This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new a0124a4f0f fix: preserve projection field metadata during physical
planning (#23981)
a0124a4f0f is described below
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]
