voonhous commented on code in PR #19808:
URL: https://github.com/apache/hudi/pull/19808#discussion_r3912192564
##########
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:
Right - `SparkParquetReaderBase.read` forces `caseSensitive=false`, so the
guard has to resolve names the way Spark then does. Both sides now compare
case-insensitively: the member names in `isUnshreddedVariantStruct`, and the
lookup of a requested column or struct member in the file schema
(`parquetFieldIgnoreCase`), since a column declared `V` over a file group `v`
was the same miss one level up.
`testValidateNoShreddedVariantStructsIgnoresCase` pins `V: struct<Value,
Metadata>` at the top level and `S.Inner` below a struct; the error keeps the
requested spelling.
##########
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:
Added `TestSpark3HoodieParquetReadSupport` under hudi-spark3-common (its
first test dir; the spark-common fixtures come in through the existing test-jar
dependency), over a hand-built `InitContext`: a shredded file fails `init` with
the guard's own message, an unshredded one initialises with the column present,
and a third leg runs the guard-free base class over the shredded file to show
the clipped request carries no typed_value - which is why `init` anchors on
`context.getFileSchema` rather than on the requested schema.
--
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]
