Optimize join queries

Supported physical operators

ApsaraDB for SelectDB supports two physical join operators for join queries in a standalone engine.

Hash Join: Builds a hash table on the right table based on the equi-join columns. The left table then streams through the hash table to perform join calculations. This operator applies only to equi-join queries.
Nest Loop Join: Performs join calculations by using two nested loops. This operator supports non-equi-join conditions such as ">" and "<" comparisons, as well as cartesian products. Although widely applicable, it has poor performance.

Shuffle methods

ApsaraDB for SelectDB is a distributed massively parallel processor (MPP) database. Join queries require data to be shuffled across nodes before physical operators can perform the join. ApsaraDB for SelectDB supports the following four shuffle methods.

In this example, Table S and Table R are joined. N indicates the number of nodes that are involved in the join calculation. T indicates the number of data records in a table.

Broadcast Join

This shuffle method requires that the full data of Table S is sent to Table R. Each node involved in the join calculation has the full data of Table S, which is T(R). This is a common shuffle method and supports the Hash Join and Nest Loop Join operators. The network overhead of this shuffle method is calculated by using the following formula: N × T(R).

The data of Table S is not moved, and the data of Table R is sent to the scan node that scans the data of Table S.
Shuffle Join

When you use the Hash Join operator to perform a join query, you can calculate the hash values of data in Table S and Table R based on the Join column and send the same hash values to the same node of a distributed system. Then, you can use the distributed system to improve the speed of the join query. The network overhead of this shuffle method is calculated based on the following formula: T(S) + T(R). However, this shuffle method supports only the Hash Join operator because the hash values of data are calculated based on the join column.

The data of Table S and Table R is calculated based on partitions. The results are sent to nodes of different partitions.
Bucket Shuffle Join

ApsaraDB for SelectDB stores table data in buckets based on hash values. When the join column matches a table's bucket column, data shuffle can leverage this distribution. For example, if the join column is the bucket column of Table S, only Table R's data needs to be moved for the join calculation.

The network overhead of this shuffle method is T(R), which indicates that the join calculation can be performed by shuffling only the data of Table R. For more information about how to use the Bucket Shuffle Join method, see Bucket Shuffle Join.

The data of Table S is not moved, and the data of Table R is sent to the node that scans the data of Table S based on the results that are calculated based on partitions.
Colocation Join

When associated tables are created with the same number of shards, data shuffle can be skipped entirely, and join calculations are performed directly on local data. For more information about Colocation Join, see Colocation Join.

The data is partitioned in advance. You do not need to consider the network overhead. You can directly perform join calculations on your on-premises computer.

The following table describes the four shuffle methods.

Shuffle method	Network overhead	Physical operator	Scenario
BroadCast	N × T(R)	Hash Join or Nest Loop Join	Common scenarios
Shuffle	T(S) + T(R)	Hash Join	Common scenarios
Bucket Shuffle	T(R)	Hash Join	Distributed columns of the left table exist in the join condition, and only the data in a single partition of the left table is used for a join query.
Colocation	0	Hash Join	Distributed columns of the left table exist in the join condition, and the left and right tables belong to the same Colocate Group.

The shuffle methods above are listed in descending order of flexibility. Methods with stricter data distribution requirements offer higher join performance.

Runtime Filter

A Runtime Filter is generated dynamically during query execution. During a join query, HashJoinNode derives a filter condition from the right table and pushes it down to the ScanNode of the left table. The ScanNode then filters and prunes data during the scan, which greatly reduces data reading and computation time. For more information about Runtime Filter, see Runtime Filter.

Join Reorder

In multi-table join scenarios, the order of joins greatly affects the performance of the entire query.

In the following example, three tables are joined. ScanA, ScanB, and ScanC represent the data scanned from Table A, Table B, and Table C based on query conditions.

In the left part of the figure, the scanned data of Table A and Table B is joined and 2,000 rows of intermediate results are generated. Then, the intermediate results are joined with the scanned data of Table C.

In the right part of the figure, the join order is adjusted. Table A and Table C are joined first, producing only 100 intermediate rows. These rows are then joined with Table B. The final result is the same, but the left approach generates 20 times more intermediate rows, causing a significant performance gap.

ApsaraDB for SelectDB supports the rule-based Join Recorder algorithm. The following items describe the logic of this algorithm:

Join large tables with small tables to generate a smaller amount of intermediate results.
Place tables with join conditions before those without in the FROM clause to enable early data filtering.
Hash Join takes priority over Nest Loop Join because it executes significantly faster.

Solutions to join query optimization

To optimize join queries, perform the following steps:

Use the profile provided by ApsaraDB for SelectDB to troubleshoot a query. The profile records various information in the entire query, which is very important for performance optimization.
Understand the join mechanism of ApsaraDB for SelectDB and analyze the reasons for a slow query.
Use session variables to modify some behaviors of join operations to optimize join operations.
Check the query plan to determine whether the optimization takes effect.

These four steps cover the standard join optimization process. If query performance does not improve, you may need to rewrite SQL join statements or adjust data distribution. Manual rewriting is more costly but may be necessary when the standard approach is insufficient.

Suggestions on optimizing join queries

Keep the following best practices in mind when you optimize join queries:

Select columns of the same type to avoid data casts, or use simple column types to accelerate join calculations.
Use key columns for joins to enable late materialization. For more information, see Runtime Filter.
Use Colocation mode for joins among large tables to reduce network overhead from data shuffle.
Use Runtime Filter when a high filter rate is expected. Runtime Filter can have side effects, so evaluate its suitability based on the join columns in your SQL statements.
For multi-table joins, ensure that the left table is the larger table and the right table is the smaller table. Hash Join outperforms Nest Loop Join in this configuration. Use SQL hints to adjust join order if needed.