peter-toth opened a new pull request, #60201:
URL: https://github.com/apache/spark/pull/60201
### Differences from #60176
This is the branch-4.0 backport of #60176. The description below is the
original one.
- The commit of the branch-4.2 backport (#60186) cherry-picks cleanly. Like
branch-4.2, branch-4.0 does not have SPARK-58310 (#57485), the check that skips
a creation side with a Python UDF, and its test. So the new check directly
follows the `aggregate` val in `injectBloomFilter`, and the new tests follow
the "simple expressions only" test. The code and the tests are otherwise the
same as on master. "Next to the existing Python UDF check" in the description
below does not hold on this branch.
- On branch-4.0, `InjectRuntimeFilterSuite`,
`BloomFilterAggregateQuerySuite` and the plan stability suites pass (347
tests), with no plan change. Without the fix, both new tests fail.
- `dev/lint-scala` passes.
### What changes were proposed in this pull request?
`InjectRuntimeFilter` no longer builds a runtime bloom filter from a
creation side whose filter or key is non-deterministic.
The check runs on the bloom filter's subquery after column pruning, next to
the existing Python UDF check. So a non-deterministic column that the bloom
filter does not read still allows the bloom filter.
Known limit: the check sees only the non-determinism that the plan shows. A
leaf can hide it when reading the leaf again returns other rows, for example:
- a cache whose plan is not repeatable, if part of the cache is computed
again;
- a V2 scan with a pushed limit, top-N or sample;
- a `LogicalRDD` that is not a materialized checkpoint, such as the
DataFrame of a `foreachBatch` micro-batch.
DPP has the same gap, see the comment on
`PartitionPruning.isCheaplyRecomputableMaterializedPlan`. Closing it needs a
repeatability check for each leaf type. Master has one only for some leaves,
and the 4.0 to 4.2 branches have none. #58635 adds such a check for its
`RUNTIME_FILTER` hint, so the default path can use it on master in a follow-up.
### Why are the changes needed?
The bloom filter's subquery evaluates the creation side a second time, apart
from the join. With a non-deterministic filter or key, the two evaluations can
keep different rows or give different keys. The bloom filter can then drop
application side rows that match a creation side row the join sees, and the
query loses rows.
Example from the ticket, where every `dim.k` is in `fact`, so `lost` must be
0:
```sql
CREATE TABLE dim USING parquet AS SELECT cast(id AS INT) k FROM range(200);
CREATE TABLE fact USING parquet AS SELECT cast(id % 200 AS INT) k, id v FROM
range(20000);
SET spark.sql.autoBroadcastJoinThreshold=-1;
SET
spark.sql.optimizer.runtime.bloomFilter.applicationSideScanSizeThreshold=1;
SELECT count(*) total, count_if(f.k IS NULL) lost
FROM (SELECT k FROM dim WHERE cast(reflect('java.lang.Math', 'random') AS
DOUBLE) < 0.5) d
LEFT JOIN fact f ON d.k = f.k;
```
Before this change, three runs gave `lost` = 47, 46 and 47. After it, `lost`
is 0.
### Does this PR introduce _any_ user-facing change?
Yes. A join whose bloom filter creation side has a non-deterministic filter
or key no longer loses rows. Such a join no longer gets a bloom filter built
from that side. That includes a sampling filter like `rand() < 0.1`, even
though its two scans usually keep the same rows.
### How was this patch tested?
New tests in `InjectRuntimeFilterSuite`:
- the bloom filters of three creation sides:
- a non-deterministic filter: no bloom filter;
- a non-deterministic key: no bloom filter;
- a non-deterministic column that the bloom filter does not read: the
bloom filter stays;
- the example above, with `checkAnswer` on `lost`, and a deterministic
creation side on the same tables that gets a bloom filter.
Before this change, the first two creation sides got a bloom filter, and the
example lost rows. Checking the creation side plan instead of the pruned
subquery fails the third creation side.
`InjectRuntimeFilterSuite`, `BloomFilterAggregateQuerySuite`, the DPP
suites, the runtime filter suites, `StorageFilterPushDownSuite` and the plan
stability suites pass, with no plan change.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: Claude Code with 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]
