dongjoon-hyun commented on code in PR #59940:
URL: https://github.com/apache/spark/pull/59940#discussion_r4470080337
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,71 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 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(a =>
isExpensive(a._2.child))
- // Split the filter's components into cheap and expensive while
keeping track of
- // 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 (cheapWithUsed, expensiveWithUsed) = splitPushable(
- originalAndRewrittenConditionsWithUsedAlias)(
- canPush = { case (_, used, _) => isCheap(used) },
- throwable = { case (_, _, replaced) => replaced.throwable })
+ // Whether a component reads an expensive alias that this projection
evaluates. A scalar
+ // Python UDF is not evaluated here: `ExtractPythonUDFs` later moves
it, with its
+ // arguments, to an eval node below the projection, and pushes the
conditions below that
+ // again, except the throwable ones written after one over its output
(SPARK-60633), so
+ // holding a deterministic condition back behind one here would gain
nothing and only cost
+ // the pushdown of the batches in between. What is left of the alias
without its Python
+ // UDFs, e.g. a Scala UDF over one, is evaluated here though. Each
Python UDF is replaced
+ // with an attribute, as `ExtractPythonUDFs` does, so that
`isExpensive` sees an input
+ // that is not foldable, as in the projection it leaves.
+ def evaluatedHere(used: AttributeMap[Alias]): Boolean = used.exists {
case (_, alias) =>
+ isExpensive(alias.child) && isExpensive(alias.child.transformDown {
+ case udf if isScalarPythonUDF(udf) =>
AttributeReference("pythonUDF", udf.dataType)()
+ })
+ }
+ // Split the filter's components, in their written order, into the
cheap ones pushed
+ // below the projection and the ones kept above it. The cheap
components written before
+ // the first one kept above are all pushed, as before SPARK-60730.
After it, a cheap
+ // component is only pushed when it can be evaluated ahead of the
components before it,
+ // and how strictly that is checked depends on why a component was
kept above:
+ // - after a component over an expensive alias that the projection
does not evaluate,
+ // i.e. one only expensive through scalar Python UDFs, a cheap
component is pushed
+ // when it is deterministic and not throwable;
+ // - after a component over an expensive alias the projection
evaluates, the first one
+ // held back, a cheap component is pushed only when `canPushAhead`
accepts it, so the
+ // others are only evaluated where none of the components before
them is false.
+ // The IS NOT NULL checks that `InferFiltersFromConstraints` puts in
front of the
+ // condition are pushed as if written there, so one over a cheap alias
that can fail,
+ // e.g. a division, still computes it ahead of the components written
before the one it
+ // was inferred from.
+ val components = originalAndRewrittenConditionsWithUsedAlias
+ val firstHeldIndex = components.indexWhere {
+ case (_, used, _) => !isCheap(used) && evaluatedHere(used)
+ }
+ val legacy = SQLConf.get.legacyPushFilterAheadOfExpensiveConjunct
+ val pushed = Seq.newBuilder[(Expression, AttributeMap[Alias],
Expression)]
+ val kept = Seq.newBuilder[(Expression, AttributeMap[Alias],
Expression)]
+ val keptHeld = Seq.newBuilder[(Expression, AttributeMap[Alias],
Expression)]
+ // 0: nothing kept above yet; 1: after a component only expensive
through Python UDFs;
+ // 2: after the first component held back.
+ var strictness = 0
+ // Whether a cheap component was kept above at strictness 1, i.e. one
that is
+ // non-deterministic or throwable, before the first one held back, if
any.
+ var keptCheapBeforeHeld = false
+ components.zipWithIndex.foreach { case (component @ (cond, used,
replaced), index) =>
+ val push = isCheap(used) && (strictness match {
+ case 0 => true
+ case 1 => !replaced.throwable && (legacy || cond.deterministic)
+ case _ => canPushAhead(replaced)
+ })
+ if (push) {
+ pushed += component
+ } else {
+ kept += component
+ if (firstHeldIndex >= 0 && index > firstHeldIndex) keptHeld +=
component
Review Comment:
**1. [Medium] The first condition held back skips the split check even when
a condition over a Python UDF is kept before it**
`keepsWrittenOrder` only checks the components after `firstHeldIndex`
(`keptHeld`), on the grounds that every component before the first one held
back is pushed. That does not hold when a condition over an alias that is only
expensive through Python UDFs is kept above before it: the first held condition
is then never checked, and the split can still place it below that condition.
For example, with `p = py1(a)`, `s = scalaUdf(py2(b))` and ANSI on:
```scala
df.where($"p" > 1 && lit(10) / ($"s" - 3) > 0 && $"s" < 100)
```
`p > 1` is kept above and sets `strictness` to 1, the division is the first
held condition, and `keptHeld` is only `s < 100`, which `canPushAhead` accepts,
so the split goes ahead. The `s` group has two conditions, so `minBy` places it
first, and the division runs below `p > 1`, failing with `DIVIDE_BY_ZERO` on a
row that `p > 1` drops. In 4.1, `p > 1` was pushed below the projection and the
division stayed above it, so the query succeeded.
Checking every component kept after the first one kept above, rather than
after the first one held back, would close this, and item 2 too.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2664,8 +2728,25 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 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.
+ // For the same reason, there is no split when a condition after the
first one held back
+ // above is rejected by `canPushAhead`, with only the cheap aliases
substituted (it can
+ // tell nothing about an expensive one such as a UDF): e.g. a division
over an expensive
+ // alias, or a cheap condition held back above, which needs no
expensive alias and so
+ // would go lowest. The filter then stays above the projection as
written, as without
+ // the split. So it does when a cheap component is kept above after
one over a Python
+ // UDF before the first one held back, or with none held back, as the
split could
+ // evaluate the later expensive ones ahead of it.
+ lazy val keepsWrittenOrder = !legacy &&
+ (keptCheapBeforeHeld || expensiveHeld.nonEmpty && {
+ val cheapAliasMap =
+ AttributeMap(aliasMap.filter { case (_, a) =>
!isExpensive(a.child) })
+ expensiveHeld.exists {
+ case (cond, _, _) => !canPushAhead(replaceAlias(cond,
cheapAliasMap))
+ }
+ })
val (splitChild, splitAliases, stayUp) =
- if (expensiveWithUsed.nonEmpty &&
SQLConf.get.splitProjectionForExpensiveFilters) {
+ if (SQLConf.get.splitProjectionForExpensiveFilters &&
expensiveWithUsed.nonEmpty &&
Review Comment:
**3. [Medium] Keeping the whole filter unsplit makes an interpreted
projection compute every alias on every row**
When `keepsWrittenOrder` is true, the split is turned off for the whole
filter, not only from the rejected condition on. Without whole-stage codegen,
i.e. when it is disabled or the plan is wider than
`spark.sql.codegen.maxFields`, the projection then computes every expensive
alias on every row the pushed conditions keep.
The split on master happens to protect some of these queries: `laterQuery`
(`f >= 0 && g + 1 > 6 && r > 0`, with `r` requiring `id > 5`) succeeds on
master with the split on, since the split computes `r` only on the rows `g + 1
> 6` keeps, but fails on this PR without codegen, which is why the
`DataFrameSuite` loop runs it only `if (codegen || !avoid)`. On a wide table it
is also a cost regression, since every UDF runs on every row.
The description notes this, and SPARK-60731, which splits in the written
order, would keep `laterQuery` working with the split on. Could SPARK-60731
land before this PR, or with it, so that `master` does not turn the split off
for these filters in between?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala:
##########
@@ -266,13 +274,25 @@ object ExprUtils extends EvalHelper with QueryErrorsBase {
e.deterministic && !e.containsPattern(PLAN_EXPRESSION) &&
canEvaluateUnconditionallyInternal(e)
+ // Only the numeric pairs are taken from `Cast.canUpCast`, which also covers
e.g. LONG to
+ // TIMESTAMP and casts to STRING or between complex types: widening a number
never fails.
+ private def isNumericUpCast(from: DataType, to: DataType): Boolean = (from,
to) match {
+ case (_: NumericType, _: NumericType) => Cast.canUpCast(from, to)
+ case _ => false
+ }
+
private def canEvaluateUnconditionallyInternal(e: Expression): Boolean = e
match {
Review Comment:
**6. [Low] The allowlist does not look at `evalMode`**
With ANSI off (`EvalMode.LEGACY`), and under `try_*` (`EvalMode.TRY`),
numeric arithmetic and most casts return a wrapped value or NULL instead of
failing, but the allowlist rejects them in every mode. So with
`spark.sql.ansi.enabled=false`, `f > 5 && a + 1 > 6`, `cast(s as int) > 0` or
`a / b > 1` stay above the projection, lose their pushdown below joins and to
the scan, and turn the split off, with no correctness gain. The only way back
is the legacy config, which also pushes the conditions that can fail.
The description lists this as a known limitation. Accepting numeric `Add`,
`Subtract`, `Multiply`, `Divide` and the like when their `evalMode` is not
`ANSI`, and the casts that can not fail in that mode, would recover most of the
cost for users who run with ANSI off. Since `canEvaluateUnconditionally` has
other callers, a follow-up may fit better, but it would help to know whether
one is planned.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2664,8 +2728,25 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 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.
+ // For the same reason, there is no split when a condition after the
first one held back
+ // above is rejected by `canPushAhead`, with only the cheap aliases
substituted (it can
+ // tell nothing about an expensive one such as a UDF): e.g. a division
over an expensive
+ // alias, or a cheap condition held back above, which needs no
expensive alias and so
+ // would go lowest. The filter then stays above the projection as
written, as without
+ // the split. So it does when a cheap component is kept above after
one over a Python
+ // UDF before the first one held back, or with none held back, as the
split could
+ // evaluate the later expensive ones ahead of it.
+ lazy val keepsWrittenOrder = !legacy &&
Review Comment:
**2. [Medium] With only Python UDF aliases, the split still orders a
condition that can fail by cost**
With no condition held back, `keptHeld` is empty, and `keptCheapBeforeHeld`
is only set by a cheap component, so a condition that can fail over a Python
UDF alias still goes through the cost order of the split, as on master. For
example, with `p = py1(a)` and `q = py2(b)`, two different Python UDFs, and
ANSI on:
```scala
df.where($"p" > 1 && lit(10) / ($"q" - 3) > 0 && $"q" < 100)
```
The `q` group has two conditions, so the split places it lowest, and the
division runs on the rows `p > 1` drops, failing with `DIVIDE_BY_ZERO`. In 4.1,
the three conditions stay in their written order above one `BatchEvalPython`.
Neither SPARK-59693 nor the split is released, so this is not a regression of a
release, but the description says that such filters are optimized as before,
while this is the reordering the PR is about.
Applying the check of item 1 to every component kept after the first one
kept above would cover this case, and would also subsume `keptCheapBeforeHeld`,
since a cheap component kept there is one that `canPushAhead` rejects. The cost
is that some filters over Python UDFs would no longer be split until
SPARK-60731.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,71 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 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(a =>
isExpensive(a._2.child))
- // Split the filter's components into cheap and expensive while
keeping track of
- // 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 (cheapWithUsed, expensiveWithUsed) = splitPushable(
- originalAndRewrittenConditionsWithUsedAlias)(
- canPush = { case (_, used, _) => isCheap(used) },
- throwable = { case (_, _, replaced) => replaced.throwable })
+ // Whether a component reads an expensive alias that this projection
evaluates. A scalar
+ // Python UDF is not evaluated here: `ExtractPythonUDFs` later moves
it, with its
+ // arguments, to an eval node below the projection, and pushes the
conditions below that
+ // again, except the throwable ones written after one over its output
(SPARK-60633), so
+ // holding a deterministic condition back behind one here would gain
nothing and only cost
+ // the pushdown of the batches in between. What is left of the alias
without its Python
+ // UDFs, e.g. a Scala UDF over one, is evaluated here though. Each
Python UDF is replaced
+ // with an attribute, as `ExtractPythonUDFs` does, so that
`isExpensive` sees an input
+ // that is not foldable, as in the projection it leaves.
+ def evaluatedHere(used: AttributeMap[Alias]): Boolean = used.exists {
case (_, alias) =>
+ isExpensive(alias.child) && isExpensive(alias.child.transformDown {
+ case udf if isScalarPythonUDF(udf) =>
AttributeReference("pythonUDF", udf.dataType)()
+ })
+ }
+ // Split the filter's components, in their written order, into the
cheap ones pushed
+ // below the projection and the ones kept above it. The cheap
components written before
+ // the first one kept above are all pushed, as before SPARK-60730.
After it, a cheap
+ // component is only pushed when it can be evaluated ahead of the
components before it,
+ // and how strictly that is checked depends on why a component was
kept above:
+ // - after a component over an expensive alias that the projection
does not evaluate,
+ // i.e. one only expensive through scalar Python UDFs, a cheap
component is pushed
+ // when it is deterministic and not throwable;
+ // - after a component over an expensive alias the projection
evaluates, the first one
+ // held back, a cheap component is pushed only when `canPushAhead`
accepts it, so the
+ // others are only evaluated where none of the components before
them is false.
+ // The IS NOT NULL checks that `InferFiltersFromConstraints` puts in
front of the
+ // condition are pushed as if written there, so one over a cheap alias
that can fail,
+ // e.g. a division, still computes it ahead of the components written
before the one it
+ // was inferred from.
+ val components = originalAndRewrittenConditionsWithUsedAlias
+ val firstHeldIndex = components.indexWhere {
+ case (_, used, _) => !isCheap(used) && evaluatedHere(used)
+ }
+ val legacy = SQLConf.get.legacyPushFilterAheadOfExpensiveConjunct
+ val pushed = Seq.newBuilder[(Expression, AttributeMap[Alias],
Expression)]
+ val kept = Seq.newBuilder[(Expression, AttributeMap[Alias],
Expression)]
+ val keptHeld = Seq.newBuilder[(Expression, AttributeMap[Alias],
Expression)]
+ // 0: nothing kept above yet; 1: after a component only expensive
through Python UDFs;
Review Comment:
**9. [Low] Most of these cases come from the cost order of the split and the
position of the inferred filters**
Items 1 to 3 and 7 here, and several items of the earlier rounds, come from
two places: the split orders the conditions by cost, and the optimizer adds
filters without a written position. Each round has added a case to this rule
for them: the three levels of `strictness`, `keptCheapBeforeHeld`, the Python
UDF rules, and the dynamic pruning and bloom filter shapes in `canPushAhead`.
The description plans both fixes: SPARK-60731 for the written order of the
split, and a separate PR for the inferred checks. With those in place, much of
this state may not be needed: a condition after the first one kept above would
be pushed when `canPushAhead` accepts it, and kept in its written order
otherwise. Could those two land first, with this PR on top of them? The
`Aggregate` and `Window` cases also still only check `throwable` for a
condition written after one that stays up, so the same rule could apply there
too, here or in a follow-up.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -3086,6 +3167,38 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
(plan, computed, conditionsOf(pending))
}
+ /**
+ * Whether a filter component written after an expensive one may be
evaluated ahead of it,
+ * on rows the components before it would have filtered out (SPARK-60730).
Such a component
+ * must not be able to fail there, e.g. on a division by zero or an invalid
cast in ANSI mode,
+ * in a UDF or in `raise_error`, nor see a different set of rows than as
written, as a
+ * non-deterministic component would.
[[ExprUtils.canEvaluateUnconditionally]] is the check
+ * for that; it only accepts deterministic comparisons, IN, IS NULL and the
like over columns
+ * and literals, so it also rejects some components that can not fail, such
as arithmetic
+ * with ANSI mode off. Two filters the optimizer adds, which have no written
position to
+ * keep, are accepted too: a runtime bloom filter over keys the check
accepts, which can not
+ * fail, and a dynamic partition pruning filter, which the scan evaluates at
run time to prune
+ * what it reads, not on the filtered rows, so its position does not matter.
The filters
+ * that `InferFiltersFromConstraints` adds in front of the condition are
pushed as if written
+ * first, also over an alias the check would reject.
+ * With [[SQLConf.LEGACY_PUSH_FILTER_AHEAD_OF_EXPENSIVE_CONJUNCT]], only
throwable components
+ * are rejected.
+ *
+ * Callers pass the component with the aliases substituted that its
evaluation would
+ * compute, so a part inside such an alias counts too.
+ */
+ private def canPushAhead(component: Expression): Boolean =
+ if (SQLConf.get.legacyPushFilterAheadOfExpensiveConjunct) {
+ !component.throwable
+ } else {
+ component match {
Review Comment:
**7. [Low] Filters inferred from a join are appended after the condition,
and so held back**
The doc above covers the filters `InferFiltersFromConstraints` puts in front
of a `Filter`'s condition, but for a join side it adds a new `Filter` on top
(`inferNewFilter`), which `CombineFilters` merges as `And(nc, ac)`, i.e. after
the existing condition. So when a join key is a cheap alias that the allowlist
rejects, e.g. `k = cast(s as int)` or `k = a + 1`, the `isnotnull(k)`, `k = 10`
or `k IN (...)` inferred from the other side land after the held UDF condition:
they stay above the projection, lose their pushdown below it, and, through
`keptHeld`, turn the split off.
Like the dynamic pruning and bloom filters, they have no written position of
their own. The separate PR that moves the inferred checks may want to cover
this case too; until then, the doc could say that only the filters inferred for
the `Filter` itself come first.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,71 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 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(a =>
isExpensive(a._2.child))
- // Split the filter's components into cheap and expensive while
keeping track of
- // 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 (cheapWithUsed, expensiveWithUsed) = splitPushable(
- originalAndRewrittenConditionsWithUsedAlias)(
- canPush = { case (_, used, _) => isCheap(used) },
- throwable = { case (_, _, replaced) => replaced.throwable })
+ // Whether a component reads an expensive alias that this projection
evaluates. A scalar
+ // Python UDF is not evaluated here: `ExtractPythonUDFs` later moves
it, with its
+ // arguments, to an eval node below the projection, and pushes the
conditions below that
+ // again, except the throwable ones written after one over its output
(SPARK-60633), so
+ // holding a deterministic condition back behind one here would gain
nothing and only cost
+ // the pushdown of the batches in between. What is left of the alias
without its Python
+ // UDFs, e.g. a Scala UDF over one, is evaluated here though. Each
Python UDF is replaced
+ // with an attribute, as `ExtractPythonUDFs` does, so that
`isExpensive` sees an input
+ // that is not foldable, as in the projection it leaves.
+ def evaluatedHere(used: AttributeMap[Alias]): Boolean = used.exists {
case (_, alias) =>
+ isExpensive(alias.child) && isExpensive(alias.child.transformDown {
+ case udf if isScalarPythonUDF(udf) =>
AttributeReference("pythonUDF", udf.dataType)()
+ })
+ }
+ // Split the filter's components, in their written order, into the
cheap ones pushed
+ // below the projection and the ones kept above it. The cheap
components written before
+ // the first one kept above are all pushed, as before SPARK-60730.
After it, a cheap
+ // component is only pushed when it can be evaluated ahead of the
components before it,
+ // and how strictly that is checked depends on why a component was
kept above:
+ // - after a component over an expensive alias that the projection
does not evaluate,
+ // i.e. one only expensive through scalar Python UDFs, a cheap
component is pushed
+ // when it is deterministic and not throwable;
+ // - after a component over an expensive alias the projection
evaluates, the first one
+ // held back, a cheap component is pushed only when `canPushAhead`
accepts it, so the
+ // others are only evaluated where none of the components before
them is false.
+ // The IS NOT NULL checks that `InferFiltersFromConstraints` puts in
front of the
+ // condition are pushed as if written there, so one over a cheap alias
that can fail,
+ // e.g. a division, still computes it ahead of the components written
before the one it
+ // was inferred from.
+ val components = originalAndRewrittenConditionsWithUsedAlias
+ val firstHeldIndex = components.indexWhere {
+ case (_, used, _) => !isCheap(used) && evaluatedHere(used)
+ }
+ val legacy = SQLConf.get.legacyPushFilterAheadOfExpensiveConjunct
+ val pushed = Seq.newBuilder[(Expression, AttributeMap[Alias],
Expression)]
+ val kept = Seq.newBuilder[(Expression, AttributeMap[Alias],
Expression)]
+ val keptHeld = Seq.newBuilder[(Expression, AttributeMap[Alias],
Expression)]
+ // 0: nothing kept above yet; 1: after a component only expensive
through Python UDFs;
+ // 2: after the first component held back.
+ var strictness = 0
+ // Whether a cheap component was kept above at strictness 1, i.e. one
that is
+ // non-deterministic or throwable, before the first one held back, if
any.
+ var keptCheapBeforeHeld = false
+ components.zipWithIndex.foreach { case (component @ (cond, used,
replaced), index) =>
+ val push = isCheap(used) && (strictness match {
+ case 0 => true
+ case 1 => !replaced.throwable && (legacy || cond.deterministic)
+ case _ => canPushAhead(replaced)
Review Comment:
**4. [Medium] The `FilterExec` subexpression path now also fails queries
that pass on every release**
This is the second item of the second round, and the description now
documents it and plans the `FilterExec` change for a separate PR. I raise it
again because this PR widens it from conditions over expensive aliases to
plain-column conditions that `canPushAhead` rejects.
With `r` requiring `id > 5`, `f >= 0 && id + 1 > 6 && id + 1 < 100 && r > 0`
succeeds on 4.1, where the UDFs are inlined, and on 4.2, 4.3 and master, where
the additions are pushed below the projection, but fails on this PR with or
without whole-stage codegen, since `id + 1` is shared and `FilterExec` computes
`r` before the first condition. Holding back a condition such as `id + 1 > 6`
behind a UDF condition is a common shape, so could the `FilterExec` change land
before this PR, or with it?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,71 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 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(a =>
isExpensive(a._2.child))
- // Split the filter's components into cheap and expensive while
keeping track of
- // 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 (cheapWithUsed, expensiveWithUsed) = splitPushable(
- originalAndRewrittenConditionsWithUsedAlias)(
- canPush = { case (_, used, _) => isCheap(used) },
- throwable = { case (_, _, replaced) => replaced.throwable })
+ // Whether a component reads an expensive alias that this projection
evaluates. A scalar
+ // Python UDF is not evaluated here: `ExtractPythonUDFs` later moves
it, with its
+ // arguments, to an eval node below the projection, and pushes the
conditions below that
+ // again, except the throwable ones written after one over its output
(SPARK-60633), so
+ // holding a deterministic condition back behind one here would gain
nothing and only cost
+ // the pushdown of the batches in between. What is left of the alias
without its Python
+ // UDFs, e.g. a Scala UDF over one, is evaluated here though. Each
Python UDF is replaced
+ // with an attribute, as `ExtractPythonUDFs` does, so that
`isExpensive` sees an input
+ // that is not foldable, as in the projection it leaves.
+ def evaluatedHere(used: AttributeMap[Alias]): Boolean = used.exists {
case (_, alias) =>
+ isExpensive(alias.child) && isExpensive(alias.child.transformDown {
+ case udf if isScalarPythonUDF(udf) =>
AttributeReference("pythonUDF", udf.dataType)()
+ })
+ }
+ // Split the filter's components, in their written order, into the
cheap ones pushed
+ // below the projection and the ones kept above it. The cheap
components written before
+ // the first one kept above are all pushed, as before SPARK-60730.
After it, a cheap
+ // component is only pushed when it can be evaluated ahead of the
components before it,
+ // and how strictly that is checked depends on why a component was
kept above:
+ // - after a component over an expensive alias that the projection
does not evaluate,
+ // i.e. one only expensive through scalar Python UDFs, a cheap
component is pushed
+ // when it is deterministic and not throwable;
+ // - after a component over an expensive alias the projection
evaluates, the first one
+ // held back, a cheap component is pushed only when `canPushAhead`
accepts it, so the
+ // others are only evaluated where none of the components before
them is false.
+ // The IS NOT NULL checks that `InferFiltersFromConstraints` puts in
front of the
+ // condition are pushed as if written there, so one over a cheap alias
that can fail,
+ // e.g. a division, still computes it ahead of the components written
before the one it
+ // was inferred from.
+ val components = originalAndRewrittenConditionsWithUsedAlias
+ val firstHeldIndex = components.indexWhere {
+ case (_, used, _) => !isCheap(used) && evaluatedHere(used)
+ }
+ val legacy = SQLConf.get.legacyPushFilterAheadOfExpensiveConjunct
Review Comment:
**8. [Low] The new pushdown is not pinned for streaming checkpoints**
The description explains why this is not recorded in the checkpoint: the
change only keeps conditions higher than before. But for a `Filter` over a
`Project` with a UDF over an `EventTimeWatermark`, a cheap condition that the
batches before a restart pushed below the watermark stays above it after the
restart. The watermark then also counts the rows the condition drops, can
advance further, and drops more rows as late, within the same query and
checkpoint.
`REGEX_PATTERN_COST_ESTIMATE_VERSION` is in
`OffsetSeqMetadata.relevantSQLConfs`, with `0` for checkpoints that do not
record it, for the same kind of plan change. Adding
`LEGACY_PUSH_FILTER_AHEAD_OF_EXPENSIVE_CONJUNCT` there, with `true` for older
checkpoints, would keep a restarted query on its earlier plan, while new
queries get the new behavior.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,71 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 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(a =>
isExpensive(a._2.child))
- // Split the filter's components into cheap and expensive while
keeping track of
- // 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 (cheapWithUsed, expensiveWithUsed) = splitPushable(
- originalAndRewrittenConditionsWithUsedAlias)(
- canPush = { case (_, used, _) => isCheap(used) },
- throwable = { case (_, _, replaced) => replaced.throwable })
+ // Whether a component reads an expensive alias that this projection
evaluates. A scalar
+ // Python UDF is not evaluated here: `ExtractPythonUDFs` later moves
it, with its
+ // arguments, to an eval node below the projection, and pushes the
conditions below that
+ // again, except the throwable ones written after one over its output
(SPARK-60633), so
+ // holding a deterministic condition back behind one here would gain
nothing and only cost
+ // the pushdown of the batches in between. What is left of the alias
without its Python
+ // UDFs, e.g. a Scala UDF over one, is evaluated here though. Each
Python UDF is replaced
+ // with an attribute, as `ExtractPythonUDFs` does, so that
`isExpensive` sees an input
+ // that is not foldable, as in the projection it leaves.
+ def evaluatedHere(used: AttributeMap[Alias]): Boolean = used.exists {
case (_, alias) =>
Review Comment:
**11. [Nit] `evaluatedHere` transforms the aliases again for every component
and pass**
`indexWhere` calls `evaluatedHere` for each component up to the first held
one, and each call runs `transformDown` on every expensive alias the component
uses, so an alias read by several components is transformed once per component,
on every pass of the batch. Each transform also creates a new
`AttributeReference` per Python UDF, which takes a new `ExprId` from the global
counter. Computing the answer once per alias for the matched `Project`, with
one placeholder attribute reused, would avoid both.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,71 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 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(a =>
isExpensive(a._2.child))
- // Split the filter's components into cheap and expensive while
keeping track of
- // 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 (cheapWithUsed, expensiveWithUsed) = splitPushable(
- originalAndRewrittenConditionsWithUsedAlias)(
- canPush = { case (_, used, _) => isCheap(used) },
- throwable = { case (_, _, replaced) => replaced.throwable })
+ // Whether a component reads an expensive alias that this projection
evaluates. A scalar
+ // Python UDF is not evaluated here: `ExtractPythonUDFs` later moves
it, with its
+ // arguments, to an eval node below the projection, and pushes the
conditions below that
+ // again, except the throwable ones written after one over its output
(SPARK-60633), so
+ // holding a deterministic condition back behind one here would gain
nothing and only cost
+ // the pushdown of the batches in between. What is left of the alias
without its Python
+ // UDFs, e.g. a Scala UDF over one, is evaluated here though. Each
Python UDF is replaced
+ // with an attribute, as `ExtractPythonUDFs` does, so that
`isExpensive` sees an input
+ // that is not foldable, as in the projection it leaves.
+ def evaluatedHere(used: AttributeMap[Alias]): Boolean = used.exists {
case (_, alias) =>
+ isExpensive(alias.child) && isExpensive(alias.child.transformDown {
+ case udf if isScalarPythonUDF(udf) =>
AttributeReference("pythonUDF", udf.dataType)()
+ })
+ }
+ // Split the filter's components, in their written order, into the
cheap ones pushed
+ // below the projection and the ones kept above it. The cheap
components written before
+ // the first one kept above are all pushed, as before SPARK-60730.
After it, a cheap
+ // component is only pushed when it can be evaluated ahead of the
components before it,
+ // and how strictly that is checked depends on why a component was
kept above:
+ // - after a component over an expensive alias that the projection
does not evaluate,
+ // i.e. one only expensive through scalar Python UDFs, a cheap
component is pushed
+ // when it is deterministic and not throwable;
+ // - after a component over an expensive alias the projection
evaluates, the first one
+ // held back, a cheap component is pushed only when `canPushAhead`
accepts it, so the
+ // others are only evaluated where none of the components before
them is false.
+ // The IS NOT NULL checks that `InferFiltersFromConstraints` puts in
front of the
+ // condition are pushed as if written there, so one over a cheap alias
that can fail,
+ // e.g. a division, still computes it ahead of the components written
before the one it
+ // was inferred from.
+ val components = originalAndRewrittenConditionsWithUsedAlias
+ val firstHeldIndex = components.indexWhere {
+ case (_, used, _) => !isCheap(used) && evaluatedHere(used)
Review Comment:
**10. [Nit] `evaluatedHere` already implies `!isCheap`, and `canPushAhead`
reads the legacy config again**
`evaluatedHere(used)` requires an alias with `isExpensive(alias.child)`, so
`!isCheap(used) &&` adds nothing here. Similarly, `canPushAhead` reads
`legacyPushFilterAheadOfExpensiveConjunct` again, although the caller already
has it as `legacy`, and `keepsWrittenOrder` only calls it when `!legacy`.
Passing `legacy` in, or checking it once in the caller, would make the two
paths easier to follow.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -3086,6 +3167,38 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
(plan, computed, conditionsOf(pending))
}
+ /**
+ * Whether a filter component written after an expensive one may be
evaluated ahead of it,
+ * on rows the components before it would have filtered out (SPARK-60730).
Such a component
+ * must not be able to fail there, e.g. on a division by zero or an invalid
cast in ANSI mode,
+ * in a UDF or in `raise_error`, nor see a different set of rows than as
written, as a
+ * non-deterministic component would.
[[ExprUtils.canEvaluateUnconditionally]] is the check
+ * for that; it only accepts deterministic comparisons, IN, IS NULL and the
like over columns
+ * and literals, so it also rejects some components that can not fail, such
as arithmetic
+ * with ANSI mode off. Two filters the optimizer adds, which have no written
position to
+ * keep, are accepted too: a runtime bloom filter over keys the check
accepts, which can not
+ * fail, and a dynamic partition pruning filter, which the scan evaluates at
run time to prune
+ * what it reads, not on the filtered rows, so its position does not matter.
The filters
+ * that `InferFiltersFromConstraints` adds in front of the condition are
pushed as if written
+ * first, also over an alias the check would reject.
+ * With [[SQLConf.LEGACY_PUSH_FILTER_AHEAD_OF_EXPENSIVE_CONJUNCT]], only
throwable components
+ * are rejected.
+ *
+ * Callers pass the component with the aliases substituted that its
evaluation would
+ * compute, so a part inside such an alias counts too.
+ */
+ private def canPushAhead(component: Expression): Boolean =
+ if (SQLConf.get.legacyPushFilterAheadOfExpensiveConjunct) {
+ !component.throwable
+ } else {
+ component match {
+ case _: DynamicPruning => true
+ case BloomFilterMightContain(_, XxHash64(keys, _)) =>
+ keys.forall(ExprUtils.canEvaluateUnconditionally)
+ case _ => ExprUtils.canEvaluateUnconditionally(component)
Review Comment:
**5. [Medium] Partition conditions outside the allowlist, such as
`year(dt)`, lose pruning**
The second round's item 5 led to `LIKE '2024%'` being pushed again, but many
partition conditions are not plain: `year(dt) = 2024`, `cast(dt as date) >=
'2024-01-01'`, `substr(dt, 1, 4) = '2024'`. Written after a condition over a
nullable UDF, e.g. `df.withColumn("f", u($"x")).where($"f" === "a" &&
year($"dt") === 2024)` on a table partitioned by `dt`, they now stay above the
projection, and the inferred `isnotnull(f)` reads `f` a second time, so the
scan does not take the filter through the projection and reads every partition.
In 4.1, `avoidDoubleFilterEval` did not exist yet, so the whole filter was
pushed with the UDF inlined and the scan pruned with `year(dt) = 2024`. For
these queries the change is a full table scan compared with 4.1 as well, not
only with 4.2 and 4.3, while the migration guide entry says the evaluation is
as in Spark 4.1. Telling partition columns apart is probably not possible in
this rule, but the entry could name partition pruning explicitly, since a full
scan is likely the most visible cost for users who upgrade.
--
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]
