sunchao opened a new pull request, #57443:
URL: https://github.com/apache/spark/pull/57443
### Why are the changes needed?
Spark's runtime Bloom-filter rule normally recognizes a small join input by
finding a selective predicate in its logical plan. Persisting that input
replaces the predicate with an `InMemoryRelation`, even after the cache has
been fully materialized. The selected keys are still available cheaply, but
Spark no longer recognizes them as a useful filter source and may shuffle the
entire other side of the join.
For example, these two joins represent the same selected keys:
```scala
val selected = requests.filter($"tenant_id" === 42).select("request_id")
events.join(selected, "request_id")
val cached = selected.persist(StorageLevel.MEMORY_AND_DISK)
cached.count()
events.join(cached, "request_id")
```
The uncached join can receive a runtime Bloom filter. The cached join
currently cannot because the optimizer sees the cache rather than the selective
predicate that produced it. The result is especially wasteful when the event
side is large and most of its keys do not appear in the already-materialized
cache.
There is also no useful default interval for a cached side that is too large
to broadcast: both `spark.sql.autoBroadcastJoinThreshold` and
`spark.sql.optimizer.runtime.bloomFilter.creationSideThreshold` default to 10
MB. A 40 MB cached key set can therefore force a shuffle while being rejected
as a Bloom-filter input, even though scanning the materialized cache does not
recompute its original query.
[SPARK-58272](https://issues.apache.org/jira/browse/SPARK-58272) addresses
the missed optimization. It revisits the unmerged cache-awareness proposed in
[#39377](https://github.com/apache/spark/pull/39377), while adding explicit
safeguards for cache materialization, replayability, and filtering benefit.
### What changes were proposed in this PR?
This change lets the existing runtime Bloom-filter rule recognize a fully
materialized, disk-backed cached relation as a safe source of join keys. The
optimizer reads the cache's exact row count and byte size and builds the Bloom
filter from cached output, while retaining its existing join-shape,
key-lineage, application-size, and filter-count checks. The resulting join
remains an ordinary Spark join; applications do not need a specialized helper
or a rewritten query.
A cache is eligible only when every partition of its current generation has
completed and its logical and physical inputs can be replayed without changing
their rows. Cache bookkeeping is generation-specific and publishes its loaded
state and partition statistics atomically; incomplete iterator consumption and
late completions from an earlier cache generation cannot make a rebuilt cache
appear complete. File-backed inputs are restricted to trusted built-in readers,
and their actual instantiated file-scan RDDs must be strict about missing and
corrupt files. Memory-only caches, partially materialized caches, opaque user
functions, nondeterministic or runtime-replaceable expressions, and best-effort
or unknown readers are excluded.
Materialization alone is not treated as proof that a Bloom filter will help.
The optimizer requires either a selective predicate in the cache's original
plan or application-side join-key statistics showing that the cache's exact row
count is smaller than the application's distinct key count. Statistics are
traced back through aliases and projections to the originating scan. This
mirrors the profitability distinction already used for materialized-input
dynamic partition pruning.
Safely materialized caches use the new
`spark.sql.optimizer.runtime.bloomFilter.materializedCreationSideThreshold`
setting, which defaults to 100 MB. The existing 10 MB creation-side limit
remains unchanged for all other inputs.
### How was this PR tested?
Focused existing and new optimizer/cache suites were run with:
```bash
build/sbt "sql/testOnly \
org.apache.spark.sql.InjectRuntimeFilterSuite \
org.apache.spark.sql.execution.columnar.InMemoryColumnarQuerySuite \
org.apache.spark.sql.execution.columnar.ConcurrentInMemoryRelationSuite \
org.apache.spark.sql.util.PartitionKeyedAccumulatorSuite"
```
The coverage exercises selective and statistics-proven cached joins,
projected join keys, size thresholds, join semantics, memory-only and partially
loaded caches, unsafe expressions, random-IV encryption, permissive and
preinitialized file readers, stale cache generations, duplicate partition
completions, and partially consumed cache iterators.
The standalone benchmark compares the same sort-merge join with runtime
Bloom filtering disabled and enabled, verifies identical query results, and
reports actual wide-side shuffle bytes:
```bash
build/sbt "sql/Test/runMain \
org.apache.spark.sql.execution.benchmark.RuntimeBloomFilterCachedInputBenchmark"
```
On 500,000 deterministic fact rows and a cached 1%-selective key set, the
measured results were:
```text
Fact-side shuffle without the runtime Bloom filter: 33,679,423 bytes
Fact-side shuffle with the runtime Bloom filter: 344,110 bytes
Fact-side shuffle reduction: 98.98%
Best query time without the runtime Bloom filter: 333 ms
Best query time with the runtime Bloom filter: 178 ms
```
The four existing affected suites passed all **67 tests**. Two additional
existing cache-lifecycle regressions passed, covering accumulator cleanup after
uncaching and materialization-bookkeeping reset after `clearCache`. Scala
source and test style checks also passed for the Catalyst and SQL modules.
### Does this PR introduce _any_ user-facing change?
Yes. Eligible joins against safely materialized cached relations can now
receive a runtime Bloom filter automatically and shuffle fewer rows. A new SQL
configuration controls the maximum size of those cached filter inputs; all
existing configuration defaults and public APIs retain their prior behavior.
### Was this patch authored or co-authored using generative AI tooling?
Generated-by: OpenAI Codex.
--
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]
