Poor parallel scaling for window queries over timestamps with time zones

オープン
  • 何か問題がありますか？
#27,101 コメント 0 件 リアクション 0 件 担当者 0 名 GitHub で見る

メンテナーはふだん 1 日以内に返信

まだ誰も着手していません。

評価

難易度
4/5
見積もり時間
3〜5日
初心者へのやさしさ
49/100
issue の種類
バグ
明瞭さ
おおむね明確
活発さ
活発
技術スタック
rust
領域
databases, performance

調査の方向性

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.

索引モデルが issue の本文から書いたものです。

説明

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

主要言語
Rust
スター
9.6k
フォーク
2.6k
平均マージ
3日 4時間
マージ済み PR（30日）
345

環境構築

Codespaces で開く

このプロジェクトの開発コンテナを、あなたの GitHub アカウントでブラウザ上に起動します。

はじめの一歩

  1. issue を最後まで読み、次にプロジェクトのコントリビューションガイドを読みます。
  2. 着手することを issue にコメントします — 二人が同じ作業をするのを防げます。
  3. リポジトリをフォークし、ブランチを切って変更します。
  4. issue 番号を参照したプルリクエストを送ります。

apache/datafusion のほかの issue

apache/datafusion の issue をすべて見る

似ている issue

Rust の issue をもっと見る

新しい issue をメールで受け取る

初心者向けの GitHub issue を短くまとめたダイジェスト。