fix(spark-sql): lay meta fields over the partial-update schema in the global-index merge - #19713
fix(spark-sql): lay meta fields over the partial-update schema in the global-index merge#19713linliu-code wants to merge 2 commits into
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## master #19713 +/- ##
============================================
+ Coverage 77.88% 77.95% +0.07%
- Complexity 33265 33453 +188
============================================
Files 2533 2539 +6
Lines 140348 140903 +555
Branches 16913 17110 +197
============================================
+ Hits 109304 109836 +532
+ Misses 23406 23402 -4
- Partials 7638 7665 +27
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 fixes MOR partial updates on the global-index merge path so meta fields are laid out against the partial-update schema (matching HoodieAppendHandle/BaseWriteHelper) instead of the full write schema, and it stops deriving the partition path from a partial record that doesn't carry the partition field. I traced the schema plumbing, config propagation, the partition-resolution guard, and the behavior-preserving full-schema/non-expression paths, and everything lines up. 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 few small readability suggestions below.
cc @yihua
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR fixes meta-field placement in the global-index partial-update merge by wrapping the merged record against its own (partial) schema so prependMetaFields' count inference lands the data at the right positions, and it derives the partition from the existing record when the partial schema omits a partition field. This round's changes are limited to naming cleanups (aligning the mergedSchema/mergedSchemaWithMetaFields locals to their parameter names) and an explanatory inline comment addressing prior-round nits; the rename is complete and consistent across the file with no behavior change. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
… global-index merge Under MOR partial updates the merge in HoodieIndexUtils produces a record carrying only the columns named in UPDATE SET, which is the intended contract: HoodieAppendHandle takes the partial schema as its writer schema, and BaseWriteHelper does the same for dedup. But the global-index merge wrapped that record against the FULL write schema, and HoodieAvroIndexedRecord#prependMetaFields infers the meta-field count as targetSchema.size() - record.size(). For a two-field record against a ten-field target that is 8, so JoinedGenericRecord treated eight slots as meta and placed the data at indices 8 and 9. A MERGE INTO with a partial UPDATE SET therefore failed with UnresolvedUnionException: Not in union ["null","long"]: 15.0 (field=ts) as amount's double landed in ts's long union, during Kryo serialization of the tagged records. Reachable on MOR with a global bloom or simple index, which are the only configurations that pre-merge at tagging time and then inspect the merged record: mayContainDuplicateLookup is tableType == MERGE_ON_READ for those two, while the record-index implementations pass false and short-circuit, and CoW defers the merge to the file rewrite. Wrap the merged record against its own schema instead. The partial schema is read from the same config the append handle and the dedup path read, resolved once in the caller rather than per record, and falls back to the write schema when partial updates are off, so a full merge result behaves exactly as before. That also makes prependMetaFields' existing count inference correct rather than needing its own change, so HoodieAvroIndexedRecord and JoinedGenericRecord are untouched. Second change, required with the first: inferPartitionPath was deriving a partition from the merged record, and KeyGenUtils#getPartitionPath substitutes the default partition for a field it cannot find, so the payload already carried dt=__HIVE_DEFAULT_PARTITION__ while both the incoming and existing records were in the real partition. Harmless only because the write died first; fixing the schema alone would have turned a loud failure into a silently mis-partitioned row. A merge cannot change a field the record does not carry, so the existing record's partition is correct by construction. Gated on partial updates so the full-schema paths keep resolving through the key generator, and the partition-field check strips the mandatory "field:TYPE" spec the custom key generators keep verbatim in getPartitionPathFields. Tests: five parameterized cases over GLOBAL_BLOOM, GLOBAL_SIMPLE, RECORD_INDEX and GLOBAL_RECORD_LEVEL_INDEX, plus a boundary case pinning the analysis-time rejection when the index updates the partition path. They assert the row values, _hoodie_partition_path, and the log block itself via validateLogBlock, which checks IS_PARTIAL and that the block schema is the meta fields over the assigned columns only. The two global bloom and simple cases fail on the parent commit with the exception above. validateLogBlock gained a defaulted partitionPath parameter so it can address a partitioned table.
…schema locals Addresses review nits. The non-partial-update call site passed a bare true for partitionResolvableFromRecord, which read as an unexplained literal; it now carries the parameter name inline. That branch merges against the full write schema, so the partition is always resolvable from the record. Renamed the mergedRecordSchema / mergedRecordSchemaWithMetaFields locals to mergedSchema / mergedSchemaWithMetaFields, matching the parameter names they are passed into.
b186a3a to
4603257
Compare
github-actions
Bot
added
size:M
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the follow-up! This round is limited to a variable rename (mergedRecordSchema → mergedSchema and its meta-fields counterpart) and an inline clarifying comment on the partitionResolvableFromRecord argument, both addressing prior nits. The renames are applied consistently at the declaration and the sole call site, with no change in behavior. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
voonhous
left a comment
voonhous
left a comment
There was a problem hiding this comment.
Round 1 on 4603257: the three major items first, minor ones to follow.
| Option<HoodieRecord<R>> mergedOpt = mergeIncomingWithExistingRecord( | ||
| incoming, existing, writerSchema, writerSchemaWithMetaFields, updatedConfig, | ||
| incoming, existing, writerSchema, writerSchemaWithMetaFields, | ||
| mergedSchema, mergedSchemaWithMetaFields, partitionResolvableFromRecord, updatedConfig, |
There was a problem hiding this comment.
The arm this feeds at line 616 (merged.newInstance(existing.getKey()), changed partition with update.partition.path=false) returns an untagged record: HoodieAvroRecord.newInstance(HoodieKey) sets the location to Option.empty(), unlike the sibling arm at 608 that calls tagRecord. Untagged means it is written as an insert: the key lands in a second file group unless the slice is still a small-file candidate (no log file, base under parquet.small.file.limit, SparkUpsertDeltaCommitPartitioner.java:107-116), where BaseSparkDeltaCommitActionExecutor.java:76-79 merges it through the COW handle and dedups by key, which is why TestMergeIntoTable.scala:393 passes. Pre-existing from #13830 and shipped (git tag --contains 7943e1e18849: release-1.1.0, 1.1.1, 1.2.0); with #19709 merged, a partial update set t.dt = s.dt, t.amount = s.amount reaches it too.
Given the release exposure, could tagRecord(merged.newInstance(existing.getKey()), existing.getCurrentLocation()) be its own one-line PR, backportable on its own, with a test that sets hoodie.merge.small.file.group.candidates.limit=0 and asserts a single file slice after the partition change?
| ("GLOBAL_RECORD_LEVEL_INDEX", Map( | ||
| "hoodie.index.type" -> "GLOBAL_RECORD_LEVEL_INDEX", | ||
| "hoodie.record.index.update.partition.path" -> "false", | ||
| "hoodie.metadata.enable" -> "true", | ||
| "hoodie.metadata.record.index.enable" -> "true")) |
There was a problem hiding this comment.
This cell is a runtime duplicate of RECORD_INDEX: SparkHoodieIndexFactory.java:65-67 maps both spellings to SparkMetadataTableGlobalRecordLevelIndex, reading the same hoodie.record.index.update.partition.path. The one cell where the two spellings diverge, and where a wrong __HIVE_DEFAULT_PARTITION__ would surface as a delete against the real partition plus a partial record inserted into the default one, is this spelling with update.partition.path=true, which the NOTE below names and skips; #19610 is the open PR adding the enum mapping. The flag is not even needed to reach that arm: while the RLI partition is not yet built, SparkMetadataTableGlobalRecordLevelIndex.java:79-88 falls back to GLOBAL_SIMPLE, whose flag defaults to true.
Could we make this cell the update.partition.path=true variant with a source dt that differs, asserting the record stays in its partition, and cite #19610 so the cell is retired when the mapping lands?
| } | ||
| } | ||
|
|
||
| // A partial update names only the columns being changed, so the record key is normally absent |
There was a problem hiding this comment.
The existing test at lines 611-681 is the nearest shape to #19712 (partitioned MOR, GLOBAL_SIMPLE, partial UPDATE SET, mixed INSERT, partition change) and has no assertions at all; it also runs with the default update.partition.path=true, so partial updates are off in it, which is why it never caught this, and line 634 sets hoodie.index.global.index.enable, which is not a config anywhere in the tree. Relatedly, TestMergeIntoTable.scala:1496-1498 has its MOR rows commented out citing HUDI-8835, fixed 2025-05-15 by #13263; that table is partitioned, partial-update, has no preCombineField and already sets the small-file limit to 0.
Could we give the test here a checkAnswer on the rows plus _hoodie_partition_path, set update.partition.path=false and drop the bogus key, and re-enable those two MOR rows in TestMergeIntoTable, so the mixed-clause, compaction and commit-time-ordering shapes are covered without another cell?
There was a problem hiding this comment.
minor -- body edits: could we keep Closes #19712 but add a "Not covered" line naming #19610 for GLOBAL_RECORD_LEVEL_INDEX + update.partition.path=true with dt assigned; say that master already carries #19709 without this fix (the silent type-aligned shape in #19709's Risk section is live today); add the release exposure (#19712: present on 1.1.1 and 1.2.x) and that #19709 must not be backported without this; "two methods in one file" is three widened signatures, two new helpers and a hoisted block; "five" cases is four; cite #16854 next to the multi-clause caveat and #5942 as the 2022 prior report; "checkstyle and scalastyle"?
Outside this PR, for awareness: with the default merge.small.file.group.candidates.limit=1, an INSERT clause in the same partition opens a small-file UPDATE bucket at the existing fileId and BaseSparkDeltaCommitActionExecutor.java:76-79 diverts the tagged partial updates sharing it to the COW merge handle, which reads them with the full schema; every partial test forcing the limit to 0 is the tell. Code-traced, not reproduced; worth its own issue.
| existingRecordContext, mergeResult, partitionResolvableFromRecord); | ||
| HoodieRecord<R> result = existingRecordContext.constructHoodieRecord(mergeResult, partitionPath); | ||
| HoodieRecord<R> withMeta = result.prependMetaFields(writeSchema, writeSchemaWithMetaFields, | ||
| HoodieRecord<R> withMeta = result.prependMetaFields(mergedSchema, mergedSchemaWithMetaFields, |
There was a problem hiding this comment.
major -- generic guard behind this fix. HoodieAvroIndexedRecord.prependMetaFields:206 infers metaFieldSize = target.size - record.size (positional since #13860, in 1.1.0 through 1.2.0) and nothing bounds it; #19712's shape gives 8 and JoinedGenericRecord writes the shifted slots silently. Seven of the eight callers are pinned to 0/5/6; HoodieBootstrapRecordIterator.java:67 is the exception and is already malformed when a bootstrap writer schema gained more than one column.
Could we add checkArgument(0 <= metaFieldSize && metaFieldSize <= HOODIE_META_COLUMNS.size() + 1) there, so the next mismatched caller fails loudly instead of writing shifted values?
| @@ -377,6 +378,9 @@ private static <R> Option<HoodieRecord<R>> mergeIncomingWithExistingRecordWithEx | |||
| HoodieRecord<R> existing, | |||
| HoodieSchema writeSchema, | |||
There was a problem hiding this comment.
nit, feel free to ignore -- writeSchema is now unused in this method: both former uses (prependMetaFields and Option.of(writeSchema)) became mergedSchema. Could we drop the parameter and the argument at line 482?
| * mandatory "field:TYPE" spec verbatim in getPartitionPathFields, so the type suffix is stripped before | ||
| * the nested path is reduced to its root. | ||
| */ | ||
| private static String partitionFieldRootName(String partitionPathField) { |
There was a problem hiding this comment.
nit, feel free to ignore -- CustomAvroKeyGenerator.getPartitionFieldAndKeyType(field).getLeft() already strips the :TYPE suffix, and the root split is split("\\.")[0], which would avoid re-adding the HoodieAvroUtils import that #17599 removed from this file. Separately, inferPartitionPath (405) and this guard (565) are handed mergedSchemaWithMetaFields while the record carries mergedSchema; inert on the Avro path, but could we pass mergedSchema so the argument matches the record?
| // config. That combination therefore reaches the writer WITH partial updates enabled. The | ||
| // partition-resolution guard in HoodieIndexUtils covers it, and the missing enum mapping is being | ||
| // added separately, so this test asserts the rejection only for the spellings that actually reject. | ||
| test("Test MOR merge without assigning the record key is rejected when the global index updates the partition path") { |
There was a problem hiding this comment.
minor -- this is a near-clone of the test at line 686: same table shape, same statement, same checkExceptionContain on the message from MergeIntoHoodieTableCommand.scala:1051, differing only in which isPartialUpdateActionForMOR conjunct fails (custom merge mode there, useGlobalIndex here). The insert into is dead, since validate runs at the top of run() before any read, and the rejection is in run(), not analysis.
Could we fold this into 686 as a two-element Seq(...).foreach, or drop it?
| // already-working path is guarded against regression. The source projects the partition column in | ||
| // every case, to keep this independent of partition-column resolution (ENG-46864). | ||
| Seq( | ||
| ("GLOBAL_BLOOM re-keying", Map( |
There was a problem hiding this comment.
minor -- the four cells are two behaviours. GLOBAL_BLOOM and GLOBAL_SIMPLE call tagGlobalLocationBackToRecords with identical flags (HoodieGlobalBloomIndex.java:106-109, HoodieGlobalSimpleIndex.java:77-80), and both record-index spellings map to SparkMetadataTableGlobalRecordLevelIndex (SparkHoodieIndexFactory.java:65-67). Also hoodie.metadata.record.index.enable is a deprecated alias of hoodie.metadata.global.record.level.index.enable; if it stops resolving, the index falls back to GLOBAL_SIMPLE on a warning and the cell stays green.
Could we collapse to two cells and use the current key (or assert the record_index partition exists)?
| // set mayContainDuplicateLookup on MOR and so reach it, while the record-index spellings pass | ||
| // false and short-circuit. Both cells are covered, so the fix is pinned where it applies and the | ||
| // already-working path is guarded against regression. The source projects the partition column in | ||
| // every case, to keep this independent of partition-column resolution (ENG-46864). |
There was a problem hiding this comment.
nit, feel free to ignore -- comment accuracy: this paragraph describes the #19708 key-generator problem rather than the meta-field layout this PR fixes; ENG-46864 is an internal tracker id; the validateLogBlock note at 805-809 overclaims, since it reads only the block header whose schema comes from config on both sides of the fix (the discriminator is the merge no longer throwing); the "re-keying" labels say the opposite of update.partition.path=false; "short-circuit" should read "skip the merged lookup" (those cells do enter the stage); double blank line at 815; commit message and body say five cases for four.
Could we tidy these in one pass?
5 participants
Describe the issue this Pull Request addresses
Closes #19712
#19709 is merged, so this PR is rebased onto master and its diff is now only this change: two methods in
HoodieIndexUtilsand the test file.Summary and Changelog
Under MOR partial updates the merge in
HoodieIndexUtilsproduces a record carrying only the columnsnamed in
UPDATE SET. That is the intended contract:HoodieAppendHandletakes the partial schema asits writer schema and
BaseWriteHelperdoes the same for dedup. But the global-index merge wrappedthat record against the full write schema, and
HoodieAvroIndexedRecord#prependMetaFieldsinfersthe meta-field count as
targetSchema.size() - record.size(). For a two-field record against aten-field target that is 8, so
JoinedGenericRecordtreated eight slots as meta and the data landedat indices 8 and 9 rather than 7 and 8.
Wrap the merged record against its own schema instead, taken from the same config the append handle
and dedup path read, resolved once in the caller rather than per record, falling back to the write
schema when partial updates are off. That makes
prependMetaFields' existing inference correct, soHoodieAvroIndexedRecordandJoinedGenericRecordare untouched.Second change, required with the first:
inferPartitionPathwas deriving a partition from that samemerged record, and
KeyGenUtils#getPartitionPathsubstitutes the default partition for a field itcannot find, so the payload already carried
dt=__HIVE_DEFAULT_PARTITION__. Harmless only because thewrite failed first; fixing the schema alone would turn a loud failure into a silently mis-partitioned
row. A merge cannot change a field the record does not carry, so the existing record's partition is
correct by construction. Gated on partial updates so full-schema paths keep resolving through the key
generator, and the partition-field check strips the mandatory
field:TYPEspec that the custom keygenerators keep verbatim in
getPartitionPathFields.Tests: five parameterized cases over
GLOBAL_BLOOM,GLOBAL_SIMPLE,RECORD_INDEXandGLOBAL_RECORD_LEVEL_INDEX, plus a boundary case pinning the analysis-time rejection when the indexupdates the partition path. They assert row values,
_hoodie_partition_path, and the log block viathe existing
validateLogBlock, which checksIS_PARTIALand that the block schema is the metafields over the assigned columns only.
validateLogBlockgained a defaultedpartitionPathparameter so it can address a partitioned table.
Impact
Fixes
MERGE INTOwith a partialUPDATE SETon MOR with a global bloom or simple index. Nobehaviour change when partial updates are off, since both new schemas then resolve to the write schema
and the partition guard is gated off.
Two points a reviewer may want to push on, stated rather than left to be found:
The partial schema is read from
config.getPartialUpdateSchema().RecordContext#getSchemaFromBufferRecordwould give the merged record's schema directly and be correct by construction; it returned null when
tried, because each
RecordContextholds its own schema cache and the merger is built from theincoming context. The config route mirrors
HoodieAppendHandleandBaseWriteHelper, but the othershape may be preferable and is easy to swap if so.
Relatedly,
WRITE_PARTIAL_UPDATE_SCHEMAis the union across update clauses while each clause projectsonly its own assignments, so config-equals-record holds for the single-clause shape the tests cover.
A divergent multi-clause statement faults earlier today, inside the payload's serializer, so this
change does not introduce that gap, but it is not an invariant either.
Risk Level
low
Two methods in one file, gated so that only the partial-update path changes. Re-verified after the
rebase onto master, so the baseline below is plain master with #19709 in it rather than a stacked
parent: the two global bloom and simple cases fail there with the exception from #19712 and pass
here;
TestPartialUpdateForMergeInto,TestMergeIntoTableandTestMergeIntoTable2pass together(67 tests,
Expected test count is: 67); checkstyle clean.An earlier revision of the partition guard was not gated and checked the partition fields by raw name.
Since the custom key generators return
field:TYPEverbatim, that would have silently disabledpartition re-keying for every table using them, including full-schema merges unrelated to partial
updates. Both the gate and the suffix stripping exist to close that.
Documentation Update
none
Contributor's checklist