How ClickHouse Rewrites a Distributed Query for Every Shard

TL;DR: The initiator does not send one query to N shards. It sends N different queries, each derived from a clone of the original query tree. The rewrite prunes IN sets down to the values that can live on that shard, materializes the right-hand side of a GLOBAL IN/JOIN into a temporary table that travels with the query, and renames the distributed table to the remote one so the shard reads its local data.

A Distributed table is a view of a cluster. When you write

sql
SELECT count() FROM events_distributed WHERE user_id IN (1, 2, 3)

the text of that query is meaningless on a shard. On a shard there is no events_distributed; there is events_local, and it holds a fraction of the rows. The shard cannot even be asked to “read the distributed table” — it would recursively fan out again.

So the initiator must turn one query into N queries, each of which:

names the remote table, not the distributed one;
carries only the constraints that are meaningful on that shard;
carries any data that has to be broadcast, in a form the shard can read.

This is the “split the query into shards and rewrite the AST” step, and it is where a lot of distributed-query weirdness actually lives. ClickHouse does it in the query tree, not the parsed AST, and the rewrite happens in three distinct places.

text
                     initiator
                        |
        +---------------+----------------+
        |               |                |
   shard 0 query   shard 1 query    shard 2 query
        |               |                |
   clone tree       clone tree        clone tree
   + prune IN       + prune IN        + prune IN
   + materialize    + materialize     + materialize
     GLOBAL IN        GLOBAL IN         GLOBAL IN
   + rename table   + rename table    + rename table
        |               |                |
        v               v                v
   events_local     events_local      events_local

Where the rewrite happens

The fan-out lives in ClusterProxy::executeQuery (src/Interpreters/ClusterProxy/executeQuery.cpp). The interesting loop is short:

cpp
// src/Interpreters/ClusterProxy/executeQuery.cpp:601
        auto query_for_shard = query_info.query_tree->clone();
        if (sharding_key_expr && query_info.optimized_cluster && settings[Setting::optimize_skip_unused_shards_rewrite_in] && shards > 1 &&
            /// TODO: support composite sharding key
            sharding_key_expr->getRequiredColumns().size() == 1)
        {
            OptimizeShardingKeyRewriteInVisitor::Data visitor_data{
                sharding_key_expr,
                sharding_key_column_name,
                shard_info,
                not_optimized_cluster->getSlotToShard(),
            };
            optimizeShardingKeyRewriteIn(query_for_shard, std::move(visitor_data), query_info.table_expression->getAlias(), new_context);
        }

Three things to notice.

The tree is cloned per shard. Each shard gets an independent copy, so the rewrites for shard 0 cannot leak into shard 1. That is the whole reason the rewrite is safe at all.
The pruning is opt-in (optimize_skip_unused_shards_rewrite_in) and only applies to a single-column sharding key — the composite case is an explicit TODO in the source.
The shard number is decided by the unoptimized cluster. The slot_to_shard mapping comes from not_optimized_cluster, because pruning the IN set must not change which shard a value belongs to.

The sharding-key trick: pruning the `IN` set

The idea is simple. If the sharding key is cityHash64(user_id) and the shard mapping is value % slot_count, then a value of user_id = 1 can only live on one particular shard. Sending user_id IN (1, 2, 3) to all shards makes every shard scan for three keys, two of which it cannot have.

So the IN set is pruned per shard. The visitor has to decide, for each literal in the set, “could this value be on my shard?” — and that requires evaluating the sharding key expression on the value:

cpp
// src/Interpreters/OptimizeShardingKeyRewriteInVisitor.cpp:81
    UInt64 value = sharding_value.safeGet<UInt64>();
    const auto shard_num = data.slots[value % data.slots.size()] + 1;
    return data.shard_info.shard_num == shard_num;

data.slots is the cluster’s slot_to_shard vector. It is not a table of shard numbers; it is a weighted table, where each shard occupies as many consecutive slots as its configured weight:

cpp
// src/Interpreters/Cluster.cpp:514
                slot_to_shard.insert(std::end(slot_to_shard), weight, shards_info.size());

so slot_to_shard[i] is the shard number (1-based) that owns slot i, and a value is routed by value % slot_to_shard.size(). A shard with weight 3 gets three times the traffic of a shard with weight 1 — without any change to the routing formula.

The evaluation of the sharding key on a literal goes through the same ExpressionActions that the insert path uses, which is what makes the two paths agree:

cpp
// src/Interpreters/OptimizeShardingKeyRewriteInVisitor.cpp:43
    ColumnWithTypeAndName & ret = block.getByName(sharding_key_column_name);
    return (*ret.column)[0];

Note the two-step nature: the value is first converted to the sharding key’s column type, then the sharding key expression is executed over a one-row block.

The pruning itself is a rewrite of the IN function’s right-hand side, and it has to cope with two different syntactic forms — a tuple function and a tuple literal:

cpp
// src/Interpreters/OptimizeShardingKeyRewriteInVisitor.cpp:117
        if (auto * tuple_func = right->as<ASTFunction>(); tuple_func && tuple_func->name == "tuple")
        {
            auto * tuple_elements = tuple_func->children.front()->as<ASTExpressionList>();
            std::erase_if(tuple_elements->children, [&](auto & child)
            {
                auto * literal = child->template as<ASTLiteral>();
                return tuple_elements->children.size() > 1 && literal && !shardContains(literal->value, name, data);
            });
        }

and the query-tree version does the same for a ConstantNode:

cpp
// src/Interpreters/OptimizeShardingKeyRewriteInVisitor.cpp:230
                for (const auto & child : tuple)
                {
                    if (shardContains(child, name, data))
                        new_tuple.push_back(child);
                }

                if (new_tuple.empty())
                    new_tuple.push_back(tuple.back());

That last if is a deliberate lie. If every value in the IN set belongs to some other shard, the pruned set would be empty — and x IN () is not valid SQL, and would change the meaning of the query from “these rows” to “no rows”. So the code keeps one arbitrary element. The result is a shard that returns a possibly non-empty result for a set that contains none of its values… which is fine, because the initiator merges the shards and the extra shard contributes nothing: the rows it returns are filtered by the same IN predicate on the shard. The IN was never the only filter; it is a filter plus a sharding hint.

The visitor is also careful about where in the tree it rewrites. It only touches the filter clauses and the join tree:

cpp
// src/Interpreters/OptimizeShardingKeyRewriteInVisitor.cpp:170
    static bool needChildVisit(QueryTreeNodePtr & parent, QueryTreeNodePtr & child)
    {
        if (const auto * query_node = parent->as<QueryNode>())
            return child == query_node->getWhere() || child == query_node->getPrewhere() || child == query_node->getJoinTreeNode();

        if (const auto * join_node = parent->as<JoinNode>())
            return child == join_node->getLeftTableExpressionNode() || child == join_node->getRightTableExpressionNode();

        if (const auto * array_join_node = parent->as<ArrayJoinNode>())
            return child == array_join_node->getTableExpressionNode();

        if (parent->as<TableFunctionNode>())
            return false;

        return true;
    }

This is not fussiness. The IN set is also part of the expression’s identity: if the pruned set has a different size, the expression’s name changes, and the initiator matches the shard’s returned columns by name. A pruned set that leaked into the projection would produce a column the initiator cannot match, and the query would fail with NOT_FOUND_COLUMN_IN_BLOCK.

There is one more subtlety, and it is the same class of bug: an aliased node is shared between the clauses that reference it. Rewriting it through a filter would rewrite it in the projection too:

cpp
// src/Interpreters/OptimizeShardingKeyRewriteInVisitor.cpp:193
        /// An aliased node is shared between the clauses that reference the alias, so rewriting it
        /// through a filter would also rewrite it in the projection or in `ORDER BY`.
        if (node->hasAlias())
            return;

The table trick: `GLOBAL IN` and `GLOBAL JOIN`

Now the harder half. Consider:

sql
SELECT * FROM events_distributed WHERE user_id IN (SELECT user_id FROM blacklist)

The subquery is on a different table, possibly a different cluster. If each shard ran the subquery itself, the shards would have to read blacklist themselves — which may not even exist there, and would multiply the work by the number of shards.

The GLOBAL variants solve this the only way you can solve it over a network: materialize the right-hand side on the initiator and ship it with the query. ClickHouse calls this the “temporary table” trick, and it is the single most expensive rewrite in the distributed path.

The decision of whether to globalize is made by DistributedProductModeRewriteInJoinVisitor, whose name is a mouthful for a policy:

distributed_product_mode = 'deny' (the default) — a local IN/JOIN against a Distributed table is an error.
distributed_product_mode = 'local' — the right-hand side is read locally, which only works if it is on the same node.
distributed_product_mode = 'global' — the right-hand side is materialized on the initiator and broadcast.

prefer_global_in_and_join = 1 upgrades local to global automatically. The rewrite is done on the tree in buildQueryTreeForShard (src/Storages/buildQueryTreeForShard.cpp):

cpp
// src/Storages/buildQueryTreeForShard.cpp:987
QueryTreeNodePtr buildQueryTreeForShard(const PlannerContextPtr & planner_context, QueryTreeNodePtr query_tree_to_modify, bool allow_global_join_for_right_table)
{
    CollectColumnSourceToColumnsVisitor collect_column_source_to_columns_visitor;
    collect_column_source_to_columns_visitor.visit(query_tree_to_modify);

    const auto & column_source_to_columns = collect_column_source_to_columns_visitor.getColumnSourceToColumns();

    DistributedProductModeRewriteInJoinVisitor visitor(planner_context->getQueryContext(), allow_global_join_for_right_table);
    visitor.visit(query_tree_to_modify);

    auto replacement_map = visitor.getReplacementMap();
    const auto & global_in_or_join_nodes = visitor.getGlobalInOrJoinNodes();

and the visitor rewrites a local IN into a globalIn and records that the right-hand side must be materialized:

cpp
// src/Storages/buildQueryTreeForShard.cpp:423
        else if ((distributed_product_mode == DistributedProductMode::GLOBAL || getSettings()[Setting::prefer_global_in_and_join]) &&
            !in_function_or_join_stack.empty())
        {
            auto * in_or_join_node_to_modify = in_function_or_join_stack.back().query_node.get();

The materialization itself is executeSubqueryNode. It runs the subquery on the initiator, writes the result into a TemporaryTableHolder, and registers it as an external table:

cpp
// src/Storages/buildQueryTreeForShard.cpp:693
    auto external_storage_holder = TemporaryTableHolder(
        mutable_context,
        ColumnsDescription(columns),
        ConstraintsDescription{},
        nullptr /*query*/,
        true /*create_for_global_subquery*/);

cpp
// src/Storages/buildQueryTreeForShard.cpp:750
    mutable_context->addExternalTable(temporary_table_name, std::move(external_storage_holder));

The temporary table’s name is derived from the hash of the subquery tree, so two identical subqueries in one query are materialized once:

cpp
// src/Storages/buildQueryTreeForShard.cpp:655
    const auto subquery_hash = subquery_node->getTreeHash();
    const auto temporary_table_name = fmt::format("_data_{}", toString(subquery_hash));

and the reuse is explicit at the top of the function:

cpp
// src/Storages/buildQueryTreeForShard.cpp:658
    const auto & external_tables = mutable_context->getExternalTables();
    auto external_table_it = external_tables.find(temporary_table_name);
    if (external_table_it != external_tables.end())
    {
        auto temporary_table_expression_node = std::make_shared<TableNode>(external_table_it->second, mutable_context);
        temporary_table_expression_node->setTemporaryTableName(temporary_table_name);
        return temporary_table_expression_node;
    }

This is where the “table trick” earns its name. The subquery is not sent as SQL. It is sent as data: a block stream that is shipped over the same connection as the query, ahead of the query’s own results, and materialized on the shard into a temporary table that the query then reads.

The materialization is not unbounded. The transfer limits apply here, which is why the code goes out of its way to reuse the machinery that implements them:

cpp
// src/Storages/buildQueryTreeForShard.cpp:725
    SizeLimits network_transfer_limits(
        subquery_settings[Setting::max_rows_to_transfer],
        subquery_settings[Setting::max_bytes_to_transfer],
        subquery_settings[Setting::transfer_overflow_mode]);

Both max_rows_to_transfer and max_bytes_to_transfer default to 0, and a zero limit means “no check” — so the default behaviour is “materialize whatever the subquery returns, no matter how big”, which is exactly the footgun this mechanism is famous for.

The renaming trick: `events_distributed` becomes `events_local`

The last rewrite is the simplest and the most confusing. The query text that arrives at the shard must not mention the distributed table.

In the query-tree path, this is handled by the fact that the table node in the tree points at the remote storage. In the text path, it is an AST substitution, done by RestoreQualifiedNamesVisitor:

cpp
// src/Interpreters/TranslateQualifiedNamesVisitor.cpp:481
void RestoreQualifiedNamesMatcher::visit(ASTIdentifier & identifier, ASTPtr &, Data & data)
{
    if (IdentifierSemantic::getColumnName(identifier))
    {
        if (IdentifierSemantic::getMembership(identifier))
        {
            identifier.restoreTable();  // TODO(ilezhankin): should restore qualified name here - why exactly here?
            data.changeTable(identifier);
        }
    }
}

and changeTable swaps the qualified name:

cpp
// src/Interpreters/TranslateQualifiedNamesVisitor.cpp:452
void RestoreQualifiedNamesMatcher::Data::changeTable(ASTIdentifier & identifier) const
{
    auto match = IdentifierSemantic::canReferColumnToTable(identifier, distributed_table);
    switch (match)
    {
        case IdentifierSemantic::ColumnMatch::AliasedTableName:
        case IdentifierSemantic::ColumnMatch::TableName:
        case IdentifierSemantic::ColumnMatch::DBAndTable:
            IdentifierSemantic::setColumnLongName(identifier, remote_table);
            break;

So the query the shard receives reads like:

sql
SELECT count() FROM events_local WHERE user_id IN (2)

— the local table, with the IN set pruned. Two rewrites, one text.

There is a third form of the same trick, for the cluster/clusterAllReplicas table functions, where the whole cluster is addressed with a table function rather than a table name. There, the substitution goes the other way: the function is replaced with the name of the table on the remote side.

The housekeeping rewrites

Two more rewrites happen on the way out, both of them about making the shard’s re-analysis of the query produce the same result as the initiator’s.

Alias renumbering. The query tree carries __tableN aliases numbered by the whole outer query’s analysis. A replica re-analyzes the query from scratch, and its numbering starts at 1 again:

cpp
// src/Interpreters/ClusterProxy/executeQuery.cpp:1403
    /// The query tree carries `__tableN` table aliases numbered by the whole outer query's analysis.
    /// When this read is nested inside a subquery the target table is `__tableK` (K > 1), but a replica
    /// re-analyzes the query sent to it from scratch and `createUniqueAliasesIfNecessary` restarts the
    /// numbering at 1, so its result columns (`__table1.*`) would not match an initiator header computed
    /// from the original tree (NOT_FOUND_COLUMN_IN_BLOCK). Renumber the tree the same way here so the
    /// header and the query sent to replicas agree (this mirrors `buildQueryTreeForShard`, which the
    /// task-based parallel-replicas path runs for the same reason).

Settings stripping. The initiator-only settings must not be forwarded:

cpp
// src/Interpreters/ClusterProxy/executeQuery.cpp:173
void stripInitiatorOnlySettings(Settings & settings)
{
    /// All of the settings below are interpreted only at the initiator: they either shape the
    /// final query (`select`, `order`, `sort`, `filter`, `limit`, `offset`, `page`,
    /// `additional_result_filter`) or shape how the result is serialised to the user
    /// (`format`, `output_format`, `default_format`, `compression`). Forwarding them to remote
    /// shards is at best wasted work and at worst breaks distributed queries — `format = 'Null'`,
    /// for example, sets `null_format = true` on each shard, suppressing TCP `sendData` and
    /// producing empty blocks; `getStructureOfRemoteTable` then `continue`s without throwing
    /// `NetException`, leaving `fail_messages` empty and surfacing as
    /// `NO_REMOTE_SHARD_AVAILABLE. Log: ` with an empty body. The query-shaping settings would
    /// similarly cause the per-shard subquery to be re-shaped a second time. Strip the settings
    /// here so the inter-server `Settings` packet does not carry them. This is shared by the
    /// `Distributed` fan-out and the `*Cluster` table functions (`IStorageCluster`), which both
    /// materialize these settings on the initiator before reaching the remote servers.

That comment is worth reading twice. format = 'Null' on a shard means “do not send data”, which turns a working distributed query into NO_REMOTE_SHARD_AVAILABLE with an empty error message — the kind of bug that costs a day.

The SETTINGS clause of the query text is dropped too, for the same reason plus one more: a newly-introduced setting would make an older shard reject the query with UNKNOWN_SETTING:

cpp
// src/Storages/buildQueryTreeForShard.cpp:1137
    // Get rid of the settings clause so we don't send them to remote. Thus newly non-important
    // settings won't break any remote parser. It's also more reasonable since the query settings
    // are written into the query context and will be sent by the query pipeline.

Putting it together

text
  initiator: SELECT * FROM dist WHERE user_id IN (1,2,3)
        |
        |  ClusterProxy::executeQuery
        v
  for each shard:
        |
        |  1. clone the query tree
        |  2. prune the IN set by the sharding key      (optimize_sharding_key_rewrite_in)
        |  3. materialize GLOBAL IN / JOIN sides          (buildQueryTreeForShard)
        |        -> run the subquery here, ship the rows as a temporary table
        |  4. rename the table to the remote one         (RestoreQualifiedNamesVisitor)
        |  5. renumber __tableN aliases                  (createUniqueAliasesIfNecessary)
        |  6. strip the SETTINGS clause and the initiator-only settings
        v
  shard 0 gets: SELECT * FROM events_local WHERE user_id IN (2)
  shard 1 gets: SELECT * FROM events_local WHERE user_id IN (1, 3)

Trying it

sql
-- Look at the per-shard queries: the Distributed table is resolved to its
-- remote counterpart, and the IN set is pruned.
EXPLAIN SYNTAX
SELECT count() FROM events_distributed WHERE user_id IN (1, 2, 3, 4, 5, 6, 7, 8);

-- Turn the pruning off and compare the plans.
SELECT count() FROM events_distributed WHERE user_id IN (1, 2, 3, 4, 5, 6, 7, 8)
SETTINGS optimize_skip_unused_shards_rewrite_in = 0;

Trade-offs

What the rewriting buys you:

Correctness of the table reference. The shard reads its local table; the query text on the wire is a valid single-node query.
Less work per shard. The IN pruning turns a broadcast of the whole key set into a broadcast of the shard’s own keys. With a 100-shard cluster and a 1000-element IN list, that is a 100× reduction in index lookups.
One place for the broadcast semantics. GLOBAL IN/JOIN are materialized once, on the initiator, and the same bytes go to every shard.
Rolling-upgrade safety. Settings and syntax that the target shard might not understand are stripped before the query is sent.

What it costs:

A clone per shard. For a 100-shard cluster, 100 copies of the query tree, 100 analyses’ worth of rewrite work, on the initiator, before the first byte goes out.
The broadcast is a real broadcast. GLOBAL IN with a million-row subquery sends a million rows to every shard. max_rows_to_transfer is the only guard, and it is off by default for bytes.
The pruned IN set is a lie, deliberately. The new_tuple.push_back(tuple.back()) fallback means the shard’s filter text no longer matches the initiator’s. The query is still correct, because the shard re-applies the same predicate to the rows it returns, but reading the shard’s query text and expecting it to be the initiator’s query will confuse you.
Alias renumbering is a fragile contract. The initiator and the shard must agree on the __tableN numbering, and the code has to fix it up explicitly in two different places (buildQueryTreeForShard and the parallel-replicas path) because the two renumber independently.

The part people get wrong: distributed_product_mode = 'deny' does not mean “the product is denied at the shard”. It means the initiator refuses to plan a local IN/JOIN across a Distributed table, because the answer would be wrong. Setting it to 'local' does not make the query correct on a multi-shard cluster; it makes the initiator read the right-hand side from the local node and hope it is the same everywhere. prefer_global_in_and_join = 1 is the setting that actually changes the semantics, at the cost of the broadcast.