Implement improved handling of same-type casts

Refine cast handling in both cast.rs and schema_rewriter.rs.
Explicit same-type casts now preserve CastExpr semantics, while
default type-only casts are elided. Update planner tests to
properly distinguish between the two and ensure consistent
unified behavior across adapters. Added low-level tests for
preserved and elided same-type cases for better coverage.

Author: kosiew

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

@@ -35,11 +35,10 @@ use datafusion_common::{
 };
 use datafusion_functions::core::getfield::GetFieldFunc;
 use datafusion_physical_expr::PhysicalExprSimplifier;
-use datafusion_physical_expr::expressions::CastColumnExpr;
 use datafusion_physical_expr::projection::{ProjectionExprs, Projector};
 use datafusion_physical_expr::{
     ScalarFunctionExpr,
-    expressions::{self, Column},
+    expressions::{self, CastExpr, Column},
 };
 use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
 use itertools::Itertools;
@@ -440,7 +439,7 @@ impl DefaultPhysicalExprAdapterRewriter {
         // TODO: add optimization to move the cast from the column to literal expressions in the case of `col = 123`
         // since that's much cheaper to evalaute.
         // See https://github.com/apache/datafusion/issues/15780#issuecomment-2824716928
-        self.create_cast_column_expr(resolved_column, physical_field, logical_field)
+        self.create_cast_expr(resolved_column, physical_field, logical_field)
     }
 
     /// Resolves a logical column to the corresponding physical column and field.
@@ -476,12 +475,12 @@ impl DefaultPhysicalExprAdapterRewriter {
         )))
     }
 
-    /// Validates type compatibility and creates a CastColumnExpr if needed.
+    /// 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 CastColumnExpr with the
-    /// appropriate configuration.
-    fn create_cast_column_expr(
+    /// handling both struct and scalar types. Returns a CastExpr with the
+    /// appropriate logical target field configuration.
+    fn create_cast_expr(
         &self,
         column: Column,
         physical_field: FieldRef,
@@ -513,9 +512,15 @@ impl DefaultPhysicalExprAdapterRewriter {
             }
         }
 
-        let cast_expr = Arc::new(CastColumnExpr::new(
+        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),
-            physical_field,
             Arc::new(logical_field.clone()),
             None,
         ));
@@ -669,7 +674,7 @@ mod tests {
     use arrow::datatypes::{DataType, Field, Fields, Schema, SchemaRef};
     use datafusion_common::{Result, ScalarValue, assert_contains, record_batch};
     use datafusion_expr::Operator;
-    use datafusion_physical_expr::expressions::{Column, Literal, col, lit};
+    use datafusion_physical_expr::expressions::{CastExpr, Column, Literal, col, lit};
     use datafusion_physical_expr_common::physical_expr::PhysicalExpr;
     use itertools::Itertools;
     use std::sync::Arc;
@@ -702,7 +707,7 @@ mod tests {
         let result = adapter.rewrite(column_expr).unwrap();
 
         // Should be wrapped in a cast expression
-        assert!(result.as_any().downcast_ref::<CastColumnExpr>().is_some());
+        assert!(result.as_any().downcast_ref::<CastExpr>().is_some());
     }
 
     #[test]
@@ -723,8 +728,8 @@ mod tests {
         let result = adapter.rewrite(Arc::new(Column::new("a", 0)))?;
         let cast = result
             .as_any()
-            .downcast_ref::<CastColumnExpr>()
-            .expect("Expected CastColumnExpr");
+            .downcast_ref::<CastExpr>()
+            .expect("Expected CastExpr");
 
         assert_eq!(cast.target_field().data_type(), &DataType::Int64);
         assert!(!cast.target_field().is_nullable());
@@ -736,8 +741,10 @@ mod tests {
             Some("1")
         );
 
-        // Ensure the expression reports the logical nullability regardless of input schema
-        assert!(!result.nullable(physical_schema.as_ref())?);
+        // Runtime nullability follows the child expression, but the logical
+        // target field nullability is still preserved via return_field().
+        assert!(result.nullable(physical_schema.as_ref())?);
+        assert!(!result.return_field(physical_schema.as_ref())?.is_nullable());
 
         Ok(())
     }
@@ -772,9 +779,8 @@ mod tests {
         println!("Rewritten expression: {result}");
 
         let expected = expressions::BinaryExpr::new(
-            Arc::new(CastColumnExpr::new(
+            Arc::new(CastExpr::new_with_target_field(
                 Arc::new(Column::new("a", 0)),
-                Arc::new(Field::new("a", DataType::Int32, false)),
                 Arc::new(Field::new("a", DataType::Int64, false)),
                 None,
             )),
@@ -860,17 +866,6 @@ mod tests {
 
         let result = adapter.rewrite(column_expr).unwrap();
 
-        let physical_struct_fields: Fields = vec![
-            Field::new("id", DataType::Int32, false),
-            Field::new("name", DataType::Utf8, true),
-        ]
-        .into();
-        let physical_field = Arc::new(Field::new(
-            "data",
-            DataType::Struct(physical_struct_fields),
-            false,
-        ));
-
         let logical_struct_fields: Fields = vec![
             Field::new("id", DataType::Int64, false),
             Field::new("name", DataType::Utf8View, true),
@@ -882,9 +877,8 @@ mod tests {
             false,
         ));
 
-        let expected = Arc::new(CastColumnExpr::new(
+        let expected = Arc::new(CastExpr::new_with_target_field(
             Arc::new(Column::new("data", 0)),
-            physical_field,
             logical_field,
             None,
         )) as Arc<dyn PhysicalExpr>;
@@ -1558,11 +1552,11 @@ mod tests {
 
         let result = adapter.rewrite(column_expr).unwrap();
 
-        // Should be a CastColumnExpr
+        // Should be a CastExpr
         let cast_expr = result
             .as_any()
-            .downcast_ref::<CastColumnExpr>()
-            .expect("Expected CastColumnExpr");
+            .downcast_ref::<CastExpr>()
+            .expect("Expected CastExpr");
 
         // Verify the inner column points to the correct physical index (1)
         let inner_col = cast_expr
@@ -1581,7 +1575,7 @@ mod tests {
     }
 
     #[test]
-    fn test_create_cast_column_expr_uses_name_lookup_not_column_index() {
+    fn test_create_cast_expr_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),
@@ -1601,7 +1595,7 @@ mod tests {
         // Deliberately provide the wrong index for column `a`.
         // Regression: this must still resolve against physical field `a` by name.
         let transformed = rewriter
-            .create_cast_column_expr(
+            .create_cast_expr(
                 Column::new("a", 0),
                 Arc::new(physical_schema.field_with_name("a").unwrap().clone()),
                 logical_schema.field_with_name("a").unwrap(),
@@ -1611,11 +1605,16 @@ mod tests {
         let cast_expr = transformed
             .data
             .as_any()
-            .downcast_ref::<CastColumnExpr>()
-            .expect("Expected CastColumnExpr");
+            .downcast_ref::<CastExpr>()
+            .expect("Expected CastExpr");
 
-        assert_eq!(cast_expr.input_field().name(), "a");
-        assert_eq!(cast_expr.input_field().data_type(), &DataType::Int32);
+        let inner_col = cast_expr
+            .expr()
+            .as_any()
+            .downcast_ref::<Column>()
+            .expect("Expected inner Column");
+        assert_eq!(inner_col.name(), "a");
+        assert_eq!(inner_col.index(), 1);
         assert_eq!(cast_expr.target_field().data_type(), &DataType::Int64);
     }
 }

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

@@ -176,6 +176,10 @@ impl CastExpr {
         }
     }
 
+    fn preserves_child_field_semantics(&self, input_schema: &Schema) -> Result<bool> {
+        Ok(self.resolved_target_field(input_schema)? == self.expr.return_field(input_schema)?)
+    }
+
     /// Check if casting from the specified source type to the target type is a
     /// widening cast (e.g. from `Int8` to `Int16`).
     pub fn check_bigger_cast(cast_type: &DataType, src: &DataType) -> bool {
@@ -337,7 +341,13 @@ 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();
-    if expr_type == *cast_type {
+    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(Arc::clone(&expr));
     }
 
@@ -353,11 +363,7 @@ pub(crate) fn cast_with_target_field_and_options(
     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))
     }
 }
 
@@ -377,7 +383,7 @@ pub fn cast(
 mod tests {
     use super::*;
 
-    use crate::expressions::column::col;
+    use crate::expressions::{Column, column::col};
 
     use arrow::{
         array::{
@@ -921,6 +927,51 @@ mod tests {
         Ok(())
     }
 
+    #[test]
+    fn field_aware_same_type_cast_preserves_explicit_target_field() -> Result<()> {
+        let schema = Schema::new(vec![Field::new("a", Int32, false)]);
+        let expr = cast_with_target_field_and_options(
+            col("a", &schema)?,
+            &schema,
+            Arc::new(Field::new("logical_a", Int32, true).with_metadata(HashMap::from([(
+                "target_meta".to_string(),
+                "1".to_string(),
+            )]))),
+            None,
+        )?;
+
+        let cast_expr = expr
+            .as_any()
+            .downcast_ref::<CastExpr>()
+            .expect("explicit same-type target should preserve CastExpr");
+        let field = cast_expr.return_field(&schema)?;
+
+        assert_eq!(field.name(), "logical_a");
+        assert!(field.is_nullable());
+        assert_eq!(
+            field.metadata().get("target_meta").map(String::as_str),
+            Some("1")
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn default_same_type_cast_is_elided() -> Result<()> {
+        let schema = Schema::new(vec![Field::new("a", Int32, false)]);
+        let expr = cast_with_target_field_and_options(
+            col("a", &schema)?,
+            &schema,
+            Int32.into_nullable_field_ref(),
+            None,
+        )?;
+
+        assert!(expr.as_any().downcast_ref::<Column>().is_some());
+        assert!(expr.as_any().downcast_ref::<CastExpr>().is_none());
+
+        Ok(())
+    }
+
     #[test]
     fn field_aware_cast_nullable_prefers_child_nullability() -> Result<()> {
         // When the child expression is nullable the cast must be treated as

datafusion/physical-expr/src/planner.rs

@@ -542,7 +542,7 @@ mod tests {
     }
 
     #[test]
-    fn test_create_physical_expr_same_type_cast_is_elided() -> Result<()> {
+    fn test_create_physical_expr_same_type_cast_preserves_explicit_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(
@@ -555,6 +555,25 @@ mod tests {
             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("same-type cast with explicit field should preserve CastExpr");
+        assert_eq!(cast_expr.target_field(), &target_field);
+        assert_eq!(physical_expr.return_field(&schema)?.name(), "same_type_cast");
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_create_physical_expr_default_same_type_cast_is_elided() -> 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::Int32));
+
         let physical_expr =
             create_physical_expr(&expr, &df_schema, &ExecutionProps::new())?;