[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,14 @@ 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<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 +3294,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 <= 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() {
         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

@@ -67,11 +67,12 @@ public MultipleBlobFileWriter(
             Set<String> 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);

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;
@@ -61,7 +62,8 @@ public PrimaryKeyBlobExternalizer(
             RowType valueType,
             Set<String> managedBlobFields,
             DataFilePathFactory pathFactory,
-            long targetFileSize) {
+            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 +99,8 @@ public PrimaryKeyBlobExternalizer(
                                     : new RowType(Collections.singletonList(field)),
                             pathFactory,
                             targetFileSize,
-                            uncommittedPacks));
+                            uncommittedPacks,
+                            copyBufferSize));
         }
         checkArgument(
                 unknownFields.isEmpty(),
@@ -223,6 +226,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 +238,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 +277,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,30 @@ 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");
+
+        // 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"));
+        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");
+    }
 }

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

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<String> managedBlobFields,
+            DataFilePathFactory pathFactory,
+            long targetFileSize) {
+        return new PrimaryKeyBlobExternalizer(
+                fileIO,
+                valueType,
+                managedBlobFields,
+                pathFactory,
+                targetFileSize,
+                BlobFormatWriter.DEFAULT_COPY_BUFFER_SIZE);
+    }
 }