apache / apache/paimon

[Bug][Flink] 'scan.parallelism' cannot be scoped to source because PaimonDataStreamScanProvider does not override ParallelismProvider#getParallelism()

Open
#7,905 0 comments 0 reactions 0 assignees View on GitHub
bug
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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.