feat: Implement to_json for subset of types

docs/source/user-guide/expressions.md

@@ -182,6 +182,14 @@ The following Spark expressions are currently available. Any known compatibility
 | VariancePop   |       |
 | VarianceSamp  |       |
 
+## Complex Types
+
+| Expression        | Notes |
+| ----------------- | ----- |
+| CreateNamedStruct |       |
+| GetElementAt      |       |
+| StructsToJson     |       |
+
 ## Other
 
 | Expression              | Notes                                                                           |
@@ -191,4 +199,3 @@ The following Spark expressions are currently available. Any known compatibility
 | ScalarSubquery          |                                                                                 |
 | Coalesce                |                                                                                 |
 | NormalizeNaNAndZero     |                                                                                 |
-| CreateNamedStruct       |                                                                                 |

native/core/src/execution/datafusion/planner.rs

@@ -95,7 +95,7 @@ use datafusion_comet_proto::{
 };
 use datafusion_comet_spark_expr::{
     Cast, CreateNamedStruct, DateTruncExpr, GetStructField, HourExpr, IfExpr, MinuteExpr, RLike,
-    SecondExpr, TimestampTruncExpr,
+    SecondExpr, TimestampTruncExpr, ToJson,
 };
 use datafusion_common::scalar::ScalarStructBuilder;
 use datafusion_common::{
@@ -655,6 +655,10 @@ impl PhysicalPlanner {
                     self.create_expr(expr.child.as_ref().unwrap(), Arc::clone(&input_schema))?;
                 Ok(Arc::new(GetStructField::new(child, expr.ordinal as usize)))
             }
+            ExprStruct::ToJson(expr) => {
+                let child = self.create_expr(expr.child.as_ref().unwrap(), input_schema)?;
+                Ok(Arc::new(ToJson::new(child, &expr.timezone)))
+            }
             expr => Err(ExecutionError::GeneralError(format!(
                 "Not implemented: {:?}",
                 expr

native/proto/src/proto/expr.proto

@@ -79,6 +79,7 @@ message Expr {
     BloomFilterMightContain bloom_filter_might_contain = 52;
     CreateNamedStruct create_named_struct = 53;
     GetStructField get_struct_field = 54;
+    ToJson to_json = 55;
   }
 }
 
@@ -343,6 +344,15 @@ message StringSpace {
   Expr child = 1;
 }
 
+message ToJson {
+  Expr child = 1;
+  string timezone = 2;
+  string date_format = 3;
+  string timestamp_format = 4;
+  string timestamp_ntz_format = 5;
+  bool ignore_null_fields = 6;
+}
+
 message Hour {
   Expr child = 1;
   string timezone = 2;

native/spark-expr/src/cast.rs

@@ -582,7 +582,7 @@ pub fn spark_cast(
     arg: ColumnarValue,
     data_type: &DataType,
     eval_mode: EvalMode,
-    timezone: String,
+    timezone: &str,
     allow_incompat: bool,
 ) -> DataFusionResult<ColumnarValue> {
     match arg {
@@ -1414,7 +1414,7 @@ impl PhysicalExpr for Cast {
             arg,
             &self.data_type,
             self.eval_mode,
-            self.timezone.clone(),
+            &self.timezone,
             self.allow_incompat,
         )
     }

native/spark-expr/src/lib.rs

@@ -30,6 +30,7 @@ pub mod spark_hash;
 mod structs;
 mod temporal;
 pub mod timezone;
+mod to_json;
 pub mod utils;
 mod xxhash64;
 
@@ -39,6 +40,7 @@ pub use if_expr::IfExpr;
 pub use regexp::RLike;
 pub use structs::{CreateNamedStruct, GetStructField};
 pub use temporal::{DateTruncExpr, HourExpr, MinuteExpr, SecondExpr, TimestampTruncExpr};
+pub use to_json::ToJson;
 
 /// Spark supports three evaluation modes when evaluating expressions, which affect
 /// the behavior when processing input values that are invalid or would result in an