voonhous commented on code in PR #19808:
URL: https://github.com/apache/hudi/pull/19808#discussion_r3912767509
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala:
##########
@@ -271,6 +272,88 @@ 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 =>
+ parquetFieldIgnoreCase(fileParquetSchema, field.name)
+ .foreach(validateNoShreddedVariantStruct(field.dataType, _,
field.name))
+ }
+ }
+
+ /**
+ * The file field a requested name resolves to. Case-insensitive on purpose:
the Spark 3.x
+ * readers this guards force spark.sql.caseSensitive=false
(SparkParquetReaderBase.read), so a
+ * column declared `V` or a member declared `Value` still lands on the
file's lower-case group,
+ * and a guard that only matched exactly would let that request straight
through to the
+ * null-value read.
+ */
+ private def parquetFieldIgnoreCase(group: GroupType, name: String):
Option[ParquetType] =
+ group.getFields.find(_.getName.equalsIgnoreCase(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)) {
Review Comment:
Added - `isShreddedVariantGroup` now requires `typed_value` next to a binary
`metadata`, the file-side anchor the Hive and Spark 4.0 guards use (`value`
stays optional, as the spec allows). A leg in
`testValidateNoShreddedVariantStructsLeavesOtherRequestsAlone` pins a user
struct `{value: binary, typed_value: int}` under a request pruned to `value`,
which now reads.
--
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]
