physical-expr: Normalize CastColumnExpr with default format options

Simplify CastColumnExpr constructor to ensure format options are always
present by using the FormatOptionsSlot trait. Add new_with_schema method
for cases requiring full schema. Update schema_rewriter.rs to properly
wrap Schema references in Arc. Add default format options fallback in
serialization for CastColumnExpr to protobuf.

Author: kosiew

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

@@ -471,7 +471,7 @@ impl<'a> DefaultPhysicalExprAdapterRewriter<'a> {
             Arc::new(physical_field.clone()),
             Arc::new(logical_field.clone()),
             None,
-            Arc::clone(&self.physical_file_schema),
+            Arc::new(self.physical_file_schema.clone()),
         )?);
 
         Ok(Transformed::yes(cast_expr))

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

@@ -22,9 +22,11 @@ use arrow::{
     compute::{CastOptions, can_cast_types},
     datatypes::{DataType, FieldRef, Schema},
     record_batch::RecordBatch,
+    util::display::FormatOptions as ArrowFormatOptions,
 };
 use datafusion_common::{
-    Result, ScalarValue, format::DEFAULT_CAST_OPTIONS,
+    Result, ScalarValue,
+    format::DEFAULT_CAST_OPTIONS,
     nested_struct::{cast_column, validate_struct_compatibility},
     plan_err,
 };
@@ -82,37 +84,94 @@ impl Hash for CastColumnExpr {
     }
 }
 
+trait FormatOptionsSlot {
+    fn ensure_present(&mut self, default: ArrowFormatOptions<'static>);
+}
+
+impl FormatOptionsSlot for ArrowFormatOptions<'static> {
+    fn ensure_present(&mut self, _default: ArrowFormatOptions<'static>) {}
+}
+
+impl FormatOptionsSlot for Option<ArrowFormatOptions<'static>> {
+    fn ensure_present(&mut self, default: ArrowFormatOptions<'static>) {
+        if self.is_none() {
+            *self = Some(default);
+        }
+    }
+}
+
 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.
+    /// This constructor ensures that format options are populated with defaults,
+    /// normalizing the CastOptions for consistent behavior during serialization
+    /// and evaluation.
     pub fn new(
         expr: Arc<dyn PhysicalExpr>,
         input_field: FieldRef,
         target_field: FieldRef,
         cast_options: Option<CastOptions<'static>>,
     ) -> Result<Self> {
-        let input_schema = Arc::new(Schema::new(vec![input_field.as_ref().clone()]));
-        Self::new_with_schema(
+        let mut cast_options = cast_options.unwrap_or(DEFAULT_CAST_OPTIONS);
+        cast_options
+            .format_options
+            .ensure_present(DEFAULT_CAST_OPTIONS.format_options.clone());
+        let input_schema = Schema::new(vec![input_field.as_ref().clone()]);
+        let expr_data_type = expr.data_type(&input_schema)?;
+        if input_field.data_type() != &expr_data_type {
+            return plan_err!(
+                "CastColumnExpr input field data type '{}' does not match expression data type '{}'",
+                input_field.data_type(),
+                expr_data_type
+            );
+        }
+
+        match (input_field.data_type(), target_field.data_type()) {
+            (DataType::Struct(source_fields), DataType::Struct(target_fields)) => {
+                validate_struct_compatibility(source_fields, target_fields)?;
+            }
+            (_, DataType::Struct(_)) => {
+                return plan_err!(
+                    "CastColumnExpr cannot cast non-struct input '{}' to struct target '{}'",
+                    input_field.data_type(),
+                    target_field.data_type()
+                );
+            }
+            _ => {
+                if !can_cast_types(input_field.data_type(), target_field.data_type()) {
+                    return plan_err!(
+                        "CastColumnExpr cannot cast input type '{}' to target type '{}'",
+                        input_field.data_type(),
+                        target_field.data_type()
+                    );
+                }
+            }
+        }
+
+        Ok(Self {
             expr,
             input_field,
             target_field,
             cast_options,
-            input_schema,
-        )
+            input_schema: Arc::new(input_schema),
+        })
     }
 
-    /// Create a new [`CastColumnExpr`] using the full input schema.
+    /// Create a new [`CastColumnExpr`] with a specific input schema.
+    ///
+    /// This constructor is useful when the expression depends on multiple
+    /// fields from a broader 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 mut cast_options = cast_options.unwrap_or(DEFAULT_CAST_OPTIONS);
+        cast_options
+            .format_options
+            .ensure_present(DEFAULT_CAST_OPTIONS.format_options.clone());
         let expr_data_type = expr.data_type(input_schema.as_ref())?;
         if input_field.data_type() != &expr_data_type {
             return plan_err!(
@@ -148,7 +207,7 @@ impl CastColumnExpr {
             expr,
             input_field,
             target_field,
-            cast_options: cast_options.unwrap_or(DEFAULT_CAST_OPTIONS),
+            cast_options,
             input_schema,
         })
     }

datafusion/proto/src/physical_plan/to_proto.rs

@@ -23,7 +23,8 @@ use arrow::datatypes::Schema;
 use arrow::ipc::writer::StreamWriter;
 use arrow::util::display::{DurationFormat, FormatOptions as ArrowFormatOptions};
 use datafusion_common::{
-    DataFusionError, Result, internal_datafusion_err, internal_err, not_impl_err,
+    DataFusionError, Result, format::DEFAULT_CAST_OPTIONS, internal_datafusion_err,
+    internal_err, not_impl_err,
 };
 use datafusion_datasource::file_scan_config::FileScanConfig;
 use datafusion_datasource::file_sink_config::FileSink;
@@ -375,12 +376,10 @@ pub fn serialize_physical_expr(
         })
     } else if let Some(cast_column) = expr.downcast_ref::<CastColumnExpr>() {
         let cast_options = serialize_cast_options(cast_column.cast_options())?;
-        let format_options = cast_options
-            .format_options
-            .clone()
-            .ok_or_else(|| {
-                internal_datafusion_err!("Missing format options for cast column")
-            })?;
+        let format_options = match cast_options.format_options.clone() {
+            Some(format_options) => format_options,
+            None => serialize_format_options(&DEFAULT_CAST_OPTIONS.format_options)?,
+        };
         Ok(protobuf::PhysicalExprNode {
             expr_type: Some(protobuf::physical_expr_node::ExprType::CastColumn(
                 Box::new(protobuf::PhysicalCastColumnNode {

datafusion/proto/tests/cases/roundtrip_physical_plan.rs

@@ -108,8 +108,8 @@ use datafusion_common::file_options::json_writer::JsonWriterOptions;
 use datafusion_common::parsers::CompressionTypeVariant;
 use datafusion_common::stats::Precision;
 use datafusion_common::{
-    DataFusionError, NullEquality, Result, UnnestOptions, internal_datafusion_err,
-    internal_err, not_impl_err,
+    DataFusionError, NullEquality, Result, UnnestOptions, format::DEFAULT_CAST_OPTIONS,
+    internal_datafusion_err, internal_err, not_impl_err,
 };
 use datafusion_expr::async_udf::{AsyncScalarUDF, AsyncScalarUDFImpl};
 use datafusion_expr::{
@@ -118,6 +118,20 @@ use datafusion_expr::{
     WindowFrame, WindowFrameBound, WindowUDF,
 };
 use datafusion_functions_aggregate::average::avg_udaf;
+
+trait FormatOptionsSlot {
+    fn clear(&mut self);
+}
+
+impl FormatOptionsSlot for FormatOptions<'static> {
+    fn clear(&mut self) {}
+}
+
+impl FormatOptionsSlot for Option<FormatOptions<'static>> {
+    fn clear(&mut self) {
+        *self = None;
+    }
+}
 use datafusion_functions_aggregate::nth_value::nth_value_udaf;
 use datafusion_functions_aggregate::string_agg::string_agg_udaf;
 use datafusion_proto::physical_plan::{
@@ -264,6 +278,49 @@ fn roundtrip_cast_column_expr() -> Result<()> {
     Ok(())
 }
 
+#[test]
+fn roundtrip_cast_column_expr_with_missing_format_options() -> Result<()> {
+    let input_field = Field::new("a", DataType::Int32, true);
+    let target_field = Field::new("a", DataType::Int64, true);
+
+    let mut cast_options = CastOptions {
+        safe: true,
+        format_options: DEFAULT_CAST_OPTIONS.format_options.clone(),
+    };
+    cast_options.format_options.clear();
+    let expr: Arc<dyn PhysicalExpr> = Arc::new(CastColumnExpr::new(
+        Arc::new(Column::new("a", 0)),
+        Arc::new(input_field.clone()),
+        Arc::new(target_field.clone()),
+        Some(cast_options),
+    )?);
+
+    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(),
+        &input_schema,
+        &codec,
+    )?;
+
+    let cast_expr = round_trip
+        .as_any()
+        .downcast_ref::<CastColumnExpr>()
+        .ok_or_else(|| internal_datafusion_err!("Expected CastColumnExpr"))?;
+
+    assert_eq!(
+        cast_expr.cast_options().format_options,
+        DEFAULT_CAST_OPTIONS.format_options
+    );
+
+    Ok(())
+}
+
 #[test]
 fn roundtrip_cast_column_expr_with_target_field_change() -> Result<()> {
     let mut input_metadata = HashMap::new();