feat!: preserve reader timezone in parsed scan partitions

Author: DrakeLin

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,16 @@ 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 an IANA timezone or fixed offset when the reader uses another
+timezone. An explicit offset in a partition value takes precedence. This setting affects typed
+`scan_metadata` output and its internal partition pruning; it does not change the transforms used
+by `Scan::execute`. Incremental scans expose the raw partition-value map.
+
 > [!TIP]
 > When the checkpoint already stores typed partition values, Kernel reads that column directly
-> and skips parsing entirely.
+> and reparses each zoned `TIMESTAMP` field when a reader timezone is requested. Other fields retain
+> their native checkpoint values.
 
 ## Cancelling a scan
 

kernel/src/checkpoint/checkpoint_shape.rs

@@ -3,9 +3,6 @@
 //! When stats are requested, also reports whether the checkpoint has compatible parsed stats.
 //! Driven through a [`PlanExecutor`].
 
-// No in-crate caller yet; following PRs will use this.
-#![allow(dead_code)]
-
 use url::Url;
 
 use crate::actions::visitors::SidecarVisitor;
@@ -29,23 +26,27 @@ pub(crate) enum CheckpointType {
     Manifest,
 }
 
-/// A snapshot's resolved checkpoint type and parsed-stats schema.
+/// A snapshot's resolved checkpoint type and compatible parsed schemas.
 #[derive(Clone, Debug, PartialEq)]
 pub(crate) struct CheckpointShape {
     /// What kind of checkpoint this is.
     pub(crate) checkpoint_type: CheckpointType,
     /// The requested stats schema when the checkpoint has a compatible `add.stats_parsed` struct
     /// to read it from; `None` when stats were not requested or no compatible parsed stats exist.
     pub(crate) parsed_stats_schema: Option<SchemaRef>,
+    /// The requested partition schema when the checkpoint has a compatible
+    /// `add.partitionValues_parsed` struct; `None` when partition values were not requested or no
+    /// compatible parsed partition values exist.
+    pub(crate) parsed_partition_schema: Option<SchemaRef>,
 }
 
 impl CheckpointShape {
-    /// Resolve `snapshot`'s checkpoint shape. Determines the checkpoint type and, when
-    /// `stats_schema` is `Some`, whether the checkpoint contains parsed stats compatible with it.
+    /// Resolve `snapshot`'s checkpoint shape and compatible parsed stats and partition schemas.
     pub(crate) fn try_new(
         exec: &dyn PlanExecutor,
         snapshot: &Snapshot,
         stats_schema: Option<&SchemaRef>,
+        partition_schema: Option<&SchemaRef>,
     ) -> DeltaResult<CheckpointShape> {
         let segment = snapshot.log_segment();
 
@@ -56,15 +57,21 @@ impl CheckpointShape {
                 return Ok(CheckpointShape {
                     checkpoint_type: CheckpointType::None,
                     parsed_stats_schema: None,
+                    parsed_partition_schema: None,
                 })
             }
         };
 
         // Classify from a V2 checkpoint's `_last_checkpoint` hint when possible, else inspect the
         // file.
-        if let Some(shape) =
-            Self::from_v2_checkpoint_hint(exec, segment, root_checkpoint, file_type, stats_schema)?
-        {
+        if let Some(shape) = Self::from_v2_checkpoint_hint(
+            exec,
+            segment,
+            root_checkpoint,
+            file_type,
+            stats_schema,
+            partition_schema,
+        )? {
             return Ok(shape);
         }
 
@@ -77,22 +84,34 @@ impl CheckpointShape {
                 };
                 // No `sidecar` column means the file actions are inline, so this is a leaf.
                 if !cp_schema.contains(SIDECAR_NAME) {
-                    return Ok(Self::try_new_leaf(Some(cp_schema), stats_schema));
+                    return Ok(Self::try_new_leaf(
+                        Some(cp_schema),
+                        stats_schema,
+                        partition_schema,
+                    ));
                 }
                 // The `sidecar` column may still be all-null (not a manifest), so scan it to
                 // confirm whether a sidecar is actually present.
                 match collect_single_sidecar(exec, root_checkpoint, file_type, &segment.log_root)? {
-                    Some(sidecar) => Self::try_new_manifest(exec, sidecar, stats_schema),
-                    None => Ok(Self::try_new_leaf(Some(cp_schema), stats_schema)),
+                    Some(sidecar) => {
+                        Self::try_new_manifest(exec, sidecar, stats_schema, partition_schema)
+                    }
+                    None => Ok(Self::try_new_leaf(
+                        Some(cp_schema),
+                        stats_schema,
+                        partition_schema,
+                    )),
                 }
             }
             // A JSON checkpoint has no footer schema to inspect, so try to collect a sidecar to
             // decide if it is a manifest or a leaf. A JSON leaf has no readable schema,
             // hence no parsed stats.
             FileType::Json => {
                 match collect_single_sidecar(exec, root_checkpoint, file_type, &segment.log_root)? {
-                    Some(sidecar) => Self::try_new_manifest(exec, sidecar, stats_schema),
-                    None => Ok(Self::try_new_leaf(None, stats_schema)),
+                    Some(sidecar) => {
+                        Self::try_new_manifest(exec, sidecar, stats_schema, partition_schema)
+                    }
+                    None => Ok(Self::try_new_leaf(None, stats_schema, partition_schema)),
                 }
             }
         }
@@ -109,26 +128,32 @@ impl CheckpointShape {
         root_checkpoint: &FileMeta,
         file_type: FileType,
         stats_schema: Option<&SchemaRef>,
+        partition_schema: Option<&SchemaRef>,
     ) -> DeltaResult<Option<CheckpointShape>> {
         match segment.checkpoint_hint_sidecars().map(Vec::as_slice) {
             Some([sidecar, ..]) => {
                 let sidecar_meta = sidecar.to_filemeta(&segment.log_root)?;
-                let result = Self::try_new_manifest(exec, sidecar_meta, stats_schema)?;
+                let result =
+                    Self::try_new_manifest(exec, sidecar_meta, stats_schema, partition_schema)?;
                 Ok(Some(result))
             }
             Some([]) => {
                 // A parquet leaf's stats live in its own schema; read it only when stats are
                 // requested. A JSON leaf has no readable schema.
                 let leaf_schema = match file_type {
-                    FileType::Parquet if stats_schema.is_some() => {
+                    FileType::Parquet if stats_schema.is_some() || partition_schema.is_some() => {
                         Some(match segment.checkpoint_hint_schema() {
                             Some(schema) => schema,
                             None => exec.read_parquet_footer(root_checkpoint.clone())?.schema,
                         })
                     }
                     _ => None,
                 };
-                Ok(Some(Self::try_new_leaf(leaf_schema, stats_schema)))
+                Ok(Some(Self::try_new_leaf(
+                    leaf_schema,
+                    stats_schema,
+                    partition_schema,
+                )))
             }
             None => Ok(None),
         }
@@ -141,18 +166,30 @@ impl CheckpointShape {
         exec: &dyn PlanExecutor,
         sidecar: FileMeta,
         stats_schema: Option<&SchemaRef>,
+        partition_schema: Option<&SchemaRef>,
     ) -> DeltaResult<CheckpointShape> {
-        let parsed_stats_schema = match stats_schema {
-            Some(stats_schema) => {
-                let footer_schema = exec.read_parquet_footer(sidecar)?.schema;
-                LogSegment::schema_has_compatible_stats_parsed(footer_schema.as_ref(), stats_schema)
-                    .then(|| stats_schema.clone())
-            }
-            None => None,
+        let footer_schema = if stats_schema.is_some() || partition_schema.is_some() {
+            Some(exec.read_parquet_footer(sidecar)?.schema)
+        } else {
+            None
         };
+        let parsed_stats_schema = stats_schema.filter(|stats_schema| {
+            footer_schema.as_ref().is_some_and(|footer_schema| {
+                LogSegment::schema_has_compatible_stats_parsed(footer_schema, stats_schema)
+            })
+        });
+        let parsed_partition_schema = partition_schema.filter(|partition_schema| {
+            footer_schema.as_ref().is_some_and(|footer_schema| {
+                LogSegment::schema_has_compatible_partition_values_parsed(
+                    footer_schema,
+                    partition_schema,
+                )
+            })
+        });
         Ok(CheckpointShape {
             checkpoint_type: CheckpointType::Manifest,
-            parsed_stats_schema,
+            parsed_stats_schema: parsed_stats_schema.cloned(),
+            parsed_partition_schema: parsed_partition_schema.cloned(),
         })
     }
 
@@ -161,15 +198,25 @@ impl CheckpointShape {
     fn try_new_leaf(
         leaf_schema: Option<SchemaRef>,
         stats_schema: Option<&SchemaRef>,
+        partition_schema: Option<&SchemaRef>,
     ) -> CheckpointShape {
         let parsed_stats_schema = stats_schema.filter(|stats_schema| {
             leaf_schema.as_ref().is_some_and(|leaf_schema| {
                 LogSegment::schema_has_compatible_stats_parsed(leaf_schema.as_ref(), stats_schema)
             })
         });
+        let parsed_partition_schema = partition_schema.filter(|partition_schema| {
+            leaf_schema.as_ref().is_some_and(|leaf_schema| {
+                LogSegment::schema_has_compatible_partition_values_parsed(
+                    leaf_schema,
+                    partition_schema,
+                )
+            })
+        });
         CheckpointShape {
             checkpoint_type: CheckpointType::Leaf,
             parsed_stats_schema: parsed_stats_schema.cloned(),
+            parsed_partition_schema: parsed_partition_schema.cloned(),
         }
     }
 }
@@ -301,8 +348,8 @@ mod tests {
         let exec = SyncPlanExecutor::default();
         let stats_schema = expect_parsed.map(|_| probe_stats_schema());
 
-        let shape =
-            CheckpointShape::try_new(&exec, snapshot.as_ref(), stats_schema.as_ref()).unwrap();
+        let shape = CheckpointShape::try_new(&exec, snapshot.as_ref(), stats_schema.as_ref(), None)
+            .unwrap();
 
         assert_eq!(
             shape.checkpoint_type, expected_checkpoint,
@@ -353,8 +400,9 @@ mod tests {
             load_test_table("v2-checkpoints-parquet-with-sidecars").unwrap();
         let exec = CountingExecutor::new();
 
-        let shape = CheckpointShape::try_new(&exec, snapshot.as_ref(), Some(&probe_stats_schema()))
-            .unwrap();
+        let shape =
+            CheckpointShape::try_new(&exec, snapshot.as_ref(), Some(&probe_stats_schema()), None)
+                .unwrap();
 
         assert_eq!(shape.checkpoint_type, CheckpointType::Manifest);
         assert_eq!(
@@ -391,8 +439,9 @@ mod tests {
             .unwrap();
 
         let exec = CountingExecutor::new();
-        let shape = CheckpointShape::try_new(&exec, snapshot.as_ref(), Some(&probe_stats_schema()))
-            .unwrap();
+        let shape =
+            CheckpointShape::try_new(&exec, snapshot.as_ref(), Some(&probe_stats_schema()), None)
+                .unwrap();
 
         assert_eq!(shape.checkpoint_type, CheckpointType::Manifest);
         assert!(
@@ -468,6 +517,7 @@ mod tests {
             root,
             file_type,
             stats_schema.as_ref(),
+            None,
         )
         .unwrap()
         .expect("an empty-sidecars hint must classify without falling through");

kernel/src/engine/arrow_expression/evaluate_expression.rs

@@ -3267,6 +3267,10 @@ mod tests {
             None => Expr::cast(col!("s"), target.clone()),
         };
         let result = evaluate_expression(&expr, &batch, Some(&target))?;
+        assert_eq!(
+            result.data_type(),
+            &ArrowDataType::try_from_kernel(&target)?
+        );
         let timestamps = result
             .as_any()
             .downcast_ref::<TimestampMicrosecondArray>()