Repository navigation
TPC-H Q18 regression #27108
Description
Describe the bug
When running TPC-H Q18 on my M3 Max MBP it takes around ~233s utilizing a large amount of memory hitting a peak RSS of 22GB. On DataFusion 55.1.0 this query took ~7s hitting a peak RSS of 4GB. On my x86_64 desktop datafusion-cli gets oom killed during Q18.
It looks like the join ordering flipped.
Q18 produced the following explains:
DataFusion 55.1.0 (q18 7.0s): RightSemi, filter estimated 120M rows
SortPreservingMergeExec: [o_totalprice@4 DESC, o_orderdate@3 ASC NULLS LAST], fetch=100, statistics=[Rows=Inexact(100), Bytes=Inexact(9519), ...]
SortExec: TopK(fetch=100), expr=[o_totalprice@4 DESC, o_orderdate@3 ASC NULLS LAST], preserve_partitioning=[true], statistics=[Rows=Inexact(1400), Bytes=Inexact(133269), ...]
AggregateExec: mode=SinglePartitioned, gby=[c_name@1 as c_name, c_custkey@0 as c_custkey, o_orderkey@2 as o_orderkey, o_orderdate@4 as o_orderdate, o_totalprice@3 as o_totalprice], aggr=[sum(lineitem.l_quantity)], statistics=[Rows=Inexact(120007581), Bytes=Inexact(11423844851), ...]
HashJoinExec: mode=Partitioned, join_type=RightSemi, on=[(l_orderkey@0, o_orderkey@2)], statistics=[Rows=Inexact(120007581), Bytes=Inexact(11423844851), ...]
FilterExec: sum(lineitem.l_quantity)@1 > 300.00, projection=[l_orderkey@0], statistics=[Rows=Inexact(120007581), Bytes=Inexact(2880181930), [(Col[0]: Min=Inexact(Int64(1)) Max=Inexact(Int64(600000000)))]]
AggregateExec: mode=FinalPartitioned, gby=[l_orderkey@0 as l_orderkey], aggr=[sum(lineitem.l_quantity)], statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), [(Col[0]: Min=Exact(Int64(1)) Max=Exact(Int64(600000000))),(Col[1]:)]]
RepartitionExec: partitioning=Hash([l_orderkey@0], 14), input_partitions=14, statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), ...]
AggregateExec: mode=Partial, gby=[l_orderkey@0 as l_orderkey], aggr=[sum(lineitem.l_quantity)], statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), [(Col[0]: Min=Exact(Int64(1)) Max=Exact(Int64(600000000))),(Col[1]:)]]
DataSourceExec: file_groups={14 groups: [[lineitem.parquet ...]]}, projection=[l_orderkey, l_quantity], file_type=parquet, statistics=[Rows=Exact(600037902), Bytes=Exact(14400909648), [(Col[0]: Min=Exact(Int64(1)) Max=Exact(Int64(600000000)) Null=Exact(0) ScanBytes=Exact(4800303216)),(Col[1]: Min=Exact(Decimal128(1.00,15,2)) Max=Exact(Decimal128(50.00,15,2)) Null=Exact(0) ScanBytes=Exact(9600606432))]]
HashJoinExec: mode=Partitioned, join_type=Inner, on=[(o_orderkey@2, l_orderkey@0)], projection=[c_custkey@0, c_name@1, o_orderkey@2, o_totalprice@3, o_orderdate@4, l_quantity@6], statistics=[Rows=Inexact(150009475), Bytes=Absent, ...]
RepartitionExec: partitioning=Hash([o_orderkey@2], 14), input_partitions=14, statistics=[Rows=Inexact(150000000), Bytes=Absent, ...]
HashJoinExec: mode=Partitioned, join_type=Inner, on=[(c_custkey@0, o_custkey@1)], projection=[c_custkey@0, c_name@1, o_orderkey@2, o_totalprice@4, o_orderdate@5], statistics=[Rows=Inexact(150000000), Bytes=Absent, ...]
RepartitionExec: partitioning=Hash([c_custkey@0], 14), input_partitions=14, statistics=[Rows=Exact(15000000), Bytes=Absent, ...]
DataSourceExec: file_groups={14 groups: [[customer.parquet ...]]}, projection=[c_custkey, c_name], file_type=parquet, statistics=[Rows=Exact(15000000), Bytes=Absent, ...]
RepartitionExec: partitioning=Hash([o_custkey@1], 14), input_partitions=14, statistics=[Rows=Inexact(150000000), Bytes=Inexact(5400000000), ...]
DataSourceExec: file_groups={14 groups: [[orders.parquet ...]]}, projection=[o_orderkey, o_custkey, o_totalprice, o_orderdate], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, statistics=[Rows=Inexact(150000000), Bytes=Inexact(5400000000), ...]
RepartitionExec: partitioning=Hash([l_orderkey@0], 14), input_partitions=14, statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), ...]
DataSourceExec: file_groups={14 groups: [[lineitem.parquet ...]]}, projection=[l_orderkey, l_quantity], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), ...]
DataFusion main ee55eff / branch-56 (q18 217s): LeftSemi, filter estimated 600M rows
SortPreservingMergeExec: [o_totalprice@4 DESC, o_orderdate@3 ASC NULLS LAST], fetch=100, statistics=[Rows=Inexact(100), Bytes=Inexact(9519), ...]
SortExec: TopK(fetch=100), expr=[o_totalprice@4 DESC, o_orderdate@3 ASC NULLS LAST], preserve_partitioning=[true], statistics=[Rows=Inexact(1400), Bytes=Inexact(133269), ...]
AggregateExec: mode=SinglePartitioned, gby=[c_name@1 as c_name, c_custkey@0 as c_custkey, o_orderkey@2 as o_orderkey, o_orderdate@4 as o_orderdate, o_totalprice@3 as o_totalprice], aggr=[sum(lineitem.l_quantity)], statistics=[Rows=Inexact(150009475), Bytes=Inexact(14279805944), ...]
HashJoinExec: mode=Partitioned, join_type=LeftSemi, on=[(o_orderkey@2, l_orderkey@0)], statistics=[Rows=Inexact(150009475), Bytes=Inexact(14279805944), ...]
HashJoinExec: mode=Partitioned, join_type=Inner, on=[(o_orderkey@2, l_orderkey@0)], projection=[c_custkey@0, c_name@1, o_orderkey@2, o_totalprice@3, o_orderdate@4, l_quantity@6], statistics=[Rows=Inexact(150009475), Bytes=Absent, ...]
RepartitionExec: partitioning=Hash([o_orderkey@2], 14), input_partitions=14, statistics=[Rows=Inexact(150000000), Bytes=Absent, ...]
HashJoinExec: mode=Partitioned, join_type=Inner, on=[(c_custkey@0, o_custkey@1)], projection=[c_custkey@0, c_name@1, o_orderkey@2, o_totalprice@4, o_orderdate@5], statistics=[Rows=Inexact(150000000), Bytes=Absent, ...]
RepartitionExec: partitioning=Hash([c_custkey@0], 14), input_partitions=14, statistics=[Rows=Exact(15000000), Bytes=Absent, ...]
DataSourceExec: file_groups={14 groups: [[customer.parquet ...]]}, projection=[c_custkey, c_name], file_type=parquet, statistics=[Rows=Exact(15000000), Bytes=Absent, ...]
RepartitionExec: partitioning=Hash([o_custkey@1], 14), input_partitions=14, statistics=[Rows=Inexact(150000000), Bytes=Inexact(5400000000), ...]
DataSourceExec: file_groups={14 groups: [[orders.parquet ...]]}, projection=[o_orderkey, o_custkey, o_totalprice, o_orderdate], file_type=parquet, predicate=DynamicFilter [ empty ] AND DynamicFilter [ empty ], dynamic_rg_pruning=eligible, statistics=[Rows=Inexact(150000000), Bytes=Inexact(5400000000), ...]
RepartitionExec: partitioning=Hash([l_orderkey@0], 14), input_partitions=14,(600037902), Bytes=Inexact(14400909648), ...]
DataSourceExec: file_groups={14 groups: [[lineitem.parquet ...]]}, projection=[l_orderkey, l_quantity], file_type=parquet, predicate=DynamicFilter [ empty ], dynamic_rg_pruning=eligible,
statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), ...]
FilterExec: sum(lineitem.l_quantity)@1 > 300.00, projection=[l_orderkey@0], statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), [(Col[0]: Min=Inexact(Int64(1))
Max=Inexact(Int64(600000000)))]]
AggregateExec: mode=FinalPartitioned, gby=[l_orderkey@0 as l_orderkey], aggr=[sum(lineitem.l_quantity)], statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), [(Col[0]:
Min=Inexact(Int64(1)) Max=Inexact(Int64(600000000))),(Col[1]:)]]
RepartitionExec: partitioning=Hash([l_orderkey@0], 14), input_partitions=14, statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), ...]
AggregateExec: mode=Partial, gby=[l_orderkey@0 as l_orderkey], aggr=[sumtatistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), [(Col[0]:Min=Inexact(Int64(1)) Max=Inexact(Int64(600000000))),(Col[1]:)]]
DataSourceExec: file_groups={14 groups: [[lineitem.parquet ...]]}, proantity], file_type=parquet, predicate=DynamicFilter [ empty ],dynamic_rg_pruning=eligible, statistics=[Rows=Inexact(600037902), Bytes=Inexact(14400909648), ...]
To Reproduce
- Generate SF100 TPC-H data using https://github.com/datafusion-contrib/tpcgen-rs
- Checkout https://github.com/apache/datafusion/tree/branch-56 and build the release version of
datafusion-clicargo build --release -p datafusion-cli
- Run TPC-H Q18 against the benchmark data generated
Expected behavior
- TPC-H Q18 runs in the same amount of time and RSS as v55.1.0 or better.
Activity
devanbenz commentedon Nov 4, 2026
Traversing through commits, I found that reverting just: #25790 out of the branch resolves the Q18 regression. I haven't done an analysis on why yet but Q18 goes back down to ~7s after revert.
Revert apache#25790 (Decimal filter selectivity) due to TPC-H Q18 reg…
[branch-56] Revert #25790 (Decimal filter selectivity) due to TPC-H Q…