Poor parallel scaling for window queries over timestamps with time zones

Abierto
  • ¿Algo no está bien?
#27,101 0 comentarios 0 reacciones 0 asignados Ver en GitHub

Los mantenedores suelen responder en 1 día

Nadie ha tomado este issue todavía.

Evaluación

Dificultad
4/5
Tiempo estimado
3-5 días
Aptitud para principiantes
49/100
Tipo de issue
Error
Claridad
Bastante claro
Estado de actividad
Activo
Stack tecnológico
rust
Área
databases, performance

Línea de trabajo

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.

Escrito por el modelo de indexación a partir del texto del issue.

Descripción

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

Lenguaje dominante
Rust
Estrellas
9.6k
Forks
2.6k
Merge medio
3 d 4 h
PR fusionados (30 d)
345

Preparar el entorno

Abrir en Codespaces

Inicia el contenedor de desarrollo del proyecto en tu navegador, con tu propia cuenta de GitHub.

Primeros pasos

  1. Lee el issue completo y luego la guía de contribución del proyecto.
  2. Comenta en el issue que vas a ocuparte — evita que dos personas hagan lo mismo.
  3. Haz un fork del repositorio y trabaja en una rama.
  4. Abre un pull request que haga referencia al número del issue.

Más de apache/datafusion

Todos los issues de apache/datafusion

Issues similares

Más issues de Rust

Recibe los nuevos issues en tu correo

Un resumen breve de issues de GitHub para principiantes.