voonhous commented on code in PR #19783:
URL: https://github.com/apache/hudi/pull/19783#discussion_r3893268251
##########
hudi-spark-datasource/hudi-spark4-common/src/main/scala/org/apache/spark/sql/adapter/BaseSpark4Adapter.scala:
##########
@@ -320,6 +324,89 @@ abstract class BaseSpark4Adapter extends SparkAdapter with
Logging {
if (rewritten) Some(StructType(fields)) else None
}
+ /**
+ * Shared implementation behind [[SparkAdapter#buildVariantProjector]] for
the 4.x adapters
+ * whose planner rewrites variants into projection structs (4.1+).
+ *
+ * Recurses into struct members, mirroring PushVariantIntoScan's
`VariantInRelation.rewriteType`:
+ * a variant is rewritten at the root of the relation output or below a
STRUCT path, while
+ * arrays and maps keep their native VariantType, so nothing under a
collection is projected
+ * here either. Before #19775 this walked top-level fields only, and a
projection struct sitting
+ * one struct member down was left holding a raw variant that the plan then
read as its
+ * projected children.
+ */
+ protected final def buildVariantProjectorForStructPaths(
+ sparkDataSchema: StructType,
+ sparkRequiredSchema: StructType): Option[InternalRow => InternalRow] = {
+ // Quick check: does any required field carry a variant projection struct,
at any depth?
+ if (!sparkRequiredSchema.fields.exists(f =>
containsVariantProjection(f.dataType))) {
+ None
+ } else {
+ // Surface mismatched schemas with both field lists rather than Spark's
bare
+ // IllegalArgumentException from fieldIndex. `path` is the dotted field
path of `name`.
+ def lookupDataField(dataStruct: StructType, requiredStruct: StructType,
+ name: String, path: String): (Int, StructField) = {
+ val idx = dataStruct.getFieldIndex(name).getOrElse(
+ throw new IllegalStateException(
+ s"Required field '$path' is absent from sparkDataSchema; " +
+ s"required=${requiredStruct.fieldNames.mkString("[", ",", "]")},
" +
+ s"data=${dataStruct.fieldNames.mkString("[", ",", "]")}"))
+ (idx, dataStruct.fields(idx))
+ }
+
+ // `ref` reads the data-schema value of type `dataType`; the result has
type `requiredType`.
+ def projectionExpr(ref: Expression, dataType: DataType, requiredType:
DataType,
+ path: String): Expression = requiredType match {
+ case projectedStruct: StructType if
VariantMetadata.isVariantStruct(projectedStruct) =>
+ require(isVariantType(dataType),
+ s"Expected VariantType for field '$path' in data schema, got
$dataType")
+ val childExprs: Seq[Expression] =
projectedStruct.fields.toSeq.flatMap { child =>
+ val vm = VariantMetadata.fromMetadata(child.metadata)
+ val pathLit = Literal(UTF8String.fromString(vm.path), StringType)
+ val variantGet: Expression =
+ VariantGet(ref, pathLit, child.dataType, vm.failOnError,
Option(vm.timeZoneId))
+ Seq(Literal(UTF8String.fromString(child.name), StringType),
variantGet)
+ }
+ CreateNamedStruct(childExprs)
+ case requiredStruct: StructType =>
+ dataType match {
+ // Rebuild the struct member by member only when something below
it is projected;
+ // otherwise the reference is already in the required shape and is
cheaper untouched.
+ case dataStruct: StructType if
containsVariantProjection(requiredStruct) =>
+ val childExprs: Seq[Expression] =
requiredStruct.fields.toSeq.flatMap { rf =>
+ val childPath = s"$path.${rf.name}"
+ val (childIdx, childField) = lookupDataField(dataStruct,
requiredStruct, rf.name, childPath)
+ val childRef = GetStructField(ref, childIdx, Some(rf.name))
+ Seq(Literal(UTF8String.fromString(rf.name), StringType),
+ projectionExpr(childRef, childField.dataType, rf.dataType,
childPath))
+ }
+ val rebuilt = CreateNamedStruct(childExprs)
+ // CreateNamedStruct is never null, so a null struct would come
back as a struct of
+ // nulls without this guard.
+ If(IsNull(ref), Literal(null, rebuilt.dataType), rebuilt)
+ case _ => ref
+ }
+ case _ => ref
+ }
+
+ val exprs: Array[Expression] = sparkRequiredSchema.fields.map { rf =>
+ val (dataIdx, dataField) = lookupDataField(sparkDataSchema,
sparkRequiredSchema, rf.name, rf.name)
+ val ref: Expression = BoundReference(dataIdx, dataField.dataType,
dataField.nullable)
+ projectionExpr(ref, dataField.dataType, rf.dataType, rf.name)
+ }
+
+ val projection = UnsafeProjection.create(exprs.toIndexedSeq,
DataTypeUtils.toAttributes(sparkDataSchema))
+ Some(row => projection(row))
+ }
+ }
+
+ /** True when `dataType` is a variant projection struct or holds one below a
struct path. */
+ private def containsVariantProjection(dataType: DataType): Boolean =
dataType match {
Review Comment:
Moved to `SparkAdapter` as a concrete method over
`isVariantProjectionStruct`; both private copies are gone.
--
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]
