Improve performance of TPCH q4, q7, q9 at TPCH SF 100

[alamb]

Is your feature request related to a problem or challenge?

@MrPowers reportes in discord: https://discord.com/channels/885562378132000778/1290751484807352412/1407568961561952277

I ran the TPC-H queries on my Macbook M3 with 16GB of RAM with different scale factors. The DataFusion, DuckDB, and Polars Streaming results are similar for scale factor 5:

I ran the TPC-H queries on my Macbook M3 with 16GB of RAM with different scale factors. The DataFusion, DuckDB, and Polars Streaming results are similar for scale factor 5:

Describe the solution you'd like

Figure out why q4, q7 and q9 are very slow

The TPCH queries are here: https://github.com/apache/datafusion/tree/main/benchmarks/queries

q4

select
    o_orderpriority,
    count(*) as order_count
from
    orders
where
        o_orderdate >= '1993-07-01'
  and o_orderdate < date '1993-07-01' + interval '3' month
  and exists (
        select
            *
        from
            lineitem
        where
                l_orderkey = o_orderkey
          and l_commitdate < l_receiptdate
    )
group by
    o_orderpriority
order by
    o_orderpriority;

q7

select
    supp_nation,
    cust_nation,
    l_year,
    sum(volume) as revenue
from
    (
        select
            n1.n_name as supp_nation,
            n2.n_name as cust_nation,
            extract(year from l_shipdate) as l_year,
            l_extendedprice * (1 - l_discount) as volume
        from
            supplier,
            lineitem,
            orders,
            customer,
            nation n1,
            nation n2
        where
                s_suppkey = l_suppkey
          and o_orderkey = l_orderkey
          and c_custkey = o_custkey
          and s_nationkey = n1.n_nationkey
          and c_nationkey = n2.n_nationkey
          and (
                (n1.n_name = 'FRANCE' and n2.n_name = 'GERMANY')
                or (n1.n_name = 'GERMANY' and n2.n_name = 'FRANCE')
            )
          and l_shipdate between date '1995-01-01' and date '1996-12-31'
    ) as shipping
group by
    supp_nation,
    cust_nation,
    l_year
order by
    supp_nation,
    cust_nation,
    l_year;

q9

select
    nation,
    o_year,
    sum(amount) as sum_profit
from
    (
        select
            n_name as nation,
            extract(year from o_orderdate) as o_year,
            l_extendedprice * (1 - l_discount) - ps_supplycost * l_quantity as amount
        from
            part,
            supplier,
            lineitem,
            partsupp,
            orders,
            nation
        where
                s_suppkey = l_suppkey
          and ps_suppkey = l_suppkey
          and ps_partkey = l_partkey
          and p_partkey = l_partkey
          and o_orderkey = l_orderkey
          and s_nationkey = n_nationkey
          and p_name like '%green%'
    ) as profit
group by
    nation,
    o_year
order by
    nation,
    o_year desc;

Describe alternatives you've considered

No response

Additional context

No response

[alamb]

I strongly suspect the issue is Join Order

Someone could prove / disprove this theory by looking at the plans from DataFusion and DuckDB and comparing the join order

Also, I suspect that the TPCH performance will improve with the Dynamic join filters @adriangb added in

• Add dynamic filter (bounds) pushdown to HashJoinExec #16445

[alamb]

@MrPowers, how are you running these benchmarks?

[2010YOUY01]

I checked q4, the join order is not optimal. After applying the filter left table is still way larger than the right table.

I think another reason to make the performance gap larger is: this bad join order caused memory usage greater than physical memory limit (I tried it's using 21GB peak memory, which is > 16GB in the report), and the OS swapping makes it significantly slower.

DataFusion CLI v49.0.1
> CREATE EXTERNAL TABLE IF NOT EXISTS lineitem (
        l_orderkey BIGINT,
        l_partkey BIGINT,
        l_suppkey BIGINT,
        l_linenumber INTEGER,
        l_quantity DECIMAL(15, 2),
        l_extendedprice DECIMAL(15, 2),
        l_discount DECIMAL(15, 2),
        l_tax DECIMAL(15, 2),
        l_returnflag VARCHAR,
        l_linestatus VARCHAR,
        l_shipdate DATE,
        l_commitdate DATE,
        l_receiptdate DATE,
        l_shipinstruct VARCHAR,
        l_shipmode VARCHAR,
        l_comment VARCHAR,
) STORED AS parquet
LOCATION '/Users/yongting/Code/datafusion/benchmarks/data/tpch_sf1/lineitem';

CREATE EXTERNAL TABLE orders
STORED AS PARQUET
LOCATION '/Users/yongting/Code/datafusion/benchmarks/data/tpch_sf1/orders/';
0 row(s) fetched.
Elapsed 0.023 seconds.

0 row(s) fetched.
Elapsed 0.009 seconds.

> explain select
    o_orderpriority,
    count(*) as order_count
from
    orders
where
        o_orderdate >= '1993-07-01'
  and o_orderdate < date '1993-07-01' + interval '3' month
  and exists (
        select
            *
        from
            lineitem
        where
                l_orderkey = o_orderkey
          and l_commitdate < l_receiptdate
    )
group by
    o_orderpriority
order by
    o_orderpriority;
+---------------+------------------------------------------------------------+
| plan_type     | plan                                                       |
+---------------+------------------------------------------------------------+
| physical_plan | ┌───────────────────────────┐                              |
|               | │  SortPreservingMergeExec  │                              |
|               | │    --------------------   │                              |
|               | │ o_orderpriority ASC NULLS │                              |
|               | │            LAST           │                              |
|               | └─────────────┬─────────────┘                              |
|               | ┌─────────────┴─────────────┐                              |
|               | │          SortExec         │                              |
|               | │    --------------------   │                              |
|               | │   o_orderpriority@0 ASC   │                              |
|               | │         NULLS LAST        │                              |
|               | └─────────────┬─────────────┘                              |
|               | ┌─────────────┴─────────────┐                              |
|               | │       ProjectionExec      │                              |
|               | │    --------------------   │                              |
|               | │      o_orderpriority:     │                              |
|               | │      o_orderpriority      │                              |
|               | │                           │                              |
|               | │        order_count:       │                              |
|               | │      count(Int64(1))      │                              |
|               | └─────────────┬─────────────┘                              |
|               | ┌─────────────┴─────────────┐                              |
|               | │       AggregateExec       │                              |
|               | │    --------------------   │                              |
|               | │       aggr: count(1)      │                              |
|               | │                           │                              |
|               | │         group_by:         │                              |
|               | │      o_orderpriority      │                              |
|               | │                           │                              |
|               | │           mode:           │                              |
|               | │      FinalPartitioned     │                              |
|               | └─────────────┬─────────────┘                              |
|               | ┌─────────────┴─────────────┐                              |
|               | │    CoalesceBatchesExec    │                              |
|               | │    --------------------   │                              |
|               | │     target_batch_size:    │                              |
|               | │            8192           │                              |
|               | └─────────────┬─────────────┘                              |
|               | ┌─────────────┴─────────────┐                              |
|               | │      RepartitionExec      │                              |
|               | │    --------------------   │                              |
|               | │ partition_count(in->out): │                              |
|               | │          14 -> 14         │                              |
|               | │                           │                              |
|               | │    partitioning_scheme:   │                              |
|               | │ Hash([o_orderpriority@0], │                              |
|               | │             14)           │                              |
|               | └─────────────┬─────────────┘                              |
|               | ┌─────────────┴─────────────┐                              |
|               | │       AggregateExec       │                              |
|               | │    --------------------   │                              |
|               | │       aggr: count(1)      │                              |
|               | │                           │                              |
|               | │         group_by:         │                              |
|               | │      o_orderpriority      │                              |
|               | │                           │                              |
|               | │       mode: Partial       │                              |
|               | └─────────────┬─────────────┘                              |
|               | ┌─────────────┴─────────────┐                              |
|               | │    CoalesceBatchesExec    │                              |
|               | │    --------------------   │                              |
|               | │     target_batch_size:    │                              |
|               | │            8192           │                              |
|               | └─────────────┬─────────────┘                              |
|               | ┌─────────────┴─────────────┐                              |
|               | │        HashJoinExec       │                              |
|               | │    --------------------   │                              |
|               | │    join_type: RightSemi   │                              |
|               | │                           ├──────────────┐               |
|               | │            on:            │              │               |
|               | │ (l_orderkey = o_orderkey) │              │               |
|               | └─────────────┬─────────────┘              │               |
|               | ┌─────────────┴─────────────┐┌─────────────┴─────────────┐ |
|               | │    CoalesceBatchesExec    ││    CoalesceBatchesExec    │ |
|               | │    --------------------   ││    --------------------   │ |
|               | │     target_batch_size:    ││     target_batch_size:    │ |
|               | │            8192           ││            8192           │ |
|               | └─────────────┬─────────────┘└─────────────┬─────────────┘ |
|               | ┌─────────────┴─────────────┐┌─────────────┴─────────────┐ |
|               | │      RepartitionExec      ││      RepartitionExec      │ |
|               | │    --------------------   ││    --------------------   │ |
|               | │ partition_count(in->out): ││ partition_count(in->out): │ |
|               | │          14 -> 14         ││          14 -> 14         │ |
|               | │                           ││                           │ |
|               | │    partitioning_scheme:   ││    partitioning_scheme:   │ |
|               | │  Hash([l_orderkey@0], 14) ││  Hash([o_orderkey@0], 14) │ |
|               | └─────────────┬─────────────┘└─────────────┬─────────────┘ |
|               | ┌─────────────┴─────────────┐┌─────────────┴─────────────┐ |
|               | │    CoalesceBatchesExec    ││    CoalesceBatchesExec    │ |
|               | │    --------------------   ││    --------------------   │ |
|               | │     target_batch_size:    ││     target_batch_size:    │ |
|               | │            8192           ││            8192           │ |
|               | └─────────────┬─────────────┘└─────────────┬─────────────┘ |
|               | ┌─────────────┴─────────────┐┌─────────────┴─────────────┐ |
|               | │         FilterExec        ││         FilterExec        │ |
|               | │    --------------------   ││    --------------------   │ |
|               | │         predicate:        ││         predicate:        │ |
|               | │      l_receiptdate >      ││ o_orderdate >= 1993-07-01 │ |
|               | │        l_commitdate       ││   AND o_orderdate < 1993  │ |
|               | │                           ││           -10-01          │ |
|               | └─────────────┬─────────────┘└─────────────┬─────────────┘ |
|               | ┌─────────────┴─────────────┐┌─────────────┴─────────────┐ |
|               | │       DataSourceExec      ││       DataSourceExec      │ |
|               | │    --------------------   ││    --------------------   │ |
|               | │         files: 21         ││         files: 21         │ |
|               | │      format: parquet      ││      format: parquet      │ |
|               | │                           ││                           │ |
|               | │         predicate:        ││         predicate:        │ |
|               | │      l_receiptdate >      ││ o_orderdate >= 1993-07-01 │ |
|               | │        l_commitdate       ││   AND o_orderdate < 1993  │ |
|               | │                           ││           -10-01          │ |
|               | └───────────────────────────┘└───────────────────────────┘ |
|               |                                                            |
+---------------+------------------------------------------------------------+
1 row(s) fetched.
Elapsed 0.050 seconds

[MrPowers]

@alamb - Here's how I am running these benchmarks: https://github.com/MrPowers/querybench

Here's how to generate the datasets and run the benchmarks: https://github.com/MrPowers/querybench/blob/main/CONTRIBUTING.md

Feel free to provide feedback or let me know how I can make these benchmarks more useful. I will continue interating to make them better!

[alamb]

I think another reason to make the performance gap larger is: this bad join order caused memory usage greater than physical memory limit (I tried it's using 21GB peak memory, which is > 16GB in the report), and the OS swapping makes it significantly slower.

• This is something that @adriangb 's recent work for Push Dynamic Join Predicates into Scan ("Sideways Information Passing", etc) #7955 may also help with (it would apply more filtering on the lineitem table maybe)

[MrPowers]

As soon as #7955 is released, I am happy to rerun these benchmarks and report back!

[alamb]

As soon as #7955 is released, I am happy to rerun these benchmarks and report back!

This will be in DataFusion 50.0.0:

• Release DataFusion 50.0.0 (Aug/Sep 2025) #16799

[nuno-faria]

I leave here some previous analyzes I had done on TPC-H when comparing to DuckDB, related to join order and join parameterization (#16710 (comment)):

[ctsk]

Looks like you have a copy/paste error in your benchmark code for datafusion @MrPowers

https://github.com/MrPowers/querybench/blob/03b615d51f3f3b700516738b543c6b6fdec0665c/querybench/datafusion/tpch_queries.py#L110-L122

[MrPowers]

@ctsk - good catch, updated the code. Looks like we should investigate q18 as well.

[ctsk]

Predicate filters are likely not a large help on tpch, since that data is notoriously uniform - thus little skipping will occur.

Ultimately, the issue is that the machine is running out of ram - minor optimizations of the hash joins wont really help it: Instead we must do better choices on what we pick as build and probe side.

I've had a look at polars join impl and it seems that they are doing something smart: They pick build and probe side during execution. A rough description would be:

• First buffer a constant amount of rows of both inputs to the join.
• If one side is exhausted before the threshold is reached, then that side becomes the build side.
  - if the exhausted side is very small, pick a nested loop join
• ...else the side with the smaller estimated cardinality (HLL sketched) becomes the build side.

Perhaps this can side-step some of the bad join decisions / large build sides that datafusion makes.

Thoughts on incorporating such an approach in datafusion? @alamb

[alamb]

Thoughts on incorporating such an approach in datafusion? @alamb

I think it is very much a good idea personally as the plan time choice of join algorithms (and orders) is notoriously tricky.

FYI @2010YOUY01 who made a PR recently to provide more control over join algorithms -- this is an interesting alternative (make the join operators themselves more dynamic)