This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 6ef05fe2f1 [core] Bypass append write buffer for BLOB payload writers
(#10535)
6ef05fe2f1 is described below
commit 6ef05fe2f1c36679317a95a5ebb701e822e35baf
Author: zhoulii <[email protected]>
AuthorDate: Wed Oct 14 09:57:32 2026 +0800
[core] Bypass append write buffer for BLOB payload writers (#10535)
---
docs/docs/multimodal-table/blob.mdx | 2 ++
.../paimon/operation/BaseAppendFileStoreWrite.java | 8 +++++-
.../org/apache/paimon/flink/BlobTableITCase.java | 30 ++++++++++++++++++++++
3 files changed, 39 insertions(+), 1 deletion(-)
diff --git a/docs/docs/multimodal-table/blob.mdx
b/docs/docs/multimodal-table/blob.mdx
index 315ae663cb..1809bb95fa 100644
--- a/docs/docs/multimodal-table/blob.mdx
+++ b/docs/docs/multimodal-table/blob.mdx
@@ -537,6 +537,8 @@ for Python partial writes.
4. **Append Table Options**: For append tables, `row-tracking.enabled` and
`data-evolution.enabled` must be set to `true`.
5. **Blob View Dependency**: Blob view fields depend on the referenced
upstream table and row. If the upstream data is removed or no longer readable,
the view cannot be resolved.
6. **Collection Storage Mode**: `ARRAY<BLOB>` and `MAP<K, BLOB>` are supported
only for raw managed blob storage (`__BLOB_FIELD`), not for
`__BLOB_DESCRIPTOR_FIELD` or `__BLOB_VIEW_FIELD`.
+7. **Append Write Buffer**: Writers that write BLOB payload files ignore
`write-buffer-for-append`
+ and write directly, because the buffer cannot preserve descriptor source
readers.
## Best Practices
diff --git
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
index 07d15637fe..61c2cef4c8 100644
---
a/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
+++
b/paimon-core/src/main/java/org/apache/paimon/operation/BaseAppendFileStoreWrite.java
@@ -164,6 +164,12 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
boolean ignorePreviousFiles) {
DataFilePathFactory dataPathFactory =
pathFactory.createDataFilePathFactory(partition, bucket);
+ boolean useWriteBuffer = options.useWriteBufferForAppend() ||
forceBufferSpill;
+ if (blobContext != null
+ && !fieldNamesInBlobFile(writeType,
blobContext.blobInlineFields()).isEmpty()) {
+ // Buffer serialization drops source readers needed to copy
descriptor-backed BLOBs.
+ useWriteBuffer = false;
+ }
return new AppendOnlyWriter(
fileIO,
ioManager,
@@ -183,7 +189,7 @@ public abstract class BaseAppendFileStoreWrite extends
MemoryFileStoreWrite<Inte
options.commitForceCompact(),
dataPathFactory,
restoreIncrement,
- options.useWriteBufferForAppend() || forceBufferSpill,
+ useWriteBuffer,
options.writeBufferSpillable() || forceBufferSpill,
options.fileCompression(),
options.spillCompressOptions(),
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
index 839483f970..abbfb34079 100644
---
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java
@@ -48,6 +48,7 @@ import org.junit.jupiter.params.provider.ValueSource;
import java.io.OutputStream;
import java.net.URI;
+import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Arrays;
import java.util.LinkedHashMap;
@@ -667,6 +668,35 @@ public class BlobTableITCase extends CatalogITCaseBase {
.containsExactlyInAnyOrder(Row.of(1, "paimon", blobData));
}
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ public void testWriteBlobDescriptorWithAppendBuffer(boolean streaming)
throws Exception {
+ byte[] payload = new byte[] {1, 2, 3};
+ Path source = warehouse.resolve("buffered_blob_source");
+ Files.write(source, payload);
+ tEnv.executeSql(
+ "CREATE TABLE buffered_blob_table (id INT, picture BYTES)"
+ + " WITH ('row-tracking.enabled'='true',"
+ + " 'data-evolution.enabled'='true',"
+ + " 'blob-field'='picture',"
+ + " 'write-buffer-for-append'='true',"
+ + " 'write-buffer-spillable'='false',"
+ + " 'blob-write-null-on-fetch-failure'='true')");
+
+ (streaming ? sEnv : tEnv)
+ .executeSql(
+ "INSERT INTO buffered_blob_table "
+ + "/*+ OPTIONS('sink.parallelism'='2') */ "
+ + "VALUES (1, sys.path_to_descriptor('"
+ + source.toUri()
+ + "'))")
+ .await();
+
+ Files.delete(source);
+ assertThat(batchSql("SELECT * FROM buffered_blob_table"))
+ .containsExactly(Row.of(1, payload));
+ }
+
@Test
public void testDescriptorToPresignedUrlBuiltInFunctions() {
assertThat(batchSql("SHOW FUNCTIONS IN sys"))
