akupchinskiy opened a new issue, #27101:
URL: https://github.com/apache/datafusion/issues/27101
## 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:
```sql
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`/`lag` could take
values by index, and
partition keys and `RANGE` bounds could be compared on arrays or in row
format (`arrow::row`)
instead of as `ScalarValue`s. 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 own `Arc` once 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 the `DataType`.
- **Dropping the time zone where it can't matter**, as #27047 does for the
ordering state of
`FirstLastGroupsAccumulator`: partition keys, say, or the values of
built-in functions like
`lag` and `max`. This can't be applied to the window operator as a whole.
`RANGE` frames 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).
--
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]
