fix(core): reject a second INSERT_OVERWRITE of a partition already overwritten by a concurrent writer - #19720
Conversation
…erwritten by a concurrent writer Under OCC, SimpleConcurrentFileWritesConflictResolutionStrategy only sees a conflict on shared file ids. Two INSERT_OVERWRITE writers that both planned against an empty partition replace nothing and write disjoint file ids, so both commit and every row in the partition is present twice; clean, clustering and compaction then keep both copies. Treat two overwrite operations of a common partition as conflicting once one of them has completed (first to commit wins, as elsewhere in OCC). Pending overwrites keep file-id semantics so an abandoned pending overwrite cannot block the partition. Tests, all red on the old predicate: the predicate itself (TestSimpleConcurrentFileWritesConflictResolutionStrategy), two SparkRDDWriteClients (TestConcurrentInsertOverwriteSamePartition), and two Spark SQL sessions running INSERT OVERWRITE ... PARTITION concurrently (TestConcurrentInsertOverwritePartition).
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19720 +/- ##
============================================
+ Coverage 77.88% 77.90% +0.01%
- Complexity 33219 33230 +11
============================================
Files 2533 2533
Lines 140218 140227 +9
Branches 16887 16891 +4
============================================
+ Hits 109215 109240 +25
+ Misses 23369 23355 -14
+ Partials 7634 7632 -2
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 closes a real OCC gap where two concurrent INSERT_OVERWRITE / INSERT_OVERWRITE_TABLE writers that both plan against an empty partition write fresh file ids and both commit, duplicating the partition; it adds a partition-level conflict check once one overwrite has completed. I traced the argument convention in TransactionUtils, the mutated-partition derivation in ConcurrentOperation.init() (which includes written write-stat partitions, so empty-partition overwrites are still detected), the asymmetric completed-only guard, and the additive placement after the file-id check — the logic holds up and the scoping (pending-skip, overwrite-vs-overwrite only) is deliberate and tested. 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 small readability nits in the new test files.
cc @yihua
|
|
||
| HoodieTableMetaClient metaClient = HoodieTestUtils.createMetaClient(storageConf, basePath); | ||
| HoodieTimeline replaceTimeline = metaClient.getActiveTimeline().getCompletedReplaceTimeline(); | ||
| assertEquals(Arrays.asList(firstInstant), |
There was a problem hiding this comment.
🤖 nit: Collections.singletonList(firstInstant) would be slightly more idiomatic here than Arrays.asList with a single element — makes the intent (exactly one element) immediately clear.
| s"select count(distinct rank_source_id, locode, device, normalized_query, serp_item_position) from $tableName where day = '$targetPartition'") | ||
| .head().getLong(0) | ||
|
|
||
| println(s"overwrites committed=${completedOverwrites.map { case (i, md) => s"$i replaced=${md.getPartitionToReplaceFileIds}" }} " + |
There was a problem hiding this comment.
🤖 nit: could you swap println for a logger (e.g. log.info(...))? Raw println gets swallowed by test frameworks in CI and mixed with unrelated output, making it harder to trace a failure.
4 participants
Describe the issue this Pull Request addresses
Under OCC,
SimpleConcurrentFileWritesConflictResolutionStrategy.hasConflictonly reports a conflict when two operations share a (partition, file id) pair. TwoINSERT_OVERWRITEwriters that both planned against an empty partition replace nothing and write fresh file ids, so both commit and every row of the partition is present twice; clean, clustering and compaction then keep both copies.INSERT_OVERWRITEhas partition semantics but OCC only records it as file ids.Summary and Changelog
Two overwrite operations (
INSERT_OVERWRITE/INSERT_OVERWRITE_TABLE) that touch a common partition now conflict once the other one has completed, regardless of file ids. Scope, deliberately narrow:Known conservative case: an
INSERT_OVERWRITEof partition P that planned after a concurrentINSERT_OVERWRITE_TABLEcompleted is now rejected as well (the reverse ordering already was, via file ids). Not in this PR:BucketIndexConcurrentFileWritesConflictResolutionStrategyoverrideshasConflictentirely and does not get this rule.Changes:
SimpleConcurrentFileWritesConflictResolutionStrategy.hasConflict: partition-level check for a completed overwrite of a common partition.TestSimpleConcurrentFileWritesConflictResolutionStrategy: predicate-level test (same partition / pending other / different partition / non-overwrite).TestConcurrentInsertOverwriteSamePartition: twoSparkRDDWriteClients overwriting the same empty partition, plus a populated-partition contrast.TestConcurrentInsertOverwritePartition: two Spark SQL sessions runningINSERT OVERWRITE ... PARTITIONconcurrently.No code copied.
Impact
A second concurrent
INSERT OVERWRITEof the same partition now fails withHoodieWriteConflictExceptioninstead of silently duplicating the partition's data. No API or config change. Inherited by strategies that subclassSimpleConcurrentFileWritesConflictResolutionStrategywithout overridinghasConflict.Risk Level
Low. Tests at three levels, all red on the old predicate (see Changelog).
TestHoodieClientMultiWriterpasses unchanged.Documentation Update
none
Contributor's checklist