fix(avro): detect shredded variant base files by shape so reconstruct… - #19582
fix(avro): detect shredded variant base files by shape so reconstruct…#19582voonhous wants to merge 2 commits into
Conversation
…ion engages
HoodieVariantReconstruction never engaged on real files: the reader's file
schema comes from converting the parquet footer MessageType, which loses the
variant logical type, so a shredded variant group arrives as a plain
{metadata, value, typed_value} record. Detection required a
HoodieSchema.Variant with isShredded(), found none, and the reader proceeded
at the unshredded schema, silently clipping typed_value away. Any avro-path
rewrite of a shredded base file persisted the nulls; the easiest trigger is
the COW small-file merge, where a plain second INSERT bin-packs into an
existing file group (apache#19567). This is the second leg of apache#19556; the Spark
reader-context leg was apache#19558.
Detect the on-disk side by SHAPE, anchored by the requested side: the
requested column (table schema, logical type intact) must be a variant for
the shape match to count, so plain user structs of the same shape are
unaffected. A file schema that kept its logical type still short-circuits on
isShredded. With detection engaging, the existing fail-fast branches
(shredded reading disabled, no provider available) now fire where they
previously never triggered, turning silent corruption into loud errors.
Extracted from the shredding-inference branch (c2ec87d), leaving the
inference-specific parts behind.
- Unit tests: footer-style plain-record schema engages and reconstructs end
to end; the same shape without a variant requested column stays disengaged.
- Functional test: the issue's repro, a COW small-file merge round trip with
the shredded layout and the same-file-group bin-pack pinned.
|
@wombatu-kun Can you please help to review this too? |
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 HoodieVariantReconstruction so it detects shredded VARIANT base files by shape (anchored on the requested column being a variant), since the footer-derived file schema loses the variant logical type — which previously caused the AVRO read path to silently null out shredded values during a COW small-file merge. I traced the detection and reconstruction path end-to-end and cross-checked the shape against the write path in HoodieSchema.createVariantShredded, confirming the requested-side anchor prevents plain user structs from matching. No issues flagged from this automated pass — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19582 +/- ##
============================================
+ Coverage 77.49% 77.51% +0.02%
- Complexity 32799 32821 +22
============================================
Files 2522 2522
Lines 139179 139224 +45
Branches 16734 16742 +8
============================================
+ Hits 107855 107923 +68
+ Misses 23748 23722 -26
- Partials 7576 7579 +3
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
…chored requested schema
Round-1 review traced the repro path further than the fix went: the COW
small-file bin-pack runs HoodieConcatHandle -> HoodieMergeHelper, where
isStrictProjectionOf(readerSchema, writerSchema) fails on RECORD vs VARIANT
because the footer-derived reader schema lost the variant logical type. The
merge then read the base file AT the footer schema, so the requested side had
no variant column for the shape detection to anchor on, reconstruction stayed
disengaged, and the subsequent rewriteRecordWithNewSchema silently dropped
typed_value. Detection alone left the E2E scenario broken.
- Share the anchored predicate as VariantSchemaUtils.isShreddedVariantTarget
and add alignShreddedVariants, which swaps footer-plain shredded variant
columns for their requested form; HoodieVariantReconstruction delegates.
- HoodieMergeHelper aligns shredded variant columns before the strict
projection check, so recordSchema resolves to the writer schema, the avro
reader engages reconstruction, and the lossy rewrite is skipped.
- Pin the E2E test to the AVRO record type: the record type picks the reader
the merge uses, and this PR covers the avro leg.
- Add the requested negative test: a footer-derived plain unshredded
{metadata, value} record against a variant requested column stays
disengaged; add a compatibility test for align + strict projection.
Residual: a shredded base file merged while the writer schema evolves (added
columns, or the internal-schema transformer) still takes the rewrite path and
drops typed_value; that needs a variant-aware record rewrite and is left to a
follow-up.
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for the follow-ups here — this round is a solid extension of the fix. The new changes move the shredded-variant shape detection into VariantSchemaUtils and, more importantly, add alignShreddedVariants into HoodieMergeHelper's strict-projection check so the COW small-file bin-pack merge (HoodieConcatHandle → HoodieMergeHelper) reads at the variant-bearing writer schema and lets reconstruction engage instead of silently dropping typed_value.
Both prior reviewer comments look addressed: wombatu-kun's request for a test on the ordinary unshredded {metadata, value} footer layout is covered by the new returnsNullForFooterDerivedPlainUnshreddedShape test, and the observation that the default bin-pack routes through HoodieMergeHelper (not the reader context the clustering tests use) is handled by the new alignment plus pinning the e2e test to the AVRO record type. I confirmed the alignment and the reader's reconstruction share the same file/requested schemas and the same detection predicate, so the two sites can't disagree, and the alignment is surgical enough not to mask genuine non-variant schema differences.
One non-blocking question worth double-checking in the inline comment about how this interacts with Advanced Schema Evolution. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here.
| // schema surfaces a plain {metadata, value, typed_value} record where the writer schema has a | ||
| // variant, which can never pass the strict check even though the reader reconstructs such | ||
| // columns when handed a variant-bearing requested schema. Without the alignment recordSchema | ||
| // degenerates to the footer schema, reconstruction cannot anchor on a variant requested |
There was a problem hiding this comment.
🤖 The alignment only takes effect when schemaEvolutionTransformerOpt.isEmpty(). If Advanced Schema Evolution (schema-on-read) is enabled on a table with a shredded variant column, wouldn't this same small-file merge fall back to recordSchema = readerSchema (the footer schema, no variant to anchor on) and drop typed_value again via the transformer's rewrite? Is that combination out of scope here, or worth a follow-up?
| boolean isPureProjection = schemaEvolutionTransformerOpt.isEmpty() | ||
| && HoodieSchemaCompatibility.isStrictProjectionOf(readerSchema, writerSchema); | ||
| && HoodieSchemaCompatibility.isStrictProjectionOf( | ||
| VariantSchemaUtils.alignShreddedVariants(readerSchema, writerSchema), writerSchema); |
There was a problem hiding this comment.
The Summary and Risk Level sections still describe detection only, with no mention that runMerge now flips isPureProjection for shredded variant base files and skips the writer-schema rewrite - a change on the code path every COW merge shares. Could you fold that and the residual from the commit message (a schema-evolution transformer or an added column still takes the lossy rewrite) into the PR body?
| }) | ||
| } | ||
|
|
||
| test("Test COW small-file merge preserves shredded VARIANT values") { |
There was a problem hiding this comment.
This has no unshredded twin, so nothing pins that an ordinary variant table - the layout written without the force-shredding config - still round-trips through the same bin-pack now that alignShreddedVariants runs on every runMerge. Could you add one the way the clustering pair above is split?
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.
5 participants
…ion engages
Describe the issue this Pull Request addresses
Closes #19567. The avro internal read path silently nulls shredded VARIANT values during base-file rewrites; the easiest trigger is a COW small-file merge, where a plain second INSERT bin-packs into an existing file group, reads the old base file through HoodieAvroParquetReader, gets value=null for every shredded value, and writes the nulls into the new file version. No table service involved and no error surfaced.
Root cause: HoodieVariantReconstruction never engaged on real files. Its file schema comes from converting the parquet footer MessageType, which loses the variant logical type, so a shredded variant group arrives as a plain record; the detection required a HoodieSchema.Variant with isShredded(), found none, concluded "no shredded columns", and the reader proceeded at the unshredded schema, clipping typed_value away. The existing unit tests passed because they hand-build annotated schemas that real footers never produce.
This is the second leg of #19556; the Spark reader-context leg was fixed by #19558.
Summary and Changelog
The detection fix is extracted from the shredding-inference branch (where it was entangled with the inference feature); the inference-specific parts are not included.
Impact
Wrong-results fix. Any rewrite that reads shredded variant base files through the avro path (COW small-file merge, and avro-path rewrites generally) preserved nulls before this fix. Environments without a shredding provider now fail fast on such files instead of silently corrupting them.
Risk Level
low. Detection only widens when the requested column is a variant and the file column has the exact shredded shape; everything downstream of detection is unchanged and already covered by unit tests.
Documentation Update
none
Contributor's checklist