ClickHouse Runs Distributed DDL Through a ZooKeeper Task Queue

TL;DR: ON CLUSTER does not open N connections and fire N copies of the query. It writes one ZooKeeper node — query-0000000123 — into a shared queue directory, and every host in the cluster watches that directory. Each host that is in the query’s host list runs the DDL and creates <entry>/finished/<host_id> with its exit code. The initiator then just lists <entry>/finished — one getChildren — and reads the status of every host that finished.

The obvious way to implement CREATE TABLE t ON CLUSTER c is to open a connection to each host in the cluster, send it the query, and collect the answers. That is what a lot of databases do, and it has an obvious problem: the initiator is a single point of failure, and it holds N connections open for the duration.

ClickHouse takes the opposite approach. The initiator writes the query into a shared queue in ZooKeeper and then gets out of the way. Every host pulls from the queue itself. This means the initiator can die and the DDL still happens; it means the work is spread over the cluster rather than funnelled through one server; and it means the “which hosts have finished” bookkeeping is just a directory listing.

The mechanism is worth looking at closely, because the interesting parts are all in the details of how ZooKeeper is used.

The queue

The DDL queue is a directory in ZooKeeper. Its path comes from the server setting distributed_ddl_path, which defaults to /clickhouse/task_queue/ddl/:

cpp
// src/Core/ServerSettings.cpp:2102
    DECLARE(String, distributed_ddl_path, "/clickhouse/task_queue/ddl/", R"(the path in Keeper for the `<task_queue>` for DDL queries)", 0, "distributed_ddl.path") \
    DECLARE(String, distributed_ddl_replicas_path, "/clickhouse/task_queue/replicas/", R"(the path in Keeper for the `<task_queue>` for replicas)", 0, "distributed_ddl.replicas_path") \

The worker is started once per server, after all the tables are loaded:

cpp
// programs/server/Server.cpp:3787
        if (has_zookeeper && config().has("distributed_ddl"))
        {
            /// DDL worker should be started after all tables were loaded
            String ddl_queue_path = server_settings[ServerSetting::distributed_ddl_path];
            String ddl_replicas_path = server_settings[ServerSetting::distributed_ddl_replicas_path];
            int pool_size = server_settings[ServerSetting::distributed_ddl_pool_size];
            if (pool_size < 1)
                throw Exception(ErrorCodes::ARGUMENT_OUT_OF_BOUND, "distributed_ddl.pool_size should be greater then 0");
            global_context->setDDLWorker(
                std::make_unique<DDLWorker>(
                    pool_size,
                    ddl_queue_path,
                    ddl_replicas_path,
                    global_context,
                    &config(),
                    "distributed_ddl",
                    "default",
                    "DDLWorker",
                    &CurrentMetrics::MaxDDLEntryID,
                    &CurrentMetrics::MaxPushedDDLEntryID),
                joinTasks(load_system_metadata_tasks, load_metadata_tasks));
        }

distributed_ddl.pool_size defaults to 1, i.e. one ON CLUSTER query at a time per server. The DDL queue is a serial queue.

The entry node

To enqueue a query, the initiator creates a sequential node. ZooKeeper appends a zero-padded counter to the name, and the counter is what makes the queue ordered:

cpp
// src/Interpreters/DDLWorker.cpp:1198
String DDLWorker::enqueueQueryAttempt(DDLLogEntry & entry)
{
    auto zookeeper = getZooKeeperFromContext();

    String query_path_prefix = fs::path(queue_dir) / "query-";
    zookeeper->createAncestors(query_path_prefix);

    NameSet host_ids;
    for (const HostID & host : entry.hosts)
        host_ids.emplace(host.toString());
    createReplicaDirs(zookeeper, host_ids);

    String node_path = zookeeper->create(query_path_prefix, entry.toString(), zkutil::CreateMode::PersistentSequential);

The node’s payload is the serialized DDLLogEntry — the query text, the list of hosts that should run it, the initiator’s identity, the settings, and a version number.

The versioned entry format

The entry format is explicitly versioned, because a rolling upgrade means old and new servers read the same queue:

cpp
// src/Interpreters/DDLTask.h:73
struct DDLLogEntry
{
    static constexpr const UInt64 OLDEST_VERSION = 1;
    static constexpr const UInt64 SETTINGS_IN_ZK_VERSION = 2;
    static constexpr const UInt64 NORMALIZE_CREATE_ON_INITIATOR_VERSION = 3;
    static constexpr const UInt64 OPENTELEMETRY_ENABLED_VERSION = 4;
    static constexpr const UInt64 PRESERVE_INITIAL_QUERY_ID_VERSION = 5;
    static constexpr const UInt64 BACKUP_RESTORE_FLAG_IN_ZK_VERSION = 6;
    static constexpr const UInt64 PARENT_TABLE_UUID_VERSION = 7;
    static constexpr const UInt64 INITIATOR_USER_VERSION = 8;
    /// Add new version here

    /// Remember to update the value below once new version is added
    static constexpr const UInt64 DDL_ENTRY_FORMAT_MAX_VERSION = 8;

    UInt64 version = 1;
    String query;
    std::vector<HostID> hosts;
    String initiator; // optional
    std::optional<SettingsChanges> settings;
    OpenTelemetry::TracingContext tracing_context;
    String initial_query_id;
    bool is_backup_restore = false;
    /// If present, this entry should be executed only if table with this uuid exists.
    /// Only for DatabaseReplicated.
    std::optional<UUID> parent_table_uuid;

    String initiator_user;
    Strings initiator_user_roles;

And the writer only emits the fields the version supports:

cpp
// src/Interpreters/DDLTask.cpp:158
String DDLLogEntry::toString() const
{
    WriteBufferFromOwnString wb;

    wb << "version: " << version << "\n";
    wb << "query: " << escape << query << "\n";

    bool write_hosts = version == OLDEST_VERSION || !hosts.empty();
    if (write_hosts)
    {
        Strings host_id_strings(hosts.size());
        std::transform(hosts.begin(), hosts.end(), host_id_strings.begin(), HostID::applyToString);
        wb << "hosts: " << host_id_strings << "\n";
    }

    wb << "initiator: " << initiator << "\n";

    bool write_settings = SETTINGS_IN_ZK_VERSION <= version && settings && !settings->empty();
    if (write_settings)
    {
        ASTSetQuery ast;
        ast.is_standalone = false;
        ast.changes = *settings;
        wb << "settings: " << ast.formatWithSecretsOneLine() << "\n";
    }

The user-facing knob for this is distributed_ddl_entry_format_version (default 5). A cluster running a mix of old and new servers can pin it down so that every server can parse what every other server writes.

The name padding trick

Sequential ZooKeeper node names are query- followed by a 10-digit zero-padded counter. That padding is not cosmetic — it is what makes the string order equal the numeric order:

cpp
// src/Common/ZooKeeper/ZooKeeper.cpp:2241
String getSequentialNodeName(const String & prefix, UInt64 number)
{
    /// NOTE Sequential counter in ZooKeeper is Int32.
    chassert(number < std::numeric_limits<Int32>::max());
    constexpr size_t seq_node_digits = 10;
    String num_str = std::to_string(number);
    String name = prefix + String(seq_node_digits - num_str.size(), '0') + num_str;
    return name;
}

Without the padding, query-9 would sort after query-10 as a string, and every range scan over the queue would be wrong. With it, the queue can be scanned with plain std::upper_bound / std::lower_bound on the node names, with no numeric parsing:

cpp
// src/Interpreters/DDLWorker.cpp:435
    auto begin_node = queue_nodes.begin();
    if (first_failed_task_name)
    {
        /// If we had failed tasks, then we should start from the first failed task.
        chassert(reinitialized, fmt::format("Stale first_failed_task_name={}", *first_failed_task_name));
        begin_node = std::lower_bound(queue_nodes.begin(), queue_nodes.end(), first_failed_task_name);
    }
    else
    {
        /// We had no failed tasks. Let's just choose the maximum entry we have previously seen.
        String last_task_name;
        if (!current_tasks.empty())
            last_task_name = current_tasks.back()->entry_name;
        if (last_skipped_entry_name && last_task_name < *last_skipped_entry_name)
            last_task_name = *last_skipped_entry_name;
        begin_node = std::upper_bound(queue_nodes.begin(), queue_nodes.end(), last_task_name);
    }

std::upper_bound on the string names finds “the entry after the last one I processed”. The whole resumption logic after a ZooKeeper reconnect is built on this.

The padding has a limit, and the codebase is honest about it elsewhere:

cpp
// src/Interpreters/TransactionPayloads.cpp:232
    /// Keeper pads to a minimum of ten digits and its counter is `int64_t`, so names grow past
    /// 10^10 and stop being order-comparable as strings. `zkutil::getSequentialNodeName` assumes
    /// Int32 and underflows its padding there.
    const String num = std::to_string(csn);
    return num.size() >= 10 ? "csn-" + num : "csn-" + String(10 - num.size(), '0') + num;

The host list

The entry names the hosts that should execute the query. The initiator computes it from the cluster definition:

cpp
// src/Interpreters/executeDDLQueryOnCluster.cpp:141
    /// Enumerate hosts which will be used to send query.
    auto addresses = cluster->filterAddressesByShardOrReplica(params.only_shard_num, params.only_replica_num);
    if (addresses.empty())
        throw Exception(ErrorCodes::LOGICAL_ERROR, "No hosts defined to execute distributed DDL query");

    std::vector<HostID> hosts;
    hosts.reserve(addresses.size());
    for (const auto * address : addresses)
        hosts.emplace_back(*address);

Note that the query is not rewritten per host. The same query text goes into the entry, and each host rewrites it locally (its own cluster definition, its own default database). That is what DDLTask::setClusterInfo and formatRewrittenQuery are for.

Before the entry is created, the initiator does its own access check — but against all the databases the shard might use, because each host will resolve the unqualified table name against its own default database:

cpp
// src/Interpreters/executeDDLQueryOnCluster.cpp:153
    AccessRightsElements access_to_check = params.access_to_check;
    bool need_replace_current_database = std::any_of(
        access_to_check.begin(),
        access_to_check.end(),
        [](const AccessRightsElement & elem) { return elem.isEmptyDatabase(); })
        || needsDefaultDatabaseForBareDictionaryOnCluster(query_ptr);

The per-host result directory

This is the trick I like most. Each host that runs the query creates a node named after itself under <entry>/finished/:

cpp
// src/Interpreters/DDLTask.h:155
    String getActiveNodePath() const { return fs::path(entry_path) / "active" / host_id_str; }
    String getFinishedNodePath() const { return fs::path(entry_path) / "finished" / host_id_str; }
    String getShardNodePath() const { return fs::path(entry_path) / "shards" / getShardID(); }
    String getSyncedNodePath() const { return fs::path(entry_path) / "synced" / host_id_str; }

Because the finished node’s name is the host ID, the initiator does not need to know in advance which hosts will finish, does not need to poll each one individually, and does not need a counter. It just lists the directory.

The status source does exactly that — two getChildren calls, one for finished and one for active:

cpp
// src/Interpreters/DistributedQueryStatusSource.cpp:220
        {
            auto retries_ctl = ZooKeeperRetriesControl("executeDistributedQueryOnCluster", getLogger(getName()), getRetriesInfo());
            retries_ctl.retryLoop(
                [&]()
                {
                    auto zookeeper = context->getDefaultOrAuxiliaryZooKeeper(zookeeper_name);
                    Strings paths = getNodesToWait();
                    auto res = zookeeper->tryGetChildren(paths);
                    for (size_t i = 0; i < res.size(); ++i)
                        if (res[i].error != Coordination::Error::ZOK && res[i].error != Coordination::Error::ZNONODE)
                            throw Coordination::Exception::fromPath(res[i].error, paths[i]);

                    if (res[0].error == Coordination::Error::ZNONODE)
                        node_exists = zookeeper->exists(node_path);
                    else
                        node_exists = true;
                    tmp_hosts = res[0].names;
                    tmp_active_hosts = res[1].names;

                    if (only_running_hosts)
                        offline_hosts = getOfflineHosts(waiting_hosts, zookeeper);
                });
        }

Two directories, two round trips, and the answer is “these hosts are done, these hosts are currently running”. Compare that with the alternative: N get calls, one per host, each of which can fail independently.

The finished/<host> node’s value is the serialized ExecutionStatus — the exit code and the error message:

cpp
// src/Interpreters/DDLOnClusterQueryStatusSource.cpp:39
ExecutionStatus DDLOnClusterQueryStatusSource::checkStatus(const String & host_id)
{
    fs::path status_path = fs::path(node_path) / "finished" / host_id;
    return getExecutionStatus(status_path);
}

and the status is fetched with a tryGet that distinguishes “node absent” from “node present but corrupt”:

cpp
// src/Interpreters/DistributedQueryStatusSource.cpp:138
ExecutionStatus DistributedQueryStatusSource::getExecutionStatus(const fs::path & status_path, bool * node_exists)
{
    ExecutionStatus status(-1, "Cannot obtain error message");

    String status_data;
    bool finished_exists = false;

    auto retries_ctl = ZooKeeperRetriesControl("executeDDLQueryOnCluster", getLogger("DDLQueryStatusSource"), getRetriesInfo());
    retries_ctl.retryLoop([&]() { finished_exists = context->getDefaultOrAuxiliaryZooKeeper(zookeeper_name)->tryGet(status_path, status_data); });
    if (node_exists)
        *node_exists = finished_exists;
    /// tryDeserializeText is atomic: a present-but-corrupt payload leaves the (-1) sentinel intact, so a
    /// caller pairing node_exists with the sentinel can tell an absent node from a corrupt present one.
    if (finished_exists)
        status.tryDeserializeText(status_data);

    return status;
}

The `active` node: a liveness signal

The active/<host> node is ephemeral. It exists exactly while the host is running the query, and it disappears when the host’s ZooKeeper session ends. That gives the initiator a cheap way to distinguish “slow” from “dead”:

cpp
// src/Interpreters/DistributedQueryStatusSource.cpp:77
NameSet DistributedQueryStatusSource::getOfflineHosts(const NameSet & hosts_to_wait, const ZooKeeperPtr & zookeeper)
{
    Strings paths;
    Strings hosts_array;
    for (const auto & host : hosts_to_wait)
    {
        hosts_array.push_back(host);
        paths.push_back(fs::path(replicas_path) / host / "active");
    }

    NameSet offline;
    auto res = zookeeper->tryGet(paths);
    for (size_t i = 0; i < res.size(); ++i)
    {
        if (res[i].error == Coordination::Error::ZNONODE)
            offline.insert(hosts_array[i]);

        if (res[i].data.ends_with(DatabaseReplicated::REPLICA_UNSYNCED_MARKER))
        {
            LOG_TRACE(log, "Replica {} is not fully synced after recovery, considering it as offline", hosts_array[i]);
            offline.insert(hosts_array[i]);
        }
    }

Note the REPLICA_UNSYNCED_MARKER case: a replica that came back after a crash but has not finished syncing is alive but not usable, and the initiator treats it as offline.

The active node under the replicas directory (/clickhouse/task_queue/replicas/<host>/active) is a different thing — it marks the replica as a live participant in the cluster at all, and it is also ephemeral, with the server’s UUID as the payload so that a stale node from a previous incarnation can be removed safely.

The worker’s side

The worker watches the queue directory. This is a real ZooKeeper watch, not polling — the worker sleeps until a watch fires:

cpp
// src/Interpreters/DDLWorker.cpp:410
    Strings queue_nodes = zookeeper->getChildrenWatch(
        queue_dir,
        &queue_node_stat,
        Coordination::WatchCallbackPtrOrEventPtr{queue_updated_event, ProfileEvents::ZooKeeperWatchTriggeredDistributedDDL});
    size_t size_before_filtering = queue_nodes.size();
    filterAndSortQueueNodes(queue_nodes);

with the filter dropping anything that is not a queue entry:

cpp
// src/Interpreters/DDLWorker.cpp:324
static void filterAndSortQueueNodes(Strings & all_nodes)
{
    std::erase_if(all_nodes, [] (const String & s) { return !startsWith(s, "query-"); });
    ::sort(all_nodes.begin(), all_nodes.end());
}

When the worker picks up an entry, it first claims it with an ephemeral active node whose payload is a canary — the server’s UUID:

cpp
// src/Interpreters/DDLWorker.cpp:693
    String active_node_path = task.getActiveNodePath();
    String finished_node_path = task.getFinishedNodePath();

    /// Step 1: Create ephemeral node in active/ status dir.
    /// It allows other hosts to understand that task is currently executing (useful for system.distributed_ddl_queue)
    /// and protects from concurrent deletion or the task.

    /// It will tryRemove(...) on exception
    auto active_node = zkutil::EphemeralNodeHolder::existing(active_node_path, *zookeeper);

    /// Try fast path
    const String canary_value = Field(ServerUUID::get()).dump();
    auto create_active_res = zookeeper->tryCreate(active_node_path, canary_value, zkutil::CreateMode::Ephemeral);
    if (create_active_res != Coordination::Error::ZOK)
    {
        if (create_active_res != Coordination::Error::ZNONODE && create_active_res != Coordination::Error::ZNODEEXISTS)
        {
            chassert(Coordination::isHardwareError(create_active_res));
            throw Coordination::Exception::fromPath(create_active_res, active_node_path);
        }

The canary exists to make cleanup safe: on a reconnect the worker might find its own previous ephemeral node still there, and it must delete only that one and not somebody else’s. Hence deleteEphemeralNodeIfContentMatches.

Then the DDL is executed, and the result is written in the same ZooKeeper transaction that removes the active node and creates the finished node:

cpp
// src/Interpreters/DDLWorker.cpp:749
        /// If table and database engine supports it, they will execute task.ops by their own in a single transaction
        /// with other zk operations (such as appending something to ReplicatedMergeTree log, or
        /// updating metadata in Replicated database), so we make create request for finished_node_path with status "0",
        /// which means that query executed successfully.
        task.ops.emplace_back(zkutil::makeRemoveRequest(active_node_path, -1));
        task.ops.emplace_back(zkutil::makeCreateRequest(finished_node_path, ExecutionStatus(0).serializeText(), zkutil::CreateMode::Persistent));

and the multi is issued at the end:

cpp
// src/Interpreters/DDLWorker.cpp:813
    /// Step 3: Create node in finished/ status dir and write execution status.
    /// FIXME: if server fails right here, the task will be executed twice. We need WAL here.
    /// NOTE: If ZooKeeper connection is lost here, we will try again to write query status.
    /// NOTE: If both table and database are replicated, task is executed in single ZK transaction.

    bool status_written = task.ops.empty();
    if (!status_written)
    {
        zookeeper->multi(task.ops);
        task.ops.clear();
    }

The FIXME is honest: if the server dies between “execute the DDL” and “write the finished node”, the DDL will run again on the next start. The code acknowledges the gap rather than pretending it does not exist.

Exactly-once, sort of

For a replicated database, the DDL must run on one replica of each shard, not on all of them — running ALTER TABLE ... ADD COLUMN three times is not harmless. The worker handles this with a ZooKeeper lock per shard plus an executed marker node:

cpp
// src/Interpreters/DDLWorker.cpp:879
    String shard_path = task.getShardNodePath();
    String is_executed_path = fs::path(shard_path) / "executed";
    String tries_to_execute_path = fs::path(shard_path) / "tries_to_execute";
    String max_tries_exceeded_path = fs::path(shard_path) / "max_tries_exceeded";
    chassert(shard_path.starts_with(String(fs::path(task.entry_path) / "shards" / "")));
    zookeeper->createIfNotExists(fs::path(task.entry_path) / "shards", "");
    zookeeper->createIfNotExists(shard_path, "");

    /// Leader replica creates is_executed_path node on successful query execution.
    /// We will remove create_shard_flag from zk operations list, if current replica is just waiting for leader to execute the query.
    auto create_shard_flag = zkutil::makeCreateRequest(is_executed_path, task.host_id_str, zkutil::CreateMode::Persistent);

Every replica of the shard races for the lock; whoever gets it executes the query and creates shards/<shard>/executed. The others see the node and skip. There is a retry counter (MAX_TRIES_TO_EXECUTE = 3) so that a leader that dies mid-execution does not permanently block the shard, and a one-hour ceiling (MAX_EXECUTION_TIMEOUT_SEC = 3600).

There is a subtlety in the lock acquisition that is worth reading, because “in replicated merge tree we can have multiple leaders” is a real thing:

cpp
// src/Interpreters/DDLWorker.cpp:942
        if (execute_on_single_replica_lock->tryLock())
        {
            /// In replicated merge tree we can have multiple leaders. So we can
            /// be "leader" and took lock, but another "leader" replica may have
            /// already executed this task.
            if (zookeeper->tryGet(is_executed_path, executed_by))
            {
                LOG_DEBUG(log, "Task {} has already been executed by replica ({}) of the same shard.", task.entry_name, executed_by);
                executed_by_other_leader = true;
                if (auto op = task.getOpToUpdateLogPointer())
                    task.ops.push_back(op);
                break;
            }

Taking the lock is not enough — you must also check the executed marker, because a previous leader may have run the query and died before releasing the lock.

Collecting the result

The initiator polls with a linear backoff, capped at one second:

cpp
// src/Interpreters/DistributedQueryStatusSource.cpp:213
        sleepForMilliseconds(std::min<size_t>(1000, 50 * try_number));

so the first check is after 50 ms, then 100 ms, 150 ms, … up to 1 s. The try_number counter increments per iteration, so a fast DDL is noticed quickly and a slow one is not polled aggressively.

The timeout is distributed_ddl_task_timeout, default 180 seconds:

cpp
// src/Core/Settings.cpp:3237
    DECLARE(Int64, distributed_ddl_task_timeout, 180, R"(
Sets timeout for DDL query responses from all hosts in cluster. If a DDL request has not been performed on all hosts, a response will contain a timeout error and a request will be executed in an async mode. Negative value means infinite.

Possible values:

- Positive integer.
- 0 — Async mode.
- Negative integer — infinite timeout.
)", 0) \

When the timeout hits, the behaviour depends on distributed_ddl_output_mode (default throw):

cpp
// src/Interpreters/DDLOnClusterQueryStatusSource.cpp:79
    constexpr auto msg_format = "Distributed DDL task {} is not finished on {} of {} hosts "
                                "({} of them are currently executing the task, {} are inactive). "
                                "They are going to execute the query in background. Was waiting for {:.3f} seconds{}";

Note the wording: “They are going to execute the query in background.” A timeout does not cancel the DDL. The workers keep going; the initiator just stops waiting. That is a deliberate design choice — the DDL is a durable queue entry, and abandoning it would leave the cluster in an inconsistent state.

Retries, and the “zoo” comment

Every ZooKeeper interaction in the DDL path is wrapped in a retry controller with exponential backoff. The defaults are 5 retries, 100 ms initial backoff, 5000 ms maximum:

cpp
// src/Interpreters/DistributedQueryStatusSource.cpp:157
ZooKeeperRetriesInfo DistributedQueryStatusSource::getRetriesInfo() const
{
    const auto & config_ref = Context::getGlobalContextInstance()->getConfigRef();
    return ZooKeeperRetriesInfo(
        config_ref.getInt("distributed_ddl_keeper_max_retries", 5),
        config_ref.getInt("distributed_ddl_keeper_initial_backoff_ms", 100),
        config_ref.getInt("distributed_ddl_keeper_max_backoff_ms", 5000),
        context->getProcessListElement());
}

And the retry loop only retries hardware errors:

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());
            }

The comment in scheduleTasks about reconnection is the best summary of what distributed DDL actually is:

cpp
// src/Interpreters/DDLWorker.cpp:365
            else if (task->was_executed)
            {
                /// Connection was lost on attempt to write status. Will retry.
                bool status_written = zookeeper->exists(task->getFinishedNodePath());
                /// You might think that the following condition is redundant, because status_written implies completely_processed.
                /// But it's wrong. It's possible that (!task->completely_processed && status_written)
                /// if ZooKeeper successfully received and processed our request
                /// but we lost connection while waiting for the response.
                /// Yeah, distributed systems is a zoo.
                if (status_written)
                {
                    /// TODO We cannot guarantee that query was actually executed synchronously if connection was lost.
                    /// Let's simple create synced/ node for now, but it would be better to pass UNFINISHED status to initiator
                    /// or wait for query to actually finish (requires https://github.com/ClickHouse/ClickHouse/issues/23513)
                    task->createSyncedNodeIfNeed(zookeeper);
                    task->completely_processed = true;
                }

Trade-offs

What the design buys:

The initiator is not in the data path. If it dies, the DDL still runs.
One round trip per poll, not N. Listing finished/ and active/ is two getChildren calls regardless of cluster size.
No central broker. ZooKeeper is the queue; there is nothing else to deploy.
The result set is derived, not transported. Each host publishes its own exit code; the initiator assembles the report from the filesystem metaphor.
Resumable. After a reconnect, the worker rescans the queue using the padded names and picks up where it left off.

What it costs:

At-least-once, not exactly-once. The FIXME in processTask is explicit: a crash between execution and status-writing re-runs the DDL. For replicated databases the shards/<n>/executed marker and the retry counter mitigate this, but the mitigation is best-effort.
The queue is serial. distributed_ddl.pool_size defaults to 1, so a slow ALTER blocks every subsequent ON CLUSTER query on that host.
Polling, on the initiator side. The worker uses a watch; the initiator uses a poll with a 1-second cap. For a 100-host cluster finishing in 10 ms, the initiator still waits ~50 ms.
The host list is in the entry. Every host reads the whole entry to find out whether it is in the host list, so a 1000-host cluster writes a 1000-element list into every entry.
The finished directory grows. Every entry keeps its per-host result nodes until the queue cleaner removes the entry.

Wrapping up

The one thing to remember: the result of a distributed DDL is a directory listing, not a set of RPC replies. Each host writes a node named after itself; the initiator reads the directory. That single inversion is what makes the initiator stateless with respect to the workers.

The part people get wrong: a distributed_ddl_task_timeout does not cancel anything. It only stops the initiator from waiting. The DDL is in a durable queue and will be executed whether or not anybody is watching.

What to go do about it: if ON CLUSTER feels slow, look at system.distributed_ddl_queue. It shows the queue’s entries and their state, and the gap between “created” and “finished” tells you whether you are waiting on the queue depth (pool size 1) or on a genuinely slow host.