This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new d058558df043 fix(spark): reorder nested variants on Spark 4.0 (#20161)
d058558df043 is described below
commit d058558df04364d614c728874fcf5672dafd881f
Author: voonhous <[email protected]>
AuthorDate: Thu Oct 1 12:24:22 2026 +0800
fix(spark): reorder nested variants on Spark 4.0 (#20161)
Spark 4.0 reads a parquet variant group by position as [value,
metadata] while Hudi writes [metadata, value] (#18334), so
Spark40HoodieParquetReadSupport reorders the group in the requested
schema. Since #18674 that reorder, when handed the catalyst schema,
covered top-level variant columns only. A variant below a struct, in
an array or in a map value kept the file order on the row-based
path: once an update sent the read through the file group reader
(MOR) or the Spark-record merge handle (COW), the read failed with
MALFORMED_VARIANT, and the COW rewrite persisted the swapped bytes
for the rows it carried over.
Walk the catalyst schema through structs, arrays and maps and reorder
every group it types as VariantType. A plain struct with the same
parquet shape is still left alone, at any depth.
CI builds no Spark 4.0 profile. Run locally on Spark 4.0.2: the new
unit test and the end-to-end test fail without the change and pass
with it.
Fixes #20157.
---
.../sql/hudi/dml/schema/TestVariantDataType.scala | 46 +++++++++++++++
.../parquet/Spark40HoodieParquetReadSupport.scala | 68 +++++++++++++++++-----
.../TestSpark40HoodieParquetReadSupport.scala | 68 +++++++++++++++++++++-
3 files changed, 166 insertions(+), 16 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
index ec8637383cbd..c0b8ad06c3f9 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantDataType.scala
@@ -198,6 +198,52 @@ class TestVariantDataType extends HoodieSparkSqlTestBase
with VariantShreddingTe
})
}
+ test("Test variants nested in structs, arrays and maps read back after an
update") {
+ assume(HoodieSparkUtils.gteqSpark4_0, "Variant type requires Spark 4.0 or
higher")
+ // Spark 4.0 reads a variant group by position as [value, metadata] and
Hudi reorders the
+ // requested group to match. Before #20157 only top-level variant columns
were reordered on the
+ // row-based path, so once an update sent the read through the file group
reader (MOR) or the
+ // Spark-record merge handle (COW), a variant below a struct, an array or
a map came back with
+ // its bytes swapped: MALFORMED_VARIANT on read, and swapped bytes on disk
after the COW rewrite.
+ Seq("cow", "mor").foreach { tableType =>
+ withRecordType()(withTempDir { tmp =>
+ val tableName = generateTableName
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | v variant,
+ | s struct<inner: variant>,
+ | arr array<variant>,
+ | m map<string, variant>,
+ | ts long
+ |) using hudi
+ | location '${tmp.getCanonicalPath}'
+ | tblproperties (
+ | primaryKey = 'id',
+ | preCombineField = 'ts',
+ | type = '$tableType',
+ | hoodie.compact.inline = 'false'
+ | )
+ """.stripMargin)
+ spark.sql(s"insert into $tableName select cast(id as int) as id, " +
+ """parse_json(concat('{"k":"v', id, '"}')) as v, """ +
+ """named_struct('inner', parse_json(concat('{"k":"s', id, '"}'))) as
s, """ +
+ """array(parse_json(concat('{"k":"a', id, '"}'))) as arr, """ +
+ """map('a', parse_json(concat('{"k":"m', id, '"}'))) as m, """ +
+ "1000L as ts from range(0, 4, 1, 1)")
+ // Assigns ts alone: ids 0 and 1 are carried over by the COW rewrite,
ids 2 and 3 go through
+ // the merge (a log file on MOR).
+ spark.sql(s"update $tableName set ts = 1001 where id >= 2")
+
+ checkAnswer(s"select id, variant_get(v, '$$.k', 'string'),
variant_get(s.inner, '$$.k', 'string'), " +
+ s"variant_get(arr[0], '$$.k', 'string'), variant_get(m['a'], '$$.k',
'string'), ts " +
+ s"from $tableName order by id")(
+ (0 until 4).map(id => Seq(id, s"v$id", s"s$id", s"a$id", s"m$id", if
(id >= 2) 1001L else 1000L)): _*)
+ })
+ }
+ }
+
test("Test Spark 4.0 rejects reads rewritten by pushVariantIntoScan") {
// The conf is off by default on Spark 4.0 and on from 4.1 (SPARK-54454).
Spark 4.0 still runs
// the rewrite once it is set but cannot read the projection struct it
produces, so Hudi fails
diff --git
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40HoodieParquetReadSupport.scala
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40HoodieParquetReadSupport.scala
index 377186036f63..f94c3560eb2a 100644
---
a/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40HoodieParquetReadSupport.scala
+++
b/hudi-spark-datasource/hudi-spark4.0.x/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/Spark40HoodieParquetReadSupport.scala
@@ -55,8 +55,8 @@ class Spark40HoodieParquetReadSupport(
override def init(context: InitContext): ReadContext = {
val baseContext = super.init(context)
// Resolve the Spark catalyst requested schema so the reorder is gated on
- // VariantType -- a user struct that happens to be <value: binary,
metadata: binary>
- // shouldn't be silently reshuffled.
+ // VariantType at any depth -- a user struct that happens to be
+ // <value: binary, metadata: binary> shouldn't be silently reshuffled.
val sparkRequestedSchema = Option(context.getConfiguration.get(
ParquetReadSupport.SPARK_ROW_REQUESTED_SCHEMA))
.map(StructType.fromString)
@@ -76,26 +76,66 @@ object Spark40HoodieParquetReadSupport {
* parquet-mr reconciles the requested schema against the file schema by
field name,
* so the correct bytes still flow to the correct converters regardless of
file order.
*
- * When a Spark catalyst schema is supplied, reorder only the top-level
fields that are
- * actually typed `VariantType` in catalyst; this prevents reshuffling a
user-defined
- * `struct<value: binary, metadata: binary>` that happens to match the
parquet shape.
+ * When a Spark catalyst schema is supplied, the walk is anchored on it and
recurses through
+ * structs, arrays and maps, so only what catalyst types as `VariantType` is
reordered, at any
+ * depth; a user-defined `struct<value: binary, metadata: binary>` that
happens to match the
+ * parquet shape is left alone wherever it sits.
*
* Shredded groups are left untouched here; rejecting them is
[[rejectShreddedVariants]]'s
- * job, which walks the whole schema rather than just the top level.
+ * job.
*/
def reorderVariantFields(schema: MessageType, sparkSchema:
Option[StructType] = None): MessageType = {
- val variantFieldNames: Set[String] = sparkSchema match {
- case Some(s) => s.fields.collect { case f if
f.dataType.isInstanceOf[VariantType] => f.name }.toSet
- case None => null
+ val reordered = sparkSchema match {
+ case Some(catalyst) => reorderFields(schema, catalyst)
+ case None => schema.getFields.asScala.map(reorderVariantType).toSeq
}
- val reordered = schema.getFields.asScala.map { f =>
- if (variantFieldNames == null || variantFieldNames.contains(f.getName)) {
- reorderVariantType(f)
- } else f
- }.toArray[Type]
Types.buildMessage().addFields(reordered: _*).named(schema.getName)
}
+ /** Reorders the fields of `group` against `struct`; fields the struct does
not name are left as they are. */
+ private def reorderFields(group: GroupType, struct: StructType): Seq[Type] =
{
+ group.getFields.asScala.map { field =>
+ struct.fields.find(_.name == field.getName) match {
+ case Some(catalystField) => reorderByCatalyst(field,
catalystField.dataType)
+ case None => field
+ }
+ }.toSeq
+ }
+
+ private def reorderByCatalyst(parquetType: Type, dataType: DataType): Type =
{
+ (parquetType, dataType) match {
+ case (group: GroupType, _: VariantType) =>
+ reorderVariantType(group)
+ case (group: GroupType, struct: StructType) =>
+ group.withNewFields(reorderFields(group, struct).asJava)
+ case (group: GroupType, array: ArrayType) =>
+ ParquetSchemaEvolutionUtils.parquetListElement(group)
+ .map(element => replaceType(group, element,
reorderByCatalyst(element, array.elementType)))
+ .getOrElse(group)
+ case (group: GroupType, map: MapType) =>
+ ParquetSchemaEvolutionUtils.parquetMapValue(group)
+ .map(value => replaceType(group, value, reorderByCatalyst(value,
map.valueType)))
+ .getOrElse(group)
+ case _ => parquetType
+ }
+ }
+
+ /**
+ * Rebuilds `parent` with its descendant `target` (matched by reference)
swapped for
+ * `replacement`. The by-reference match covers both list layouts: in the
2-level one the
+ * repeated group is itself the element, so `target` can be a direct child
of `parent`.
+ */
+ private def replaceType(parent: Type, target: Type, replacement: Type): Type
= {
+ if (parent eq target) {
+ replacement
+ } else if (parent.isPrimitive) {
+ parent
+ } else {
+ val group = parent.asGroupType()
+ group.withNewFields(group.getFields.asScala.map(replaceType(_, target,
replacement)).asJava)
+ }
+ }
+
private def reorderVariantType(t: Type): Type = {
t match {
case group: GroupType if isVariantGroup(group) &&
group.containsField("typed_value") =>
diff --git
a/hudi-spark-datasource/hudi-spark4.0.x/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark40HoodieParquetReadSupport.scala
b/hudi-spark-datasource/hudi-spark4.0.x/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark40HoodieParquetReadSupport.scala
index 1e61d93180e1..74f5852ce6fa 100644
---
a/hudi-spark-datasource/hudi-spark4.0.x/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark40HoodieParquetReadSupport.scala
+++
b/hudi-spark-datasource/hudi-spark4.0.x/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestSpark40HoodieParquetReadSupport.scala
@@ -19,11 +19,11 @@ package org.apache.spark.sql.execution.datasources.parquet
import org.apache.hudi.exception.HoodieException
+import org.apache.parquet.schema.{MessageType, Type, Types}
import org.apache.parquet.schema.PrimitiveType.PrimitiveTypeName
-import org.apache.parquet.schema.Types
import org.apache.spark.sql.execution.datasources.VariantMetadata
import
org.apache.spark.sql.execution.datasources.parquet.VariantParquetTestFixtures.{shreddedVariant,
stringKeyMap, threeLevelList, twoLevelList, unshreddedVariant}
-import org.apache.spark.sql.types.{ArrayType, BinaryType, IntegerType,
MapType, MetadataBuilder, StringType, StructField, StructType, VariantType}
+import org.apache.spark.sql.types.{ArrayType, BinaryType, DataType,
IntegerType, MapType, MetadataBuilder, StringType, StructField, StructType,
VariantType}
import org.junit.jupiter.api.{Assertions, Test}
class TestSpark40HoodieParquetReadSupport {
@@ -198,6 +198,70 @@ class TestSpark40HoodieParquetReadSupport {
Assertions.assertEquals("metadata", group.getFields.get(1).getName)
}
+ /**
+ * With a catalyst schema the reorder follows it through structs, arrays and
maps (#20157): a
+ * variant at the top level, below one or two structs, as a list element in
either layout, as a
+ * map value or inside a list's struct element comes back as [value,
metadata], and nothing
+ * else in the message changes.
+ */
+ @Test
+ def testReorderVariantFieldsReordersNestedVariants(): Unit = {
+ val result = Spark40HoodieParquetReadSupport.reorderVariantFields(
+ nestedVariantMessage(unshreddedVariant),
Some(nestedVariantCatalyst(VariantType)))
+ Assertions.assertEquals(nestedVariantMessage(valueFirstVariant), result)
+ }
+
+ /**
+ * The same parquet shapes typed as a plain struct of two binaries in
catalyst keep the file
+ * order [metadata, value] at every depth.
+ */
+ @Test
+ def testReorderVariantFieldsLeavesPlainStructsAtAnyDepth(): Unit = {
+ val schema = nestedVariantMessage(unshreddedVariant)
+ val plainStruct = new StructType().add("metadata",
BinaryType).add("value", BinaryType)
+ val result = Spark40HoodieParquetReadSupport.reorderVariantFields(
+ schema, Some(nestedVariantCatalyst(plainStruct)))
+ Assertions.assertEquals(schema, result)
+ }
+
+ /**
+ * Every variant position the catalyst-anchored reorder walks, each group
built by `variant`:
+ * v, s.inner, d.mid.inner, the 3-level list l3, the 2-level (parquet-avro)
list l2 of
+ * struct<inner>, the map m and the 3-level list items of struct<inner>.
+ */
+ private def nestedVariantMessage(variant: String => Type): MessageType = {
+ def innerStruct(name: String): Type =
Types.optionalGroup().addField(variant("inner")).named(name)
+ Types.buildMessage()
+ .addField(variant("v"))
+ .addField(innerStruct("s"))
+ .addField(Types.optionalGroup().addField(innerStruct("mid")).named("d"))
+ .addField(threeLevelList("l3", variant("element")))
+ .addField(twoLevelList("l2", "array", variant("inner")))
+ .addField(stringKeyMap("m", variant("value")))
+ .addField(threeLevelList("items", innerStruct("element")))
+ .named("test")
+ }
+
+ /** The catalyst schema of [[nestedVariantMessage]], typing each
variant-shaped group as `leafType`. */
+ private def nestedVariantCatalyst(leafType: DataType): StructType = {
+ val inner = new StructType().add("inner", leafType)
+ new StructType()
+ .add("v", leafType)
+ .add("s", inner)
+ .add("d", new StructType().add("mid", inner))
+ .add("l3", ArrayType(leafType))
+ .add("l2", ArrayType(inner))
+ .add("m", MapType(StringType, leafType))
+ .add("items", ArrayType(inner))
+ }
+
+ /** [[unshreddedVariant]] in the [value, metadata] order Spark 4.0 reads by
position. */
+ private def valueFirstVariant(name: String): Type =
+ Types.optionalGroup()
+ .addField(Types.required(PrimitiveTypeName.BINARY).named("value"))
+ .addField(Types.required(PrimitiveTypeName.BINARY).named("metadata"))
+ .named(name)
+
/**
* Spark's PushVariantIntoScan rewrite of one variant column: extraction
fields named by
* ordinal, each carrying the marker metadata the rewrite arm keys on.
