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 c6964a7a7a [flink] Support Blob tables in dedicated split generation
(#8914)
c6964a7a7a is described below
commit c6964a7a7acc952370b86b8f037aafe6ff1096bf
Author: wangwj <[email protected]>
AuthorDate: Thu Jul 30 12:03:37 2026 +0800
[flink] Support Blob tables in dedicated split generation (#8914)
---
.../paimon/flink/source/FlinkSourceBuilder.java | 7 +-
.../flink/source/operator/MonitorSource.java | 40 ++++++-
.../paimon/flink/source/operator/ReadOperator.java | 32 ++++-
.../org/apache/paimon/flink/BlobTableITCase.java | 36 ++++++
.../source/operator/ReadOperatorBlobTest.java | 129 +++++++++++++++++++++
5 files changed, 240 insertions(+), 4 deletions(-)
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
index a49ba7ae41..dca0f3b119 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java
@@ -475,12 +475,13 @@ public class FlinkSourceBuilder {
throw new IllegalArgumentException(
"Cannot limit streaming source, please use batch execution
mode.");
}
+ ReadBuilder readBuilder = createReadBuilder(projectedRowType());
dataStream =
MonitorSource.buildSource(
env,
sourceName,
produceTypeInfo(),
- createReadBuilder(projectedRowType()),
+ readBuilder,
conf.get(CoreOptions.CONTINUOUS_DISCOVERY_INTERVAL).toMillis(),
watermarkStrategy == null,
conf.get(FlinkConnectorOptions.READ_SHUFFLE_BUCKET_WITH_PARTITION),
@@ -488,7 +489,9 @@ public class FlinkSourceBuilder {
outerProject(),
isBounded,
limit,
- table);
+ table,
+ readBuilder.readType(),
+ conf.get(CoreOptions.BLOB_AS_DESCRIPTOR));
if (parallelism != null) {
dataStream.getTransformation().setParallelism(parallelism);
}
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
index b6de64472b..3fbec3697b 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/MonitorSource.java
@@ -34,6 +34,7 @@ import org.apache.paimon.table.source.ReadBuilder;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.StreamTableScan;
import org.apache.paimon.table.source.TableScan;
+import org.apache.paimon.types.RowType;
import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
@@ -274,6 +275,38 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
boolean isBounded,
@Nullable Long limit,
@Nullable Table table) {
+ return buildSource(
+ env,
+ name,
+ typeInfo,
+ readBuilder,
+ monitorInterval,
+ emitSnapshotWatermark,
+ shuffleBucketWithPartition,
+ unordered,
+ nestedProjectedRowData,
+ isBounded,
+ limit,
+ table,
+ readBuilder.readType(),
+ false);
+ }
+
+ public static DataStream<RowData> buildSource(
+ StreamExecutionEnvironment env,
+ String name,
+ TypeInformation<RowData> typeInfo,
+ ReadBuilder readBuilder,
+ long monitorInterval,
+ boolean emitSnapshotWatermark,
+ boolean shuffleBucketWithPartition,
+ boolean unordered,
+ NestedProjectedRowData nestedProjectedRowData,
+ boolean isBounded,
+ @Nullable Long limit,
+ @Nullable Table table,
+ RowType readType,
+ boolean blobAsDescriptor) {
MonitorSource monitorSource =
new MonitorSource(readBuilder, monitorInterval,
emitSnapshotWatermark, isBounded);
Source<Split, SimpleSourceSplit, NoOpEnumState> source = monitorSource;
@@ -296,7 +329,12 @@ public class MonitorSource extends
AbstractNonCoordinatedSource<Split> {
return sourceDataStream.transform(
name + "-Reader",
typeInfo,
- new ReadOperator(readBuilder::newRead, nestedProjectedRowData,
limit));
+ new ReadOperator(
+ readBuilder::newRead,
+ nestedProjectedRowData,
+ limit,
+ readType,
+ blobAsDescriptor));
}
private static DataStream<Split> shuffleUnordered(
diff --git
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
index 30b2ab0d62..95bf5606e7 100644
---
a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
+++
b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java
@@ -21,12 +21,15 @@ package org.apache.paimon.flink.source.operator;
import org.apache.paimon.data.InternalRow;
import org.apache.paimon.disk.IOManager;
import org.apache.paimon.flink.FlinkRowData;
+import org.apache.paimon.flink.FlinkRowDataWithBlob;
import org.apache.paimon.flink.NestedProjectedRowData;
import org.apache.paimon.flink.source.RecordLimiter;
import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
import org.apache.paimon.table.source.DataSplit;
import org.apache.paimon.table.source.Split;
import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.BlobType;
+import org.apache.paimon.types.RowType;
import org.apache.paimon.utils.CloseableIterator;
import org.apache.paimon.utils.SerializableSupplier;
@@ -42,6 +45,9 @@ import org.slf4j.LoggerFactory;
import javax.annotation.Nullable;
+import java.util.HashSet;
+import java.util.Set;
+
/**
* The operator that reads the {@link Split splits} received from the
preceding {@link
* MonitorSource}. Contrary to the {@link MonitorSource} which has a
parallelism of 1, this operator
@@ -56,6 +62,8 @@ public class ReadOperator extends
AbstractStreamOperator<RowData>
private final SerializableSupplier<TableRead> readSupplier;
@Nullable private final NestedProjectedRowData nestedProjectedRowData;
+ @Nullable private final RowType readType;
+ private final boolean blobAsDescriptor;
private transient TableRead read;
private transient StreamRecord<RowData> reuseRecord;
@@ -76,9 +84,20 @@ public class ReadOperator extends
AbstractStreamOperator<RowData>
SerializableSupplier<TableRead> readSupplier,
@Nullable NestedProjectedRowData nestedProjectedRowData,
@Nullable Long limit) {
+ this(readSupplier, nestedProjectedRowData, limit, null, false);
+ }
+
+ public ReadOperator(
+ SerializableSupplier<TableRead> readSupplier,
+ @Nullable NestedProjectedRowData nestedProjectedRowData,
+ @Nullable Long limit,
+ @Nullable RowType readType,
+ boolean blobAsDescriptor) {
this.readSupplier = readSupplier;
this.nestedProjectedRowData = nestedProjectedRowData;
this.limit = limit;
+ this.readType = readType;
+ this.blobAsDescriptor = blobAsDescriptor;
}
@Override
@@ -101,7 +120,18 @@ public class ReadOperator extends
AbstractStreamOperator<RowData>
.getSpillingDirectoriesPaths());
this.read = readSupplier.get().withIOManager(ioManager);
this.recordLimiter = RecordLimiter.create(limit);
- this.reuseRow = new FlinkRowData(null);
+ Set<Integer> blobFields = new HashSet<>();
+ if (readType != null) {
+ for (int i = 0; i < readType.getFieldCount(); i++) {
+ if (BlobType.isBlobFileField(readType.getTypeAt(i))) {
+ blobFields.add(i);
+ }
+ }
+ }
+ this.reuseRow =
+ blobFields.isEmpty()
+ ? new FlinkRowData(null)
+ : new FlinkRowDataWithBlob(null, blobFields,
blobAsDescriptor);
this.reuseRecord = new StreamRecord<>(null);
this.idlingStarted();
}
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 99b785f641..d9e894646e 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
@@ -83,6 +83,42 @@ public class BlobTableITCase extends CatalogITCaseBase {
assertThat(batchSql("SELECT file_path FROM
`blob_table$files`").size()).isEqualTo(2);
}
+ @Test
+ public void testDedicatedSplitGenerationWithBlobProjection() {
+ tEnv.executeSql(
+ "CREATE TABLE dedicated_blob_table (id INT, data STRING,
picture BYTES)"
+ + " WITH ('row-tracking.enabled'='true',"
+ + " 'data-evolution.enabled'='true',"
+ + " 'blob-field'='picture')");
+ batchSql(
+ "INSERT INTO dedicated_blob_table VALUES"
+ + " (1, 'paimon', X'48656C6C6F'),"
+ + " (2, 'flink', X'5945')");
+
+ assertThat(
+ batchSql(
+ "SELECT picture, id FROM dedicated_blob_table"
+ + " /*+
OPTIONS('scan.dedicated-split-generation'='true') */"
+ + " ORDER BY id"))
+ .containsExactly(
+ Row.of(new byte[] {72, 101, 108, 108, 111}, 1),
+ Row.of(new byte[] {89, 69}, 2));
+
+ batchSql("ALTER TABLE dedicated_blob_table SET
('blob-as-descriptor'='true')");
+ List<Row> descriptorRows =
+ batchSql(
+ "SELECT picture, id FROM dedicated_blob_table"
+ + " /*+
OPTIONS('scan.dedicated-split-generation'='true') */"
+ + " ORDER BY id");
+ assertThat(descriptorRows).hasSize(2);
+ assertThat(BlobDescriptor.deserialize((byte[])
descriptorRows.get(0).getField(0)).length())
+ .isEqualTo(5);
+ assertThat(descriptorRows.get(0).getField(1)).isEqualTo(1);
+ assertThat(BlobDescriptor.deserialize((byte[])
descriptorRows.get(1).getField(0)).length())
+ .isEqualTo(2);
+ assertThat(descriptorRows.get(1).getField(1)).isEqualTo(2);
+ }
+
@Test
public void testMultipleBlobs() {
batchSql("SELECT * FROM multiple_blob_table");
diff --git
a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/ReadOperatorBlobTest.java
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/ReadOperatorBlobTest.java
new file mode 100644
index 0000000000..19ddab1a38
--- /dev/null
+++
b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/operator/ReadOperatorBlobTest.java
@@ -0,0 +1,129 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.paimon.flink.source.operator;
+
+import org.apache.paimon.data.BinaryRow;
+import org.apache.paimon.data.Blob;
+import org.apache.paimon.data.BlobDescriptor;
+import org.apache.paimon.data.GenericRow;
+import org.apache.paimon.data.InternalRow;
+import org.apache.paimon.disk.IOManager;
+import org.apache.paimon.metrics.MetricRegistry;
+import org.apache.paimon.reader.RecordReader;
+import org.apache.paimon.table.source.DataSplit;
+import org.apache.paimon.table.source.Split;
+import org.apache.paimon.table.source.TableRead;
+import org.apache.paimon.types.DataTypes;
+import org.apache.paimon.types.RowType;
+import org.apache.paimon.utils.IteratorRecordReader;
+import org.apache.paimon.utils.UriReader;
+
+import org.apache.flink.streaming.runtime.streamrecord.StreamRecord;
+import org.apache.flink.streaming.util.OneInputStreamOperatorTestHarness;
+import org.apache.flink.table.data.RowData;
+import org.apache.flink.table.runtime.typeutils.InternalSerializers;
+import org.junit.jupiter.api.Test;
+
+import java.io.IOException;
+import java.util.Collections;
+
+import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** Tests {@link ReadOperator} reading BLOB values on the dedicated split
path. */
+public class ReadOperatorBlobTest {
+
+ private static final RowType READ_TYPE = RowType.of(DataTypes.BLOB());
+
+ @Test
+ public void testReadBlobAsDescriptor() throws Exception {
+ BlobDescriptor descriptor = new BlobDescriptor("file:///blob", 7, 11);
+ byte[] result = readBlob(Blob.fromDescriptor(UriReader.fromHttp(),
descriptor), true);
+
+ assertThat(result).isEqualTo(descriptor.serialize());
+ }
+
+ @Test
+ public void testReadBlobAsData() throws Exception {
+ byte[] data = new byte[] {1, 2, 3};
+
+ assertThat(readBlob(Blob.fromData(data), false)).isEqualTo(data);
+ }
+
+ private byte[] readBlob(Blob blob, boolean blobAsDescriptor) throws
Exception {
+ ReadOperator operator =
+ new ReadOperator(
+ () -> new TestingTableRead(GenericRow.of(blob)),
+ null,
+ null,
+ READ_TYPE,
+ blobAsDescriptor);
+ OneInputStreamOperatorTestHarness<Split, RowData> harness =
+ new OneInputStreamOperatorTestHarness<>(operator);
+ harness.setup(InternalSerializers.create(toLogicalType(READ_TYPE)));
+ harness.open();
+ try {
+ DataSplit split =
+ DataSplit.builder()
+ .withPartition(BinaryRow.EMPTY_ROW)
+ .withBucket(0)
+ .withBucketPath("bucket-0")
+ .withDataFiles(Collections.emptyList())
+ .build();
+ harness.processElement(new StreamRecord<>(split));
+
+ assertThat(harness.getOutput()).hasSize(1);
+ @SuppressWarnings("unchecked")
+ StreamRecord<RowData> result =
+ (StreamRecord<RowData>)
harness.getOutput().iterator().next();
+ return result.getValue().getBinary(0);
+ } finally {
+ harness.close();
+ }
+ }
+
+ private static class TestingTableRead implements TableRead {
+
+ private final InternalRow row;
+
+ private TestingTableRead(InternalRow row) {
+ this.row = row;
+ }
+
+ @Override
+ public TableRead withMetricRegistry(MetricRegistry registry) {
+ return this;
+ }
+
+ @Override
+ public TableRead executeFilter() {
+ return this;
+ }
+
+ @Override
+ public TableRead withIOManager(IOManager ioManager) {
+ return this;
+ }
+
+ @Override
+ public RecordReader<InternalRow> createReader(Split split) throws
IOException {
+ return new
IteratorRecordReader<>(Collections.singleton(row).iterator());
+ }
+ }
+}
