dongjoon-hyun commented on code in PR #59848:
URL: https://github.com/apache/spark/pull/59848#discussion_r4403206075
##########
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:
**[1] The cheap path checks `cond.throwable`, but what it pushes is
`replaced`, so a throwable inside a cheap projected alias still runs ahead of
an expensive conjunct written before it.**
`cond` reads the alias, so `cond.throwable` is false when the throwable part
sits in the alias, while the pushed `replaced` is throwable. A variant of
`cheapQuery` with `assert_true` moved into the projection shows it:
```scala
spark.range(0, 10, 1, 1)
.select($"id", f($"id") as "f", assert_true($"id" > 5) as "t")
.where($"f" > 5 && $"t".isNull)
```
`t IS NULL` reads only the cheap alias `t`, so it is pushed below the
projection as `isnull(if(id > 5, null, raise_error(...)))` and raises
`USER_RAISED_EXCEPTION` on ids 0 to 5. With `avoidDoubleFilterEval=false`, the
whole condition is pushed in written order and the query returns ids 6 to 9,
the same as `cheapQuery`. Since this path pushes `replaced`, checking it covers
both shapes:
```suggestion
.partition { case ((_, used, replaced), idx) =>
isCheap(used) && (!replaced.throwable || firstExpensive < 0 ||
idx < firstExpensive)
```
Could you add this variant to the end-to-end matrix as well? I traced it
from source and did not run it.
##########
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 {
Review Comment:
**[3] Nit: `span` can express the pushable prefix directly.**
Every conjunct before the first expensive one is pushable, whatever it
contains, and after that a conjunct is pushable when it is cheap and not
throwable. With `span`, `firstExpensive`, `zipWithIndex`, the `firstExpensive <
0` case and the two `.map(_._1)` can go:
```scala
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
```
This keeps the conjuncts in written order, as the current partition does,
and already includes item 1.
##########
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
Review Comment:
**[2] The split also checks only the conjunct, so a non-throwable conjunct
over a throwable expensive alias is still split off first, and the lower
projection computes that alias on every row.**
Splitting a group off also moves the computation of its aliases into a lower
`Project`, so the aliases matter here as much as the conjunct. Take, with `f`,
`g` and `h` identity UDFs as in the new test:
```scala
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)
```
The `{g}` group needs one alias and `{f, h}` needs two, so the split puts
`g` in the bottom layer, and its `raise_error` fires on ids 0 to 5. With
whole-stage codegen and the split off, `g` is only computed for the rows `f + h
> 10` keeps, and with `avoidDoubleFilterEval=false` the query returns ids 6 to
9 in every mode. This is the `splitQuery` shape with the throwable part moved
into the alias. Checking the rewritten conjunct covers the aliases it reads:
```suggestion
case (cond, _, replaced) => cond.deterministic &&
!replaced.throwable
```
An alias that `unitOf` adds because it shares expensive work with these
could still be throwable, but that looks rare enough to leave. As with item 1,
I traced this from source and did not run it.
--
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]
