清理 ClickHouse 重复行的 REPLACE PARTITION runbook ::

这条 Kafka 到 ClickHouse 的链路上，已经有两批重复行写进了明细表：Kafka Connect 超时重投带来的 234 行（超时重投那篇），和 producer 重试带来的一百多行（四个入口那篇）。去重方案管的是以后的重复，已经落库的这些要单独清，而且重复还会再来，要的是一套能反复跑、跑坏了能回滚的 runbook。

ClickHouse 没有 UPSERT，改数据以 part 为单位（写入路径那篇）。这里的做法是把出问题的那一天从快照重建一遍，验过之后用 REPLACE PARTITION 整个换进去。

这篇最早的版本直接从正在写入的原表建临时表，换分区前数两次行数。后来评审一份同类的清理方案，在 lab 上跑出了几种行数对得上、数据却写坏或丢掉的情形（行数校验拦不住的情形），runbook 按评审的结论改成了下面的 R0 到 R8。整套在 clickhouse-lab 的三副本集群上跑过（25.3.14.14，Aiven 上的现有集群是 25.3.14.1），定稿那一版是实验 19；还没在现有集群上执行。

先低成本地数出有多少重复#

清理之前要先知道有多少、集中在哪几个小时。直接对整个分区 GROUP BY 也能数，但两亿行那套部署一个日分区就有 30 GiB，过了 45 天的分区还已经搬到了对象存储上，在现有集群上这么扫一次代价太大。重复只需要一个数字就能看出来：

SELECT count() AS rows,
       uniqExact(id, settle_ms, rev) AS keys
FROM events
WHERE _partition_id = '20260730'
  AND settle_ms >= {hour_start_ms}
  AND settle_ms <  {hour_start_ms} + 3600000

count() - uniqExact(...) 就是这一小时的多余行数。省在 WHERE 的两层裁剪：_partition_id 把日分区写死，settle_ms 是排序键的第一列，按小时切能让主键索引生效。24 条这样的查询覆盖一整天，哪一小时有问题也直接看出来。

内存跟着这一小时的行数走。两亿行那套部署一小时 500 万到 1600 万行；lab 在 25.8 上测得精确去重每行约 60–65 字节（实验 31–33），按这个推算，1600 万行的那一小时要 1 GB 左右，正好贴着下面那条 1 GB 的限制。三亿多行那套的高峰小时只会更多。被限制挡下来（报 Code: 241）的小时，再切成半小时重跑。

在现有集群上跑的时候把限制写在 URL 参数里，服务端强制，比在客户端约束可靠：

curl "https://host:port/?readonly=2&max_threads=1&max_memory_usage=1000000000&priority=10" \
     --data-binary @query.sql

为什么用快照加换分区#

清一个历史分区里的重复，还有几条路，各有缺口：

轻量删除，只给命中的行写删除标记。25.3 上必须带 IN PARTITION，否则全表每个 part 都要过一遍；同一秒写进来的两份按 create_time 分不开，只能按 _part_offset 删；执行前看不到结果，也没有回滚（轻量删除那篇）。
OPTIMIZE TABLE … PARTITION … FINAL DEDUPLICATE BY …，把整个分区合成一个 part、合并时去掉重复。它同样重写整个分区，同一组留下哪一行由合并决定，执行前不能验结果，也没有回滚（OPTIMIZE）。
换成 ReplacingMergeTree，管的是以后，清不了这张表现在的重复，除非迁移（在线换引擎那篇）。

选快照加换分区，是因为执行前能把结果完整验一遍，执行后能原样回滚。代价是要处理快照之后的写入和多副本之间的先后，R0、R4、R6 三道闸就是为这两件事设的。

runbook：R0 到 R8#

九步按顺序跑，表名里的 bak、dupkeys、dedup 分别是快照表、重复键表和去重后的临时表：

步	做什么	闸	没过就
R0	只读前置检查，并确认执行前提：已经停掉对这个分区的补发；sink 已经不再写这一天；本地盘放得下一整天的新分区	三个副本当天行数相等；当天分区的复制队列为空（不算后台 merge 和 mutation）；没有未完成的 mutation；三张工作表的名字没被占；两张表不会撞同一个 Keeper 路径	等复制追平再跑；名字被占见下文
R1	用 `ATTACH PARTITION … FROM` 给当天做快照，记下快照时刻	快照在三个副本上的行数都等于原表当天的行数，记为 C0	停下，交给人
R2	从快照找出重复键，存进一张小表	组数、多余行数等于预期；每组的业务列哈希数是 1	停下，交给人
R3	从快照建去重后的分区，不读原表，写入带 `insert_quorum = 3`	行数 = C0 − 多余行数；键数 = 快照键数；没有重复键	停下，交给人
R4	换分区前一刻，把原表、去重表、快照表在每个副本上的当天行数各数一遍	原表和快照表都仍等于 C0，去重表都等于 C0 − 多余行数；当天分区的复制队列为空	停掉写入源，删掉三张工作表，从 R1 重来（此时还没换分区）
R5	`REPLACE PARTITION`，紧跟 R4 执行，记下换分区时刻	—	—
R6	事后核对	每个副本行数 = C0 − 多余行数、没有重复键；队列为空；快照之后、换分区之前没有新写入	报出来的行让写入源按键重发
R7	回滚：从快照反向再换一次	每个副本当天行数回到 C0，队列为空	—
R8	收尾：删掉临时表和重复键表，快照表留到下游确认之后	—	—

SQL 的结构照实验 19，按现有集群的 Replicated 库改了两处：DDL 不写 ON CLUSTER，复制表不写引擎参数（实验 19 是 Atomic 库，带 ON CLUSTER 和显式路径）。现有集群的连接随机落到某个节点，整套最好放在同一个 clickhouse-client 会话里跑，落在同一个节点上。做不到的时候（比如经负载均衡一条条发的 HTTP 请求），往临时表写完之后，换分区那一条可能落在还没拉到临时表的副本上，整天的数据会被换成空分区（实验 17 的 D 段）。R3 的 insert_quorum = 3 和 R4 数三张表防的就是这种情况，在同一个会话里跑也照样带上。集群名按 Aiven 文档的示例写 default（Query a database），下面的 {D} 是要清的那一天的分区 ID。

快照和重建（R0 到 R3）#

R0 有五项。前两项要问遍所有副本，连接落到哪个节点，读到的就是哪个副本。复制队列只看当天分区、不算 MERGE_PARTS 和 MUTATE_PART：别的分区随时可能挂着 merge，按整张表数，这道闸会被和当天无关的任务拦住。实验 19 第 5b 段让后一天挂着一个 MERGE_PARTS，整表的队列不空，按当天过滤之后照常通过：

-- 三个副本当天的行数，必须相等
SELECT hostName() AS h, count()
FROM clusterAllReplicas('default', currentDatabase(), events)
WHERE _partition_id = '{D}' GROUP BY h ORDER BY h;

-- 全副本上这个分区的复制队列，不算后台 merge 和 mutation，必须为空
SELECT count() FROM clusterAllReplicas('default', system.replication_queue)
WHERE database = currentDatabase() AND table = 'events'
  AND type NOT IN ('MERGE_PARTS', 'MUTATE_PART')
  AND (new_part_name = '' OR startsWith(new_part_name, '{D}_'));

-- 没有未完成的 mutation
SELECT count() FROM system.mutations
WHERE database = currentDatabase() AND table = 'events' AND NOT is_done;

第四项是三张工作表的名字没被占。被占了，说明上一次跑到一半停了，先停下来由人确认上一次跑到了哪一步：还没执行过 R5，就把三张表 DROP TABLE … SYNC 掉，从 R1 重来；已经执行过 R5，events_bak 是唯一的回滚来源，不能删。重来时不要用 CREATE TABLE IF NOT EXISTS 接着写：它会复用上次写了一半的临时表，两条 INSERT 追加上去就错了（实验 07）。删表一定带 SYNC，Atomic 和 Replicated 库的 DROP 默认是延迟的，不加 SYNC 紧接着建同名、同 Keeper 路径的表会报 REPLICA_ALREADY_EXISTS。现有集群把延迟删除的宽限期和删表的大小保险都设成了 0，删掉就没有 UNDROP，删之前要再确认一次表名（误删之后的恢复）。

第五项是 R1、R3 要建的表不会和原表撞在同一个 Keeper 路径上。CREATE TABLE … AS events 连复制表的 Keeper 路径一起复制：原表路径写死成字面量时，克隆当场报 REPLICA_ALREADY_EXISTS；路径里带 {uuid} 时，克隆拿到自己的 uuid，两个路径不同（实验 04）。Aiven 上 SHOW CREATE TABLE 被拒、system.tables.engine_full 被抹掉，看不到 {uuid} 字样，要拿路径和表自己的 uuid 对比：

SELECT zookeeper_path FROM system.replicas WHERE database = currentDatabase() AND table = 'events';
SELECT uuid FROM system.tables WHERE database = currentDatabase() AND name = 'events';
-- 路径里含这张表自己的 uuid，克隆就会拿到新路径

几套 region 部署里查过的那一套，路径是带 uuid 的；其余几套执行前要先查。Replicated 库里不写引擎参数时克隆拿到新路径，是 lab 上用一次性探针看到的。

R1 到 R3：

-- R1：快照，等三个副本都挂上
SELECT now64(6);                                             -- 快照时刻 t_snap
CREATE TABLE events_bak AS events;
ALTER TABLE events_bak ATTACH PARTITION ID '{D}' FROM events SETTINGS alter_sync = 2;
SELECT hostName() AS h, count()                              -- 三行都是 C0
FROM clusterAllReplicas('default', currentDatabase(), events_bak)
WHERE _partition_id = '{D}' GROUP BY h ORDER BY h;

-- R2：重复键
CREATE TABLE events_dupkeys (id String, settle_ms UInt64, rev Int8)
ENGINE = ReplicatedMergeTree ORDER BY (settle_ms, id, rev);
INSERT INTO events_dupkeys SETTINGS insert_quorum = 3
SELECT id, settle_ms, rev FROM events_bak WHERE _partition_id = '{D}'
GROUP BY id, settle_ms, rev HAVING count() > 1;

-- R2 的闸：组数和多余行数等于预期
SELECT count() AS groups, sum(c - 1) AS extra FROM (
  SELECT count() AS c FROM events_bak
  WHERE _partition_id = '{D}' AND (id, settle_ms, rev) IN (SELECT id, settle_ms, rev FROM events_dupkeys)
  GROUP BY id, settle_ms, rev);

-- R2 的闸：每组重复除 create_time 外内容一样，必须是 0
SELECT count() FROM (
  SELECT uniqExact(cityHash64(tuple(列1, 列2, ……))) AS h      -- 除 create_time 外的全部业务列
  FROM events_bak
  WHERE _partition_id = '{D}' AND (id, settle_ms, rev) IN (SELECT id, settle_ms, rev FROM events_dupkeys)
  GROUP BY id, settle_ms, rev HAVING h > 1);

-- R3：从快照建去重后的分区，三个副本都写进去才返回
CREATE TABLE events_dedup AS events;
INSERT INTO events_dedup SETTINGS insert_quorum = 3
SELECT * FROM events_bak WHERE _partition_id = '{D}'
  AND (id, settle_ms, rev) NOT IN (SELECT id, settle_ms, rev FROM events_dupkeys);
INSERT INTO events_dedup SETTINGS insert_quorum = 3
SELECT * FROM events_bak WHERE _partition_id = '{D}'
  AND (id, settle_ms, rev) IN (SELECT id, settle_ms, rev FROM events_dupkeys)
ORDER BY create_time LIMIT 1 BY id, settle_ms, rev;

-- R3 的闸：三个数都应等于 C0 − 多余行数
SELECT count(), uniqExact(id, settle_ms, rev),
       (SELECT uniqExact(id, settle_ms, rev) FROM events_bak WHERE _partition_id = '{D}')
FROM events_dedup WHERE _partition_id = '{D}';

几个写法上的选择，各有原因：

快照用 ATTACH PARTITION … FROM，在本地盘上只挂硬链接：快照表和原表的列文件是同一个 inode，三个副本都是本地挂、没去拉数据，这条语句写入 0 行（实验 20）。之后只从快照读，原表再有写入也不影响重建。
分区写 PARTITION ID '{D}'。PARTITION ID 后面必须是带单引号的字符串，PARTITION ID 20260310 报语法错误。不带 ID 时跟的是分区表达式，文档说 String 要加引号、Date 和 Int 不需要，实测 Int 加了引号也能用（实验 05，分区操作文档）。PARTITION ID 的规则只有一条，和查询里用的 _partition_id 也是同一个口径。
重建拆成两条 INSERT。没有重复的行用 SELECT * 原样拷，LIMIT 1 BY 只作用在几百个重复键上；整个分区一起 LIMIT 1 BY，要在内存里存下两亿个键。整行原样拷还躲开了按位置对列的坑：手写列表的 INSERT … SELECT 列序一错，整段数据会写进别的列，行数校验照样通过。
R2 本身就是对整个分区按键 GROUP BY，内存跟着键数涨。lab 在 25.8 上测过同一类精确去重：count(DISTINCT id) 每行约 60–65 字节，3 亿行要约 18 GiB，4 vCPU / 16 GB 的服务端内存上限 9.94 GiB，2 亿行时就报 Code: 241（实验 31–33）。GROUP BY 要存完整的键，按理只会更多。分区大到这个量级，R2 按小时切成 24 条 INSERT 写进同一张 events_dupkeys（键里有 settle_ms，同一个键的几份一定落在同一个小时），或者给它设 max_bytes_before_external_group_by，让聚合超过阈值时落盘（实验 34 验过 GROUP BY 子查询能这样落盘跑完）。
写 events_dupkeys 和 events_dedup 都带 insert_quorum = 3，三个副本都写进去才返回，有副本拉不到就报 Code: 319 停下（实验 19 第 5b 段）。默认的 0 只等收到 INSERT 的那个副本，后面换分区要是落在别的副本上，那里可能还是空表。
重复的几份留 create_time 最早的那份，排序写在 ORDER BY create_time 里。create_time 相同的两份仍由 LIMIT 1 BY 任选一份，R2 的哈希闸已经保证两份除 create_time 外一样，选哪份都对。
哈希校验写成 cityHash64(tuple(…))。不包 tuple 时，某一列是 NULL 会让整个哈希变成 NULL，uniqExact 再把它跳过，两种内容只数出一种。这一项不是 0，说明重复的两份内容其实不同，LIMIT 1 BY 会不报错地丢掉一份，要中止交给人判断。
R2 闸里的「预期」来自前面按小时数出来的多余行数。另外设一条占比阈值（多余行不超过当天的 0.1%），高过这条线说明不是零星写重，换分区只是把问题盖住，也该停下来交给人判断。

换分区前后的闸（R4 到 R6）#

R4 和 R5 连着跑，中间不留人工步骤：

-- R4：三张表各数一遍每个副本，每条都要返回三行
SELECT hostName() AS h, count()                              -- 都 = C0
FROM clusterAllReplicas('default', currentDatabase(), events)
WHERE _partition_id = '{D}' GROUP BY h ORDER BY h;
SELECT hostName() AS h, count()                              -- 都 = C0 − 多余行数
FROM clusterAllReplicas('default', currentDatabase(), events_dedup)
WHERE _partition_id = '{D}' GROUP BY h ORDER BY h;
SELECT hostName() AS h, count()                              -- 都 = C0
FROM clusterAllReplicas('default', currentDatabase(), events_bak)
WHERE _partition_id = '{D}' GROUP BY h ORDER BY h;
-- 当天分区的复制队列，写法同 R0，必须为空
SELECT count() FROM clusterAllReplicas('default', system.replication_queue)
WHERE database = currentDatabase() AND table = 'events'
  AND type NOT IN ('MERGE_PARTS', 'MUTATE_PART')
  AND (new_part_name = '' OR startsWith(new_part_name, '{D}_'));

-- R5：换分区，等三个副本都换完
SELECT now64(6);                                             -- 换分区时刻 t_swap
ALTER TABLE events REPLACE PARTITION ID '{D}' FROM events_dedup SETTINGS alter_sync = 2;

R4 要数全部副本：REPLACE PARTITION 换掉的是这个分区里块号在范围内的全部 part，不管执行节点自己有没有拉到，只数执行节点会放行还没复制过来的写入（实验 19 第 3b 段，经过写在行数校验拦不住的情形）。去重表也要数全部副本：换进去的是执行节点本地的那份 events_dedup，执行节点还没拉到它，换进去的就是一个空分区，三个副本一起变成 0 行（实验 17 的 D 段）；快照表是回滚的来源，同样要在每个副本上都齐。某个副本上一行都没有时，GROUP BY h 的结果里不会出现这个副本，所以每条都要数到三行，少一行就当没过。R5 带 alter_sync = 2，等三个副本都执行完才返回，默认的 1 只等当前副本，紧接着的 R6 可能还看到落后副本上的旧数据。

R6 核对每个副本，再查 part_log：

SELECT hostName() AS h, count() AS c, count() - uniqExact(id, settle_ms, rev) AS dup
FROM clusterAllReplicas('default', currentDatabase(), events)
WHERE _partition_id = '{D}' GROUP BY h ORDER BY h;

SYSTEM FLUSH LOGS ON CLUSTER default;
SELECT sum(rows) FROM clusterAllReplicas('default', system.part_log)  -- 必须是 0
WHERE database = currentDatabase() AND table = 'events' AND partition_id = '{D}'
  AND event_type = 'NewPart' AND error = 0
  AND event_time_microseconds >= '{t_snap}' AND event_time_microseconds < '{t_swap}';

part_log 那一条的两个条件都不能省。error = 0 排除被块级去重拦下的插入，它们也记成 NewPart，只是 error = 389（实验 03）。上界卡在换分区时刻之前，因为 REPLACE_RANGE 换进来的 part 在 part_log 里同样记 NewPart，哪个副本是从别处拉的也一样（实验 17），不卡上界就会把换进来的整个分区算成新写入。FLUSH LOGS ON CLUSTER 在 Aiven 上跑不了的话，连到每个节点各跑一次。

R4 和 R5 之间那一瞬间写进来的行，任何闸都拦不住，只能由 R6 事后报出来；取 t_swap 到 ALTER 真正执行之间的那几毫秒，R6 也报不出来。t_snap、t_swap 取的是执行节点的时钟，副本之间有时钟差时，边界上那一小段的写入也可能漏报。回滚只能回到快照，快照里同样没有这些行，补救要靠写入源按键重发。所以执行窗口里要先停掉对这个分区的补发。

回滚和收尾（R7、R8）#

-- R7：回滚，等三个副本都换完
ALTER TABLE events REPLACE PARTITION ID '{D}' FROM events_bak SETTINGS alter_sync = 2;

-- R8：收尾
DROP TABLE IF EXISTS events_dedup SYNC;
DROP TABLE IF EXISTS events_dupkeys SYNC;
-- events_bak 留到对账、下游报表确认之后再 DROP … SYNC

R7 在本地盘上也是硬链接，写入 0 行，lab 上回滚之后，分区里每一行都能在另存的标准答案里找到，键一个不少，行数对得上（实验 19、20）。R7 同样是一次 REPLACE PARTITION，会把快照之后写进这一天的行一起抹掉，包括 R5 之后正常进来的迟到写入。回滚前先停补发，按 R4 的写法数一遍每个副本（应该仍是 C0 − 多余行数）；回滚后按 R6 的写法查 part_log，下界换成 t_swap。快照表在换分区之后还引用着旧 part，删掉它之前这部分空间不会释放。

明细清完之后，下游按 create_time 窗口增量累加的聚合表不会跟着变，受影响的时间桶要按被删那些行的 create_time 定位后重算。报表改成物化视图之后怎么对账、重算，写在报表的对账和重算，那条链路还没部署到现有集群。

实验 19：正常跑、回滚和耗时#

实验 19 的数据照现有集群的情形按比例缩小：当天 10 万行加 18 份重复（12 份和原行的 create_time 相同，6 份晚 5 秒），前后两天各 1000 行。

整套正常跑完，三个副本都是 100000 行、0 个重复键，前后两天没动；拿换分区之前另存的标准答案比，新分区每一行都能在里面找到，键一个不少。回滚之后分区回到原样，快照表还在，可以再换回去重后的版本。几种时机问题注进去之后，各道闸分别拦下了什么，见行数校验拦不住的情形的最后一节。

10 万行的规模上，R3 重建（两条 INSERT，写入 10 万行）用了 46 ms，R5 换分区用了 14 ms、写入 0 行，看不出差别。换到一千多万到三亿多行的日分区上，时间应该主要花在 R2、R3：两步都要把整个分区读一遍，R3 还要写一遍，读写量跟着分区行数线性增长，R5 只挂硬链接。

在现有集群上执行时还要顾到两样：

part_log 的保留期。现有集群上只留约 4 天，R6 要在换完之后及时跑。
权限。R6 的 SYSTEM FLUSH LOGS ON CLUSTER default，执行前按在线换引擎那篇的办法确认。

小结#

数重复用 count() - uniqExact(key)，按分区加排序键首列切成小时窗，比整分区 GROUP BY 便宜得多，也能定位到出问题的时段。
选快照加换分区，是因为执行前能完整验结果、执行后能原样回滚；快照、换分区、回滚在本地盘上都是硬链接，写入 0 行。
从快照重建，不读原表；没有重复的行整行原样拷，重复键按 create_time 留最早一份，哈希校验包一层 tuple。
工作表写入带 insert_quorum = 3；换分区前一刻把原表、去重表、快照表在每个副本上各数一遍，看当天分区的复制队列；换分区带 alter_sync = 2；事后查 part_log，带 error = 0，上界卡在换分区时刻之前。
数重复的查询和 R2 的内存都跟着键数涨，精确去重在 25.8 上实测每行约 60–65 字节；两三亿行的分区要按小时切，或者让 R2 的 GROUP BY 落盘。
工作表的名字被占时先停下，已经换过分区的话快照表是唯一的回滚来源，不能删。

参考资料#

ALTER TABLE … PARTITION — ATTACH PARTITION FROM、REPLACE PARTITION 和两张表要满足的条件，分区表达式和 PARTITION ID 的引号规则
OPTIMIZE … DEDUPLICATE BY
Atomic 库引擎 — DROP 为什么是延迟的，以及 SYNC 修饰符
cluster / clusterAllReplicas — 一条查询问遍所有副本
Aiven for ClickHouse 服务架构 — 单 shard 三节点、无主从、连接随机落到任一节点
Query a database（Aiven） — clusterAllReplicas(default, …) 的示例，集群名 default 的出处
clickhouse-lab — 实验 03、04、05、07、17、19、20，定稿 runbook 的脚本是 experiments/19-dedup-replace-runbook-revised.sh；精确去重的内存出自 25.8 上的规模压测（实验 31–34，脚本在 experiments/scale/，结论汇总在 docs/scale-300m.md）

清理 ClickHouse 重复行的 REPLACE PARTITION runbook

目录