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 a41f6eaf7df0 fix(spark): keep variant guard under nested filter
pushdown (#20201)
a41f6eaf7df0 is described below
commit a41f6eaf7df0de8c9aefdd80a3cbb03967623e96
Author: voonhous <[email protected]>
AuthorDate: Mon Oct 12 15:52:24 2026 +0800
fix(spark): keep variant guard under nested filter pushdown (#20201)
* test(variant): pin nested filters on the projection
With shortName() back to the base file format (#20029), Spark pushes
nested predicates into Hudi scans again. A predicate on a variant that
PushVariantIntoScan rewrote is then pushed as a filter on the
projection struct's ordinal-named member, e.g. GreaterThan(v.`0`,5).
No file carries that name as a parquet leaf, so ParquetFilters drops it
per file: it skips no row group and lands on no other column.
Pin that on the two existing variant_get filter tests: the
PushedFilters of the scan per pushVariantIntoScan arm, top-level on COW
and MOR and nested (s.inner), next to the row counts both arms already
share, plus an IS NULL count over files where the path is typed,
residual, type-conflicted and absent.
* fix(spark): drop pushed filters on unknown names
Under schema-on-read the parquet readers re-spell each pushed filter
onto the file's column names before the variant guard runs, and
InternalSchemaUtils.reBuildFilterName throws on a name the query schema
does not hold. With nested predicate pushdown restored (#20029), a
predicate on a variant that PushVariantIntoScan rewrote arrives as a
filter on the projection struct's ordinal-named member (v.`0`), so the
read died with "cannot find filter col name" instead of the guard's
message that names the conf to turn off.
Collapse such a filter to AlwaysTrue, as a filter on a column the file
lacks already is. Spark evaluates pushed filters again above the scan,
so dropping one is safe, and the read reaches
validateNoShreddedVariants as before.
TestParquetSchemaEvolutionUtils pins the new arm for every leaf filter
type; the schema-on-read fail-fast test gains a variant_get predicate
leg that expects the guard.
---
.../parquet/ParquetSchemaEvolutionUtils.scala | 24 +++++++++++++--
.../parquet/TestParquetSchemaEvolutionUtils.scala | 10 +++++--
.../schema/TestVariantShreddingMixedLayouts.scala | 35 ++++++++++++++++++++++
.../dml/schema/VariantShreddingTestSupport.scala | 24 +++++++++++----
4 files changed, 83 insertions(+), 10 deletions(-)
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
index b99ba83850e7..47ee1c02cda9 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/main/scala/org/apache/spark/sql/execution/datasources/parquet/ParquetSchemaEvolutionUtils.scala
@@ -187,7 +187,8 @@ object ParquetSchemaEvolutionUtils {
* Maps a pushed-down query filter onto the names the file actually carries:
a column renamed
* under schema-on-read is rewritten to its file-schema name, and one the
file does not hold at
* all collapses to AlwaysTrue, since a filter on an absent column cannot
skip any of its row
- * groups. A table without an internal schema passes its filters through
untouched.
+ * groups. A name the query schema does not hold at all (see fileFilterName)
collapses to
+ * AlwaysTrue as well. A table without an internal schema passes its filters
through untouched.
*/
def rebuildFilterFromParquet(oldFilter: Filter, fileSchema: InternalSchema,
querySchema: InternalSchema): Filter = {
if (fileSchema == null || querySchema == null) {
@@ -258,7 +259,7 @@ object ParquetSchemaEvolutionUtils {
Try(CatalystSqlParser.parseMultipartIdentifier(name)).toOption match {
case Some(parts) =>
val fullName = parts.mkString(".")
- val fileFullName = InternalSchemaUtils.reBuildFilterName(fullName,
fileSchema, querySchema)
+ val fileFullName = fileFilterName(fullName, fileSchema, querySchema)
if (fileFullName.isEmpty) {
""
} else if (fileFullName == fullName) {
@@ -270,7 +271,24 @@ object ParquetSchemaEvolutionUtils {
}.quoted
}
case None =>
- InternalSchemaUtils.reBuildFilterName(name, fileSchema, querySchema)
+ fileFilterName(name, fileSchema, querySchema)
+ }
+ }
+
+ /**
+ * InternalSchemaUtils.reBuildFilterName, except a name the query schema
does not hold comes back
+ * empty, so its filter collapses to AlwaysTrue instead of failing the read.
Such a name comes from
+ * a predicate on a variant that Spark's PushVariantIntoScan rewrote: it is
pushed as a nested
+ * filter on the projection struct's ordinal-named member (v.`0`), which is
no column of the table.
+ * Dropping it is always safe, since Spark evaluates every pushed filter
again above the scan, and
+ * it lets the read reach validateNoShreddedVariants, which rejects that
request with the
+ * actionable message.
+ */
+ private def fileFilterName(fullName: String, fileSchema: InternalSchema,
querySchema: InternalSchema): String = {
+ if (querySchema.findIdByName(fullName) == -1) {
+ ""
+ } else {
+ InternalSchemaUtils.reBuildFilterName(fullName, fileSchema, querySchema)
}
}
diff --git
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
index 686288fb8a23..87b0f5f3b59e 100644
---
a/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
+++
b/hudi-spark-datasource/hudi-spark-common/src/test/scala/org/apache/spark/sql/execution/datasources/parquet/TestParquetSchemaEvolutionUtils.scala
@@ -363,8 +363,10 @@ class TestParquetSchemaEvolutionUtils {
// cannot be evaluated must not skip any of the file's row groups.
Assertions.assertEquals(AlwaysTrue, rebuild(IsNotNull("added")))
- // Every leaf filter type takes both arms: re-spelled when the file holds
the column under
- // another name, AlwaysTrue when it does not hold it at all.
+ // Every leaf filter type takes all three arms: re-spelled when the file
holds the column under
+ // another name, AlwaysTrue when it does not hold it at all, and
AlwaysTrue when even the query
+ // schema does not hold the name, which is how a predicate on a
PushVariantIntoScan projection
+ // member arrives (before, that threw IllegalArgumentException).
val leafFilters: Seq[String => Filter] = Seq(
EqualTo(_, "x"), EqualNullSafe(_, "x"), GreaterThan(_, "x"),
GreaterThanOrEqual(_, "x"),
LessThan(_, "x"), LessThanOrEqual(_, "x"), In(_, Array[Any]("x", "y")),
IsNull(_), IsNotNull(_),
@@ -372,6 +374,7 @@ class TestParquetSchemaEvolutionUtils {
leafFilters.foreach { leaf =>
Assertions.assertEquals(leaf("original"), rebuild(leaf("renamed")))
Assertions.assertEquals(AlwaysTrue, rebuild(leaf("added")),
s"${leaf("added")} on an absent column")
+ Assertions.assertEquals(AlwaysTrue, rebuild(leaf("v.`0`")),
s"${leaf("v.`0`")} on a name the query schema lacks")
}
// The constant filters reference no column and pass through as they are.
@@ -382,6 +385,9 @@ class TestParquetSchemaEvolutionUtils {
Assertions.assertEquals(
And(EqualTo("original", "x"), AlwaysTrue),
rebuild(And(EqualTo("renamed", "x"), IsNull("added"))))
+ Assertions.assertEquals(
+ And(EqualTo("original", "x"), AlwaysTrue),
+ rebuild(And(EqualTo("renamed", "x"), GreaterThan("v.`0`", 5))))
Assertions.assertEquals(
Or(Not(EqualTo("original", "x")), GreaterThanOrEqual("id", 2)),
rebuild(Or(Not(EqualTo("renamed", "x")), GreaterThanOrEqual("id", 2))))
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 544b01db4e25..c0b6bb5c7934 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
@@ -607,6 +607,17 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
assert(scanReadSchema(sql) == s"struct<id:int,v:$readV>", s"[$leg]
ReadSchema of the scan of: $sql")
}
+ // The scan reports the base file format as its short name (#20029), so
Spark pushes nested
+ // predicates: with the rule on, a variant_get predicate arrives as a
filter on the projection
+ // struct's ordinal-named member, a name no file carries as a parquet
leaf. ParquetFilters drops
+ // it per file - it neither skips a row group nor lands on another column
- and the checkAnswer
+ // counts in each leg, identical on both arms, are what pin that. This
pins that the filter
+ // really was pushed, so those counts cannot go vacuous if the pushdown is
lost again.
+ def assertPushedFilters(sql: String, pushIntoScan: String, on: String,
off: String, leg: String): Unit = {
+ val expected = if (pushIntoScan.toBoolean) on else off
+ assert(scanPushedFilters(sql) == expected, s"[$leg] PushedFilters 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) {
@@ -640,6 +651,13 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
checkAnswer(
s"select count(*) from $tableName where try_variant_get(v, '$$.d',
'boolean')")(Seq(10))
checkAnswer(s"select count(*) from $tableName where v is
null")(Seq(0))
+ // An IsNull filter is the one that would drop rows if it resolved
against the wrong leaf.
+ checkAnswer(
+ s"select count(*) from $tableName where try_variant_get(v, '$$.a',
'bigint') is null")(Seq(20))
+ assertPushedFilters(s"select count(*) from $tableName where
try_variant_get(v, '$$.a', 'bigint') > 5",
+ pushIntoScan, on = "[IsNotNull(v), GreaterThan(v.`0`,5)]", off =
"[IsNotNull(v)]", leg)
+ assertPushedFilters(s"select count(*) from $tableName where
try_variant_get(v, '$$.a', 'bigint') is null",
+ pushIntoScan, on = "[IsNull(v.`0`)]", off = "[]", leg)
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",
@@ -684,6 +702,8 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
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)
+ assertPushedFilters(s"select count(*) from $tableName where
try_variant_get(v, '$$.a', 'bigint') > 100",
+ pushIntoScan, on = "[IsNotNull(v), GreaterThan(v.`0`,100)]", off =
"[IsNotNull(v)]", 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
@@ -942,6 +962,15 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
checkNestedExceptionContains(
() => spark.sql(s"select id, cast(v as string), note from
$tableName").collect())(
"pushVariantIntoScan")
+ // A variant_get predicate reaches the reader as a pushed filter on
the projection struct's
+ // ordinal-named member, a name the query schema does not hold. The
filter rebuild
+ // (ParquetSchemaEvolutionUtils.rebuildFilterFromParquet) runs before
the guard and drops it
+ // rather than failing on the unknown name, so this read is rejected
by the guard as well
+ // instead of an IllegalArgumentException from the rebuild.
+ val filtered = s"select id from $tableName where variant_get(v,
'$$.a', 'int') > 0"
+ assert(scanPushedFilters(filtered) == "[IsNotNull(v),
GreaterThan(v.`0`,0)]",
+ s"[$leg] PushedFilters of the scan of: $filtered")
+ checkNestedExceptionContains(() =>
spark.sql(filtered).collect())("pushVariantIntoScan")
// The guard's empty-projection carve-out: count(*) reads no column
data, and the query
// schema it would be checked against is the UNPRUNED table schema,
variant included, so
// the guard must not run at all. Drop the requiredSchema.nonEmpty
gate at any of its five
@@ -1525,6 +1554,12 @@ class TestVariantShreddingMixedLayouts extends
HoodieSparkSqlTestBase with Varia
assert(variantProjectionPushedIntoScan(
s"select id, variant_get(s.inner, '$$.k', 'string') from
$tableName") == pushed,
s"[$leg] PushVariantIntoScan $verdict rewritten s.inner into a
projection struct")
+ // The nested twin of the top-level pin: the member filter sits one
struct further down
+ // and is dropped the same way; IsNotNull(s.inner) is itself a
nested filter on a group,
+ // which ParquetFilters does not serve either.
+ val filtered = s"select id from $tableName where
variant_get(s.inner, '$$.k', 'string') = 'x7'"
+ val expectedFilters = if (pushed) "[IsNotNull(s.inner),
EqualTo(s.inner.`0`,x7)]" else "[IsNotNull(s.inner)]"
+ assert(scanPushedFilters(filtered) == expectedFilters, s"[$leg]
PushedFilters of the scan of: $filtered")
}
}
}
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 8f1181093fa7..3dbc9e9b955c 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
@@ -657,11 +657,18 @@ trait VariantShreddingTestSupport { self:
HoodieSparkSqlTestBase =>
* `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
- }
+ protected def scanReadSchema(sql: String): String =
+ singleFileScanOf(sql).requiredSchema.catalogString
+
+ /**
+ * The PushedFilters of the one file scan in the physical plan of `sql`, as
its explain line
+ * prints them. A predicate on a variant that PushVariantIntoScan rewrote
shows there as a nested
+ * filter on the projection struct's ordinal-named member, e.g.
GreaterThan(v.`0`,5). These are
+ * the filters Spark hands the reader, not the ones a file can serve -
ParquetFilters drops a
+ * name that is no parquet leaf.
+ */
+ protected def scanPushedFilters(sql: String): String =
+ singleFileScanOf(sql).metadata.getOrElse("PushedFilters", "")
/**
* The file scans in the physical plan of `sql`. `sparkPlan` rather than
`executedPlan`: under
@@ -673,6 +680,13 @@ trait VariantShreddingTestSupport { self:
HoodieSparkSqlTestBase =>
scans
}
+ /** The one file scan in the physical plan of `sql`. */
+ private def singleFileScanOf(sql: String): FileSourceScanExec = {
+ val scans = fileScansOf(sql)
+ assert(scans.size == 1, s"expected a single file scan in the plan of:
$sql")
+ scans.head
+ }
+
//
---------------------------------------------------------------------------------------------
// Write-layout toggle
//
---------------------------------------------------------------------------------------------
