This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch branch-4.2
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-4.2 by this push:
new eef259d0e42f [SPARK-60602][4.2][SQL] Do not evaluate throwable filter
conditions ahead of the conditions before them when pushing filters through a
projection
eef259d0e42f is described below
commit eef259d0e42f20cc7baeafa939fd6fdc746d7d62
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Thu Oct 29 08:05:46 2026 -0700
[SPARK-60602][4.2][SQL] Do not evaluate throwable filter conditions ahead
of the conditions before them when pushing filters through a projection
### What changes were proposed in this pull request?
Backport of #59848 to branch-4.2.
`PushPredicateThroughNonJoin` pushes the cheap conjuncts of a `Filter` over
a `Project` with expensive (UDF, expensive regex) aliases below the projection,
and keeps the expensive ones above (SPARK-47672,
`spark.sql.optimizer.avoidDoubleFilterEval`). It did not check
`Expression.throwable`, so a cheap `raise_error` / `assert_true` conjunct could
be pushed below the projection ahead of an expensive conjunct written before it.
This PR pushes every cheap conjunct written before the first expensive one.
After it, only the cheap conjuncts that are not throwable are pushed, checking
the conjunct with the projection's aliases substituted so a throwable part
inside a projected alias counts too. It also documents the restriction in the
rule's algorithm overview.
branch-4.2 does not have the projection split from SPARK-55014, so the
split-related parts of #59848 do not apply here and are left out.
### Why are the changes needed?
A throwable condition can be evaluated on rows that the `AND` short-circuit
would have kept from it, so a query that should succeed fails. With `f` an
expensive UDF:
```scala
spark.range(0, 10)
.select($"id", f($"id") as "f")
.where($"f" > 5 && assert_true($"id" > 5).isNull)
```
The cheap `assert_true` condition is pushed below the projection, ahead of
`f > 5`, and the query fails with `USER_RAISED_EXCEPTION` whenever
`avoidDoubleFilterEval` is true.
### Does this PR introduce _any_ user-facing change?
Yes. Queries like the one above failed with `USER_RAISED_EXCEPTION` in
4.2.0 and later. They now succeed. Filters without throwable conditions are
optimized as before.
### How was this patch tested?
The `FilterPushdownSuite` and `DataFrameSuite` tests from #59848 that apply
without the projection split: the cheap pushdown path, with the throwable part
in the condition and in a projected alias, and an end-to-end test over
whole-stage codegen and `avoidDoubleFilterEval`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #59907 from viirya/SPARK-60602-4.2.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../spark/sql/catalyst/optimizer/Optimizer.scala | 37 ++++++++-------
.../catalyst/optimizer/FilterPushdownSuite.scala | 55 ++++++++++++++++++++++
.../org/apache/spark/sql/DataFrameSuite.scala | 36 ++++++++++++++
3 files changed, 112 insertions(+), 16 deletions(-)
diff --git
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
index 47d61eb0ee54..f97c11292d40 100644
---
a/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
+++
b/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala
@@ -2113,7 +2113,11 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 3) When an a filter references expensive to compute references we do
not push it.
// Note that a given filter may contain parts (sepereated by logical ands)
from all cases.
// We handle each part separately according to the logic above.
- // Additional restriction:
+ // Additional restrictions:
+ // SPARK-60602: A throwable part (including one over a throwable alias)
must not be
+ // evaluated on rows the parts written before it would have filtered out.
So it is pushed
+ // only when no expensive part is written before it, and the parts left
above the projection
+ // keep their written order.
// SPARK-13473: We can't push the predicate down when the underlying
projection output non-
// deterministic field(s). Non-deterministic expressions are essentially
stateful. This
// implies that, for a given input row, the output are determined by the
expression's initial
@@ -2137,22 +2141,23 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
val (replaced, usedAliases) = replaceAliasWhileTracking(cond,
aliasMap)
(cond, usedAliases, replaced)
}
+ // A component is cheap if it uses nothing expensive from the
projection: either it
+ // uses nothing at all, or only inexpensive items. Then we can push it
because it might
+ // eliminate more data quickly and it may also be something which
could be evaluated at
+ // the storage layer. We may wish to improve this heuristic in the
future.
+ def isCheap(used: AttributeMap[Alias]): Boolean =
!used.exists(_._2.child.expensive)
// Split the filter's components into cheap and expensive while
keeping track of
- // what each references from the projection.
- val (cheapWithUsed, expensiveWithUsed) =
originalAndRewrittenConditionsWithUsedAlias
- .partition { case (cond, used, replaced) =>
- // Didn't use anything? We're good
- if (used.isEmpty) {
- true
- } else if (!used.exists(_._2.child.expensive)) {
- // If it's cheap we can push it because it might eliminate more
data quickly and
- // it may also be something which could be evaluated at the
storage layer.
- // We may wish to improve this heuristic in the future.
- true
- } else {
- false
- }
- }
+ // what each references from the projection. The cheap components
written before the
+ // first expensive one are all pushed. After it, a throwable component
(e.g.
+ // `raise_error`, possibly inside a projected alias) stays above with
the expensive ones,
+ // so it is not evaluated on rows the components before it would have
filtered out.
+ val (leading, rest) = originalAndRewrittenConditionsWithUsedAlias.span
{
+ case (_, used, _) => isCheap(used)
+ }
+ val (cheapRest, expensiveWithUsed) = rest.partition {
+ case (_, used, replaced) => isCheap(used) && !replaced.throwable
+ }
+ val cheapWithUsed = leading ++ cheapRest
// Short circuit if we do not have any cheap filters return the
original filter as is.
if (cheapWithUsed.isEmpty) {
f
diff --git
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
index f457eced4d98..1c9b5b6f6e57 100644
---
a/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
+++
b/sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala
@@ -271,6 +271,61 @@ class FilterPushdownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}
+ test("SPARK-60602: a cheap throwable condition is not pushed ahead of an
expensive one") {
+ val raiseError = IsNull(RaiseError(Literal("boom")))
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"b")
+ .where($"f" && ($"a" > 5 || raiseError) && $"b" > 1)
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // Only the non-throwable cheap condition moves below the expensive one.
+ val correctAnswer = testStringRelation
+ .where($"b" > 1)
+ .select($"a", $"e".rlike("magic") as "f", $"b")
+ .where($"f" && ($"a" > 5 || raiseError))
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+
+ // Written ahead of every expensive condition, it is pushed as usual.
+ val throwableFirst = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"b")
+ .where(($"a" > 5 || raiseError) && $"f")
+ .analyze
+
+ val throwableFirstAnswer = testStringRelation
+ .where($"a" > 5 || raiseError)
+ .select($"a", $"e".rlike("magic") as "f", $"b")
+ .where($"f")
+ .analyze
+
+ comparePlans(Optimize.execute(throwableFirst), throwableFirstAnswer)
+ }
+
+ test("SPARK-60602: a cheap condition over a throwable alias is not pushed
ahead of an " +
+ "expensive one") {
+ val raiseError = IsNull(RaiseError(Literal("boom")))
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", If($"a" > 5, Literal(true),
raiseError) as "t",
+ $"b")
+ .where($"f" && $"t" && $"b" > 1)
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // `t` is cheap, but reading it evaluates the throwable expression behind
it.
+ val correctAnswer = testStringRelation
+ .where($"b" > 1)
+ .select($"a", $"e".rlike("magic") as "f", If($"a" > 5, Literal(true),
raiseError) as "t",
+ $"b")
+ .where($"f" && $"t")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
test("nondeterministic: can always push down filter through project with
deterministic field") {
val originalQuery = testRelation
.select($"a")
diff --git a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
index 577ae025f1df..273c24d00c03 100644
--- a/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
+++ b/sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala
@@ -2786,6 +2786,42 @@ class DataFrameSuite extends SharedSparkSession
checkAnswer(df1, Seq(Row(1), Row(4)))
}
+ test("SPARK-60602: a throwable filter condition is not evaluated ahead of
its order") {
+ val f = udf((i: Long) => i)
+ val g = udf((i: Long) => i)
+ // `assert_true(g > 5)` only holds for the ids the first conjunct (`f + g
> 10`) keeps, so the
+ // `AND` short-circuit must keep it away from the other rows however the
filter is pushed.
+ // With `id >= 0` the rule also rebuilds the plan, pushing that conjunct
below the projection,
+ // and only the order of the conjuncts left above keeps the throwable one
behind the first.
+ def expensiveQuery(extra: Column): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f", g($"id") as "g")
+ .where($"f" + $"g" > 10 && assert_true($"g" > 5).isNull && extra)
+ // The same with the throwable conjunct referencing no projected UDF at
all.
+ def cheapQuery(extra: Column): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f")
+ .where($"f" > 5 && assert_true($"id" > 5).isNull && extra)
+ // The same with the throwable part inside a projected alias.
+ def cheapAliasQuery(): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f", assert_true($"id" > 5) as "t")
+ .where($"f" > 5 && $"t".isNull)
+ for (codegen <- Seq(true, false); avoid <- Seq(true, false)) {
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString,
+ SQLConf.AVOID_DOUBLE_FILTER_EVAL.key -> avoid.toString) {
+ Seq(lit(true), $"id" >= 0).foreach { extra =>
+ checkAnswer(expensiveQuery(extra), (6L until 10L).map(i => Row(i, i,
i)))
+ checkAnswer(cheapQuery(extra), (6L until 10L).map(i => Row(i, i)))
+ }
+ // An interpreted projection computes every projected column for every
row it gets, so
+ // a throwable alias only waits for the filter when codegen defers it
past the filter
+ // or the whole filter is pushed below the projection.
+ if (codegen || !avoid) {
+ checkAnswer(cheapAliasQuery(), (6L until 10L).map(i => Row(i, i,
null)))
+ }
+ }
+ }
+ }
+
test("SPARK-57725: resolve columns when the input plan has a null-named
attribute") {
// A null-named AttributeReference can reach the analyzer (e.g. via a
StructField built with a
// null name). Selecting another column must still resolve through the
full analyzer path
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
