Conversation
…able commit With error table write unification enabled, the streamer committed the error table and afterwards summed the error-table write-status RDD for the record counts. The commit releases that RDD, so the sum recomputed the bulk insert and wrote every error record a second time under the committed instant.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19940 +/- ##
============================================
+ Coverage 80.23% 80.31% +0.07%
+ Complexity 34747 34743 -4
============================================
Files 2546 2546
Lines 142608 142515 -93
Branches 17362 17356 -6
============================================
+ Hits 114419 114455 +36
+ Misses 20277 20159 -118
+ Partials 7912 7901 -11
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
…e regression test The only error-table writer in the tree is a stub whose commit stores the write-status RDD and releases nothing, so no test could observe the recompute this PR fixes. RddBackedErrorTableWriter drives a real SparkRDDWriteClient over a real Hudi table and releases both caches on commit the way a production error-table writer does. TestErrorTableWriteOnce asserts the record keys the error table holds, one test on the committer's ordering and one pinning the previous ordering that wrote every record twice.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! The PR materializes the error-table write statuses on the driver before the error-table commit releases the write-status RDD, so the later record count no longer re-evaluates the bulk-insert lineage and lands every error record a second time under the completed instant. I traced the persist/release mechanics through the write client and the fix ordering holds; one small question inline about the storage level of the fallback cache. 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. A few minor javadoc clarity nits; the core refactor (collect-then-commit, RDD→List propagation) is clean and well-tested.
cc @yihua
| JavaRDD<WriteStatus> writeStatusRDD = errorTableWriteStatusRDDOpt.get(); | ||
| boolean cacheHere = writeStatusRDD.getStorageLevel().equals(StorageLevel.NONE()); | ||
| if (cacheHere) { | ||
| writeStatusRDD.cache(); |
There was a problem hiding this comment.
🤖 cache() is MEMORY_ONLY, so a partition that doesn't fit (or gets evicted) is silently dropped and recomputed by the commit's evaluation — which is exactly the double write this PR is guarding against. Would persist(StorageLevel.MEMORY_AND_DISK()) (or the writer's hoodie.write.status.storage.level, default MEMORY_AND_DISK_SER) be a safer net here? I realize this mirrors the data-table cache() in StreamSync, so happy to leave it if you'd rather keep them consistent.
| * | ||
| * @param dataTableWriteStatuses Pre-collected data-table write statuses. Must not be null. | ||
| * @param errorTableWriteStatusRDDOpt Optional error-table write status RDD; only consulted | ||
| * @param errorTableWriteStatuses Error-table write statuses collected before the error table |
There was a problem hiding this comment.
🤖 nit: the rewrapped javadoc now reads as one run-on sentence ("...or empty when none were written when unification is enabled") which is ambiguous about what depends on unification being enabled. Might be worth splitting back into two sentences like the original.
… table Replaces the collect-based approach with the ordering the pre-apache#18750 validator used: sum the error-table write statuses on the executors while the error table's own write still has them cached, then commit. Reverts ErrorTableCommitter and SuccessfulRecordCounter to their previous shape.
yihua
changed the title
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for addressing the feedback! The PR now takes a simpler route: the record count moves ahead of the error-table commit and sums the write-status RDD directly via aggregate, so the count materializes the writer's cache and the commit reuses it, instead of collecting statuses on the driver through a separate collectAndCommit helper. Both of my prior comments are resolved — the javadoc for the error-table parameter is clear again, and the cache() storage-level question is moot since that fallback was removed along with collectAndCommit. That removal does raise one follow-on question inline: the new ordering relies on the pluggable writer's upsert returning a persisted RDD, which the BaseErrorTableWriter contract doesn't state, so a custom writer returning an unpersisted or derived RDD would still see the write evaluated twice. 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.
| // statuses are held only by the cache that commit releases, so reading them afterwards | ||
| // re-runs the write and lands every error record twice. | ||
| SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( | ||
| writeStatuses, errorTableWriteStatusRDDOpt, isErrorTableWriteUnificationEnabled); |
There was a problem hiding this comment.
🤖 This now relies on the writer's upsert handing back an RDD that is already persisted (the bulk insert's instant cache), but BaseErrorTableWriter#upsert is a pluggable EVOLVING API with no such contract. If a custom writer returns an unpersisted or derived RDD (e.g. a map over the bulk-insert result), aggregate evaluates the write once here and writeClient.commit evaluates it again — the same double-write this PR fixes. The previous revision's cacheHere fallback covered that; was dropping it intentional, or would a StorageLevel.NONE() check + persist around the count (like the data-table block just above) be worth keeping?
danny0405
left a comment
danny0405
left a comment
There was a problem hiding this comment.
Reviewed the current diff and traced write-status persistence/release through the writer and commit path. The count-before-commit ordering addresses the reported recomputation for the persisted write-status RDD. One regression-test gap is noted inline. Validation: source inspection and current CI results; no local test execution.
| SuccessfulRecordCounter.Counts counts = SuccessfulRecordCounter.compute( | ||
| new ArrayList<>(), Option.of(writeStatusRDD), true); | ||
| assertTrue(ErrorTableCommitter.commit(writer, Option.of(writeStatusRDD), true, | ||
| BASE_TABLE_INSTANT, Option.empty()), "error table commit should succeed"); |
There was a problem hiding this comment.
[P2] Exercise the production ordering in the regression test
This test calls SuccessfulRecordCounter.compute and ErrorTableCommitter.commit directly in the desired order, but never invokes StreamSync. Both helpers are unchanged from the base branch, so reverting the entire production change in StreamSync would leave both new tests passing, including the test that explicitly expects duplicates. Could we add a test that runs a streamer batch with this RDD-backed writer and write unification enabled, then asserts each error-table key appears once? That assertion should fail when only the StreamSync reorder is reverted, so the suite catches a recurrence of the actual bug.
5 participants
Describe the issue this Pull Request addresses
With
hoodie.errortable.write.unification.enabled=true,StreamSynccommits the error table and afterwards sums the error-table write statuses for the record counts. That ordering duplicates every error record.The error-table write-status RDD is kept alive only by a Spark cache. The bulk insert persists it in
BaseSparkCommitActionExecutor.updateIndexunderHoodieDataCacheKey.of(basePath, instantTime), and the error table's ownwriteClient.commitreleases it throughreleaseResources->SparkReleaseResources.releaseCachedData. A real error-table writer typically also drops its cache of the upstream error events insidecommit. So by the time the count runs, nothing backs the RDD, theaggregatere-evaluates its lineage, and the executors run the bulk insert a second time under an instant that has already completed. The re-run allocates fresh file ids, so the duplicates land in new file groups stamped with that instant and a snapshot read returns two rows per error record.The counts themselves stay correct, which is why this is invisible from the outside: the recomputed statuses still describe one write's worth of records.
This arrived with #18750. That PR's first cut counted before committing the error table. A later revision hoisted the error-table commit ahead of the validators so error records would survive a validator abort, and the count moved behind it as a consequence. In the same round the data-table RDD was given a cache and a driver collect while the error-table RDD was left as a lazy
aggregate, so the asymmetry went unnoticed.Summary and Changelog
StreamSync.writeToSinkAndDoMetaSynccounts records before committing the error table, restoring the ordering the pre-[RFC] Migrate HoodieStreamerWriteStatusValidator into the pre-commit validator framework #18750HoodieStreamerWriteStatusValidatorused. The error-table commit still runs ahead of the validators, so the property the reorder was introduced for is preserved. Steps renumbered accordingly.ErrorTableCommitterandSuccessfulRecordCounterare unchanged. The error-table statuses are still summed on the executors with a singleaggregate; nothing is materialized on the driver.RddBackedErrorTableWriteris an error-table writer backed by a realSparkRDDWriteClientover a real Hudi table, whosecommitreleases the write statuses it is given and the upstream error events. The onlyBaseErrorTableWriterimplementation in the tree is a stub that stores the RDD and releases nothing, so no existing test could observe this.TestErrorTableWriteOnceasserts the record keys the error table actually holds: one case on the fixed ordering, one pinning the previous ordering that wrote every record twice.Ordering, for review
The whole change is the position of the counting block.
ErrorTableCommitterandSuccessfulRecordCounterare byte-identical to the base branch.Before, in
writeToSinkAndDoMetaSync:After:
This restores the order
HoodieStreamerWriteStatusValidatorused before #18750, where the same two operations ran inside thewriteClient.commitcallback rather than ahead of it. Onrelease-1.2.0that isStreamSync.HoodieStreamerWriteStatusValidator.validate: the error-table sum, thenerrorTableWriter.commit(...), then the write-error gate, and only then doesSparkRDDWriteClient.commitproceed tocommitStats.aggregateStep 2 stays ahead of the validators, so the property #18750 introduced the reorder for (error records survive a validator-driven abort) is unchanged.
Two things that are deliberately not restored here, both introduced by #18750 and out of scope: the sequence runs before
writeClient.commitrather than inside its callback, so failures roll the inflight instant back explicitly; and the data-table statuses are collected inStreamSyncas fullWriteStatusobjects rather than inside the write client asSlimWriteStats.The error table is a separate table with its own write client, instant and
HoodieDataCacheKey, so its commit releases only its own write statuses. The data-table RDD is unaffected either way, which is why only error-table rows were ever duplicated.Impact
Error-table writes land once under write unification. No config, API or public behaviour change.
Risk Level
low. The change is a move of the counting block within one method; no helper signatures change. Red-then-green verified: with the count after the commit the new test fails with each of the 20 expected record keys appearing twice, and passes with the count before it.
Documentation Update
none
Contributor's checklist