How DataFusion Decorrelates a Correlated Subquery Into a Join (and Pays for It With a CASE Expression)

TL;DR: A correlated subquery — one that references a column from the outer query — is the hardest thing in SQL to optimise, because the naive execution runs the inner query once per outer row. DataFusion rewrites it into a join: the correlated predicate is pulled up out of the subquery’s filter and becomes the join’s ON condition. The subtlety is that aggregates cannot distinguish “no rows” from “a row whose value is NULL”, so a naive rewrite silently changes COUNT(*) from 0 to NULL. DataFusion fixes this with a magic column called __always_true and a CASE expression at the end.

sql
SELECT c.name,
       (SELECT max(o.amount) FROM orders o WHERE o.cid = c.id)
FROM customers c;

Written literally, this means: for each customer, scan the orders and find the maximum amount. A smart DBMS does not do that. It decorrelates the subquery into a join:

sql
SELECT c.name, agg.max_amount
FROM customers c
LEFT JOIN (SELECT cid, max(amount) AS max_amount FROM orders GROUP BY cid) agg
  ON agg.cid = c.id;

One pass over orders, one hash aggregate, one join. From “O(rows × customers)” to O(rows + customers).

DataFusion’s implementation of this is in datafusion/optimizer/src/decorrelate.rs and datafusion/optimizer/src/scalar_subquery_to_join.rs — 1,142 and 1,286 lines respectively, and they are the densest, most careful code in the optimizer.

The pull-up

The core type is PullUpCorrelatedExpr, and its doc comment states the plan:

This struct rewrite the sub query plan by pull up the correlated expressions (contains outer reference columns) from the inner subquery’s ‘Filter’. It adds the inner reference columns to the ‘Projection’ or ‘Aggregate’ of the subquery if they are missing, so that they can be evaluated by the parent operator as the join condition.

So the transformation has three parts:

Find the correlated conjuncts in the subquery’s filter.
Remove them from the filter and remember them as join_filters.
Make sure the columns they reference survive to the subquery’s output — by adding them to a Projection or as extra grouping columns of an Aggregate.

The filter arm is where this happens:

rust
LogicalPlan::Filter(plan_filter) => {
    let subquery_filter_exprs = split_conjunction(&plan_filter.predicate);
    ...
    let (mut join_filters, subquery_filters) =
        find_join_exprs(subquery_filter_exprs)?;
    if let Some(in_predicate) = &self.in_predicate_opt {
        // in_predicate may be already included in the join filters, remove it from the join filters first.
        join_filters = remove_duplicated_filter(join_filters, in_predicate)?;
    }
    let correlated_subquery_cols =
        collect_subquery_cols(&join_filters, subquery_schema)?;
    for expr in join_filters {
        if !self.join_filters.contains(&expr) {
            self.join_filters.push(expr)
        }
    }

find_join_exprs is the key primitive: it splits the filter’s conjuncts into those that can become equality join predicates (correlated columns on both sides) and those that must stay as a residual filter. collect_subquery_cols then finds which inner columns the join predicate needs, so they can be added to the subquery’s output.

The “can’t pull up here” list

The really interesting part of f_down is the exhaustive match that decides which plan nodes a correlated filter may move above. Every arm is a correctness argument. A selection:

rust
// A correlated filter can only move above a join from a side whose
// rows the join preserves. Below the side an outer join fills with
// NULLs, the filter decides which rows are unmatched, so pulling it
// above the join changes the result. The side a semi, anti or mark
// join does not output cannot give its columns to a pulled up
// filter either.
LogicalPlan::Join(ref join) if !correlated_inputs_are_preserved(join) => {
    Ok(self.stop_pull_up(plan))
}

and

rust
// An ASOF join keeps each left row once, so a filter on the left
// side can move above it. A filter on the right side selects the
// rows the join can match, so it must stay below the join.
LogicalPlan::AsOfJoin(ref join) if holds_outer_reference(&join.right) => {
    Ok(self.stop_pull_up(plan))
}

and, with an explicit issue reference:

rust
// - A `DISTINCT` shares the exemption of an Aggregate without
//   aggregate functions: its output takes the columns the node
//   below it adds, so they become part of its keys. This is only
//   right when the pulled up filter fixes those columns for each
//   outer row. A non-equality filter can return duplicates, see
//   https://github.com/apache/datafusion/issues/25808.

The comment above the whole match is a design statement:

This match is exhaustive on purpose. A correlated filter moves above a node only when the node is known to keep the result the same. Add each new LogicalPlan variant to one of the groups below.

That is a much better failure mode than the alternative — a new plan variant silently inheriting the wrong behaviour.

The count bug

Here is the part that makes this hard.

Consider:

sql
SELECT c.name, (SELECT count(*) FROM orders o WHERE o.cid = c.id) FROM customers c;

For a customer with no orders, the correct answer is 0. Now decorrelate it. The inner plan becomes SELECT o.cid, count(*) FROM orders GROUP BY o.cid. The outer becomes a LEFT JOIN. For a customer with no orders, the LEFT JOIN produces no matching row, so the right side is NULL-extended — and count(*) comes out as NULL, not 0.

The SQL standard says count(*) over an empty set is 0. The rewrite produced NULL. That is the “count bug”, and it is not specific to count:

The anomaly arises because aggregates fail to distinguish between an empty set and null values when optimizing a correlated query as a join. Here, we use “the count bug” to refer to all such cases.

(Originally described in Optimization of Nested SQL Queries Revisited, and tracked in DataFusion as issue #10553.)

The fix is a two-part trick.

Part one: compute the aggregate’s value on an empty input. For each aggregate expression in the subquery, work out what it evaluates to when the input is empty. count(*) → 0, sum(x) → NULL, max(x) → NULL, and anything built on top of those (count(*) + 2 → 2) is computed by running the expression simplifier on the substituted defaults:

rust
fn agg_exprs_evaluation_result_on_empty_batch(
    agg_expr: &[Expr],
    schema: &DFSchemaRef,
    expr_result_map_for_count_bug: &mut ExprResultMap,
) -> Result<()> {
    ...
    let result_expr = e
        .clone()
        .transform_up(|expr| {
            let new_expr = if let Expr::AggregateFunction(agg) = &expr {
                let return_type = expr.get_type(schema.as_ref())?;
                let default_value = agg.func.default_value(&return_type)?;
                Transformed::yes(Expr::Literal(default_value, None))
            } else {
                Transformed::no(expr)
            };
            Ok(new_expr)
        })
        .data()?;

The default_value of the aggregate function is the answer for the empty set, and the simplifier propagates it through the surrounding arithmetic. Note that this uses the function’s declared default rather than a hardcoded count → 0, so a user-defined aggregate with a different empty set value works too.

Part two: ship a “did this row actually match?” flag through the join. The decorrelator adds a projection of the literal true to the subquery, aliased to a magic name:

rust
/// Used to indicate the unmatched rows from the inner(subquery) table after the left out Join
/// This is used to handle [the Count bug]
pub const UN_MATCHED_ROW_INDICATOR: &str = "__always_true";

The subquery’s projection becomes count(Int64(1)) AS count(*), t2.a, Boolean(true) AS __always_true. Because the LEFT JOIN null-extends the right side, __always_true is NULL exactly when no inner row matched. That NULL is the signal.

Part three: the CASE. The final projection uses the flag to pick the right value:

rust
let mut builder = when(Expr::Column(indicator_col).is_null(), result);
if let Some((when_expr, then_expr)) = &having_arm {
    builder = builder.when(when_expr.clone(), then_expr.clone());
}
let compensation_expr = builder.otherwise(Expr::Column(value_col.clone()))?;

which renders as:

sql
CASE WHEN __scalar_sq_1.__always_true IS NULL
     THEN 0              -- the empty-set value
     ELSE __scalar_sq_1.count(*)
END

And the aggregates that do return NULL on an empty set are skipped entirely:

rust
if evaluates_to_null(result.clone(), result.column_refs())? {
    // Aggregates whose empty-input value is NULL (max/min/sum/…)
    // need no compensation: the LEFT JOIN already produces NULL
    // for unmatched outer rows.
    continue;
}

So max(amount) — which is NULL for an empty set — needs no CASE at all. Only the aggregates whose empty-set value is not NULL need the compensation. That is a nice touch: the fix is applied exactly where it is needed and nowhere else.

The generated plan, from the test suite, reads:

text
Filter: CASE WHEN __scalar_sq_1.__always_true IS NULL THEN Int64(0) ELSE __scalar_sq_1.count(*) END > Int64(0)
  Projection: t1.a, t1.b, __scalar_sq_1.count(*), __scalar_sq_1.__always_true
    HashJoinExec: mode=CollectLeft, join_type=Right, on=[(a@1, a@0)]
      ProjectionExec: expr=[count(Int64(1))@1 as count(*), a@0 as a, true as __always_true]

The `IN` subquery subtlety

There is one more trap, documented at length on the correlated_filters field:

Every correlated conjunct that a Filter of the subquery applies, before remove_duplicated_filter drops the ones that the IN predicate already covers. The outer references are kept, so that a subquery column and an outer column with the same qualified name stay apart.
join_filters holds only the conjuncts that the join still needs. This list is what the subquery enforces on its own rows. The caller uses it to tell if a join key can be NULL inside the scope of an outer row: x IN (SELECT y FROM .. WHERE y = x) keeps every NULL y out of its result, although join_filters no longer says so.
The list is cleared when the pull up passes a node that can put a NULL back into such a column: an outer join, a union or a grouping set.

The problem: x IN (SELECT y FROM t WHERE y = x). The predicate y = x becomes the join key, so it disappears from join_filters. But the original predicate also had the effect of excluding NULL y values, and after the rewrite the join is an equi-join, which by SQL semantics does not match NULLs anyway — so the effect is preserved by accident. But if the pull-up passes through an outer join that can reintroduce a NULL into the join key, the exclusion is lost. Hence the separate list, and hence the “cleared when the pull up passes a node that can put a NULL back” rule.

The catch

The rewrite is not always possible. can_pull_up goes to false for a variety of shapes — an aggregate that the correlated filter cannot be moved above, a DISTINCT ON, a recursive query. When that happens, the subquery stays a subquery and is executed by ScalarSubqueryExec (or, for IN/EXISTS, by a semi-join) at runtime. The rewrite is an optimisation, not a requirement.
The CASE costs something. Every projected aggregate now carries a CASE expression, evaluated per row. For a correlated scalar subquery over a large outer table, that is a real cost — the price of correctness.
The magic column is a real column. __always_true flows through the join, the projection, and the physical plan. The name is chosen to be unlikely to collide, and the code qualifies it with the subquery alias to avoid ambiguity with an outer table column of the same name. But it is visible in EXPLAIN output, and it is a column you pay for.
Correctness here is a moving target. The DISTINCT exemption, the GROUPING SETS handling, the column_refs_above tracking — these are all responses to specific bugs, and each new plan node type or SQL feature can find a new hole. The exhaustive match is the safety net.

The general lesson is worth stating plainly: rewriting a subquery into a join is one of the highest-value optimisations in a query engine, and it is also one of the highest-risk, because SQL’s semantics for “empty set” and “NULL” are finer than most engines’ internal representations. DataFusion’s answer is to compute the empty-set semantics at rewrite time and carry the information needed to reconstruct them through the join.

The pull-up

The “can’t pull up here” list

The count bug

The IN subquery subtlety

The catch

The `IN` subquery subtlety