Make Physical CastExpr Field-aware and unify cast semantics across physical expressions (#20814)

## Which issue does this PR close?

* Part of #20164

## Rationale for this change

Physical `CastExpr` previously stored only a target `DataType`. This
caused field-level semantics (name, nullability, and metadata) to be
lost when casts were represented in the physical layer. In contrast,
logical expressions already carry this information through `FieldRef`.

This mismatch created several issues:

* Physical and logical cast representations diverged in how they
preserve schema semantics.
* Struct casting logic behaved differently depending on whether the cast
was represented as `CastExpr` or `CastColumnExpr`.
* Downstream components (such as schema rewriting and ordering
equivalence analysis) required additional branching and duplicated
logic.

Making `CastExpr` field-aware aligns the physical representation with
logical semantics and enables consistent schema propagation across
execution planning and expression evaluation.

## What changes are included in this PR?

This PR introduces field-aware semantics to `CastExpr` and simplifies
several areas that previously relied on type-only casting.

Key changes include:

1. **Field-aware CastExpr**

* Replace the `cast_type: DataType` field with `target_field: FieldRef`.
* Add `new_with_target_field` constructor to explicitly construct
field-aware casts.
* Keep the existing `new(expr, DataType)` constructor as a compatibility
shim that creates a canonical field.

2. **Return-field and nullability behavior**

* `return_field` now returns the full `target_field`, preserving name,
nullability, and metadata.
* `nullable()` now derives its result from the resolved target field
rather than the input expression.
* Add compatibility logic for legacy type-only casts to preserve
previous behavior.

3. **Struct cast validation improvements**

* Struct-to-struct casting now validates compatibility using field
information before execution.
* Planning-time validation prevents unsupported casts from reaching
execution.

4. **Shared cast property logic**

* Introduce shared helper functions (`cast_expr_properties`,
`is_order_preserving_cast_family`) for determining ordering
preservation.
* Reuse this logic in both `CastExpr` and `CastColumnExpr` to avoid
duplicated implementations.

5. **Schema rewriter improvements**

   * Refactor physical column resolution into `resolve_physical_column`.
* Simplify cast insertion logic when logical and physical fields differ.
* Pass explicit physical and logical fields to cast creation for
improved correctness.

6. **Ordering equivalence simplification**

* Introduce `substitute_cast_like_ordering` helper to unify handling of
`CastExpr` and `CastColumnExpr` in ordering equivalence analysis.

7. **Additional unit tests**

   * Validate metadata propagation through `return_field`.
   * Verify nullability behavior for field-aware casts.
   * Ensure legacy type-only casts preserve existing semantics.
   * Test struct-cast validation with nested field semantics.

## Are these changes tested?

Yes.

New unit tests were added in `physical-expr/src/expressions/cast.rs` to
verify:

* Metadata propagation through field-aware casts
* Correct nullability behavior derived from the target field
* Backward compatibility with legacy type-only constructors
* Struct cast compatibility validation using nested fields

Existing tests continue to pass and validate compatibility with the
previous API behavior.

## Are there any user-facing changes?

There are no direct user-facing behavior changes.

This change primarily improves internal schema semantics and consistency
in the physical expression layer. Existing APIs remain compatible
through the legacy constructor that accepts only a `DataType`.

## LLM-generated code disclosure

This PR includes LLM-generated code and comments. All LLM-generated
content has been manually reviewed and tested.

Author: kosiew

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

@@ -26,7 +26,7 @@ use std::sync::Arc;
 
 use arrow::array::RecordBatch;
 use arrow::compute::can_cast_types;
-use arrow::datatypes::{DataType, Field, SchemaRef};
+use arrow::datatypes::{DataType, Field, FieldRef, SchemaRef};
 use datafusion_common::{
     Result, ScalarValue, exec_err,
     metadata::FieldMetadata,
@@ -404,71 +404,76 @@ impl DefaultPhysicalExprAdapterRewriter {
             }
         };
 
-        // Check if the column exists in the physical schema
-        let physical_column_index = match self
-            .physical_file_schema
-            .index_of(column.name())
-        {
-            Ok(index) => index,
-            Err(_) => {
-                if !logical_field.is_nullable() {
-                    return exec_err!(
-                        "Non-nullable column '{}' is missing from the physical schema",
-                        column.name()
-                    );
-                }
-                // If the column is missing from the physical schema fill it in with nulls.
-                // For a different behavior, provide a custom `PhysicalExprAdapter` implementation.
-                let null_value = ScalarValue::Null.cast_to(logical_field.data_type())?;
-                return Ok(Transformed::yes(Arc::new(
-                    expressions::Literal::new_with_metadata(
-                        null_value,
-                        Some(FieldMetadata::from(logical_field)),
-                    ),
-                )));
+        let Some((resolved_column, physical_field)) =
+            self.resolve_physical_column(column)?
+        else {
+            if !logical_field.is_nullable() {
+                return exec_err!(
+                    "Non-nullable column '{}' is missing from the physical schema",
+                    column.name()
+                );
             }
+            // If the column is missing from the physical schema fill it in with nulls.
+            // For a different behavior, provide a custom `PhysicalExprAdapter` implementation.
+            let null_value = ScalarValue::Null.cast_to(logical_field.data_type())?;
+            return Ok(Transformed::yes(Arc::new(
+                expressions::Literal::new_with_metadata(
+                    null_value,
+                    Some(FieldMetadata::from(logical_field)),
+                ),
+            )));
         };
-        let physical_field = self.physical_file_schema.field(physical_column_index);
 
-        if column.index() == physical_column_index && logical_field == physical_field {
+        if resolved_column.index() == column.index()
+            && logical_field == physical_field.as_ref()
+        {
             return Ok(Transformed::no(expr));
         }
 
-        let column = self.resolve_column(column, physical_column_index)?;
-
-        if logical_field == physical_field {
+        if logical_field == physical_field.as_ref() {
             // If the fields match (including metadata/nullability), we can use the column as is
-            return Ok(Transformed::yes(Arc::new(column)));
+            return Ok(Transformed::yes(Arc::new(resolved_column)));
         }
 
-        if logical_field.data_type() == physical_field.data_type() {
-            // The data type matches, but the field metadata / nullability differs.
-            // Emit a CastColumnExpr so downstream schema construction uses the logical field.
-            return self.create_cast_column_expr(column, logical_field);
-        }
-
-        // We need to cast the column to the logical data type
+        // We need a cast expression whenever the logical and physical fields differ,
+        // whether that difference is only metadata/nullability or also data type.
         // 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(column, logical_field)
+        self.create_cast_column_expr(resolved_column, physical_field, logical_field)
     }
 
-    /// Resolves a column expression, handling index and type mismatches.
-    ///
-    /// Returns the appropriate Column expression when the column's index or data type
-    /// don't match the physical schema. Assumes that the early-exit case (both index
-    /// and type match) has already been checked by the caller.
-    fn resolve_column(
+    /// Resolves a logical column to the corresponding physical column and field.
+    fn resolve_physical_column(
         &self,
         column: &Column,
-        physical_column_index: usize,
-    ) -> Result<Column> {
-        if column.index() == physical_column_index {
-            Ok(column.clone())
+    ) -> Result<Option<(Column, FieldRef)>> {
+        // The physical schema adaptation step intentionally resolves columns by **name first**
+        // rather than trusting the incoming index. This mirrors what the old refactoring
+        // did before `resolve_physical_column()` was extracted: the planner might hand us a
+        // `Column` whose `index` field is stale (e.g. after projection/rename rewrites), so
+        // resolving by name ensures we match the correct physical slot. Once we know the
+        // proper index we rebuild the `Column` with `new_with_schema` so callers can rely
+        // on `column.index()` later without having to re-query the schema.
+        let Ok(physical_column_index) = self.physical_file_schema.index_of(column.name())
+        else {
+            return Ok(None);
+        };
+
+        let column = if column.index() == physical_column_index {
+            column.clone()
         } else {
-            Column::new_with_schema(column.name(), self.physical_file_schema.as_ref())
-        }
+            Column::new_with_schema(column.name(), self.physical_file_schema.as_ref())?
+        };
+
+        Ok(Some((
+            column,
+            Arc::new(
+                self.physical_file_schema
+                    .field(physical_column_index)
+                    .clone(),
+            ),
+        )))
     }
 
     /// Validates type compatibility and creates a CastColumnExpr if needed.
@@ -479,35 +484,29 @@ impl DefaultPhysicalExprAdapterRewriter {
     fn create_cast_column_expr(
         &self,
         column: Column,
+        physical_field: FieldRef,
         logical_field: &Field,
     ) -> Result<Transformed<Arc<dyn PhysicalExpr>>> {
-        // Look up the column index in the physical schema by name to ensure correctness.
-        let physical_column_index = self.physical_file_schema.index_of(column.name())?;
-        let actual_physical_field =
-            self.physical_file_schema.field(physical_column_index);
-
         // For struct types, use validate_struct_compatibility which handles:
         // - Missing fields in source (filled with nulls)
         // - Extra fields in source (ignored)
         // - Recursive validation of nested structs
         // For non-struct types, use Arrow's can_cast_types
-        match (actual_physical_field.data_type(), logical_field.data_type()) {
+        match (physical_field.data_type(), logical_field.data_type()) {
             (DataType::Struct(physical_fields), DataType::Struct(logical_fields)) => {
                 validate_struct_compatibility(
                     physical_fields.as_ref(),
                     logical_fields.as_ref(),
                 )?;
             }
             _ => {
-                let is_compatible = can_cast_types(
-                    actual_physical_field.data_type(),
-                    logical_field.data_type(),
-                );
+                let is_compatible =
+                    can_cast_types(physical_field.data_type(), logical_field.data_type());
                 if !is_compatible {
                     return exec_err!(
                         "Cannot cast column '{}' from '{}' (physical data type) to '{}' (logical data type)",
                         column.name(),
-                        actual_physical_field.data_type(),
+                        physical_field.data_type(),
                         logical_field.data_type()
                     );
                 }
@@ -516,7 +515,7 @@ impl DefaultPhysicalExprAdapterRewriter {
 
         let cast_expr = Arc::new(CastColumnExpr::new(
             Arc::new(column),
-            Arc::new(actual_physical_field.clone()),
+            physical_field,
             Arc::new(logical_field.clone()),
             None,
         ));
@@ -1604,6 +1603,7 @@ mod tests {
         let transformed = rewriter
             .create_cast_column_expr(
                 Column::new("a", 0),
+                Arc::new(physical_schema.field_with_name("a").unwrap().clone()),
                 logical_schema.field_with_name("a").unwrap(),
             )
             .unwrap();

datafusion/physical-expr/src/equivalence/properties/mod.rs

@@ -39,7 +39,7 @@ use crate::{
     PhysicalSortRequirement,
 };
 
-use arrow::datatypes::SchemaRef;
+use arrow::datatypes::{DataType, SchemaRef};
 use datafusion_common::tree_node::{Transformed, TransformedResult, TreeNode};
 use datafusion_common::{Constraint, Constraints, HashMap, Result, plan_err};
 use datafusion_expr::interval_arithmetic::Interval;
@@ -195,6 +195,39 @@ impl OrderingEquivalenceCache {
 }
 
 impl EquivalenceProperties {
+    /// Helper used by the ordering equivalence rule when considering whether a
+    /// cast-bearing expression can replace an existing sort key without invalidating
+    /// the ordering.
+    ///
+    /// This function handles *both* `CastExpr` (generic cast) and
+    /// `CastColumnExpr` (field-aware cast) because the planner may introduce either
+    /// form during rewrite steps; the core logic is the same in both cases.  The
+    /// substitution is only allowed when the cast wraps **the very same child
+    /// expression** that the original sort used (an exact-child-match invariant),
+    /// and the casted type must be a widening/order-preserving conversion
+    /// `CastExpr::check_bigger_cast(...)` ensures.  Without those restrictions the
+    /// existing sort order could be violated (e.g. a narrowing cast could collapse
+    /// distinct values together).
+    fn substitute_cast_like_ordering(
+        r_expr: Arc<dyn PhysicalExpr>,
+        sort_expr: &PhysicalSortExpr,
+        expr_type: &DataType,
+    ) -> Option<PhysicalSortExpr> {
+        let (child_expr, cast_type) = if let Some(cast_expr) =
+            r_expr.as_any().downcast_ref::<CastExpr>()
+        {
+            (cast_expr.expr(), cast_expr.cast_type())
+        } else if let Some(cast_expr) = r_expr.as_any().downcast_ref::<CastColumnExpr>() {
+            (cast_expr.expr(), cast_expr.target_field().data_type())
+        } else {
+            return None;
+        };
+
+        (child_expr.eq(&sort_expr.expr)
+            && CastExpr::check_bigger_cast(cast_type, expr_type))
+        .then(|| PhysicalSortExpr::new(r_expr, sort_expr.options))
+    }
+
     /// Creates an empty `EquivalenceProperties` object.
     pub fn new(schema: SchemaRef) -> Self {
         Self {
@@ -844,32 +877,10 @@ impl EquivalenceProperties {
                     let expr_type = sort_expr.expr.data_type(schema).unwrap();
                     // TODO: Add one-to-one analysis for ScalarFunctions.
                     for r_expr in referring_exprs {
-                        // We check whether this expression is substitutable.
-                        if let Some(cast_expr) =
-                            r_expr.as_any().downcast_ref::<CastExpr>()
-                        {
-                            // For casts, we need to know whether the cast
-                            // expression matches:
-                            if cast_expr.expr.eq(&sort_expr.expr)
-                                && cast_expr.is_bigger_cast(&expr_type)
-                            {
-                                result.push(PhysicalSortExpr::new(
-                                    r_expr,
-                                    sort_expr.options,
-                                ));
-                            }
-                        } else if let Some(cast_expr) =
-                            r_expr.as_any().downcast_ref::<CastColumnExpr>()
-                        {
-                            let cast_type = cast_expr.target_field().data_type();
-                            if cast_expr.expr().eq(&sort_expr.expr)
-                                && CastExpr::check_bigger_cast(cast_type, &expr_type)
-                            {
-                                result.push(PhysicalSortExpr::new(
-                                    r_expr,
-                                    sort_expr.options,
-                                ));
-                            }
+                        if let Some(substituted) = Self::substitute_cast_like_ordering(
+                            r_expr, &sort_expr, &expr_type,
+                        ) {
+                            result.push(substituted);
                         }
                     }
                     result.push(sort_expr);