TPC-H Q01: DataFusion vs Polars comparison

Aberta
  • Algo errado?
#27,279 0 comentários 1 reação 0 responsáveis Ver no GitHub

Mantenedores costumam responder em até 1 dia

Ninguém assumiu esta issue ainda.

Avaliação

Dificuldade
4/5
Tempo estimado
3-5 dias
Facilidade para iniciantes
35/100
Tipo de issue
Refatoração
Clareza
Razoavelmente clara
Status de atividade
Ativa
Stack de tecnologia
rust
Domínio
backend, performance

Direção de pesquisa

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.

Escrita pelo modelo de indexação a partir do texto da issue.

Descrição

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.

Linguagem predominante
Rust
Estrelas
9.6k
Forks
2.6k
Merge médio
3d 4h
PRs com merge (30d)
345

Preparar o ambiente

Abrir no Codespaces

Inicia o contêiner de desenvolvimento do projeto no navegador, com a sua própria conta do GitHub.

Primeiros passos

  1. Leia a issue inteira e depois o guia de contribuição do projeto.
  2. Comente na issue dizendo que vai assumir — evita que duas pessoas façam o mesmo trabalho.
  3. Faça um fork do repositório e trabalhe em uma branch.
  4. Abra um pull request que referencie o número da issue.

Mais de apache/datafusion

Todas as issues de apache/datafusion

Issues semelhantes

Mais issues de Rust

Receba novas issues na sua caixa de entrada

Um resumo curto de issues do GitHub para quem está começando.