Module runtime_filter_exec 

Cross-stage runtime filter: the two plan nodes that carry a bloom filter from a join’s build side into the probe side’s map stage.

Why this exists at all

DataFusion builds a dynamic filter from a join’s build side and pushes it into the probe side’s scan. That works single-node because both live in one plan. We cut the plan at every exchange, so the probe’s scan runs in its own stage before the join stage exists — there is no build side to learn from, and every SF100 plan dump shows the placeholder unfilled:

lineitem ... predicate=l_returnflag = R AND DynamicFilter [ empty ]

The cost is not the scan, it is the shuffle. TPC-H q10 hash-partitions ALL ~150M returned lineitem rows across a pod network measured at ~11 MiB/s when the orders side selects one quarter (~3.5%) and only ~7M of those rows can possibly join. Dropping the other 95% before the shuffle write removes the bytes from the wire, which is the binding constraint.

The shape

  stage B (build side)          stage F (filter)         stage P (probe)
  ────────────────────          ────────────────         ───────────────
  scan orders                   scan orders              scan lineitem
  filter o_orderdate            filter o_orderdate       filter l_returnflag
  └─> shuffle(o_orderkey)       project o_orderkey       └─> RuntimeFilterProbeExec
                                coalesce -> 1 task            ├── data
                                RuntimeFilterBuildExec        └── ShuffleReadExec(F)
                                └─> shuffle(keyless, 1)   └─> shuffle(l_orderkey)

Stage F is a clone of stage B’s subtree, not a read of its output: the build side is small by construction (the planner’s selectivity gate refuses otherwise) and re-scanning it from local disk at ~300 MB/s beats re-reading its shuffle output across a ~11 MiB/s pod network. It is also what Spark’s InjectRuntimeFilter does.

Coalescing F to a single task is deliberate. The filter is a broadcast: every task of stage P must fetch the whole thing. One task producing one ~6.5 MB blob costs P × 6.5 MB; N tasks producing N partials costs P × N × 6.5 MB, which at q10’s shape is 2 GB of wire to save a shuffle — the fix paying for itself many times over in the wrong direction.

Why the filter travels as ordinary shuffle data

It is a one-row RecordBatch with a single Binary column, written through the same keyless single-partition gather the stage cutter already emits for ungrouped aggregates, and read back through an ordinary [ShuffleReadExec]. So there is no new RPC, no new store key, no new transport, and the stage DAG edge P → F is an edge the scheduler already knows how to order and cycle-check.

Correctness

A bloom filter has false positives but never false negatives, so a row that can join is never dropped and the join’s output is unchanged. Everything here is built to keep that one property true: see krishiv_shuffle::RuntimeFilter for the encoding, union and fail-open rules, and runtime_filter_candidates in distributed_plan for the join shapes this may fire on (inner only).

Structs

RuntimeFilterBuildExec

Consumes its input and emits ONE row: the serialized bloom filter of a single key column.

RuntimeFilterProbeExec

Drops input rows whose key is provably absent from the build side.

Constants

FILTER_COLUMN

Column name of the single Binary column a filter stage emits.

RUNTIME_FILTER_ENV

Env flag gating the whole feature. Off by default.

Functions

enabled

Is cross-stage runtime filtering enabled?

filter_schema

Schema of a filter stage’s output: one row, one serialized bloom.