Architecture Overview

**TTL (time-to-live) merges** provide aging for historical data. Unlike deleting and aggregating merges, TTL merges process only one part at a time. TTL merges are defined in terms of rules with triggers and actions. A trigger is an expression computing a timestamp for every row, which is compared against the time at which the TTL merge runs. While this allows users to control actions at row granularity, we found it sufficient to check whether all rows satisfy a given condition and run the action on the entire part. Possible actions include 1. move the part to another volume (e.g. cheaper and slower storage), 2. re-compress the part (e.g. with a more heavy-weight codec), 3. delete the part, and 4. roll-up, i.e. aggregate the rows using a grouping key and aggregate functions.

As an example, consider the logging table definition in [Listing 1.](#page-4-2) ClickHouse will move parts with timestamp column values older than one week to slow but inexpensive S3 object storage.

```
1 CREATE TABLE tab ( ts DateTime , msg String )
2 ENGINE MergeTree PRIMARY KEY ts
3 TTL ( ts + INTERVAL 1 WEEK ) TO VOLUME 's3 '
```

Listing 1: Move part to object storage after one week.

### 3.4 Updates and Deletes[​](#3-4-updates-and-deletes "Direct link to 3-4-updates-and-deletes")

The design of the MergeTree\* table engines favors append-only workloads, yet some use cases require to modify existing data occasionally, e.g. for regulatory compliance. Two approaches for updating or deleting data exist, neither of which block parallel inserts.

**Mutations** rewrite all parts of a table in-place. To prevent a table (delete) or column (update) from doubling temporarily in size, this operation is non-atomic, i.e. parallel SELECT statements may read mutated and non-mutated parts. Mutations guarantee that the data is physically changed at the end of the operation. Delete mutations are still expensive as they rewrite all columns in all parts.

As an alternative, **lightweight deletes** only update an internal bitmap column, indicating if a row is deleted or not. ClickHouse amends SELECT queries with an additional filter on the bitmap column to exclude deleted rows from the result. Deleted rows are physically removed only by regular merges at an unspecified time in future. Depending on the column count, lightweight deletes can be much faster than mutations, at the cost of slower SELECTs.

Update and delete operations performed on the same table are expected to be rare and serialized to avoid logical conflicts.

### 3.5 Idempotent Inserts[​](#3-5-idempotent-inserts "Direct link to 3-5-idempotent-inserts")

A problem that frequently occurs in practice is how clients should handle connection timeouts after sending data to the server for insertion into a table. In this situation, it is difficult for clients to distinguish between whether the data was successfully inserted or not. The problem is traditionally solved by re-sending the data from the client to the server and relying on primary key or unique constraints to reject duplicate inserts. Databases perform the required point lookups quickly using index structures based on binary trees [[39,](#page-12-15) [68]](#page-13-16), radix trees [[45]](#page-13-17), or hash tables [[29]](#page-12-16). Since these data structures index every tuple, their space and update overhead becomes prohibitive for large data sets and high ingest rates.

ClickHouse provides a more light-weight alternative based on the fact that each insert eventually creates a part. More specifically, the server maintains hashes of the N last inserted parts (e.g. N=100) and ignores re-inserts of parts with a known hash. Hashes for non-replicated and replicated tables are stored locally, respectively, in Keeper. As a result, inserts become idempotent, i.e. clients can simply re-send the same batch of rows after a timeout and assume that the server takes care of deduplication. For more control over the deduplication process, clients can optionally provide an insert token that acts as a part hash. While hash-based deduplication incurs an overhead associated with hashing the new rows, the cost of storing and comparing hashes is negligible.

### 3.6 Data Replication[​](#3-6-data-replication "Direct link to 3-6-data-replication")

Replication is a prerequisite for high availability (tolerance against node failures), but also used for load balancing and zero-downtime upgrades [[14]](#page-12-17). In ClickHouse, replication is based on the notion of table states which consist of a set of table parts (Section [3.1)](#page-2-2) and table metadata, such as column names and types. Nodes advance the state of a table using three operations: 1. Inserts add a new part to the state, 2. merges add a new part and delete existing parts to/from the state, 3. mutations and DDL statements add parts, and/or delete parts, and/or change table metadata, depending on the concrete operation. Operations are performed locally on a single node and recorded as a sequence of state transition in a global replication log.

The replication log is maintained by an ensemble of typically three ClickHouse Keeper processes which use the Raft consensus algorithm [[59]](#page-13-4) to provide a distributed and fault-tolerant coordination layer for a cluster of ClickHouse nodes. All cluster nodes initially point to the same position in the replication log. While the nodes execute local inserts, merges, mutations, and DDL statements, the replication log is replayed asynchronously on all other nodes. As a result, replicated tables are only eventually consistent, i.e. nodes can temporarily read old table states while converging towards the latest state. Most aforementioned operations can alternatively be executed synchronously until a quorum of nodes (e.g. a majority of nodes or all nodes) adopted the new state.

As an example, [Figure 6](#page-5-3) shows an initially empty replicated table in a cluster of three ClickHouse nodes. Node 1 first receives two insert statements and records them ( 1 2 ) in the replication log stored in the Keeper ensemble. Next, Node 2 replays the first log entry by fetching it ( 3 ) and downloading the new part from Node 1 ( 4 ), whereas Node 3 replays both log entries ( 3 4 5 6 ). Finally, Node 3 merges both parts to a new part, deletes the input parts, and records a merge entry in the replication log ( 7 ).

Figure 6: Replication in a cluster of three nodes.

Three optimizations to speed up synchronization exist: First, new nodes added to the cluster do replay the replication log from scratch, instead they simply copy the state of the node which wrote the last replication log entry. Second, merges are replayed by repeating them locally or by fetching the result part from another node. The exact behavior is configurable and allows to balance CPU consumption and network I/O. For example, cross-data-center replication typically prefers local merges to minimize operating costs. Third, nodes replay mutually independent replication log entries in parallel. This includes, for example, fetches of new parts inserted consecutively into the same table, or operations on different tables.

### 3.7 ACID Compliance[​](#3-7-acid-compliance "Direct link to 3-7-acid-compliance")

To maximize the performance of concurrent read and write operations, ClickHouse avoids latching as much as possible. Queries are executed against a snapshot of all parts in all involved tables created at the beginning of the query. This ensures that new parts inserted by parallel INSERTs or merges (Section [3.1)](#page-2-2) don't participate in execution. To prevent parts from being modified or removed simultaneously (Section [3.4)](#page-4-0), the reference count of the processed parts is incremented for the duration of the query. Formally, this corresponds to snapshot isolation realized by an MVCC variant [[6]](#page-12-18) based on versioned parts. As a result, statements are generally not ACID-compliant except for the rare case that concurrent writes at the time the snapshot is taken each affect only a single part.