Repository navigation
TPC-H Q15: DataFusion vs Polars comparison #27310
Description
Part of #26169, following #27279 (Q01) and #27282 (Q02). At SF1000, DataFusion was 5.43x slower than Polars on Q15 (results). This compares the two engines locally at SF1 and SF10.
Setup: the same as #27279. DataFusion main at 14b3edd and Polars main at pola-rs/polars@09251856d4 (streaming engine) run on the same Parquet files (SF10: lineitem 60M rows in 64 row groups, supplier 100K), on an Apple M5 Max (6 + 12 cores). Polars SQL has no CREATE VIEW, so both engines run revenue0 as a CTE. DataFusion plans the CTE and the view in benchmarks/queries/q15.sql identically, and its totals below come from dfbench, which runs the view. Polars SQL and the polars-benchmark DataFrame version take the same time (55.8 and 54.3 ms at SF10).
Query text used for both engines
with revenue0 (supplier_no, total_revenue) as (
select
l_suppkey,
sum(l_extendedprice * (1 - l_discount))
from
lineitem
where
l_shipdate >= date '1996-01-01'
and l_shipdate < date '1996-01-01' + interval '3' month
group by
l_suppkey
)
select
s_suppkey,
s_name,
s_address,
s_phone,
total_revenue
from
supplier,
revenue0
where
s_suppkey = supplier_no
and total_revenue = (
select
max(total_revenue)
from
revenue0
)
order by
s_suppkey;Plans
There is one join, supplier against revenue0, built on supplier in both engines. The plans differ in how often revenue0 is computed.
DataFusion logical and physical plan (SF10)
Sort: supplier.s_suppkey ASC NULLS LAST
Projection: supplier.s_suppkey, supplier.s_name, supplier.s_address, supplier.s_phone, revenue0.total_revenue
Inner Join: supplier.s_suppkey = revenue0.supplier_no
TableScan: supplier projection=[s_suppkey, s_name, s_address, s_phone]
SubqueryAlias: revenue0
Projection: lineitem.l_suppkey AS supplier_no, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount) AS total_revenue
Filter: sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount) = (<subquery>)
Subquery:
Aggregate: groupBy=[[]], aggr=[[max(revenue0.total_revenue)]]
SubqueryAlias: revenue0
Projection: sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount) AS total_revenue
Aggregate: groupBy=[[lineitem.l_suppkey]], aggr=[[sum(lineitem.l_extendedprice * (Decimal128(1,20,0) - lineitem.l_discount)) AS sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]]
Projection: lineitem.l_suppkey, lineitem.l_extendedprice, lineitem.l_discount
Filter: lineitem.l_shipdate >= Date32("1996-01-01") AND lineitem.l_shipdate < Date32("1996-04-01")
TableScan: lineitem projection=[l_suppkey, l_extendedprice, l_discount, l_shipdate], partial_filters=[lineitem.l_shipdate >= Date32("1996-01-01"), lineitem.l_shipdate < Date32("1996-04-01")]
Aggregate: groupBy=[[lineitem.l_suppkey]], aggr=[[sum(lineitem.l_extendedprice * (Decimal128(1,20,0) - lineitem.l_discount)) AS sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]]
Projection: lineitem.l_suppkey, lineitem.l_extendedprice, lineitem.l_discount
Filter: lineitem.l_shipdate >= Date32("1996-01-01") AND lineitem.l_shipdate < Date32("1996-04-01")
TableScan: lineitem projection=[l_suppkey, l_extendedprice, l_discount, l_shipdate], partial_filters=[lineitem.l_shipdate >= Date32("1996-01-01"), lineitem.l_shipdate < Date32("1996-04-01")]
ScalarSubqueryExec: subqueries=1
SortPreservingMergeExec: [s_suppkey@0 ASC NULLS LAST]
SortExec: expr=[s_suppkey@0 ASC NULLS LAST], preserve_partitioning=[true]
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(s_suppkey@0, supplier_no@0)], projection=[s_suppkey@0, s_name@1, s_address@2, s_phone@3, total_revenue@5]
CoalescePartitionsExec
DataSourceExec: file_groups={18 groups: [[tpch_sf10/supplier/part-0.parquet:0..349257], ...]}, projection=[s_suppkey, s_name, s_address, s_phone], file_type=parquet
ProjectionExec: expr=[l_suppkey@0 as supplier_no, sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@1 as total_revenue]
FilterExec: sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@1 = scalar_subquery(<pending>)
AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as l_suppkey], aggr=[sum(lineitem.l_extendedprice * 1 - lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
RepartitionExec: partitioning=Hash([l_suppkey@0], 18), input_partitions=18
AggregateExec: mode=Partial, gby=[l_suppkey@0 as l_suppkey], aggr=[sum(lineitem.l_extendedprice * 1 - lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
FilterExec: l_shipdate@3 >= 1996-01-01 AND l_shipdate@3 < 1996-04-01, projection=[l_suppkey@0, l_extendedprice@1, l_discount@2]
DataSourceExec: file_groups={18 groups: [[tpch_sf10/lineitem/part-0.parquet:0..91990403], ...]}, projection=[l_suppkey, l_extendedprice, l_discount, l_shipdate], file_type=parquet, predicate=l_shipdate@10 >= 1996-01-01 AND l_shipdate@10 < 1996-04-01 AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=l_shipdate_null_count@1 != row_count@2 AND l_shipdate_max@0 >= 1996-01-01 AND l_shipdate_null_count@1 != row_count@2 AND l_shipdate_min@3 < 1996-04-01, required_guarantees=[]
AggregateExec: mode=Final, gby=[], aggr=[max(revenue0.total_revenue)]
CoalescePartitionsExec
AggregateExec: mode=Partial, gby=[], aggr=[max(revenue0.total_revenue)]
ProjectionExec: expr=[sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)@1 as total_revenue]
AggregateExec: mode=FinalPartitioned, gby=[l_suppkey@0 as l_suppkey], aggr=[sum(lineitem.l_extendedprice * 1 - lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
RepartitionExec: partitioning=Hash([l_suppkey@0], 18), input_partitions=18
AggregateExec: mode=Partial, gby=[l_suppkey@0 as l_suppkey], aggr=[sum(lineitem.l_extendedprice * 1 - lineitem.l_discount) as sum(lineitem.l_extendedprice * Int64(1) - lineitem.l_discount)]
FilterExec: l_shipdate@3 >= 1996-01-01 AND l_shipdate@3 < 1996-04-01, projection=[l_suppkey@0, l_extendedprice@1, l_discount@2]
DataSourceExec: file_groups={18 groups: [[tpch_sf10/lineitem/part-0.parquet:0..91990403], ...]}, projection=[l_suppkey, l_extendedprice, l_discount, l_shipdate], file_type=parquet, predicate=l_shipdate@10 >= 1996-01-01 AND l_shipdate@10 < 1996-04-01, pruning_predicate=l_shipdate_null_count@1 != row_count@2 AND l_shipdate_max@0 >= 1996-01-01 AND l_shipdate_null_count@1 != row_count@2 AND l_shipdate_min@3 < 1996-04-01, required_guarantees=[]
Polars optimized and physical plan (SF10)
SORT BY [nulls_last: [true]] [col("s_suppkey")]
WITH_COLUMNS:
[col("s_suppkey"), col("s_name"), col("s_address"), col("s_phone"), col("total_revenue")]
simple π 5/5 ["s_suppkey", "s_name", ... 3 other columns]
FILTER col("total_revenue") == col("_POLARS_TMP_SQL_6").first()
FROM
HCONCAT
PLAN 0:
INNER JOIN:
BUILD SIDE: PreferLeft
LEFT PLAN ON: [col("s_suppkey")]
Parquet SCAN [tpch_sf10/supplier/part-0.parquet, ... 15 other sources]
PROJECT 4/7 COLUMNS
ESTIMATED ROWS: 100000
RIGHT PLAN ON: [col("supplier_no")]
CACHE[id: 9d868e2f-b2f0-4b34-b82f-d47ad3cd1caf]
SELECT [col("l_suppkey").alias("supplier_no"), col("l_extendedprice").alias("total_revenue")]
AGGREGATE[maintain_order: false]
[col("l_extendedprice").decimal_mul([(1 - col("l_discount"))]).sum(null_on_empty=true)] BY [col("l_suppkey")]
FROM
simple π 3/3 ["l_suppkey", "l_extendedprice", ... 1 other column]
Parquet SCAN [tpch_sf10/lineitem/part-0.parquet, ... 15 other sources]
PROJECT 4/16 COLUMNS
SELECTION: (col("l_shipdate") >= 1996-01-01) & (col("l_shipdate") < 1996-04-01)
ESTIMATED ROWS: 59986052
END INNER JOIN
PLAN 1:
SELECT [col("total_revenue").item(allow_empty=true).alias("_POLARS_TMP_SQL_6")]
SELECT [col("total_revenue").max()]
simple π 1/1 ["total_revenue"]
CACHE[id: 9d868e2f-b2f0-4b34-b82f-d47ad3cd1caf]
END HCONCAT
Streaming physical plan, from POLARS_VISUALIZE_PHYSICAL_PLAN (_NN stands for _POLARS_TMP_PHYS_NN and _SQL_14 for _POLARS_TMP_SQL_14, the table names are added):
in-memory-sink
sort: s_suppkey
with-columns: s_suppkey, s_name, s_address, s_phone, total_revenue
filter: _34, project 5/7: s_suppkey, s_name, s_address, s_phone, total_revenue
select: s_suppkey, s_name, s_address, s_phone, total_revenue, _SQL_14, _34 = col("total_revenue") == col("_35")
zip-broadcast
multiplexer J
multiplexer S
reduce: _35 = col("_SQL_14").first()
zip-broadcast
multiplexer J
multiplexer S
multiplexer J (the join)
equi-join INNER: s_suppkey = supplier_no
multi-scan[parquet] (supplier): 16 sources, project: 4, filter: col("s_suppkey").dynamic_predicate()
multiplexer R
multiplexer S (the max subquery)
select: _SQL_14 = _33
reduce: _33 = col("total_revenue").item(allow_empty=true)
select: total_revenue = _32
reduce: _32 = col("total_revenue").max()
select: total_revenue
multiplexer R
multiplexer R (revenue0, computed once)
select: supplier_no = l_suppkey, total_revenue = l_extendedprice
select: l_suppkey = _28, l_extendedprice = _30
group-by: key=[_28], aggs=[_30 = col("_31").sum(null_on_empty=true)]
select: _28 = col("l_suppkey"), _31 = col("l_extendedprice").decimal_mul([(1 - col("l_discount"))])
select: l_suppkey, l_extendedprice, l_discount
multi-scan[parquet] (lineitem): 16 sources, project: 4, filter: (col("l_shipdate") >= 1996-01-01) & (col("l_shipdate") < 1996-04-01)
Timings per finding
SF10. The wall times in findings 1 and 2 come from datafusion-cli for both DataFusion columns. The CPU rows come from EXPLAIN ANALYZE for DataFusion and POLARS_LOG_METRICS for Polars, with 1 thread, for a single pass over lineitem.
| DataFusion | Polars | |
|---|---|---|
1. revenue0 computed twice: CTE / MATERIALIZED CTE, 18 threads |
141.5 / 71.5 ms | 55.8 ms |
| 1. The same, 1 thread | 1281 / 639 ms | 574 ms |
2. Scan work granularity: MATERIALIZED CTE with default / target_partitions = 64 |
71.5 / 62.0 ms | 55.8 ms |
3. CPU of the scan and date filter: default / pushdown_filters = true |
568 / 626 ms | 524 ms |
| CPU of the aggregate | 25 ms | 23 ms |
Total timing
Median ms.
| DataFusion | Polars | DataFusion / Polars | |
|---|---|---|---|
| SF1, 18 threads | 14.4 | 9.6 | 1.50x |
| SF10, 18 threads | 143.6 | 55.8 | 2.57x |
| SF10, 1 thread | 1271 | 574 | 2.21x |
With 18 threads at SF10, DataFusion uses about 1.59 s of CPU and Polars 0.68 s.
Findings
revenue0is computed twice, one pass after the other. DataFusion inlines the view into both the FROM and themaxsubquery, solineitemis scanned (60M rows, 450 MB) and aggregated twice.ScalarSubqueryExecruns the subquery to completion before it starts the main plan: the per-partitionBaselineMetricsshow the first pass from 0 to 76 ms and the second from 76 to 148 ms. Polars computesrevenue0once (CACHE), and both the join and themaxread it. WithWITH revenue0 AS MATERIALIZED (...)(PostgreSQL dialect), DataFusion returns the same row in 71.5 ms instead of 141.5 ms. Reusing repeated CTEs and subplans is Avoid recompute CTEs (common table expressions) / share input plans #8777.- Scan work granularity, as in TPC-H Q01: DataFusion vs Polars comparison #27279. In the first pass, the scan partitions finish between 43 and 76 ms. With the materialized CTE,
target_partitions = 64gives 62.0 ms. - The date filter selects 3.78% of the rows. Polars decodes
l_shipdatefirst and the other three columns only for the rows that match. DataFusion decodes all four columns and then runsFilterExec.pushdown_filters = truereturns the same 2.27M rows from the scan but costs more (626 ms, and 159 vs 144 ms at 18 threads), althoughrow_pushdown_eval_timeis only about 11 ms. The pushdown cost is tracked in [EPIC] Fix performance regressions when enabling parquet filter pushdown (late materialization) #20324.
All numbers are from a local machine, not the SF1000 S3 setup. With findings 1 and 2 applied, DataFusion is at 62.0 ms against 55.8 ms for Polars.
Activity