sarutak commented on code in PR #59940:
URL: https://github.com/apache/spark/pull/59940#discussion_r4469544808
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,69 @@ 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))
+ // 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 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 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 })
+ // first expensive one are pushed, except the IS NOT NULL checks kept
above and the
+ // throwable components written after one (see below).
+ // After the first component over an expensive alias the projection
evaluates, which is
+ // the first one held back, only the cheap components that
`canPushAhead` accepts are
+ // pushed; the others stay above with the expensive ones, so they are
only evaluated
+ // where none of the components before them is false. Before that one,
the cheap
+ // components that are not throwable are pushed, as before
SPARK-60730, except the IS NOT
+ // NULL checks kept above.
+ val (written, rest) = originalAndRewrittenConditionsWithUsedAlias.span
{
+ case (_, used, _) => isCheap(used)
+ }
+ val (beforeHeld, fromHeld) = rest.span {
+ case (_, used, _) => isCheap(used) || !evaluatedHere(used)
+ }
+ val (firstHeld, held) = fromHeld.splitAt(1)
+ // `InferFiltersFromConstraints` puts the IS NOT NULL checks it infers
in front of the
+ // condition, so such a check over a cheap alias that a held component
after the first
+ // one reads, e.g. a division, would be pushed and compute the alias
ahead of the
+ // components before that one. It stays above when `canPushAhead`
rejects it with the
+ // alias substituted, also when the user wrote it there. With
whole-stage codegen,
+ // `FilterExec` then evaluates an IS NOT NULL check on a column just
before the first
+ // component that reads the column, or earlier after a component that
can be NULL (see
+ // `precedingNotNullChecks`). A held component that reads the alias is
rejected
+ // too, so the projection is not split then (see `keepsWrittenOrder`).
+ val heldRefs = AttributeSet(held.flatMap(_._1.references))
+ def keepsNotNull(cond: Expression, used: AttributeMap[Alias],
replaced: Expression) =
+ cond.isInstanceOf[IsNotNull] && used.keys.exists(heldRefs.contains)
&&
+ !canPushAhead(replaced)
+ // As in `splitPushable`, a throwable component written after a check
kept above stays
+ // above too.
+ val (beforeKept, fromKept) = written.span {
+ case (cond, used, replaced) => !keepsNotNull(cond, used, replaced)
+ }
+ val (keptWritten, afterKept) = fromKept.partition {
+ case (cond, used, replaced) => keepsNotNull(cond, used, replaced) ||
replaced.throwable
+ }
+ val leading = beforeKept ++ afterKept
Review Comment:
Reproduced: `id / b > 1` was pushed alone over the `Range` and failed on
`id` 0. I first tried your suggestion, and the query then still failed with
whole-stage codegen. The filter above the projection was `isnotnull(x) AND id /
b > 1 AND f > 5 AND x < 10`, in the written order, but `FilterExec` emits an
`IS NOT NULL` check on a column just before the first conjunct that reads the
column, here `x < 10`, so the division, which does not read `x`, ran first.
Without codegen it returned rows 7 and 8. 4.3.0 and master push the check as
`isnotnull(try_divide(id, id % 3))`, which `FilterExec` evaluates in its
written position. So keeping a check above can not keep its written order for
the conditions after it that do not read the column, and this rule can not tell
a written check from an inferred one. I removed the kept check: a check
written, or inferred, before the first expensive condition is pushed with the
conditions after it as on master, so your query returns rows 7 and 8 in every
combination of codegen, the split and `avoidDoubleFilterEval`; it is in the
`DataFrameSuite` loop now, and `FilterPushdownSuite` checks the pushed order.
This brings back the case of the second round's items 1 and 6, an inferred
`isnotnull(x)` over a cheap alias that can fail, which also fails on master. So
I took out the end-to-end queries and the description's table row for the
division and the cast inside a projected alias that I added for item 6, and
named the case in the description as not covered here. I think item 3 is the
right fix for it.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,69 @@ 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))
+ // 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 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 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 })
+ // first expensive one are pushed, except the IS NOT NULL checks kept
above and the
+ // throwable components written after one (see below).
+ // After the first component over an expensive alias the projection
evaluates, which is
+ // the first one held back, only the cheap components that
`canPushAhead` accepts are
+ // pushed; the others stay above with the expensive ones, so they are
only evaluated
+ // where none of the components before them is false. Before that one,
the cheap
+ // components that are not throwable are pushed, as before
SPARK-60730, except the IS NOT
+ // NULL checks kept above.
+ val (written, rest) = originalAndRewrittenConditionsWithUsedAlias.span
{
+ case (_, used, _) => isCheap(used)
+ }
+ val (beforeHeld, fromHeld) = rest.span {
+ case (_, used, _) => isCheap(used) || !evaluatedHere(used)
+ }
+ val (firstHeld, held) = fromHeld.splitAt(1)
+ // `InferFiltersFromConstraints` puts the IS NOT NULL checks it infers
in front of the
+ // condition, so such a check over a cheap alias that a held component
after the first
+ // one reads, e.g. a division, would be pushed and compute the alias
ahead of the
+ // components before that one. It stays above when `canPushAhead`
rejects it with the
+ // alias substituted, also when the user wrote it there. With
whole-stage codegen,
+ // `FilterExec` then evaluates an IS NOT NULL check on a column just
before the first
+ // component that reads the column, or earlier after a component that
can be NULL (see
+ // `precedingNotNullChecks`). A held component that reads the alias is
rejected
+ // too, so the projection is not split then (see `keepsWrittenOrder`).
+ val heldRefs = AttributeSet(held.flatMap(_._1.references))
+ def keepsNotNull(cond: Expression, used: AttributeMap[Alias],
replaced: Expression) =
+ cond.isInstanceOf[IsNotNull] && used.keys.exists(heldRefs.contains)
&&
+ !canPushAhead(replaced)
+ // As in `splitPushable`, a throwable component written after a check
kept above stays
+ // above too.
+ val (beforeKept, fromKept) = written.span {
+ case (cond, used, replaced) => !keepsNotNull(cond, used, replaced)
+ }
+ val (keptWritten, afterKept) = fromKept.partition {
+ case (cond, used, replaced) => keepsNotNull(cond, used, replaced) ||
replaced.throwable
+ }
+ val leading = beforeKept ++ afterKept
+ val (cheapBefore, expensiveBefore) = beforeHeld.partition {
+ case (cond, used, replaced) =>
+ isCheap(used) && !replaced.throwable && !keepsNotNull(cond, used,
replaced)
Review Comment:
Reproduced: 29 rows, with `rand(10) > 0.5` directly over the `Range`. Done
as you suggest: after a condition over an alias that is only expensive through
Python UDFs, a cheap condition is now pushed only when it is deterministic and
not throwable. Your query now returns the 31 rows of 4.1, with the split on and
off: `rand(10) > 0.5` stays after `p >= 50`, and after `ExtractPythonUDFs` both
are evaluated in that order over `BatchEvalPython`. As `rand(10) > 0.5` is kept
above there, the projection is not split either, also when no condition is held
back, so the split does not move a later expensive condition, over a Python UDF
or over an alias the projection evaluates, below it. The split still does so
below a non-deterministic condition that is itself over an expensive alias,
such as `f > rand(10) && g > 6` with `f` and `g` UDFs, as on master since
SPARK-55014; SPARK-60731, which makes the split follow the written order, fixes
that, and the description names it. The description and
the config doc now say that a non-deterministic condition after a condition
over a Python UDF stays above too; the legacy config pushes it again. The
migration guide entry leaves the Python UDF case out, since a Python UDF is
only expensive since SPARK-59693, which is not in a release yet.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,69 @@ 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))
+ // 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 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 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 })
+ // first expensive one are pushed, except the IS NOT NULL checks kept
above and the
+ // throwable components written after one (see below).
+ // After the first component over an expensive alias the projection
evaluates, which is
+ // the first one held back, only the cheap components that
`canPushAhead` accepts are
+ // pushed; the others stay above with the expensive ones, so they are
only evaluated
+ // where none of the components before them is false. Before that one,
the cheap
+ // components that are not throwable are pushed, as before
SPARK-60730, except the IS NOT
+ // NULL checks kept above.
+ val (written, rest) = originalAndRewrittenConditionsWithUsedAlias.span
{
+ case (_, used, _) => isCheap(used)
+ }
+ val (beforeHeld, fromHeld) = rest.span {
+ case (_, used, _) => isCheap(used) || !evaluatedHere(used)
+ }
+ val (firstHeld, held) = fromHeld.splitAt(1)
+ // `InferFiltersFromConstraints` puts the IS NOT NULL checks it infers
in front of the
+ // condition, so such a check over a cheap alias that a held component
after the first
+ // one reads, e.g. a division, would be pushed and compute the alias
ahead of the
+ // components before that one. It stays above when `canPushAhead`
rejects it with the
+ // alias substituted, also when the user wrote it there. With
whole-stage codegen,
Review Comment:
Agreed, and item 1 shows why the checks are better handled where they are
inferred than here. I would like to do it in a separate PR, since it changes
the conjunct order of the plans with inferred filters, and so the plan
stability golden files, and it is independent of this one. Until then the
inferred checks are pushed as on master, which the description now names.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala:
##########
@@ -984,6 +984,461 @@ class FilterPushdownSuite extends PlanTest {
comparePlans(Optimize.execute(originalQuery.analyze),
correctAnswer.analyze)
}
+ test("SPARK-60730: a cheap condition that can fail is not pushed ahead of an
expensive one") {
+ Seq(true, false).foreach { split =>
+ withSQLConf(SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key ->
split.toString) {
+ val originalQuery = testStringRelation
+ .select($"a", expensiveRLike("magic") as "f", $"b")
+ .where($"f" && $"a" + 1 > 6 && $"b" > 1 && $"e".cast(IntegerType) >
0 &&
+ IsNotNull($"a"))
+ .analyze
+ val Seq(expensive, arithmetic, _, cast, _) =
splitConjunctivePredicates(
+ originalQuery.collectFirst { case Filter(condition, _) => condition
}.get)
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // Only the conditions that can not fail move below the expensive one.
The others stay
+ // above it in their written order, also with the split, which would
otherwise put
+ // them lowest since they need no alias.
+ val correctAnswer = testStringRelation
+ .where($"b" > 1 && IsNotNull($"a"))
+ .select($"a", expensiveRLike("magic") as "f", $"b")
+ .where($"f" && $"a" + 1 > 6 && $"e".cast(IntegerType) > 0)
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ assertTopFilterConjuncts(optimized, Seq(expensive, arithmetic, cast))
+
+ // Written ahead of every expensive condition, they are pushed as
usual.
+ val failableFirst = testStringRelation
+ .select($"a", expensiveRLike("magic") as "f", $"b")
+ .where($"a" + 1 > 6 && $"f")
+ .analyze
+
+ val failableFirstAnswer = testStringRelation
+ .where($"a" + 1 > 6)
+ .select($"a", expensiveRLike("magic") as "f", $"b")
+ .where($"f")
+ .analyze
+
+ comparePlans(Optimize.execute(failableFirst), failableFirstAnswer)
+ }
+ }
+ }
+
+ test("SPARK-60730: a cheap condition over an alias that can fail is not
pushed ahead of an " +
+ "expensive one") {
+ val originalQuery = testStringRelation
+ .select($"a", expensiveRLike("magic") as "f", $"a" + 1 as "x", $"b")
+ .where($"f" && $"x" > 6 && $"b" > 1)
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // `x > 6` reads no expensive alias, but reading `x` evaluates `a + 1`.
+ val correctAnswer = testStringRelation
+ .where($"b" > 1)
+ .select($"a", expensiveRLike("magic") as "f", $"a" + 1 as "x", $"b")
+ .where($"f" && $"x" > 6)
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-60730: an IS NOT NULL check in front of the condition over an
alias that can " +
+ "fail is not pushed ahead of an expensive one") {
+ // `InferFiltersFromConstraints`, which this suite does not run, puts the
IS NOT NULL
+ // checks it infers in front of the condition, as written here.
+ def project: LogicalPlan = testStringRelation
+ .select($"a", expensiveRLike("magic") as "f", $"a" + 1 as "x", $"a" as
"y", $"b")
+ Seq(true, false).foreach { split =>
+ withSQLConf(SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key ->
split.toString) {
+ // Pushing `isnotnull(x)` would compute `a + 1` ahead of `f`.
+ val originalQuery = project
+ .where(IsNotNull($"x") && $"f" && $"x" > 6 && $"b" > 1)
+ .analyze
+ val Seq(notNull, expensive, failable, _) = splitConjunctivePredicates(
+ originalQuery.collectFirst { case Filter(condition, _) => condition
}.get)
+ val optimized = Optimize.execute(originalQuery)
+ val correctAnswer = testStringRelation
+ .where($"b" > 1)
+ .select($"a", expensiveRLike("magic") as "f", $"a" + 1 as "x", $"a"
as "y", $"b")
+ .where(IsNotNull($"x") && $"f" && $"x" > 6)
+ .analyze
+ comparePlans(optimized, correctAnswer)
+ assertTopFilterConjuncts(optimized, Seq(notNull, expensive, failable))
+
+ // It is pushed when no held condition reads the alias, also when the
first expensive
+ // condition does, which is evaluated first anyway; or when the alias
can not fail.
+ Seq[Expression]($"f", $"f" || $"x" > 6).foreach { first =>
+ val notReadQuery = project
+ .where(IsNotNull($"x") && first && $"a" + 2 > 6)
+ .analyze
+ val notReadAnswer = testStringRelation
+ .where(IsNotNull($"a" + 1))
+ .select($"a", expensiveRLike("magic") as "f", $"a" + 1 as "x",
$"a" as "y", $"b")
+ .where(first && $"a" + 2 > 6)
+ .analyze
+ comparePlans(Optimize.execute(notReadQuery), notReadAnswer)
+ }
+ val plainQuery = project
+ .where(IsNotNull($"y") && $"f" && $"y" + 1 > 6)
+ .analyze
+ val plainAnswer = testStringRelation
+ .where(IsNotNull($"a"))
+ .select($"a", expensiveRLike("magic") as "f", $"a" + 1 as "x", $"a"
as "y", $"b")
+ .where($"f" && $"y" + 1 > 6)
+ .analyze
+ comparePlans(Optimize.execute(plainQuery), plainAnswer)
+ // Only an IS NOT NULL check stays above. A condition written first is
evaluated first
+ // anyway, so it is pushed.
+ val writtenQuery = project
+ .where($"x" > 0 && $"f" && $"x" < 100)
+ .analyze
+ val writtenAnswer = testStringRelation
+ .where($"a" + 1 > 0)
+ .select($"a", expensiveRLike("magic") as "f", $"a" + 1 as "x", $"a"
as "y", $"b")
+ .where($"f" && $"x" < 100)
+ .analyze
+ comparePlans(Optimize.execute(writtenQuery), writtenAnswer)
+ }
+ }
+ // Also after a condition over a Python UDF, which does not hold
conditions back.
+ val pythonUDF = PythonUDF("pythonUDF", null, IntegerType, Seq(attrA),
+ PythonEvalType.SQL_BATCHED_UDF, udfDeterministic = true)
+ val afterPython = testStringRelation
+ .select($"a", pythonUDF.as("p"), expensiveRLike("magic") as "f", $"a" +
1 as "x", $"b")
+ .where($"p" > 1 && IsNotNull($"x") && $"f" && $"x" > 6)
+ .analyze
+ comparePlans(Optimize.execute(afterPython), afterPython)
+ // A throwable condition written after the check stays above with it, in
written order,
+ // while a plain one is still pushed.
+ val raiseError = IsNull(RaiseError(Literal("boom")))
+ val throwableAfter = project
+ .where(IsNotNull($"x") && ($"b" > 1 || raiseError) && $"b" < 10 && $"f"
&& $"x" > 6)
Review Comment:
Added. The `IS NOT NULL` test now checks that a check written in front is
pushed with a division after it in their written order, and the Python UDF test
has `Rand(10) > 0.5` after `p > 1`, which stays above, while `a + 1 > 6` after
it is still pushed, plus the legacy config, which pushes it, a throwable
condition after `p > 1`, and a condition over a Python UDF before or after one
held back. The `throwableAfter` variant and the legacy case with a throwable
alias went away with the kept check.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2642,15 +2650,69 @@ 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))
+ // 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 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 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 })
+ // first expensive one are pushed, except the IS NOT NULL checks kept
above and the
+ // throwable components written after one (see below).
+ // After the first component over an expensive alias the projection
evaluates, which is
+ // the first one held back, only the cheap components that
`canPushAhead` accepts are
+ // pushed; the others stay above with the expensive ones, so they are
only evaluated
+ // where none of the components before them is false. Before that one,
the cheap
+ // components that are not throwable are pushed, as before
SPARK-60730, except the IS NOT
+ // NULL checks kept above.
+ val (written, rest) = originalAndRewrittenConditionsWithUsedAlias.span
{
+ case (_, used, _) => isCheap(used)
+ }
+ val (beforeHeld, fromHeld) = rest.span {
+ case (_, used, _) => isCheap(used) || !evaluatedHere(used)
+ }
+ val (firstHeld, held) = fromHeld.splitAt(1)
Review Comment:
Done. The rule now classifies the components in one loop, in their written
order, and keeps how strictly the cheap ones are checked: none before the first
component kept above, deterministic and not throwable after one only expensive
through Python UDFs, and `canPushAhead` after the first one held back. Only a
component over an expensive alias raises the level: a cheap one kept above,
such as `rand(10) > 0.5` after `p >= 50`, still lets a later deterministic one
be pushed ahead of it, as Spark allows for deterministic conditions, but it
keeps the projection from being split, so the split does not move a later
expensive condition ahead of it. The kept components keep their written order
by construction.
--
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]
