This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-23981-e7304eec73607be8d82846fb279d64c301371238 in repository https://gitbox.apache.org/repos/asf/datafusion.git
commit a0124a4f0f7d9614d13c8d1753df5263a2c42b51 Author: subotac <[email protected]> AuthorDate: Wed Aug 5 10:01:33 2026 +0300 fix: preserve projection field metadata during physical planning (#23981) ## Which issue does this PR close? - Closes #23529. ## Rationale for this change Logical projection aliases can attach field metadata, but physical projection planning derived its output schema only from the unwrapped physical expressions. This caused schema validation to reject aggregates above such projections. ## What changes are included in this PR? - Apply logical projection schema and field metadata when constructing `ProjectionExec`. - Preserve the existing physical derivation of field names, types, and nullability. - Add metadata-aware constructors across the physical-expression and physical-plan crate boundary. - Cover projections with alias metadata followed by aggregation. ## Are these changes tested? Yes. A focused physical planner regression test reproduces the original schema mismatch and passes with the fix. The relevant core and physical-plan unit suites, targeted all-targets/all-features clippy, formatting, and diff checks also pass. ## Are there any user-facing changes? Projection field metadata is now preserved in physical plans. Two additive constructor methods support passing projection schema metadata across crate boundaries; no existing API is changed. --- datafusion/core/src/physical_planner.rs | 55 +++++++++++++++++++++-- datafusion/physical-expr/src/projection.rs | 50 +++++++++++++++++++++ datafusion/physical-plan/src/projection.rs | 70 +++++++++++++++++++++++++++++- 3 files changed, 171 insertions(+), 4 deletions(-) diff --git a/datafusion/core/src/physical_planner.rs b/datafusion/core/src/physical_planner.rs index 46e2957af0..da8e0f2f57 100644 --- a/datafusion/core/src/physical_planner.rs +++ b/datafusion/core/src/physical_planner.rs @@ -1106,6 +1106,7 @@ impl DefaultPhysicalPlanner { children.one()?, input, expr, + node.schema(), )?, LogicalPlan::Filter(Filter { predicate, input, .. @@ -1329,6 +1330,7 @@ impl DefaultPhysicalPlanner { physical_left, input, expr, + left.schema(), )?, _ => physical_left, }; @@ -1343,6 +1345,7 @@ impl DefaultPhysicalPlanner { physical_right, input, expr, + right.schema(), )?, _ => physical_right, }; @@ -1727,6 +1730,7 @@ impl DefaultPhysicalPlanner { join, input, expr, + new_logical.schema(), )? } else { join @@ -2958,6 +2962,7 @@ impl DefaultPhysicalPlanner { input_exec: Arc<dyn ExecutionPlan>, input: &Arc<LogicalPlan>, expr: &[Expr], + output_schema: &DFSchema, ) -> Result<Arc<dyn ExecutionPlan>> { let input_logical_schema = input.as_ref().schema(); let input_physical_schema = input_exec.schema(); @@ -3015,7 +3020,11 @@ impl DefaultPhysicalPlanner { .into_iter() .map(|(expr, alias)| ProjectionExpr { expr, alias }) .collect(); - Ok(Arc::new(ProjectionExec::try_new(proj_exprs, input_exec)?)) + Ok(Arc::new(ProjectionExec::try_new_with_schema_metadata( + proj_exprs, + input_exec, + output_schema.as_arrow(), + )?)) } PlanAsyncExpr::Async( async_map, @@ -3027,8 +3036,11 @@ impl DefaultPhysicalPlanner { .into_iter() .map(|(expr, alias)| ProjectionExpr { expr, alias }) .collect(); - let new_proj_exec = - ProjectionExec::try_new(proj_exprs, Arc::new(async_exec))?; + let new_proj_exec = ProjectionExec::try_new_with_schema_metadata( + proj_exprs, + Arc::new(async_exec), + output_schema.as_arrow(), + )?; Ok(Arc::new(new_proj_exec)) } _ => internal_err!("Unexpected PlanAsyncExpressions variant"), @@ -3579,6 +3591,43 @@ mod tests { Ok(()) } + #[tokio::test] + async fn test_projection_preserves_field_metadata_for_aggregate() -> Result<()> { + use datafusion_common::metadata::FieldMetadata; + use datafusion_expr::expr::AggregateFunction; + use datafusion_functions_aggregate::min_max::max_udaf; + + let schema = Schema::new(vec![Field::new("value", DataType::Utf8, false)]); + let input = LogicalPlan::EmptyRelation(EmptyRelation { + produce_one_row: false, + schema: Arc::new(schema.to_dfschema()?), + }); + let metadata = + FieldMetadata::from(HashMap::from([("foo".to_string(), "bar".to_string())])); + let projection = LogicalPlan::Projection(Projection::try_new( + vec![col("value").alias_with_metadata("value", Some(metadata))], + Arc::new(input), + )?); + let aggregate = LogicalPlan::Aggregate(Aggregate::try_new( + Arc::new(projection), + vec![], + vec![Expr::AggregateFunction(AggregateFunction::new_udf( + max_udaf(), + vec![col("value")], + false, + None, + vec![], + None, + ))], + )?); + + DefaultPhysicalPlanner::default() + .create_physical_plan(&aggregate, &SessionContext::new().state()) + .await?; + + Ok(()) + } + #[derive(Debug, Default)] struct NullAccumulator; diff --git a/datafusion/physical-expr/src/projection.rs b/datafusion/physical-expr/src/projection.rs index bfa40c7838..e3fd6ddf74 100644 --- a/datafusion/physical-expr/src/projection.rs +++ b/datafusion/physical-expr/src/projection.rs @@ -539,6 +539,56 @@ impl ProjectionExprs { }) } + /// Create a new [`Projector`] using field and schema metadata from + /// `projected_schema`. + /// + /// Field names, data types, and nullability are still derived from the physical + /// projection expressions and `input_schema`; only field and schema metadata are + /// taken from `projected_schema`. + /// + /// # Errors + /// + /// Returns an error if the projection cannot be applied to `input_schema`, or if + /// `projected_schema` has a different number of fields than the projection. + pub fn make_projector_with_schema_metadata( + &self, + input_schema: &Schema, + projected_schema: &Schema, + ) -> Result<Projector> { + let output_schema = self.project_schema(input_schema)?; + if output_schema.fields().len() != projected_schema.fields().len() { + return Err(internal_datafusion_err!( + "Projection has {} output fields but metadata schema has {} fields", + output_schema.fields().len(), + projected_schema.fields().len() + )); + } + + let fields = output_schema + .fields() + .iter() + .zip(projected_schema.fields()) + .map(|(field, projected_field)| { + Arc::new( + field + .as_ref() + .clone() + .with_metadata(projected_field.metadata().clone()), + ) + }) + .collect::<Vec<_>>(); + let output_schema = Arc::new(Schema::new_with_metadata( + fields, + projected_schema.metadata().clone(), + )); + + Ok(Projector { + projection: self.clone(), + output_schema, + expression_metrics: None, + }) + } + pub fn create_expression_metrics( &self, metrics: &ExecutionPlanMetricsSet, diff --git a/datafusion/physical-plan/src/projection.rs b/datafusion/physical-plan/src/projection.rs index 42501f2239..fac837b09f 100644 --- a/datafusion/physical-plan/src/projection.rs +++ b/datafusion/physical-plan/src/projection.rs @@ -40,7 +40,7 @@ use std::pin::Pin; use std::sync::Arc; use std::task::{Context, Poll}; -use arrow::datatypes::SchemaRef; +use arrow::datatypes::{Schema, SchemaRef}; use arrow::record_batch::RecordBatch; use datafusion_common::config::ConfigOptions; use datafusion_common::tree_node::{ @@ -143,6 +143,34 @@ impl ProjectionExec { Self::try_from_projector(projector, input) } + /// Create a projection using field and schema metadata from + /// `projected_schema`. + /// + /// Field names, data types, and nullability are still derived from the physical + /// projection expressions and the input plan; only field and schema metadata are + /// taken from `projected_schema`. + /// + /// # Errors + /// + /// Returns an error if the projection cannot be applied to the input plan, or if + /// `projected_schema` has a different number of fields than the projection. + pub fn try_new_with_schema_metadata<I, E>( + expr: I, + input: Arc<dyn ExecutionPlan>, + projected_schema: &Schema, + ) -> Result<Self> + where + I: IntoIterator<Item = E>, + E: Into<ProjectionExpr>, + { + let input_schema = input.schema(); + let expr_arc = expr.into_iter().map(Into::into).collect::<Arc<_>>(); + let projection = ProjectionExprs::from_expressions(expr_arc); + let projector = projection + .make_projector_with_schema_metadata(&input_schema, projected_schema)?; + Self::try_from_projector(projector, input) + } + fn try_from_projector( projector: Projector, input: Arc<dyn ExecutionPlan>, @@ -1395,6 +1423,46 @@ mod tests { BinaryExpr, Column, DynamicFilterPhysicalExpr, Literal, binary, col, lit, }; + #[test] + fn test_try_new_with_schema_metadata_only_replaces_metadata() -> Result<()> { + let input_schema = Arc::new(Schema::new(vec![Field::new( + "input", + DataType::Int32, + false, + )])); + let input: Arc<dyn ExecutionPlan> = Arc::new(EmptyExec::new(input_schema)); + let field_metadata = + HashMap::from([("field-key".to_string(), "field-value".to_string())]); + let schema_metadata = + HashMap::from([("schema-key".to_string(), "schema-value".to_string())]); + let metadata_schema = Schema::new_with_metadata( + vec![ + Field::new("ignored", DataType::Utf8, true) + .with_metadata(field_metadata.clone()), + ], + schema_metadata.clone(), + ); + + let projection = ProjectionExec::try_new_with_schema_metadata( + [ProjectionExpr { + expr: Arc::new(Column::new("input", 0)), + alias: "output".to_string(), + }], + input, + &metadata_schema, + )?; + + let expected_schema = Arc::new(Schema::new_with_metadata( + vec![ + Field::new("output", DataType::Int32, false) + .with_metadata(field_metadata), + ], + schema_metadata, + )); + assert_eq!(projection.schema(), expected_schema); + Ok(()) + } + #[test] fn test_collect_column_indices() -> Result<()> { let expr = Arc::new(BinaryExpr::new( --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
