pingcap / pingcap/tidb

Executing broadcast join with union all sub-queries consume much memory than expected

Open
#58,251 3 comments 0 reactions 0 assignees View on GitHub
sig/execution type/enhancement
Dominant language
Go
Stars
40.5k
Forks
6.2k
PR merge metrics
PR metrics pending

Description

## Enhancement
Load TPC-H 50GB into a TiDB cluster with TiFlash write node and TiFlash compute node.

The default execution plan of following query use boardcast-join instead of shuffle join

```
-- 5 sub-queries
select count(*) from (
WITH `op` AS (
select *, row_number() over (partition by l_partkey,date(l_shipdate) order by l_shipdate) times from lineitem where date(l_shipdate) between '1996-01-01' and '1996-03-01'
UNION ALL
select *, row_number() over (partition by l_partkey,date(l_shipdate) order by l_shipdate) times from lineitem where date(l_shipdate) between '1996-03-01' and '1996-05-01'
UNION ALL
select *, row_number() over (partition by l_partkey,date(l_shipdate) order by l_shipdate) times from lineitem where date(l_shipdate) between '1996-05-01' and '1996-07-01'
UNION ALL
select *, row_number() over (partition by l_partkey,date(l_shipdate) order by l_shipdate) times from lineitem where date(l_shipdate) between '1996-07-01' and '1996-09-01'
UNION ALL
select *, row_number() over (partition by l_partkey,date(l_shipdate) order by l_shipdate) times from lineitem where date(l_shipdate) between '1996-09-01' and '1996-11-01'
)
SELECT
`op`.*,
`u`.`p_brand` `p_brand`
FROM
`op`
JOIN part `u` ON `u`.`p_partkey` = `op`.`l_partkey`
WHERE
`op`.`times` = 1
) t;
```

Execution plan details:

[shuffle-join-exec-details.sql.zip](https://github.com/user-attachments/files/18123205/shuffle-join-exec-details.sql.zip)
[boardcast-join-exec-details.sql.zip](https://github.com/user-attachments/files/18123206/boardcast-join-exec-details.sql.zip)

Execution the query take almost 18GB memory on tiflash. The hash-join table takes more than 11GB.

![Image](https://github.com/user-attachments/assets/6a3c2310-2466-4725-9376-21811bb3e640)

![Image](https://github.com/user-attachments/assets/eeaafb5c-90be-4d99-a14b-8a6ca813194a)

* * *

For a TPC-H 50GiB dataset, when broadcasting the `part` table (10,000,000 rows) to join with tables containing `UNION ALL` sub-queries, each additional `UNION ALL` requires broadcasting an extra copy of the `part` table. Moreover, for every additional tiflash compute node, a copy of the `part` table is needed on that compute node as well.

Consequently, in the case of using a broadcast join, the part table needs to generate `num-of-sub-queries * num-of-tiflash-compute-node` complete copies. This results in additional overhead in terms of network bandwidth, CPU, and memory.

![Image](https://github.com/user-attachments/assets/8a560fb9-6d25-4cbc-b959-6c0c4ee01b6f)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.