first_value/last_value ORDER BY a time zone aware timestamp gets slower with more partitions (Arc<str> refcount contention in FirstLastGroupsAccumulator) #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 with a benchmark of my own 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.

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):

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).
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,
  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);
• resize_states clones default_orderings (a Timestamp*(None, Some(tz))) for every new group
  (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 ScalarValues 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 feat(optimizer): coalesce peer first_value / last_value into a single struct aggregate #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