Repository navigation
TPC-H Q01: DataFusion vs Polars comparison #27279
Description
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 14b3edd (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
- Scan work granularity.
SharedWorkSource(Dynamic work scheduling in FileStream #21351) shares the 33 byte-range file splits, not row groups. The per-partitionBaselineMetrics(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. Withtarget_partitions = 64each split holds about one row group. Polars hands out morsels of about 105K rows. Row-group sharing was prototyped in feat(parquet): row-group morselization for sibling FileStream stealing #21766 and perf: split row groups by file range (morsel splitting) #23285, and Create a pipeline/morsel scheduler #21719 tracks a morsel scheduler. - Group accumulators. Q01 has 4 groups, and 64% of consecutive rows fall in the same group as the previous row. So
values[g] += vinPrimitiveGroupsAccumulator,AvgGroupsAccumulatorandCountGroupsAccumulatorwaits on the previous store. Polars spreads a few groups over 64 slots (perf: Add eviction-free path to GroupedReduction, with lanes pola-rs/polars#29613). Ani128sum over Q01's group sequence takes 1.55 ns/row with one slot per group and 0.43 ns/row with 16 lanes. - Decimal
+,-,*. Every element goes through checkedi128arithmetic. As intry_unarycalls thedecimal_opclosure out of line for every value arrow-rs#11529,try_unarycalls thedecimal_opclosure out of line for every value of1 - l_discountand1 + l_tax. perf(arrow-arith): vectorize checked decimaladd,subandmularrow-rs#12062 replaces these kernels and multiplies ini64when 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.
Reacted by Tobias Schwarzinger
Activity