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

Author: DrakeLin

datafusion-executor/src/expression.rs

@@ -407,15 +407,22 @@ fn struct_columns_from_patch(
 /// - Decimal: arrow's cast silently rescales/rounds to the target scale, while kernel requires the
 ///   value's scale to match the target's exactly (and hard-errors otherwise).
 ///
+/// Configured options are rejected rather than silently lowered with default semantics.
+///
 /// # 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 !map_to_struct.options.is_default() {
+        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);
@@ -547,7 +554,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 +1008,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 +1070,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 }`.

ffi/src/expressions/engine_visitor.rs

@@ -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 !map_to_struct.options.is_default() => {
+            visit_unknown(visitor, sibling_list_id, "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);
@@ -771,7 +774,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 +794,10 @@ mod tests {
             sibling_list_id: usize,
             parts: Vec<String>,
         },
+        Unknown {
+            sibling_list_id: usize,
+            name: String,
+        },
     }
 
     #[derive(Default)]
@@ -847,6 +854,18 @@ 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(
@@ -925,7 +944,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 +1011,25 @@ 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: "map_to_struct".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;
@@ -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/test_ffi.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()
 }

kernel/proto/expressions.proto

@@ -150,10 +150,25 @@ 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. For named timezones,
+  // repeated local times use the earlier occurrence. Local times skipped by a forward clock change
+  // use the prior offset, shifting the result forward by the size of the gap.
+  optional string timestamp_timezone = 1;
+}
+
 message MapToStructExpression {
   Expression map_expr = 1;
 }
 
+message ConfiguredMapToStructExpression {
+  Expression map_expr = 1;
+  MapToStructOptions options = 2;
+}
+
 // `nullability_predicate` is optional: when set and it evaluates to false/null, the whole
 // struct is null.
 message StructExpression {
@@ -237,7 +252,10 @@ message Expression {
     IfExpression if_expr = 9;
     OpaqueExpression opaque = 10;
     ParseJsonExpression parse_json = 11;
+    // This tag carries only default MapToStruct semantics.
     MapToStructExpression map_to_struct = 12;
     string unknown = 13;
+    // Configured semantics use a distinct tag so executors without options support reject them.
+    ConfiguredMapToStructExpression configured_map_to_struct = 14;
   }
 }

kernel/src/checkpoint/checkpoint_transform.rs

@@ -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,12 @@ 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 caller-supplied timezone, so reconstruct with
+        // deterministic UTC rather than guessing from the host environment.
+        Expression::map_to_struct(
+            col!(ADD_NAME, PARTITION_VALUES_FIELD),
+            MapToStructOptions::default(),
+        ),
     ]))
 }