Analytics PipelineClickHouse & Materialized Views
Guide ClickHouse Core Architecture & Analytics Fundamentals

Load Balancing Reads Across Replicas

A Distributed table does not spread reads evenly by default — it walks each shard’s replica list in a fixed order and hammers the first reachable one until it dies, which is why a three-replica cluster can show one node at 90% CPU while the other two idle. The lever is the load_balancing setting, backed by max_replica_delay_for_distributed_queries and fallback_to_stale_replicas_for_distributed_queries so a busy or lagging replica is skipped instead of served. This page explains each policy, how to pin them per query or per user, and how to prove the traffic actually moved.

Prerequisites

How a distributed read picks a replica

When you query a Distributed table, the initiator node fans the query out to one replica per shard. Which replica it picks per shard is decided entirely by load_balancing. The delay-aware settings act as a filter applied before the policy chooses: any replica whose absolute_delay exceeds max_replica_delay_for_distributed_queries is removed from the candidate list, and only then does the policy rank whoever is left.

Replica selection for a distributed read: delay filter then load_balancing policy The initiator fans a Distributed query to one replica per shard. A delay filter removes replicas past max_replica_delay_for_distributed_queries, then the load_balancing policy (random, nearest_hostname, in_order, round_robin) ranks the survivors and one replica per shard executes its slice. prefer_localhost_replica short-circuits to a local replica. Initiator Distributed query Delay filter drop replica if delay > max_replica_delay candidate set remains load_balancing random nearest_hostname in_order round_robin shard 1 · replica B chosen · executes slice shard 2 · replica A chosen · executes slice shard 3 · localhost prefer_localhost_replica prefer_localhost_replica short-circuits selection to the local replica when the initiator is itself a replica of the shard.

Step 1 — Read the current policy and cluster topology

Confirm what each session is using and enumerate the replicas the policy has to choose from.

sql
SELECT name, value FROM system.settings
WHERE name IN (
    'load_balancing',
    'max_replica_delay_for_distributed_queries',
    'fallback_to_stale_replicas_for_distributed_queries',
    'prefer_localhost_replica'
);

SELECT cluster, shard_num, replica_num, host_name, is_local
FROM system.clusters
WHERE cluster = 'analytics_cluster'
ORDER BY shard_num, replica_num;

The default load_balancing is random, max_replica_delay_for_distributed_queries is 300 seconds, and both fallback_to_stale_replicas_for_distributed_queries and prefer_localhost_replica default to 1.

Step 2 — Choose a load_balancing policy

Set the policy per session to test it, then promote the winner into the user profile. The four policies trade evenness against cache locality:

sql
-- Even spread, no locality: best for a homogeneous cluster behind one LB.
SET load_balancing = 'random';

-- Route to the topologically closest replica (fewest hostname edits) to cut cross-AZ latency.
SET load_balancing = 'nearest_hostname';

-- Deterministic: always prefer replica 1, fall through only on failure. Predictable cache warmth.
SET load_balancing = 'in_order';

-- Rotate the starting replica each query so long-running sessions spread load over time.
SET load_balancing = 'round_robin';

Use nearest_hostname when replicas span availability zones and you want to keep traffic local; use round_robin to spread a small number of heavy analytical queries; use in_order only when one replica has warmer page cache and you accept uneven load. random remains the safe default for many small queries.

Step 3 — Fence off lagging replicas

A replica that is behind on replication will happily answer a SELECT with stale data unless you cap the tolerated lag. max_replica_delay_for_distributed_queries removes any replica whose absolute_delay exceeds the threshold before the policy runs.

sql
-- Skip any replica more than 30s behind for this workload.
SET max_replica_delay_for_distributed_queries = 30;

-- If EVERY replica of a shard is stale, decide fail vs. serve-stale:
SET fallback_to_stale_replicas_for_distributed_queries = 1;  -- 1 = serve stale rather than error

With fallback_to_stale_replicas_for_distributed_queries = 0, a shard whose replicas are all past the delay threshold returns an error instead of stale rows — the right choice for correctness-critical reporting. Leave it at 1 for availability-first dashboards where a slightly stale answer beats a failed query. The replica states these settings read are exactly the ones you repair in executing replica failover DDL.

Step 4 — Control localhost preference

When the initiator node is itself a replica of a shard, prefer_localhost_replica = 1 short-circuits selection and reads locally, skipping a network hop. That is a latency win but it concentrates load on whichever node clients connect to.

sql
-- Force fan-out even to remote replicas so a single entry node does not absorb all shard-local reads.
SET prefer_localhost_replica = 0;

Turn it off when clients all connect through one address and you want that node’s local shard to share load with its peers; keep it on when clients connect round-robin across all nodes.

Step 5 — Pin the policy in the user profile

Session settings vanish on reconnect. Bake the production policy into the user profile so every connection inherits it.

xml
<!-- users.xml / profile -->
<profiles>
    <analytics_reader>
        <load_balancing>nearest_hostname</load_balancing>
        <max_replica_delay_for_distributed_queries>30</max_replica_delay_for_distributed_queries>
        <fallback_to_stale_replicas_for_distributed_queries>0</fallback_to_stale_replicas_for_distributed_queries>
        <prefer_localhost_replica>0</prefer_localhost_replica>
    </analytics_reader>
</profiles>

Verification

Prove the reads actually spread by attributing each shard slice to the replica that ran it. system.query_log records the initiating query; the per-replica child queries carry the real host. Flush logs first, then group by host:

sql
SYSTEM FLUSH LOGS;

SELECT
    hostName()               AS executed_on,
    count()                  AS query_slices,
    round(avg(query_duration_ms)) AS avg_ms
FROM clusterAllReplicas('analytics_cluster', system.query_log)
WHERE type = 'QueryFinish'
  AND is_initial_query = 0                 -- the fanned-out child queries
  AND event_time > now() - INTERVAL 15 MINUTE
GROUP BY executed_on
ORDER BY query_slices DESC;

With random or round_robin the query_slices count should be roughly even across replicas of the same shard. A single host dominating means either in_order is in effect, prefer_localhost_replica is pinning traffic, or the other replicas are being filtered out by the delay threshold — check absolute_delay in system.replicas to tell those apart.

Gotchas and edge cases

  • max_replica_delay can silently collapse your read pool. Set it too tight during a merge storm and every replica trips the threshold, so with fallback_to_stale_replicas_for_distributed_queries = 0 the whole shard starts erroring. Tune the delay against the real steady-state absolute_delay from monitoring replication lag with system.replicas, not an arbitrary small number.
  • in_order plus a flaky replica 1 causes a thundering fallback. Because in_order always tries the same replica first, a replica that flaps up and down makes every query pay a failed-connection timeout before falling through. Use round_robin or random for any replica set that is not rock-solid.
  • load_balancing governs reads, not the write path. Insert routing into a Distributed table is controlled by insert_distributed_sync and shard weights, not these settings. Do not expect round_robin to spread inserts — it only affects which replica serves each shard slice of a SELECT.
  • Localhost preference hides uneven client connections. With prefer_localhost_replica = 1 a load test driven from one node looks perfectly balanced in local latency while remote replicas sit idle. Always verify with is_initial_query = 0 grouping across clusterAllReplicas, not from a single node’s view.

Up: Fallback Routing & High Availability