Opensource internals

How DataFusion pushes filters through joins without breaking null semantics

11 min read Jay Hubbert

A deep dive into DataFusion's filter pushdown rules, transitive predicate inference across joins, and why LEFT JOIN null-extension forbids half the pushes

TL;DR: PushDownFilter splits a filter into conjuncts and pushes each one as far down as the algebra allows. A predicate is pushed through a join only if the side it references is preserved — the join returns a subset of that side’s rows, one output row per input row. For WHERE clauses a LEFT JOIN preserves the left; for ON clauses it preserves the right. Get that backwards and you change the query’s meaning. DataFusion also infers predicates across the join: ON A.x = B.x plus WHERE B.x = 5 implies A.x = 5, so the constant reaches the other table scan too.

The problem: a filter above a join is a filter above a materialization

Consider SELECT * FROM orders o JOIN customers c ON o.cid = c.id WHERE c.region = 'EU'. The naive plan reads every order, reads every customer, builds a hash table, executes the join, and then throws away 95% of the result.

The fix is to push c.region = 'EU' below the join, into the customers scan: the Parquet reader can use row-group statistics to skip chunks, and the hash table gets smaller, so fewer order rows survive to be probed. Doing this correctly is the entire difficulty. A predicate may be pushed through a node only if the node is filter-commutative with it: filter(op(data)) == op(filter(data)). That property is per-node and per-predicate, which is why the rule is a general tree rewrite rather than a pile of special cases.

The rule

PushDownFilter (datafusion/optimizer/src/push_down_filter.rs:851-860) declares apply_order() == Some(ApplyOrder::TopDown): the rule must see a Filter and its child before it rewrites the child, because the whole game is “move this filter toward the leaves”. (Contrast with SimplifyExpressions, which is BottomUp because simplification is local.)

The implementation is a single top-down pass that carries a stack of pending predicates down the tree (push_down_filter.rs:160-168): “When it passes through a filter, it stores that filter, and when it reaches a plan node that does not commute with that filter, it adds the filter to that place.”

Conjunction splitting: push what you can, keep what you can’t

The predicate is decomposed immediately:

rust 
// datafusion/optimizer/src/push_down_filter.rs:874-880
let predicate = split_conjunction_owned(filter.predicate.clone());
let old_predicate_len = predicate.len();
let new_predicates =
    with_debug_timing("simplify_predicates", || simplify_predicates(predicate))?;

split_conjunction_owned turns a > 1 AND b < 2 AND sum(c) > 3 into three independent predicates. This is what makes partial pushdown possible: a > 1 may be pushable below an Aggregate while sum(c) > 3 obviously is not.

Two things happen to the conjuncts before they travel. simplify_predicates (simplify_predicates.rs:60-116) applies bounded boolean reasoning — x AND x → x, contradiction detection — under a work budget (MAX_DISJUNCTION_WORK, MAX_DISJUNCTION_DEPTH) so a pathological disjunction cannot make planning quadratic. Then reorder_predicates puts cheap predicates first, using a two-class partition: a predicate is cheap iff every node in it is on the allow-list in is_cheap_node (reorder_predicates.rs:89-131), and anything unrecognized is expensive. The rationale is in the module doc: “DataFusion’s AND evaluator short-circuits the right-hand side when the left-hand side keeps few rows, so leading with a cheap predicate shrinks the batch that expensive ones see.” The heuristic refuses to reorder at all if any conjunct is volatile, since reordering changes how many times a side-effecting expression is evaluated.

Pushing through each kind of node

  • Sort — always pushable (push_down_filter.rs:982-988). Sorting fewer rows is strictly better.
  • Projection — needs care: a filter on an alias b must be rewritten back to the aliased expression a + 1 before it can go below the projection (rewrite_projection, push_down_filter.rs:1529+).
  • Distinct and Union — pushable; for a union the predicate goes into each branch, with column references remapped per branch (push_down_filter.rs:1091-1114).
  • Limit/Offset — not pushable, checked up front (push_down_filter.rs:936-945): if the child has a fetch or a skip, the rule returns the Filter unchanged, because LIMIT 3 then WHERE a > 10 is not the same as WHERE a > 10 then LIMIT 3. The other direction is handled by PushDownLimit, which the optimizer runs first (optimizer.rs:307-308: “Filters can’t be pushed down past Limits, we should do PushDownFilter after PushDownLimit”).
  • Aggregate — only predicates on the grouping keys can be pushed, and the group expression must be substituted with the alias the aggregate exposes (push_down_filter.rs:1116-1160). The code also filters out volatile group keys: a volatile group key is re-evaluated per row, so a predicate on it evaluated below the aggregate could see a different value than the one used for grouping.
  • Window — a predicate can be pushed below a window function only if it reads nothing but partition keys, and only if that key set is the intersection over all window functions in the plan (push_down_filter.rs:1172-1216): two windows partitioned by different keys see different row sets, so a predicate on a can only be pushed if every window is partitioned by a.

The join: the preserved-side analysis

Here is where correctness lives. A join side is preserved if the join’s output maps one-to-one back to that side’s rows (push_down_filter.rs:178-192):

We say a join side is preserved if the join returns all or a subset of the rows from the relevant side, such that each row of the output table directly maps to a row of the preserved input table. If a table is not preserved, it can provide extra null rows.

rust 
// datafusion/optimizer/src/push_down_filter.rs:193-210
pub(crate) fn lr_is_preserved(join_type: JoinType) -> (bool, bool) {
    match join_type {
        JoinType::Inner => (true, true),
        JoinType::Left => (true, false),
        JoinType::Right => (false, true),
        JoinType::Full => (false, false),
        JoinType::LeftSemi | JoinType::LeftAnti | JoinType::LeftMark => (true, false),
        JoinType::RightSemi | JoinType::RightAnti | JoinType::RightMark => (false, true),
    }
}

So for a WHERE clause above a LEFT JOIN, the left side is preserved and the right side is not. That is not a stylistic choice; it is a semantic fact:

text 
  customers (right)                customers (right, filtered)
  ┌──────────────┐                 ┌──────────────┐
  │ id=1, EU     │                 │ id=1, EU     │  ┌──────────┐
  │ id=2, US     │   LEFT JOIN     │ id=2, US     │  │ orders   │
  │ id=3, US     │  ───────────►   └──────────────┘  │ o.cid=2  │
  └──────────────┘                 NULL, NULL, NULL  │ o.cid=9  │
                                    (row for o.cid=9)  └──────────┘

Pushing c.region = 'EU' below the right input would delete id=2, US, and the order o.cid=2 would then become a null-extended row instead of a matched row. The join’s output changes, so the right side is not preserved and the predicate stays above.

The ON-clause asymmetry

But the ON clause of the same join behaves oppositely, and this trips up almost everyone who reads the code the first time:

rust 
// datafusion/common/src/join_type.rs:116-131
pub fn on_lr_is_preserved(&self) -> (bool, bool) {
    match self {
        JoinType::Inner => (true, true),
        JoinType::Left => (false, true),      // <-- right preserved for ON
        JoinType::Right => (true, false),
        JoinType::Full => (false, false),
        JoinType::LeftSemi | JoinType::RightSemi => (true, true),
        JoinType::LeftAnti => (false, true),
        JoinType::RightAnti => (true, false),
        JoinType::LeftMark => (false, true),
        JoinType::RightMark => (true, false),
    }
}

Why? Because an ON condition filters matches, not rows. LEFT JOIN c ON o.cid = c.id AND c.region = 'EU' does not remove left rows — it only prevents the non-EU customers from matching. The right side of a LEFT JOIN is not null-extended (only the left is), so filtering the right input is safe. The left side is null-extended, so filtering it below the join would be wrong. The code keeps the two tables separate on purpose: lr_is_preserved for WHERE, on_lr_is_preserved for ON.

The transitive closure: inferring predicates across the join

Given SELECT * FROM a JOIN b ON a.x = b.x WHERE b.x = 5, the join tells us a.x = b.x, so a.x = 5 is also true for every surviving row. DataFusion derives that:

rust 
// datafusion/optimizer/src/push_down_filter.rs:614-660
fn infer_join_predicates(
    join: &Join, predicates: &[Expr], on_filters: &[Expr],
) -> Result<Vec<Expr>> {
    // Null-aware joins ... Skip inference entirely for null-aware joins.
    if join.null_aware {
        return Ok(vec![]);
    }
    // Only allow both side key is column.
    let join_col_keys = join.on.iter()
        .filter_map(|(l, r)| {
            let left_col = l.try_as_col()?;
            let right_col = r.try_as_col()?;
            Some((left_col, right_col))
        })
        .collect::<Vec<_>>();
    ...
}

The substitution is a column-to-column map with the direction switch encoded as two const generics:

rust 
// datafusion/optimizer/src/push_down_filter.rs:779-816
fn infer_join_predicates_impl<
    const ENABLE_LEFT_TO_RIGHT: bool,
    const ENABLE_RIGHT_TO_LEFT: bool,
>(...) {
    for predicate in input_predicates {
        for &col in &predicate.column_refs() {
            for (l, r) in join_col_keys.iter() {
                if ENABLE_LEFT_TO_RIGHT && col == *l { /* col -> r */ }
                if ENABLE_RIGHT_TO_LEFT && col == *r { /* col -> l */ }
            }
        }
        inferred_predicates.try_build_predicate(predicate.clone(), &join_cols_to_replace)?;
    }
}

For an inner join both directions are enabled, so b.x = 5 yields a.x = 5 and vice versa. For a LEFT join, only the ON-filter direction right-to-left is enabled (push_down_filter.rs:734-757).

For non-inner joins there is an extra safety check, because inference is only sound if the inferred predicate rejects nulls:

rust 
// datafusion/optimizer/src/push_down_filter.rs:660-701
fn try_build_predicate(
    &mut self, predicate: Expr, replace_map: &HashMap<&Column, &Column>,
) -> Result<()> {
    if self.is_inner_join
        || matches!(
            is_restrict_null_predicate(predicate.clone(), replace_map.keys().copied()),
            Ok(true)
        )
    {
        self.predicates.push(replace_col(predicate, replace_map)?);
    }
    Ok(())
}

The doc comment gives the canonical counter-example: “When the JoinType is not Inner, we need to detect whether the inferred predicate can strictly filter out NULL, otherwise ignore it. e.g. SELECT * FROM t1 LEFT JOIN t2 ON t1.c0 = t2.c0 WHERE t2.c0 IS NULL; We cannot infer the predicate t1.c0 IS NULL, otherwise the predicate will be pushed down to the left side, resulting in the wrong result.”

is_restrict_null_predicate (datafusion/optimizer/src/utils.rs:220-246) implements this by evaluating the predicate against a column of nulls and checking the result is false or NULL. That is a nice trick: instead of a syntactic analysis of null-rejecting predicates, it asks “if this column were NULL, would the predicate still pass?” — a semantic test implemented with the executor.

OR extraction

Filters under an OR are usually not pushable. DataFusion does a best-effort extraction: from (A AND B) OR (C AND D) it can pull out a clause that is a necessary condition for the disjunction, such as A OR C. Adding a necessary condition to a child is sound — it can only remove rows the original filter would have removed anyway. extract_or_clauses_for_join (push_down_filter.rs:336+) only does this for the sides that are preserved.

Pushing into the scan

The last stop is the TableScan, where the predicate meets a storage engine. The provider answers per-predicate with TableProviderFilterPushDown::{Exact, Inexact, Unsupported}. Exact means “I applied this filter and return exactly the matching rows”, so the predicate is removed from the plan. Inexact means “I applied it, but you must apply it again” (a Parquet row-group prune, for instance) so the predicate is kept above the scan as well. Unsupported means it stays where it is.

Two guards appear here: predicates containing scalar subqueries are never handed to a provider (the subquery’s value is not known at plan time), and volatile predicates are never pushed (a scan that evaluates them once per file group rather than once per row changes the answer). There is also a fix-up for the FALSE predicate: a provider that accepts Exact pushdown would absorb FALSE into the scan, hiding the contradiction from EliminateFilter and still invoking scan() (push_down_filter.rs:921-931), so DataFusion short-circuits to an EmptyRelation instead.

The physical half

The logical rule handles the plan shape. The physical optimizer does the same job one level lower with a different protocol — a negotiation between parent and child, because the children are ExecutionPlans that know things the optimizer does not (whether a Parquet reader can prune, whether a HashJoinExec accepts a filter on its build side, whether a WindowAggExec has a partition key).

The protocol is documented at the top of datafusion/physical-optimizer/src/filter_pushdown.rs:21-38: the optimizer asks the parent for a FilterDescription (which parent filters and which of the parent’s own filters may be offered to each child), recurses into the children with those predicates, collects the FilterPushdownPropagation results, and finally calls handle_child_pushdown_result on the parent so it can decide what to do with the leftovers.

The bookkeeping is the interesting part. Each predicate has to be tracked across the recursion even when it is not pushed, so the parent can be told “this one went through, that one did not”. FilteredVec (filter_pushdown.rs:609-705) does that with index arithmetic: each node filters out the predicates it may not push (volatile ones, per allow_pushdown_for_expr at filter_pushdown.rs:707), records the surviving indices, and maps the child’s answers back to the original positions. The map_results_to_original and trace_to_original_index helpers are pure index bookkeeping — the kind of code that is trivial to get subtly wrong and is exactly why it is unit-tested in isolation (filter_pushdown.rs:716-860).

There are two rounds: FilterPushdown::new() runs before EnsureRequirements, and FilterPushdown::new_post_optimization() runs at the very end (optimizer.rs:223, 295). The second round exists because dynamic filters (produced by a SortExec in TopK mode, or by a hash join build side) hold references to specific source ExecutionPlans; if the plan is mutated after they are created, the references dangle. So dynamic filters are pushed last, when nothing else will move the source again.

Trade-offs and traps

Pushdown is not free. Every pushed predicate must be re-serialized, re-planned and re-evaluated at the scan. A predicate that selects 99% of rows and lives in a Parquet file with no useful statistics makes the query slower.

Partial pushdown changes the shape of the plan, which is why the rule returns Transformed::yes only when something actually moved, and re-inserts the leftover predicates with with_filters (push_down_filter.rs:1663-1680).

The one rule that is deliberately not applied: pushing a filter below a LEFT JOIN’s right input, and pushing an ON filter below a LEFT JOIN’s left input. Both are blocked not by a heuristic but by a semantic argument about null extension, encoded in the two *_is_preserved functions. Getting this wrong produces a query that returns the right number of rows with the wrong contents — the kind of bug that survives a year of production because the row count still looks plausible.

DatafusionOptimizerPerformanceInternals