Two recent pull requests to ClickHouse substantially improve JOIN performance in common scenarios.
Two new pull requests have been merged into ClickHouse and will be available in ClickHouse 24.4. These changes improve JOIN performance in many production scenarios, in some cases speeding up query execution by several orders of magnitude.
Pull Request #1: JOIN predicate pushdown using equivalence classes
Predicate pushdown is a query optimization technique used in database management systems to greatly reduce the amount of data a query processes.
In ClickHouse, as in most other SQL databases, query execution is split into several stages:
- Query parsing.
- Query analysis.
- Building the logical query plan.
- Optimizing the logical query plan.
- Building the physical query plan.
- Optimizing the physical query plan.
- Executing the physical query plan.
In most databases the logical query plan is a tree where each node is an operator of relational algebra, and the leaves of the tree are data sources, usually table scans.
It is convenient to describe query plan steps in terms of relational algebra. Relational algebra and its operators are well studied, and there are many known optimizations.
Predicate pushdown is one of them.
Predicate pushdown improves query performance by pushing filter conditions down to the operators that scan the data. Filtering earlier lets the later steps of the query plan process far less data, makes index usage more efficient, and in distributed databases greatly reduces the volume of data transferred between nodes.
Pushdown can be applied to most relational algebra operators, such as projections, aggregations, sorts, and joins. The most important case is pushing predicates down at the JOIN stage, simply because JOIN operators usually produce huge volumes of data.
Note that for some operators it is safe to push down only part of a predicate. In that case the predicate has to be split into parts. This is why it makes sense to store predicates in databases in conjunctive normal form, applying the pushdown optimization to each conjunct separately when needed.
Here is an example:
CREATE TABLE test_table_1 (id UInt64, value String) ENGINE=MergeTree ORDER BY id;
CREATE TABLE test_table_2 (id UInt64, value String) ENGINE=MergeTree ORDER BY id;
INSERT INTO test_table_1 SELECT number, number FROM numbers(150000000);
INSERT INTO test_table_2 SELECT number, number FROM numbers(150000000);
SELECT * FROM test_table_1 AS lhs
INNER JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE test_table_1.id = 5;
Elapsed: 22.130 sec.
Processed 150.01 million rows, 3.79 GB (6.78 million rows/s., 171.21 MB/s.)
Peak memory usage: 18.41 GiB.
Let's look at the ClickHouse logical query plan for this query:
EXPLAIN
SELECT * FROM test_table_1 AS lhs
INNER JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE test_table_1.id = 5
SETTINGS optimize_move_to_prewhere = 0
┌─explain──────────────────────────────────────────────────────────────────────┐
│ Expression ((Project names + (Projection + ))) │
│ Join (JOIN FillRightFirst) │
│ Filter (( + (JOIN actions + Change column names to column identifiers))) │
│ ReadFromMergeTree (default.test_table_1) │
│ Expression ((JOIN actions + Change column names to column identifiers)) │
│ ReadFromMergeTree (default.test_table_2) │
└──────────────────────────────────────────────────────────────────────────────┘
In this example the predicate is moved to the LEFT side of the JOIN. Note the SETTINGS optimize_move_to_prewhere = 0: without it, this filtering step would be turned into a PREWHERE for the left table.
Before ClickHouse 24.4, JOIN used a simpler version of the predicate pushdown optimization. Notably, it did not take equivalence classes of the joined columns into account (that is, columns that become equivalent after the JOIN executes).
PR #61216 introduces a more sophisticated predicate analysis that uses equivalence classes: it can rewrite predicates applied to one side of the JOIN as predicates that apply to the other side. The predicate is also split into parts, and only the safe parts are moved down when necessary.
Here is another example:
SELECT * FROM test_table_1 AS lhs
INNER JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE test_table_1.id = 5;
In this example the id column of test_table_1 is known to be equivalent to the id column of test_table_2, so the predicate test_table_1.id = 5 can be rewritten as test_table_2.id = 5 and pushed down to the right table.
Filter pushdown follows different logic for different JOIN types:
- For an INNER JOIN, all predicates can be pushed to either side of the JOIN. Predicates that use only equivalent columns can also be rewritten from the left side to the right and vice versa.
- For a LEFT/RIGHT JOIN, conditions that use only columns from the LEFT/RIGHT table can be pushed to the LEFT/RIGHT side of the JOIN. Predicates that use only equivalent columns can also be rewritten from the left side to the right for LEFT JOINs, and from the right side to the left for RIGHT JOINs.
Let's look at the query plan after this optimization is applied:
EXPLAIN
SELECT * FROM test_table_1 AS lhs
INNER JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE test_table_1.id = 5
SETTINGS optimize_move_to_prewhere = 0
┌─explain──────────────────────────────────────────────────────────────────────┐
│ Expression ((Project names + (Projection + ))) │
│ Join (JOIN FillRightFirst) │
│ Filter (( + (JOIN actions + Change column names to column identifiers))) │
│ ReadFromMergeTree (default.test_table_1) │
│ Filter (( + (JOIN actions + Change column names to column identifiers))) │
│ ReadFromMergeTree (default.test_table_2) │
└──────────────────────────────────────────────────────────────────────────────┘
Now the predicate is pushed to both the left and right sides of the JOIN. The performance improvement for INNER, LEFT and RIGHT JOINs is also visible in real numbers:
inner join before and after
SELECT * FROM test_table_1 AS lhs
INNER JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE lhs.id = 5
Before:
Elapsed: 22.130 sec
Processed 150.01 million rows, 3.79 GB (6.78 million rows/s, 171.21 MB/s)
Peak memory usage: 18.41 GiB
After:
Elapsed: 0.005 sec
Processed 16.38 thousand rows, 131.19 KB (3.21 million rows/s, 25.73 MB/s)
Peak memory usage: 579.28 KB
left join before and after
SELECT * FROM test_table_1 AS lhs
LEFT JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE lhs.id = 5;
Before:
Elapsed: 22.680 sec
Processed 150.01 million rows, 3.79 GB (6.61 million rows/s, 167.06 MB/s)
Peak memory usage: 18.42 GiB
After:
Elapsed: 0.005 sec
Processed 16.38 thousand rows, 131.19 KB (3.30 million rows/s, 26.45 MB/s)
Peak memory usage: 579.28 KB
Here are the full results of the ClickHouse performance tests, with the highlights below.
These changes made INNER and LEFT JOINs more than 180 times faster.
This optimization also fixes several related, long-standing issues in ClickHouse. A few examples:
Pull Request #2: Automatic conversion of OUTER JOIN to INNER JOIN
Another change lets ClickHouse automatically convert an OUTER JOIN into an INNER JOIN when the predicate after the JOIN filters out every non-joined row filled with default values.
This opens up more optimization opportunities: once an OUTER JOIN has become an INNER JOIN, predicate pushdown can be applied in more scenarios.
Using the table from the demo above:
query
SELECT * FROM test_table_1 AS lhs
LEFT JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE test_table_2.id = 5
Elapsed: 27.680 sec
Processed 300.00 million rows, 7.58 GB (10.84 million rows/s, 273.77 MB/s)
Peak memory usage: 18.46 GiB
Here is the ClickHouse logical query plan:
query plan
EXPLAIN actions = 1
SELECT * FROM test_table_1 AS lhs
LEFT JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE test_table_2.id = 5
┌─explain─────────────────────────────────────────────┐
│ Expression ((Project names + Projection)) │
│ Filter ((WHERE + DROP unused columns after JOIN)) │
│ Join (JOIN FillRightFirst) │
│ Type: LEFT │
│ Strictness: ALL │
│ Algorithm: HashJoin │
│ Clauses: [(__table1.id) = (__table2.id)] │
│ ... │
└─────────────────────────────────────────────────────┘
Note
With actions = 1 more details of the query plan become visible, such as the JOIN type, the specific actions that will be executed, and other details. Only the key part of the query plan is shown here, to confirm that this is a LEFT JOIN.
In this example, the predicate test_table_2.id = 5 will always filter out the unmatched rows with default values produced by the LEFT JOIN.
PR #62907 introduces an analysis that can automatically convert an OUTER JOIN into an INNER JOIN. This analysis determines that the predicate after the OUTER JOIN will always filter out non-joined rows with default values. When that holds, the OUTER JOIN can be converted to an INNER JOIN.
To do this, the analysis attempts constant folding of the predicate that executes after the OUTER JOIN: all right/left-side columns of a LEFT/RIGHT JOIN are replaced with columns that hold constant default values. If the constant-folded predicate evaluates to a constant False or NULL, the OUTER JOIN can be converted to an INNER JOIN.
Here is the query plan after this optimization is applied:
query plan after the optimization
EXPLAIN actions = 1
SELECT * FROM test_table_1 AS lhs
LEFT JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE test_table_2.id = 5
┌─explain────────────────────────────────────────┐
│ Expression ((Project names + (Projection + ))) │
│ Join (JOIN FillRightFirst) │
│ Type: INNER │
│ Strictness: ALL │
│ Algorithm: HashJoin │
│ Clauses: [(__table1.id) = (__table2.id)] │
│ ... │
└────────────────────────────────────────────────┘
The LEFT JOIN has already been replaced with an INNER JOIN. The Filter step has moved down, because for an INNER JOIN test_table_2.id = 5 can safely be pushed to both sides of the JOIN.
The original queries also run faster after the optimization:
performance before and after
SELECT * FROM test_table_1 AS lhs
LEFT JOIN test_table_2 AS rhs ON lhs.id = rhs.id
WHERE test_table_2.id = 5
Before:
Elapsed: 27.680 sec
Processed 300.00 million rows, 7.58 GB (10.84 million rows/s, 273.77 MB/s)
Peak memory usage: 18.46 GiB
After:
Elapsed: 0.004 sec
Processed 16.38 thousand rows, 131.19 KB (3.96 million rows/s, 31.74 MB/s)
Peak memory usage: 578.27 KB
The ClickHouse performance test results show large improvements for INNER, LEFT and RIGHT JOINs:
These changes substantially improved ClickHouse JOIN performance, in some cases by several orders of magnitude.
Conclusion
Significant performance gains in databases come from high-level logical optimizations applied over the query plan. Such optimizations work well together and can be combined for even greater gains, as these two pull requests show.
These two pull requests, available in ClickHouse 24.4, substantially improved JOIN performance in many production scenarios and should noticeably speed up queries for ClickHouse users.