comphead opened a new issue, #27310: URL: https://github.com/apache/datafusion/issues/27310
Part of #26169, following #27279 (Q01) and #27282 (Q02). At SF1000, DataFusion was 5.43x slower than Polars on Q15 ([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 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). <details><summary>Query text used for both engines</summary> ```sql 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; ``` </details> ### Plans There is one join, `supplier` against `revenue0`, built on `supplier` in both engines. The plans differ in how often `revenue0` is computed. <details><summary>DataFusion logical and physical plan (SF10)</summary> ```text 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=[] ``` </details> <details><summary>Polars optimized and physical plan (SF10)</summary> ```text 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): ```text 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) ``` </details> ### 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 1. **`revenue0` is computed twice, one pass after the other.** DataFusion inlines the view into both the FROM and the `max` subquery, so `lineitem` is scanned (60M rows, 450 MB) and aggregated twice. `ScalarSubqueryExec` runs the subquery to completion before it starts the main plan: the per-partition `BaselineMetrics` show the first pass from 0 to 76 ms and the second from 76 to 148 ms. Polars computes `revenue0` once (`CACHE`), and both the join and the `max` read it. With `WITH 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 #8777. 2. **Scan work granularity**, as in #27279. In the first pass, the scan partitions finish between 43 and 76 ms. With the materialized CTE, `target_partitions = 64` gives 62.0 ms. 3. **The date filter selects 3.78% of the rows.** Polars decodes `l_shipdate` first and the other three columns only for the rows that match. DataFusion decodes all four columns and then runs `FilterExec`. `pushdown_filters = true` returns the same 2.27M rows from the scan but costs more (626 ms, and 159 vs 144 ms at 18 threads), although `row_pushdown_eval_time` is only about 11 ms. The pushdown cost is tracked in #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. -- 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]
