sarutak opened a new pull request, #59940:
URL: https://github.com/apache/spark/pull/59940
### What changes were proposed in this pull request?
`PushPredicateThroughNonJoin` splits the conditions of a `Filter` over a
`Project` with expensive aliases (UDFs, expensive regexes) into cheap and
expensive ones when `spark.sql.optimizer.avoidDoubleFilterEval` is true (the
default, SPARK-47672). It pushes the cheap ones below the projection and keeps
the expensive ones above it, so a cheap condition written after an expensive
one is evaluated first, also on the rows the expensive one would have filtered
out. SPARK-60602 stopped this for `throwable` conditions, but
`Expression.throwable` is opt-in metadata that ANSI runtime errors (division by
zero, invalid casts, overflows) and UDFs do not report.
This PR:
- pushes every cheap condition written before the first expensive one, as
before, and after it only the cheap conditions that
`ExprUtils.canEvaluateUnconditionally` accepts. That is the existing check for
moving an expression out of a short-circuited position (SPARK-58066). It is an
allowlist: deterministic comparisons, `IN`, `IS NULL`, `AND` / `OR` / `NOT` and
the like over columns and literals. It checks the condition with the
projection's aliases substituted, so a part inside a projected alias counts
too. It subsumes the `throwable` check from SPARK-60602. Being an allowlist, it
also rejects some conditions that can not fail, such as arithmetic with ANSI
mode off, a comparison that needs an implicit cast, a string function or a
subquery. 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 does not matter.
They are matched by shape, so a written `size(col) > 0` is accepted too. Other
kinds of added filters stay above, which only costs pushdown;
- applies the same check, with only the cheap aliases substituted, to the
conditions after the first expensive one that would enter the projection split
(SPARK-55014, unreleased). The split orders conditions by cost, so it would
otherwise put a division over an expensive alias, or a cheap condition the
first change left above, which needs no expensive alias, in the lowest layer. A
rejected condition stays above in that pass together with every condition
written after it, so the split does not put a later condition, or compute the
aliases it needs, below it then. 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. Throwable
and non-deterministic conditions keep the SPARK-60602 behavior, staying above
without holding back the later ones. The first expensive condition is
exempt, since every condition written before it is already pushed below;
- adds `spark.sql.legacy.pushFilterAheadOfExpensiveConjunct` (internal,
default `false`) to go back to only keeping `throwable` conditions above;
- documents the change in the migration guide and the rule's algorithm
overview.
This PR only covers the `Filter` over `Project` case. SPARK-60633 tracks a
related problem with throwable conditions in the generic `UnaryNode` case of
the same rule. The projection split can still compute an expensive alias, such
as a UDF, ahead of the conditions written before the condition reading it; that
is tracked separately in SPARK-60731.
### Why are the changes needed?
Since 4.2.0, queries that ran in 4.1 fail with the default configs. With `f`
a no-op UDF, so `f` is an expensive alias, and ANSI mode on:
```scala
val f = udf((i: Long) => i)
val df = spark.range(0, 10, 1, 1).select($"id", f($"id") as "f")
// Division by zero on id = 3, which f > 5 filters out.
df.where($"f" > 5 && lit(10L) / ($"id" - 3) > 0).collect()
// A UDF written directly in the filter that requires id > 5.
val h = udf((i: Long) => { require(i > 5); i })
df.where($"f" > 5 && h($"id") > 0).collect()
```
| Query | 4.1.4 | 4.2.0 | master before this change |
|---|---|---|---|
| division | OK | `DIVIDE_BY_ZERO` | `DIVIDE_BY_ZERO` |
| invalid cast of a cheap alias that is `'x'` on the rows `f > 5` drops | OK
| `CAST_INVALID_INPUT` | `CAST_INVALID_INPUT` |
| UDF in the filter | OK | `FAILED_EXECUTE_UDF` | `FAILED_EXECUTE_UDF` |
All of them succeed with `spark.sql.optimizer.avoidDoubleFilterEval=false`,
at the cost of evaluating the expensive aliases twice. SPARK-47672 stated that
results should be the same. Before this change, the projection split on master
moves conditions the same way: `f + g > 10 && 10 / (g - 3) > 0`, with `f` and
`g` UDFs, fails with the split on.
The same reordering changes which rows a non-deterministic condition sees:
`f >= 50 && rand(10) > 0.5` over `range(0, 100)` returns 31 rows in 4.1.4 and a
different set of 29 rows in 4.2.0 and on master.
### Does this PR introduce _any_ user-facing change?
Yes. With `avoidDoubleFilterEval` on, a condition written after a condition
over an expensive alias is evaluated after it, as in 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 is not plain, and a
column that the projection defines with a cheap expression counts as that
expression. Queries like the ones above, which failed in 4.2.0 and later, now
succeed. A non-deterministic condition now stays behind the expensive one, as
in 4.1, so `rand(10)` above sees the same rows as in 4.1. A plain condition
written after it can still be pushed ahead of it, as Spark allows for
deterministic conditions since 2.3, and so can an expensive condition the
projection split places below it.
Other conditions written after an expensive one, such as ones with
arithmetic, casts, string functions or subqueries, are no longer pushed below
the projection, even when they can not fail. So they are no longer pushed
further down either, for example below a join, and the expensive condition now
also runs on the rows they used to drop first. A query that relied on that can
now fail, as it did in 4.1. For example, with `p` a UDF that requires `id > 5`,
`p > 0 && id > 5` still succeeds since `id > 5` is a plain comparison and is
still pushed, but `p > 0 && id + 1 > 6` now fails, also with ANSI mode off.
Setting `spark.sql.legacy.pushFilterAheadOfExpensiveConjunct=true` pushes such
conditions again, except throwable ones such as `raise_error`. The migration
guide has an entry for this.
On master, the projection split also no longer places such a condition, with
the expensive aliases read as columns, ahead of the conditions written before
it. It can still compute an expensive alias, and evaluate a plain condition
over it, ahead of the conditions written before that condition, which
SPARK-60731 tracks.
Filters without conditions after the first expensive one, and filters over
projections without expensive aliases, are optimized as before.
### How was this patch tested?
New tests in `FilterPushdownSuite`: cheap conditions that can fail
(arithmetic, a cast) after an expensive one, including the order of the
conditions left above and with the split on and off; a cheap condition over an
alias that can fail; a non-deterministic condition; a condition that can fail
over an expensive alias in the split; the first expensive condition still split
off when the check rejects it; the conditions after a rejected one not split
off ahead of it; the filters the optimizer adds (dynamic pruning, runtime bloom
filter, `size(...) > 0`) still pushed, but not over a key or an array that is
not plain; throwable and non-deterministic conditions still not holding back
the later conditions in the split, as with SPARK-60602; and the legacy config.
End-to-end tests in `DataFrameSuite` run the division, cast, UDF, split and
`rand` queries with ANSI on over every combination of whole-stage codegen,
`splitProjectionForExpensiveFilters` and `avoidDoubleFilterEval`, and two
queries over UDFs that only work for the rows an earlier conjunct keeps over
every combination but interpreted projections without the split and with
`avoidDoubleFilterEval` on, which compute every alias on every row, and check
that the legacy config brings the `DIVIDE_BY_ZERO` back. Another runs the `p`
queries above, with and without the legacy config.
Each of these mutants fails at least one new or existing test: checking only
`throwable` again, checking the condition without substituting the aliases,
dropping the check from the split, applying it to the first expensive condition
too, not holding back the conditions after a rejected one, holding them back
after a throwable or a non-deterministic one too, not substituting the cheap
aliases in the split, not accepting each kind of filter the optimizer adds,
accepting a bloom filter over any key or a `size` over any array, ignoring the
legacy config, dropping the `throwable` check under the legacy config, and
pushing nothing ahead of the first expensive condition. All tests in
`org.apache.spark.sql.catalyst.optimizer`, `DataFrameSuite` and the TPC-DS
`PlanStabilitySuite`s pass, with no golden plan changes.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Kiro CLI
--
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]
