Repository navigation
Description
Describe the bug
Some window queries over a Timestamp(_, Some(tz)) column get slower as target_partitions
grows, while the same queries over the same instants without a time zone scale normally. At 1
partition the two take about the same time, and the profiles show clone/drop of ScalarValue and
DataType as hotspots. That points to contention on the time zone's Arc<str> refcount: every
partition shares it, and these paths clone and drop it about once per row.
It looks like the same mechanism as #27046, on window code paths.
To Reproduce
datafusion-cli from main (238ae32), release build, Apple M2 Pro (10 cores, on AC power).
2M rows; ts is Timestamp(µs, "UTC"), tsn holds the same instants without a time zone:
COPY (
SELECT value / 2 AS k1,
(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; -- and 1
SELECT max(r) FROM (SELECT max(ts) OVER (PARTITION BY k1 % 1000 ORDER BY o) r FROM t);
SELECT max(r) FROM (SELECT max(tsn) OVER (PARTITION BY k1 % 1000 ORDER BY o) r FROM t);Each query ran 4 times per setting; the table shows the median of the last 3, in ms:
query (x = ts or tsn) |
ts, 1 partition |
tsn, 1 partition |
ts, 10 partitions |
tsn, 10 partitions |
ts / tsn at 10 |
|---|---|---|---|---|---|
max(x) OVER (PARTITION BY k1 % 1000 ORDER BY o) |
500 | 497 | 1091 | 117 | 9.3x |
lag(x) OVER (PARTITION BY k1 ORDER BY o) |
635 | 635 | 1020 | 465 | 2.2x |
row_number() OVER (PARTITION BY x ORDER BY o) |
699 | 690 | 895 | 612 | 1.5x |
sum(v) OVER (PARTITION BY k1 ORDER BY x) |
1018 | 1008 | 862 | 602 | 1.4x |
first_value(x) OVER (PARTITION BY k1 ORDER BY o) |
800 | 776 | 656 | 610 | 1.1x |
GROUP BY x (control) |
61 | 60 | 23 | 23 | 1.0x |
The running max(ts) takes 2.2x longer with 10 partitions than with 1.
Where the time goes
I sampled main (c120619, which has the same window code as 238ae32; release-nonlto build for
symbols) while each query ran repeatedly at 10 partitions. I counted the non-idle samples whose top
frame is a clone or drop of ScalarValue or DataType. That is where the refcount updates are
inlined, though not every such sample is contention.
| query | share of samples (ts / tsn) |
where the per-row scalars come from |
|---|---|---|
running max(ts) |
41% / 4% | PlainAggregateWindowExpr::get_aggregate_result_inside_range (physical-expr/src/window/aggregate.rs) slices the input as the frame advances, and each slice clones the DataType. MaxAccumulator::update_batch (functions-aggregate-common/src/min_max.rs) replaces its scalar and evaluate clones it. The results go through ScalarValue::iter_to_array. |
lag(ts) |
32% | The stateful evaluator in functions-window/src/lead_lag.rs builds each output with ScalarValue::try_from_array, then iter_to_array. (evaluate_all already shifts arrays.) |
row_number() … PARTITION BY ts |
38% | compute_partition_permutation in physical-plan/src/windows/bounded_window_agg_exec.rs builds a Vec<ScalarValue> key per row (get_row_at_idx) to compare with the partition keys (PartitionKey, physical-expr/src/window/window_expr.rs). |
sum(v) … ORDER BY ts |
23% / 8% | The RANGE frame bounds: WindowFrameStateRange::calculate_index_of_row → search_in_slice builds comparison scalars per probe (extract_row_at_idx_to_buf). |
The query that runs the most of these per row, the running aggregate, degrades the most.
Possible fixes
- Fewer per-row scalars on these paths. The running aggregate could build its output without
cloning the accumulator state for every row,lead/lagcould take values by index, and
partition keys andRANGEbounds could be compared on arrays or in row format (arrow::row)
instead of asScalarValues. These are separate, partial changes, but they would save
allocations for every data type, not just timestamps. - One time zone
Arc<str>per execution partition. If each partition cloned the time zone
string into its ownArconce and used it for the arrays and scalars it creates, partitions
would no longer share a refcount, and the time zone and semantics would stay the same. That goes
at the contention itself and would help other operators too, but it needs a point where arrays
enter a partition to swap theDataType. - Dropping the time zone where it can't matter, as perf: avoid Arc<str> refcount contention in grouped first_value/last_value ordered by a time zone aware timestamp #27047 does for the ordering state of
FirstLastGroupsAccumulator: partition keys, say, or the values of built-in functions like
lagandmax. This can't be applied to the window operator as a whole.RANGEframes with
interval offsets add intervals to the timestamp, which depends on the time zone (DST, calendar
days), and user-defined window or aggregate functions may look at it.
Expected behavior
Window queries over tz-aware timestamps scale with partitions like the same queries over naive
timestamps.
Additional context
I found these while comparing engines on a benchmark of my own (event data with UTC timestamps).
Activity