ClickHouse Sharding Troubleshooting and Performance Optimization

Most ClickHouse sharding troubleshooting comes down to three tables and one question: is the problem in the write path (the Distributed engine’s asynchronous send queue), the data layout (an uneven or wrong sharding key), or the read path (the initiator doing work the shards should have done)? system.distribution_queue, system.parts queried through clusterAllReplicas(), and system.query_log joined on initial_query_id answer that question in under ten minutes.

What follows is the ClickHouse sharding troubleshooting field guide we use on ChistaDATA support calls, written against ClickHouse 26.3 LTS on self-managed clusters (ReplicatedMergeTree locals behind a Distributed table, ClickHouse Keeper for coordination). Where behaviour changed in a specific release, the version is called out inline.

The ClickHouse sharding topology this guide assumes

A conventional ClickHouse sharding layout: two shards, two replicas per shard, a three-node Keeper ensemble on its own hosts, and a Distributed table on every node so any node can act as the query initiator. The cluster definition that matters most for the rest of this article is the internal_replication flag; get it wrong and section two of this guide is your Monday morning.

<!-- /etc/clickhouse-server/config.d/cluster.xml (per node) -->
<clickhouse>
  <remote_servers>
    <events_cluster>
      <shard>
        <internal_replication>true</internal_replication>
        <replica><host>ch-s1-r1</host><port>9000</port></replica>
        <replica><host>ch-s1-r2</host><port>9000</port></replica>
      </shard>
      <shard>
        <internal_replication>true</internal_replication>
        <replica><host>ch-s2-r1</host><port>9000</port></replica>
        <replica><host>ch-s2-r2</host><port>9000</port></replica>
      </shard>
    </events_cluster>
  </remote_servers>
</clickhouse>

CREATE TABLE events_local ON CLUSTER events_cluster
(
    event_date   Date,
    event_time   DateTime64(3, 'UTC'),
    tenant_id    UInt32,
    user_id      UInt64,
    event_type   LowCardinality(String),
    payload      String
)
ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/events_local', '{replica}')
PARTITION BY toYYYYMM(event_date)
ORDER BY (tenant_id, event_time, user_id)
SETTINGS index_granularity = 8192;

CREATE TABLE events_dist ON CLUSTER events_cluster
AS events_local
ENGINE = Distributed('events_cluster', 'default', 'events_local', cityHash64(tenant_id));

Every ClickHouse sharding troubleshooting case below is presented the same way: the symptom as you meet it, the two or three commands that localise it, the mechanism, the fix in stages, and what to monitor so it does not come back.

1. ClickHouse sharding troubleshooting: inserts succeed but rows never arrive on the shards

Symptom. The application reports successful INSERT INTO events_dist, row counts on the Distributed table lag by minutes or hours, and the initiator’s data disk is filling up. On a bad day you also see:

Code: 243. DB::Exception: Cannot reserve 1.00 MiB, not enough space. (NOT_ENOUGH_SPACE)

Triage. ClickHouse sharding troubleshooting starts at the send queue. By default the Distributed engine acknowledges an insert once it has written the batch to a local .bin file under data/<db>/events_dist/shard<N>_replica<M>/. A background thread then ships those files. If shipping stalls, the acknowledgement was a promise the initiator cannot keep yet.

SELECT
    database,
    table,
    data_path,
    is_blocked,
    error_count,
    data_files,
    formatReadableSize(data_compressed_bytes) AS pending,
    broken_data_files,
    last_exception_time,
    substring(last_exception, 1, 160)         AS last_exception
FROM system.distribution_queue
ORDER BY data_compressed_bytes DESC;

Healthy looks like data_files in the low tens and pending in megabytes, draining between refreshes. Afflicted looks like thousands of files, gigabytes pending, a non-zero error_count, and a last_exception that names the real problem, most often Code: 210. Connection refused, an authentication failure after a password rotation, or Code: 252. Too many parts coming back from the target shard.

Mechanism. Each Distributed insert is split by the sharding key, and each shard’s slice becomes one file. With small application batches you create one small file per shard per insert, and the target shard then has to create one part per file. In ClickHouse sharding the send queue is therefore a part-count amplifier: an application inserting 200 rows every 50 ms against two shards produces 40 parts per second downstream before merges even start.

The Too many parts error from the shard is not a limit to raise on the shard; it is the send queue telling you the batching is wrong. Our earlier note on high-velocity ingestion through Distributed tables covers the parameter surface in more depth.

Fix, staged. Immediate mitigation is to flush and confirm the queue drains:

SYSTEM FLUSH DISTRIBUTED events_dist;

-- verify
SELECT table, data_files, error_count
FROM system.distribution_queue
WHERE table = 'events_dist';

If is_blocked = 1 persists after the exception is resolved, SYSTEM START DISTRIBUTED SENDS events_dist restarts the sender. Broken files (broken_data_files > 0) have been moved to a broken/ subdirectory on disk; they will not be retried automatically, and whether you replay or discard them is a data-loss decision that belongs to the owning team, not the DBA.

The proper fix is to make the engine batch for you and, at high volume, to stop routing writes through the Distributed table at all:

-- Table-level: coalesce queued files into larger sends (26.3 LTS names)
ALTER TABLE events_dist
    MODIFY SETTING
        background_insert_batch = 1,
        background_insert_split_batch_on_failure = 1;

-- Session/profile level for the ingestion user
SET distributed_background_insert_batch = 1;

At sustained rates above a few thousand inserts per second per initiator, the ingestion layer should compute cityHash64(tenant_id) % 2 itself and insert directly into events_local on the correct shard. That removes the intermediate disk write, makes backpressure visible to the producer, and is the pattern we default to in ChistaDATA managed estates. Note that since 26.3 LTS async_insert is on by default; if your fleet crossed that version boundary recently, check system.asynchronous_insert_log before assuming the Distributed queue is the only buffer in play.

Prevention. Alert on data_compressed_bytes per Distributed table above a threshold you derive from disk headroom (we use 5 % of the data volume), and on error_count increasing across two consecutive scrapes. Both are cheap to scrape from system.distribution_queue.

2. ClickHouse sharding troubleshooting: row counts are exactly double after a cluster rebuild

Symptom. SELECT count() FROM events_dist returns twice the number of rows inserted. Sometimes the discrepancy is not exactly double, because some inserts landed while a replica was down.

Triage.

SELECT
    hostName()  AS host,
    count()     AS rows
FROM clusterAllReplicas('events_cluster', default, events_local)
GROUP BY host
ORDER BY host;

In a healthy ClickHouse sharding setup, replicas within a shard agree on rows. If they agree and the Distributed count is still double, read the cluster XML: internal_replication is false (or absent, which is the same thing).

Mechanism. With internal_replication = false, the Distributed engine writes every batch to every replica in the shard. That is correct ClickHouse sharding behaviour when the local tables are plain MergeTree and the Distributed table is the only replication mechanism. With ReplicatedMergeTree locals, each replica also receives the other replica’s copy through the Keeper replication log, so every row exists twice.

Insert deduplication by block hash usually masks this for identical blocks, which is why the count is sometimes not exactly double: the dedup window (replicated_deduplication_window, default 1000 blocks on 26.3) only covers recent inserts, and files sent at different times carry different block ids.

Fix. Set internal_replication to true on every shard, reload with SYSTEM RELOAD CONFIG on every node, then deduplicate the historical data. The only safe dedup on a sharded ReplicatedMergeTree is per partition with a confirmation gate, in a maintenance window, after a verified backup:

-- Verification before: row count per partition, per replica
SELECT partition, sum(rows)
FROM system.parts
WHERE table = 'events_local' AND active
GROUP BY partition
ORDER BY partition;

-- Gate: proceed only if the affected partitions are confirmed in writing
OPTIMIZE TABLE events_local PARTITION '202608' FINAL DEDUPLICATE;

-- Validation after: repeat the count and compare with the source of truth

OPTIMIZE ... DEDUPLICATE rewrites the whole partition; on a multi-terabyte partition that is hours of merge I/O on every replica in the shard, so budget for it in system.merges and stage it partition by partition.

3. Uneven ClickHouse sharding: one shard runs hot, the others idle

Symptom. p95 latency on events_dist queries is set by a single node. CPU on ch-s1-* sits at 90 % while ch-s2-* rests at 20 %. Merges on shard 1 fall behind.

Triage. ClickHouse sharding troubleshooting for skew means measuring the layout, not the CPU graph:

SELECT
    hostName()                              AS host,
    sum(rows)                               AS rows,
    formatReadableSize(sum(bytes_on_disk))  AS on_disk,
    count()                                 AS active_parts
FROM clusterAllReplicas('events_cluster', system.parts)
WHERE table = 'events_local' AND active
GROUP BY host
ORDER BY host;

For ClickHouse sharding, rows per shard within 10 % of each other is balanced. A 60/40 split from a hash key is worth investigating; anything beyond that, or a 95/5 split, means the key has low cardinality or a dominant value. Confirm with the key’s distribution:

SELECT
    cityHash64(tenant_id) % 2 AS shard_slot,
    count()                    AS rows,
    uniq(tenant_id)            AS tenants
FROM events_dist
GROUP BY shard_slot;

Mechanism. In ClickHouse sharding, the key is a modulo over the shard weights, so it inherits the skew of the column it hashes. cityHash64(tenant_id) is uniform across tenants but not across rows: one tenant producing 40 % of traffic lands 40 % of rows on one shard, forever. The trade-off every ClickHouse sharding key makes is query locality against balance. Keeping a tenant on one shard lets optimize_skip_unused_shards prune the other shard for tenant-scoped queries and keeps per-tenant joins local; splitting tenants across shards balances load and destroys both properties. There is no free option.

Fix. For a handful of heavy tenants, a composite key preserves locality for everyone else:

CREATE TABLE events_dist_v2 ON CLUSTER events_cluster
AS events_local
ENGINE = Distributed(
    'events_cluster', 'default', 'events_local',
    cityHash64(tenant_id, intDiv(toUnixTimestamp(event_time), 3600))
);

That spreads each tenant’s data across shards by hour while keeping an hour’s worth of one tenant together. Queries filtered by tenant_id alone can no longer be pruned to one shard, and you should say so to whoever owns the dashboards. Moving already-written data between shards is a separate operation; the honest options are covered in our shard rebalancing methods write-up, and the aggregation-side consequences in troubleshooting data skew in distributed aggregation.

Prevention. Put the per-host rows query on a weekly schedule and record the ratio. ClickHouse sharding skew drifts as customers grow; catching it at 60/40 is a key change, catching it at 90/10 is a resharding project.

4. ClickHouse sharding troubleshooting: queries are slower than the sum of their shards

Symptom. The same aggregation takes 400 ms directly on events_local on each shard and 6 seconds through events_dist. Or the initiator throws:

Code: 241. DB::Exception: Memory limit (for query) exceeded: would use 28.5 GiB
(attempt to allocate chunk of 4194304 bytes), maximum: 28.0 GiB. (MEMORY_LIMIT_EXCEEDED)

Triage. The most useful ClickHouse sharding troubleshooting fact is that every shard-side query carries the initiator’s query id, so you can see exactly where the time went:

SELECT
    hostName()                 AS host,
    is_initial_query,
    query_duration_ms,
    read_rows,
    result_rows,
    formatReadableSize(memory_usage) AS mem,
    ProfileEvents['SelectedMarks']   AS selected_marks
FROM clusterAllReplicas('events_cluster', system.query_log)
WHERE initial_query_id = '${QUERY_ID}'
  AND type = 'QueryFinish'
ORDER BY is_initial_query DESC, host;

Two ClickHouse sharding patterns account for most cases. First, result_rows on the shards is enormous (millions of partial-aggregate groups shipped to the initiator), and the initiator’s row shows most of the duration and memory: the merge phase is the bottleneck. Second, every shard shows read_rows equal to its full table even though the query has a WHERE tenant_id = 42: no shard pruning happened.

Mechanism and fix for the merge bottleneck. Distributed aggregation in ClickHouse sharding is two-phase: shards return aggregate states per group, the initiator merges them. When the group cardinality is high (a GROUP BY user_id over 50 million users), the initiator receives 50 million states from each shard and merges them single-handedly, on one node, in memory. Three levers, in order of preference:

-- (a) If GROUP BY contains the sharding key, groups cannot span shards:
--     let each shard finish its own aggregation and skip the merge.
SET optimize_distributed_group_by_sharding_key = 1;

-- (b) Stream states from shards and merge incrementally (default 1 on 26.3;
--     confirm it has not been pinned off in a profile)
SET distributed_aggregation_memory_efficient = 1;

-- (c) Spill the merge to disk rather than fail
SET max_bytes_before_external_group_by = 10000000000;  -- 10 GB, bytes

Lever (a) is the structural ClickHouse sharding win, and it is the strongest argument for choosing a sharding key that matches the dominant GROUP BY. For queries that need the full merge, distributed_group_by_no_merge = 2 is a diagnostic rather than a fix: it returns the per-shard results unmerged so you can see how much state each shard is contributing.

Mechanism and fix for missing shard pruning. The initiator can skip shards only when the WHERE clause constrains the sharding key with equality or IN, the sharding key expression is deterministic, and optimize_skip_unused_shards is on (it is off by default). Turn it on per profile, and use the force variant in CI so a query that silently fans out to every shard fails loudly:

SET optimize_skip_unused_shards = 1;
SET force_optimize_skip_unused_shards = 1;   -- throws if pruning is impossible

EXPLAIN
SELECT count()
FROM events_dist
WHERE tenant_id = 42;

Compare SelectedMarks and read_rows per shard before and after; the pruned shard should not appear in system.query_log at all. For the shard-side plan itself, EXPLAIN PIPELINE on events_local tells you whether the scan is granule-efficient; our EXPLAIN PIPELINE guide walks through reading that output.

5. JOIN and IN subqueries fail or explode across ClickHouse shards

Symptom.

Code: 288. DB::Exception: Double-distributed IN/JOIN subqueries is denied
(distributed_product_mode = 'deny'). You may rewrite query to use local tables
in subqueries, or use GLOBAL keyword, or set distributed_product_mode to
suitable value. (DISTRIBUTED_IN_JOIN_SUBQUERY_DENIED)

Or no error, but a ClickHouse sharding troubleshooting classic: a join that takes minutes and transfers gigabytes between nodes.

Mechanism. A query like SELECT ... FROM events_dist JOIN users_dist USING (user_id) is rewritten and sent to each shard as events_local JOIN users_dist. The right-hand side is still a Distributed table, so each shard fans out to every shard for the join input: with N shards that is N² sub-queries, and the default distributed_product_mode = 'deny' refuses to run it. The three sound rewrites, in order of cost:

-- (1) Co-locate: shard both tables on the same key, then join local-to-local.
--     Correct only if events_local and users_local use the same sharding key.
SET distributed_product_mode = 'local';
SELECT e.tenant_id, count()
FROM events_dist AS e
JOIN users_dist AS u ON e.user_id = u.user_id
GROUP BY e.tenant_id;

-- (2) GLOBAL: build the right side once on the initiator, ship it to every shard
--     as a temporary table. Cost = size of the right side × number of shards.
SELECT e.tenant_id, count()
FROM events_dist AS e
GLOBAL JOIN users_dist AS u ON e.user_id = u.user_id
GROUP BY e.tenant_id;

-- (3) Dictionary: for slowly changing dimensions, replace the join entirely
SELECT dictGet('users_dict', 'plan', user_id) AS plan, count()
FROM events_dist
GROUP BY plan;

Option (1) is why a ClickHouse sharding key is a data-model decision and not an operations one: it only works when the join key is the sharding key on both sides. Option (2) is correct in every case but should be sized: system.query_log on the initiator shows ProfileEvents['NetworkSendBytes'], and if the right side is hundreds of megabytes, every query pays that transfer N times.

Option (3) is the fastest for dimension tables under a few gigabytes and removes the network entirely. The ClickHouse documentation on the Distributed table engine is precise on how the rewrite is performed if you need the exact semantics for a corner case.

6. A ClickHouse shard goes away and every query fails

Symptom.

Code: 279. DB::Exception: All connection tries failed. Log:
Code: 210. DB::NetException: Connection refused (ch-s2-r1:9000). (NETWORK_ERROR)
Code: 210. DB::NetException: Connection refused (ch-s2-r2:9000). (NETWORK_ERROR)
(ALL_CONNECTION_TRIES_FAILED)

Or a subtler one: a replica is up, accepts connections, and returns stale data because it is read-only.

Triage. In ClickHouse sharding troubleshooting, check replica state before network state, because a Keeper session loss produces read-only replicas that look perfectly healthy to a TCP probe:

SELECT
    hostName()      AS host,
    table,
    is_readonly,
    absolute_delay,
    queue_size,
    zookeeper_exception
FROM clusterAllReplicas('events_cluster', system.replicas)
WHERE table = 'events_local'
ORDER BY host;

-- Keeper ensemble view, since 26.1
SELECT * FROM system.zookeeper_info;

is_readonly = 1 with a populated zookeeper_exception is a Keeper problem, not a shard problem; the Keeper mntr output (zk_server_state, zk_outstanding_requests) is the next stop, and the ClickHouse Keeper guide documents every field. Fix the quorum first; the replicas rejoin on their own.

Fix and the trade-off it carries. In ClickHouse sharding, for genuinely unavailable shards you choose between failing fast and answering partially:

-- Profile for dashboards that prefer a partial answer to no answer
SET skip_unavailable_shards = 1;

-- Profile for billing / reconciliation queries: keep the default
SET skip_unavailable_shards = 0;

Silent partial results are the more dangerous ClickHouse sharding failure. Put skip_unavailable_shards = 1 only in a named profile assigned to users whose queries can tolerate it, never in the default profile. For the replica-selection side, load_balancing decides which replica in a shard the initiator tries first (random by default; nearest_hostname or in_order for topology-aware routing), and max_replica_delay_for_distributed_queries (default 300 s) fences off replicas whose absolute_delay exceeds the threshold, with fallback_to_stale_replicas_for_distributed_queries deciding whether a stale answer is better than an error. Set those three per workload profile and write down why.

ClickHouse sharding performance optimisation once the cluster is healthy

With the six failure modes closed, three settings deliver most of the remaining ClickHouse sharding headroom, and each has a measurable before-and-after in system.query_log.

Parallel replicas. Since the 26.x line, core parallel replicas is no longer listed as experimental or beta: a single query can read from every replica of a shard, splitting the scan by mark ranges. On a two-replica shard that is close to a 2× read-throughput ceiling for scan-bound queries.

SET enable_analyzer = 1;
SET enable_parallel_replicas = 1;
SET max_parallel_replicas = 2;
SET cluster_for_parallel_replicas = 'events_cluster';
SET parallel_replicas_min_number_of_rows_per_replica = 1000000;

Two exclusions decide whether it applies to your workload: parallel replicas are disabled for queries using FINAL, and they are incompatible with projections as of 26.7 per the parallel replicas deployment guide. A design picks projections or parallel replicas per query class, not both; verify the choice with EXPLAIN PIPELINE and confirm the extra replicas appear in system.query_log under the same initial_query_id.

Local replica preference in ClickHouse sharding. prefer_localhost_replica = 1 (the default) makes an initiator that is also a shard member read its own local table over the loopback rather than the network. Keep it on unless you deliberately run dedicated query-only initiators, in which case it is irrelevant. With parallel replicas, parallel_replicas_prefer_local_replica (26.5) plays the same role.

ClickHouse sharding connection fan-out. max_distributed_connections (default 1024) caps concurrent shard connections per query and rarely needs changing; connections_with_failover_max_tries and connect_timeout_with_failover_ms (default 1000 ms) decide how quickly the initiator gives up on a dead replica and moves to the next, and on a cross-availability-zone cluster the default timeout is often too short. Measure with ProfileEvents['DistributedConnectionFailTry'] in system.query_log before raising it.

Monitoring ClickHouse sharding health so the pager fires first

Every ClickHouse sharding troubleshooting query in this article reduces to a handful of scrape-able queries. Run them on a one-minute cadence from the monitoring host, alert on the thresholds, and the mean time to detect any of these failure modes drops from a user complaint to a page.

Where this guide stops

Everything above assumes ClickHouse sharding on self-managed open-source ClickHouse with ReplicatedMergeTree locals and a Distributed table. ClickHouse Cloud uses SharedMergeTree, which has no Distributed send queue and no internal_replication flag, so sections one and two do not apply there and the parallel-replicas behaviour differs. The settings names are those of 26.3 LTS; on 24.x fleets several carry their older distributed_directory_monitor_* names, and if you are planning the jump, the rolling LTS upgrade guide lists the renamed and removed settings to audit first.

ClickHouse sharding troubleshooting changes are still production changes: test every one on a staging cluster with a replayed workload before it reaches production, and keep a verified backup and a rehearsed restore path in front of any OPTIMIZE ... DEDUPLICATE or resharding operation.

If you would rather have the send queue, the sharding-key review, and the 03:00 Keeper page of your ClickHouse sharding estate handled by people who do this every week, ChistaDATA runs 24×7 ClickHouse support with a 15-minute Severity 1 response on 100 % open-source ClickHouse, and fully managed ClickHouse operations for teams that want the cluster to be someone else’s problem entirely.