From 9d88fdbc4ddfeae697d3b4ecfccc939f57701f36 Mon Sep 17 00:00:00 2001 From: Claude YOLO non-root runner Date: Sat, 18 Jul 2026 09:54:40 -0700 Subject: [PATCH 1/7] [core] Make BLOB copy buffer size configurable via blob.copy-buffer-size Introduce the `blob.copy-buffer-size` CoreOption and thread it through every BlobFormatWriter construction path, so the copy buffer used when writing BLOB payloads is no longer hard-coded to 4 KiB. The default is kept at 4 KiB, so behavior is unchanged unless the option is set explicitly; this only makes the buffer configurable. It covers the normal append / primary-key write paths (BlobFileContext -> DedicatedFormatRollingFileWriter -> MultipleBlobFileWriter, and KeyValueFileWriterFactory -> PrimaryKeyBlobExternalizer), the blob format factory, and blob compaction (DataEvolutionBlobCompactTask). Existing public constructors are preserved and delegate to the default so the change is API-compatible. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_01Ab8nexcSeG4eJ4avGKbWvd --- .../java/org/apache/paimon/CoreOptions.java | 11 ++ .../DedicatedFormatRollingFileWriter.java | 3 +- .../paimon/append/MultipleBlobFileWriter.java | 5 +- .../DataEvolutionBlobCompactTask.java | 2 +- .../blob/PrimaryKeyBlobExternalizer.java | 31 ++- .../paimon/io/KeyValueFileWriterFactory.java | 3 +- .../paimon/operation/BlobFileContext.java | 12 +- .../paimon/format/blob/BlobFileFormat.java | 9 +- .../format/blob/BlobFileFormatFactory.java | 5 +- .../paimon/format/blob/BlobFormatWriter.java | 22 ++- .../format/blob/BlobFormatWriterTest.java | 176 ++++++++++++++++++ 11 files changed, 266 insertions(+), 13 deletions(-) diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 4f1cb3111193..4434728a9fc7 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -830,6 +830,13 @@ public InlineElement getDescription() { "Whether to consider blob file size as a factor when performing scan splitting.") .build()); + public static final ConfigOption BLOB_COPY_BUFFER_SIZE = + key("blob.copy-buffer-size") + .memoryType() + .defaultValue(MemorySize.parse("4 kb")) + .withDescription( + "Buffer size used when copying BLOB payloads into BLOB files."); + public static final ConfigOption NUM_SORTED_RUNS_COMPACTION_TRIGGER = key("num-sorted-run.compaction-trigger") .intType() @@ -3286,6 +3293,10 @@ public long blobTargetFileSize() { .orElse(targetFileSize(false)); } + public int blobCopyBufferSize() { + return Math.toIntExact(options.get(BLOB_COPY_BUFFER_SIZE).getBytes()); + } + public boolean blobSplitByFileSize() { return options.getOptional(BLOB_SPLIT_BY_FILE_SIZE) .orElse(!options.get(BLOB_AS_DESCRIPTOR)); diff --git a/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java index 86df6b10b265..9b8897862a8b 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/DedicatedFormatRollingFileWriter.java @@ -201,7 +201,8 @@ public DedicatedFormatRollingFileWriter( context.blobInlineFields(), context.writeNullOnMissingFile(), context.writeNullOnFetchFailure(), - context.blobFetchMetricReporter()); + context.blobFetchMetricReporter(), + context.copyBufferSize()); } else { this.blobWriterFactory = null; } diff --git a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java index d952018ee1f0..39d6c6eac833 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java @@ -67,11 +67,12 @@ public MultipleBlobFileWriter( Set blobInlineFields, boolean writeNullOnMissingFile, boolean writeNullOnFetchFailure, - BlobFetchMetricReporter blobFetchMetricReporter) { + BlobFetchMetricReporter blobFetchMetricReporter, + int copyBufferSize) { RowType blobRowType = new RowType(fieldsInBlobFile(writeSchema, blobInlineFields)); this.blobWriters = new ArrayList<>(); for (String blobFieldName : blobRowType.getFieldNames()) { - BlobFileFormat blobFileFormat = new BlobFileFormat(); + BlobFileFormat blobFileFormat = new BlobFileFormat(false, copyBufferSize); blobFileFormat.setWriteConsumer(blobConsumer); blobFileFormat.setWriteNullOnMissingFile(writeNullOnMissingFile); blobFileFormat.setWriteNullOnFetchFailure(writeNullOnFetchFailure); diff --git a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java index 1069e7c62d86..675aaf29b986 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java @@ -128,7 +128,7 @@ private FileWriter createBlobFileWriter( RowType blobWriteType, String blobFieldName, DataFilePathFactory pathFactory) { - BlobFileFormat blobFileFormat = new BlobFileFormat(); + BlobFileFormat blobFileFormat = new BlobFileFormat(false, options.blobCopyBufferSize()); return new RowDataFileWriter( table.fileIO(), RollingFileWriter.createFileWriterContext( diff --git a/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java b/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java index 9216f7597c3e..b51093d0194a 100644 --- a/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java +++ b/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java @@ -20,6 +20,7 @@ import org.apache.paimon.data.Blob; import org.apache.paimon.data.BlobDescriptor; +import org.apache.paimon.data.BlobFetchMetricReporter; import org.apache.paimon.data.GenericArray; import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalArray; @@ -62,6 +63,22 @@ public PrimaryKeyBlobExternalizer( Set managedBlobFields, DataFilePathFactory pathFactory, long targetFileSize) { + this( + fileIO, + valueType, + managedBlobFields, + pathFactory, + targetFileSize, + BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + } + + public PrimaryKeyBlobExternalizer( + FileIO fileIO, + RowType valueType, + Set managedBlobFields, + DataFilePathFactory pathFactory, + long targetFileSize, + int copyBufferSize) { checkArgument(targetFileSize > 0, "Managed BLOB target file size must be positive."); this.fileIO = fileIO; this.rowConverter = new RowDataToObjectArrayConverter(valueType); @@ -97,7 +114,8 @@ public PrimaryKeyBlobExternalizer( : new RowType(Collections.singletonList(field)), pathFactory, targetFileSize, - uncommittedPacks)); + uncommittedPacks, + copyBufferSize)); } checkArgument( unknownFields.isEmpty(), @@ -223,6 +241,7 @@ private static class ManagedBlobPackWriter { private final DataFilePathFactory pathFactory; private final long targetFileSize; private final List uncommittedPacks; + private final int copyBufferSize; private Path currentPath; private PositionOutputStream out; @@ -234,12 +253,14 @@ private ManagedBlobPackWriter( RowType blobType, DataFilePathFactory pathFactory, long targetFileSize, - List uncommittedPacks) { + List uncommittedPacks, + int copyBufferSize) { this.fileIO = fileIO; this.blobType = blobType; this.pathFactory = pathFactory; this.targetFileSize = targetFileSize; this.uncommittedPacks = uncommittedPacks; + this.copyBufferSize = copyBufferSize; } private BlobDescriptor write(Blob blob) throws IOException { @@ -271,7 +292,11 @@ private void openCurrent() throws IOException { lastDescriptor = descriptor; return false; }, - blobType); + blobType, + false, + false, + BlobFetchMetricReporter.NOOP, + copyBufferSize); writer.setFile(currentPath); } diff --git a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java index 547d827a73b4..dfe875120dcf 100644 --- a/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java +++ b/paimon-core/src/main/java/org/apache/paimon/io/KeyValueFileWriterFactory.java @@ -96,7 +96,8 @@ private KeyValueFileWriterFactory( valueType, managedBlobFields, formatContext.pathFactory(new WriteFormatKey(0, false)), - options.blobTargetFileSize()); + options.blobTargetFileSize(), + options.blobCopyBufferSize()); } public RowType keyType() { diff --git a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java index 9d79ebe6eb6c..b1b50e157b82 100644 --- a/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java +++ b/paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java @@ -36,6 +36,7 @@ public class BlobFileContext { private final Set blobInlineFields; private final boolean writeNullOnMissingFile; private final boolean writeNullOnFetchFailure; + private final int copyBufferSize; private @Nullable BlobConsumer blobConsumer; private BlobFetchMetricReporter blobFetchMetricReporter = BlobFetchMetricReporter.NOOP; @@ -44,11 +45,13 @@ private BlobFileContext( Set blobDescriptorFields, Set blobInlineFields, boolean writeNullOnMissingFile, - boolean writeNullOnFetchFailure) { + boolean writeNullOnFetchFailure, + int copyBufferSize) { this.blobDescriptorFields = blobDescriptorFields; this.blobInlineFields = blobInlineFields; this.writeNullOnMissingFile = writeNullOnMissingFile; this.writeNullOnFetchFailure = writeNullOnFetchFailure; + this.copyBufferSize = copyBufferSize; } @Nullable @@ -72,7 +75,8 @@ public static BlobFileContext create(RowType rowType, CoreOptions options) { descriptorFields, inlineFields, options.blobWriteNullOnMissingFile(), - options.blobWriteNullOnFetchFailure()); + options.blobWriteNullOnFetchFailure(), + options.blobCopyBufferSize()); } public BlobFileContext withBlobConsumer(BlobConsumer blobConsumer) { @@ -114,6 +118,10 @@ public boolean writeNullOnFetchFailure() { return writeNullOnFetchFailure; } + public int copyBufferSize() { + return copyBufferSize; + } + public BlobFetchMetricReporter blobFetchMetricReporter() { return blobFetchMetricReporter; } diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java index 5b0f359c7df1..e3bf4851b58e 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java @@ -53,6 +53,7 @@ public class BlobFileFormat extends FileFormat { private final boolean blobAsDescriptor; + private final int copyBufferSize; private boolean writeNullOnMissingFile; private boolean writeNullOnFetchFailure; private BlobFetchMetricReporter blobFetchMetricReporter = BlobFetchMetricReporter.NOOP; @@ -64,8 +65,13 @@ public BlobFileFormat() { } public BlobFileFormat(boolean blobAsDescriptor) { + this(blobAsDescriptor, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + } + + public BlobFileFormat(boolean blobAsDescriptor, int copyBufferSize) { super(BlobFileFormatFactory.IDENTIFIER); this.blobAsDescriptor = blobAsDescriptor; + this.copyBufferSize = copyBufferSize; } public static boolean isBlobFile(String fileName) { @@ -132,7 +138,8 @@ public FormatWriter create(PositionOutputStream out, String compression) { type, writeNullOnMissingFile, writeNullOnFetchFailure, - blobFetchMetricReporter); + blobFetchMetricReporter, + copyBufferSize); } } diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormatFactory.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormatFactory.java index 2a54d497093b..82545fcfa358 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormatFactory.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormatFactory.java @@ -35,6 +35,9 @@ public String identifier() { @Override public FileFormat create(FormatContext formatContext) { boolean blobAsDescriptor = formatContext.options().get(CoreOptions.BLOB_AS_DESCRIPTOR); - return new BlobFileFormat(blobAsDescriptor); + int copyBufferSize = + Math.toIntExact( + formatContext.options().get(CoreOptions.BLOB_COPY_BUFFER_SIZE).getBytes()); + return new BlobFileFormat(blobAsDescriptor, copyBufferSize); } } diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java index 9de5a169d1e2..ac87e4ec9023 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java @@ -64,6 +64,7 @@ public class BlobFormatWriter implements FileAwareFormatWriter { public static final long NULL_LENGTH = -1L; public static final long PLACE_HOLDER_LENGTH = -2L; public static final long ARRAY_NULL_ELEMENT_LENGTH = -1L; + public static final int DEFAULT_COPY_BUFFER_SIZE = 4 * 1024; private final PositionOutputStream out; @Nullable private final BlobConsumer writeConsumer; @@ -113,6 +114,25 @@ public BlobFormatWriter( boolean writeNullOnMissingFile, boolean writeNullOnFetchFailure, BlobFetchMetricReporter blobFetchMetricReporter) { + this( + out, + writeConsumer, + type, + writeNullOnMissingFile, + writeNullOnFetchFailure, + blobFetchMetricReporter, + DEFAULT_COPY_BUFFER_SIZE); + } + + public BlobFormatWriter( + PositionOutputStream out, + @Nullable BlobConsumer writeConsumer, + RowType type, + boolean writeNullOnMissingFile, + boolean writeNullOnFetchFailure, + BlobFetchMetricReporter blobFetchMetricReporter, + int copyBufferSize) { + checkArgument(copyBufferSize > 0, "BLOB copy buffer size must be positive."); this.out = out; this.writeConsumer = writeConsumer; this.blobFetchMetricReporter = blobFetchMetricReporter; @@ -125,7 +145,7 @@ public BlobFormatWriter( ? new ArrayBlobElementWriter() : new RawBlobElementWriter(); this.crc32 = new CRC32(); - this.tmpBuffer = new byte[4096]; + this.tmpBuffer = new byte[copyBufferSize]; this.lengths = new LongArrayList(16); } diff --git a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java index 85e431458953..f1d66e102bbb 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java @@ -43,6 +43,10 @@ import org.junit.jupiter.api.io.TempDir; import java.nio.file.Files; +import java.util.ArrayList; +import java.util.LinkedHashMap; +import java.util.List; +import java.util.Map; import static org.assertj.core.api.Assertions.assertThat; import static org.assertj.core.api.Assertions.assertThatThrownBy; @@ -477,6 +481,178 @@ public void testArrayBlobFetchMetricReporterForUnhandledFailure( assertThat(metricReporter.fetchFailureNullWritten).isEqualTo(0); } + @Test + public void testCopyBufferSizeIsRespectedForBlobRef(@TempDir java.nio.file.Path tempDir) + throws Exception { + String uri = "mem://file"; + byte[] source = sequentialBytes(20); + RecordingUriReader reader = new RecordingUriReader(singleFile(uri, source)); + java.nio.file.Path outputFile = tempDir.resolve("blob.out"); + + BlobFormatWriter writer = newWriter(outputFile, RowType.of(DataTypes.BLOB()), 8); + writer.addElement(GenericRow.of(new BlobRef(reader, new BlobDescriptor(uri, 0, 20)))); + writer.close(); + + // With an 8-byte copy buffer, no single read request exceeds 8 bytes. + assertThat(reader.opened).hasSize(1); + assertThat(reader.opened.get(0).maxReadRequest).isEqualTo(8); + assertThat(readBackBlobs(outputFile, 1)).containsExactly(source); + } + + @Test + public void testDefaultCopyBufferSize(@TempDir java.nio.file.Path tempDir) throws Exception { + // default preserves the historical 4 KiB copy buffer. + assertThat(BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE).isEqualTo(4 * 1024); + + String uri = "mem://file"; + byte[] source = sequentialBytes(5000); + RecordingUriReader reader = new RecordingUriReader(singleFile(uri, source)); + java.nio.file.Path outputFile = tempDir.resolve("blob.out"); + + // default constructor -> default 4 KiB copy buffer. + BlobFormatWriter writer = + new BlobFormatWriter( + new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), + null, + RowType.of(DataTypes.BLOB())); + writer.addElement(GenericRow.of(new BlobRef(reader, new BlobDescriptor(uri, 0, 5000)))); + writer.close(); + + assertThat(reader.opened.get(0).maxReadRequest).isEqualTo(4 * 1024); + assertThat(readBackBlobs(outputFile, 1)).containsExactly(source); + } + + private static BlobFormatWriter newWriter( + java.nio.file.Path outputFile, RowType rowType, int copyBufferSize) + throws java.io.FileNotFoundException { + return new BlobFormatWriter( + new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), + null, + rowType, + false, + false, + BlobFetchMetricReporter.NOOP, + copyBufferSize); + } + + private static List readBackBlobs(java.nio.file.Path outputFile, int expectedCount) + throws Exception { + LocalFileIO fileIO = new LocalFileIO(); + Path filePath = new Path(outputFile.toUri()); + long fileSize = Files.size(outputFile); + List result = new ArrayList<>(); + try (SeekableInputStream in = fileIO.newInputStream(filePath)) { + BlobFileMeta fileMeta = new BlobFileMeta(in, fileSize, null); + assertThat(fileMeta.recordNumber()).isEqualTo(expectedCount); + BlobFormatReader reader = + new BlobFormatReader( + fileIO, filePath, fileMeta, in, 1, 0, DataTypes.BLOB(), false); + FileRecordIterator iterator = reader.readBatch(); + for (int i = 0; i < expectedCount; i++) { + InternalRow row = iterator.next(); + assertThat(row).isNotNull(); + result.add(readAll(row.getBlob(0))); + } + } + return result; + } + + private static byte[] readAll(Blob blob) throws Exception { + try (SeekableInputStream in = blob.newInputStream()) { + return org.apache.paimon.utils.IOUtils.readFully(in, false); + } + } + + private static byte[] sequentialBytes(int length) { + byte[] bytes = new byte[length]; + for (int i = 0; i < length; i++) { + bytes[i] = (byte) i; + } + return bytes; + } + + private static Map singleFile(String uri, byte[] data) { + Map files = new LinkedHashMap<>(); + files.put(uri, data); + return files; + } + + /** A {@link UriReader} over in-memory files that records opened streams. */ + private static final class RecordingUriReader implements UriReader { + + private final Map files; + private final List opened = new ArrayList<>(); + private int openCount; + + private RecordingUriReader(Map files) { + this.files = files; + } + + @Override + public SeekableInputStream newInputStream(String uri) { + byte[] data = files.get(uri); + if (data == null) { + throw new IllegalArgumentException("Unknown uri: " + uri); + } + openCount++; + CountingSeekableInputStream stream = new CountingSeekableInputStream(data); + opened.add(stream); + return stream; + } + } + + /** A seekable stream over a byte array that records close count and max read request size. */ + private static final class CountingSeekableInputStream extends SeekableInputStream { + + private final byte[] data; + private int pos; + private int closeCount; + private int maxReadRequest; + + private CountingSeekableInputStream(byte[] data) { + this.data = data; + } + + @Override + public void seek(long desired) { + this.pos = (int) desired; + } + + @Override + public long getPos() { + return pos; + } + + @Override + public int read() { + maxReadRequest = Math.max(maxReadRequest, 1); + if (pos >= data.length) { + return -1; + } + return data[pos++] & 0xFF; + } + + @Override + public int read(byte[] b, int off, int len) { + maxReadRequest = Math.max(maxReadRequest, len); + if (len == 0) { + return 0; + } + if (pos >= data.length) { + return -1; + } + int n = Math.min(len, data.length - pos); + System.arraycopy(data, pos, b, off, n); + pos += n; + return n; + } + + @Override + public void close() { + closeCount++; + } + } + private static void assertBlobPayload(Blob blob, byte[] expected) throws Exception { try (SeekableInputStream blobIn = blob.newInputStream()) { byte[] actual = new byte[expected.length]; From 16b03bda5cba227deea14f243e480692857ad4b1 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 18 Jul 2026 21:59:53 -0700 Subject: [PATCH 2/7] [core] Validate blob.copy-buffer-size and add it to generated config docs Reject zero / negative / oversized values at config-read time with a clear, option-named message, instead of a mid-write failure or a bare ArithmeticException. Add the option to docs/generated/core_configuration.html so ConfigOptionsDocsCompletenessITCase passes, and add zero/oversized tests. --- docs/generated/core_configuration.html | 6 +++++ .../java/org/apache/paimon/CoreOptions.java | 14 ++++++++++- .../org/apache/paimon/CoreOptionsTest.java | 23 +++++++++++++++++++ .../format/blob/BlobFileFormatFactory.java | 2 +- .../format/blob/BlobFormatWriterTest.java | 9 ++------ 5 files changed, 45 insertions(+), 9 deletions(-) diff --git a/docs/generated/core_configuration.html b/docs/generated/core_configuration.html index 143c244451a9..49500e1b61fc 100644 --- a/docs/generated/core_configuration.html +++ b/docs/generated/core_configuration.html @@ -104,6 +104,12 @@ Boolean Whether to write NULL for a descriptor BLOB value when the referenced file or HTTP resource does not exist during Flink writes. When false, the write fails when the descriptor is read. + +
blob.copy-buffer-size
+ 4 kb + MemorySize + Buffer size used when copying BLOB payloads into BLOB files. +
blob.split-by-file-size
(none) diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 4434728a9fc7..04ef7cbcb4b3 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -830,6 +830,7 @@ public InlineElement getDescription() { "Whether to consider blob file size as a factor when performing scan splitting.") .build()); + // Keep this default in sync with BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE (= 4 * 1024). public static final ConfigOption BLOB_COPY_BUFFER_SIZE = key("blob.copy-buffer-size") .memoryType() @@ -3294,7 +3295,18 @@ public long blobTargetFileSize() { } public int blobCopyBufferSize() { - return Math.toIntExact(options.get(BLOB_COPY_BUFFER_SIZE).getBytes()); + return checkedBlobCopyBufferSize(options.get(BLOB_COPY_BUFFER_SIZE).getBytes()); + } + + /** Validates {@link #BLOB_COPY_BUFFER_SIZE} bytes and narrows to a positive int. */ + public static int checkedBlobCopyBufferSize(long bytes) { + checkArgument( + bytes > 0 && bytes <= Integer.MAX_VALUE, + "'%s' must be between 1 byte and %s bytes, but was %s bytes.", + BLOB_COPY_BUFFER_SIZE.key(), + Integer.MAX_VALUE, + bytes); + return (int) bytes; } public boolean blobSplitByFileSize() { diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index 3399990a8a4b..568e649f1f64 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -18,6 +18,7 @@ package org.apache.paimon; +import org.apache.paimon.options.MemorySize; import org.apache.paimon.options.Options; import org.junit.jupiter.api.Test; @@ -157,4 +158,26 @@ public void testMapStorageLayout() { assertThatThrownBy(() -> negativeMaxColumnsOptions.mapSharedShreddingMaxColumns("metrics")) .hasMessageContaining("options map.shared-shredding.max-columns must > 0"); } + + @Test + public void testBlobCopyBufferSize() { + Options conf = new Options(); + // default preserves the historical 4 KiB buffer. + assertThat(new CoreOptions(conf).blobCopyBufferSize()).isEqualTo(4 * 1024); + + conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("64 kb")); + assertThat(new CoreOptions(conf).blobCopyBufferSize()).isEqualTo(64 * 1024); + + // zero is rejected early with an option-named message. + conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("0 bytes")); + assertThatThrownBy(() -> new CoreOptions(conf).blobCopyBufferSize()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("blob.copy-buffer-size"); + + // oversized (> Integer.MAX_VALUE) is rejected instead of throwing ArithmeticException. + conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("3 gb")); + assertThatThrownBy(() -> new CoreOptions(conf).blobCopyBufferSize()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("blob.copy-buffer-size"); + } } diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormatFactory.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormatFactory.java index 82545fcfa358..625203bfbdde 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormatFactory.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormatFactory.java @@ -36,7 +36,7 @@ public String identifier() { public FileFormat create(FormatContext formatContext) { boolean blobAsDescriptor = formatContext.options().get(CoreOptions.BLOB_AS_DESCRIPTOR); int copyBufferSize = - Math.toIntExact( + CoreOptions.checkedBlobCopyBufferSize( formatContext.options().get(CoreOptions.BLOB_COPY_BUFFER_SIZE).getBytes()); return new BlobFileFormat(blobAsDescriptor, copyBufferSize); } diff --git a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java index f1d66e102bbb..c350a1619124 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java @@ -582,7 +582,6 @@ private static final class RecordingUriReader implements UriReader { private final Map files; private final List opened = new ArrayList<>(); - private int openCount; private RecordingUriReader(Map files) { this.files = files; @@ -594,7 +593,6 @@ public SeekableInputStream newInputStream(String uri) { if (data == null) { throw new IllegalArgumentException("Unknown uri: " + uri); } - openCount++; CountingSeekableInputStream stream = new CountingSeekableInputStream(data); opened.add(stream); return stream; @@ -606,7 +604,6 @@ private static final class CountingSeekableInputStream extends SeekableInputStre private final byte[] data; private int pos; - private int closeCount; private int maxReadRequest; private CountingSeekableInputStream(byte[] data) { @@ -634,10 +631,10 @@ public int read() { @Override public int read(byte[] b, int off, int len) { - maxReadRequest = Math.max(maxReadRequest, len); if (len == 0) { return 0; } + maxReadRequest = Math.max(maxReadRequest, len); if (pos >= data.length) { return -1; } @@ -648,9 +645,7 @@ public int read(byte[] b, int off, int len) { } @Override - public void close() { - closeCount++; - } + public void close() {} } private static void assertBlobPayload(Blob blob, byte[] expected) throws Exception { From 979d445575758c0bdea98dee288f067d7addf9d0 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 18 Jul 2026 21:59:53 -0700 Subject: [PATCH 3/7] [python] Support 'blob.copy-buffer-size' in pypaimon Make the BLOB copy buffer size configurable in pypaimon, mirroring the Java option (default 4 KiB). The value is read via CoreOptions.blob_copy_buffer_size and threaded through BlobWriter -> BlobFileWriter -> BlobFormatWriter, replacing the hard-coded 4096 copy buffer. --- .../pypaimon/common/options/core_options.py | 14 ++++++++++++++ paimon-python/pypaimon/write/blob_format_writer.py | 10 +++++++--- .../pypaimon/write/writer/blob_file_writer.py | 4 +++- paimon-python/pypaimon/write/writer/blob_writer.py | 8 +++++++- 4 files changed, 31 insertions(+), 5 deletions(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 4d2e6a7bb8bc..aaddcc8610e1 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -395,6 +395,13 @@ class CoreOptions: .with_description("The target file size for blob files.") ) + BLOB_COPY_BUFFER_SIZE: ConfigOption[MemorySize] = ( + ConfigOptions.key("blob.copy-buffer-size") + .memory_type() + .default_value(MemorySize.of_kibi_bytes(4)) + .with_description("Buffer size used when copying BLOB payloads into BLOB files.") + ) + VECTOR_FILE_FORMAT: ConfigOption[str] = ( ConfigOptions.key("vector.file.format") .string_type() @@ -1111,6 +1118,13 @@ def blob_target_file_size(self, default=None): else: return self.target_file_size(has_primary_key=False) + def blob_copy_buffer_size(self): + size = self.options.get(CoreOptions.BLOB_COPY_BUFFER_SIZE, None).get_bytes() + if size <= 0: + raise ValueError( + f"'{CoreOptions.BLOB_COPY_BUFFER_SIZE.key()}' must be positive, but was {size} bytes.") + return size + def vector_file_format(self, default=None): return self.options.get(CoreOptions.VECTOR_FILE_FORMAT, default) diff --git a/paimon-python/pypaimon/write/blob_format_writer.py b/paimon-python/pypaimon/write/blob_format_writer.py index 36c1044443f4..24f937ba0bde 100644 --- a/paimon-python/pypaimon/write/blob_format_writer.py +++ b/paimon-python/pypaimon/write/blob_format_writer.py @@ -41,10 +41,14 @@ class BlobFormatWriter: def __init__(self, output_stream: BinaryIO, blob_consumer: Optional[BlobConsumer] = None, - file_path: Optional[str] = None): + file_path: Optional[str] = None, + copy_buffer_size: int = BUFFER_SIZE): + if copy_buffer_size <= 0: + raise ValueError("BLOB copy buffer size must be positive") self.output_stream = output_stream self._blob_consumer = blob_consumer self._file_path = file_path + self.copy_buffer_size = copy_buffer_size self.lengths: List[int] = [] self.position = 0 @@ -176,10 +180,10 @@ def _write_blob_data(self, blob_value: Blob, crc32: int): else: stream = blob_value.new_input_stream() try: - chunk = stream.read(self.BUFFER_SIZE) + chunk = stream.read(self.copy_buffer_size) while chunk: crc32 = self._write_with_crc(chunk, crc32) - chunk = stream.read(self.BUFFER_SIZE) + chunk = stream.read(self.copy_buffer_size) finally: stream.close() diff --git a/paimon-python/pypaimon/write/writer/blob_file_writer.py b/paimon-python/pypaimon/write/writer/blob_file_writer.py index 728d86c50fab..b366dab96ea1 100644 --- a/paimon-python/pypaimon/write/writer/blob_file_writer.py +++ b/paimon-python/pypaimon/write/writer/blob_file_writer.py @@ -36,7 +36,8 @@ class BlobFileWriter: Writes rows one by one and tracks file size. """ - def __init__(self, file_io, file_path: Path, blob_consumer: Optional[BlobConsumer] = None): + def __init__(self, file_io, file_path: Path, blob_consumer: Optional[BlobConsumer] = None, + copy_buffer_size: int = BlobFormatWriter.BUFFER_SIZE): self.file_io = file_io self.file_path = file_path self._blob_consumer = blob_consumer @@ -45,6 +46,7 @@ def __init__(self, file_io, file_path: Path, blob_consumer: Optional[BlobConsume self.output_stream, blob_consumer=blob_consumer, file_path=str(file_path), + copy_buffer_size=copy_buffer_size, ) self.row_count = 0 self.closed = False diff --git a/paimon-python/pypaimon/write/writer/blob_writer.py b/paimon-python/pypaimon/write/writer/blob_writer.py index 2e0e682130c1..0ab079959bdd 100644 --- a/paimon-python/pypaimon/write/writer/blob_writer.py +++ b/paimon-python/pypaimon/write/writer/blob_writer.py @@ -44,6 +44,7 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, bl options = self.table.options self.blob_target_file_size = CoreOptions.blob_target_file_size(options) + self.blob_copy_buffer_size = CoreOptions.blob_copy_buffer_size(options) self._blob_consumer = blob_consumer self.current_writer: Optional[BlobFileWriter] = None @@ -100,7 +101,12 @@ def open_current_writer(self): self.file_count += 1 # Increment counter for next file file_path = self._generate_file_path(file_name) self.current_file_path = file_path - self.current_writer = BlobFileWriter(self.file_io, file_path, blob_consumer=self._blob_consumer) + self.current_writer = BlobFileWriter( + self.file_io, + file_path, + blob_consumer=self._blob_consumer, + copy_buffer_size=self.blob_copy_buffer_size, + ) def rolling_file(self) -> bool: if self.current_writer is None: From 92f815960a48f57246da924b79dd667eff303f97 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sat, 18 Jul 2026 23:09:38 -0700 Subject: [PATCH 4/7] [python] Cap blob.copy-buffer-size at Integer.MAX_VALUE in pypaimon The pypaimon accessor and BlobFormatWriter only rejected non-positive sizes, so a value such as '3 gb' was accepted (3221225472) and passed to stream.read(), risking a huge allocation and diverging from Java, which caps the copy buffer at an int. Reject sizes outside 1 .. 2**31 - 1 in both CoreOptions.blob_copy_buffer_size and BlobFormatWriter, and add boundary tests. --- .../pypaimon/common/options/core_options.py | 6 ++++-- paimon-python/pypaimon/tests/blob_test.py | 21 +++++++++++++++++++ .../pypaimon/write/blob_format_writer.py | 7 +++++-- 3 files changed, 30 insertions(+), 4 deletions(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index aaddcc8610e1..1deaa51307cc 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -1120,9 +1120,11 @@ def blob_target_file_size(self, default=None): def blob_copy_buffer_size(self): size = self.options.get(CoreOptions.BLOB_COPY_BUFFER_SIZE, None).get_bytes() - if size <= 0: + # Upper bound mirrors the Java int copy buffer (Integer.MAX_VALUE). + if not 1 <= size <= 2 ** 31 - 1: raise ValueError( - f"'{CoreOptions.BLOB_COPY_BUFFER_SIZE.key()}' must be positive, but was {size} bytes.") + f"'{CoreOptions.BLOB_COPY_BUFFER_SIZE.key()}' must be between 1 byte and " + f"{2 ** 31 - 1} bytes, but was {size} bytes.") return size def vector_file_format(self, default=None): diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index 9da03898e4a2..a57a0f9a4def 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -130,6 +130,27 @@ def tearDown(self): except OSError: pass # Ignore cleanup errors + def test_blob_copy_buffer_size_validation(self): + """blob.copy-buffer-size must be within 1 .. Integer.MAX_VALUE, matching Java.""" + from pypaimon.write.blob_format_writer import BlobFormatWriter + from pypaimon.common.options.core_options import CoreOptions + + # BlobFormatWriter rejects non-positive and oversized (> 2**31 - 1) buffers. + for bad in [0, -1, 2 ** 31]: + with self.assertRaises(ValueError): + BlobFormatWriter(io.BytesIO(), copy_buffer_size=bad) + # a valid custom buffer is accepted. + BlobFormatWriter(io.BytesIO(), copy_buffer_size=8).close() + + # CoreOptions accessor defaults to 4 KiB and validates the same bounds. + self.assertEqual(CoreOptions(Options({})).blob_copy_buffer_size(), 4096) + self.assertEqual( + CoreOptions(Options({'blob.copy-buffer-size': '256 kb'})).blob_copy_buffer_size(), + 256 * 1024) + for bad in ['0 bytes', '3 gb']: + with self.assertRaises(ValueError): + CoreOptions(Options({'blob.copy-buffer-size': bad})).blob_copy_buffer_size() + def test_from_data(self): """Test Blob.from_data() method.""" test_data = b"test data" diff --git a/paimon-python/pypaimon/write/blob_format_writer.py b/paimon-python/pypaimon/write/blob_format_writer.py index 24f937ba0bde..5aa3119ec16b 100644 --- a/paimon-python/pypaimon/write/blob_format_writer.py +++ b/paimon-python/pypaimon/write/blob_format_writer.py @@ -37,14 +37,17 @@ class BlobFormatWriter: PLACE_HOLDER_LENGTH = -2 ARRAY_NULL_ELEMENT_LENGTH = -1 BUFFER_SIZE = 4096 + MAX_BUFFER_SIZE = 2 ** 31 - 1 # keep in sync with the Java int copy-buffer upper bound METADATA_SIZE = 12 # 8-byte length + 4-byte CRC def __init__(self, output_stream: BinaryIO, blob_consumer: Optional[BlobConsumer] = None, file_path: Optional[str] = None, copy_buffer_size: int = BUFFER_SIZE): - if copy_buffer_size <= 0: - raise ValueError("BLOB copy buffer size must be positive") + if not 1 <= copy_buffer_size <= self.MAX_BUFFER_SIZE: + raise ValueError( + f"BLOB copy buffer size must be between 1 and {self.MAX_BUFFER_SIZE}, " + f"but was {copy_buffer_size}.") self.output_stream = output_stream self._blob_consumer = blob_consumer self._file_path = file_path From 20eda7a2e222b5eb2f5b91374de3680e84769c0f Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 19 Jul 2026 01:09:19 -0700 Subject: [PATCH 5/7] [core] Cap blob.copy-buffer-size at 256 MiB and restore MultipleBlobFileWriter overload Integer.MAX_VALUE was accepted but new byte[copyBufferSize] then fails with OOM (exceeds the JVM array limit) and is a per-writer allocation footgun. Reject sizes above CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE (256 MiB) at config-read time and in the writer, with a clear message. Also restore the previous 14-arg MultipleBlobFileWriter constructor (delegating to the default buffer) so the change stays source/binary compatible, as the PR claims. --- .../java/org/apache/paimon/CoreOptions.java | 7 ++-- .../paimon/append/MultipleBlobFileWriter.java | 34 +++++++++++++++++++ .../org/apache/paimon/CoreOptionsTest.java | 11 +++++- .../paimon/format/blob/BlobFormatWriter.java | 7 +++- 4 files changed, 55 insertions(+), 4 deletions(-) diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index 04ef7cbcb4b3..c53f1e6a23c8 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -830,6 +830,9 @@ public InlineElement getDescription() { "Whether to consider blob file size as a factor when performing scan splitting.") .build()); + // Safety ceiling for blob.copy-buffer-size (allocated as new byte[size] per writer). + public static final int MAX_BLOB_COPY_BUFFER_SIZE = 256 * 1024 * 1024; + // Keep this default in sync with BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE (= 4 * 1024). public static final ConfigOption BLOB_COPY_BUFFER_SIZE = key("blob.copy-buffer-size") @@ -3301,10 +3304,10 @@ public int blobCopyBufferSize() { /** Validates {@link #BLOB_COPY_BUFFER_SIZE} bytes and narrows to a positive int. */ public static int checkedBlobCopyBufferSize(long bytes) { checkArgument( - bytes > 0 && bytes <= Integer.MAX_VALUE, + bytes > 0 && bytes <= MAX_BLOB_COPY_BUFFER_SIZE, "'%s' must be between 1 byte and %s bytes, but was %s bytes.", BLOB_COPY_BUFFER_SIZE.key(), - Integer.MAX_VALUE, + MAX_BLOB_COPY_BUFFER_SIZE, bytes); return (int) bytes; } diff --git a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java index 39d6c6eac833..8c37f28f83da 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java @@ -23,6 +23,7 @@ import org.apache.paimon.data.InternalRow; import org.apache.paimon.fileindex.FileIndexOptions; import org.apache.paimon.format.blob.BlobFileFormat; +import org.apache.paimon.format.blob.BlobFormatWriter; import org.apache.paimon.fs.FileIO; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFilePathFactory; @@ -53,6 +54,39 @@ public class MultipleBlobFileWriter implements Closeable { private final List blobWriters; + public MultipleBlobFileWriter( + FileIO fileIO, + long schemaId, + RowType writeSchema, + DataFilePathFactory pathFactory, + Supplier seqNumCounterSupplier, + FileSource fileSource, + boolean asyncFileWrite, + boolean statsDenseStore, + long targetFileSize, + @Nullable BlobConsumer blobConsumer, + Set blobInlineFields, + boolean writeNullOnMissingFile, + boolean writeNullOnFetchFailure, + BlobFetchMetricReporter blobFetchMetricReporter) { + this( + fileIO, + schemaId, + writeSchema, + pathFactory, + seqNumCounterSupplier, + fileSource, + asyncFileWrite, + statsDenseStore, + targetFileSize, + blobConsumer, + blobInlineFields, + writeNullOnMissingFile, + writeNullOnFetchFailure, + blobFetchMetricReporter, + BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + } + public MultipleBlobFileWriter( FileIO fileIO, long schemaId, diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index 568e649f1f64..f89f10f65d26 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -174,7 +174,16 @@ public void testBlobCopyBufferSize() { .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("blob.copy-buffer-size"); - // oversized (> Integer.MAX_VALUE) is rejected instead of throwing ArithmeticException. + // the max (256 MiB) is accepted, just above it is rejected. + conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("256 mb")); + assertThat(new CoreOptions(conf).blobCopyBufferSize()) + .isEqualTo(CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE); + conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("512 mb")); + assertThatThrownBy(() -> new CoreOptions(conf).blobCopyBufferSize()) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("blob.copy-buffer-size"); + + // way oversized is rejected instead of throwing ArithmeticException / OOM. conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("3 gb")); assertThatThrownBy(() -> new CoreOptions(conf).blobCopyBufferSize()) .isInstanceOf(IllegalArgumentException.class) diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java index ac87e4ec9023..dbec5c7a55c1 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java @@ -18,6 +18,7 @@ package org.apache.paimon.format.blob; +import org.apache.paimon.CoreOptions; import org.apache.paimon.data.Blob; import org.apache.paimon.data.BlobArrayPlaceholder; import org.apache.paimon.data.BlobConsumer; @@ -132,7 +133,11 @@ public BlobFormatWriter( boolean writeNullOnFetchFailure, BlobFetchMetricReporter blobFetchMetricReporter, int copyBufferSize) { - checkArgument(copyBufferSize > 0, "BLOB copy buffer size must be positive."); + checkArgument( + copyBufferSize > 0 && copyBufferSize <= CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE, + "BLOB copy buffer size must be between 1 and %s, but was %s.", + CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE, + copyBufferSize); this.out = out; this.writeConsumer = writeConsumer; this.blobFetchMetricReporter = blobFetchMetricReporter; From 8c1b3c8ebd43b84f84cbf82c4ca1b0db68df259f Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 19 Jul 2026 01:09:19 -0700 Subject: [PATCH 6/7] [python] Read effective options and cap blob.copy-buffer-size at 256 MiB in pypaimon BlobWriter read the copy buffer size from self.table.options, ignoring the effective options passed to the writer, so a custom value was dropped. Use self.options.blob_copy_buffer_size(), consistent with the other accessors. Also cap the size at 256 MiB (was 2**31 - 1) to match Java and avoid a huge per-writer allocation, and pin it with boundary tests. --- .../pypaimon/common/options/core_options.py | 7 ++++--- paimon-python/pypaimon/tests/blob_test.py | 15 ++++++++++----- .../pypaimon/write/blob_format_writer.py | 2 +- .../pypaimon/write/writer/blob_writer.py | 3 ++- 4 files changed, 17 insertions(+), 10 deletions(-) diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 1deaa51307cc..4595e39de98f 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -1120,11 +1120,12 @@ def blob_target_file_size(self, default=None): def blob_copy_buffer_size(self): size = self.options.get(CoreOptions.BLOB_COPY_BUFFER_SIZE, None).get_bytes() - # Upper bound mirrors the Java int copy buffer (Integer.MAX_VALUE). - if not 1 <= size <= 2 ** 31 - 1: + # Upper bound matches Java CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE (256 MiB). + max_size = 256 * 1024 * 1024 + if not 1 <= size <= max_size: raise ValueError( f"'{CoreOptions.BLOB_COPY_BUFFER_SIZE.key()}' must be between 1 byte and " - f"{2 ** 31 - 1} bytes, but was {size} bytes.") + f"{max_size} bytes, but was {size} bytes.") return size def vector_file_format(self, default=None): diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index a57a0f9a4def..b574a24079c8 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -131,23 +131,28 @@ def tearDown(self): pass # Ignore cleanup errors def test_blob_copy_buffer_size_validation(self): - """blob.copy-buffer-size must be within 1 .. Integer.MAX_VALUE, matching Java.""" + """blob.copy-buffer-size must be within 1 .. 256 MiB, matching Java.""" from pypaimon.write.blob_format_writer import BlobFormatWriter from pypaimon.common.options.core_options import CoreOptions - # BlobFormatWriter rejects non-positive and oversized (> 2**31 - 1) buffers. - for bad in [0, -1, 2 ** 31]: + max_size = 256 * 1024 * 1024 + # BlobFormatWriter rejects non-positive and oversized (> 256 MiB) buffers. + for bad in [0, -1, max_size + 1]: with self.assertRaises(ValueError): BlobFormatWriter(io.BytesIO(), copy_buffer_size=bad) - # a valid custom buffer is accepted. + # valid custom buffers (including the max) are accepted. BlobFormatWriter(io.BytesIO(), copy_buffer_size=8).close() + BlobFormatWriter(io.BytesIO(), copy_buffer_size=max_size).close() # CoreOptions accessor defaults to 4 KiB and validates the same bounds. self.assertEqual(CoreOptions(Options({})).blob_copy_buffer_size(), 4096) self.assertEqual( CoreOptions(Options({'blob.copy-buffer-size': '256 kb'})).blob_copy_buffer_size(), 256 * 1024) - for bad in ['0 bytes', '3 gb']: + self.assertEqual( + CoreOptions(Options({'blob.copy-buffer-size': '256 mb'})).blob_copy_buffer_size(), + max_size) + for bad in ['0 bytes', '512 mb', '3 gb']: with self.assertRaises(ValueError): CoreOptions(Options({'blob.copy-buffer-size': bad})).blob_copy_buffer_size() diff --git a/paimon-python/pypaimon/write/blob_format_writer.py b/paimon-python/pypaimon/write/blob_format_writer.py index 5aa3119ec16b..b64cd19cf7d5 100644 --- a/paimon-python/pypaimon/write/blob_format_writer.py +++ b/paimon-python/pypaimon/write/blob_format_writer.py @@ -37,7 +37,7 @@ class BlobFormatWriter: PLACE_HOLDER_LENGTH = -2 ARRAY_NULL_ELEMENT_LENGTH = -1 BUFFER_SIZE = 4096 - MAX_BUFFER_SIZE = 2 ** 31 - 1 # keep in sync with the Java int copy-buffer upper bound + MAX_BUFFER_SIZE = 256 * 1024 * 1024 # matches Java CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE METADATA_SIZE = 12 # 8-byte length + 4-byte CRC def __init__(self, output_stream: BinaryIO, diff --git a/paimon-python/pypaimon/write/writer/blob_writer.py b/paimon-python/pypaimon/write/writer/blob_writer.py index 0ab079959bdd..c5bf143728a7 100644 --- a/paimon-python/pypaimon/write/writer/blob_writer.py +++ b/paimon-python/pypaimon/write/writer/blob_writer.py @@ -44,7 +44,8 @@ def __init__(self, table, partition: Tuple, bucket: int, max_seq_number: int, bl options = self.table.options self.blob_target_file_size = CoreOptions.blob_target_file_size(options) - self.blob_copy_buffer_size = CoreOptions.blob_copy_buffer_size(options) + # Use the effective options (constructor-provided), consistent with the other accessors. + self.blob_copy_buffer_size = self.options.blob_copy_buffer_size() self._blob_consumer = blob_consumer self.current_writer: Optional[BlobFileWriter] = None From b61dc8bc704c13ef263dfaff0153395a5e7c81e3 Mon Sep 17 00:00:00 2001 From: xiaohongbo Date: Sun, 19 Jul 2026 23:20:29 -0700 Subject: [PATCH 7/7] [core] Address BLOB copy buffer review comments --- .../java/org/apache/paimon/CoreOptions.java | 7 +- .../paimon/append/MultipleBlobFileWriter.java | 34 ---- .../blob/PrimaryKeyBlobExternalizer.java | 15 -- .../org/apache/paimon/CoreOptionsTest.java | 15 +- .../apache/paimon/append/BlobUpdateTest.java | 3 +- .../blob/PrimaryKeyBlobExternalizerTest.java | 37 ++-- .../paimon/format/blob/BlobFileFormat.java | 8 - .../paimon/format/blob/BlobFormatWriter.java | 51 +---- .../format/blob/BlobFileFormatTest.java | 15 +- .../format/blob/BlobFormatWriterTest.java | 187 +++++++----------- .../pypaimon/common/options/core_options.py | 4 +- paimon-python/pypaimon/tests/blob_test.py | 24 ++- .../pypaimon/write/blob_format_writer.py | 6 +- 13 files changed, 138 insertions(+), 268 deletions(-) diff --git a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java index c53f1e6a23c8..04ef7cbcb4b3 100644 --- a/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java +++ b/paimon-api/src/main/java/org/apache/paimon/CoreOptions.java @@ -830,9 +830,6 @@ public InlineElement getDescription() { "Whether to consider blob file size as a factor when performing scan splitting.") .build()); - // Safety ceiling for blob.copy-buffer-size (allocated as new byte[size] per writer). - public static final int MAX_BLOB_COPY_BUFFER_SIZE = 256 * 1024 * 1024; - // Keep this default in sync with BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE (= 4 * 1024). public static final ConfigOption BLOB_COPY_BUFFER_SIZE = key("blob.copy-buffer-size") @@ -3304,10 +3301,10 @@ public int blobCopyBufferSize() { /** Validates {@link #BLOB_COPY_BUFFER_SIZE} bytes and narrows to a positive int. */ public static int checkedBlobCopyBufferSize(long bytes) { checkArgument( - bytes > 0 && bytes <= MAX_BLOB_COPY_BUFFER_SIZE, + bytes > 0 && bytes <= Integer.MAX_VALUE, "'%s' must be between 1 byte and %s bytes, but was %s bytes.", BLOB_COPY_BUFFER_SIZE.key(), - MAX_BLOB_COPY_BUFFER_SIZE, + Integer.MAX_VALUE, bytes); return (int) bytes; } diff --git a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java index 8c37f28f83da..39d6c6eac833 100644 --- a/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java +++ b/paimon-core/src/main/java/org/apache/paimon/append/MultipleBlobFileWriter.java @@ -23,7 +23,6 @@ import org.apache.paimon.data.InternalRow; import org.apache.paimon.fileindex.FileIndexOptions; import org.apache.paimon.format.blob.BlobFileFormat; -import org.apache.paimon.format.blob.BlobFormatWriter; import org.apache.paimon.fs.FileIO; import org.apache.paimon.io.DataFileMeta; import org.apache.paimon.io.DataFilePathFactory; @@ -54,39 +53,6 @@ public class MultipleBlobFileWriter implements Closeable { private final List blobWriters; - public MultipleBlobFileWriter( - FileIO fileIO, - long schemaId, - RowType writeSchema, - DataFilePathFactory pathFactory, - Supplier seqNumCounterSupplier, - FileSource fileSource, - boolean asyncFileWrite, - boolean statsDenseStore, - long targetFileSize, - @Nullable BlobConsumer blobConsumer, - Set blobInlineFields, - boolean writeNullOnMissingFile, - boolean writeNullOnFetchFailure, - BlobFetchMetricReporter blobFetchMetricReporter) { - this( - fileIO, - schemaId, - writeSchema, - pathFactory, - seqNumCounterSupplier, - fileSource, - asyncFileWrite, - statsDenseStore, - targetFileSize, - blobConsumer, - blobInlineFields, - writeNullOnMissingFile, - writeNullOnFetchFailure, - blobFetchMetricReporter, - BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); - } - public MultipleBlobFileWriter( FileIO fileIO, long schemaId, diff --git a/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java b/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java index b51093d0194a..edb9ed4a0b38 100644 --- a/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java +++ b/paimon-core/src/main/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizer.java @@ -57,21 +57,6 @@ public class PrimaryKeyBlobExternalizer { private final ManagedBlobPackWriter[] packWriters; private final List uncommittedPacks; - public PrimaryKeyBlobExternalizer( - FileIO fileIO, - RowType valueType, - Set managedBlobFields, - DataFilePathFactory pathFactory, - long targetFileSize) { - this( - fileIO, - valueType, - managedBlobFields, - pathFactory, - targetFileSize, - BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); - } - public PrimaryKeyBlobExternalizer( FileIO fileIO, RowType valueType, diff --git a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java index f89f10f65d26..031dd296792f 100644 --- a/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/CoreOptionsTest.java @@ -174,17 +174,12 @@ public void testBlobCopyBufferSize() { .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("blob.copy-buffer-size"); - // the max (256 MiB) is accepted, just above it is rejected. - conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("256 mb")); - assertThat(new CoreOptions(conf).blobCopyBufferSize()) - .isEqualTo(CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE); + // There is no arbitrary memory ceiling; only the Java int-sized array limit applies. conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("512 mb")); - assertThatThrownBy(() -> new CoreOptions(conf).blobCopyBufferSize()) - .isInstanceOf(IllegalArgumentException.class) - .hasMessageContaining("blob.copy-buffer-size"); - - // way oversized is rejected instead of throwing ArithmeticException / OOM. - conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("3 gb")); + assertThat(new CoreOptions(conf).blobCopyBufferSize()).isEqualTo(512 * 1024 * 1024); + conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse(Integer.MAX_VALUE + " bytes")); + assertThat(new CoreOptions(conf).blobCopyBufferSize()).isEqualTo(Integer.MAX_VALUE); + conf.set(CoreOptions.BLOB_COPY_BUFFER_SIZE, MemorySize.parse("2 gb")); assertThatThrownBy(() -> new CoreOptions(conf).blobCopyBufferSize()) .isInstanceOf(IllegalArgumentException.class) .hasMessageContaining("blob.copy-buffer-size"); diff --git a/paimon-core/src/test/java/org/apache/paimon/append/BlobUpdateTest.java b/paimon-core/src/test/java/org/apache/paimon/append/BlobUpdateTest.java index c8c2a7f90c4b..b86b020cd199 100644 --- a/paimon-core/src/test/java/org/apache/paimon/append/BlobUpdateTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/append/BlobUpdateTest.java @@ -29,6 +29,7 @@ import org.apache.paimon.data.Timestamp; import org.apache.paimon.format.FormatWriter; import org.apache.paimon.format.blob.BlobFileFormat; +import org.apache.paimon.format.blob.BlobFormatWriter; import org.apache.paimon.fs.FileIO; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; @@ -473,7 +474,7 @@ private static DataFileMeta writeBlobFile( throws IOException { try (PositionOutputStream out = fileIO.newOutputStream(path, false)) { FormatWriter writer = - new BlobFileFormat() + new BlobFileFormat(false, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE) .createWriterFactory(RowType.of(DataTypes.BLOB())) .create(out, "none"); for (Blob blob : blobs) { diff --git a/paimon-core/src/test/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizerTest.java b/paimon-core/src/test/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizerTest.java index 5fe4686c9f75..da9fa84032e8 100644 --- a/paimon-core/src/test/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizerTest.java +++ b/paimon-core/src/test/java/org/apache/paimon/blob/PrimaryKeyBlobExternalizerTest.java @@ -25,6 +25,7 @@ import org.apache.paimon.data.GenericRow; import org.apache.paimon.data.InternalArray; import org.apache.paimon.data.InternalRow; +import org.apache.paimon.format.blob.BlobFormatWriter; import org.apache.paimon.fs.Path; import org.apache.paimon.fs.PositionOutputStream; import org.apache.paimon.fs.PositionOutputStreamWrapper; @@ -79,7 +80,7 @@ public void close() throws IOException { new DataFilePathFactory( bucketPath, "avro", "data-", "changelog-", false, null, null); PrimaryKeyBlobExternalizer externalizer = - new PrimaryKeyBlobExternalizer( + newExternalizer( fileIO, RowType.of(DataTypes.BLOB()), Collections.singleton("f0"), @@ -104,7 +105,7 @@ void testRejectsNonBlobManagedField() throws Exception { assertThatThrownBy( () -> - new PrimaryKeyBlobExternalizer( + newExternalizer( fileIO, RowType.of(DataTypes.INT()), Collections.singleton("f0"), @@ -125,7 +126,7 @@ void testRejectsUnknownManagedField() throws Exception { assertThatThrownBy( () -> - new PrimaryKeyBlobExternalizer( + newExternalizer( fileIO, RowType.of(DataTypes.BLOB()), Collections.singleton("missing"), @@ -145,8 +146,7 @@ void testExternalizeRawBlobBeforeBuffering() throws Exception { bucketPath, "avro", "data-", "changelog-", false, null, null); RowType valueType = RowType.of(DataTypes.INT(), DataTypes.BLOB()); PrimaryKeyBlobExternalizer externalizer = - new PrimaryKeyBlobExternalizer( - fileIO, valueType, Collections.singleton("f1"), pathFactory, 1024L); + newExternalizer(fileIO, valueType, Collections.singleton("f1"), pathFactory, 1024L); byte[] expected = "managed-blob".getBytes(StandardCharsets.UTF_8); InternalRow result = @@ -172,7 +172,7 @@ void testRematerializesBlobRefInput() throws Exception { new DataFilePathFactory( bucketPath, "avro", "data-", "changelog-", false, null, null); PrimaryKeyBlobExternalizer externalizer = - new PrimaryKeyBlobExternalizer( + newExternalizer( fileIO, RowType.of(DataTypes.INT(), DataTypes.BLOB()), Collections.singleton("f1"), @@ -212,7 +212,7 @@ void testExternalizesOnlyDeclaredFields() throws Exception { }, new String[] {"id", "managed", "unmanaged"}); PrimaryKeyBlobExternalizer externalizer = - new PrimaryKeyBlobExternalizer( + newExternalizer( fileIO, valueType, Collections.singleton("managed"), pathFactory, 1024L); Blob unmanaged = Blob.fromData(new byte[] {2}); @@ -233,7 +233,7 @@ void testRetractSkipsPayloadAndAbortDeletesPrivatePack() throws Exception { new DataFilePathFactory( bucketPath, "avro", "data-", "changelog-", false, null, null); PrimaryKeyBlobExternalizer externalizer = - new PrimaryKeyBlobExternalizer( + newExternalizer( fileIO, RowType.of(DataTypes.INT(), DataTypes.BLOB()), Collections.singleton("f1"), @@ -268,7 +268,7 @@ void testExternalizeBlobArrayElements() throws Exception { new DataFilePathFactory( bucketPath, "avro", "data-", "changelog-", false, null, null); PrimaryKeyBlobExternalizer externalizer = - new PrimaryKeyBlobExternalizer( + newExternalizer( fileIO, RowType.of(DataTypes.INT(), DataTypes.ARRAY(DataTypes.BLOB())), Collections.singleton("f1"), @@ -309,7 +309,7 @@ void testRematerializesBlobRefArrayElement() throws Exception { new DataFilePathFactory( bucketPath, "avro", "data-", "changelog-", false, null, null); PrimaryKeyBlobExternalizer externalizer = - new PrimaryKeyBlobExternalizer( + newExternalizer( fileIO, RowType.of(DataTypes.INT(), DataTypes.ARRAY(DataTypes.BLOB())), Collections.singleton("f1"), @@ -352,7 +352,7 @@ void testBlobArrayNullEmptyRetractAndPlaceholder() throws Exception { new DataFilePathFactory( bucketPath, "avro", "data-", "changelog-", false, null, null); PrimaryKeyBlobExternalizer externalizer = - new PrimaryKeyBlobExternalizer( + newExternalizer( fileIO, RowType.of(DataTypes.INT(), DataTypes.ARRAY(DataTypes.BLOB())), Collections.singleton("f1"), @@ -380,4 +380,19 @@ void testBlobArrayNullEmptyRetractAndPlaceholder() throws Exception { .hasMessageContaining("placeholder blob array"); assertThat(fileIO.listStatus(bucketPath)).isEmpty(); } + + private static PrimaryKeyBlobExternalizer newExternalizer( + LocalFileIO fileIO, + RowType valueType, + java.util.Set managedBlobFields, + DataFilePathFactory pathFactory, + long targetFileSize) { + return new PrimaryKeyBlobExternalizer( + fileIO, + valueType, + managedBlobFields, + pathFactory, + targetFileSize, + BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + } } diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java index e3bf4851b58e..67d0baaa4471 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFileFormat.java @@ -60,14 +60,6 @@ public class BlobFileFormat extends FileFormat { @Nullable public BlobConsumer writeConsumer; - public BlobFileFormat() { - this(false); - } - - public BlobFileFormat(boolean blobAsDescriptor) { - this(blobAsDescriptor, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); - } - public BlobFileFormat(boolean blobAsDescriptor, int copyBufferSize) { super(BlobFileFormatFactory.IDENTIFIER); this.blobAsDescriptor = blobAsDescriptor; diff --git a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java index dbec5c7a55c1..5e2bc9f9bb24 100644 --- a/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java +++ b/paimon-format/src/main/java/org/apache/paimon/format/blob/BlobFormatWriter.java @@ -18,7 +18,6 @@ package org.apache.paimon.format.blob; -import org.apache.paimon.CoreOptions; import org.apache.paimon.data.Blob; import org.apache.paimon.data.BlobArrayPlaceholder; import org.apache.paimon.data.BlobConsumer; @@ -80,51 +79,6 @@ public class BlobFormatWriter implements FileAwareFormatWriter { private String pathString; - public BlobFormatWriter( - PositionOutputStream out, @Nullable BlobConsumer writeConsumer, RowType type) { - this(out, writeConsumer, type, false, false); - } - - public BlobFormatWriter( - PositionOutputStream out, - @Nullable BlobConsumer writeConsumer, - RowType type, - boolean writeNullOnMissingFile) { - this(out, writeConsumer, type, writeNullOnMissingFile, false); - } - - public BlobFormatWriter( - PositionOutputStream out, - @Nullable BlobConsumer writeConsumer, - RowType type, - boolean writeNullOnMissingFile, - boolean writeNullOnFetchFailure) { - this( - out, - writeConsumer, - type, - writeNullOnMissingFile, - writeNullOnFetchFailure, - BlobFetchMetricReporter.NOOP); - } - - public BlobFormatWriter( - PositionOutputStream out, - @Nullable BlobConsumer writeConsumer, - RowType type, - boolean writeNullOnMissingFile, - boolean writeNullOnFetchFailure, - BlobFetchMetricReporter blobFetchMetricReporter) { - this( - out, - writeConsumer, - type, - writeNullOnMissingFile, - writeNullOnFetchFailure, - blobFetchMetricReporter, - DEFAULT_COPY_BUFFER_SIZE); - } - public BlobFormatWriter( PositionOutputStream out, @Nullable BlobConsumer writeConsumer, @@ -134,9 +88,8 @@ public BlobFormatWriter( BlobFetchMetricReporter blobFetchMetricReporter, int copyBufferSize) { checkArgument( - copyBufferSize > 0 && copyBufferSize <= CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE, - "BLOB copy buffer size must be between 1 and %s, but was %s.", - CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE, + copyBufferSize > 0, + "BLOB copy buffer size must be positive, but was %s.", copyBufferSize); this.out = out; this.writeConsumer = writeConsumer; diff --git a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java index 2b72cbb08e9f..ff5315e6684f 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFileFormatTest.java @@ -95,7 +95,8 @@ public void testReadArrayBlobInlineBytes() throws IOException { @Test public void testWriteArrayBlobPlaceholderWithProjectedRow() throws IOException { - BlobFileFormat format = new BlobFileFormat(); + BlobFileFormat format = + new BlobFileFormat(false, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); RowType rowType = RowType.of(DataTypes.ARRAY(DataTypes.BLOB())); try (PositionOutputStream out = fileIO.newOutputStream(file, false)) { @@ -162,7 +163,8 @@ public void testRejectArrayBlobElementLengthMismatch() throws IOException { } private void innerTest(boolean blobAsDescriptor) throws IOException { - BlobFileFormat format = new BlobFileFormat(blobAsDescriptor); + BlobFileFormat format = + new BlobFileFormat(blobAsDescriptor, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); RowType rowType = RowType.of(DataTypes.BLOB()); // write @@ -235,7 +237,8 @@ private void innerTest(boolean blobAsDescriptor) throws IOException { private void assertMalformedArrayPayload( ArrayPayloadCorruptor corruptor, String expectedMessage) throws IOException { - BlobFileFormat format = new BlobFileFormat(); + BlobFileFormat format = + new BlobFileFormat(false, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); RowType rowType = RowType.of(DataTypes.ARRAY(DataTypes.BLOB())); try (PositionOutputStream out = fileIO.newOutputStream(file, false)) { FormatWriter writer = format.createWriterFactory(rowType).create(out, null); @@ -272,7 +275,8 @@ private void assertMalformedArrayPayload( @Test public void testReadWithProjectedRowTypeContainingExtraFields() throws IOException { - BlobFileFormat format = new BlobFileFormat(false); + BlobFileFormat format = + new BlobFileFormat(false, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); RowType writeRowType = RowType.of(DataTypes.BLOB()); // write blob data @@ -310,7 +314,8 @@ public void testReadWithProjectedRowTypeContainingExtraFields() throws IOExcepti } private void innerTestArray(boolean blobAsDescriptor) throws IOException { - BlobFileFormat format = new BlobFileFormat(blobAsDescriptor); + BlobFileFormat format = + new BlobFileFormat(blobAsDescriptor, BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); RowType rowType = RowType.of(DataTypes.ARRAY(DataTypes.BLOB())); GenericArray first = diff --git a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java index c350a1619124..633d32e44126 100644 --- a/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java +++ b/paimon-format/src/test/java/org/apache/paimon/format/blob/BlobFormatWriterTest.java @@ -62,11 +62,7 @@ public void testTwoConsecutiveBlobsPreserveReadback(@TempDir java.nio.file.Path byte[] firstPayload = "first-blob".getBytes(); byte[] secondPayload = "second-blob-payload".getBytes(); - BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), - null, - rowType); + BlobFormatWriter writer = newWriter(outputFile, rowType); writer.addElement(GenericRow.of(Blob.fromData(firstPayload))); writer.addElement(GenericRow.of(Blob.fromData(secondPayload))); writer.close(); @@ -92,13 +88,7 @@ public void testWriteNullOnFetchFailureFallbackForHttpBadRequest( @TempDir java.nio.file.Path tempDir) throws Exception { RowType rowType = RowType.of(DataTypes.BLOB()); java.nio.file.Path outputFile = tempDir.resolve("blob.out"); - BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), - null, - rowType, - false, - true); + BlobFormatWriter writer = newWriter(outputFile, rowType, false, true); writer.addElement( GenericRow.of( @@ -116,13 +106,7 @@ public void testHttpRateLimitWritesNullWhenFetchFailureEnabled( @TempDir java.nio.file.Path tempDir) throws Exception { RowType rowType = RowType.of(DataTypes.BLOB()); java.nio.file.Path outputFile = tempDir.resolve("blob.out"); - BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), - null, - rowType, - false, - true); + BlobFormatWriter writer = newWriter(outputFile, rowType, false, true); writer.addElement( GenericRow.of( @@ -139,14 +123,7 @@ public void testHttpRateLimitWritesNullWhenFetchFailureEnabled( public void testHttpRateLimitFailsWhenFetchFailureDisabled(@TempDir java.nio.file.Path tempDir) throws Exception { RowType rowType = RowType.of(DataTypes.BLOB()); - BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream( - tempDir.resolve("blob.out").toFile()), - null, - rowType, - false, - false); + BlobFormatWriter writer = newWriter(tempDir.resolve("blob.out"), rowType, false, false); assertThatThrownBy( () -> @@ -166,14 +143,7 @@ public void testHttpRateLimitFailsWhenFetchFailureDisabled(@TempDir java.nio.fil public void testHttpNotFoundPropagatesWhenFetchFailureDisabled( @TempDir java.nio.file.Path tempDir) throws Exception { RowType rowType = RowType.of(DataTypes.BLOB()); - BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream( - tempDir.resolve("blob.out").toFile()), - null, - rowType, - false, - false); + BlobFormatWriter writer = newWriter(tempDir.resolve("blob.out"), rowType, false, false); assertThatThrownBy( () -> @@ -200,13 +170,7 @@ public void testWriteNullOnFetchFailureForInvalidUriDescriptor( new BlobDescriptor("https://img.alicdn.com/imgextra/##1304008055350781673", 0, -1) .serialize(); - BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), - null, - rowType, - false, - true); + BlobFormatWriter writer = newWriter(outputFile, rowType, false, true); writer.addElement(new DescriptorBytesRow(descriptorBytes, uriReaderFactory)); writer.close(); @@ -232,13 +196,7 @@ public void testArrayWriteNullOnFetchFailureForInvalidUriDescriptor( new BlobDescriptor("https://img.alicdn.com/imgextra/##1304008055350781673", 0, -1) .serialize(); - BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), - null, - rowType, - false, - true); + BlobFormatWriter writer = newWriter(outputFile, rowType, false, true); writer.addElement( GenericRow.of(new DescriptorBytesArray(descriptorBytes, uriReaderFactory))); @@ -273,13 +231,7 @@ public void testHttpNotFoundWritesNullWhenMissingFileEnabled( @TempDir java.nio.file.Path tempDir) throws Exception { RowType rowType = RowType.of(DataTypes.BLOB()); java.nio.file.Path outputFile = tempDir.resolve("blob.out"); - BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), - null, - rowType, - true, - false); + BlobFormatWriter writer = newWriter(outputFile, rowType, true, false); writer.addElement( GenericRow.of( @@ -304,14 +256,7 @@ public void testBlobFetchMetricReporterForSuccessAndNullWritten( RowType rowType = RowType.of(DataTypes.BLOB()); TestingBlobFetchMetricReporter metricReporter = new TestingBlobFetchMetricReporter(); BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream( - tempDir.resolve("blob.out").toFile()), - null, - rowType, - false, - true, - metricReporter); + newWriter(tempDir.resolve("blob.out"), rowType, false, true, metricReporter); writer.addElement(GenericRow.of(Blob.fromData("image".getBytes()))); writer.addElement( @@ -333,14 +278,7 @@ public void testBlobFetchMetricReporterForUnhandledFailure(@TempDir java.nio.fil RowType rowType = RowType.of(DataTypes.BLOB()); TestingBlobFetchMetricReporter metricReporter = new TestingBlobFetchMetricReporter(); BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream( - tempDir.resolve("blob.out").toFile()), - null, - rowType, - false, - false, - metricReporter); + newWriter(tempDir.resolve("blob.out"), rowType, false, false, metricReporter); assertThatThrownBy( () -> @@ -368,14 +306,7 @@ public void testBlobFetchMetricReporterForPreCheckedMissingFile( new BlobDescriptor("https://example.com/missing.jpg", 0, -1).serialize(); TestingBlobFetchMetricReporter metricReporter = new TestingBlobFetchMetricReporter(); BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream( - tempDir.resolve("blob.out").toFile()), - null, - rowType, - true, - false, - metricReporter); + newWriter(tempDir.resolve("blob.out"), rowType, true, false, metricReporter); writer.addElement(new DescriptorBytesRow(descriptorBytes, uriReaderFactory, true)); writer.close(); @@ -390,14 +321,7 @@ public void testBlobFetchMetricReporterIgnoresUserNull(@TempDir java.nio.file.Pa RowType rowType = RowType.of(DataTypes.BLOB()); TestingBlobFetchMetricReporter metricReporter = new TestingBlobFetchMetricReporter(); BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream( - tempDir.resolve("blob.out").toFile()), - null, - rowType, - true, - false, - metricReporter); + newWriter(tempDir.resolve("blob.out"), rowType, true, false, metricReporter); writer.addElement(GenericRow.of((Object) null)); writer.close(); @@ -412,14 +336,7 @@ public void testArrayBlobFetchMetricReporter(@TempDir java.nio.file.Path tempDir RowType rowType = RowType.of(DataTypes.ARRAY(DataTypes.BLOB())); TestingBlobFetchMetricReporter metricReporter = new TestingBlobFetchMetricReporter(); BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream( - tempDir.resolve("blob.out").toFile()), - null, - rowType, - true, - true, - metricReporter); + newWriter(tempDir.resolve("blob.out"), rowType, true, true, metricReporter); writer.addElement( GenericRow.of( @@ -453,14 +370,7 @@ public void testArrayBlobFetchMetricReporterForUnhandledFailure( RowType rowType = RowType.of(DataTypes.ARRAY(DataTypes.BLOB())); TestingBlobFetchMetricReporter metricReporter = new TestingBlobFetchMetricReporter(); BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream( - tempDir.resolve("blob.out").toFile()), - null, - rowType, - false, - false, - metricReporter); + newWriter(tempDir.resolve("blob.out"), rowType, false, false, metricReporter); assertThatThrownBy( () -> @@ -501,7 +411,7 @@ public void testCopyBufferSizeIsRespectedForBlobRef(@TempDir java.nio.file.Path @Test public void testDefaultCopyBufferSize(@TempDir java.nio.file.Path tempDir) throws Exception { - // default preserves the historical 4 KiB copy buffer. + // The configured default preserves the historical 4 KiB copy buffer. assertThat(BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE).isEqualTo(4 * 1024); String uri = "mem://file"; @@ -509,12 +419,7 @@ public void testDefaultCopyBufferSize(@TempDir java.nio.file.Path tempDir) throw RecordingUriReader reader = new RecordingUriReader(singleFile(uri, source)); java.nio.file.Path outputFile = tempDir.resolve("blob.out"); - // default constructor -> default 4 KiB copy buffer. - BlobFormatWriter writer = - new BlobFormatWriter( - new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), - null, - RowType.of(DataTypes.BLOB())); + BlobFormatWriter writer = newWriter(outputFile, RowType.of(DataTypes.BLOB())); writer.addElement(GenericRow.of(new BlobRef(reader, new BlobDescriptor(uri, 0, 5000)))); writer.close(); @@ -522,16 +427,70 @@ public void testDefaultCopyBufferSize(@TempDir java.nio.file.Path tempDir) throw assertThat(readBackBlobs(outputFile, 1)).containsExactly(source); } + private static BlobFormatWriter newWriter(java.nio.file.Path outputFile, RowType rowType) + throws java.io.FileNotFoundException { + return newWriter( + outputFile, + rowType, + false, + false, + BlobFetchMetricReporter.NOOP, + BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + } + + private static BlobFormatWriter newWriter( + java.nio.file.Path outputFile, + RowType rowType, + boolean writeNullOnMissingFile, + boolean writeNullOnFetchFailure) + throws java.io.FileNotFoundException { + return newWriter( + outputFile, + rowType, + writeNullOnMissingFile, + writeNullOnFetchFailure, + BlobFetchMetricReporter.NOOP, + BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + } + + private static BlobFormatWriter newWriter( + java.nio.file.Path outputFile, + RowType rowType, + boolean writeNullOnMissingFile, + boolean writeNullOnFetchFailure, + BlobFetchMetricReporter blobFetchMetricReporter) + throws java.io.FileNotFoundException { + return newWriter( + outputFile, + rowType, + writeNullOnMissingFile, + writeNullOnFetchFailure, + blobFetchMetricReporter, + BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE); + } + private static BlobFormatWriter newWriter( java.nio.file.Path outputFile, RowType rowType, int copyBufferSize) throws java.io.FileNotFoundException { + return newWriter( + outputFile, rowType, false, false, BlobFetchMetricReporter.NOOP, copyBufferSize); + } + + private static BlobFormatWriter newWriter( + java.nio.file.Path outputFile, + RowType rowType, + boolean writeNullOnMissingFile, + boolean writeNullOnFetchFailure, + BlobFetchMetricReporter blobFetchMetricReporter, + int copyBufferSize) + throws java.io.FileNotFoundException { return new BlobFormatWriter( new LocalFileIO.LocalPositionOutputStream(outputFile.toFile()), null, rowType, - false, - false, - BlobFetchMetricReporter.NOOP, + writeNullOnMissingFile, + writeNullOnFetchFailure, + blobFetchMetricReporter, copyBufferSize); } diff --git a/paimon-python/pypaimon/common/options/core_options.py b/paimon-python/pypaimon/common/options/core_options.py index 4595e39de98f..5af7a88a8f71 100644 --- a/paimon-python/pypaimon/common/options/core_options.py +++ b/paimon-python/pypaimon/common/options/core_options.py @@ -1120,8 +1120,8 @@ def blob_target_file_size(self, default=None): def blob_copy_buffer_size(self): size = self.options.get(CoreOptions.BLOB_COPY_BUFFER_SIZE, None).get_bytes() - # Upper bound matches Java CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE (256 MiB). - max_size = 256 * 1024 * 1024 + # Java BlobFormatWriter stores the byte-array size in an int. + max_size = (1 << 31) - 1 if not 1 <= size <= max_size: raise ValueError( f"'{CoreOptions.BLOB_COPY_BUFFER_SIZE.key()}' must be between 1 byte and " diff --git a/paimon-python/pypaimon/tests/blob_test.py b/paimon-python/pypaimon/tests/blob_test.py index b574a24079c8..95e02047c691 100644 --- a/paimon-python/pypaimon/tests/blob_test.py +++ b/paimon-python/pypaimon/tests/blob_test.py @@ -131,28 +131,32 @@ def tearDown(self): pass # Ignore cleanup errors def test_blob_copy_buffer_size_validation(self): - """blob.copy-buffer-size must be within 1 .. 256 MiB, matching Java.""" + """blob.copy-buffer-size must be positive and fit Java's int-sized array.""" from pypaimon.write.blob_format_writer import BlobFormatWriter from pypaimon.common.options.core_options import CoreOptions - max_size = 256 * 1024 * 1024 - # BlobFormatWriter rejects non-positive and oversized (> 256 MiB) buffers. - for bad in [0, -1, max_size + 1]: + # The internal writer already receives an int-sized value and only rejects non-positive + # buffers. + for bad in [0, -1]: with self.assertRaises(ValueError): BlobFormatWriter(io.BytesIO(), copy_buffer_size=bad) - # valid custom buffers (including the max) are accepted. BlobFormatWriter(io.BytesIO(), copy_buffer_size=8).close() - BlobFormatWriter(io.BytesIO(), copy_buffer_size=max_size).close() - # CoreOptions accessor defaults to 4 KiB and validates the same bounds. + # CoreOptions defaults to 4 KiB, accepts values above the old 256 MiB ceiling, + # and retains only the Java int technical limit for cross-language consistency. self.assertEqual(CoreOptions(Options({})).blob_copy_buffer_size(), 4096) self.assertEqual( CoreOptions(Options({'blob.copy-buffer-size': '256 kb'})).blob_copy_buffer_size(), 256 * 1024) self.assertEqual( - CoreOptions(Options({'blob.copy-buffer-size': '256 mb'})).blob_copy_buffer_size(), - max_size) - for bad in ['0 bytes', '512 mb', '3 gb']: + CoreOptions(Options({'blob.copy-buffer-size': '512 mb'})).blob_copy_buffer_size(), + 512 * 1024 * 1024) + self.assertEqual( + CoreOptions(Options({ + 'blob.copy-buffer-size': f'{(1 << 31) - 1} bytes' + })).blob_copy_buffer_size(), + (1 << 31) - 1) + for bad in ['0 bytes', '2 gb', '3 gb']: with self.assertRaises(ValueError): CoreOptions(Options({'blob.copy-buffer-size': bad})).blob_copy_buffer_size() diff --git a/paimon-python/pypaimon/write/blob_format_writer.py b/paimon-python/pypaimon/write/blob_format_writer.py index b64cd19cf7d5..6efdf9e49fa2 100644 --- a/paimon-python/pypaimon/write/blob_format_writer.py +++ b/paimon-python/pypaimon/write/blob_format_writer.py @@ -37,17 +37,15 @@ class BlobFormatWriter: PLACE_HOLDER_LENGTH = -2 ARRAY_NULL_ELEMENT_LENGTH = -1 BUFFER_SIZE = 4096 - MAX_BUFFER_SIZE = 256 * 1024 * 1024 # matches Java CoreOptions.MAX_BLOB_COPY_BUFFER_SIZE METADATA_SIZE = 12 # 8-byte length + 4-byte CRC def __init__(self, output_stream: BinaryIO, blob_consumer: Optional[BlobConsumer] = None, file_path: Optional[str] = None, copy_buffer_size: int = BUFFER_SIZE): - if not 1 <= copy_buffer_size <= self.MAX_BUFFER_SIZE: + if copy_buffer_size <= 0: raise ValueError( - f"BLOB copy buffer size must be between 1 and {self.MAX_BUFFER_SIZE}, " - f"but was {copy_buffer_size}.") + f"BLOB copy buffer size must be positive, but was {copy_buffer_size}.") self.output_stream = output_stream self._blob_consumer = blob_consumer self._file_path = file_path