sarutak commented on code in PR #59940:
URL: https://github.com/apache/spark/pull/59940#discussion_r4429290160
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2627,17 +2633,40 @@ 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 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, 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.
+ def evaluatedHere(used: AttributeMap[Alias]): Boolean = used.exists {
case (_, alias) =>
+ alias.child.transformDown {
+ case udf if isScalarPythonUDF(udf) => Literal(null, udf.dataType)
+ }.expensive
+ }
// 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.
+ // first expensive one are all pushed. 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.
val (leading, rest) = originalAndRewrittenConditionsWithUsedAlias.span
{
case (_, used, _) => isCheap(used)
}
Review Comment:
Reproduced, thanks: with the non-nullable `f`, `isnotnull(10 / (id - 3))`
was pushed over the `Range`. Master fails these queries too, with a nullable
`f` as well, since it pushes `x > 0` itself, but this PR claimed to cover them.
Now an `IS NOT NULL` check written, or inferred, before the first condition
held back stays above the projection when it reads a cheap alias that a held
condition after the first one reads and `canPushAhead` rejects it with the
alias substituted, as you suggested first. That includes the case where a
condition over a Python UDF comes first, e.g. `p > 0 && f > 5 && x > 0`, where
the inferred `isnotnull(p)` comes before `isnotnull(x)`. With whole-stage
codegen, `FilterExec` then evaluates it just before the first conjunct that
reads the column, so after `f > 5`. The same holds for such a check the user
wrote in front, so `f > 5` now also sees the rows it drops; the description and
the migration guide say so. A held condition that reads the alias is rejecte
d too, so the projection is not split then. Under the legacy config,
`canPushAhead` only rejects throwable components, so the check is pushed as
before unless the alias is throwable. A condition written first that reads the
alias is still pushed, since it is evaluated first anyway. I did not take the
second option: the rule can not tell an inferred check from one the user wrote,
which no later condition may imply, and an expression can be NULL on non-null
inputs, e.g. a division by zero with ANSI off, so `IS NOT NULL` over its inputs
could only be added next to the check, not replace it. Two limits, now in the
description: without whole-stage codegen these queries still fail, since an
interpreted projection computes `x` on every row it gets, as before this PR;
and the other conditions that `InferFiltersFromConstraints` infers in front,
from equalities, are still pushed as if written first, also over such an alias.
Keeping those above would not help, since `FilterExec` evaluates them
in their position, which is first. In my first-round reply I wrote that I
would rather not special-case the inferred conditions; that was about the
projection split, which SPARK-60731 still handles, while this one is about the
pushdown, where only the `IS NOT NULL` checks get the lazy placement that makes
keeping them above help.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2627,17 +2633,40 @@ 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 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, 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.
+ def evaluatedHere(used: AttributeMap[Alias]): Boolean = used.exists {
case (_, alias) =>
+ alias.child.transformDown {
+ case udf if isScalarPythonUDF(udf) => Literal(null, udf.dataType)
+ }.expensive
+ }
// 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.
+ // first expensive one are all pushed. 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.
val (leading, rest) = originalAndRewrittenConditionsWithUsedAlias.span
{
case (_, used, _) => isCheap(used)
}
- val (cheapRest, expensiveWithUsed) = rest.partition {
+ val (beforeHeld, fromHeld) = rest.span {
+ case (_, used, _) => isCheap(used) || !evaluatedHere(used)
+ }
+ val (firstHeld, held) = fromHeld.splitAt(1)
+ val (cheapBefore, expensiveBefore) = beforeHeld.partition {
case (_, used, replaced) => isCheap(used) && !replaced.throwable
}
+ val (cheapHeld, expensiveHeld) = held.partition {
+ case (_, used, replaced) => isCheap(used) && canPushAhead(replaced)
+ }
Review Comment:
Reproduced. The mechanism is older than this PR: on master, with the split
off, `f >= 0 && g + 1 > 6 && g + 1 < 100 && r > 0`, with `g` a UDF too, fails
the same way, because those conditions are held back there too. This PR widens
it to conditions over plain columns. I also corrected the description, which
said "with whole-stage codegen, an alias is only computed where a condition
needs it", as did my reply in the first round; it now names this path and item
3. I would rather not change `FilterExec` in this PR. Computing only the
columns that the common subexpressions read would not cover a shared expression
over the alias itself, e.g. `x + 1 > 6 && x + 1 < 100` with `x` a division, so
the prologue would have to emit each column just before its first use. I would
like to do that in a separate PR. With the split on, SPARK-60731 should also
avoid your example, since the conditions written before `r > 0` would then sit
below the projection that computes `r`; I will check it there.
##########
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
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` or
`size`, 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.
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, 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, and when whole-stage codegen is not used, for example when it is
disabled or the plan is wider than `spark.sql.codegen.maxFields`, so does every
expensive expression the projection computes, since an interpreted projection
computes them all 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. This does not cover Python
user-defined functions, which 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. With
`spark.sql.optimizer.splitProjectionForExpensiveFilters` (the default), the
projection split, which orders the conditions over expensive expressions by
cost, may still compute the expensive expression a later condition reads ahead
of the conditions written before it. To push such conditions below the
projection again, except throwable ones such as `raise_error`, set
`spark.sql.legacy.pushFilterAheadOfExpensiveConjunct` to `true`.
Review Comment:
Reproduced. As with item 2, master has the same exposure for conditions it
holds back: with the split off, `f >= 0 && g > 5 && a > 0`, with `g` a UDF and
`a` and `b` sharing `p`, fails there too. The migration guide entry and the
description now name an expression that two projected expressions share, such
as a UDF read field by field. I left `ProjectExec` as is. Deferring a shared
subexpression past the filter is a general codegen change, and SPARK-60731 is
the structural answer for this shape with the split on.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2627,17 +2633,40 @@ 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 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, 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.
+ def evaluatedHere(used: AttributeMap[Alias]): Boolean = used.exists {
case (_, alias) =>
+ alias.child.transformDown {
+ case udf if isScalarPythonUDF(udf) => Literal(null, udf.dataType)
+ }.expensive
+ }
Review Comment:
Confirmed: `id + 1 > 6` now stays above the projection, over
`BatchEvalPython`. I kept the code. `f > 0` is written first and needs `py(id)`
on every row it gets, so `py` seeing those rows follows the written order; 4.1
and master only avoided it by evaluating `id + 1 > 6` first, which is the
reordering this PR removes for `s`. The migration guide and the description now
say that the Python UDF under such an expression runs on the rows that the
conditions after it used to drop first.
##########
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
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` or
`size`, 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.
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, 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, and when whole-stage codegen is not used, for example when it is
disabled or the plan is wider than `spark.sql.codegen.maxFields`, so does every
expensive expression the projection computes, since an interpreted projection
computes them all 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. This does not cover Python
user-defined functions, which 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. With
`spark.sql.optimizer.splitProjectionForExpensiveFilters` (the default), the
projection split, which orders the conditions over expensive expressions by
cost, may still compute the expensive expression a later condition reads ahead
of the conditions written before it. 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 for `substr(dt, 1, 4) = '2024'` and `dt LIKE '2024%'`, also with a
UDF marked `asNonNullable`, since a Scala UDF with a primitive parameter over a
nullable column is wrapped in a null check, so `isnotnull(f)` is inferred and
`f` is read twice. With item 8, `LIKE '2024%'` is pushed again and prunes the
partitions; `substr` and `year` are not yet. The migration guide and the
description now name partition pruning and data source filter pushdown next to
the join, for a filter that reads a column the projection computes more than
once. I did not special-case partition columns: pruning evaluates the condition
on every partition value, so one that can fail there, e.g. `cast(dt AS INT) >
0` in ANSI mode, would be evaluated ahead of `f > 0` again. Adding more
functions that can not fail to the allowlist, such as `substr` with literal
positions, `year` over a date or `lower`, which `ILIKE` goes through, would
reduce this further; I think that is better done separately, since each
one changes the other callers of `canEvaluateUnconditionally` too.
##########
sql/core/src/test/scala/org/apache/spark/sql/DataFrameSuite.scala:
##########
@@ -2949,6 +2949,117 @@ 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)
Review Comment:
Added. The `DataFrameSuite` loop now has the division and the cast inside
the projected alias, each with a non-nullable and a nullable `f`; the
non-nullable ones failed on the previous head, and all of them on master. Like
`laterQuery` and `inferredQuery`, they only run where whole-stage codegen is
used or the filter is pushed whole. `FilterPushdownSuite` has a test with the
`IS NOT NULL` check written in front, since its `Optimize` does not infer
filters, with the split on and off, after a condition over a Python UDF and
under the legacy config, and with the cases where it is still pushed: no held
condition after the first one reads the alias, the alias can not fail, or it is
not an `IS NOT NULL` check. The description has a row for the alias case now,
next to the "invalid cast" one, which keeps the cast in the filter.
--
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]
