spark-test commented on code in PR #59848:
URL: https://github.com/apache/spark/pull/59848#discussion_r4395819842
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2644,15 +2647,20 @@ 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 stay there
too: ordering by
+ // cost could evaluate them 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, _, _) => cond.deterministic && !cond.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))
Review Comment:
**Nit (P3):** Emitting the top filter in written order also changes filters
that have no throwable conjunct. With the split on, a non-deterministic
expensive conjunct used to go after the conjuncts the split could not place
(`unplaced ++ nonDeterministic`). Now it keeps its written position whenever
the rule rewrites the plan. Take `(rand(10) > 0.5 || f) && (f || g) && a > 1`
over `rlike` aliases `f` and `g`: `a > 1` is pushed, `f || g` is left unplaced,
and the top filter goes from `(f || g) AND (rand(10) > 0.5 || f)` to `(rand(10)
> 0.5 || f) AND (f || g)`. `rand(10)` now runs on every row that passes `a >
1`, so a seeded query keeps a different set of rows than on master.
The new position looks like the better choice. It is what the split-disabled
branch does, and no release ever had the old order, which arrived with
SPARK-55014. But it contradicts the description's "Filters without throwable
conditions are optimized as before", and no test pins either order:
`comparePlans` sorts each Filter's conjuncts, and the next fixed-point pass
leaves this filter as it is. I'd keep written order, mention the change in the
description, and add a plan test that asserts the top filter's conjunct
sequence directly. If you'd rather keep master's order, `unplaced ++
notSplittable.map(_._1)` is just as safe for throwable conjuncts.
**Verification:**
- **Regression:** Take a filter whose non-deterministic expensive conjunct
is written before a conjunct that needs every expensive alias, together with a
cheap conjunct that is pushed below the projection. The optimized top filter
lists its conjuncts in written order, and the check fails if the
non-deterministic conjunct is moved after the other one.
- **Inspection:** The PR description's statement about filters without
throwable conditions says that non-deterministic conjuncts left above the
projection now keep their written position instead of moving after the
conjuncts the split could not place.
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala:
##########
@@ -2905,6 +2905,29 @@ 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)
+ // `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.
+ def splitQuery(): 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)
Review Comment:
**Nit (P3):** Neither query here reaches the plans this PR rebuilds. Neither
has a conjunct the rule can push or a group it can split. With
`avoidDoubleFilterEval` on, the rule returns the original Filter, and with it
off, it pushes the whole condition. So the end-to-end matrix never runs a plan
where other conjuncts move while `assert_true` is held in the top filter.
The gap matters most for this first example. `assert_true($"g" > 5)` uses
the expensive alias `g`, so no later pass pushes it, and in a rebuilt plan only
the written order of the top filter keeps it behind `f + g > 10`. Adding one
pushable conjunct, e.g. `&& $"id" >= 0`, produces such a plan: `id >= 0` is
pushed and the top filter is `f + g > 10 AND assert_true(g > 5) IS NULL`. If
the stay-up list were emitted with the non-splittable conjuncts first, this
variant would raise with the split on, while both current queries would keep
passing. `FilterPushdownSuite`'s cheap-throwable test would still catch that
particular regression today, but only indirectly. `comparePlans` ignores
conjunct order, and the test fails only because its fixed-point batch re-pushes
the misplaced cheap throwable. Running this variant, and `cheapQuery` with the
same extra conjunct, in the existing matrix would test the PR's written-order
change directly.
**Verification:**
- **Regression:** When the conjuncts left above the projection are emitted
with the non-splittable ones ahead of the unplaced ones, the added
first-example query fails with USER_RAISED_EXCEPTION in the configurations with
both avoidDoubleFilterEval and splitProjectionForExpensiveFilters enabled, and
it passes with the written-order emission.
- **Behavior:** The added queries return exactly the rows that written-order
AND evaluation allows (ids 6 to 9 for both example shapes with the extra `id >=
0` conjunct) under every combination of whole-stage codegen,
avoidDoubleFilterEval and splitProjectionForExpensiveFilters.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2618,22 +2618,25 @@ 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)
+ val firstExpensive =
originalAndRewrittenConditionsWithUsedAlias.indexWhere {
+ case (_, used, _) => !isCheap(used)
+ }
// 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. A throwable component
(e.g. `raise_error`)
+ // must not be evaluated on rows the components before it would have
filtered out, so
+ // it is only pushed when no expensive component, which stays above,
comes before it.
+ val (cheapWithIndex, expensiveWithIndex) =
originalAndRewrittenConditionsWithUsedAlias
+ .zipWithIndex
+ .partition { case ((cond, used, _), idx) =>
+ isCheap(used) && (!cond.throwable || firstExpensive < 0 || idx <
firstExpensive)
Review Comment:
**Nit (P3):** The algorithm overview at the top of this case is now out of
date. It still says that parts referencing nothing, or only inexpensive items,
are "pushed", that "We handle each part separately according to the logic
above.", and that SPARK-13473 is the only "Additional restriction:". With this
change, a cheap throwable part written after an expensive one stays above the
projection, so where it goes depends on the parts before it. Throwable parts
also never enter the split. Only the inline comments here and at the split say
so, which makes the restriction easy to drop in a later edit without noticing
that SPARK-60602 depends on it. Could you add it to the overview, e.g. as a
second additional restriction? Something like: a throwable part is pushed only
when no expensive part is written before it, it never enters the projection
split, and the parts left above keep their written order.
**Verification:**
- **Inspection:** The overview's account of where cheap and expensive parts
go matches the partition and split predicates, including the throwable
restriction, and the updated comment lines stay ASCII and within the
100-character line limit.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
