Skip to content

Commit 0b7433a

Browse files
committed
feat: honor reader timezone in partition values
1 parent bbe47bc commit 0b7433a

21 files changed

Lines changed: 1335 additions & 409 deletions

File tree

datafusion-executor/src/expression.rs

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -472,6 +472,7 @@ fn map_to_struct_primitive(field: &StructField) -> DeltaResult<&PrimitiveType> {
472472
#[derive(Debug, PartialEq, Eq)]
473473
struct KernelMapToStructUdf {
474474
output_schema: KernelSchemaRef,
475+
output_type: KernelDataType,
475476
options: MapToStructOptions,
476477
return_type: ArrowDataType,
477478
signature: Signature,
@@ -493,9 +494,11 @@ impl KernelMapToStructUdf {
493494
.as_ref()
494495
.try_into_arrow()
495496
.map_err(Error::generic_err)?;
497+
let output_type = KernelDataType::from(output_schema.as_ref().clone());
496498
Ok(Self {
497499
return_type: ArrowDataType::Struct(arrow_schema.fields().clone()),
498500
output_schema,
501+
output_type,
499502
options,
500503
signature: Signature::any(1, Volatility::Immutable),
501504
})
@@ -523,9 +526,8 @@ impl ScalarUDFImpl for KernelMapToStructUdf {
523526
KernelExpression::column(["map"]),
524527
self.options.clone(),
525528
);
526-
let output_type = KernelDataType::from(self.output_schema.as_ref().clone());
527529
let result =
528-
kernel_expression::evaluate_expression(&expression, &batch, Some(&output_type))
530+
kernel_expression::evaluate_expression(&expression, &batch, Some(&self.output_type))
529531
.map_err(|e| DataFusionError::External(Box::new(e)))?;
530532
Ok(ColumnarValue::Array(result))
531533
}

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

Lines changed: 20 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -249,9 +249,9 @@ If the scan has no predicate, this returns `None`.
249249
## Typed partition values
250250

251251
Kernel reads each file's partition values from the Delta log and exposes them on its
252-
`ScanFile` as a raw string map. Kernel's `transform_to_logical` could materialize them as
253-
typed columns. If your connector assembles output rows itself instead of using that transform,
254-
it parses the string map per file.
252+
`ScanFile` as a raw string map. `transform_to_logical` materializes them as typed columns. A
253+
connector that assembles output rows itself instead of using that transform would otherwise need
254+
to parse the string map per file.
255255

256256
To have Kernel hand you the typed values directly, opt in with `with_partition_values`:
257257

@@ -270,7 +270,10 @@ To have Kernel hand you the typed values directly, opt in with `with_partition_v
270270
# let snapshot = Snapshot::builder_for(url).build(&engine)?;
271271
let scan = snapshot
272272
.scan_builder()
273-
.with_partition_values(PartitionValuesOptions::with_struct())
273+
.with_partition_values(
274+
PartitionValuesOptions::with_struct()
275+
.with_timestamp_timezone("America/Los_Angeles"),
276+
)
274277
.build()?;
275278
# Ok(())
276279
# }
@@ -281,9 +284,20 @@ nullable field per partition column (by physical name). You read it as a typed c
281284
of parsing the string map per file. The raw string map is still present, so this option only
282285
adds the typed column.
283286

287+
By default, offset-less `TIMESTAMP` partition strings are interpreted in UTC. Use
288+
`with_timestamp_timezone` with a recognized IANA timezone or a normalized `+HH:MM`, `-HH:MM`,
289+
fixed offset when the reader uses another timezone. An explicit offset or embedded time zone in a
290+
partition value takes precedence. This setting affects typed `scan_metadata` output, partition
291+
predicate evaluation after log replay, and the partition-column row transforms used by
292+
`Scan::execute`. Checkpoint footer pruning continues to use the checkpoint's native parsed
293+
partition values. Incremental scans expose the raw partition-value map. `TIMESTAMP_NTZ` remains
294+
timezone-independent. For daylight-saving transitions, ambiguous local times use the earlier
295+
instant, and nonexistent local times use the offset from before the transition.
296+
284297
> [!TIP]
285-
> When the checkpoint already stores typed partition values, Kernel reads that column directly
286-
> and skips parsing entirely.
298+
> Kernel reparses surviving commit and checkpoint rows from the raw map for typed output and final
299+
> predicate evaluation. Checkpoint footer pruning happens first and continues to use the
300+
> checkpoint's native parsed partition values.
287301
288302
## Cancelling a scan
289303

kernel/Cargo.toml

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -32,7 +32,7 @@ pre-release-hook = [
3232
delta_kernel_derive = { path = "../derive-macros", version = "0.28.0" }
3333
bytes = "1.10"
3434
chrono = "0.4.41"
35-
chrono-tz = { version = "0.10.4", optional = true }
35+
chrono-tz = "0.10.4"
3636
crc = "3.2.2"
3737
indexmap = "2.10.0"
3838
itertools = "0.14"
@@ -100,7 +100,7 @@ need-arrow = [] # need-arrow is a marker that the feature needs arrow dep
100100
arrow-58 = ["dep:arrow_58", "dep:parquet_58", "dep:object_store_13"]
101101
arrow-59 = ["dep:arrow_59", "dep:parquet_59", "dep:object_store_13"]
102102
arrow-conversion = ["need-arrow"]
103-
arrow-expression = ["need-arrow", "dep:chrono-tz"]
103+
arrow-expression = ["need-arrow"]
104104

105105
# Schema diffing functionality (experimental)
106106
schema-diff = []

0 commit comments

Comments
 (0)