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 776575dc0a72..f10a48512883 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.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> producer; private final String name; private final Table table; + private final Optional parallelism; public PaimonDataStreamScanProvider( boolean isBounded, Function> producer, String name, Table table) { + this(isBounded, producer, name, table, Optional.empty()); + } + + public PaimonDataStreamScanProvider( + boolean isBounded, + Function> producer, + String name, + Table table, + Optional 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 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 40a8fe112574..9bdbf0e7dc72 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.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() { 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 c4968729937f..7a18546a1fd6 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 ScanRuntimeProvider getScanRuntimeProvider(ScanContext scanContext) { 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 a78f44bc6fc4..f6cba0e69698 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 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 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 def5be553ae8..e0b4172448e0 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.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() { - { - put(INFER_SCAN_PARALLELISM.key(), "false"); - put(SCAN_PARALLELISM.key(), "66"); - } - }))) - .isEqualTo(66); + String queryWithHint = + buildQueryWithTableOptions( + table, + "*", + "", + new HashMap() { + { + put(INFER_SCAN_PARALLELISM.key(), "false"); + put(SCAN_PARALLELISM.key(), "66"); + } + }); + DataStream 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 stream = ((StreamTableEnvironment) bEnv).toChangelogStream(bEnv.sqlQuery(sql)); - return stream.getParallelism(); + return sourceParallelism(stream); } private int sourceParallelismStreaming(String sql) { DataStream stream = ((StreamTableEnvironment) sEnv).toChangelogStream(sEnv.sqlQuery(sql)); - return stream.getParallelism(); + return sourceParallelism(stream); + } + + private int sourceParallelism(DataStream 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 23614f13b3fc..e3cbb43078c5 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.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 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 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 sourceStream1 = runtimeProvider.produceDataStream(s -> Optional.empty(), sEnv1); + assertThat(runtimeProvider.getParallelism()).contains(3); assertThat(sourceStream1.getParallelism()).isEqualTo(3); }