Refactor cast expression handling in planner

Expose shared cast_with_target_field helper to validate and
build field-aware CastExprs in one place. Update planner to
directly call this helper, removing the need for temporary
type-only casts. Add regression tests to cover standard casts,
metadata-bearing casts, and same-type semantic-preserving
casts.

Author: kosiew

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

@@ -201,6 +201,12 @@ impl CastExpr {
     }
 }
 
+fn is_default_target_field(target_field: &FieldRef) -> bool {
+    target_field.name().is_empty()
+        && target_field.is_nullable()
+        && target_field.metadata().is_empty()
+}
+
 pub(crate) fn is_order_preserving_cast_family(
     source_type: &DataType,
     target_type: &DataType,
@@ -315,23 +321,57 @@ pub fn cast_with_options(
     input_schema: &Schema,
     cast_type: DataType,
     cast_options: Option<CastOptions<'static>>,
+) -> Result<Arc<dyn PhysicalExpr>> {
+    cast_with_target_field(
+        expr,
+        input_schema,
+        cast_type.into_nullable_field_ref(),
+        cast_options,
+    )
+}
+
+/// Return a PhysicalExpression representing `expr` casted to `target_field`,
+/// preserving any explicit field semantics such as name, nullability, and
+/// metadata.
+pub fn cast_with_target_field(
+    expr: Arc<dyn PhysicalExpr>,
+    input_schema: &Schema,
+    target_field: FieldRef,
+    cast_options: Option<CastOptions<'static>>,
 ) -> Result<Arc<dyn PhysicalExpr>> {
     let expr_type = expr.data_type(input_schema)?;
-    if expr_type == cast_type {
-        Ok(Arc::clone(&expr))
-    } else if requires_nested_struct_cast(&expr_type, &cast_type) {
-        if can_cast_named_struct_types(&expr_type, &cast_type) {
+    let cast_type = target_field.data_type();
+    if expr_type == *cast_type {
+        if is_default_target_field(&target_field) {
+            return Ok(Arc::clone(&expr));
+        }
+
+        Ok(Arc::new(CastExpr::new_with_target_field(
+            expr,
+            target_field,
+            cast_options,
+        )))
+    } else if requires_nested_struct_cast(&expr_type, cast_type) {
+        if can_cast_named_struct_types(&expr_type, cast_type) {
             // Allow casts involving structs (including nested inside Lists, Dictionaries,
             // etc.) that pass name-based compatibility validation. This validation is
             // applied at planning time (now) to fail fast, rather than deferring errors
             // to execution time. The name-based casting logic will be executed at runtime
             // via ColumnarValue::cast_to.
-            Ok(Arc::new(CastExpr::new(expr, cast_type, cast_options)))
+            Ok(Arc::new(CastExpr::new_with_target_field(
+                expr,
+                target_field,
+                cast_options,
+            )))
         } else {
             not_impl_err!("Unsupported CAST from {expr_type} to {cast_type}")
         }
-    } else if can_cast_types(&expr_type, &cast_type) {
-        Ok(Arc::new(CastExpr::new(expr, cast_type, cast_options)))
+    } else if can_cast_types(&expr_type, cast_type) {
+        Ok(Arc::new(CastExpr::new_with_target_field(
+            expr,
+            target_field,
+            cast_options,
+        )))
     } else {
         not_impl_err!("Unsupported CAST from {expr_type} to {cast_type}")
     }

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

@@ -41,7 +41,7 @@ pub use crate::aggregate::stats::StatsType;
 
 pub use binary::{BinaryExpr, binary, similar_to};
 pub use case::{CaseExpr, case};
-pub use cast::{CastExpr, cast};
+pub use cast::{CastExpr, cast, cast_with_target_field};
 pub use cast_column::CastColumnExpr;
 pub use column::{Column, col, with_new_schema};
 pub use datafusion_expr::utils::format_state_name;

datafusion/physical-expr/src/planner.rs

@@ -289,17 +289,12 @@ pub fn create_physical_expr(
             Ok(expressions::case(expr, when_then_expr, else_expr)?)
         }
         Expr::Cast(Cast { expr, field }) => {
-            let expr = create_physical_expr(expr, input_dfschema, execution_props)?;
-
-            // Reuse the standard CAST validation path, but preserve the logical
-            // target field instead of lowering to a type-only physical cast.
-            expressions::cast(Arc::clone(&expr), input_schema, field.data_type().clone())?;
-
-            Ok(Arc::new(expressions::CastExpr::new_with_target_field(
-                expr,
+            expressions::cast_with_target_field(
+                create_physical_expr(expr, input_dfschema, execution_props)?,
+                input_schema,
                 Arc::clone(field),
                 None,
-            )))
+            )
         }
         Expr::TryCast(TryCast { expr, field }) => {
             if !field.metadata().is_empty() {
@@ -497,6 +492,27 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn test_cast_lowering_preserves_standard_cast_semantics() -> Result<()> {
+        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
+        let df_schema = DFSchema::try_from(schema.clone())?;
+        let cast_expr = Expr::Cast(Cast::new(Box::new(col("a")), DataType::Int64));
+
+        let physical =
+            create_physical_expr(&cast_expr, &df_schema, &ExecutionProps::new())?;
+        let cast = physical
+            .as_any()
+            .downcast_ref::<expressions::CastExpr>()
+            .expect("planner should lower ordinary CAST to CastExpr");
+
+        assert_eq!(cast.cast_type(), &DataType::Int64);
+        assert_eq!(physical.return_field(&schema)?.name(), "a");
+        assert_eq!(physical.return_field(&schema)?.data_type(), &DataType::Int64);
+        assert!(!physical.nullable(&schema)?);
+
+        Ok(())
+    }
+
     #[test]
     fn test_cast_lowering_preserves_same_type_field_semantics() -> Result<()> {
         let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);