feat: honor reader timezone in partition values

Author: DrakeLin

datafusion-executor/src/expression.rs

@@ -472,6 +472,7 @@ fn map_to_struct_primitive(field: &StructField) -> DeltaResult<&PrimitiveType> {
 #[derive(Debug, PartialEq, Eq)]
 struct KernelMapToStructUdf {
     output_schema: KernelSchemaRef,
+    output_type: KernelDataType,
     options: MapToStructOptions,
     return_type: ArrowDataType,
     signature: Signature,
@@ -493,9 +494,11 @@ impl KernelMapToStructUdf {
             .as_ref()
             .try_into_arrow()
             .map_err(Error::generic_err)?;
+        let output_type = KernelDataType::from(output_schema.as_ref().clone());
         Ok(Self {
             return_type: ArrowDataType::Struct(arrow_schema.fields().clone()),
             output_schema,
+            output_type,
             options,
             signature: Signature::any(1, Volatility::Immutable),
         })
@@ -523,9 +526,8 @@ impl ScalarUDFImpl for KernelMapToStructUdf {
             KernelExpression::column(["map"]),
             self.options.clone(),
         );
-        let output_type = KernelDataType::from(self.output_schema.as_ref().clone());
         let result =
-            kernel_expression::evaluate_expression(&expression, &batch, Some(&output_type))
+            kernel_expression::evaluate_expression(&expression, &batch, Some(&self.output_type))
                 .map_err(|e| DataFusionError::External(Box::new(e)))?;
         Ok(ColumnarValue::Array(result))
     }

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,20 @@ 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`,
+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