DataFusion's Dynamic Filters: Letting the Producer Tell the Consumer What It Learned at Runtime

TL;DR: A query planner works with facts it knows at plan time. But a HashJoinExec only learns the range of its build side after it has read the build side, and a TopK only learns its threshold after it has filled its heap. DataFusion lets those operators publish that fact into a shared DynamicFilterPhysicalExpr — a cell containing a physical expression that can be replaced at runtime. The consumer (a Parquet scan, a file pruner) subscribes to it and re-prunes as the value tightens. The interesting parts are the concurrency: a generation counter, a tokio::sync::watch channel for the async waiters, an atomic-load fast path for the synchronous pollers, and a per-generation cache so the per-batch hot path does not re-walk the expression tree.

Static predicate pushdown is the oldest trick in query engines: if the predicate is t.d > 2024-01-01, and the file’s minimum t.d is 2023-06-01, the file cannot contain a matching row, so skip it.

But a static predicate is a promise about the data made before the query runs. Plenty of the most valuable facts are only learnable at runtime:

SELECT * FROM orders o
JOIN customers c ON o.customer_id = c.id
WHERE o.amount > (SELECT percentile_cont(0.9) WITHIN GROUP (ORDER BY amount) FROM orders);

The subquery result is not known until the subquery executes — which may be after the scan of orders has already started. Pushing that filter down at plan time is impossible. Pushing it down at runtime is a much bigger win.

DataFusion’s answer is DynamicFilterPhysicalExpr (datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs): a physical expression whose value is a mutable cell that any operator with a reference to it can rewrite.

The shape of the cell

// datafusion/physical-expr/src/expressions/dynamic_filters/mod.rs
pub struct DynamicFilterPhysicalExpr {
    /// The original children of this PhysicalExpr, if any.
    ///
    /// This is necessary because the dynamic filter may be initialized with a
    /// placeholder (e.g. `lit(true)`) and later remapped to the actual
    /// expressions that are being filtered.
    children: Vec<Arc<dyn PhysicalExpr>>,
    /// If any of the children were remapped / modified (e.g. to adjust for
    /// projections) we need to keep track of the new children
    remapped_children: Option<Vec<Arc<dyn PhysicalExpr>>>,
    /// Cache of the last (generation, remapped-expression) pair returned by
    /// `current()`.
    current_cache: CurrentExprCache,
    /// The source of dynamic filters.
    inner: Arc<RwLock<Inner>>,
    /// Broadcasts filter state (updates and completion) to all waiters.
    state_watch: watch::Sender<FilterState>,
    ...
}

Four fields do all the work.

children is fixed at construction time. This is the crucial design constraint, and the doc comment explains why at length:

Generally the important bit is that the leaf children that reference columns do not change since those will be used to determine what columns need to read or projected when evaluating the expression.

In other words, the filter may change value at runtime, but it may not change shape. That invariant is what allows the Parquet scan to decide at open time which columns it must read: the column set of a dynamic filter is knowable before the filter has any value. A filter starts life as lit(true) and later becomes customer_id IN (1, 2, 3) — but it always references customer_id, so the scan always reads customer_id.

inner is the current value plus a monotonic generation counter. state_watch is a tokio::sync::watch channel for asynchronous waiters.

Updating and waiting

The write side is update():

pub fn update(&self, new_expr: Arc<dyn PhysicalExpr>) -> Result<()> {
    // Remap the children of the new expression to match the original children
    let new_expr = Self::remap_children(
        &self.children,
        self.remapped_children.as_ref(),
        new_expr,
    )?;

    // Load the current inner, increment generation, and store the new one
    let mut current = self.inner.write();
    let new_generation = current.generation + 1;
    *current = Inner {
        // Preserve the expression id across updates.
        expression_id: current.expression_id,
        generation: new_generation,
        expr: new_expr,
        is_complete: current.is_complete,
    };
    drop(current); // Release the lock before broadcasting

    // Broadcast the new state to all waiters
    let _ = self.state_watch.send(FilterState::InProgress {
        generation: new_generation,
    });
    Ok(())
}

Note the drop(current) before the broadcast. Holding a write lock across watch::send would be a lock-ordering hazard: a woken waiter might immediately try to read the filter and deadlock against the sender.

There are two flavours of waiting, and the distinction matters:

• wait_update() — returns when the generation increases. Used by operators that want to react to a tightening filter.
• wait_complete() — returns when the producer says “that was the last update”. Used by operators that need the final value before they can decide something, e.g. a hash join that must publish the union of all partitions’ build-side bounds.

pub async fn wait_complete(&self) {
    if self.inner.read().is_complete {
        return;
    }
    let mut rx = self.state_watch.subscribe();
    let _ = rx
        .wait_for(|state| matches!(state, FilterState::Complete { .. }))
        .await;
}

The is_complete fast path exists so that a consumer joining late (after the producer already finished) does not park forever on a channel that will never fire again.

The hot path problem: current()

Here is where a naive implementation falls apart. A consumer that applies the filter to every incoming batch needs the current expression. But the expression stored in inner refers to the original children, while the consumer may have rewritten the expression tree for its own schema (a projection pushdown renames/reorders columns). So current() has to remap the stored expression’s children to match what the consumer expects.

That remap is expr.transform_up(...) — a full tree walk. And for a join key filter it is a large tree walk, because the expression is an InList holding hundreds or thousands of literal values.

The fix is the current_cache:

pub fn current(&self) -> Result<Arc<dyn PhysicalExpr>> {
    // Fast path: cache hit for the current generation.
    let (expr, generation) = {
        let inner = self.inner.read();
        (Arc::clone(inner.expr()), inner.generation)
    };
    if let Some((cached_gen, cached_expr)) = self.current_cache.read().as_ref()
        && *cached_gen == generation
    {
        return Ok(Arc::clone(cached_expr));
    }
    // Slow path: (re)compute the remap and store it under a write lock.
    ...
}

The cache is keyed by generation, so it is invalidated exactly when it should be — when update() bumps the generation. Between updates (which for a hash join happen once per build, and for a TopK once per threshold refresh) every per-batch evaluation is a read-lock plus a pointer clone.

There is a subtle correctness detail in the write path, documented at length in the source:

// Only publish our result if it is strictly newer than whatever is
// currently cached. Without this guard a slow computation that
// observed an older `inner` could clobber a newer entry that a
// concurrent caller has already published (see #23532 review)
let should_write = match cache.as_ref() {
    Some((cached_gen, _)) => generation > *cached_gen,
    None => true,
};

Two threads racing in the slow path can observe different generations. The one that observed the newer generation must win, even if it finishes second. A plain “last writer wins” cache would install the older expression and force every subsequent reader to redo the remap — or worse, apply a stale filter. The generation > *cached_gen guard makes the cache monotone.

Cheap change detection: the tracker

Not every consumer wants to await an update. The Parquet row-group pruner wants to poll: “has anything changed since I last looked?”. The old way to answer that was to recursively fold PhysicalExpr::snapshot_generation over the whole predicate tree on every check — an O(tree) walk per row group.

DynamicFilterTracker (datafusion/physical-expr/src/expressions/dynamic_filters/tracker.rs) replaces that with a one-time subscription:

pub enum DynamicFilterTracking {
    /// The predicate contains no DynamicFilterPhysicalExpr at all.
    Static,
    /// The predicate contains one or more dynamic filters, but all of them have
    /// already been marked complete.
    AllComplete,
    /// The predicate contains at least one dynamic filter that can still change.
    Watching(DynamicFilterTracker),
}

impl DynamicFilterTracking {
    pub fn classify(predicate: &Arc<dyn PhysicalExpr>) -> Self {
        let mut subscriptions = Vec::new();
        let mut found_any = false;
        predicate.apply(|expr| {
            if let Some(filter) = expr.downcast_ref::<DynamicFilterPhysicalExpr>() {
                found_any = true;
                // Already-complete filters can never change again, so there is
                // no point subscribing to them.
                if !filter.is_complete() {
                    subscriptions.push(filter.subscribe());
                }
            }
            Ok(TreeNodeRecursion::Continue)
        })
        .expect("traversal closure is infallible");

        if !found_any {
            DynamicFilterTracking::Static
        } else if subscriptions.is_empty() {
            DynamicFilterTracking::AllComplete
        } else {
            DynamicFilterTracking::Watching(DynamicFilterTracker { subscriptions })
        }
    }
}

Three classifications, three behaviours:

• Static — no dynamic filter. The predicate can never change; build the pruning predicate once and be done.
• AllComplete — dynamic filters exist but are final. Their current value may differ from what planning saw (so a one-shot re-prune is still worth doing), but there is nothing to watch.
• Watching — at least one filter can still change. Keep the subscriptions and poll them.

The subscription itself is a watch::Receiver plus a last-seen generation, and observe() is “a cheap atomic load in the common ’nothing changed' case”. Crucially, the tracker shrinks: once a filter completes, its subscription is dropped, so the watch set only gets smaller over the life of a query.

The consumer side: file pruning

The FilePruner (datafusion/pruning/src/file_pruner.rs) is a compact demonstration of the whole pattern:

pub struct FilePruner {
    predicate: Arc<dyn PhysicalExpr>,
    /// Tracks the dynamic filters inside `predicate` so we only rebuild the
    /// pruning predicate when one of them has actually moved.
    tracking: DynamicFilterTracking,
    /// Whether `should_prune` has built+evaluated the pruning predicate at
    /// least once.
    checked_once: bool,
    file_schema: SchemaRef,
    file_stats_pruning: PrunableStatistics,
    predicate_creation_errors: Count,
}

should_prune() is guarded by a “has anything changed?” check, because building a PruningPredicate involves expression analysis and is not free:

pub fn should_prune(&mut self) -> Result<bool> {
    // Building the pruning predicate is expensive (it involves expression
    // analysis), so we only do it on the first check and whenever a dynamic
    // filter inside the predicate has actually moved.

And try_new refuses to build a pruner at all when it provably cannot help:

// Only build a pruner when it could prune something planning didn't
// already: the file has real column statistics, or the predicate carries
// a dynamic filter (whose value, or folded partition columns, can prune
// even without column statistics).
if !partitioned_file.has_statistics() && !tracking.contains_dynamic_filter() {
    return None;
}

That is a nice piece of engineering taste: the pruner is not just “always on”, it is “on only when it can do something new”.

The catch

• A dynamic filter is only as good as its producer. If the producer publishes a filter that is too weak, nothing is lost but a missed optimisation. If it publishes one that is too strong, rows are silently dropped. DataFusion’s hash join goes to considerable trouble here — merging bounds across all partitions under a mutex, electing exactly one finalizer so the filter is updated once, and keeping NULL rows when the join’s null equality requires them.
• wait_update can hang. The docs are explicit: a producer that never updates and never completes will park a consumer forever. DataFusion’s built-in producers always call mark_complete(), but a third-party producer that creates a dynamic filter nobody consumes would deadlock.
• The expression’s shape is frozen at plan time. If a producer wants to publish a filter over a column the consumer never read, it cannot. The children list has to be a superset of what the final expression will reference, and that list is decided before anyone knows the value.
• Cache invalidation is subtle. The generation-keyed cache and the monotone write guard are load-bearing. Removing either reintroduces either a stale-value bug or a performance cliff.

The next article in this series looks at the most important producer: how a hash join turns its build side into a filter that the probe side can execute.