feat!: honor reader timezone in partition values

docs/user-guide/src/reading/scan_metadata.md

@@ -249,9 +249,9 @@ If the scan has no predicate, this returns `None`.
 ## Typed partition values
 
 Kernel reads each file's partition values from the Delta log and exposes them on its
-`ScanFile` as a raw string map. Kernel's `transform_to_logical` could materialize them as
-typed columns. If your connector assembles output rows itself instead of using that transform,
-it parses the string map per file.
+`ScanFile` as a raw string map. `transform_to_logical` materializes them as typed columns. A
+connector that assembles output rows itself instead of using that transform would otherwise need
+to parse the string map per file.
 
 To have Kernel hand you the typed values directly, opt in with `with_partition_values`:
 
@@ -270,7 +270,10 @@ To have Kernel hand you the typed values directly, opt in with `with_partition_v
 # let snapshot = Snapshot::builder_for(url).build(&engine)?;
 let scan = snapshot
     .scan_builder()
-    .with_partition_values(PartitionValuesOptions::with_struct())
+    .with_partition_values(
+        PartitionValuesOptions::with_struct()
+            .with_timestamp_timezone("America/Los_Angeles"),
+    )
     .build()?;
 # Ok(())
 # }
@@ -281,9 +284,21 @@ nullable field per partition column (by physical name). You read it as a typed c
 of parsing the string map per file. The raw string map is still present, so this option only
 adds the typed column.
 
+By default, offset-less `TIMESTAMP` partition strings are interpreted in UTC. Use
+`with_timestamp_timezone` with a recognized IANA timezone or a normalized `+HH:MM`, `-HH:MM`,
+`+HH:MM:SS`, or `-HH:MM:SS` fixed offset when the reader uses another timezone. An explicit offset
+or embedded time zone in a partition value takes precedence. This setting affects typed
+`scan_metadata` output, partition predicate evaluation after log replay, and the partition-column
+row transforms used by `Scan::execute`. Checkpoint footer pruning continues to use the
+checkpoint's native parsed partition values. Incremental scans expose the raw partition-value map.
+`TIMESTAMP_NTZ` remains timezone-independent. For daylight-saving transitions, ambiguous local
+times use the earlier instant, and nonexistent local times use the offset from before the
+transition.
+
 > [!TIP]
-> When the checkpoint already stores typed partition values, Kernel reads that column directly
-> and skips parsing entirely.
+> Kernel reparses surviving commit and checkpoint rows from the raw map for typed output and final
+> predicate evaluation. Checkpoint footer pruning happens first and continues to use the
+> checkpoint's native parsed partition values.
 
 ## Cancelling a scan
 

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),
+                }
+            ]
+        );
+    }
 }