sarutak commented on code in PR #59940:
URL: https://github.com/apache/spark/pull/59940#discussion_r4423134506
##########
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:
Reproduced, with one change: `rand(10) > 2` is folded to false, which leaves
an empty relation, so I used `rand(10) * 10 > 100` instead. With the split on
and ANSI on, that query failed with `DIVIDE_BY_ZERO` on the previous head,
while it returned no rows before this PR. I removed the hold-back. When a
condition after the first one held back is rejected by the check, the
projection is now not split at all, so the filter stays above the projection as
written, as with the split off, and no later pass pushes part of it onto a
split layer. The query is in the `DataFrameSuite` loop now, and the cheap
throwable variant is a case of the new `FilterPushdownSuite` test "the
projection is not split when a condition after the first expensive one can
fail". SPARK-60731 (separate PR to follow) replaces the cost order of the split
with the written order, which lets it split such filters again.
##########
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:
Confirmed: with a Python UDF the division fails before this PR, on the
previous head, and still on the new one, and with a join below, a condition
held back above the projection lost its pushdown to its side of the join. Since
`ExtractPythonUDFs` moves a scalar Python UDF below the projection and the
batch after it pushes the conditions below it again, holding them back gains
nothing. So conditions are now only held back after a condition over an alias
that is still expensive without its scalar Python UDFs, i.e. one the projection
evaluates itself. Behind a Python UDF they are pushed as before, also when the
UDF is over an expensive expression, which it takes along to the eval node; an
expensive expression over a Python UDF, such as a Scala UDF over one, still
holds them back. With the join, the plan is now the same as before this PR. The
migration guide and the config doc now say that Python UDFs are not covered.
Keeping the conditions above the eval node in the generic case woul
d be a larger change, close to SPARK-60633, so I left it out of this PR.
##########
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:
Reproduced with nullable Scala UDFs: the query failed on the previous head
and succeeded before this PR. With the hold-back gone, the filter is not split
while `g + 1 > 6` is held, so the inferred `IS NOT NULL` conditions no longer
let the split compute `r` early. The query is in the `DataFrameSuite` loop now.
One caveat, now in the description and the migration guide: when whole-stage
codegen is not used, an interpreted projection computes every alias on every
row it gets, so holding conditions back above it lets every alias see the rows
they would have dropped. This query, and e.g. `f >= 0 && id + 1 > 6 && r > 0`
with `r` requiring `id > 5` even with the split off, now fail there, though
they succeeded before this PR. With whole-stage codegen an alias is only
computed where a condition needs it, so they succeed. And when the conditions
held back are all plain, the split still runs in cost order, so inferred `IS
NOT NULL` conditions can still let it compute an alias early. Rather
than special-casing the inferred conditions, SPARK-60731 (separate PR to
follow) makes the split follow the written order, which handles both.
##########
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:
Done: `Size`, and `Cast` when `Cast.canUpCast` holds for two numeric types,
are now in the allowlist of `ExprUtils.canEvaluateUnconditionally`, so the
`size(...) > 0` shape is gone. I only took the numeric pairs, which can not
fail, since `Cast.canUpCast` also covers e.g. `LONG` to `TIMESTAMP`, casts to
`STRING` and between complex types. `ExprUtilsSuite` covers both sides, and the
suites of the other callers pass.
##########
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:
Added: a new test, "the legacy config splits the projection as before", runs
the legacy config on a division over an expensive alias and on a filter with a
throwable condition, and checks the plans of the split as before this PR.
Keeping the projection unsplit under the legacy config fails it.
##########
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:
It targets master and branch-4.x (4.4.0) first. Whether it also goes to
branch-4.3 and branch-4.2 is open; I added a paragraph on it to the
description. If it does, the version and the migration guide entry would name
the release it first ships in there, and only the pushdown part would apply.
--
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]
