Refactor cast logic and simplify planner

Moved shared field-aware cast logic to field_aware_cast.rs,
removing try_cast.rs dependency on cast.rs for generic behavior.
Inlined shared child-lowering step in Cast/TryCast branches
by removing the generic planner helper. Added an invariant
comment in schema_rewriter.rs to clarify that create_cast_expr
expects a column already canonicalized by resolve_physical_column.

Author: kosiew

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

@@ -477,8 +477,10 @@ impl DefaultPhysicalExprAdapterRewriter {
     /// Validates type compatibility and creates a field-aware CastExpr if needed.
     ///
     /// Checks whether the physical field can be cast to the logical field type,
-    /// handling both struct and scalar types. Returns a CastExpr with the
-    /// appropriate logical target field configuration.
+    /// handling both struct and scalar types. The `column` passed here is
+    /// expected to already be canonicalized by [`Self::resolve_physical_column`]
+    /// so this helper can focus only on compatibility validation and preserving
+    /// the logical target field configuration.
     fn create_cast_expr(
         &self,
         column: Column,

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

@@ -20,6 +20,9 @@ use std::fmt;
 use std::hash::Hash;
 use std::sync::Arc;
 
+use crate::expressions::field_aware_cast::{
+    resolve_target_field, should_elide_same_type_cast,
+};
 use crate::physical_expr::PhysicalExpr;
 
 use arrow::compute::{CastOptions, can_cast_types};
@@ -45,46 +48,6 @@ const DEFAULT_SAFE_CAST_OPTIONS: CastOptions<'static> = CastOptions {
     format_options: DEFAULT_FORMAT_OPTIONS,
 };
 
-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
 #[derive(Debug, Clone, Eq)]
 pub struct CastExpr {

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

@@ -0,0 +1,62 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use crate::physical_expr::PhysicalExpr;
+use arrow::datatypes::{FieldRef, Schema};
+use datafusion_common::Result;
+
+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)?)
+}

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

@@ -24,6 +24,7 @@ mod cast;
 mod cast_column;
 mod column;
 mod dynamic_filters;
+mod field_aware_cast;
 mod in_list;
 mod is_not_null;
 mod is_null;

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

@@ -21,7 +21,9 @@ use std::hash::Hash;
 use std::sync::Arc;
 
 use crate::PhysicalExpr;
-use crate::expressions::cast::{resolve_target_field, should_elide_same_type_cast};
+use crate::expressions::field_aware_cast::{
+    resolve_target_field, should_elide_same_type_cast,
+};
 use arrow::compute;
 use arrow::compute::CastOptions;
 use arrow::datatypes::{DataType, FieldRef, Schema};

datafusion/physical-expr/src/planner.rs

@@ -23,7 +23,7 @@ use crate::{
     expressions::{self, Column, Literal, binary, like, similar_to},
 };
 
-use arrow::datatypes::{FieldRef, Schema};
+use arrow::datatypes::Schema;
 use datafusion_common::config::ConfigOptions;
 use datafusion_common::metadata::FieldMetadata;
 use datafusion_common::{
@@ -288,29 +288,23 @@ pub fn create_physical_expr(
                 };
             Ok(expressions::case(expr, when_then_expr, else_expr)?)
         }
-        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::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 }) => {
+            let child = create_physical_expr(expr, input_dfschema, execution_props)?;
+            expressions::try_cast_with_target_field(
+                child,
+                input_schema,
+                Arc::clone(field),
+            )
+        }
         Expr::Not(expr) => {
             expressions::not(create_physical_expr(expr, input_dfschema, execution_props)?)
         }
@@ -403,21 +397,6 @@ 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,