voonhous commented on code in PR #19808:
URL: https://github.com/apache/hudi/pull/19808#discussion_r3911178013
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala:
##########
@@ -271,6 +272,76 @@ object ParquetSchemaEvolutionUtils {
}
}
+ /**
+ * Fails the read when a column requested as the unshredded variant struct
sits over a parquet
+ * group that carries typed_value. This is the shape Spark 3.x readers use
for a variant column:
+ * Spark 3.x has no VariantType, so the table's own schema does not convert
(see
+ * BaseSpark3Adapter) and the documented way to read such a table 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
without this guard a
+ * shredded group's typed_value is simply not projected and the rows come
back with a null
+ * `value`: the payload is dropped silently. Reconstruction is not an option
on Spark 3.x, whose
+ * classpath carries no VariantShreddingProvider (the only implementation
ships in spark4-common),
+ * so the read fails instead, as it already does on Spark 4.0, Flink and
Hive.
+ *
+ * The anchor is two-sided: the requested side must be exactly two binary
members named
+ * `metadata` and `value` (a struct carrying any further member is a plain
user struct, exempt
+ * here as it is in the sibling Hive and Spark 4.0 guards), and the file
side must carry
+ * typed_value at that same path. A column the query does not project is
never walked, so a read
+ * that does not touch the variant keeps working, as does an unshredded file.
+ */
+ def validateNoShreddedVariantStructs(requiredSchema: StructType,
footerFileMetaData: FileMetaData): Unit = {
+ val fileParquetSchema = footerFileMetaData.getSchema
+ requiredSchema.fields.foreach { field =>
+ if (fileParquetSchema.containsField(field.name)) {
+ validateNoShreddedVariantStruct(
+ field.dataType,
fileParquetSchema.getType(fileParquetSchema.getFieldIndex(field.name)),
field.name)
+ }
+ }
+ }
+
+ private def validateNoShreddedVariantStruct(dataType: DataType, parquetType:
ParquetType, path: String): Unit = {
+ if (!parquetType.isPrimitive) {
+ val group = parquetType.asGroupType()
+ dataType match {
+ case struct: StructType if isUnshreddedVariantStruct(struct) =>
+ if
(group.containsField(HoodieSchema.Variant.VARIANT_TYPED_VALUE_FIELD)) {
+ throw new HoodieException(String.format(
+ "Column '%s' is a shredded variant (typed_value present)
requested as its unshredded "
+ + "struct shape; Spark 3.x cannot reconstruct shredded
variants, and reading it "
+ + "here would return a null value for every shredded row. Read
the table with "
+ + "Spark 4.1+, or rewrite it unshredded (e.g. cluster with "
+ + "hoodie.parquet.variant.write.shredding.enabled=false).",
path))
+ }
+ case struct: StructType =>
+ struct.fields.foreach { field =>
+ if (group.containsField(field.name)) {
+ validateNoShreddedVariantStruct(field.dataType,
group.getType(field.name), concatPath(path, field.name))
+ }
+ }
+ case array: ArrayType =>
+
parquetListElement(group).foreach(validateNoShreddedVariantStruct(array.elementType,
_, concatPath(path, "element")))
+ case map: MapType =>
+
parquetMapValue(group).foreach(validateNoShreddedVariantStruct(map.valueType,
_, concatPath(path, "value")))
+ case _ =>
+ }
+ }
+ }
+
+ /**
+ * Whether `struct` is the unshredded variant shape: exactly the two binary
members a variant
+ * group carries, in either order. The member count is exact on purpose: a
struct holding a third
+ * member is a user struct that happens to carry those two names, including
the
+ * {metadata, value, typed_value} shape, whose caller already sees the
shredded layout and is
+ * reading it deliberately.
+ */
+ private def isUnshreddedVariantStruct(struct: StructType): Boolean = {
+ struct.fields.length == 2 &&
Review 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.
##########
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)
.sinceVersion("1.3.0")
- .withDocumentation("When enabled, the shredding schema for variant
columns without an explicit "
- + "typed_value in the write schema is inferred automatically per
parquet file from a sample of "
- + "the records written to that file, mirroring Spark 4.1's "
- + "spark.sql.variant.inferShreddingSchema. Requires Spark 4.1+ on
the writer classpath; "
- + "writes stay unshredded otherwise (Spark 4.0, Flink, Java
engines). Applies to every "
- + "parquet file the writer produces: base files and, on table
version 10+, the native "
+ .withDocumentation("Infers the shredding schema of variant columns that
have no explicit "
+ + "typed_value in the write schema, per parquet file, from a sample
of the records written "
+ + "to that file, mirroring Spark 4.1's
spark.sql.variant.inferShreddingSchema (also on by "
+ + "default there). Takes effect only when a Spark 4.1+ writer is on
the classpath; other "
+ + "writers (Spark 3.x, Spark 4.0, Flink, Java) ignore it and write
unshredded. Applies to "
+ + "every parquet file the writer produces: base files and, on table
version 10+, the native "
+ "parquet log files of MOR tables (each infers its own schema).
Data blocks inside "
+ "Avro-format log files, whether Avro or parquet
(hoodie.logfile.data.block.format), stay "
+ "unshredded and shred at compaction. Applies to top-level variant
columns only; a variant "
- + "nested inside a struct, array or map stays unshredded. This is a
write config rather than "
- + "a table config: SQL DML and procedures called by table name pick
it up from the table's "
- + "catalog properties, while path-based procedures, the DataSource
writer and the streamer "
- + "must be handed it explicitly. Up to 4096 records or 64MB are
buffered per "
- + "open file writer before the writer is created, on top of
parquet's own row-group "
+ + "nested inside a struct, array or map stays unshredded. Shredded
files can only be read "
+ + "back by Spark 4.1+: Spark 4.0, Hive and Flink readers fail fast
on them, so disable this "
Review Comment:
Added - the sentence now reads "Spark 4.0, Spark 3.x, Hive and Flink readers
fail fast on them".
--
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]
