Why architecture matters here
The architecture matters because the alternative — scanning the whole fact table and filtering after the join — scales with the size of the fact table, which is the one thing you cannot control and the one thing that keeps growing. A predicate that selects 0.1% of customers still forces a full read of a table that may be tens of terabytes. Runtime filters change the scaling law: the cost becomes proportional to the number of fact rows that survive the filter, not the number that exist. For a selective star query that is the difference between a query that returns in seconds and one that saturates the cluster for minutes.
It matters, too, because the savings compound with the storage format. Columnar formats like Parquet and ORC store data in row groups, each carrying min/max statistics per column. A min/max runtime filter can eliminate an entire row group by comparing the filter's range against the group's statistics — no decoding required. A Bloom filter goes finer, rejecting individual rows during the scan. Because the scanner is the component closest to the raw bytes, a filter applied there prevents I/O, decompression, and decoding all at once. Push the same predicate later in the plan and you have already paid those costs.
The trade-off is that runtime filters are speculative. Building and broadcasting a filter has a cost, and a filter that rejects few rows — because the join is not selective, or the fact keys are uniformly spread across the whole domain — buys nothing while still consuming memory and network. The architecture therefore has to be adaptive: produce filters cheaply, apply them where they help, and disable those that prove ineffective so they do not tax an already-tight query. A good runtime-filter implementation is as much about knowing when not to filter as about filtering.
Finally, this is an architecture that only pays off in a massively parallel, shared-nothing engine. Impala runs a query as a tree of fragments spread across many daemons, each scanning a slice of the data. A runtime filter is a small message that lets one part of that tree teach another part what to skip — a form of cross-fragment cooperation that would be pointless in a single-node engine but is transformative when hundreds of scan threads would otherwise each read their slice blind. Understanding the filter means understanding it as a distributed communication pattern, not just a predicate.
The architecture: every piece explained
A runtime filter has two roles in the plan: a producer and one or more consumers. The producer sits on the build side of a hash join — the side whose rows populate the hash table, conventionally the smaller, more filtered input. As the build completes, the producer walks the join keys it has collected and constructs a filter data structure. The consumer sits on a scan node feeding the probe side of the join — the large fact table. The plan wires the two together with a filter ID, and the runtime carries the finished filter from producer to consumer.
Impala builds three kinds of filter. A Bloom filter is a bit array with several hash functions; it answers 'is this key possibly in the set?' with no false negatives and a tunable false-positive rate. It is the general-purpose choice, good for high-cardinality keys. A min/max filter records only the smallest and largest key value on the build side; it is nearly free to build and lets scanners prune whole Parquet row groups by comparing against stored statistics, but it only prunes when the surviving keys cluster in a range. An IN-list filter holds the exact set of build-side keys when that set is small; it gives perfect selectivity with zero false positives, but is only viable below a cardinality threshold.
Routing depends on the join distribution mode. In a broadcast join the small build side is replicated to every node, so each node builds a complete local filter and applies it to its own scan — no cross-node aggregation needed. In a partitioned (shuffle) join the build side is split across nodes by hash of the key, so each node holds only part of the key set; the partial filters must be aggregated at the coordinator (or a designated node) into a global filter before being broadcast back to all scanners. That aggregation step adds latency, which is why partitioned-join filters have a longer natural wait time than broadcast ones.
The consumer side has two application points. At row-group granularity, before decoding, the scanner compares the filter against Parquet/ORC column statistics and dictionary, discarding groups that cannot match — this is the cheapest and highest-leverage form. At row granularity, during the scan, each surviving row's key is probed against the Bloom or IN-list filter and rejected individually. The scanner also needs a policy for the moment before the filter arrives: it can briefly wait, controlled by RUNTIME_FILTER_WAIT_TIME_MS, and if the filter has not appeared it proceeds unfiltered rather than stall the whole query.
End-to-end flow
Walk a selective star query end to end. The planner identifies the hash join between sales and the filtered customers, marks the customers side as the filter producer, and attaches a consumer to the sales scan, tied together by a filter ID. Execution begins: scan fragments for both tables start, and the customers scan applies its country = 'IN' predicate, emitting only Indian customer rows.
Those rows flow into the hash-join build. As the hash table fills, the producer accumulates the distinct cust_id values and, at build completion, finalizes the filter — say a Bloom filter sized for the observed cardinality. In a broadcast join each node has done this locally; in a partitioned join the partial filters are shipped to the coordinator, OR-ed together into a global Bloom filter (or unioned min/max ranges), and the result is published back out.
Meanwhile the sales scanners have started reading their slices. Each consumer checks whether its filter has arrived. If yes, it begins skipping: row groups whose min/max cannot overlap the filter's range are dropped wholesale, and surviving rows are probed against the Bloom filter, with only possible matches passed upward. If the filter has not yet arrived, the scanner waits up to the configured timeout; on timeout it proceeds unfiltered so the query cannot deadlock waiting on a slow producer.
The surviving fact rows — now a tiny fraction of the original — flow into the join probe, match against the hash table, and feed the aggregation. The query profile records, per filter, how many rows were considered and how many were rejected, plus the wait time each scanner incurred. That is the feedback loop: an operator reading the profile can see whether a filter rejected 95% of rows (excellent) or 3% (dead weight worth disabling), and whether scanners spent their time productively or blocked on a filter that arrived too late to matter.