Implement schema-aware CastColumnExpr constructor

Add a new CastColumnExpr::new_with_schema constructor
that accepts and stores the input schema. Document the
column-only helper for single-field validation paths.

Update CastColumnExpr construction to include full input
schemas during schema rewriting and proto parsing, ensuring
correct type resolution.

Author: kosiew

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

@@ -466,11 +466,12 @@ impl<'a> DefaultPhysicalExprAdapterRewriter<'a> {
             }
         }
 
-        let cast_expr = Arc::new(CastColumnExpr::new(
+        let cast_expr = Arc::new(CastColumnExpr::new_with_schema(
             Arc::new(column),
             Arc::new(physical_field.clone()),
             Arc::new(logical_field.clone()),
             None,
+            Arc::clone(&self.physical_file_schema),
         )?);
 
         Ok(Transformed::yes(cast_expr))
@@ -659,8 +660,11 @@ mod tests {
     #[test]
     fn test_rewrite_multi_column_expr_with_type_cast() {
         let (physical_schema, logical_schema) = create_test_schema();
+        let physical_schema = Arc::new(physical_schema);
+        let logical_schema = Arc::new(logical_schema);
         let factory = DefaultPhysicalExprAdapterFactory;
-        let adapter = factory.create(Arc::new(logical_schema), Arc::new(physical_schema));
+        let adapter =
+            factory.create(Arc::clone(&logical_schema), Arc::clone(&physical_schema));
 
         // Create a complex expression: (a + 5) OR (c > 0.0) that tests the recursive case of the rewriter
         let column_a = Arc::new(Column::new("a", 0)) as Arc<dyn PhysicalExpr>;
@@ -685,11 +689,12 @@ mod tests {
 
         let expected = expressions::BinaryExpr::new(
             Arc::new(
-                CastColumnExpr::new(
+                CastColumnExpr::new_with_schema(
                     Arc::new(Column::new("a", 0)),
                     Arc::new(Field::new("a", DataType::Int32, false)),
                     Arc::new(Field::new("a", DataType::Int64, false)),
                     None,
+                    Arc::clone(&physical_schema),
                 )
                 .expect("cast column expr"),
             ),
@@ -765,15 +770,18 @@ mod tests {
             false,
         )]);
 
+        let physical_schema = Arc::new(physical_schema);
+        let logical_schema = Arc::new(logical_schema);
         let factory = DefaultPhysicalExprAdapterFactory;
-        let adapter = factory.create(Arc::new(logical_schema), Arc::new(physical_schema));
+        let adapter =
+            factory.create(Arc::clone(&logical_schema), Arc::clone(&physical_schema));
         let column_expr = Arc::new(Column::new("data", 0));
 
         let result = adapter.rewrite(column_expr).unwrap();
 
         let expected = Arc::new(
-            CastColumnExpr::new(
-                Arc::new(Column::new("data", 0)),
+                CastColumnExpr::new_with_schema(
+                    Arc::new(Column::new("data", 0)),
                 Arc::new(Field::new(
                     "data",
                     DataType::Struct(
@@ -797,6 +805,7 @@ mod tests {
                     false,
                 )),
                 None,
+                Arc::clone(&physical_schema),
             )
             .expect("cast column expr"),
         ) as Arc<dyn PhysicalExpr>;

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

@@ -516,11 +516,12 @@ mod tests {
 
         let input_field = Arc::new(input_schema.field(0).clone());
         let target_field = Arc::new(Field::new("a_cast", DataType::Int64, true));
-        let cast_col = Arc::new(CastColumnExpr::new(
+        let cast_col = Arc::new(CastColumnExpr::new_with_schema(
             Arc::clone(&col_a),
             input_field,
             target_field,
             None,
+            Arc::clone(&input_schema),
         )?) as Arc<dyn PhysicalExpr>;
 
         let proj_exprs = vec![

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

@@ -58,6 +58,8 @@ pub struct CastColumnExpr {
     target_field: FieldRef,
     /// Options forwarded to [`cast_column`].
     cast_options: CastOptions<'static>,
+    /// Schema used to resolve expression data types during construction.
+    input_schema: Arc<Schema>,
 }
 
 // Manually derive `PartialEq`/`Hash` as `Arc<dyn PhysicalExpr>` does not
@@ -82,14 +84,36 @@ impl Hash for CastColumnExpr {
 
 impl CastColumnExpr {
     /// Create a new [`CastColumnExpr`].
+    ///
+    /// This constructor assumes `expr` is a column expression and validates it
+    /// against a single-field schema derived from `input_field`. If the
+    /// expression depends on a broader schema (for example, computed
+    /// expressions), use [`Self::new_with_schema`] instead.
     pub fn new(
         expr: Arc<dyn PhysicalExpr>,
         input_field: FieldRef,
         target_field: FieldRef,
         cast_options: Option<CastOptions<'static>>,
     ) -> Result<Self> {
-        let input_schema = Schema::new(vec![input_field.as_ref().clone()]);
-        let expr_data_type = expr.data_type(&input_schema)?;
+        let input_schema = Arc::new(Schema::new(vec![input_field.as_ref().clone()]));
+        Self::new_with_schema(
+            expr,
+            input_field,
+            target_field,
+            cast_options,
+            input_schema,
+        )
+    }
+
+    /// Create a new [`CastColumnExpr`] using the full input schema.
+    pub fn new_with_schema(
+        expr: Arc<dyn PhysicalExpr>,
+        input_field: FieldRef,
+        target_field: FieldRef,
+        cast_options: Option<CastOptions<'static>>,
+        input_schema: Arc<Schema>,
+    ) -> Result<Self> {
+        let expr_data_type = expr.data_type(input_schema.as_ref())?;
         if input_field.data_type() != &expr_data_type {
             return plan_err!(
                 "CastColumnExpr input field data type '{}' does not match expression data type '{}'",
@@ -125,6 +149,7 @@ impl CastColumnExpr {
             input_field,
             target_field,
             cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS),
+            input_schema,
         })
     }
 
@@ -208,11 +233,12 @@ impl PhysicalExpr for CastColumnExpr {
     ) -> Result<Arc<dyn PhysicalExpr>> {
         assert_eq!(children.len(), 1);
         let child = children.pop().expect("CastColumnExpr child");
-        Ok(Arc::new(Self::new(
+        Ok(Arc::new(Self::new_with_schema(
             child,
             Arc::clone(&self.input_field),
             Arc::clone(&self.target_field),
             Some(self.cast_options.clone()),
+            Arc::clone(&self.input_schema),
         )?))
     }
 
@@ -253,11 +279,12 @@ mod tests {
         let batch = RecordBatch::try_new(Arc::clone(&schema), vec![values])?;
 
         let column = Arc::new(Column::new_with_schema("a", schema.as_ref())?);
-        let expr = CastColumnExpr::new(
+        let expr = CastColumnExpr::new_with_schema(
             column,
             Arc::new(input_field.clone()),
             Arc::new(target_field.clone()),
             None,
+            Arc::clone(&schema),
         )?;
 
         let result = expr.evaluate(&batch)?;
@@ -307,11 +334,12 @@ mod tests {
         )?;
 
         let column = Arc::new(Column::new_with_schema("s", schema.as_ref())?);
-        let expr = CastColumnExpr::new(
+        let expr = CastColumnExpr::new_with_schema(
             column,
             Arc::new(input_field.clone()),
             Arc::new(target_field.clone()),
             None,
+            Arc::clone(&schema),
         )?;
 
         let result = expr.evaluate(&batch)?;
@@ -377,11 +405,12 @@ mod tests {
         )?;
 
         let column = Arc::new(Column::new_with_schema("root", schema.as_ref())?);
-        let expr = CastColumnExpr::new(
+        let expr = CastColumnExpr::new_with_schema(
             column,
             Arc::new(outer_field.clone()),
             Arc::new(target_field.clone()),
             None,
+            Arc::clone(&schema),
         )?;
 
         let result = expr.evaluate(&batch)?;
@@ -428,11 +457,12 @@ mod tests {
         );
         let literal =
             Arc::new(Literal::new(ScalarValue::Struct(Arc::new(scalar_struct))));
-        let expr = CastColumnExpr::new(
+        let expr = CastColumnExpr::new_with_schema(
             literal,
             Arc::new(input_field.clone()),
             Arc::new(target_field.clone()),
             None,
+            Arc::clone(&schema),
         )?;
 
         let batch = RecordBatch::new_empty(Arc::clone(&schema));

datafusion/physical-expr/src/intervals/utils.rs

@@ -214,8 +214,14 @@ mod tests {
 
         let column_expr = col("a", &schema).unwrap();
         let cast_expr = Arc::new(
-            CastColumnExpr::new(column_expr, input_field, target_field, None)
-                .expect("cast column expr"),
+            CastColumnExpr::new_with_schema(
+                column_expr,
+                input_field,
+                target_field,
+                None,
+                Arc::clone(&schema),
+            )
+            .expect("cast column expr"),
         ) as Arc<dyn PhysicalExpr>;
 
         assert!(check_support(&cast_expr, &schema));

datafusion/physical-expr/src/simplifier/unwrap_cast.rs

@@ -222,8 +222,14 @@ mod tests {
         // Create: cast_column(c1 as INT64) > INT64(10)
         let column_expr = col("c1", &schema).unwrap();
         let cast_expr = Arc::new(
-            CastColumnExpr::new(column_expr, input_field, target_field, None)
-                .expect("cast column expr"),
+            CastColumnExpr::new_with_schema(
+                column_expr,
+                input_field,
+                target_field,
+                None,
+                Arc::new(schema.clone()),
+            )
+            .expect("cast column expr"),
         );
         let literal_expr = lit(10i64);
         let binary_expr =

datafusion/proto/src/physical_plan/from_proto.rs

@@ -364,7 +364,7 @@ pub fn parse_physical_expr(
                 e.safe,
                 e.format_options.as_ref(),
             )?;
-            Arc::new(CastColumnExpr::new(
+            Arc::new(CastColumnExpr::new_with_schema(
                 parse_required_physical_expr(
                     e.expr.as_deref(),
                     ctx,
@@ -375,6 +375,7 @@ pub fn parse_physical_expr(
                 Arc::new(Field::try_from(input_field)?),
                 Arc::new(Field::try_from(target_field)?),
                 cast_options,
+                Arc::new(input_schema.clone()),
             )?)
         }
         ExprType::TryCast(e) => Arc::new(TryCastExpr::new(

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -219,19 +219,20 @@ fn roundtrip_cast_column_expr() -> Result<()> {
         safe: true,
         format_options,
     };
-    let expr: Arc<dyn PhysicalExpr> = Arc::new(CastColumnExpr::new(
+    let input_schema = Schema::new(vec![input_field.clone()]);
+    let expr: Arc<dyn PhysicalExpr> = Arc::new(CastColumnExpr::new_with_schema(
         Arc::new(Column::new("a", 0)),
         Arc::new(input_field.clone()),
         Arc::new(target_field.clone()),
         Some(cast_options.clone()),
+        Arc::new(input_schema.clone()),
     )?);
 
     let ctx = SessionContext::new();
     let codec = DefaultPhysicalExtensionCodec {};
     let proto = datafusion_proto::physical_plan::to_proto::serialize_physical_expr(
         &expr, &codec,
     )?;
-    let input_schema = Schema::new(vec![input_field.clone()]);
     let round_trip = datafusion_proto::physical_plan::from_proto::parse_physical_expr(
         &proto,
         &ctx.task_ctx(),
@@ -244,11 +245,12 @@ fn roundtrip_cast_column_expr() -> Result<()> {
         .downcast_ref::<CastColumnExpr>()
         .ok_or_else(|| internal_datafusion_err!("Expected CastColumnExpr"))?;
 
-    let expected = CastColumnExpr::new(
+    let expected = CastColumnExpr::new_with_schema(
         Arc::new(Column::new("a", 0)),
         Arc::new(input_field.clone()),
         Arc::new(target_field.clone()),
         Some(cast_options),
+        Arc::new(input_schema.clone()),
     )?;
 
     assert_eq!(cast_expr, &expected);
@@ -274,19 +276,20 @@ fn roundtrip_cast_column_expr_with_target_field_change() -> Result<()> {
     let target_field =
         Field::new("payload_cast", DataType::Utf8, false).with_metadata(target_metadata);
 
-    let expr: Arc<dyn PhysicalExpr> = Arc::new(CastColumnExpr::new(
+    let input_schema = Schema::new(vec![input_field.clone()]);
+    let expr: Arc<dyn PhysicalExpr> = Arc::new(CastColumnExpr::new_with_schema(
         Arc::new(Column::new("payload", 0)),
         Arc::new(input_field.clone()),
         Arc::new(target_field.clone()),
         None,
+        Arc::new(input_schema.clone()),
     )?);
 
     let ctx = SessionContext::new();
     let codec = DefaultPhysicalExtensionCodec {};
     let proto = datafusion_proto::physical_plan::to_proto::serialize_physical_expr(
         &expr, &codec,
     )?;
-    let input_schema = Schema::new(vec![input_field.clone()]);
     let round_trip = datafusion_proto::physical_plan::from_proto::parse_physical_expr(
         &proto,
         &ctx.task_ctx(),