dongjoon-hyun commented on code in PR #59940:
URL: https://github.com/apache/spark/pull/59940#discussion_r4424301716
##########
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:
**2. [Medium] `FilterExec` computes every alias the conditions read when
they share a sub-expression**
The description says that "with whole-stage codegen, an alias is only
computed where a condition needs it". That does not hold when
`FilterExec.doConsume` takes its CSE path
(`spark.sql.subexpressionElimination.filterExec.enabled`, on by default),
because `inputVarsEvalCode` computes the input columns of all `otherPreds` at
the top of the row loop. Before this PR, cheap conditions that share a
sub-expression were pushed below the projection. Now they stay in one `Filter`
with the conditions over the aliases. For example, with `r` a UDF that requires
`id > 5`:
```scala
spark.range(0, 10, 1, 1)
.select($"id", f($"id") as "f", r($"id") as "r")
.where($"f" >= 0 && $"id" + 1 > 6 && $"id" + 1 < 100 && $"r" > 0)
```
The two `id + 1` select that path, so `r` runs on `id` 0 ahead of every
condition, and the query fails with `FAILED_EXECUTE_UDF`, also with whole-stage
codegen. It succeeds in 4.1, where `r` is inlined into its own conjunct, and
before this PR, where the additions are pushed below. A cheap alias that can
fail is not protected on this path either.
Could the prologue only compute the columns that the common sub-expressions
read? Its comment says those are the ones CSE consumes. Otherwise, could the
description and the migration guide name this case too?
##########
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:
**1. [High] An inferred `IS NOT NULL` over a cheap alias that can fail is
still pushed first**
`InferFiltersFromConstraints` puts the conditions it infers in front of the
condition. So when a held condition reads a nullable cheap alias, the batch
after inferring filters sees `isnotnull(x)` as the first condition. It is
cheap, so it is `leading`, and `canPushAhead` is only asked about the
conditions after the first one held back. For example, with `f` the no-op UDF
of the description, which is not nullable, and ANSI mode on:
```scala
spark.range(0, 10, 1, 1)
.select($"id", f($"id") as "f", lit(10L) / ($"id" - 3) as "x")
.where($"f" > 5 && $"x" > 0)
```
1. Before inferring filters, `x > 0` is rejected, so the whole filter stays
above the projection.
2. `Divide` is always nullable, so `Filter.constraints` has `isnotnull(x)`
and the `Project` does not, and the filter becomes `isnotnull(x) AND f > 5 AND
x > 0`.
3. After inferring filters, `isnotnull(x)` is pushed as `isnotnull(10 / (id
- 3))` over the `Range`, and the query fails with `DIVIDE_BY_ZERO` on `id` 3.
In 4.1, and with `avoidDoubleFilterEval` off, the whole filter is pushed, so
nothing computes the division ahead of `f > 5`, and the query returns 4 rows.
The cast goes the same way once it is inside the alias, as in `.cast("int") as
"i"` with `i > 0`, since `Cast.forceNullable` is true from a string, or `CAST(s
AS INT) AS i` in a view. A nullable expensive alias hides this, because its own
`isnotnull(f)` then comes first and holds `isnotnull(x)` back.
So the migration guide's "A column that a view, a subquery or a DataFrame
`select` defines with a cheap expression counts as that expression" does not
hold yet for these, and neither does "Other filters the optimizer adds stay
above when the check rejects them" in the Scaladoc of `canPushAhead`.
Could an `IsNotNull` whose substituted form `canPushAhead` rejects stay
above the projection too? `FilterExec` already generates such a check just
before the first conjunct that reads the column. Or, for a null-intolerant
alias, it could push the `IsNotNull` of the alias's input columns instead,
which is what 4.1 infers.
##########
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:
**4. [Low] A Python UDF under a Scala UDF now runs on the rows a held
condition used to drop**
When the alias is a Scala UDF over a Python UDF, `evaluatedHere` holds the
later conditions back. But `ExtractPythonUDFs` still moves the Python UDF below
the projection, and the alias stays expensive, so the batch after it does not
push the held conditions either. For example, with `py` a Python UDF that only
works for `id > 5` and `s` a Scala UDF:
```sql
SELECT * FROM (SELECT id, s(py(id)) AS f FROM range(0, 10, 1, 1)) WHERE f >
0 AND id + 1 > 6
```
`py` now runs on `id` 0 in the eval node below the projection, and the query
fails. In 4.1 and before this PR, `id + 1 > 6` ends up below the eval node, so
`py` only sees the rows it keeps. Holding the condition back is still right for
`s`, so this may only need wording: the migration guide says "This does not
cover Python user-defined functions". Could it also say that a Python UDF under
an expression the projection evaluates now runs on those rows?
##########
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:
**5. [Medium] A held condition on a partition column can lose partition
pruning**
This entry describes the cost of holding a condition back as "so it is no
longer pushed further down either, for example below a join". The larger cost
may be at the scan. `PhysicalOperation` and `ScanOperation` only take a
`Filter` through a `Project` when `canCollapseExpressions` allows it, and it
refuses when the condition reads an expensive alias twice. That is the usual
case for a nullable UDF, since `InferFiltersFromConstraints` adds
`isnotnull(f)`. Then `PruneFileSourcePartitions` and `FileSourceStrategy` only
see the `Project` under the filter. For example, with `u` a nullable UDF and
`t` partitioned by `dt`:
```sql
SELECT * FROM (SELECT dt, u(x) AS f FROM t) WHERE f > 0 AND substr(dt, 1, 4)
= '2024'
```
`substr(dt, 1, 4) = '2024'` stays above the projection, next to
`isnotnull(f) AND f > 0`, so the query reads every partition. In 4.1 and before
this PR, it is pruned. `year(d) = 2024` and `dt LIKE '2024%'` behave the same
way. The result does not change, but on a large table this could be the largest
cost of the change.
Could the migration guide and the description name partition pruning and
data source filter pushdown, next to the join? Item 8 would also reduce this
for some conditions.
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/ExprUtils.scala:
##########
@@ -266,13 +269,22 @@ object ExprUtils extends EvalHelper with QueryErrorsBase {
e.deterministic && !e.containsPattern(PLAN_EXPRESSION) &&
canEvaluateUnconditionallyInternal(e)
+ // Only the numeric pairs are taken from `Cast.canUpCast`, which also covers
e.g. LONG to
+ // TIMESTAMP and casts to STRING or between complex types: widening a number
never fails.
+ private def isNumericUpCast(from: DataType, to: DataType): Boolean = (from,
to) match {
+ case (_: NumericType, _: NumericType) => Cast.canUpCast(from, to)
+ case _ => false
+ }
+
private def canEvaluateUnconditionallyInternal(e: Expression): Boolean = e
match {
case _: AttributeReference | _: Literal => true
case _: GetStructField | _: GetArrayStructFields | _: GetMapValue =>
e.children.forall(canEvaluateUnconditionallyInternal)
case _: And | _: Or | _: Not | _: BinaryComparison | _: IsNull | _:
IsNotNull |
- _: IsNaN | _: NullIf | _: Coalesce | _: In | _: InSet =>
+ _: IsNaN | _: NullIf | _: Coalesce | _: In | _: InSet | _: Size =>
e.children.forall(canEvaluateUnconditionallyInternal)
Review Comment:
**8. [Low] `StartsWith`, `EndsWith`, `Contains` and a field of a struct
built in the projection are rejected too**
As with `Size`, these can not fail, and they are common. `name LIKE 'abc%'`
becomes `StartsWith`, which data sources accept as `StringStartsWith`, so `f >
0 AND name LIKE 'abc%'` loses that pushdown, and with more expensive conditions
after it, the split too. `Like` and `RLike` themselves can fail on an invalid
pattern, so they should stay out.
Also, with `struct(a, b) AS s` in the projection, `s.a > 1` is checked as
`GetStructField(CreateNamedStruct(...)) > 1`, before `SimplifyExtractValueOps`
reduces it to `a > 1`, so a plain condition is rejected.
Could these go in the allowlist, with `CreateNamedStruct` taking the check
of its children? The other callers would accept them too, and item 5 would
apply to fewer conditions.
##########
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:
**3. [Medium] `ProjectExec` computes a sub-expression that aliases share on
every row**
The same statement has a second exception, in the projection itself.
`ProjectExec.doConsume` emits `subExprsCode` before `consume`, so a
sub-expression shared by two aliases is computed for every row that reaches the
projection, whatever the filter above it needs. A struct-returning UDF read
field by field is the common shape, as the Scaladoc of
`splitProjectForExpensiveConditions` notes. For example, with `p` a UDF that
requires `id > 5`:
```scala
spark.range(0, 10, 1, 1)
.select($"id", f($"id") as "f", p($"id") as "a", p($"id") + 1 as "b")
.where($"f" >= 0 && $"id" + 1 > 6 && $"a" > 0)
```
`id + 1 > 6` now stays above the projection, so `p` runs on `id` 0, and the
query fails with whole-stage codegen too. It succeeds in 4.1 and before this
PR. For the same reason, a cast or a division shared by two aliases is not
protected by the conditions written before it, even with item 1 fixed.
This entry limits the effect to "when whole-stage codegen is not used".
Could it also name a sub-expression that several aliases share?
##########
sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/Optimizer.scala:
##########
@@ -2652,8 +2681,20 @@ object PushPredicateThroughNonJoin extends
Rule[LogicalPlan] with PredicateHelpe
// 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.
+ // For the same reason, there is no split when a condition after the
first one held back
+ // above is rejected by `canPushAhead`, with only the cheap aliases
substituted (it can
+ // tell nothing about an expensive one such as a UDF): e.g. a division
over an expensive
+ // alias, or a cheap condition held back above, which needs no
expensive alias and so
+ // would go lowest. The filter then stays above the projection as
written, as without
+ // the split.
+ val cheapAliasMap = AttributeMap(aliasMap.filter { case (_, a) =>
!a.child.expensive })
+ val keepsWrittenOrder =
!SQLConf.get.legacyPushFilterAheadOfExpensiveConjunct &&
+ expensiveHeld.exists {
+ case (cond, _, _) => !canPushAhead(replaceAlias(cond,
cheapAliasMap))
+ }
Review Comment:
**9. [Nit] `cheapAliasMap` and `keepsWrittenOrder` are computed when nothing
needs them**
This builds a new `AttributeMap` for every `Filter` over a `Project`, on
every pass, also when `expensiveHeld` is empty or the split is off. It could be
a `lazy val`, with `splitProjectionForExpensiveFilters` checked first.
Similarly, `evaluatedHere` copies the tree of every alias in `used`, also the
cheap ones, where `alias.child.expensive && ...` would skip them.
##########
sql/catalyst/src/test/scala/org/apache/spark/sql/catalyst/optimizer/FilterPushdownSuite.scala:
##########
@@ -750,6 +750,314 @@ 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")
Review Comment:
**7. [Low] The `rlike("magic")` fixture and `evaluatedHere` depend on the
regex heuristic that #59879 changes**
`$"e".rlike("magic")` is only expensive because `_expensiveRegex` looks at
`left`, the subject, not at the pattern. That comes from before this PR, and
#59879 (SPARK-60632) fixes it. With that fix `rlike("magic")` is cheap, so
whichever PR goes in second has to rewrite these tests, and the expectation of
`regexOverPythonQuery` changes too. That PR adds a helper to this suite for the
existing tests, `$"e".rlike(Concat(Seq($"e", Literal(tag))))`, which is
expensive with both heuristics. Could the new tests use the same shape?
`evaluatedHere` has a similar dependency. With that fix, `Literal(null,
udf.dataType)` makes the pattern of `e RLIKE py(a)` foldable, and a NULL
pattern counts as cheap, so the alias no longer counts as evaluated here,
although the regex is still compiled per row in the projection.
Until then, every `RLIKE` alias over a column counts as expensive, so the
hold-back and the lost split apply to those aliases too, whatever the pattern.
The two PRs may be worth ordering.
##########
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:
**6. [Low] No end-to-end test computes an expression that can fail inside an
alias**
`divideQuery` and `castQuery` keep the division and the cast in the filter,
and the alias `s` here is not nullable. "a cheap condition over an alias that
can fail" in `FilterPushdownSuite` does have it inside the alias, but
`Optimize` in that suite has no `InferFiltersFromConstraints`, so
`isnotnull(x)` never appears. Adding the query of item 1 to this loop, and
`when($"id" > 5, lit("1")).otherwise(lit("x")).cast("int") as "i"` with `i >
0`, would catch item 1. The second one is also what the "invalid cast of a
cheap alias" row of the description's table reads like.
--
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]
