Add cast column format option fields to protobuf

Implement cast column format option fields in the protobuf
schema for physical plans. Serialize and reconstruct full
cast format options during protobuf conversions, handling
format option mapping and fallbacks. Extend the cast
column roundtrip test to verify serialization fidelity
with non-default format options.

Author: kosiew

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

@@ -684,12 +684,15 @@ mod tests {
         println!("Rewritten expression: {result}");
 
         let expected = expressions::BinaryExpr::new(
-            Arc::new(CastColumnExpr::new(
-                Arc::new(Column::new("a", 0)),
-                Arc::new(Field::new("a", DataType::Int32, false)),
-                Arc::new(Field::new("a", DataType::Int64, false)),
-                None,
-            ).unwrap()),
+            Arc::new(
+                CastColumnExpr::new(
+                    Arc::new(Column::new("a", 0)),
+                    Arc::new(Field::new("a", DataType::Int32, false)),
+                    Arc::new(Field::new("a", DataType::Int64, false)),
+                    None,
+                )
+                .unwrap(),
+            ),
             Operator::Plus,
             Arc::new(expressions::Literal::new(ScalarValue::Int64(Some(5)))),
         );
@@ -768,32 +771,35 @@ mod tests {
 
         let result = adapter.rewrite(column_expr).unwrap();
 
-        let expected = Arc::new(CastColumnExpr::new(
-            Arc::new(Column::new("data", 0)),
-            Arc::new(Field::new(
-                "data",
-                DataType::Struct(
-                    vec![
-                        Field::new("id", DataType::Int32, false),
-                        Field::new("name", DataType::Utf8, true),
-                    ]
-                    .into(),
-                ),
-                false,
-            )),
-            Arc::new(Field::new(
-                "data",
-                DataType::Struct(
-                    vec![
-                        Field::new("id", DataType::Int64, false),
-                        Field::new("name", DataType::Utf8View, true),
-                    ]
-                    .into(),
-                ),
-                false,
-            )),
-            None,
-        ).unwrap()) as Arc<dyn PhysicalExpr>;
+        let expected = Arc::new(
+            CastColumnExpr::new(
+                Arc::new(Column::new("data", 0)),
+                Arc::new(Field::new(
+                    "data",
+                    DataType::Struct(
+                        vec![
+                            Field::new("id", DataType::Int32, false),
+                            Field::new("name", DataType::Utf8, true),
+                        ]
+                        .into(),
+                    ),
+                    false,
+                )),
+                Arc::new(Field::new(
+                    "data",
+                    DataType::Struct(
+                        vec![
+                            Field::new("id", DataType::Int64, false),
+                            Field::new("name", DataType::Utf8View, true),
+                        ]
+                        .into(),
+                    ),
+                    false,
+                )),
+                None,
+            )
+            .unwrap(),
+        ) as Arc<dyn PhysicalExpr>;
 
         assert_eq!(result.to_string(), expected.to_string());
     }

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

@@ -444,8 +444,6 @@ mod tests {
         )
         .expect_err("expected mismatched schema metadata error");
 
-        assert!(err
-            .to_string()
-            .contains("mismatched schema metadata"));
+        assert!(err.to_string().contains("mismatched schema metadata"));
     }
 }

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

@@ -209,12 +209,9 @@ mod tests {
         let target_field = Arc::new(Field::new("a", DataType::Int64, true));
 
         let column_expr = col("a", &schema).unwrap();
-        let cast_expr = Arc::new(CastColumnExpr::new(
-            column_expr,
-            input_field,
-            target_field,
-            None,
-        ).unwrap()) as Arc<dyn PhysicalExpr>;
+        let cast_expr = Arc::new(
+            CastColumnExpr::new(column_expr, input_field, target_field, None).unwrap(),
+        ) as Arc<dyn PhysicalExpr>;
 
         assert!(check_support(&cast_expr, &schema));
     }

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

@@ -221,12 +221,9 @@ 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,
-        ).unwrap());
+        let cast_expr = Arc::new(
+            CastColumnExpr::new(column_expr, input_field, target_field, None).unwrap(),
+        );
         let literal_expr = lit(10i64);
         let binary_expr =
             Arc::new(BinaryExpr::new(cast_expr, Operator::Gt, literal_expr));

datafusion/proto/proto/datafusion.proto

@@ -991,6 +991,25 @@ message PhysicalCastColumnNode {
   datafusion_common.Field input_field = 2;
   datafusion_common.Field target_field = 3;
   bool safe = 4;
+  FormatOptions format_options = 5;
+}
+
+enum DurationFormat {
+  DURATION_FORMAT_UNSPECIFIED = 0;
+  DURATION_FORMAT_ISO8601 = 1;
+  DURATION_FORMAT_PRETTY = 2;
+}
+
+message FormatOptions {
+  bool safe = 1;
+  string null = 2;
+  optional string date_format = 3;
+  optional string datetime_format = 4;
+  optional string timestamp_format = 5;
+  optional string timestamp_tz_format = 6;
+  optional string time_format = 7;
+  DurationFormat duration_format = 8;
+  bool types_info = 9;
 }
 
 message PhysicalNegativeNode {