viirya opened a new pull request, #59944:
URL: https://github.com/apache/spark/pull/59944
### What changes were proposed in this pull request?
`PushPredicateThroughNonJoin` splits a filter into the conjuncts it can push
below an operator and the ones it keeps above. Apart from the `Filter` over
`Project` case fixed by SPARK-60602, these cases ignore `Expression.throwable`:
- the generic `UnaryNode` case (`pushDownPredicate`), for every operator in
`canPushThrough`: `BatchEvalPython`, `ArrowEvalPython`, `Generate`, `Expand`,
`Sort`, `Repartition`, ...;
- `Aggregate`, `Window`, `Union` and `EventTimeWatermark`.
So a throwable conjunct (`raise_error`, `assert_true`) can be pushed below
the operator on its own, ahead of a conjunct written before it that has to stay
above.
This PR gives all of these cases one shared split, `splitPushableConjuncts`.
Every conjunct written before the first one that cannot be pushed is pushed, as
before. After that one, only conjuncts that are not throwable are pushed. The
conjuncts left above keep the order they were written in. Before, the generic
and `Window` cases moved the non-deterministic conjuncts to the end, which
could also put a throwable conjunct ahead of a non-deterministic one written
before it.
Moving a non-throwable conjunct below a throwable one is still allowed,
since that can only reduce the rows the throwable conjunct sees. A throwable
conjunct with only pushable conjuncts before it is still pushed, so the
`Aggregate` pushdown of throwable predicates from SPARK-46707 still works.
`CombineFilters`, `PushPredicateThroughJoin` and SPARK-60602 already follow
these rules.
### Why are the changes needed?
A throwable conjunct can run on rows that the `AND` short-circuit would have
kept from it, so a query that should succeed fails. For example, with a Python
UDF `f`:
```python
spark.range(0, 10, 1, 1).select("id", f("id").alias("f")) \
.where((col("f") > 5) & assert_true(col("id") > 5).isNull()).collect()
```
After `ExtractPythonUDFs`, `SparkOptimizer` runs
`PushPredicateThroughNonJoin` again. It pushes `assert_true(id > 5) IS NULL`
alone below the `BatchEvalPython` / `ArrowEvalPython` node, and the query fails
with `USER_RAISED_EXCEPTION`, whatever
`spark.sql.optimizer.avoidDoubleFilterEval` is set to. The same happens without
Python:
```scala
spark.range(10).select($"id", explode(array($"id")) as "x")
.where($"x" > 5 && assert_true($"id" > 5).isNull) // Generate
spark.range(10).groupBy($"id").agg(max($"id") as "m")
.where($"m" > 5 && assert_true($"id" > 5).isNull) // Aggregate
spark.range(10).select($"id", max($"id").over(Window.partitionBy($"id")) as
"m")
.where($"m" > 5 && assert_true($"id" > 5).isNull) // Window
df.union(df).where(rand(0) < 0 && assert_true($"id" > 100).isNull) // Union
```
### Does this PR introduce _any_ user-facing change?
Yes. Queries like the ones above failed with `USER_RAISED_EXCEPTION` and now
succeed. This affects every release that has `Expression.throwable` (4.0.0 and
later).
Filters without throwable conjuncts are pushed as before. The one other
change is that conjuncts left above the operator in the generic and `Window`
cases now keep their written order, instead of the non-deterministic ones
moving to the end.
### How was this patch tested?
New tests in `FilterPushdownSuite` for `BatchEvalPython` /
`ArrowEvalPython`, `Sort`, `Aggregate`, `Window`, `Union` and
`EventTimeWatermark`. They check that a throwable conjunct stays above, that a
later non-throwable conjunct is still pushed, that a throwable conjunct with
nothing kept above before it is still pushed, and, using
`assertTopFilterConjuncts`, the written order of the conjuncts left above.
End-to-end tests:
- `DataFrameSuite` covers `Generate`, `Aggregate`, `Window` and `Union`,
with whole-stage codegen on and off.
- `PythonUDFSuite` covers a regular Python UDF and a scalar pandas UDF, with
`avoidDoubleFilterEval` on and off.
Each new test fails without the change. All tests in
`org.apache.spark.sql.catalyst.optimizer` pass, and so do `DataFrameSuite`,
`PythonUDFSuite`, `ExtractPythonUDFsSuite`, `BatchEvalPythonExecSuite` and the
`FilterPushdown*` suites in `sql/core`.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code (Claude Opus 5.5)
This pull request and its description were written by Isaac.
--
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]
