Bucket Shuffle Join is a new function officially added in Doris 0.14. The purpose is to provide local optimization for some join queries to reduce the time-consuming of data transmission between nodes and speed up the query.
It's design, implementation can be referred to ISSUE 4394。
The conventional distributed join methods supported by Doris is: Shuffle Join, Broadcast Join
. Both of these join will lead to some network overhead.
For example, there are join queries for table A and table B. the join method is hashjoin. The cost of different join types is as follows:
3B
, and its memory overhead is 3B
.A + B
and memory overhead is B
.The data distribution information of each Doris table is saved in FE. If the join statement hits the data distribution column of the left table, we should use the data distribution information to reduce the network and memory overhead of the join query. This is the source of the idea of bucket shuffle join.
The picture above shows how the Bucket Shuffle Join works. The SQL query is A table join B table. The equivalent expression of join hits the data distribution column of A. According to the data distribution information of table A. Bucket Shuffle Join sends the data of table B to the corresponding data storage and calculation node of table A. The cost of Bucket Shuffle Join is as follows:
network cost: B < min(3B, A + B)
memory cost: B <= min(3B, B)
Therefore, compared with Broadcast Join and Shuffle Join, Bucket shuffle join has obvious performance advantages. It reduces the time-consuming of data transmission between nodes and the memory cost of join. Compared with Doris's original join method, it has the following advantages
Set session variable enable_bucket_shuffle_join
to true
, FE will automatically plan queries that can be converted to Bucket Shuffle Join.
set enable_bucket_shuffle_join = true;
In FE's distributed query planning, the priority order is Colocate Join -> Bucket Shuffle Join -> Broadcast Join -> Shuffle Join. However, if the user explicitly hints the type of join, for example:
select * from test join [shuffle] baseall on test.k1 = baseall.k1;
the above order of preference will not take effect.
The session variable is set to true
by default in version 0.14, while it needs to be set to true
manually in version 0.13.
You can use the explain
command to check whether the join is a Bucket Shuffle Join
| 2:HASH JOIN | | | join op: INNER JOIN (BUCKET_SHUFFLE) | | | hash predicates: | | | colocate: false, reason: table not in the same group | | | equal join conjunct: `test`.`k1` = `baseall`.`k1`
The join type indicates that the join method to be used is:BUCKET_SHUFFLE
。
In most scenarios, users only need to turn on the session variable by default to transparently use the performance improvement brought by this join method. However, if we understand the planning rules of Bucket Shuffle Join, we can use it to write more efficient SQL.
where
condition as far as possible to make the partition clipping policy effective.