dongjoon-hyun commented on code in PR #59940:
URL: https://github.com/apache/spark/pull/59940#discussion_r4451396729
##########
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:
**1. [Medium] A condition written after a kept `IS NOT NULL` check is pushed
ahead of it**
The check now stays above the projection, but `afterKept`, and `cheapBefore`
after a condition over a Python UDF, still push the components written after it
unless they are throwable. So a condition that can fail now runs ahead of a
check that guarded it. For example, in ANSI mode, with `f` a non-nullable no-op
Scala UDF:
```scala
spark.range(0, 10, 1, 1)
.select($"id", ($"id" % 3) as "b", f($"id") as "f",
expr("try_divide(id, id % 3)") as "x")
.where($"x".isNotNull && $"id" / $"b" > 1 && $"f" > 5 && $"x" < 10)
```
1. `x < 10` is held back and reads `x`, and `canPushAhead` rejects
`isnotnull(try_divide(id, id % 3))`, so the check is kept above.
2. `id / b > 1` is in `afterKept`, so it is part of `leading` and is pushed
over the `Range`.
3. On `id` 0, where `x IS NOT NULL` is false and the written `AND` stops,
the division now runs, so the query should fail with `DIVIDE_BY_ZERO`.
On the released 4.1.3 and 4.3.0 the query returns the rows for `id` 7 and 8:
4.1.3 pushes the whole filter, and 4.3.0 the three cheap conditions, in their
written order. Master does the same as 4.3.0 before this PR.
Could the components written after the first kept check get the same rule as
the held ones, i.e. be pushed only when `canPushAhead` accepts them with the
aliases substituted? That rule also rejects a throwable component, so the
separate `replaced.throwable` partition would no longer be needed.
##########
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:
**2. [Medium] A non-deterministic condition after a condition over a Python
UDF is still pushed ahead of it**
Since SPARK-59693, which is in master and `branch-4.x` but not in 4.3, a
Python UDF is expensive, so a condition over its alias stays above the
projection. `evaluatedHere` does not hold the later conditions back behind it,
and `cheapBefore` pushes every cheap one that is not throwable, including a
non-deterministic one. `ExtractPythonUDFs` does not undo that. The batch after
it pushes only deterministic conditions through `BatchEvalPython`
(`pushDownPredicate`), so a non-deterministic condition left above would stay
after the UDF condition, while one pushed here stays below it. For example,
with `py = udf(lambda i: i, "long")`:
```python
(spark.range(0, 100, 1, 1)
.select(col("id"), py(col("id")).alias("p"))
.where((col("p") >= 50) & (rand(10) > 0.5)))
```
On the released 4.1.3 and 4.3.0 this returns 31 rows: the whole filter goes
below the projection together, so `rand(10)` only sees the rows where `p >= 50`
holds. Here `rand(10) > 0.5` is pushed below the projection, and after
`ExtractPythonUDFs` it sits directly over the `Range`. That plan, run on 4.3.0
as `spark.range(0, 100, 1, 1).where(rand(10) > 0.5).where($"id" >= 50)`,
returns a different set of 29 rows, the same 31 and 29 that the description
shows for a Scala UDF. I did not run master or this PR.
So the description's "With a Python UDF, the conditions are still pushed
below the eval node after `ExtractPythonUDFs`, as in 4.1", and the matching
sentence in the migration guide, do not hold for a non-deterministic condition.
Master already behaves this way, but this PR fixes the same reordering for the
other expensive expressions. Could `cheapBefore` also require
`cond.deterministic`? The condition would then stay above with the UDF
condition. After `ExtractPythonUDFs` the alias is a plain attribute, so the
whole filter would be pushed in its written order, as in 4.3.0.
##########
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:
**5. [Nit] Seven `span` and `partition` steps classify the components**
`written`/`rest`, `beforeHeld`/`fromHeld`, `firstHeld`/`held`,
`beforeKept`/`fromKept`, `keptWritten`/`afterKept`,
`cheapBefore`/`expensiveBefore` and `cheapHeld`/`expensiveHeld` each apply a
slightly different rule, and `expensiveWithUsed` relies on their concatenation
to keep the written order. In items 1 and 2, one of these groups gets a weaker
rule than the others.
A single pass over the components could keep a flag that is set at the first
component kept above for any reason, and require `canPushAhead` of every cheap
component after it. That would state the rule once. A condition over a Python
UDF could set a weaker flag that only requires a deterministic component that
is not throwable.
##########
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:
**4. [Low] No test covers a non-plain condition after a kept check, or a
non-deterministic one after a Python UDF condition**
This test only puts a plain `b < 10` or a throwable condition after
`IsNotNull($"x")`, and the Python UDF test has no non-deterministic condition
after the UDF condition, so items 1 and 2 pass the suite. A variant of
`throwableAfter` with `$"b" + 1 < 10` in place of `$"b" < 10`, and a `Rand(10)
> 0.5` after `$"p" > 1` in the Python UDF test, would pin them.
The legacy config with a throwable alias has no case either. There the check
now stays above, unlike in 4.3, but the legacy case in this test uses `a + 1`,
which is pushed.
##########
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:
**3. [Low] The inferred checks could be placed where they are inferred
instead**
As [the second-round
reply](https://github.com/apache/spark/pull/59940#discussion_r4429290160) says,
a check the user wrote in front stays above too, because this rule can not tell
it from one that `InferFiltersFromConstraints` inferred. Besides the cost noted
there, item 1 shows that holding a written check back also changes what runs
ahead of it. And the description lists the filters inferred from equalities as
not covered. Both come from putting the inferred conditions in front of the
condition, which is the only reason a check in front needs a special case here.
Could `InferFiltersFromConstraints` add them after the condition instead?
Then an inferred `isnotnull(x)` for a held `x < 10` would come after `f > 5`
and take the held path, so `keepsNotNull` would not be needed. A written check
would be pushed as in 4.3, and the filters inferred from equalities would be
covered the same way. `FilterExec` already evaluates an `IS NOT NULL` check
just before the first conjunct that reads the column, so the position should
matter little at run time. The plan stability golden files would change,
though, so this may be better as its own PR.
##########
docs/sql-migration-guide.md:
##########
@@ -67,6 +67,7 @@ license: |
- Since Spark 4.4, when `spark.sql.parquet.fieldId.read.enabled` is true, a
Parquet read fails with `PARQUET_FIELD_REQUESTED_MORE_THAN_ONCE` if two or more
requested fields of a struct (or of the top-level schema) resolve to the same
Parquet field. This happens when one requested field is matched by field ID and
another by name, or when two requested fields have the same field ID. In Spark
4.3 and earlier, such a read returned wrong values without an error: with the
vectorized reader one of the fields read as non-null default values (for
example `0`) or uninitialized data, and with the parquet-mr reader
(`spark.sql.parquet.enableVectorizedReader` set to `false`) one of the fields
read as `null`. To read the Parquet field more than once, request it once and
refer to the resulting column more than once in the query.
- Since Spark 4.4, when `spark.sql.groupByOrdinal` is true (the default), the
pipe `AGGREGATE` operator rejects an aliased integer literal in its `GROUP BY`
clause, such as `TABLE t |> AGGREGATE sum(x) AS s GROUP BY 2 AS grp`, with
`UNSUPPORTED_FEATURE.PIPE_OPERATOR_AGGREGATE_UNSUPPORTED_CASE`, because a bare
integer literal in the same position is an ordinal. Previously the aliased
literal was grouped as a constant. Bare ordinals and aliases on other grouping
expressions are not affected. To restore the previous behavior, set
`spark.sql.legacy.allowAliasedPipeAggregateOrdinals` to `true`.
- Since Spark 4.4, with whole-stage codegen enabled, a filter condition,
including the `FILTER` clause of an aggregate function, evaluates the remaining
top-level conjuncts of the condition on a row after one of them evaluates to
`NULL`, as the SQL standard requires, and does so in the order they appear in
the condition, as Spark already did with whole-stage codegen disabled, which
also evaluates a subexpression shared by several conjuncts before the
condition. Only a `FALSE` conjunct stops the evaluation. One exception: an `IS
NOT NULL` check on a column, or on an expression over it, that a later conjunct
references is still evaluated before that conjunct and the conjuncts up to the
check's own position, even when it is written after them, and can drop the row
before an error one of them raises. Once a condition has a conjunct that can be
`NULL`, the `IS NOT NULL` checks after that conjunct are evaluated in the order
of the condition on every row, not only on rows where a conjunct
is `NULL`. A row whose condition is `NULL` is still dropped, but an error
raised by a later conjunct on that row, such as `ARITHMETIC_OVERFLOW` in ANSI
mode, now fails the query. For example, `SELECT a FROM t WHERE NOT (a > 0 AND b
> 0) AND y + 1 > 0` can fail with `ARITHMETIC_OVERFLOW` on a row where `a` is
`NULL`, `b` is `1`, and `y` is `2147483647`, unless the row is dropped earlier
by a `FALSE` conjunct or `IS NOT NULL` check that precedes `y + 1 > 0` in the
condition, or skipped before the filter, for example by data source filter
pushdown such as Parquet row group statistics. Previously the query skipped
such a row with whole-stage codegen enabled. The later conjuncts also run on
more rows than before, so a nondeterministic conjunct such as `rand(seed)` can
return different values on the remaining rows, and a side-effecting
user-defined function is called on rows that are dropped. To restore the
previous behavior, set `spark.sql.legacy.filterShortCircuitOnNullConjunct` to
`tru
e`.
+- Since Spark 4.4, when `spark.sql.optimizer.avoidDoubleFilterEval` is true
(the default), a filter condition written after a condition over an expensive
expression that a projection evaluates, such as a Scala user-defined function,
is evaluated after it, as in Spark 4.1, unless it is a plain deterministic
condition over columns and literals, such as a comparison, `IN`, `IS NULL`,
`size`, `length`, `octet_length`, `startswith`, `endswith` or `contains` on
strings, or a `LIKE` without `_` that the optimizer turns into these, such as
`LIKE 'abc%'` or `LIKE 'ab%c'`, including a comparison with an implicit numeric
widening cast, such as between an `INT` and a `BIGINT` column. A column that a
view, a subquery or a DataFrame `select` defines with a cheap expression counts
as that expression, also in an `IS NOT NULL` check written or inferred before
the expensive condition, which then stays above the projection too when it is
not plain and a condition after the expensive one reads the colu
mn. The conditions written between the check and the expensive condition that
are still pushed then run ahead of it, and with whole-stage codegen it can be
evaluated just before the first condition that reads the column, so the
expensive condition can also run on the rows it drops. A throwable condition
written after such a check stays above with it. This does not cover the other
filters that the optimizer infers from equalities, which are still pushed as if
written first. In Spark 4.2 and 4.3, such a condition could be pushed below the
projection and evaluated first, also on the rows the earlier condition drops,
so a query could fail with an error such as `DIVIDE_BY_ZERO` or
`CAST_INVALID_INPUT` in ANSI mode, or with an exception from a user-defined
function. A condition that is not plain, such as one with arithmetic, another
cast, another string function, an `RLIKE` or another `LIKE`, or a subquery, is
now kept above the projection even when it can not fail, so it is no longer pus
hed further down either, for example below a join or a streaming event-time
watermark, which then also counts the rows it drops, or, when the filter reads
a column that the projection computes more than once, such as with the `IS NOT
NULL` check inferred for a nullable user-defined function, to the scan, where
it pruned partitions or was pushed to the data source. The earlier condition
now also runs on the rows that such a condition used to drop first, and so do
the expressions the projection computes when whole-stage codegen is not used,
for example when it is disabled or the plan is wider than
`spark.sql.codegen.maxFields`, since an interpreted projection computes them
all on every row it gets. With whole-stage codegen, this also happens to the
columns the filter reads when two of its conditions share an expression such as
`id + 1`, since the filter then computes them all before its first condition,
and to an expression that two projected expressions share, such as a
user-defined
function read field by field, since the projection computes it on every row it
gets. So a query that relied on that can now fail, and a non-deterministic
condition such as `rand(seed)` can select different rows than in Spark 4.2 and
4.3. Python user-defined functions are evaluated below the projection, so the
conditions written after a condition over one, and before any condition over an
expensive expression the projection computes, are still pushed below it, except
throwable ones such as `raise_error` and such an `IS NOT NULL` check. An
expensive expression over a Python user-defined function, such as a Scala
user-defined function over one, is evaluated in the projection though, so the
conditions after it are kept above, and the Python user-defined function now
also runs on the rows they used to drop first. With
`spark.sql.optimizer.splitProjectionForExpensiveFilters` (the default), the
projection split, which orders the conditions over expensive expressions by
cost, may still comp
ute the expensive expression a later condition reads ahead of the conditions
written before it. Whether a `LIKE` or `RLIKE` is an expensive expression
follows `spark.sql.optimizer.regexPatternCostEstimateVersion`, so for a
streaming query restarted from a checkpoint that does not record it, every
`RLIKE` and every `LIKE` that the optimizer does not rewrite is. To push such
conditions below the projection again, except throwable ones such as
`raise_error`, set `spark.sql.legacy.pushFilterAheadOfExpensiveConjunct` to
`true`.
Review Comment:
**6. [Nit] The migration guide entry is a single bullet of about 720 words**
It covers the allowlist, the `IS NOT NULL` handling, the codegen paths,
Python UDFs, the split, the regex estimate with checkpoints, and the legacy
config. So a user who upgrades has to read all of it to find what changed and
how to restore it, while the other 4.4 entries are under 350 words. Could it
state the change and the config in two or three sentences, as its first lines
do, and leave the list of exceptions to the config doc or the description?
--
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]
