Remove CastColumnExpr and custom_file_casts example; unify on field-aware CastExpr (#21563)

## Which issue does this PR close?

* Closes #20164

---

## Rationale for this change

This PR completes the cast unification effort by removing the remaining
public and example surface of `CastColumnExpr`. Keeping both `CastExpr`
and `CastColumnExpr` introduced duplication in casting logic, increased
maintenance overhead, and made it easier for downstream code to
reintroduce divergent behavior.

With the introduction of field-aware `CastExpr` (via
`new_with_target_field`), all functionality previously covered by
`CastColumnExpr` can now be expressed through a single, consistent
abstraction. This simplifies the mental model for casting, ensures
consistent behavior across execution paths, and aligns documentation and
examples with the unified design.

Additionally, this change ensures that test coverage and examples
reflect the final architecture, reducing confusion for contributors and
users.

---

## What changes are included in this PR?

* **Removal of `CastColumnExpr`**

  * Deleted `datafusion/physical-expr/src/expressions/cast_column.rs`.
* Removed module declaration and public re-export from
`expressions/mod.rs`.

* **Migration to field-aware `CastExpr`**

  * Updated code paths to rely exclusively on `CastExpr`.
* Replaced usages of `data_type(&schema)` with
`target_field().data_type()` where appropriate.

* **Example updates**

  * Updated `custom_file_casts` example to only handle `CastExpr`.
* Removed branching logic that handled both `CastExpr` and
`CastColumnExpr`.
  * Updated CLI usage string to reflect current examples.

* **Test consolidation and improvements**

* Migrated struct, nested struct, and scalar struct cast tests into
`cast.rs`.
  * Added comprehensive field-aware cast tests covering:

    * Target field metadata preservation
    * Nullability semantics
    * Struct casting with missing fields
    * Nested struct casting
    * Struct scalar casting

* **Refactoring and cleanup**

* Renamed helper functions for clarity (e.g., `assert_cast_column` →
`assert_cast_input_column`).
* Simplified adapter logic to operate on a single cast expression type.

* **Documentation updates**

* Added upgrade note documenting removal of `CastColumnExpr` and
migration guidance.

---

## Are these changes tested?

Yes.

* Existing tests were updated to reflect the unified cast model.
* Coverage from `cast_column.rs` was preserved and migrated into
`cast.rs`.
* New tests validate:

  * Field-aware `CastExpr` semantics
  * Struct and nested struct casting behavior
  * Scalar struct casting
  * Metadata and nullability propagation
* Adapter and schema rewriting tests were updated and remain green.

---

## Are there any user-facing changes?

Yes.

* **Breaking change**: `CastColumnExpr` has been removed from the public
API.

* Users must migrate to `CastExpr`:

  * Use `CastExpr::new_with_target_field(...)` for field-aware casting.
  * Access output metadata via `target_field()`.
  * Access input expressions via `expr()`.

* Documentation has been updated with migration guidance.

---

## 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-examples/examples/custom_data_source/custom_file_casts.rs

@@ -33,7 +33,7 @@ use datafusion::execution::context::SessionContext;
 use datafusion::execution::object_store::ObjectStoreUrl;
 use datafusion::parquet::arrow::ArrowWriter;
 use datafusion::physical_expr::PhysicalExpr;
-use datafusion::physical_expr::expressions::{CastColumnExpr, CastExpr};
+use datafusion::physical_expr::expressions::CastExpr;
 use datafusion::prelude::SessionConfig;
 use datafusion_physical_expr_adapter::{
     DefaultPhysicalExprAdapterFactory, PhysicalExprAdapter, PhysicalExprAdapterFactory,
@@ -43,9 +43,10 @@ use object_store::path::Path;
 use object_store::{ObjectStore, ObjectStoreExt, PutPayload};
 
 // Example showing how to implement custom casting rules to adapt file schemas.
-// This example enforces that casts must be strictly widening: if the file type is Int64 and the table type is Int32, it will error
-// before even reading the data.
-// Without this custom cast rule DataFusion would happily do the narrowing cast, potentially erroring only if it found a row with data it could not cast.
+// This example enforces strictly widening casts: if the file type is Int64 and
+// the table type is Int32, it errors before reading the data. Without this
+// custom cast rule DataFusion would apply the narrowing cast and might only
+// error after reading a row that it could not cast.
 pub async fn custom_file_casts() -> Result<()> {
     println!("=== Creating example data ===");
 
@@ -139,7 +140,7 @@ async fn write_data(
     Ok(())
 }
 
-/// Factory for creating DefaultValuePhysicalExprAdapter instances
+/// Factory for creating custom cast physical expression adapters
 #[derive(Debug)]
 struct CustomCastPhysicalExprAdapterFactory {
     inner: Arc<dyn PhysicalExprAdapterFactory>,
@@ -167,8 +168,8 @@ impl PhysicalExprAdapterFactory for CustomCastPhysicalExprAdapterFactory {
     }
 }
 
-/// Custom PhysicalExprAdapter that handles missing columns with default values from metadata
-/// and wraps DefaultPhysicalExprAdapter for standard schema adaptation
+/// Custom `PhysicalExprAdapter` that wraps the default adapter and rejects
+/// narrowing file-schema casts.
 #[derive(Debug, Clone)]
 struct CustomCastsPhysicalExprAdapter {
     physical_file_schema: SchemaRef,
@@ -177,34 +178,23 @@ struct CustomCastsPhysicalExprAdapter {
 
 impl PhysicalExprAdapter for CustomCastsPhysicalExprAdapter {
     fn rewrite(&self, mut expr: Arc<dyn PhysicalExpr>) -> Result<Arc<dyn PhysicalExpr>> {
-        // First delegate to the inner adapter to handle missing columns and discover any necessary casts
+        // First delegate to the inner adapter to handle standard schema adaptation
+        // and discover any necessary casts.
         expr = self.inner.rewrite(expr)?;
-        // Now we can apply custom casting rules or even swap out all CastExprs for a custom cast kernel / expression
-        // For example, [DataFusion Comet](https://github.com/apache/datafusion-comet) has a [custom cast kernel](https://github.com/apache/datafusion-comet/blob/b4ac876ab420ed403ac7fc8e1b29f42f1f442566/native/spark-expr/src/conversion_funcs/cast.rs#L133-L138).
+        // Now apply custom casting rules or swap CastExprs for a custom cast
+        // kernel / expression. For example, DataFusion Comet has a custom cast
+        // kernel in its native Spark expression implementation.
         expr.transform(|expr| {
             if let Some(cast) = expr.as_any().downcast_ref::<CastExpr>() {
                 let input_data_type =
                     cast.expr().data_type(&self.physical_file_schema)?;
-                let output_data_type = cast.data_type(&self.physical_file_schema)?;
+                let output_data_type = cast.target_field().data_type();
                 if !cast.is_bigger_cast(&input_data_type) {
                     return not_impl_err!(
                         "Unsupported CAST from {input_data_type} to {output_data_type}"
                     );
                 }
             }
-            if let Some(cast) = expr.as_any().downcast_ref::<CastColumnExpr>() {
-                let input_data_type =
-                    cast.expr().data_type(&self.physical_file_schema)?;
-                let output_data_type = cast.data_type(&self.physical_file_schema)?;
-                if !CastExpr::check_bigger_cast(
-                    cast.target_field().data_type(),
-                    &input_data_type,
-                ) {
-                    return not_impl_err!(
-                        "Unsupported CAST from {input_data_type} to {output_data_type}"
-                    );
-                }
-            }
             Ok(Transformed::no(expr))
         })
         .data()

datafusion-examples/examples/custom_data_source/main.rs

@@ -21,7 +21,7 @@
 //!
 //! ## Usage
 //! ```bash
-//! cargo run --example custom_data_source -- [all|csv_json_opener|csv_sql_streaming|custom_datasource|custom_file_casts|custom_file_format|default_column_values|file_stream_provider]
+//! cargo run --example custom_data_source -- [all|adapter_serialization|csv_json_opener|csv_sql_streaming|custom_datasource|custom_file_casts|custom_file_format|default_column_values|file_stream_provider]
 //! ```
 //!
 //! Each subcommand runs a corresponding example:

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

@@ -644,7 +644,7 @@ mod tests {
             .expect("Expected CastExpr")
     }
 
-    fn assert_cast_column(cast_expr: &CastExpr, name: &str, index: usize) {
+    fn assert_cast_input_column(cast_expr: &CastExpr, name: &str, index: usize) {
         let inner_col = cast_expr
             .expr()
             .as_any()
@@ -771,7 +771,7 @@ mod tests {
 
         let left_cast = assert_cast_expr(left.left());
         assert_eq!(left_cast.target_field().data_type(), &DataType::Int64);
-        assert_cast_column(left_cast, "a", 0);
+        assert_cast_input_column(left_cast, "a", 0);
 
         let right = outer
             .right()
@@ -1672,7 +1672,7 @@ mod tests {
         let cast_expr = assert_cast_expr(&result);
 
         // Verify the inner column points to the correct physical index (1)
-        assert_cast_column(cast_expr, "a", 1);
+        assert_cast_input_column(cast_expr, "a", 1);
 
         // Verify cast types
         assert_eq!(
@@ -1692,7 +1692,7 @@ mod tests {
         // Regression: this must still resolve against physical field `a` by name.
         let rewritten = adapter.rewrite(Arc::new(Column::new("a", 0))).unwrap();
         let cast_expr = assert_cast_expr(&rewritten);
-        assert_cast_column(cast_expr, "a", 1);
+        assert_cast_input_column(cast_expr, "a", 1);
         assert_eq!(cast_expr.target_field().data_type(), &DataType::Int64);
     }
 }