feat(variant): infer shredding schemas by default on Spark 4.1+ - #19808
Conversation
Flip hoodie.parquet.variant.shredding.schema.inference.enabled to true, closing apache#19690. Spark 4.1 itself defaults both spark.sql.variant.writeShredding.enabled and spark.sql.variant.inferShreddingSchema to true, so a Hudi table on Spark 4.1 was the one parquet writer in that stack still producing unshredded variant columns. Breaking change: base files and native parquet log files written by Spark 4.1+ writers now carry typed_value for top-level variant columns, and only Spark 4.1+ can read them. Spark 4.0, Hive and Flink readers of such tables fail fast (apache#19687) where they read the unshredded layout before. Opt out per writer or per table with hoodie.parquet.variant.shredding.schema.inference.enabled=false (or hoodie.parquet.variant.write.shredding.enabled=false); already shredded files are rewritten unshredded by clustering with the option off. Existing files are not touched. Only writers with a Spark 4.1+ inferrer on the classpath change behavior: Spark 3.x, Spark 4.0, Flink and Java writers keep writing unshredded files, which the Avro-factory test now pins with a stock config. The config doc carries the reader matrix, the opt-out and the un-shred recipe. TestVariantSchemaUtils pins the new default and sets the flag off explicitly where it tested the disabled path. No Spark assertion moved: every unshredded-layout pin already disables write shredding or inference explicitly.
203830a to
4d8078c
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 flips hoodie.parquet.variant.shredding.schema.inference.enabled to default on, so Spark 4.1+ writers shred top-level variant columns by default, and refreshes the config docs and the two pinning tests accordingly. I traced the three writer factories (Avro, Spark row, Spark) and confirmed the inference path is consistently gated on VariantShreddingRuntime.lookupInferrer().isPresent(), so non-4.1 writers still degrade to unshredded — the gating looks correct and the tests are self-consistent with the new default. One cross-engine durability point on the default-on decision is worth a committer's eyes (see inline). Please take a look at the inline comment, and this should be ready for a Hudi committer or PMC member to take it from here.
| public static final ConfigProperty<Boolean> PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED = ConfigProperty | ||
| .key("hoodie.parquet.variant.shredding.schema.inference.enabled") | ||
| .defaultValue(false) | ||
| .defaultValue(true) |
There was a problem hiding this comment.
🤖 Flipping this to default-on while it stays a write config (not persisted per table) inverts the failure mode of the "writer forgot the config" case: before, a writer missing this wrote unshredded and stayed cross-engine readable; now any Spark 4.1+ writer that misses the opt-out (a path-based write, the DataSource writer, a streamer) silently shreds and breaks Flink/Hive/Spark-4.0 reads of that same table. Have you considered persisting the shredding decision as a table property so the opt-out is durable across every writer, rather than relying on each write path to pass it? @yihua since this is effectively a storage-format-default change, worth a committer/PMC look at the default-on choice and confirming the non-4.1 reader fail-fast is complete (no silent payload drop) across all read paths.
There was a problem hiding this comment.
Keeping it on by default. The audit this asks for did turn up one real silent-drop path, now fixed in this PR.
Default-on stays. Spark 4.1 defaults both spark.sql.variant.writeShredding.enabled and spark.sql.variant.inferShreddingSchema to true; a plain variant column in Hudi should land on disk the way the same column lands in plain parquet. The key is sinceVersion("1.3.0") and unreleased, so no existing table changes behavior on upgrade -- flipping it after the first release is the change that would be an upgrade regression.
On persisting the decision as a table property. Shredded-ness is per file, not per table, and already was before this PR: inference samples the first records of each open file and splices a typed_value schema for that file alone, a file whose sample has nothing to shred stays unshredded, Avro log blocks never shred, and hoodie.parquet.variant.write.shredding.enabled has defaulted to true since 1.1.0 -- a writer that missed the config already shredded whenever the write schema carried a typed_value. So there is no table-level invariant to persist, and none this PR removes. The value is also not out of reach per table: SQL DML and table-name procedures pick it up from the table's catalog properties. The remaining gap for path-based procedures, the DataSource writer and the streamer is the same for every hoodie.parquet.* write config (codec, block size, bloom filter).
Fail-fast audit for non-4.1 readers
| Reader | On a shredded file | Where |
|---|---|---|
| Spark 4.1 / 4.2 | reconstructs | supported |
| Spark 4.0 | throws | Spark40HoodieParquetReadSupport.rejectShreddedVariants, off the footer from both the read support and Spark40ParquetReader |
| Spark 3.x, schema from commit metadata | throws | no VariantType, the schema does not convert (BaseSpark3Adapter) |
| Spark 3.x / 4.x schema-on-read | throws | ParquetSchemaEvolutionUtils.validateNoShreddedVariants |
| Flink 2.1 / 2.2 | throws | ParquetSplitReaderUtil.validateVariantType, ParquetSchemaConverter.convertToRowType |
| Flink <= 2.0 | no VARIANT type, the column cannot be declared | -- |
| Hive | throws | HoodieParquetInputFormat.validateNoShreddedVariantRead, HiveHoodieReaderContext |
| Java / Avro reader | throws on no-provider and on allow.reading.shredded=false |
HoodieVariantReconstruction.create |
| Compaction / merge / bootstrap | reconstructs, keeps typed_value | VariantSchemaUtils.alignShreddedVariants (#19567) |
Spark 4.0 is in that list because it can write shredded from an explicit typed_value schema; what it cannot do is infer one, since the inferrer ships only in the 4.1 and 4.2 modules. So this flip never produces a shredded file on a 4.0-only pipeline.
On Flink, base files and native parquet log blocks both reach the guard -- FlinkRowDataReaderContext.getFileRecordIterator routes both through HoodieRowDataParquetReader -> RecordIterators -> ParquetSplitReaderUtil. It fires at reader construction, before any row, and only for a projected column. Flink writes over such a table (upsert, compaction) read through the same path, so they fail rather than round-tripping a dropped typed_value. ITTestVariantCrossEngineCompatibility#testFlinkReadShreddedVariantCOWTableFailsFast pins it against a fixture whose first file group is shredded and second is not, and asserts the guard's own message.
The gap, fixed in this PR. Spark 3.x has no VariantType, so the table's own schema does not convert and the documented way to read a variant table there is to declare the column as struct<value: binary, metadata: binary> -- the same shape Hive sync writes to the metastore. Parquet reconciles requested against file fields by name, so a shredded group's typed_value was simply not projected and those rows came back with a null value: dropped silently, which is precisely what this thread asks about. Added ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs, called from Spark33/34/35ParquetReader and Spark3LegacyHoodieParquetFileFormat. The anchor is two-sided -- the requested side exactly two binary members named metadata and value, the file side carrying typed_value at that same path -- so a plain user struct is exempt, an unshredded file still reads byte for byte as before, and a query that does not project the column is untouched. Reconstruction is not an alternative on Spark 3.x: the only VariantShreddingProvider ships in spark4-common. Unit tests in TestParquetSchemaEvolutionUtils. 404c531.
Trino has no variant support and no guard on our side, and the synced Hive type there is the same two-binary struct. That is out of scope here -- variant is not usable through that connector today regardless.
Opt-out for tables other engines read. hoodie.parquet.variant.shredding.schema.inference.enabled=false keeps schema-declared shredding; hoodie.parquet.variant.write.shredding.enabled=false disables shredding entirely. Files already shredded return to the unshredded layout by clustering with either set to false.
P.S. comment is a little long and may read like it's LLM, but it's long to cover every aspect of cross version support, configuration location and ownership. So please bear with the lengthiness.
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## master #19808 +/- ##
============================================
- Coverage 78.31% 78.31% -0.01%
- Complexity 33888 33897 +9
============================================
Files 2541 2542 +1
Lines 141728 141775 +47
Branches 17182 17197 +15
============================================
+ Hits 111001 111033 +32
- Misses 23017 23029 +12
- Partials 7710 7713 +3
Flags with carried forward coverage won't be shown. Click here to find out more.
🚀 New features to boost your workflow:
|
Spark 3.x has no VariantType, so a variant column is read there by declaring it as struct<value: binary, metadata: binary>, the same shape Hive sync writes to the metastore. Parquet reconciles the requested against the file fields by name, so a shredded group's typed_value was never projected and those rows came back with a null value: the payload was dropped silently. Reconstruction is not an option, as the only VariantShreddingProvider ships in spark4-common, so reject the read instead, the way Spark 4.0, Flink and Hive already do. Add ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs and call it from Spark33/34/35ParquetReader and the legacy file format. The anchor is two-sided - the request must be exactly the two binary members named metadata and value, and the file must carry typed_value at that path - so a plain user struct, an unshredded file and a query that does not project the column are all untouched.
github-actions
Bot
added
size:M
testValidateNoShreddedVariantStructsRejectsTopLevelShreddedVariant passed the bare struct<metadata, value> as the required schema, so the guard walked its two members against a file whose only column is v, matched nothing and returned without throwing. Request the struct as column v, the way the nested and exemption legs already do.
|
@wombatu-kun Can you. please help to review this PR, this should be the last PR for toggling shredding inference to true by default for Spark4.1+ writers. This should wrap up our work for Variant + shredding on Spark, allowing us to work on other more important priorities. |
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR flips hoodie.parquet.variant.shredding.schema.inference.enabled to default-on for Spark 4.1+ writers and adds a Spark 3.x read guard (validateNoShreddedVariantStructs) that fails fast when a variant column requested as its unshredded struct<value, metadata> shape sits over a shredded parquet group. The guards are wired into the Spark 3.3/3.4/3.5 readers and the legacy file format, and I confirmed the MOR file-group-reader path reuses the same createParquetFileReader().read, so it's covered too. One edge case around partial (nested-pruned) projection of the variant struct is worth double-checking — see the inline comment. The default-on failure-mode tradeoff has already been discussed in the earlier thread. Please take a look at the inline comment, and this should be ready for a Hudi committer or PMC member to take it from here.
| * reading it deliberately. | ||
| */ | ||
| private def isUnshreddedVariantStruct(struct: StructType): Boolean = { | ||
| struct.fields.length == 2 && |
There was a problem hiding this comment.
🤖 isUnshreddedVariantStruct requires exactly 2 fields, but nested schema pruning can hand this guard a pruned variant struct. SELECT v.value on a CoW read (via the HadoopFsRelation fallback in BaseFileOnlyRelation, where nested pruning applies) yields requiredSchema = v: struct<value: binary> — length 1, so this returns false, the walk falls through to the plain-struct recursion into the primitive value, and typed_value is never checked. That silently returns null for every shredded row, exactly the drop this guard aims to stop. The sibling validateNoShreddedVariant avoids this by keying on variant field-id markers rather than field count. Could the subset-projection case be covered too?
There was a problem hiding this comment.
Confirmed reachable: HoodieFileGroupReaderBasedFileFormat extends ParquetFileFormat, so Spark's SchemaPruning narrows the relation's data schema exactly as TestNestedSchemaPruningOptimization asserts, and FileGroupReaderSchemaHandler forwards that narrowed struct verbatim on a base-file-only read. Accepting a non-empty subset of the two binary members in isUnshreddedVariantStruct closes it without weakening the third-member exemption the new tests pin.
There was a problem hiding this comment.
Confirmed and fixed. isUnshreddedVariantStruct now accepts a non-empty subset: every member must be binary and named metadata or value. A member outside those two names still exempts the struct, so the three-member case the tests pin is unchanged.
One consequence worth naming: pruning a deliberately declared {metadata, value, typed_value} struct down to value alone now lands on the guard, since at that point nothing distinguishes it from the variant request. Called out in the scaladoc.
Coverage: testValidateNoShreddedVariantStructsRejectsPrunedVariantStruct pins both single-member spellings against a shredded file and its unshredded twin, and the new functional test runs a pruned select id, v.value leg.
Review follow-ups on the Spark 3.x shredded-variant guard. Nested schema pruning narrows a request to the leaves a query touches, so `SELECT v.value` reached the guard as a one-member struct and slipped past a shape test that wanted both members. Accept a non-empty subset instead: every member binary and named metadata or value. A member outside those two names still exempts the struct, so the three-member shape stays exempt unless pruning strips it back to one of the two. Log blocks never met the guard at all. They are read through HoodieSparkParquetReader, which builds its reader on whatever SparkAdapter.createParquetReadSupport returns, and Spark 3.x got the plain support. Return Spark3HoodieParquetReadSupport instead, guarding in init against the file schema the way Spark 4.0 already does. validateNoShreddedVariantStructs now takes a MessageType so both routes share the walk. Also name Spark 3.x in the config's reader list, name write.shredding.enabled in the un-shred recipe (the key that actually strips typed_value), correct the Avro factory test's gate description, and pin the guard's call sites with a Spark 3 leg over the mixed shredded fixture.
github-actions
Bot
added
size:L
hudi-agent
left a comment
hudi-agent
left a comment
There was a problem hiding this comment.
Thanks for working on this! This PR flips variant shredding-schema inference on by default for Spark 4.1+ writers and adds a fail-fast guard (validateNoShreddedVariantStructs) so Spark 3.x readers reject shredded-variant files instead of silently returning null values. I traced the guard's schema walk (top-level, nested, and pruned-subset cases), confirmed all Spark 3.x read routes are covered (per-version base readers, the legacy format, and the log-block read-support path), and checked that the inference gating still skips schema-on-read tables under the new default. No new issues flagged from this automated pass beyond what earlier rounds already covered — a Hudi committer or PMC member can take it from here for a final review.
cc @yihua
Review follow-ups on the Spark 3.x shredded-variant guard. The readers it protects force spark.sql.caseSensitive=false, so a column declared V or a member declared Value still resolves onto the file's lower-case group. Resolve the same way in the guard: member names compare case-insensitively, and requested columns and struct members are looked up in the file schema ignoring case. Cover Spark3HoodieParquetReadSupport directly over a hand-built InitContext, in a new test dir for hudi-spark3-common, so a wiring slip in init cannot fall silently back to the null-value read. A third leg shows the clipped request carries no typed_value, which is why init anchors on the file schema. Drop the explicit inference tblproperty from the COW inference test so the default flip is pinned end to end.
| public static final ConfigProperty<Boolean> PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED = ConfigProperty | ||
| .key("hoodie.parquet.variant.shredding.schema.inference.enabled") | ||
| .defaultValue(false) | ||
| .defaultValue(true) |
There was a problem hiding this comment.
With inference on by default, HoodieInternalRowFileWriterFactory.newParquetInternalRowFileWriter calls getInferableVariantColumnsFromConfig before checking VariantShreddingRuntime.lookupInferrer, so every row-writer file handle now pays an uncached HoodieSchema.parse of hoodie.avro.schema - on every table, and on classpaths that can never infer. Check the inferrer first (it is a static field), the way HoodieSparkFileWriterFactory and HoodieAvroFileWriterFactory gate on the already-parsed schema argument.
There was a problem hiding this comment.
Reordered by cost: the inferrer lookup (a static field) goes first, so a Spark 3.x or 4.0 classpath pays nothing; then a scan of the StructType argument for a top-level variant through the adapter's isVariantType; only a variant table on an inferring classpath then reaches getInferableVariantColumnsFromConfig and its parse. The other two factories already gate on their parsed schema argument; this one now does the equivalent on its StructType.
| val group = parquetType.asGroupType() | ||
| dataType match { | ||
| case struct: StructType if isUnshreddedVariantStruct(struct) => | ||
| if (group.containsField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD)) { |
There was a problem hiding this comment.
The file side fires on typed_value alone, while Spark40HoodieParquetReadSupport.isVariantGroup and HoodieParquetInputFormat.collectShreddedVariantPaths both also require a binary metadata in the same group - so a plain user struct that happens to carry a typed_value member is rejected once pruning narrows the request to value. Add the metadata requirement here too so the file-side anchor matches the siblings this mirrors.
There was a problem hiding this comment.
Added - isShreddedVariantGroup now requires typed_value next to a binary metadata, the file-side anchor the Hive and Spark 4.0 guards use (value stays optional, as the spec allows). A leg in testValidateNoShreddedVariantStructsLeavesOtherRequestsAlone pins a user struct {value: binary, typed_value: int} under a request pruned to value, which now reads.
Review follow-ups. The row-writer factory called getInferableVariantColumnsFromConfig before looking up the inferrer, and with inference on by default its flag gates pass on every table, so every file handle paid an uncached parse of hoodie.avro.schema - on classpaths that can never infer too. Order the gates by cost: the static inferrer lookup, then a scan of the StructType for a top-level variant, and only then the config-schema parse. The Spark 3.x guard fired on typed_value alone, so once pruning had narrowed a request to a lone value member, a user struct that merely carried a typed_value was rejected too. Require a binary metadata beside it, the file-side anchor the Hive and Spark 4.0 guards already use.
5 participants
Describe the issue this Pull Request addresses
Closes #19690 (last step after #18961, #19687, #19777, #19783). Spark 4.1 itself defaults
spark.sql.variant.inferShreddingSchemaandspark.sql.variant.writeShredding.enabledtotrue; Hudi on Spark 4.1 was the one parquet writer in that stack still writing unshredded variants.Summary and Changelog
hoodie.parquet.variant.shredding.schema.inference.enableddefaults totrue; its doc now carries the reader matrix, the opt-out and the un-shred recipe.TestHoodieAvroFileWriterFactoryVariantInferencepins that with a stock config.TestVariantSchemaUtilspins the default; its disabled legs set the flag off explicitly.TestVariantDataType,TestVariantShreddingMixedLayouts,TestStreamingSourcevariant legs).struct<value: binary, metadata: binary>; that path projected a shredded group by name and returned a nullvalueper shredded row.ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructsnow rejects it, called fromSpark33/34/35ParquetReaderand the Spark 3 legacy file format, with unit tests inTestParquetSchemaEvolutionUtils.Impact
New base files and native parquet log files (table version 10+) from Spark 4.1+ writers carry
typed_valuefor top-level variant columns. Commit schema unchanged; mixed layouts supported (#19687).Risk Level
medium: the default changes on-disk layout and cross-engine readability. Other engines fail with an explicit error, never dropped payloads; the opt-out is one config.
Documentation Update
Config doc in this PR; the variant page on asf-site in a separate PR.
More detail: mechanics, verification, cross-engine notes, release note
Mechanics
VariantShreddingRuntime.lookupInferrer()finds a Spark 4.1+/4.2 inferrer on the classpath; Flink'sHoodieRowDataFileWriterFactorynever consults the flag, and the Avro factory (Flink AVRO merger, Java client) declines when no Spark provider or inferrer is present.typed_valueis per-file physical layout:TableSchemaResolverstrips it, so the commit schema and the catalog schema do not change.PARQUET_VARIANT_WRITE_SHREDDING_ENABLED's doc now names inference as the second source of a shredding schema next to a schema-declaredtyped_value.Verification (default on)
TestVariantDataType20 passed / 3 canceled (Spark 3 legs),TestVariantShreddingMixedLayouts27/27,TestStreamingSourceshredded-variant legs 2/2,TestVariantSchemaUtils15/15,TestHoodieAvroFileWriterFactoryVariantInference1/1,TestHoodieVariantReconstruction13/13; checkstyle clean on hudi-common and hudi-hadoop-common.VariantShreddingTestSupport.layoutConfssets write shredding and inference explicitly per layout, and the unshredded twins inTestVariantDataType(CDC, small-file merge, clustering) sethoodie.parquet.variant.write.shredding.enabled=false, which also gates inference.Cross-engine
ParquetSplitReaderUtil(ITTestVariantCrossEngineCompatibility.testFlinkReadShreddedVariantCOWTableFailsFast), Spark 3.x throughParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs- whether the column is auto-resolved from the table schema or declared as the two-binary struct. None of them returns partial payloads. The Trino connector does not read variant yet.hoodie.parquet.variant.shredding.schema.inference.enabled=false(inference only) orhoodie.parquet.variant.write.shredding.enabled=false(all shredding). For SQL tables that is a tblproperty set once at CREATE or ALTER; path-based procedures, the DataSource writer and the streamer take it as a write option.Release note
Risk mitigation
Contributor's checklist