Embedded Projections: How DataFusion Removes a Whole Plan Node by Folding It Into the Join

TL;DR: A ProjectionExec on top of a HashJoinExec is not free. The join builds its output by gathering rows — for every matched build row, it has to take() the build side’s columns into a new batch. If the projection above only wants two of the ten build columns, the join still materialised all ten. DataFusion’s EmbeddedProjection trait lets the join take a column list directly, so it only ever gathers the columns that survive — and the separate projection node disappears too.

Projection pushdown is one of the oldest optimisations in query engines, and it is usually taught as “push the projection below the operator”. That works when the operator’s own logic does not need the dropped columns — but a join needs its key columns, and its filter might need more.

The gap

The comment in datafusion/physical-plan/src/projection.rs states the problem exactly:

Some projection can’t be pushed down left input or right input of hash join because filter or on need may need some columns that won’t be used in later. By embed those projection to hash join, we can reduce the cost of build_batch_from_indices in hash join (build_batch_from_indices need to can compute::take() for each column) and avoid unnecessary output creation.

Two costs:

1. build_batch_from_indices does a take() per column. The hash join’s output is built by gathering the matched build rows. If the build side has 10 columns and the output only uses 2, the other 8 are take()d and then thrown away by the projection above.
2. A separate plan node. The ProjectionExec is another operator in the pipeline: another stream, another set of metrics, another poll per batch.

Embedding fixes both.

The trait

/// Trait for execution plans that can embed a projection, avoiding a separate
/// [`ProjectionExec`] wrapper.
///
/// # Empty projections
///
/// `Some(vec![])` is a valid projection that produces zero output columns while
/// preserving the correct row count. Implementors must ensure that runtime batch
/// construction still returns batches with the right number of rows even when no
/// columns are selected (e.g. for `SELECT count(1) … JOIN …`).
pub trait EmbeddedProjection: ExecutionPlan + Sized {
    fn with_projection(&self, projection: Option<Vec<usize>>) -> Result<Self>;
}

The trait is small, and the doc comment is where the subtlety lives. An empty projection — Some(vec![]) — is legal, and it produces a batch with zero columns but the correct row count.

That matters for SELECT count(1) FROM a JOIN b. The join still has to know how many rows matched, and the output batch has to have that many rows, even though it has no columns at all. A naive implementation would produce a zero-row batch and the count would be wrong.

And the embedding is used to avoid exactly that waste:

// If the projection has no expressions at all (e.g., ProjectionExec: expr=[]),
// embed an empty projection into the execution plan so it outputs zero columns.
// This avoids allocating throwaway null arrays for build-side columns
// when no output columns are actually needed (e.g., count(1) over a right join).
if projection.expr().is_empty() {
    let new_execution_plan = Arc::new(execution_plan.with_projection(Some(vec![]))?);
    return Ok(Some(new_execution_plan));
}

“throwaway null arrays” — for a count(1) over a right join, the join still has to emit the NULL-extended rows for the unmatched side. Without the embedding, those rows are materialised with every build column filled with NULLs, and then all the columns are dropped. That is a per-row allocation of null arrays for data nobody reads.

The three outcomes

The interesting part of try_embed_projection is its decision logic at the end:

if is_projection_removable(&new_projection) {
    // Residual is identity — embedding fully absorbed the projection.
    Ok(Some(new_execution_plan))
} else if columns_reduced {
    // Embedding reduced columns even though a residual is still needed
    // for renames or expressions — worth keeping.
    Ok(Some(new_projection))
} else {
    // No columns eliminated and residual still needed — embedding just
    // adds an unnecessary column reorder inside the operator.
    Ok(None)
}

Three cases:

1. The projection was fully absorbed. The join now produces exactly the columns the projection wanted, in the right order, with the right names. The ProjectionExec node disappears from the plan entirely. One fewer operator, one fewer stream, one fewer poll per batch.

2. Some columns were eliminated but a residual is still needed — the projection contained an expression (a + 1) or a rename, not just column references. The embedded projection still saves the join from materialising the dropped columns, and a thin ProjectionExec sits on top to do the arithmetic. Worth keeping.

3. Nothing was eliminated and a residual is needed. The embedding would only add a column reorder inside the operator, which is pure overhead. So the whole thing is rejected and the original plan is kept.

That third case is the one that shows the care: an optimisation that fires unconditionally is not always an optimisation.

Rewriting the expressions

The residual projection’s expressions have to be rewritten, because the input schema changed. The code builds a mapping from the original column indices to the new ones:

// Build projection expressions for update_expr. Zip the projection_index with the new_execution_plan output schema fields.
let embed_project_exprs = projection_index
    .iter()
    .zip(new_execution_plan.schema().fields())
    .map(|(index, field)| ProjectionExpr {
        expr: Arc::new(Column::new(field.name(), *index)) as Arc<dyn PhysicalExpr>,
        alias: field.name().to_owned(),
    })
    .collect::<Vec<_>>();

and then update_expr rewrites every column reference in the residual projection through that mapping:

for proj_expr in projection.expr() {
    // update column index for projection expression since the input schema has been changed.
    let Some(expr) =
        update_expr(&proj_expr.expr, embed_project_exprs.as_slice(), false)?
    else {
        return Ok(None);
    };

Note the else { return Ok(None) }: if the expression cannot be rewritten, the whole embedding is abandoned and the original plan is returned. Better to give up than to produce a plan with a subtly wrong projection.

And the comment on the residual explains why it is needed at all:

Old projection may contain some alias or expression such as a + 1 and CAST('true' AS BOOLEAN), but our projection_exprs in hash join just contain column, so we need to create the new projection to keep the original projection.

The join’s embedded projection can only be column indices — it is a Vec<usize> — so anything that is not a plain column reference has to stay in a projection above.

The catch

• Only column indices, no expressions. The embedded projection is a Vec<usize>. Anything with an expression in it must remain in a ProjectionExec above.
• The empty-projection case is a correctness trap. The doc calls it out explicitly, because getting it wrong changes the row count and therefore the answer to a count(*).
• It only applies to operators that opt in. The trait has to be implemented; HashJoinExec and the ASOF join do it, other operators do not.
• The decision is heuristic. The “columns reduced” test is a proxy for “this is worth it”. A join that reorders columns but drops none might still be worth embedding in some workloads, but the code conservatively declines.

Why it matters

This is a small optimisation with a nice shape: it removes a plan node and reduces the work the node below it does. Those two goals usually conflict — the usual projection pushdown trades a plan node for extra work in the operator below — and embedding gets both.

And the empty-projection case is the sort of thing that shows up in every engine eventually. SELECT count(1) FROM a JOIN b produces a batch with no columns and N rows, and every implementation has to get that right. DataFusion’s doc comment is a warning to the next person who touches it.