feat: add ExpressionPlacement enum for optimizer expression placement decisions (#20065)

## Summary

This PR is part of work towards
#19387

Extracts the `ExpressionPlacement` enum from #20036 to
provide a mechanism for expressions to indicate where they should be
placed in the query plan for optimal execution.

I've opted to go the route of having expressions declare their behavior
via a new API on `enum Expr` and `trait PhysicalExpr`:

```rust
enum Expr {
    pub fn placement(&self) -> ExpressionPlacement { ... }
   ...
}
```
And:

```rust
trait PhysicalExpr {
   fn placement(&self) -> ExpressionPlacement { ... }
}
```

Where `ExpressionPlacement`:

```rust
enum ExpressionPlacement {
    /// Argument is a literal constant value or an expression that can be
    /// evaluated to a constant at planning time.
    Literal,
    /// Argument is a simple column reference.
    Column,
    /// Argument is a complex expression that can be safely placed at leaf nodes.
    /// For example, if `get_field(struct_col, 'field_name')` is implemented as a
    /// leaf-pushable expression, then it would return this variant.
    /// Then `other_leaf_function(get_field(...), 42)` could also be classified as
    /// leaf-pushable using the knowledge that `get_field(...)` is leaf-pushable.
    PlaceAtLeaves,
    /// Argument is a complex expression that should be placed at root nodes.
    /// For example, `min(col1 + col2)` is not leaf-pushable because it requires per-row computation.
    PlaceAtRoot,
}
```

We arrived at `ExprPlacement` after iterating through a version that
had:

```rust
enum ArgTriviality {
    Literal,
    Column,
    Trivial,
    NonTrivial,
}
```

This terminology came from existing concepts in the codebase that were
sprinkled around various places in the logical and physical layers. Some
examples:


https://github.com/apache/datafusion/blob/f819061833d0ee4d7899ed6a0a431c584533b241/datafusion/physical-plan/src/projection.rs#L282-L290


https://github.com/apache/datafusion/blob/f819061833d0ee4d7899ed6a0a431c584533b241/datafusion/physical-plan/src/projection.rs#L1120-L1125


https://github.com/apache/datafusion/blob/f819061833d0ee4d7899ed6a0a431c584533b241/datafusion/optimizer/src/optimize_projections/mod.rs#L589-L592

The new API adds the nuance / distinction of the case of `get_field(col,
'a')` where it is neither a column nor a literal but it is trivial.

It also gives scalar functions the ability to classify themselves.
This part was a bit tricky because `ScalarUDFImpl` (the scalar function
trait that users implement) lives in `datafuions-expr` which cannot have
references to `datafusion-physical-expr-common` (where `PhysicalExpr` is
defined).
But once we are in the physical layer scalar functions are represented
as `func: ScalarUDFImpl, args: Vec<Arc<dyn PhysicalExpr>>`.
And since we can't have a trait method referencing `PhysicalExpr` there
would be no way to ask a function to classify itself in the physical
layer.

Additionally even if we could refer to `PhysicalExpr` from the
`ScalarUDFImpl` trait we would then need 2 methods with similar but
divergent logic (match on the `Expr` enum in one, downcast to various
known types in the physical version) that adds boilerplate for
implementers.

The `ExprPlacement` enum solves this problem: we can have a single
method `ScalarUDFImpl::placement(args: &[ExpressionPlacement])`.
The parent of `ScalarUDFImpl` will call either `Expr::placement` or
`PhysicalExpr::placement` depending on which one it has.

## Changes

- Add `ExpressionPlacement` enum in `datafusion-expr-common` with four
variants:
  - `Literal` - constant values
  - `Column` - simple column references
- `PlaceAtLeaves` - cheap expressions (like `get_field`) that can be
pushed to leaf nodes
  - `PlaceAtRoot` - expensive expressions that should stay at root

- Add `placement()` method to:
  - `Expr` enum
- `ScalarUDF` / `ScalarUDFImpl` traits (with default returning
`PlaceAtRoot`)
  - `PhysicalExpr` trait (with default returning `PlaceAtRoot`)
- Physical expression implementations for `Column`, `Literal`, and
`ScalarFunctionExpr`

- Implement `placement()` for `GetFieldFunc` that returns
`PlaceAtLeaves` when accessing struct fields with literal keys

- Replace `is_expr_trivial()` function checks with `placement()` checks
in:
  - `datafusion/optimizer/src/optimize_projections/mod.rs`
  - `datafusion/physical-plan/src/projection.rs`

## Test Plan

- [x] `cargo check` passes on all affected packages
- [x] `cargo test -p datafusion-optimizer` passes
- [x] `cargo test -p datafusion-physical-plan` passes (except unrelated
zstd feature test)
- [x] `cargo test -p datafusion-functions --lib getfield` passes

🤖 Generated with [Claude Code](https://claude.ai/code)

---------

Co-authored-by: Claude Opus 4.5 <noreply@anthropic.com>

Author: adriangb

datafusion/expr-common/src/lib.rs

@@ -40,7 +40,10 @@ pub mod dyn_eq;
 pub mod groups_accumulator;
 pub mod interval_arithmetic;
 pub mod operator;
+pub mod placement;
 pub mod signature;
 pub mod sort_properties;
 pub mod statistics;
 pub mod type_coercion;
+
+pub use placement::ExpressionPlacement;

datafusion/expr-common/src/placement.rs

@@ -0,0 +1,62 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Expression placement information for optimization decisions.
+
+/// Describes where an expression should be placed in the query plan for
+/// optimal execution. This is used by optimizers to make decisions about
+/// expression placement, such as whether to push expressions down through
+/// projections.
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum ExpressionPlacement {
+    /// A constant literal value.
+    Literal,
+    /// A simple column reference.
+    Column,
+    /// A cheap expression that can be pushed to leaf nodes in the plan.
+    /// Examples include `get_field` for struct field access.
+    /// Pushing these expressions down in the plan can reduce data early
+    /// at low compute cost.
+    /// See [`ExpressionPlacement::should_push_to_leaves`] for details.
+    MoveTowardsLeafNodes,
+    /// An expensive expression that should stay where it is in the plan.
+    /// Examples include complex scalar functions or UDFs.
+    KeepInPlace,
+}
+
+impl ExpressionPlacement {
+    /// Returns true if the expression can be pushed down to leaf nodes
+    /// in the query plan.
+    ///
+    /// This returns true for:
+    /// - [`ExpressionPlacement::Column`]: Simple column references can be pushed down. They do no compute and do not increase or
+    ///   decrease the amount of data being processed.
+    ///   A projection that reduces the number of columns can eliminate unnecessary data early,
+    ///   but this method only considers one expression at a time, not a projection as a whole.
+    /// - [`ExpressionPlacement::MoveTowardsLeafNodes`]: Cheap expressions can be pushed down to leaves to take advantage of
+    ///   early computation and potential optimizations at the data source level.
+    ///   For example `struct_col['field']` is cheap to compute (just an Arc clone of the nested array for `'field'`)
+    ///   and thus can reduce data early in the plan at very low compute cost.
+    ///   It may even be possible to eliminate the expression entirely if the data source can project only the needed field
+    ///   (as e.g. Parquet can).
+    pub fn should_push_to_leaves(&self) -> bool {
+        matches!(
+            self,
+            ExpressionPlacement::Column | ExpressionPlacement::MoveTowardsLeafNodes
+        )
+    }
+}

datafusion/expr/src/expr.rs

@@ -38,6 +38,7 @@ use datafusion_common::tree_node::{
 use datafusion_common::{
     Column, DFSchema, HashMap, Result, ScalarValue, Spans, TableReference,
 };
+use datafusion_expr_common::placement::ExpressionPlacement;
 use datafusion_functions_window_common::field::WindowUDFFieldArgs;
 #[cfg(feature = "sql")]
 use sqlparser::ast::{
@@ -1536,6 +1537,23 @@ impl Expr {
         }
     }
 
+    /// Returns placement information for this expression.
+    ///
+    /// This is used by optimizers to make decisions about expression placement,
+    /// such as whether to push expressions down through projections.
+    pub fn placement(&self) -> ExpressionPlacement {
+        match self {
+            Expr::Column(_) => ExpressionPlacement::Column,
+            Expr::Literal(_, _) => ExpressionPlacement::Literal,
+            Expr::ScalarFunction(func) => {
+                let arg_placements: Vec<_> =
+                    func.args.iter().map(|arg| arg.placement()).collect();
+                func.func.placement(&arg_placements)
+            }
+            _ => ExpressionPlacement::KeepInPlace,
+        }
+    }
+
     /// Return String representation of the variant represented by `self`
     /// Useful for non-rust based bindings
     pub fn variant_name(&self) -> &str {

datafusion/expr/src/lib.rs

@@ -95,6 +95,7 @@ pub use datafusion_expr_common::accumulator::Accumulator;
 pub use datafusion_expr_common::columnar_value::ColumnarValue;
 pub use datafusion_expr_common::groups_accumulator::{EmitTo, GroupsAccumulator};
 pub use datafusion_expr_common::operator::Operator;
+pub use datafusion_expr_common::placement::ExpressionPlacement;
 pub use datafusion_expr_common::signature::{
     ArrayFunctionArgument, ArrayFunctionSignature, Coercion, Signature,
     TIMEZONE_WILDCARD, TypeSignature, TypeSignatureClass, Volatility,

datafusion/expr/src/udf.rs

@@ -31,6 +31,7 @@ use datafusion_common::config::ConfigOptions;
 use datafusion_common::{ExprSchema, Result, ScalarValue, not_impl_err};
 use datafusion_expr_common::dyn_eq::{DynEq, DynHash};
 use datafusion_expr_common::interval_arithmetic::Interval;
+use datafusion_expr_common::placement::ExpressionPlacement;
 use std::any::Any;
 use std::cmp::Ordering;
 use std::fmt::Debug;
@@ -361,6 +362,13 @@ impl ScalarUDF {
     pub fn as_async(&self) -> Option<&AsyncScalarUDF> {
         self.inner().as_any().downcast_ref::<AsyncScalarUDF>()
     }
+
+    /// Returns placement information for this function.
+    ///
+    /// See [`ScalarUDFImpl::placement`] for more details.
+    pub fn placement(&self, args: &[ExpressionPlacement]) -> ExpressionPlacement {
+        self.inner.placement(args)
+    }
 }
 
 impl<F> From<F> for ScalarUDF
@@ -964,6 +972,20 @@ pub trait ScalarUDFImpl: Debug + DynEq + DynHash + Send + Sync {
     fn documentation(&self) -> Option<&Documentation> {
         None
     }
+
+    /// Returns placement information for this function.
+    ///
+    /// This is used by optimizers to make decisions about expression placement,
+    /// such as whether to push expressions down through projections.
+    ///
+    /// The default implementation returns [`ExpressionPlacement::KeepInPlace`],
+    /// meaning the expression should be kept where it is in the plan.
+    ///
+    /// Override this method to indicate that the function can be pushed down
+    /// closer to the data source.
+    fn placement(&self, _args: &[ExpressionPlacement]) -> ExpressionPlacement {
+        ExpressionPlacement::KeepInPlace
+    }
 }
 
 /// ScalarUDF that adds an alias to the underlying function. It is better to
@@ -1091,6 +1113,10 @@ impl ScalarUDFImpl for AliasedScalarUDFImpl {
     fn documentation(&self) -> Option<&Documentation> {
         self.inner.documentation()
     }
+
+    fn placement(&self, args: &[ExpressionPlacement]) -> ExpressionPlacement {
+        self.inner.placement(args)
+    }
 }
 
 #[cfg(test)]

datafusion/functions/src/core/getfield.rs

@@ -33,8 +33,8 @@ use datafusion_common::{
 use datafusion_expr::expr::ScalarFunction;
 use datafusion_expr::simplify::ExprSimplifyResult;
 use datafusion_expr::{
-    ColumnarValue, Documentation, Expr, ReturnFieldArgs, ScalarFunctionArgs, ScalarUDF,
-    ScalarUDFImpl, Signature, Volatility,
+    ColumnarValue, Documentation, Expr, ExpressionPlacement, ReturnFieldArgs,
+    ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature, Volatility,
 };
 use datafusion_macros::user_doc;
 
@@ -499,6 +499,32 @@ impl ScalarUDFImpl for GetFieldFunc {
     fn documentation(&self) -> Option<&Documentation> {
         self.doc()
     }
+
+    fn placement(&self, args: &[ExpressionPlacement]) -> ExpressionPlacement {
+        // get_field can be pushed to leaves if:
+        // 1. The base (first arg) is a column or already placeable at leaves
+        // 2. All field keys (remaining args) are literals
+        if args.is_empty() {
+            return ExpressionPlacement::KeepInPlace;
+        }
+
+        let base_placement = args[0];
+        let base_is_pushable = matches!(
+            base_placement,
+            ExpressionPlacement::Column | ExpressionPlacement::MoveTowardsLeafNodes
+        );
+
+        let all_keys_are_literals = args
+            .iter()
+            .skip(1)
+            .all(|p| matches!(p, ExpressionPlacement::Literal));
+
+        if base_is_pushable && all_keys_are_literals {
+            ExpressionPlacement::MoveTowardsLeafNodes
+        } else {
+            ExpressionPlacement::KeepInPlace
+        }
+    }
 }
 
 #[cfg(test)]
@@ -542,4 +568,92 @@ mod tests {
 
         Ok(())
     }
+
+    #[test]
+    fn test_placement_literal_key() {
+        let func = GetFieldFunc::new();
+
+        // get_field(col, 'literal') -> leaf-pushable (static field access)
+        let args = vec![ExpressionPlacement::Column, ExpressionPlacement::Literal];
+        assert_eq!(
+            func.placement(&args),
+            ExpressionPlacement::MoveTowardsLeafNodes
+        );
+
+        // get_field(col, 'a', 'b') -> leaf-pushable (nested static field access)
+        let args = vec![
+            ExpressionPlacement::Column,
+            ExpressionPlacement::Literal,
+            ExpressionPlacement::Literal,
+        ];
+        assert_eq!(
+            func.placement(&args),
+            ExpressionPlacement::MoveTowardsLeafNodes
+        );
+
+        // get_field(get_field(col, 'a'), 'b') represented as MoveTowardsLeafNodes for base
+        let args = vec![
+            ExpressionPlacement::MoveTowardsLeafNodes,
+            ExpressionPlacement::Literal,
+        ];
+        assert_eq!(
+            func.placement(&args),
+            ExpressionPlacement::MoveTowardsLeafNodes
+        );
+    }
+
+    #[test]
+    fn test_placement_column_key() {
+        let func = GetFieldFunc::new();
+
+        // get_field(col, other_col) -> NOT leaf-pushable (dynamic per-row lookup)
+        let args = vec![ExpressionPlacement::Column, ExpressionPlacement::Column];
+        assert_eq!(func.placement(&args), ExpressionPlacement::KeepInPlace);
+
+        // get_field(col, 'a', other_col) -> NOT leaf-pushable (dynamic nested lookup)
+        let args = vec![
+            ExpressionPlacement::Column,
+            ExpressionPlacement::Literal,
+            ExpressionPlacement::Column,
+        ];
+        assert_eq!(func.placement(&args), ExpressionPlacement::KeepInPlace);
+    }
+
+    #[test]
+    fn test_placement_root() {
+        let func = GetFieldFunc::new();
+
+        // get_field(root_expr, 'literal') -> NOT leaf-pushable
+        let args = vec![
+            ExpressionPlacement::KeepInPlace,
+            ExpressionPlacement::Literal,
+        ];
+        assert_eq!(func.placement(&args), ExpressionPlacement::KeepInPlace);
+
+        // get_field(col, root_expr) -> NOT leaf-pushable
+        let args = vec![
+            ExpressionPlacement::Column,
+            ExpressionPlacement::KeepInPlace,
+        ];
+        assert_eq!(func.placement(&args), ExpressionPlacement::KeepInPlace);
+    }
+
+    #[test]
+    fn test_placement_edge_cases() {
+        let func = GetFieldFunc::new();
+
+        // Empty args -> NOT leaf-pushable
+        assert_eq!(func.placement(&[]), ExpressionPlacement::KeepInPlace);
+
+        // Just base, no key -> MoveTowardsLeafNodes (not a valid call but should handle gracefully)
+        let args = vec![ExpressionPlacement::Column];
+        assert_eq!(
+            func.placement(&args),
+            ExpressionPlacement::MoveTowardsLeafNodes
+        );
+
+        // Literal base with literal key -> NOT leaf-pushable (would be constant-folded)
+        let args = vec![ExpressionPlacement::Literal, ExpressionPlacement::Literal];
+        assert_eq!(func.placement(&args), ExpressionPlacement::KeepInPlace);
+    }
 }

datafusion/optimizer/src/common_subexpr_eliminate.rs

@@ -702,6 +702,11 @@ impl CSEController for ExprCSEController<'_> {
         #[expect(deprecated)]
         let is_normal_minus_aggregates = matches!(
             node,
+            // TODO: there's an argument for removing `Literal` from here,
+            // maybe using `Expr::placemement().should_push_to_leaves()` instead
+            // so that we extract common literals and don't broadcast them to num_batch_rows multiple times.
+            // However that currently breaks things like `percentile_cont()` which expect literal arguments
+            // (and would instead be getting `col(__common_expr_n)`).
             Expr::Literal(..)
                 | Expr::Column(..)
                 | Expr::ScalarVariable(..)

datafusion/optimizer/src/optimize_projections/mod.rs

@@ -525,15 +525,14 @@ fn merge_consecutive_projections(proj: Projection) -> Result<Transformed<Project
     expr.iter()
         .for_each(|expr| expr.add_column_ref_counts(&mut column_referral_map));
 
-    // If an expression is non-trivial and appears more than once, do not merge
+    // If an expression is non-trivial (KeepInPlace) and appears more than once, do not merge
     // them as consecutive projections will benefit from a compute-once approach.
     // For details, see: https://github.com/apache/datafusion/issues/8296
     if column_referral_map.into_iter().any(|(col, usage)| {
         usage > 1
-            && !is_expr_trivial(
-                &prev_projection.expr
-                    [prev_projection.schema.index_of_column(col).unwrap()],
-            )
+            && !prev_projection.expr[prev_projection.schema.index_of_column(col).unwrap()]
+                .placement()
+                .should_push_to_leaves()
     }) {
         // no change
         return Projection::try_new_with_schema(expr, input, schema).map(Transformed::no);
@@ -586,11 +585,6 @@ fn merge_consecutive_projections(proj: Projection) -> Result<Transformed<Project
     }
 }
 
-// Check whether `expr` is trivial; i.e. it doesn't imply any computation.
-fn is_expr_trivial(expr: &Expr) -> bool {
-    matches!(expr, Expr::Column(_) | Expr::Literal(_, _))
-}
-
 /// Rewrites a projection expression using the projection before it (i.e. its input)
 /// This is a subroutine to the `merge_consecutive_projections` function.
 ///

datafusion/physical-expr-common/src/physical_expr.rs

@@ -35,6 +35,7 @@ use datafusion_common::{
 };
 use datafusion_expr_common::columnar_value::ColumnarValue;
 use datafusion_expr_common::interval_arithmetic::Interval;
+use datafusion_expr_common::placement::ExpressionPlacement;
 use datafusion_expr_common::sort_properties::ExprProperties;
 use datafusion_expr_common::statistics::Distribution;
 
@@ -430,6 +431,16 @@ pub trait PhysicalExpr: Any + Send + Sync + Display + Debug + DynEq + DynHash {
     fn is_volatile_node(&self) -> bool {
         false
     }
+
+    /// Returns placement information for this expression.
+    ///
+    /// This is used by optimizers to make decisions about expression placement,
+    /// such as whether to push expressions down through projections.
+    ///
+    /// The default implementation returns [`ExpressionPlacement::KeepInPlace`].
+    fn placement(&self) -> ExpressionPlacement {
+        ExpressionPlacement::KeepInPlace
+    }
 }
 
 #[deprecated(

datafusion/physical-expr/src/expressions/column.rs

@@ -30,6 +30,7 @@ use arrow::{
 use datafusion_common::tree_node::{Transformed, TreeNode};
 use datafusion_common::{Result, internal_err, plan_err};
 use datafusion_expr::ColumnarValue;
+use datafusion_expr_common::placement::ExpressionPlacement;
 
 /// Represents the column at a given index in a RecordBatch
 ///
@@ -146,6 +147,10 @@ impl PhysicalExpr for Column {
     fn fmt_sql(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(f, "{}", self.name)
     }
+
+    fn placement(&self) -> ExpressionPlacement {
+        ExpressionPlacement::Column
+    }
 }
 
 impl Column {