Refactor cast node and resolve column logic

Reuse existing candidate cast node instead of rebuilding
it. Extract shared target-field resolution and same-type
elision logic. Inline the one-off nested-struct validation
wrapper and remove redundant column re-resolution inside
create_cast_expr. Collapse cast and try-cast lowering in
the planner behind a small helper function.

Update the adapter test to ensure that name-based
canonicalization occurs in resolve_physical_column,
rather than again in create_cast_expr.

Author: kosiew

datafusion/physical-expr-adapter/src/schema_rewriter.rs

@@ -498,13 +498,6 @@ impl DefaultPhysicalExprAdapterRewriter {
                         logical_field.data_type()
                     )))?;
 
-        let physical_column_index = self.physical_file_schema.index_of(column.name())?;
-        let column = if column.index() == physical_column_index {
-            column
-        } else {
-            Column::new_with_schema(column.name(), self.physical_file_schema.as_ref())?
-        };
-
         let cast_expr = Arc::new(CastExpr::new_with_target_field(
             Arc::new(column),
             Arc::new(logical_field.clone()),
@@ -1693,7 +1686,7 @@ mod tests {
     }
 
     #[test]
-    fn test_create_cast_expr_uses_name_lookup_not_column_index() {
+    fn test_resolve_physical_column_uses_name_lookup_not_column_index() {
         // Physical schema has column `a` at index 1; index 0 is an incompatible type.
         let physical_schema = Arc::new(Schema::new(vec![
             Field::new("b", DataType::Binary, true),
@@ -1711,11 +1704,19 @@ mod tests {
         };
 
         // Deliberately provide the wrong index for column `a`.
-        // Regression: this must still resolve against physical field `a` by name.
+        // Regression: resolution must still find physical field `a` by name.
+        let (resolved_column, physical_field) = rewriter
+            .resolve_physical_column(&Column::new("a", 0))
+            .unwrap()
+            .expect("Expected resolved column");
+        assert_eq!(resolved_column.name(), "a");
+        assert_eq!(resolved_column.index(), 1);
+        assert_eq!(physical_field.data_type(), &DataType::Int32);
+
         let transformed = rewriter
             .create_cast_expr(
-                Column::new("a", 0),
-                &Arc::new(physical_schema.field_with_name("a").unwrap().clone()),
+                resolved_column,
+                &physical_field,
                 logical_schema.field_with_name("a").unwrap(),
             )
             .unwrap();

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

@@ -45,14 +45,44 @@ const DEFAULT_SAFE_CAST_OPTIONS: CastOptions<'static> = CastOptions {
     format_options: DEFAULT_FORMAT_OPTIONS,
 };
 
-/// Check if name-based struct casting is allowed by validating field compatibility.
-///
-/// This function applies the same validation rules as execution time to ensure
-/// planning-time validation matches runtime validation, enabling fail-fast behavior
-/// instead of deferring errors to execution. Handles structs at any nesting level
-/// (e.g., `List<Struct>`, `Dictionary<_, Struct>`).
-fn can_cast_named_struct_types(source: &DataType, target: &DataType) -> bool {
-    validate_data_type_compatibility("", source, target).is_ok()
+pub(crate) 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 resolve_target_field(
+    expr: &Arc<dyn PhysicalExpr>,
+    target_field: &FieldRef,
+    input_schema: &Schema,
+    force_nullable: bool,
+) -> Result<FieldRef> {
+    if is_default_target_field(target_field) {
+        let field = expr.return_field(input_schema)?;
+        let field = field
+            .as_ref()
+            .clone()
+            .with_data_type(target_field.data_type().clone());
+        let field = if force_nullable {
+            field.with_nullable(true)
+        } else {
+            field
+        };
+        Ok(Arc::new(field))
+    } else {
+        Ok(Arc::clone(target_field))
+    }
+}
+
+pub(crate) fn should_elide_same_type_cast(
+    expr: &Arc<dyn PhysicalExpr>,
+    target_field: &FieldRef,
+    input_schema: &Schema,
+    force_nullable: bool,
+) -> Result<bool> {
+    Ok(is_default_target_field(target_field)
+        || resolve_target_field(expr, target_field, input_schema, force_nullable)?
+            == expr.return_field(input_schema)?)
 }
 
 /// CAST expression casts an expression to a specific data type and returns a runtime error on invalid cast
@@ -151,30 +181,12 @@ 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() {
-            self.expr.return_field(input_schema).map(|field| {
-                Arc::new(
-                    field
-                        .as_ref()
-                        .clone()
-                        .with_data_type(self.cast_type().clone()),
-                )
-            })
-        } else {
-            Ok(Arc::clone(&self.target_field))
-        }
+        resolve_target_field(&self.expr, &self.target_field, input_schema, false)
     }
 
     fn preserves_child_field_semantics(&self, input_schema: &Schema) -> Result<bool> {
-        Ok(self.resolved_target_field(input_schema)?
-            == self.expr.return_field(input_schema)?)
+        should_elide_same_type_cast(&self.expr, &self.target_field, input_schema, false)
     }
 
     /// Check if casting from the specified source type to the target type is a
@@ -338,20 +350,15 @@ pub(crate) fn cast_with_target_field_and_options(
 ) -> Result<Arc<dyn PhysicalExpr>> {
     let expr_type = expr.data_type(input_schema)?;
     let cast_type = target_field.data_type().clone();
-    let candidate = CastExpr::new_with_target_field(
-        Arc::clone(&expr),
-        Arc::clone(&target_field),
-        cast_options.clone(),
-    );
-
-    if expr_type == cast_type
-        && candidate.preserves_child_field_semantics(input_schema)?
-    {
-        return Ok(expr);
-    }
+    let candidate =
+        CastExpr::new_with_target_field(Arc::clone(&expr), target_field, cast_options);
 
     if expr_type == cast_type {
-        return Ok(Arc::new(candidate));
+        return if candidate.preserves_child_field_semantics(input_schema)? {
+            Ok(expr)
+        } else {
+            Ok(Arc::new(candidate))
+        };
     }
 
     let is_valid_cast = if requires_nested_struct_cast(&expr_type, &cast_type) {
@@ -360,19 +367,15 @@ pub(crate) fn cast_with_target_field_and_options(
         // 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)
+        validate_data_type_compatibility("", &expr_type, &cast_type).is_ok()
     } else {
         can_cast_types(&expr_type, &cast_type)
     };
 
     if !is_valid_cast {
         not_impl_err!("Unsupported CAST from {expr_type} to {cast_type}")
     } else {
-        Ok(Arc::new(CastExpr::new_with_target_field(
-            expr,
-            target_field,
-            cast_options,
-        )))
+        Ok(Arc::new(candidate))
     }
 }
 

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

@@ -21,6 +21,7 @@ use std::hash::Hash;
 use std::sync::Arc;
 
 use crate::PhysicalExpr;
+use crate::expressions::cast::{resolve_target_field, should_elide_same_type_cast};
 use arrow::compute;
 use arrow::compute::CastOptions;
 use arrow::datatypes::{DataType, FieldRef, Schema};
@@ -83,31 +84,12 @@ impl TryCastExpr {
         &self.target_field
     }
 
-    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() {
-            self.expr.return_field(input_schema).map(|field| {
-                Arc::new(
-                    field
-                        .as_ref()
-                        .clone()
-                        .with_data_type(self.cast_type().clone())
-                        .with_nullable(true),
-                )
-            })
-        } else {
-            Ok(Arc::clone(&self.target_field))
-        }
+        resolve_target_field(&self.expr, &self.target_field, input_schema, true)
     }
 
     fn preserves_child_field_semantics(&self, input_schema: &Schema) -> Result<bool> {
-        Ok(self.resolved_target_field(input_schema)?
-            == self.expr.return_field(input_schema)?)
+        should_elide_same_type_cast(&self.expr, &self.target_field, input_schema, true)
     }
 }
 
@@ -187,8 +169,7 @@ pub(crate) fn try_cast_with_target_field(
     let candidate = TryCastExpr::new_with_target_field(Arc::clone(&expr), target_field);
 
     if expr_type == cast_type
-        && (candidate.is_default_target_field()
-            || candidate.preserves_child_field_semantics(input_schema)?)
+        && candidate.preserves_child_field_semantics(input_schema)?
     {
         Ok(expr)
     } else if expr_type == cast_type || can_cast_types(&expr_type, &cast_type) {

datafusion/physical-expr/src/planner.rs

@@ -23,7 +23,7 @@ use crate::{
     expressions::{self, Column, Literal, binary, like, similar_to},
 };
 
-use arrow::datatypes::Schema;
+use arrow::datatypes::{FieldRef, Schema};
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::metadata::FieldMetadata;
 use datafusion_common::{
@@ -288,22 +288,29 @@ pub fn create_physical_expr(
                 };
             Ok(expressions::case(expr, when_then_expr, else_expr)?)
         }
-        Expr::Cast(Cast { expr, field }) => {
-            let child = create_physical_expr(expr, input_dfschema, execution_props)?;
-            expressions::cast_with_target_field_and_options(
-                child,
-                input_schema,
-                Arc::clone(field),
-                None,
-            )
-        }
-        Expr::TryCast(TryCast { expr, field }) => {
-            expressions::try_cast_with_target_field(
-                create_physical_expr(expr, input_dfschema, execution_props)?,
-                input_schema,
-                Arc::clone(field),
-            )
-        }
+        Expr::Cast(Cast { expr, field }) => create_physical_cast_like_expr(
+            expr,
+            field,
+            input_dfschema,
+            input_schema,
+            execution_props,
+            |child, input_schema, field| {
+                expressions::cast_with_target_field_and_options(
+                    child,
+                    input_schema,
+                    field,
+                    None,
+                )
+            },
+        ),
+        Expr::TryCast(TryCast { expr, field }) => create_physical_cast_like_expr(
+            expr,
+            field,
+            input_dfschema,
+            input_schema,
+            execution_props,
+            expressions::try_cast_with_target_field,
+        ),
         Expr::Not(expr) => {
             expressions::not(create_physical_expr(expr, input_dfschema, execution_props)?)
         }
@@ -396,6 +403,21 @@ pub fn create_physical_expr(
     }
 }
 
+fn create_physical_cast_like_expr<F>(
+    expr: &Expr,
+    field: &FieldRef,
+    input_dfschema: &DFSchema,
+    input_schema: &Schema,
+    execution_props: &ExecutionProps,
+    lower: F,
+) -> Result<Arc<dyn PhysicalExpr>>
+where
+    F: FnOnce(Arc<dyn PhysicalExpr>, &Schema, FieldRef) -> Result<Arc<dyn PhysicalExpr>>,
+{
+    let child = create_physical_expr(expr, input_dfschema, execution_props)?;
+    lower(child, input_schema, Arc::clone(field))
+}
+
 /// Create vector of Physical Expression from a vector of logical expression
 pub fn create_physical_exprs<'a, I>(
     exprs: I,