voonhous commented on code in PR #19808:
URL: https://github.com/apache/hudi/pull/19808#discussion_r3904658747
##########
hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java:
##########
@@ -308,22 +310,26 @@ public class HoodieStorageConfig extends HoodieConfig {
public static final ConfigProperty<Boolean>
PARQUET_VARIANT_SHREDDING_SCHEMA_INFERENCE_ENABLED = ConfigProperty
.key("hoodie.parquet.variant.shredding.schema.inference.enabled")
- .defaultValue(false)
+ .defaultValue(true)
Review 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.
--
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]
