[Flink] Support withShard read of Flink

Author: wwj6591812

docs/layouts/shortcodes/generated/flink_connector_configuration.html

@@ -162,7 +162,7 @@
             <td><h5>scan.split-enumerator.mode</h5></td>
             <td style="word-wrap: break-word;">fair</td>
             <td><p>Enum</p></td>
-            <td>The mode used by StaticFileStoreSplitEnumerator to assign splits.<br /><br />Possible values:<ul><li>"fair": Distribute splits evenly when batch reading to prevent a few tasks from reading all.</li><li>"preemptive": Distribute splits preemptively according to the consumption speed of the task.</li></ul></td>
+            <td>The mode used by StaticFileStoreSplitEnumerator to assign splits.<br /><br />Possible values:<ul><li>"fair": Distribute splits evenly when batch reading to prevent a few tasks from reading all.</li><li>"preemptive": Distribute splits preemptively according to the consumption speed of the task.</li><li>"empty": Splits will generate on TaskManager.</li></ul></td>
         </tr>
         <tr>
             <td><h5>scan.watermark.alignment.group</h5></td>

paimon-core/src/main/java/org/apache/paimon/table/source/EmptyDataSplit.java

@@ -0,0 +1,28 @@
+/*
+ * 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.table.source;
+
+/** Input splits. Needed by most batch computation engines. */
+public class EmptyDataSplit implements Split {
+
+    @Override
+    public long rowCount() {
+        return 0;
+    }
+}

paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/FlinkConnectorOptions.java

@@ -547,7 +547,8 @@ public enum SplitAssignMode implements DescribedEnum {
                 "Distribute splits evenly when batch reading to prevent a few tasks from reading all."),
         PREEMPTIVE(
                 "preemptive",
-                "Distribute splits preemptively according to the consumption speed of the task.");
+                "Distribute splits preemptively according to the consumption speed of the task."),
+        EMPTY("empty", "Splits will generate on TaskManager.");
 
         private final String value;
         private final String description;

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

@@ -22,18 +22,20 @@
 import org.apache.paimon.flink.NestedProjectedRowData;
 import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
 import org.apache.paimon.flink.utils.TableScanUtils;
-import org.apache.paimon.table.source.TableRead;
 
 import org.apache.flink.api.connector.source.SourceReader;
 import org.apache.flink.api.connector.source.SourceReaderContext;
 import org.apache.flink.connector.base.source.reader.SingleThreadMultiplexSourceReaderBase;
+import org.apache.flink.connector.base.source.reader.splitreader.SplitReader;
+import org.apache.flink.connector.file.src.reader.BulkFormat;
 import org.apache.flink.connector.file.src.reader.BulkFormat.RecordIterator;
 import org.apache.flink.table.data.RowData;
 
 import javax.annotation.Nullable;
 
 import java.util.Map;
 import java.util.Optional;
+import java.util.function.Supplier;
 
 /** A {@link SourceReader} that read records from {@link FileStoreSourceSplit}. */
 public class FileStoreSourceReader
@@ -45,19 +47,14 @@ public class FileStoreSourceReader
     private long lastConsumeSnapshotId = Long.MIN_VALUE;
 
     public FileStoreSourceReader(
+            Supplier<SplitReader<BulkFormat.RecordIterator<RowData>, FileStoreSourceSplit>>
+                    splitReaderSupplier,
             SourceReaderContext readerContext,
-            TableRead tableRead,
             FileStoreSourceReaderMetrics metrics,
             IOManager ioManager,
-            @Nullable Long limit,
             @Nullable NestedProjectedRowData rowData) {
-        // limiter is created in SourceReader, it can be shared in all split readers
         super(
-                () ->
-                        new FileStoreSourceSplitReader(
-                                tableRead.withIOManager(ioManager),
-                                RecordLimiter.create(limit),
-                                metrics),
+                splitReaderSupplier,
                 (element, output, state) ->
                         FlinkRecordsWithSplitIds.emitRecord(
                                 readerContext, element, output, state, metrics, rowData),

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

@@ -61,7 +61,7 @@ public class FileStoreSourceSplitReader
 
     @Nullable private final RecordLimiter limiter;
 
-    private final Queue<FileStoreSourceSplit> splits;
+    protected final Queue<FileStoreSourceSplit> splits;
 
     private final Pool<FileStoreRecordIterator> pool;
 

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

@@ -67,12 +67,16 @@ public SourceReader<RowData, FileStoreSourceSplit> createReader(SourceReaderCont
                 new FileStoreSourceReaderMetrics(metricGroup);
         TableRead tableRead =
                 readBuilder.newRead().withMetricRegistry(new FlinkMetricRegistry(metricGroup));
+        // limiter is created in SourceReader, it can be shared in all split readers
         return new FileStoreSourceReader(
+                () ->
+                        new FileStoreSourceSplitReader(
+                                tableRead.withIOManager(ioManager),
+                                RecordLimiter.create(limit),
+                                sourceReaderMetrics),
                 context,
-                tableRead,
                 sourceReaderMetrics,
                 ioManager,
-                limit,
                 NestedProjectedRowData.copy(rowData));
     }
 

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

@@ -28,6 +28,7 @@
 import org.apache.paimon.flink.sink.FlinkSink;
 import org.apache.paimon.flink.source.align.AlignedContinuousFileStoreSource;
 import org.apache.paimon.flink.source.operator.MonitorSource;
+import org.apache.paimon.flink.source.shardread.ShardStaticFileStoreSource;
 import org.apache.paimon.flink.utils.TableScanUtils;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.predicate.Predicate;
@@ -66,6 +67,7 @@
 import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;
 import static org.apache.paimon.CoreOptions.StreamingReadMode.FILE;
 import static org.apache.paimon.flink.FlinkConnectorOptions.SOURCE_OPERATOR_UID_SUFFIX;
+import static org.apache.paimon.flink.FlinkConnectorOptions.SplitAssignMode.EMPTY;
 import static org.apache.paimon.flink.FlinkConnectorOptions.generateCustomUid;
 import static org.apache.paimon.flink.LogicalTypeConversion.toLogicalType;
 import static org.apache.paimon.flink.utils.ParallelismUtils.forwardParallelism;
@@ -198,6 +200,18 @@ private DataStream<RowData> buildStaticFileSource() {
                         outerProject()));
     }
 
+    private DataStream<RowData> buildShardStaticFileSource() {
+        Options options = Options.fromMap(table.options());
+        return toDataStream(
+                new ShardStaticFileStoreSource(
+                        createReadBuilder(projectedRowType()),
+                        limit,
+                        options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_BATCH_SIZE),
+                        options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_ASSIGN_MODE),
+                        dynamicPartitionFilteringInfo,
+                        outerProject()));
+    }
+
     private DataStream<RowData> buildContinuousFileSource() {
         return toDataStream(
                 new ContinuousFileStoreSource(
@@ -295,6 +309,11 @@ public DataStream<RowData> build() {
         }
 
         if (sourceBounded) {
+            Options options = Options.fromMap(table.options());
+            if (EMPTY.equals(
+                    options.get(FlinkConnectorOptions.SCAN_SPLIT_ENUMERATOR_ASSIGN_MODE))) {
+                return buildShardStaticFileSource();
+            }
             return buildStaticFileSource();
         }
         TableScanUtils.streamingReadingValidate(table);

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

@@ -23,6 +23,7 @@
 import org.apache.paimon.flink.source.assigners.FIFOSplitAssigner;
 import org.apache.paimon.flink.source.assigners.PreAssignSplitAssigner;
 import org.apache.paimon.flink.source.assigners.SplitAssigner;
+import org.apache.paimon.flink.source.shardread.EmptySplitAssigner;
 import org.apache.paimon.table.source.InnerTableScan;
 import org.apache.paimon.table.source.ReadBuilder;
 import org.apache.paimon.table.source.TableScan;
@@ -87,7 +88,7 @@ public SplitEnumerator<FileStoreSourceSplit, PendingSplitsCheckpoint> restoreEnu
                 context, null, splitAssigner, dynamicPartitionFilteringInfo);
     }
 
-    private List<FileStoreSourceSplit> getSplits(SplitEnumeratorContext context) {
+    public List<FileStoreSourceSplit> getSplits(SplitEnumeratorContext context) {
         FileStoreSourceSplitGenerator splitGenerator = new FileStoreSourceSplitGenerator();
         TableScan scan = readBuilder.newScan();
         // register scan metrics
@@ -108,6 +109,8 @@ public static SplitAssigner createSplitAssigner(
                 return new PreAssignSplitAssigner(splitBatchSize, context, splits);
             case PREEMPTIVE:
                 return new FIFOSplitAssigner(splits);
+            case EMPTY:
+                return new EmptySplitAssigner(splitBatchSize, context, splits);
             default:
                 throw new UnsupportedOperationException(
                         "Unsupported assign mode " + splitAssignMode);

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

@@ -22,6 +22,7 @@
 import org.apache.paimon.flink.NestedProjectedRowData;
 import org.apache.paimon.flink.PaimonDataStreamScanProvider;
 import org.apache.paimon.flink.Projection;
+import org.apache.paimon.flink.source.shardread.ShardStaticFileStoreSource;
 import org.apache.paimon.options.Options;
 import org.apache.paimon.predicate.Predicate;
 import org.apache.paimon.table.BucketMode;
@@ -39,6 +40,8 @@
 
 import javax.annotation.Nullable;
 
+import static org.apache.paimon.flink.FlinkConnectorOptions.SplitAssignMode.EMPTY;
+
 /** A {@link FlinkTableSource} for system table. */
 public class SystemTableSource extends FlinkTableSource {
 
@@ -100,9 +103,15 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
                     new ContinuousFileStoreSource(
                             readBuilder, table.options(), limit, BucketMode.HASH_FIXED, rowData);
         } else {
-            source =
-                    new StaticFileStoreSource(
-                            readBuilder, limit, splitBatchSize, splitAssignMode, null, rowData);
+            if (EMPTY.equals(splitAssignMode)) {
+                source =
+                        new ShardStaticFileStoreSource(
+                                readBuilder, limit, splitBatchSize, splitAssignMode, null, rowData);
+            } else {
+                source =
+                        new StaticFileStoreSource(
+                                readBuilder, limit, splitBatchSize, splitAssignMode, null, rowData);
+            }
         }
         return new PaimonDataStreamScanProvider(
                 source.getBoundedness() == Boundedness.BOUNDED,

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

@@ -22,7 +22,9 @@
 import org.apache.paimon.flink.NestedProjectedRowData;
 import org.apache.paimon.flink.source.FileStoreSourceReader;
 import org.apache.paimon.flink.source.FileStoreSourceSplit;
+import org.apache.paimon.flink.source.FileStoreSourceSplitReader;
 import org.apache.paimon.flink.source.FileStoreSourceSplitState;
+import org.apache.paimon.flink.source.RecordLimiter;
 import org.apache.paimon.flink.source.metrics.FileStoreSourceReaderMetrics;
 import org.apache.paimon.table.source.TableRead;
 import org.apache.paimon.utils.ReflectionUtils;
@@ -61,7 +63,16 @@ public AlignedSourceReader(
             IOManager ioManager,
             @Nullable Long limit,
             @Nullable NestedProjectedRowData rowData) {
-        super(readerContext, tableRead, metrics, ioManager, limit, rowData);
+        super(
+                () ->
+                        new FileStoreSourceSplitReader(
+                                tableRead.withIOManager(ioManager),
+                                RecordLimiter.create(limit),
+                                metrics),
+                readerContext,
+                metrics,
+                ioManager,
+                rowData);
         this.nextCheckpointId = null;
         try {
             // In lower versions of Flink, the SplitFetcherManager does not provide the getQueue