[SPARK-59345][SQL] Add RUNTIME_FILTER join hint

[pan3793]

What changes were proposed in this pull request?

Add a RUNTIME_FILTER(relation) join hint. It names a join side as the runtime filter source:
Spark builds a runtime filter from that side and prunes the other side of the join.

• One hint for both runtime filter kinds, and not a join strategy. HintInfo gains a
  runtimeFilterSource facet next to strategy, resolved with the same relation matching as
  join strategy hints, so it composes with BROADCAST, MERGE, etc. on the same relation.
  Spark picks the mechanism: partition pruning when the pruned side's join key allows it, a
  Bloom filter otherwise.
• The hint overrides cost estimates, not correctness. It waives the selective-predicate search,
  the size thresholds, the lineage and shuffle requirements, and DPP's benefit estimate and
  reuseBroadcastOnly gate. It keeps the join-type and simple-key requirements, requires the
  hinted side to be a repeatable source (JoinSelectionHelper.isRepeatableRuntimeFilterSource),
  and leaves the filter count and Bloom filter size limits in place. A hinted side is never
  itself pruned.
• The hint is never silently dropped. InjectRuntimeFilter warns with the reason whenever it is
  not applied, including when Bloom filters are disabled. Hinting both sides is ambiguous: Spark
  warns and falls back to the heuristics.

This supersedes SPARK-32842 (#29709, a DPP-only hint) and SPARK-42064 (#39571,
BLOOM_FILTER_JOIN as a JoinStrategyHint).

Why are the changes needed?

Spark's runtime filter heuristics depend on statistics and a small set of recognized plan shapes.
When statistics are missing or misleading, or the filter source is an aggregate, a subquery, or
a join, no runtime filter is built even though the user knows the pruning would pay off. Earlier
attempts to relax the DPP cost model automatically (SPARK-34884) were reverted because the
estimates are unreliable, and reuseBroadcastOnly applies to every join in the session. A hint
makes the choice explicit for one join, the same way join strategy hints override join selection.

Does this PR introduce any user-facing change?

Yes. A new SQL hint RUNTIME_FILTER and the corresponding Dataset.hint("runtime_filter", ...)
name are recognized. Previously the name was reported as an unrecognized hint and ignored. The
hint parameter error message now reads "Join hint parameter ..." instead of "Join strategy hint
parameter ...".

How was this patch tested?

New tests in InjectRuntimeFilterSuite (which heuristics the hint waives, which requirements it
keeps, DPP deduplication and the non-pushable DPP case, composition with strategy hints,
ambiguous and unapplicable hints with their warning text, subquery and parameter-less forms),
DynamicPartitionPruningSuite (standalone DPP subquery under the hint, broadcast reuse,
partition-key requirement, both-sides hint), and ResolveHintsSuite (resolution and merging).
Existing JoinHintSuite and hint-related suites pass.

Was this patch authored or co-authored using generative AI tooling?

Generated-by: Claude Fable 5.1

[pan3793]

cc @wangyum @sunchao @ulysses-you @cloud-fan, looking forward to your thoughts on this idea.

[pan3793]

the name becomes inaccurate, not sure if it's worth renaming, maybe ResolveJoinHints?

[cloud-fan]

Review summary

Independent runtime-filter source execution is existing behavior, and DPP broadcast reuse is already configuration-dependent; this PR does not introduce that execution model. The new correctness gap is that the hint admits complete order-sensitive aggregate and tied top-N sources that the unhinted heuristics do not select. Catalyst plan determinism does not make their key sets stable, so filter construction and the join can choose different keys and the filter can remove a row that should match. This wrong-result path should be closed before merge by rejecting hinted sources whose selected key set is not stable.

Separately, a deterministic operator can block DPP pushdown even though the new survival proxy credits the hint. Cleanup then removes the DPP after Bloom fallback and warning have already been suppressed. The decision should use actual scan reachability, with a final-plan regression for a deterministic barrier.

Findings

2 total: 0 P0, 1 P1, 1 P2, 0 P3.

Blocking (P1)

• Reject order-sensitive runtime-filter sources — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/joins.scala:625 — see inline.

Non-blocking (P2)

• Do not credit DPP before proving it reaches the scan — sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/optimizer/InjectRuntimeFilter.scala:521 — see inline.

Verification

• The P1 path is PR-specific because the hint newly admits complete order-sensitive sources into a pre-existing independently evaluated runtime-filter path; both Bloom and DPP consumers can then evaluate that admitted source separately from the join.
• InjectRuntimeFilter runs before predicate pushdown and cleanup, so a deterministic pushdown barrier can invalidate the DPP survival proxy after the hint is credited.

[cloud-fan]

Blocking (P1): Independent runtime-filter source execution already exists, and DPP broadcast reuse is configuration-dependent; the new risk here is that this hint admits the complete source plan after only this check. plan.deterministic covers expression determinism, so it still accepts order-sensitive first/last/any_value aggregates and top-N sources with ties. Separate filter and join evaluations can then choose different keys, causing the filter to discard a row the join would match. Please reject row-order-dependent hinted sources unless their selected key set is proven stable, and cover both an order-sensitive aggregate and a tied top-N case.

Recommended change: Conservatively reject hinted source plans whose output depends on input order, including order-sensitive aggregates and LIMIT/OFFSET whose ordering is not proven total.

Why this works: Extend isRepeatableRuntimeFilterSource to recognize these logical and aggregate shapes and return false, allowing the existing not-applied warning path to handle them.

Scope: JoinSelectionHelper's repeatability predicate and focused runtime-filter tests for both Bloom-filter and DPP consumers.

Compatibility: This only narrows acceptance of hint shapes that are currently unsafe; stable sources retain existing behavior, while rejected sources keep correct join results and receive the existing warning.

Risks: A conservative predicate may decline some sources that happen to be stable at runtime. Aggregate aliases and equivalent top-N plan shapes must be classified consistently.

Constraints: Do not treat Expression.deterministic or user responsibility for sort ties as a row-set repeatability proof. Apply the same source eligibility decision to both Bloom-filter and DPP paths.

Success: No accepted source can yield a join-key set in the join evaluation that was absent from the independent filter evaluation, with regressions covering first/last/any_value and tied top-N inputs.

[cloud-fan]

Non-blocking (P2): Determinism is not sufficient to prove this DPP survives. For example, a deterministic Window whose partition columns exclude the join key blocks predicate pushdown; CleanupDynamicPruningFilters then replaces the stranded DPP with true after this branch has already credited the hint. The final plan has neither DPP nor the Bloom fallback, and no warning is emitted. Please use the same scan-reachability criterion as DPP cleanup before setting appliedHint; otherwise continue to the Bloom-filter path, with a deterministic barrier regression test that inspects the final optimized plan.

Recommended change: Credit an existing DPP predicate only when it can reach a supported scan; otherwise continue through the existing Bloom-filter fallback and warning logic.

Why this works: Reuse the scan-reachability condition enforced by DPP pushdown and cleanup instead of application-subtree determinism when computing prunedByDpp.

Scope: InjectRuntimeFilter's DPP decision plus a focused InjectRuntimeFilterSuite regression using a deterministic pushdown barrier.

Compatibility: Plans whose DPP already survives are unchanged; plans where cleanup would remove DPP now receive a Bloom filter when eligible or the documented not-applied warning.

Risks: Duplicating scan-reachability logic could drift from CleanupDynamicPruningFilters. The fallback may add a Bloom filter in plans that previously ended with no runtime filter.

Constraints: Keep the survival check aligned with the actual pushdown and cleanup rules. Preserve the existing Bloom-filter count, key-shape, enablement, and warning gates.

Success: Every credited hinted DPP remains in the final optimized plan; when it cannot survive, the final plan has the Bloom fallback or the user receives the documented reason-bearing warning.