Executing broadcast join with union all sub-queries consume much memory than expected
- 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.


* * *
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.

Contributor guide
Assessment
This issue has not been assessed yet.