Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 25 additions & 2 deletions crates/datafusion/src/read/predicate.rs
Original file line number Diff line number Diff line change
Expand Up @@ -462,10 +462,21 @@ fn scalar_value_to_datum(value: &ScalarValue) -> Option<Datum> {
// DataFusion's type coercion always converts them to match the column type
// (either TimestampMicrosecond or TimestampNanosecond) before predicate pushdown.
// See unit tests for how those conversions would work if needed.
ScalarValue::TimestampMicrosecond(Some(v), _) => {
ScalarValue::TimestampMicrosecond(Some(v), None) => {
Some(Datum::timestamp_micros(*v))
}
ScalarValue::TimestampNanosecond(Some(v), _) => Some(Datum::timestamp_nanos(*v)),
ScalarValue::TimestampNanosecond(Some(v), None) => {
Some(Datum::timestamp_nanos(*v))
}
// Note: Arrow stores tz-aware timestamps as UTC epoch values; the tz
// only affects display. The same is true for iceberg's tz-aware datum
// types.
ScalarValue::TimestampMicrosecond(Some(v), Some(_)) => {
Some(Datum::timestamptz_micros(*v))
}
ScalarValue::TimestampNanosecond(Some(v), Some(_)) => {
Some(Datum::timestamptz_nanos(*v))
}
_ => None,
}
}
Expand Down Expand Up @@ -707,6 +718,18 @@ mod tests {
));
assert_eq!(datum, Some(Datum::timestamp_nanos(ts_nanos)));

let datum = super::scalar_value_to_datum(&ScalarValue::TimestampMicrosecond(
Some(ts_micros),
Some("+01:00".into()),
));
assert_eq!(datum, Some(Datum::timestamptz_micros(ts_micros)));

let datum = super::scalar_value_to_datum(&ScalarValue::TimestampNanosecond(
Some(ts_nanos),
Some("UTC".into()),
));
assert_eq!(datum, Some(Datum::timestamptz_nanos(ts_nanos)));

// Test None timestamp
let datum =
super::scalar_value_to_datum(&ScalarValue::TimestampMicrosecond(None, None));
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -181,3 +181,92 @@ CREATE TABLE default.default.test_timestamp_millis (id INT NOT NULL, ts TIMESTAM
# This should fail because Iceberg doesn't support second precision
statement error (?s)DataFusion error: IcebergError\(DataInvalid\).*Unsupported Arrow data type: Timestamp\(s\)
CREATE TABLE default.default.test_timestamp_seconds (id INT NOT NULL, ts TIMESTAMP(0))

# Without this, `TIMESTAMP WITH TIME ZONE` in `CREATE TABLE` ends up as a normal
# timestamp column.
statement ok
SET datafusion.execution.time_zone = 'UTC'

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Could you add a short comment on why this is needed? I'm guessing it's so ::TIMESTAMPTZ ends up as +00:00. Also it sticks around for anything added to this file later, so maybe reset it at the end next to the DROP.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It's confusing, but without it the CREATE TABLE ends up with a normal timestamp column, even with TIMESTAMP WITH TIME ZONE specified.


statement ok
CREATE TABLE default.default.test_timestamptz (id INT NOT NULL, ts_us TIMESTAMP(6) WITH TIME ZONE, ts_ns TIMESTAMP WITH TIME ZONE, ts_naive TIMESTAMP)

query I
INSERT INTO default.default.test_timestamptz VALUES
(1, '2023-01-04T12:00:00+00:00'::TIMESTAMPTZ, '2023-01-04T12:00:00+00:00'::TIMESTAMPTZ, CAST('2023-01-04 12:00:00' AS TIMESTAMP)),
(2, '2023-01-06T12:00:00+00:00'::TIMESTAMPTZ, '2023-01-06T12:00:00+00:00'::TIMESTAMPTZ, CAST('2023-01-06 12:00:00' AS TIMESTAMP))
----
2

query TT
EXPLAIN SELECT id FROM default.default.test_timestamptz WHERE ts_us > '2023-01-05T00:00:00+00:00'::TIMESTAMPTZ

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Can we also cover the mixed cases? e.g. ts_us > CAST('2023-01-05' AS TIMESTAMP) (tz column vs naive literal), and a naive TIMESTAMP column vs a ::TIMESTAMPTZ literal. Once the upstream conversion is gone a mismatched datum would fail to bind and I think the query would just error, so it'd be good to lock in that coercion either gives us a matching literal or skips pushdown.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done.

----
logical_plan
01)Projection: default.default.test_timestamptz.id
02)--Filter: default.default.test_timestamptz.ts_us > TimestampMicrosecond(1672876800000000, Some("+00:00"))
03)----TableScan: default.default.test_timestamptz projection=[id, ts_us], partial_filters=[default.default.test_timestamptz.ts_us > TimestampMicrosecond(1672876800000000, Some("+00:00"))]
physical_plan
01)FilterExec: ts_us@1 > 1672876800000000, projection=[id@0]
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
03)----DataSourceExec: format=iceberg, projection=[id, ts_us], predicate=ts_us > 2023-01-05 00:00:00 UTC

query I
SELECT id FROM default.default.test_timestamptz WHERE ts_us > '2023-01-05T00:00:00+00:00'::TIMESTAMPTZ
----
2

query TT
EXPLAIN SELECT id FROM default.default.test_timestamptz WHERE ts_ns > '2023-01-05T00:00:00+00:00'::TIMESTAMPTZ
----
logical_plan
01)Projection: default.default.test_timestamptz.id
02)--Filter: default.default.test_timestamptz.ts_ns > TimestampNanosecond(1672876800000000000, Some("+00:00"))
03)----TableScan: default.default.test_timestamptz projection=[id, ts_ns], partial_filters=[default.default.test_timestamptz.ts_ns > TimestampNanosecond(1672876800000000000, Some("+00:00"))]
physical_plan
01)FilterExec: ts_ns@1 > 1672876800000000000, projection=[id@0]
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
03)----DataSourceExec: format=iceberg, projection=[id, ts_ns], predicate=ts_ns > 2023-01-05 00:00:00 UTC

query I
SELECT id FROM default.default.test_timestamptz WHERE ts_ns > '2023-01-05T00:00:00+00:00'::TIMESTAMPTZ
----
2

query TT
EXPLAIN SELECT id FROM default.default.test_timestamptz WHERE ts_us > CAST('2023-01-05 00:00:00' AS TIMESTAMP)
----
logical_plan
01)Projection: default.default.test_timestamptz.id
02)--Filter: default.default.test_timestamptz.ts_us > TimestampMicrosecond(1672876800000000, Some("+00:00"))
03)----TableScan: default.default.test_timestamptz projection=[id, ts_us], partial_filters=[default.default.test_timestamptz.ts_us > TimestampMicrosecond(1672876800000000, Some("+00:00"))]
physical_plan
01)FilterExec: ts_us@1 > 1672876800000000, projection=[id@0]
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
03)----DataSourceExec: format=iceberg, projection=[id, ts_us], predicate=ts_us > 2023-01-05 00:00:00 UTC

query I
SELECT id FROM default.default.test_timestamptz WHERE ts_us > CAST('2023-01-05 00:00:00' AS TIMESTAMP)
----
2

query TT
EXPLAIN SELECT id FROM default.default.test_timestamptz WHERE ts_naive > '2023-01-05T00:00:00+00:00'::TIMESTAMPTZ
----
logical_plan
01)Projection: default.default.test_timestamptz.id
02)--Filter: default.default.test_timestamptz.ts_naive > TimestampNanosecond(1672876800000000000, None)
03)----TableScan: default.default.test_timestamptz projection=[id, ts_naive], partial_filters=[default.default.test_timestamptz.ts_naive > TimestampNanosecond(1672876800000000000, None)]
physical_plan
01)FilterExec: ts_naive@1 > 1672876800000000000, projection=[id@0]
02)--RepartitionExec: partitioning=RoundRobinBatch(4), input_partitions=1
03)----DataSourceExec: format=iceberg, projection=[id, ts_naive], predicate=ts_naive > 2023-01-05 00:00:00

query I
SELECT id FROM default.default.test_timestamptz WHERE ts_naive > '2023-01-05T00:00:00+00:00'::TIMESTAMPTZ
----
2

statement ok
DROP TABLE default.default.test_timestamptz

statement ok
RESET datafusion.execution.time_zone
Loading