This is an automated email from the ASF dual-hosted git repository. voonhous pushed a commit to branch variant-shredding-inference-default-on in repository https://gitbox.apache.org/repos/asf/hudi.git
commit 404c531ada8db3cc5f9fe684f413c46f7434a7ee Author: voon <[email protected]> AuthorDate: Tue Sep 1 21:31:08 2026 +0800 fix(spark3): reject shredded variant struct reads 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. --- .../parquet/ParquetSchemaEvolutionUtils.scala | 73 ++++++++++++++++++++- .../parquet/TestParquetSchemaEvolutionUtils.scala | 74 ++++++++++++++++++++-- .../Spark3LegacyHoodieParquetFileFormat.scala | 8 +++ .../datasources/parquet/Spark33ParquetReader.scala | 5 ++ .../datasources/parquet/Spark34ParquetReader.scala | 5 ++ .../datasources/parquet/Spark35ParquetReader.scala | 5 ++ 6 files changed, 165 insertions(+), 5 deletions(-) diff --git a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala index 113d7fba1358..5c04582070e5 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala @@ -20,6 +20,7 @@ package org.apache.spark.sql.execution.datasources.parquet import org.apache.hudi.SparkAdapterSupport import org.apache.hudi.client.utils.SparkInternalSchemaConverter import org.apache.hudi.common.fs.FSUtils +import org.apache.hudi.common.schema.HoodieSchema import org.apache.hudi.common.schema.internal.{InternalSchema, Type => InternalType} import org.apache.hudi.common.schema.internal.Types import org.apache.hudi.common.schema.internal.action.InternalSchemaMerger @@ -42,7 +43,7 @@ import org.apache.spark.sql.catalyst.expressions.{AttributeReference, UnsafeProj import org.apache.spark.sql.execution.datasources.SparkSchemaTransformUtils import org.apache.spark.sql.execution.datasources.parquet.ParquetSchemaEvolutionUtils.pruneInternalSchema import org.apache.spark.sql.sources._ -import org.apache.spark.sql.types.{ArrayType, AtomicType, DataType, MapType, StructType} +import org.apache.spark.sql.types.{ArrayType, AtomicType, BinaryType, DataType, MapType, StructType} import java.time.ZoneId @@ -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 && + struct.fields.forall(_.dataType == BinaryType) && + struct.fields.exists(_.name == HoodieSchema.Variant.VARIANT_METADATA_FIELD) && + struct.fields.exists(_.name == HoodieSchema.Variant.VARIANT_VALUE_FIELD) + } + /** * The dotted path of the first PushVariantIntoScan rewrite struct in the schema, if any (see * SparkInternalSchemaConverter.isVariantRewriteStruct for the marker). diff --git a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala index f462633f7347..f03ae591ef3c 100644 --- a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala +++ b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala @@ -27,15 +27,17 @@ import org.apache.hudi.exception.HoodieException import org.apache.parquet.hadoop.metadata.FileMetaData import org.apache.parquet.schema.{Type, Types} import org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{shreddedVariant, stringKeyMap, threeLevelList, twoLevelList, unshreddedVariant} -import org.apache.spark.sql.types.{BinaryType, MetadataBuilder, StructField, StructType} +import org.apache.spark.sql.types.{ArrayType, BinaryType, IntegerType, MapType, MetadataBuilder, StringType, StructField, StructType} import org.junit.jupiter.api.{Assertions, Test} import java.util.{Arrays, Collections, HashMap} /** - * Unit tests for [[ParquetSchemaEvolutionUtils.validateNoShreddedVariants]], the schema-on-read - * guard that fails a read the merged internal-schema request would otherwise serve with the - * typed_value clipped away. No SparkSession: the guard is a pure schema walk. + * Unit tests for the two shredded-variant read guards of [[ParquetSchemaEvolutionUtils]]: + * validateNoShreddedVariants, the schema-on-read guard that fails a read the merged + * internal-schema request would otherwise serve with the typed_value clipped away, and + * validateNoShreddedVariantStructs, the Spark 3.x guard for a variant requested as its + * unshredded struct shape. No SparkSession: both guards are pure schema walks. */ class TestParquetSchemaEvolutionUtils { @@ -199,6 +201,70 @@ class TestParquetSchemaEvolutionUtils { private def footerOf(column: Type): FileMetaData = new FileMetaData(Types.buildMessage().addField(column).named("test"), new HashMap[String, String](), "test") + /** + * The Spark 3.x shape: no VariantType there, so a variant column is declared as + * struct<value: binary, metadata: binary> (the shape Hive sync also writes). Either member + * order is the same column, and the unshredded twin of the same file must still read. + */ + @Test + def testValidateNoShreddedVariantStructsRejectsTopLevelShreddedVariant(): Unit = { + Seq( + ("metadata first", new StructType().add("metadata", BinaryType).add("value", BinaryType)), + ("value first", new StructType().add("value", BinaryType).add("metadata", BinaryType)) + ).foreach { case (order, requiredSchema) => + val failure = Assertions.assertThrows(classOf[HoodieException], () => + ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema, footerOf(shreddedVariant("v")))) + Assertions.assertTrue( + failure.getMessage.contains("shredded variant") && failure.getMessage.contains("'v'"), + s"The $order error must name the shredded variant column, got: ${failure.getMessage}") + + ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema, footerOf(unshreddedVariant("v"))) + } + } + + /** The walk has to reach a variant below a struct, a list element and a map value. */ + @Test + def testValidateNoShreddedVariantStructsRejectsNestedShreddedVariant(): Unit = { + Seq( + ("struct", new StructType().add("s", new StructType().add("inner", variantStruct)), + footerOf(Types.optionalGroup().addField(shreddedVariant("inner")).named("s")), "'s.inner'"), + ("list", new StructType().add("v", ArrayType(variantStruct)), + footerOf(threeLevelList("v", shreddedVariant("element"))), "'v.element'"), + ("map", new StructType().add("v", MapType(StringType, variantStruct)), + footerOf(stringKeyMap("v", shreddedVariant("value"))), "'v.value'") + ).foreach { case (leg, requiredSchema, footer, path) => + val failure = Assertions.assertThrows(classOf[HoodieException], () => + ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema, footer)) + Assertions.assertTrue(failure.getMessage.contains(path), + s"The $leg error must name $path, got: ${failure.getMessage}") + } + } + + /** + * The requested side must be the variant shape exactly, so a user struct that merely contains + * those two names, or carries them with another type, reads the same file untouched - as does a + * column the file does not hold at all. Each leg fails the test by throwing. + */ + @Test + def testValidateNoShreddedVariantStructsLeavesOtherRequestsAlone(): Unit = { + Seq( + new StructType().add("metadata", BinaryType).add("value", BinaryType).add("extra", BinaryType), + new StructType().add("metadata", BinaryType).add("value", IntegerType), + new StructType().add("a", BinaryType).add("b", BinaryType) + ).foreach { requested => + ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs( + new StructType().add("v", requested), footerOf(shreddedVariant("v"))) + } + + // A column added after the file was written has no footer field to walk. + ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs( + new StructType().add("added", variantStruct), footerOf(shreddedVariant("v"))) + } + + /** How a variant column is declared on Spark 3.x, which has no VariantType. */ + private def variantStruct: StructType = + new StructType().add("value", BinaryType).add("metadata", BinaryType) + /** * What a 2-level repeated group wraps here: a single struct field "e" holding the shredded * variant "inner". The repeated group is itself the element record, so without the name arms diff --git a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala index 1e4411943e0e..77fe3901bc6b 100644 --- a/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala +++ b/hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3LegacyHoodieParquetFileFormat.scala @@ -247,6 +247,14 @@ abstract class Spark3LegacyHoodieParquetFileFormat(shouldAppendPartitionValues: // Clone new conf val hadoopAttemptConf = new Configuration(broadcastedHadoopConf.value.value) + // A variant column is declared as its unshredded struct shape on Spark 3.x (no VariantType); + // reject a file that shreds it before either branch below, so the read fails naming the + // column instead of projecting the group by name and returning a null value for every + // shredded row. Gated like the schema-on-read guard: an empty projection (count(*)) reads no + // column data and must not pay a footer read. + if (requiredSchema.nonEmpty) { + ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema, footerFileMetaData) + } val typeChangeInfos: java.util.Map[Integer, Pair[DataType, DataType]] = if (shouldUseInternalSchema) { // Same guard as ParquetSchemaEvolutionUtils.getHadoopConfClone: schema-on-read cannot // reconstruct shredded variants, so fail loudly instead of silently dropping typed_value. diff --git a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala index becf3911c426..de60a1476563 100644 --- a/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala +++ b/hudi-spark-datasource/hudi-spark3.3.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark33ParquetReader.scala @@ -113,6 +113,11 @@ class Spark33ParquetReader(enableVectorizedReader: Boolean, } lazy val footerFileMetaData = fileFooter.getFileMetaData + // A variant column is declared as its unshredded struct shape on Spark 3.x (no VariantType); + // reject a file that shreds it here, before the reader is built, so the read fails naming the + // column instead of projecting the group by name and returning a null value for every + // shredded row. + ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema, footerFileMetaData) val datetimeRebaseSpec = DataSourceUtils.datetimeRebaseSpec( footerFileMetaData.getKeyValueMetaData.get, datetimeRebaseModeInRead) diff --git a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala index 9f09d03dba07..e94545207877 100644 --- a/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala +++ b/hudi-spark-datasource/hudi-spark3.4.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark34ParquetReader.scala @@ -110,6 +110,11 @@ class Spark34ParquetReader(enableVectorizedReader: Boolean, } lazy val footerFileMetaData = fileFooter.getFileMetaData + // A variant column is declared as its unshredded struct shape on Spark 3.x (no VariantType); + // reject a file that shreds it here, before the reader is built, so the read fails naming the + // column instead of projecting the group by name and returning a null value for every + // shredded row. + ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema, footerFileMetaData) val datetimeRebaseSpec = DataSourceUtils.datetimeRebaseSpec( footerFileMetaData.getKeyValueMetaData.get, datetimeRebaseModeInRead) diff --git a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala index e91d01d07f22..52c5ac65bbfb 100644 --- a/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala +++ b/hudi-spark-datasource/hudi-spark3.5.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark35ParquetReader.scala @@ -117,6 +117,11 @@ class Spark35ParquetReader(enableVectorizedReader: Boolean, } val footerFileMetaData = fileFooter.getFileMetaData + // A variant column is declared as its unshredded struct shape on Spark 3.x (no VariantType); + // reject a file that shreds it here, before the reader is built, so the read fails naming the + // column instead of projecting the group by name and returning a null value for every + // shredded row. + ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(requiredSchema, footerFileMetaData) val datetimeRebaseSpec = DataSourceUtils.datetimeRebaseSpec( footerFileMetaData.getKeyValueMetaData.get, datetimeRebaseModeInRead)
