[SPARK-58272][SQL] Enable runtime Bloom filters for materialized cached inputs

[sunchao]

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:

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 addresses the missed optimization. It revisits the unmerged cache-awareness proposed in #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:

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:

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:

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.

[sunchao]

cc @cloud-fan @viirya @peter-toth @dongjoon-hyun

[dongjoon-hyun]

The getClass.getName.startsWith("org.apache.spark.sql.catalyst.expressions.") check looks like a fragile allow-by-package heuristic. It rejects unknown expressions (safe direction), but it silently
accepts every current and future expression under that package whose deterministic flag is
true — including ones whose output is not actually repeatable across evaluations.

AesEncrypt is exactly such a case, and the fact that it needs an explicit carve-out in the first pattern shows the failure mode: nothing prevents the next AesEncrypt-like expression (deterministic-looking flag, runtime randomness or external state) from being added to the package without anyone updating this denylist.

[sunchao]

Good point—the package check is an intentional namespace trust boundary, not an independent proof of repeatability. For expressions whose class is in the Catalyst namespace, this code relies on the documented Expression.deterministic contract: an expression must return the same result for fixed child inputs, and expressions using mutable state or implicit inputs must report themselves as non-deterministic. It separately rejects CURRENT_LIKE, NonSQLExpression, and user-defined expressions.

AesEncrypt contains a known pre-existing exception when the IV is omitted (for example, default GCM): evaluation generates a fresh random IV even though the wrapper inherits deterministic from its children. This change rejects AesEncrypt in the analyzed plan before optimization replaces it with StaticInvoke; the optimized StaticInvoke is also rejected through NonSQLExpression. Commit e783e16ff6a adds a source comment clarifying this trust boundary.

I did a targeted scan of current built-ins involving randomness, current time, TaskContext, reflection, or runtime replacement and did not find a second current expression that passes these checks while producing non-repeatable output, though that is not an absolute proof against future additions. An exact-class allowlist would duplicate the evolving Catalyst expression set and disable safe expressions by default, so I kept deterministic as the core contract. Do you know of another concrete current case we should cover here?

[dongjoon-hyun]

Shall we split the cache-materialization bookkeeping changes into a separate PR, @sunchao ?

The fullyConsumed gate in CachedRDDBuilder.buildBuffers and the per-generation
accumulator isolation in clearCache are standalone correctness fixes that affect
all cached relations, not just the new Bloom-filter path:

• Without the fullyConsumed gate, a task that recomputes a partition (e.g. after a failed block store) but only partially consumes its iterator can overwrite a completed partition's statistics with partial values (last-write-wins), which AQE then consumes.
• Without generation isolation, late accumulator updates from tasks of a previous cache generation can make a rebuilt cache appear fully materialized.

Since SPARK-57547 is already in Spark 4.2.0, these two fixes stand on their own and could be reviewed (and, if needed, backported) independently of the optimizer feature. Keeping this PR focused on the runtime-filter change would also make it easier to reason about.

[sunchao]

Thanks @dongjoon-hyun ! That is a good idea. I've split the cache-materialization related fixes into #57617. Please take a look!

[cloud-fan]

0 blocking, 1 non-blocking, 0 nits.
The safety and correctness gates are well covered; one non-blocking optimizer-efficiency cleanup remains.

Suggestions (1)

• sql/core/src/main/scala/org/apache/spark/sql/execution/columnar/InMemoryRelation.scala:658: Read the materialized cache metadata once per eligibility check. statsAvailable and isOutputRepeatable each call repeatableMaterializedStats, so every candidate cache folds all partition statistics twice before leaf.stats.rowCount may perform another fold. Exposing one snapshot containing durability, repeatability, row count, and byte size would keep optimizer work linear with a single pass and ensure all gates use the same observation. -- see inline

Verification

I traced eligibility from InjectRuntimeFilter through MaterializedLeafNode into CachedRDDBuilder: a materialized creation side must have a complete generation snapshot, repeatable logical and physical lineage, disk-backed storage, binary-stable join keys, and either selectivity or a favorable distinct-count comparison. The injected predicate uses the existing Bloom-filter aggregate and does not alter join output or schema. I also checked that accumulator completeness and folding share one synchronized critical section and that cache rebuilds replace the generation accumulator and reset strict-read state.

[cloud-fan]

Could we expose one materialized-metadata snapshot and reuse it for all eligibility checks? statsAvailable and isOutputRepeatable each fold every partition's statistics, and leaf.stats.rowCount may fold them again, so one optimizer check can scan the partition map three times.

[sunchao]

Thanks @cloud-fan — addressed in 9fafde7. MaterializedLeafMetadata now captures row count, byte size, durability, and repeatability from one synchronized cache-generation snapshot, and InjectRuntimeFilter reads that snapshot once per eligible cache. A cheap precheck skips memory-only and non-repeatable caches without folding their partition statistics. The selected creation-side size is reused by both threshold checks while preserving projected-plan size estimates. I also added coverage for cold/partially materialized caches, memory-only caches, projected thresholds, cache rebuilds, and partition recomputation. All 179 focused tests and the Catalyst/SQL Scala style checks pass.