3.1 Exhaustive
The exhaustive join reordering algorithm is based on systematically enumerating all possible join orders. In practice, this is achieved through two fundamental rules, which together cover nearly the entire space of join permutations.
Rule 1: Join Commutativity
A join between two relations can be reordered by swapping its inputs:A JOIN B → B JOIN A
During this transformation, the join type must be adjusted accordingly. For example, a LEFT OUTER JOIN becomes a RIGHT OUTER JOIN after swapping the join operands.
Rule 2: Join Associativity
Join associativity allows the join order among three relations to be rearranged:(A JOIN B) JOIN C → A JOIN (B JOIN C)
In StarRocks, associativity is handled differently depending on the join type. Specifically, StarRocks distinguishes between:
- Associativity for Inner / Cross Joins
- Associativity for Semi Joins
3.2 Greedy
For its greedy join reordering strategy, StarRocks primarily draws inspiration from multi-sequence greedy algorithms, with a small but important enhancement: at each iteration level, instead of keeping only a single best result, StarRocks retains the top 10 candidate plans (which may not be globally optimal). These candidates are then carried forward into the next iteration, ultimately producing 10 greedy-optimized plans.
Due to the inherent limitations of greedy algorithms, this approach does not guarantee a globally optimal plan. However, by preserving multiple high-quality candidates at each step, it significantly increases the likelihood of finding a near-optimal or optimal solution.
3.3 Cost Model
StarRocks uses these join reordering algorithms to generate N candidate plans. It then evaluates them with a cost model that estimates the cost of each join. The overall cost is computed as: Join Cost = CPU × (Row(L) + Row(R)) + Memory × Row(R)
Here, Row(L) and Row(R) are the estimated output row counts of the join’s left and right children, respectively. This formula primarily accounts for the CPU cost of processing both inputs, as well as the memory cost of building the hash table on the right side of a hash join. The figure below shows how StarRocks estimates join output row counts in more detail.
Because different join reordering algorithms explore search spaces of varying sizes and have different time complexities, StarRocks benchmarks their execution time and complexity characteristics, as shown below.
Based on the observed execution costs, StarRocks applies practical limits to how different join reordering algorithms are used:
- For joins involving up to 4 tables, StarRocks uses the exhaustive algorithm.
- For joins with 4-10 tables, StarRocks generates:
- 1 plan using the left-deep strategy,
- 10 plans using the greedy algorithm,
- 1 plan using dynamic programming.
On top of these, StarRocks further explores additional plans using join commutativity.
- For joins with more than 10 tables, StarRocks relies only on the greedy and left-deep strategies, producing a total of 11 candidate plans as the basis for reordering.
- When statistics are unavailable, cost-based greedy and dynamic programming approaches become unreliable. In this case, StarRocks falls back to using a single left-deep plan as the basis for join reordering.
Distributed Join Planning
After covering the logical optimizations involved in join queries, we now turn to join execution in a distributed environment, focusing on how StarRocks optimizes distributed join planning as a distributed database.
4.1 MPP Parallel Execution
StarRocks is built on an MPP (Massively Parallel Processing) execution framework. The overall architecture is illustrated below. Using a simple join query as an example, the execution of A JOIN B in StarRocks typically proceeds as follows:
- Data from tables A and B is read in parallel from different nodes, based on their respective data distributions.
- According to the join predicate, data from A and B is reshuffled so that matching rows are sent to the same set of nodes.
- The join is executed locally on each node, and the partial results are produced.
As shown, query execution usually involves multiple sets of machines: the nodes reading table A, the nodes reading table B, and the nodes performing the join are not necessarily the same. As a result, execution inevitably involves network transfers and data exchanges.
These network operations introduce significant overhead. Therefore, a key goal in optimizing distributed join execution in StarRocks is to minimize network cost, while more intelligently partitioning and distributing the query plan to fully leverage the benefits of parallel execution.
4.2 Distributed Join Optimization
We begin by introducing the distributed execution plans that StarRocks can generate. Using a simple join query as an example:
Select * From A Join B on A.a = B.bIn practice, StarRocks can generate five basic types of distributed join plans:
- Shuffle Join Data from both tables A and B is shuffled based on the join key so that matching rows are sent to the same set of nodes, where the join is then executed.
- Broadcast Join The entire table B is broadcast to all nodes that hold table A, and the join is performed locally on those nodes. Compared to a shuffle join, this avoids shuffling table A, but requires broadcasting all of table B. This strategy is suitable when B is a small table.
- Bucket Shuffle Join An optimization over broadcast join. Instead of broadcasting table B to all nodes, B is shuffled according to A’s data distribution and sent only to the corresponding nodes that hold matching buckets of A. Globally, the shuffled data from B exists only once, significantly reducing network traffic compared to broadcast join. This strategy has an important constraint: the join key must be consistent with A’s distribution key.
- Colocate Join When tables A and B are created within the same colocate group, their data distributions are guaranteed to be identical. If the join key matches the distribution key, StarRocks can execute the join directly on the local nodes holding A and B, without any data shuffle.
- Replicate Join An experimental feature in StarRocks. If every node holding table A also contains a full copy of table B, the join can be executed locally. This approach has very strict requirements — essentially requiring the replication factor of table B to match the total number of nodes in the cluster — making it impractical in most real-world scenarios.
4.3 Exploring Distributed Join Plans
StarRocks derives distributed join plans through distribution property inference. Using a shuffle join as an example:SELECT * FROM A JOIN B ON A.a = B.b, the join operator propagates shuffle requirements top-down to tables A and B. If a scan node cannot satisfy the required distribution, StarRocks inserts an Enforce operator to introduce a shuffle. In the final execution plan, this shuffle is translated into an Exchange node responsible for network data transfer.
Other distributed join strategies are derived in the same way: the join operator requests different distribution properties from its input operators, and the optimizer generates the corresponding distributed execution plans accordingly.
4.4 Complex Distributed Joins
In real-world workloads, user queries are far more complex than a simple A JOIN B. They often involve three or more tables. For such queries, StarRocks generates a richer set of distributed execution plans, all derived from the same fundamental join strategies described earlier.
For example:
Select * From A Join B on A.a = B.b Join C on A.a = C.cUsing combinations of Shuffle Join and Broadcast Join, StarRocks can derive multiple distributed plans, as illustrated below.
If Colocate Join and Bucket Shuffle Join are also considered, even more execution plans become possible:
Despite their increased complexity, the underlying derivation logic remains the same. Distribution properties are propagated downward through the plan tree, allowing the optimizer to infer different combinations of distributed join strategies.
4.5 Global Runtime Filters
Beyond exploring distributed execution plans, StarRocks further optimizes join performance by leveraging the execution characteristics of join operators to build Global Runtime Filters.
The execution flow of a Hash Join in StarRocks is as follows:
- Retrieve the complete data set from the right table.
- Build a hash table from the right table.
- Fetch data from the left table.
- Probe the hash table to evaluate join conditions.
- Produce the join results.
Global Runtime Filters are applied between Step 2 and Step 3. After constructing the hash table on the right side, StarRocks derives runtime filter predicates from the observed data and pushes these filters down to the scan nodes of the left table before left-side data is read. This allows the left table to filter out irrelevant rows early, significantly reducing join input size.
At present, Global Runtime Filters in StarRocks support the following filtering techniques: Min/Max filters, IN predicates, and Bloom filters. The diagram below illustrates how these filters work in practice.
Summary
This article has explored StarRocks’ practical experience and ongoing work in join query optimization. All of the techniques discussed are closely aligned with the core optimization principles outlined throughout the article. When optimizing SQL queries in practice, users can also apply the following guidelines together with the features provided by StarRocks to achieve better performance:
- Join operators vary significantly in performance. Prefer high-performance join types whenever possible and avoid expensive ones. Based on typical join output sizes, the rough performance ranking is: Semi Join / Anti Join > Inner Join > Outer Join > Full Outer Join > Cross Join.
- For hash joins, building the hash table on a smaller input is far more efficient than building it on a large table.
- In multi-table joins, execute highly selective joins first to substantially reduce the cost of subsequent joins.
- Minimize the amount of data participating in joins through early filtering and pruning.
- Reduce network overhead in distributed joins as much as possible to fully benefit from parallel execution.
Case Studies
Demandbase
By leveraging StarRocks’ On-the-Fly JOIN capabilities, Demandbase successfully replaced its existing ClickHouse clusters, optimizing performance while significantly reducing costs across multiple areas.
Read the case study: Demandbase Ditches Denormalization By Switching off ClickHouse
Naver
NAVER modernized its data infrastructure with StarRocks by enabling scalable, real-time analytics over multi-table joins without denormalization. The case study highlights the critical role of efficient, on-the-fly join execution in supporting production-scale analytical workloads.
Read the case study: How JOIN Changed How We Approach Data Infra At NAVER
Shopee
Data Go is a no-code query platform where Shopee business users build queries from multiple tables. Presto struggled with complex join performance and high resource usage. When Shopee switched to StarRocks for multi-table joins, they observed 3×-10× performance improvements and a ~60% reduction in CPU usage compared with Presto on external Hive data.
Read the case study: How Shopee 3xed Their Query Performance With StarRocks
