ClickHouse Coordinates a Cluster-Wide Backup with ZooKeeper

TL;DR: BACKUP ON CLUSTER is not a fan-out of independent backups — it is one logical backup with a shared manifest. Each host publishes a file info record in ZooKeeper, every host reads all of them, sorts them by name, and assigns each file a global data file index. Who writes which file is decided by a create on /writing_files/<index> — a ZooKeeper node creation is the mutex. Between phases, every host parks at a named stage and the group does not move on until every host has arrived.

Backing up a cluster looks like a solved thing until you try to solve it. The naive implementation is: send BACKUP to every host, let each one dump its own local data, call it a day. That works until you want the result to be a backup rather than N backups — one manifest, one file namespace, restorable as a unit.

The trouble is that a MergeTree cluster is not N independent databases. A replicated table has the same logical data on every replica; a Replicated database replicates its DDL history; system.users on one host is not the same as on another. If each host writes its own files, you get N copies of everything and no way to tell which copy is the real one.

ClickHouse’s answer is to make the backup global. There is exactly one manifest, exactly one file namespace, and a rule that decides who writes what. That rule is implemented with ZooKeeper primitives, and it is much cleverer than it first looks.

The shape of the thing

First, the trick that makes the whole thing possible: BACKUP ON CLUSTER reuses the distributed DDL machinery to ship the query. There is no separate “backup cluster protocol”.

cpp
// src/Backups/BackupsWorker.cpp:1255
void BackupsWorker::sendQueryToOtherHosts(const ASTBackupQuery & backup_or_restore_query, const ClusterPtr & cluster,
    size_t only_shard_num, size_t only_replica_num, ContextMutablePtr context, const AccessRightsElements & access_to_check,
    const ZooKeeperRetriesInfo & retries_info) const
{
    chassert(cluster);

    DDLQueryOnClusterParams params;
    params.cluster = cluster;
    params.only_shard_num = only_shard_num;
    params.only_replica_num = only_replica_num;
    params.access_to_check = access_to_check;
    params.retries_info = retries_info;

    context->setSetting("distributed_ddl_task_timeout", Field{0});
    context->setSetting("distributed_ddl_output_mode", Field{"never_throw"});

    // executeDDLQueryOnCluster() will return without waiting for completion
    executeDDLQueryOnCluster(backup_or_restore_query.clone(), context, params);

The initiator does not run the backup itself. It ships the query, sets distributed_ddl_task_timeout to zero so it does not wait, and then watches ZooKeeper for the result of the internal per-host operations. That is why the backup coordinator has two roles: the initiator has an empty host name (kInitiator), every other host has its own host_id.

cpp
// src/Backups/BackupsWorker.cpp:1286
    bool is_internal_backup = backup_settings.internal;

    String root_zk_path = context->getConfigRef().getString("backups.zookeeper_path", "/clickhouse/backups");
    auto get_zookeeper = [global_context = context->getGlobalContext()] { return global_context->getZooKeeper(); };
    auto keeper_settings = BackupKeeperSettings(context);

    auto all_hosts = BackupSettings::Util::filterHostIDs(
        backup_settings.cluster_host_ids, backup_settings.shard_num, backup_settings.replica_num);
    all_hosts.emplace_back(BackupCoordinationOnCluster::kInitiator);

    String current_host = is_internal_backup ? backup_settings.host_id : String{BackupCoordinationOnCluster::kInitiator};

Note all_hosts.emplace_back(BackupCoordinationOnCluster::kInitiator): the initiator is in the list as a pseudo-host, even though it does no work. That matters for the barrier — other hosts wait on the initiator’s stage node too.

The ZooKeeper layout

Everything for one backup lives under one subtree. The paths are constructed once, in the constructor, and every node name is derived from them:

cpp
// src/Backups/BackupCoordinationStageSync.cpp:126
    , zookeeper_path(zookeeper_path_)
    , root_zookeeper_path(zookeeper_path.parent_path().parent_path())
    , operation_zookeeper_path(zookeeper_path.parent_path())
    , operation_node_name(zookeeper_path.parent_path().filename())
    , start_node_path(zookeeper_path / ("started|" + current_host))
    , finish_node_path(zookeeper_path / ("finished|" + current_host))
    , initiator_start_node_path(zookeeper_path / ("started|" + String{kInitiator}))
    , num_hosts_node_path(zookeeper_path / "num_hosts")
    , error_node_path(zookeeper_path / "error")
    , alive_node_path(zookeeper_path / ("alive|" + current_host))
    , alive_tracker_node_path(fs::path{root_zookeeper_path} / "alive_tracker")

With backups.zookeeper_path = /clickhouse/backups and a backup UUID:

text
/clickhouse/backups/
├── alive_tracker                       (persistent; "someone is alive" version counter)
└── backup-<uuid>/
    ├── stage/
    │   ├── started|<host>               (persistent; host has begun)
    │   ├── started|                     (persistent; the initiator)
    │   ├── finished|<host>              (persistent; host is done)
    │   ├── alive|<host>                 (EPHEMERAL; host is alive)
    │   ├── num_hosts                    (persistent; counter)
    │   ├── error                        (persistent; someone failed)
    │   ├── current|<host>|gathering metadata (0)
    │   ├── current|<host>|writing backup   (0)
    │   └── current|<host>|completed       (0)
    ├── file_infos/<host>/000000         (persistent; the file manifest, chunked)
    ├── writing_files/17                 (persistent; "host #17 owns data file 17")
    ├── repl_part_names/…  repl_mutations/…  repl_access/…
    ├── repl_sql_objects/…  repl_workload_entities/…
    └── keeper_map_tables/…  rocksdb_tables/…

Three things to notice:

alive|<host> is ephemeral — it disappears automatically when the host’s ZooKeeper session ends. That is the liveness signal, and it is why a crashed host does not deadlock the whole backup.
current|<host>|<stage> encodes both the stage and a small result payload. For most stages the result is empty; for metadata gathering it is "consistent" or "need_another_attempt". The stage barrier is thus a general barrier that also carries a vote.
file_infos/<host> holds the per-host manifest, chunked because ZooKeeper nodes have a size limit.

Phase 0: claiming the operation

Before any work happens, a host must announce itself and check that it is allowed to start at all. This happens in one ZooKeeper multi request, which matters: the whole claim must be atomic, or two hosts could both believe they own the backup.

cpp
// src/Backups/BackupCoordinationStageSync.cpp:304
        Coordination::Requests requests;
        requests.reserve(6);

        size_t operation_node_pos = static_cast<size_t>(-1);
        if (!zookeeper->exists(operation_zookeeper_path))
        {
            operation_node_pos = requests.size();
            requests.emplace_back(zkutil::makeCreateRequest(operation_zookeeper_path, "", zkutil::CreateMode::Persistent));
        }

        size_t zookeeper_path_pos = static_cast<size_t>(-1);
        if (!zookeeper->exists(zookeeper_path))
        {
            zookeeper_path_pos = requests.size();
            requests.emplace_back(zkutil::makeCreateRequest(zookeeper_path, "", zkutil::CreateMode::Persistent));
        }

        size_t num_hosts_node_pos = requests.size();
        if (num_hosts)
            requests.emplace_back(zkutil::makeSetRequest(num_hosts_node_path, toString(*num_hosts + 1), num_hosts_version));
        else
            requests.emplace_back(zkutil::makeCreateRequest(num_hosts_node_path, "1", zkutil::CreateMode::Persistent));

        size_t alive_tracker_node_pos = requests.size();
        requests.emplace_back(zkutil::makeSetRequest(alive_tracker_node_path, "", alive_tracker_version));

        requests.emplace_back(zkutil::makeCreateRequest(start_node_path, std::to_string(kCurrentVersion), zkutil::CreateMode::Persistent));
        requests.emplace_back(zkutil::makeCreateRequest(alive_node_path, "", zkutil::CreateMode::Ephemeral));

        Coordination::Responses responses;
        auto code = zookeeper->tryMulti(requests, responses);

Six operations, one transaction: create the operation node, create the stage node, bump the num_hosts counter with a versioned set, bump the alive_tracker version, create my started|<host> marker, create my ephemeral alive|<host> node.

The two sets are versioned deliberately. The version acts as an optimistic-concurrency token, and a ZBADVERSION on either means “somebody else changed the world under me, start over”:

cpp
// src/Backups/BackupCoordinationStageSync.cpp:372
        else if ((num_hosts_node_pos < responses.size()) && num_hosts &&
            (responses[num_hosts_node_pos]->error == Coordination::Error::ZBADVERSION))
        {
            show_error_before_next_attempt(fmt::format("The version of node {} changed", num_hosts_node_path));
            num_hosts.reset(); /// needs to reread 'num_hosts' again
        }
        else if ((alive_tracker_node_pos < responses.size()) &&
            (responses[alive_tracker_node_pos]->error == Coordination::Error::ZBADVERSION))
        {
            show_error_before_next_attempt(fmt::format("The version of node {} changed", alive_tracker_node_path));
            check_concurrency = true; /// needs to recheck for concurrency again
        }

Read that second branch carefully: a failed alive_tracker bump forces a re-check of concurrency. That is the whole concurrency protocol in one line. The alive_tracker version is a cheap “has anything changed underneath me” flag; if it moved, the world may now contain another running backup.

The concurrency check

allow_concurrent_backups defaults to true, but the check exists for when it is turned off. “Another operation is running” is defined precisely as: another backup-<uuid> directory exists whose stage/ contains a node starting with alive.

cpp
// src/Backups/BackupCoordinationStageSync.cpp:408
    for (const String & found_operation : found_operations)
    {
        if (found_operation.starts_with(is_restore ? "restore-" : "backup-") && (found_operation != operation_node_name))
        {
            Strings stages;
            code = zookeeper->tryGetChildren(fs::path{root_zookeeper_path} / found_operation / "stage", stages);

            if (!((code == Coordination::Error::ZOK) || (code == Coordination::Error::ZNONODE)))
                throw zkutil::KeeperException::fromPath(code, fs::path{root_zookeeper_path} / found_operation / "stage");

            if (code == Coordination::Error::ZOK)
            {
                for (const String & stage : stages)
                {
                    if (stage.starts_with("alive"))
                        BackupConcurrencyCheck::throwConcurrentOperationNotAllowed(is_restore);
                }
            }
        }
    }

Because alive nodes are ephemeral, a dead host’s backup does not block the next one.

There is also a purely local concurrency check, in the server process rather than in ZooKeeper:

cpp
// src/Backups/BackupConcurrencyCheck.cpp:44
        else
        {
            size_t num_local_backups = counters.local_backups;
            size_t num_on_cluster_backups = counters.on_cluster_backups.size();
            if (on_cluster)
            {
                if (!counters.on_cluster_backups.contains(zookeeper_path))
                    ++num_on_cluster_backups;
            }
            else
            {
                ++num_local_backups;
            }
            found_concurrent_operation = (num_local_backups + num_on_cluster_backups > 1);
        }

The counters are reference-counted per ZooKeeper path, incremented in the constructor and decremented in the destructor — a scoped lock. Keying by path means two BACKUP ON CLUSTER operations against different clusters do not collide, but two against the same path do.

Phase 1: the metadata consensus loop

Here is the part that is genuinely unusual. Before copying any data, every host independently collects the list of databases and tables it will back up. If a table is renamed mid-collection, one host may see the old name and another the new one, and the resulting backup would be garbage.

ClickHouse’s answer is not a lock — it is a convergent retry loop with exponential backoff and a global agreement check. The loop is documented in a comment that reads like a design document:

cpp
// src/Backups/BackupEntriesCollector.cpp:237
void BackupEntriesCollector::gatherMetadataAndCheckConsistency()
{
    /// With the default values the metadata collecting works in the following way:
    /// 1) it tries to collect the metadata for the first time; then
    /// 2) it tries to collect it again, and compares the results from the first and the second collecting; if they match, it's done; otherwise
    /// 3) it sleeps 100 millisecond and tries to collect again, then compares the results from the second and the third collecting; if they match, it's done; otherwise
    /// 4) it tries to collect again, then compares the results from the third and the fourth collecting; if they match, it's done; otherwise
    /// 5) it sleeps 200 milliseconds and tries to collect again, then compares the results from the fourth and the fifth collecting; if they match, it's done;
    /// ...
    /// and so on, the sleep time is doubled each time until it reaches 5000 milliseconds.
    /// And such attempts will be continued until 600000 milliseconds pass.

The numbers come from the server config: backups.attempts_to_collect_metadata_before_sleep (default 2), backups.min_sleep_before_next_attempt_to_collect_metadata (default 100 ms), backups.max_sleep_before_next_attempt_to_collect_metadata (default 5000 ms), backups.collect_metadata_timeout (default 600000 ms). The backoff is a clean power-of-two ramp, capped at 2^10:

cpp
// src/Backups/BackupEntriesCollector.cpp:91
    /// How long we should sleep after finding an inconsistency error.
    std::chrono::milliseconds getSleepTimeAfterInconsistencyError(int attempt_no, unsigned int attempts_before_sleep, std::chrono::milliseconds min_sleep, std::chrono::milliseconds max_sleep)
    {
        attempts_before_sleep = std::max(attempts_before_sleep, 1U);
        if (((attempt_no + 1) % attempts_before_sleep) != 0)
            return std::chrono::milliseconds::zero(); /// no sleep

        int sleep_counter = attempt_no / attempts_before_sleep;
        std::chrono::milliseconds sleep_time = intExp2(std::min(sleep_counter, 10)) * min_sleep;
        return std::min(sleep_time, max_sleep);
    }

Two kinds of consistency are checked. The local one: collect twice in a row and compare. A table renamed between the two passes shows up as a mismatch.

cpp
// src/Backups/BackupEntriesCollector.cpp:312
    if (!compare_collected_metadata)
        return;

    /// We have to check consistency of collected information to protect from the case when some table or database is
    /// renamed during this collecting making the collected information invalid.
    String mismatch_description;
    if (!compareWithPrevious(mismatch_description))
    {
        /// Usually two passes is minimum.
        /// (Because we need to compare with table names from the previous pass to be sure we are not going to miss anything).
        if (attempt_no >= 1)
            inconsistency_error = Exception{ErrorCodes::INCONSISTENT_METADATA_FOR_BACKUP, "{}", mismatch_description};
        need_another_attempt = true;
    }

And the global one: every host publishes its verdict as the stage result and waits for everybody else’s verdict. The stage barrier does double duty as a consensus mechanism:

cpp
// src/Backups/BackupEntriesCollector.cpp:340
    /// We've gathered metadata, cool! But we have to check that other hosts coped with that too.
    auto all_hosts_results = setStage(next_stage, need_another_attempt ? "need_another_attempt" : "consistent");
    size_t count = std::min(all_hosts.size(), all_hosts_results.size());

    for (size_t i = 0; i != count; ++i)
    {
        const auto & other_host = all_hosts[i];
        const auto & other_host_result = all_hosts_results[i];
        if ((other_host_result != "consistent") && (other_host_result != "need_another_attempt"))
        {
            LOG_WARNING(log, "Found inconsistency on host {}: {}", other_host, other_host_result);
            inconsistency_error = Exception{ErrorCodes::INCONSISTENT_METADATA_FOR_BACKUP, "Found inconsistency on host {}: {}", other_host, other_host_result};
            need_another_attempt = true;
            return;
        }
    }

The stage name is parameterised by the attempt number — "gathering metadata (0)", "gathering metadata (1)", and so on — because the barrier is per attempt. Every host must reach attempt N before anyone proceeds to attempt N+1.

Note the subtlety: a host that failed to collect metadata publishes its error text as the stage result and returns immediately. The other hosts, on reading that result, see a string that is neither "consistent" nor "need_another_attempt", and they treat it as a failure and start their own next attempt.

The stage barrier itself

The barrier is a condition variable fed by a background polling thread. There is no watch on the stage nodes — the implementation polls.

cpp
// src/Backups/BackupCoordinationStageSync.cpp:504
        try
        {
            /// Recreate the 'alive' node if necessary and read a new state from ZooKeeper.
            auto holder = with_retries.createRetriesControlHolder("BackupCoordinationStageSync::watchingThread");
            auto & zookeeper = holder.faulty_zookeeper;
            with_retries.renewZooKeeper(zookeeper);

            if (should_stop())
                return;

            /// Recreate the 'alive' node if it was removed.
            createAliveNode(zookeeper);

            /// Reads the current state from nodes in ZooKeeper.
            readCurrentState(zookeeper);
        }

The alive node must be recreated on every session loss — an ephemeral node is tied to the session, so a reconnect means a new session means the node is gone. The recreation is a two-op multi (create the node, bump the alive_tracker version), which is what makes the tracker version a reliable “something happened” flag:

cpp
// src/Backups/BackupCoordinationStageSync.cpp:547
void BackupCoordinationStageSync::createAliveNode(Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper)
{
    if (zookeeper->exists(alive_node_path))
        return;

    Coordination::Requests requests;
    requests.emplace_back(zkutil::makeCreateRequest(alive_node_path, "", zkutil::CreateMode::Ephemeral));
    requests.emplace_back(zkutil::makeSetRequest(alive_tracker_node_path, "", -1));
    zookeeper->multi(requests);

    LOG_INFO(log, "The alive node was recreated for {}", current_host_desc);
}

The polling period is backups.sync_period_ms, default 5000 ms. So the worst-case latency for a stage barrier is five seconds. That is a deliberate trade: polling is simple and survives any kind of ZooKeeper weirdness, where a watch would need careful re-arming.

Publishing a stage is a plain create of a persistent node whose name encodes the stage, with the result as its payload:

cpp
// src/Backups/BackupCoordinationStageSync.cpp:869
void BackupCoordinationStageSync::createStageNode(const String & stage, const String & stage_result, Coordination::ZooKeeperWithFaultInjection::Ptr zookeeper)
{
    if (isErrorSet())
        rethrowSetError();

    String serialized_error;
    if (zookeeper->tryGet(error_node_path, serialized_error))
    {
        auto [exception, host] = parseErrorNode(serialized_error);
        if (exception)
        {
            std::lock_guard lock{mutex};
            state.addErrorInfo(exception, host);
            std::rethrow_exception(exception);
        }
    }

    auto code = zookeeper->tryCreate(getStageNodePath(stage), stage_result, zkutil::CreateMode::Persistent);
    if (code == Coordination::Error::ZOK)
    {
        std::lock_guard lock{mutex};
        state.hosts.at(current_host).stages[stage] = stage_result;
        return;
    }

    if (code == Coordination::Error::ZNODEEXISTS)
    {
        String another_result = zookeeper->get(getStageNodePath(stage));
        std::lock_guard lock{mutex};
        state.hosts.at(current_host).stages[stage] = another_result;
        return;
    }

    throw zkutil::KeeperException::fromPath(code, getStageNodePath(stage));
}

ZNODEEXISTS is treated as success — the host already published this stage (possibly in a previous attempt that appeared to fail). Idempotence by tolerance.

And the waiting side:

cpp
// src/Backups/BackupCoordinationStageSync.cpp:945
        const HostInfo & host_info = it->second;
        auto stage_it = host_info.stages.find(stage_to_wait);
        if (stage_it != host_info.stages.end())
        {
            results[i] = stage_it->second;
            continue;
        }

        if (host_info.finished)
        {
            if (stage_to_wait == BackupCoordinationStage::FINALIZING_TABLES)
            {
                /// This is a newly added stage. For compatibility with older server versions,
                /// allow other replicas to skip this stage. This doesn't break anything: this stage
                /// unpauses refreshable materialized views, but older server versions don't pause
                /// them in the first place.
                results[i] = "";
                continue;
            }
            throw Exception(ErrorCodes::FAILED_TO_SYNC_BACKUP_OR_RESTORE,
                            "{} finished without coming to stage {}", getHostDesc(host), stage_to_wait);
        }

The FINALIZING_TABLES special case is a lovely piece of pragmatism: a stage was added to the protocol later, and an older server that has already finished will never reach it. Rather than deadlock the cluster, the barrier treats “finished” as “passed every stage” for that one stage only.

The stage list

The stage names are constants in one header (src/Backups/BackupCoordinationStage.h), which makes the protocol readable at a glance:

cpp
// src/Backups/BackupCoordinationStage.h:9
namespace BackupCoordinationStage
{
    /// Finding all tables and databases which we're going to put to the backup and collecting their metadata.
    constexpr const char * GATHERING_METADATA = "gathering metadata";

    String formatGatheringMetadata(int attempt_no);

    /// Making temporary hard links and prepare backup entries.
    constexpr const char * EXTRACTING_DATA_FROM_TABLES = "extracting data from tables";

    /// Running special tasks for replicated tables which can also prepare some backup entries.
    constexpr const char * RUNNING_POST_TASKS = "running post-tasks";

    /// Building information about all files which will be written to a backup.
    constexpr const char * BUILDING_FILE_INFOS = "building file infos";

    /// Writing backup entries to the backup and removing temporary hard links.
    constexpr const char * WRITING_BACKUP = "writing backup";

    /// Finding databases and tables in the backup which we're going to restore.
    constexpr const char * FINDING_TABLES_IN_BACKUP = "finding tables in backup";

The backup half is: gathering metadata → extracting data → running post-tasks → building file infos → writing backup → completed. The restore half is separate (finding tables → checking access rights → creating databases → creating tables → inserting data → finalizing tables → completed). BackupCoordinationStageSync is constructed with an is_restore flag and serves both protocols.

Phase 2: the file manifest — the heart of it

Each host has its own set of files (its own MergeTree parts, its own system tables, its own access files). The backup needs one list.

Each host computes a BackupFileInfo for each of its files — name, size, checksum, plus the incremental-backup fields — and publishes them to ZooKeeper:

cpp
// src/Backups/BackupCoordinationOnCluster.cpp:945
void BackupCoordinationOnCluster::addFileInfos(BackupFileInfos && file_infos_)
{
    {
        std::lock_guard lock{file_infos_mutex};
        if (file_infos)
            throw Exception(ErrorCodes::LOGICAL_ERROR, "addFileInfos() must not be called after preparing");
    }

    auto component_guard = Coordination::setCurrentComponent("BackupCoordinationOnCluster::addFileInfos");
    /// Serialize `file_infos_` and write it to ZooKeeper's nodes.
    String file_infos_str = FileInfos::serialize(file_infos_);
    serializeToMultipleZooKeeperNodes(zookeeper_path + "/file_infos/" + current_host, file_infos_str, "addFileInfos");
}

The serialized form is compact and deliberately omits the fields that are derived rather than observed:

cpp
// src/Backups/BackupCoordinationOnCluster.cpp:106
        static String serialize(const BackupFileInfos & file_infos_)
        {
            WriteBufferFromOwnString out;
            writeBinary(file_infos_.size(), out);
            for (const auto & info : file_infos_)
            {
                writeBinary(info.file_name, out);
                writeBinary(info.size, out);
                writeBinary(info.checksum, out);
                writeBinary(info.base_size, out);
                writeBinary(info.base_checksum, out);
                writeBinary(info.encrypted_by_disk, out);
                writeBinary(info.reference_target, out);
                /// We don't store `info.data_file_name` and `info.data_file_index` because they're determined automalically
                /// after reading file infos for all the hosts (see the class BackupCoordinationFileInfos).
            }
            return out.str();
        }

The payload can be large — a backup of a big cluster has millions of files — so it is chopped into chunks of at most backup_restore_keeper_value_max_size bytes (default 1 MiB) and stored as numbered child nodes:

cpp
// src/Backups/BackupCoordinationOnCluster.cpp:308
    size_t max_part_size = keeper_settings.value_max_size;
    if (!max_part_size)
        max_part_size = value.size();

    size_t num_parts = (value.size() + max_part_size - 1) / max_part_size; /// round up

    for (size_t i = 0; i != num_parts; ++i)
    {
        size_t begin = i * max_part_size;
        size_t end = std::min(begin + max_part_size, value.size());
        String part = value.substr(begin, end - begin);
        String part_path = fmt::format("{}/{:06}", path, i);

The {:06} zero-padding makes the lexicographic order of the child names equal to the numeric order of the parts — so reading them back is just getChildren + sort.

Then every host reads everyone’s file infos and computes the global plan:

cpp
// src/Backups/BackupCoordinationOnCluster.cpp:975
void BackupCoordinationOnCluster::prepareFileInfos() const
{
    if (file_infos)
        return;

    file_infos.emplace(BackupCoordinationFileInfos::Config{plain_backup, data_file_name_gen, data_file_name_prefix_length});

    Strings hosts_with_file_infos;
    {
        auto holder = with_retries.createRetriesControlHolder("prepareFileInfos::get_hosts");
        holder.retries_ctl.retryLoop(
            [&, &zk = holder.faulty_zookeeper]()
        {
            with_retries.renewZooKeeper(zk);
            hosts_with_file_infos = zk->getChildren(zookeeper_path + "/file_infos");
        });
    }

    for (const String & host : hosts_with_file_infos)
    {
        String file_infos_str = deserializeFromMultipleZooKeeperNodes(zookeeper_path + "/file_infos/" + host, "prepareFileInfos");
        auto deserialized_file_infos = FileInfos::deserialize(file_infos_str).file_infos;
        file_infos->addFileInfos(std::move(deserialized_file_infos), host);
    }
}

Phase 3: assigning a global name and index to every file

BackupCoordinationFileInfos::prepare() is where the global plan is computed. Every host runs this code over the same input and must get the same answer — which is why the sort key is the file name and nothing else.

cpp
// src/Backups/BackupCoordinationFileInfos.cpp:67
    /// Make a list of all file infos from all hosts.
    size_t total_num_infos = 0;
    for (const auto & [_, infos] : file_infos)
        total_num_infos += infos.size();

    file_infos_for_all_hosts.reserve(total_num_infos);
    for (auto & [_, infos] : file_infos)
        for (auto & info : infos)
            file_infos_for_all_hosts.emplace_back(&info);

    /// Sort the list of all file infos by file name (file names must be unique).
    std::sort(file_infos_for_all_hosts.begin(), file_infos_for_all_hosts.end(), BackupFileInfo::LessByFileName{});

    auto adjacent_it = std::adjacent_find(file_infos_for_all_hosts.begin(), file_infos_for_all_hosts.end(), BackupFileInfo::EqualByFileName{});
    if (adjacent_it != file_infos_for_all_hosts.end())
    {
        throw Exception(
            ErrorCodes::BACKUP_ENTRY_ALREADY_EXISTS, "Entry {} added multiple times to backup", quoteString((*adjacent_it)->file_name));
    }

The file names are unique because each host stores its data under shards/<n>/replicas/<m>/ in the backup. Two hosts cannot produce the same name, and a duplicate is a hard error rather than a silent overwrite.

Now the interesting branch — the non-plain (deduplicating) path:

cpp
// src/Backups/BackupCoordinationFileInfos.cpp:162
        /// For non-plain backups files with the same size and checksum are stored only once,
        /// in order to find those files we'll use this map.
        std::map<SizeAndChecksum, size_t> data_file_index_by_checksum;

        for (size_t i = 0; i != file_infos_for_all_hosts.size(); ++i)
        {
            auto & info = *(file_infos_for_all_hosts[i]);

            if (!info.reference_target.empty())
            {
                if (!try_resolve_reference(info))
                    unresolved_references.push_back(&info);

                continue;
            }

            if (info.size == info.base_size)
            {
                /// A file is either empty or can be get from the base backup as a whole.
                info.data_file_name.clear();
                info.data_file_index = static_cast<size_t>(-1);
            }
            else
            {
                SizeAndChecksum size_and_checksum{info.size, info.checksum};
                auto [it, inserted] = data_file_index_by_checksum.emplace(size_and_checksum, i);
                if (inserted)
                {
                    /// Found a new file.
                    info.data_file_name = getBackupDataFileName(info, config.data_file_name_generator, config.data_file_name_prefix_length);
                    info.data_file_index = i;
                    ++num_files;
                    total_size_of_files += info.size - info.base_size;
                }
                else
                {
                    /// Found a file with the same size and checksum as some file before, reuse old `data_file_index` and `data_file_name`.
                    info.data_file_index = it->second;
                    info.data_file_name = file_infos_for_all_hosts[it->second]->data_file_name;
                }
            }

            file_name_to_info.emplace(info.file_name, &info);
        }

Three things happen here:

data_file_index is simply the position in the globally sorted list. That is the coordination key: index 17 is the same file on every host.
Files with identical (size, checksum) collapse to one physical data file. On a replicated cluster, the same part exists on every replica with the same checksum, so a 3-replica cluster stores one copy of the data and three manifest entries pointing at it.
total_size_of_files accumulates info.size - info.base_size, i.e. only the bytes that are new relative to the incremental base.

Phase 4: electing the writer — ZooKeeper `create` as a mutex

Now the question: host A has file 17, host B has file 23, host C has file 17 too (same content). Somebody has to write the physical file for index 17. Exactly one host must do it.

The answer is the simplest possible use of ZooKeeper: create is atomic, and fails with ZNODEEXISTS if the node is there.

cpp
// src/Backups/BackupCoordinationOnCluster.cpp:1001
bool BackupCoordinationOnCluster::startWritingFile(size_t data_file_index)
{
    {
        /// Check if this host is already writing this file.
        std::lock_guard lock{writing_files_mutex};
        if (writing_files.contains(data_file_index))
            return false;
    }
    auto component_guard = Coordination::setCurrentComponent("BackupCoordinationOnCluster::startWritingFile");

    /// Store in Zookeeper that this host is the only host which is allowed to write this file.
    bool host_is_assigned = false;
    String full_path = zookeeper_path + "/writing_files/" + std::to_string(data_file_index);
    String host_index_str = std::to_string(current_host_index);

    auto holder = with_retries.createRetriesControlHolder("startWritingFile");
    holder.retries_ctl.retryLoop(
            [&, &zk = holder.faulty_zookeeper]()
    {
        with_retries.renewZooKeeper(zk);
        auto code = zk->tryCreate(full_path, host_index_str, zkutil::CreateMode::Persistent);

        if (code == Coordination::Error::ZOK)
            host_is_assigned = true; /// If we've just created this ZooKeeper's node, this host is assigned.
        else if (code == Coordination::Error::ZNODEEXISTS)
            host_is_assigned = (zk->get(full_path) == host_index_str); /// The previous retry could write this ZooKeeper's node and then fail.
        else
            throw zkutil::KeeperException::fromPath(code, full_path);
    });

    if (!host_is_assigned)
        return false; /// Other host is writing this file.

    {
        /// Check if this host is already writing this file,
        /// and if it's not, mark that this host is writing this file.
        /// We have to check that again because we were accessing ZooKeeper with the mutex unlocked.
        std::lock_guard lock{writing_files_mutex};
        return writing_files.emplace(data_file_index).second; /// Return false if this host is already writing this file.
    }
}

The ZNODEEXISTS branch is the interesting one. The node’s value is the index of the host that won the election. If the value equals my index, I won — even though I did not create the node. That covers the case where a retry after a network hiccup created the node successfully but the response was lost. Idempotent election without a separate “am I the owner?” round trip.

And the caller simply skips files it lost:

cpp
// src/Backups/BackupImpl.cpp:1741
    if (!coordination->startWritingFile(info.data_file_index))
    {
        LOG_TRACE(log, "Writing backup for file {} from {}: skipped, data file #{} is already being written", info.data_file_name, src_file_desc, info.data_file_index);
        return;
    }

So on a 3-replica cluster with identical parts, exactly one replica uploads each distinct file and the other two skip it. The result is a backup that contains the data once.

text
  host A                     host B                     host C
    |                           |                           |
    |-- file_infos/A ------>  /file_infos/  <--- file_infos/B --|
    |                           |                           |
    |   all three read all three manifests, sort by name     |
    |   all three compute the same data_file_index map       |
    |                           |                           |
    | create /writing_files/17  -> ZOK   (A wins)           |
    |                           | create -> ZNODEEXISTS     |
    |                           |   value != my index       |
    |                           |   -> skip file 17         |
    |                           |                           |
    | write file 17             | write file 23             | write file 41

Retries: everything is wrapped

Every ZooKeeper operation in the backup path goes through a retry controller with exponential backoff. The default is deliberately generous, because a backup that fails halfway because of a 30-second Keeper blip is worse than one that takes longer:

cpp
// src/Core/Settings.cpp:4848
    DECLARE(UInt64, backup_restore_keeper_max_retries, 1000, R"(
Max retries for [Zoo]Keeper operations in the middle of a BACKUP or RESTORE operation.
Should be big enough so the whole operation won't fail because of a temporary [Zoo]Keeper failure.
)", 0, \

The backoff is a plain doubling with a cap:

cpp
// src/Common/ZooKeeper/ZooKeeperRetries.h:232
        /// retries
        logLastError("will retry due to error");
        sleepForMilliseconds(current_backoff_ms);
        current_backoff_ms = std::min(current_backoff_ms * 2, retries_info.max_backoff_ms);

with backup_restore_keeper_retry_initial_backoff_ms = 100 and backup_restore_keeper_retry_max_backoff_ms = 5000. Critically, only hardware errors are retried:

cpp
// src/Common/ZooKeeper/ZooKeeperRetries.h:81
            catch (const zkutil::KeeperException & e)
            {
                iteration_cleanup();

                if (!Coordination::isHardwareError(e.code))
                    throw;

                setKeeperError(std::current_exception(), e.code, e.message());
            }

A ZNODEEXISTS or a ZBADVERSION is a semantic answer, not a failure — retrying it would be wrong. A connection reset is a hardware error and gets retried.

Trade-offs

What the design buys:

One manifest, one file namespace. A BACKUP ON CLUSTER is restorable as a unit; RESTORE ON CLUSTER puts the data back on the right shards and replicas.
No data duplication. The (size, checksum) dedup means a 3-replica cluster stores the part once.
No coordination server. The only shared state is ZooKeeper, and the only ZooKeeper operation used for the critical election is create.
Crash tolerance. Every phase is idempotent. A host that dies leaves ephemeral nodes behind; the survivors notice within sync_period_ms and either retry or abort.

What it costs:

Polling latency. backups.sync_period_ms defaults to 5000 ms, so each stage barrier costs up to five seconds of wall clock even when everything is ready. A backup with six stages pays this repeatedly.
ZooKeeper is a hard dependency for the whole operation. It is not in the data path — the actual bytes go straight to S3 — but it is in the control path, and a Keeper outage stalls every backup in the cluster.
The manifest is O(files) in ZooKeeper. A backup of millions of parts writes millions of file-info records into Keeper, in 1 MiB chunks. This is the single biggest scaling constraint on BACKUP ON CLUSTER.
The metadata consensus loop is a heuristic, not a proof. Two consecutive identical reads on every host, plus a cross-host vote, gives strong practical safety — but there is no fencing token, so in principle a rename that happens to straddle the window identically on every host could slip through. The INCONSISTENT_METADATA_FOR_BACKUP error and the 600-second timeout are the safety nets.

Wrapping up

The one thing to remember: a cluster-wide backup is a global file namespace, and the global namespace is produced by sorting every host’s file list by name. The sort is the whole trick — it gives every file a cluster-unique index without any central allocation, because sorting is commutative and deterministic.

The part people get wrong: the deduplication is by (size, checksum), not by path. Two hosts with the same file at different paths in the backup share one physical object. That is why getNumFiles() and getNumEntries() differ, and why the backup’s on-disk size is much smaller than the sum of the replicas’ data.

What to go do about it: if your BACKUP ON CLUSTER is slow, look at the log lines Setting stage: ... and Hosts [...] reached stage .... The gap between them is barrier latency — and with the default 5-second sync period, that gap is almost certainly the problem, not the data transfer.