ericyuan915 opened a new pull request, #19520: URL: https://github.com/apache/hudi/pull/19520
### Describe the issue this Pull Request addresses Closes #19516 Bounded (batch) reads in the Flink Source V2 inherit the streaming path's split provider, which pins every split to one subtask at discovery and never rebalances. `DefaultHoodieSplitProvider` keeps one queue per subtask and assigns on arrival, and `getNext` serves a reader only from its own queue, so a subtask that drew a heavier share keeps working while its peers sit idle. The assignment also balances split *count*, not bytes or records, so the imbalance is decided before any reading happens and cannot be recovered. On a bounded backfill of one date partition of a COW table (~16.4K base files, parallelism 32, Flink 1.18), the 32 reader subtasks finished **78 minutes apart** (fastest 125 min, slowest 203 min) and one subtask read alone for the last ~28 minutes. Results were correct; the loss was pure idle capacity. Note that `createEnumerator` builds the provider *before* branching on streaming vs bounded, so the bounded path inherits the streaming provider structurally rather than by explicit choice. ### Summary and Changelog Bounded reads now serve splits from a single shared, work-stealing pool, so whichever reader finishes first takes the next split and all readers stay busy until the pool is drained. Streaming behaviour is unchanged. - **`GlobalHoodieSplitProvider` (new)** — a `HoodieSplitProvider` backed by one `PriorityBlockingQueue` ordered by the existing `HoodieSourceSplitComparator` (oldest commit first, the same ordering the per-subtask queues use). `getNext(taskId, hostname)` ignores both arguments. `onUnassignedSplits` (the `addSplitsBack` path) returns splits to the same pool. - **`HoodieSource.createEnumerator`** — the provider is now chosen on the streaming/bounded branch rather than before it: streaming keeps `DefaultHoodieSplitProvider` plus the existing assigners, bounded gets the shared pool. Restore then replays the checkpointed pending splits into whichever provider was chosen. The split assigner is only constructed on the streaming branch. No enumerator change is required. With one shared pool, `getNext` returning empty already means "globally drained", so `HoodieStaticSplitEnumerator`'s existing `signalNoMoreSplits` logic stays correct. **Why the affinity is load-bearing for streaming but not for bounded.** `DefaultHoodieSplitAssigner` uses Flink's own `KeyGroupRangeAssignment.assignKeyToParallelOperator(split.getFileId(), ...)`. 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; `HoodieSplitBucketAssigner` similarly aligns bucket id to subtask. A bounded read has none of that: **exactly one split per file group, no cross-commit continuation, and no ordering relationship between splits.** That claim covers every mode `createBatchHoodieSplits()` routes to the static enumerator, not just the COW snapshot case that was measured: | Bounded mode | Split builder | One split per file group because | |---|---|---| | COW snapshot | `FileIndexReader.baseFileOnlyHoodieSourceSplits` | `fsView.getLatestBaseFiles(par)` yields one latest base file per file group | | Read-optimized | same builder | as above | | MOR snapshot | `FileIndexReader.buildHoodieSplits` → `readFileSlice` | `getLatestMergedFileSlicesBeforeOrOn` yields one merged slice per file group | | Bounded incremental | `IncrementalInputSplits.inputSplits` → `getInputSplits` | slices also come from `getLatestMergedFileSlicesBeforeOrOn`, one split per slice | | Bounded incremental CDC | `IncrementalInputSplits.getCdcInputSplits` | the extractor returns `Map<HoodieFileGroupId, List<HoodieCDCFileSplit>>`, so one split per file group with the file group's `changes[]` sorted by instant **inside** the split — cross-commit order is intra-split, never cross-split | `TestHoodieSourceEnumeratorRouting` asserts this invariant (distinct file ids) for each of those five modes rather than leaving it as prose, so a future change that starts emitting multiple splits per file group in a bounded mode fails the test. I also could not find a Source V2 partitioning contract exposed to downstream operators that would make bucket/file-id affinity load-bearing for a bounded scan: `HoodieTableSource.addFileDistributionStrategy` is applied only to the V1 `DataStream<MergeOnReadInputSplit>` monitoring stream, never to `HoodieSource`. **Consistency with the V1 source and with this provider's own history.** V1 bounded reads already use a shared pool — MOR/incremental/CDC via `DefaultInputSplitAssigner`, COW via the locality-aware `LocatableInputSplitAssigner` — pulling splits as readers finish rather than pinning them. And `DefaultHoodieSplitProvider` itself was a single shared queue before #18082, which introduced per-subtask assignment for streaming distribution parity and applied it to the bounded branch as well. This PR restores shared pulling for bounded only. Thanks @cshuo for both data points. **Tests** - `TestGlobalHoodieSplitProvider` (new, 15 cases): work stealing across arbitrary subtask ids, a single subtask draining the whole pool, oldest-commit-first ordering regardless of requester, `onUnassignedSplits` returning a split that a *different* subtask claims, checkpoint state round-trip, `isAvailable()` completion, and a concurrent 8-thread drain of 500 splits asserting each split is served exactly once. - `TestHoodieStaticSplitEnumerator` (+3): work stealing at the enumerator level; no-more-splits fires only when the pool is globally drained; and the failure case @danny0405 asked for — `addSplitsBack` *after* another reader has already received `NoMoreSplits`, asserting the returned split lands in the shared pool and is claimed by a third subtask that has neither failed nor finished. - `TestHoodieSourceEnumeratorRouting` (new, 16 cases): parameterized over the five bounded modes above, asserting fresh creation and restore both produce `HoodieStaticSplitEnumerator` + `GlobalHoodieSplitProvider`; that restore replays exactly the checkpointed splits and does not re-run discovery; and, parameterized over the requesting subtask 0-3, that a restored pending split goes to whichever subtask asks (under pinning only the one subtask its file id hashes to could ever receive it). Streaming fresh and restore are asserted to still produce `HoodieContinuousSplitEnumerator` + `DefaultHoodieSplitProvider`. ### Impact Performance only for bounded Source V2 reads; no config, no API change, no change to checkpoint contents or format. Same table, partition, and parallelism as the run above, with 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. Answering @danny0405's question on the intervals: **wall clock** is job submission to job `FINISHED`, so it includes resource allocation, split discovery, DAG deployment and teardown; **reader minutes** are per-subtask, first record to last record. That is the 228 vs 203 min gap in the pinned run (~25 min of setup); the shared run's 166 / 166 line up because its setup overlapped the read. 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 rather than by accumulated imbalance — stealing cannot preempt a split already being read, so one pathologically large file group remains the only exposure. Checkpoint size and restore semantics are unchanged: the enumerator still snapshots the same set of pending splits, in 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. ### Risk Level low Scoped to the non-streaming branch of `HoodieSource.createEnumerator`; streaming keeps `DefaultHoodieSplitProvider` and the existing assigners byte for byte. The enumerator, split serialization and checkpoint state are untouched. Restore is the one place this could regress quietly, since the provider used to be built before the streaming/bounded branch, so both fresh creation and restore are covered for every mode, including the failed-reader path after other readers have finished. Verified with the unit tests above plus the existing `TestHoodieSource`, `TestDefaultHoodieSplitProvider`, `TestHoodieContinuousSplitEnumerator` and `TestHoodieEnumeratorStateSerializer` suites, and the `read.source-v2.enabled` batch-read integration tests in `ITTestHoodieDataSource`. ### Documentation Update none — no new config and no user-facing behaviour change beyond the scheduling of bounded reads. ### Contributor's checklist - [x] Read through [contributor's guide](https://hudi.apache.org/contribute/how-to-contribute) - [x] Enough context is provided in the sections above - [x] Adequate tests were added if applicable -- 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]
