[flink] Support Blob tables in dedicated split generation

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/FlinkSourceBuilder.java

@@ -475,20 +475,23 @@ private DataStream<RowData> buildDedicatedSplitGenSource(boolean isBounded) {
             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),
                         unordered,
                         outerProject(),
                         isBounded,
                         limit,
-                        table);
+                        table,
+                        readBuilder.readType(),
+                        conf.get(CoreOptions.BLOB_AS_DESCRIPTOR));
         if (parallelism != null) {
             dataStream.getTransformation().setParallelism(parallelism);
         }

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.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 static DataStream<RowData> buildSource(
             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 static DataStream<RowData> buildSource(
         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(

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/operator/ReadOperator.java

@@ -21,12 +21,15 @@
 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 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 ReadOperator(
             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 void open() throws Exception {
                                 .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();
     }

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/BlobTableITCase.java

@@ -83,6 +83,42 @@ public void testBasic() {
         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");

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