[Feature Request][MPP] Join runtime filters for the analytics engine

Is your feature request related to a problem? Please describe

When the analytics engine distributes a join as a hash shuffle, both inputs are hash-partitioned and shipped to a worker tier in full, even when most probe-side rows have no match on the build side. Nothing prunes them: the probe producer has no way to know which keys the build side holds, because the two sides are separate Substrait fragments dispatched together to different nodes. On a five-way join over a 600M-row fact table the dominant cost is shipping rows that the join then discards.

DataFusion's native dynamic filter cannot close this gap, and not because it works poorly, it is structurally inapplicable. It works by having HashJoinExec and the leaf scan share one Arc<DynamicFilterPhysicalExpr> that the join updates once its build side is collected, which requires the join and the scan to be in the same plan, in the same process, with the value arriving during execution. A shuffle join satisfies none of the three. It does apply to broadcast joins, where the build side is registered as an in-process memtable, so the gap is specifically the shuffle family.

Describe the solution you'd like

Produce the filter from an independent build-side aggregation rather than from the join operator, so it does not need the join and the scan to share a process. Before the main dispatch, run a small pre-pass over the build side that aggregates its join key into one fixed-size Bloom filter per shard; union those on the coordinator; then deliver the result to the probe-side producer as an execution instruction while the plan itself carries only an integer filter id. The probe producer tests each row against the filter and drops non-matching rows before they enter the shuffle, which is where the saving is, this reduces shuffle and join volume, not bytes read.

The engine already owns most of the required parts: a plan-carries-id / value-delivered-separately convention used today for predicate delegation, a coordinator-side phase that runs an arbitrary stage subtree in isolation and captures its output, and a shard-level pre-dispatch pruning phase. The genuinely new piece is a serializable payload with a well-defined union, for which the parquet split-block Bloom filter is reused so the build and probe sides cannot disagree on hash function or block layout.

Correctness rules, all of which refuse rather than guess: never filter a side the join preserves (outer, semi, anti), never move a predicate below an aggregate, a limit, a union, or a non-inner join, and never apply a filter whose key column name resolves to more than one scan. A false positive costs one row through the join; a false negative would lose a result, so every union refuses a size mismatch instead of truncating.

Related component

Search:Performance

Describe alternatives you've considered

Extending DataFusion's native dynamic filter across the process boundary. Structurally impossible for shuffle joins, per above. Two further limits make it unattractive even where it applies: only the hash join implements the pushdown hook, so the largest joins — deliberately planned as sort-merge — produce no filter at all; and a serialized dynamic filter loses its hash-table lookup and degrades to min/max above roughly 150 distinct values.

Pre-dispatch shard elimination for wide keys. Attempted and disproven, which is worth recording because it looks obviously right. A shard-level check cannot query a Bloom filter, because a Bloom answers "is this key present" and cannot be enumerated; the only pre-scan enumerator is the column's dictionary. Measured on real files, the high-cardinality join keys are dictionary-encoded in 0 of 3 row groups while the low-cardinality columns are in 3 of 3 — parquet stops dictionary-encoding exactly the columns a join filter targets. With no enumerable source, a value set unbounded above half a million distinct keys, and a min/max envelope measured too broad to prune, there is no payload a pre-dispatch evaluator can use. Shard elimination therefore remains available only for narrow key sets.

Mid-flight delivery to already-running shard tasks. Needs a new transport action and a new native entry point to mutate a filter on a live session, in exchange for the weakest application point — too late for pre-dispatch pruning, and nothing for joins planned as sort-merge.

Additional context

Measured on a 8-node cluster at TPC-H sf=100, alternating on/off/on/off arms so cache warmth cannot explain the result, with results identical across every run:

probe side	filter on	filter off
600M-row fact table, filtered by a dimension carrying a local predicate and joined to it directly	7.5 / 6.8 s	27.9 / 29.7 s	+75%
80M-row table, same shape	5.0–5.5 s	6.8–6.9 s	+24%
fact table one join further from the locally-filtered dimension	23.9 / 24.0 s	24.3 / 24.7 s	inside noise
no local predicate on any dimension; selectivity only in a three-table chain	48.4–49.5 s	48.1–50.4 s	none