feat!: support map-to-struct options in engine adapters

ffi/examples/visit-expression/engine_to_kernel_expression.h

@@ -253,7 +253,17 @@ uintptr_t convert_engine_to_kernel_expression_item(
       assert(m2s->child_expr.len == 1);
       uintptr_t child = convert_engine_to_kernel_expression_item(
           state, m2s->child_expr.list[0]);
-      return visit_expression_map_to_struct(state, child);
+      struct OptionalValueKernelStringSlice timestamp_timezone = {
+        .tag = NoneKernelStringSlice
+      };
+      if (m2s->timestamp_timezone != NULL) {
+        timestamp_timezone.tag = SomeKernelStringSlice;
+        timestamp_timezone.some = (struct KernelStringSlice) {
+          .ptr = m2s->timestamp_timezone,
+          .len = strlen(m2s->timestamp_timezone)
+        };
+      }
+      return visit_expression_map_to_struct(state, child, timestamp_timezone);
     }
     case StructPatch:
     case FieldPatch:

ffi/examples/visit-expression/expression.h

@@ -131,6 +131,7 @@ struct Column {
 };
 struct MapToStructExpr {
   ExpressionItemList child_expr;
+  char* timestamp_timezone;
 };
 struct BinaryData {
   uint8_t* buf;
@@ -428,9 +429,13 @@ void visit_unknown(void *data, uintptr_t sibling_list_id, struct KernelStringSli
 
 void visit_map_to_struct_expr(void* data,
                               uintptr_t sibling_list_id,
-                              uintptr_t child_list_id) {
+                              uintptr_t child_list_id,
+                              struct OptionalValueKernelStringSlice timestamp_timezone) {
   struct MapToStructExpr* m2s = malloc(sizeof(struct MapToStructExpr));
   m2s->child_expr = get_expr_list(data, child_list_id);
+  m2s->timestamp_timezone = timestamp_timezone.tag == SomeKernelStringSlice
+      ? allocate_string(timestamp_timezone.some)
+      : NULL;
   put_expr_item(data, sibling_list_id, m2s, MapToStruct);
 }
 
@@ -729,6 +734,7 @@ void free_expression_item(ExpressionItem ref) {
     case MapToStruct: {
       struct MapToStructExpr* m2s = ref.ref;
       free_expression_list(m2s->child_expr);
+      free(m2s->timestamp_timezone);
       free(m2s);
       break;
     }

ffi/examples/visit-expression/expression_print.h

@@ -295,7 +295,11 @@ void print_tree_helper(ExpressionItem ref, int depth) {
     }
     case MapToStruct: {
       struct MapToStructExpr* m2s = ref.ref;
-      printf("MapToStruct\n");
+      if (m2s->timestamp_timezone == NULL) {
+        printf("MapToStruct\n");
+      } else {
+        printf("MapToStruct(timestamp_timezone=%s)\n", m2s->timestamp_timezone);
+      }
       print_expression_item_list(m2s->child_expr, depth + 1);
       break;
     }

ffi/src/expressions/engine_visitor.rs

@@ -16,7 +16,7 @@ use crate::expressions::{
     SharedExpression, SharedOpaqueExpressionOp, SharedOpaquePredicateOp, SharedPredicate,
 };
 use crate::handle::Handle;
-use crate::{kernel_string_slice, KernelStringSlice, SharedSchema};
+use crate::{kernel_string_slice, KernelStringSlice, OptionalValue, SharedSchema};
 
 type VisitLiteralFn<T> = extern "C" fn(data: *mut c_void, sibling_list_id: usize, value: T);
 type VisitUnaryFn = extern "C" fn(data: *mut c_void, sibling_list_id: usize, child_list_id: usize);
@@ -31,6 +31,12 @@ type VisitParseJsonFn = extern "C" fn(
     child_list_id: usize,
     output_schema: Handle<SharedSchema>,
 );
+type VisitMapToStructFn = extern "C" fn(
+    data: *mut c_void,
+    sibling_list_id: usize,
+    child_list_id: usize,
+    timestamp_timezone: OptionalValue<KernelStringSlice>,
+);
 type VisitColumnFn = extern "C" fn(
     data: *mut c_void,
     sibling_list_id: usize,
@@ -174,10 +180,10 @@ pub struct EngineExpressionVisitor {
     /// `child_list_id`. The `output_schema` handle specifies the schema to parse the JSON
     /// into.
     pub visit_parse_json: VisitParseJsonFn,
-    /// Visits the `MapToStruct` expression belonging to the list identified by `sibling_list_id`.
-    /// The sub-expression (map column) will be in a _one_ item list identified by `child_list_id`.
-    /// The output struct schema is determined by the evaluator's result type.
-    pub visit_map_to_struct: VisitUnaryFn,
+    /// Visits a `MapToStruct` expression. The sub-expression is in the one-item list identified by
+    /// `child_list_id`. `timestamp_timezone` carries the configured reader timezone, or is `None`
+    /// when the expression uses the default UTC interpretation.
+    pub visit_map_to_struct: VisitMapToStructFn,
     /// Visits the `LessThan` binary operator belonging to the list identified by
     /// `sibling_list_id`. The operands will be in a _two_ item list identified by
     /// `child_list_id`
@@ -696,10 +702,22 @@ fn visit_expression_impl(
                 schema_handle
             );
         }
-        Expression::MapToStruct(MapToStructExpression { map_expr }) => {
+        Expression::MapToStruct(MapToStructExpression { map_expr, options }) => {
             let child_list_id = call!(visitor, make_field_list, 1);
             visit_expression_impl(visitor, map_expr, child_list_id);
-            call!(visitor, visit_map_to_struct, sibling_list_id, child_list_id);
+            let timestamp_timezone = match options.timestamp_timezone() {
+                Some(timestamp_timezone) => {
+                    OptionalValue::Some(kernel_string_slice!(timestamp_timezone))
+                }
+                None => OptionalValue::None,
+            };
+            call!(
+                visitor,
+                visit_map_to_struct,
+                sibling_list_id,
+                child_list_id,
+                timestamp_timezone
+            );
         }
         // TODO(#2975): Add a dedicated visitor callback for cast expressions.
         Expression::Cast(cast) => visit_unknown(
@@ -771,7 +789,7 @@ fn visit_predicate_internal(predicate: &Predicate, visitor: &mut EngineExpressio
 
 #[cfg(test)]
 mod tests {
-    use delta_kernel::expressions::{lit, Expression, Scalar};
+    use delta_kernel::expressions::{lit, Expression, MapToStructOptions, Scalar};
     use rstest::rstest;
 
     use super::*;
@@ -791,6 +809,15 @@ mod tests {
             sibling_list_id: usize,
             parts: Vec<String>,
         },
+        Unknown {
+            sibling_list_id: usize,
+            name: String,
+        },
+        MapToStruct {
+            sibling_list_id: usize,
+            child_list_id: usize,
+            timestamp_timezone: Option<String>,
+        },
     }
 
     #[derive(Default)]
@@ -806,6 +833,22 @@ mod tests {
         list_id
     }
 
+    extern "C" fn visit_map_to_struct(
+        data: *mut c_void,
+        sibling_list_id: usize,
+        child_list_id: usize,
+        timestamp_timezone: OptionalValue<KernelStringSlice>,
+    ) {
+        let builder = unsafe { &mut *(data as *mut TestExpressionBuilder) };
+        let timestamp_timezone = Option::from(timestamp_timezone)
+            .map(|timezone| unsafe { String::try_from_slice(&timezone).unwrap() });
+        builder.events.push(LiteralEvent::MapToStruct {
+            sibling_list_id,
+            child_list_id,
+            timestamp_timezone,
+        });
+    }
+
     extern "C" fn visit_literal_interval_year_month(
         data: *mut c_void,
         sibling_list_id: usize,
@@ -847,6 +890,19 @@ mod tests {
         });
     }
 
+    extern "C" fn visit_unknown_name(
+        data: *mut c_void,
+        sibling_list_id: usize,
+        name: KernelStringSlice,
+    ) {
+        let builder = unsafe { &mut *(data as *mut TestExpressionBuilder) };
+        let name = unsafe { String::try_from_slice(&name) }.unwrap();
+        builder.events.push(LiteralEvent::Unknown {
+            sibling_list_id,
+            name,
+        });
+    }
+
     macro_rules! ignore_fn {
         ($fn_name:ident $(, $arg_type:ty)*) => {
             extern "C" fn $fn_name(
@@ -907,7 +963,7 @@ mod tests {
             visit_is_null: ignore_child_list,
             visit_to_json: ignore_child_list,
             visit_parse_json: ignore_parse_json,
-            visit_map_to_struct: ignore_child_list,
+            visit_map_to_struct,
             visit_lt: ignore_child_list,
             visit_gt: ignore_child_list,
             visit_eq: ignore_child_list,
@@ -925,7 +981,7 @@ mod tests {
             visit_field_patch: ignore_field_patch,
             visit_opaque_expr: ignore_opaque_expr,
             visit_opaque_pred: ignore_opaque_pred,
-            visit_unknown: ignore_string_slice,
+            visit_unknown: visit_unknown_name,
         }
     }
 
@@ -992,4 +1048,35 @@ mod tests {
         assert_eq!(top_level_id, 0);
         assert_eq!(builder.events, vec![expected]);
     }
+
+    #[rstest]
+    #[case::default(None)]
+    #[case::configured(Some("America/Los_Angeles"))]
+    fn map_to_struct_visits_options(#[case] timestamp_timezone: Option<&str>) {
+        let options = timestamp_timezone.map_or_else(MapToStructOptions::default, |timezone| {
+            MapToStructOptions::default().with_timestamp_timezone(timezone)
+        });
+        let expression =
+            Expression::map_to_struct(Expression::column(["partitionValues"]), options);
+        let mut builder = TestExpressionBuilder::default();
+        let mut visitor = test_visitor(&mut builder);
+
+        let top_level_id = visit_expression_internal(&expression, &mut visitor);
+
+        assert_eq!(top_level_id, 0);
+        assert_eq!(
+            builder.events,
+            vec![
+                LiteralEvent::Column {
+                    sibling_list_id: 1,
+                    parts: vec!["partitionValues".to_string()],
+                },
+                LiteralEvent::MapToStruct {
+                    sibling_list_id: 0,
+                    child_list_id: 1,
+                    timestamp_timezone: timestamp_timezone.map(str::to_string),
+                }
+            ]
+        );
+    }
 }

ffi/src/expressions/kernel_visitor.rs

@@ -6,7 +6,7 @@ use std::sync::Arc;
 use delta_kernel::engine::arrow_expression::opaque::ArrowOpaquePredicate;
 use delta_kernel::expressions::{
     lit, null_lit, BinaryExpressionOp, BinaryPredicateOp, ColumnName, Expression,
-    JunctionPredicateOp, Predicate, Scalar, UnaryPredicateOp,
+    JunctionPredicateOp, MapToStructOptions, Predicate, Scalar, UnaryPredicateOp,
 };
 use delta_kernel::schema::{DataType, PrimitiveType};
 use delta_kernel::DeltaResult;
@@ -20,7 +20,7 @@ use crate::handle::Handle;
 use crate::scan::{EngineExpression, EnginePredicate};
 use crate::{
     AllocateErrorFn, EngineIterator, ExternResult, IntoExternResult, KernelStringSlice,
-    ReferenceSet, TryFromStringSlice,
+    OptionalValue, ReferenceSet, TryFromStringSlice,
 };
 
 pub(crate) enum ExpressionOrPredicate {
@@ -685,14 +685,34 @@ pub extern "C" fn visit_expression_struct(
     wrap_expression(state, Expression::struct_from(exprs))
 }
 
-/// Visit a MapToStruct expression. The `child_expr` is the map expression.
+/// Builds a `MapToStruct` expression from its map child and optional reader timezone.
+///
+/// `timestamp_timezone` is `None` for the default UTC interpretation. A provided string is copied
+/// into the expression before this function returns.
+///
+/// Returns zero when `child_expr` is invalid or the timezone is not valid UTF-8.
+///
+/// # Safety
+///
+/// A provided `timestamp_timezone` slice must have a non-null pointer to a readable buffer of its
+/// declared number of initialized bytes and remain valid for this call.
 #[no_mangle]
-pub extern "C" fn visit_expression_map_to_struct(
+pub unsafe extern "C" fn visit_expression_map_to_struct(
     state: &mut KernelExpressionVisitorState,
     child_expr: usize,
+    timestamp_timezone: OptionalValue<KernelStringSlice>,
 ) -> usize {
+    let options = match Option::from(timestamp_timezone) {
+        Some(timestamp_timezone) => match unsafe { String::try_from_slice(&timestamp_timezone) } {
+            Ok(timestamp_timezone) => {
+                MapToStructOptions::default().with_timestamp_timezone(timestamp_timezone)
+            }
+            Err(_) => return 0,
+        },
+        None => MapToStructOptions::default(),
+    };
     unwrap_kernel_expression(state, child_expr).map_or(0, |expr| {
-        wrap_expression(state, Expression::map_to_struct(expr))
+        wrap_expression(state, Expression::map_to_struct(expr, options))
     })
 }
 
@@ -868,6 +888,49 @@ mod tests {
 
     use super::*;
 
+    #[rstest]
+    #[case::default(None)]
+    #[case::configured(Some("America/Los_Angeles"))]
+    fn map_to_struct_preserves_options(#[case] timestamp_timezone: Option<&str>) {
+        let mut state = KernelExpressionVisitorState::default();
+        let child = wrap_expression(&mut state, col!("partitionValues"));
+        let ffi_timezone = match timestamp_timezone {
+            Some(timestamp_timezone) => {
+                OptionalValue::Some(crate::kernel_string_slice!(timestamp_timezone))
+            }
+            None => OptionalValue::None,
+        };
+
+        let expression_id =
+            unsafe { visit_expression_map_to_struct(&mut state, child, ffi_timezone) };
+        let expression = unwrap_kernel_expression(&mut state, expression_id).unwrap();
+        let options = timestamp_timezone.map_or_else(MapToStructOptions::default, |timezone| {
+            MapToStructOptions::default().with_timestamp_timezone(timezone)
+        });
+
+        assert_eq!(
+            expression,
+            Expression::map_to_struct(col!("partitionValues"), options)
+        );
+    }
+
+    #[test]
+    fn map_to_struct_rejects_invalid_timezone_utf8() {
+        let mut state = KernelExpressionVisitorState::default();
+        let child = wrap_expression(&mut state, col!("partitionValues"));
+        let invalid_utf8 = [0xff_u8];
+        let timezone = KernelStringSlice {
+            ptr: invalid_utf8.as_ptr().cast(),
+            len: invalid_utf8.len(),
+        };
+
+        let expression_id = unsafe {
+            visit_expression_map_to_struct(&mut state, child, OptionalValue::Some(timezone))
+        };
+
+        assert_eq!(expression_id, 0);
+    }
+
     // ============================================================================
     // NullTypeTag::from_data_type
     // ============================================================================