Architecture Overview

### 4.3 Multi-Node Parallelization[​](#4-3-multi-node-parallelization "Direct link to 4.3 Multi-Node Parallelization")

If the source table of a query is sharded, the query optimizer on the node that received the query (initiator node) tries to perform as much work as possible on other nodes. Results from other nodes can be integrated into different points of the query plan. Depending on the query, remote nodes may either 1. stream raw source table columns to the initiator node, 2. filter the source columns and send the surviving rows, 3. execute filter and aggregation steps and send local result groups with partial aggregation states, or 4. run the entire query including filters, aggregation, and sorting.

Node 2 ... N in [Figure 8](#page-7-1) show plan fragments executed on other nodes holding shards of the hits table. These nodes filter and group the local data and send the result to the initiator node. The GroupStateMerge operator on node 1 merges the local and remote results before the results groups are finally sorted.

### 4.4 Holistic Performance Optimization[​](#4-4-holistic-performance-optimization "Direct link to 4-4-holistic-performance-optimization")

This section presents selected key performance optimizations applied to different stages of query execution.

**Query optimization**. The first set of optimizations is applied on top of a semantic query representation obtained from the query's AST. Examples of such optimizations include constant folding (e.g. concat(lower('a'),upper('b')) becomes 'aB'), extracting scalars from certain aggregation functions (e.g. sum(a*2) becomes 2 \* sum(a)), common subexpression elimination, and transforming disjunctions of equality filters to IN-lists (e.g. x=c OR x=d becomes x IN (c,d)). The optimized semantic query representation is subsequently transformed to a logical operator plan. Optimizations on top of the logical plan include filter pushdown, reordering function evaluation and sorting steps, depending on which one is estimated to be more expensive. Finally, the logical query plan is transformed into a physical operator plan. This transformation can exploit the particularities of the involved table engines. For example, in the case of a MergeTree*-table engine, if the ORDER BY columns form a prefix of the primary key, the data can be read in disk order, and sorting operators can be removed from the plan. Also, if the grouping columns in an aggregation form a prefix of the primary key, ClickHouse can use sort aggregation [[33]](#page-12-21), i.e. aggregate runs of the same value in the pre-sorted inputs directly. Compared to hash aggregation, sort aggregation is significantly less memory-intensive, and the aggregate value can be passed to the next operator immediately after a run has been processed.

**Query compilation**. ClickHouse employs [query compilation based on LLVM](https://clickhou.se/jit) to dynamically fuse adjacent plan operators [[38,](#page-12-22) [53]](#page-13-0). For example, the expression a \* b + c + 1 can be combined into a single operator instead of three operators. Besides expressions, ClickHouse also employs compilation to evaluate multiple aggregation functions at once (i.e. for GROUP BY) and for sorting with more than one sort key. Query compilation decreases the number of virtual calls, keeps data in registers or CPU caches, and helps the branch predictor as less code needs to execute. Additionally, runtime compilation enables a rich set of optimizations, such as logical optimizations and peephole optimizations implemented in compilers, and gives access to the fastest locally available CPU instructions. The compilation is initiated only when the same regular, aggregation, or sorting expression is executed by different queries more than a configurable number of times. Compiled query operators are cached and can be reused by future queries.[7]

**Primary key index evaluation**. ClickHouse evaluates WHERE conditions using the primary key index if a subset of filter clauses in the condition's conjunctive normal form constitutes a prefix of the primary key columns. The primary key index is analyzed left-to-right on lexicographically sorted ranges of key values. Filter clauses corresponding to a primary key column are evaluated using ternary logic - they're all true, all false, or mixed true/false for the values in the range. In the latter case, the range is split into sub-ranges which are analyzed recursively. Additional optimizations exist for functions in filter conditions. First, functions have traits describing their monotonicity, e.g, toDayOfMonth(date) is piecewise monotonic within a month. Monotonicity traits allow to infer if a function produces sorted results on sorted input key value ranges. Second, some functions can compute the preimage of a given function result. This is used to replace comparisons of constants with function calls on the key columns by comparing the key column value with the preimage. For example, toYear(k) = 2024 can be replaced by k >= 2024-01-01 && k < 2025-01-01.

**Data skipping**. ClickHouse tries to avoid data reads at query runtime using the data structures presented in Section [3.2.](#page-3-0) Additionally, filters on different columns are evaluated sequentially in order of descending estimated selectivity based on heuristics and (optional) column statistics. Only data chunks that contain at least one matching row are passed to the next predicate. This gradually decreases the amount of read data and the number of computations to be performed from predicate to predicate. The optimization is only applied when at least one highly selective predicate is present; otherwise, the latency of the query would deteriorate compared to an evaluation of all predicates in parallel.

**Hash tables**. Hash tables are fundamental data structures for aggregation and hash joins. Choosing the right type of hash table is critical to performance. ClickHouse [instantiates](https://clickhou.se/hashtables) various hash tables (over 30 as of March 2024) from a generic hash table template with the hash function, allocator, cell type, and resize policy as variation points. Depending on the data type of the grouping columns, the estimated hash table cardinality, and other factors, the fastest hash table is selected for each query operator individually. Further optimizations implemented for hash tables include:

- a two-level layout with 256 sub-tables (based on the first byte of the hash) to support huge key sets,
- string hash tables [[79]](#page-13-20) with four sub-tables and different hash functions for different string lengths,
- lookup tables which use the key directly as bucket index (i.e. no hashing) when there are only few keys,
- values with embedded hashes for faster collision resolution when comparison is expensive (e.g. strings, ASTs),
- creation of hash tables based on predicted sizes from runtime statistics to avoid unnecessary resizes,
- allocation of multiple small hash tables with the same creation/destruction lifecycle on a single memory slab,
- instant clearing of hash tables for reuse using per-hash-map and per-cell version counters,
- usage of CPU prefetch (\_\_builtin\_prefetch) to speed up the retrieval of values after hashing the key.

**Joins**. As ClickHouse originally supported joins only rudimentarily, many use cases historically resorted to denormalized tables. Today, the database [offers](https://clickhou.se/joins) all join types available in SQL (inner, left- /right/full outer, cross, as-of), as well as different join algorithms such as hash join (naïve, grace), sort-merge join, and index join for table engines with fast key-value lookup (usually dictionaries).