TPC-H Q01: DataFusion vs Polars comparison

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

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

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

評価

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

調査の方向性

Start with the benchmark query benchmarks/queries/q1.sql and the reported timings and plans; this issue is part of #26169. The findings point to SharedWorkSource and the group accumulator types PrimitiveGroupsAccumulator, AvgGroupsAccumulator, and CountGroupsAccumulator, while related work is tracked in #21351, #21719, #21766, #23285, and #18411. Compare a focused change against the listed SF1 and SF10 timings; the issue does not specify one selected finding or a completion criterion.

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

説明

enhancement

Part of #26169, following this plan. At SF1000, DataFusion was 1.74x slower than Polars on Q01 (results). This compares the two engines locally at SF1 and SF10.

Setup: DataFusion main at 14b3edd9a (dfbench, fat LTO) and Polars main at pola-rs/polars@09251856d4 (streaming engine, release build features). Both run the same benchmarks/queries/q1.sql on the same Parquet files (SF10: 16 files, 64 row groups), on an Apple M5 Max (6 + 12 cores). Runs are warm, with the engines interleaved.

Plans

The plans are the same. Q01 has no joins. Both engines read 7 of 16 columns and compute l_extendedprice * (1 - l_discount) once. Statistics prune nothing. Polars applies the filter inside the scan, while DataFusion uses FilterExec (pushdown_filters = true does not change the timing).

DataFusion logical and physical plan (SF10) 
Sort: lineitem.l_returnflag ASC NULLS LAST, lineitem.l_linestatus ASC NULLS LAST
  Projection: lineitem.l_returnflag, lineitem.l_linestatus, sum(lineitem.l_quantity) AS sum_qty, sum(lineitem.l_extendedprice) AS sum_base_price, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount) AS sum_disc_price, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) + lineitem.l_tax) AS sum_charge, avg(lineitem.l_quantity) AS avg_qty, avg(lineitem.l_extendedprice) AS avg_price, avg(lineitem.l_discount) AS avg_disc, count(Int64(1)) AS count(*) AS count_order
    Aggregate: groupBy=[[lineitem.l_returnflag, lineitem.l_linestatus]], aggr=[[sum(lineitem.l_quantity), sum(lineitem.l_extendedprice), sum(__common_expr_1) AS sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount), sum(__common_expr_1 * (Decimal128(1,20,0) + lineitem.l_tax)) AS sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) + lineitem.l_tax), avg(lineitem.l_quantity), avg(lineitem.l_extendedprice), avg(lineitem.l_discount), count(Int64(1))]]
      Projection: lineitem.l_extendedprice * (Decimal128(1,20,0) - lineitem.l_discount) AS __common_expr_1, lineitem.l_quantity, lineitem.l_extendedprice, lineitem.l_discount, lineitem.l_tax, lineitem.l_returnflag, lineitem.l_linestatus
        Filter: lineitem.l_shipdate <= Date32("1998-09-02")
          TableScan: lineitem projection=[l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate], partial_filters=[lineitem.l_shipdate <= Date32("1998-09-02")]

SortPreservingMergeExec: [l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC NULLS LAST]
  ProjectionExec: expr=[l_returnflag@0 as l_returnflag, l_linestatus@1 as l_linestatus, sum(lineitem.l_quantity)@2 as sum_qty, sum(lineitem.l_extendedprice)@3 as sum_base_price, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@4 as sum_disc_price, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) + lineitem.l_tax)@5 as sum_charge, avg(lineitem.l_quantity)@6 as avg_qty, avg(lineitem.l_extendedprice)@7 as avg_price, avg(lineitem.l_discount)@8 as avg_disc, count(Int64(1))@9 as count_order]
    SortExec: expr=[l_returnflag@0 ASC NULLS LAST, l_linestatus@1 ASC NULLS LAST], preserve_partitioning=[true]
      AggregateExec: mode=FinalPartitioned, gby=[l_returnflag@0 as l_returnflag, l_linestatus@1 as l_linestatus], aggr=[sum(lineitem.l_quantity), sum(lineitem.l_extendedprice), sum(__common_expr_1) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount), sum(__common_expr_1 * 1 + lineitem.l_tax) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) + lineitem.l_tax), avg(lineitem.l_quantity), avg(lineitem.l_extendedprice), avg(lineitem.l_discount), count(Int64(1))]
        RepartitionExec: partitioning=Hash([l_returnflag@0, l_linestatus@1], 18), input_partitions=18
          AggregateExec: mode=Partial, gby=[l_returnflag@5 as l_returnflag, l_linestatus@6 as l_linestatus], aggr=[sum(lineitem.l_quantity), sum(lineitem.l_extendedprice), sum(__common_expr_1) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount), sum(__common_expr_1 * 1 + lineitem.l_tax) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount * Int64(1) + lineitem.l_tax), avg(lineitem.l_quantity), avg(lineitem.l_extendedprice), avg(lineitem.l_discount), count(Int64(1))]
            ProjectionExec: expr=[l_extendedprice@0 * (1 - l_discount@1) as __common_expr_1, l_quantity@2 as l_quantity, l_extendedprice@0 as l_extendedprice, l_discount@1 as l_discount, l_tax@3 as l_tax, l_returnflag@4 as l_returnflag, l_linestatus@5 as l_linestatus]
              FilterExec: l_shipdate@6 <= 1998-09-02, projection=[l_extendedprice@1, l_discount@2, l_quantity@0, l_tax@3, l_returnflag@4, l_linestatus@5]
                DataSourceExec: file_groups={18 groups: [[tpch_sf10/lineitem/part-0.parquet:0..91990403], [tpch_sf10/lineitem/part-0.parquet:91990403..104314377, tpch_sf10/lineitem/part-1.parquet:0..79666429], ...]}, projection=[l_quantity, l_extendedprice, l_discount, l_tax, l_returnflag, l_linestatus, l_shipdate], file_type=parquet, predicate=l_shipdate@10 <= 1998-09-02, pruning_predicate=l_shipdate_null_count@1 != row_count@2 AND l_shipdate_min@0 <= 1998-09-02, required_guarantees=[]
Polars optimized and physical plan (SF10) 
SORT BY [nulls_last: [true, true]] [col("l_returnflag"), col("l_linestatus")]
  AGGREGATE[maintain_order: false]
    [col("l_quantity").sum(null_on_empty=true).alias("sum_qty"), col("l_extendedprice").sum(null_on_empty=true).alias("sum_base_price"), col("__POLARS_CSER_0x100000005").sum(null_on_empty=true).alias("sum_disc_price"), col("__POLARS_CSER_0x100000005").alias("l_extendedprice").decimal_mul([(1 + col("l_tax"))]).sum(null_on_empty=true).alias("sum_charge"), col("l_quantity").mean().alias("avg_qty"), col("l_extendedprice").mean().alias("avg_price"), col("l_discount").mean().alias("avg_disc"), len().cast(Int64).alias("count_order")] BY [col("l_returnflag"), col("l_linestatus")]
    FROM
     WITH_COLUMNS:
     [col("l_extendedprice").decimal_mul([(1 - col("l_discount"))]).alias("__POLARS_CSER_0x100000005")]
      simple π 6/6 ["l_returnflag", "l_linestatus", ... 4 other columns]
        Parquet SCAN [tpch_sf10/lineitem/part-0.parquet, ... 15 other sources]
        PROJECT 7/16 COLUMNS
        SELECTION: col("l_shipdate") <= 1998-09-02
        ESTIMATED ROWS: 59986052

Streaming physical plan, from POLARS_VISUALIZE_PHYSICAL_PLAN (_NN stands for _POLARS_TMP_PHYS_NN):

in-memory-sink
  select: l_returnflag, l_linestatus, sum_qty, sum_base_price, sum_disc_price, sum_charge, avg_qty, avg_price, avg_disc, count_order
    sort: _50
      with-columns: _50 = col("l_returnflag").row_encode([col("l_linestatus")])
        select: l_returnflag = _35, l_linestatus = _36, sum_qty = _38, ..., count_order = _49.cast(Int64)
          group-by: key=[_35, _36]
                    fused agg inputs: _41 = col("l_extendedprice").decimal_mul([(1 - col("l_discount"))]), _43 = col("_41").decimal_mul([(1 + col("l_tax"))])
                    aggs: sum(l_quantity), sum(l_extendedprice), sum(_41), sum(_43), mean(l_quantity), mean(l_extendedprice), mean(l_discount), len()
            select: _35 = l_returnflag, _36 = l_linestatus, l_quantity, l_extendedprice, l_discount, l_tax
              select: l_returnflag, l_linestatus, l_quantity, l_extendedprice, l_discount, l_tax
                multi-scan[parquet]: 16 sources, project: 7 total, 7 from file, filter: col("l_shipdate") <= 1998-09-02
Timings per finding

SF10. Finding 1 is wall time with 18 threads. The other rows are CPU time per query with 1 thread, from EXPLAIN ANALYZE for DataFusion and POLARS_LOG_METRICS for Polars. The split inside the Polars group-by node (marked ~) comes from a sampling profile.

DataFusion Polars
1. Scan work granularity: default / target_partitions = 64 237 / 202 ms 211 ms
2. Group accumulators (8 aggregates) 681 ms ~330 ms
3. Decimal +, -, * 433 ms ~220 ms
Group keys 241 ms ~300 ms
Parquet scan and filter, including ZSTD 765 ms 1210 ms
Total timing

Median ms.

DataFusion Polars DataFusion / Polars
SF1, 18 threads 22.6 20.5 1.10x
SF10, 18 threads 237 211 1.12x
SF10, 1 thread 2160 2060 1.05x

With 18 threads DataFusion reports less CPU than Polars (2.7 s vs 3.3 s, as each engine measures it). So the multi-threaded gap is mostly how the work is spread.

Findings
  1. Scan work granularity. SharedWorkSource (#21351) shares the 33 byte-range file splits, not row groups. The per-partition BaselineMetrics (start_timestamp, end_timestamp, output_rows) show partitions reading 2.69M to 4.79M rows, and partial aggregates finishing between 125 and 211 ms of a 211 ms query. With target_partitions = 64 each split holds about one row group. Polars hands out morsels of about 105K rows. Row-group sharing was prototyped in #21766 and #23285, and #21719 tracks a morsel scheduler.
  2. Group accumulators. Q01 has 4 groups, and 64% of consecutive rows fall in the same group as the previous row. So values[g] += v in PrimitiveGroupsAccumulator, AvgGroupsAccumulator and CountGroupsAccumulator waits on the previous store. Polars spreads a few groups over 64 slots (pola-rs/polars#29613). An i128 sum over Q01's group sequence takes 1.55 ns/row with one slot per group and 0.43 ns/row with 16 lanes.
  3. Decimal +, -, *. Every element goes through checked i128 arithmetic. As in apache/arrow-rs#11529, try_unary calls the decimal_op closure out of line for every value of 1 - l_discount and 1 + l_tax. apache/arrow-rs#12062 replaces these kernels and multiplies in i64 when the values fit.

Group keys (see #18411) are not a gap against Polars here. Polars' Parquet scan costs more than DataFusion's, which offsets findings 2 and 3 in the single-threaded total. All numbers are from a local machine, not the SF1000 S3 setup.

主要言語
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 を短くまとめたダイジェスト。