akupchinskiy opened a new issue, #27046:
URL: https://github.com/apache/datafusion/issues/27046
## Describe the bug
A grouped `first_value(x ORDER BY ts)` / `last_value(x ORDER BY ts)` where
`ts` is
`Timestamp(_, Some(tz))` does not scale with `target_partitions`; it gets
slower as partitions are
added. The same query ordered by a timestamp without a time zone (same
instants) scales normally.
At 10 partitions the time zone aware version is about 5x slower here, and up
to 14x on a
latest-state dedup query.
`DISTINCT ON (k, ts) ... ORDER BY k, ts, ...` is planned as one
`first_value(col ORDER BY k, ts, ...)`
per output column, so every `DISTINCT ON` with a time zone aware key is
affected.
Found by a benchmark (digibench, a TPC-style benchmark on synthetic
digital-economy data that
runs the same analytical queries on several engines): a `DISTINCT ON
(session_id, event_name, client_ts)`
dedup query spent 92% of its DataFusion CPU in these aggregates and was 25x
slower than DuckDB.
## To Reproduce
`datafusion-cli` (release build of `main` at f058e2d), data generated in
SQL: 2M rows, 1M groups of
2 rows whose order in `ts` is scrambled (the second row wins in about two
thirds of the groups);
`ts` is `Timestamp(µs, "UTC")`, `tsn` the same instants without a time zone.
```sql
COPY (
SELECT value AS id,
value / 2 AS k1,
(value / 2) % 3 AS k2,
(value * 7919) % 1000003 AS o,
value % 977 AS v,
arrow_cast(to_timestamp_micros(1735689600000000 + ((value *
2654435761) % 1000000007) * 1000), 'Timestamp(Microsecond, Some("UTC"))') AS ts,
to_timestamp_micros(1735689600000000 + ((value * 2654435761) %
1000000007) * 1000) AS tsn
FROM generate_series(0, 1999999)
) TO 'fvtz.parquet' STORED AS PARQUET;
CREATE EXTERNAL TABLE t STORED AS PARQUET LOCATION 'fvtz.parquet';
SET datafusion.execution.target_partitions = 10;
-- slow
SELECT count(*), sum(a) FROM (SELECT k1, k2, first_value(v ORDER BY ts) a
FROM t GROUP BY k1, k2);
-- fast
SELECT count(*), sum(a) FROM (SELECT k1, k2, first_value(v ORDER BY tsn) a
FROM t GROUP BY k1, k2);
-- same for DISTINCT ON
SELECT count(*), sum(v) FROM (SELECT DISTINCT ON (k1) v FROM t ORDER BY k1,
ts, o);
SELECT count(*), sum(v) FROM (SELECT DISTINCT ON (k1) v FROM t ORDER BY k1,
tsn, o);
```
Median of 5 hot runs after a warm-up, ms (Apple M2 Pro, 10 cores, otherwise
idle):
| target_partitions | first_value ORDER BY ts (UTC) | ORDER BY tsn (no tz) |
slowdown | last_value ts / tsn | DISTINCT ON ts / tsn |
|---|---|---|---|---|---|
| 1 | 193 | 190 | 1.0x | 191 / 190 | 266 / 264 |
| 4 | 324 | 106 | 3.1x | 337 / 105 | 443 / 164 |
| 10 | 491 | 98 | 5.0x | 485 / 97 | 558 / 154 |
The time zone aware version gets slower with every partition added, the
naive one faster.
A latest-state-per-order query (10M events, `DISTINCT ON (order_id) ...
ORDER BY order_id,
ingest_ts DESC` with a UTC `ingest_ts`, default partitions) takes 6.8 s
against 0.5 s with the
same instants as a naive timestamp.
## Expected behavior
Ordering by a time zone aware timestamp costs the same as ordering by a
naive one: the time zone
does not take part in the comparison (the values are UTC instants).
Time zone aware timestamps are the common case for lakehouse data: Parquet
written by Spark, Trino,
DuckDB (`TIMESTAMPTZ`) or pandas with `isAdjustedToUTC = true`, Delta
`timestamp` and Iceberg
`timestamptz` columns all arrive as `Timestamp(_, Some("UTC"))`. "Latest row
per key" over such
data (`DISTINCT ON (key) ... ORDER BY key, ts DESC`, last price per
instrument, CDC dedup) hits
this path whenever DataFusion runs with more than one partition (the
default).
## Additional context
`FirstLastGroupsAccumulator` keeps the ordering values of each group's
current winner as
`orderings: Vec<Vec<ScalarValue>>`
([first_last.rs#L387](https://github.com/apache/datafusion/blob/f058e2d5367f761fc17ce62879339853e141f654/datafusion/functions-aggregate/src/first_last.rs#L387)).
A `ScalarValue::Timestamp*(v, Some(tz))` holds the time zone as an
`Arc<str>`, and that `Arc` is the
one from the column's `DataType`, shared by every partition. Per updated
group and batch it is cloned
and dropped several times:
- `extract_row_at_idx_to_buf` → `ScalarValue::try_from_array` clones the tz
from the array's data type
([common/src/utils/mod.rs#L84-L99](https://github.com/apache/datafusion/blob/f058e2d5367f761fc17ce62879339853e141f654/datafusion/common/src/utils/mod.rs#L84-L99),
[scalar/mod.rs#L4379](https://github.com/apache/datafusion/blob/f058e2d5367f761fc17ce62879339853e141f654/datafusion/common/src/scalar/mod.rs#L4379)),
and drops the previous row's values when the buffer is cleared;
- `update_state` drops the old winner and clones the new one: `clear()` +
`extend_from_slice()`
([first_last.rs#L538-L542](https://github.com/apache/datafusion/blob/f058e2d5367f761fc17ce62879339853e141f654/datafusion/functions-aggregate/src/first_last.rs#L538-L542));
- `resize_states` clones `default_orderings` (a `Timestamp*(None,
Some(tz))`) for every new group
([first_last.rs#L509-L510](https://github.com/apache/datafusion/blob/f058e2d5367f761fc17ce62879339853e141f654/datafusion/functions-aggregate/src/first_last.rs#L509-L510)),
and all of them are dropped on emit.
Each clone or drop is an atomic increment or decrement of the same reference
count, so with N
partitions N threads hammer one cache line. With 1M groups that is several
million contended atomics
per aggregation phase; the per-row work is otherwise small, so the
contention dominates.
The non-grouped `FirstValueAccumulator` / `LastValueAccumulator` (used
without GROUP BY, or for
value types the groups accumulator does not support) extract their ordering
row once per
accumulator and batch rather than once per updated group, so they are much
less exposed.
Possible fixes:
1. Narrow: keep the ordering values in the accumulator without the time zone
and put it back on the
arrays returned by `state()`. No change to results, state schema or
plans. I have a PR for this.
2. Broader (follow-up): stop storing per-group orderings as `ScalarValue`s
at all, e.g. in the
`arrow::row` format or as typed columns. That would also remove the
remaining per-group
`ScalarValue` work for every ordering type and could share infrastructure
with the coalescing work in #23682.
This also matters for #23603 / #23824 (rewriting `row_number() = 1` dedup to
`first_value`): on a time zone
aware key the rewrite would currently route the query into this slow path.
Environment: DataFusion `main` at f058e2d (2026-11-02), also reproduced on
55.0.0
(Python bindings), macOS 15.3.1, Apple M2 Pro (10 cores), Rust 1.99.1.
## Willingness to contribute
I plan to contribute a fix for this bug independently
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
