perf(flink): use a shared work-stealing split pool for Source V2 bounded reads - #19520
Conversation
ericyuan915
changed the title
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR introduces a shared work-stealing split pool (GlobalHoodieSplitProvider) for bounded Flink Source V2 reads so readers stay busy until the pool drains, while leaving the streaming path on per-subtask assignment. I traced the threading model (PBQ makes the concurrent pendingSplitCount() gauge read safe), the fresh/restore branches in createEnumerator (no double-discovery, checkpoint state format unchanged), the work-stealing drain and signalNoMoreSplits logic, and the addSplitsBack recovery path — the last is actually more robust than the existing Default provider's re-pinning. The one-split-per-file-group safety invariant is enforced by the added routing tests across all bounded modes. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of minor naming and visibility nits in GlobalHoodieSplitProvider; the rest of the code is clean and well-documented.
cc @yihua
| "Pending records is not supported in GlobalHoodieSplitProvider."); | ||
| } | ||
|
|
||
| private synchronized void completeAvailableFuturesIfNeeded() { |
There was a problem hiding this comment.
🤖 nit: the method name says Futures (plural) but there is only one availableFuture field and it completes exactly one future per call — could you rename it to completeAvailableFutureIfNeeded?
| * while the coordinator thread assigns. | ||
| */ | ||
| public class GlobalHoodieSplitProvider implements HoodieSplitProvider { | ||
| public static final int INITIAL_POOL_CAPACITY = 20; |
There was a problem hiding this comment.
🤖 nit: INITIAL_POOL_CAPACITY is public but it is only an internal queue-sizing detail with no meaningful contract for callers — would private (or at most package-private) be more appropriate here?
| // Reading from earliest triggers a full table scan, which bypasses the CDC branch, so the | ||
| // CDC mode starts from the last completed commit instead. | ||
| conf.set(FlinkOptions.READ_START_COMMIT, | ||
| mode.cdcEnabled ? lastCompletionTime() : FlinkOptions.START_COMMIT_EARLIEST); |
There was a problem hiding this comment.
COW_INCREMENTAL sets READ_START_COMMIT to earliest here, which makes IncrementalInputSplits.inputSplits() take its fullTableScan branch. As a result, this case does not exercise the non-full-scan bounded incremental path cited in the PR one-split-per-file-group safety argument. Could this use an actual completed commit (and, if needed, a bounded end commit) so that fullTableScan is false? A separate earliest case can remain if that fallback also needs coverage.
There was a problem hiding this comment.
Good catch, you're right — earliest leaves the analyzer's startInstant empty (IncrementalQueryAnalyzer:205, isConsumingFromEarliest() is startInstant.isEmpty()), so fullTableScan was true and that parameter never reached the metadata-driven branch the safety argument cites. Fixed in 4d55875.
COW_INCREMENTAL now starts from a real completed commit, and I added COW_INCREMENTAL_FROM_EARLIEST so the full-scan fallback stays covered — as you suggested, both sides of the branch.
I also wanted the parameter to fail loudly rather than silently stop covering the branch again, so the modes that start from a real commit now write a last commit touching only par5/par6. The metadata-driven branch derives its read partitions from that commit's metadata alone, while a full table scan lists par1 through par6, so asserting the split partitions pins which branch actually produced them:
if (mode.incrementalStart == IncrementalStart.LAST_COMMIT) {
assertEquals(new HashSet<>(Arrays.asList("par5", "par6")),
splits.stream().map(HoodieSourceSplit::getPartitionPath).collect(Collectors.toSet()),
"Mode " + mode + " should read only the partitions written by the start commit, "
+ "which is what distinguishes the incremental branch from a full table scan");
}I verified the assertion actually bites by temporarily pointing the start commit back at earliest: COW_INCREMENTAL and COW_INCREMENTAL_CDC both fail with expected: <[par5, par6]> but was: <[par5, par6, par1, par2, par3, par4]>.
For the record, all three incremental shapes end at the same getInputSplits(fileSlices, ...), one split per slice, and both branches source their slices from getLatestMergedFileSlicesBeforeOrOn; the difference is only how the partition/file set is derived. So the one-split-per-file-group claim held either way, but the test now demonstrates it on both paths instead of asserting it on one and claiming it for the other.
18 routing cases green, and 135 tests across the touched source suites with checkstyle clean.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19520 +/- ##
=========================================
Coverage 76.60% 76.61%
- Complexity 32238 32262 +24
=========================================
Files 2520 2521 +1
Lines 139052 139081 +29
Branches 16708 16710 +2
=========================================
+ Hits 106527 106560 +33
+ Misses 24913 24912 -1
+ Partials 7612 7609 -3
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the contribution! This PR adds a GlobalHoodieSplitProvider — a shared, work-stealing PriorityBlockingQueue-backed split pool for bounded (batch) Source V2 reads — and routes the enumerator so streaming keeps per-subtask affinity while bounded reads drain a single pool, eliminating the straggler tail. I traced the routing/restore logic in createEnumerator, verified the "one split per file group / no cross-commit continuation" safety invariant against IncrementalInputSplits.getInputSplits/getCdcInputSplits, and checked the coordinator-thread concurrency model and the addSplitsBack failover path; the design mirrors Flink's canonical bounded-source pattern and the tests exercise every bounded mode. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
|
@ericyuan915 can you rebase with the latest master to make the CI pass. |
…ded reads Bounded reads inherit the streaming split provider, which pins every split to one subtask at discovery and never rebalances, so a subtask that drew a heavier share keeps working while its peers sit idle. On a bounded COW backfill (~16.4K splits, parallelism 32) the readers finished 78 minutes apart and the job took 3.80 h instead of 2.77 h. The affinity pays for itself only in streaming, where a file group accumulates log files across commits and successive splits of one file id must stay on one reader. A bounded read has exactly one split per file group, no cross-commit continuation and no ordering relationship between splits, so any reader can read any split. Add GlobalHoodieSplitProvider, a single shared pool ordered by the existing HoodieSourceSplitComparator whose getNext ignores the subtask id, and select it on the non-streaming branch of HoodieSource.createEnumerator. Streaming keeps DefaultHoodieSplitProvider and the existing assigners. The provider is now chosen on the streaming/bounded branch rather than before it, so restore replays checkpointed splits into whichever provider was chosen. No enumerator change is needed: with one pool, getNext returning empty already means globally drained. Closes apache#19516
…t just full scan READ_START_COMMIT=earliest leaves the analyzer's startInstant empty, so IncrementalInputSplits.inputSplits() took its fullTableScan branch and the COW_INCREMENTAL parameter never reached the metadata-driven branch that the one-split-per-file-group argument cites. Split the parameter in two: COW_INCREMENTAL now starts from a real completed commit so fullTableScan is false, and COW_INCREMENTAL_FROM_EARLIEST keeps the full-scan fallback covered. To make the branch observable, the modes that start from a real commit write a last commit that only touches par5 and par6. The metadata-driven branch derives its read partitions from that commit alone, while a full table scan lists par1 through par6, so asserting the split partitions pins which branch produced them.
Head branch was pushed to by a user without write access
4d55875 to
44bb694
Compare
…ded reads (#19520) * perf(flink): use a shared work-stealing split pool for Source V2 bounded reads Bounded reads inherit the streaming split provider, which pins every split to one subtask at discovery and never rebalances, so a subtask that drew a heavier share keeps working while its peers sit idle. On a bounded COW backfill (~16.4K splits, parallelism 32) the readers finished 78 minutes apart and the job took 3.80 h instead of 2.77 h. The affinity pays for itself only in streaming, where a file group accumulates log files across commits and successive splits of one file id must stay on one reader. A bounded read has exactly one split per file group, no cross-commit continuation and no ordering relationship between splits, so any reader can read any split. Add GlobalHoodieSplitProvider, a single shared pool ordered by the existing HoodieSourceSplitComparator whose getNext ignores the subtask id, and select it on the non-streaming branch of HoodieSource.createEnumerator. Streaming keeps DefaultHoodieSplitProvider and the existing assigners. The provider is now chosen on the streaming/bounded branch rather than before it, so restore replays checkpointed splits into whichever provider was chosen. No enumerator change is needed: with one pool, getNext returning empty already means globally drained. Closes #19516 * test(flink): cover the metadata-driven bounded incremental branch, not just full scan READ_START_COMMIT=earliest leaves the analyzer's startInstant empty, so IncrementalInputSplits.inputSplits() took its fullTableScan branch and the COW_INCREMENTAL parameter never reached the metadata-driven branch that the one-split-per-file-group argument cites. Split the parameter in two: COW_INCREMENTAL now starts from a real completed commit so fullTableScan is false, and COW_INCREMENTAL_FROM_EARLIEST keeps the full-scan fallback covered. To make the branch observable, the modes that start from a real commit write a last commit that only touches par5 and par6. The metadata-driven branch derives its read partitions from that commit alone, while a full table scan lists par1 through par6, so asserting the split partitions pins which branch produced them. (cherry picked from commit 2490309) Adaptations for release-1.2.1: - HoodieSource keeps its LOG field. Master converted the class to Lombok @slf4j, so the diff's log.info collided with LOG.info. The restructure itself applies unchanged, and the per-file delta is identical to upstream. - Two tests in the TestHoodieStaticSplitEnumerator conflict region, testConstructorWithNullMetricGroup and testHandleSourceEventWithUnknownEventThrows, are master context rather than additions of this commit, so they are not taken. The first would not compile here: it needs a MockSplitEnumeratorContext(boolean) constructor that this branch's mock does not have. - Added a static import for assertFalse. The three new tests use it, but master already imported it for testConstructorWithNullMetricGroup above, so it arrived as context rather than as an added line. This is the only line in which the applied delta differs from upstream.
…ded reads (#19520) * perf(flink): use a shared work-stealing split pool for Source V2 bounded reads Bounded reads inherit the streaming split provider, which pins every split to one subtask at discovery and never rebalances, so a subtask that drew a heavier share keeps working while its peers sit idle. On a bounded COW backfill (~16.4K splits, parallelism 32) the readers finished 78 minutes apart and the job took 3.80 h instead of 2.77 h. The affinity pays for itself only in streaming, where a file group accumulates log files across commits and successive splits of one file id must stay on one reader. A bounded read has exactly one split per file group, no cross-commit continuation and no ordering relationship between splits, so any reader can read any split. Add GlobalHoodieSplitProvider, a single shared pool ordered by the existing HoodieSourceSplitComparator whose getNext ignores the subtask id, and select it on the non-streaming branch of HoodieSource.createEnumerator. Streaming keeps DefaultHoodieSplitProvider and the existing assigners. The provider is now chosen on the streaming/bounded branch rather than before it, so restore replays checkpointed splits into whichever provider was chosen. No enumerator change is needed: with one pool, getNext returning empty already means globally drained. Closes #19516 * test(flink): cover the metadata-driven bounded incremental branch, not just full scan READ_START_COMMIT=earliest leaves the analyzer's startInstant empty, so IncrementalInputSplits.inputSplits() took its fullTableScan branch and the COW_INCREMENTAL parameter never reached the metadata-driven branch that the one-split-per-file-group argument cites. Split the parameter in two: COW_INCREMENTAL now starts from a real completed commit so fullTableScan is false, and COW_INCREMENTAL_FROM_EARLIEST keeps the full-scan fallback covered. To make the branch observable, the modes that start from a real commit write a last commit that only touches par5 and par6. The metadata-driven branch derives its read partitions from that commit alone, while a full table scan lists par1 through par6, so asserting the split partitions pins which branch produced them. (cherry picked from commit 2490309) Adaptations for release-1.2.1: - HoodieSource keeps its LOG field. Master converted the class to Lombok @slf4j, so the diff's log.info collided with LOG.info. The restructure itself applies unchanged, and the per-file delta is identical to upstream. - Two tests in the TestHoodieStaticSplitEnumerator conflict region, testConstructorWithNullMetricGroup and testHandleSourceEventWithUnknownEventThrows, are master context rather than additions of this commit, so they are not taken. The first would not compile here: it needs a MockSplitEnumeratorContext(boolean) constructor that this branch's mock does not have. - Added a static import for assertFalse. The three new tests use it, but master already imported it for testConstructorWithNullMetricGroup above, so it arrived as context rather than as an added line. This is the only line in which the applied delta differs from upstream.
5 participants
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.
DefaultHoodieSplitProviderkeeps one queue per subtask and assigns on arrival, andgetNextserves 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
createEnumeratorbuilds 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) — aHoodieSplitProviderbacked by onePriorityBlockingQueueordered by the existingHoodieSourceSplitComparator(oldest commit first, the same ordering the per-subtask queues use).getNext(taskId, hostname)ignores both arguments.onUnassignedSplits(theaddSplitsBackpath) returns splits to the same pool.HoodieSource.createEnumerator— the provider is now chosen on the streaming/bounded branch rather than before it: streaming keepsDefaultHoodieSplitProviderplus 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,
getNextreturning empty already means "globally drained", soHoodieStaticSplitEnumerator's existingsignalNoMoreSplitslogic stays correct.Why the affinity is load-bearing for streaming but not for bounded.
DefaultHoodieSplitAssigneruses Flink's ownKeyGroupRangeAssignment.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;HoodieSplitBucketAssignersimilarly 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:FileIndexReader.baseFileOnlyHoodieSourceSplitsfsView.getLatestBaseFiles(par)yields one latest base file per file groupFileIndexReader.buildHoodieSplits→readFileSlicegetLatestMergedFileSlicesBeforeOrOnyields one merged slice per file groupIncrementalInputSplits.inputSplits→getInputSplitsgetLatestMergedFileSlicesBeforeOrOn, one split per sliceIncrementalInputSplits.getCdcInputSplitsMap<HoodieFileGroupId, List<HoodieCDCFileSplit>>, so one split per file group with the file group'schanges[]sorted by instant inside the split — cross-commit order is intra-split, never cross-splitTestHoodieSourceEnumeratorRoutingasserts 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.addFileDistributionStrategyis applied only to the V1DataStream<MergeOnReadInputSplit>monitoring stream, never toHoodieSource.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-awareLocatableInputSplitAssigner— pulling splits as readers finish rather than pinning them. AndDefaultHoodieSplitProvideritself 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,onUnassignedSplitsreturning 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 —addSplitsBackafter another reader has already receivedNoMoreSplits, 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 produceHoodieStaticSplitEnumerator+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 produceHoodieContinuousSplitEnumerator+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:
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 keepsDefaultHoodieSplitProviderand 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 existingTestHoodieSource,TestDefaultHoodieSplitProvider,TestHoodieContinuousSplitEnumeratorandTestHoodieEnumeratorStateSerializersuites, and theread.source-v2.enabledbatch-read integration tests inITTestHoodieDataSource.Documentation Update
none — no new config and no user-facing behaviour change beyond the scheduling of bounded reads.
Contributor's checklist