This is an automated email from the ASF dual-hosted git repository.
voonhous pushed a commit to branch branch-1.2.x
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/branch-1.2.x by this push:
new eb1213214ef3 fix(variant): port #19783 to release-1.2.1 (#20039)
eb1213214ef3 is described below
commit eb1213214ef3a742583e3d5326a07a99ccb0c758
Author: voonhous <[email protected]>
AuthorDate: Wed Sep 30 12:24:05 2026 +0800
fix(variant): port #19783 to release-1.2.1 (#20039)
* test(spark): pushVariantIntoScan legs for release-1.2.1
Ported from master's TestVariantShreddingMixedLayouts (#19688, #19783)
over unshredded data. Evidence for the #19783 backport decision: on
Spark 4.1.1 the MOR pushVariantIntoScan=true legs fail (top-level: a
null variant written through a log is lost by 'v is null'; nested:
SIGBUS in StubRoutines::forward_copy_longs), every COW leg and every
pushVariantIntoScan=false leg passes.
* fix(variant): port #19783 to release-1.2.1
Under Spark 4.1's default spark.sql.variant.pushVariantIntoScan=true, a
MOR read that merges a log file lost the row of a variant nulled through
the log (`where v is null`) and failed the task on a struct-nested
variant (SparkOutOfMemoryError on x86, SIGBUS on aarch64): the reader
context overlaid and projected the pushed-down struct at the top level
only, and the projector had no null guard.
Hand-port of master c59987a024cd (#19783); a cherry-pick conflicts in
10 of 13 files on this branch. The overlay and its detection recurse
into struct members, SparkAdapter gains containsVariantProjection, the
4.1 projector is recursive and projects a null variant to a NULL
struct, and the SparkSchemaTransformUtils arm and the two adapter unit
tests are applied verbatim. The comment-only CDC and MergeOnReadRDDV2
hunks are not ported.
Closes #20037
* fix(variant): read nested variants row-based
Port the recursive supportBatch gates from #19777 so a variant or a
PushVariantIntoScan projection nested in a struct, array or map reads
row-based, as a top-level one already does. Without it a nested
variant still took the vectorized path on Spark 4.1 (#18605).
Drop the VARIANT_LEGS / VARIANT_SKIP_LEGS leg filter from
TestVariantPushVariantIntoScan so the suite cannot pass as skipped.
---
.../SparkFileFormatInternalRowReaderContext.scala | 19 ++-
.../datasources/SparkSchemaTransformUtils.scala | 13 ++
.../org/apache/spark/sql/hudi/SparkAdapter.scala | 12 ++
.../HoodieFileGroupReaderBasedFileFormat.scala | 38 ++++-
.../TestBaseSpark4AdapterVariantMethods.scala | 109 +++++++++++-
.../schema/TestVariantPushVariantIntoScan.scala | 187 +++++++++++++++++++++
.../apache/spark/sql/adapter/Spark4_1Adapter.scala | 82 ++++++---
7 files changed, 420 insertions(+), 40 deletions(-)
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
index edfcc44e9aa2..1429f5d5363f 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/hudi/SparkFileFormatInternalRowReaderContext.scala
@@ -84,12 +84,19 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
private lazy val allFilters = filters ++ requiredFilters
// For each field of `target`, replace its dataType with the matching
field's projected
- // variant struct from `source` (when present). Non-matching fields pass
through.
+ // variant struct from `source` (when present), recursing into struct
members so a variant
+ // reached through a struct path is overlaid too. Fields are matched by name
(findFieldByName);
+ // non-matching fields pass through. The recursion mirrors
PushVariantIntoScan's
+ // VariantInRelation.rewriteType, which rewrites variants at the root of the
relation output
+ // and below STRUCT paths only, so an array element or a map value is never
overlaid here
+ // either (#19783).
private def overlayVariantProjections(target: StructType, source:
StructType): StructType = {
StructType(target.fields.map { f =>
- SparkFileFormatInternalRowReaderContext.findFieldByName(source,
f.name).map(_.dataType) match {
- case Some(projStruct: StructType) if
sparkAdapter.isVariantProjectionStruct(projStruct) =>
+ (f.dataType,
SparkFileFormatInternalRowReaderContext.findFieldByName(source,
f.name).map(_.dataType)) match {
+ case (_, Some(projStruct: StructType)) if
sparkAdapter.isVariantProjectionStruct(projStruct) =>
f.copy(dataType = projStruct)
+ case (targetStruct: StructType, Some(sourceStruct: StructType)) =>
+ f.copy(dataType = overlayVariantProjections(targetStruct,
sourceStruct))
case _ => f
}
})
@@ -101,10 +108,8 @@ class
SparkFileFormatInternalRowReaderContext(baseFileReader: SparkColumnarFileR
// bare required schema would drop them and the merger would read garbage
offsets.
override def getLogBlockRecordProjection(
dataBlockSchema: HoodieSchema): HOption[JFunction[InternalRow,
InternalRow]] = {
- val needsProjection = sparkRequiredSchema.exists(_.fields.exists(f =>
f.dataType match {
- case st: StructType => sparkAdapter.isVariantProjectionStruct(st)
- case _ => false
- }))
+ val needsProjection =
+ sparkRequiredSchema.exists(_.fields.exists(f =>
sparkAdapter.containsVariantProjection(f.dataType)))
if (!needsProjection) {
return HOption.empty[JFunction[InternalRow, InternalRow]]()
}
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
index c2676c710b9b..3150216c5e1e 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/execution/datasources/SparkSchemaTransformUtils.scala
@@ -426,6 +426,13 @@ object SparkSchemaTransformUtils {
case (ArrayType(rt, _), ArrayType(ft, _)) =>
ArrayType(addMissingFields(rt, ft))
case (MapType(requiredKey, requiredValue, _), MapType(fileKey, fileValue,
_)) =>
MapType(addMissingFields(requiredKey, fileKey),
addMissingFields(requiredValue, fileValue))
+ // A Spark 4.1 variant projection struct requested over a file VariantType
is the pair
+ // isDataTypeEqual declares equal through the adapter, and the reader has
to be handed the
+ // projection struct, not the file's VariantType: parquet-mr decodes into
the projected shape
+ // natively, while the type-change Cast applied afterwards would cast the
variant VALUE to that
+ // struct. Without this arm a type change on a SIBLING member of the
enclosing struct rewrote
+ // the projection back to VariantType (#19775).
+ case (requiredStruct: StructType, _) if
isVariantProjectionOverVariant(requiredStruct, fileType) => requiredStruct
case (StructType(requiredFields), StructType(fileFields)) =>
val fileFieldMap = fileFields.map(f => f.name -> f).toMap
StructType(requiredFields.map(f => {
@@ -436,4 +443,10 @@ object SparkSchemaTransformUtils {
}))
case _ => fileType
}
+
+ // Same Try as in isDataTypeEqual: the adapter module may be absent from the
classpath
+ // (hudi-spark-client tests), in which case there is no projection struct to
recognise anyway.
+ private def isVariantProjectionOverVariant(requiredType: StructType,
fileType: DataType): Boolean =
+ Try(HoodieSparkUtils.sparkAdapter.isVariantProjectionStruct(requiredType)
+ &&
HoodieSparkUtils.sparkAdapter.isVariantType(fileType)).getOrElse(false)
}
diff --git
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
index 2934536f4ec9..00e609f09915 100644
---
a/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
+++
b/hudi-client/hudi-spark-client/src/main/scala/org/apache/spark/sql/hudi/SparkAdapter.scala
@@ -490,6 +490,18 @@ trait SparkAdapter extends Serializable {
*/
def isVariantProjectionStruct(structType: StructType): Boolean = false
+ /**
+ * True when `dataType` is a variant projection struct or holds one below a
struct path: the two
+ * places PushVariantIntoScan puts them (the root of the relation output and
STRUCT members), so
+ * an array element or a map value never matches. The reader context's
schema overlay and the
+ * adapter's row projector both key off this, and they have to agree on it.
+ */
+ def containsVariantProjection(dataType: DataType): Boolean = dataType match {
+ case st: StructType =>
+ isVariantProjectionStruct(st) || st.fields.exists(f =>
containsVariantProjection(f.dataType))
+ case _ => false
+ }
+
/**
* If `sparkRequiredSchema` contains any field that's a Spark 4.1 variant
projection struct
* (i.e., the same-named field in `sparkDataSchema` is `VariantType`),
returns a row
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
index 34e477895c8b..1955f45c2622 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/HoodieFileGroupReaderBasedFileFormat.scala
@@ -55,7 +55,7 @@ import
org.apache.spark.sql.execution.vectorized.{OffHeapColumnVector, OnHeapCol
import org.apache.spark.sql.hudi.MultipleColumnarFileFormatReader
import org.apache.spark.sql.internal.SQLConf
import org.apache.spark.sql.sources.Filter
-import org.apache.spark.sql.types.StructType
+import org.apache.spark.sql.types.{ArrayType, DataType, MapType, StructType}
import org.apache.spark.sql.vectorized.{ColumnarBatch, ColumnarBatchUtils}
import org.apache.spark.util.SerializableConfiguration
@@ -133,6 +133,25 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
}
}
+ /**
+ * True when `dataType` itself, or anything it nests, satisfies `predicate`.
A struct that
+ * matches is reported without descending into it, so a variant projection
struct is answered
+ * as one type rather than as its pushed-down extraction fields.
+ */
+ private def containsType(dataType: DataType, predicate: DataType =>
Boolean): Boolean = {
+ predicate(dataType) || (dataType match {
+ case s: StructType => s.fields.exists(f => containsType(f.dataType,
predicate))
+ case a: ArrayType => containsType(a.elementType, predicate)
+ case m: MapType => containsType(m.valueType, predicate)
+ case _ => false
+ })
+ }
+
+ private def isVariantProjection(dataType: DataType): Boolean = dataType
match {
+ case s: StructType => sparkAdapter.isVariantProjectionStruct(s)
+ case _ => false
+ }
+
/**
* Checks if the file format supports vectorized reading, please refer to
SPARK-40918.
*
@@ -151,8 +170,7 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
supportVectorizedRead = false
supportReturningBatch = false
false
- } else if (schema.fields.exists(f => f.dataType.isInstanceOf[StructType]
- &&
sparkAdapter.isVariantProjectionStruct(f.dataType.asInstanceOf[StructType]))) {
+ } else if (schema.fields.exists(f => containsType(f.dataType,
isVariantProjection))) {
// Spark 4.1's PushVariantIntoScan rewrites a variant column to a struct
of pushed-down
// extractions. The Spark vectorized parquet reader treats this as a
nested type change
// (data column is VariantType, required is a struct) and refuses to
read in vectorized
@@ -160,9 +178,21 @@ class HoodieFileGroupReaderBasedFileFormat(tablePath:
String,
supportVectorizedRead = false
supportReturningBatch = false
false
- } else if (HoodieSparkUtils.gteqSpark4_1 && schema.fields.exists(f =>
sparkAdapter.isVariantType(f.dataType))) {
+ } else if (HoodieSparkUtils.gteqSpark4_1
+ && schema.fields.exists(f => containsType(f.dataType,
sparkAdapter.isVariantType))) {
// #18605: Spark 4.1's vectorized variant read produces UnsafeRow
encodings that SIGBUS
// during RangePartitioner sampling. Force row-based reads. Spark 4.0
unaffected.
+ //
+ // Both checks above walk the schema instead of scanning top-level
fields only. Spark's
+ // ParquetUtils.isBatchReadSupported treats VariantType as an atomic
type and, once
+ // spark.sql.parquet.enableNestedColumnVectorizedReader is on - on by
default since Spark
+ // 3.4.0 (added in 3.3.0, off) - nested columns as batch-readable, so at
stock settings a
+ // variant reached the vectorized reader whenever it sat inside a
struct/array/map even
+ // though a top-level one did not. The SIGBUS is in the UnsafeRow
encoding of the vectorized
+ // variant vectors that RangePartitioner samples, and it samples WHOLE
rows, so a variant
+ // carried inside a struct is exposed exactly the same way under any
range-partitioned query.
+ // The cost is that nested-variant tables lose vectorization on 4.1+ by
default, exactly as
+ // top-level ones do.
supportVectorizedRead = false
supportReturningBatch = false
false
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/adapter/TestBaseSpark4AdapterVariantMethods.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/adapter/TestBaseSpark4AdapterVariantMethods.scala
index 8f14dd1ceed8..b9a7dbe07500 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/adapter/TestBaseSpark4AdapterVariantMethods.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/adapter/TestBaseSpark4AdapterVariantMethods.scala
@@ -23,7 +23,10 @@ import org.apache.hudi.common.schema.{HoodieSchema,
HoodieSchemaType}
import org.apache.parquet.schema.PrimitiveType
import org.apache.parquet.schema.Type.Repetition
-import org.apache.spark.sql.types.{BinaryType, IntegerType, StringType,
StructField, StructType}
+import org.apache.spark.sql.catalyst.InternalRow
+import org.apache.spark.sql.execution.datasources.SparkSchemaTransformUtils
+import org.apache.spark.sql.types.{BinaryType, IntegerType, LongType,
Metadata, MetadataBuilder, StringType, StructField, StructType}
+import org.apache.spark.unsafe.types.UTF8String
import org.junit.jupiter.api.Assertions.{assertEquals, assertFalse,
assertThrows, assertTrue}
import org.junit.jupiter.api.Assumptions.assumeTrue
import org.junit.jupiter.api.Test
@@ -259,4 +262,108 @@ class TestBaseSpark4AdapterVariantMethods extends
SparkAdapterSupport {
val result = sparkAdapter.isDataTypeEqualForPhysicalSchema(invalidStruct,
variantType)
assertTrue(result.isEmpty, "Should return None when reversed struct
doesn't match variant physical schema")
}
+
+ /**
+ * The per-child metadata PushVariantIntoScan attaches to a projection
struct. Built key by key
+ * rather than through Spark 4.1's `VariantMetadata`, which does not exist
on the Spark 3.x
+ * profiles this suite also compiles for; the test pins the shape against
the adapter before it
+ * relies on it.
+ */
+ private def variantProjectionMetadata(path: String): Metadata =
+ new MetadataBuilder().putMetadata("__VARIANT_METADATA_KEY",
+ new MetadataBuilder()
+ .putString("path", path)
+ .putBoolean("failOnError", true)
+ .putString("timeZoneId", "UTC")
+ .build()).build()
+
+ /** A `VariantVal` holding `json`, reached by reflection for the same
cross-version reason. */
+ private def parseJson(json: String): Any = {
+ val cls =
Class.forName("org.apache.spark.sql.catalyst.expressions.variant.VariantExpressionEvalUtils$")
+ val module = cls.getField("MODULE$").get(null)
+ // parseJson(UTF8String, allowDuplicateKeys, ...): the trailing flags
gained an entry across
+ // 4.x, so every one of them is passed as its default false.
+ val method = cls.getMethods.find(m => m.getName == "parseJson"
+ && m.getParameterTypes.headOption.contains(classOf[UTF8String])).get
+ val args: Seq[AnyRef] = UTF8String.fromString(json) +:
+ Seq.fill(method.getParameterCount - 1)(Boolean.box(false))
+ method.invoke(module, args: _*)
+ }
+
+ @Test
+ def testBuildVariantProjectorRecursesIntoStructPaths(): Unit = {
+ assumeTrue(HoodieSparkUtils.gteqSpark4_1, "Variant projection structs only
exist on Spark 4.1+")
+ val variantType = sparkAdapter.getVariantDataType.get
+ val projectionStruct = StructType(Seq(
+ StructField("0", StringType, metadata =
variantProjectionMetadata("$.k"))))
+ assertTrue(sparkAdapter.isVariantProjectionStruct(projectionStruct),
+ "the hand-built metadata must be what Spark recognizes as a projection
struct")
+
+ // `cast(s.inner as string)` on `s struct<inner: variant, other: int>`:
PushVariantIntoScan
+ // rewrites the variant one struct member down, leaving the struct itself
in place.
+ val dataSchema = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("s", StructType(Seq(
+ StructField("inner", variantType),
+ StructField("other", IntegerType))))))
+ val requiredSchema = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("s", StructType(Seq(
+ StructField("inner", projectionStruct),
+ StructField("other", IntegerType))))))
+
+ val projector = sparkAdapter.buildVariantProjector(dataSchema,
requiredSchema)
+ assertTrue(projector.isDefined, "a projection below a struct path must
build a projector")
+ val project = projector.get
+
+ val row = project(InternalRow(1, InternalRow(parseJson("""{"k":"n1"}"""),
7)))
+ assertEquals(1, row.getInt(0))
+ val struct = row.getStruct(1, 2)
+ assertEquals("n1", struct.getStruct(0, 1).getUTF8String(0).toString,
+ "the nested variant must be projected to its requested extraction")
+ assertEquals(7, struct.getInt(1), "the sibling of the projected variant
must survive")
+
+ // CreateNamedStruct is never null on its own, so a null struct has to be
preserved explicitly.
+ assertTrue(project(InternalRow(2, null)).isNullAt(1), "a null struct must
stay null")
+
+ // A null variant inside a live struct: its projection struct must be
NULL, not a struct of
+ // nulls, because PushVariantIntoScan rewrites `s.inner is null` onto that
struct.
+ val nullVariant = project(InternalRow(3, InternalRow(null, 9)))
+ val liveStruct = nullVariant.getStruct(1, 2)
+ assertTrue(liveStruct.isNullAt(0), "a null variant must project to a null
struct")
+ assertEquals(9, liveStruct.getInt(1), "the sibling of a null variant must
survive")
+
+ assertTrue(sparkAdapter.buildVariantProjector(dataSchema,
dataSchema).isEmpty,
+ "a required schema with no projection struct anywhere must not build a
projector")
+ }
+
+ @Test
+ def testImplicitSchemaChangeKeepsNestedVariantProjection(): Unit = {
+ assumeTrue(HoodieSparkUtils.gteqSpark4_1, "Variant projection structs only
exist on Spark 4.1+")
+ val variantType = sparkAdapter.getVariantDataType.get
+ val projectionStruct = StructType(Seq(
+ StructField("0", StringType, metadata =
variantProjectionMetadata("$.k"))))
+
+ // The file wrote s.n as int and the table has since widened it to long,
so `s` is an implicit
+ // type change on this file and every member of `s` is reconciled by
addMissingFields --
+ // including the projected variant next to the widened column.
+ val fileStruct = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("s", StructType(Seq(
+ StructField("inner", variantType),
+ StructField("n", IntegerType))))))
+ val requiredSchema = StructType(Seq(
+ StructField("id", IntegerType),
+ StructField("s", StructType(Seq(
+ StructField("inner", projectionStruct),
+ StructField("n", LongType))))))
+
+ val (typeChanges, readerSchema) =
SparkSchemaTransformUtils.buildImplicitSchemaChangeInfo(fileStruct,
requiredSchema)
+ val readerS = readerSchema("s").dataType.asInstanceOf[StructType]
+ assertEquals(projectionStruct, readerS("inner").dataType,
+ "the reader must be handed the projection struct, not the file's
VariantType")
+ assertEquals(IntegerType, readerS("n").dataType, "the widened sibling is
still read at its file type")
+ assertTrue(typeChanges.containsKey(1), "s is an implicit type change")
+ assertEquals(readerS, typeChanges.get(1).getRight, "the recorded reader
type is the reconciled struct")
+ }
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
new file mode 100644
index 000000000000..790317b5c86d
--- /dev/null
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantPushVariantIntoScan.scala
@@ -0,0 +1,187 @@
+/*
+ * 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.hudi.dml.schema
+
+import org.apache.hudi.{HoodieSparkUtils, SparkAdapterSupport}
+import org.apache.hudi.common.model.HoodieRecord.HoodieRecordType
+
+import org.apache.spark.sql.execution.FileSourceScanExec
+import org.apache.spark.sql.hudi.common.HoodieSparkSqlTestBase
+import org.apache.spark.sql.types.{DataType, StructType}
+
+import scala.collection.mutable.ListBuffer
+import scala.util.{Failure, Success, Try}
+
+/**
+ * The pushVariantIntoScan legs of master's TestVariantShreddingMixedLayouts
(#19688, #19783),
+ * ported to release-1.2.1 over unshredded data only, since this branch has no
shredding writer.
+ * spark.sql.variant.pushVariantIntoScan is swept on and off: both arms expect
the very same rows,
+ * and only the plan tells them apart (whether the rule rewrote the variant
into a projection
+ * struct inside the scan). Record types are swept too, which on table version
9 gives avro
+ * (AVRO) and parquet (SPARK) log blocks on the MOR legs.
+ *
+ * Every leg runs to the end and its verdict is recorded, so one wrong result
does not hide the
+ * others.
+ */
+class TestVariantPushVariantIntoScan extends HoodieSparkSqlTestBase {
+
+ private val SPARK_4_1_GATE = "PushVariantIntoScan is on by default from
Spark 4.1"
+
+ // ids 0-2 keep their inserted value, id 3 is updated and id 4's variant is
nulled by the update.
+ private def merged(id: Int): String = if (id < 3) s"x$id" else if (id == 3)
"y3" else null
+ private def mergedJson(id: Int): String = Option(merged(id)).map(k =>
s"""{"k":"$k"}""").orNull
+
+ private def variantProjectionPushedIntoScan(sql: String): Boolean = {
+ def containsProjection(dataType: DataType): Boolean = dataType match {
+ case st: StructType =>
+ SparkAdapterSupport.sparkAdapter.isVariantProjectionStruct(st) ||
+ st.fields.exists(f => containsProjection(f.dataType))
+ case _ => false
+ }
+ val scans = spark.sql(sql).queryExecution.sparkPlan.collect { case scan:
FileSourceScanExec => scan }
+ assert(scans.nonEmpty, s"expected a file scan in the plan of: $sql")
+ scans.exists(_.requiredSchema.fields.exists(f =>
containsProjection(f.dataType)))
+ }
+
+ private def assertPushed(sql: String, pushed: Boolean, leg: String): Unit = {
+ val verdict = if (pushed) "should have" else "must not have"
+ assert(variantProjectionPushedIntoScan(sql) == pushed,
+ s"[$leg] PushVariantIntoScan $verdict rewritten the variant into a
projection struct")
+ }
+
+ /**
+ * Sweeps table type x pushVariantIntoScan x record type, runs `body` once
per leg on a fresh
+ * table, and fails at the end with every leg that failed.
+ */
+ private def sweep(scope: String)(body: (String, String, String, Boolean) =>
Unit): Unit = {
+ val failures = ListBuffer.empty[String]
+ var ran = 0
+ // The conf-off arm runs first so that, when a conf-on leg crashes the
JVM, every other verdict
+ // is already in the log.
+ Seq("cow", "mor").foreach { tableType =>
+ Seq("false", "true").foreach { pushIntoScan =>
+ Seq(HoodieRecordType.AVRO, HoodieRecordType.SPARK).foreach {
recordType =>
+ val key = s"$scope:$tableType:$pushIntoScan:$recordType"
+ ran += 1
+ withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan)
{
+ withRecordType(Seq(recordType))(withTempDir { tmp =>
+ val tableName = generateTableName
+ val leg = s"$key, $tableName"
+ // scalastyle:off println
+ println(s"LEG START $key")
+ Try(body(tableName, tmp.getCanonicalPath, tableType,
pushIntoScan.toBoolean)) match {
+ case Success(_) => println(s"LEG PASS $key")
+ case Failure(e) =>
+ val msg =
Option(e.getMessage).getOrElse(e.toString).linesIterator.take(3).mkString(" | ")
+ println(s"LEG FAIL $key: ${e.getClass.getSimpleName}: $msg")
+ failures += s"[$leg] ${e.getClass.getSimpleName}: $msg"
+ }
+ // scalastyle:on println
+ })
+ }
+ }
+ }
+ }
+ assert(failures.isEmpty, s"${failures.size} of $ran $scope legs failed:\n"
+ failures.mkString("\n"))
+ }
+
+ private def createTable(tableName: String, tablePath: String, tableType:
String, variantColumnDdl: String): Unit = {
+ spark.sql(
+ s"""
+ |create table $tableName (
+ | id int,
+ | $variantColumnDdl,
+ | ts long
+ |) using hudi
+ | location '$tablePath'
+ | tblproperties (
+ | primaryKey = 'id',
+ | preCombineField = 'ts',
+ | type = '$tableType',
+ | hoodie.compact.inline = 'false'
+ | )
+ """.stripMargin)
+ }
+
+ test("top-level variant_get projections and filters under
pushVariantIntoScan on and off") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ sweep("toplevel") { (tableName, tablePath, tableType, pushed) =>
+ createTable(tableName, tablePath, tableType, "v variant")
+ spark.sql(s"insert into $tableName select cast(id as int) as id, " +
+ """parse_json(concat('{"k":"x', id, '"}')) as v, 1000L as ts from
range(0, 5, 1, 1)""")
+ // ids >= 3 go through a log file on MOR (a rewritten base file on COW);
id 4 is nulled.
+ spark.sql(s"update $tableName set " +
+ """v = parse_json(case when id = 4 then cast(null as string) """ +
+ """else concat('{"k":"y', id, '"}') end), ts = 1001 """ +
+ "where id >= 3")
+
+ checkAnswer(s"select id, variant_get(v, '$$.k', 'string') from
$tableName order by id")(
+ (0 until 5).map(id => Seq(id, merged(id))): _*)
+ checkAnswer(s"select id from $tableName where variant_get(v, '$$.k',
'string') = 'y3'")(Seq(3))
+ checkAnswer(s"select id from $tableName where variant_get(v, '$$.k',
'string') = 'x1'")(Seq(1))
+ checkAnswer(s"select id, cast(v as string) from $tableName order by id")(
+ (0 until 5).map(id => Seq(id, mergedJson(id))): _*)
+ // The rule rewrites `v is null` onto the projection struct itself, so a
null variant
+ // projected as a struct OF nulls instead of a NULL struct loses this
row.
+ checkAnswer(s"select id from $tableName where v is null")(Seq(4))
+ checkAnswer(s"select count(*) from $tableName where v is not
null")(Seq(4))
+ assertPushed(s"select id, variant_get(v, '$$.k', 'string') from
$tableName", pushed, tableName)
+ }
+ }
+
+ test("nested variant_get projections and filters under pushVariantIntoScan
on and off") {
+ assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+
+ // MOR with a log file is the #19783 case: before that fix the internal
reader applied the
+ // PushVariantIntoScan projection to TOP-LEVEL fields only, so the merged
row still held a raw
+ // VariantVal at s.inner while the plan read that memory as the projected
struct s.inner.0
+ // (SIGBUS / InternalError / OOM, or silent nulls).
+ sweep("nested") { (tableName, tablePath, tableType, pushed) =>
+ createTable(tableName, tablePath, tableType, "s struct<inner: variant>")
+ spark.sql(s"insert into $tableName select cast(id as int) as id, " +
+ """named_struct('inner', parse_json(concat('{"k":"x', id, '"}'))) as
s, """ +
+ "1000L as ts from range(0, 5, 1, 1)")
+ spark.sql(s"update $tableName set " +
+ """s = named_struct('inner', parse_json(case when id = 4 then
cast(null as string) """ +
+ """else concat('{"k":"y', id, '"}') end)), ts = 1001 """ +
+ "where id >= 3")
+
+ checkAnswer(s"select id, variant_get(s.inner, '$$.k', 'string') from
$tableName order by id")(
+ (0 until 5).map(id => Seq(id, merged(id))): _*)
+ checkAnswer(s"select id from $tableName where variant_get(s.inner,
'$$.k', 'string') = 'y3'")(Seq(3))
+ checkAnswer(s"select id from $tableName where variant_get(s.inner,
'$$.k', 'string') = 'x1'")(Seq(1))
+ checkAnswer(s"select id, cast(s.inner as string) from $tableName order
by id")(
+ (0 until 5).map(id => Seq(id, mergedJson(id))): _*)
+ checkAnswer(s"select id from $tableName where s.inner is null")(Seq(4))
+ checkAnswer(s"select count(*) from $tableName where s.inner is not
null")(Seq(4))
+ // The whole struct: no extraction, nothing is rewritten, native
VariantType at depth.
+ val wholeStruct = spark.sql(s"select id, s from $tableName order by
id").collect()
+ assert(wholeStruct.length == 5, s"[$tableName] whole-struct read should
return 5 rows")
+ wholeStruct.foreach { row =>
+ val id = row.getInt(0)
+ assert(!row.isNullAt(1), s"[$tableName] whole-struct read nulled out s
for id $id")
+ val inner = row.getStruct(1).getAs[Any]("inner")
+ assert(Option(inner).map(_.toString).orNull == mergedJson(id),
+ s"[$tableName] whole-struct read of s.inner for id $id should be
${mergedJson(id)}, got $inner")
+ }
+ assertPushed(s"select id, variant_get(s.inner, '$$.k', 'string') from
$tableName", pushed, tableName)
+ }
+ }
+}
diff --git
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
index f6a8e19b0e6c..b8a779968caa 100644
---
a/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
+++
b/hudi-spark-datasource/hudi-spark4.1.x/src/main/scala/org/apache/spark/sql/adapter/Spark4_1Adapter.scala
@@ -32,7 +32,7 @@ import org.apache.spark.sql.avro._
import org.apache.spark.sql.catalyst.InternalRow
import org.apache.spark.sql.catalyst.analysis.{EliminateSubqueryAliases,
ResolvedTable}
import org.apache.spark.sql.catalyst.catalog.CatalogTable
-import org.apache.spark.sql.catalyst.expressions.{BoundReference,
CreateNamedStruct, Expression, Literal, UnsafeProjection}
+import org.apache.spark.sql.catalyst.expressions.{BoundReference,
CreateNamedStruct, Expression, GetStructField, If, IsNull, Literal,
UnsafeProjection}
import org.apache.spark.sql.catalyst.expressions.variant.VariantGet
import org.apache.spark.sql.catalyst.parser.{ParseException, ParserInterface}
import org.apache.spark.sql.catalyst.planning.PhysicalOperation
@@ -230,39 +230,65 @@ class Spark4_1Adapter extends BaseSpark4Adapter {
override def buildVariantProjector(sparkDataSchema: StructType,
sparkRequiredSchema: StructType):
Option[InternalRow => InternalRow] = {
- // Quick check: any required field a variant projection struct?
- if (!sparkRequiredSchema.fields.exists(f =>
VariantMetadata.isVariantStruct(f.dataType))) {
+ // 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.
- def lookupDataField(name: String): (Int, StructField) = {
- val idx = sparkDataSchema.getFieldIndex(name).getOrElse(
+ // 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 '$name' is absent from sparkDataSchema; " +
- s"required=${sparkRequiredSchema.fieldNames.mkString("[", ",",
"]")}, " +
- s"data=${sparkDataSchema.fieldNames.mkString("[", ",", "]")}"))
- (idx, sparkDataSchema.fields(idx))
+ 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),
DataTypes.StringType)
+ val variantGet: Expression =
+ VariantGet(ref, pathLit, child.dataType, vm.failOnError,
Option(vm.timeZoneId))
+ Seq(Literal(UTF8String.fromString(child.name),
DataTypes.StringType), variantGet)
+ }
+ val projected = CreateNamedStruct(childExprs)
+ // A null variant has to come out as a NULL struct, not a struct of
nulls: CreateNamedStruct
+ // is never null, the parquet paths leave the field null, and
PushVariantIntoScan rewrites
+ // IsNull(v) / IsNotNull(v) onto this struct directly.
+ If(IsNull(ref), Literal(null, projected.dataType), projected)
+ 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),
DataTypes.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 =>
- rf.dataType match {
- case projectedStruct: StructType if
VariantMetadata.isVariantStruct(projectedStruct) =>
- val (dataIdx, dataField) = lookupDataField(rf.name)
- require(isVariantType(dataField.dataType),
- s"Expected VariantType for field '${rf.name}' in data schema,
got ${dataField.dataType}")
- val variantRef: Expression = BoundReference(dataIdx,
dataField.dataType, dataField.nullable)
- val childExprs: Seq[Expression] =
projectedStruct.fields.toSeq.flatMap { child =>
- val vm = VariantMetadata.fromMetadata(child.metadata)
- val pathLit = Literal(UTF8String.fromString(vm.path),
DataTypes.StringType)
- val tz: Option[String] = Option(vm.timeZoneId)
- val variantGet: Expression = VariantGet(variantRef, pathLit,
child.dataType, vm.failOnError, tz)
- Seq(Literal(UTF8String.fromString(child.name),
DataTypes.StringType), variantGet)
- }
- CreateNamedStruct(childExprs)
- case _ =>
- val (dataIdx, dataField) = lookupDataField(rf.name)
- BoundReference(dataIdx, dataField.dataType, dataField.nullable)
- }
+ 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))
