How DataFusion replaces a projection with an index instead of materializing it

TL;DR: The physical optimizer is a single-pass pipeline of ~20 rules that run after the plan is built, when the operators know things the planner did not. ProjectionPushdown does not “push a projection down” in the arrow-copying sense: it hands the scan a list of column indices and deletes the ProjectionExec node. file_row_index() is rewritten into a reference to a virtual column the Parquet reader produces as an offset — the function is never evaluated per row. A NestedLoopJoin filter that depends on only one side is computed once per side instead of once per cross-product row.

Two optimizers, two jobs

The logical optimizer runs on a LogicalPlan: it does not know what a Parquet file is, how many partitions a scan has, or whether a hash join can accept a filter on its build side. The physical optimizer runs on an ExecutionPlan, where all of that is known, and it does not get to change the query’s semantics.

The contract is spelled out in the source (datafusion/physical-optimizer/src/optimizer.rs:97-125):

1. Correctness. The returned plan must a) be valid (it can be executed without internal errors) and b) produce the same results as the input plan. A rule may change the operators in the plan, such as changing which operator implements a join, but it may never change the query’s semantics.
2. Reorderable. Because each rule preserves correctness, running the rules in a different order must still yield a valid plan that produces the same results.

…with an honest list of exceptions, because three rules exist to make a plan executable rather than to make it fast: EnsureRequirements (inserts the repartitions and sorts), OutputRequirements (records output requirements) and SanityCheckPlan (rejects plans whose requirements are not satisfied).

The rule order is fixed and load-bearing (datafusion/physical-optimizer/src/optimizer.rs:192-300):

let rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>> = vec![
    Arc::new(OutputRequirements::new_add_mode()),
    Arc::new(AggregateStatistics::new()),
    Arc::new(EliminateConstantJoin::new()),
    Arc::new(JoinSelection::new()),
    Arc::new(LimitedDistinctAggregation::new()),
    Arc::new(FilterPushdown::new()),
    Arc::new(WindowTopN::new()),
    Arc::new(EnsureRequirements::new()),
    Arc::new(CombinePartialFinalAggregate::new()),
    Arc::new(OptimizeAggregateOrder::new()),
    Arc::new(ProjectionPushdown::new()),
    ...
    Arc::new(FilterPushdown::new_post_optimization()),
    Arc::new(SanityCheckPlan::new()),
];

Note the two ProjectionPushdown entries and the two FilterPushdown entries — the same rule appears twice at different points because the plan shape it cares about only exists at that point in the pipeline. The docs are blunt: “Optimization Coverage Is Not Guaranteed. … a built-in rule may silently miss an optimization that would apply in the default order.” (optimizer.rs:127-155)

The projection trick

Start with the simplest possible plan:

Projection: t.a, t.b
  DataSourceExec: file_groups={...}, projection=[a, b, c]

There are three possible fates for the ProjectionExec, and the rule picks the cheapest.

1. It is the identity → delete it

// datafusion/physical-plan/src/projection.rs:1075-1085
fn is_projection_removable(projection: &ProjectionExec) -> bool {
    exprs.iter().enumerate().all(|(idx, proj_expr)| {
        let Some(col) = proj_expr.expr.downcast_ref::<Column>() else {
            return false;
        };
        col.name() == proj_expr.alias && col.index() == idx
    }) && exprs.len() == projection.input().schema().fields().len()
        && projection.schema() == projection.input().schema()
}

All three conditions matter: every expression must be a bare Column with no alias, the projection must have exactly as many entries as the input has fields, and the schemas must be equal (not merely the same width). SELECT a, b over (a, b) goes; SELECT b, a does not; SELECT a AS c, b does not.

2. It is a permutation or a subset → become a list of indices

This is the interesting case. The ProjectionExec is handed to its child via try_swapping_with_projection, and the scan swallows it: DataSourceExec::try_swapping_with_projection (datafusion/datasource/src/source.rs:529-545) forwards to the DataSource, which forwards to the file source’s own try_pushdown_projection.

// datafusion/physical-plan/src/projection.rs:1046-1066
pub fn remove_unnecessary_projections(
    plan: Arc<dyn ExecutionPlan>,
) -> Result<Transformed<Arc<dyn ExecutionPlan>>> {
    let maybe_modified = if let Some(projection) = plan.downcast_ref::<ProjectionExec>() {
        if is_projection_removable(projection) {
            return Ok(Transformed::yes(Arc::clone(projection.input())));
        }
        if projection.overrides_metadata() {
            return Ok(Transformed::no(plan));
        }
        projection.input().try_swapping_with_projection(projection)?
    } else {
        return Ok(Transformed::no(plan));
    };
    Ok(maybe_modified.map_or_else(|| Transformed::no(plan), Transformed::yes))
}

The scan’s own representation of “the columns I want” is a ProjectionExprs — a list of expressions — and for a column-only projection that list is exactly a list of indices:

// datafusion/physical-expr/src/projection.rs:396-404
/// Extract the column indices used in this projection.
/// For example, for a projection `SELECT a AS x, b + 1 AS y`, where `a` is at index 0 and `b` is at index 1,
/// this function would return `[0, 1]`.
pub fn column_indices(&self) -> Vec<usize> {
    self.exprs.iter()
        .flat_map(|e| collect_columns(&e.expr).into_iter().map(|col| col.index()))
        .sorted_unstable().dedup().collect_vec()
}

So the transformation of SELECT b, a FROM t is:

    Before:
    ProjectionExec: exprs=[b@1, a@0]
      DataSourceExec: file_groups={...}, projection=[a, b, c]

    After:
    DataSourceExec: file_groups={...}, projection=[b@1, a@0]

The ProjectionExec node is gone. There is no Arrow array copy, no per-batch take, no extra allocation. The scan reads two leaves out of the Parquet file, in the order the projection asked for, and hands the reader a ProjectionMask built from the column references.

This is the real meaning of “projection pushdown” in DataFusion. The projection does not move; it dissolves into the scan’s column list. The ProjectionExec operator exists only for the cases where the scan cannot absorb the expression — a computed column, a UDF, a reordering the source cannot express.

3. It is a computation → it stays, but the plan below it shrinks

When the projection contains real expressions, remove_unnecessary_projections leaves it alone, but the input to the projection still gets narrowed. That is what try_swapping_with_projection does on a FilterExec (datafusion/physical-plan/src/filter.rs:774-800): if the projection narrows the schema, the filter’s predicate is rewritten into the projection’s output coordinates with update_expr(self.predicate(), projection.expr(), false) — the false means “project”, not “unproject” — and the filter is rebuilt on top of the projection.

update_expr: rewriting references in both directions

// datafusion/physical-expr/src/projection.rs:1149-1160
/// 1) When `unproject` is `true`:
///    Rewrites an expression with respect to the projection expressions,
///    effectively "unprojecting" it to reference the original input columns.
///    For example, given
///    * the expressions `a@1 + b@2` and `c@0`
///    * and projection expressions `c@2, a@0, b@1`
///    Then
///    * `a@1 + b@2` becomes `a@0 + b@1`
///    * `c@0` becomes `c@2`
///
/// 2) When `unproject` is `false`:
///    Rewrites the expression to reference the projected expressions,
///    effectively "projecting" it. The resulting expression will reference
///    the indices as they appear in the projection.

The implementation carries a three-state “did I manage to rewrite everything?” flag, because a partial rewrite is worse than none: RewriteState::{Unchanged, RewrittenValid, RewrittenInvalid}, and the function returns None when the state is RewrittenInvalid — meaning “I could not express this expression over the new schema, so do not do the rewrite at all” (projection.rs:1163-1230).

The same machinery shows up again when a filter is pushed into a scan. The filter refers to the scan’s output schema, but the scan’s filter machinery wants table schema coordinates (datafusion/datasource/src/file_scan_config/mod.rs:1040-1058): unproject_expr maps output coordinates back to input coordinates, and reassign_expr_columns then re-bases the indices onto the table schema. Two different coordinate systems, two different functions, and a great way to produce a wrong-but-plausible answer if you mix them up.

The index trick: file_row_index()

Now the trick with the best name. file_row_index() is a scalar UDF that returns the row’s index within the file. Left alone, it is a ScalarFunctionExpr evaluated for every row of every batch — a materialized column of row numbers, computed by a function that has no idea what a row number is.

DataFusion rewrites it into a reference to a virtual column that the Parquet reader produces natively:

// datafusion/physical-expr-adapter/src/rewrite.rs:97-110
pub fn rewrite_file_row_index_expr(
    expr: Arc<dyn PhysicalExpr>,
    row_index_name: &str,
    row_index_idx: usize,
) -> Result<Arc<dyn PhysicalExpr>> {
    rewrite_scalar_udf::<FileRowIndexFunc, _>(expr, |_| {
        let source = Arc::new(Column::new(row_index_name, row_index_idx));
        let target_field = Arc::new(Field::new("file_row_index", DataType::Int64, true));
        Ok(Arc::new(CastExpr::new_with_target_field(source, target_field, None)))
    })
}

and the projection that contains it is rewritten as a whole (rewrite.rs:125-158): the row-index column is appended to the scan’s base projection if it is not already there, and every file_row_index() in the pushed projection is rewritten to read from it.

The plumbing on the Parquet side is where the “index” becomes real (datafusion/datasource-parquet/src/source.rs:702-729):

fn try_pushdown_projection(
    &self, projection: &ProjectionExprs,
) -> datafusion_common::Result<Option<Arc<dyn FileSource>>> {
    let mut source = self.clone();
    if !projection.iter().any(|projection_expr| {
        expr_references_scalar_udf::<FileRowIndexFunc>(&projection_expr.expr)
    }) {
        source.projection = self.projection.try_merge(projection)?;
        return Ok(Some(Arc::new(source)));
    }
    // If we can find a reference to `FileRowIndexFunc`, we add it as a virtual column
    // or re-use an existing one in the table's schema.
    let (table_schema, row_index_col) = table_schema_with_row_index_col(self.table_schema());
    source.table_schema = table_schema;
    source.projection = rewrite_file_row_index_projection(
        &self.projection, projection, &row_index_col,
    )?;
    Ok(Some(Arc::new(source)))
}

and the virtual column is a Parquet RowNumber extension type, which the arrow reader fills in as an offset into the row group (source.rs:1324-1332):

let row_index_field = Arc::new(
    Field::new(&row_index_name, DataType::Int64, true).with_extension_type(RowNumber),
);

So the shape of the trick is:

    Before:
    ProjectionExec: exprs=[file_row_index() AS idx, t.a]
      DataSourceExec: file_groups={...}, projection=[a]

    After:
    DataSourceExec: file_groups={...}, projection=[__datafusion_file_row_index@N, a@0]
      (the reader emits __datafusion_file_row_index as the row offset)

A per-row function evaluation is replaced by an index the reader already knows how to produce, and the projection node disappears. That is “replace a projection with an index instead of materializing it” in the most literal sense available in this codebase. The same idea shows up in a smaller way for input_file_name(), which is rewritten into a per-file Literal (rewrite.rs:189-193) — a constant, not a per-row function.

The nested loop join: compute once, not per cross-product

The other half of ProjectionPushdown is a rule that has nothing to do with columns and everything to do with arithmetic. A nested loop join evaluates its filter for every pair of rows: if the filter is upper(l.name) = r.name, upper(l.name) is recomputed |R| times per left row. The rule finds subexpressions that depend on only one side and computes them once per row of that side:

// datafusion/physical-optimizer/src/projection_pushdown.rs:339-385
fn rewrite(&mut self, expr: Arc<dyn PhysicalExpr>) -> Result<Transformed<Arc<dyn PhysicalExpr>>> {
    if expr.is::<LambdaExpr>() {
        return Ok(Transformed::no(expr));
    }
    let depends_on_this_side = self.depends_on_join_side(&expr, self.join_side)?;
    if !depends_on_this_side {
        return Ok(Transformed::no(expr));
    }
    let depends_on_other_side = self.depends_on_join_side(&expr, self.join_side.negate())?;
    if depends_on_other_side || is_volatile(&expr) {
        return expr.map_children(|expr| self.rewrite(expr));
    }
    // There is only a dependency on this side.
    // If this expression has no children, we do not push down, as it should already be a column reference
    if expr.children().is_empty() {
        return Ok(Transformed::no(expr));
    }
    // Otherwise, we push down a projection.
    let alias = self.alias_generator.next("join_proj_push_down");
    let idx = self.create_new_column(alias.clone(), expr)?;
    Ok(Transformed::yes(Arc::new(Column::new(&alias, idx)) as Arc<dyn PhysicalExpr>))
}

Note the two guards: an expression that depends on both sides is not pushed (it is genuinely a cross-product expression), and a volatile expression is not pushed (pushing random() > 0.5 would change how many times it is evaluated). The rewrite is a column-reference substitution: the subexpression becomes a new projected column on that side, and the filter refers to the new column.

    Before:
    NestedLoopJoinExec: filter=(upper(l.name) = r.name)
      ProjectionExec: l.*
      ProjectionExec: r.*

    After:
    NestedLoopJoinExec: filter=(join_proj_push_down_1@N = r.name)
      ProjectionExec: l.*, upper(l.name) AS join_proj_push_down_1
      ProjectionExec: r.*

The join filter is then minimized against the new schemas (minimize_join_filter, projection_pushdown.rs:244+), because the columns that were only used to compute the pushed-down expression are no longer needed by the filter at all.

Filter pushdown as a negotiation

The physical filter pushdown is not a rewrite; it is a protocol. Four steps, from the module doc (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 — for a FilterExec, that is to combine them back into a single ANDed predicate with conjunction(unhandled_filters) (datafusion/physical-plan/src/filter.rs:874-882). If the combined predicate is true, the FilterExec disappears entirely.

The bookkeeping across the recursion is where the bugs live, so it is isolated in FilteredVec (filter_pushdown.rs:609-705), which tracks the original index of every predicate through the filtering and result-mapping steps. Predicates that a node refuses to push (volatile ones) still have to be reported back to its parent as “not supported”, so the index mapping has to be exact. There are two rounds: FilterPushdown::new() runs before EnsureRequirements, and FilterPushdown::new_post_optimization() runs at the very end, because dynamic filters (produced by a TopK SortExec or by a hash join’s build side) hold references to specific source ExecutionPlans, and any later plan mutation could dangle them.

Distribution: adding shuffles, and knowing when not to

EnsureRequirements is the rule that inserts RepartitionExec and SortExec to satisfy the distribution and ordering requirements declared by each operator. The interesting part is the set of refusals. It removes distribution-only operators that are no longer needed (remove_dist_changing_operators, enforce_distribution.rs:1016+), and it refuses to add a round-robin repartition that would only increase the partition count without a reason (enable_round_robin_repartition, repartition_file_scans, target_partitions — enforce_distribution.rs:1524-1540).

The “bubble up the hash partitioning” trick is a good example of a rule that changes the shape of the plan to avoid a shuffle (enforce_distribution.rs:1907-1930). Instead of

Agg:
  Repartition (hash):
    Union:
      - Agg: Repartition (hash): Data
      - Agg: Repartition (hash): Data

it produces

Agg:
  Interleave:
    - Agg: Repartition (hash): Data
    - Agg: Repartition (hash): Data

InterleaveExec merges the partitions of N children into one partitioned output, letting a single hash-partitioned aggregate sit above the union instead of a per-branch aggregate plus a union of results. One shuffle instead of two.

JoinSelection (datafusion/physical-optimizer/src/join_selection.rs) does the statistics-driven work: it picks the build side of a hash join by comparing estimated sizes, and it converts a non-runnable plan (one with a pipeline-breaking join over an infinite source) into a runnable one. The swap decision is deliberately conservative (join_selection.rs:88-112): byte sizes first, row counts as a fallback, and no swap at all if neither is known. A heuristic that guesses when it has no data is worse than a heuristic that declines.## Trade-offs and traps

The physical optimizer may make the plan worse. ProjectionPushdown runs twice, and the second run happens after LimitPushdown and TopKRepartition have reshaped the plan. The rules are written to be shape-specific so this does not happen, but it is a real hazard.

Absorbing a projection can duplicate work. If a scan already has a projection containing an expensive expression, and the new projection references it twice, merging them would evaluate the expression twice. The scan refuses (datafusion/datasource/src/file_scan_config/mod.rs:688-715): a column is duplicated-and-kept only if the underlying expression is volatile or its ExpressionPlacement is not should_push_to_leaves(). ExpressionPlacement (datafusion/expr-common/src/placement.rs) is the taxonomy that makes this decidable: a Column or a cheap expression like struct_col['field'] may be pushed to the leaves; an expensive UDF is KeepInPlace.

Volatile functions are a hard stop. Pushing random() into a scan, or into a join side, changes the number of times it is evaluated and therefore the answer. The physical rules check is_volatile at every pushdown point.

Metadata is a correctness constraint, not a nicety. remove_unnecessary_projections refuses to touch a projection that “overrides metadata”, because swapping it with its child would change the metadata visible to the child’s expressions.

Reordering the pipeline is legal but lossy. The contract says the rules are reorderable, and also says that a rule may silently stop matching. The recommendation in the source is to keep the default order and use the configuration knobs instead — optimizer.enable_distinct_aggregation_soft_limit, optimizer.enable_round_robin_repartition, optimizer.join_reordering, optimizer.prefer_hash_join, optimizer.subset_repartition_threshold. Turn a knob off; do not reorder the pipeline.

The deepest lesson. Every trick in this file is the same trick in different clothing: find work that is being done per row or per pair of rows and move it somewhere it can be done once. The projection becomes a column index. The row index becomes a reader-provided offset. The join filter subexpression becomes a precomputed column. The filter becomes a predicate the scan can evaluate against statistics. The optimizer’s job is not to make the operators faster; it is to move the work up to a place where the operator can be smarter about it.