Skip to content

TPC-H Q02: DataFusion vs Polars comparison #27282

New issue
New issue
Open
Open
TPC-H Q02: DataFusion vs Polars comparison#27282
Labels
enhancementNew feature or request

Description

@comphead
comphead
opened on Nov 10, 2026

Part of #26169, following the Q01 comparison in #27279. At SF1000, DataFusion was 5.42x slower than Polars on Q02 (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 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.

DataFusion logical and physical plan (SF10) 
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
Polars optimized and physical plan (SF10) 
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):

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"

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 (perf: Read correlated SQL aggregates from the outer join when the subquery repeats it pola-rs/polars#29685). Its fallback restricts the subquery input to the outer keys with a semi join (perf: Restrict correlated SQL aggregates to requested keys 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. perf: push semi joins below inner joins and restrict decorrelated aggregates #26218 (PushDownSemiJoin, which semi joins the aggregate input with the filtered scan of the outer keys and cites TPC-H Q2) potentially targets this. Avoid recompute CTEs (common table expressions) / share input plans #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 [EPIC] Fix performance regressions when enabling parquet filter pushdown (late materialization) #20324.

All numbers are from a local machine, not the SF1000 S3 setup. SF1 shows no gap.

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Metadata

Metadata

Assignees

No one assigned

    Labels

    enhancementNew feature or request

    Type

    No type

    Projects

    No projects

      Milestone

      No milestone

      Relationships

      None yet

      Development

      No branches or pull requests

      Issue actions