fix(nbcc): reject insert overwrite combined with non-blocking concurrency control - #19791
fix(nbcc): reject insert overwrite combined with non-blocking concurrency control#19791Joy-2000 wants to merge 4 commits into
Conversation
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## master #19791 +/- ##
============================================
- Coverage 78.14% 74.06% -4.09%
+ Complexity 33692 32040 -1652
============================================
Files 2540 2541 +1
Lines 141413 141588 +175
Branches 17123 17182 +59
============================================
- Hits 110513 104864 -5649
- Misses 23200 29482 +6282
+ Partials 7700 7242 -458
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
|
+1 on this to fail loudly. IIUC, this is only relevant to simple bucket index (i.e. index based bucket number) right? |
|
@voonhous Yeah, non-blocking concurrency control requires the MOR table with simple bucket index. |
voonhous
left a comment
voonhous
left a comment
There was a problem hiding this comment.
Inline comments below. Two notes that do not anchor to a diff line:
- The Spark RDD path was already rejected before this PR:
BaseSparkBucketIndexBucketInfoGetter.java:40has thrown this exact message since HUDI-8866 (a93c6eec6788). What the new Spark guard adds is failing beforeinitTable/startCommit(the old check fires after the requested and inflight replacecommit and the workload-profile shuffle) and covering the row-writer bulk_insert overwrite path, which had no check. Could the description say so, so the partitioner check is not read as missed? - nit, optional: the
@EnumFieldDescriptiononNON_BLOCKING_CONCURRENCY_CONTROL(WriteConcurrencyMode.java:42-46) states no restriction at all, and the PR marks "Documentation Update: none". Would one clause ("MOR with simple bucket index only; insert overwrite is not supported") be worth adding there? While in that enum,isNonBlockingConcurrencyControl(String)at:65lacks theLocale.ROOTits sibling at:54uses, and this PR now calls it on every Spark write. - FYI only, pre-existing and out of scope: with
hoodie.datasource.write.row.writer.enable=false,insert overwrite+ bulk_insert goes to plainclient.bulkInsert(DataSourceUtils.java:192-195) andBULKINSERT_OVERWRITE_OPERATION_TYPEhas no reader in hudi-client, so the overwrite is silently dropped.
|
The existing Spark check is a drive-by from HUDI-8866 ( @danny0405 for a root cause fix discussion/visibility |
github-actions
Bot
added
size:L
|
Thanks for the detailed review @voonhous — all points addressed. Summary below. Engine-agnostic backstop. You're right that the per-engine guards left the Hudi Streamer path exposed. I moved the real safety net down to Flink single funnel. Consolidated the check to the top of Message literal → one validator. Went a step further than hoisting the string: the whole rule now lives in one place — Enum + Locale. Added the restriction clause to the Case sensitivity. Both scaladoc / ExceptionUtils. Restored the Tests (moved to their conventional home in
Description. Updated to state that the Spark RDD path already rejected this since HUDI-8866 ( On the broader invariant. Fully agree with you: the real invariant is a fixed-id file group must never be listed in a replacecommit, and Also noting your FYI on the |
|
@Joy-2000 verified all of the above at 7c36e9b and resolved the threads. Two notes:
Side note, could you please create an issue/dev-task for the root-cause fix by bumping the |
|
Deferring this PR to @danny0405 now, thank you for. the contribution, no other comments from me. |
|
@hudi-bot run azure |
danny0405
left a comment
danny0405
left a comment
There was a problem hiding this comment.
A couple of suggestions on validating this unsupported option combination before write-client work begins. The Spark SQL placement after deduceOperation and before handleSaveModes already matches that goal, including the row-writer overwrite flag. The comments below concern the remaining Flink and Spark Streamer timing. These are based on tracing the current head; I have not run additional tests for this review.
| // Validate the finalized write operation (after overwrite/static-partition resolution) at the | ||
| // single point all production callers (table sink, flink streamer, sink v2) funnel through. | ||
| OptionsResolver.checkNonBlockingConcurrencyControl(conf); |
There was a problem hiding this comment.
Could we also call OptionsResolver.checkNonBlockingConcurrencyControl(conf) at the very beginning of HoodieTableSink.getSinkRuntimeProvider(), before returning the DataStreamSinkProviderAdapter lambda?
By that point, the planner has applied applyOverwrite() and applyStaticPartition(), so conf contains the resolved operation. This gives the Table/SQL path a single early validation point for both explicit write.operation options and SQL INSERT OVERWRITE, without adding separate checks in HoodieTableFactory.sanityCheck() and applyOverwrite().
The placement before the lambda matters: the lambda performs client-ID setup and calls StreamerUtil.initTableFromClientIfNecessary() before reaching this pipeline guard. With incremental job-graph initialization or partition-level simple bucket indexing, the latter can create the table before the invalid combination is rejected.
Please keep this shared pipeline guard as well, since HoodieFlinkStreamer and direct pipeline callers bypass HoodieTableSink. A sink-level test could assert that getSinkRuntimeProvider() itself rejects the resolved overwrite operation under NBCC, without invoking the returned provider.
Updated to recommend this single sink-level check instead of my earlier factory-plus-applyOverwrite() suggestion.
| // Engine-agnostic backstop rejecting insert overwrite under non-blocking concurrency control. | ||
| // Complements the engine-specific guards by covering entry points that bypass them (e.g. Hudi | ||
| // Streamer, which reaches the write client directly). | ||
| WriteConcurrencyMode.checkInsertOverwriteSupported( | ||
| config.getWriteConcurrencyMode().isNonBlockingConcurrencyControl(), | ||
| WriteOperationType.isOverwrite(writeOperationType)); |
There was a problem hiding this comment.
Could we add an early Spark HoodieStreamer option check alongside this backstop? StreamSync.writeToSinkAndDoMetaSync calls startCommit before writeToSink dispatches to insertOverwrite/insertOverwriteTable, and those client methods call initTable before reaching preWrite. This guard prevents the overwrite, but the invalid NBCC configuration has already caused client initialization and commit-instant creation, unlike the Spark SQL guard. StreamSync.getHoodieClientConfigAndWriterSchema already has the built HoodieWriteConfig and cfg.operation; calling the shared validator immediately after builder.build() would reject both overwrite variants before constructing the write client. Please use cfg.operation rather than assuming the datasource operation property is present. Keep this preWrite check as protection for direct client callers, and add a Streamer test asserting that rejection creates no new timeline instant.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR rejects insert overwrite under non-blocking concurrency control (which reuses the deterministic bucket file id but emits a replacecommit that hides the freshly-written data by file id → silent data loss), centralizing the rule in WriteConcurrencyMode.checkInsertOverwriteSupported and wiring it into an engine-agnostic preWrite backstop plus fail-fast guards at the Spark and Flink entry points. I traced the overwrite paths across Spark SQL, the DataSource writer, the row-writer bulk_insert path, and Flink, and the guard placement lines up precisely with where the replacecommit hazard occurs; the DELETE_PARTITION and delete-and-recreate paths are correctly left untouched. No new issues flagged from this automated pass beyond what other reviewers have already raised — a Hudi committer or PMC member can take it from here for a final review. A couple of small naming and comment-formatting nits below.
cc @yihua
|
|
||
| /** | ||
| * Returns whether the operation is INSERT OVERWRITE dynamic partition. | ||
| */ |
There was a problem hiding this comment.
🤖 nit: could you rename this to something like checkInsertOverwriteSupported(Configuration conf) to mirror the underlying WriteConcurrencyMode method? checkNonBlockingConcurrencyControl reads more like a predicate on the mode itself and doesn't hint at what the check is actually validating.
| + "by the query reader and the compactor. Insert overwrite is not supported in this mode.") | ||
| NON_BLOCKING_CONCURRENCY_CONTROL; | ||
|
|
||
| //Error message thrown when insert overwrite is combined with non-blocking concurrency control. |
There was a problem hiding this comment.
🤖 nit: missing space after // — //Error message should be // Error message to match the surrounding comment style.
6 participants
Describe the issue this Pull Request addresses
Under NBCC, insert overwrite reuses the deterministic bucket file id, but the replace commit it generates records that same file id as replaced. Because the file system view hides a replaced file group by file id (ignoring the replace instant), the freshly-overwritten data becomes invisible → silent data loss.
Summary and Changelog
Reject
insert overwritewhen non-blocking concurrency control (NB-CC) is enabled, and fail early — before any instant is created.Why this is unsafe. NB-CC requires a MOR table with a simple bucket index, where each bucket has one deterministic, fixed file id.
insert overwritereuses that same file id but generates a replacecommit that records the id as replaced. Because the file system view hides a replaced file group by file id (ignoring the replace instant), the freshly-overwritten data becomes invisible → silent data loss.What this PR does:
BaseHoodieWriteClient#preWrite, the single chokepoint every overwrite entry point funnels through (Spark SQL, the DataSource writer, Flink, and Hudi Streamer). Keyed onisNonBlockingConcurrencyControl() && WriteOperationType.isOverwrite(writeOperationType). This closes the Hudi Streamer path, which bypasses the writer-level guards and previously only tripped the late partitioner check.HoodieSparkSqlWriter) and Flink entry points for a friendlier early error. On Flink they're consolidated to a single call at the top ofPipelines#hoodieStreamWrite— the funnel for the table sink,HoodieFlinkStreamer --op insert_overwrite, and Sink V2 — replacing the duplicated checks inHoodieTableSinkandPipelinesV2.WriteConcurrencyMode.checkInsertOverwriteSupported(isNbcc, isOverwrite), which owns the message constant and throws a single exception type (HoodieException). All four throw sites route through it —preWrite,HoodieSparkSqlWriter,OptionsResolver, and the pre-existingBaseSparkBucketIndexBucketInfoGetterpartitioner check — replacing the previous mix ofHoodieExceptionandIllegalArgumentException.bulk_insertoverwrite and case-insensitive operation strings are handled viaWriteOperationType.isOverwrite(...)+Locale.ROOTnormalization in bothsparkSqlInsertIntoOperationanddeduceOverwriteConfig(the latter previously risked deleting the table path on an uppercase operation).Relationship to the existing check. The Spark RDD path already rejected this since HUDI-8866 (
BaseSparkBucketIndexBucketInfoGetter), but only after the requested/inflight replacecommit and the workload-profile shuffle. This PR adds (a) failing beforeinitTable/startCommit, and (b) covering the row-writer bulk_insert overwrite path, which had no check — so the pre-existing partitioner check is superseded, not missed.Underlying invariant (follow-up, not in this PR)
The root invariant is broader than insert overwrite: a fixed-id bucket file group must never be listed in a replacecommit.
delete_partitionfollowed by a re-insert into the same partition hits the identical trap. This PR is intentionally a narrow, safe guard forinsert overwriteonly (isOverwrite(...)excludesDELETE_PARTITION, since a standalonedelete_partitionunder NB-CC is valid and loses no data). The root-cause fix — bumping the-0generation innewBucketFileIdForNBCCon replace so re-inserts land on a fresh id — is deferred to a separate ISSUE for discussion.Impact
Users combining
insert overwritewith non-blocking concurrency control now get a clear error at submission time instead of silent data loss. No change to any existing valid write path.Risk Level
low
Adds a validation that rejects a previously data-losing combination; existing valid writes are unaffected.
Documentation Update
The
@EnumFieldDescriptiononNON_BLOCKING_CONCURRENCY_CONTROLnow states the restriction. No site docs change needed.Contributor's checklist
TestWriteConcurrencyMode; Spark positive + negative inTestInsertTable3; Flink inITTestHoodieDataSource/TestOptionsResolver)