This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 1d8226d7a9 [flink] Scope scan.parallelism to source operators (#9803)
1d8226d7a9 is described below
commit 1d8226d7a9e702cbd0ce386cb283961065a9916b
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);
}
