Repository navigation
[flink] Scope scan.parallelism to source operators - #9803
Conversation
Report explicit scan.parallelism through ParallelismProvider while retaining producer side parallelism assignment for older Flink versions Signed-off-by: Akash Reddy Jammula <akashreddyjammula@gmail.com>
Verify configured parallelism is exposed by the scan provider and applies only to the source while inferred and downstream parallelism behaviour remains unchanged Signed-off-by: Akash Reddy Jammula <akashreddyjammula@gmail.com>
cee9599 to
81670c8
Compare
JingsongLi
left a comment
There was a problem hiding this comment.
Requirement fit: SUPPORTED. Implementation: CLEAN.
Reviewed 81670c873487. Reporting explicitly configured source parallelism through the Flink planner contract addresses #7905. The old constructor remains available, inferred parallelism stays on the runtime path, and normal/system/empty modification sources expose their intended values. I checked the provider callers and the updated source/downstream graph assertions.
Validation: code and test review; the Flink matrix was not rerun locally. Current CI has a cancelled Java / Flink 1 / Common job, and its aggregate result is failed. The job log reports cancellation rather than an identified test failure; complete that lane before merging.
No actionable implementation regression found in this review.
Purpose
Fixes the behavior reported in issue #7905.
Paimon currently applies
scan.parallelismdirectly to theDataStreamproduced byPaimonDataStreamScanProvider. However, on Flink 1.19 and later, the Table planner usesParallelismProvider#getParallelism()to distinguish connector-defined source parallelism from the parallelism of downstream operators.Because
PaimonDataStreamScanProviderdid not expose the configured parallelism through this contract, the source parallelism propagated through the forward chain. For example:This change reports the explicitly configured scan parallelism to the Flink planner so that it is scoped to the source operator.
Tests
Added and updated coverage to verify:
Validated with:
mvn -pl paimon-flink/paimon-flink-common -am -Pfast-build -Pflink1
-DwildcardSuites=none -DfailIfNoTests=false
-Dtest=DataTableSourceTest,ReadWriteTableITCase#testSourceParallelism+testInferParallelism test
mvn -pl paimon-flink/paimon-flink-common -am -Pfast-build -Pflink2
-DwildcardSuites=none -DfailIfNoTests=false
-Dtest=DataTableSourceTest,ReadWriteTableITCase#testSourceParallelism+testInferParallelism test
Also validated checked compilation for the Flink 1 and Flink 2 profiles and packaged the Flink 1.16, 1.17, and 1.18 adapters.
Extra info
####Changes
####Compatibility
Flink 1.19+ consumes ParallelismProvider#getParallelism() for table sources and can isolate the configured source parallelism from downstream operators.
Flink 1.16–1.18 do not consume this metadata for sources. The existing direct DataStream#setParallelism() behavior is intentionally retained as a compatibility fallback.
Only an explicitly configured scan.parallelism is reported to the planner. Inferred parallelism depends on the runtime StreamExecutionEnvironment and split discovery, which are unavailable when the planner queries the provider, so inference behavior remains unchanged.
Changelog
Notes for reviewers