Skip to content

Allow struct field access projections to be pushed down into scans#19538

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
adriangb wants to merge 4 commits into from
Closed

Allow struct field access projections to be pushed down into scans #19538

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9641322
feat: Allow struct field access projections to be pushed down into scans
adriangb Jan 26, 2026
2d80d7f
rename placement_with_args -> placement
adriangb Jan 27, 2026
1a6e536
fix typos
adriangb Jan 27, 2026
9f0bd84
fix more typo
adriangb Jan 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions Cargo.lock
View file
Open in desktop

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 3 additions & 0 deletions datafusion/expr-common/src/lib.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -41,7 +41,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;
68 changes: 68 additions & 0 deletions datafusion/expr-common/src/placement.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,68 @@
// 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 classification for scalar functions.
//!
//! This module determines where in the query plan expressions should be placed
//! to optimize data flow:
//!
//! - **Leaf placement**: Cheap expressions (field accessors, column references)
//! are pushed down to leaf nodes (near data sources) to reduce data volume early.
//! - **Root placement**: Expensive expressions (computations, aggregates) are kept
//! at root nodes (after filtering) to operate on less data.

/// Classification of expression placement for scalar functions.
///
/// This enum is used by [`ScalarUDFImpl::placement`] to allow
/// functions to make context-dependent decisions about where they should
/// be placed in the query plan based on the nature of their arguments.
///
/// For example, `get_field(struct_col, 'field_name')` is
/// leaf-pushable (static field lookup), but `string_col like '%foo%'`
/// performs expensive per-row computation and should be placed
/// as further up the tree so that it can be run after filtering, sorting, etc.
///
/// # Why not pass in expressions directly to decide placement?
///
/// There are two reasons for using this enum instead of passing in the full expressions:
///
/// 1. **ScalarUDFImpl cannot reference PhysicalExpr**: The trait is defined in datafusion-expr,
/// which cannot reference datafusion-physical-expr since the latter depends on the former
/// (it would create a circular dependency).
/// 2. **Simplicity**: Without this enum abstracting away logical / physical distinctions,
/// we would need two distinct methods on ScalarUDFImpl: one for logical expression placement
/// and one for physical expression placement. This would require implementors to duplicate logic
/// and increases complexity for UDF authors.
///
/// [`ScalarUDFImpl::placement`]: https://docs.rs/datafusion-expr/latest/datafusion_expr/trait.ScalarUDFImpl.html#tymethod.placement
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub 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,
}
24 changes: 23 additions & 1 deletion datafusion/expr/src/expr.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -27,7 +27,7 @@ use std::sync::Arc;
use crate::expr_fn::binary_expr;
use crate::function::WindowFunctionSimplification;
use crate::logical_plan::Subquery;
use crate::{AggregateUDF, Volatility};
use crate::{AggregateUDF, ExpressionPlacement, Volatility};
use crate::{ExprSchemable, Operator, Signature, WindowFrame, WindowUDF};

use arrow::datatypes::{DataType, Field, FieldRef};
Expand DownExpand Up@@ -1933,6 +1933,28 @@ impl Expr {
}
}

/// Returns the placement classification of this expression.
///
/// This tells us if optimizers should preferentially
/// move this expression towards the leaves of the execution plan
/// tree (for cheap expressions or expressions that reduce the data size)
/// or towards the root of the execution plan tree (for expensive expressions
/// that should be run after filtering or parallelization, or expressions that increase the data size).
pub fn placement(&self) -> ExpressionPlacement {
match self {
Expr::Column(_) => ExpressionPlacement::Column,
Expr::Literal(_, _) => ExpressionPlacement::Literal,
Expr::ScalarFunction(func) => {
// Classify each argument's placement for context-aware decision making
let arg_placements: Vec<ExpressionPlacement> =
func.args.iter().map(|arg| arg.placement()).collect();

func.func.placement(&arg_placements)
}
_ => ExpressionPlacement::PlaceAtRoot,
}
}

/// Return all references to columns in this expression.
///
/// # Example
Expand Down
1 change: 1 addition & 0 deletions datafusion/expr/src/lib.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -92,6 +92,7 @@ pub use datafusion_doc::{
DocSection, Documentation, DocumentationBuilder, aggregate_doc_sections,
scalar_doc_sections, window_doc_sections,
};
pub use datafusion_expr_common::ExpressionPlacement;
pub use datafusion_expr_common::accumulator::Accumulator;
pub use datafusion_expr_common::columnar_value::ColumnarValue;
pub use datafusion_expr_common::groups_accumulator::{EmitTo, GroupsAccumulator};
Expand Down
49 changes: 49 additions & 0 deletions datafusion/expr/src/udf.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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;
Expand DownExpand Up@@ -123,6 +124,19 @@ impl ScalarUDF {
Self { inner: fun }
}

/// Returns the placement classification of this function given its arguments' placement.
///
/// This allows functions to make context-dependent decisions about where they should
/// be placed in the query plan. For example, `get_field(struct_col, 'field_name')` is
/// leaf-pushable (static field lookup), but `string_col like '%foo%'`
/// performs expensive per-row computation and should be placed
/// as further up the tree so that it can be run after filtering, sorting, etc.
///
/// See [`ScalarUDFImpl::placement`] for more details.
pub fn placement(&self, args: &[ExpressionPlacement]) -> ExpressionPlacement {
self.inner.placement(args)
}

/// Return the underlying [`ScalarUDFImpl`] trait object for this function
pub fn inner(&self) -> &Arc<dyn ScalarUDFImpl> {
&self.inner
Expand DownExpand Up@@ -885,6 +899,37 @@ pub trait ScalarUDFImpl: Debug + DynEq + DynHash + Send + Sync {
fn documentation(&self) -> Option<&Documentation> {
None
}

/// Returns the placement classification of this function given its arguments' placement.
///
/// This method allows functions to make context-dependent decisions about
/// where they should be placed in the query plan. The default implementation
/// returns [`ExpressionPlacement::PlaceAtRoot`] (conservative default).
///
/// Leaf-pushable functions are lightweight accessor functions like `get_field`
/// (struct field access) that simply access nested data within a column
/// without significant computation.
/// These can be pushed down to leaf nodes near data sources to reduce data volume early in the plan.
///
/// [`ExpressionPlacement::PlaceAtRoot`] represents expressions that should be kept after filtering,
/// such as expensive computations or aggregates that benefit from operating
/// on fewer rows.
///
/// # Example
///
/// - `get_field(struct_col, 'field_name')` with a literal key is leaf-pushable as it
/// performs metadata only (cheap) extraction of a sub-array from a struct column.
/// Thus, it can be placed near the data source to minimize data early.
/// - `string_col like '%foo%'` performs expensive per-row computation and should be placed
/// further up the tree so that it can be run after filtering, sorting, etc.
///
/// # Arguments
///
/// * `args` - Classification of each argument's placement, collected from the expression tree
/// by the caller.
fn placement(&self, _args: &[ExpressionPlacement]) -> ExpressionPlacement {
ExpressionPlacement::PlaceAtRoot
}
}

/// ScalarUDF that adds an alias to the underlying function. It is better to
Expand DownExpand Up@@ -1012,6 +1057,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)]
Expand Down
111 changes: 109 additions & 2 deletions datafusion/functions/src/core/getfield.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -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;

Expand DownExpand Up@@ -499,6 +499,37 @@ impl ScalarUDFImpl for GetFieldFunc {
fn documentation(&self) -> Option<&Documentation> {
self.doc()
}

fn placement(&self, args: &[ExpressionPlacement]) -> ExpressionPlacement {
// get_field is leaf-pushable if:
// 1. The struct/map argument is a Column or PlaceAtLeaves (not Literal or PlaceAtRoot)
// 2. All key arguments are literals (static field access, not dynamic per-row lookup)
//
// Literal base is not considered leaf-pushable because it would be constant-folded anyway.
if args.is_empty() {
return ExpressionPlacement::PlaceAtRoot;
}

// Check if the base (struct/map) argument is Column or PlaceAtLeaves
if !matches!(
args[0],
ExpressionPlacement::Column | ExpressionPlacement::PlaceAtLeaves
) {
return ExpressionPlacement::PlaceAtRoot;
}

// All key arguments (after the first) must be literals for static field access
let keys_literal = args
.iter()
.skip(1)
.all(|a| *a == ExpressionPlacement::Literal);

if keys_literal {
ExpressionPlacement::PlaceAtLeaves
} else {
ExpressionPlacement::PlaceAtRoot
}
}
}

#[cfg(test)]
Expand DownExpand Up@@ -542,4 +573,80 @@ 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::PlaceAtLeaves);

// 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::PlaceAtLeaves);

// get_field(get_field(col, 'a'), 'b') represented as PlaceAtLeaves for base
let args = vec![
ExpressionPlacement::PlaceAtLeaves,
ExpressionPlacement::Literal,
];
assert_eq!(func.placement(&args), ExpressionPlacement::PlaceAtLeaves);
}

#[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::PlaceAtRoot);

// 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::PlaceAtRoot);
}

#[test]
fn test_placement_root() {
let func = GetFieldFunc::new();

// get_field(root_expr, 'literal') -> NOT leaf-pushable
let args = vec![
ExpressionPlacement::PlaceAtRoot,
ExpressionPlacement::Literal,
];
assert_eq!(func.placement(&args), ExpressionPlacement::PlaceAtRoot);

// get_field(col, root_expr) -> NOT leaf-pushable
let args = vec![
ExpressionPlacement::Column,
ExpressionPlacement::PlaceAtRoot,
];
assert_eq!(func.placement(&args), ExpressionPlacement::PlaceAtRoot);
}

#[test]
fn test_placement_edge_cases() {
let func = GetFieldFunc::new();

// Empty args -> NOT leaf-pushable
assert_eq!(func.placement(&[]), ExpressionPlacement::PlaceAtRoot);

// Just base, no key -> PlaceAtLeaves (not a valid call but should handle gracefully)
let args = vec![ExpressionPlacement::Column];
assert_eq!(func.placement(&args), ExpressionPlacement::PlaceAtLeaves);

// 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::PlaceAtRoot);
}
}
17 changes: 7 additions & 10 deletions datafusion/optimizer/src/optimize_projections/mod.rs
View file
Open in desktop
Original file line numberDiff line numberDiff line change
Expand Up@@ -30,8 +30,8 @@ use datafusion_common::{
};
use datafusion_expr::expr::Alias;
use datafusion_expr::{
Aggregate, Distinct, EmptyRelation, Expr, Projection, TableScan, Unnest, Window,
logical_plan::LogicalPlan,
Aggregate, Distinct, EmptyRelation, Expr, ExpressionPlacement, Projection, TableScan,
Unnest, Window, logical_plan::LogicalPlan,
};

use crate::optimize_projections::required_indices::RequiredIndices;
Expand DownExpand Up@@ -530,9 +530,11 @@ fn merge_consecutive_projections(proj: Projection) -> Result<Transformed<Project
// 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()],
&& matches!(
prev_projection.expr
[prev_projection.schema.index_of_column(col).unwrap()]
.placement(),
ExpressionPlacement::PlaceAtRoot
)
}) {
// no change
Expand DownExpand Up@@ -586,11 +588,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(_, _))
}
Comment on lines -589 to -592

@adriangbadriangbDec 31, 2025

Copy link
Copy Markdown
ContributorAuthor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As evidenced by the existing functions for both logical and physical expressions this was already a concept and implementation within the codebase, so all this PR is really doing is allowing arbitrary functions / expressions to declare themselves as trivial.


/// Rewrites a projection expression using the projection before it (i.e. its input)
/// This is a subroutine to the `merge_consecutive_projections` function.
///
Expand Down
Loading