ApsaraDB for SelectDB:Use Bucket Shuffle Join

How it works

ApsaraDB for SelectDB supports two common distributed Join strategies:

• Broadcast Join: sends the full right table to every HashJoinNode that holds left-table data. With three HashJoinNodes, that is 3x the right table's data volume in both network and memory.

• Shuffle Join: hashes both tables and redistributes all rows across cluster nodes. Network overhead equals the combined size of both tables; memory overhead equals the right table's size.

The frontend (FE) node stores the data distribution information for every table. When a Join's equality condition matches the left table's bucket column, the FE uses that distribution map to send right-table data directly to the left table's storage nodes — no redistribution of the left table required.

The result: both network and memory overhead equal only the right table's size. This advantage grows when the FE can also apply partition pruning or bucket pruning on the left table, further reducing the data each node needs to process.

How Bucket Shuffle Join differs from Colocate Join: Bucket Shuffle Join requires no knowledge of or changes to either table's data distribution, and it avoids data skew because it imposes no matching distribution rules between the two tables. It also gives Join Reorder more paths to explore during query planning.

Key concepts

Enable Bucket Shuffle Join

Set the session variable enable_bucket_shuffle_join to true. The FE then automatically selects Bucket Shuffle Join for any eligible query.

set enable_bucket_shuffle_join = true;

When planning a distributed query, the FE applies this priority order:

Colocate Join > Bucket Shuffle Join > Broadcast Join > Shuffle Join

To override the automatic selection, use a hint:

SELECT * FROM test JOIN [shuffle] baseall ON test.k1 = baseall.k1;

Verify that Bucket Shuffle Join is used

Run explain on a query to inspect its execution plan. Look for BUCKET_SHUFFLE in the join op field.

Bucket Shuffle Join applied:

|   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`            |

Fallback to another Join method (Bucket Shuffle Join did not apply):

|   2:HASH JOIN                                                       |
|   |  join op: INNER JOIN (BROADCAST)                               |
|   |  hash predicates:                                              |
|   |  colocate: false, reason: table not in the same group          |
|   |  equal join conjunct: `test`.`k1` = `baseall`.`k1`            |

If the plan shows BROADCAST or SHUFFLE instead of BUCKET_SHUFFLE, check the planning rules below to identify why the optimization did not apply.

Planning rules

Bucket Shuffle Join routes data based on hash-computed bucket assignments. All planning requirements follow from this core constraint: the FE must be able to determine, at planning time, exactly which storage nodes hold each row of the left table.