Poor parallel scaling for window queries over timestamps with time zones

Ouverte
  • Quelque chose ne va pas ?
#27,101 0 commentaires 0 réactions 0 personnes assignées Voir sur GitHub

Les mainteneurs répondent en général sous 1 jour

Personne n'a encore pris cette issue.

Évaluation

Difficulté
4/5
Temps estimé
3-5 jours
Accessibilité débutants
49/100
Type d'issue
Bug
Clarté
Plutôt claire
Activité
Active
Stack technique
rust
Domaine
databases, performance

Piste de recherche

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.

Rédigé par le modèle d'indexation à partir du texte de l'issue.

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

Langage dominant
Rust
Étoiles
9.6k
Forks
2.6k
Merge moyen
3 j 4 h
PR mergées (30 j)
345

Préparer son environnement

Ouvrir dans Codespaces

Lance le conteneur de développement du projet dans votre navigateur, avec votre propre compte GitHub.

Par où commencer

  1. Lisez l'issue en entier, puis le guide de contribution du projet.
  2. Signalez en commentaire que vous la prenez — cela évite que deux personnes fassent le même travail.
  3. Forkez le dépôt et travaillez sur une branche.
  4. Ouvrez une pull request qui référence le numéro de l'issue.

Autres issues de apache/datafusion

Toutes les issues de apache/datafusion

Issues similaires

Plus d'issues Rust

Recevez les nouvelles issues par e-mail

Un résumé court des issues GitHub adaptées aux débutants.