[core] Support configurable BLOB copy buffer size

docs/generated/core_configuration.html

@@ -104,6 +104,12 @@
             <td>Boolean</td>
             <td>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.</td>
         </tr>
+        <tr>
+            <td><h5>blob.copy-buffer-size</h5></td>
+            <td style="word-wrap: break-word;">4 kb</td>
+            <td>MemorySize</td>
+            <td>Buffer size used when copying BLOB payloads into BLOB files.</td>
+        </tr>
         <tr>
             <td><h5>blob.split-by-file-size</h5></td>
             <td style="word-wrap: break-word;">(none)</td>

paimon-api/src/main/java/org/apache/paimon/CoreOptions.java

@@ -830,6 +830,17 @@ 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<MemorySize> 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<Integer> NUM_SORTED_RUNS_COMPACTION_TRIGGER =
             key("num-sorted-run.compaction-trigger")
                     .intType()
@@ -3286,6 +3297,21 @@ public long blobTargetFileSize() {
                 .orElse(targetFileSize(false));
     }
 
+    public int blobCopyBufferSize() {
+        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 <= MAX_BLOB_COPY_BUFFER_SIZE,
+                "'%s' must be between 1 byte and %s bytes, but was %s bytes.",
+                BLOB_COPY_BUFFER_SIZE.key(),
+                MAX_BLOB_COPY_BUFFER_SIZE,
+                bytes);
+        return (int) bytes;
+    }
+
     public boolean blobSplitByFileSize() {
         return options.getOptional(BLOB_SPLIT_BY_FILE_SIZE)
                 .orElse(!options.get(BLOB_AS_DESCRIPTOR));

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;
         }

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;
@@ -68,10 +69,44 @@ public MultipleBlobFileWriter(
             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,
+            RowType writeSchema,
+            DataFilePathFactory pathFactory,
+            Supplier<LongCounter> seqNumCounterSupplier,
+            FileSource fileSource,
+            boolean asyncFileWrite,
+            boolean statsDenseStore,
+            long targetFileSize,
+            @Nullable BlobConsumer blobConsumer,
+            Set<String> blobInlineFields,
+            boolean writeNullOnMissingFile,
+            boolean writeNullOnFetchFailure,
+            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);

paimon-core/src/main/java/org/apache/paimon/append/dataevolution/DataEvolutionBlobCompactTask.java

@@ -128,7 +128,7 @@ private FileWriter<InternalRow, DataFileMeta> 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(

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<String> managedBlobFields,
             DataFilePathFactory pathFactory,
             long targetFileSize) {
+        this(
+                fileIO,
+                valueType,
+                managedBlobFields,
+                pathFactory,
+                targetFileSize,
+                BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE);
+    }
+
+    public PrimaryKeyBlobExternalizer(
+            FileIO fileIO,
+            RowType valueType,
+            Set<String> 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<Path> uncommittedPacks;
+        private final int copyBufferSize;
 
         private Path currentPath;
         private PositionOutputStream out;
@@ -234,12 +253,14 @@ private ManagedBlobPackWriter(
                 RowType blobType,
                 DataFilePathFactory pathFactory,
                 long targetFileSize,
-                List<Path> uncommittedPacks) {
+                List<Path> 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);
         }
 

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() {

paimon-core/src/main/java/org/apache/paimon/operation/BlobFileContext.java

@@ -36,6 +36,7 @@ public class BlobFileContext {
     private final Set<String> 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<String> blobDescriptorFields,
             Set<String> 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;
     }

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,35 @@ 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");
+
+        // 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)
+                .hasMessageContaining("blob.copy-buffer-size");
+    }
 }

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);
         }
     }
 

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 =
+                CoreOptions.checkedBlobCopyBufferSize(
+                        formatContext.options().get(CoreOptions.BLOB_COPY_BUFFER_SIZE).getBytes());
+        return new BlobFileFormat(blobAsDescriptor, copyBufferSize);
     }
 }

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;
@@ -64,6 +65,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 +115,29 @@ 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 && 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;
@@ -125,7 +150,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);
     }