Inside StarRocks: Why Joins Are Faster Than You’d Expect

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:

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:

On top of these, StarRocks further explores additional plans using join commutativity.

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:

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

In practice, StarRocks can generate five basic types of distributed join plans:

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

Using 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:

  1. Retrieve the complete data set from the right table.
  2. Build a hash table from the right table.
  3. Fetch data from the left table.
  4. Probe the hash table to evaluate join conditions.
  5. 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:

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

Link nội dung: https://ohanapreschool.edu.vn/in-join-a57233.html