Preserve logical cast field semantics during physical lowering with field-aware CastExpr (apache#20836)

## Which issue does this PR close?

* Part of apache#20164

---

## Rationale for this change

The current physical planning path for `Expr::Cast` discards logical
field information (name, nullability, and metadata) by lowering casts
using only the target `DataType`. This results in a loss of semantic
fidelity between logical and physical plans, particularly for
metadata-bearing fields and same-type casts with explicit field intent.

Additionally, the planner previously rejected casts with metadata due to
limitations of the type-only casting API, creating inconsistencies with
other parts of the system (e.g. adapter-generated expressions).

This change introduces a field-aware casting path that preserves logical
intent throughout physical lowering, ensuring consistent semantics
across planner and adapter outputs.

---

## What changes are included in this PR?

* Introduced `cast_with_target_field` to construct `CastExpr` using full
`FieldRef` semantics (name, nullability, metadata).
* Refactored existing `cast_with_options` to delegate to the new
field-aware helper.
* Moved `is_default_target_field` to a shared helper function for reuse.
* Updated planner (`planner.rs`) to use `cast_with_target_field` instead
of type-only casting.
* Removed metadata rejection logic during cast lowering.
* Ensured same-type casts preserve explicit field semantics unless the
target field is default.
* Adjusted cast construction to validate compatibility before building
expressions.
* Exported `cast_with_target_field` for internal planner use.

---

## Are these changes tested?

Yes.

Added planner-focused unit tests to validate:

* Preservation of target field metadata during cast lowering
* Correct propagation of nullability semantics
* Proper handling of same-type casts with explicit field overrides
* No regression for standard type-only casts
* Rejection behavior for unsupported extension type casts via `TryCast`

These tests ensure both backward compatibility and correctness of the
new semantics.

---

## Are there any user-facing changes?

Yes, behaviorally (but not API-breaking):

* Cast expressions now preserve logical field metadata and nullability
in physical plans.
* Previously rejected metadata-bearing casts are now supported.
* Same-type casts may now produce a `CastExpr` when explicit field
semantics are provided.

There are no breaking changes to public APIs, but downstream consumers
that relied on previous planner behavior (e.g. metadata stripping or
cast elision) may observe differences.

---

## 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/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
@@ -201,6 +195,12 @@ impl CastExpr {
     }
 }
 
+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 is_order_preserving_cast_family(
     source_type: &DataType,
     target_type: &DataType,
@@ -315,26 +315,55 @@ pub fn cast_with_options(
     input_schema: &Schema,
     cast_type: DataType,
     cast_options: Option<CastOptions<'static>>,
+) -> Result<Arc<dyn PhysicalExpr>> {
+    cast_with_target_field(
+        expr,
+        input_schema,
+        cast_type.into_nullable_field_ref(),
+        cast_options,
+    )
+}
+
+/// Return a PhysicalExpression representing `expr` casted to `target_field`,
+/// preserving any explicit field semantics such as name, nullability, and
+/// metadata.
+///
+/// If the input expression already has the same data type, this helper still
+/// preserves an explicit `target_field` by constructing a field-aware
+/// [`CastExpr`]. Only the default synthesized field created by the legacy
+/// type-only API is elided back to the original child expression.
+pub fn cast_with_target_field(
+    expr: Arc<dyn PhysicalExpr>,
+    input_schema: &Schema,
+    target_field: FieldRef,
+    cast_options: Option<CastOptions<'static>>,
 ) -> Result<Arc<dyn PhysicalExpr>> {
     let expr_type = expr.data_type(input_schema)?;
-    if expr_type == cast_type {
-        Ok(Arc::clone(&expr))
-    } 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(expr, cast_type, 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(expr, cast_type, cast_options)))
+    let cast_type = target_field.data_type();
+    if expr_type == *cast_type && is_default_target_field(&target_field) {
+        return Ok(Arc::clone(&expr));
+    }
+
+    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/expressions/mod.rs

@@ -56,3 +56,5 @@ pub use no_op::NoOp;
 pub use not::{NotExpr, not};
 pub use try_cast::{TryCastExpr, try_cast};
 pub use unknown_column::UnKnownColumn;
+
+pub(crate) use cast::cast_with_target_field;

datafusion/physical-expr/src/planner.rs

@@ -288,25 +288,12 @@ pub fn create_physical_expr(
                 };
             Ok(expressions::case(expr, when_then_expr, else_expr)?)
         }
-        Expr::Cast(Cast { expr, field }) => {
-            if !field.metadata().is_empty() {
-                let (_, src_field) = expr.to_field(input_dfschema)?;
-                return plan_err!(
-                    "Cast from {} to {} is not supported",
-                    format_type_and_metadata(
-                        src_field.data_type(),
-                        Some(src_field.metadata()),
-                    ),
-                    format_type_and_metadata(field.data_type(), Some(field.metadata()))
-                );
-            }
-
-            expressions::cast(
-                create_physical_expr(expr, input_dfschema, execution_props)?,
-                input_schema,
-                field.data_type().clone(),
-            )
-        }
+        Expr::Cast(Cast { expr, field }) => expressions::cast_with_target_field(
+            create_physical_expr(expr, input_dfschema, execution_props)?,
+            input_schema,
+            Arc::clone(field),
+            None,
+        ),
         Expr::TryCast(TryCast { expr, field }) => {
             if !field.metadata().is_empty() {
                 let (_, src_field) = expr.to_field(input_dfschema)?;
@@ -445,11 +432,26 @@ pub fn logical2physical(expr: &Expr, schema: &Schema) -> Arc<dyn PhysicalExpr> {
 mod tests {
     use arrow::array::{ArrayRef, BooleanArray, RecordBatch, StringArray};
     use arrow::datatypes::{DataType, Field};
-    use datafusion_common::datatype::DataTypeExt;
     use datafusion_expr::col;
 
     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())
+    }
+
+    fn as_planner_cast(physical: &Arc<dyn PhysicalExpr>) -> &expressions::CastExpr {
+        physical
+            .as_any()
+            .downcast_ref::<expressions::CastExpr>()
+            .expect("planner should lower logical CAST to CastExpr")
+    }
+
     #[test]
     fn test_create_physical_expr_scalar_input_output() -> Result<()> {
         let expr = col("letter").eq(lit("A"));
@@ -476,36 +478,63 @@ mod tests {
     }
 
     #[test]
-    fn test_cast_to_extension_type() -> Result<()> {
-        let extension_field_type = Arc::new(
-            DataType::FixedSizeBinary(16)
-                .into_nullable_field()
-                .with_metadata(
-                    [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())]
-                        .into(),
-                ),
+    fn test_cast_lowering_preserves_target_field_metadata() -> Result<()> {
+        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()),
         );
-        let expr = lit("3230e5d4-888e-408b-b09b-831f44aa0c58");
         let cast_expr = Expr::Cast(Cast::new_from_field(
-            Box::new(expr.clone()),
-            Arc::clone(&extension_field_type),
+            Box::new(col("a")),
+            Arc::clone(&target_field),
         ));
-        let err =
-            create_physical_expr(&cast_expr, &DFSchema::empty(), &ExecutionProps::new())
-                .unwrap_err();
-        assert!(err.message().contains("arrow.uuid"));
-
-        let try_cast_expr = Expr::TryCast(TryCast::new_from_field(
-            Box::new(expr.clone()),
-            Arc::clone(&extension_field_type),
+
+        let physical = lower_cast_expr(&cast_expr, &schema)?;
+        let cast = as_planner_cast(&physical);
+
+        assert_eq!(cast.target_field(), &target_field);
+        assert_eq!(physical.return_field(&schema)?, target_field);
+        assert!(physical.nullable(&schema)?);
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_cast_lowering_preserves_standard_cast_semantics() -> Result<()> {
+        let schema = test_cast_schema();
+        let cast_expr = Expr::Cast(Cast::new(Box::new(col("a")), DataType::Int64));
+
+        let physical = lower_cast_expr(&cast_expr, &schema)?;
+        let cast = as_planner_cast(&physical);
+        let returned_field = physical.return_field(&schema)?;
+
+        assert_eq!(cast.cast_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 = 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(),
+            ),
+        );
+        let cast_expr = Expr::Cast(Cast::new_from_field(
+            Box::new(col("a")),
+            Arc::clone(&target_field),
         ));
-        let err = create_physical_expr(
-            &try_cast_expr,
-            &DFSchema::empty(),
-            &ExecutionProps::new(),
-        )
-        .unwrap_err();
-        assert!(err.message().contains("arrow.uuid"));
+
+        let physical = lower_cast_expr(&cast_expr, &schema)?;
+        let cast = as_planner_cast(&physical);
+
+        assert_eq!(cast.target_field(), &target_field);
+        assert_eq!(physical.return_field(&schema)?, target_field);
+        assert!(physical.nullable(&schema)?);
 
         Ok(())
     }

datafusion/sqllogictest/src/test_context.rs

@@ -28,14 +28,17 @@ use arrow::array::{
     LargeStringArray, StringArray, TimestampNanosecondArray, UnionArray,
 };
 use arrow::buffer::ScalarBuffer;
-use arrow::datatypes::{DataType, Field, Schema, SchemaRef, TimeUnit, UnionFields};
+use arrow::datatypes::{
+    DataType, Field, FieldRef, Schema, SchemaRef, TimeUnit, UnionFields,
+};
 use arrow::record_batch::RecordBatch;
 use datafusion::catalog::{
     CatalogProvider, MemoryCatalogProvider, MemorySchemaProvider, SchemaProvider, Session,
 };
 use datafusion::common::{DataFusionError, Result, not_impl_err};
 use datafusion::functions::math::abs;
 use datafusion::logical_expr::async_udf::{AsyncScalarUDF, AsyncScalarUDFImpl};
+use datafusion::logical_expr::planner::TypePlanner;
 use datafusion::logical_expr::{
     ColumnarValue, Expr, ScalarFunctionArgs, ScalarUDF, ScalarUDFImpl, Signature,
     Volatility, create_udf,
@@ -54,6 +57,7 @@ use datafusion::common::cast::as_float64_array;
 use datafusion::execution::SessionStateBuilder;
 use datafusion::execution::runtime_env::RuntimeEnv;
 use log::info;
+use sqlparser::ast;
 use tempfile::TempDir;
 
 /// Context for running tests
@@ -64,6 +68,23 @@ pub struct TestContext {
     test_dir: Option<TempDir>,
 }
 
+#[derive(Debug)]
+struct SqlLogicTestTypePlanner;
+
+impl TypePlanner for SqlLogicTestTypePlanner {
+    fn plan_type_field(&self, sql_type: &ast::DataType) -> Result<Option<FieldRef>> {
+        match sql_type {
+            ast::DataType::Uuid => Ok(Some(Arc::new(
+                Field::new("", DataType::FixedSizeBinary(16), true).with_metadata(
+                    [("ARROW:extension:name".to_string(), "arrow.uuid".to_string())]
+                        .into(),
+                ),
+            ))),
+            _ => Ok(None),
+        }
+    }
+}
+
 impl TestContext {
     pub fn new(ctx: SessionContext) -> Self {
         Self {
@@ -92,6 +113,14 @@ impl TestContext {
             state_builder = state_builder.with_spark_features();
         }
 
+        if matches!(
+            relative_path.file_name().and_then(|name| name.to_str()),
+            Some("cast_extension_type_metadata.slt")
+        ) {
+            state_builder =
+                state_builder.with_type_planner(Arc::new(SqlLogicTestTypePlanner));
+        }
+
         let state = state_builder.build();
 
         let mut test_ctx = TestContext::new(SessionContext::new_with_state(state));

datafusion/sqllogictest/test_files/cast_extension_type_metadata.slt

@@ -0,0 +1,49 @@
+# 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.
+
+# Regression tests for logical CAST targets that carry explicit field metadata.
+
+query ?T
+SELECT
+    CAST(
+        arrow_cast(X'00010203040506070809000102030506', 'FixedSizeBinary(16)')
+        AS UUID
+    ),
+    arrow_metadata(
+        CAST(
+            arrow_cast(X'00010203040506070809000102030506', 'FixedSizeBinary(16)')
+            AS UUID
+        ),
+        'ARROW:extension:name'
+    );
+----
+00010203040506070809000102030506 arrow.uuid
+
+query ?T
+SELECT
+    CAST(raw AS UUID),
+    arrow_metadata(CAST(raw AS UUID), 'ARROW:extension:name')
+FROM (
+    VALUES (
+        arrow_cast(X'00010203040506070809000102030506', 'FixedSizeBinary(16)')
+    )
+) AS uuids(raw);
+----
+00010203040506070809000102030506 arrow.uuid
+
+statement error DataFusion error: Optimizer rule 'simplify_expressions' failed[\s\S]*TryCast from FixedSizeBinary\(16\) to FixedSizeBinary\(16\)<\{"ARROW:extension:name": "arrow\.uuid"\}> is not supported
+SELECT TRY_CAST(arrow_cast(X'00010203040506070809000102030506', 'FixedSizeBinary(16)') AS UUID);