ClickHouse: репликация через Keeper — журнал get, дедупликация блоков, insert_quorum и Distributed

На одной реплике ClickHouse строки есть, на другой их нет, и по счётчику не понять, догонит она или застряла. Репликация здесь устроена не как поток WAL: каждая вставка создаёт неизменяемый part (его устройство описано в статье про первичный ключ MergeTree, а судьбу part’ов после записи — в статье про слияния и мутации), реплика-автор записывает в Keeper строку журнала get, и остальные реплики сами скачивают этот part. Рядом в Keeper лежат хеши принятых блоков, они защищают от дублей при повторе вставки, а для insert_quorum число живых реплик берётся оттуда же. Поверх репликации стоит Distributed, который раскладывает строки по шардам через очередь файлов на диске. Ниже всё это на двух репликах и одном Keeper; объёмы — тысячи строк, цель опыта — видеть записи в журнале, а не измерять скорость.

DBA / 01koddo

Вставка на r1: part, журнал в Keeper, fetch на r2

Стенд

Три контейнера в сети dba-clickhouse-replication-keeper-net: Keeper, реплики r1 и r2. Keeper один: для опыта хватает, но своей потери он не переживёт. Три узла Keeper на Raft устроены как в статье про консенсус. Конфиги лежат в каталоге w: у Keeper свой файл, у реплик remote_servers, macros и zookeeper.

N=dba-clickhouse-replication-keeper
cat > w/node1.xml <<X
<clickhouse>
  <listen_host>0.0.0.0</listen_host>
  <remote_servers><demo><shard>
    <internal_replication>true</internal_replication>
    <replica><host>$N-r1</host><port>9000</port></replica>
    <replica><host>$N-r2</host><port>9000</port></replica>
  </shard></demo></remote_servers>
  <macros><shard>01</shard><replica>r1</replica></macros>
  <zookeeper><node><host>$N-k1</host><port>9181</port></node></zookeeper>
</clickhouse>
X
sed 's/>r1</>r2</' w/node1.xml > w/node2.xml
docker network create $N-net
docker run -d --name $N-k1 --network $N-net --memory 500m \
  -v $PWD/w/keeper.xml:/etc/clickhouse-keeper/keeper_config.xml:ro clickhouse/clickhouse-keeper:25.3
for n in 1 2; do docker run -d --name $N-r$n --network $N-net --memory 1500m \
  -v $PWD/w/node$n.xml:/etc/clickhouse-server/config.d/node.xml:ro clickhouse/clickhouse-server:25.3; done
c1() { docker exec $N-r1 clickhouse-client "$@"; }
c2() { docker exec $N-r2 clickhouse-client "$@"; }

keeper.xml — минимальный keeper_server на порту 9181 с одним узлом в raft_configuration (server_id 1). Версия сервера по SELECT version() — 25.3.14.14.

Опыт 1: part, журнал и fetch

Таблица создаётся на обеих репликах сразу. Путь в Keeper и имя реплики берутся из макросов:

c1 -q "CREATE TABLE ev ON CLUSTER demo (id UInt64, v String)
       ENGINE=ReplicatedMergeTree('/clickhouse/tables/{shard}/ev','{replica}') ORDER BY id"
c1 -q "INSERT INTO ev SELECT number, 'a' FROM numbers(1000)"

Вывод CREATE: два хоста, статус 0 у обоих.

dba-clickhouse-replication-keeper-r2	9000	0		1	0
dba-clickhouse-replication-keeper-r1	9000	0		0	0

Вставка шла на r1. Журнал событий part’ов показывает, что на r1 part родился, а на r2 был скачан (hostName() — это id контейнера, 0ebeea16352d — r1, d6fb08ce65b1 — r2):

for c in c1 c2; do $c -q "SYSTEM FLUSH LOGS"; $c -q "SELECT hostName(), event_type, part_name, rows
  FROM system.part_log WHERE table='ev' ORDER BY event_time_microseconds"; done

Вывод:

0ebeea16352d	NewPart	all_0_0_0	1000
d6fb08ce65b1	DownloadPart	all_0_0_0	1000

Тот же part на обеих репликах, и на второй он не вставлен, а скачан. Что для этого записано в Keeper:

c1 -q "SELECT name, value FROM system.zookeeper WHERE path='/clickhouse/tables/01/ev/log' FORMAT Vertical"

Вывод:

Row 1:
──────
name:  log-0000000000
value: format version: 4
create_time: 2026-10-28 00:20:02
source replica: r1
block_id: all_718426935619744589_17257296687255910572
get
all_0_0_0
part_type: Compact

Запись get с source replica: r1 и есть команда «скачай part all_0_0_0». Каждая реплика забирает такие записи из журнала в свою очередь (в коде — pullLogsToQueue) и выполняет их по порядку. Позже в том же журнале появляются и записи merge: слияние тоже решается один раз и выполняется на каждой реплике (подробнее — в статье про слияния).

Опыт 2: тот же блок второй раз

Повтор вставки — обычный случай: клиент не получил ответ и отправил запрос снова. В журнале выше есть block_id: хеш данных вместе с идентификатором партиции. Код присваивает его блоку перед фиксацией (ReplicatedMergeTreeSink.cpp, строки 399–401, тег v25.3.14.14-lts):

                /// We add the hash from the data and partition identifier to deduplication ID.
                /// That is, do not insert the same data to the same partition twice.
                block_id = temp_part.part->getNewPartBlockID(block_dedup_token);

Part фиксируется в Keeper под blocks/<block_id>, и если такой узел уже есть, part выбрасывается. Тот же INSERT ещё раз, затем другие данные, затем те же другие данные через вторую реплику:

c1 -q "INSERT INTO ev SELECT number, 'a' FROM numbers(1000)"
c1 -q "SELECT count() parts, sum(rows) FROM system.parts WHERE table='ev' AND active"
c1 -q "INSERT INTO ev SELECT number, 'b' FROM numbers(1000)"
c1 -q "SELECT name, value FROM system.zookeeper WHERE path='/clickhouse/tables/01/ev/blocks'"
c2 -q "INSERT INTO ev SELECT number, 'b' FROM numbers(1000)"; c2 -q "SELECT count() FROM ev"
c1 -q "SELECT event_type, rows, error, errorCodeToName(error) FROM system.part_log WHERE table='ev' AND error != 0"

Вывод по порядку:

1	1000
all_718426935619744589_17257296687255910572	all_0_0_0
all_3016996774311053399_9420078176866479432	all_1_1_0
2000
NewPart	1000	389	INSERT_WAS_DEDUPLICATED

После повтора part по-прежнему один с 1000 строками. Строки 'b' дали второй узел в blocks. Вставка тех же 'b' на r2 не добавила строк (счётчик остался 2000): блок проверяется в общем Keeper, а не в памяти реплики. В system.part_log дубль виден событием с ошибкой 389. Окно хранения хешей — replicated_deduplication_window = 1000 блоков и replicated_deduplication_window_seconds = 604800 (из system.merge_tree_settings): после выхода блока из окна повтор создаст дубль.

Опыт 3: insert_quorum при остановленной реплике

По умолчанию INSERT подтверждается после записи на одну реплику, остальные догоняют асинхронно. С insert_quorum=2 клиент получает OK, только когда part принят двумя репликами. Проверка на остановленной r2:

docker stop $N-r2
c1 --insert_quorum=2 --insert_quorum_timeout=5000 -q "INSERT INTO ev SELECT number+10000, 'q' FROM numbers(10)"
c1 -q "INSERT INTO ev SELECT number+10000, 'q' FROM numbers(10)" && echo ok

Вывод:

Received exception from server (version 25.3.14):
Code: 285. DB::Exception: Received from localhost:9000. DB::Exception: Number of alive replicas (1) is less than requested quorum (2/2).. (TOO_FEW_LIVE_REPLICAS)
(query: INSERT INTO ev SELECT number+10000, 'q' FROM numbers(10))
ok

Проверка идёт до записи, по числу живых реплик в Keeper (ReplicatedMergeTreeSink.cpp, строки 257–262):

                throw Exception(
                    ErrorCodes::TOO_FEW_LIVE_REPLICAS,
                    "Number of alive replicas ({}) is less than requested quorum ({}/{}).",
                    active_replicas,
                    quorum_size,
                    replicas_number);

Вставка без кворума прошла (ok): на r1 стало 2010 строк, r2 их получит при запуске. Кворум на двух репликах означает, что любой отказ одной из них останавливает запись; с двумя репликами insert_quorum=2 — выбор «не потерять подтверждённое» ценой доступности.

Опыт 4: отстающая реплика и очередь

r2 запускается снова, и у неё отключается только скачивание part’ов (контейнер работает), чтобы очередь была видна, а не исчезла вместе с процессом. Затем пять вставок на r1:

docker start $N-r2; sleep 12
c2 -q "SYSTEM STOP FETCHES ev"
for i in 1 2 3 4 5; do c1 -q "INSERT INTO ev SELECT number+30000+$i*100, 'g' FROM numbers(100)"; done
c2 -q "SELECT queue_size, inserts_in_queue, merges_in_queue, log_pointer, log_max_index, absolute_delay
       FROM system.replicas WHERE table='ev'"
c2 -q "SELECT type, new_part_name, source_replica, num_tries FROM system.replication_queue WHERE table='ev'"
c1 -q "SELECT count() FROM ev"; c2 -q "SELECT count() FROM ev"

Вывод:

6	5	1	15	14	3
GET_PART	all_8_8_0	r1	0
GET_PART	all_9_9_0	r1	0
GET_PART	all_10_10_0	r1	0
GET_PART	all_11_11_0	r1	0
GET_PART	all_12_12_0	r1	0
MERGE_PARTS	all_0_10_2	r1	5
3010
2510

log_max_index (14) — последняя запись журнала в Keeper, log_pointer (15) — номер следующей записи, которую r2 возьмёт из журнала: всё до 14 включительно она уже перенесла в свою очередь, но не выполнила. В очереди шесть записей, пять GET_PART и слияние all_0_10_2, которое r1 успела решить. У слияния num_tries равно 5: пока скачивание остановлено, r2 его не выполнила. absolute_delay = 3 секунды отставания; 3010 строк на r1 против 2510 на r2. Скачивание включается обратно:

c2 -q "SYSTEM START FETCHES ev"
c2 -q "SELECT queue_size, absolute_delay FROM system.replicas WHERE table='ev'"; c2 -q "SELECT count() FROM ev"

Вывод: 0 0 и 3010. Очередь опустела сама: строки не пересылались заново, r2 выполнила записи журнала. Реальный сетевой разрыв даёт ту же картину; SYSTEM STOP FETCHES её только воспроизводит.

Опыт 5: Distributed и очередь на диске

Distributed — не хранилище, а маршрутизатор. Вторая конфигурация разводит те же узлы по двум шардам с одной репликой в каждом; ключ шардирования id делит строки по чётности (равные веса шардов). Реплики должны принимать подключения друг от друга под пользователем default: в образе он по умолчанию ограничен localhost, поэтому добавлен users.d/net.xml с <networks><ip>::/0</ip></networks>.

c1 -q "SELECT cluster, shard_num, replica_num, host_name FROM system.clusters WHERE cluster='two_shards'"
for c in c1 c2; do $c -q "CREATE TABLE loc (id UInt64, v String) ENGINE=MergeTree ORDER BY id"; done
c1 -q "CREATE TABLE dist AS loc ENGINE=Distributed(two_shards, default, loc, id)"
c1 -q "INSERT INTO dist SELECT number, 'd' FROM numbers(1000)"
for c in c1 c2; do $c -q "SELECT count(), min(id), max(id) FROM loc"; done

Вывод: два шарда по одной реплике, затем строки по шардам.

two_shards	1	1	dba-clickhouse-replication-keeper-r1
two_shards	2	1	dba-clickhouse-replication-keeper-r2
500	0	998
500	1	999

Чётные id ушли на r1, нечётные на r2 (запрос count() через dist вернул 1000). Теперь r2 остановлен, и вставка идёт через dist:

docker stop $N-r2
c1 -q "INSERT INTO dist SELECT number+2000, 'd' FROM numbers(1000)" && echo insert-ok
sleep 6
c1 -q "SELECT table, is_blocked, error_count, data_files, data_compressed_bytes FROM system.distribution_queue"

Вывод insert-ok и строка очереди: dist 0 1 1 2387 (table, is_blocked, error_count, data_files, data_compressed_bytes). Вставка завершилась успехом: нечётная половина лежит файлом .bin в каталоге шарда на диске r1. Код кладёт файл в каталог и ставит в очередь отправки (DistributedSink.cpp, строки 905–912; сам Distributed создаёт этот sink в StorageDistributed::write):

        // Create hardlink here to reuse increment number
        auto bin_file = (fs::path(path) / file_name).string();
        auto & directory_queue = storage.getDirectoryQueue(disk, *it);
        {
            createHardLink(first_file_tmp_path, bin_file);
            auto dir_sync_guard = make_directory_sync_guard(*it);
        }
        directory_queue.addFileAndSchedule(bin_file, file_size, sleep_ms);

Чтение через dist при остановленной r2 упало с All connection tries failed, а после docker start через 15 секунд очередь опустела (data_files 0, error_count 0), и на r2 стало 1500 строк. Файл очереди не защищён от потери диска r1: пока он не отправлен, подтверждённая вставка живёт в одном месте. Если шард состоит из реплик ReplicatedMergeTree, по документации internal_replication=true заставляет Distributed писать только на одну из них, а остальные получают part по журналу из опыта 1; в этом опыте шарды без реплик, и настройка ничего не меняет.

Что из этого следует

• Лаг реплики — это queue_size и absolute_delay, а не разница счётчиков. В опыте 4 разница в 500 строк стояла на пяти GET_PART, а не на потере данных.
• Повтор вставки безопасен только внутри окна и только с тем же содержимым. 1000 блоков и 604800 секунд по умолчанию; дубль после окна проходит.
• insert_quorum=2 при двух репликах делает недоступной запись при отказе любой. Три реплики с кворумом 2 терпят одну потерю.
• Один Keeper — единая точка отказа всей репликации таблицы. Продакшен-кластер ставит три узла.
• Асинхронный Distributed подтверждает вставку раньше, чем строки доехали. Очередь на диске инициатора — место, где данные могут задержаться.

Чек-лист

1. Следи за system.replicas: queue_size, absolute_delay, active_replicas против total_replicas, is_readonly.
2. При росте очереди смотри system.replication_queue: num_tries и last_exception называют причину.
3. Повторяй вставки клиентом только с тем же содержимым: хеш считается по блоку, и другой блок не распознается как дубль.
4. Для Distributed смотри system.distribution_queue: data_files и error_count выше нуля значат недоставленные файлы.
5. Убери стенд вместе с данными: docker rm -f -v $N-k1 $N-r1 $N-r2 && docker network rm $N-net.