This is an automated email from the ASF dual-hosted git repository. XiaoHongbo-Hope pushed a commit to branch release-2.1 in repository https://gitbox.apache.org/repos/asf/paimon.git
commit dd4837d595d061faaaa3aab69bcf4e8e358b6adb Author: Akash Reddy Jammula <[email protected]> AuthorDate: Thu Sep 17 02:29:54 2026 -0700 [flink] Scope scan.parallelism to source operators (#9803) --- .../paimon/flink/PaimonDataStreamScanProvider.java | 21 +++++++++- .../paimon/flink/source/BaseDataTableSource.java | 4 +- .../paimon/flink/source/DataTableSource.java | 3 +- .../paimon/flink/source/SystemTableSource.java | 5 ++- .../apache/paimon/flink/ReadWriteTableITCase.java | 46 ++++++++++++++-------- .../paimon/flink/source/DataTableSourceTest.java | 43 ++++++++++++++++++++ 6 files changed, 101 insertions(+), 21 deletions(-) diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PaimonDataStreamScanProvider.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PaimonDataStreamScanProvider.java index 776575dc0a..f10a485128 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PaimonDataStreamScanProvider.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/PaimonDataStreamScanProvider.java @@ -25,32 +25,46 @@ import org.apache.flink.streaming.api.datastream.DataStream; 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 class PaimonDataStreamScanProvider implements DataStreamScanProvider, Lin return isBounded; } + @Override + public Optional<Integer> getParallelism() { + return parallelism; + } + @Override public LineageVertex getLineageVertex() { return LineageUtils.sourceLineageVertex(name, isBounded, table); diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java index 40a8fe1125..9bdbf0e7dc 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/BaseDataTableSource.java +++ b/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.CHANGELOG_PRODUCER; 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 abstract class BaseDataTableSource extends FlinkTableSource .env(env) .build(), tableIdentifier.asSummaryString(), - table); + table, + options.getOptional(SCAN_PARALLELISM)); } protected Table tableForScan() { diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java index c496872993..7a18546a1f 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/DataTableSource.java @@ -216,7 +216,8 @@ public class DataTableSource extends BaseDataTableSource tableIdentifier.asSummaryString()) .setParallelism(1), tableIdentifier.asSummaryString(), - table); + table, + Optional.of(1)); } @Override diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SystemTableSource.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SystemTableSource.java index a78f44bc6f..f6cba0e696 100644 --- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SystemTableSource.java +++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/source/SystemTableSource.java @@ -39,6 +39,8 @@ import org.apache.flink.table.data.RowData; 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 class SystemTableSource extends FlinkTableSource { return dataStreamSource; }, tableIdentifier.asSummaryString(), - table); + table, + options.getOptional(SCAN_PARALLELISM)); } @Override diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java index def5be553a..e0b4172448 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/ReadWriteTableITCase.java +++ b/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.DataStream; 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 class ReadWriteTableITCase extends AbstractTestBase { 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 class ReadWriteTableITCase extends AbstractTestBase { 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 class ReadWriteTableITCase extends AbstractTestBase { 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 @@ public class ReadWriteTableITCase extends AbstractTestBase { 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( diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java index 23614f13b3..e3cbb43078 100644 --- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java +++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/source/DataTableSourceTest.java @@ -48,6 +48,7 @@ import org.apache.flink.table.catalog.ObjectIdentifier; 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 @@ class DataTableSourceTest { 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 @@ class DataTableSourceTest { 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 @@ class DataTableSourceTest { StreamExecutionEnvironment sEnv1 = StreamExecutionEnvironment.createLocalEnvironment(); DataStream<RowData> sourceStream1 = runtimeProvider.produceDataStream(s -> Optional.empty(), sEnv1); + assertThat(runtimeProvider.getParallelism()).contains(3); assertThat(sourceStream1.getParallelism()).isEqualTo(3); }
