sarutak commented on code in PR #59940:
URL: https://github.com/apache/spark/pull/59940#discussion_r4471001389
##########
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:
Reproduced: with `p = py1(id)`, `s = scalaUdf(py2(id))` and `10 / (s - 1)`,
which fails on `id` 1, the query failed with `DIVIDE_BY_ZERO` with the split on
and returned rows 2 to 9 with it off. Done as you suggest: the split is now
checked for every condition kept above after the first one kept above, not only
after the first one held back, so here the division is checked and the filter
stays as written. `FilterPushdownSuite` has a query of that shape, with an
expensive expression over a Python UDF, and a `DataFrameSuite` test runs yours,
with `scalaUDF(py2(id))`, over the combinations of the split and
`avoidDoubleFilterEval`.
##########
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:
Reproduced with `p = py1(id)` and `q = py2(id)` as for item 1:
`DIVIDE_BY_ZERO` with the split on, rows 2 to 9 with it off. The check of item
1 covers it, and it replaces `keptCheapBeforeHeld`, since a cheap condition
kept there is one that `canPushAhead` rejects. As you note, a filter over
Python UDFs with a condition after the first one kept above that is not plain
is no longer split until SPARK-60731; the description now says that the filters
over Python UDFs are optimized as before only when their conditions kept above
after the first one are all plain.
##########
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:
Agreed. I will open SPARK-60731 against master first and rebase this PR on
it, so that `master` does not turn the split off for these filters in between.
With the split in the written order, `laterQuery` should then also run in the
`DataFrameSuite` loop without whole-stage codegen and with the split on.
##########
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:
Agreed. I will open the `FilterExec` change first too, and this PR will go
on top of it, so that `f >= 0 && id + 1 > 6 && id + 1 < 100 && r > 0` keeps
working with whole-stage codegen. Without it, the interpreted projection still
computes `r` on every row, as in item 3, which the written-order split of
SPARK-60731 can only avoid with the split on.
##########
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:
Agreed, a full scan is the most visible part. The migration guide entry now
names it: a condition on a partition column such as `year(dt) = 2024` then no
longer prunes partitions. The allowlist change for item 6 can also add
functions that can not fail, such as `year` over a date column, which would
bring back the pruning for some of these conditions; a cast from a string, as
in `cast(dt as date)`, can fail in ANSI mode, so that one stays above.
##########
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:
Agreed, with ANSI mode off this PR only has a cost for such conditions. I
plan a separate PR for it, before this one: `canEvaluateUnconditionally` would
accept numeric arithmetic and the casts that can not fail when their `evalMode`
is not `ANSI`. As its other callers would accept them too, it seems better
reviewed on its own.
--
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]
