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 85f9de41e04f test(variant): pin the scan ReadSchema per arm (#20155)
85f9de41e04f is described below
commit 85f9de41e04f6f73107d94ceba3342bdb984add5
Author: voonhous <[email protected]>
AuthorDate: Wed Sep 30 16:41:33 2026 +0800
test(variant): pin the scan ReadSchema per arm (#20155)
Both arms of a spark.sql.variant.pushVariantIntoScan sweep return the
same rows, so only the plan proves that the true arm ran with the
rewrite. The other sweeps of TestVariantShreddingMixedLayouts assert
that through variantProjectionPushedIntoScan; the mixed-layout
variant_get test, the first sweep of the suite, had no plan assertion
on its COW and MOR legs.
Add scanReadSchema to the test support, the ReadSchema of the file
scan as its explain line prints it, and pin it on those legs:
v:struct<0:bigint> with the rule on and v:variant with it off, plus
the eight-member struct behind the partial-update read, whose
expectation rests on seven of the eight pushed fields being null.
Test only.
Part of #18285.
---
.../schema/TestVariantShreddingMixedLayouts.scala | 18 ++++++++++++++
.../dml/schema/VariantShreddingTestSupport.scala | 29 ++++++++++++++++++----
2 files changed, 42 insertions(+), 5 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
index 1be340407c56..1ccaa716c343 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/TestVariantShreddingMixedLayouts.scala
@@ -594,6 +594,18 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
test("variant_get filters and projections resolve per file across mixed
layouts") {
assume(HoodieSparkUtils.gteqSpark4_1, SPARK_4_1_GATE)
+ // Both arms of a leg expect the very same rows, so the scan is what tells
them apart: with the
+ // rule on, its ReadSchema carries v as a projection struct with one
member per pushed path. A
+ // true arm that silently fell back to reading the whole variant would
show `v:variant` there.
+ def assertScanReadsV(sql: String, pushIntoScan: String, pushedPaths: Int,
leg: String): Unit = {
+ val readV = if (pushIntoScan.toBoolean) {
+ (0 until pushedPaths).map(i => s"$i:bigint").mkString("struct<", ",",
">")
+ } else {
+ "variant"
+ }
+ assert(scanReadSchema(sql) == s"struct<id:int,v:$readV>", s"[$leg]
ReadSchema of the scan of: $sql")
+ }
+
// Read-mode test; SPARK pinned (see the mixed-files test above).
Seq("true", "false").foreach { pushIntoScan =>
withSQLConf("spark.sql.variant.pushVariantIntoScan" -> pushIntoScan) {
@@ -629,6 +641,8 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
checkAnswer(s"select count(*) from $tableName where v is
null")(Seq(0))
assertVariantSegments(tableName, leg, Seq(("v", Seq(
(0 until 20, ObjA), (20 until 30, ObjAConflict), (30 until 40,
ObjB)))))
+ assertScanReadsV(s"select id, try_variant_get(v, '$$.a', 'bigint')
from $tableName",
+ pushIntoScan, pushedPaths = 1, leg)
}
}
}
@@ -667,6 +681,8 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
s"select count(*) from $tableName where try_variant_get(v, '$$.a',
'bigint') > 100")(Seq(2))
checkAnswer(
s"select id from $tableName where variant_get(v, '$$.b', 'string')
= 'b7'")(Seq(7))
+ assertScanReadsV(s"select id, try_variant_get(v, '$$.a', 'bigint')
from $tableName",
+ pushIntoScan, pushedPaths = 1, leg)
// A MERGE INTO that assigns ts alone writes a partial log block
// (hoodie.spark.sql.merge.into.partial.updates defaults to true on
MOR), so the read
@@ -682,6 +698,8 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
val widePaths = ('a' to 'h').map(c => s"try_variant_get(v, '$$.$c',
'bigint')").mkString(", ")
checkAnswer(s"select $widePaths from $tableName where id = 4")(
Seq(4L, null, null, null, null, null, null, null))
+ // The eight paths are what the scan was asked for, not what the row
happens to hold.
+ assertScanReadsV(s"select $widePaths from $tableName where id = 4",
pushIntoScan, pushedPaths = 8, leg)
}
}
}
diff --git
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
index 3e141478b488..8f1181093fa7 100644
---
a/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
+++
b/hudi-spark-datasource/hudi-spark/src/test/scala/org/apache/spark/sql/hudi/dml/schema/VariantShreddingTestSupport.scala
@@ -644,14 +644,33 @@ trait VariantShreddingTestSupport { self:
HoodieSparkSqlTestBase =>
* Whether the physical plan of `sql` reads a variant through a
PushVariantIntoScan projection
* struct: the file scan's required schema carries the projected struct at
some struct path.
* Pins that the rule actually fired for a true arm, so it cannot silently
become a copy of the
- * arm with the rule off. `sparkPlan` rather than `executedPlan`: under AQE
the latter is a
- * placeholder whose children only exist once the query runs.
+ * arm with the rule off.
*/
- protected def variantProjectionPushedIntoScan(sql: String): Boolean = {
+ protected def variantProjectionPushedIntoScan(sql: String): Boolean =
+ fileScansOf(sql).exists(_.requiredSchema.fields.exists(f =>
+ SparkAdapterSupport.sparkAdapter.containsVariantProjection(f.dataType)))
+
+ /**
+ * The ReadSchema of the one file scan in the physical plan of `sql`, as its
explain line prints
+ * it (the catalog string of the scan's required schema). A variant that
PushVariantIntoScan
+ * rewrote shows there as a projection struct with one ordinal-named member
per pushed path,
+ * `v:struct<0:bigint>`; one that reaches the reader whole shows as
`v:variant`. Where
+ * [[variantProjectionPushedIntoScan]] says whether the rule fired, this
pins what it pushed.
+ */
+ protected def scanReadSchema(sql: String): String = {
+ val scans = fileScansOf(sql)
+ assert(scans.size == 1, s"expected a single file scan in the plan of:
$sql")
+ scans.head.requiredSchema.catalogString
+ }
+
+ /**
+ * The file scans in the physical plan of `sql`. `sparkPlan` rather than
`executedPlan`: under
+ * AQE the latter is a placeholder whose children only exist once the query
runs.
+ */
+ private def fileScansOf(sql: String): Seq[FileSourceScanExec] = {
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 =>
- SparkAdapterSupport.sparkAdapter.containsVariantProjection(f.dataType)))
+ scans
}
//
---------------------------------------------------------------------------------------------
