Distributed queries

A ClickHouse cluster is just a set of independent servers configured to know about each other. There is no shared catalogue and no central coordinator (Keeper handles only metadata, not query execution). Distributed queries are routed by Distributed tables — stateless query planners that fan out to shards.

The cluster model

<remote_servers>
  <my_cluster>
    <shard>
      <replica><host>n1</host><port>9000</port></replica>
      <replica><host>n2</host><port>9000</port></replica>
    </shard>
    <shard>
      <replica><host>n3</host><port>9000</port></replica>
      <replica><host>n4</host><port>9000</port></replica>
    </shard>
  </my_cluster>
</remote_servers>

A cluster has shards; each shard has one or more replicas. The configuration is loaded by src/Interpreters/Cluster.cpp. Cluster-discovery (src/Interpreters/ClusterDiscovery.cpp) extends this by auto-populating clusters from Keeper.

Distributed tables

StorageDistributed (src/Storages/StorageDistributed.cpp, ~90 KB) is a query router with no data. Created with:

CREATE TABLE events_dist AS events
ENGINE = Distributed(my_cluster, my_db, events, sharding_key);

Reads:

1. The query is rewritten so that the local part runs against my_db.events on each shard.
2. Sub-queries are issued in parallel to one replica per shard via ClusterProxy::SelectStreamFactory.cpp.
3. Partial results are merged on the initiator (or push-down through merge-aggregated steps).

Writes (when the distributed table itself is the target):

• Sync (distributed_foreground_insert=1) — each block is split by the sharding key and forwarded directly.
• Async (default in older versions; toggled by distributed_foreground_insert) — blocks are written to a local data/<dist_table>/ queue and flushed by a background thread.

ClusterProxy

src/Interpreters/ClusterProxy/ is the layer that rewrites a query for distributed execution and creates RemoteSource processors:

• SelectStreamFactory.cpp — creates per-shard streams.
• executeQuery.cpp — picks replicas, optionally hedges, distributes mark ranges (parallel replicas).

Replica selection

load_balancing setting:

• random (default).
• nearest_hostname — by hostname distance.
• in_order — replica list order.
• first_or_random — try first; fall back to random.
• round_robin — rotates per query.

HedgedConnectionsFactory.cpp (in src/Client/) implements hedged requests: try one replica, after hedged_connection_timeout_ms start a second, take whichever responds first.

Parallel replicas

A different fan-out: a single SELECT against a ReplicatedMergeTree is split across the replicas of one shard. The leader splits mark ranges; followers pull. Toggled by allow_experimental_parallel_reading_from_replicas / max_parallel_replicas. See Replication for the read path.

DDL across the cluster

SOMETHING ON CLUSTER cluster_name runs the statement on every shard's selected replicas. It is queued in /clickhouse/task_queue/ddl/ in Keeper and consumed by DDLWorker (src/Interpreters/DDLWorker.cpp). Status is in system.distributed_ddl_queue. Replicated databases automate this away.

Insert deduplication

Because inserts can be routed to multiple replicas of a shard, ClickHouse deduplicates by hashing the inserted block (within a window of replicated_deduplication_window blocks). This makes idempotent retries safe.

Sharding key

sharding_key is an arbitrary expression. Its value modulo the shard count picks the destination shard. Using cityHash64(user_id) is typical. With internal_replication=1 the destination shard handles its own replication.

Aggregation strategies

For SELECT count(), sum(x), etc. the planner emits an AggregatingStep per shard, then a MergingAggregatedStep on the initiator. For GROUP BY with high cardinality the planner uses two-level hashing across shards. Aggregator.cpp implements the merge.

For non-additive operations (uniqExact, quantileExact, …) shards send full data; for additive ones (sum, avg, uniq, quantileTDigest, …) shards send partial states and the initiator merges.

Network compression

Inter-server traffic is compressed by default (network_compression_method, defaulting to LZ4). For wide networks ZSTD may be cheaper.

Diagnostics

• system.clusters — current topology.
• system.distributed_ddl_queue — pending ON CLUSTER work.
• system.distribution_queue — async-distributed-insert backlog.
• system.distributed_replicas — replica info.
• Add SETTINGS log_queries=1 and use system.query_log filtered by type='QueryFinish' to see how a query was decomposed.

Settings of note

• prefer_localhost_replica — read from the local replica when possible.
• distributed_aggregation_memory_efficient — streaming two-phase aggregation.
• optimize_distributed_group_by_sharding_key — push GROUP BY into shards when the key matches the sharding key.
• parallel_distributed_insert_select — parallelise INSERT ... SELECT across shards.

Related pages

• Replication
• Other engines — StorageDistributed and *Cluster variants.
• Server protocols — interserver HTTP and native TCP.