ericyuan915 opened a new issue, #19516: URL: https://github.com/apache/hudi/issues/19516
### Task Description ### Task Description **What needs to be done:** Give **bounded (batch) reads** in Flink Source V2 a shared, work-stealing split pool instead of the per-subtask split pinning they currently inherit from the streaming path. Concretely: - add a `HoodieSplitProvider` implementation backed by a single shared queue, whose `getNext(subtaskId, hostname)` ignores the subtask id so any idle reader takes the next pending split; - select it in `HoodieSource.createEnumerator` **only** on the non-streaming (`HoodieStaticSplitEnumerator`) branch. Streaming keeps `DefaultHoodieSplitProvider` and the existing assigners, unchanged. No enumerator change is required: with one shared pool, `getNext` returning empty already means "globally drained", so the existing `signalNoMoreSplits` logic stays correct, and `addSplitsBack` (failed reader) returns splits to the shared pool where any reader can pick them up. **Why this task is needed:** Today a bounded read pins every split to one subtask at discovery and never rebalances, so readers that draw a heavier share keep working while their peers sit idle. On a bounded backfill of one date partition of a COW table (~16.4K base files → ~16.4K splits, parallelism 32, Flink 1.18) the 32 reader subtasks finished **78 minutes apart** (fastest 125 min, slowest 203 min); for the last ~28 minutes one subtask read alone while the other 31 were idle. A prototype of the change above cut the same job from **3.80 h to 2.77 h**. --- ## What happens today 1. Run a bounded (batch) `HoodieSource` read over a partition with thousands of base files, parallelism N ≫ 1 (e.g. a COW snapshot read restricted to one partition). 2. Watch per-subtask `numRecordsIn` / subtask finish times. 3. Subtasks finish at widely spread times, aggregate throughput decays in a staircase as readers drop out one by one, and one or more stragglers run alone at the end. Results are correct — this is purely wasted capacity. ## Root cause Split → subtask binding happens at discovery and is permanent. `DefaultHoodieSplitProvider` keeps **one queue per subtask** and pins each split on arrival ([`DefaultHoodieSplitProvider.java#L80-L82`](https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/DefaultHoodieSplitProvider.java#L80-L82)): ```java private void addSplits(Collection<HoodieSourceSplit> splits) { for (HoodieSourceSplit split : splits) { int taskId = splitAssigner.assign(split); Queue<HoodieSourceSplit> queue = pendingSplits.computeIfAbsent( taskId, k -> new PriorityBlockingQueue<>(DEFAULT_SPLIT_QUEUE_SIZE, comparator)); queue.add(split); ... ``` and `getNext` serves a reader **only from its own queue** ([`#L59-L68`](https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/DefaultHoodieSplitProvider.java#L59-L68)): ```java public Option<HoodieSourceSplit> getNext(int subTaskId, @Nullable String hostname) { if (pendingSplits.containsKey(subTaskId)) { Queue<HoodieSourceSplit> splits = pendingSplits.get(subTaskId); if (!splits.isEmpty()) { return Option.of(splits.poll()); } } return Option.empty(); } ``` Two consequences for bounded reads: 1. **Assignment balances split *count*, not bytes or records.** `DefaultHoodieSplitAssigner` hashes the file id ([`#L48-L49`](https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/split/assign/DefaultHoodieSplitAssigner.java#L48-L49)); `HoodieSplitNumberAssigner` uses `splitNum % parallelism`. Equal split counts ≠ equal work when file groups differ in size or per-record cost. 2. **No rebalancing.** Once a subtask's queue empties, `getNext` returns empty and the static enumerator permanently signals no-more-splits for it ([`AbstractHoodieSplitEnumerator.java#L157-L171`](https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/enumerator/AbstractHoodieSplitEnumerator.java#L157-L171)) — even while other subtasks still have a backlog. That idle capacity cannot be reclaimed. Because per-subtask split count is a hash-fixed *input* rather than an *outcome* of reader speed, it is uncorrelated with throughput: in our run `corr(splits taken, read rate) = −0.37`. Note that `createEnumerator` builds the provider **before** branching on streaming vs bounded ([`HoodieSource.java#L135-L168`](https://github.com/apache/hudi/blob/master/hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/HoodieSource.java#L135-L168)) — the bounded path inherits the streaming provider structurally rather than by explicit choice. ## Why the current design does this (and why bounded is different) The pinning looks deliberate and correct **for streaming**: `DefaultHoodieSplitAssigner` uses Flink's own key-group function, `KeyGroupRangeAssignment.assignKeyToParallelOperator(split.getFileId(), ...)` — i.e. file-group affinity mirroring keyed-state partitioning. For a continuous read that matters: - a MOR file group accumulates log files across commits and the continuous enumerator keeps emitting new splits for the **same** file id, so pinning keeps successive splits of one file group on one reader (no two readers concurrently merging the same file group, per-file-group commit order preserved); - `HoodieSplitBucketAssigner` similarly aligns bucket id → subtask for bucket-index tables, letting a read line up with a bucketed/keyed downstream without a shuffle. None of that applies to a bounded snapshot / read-optimized read: exactly one split per file group, no cross-commit continuation, no ordering relationship between splits. The affinity buys nothing in batch, while its cost (no rebalancing, count-based balance) is paid in full. So this isn't the streaming design being wrong — it's the bounded path inheriting a constraint that only pays for itself in streaming. ## Measured impact Prototype patch in a fork, same table / partition / parallelism, **only the split provider changed**: | | pinned (current) | shared pool | |---|---|---| | Per-subtask finish spread | **78 min** (125–203 min) | **0 min** (all 32 at 166 min) | | Splits per subtask | 491–608 | 364–644 | | corr(splits taken, read rate) | −0.37 | **+0.997** | | Wall clock | **3.80 h** | **2.77 h** | All 16,395 splits processed in both runs, 0 restarts. The correlation flip is the clearest signal: under a shared pool the split count becomes an *output* (faster readers pull more, everyone finishes together) instead of a hash-fixed input. The residual tail is then bounded by the duration of a single in-flight split (~tens of seconds) rather than by accumulated imbalance — stealing cannot preempt a split already being read, so one pathologically large file group remains the only exposure. ## Properties of the proposed change - **Checkpoint size and restore semantics unchanged.** The enumerator still snapshots the same set of pending splits — one queue instead of N. On restore a pending split may be picked up by a different subtask, which is safe precisely because bounded splits are independent. - Splits can still be served in commit-time order via the existing `HoodieSourceSplitComparator`. - Can be made opt-in behind a config rather than the default for bounded reads, if preferred. ## Environment - Hudi: 1.2.x line (`hudi-flink-datasource`, Flink Source V2). Verified against `master` at time of filing: `DefaultHoodieSplitProvider` is unchanged in the respects above, and no shared-pool provider exists. - Flink: 1.18 - Table type: COPY_ON_WRITE, bounded snapshot read with partition pruning - Parallelism: 32; ~16.4K base files in the read partition - Storage: object store ## Questions for maintainers 1. Should the shared pool be the **default** for bounded reads, or opt-in behind a config? 2. Is there a bounded-read scenario where file-group → subtask affinity is load-bearing (e.g. a bucketed bounded read feeding a bucketed write that relies on the source's partitioning to avoid a shuffle)? If so, the selection should key off the assigner strategy rather than simply `isStreaming()`. Happy to open a PR with the shared-pool provider plus unit tests (work stealing across subtasks; no-more-splits only when the pool is globally drained; `addSplitsBack` into the shared pool; checkpoint state round-trip) if the approach and the bounded-only scoping look right. ### Task Type Performance optimization ### Related Issues **Parent feature issue:** n/a **Related issues:** no blockers. Context only — the Flink Source V2 components involved were introduced in #17503 (static split enumerator), #17773 (source reader) and #18082 (split distribution strategy). ### Task Type Code improvement/refactoring ### Related Issues **Parent feature issue:** (if applicable ) **Related issues:** NOTE: Use `Relationships` button to add parent/blocking issues after issue is created. -- 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]
