Fix.

Author: mbutrovich

native/Cargo.lock

@@ -370,21 +370,29 @@ impl SparkPhysicalExprAdapter {
         .data()
     }
 
-    /// Replace CastColumnExpr (DataFusion's cast) with Spark's Cast expression.
+    /// Replace CastExpr (DataFusion's cast) with Spark's Cast expression.
     fn replace_with_spark_cast(
         &self,
         expr: Arc<dyn PhysicalExpr>,
     ) -> DataFusionResult<Transformed<Arc<dyn PhysicalExpr>>> {
-        // Check for CastColumnExpr and replace with spark_expr::Cast
-        // CastColumnExpr is in datafusion_physical_expr::expressions
+        // Check for CastExpr and replace with spark_expr::Cast
         if let Some(cast) = expr
             .as_any()
-            .downcast_ref::<datafusion::physical_expr::expressions::CastColumnExpr>()
+            .downcast_ref::<datafusion::physical_expr::expressions::CastExpr>()
         {
             let child = Arc::clone(cast.expr());
-            let physical_type = cast.input_field().data_type();
             let target_type = cast.target_field().data_type();
 
+            // Derive input field from the child Column expression and the physical schema
+            let input_field = if let Some(col) = child.as_any().downcast_ref::<Column>() {
+                Arc::new(self.physical_file_schema.field(col.index()).clone())
+            } else {
+                // Fallback: synthesize a field from the target field name and child data type
+                let child_type = cast.expr().data_type(&self.physical_file_schema)?;
+                Arc::new(Field::new(cast.target_field().name(), child_type, true))
+            };
+            let physical_type = input_field.data_type();
+
             // For complex nested types (Struct, List, Map), Timestamp timezone
             // mismatches, and Timestamp→Int64 (nanosAsLong), use CometCastColumnExpr
             // with spark_parquet_convert which handles field-name-based selection,
@@ -413,7 +421,7 @@ impl SparkPhysicalExprAdapter {
                 let comet_cast: Arc<dyn PhysicalExpr> = Arc::new(
                     CometCastColumnExpr::new(
                         child,
-                        Arc::clone(cast.input_field()),
+                        input_field,
                         Arc::clone(cast.target_field()),
                         None,
                     )

native/core/src/parquet/schema_adapter.rs

@@ -21,7 +21,6 @@ use datafusion::common::{utils::take_function_args, DataFusionError, Result, Sca
 use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
-use std::any::Any;
 use std::sync::Arc;
 
 /// Spark-compatible date_from_unix_date function.
@@ -48,10 +47,6 @@ impl Default for SparkDateFromUnixDate {
 }
 
 impl ScalarUDFImpl for SparkDateFromUnixDate {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "date_from_unix_date"
     }

native/spark-expr/src/datetime_funcs/date_from_unix_date.rs

@@ -32,7 +32,7 @@ use datafusion::logical_expr::{
     ColumnarValue, ScalarFunctionArgs, ScalarUDFImpl, Signature, Volatility,
 };
 use num::integer::div_floor;
-use std::{any::Any, fmt::Debug, sync::Arc};
+use std::{fmt::Debug, sync::Arc};
 
 const MICROS_PER_HOUR: i64 = 3_600_000_000;
 
@@ -56,10 +56,6 @@ impl Default for SparkHoursTransform {
 }
 
 impl ScalarUDFImpl for SparkHoursTransform {
-    fn as_any(&self) -> &dyn Any {
-        self
-    }
-
     fn name(&self) -> &str {
         "hours_transform"
     }