comphead opened a new issue, #27282: URL: https://github.com/apache/datafusion/issues/27282
Part of #26169, following the Q01 comparison in #27279. At SF1000, DataFusion was 5.42x slower than Polars on Q02 ([results](https://github.com/apache/datafusion/issues/26169#issuecomment-6744781906)). This compares the two engines locally at SF1 and SF10. **Setup:** the same as #27279. DataFusion `main` at 14b3edd9a and Polars `main` at pola-rs/polars@09251856d4 (streaming engine) run the same `benchmarks/queries/q2.sql` on the same Parquet files (SF10: `part` 2M rows, `partsupp` 8M, `supplier` 100K, one row group per file), on an Apple M5 Max (6 + 12 cores). Polars SQL and the polars-benchmark DataFrame version produce the same plan shape and take the same time (16.8 and 16.4 ms at SF10). ### Plans The join order is the same in both engines: `part` (filtered), `partsupp`, `supplier`, `nation`, `region` (filtered), which is the FROM order. The projections and scan filters are the same too. The plans differ in how the correlated subquery is decorrelated. <details><summary>DataFusion logical and physical plan (SF10)</summary> ```text Sort: supplier.s_acctbal DESC NULLS FIRST, nation.n_name ASC NULLS LAST, supplier.s_name ASC NULLS LAST, part.p_partkey ASC NULLS LAST, fetch=100 Projection: supplier.s_acctbal, supplier.s_name, nation.n_name, part.p_partkey, part.p_mfgr, supplier.s_address, supplier.s_phone, supplier.s_comment LeftSemi Join: part.p_partkey = __scalar_sq_1.ps_partkey, partsupp.ps_supplycost = __scalar_sq_1.min(partsupp.ps_supplycost) Projection: part.p_partkey, part.p_mfgr, supplier.s_name, supplier.s_address, supplier.s_phone, supplier.s_acctbal, supplier.s_comment, partsupp.ps_supplycost, nation.n_name Inner Join: nation.n_regionkey = region.r_regionkey Projection: part.p_partkey, part.p_mfgr, supplier.s_name, supplier.s_address, supplier.s_phone, supplier.s_acctbal, supplier.s_comment, partsupp.ps_supplycost, nation.n_name, nation.n_regionkey Inner Join: supplier.s_nationkey = nation.n_nationkey Projection: part.p_partkey, part.p_mfgr, supplier.s_name, supplier.s_address, supplier.s_nationkey, supplier.s_phone, supplier.s_acctbal, supplier.s_comment, partsupp.ps_supplycost Inner Join: partsupp.ps_suppkey = supplier.s_suppkey Projection: part.p_partkey, part.p_mfgr, partsupp.ps_suppkey, partsupp.ps_supplycost Inner Join: part.p_partkey = partsupp.ps_partkey Projection: part.p_partkey, part.p_mfgr Filter: part.p_size = Int32(15) AND part.p_type LIKE Utf8View("%BRASS") TableScan: part projection=[p_partkey, p_mfgr, p_type, p_size], partial_filters=[part.p_size = Int32(15), part.p_type LIKE Utf8View("%BRASS")] TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost] TableScan: supplier projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment] TableScan: nation projection=[n_nationkey, n_name, n_regionkey] Projection: region.r_regionkey Filter: region.r_name = Utf8View("EUROPE") TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8View("EUROPE")] SubqueryAlias: __scalar_sq_1 Projection: min(partsupp.ps_supplycost), partsupp.ps_partkey Aggregate: groupBy=[[partsupp.ps_partkey]], aggr=[[min(partsupp.ps_supplycost)]] Projection: partsupp.ps_partkey, partsupp.ps_supplycost LeftSemi Join: nation.n_regionkey = region.r_regionkey Projection: partsupp.ps_partkey, partsupp.ps_supplycost, nation.n_regionkey Inner Join: supplier.s_nationkey = nation.n_nationkey Projection: partsupp.ps_partkey, partsupp.ps_supplycost, supplier.s_nationkey Inner Join: partsupp.ps_suppkey = supplier.s_suppkey TableScan: partsupp projection=[ps_partkey, ps_suppkey, ps_supplycost] TableScan: supplier projection=[s_suppkey, s_nationkey] TableScan: nation projection=[n_nationkey, n_regionkey] Projection: region.r_regionkey Filter: region.r_name = Utf8View("EUROPE") TableScan: region projection=[r_regionkey, r_name], partial_filters=[region.r_name = Utf8View("EUROPE")] SortPreservingMergeExec: [s_acctbal@0 DESC, n_name@2 ASC NULLS LAST, s_name@1 ASC NULLS LAST, p_partkey@3 ASC NULLS LAST], fetch=100 SortExec: TopK(fetch=100), expr=[s_acctbal@0 DESC, n_name@2 ASC NULLS LAST, s_name@1 ASC NULLS LAST, p_partkey@3 ASC NULLS LAST], preserve_partitioning=[true] HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(p_partkey@0, ps_partkey@1), (ps_supplycost@7, min(partsupp.ps_supplycost)@0)], projection=[s_acctbal@5, s_name@2, n_name@8, p_partkey@0, p_mfgr@1, s_address@3, s_phone@4, s_comment@6] RepartitionExec: partitioning=Hash([p_partkey@0, ps_supplycost@7], 18), input_partitions=18 HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(r_regionkey@0, n_regionkey@9)], projection=[p_partkey@1, p_mfgr@2, s_name@3, s_address@4, s_phone@5, s_acctbal@6, s_comment@7, ps_supplycost@8, n_name@9] CoalescePartitionsExec FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0] RepartitionExec: partitioning=RoundRobinBatch(18), input_partitions=1 DataSourceExec: file_groups={1 groups: [[tpch_sf10/region/part-0.parquet], ...]}, projection=[r_regionkey, r_name], file_type=parquet, predicate=r_name@1 = EUROPE, pruning_predicate=r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1, required_guarantees=[r_name in (EUROPE)] HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@4)], projection=[p_partkey@3, p_mfgr@4, s_name@5, s_address@6, s_phone@8, s_acctbal@9, s_comment@10, ps_supplycost@11, n_name@1, n_regionkey@2] DataSourceExec: file_groups={1 groups: [[tpch_sf10/nation/part-0.parquet], ...]}, projection=[n_nationkey, n_name, n_regionkey], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(s_suppkey@0, ps_suppkey@2)], projection=[p_partkey@7, p_mfgr@8, s_name@1, s_address@2, s_nationkey@3, s_phone@4, s_acctbal@5, s_comment@6, ps_supplycost@10] CoalescePartitionsExec DataSourceExec: file_groups={18 groups: [[tpch_sf10/supplier/part-0.parquet:0..349257], ...]}, projection=[s_suppkey, s_name, s_address, s_nationkey, s_phone, s_acctbal, s_comment], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible HashJoinExec: mode=Partitioned, join_type=Inner, on=[(p_partkey@0, ps_partkey@0)], projection=[p_partkey@0, p_mfgr@1, ps_suppkey@3, ps_supplycost@4] RepartitionExec: partitioning=Hash([p_partkey@0], 18), input_partitions=18 FilterExec: p_size@3 = 15 AND p_type@2 LIKE %BRASS, projection=[p_partkey@0, p_mfgr@1] DataSourceExec: file_groups={18 groups: [[tpch_sf10/part/part-0.parquet:0..2809848], ...]}, projection=[p_partkey, p_mfgr, p_type, p_size], file_type=parquet, predicate=p_size@5 = 15 AND p_type@4 LIKE %BRASS, pruning_predicate=p_size_null_count@2 != row_count@3 AND p_size_min@0 <= 15 AND 15 <= p_size_max@1, required_guarantees=[p_size in (15)] RepartitionExec: partitioning=Hash([ps_partkey@0], 18), input_partitions=18 DataSourceExec: file_groups={18 groups: [[tpch_sf10/partsupp/part-0.parquet:0..17473198], ...]}, projection=[ps_partkey, ps_suppkey, ps_supplycost], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible RepartitionExec: partitioning=Hash([ps_partkey@1, min(partsupp.ps_supplycost)@0], 18), input_partitions=18 ProjectionExec: expr=[min(partsupp.ps_supplycost)@1 as min(partsupp.ps_supplycost), ps_partkey@0 as ps_partkey] AggregateExec: mode=FinalPartitioned, gby=[ps_partkey@0 as ps_partkey], aggr=[min(partsupp.ps_supplycost)] RepartitionExec: partitioning=Hash([ps_partkey@0], 18), input_partitions=18 AggregateExec: mode=Partial, gby=[ps_partkey@0 as ps_partkey], aggr=[min(partsupp.ps_supplycost)] HashJoinExec: mode=CollectLeft, join_type=RightSemi, on=[(r_regionkey@0, n_regionkey@2)], projection=[ps_partkey@0, ps_supplycost@1] CoalescePartitionsExec FilterExec: r_name@1 = EUROPE, projection=[r_regionkey@0] RepartitionExec: partitioning=RoundRobinBatch(18), input_partitions=1 DataSourceExec: file_groups={1 groups: [[tpch_sf10/region/part-0.parquet], ...]}, projection=[r_regionkey, r_name], file_type=parquet, predicate=r_name@1 = EUROPE, pruning_predicate=r_name_null_count@2 != row_count@3 AND r_name_min@0 <= EUROPE AND EUROPE <= r_name_max@1, required_guarantees=[r_name in (EUROPE)] HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(n_nationkey@0, s_nationkey@2)], projection=[ps_partkey@2, ps_supplycost@3, n_regionkey@1] DataSourceExec: file_groups={1 groups: [[tpch_sf10/nation/part-0.parquet], ...]}, projection=[n_nationkey, n_regionkey], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible HashJoinExec: mode=CollectLeft, join_type=Inner, on=[(s_suppkey@0, ps_suppkey@1)], projection=[ps_partkey@2, ps_supplycost@4, s_nationkey@1] CoalescePartitionsExec DataSourceExec: file_groups={18 groups: [[tpch_sf10/supplier/part-0.parquet:0..349257], ...]}, projection=[s_suppkey, s_nationkey], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible DataSourceExec: file_groups={18 groups: [[tpch_sf10/partsupp/part-0.parquet:0..17473198], ...]}, projection=[ps_partkey, ps_suppkey, ps_supplycost], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible ``` </details> <details><summary>Polars optimized and physical plan (SF10)</summary> ```text SORT BY [slice: (0, 100, dynamic_pred: f0fbee75-90c3-4d22-881d-4a84dc4b5d7e), descending: [true, false, false, false], nulls_last: [false, true, true, true]] [col("s_acctbal"), col("n_name"), col("s_name"), col("p_partkey")] simple π 8/8 ["s_acctbal", "s_name", ... 6 other columns] WITH_COLUMNS: [col("s_acctbal"), col("s_name"), col("n_name"), col("p_partkey"), col("p_mfgr"), col("s_address"), col("s_phone"), col("s_comment")] simple π 8/8 ["p_partkey", "p_mfgr", ... 6 other columns] INNER JOIN: LEFT PLAN ON: [col("p_partkey"), col("ps_supplycost")] FILTER col("s_acctbal").dynamic_predicate() FROM CACHE[id: 4530fe76-0c6a-42e9-8dce-f3fbf64f6804] simple π 9/9 ["p_partkey", "p_mfgr", ... 7 other columns] INNER JOIN: LEFT PLAN ON: [col("n_regionkey")] simple π 10/10 ["p_partkey", "p_mfgr", ... 8 other columns] INNER JOIN: LEFT PLAN ON: [col("s_nationkey")] simple π 9/9 ["p_partkey", "p_mfgr", ... 7 other columns] INNER JOIN: LEFT PLAN ON: [col("ps_suppkey")] simple π 4/4 ["p_partkey", "p_mfgr", ... 2 other columns] INNER JOIN: LEFT PLAN ON: [col("p_partkey")] simple π 2/2 ["p_partkey", "p_mfgr"] Parquet SCAN [tpch_sf10/part/part-0.parquet, ... 15 other sources] PROJECT 4/9 COLUMNS SELECTION: (col("p_size") == 15) & col("p_type").str.ends_with(["BRASS"]) 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("s_suppkey")] Parquet SCAN [tpch_sf10/supplier/part-0.parquet, ... 15 other sources] PROJECT */7 COLUMNS ESTIMATED ROWS: 100000 END INNER JOIN RIGHT PLAN ON: [col("n_nationkey")] Parquet SCAN [tpch_sf10/nation/part-0.parquet] PROJECT 3/4 COLUMNS ESTIMATED ROWS: 25 END INNER JOIN RIGHT PLAN ON: [col("r_regionkey")] simple π 1/1 ["r_regionkey"] Parquet SCAN [tpch_sf10/region/part-0.parquet] PROJECT 2/3 COLUMNS SELECTION: col("r_name") == "EUROPE" ESTIMATED ROWS: 5 END INNER JOIN RIGHT PLAN ON: [col("p_partkey"), col("__POLARS_CORR__POLARS_TMP_SQL_2_res").alias("__POLARS_JOIN_KEY_0___POLARS_CORR__POLARS_TMP_SQL_2_res")] AGGREGATE[maintain_order: false] [col("ps_supplycost").min().alias("__POLARS_CORR__POLARS_TMP_SQL_2_res")] BY [col("p_partkey")] FROM simple π 2/2 ["p_partkey", "ps_supplycost"] CACHE[id: 4530fe76-0c6a-42e9-8dce-f3fbf64f6804] END INNER JOIN ``` Streaming physical plan, from `POLARS_VISUALIZE_PHYSICAL_PLAN` (`_NN` stands for `_POLARS_TMP_PHYS_NN`, the table names are added): ```text in-memory-sink select: s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment sort: _19 = col("s_acctbal").row_encode([col("n_name"), col("s_name"), col("p_partkey")]) top-k: by s_acctbal, n_name, s_name, p_partkey, k = 100 select: s_acctbal, s_name, n_name, p_partkey, p_mfgr, s_address, s_phone, s_comment equi-join INNER: left_on=[p_partkey, ps_supplycost], right_on=[p_partkey, __POLARS_CORR__POLARS_TMP_SQL_6_res] filter: col("s_acctbal").dynamic_predicate() multiplexer M select: p_partkey, __POLARS_CORR__POLARS_TMP_SQL_6_res = _18 group-by: key=[p_partkey], aggs=[_18 = col("ps_supplycost").min()] select: p_partkey, ps_supplycost multiplexer M multiplexer M (the outer join, computed once) select: p_partkey, p_mfgr, s_name, s_address, s_phone, s_acctbal, s_comment, n_name, ps_supplycost equi-join INNER: n_regionkey = r_regionkey equi-join INNER: s_nationkey = n_nationkey equi-join INNER: ps_suppkey = s_suppkey equi-join INNER: p_partkey = ps_partkey multi-scan[parquet] (part): 16 sources, project: 4, filter: (col("p_size") == 15) & col("p_type").str.ends_with(["BRASS"]) multi-scan[parquet] (partsupp): 16 sources, project: 3, filter: col("ps_partkey").dynamic_predicate() & col("ps_partkey").dynamic_predicate() multi-scan[parquet] (supplier): 16 sources, project: 7 multi-scan[parquet] (nation): 1 source, project: 3 select: r_regionkey multi-scan[parquet] (region): 1 source, project: 2, filter: col("r_name") == "EUROPE" ``` </details> ### Timings per finding SF10. The wall times in finding 1 come from `datafusion-cli` for both DataFusion columns. The CPU rows come from `EXPLAIN ANALYZE` for DataFusion and `POLARS_LOG_METRICS` for Polars, with 18 threads. | | DataFusion | Polars | |---|---:|---:| | 1. Decorrelation: `q2.sql` / the Polars plan written as DataFusion SQL, 18 threads | 39.0 / 21.0 ms | 16.8 ms | | 1. The same, 1 thread | 294 / 120 ms | 102 ms | | 1. CPU of the subquery side | 246 ms | 0.6 ms | | 2. CPU of the outer `partsupp` scan and probe: default / `pushdown_filters = true` | 127 / 165 ms | 76 ms | ### Total timing Median ms. | | DataFusion | Polars | DataFusion / Polars | |---|---:|---:|---:| | SF1, 18 threads | 8.9 | 9.2 | 0.97x | | SF10, 18 threads | 37.3 | 16.8 | 2.22x | | SF10, 1 thread | 290 | 102 | 2.84x | With 18 threads at SF10, DataFusion uses about 415 ms of CPU and Polars 126 ms. ### Findings 1. **The decorrelated aggregate covers every part.** `scalar_subquery_to_join` turns the subquery into a `LeftSemi` join against `min(ps_supplycost)` grouped by `ps_partkey` over every EUROPE supplier. That is a second `partsupp` scan (8M rows), three joins and 1.18M groups, while the outer side has 6.35K rows. The semi join's dynamic filter covers both join keys, one of them the aggregate output, and `AggregateExec` only forwards filters on grouping columns. So the subquery's `partsupp` scan only gets the filter from the `supplier` join. Polars SQL sees that the subquery repeats the outer join tree and reads the minimum from the cached outer join (pola-rs/polars#29685). Its fallback restricts the subquery input to the outer keys with a semi join (pola-rs/polars#29383). Writing the Polars plan in DataFusion SQL, with the outer join as a `MATERIALIZED` CTE, returns the same 100 rows in 21.0 ms. #26218 (`PushDownSemiJoin`, which semi joins the aggregate input with the filtered scan of the outer keys and cites TPC-H Q2) potentially targets this. #8777 covers sharing repeated subplans. 2. **Join dynamic filters only prune row groups.** The outer `partsupp` scan gets `ps_partkey >= 249 AND ps_partkey <= 1999754 AND hash_lookup`. With the default `pushdown_filters = false` it prunes nothing, so all 8M rows are decoded and probed (and `nation` returns 25 rows, not 5). With `pushdown_filters = true` the scan returns 31.4K rows but costs more CPU (164 ms, of which `row_pushdown_eval_time` is 62 ms). Polars publishes a key range plus a bitset over the 8K build keys and applies it while decoding, so its scan returns the same 31.4K rows for 71 ms. The pushdown cost is tracked in #20324. All numbers are from a local machine, not the SF1000 S3 setup. SF1 shows no gap. -- 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]
