Poor parallel scaling for window queries over timestamps with time zones

Aperta
  • Qualcosa non va?
#27,101 0 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
49/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
rust
Ambito
databases, performance

Direzione di ricerca

Start with the window execution paths named in the report: physical-expr/src/window/aggregate.rs, functions-aggregate-common/src/min_max.rs, functions-window/src/lead_lag.rs, and physical-plan/src/windows/bounded_window_agg_exec.rs. Compare their per-row ScalarValue handling with the timestamp partitioning and frame paths described in the issue. Reproduce the reported partition scaling with the supplied SQL; done means window queries over time-zone-aware timestamps scale more like their naive timestamp equivalents without changing time-zone semantics.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

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/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 ScalarValues. 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).

Lingua principale
Rust
Stelle
9.6k
Fork
2.6k
Merge medio
3g 4h
PR unite (30g)
345

Preparare l'ambiente

Apri in Codespaces

Avvia il container di sviluppo del progetto nel browser, con il tuo account GitHub.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/datafusion

Tutte le issue di apache/datafusion

Issue simili

Altre issue su Rust

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.