Poor parallel scaling for window queries over timestamps with time zones

Offen
  • Stimmt etwas nicht?
#27,101 0 Kommentare 0 Reaktionen 0 zugewiesene Personen Auf GitHub ansehen

Maintainer antworten meist innerhalb von 1 Tag

Dieses Issue hat noch niemand übernommen.

Bewertung

Schwierigkeit
4/5
Geschätzter Aufwand
3-5 Tage
Anfängerfreundlichkeit
49/100
Issue-Typ
Bug
Klarheit
Größtenteils klar
Aktivitätsstatus
Aktiv
Tech-Stack
rust
Bereich
databases, performance

Rechercherichtung

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.

Vom Indexierungsmodell aus dem Issue-Text verfasst.

Beschreibung

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

Vorherrschende Sprache
Rust
Sterne
9.6k
Forks
2.6k
Ø Merge
3 T. 4 Std.
Gemergte PRs (30 T.)
345

Entwicklungsumgebung

In Codespaces öffnen

Startet den Dev-Container des Projekts im Browser, mit Ihrem eigenen GitHub-Konto.

Erste Schritte

  1. Lesen Sie das ganze Issue und danach den Beitragsleitfaden des Projekts.
  2. Schreiben Sie ins Issue, dass Sie es übernehmen — das erspart doppelte Arbeit.
  3. Forken Sie das Repository und arbeiten Sie in einem Branch.
  4. Öffnen Sie einen Pull Request, der die Issue-Nummer nennt.

Mehr aus apache/datafusion

Alle Issues in apache/datafusion

Ähnliche Issues

Weitere Issues zu Rust

Neue Issues direkt in Ihr Postfach

Eine kurze Übersicht über anfängerfreundliche GitHub-Issues.