fix(variant): close the shredded-read gaps exposed by a mixed-layout test matrix - #19687
fix(variant): close the shredded-read gaps exposed by a mixed-layout test matrix#19687voonhous wants to merge 19 commits into
Conversation
voonhous
changed the title
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR expands variant-shredding test coverage to mixed layouts and adds production hardening: fail-fast guards in the Hive reader, Spark 4.0 read support, and the schema-on-read path so shredded-variant files error loudly instead of silently returning nulls; a bootstrap-reader fix so reconstruction engages at the caller schema; a typed_value nullability fix per the shredding spec; and up-front rejection of clustering sort on ordering-less types. I traced each production change against its surrounding source and callers: the guards are correctly scoped to fire only on genuine shredded variants, the bootstrap change is consistent with the two-arg overload, and the schema change is backward-readable within this unreleased feature. No correctness issues found. A few style/readability suggestions in the inline comments. Please take a look, and this should be ready for a Hudi committer or PMC member to take it from here.
. A couple of minor naming and message-consistency nits below.
cc @yihua
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19687 +/- ##
============================================
- Coverage 77.85% 77.80% -0.06%
- Complexity 33189 33198 +9
============================================
Files 2532 2532
Lines 140050 140208 +158
Branches 16861 16910 +49
============================================
+ Hits 109040 109082 +42
- Misses 23386 23487 +101
- Partials 7624 7639 +15
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
|
Pushed d8ffd05 after a self-review round. Short version: the guards this PR added had bypasses, and a few of the new tests could not fail for the reason they exist. This commit closes both. Production
Tests
Deliberately left as follow-ups: the flink 1.18-2.0 copies of |
VariantShreddingTestSupport carries the parquet-footer inspectors that TestVariantDataType grew (listDataParquetFiles, readParquetSchema, getFieldAsGroup, assertVariantLayout, assertSingleFileGroup) plus the new mixed-layout machinery: a per-file layout map split by base vs native log, a parquet-hadoop row inspector that counts typed vs residual rows inside one file, a shape-drift SQL generator, a session-conf write-layout toggle (Unshredded / Forced / Inferred with a forced stand-in until apache#18961 lands), and the table-service idioms. TestVariantDataType mixes the trait in; no test behavior changes.
…e shredding spec The shredding spec makes typed_value OPTIONAL: a row whose value does not match the shredding schema (a scalar or array under an object schema, a JSON null) leaves typed_value null and carries everything in the value residual. createVariantShredded attached the typed_value schema as-is, so an object shredding schema produced a REQUIRED record and any non-object row crashed the Avro write path with parquet-avro's 'Null-value for required field: typed_value'. The Spark row writer already writes typed_value optional, and Spark4VariantShreddingProvider already unwraps nullable typed_value at every touch point and puts an explicit null for non-matching rows - only the schema builder disagreed. typed_value is now wrapped nullable (idempotently), the Variant accessor unwraps the union so consumers keep seeing the value type, and the Avro-to-parquet conversion pin flips to 'optional ... typed_value'. Existing files with a required typed_value stay readable: read-side detection is shape-based and never inspects typed_value nullability.
…ing and read modes Every shredded test so far forces ONE layout per table. This matrix pins what production will produce once per-file inference (apache#18961) lands: files with different typed_value layouts in one table, shredded bases under unshredded logs and the reverse, and rows inside one file that fell back to the residual. - same-file mix: forced schema with conflicting / disjoint / non-object rows, physical typed-vs-residual counts pinned per the spec - same-table mix: per-commit layouts; snapshot, time travel, incremental and read-optimized over all of them; small-file bin-pack re-deriving the layout from the incoming commit (inputs never consulted) - MOR compaction: three-layout log split compacted three times under three layouts (asserting the base re-derives each time), table version 9 legacy blocks, event-time vs commit-time ordering with deletes - clustering: heterogeneous inputs rewritten under a configured layout on both the row-writer and RDD paths, COW and MOR-with-logs - reads: variant_get/try_variant_get across files where the path is typed, residual, conflicted or absent, with pushVariantIntoScan on and off; CDC images across layout changes - nested variant forced-shredded by the row writer; add-column evolution followed by compaction and clustering; rollback and savepoint restore across layouts Inference legs substitute a forced stand-in until apache#18961 lands, so the matrix keeps its shape on every profile.
The package holds TestBaseSpark3AdapterVariantMethods and TestBaseSpark4AdapterVariantMethods, which no lane's wildcard filter matched, so they never ran in CI.
variant_shredded_mixed_cow.zip carries a forced-shredded file group (with one per-field residual row) and an unshredded file group, for the engines that cannot write shredded files themselves: the Spark 4.0, Hive and Flink read-behavior pins.
Flink has no shredded-variant read support; reading the mixed-layout fixture must fail with an error naming the variant column rather than return nulls or a partial payload. Verified on the flink2.1 profile (the shredded file group in the fixture carries no parquet VARIANT annotation, matching what Hudi's writers produce today, so this pins the shape-based guard, not the annotation-based one).
The Hive reader hands base files to a plain parquet-avro read at the
requested {metadata, value} projection, so a file whose variant group
carries typed_value came back with silent nulls: the typed rows keep
their payload in typed_value, which the projection drops. The footer is
already read for schema pruning, so detect the shredded shape there and
throw an error naming the column and the remediation instead.
Detection reuses VariantSchemaUtils.isShreddedVariantTarget: shape-based
on the footer schema, anchored on the requested column being a variant,
so plain user structs of the same shape are left alone. Top-level
columns only, matching the writers' shredding scope. Queries that do
not project the variant column (e.g. count(*)) and unshredded variant
files keep reading as before, pinned by the new tests. The legacy
non-file-group-reader path (hoodie.file.group.reader.enabled=false)
has no requested-schema anchor and stays unguarded.
…le read
The one-argument getRecordIterator overload (used by HoodieMergeHelper's
bootstrap branch) read the data file at its OWN footer schema. A
shredded variant column surfaces there as a plain {metadata, value,
typed_value} record with the logical type lost, so
HoodieVariantReconstruction never engaged and the later rewrite to the
writer schema silently dropped typed_value. Requesting the caller's
schema minus meta fields - exactly what the two-argument overload does -
anchors reconstruction on the requested variant. Pinned by a round-trip
test with the real provider: red at the footer schema, green now.
Sorting a clustering plan by a VARIANT (or BLOB/VECTOR) column used to fail deep in the Spark job - an AnalysisException from the row partitioner or a ClassCastException from the RDD one. The procedure's order parameter and the execution strategy's configured sort columns (hoodie.clustering.plan.strategy.sort.columns, covering the inline and async paths) now fail up front with an error naming the column.
The internal schema models a variant as a two-field {metadata, value}
record (sentinel negative field ids), so the merged schema-on-read
request clips the file's typed_value away and typed rows read back with
a null value residual - silent data loss. Until reconstruction under
schema-on-read lands (apache#18285), detect the shredded footer shape under
an internal-schema read and throw an error naming the column. The check
anchors on the sentinel ids, which no real user field can carry.
Under the default PushVariantIntoScan rewrite the same read already
fails before this hook (the internal-schema pruning cannot resolve the
projection struct's synthetic children); the guard closes the
pushVariantIntoScan=false leg, where the read otherwise succeeded with
nulls. Known apache#18285 residue documented in the test: the schema-on-read
DDL also degrades the CATALOG schema's variant column to a plain
struct, so plain reads after the DDL fail in Spark before any Hudi
hook.
Spark 4.0's unshredded converter reads only [value, metadata]; the reorder workaround (apache#18334) dropped typed_value from the requested schema, so shredded files written by Spark 4.1+ (or the forced test DDL) read back with a partial or null payload. Throw an error naming the column instead; unshredded groups keep the reorder. Spark 4.0 cannot reconstruct shredded variants (no SPARK-54410), so failing loudly is the only safe behavior.
- clustering sort on a variant column rejected via the procedure order parameter AND the configured plan-strategy sort columns - schema-on-read of a shredded file fails with the guard error on the pushVariantIntoScan=false leg and stays loud on the default leg - Spark 4.0 read-back leg of the shredded write test asserts the fail fast instead of skipping - assertQueryFailsWith joins the shared trait: checkNestedExceptionContains never executes the query (spark.sql is lazy), so it cannot see execution-time failures like per-file read guards
…-fast leg Empirically Spark 4.0 rejects the 3-field group during file-schema conversion (INVALID_VARIANT_FROM_PARQUET.WRONG_NUM_FIELDS) before the Hudi read-support guard runs on this path; the guard remains defense in depth for requested schemas that bypass the converter. Either way the read fails naming the variant, which is what the leg pins.
…ort-rejection hint - bootstrapReaderReconstructsShreddedDataFileAtTheTableSchema -> ...UsingTableSchema - the procedure's sort-column rejection now ends with the same actionable hint as the execution strategy's twin
Guards: - schema-on-read: the guard now recurses into structs, array elements and map values (the row writer shreds nested variants), skips empty projections so count(*) keeps working, and is shared with the four per-version Legacy parquet file formats that carry the same merge block - hive: detect shredded variants at any depth via toShreddedReadSchema, and repeat the fail-fast on the non-file-group-reader path (fgr disabled, schema-on-read enabled, bootstrap splits), which lands on Hive's plain parquet reader with no guard - sort columns: one shared SortUtils.validateSortableColumns - case insensitive, rejects MAP alongside VARIANT/BLOB/VECTOR, applied in the partitioner constructors (custom-columns row/RDD/java and both consistent-bucket partitioners) so the consistent-bucket strategy and plain bulk_insert sort columns cannot bypass the check; the execution strategy and run_clustering keep their early copies for a cleaner error Tests: - nested-variant fail-fast legs for the hive and schema-on-read guards - BLOB/VECTOR/MAP rejection pinned in TestClusteringProcedure, not gated on Spark 4.1, so the validation is covered on every lane - incremental, read-optimized and time-travel legs assert values, not counts; the pushVariantIntoScan default leg asserts a variant-related failure instead of a bare intercept[Throwable] - the nested row-writer test sweeps the vectorized reader (apache#18605) - the Flink pin asserts the guard's own message - the CDC layout-flip folded into TestVariantDataType's CDC test; the ordering-modes test folded into the three-layout compaction test (its delete-block leg); the unshredded clustering twin retired in favor of the mixed-layout clustering matrix - dead Inferred scaffolding pruned (unused shapes and helpers); shared createVariantTable and checkNestedExceptionContains helpers Reverted the bot.yml filter hunk: the org.apache.spark.sql.adapter tests are JUnit classes the java UT lanes already run; wildcardSuites only discovers scalatest suites, so the hunk was a no-op.
6bc3e16 to
c5691f3
Compare
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR closes several shredded-variant read gaps across the Spark legacy parquet formats and the Hive/legacy-input-format paths — wiring in a shared schema-on-read guard, adding fail-fast checks for shredded variant groups, fixing bootstrap reconstruction, and rejecting unsortable clustering sort columns, with supporting tests. The main thing to double-check is the legacy HoodieParquetInputFormat guard, which detects shredding via the variant logical type that the footer-derived schema may no longer carry, so it might never fire. Please take a look at the inline comments, and this should be ready for a Hudi committer or PMC member to take it from here. A couple of small maintainability nits — a duplicated magic-string constant and a redundant split — otherwise the code is clean and well-commented.
| if (fileSchema.getType() != HoodieSchemaType.RECORD) { | ||
| return; | ||
| } | ||
| HoodieSchema strippedSchema = VariantSchemaUtils.stripVariantShredding(fileSchema); |
There was a problem hiding this comment.
🤖 stripVariantShredding only strips nodes typed VARIANT (via stripVariantShreddingAt -> case VARIANT -> isShredded()), but the readSchema here goes through AvroSchemaConverterWithTimestampNTZ.convert(MessageType), which doesn't recognize the parquet VariantLogicalTypeAnnotation and falls back to plain-record conversion — so a shredded variant comes back as a plain {metadata, value, typed_value} record with no VARIANT node. stripVariantShredding never matches it, strippedSchema == fileSchema is always true, and this guard never throws, leaving the legacy Hive path silently nulling shredded columns. The FGR guard in HiveHoodieReaderContext avoids this by anchoring on the requested variant via toShreddedReadSchema/isShreddedVariantShape. Could you switch to shape-based detection here (the variant group already exposes typed_value in the footer schema)? @yihua
| public static final String HOODIE_VALID_COMMITS_LIST = "hoodie.valid.commits.list"; | ||
| // Mirrors Spark's VariantMetadata.METADATA_KEY (Spark 4.x only), referenced by literal | ||
| // because the class does not exist on Spark 3 classpaths. | ||
| private static final String SPARK_VARIANT_METADATA_KEY = "__VARIANT_METADATA_KEY"; |
There was a problem hiding this comment.
🤖 nit: SPARK_VARIANT_METADATA_KEY is private here, which forces ParquetSchemaEvolutionUtils.scala to define its own independent copy of the same string literal. If the key ever changes in Spark it would need updating in two places — could you move this to a shared (at least package-visible) constant so there's a single source of truth?
| throw new HoodieClusteringException("Order column not exist:" + col) | ||
| } | ||
| }) | ||
| // The same validation the partitioners apply at execution time (see |
There was a problem hiding this comment.
🤖 nit: orderColumns.split(",") is computed twice — once on line 236 for the foreach existence check and again here for validateSortableColumns. Could you extract it to a val above the loop so the split is done once and both call sites share the same array?
4 participants
Describe the issue this Pull Request addresses
Closes #19688. Part of #18937 (variant shredding feature completeness). Follow-up #19689 (nested shredding parity) is blocked on this PR and #18961.
Shredded-variant tests force ONE
typed_valuelayout per table; per-file inference (#18961) will make MIXED layouts routine (different layouts per file, shredded bases under unshredded logs, typed and residual rows in one file). Auditing the paths that meet these mixes found six defects, several of them silent data loss. Full audit narrative in #19688.Summary and Changelog
Fixes (one commit each, each pinned by a test that was red before the fix):
fix(schema): shreddedtyped_valuedeclared nullable, per the specNull-value for required field: typed_value); also the crash #18961's Avro inference hits on post-sample rowsfix(hive): fail fast on a requested shredded variant column, at any nesting depth, on the file-group-reader path and the legacyMapredParquetInputFormatfall-through (fgr disabled, schema-on-read, bootstrap splits){metadata, value})count(*), no-variant projections, unshredded files keep readingfix(bootstrap): one-arggetRecordIteratorrequests the caller's schematyped_valuesilently droppedfix(clustering): reject unsortable sort columns (VARIANT/MAP/BLOB/VECTOR), case-insensitively, via a sharedSortUtils.validateSortableColumnsapplied in the partitioner constructorsAnalysisException/ClassCastExceptiondeep in the job; consistent-bucket clustering and plainbulk_insertsort columns bypassed the check entirelyorder, plan-strategy sort columns, bulk_insert sort columns)fix(spark): schema-on-read guard, recursing into nested types, shared with the four per-version Legacy parquet file formats; empty projections (count(*)) keep workingpushVariantIntoScan=false(internal schema clipstyped_value), on the fgr path and the legacy relations alikefix(spark4.0): guard inreorderVariantTypetyped_valuefrom the requested schemaTests: new
TestVariantShreddingMixedLayouts(15 tests) +VariantShreddingTestSupporttrait extracted fromTestVariantDataType(no behavior change there). Coverage matrix in #19688; dimensions: same-file typed-vs-residual placement, four layouts in one table across snapshot/time-travel/incremental/RO, three-layout MOR compaction under three service-time layouts, table version 9, ordering modes, heterogeneous clustering (row-writer + RDD, COW + MOR-with-logs),variant_getacross mixed files, CDC, nested, evolution, rollback/savepoint. Plus a Spark 4.1-written mixed-layout fixture zip for the Flink/Spark 4.0 fail-fast pins.Impact
typed_valueoptional (spec-compliant, matching the row writer).Risk Level: low
Off-default surfaces; every fix pinned red-to-green. Verified locally: variant suites green on spark4.1 + spark4.2 (scala-2.13), spark4.0 (incl.
TestHoodieRowParquetWriteSupportVariant), flink2.1ITTestVariantCrossEngineCompatibility4/4, hudi-hadoop-mr module. Note: everyspark4.0CI lane is[CI-TRIM]-commented, so the Spark 4.0 fix and its tests are covered by the local-Dspark4.0run only.Documentation Update
None. No configs added or changed; guard errors carry their own remediation.
Contributor's checklist