fix(variant): project PushVariantIntoScan struct paths in the ... - #19783
Merged
voonhous merged 4 commits intoSep 1, 2026
Merged
Conversation
voonhous
changed the title
voonhous
changed the title
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19783 +/- ##
============================================
+ Coverage 78.29% 78.31% +0.01%
- Complexity 33861 33887 +26
============================================
Files 2541 2541
Lines 141676 141728 +52
Branches 17172 17182 +10
============================================
+ Hits 110928 110990 +62
+ Misses 23039 23030 -9
+ Partials 7709 7708 -1
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
voonhous
force-pushed
the
variant-nested-shredded-internal-reads
branch
from
August 28, 2026 11:28
e268d40 to
115b202
Compare
3 tasks
voonhous
force-pushed
the
variant-nested-shredded-internal-reads
branch
from
August 28, 2026 12:43
115b202 to
a51b75e
Compare
wombatu-kun
reviewed
Aug 30, 2026
…oup reader and pin nested-shredded internal reads 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. The Spark file-group-reader context applied that projection to top-level fields only, so on a MOR table with logs the merged row still held a raw VariantVal at s.inner while the plan read that memory as the projected struct: SIGBUS / InternalError / OOM or a silent null, on any log format and record type, shredded or not. COW was unaffected because the whole catalyst schema goes to Spark's parquet reader, which handles a projection struct at any depth. - SparkFileFormatInternalRowReaderContext: overlayVariantProjections and shouldProjectVariants recurse into struct members, mirroring VariantInRelation.rewriteType (struct paths only; arrays and maps stay native). - BaseSpark4Adapter: buildVariantProjector implementation hoisted out of the identical Spark4_1Adapter / Spark4_2Adapter copies and made recursive: a projection struct below a struct path is rebuilt with VariantGet children, enclosing structs are recreated null-preservingly, untouched subtrees are passed through by reference. - The four comment blocks that deferred the nested case (reader context, SparkAdapter.buildFullVariantReadSchema, HoodieMergeOnReadRDDV2, CDCFileGroupIterator) described the wrong mechanism: the Spark 4.1.1 row reader reconstructs a native VariantType request at any depth, so the full-variant projection shape is a contract, not a workaround for a reader clip, and stays top-level. Verified with the rewrite disabled (every existing top-level leg green, including the apache#19556 repro at its own fix commit); the only thing the shape still carries is the schema-on-read fail-fast wording. Tests (TestVariantShreddingMixedLayouts section F, nested-only tables through a new withNestedOnlyVariantTable scaffold; TestBaseSpark4AdapterVariantMethods; TestStreamingSource): MOR merge, read-optimized, compaction under the forced DDL and unshredded, clustering on both writer paths; CDC images through hudi_table_changes in both logging modes; variant_get / cast on the nested path over typed and residual files with pushVariantIntoScan on and off, on COW and on MOR over native parquet logs and table-version-9 avro blocks (red before the fix: JVM abort); the recursive projector as a unit; the streaming legacy RDD leg extended with a nested-shredded column and a footer pin. Closes apache#19775. Part of apache#18937. Stacked on apache#19777.
… schema across snapshot reads and compaction No DDL forces a variant that is directly an array element and Spark's PushVariantIntoScan rewrites struct paths only, so array<variant> is the shape every reader handles natively, shredded or not. The only Hudi write that produces a shredded element is a write schema declaring typed_value there (hoodie.write.schema, honoured by the row write support; the AVRO record-type insert cannot take an override of a different shape). Pins the footer, snapshot reads with the vectorized reader and pushVariantIntoScan swept, variant_get on an element, a log update, read-optimized, compaction under the declared schema (element re-shredded) and without it (element unshredded). Part of apache#19775.
voonhous
force-pushed
the
variant-nested-shredded-internal-reads
branch
from
August 31, 2026 08:48
a51b75e to
4e1524d
Compare
github-actions
Bot
added
size:L
Addresses the six review threads on apache#19783. - BaseSpark4Adapter: a null variant in an avro log record projects to a NULL struct, not a struct of nulls; PushVariantIntoScan rewrites IsNull/IsNotNull onto the projection struct, so `s.inner is null` lost the row. Same If(IsNull) guard the struct rebuild carries. - SparkSchemaTransformUtils.addMissingFields: a projection struct over a file VariantType stays the reader type when a sibling member has an implicit type change; it was folded back to VariantType and the type-change Cast then cast the variant value to the struct. - SparkAdapter.containsVariantProjection is the one definition; the private copies in the adapter and the reader context are gone. - HoodieMergeOnReadRDDV2: the comment claimed TestStreamingSource pins the skip-merging reader over a nested-shredded base; that test's table has a top-level variant, so its splits are re-routed. The leg now runs twice, with and without the top-level variant, and the nested-only leg is what the comment names. - TestStreamingSource lists base files through FSUtils.isBaseFile (native parquet logs matched the ".parquet" suffix). - TestVariantShreddingMixedLayouts: each pushVariantIntoScan arm pins whether the scan's required schema carries the projection struct; the MOR legs null one variant through the log; a new leg widens a sibling (s.n int -> long) through the DataFrame API and reads the int file with the projection pushed. The catalog view of the SQL table keeps reporting the old type, so that leg reads through a path-based view. - Post-rebase compile fix: the array-element write schema uses the 4-arg createVariantShreddedObject.
github-actions
Bot
added
size:XL
wombatu-kun
approved these changes
Sep 1, 2026
| // below, whose reader context requests the full-variant projection shape for parquet base | ||
| // files (#19578), so a SHREDDED base file is read on this legacy path through the same | ||
| // contract as everywhere else. Without a top-level variant the base-only split stays on | ||
| // requiredSchemaReaderSkipMerging, whose native VariantType request the Spark 4.1+ row reader |
Contributor
There was a problem hiding this comment.
Spark41LegacyHoodieParquetFileFormat extends ParquetFileFormat without overriding supportBatch, and VariantType is an AtomicType, so this split reads through the vectorized reader at stock settings - the variant veto lives only in HoodieFileGroupReaderBasedFileFormat. Name the parquet reader rather than the row reader here and in TestStreamingSource's scaladoc?
Member
Author
There was a problem hiding this comment.
Right: the legacy file format takes enableVectorizedReader straight from the inherited ParquetFileFormat.supportBatch, and VariantType is atomic, so that split reads through the vectorized reader unless the user turns it off. The comment here and the scaladoc in TestStreamingSource now say "parquet reader (the vectorized one at stock settings)" and name where the veto actually lives.
| import org.apache.hudi.common.table.log.HoodieLogFormat | ||
| import org.apache.hudi.common.table.log.block.HoodieLogBlock.HoodieLogBlockType | ||
| import org.apache.hudi.common.testutils.HoodieTestUtils | ||
| import org.apache.hudi.common.util.StringUtils |
Contributor
There was a problem hiding this comment.
Moving listLogBlockTypes to VariantShreddingTestSupport left scala.collection.JavaConverters._ and getMetaClientAndFileSystemView imported here with no remaining use. Drop both.
| // null - only its `inner` member is, and only for id 4. | ||
| val wholeStruct = spark.sql(s"select id, s from $tableName order by id").collect() | ||
| assert(wholeStruct.length == 5, s"[$leg] whole-struct read should return 5 rows") | ||
| assert(wholeStruct.forall(!_.isNullAt(1)), s"[$leg] whole-struct read nulled out s") |
Contributor
There was a problem hiding this comment.
This pins only the row count and that s is non-null, so s.inner coming back null or stale for every row would still pass. Check the nested payload here too, the way mergedJson already does for the rewritten reads?
Member
Author
There was a problem hiding this comment.
Done: the whole-struct read now checks s.inner per row against mergedJson (VariantVal.toString is its JSON), null for id 4, alongside the non-null s check.
wombatu-kun
pushed a commit
to voonhous/hudi
that referenced
this pull request
Sep 1, 2026
Addresses the six review threads on apache#19783. - BaseSpark4Adapter: a null variant in an avro log record projects to a NULL struct, not a struct of nulls; PushVariantIntoScan rewrites IsNull/IsNotNull onto the projection struct, so `s.inner is null` lost the row. Same If(IsNull) guard the struct rebuild carries. - SparkSchemaTransformUtils.addMissingFields: a projection struct over a file VariantType stays the reader type when a sibling member has an implicit type change; it was folded back to VariantType and the type-change Cast then cast the variant value to the struct. - SparkAdapter.containsVariantProjection is the one definition; the private copies in the adapter and the reader context are gone. - HoodieMergeOnReadRDDV2: the comment claimed TestStreamingSource pins the skip-merging reader over a nested-shredded base; that test's table has a top-level variant, so its splits are re-routed. The leg now runs twice, with and without the top-level variant, and the nested-only leg is what the comment names. - TestStreamingSource lists base files through FSUtils.isBaseFile (native parquet logs matched the ".parquet" suffix). - TestVariantShreddingMixedLayouts: each pushVariantIntoScan arm pins whether the scan's required schema carries the projection struct; the MOR legs null one variant through the log; a new leg widens a sibling (s.n int -> long) through the DataFrame API and reads the int file with the projection pushed. The catalog view of the SQL table keeps reporting the old type, so that leg reads through a path-based view. - Post-rebase compile fix: the array-element write schema uses the 4-arg createVariantShreddedObject.
wombatu-kun
force-pushed
the
variant-nested-shredded-internal-reads
branch
2 times, most recently
from
September 1, 2026 02:49
2ea92de to
6b4ffe7
Compare
Addresses the three follow-up threads on apache#19783. - HoodieMergeOnReadRDDV2 and TestStreamingSource said the legacy skip-merging split is read by Spark's row reader; the legacy file format inherits ParquetFileFormat.supportBatch and VariantType is atomic, so at stock settings it is the vectorized reader. The variant veto lives in HoodieFileGroupReaderBasedFileFormat only. - The whole-struct read in the nested MOR legs checks the s.inner payload per row, not just the row count and a non-null s. - TestVariantDataType drops the JavaConverters and getMetaClientAndFileSystemView imports left behind when listLogBlockTypes moved to VariantShreddingTestSupport.
Collaborator
3 tasks
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
4 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.
...file-group reader and pin nested-shredded internal reads
TL;DR
spark.sql.variant.pushVariantIntoScan=true, a plain query touching a struct-nested variant (cast(s.inner as string),variant_get(s.inner, ...)) on a MOR table with log files crashed the JVM or returned nulls.Describe the issue this Pull Request addresses
Closes #19775. Part of #18937. Stacked on #19777: its squashed commit comes first, review the rest.
The bug:
s: struct<inner: struct<"0": string>>.SparkFileFormatInternalRowReaderContextoverlaid and projected that struct for top-level fields only.VariantValats.innerwhile the plan read it as the projected struct: SIGBUS,InternalError, OOM, or silent nulls. Any log format, any record type.(id int, s struct<inner: variant>, ts long)withhoodie.index.type=INMEMORY, one insert, thenselect id, cast(s.inner as string) from t.The comments:
VariantTyperequest clips a shredded file to{metadata, value}.Summary and Changelog
Struct-nested variants now read correctly through the Spark file-group reader on MOR tables with logs.
SparkFileFormatInternalRowReaderContext: the projection overlay and its detection recurse into struct members, mirroring Spark'sVariantInRelation.rewriteType. Struct paths only; arrays and maps stay native.BaseSpark4Adapter:buildVariantProjectorhoisted from the identical 4.1/4.2 adapter copies and made recursive. A null variant in an avro log record now projects to a NULL struct instead of a struct of nulls:PushVariantIntoScanrewritesIsNull(v)/IsNotNull(v)onto that struct, sowhere s.inner is nulllost the row before.SparkSchemaTransformUtils.addMissingFields: a projection struct requested over a fileVariantTypestays the reader type when a sibling member of the same struct has an implicit type change (a file withs.nint under a table widened to long); it used to be folded back toVariantType, and the type-changeCastthen cast the variant value to the projected struct.SparkAdapter.containsVariantProjection: one definition for the reader context's overlay and the adapter's projector.Tests, all over tables whose only variant is the nested one:
TestVariantShreddingMixedLayouts, section F:variant_getandcaston COW, and on MOR over native parquet logs and table-version-9 avro blocks, withpushVariantIntoScanon and off. JVM abort before the fix.array<variant>element shredded through a declaredhoodie.write.schema, across reads and compaction.is null/is not null; eachpushVariantIntoScanarm pins whether the scan's required schema carries the projection struct, so the two arms cannot silently collapse into one.s.nint to long) beside a projected nested variant.TestBaseSpark4AdapterVariantMethods: the recursive projector as a unit, a null variant inside a live struct, and the implicit-change reconciliation keeping the projection struct.TestStreamingSource: the streaming legacy-RDD leg twice, with a top-level variant beside the nested one (re-routed to the file-group reader) and nested-only (base-only split served by Spark's own row reader over the shredded group).Impact
Reads of struct-nested variants on MOR tables with log files no longer crash or return nulls;
is nullover avro log records and reads beside an implicitly widened sibling are correct. No config or API changes.Risk Level
low: the projector change is exercised by the new MOR legs on both log formats and both record types; the existing top-level suites are unchanged in behaviour.
Documentation Update
none
Contributor's checklist