fix(metadata): use committed Spark write statuses after retries - #19801
fix(metadata): use committed Spark write statuses after retries#19801chrevanthreddy wants to merge 3 commits into
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19801 +/- ##
============================================
- Coverage 78.29% 78.23% -0.06%
+ Complexity 33850 33826 -24
============================================
Files 2541 2541
Lines 141674 141665 -9
Branches 17172 17172
============================================
- Hits 110917 110835 -82
- Misses 23048 23111 +63
- Partials 7709 7719 +10
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 working on this! The PR moves the coalescing shuffle ahead of streaming metadata generation so RLI/SI records are derived from the scheduler-selected data-table write statuses rather than raw pre-shuffle lineage. A few things are worth resolving before this goes further: the reworked method references an undefined variable (won't compile), and routing metadata derivation through the coalesced RDD looks like it collapses derivation parallelism at the default divisor — details in the inline comments, plus a note on whether the new regression test actually exercises the bug. Please take a look at any inline comments, and this should be ready for a Hudi committer or PMC member to take it from here.
| // Materialize data-table statuses behind a shuffle before deriving metadata records. Spark's | ||
| // scheduler exposes only the successful attempt for each shuffle partition, so failed or losing | ||
| // speculative attempts cannot contribute stale RLI/SI locations to the metadata write. | ||
| HoodieData<WriteStatus> successfulDataWriteStatuses = HoodieJavaRDD.of( |
There was a problem hiding this comment.
🤖 This won't compile as written — the local is declared successfulDataWriteStatuses here, but it's referenced as committedDataWriteStatuses on lines 68 and 71, which is never defined. Looks like a rename that got left half-applied; could you reconcile the two names?
| return coalescedDataWriteStatuses.union(mdtWriteStatuses); | ||
|
|
||
| HoodieData<WriteStatus> metadataWriteStatuses = | ||
| metadataWriter.streamWriteToMetadataPartitions(committedDataWriteStatuses, instantTime); |
There was a problem hiding this comment.
🤖 Deriving metadata from the coalesced RDD ties RLI/SI generation to coalesceParallelism = max(1, numPartitions/coalesceDivisor). With the default divisor of 5000 that's 1 partition for nearly all writes, so the whole writeStatus.flatMap(indexMapper) derivation is funneled through a single reducer, whereas before it ran at full data-write parallelism. For large writes that looks like a serious throughput/OOM regression. @nsivabalan @yihua is the metadata derivation intended to run at coalesced parallelism? Could we instead persist the pre-coalesce statuses (so both branches consume the same committed output) and coalesce only the union branch, keeping derivation parallelism intact?
| assertTrue(paths.stream().noneMatch(path -> path.equals("file-0-attempt-0"))); | ||
| assertTrue(metadataPaths.contains("file-0-attempt-1"), | ||
| "Metadata generation must consume the successful attempt"); | ||
| assertTrue(metadataPaths.stream().noneMatch(path -> path.equals("file-0-attempt-0"))); |
There was a problem hiding this comment.
🤖 Since attempt 0 throws before returning a WriteStatus, file-0-attempt-0 is never emitted by any attempt, so the noneMatch("file-0-attempt-0") checks hold trivially — this test would pass on master too and doesn't actually exercise the stale-lineage path. The bug you describe is a losing/slower speculative attempt that does emit output; could the test be shaped to reproduce that so it genuinely guards the regression?
4 participants
Describe the issue this Pull Request addresses
Spark can produce a data-table
WriteStatusfrom a task attempt that later fails or loses to a retry/speculative attempt. The streaming metadata path previously derived RLI/SI records from that raw pre-shuffle lineage, allowing a losing attempt's stale file location to leak into committed metadata.This is a core metadata-index correctness issue discovered while validating RFC-109, but the fix is independent of vector indexing.
Summary and Changelog
Spark streaming metadata writes now place the data-table statuses behind the existing coalescing shuffle before deriving RLI/SI metadata records.
local[8,2]and verifies both collected data output and metadata input contain only attempt 1.No code was copied.
Impact
Correctness fix for Spark writers using streaming metadata writes, particularly RLI and SI under task retry/speculation. No storage-format, public API, timeline-protocol, metadata-schema, or configuration change.
Risk Level
medium — this moves the existing coalescing shuffle ahead of streaming metadata generation. Mitigation:
git diff --checkpasses.Documentation Update
none
Contributor's checklist