Chapter 6: Replication and ZooKeeper Coordination #

Welcome back! In Chapter 5: Background Merges, Mutations, and Compaction Selection, you saw how a single MergeTree table tidies itself up with merges and mutations. But what if the table lives on several machines? What if one machine is offline? How do all the machines agree on which data parts exist?

That is the job of Replication and ZooKeeper Coordination.

Think of a team of chefs cooking the same recipe in different kitchens. They cannot see each other, but they all keep checking the same shared whiteboard. Whenever one chef makes progress, they write it on the whiteboard. The other chefs copy the note and do the same work in their own kitchen.

In ClickHouse, the whiteboard is ZooKeeper. The chefs are replicas.

The Use Case: One Insert, Many Replicas #

Imagine you have two replicas of a weather table:

CREATE TABLE weather
(
    city       String,
    temp_c     Int32,
    event_date Date
)
ENGINE = ReplicatedMergeTree('/clickhouse/weather', '{replica}')
PARTITION BY toYYYYMM(event_date)
ORDER BY (event_date, city);

Now you insert one row into Replica A:

INSERT INTO weather VALUES ('Berlin', 21, '2026-01-01');

A moment later, you can run a SELECT on Replica B and see that same row:

SELECT city, temp_c FROM weather;

Output on Replica B:

Berlin  21

But no direct network connection was made from Replica A to Replica B. How did Replica B know about the new row?

The answer: both replicas coordinate through ZooKeeper.

Key Concept 1: ZooKeeper Is a Shared Whiteboard #

ZooKeeper stores small named data nodes called znodes. For a ReplicatedMergeTree table, ZooKeeper holds a small tree of useful paths.

A typical tree looks like this:

/clickhouse/weather/
├── log/
├── mutations/
├── block_numbers/
└── replicas/
    ├── replica-1/queue/
    ├── replica-1/is_active
    ├── replica-2/queue/
    └── replica-2/is_active

Let’s decode these paths:

• log/ — the shared work log. Every replica writes work items here.
• mutations/ — records of ALTER TABLE mutations that need block numbers.
• block_numbers/ — counters that hand out unique block numbers, one set per partition.
• replicas/<name>/queue/ — a replica’s local to-do list.
• replicas/<name>/is_active — an ephemeral znodes that says “this replica is alive right now.”

A znode is just a tiny data object. Some znodes are ephemeral, which means they disappear if the replica that created them loses its ZooKeeper connection. That is how ZooKeeper knows a replica has gone offline.

Key Concept 2: Work Items Are ReplicatedMergeTreeLogEntry Objects #

Every action that must happen on all replicas becomes a work item called a log entry.

The important types are:

• GET_PART — “copy this data part from another replica.”
• MERGE_PARTS — “merge these parts into one bigger part.”
• MUTATE_PART — “apply a mutation to this part.”
• DROP_RANGE — “delete parts in a certain range.”

A simplified log entry looks like this:

struct ReplicatedMergeTreeLogEntry
{
    enum Type { GET_PART, MERGE_PARTS, MUTATE_PART, DROP_RANGE };
    Type type;
    String new_part_name;
    Strings source_parts;
};

This code is based on ReplicatedMergeTreeLogEntry.h. An entry knows what to do and which parts are involved.

Log entries can be serialized to text, stored in ZooKeeper, and read back:

void ReplicatedMergeTreeLogEntryData::writeText(WriteBuffer & out) const;
void ReplicatedMergeTreeLogEntryData::readText(ReadBuffer & in, MergeTreeDataFormatVersion v);

So when Replica A commits a new insert, it writes one of these entries into log/. Every other replica watches log/, copies the entry into its own queue, and executes it.

Key Concept 3: Mutations Use Block Numbers #

Mutations are ALTER TABLE UPDATE or DELETE commands. They are special because they can affect many partitions and many data parts at once.

A mutation is stored as a ReplicatedMergeTreeMutationEntry:

struct ReplicatedMergeTreeMutationEntry
{
    String znode_name;
    time_t create_time = 0;
    String source_replica;
    std::map<String, Int64> block_numbers;
    MutationCommands commands;
};

The key part is block_numbers. This is a map from partition_id to block_number.

Why do mutations need block numbers?

Imagine a mutation that says “change all temperatures before March.” Each partition has its own list of parts. The mutation needs to know which parts are old enough to be changed. A part’s max_block tells how new it is. If a part’s max_block is less than or equal to the mutation’s block number for that partition, the part must be mutated.

This is how a mutation entry might look as text:

format version: 1
source replica: replica-1
block numbers count: 1
202601  42
commands: UPDATE temp_c = ...

The block number 42 is the boundary. Any part in partition 202601 with block number <= 42 must be mutated.

Key Concept 4: Ephemeral Locks Serialize Block Numbers #

How do replicas get those unique block numbers? They use an ephemeral lock.

A replica creates an ephemeral sequential znode in ZooKeeper:

EphemeralLockInZooKeeper createEphemeralLockInZooKeeper(...)
{
    String path = zookeeper->create(prefix, holder,
        zkutil::CreateMode::EphemeralSequential);
    UInt64 number = parseSequentialNodeNumber(path, prefix.size());
    return {prefix, zookeeper, path, number};
}

ZooKeeper appends a unique sequence number to the znode path. That sequence number becomes the new block number.

Why EphemeralSequential?

• Sequential means each create gets the next number, so no two replicas can get the same block number.
• Ephemeral means if the replica crashes, the lock znode disappears automatically. No one is stuck waiting forever.

For mutations, ClickHouse needs to lock block numbers in several partitions at once. It uses a helper called EphemeralLocksInAllPartitions:

struct LockInfo
{
    String path;
    String partition_id;
    UInt64 number = 0;
};

This makes sure the mutation gets a consistent snapshot of block numbers across all partitions.

Key Concept 5: Every Replica Has a Queue #

The shared log/ is like a newspaper. Every replica reads the same articles.

But each replica also has its own local to-do list in replicas/<name>/queue/. It copies log entries into this queue and then executes them one by one.

The queue tracks how execution is going:

bool currently_executing = false;
size_t num_tries = 0;
std::exception_ptr exception;

If an entry fails, the queue remembers the exception and tries again later. If an entry is affected by a DROP_RANGE, the queue can postpone it. There is even a helper called DropPartsRanges for that:

class DropPartsRanges
{
public:
    bool isAffectedByDropPart(const ReplicatedMergeTreeLogEntry & entry,
                              std::string & postpone_reason) const;
    void addDropPart(const ReplicatedMergeTreeLogEntryPtr & entry);
};

Because every replica executes the same log entries, they all move toward the same set of active parts.

Key Concept 6: Attach and Restart Threads Help Replicas Rejoin #

Replicas are not always online. They may restart, lose network connections, or have ZooKeeper sessions expire.

ClickHouse has two background threads to help.

ReplicatedMergeTreeAttachThread #

This thread runs when the table starts. It:

1. Checks whether the table metadata exists in ZooKeeper.
2. Creates missing ZooKeeper nodes.
3. Checks local parts.
4. Starts the normal background threads.

If ZooKeeper is not available yet, it retries later:

void ReplicatedMergeTreeAttachThread::run()
{
    try { runImpl(); finalizeInitialization(); }
    catch (...) { task->scheduleAfter(retry_period * 1000); }
}

ReplicatedMergeTreeRestartingThread #

This thread runs continuously. It watches the ZooKeeper session. If the session expires, it reinitializes everything:

class ReplicatedMergeTreeRestartingThread
{
public:
    void start(bool schedule);
    void shutdown(bool part_of_full_shutdown);
    void run();
private:
    StorageReplicatedMergeTree & storage;
    bool runImpl();
};

It also writes the ephemeral is_active node. That is how other replicas know this replica is alive.

Solving the Use Case Step by Step #

Let’s follow our insert from Replica A to Replica B.

Step by step:

1. A client sends INSERT to Replica A.
2. Replica A asks ZooKeeper for a new block number. ZooKeeper gives 42.
3. Replica A writes a temporary data part, just like in Chapter 2: Write Path and Part Creation.
4. Replica A writes a log entry to log/ saying: “here is a part named 202601_42_42_0.”
5. Replica A executes its own entry and makes the part visible locally.
6. Replica B is watching log/. It sees the new entry.
7. Replica B copies the entry into its queue.
8. Replica B fetches the part from Replica A and makes it active.

Now both replicas have the same active data part.

Mutations Follow a Similar Pattern #

Mutations are slightly more complex because they use block numbers and two kinds of ZooKeeper entries.

First, a mutation entry is stored in /mutations. Then, for each affected part, a MUTATE_PART entry is appended to the shared log. Every replica sees those log entries and rewrites the affected parts.

Under the Hood: A Small Code Tour #

Let’s peek at the real files behind these ideas. We will simplify the code, but keep the important names.

ReplicatedMergeTreeLogEntry.h #

The log entry is the heart of replication. It is small and serializable by design.

struct ReplicatedMergeTreeLogEntry
{
    enum Type { GET_PART, MERGE_PARTS, MUTATE_PART, DROP_RANGE };
    Type type;
    String new_part_name;
    Strings source_parts;
};

The real struct also stores source_replica, quorum, deduplication_block_ids, and execution stats like num_tries and currently_executing.

ReplicatedMergeTreeMutationEntry.h #

A mutation is a separate record. It is not written directly into the log. Instead, it is stored in /mutations, and later converted into MUTATE_PART log entries.

struct ReplicatedMergeTreeMutationEntry
{
    String znode_name;
    time_t create_time = 0;
    String source_replica;
    std::map<String, Int64> block_numbers;
    MutationCommands commands;
};

The block_numbers map is the most interesting part. It tells every replica exactly where the mutation boundary lies.

EphemeralLockInZooKeeper.cpp #

Block numbers are allocated with ephemeral sequential znodes.

EphemeralLockInZooKeeper createEphemeralLockInZooKeeper(...)
{
    String path = zookeeper->create(prefix, holder,
        zkutil::CreateMode::EphemeralSequential);
    UInt64 number = parseSequentialNodeNumber(path, prefix.size());
    return {prefix, zookeeper, path, number};
}

This tiny function is how “one replica asks for the next block number” works in real code.

ReplicatedMergeTreeAttachThread.cpp #

When a replica first starts, the attach thread tries to initialize the table. If ZooKeeper is not ready, it schedules another try.

void ReplicatedMergeTreeAttachThread::run()
{
    try { runImpl(); finalizeInitialization(); }
    catch (...) { task->scheduleAfter(retry_period * 1000); }
}

Once initialization succeeds, all normal background threads can start.

ReplicatedMergeTreeRestartingThread.h #

The restarting thread keeps the replica healthy during its lifetime.

class ReplicatedMergeTreeRestartingThread
{
public:
    void start(bool schedule);
    void shutdown(bool part_of_full_shutdown);
    void run();
private:
    StorageReplicatedMergeTree & storage;
    bool runImpl();
};

If the ZooKeeper session expires, this thread notices and restarts the replica.

ReplicatedMergeTreeSink.h #

The ReplicatedMergeTreeSink is the entry point for inserted data. It is similar to the MergeTreeSink from Chapter 2, but it also commits to ZooKeeper.

class ReplicatedMergeTreeSink : public SinkToStorage
{
public:
    void consume(Chunk & chunk) override;
    void onFinish() override;
private:
    StorageReplicatedMergeTree & storage;
};

It writes a temporary part, then calls commitPart, which creates the ZooKeeper log entry.

Why This Design Is Powerful #

The design gives us three huge benefits:

1. No direct replica-to-replica metadata communication. Replicas only talk to ZooKeeper. They do not need to know each other’s IP addresses for coordination.

2. Automatic catch-up. If a replica is offline for a while, it misses several log entries. When it comes back, it reads the log from where it stopped and catches up.

3. Convergence. Since every replica executes the same log entries, they all converge to the same active data part set. Even if one replica is temporarily behind, it eventually catches up.

Summary #

In this chapter, you learned:

• ReplicatedMergeTree uses ZooKeeper as a shared coordination board.
• The shared log/ contains work items called ReplicatedMergeTreeLogEntry.
• Each replica copies log entries into its own queue and executes them.
• Mutations are stored as ReplicatedMergeTreeMutationEntry objects with per-partition block numbers.
• Ephemeral locks allocate block numbers safely and automatically clean up if a replica dies.
• ReplicatedMergeTreeAttachThread and ReplicatedMergeTreeRestartingThread help replicas rejoin after disconnects.
• All replicas converge to the same active data part set.

Replication may sound complex, but the core idea is simple: everyone reads the same whiteboard, writes down the same tasks, and does them on their own machine.

You have now completed the MergeTree tutorial! You have seen how data parts are born, how queries find them, how indexes make them fast, how background work reshapes them, and how multiple replicas stay in sync. Happy querying!

Generated by AI Codebase Knowledge Builder