背景#

去重方案的主体是把明细表从 ReplicatedMergeTree 换成 ReplicatedReplacingMergeTree(create_time)，键不变。难处在于这张表一直在被 Kafka Connect 的 sink 写，两周前的分区还会被人工补发写进几千行。要做到四件事：不丢、不重、能回滚，sink 只停很短的一会儿。

下面的步骤在 clickhouse-lab 的实验 27 里整套跑过：25.3.14.14，三副本，库引擎和 Aiven 上的现有集群一样是 Replicated，sink 用脚本模拟、一直在写，中途还人工补发了一次。还没在现有集群上执行。

前提有五条，后四条和现有集群有关，动手前逐条过一遍：

  • 只换引擎，别的一概不动。ATTACH PARTITION … FROM 和 REPLACE PARTITION 要求两张表结构相同，分区键、ORDER BY、主键、存储策略相同，新表包含旧表全部的跳数索引和 projection（ATTACH PARTITION FROM）。默认值和列序文档没列，照旧表抄是为了稳妥。现在那张明细表上还有一条 create_time 满 45 天搬到对象存储的 TTL，新表也要带上，不然切换之后的新数据不再往冷层搬；这条 TTL 在 Aiven 上看不到建表语句，是从 system.parts 的 move_ttl_info 读出来的，lab 没有对象存储，这一项没验过。排序键或分区键也要改的话，只能 INSERT … SELECT 重写数据，这份步骤不适用。
  • SYSTEM STOP MERGES 要能用，行数闸靠它。Aiven 的文档没提这条语句，只说主账号 avnadmin 拥有服务开放的全部权限、用户不能再给它加权限（Manage users and roles）。执行前用 SHOW GRANTS FINAL（连同角色带来的权限一起列）查一下，结果里要有 SYSTEM MERGES，或者覆盖它的 SYSTEM、ALL；后面用到的 SYSTEM SYNC REPLICA、SYSTEM FLUSH LOGS 也要在里面。
  • SYSTEM 语句在 Replicated 库里不会自动传播，下面写成 ON CLUSTER default。Aiven 的文档以前写着不支持 ON CLUSTER，2026-05-28 删掉了这一条，改成对 Replicated 库的 DDL 不必写（Query a database、aiven-docs #1585），示例里的集群名是 default。SYSTEM 语句文档没提，开始前先跑一条没有副作用的 SYSTEM FLUSH LOGS ON CLUSTER default，三个节点都返回、状态都是 0 再往下走。不能用的话，要连到每个节点各跑一遍，STOP MERGES 只落在一个副本上，行数闸就不成立了。
  • 现有集群的 default profile 把 distributed_ddl_output_mode 设成了 null_status_on_timeout：带 ON CLUSTER 的语句在某个节点上超过 distributed_ddl_task_timeout 还没执行完，不报错，返回里那个节点的状态是 NULL（Settings.cpp @ v25.3.13.19-lts），STOP MERGES 漏了一个节点也不会报错。这一项没被锁成常量，在 clickhouse-client 会话开头先 SET distributed_ddl_output_mode = 'throw'，超时就会报错。
  • 分区在本地盘上时，ATTACH 和 REPLACE 只挂硬链接（实验 20、27）。两亿行那套部署约三分之二、三亿多行那套近九成的数据已经在对象存储上，这部分分区走不走硬链接、要不要真拷数据，没验过。

下面的 {db} 是库名，集群名按 Aiven 文档的示例写 default，别的环境用 SELECT DISTINCT cluster FROM system.clusters 查。Replicated 库里的建表、EXCHANGE 这类 DDL 自动在三个节点上执行，不写 ON CLUSTER。

建新表，停掉它的 merge# 

CREATE TABLE {db}.events_new (...与 events 完全相同的列和索引...)
ENGINE = ReplicatedReplacingMergeTree(create_time)
PARTITION BY toYYYYMMDD(toDateTime(settle_ms / 1000))
ORDER BY (settle_ms, id, rev)
PRIMARY KEY (settle_ms, id)
TTL create_time + INTERVAL 45 DAY TO VOLUME 'remote'  -- 照旧表；卷名以 move_ttl_info 读到的为准
SETTINGS storage_policy = 'tiered';

SYSTEM STOP MERGES ON CLUSTER default {db}.events_new;

新表停 merge 有两个原因。ReplacingMergeTree 合并时会折掉重复、改变行数，停了之后后面才能拿行数做闸；新表在切换之前也没人读，合并没有意义。旧表的 merge 不停，它一直在被 sink 写，停了会堆 part。

建完之后逐项比两张表的 system.columns（position、name、type、default_kind、default_expression）、system.tables（partition_key、sorting_key、primary_key、storage_policy）和 system.data_skipping_indices，必须完全一样。

在线搬历史分区#

先在服务端取一次时刻（SELECT now64(6)），记作 T1，后面查 part_log 用的是同一个时钟。然后一个分区一个分区地挂过去，sink 不用停：

ALTER TABLE {db}.events_new ATTACH PARTITION ID '{p}' FROM {db}.events;

实验 27 里五天、每天 20 万行的历史，ATTACH 一共写了 0 行，只挂硬链接；另外两个副本也是在本地挂，没去拉数据（实验 20）。在 Replicated 库里，分区操作不走库的复制日志，库引擎只在当前副本上挂载或删除 part（Replicated 库引擎），另外两个副本靠表自己的复制拿到结果，lab 上用一次性探针看到的也是这样。分区可以慢慢搬。

搬的过程中，旧表当天的分区还在被 sink 写，已经搬过的历史分区也可能被人工补发。这些变化留到下一步一次补齐。

停 sink、排空、按 part_log 补齐#

先在 Confluent Cloud 上暂停 connector，消息留在 Kafka 里，恢复之后接着写。

排空要看两件事。第一，三个副本上都没有 sink 的 INSERT 还在跑：查 clusterAllReplicas('default', system.processes)，按 sink 的用户和 query_kind = 'Insert' 筛。要看服务端，因为客户端超时的那条 INSERT 在服务端可能还没结束，还会晚到提交（超时重投那篇）。connector 默认同步插入，服务端没有攒着没落盘的数据。第二，两张表在每个副本上追平，复制队列里除了合并任务没有别的条目：

SYSTEM SYNC REPLICA ON CLUSTER default {db}.events LIGHTWEIGHT;
SYSTEM SYNC REPLICA ON CLUSTER default {db}.events_new LIGHTWEIGHT;

SELECT count()
FROM clusterAllReplicas('default', system.replication_queue)
WHERE database = '{db}' AND table IN ('events', 'events_new') AND type != 'MERGE_PARTS';  -- 必须是 0

这里别等 queue_size 归零，也别用不带 LIGHTWEIGHT 的 SYNC REPLICA。新表停着 merge，有可合并的 part 时 leader 照样给它排合并任务，任务执行不了，就一直挂在队列里。lab 里 6 个小 part 停着 merge，队列始终不归零，不带 LIGHTWEIGHT 的 SYNC REPLICA 一直等到超时报 Code: 159，带上 LIGHTWEIGHT 立刻返回（实验 27）。

然后补齐。T1 之后旧表有新 part 的分区，就是要补的分区，逐个用 REPLACE PARTITION 重新同步一遍：

SYSTEM FLUSH LOGS ON CLUSTER default;

SELECT DISTINCT partition_id
FROM clusterAllReplicas('default', system.part_log)
WHERE database = '{db}' AND table = 'events'
  AND event_type IN ('NewPart', 'MutatePart') AND error = 0
  AND event_time_microseconds >= '{T1}';

-- 对每个 {p}：
ALTER TABLE {db}.events_new REPLACE PARTITION ID '{p}' FROM {db}.events SETTINGS alter_sync = 2;

alter_sync = 2 让这条语句等三个副本都执行完才返回，默认的 1 只等当前副本。实验 27 里查出来的正好是两个分区：sink 在写的当天，和搬完之后被补发了 50 行的那一天。error = 0 用来排除被块级去重拦下的插入，它们在 part_log 里也记成 NewPart。现有集群上 part_log 只留约 4 天，搬历史和补齐之间别隔太久，不然查不全。

行数闸：每个副本、每个分区都一样#

只读 system.parts，不扫数据：

SELECT hostName() AS h, partition_id,
       sumIf(rows, table = 'events') AS a, sumIf(rows, table = 'events_new') AS b
FROM clusterAllReplicas('default', system.parts)
WHERE database = '{db}' AND table IN ('events', 'events_new') AND active
GROUP BY h, partition_id
HAVING a != b;  -- 必须是空的

这道闸成立的前提就是新表停着 merge：两边的 part 都还没被 ReplacingMergeTree 折叠过，行数应该逐分区、逐副本相等。实验 27 里这条查询是空的，搬迁前混进去的 300 行重复也原样搬了过来，那一天新表的物理行数是 200300。

用 EXCHANGE 切换# 

EXCHANGE TABLES {db}.events AND {db}.events_new;
-- 恢复 connector
SYSTEM START MERGES ON CLUSTER default {db}.events;

EXCHANGE 原子地交换两张表的名字。用多表 RENAME 不行，文档写明一条 RENAME 改多张表不是原子的，其他会话可能读到 Table ... does not exist（RENAME），sink 正好在那一刻写进来就会失败。

挂在旧表名上的物化视图跟着名字走，切换之后写进新 events 的数据照常触发它（实验 27）。sink 从暂停到恢复，results/ 里这一轮是 443 ms，其中要补的是 2 个分区；按 lab 文档的记录，前后 5 次运行在 0.34 到 0.50 秒之间。到了现有集群上，这段时间主要看要补几个分区，每个是一条 REPLACE。

切换之后有一处缺口，是按机制推出来的，还没测过：ATTACH 和 REPLACE 搬的是 part，旧表记在 Keeper 里的块级去重标记不会跟过来，新表的去重窗口是空的。比如暂停之前有一批 INSERT 客户端超时、服务端晚到提交进了旧表，补齐时被一起搬了过来；connector 恢复之后，框架把内存里还留着的这一批带着同一个 token 重投，写进的是新表，新表认不出来，会多一份物理行。FINAL 折得掉它，物化视图喂的报表却可能多算，切换那天按报表对账那篇对一次账。

切换之后核对的五项#

实验 27 切换之后核了这几项，全部通过：

核对项 结果
sink 写过的每个 offset 在新表里都在、只有一份 6300 个 offset，物理行数 6300
历史 5 天新表 FINAL 的行数等于旧表里不同键的个数 1000000 / 1000000
搬迁前就有的 300 行重复被 FINAL 折掉 那一天 FINAL 读出 200000 行
人工补发的行 50 行都在
下游物化视图在切换前后 当天的行数等于 sink 写到的 offset，不漏也不多

回滚#

切换之后发现问题，按同样的思路倒回去：

  1. 停 sink，在服务端取时刻 T2，按前面的办法排空；两张表都 SYSTEM STOP MERGES。
  2. 在一个副本上，把新表（现在叫 events）的每个分区都 REPLACE PARTITION … FROM 回旧表（现在叫 events_new），带 alter_sync = 2。这里不按 part_log 挑分区：新表切换之后开了 merge，合并会改变行数，没被写过的分区行数也对不上了。全部同步一遍，本地盘上都是硬链接，代价只是 REPLACE 的条数。
  3. 过闸。拿执行 REPLACE 的那个副本上的新表当基准，和每个副本上的旧表逐分区比行数；T2 之后新表不能再有新写进来的 part。
  4. 再 EXCHANGE 回去，开 merge，恢复 sink。

第 3 步的闸不能照搬前面那种各副本各比各的写法。新表停 merge 的那一刻，各副本的合并进度不一定相同。实验 27 特意造了这个局面：只在 ch2 上停掉 merge，往当天写进 100 行重复，ch1 上合并掉了，ch2 上还留着；各比各的闸在这里会报对不上。拿执行副本当基准的写法：

-- 在执行 REPLACE 的那个副本上跑：三个副本都要出现，same 都是 1
WITH (SELECT arraySort(groupArray((partition_id, c))) FROM
        (SELECT partition_id, sum(rows) AS c FROM system.parts
         WHERE database = '{db}' AND table = 'events' AND active GROUP BY partition_id)) AS ref
SELECT h, got = ref AS same FROM
  (SELECT h, arraySort(groupArray((partition_id, c))) AS got FROM
     (SELECT hostName() AS h, partition_id, sum(rows) AS c
      FROM clusterAllReplicas('default', system.parts)
      WHERE database = '{db}' AND table = 'events_new' AND active GROUP BY h, partition_id)
   GROUP BY h);

实验 27 里回滚之后，切换前后 sink 写过的每个 offset 都在、只有一份（8200 个 offset，物理行数 8200），补发的 50 行也在。

旧表留到确认不回滚再删。它和新表共用硬链接，删掉之前，旧 part 占的空间不会释放。上游默认删超过 50 GB 的表会被 max_table_size_to_drop 拦下，现有集群上这道保险是 0，删的时候不会被拦，动手前要再确认一次删的是旧表（误删之后的恢复）。

不要这样做#

下面三种看上去更省事的做法，前两种在 lab 上跑出了问题，第三种文档写明了不行：

  • 不排空就切换。正在跑的 INSERT 在开始时就绑定了那张表本身，不跟着名字走。lab 里一条要跑 3 秒的 INSERT 写到一半时 EXCHANGE，那 30 行全落进了切换之后的旧表，新表里一行没有（实验 27）。
  • 用同名的 UNION ALL 视图拼新旧两张表。sink 恢复之后写的是这个视图，普通视图不能写，lab 上往它 INSERT 直接报 Code: 48（实验 27）。48 不在 connector 当作可重试的错误码里，默认 errors.tolerance=none 时 task 会停下，这一句是按源码推的。
  • 用多表 RENAME 当原子切换。原因见上一节。

小结#

  • 只换引擎时，ATTACH PARTITION … FROM 能在线把历史分区挂进新表，本地盘上只挂硬链接、写入 0 行；新表要带上旧表的搬迁 TTL，对象存储上的分区怎么走还没验过。
  • 新表停着 merge，是为了拿行数做闸；停 merge 的表复制队列不会归零，排空要用 SYNC REPLICA … LIGHTWEIGHT，再看队列里除了 MERGE_PARTS 还有没有别的条目。
  • 停 sink 之后按 part_log 找出搬迁期间被写过的分区，用 REPLACE PARTITION 补齐，逐副本、逐分区比行数过闸，再 EXCHANGE 切换。lab 上 sink 只停了零点几秒。
  • 回滚是同样的路子反过来，闸要拿执行 REPLACE 的副本当基准。
  • 新表的块级去重窗口是空的，切换那天要对一次报表的账。

参考资料#

相关文章#