Opensource internals

No Row-by-Row Interpreter: How DataFusion's Physical Expression Tree Compiles Into Whole-Batch Arrow Kernel Calls

15 min read Jay Hubbert

DataFusion's physical expressions are a tree of typed nodes that dispatch once per batch into Arrow's precompiled SIMD kernels

TL;DR: The classic query-engine mistake is a row-at-a-time interpreter: for each row, walk the expression tree, switch on the operator, switch on the data type, compute. DataFusion does the opposite. The physical expression tree is walked once per RecordBatch, the switch on the operator happens once per node, and the work itself is delegated to Arrow’s precompiled, SIMD-vectorized kernels, which see the entire array at once. a < b is not a Rust for loop over rows — it is one call to arrow::compute::kernels::cmp::lt on two &dyn Datums.

The shape of the problem

A logical expression like

sql 
WHERE a + b > 100 AND c LIKE 'foo%'

has a tree of operators and leaves. Every engine has to answer: where does the work happen?

The naive design is a bytecode interpreter:

text 
for row in rows:                      # 1_000_000 iterations
    stack = []
    for op in bytecode:               # 7 iterations
        match op:
            ADD => stack.push(stack.pop() + stack.pop())
            LT  => stack.push(stack.pop() < stack.pop())
            ...

This is what every row-oriented engine did for decades, and it is catastrophically slow on modern CPUs. The inner loop is a match on a discriminant, the operands are boxed, the CPU branch predictor is doing nothing but predicting the match, and the memory access pattern is scattered across a heap-allocated value per intermediate result. None of that is vectorizable.

The fix that won is the “vectorized” or “column-at-a-time” design: don’t evaluate one row, evaluate one column. Compute the whole a + b column, then the whole > 100 column, then the whole AND. Every step is a tight, branch-free, SIMD-friendly loop over a dense buffer.

DataFusion’s actual design: a typed tree, dispatched once per batch

DataFusion’s physical expression is a trait:

rust 
// datafusion/physical-expr-common/src/physical_expr.rs:76-96
pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash {
    /// Get the data type of this expression, given the schema of the input.
    fn data_type(&self, input_schema: &Schema) -> Result<DataType> { ... }
    /// Determine whether this expression is nullable, given the schema of the input
    fn nullable(&self, input_schema: &Schema) -> Result<bool> { ... }
    /// Evaluate an expression against a RecordBatch
    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue>;
    ...
}

The return type is ColumnarValue, which has exactly two variants:

rust 
// datafusion/expr-common/src/columnar_value.rs
pub enum ColumnarValue {
    /// Array of values
    Array(ArrayRef),
    /// A single value
    Scalar(ScalarValue),
}

That enum is the whole abstraction. A node in the tree is asked to evaluate against a batch, and it answers with either “here is the whole column” or “here is one value that is the same for every row”. There is no per-row iteration anywhere in the trait.

A leaf: the physical Column

The simplest leaf does almost nothing:

rust 
// datafusion/physical-expr/src/expressions/column.rs:120-124
    /// Evaluate the expression
    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
        self.bounds_check(batch.schema().as_ref())?;
        Ok(ColumnarValue::Array(Arc::clone(batch.column(self.index))))
    }

Note what this is not: it does not copy the column, does not slice it, does not box it into a Value. It clones an Arc — a refcount bump — and returns a borrowed view of the batch’s buffer. The physical Column is resolved to a schema index at plan time, not a name:

rust 
// datafusion/physical-expr/src/expressions/column.rs:66-72
pub struct Column {
    /// The name of the column (used for debugging and display purposes)
    name: String,
    /// The index of the column in its schema
    index: usize,
}

The name is carried along only for Display and Debug. At runtime the lookup is batch.column(self.index) — an array index.

An internal node: BinaryExpr

BinaryExpr is where the “compile once, dispatch once” trick becomes visible. Its evaluate looks like this:

rust 
// datafusion/physical-expr/src/expressions/binary.rs:561-583
    fn evaluate(&self, batch: &RecordBatch) -> Result<ColumnarValue> {
        use arrow::compute::kernels::numeric::*;

        if self.op == Operator::And {
            return self.evaluate_conjunction(batch);
        }

        // Evaluate left-hand side expression.
        let lhs = self.left.evaluate(batch)?;

        // Check if we can apply short-circuit evaluation.
        match check_short_circuit(&lhs, &self.op) {
            ShortCircuitStrategy::None => {}
            ShortCircuitStrategy::ReturnLeft => return Ok(lhs),
            ...

and then the meat:

rust 
// datafusion/physical-expr/src/expressions/binary.rs:641-683
        match self.op {
            Operator::Plus if self.fail_on_overflow => return apply(&lhs, &rhs, add),
            Operator::Plus => return apply(&lhs, &rhs, add_wrapping),
            Operator::Minus if self.fail_on_overflow => return apply(&lhs, &rhs, sub),
            Operator::Minus => return apply(&lhs, &rhs, sub_wrapping),
            Operator::Multiply if self.fail_on_overflow => return apply(&lhs, &rhs, mul),
            Operator::Multiply => return apply(&lhs, &rhs, mul_wrapping),
            Operator::Divide => return apply(&lhs, &rhs, div),
            Operator::Modulo => return apply(&lhs, &rhs, rem),

            Operator::Eq
            | Operator::NotEq
            | Operator::Lt
            | Operator::Gt
            | Operator::LtEq
            | Operator::GtEq
            | Operator::IsDistinctFrom
            | Operator::IsNotDistinctFrom
            | Operator::LikeMatch
            | Operator::ILikeMatch
            | Operator::NotLikeMatch
            | Operator::NotILikeMatch => {
                return apply_cmp(self.op, &lhs, &rhs);
            }
            _ => {}
        }

Read that carefully. The match self.op is a Rust match on an enum discriminant. The compiler turns it into a jump table, or inlines the branches. It happens once, not once per row. And each arm names an Arrow kernel function — add, sub_wrapping, div, rem — which are monomorphized over the concrete numeric type inside Arrow.

apply_cmp does the same thing for comparisons, by turning the operator into a function pointer:

rust 
// datafusion/physical-expr-common/src/datum.rs:62-98
pub fn apply_cmp(
    op: Operator,
    lhs: &ColumnarValue,
    rhs: &ColumnarValue,
) -> Result<ColumnarValue> {
    if lhs.data_type().is_nested() {
        apply_cmp_for_nested(op, lhs, rhs)
    } else {
        let f = match op {
            Operator::Eq => eq,
            Operator::NotEq => neq,
            Operator::Lt => lt,
            Operator::LtEq => lt_eq,
            Operator::Gt => gt,
            Operator::GtEq => gt_eq,
            Operator::IsDistinctFrom => distinct,
            Operator::IsNotDistinctFrom => not_distinct,

            Operator::LikeMatch => like,
            Operator::ILikeMatch => ilike,
            Operator::NotLikeMatch => nlike,
            Operator::NotILikeMatch => nilike,

            _ => {
                return internal_err!("Invalid compare operator: {}", op);
            }
        };
        ...
        apply(&lhs, &rhs, |l, r| Ok(Arc::new(f(l, r)?)))
    }
}

f is a function pointer to an Arrow kernel — arrow_ord::cmp::lt and friends, which take &dyn Datum and return a BooleanArray. Selecting the function pointer happens once per evaluation of the node; the call itself processes the whole array.

Datum: the scalar/array duality, resolved inside Arrow

apply is the shim that maps Arrow’s Datum kernels onto ColumnarValue:

rust 
// datafusion/physical-expr-common/src/datum.rs:38-59
pub fn apply(
    lhs: &ColumnarValue,
    rhs: &ColumnarValue,
    f: impl Fn(&dyn Datum, &dyn Datum) -> Result<ArrayRef, ArrowError>,
) -> Result<ColumnarValue> {
    match (&lhs, &rhs) {
        (ColumnarValue::Array(left), ColumnarValue::Array(right)) => {
            Ok(ColumnarValue::Array(f(&left.as_ref(), &right.as_ref())?))
        }
        (ColumnarValue::Scalar(left), ColumnarValue::Array(right)) => Ok(
            ColumnarValue::Array(f(&left.to_scalar()?, &right.as_ref())?),
        ),
        (ColumnarValue::Array(left), ColumnarValue::Scalar(right)) => Ok(
            ColumnarValue::Array(f(&left.as_ref(), &right.to_scalar()?)?),
        ),
        (ColumnarValue::Scalar(left), ColumnarValue::Scalar(right)) => {
            let array = f(&left.to_scalar()?, &right.to_scalar()?)?;
            let scalar = ScalarValue::try_from_array(array.as_ref(), 0)?;
            Ok(ColumnarValue::Scalar(scalar))
        }
    }
}

Arrow’s Datum trait is the key to the scalar/array duality. It has exactly one method:

rust 
// arrow-array-60.0.0/src/scalar.rs:79-100
pub trait Datum {
    /// Returns the value for this [`Datum`] and a boolean indicating if the value is scalar
    fn get(&self) -> (&dyn Array, bool);
}

A scalar is just a one-element array plus a flag saying “treat me as a scalar”. Arrow’s kernels then have one code path for the array case and one for the broadcast case, and they are written once, in Arrow, with SIMD intrinsics — not in DataFusion, and not per-row.

So WHERE x = 42 does not materialize a 1-million-element array of 42s. 42 becomes a Scalar, Arrow’s eq kernel sees a scalar on one side and broadcasts inside its own loop.

Where the “compilation” actually happens

The word “compile” in the title is doing real work here, but the compilation is not code generation. It is resolution of everything that can be resolved at plan time, so that the runtime path has nothing left to decide.

Type coercion is resolved by the logical planner. The physical planner explicitly refuses to do it:

rust 
// datafusion/physical-expr/src/planner.rs:274-281
            // Note that the logical planner is responsible
            // for type coercion on the arguments (e.g. if one
            // argument was originally Int32 and one was
            // Int64 they will both be coerced to Int64).
            //
            // There should be no coercion during physical
            // planning.
            binary(lhs, *op, rhs, input_schema)

So by the time a BinaryExpr exists, its two children are guaranteed to have compatible types, and the runtime match self.op arm can call add/sub/etc. without any type check, any cast, any error path. When a coercion is needed, the logical type-coercion analyzer inserts an Expr::Cast node into the logical tree (datafusion/optimizer/src/analyzer/type_coercion.rs:535-570, via BinaryTypeCoercer::new(...).get_input_types()), and the physical planner lowers that node into a physical CastExpr:

rust 
// datafusion/physical-expr/src/planner.rs:387-404
        Expr::Cast(Cast { expr, field }) => expressions::cast_with_target_field(
            create_physical_expr(expr, input_dfschema, execution_props, planning_ctx)?,
            input_schema,
            field,
            None,
        ),
        Expr::TryCast(TryCast { expr, field }) => {
            expressions::try_cast_with_target_field(
                create_physical_expr(
                    expr,
                    input_dfschema,
                    execution_props,
                    planning_ctx,
                )?,
                input_schema,
                field,
            )
        }

So the cast is a node in the tree, evaluated once per batch by the CastExpr — never a per-row type check buried inside the arithmetic kernel.

The operator-to-kernel mapping is resolved once. The match op above is a compile-time-known set of arms. Arrow’s kernels are pub fn lt(lhs: &dyn Datum, rhs: &dyn Datum) -> Result<BooleanArray, ArrowError> — inside, they downcast once and then run a monomorphized loop.

Null handling is resolved inside the kernel. lt on two Int32Arrays knows how to combine two validity bitmaps, and it does it with bitwise AND over 64-bit words, not with a per-row if is_null.

The physical tree is a different tree from the logical one

This separation is not cosmetic; it is what makes the runtime path fast.

The logical Expr enum (datafusion/expr/src/expr.rs:327+) is a planning language:

rust 
pub enum Expr {
    Alias(Alias),
    Column(Column),                 // named, qualified, relation-aware
    ScalarVariable(FieldRef, Vec<String>),
    Literal(ScalarValue, Option<FieldMetadata>),
    BinaryExpr(BinaryExpr),
    ...
    Like(Like),
    SimilarTo(Like),
    Between(Between),
    Case(Case),
    Cast(Cast),
    TryCast(TryCast),
    InList(InList),
    ...
}

It refers to columns by name and qualifier, may contain scalar subqueries and correlated references, and its operators are un-coerced.

The physical PhysicalExpr is a execution language. It refers to columns by index, contains no subqueries (those are resolved into separate operators), and its operators are pre-coerced. The translation happens exactly once, in create_physical_expr:

rust 
// datafusion/physical-expr/src/planner.rs:133-168
pub fn create_physical_expr(
    e: &Expr,
    input_dfschema: &DFSchema,
    execution_props: &ExecutionProps,
    planning_ctx: &PhysicalPlanningContext,
) -> Result<Arc<dyn PhysicalExpr>> {
    let input_schema = input_dfschema.as_arrow();

    match e {
        ...
        Expr::Column(c) => {
            let idx = input_dfschema.index_of_column(c)?;
            Ok(Arc::new(Column::new(&c.name, idx)))
        }
        Expr::Literal(value, metadata) => Ok(Arc::new(Literal::new_with_metadata(
            value.clone(),
            metadata.clone(),
        ))),

That single function is the entire “compiler”. It walks the logical tree once and produces a tree of Arc<dyn PhysicalExpr> where:

  • Expr::Column (name + qualifier) becomes Column { name, index } — a resolved slot.
  • Expr::Literal becomes Literal, whose evaluate is a constant:
    rust 
    // datafusion/physical-expr/src/expressions/literal.rs:106-108
    fn evaluate(&self, _batch: &RecordBatch) -> Result<ColumnarValue> {
        Ok(ColumnarValue::Scalar(self.value.clone()))
    }
    A literal never allocates, never materializes a column. WHERE x = 42 hands 42 to Arrow as a scalar.
  • Expr::BinaryExpr becomes BinaryExpr, with coercion already applied.

Because the physical tree is a tree of trait objects (Arc<dyn PhysicalExpr>), the runtime walk is a virtual call per node per batch. For a 7-node expression on a 8192-row batch, that is 7 virtual calls for 8192 rows — a per-batch cost, not a per-row cost. This is the crucial distinction from the bytecode interpreter, and it is why the design scales.

Specialization at the node level, not just at the batch level

Once you’re evaluating whole arrays, a whole class of optimizations opens up that are impossible in a row interpreter.

Short-circuiting on the shape of the result, not row-by-row. BinaryExpr::evaluate inspects the boolean LHS array and, if the RHS is not needed at all, skips evaluating it entirely:

rust 
// datafusion/physical-expr/src/expressions/binary.rs:1285-1316
        ColumnarValue::Array(array) => {
            // Check Boolean arrays directly.
            if let Ok(bool_array) = as_boolean_array(array) {
                // NULLs still need the RHS.
                if bool_array.null_count() > 0 {
                    return ShortCircuitStrategy::None;
                }

                let len = bool_array.len();
                if len == 0 {
                    return ShortCircuitStrategy::None;
                }

                let true_count = bool_array.values().count_set_bits();
                if true_count == len {
                    return ShortCircuitStrategy::ReturnLeft;
                }

                if true_count == 0 {
                    return ShortCircuitStrategy::ReturnRight;
                }

                let false_count = len - true_count;
                if false_count as f32 / len as f32 <= PRE_SELECTION_THRESHOLD {
                    // Only false rows need the RHS; the LHS has no NULLs here.
                    let mask = BooleanArray::new(!bool_array.values(), None);
                    return ShortCircuitStrategy::PreSelection {
                        mask,
                        fill_value: true,
                    };
                }
            }
        }

count_set_bits() on the validity/values buffer tells you, in O(n/64) word operations, whether the whole column is true, all false, or mostly false. In the last case, DataFusion filters the batch down to just the false rows, evaluates the RHS on that smaller batch, and scatters the result back — trading a full-width RHS evaluation for a filtered one.

Nested AND flattening. evaluate_conjunction walks a chain of ANDs and filters between conjuncts, so a cheap first conjunct prunes the input for an expensive second one:

rust 
// datafusion/physical-expr/src/expressions/binary.rs:1210-1254
    /// Evaluates nested `AND`s together to avoid repeated filtering (#25035).
    ///
    /// When at most [`PRE_SELECTION_THRESHOLD`] of original rows remain,
    /// filters before conjuncts that may be costly or fail, so they never see
    /// rows nested evaluation would skip. `NULL` rows stay for a later `false`.
    fn evaluate_conjunction(&mut self, batch: &RecordBatch) -> Result<ColumnarValue> {

Array-vs-scalar kernel specialization. Some operators have a dedicated Arrow kernel for the array-scalar case that is materially faster than the general array-array one (regex matching, bitwise ops):

rust 
// datafusion/physical-expr/src/expressions/binary.rs:1142-1163
    fn evaluate_array_scalar(
        &self,
        array: &dyn Array,
        scalar: ScalarValue,
    ) -> Result<Option<Result<ArrayRef>>> {
        use Operator::*;
        let scalar_result = match &self.op {
            RegexMatch => regex_match_dyn_scalar(array, &scalar, false, false),
            RegexIMatch => regex_match_dyn_scalar(array, &scalar, false, true),
            ...
            // if scalar operation is not supported - fallback to array implementation
            _ => None,
        };

        Ok(scalar_result)
    }

Selection vectors instead of filtering. PhysicalExpr::evaluate_selection is a trait method that evaluates an expression only on the rows selected by a boolean mask, and scatters the results back:

rust 
// datafusion/physical-expr-common/src/physical_expr.rs:121-180
    fn evaluate_selection(
        &self,
        batch: &RecordBatch,
        selection: &BooleanArray,
    ) -> Result<ColumnarValue> {
        let row_count = batch.num_rows();
        ...
        // First, check if we can avoid filtering altogether.
        if selection.null_count() == 0 && !selection.has_false() {
            // All values from the `selection` filter are true and match the input batch.
            // No need to perform any filtering.
            return self.evaluate(batch);
        }

        // Next, prepare the result array for each 'true' row in the selection vector.
        let filtered_result = if !selection.has_true() {
            // Do not call `evaluate` when the selection is empty.
            // `evaluate_selection` is used to conditionally evaluate expressions.
            // When the expression in question is fallible, evaluating it with an empty
            // record batch may trigger a runtime error (e.g. division by zero).
            ...
            ColumnarValue::Array(new_empty_array(&datatype))
        } else {
            // If we reach this point, there's no other option than to filter the batch.
            // This is a fairly costly operation since it requires creating partial copies
            // (worst case of length `row_count - 1`) of all the arrays in the record batch.
            let filtered_batch = filter_record_batch(batch, selection)?;
            self.evaluate(&filtered_batch)?
        };

        // Finally, scatter the filtered result array so that the indices match the input rows again.
        match &filtered_result {
            ColumnarValue::Array(a) => {
                scatter(selection, a.as_ref()).map(ColumnarValue::Array)
            }
            ...

This is the vectorized analogue of “don’t evaluate 1/0 for rows that will be discarded”. A row interpreter would need a per-row branch; a columnar evaluator filters the batch, evaluates the cheap subset, and scatters.

CASE becomes a sequence of filtered batches. The CaseExpr evaluates one branch at a time on the remaining rows, which shrink as branches match:

rust 
// datafusion/physical-expr/src/expressions/case.rs:864-890
            // If the 'when' predicate did not match any rows, continue to the next branch immediately.
            // Only counts valid slots that are true (masked-null predicate slots are ignored),
            // so no `prep_null_mask_filter` needed here.
            if !when_value.has_true() {
                continue;
            }

            // If the 'when' predicate matched all remaining rows, there is no need to filter
            if when_value.null_count() == 0 && !when_value.has_false() {
                let then_expression = &self.when_then_expr[i].1;
                let then_value = then_expression.evaluate(&remainder_batch)?;
                result_builder.add_branch_result(&remainder_rows, then_value)?;
                return result_builder.finish();
            }

            // Filter the remainder batch based on the 'when' value
            // This results in a batch containing only the rows that need to be evaluated
            // for the current branch

There is even a fully-memoized specialization: CASE WHEN <literal> THEN <literal> ... becomes a lookup table over the values (datafusion/physical-expr/src/expressions/case/literal_lookup_table/).

Trade-offs and failure modes

  • The virtual call per node per batch is not free. For a batch of 8192 rows and a 20-node expression, that’s 20 dynamic dispatches — utterly negligible. But for tiny batches (say 8 rows) the fixed overhead of the walk can dominate. DataFusion mitigates this by coalescing small batches upstream (see the batch-merging article in this series), which raises the row count per evaluation.

  • The tree must be re-planned for schema changes. Because Column is an index, inserting a column into the batch invalidates the whole physical expression tree. with_new_schema in datafusion/physical-expr/src/expressions/column.rs:246-266 exists precisely to re-index the tree, and it has to be called at every point where the input schema shifts (projections, joins). The alternative — name-based lookup at runtime — would be catastrophically slower.

  • Filters copy data. filter_record_batch makes partial copies of every column in the batch, as the doc comment above says: “worst case of length row_count - 1”. This is the cost of the “evaluate fewer rows” optimizations. DataFusion guards it with the PRE_SELECTION_THRESHOLD (20%) check so it only filters when it is clearly worth it, and with is_cheap_and_infallible to avoid filtering before a conjunct that cannot fail.

  • ColumnarValue::Scalar is a footgun. The doc on into_array says plainly that converting a scalar to an array “is not as efficient as handling the scalar value directly”. A UDF that eagerly calls into_array on all its arguments throws away the entire scalar broadcast optimization. The ScalarFunctionExpr wrapper at least checks the result length:

    rust 
    // datafusion/physical-expr/src/scalar_function.rs:263-279
        if let ColumnarValue::Array(array) = &output
            && array.len() != batch.num_rows()
        {
            // If the arguments are a non-empty slice of scalar values, we can assume that
            // returning a one-element array is equivalent to returning a scalar.
            let preserve_scalar = array.len() == 1 && !input_empty && input_all_scalar;
            return if preserve_scalar {
                ScalarValue::try_from_array(array, 0).map(ColumnarValue::Scalar)
            } else {
                internal_err!(...)
            };
        }

  • Kernel coverage gaps leak into the physical layer. time ± interval returns a wrapped time (PostgreSQL/DuckDB semantics) which Arrow’s arithmetic kernels do not implement, so DataFusion has to implement it itself in apply_time_interval (datafusion/physical-expr/src/expressions/binary.rs:395-433). Every such gap is a hand-written kernel that does not get Arrow’s SIMD treatment, and it is a maintenance liability.

  • The physical tree is not the logical tree. Anyone reasoning about “why is this query slow” has to remember that the expression they wrote is not the expression that runs. The logical planner may have rewritten it, the physical planner may have inserted casts, and the physical expression tree may have been re-indexed. This is a genuine source of confusion, and it’s why DataFusion has fmt_sql on PhysicalExpr so you can print the thing that will actually execute.

The takeaway

DataFusion’s expression engine is not an interpreter, and it is not a compiler either. It is a typed tree of node objects, walked once per batch, where each node resolves its operator to an Arrow kernel and hands over whole arrays. All the “deciding” work — name resolution, type coercion, operator selection, null semantics, SIMD details — is pushed either to plan time or into Arrow’s precompiled kernels, so that the only thing left at runtime is a handful of virtual calls and a straight-line call into a vectorized function. The result is that the per-row cost of expression interpretation is essentially zero, and all the time is spent in the kernels doing actual arithmetic on dense buffers.

DatafusionArrowPerformanceInternals