sarutak commented on code in PR #59940:
URL: https://github.com/apache/spark/pull/59940#discussion_r4471004349
##########
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:
Right, `CombineFilters` appends the filter inferred for a join side as
`And(nc, ac)`. The `canPushAhead` doc now says that only the filters inferred
for the `Filter` itself come first, and that the ones inferred for a join side
are checked like conditions written last, as does the description. Accepting
them as they are would compute a key such as `cast(s as int)` ahead of the
condition again, so I would rather look at them in the
`InferFiltersFromConstraints` PR.
##########
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:
Done as you suggest: `spark.sql.legacy.pushFilterAheadOfExpensiveConjunct`
is now in `OffsetSeqMetadata.relevantSQLConfs`, and a checkpoint that does not
record it uses `true`, so a restarted query keeps pushing the conditions this
PR holds back, as before this PR. SPARK-47672's own change and the split are
not recorded, so a query from a 4.1 checkpoint still gets those, as the
description now says. `OffsetSeqLogSuite` checks the recorded and the default
values, and `StreamingQueryManagerSuite` that a query restarted from a
checkpoint without it pushes `value + 1 > 6` below the projection again. The
migration guide entry, the config doc and the description say that such a query
has to start from a new checkpoint to 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) =>
+ 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:
Agreed, and I plan to do it in that order, as in the top-level comment:
SPARK-60731 and the `InferFiltersFromConstraints` change first, then this PR on
top, where I expect the levels of `strictness`, the Python UDF rules and the
split check to shrink to what you describe. I would like to look at the
`Aggregate` and `Window` cases in a follow-up after this one.
##########
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:
Done: the `indexWhere` only calls `evaluatedHere`, and `canPushAhead` no
longer reads the legacy config; the caller checks `legacy` once, and only
throwable conditions are rejected under it, as before.
##########
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:
Done: `evaluatedHere` now caches its answer per alias for the matched
`Project`, and all the placeholders of one projection share one `ExprId`. It is
still computed again on each pass of the batch, as the rest of the rule is.
--
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]
