Adaptive Execution: Runtime Filters and Mid-Query Replanning

A plan is a set of decisions made before any row is read: which side of a join to build, whether to repartition, how much memory each operator gets. Those decisions rest on estimates, and estimates are wrong often enough that a plan can be correct and still be an order of magnitude slower than necessary. Adaptive execution is the family of techniques that let a running plan observe reality and adjust, either by changing an operator’s behaviour or by changing the plan’s shape.

Two approaches are worth distinguishing. Runtime filters change what work is done without changing the plan: a filter built from one side of a join is applied to the other side’s scan, so rows that cannot match are never read. Mid-query replanning changes the plan itself: after a blocking operator finishes, the actual cardinality is known, and the remaining plan can be re-optimized with that knowledge.

Statistics are wrong in one direction

Estimates are derived from per-column statistics — distinct value counts, histograms, null fractions — combined under an assumption of independence. That assumption fails for correlated columns: a predicate on city and a predicate on postal_code are nearly redundant, so multiplying their selectivities underestimates the result by orders of magnitude. It fails for joins whose keys are correlated with filters, and it fails for expressions the statistics do not cover, such as a function applied to a column.

The error distribution is not symmetric. Independence-based estimates for conjunctive predicates with positive correlation underestimate; estimates for a join whose key is filtered on the build side underestimate the build size and therefore the memory need. Underestimating a build side causes a hash join to spill; overestimating it causes the planner to avoid the hash join and choose something slower but bounded. Systems accept the second failure and try to detect the first.

A single number captures the problem: an estimate of a thousand rows against an actual of ten million turns a linear operator into one that spills and re-reads its input several times. The plan text does not change, the query does not error, and the only visible symptom is time.

Bloom filters as a join-side prefilter

A bloom filter is a bit array with k hash functions. Inserting a value sets k bits; testing a value reports “possibly present” or “definitely absent”. False positives are possible at a rate set by the array size and the number of inserted values; false negatives are not. The standard sizing gives a false-positive rate near one percent with about ten bits per element and seven hash functions, and the rate falls exponentially as the array grows relative to the element count.

For a join, the build side can produce a bloom filter over its join keys, and the probe side can test each row before doing anything expensive. A row that fails the test cannot have a match, so it can be discarded before the hash lookup, before the network hop in a parallel plan, or before the heap fetch in an index scan. The saving is largest when the build side is highly selective: a filter built from a thousand keys rejects almost everything.

The filter is most valuable when it can be pushed down to the scan. In a parallel plan, the filter is built by one fragment and shipped to the fragments that scan the other table; each scan tests rows before sending them across the exchange. This turns a filter that would have rejected rows after a network transfer into one that rejects them before it, and the transfer cost is often the dominant term.

Two constraints limit the technique. The filter must be complete before the probe side starts, which forces a synchronization point: the build side must finish before any probe-side row is tested. That serializes the two inputs at the filter’s creation, which can cost more than it saves if the build side is slow. And the filter’s false positives are silent: a query with a filter is correct, but its performance depends on a probabilistic structure whose error rate is a function of the actual key distribution, which may differ from the estimate used to size it.

-- the filter is not visible in SQL; it appears in the plan as an extra
-- predicate on the scan of the probe-side table
SELECT o.id, c.name
FROM orders o JOIN customers c ON c.id = o.customer_id
WHERE c.region = 'north';
-- plan: bloom filter built from the ~few-thousand matching customers,
-- applied to the orders scan before the join probe

Dynamic partition pruning

When a table is physically partitioned, the partitioning column’s predicate can eliminate whole partitions before they are opened. Static pruning uses the predicate’s constants; dynamic pruning uses values produced at run time by the other side of a join.

The mechanism is a dependency: the scan of the partitioned table waits for the other input to produce its distinct partitioning-key values, then opens only the partitions whose ranges intersect those values. In a star-schema query joining a large fact table partitioned by date to a small dimension filtered to a few dates, the fact scan can skip every partition outside the selected range. The saving is proportional to the fraction of partitions excluded, and it applies to I/O, to the number of files opened, and to the metadata work of planning each partition.

The dependency is a scheduling constraint. The scan cannot start until the dimension side has produced its keys, so the two inputs are no longer independent and the plan’s critical path lengthens. Systems express this with a special exchange or a “wait for” edge in the plan, and the benefit only materializes if the excluded partitions are numerous enough to outweigh the lost overlap.

A variant applies to a partitioned join key rather than to a filter: if both sides are partitioned on the join key with compatible schemes, each partition pair can be joined independently, with no exchange and no global shuffle. This partition-wise join is the largest single win available from physical design, and it requires that the partitioning be declared in the catalog rather than inferred.

Replanning after a blocking operator

A blocking operator — a sort, a hash build, an aggregation — completes before its parent starts. At that moment the actual row count is known, and it may differ from the estimate by orders of magnitude. Replanning uses the actual count to re-cost the remainder of the plan and, if the new cost is better, replace the remaining subtree.

The classic case is a join whose inputs are themselves joins. The first join completes with far more rows than estimated; the second join’s build side is now known to be too large for a hash join, and a merge join over sorted input becomes the better choice. Without replanning the plan spills; with it the plan switches algorithms mid-flight.

Implementations differ in how much they can change. A conservative form re-costs only the choice of algorithm at a single join, keeping the tree shape. An aggressive form re-optimizes the whole remaining subtree, which can change join order, exchange placement, and parallelism. The aggressive form needs the intermediate result to be materialized in a form the new plan can consume, which means the boundary must be a blocking operator whose output is a stable, re-readable relation.

The cost of replanning is the planning time itself, paid in the middle of execution, plus the risk of oscillation: a plan that replans after every blocking operator can thrash if the estimates remain wrong. Implementations cap the number of replans and require a cost improvement threshold before switching.

What adaptivity cannot fix

Adaptivity observes cardinalities, not distributions. A join with a single hot key has the right total row count and a skewed distribution that defeats any partition-based algorithm; replanning on the count does not detect the skew. Handling it requires observing the key frequencies, which is a different measurement and a different mechanism.

Adaptivity also cannot undo an early decision. If the first operator in the plan chose the wrong access path, the rows are already read by the time any feedback exists. The techniques that help there are the ones that make the early decision cheap to revise: a scan that can switch from index to sequential mid-stream, or a filter that is applied before the expensive access. Both exist in some systems and neither is universal.

The honest summary is that adaptive execution narrows the gap between the best plan and the chosen plan, but it does not close it, and it adds a scheduling dependency and a planning cost of its own. The estimates it reacts to remain the foundation, which is why the cost model and the statistics it consumes are the subject of a separate treatment in the query optimization material. Within execution, the interaction that matters most is with parallelism: replanning a parallel plan must also decide the degree of parallelism for the new subtree, and a change in worker count invalidates any partitioning the previous plan established, as described in parallel execution and exchange operators.

The operator-level view

Set against the fixed operators described elsewhere in this cluster, adaptivity is a thin layer of decisions layered over them. A hash join that discovers its build side is too large can fall back to a partitioned variant; a sort-merge join that discovers its inputs are already ordered can skip its sorts; a scan that discovers its predicate is more selective than estimated can switch to an index. Each of these is a local decision made possible by an observation, and the aggregate effect is a plan that behaves better than its estimates.