devanbenz opened a new issue, #27108:
URL: https://github.com/apache/datafusion/issues/27108
### 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:
<details>
<summary>DataFusion 55.1.0 (q18 7.0s): RightSemi, filter estimated 120M
rows</summary>
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](tel: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](tel:4800303216))),(Col[1]:
Min=Exact(Decimal128(1.00,15,2)) Max=Exact(Decimal128(50.00,15,2))
Null=Exact(0) ScanBytes=Exact([9600606432](tel: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](tel: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](tel: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), ...]
</details>
<details>
<summary>DataFusion main ee55effcc / branch-56 (q18 217s): LeftSemi, filter
estimated 600M rows</summary>
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](tel: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](tel: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), ...]
</details>
### 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-cli`
- `cargo build --release -p datafusion-cli`
- Run TPC-H Q18 against the benchmark data generated
### Expected behavior
_No response_
### Additional context
_No response_
### Willingness to contribute
None
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
