ClickHouse Distributed Tables

A Distributed table stores no data. It routes queries to the underlying local tables on each shard and merges the results.

Setup

<!-- config.xml: the cluster topology -->
<remote_servers>
    <main>
        <shard>
            <internal_replication>true</internal_replication>
            <replica><host>ch-1a</host><port>9000</port></replica>
            <replica><host>ch-1b</host><port>9000</port></replica>
        </shard>
        <shard>
            <internal_replication>true</internal_replication>
            <replica><host>ch-2a</host><port>9000</port></replica>
            <replica><host>ch-2b</host><port>9000</port></replica>
        </shard>
    </main>
</remote_servers>

-- Local table on every node.
CREATE TABLE events_local ON CLUSTER main
(
    event_time DateTime, tenant_id UInt64, event_type LowCardinality(String)
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events_local', '{replica}')
ORDER BY (tenant_id, event_time);
 
-- Distributed table that fans out across shards.
CREATE TABLE events ON CLUSTER main AS events_local
ENGINE = Distributed(main, currentDatabase(), events_local, cityHash64(tenant_id));

Sharding

The last argument to Distributed is the sharding key expression. cityHash64(tenant_id) distributes by tenant, so all of one tenant's rows land on one shard.

Choosing it follows the usual rules — see Sharding:

• High cardinality and even distribution, or one shard becomes hot.
• Present in the common queries, so a query can target one shard instead of all of them.
• Aligned with joins: a distributed join is far cheaper when both tables are sharded on the join key, because no data has to move.

Sharding by a monotonically increasing timestamp is the classic mistake: every current write goes to one shard.

Insert paths

Two options, and the second is usually better:

1. Insert into the Distributed table. It buffers rows locally and forwards them asynchronously (distributed_foreground_insert = 0 by default). Convenient, but the buffer is a durability gap and a source of surprise when a node is lost.
2. Insert directly into the local table on a node chosen by the client. The application does the sharding, which is more code and removes the buffer entirely.

SELECT * FROM system.distribution_queue;   -- pending forwards from Distributed inserts

Query behaviour

The initiating node sends a rewritten subquery to each shard, receives partial results and merges them. Aggregations combine cleanly. The patterns that do not scale well:

• GROUP BY with very high cardinality — every shard returns a large partial result to be merged on one node. distributed_group_by_no_merge and two-level aggregation settings help.
• Joins where the right table is not co-located — the right side is broadcast to every shard. Use dictionaries for small dimension tables, or shard both tables on the join key.
• ORDER BY ... LIMIT without the sharding key — every shard must return its candidates.

-- Which shards a query touched and how much each read.
SELECT hostName() AS host, count(), sum(read_rows)
FROM clusterAllReplicas(main, system.query_log)
WHERE type = 'QueryFinish' AND event_time > now() - INTERVAL 10 MINUTE
GROUP BY host;

Adding shards