batched clean plans for very large partition tables - #19377
Closed
ehurheap wants to merge 742 commits into
Closed
Conversation
…mmit metadata (apache#10837) * Fixing handling null values of extra metadata in clean commit metadata * fixing tests
…reserveCommitMetadata field (apache#8393)
…submit job (apache#7074) Co-authored-by: Y Ethan Guo <ethan.guoyihua@gmail.com>
…platform-service/hudi-metaserver/hudi-metaserver-server (apache#7674) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
…e/hudi-metaserver/hudi-metaserver-server (apache#7673) Signed-off-by: dependabot[bot] <support@github.com> Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
* [HUDI-1517] create marker file for every log file (apache#4913) (apache#524) --------- Co-authored-by: guanziyue <30882822+guanziyue@users.noreply.github.com> Co-authored-by: Lokesh Jain <ljain@apache.org>
Signed-off-by: studystill <chenghuiyue@outlook.com>
… path (apache#10836) * get rid of collect in row writer clustering * fix race condition * add logging --------- Co-authored-by: Jonathan Vexler <=>
Co-authored-by: Vova Kolmakov <vlvkolmakov@nsk.beeline.ru>
…rementalFetchInContinuousMode (apache#10867) Co-authored-by: Vova Kolmakov <kolmakov.vladimir@huawei-partners.com>
…en building clustering plan (apache#10621) * Get partitions from active timeline instead of listing when building clustering plan * fix checkstyle
Co-authored-by: Vova Kolmakov <kolmakov.vladimir@huawei-partners.com>
…og sync (apache#10460) * Add parallel listing of existing partitions * Improve with new approach * Fix checkstyle * Fix listing for empty list of commits * Fix logic for HiveSyncTool * Fix lint errors * Fix IT * Use custom thread names * Address review comments --------- Co-authored-by: vmakarevich <vitali.makarevich@instructure.com>
apache#10870) Co-authored-by: Vova Kolmakov <kolmakov.vladimir@huawei-partners.com>
…rtition or multi primary key tables creation (apache#10840)
Co-authored-by: 含风 <qidian.qd@alibaba-inc.com>
…s tests (apache#10889) Co-authored-by: Vova Kolmakov <vlvkolmakov@nsk.beeline.ru>
Co-authored-by: Vova Kolmakov <kolmakov.vladimir@huawei-partners.com>
…ors and HoodieTables (apache#10908) Co-authored-by: Vova Kolmakov <kolmakov.vladimir@huawei-partners.com>
* support rollback only * update project version --------- Co-authored-by: pawel <pawel.sendyk@heapanalytics.com>
PA-9395: Parallelize partition deletes. More logging.
…restart before write" #23
Versioning hudi heap-fork
…bles Adds hoodie.clean.plan.partitions.batch.size (default -1, disabled). When set to a positive value, a single logical clean is paginated into multiple internally-batched clean instants, each covering at most N partitions. This avoids serializing one giant HoodieCleanerPlan Avro array and hitting the JVM Integer.MAX_VALUE byte-array ceiling on tables with millions of partitions (events-v1: ~6.7M). Design: * HoodieCleanConfig: new CLEANER_PLAN_PARTITIONS_BATCH_SIZE property and builder method. HoodieWriteConfig: matching getter. * CleanPlanActionExecutor: added BatchOverride carrying both the partition slice AND the pinned earliestInstant. Both must move together because CleanPlanner.getDeletePaths(partition, earliest) drives per-file delete decisions off earliestInstant; letting each batch recompute it would silently break the spec's "same earliestCommitToRetain across all batches" invariant. The existing constructor is retained as a delegating shim. Also fixes a pre-existing dead-code issue in the same file: the Option<Map<String,String>> extraMetadata constructor argument was stored on a field but never propagated into the plan. Now merged into the plan's extraMetadata alongside savepointed timestamps. Promoted requestClean(String) from protected to public so the batched-clean orchestrator can drive per-batch requests directly, bypassing execute()'s needsCleaning trigger gate (which would return false for batches 2..N since the just-completed batch 1 is itself the "last clean"). * BaseHoodieTableServiceClient: at the top of clean(...), when the new config is > 0, delegate to a new cleanInBatches() path. That path finishes any pending stragglers first (via CleanActionExecutor), computes earliestInstant + fullPartitionList exactly once, sorts the partitions for deterministic chunking, scans completed clean instants for extraMetadata matching this run's targetEarliestCommit + totalBatches to build the resume set, then iterates the remaining batches: for each, tags extraMetadata with batchIndex/totalBatches/ targetEarliestCommitToRetain, calls CleanPlanActionExecutor.requestClean(instantTime) directly, and runs CleanActionExecutor.execute() to complete that instant. * HoodieCleaner (utility): no change. The new config key is picked up automatically via --hoodie-conf / withProps, and client.clean() routes through the modified clean(String, boolean) path. Non-goals honored: partitionsToClean computation is unchanged, policy semantics are unchanged, and the incremental-mode/archiver interaction is untouched. Legacy behavior (batch.size=-1) is preserved. Tests to follow in a separate commit. 🤖 Generated with [Claude Code](https://claude.com/claude-code)
Starts the 0.2 development line. The 0.2 series will carry the opt-in paginated cleaner-plan batching feature. 🤖 Generated with [Claude Code](https://claude.com/claude-code)
…chOverride
Extract the pure logic backing cleanInBatches into two package-private
static helpers on BaseHoodieTableServiceClient so it can be exercised
without setting up a full HoodieTable / Spark stack:
chunkForBatchedClean(partitions, batchSize)
- deterministic natural-string sort before chunking so a restarted
job aligns on the same batchIndex as the original run
parseCompletedBatchIndex(extra, targetEarliest, totalBatches)
- identifies which completed .clean instants belong to the current
logical paginated run; rejects legacy instants, other-run
instants, and malformed bookkeeping
TestBatchedCleanHelpers covers the spec's testing-plan point 1:
- chunk sizing edge cases (empty, exact multiple, partial trailing,
smaller-than-batch, size-equal-to-batch)
- deterministic sort even when the input is reverse-ordered
- input list is not mutated
- batchSize <= 0 rejected
- parse rejects null/empty metadata, missing keys, malformed values,
mismatched targetEarliestCommit, mismatched totalBatches
- parse accepts a matching entry and returns the index
- a legacy (savepointed-only) extraMetadata is not mistaken for a
batched instant
- combined chunking + a simulated doneSet produces the expected
remaining slice with every index accounted for exactly once
TestCleanPlanActionExecutorBatchOverride covers the invariant that the
partition slice and earliestInstant are pinned together in one value
type — the spec's correctness requirement expressed as a source-level
invariant. End-to-end verification that requestClean() honors the
override is left to a future integration test (spec's testing-plan
point 3) where a real HoodieTable timeline is available; adding that
here would require the mock-heavy scaffolding of TestCleanPlanner and
still not exercise the actual persistence path.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
… and resume
Adds three tests to TestCleanPlanExecutor, covering the spec's
testing-plan point 3 against a real Hudi timeline (KEEP_LATEST_FILE_VERSIONS
policy, MDT enabled, HoodieMetadataTestTable-driven writes):
testCleanInBatchesDisabledPreservesLegacyPath
- batch.size = -1 (default) still routes through the legacy path:
exactly one .clean instant, no batching extraMetadata keys.
testCleanInBatchesProducesMultipleInstants
- 5 partitions with batch.size = 2 produces 3 completed .clean
instants. Every batch tags its extraMetadata with totalBatches,
targetEarliestCommitToRetain (shared across the whole logical
run — the spec's correctness invariant, checked here), and a
unique batchIndex in [0, totalBatches).
testCleanInBatchesResumeSkipsCompletedBatches
- Re-invoking clean when every batch of one logical run has already
completed is a no-op: the resume-detection scan sees each
batchIndex on the timeline and creates no new .clean instants.
This is the direct guarantee that a crashed-then-resumed run
does not reprocess completed batches.
Left for a follow-up (spec's testing-plan point 4): running against a
scaled-down clone of events-v1 to confirm the plan-size fix actually
avoids the OOM at ~6.7M-partition scale.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
ehurheap
marked this pull request as draft
July 25, 2026 00:14
ehurheap
changed the title
Collaborator
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
20 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
opt-in batched clean plans for very-large-partition tables