Akash3121 opened a new pull request, #9803:
URL: https://github.com/apache/paimon/pull/9803
### Purpose
Fixes the behavior reported in issue #7905.
Paimon currently applies `scan.parallelism` directly to the `DataStream`
produced by `PaimonDataStreamScanProvider`. However, on Flink 1.19 and later,
the Table planner uses `ParallelismProvider#getParallelism()` to distinguish
connector-defined source parallelism from the parallelism of downstream
operators.
Because `PaimonDataStreamScanProvider` did not expose the configured
parallelism through this contract, the source parallelism propagated through
the forward chain. For example:
```text
parallelism.default = 4
scan.parallelism = 16
Before:
Source (16) --forward--> Calc (16)
Expected:
Source (16) --exchange--> Calc (4)
```
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:
- An explicitly configured scan.parallelism is returned by the runtime
provider.
- The produced source retains the configured parallelism.
- Inferred parallelism is not incorrectly exposed as connector-configured
planner metadata.
- Existing inferred source-parallelism behavior is preserved.
- System-table scans expose their configured parallelism.
- Empty row-level-modification scans expose their fixed parallelism of 1 .
- The source transformation uses scan.parallelism while the downstream
result keeps the environment default parallelism.
- Invalid configured parallelism is rejected by the Flink planner.
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
- Make PaimonDataStreamScanProvider explicitly implement
ParallelismProvider .
- Add an optional source-parallelism value and return it from
getParallelism() .
- Preserve the existing four-argument constructor for source and binary
compatibility.
- Propagate explicitly configured scan.parallelism from normal table
scans and system-table scans to the runtime provider.
- Report parallelism 1 for the internally fixed empty
row-level-modification source.
- Retain the existing producer-side parallelism assignment for Flink
1.16–1.18, whose planners do not consume source parallelism from
ParallelismProvider .
- Keep inferred scan parallelism in the existing runtime path instead of
exposing it through getParallelism() .
####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
- Bug fix: scan.parallelism is now scoped to the Paimon source operator
on Flink 1.19 and later instead of propagating to downstream operators.
### Notes for reviewers
- Keeping both getParallelism() and the existing producer-side
setParallelism() is intentional. The former fixes planner behavior on Flink
1.19+, while the latter preserves support for older Flink versions.
- Inferred parallelism intentionally returns Optional.empty() from the
provider because it is calculated later using the runtime environment and scan
splits.
- The provider explicitly implements ParallelismProvider even though
newer DataStreamScanProvider versions inherit it. This keeps the method valid
against older Flink APIs where ParallelismProvider already exists but is not
a parent of DataStreamScanProvider
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
