Refactor cast expression handling

Consolidate default-target-field predicate and success
construction path in cast_with_target_field to reduce
duplicate code in cast.rs. Simplify tests in planner.rs
by implementing shared setup helpers and caching
return_field() results for standard casts.

Author: kosiew

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

@@ -151,14 +151,8 @@ impl CastExpr {
         &self.cast_options
     }
 
-    fn is_default_target_field(&self) -> bool {
-        self.target_field.name().is_empty()
-            && self.target_field.is_nullable()
-            && self.target_field.metadata().is_empty()
-    }
-
     fn resolved_target_field(&self, input_schema: &Schema) -> Result<FieldRef> {
-        if self.is_default_target_field() {
+        if is_default_target_field(&self.target_field) {
             self.expr.return_field(input_schema).map(|field| {
                 Arc::new(
                     field
@@ -345,36 +339,28 @@ pub fn cast_with_target_field(
         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_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_with_target_field(
-            expr,
-            target_field,
-            cast_options,
-        )))
+    let can_build_cast = if requires_nested_struct_cast(&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.
+        can_cast_named_struct_types(&expr_type, cast_type)
     } else {
-        not_impl_err!("Unsupported CAST from {expr_type} to {cast_type}")
+        can_cast_types(&expr_type, cast_type)
+    };
+
+    if !can_build_cast {
+        return not_impl_err!("Unsupported CAST from {expr_type} to {cast_type}");
     }
+
+    Ok(Arc::new(CastExpr::new_with_target_field(
+        expr,
+        target_field,
+        cast_options,
+    )))
 }
 
 /// Return a PhysicalExpression representing `expr` casted to

datafusion/physical-expr/src/planner.rs

@@ -439,6 +439,15 @@ mod tests {
 
     use super::*;
 
+    fn test_cast_schema() -> Schema {
+        Schema::new(vec![Field::new("a", DataType::Int32, false)])
+    }
+
+    fn lower_cast_expr(expr: &Expr, schema: &Schema) -> Result<Arc<dyn PhysicalExpr>> {
+        let df_schema = DFSchema::try_from(schema.clone())?;
+        create_physical_expr(expr, &df_schema, &ExecutionProps::new())
+    }
+
     #[test]
     fn test_create_physical_expr_scalar_input_output() -> Result<()> {
         let expr = col("letter").eq(lit("A"));
@@ -466,8 +475,7 @@ mod tests {
 
     #[test]
     fn test_cast_lowering_preserves_target_field_metadata() -> Result<()> {
-        let schema = Schema::new(vec![Field::new("a", DataType::Int32, false)]);
-        let df_schema = DFSchema::try_from(schema.clone())?;
+        let schema = test_cast_schema();
         let target_field = Arc::new(
             Field::new("cast_target", DataType::Int64, true).with_metadata(
                 [("target_meta".to_string(), "1".to_string())].into(),
@@ -478,8 +486,7 @@ mod tests {
             Arc::clone(&target_field),
         ));
 
-        let physical =
-            create_physical_expr(&cast_expr, &df_schema, &ExecutionProps::new())?;
+        let physical = lower_cast_expr(&cast_expr, &schema)?;
         let cast = physical
             .as_any()
             .downcast_ref::<expressions::CastExpr>()
@@ -494,29 +501,27 @@ mod tests {
 
     #[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 schema = test_cast_schema();
         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 physical = lower_cast_expr(&cast_expr, &schema)?;
         let cast = physical
             .as_any()
             .downcast_ref::<expressions::CastExpr>()
             .expect("planner should lower ordinary CAST to CastExpr");
+        let returned_field = physical.return_field(&schema)?;
 
         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_eq!(returned_field.name(), "a");
+        assert_eq!(returned_field.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)]);
-        let df_schema = DFSchema::try_from(schema.clone())?;
+        let schema = test_cast_schema();
         let target_field = Arc::new(
             Field::new("same_type_cast", DataType::Int32, true).with_metadata(
                 [("target_meta".to_string(), "same-type".to_string())].into(),
@@ -527,8 +532,7 @@ mod tests {
             Arc::clone(&target_field),
         ));
 
-        let physical =
-            create_physical_expr(&cast_expr, &df_schema, &ExecutionProps::new())?;
+        let physical = lower_cast_expr(&cast_expr, &schema)?;
         let cast = physical
             .as_any()
             .downcast_ref::<expressions::CastExpr>()