wombatu-kun commented on code in PR #19808: URL: https://github.com/apache/hudi/pull/19808#discussion_r3911864589
########## hudi-spark-datasource/hudi-spark3-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark3HoodieParquetReadSupport.scala: ########## @@ -0,0 +1,61 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.spark.sql.execution.datasources.parquet + +import org.apache.hudi.common.util.{Option => HOption} + +import org.apache.parquet.hadoop.api.InitContext +import org.apache.parquet.hadoop.api.ReadSupport.ReadContext +import org.apache.parquet.schema.MessageType +import org.apache.spark.sql.catalyst.util.RebaseDateTime.RebaseSpec +import org.apache.spark.sql.types.StructType + +import java.time.ZoneId + +/** + * The Spark 3.x [[HoodieParquetReadSupport]], rejecting a shredded variant the request cannot + * reconstruct. Mirrors [[org.apache.spark.sql.adapter.BaseSpark3Adapter#createParquetReadSupport]]'s + * Spark 4.0 sibling, which rejects at the same point for the same reason. + * + * The per-version parquet readers guard base-file reads, but they are not the only route: log + * blocks - native parquet log files and the inline blocks of an avro log file - are read by + * {@code HoodieSparkParquetReader.getUnsafeRowIterator}, which builds a {@code ParquetReader} on + * this read support instead. A shredded variant in a log block therefore only meets a guard here. + */ +class Spark3HoodieParquetReadSupport(convertTz: Option[ZoneId], + enableVectorizedReader: Boolean, + enableTimestampFieldRepair: Boolean, + datetimeRebaseSpec: RebaseSpec, + int96RebaseSpec: RebaseSpec, + tableSchemaOpt: HOption[MessageType] = HOption.empty()) + extends HoodieParquetReadSupport( + convertTz, enableVectorizedReader, enableTimestampFieldRepair, + datetimeRebaseSpec, int96RebaseSpec, tableSchemaOpt) { + + override def init(context: InitContext): ReadContext = { + val readContext = super.init(context) + // Anchored on the catalyst request and the file schema, not on the requested parquet schema: + // a Spark 3.x read asks for the variant's binary members alone, so the requested schema has + // already had typed_value clipped away by the time it gets here and only the file can show + // that the column is shredded. + Option(context.getConfiguration.get(ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA)) + .map(StructType.fromString) + .foreach(ParquetSchemaEvolutionUtils.validateNoShreddedVariantStructs(_, context.getFileSchema)) Review Comment: Nothing tests this class: the schema walk is pinned in `TestParquetSchemaEvolutionUtils`, but no test pins that `init` reads `SPARK_ROW_REQUESTED_SCHEMA` and passes the file schema, so a wiring slip here degrades silently back to the null-value read it exists to stop. A direct unit test over a hand-built `InitContext`, like `TestSpark40HoodieParquetReadSupport` does for the 4.0 sibling, would cover it. ########## hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala: ########## @@ -271,6 +272,80 @@ 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 binary members named `metadata` and + * `value`, either or both and nothing else (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 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, fileParquetSchema: MessageType): Unit = { + 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: a non-empty set of the binary members a + * variant group carries, in any order. A subset counts because Spark's nested schema pruning + * narrows the request to the leaves a query touches - `SELECT v.value` reaches this guard as a + * one-member struct (see TestNestedSchemaPruningOptimization for the pruning itself), and that + * single member is exactly the one a shredded file would return null for. + * + * Any member outside those two names exempts the struct, so a plain user struct is untouched, as + * is the {metadata, value, typed_value} shape whose caller already sees the shredded layout and + * is reading it deliberately - though pruning that shape down to `value` alone does land here, + * since nothing then distinguishes it from the variant request this guards. + */ + private def isUnshreddedVariantStruct(struct: StructType): Boolean = { + struct.fields.nonEmpty && struct.fields.forall(field => + field.dataType == BinaryType + && (field.name == HoodieSchema.Variant.VARIANT_METADATA_FIELD Review Comment: `isUnshreddedVariantStruct` compares `metadata` and `value` case-exactly, while `SparkParquetReaderBase.read` forces `spark.sql.caseSensitive=false` on the conf it hands the reader, so a column declared `struct<Value: binary, Metadata: binary>` misses the guard and is still resolved onto the shredded group. Lower-case both names before comparing, as `HoodieParquetInputFormat.isVariantShapedStruct` does. ########## hudi-common/src/main/java/org/apache/hudi/common/config/HoodieStorageConfig.java: ########## @@ -308,22 +310,28 @@ 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: Every inference test in `TestVariantDataType` still sets `hoodie.parquet.variant.shredding.schema.inference.enabled = 'true'` in tblproperties, so nothing fails if this default goes back to false. Dropping that property from one of them would pin the flip end to end. -- 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]
