Skip to main content

ClickHouse Distributed parity

The sink writes directly to shard-local tables, never through a Distributed table — that is the whole point of the connector (bigger blocks, less merge pressure, a synchronous durable ack; see Why direct-to-shard, not a Distributed table). But you may still want to read through a Distributed table and have ClickHouse prune shards with optimize_skip_unused_shards, so a query filtered on the sharding key only visits the one shard that can hold the row.

That works only if the sink's record→shard placement is bit-for-bit identical to the Distributed table's sharding expression. This page covers how to line the two up, and the parity check that keeps them lined up.

[!WARNING] If placement and the DDL's sharding expression diverge, pruned queries silently return wrong results — not an error. ClickHouse trusts the sharding key: a query pruned to shard 2 never visits shard 1, so a row the sink placed on the wrong shard is simply invisible. This is documented ClickHouse behavior. The opt-in startup check below is the guard.

The topology contract​

Shards are matched to the cluster by position. The sink's shards: list must be in the same order as the <shard> entries in the cluster's remote_servers config:

  sink config shards[i]   ⟷   remote_servers <shard> number i+1  (shard_num i+1)

This 1:1 positional mapping is the one thing the framework cannot verify — nothing in the wire protocol ties your shards[0] to the cluster's shard_num = 1. The startup check verifies counts, weights, and the sharding expression; the ordering is yours to get right. As a best effort it also cross-checks your replica URL hostnames against system.clusters and logs a warning when one appears under a different shard_num — advisory only (HTTP URLs are not reliably mappable onto the cluster's native host:port entries), it never fails startup, and its silence proves nothing. Own the ordering operationally: list the shards in cluster order and keep the two configs under one review.

A worked example: shard by sensor​

1. The cluster (remote_servers)​

Two single-replica shards, written with internal_replication=true (the topology this sink assumes — see the README):

<clickhouse>
    <remote_servers>
        <sensors>
            <shard>
                <internal_replication>true</internal_replication>
                <weight>1</weight>
                <replica><host>ch-0</host><port>9000</port></replica>
            </shard>
            <shard>
                <internal_replication>true</internal_replication>
                <weight>1</weight>
                <replica><host>ch-1</host><port>9000</port></replica>
            </shard>
        </sensors>
    </remote_servers>
</clickhouse>

2. The tables​

A local table on every shard node (what the sink writes to), and a Distributed table on a query node (what you read through):

-- On each shard node: the local table the sink inserts into.
CREATE TABLE analytics.sensor_events
(
    sensor  String,
    ts_ms   Int64,
    value   Float64
)
ENGINE = MergeTree
ORDER BY (sensor, ts_ms)
SETTINGS non_replicated_deduplication_window = 100;

-- On a query node: reads fan out (and prune) across the cluster.
CREATE TABLE analytics.sensor_events_dist AS analytics.sensor_events
ENGINE = Distributed(sensors, analytics, sensor_events, xxHash64(sensor));

The Distributed engine's fourth argument, xxHash64(sensor), is the sharding expression the sink must reproduce exactly.

3. The sink​

table: names the local table — the sink never inserts through the Distributed one. Weights must equal the cluster's <weight> values, and distributed_check names the Distributed table for the startup guard:

sink:
  clickhouse:
    table: analytics.sensor_events      # the LOCAL table, never the Distributed one
    columns: [sensor, ts_ms, value]
    shards:
      - replicas: ["http://ch-0:8123"]
        weight: 1
      - replicas: ["http://ch-1:8123"]
        weight: 1
    distributed_check:
      cluster: sensors
      table: analytics.sensor_events_dist
      sharding_key: sensor
      # endpoint: "http://query-node:8123"  # if the Distributed table lives off-shard
      # sharding_expr: "xxHash64(sensor)"    # escape hatch for non-identifier keys

distributed_check takes exactly one of sharding_key (the shipped, verified form — the check builds xxHash64(<key>) for you) or sharding_expr (a raw-expression escape hatch, compared textually and therefore brittle). endpoint defaults to the first replica of shard 0; override it when the Distributed table lives on a front node outside the shards list.

4. The router​

Supply a key extractor — a plain fn item, not a closure (the extractor is higher-ranked over the payload lifetime) — returning a ShardKey. The router owns the xxHash64(seed 0) and the weight-interval selection:

use etl::clickhouse::ShardKey;

/// Sharding key: the `sensor` column. One sensor always lands on one shard,
/// matching the Distributed DDL's `xxHash64(sensor)`.
fn sensor_key<'a>(row: &'a SensorEvent<'_>) -> ShardKey<'a> {
    ShardKey::Str(row.sensor)
}
let sink = etl::clickhouse::config::from_component_config(&pipeline.config().sink)?;
// No-op unless the YAML opts into `distributed_check`; with it, startup fails
// fast if the sink topology drifts from the cluster + DDL.
pipeline.block_on(sink.validate_distributed())?;
// Weights come from the validated YAML — router and endpoints can't drift.
let router = sink.router::<EventFam>(sensor_key);

router then drops into the chain's terminal stage in place of KeyHashRouter (.sink(encoder.clone(), router.clone(), ...)). It is shard-count-agnostic: a single-shard config behaves identically — every hash lands on shard 0 — so the same code runs in dev and scales out unchanged.

Parity rules​

For placement to match ClickHouse bit-for-bit:

  • Single-column key. The sharding expression must be xxHash64(<one column>). Multi-argument ClickHouse hash calls are not externally reproducible.
  • xxHash64 only. The router implements canonical XXH64 with seed 0. cityHash64 is a frozen fork of Google's CityHash v1.0.2 that the router does not reproduce; do not use it (or any other hash) as the sharding key.
  • Integer keys hash at their declared column width, little-endian. ShardKey::U64 hashes 8 LE bytes; ShardKey::U32 hashes 4 LE bytes. They differ — xxHash64(toUInt64(42)) ≠ xxHash64(toUInt32(42)) — so the ShardKey variant must match the column's declared type (Int64: pass the value as ShardKey::U64; DateTime: ShardKey::U32).
  • Non-Nullable key column. A Nullable key changes the value ClickHouse hashes; parity requires the sharding-key column to be non-Nullable.

Pruning: optimize_skip_unused_shards​

Once placement matches, prune reads by enabling the setting per query:

SELECT count()
FROM analytics.sensor_events_dist
WHERE sensor = 'sensor-42'
SETTINGS optimize_skip_unused_shards = 1;

Preconditions — pruning fires only when all hold:

  • The predicate is a literal = or IN (combined with AND/OR) over the deterministic sharding key. Ranges (>, BETWEEN) and query parameters ({p:String}) do not prune — the optimizer can't evaluate the sharding expression at planning time.
  • The sharding key is deterministic (it is — xxHash64 of a column).

To assert the query is actually prunable in a test or CI, add force_optimize_skip_unused_shards = 2, which makes ClickHouse throw if the optimizer could not engage.

[!NOTE] force_optimize_skip_unused_shards = 2 proves the optimizer engaged, not that fewer shards were contacted. To prove a specific shard was skipped end to end, correlate each node's system.query_log by initial_query_id (with SYSTEM FLUSH LOGS on every node) and confirm the non-owning shard ran no subquery.

Weights: the add-a-shard knob​

Selection is xxHash64(key) % sum_of_weights, with the remainder mapped through consecutive half-open weight intervals in shard order. All-weights-1 degenerates to hash % num_shards. Weights are the documented rebalancing lever: when you add capacity, give the new (or larger) shard a higher weight so it takes proportionally more keys, and keep the sink's weights equal to the cluster's <weight> values — the parity check enforces this on startup, and drift would misplace rows.

[!IMPORTANT] Changing weights, changing the sharding key, or adding/removing a shard re-homes keys to different shards — and rows already written do not move. Nothing merges across shards: the old shard keeps its copies permanently (ReplacingMergeTree collapses within one shard's local table only), so unpruned queries through the Distributed table return both homes' rows, while pruned queries return only the new home's. Plan such a change as a migration — backfill or rebuild the moved keys' rows, or delete the old shard's copies. A plain restart with unchanged topology is the benign case: replayed rows land on the same shard, where dedup tokens and the collapse pattern apply as usual. See Delivery guarantees.