voonhous opened a new pull request, #19783:
URL: https://github.com/apache/hudi/pull/19783
### Change Logs
Closes #19775. Part of #18937. Stacked on #19777 (the first commit is that
PR's; review the second commit): the legs need its
test scaffolds and its Avro-path nested write parity, and it merges first.
Pins that a variant shredded below the top level reads correctly through
Hudi's Spark-native internal read paths on Spark 4.1+,
corrects the four comment blocks that deferred the nested case with the
wrong mechanism, and fixes the one real defect the legs
found: the file-group reader applied a PushVariantIntoScan projection to
top-level fields only.
**The defect (issue outcome B, shredding-independent).** Spark 4.1+'s
`PushVariantIntoScan` rewrites a variant reached through
a struct path (`cast(s.inner as string)`, `variant_get(s.inner, ...)`) into
a projection struct nested inside the scan's required
schema (`s: struct<inner: struct<"0": string>>`).
`SparkFileFormatInternalRowReaderContext` overlaid and projected such structs at
the top level only, so on a MOR table with logs the merged row still carried
a raw `VariantVal` at `s.inner` while the plan read
that memory as `s.inner.0`: JVM SIGBUS / `InternalError` / OOM or a silent
null, on native parquet logs and avro blocks, both record
types, shredded or not. Minimal repro: `create table t (id int, s
struct<inner: variant>, ts long) using hudi tblproperties
(type='mor', hoodie.index.type='INMEMORY', ...)`, one insert, `select id,
cast(s.inner as string) from t`. COW was unaffected
because the whole catalyst schema goes to Spark's parquet reader, which
handles a projection struct at any depth; with
`spark.sql.variant.pushVariantIntoScan=false` everything worked.
Fix: `overlayVariantProjections` / `shouldProjectVariants` recurse into
struct members, and `buildVariantProjector` (hoisted out of
the identical `Spark4_1Adapter` / `Spark4_2Adapter` copies into
`BaseSpark4Adapter`) rebuilds a projection struct below a struct
path with `VariantGet` children, recreating enclosing structs
null-preservingly and passing untouched subtrees through by
reference. Struct paths only, mirroring `VariantInRelation.rewriteType`;
arrays and maps stay native on both sides.
**What was verified before writing anything** (all on Spark 4.1.1, the
pinned version):
- The parquet row reader reconstructs a shredded variant for a NATIVE
`VariantType` request at any depth: top-level, struct member
and array element. `ParquetReadSupport.clipParquetType` passes a
`VariantType` leaf's group through unchanged (its
`clipVariantSchema` is a no-op TODO),
`ParquetToSparkSchemaConverter.convertField` carries the catalyst target type
into struct
members, list elements and map values, and `ParquetRowConverter` builds a
`ParquetVariantConverter` for it. `PushVariantIntoScan`
itself only rewrites struct paths; arrays and maps stay native even for
user queries.
- What does read a shredded group as `value=null` is a request in the
physical `{metadata, value}` struct shape. The only
producer of that shape is the schema-on-read internal-schema branch, which
#19687 guards.
- The full-variant projection rewrite
(`SparkAdapter.buildFullVariantReadSchema`) is therefore a contract, not a
workaround:
with it disabled (patched class shadowed on the test classpath, class
origin verified), `TestVariantDataType` is 20/20 and
`TestVariantShreddingMixedLayouts` 21/22 -- the one red is the
schema-on-read fail-fast leg, where the projection shape is what
lets `validateNoShreddedVariants` name the route instead of a
`ClassCastException` inside the read. The same holds at the #19558
fix commit itself (rebuilt in a worktree): the #19556 clustering repro
passes there with the rewrite disabled and the logged
parquet requested schema keeps `typed_value`. With
`HoodieMergeOnReadRDDV2.shouldRerouteVariantSplit` forced off, the legacy
streaming shredded leg passes too.
So this is issue #19775's outcome A: the rewrite stays top-level and the
code is unchanged; the comments now say what the shape
is for and why nested variants need none.
**Tests** (`TestVariantShreddingMixedLayouts`, section F; every table has
ONLY the nested variant so nothing top-level can mask the
nested path -- a new `withNestedOnlyVariantTable` scaffold in
`VariantShreddingTestSupport`):
- MOR: nested-shredded base + nested-shredded log; snapshot merge with the
vectorized reader on and off, read-optimized, the legacy
RDD path (`hoodie.file.group.reader.enabled=false`) over a base + log
slice and over the compacted log-free slice, compaction under
the forced DDL (base re-shredded on both record types, the AVRO one
through #19777's Avro-path parity) and under `Unshredded`
(typed_value stripped at depth), clustering under the forced DDL for both
`row.writer.enable` values.
- CDC (`hudi_table_changes(..., 'cdc', 'earliest')`, both supplemental
logging modes): insert and update images of `s.inner`
across a layout flip.
- `variant_get` projection and filter on `s.inner` across a typed file and a
residual file, `pushVariantIntoScan` on and off: on
COW, and on MOR over native parquet log files (SPARK record type) and
table-version-9 avro data blocks (AVRO record type, pinned
through the log block types). Red before the fix as a JVM abort
(SIGSEGV/SIGBUS in an unsafe copy stub).
- `TestBaseSpark4AdapterVariantMethods`: the recursive projector as a unit
(nested extraction, sibling preserved, null struct
preserved, no projection -> None).
- `TestStreamingSource`'s legacy RDD leg (the only remaining
`hoodie.file.group.reader.enabled=false` consumer; batch reads no
longer consult that key) gains a nested-shredded column on every split
plus a footer pin on the compacted base.
- `array<variant>` shredded at the element through a declared
`hoodie.write.schema` (the only way a Hudi write produces that layout;
SPARK record type, since the AVRO record-type insert cannot take a
write-schema override of a different shape): snapshot read with
the vectorized reader on and off, `variant_get` on an element, a log
update, compaction under the declared schema (element
re-shredded) and without it (element unshredded).
**Not done here, on purpose:** removing the projection rewrite and the
legacy re-route now that both are shown to be contract rather
than correctness. That is a behaviour change (it would put variant tables
back on the skip-merging fast path) and belongs in its
own PR if wanted.
### Impact
Reads of a struct-nested variant through the Spark file-group reader with
`pushVariantIntoScan` on (the default) no longer crash
or return nulls on MOR tables with log files. No config changes.
### Risk level
Low
### Documentation Update
None
### Contributor's checklist
- [x] Read through [contributor's
guide](https://hudi.apache.org/contribute/how-to-contribute)
- [x] Change Logs and Impact were stated clearly
- [x] Adequate tests were added if applicable
- [ ] CI passed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
