feat!: add reader timezone to map-to-struct

Author: DrakeLin

Cargo.lock

@@ -390,13 +390,11 @@ fn struct_columns_from_patch(
 /// types come from `output_type`, which must be a struct holding only primitive fields (matching
 /// the kernel evaluator, which supports only primitive targets).
 ///
-/// Each field extracts its value with `cast(get_field(map, name), T)`. For a numeric or temporal
-/// type the raw value is first wrapped in `nullif(.., '')`, mapping an empty string to null before
-/// the cast, so an empty string becomes null (kernel's `empty_string_partition_cast`) while an
-/// unparseable value fails the cast (kernel's hard parse error). String and Binary keep the raw
-/// value (empty is a valid empty string / empty bytes). A missing key or null value is already null
-/// via [`get_field`]. The whole struct is nulled where the input map row is null, via `<map> IS NOT
-/// NULL`.
+/// Each field extracts its value with `cast(get_field(map, name), T)`. For every type except String
+/// and Binary, the raw value is first wrapped in `nullif(.., '')`, mapping an empty string to null
+/// before the cast. String and Binary keep the raw value because empty strings and bytes are valid.
+/// A missing key or null value is already null via [`get_field`]. The whole struct is nulled where
+/// the input map row is null, via `<map> IS NOT NULL`.
 ///
 /// KNOWN DIVERGENCES from the kernel parser, confined to malformed or non-spec-compliant values
 /// (spec-compliant writers never emit them):
@@ -408,14 +406,20 @@ fn struct_columns_from_patch(
 ///   value's scale to match the target's exactly (and hard-errors otherwise).
 ///
 /// # Errors
-/// Returns an error when `output_type` is absent, not a struct, or has a non-primitive field, or
-/// from lowering the map expression.
+///
+/// Returns an error when options are configured, `output_type` is absent, not a struct, or has a
+/// non-primitive field, or from lowering the map expression.
 fn map_to_struct_to_df_expr(
     map_to_struct: &MapToStructExpression,
     input_schema: &StructType,
     output_type: Option<&KernelDataType>,
 ) -> DeltaResult<DFExpr> {
     let target = require_struct_output(output_type, "MapToStruct")?;
+    if !can_lower_map_to_struct_options(map_to_struct) {
+        return Err(Error::unsupported(
+            "DataFusion execution of MapToStruct with configured options",
+        ));
+    }
     let map = to_df_expr(&map_to_struct.map_expr, input_schema, None)?;
 
     let mut args = Vec::with_capacity(target.num_fields() * 2);
@@ -445,6 +449,10 @@ fn map_to_struct_to_df_expr(
     Ok(struct_null_when_not(map.is_not_null(), named_struct(args)))
 }
 
+fn can_lower_map_to_struct_options(map_to_struct: &MapToStructExpression) -> bool {
+    map_to_struct.options.is_default()
+}
+
 /// Lowers a `ParseJson` (parse a JSON-string column into a struct) to a call of the
 /// [`ParseJsonUdf`] scalar UDF, which delegates to kernel's own JSON parser. Unlike the
 /// struct-shaped arms, `ParseJson` is self-typed -- it carries its target `output_schema` -- so it
@@ -547,7 +555,7 @@ mod tests {
     use datafusion::physical_expr::execution_props::ExecutionProps;
     use delta_kernel::expressions::{
         col, lit, null_lit, ColumnName as KernelColumnName, Expression as KernelExpr,
-        ExpressionStructPatch, ExpressionStructPatchBuilder,
+        ExpressionStructPatch, ExpressionStructPatchBuilder, MapToStructOptions,
     };
     use delta_kernel::schema::{schema, schema_ref, ArrayType, DataType, MapType, StructType};
     use rstest::rstest;
@@ -1001,7 +1009,7 @@ mod tests {
     /// Lowers a `MapToStruct` over `pv` targeting `output_schema` and renders it as a `Display`
     /// string.
     fn lower_map_to_struct(output_schema: StructType) -> String {
-        let kernel = KernelExpr::map_to_struct(col!("pv"));
+        let kernel = KernelExpr::map_to_struct(col!("pv"), MapToStructOptions::default());
         let target: DataType = output_schema.into();
         to_df_expr(&kernel, &pv_map_schema(), Some(&target))
             .unwrap()
@@ -1063,13 +1071,28 @@ mod tests {
         #[case] output_type: Option<DataType>,
         #[case] expected_message: &str,
     ) {
-        let kernel = KernelExpr::map_to_struct(col!("pv"));
+        let kernel = KernelExpr::map_to_struct(col!("pv"), MapToStructOptions::default());
         let err = to_df_expr(&kernel, &pv_map_schema(), output_type.as_ref())
             .unwrap_err()
             .to_string();
         assert!(err.contains(expected_message), "{err}");
     }
 
+    #[test]
+    fn configured_map_to_struct_is_unsupported() {
+        let target = DataType::from(schema! { nullable "ts": TIMESTAMP });
+        let kernel = KernelExpr::map_to_struct(
+            col!("pv"),
+            MapToStructOptions::default().with_timestamp_timezone("America/Los_Angeles"),
+        );
+
+        let error = to_df_expr(&kernel, &pv_map_schema(), Some(&target))
+            .unwrap_err()
+            .to_string();
+
+        assert!(error.contains("MapToStruct with configured options"));
+    }
+
     // === ParseJson Shared Helpers ===
 
     /// Input schema for JSON tests: `{ j: string }`.

datafusion-executor/Cargo.lock

@@ -174,9 +174,9 @@ 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.
+    /// Visits a `MapToStruct` expression with default options. Expressions with configured options
+    /// are reported through `visit_unknown` without visiting the child expression. The
+    /// sub-expression is in the one-item list identified by `child_list_id`.
     pub visit_map_to_struct: VisitUnaryFn,
     /// 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
@@ -696,7 +696,10 @@ fn visit_expression_impl(
                 schema_handle
             );
         }
-        Expression::MapToStruct(MapToStructExpression { map_expr }) => {
+        Expression::MapToStruct(map_to_struct) if !can_visit_map_to_struct(map_to_struct) => {
+            visit_unknown(visitor, sibling_list_id, "configured_map_to_struct")
+        }
+        Expression::MapToStruct(MapToStructExpression { map_expr, .. }) => {
             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);
@@ -711,6 +714,10 @@ fn visit_expression_impl(
     }
 }
 
+fn can_visit_map_to_struct(map_to_struct: &MapToStructExpression) -> bool {
+    map_to_struct.options.is_default()
+}
+
 fn visit_predicate_impl(
     visitor: &mut EngineExpressionVisitor,
     predicate: &Predicate,
@@ -771,7 +778,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 +798,14 @@ 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,
+        },
     }
 
     #[derive(Default)]
@@ -847,6 +862,30 @@ 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,
+        });
+    }
+
+    extern "C" fn visit_map_to_struct(
+        data: *mut c_void,
+        sibling_list_id: usize,
+        child_list_id: usize,
+    ) {
+        let builder = unsafe { &mut *(data as *mut TestExpressionBuilder) };
+        builder.events.push(LiteralEvent::MapToStruct {
+            sibling_list_id,
+            child_list_id,
+        });
+    }
     macro_rules! ignore_fn {
         ($fn_name:ident $(, $arg_type:ty)*) => {
             extern "C" fn $fn_name(
@@ -907,7 +946,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 +964,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 +1031,52 @@ mod tests {
         assert_eq!(top_level_id, 0);
         assert_eq!(builder.events, vec![expected]);
     }
+
+    #[test]
+    fn timezone_aware_map_to_struct_visits_unknown() {
+        let expression = Expression::map_to_struct(
+            Expression::column(["partitionValues"]),
+            MapToStructOptions::default().with_timestamp_timezone("America/Los_Angeles"),
+        );
+        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::Unknown {
+                sibling_list_id: 0,
+                name: "configured_map_to_struct".to_string(),
+            }]
+        );
+    }
+
+    #[test]
+    fn default_map_to_struct_visits_child_then_map_to_struct() {
+        let expression = Expression::map_to_struct(
+            Expression::column(["partitionValues"]),
+            MapToStructOptions::default(),
+        );
+        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,
+                },
+            ]
+        );
+    }
 }

datafusion-executor/src/expression.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;
@@ -692,7 +692,10 @@ pub extern "C" fn visit_expression_map_to_struct(
     child_expr: usize,
 ) -> usize {
     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, MapToStructOptions::default()),
+        )
     })
 }
 

ffi/src/expressions/engine_visitor.rs

@@ -6,8 +6,9 @@ use std::sync::Arc;
 
 use delta_kernel::expressions::{
     col, column_name, column_pred, lit, null_lit, ArrayData, BinaryExpressionOp, BinaryPredicateOp,
-    Expression as Expr, ExpressionStructPatchBuilder, MapData, OpaqueExpressionOp,
-    OpaquePredicateOp, Predicate as Pred, Scalar, ScalarExpressionEvaluator, StructData,
+    Expression as Expr, ExpressionStructPatchBuilder, MapData, MapToStructOptions,
+    OpaqueExpressionOp, OpaquePredicateOp, Predicate as Pred, Scalar, ScalarExpressionEvaluator,
+    StructData,
 };
 use delta_kernel::kernel_predicates::{
     DirectDataSkippingPredicateEvaluator, DirectPredicateEvaluator,
@@ -158,7 +159,7 @@ pub unsafe extern "C" fn get_testing_kernel_expression() -> Handle<SharedExpress
         Expr::struct_from([lit(5_i32), lit(20_i64)]),
         Expr::opaque(OpaqueTestOp("foo".to_string()), vec![lit(42), lit(1.111)]),
         Expr::unknown("mystery"),
-        Expr::map_to_struct(col!("pv")),
+        Expr::map_to_struct(col!("pv"), MapToStructOptions::default()),
         Expr::coalesce([col!("col"), lit(0_i32)]),
         Expr::array([lit(1_i32), lit(2_i32)]),
     ];
@@ -254,7 +255,7 @@ pub unsafe extern "C" fn get_simple_testing_kernel_expression() -> Handle<Shared
         Expr::binary(BinaryExpressionOp::Multiply, lit(5), lit(6)),
         Expr::binary(BinaryExpressionOp::Divide, lit(100), lit(4)),
         Expr::struct_from([lit(1_i32), lit(2_i64), lit(3.0_f64)]),
-        Expr::map_to_struct(col!("partitionValues")),
+        Expr::map_to_struct(col!("partitionValues"), MapToStructOptions::default()),
     ];
     Arc::new(Expr::struct_from(sub_exprs)).into()
 }

ffi/src/expressions/kernel_visitor.rs

@@ -42,6 +42,7 @@ pre-release-hook = [
 delta_kernel_derive = { path = "../derive-macros", version = "0.28.0" }
 bytes = "1.10"
 chrono = "0.4.41"
+chrono-tz = { version = "0.10.4", optional = true }
 crc = "3.2.2"
 indexmap = "2.10.0"
 itertools = "0.14"
@@ -109,7 +110,7 @@ need-arrow = [] # need-arrow is a marker that the feature needs arrow dep
 arrow-58 = ["dep:arrow_58", "dep:parquet_58", "dep:object_store_13"]
 arrow-59 = ["dep:arrow_59", "dep:parquet_59", "dep:object_store_13"]
 arrow-conversion = ["need-arrow"]
-arrow-expression = ["need-arrow"]
+arrow-expression = ["need-arrow", "dep:chrono-tz"]
 
 # Schema diffing functionality (experimental)
 schema-diff = []

ffi/src/test_ffi.rs

@@ -150,8 +150,17 @@ message ParseJsonExpression {
   delta.kernel.schema.StructType output_schema = 2;
 }
 
+message MapToStructOptions {
+  // Optional connector-supplied IANA timezone or normalized fixed offset (`+HH:MM`, `-HH:MM`,
+  // `+HH:MM:SS`, or `-HH:MM:SS`) for offset-less TIMESTAMP values. Kernel never infers this from
+  // the host or table metadata. A timezone or offset in a value takes precedence; absence means
+  // UTC.
+  optional string timestamp_timezone = 1;
+}
+
 message MapToStructExpression {
   Expression map_expr = 1;
+  MapToStructOptions options = 2;
 }
 
 // `nullability_predicate` is optional: when set and it evaluates to false/null, the whole

kernel/Cargo.toml

@@ -16,7 +16,7 @@
 use std::sync::{Arc, LazyLock};
 
 use crate::actions::{ADD_NAME, STATS_PARSED as STATS_PARSED_FIELD};
-use crate::expressions::{col, Expression, ExpressionRef, UnaryExpressionOp};
+use crate::expressions::{col, Expression, ExpressionRef, MapToStructOptions, UnaryExpressionOp};
 use crate::schema::{DataType, SchemaRef, SchemaStructPatchBuilder, StructField, StructType};
 use crate::struct_patch::ProjectionStructPatchBuilder;
 use crate::table_properties::TableProperties;
@@ -211,7 +211,11 @@ fn build_stats_parsed_expr(stats_schema: &SchemaRef) -> ExpressionRef {
 fn build_partition_values_parsed_expr() -> ExpressionRef {
     Arc::new(Expression::coalesce([
         col!(ADD_NAME, PARTITION_VALUES_PARSED_FIELD),
-        Expression::map_to_struct(col!(ADD_NAME, PARTITION_VALUES_FIELD)),
+        // Checkpoint construction has no reader timezone, so use UTC.
+        Expression::map_to_struct(
+            col!(ADD_NAME, PARTITION_VALUES_FIELD),
+            MapToStructOptions::default(),
+        ),
     ]))
 }