[Bug][Flink] 'scan.parallelism' cannot be scoped to source because PaimonDataStreamScanProvider does not override ParallelismProvider#getParallelism()
- Dominant language
- Java
- Stars
- 3.4k
- Forks
- 1.4k
- Avg merge
- 1d 11h
- Merged PRs (30d)
- 396
Description
### Search before asking
- [x] I searched in the [issues](https://github.com/apache/paimon/issues) and found nothing similar.
### Paimon version
1.3.1
### Compute Engine
Flink
### Minimal reproduce step
-- Flink 1.19+
SET 'parallelism.default' = '4';
CREATE TABLE t (
id BIGINT,
v STRING
) WITH (
'connector' = 'paimon',
'bucket' = '16',
'scan.parallelism' = '16'
);
EXPLAIN PLAN FOR
SELECT id, UPPER(v) FROM t;
Inspect the resulting job graph in the Flink Web UI / EXPLAIN.
### What doesn't meet your expectations?
Expected
The scan.parallelism option, by its name and by Flink's standard Table connector contract, is supposed to control only the source operator's parallelism. Downstream operators should
run at parallelism.default, with a rebalance edge inserted between them and the source:
Source[t] (parallelism=16) --rebalance--> Calc(UPPER(v)) (parallelism=4)
Actual
The configured scan.parallelism = 16 propagates to every downstream operator in the forward chain. The whole pipeline runs at parallelism 16, ignoring parallelism.default:
Source[t] (parallelism=16) --forward--> Calc(UPPER(v)) (parallelism=16)
There is no way through SQL today to read a Paimon table with high source parallelism while keeping downstream computation at the global default.
### Anything else?
Root cause
PaimonDataStreamScanProvider already inherits the ParallelismProvider interface (Flink's DataStreamScanProvider extends ParallelismProvider), but it never overrides
getParallelism(), so the planner always receives the default Optional.empty(). Paimon then applies the configured parallelism by mutating the produced DataStream directly
References
- Flink — [Table connector ParallelismProvider](https://nightlies.apache.org/flink/flink-docs-stable/docs/dev/table/sourcessinks/#parallelism)
- [FLIP-367: Support Setting Parallelism for Table/SQL Sources](https://cwiki.apache.org/confluence/pages/viewpage.action?pageId=263429150)
### Are you willing to submit a PR?
- [x] I'm willing to submit a PR!
Contributor guide
No contributing guide indexed for this repository
Research direction
Locate PaimonDataStreamScanProvider and inspect its inherited ParallelismProvider#getParallelism() behavior, then review the Flink DataStreamScanProvider parallelism contract. Run the provided Flink 1.19+ SQL reproduction and inspect EXPLAIN or the Flink Web UI. Done means scan.parallelism controls only the source, downstream operators use parallelism.default, and a rebalance edge appears between them.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- stream-processing
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 68/100