如何通过pgx实现PostgreSQL逻辑复制的全量+增量数据同步

问题分析 #

你当前的逻辑仅能捕获复制槽创建后的增量变更，无法获取已有全量数据；尝试使用USE_SNAPSHOT参数创建复制槽时，触发错误：

CREATE_REPLICATION_SLOT ... USE_SNAPSHOT 必须在事务内调用 (SQLSTATE XX000)

这是因为PostgreSQL要求USE_SNAPSHOT必须在只读事务上下文中执行，且事务内不能有其他前置操作。

解决方案步骤 #

1. 在只读事务中创建带快照的复制槽 #

使用pgx库时，需手动开启只读事务，在事务内执行创建复制槽命令：

// 假设conn是已建立的pgx连接
tx, err := conn.BeginTx(context.Background(), pgx.TxOptions{ReadOnly: true})
if err != nil {
    // 处理错误逻辑
}
defer tx.Rollback(context.Background())

// 替换占位符为你的复制槽名和逻辑插件名（如wal2json）
createSlotQuery := fmt.Sprintf("CREATE_REPLICATION_SLOT %s LOGICAL %s USE_SNAPSHOT", "my_custom_slot", "wal2json")
_, err = tx.Exec(context.Background(), createSlotQuery)
if err != nil {
    // 处理错误逻辑
}

2. 利用事务快照导出全量数据 #

在同一个只读事务内，基于当前事务的一致性快照，查询需要同步的所有表的全量数据：

// 示例：导出users表全量数据
rows, err := tx.Query(context.Background(), "SELECT * FROM users")
if err != nil {
    // 处理错误逻辑
}
defer rows.Close()

// 遍历并处理全量数据（如写入目标存储）
for rows.Next() {
    // 解析行数据、落地存储的逻辑
}

3. 提交事务并启动CDC增量复制 #

全量数据导出完成后，提交事务，随后启动逻辑复制接收后续增量变更：

// 提交事务，快照会绑定到复制槽的起始LSN
err = tx.Commit(context.Background())
if err != nil {
    // 处理错误逻辑
}

// 启动复制，指定复制槽名，用'0/0'让PostgreSQL自动从快照后的LSN开始发送增量
startReplicationQuery := fmt.Sprintf("START_REPLICATION SLOT %s LOGICAL 0/0", "my_custom_slot")
// 参考pgx官方文档中ReplicationConn的用法，处理WAL变更流

关键注意事项 #

• 确保使用的逻辑复制插件（如wal2json、test_decoding）支持快照功能，PostgreSQL 13对主流插件兼容性无问题
• 全量查询必须与创建复制槽在同一个只读事务内，才能保证数据一致性
• 启动复制时用'0/0'作为起始LSN，PostgreSQL会自动从复制槽关联的快照对应的LSN开始推送增量，不会重复处理全量数据

内容的提问来源于stack exchange，提问作者Vladislav Len