Architecture Overview

In practice, most of ClickHouse's write-heavy decision making use cases even tolerate a small risk of losing new data in case of a power outage. The database takes advantage of this by not forcing a commit (fsync) of newly inserted parts to disk by default, allowing the kernel to batch writes at the cost of forgoing atomicity.

## 4 QUERY PROCESSING LAYER[​](#4-query-processing-layer "Direct link to 4-query-processing-layer")

Figure 7: Parallelization across SIMD units, cores and nodes.

As illustrated by [Figure 7,](#page-6-1) ClickHouse parallelizes queries at the level of data elements, data chunks, and table shards. Multiple data elements can be processed within operators at once using SIMD instructions. On a single node, the query engine executes operators simultaneously in multiple threads. ClickHouse uses the same vectorization model as MonetDB/X100 [[11]](#page-12-0), i.e. operators produce, pass, and consume multiple rows (data chunks) instead of single rows to minimize the overhead of virtual function calls. If a source table is split into disjoint table shards, multiple nodes can scan the shards simultaneously. As a result, all hardware resources are fully utilized, and query processing can be scaled horizontally by adding nodes and vertically by adding cores.

The rest of this section first describes parallel processing at data element, data chunk, and shard granularity in more detail. We then present selected key optimizations to maximize query performance. Finally, we discuss how ClickHouse manages shared system resources in the presence of simultaneous queries.

### 4.1 SIMD Parallelization[​](#4-1-simd-parallelization "Direct link to 4.1 SIMD Parallelization")

Passing multiple rows between operators creates an opportunity for vectorization. Vectorization is either based on manually written intrinsics [[64,](#page-13-18) [80]](#page-13-19) or compiler auto-vectorization [[25]](#page-12-19). Code that benefits from vectorization is compiled into different compute kernels. For example, the inner hot loop of a query operator can be implemented in terms of a non-vectorized kernel, an auto-vectorized AVX2 kernel, and a manually vectorized AVX-512 kernel. The fastest kernel is [chosen at runtime](https://clickhou.se/cpu-dispatch) based on the cpuid instruction. This approach allows ClickHouse to run on systems as old as 15 years (requiring SSE 4.2 as a minimum), while still providing significant speedups on recent hardware.

### 4.2 Multi-Core Parallelization[​](#4-2-multi-core-parallelization "Direct link to 4.2 Multi-Core Parallelization")

Figure 8: A physical operator plan with three lanes.

ClickHouse follows the conventional approach [[31]](#page-12-13) of transforming SQL queries into a directed graph of physical plan operators. The input of the operator plan is represented by special source operators that read data in the native or any of the supported 3rd-party formats (see Section [5)](#page-9-0). Likewise, a special sink operator converts the result into the desired output format. The physical operator plan is unfolded at query compilation time into independent execution lanes based on a configurable maximum number of worker threads (by default, the number of cores) and the source table size. Lanes decompose the data to be processed by parallel operators into non-overlapping ranges. To maximize the opportunity for parallel processing, lanes are merged as late as possible.

As an example, the box for Node 1 in [Figure 8](#page-7-1) shows the operator graph of a typical OLAP query against a table with page impression statistics. In the first stage, three disjoint ranges of the source table are filtered simultaneously. A Repartition exchange operator dynamically routes result chunks between the first and second stages to keep the processing threads evenly utilized. Lanes may become imbalanced after the first stage if the scanned ranges have significantly different selectivities. In the second stage, the rows that survived the filter are grouped by RegionID. The Aggregate operators maintain local result groups with RegionID as a grouping column and a per-group sum and count as a partial aggregation state for avg(). The local aggregation results are eventually merged by a GroupStateMerge operator into a global aggregation result. This operator is also a pipeline breaker, i.e., the third stage can only start once the aggregation result has been fully computed. In the third stage, the result groups are first divided by a Distribute exchange operator into three equally large disjoint partitions, which are then sorted by AvgLatency. Sorting is performed in three steps: First, ChunkSort operators sort the individual chunks of each partition. Second, StreamSort operators maintain a local sorted result which is combined with incoming sorted chunks using 2-way merge sorting. Finally, a MergeSort operator combines the local results using k-way sorting to obtain the final result.

Operators are state machines and connected to each other via input and output ports. The three possible states of an operator are need-chunk, ready, and done. To move from need-chunk to ready, a chunk is placed in the operator's input port. To move from ready to done, the operator processes the input chunk and generates an output chunk. To move from done to need-chunk, the output chunk is removed from the operator's output port. The first and third state transitions in two connected operators can only be performed in a combined step. Source operators (sink operators) only have states ready and done (need-chunk and done).

Worker threads continuously traverse the physical operator plan and perform state transitions. To keep CPU caches hot, the plan contains hints that the same thread should process consecutive operators in the same lane. Parallel processing happens both horizontally across disjoint inputs within a stage (e.g. in [Figure 8](#page-7-1) the Aggregate operators are executed concurrently) and vertically across stages not separated by pipeline breakers (e.g. in [Figure 8](#page-7-1) the Filter and Aggregate operator in the same lane can run simultaneously). To avoid over and undersubscription when new queries start, or concurrent queries finish, the degree of parallelism can be changed midquery between one and the maximum number of worker threads for the query specified at query start (see Section [4.5)](#page-9-1).

Operators can further affect query execution at runtime in two ways. First, operators can dynamically create and connect new operators. This is mainly used to switch to external aggregation, sort, or join algorithms instead of canceling a query when the memory consumption exceeds a configurable threshold. Second, operators can request worker threads to move into an asynchronous queue. This provides more effective use of worker threads when waiting for remote data.

ClickHouse's query execution engine and morsel-driven parallelism [[44]](#page-12-20) are similar in that lanes are normally executed on different cores / NUMA sockets and that worker threads can steal tasks from other lanes. Also, there is no central scheduling component; instead, worker threads select their tasks individually by continuously traversing the operator plan. Unlike morsel-driven parallelism, ClickHouse bakes the maximum degree of parallelism into the plan and uses much bigger ranges to partition the source table compared to default morsel sizes of ca. 100.000 rows. While this may in some cases cause stalls (e.g. when the runtime of filter operators in different lanes differ vastly) we find that liberal use of exchange operators such as Repartition at least avoids such imbalances from accumulating across stages.

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