dongjoon-hyun commented on code in PR #59940:
URL: https://github.com/apache/spark/pull/59940#discussion_r4420313100
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2651,17 +2688,46 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// split orders conditions by cost, and cost is no way to decide how
many rows a
// 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.
+ // in a lower projection) ahead of the conditions written before them.
So do the
+ // conditions after the first expensive one that `canPushAhead`
rejects, e.g. a division
+ // over an expensive alias, or a cheap condition left above by
`canPushAhead`, which
+ // needs no expensive alias and so would go lowest, and the conditions
after those.
val (splitChild, splitAliases, stayUp) =
if (expensiveWithUsed.nonEmpty &&
SQLConf.get.splitProjectionForExpensiveFilters) {
- val (splittable, notSplittable) = expensiveWithUsed.partition {
- case (cond, _, replaced) => cond.deterministic &&
!replaced.throwable
+ // Only the cheap aliases are substituted: `canPushAhead` never
accepts an expensive
+ // one such as a UDF, so substituting it would reject every
condition over an
+ // expensive alias and stop the split. Whether the split may
compute an expensive
+ // alias ahead of the conditions written before it is a separate
question.
+ val cheapAliasMap = AttributeMap(aliasMap.filter { case (_, a) =>
!a.child.expensive })
+ // A condition after the first expensive one that `canPushAhead`
rejects stays above
+ // in this pass, and so does every condition written after it, so
the split does not
+ // put a later condition, or compute the aliases it needs, below
the rejected one now.
+ // If every condition written before it is placed by a later pass,
it is then the
+ // first one of the filter left above, and is pushed or split off
on top of them; the
+ // split orders the conditions after it by cost again. Otherwise
it stays above,
Review Comment:
**1. [Medium] `CombineFilters` can still move a held condition ahead of a
non-deterministic or throwable one**
When the split takes the first expensive condition off, the conditions held
above come back in the next pass. If the first of them are cheap, they are
pushed together as `leading` onto the `Filter` of the split layer.
`CombineFilters` then merges only the deterministic, non-throwable conditions
of that block into the lower `Filter`, and keeps the others on top. So a
condition that can fail ends up ahead of a non-deterministic or throwable
condition written before it. For example, with `f` and `h` Scala UDFs, ANSI
mode on and the split on:
```scala
spark.range(0, 10, 1, 1)
.select($"id", f($"id") as "f", h($"id") as "h")
.where($"f" >= 0 && rand(10) > 2 && lit(10L) / ($"id" - 3) > 0 && $"h" > 0)
```
The first pass splits `f >= 0` off, and the division sets `firstRejected` to
2, so `rand(10) > 2`, the division and `h > 0` stay above. In the second pass,
`rand(10) > 2` and the division are cheap and come first, so they are pushed as
one `Filter` onto `Filter(f >= 0)`. `CombineFilters` turns that into
`Filter(rand(10) > 2, Filter(f >= 0 AND division))`, and the query fails with
`DIVIDE_BY_ZERO` on `id` 3. In 4.1, and with the split off, `rand(10) > 2` is
false on every row, so the division is never evaluated and the query returns no
rows. A cheap throwable condition in place of `rand(10) > 2` behaves the same
way, since `CombineFilters` keeps it on top too.
So the held condition is not always "pushed or split off on top of them", as
the comment above says. Could the second pass push the conditions after a
non-deterministic or throwable leading one in a separate `Filter` on top of it?
Or could `CombineFilters` merge only the outer conditions before the first
non-deterministic or throwable one, with a `span` instead of the `partition`?
The `DataFrameSuite` loop would catch this with a query like the one above.
##########
docs/sql-migration-guide.md:
##########
@@ -63,6 +63,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
projected expression, such as a 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` or `IS NULL`. A comparison that needs an
implicit cast, such as between an `INT` and a `BIGINT` column, is not plain,
and a column that a view, a subquery or a DataFrame `select` defines with an
expression counts as that expression. 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, a cast, a string function or a subquery, is now kept above the
projection even when it can not fail, so it is no longer pushed further down
either, for example below a join. The earlier condition now also runs on the
rows that such a condition used to drop first, 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. To push such conditions below the
projection again, except throwable ones such as `raise_error`, set
`spark.sql.legacy.pushFilterAheadOfExpensiveConjunct` to `true`.
Review Comment:
**2. [Medium] With Python UDFs, the held conditions are pushed below the UDF
again after `ExtractPythonUDFs`**
`PythonUDF` is expensive, so the main batches keep the later conditions
above the projection. But the "Extract Python UDFs" batch of `SparkOptimizer`
runs `PushPredicateThroughNonJoin` again, after `ExtractPythonUDFs` has
replaced the UDF with a `pythonUDF0` attribute that a `BatchEvalPython` or
`ArrowEvalPython` below the `Project` computes. The alias is cheap now, so the
held conditions are pushed below the `Project`, and the generic `UnaryNode`
case then pushes the ones that only read the input columns below the eval node
too. For example, with a Python UDF `pf` and ANSI mode on, this still fails
with `DIVIDE_BY_ZERO`, as in 4.1:
```python
spark.range(0, 10, 1, 1).select("id", pf("id").alias("f")) \
.where((col("f") > 5) & (lit(10) / (col("id") - 3) > 0))
```
So for Python UDFs this PR does not change the order, but the held
conditions now miss the pushdown of the main batches. That batch only has
`PushPredicateThroughNonJoin`, so a condition held above a `Project` over a
`Join` no longer reaches the join side. SPARK-60633 only covers throwable
conditions in the generic case, so it does not change this either.
The migration guide (here) and the config doc say that a condition written
after one over "an expensive projected expression, such as a user-defined
function, is evaluated after it, as in Spark 4.1". Could they say that this
covers the expressions evaluated in the projection, such as Scala UDFs, but not
Python UDFs? Or should the generic case keep the conditions that `canPushAhead`
rejects above the eval node, when an expensive condition over its output is
written before them? The end-to-end tests only use Scala UDFs that return
`Long`.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala:
##########
@@ -750,6 +750,287 @@ class FilterPushdownSuite extends PlanTest {
assertTopFilterConjuncts(Optimize.execute(originalQuery),
Seq(nonDeterministic, unplaced))
}
+ 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", $"e".rlike("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", $"e".rlike("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", $"e".rlike("magic") as "f", $"b")
+ .where($"a" + 1 > 6 && $"f")
+ .analyze
+
+ val failableFirstAnswer = testStringRelation
+ .where($"a" + 1 > 6)
+ .select($"a", $"e".rlike("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", $"e".rlike("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", $"e".rlike("magic") as "f", $"a" + 1 as "x", $"b")
+ .where($"f" && $"x" > 6)
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-60730: a cheap non-deterministic condition is not pushed ahead
of an " +
+ "expensive one") {
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"b")
+ .where($"f" && Rand(10) > 0.5 && $"b" > 1)
+ .analyze
+ val Seq(expensive, nonDeterministic, _) = splitConjunctivePredicates(
+ originalQuery.collectFirst { case Filter(condition, _) => condition
}.get)
+
+ val optimized = Optimize.execute(originalQuery)
+
+ val correctAnswer = testStringRelation
+ .where($"b" > 1)
+ .select($"a", $"e".rlike("magic") as "f", $"b")
+ .where($"f" && Rand(10) > 0.5)
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ assertTopFilterConjuncts(optimized, Seq(expensive, nonDeterministic))
+ }
+
+ test("SPARK-60730: a condition that can fail is not split off ahead of the
conditions " +
+ "before it") {
+ val expensiveInt = If($"e".rlike("other"), $"a", $"b")
+ val originalQuery = testStringRelation
+ .select($"a", $"e".rlike("magic") as "f", $"e".rlike("third") as "h",
+ expensiveInt as "g", $"b")
+ .where($"f" =!= $"h" && Literal(10) / ($"g" - 3) > 0)
+ .analyze
+
+ val optimized = Optimize.execute(originalQuery)
+
+ // On cost alone the division over g would go lowest, running on every
row. Splitting f and
+ // h off instead keeps it behind the condition written before it.
+ val correctAnswer = testStringRelation
+ .select($"a", $"b", $"e", $"e".rlike("magic") as "f",
$"e".rlike("third") as "h")
+ .where($"f" =!= $"h")
+ .select($"a", $"f", $"h", expensiveInt as "g", $"b")
+ .where(Literal(10) / ($"g" - 3) > 0)
+ .analyze
+
+ comparePlans(optimized, correctAnswer)
+ }
+
+ test("SPARK-60730: the legacy config pushes a cheap condition that can fail
ahead of an " +
+ "expensive one") {
Review Comment:
**5. [Low] No test runs the legacy config on a condition that the split
holds back**
This test has one expensive alias, so the split has nothing to place, and
the `DataFrameSuite` check of the legacy config only runs the division query,
which does not reach the split either. So if `firstRejected` ignored the legacy
config, e.g. by calling `ExprUtils.canEvaluateUnconditionally` directly, every
test would still pass. Yet under the legacy config the query of "a condition
that can fail is not split off ahead of the conditions before it" would then be
split as with this change, not as before it. Could this test also check that
query? The list of mutants in the description includes "ignoring the legacy
config", which only holds for the first change.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2651,17 +2688,46 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// split orders conditions by cost, and cost is no way to decide how
many rows a
// 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.
+ // in a lower projection) ahead of the conditions written before them.
So do the
+ // conditions after the first expensive one that `canPushAhead`
rejects, e.g. a division
+ // over an expensive alias, or a cheap condition left above by
`canPushAhead`, which
+ // needs no expensive alias and so would go lowest, and the conditions
after those.
val (splitChild, splitAliases, stayUp) =
if (expensiveWithUsed.nonEmpty &&
SQLConf.get.splitProjectionForExpensiveFilters) {
- val (splittable, notSplittable) = expensiveWithUsed.partition {
- case (cond, _, replaced) => cond.deterministic &&
!replaced.throwable
+ // Only the cheap aliases are substituted: `canPushAhead` never
accepts an expensive
+ // one such as a UDF, so substituting it would reject every
condition over an
+ // expensive alias and stop the split. Whether the split may
compute an expensive
+ // alias ahead of the conditions written before it is a separate
question.
+ val cheapAliasMap = AttributeMap(aliasMap.filter { case (_, a) =>
!a.child.expensive })
+ // A condition after the first expensive one that `canPushAhead`
rejects stays above
+ // in this pass, and so does every condition written after it, so
the split does not
+ // put a later condition, or compute the aliases it needs, below
the rejected one now.
+ // If every condition written before it is placed by a later pass,
it is then the
+ // first one of the filter left above, and is pushed or split off
on top of them; the
+ // split orders the conditions after it by cost again. Otherwise
it stays above,
+ // with the conditions after it. The first expensive condition
+ // is exempt, since every condition written before it is already
pushed below.
+ // Throwable and non-deterministic conditions stay above as before
(SPARK-60602),
+ // without holding back the later ones.
+ val firstRejected = expensiveWithUsed.indexWhere({ case (cond, _,
replaced) =>
+ cond.deterministic && !replaced.throwable &&
+ !canPushAhead(replaceAlias(cond, cheapAliasMap))
+ }, 1) match {
+ case -1 => expensiveWithUsed.length
+ case i => i
}
+ val (splittableWithIndex, notSplittableWithIndex) =
+ expensiveWithUsed.zipWithIndex.partition { case ((cond, _,
replaced), i) =>
+ cond.deterministic && !replaced.throwable && i < firstRejected
+ }
+ val splittable = splittableWithIndex.map(_._1)
+ val notSplittable = notSplittableWithIndex.map(_._1)
Review Comment:
**8. [Nit] `splitAt` and `span` would replace the index bookkeeping**
The `indexWhere` from 1 with the `-1` sentinel, the `zipWithIndex.partition`
and the two `map(_._1)` say that every condition from the first rejected one on
is held. The `span` above says the same for the cheap part, so this could read
the same way:
```scala
def canSplit(c: (Expression, AttributeMap[Alias], Expression)):
Boolean =
c._1.deterministic && !c._3.throwable
val (first, later) = expensiveWithUsed.splitAt(1)
val (beforeRejected, held) = later.span { c =>
!canSplit(c) || canPushAhead(replaceAlias(c._1, cheapAliasMap))
}
val (splittable, unsplittable) = (first ++
beforeRejected).partition(canSplit)
val notSplittable = unsplittable ++ held
```
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala:
##########
@@ -2949,6 +2949,95 @@ class DataFrameSuite extends SharedSparkSession
}
}
+ test("SPARK-60730: a filter condition that can fail is not evaluated ahead
of its order") {
+ val f = udf((i: Long) => i)
+ val g = udf((i: Long) => i)
+ val h = udf((i: Long) => { require(i > 5, s"h called with $i"); i })
+ // Each query fails if the conjunct after the first is evaluated on an id
<= 5, which the
+ // first conjunct filters out.
+ def divideQuery(): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f")
+ .where($"f" > 5 && lit(10L) / ($"id" - 3) > 0)
+ def castQuery(): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f", when($"id" > 5,
lit("1")).otherwise(lit("x")) as "s")
+ .where($"f" > 5 && $"s".cast("int") > 0)
+ def udfQuery(): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f")
+ .where($"f" > 5 && h($"id") > 0)
+ // With three expensive aliases, so the projection split is involved too.
It splits `f` and
+ // `h` off first, as written, and leaves the division over `g` behind them.
+ val k = udf((i: Long) => i)
+ def splitQuery(): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f", k($"id") as "h", g($"id") as "g")
+ .where($"f" + $"h" > 10 && lit(10L) / ($"g" - 3) > 0)
+ // `q` only works for the ids the first conjunct keeps. The split must put
that conjunct
+ // lowest, as written, even though the check rejects its addition.
+ val q = udf((i: Long) => { require(i > 4, s"q called with $i"); i })
+ def firstQuery(): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f", q($"id") as "q")
+ .where($"f" + 1 > 5 && $"q" > 0)
+ // `r` only works for the ids the second conjunct keeps, so the split must
not put the third
+ // conjunct below the second, which can fail. Without the split, an
interpreted projection
+ // computes `q` and `r` for every row it gets.
+ val r = udf((i: Long) => { require(i > 5, s"r called with $i"); i })
+ def laterQuery(): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", f($"id") as "f", g($"id") as "g", r($"id") as "r")
+ .where($"f" >= 0 && $"g" + 1 > 6 && $"r" > 0)
+ // A non-deterministic conjunct sees the same rows as with the whole
filter pushed.
+ def randQuery(): DataFrame = spark.range(0, 100, 1, 1)
+ .select($"id", f($"id") as "f")
+ .where($"f" >= 50 && rand(10) > 0.5)
+ val randExpected = withSQLConf(SQLConf.AVOID_DOUBLE_FILTER_EVAL.key ->
"false") {
+ randQuery().collect().toSeq
+ }
+ for (codegen <- Seq(true, false); split <- Seq(true, false); avoid <-
Seq(true, false)) {
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.WHOLESTAGE_CODEGEN_ENABLED.key -> codegen.toString,
+ SQLConf.AVOID_DOUBLE_FILTER_EVAL.key -> avoid.toString,
+ SQLConf.SPLIT_PROJECTION_FOR_EXPENSIVE_FILTERS.key -> split.toString) {
+ checkAnswer(divideQuery(), (6L until 10L).map(i => Row(i, i)))
+ checkAnswer(castQuery(), (6L until 10L).map(i => Row(i, i, "1")))
+ checkAnswer(udfQuery(), (6L until 10L).map(i => Row(i, i)))
+ checkAnswer(splitQuery(), (6L until 10L).map(i => Row(i, i, i, i)))
+ checkAnswer(randQuery(), randExpected)
+ if (codegen || split || !avoid) {
+ checkAnswer(firstQuery(), (5L until 10L).map(i => Row(i, i, i)))
+ checkAnswer(laterQuery(), (6L until 10L).map(i => Row(i, i, i, i)))
+ }
+ }
+ }
+ // The legacy config pushes the division below the projection again.
+ withSQLConf(
+ SQLConf.ANSI_ENABLED.key -> "true",
+ SQLConf.LEGACY_PUSH_FILTER_AHEAD_OF_EXPENSIVE_CONJUNCT.key -> "true") {
+ val e = intercept[SparkArithmeticException](divideQuery().collect())
+ assert(e.getCondition === "DIVIDE_BY_ZERO")
+ }
+ }
+
+ test("SPARK-60730: an expensive filter condition sees the rows a later
condition drops " +
+ "unless that condition is a plain comparison") {
+ val p = udf((i: Long) => { require(i > 5, s"p called with $i"); i })
+ // `p` only works for the ids the later conjunct keeps, so these queries
depend on that
+ // conjunct being pushed below the projection and evaluated first.
+ def query(later: Column): DataFrame = spark.range(0, 10, 1, 1)
+ .select($"id", p($"id") as "p")
+ .where($"p" > 0 && later)
+ val expected = (6L until 10L).map(i => Row(i, i))
+ // A plain comparison can not fail, so it is still pushed.
+ checkAnswer(query($"id" > 5), expected)
+ // An addition is not a plain comparison, so it now stays behind `p > 0`,
as in Spark 4.1,
+ // even with ANSI mode off, where it can not fail.
+ withSQLConf(SQLConf.ANSI_ENABLED.key -> "false") {
+ val e = intercept[SparkException](query($"id" + 1 > 6).collect())
+ assert(e.getCondition === "FAILED_EXECUTE_UDF")
Review Comment:
**7. [Low] This test pins a failure that a later improvement would remove**
The `intercept` asserts that `p > 0 AND id + 1 > 6` fails with ANSI mode
off, where the addition can not fail. That pins the conservative part of the
allowlist, not what the change is for. Accepting such an addition later (for
example `Add` with ANSI mode off, or an up-cast as in item 4) would be correct,
and would break this test. Could it check the plan instead, e.g. that the
addition stays in the top `Filter`, and keep the end-to-end part for the legacy
config?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2651,17 +2688,46 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// split orders conditions by cost, and cost is no way to decide how
many rows a
// 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.
+ // in a lower projection) ahead of the conditions written before them.
So do the
+ // conditions after the first expensive one that `canPushAhead`
rejects, e.g. a division
+ // over an expensive alias, or a cheap condition left above by
`canPushAhead`, which
+ // needs no expensive alias and so would go lowest, and the conditions
after those.
val (splitChild, splitAliases, stayUp) =
if (expensiveWithUsed.nonEmpty &&
SQLConf.get.splitProjectionForExpensiveFilters) {
- val (splittable, notSplittable) = expensiveWithUsed.partition {
- case (cond, _, replaced) => cond.deterministic &&
!replaced.throwable
+ // Only the cheap aliases are substituted: `canPushAhead` never
accepts an expensive
+ // one such as a UDF, so substituting it would reject every
condition over an
+ // expensive alias and stop the split. Whether the split may
compute an expensive
+ // alias ahead of the conditions written before it is a separate
question.
+ val cheapAliasMap = AttributeMap(aliasMap.filter { case (_, a) =>
!a.child.expensive })
+ // A condition after the first expensive one that `canPushAhead`
rejects stays above
+ // in this pass, and so does every condition written after it, so
the split does not
+ // put a later condition, or compute the aliases it needs, below
the rejected one now.
+ // If every condition written before it is placed by a later pass,
it is then the
+ // first one of the filter left above, and is pushed or split off
on top of them; the
+ // split orders the conditions after it by cost again. Otherwise
it stays above,
+ // with the conditions after it. The first expensive condition
+ // is exempt, since every condition written before it is already
pushed below.
+ // Throwable and non-deterministic conditions stay above as before
(SPARK-60602),
+ // without holding back the later ones.
+ val firstRejected = expensiveWithUsed.indexWhere({ case (cond, _,
replaced) =>
+ cond.deterministic && !replaced.throwable &&
+ !canPushAhead(replaceAlias(cond, cheapAliasMap))
Review Comment:
**3. [Low] Inferred `IsNotNull` conditions are written first, so the split
can compute held aliases early**
`InferFiltersFromConstraints` puts the conditions it infers in front of the
condition, e.g. `isnotnull(r)` for `r > 0` when the UDF behind `r` is nullable
(every Python UDF, or a Scala UDF that returns an object). `canPushAhead`
accepts them, and they come before the held conditions. So in the batch after
inferring filters, the split places them, and computes their aliases, ahead of
the conditions that this PR holds them behind. For example, with nullable UDFs
`f`, `g`, `r` and `s`, and the split on:
```scala
.where($"f" > rand(10) && $"g" + 1 > 6 && $"r" > 0 && $"s" > 0)
```
Before inferring filters, nothing is split: the first condition is
non-deterministic, and `g + 1 > 6` holds back the rest. After it, the filter
starts with `isnotnull(f)`, `isnotnull(g)`, `isnotnull(r)` and `isnotnull(s)`,
`firstRejected` is 5, and the split computes `f`, `g` and `r` below the rest,
so `r` runs on every non-null row. Without this PR, the first batch already
splits `g + 1 > 6` and `r > 0` off, so `r` only runs on the rows that `g + 1 >
6` keeps. So with an `r` that only works on those rows, and an interpreted
projection or a Python UDF, the query fails only with this PR.
This is only from reading the code, and the shape is narrow. But it goes
against the comment above, which says that the split does not "compute the
aliases it needs, below the rejected one". Could the inferred `IsNotNull`
conditions over expensive aliases be left out of the split, or be placed after
the conditions they come from?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/internal/SQLConf.scala:
##########
@@ -670,6 +670,26 @@ object SQLConf {
.booleanConf
.createWithDefault(true)
+ val LEGACY_PUSH_FILTER_AHEAD_OF_EXPENSIVE_CONJUNCT =
+ buildConf("spark.sql.legacy.pushFilterAheadOfExpensiveConjunct")
+ .internal()
+ .doc(s"When ${AVOID_DOUBLE_FILTER_EVAL.key} is true, a filter over a
projection with " +
+ "expensive (UDF, etc.) elements pushes its cheap conditions below the
projection and " +
+ "keeps the expensive ones above it. When this config is true, a cheap
condition written " +
+ "after an expensive one is pushed, and so evaluated ahead of it,
unless it is " +
+ "throwable (e.g. raise_error). When false, it is pushed only when it
is a plain " +
+ "deterministic condition over columns and literals, such as a
comparison, IN or IS " +
+ "NULL, where a column the projection defines with an expression counts
as that " +
+ "expression; otherwise it stays above in its written order, so it is
only evaluated " +
+ "where none of the conditions before it is false, and an error it
would raise on other " +
+ "rows (e.g. a division by zero or an invalid cast in ANSI mode, or an
exception in a " +
+ "UDF) is not raised. The projection split does not place such a
condition, with the " +
+ "expensive elements read as columns, ahead of the conditions written
before it either.")
+ .version("4.4.0")
Review Comment:
**6. [Low] Is this meant for `branch-4.3` and `branch-4.2` too?**
The description says the failures started in 4.2.0, and SPARK-60602 went to
`branch-4.3` and `branch-4.2` as well. If this goes there too,
`.version("4.4.0")` and the "Since Spark 4.4" entry would need to name the
release it first ships in (4.3.0 is still at the RC stage). Those branches also
have no projection split (SPARK-55014 is only in `master` and `branch-4.x`), so
they would only take the first change. If it is meant for 4.4 only, could the
description say so, since 4.2 and 4.3 users would keep the failures?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2627,16 +2633,47 @@ 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(_._2.child.expensive)
+ // Whether a component written after an expensive one may be evaluated
ahead of it, on
+ // rows the components before it would have filtered out. 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. It also accepts some filters
the optimizer
+ // adds, which have no written position to keep: a runtime bloom
filter over plain keys
+ // and the `size(...) > 0` that `InferFiltersFromGenerate` infers over
a plain array or
+ // map, 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 among the components does not matter. These are matched by
shape, so a
+ // written `size(col) > 0` is accepted too.
+ // Other kinds of added filters stay above, which only costs pushdown.
The legacy config
+ // only rejects throwable components. Callers pass the component with
the aliases
+ // substituted that its evaluation would compute, so a part inside
such an alias counts
+ // too.
+ 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 GreaterThan(Size(array, _), Literal(0, IntegerType)) =>
+ ExprUtils.canEvaluateUnconditionally(array)
Review Comment:
**4. [Low] Could `ExprUtils.canEvaluateUnconditionally` accept these
expressions instead of matching their shapes here?**
`Size` can not fail. With it in the allowlist of
`ExprUtils.canEvaluateUnconditionally`, the `size(...) > 0` case would not need
its own shape, and the check would also accept `size(arr) >= 1`, `size(arr) =
0` or `size(arr) > 1`, which the shape rejects although they can not fail
either. The same goes for an up-cast that type coercion adds
(`Cast.canUpCast`), as in the `INT` and `BIGINT` comparison that the migration
guide names. That is a common shape, and this PR now keeps it above the
projection, and so above a join below it. The other callers
(`FilterExec.mayRaise`, `StreamedSideJoinCondition` and
`RewriteDistinctAggregates`) would get the same benefit. The dynamic pruning
and bloom filter cases would still need their own match, since they hold a
subquery.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2627,16 +2633,47 @@ 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(_._2.child.expensive)
+ // Whether a component written after an expensive one may be evaluated
ahead of it, on
+ // rows the components before it would have filtered out. 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. It also accepts some filters
the optimizer
+ // adds, which have no written position to keep: a runtime bloom
filter over plain keys
+ // and the `size(...) > 0` that `InferFiltersFromGenerate` infers over
a plain array or
+ // map, 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 among the components does not matter. These are matched by
shape, so a
+ // written `size(col) > 0` is accepted too.
+ // Other kinds of added filters stay above, which only costs pushdown.
The legacy config
+ // only rejects throwable components. Callers pass the component with
the aliases
+ // substituted that its evaluation would compute, so a part inside
such an alias counts
+ // too.
+ def canPushAhead(component: Expression): Boolean =
+ if (SQLConf.get.legacyPushFilterAheadOfExpensiveConjunct) {
Review Comment:
**9. [Nit] `canPushAhead` could be a private method of the object**
It is a local `def` in the `Project` case that captures nothing from it, and
the doc of `splitProjectForExpensiveConditions` refers to it by name, where a
reader can not find it. As a private method next to `canPushThroughCondition`,
with its comment as Scaladoc, the case body would be shorter, and the method
could be tested 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]
