TPC-H Q01: DataFusion vs Polars comparison

Abierto
  • ¿Algo no está bien?
#27,279 0 comentarios 1 reacción 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
35/100
Tipo de issue
Refactorización
Claridad
Bastante claro
Estado de actividad
Activo
Stack tecnológico
rust
Área
backend, performance

Línea de trabajo

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.

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

Descripción

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.

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.