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,17 @@ 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, internal partition pruning, and the partition-column row transforms used
+by `Scan::execute`. 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.27.1" }
 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

@@ -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;

kernel/Cargo.toml

@@ -3,19 +3,16 @@ use std::borrow::Cow;
 use std::collections::HashMap;
 use std::sync::Arc;
 
-use chrono::{FixedOffset, NaiveDateTime, Offset, TimeDelta, TimeZone, Utc};
 use itertools::Itertools;
 use tracing::warn;
 
-use crate::arrow::array::timezone::Tz;
 use crate::arrow::array::types::*;
 use crate::arrow::array::{
     self as arrow_array, make_array, new_null_array, Array, ArrayBuilder, ArrayData, ArrayRef,
     AsArray, BooleanArray, Datum, ListArray, MapArray, MutableArrayData, NullBufferBuilder,
     RecordBatch, StringArray, StructArray,
 };
 use crate::arrow::buffer::{NullBuffer, OffsetBuffer};
-use crate::arrow::compute::kernels::cast_utils::{string_to_datetime, Parser};
 use crate::arrow::compute::kernels::cmp::{distinct, eq, gt, gt_eq, lt, lt_eq, neq, not_distinct};
 use crate::arrow::compute::kernels::comparison::in_list_utf8;
 use crate::arrow::compute::kernels::numeric::{add, div, mul, sub};
@@ -44,6 +41,7 @@ use crate::expressions::{
     UnaryPredicateOp, VariadicExpression, VariadicExpressionOp,
 };
 use crate::schema::{DataType, PrimitiveType, StructField, StructType};
+use crate::timestamp_timezone::{parse_partition_scalar, TimestampTimezone};
 
 #[internal_api]
 pub(crate) trait ProvidesColumnByName {
@@ -414,95 +412,6 @@ pub fn evaluate_expression(
     }
 }
 
-#[derive(Clone, Copy)]
-enum TimestampTimezone {
-    Arrow(Tz),
-    Fixed(FixedOffset),
-}
-
-impl TimestampTimezone {
-    fn parse(value: &str) -> DeltaResult<Self> {
-        if value.starts_with('+') || value.starts_with('-') {
-            return parse_fixed_offset(value)
-                .map(Self::Fixed)
-                .ok_or_else(|| Error::generic(format!("Invalid timestamp timezone: {value}")));
-        }
-        value
-            .parse::<Tz>()
-            .map(Self::Arrow)
-            .map_err(|_| Error::generic(format!("Invalid timestamp timezone: {value}")))
-    }
-
-    fn parse_timestamp(self, raw: &str) -> Result<i64, ArrowError> {
-        match self {
-            Self::Arrow(timezone) => parse_timestamp_with_timezone(timezone, raw),
-            Self::Fixed(timezone) => parse_timestamp_with_timezone(timezone, raw),
-        }
-    }
-}
-
-fn parse_fixed_offset(value: &str) -> Option<FixedOffset> {
-    let (sign, value) = match value.as_bytes().first()? {
-        b'+' => (1, &value[1..]),
-        b'-' => (-1, &value[1..]),
-        _ => return None,
-    };
-    let mut parts = value.split(':');
-    let parse_component = |part: &str| {
-        (!part.is_empty() && part.bytes().all(|byte| byte.is_ascii_digit()))
-            .then(|| part.parse::<i32>().ok())
-            .flatten()
-    };
-    let hours = parse_component(parts.next()?)?;
-    let minutes = parts.next().map_or(Some(0), parse_component)?;
-    let seconds = parts.next().map_or(Some(0), parse_component)?;
-    if parts.next().is_some()
-        || value.is_empty()
-        || hours > 18
-        || minutes > 59
-        || seconds > 59
-        || (hours == 18 && (minutes != 0 || seconds != 0))
-    {
-        return None;
-    }
-    FixedOffset::east_opt(sign * (hours * 3_600 + minutes * 60 + seconds))
-}
-
-fn parse_timestamp_with_timezone<T: TimeZone + Copy>(
-    timezone: T,
-    raw: &str,
-) -> Result<i64, ArrowError> {
-    if let Ok(timestamp) = string_to_datetime(&timezone, raw) {
-        return Ok(timestamp.timestamp_micros());
-    }
-    let local_datetime = string_to_datetime(&Utc, raw)?.naive_utc();
-    resolve_local_timestamp(local_datetime, timezone).ok_or_else(|| {
-        ArrowError::ParseError(format!("Error resolving local timestamp from '{raw}'"))
-    })
-}
-
-fn resolve_local_timestamp<T: TimeZone + Copy>(
-    local_datetime: NaiveDateTime,
-    timezone: T,
-) -> Option<i64> {
-    if let Some(timestamp) = timezone.from_local_datetime(&local_datetime).earliest() {
-        return Some(timestamp.timestamp_micros());
-    }
-
-    // A forward transition can skip more than one hour, so walk backward until reaching the
-    // pre-transition side of the gap.
-    let offset = (1..=48).find_map(|hours| {
-        let before_transition = local_datetime.checked_sub_signed(TimeDelta::hours(hours))?;
-        timezone
-            .from_local_datetime(&before_transition)
-            .earliest()
-            .map(|timestamp| timestamp.offset().fix())
-    })?;
-    local_datetime
-        .checked_sub_signed(TimeDelta::seconds(i64::from(offset.local_minus_utc())))
-        .map(|utc_datetime| utc_datetime.and_utc().timestamp_micros())
-}
-
 /// Evaluate an `ARRAY(e0, e1, ..., eN-1)` constructor expression into an Arrow `ListArray`.
 ///
 /// Each input expression produces one column of length M (rows in the batch); the output
@@ -1032,44 +941,6 @@ pub fn coalesce_arrays(
     Ok(make_array(mutable.freeze()))
 }
 
-/// Parses one raw partition-value string into its target [`Scalar`], or `None` for a null value.
-///
-/// An empty string casts via [`PrimitiveType::empty_string_partition_cast`].
-///
-/// Uses map-to-struct's date and timestamp semantics.
-fn parse_partition_scalar(
-    prim: &PrimitiveType,
-    raw: &str,
-    timestamp_timezone: TimestampTimezone,
-) -> DeltaResult<Option<Scalar>> {
-    if raw.is_empty() {
-        return Ok(prim.empty_string_partition_cast());
-    }
-    match prim {
-        PrimitiveType::Date => {
-            let days = Date32Type::parse(raw).ok_or_else(|| {
-                Error::ParseError(raw.to_string(), DataType::Primitive(prim.clone()))
-            })?;
-            return Ok(Some(Scalar::Date(days)));
-        }
-        PrimitiveType::Timestamp => {
-            let micros = timestamp_timezone.parse_timestamp(raw).map_err(|_| {
-                Error::ParseError(raw.to_string(), DataType::Primitive(prim.clone()))
-            })?;
-            return Ok(Some(Scalar::Timestamp(micros)));
-        }
-        PrimitiveType::TimestampNtz => {
-            let micros = string_to_datetime(&Utc, raw)
-                .map_err(|_| Error::ParseError(raw.to_string(), DataType::Primitive(prim.clone())))?
-                .timestamp_micros();
-            return Ok(Some(Scalar::TimestampNtz(micros)));
-        }
-        _ => {}
-    }
-    let scalar = prim.parse_scalar(raw)?;
-    Ok((!matches!(scalar, Scalar::Null(_))).then_some(scalar))
-}
-
 /// Evaluates `MAP_TO_STRUCT(map_col, output_schema)`: extracts keys from a `Map<String, String>`
 /// and parses each value into its target type, producing a `StructArray`. An empty-string value
 /// casts via [`PrimitiveType::empty_string_partition_cast`].
@@ -3008,35 +2879,6 @@ mod tests {
         );
     }
 
-    #[rstest]
-    #[case::zero("+0", 0)]
-    #[case::hours("+5", 18_000)]
-    #[case::hours_minutes("-05:30", -19_800)]
-    #[case::hours_minutes_seconds("+12:45:30", 45_930)]
-    #[case::positive_limit("+18:00:00", 64_800)]
-    #[case::negative_limit("-18", -64_800)]
-    fn test_parse_fixed_offset_accepts_supported_forms(
-        #[case] timezone: &str,
-        #[case] expected_seconds: i32,
-    ) {
-        assert_eq!(
-            parse_fixed_offset(timezone).map(|offset| offset.local_minus_utc()),
-            Some(expected_seconds)
-        );
-    }
-
-    #[rstest]
-    #[case::empty("")]
-    #[case::bare_positive_sign("+")]
-    #[case::over_max_second("+18:00:01")]
-    #[case::over_max_hour("+19")]
-    #[case::minutes_out_of_range("+00:60")]
-    #[case::seconds_out_of_range("+00:00:60")]
-    #[case::extra_component("+05:30:00:00")]
-    fn test_parse_fixed_offset_rejects_invalid_forms(#[case] timezone: &str) {
-        assert_eq!(parse_fixed_offset(timezone), None);
-    }
-
     #[test]
     fn test_map_to_struct_timestamp_timezone_does_not_affect_timestamp_ntz() {
         let raw = "2024-01-15 12:30:45.123456";

kernel/src/checkpoint/checkpoint_shape.rs

@@ -114,6 +114,7 @@ pub mod table_changes;
 pub mod table_configuration;
 pub mod table_features;
 pub mod table_properties;
+mod timestamp_timezone;
 pub mod transaction;
 pub mod transforms;