clickhouse.guru
Runbooks

Runbook: replication, cluster operations and Kafka ingestion

On-call queries for replicated clusters: replica health, orphan tables, replication-queue oddities, ON CLUSTER drops, setting drift, BACKUP progress, live config reload, remote() and Kafka consumers.

This page collects what you reach for when a cluster misbehaves: a replica went read-only, a log line looks alarming, a distributed DDL timed out, one node behaves differently from the rest, or a Kafka table stopped moving. The queries are short because ClickHouse exposes its internals as tables, and most questions are one SELECT away.

What runs where. Every query here reads system tables that exist on a server: system.replicas, system.replication_queue, system.distributed_ddl_queue, system.backups, system.kafka_consumers. The ones against system.replicas, system.replication_queue, system.backups and system.kafka_consumers parse and run in clickhouse local 26.4 but return no rows there, so they were run for validity, not for output. Anything with clusterAllReplicas, ON CLUSTER or remote() needs a configured cluster or a second server and was only syntax-checked with clickhouse format. The Kafka command-line cheat sheet needs a Kafka installation. Sections that could run locally (the column diff, system.tables dependencies, server_settings) did, and show real output. The cluster is called my_cluster and the database analytics throughout.

Related reading: how replication works, inserting into Distributed tables, operating Kafka engine tables, Kafka concepts for ClickHouse users.

Contents

Which tables are replicated, and is any replica unhealthy?

In a cluster not every table is replicated, and replicas of one table can be spread unevenly across hosts. system.tables tells you what exists; system.replicas tells you how each replicated table is doing. Use this section to audit redundancy, to look for read-only, lagging or missing replicas, and to work out which tables form one shard.

Replicated tables on this node:

SELECT
    database,
    name,
    engine,
    formatReadableSize(total_bytes) AS size
FROM system.tables
WHERE engine LIKE 'Replicated%'
ORDER BY total_bytes DESC;

The alert query. Run this on each replica, or through clusterAllReplicas. An empty result means every replicated table on that server looks healthy; any row is a table that needs attention.

SELECT
    database, `table`, is_leader, is_readonly, is_session_expired,
    future_parts, parts_to_check, queue_size, inserts_in_queue, merges_in_queue,
    log_max_index, log_pointer, total_replicas, active_replicas, absolute_delay
FROM system.replicas
WHERE is_readonly
   OR is_session_expired
   OR future_parts > 20
   OR parts_to_check > 10
   OR queue_size > 20
   OR inserts_in_queue > 10
   OR log_max_index - log_pointer > 10
   OR total_replicas < 2
   OR active_replicas < total_replicas;
Column What a bad value means
is_readonly, is_session_expired The replica lost its Keeper session or metadata and refuses inserts; zookeeper_exception says why
queue_size, inserts_in_queue Fetches are not keeping up with the log
log_max_index - log_pointer The replica is not even copying the shared log into its queue
parts_to_check Parts are being re-verified because something looks corrupt
active_replicas < total_replicas Another replica is down (replica_is_active shows which), not necessarily this one
absolute_delay Replica lag in seconds behind the freshest one

Adjust the thresholds to your workload (a table with heavy merges can have a legitimately long queue) and total_replicas < 2 to your topology. Reading log_max_index, log_pointer, total_replicas and active_replicas costs Keeper round trips, so keep this check infrequent on servers with thousands of tables.

The full table, one row per replicated table on the node, worst first:

SELECT
    database,
    `table`,
    replica_name,
    is_leader,
    is_readonly,
    is_session_expired,
    queue_size,
    inserts_in_queue,
    merges_in_queue,
    absolute_delay,
    active_replicas,
    total_replicas,
    zookeeper_path
FROM system.replicas
ORDER BY absolute_delay DESC, queue_size DESC;

zookeeper_path identifies a replicated table across hosts: replicas of one shard share it, different shards have different paths (often because the path contains {shard}). Across the cluster, group the replicas by Keeper path to see which hosts hold each table:

SELECT
    database,
    `table`,
    zookeeper_path,
    groupArray(hostName()) AS hosts,
    count() AS replica_count
FROM clusterAllReplicas('my_cluster', system.replicas)
GROUP BY database, `table`, zookeeper_path
ORDER BY database, `table`;

A replica_count of 1 means no second copy. Tables that exist on every node with a plain MergeTree engine are held per node and are not synchronised at all.

How do I find tables nobody uses?

Tables outlive the projects that created them. Before dropping one, check two things: does anything read or write it, and does anything depend on it? Neither answer is certain, but together they remove most risk.

Tables not touched by any query recently. This compares system.tables with the tables array of the query log over a window:

SELECT
    t.database,
    t.name,
    t.engine,
    formatReadableSize(t.total_bytes) AS size
FROM system.tables AS t
WHERE t.database NOT IN ('system', 'INFORMATION_SCHEMA', 'information_schema')
  AND t.engine LIKE '%MergeTree'
  AND concat(t.database, '.', t.name) NOT IN (
        SELECT arrayJoin(tables)
        FROM system.query_log
        WHERE event_date >= today() - 30)
ORDER BY t.total_bytes DESC;

The window is limited by how long your query log is kept; if it holds a week, “not used in 30 days” means “not used in a week”. A table fed only by a materialized view is written by the server on every insert into the source rather than by a query of its own, so it may not appear in the log under its own name and can look unused while live. In a local test the view’s inner storage table (.inner_id.<uuid>) showed up in this list for exactly that reason. Check dependencies next.

What depends on a table. system.tables records, for each table, the materialized views that read from it:

SELECT
    d.database,
    d.name AS source_table,
    concat(dep_db, '.', dep_table) AS dependent
FROM system.tables AS d
ARRAY JOIN
    dependencies_database AS dep_db,
    dependencies_table AS dep_table
WHERE d.database NOT IN ('system', 'INFORMATION_SCHEMA', 'information_schema')
ORDER BY d.database, d.name;
analytics  events  analytics.events_per_day

A view whose source was dropped, or a table that is the TO target of a view but is never read, are the typical orphans. To see a view’s target and source, read create_table_query of the view in system.tables.

Before you drop:

  • Rename first (RENAME TABLE a TO a_unused) and wait a release cycle; a rename is cheap to undo, a drop is not.
  • Take a backup, or DETACH TABLE if in doubt: the data stays on disk.

For the usage side, the queries runbook ranks the most-queried tables.

What does “Found parts with the same min block and with the same max block” mean?

In the server log of a replicated table you may see:

Found parts with the same min block and with the same max block as the missing part <name>. Hoping that it will eventually appear as a result of a merge.

ClickHouse scheduled a check on a part it thought was missing (a replica is expected to have it but did not at that instant). Later another background task looked at the same part to decide how to repair it and found that a part with the same block range had appeared in the meantime: the result of a merge or of a fetch from another replica. The “missing” part was only missing for a moment.

Nothing to do in most cases; the message should disappear once the merge or fetch completes. Worry only if it repeats for the same part name for a long time. Then check whether something is stuck, and whether the replica is healthy at all:

SELECT type, new_part_name, source_replica, num_tries, last_exception, postpone_reason
FROM system.replication_queue
WHERE `table` = 'events_local'
ORDER BY create_time
LIMIT 20;

SELECT is_readonly, is_session_expired, queue_size, absolute_delay, lost_part_count
FROM system.replicas
WHERE `table` = 'events_local';

A growing lost_part_count means parts really were lost on every replica; that needs a different procedure. See how replication works.

Why does the replication queue say TTL merges already executing?

Entries in system.replication_queue whose postpone_reason reads like this:

Not executing log entry queue-0000000123 for part 715_0_286_108 because 2 merges with TTL already executing, maximum 2.

List them with:

SELECT database, `table`, replica_name, type, new_part_name, merge_type,
       num_postponed, postpone_reason, last_postpone_time
FROM system.replication_queue
WHERE is_currently_executing = 0 AND postpone_reason != ''
ORDER BY num_postponed DESC
LIMIT 10
FORMAT Vertical;

The entry is a merge triggered by a TTL (merge_type = 'TTL_DELETE' or a TTL move) that cannot start yet. Merges that apply a TTL are limited by the MergeTree setting max_number_of_merges_with_ttl_in_pool, 2 by default. While two TTL merges run, further ones wait in the queue and are retried later, each retry increasing num_postponed. A TTL merge for the same part that another replica is already performing is postponed for the same reason.

It is not an error, and last_exception is empty. It signals that TTL merges are slow or numerous: large parts, a busy merge pool, or a TTL rule that suddenly applies to a lot of data (for example after MODIFY TTL).

What to do:

  • Usually nothing; entries run when a slot frees up.
  • If the postponed count grows for days, see what the running merges are doing (SELECT * FROM system.merges, running merges) and the size of the parts involved.
  • Raising max_number_of_merges_with_ttl_in_pool runs more TTL merges in parallel at the cost of I/O. It is a MergeTree setting: set it on the table or in the server’s merge_tree section.
  • Partitioning by time and using ttl_only_drop_parts = 1 turns TTL into cheap whole-part drops instead of rewrites. See mutations, lightweight DELETE and TTL and TTL that deletes only matching rows.

Why did DROP TABLE ON CLUSTER time out, and what does SYNC do?

ON CLUSTER queues the statement in a shared DDL queue (in Keeper), and every host picks it up and runs it. The client waits for the hosts to report back. When a host is busy or the drop is heavy, the wait can exceed distributed_ddl_task_timeout (180 seconds by default, read from system.settings on 26.4) and you get a timeout even though the work continues in the background.

DROP TABLE analytics.events_old ON CLUSTER my_cluster SYNC;

What SYNC changes. A plain DROP TABLE on an Atomic database only detaches the table; its files are removed after a delay (server setting database_atomic_delay_before_drop_table_sec, 480 seconds by default according to the docs). SYNC removes the data immediately and makes the statement wait, so you know the space is free when the command returns. Without it, the table disappears from queries at once but not from disk.

Dealing with timeouts:

  • A timeout does not mean the statement failed. Check each host: system.distributed_ddl_queue has one row per host and task with status and exception_code, or run SELECT count() FROM system.tables WHERE name = 'events_old' on the hosts.
  • To see per-host results in the response, use distributed_ddl_output_mode. The value throw (the default) reports each host and fails on timeout; never_throw or null_status_on_timeout return the host table without raising.
  • Raise distributed_ddl_task_timeout for long operations, or set it to 0 to return immediately (the DDL then runs asynchronously and you poll the queue).
  • Do not retry the same DDL blindly. IF EXISTS / IF NOT EXISTS make a retry safe:
DROP TABLE IF EXISTS analytics.events_old ON CLUSTER my_cluster SYNC
SETTINGS distributed_ddl_task_timeout = 600;

How do I compare settings or table structure across nodes?

Configuration drift is a classic cause of “it only happens on one node”. Because every setting is exposed in a system table, you can compare nodes with SQL instead of diffing config files. Use it after an upgrade or config change to confirm every node got it, and when one replica behaves differently.

Server-wide MergeTree settings that differ between nodes. clusterAllReplicas runs the query on every replica. Group by setting name and keep those with more than one distinct value, or reported by fewer nodes than expected (which happens when one node knows a setting another does not, as after a partial upgrade):

WITH (
    SELECT count()
    FROM system.clusters
    WHERE cluster = 'my_cluster'
) AS cluster_size
SELECT
    name,
    groupUniqArray(value) AS distinct_values,
    count() AS reporting_nodes,
    cluster_size
FROM clusterAllReplicas('my_cluster', system.merge_tree_settings)
GROUP BY name
HAVING length(distinct_values) > 1 OR reporting_nodes < cluster_size;

Replace system.merge_tree_settings with system.settings, system.server_settings or system.build_options to compare other things; they all have a name/value shape. Filter out settings that legitimately differ per node, such as anything containing a hostname.

Columns that differ between two tables. The same trick works on one server: count how many of the two tables each column name appears in; anything not in both shows up.

WITH
    'events' AS table_a,
    'events_new' AS table_b
SELECT
    name,
    count() AS tables_with_column
FROM system.columns
WHERE database = 'analytics'
  AND `table` IN (table_a, table_b)
GROUP BY name
HAVING tables_with_column != 2
ORDER BY name;
referrer  1

(events_new was created AS events plus ADD COLUMN referrer String.) This is the quick way to compare a table with its renamed copy, or query_log with the query_log_0 that was rotated out after an upgrade added columns. It compares names only; group by name, type instead to compare types too — the same test then returned referrer String 1.

How far along is my BACKUP?

BACKUP ... ASYNC returns an id immediately and the work continues in the background. system.backups is where to watch it. Status of all backups:

SELECT
    id,
    name,
    status,
    start_time,
    end_time,
    formatReadableSize(total_size) AS size,
    num_files,
    error
FROM system.backups
ORDER BY start_time DESC;

status is one of CREATING_BACKUP, BACKUP_CREATED, BACKUP_FAILED, BACKUP_CANCELLED, and for restores RESTORING, RESTORED, RESTORE_FAILED, RESTORE_CANCELLED. A failed backup has its reason in error. The table is empty if no backup has run since the server started, as it was in the local test.

Rough progress and ETA for a running backup. ClickHouse reports no percentage, but total_size grows as files are written, so compare it with the size of the data being backed up. The result is an estimate: it assumes the backup is about as big as the compressed data on disk, which is roughly true for a plain backup and not for an incremental one.

WITH
    (SELECT sum(total_bytes)
     FROM system.tables
     WHERE database = 'analytics' AND has_own_data AND total_bytes IS NOT NULL) AS database_bytes
SELECT
    id,
    start_time,
    round(100 * total_size / database_bytes, 1) AS progress_pct,
    formatReadableSize(total_size) AS backed_up,
    formatReadableSize(database_bytes) AS database_size,
    toUInt32(now() - start_time) AS elapsed_seconds,
    now() + toIntervalSecond(toUInt32((now() - start_time) * (100 / progress_pct - 1))) AS eta
FROM system.backups
WHERE status = 'CREATING_BACKUP' AND progress_pct > 0;

Adjust the database = 'analytics' filter to the databases your BACKUP statement covers; for a single table filter on name too.

  • If progress_pct stalls while the status stays CREATING_BACKUP, look at system.processes for the backup query and at the destination (disk or S3) for throttling.
  • total_size counts what has been written to the backup; uncompressed_size and compressed_size give the other views. Compare like with like.
  • The table holds history only until the server restarts or entries age out of the in-memory list.

How do I change ClickHouse config without a restart?

Restarting a production node to change a limit is disruptive. Many settings can change live, and the server tells you which. system.server_settings has a column for it:

SELECT name, value, changeable_without_restart
FROM system.server_settings
WHERE name IN ('max_concurrent_queries', 'max_server_memory_usage', 'mark_cache_size', 'max_thread_pool_size')
ORDER BY name;
mark_cache_size          5368709120   Yes
max_concurrent_queries   0            Yes
max_server_memory_usage  23192823398  Yes
max_thread_pool_size     10000        No

changeable_without_restart is Yes, No or IncreaseOnly (can be raised live, lowering needs a restart). Anything No needs a restart.

Applying a change:

  1. Put it in its own file under /etc/clickhouse-server/config.d/ (server settings) or users.d/ (users and profiles), not in the main config.xml. Override files are merged into the main config and survive package upgrades.
  2. For config.d, ask the server to reread it: SYSTEM RELOAD CONFIG;
  3. Confirm in system.server_settings that the new value is there.

User, role and profile definitions in XML under users.d are watched and reloaded automatically when the file changes. Users created with SQL (CREATE USER) take effect immediately and never touch the files. Dictionaries have their own reload statements (SYSTEM RELOAD DICTIONARIES).

In containers and orchestrators, config files from a mounted volume or a Kubernetes ConfigMap change after a short propagation delay with no pod restart. Verify the file inside the pod, then run SYSTEM RELOAD CONFIG. Make sure your deployment tool (Helm, an operator) does not roll the pods when it applies the change: some treat any config change as a reason for a rolling restart.

SYSTEM RELOAD CONFIG needs a server; clickhouse local answers SYSTEM RELOAD CONFIG query is not supported in clickhouse-local. See the server settings reference.

What does the “Include not found” message in the log mean?

On startup the log contains lines like Include not found: clickhouse_remote_servers or Include not found: zookeeper-servers. It looks alarming and is almost always nothing.

Config files (config.xml, users.xml) support substitutions: a tag can carry an incl="name" attribute, and the server then replaces its content with the section called name from a separate substitutions file, by default /etc/metrika.xml, or wherever <include_from> points. The stock config.xml has incl attributes on several tags, for example:

<zookeeper incl="zookeeper-servers">
    <node>
        <host>keeper-1</host>
        <port>2181</port>
    </node>
</zookeeper>

If the substitutions file is missing or has no section with that name, the server logs Include not found: zookeeper-servers and carries on with whatever is written inline, here the node entries. Nothing is broken.

What to do:

  • Nothing, if you configure everything directly in config.xml or config.d/. The message is informational.
  • To silence it, delete the incl="..." attribute, or provide the section in the substitutions file:
<clickhouse>
    <include_from>/etc/clickhouse-server/substitutions.xml</include_from>
</clickhouse>
  • If you do rely on a substitution file and the message appears for a name that should exist, check the path, the file permissions, and that the section name matches the incl value exactly.

The substitution mechanism is older than the config.d/ override-file approach, which is simpler and what current docs recommend. See configuration files.

How do I query another ClickHouse server with remote()?

remote() reads a table from another ClickHouse server over the native protocol. No setup is needed beyond network access and a user with SELECT on the table. This needs a second server, so it was syntax-checked, not run.

-- user 'default', no password
SELECT count() FROM remote('host:9000', analytics, events);

-- with credentials
SELECT count() FROM remote('host:9000', analytics, events, 'reader', '<password>');

-- the sixth argument is a sharding key, used when the address pattern names several servers
SELECT count() FROM remote('host-{1..3}:9000', analytics, events, 'reader', '<password>', rand());

Arguments in order: address (or address pattern), database, table, user, password, sharding key. 'host-{1..3}' expands to three servers and 'a,b' lists them; the result is the union of the tables on all of them, a cheap way to query a shard set without defining a cluster. A table definition that hides the call:

CREATE TABLE analytics.events_remote AS remote('host:9000', analytics, events, 'reader', '<password>');

Notes:

  • Use remoteSecure() for TLS (port 9440 by default).
  • Passwords written in CREATE TABLE ... AS remote(...) end up in the stored metadata; a named collection or a profile on the server is the better home for secrets.
  • remote() and url() can be restricted to an allow-list of hosts; see restricting the hosts.

Which Kafka commands do I need around a Kafka engine table?

All scripts are in the bin/ directory of the Kafka distribution and use --bootstrap-server (the old --zookeeper flag no longer exists). Hostnames are placeholders; this needs a Kafka installation, so none of it was run.

# topics
bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --list
bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --create --topic sample --partitions 3 --replication-factor 3
bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --describe --topic sample
bin/kafka-topics.sh --bootstrap-server kafka-1:9092 --delete --topic sample

# send and read test messages
bin/kafka-console-producer.sh --bootstrap-server kafka-1:9092 --topic sample
bin/kafka-console-consumer.sh --bootstrap-server kafka-1:9092 --topic sample --from-beginning

# lag of the group behind a ClickHouse Kafka table
bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --list
bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --describe --group clickhouse-events-v1

# newest offset of every partition
bin/kafka-get-offsets.sh --bootstrap-server kafka-1:9092 --topic sample

# replay: stop the ClickHouse consumer first, preview, then execute
bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --group clickhouse-events-v1 --topic sample --reset-offsets --to-earliest --dry-run
bin/kafka-consumer-groups.sh --bootstrap-server kafka-1:9092 --group clickhouse-events-v1 --topic sample --reset-offsets --to-earliest --execute

Reset targets: --to-earliest, --to-latest, --to-datetime, --to-offset, --shift-by, --by-duration, --to-current. The group must have no active members. Typical retention settings (broker defaults; topics override them with retention.ms and retention.bytes):

log.cleanup.policy=delete
log.retention.hours=96
log.retention.bytes=1073741824
log.retention.check.interval.ms=30000

Retention is per partition and whichever limit hits first deletes the data. It is also your recovery window if ClickHouse is down. When a Kafka table lags or stalls, operating Kafka engine tables explains what to check, including the DETACH/ATTACH restart.

What does ClickHouse think its Kafka consumers are doing?

SELECT
    database,
    `table`,
    consumer_id,
    assignments.topic,
    assignments.partition_id,
    assignments.current_offset,
    num_messages_read,
    last_poll_time,
    last_commit_time,
    num_rebalance_assignments,
    exceptions.text
FROM system.kafka_consumers
ORDER BY `table`, consumer_id;

This is ClickHouse’s own view; the broker’s view of lag comes from kafka-consumer-groups.sh --describe (previous section). The query parses and returns no rows in clickhouse local, which has no Kafka tables. How to read it on a server:

  • No rows for a table: no consumer is running. The Kafka table has no attached materialized view, or it is detached.
  • last_poll_time recent but last_commit_time old: the consumer reads messages but cannot finish writing them. Look at exceptions.text and at the target table (read-only replica, too many parts).
  • num_rebalance_assignments climbing: the consumer group keeps rebalancing, usually because the consumer stalls longer than the poll interval.
  • Empty assignments.partition_id: this consumer owns no partitions, because there are more consumers than partitions.

Full guide: operating Kafka engine tables; the pipeline itself is in the Kafka engine and materialized view pipeline.

runbookreplication

Discussion

Comments live in GitHub Discussions. Sign in with GitHub to join in.