feat: honor reader timezone in partition values

Author: DrakeLin

Cargo.lock

@@ -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,19 @@ 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`,
+`+HH:MM:SS`, or `-HH:MM:SS` fixed offset when the reader uses another timezone. An explicit offset
+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. 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 parses typed partition values from the raw map for both commits and checkpoints. This
+> keeps reader-timezone behavior independent of the timezone used to write a checkpoint.
 
 ## Cancelling a scan
 

datafusion-executor/Cargo.lock

@@ -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 = "0.10.4"
 crc = "3.2.2"
 indexmap = "2.10.0"
 itertools = "0.14"

docs/user-guide/src/reading/scan_metadata.md

@@ -1,9 +1,9 @@
 //! Resolves a checkpoint shape. This falls into the following cases: no checkpoint,
 //! leaf (file actions inline, including multi-part), or manifest (which references sidecar files).
-//! When stats are requested, also reports whether the checkpoint has compatible parsed stats.
+//! When typed metadata is requested, also reports whether the checkpoint has compatible parsed
+//! stats and partition values.
 //! Driven through a [`PlanExecutor`].
-// No in-crate caller yet; following PRs will use this.
-#![allow(dead_code)]
+use std::sync::Arc;
 
 use url::Url;
 
@@ -28,23 +28,27 @@ pub(crate) enum CheckpointType {
     Manifest,
 }
 
-/// A snapshot's resolved checkpoint type and parsed-stats schema.
+/// A snapshot's resolved checkpoint type and compatible typed metadata 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 partitions were not requested or no
+    /// compatible parsed values exist.
+    pub(crate) parsed_partition_values_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 typed metadata fields.
     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();
 
@@ -55,15 +59,21 @@ impl CheckpointShape {
                 return Ok(CheckpointShape {
                     checkpoint_type: CheckpointType::None,
                     parsed_stats_schema: None,
+                    parsed_partition_values_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);
         }
 
@@ -76,7 +86,11 @@ 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.
@@ -85,9 +99,14 @@ impl CheckpointShape {
                         exec,
                         sidecar,
                         stats_schema,
+                        partition_schema,
                         segment.checkpoint_hint_sidecar_file_schema(),
                     ),
-                    None => Ok(Self::try_new_leaf(Some(cp_schema), stats_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
@@ -99,9 +118,10 @@ impl CheckpointShape {
                         exec,
                         sidecar,
                         stats_schema,
+                        partition_schema,
                         segment.checkpoint_hint_sidecar_file_schema(),
                     ),
-                    None => Ok(Self::try_new_leaf(None, stats_schema)),
+                    None => Ok(Self::try_new_leaf(None, stats_schema, partition_schema)),
                 }
             }
         }
@@ -118,6 +138,7 @@ 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, ..]) => {
@@ -126,62 +147,67 @@ impl CheckpointShape {
                     exec,
                     sidecar_meta,
                     stats_schema,
+                    partition_schema,
                     segment.checkpoint_hint_sidecar_file_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.
+                // A parquet leaf's typed metadata lives in its own schema; read it only when typed
+                // metadata was 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),
         }
     }
 
-    /// Build the shape for a manifest checkpoint. Its file actions and their stats live in the
-    /// sidecars, so we need a sidecar's schema to answer the stats-compatibility question -- but
-    /// only when stats were requested. All sidecars of a checkpoint share one schema, so any one is
-    /// sufficient.
-    ///
-    /// If the `_last_checkpoint` hint carries a `sidecarFileSchema`, use it directly,
-    /// Otherwise read the sidecar's footer to get the schema.
+    /// Build the shape for a manifest checkpoint. Its file actions and typed metadata live in the
+    /// sidecars. Use the `_last_checkpoint` sidecar schema when present; otherwise read one
+    /// sidecar footer when typed metadata was requested. All sidecars share one schema.
     fn try_new_manifest(
         exec: &dyn PlanExecutor,
         sidecar: FileMeta,
         stats_schema: Option<&SchemaRef>,
+        partition_schema: Option<&SchemaRef>,
         hint_sidecar_schema: Option<StructType>,
     ) -> DeltaResult<CheckpointShape> {
-        let parsed_stats_schema = match stats_schema {
-            Some(stats_schema) => {
-                let compatible = match hint_sidecar_schema {
-                    Some(schema) => {
-                        LogSegment::schema_has_compatible_stats_parsed(&schema, stats_schema)
-                    }
-                    None => {
-                        let footer_schema = exec.read_parquet_footer(sidecar)?.schema;
-                        LogSegment::schema_has_compatible_stats_parsed(
-                            footer_schema.as_ref(),
-                            stats_schema,
-                        )
-                    }
-                };
-                compatible.then(|| stats_schema.clone())
+        let sidecar_schema = match hint_sidecar_schema {
+            Some(schema) => Some(Arc::new(schema)),
+            None if stats_schema.is_some() || partition_schema.is_some() => {
+                Some(exec.read_parquet_footer(sidecar)?.schema)
             }
             None => None,
         };
+        let parsed_stats_schema = stats_schema.filter(|stats_schema| {
+            sidecar_schema.as_ref().is_some_and(|sidecar_schema| {
+                LogSegment::schema_has_compatible_stats_parsed(sidecar_schema, stats_schema)
+            })
+        });
+        let parsed_partition_values_schema = partition_schema.filter(|partition_schema| {
+            sidecar_schema.as_ref().is_some_and(|sidecar_schema| {
+                LogSegment::schema_has_compatible_partition_values_parsed(
+                    sidecar_schema,
+                    partition_schema,
+                )
+            })
+        });
         Ok(CheckpointShape {
             checkpoint_type: CheckpointType::Manifest,
-            parsed_stats_schema,
+            parsed_stats_schema: parsed_stats_schema.cloned(),
+            parsed_partition_values_schema: parsed_partition_values_schema.cloned(),
         })
     }
 
@@ -190,15 +216,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_values_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_values_schema: parsed_partition_values_schema.cloned(),
         }
     }
 }
@@ -334,8 +370,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,
@@ -381,8 +417,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!(
@@ -419,8 +456,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!(
@@ -492,6 +530,7 @@ mod tests {
             root,
             file_type,
             stats_schema.as_ref(),
+            None,
         )
         .unwrap()
         .expect("an empty-sidecars hint must classify without falling through");
@@ -602,6 +641,7 @@ mod tests {
             root,
             file_type,
             Some(&stats_schema),
+            None,
         )
         .unwrap()
         .expect("a sidecar-listing hint must classify as a manifest");
@@ -636,7 +676,7 @@ mod tests {
 
         // Full resolution reads the sidecar footer to answer compatibility.
         let footer_shape =
-            CheckpointShape::try_new(&exec, snapshot.as_ref(), Some(&stats_schema)).unwrap();
+            CheckpointShape::try_new(&exec, snapshot.as_ref(), Some(&stats_schema), None).unwrap();
         assert_eq!(footer_shape.checkpoint_type, CheckpointType::Manifest);
 
         // Read that same footer schema ourselves and feed it as the hint schema.
@@ -651,6 +691,7 @@ mod tests {
             &exec,
             sidecar,
             Some(&stats_schema),
+            None,
             Some(footer_schema.as_ref().clone()),
         )
         .unwrap();

kernel/Cargo.toml

@@ -442,8 +442,8 @@ mod tests {
     }
 
     /// The checkpoint falls back to `MAP_TO_STRUCT` over `partitionValues` when no native
-    /// `partitionValues_parsed` column is present, so a checkpoint reconstructs the same typed
-    /// struct the scan reads and the two can never disagree on a value.
+    /// `partitionValues_parsed` column is present, so footer skipping still has typed partition
+    /// values for checkpoints that do not provide them natively.
     #[test]
     fn build_partition_values_parsed_expr_falls_back_to_map_to_struct() {
         let expr = build_partition_values_parsed_expr();