TPC-H Q02: DataFusion vs Polars comparison

Aberta
  • Algo errado?
#27,282 0 comentários 1 reação 0 responsáveis Ver no GitHub

Mantenedores costumam responder em até 1 dia

Ninguém assumiu esta issue ainda.

Avaliação

Dificuldade
4/5
Tempo estimado
3-5 dias
Facilidade para iniciantes
48/100
Tipo de issue
Bug
Clareza
Razoavelmente clara
Status de atividade
Ativa
Stack de tecnologia
rust, sql
Domínio
databases, performance

Direção de pesquisa

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.

Escrita pelo modelo de indexação a partir do texto da issue.

Descrição

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.

Linguagem predominante
Rust
Estrelas
9.6k
Forks
2.6k
Merge médio
3d 4h
PRs com merge (30d)
345

Preparar o ambiente

Abrir no Codespaces

Inicia o contêiner de desenvolvimento do projeto no navegador, com a sua própria conta do GitHub.

Primeiros passos

  1. Leia a issue inteira e depois o guia de contribuição do projeto.
  2. Comente na issue dizendo que vai assumir — evita que duas pessoas façam o mesmo trabalho.
  3. Faça um fork do repositório e trabalhe em uma branch.
  4. Abra um pull request que referencie o número da issue.

Mais de apache/datafusion

Todas as issues de apache/datafusion

Issues semelhantes

Mais issues de Rust

Receba novas issues na sua caixa de entrada

Um resumo curto de issues do GitHub para quem está começando.