voonhous commented on code in PR #19687:
URL: https://github.com/apache/hudi/pull/19687#discussion_r3850695522
##########
hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala:
##########
@@ -203,4 +212,125 @@ object ParquetSchemaEvolutionUtils {
internalSchemaOpt
}
}
+
+ /**
+ * Fails fast when schema-on-read meets a shredded variant file. The
internal schema models a
+ * variant as a two-field {metadata, value} record (with sentinel negative
field ids, see
+ * InternalSchemaConverter), so the merged request clips the file's
typed_value away and the
+ * typed rows would read back with a null value residual - silent data loss.
Reconstruction
+ * under schema-on-read is tracked by #18285; until then the read must fail
loudly. The check
+ * anchors on the sentinel ids, which no real user field can carry, so plain
user structs of
+ * the same shape are left alone. The walk recurses through structs, arrays
and maps because
+ * the row writer shreds nested variants too (see VariantSchemaUtils).
+ *
+ * A scan rewritten by Spark's PushVariantIntoScan (4.x) fails fast
regardless of the file's
+ * layout: the merged internal-schema request materializes the variant as
{metadata, value}
+ * while downstream codegen expects the rewrite's ordinal-named extraction
struct, so the
+ * read cannot be served either way (pruning treats the rewritten struct as
the variant
+ * column itself, see SparkInternalSchemaConverter.isVariantRewriteStruct).
+ *
+ * Shared by [[ParquetSchemaEvolutionUtils.getHadoopConfClone]] and the
per-version legacy
+ * file formats, which carry a copy of the same schema-merge block. Callers
gate on a
+ * non-empty projection: empty-projection queries (count(*), select 1) read
no column data
+ * and must keep working, and the query schema is unpruned in that case.
+ */
+ def validateNoShreddedVariants(requiredSchema: StructType, querySchema:
InternalSchema, footerFileMetaData: FileMetaData): Unit = {
+ findVariantRewritePath(requiredSchema).foreach { path =>
+ throw new HoodieException(String.format(
+ "Column '%s' is a variant projected through Spark's variant rewrite "
+ + "(spark.sql.variant.pushVariantIntoScan) and the table is read
with schema-on-read "
+ + "(hoodie.schema.on.read.enable), which cannot reconstruct variants
(see issue "
+ + "#18285). Read without schema-on-read.", path))
+ }
+ val fileParquetSchema = footerFileMetaData.getSchema
+ querySchema.getRecord.fields().foreach { field =>
+ if (fileParquetSchema.containsField(field.name())) {
+ validateNoShreddedVariant(
+ field.`type`(),
fileParquetSchema.getType(fileParquetSchema.getFieldIndex(field.name())),
field.name())
+ }
+ }
+ }
+
+ /**
+ * The dotted path of the first PushVariantIntoScan rewrite struct in the
schema, if any (see
+ * SparkInternalSchemaConverter.isVariantRewriteStruct for the marker).
+ */
+ private def findVariantRewritePath(dataType: DataType, path: String = ""):
Option[String] = dataType match {
+ case struct: StructType if
SparkInternalSchemaConverter.isVariantRewriteStruct(struct) =>
+ Some(path)
+ case struct: StructType =>
+ struct.fields.foldLeft(Option.empty[String]) { (found, field) =>
+ found.orElse(findVariantRewritePath(field.dataType, concatPath(path,
field.name)))
+ }
+ case array: ArrayType => findVariantRewritePath(array.elementType,
concatPath(path, "element"))
+ case map: MapType => findVariantRewritePath(map.valueType,
concatPath(path, "value"))
+ case _ => None
+ }
+
+ private def concatPath(path: String, name: String): String =
+ if (path.isEmpty) name else path + "." + name
+
+ private def validateNoShreddedVariant(internalType: InternalType,
parquetType: ParquetType, path: String): Unit = {
+ internalType match {
+ // A variant: two fields, both carrying the sentinel negative ids
(BLOB's sentinel record
+ // has three). The parquet side decides shredded-ness.
+ case record: Types.RecordType if record.fields().size() == 2 &&
record.fields().forall(_.fieldId() < 0) =>
+ if (!parquetType.isPrimitive &&
parquetType.asGroupType().containsField("typed_value")) {
+ throw new HoodieException(String.format(
+ "Column '%s' is a shredded variant (typed_value present) and the
table is read "
+ + "with schema-on-read (hoodie.schema.on.read.enable), which
cannot reconstruct "
+ + "shredded variants (see issue #18285). Read without
schema-on-read, or rewrite "
+ + "the table unshredded (e.g. cluster with "
+ + "hoodie.parquet.variant.write.shredding.enabled=false).",
path))
+ }
+ case record: Types.RecordType if !parquetType.isPrimitive =>
+ val group = parquetType.asGroupType()
+ record.fields().foreach { field =>
+ if (group.containsField(field.name())) {
+ validateNoShreddedVariant(field.`type`(),
group.getType(field.name()), path + "." + field.name())
+ }
+ }
+ case array: Types.ArrayType =>
+
parquetListElement(parquetType).foreach(validateNoShreddedVariant(array.elementType(),
_, path + ".element"))
+ case map: Types.MapType =>
+
parquetMapValue(parquetType).foreach(validateNoShreddedVariant(map.valueType(),
_, path + ".value"))
+ case _ =>
+ }
+ }
+
+ /**
+ * Resolves the element type of a parquet LIST group, covering both the
3-level layout the
+ * Spark writer produces (group -> repeated "list" -> element) and the
2-level layout
+ * parquet-avro produces (group -> repeated element). The 3-level test
mirrors Spark's
+ * ParquetSchemaConverter.isElementType. An unrecognized shape returns None,
which stops the
+ * walk without failing the read.
+ */
+ private def parquetListElement(parquetType: ParquetType):
Option[ParquetType] = {
+ if (parquetType.isPrimitive || parquetType.asGroupType().getFieldCount !=
1) {
+ None
+ } else {
+ val repeated = parquetType.asGroupType().getType(0)
+ if (!repeated.isRepetition(ParquetType.Repetition.REPEATED)) {
+ None
+ } else if (!repeated.isPrimitive && repeated.asGroupType().getFieldCount
== 1
+ && repeated.getName != "array" &&
!repeated.getName.endsWith("_tuple")) {
Review Comment:
Bound to the parent name: `repeated.getName != parquetType.getName +
"_tuple"`.
--
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]
