comphead opened a new issue, #27311: URL: https://github.com/apache/datafusion/issues/27311
Part of #26169, following #27279 (Q01), #27282 (Q02) and #27310 (Q15). At SF1000, DataFusion was 2.42x slower than Polars on Q20 ([results](https://github.com/apache/datafusion/issues/26169#issuecomment-6744781906)). 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 14b3edd9a 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). <details><summary>DataFusion logical and physical plan (SF10)</summary> ```text 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=[] ``` </details> <details><summary>Polars SQL optimized and physical plan (SF10)</summary> ```text 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): ```text 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"]) ``` </details> <details><summary>Polars DataFrame API (polars-benchmark) optimized plan (SF10)</summary> ```text 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 ``` </details> ### 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 1. **DataFusion's plan is better than Polars SQL's.** DataFusion semi-joins `partsupp` (8M rows) with the 21.55K `forest%` parts to 86.2K rows before the semi join against the aggregate on `(ps_partkey, ps_suppkey)`. Polars SQL joins all 8M `partsupp` rows 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 (pola-rs/polars#29383) does not apply here. 2. **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 `AggregateExec` to the `lineitem` scan, and the filters chain (CANADA suppliers, then `partsupp`, then `lineitem`). With the default `pushdown_filters = false` they prune no row groups, so the aggregate builds 5.44M groups while only 58.66K `partsupp` rows survive the semi join. With `pushdown_filters = true` the 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 (#27310) it made the query slower. #25727 (adaptive placement of Parquet scan filters) is related. #26218 (`PushDownSemiJoin`, which semi joins an aggregate's input with the filtered scan of the outer keys) potentially restricts this aggregate at plan time. 3. **DataFusion's aggregate is cheaper.** Over 5.44M groups of two `Int64` keys 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. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
