This is an automated email from the ASF dual-hosted git repository.
viirya pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new d3fe206639a0 [SPARK-60602][SQL] Do not evaluate throwable filter
conditions ahead of the conditions before them when pushing filters through a
projection
d3fe206639a0 is described below
commit d3fe206639a059007872617590aff9eeeb6ba8ce
Author: Liang-Chi Hsieh <[email protected]>
AuthorDate: Wed Oct 28 23:35:10 2026 -0700
[SPARK-60602][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?
`PushPredicateThroughNonJoin` reorders the conjuncts of a `Filter` over a
`Project` that has expensive (UDF, expensive regex) aliases. It does so in two
places, and neither checks `Expression.throwable`:
1. The projection split added by SPARK-55014
(`spark.sql.optimizer.splitProjectionForExpensiveFilters`). It orders condition
groups by how many not-yet-computed aliases they need, so a `raise_error` /
`assert_true` condition can be split below the conditions written before it.
Splitting a group off also computes its aliases in the lower projection, so the
same happens for a condition over a throwable alias.
2. The cheap/expensive partition added by SPARK-47672
(`spark.sql.optimizer.avoidDoubleFilterEval`). It pushes a cheap conjunct below
the projection, ahead of an expensive conjunct written before it.
This PR checks the condition with the projection's aliases substituted
(`replaced.throwable`), so a throwable part inside a projected alias counts
too. It then:
- keeps throwable conditions out of the split, using `deterministic &&
!throwable`, the same check `PushPredicateThroughJoin` uses;
- pushes every cheap condition written before the first expensive one, and
after it, only the cheap conditions that are not throwable;
- keeps the conditions left in the top filter in the order they were
written;
- documents the restriction in the rule's algorithm overview.
Moving a non-throwable condition below a throwable one is still allowed,
since it can only reduce the rows the throwable condition sees.
`CombineFilters` and the join pushdown rules already behave this way.
This PR only covers the `Filter` over `Project` case. The generic
`UnaryNode` case of the same rule has the same problem, for example with Python
UDFs once `ExtractPythonUDFs` has run. That is tracked separately in
SPARK-60633.
### 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` and `g`
expensive UDFs:
```scala
spark.range(0, 10)
.select($"id", f($"id") as "f", g($"id") as "g")
.where($"f" + $"g" > 10 && assert_true($"g" > 5).isNull)
```
The `assert_true` group needs fewer aliases, so the split evaluates it
first, on every row, and the query fails with `USER_RAISED_EXCEPTION`. It
succeeds with `splitProjectionForExpensiveFilters=false`.
```scala
spark.range(0, 10)
.select($"id", f($"id") as "f")
.where($"f" > 5 && assert_true($"id" > 5).isNull)
```
Here the cheap `assert_true` condition is pushed below the projection,
ahead of `f > 5`. This fails whenever `avoidDoubleFilterEval` is true, whatever
the split config, and it is already in 4.2.0. The same holds when `assert_true`
is moved into a projected alias and the filter only reads it.
### Does this PR introduce _any_ user-facing change?
Yes. Queries like the ones above failed with `USER_RAISED_EXCEPTION` in
4.2.0 and later (the second shape), and on unreleased master / branch-4.x (the
first shape). They now succeed.
One more change applies only with the unreleased projection split enabled.
A non-deterministic condition left above the projection now keeps its written
position among the conditions the split could not place, instead of moving
after them. This is the order the split-disabled path already uses. Other
filters without throwable conditions are optimized as before.
### How was this patch tested?
New tests in `FilterPushdownSuite`:
- the split, a throwable condition not blocking later conditions, and the
cheap pushdown path;
- a condition over a throwable alias, in both the cheap pushdown and the
split;
- the order of the conditions left above the projection, for a throwable
and for a non-deterministic condition. These check the conjunct sequence
directly, since `comparePlans` sorts conjuncts.
An end-to-end test in `DataFrameSuite` runs every combination of
whole-stage codegen, `splitProjectionForExpensiveFilters` and
`avoidDoubleFilterEval`. It covers both queries above, each with and without an
extra `id >= 0` conjunct, which makes the rule rebuild the plan. It also covers
the throwable-alias variants.
Each new test fails without the change, or with the top filter emitted in
the old order. All tests in `org.apache.spark.sql.catalyst.optimizer` pass.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
Closes #59848 from viirya/SPARK-60602.
Authored-by: Liang-Chi Hsieh <[email protected]>
Signed-off-by: Liang-Chi Hsieh <[email protected]>
---
.../spark/sql/catalyst/optimizer/Optimizer.scala | 58 ++++----
.../catalyst/optimizer/FilterPushdownSuite.scala | 151 +++++++++++++++++++++
.../org/apache/spark/sql/DataFrameSuite.scala | 44 ++++++
3 files changed, 230 insertions(+), 23 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 3bb998dca253..9e2c4c6b972f 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
@@ -2594,7 +2594,11 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// is worth splitting.
// Note that a given filter may contain parts (separated 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, it never enters the
projection split,
+ // 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
@@ -2618,22 +2622,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
// Cheap filters (cases 1 & 2) pushed below the projection.
val baseChild: LogicalPlan = if (cheapWithUsed.nonEmpty) {
val combinedCheapFilter = cheapWithUsed.map(_._3).reduce(And)
@@ -2644,15 +2649,21 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// Case 3: split the projection around the expensive conditions.
Non-deterministic
// conditions stay in the top filter, as they do everywhere else in
this rule -- the
// split orders conditions by cost, and cost is no way to decide how
many rows a
- // non-deterministic expression sees.
+ // non-deterministic expression sees. Throwable conditions, or
conditions over throwable
+ // aliases, stay there too: ordering by cost could evaluate them (or
compute the aliases
+ // in a lower projection) ahead of the conditions written before them.
val (splitChild, splitAliases, stayUp) =
if (expensiveWithUsed.nonEmpty &&
SQLConf.get.splitProjectionForExpensiveFilters) {
- val (deterministic, nonDeterministic) =
- expensiveWithUsed.partition { case (cond, _, _) =>
cond.deterministic }
+ val (splittable, notSplittable) = expensiveWithUsed.partition {
+ case (cond, _, replaced) => cond.deterministic &&
!replaced.throwable
+ }
val (child, aliases, unplaced) =
splitProjectForExpensiveConditions(
project, aliasMap, baseChild,
- deterministic.map { case (cond, used, _) => (cond, used) })
- (child, aliases, unplaced ++ nonDeterministic.map(_._1))
+ splittable.map { case (cond, used, _) => (cond, used) })
+ // Keep the top filter in the order the conditions were written,
so a throwable
+ // condition still only sees the rows the conditions before it
kept.
+ val stayUpSet = (unplaced ++ notSplittable.map(_._1)).toSet
+ (child, aliases,
expensiveWithUsed.map(_._1).filter(stayUpSet.contains))
} else {
(baseChild, AttributeSet.empty, expensiveWithUsed.map(_._1))
}
@@ -2872,8 +2883,9 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
* The most demanding group is left un-placed for the caller to put above
the projection --
* where expensive conditions go when there is nothing worth splitting.
*
- * Callers must pass deterministic conditions only -- the grouping below
orders conditions by
- * cost, and cost is no way to decide how many rows a non-deterministic
expression sees.
+ * Callers must pass deterministic, non-throwable conditions only -- the
grouping below orders
+ * conditions by cost, and cost is no way to decide how many rows a
non-deterministic
+ * expression sees, nor may it evaluate a throwable one ahead of the
conditions before it.
*
* Returns the new plan, the alias attributes it has already computed, and
the un-placed
* conditions (in their original, alias-referencing form).
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 27ad4da6a917..5e1d2b9f2fd6 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
@@ -599,6 +599,157 @@ class FilterPushdownSuite extends PlanTest {
comparePlans(optimized, correctAnswer)
}
+ test("SPARK-60602: a throwable condition is not split off ahead of the
conditions before it") {
+ val raiseError = IsNull(RaiseError(Literal("boom")))
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g",
$"b")
+ .where($"f" =!= $"g" && ($"g" || raiseError))
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // On cost alone (g || raiseError) would go lowest, raising on rows (f !=
g) filters out.
+ // It stays in the top filter instead, which leaves nothing worth
splitting.
+ comparePlans(optimized, originalQuery)
+ }
+
+ test("SPARK-60602: a throwable condition does not keep later conditions from
being split off") {
+ val raiseError = IsNull(RaiseError(Literal("boom")))
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g",
$"b")
+ .where(($"g" || raiseError) && $"f")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // Filtering on f first only leaves the throwable condition fewer rows to
raise on.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f")
+ .where($"f")
+ .select($"a", $"f", $"e".rlike("other") as "g", $"b")
+ .where($"g" || raiseError)
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-60602: a cheap throwable condition is not pushed ahead of an
expensive one") {
+ val raiseError = IsNull(RaiseError(Literal("boom")))
+ Seq(true, false).foreach { split =>
+ withSQLConf(SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key ->
split.toString) {
+ 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("SPARK-60602: a condition over a throwable alias is not split off ahead
of the " +
+ "conditions before it") {
+ val raiseError = IsNull(RaiseError(Literal("boom")))
+ val throwableAlias = If($"e".rlike("other"), Literal(true), raiseError)
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("third") as "h",
+ throwableAlias as "g", $"b")
+ .where($"f" =!= $"h" && $"g")
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // On cost alone the condition on g would go lowest, computing g on every
row. Splitting
+ // f and h off instead only leaves g fewer rows to raise on.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f",
$"e".rlike("third") as "h")
+ .where($"f" =!= $"h")
+ .select($"a", $"f", $"h", throwableAlias as "g", $"b")
+ .where($"g")
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ // `comparePlans` sorts the conjuncts of each filter, so check the top
filter's order directly.
+ private def assertTopFilterConjuncts(plan: LogicalPlan, expected:
Seq[Expression]): Unit = {
+ val actual = plan.collectFirst { case Filter(condition, _) => condition }
+ .map(splitConjunctivePredicates).getOrElse(Nil)
+ assert(actual.map(_.canonicalized) === expected.map(_.canonicalized),
+ s"Top filter conjuncts ${actual.mkString(", ")} are not
${expected.mkString(", ")}")
+ }
+
+ test("SPARK-60602: the conditions left above the projection keep their
written order") {
+ val raiseError = IsNull(RaiseError(Literal("boom")))
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g",
$"b")
+ .where($"f" =!= $"g" && ($"g" || raiseError) && $"a" > 1)
+ .analyze
+ val Seq(unplaced, throwable, _) = splitConjunctivePredicates(
+ originalQuery.asInstanceOf[Filter].condition)
+
+ // a > 1 is pushed, (f != g) needs every expensive element and is left
un-placed, and the
+ // throwable condition is not split. It must still come after (f != g).
+ assertTopFilterConjuncts(Optimize.execute(originalQuery), Seq(unplaced,
throwable))
+ }
+
+ test("SPARK-60602: a non-deterministic condition left above the projection
keeps its " +
+ "written order") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("other") as "g",
$"b")
+ .where((Rand(10) > 0.5 || $"g") && $"f" =!= $"g" && $"a" > 1)
+ .analyze
+ val Seq(nonDeterministic, unplaced, _) = splitConjunctivePredicates(
+ originalQuery.asInstanceOf[Filter].condition)
+
+ assertTopFilterConjuncts(Optimize.execute(originalQuery),
Seq(nonDeterministic, unplaced))
+ }
+
test("SPARK-55014: do not split the projection when configured") {
withSQLConf(SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key -> "false")
{
val originalQuery = testStringRelation
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 b579924c67ad..d7d38f637c74 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
@@ -2905,6 +2905,50 @@ class DataFrameSuite extends SharedSparkSession
}
}
+ 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)
+ val h = 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 splitQuery(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 two shapes 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)
+ def splitAliasQuery(): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f", h($"id") as "h",
+ when(g($"id") > 5, g($"id")).otherwise(raise_error(lit("boom"))) as
"g")
+ .where($"f" + $"h" > 10 && $"g" > 5)
+ val expected = (6L until 10L).map(i => Row(i, i, i))
+ for (codegen <- Seq(true, false); split <- Seq(true, false); avoid <-
Seq(true, false)) {
+ withSQLConf(
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString,
+ SQLConf.AVOID_DOUBLE_FILTER_EVAL.key -> avoid.toString,
+ SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key -> split.toString) {
+ Seq(lit(true), $"id" >= 0).foreach { extra =>
+ checkAnswer(splitQuery(extra), expected)
+ 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)))
+ checkAnswer(splitAliasQuery(), (6L until 10L).map(i => Row(i, i, i,
i)))
+ }
+ }
+ }
+ }
+
test("SPARK-60528: collect a nested empty struct") {
val df = spark.range(1).selectExpr("id AS i", "named_struct('a', struct())
AS s")
checkAnswer(df, Row(0L, Row(Row())))
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
