How does ClickHouse execute a query against a Distributed table across shards?

## The two table layers A ClickHouse cluster separates storage from routing. Each shard holds a **local** MergeTree table — say `events_local`. On top sits a `Distributed` table, `events`, declared with the cluster name, the database, the local table's name and a sharding key. The distributed table stores no data of its own; it knows how to fan a query out and how to route inserts. ## The execution flow 1. A client sends a query to some node; that node becomes the **initiator**. 2. The initiator analyses the query and **rewrites** it so the distributed table is replaced by the local table name. 3. It sends that rewritten query to one replica of each shard (choosing among replicas by a load-balancing policy; `prefer_localhost_replica` makes it use its own copy when it hosts one). 4. Each shard executes locally, in full parallel, applying its own `max_threads` — the setting is per server, so a five-shard cluster with `max_threads = 8` may run forty query threads plus the initiator's. 5. Shards stream results back to the initiator, which performs the final combination and returns to the client. In a query plan on the initiator, the remote legs appear as a read-from-remote step; EXPLAIN there does not expand into each shard's own pipeline. To see a shard's plan you run EXPLAIN against the local table on that shard. ## Partial aggregation states are the key idea For a `GROUP BY`, the shards do not ship rows. Each shard aggregates its own data and returns **intermediate aggregation state** — for `sum` that is a running total per group, for a distinct-count sketch it is the sketch itself — and the initiator merges the states per group. This is what makes MPP aggregation cheap: network traffic scales with distinct groups, not with rows scanned. It also explains the failure mode. If the grouping key has tens of millions of distinct values, every shard sends a large state set and the **initiator** must hold and merge all of them on one machine. The memory limit that fires will be on the initiator, not the shards, and the fix is to reduce cardinality, not to add shards. Several settings shape this. `distributed_group_by_no_merge` skips the merge entirely — correct only when the data is already sharded such that no group spans shards. `optimize_distributed_group_by_sharding_key` lets the engine apply that reasoning automatically when the grouping key is compatible with the sharding key. `distributed_aggregation_memory_efficient` merges incoming states in a streaming fashion rather than accumulating everything first. ## The GLOBAL trap This is the correctness question interviewers actually ask. Consider: ```sql SELECT count() FROM events WHERE user_id IN (SELECT user_id FROM vip_users); ``` If `vip_users` is itself distributed, the rewritten query sent to each shard evaluates the subquery **against that shard's local slice** of `vip_users`. Each shard therefore filters against a different, partial set, and the total is wrong — quietly, with no error. `GLOBAL IN` changes the protocol: the initiator evaluates the subquery once, materialises the result as a temporary table, and ships it to every shard, so all shards filter against the same complete set. `GLOBAL JOIN` does the same for the right-hand side of a join. The cost is the broadcast: the temporary set is sent to every shard, so it must be small enough that shipping it is cheaper than the alternative. `distributed_product_mode` governs how the server treats such distributed subqueries — from allowing them through to rejecting them outright — and setting it to reject is a reasonable guardrail on a cluster where this mistake keeps happening. ## Sorting, limits and skew A global `ORDER BY ... LIMIT n` pushes a local sort and limit to each shard, then merges the pre-limited streams on the initiator — efficient. A global `ORDER BY` with no limit, or one over a large result, concentrates the sort on the initiator. Data skew is the other operational hazard: if the sharding key sends a disproportionate share of rows to one shard, that shard becomes the straggler and the query is as slow as it is, no matter how idle the rest are. Compare per-shard rows in `system.parts` on each node, and compare per-shard timings by joining the shard rows in `system.query_log` on `initial_query_id`. ## Observing it Every participating server writes its own `system.query_log` rows. On the initiator the user's query has `is_initial_query = 1`; each shard's sub-query has the same `initial_query_id` with `is_initial_query = 0`. That join is how you find out which shard was slow, which one read the most, and whether the imbalance is data or hardware.