TPC-H Q01: DataFusion vs Polars comparison

Đang mở
  • Có gì không ổn?
#27,279 0 bình luận 1 reaction 0 người được giao Xem trên GitHub

Maintainer thường phản hồi trong vòng 1 ngày

Chưa có ai nhận issue này.

Đánh giá

Độ khó
4/5
Thời gian dự kiến
3-5 ngày
Mức phù hợp với người mới
35/100
Loại issue
Tái cấu trúc
Độ rõ ràng
Khá rõ ràng
Mức độ hoạt động
Sôi nổi
Công nghệ
rust
Lĩnh vực
backend, performance

Hướng nghiên cứu

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.

Do mô hình lập chỉ mục viết ra từ nội dung của issue.

Mô tả

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.

Ngôn ngữ chính
Rust
Star
9.6k
Fork
2.6k
Merge trung bình
3 ngày 4 giờ
Pull request đã merge (30 ngày)
345

Chuẩn bị môi trường

Mở trong Codespaces

Khởi chạy dev container của dự án ngay trên trình duyệt, bằng tài khoản GitHub của bạn.

Bắt đầu từ đâu

  1. Đọc hết issue, rồi đọc hướng dẫn đóng góp của dự án.
  2. Bình luận trên issue rằng bạn sẽ nhận — tránh hai người làm cùng một việc.
  3. Fork repository và làm thay đổi trên một nhánh.
  4. Mở pull request có tham chiếu số hiệu của issue.

Issue khác của apache/datafusion

Tất cả issue của apache/datafusion

Issue tương tự

Thêm issue về Rust

Nhận issue mới trong hộp thư của bạn

Bản tóm tắt ngắn những issue GitHub phù hợp với người mới.