How replication works: parts, replicas, Keeper and mutations

Replication in ClickHouse looks like magic from the outside: insert on one node, query on another. The mechanism is simple once you see which parts are shared and which are not. This article builds the mental model and maps each piece to the system table where you can see it. The statements need a replicated cluster, so they were parsed with clickhouse format on 26.4 and the columns checked against that version’s system tables; they were not run against a live cluster.

What is and is not shared

A replicated table is many independent tables that agree on a shared log. There is no shared storage. Every replica has its own copy of the data parts on its own disk. What they share is a small amount of coordination state in ClickHouse Keeper (or ZooKeeper; the protocol is the same and Keeper is the recommended server).

CREATE TABLE analytics.events_local ON CLUSTER my_cluster
(
    event_time DateTime, user_id UInt64, event_type LowCardinality(String),
    url String, duration_ms UInt32, country LowCardinality(String)
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events_local', '{replica}')
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, event_time);

The two engine arguments identify the table in Keeper: the first is the table path, the same on every replica of a shard; the second is the replica name, unique per replica. Both usually come from macros defined in each server’s config. Two replicas with the same table path belong to the same replicated table. Two shards must use different paths, which is what {shard} is for.

Keeper holds, under that table path, roughly:

Keeper node	Contents
log	The shared, ordered log of actions all replicas must perform: new part, merge, mutation, drop
replicas/<name>/queue	Each replica’s private to-do list, copied from log
replicas/<name>/parts	The names (and checksums) of the parts each replica has
block_numbers, blocks	Block number allocation per partition and insert deduplication hashes
mutations	Pending and finished mutations
leader_election	Who may schedule merges

You can browse it from SQL with system.zookeeper. The path condition is mandatory, and the table adds load on Keeper, so use it for diagnosis only:

SELECT name, value
FROM system.zookeeper
WHERE path = '/clickhouse/tables/01/events_local';

Keeper never stores your rows. It stores names: part names, log entries and counters. That is why a Keeper ensemble can serve thousands of tables with modest hardware, and why it is the first thing to look at when many tables stall at once.

Life of an INSERT

1. The client inserts into replica A. A writes the block to a new part on its own disk, e.g. 202601_5_5_0. The block number comes from Keeper, so it is unique across replicas.
2. A records the new part under its replicas/A/parts and adds a GET_PART entry to the shared log. If the insert carries a deduplication hash that Keeper already holds, the insert is recognised as a retry and dropped.
3. Replica B, watching the log, copies the entry into its own queue.
4. B executes the entry by fetching the part from A over HTTP. It does not redo the insert; it downloads the finished bytes and verifies the checksums.

After this, both replicas have a part with the same name and the same content. The name is the identity: replicas exchange names, not rows.

Life of a merge

Merges are decided once and executed everywhere. A replica that holds the right to schedule merges chooses source parts and writes a MERGE_PARTS entry into the log, naming the result part (say 202601_1_5_1). Each replica then executes that entry. If it already has all the source parts it merges them locally, which is cheaper than a download. If the result part already exists on another replica and a local merge would take too long or a source part is missing, it fetches the result instead. Either way, the end state is identical part names on every replica.

In current versions any replica can act as a scheduler, so you may see several replicas with is_leader = 1 at the same time. The setting replicated_can_become_leader = 0 keeps a replica from scheduling merges, but it still executes them.

Life of a mutation

A mutation is first recorded under mutations in Keeper with the block number it applies up to. Each replica then turns it into MUTATE_PART entries in its queue (this is what part_mutations_in_queue counts), one per part it must rewrite. The rewritten part gets a data-version suffix in its name (202601_1_5_1_9). Parts created by later inserts are not touched. A mutation is done on a replica when every relevant part has been rewritten, and is_done in system.mutations reports that for the replica you are asking. For the full story see mutations, lightweight DELETE and TTL.

SELECT mutation_id, command, parts_to_do, latest_fail_reason
FROM system.mutations
WHERE database = 'analytics' AND `table` = 'events_local' AND NOT is_done;

Reading system.replicas

Each Replicated* table on a server has one row in system.replicas: the local replica’s view of the table plus selected Keeper data.

SELECT * FROM system.replicas
WHERE database = 'analytics' AND `table` = 'events_local'
FORMAT Vertical;

Group the columns by question:

Is this replica able to work?

• is_readonly: the replica refuses inserts. It is set when Keeper is unreachable, when the table’s metadata in Keeper is missing, or while the session is being re-established.
• is_session_expired: the Keeper session is gone. Usually transient, persistent if Keeper is down.
• zookeeper_exception: the last error when reading from Keeper.

What is it still supposed to do?

• queue_size, inserts_in_queue, merges_in_queue, part_mutations_in_queue: how many log entries are waiting, split by kind. A queue that stays at zero or drains in seconds is healthy.
• queue_oldest_time, inserts_oldest_time, merges_oldest_time: when the oldest entry of each kind was queued. Age matters more than count.
• oldest_part_to_get, oldest_part_to_merge_to, oldest_part_to_mutate_to: the part names the oldest entries are producing.
• future_parts, parts_to_check: parts that will appear after queued actions, and parts being verified because something looks corrupt.

Is it keeping up with the log?

• log_max_index is the newest entry in the shared log; log_pointer is the next entry this replica will copy to its queue. A gap between them means the replica is not even reading the log.
• absolute_delay is the replica’s lag in seconds, last_queue_update when it last refreshed its queue.

Who else is there?

• total_replicas and active_replicas: registered replicas versus those holding a live Keeper session.
• replica_is_active: the same as a map from replica name to 0 or 1, which tells you which replica is down.
• lost_part_count: parts that were lost on all replicas since table creation; it only ever grows.

Some columns require extra Keeper round trips (log_max_index, log_pointer, total_replicas, active_replicas and friends), so selecting * is slower than naming what you need. A ready-made health check built on these columns is in this snippet.

The queue itself

system.replication_queue lists the pending entries one by one. When queue_size is stuck, look here for the reason:

SELECT database, `table`, replica_name, type, new_part_name, source_replica,
       create_time, num_tries, last_exception, num_postponed, postpone_reason
FROM system.replication_queue
WHERE database = 'analytics'
ORDER BY create_time
LIMIT 20;

type tells you what the entry is (GET_PART, MERGE_PARTS, MUTATE_PART, DROP_RANGE and others). last_exception explains a failure (typically a part that cannot be fetched because the source replica no longer has it), and postpone_reason explains why an entry is waiting rather than failing, for example because TTL merges hit their concurrency limit or because a needed part is not there yet.

Parts versus system.parts

system.parts is local: it lists the parts this server has. Comparing replicas is a matter of running the same query on all of them:

SELECT hostName() AS host, `table`, count() AS active_parts, sum(rows) AS rows
FROM clusterAllReplicas('my_cluster', system.parts)
WHERE database = 'analytics' AND `table` = 'events_local' AND active
GROUP BY host, `table`
ORDER BY host;

Matching active_parts and rows for the replicas of one shard means they converged. Row counts can differ transiently while inserts, merges and fetches are in flight. Part names can differ for a while too, since replicas may have merged different sources into different results before settling on the one chosen in the log.

Operations you will need

• SYSTEM SYNC REPLICA analytics.events_local blocks until this replica has executed everything currently in its queue. Use it before comparing counts.
• SYSTEM RESTART REPLICA re-initialises a table’s Keeper session without restarting the server.
• SYSTEM RESTORE REPLICA rebuilds a replica’s Keeper metadata from its local parts when the metadata was lost; the replica must be read-only for this to apply.
• SYSTEM DROP REPLICA 'name' FROM TABLE ... removes the Keeper metadata of a replica that no longer exists. It refuses to drop the replica you are connected to.
• detached parts hold data ClickHouse refused to use; see dropping detached parts.

Failure modes in one table

Symptom	Likely cause	Look at
Inserts fail with “table is in readonly mode”	Keeper session lost or metadata missing	is_readonly, zookeeper_exception, Keeper health
absolute_delay grows, queue_size grows	Fetches or merges slower than inserts	replication_queue, network, merge threads
Queue entry retried forever with “no replica has part”	The part was dropped or lost everywhere	last_exception, lost_part_count
log_pointer far below log_max_index	Replica is not reading the log	Keeper latency, last_queue_update_exception
Replica missing from replica_is_active	Its server is down or cut off from Keeper	That server’s log

Reference: data replication in the docs and system.replicas.