TPC-H Q02: DataFusion vs Polars comparison

Aperta
  • Qualcosa non va?
#27,282 0 commenti 1 reazione 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

Nessuno ha ancora preso questa issue.

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
48/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
rust, sql
Ambito
databases, performance

Direzione di ricerca

Start with benchmarks/queries/q2.sql and the setup and plans in this issue; compare DataFusion's and Polars' execution of Q02 at SF1 and SF10. The issue points to the benchmark data and query, but does not identify a specific operator or code location to change. Done means locating and addressing the source of the performance gap, then reporting comparable timings for both scale factors.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

enhancement

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 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.

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 (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.

Lingua principale
Rust
Stelle
9.6k
Fork
2.6k
Merge medio
3g 4h
PR unite (30g)
345

Preparare l'ambiente

Apri in Codespaces

Avvia il container di sviluppo del progetto nel browser, con il tuo account GitHub.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/datafusion

Tutte le issue di apache/datafusion

Issue simili

Altre issue su Rust

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.