fix(variant): project PushVariantIntoScan struct paths in the ... - #19783
fix(variant): project PushVariantIntoScan struct paths in the ...#19783voonhous wants to merge 3 commits into
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.29% -0.01%
- Complexity 33861 33870 +9
============================================
Files 2541 2541
Lines 141676 141696 +20
Branches 17172 17177 +5
============================================
+ Hits 110928 110943 +15
- Misses 23039 23047 +8
+ Partials 7709 7706 -3
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
e268d40 to
115b202
Compare
115b202 to
a51b75e
Compare
| (f.dataType, SparkFileFormatInternalRowReaderContext.findFieldByName(source, f.name).map(_.dataType)) match { | ||
| case (_, Some(projStruct: StructType)) if sparkAdapter.isVariantProjectionStruct(projStruct) => | ||
| f.copy(dataType = projStruct) | ||
| case (targetStruct: StructType, Some(sourceStruct: StructType)) => |
There was a problem hiding this comment.
SparkSchemaTransformUtils.addMissingFields has no variant-projection case, so once any sibling member of the enclosing struct triggers an implicit type change the nested projection installed here is replaced by the file's VariantType in the read schema. Should it get the same special case isDataTypeEqualForPhysicalSchema already has, or is implicit evolution alongside a nested variant out of scope for this PR?
There was a problem hiding this comment.
In scope: addMissingFields now has the arm for the pair isDataTypeEqual already declares equal. A projection struct requested over a file VariantType stays the reader type instead of being folded back to VariantType -- which parquet then decoded as a raw variant that the type-change Cast cast to the projected struct, i.e. nulls for that member. Pinned in TestBaseSpark4AdapterVariantMethods.testImplicitSchemaChangeKeepsNestedVariantProjection (file s.n int, table s.n long, s.inner projected: the reconciled reader type keeps the projection struct) and functionally by the new "Implicit widening of a sibling keeps the nested variant projection" leg in TestVariantShreddingMixedLayouts.
| // 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. The skip-merging reader's native VariantType request is | ||
| // reconstructed by the Spark 4.1+ row reader as well (pinned by TestStreamingSource), so this |
There was a problem hiding this comment.
TestStreamingSource's table also carries a top-level v, so shouldRerouteVariantSplit is true there and the base-only split never reaches requiredSchemaReaderSkipMerging - that leg is not what the test pins. Drop the reference here and in the matching "Not swept here" note in TestVariantShreddingMixedLayouts, or add a nested-only leg that actually takes the branch?
There was a problem hiding this comment.
Added the nested-only leg rather than dropping the claim. TestStreamingSource now runs the legacy shredded-variant stream twice, with and without the top-level v. Without it shouldRerouteVariantSplit is false, so the base-only split goes through requiredSchemaReaderSkipMerging and Spark's own row reader rebuilds s.inner from the shredded group; the footer pin asserts s.inner is shredded and that the file carries no v. This comment and the "Not swept here" note in TestVariantShreddingMixedLayouts now name that leg.
| // Pin that the compacted base file is shredded at both depths: without it the streams below | ||
| // would pass just the same over an unshredded base and pin nothing about shredded reads. | ||
| val conf = spark.sessionState.newHadoopConf() | ||
| val baseFiles = new Path(tablePath).getFileSystem(conf).listStatus(new Path(tablePath)) |
There was a problem hiding this comment.
A native parquet log file is named <fileId>_<token>_<instant>_<v>.log.parquet with no dot prefix, so this listing also picks up the two deltacommit logs and "expected a compacted base file" is satisfied even if inline compaction never ran. Filter with FSUtils.isBaseFile the way assertNestedBaseLayout does.
There was a problem hiding this comment.
Done: the listing filters with FSUtils.isBaseFile, and the message says log files are excluded.
| | 1000L as ts from range($lo, $hi, 1, 1)""".stripMargin | ||
|
|
||
| Seq("true", "false").foreach { pushIntoScan => | ||
| withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) { |
There was a problem hiding this comment.
Nothing in these legs asserts that PushVariantIntoScan actually fired, and both arms expect the same rows, so if the rule ever stops matching HoodieFileGroupReaderBasedFileFormat the projected arm silently becomes a duplicate of the other one. Worth pinning the plan in the true arm the way assertLegacyRddPlan pins its own path in TestStreamingSource - follow-up, not a blocker.
There was a problem hiding this comment.
Added variantProjectionPushedIntoScan(sql) to VariantShreddingTestSupport: it collects the FileSourceScanExec off sparkPlan (AQE hides it on executedPlan) and asks whether its requiredSchema carries a projection struct, through the new SparkAdapter.containsVariantProjection. The COW loop and both MOR legs assert it equals their pushVariantIntoScan arm, and the new widening leg asserts it stays pushed.
| VariantGet(ref, pathLit, child.dataType, vm.failOnError, Option(vm.timeZoneId)) | ||
| Seq(Literal(UTF8String.fromString(child.name), StringType), variantGet) | ||
| } | ||
| CreateNamedStruct(childExprs) |
There was a problem hiding this comment.
A null variant in an avro log record is projected into a non-null struct of nulls here, while the parquet paths leave the field null - and PushVariantIntoScan rewrites IsNull(v) / IsNotNull(v) directly onto that struct. Wrap this in the same If(IsNull(...)) guard the struct rebuild below already carries?
There was a problem hiding this comment.
Fixed: the projection-struct arm carries the same If(IsNull(ref), null, ...) guard as the struct rebuild below it. Pinned twice: the projector unit test feeds a null variant inside a live struct and expects a null struct with the sibling intact, and the nested MOR legs now null id 4's variant through the log and check where s.inner is null / is not null; the avro-block leg with the rule on returned no row for is null before the fix.
| } | ||
|
|
||
| /** True when `dataType` is a variant projection struct or holds one below a struct path. */ | ||
| private def containsVariantProjection(dataType: DataType): Boolean = dataType match { |
There was a problem hiding this comment.
containsVariantProjection is now defined identically here and in SparkFileFormatInternalRowReaderContext. SparkAdapter already hosts isVariantProjectionStruct, so a default method there would give both call sites one definition.
There was a problem hiding this comment.
Moved to SparkAdapter as a concrete method over isVariantProjectionStruct; both private copies are gone.
…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.
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
4 participants
...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