Improve cast lowering to preserve field metadata

Update cast handling in planner.rs to retain logical target
FieldRef metadata during the cast lowering process. Introduced
a new field-aware helper in cast.rs for this purpose, which
ensures that metadata, name, and nullability intent are kept
intact, even for same-type casts.

Updated planner tests to verify:
- Preservation of metadata and nullability in lowered casts
- CastExpr production for same-type casts with different
  field semantics
- Regression checks for standard non-metadata casts
- Enhanced CAST behavior to retain extension metadata while
  ensuring TRY_CAST still rejects it.

Author: kosiew

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

@@ -319,22 +319,51 @@ pub fn cast_with_options(
     input_schema: &Schema,
     cast_type: DataType,
     cast_options: Option<CastOptions<'static>>,
+) -> Result<Arc<dyn PhysicalExpr>> {
+    cast_with_target_field_and_options(
+        expr,
+        input_schema,
+        cast_type.into_nullable_field_ref(),
+        cast_options,
+    )
+}
+
+/// Return a [`PhysicalExpr`] representing `expr` casted to `target_field`,
+/// preserving the field metadata in the resulting expression.
+pub fn cast_with_target_field_and_options(
+    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)?;
+    let cast_type = target_field.data_type().clone();
     if expr_type == cast_type {
-        Ok(Arc::clone(&expr))
+        Ok(Arc::new(CastExpr::new_with_target_field(
+            expr,
+            target_field,
+            cast_options,
+        )))
     } else if matches!((&expr_type, &cast_type), (Struct(_), Struct(_))) {
         if can_cast_struct_types(&expr_type, &cast_type) {
             // Allow struct-to-struct casts 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)))
+        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_and_options};
 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,22 +289,11 @@ pub fn create_physical_expr(
             Ok(expressions::case(expr, when_then_expr, else_expr)?)
         }
         Expr::Cast(Cast { expr, field }) => {
-            if !field.metadata().is_empty() {
-                let (_, src_field) = expr.to_field(input_dfschema)?;
-                return plan_err!(
-                    "Cast from {} to {} is not supported",
-                    format_type_and_metadata(
-                        src_field.data_type(),
-                        Some(src_field.metadata()),
-                    ),
-                    format_type_and_metadata(field.data_type(), Some(field.metadata()))
-                );
-            }
-
-            expressions::cast(
+            expressions::cast_with_target_field_and_options(
                 create_physical_expr(expr, input_dfschema, execution_props)?,
                 input_schema,
-                field.data_type().clone(),
+                Arc::clone(field),
+                None,
             )
         }
         Expr::TryCast(TryCast { expr, field }) => {
@@ -443,10 +432,12 @@ pub fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
 
 #[cfg(test)]
 mod tests {
+    use crate::expressions::{CastExpr, Column};
     use arrow::array::{ArrayRef, BooleanArray, RecordBatch, StringArray};
     use arrow::datatypes::{DataType, Field};
     use datafusion_common::datatype::DataTypeExt;
     use datafusion_expr::{Operator, col, lit};
+    use std::collections::HashMap;
 
     use super::*;
 
@@ -477,39 +468,123 @@ mod tests {
 
     #[test]
     fn test_cast_to_extension_type() -> Result<()> {
-        let extension_field_type = Arc::new(
-            DataType::FixedSizeBinary(16)
-                .into_nullable_field()
-                .with_metadata(
-                    [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())]
-                        .into(),
-                ),
-        );
-        let expr = lit("3230e5d4-888e-408b-b09b-831f44aa0c58");
+        let extension_field_type =
+            Arc::new(DataType::Int64.into_nullable_field().with_metadata(
+                [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())].into(),
+            ));
+        let expr = col("value");
+        let schema = Schema::new(vec![Field::new("value", DataType::Int32, false)]);
+        let df_schema = DFSchema::try_from(schema.clone())?;
         let cast_expr = Expr::Cast(Cast::new_from_field(
             Box::new(expr.clone()),
             Arc::clone(&extension_field_type),
         ));
-        let err =
-            create_physical_expr(&cast_expr, &DFSchema::empty(), &ExecutionProps::new())
-                .unwrap_err();
-        assert!(err.message().contains("arrow.uuid"));
+        let physical_expr =
+            create_physical_expr(&cast_expr, &df_schema, &ExecutionProps::new())?;
+
+        let field = physical_expr.return_field(&schema)?;
+        assert_eq!(field.data_type(), extension_field_type.data_type());
+        assert_eq!(field.metadata(), extension_field_type.metadata());
+        assert!(physical_expr.nullable(&schema)?);
 
         let try_cast_expr = Expr::TryCast(TryCast::new_from_field(
             Box::new(expr.clone()),
             Arc::clone(&extension_field_type),
         ));
-        let err = create_physical_expr(
-            &try_cast_expr,
-            &DFSchema::empty(),
-            &ExecutionProps::new(),
-        )
-        .unwrap_err();
+        let err =
+            create_physical_expr(&try_cast_expr, &df_schema, &ExecutionProps::new())
+                .unwrap_err();
         assert!(err.message().contains("arrow.uuid"));
 
         Ok(())
     }
 
+    #[test]
+    fn test_create_physical_expr_cast_preserves_target_field_semantics() -> Result<()> {
+        let schema = Schema::new(vec![Field::new("value", DataType::Int32, false)]);
+        let df_schema = DFSchema::try_from(schema.clone())?;
+        let target_field = Arc::new(
+            Field::new("logical_cast", DataType::Int64, true).with_metadata(
+                HashMap::from([("semantic".to_string(), "target".to_string())]),
+            ),
+        );
+        let expr = Expr::Cast(Cast::new_from_field(
+            Box::new(col("value")),
+            Arc::clone(&target_field),
+        ));
+
+        let physical_expr =
+            create_physical_expr(&expr, &df_schema, &ExecutionProps::new())?;
+        let cast_expr = physical_expr
+            .as_any()
+            .downcast_ref::<CastExpr>()
+            .expect("logical cast should lower to CastExpr");
+
+        assert_eq!(cast_expr.target_field(), &target_field);
+
+        let field = physical_expr.return_field(&schema)?;
+        assert_eq!(field.name(), "logical_cast");
+        assert_eq!(field.data_type(), &DataType::Int64);
+        assert!(field.is_nullable());
+        assert_eq!(
+            field.metadata().get("semantic").map(String::as_str),
+            Some("target")
+        );
+        assert!(physical_expr.nullable(&schema)?);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_create_physical_expr_same_type_cast_keeps_target_field() -> Result<()> {
+        let schema = Schema::new(vec![Field::new("value", DataType::Int32, false)]);
+        let df_schema = DFSchema::try_from(schema.clone())?;
+        let target_field = Arc::new(
+            Field::new("same_type_cast", DataType::Int32, true).with_metadata(
+                HashMap::from([("semantic".to_string(), "same_type".to_string())]),
+            ),
+        );
+        let expr = Expr::Cast(Cast::new_from_field(
+            Box::new(col("value")),
+            Arc::clone(&target_field),
+        ));
+
+        let physical_expr =
+            create_physical_expr(&expr, &df_schema, &ExecutionProps::new())?;
+
+        assert!(physical_expr.as_any().downcast_ref::<Column>().is_none());
+        assert!(physical_expr.as_any().downcast_ref::<CastExpr>().is_some());
+
+        let field = physical_expr.return_field(&schema)?;
+        assert_eq!(field.name(), "same_type_cast");
+        assert_eq!(field.data_type(), &DataType::Int32);
+        assert!(field.is_nullable());
+        assert_eq!(
+            field.metadata().get("semantic").map(String::as_str),
+            Some("same_type")
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_create_physical_expr_standard_cast_still_validates() -> Result<()> {
+        let schema = Schema::new(vec![Field::new("value", DataType::Int32, false)]);
+        let df_schema = DFSchema::try_from(schema.clone())?;
+        let expr = Expr::Cast(Cast::new(Box::new(col("value")), DataType::Int64));
+
+        let physical_expr =
+            create_physical_expr(&expr, &df_schema, &ExecutionProps::new())?;
+
+        assert_eq!(physical_expr.data_type(&schema)?, DataType::Int64);
+        assert_eq!(
+            physical_expr.return_field(&schema)?.data_type(),
+            &DataType::Int64
+        );
+
+        Ok(())
+    }
+
     /// Test that deeply nested expressions do not cause a stack overflow.
     ///
     /// This test only runs when the `recursive_protection` feature is enabled,