You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@paimon.apache.org by ho...@apache.org on 2026/09/20 04:08:18 UTC
(paimon) 15/20: [flink] Scope scan.parallelism to source operators (#9803)
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 <ak...@gmail.com>
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);
}