refactor!: share partition timestamp parsing

datafusion-executor/src/expression.rs

@@ -404,10 +404,10 @@ fn struct_columns_from_patch(
 ///   while kernel accepts only `"true"`/`"false"`.
 /// - Decimal: arrow's cast silently rescales/rounds to the target scale, while kernel requires the
 ///   value's scale to match the target's exactly (and hard-errors otherwise).
-/// - A timestamp with a trailing named timezone is accepted by the kernel parser but not by the
-///   native DataFusion cast.
+/// - A timestamp with a trailing named timezone is accepted by the native DataFusion cast but not
+///   by the kernel parser.
 ///
-/// Configured options use a kernel-backed UDF so reader-timezone parsing follows kernel semantics.
+/// Configured options use a kernel-backed UDF so timezone parsing follows kernel semantics.
 ///
 /// # Errors
 ///

kernel/Cargo.toml

@@ -32,7 +32,7 @@ pre-release-hook = [
 delta_kernel_derive = { path = "../derive-macros", version = "0.28.0" }
 bytes = "1.10"
 chrono = "0.4.41"
-chrono-tz = { version = "0.10.4", optional = true }
+chrono-tz = "0.10.4"
 crc = "3.2.2"
 indexmap = "2.10.0"
 itertools = "0.14"
@@ -100,7 +100,7 @@ need-arrow = [] # need-arrow is a marker that the feature needs arrow dep
 arrow-58 = ["dep:arrow_58", "dep:parquet_58", "dep:object_store_13"]
 arrow-59 = ["dep:arrow_59", "dep:parquet_59", "dep:object_store_13"]
 arrow-conversion = ["need-arrow"]
-arrow-expression = ["need-arrow", "dep:chrono-tz"]
+arrow-expression = ["need-arrow"]
 
 # Schema diffing functionality (experimental)
 schema-diff = []

kernel/src/engine/arrow_expression/evaluate_expression.rs

@@ -7,7 +7,6 @@ use chrono::Utc;
 use itertools::Itertools;
 use tracing::warn;
 
-use super::timestamp_timezone::TimestampTimezone;
 use crate::arrow::array::types::*;
 use crate::arrow::array::{
     self as arrow_array, make_array, new_null_array, Array, ArrayBuilder, ArrayData, ArrayRef,
@@ -43,6 +42,7 @@ use crate::expressions::{
     OpaquePredicate, Predicate, Scalar, UnaryExpression, UnaryExpressionOp, UnaryPredicate,
     UnaryPredicateOp, VariadicExpression, VariadicExpressionOp,
 };
+use crate::partition_values::TimestampTimezone;
 use crate::schema::{DataType, PrimitiveType, StructField, StructType};
 
 #[internal_api]
@@ -937,8 +937,8 @@ fn coalesce_arrays(
 /// 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`].
-/// `timestamp_timezone` applies only to `TIMESTAMP` values without an embedded offset or named
-/// timezone; it does not affect `DATE` or `TIMESTAMP_NTZ`.
+/// `timestamp_timezone` applies only to space-separated `TIMESTAMP` values; it does not affect
+/// ISO 8601 timestamps with an explicit offset, `DATE`, or `TIMESTAMP_NTZ`.
 fn parse_partition_scalar(
     prim: &PrimitiveType,
     raw: &str,
@@ -975,7 +975,7 @@ fn parse_partition_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`].
-/// `timestamp_timezone` controls `TIMESTAMP` values without an embedded offset or named timezone.
+/// `timestamp_timezone` controls space-separated `TIMESTAMP` values without an explicit offset.
 ///
 /// - Missing keys produce null values
 /// - Parse errors are propagated (indicating a broken table)
@@ -2895,29 +2895,19 @@ mod tests {
     )]
     #[case::explicit_input_offset(
         Some("America/Los_Angeles"),
-        "2024-01-15 12:30:45+02:00",
+        "2024-01-15T12:30:45+02:00",
         "2024-01-15T10:30:45Z"
     )]
     #[case::explicit_input_offset_over_fixed_reader(
         Some("+05:30"),
-        "2024-01-15 12:30:45+02:00",
+        "2024-01-15T12:30:45+02:00",
         "2024-01-15T10:30:45Z"
     )]
     #[case::normalized_utc(
         Some("America/Los_Angeles"),
         "2024-01-15T12:30:45.123456Z",
         "2024-01-15T12:30:45.123456Z"
     )]
-    #[case::embedded_iana_timezone(
-        Some("Europe/Berlin"),
-        "2024-01-15 12:30:45 America/New_York",
-        "2024-01-15T17:30:45Z"
-    )]
-    #[case::embedded_iana_timezone_with_default_options(
-        None,
-        "2024-01-15 12:30:45 America/New_York",
-        "2024-01-15T17:30:45Z"
-    )]
     #[case::dst_overlap(
         Some("America/Los_Angeles"),
         "2024-11-03 01:30:00",

kernel/src/engine/arrow_expression/mod.rs

@@ -20,7 +20,6 @@ use crate::{EngineData, EvaluationHandler, ExpressionEvaluator, PredicateEvaluat
 
 pub mod evaluate_expression;
 pub mod opaque;
-mod timestamp_timezone;
 
 #[cfg(test)]
 mod tests;

kernel/src/expressions/mod.rs

@@ -662,8 +662,8 @@ impl ParseJsonExpression {
 /// Connector-supplied options controlling how a [`MapToStructExpression`] parses map values.
 ///
 /// Kernel does not infer these settings from the host environment or table metadata.
-/// Expression producers must use one reader timezone for all partition-value expressions in a
-/// scan so materialization and pruning cannot interpret the same value differently.
+/// Expression producers must use one timezone for all partition-value expressions in a scan so
+/// materialization and pruning cannot interpret the same value differently.
 #[derive(Clone, Debug, Default, Eq, Hash, PartialEq, Serialize, Deserialize)]
 pub struct MapToStructOptions {
     #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -738,18 +738,16 @@ impl MapToStructOptions {
 ///   round or rescale.
 /// - BOOLEAN: accept case-insensitive `true` or `false`, with no numeric or yes/no aliases.
 /// - DATE: parse `{year}-{month}-{day}`.
-/// - TIMESTAMP: accept date-only values and timestamps with a space, `T`, or `t` separator,
-///   optional fractional seconds, an optional numeric offset, or a trailing IANA timezone. Parse
-///   values with an offset or timezone as absolute instants; parse offset-less values in the reader
-///   timezone from [`MapToStructOptions`], or UTC by default.
+/// - TIMESTAMP: accept the protocol's space-separated form and ISO 8601 values with an explicit
+///   offset. Parse space-separated values in the timezone from [`MapToStructOptions`], or UTC by
+///   default; an explicit offset takes precedence.
 /// - TIMESTAMP_NTZ: parse a space-separated timestamp without an offset and preserve the local
 ///   wall-clock value.
 /// - Interval types: parse an ANSI interval literal accepted by [`PrimitiveType::parse_scalar`].
 /// - VOID: reject every non-empty value.
 ///
-/// The reader timezone does not affect a timestamp carrying its own time zone or offset. Modern
-/// writers use the protocol's UTC-adjusted ISO 8601 form, which therefore reads independently of
-/// the configured reader timezone.
+/// The configured timezone does not affect a timestamp carrying an explicit offset. The protocol's
+/// UTC-adjusted ISO 8601 form therefore reads independently of the configured timezone.
 ///
 /// Non-empty geometry and geography values are unsupported. Struct, array, map, and variant target
 /// fields are not primitive partition types and are rejected. Any other unparseable non-empty value

kernel/src/lib.rs

@@ -104,6 +104,7 @@ mod log_path;
 mod log_reader;
 pub mod metrics;
 pub mod partition;
+pub(crate) mod partition_values;
 #[cfg(feature = "declarative-plans")]
 pub mod plans;
 pub mod scan;

kernel/src/engine/arrow_expression/timestamp_timezone.rs → kernel/src/partition_values.rs

@@ -1,9 +1,8 @@
-//! Arrow reader-timezone handling for `TIMESTAMP` partition values.
+//! Engine-independent timezone handling for `TIMESTAMP` partition values.
 //!
 //! Kernel never infers this timezone from the host or the Delta table. A connector supplies its
-//! reader/session timezone through [`MapToStructOptions`], and no option means UTC. Timestamp
-//! strings may carry their own offset or named timezone. This preserves the Arrow evaluator's
-//! partition-value compatibility; the embedded value takes precedence.
+//! timezone through [`MapToStructOptions`], and no option means UTC. A protocol-formatted ISO 8601
+//! timestamp carries its own offset, which takes precedence.
 //!
 //! # Terminology
 //!
@@ -12,10 +11,10 @@
 
 use std::str::FromStr;
 
-use chrono::{FixedOffset, LocalResult, NaiveDateTime, Offset, TimeDelta, TimeZone, Utc};
+use chrono::{DateTime, FixedOffset, LocalResult, NaiveDateTime, Offset, TimeDelta, TimeZone};
 use chrono_tz::Tz;
 
-use crate::arrow::compute::kernels::cast_utils::string_to_datetime;
+#[cfg(feature = "arrow-expression")]
 use crate::expressions::MapToStructOptions;
 use crate::{DeltaResult, Error};
 
@@ -53,12 +52,13 @@ impl FromStr for TimestampTimezone {
 }
 
 impl TimestampTimezone {
-    /// Resolves the reader timezone from map-to-struct options, defaulting to UTC.
+    /// Resolves the timezone from map-to-struct options, defaulting to UTC.
     ///
     /// # Errors
     ///
     /// Returns an error when the configured value is neither a recognized IANA timezone nor a
     /// fixed offset in `+HH:MM` or `-HH:MM` form.
+    #[cfg(feature = "arrow-expression")]
     pub(crate) fn try_from_options(options: &MapToStructOptions) -> DeltaResult<Self> {
         match options.timestamp_timezone() {
             Some(value) => value.parse(),
@@ -68,23 +68,15 @@ impl TimestampTimezone {
 
     /// Parses a partition timestamp into microseconds since the Unix epoch.
     ///
-    /// Arrow handles the timestamp grammar. An offset or timezone carried by `raw` takes
-    /// precedence over the configured reader timezone. For an offset-less value in a named reader
-    /// timezone, local clock transitions follow reader timestamp-cast semantics:
+    /// A protocol-formatted ISO 8601 value carries its own offset. A space-separated value uses
+    /// the configured timezone. For an offset-less value in a named timezone, local clock
+    /// transitions follow timestamp-cast semantics:
     ///
     /// - If a backward clock change makes a local time occur twice, use the earlier instant.
     /// - If a forward clock change skips a local time, use the offset from before the change. For
     ///   example, `02:30` in a one-hour spring-forward gap is interpreted as `03:30`.
-    ///
-    /// Arrow rejects both transition cases, so named timezones use a fallback only when Arrow
-    /// cannot resolve the local timestamp.
     pub(crate) fn parse_timestamp(self, raw: &str) -> Option<i64> {
-        match self {
-            Self::Named(timezone) => parse_timestamp_in_named_timezone(raw, timezone),
-            Self::Fixed(timezone) => string_to_datetime(&timezone, raw)
-                .ok()
-                .map(|timestamp| timestamp.timestamp_micros()),
-        }
+        parse_timestamp(raw, self)
     }
 }
 
@@ -109,23 +101,49 @@ fn parse_two_digits(value: &str) -> Option<i32> {
     let [tens, ones] = value.as_bytes() else {
         return None;
     };
-    (tens.is_ascii_digit() && ones.is_ascii_digit())
-        .then_some(i32::from(*tens - b'0') * 10 + i32::from(*ones - b'0'))
+    if !tens.is_ascii_digit() || !ones.is_ascii_digit() {
+        return None;
+    }
+    Some(i32::from(*tens - b'0') * 10 + i32::from(*ones - b'0'))
 }
 
-/// Parses with Arrow, resolving only named-zone clock transitions that Arrow rejects.
-fn parse_timestamp_in_named_timezone(raw: &str, timezone: Tz) -> Option<i64> {
-    if let Ok(timestamp) = string_to_datetime(&timezone, raw) {
-        return Some(timestamp.timestamp_micros());
+fn parse_timestamp(raw: &str, timezone: TimestampTimezone) -> Option<i64> {
+    if raw
+        .split_once('.')
+        .is_some_and(|(_, suffix)| suffix.bytes().take_while(u8::is_ascii_digit).count() > 6)
+    {
+        return None;
     }
-    let local_datetime = string_to_datetime(&Utc, raw).ok()?.naive_utc();
-    match timezone.from_local_datetime(&local_datetime) {
-        LocalResult::Ambiguous(first, second) => Some(first.min(second).timestamp_micros()),
-        LocalResult::None => resolve_nonexistent_local_timestamp(local_datetime, timezone),
-        LocalResult::Single(_) => {
-            // A unique local time means Arrow rejected the input for another reason.
-            None
-        }
+    if let Ok(local_datetime) = NaiveDateTime::parse_from_str(raw, "%Y-%m-%d %H:%M:%S%.f") {
+        return resolve_local_timestamp(local_datetime, timezone);
+    }
+    parse_explicit_offset_timestamp(raw)
+}
+
+fn parse_explicit_offset_timestamp(raw: &str) -> Option<i64> {
+    // RFC 3339 permits lowercase separators, but the protocol form uses uppercase `T` and `Z`.
+    if raw.as_bytes().get(10) != Some(&b'T') || raw.ends_with('z') {
+        return None;
+    }
+    let timestamp = DateTime::parse_from_rfc3339(raw).ok()?;
+    (timestamp.offset().local_minus_utc().unsigned_abs() <= 18 * 3_600)
+        .then(|| timestamp.timestamp_micros())
+}
+
+fn resolve_local_timestamp(
+    local_datetime: NaiveDateTime,
+    timezone: TimestampTimezone,
+) -> Option<i64> {
+    match timezone {
+        TimestampTimezone::Fixed(timezone) => timezone
+            .from_local_datetime(&local_datetime)
+            .single()
+            .map(|timestamp| timestamp.timestamp_micros()),
+        TimestampTimezone::Named(timezone) => match timezone.from_local_datetime(&local_datetime) {
+            LocalResult::Ambiguous(first, second) => Some(first.min(second).timestamp_micros()),
+            LocalResult::None => resolve_nonexistent_local_timestamp(local_datetime, timezone),
+            LocalResult::Single(timestamp) => Some(timestamp.timestamp_micros()),
+        },
     }
 }
 
@@ -148,25 +166,27 @@ fn resolve_nonexistent_local_timestamp(local_datetime: NaiveDateTime, timezone:
 mod tests {
     use rstest::rstest;
 
-    use super::super::expected_timestamp_micros;
     use super::*;
 
-    fn options(timezone: &str) -> MapToStructOptions {
-        MapToStructOptions::default().with_timestamp_timezone(timezone)
+    fn expected_timestamp_micros(timestamp: &str) -> i64 {
+        DateTime::parse_from_rfc3339(timestamp)
+            .unwrap()
+            .timestamp_micros()
     }
 
     #[rstest]
     #[case::named("America/Los_Angeles")]
     #[case::minute_offset("+05:30")]
     #[case::positive_limit("+18:00")]
     #[case::negative_limit("-18:00")]
-    fn accepts_normalized_reader_timezones(#[case] timezone: &str) {
-        assert!(TimestampTimezone::try_from_options(&options(timezone)).is_ok());
+    fn accepts_normalized_timezones(#[case] timezone: &str) {
+        assert!(timezone.parse::<TimestampTimezone>().is_ok());
     }
 
     #[rstest]
     #[case::empty("")]
     #[case::bare_sign("+")]
+    #[case::malformed_sign("+-5:00")]
     #[case::compact_hour("+05")]
     #[case::compact_hour_minute("+0530")]
     #[case::invalid_minutes("+05:60")]
@@ -176,37 +196,38 @@ mod tests {
     #[case::invalid_hours("+19:00")]
     #[case::extra_component("+05:00:00:00")]
     #[case::unknown_name("Not/AZone")]
-    fn rejects_noncanonical_reader_timezones(#[case] timezone: &str) {
-        assert!(TimestampTimezone::try_from_options(&options(timezone)).is_err());
+    fn rejects_noncanonical_timezones(#[case] timezone: &str) {
+        assert!(timezone.parse::<TimestampTimezone>().is_err());
     }
 
     #[rstest]
     #[case::local("2024-01-15 12:30:45.123456", "2024-01-15T12:30:45.123456Z")]
-    #[case::compact_offset("2024-01-15T17:30:45+0530", "2024-01-15T12:00:45Z")]
-    #[case::offset_beyond_reader_limit("2024-01-15T12:30:45+19:00", "2024-01-14T17:30:45Z")]
+    #[case::utc("2024-01-15T12:30:45.123456Z", "2024-01-15T12:30:45.123456Z")]
+    #[case::positive_offset("2024-01-15T17:30:45+05:30", "2024-01-15T12:00:45Z")]
     #[case::negative_offset("2024-01-15T07:00:45-05:30", "2024-01-15T12:30:45Z")]
-    #[case::spaced_offset("2024-01-15 17:30:45 +05:30", "2024-01-15T12:00:45Z")]
-    #[case::named("2024-01-15 12:30:45 America/New_York", "2024-01-15T17:30:45Z")]
-    #[case::lowercase_t("2024-01-15t12:30:45", "2024-01-15T12:30:45Z")]
-    #[case::compact_clock("2024-01-15 123045", "2024-01-15T12:30:45Z")]
-    #[case::date_only("2024-01-15", "2024-01-15T00:00:00Z")]
-    #[case::excess_fraction("2024-01-15 12:30:45.123456789123", "2024-01-15T12:30:45.123456Z")]
-    fn parses_compatible_partition_timestamps(#[case] raw: &str, #[case] expected: &str) {
+    fn parses_protocol_partition_timestamps(#[case] raw: &str, #[case] expected: &str) {
         assert_eq!(
             TimestampTimezone::default().parse_timestamp(raw),
             Some(expected_timestamp_micros(expected))
         );
     }
 
     #[rstest]
-    #[case::conflicting("2024-01-15 12:30:45+02:00 America/New_York")]
-    #[case::unknown_embedded("2024-01-15 12:30:45 Foo/Bar")]
-    #[case::multibyte_prefix("日時 America/New_York")]
+    #[case::embedded_timezone("2024-01-15 12:30:45 America/New_York")]
+    #[case::spaced_offset("2024-01-15 17:30:45 +05:30")]
+    #[case::lowercase_t("2024-01-15t12:30:45")]
+    #[case::compact_clock("2024-01-15 123045")]
+    #[case::date_only("2024-01-15")]
+    #[case::zoneless_t("2024-01-15T12:30:45")]
+    #[case::compact_offset("2024-01-15T12:30:45+0530")]
+    #[case::hour_only_offset("2024-01-15T12:30:45+07")]
+    #[case::offset_past_limit("2024-01-15T12:30:45+19:00")]
+    #[case::sub_microsecond_precision("1969-12-31 23:59:59.999999500")]
+    #[case::ten_digit_fraction("2024-01-15 12:30:45.1234567891")]
     #[case::extra_offset_component("2024-01-15T12:30:45+05:30:15:00")]
     #[case::bad_compact_offset("2024-01-15T12:30:45+053")]
-    #[case::ampm("2024-01-15 01:30:45 PM")]
-    #[case::day_first("15/01/2024 12:30:45")]
-    #[case::time_only("12:30:45")]
+    #[case::invalid_suffix("2024-01-15T12:30:45XYZ")]
+    #[case::trailing_garbage_after_zone("2024-01-15T12:30:45ZXYZ")]
     fn rejects_unsupported_partition_timestamps(#[case] raw: &str) {
         assert_eq!(TimestampTimezone::default().parse_timestamp(raw), None);
     }

kernel/src/scan/tests.rs

@@ -748,11 +748,6 @@ fn test_get_partition_value() {
             PrimitiveType::Timestamp,
             Scalar::Timestamp(123456),
         ),
-        (
-            "1970-01-01 00:00:00.123456789",
-            PrimitiveType::Timestamp,
-            Scalar::Timestamp(123456),
-        ),
         (
             // RFC 3339 with a non-UTC offset: normalized to UTC (1969-12-31T19:00:00Z)
             "1970-01-01T00:00:00+05:00",