dongjoon-hyun opened a new pull request, #60117:
URL: https://github.com/apache/spark/pull/60117
### What changes were proposed in this pull request?
This PR fixes restarts of streaming queries from checkpoints written by
Spark 4.3.x or before, when a filter uses the result of a Python UDF.
SPARK-59693 made every Python UDF (`PythonFuncExpression`) expensive. With
`spark.sql.optimizer.avoidDoubleFilterEval=true` (the default),
`PushPredicateThroughNonJoin` now keeps a filter on a Python UDF alias above
its `Project`. Before, the filter was pushed below the `Project`. From there it
could move further below a stateful operator, such as into the input of a
stream-stream join, or below `EventTimeWatermark`. The checkpoint does not
record which classification a query ran with, so a query restarted from a
checkpoint of a released Spark version gets a different plan.
This PR:
- adds an internal config,
`spark.sql.optimizer.avoidDoubleFilterEvalForPythonUDF` (default `true`). With
`false`, a Python UDF is only as expensive as its arguments, as in Spark 4.3
and before.
- adds it to `OffsetSeqMetadata.relevantSQLConfs`, with `false` in
`relevantSQLConfDefaultValues` for checkpoints without the key. Every released
version treats Python UDFs as cheap, so a missing key has one meaning and a
static default is enough.
- checks it in `PushPredicateThroughNonJoin.isExpensive` rather than in
`PythonFuncExpression.expensive`, because a parent expression caches
`hasExpensiveChild`. SPARK-60632 does the same for
`spark.sql.optimizer.regexPatternCostEstimateVersion`. With `false`, it
classifies a copy of the expression in which each Python UDF is replaced by
`PrettyPythonUDF`. That expression is not foldable either, and it is expensive
only through its arguments.
| Checkpoint written by | Value used |
|---|---|
| Spark 4.3.x or before | `false`, so the filter stays where those versions
put it |
| Spark with this change | the recorded value, `true` by default |
This is a separate config rather than a new version of
`spark.sql.optimizer.regexPatternCostEstimateVersion`. That config is about
LIKE and RLIKE patterns, and its versions are shared with `branch-4.2` and
`branch-4.3`, which do not have SPARK-59693.
### Why are the changes needed?
Without this change, a query that ran on Spark 4.3.0 fails or returns
different rows after an upgrade to the unreleased `master` or `branch-4.x`. No
config is changed in either case.
Issue 1: When only the filter uses the input column of the UDF, the state
schema of the stream-stream join differs between the two plans, so the restart
fails.
```python
from pyspark.sql.functions import col, udf
@udf("int")
def identity(v):
return v
left = spark.readStream.schema("k INT, lv INT").json(left_path)
right = spark.readStream.schema("k INT, rv INT").json(right_path)
df = (left.join(right, "k")
.select(identity(col("lv")).alias("u"), "rv")
.where(col("u") > 0)
.select("rv"))
# Run 1 on Spark 4.3.0 with the left rows (1, -1) and (2, 10).
# Run 2 on master from the same checkpoint with the right rows (1, 100) and
(2, 200).
# Issue 1: It should return [200], but it fails with
STATE_STORE_VALUE_SCHEMA_NOT_COMPATIBLE.
```
Issue 2: A filter over `withWatermark` stays above the watermark after the
restart. Rows that fail the filter then advance the watermark, so later rows
can be dropped as late.
```python
from pyspark.sql.functions import window
events = spark.readStream.schema("t TIMESTAMP, v INT").json(events_path)
df = (events.withWatermark("t", "10 seconds")
.select("t", identity(col("v")).alias("u"))
.where(col("u") > 0)
.groupBy(window("t", "10 seconds"))
.count())
# Run 1 on Spark 4.3.0 with (00:01:40, 1).
# Run 2 on master with (00:16:40, -1), then run 3 with (00:01:45, 1) and
(00:33:20, 1).
# Issue 2: It should return the window [00:01:40, 00:01:50) with count 2,
but it returns count 1.
```
Setting `spark.sql.optimizer.avoidDoubleFilterEval=false` works around both
issues. However, it also moves the filters on other expensive expressions, such
as Scala UDFs. In addition, since SPARK-60885 a failed first restart records
`true`, so setting `false` afterwards no longer recovers the query.
### Does this PR introduce _any_ user-facing change?
No, compared to the released versions. A query restarted from a checkpoint
of Spark [4.3.0](https://github.com/apache/spark/releases/tag/v4.3.0)
(2026-10-27) or before keeps the plan of that version.
Within the unreleased `master` and `branch-4.x`, yes:
- A query restarted from a checkpoint of Spark 4.3.x or before no longer
treats Python UDFs as expensive.
- A new query still treats them as expensive.
- A checkpoint written by a snapshot build after SPARK-59693 has no key, so
it now reads `false`.
### How was this patch tested?
New tests:
- `FilterPushdownSuite`: with `false`, a filter on a `pythonUDF(a) + 1`
alias is pushed below the `Project`, and a filter on a Python UDF over an
expensive RLIKE stays above it. The plan is optimized with the default first,
so the test also covers the cached `hasExpensiveChild`.
- `OffsetSeqLogSuite`: a checkpoint without the key reads `false`, and a new
checkpoint records `true`.
Existing tests: `catalyst/testOnly *FilterPushdownSuite` and `sql/testOnly
*OffsetSeqLogSuite`.
The test suite has no checkpoints written by released Spark versions, so the
issues were also checked manually with PySpark (file source,
`Trigger.AvailableNow`). Run 1 used the released Spark 4.3.0. Later runs used a
build of `master` without this PR and a build with it, from the same
checkpoint. `S1` is Issue 1 with the UDF result in the output, which should not
change.
| Query | 4.3.0 -> 4.3.0 | 4.3.0 -> master | 4.3.0 -> this PR | this PR ->
this PR |
|---|---|---|---|---|
| S1 | `[2,10,200]` | `[2,10,200]` | `[2,10,200]` | not run |
| Issue 1 | `[200]` | `STATE_STORE_VALUE_SCHEMA_NOT_COMPATIBLE` | `[200]` |
`[200]` |
| Issue 2 | count 2 | count 1 | count 2 | count 1 |
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Opus 5.5
--
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]
