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 b18c14d647 [common][flink] Fix BlobDescriptor lost during shuffle
serialization (#8494)
b18c14d647 is described below
commit b18c14d6471c71d18404eb8491a12650969ef0c4
Author: liming.1018 <[email protected]>
AuthorDate: Tue Jul 7 20:48:29 2026 +0800
[common][flink] Fix BlobDescriptor lost during shuffle serialization (#8494)
---
.../apache/paimon/data/AbstractBinaryWriter.java | 7 +-
.../paimon/data/serializer/BlobSerializer.java | 10 ++-
.../org/apache/paimon/flink/BlobTableITCase.java | 83 ++++++++++++++++++++++
3 files changed, 97 insertions(+), 3 deletions(-)
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/AbstractBinaryWriter.java
b/paimon-common/src/main/java/org/apache/paimon/data/AbstractBinaryWriter.java
index bb632d5b31..bbbab73dd6 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/AbstractBinaryWriter.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/AbstractBinaryWriter.java
@@ -203,7 +203,12 @@ abstract class AbstractBinaryWriter implements
BinaryWriter {
@Override
public void writeBlob(int pos, Blob blob) {
- byte[] bytes = blob.toData();
+ byte[] bytes;
+ if (blob instanceof BlobData) {
+ bytes = blob.toData();
+ } else {
+ bytes = Blob.serializeBlob(blob);
+ }
writeBinary(pos, bytes, 0, bytes.length);
}
diff --git
a/paimon-common/src/main/java/org/apache/paimon/data/serializer/BlobSerializer.java
b/paimon-common/src/main/java/org/apache/paimon/data/serializer/BlobSerializer.java
index 28f9233a37..55f6c8667b 100644
---
a/paimon-common/src/main/java/org/apache/paimon/data/serializer/BlobSerializer.java
+++
b/paimon-common/src/main/java/org/apache/paimon/data/serializer/BlobSerializer.java
@@ -39,12 +39,18 @@ public class BlobSerializer extends
SerializerSingleton<Blob> {
@Override
public void serialize(Blob blob, DataOutputView target) throws IOException
{
- BinarySerializer.INSTANCE.serialize(blob.toData(), target);
+ byte[] bytes;
+ if (blob instanceof BlobData) {
+ bytes = blob.toData();
+ } else {
+ bytes = Blob.serializeBlob(blob);
+ }
+ BinarySerializer.INSTANCE.serialize(bytes, target);
}
@Override
public Blob deserialize(DataInputView source) throws IOException {
byte[] bytes = BinarySerializer.INSTANCE.deserialize(source);
- return new BlobData(bytes);
+ return Blob.fromBytes(bytes, null, null);
}
}
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 d09ccedb16..280535d6e3 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
@@ -724,6 +724,89 @@ public class BlobTableITCase extends CatalogITCaseBase {
assertThat(AbstractFlinkTableFactory.schemaEquals(convertedRowType,
flinkRowType)).isTrue();
}
+ @Test
+ public void testBlobDescriptorFieldStreamingWriteWithShuffle() throws
Exception {
+ // Verifies that blob-descriptor-field works when shuffle
serialization occurs
+ // in streaming mode. When source and sink have different parallelism,
Flink
+ // inserts a REBALANCE that serializes InternalRow to BinaryRow via
+ // InternalRowTypeSerializer. writeBlob() must preserve the
BlobDescriptor so
+ // that BlobDescriptorWriter can serialize it at the sink.
+
+ byte[] blobData = "multimodal-content".getBytes();
+ FileIO fileIO = new LocalFileIO();
+ String uri = "file://" + warehouse + "/multimodal_blob_stream";
+ try (OutputStream outputStream =
+ fileIO.newOutputStream(new org.apache.paimon.fs.Path(uri),
true)) {
+ outputStream.write(blobData);
+ }
+
+ sEnv.executeSql(
+ "CREATE TABLE multimodal_assets_metadata ("
+ + "id STRING, "
+ + "source_uri STRING, "
+ + "media_content BYTES, "
+ + "dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + "'row-tracking.enabled'='true',"
+ + "'data-evolution.enabled'='true',"
+ + "'blob-field'='media_content',"
+ + "'blob-descriptor-field'='media_content'"
+ + ")");
+
+ // Streaming INSERT with sys.path_to_descriptor
+ // The /*+ OPTIONS('sink.parallelism' = '2') */ hint forces a shuffle
+ // from the source (parallelism 1 for VALUES) to the sink (parallelism
2).
+ // After fix, writeBlob() preserves BlobDescriptor through the shuffle.
+ sEnv.getConfig().set("table.dml-sync", "true");
+ sEnv.executeSql(
+ "INSERT INTO multimodal_assets_metadata "
+ + "/*+ OPTIONS('sink.parallelism' = '2') */ "
+ + "VALUES ('test-id', 'test-uri', "
+ + "sys.path_to_descriptor('"
+ + uri
+ + "'), '2025-07-07')")
+ .await();
+ }
+
+ @Test
+ public void testBlobDescriptorFieldBatchWriteWithShuffle() throws
Exception {
+ // Same as streaming test but in batch mode.
+ // Verifies that blob-descriptor-field works when shuffle serialization
+ // occurs in batch mode.
+
+ byte[] blobData = "batch-multimodal-content".getBytes();
+ FileIO fileIO = new LocalFileIO();
+ String uri = "file://" + warehouse + "/multimodal_blob_batch";
+ try (OutputStream outputStream =
+ fileIO.newOutputStream(new org.apache.paimon.fs.Path(uri),
true)) {
+ outputStream.write(blobData);
+ }
+
+ tEnv.executeSql(
+ "CREATE TABLE batch_multimodal_metadata ("
+ + "id STRING, "
+ + "source_uri STRING, "
+ + "media_content BYTES, "
+ + "dt STRING"
+ + ") PARTITIONED BY (dt) WITH ("
+ + "'row-tracking.enabled'='true',"
+ + "'data-evolution.enabled'='true',"
+ + "'blob-field'='media_content',"
+ + "'blob-descriptor-field'='media_content'"
+ + ")");
+
+ // Batch INSERT with sys.path_to_descriptor
+ // The /*+ OPTIONS('sink.parallelism' = '2') */ hint forces a shuffle,
+ // same as streaming mode.
+ batchSql(
+ "INSERT INTO batch_multimodal_metadata "
+ + "/*+ OPTIONS('sink.parallelism' = '2') */ "
+ + "VALUES ('test-id', 'test-uri', "
+ + "sys.path_to_descriptor('"
+ + uri
+ + "'), '2025-07-07')");
+ }
+
private static final char[] HEX_ARRAY = "0123456789ABCDEF".toCharArray();
public static String bytesToHex(byte[] bytes) {
