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

Author: DrakeLin

datafusion-executor/src/expression.rs

@@ -5,10 +5,10 @@ use std::ffi::c_void;
 use delta_kernel::expressions::{
     ArrayData, BinaryExpression, BinaryExpressionOp, BinaryPredicate, BinaryPredicateOp,
     ColumnName, Expression, ExpressionRef, ExpressionStructPatch, JunctionPredicate,
-    JunctionPredicateOp, MapData, MapToStructExpression, OpaqueExpression, OpaqueExpressionOpRef,
-    OpaquePredicate, OpaquePredicateOpRef, ParseJsonExpression, Predicate, Scalar, StructData,
-    UnaryExpression, UnaryExpressionOp, UnaryPredicate, UnaryPredicateOp, VariadicExpression,
-    VariadicExpressionOp,
+    JunctionPredicateOp, MapData, MapToStructExpression, MapToStructOptions, OpaqueExpression,
+    OpaqueExpressionOpRef, OpaquePredicate, OpaquePredicateOpRef, ParseJsonExpression, Predicate,
+    Scalar, StructData, UnaryExpression, UnaryExpressionOp, UnaryPredicate, UnaryPredicateOp,
+    VariadicExpression, VariadicExpressionOp,
 };
 
 use super::kernel_visitor::NullTypeTag;
@@ -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. The sub-expression is in the
+    /// one-item list identified by `child_list_id`. Expressions with configured options are
+    /// reported through `visit_unknown` without visiting the child expression.
     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,12 @@ fn visit_expression_impl(
                 schema_handle
             );
         }
-        Expression::MapToStruct(MapToStructExpression { map_expr }) => {
+        Expression::MapToStruct(map_to_struct)
+            if map_to_struct.options != MapToStructOptions::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 +776,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 +796,10 @@ mod tests {
             sibling_list_id: usize,
             parts: Vec<String>,
         },
+        Unknown {
+            sibling_list_id: usize,
+            name: String,
+        },
     }
 
     #[derive(Default)]
@@ -847,6 +856,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 +946,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 +1013,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/engine_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/expressions/kernel_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/test_ffi.rs

@@ -150,8 +150,16 @@ message ParseJsonExpression {
   delta.kernel.schema.StructType output_schema = 2;
 }
 
+message MapToStructOptions {
+  // Optional IANA timezone or fixed offset for offset-less TIMESTAMP values. Explicit offsets win;
+  // absence means UTC. Ambiguous local times use the earlier instant, nonexistent local times use
+  // the pre-transition offset, and invalid zones must fail evaluation.
+  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/proto/expressions.proto

@@ -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,10 @@ 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)),
+        Expression::map_to_struct(
+            col!(ADD_NAME, PARTITION_VALUES_FIELD),
+            MapToStructOptions::default(),
+        ),
     ]))
 }