Inserting into Distributed tables

A Distributed table stores no data. It is a routing layer over a set of local tables on every shard. Reads fan out to the shards, and writes are forwarded to them. This part explains the write path, because it hides a disk queue that can surprise you. The statements here need a cluster; they were parsed with clickhouse format on 26.4 and the settings were checked against system.settings.

The setup

Take a cluster of three shards with two replicas each, six nodes in total. Each node has the local table, and each node also has the Distributed table pointing at it:

-- on every node (ON CLUSTER is the easy way)
CREATE TABLE analytics.events_local ON CLUSTER my_cluster
(
    event_time DateTime, user_id UInt64, event_type LowCardinality(String),
    url String, duration_ms UInt32, country LowCardinality(String)
)
ENGINE = ReplicatedMergeTree
PARTITION BY toYYYYMM(event_time)
ORDER BY (event_type, event_time);

CREATE TABLE analytics.events_all ON CLUSTER my_cluster AS analytics.events_local
ENGINE = Distributed(my_cluster, analytics, events_local, cityHash64(user_id));

The last argument is the sharding key: a row goes to the shard selected by sharding_key % total_shard_weight. Pick an expression that is stable and well distributed. Without a sharding key, inserting into a multi-shard Distributed table is an error.

What the initiator does

Your client connects to any node. That node, the initiator, receives the whole insert. For a default (asynchronous) insert it then does this:

1. Splits the data per destination shard according to the sharding key.
2. Writes each piece to its local disk, in the Distributed table’s directory.
3. Acknowledges the insert to your client.
4. In the background, sends each stored piece to a replica of the target shard.

So a successful INSERT into a Distributed table only means “the initiator has it on disk”, not “the shards have it”. A replica that is down is skipped and another one is chosen. With internal_replication = true on the cluster definition, which is the correct setting for Replicated* tables, the data is sent to one replica per shard and the replication mechanism copies it to the others. With the setting off the initiator writes to every replica itself, which is how you get replicas diverging.

If a whole shard is unavailable, the initiator keeps the data and retries. That has an operational consequence:

If a shard stays down for a long time, the pending data accumulates on the initiators’ disks. Watch the free space of the nodes that receive your inserts.

Watching the queue

system.distribution_queue shows the pending pieces per destination:

SELECT
    database,
    `table`,
    data_path,
    is_blocked,
    error_count,
    data_files,
    formatReadableSize(data_compressed_bytes) AS pending,
    last_exception
FROM system.distribution_queue
ORDER BY data_files DESC;

data_files growing without bound, or error_count rising with a last_exception, means a shard cannot be reached or rejects the data (for example, a column that exists on the Distributed table and not in the local one). SYSTEM FLUSH DISTRIBUTED analytics.events_all forces a send attempt, and SYSTEM STOP DISTRIBUTED SENDS / START DISTRIBUTED SENDS pause and resume the background work.

Synchronous inserts

If you need to know that the shards have the data when the insert returns, turn on foreground mode:

INSERT INTO analytics.events_all SETTINGS distributed_foreground_insert = 1
SELECT ... ;

The old name of this setting is insert_distributed_sync; it still works as an alias. In this mode the initiator does not keep the data locally. It forwards it to the shards and waits for them to confirm (at least one replica per shard with internal_replication). A shard that is down makes the insert fail or time out instead of queueing. The wait is bounded by distributed_background_insert_timeout (alias insert_distributed_timeout; 0 means no timeout).

The trade-off is plain: asynchronous mode never fails because of a shard outage but can lose data if the initiator’s disk is lost, while synchronous mode tells you immediately but couples the insert to the health of every shard.

Choosing the shard yourself

To bypass the sharding key and write to one specific shard synchronously, set insert_shard_id to the shard number (1-based). 0, the default, disables it.

INSERT INTO analytics.events_all SETTINGS insert_shard_id = 2
SELECT ... ;

This is the tool when each shard should receive different data from a pipeline you control. A value outside the range of shards raises an error.

The better pattern for heavy ingestion

Inserting through Distributed makes the initiator a bottleneck: it receives, splits and re-sends every row. For high volume, insert directly into the local tables from many clients, with a load balancer or client-side round robin choosing the shard, and use the Distributed table for reads. Each insert then lands as one part on one shard, and the replication machinery does the rest. You give up the automatic sharding by key, so use it when any shard may hold any row, or when the client can do the routing.

Next: moving data between tables with ATTACH PARTITION FROM.

Reference: the Distributed engine.