PostgreSQL自定义CDC如何实现全量同步及WAL格式适配？

一、解决逻辑复制槽无历史数据的核心思路 #

要实现全量+增量的无缝同步，关键是保证全量数据导出和增量WAL捕获的一致性，避免数据丢失或重复，具体步骤如下：

• 1. 创建一致性快照：开启事务执行BEGIN TRANSACTION ISOLATION LEVEL REPEATABLE READ;，后续全量导出的数据均为该时间点的快照，且不会阻塞正常读写（除非存在长事务）。
• 2. 创建逻辑复制槽：在快照生效期间，执行SELECT pg_create_logical_replication_slot('your_slot_name', 'wal2json');，确保复制槽的起始LSN与快照时间点对齐。
• 3. 分块导出全量数据：在快照事务内，通过主键范围或分页方式分块查询目标表数据（比如WHERE id > :last_id LIMIT 1000），避免一次性加载大量数据导致内存溢出。
• 4. 切换增量同步：全量导出完成后提交快照事务，随后开始通过pg_logical_slot_get_changes消费复制槽中的WAL变更，此时捕获的增量数据均为全量快照之后的内容。

二、分块查询全量数据的可行性及格式转换 #

可行性 #

完全可行，这是自定义CDC初始化阶段的标准方案，只要配合上述一致性快照机制，就能实现全量与增量的无缝衔接，不会出现数据不一致问题。

格式转换方法 #

wal2json通过pg_logical_slot_get_changes返回的典型INSERT操作JSON格式如下：

{
  "change": [
    {
      "kind": "insert",
      "schema": "public",
      "table": "users",
      "columnnames": ["id", "name", "email"],
      "columntypes": ["integer", "text", "text"],
      "columnvalues": [1, "Alice", "alice@example.com"]
    }
  ]
}

要将分块查询结果转换成该格式，可直接在SQL中构造对应结构，示例如下：

SELECT json_build_object(
  'change', json_agg(
    json_build_object(
      'kind', 'insert',
      'schema', 'public',
      'table', 'users',
      'columnnames', ARRAY['id', 'name', 'email'],
      'columntypes', ARRAY['integer', 'text', 'text'],
      'columnvalues', ARRAY[id, name, email]
    )
  )
)
FROM public.users
WHERE id > 0 LIMIT 1000; -- 按主键范围分页

通用化处理可通过查询information_schema.columns获取表的列名和类型，再动态生成JSON结构；也可在应用层（如Python/Go）中读取表结构后，将每行数据映射为对应格式。

注意事项：

• 确保columnnames、columntypes、columnvalues的列顺序完全一致，避免值与列不匹配。
• 保留columnvalues中的NULL值，与wal2json输出逻辑保持一致。
• 优先使用主键范围分页替代OFFSET，提升大数据量表的查询效率。

内容的提问来源于stack exchange，提问作者ayb jax