Opensource internals

How ClickHouse Replicates MergeTree: A ZooKeeper Log and a Hash Deduplication

16 min read Jay Hubbert

ReplicatedMergeTree keeps an ordered text log in ZooKeeper, copies it into a per-replica queue, executes the entries out of order, and deduplicates retried inserts with a content hash checked in the same atomic multi as the block-number allocation.

TL;DR: ReplicatedMergeTree is a log in ZooKeeper plus a set of workers that execute the log’s entries. The log is a directory of sequential znodes holding text; each replica copies the log into its own queue and drains it out of order. The interesting part is the insert path: the block number is allocated by an ephemeral-sequential znode, and in the same atomic multi the insert checks whether the block’s content hash already exists. If it does, the duplicate rows are removed and the part is rewritten.

Replication is a distributed systems problem wearing a database’s clothes. Three replicas of a table must end up with the same data. The network partitions. A replica goes down mid-merge. The client retries an insert because it did not see the acknowledgement.

The requirements that fall out of this:

  1. A total order. The replicas must agree on what happened, not just on the final state — otherwise a merge on one replica and a drop on another produce two different tables.
  2. Idempotence. Any operation may be executed twice, and the second execution must be a no-op.
  3. Convergence without a coordinator. No replica may be the one that decides what the others do.

ClickHouse’s answer to (1) is a ZooKeeper log. Its answer to (2) is a content hash. Its answer to (3) is that every replica reads the same log and decides independently what it can execute right now.

The log

The log is a directory of sequential znodes under /log. Each znode is named log-0000000001, log-0000000002, … and its content is a text blob describing one operation:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeLogEntry.h:27
    enum Type
    {
        EMPTY,          /// Not used.
        GET_PART,       /// Get the part from another replica.
        ATTACH_PART,    /// Attach the part, possibly from our own replica (if found in /detached folder).
                        /// You may think of it as a GET_PART with some optimizations as they're nearly identical.
        MERGE_PARTS,    /// Merge the parts.
        DROP_RANGE,     /// Delete the parts in the specified partition in the specified number range.
        CLEAR_COLUMN,   /// NOTE: Deprecated. Drop specific column from specified partition.
        CLEAR_INDEX,    /// NOTE: Deprecated. Drop specific index from specified partition.
        REPLACE_RANGE,  /// Drop certain range of parts and replace them by new ones
        MUTATE_PART,    /// Apply one or several mutations to the part.
        ALTER_METADATA, /// Apply alter modification according to global /metadata and /columns paths
        SYNC_PINNED_PART_UUIDS, /// Synchronization point for ensuring that all replicas have up to date in-memory state.
        CLONE_PART_FROM_SHARD,  /// Clone part from another shard.
        DROP_PART,      /// NOTE: Virtual (has the same (de)serialization format as DROP_RANGE). Deletes the specified part.
    };

The log entry is human-readable, and it is versioned:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeLogEntry.cpp:77
    out << "format version: " << format_version << "\n"
        << "create_time: " << LocalDateTime(create_time ? create_time : time(nullptr), DateLUT::serverTimezoneInstance()) << "\n"
        << "source replica: " << source_replica << '\n'
        << "block_id: " << escape << block_hashes_string << '\n';

with a version enum that grows as features are added:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeLogEntry.cpp:26
enum FormatVersion : UInt8
{
    FORMAT_WITH_CREATE_TIME = 2,
    FORMAT_WITH_BLOCK_ID = 3,
    FORMAT_WITH_DEDUPLICATE = 4,
    FORMAT_WITH_UUID = 5,
    FORMAT_WITH_DEDUPLICATE_BY_COLUMNS = 6,
    FORMAT_WITH_LOG_ENTRY_ID = 7,

    FORMAT_LAST = 8,
};

This matters because the log is shared between replicas of different versions during a rolling upgrade. An old replica must be able to read a new replica’s entries (at least the subset it understands), and a new replica must be able to read an old replica’s entries. A text format with an explicit version number, where unknown trailing fields are skipped, is far easier to evolve than a binary one.

Copying the log into a queue

A replica does not execute the shared log directly. It copies the entries into its own queue, under /replicas/<name>/queue/, and remembers how far it has read with a log_pointer. The copy is a PersistentSequential create, so the queue entries are numbered in the order they were copied:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeQueue.cpp:938
                ops.emplace_back(zkutil::makeCreateRequest(
                    fs::path(replica_path) / "queue/queue-", res.data, zkutil::CreateMode::PersistentSequential));

and the pointer is advanced in the same multi, so the copy and the pointer move are atomic:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeQueue.cpp:953
            ops.emplace_back(zkutil::makeSetRequest(
                fs::path(replica_path) / "log_pointer", toString(last_entry_index + 1), -1));

Why copy at all? Because a replica needs a private queue: entries it has already executed are removed from it, entries it cannot execute yet (because they conflict with a running merge) stay in it, and the shared log is append-only for everyone. The queue is the replica’s private “to-do list”, derived from the shared “history”.

The copy happens in batches, and the batch size grows exponentially so that a replica that has been offline for a long time catches up with a handful of round trips instead of thousands:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeQueue.cpp:903
            /// Increase the batch size exponentially, so it will saturate to MAX_MULTI_OPS.
            if (current_multi_batch_size < MAX_MULTI_OPS)
                current_multi_batch_size = std::min<size_t>(MAX_MULTI_OPS, current_multi_batch_size * 2);

Executing out of order

Here is the part that surprises people. The queue is a queue, but the execution order is not the queue order:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeQueue.h:96
    /** The queue of what you need to do on this line to catch up. It is taken from ZooKeeper (/replicas/me/queue/).
      * In ZK records in chronological order. Here they are executed in parallel and reorder after entry execution.
      * Order of execution is not "queue" at all. Look at selectEntryToProcess.
      */
    Queue queue;

The selection walks the queue and takes the first entry that is currently executable:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeQueue.cpp:2202
    for (auto it = queue.begin(); it != queue.end(); ++it)
    {
        if ((*it)->currently_executing)
            continue;

        if (shouldExecuteLogEntry(**it, (*it)->postpone_reason, merger_mutator, data, committing_blocks, lock))
        {
            entry = *it;
            /// We gave a chance for the entry, move it to the tail of the queue, after that
            /// we move it to the end of the queue.
            queue.splice(queue.end(), queue, it);
            break;
        }

        ++(*it)->num_postponed;
        (*it)->last_postpone_time = time(nullptr);
    }

Two details are worth pausing on.

The entry is moved to the tail when it is selected. So a failing entry goes to the back of the line and does not block the entries behind it. This is the difference between “one bad merge stalls the whole replica” and “one bad merge is retried while the rest of the queue drains”.

“Currently executable” is a real predicate. An entry may be postponed because:

  • it failed recently, and there is an exponential backoff on it;
  • it produces a part that a currently executing entry also produces (isCoveredByFuturePartsImpl);
  • it intersects a DROP_RANGE/REPLACE_RANGE intent;
  • it produces a part that is going to be dropped anyway — “Optimization: it does not really make sense to generate parts that are going to be dropped anyway”;
  • the fetch pool is saturated.

The invariant that makes all of this safe is spelled out in the header:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeQueue.h:90
    /// A set of parts that should be on this replica according to the queue entries that have been done
    /// up to this point. The invariant holds: `virtual_parts` = `current_parts` + `queue`.
    /// Note: it can be different from the actual set of parts because the replica can decide to fetch
    /// a bigger part instead of the part mentioned in the log entry.
    ActiveDataPartSet current_parts;

virtual_parts is what the replica will have once the queue drains; it is computed, not observed. That is what lets the merge-assignment logic reason about the future without talking to anybody.

Which pool

Not all entries are equal. Fetching a part is I/O-bound and small; a merge is CPU- and disk-bound and large. So the entry type picks the thread pool:

cpp 
// src/Storages/StorageReplicatedMergeTree.cpp:4364
    /// Depending on entry type execute in fetches (small) pool or big merge_mutate pool
    if (job_type == LogEntry::GET_PART || job_type == LogEntry::ATTACH_PART)
    {
        assignee.scheduleFetchTask(std::make_shared<ExecutableLambdaAdapter>(
            [this, selected_entry] () mutable
            {
                return processQueueEntry(selected_entry);
            }, common_assignee_trigger, getStorageID()));
        return true;
    }
    if (job_type == LogEntry::MERGE_PARTS)
    {
        auto task = std::make_shared<MergeFromLogEntryTask>(selected_entry, *this, common_assignee_trigger);
        assignee.scheduleMergeMutateTask(task);
        return true;
    }

A replica that is catching up after being offline for an hour has hundreds of GET_PART entries and a handful of MERGE_PARTS entries; putting them in separate pools means the merges do not starve the fetches (or the other way round).

When an entry completes, it is removed from the RAM queue and then from ZooKeeper:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeQueue.cpp:788
            /// Remove the job from the queue in the RAM.
            /// You can not just refer to a pre-saved iterator, because someone else might be able to delete the task.
            /// Why do we view the queue from the end?
            ///  - because the task for execution first is moved to the end of the queue, so that in case of failure it remains at the end.
            for (Queue::iterator it = queue.end(); it != queue.begin();)

The iteration is from the end precisely because of the “move to the tail” rule above.

The insert path: block number and deduplication in one round trip

Now the heart of it. A client inserts a block. The replica must:

  1. give the resulting part a name;
  2. know that this block has not been inserted before;
  3. do both atomically with respect to other replicas doing the same thing.

The block number

Every MergeTree part has a block number — the number of the last block of rows in it — and in a replicated table that number has to be allocated by a coordinator, because two replicas inserting concurrently must not pick the same number.

ClickHouse allocates it with an ephemeral sequential znode:

cpp 
// src/Storages/MergeTree/EphemeralLockInZooKeeper.cpp:53
EphemeralLockInZooKeeper createEphemeralLockInZooKeeper(
    const String & path_prefix_,
    const String & temp_path,
    const ZooKeeperWithFaultInjectionPtr & zookeeper_,
    const std::vector<String> & deduplication_paths,
    const std::optional<String> & znode_data)
{
    String path;

    if (deduplication_paths.empty())
    {
        String holder_path = znode_data.value_or(temp_path + "/" + EphemeralLockInZooKeeper::LEGACY_LOCK_OTHER);
        path = zookeeper_->create(path_prefix_, holder_path, zkutil::CreateMode::EphemeralSequential);
    }

ZooKeeper’s sequential create appends a 10-digit counter to the path prefix. The counter is the block number:

cpp 
// src/Storages/MergeTree/EphemeralLockInZooKeeper.cpp:106
    const UInt64 number = parseSequentialNodeNumber(path, path_prefix_.size());
    warnIfBlockNumberIsHigh(number, path);
    CurrentMetrics::max(CurrentMetrics::MaxAllocatedEphemeralLockSequentialNumber, number);
    return EphemeralLockInZooKeeper{path_prefix_, zookeeper_, path, number};

This is a lovely piece of engineering: the block number is allocated by the same operation that creates the lock, in one round trip, with no read- modify-write. The node is ephemeral, so if the session dies the number is released — but the number is not reused, because the sequential counter is monotone regardless.

The path is per partition:

cpp 
// src/Storages/StorageReplicatedMergeTree.cpp:7987
    auto lock = createEphemeralLockInZooKeeper(
        fs::path(partition_path) / "block-", fs::path(zookeeper_table_path) / "temp", zookeeper, zookeeper_block_id_paths, znode_data);

so /block_numbers/<partition>/block-0000000123 is the lock for block number 123 of that partition. The lock is deleted as part of the commit multi.

The deduplication check rides along

The second thing the insert needs is a duplicate check. The deduplication id is a hash of the block’s content, not the block number:

cpp 
// src/Interpreters/InsertDeduplication.cpp:69
std::string DeduplicationHash::getBlockId() const
{
    return fmt::format("{}_{}_{}", partition_id, hash.items[0], hash.items[1]);
}

and it lives in a znode whose content is the name of the part that holds that data:

cpp 
// src/Interpreters/InsertDeduplication.cpp:75
std::string DeduplicationHash::getPath(const std::string & storage_path) const
{
    std::string hashes_directory = [&] ()
    {
        switch (hash_type)
        {
            case HashType::SYNC:
                return "blocks";
            case HashType::ASYNC:
                return "async_blocks";
            case HashType::UNIFIED:
                return "deduplication_hashes";
        }
    }();
    return (std::filesystem::path(storage_path) / hashes_directory / getBlockId()).string();
}

The three directories are a compatibility story: blocks is the old sync-insert location, async_blocks the old async-insert one, and deduplication_hashes the unified one that both now share.

Now the trick. The duplicate check and the lock creation are performed in a single multi:

cpp 
// src/Storages/MergeTree/EphemeralLockInZooKeeper.cpp:67
    else
    {
        String holder_path = znode_data.value_or(temp_path + "/" + EphemeralLockInZooKeeper::LEGACY_LOCK_INSERT);

        /// Check for duplicates in advance, to avoid superfluous block numbers allocation
        Coordination::Requests ops;

        for (const auto & single_dedup_path : deduplication_paths)
            zkutil::addCheckNotExistsRequest(ops, *zookeeper_, single_dedup_path);

        auto deduplication_path_ops_size = ops.size();

        ops.emplace_back(zkutil::makeCreateRequest(path_prefix_, holder_path, zkutil::CreateMode::EphemeralSequential));
        Coordination::Responses responses;
        Coordination::Error e = zookeeper_->tryMulti(ops, responses);
        if (e == Coordination::Error::ZNODEEXISTS)
        {
            auto failed_idx = zkutil::getFailedOpIndex(Coordination::Error::ZNODEEXISTS, responses);

            if (failed_idx < deduplication_path_ops_size)
            {
                const String & failed_op_path = ops[failed_idx]->getPath();
                LOG_DEBUG(
                    getLogger("createEphemeralLockInZooKeeper"),
                    "Deduplication path already exists: deduplication_path={}",
                    failed_op_path);
                return EphemeralLockInZooKeeper{"", nullptr, "", 0, failed_op_path};
            }
        }

Read that carefully, because it is the whole trick in one function:

  • The multi is [check-not-exists(dedup_path_1), ..., create(block-)].
  • If the dedup path does not exist, the create succeeds and the block number is allocated — atomically, in one round trip.
  • If the dedup path does exist, the multi fails with ZNODEEXISTS, the failed op index tells you which dedup path conflicted, and the function returns a lock with isLocked() == false and a conflict_path.

The caller then knows, without any extra round trip, that this exact content has already been inserted:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeSink.cpp:934
        auto block_number_lock = storage.allocateBlockNumber(part->info.getPartitionId(), zookeeper, block_id_pathes, "", block_data); /// 1 RTT

        ThreadFuzzer::maybeInjectSleep();

        {
            std::filesystem::path conflict_path = block_number_lock.getConflictPath();
            if (!conflict_path.empty())
            {
                LOG_DEBUG(log, "Cannot get lock, the conflict path is {}", conflict_path);
                auto conflicted_hash_it = std::find_if(
                    deduplication_hashes.begin(),
                    deduplication_hashes.end(),
                    [&](const DeduplicationHash & hash)
                    {
                        return hash.getBlockId() == conflict_path.filename().string();
                    });
                chassert(conflicted_hash_it != deduplication_hashes.end());
                retry_context.conflict_deduplication_hashes.push_back(*conflicted_hash_it);

                return CommitRetryContext::RESOLVE_CONFLICTS;
            }
        }

The comment /// 1 RTT on the allocateBlockNumber call is the author telling you that this whole thing — check for duplicates and allocate the block number — costs exactly one round trip to ZooKeeper. That is the entire point. A naive implementation would be: read the dedup znode (1 RTT), decide, create the block number (1 RTT), check again (1 RTT). Three round trips, and a race window between the check and the create.

Resolving the conflict

When the multi fails with a conflict, the replica needs to know which part holds the duplicate data — that is what the dedup znode’s content is for. It reads the conflicting znode:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeSink.cpp:767
        auto response = zookeeper->tryGet(getDeduplicationPaths(storage.zookeeper_path, retry_context.conflict_deduplication_hashes));
        for (size_t i = 0; i < retry_context.conflict_deduplication_hashes.size(); ++i)
        {
            auto & deduplication_hash = retry_context.conflict_deduplication_hashes[i];
            const auto & resp = response[i];

and then the rows that duplicate the existing part are removed from the block, and the part is rewritten with what is left:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeSink.cpp:512
                    auto result = partition.deduplication_info->deduplicateBlock(
                        getDeduplicationBlockIds(conflicts),
                        partition.block_with_partition.partition_id,
                        context);

If everything was a duplicate, the part is dropped and the insert reports that it was deduplicated:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeSink.cpp:538
                if (partition.block_with_partition.block->rows() == 0)
                {
                    // Whole block was deduplicated

                    if (!is_async_insert)
                    {
                        chassert(conflicts.size() == 1);
                        auto block_id = conflicts.front().getBlockId();

and the retry loop goes back to the top and re-commits the remaining rows with a fresh block number.

Note what this means: the deduplication is not “skip the whole insert”. It is “remove the rows that are already there and insert the rest”. A block that partially overlaps with an existing part is inserted partially. That is why the dedup id is a content hash and not the block number: the block number is allocated per partition and is meaningless as a duplicate detector, while the content hash is exactly the question “have I seen these rows before?”.

The whole insert, end to end

text 
  INSERT (block B into partition P)
        |
        |  hash the content of B  ->  dedup id = "<partition>_<hash>"
        v
  multi( [check-not-exists(/deduplication_hashes/<id>),
           create(/block_numbers/P/block-0000000123, EPHEMERAL_SEQUENTIAL)] )
        |
        +--- ZNODEEXISTS --> the block is a duplicate
        |                     read the znode -> the part that holds it
        |                     drop the duplicate rows, rewrite the part, retry
        |
        +--- ZOK --> block number = 123
                      part->info.min_block = max_block = 123
                      part name = <partition>_<min_block>_<max_block>_<level>
        |
        v
  multi( [create(/log/log-...), delete(block-number lock),
           create(quorum znode), create(shared data lock), ...] )
        |
        v
  the log entry "get <part name>" is now the history;
  every replica will fetch the part from whoever has it

The quorum

For a replicated table, “the insert succeeded” means more than “the part is on disk here”. The insert waits for a quorum of replicas to have the part:

cpp 
// src/Storages/MergeTree/ReplicatedMergeTreeSink.cpp:789
            /** If we need a quorum - create a node in which the quorum is monitored.
             * (If such a node already exists, then someone has managed to make another quorum record at the same time,
             *  but for it the quorum has not yet been reached.
             *  You can not do the next quorum record at this time.)
             */
            if (isQuorumEnabled())
            {
                ReplicatedMergeTreeQuorumEntry quorum_entry;
                quorum_entry.part_name = part->name;
                quorum_entry.required_number_of_replicas = getQuorumSize(total_replicas_count);
                quorum_entry.replicas.insert(storage.replica_name);

The quorum znode is a classic “watch and count” pattern: each replica that receives the part adds itself to the znode’s content, and when the count reaches the required number the znode is deleted, which wakes up the waiters.

Looking at it

sql 
-- The replication log, as text.
SELECT * FROM system.zookeeper_log WHERE path LIKE '/clickhouse/tables/%/log/%' LIMIT 5;

-- The per-replica queue.
SELECT replica_name, type, new_part_name, num_tries, num_postponed, last_exception
FROM system.replication_queue
ORDER BY create_time;

-- The deduplication znodes, with the part name they point at.
SELECT name, length(value) FROM system.zookeeper
WHERE path = '/clickhouse/tables/{shard}/default/tbl/deduplication_hashes';

Trade-offs

What the log buys you:

  • A total order that every replica can read, without a coordinator.
  • Recoverability. A replica that was down for an hour catches up by reading the log; it does not need to be told what happened.
  • Human-readable history. The log entries are text, with an explicit format version, which is why a rolling upgrade with mixed versions works and why you can debug a replication problem with system.zookeeper.

What it costs:

  • ZooKeeper is in the write path. Every insert is at least one round trip to ZooKeeper (the dedup + block-number multi) plus one more (the commit multi that creates the log entry). On a cross-availability-zone deployment, that is milliseconds.
  • The log grows forever unless something truncates it. The queue is trimmed per replica, but the shared log is the history.
  • The dedup window is finite. The deduplication znodes are only kept for the last replicated_deduplication_window blocks (10000 by default), so a retry that arrives much later is not deduplicated. This is a correctness-relevant limit, not an optimization.

What the block-number lock buys you:

  • One round trip for both the duplicate check and the number allocation.
  • No read-modify-write. The number comes from ZooKeeper’s sequential counter, so two concurrent inserts on two replicas cannot get the same number, and a crashed session cannot leave a gap that is reused.

What it costs:

  • The block number is not the dedup id. It is tempting to think “the part is named after the block number, so the block number is the duplicate detector”, but that is wrong: the block number is allocated per partition and says nothing about content. The dedup id is a content hash, and it is the hash that makes a retried insert a no-op.
  • A duplicate is not free. The conflict path is resolved by removing the duplicate rows from the block and rewriting the part, which means a partially-overlapping insert is written twice: once (partially) rejected, once rewritten with the remainder.
  • The dedup znodes are garbage that has to be collected. They are deleted lazily, when the parts they point at are dropped, and the window setting bounds how far back they go.

The part people get wrong: ReplicatedMergeTree is not exactly-once by itself. It is at-least-once delivery plus a finite-window content-hash deduplication. A client that retries an insert more than replicated_deduplication_window blocks later will duplicate the rows. If you need a stronger guarantee, the dedup id has to come from the application — the insert_deduplication_token setting lets the client supply its own id, which is then used instead of the content hash and is stable across retries forever.

ClickhouseReplicationInternalsZookeeperDistributed