Repository navigation
TPC-H Q20: DataFusion vs Polars comparison #27311
Description
Part of #26169, following #27279 (Q01), #27282 (Q02) and #27310 (Q15). At SF1000, DataFusion was 2.42x slower than Polars on Q20 (results). Locally at SF1 and SF10, DataFusion is faster than both Polars variants, so this issue records what differs and what to check at SF1000.
Setup: the same as #27279. DataFusion main at 14b3edd and Polars main at pola-rs/polars@09251856d4 (streaming engine) run the same benchmarks/queries/q20.sql on the same Parquet files (SF10: part 2M rows, partsupp 8M, lineitem 60M, supplier 100K), on an Apple M5 Max (6 + 12 cores). The polars-benchmark DataFrame version, which the Polars 2.0 benchmark runs, is measured too. All three return the same 1804 rows.
Plans
The plans differ. DataFusion applies both IN subqueries as semi joins below the join with the correlated aggregate. Polars SQL applies them as lists after the joins. The polars-benchmark DataFrame version is written in the same order as DataFusion's plan. In all three, the decorrelated sum(l_quantity) covers every 1994 lineitem row (5.44M groups).
DataFusion logical and physical plan (SF10)
Sort: supplier.s_name ASC NULLS LAST
Projection: supplier.s_name, supplier.s_address
LeftSemi Join: supplier.s_suppkey = __correlated_sq_2.ps_suppkey
Projection: supplier.s_suppkey, supplier.s_name, supplier.s_address
Inner Join: supplier.s_nationkey = nation.n_nationkey
TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey]
Projection: nation.n_nationkey
Filter: nation.n_name = Utf8View("CANADA")
TableScan: nation projection=[n_nationkey, n_name], partial_filters=[nation.n_name = Utf8View("CANADA")]
SubqueryAlias: __correlated_sq_2
Projection: partsupp.ps_suppkey
LeftSemi Join: partsupp.ps_partkey = __scalar_sq_3.l_partkey, partsupp.ps_suppkey = __scalar_sq_3.l_suppkey Filter: CAST(partsupp.ps_availqty AS Float64) > __scalar_sq_3.Float64(0.5) * sum(lineitem.l_quantity)
LeftSemi Join: partsupp.ps_partkey = __correlated_sq_1.p_partkey
TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_availqty]
SubqueryAlias: __correlated_sq_1
Projection: part.p_partkey
Filter: part.p_name LIKE Utf8View("forest%")
TableScan: part projection=[p_partkey, p_name], partial_filters=[part.p_name LIKE Utf8View("forest%")]
SubqueryAlias: __scalar_sq_3
Projection: Float64(0.5) * CAST(sum(lineitem.l_quantity) AS Float64), lineitem.l_partkey, lineitem.l_suppkey
Aggregate: groupBy=[[lineitem.l_partkey, lineitem.l_suppkey]], aggr=[[sum(lineitem.l_quantity)]]
Projection: lineitem.l_partkey, lineitem.l_suppkey, lineitem.l_quantity
Filter: lineitem.l_shipdate >= Date32("1994-01-01") AND lineitem.l_shipdate < Date32("1995-01-01")
TableScan: lineitem projection=[l_partkey, l_suppkey, l_quantity, l_shipdate], partial_filters=[lineitem.l_shipdate >= Date32("1994-01-01"), lineitem.l_shipdate < Date32("1995-01-01")]
SortPreservingMergeExec: [s_name@0 ASC NULLS LAST]
SortExec: expr=[s_name@0 ASC NULLS LAST], preserve_partitioning=[true]
HashJoinExec: mode=CollectLeft, join_type=LeftSemi, on=[(s_suppkey@0, ps_suppkey@0)], projection=[s_name@1, s_address@2]
CoalescePartitionsExec
HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@3)], projection=[s_suppkey@1, s_name@2, s_address@3]
CoalescePartitionsExec
FilterExec: n_name@1 = CANADA, projection=[n_nationkey@0]
RepartitionExec: partitioning=RoundRobinBatch(18), input_partitions=1
DataSourceExec: file_groups={1 groups: [[tpch_sf10/nation/part-0.parquet], ...]}, projection=[n_nationkey, n_name], file_type=parquet, predicate=n_name@1 = CANADA, pruning_predicate=n_name_null_count@2 != row_count@3 AND n_name_min@0 <= CANADA AND CANADA <= n_name_max@1, required_guarantees=[n_name in (CANADA)]
DataSourceExec: file_groups={18 groups: [[tpch_sf10/supplier/part-0.parquet:0..349257], ...]}, projection=[s_suppkey, s_name, s_address, s_nationkey], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible
HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(ps_partkey@0, l_partkey@1), (ps_suppkey@1, l_suppkey@2)], filter=CAST(ps_availqty@0 AS Float64) > Float64(0.5) * sum(lineitem.l_quantity)@1, projection=[ps_suppkey@1]
RepartitionExec: partitioning=Hash([ps_partkey@0, ps_suppkey@1], 18), input_partitions=18
HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(p_partkey@0, ps_partkey@0)]
CoalescePartitionsExec
FilterExec: p_name@1 LIKE forest%, projection=[p_partkey@0]
DataSourceExec: file_groups={18 groups: [[tpch_sf10/part/part-0.parquet:0..2809848], ...]}, projection=[p_partkey, p_name], file_type=parquet, predicate=p_name@1 LIKE forest%, pruning_predicate=p_name_null_count@2 != row_count@3 AND p_name_min@0 <= foresu AND forest <= p_name_max@1, required_guarantees=[]
DataSourceExec: file_groups={18 groups: [[tpch_sf10/partsupp/part-0.parquet:0..17473198], ...]}, projection=[ps_partkey, ps_suppkey, ps_availqty], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible
ProjectionExec: expr=[0.5 * CAST(sum(lineitem.l_quantity)@2 AS Float64) as Float64(0.5) * sum(lineitem.l_quantity), l_partkey@0 as l_partkey, l_suppkey@1 as l_suppkey]
AggregateExec: mode=FinalPartitioned, gby=[l_partkey@0 as l_partkey, l_suppkey@1 as l_suppkey], aggr=[sum(lineitem.l_quantity)]
RepartitionExec: partitioning=Hash([l_partkey@0, l_suppkey@1], 18), input_partitions=18
AggregateExec: mode=Partial, gby=[l_partkey@0 as l_partkey, l_suppkey@1 as l_suppkey], aggr=[sum(lineitem.l_quantity)]
FilterExec: l_shipdate@3 >= 1994-01-01 AND l_shipdate@3 < 1995-01-01, projection=[l_partkey@0, l_suppkey@1, l_quantity@2]
DataSourceExec: file_groups={18 groups: [[tpch_sf10/lineitem/part-0.parquet:0..91990403], ...]}, projection=[l_partkey, l_suppkey, l_quantity, l_shipdate], file_type=parquet, predicate=l_shipdate@10 >= 1994-01-01 AND l_shipdate@10 < 1995-01-01 AND DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, pruning_predicate=l_shipdate_null_count@1 != row_count@2 AND l_shipdate_max@0 >= 1994-01-01 AND l_shipdate_null_count@1 != row_count@2 AND l_shipdate_min@3 < 1995-01-01, required_guarantees=[]
Polars SQL optimized and physical plan (SF10)
SORT BY [nulls_last: [true]] [col("s_name")]
WITH_COLUMNS:
[col("s_name"), col("s_address")]
simple π 2/2 ["s_name", "s_address"]
FILTER when(col("_POLARS_TMP_SQL_14").first().list.length() == 0).then(false).otherwise(col("s_suppkey").is_in([col("_POLARS_TMP_SQL_14").first()]) | (col("_POLARS_TMP_SQL_14").first().list.contains([null.cast(Int64)]) & null.cast(Boolean)))
FROM
HCONCAT
PLAN 0:
simple π 3/3 ["s_name", "s_address", ... 1 other column]
INNER JOIN:
BUILD SIDE: PreferRight
LEFT PLAN ON: [col("s_nationkey")]
Parquet SCAN [tpch_sf10/supplier/part-0.parquet, ... 15 other sources]
PROJECT 4/7 COLUMNS
ESTIMATED ROWS: 100000
RIGHT PLAN ON: [col("n_nationkey")]
simple π 1/1 ["n_nationkey"]
Parquet SCAN [tpch_sf10/nation/part-0.parquet]
PROJECT 2/4 COLUMNS
SELECTION: col("n_name") == "CANADA"
ESTIMATED ROWS: 25
END INNER JOIN
PLAN 1:
SELECT [col("ps_suppkey").implode().alias("_POLARS_TMP_SQL_14")]
simple π 1/1 ["ps_suppkey"]
FILTER when(col("_POLARS_TMP_SQL_12").first().list.length() == 0).then(false).otherwise(col("ps_partkey").is_in([col("_POLARS_TMP_SQL_12").first()]) | (col("_POLARS_TMP_SQL_12").first().list.contains([null.cast(Int64)]) & null.cast(Boolean)))
FROM
HCONCAT
PLAN 0:
simple π 2/2 ["ps_suppkey", "ps_partkey"]
INNER JOIN:
BUILD SIDE: PreferLeft
FUSED PREDICATE: col("ps_availqty").cast(Decimal(38, 0)) > col("__POLARS_CORR__POLARS_TMP_SQL_9_res")
LEFT PLAN ON: [col("ps_partkey"), col("ps_suppkey")]
Parquet SCAN [tpch_sf10/partsupp/part-0.parquet, ... 15 other sources]
PROJECT 3/5 COLUMNS
ESTIMATED ROWS: 8000000
RIGHT PLAN ON: [col("__POLARS_CORR__POLARS_TMP_SQL_9_c_l_partkey"), col("__POLARS_CORR__POLARS_TMP_SQL_9_c_l_suppkey")]
AGGREGATE[maintain_order: false]
[0.5.decimal_mul([col("__POLARS_CORR__POLARS_TMP_SQL_9_c_l_quantity").sum(null_on_empty=true)]).alias("__POLARS_CORR__POLARS_TMP_SQL_9_res")] BY [col("__POLARS_CORR__POLARS_TMP_SQL_9_c_l_partkey"), col("__POLARS_CORR__POLARS_TMP_SQL_9_c_l_suppkey")]
FROM
SELECT [col("l_partkey").alias("__POLARS_CORR__POLARS_TMP_SQL_9_c_l_partkey"), col("l_suppkey").alias("__POLARS_CORR__POLARS_TMP_SQL_9_c_l_suppkey"), col("l_quantity").alias("__POLARS_CORR__POLARS_TMP_SQL_9_c_l_quantity")]
simple π 3/3 ["l_partkey", "l_suppkey", ... 1 other column]
Parquet SCAN [tpch_sf10/lineitem/part-0.parquet, ... 15 other sources]
PROJECT 4/16 COLUMNS
SELECTION: (col("l_shipdate") >= 1994-01-01) & (col("l_shipdate") < 1995-01-01)
ESTIMATED ROWS: 59986052
END INNER JOIN
PLAN 1:
SELECT [col("p_partkey").implode().alias("_POLARS_TMP_SQL_12")]
simple π 1/1 ["p_partkey"]
Parquet SCAN [tpch_sf10/part/part-0.parquet, ... 15 other sources]
PROJECT 2/9 COLUMNS
SELECTION: col("p_name").str.starts_with(["forest"])
ESTIMATED ROWS: 2000000
END HCONCAT
END HCONCAT
Streaming physical plan, from POLARS_VISUALIZE_PHYSICAL_PLAN (_NN stands for _POLARS_TMP_PHYS_NN, _SQL_NN for _POLARS_TMP_SQL_NN and c_l_* for __POLARS_CORR__POLARS_TMP_SQL_25_c_l_*, the table names are added):
in-memory-sink
sort: s_name
with-columns: s_name, s_address
filter: _56, project 2/5: s_name, s_address
select: s_name, s_address, s_suppkey, _SQL_30, _56 = s_suppkey IN _57 (empty list gives false)
zip-broadcast
multiplexer A
multiplexer B
reduce: _57 = col("_SQL_30").first()
zip-broadcast: multiplexer A, multiplexer B
multiplexer A (supplier ⋈ nation)
select: s_name, s_address, s_suppkey
equi-join INNER: s_nationkey = n_nationkey
multi-scan[parquet] (supplier): 16 sources, project: 4, filter: col("s_nationkey").dynamic_predicate()
select: n_nationkey
multi-scan[parquet] (nation): 1 source, project: 2, filter: col("n_name") == "CANADA"
multiplexer B (the list for s_suppkey IN)
select: _SQL_30 = _55
in-memory-map (implode ps_suppkey)
filter: _51, project 1/4: ps_suppkey
select: ps_suppkey, ps_partkey, _SQL_28, _51 = ps_partkey IN _53 (empty list gives false)
zip-broadcast
multiplexer C
multiplexer D
reduce: _52, _53, _54 = col("_SQL_28").first()
zip-broadcast: multiplexer C, multiplexer D
multiplexer C (partsupp ⋈ the decorrelated aggregate)
select: ps_suppkey, ps_partkey
equi-join INNER: [ps_partkey, ps_suppkey] = [c_l_partkey, c_l_suppkey]
multi-scan[parquet] (partsupp): 16 sources, project: 3, filter: col("ps_partkey").dynamic_predicate() & col("ps_suppkey").dynamic_predicate()
select: c_l_partkey = _46, c_l_suppkey = _47, res = 0.5.decimal_mul([_49])
group-by: key=[_46, _47], aggs=[_49 = col("c_l_quantity").sum(null_on_empty=true)]
select: _46 = c_l_partkey, _47 = c_l_suppkey, c_l_quantity
select: c_l_partkey = l_partkey, c_l_suppkey = l_suppkey, c_l_quantity = l_quantity
multi-scan[parquet] (lineitem): 16 sources, project: 4, filter: (col("l_shipdate") >= 1994-01-01) & (col("l_shipdate") < 1995-01-01)
multiplexer D (the list for ps_partkey IN)
select: _SQL_28 = _50
in-memory-map (implode p_partkey)
select: p_partkey
multi-scan[parquet] (part): 16 sources, project: 2, filter: col("p_name").str.starts_with(["forest"])
Polars DataFrame API (polars-benchmark) optimized plan (SF10)
SORT BY [col("s_name")]
simple π 2/2 ["s_name", "s_address"]
INNER JOIN:
LEFT PLAN ON: [col("ps_suppkey")]
SELECT [col("ps_suppkey").unique()]
simple π 1/1 ["ps_suppkey"]
INNER JOIN:
FUSED PREDICATE: col("ps_availqty").cast(Float64) > col("sum_quantity")
LEFT PLAN ON: [col("ps_suppkey"), col("p_partkey")]
simple π 3/3 ["ps_suppkey", "ps_availqty", ... 1 other column]
INNER JOIN:
LEFT PLAN ON: [col("p_partkey")]
SELECT [col("p_partkey").unique()]
simple π 1/1 ["p_partkey"]
Parquet SCAN [tpch_sf10/part/part-0.parquet, ... 15 other sources]
PROJECT 2/9 COLUMNS
SELECTION: col("p_name").str.starts_with(["forest"])
ESTIMATED ROWS: 2000000
RIGHT PLAN ON: [col("ps_partkey")]
Parquet SCAN [tpch_sf10/partsupp/part-0.parquet, ... 15 other sources]
PROJECT 3/5 COLUMNS
ESTIMATED ROWS: 8000000
END INNER JOIN
RIGHT PLAN ON: [col("l_suppkey"), col("l_partkey")]
AGGREGATE[maintain_order: false]
[(col("l_quantity").sum().cast(Float64) * 0.5).alias("sum_quantity")] BY [col("l_partkey"), col("l_suppkey")]
FROM
simple π 3/3 ["l_partkey", "l_suppkey", ... 1 other column]
Parquet SCAN [tpch_sf10/lineitem/part-0.parquet, ... 15 other sources]
PROJECT 4/16 COLUMNS
SELECTION: (col("l_shipdate") >= 1994-01-01) & (col("l_shipdate") < 1995-01-01)
ESTIMATED ROWS: 59986052
END INNER JOIN
RIGHT PLAN ON: [col("s_suppkey")]
simple π 3/3 ["s_name", "s_address", ... 1 other column]
INNER JOIN:
BUILD SIDE: PreferRight
LEFT PLAN ON: [col("s_nationkey")]
Parquet SCAN [tpch_sf10/supplier/part-0.parquet, ... 15 other sources]
PROJECT 4/7 COLUMNS
ESTIMATED ROWS: 100000
RIGHT PLAN ON: [col("n_nationkey")]
simple π 1/1 ["n_nationkey"]
Parquet SCAN [tpch_sf10/nation/part-0.parquet]
PROJECT 2/4 COLUMNS
SELECTION: col("n_name") == "CANADA"
ESTIMATED ROWS: 25
END INNER JOIN
END INNER JOIN
Timings per finding
SF10. Wall times are medians with 18 threads. The CPU rows come from EXPLAIN ANALYZE for DataFusion and POLARS_LOG_METRICS for Polars, with 1 thread. The Polars column is the DataFrame API unless it says SQL.
| DataFusion | Polars | |
|---|---|---|
| 1. Semi joins below the correlated join: DataFusion / Polars SQL / Polars DataFrame API | 126.7 ms | 192.8 / 133.0 ms |
2. Dynamic filter through the aggregate: default / pushdown_filters = true |
126.7 / 108.4 ms | 133.0 ms |
2. Rows into the aggregate: default / pushdown_filters = true |
9.10M / 4.01K | 9.10M |
| 3. CPU of the aggregate over 5.44M groups | 286 ms | 658 ms |
| 3. Peak RSS of one run | 1.0 GB | 1.5 GB (SQL 1.6 GB) |
CPU of the lineitem scan and date filter |
684 ms | 682 ms |
Total timing
Median ms. The ratio uses the Polars DataFrame API, the faster Polars variant.
| DataFusion | Polars DataFrame API | Polars SQL | DataFusion / Polars | |
|---|---|---|---|---|
| SF1, 18 threads | 14.3 | 17.9 | 23.2 | 0.80x |
| SF10, 18 threads | 126.7 | 133.0 | 192.8 | 0.95x |
| SF10, 1 thread | 1224 | 1297 | 1759 | 0.94x |
With 18 threads at SF10, DataFusion uses about 1.55 s of CPU and the Polars DataFrame API 1.92 s.
Findings
- DataFusion's plan is better than Polars SQL's. DataFusion semi-joins
partsupp(8M rows) with the 21.55Kforest%parts to 86.2K rows before the semi join against the aggregate on(ps_partkey, ps_suppkey). Polars SQL joins all 8Mpartsupprows with the 5.44M-group aggregate (938 ms of CPU, 5.43M rows out) and filters to 58.66K rows afterwards. Its key restriction for correlated aggregates (perf: Restrict correlated SQL aggregates to requested keys pola-rs/polars#29383) does not apply here. - The dynamic filter reaches the aggregate input but only prunes row groups by default. The aggregate's group keys are the semi join keys, so the semi join's dynamic filter passes through
AggregateExecto thelineitemscan, and the filters chain (CANADA suppliers, thenpartsupp, thenlineitem). With the defaultpushdown_filters = falsethey prune no row groups, so the aggregate builds 5.44M groups while only 58.66Kpartsupprows survive the semi join. Withpushdown_filters = truethe scan returns 4.01K rows, the aggregate has 2.38K groups and the query takes 108.4 ms. This is the clearest case so far where the row-level filter pays for itself. In Q15 (TPC-H Q15: DataFusion vs Polars comparison #27310) it made the query slower. feat: adaptive placement of Parquet scan filters (row filter, post-scan, skip) #25727 (adaptive placement of Parquet scan filters) is related. perf: push semi joins below inner joins and restrict decorrelated aggregates #26218 (PushDownSemiJoin, which semi joins an aggregate's input with the filtered scan of the outer keys) potentially restricts this aggregate at plan time. - DataFusion's aggregate is cheaper. Over 5.44M groups of two
Int64keys it takes 286 ms against 658 ms for Polars, and a whole run peaks at 1.0 GB against 1.5 GB.
All numbers are from a local machine, not the SF1000 S3 setup. The aggregate grows about 100x from SF10 to SF1000, and the SF1000 run gave DataFusion a 105 GB greedy pool while Polars had no memory limit. The spill_count and spilled_bytes of the AggregateExec nodes in EXPLAIN ANALYZE on that setup would show whether DataFusion spilled there. That is not verified.
Activity