[flink] Scope scan.parallelism to source operators

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

@@ -25,32 +25,46 @@
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.lineage.LineageVertex;
 import org.apache.flink.streaming.api.lineage.LineageVertexProvider;
+import org.apache.flink.table.connector.ParallelismProvider;
 import org.apache.flink.table.connector.ProviderContext;
 import org.apache.flink.table.connector.source.DataStreamScanProvider;
 import org.apache.flink.table.data.RowData;
 
+import java.util.Optional;
 import java.util.function.Function;
 
 /**
  * Paimon {@link DataStreamScanProvider} that also implements {@link LineageVertexProvider} so
  * Flink's lineage graph discovers the Paimon source table.
  */
-public class PaimonDataStreamScanProvider implements DataStreamScanProvider, LineageVertexProvider {
+public class PaimonDataStreamScanProvider
+        implements DataStreamScanProvider, ParallelismProvider, LineageVertexProvider {
 
     private final boolean isBounded;
     private final Function<StreamExecutionEnvironment, DataStream<RowData>> producer;
     private final String name;
     private final Table table;
+    private final Optional<Integer> parallelism;
 
     public PaimonDataStreamScanProvider(
             boolean isBounded,
             Function<StreamExecutionEnvironment, DataStream<RowData>> producer,
             String name,
             Table table) {
+        this(isBounded, producer, name, table, Optional.empty());
+    }
+
+    public PaimonDataStreamScanProvider(
+            boolean isBounded,
+            Function<StreamExecutionEnvironment, DataStream<RowData>> producer,
+            String name,
+            Table table,
+            Optional<Integer> parallelism) {
         this.isBounded = isBounded;
         this.producer = producer;
         this.name = name;
         this.table = table;
+        this.parallelism = parallelism;
     }
 
     @Override
@@ -64,6 +78,11 @@ public boolean isBounded() {
         return isBounded;
     }
 
+    @Override
+    public Optional<Integer> getParallelism() {
+        return parallelism;
+    }
+
     @Override
     public LineageVertex getLineageVertex() {
         return LineageUtils.sourceLineageVertex(name, isBounded, table);

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

@@ -76,6 +76,7 @@
 import static org.apache.paimon.CoreOptions.MergeEngine.FIRST_ROW;
 import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_ASYNC;
 import static org.apache.paimon.flink.FlinkConnectorOptions.LOOKUP_ASYNC_THREAD_NUMBER;
+import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_PARALLELISM;
 import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_REMOVE_NORMALIZE;
 import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_WATERMARK_ALIGNMENT_GROUP;
 import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_WATERMARK_ALIGNMENT_MAX_DRIFT;
@@ -217,7 +218,8 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
                                         .env(env)
                                         .build(),
                 tableIdentifier.asSummaryString(),
-                table);
+                table,
+                options.getOptional(SCAN_PARALLELISM));
     }
 
     protected Table tableForScan() {

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

@@ -216,7 +216,8 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
                                         tableIdentifier.asSummaryString())
                                 .setParallelism(1),
                 tableIdentifier.asSummaryString(),
-                table);
+                table,
+                Optional.of(1));
     }
 
     @Override

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

@@ -39,6 +39,8 @@
 
 import javax.annotation.Nullable;
 
+import static org.apache.paimon.flink.FlinkConnectorOptions.SCAN_PARALLELISM;
+
 /** A {@link FlinkTableSource} for system table. */
 public class SystemTableSource extends FlinkTableSource {
 
@@ -132,7 +134,8 @@ public ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) {
                     return dataStreamSource;
                 },
                 tableIdentifier.asSummaryString(),
-                table);
+                table,
+                options.getOptional(SCAN_PARALLELISM));
     }
 
     @Override

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

@@ -36,6 +36,7 @@
 import org.apache.flink.streaming.api.datastream.DataStreamSink;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.api.transformations.PartitionTransformation;
+import org.apache.flink.streaming.api.transformations.SourceTransformation;
 import org.apache.flink.table.api.EnvironmentSettings;
 import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
 import org.apache.flink.table.catalog.ObjectIdentifier;
@@ -1188,19 +1189,21 @@ public void testSourceParallelism() throws Exception {
         assertThat(sourceParallelism(buildSimpleQuery(table))).isEqualTo(bExeEnv.getParallelism());
 
         // with hint
-        assertThat(
-                        sourceParallelism(
-                                buildQueryWithTableOptions(
-                                        table,
-                                        "*",
-                                        "",
-                                        new HashMap<String, String>() {
-                                            {
-                                                put(INFER_SCAN_PARALLELISM.key(), "false");
-                                                put(SCAN_PARALLELISM.key(), "66");
-                                            }
-                                        })))
-                .isEqualTo(66);
+        String queryWithHint =
+                buildQueryWithTableOptions(
+                        table,
+                        "*",
+                        "",
+                        new HashMap<String, String>() {
+                            {
+                                put(INFER_SCAN_PARALLELISM.key(), "false");
+                                put(SCAN_PARALLELISM.key(), "66");
+                            }
+                        });
+        DataStream<Row> result =
+                ((StreamTableEnvironment) bEnv).toChangelogStream(bEnv.sqlQuery(queryWithHint));
+        assertThat(result.getParallelism()).isEqualTo(bExeEnv.getParallelism());
+        assertThat(sourceParallelism(result)).isEqualTo(66);
     }
 
     @Test
@@ -1283,7 +1286,7 @@ public void testInferParallelism() throws Exception {
                                                         put(SCAN_PARALLELISM.key(), "-2");
                                                     }
                                                 })))
-                .hasMessageContaining("The parallelism of an operator must be at least 1");
+                .hasMessageContaining("Invalid configured parallelism -2");
 
         // 2 splits, the parallelism is splits num: 2
         insertInto(table, "('Euro', 119)");
@@ -1329,7 +1332,7 @@ public void testInferParallelism() throws Exception {
                                         3L,
                                         Collections.singletonMap(
                                                 INFER_SCAN_PARALLELISM.key(), "true"))))
-                .isEqualTo(1);
+                .isEqualTo(2);
 
         // 2 splits, infer parallelism is disabled, the parallelism is scan.parallelism
         assertThat(
@@ -1879,13 +1882,22 @@ private void validatePurgingResult(
     private int sourceParallelism(String sql) {
         DataStream<Row> stream =
                 ((StreamTableEnvironment) bEnv).toChangelogStream(bEnv.sqlQuery(sql));
-        return stream.getParallelism();
+        return sourceParallelism(stream);
     }
 
     private int sourceParallelismStreaming(String sql) {
         DataStream<Row> stream =
                 ((StreamTableEnvironment) sEnv).toChangelogStream(sEnv.sqlQuery(sql));
-        return stream.getParallelism();
+        return sourceParallelism(stream);
+    }
+
+    private int sourceParallelism(DataStream<Row> stream) {
+        return stream.getTransformation().getTransitivePredecessors().stream()
+                .filter(SourceTransformation.class::isInstance)
+                .map(SourceTransformation.class::cast)
+                .findFirst()
+                .orElseThrow(() -> new AssertionError("Source transformation not found"))
+                .getParallelism();
     }
 
     private void testSinkParallelism(

paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java

@@ -48,6 +48,7 @@
 import org.apache.flink.table.connector.source.DynamicTableSource;
 import org.apache.flink.table.connector.source.LookupTableSource;
 import org.apache.flink.table.connector.source.ScanTableSource;
+import org.apache.flink.table.connector.source.abilities.SupportsRowLevelModificationScan.RowLevelModificationType;
 import org.apache.flink.table.data.RowData;
 import org.apache.flink.table.types.DataType;
 import org.apache.flink.table.types.logical.LogicalType;
@@ -80,6 +81,7 @@ void testInferScanParallelism() throws Exception {
                 new DataTableSource(
                         ObjectIdentifier.of("cat", "db", "table"), fileStoreTable, true, null);
         PaimonDataStreamScanProvider runtimeProvider = runtimeProvider(tableSource);
+        assertThat(runtimeProvider.getParallelism()).isEmpty();
         StreamExecutionEnvironment sEnv1 = StreamExecutionEnvironment.createLocalEnvironment();
         sEnv1.setParallelism(-1);
         DataStream<RowData> sourceStream1 =
@@ -102,6 +104,46 @@ void testInferScanParallelism() throws Exception {
         assertThat(sourceStream2.getParallelism()).isEqualTo(sEnv2.getParallelism());
     }
 
+    @Test
+    void testConfiguredScanParallelism() throws Exception {
+        FileStoreTable fileStoreTable =
+                createTable(
+                        ImmutableMap.of(
+                                "bucket", "1",
+                                "bucket-key", "a",
+                                "scan.parallelism", "3"));
+
+        DataTableSource tableSource =
+                new DataTableSource(
+                        ObjectIdentifier.of("cat", "db", "table"), fileStoreTable, true, null);
+        PaimonDataStreamScanProvider runtimeProvider = runtimeProvider(tableSource);
+
+        assertThat(runtimeProvider.getParallelism()).contains(3);
+
+        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
+        env.setParallelism(7);
+        DataStream<RowData> sourceStream =
+                runtimeProvider.produceDataStream(s -> Optional.empty(), env);
+        assertThat(sourceStream.getParallelism()).isEqualTo(3);
+    }
+
+    @Test
+    void testEmptyRowLevelModificationScanParallelism() throws Exception {
+        FileStoreTable fileStoreTable =
+                createTable(
+                        ImmutableMap.of(
+                                "bucket", "-1",
+                                "row-tracking.enabled", "true",
+                                "data-evolution.enabled", "true"));
+
+        DataTableSource tableSource =
+                new DataTableSource(
+                        ObjectIdentifier.of("cat", "db", "table"), fileStoreTable, true, null);
+        tableSource.applyRowLevelModificationScan(RowLevelModificationType.DELETE, null);
+
+        assertThat(runtimeProvider(tableSource).getParallelism()).contains(1);
+    }
+
     @Test
     void testInferPostponeMergeParallelism() throws Exception {
         FileStoreTable fileStoreTable = createPostponeTable(false);
@@ -168,6 +210,7 @@ public void testSystemTableParallelism() throws Exception {
         StreamExecutionEnvironment sEnv1 = StreamExecutionEnvironment.createLocalEnvironment();
         DataStream<RowData> sourceStream1 =
                 runtimeProvider.produceDataStream(s -> Optional.empty(), sEnv1);
+        assertThat(runtimeProvider.getParallelism()).contains(3);
         assertThat(sourceStream1.getParallelism()).isEqualTo(3);
     }