fix(spark): read shredded variants through the CDC and legacy streaming paths, and restore partition values there - #19583
Merged
Conversation
…ng paths Writing the coverage apache#19578 asked for surfaced three defects of the apache#19556 null-read family on the query paths that do not go through catalyst: - HoodieMergeOnReadRDDV2 base-only splits bypass the internal reader context and read through the plain skip-merging base reader, so a shredded base file with no logs read null variants on the legacy (hoodie.file.group.reader.enabled=false) streaming/relation path. Splits with variant columns now take the file-group reader branch instead. - CDCFileGroupIterator's BASE_FILE_INSERT case reads the new base file directly with the plain table schema, so CDC after-images of insert commits read null variants under every supplemental logging mode. The direct read now applies the same full-variant rewrite and restore as the reader context (helpers factored out of SparkFileFormatInternalRowReaderContext; context behavior unchanged). - InternalRowToJsonStringConverter had no VariantType case, so CDC images serialized the VariantVal bean (raw bytes as base64) instead of the variant's JSON. Variant columns now embed as real JSON nodes via VariantVal.toString. Tests: CDC round trip over a shredded COW table (insert + update, OP_KEY_ONLY and DATA_BEFORE_AFTER, shredded layout pinned; single-row table so the in-flight avro merge-path fix apache#19582 is not a prerequisite), and a legacy-path MOR streaming round trip covering both the base-only and merged split branches over a compacted shredded base file.
hudi-agent
reviewed
Aug 11, 2026
hudi-agent
left a comment
hudi-agent
left a comment
Contributor
There was a problem hiding this comment.
Thanks for working on this! This PR fixes shredded-variant null reads on three Spark paths that bypass the internal reader context — the legacy MOR base-only split, CDC's BASE_FILE_INSERT direct base-file read, and the CDC before/after JSON image converter — by factoring the full-variant rewrite/restore into shared helpers and applying them consistently. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
Member
Author
|
@wombatu-kun Can you please help to review this PR too? Thank you. |
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19583 +/- ##
============================================
+ Coverage 77.49% 77.53% +0.03%
- Complexity 32799 32887 +88
============================================
Files 2522 2524 +2
Lines 139179 139389 +210
Branches 16734 16781 +47
============================================
+ Hits 107855 108071 +216
- Misses 23748 23753 +5
+ Partials 7576 7565 -11
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
wombatu-kun
reviewed
Aug 11, 2026
Main code:
- HoodieMergeOnReadRDDV2: gate the base-only re-route on
sparkAdapter.buildFullVariantReadSchema(...).isDefined instead of the mere
presence of a variant column. It is None below Spark 4.1, where the file
group reader reads the same nulls, so re-routing only lost the fast path.
- HoodieMergeOnReadRDDV2: keep the fast path for splits carrying partition
values parsed off the partition path. Only requiredSchemaReaderSkipMerging
appends those (drop.partition.columns, extract-from-path, bootstrap fast
read); the file group reader branch builds its PartitionedFile with
InternalRow.empty and an empty partition schema, so re-routing such a split
would trade null variants for null partition columns. The same gap on the
merged branch predates this change and is left for a follow-up.
- InternalRowToJsonStringConverter: match the variant column on
dt.typeName rather than SparkAdapter.isVariantType. The guard runs for every
non-string/array/map/struct field, and resolving the adapter needs a version
module that is absent from hudi-spark-common's own test classpath: 13 of the
14 TestInternalRowToJsonStringConverter cases errored with
ClassNotFoundException: Spark4_2Adapter. Also fall back to the raw rendering
when readTree rejects the variant JSON (a non-finite double renders as a bare
NaN/Infinity token), so a CDC query degrades instead of failing.
Tests:
- TestInternalRowToJsonStringConverter: cover the JSON embedding and the
malformed-JSON fallback; suite goes 14/14 (was 13 errors before the fix).
- TestStreamingSource: write through format("hudi"), not the fully qualified
name. Only Spark4DefaultSource overrides supportsDataType to accept
VariantType and it is reachable solely via the registered short name, so the
write was dying with UNSUPPORTED_DATA_TYPE_FOR_DATASOURCE before the read.
Add the testLegacyIncrementalStreamSource plan guard so the test cannot pass
through a silent fallback to the file group reader. Correct the coverage
comment (the second batch is a log-only split, not a merged one) and add a
second testStream whose own checkpoint replays from INIT, which is what
actually produces a base + log slice on this path.
- TestVariantDataType: sweep the CDC round trip over shredded and unshredded
layouts rather than adding a second copy, so the rewrite branch is covered on
both sides. Pin the test to HoodieRecordType.SPARK: a cdc table always writes
through FileGroupReaderBasedMergeHandle, whose reader context follows the
merger's record type, and the AVRO leg's shredded base-file read is the
separate defect tracked as apache#19567/apache#19582. The previous claim that a single-row
table removed that prerequisite was wrong: what fails is the CDC before image
being written with value=null, which does not depend on carried-over rows.
Verified on Spark 4.1: TestVariantDataType 13/13 (2 cancelled as Spark-3 only),
TestStreamingSource 15/15. TestStreamingSource's suite-level leaked-file-stream
abort reproduces identically on the pre-PR revision, so it is not from this
change.
hudi-agent
reviewed
Aug 12, 2026
hudi-agent
left a comment
hudi-agent
left a comment
Contributor
There was a problem hiding this comment.
Thanks for the follow-up work here! This PR routes shredded-variant reads correctly through the legacy MOR streaming and CDC paths, fixing three null-read defects, and factors the full-variant rewrite/restore into shared helpers reused by the CDC BASE_FILE_INSERT direct read. I traced the reroute guard in HoodieMergeOnReadRDDV2, the copy-before-buffer handling in CDCFileGroupIterator, and the recursive variant handling in InternalRowToJsonStringConverter; the changed paths look internally consistent and the correctness concerns I can identify have all already been raised and addressed in the prior rounds. No new issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
github-actions
Bot
added
size:L
wombatu-kun
reviewed
Aug 12, 2026
…branch HoodieMergeOnReadRDDV2's file-group-reader branch sourced every projected column from the data files, so it returned NULL partition columns whenever those columns are not persisted there: drop.partition.columns, read-side extraction from the partition path, or a bootstrap data-queries-only read. SparkFileFormatInternalRowReaderContext builds its PartitionedFile with InternalRow.empty and an empty partition schema, and the builder's withPartitionPath only feeds the bootstrap merge, so nothing injected them. Only the skip-merging fast path ever appended them, via appendPartitionValues. The output row shape was already correct -- TableSchemaResolver re-appends dropped partition columns to the table schema, so the required schema keeps them and the reader merely fills them with nulls -- which makes this a value substitution at existing ordinals, with no column added, moved or dropped. - Carry the parsed values on HoodieMergeOnReadFileSplit, resolved on the driver by the same routine the skip-merging reader relies on. A log-only slice has no base file to carry them, so they come off a log file's path; that uses HoodieLogFile#getPath, since pathInfo is transient and null there. - Splice them in with one per-split UnsafeProjection over a JoinedRow. Bound references rather than literals: literals are inlined into generated code, so every distinct partition value would miss Spark's codegen cache. Mirrors HoodieFileGroupReaderBasedFileFormat.appendPartitionAndProject. - No-op unless it has to run: the split carries no values when the columns are read from the data files, the ordinal array is empty for a non-partitioned table (including the metadata table), and unprojected partition columns resolve to -1. The metadata-table Avro leg is untouched. With the branch fixed, the base-only re-route added earlier in this PR no longer has to avoid it, so its partition-value escape hatch is dropped and splits with shredded variant columns re-route unconditionally. Reachable today only from legacy MOR streaming (hoodie.file.group.reader.enabled =false); batch MOR always scans through HoodieFileGroupReaderBasedFileFormat, which appends partition values itself. Tests, in TestLegacyParquetReadPath since it already constructs the legacy relations directly: - MOR snapshot over drop.partition.columns with a partial upsert, so one file group merges base plus log while the others stay base-only. Verified red before this change (base-served p0 rows came back null) and green after. - MOR incremental over a log-only slice. This one passes with or without the splicing above, so it is not a repro: its rows come from log records, which do carry the partition column because HUDI-6926 makes a MOR upsert ignore drop.partition.columns. It pins the log-only resolution off HoodieLogFile#getPath and agreement with the file-group-reader oracle. Noted as such in the test.
Round-2 review pointed out the stated trigger was wrong, and it is: Spark guards the double and float arms of Variant.toJsonImpl with isFinite and sends the non-finite one through appendQuoted, so a variant renders NaN and Infinity QUOTED and readTree never sees a bare token. Verified against the bytecode and by rendering hand-built variants on Spark 4.0.2 and 4.1.1. The fallback still earns its place, for a different reason. Jackson's default StreamReadConstraints cap field names at 50k chars, strings at 20M and nesting at 1000 levels, while a variant may hold all three well inside its own 128MB limit, and castToVariant applies no such gate. All three were reproduced and all three arrive as StreamConstraintsException, a JsonProcessingException, so the existing catch covers them. - Reword the comment to name the real triggers. - Record why value.toString stays outside the try: it throws MALFORMED_VARIANT (a SparkRuntimeException, which JsonProcessingException does not cover) on corrupt bytes. That is a data-integrity signal an operator needs, not a rendering quirk to swallow into a CDC image, and there would be no rendering left to fall back to anyway. - Retarget the test, which asserted a rendering production cannot produce, onto over-deep nesting, and rename it accordingly. It expects the value as a JSON string, so it only passes when the fallback actually fires.
voonhous
changed the title
hudi-agent
reviewed
Aug 13, 2026
hudi-agent
left a comment
hudi-agent
left a comment
Contributor
There was a problem hiding this comment.
Thanks for working on this! The PR routes shredded-variant reads through the CDC and legacy MOR streaming/relation paths (HoodieMergeOnReadRDDV2, CDCFileGroupIterator, InternalRowToJsonStringConverter) and restores partition values on the file-group-reader branch. I traced the partition-value splicing (ordinal/JoinedRow binding, the empty/non-projected guards, and the checkState), the variant reroute condition, the CDC BASE_FILE_INSERT rewrite+restore+copy, and the null/exception handling in the JSON converter, cross-checking ordering against getPartitionFields and the file-group-reader oracle used by the new tests. No new correctness issues flagged from this automated pass beyond what earlier rounds already covered — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
wombatu-kun
reviewed
Aug 13, 2026
Round-3 review. The write-side gap is real: Jackson enforces a nesting cap on output too (StreamWriteConstraints, also 1000 levels), inside convert's writeValueAsString and therefore outside the branch's catch, so a variant deep enough to clear the read limit but not the write limit once the image's own object levels are added still failed the query. Embedding the validated rendering as a RawValue closes it: that goes out through JsonGenerator.writeRawValue, which keeps no nesting context at all. Measured while confirming this, on jackson 2.18.2/2.20.0/2.21.2: the window is one level per enclosing object for an object-rendered variant and one more for an array-rendered one, because jackson-core's WriterBasedJsonGenerator validates the PARENT depth in writeStartObject(Object) and so tolerates one extra level. A variant at the top of the image rendering as an array fails at exactly depth 1000, which is what the new test pins. RawValue is byte-identical to the parsed-node path for every rendering Variant.toJson can produce -- checked against the real renderer across decimals, doubles, escaping, key order and all scalar shapes -- and it drops a parse-then-rebuild round trip. It does mean unvalidated text would be spliced verbatim, so the gate is tightened to match: - FAIL_ON_TRAILING_TOKENS on the mapper, since readTree otherwise parses a valid prefix and leaves the rest unconsumed; - an explicit MissingNode check, since readTree answers blank input with MissingNode rather than throwing, and RawValue would emit nothing at all. Neither is producible by Variant.toJson, which always emits exactly one complete value; they are defence in depth for the verbatim embed. Also, from the same review: - MergeOnReadSnapshotRelation: the split's partition values are non-empty on any of shouldExtractPartitionValuesFromPartitionPath's three triggers, not only when the columns are omitted from the data files -- the extract-from-path read option applies to tables that persist them too. Comment corrected. - TestVariantDataType: the compaction test still pointed at apache#19578 as tracking the no-catalyst-schema legs, which this PR closes and covers. Repointed at the CDC round trip and the TestStreamingSource test. Verified: TestInternalRowToJsonStringConverter 16/16, with the depth test red before the RawValue change (StreamWriteConstraints exceeded) and green after; TestVariantDataType 13/13, so the CDC round trip still reads its images back through get_json_object.
Member
Author
|
@wombatu-kun Ready for another round of review. |
wombatu-kun
approved these changes
Aug 13, 2026
Collaborator
This was referenced Aug 24, 2026
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Describe the issue this Pull Request addresses
Closes #19578, and fixes #19594.
#19578 asked for streaming and CDC round-trip coverage of shredded variant reads; writing the coverage surfaced three real defects on those paths, all of the #19556 null-read family (requesting native VariantType against a shredded parquet base file clips the shredded group to
{metadata, value}and readsvalue=null). Review of the fix surfaced a fourth, unrelated defect on the same RDD, filed as #19594 and fixed here.HoodieMergeOnReadRDDV2base-only splits (dataFileOnlySplit) bypass the internal reader context entirely and read through the relation's plain skip-merging base reader, so a shredded base file with no log files read null variants. Hit by the legacy (hoodie.file.group.reader.enabled=false) streaming/relation path, e.g. any streaming batch over a freshly compacted slice.CDCFileGroupIterator's BASE_FILE_INSERT case reads the new base file directly with the plain table schema, bypassing the context, so CDC after-images of insert commits read null variants. Hit under every supplemental logging mode.InternalRowToJsonStringConverter(CDC before/after images) had no VariantType handling, so a variant column fell through to Jackson bean serialization of the rawVariantValbytes instead of the variant's JSON.HoodieMergeOnReadRDDV2's file-group-reader branch never appended partition values, so partition columns read back NULL whenever they are not persisted in the data files (drop.partition.columns, read-side extraction from the partition path, bootstrap data-queries-only). Pre-existing and independent of variants. It is fixed here rather than deferred because fix 1 re-routes base-only splits onto that same branch and would otherwise have widened the gap; carving it out would mean either shipping that regression or keeping a carve-out no test could reach.Summary and Changelog
HoodieMergeOnReadRDDV2: splits whose required schema has variant columns skip the plain fast reader and take the file-group reader branch, whose reader context requests the full-variant projection shape (fix(spark): read shredded variants through internal write-side parquet reads #19558). Gated onSparkAdapter.buildFullVariantReadSchema(...).isDefined, which isNonebelow Spark 4.1 where the file-group reader would read the same nulls. Base-only splits without variants are unchanged.HoodieMergeOnReadRDDV2: the file-group-reader branch now appends partition values. The split carries the values parsed on the driver by the same routine the skip-merging reader relies on (off a log file's path when the slice has no base file), and they are bound in with one per-splitUnsafeProjectionover aJoinedRow. The output row shape was already correct --TableSchemaResolverre-appends dropped partition columns, so the reader was merely filling them with nulls -- so this is a value substitution at existing ordinals, with no column added, moved or dropped. Inert unless it applies: no values on the split when the columns live in the data files, an empty ordinal array for a non-partitioned table (including the metadata table, whose Avro leg is untouched), and-1for unprojected columns.SparkFileFormatInternalRowReaderContext: the full-variant rewrite and restore projection are factored into reusable helpers (fullVariantReadSchemaWithOrdinals,variantRestoreProjection); behavior of the context itself is unchanged.CDCFileGroupIterator: BASE_FILE_INSERT applies the same rewrite/restore around its direct base-file read (with a defensive copy, since the restore projection reuses its output buffer).InternalRowToJsonStringConverter: variant columns are embedded into before/after images as real JSON nodes viaVariantVal.toString, soget_json_object(after, '$.v.key')works as expected. The column is matched onDataType.typeName, notSparkAdapter.isVariantType: the guard runs for every non-string/array/map/struct field, and resolving the adapter needs a version module that is not on hudi-spark-common's own test classpath.readTreefalls back to the raw rendering onJsonProcessingException, which is how a variant exceeding Jackson's defaultStreamReadConstraints(50k field names, 20M strings, 1000 nesting levels -- all reachable inside the variant size limit) arrives;VariantVal.toStringstays outside that block soMALFORMED_VARIANTon corrupt bytes still surfaces.Tests:
TestVariantDataType, sweeping shredded and unshredded layouts against OP_KEY_ONLY and DATA_BEFORE_AFTER, with the on-disk layout pinned per leg. Pinned toHoodieRecordType.SPARK, since a cdc table always writes throughFileGroupReaderBasedMergeHandleand the merger's record type picks that handle's reader context; the AVRO leg's shredded base-file read is the separate defect tracked as [BUG] CoW small-file merge silently nulls shredded VARIANT values via the avro reader path #19567.TestStreamingSourceover a compacted shredded base file, asserting the executed plan is the RDD-backedScan ExistingRDDso it cannot pass through a silent fallback to the file-group reader. One stream covers the base-only and log-only splits; a second, with its own checkpoint, replays from INIT so its first batch spans the compaction commit and the update together, which is the only way to land a base + log slice on this path.drop.partition.columnstable inTestLegacyParquetReadPath, through the directly-constructed legacy relations. The snapshot case uses a partial upsert so one file group merges base plus log while the others stay base-only, exercising the merging branch and the fast path in one query; verified red before fix 4 and green after. A second case covers a log-only slice.TestInternalRowToJsonStringConvertercovers the JSON embedding and the constraints fallback. Before fix 3's detection change, 13 of its 14 cases errored withClassNotFoundException: Spark4_2Adapter.Impact
Wrong-results fixes on non-default or CDC paths, plus one image-format improvement. CDC after-images of insert commits and legacy-path reads of base-only shredded slices returned null variants before this change; legacy-path MOR reads of a table whose partition columns live only in the partition path returned null partition columns. CDC images of variant columns now carry the variant's JSON structure; previously they carried a Jackson bean rendering of the raw bytes, which no reader could reasonably consume, so this is considered a fix rather than a breaking format change.
One deliberate behaviour change to call out: with
hoodie.datasource.read.extract.partition.values.from.path=trueon a table that does persist its partition columns, this branch previously returned the file-persisted value and now returns the partition-path-parsed one. That matches the skip-merging reader, the file-format path and the documented flag semantics, including theTimestampBasedKeyGeneratorcaveat noted onHoodieBaseRelation.shouldExtractPartitionValuesFromPartitionPath.Risk Level
low-medium. The variant routing change only triggers when the schema has variant columns and the adapter supports the rewrite; the CDC direct-read rewrite only rewrites on Spark 4.1+; the image converter change only affects variant columns. Fix 4 touches a shared read path, but
HoodieMergeOnReadRDDV2runs only for legacy MOR streaming (hoodie.file.group.reader.enabled=false) and directly-constructed relations -- batch MOR always scans throughHoodieFileGroupReaderBasedFileFormat, which appends partition values itself -- and it is a no-op unless the partition columns are absent from the data files.Documentation Update
none
Contributor's checklist