返回文章列表

PostgreSQL Logical Replication、CDC、Schema Evolution 与在线大版本升级

从 WAL、逻辑解码、Publication、Slot、Apply Worker、Replication Origin、Schema Evolution、CDC Consumer 与在线大版本升级出发，理解 PostgreSQL 逻辑复制的数据流水线。

第 22 章：Logical Replication、CDC、Schema Evolution 与在线大版本升级

技术基线：PostgreSQL 18；同时标注 PostgreSQL 14、15、16、17、18 的关键差异。示例默认发布端为 publisher，订阅端为 subscriber，两端属于不同 PostgreSQL 实例。 核心结论：逻辑复制不是“自动搬迁整个数据库”，而是一条由 WAL、逻辑解码、Publication、Slot、协议、Apply Worker、Replication Origin 和人工维护的 Schema/Sequence 共同组成的数据流水线。它可以把停机窗口压缩到秒级或分钟级，但只有在幂等、校验、回滚和 Fencing 都设计完整时，才称得上在线迁移。

1. 本章定位

1.1 本章解决什么问题

本章解决四类生产问题：

  1. 单表或一组表在线迁移：旧库持续写入时完成 Backfill，并在短暂停写后切换。
  2. CDC（Change Data Capture）：将 PostgreSQL 提交后的行级变化可靠地投递到搜索、缓存、湖仓、审计或事件系统。
  3. PostgreSQL 大版本低停机升级：使用旧版本作为 Publisher、新版本作为 Subscriber，完成跨版本全量与增量同步后切流。
  4. 跨系统数据同步：把 PostgreSQL 变更转换为目标系统可接受的事件，同时处理重复、乱序边界、Schema 变化和故障恢复。

1.2 为什么生产环境必须掌握

物理复制复制的是块级 WAL，适合同一 PostgreSQL 主版本、同一集群形态的高可用；逻辑复制复制的是经过解码和筛选的行变化，适合按表选择、跨版本迁移、跨平台同步和 CDC。它提供了灵活性，也把以下责任交给了使用者：

  • 目标端表结构、索引、约束、权限和扩展对象必须预先准备；
  • DDL、Sequence 当前值、Large Object、部分对象不会自动复制；
  • Slot 可能长期保留 WAL，最终写满磁盘；
  • 消费者崩溃后可能重放事件，因此通常只能承诺 At-Least-Once；
  • Subscriber 上的本地写入可能产生唯一键冲突或 Missing Row；
  • 切流后是否能回滚，取决于是否保留反向增量链路，而不是取决于是否“还留着旧库”。

1.3 与前后章节的依赖关系

  • 依赖第 13 章的 WAL、Checkpoint、Crash Recovery 与 LSN。
  • 依赖第 15 章的在线 DDL、Expand/Contract 和兼容窗口。
  • 依赖第 17 章的幂等、Outbox、Commit Outcome Unknown 与 Backpressure。
  • 依赖第 20 章的 Backup/PITR；逻辑复制不能替代备份。
  • 依赖第 21 章的物理复制与 Promotion；[PG17+] Logical Slot Failover 需要与物理副本协同。
  • 为第 23 章的 Patroni、Fencing 和综合 HA 提供逻辑槽切换基础。

1.4 本章不展开的内容

本章不深入自定义 Output Plugin 的 C 源码、不构建完整 Debezium/Kafka Connect 替代品、不讲分布式全局事务，也不把双向逻辑复制描述成自动冲突解决方案。Go 示例是一个可验证协议和可靠性边界的教学实现，不是完整生产 CDC 平台。

2. 可验证的学习目标

完成本章后，你应当能够：

  1. 从一次 UPDATE 的 WAL 产生开始，解释它经过 Logical Decoding、pgoutput、Slot、网络和 Apply Worker 到达 Subscriber 的完整路径。
  2. 使用 CREATE PUBLICATION、CREATE SUBSCRIPTION 建立初始同步与持续复制，并通过系统视图验证每张表的同步状态。
  3. 准确区分 restart_lsn、confirmed_flush_lsn、received_lsn、latest_end_lsn 与 Replication Origin 的 remote_lsn。
  4. 为 DEFAULT、USING INDEX、FULL、NOTHING 四种 Replica Identity 选择合适场景，并解释其 WAL、查找和并发代价。
  5. 验证 Row Filter、Column List、Partition Root 与 [PG18] Stored Generated Column 的复制行为和版本边界。
  6. 复现 insert_exists 与 update_missing 冲突，判断哪些冲突停止 Apply，哪些变化会被跳过。
  7. 设计 Publisher 与 Subscriber 的 Expand/Contract Schema Evolution 顺序，避免新旧 Schema 不兼容导致复制停滞。
  8. 设计单表迁移、CDC、大版本升级和跨系统同步的全量、增量、校验、切流、回滚与 Slot 清理流程。
  9. 通过 Slot 视图计算 WAL 保留量，判断消费者慢、长事务、失活 Slot 或 Checkpoint 配置造成的磁盘风险。
  10. 解释 [PG14+] Large Transaction Streaming、[PG15+] Two-Phase Replication、[PG16+] Parallel Apply 和 [PG17+] Logical Slot Failover 的边界。
  11. 使用 Go、pglogrepl、pgx/v5 与 pgxpool 实现带 Keepalive、Status Update、持久化 LSN Checkpoint、有界队列、幂等落库和 Schema Version 的简化 CDC Consumer。
  12. 在模拟大版本切流中，把写入冻结窗口、最终 LSN、Sequence 对齐、连接切换、验证和回滚条件写成可执行 Runbook。

3. 核心术语

中文名称English准确定义容易混淆的概念所属层次
逻辑解码Logical Decoding从 WAL 中重建事务级、行级逻辑变化的过程不是读取 SQL 文本，也不是审计日志Publisher/WAL
发布Publication定义哪些表、操作、行和列可被发布不保存消费进度，不主动发送数据Publisher/Catalog
订阅SubscriptionSubscriber 上的连接、Publication 集合、Slot 与 Apply 配置不是备份，也不是自动 Schema 同步Subscriber/Catalog
逻辑复制槽Logical Replication Slot为单个数据库保留解码状态和所需 WAL 的持久对象与物理 Slot 不同；一个 Slot 同时只能有一个活跃消费者Publisher
输出插件Output Plugin将逻辑变化编码为外部协议；内置逻辑复制使用 pgoutputwal2json 等是其他插件Publisher
复制源Replication OriginSubscriber 记录远端提交位置和来源的机制，用于进度、环路过滤和来源追踪不等同于 Slot，也不等同于 TimelineSubscriber
初始表同步Initial Table Sync对订阅表创建一致性 Snapshot，执行 COPY，再追赶增量的过程不是普通 pg_dump；每表可由独立 Worker 完成Subscriber/Publisher
表同步 WorkerTable Sync Worker为单张表执行 Snapshot、COPY 和 Catch-up 的特殊 Apply Worker与长期运行的主 Apply Worker 不同Subscriber
Apply WorkerApply Worker按远端事务顺序将逻辑变化映射并写入本地表默认以 session_replication_role=replica 运行Subscriber
一致性快照Snapshot确定初始 COPY 可见行集合的 MVCC 边界与 Slot 的永久进度不是同一个对象MVCC
COPY 阶段COPY Phase初始同步期间批量复制快照可见行会产生 Subscriber 本地 WAL；触发器行为与持续 Apply 不同Initial Sync
副本标识Replica IdentityPublisher 为 UPDATE/DELETE 提供目标行定位信息的规则不是 Subscriber 的主键定义，但两端必须能匹配Table/WAL
行过滤Row Filter在 Publisher 按表达式筛选要发布的行不是安全边界；多个 Publication 可能按 OR 合并Publication
列列表Column List限制发布的列集合不是列权限；缺列会影响目标业务语义Publication
分区根发布Partition Root使用根分区表的身份和列定义发布叶子分区变化与直接发布叶子表的 Relation Identity 不同Publication
重启 LSNrestart_lsnSlot 仍可能需要的最早 WAL 位置通常早于 confirmed_flush_lsnSlot
确认刷新 LSNconfirmed_flush_lsn逻辑消费者已确认持久接收的位置；更早变化不再可获取不代表外部业务一定已完成，取决于何时 ACKSlot
大事务流式传输Large Transaction Streaming在事务提交前分段传输其已解码变化，避免全部留在 Publisher 内存/磁盘不等于打破事务原子性Protocol
两阶段复制Two-Phase Replication在 PREPARE TRANSACTION 时把远端 Prepared Transaction 复制为本地 Prepared Transaction不等于分布式事务协调器Protocol/Apply
冲突Subscriber Conflict远端变化无法按预期应用，或与本地来源/数据不一致Constraint Error 与 Missing Row 的处理不同Subscriber
LSN 检查点LSN CheckpointCDC 消费者持久化“已安全落地”的远端提交末端 LSN不能在下游事务提交前更新Consumer
至少一次At-Least-Once故障恢复后事件不会静默丢失，但可能重复不等于 Exactly-OnceDelivery Semantics
幂等消费者Idempotent Consumer重放同一事务或事件不会产生额外业务副作用不能仅靠内存去重Consumer/Sink
Schema 版本Schema Version对 Relation 元数据生成稳定标识，使事件可按产生时结构解释不只是应用发布版本号CDC Contract
回填Backfill在增量链路运行前后复制历史数据必须与增量边界建立一致关系Migration
双写Dual Write应用同时写旧、新目标没有原子性/幂等时容易产生分叉Application
切流Cutover将读取或写入权威从旧系统切换到新系统DNS/连接池旧连接可能延长实际切换Migration
回滚Rollback把权威重新切回旧系统并保持切流后变化不丢失“旧库还在”不代表能无损回滚Migration/HA

4. 整体心智模型

PostgreSQL Logical Replication、CDC、Schema Evolution 与在线大版本升级 flow 1

4.1 数据流

  1. 业务事务修改 Heap/Index，同时写入 WAL。
  2. Slot 绑定逻辑解码上下文；WAL Sender 从 restart_lsn 之后读取必要 WAL。
  3. Logical Decoding 按事务重建变化，pgoutput 根据 Publication 过滤并编码 Relation、Begin、Insert/Update/Delete、Commit 等消息。
  4. 内置 Subscription 将消息交给 Apply Worker；自定义 CDC 则由客户端自行解码和持久化。
  5. Apply Worker 在 Subscriber 中开启本地事务，按远端事务边界执行 DML，提交后更新 Replication Origin。
  6. 消费者发送 Status Update 后，Publisher 才能推进 confirmed_flush_lsn；随着不再需要旧事务和 Catalog Snapshot，restart_lsn 才能继续推进。

4.2 控制流

  • Publication 决定“可以发送什么”。
  • Subscription/CDC 客户端决定“从哪个 Slot、哪个 LSN 开始、何时确认”。
  • Replica Identity 决定 UPDATE/DELETE 中携带什么旧值，并影响 Subscriber 如何定位目标行。
  • Replication Origin 记录远端进度；Slot 记录 Publisher 端可回收边界。
  • Schema Evolution、切流、回滚与 Slot 删除必须由运维流程控制。

4.3 状态变化

初始同步不是一个原子瞬间，而是每表状态机：

PostgreSQL Logical Replication、CDC、Schema Evolution 与在线大版本升级 flow 2

pg_subscription_rel.srsubstate 中：i=initialize，d=copying，f=finished table copy，s=synchronized，r=ready。

4.4 故障路径

  • 网络中断：Slot 保留尚未确认的 WAL，恢复后重放；会出现重复而不应静默丢失。
  • Subscriber 唯一键冲突：Apply 报错并停止或在 disable_on_error=true 时禁用 Subscription。
  • UPDATE/DELETE Missing Row：[PG18] 记录冲突统计，但操作被跳过，复制继续；数据已发生语义分叉。
  • CDC Sink 提交成功但 ACK 丢失：事件会重放；需要持久幂等键。
  • Slot 长期无消费者：restart_lsn 停滞，pg_wal 增长；达到限制后 Slot 可能失效，随后只能重建/重做初始同步。
  • Publisher 故障转移：只有 [PG17+] 正确配置并已同步的 Failover Slot 才可能在新 Primary 延续；仍必须 Fencing 旧 Primary。

5. 使用方式

5.1 PostgreSQL 14—18 关键版本矩阵

能力PG14PG15PG16PG17PG18
基础 Publication/Subscription/pgoutput✓✓✓✓✓
大事务进行中流式传输，协议 v2新增✓✓✓✓
Row Filter、Column List—新增✓✓✓
ALTER SUBSCRIPTION ... SKIP—新增✓✓✓
Two-Phase Logical Replication—新增✓✓✓
Standby 上 Logical Decoding——新增✓✓
大事务 Parallel Apply、协议 v4——新增✓✓
Logical Failover Slot———新增✓
pg_upgrade 保留 Logical Slot/Subscription 状态———起始版本要求✓
Stored Generated Column 复制————新增
Subscription streaming 默认值offoffoffoffparallel
细分逻辑复制冲突日志与计数基础错误基础统计基础统计基础统计增强
idle_replication_slot_timeout————新增

跨版本升级通常采用“旧版本 Publisher → 新版本 Subscriber”。不要假设新版本 Publisher 的新协议/类型能力能被旧 Subscriber 理解；跨版本时优先使用文本传输，即 binary=false。

5.2 Publisher 前置配置

postgresql.conf：

wal_level = logical
max_wal_senders = 20
max_replication_slots = 20

# 需要结合允许的消费者中断时长、峰值 WAL 速率和磁盘余量评估。
# -1 表示 Slot 可无限保留 WAL，风险是写满磁盘。
max_slot_wal_keep_size = '200GB'

# [PG18] 只用于治理长期失活 Slot；检查发生在 Checkpoint，不能当精确计时器。
idle_replication_slot_timeout = '24h'

# 大事务在逻辑解码端的内存阈值；超过后可能落盘或按协议流式发送。
logical_decoding_work_mem = '256MB'

修改 wal_level、max_wal_senders、max_replication_slots 通常需要重启。任何数值都必须根据写入峰值、消费者恢复时间、磁盘容量、RPO/RTO 和并发 Subscription 数量压测，而不是照抄。

创建最小权限复制角色：

CREATE ROLE repl_login
    WITH LOGIN REPLICATION PASSWORD 'replace-with-secret';

GRANT CONNECT ON DATABASE appdb TO repl_login;
GRANT USAGE ON SCHEMA app TO repl_login;
GRANT SELECT ON TABLE app.orders TO repl_login;

pg_hba.conf 需要同时允许普通数据库连接和复制连接。生产中使用 TLS、Secret 管理与地址白名单，不把密码写入脚本仓库。

5.3 创建 Publication

CREATE PUBLICATION pub_orders
FOR TABLE app.orders
WITH (
    publish = 'insert, update, delete, truncate',
    publish_via_partition_root = false
);

带行过滤和列列表：

CREATE PUBLICATION pub_tenant_42
FOR TABLE app.orders (
    id,
    tenant_id,
    order_no,
    status,
    amount,
    updated_at
)
WHERE (tenant_id = 42 AND status <> 'draft')
WITH (publish = 'insert, update, delete');

注意：

  • Row Filter 表达式在 Publisher 上求值；false 或 NULL 都不发布。
  • UPDATE 可能被转换为 INSERT/DELETE 语义：旧行不匹配而新行匹配时，相当于进入过滤集合；反之相当于离开集合。
  • 对 UPDATE/DELETE，过滤表达式只能依赖 Replica Identity 可用列及受支持的简单表达式。
  • 同表在多个 Publication 中的 Row Filter 通常按“任一匹配”发布；但不同 Column List 的组合受限制，应在建 Subscription 前验证。
  • Row Filter/Column List 不是权限或数据脱敏安全边界。Subscriber 管理员、复制连接和 WAL 仍需独立治理。
  • 初始同步会考虑 Row Filter，但不会按 publish='insert' 之类的操作列表排除历史行。
  • PG15 以前的 Subscriber 在初始同步阶段可能忽略 Row Filter/Column List；跨版本必须检查最老一端行为。

5.4 [PG18] Generated Column 复制

PG18 可以发布 stored generated column。两种常见方式：

-- 对未显式写 Column List 的表，发布 stored generated columns。
CREATE PUBLICATION pub_invoice
FOR TABLE app.invoice
WITH (publish_generated_columns = 'stored');
-- 显式列列表中包含 stored generated column。
CREATE PUBLICATION pub_invoice_explicit
FOR TABLE app.invoice (
    id,
    subtotal,
    tax,
    total_with_tax
);

边界：

  • PG18 以前，生成列不作为发布值发送。
  • Publisher 的 generated column 可以发给 Subscriber 的普通列；这适合把计算结果固化到目标端。
  • 不要把发布值直接写入 Subscriber 的 generated column；该列由 Subscriber 自己计算，写入会失败。
  • 若两端都保留同一生成表达式，常见做法是不发布生成列，让 Subscriber 重新计算；必须验证函数、Collation、时区和表达式语义一致。
  • Virtual Generated Column 与 Stored Generated Column 的存储/复制语义不同；本章的“可发布生成列”特指 PG18 文档支持的 stored generated column。

5.5 Replica Identity 选型

-- 使用主键；默认模式。
ALTER TABLE app.orders REPLICA IDENTITY DEFAULT;

-- 使用指定唯一索引。
CREATE UNIQUE INDEX CONCURRENTLY orders_replica_identity_uq
    ON app.orders (tenant_id, order_no);
ALTER TABLE app.orders
    ALTER COLUMN tenant_id SET NOT NULL,
    ALTER COLUMN order_no SET NOT NULL;
ALTER TABLE app.orders
    REPLICA IDENTITY USING INDEX orders_replica_identity_uq;

-- 发送旧行所有可用列。
ALTER TABLE app.orders REPLICA IDENTITY FULL;

-- 不为 UPDATE/DELETE 提供标识；仅 INSERT 场景才合理。
ALTER TABLE app.orders REPLICA IDENTITY NOTHING;
模式Publisher 发送的旧值Subscriber 定位方式优点主要代价/限制
DEFAULT主键列主键/兼容唯一索引最常见、消息小、查找快表必须有合适主键；主键变更要兼容
USING INDEX指定唯一索引列对应键查找可避免修改业务主键索引必须唯一、非 Partial、非 Deferrable，键列需 NOT NULL；维护额外索引
FULL旧行全部可用列全行匹配；现代版本可利用符合条件的 B-tree/Hash 索引无主键遗留表可快速接入WAL/网络放大；宽行/TOAST 代价大；无索引时目标查找昂贵；类型需有合适相等比较
NOTHING无无法可靠定位INSERT-only Publication 可用发布 UPDATE/DELETE 时相应写入会失败；不能恢复目标行

Replica Identity 的 WAL 取值规则定义在 Publisher。Subscriber 不要求使用同名约束，但当 Publisher 使用非 FULL 身份时，Subscriber 必须设置由相同或更少键列组成的兼容 Replica Identity；为了定位性能与一致性，通常配置语义等价的唯一索引。不要用 FULL 掩盖缺失主键的长期数据模型问题。

5.6 分区表与 Partition Root

CREATE PUBLICATION pub_orders_partitioned
FOR TABLE app.orders_partitioned
WITH (publish_via_partition_root = true);
  • false：按实际叶子分区的 Relation Identity 发布；Subscriber 需要能映射相应叶子关系。
  • true：将变化按分区根表的名称和列定义发布，便于 Subscriber 使用不同的分区布局或非分区表。
  • Row Filter 和 Column List 的来源会受到根/叶发布方式影响，必须用真实 DML 验证。
  • UPDATE 导致跨分区移动时，可能表现为 DELETE+INSERT；目标端唯一约束和 Row Filter 可能因此触发不同冲突。
  • TRUNCATE 在分区和外键场景中有额外风险；不要把它当成普通逐行 DELETE。

5.7 创建 Subscriber Schema 与 Subscription

逻辑复制不会自动创建 Schema。先在目标端执行经过评审的 Schema：

pg_dump --schema-only --no-owner --no-privileges \
  --table='app.orders' "$PUBLISHER_DATABASE_URL" \
  | psql "$SUBSCRIBER_DATABASE_URL"

然后在 Subscriber 上：

CREATE SUBSCRIPTION sub_orders
CONNECTION 'host=publisher.example port=5432 dbname=appdb user=repl_login password=replace-with-secret sslmode=verify-full'
PUBLICATION pub_orders
WITH (
    copy_data = true,
    create_slot = true,
    enabled = true,
    binary = false,
    streaming = parallel,
    two_phase = false,
    disable_on_error = true,
    failover = false
);

关键选项：

选项默认/版本说明
copy_datatrue是否执行初始 COPY；已有一致数据时才可安全设为 false
create_slottrue是否由 Subscription 创建主 Slot
slot_nameSubscription 名指定外部预建 Slot；NONE 仅适合禁用且不建 Slot 的配置阶段
binaryfalse可能更快，但跨版本、跨架构和类型兼容性更差
streamingPG14–17 默认 off；PG18 默认 paralleloff、on、parallel 控制大事务传输/Apply 方式
synchronous_commitoffSubscriber 崩溃后可从 Publisher 重放；同步逻辑复制场景需单独评估
two_phasefalse[PG15+] 复制 Prepared Transaction；初始同步完成后才进入启用状态
disable_on_errorfalse错误时自动禁用 Subscription，避免无限重试刷日志；需监控并人工恢复
originany环路控制时可用 none 排除带 Origin 的上游变化，但不能代替完整拓扑设计
failoverfalse[PG17+] 标记关联 Slot 可同步到物理 Standby

安全注意：默认由 CREATE SUBSCRIPTION 建 Slot 时，该命令不能放在事务块中。若 Publisher 与 Subscriber 在同一 Cluster，命令可能因“本事务等待自己创建的 Slot”而挂住，应预先创建 Slot，并使用 create_slot=false。

5.8 常用变更命令

ALTER SUBSCRIPTION sub_orders DISABLE;
ALTER SUBSCRIPTION sub_orders ENABLE;

ALTER SUBSCRIPTION sub_orders
    CONNECTION 'host=new-publisher.example port=5432 dbname=appdb user=repl_login sslmode=verify-full';

ALTER SUBSCRIPTION sub_orders
    SET (streaming = on, disable_on_error = true);

ALTER SUBSCRIPTION sub_orders
    REFRESH PUBLICATION WITH (copy_data = true);

REFRESH PUBLICATION 会发现新增/删除的表，但不会因为你改变 Row Filter 就自动重抄已处于 ready 状态的表。需要重做基线时，应显式设计临时表、重新订阅或重新 Backfill。

5.9 监控 SQL

Publisher：

SELECT
    slot_name,
    plugin,
    slot_type,
    database,
    active,
    active_pid,
    restart_lsn,
    confirmed_flush_lsn,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn))
        AS retained_from_restart,
    wal_status,
    safe_wal_size,
    invalidation_reason,
    failover,
    synced
FROM pg_replication_slots
WHERE slot_type = 'logical'
ORDER BY slot_name;
  • restart_lsn：Publisher 不能回收的最早 WAL 边界。
  • confirmed_flush_lsn：消费者已确认安全持久化的逻辑位置。
  • retained_from_restart：当前 WAL 头到保留边界的近似字节差，不等于目录实际占用。
  • wal_status：可判断 Slot 是否仍处于 reserved/extended，或已接近/进入不可用状态。
  • safe_wal_size：在限制下还能写多少 WAL；NULL 可能表示不适用或已失效。
  • invalidation_reason：一旦非空，通常需要重建 Slot/Subscription 并重新基线。
  • failover/synced：[PG17+] 用于确认 Failover Slot 属性和 Standby 同步状态。

Subscriber：

SELECT
    subname,
    worker_type,
    pid,
    leader_pid,
    relid::regclass AS relation,
    received_lsn,
    last_msg_send_time,
    last_msg_receipt_time,
    latest_end_lsn,
    latest_end_time
FROM pg_stat_subscription
ORDER BY subname, worker_type, relation NULLS FIRST;
SELECT
    s.subname,
    r.srrelid::regclass AS relation,
    r.srsubstate,
    r.srsublsn
FROM pg_subscription_rel AS r
JOIN pg_subscription AS s ON s.oid = r.srsubid
ORDER BY s.subname, relation;

[PG18] 冲突与错误统计：

SELECT
    subname,
    apply_error_count,
    sync_error_count,
    confl_insert_exists,
    confl_update_origin_differs,
    confl_update_exists,
    confl_update_missing,
    confl_delete_origin_differs,
    confl_delete_missing,
    confl_multiple_unique_conflicts,
    stats_reset
FROM pg_stat_subscription_stats
ORDER BY subname;

Replication Origin：

SELECT
    external_id,
    remote_lsn,
    local_lsn
FROM pg_replication_origin_status
ORDER BY external_id;

remote_lsn 是已应用的上游位置；local_lsn 是相应 Subscriber 本地提交位置。它们属于不同 WAL 空间，不能直接相减。

6. 底层原理

6.1 一次事务的端到端时间线

假设 Publisher 执行：

BEGIN;
UPDATE app.orders
SET status = 'paid', updated_at = clock_timestamp()
WHERE id = 1001;
INSERT INTO app.order_events(order_id, event_type)
VALUES (1001, 'paid');
COMMIT;

逻辑复制时间线如下：

Publisher Backend
  T0  获取行锁，生成新 Tuple/Index 变化
  T1  写 WAL；WAL 中保存物理/逻辑解码所需信息，而不是原 SQL 文本
  T2  COMMIT WAL Record 刷盘，业务事务对外提交

WAL Sender + Logical Decoding
  T3  从 Slot 允许的位置读取 WAL
  T4  ReorderBuffer 按 XID 收集并按事务顺序重建变化
  T5  pgoutput 发送 Relation 元数据、Begin、Update、Insert、Commit

Subscriber Apply Worker
  T6  开启本地事务，设置 replication origin session
  T7  用 Replica Identity 在 app.orders 定位行并 UPDATE
  T8  INSERT app.order_events
  T9  提交 Subscriber 本地事务并推进 Origin 进度

反馈
  T10 Subscriber/客户端发送 write/flush/apply LSN
  T11 Publisher 推进 confirmed_flush_lsn
  T12 当更早事务、Catalog Snapshot 都不再需要时推进 restart_lsn

关键点：

  • 默认异步逻辑复制下，Publisher 在 T2 提交时不等待 T9，故 RPO 取决于 Slot/Publisher 是否仍可访问，而读一致性取决于 Apply Lag。
  • 同一 Subscription 内，同一远端事务在 Subscriber 上以一个本地事务应用，事务内顺序和原子性得到保持。
  • 多个 Subscription、多个独立 CDC Sink 之间没有全局提交顺序保证。
  • Row Filter/Column List 会使“远端事务的部分变化”不可见，因此下游业务不应假设收到完整数据库不变量。

6.2 Logical Decoding、ReorderBuffer 与 Catalog Snapshot

WAL 的首要目的仍是崩溃恢复和物理复制。Logical Decoding 通过 WAL 中的 relation/tuple 相关信息与系统目录，把变化重建为逻辑消息：

  1. 为每个 XID 收集 INSERT/UPDATE/DELETE 等变化。
  2. 根据提交/中止记录决定是否输出；中止事务不会成为已提交 CDC 事件。
  3. 需要系统目录历史来解释某个 WAL 位置上的 Relation、类型和列，因此 Slot 的 catalog_xmin 可能阻止旧 Catalog Tuple 被 Vacuum 清理。
  4. 普通 xmin/catalog_xmin 长期停滞会增加表或系统目录膨胀风险。
  5. 变化超过 logical_decoding_work_mem 后，未开启合适流式协议时可能写入 Publisher 临时文件；大事务可能造成内存、临时 I/O 和尾延迟放大。

逻辑解码输出的是 提交语义下的行变化，不是 SQL：

  • UPDATE x = x + 1 不会以表达式发送，而是发送键和新值。
  • INSERT ... SELECT 会成为若干行消息。
  • Statement Trigger、执行计划、调用方身份等不在 pgoutput 行消息中。
  • DDL 不会通过内置逻辑复制自动应用。

6.3 Initial Table Sync：Snapshot、COPY、Catch-up 与交接

创建 Subscription 且 copy_data=true 后，每张表大致经历：

  1. 主 Apply Worker 发现该表需要同步，写入 pg_subscription_rel 状态。
  2. Table Sync Worker 连接 Publisher，为该表建立专用临时同步 Slot/复制上下文。
  3. 获取一致性 Snapshot，使 COPY 看到一个确定的历史集合。
  4. Subscriber 执行类似 COPY 的批量装载。
  5. COPY 期间 Publisher 仍可写入；这些增量由同步 Slot 保留。
  6. COPY 完成后，Sync Worker 从 Snapshot 对应一致点追赶到主 Apply Worker 的同步点。
  7. 状态进入 s/r，控制权交给主 Apply Worker，专用同步 Slot 被清理。

这套机制避免了“先全量、再随便找一个时间点开始增量”造成的缺口，但仍有生产边界：

  • 初始同步按表并行，不代表整个 Subscription 的所有表在同一瞬间同时可用。
  • COPY 完成后该表内容会对 Subscriber 其他会话可见；若应用过早读目标库，可能看到部分表已同步、部分未同步。
  • 初始 COPY 会触发 Subscriber 上 INSERT 的 Row Trigger 和 Statement Trigger；持续 Apply 默认 session_replication_role=replica，普通 Trigger/Rule 不触发，只触发显式配置为 REPLICA 或 ALWAYS 的相关 Row Trigger。两条路径必须分别测试。
  • Publication 的 publish 操作列表不限制历史 COPY；即使只发布 INSERT，既有行仍会被复制。
  • 初始同步错误会增加 sync_error_count，Worker 通常会重试；若根因是 Schema、权限或约束不兼容，重试不会自愈。

6.4 restart_lsn 与 confirmed_flush_lsn

用一个简化例子说明：

WAL:  A ---- B ---- C ---- D ---- E ---- F ---- G
       ^              ^                   ^
       |              |                   |
  restart_lsn   confirmed_flush_lsn   current WAL
  • confirmed_flush_lsn=C：消费者已经确认 C 之前的输出安全持久化，Slot 不再承诺重新提供更早逻辑消息。
  • restart_lsn=A：解码器仍可能需要 A 之后的原始 WAL，例如存在从 A 开始、到 E 才提交的长事务，或需要较早 Catalog 状态。
  • 因此 restart_lsn 可以明显早于 confirmed_flush_lsn。只盯 confirmed_flush_lsn 会漏掉长事务造成的 WAL 保留。
  • Slot 进度是崩溃安全的，但检查点化并不意味着每条确认都同步落盘。Publisher 崩溃恢复后可能从较早位置重放少量消息，因此消费者必须处理 Duplicate Event。

Slot WAL 保留量的容量估算应使用峰值而非平均值：

所需余量 ≈ 峰值 WAL 生成速率 × 允许的最长中断时间
         + 最大长事务覆盖的 WAL
         + Checkpoint/测量误差和安全余量

max_slot_wal_keep_size 不是实时硬限流器：它主要在 Checkpoint 时参与判断；到达限制可能让 Slot 失去所需 WAL，而不是自动让消费者“追上”。

6.5 为什么通常是 At-Least-Once

对自定义 CDC，安全顺序必须是：

解码远端事务
  -> 下游事务原子提交“事件 + 幂等记录 + LSN Checkpoint”
  -> 提交成功后发送 Standby Status Update

若先 ACK 再写 Sink：

ACK C -> Publisher 回收 C 之前 WAL -> Sink 写入失败

会造成不可恢复丢失。

若先写 Sink，提交成功后网络断开，ACK 未到 Publisher：

Sink 已有事件 -> Publisher 仍认为未确认 -> 重连后重放

会产生重复，但可由幂等键消除。因此可靠的默认语义是 At-Least-Once。

幂等键建议基于稳定的远端提交标识：

(consumer_id, transaction_end_lsn, event_index)

也可以使用业务事件 ID，但不能只用主键，因为同一主键会发生多次合法 UPDATE。对于 Kafka 等外部 Sink，“数据库 Checkpoint 与 Broker Produce”无法天然处于同一个本地事务中，需要事务性 Broker、Outbox/Inbox、幂等 Producer 或可证明的重放策略。

6.6 Large Transaction Streaming 与协议版本

无流式传输时，Logical Decoding 必须等事务提交后再整体发送；一个持续数十分钟、修改数百万行的事务会导致：

  • Publisher ReorderBuffer 占用内存或写临时文件；
  • Slot 的 restart_lsn 长期不前进；
  • 提交后形成突发网络流量与 Apply 延迟；
  • Consumer 若按事务全量内存缓存，可能 OOM。

版本演进：

协议/能力版本含义
pgoutput protocol v1基础事务提交后发送完整事务
protocol v2[PG14+]支持 Stream Start/Stop/Commit/Abort，事务提交前分段发送
protocol v3[PG15+]增加 Two-Phase 相关消息
protocol v4[PG16+]支持 Parallel Streaming/Apply 所需能力

Subscription 模式：

  • streaming=off：Publisher 完整解码，提交后整体发送。
  • streaming=on：[PG14+] 在进行中发送分段；Subscriber 先写临时文件，远端提交后再应用。
  • streaming=parallel：[PG16+] 尽量交给 Parallel Apply Worker 直接处理；无空闲 Worker 时回退到临时文件路径。PG18 默认值变为 parallel。

流式传输降低 Publisher 的峰值资源压力和提交后的突发延迟，但没有取消事务原子性。CDC Consumer 若要在远端 Commit 前向外暴露事件，就破坏了提交语义；正确做法是把分段持久化到内部 spool，在收到 Stream Commit 后才发布。

本章 Go 示例请求 proto_version '2' 与 streaming 'on'，因为当前示例使用的 pglogrepl.ParseV2 只演示 v2 消息。它不实现协议 v3/v4、Prepared Transaction 和真正的 Parallel Apply。

6.7 Two-Phase Replication

[PG15+] 当 Subscription two_phase=true 时：

Publisher PREPARE TRANSACTION
    -> Subscriber 复制为 Prepared Transaction
Publisher COMMIT PREPARED / ROLLBACK PREPARED
    -> Subscriber 执行对应完成动作

若关闭 Two-Phase，Prepared Transaction 通常要到 Publisher COMMIT PREPARED 后才以普通已提交事务发送。

生产注意：

  • Subscriber 需要为 Prepared Transaction 配置足够的 max_prepared_transactions。
  • Subscription 初始同步未完成前，内部 Two-Phase 状态保持 pending；可检查 pg_subscription.subtwophasestate。
  • Prepared Transaction 会长时间持有锁和资源；参与者故障会造成 In-Doubt 状态。
  • 目标系统若不是 PostgreSQL，CDC Consumer 不能仅靠 v2 消息声称支持 2PC。
  • 除非业务确实使用 PREPARE TRANSACTION，不要为了“更强一致性”盲目开启。

6.8 Row Filter、Column List 与 UPDATE 语义

假设过滤条件为 status = 'paid'：

旧行新行发布语义
不匹配不匹配不发送
不匹配匹配发送 INSERT 语义，使行进入集合
匹配匹配发送 UPDATE
匹配不匹配发送 DELETE 语义，使行离开集合

这意味着下游接收的是一个动态集合，而不是源表所有 UPDATE 的简单子集。

Column List 必须包含：

  • 下游正确解释事件所需的业务列；
  • UPDATE/DELETE 所需 Replica Identity 列；
  • 不可省略但无默认值的 Subscriber 必填列，或在 Subscriber 提供可接受默认值。

没有 Column List 时，未来新增普通列通常自动进入发布范围，可能让脆弱的 CDC Decoder 突然看到新列；有显式 Column List 时，新列默认不进入事件，兼容性更可控，但容易忘记扩展。两种策略都需要 Schema Registry/契约测试。

6.9 DDL 边界与 Schema Evolution

内置逻辑复制不自动复制 DDL。安全顺序遵循 Expand/Contract：

添加可空列

1. Subscriber 先 ADD COLUMN（允许 NULL 或有兼容默认值）
2. Publisher ADD COLUMN
3. 更新 Publication Column List（若显式列出）
4. ALTER SUBSCRIPTION ... REFRESH PUBLICATION（仅需要发现表变化时）
5. 发布双读/双写兼容应用
6. Backfill 历史值
7. 校验非空率与两端值
8. 最后再加 NOT NULL/删除旧列

类型变化

不要直接在 Publisher 先把 integer 改为不兼容的 jsonb。可采用：

新增 new_col -> 双写 -> Backfill -> 校验 -> 切读 -> 停旧写 -> 删除 old_col

Subscriber 可以比 Publisher 多列，只要额外列具有默认值、可空或由 Subscriber 自己填充。列按名称映射而不是只按位置盲写，但类型必须可接受发布格式。

Rename 与 Drop

Rename 在协议关系缓存和旧消费者中尤其危险。推荐新增新列而非原地 Rename；Drop 必须最后执行，并确保：

  • 所有 Subscription/CDC Consumer 已不依赖旧列；
  • 旧版本应用已完全退出；
  • 回滚窗口已结束；
  • Schema Version 事件和文档已更新。

6.10 Sequence 边界

Serial/Identity 生成出的 行值 会随 INSERT 复制，但 Sequence 对象的当前状态不会持续复制。大版本切流前必须在冻结写入后对齐：

SELECT setval(
    'app.orders_id_seq',
    GREATEST(
        COALESCE((SELECT max(id) FROM app.orders), 0),
        (SELECT last_value FROM app.orders_id_seq)
    ),
    true
);

更稳妥的做法是在 Publisher 生成每个 Sequence 的 setval 脚本，并在最终增量追平后、目标开放写入前执行。需考虑：

  • Sequence Cache 造成的空洞是正常的；不要追求连续 ID。
  • 目标 Sequence 必须高于所有已复制和潜在保留值。
  • 双写期间两端独立取号会冲突；可分配不相交区间、使用 UUID/UUIDv7，或只让一个系统拥有写权。
  • TRUNCATE ... RESTART IDENTITY 的 Sequence 副作用也不会按普通行复制自动恢复。

6.11 Subscriber Conflict

Subscriber 持续 Apply 类似普通 DML，因此本地数据、约束、权限、RLS 和 Trigger 都可能形成冲突。

[PG18] 主要冲突：

冲突行为典型根因
insert_exists报错，Apply 停止Subscriber 已有相同唯一键，或本地写入
update_exists报错，Apply 停止UPDATE 后值撞上另一个唯一键
multiple_unique_conflicts报错，Apply 停止同时违反多个非延迟唯一约束
update_missing跳过该 UPDATE，Apply 继续Subscriber 行被删、初始基线不完整、键不一致
delete_missing跳过该 DELETE，Apply 继续行已不存在
update_origin_differs记录冲突但仍应用本地行来源与远端不同；需 track_commit_timestamp 才能检测细节
delete_origin_differs记录冲突但仍应用同上

处理顺序：

  1. 先保存 Subscriber 日志、冲突 LSN、Relation、键值和当前 Origin。
  2. 判定权威：Remote-Wins、Local-Wins，还是业务合并。
  3. 优先修正 Subscriber 数据/约束后让同一事务重放。
  4. 只有确认整笔远端事务都可丢弃时才使用：
ALTER SUBSCRIPTION sub_orders SKIP (lsn = '0/14C0378');

SKIP 跳过的是整个远端事务，不只是冲突行，极易造成隐藏不一致。streaming=parallel 下失败日志可能不含 Finish LSN，可临时改为 on/off 复现以获取，但必须先评估重放副作用。

6.12 Replication Origin、环路与进度

Subscription 在 Subscriber 为上游创建 Origin。Apply 时大致执行：

setup origin session
begin local transaction
mark remote origin LSN / commit time
apply changes
commit local transaction
persist origin progress

它带来两个价值：

  1. 崩溃恢复后知道远端哪些事务已应用，避免任意重复应用。
  2. 在级联/双向拓扑中标记变化来源，可用 origin=none 请求只发送本地产生变化，降低回环风险。

但 origin=none 不是通用冲突解决器：

  • 两端同时更新同一业务行仍会形成 Last-Writer-Wins 或约束冲突。
  • Sequence、DDL、时间戳和业务不变量不会自动合并。
  • 网络分区后两端各自接收写入，会产生无法仅靠 LSN 排序解决的语义冲突。

真正的多主需要明确的写分片、单行所有权、冲突解决函数、全局 ID 和可审计合并策略；多数迁移应坚持 Single Writer。

6.13 跨版本复制与在线大版本升级

逻辑复制可跨主要版本和平台，但兼容性取决于两端共同能力：

  • 表和列名必须可映射，类型的文本输入/输出必须兼容。
  • binary=true 对跨版本不够可移植；升级链路优先 false。
  • 新版本 Publisher 的新类型、新 Generated Column 行为或更高协议版本不一定能被旧 Subscriber 接受。
  • Extension、Collation、函数、权限、RLS、Trigger、Materialized View、Large Object、Sequence 状态、定时任务和 DDL 不会由行复制完整迁移。
  • Row Filter/Column List 的初始同步行为取决于 Subscriber 版本，最老一端决定边界。

低停机升级的基本形态：

旧版本 Primary（Publisher，继续写）
      |  schema-only + initial copy + logical changes
      v
新版本 Cluster（Subscriber，只读验证）
      |
      |  短暂停写 -> 追平最终 LSN -> Sequence 对齐 -> 切连接
      v
新版本成为写权威

逻辑升级不是 pg_upgrade 的替代关系：

  • pg_upgrade 适合可接受停机、同机/共享存储或可快速搬迁文件的场景，速度通常更快。
  • 逻辑升级提供更长的在线验证窗口、数据选择和架构调整，但资源、双集群成本和操作复杂度更高。
  • [PG17+] pg_upgrade 开始支持迁移满足条件的 Logical Slot/Subscription 状态；旧 Cluster 必须至少为 17，且升级前要满足 Slot 可用、插件存在、Subscription 表状态等前置条件。不要把这一能力误套到 PG16→PG18。

7. 内部数据结构和状态

7.1 相关系统目录与视图

对象所在端关键字段/作用
pg_publicationPublisherPublication 属性、pubinsert/pubupdate/pubdelete/pubtruncate、分区根等
pg_publication_relPublisherPublication 与 Relation 的关联、Row Filter、Column List 元数据
pg_publication_tablesPublisher展开后的可读 Publication 表视图
pg_replication_slotsPublisher/StandbySlot 活跃状态、LSN、xmin/catalog_xmin、WAL 状态、Failover/Sync 状态
pg_stat_replicationPublisherWAL Sender 连接、客户端、发送/写入/刷新/回放位置
pg_subscriptionSubscriberConnection、Publication、Slot 名、Streaming、Two-Phase、Failover 等配置
pg_subscription_relSubscriber每张订阅表的 i/d/f/s/r 状态及同步 LSN
pg_stat_subscriptionSubscriberApply、Parallel Apply、Table Sync Worker 的实时状态与接收位置
pg_stat_subscription_statsSubscriberApply/Sync 错误；[PG18] 细分冲突计数
pg_replication_originSubscriberOrigin 名称与内部 ID
pg_replication_origin_statusSubscriberremote_lsn/local_lsn 进度
pg_stat_progress_copySubscriber初始 COPY 的进度，适用时可观察 bytes/tuples processed

7.2 WAL、Tuple 与 Replica Identity

  • INSERT 消息包含新 Tuple 的发布列。
  • UPDATE 通常包含新 Tuple；若 Replica Identity 键发生变化，携带旧 Key；FULL 时可携带旧 Tuple。
  • DELETE 仅需旧 Key 或旧 Tuple，不存在新值。
  • 未变化的外部 TOAST 值可能以 unchanged toast 标志出现，而不是重新发送完整大字段。CDC Consumer 不能把它误解为 NULL。
  • Column List 会改变协议 Relation 元数据和 Tuple 列集合；Decoder 必须以最近的 Relation Message 解释 Tuple。
  • Relation OID/ID 只在当前 Publisher 数据库上下文内有意义，不能作为跨重建永久 Schema ID。

7.3 Slot 状态机

PostgreSQL Logical Replication、CDC、Schema Evolution 与在线大版本升级 flow 3

active=false 不一定异常：维护窗口内短暂断开是正常的。告警应结合 inactive_since、WAL 保留字节、safe_wal_size、消费者心跳和业务 RPO，而不是只看 Boolean。

7.4 Memory Context 与临时文件

Logical Decoding 使用内部 Memory Context 管理事务变化和 Relation 元数据。大事务超过阈值后：

  • streaming=off：更可能把变化 spill 到 Publisher 临时文件，直到提交后输出。
  • streaming=on/parallel：可提前发送，但 Subscriber/Consumer 仍需在 Commit 前持久化或缓存。
  • Subscriber Parallel Apply 无 Worker 时也可能写临时文件。

监控应联合观察：

  • Publisher/Subscriber temp_bytes、temp_files；
  • pg_stat_database；
  • 进程 RSS；
  • pg_wal 和 Slot 保留；
  • 网络吞吐；
  • Apply Worker Wait Event；
  • 单事务行数和字节数。

7.5 锁与约束

  • Publisher 业务 DML 按正常方式持有行锁/表锁；逻辑解码不重新执行源端 SQL。
  • Subscriber Apply DML 会获取普通目标行锁、索引页锁和约束所需锁，因此可能被本地查询或 DDL 阻塞。
  • 初始 COPY 持有适合 COPY 的 Relation 锁，并会维护 Subscriber 索引、约束和 WAL。
  • Subscriber 上长 DDL、未提交事务、外键检查和 Replica Trigger 都可能让 Apply Lag 增长。
  • Apply 被阻塞时 Publisher 业务未必立即变慢，但 Slot WAL 增长会把延迟最终转化为磁盘风险。

8. 场景和选型决策

8.1 决策表

业务场景推荐方案不推荐方案原因性能代价并发代价一致性代价HA 代价运维复杂度
同版本整集群 HA物理流复制用逻辑复制替代物理 Standby物理复制覆盖全部对象、恢复语义完整WAL/网络同步模式增加提交延迟异步有 RPO成熟中
单表在线迁移Publication + Subscription/自定义 Backfill应用无序双写可获得一致初始点和增量追赶双端写 WAL、索引维护切流需短暂停写过滤/DDL 需人工管理Slot/旧库保留中高
PostgreSQL CDCpgoutput + 成熟 CDC 平台；教学/特殊需求可 pglogrepl轮询 updated_at 作为唯一方案WAL 能覆盖 DELETE、事务边界和低延迟解码、网络、SinkConsumer Backpressure通常 At-Least-OnceSlot Failover 要额外设计高
大版本低停机升级旧→新逻辑复制 + Freeze/Catch-up/Cutover新→旧复制或直接双写兼容路径清晰，可长时间验证双集群成本、双 WAL最终冻结写DDL/Sequence 手工回滚需反向链路高
同机快速升级pg_upgrade + 完整备份/演练逻辑复制所有表停机允许时更快、更简单短期 I/O停机窗口文件级迁移回滚仍需计划中
跨系统同步CDC + Schema Registry + 幂等 Sink把目标系统当事务同步镜像目标模型不同，需要映射和重放序列化/反序列化Queue 与连接池隔离最终一致外部 Sink HA高
双向多主明确写分片/冲突策略，或专用产品两个普通 Subscription 相互指向回环、冲突和 Sequence 无自动解决双倍复制冲突热点语义复杂脑裂高风险极高

8.2 场景一：单表迁移

示例：把 legacy.orders 迁到新 Cluster 的 app.orders，最终由新库承接读写。

阶段设计
初始同步目标先建兼容 Schema、主键和必要索引；建立单表 Publication/Subscription，copy_data=true；大表可先减少非必要二级索引，但不能牺牲 Replica Identity 和校验能力
增量同步旧库保持唯一写权威；监控 Slot、pg_subscription_rel、Apply Lag；禁止目标端业务写入
Schema 变化迁移窗口内冻结破坏性 DDL；必须变更时 Subscriber 先 Expand、Publisher 后写入新列，应用保持双版本兼容
数据校验行数、按主键范围聚合、金额总和、NULL 分布、抽样/分桶 Hash；对动态表在一致 LSN 或冻结窗口做最终校验
最终切流停止旧库写入 → 记录最终 WAL LSN → 等待 Origin/Subscription 追平 → 对齐 Sequence → 目标健康检查 → 更换写端点并排空旧连接池
回滚新库开放写入前可直接退回；开放后若没有反向 CDC，回滚会丢新库写入。可在切流前预建反向链路但禁用，或双向捕获到独立日志
Slot 清理过回滚窗口且确认无回退需求后，先 Drop Subscription/解除 Slot，再检查 Publisher pg_replication_slots 无孤儿 Slot
故障处理初始 COPY 失败修 Schema/权限后重试；Apply 冲突先恢复目标一致性；Slot 失效则重新 Backfill，不要伪造 LSN 跳过缺口

单表迁移可在 Subscriber 使用不同表名吗？内置 Subscription 要求匹配发布的 Schema/Table 名。需要改名时可：

  • 在目标建立同名中间表，切流后 Rename；
  • 使用自定义 CDC 做 Relation 映射；
  • 使用视图隔离应用逻辑名。

8.3 场景二：CDC Consumer

目标：把订单变化写入审计事件表或外部消息系统。

阶段设计
初始同步使用 Slot 导出的 Snapshot 与一致起点协调 Backfill；或先加载业务快照，再从严格记录的 LSN 消费。不能把“某次 SELECT * 结束时间”当增量起点
增量同步每个 Slot 单活消费；按远端事务 Commit 边界输出；Sink 提交成功后才 ACK；有界队列产生 Backpressure
Schema 变化Relation Message 生成 Schema Version；未知列/类型进入隔离队列；兼容 Decoder 允许新增可选列，破坏性变化先发布新事件版本
数据校验监控 LSN Lag、事件计数、事务计数、业务聚合和 Dead Letter；定期从源表与物化目标做分桶对账
最终切流CDC 通常没有数据库写权切流；若目标成为读模型，先达到目标新鲜度阈值，再逐步切读并保留源回查
回滚读流量可回源；事件消费者升级需支持旧 Schema/重新消费。保留原始事件或可重建日志
Slot 清理Consumer 永久退役前，确认无重放/审计需求；Drop Slot 后历史 WAL 不可恢复，需有独立保留层
故障处理Sink 超时视为 Commit Outcome Unknown：不 ACK，让重放由幂等键吸收；事务过大则落盘 Spool 或停止并告警，不能无限占内存

初始 Snapshot 协调的严格做法是通过复制协议创建 Slot 并导出 Snapshot，保持导出会话有效，在另一个 REPEATABLE READ 事务中 SET TRANSACTION SNAPSHOT 后执行 Backfill，再从 Slot 的一致点继续。成熟 CDC 平台通常已经实现这部分；本章 Go 示例没有实现 Snapshot Coordinator。

8.4 场景三：PostgreSQL 大版本低停机升级

示例：PG17 Publisher → PG18 Subscriber。

阶段设计
初始同步在 PG18 安装相同 Extension/Collation，导入 Schema-only；建立 Publication/Subscription；按容量安排初始 COPY 和索引构建
增量同步PG17 保持写权威；PG18 仅用于只读验证；保持 binary=false；监控所有表进入 r
Schema 变化升级冻结期禁止非兼容 DDL；必要 DDL 目标先应用；确认 PG18 Reserved Keyword、函数、Planner 和 Extension 行为
数据校验对象清单、表行数/分桶 Hash、Sequence、权限、RLS、Trigger、函数、Job、Large Object、Extension、Collation、查询计划与性能回归
最终切流应用进入维护/只读 → 终止或等待旧写事务 → 记录 pg_current_wal_lsn() → 等待 Subscription Origin 达到最终提交位置 → 同步 Sequence → Disable Subscription → 切 DNS/服务发现/Secret → 清空连接池 → PG18 开写
回滚PG18 尚未写入时立即切回 PG17；一旦 PG18 接受写入，必须有 PG18→PG17 反向复制或变更日志。反向兼容类型/DDL可能不成立，因此设定明确 Point of No Return
Slot 清理回滚窗口结束后 Drop PG18 Subscription，使其正常删除 PG17 Slot；Publisher 不可达时先 slot_name=NONE 再本地 Drop，并在旧库手工清理
故障处理Apply 冲突先停切流；目标性能不足则继续旧库写权威并调优；Slot Lost 需重建；切流中连接混写时立即 Fencing 旧写入口并审计两端增量

Final LSN 不能只看 pg_current_wal_lsn() 与 latest_end_lsn 文本相等。 应同时确认：

  • 旧库没有仍在运行/Prepared 的写事务；
  • Subscriber Apply Worker 无错误、无阻塞；
  • 对应 Replication Origin 的 remote_lsn 已达到最终远端事务位置；
  • 应用旧连接已被终止或只读化；
  • Sequence 和非表对象完成最后同步。

8.5 场景四：跨系统数据同步

示例：PostgreSQL → 搜索引擎/数据湖/键值存储。

阶段设计
初始同步从一致 Snapshot 导出历史数据；目标记录 Snapshot ID/LSN；Backfill 可并行但必须按主键分片且可重试
增量同步CDC 保存原始事务位置；转换层生成目标 Upsert/Delete；同一业务键按分区串行或使用版本号拒绝旧事件
Schema 变化建立 Source Schema Version 与 Target Mapping Version；新增字段先让目标接受，删除/改类型走双版本事件
数据校验源与目标比较文档数、按时间/租户聚合、抽样内容 Hash、删除墓碑数量和消费端高水位
最终切流目标承担查询前先达到 Lag SLO并通过 Shadow Read；分批切租户/百分比，不一次全切
回滚查询回源；保留原始 CDC Log 以重建目标；不把目标的派生数据反写源库
Slot 清理只有在上游事件已进入独立持久日志后才可删除数据库 Slot；否则删除意味着失去重放源
故障处理429/限流触发 Backpressure；毒数据进入 Dead Letter 但不能越过整笔事务悄悄 ACK；目标不可用时以磁盘 Spool 和 WAL 预算共同限流

目标系统通常没有 PostgreSQL 事务，因此只能在“单文档/单分区/单批次”级别提供幂等与顺序。不要向业务承诺数据库级 Exactly-Once，除非端到端协议可形式化证明。

8.6 Schema Evolution 兼容矩阵

变更推荐顺序是否通常在线复制风险
新增 Nullable 列Subscriber → Publisher → 应用使用是显式 Column List 需更新；旧消费者需忽略未知列
新增带常量默认列Subscriber 先建兼容列通常是版本/重写行为需按具体 DDL 验证
新增 NOT NULL 列先 Nullable/默认 → Backfill → 验证 → NOT NULL是Publisher 先写会使 Subscriber Apply 失败
扩大 varchar/数值范围Subscriber 先扩大通常是目标类型必须接受源文本/二进制格式
缩小类型范围新列迁移否/高风险目标拒绝值导致 Apply 停止
Rename 列新增别名列、双写、切换、最后删除是但复杂旧 Relation Schema、Consumer 和应用不兼容
Drop 列所有消费者停用后 Subscriber/Publisher Contract是但不可逆回滚和旧事件重放失败
修改主键/Replica Identity先建新唯一索引，两端验证，再切 Identity可在线但需谨慎切换窗口内 UPDATE/DELETE 定位和消息键变化
修改分区布局publish_via_partition_root=true 可降低映射耦合取决于方案跨分区 UPDATE、TRUNCATE、唯一约束
Generated Column明确“发送值”或“目标重算”二选一是PG18 前后行为不同，目标 Generated Column 不可直接写

9. 高性能分析

9.1 必须先建立工作负载基线

任何调参前记录：

  • 数据规模、行宽、TOAST 比例、表/索引数量；
  • 峰值 TPS、每事务行数、最大事务大小；
  • 平均与峰值 WAL 字节/秒；
  • Publisher/Subscriber CPU、内存、磁盘类型、网络带宽和 RTT；
  • 初始 COPY 并发、Subscription 数、Worker 上限；
  • P50/P95/P99 Apply Lag 与允许的恢复窗口；
  • Sink 的吞吐、批大小、限流和失败率。

9.2 CPU

Publisher CPU 消耗来自 WAL 读取、Logical Decoding、表达式 Row Filter、类型输出和协议编码。Subscriber CPU 来自协议解析、类型输入、约束、索引维护和 Trigger。以下情况放大 CPU：

  • 大量宽行使用 REPLICA IDENTITY FULL；
  • Row Filter 复杂且每行执行；
  • Subscriber 拥有过多二级索引；
  • 文本格式需要昂贵的类型转换；
  • 自定义 CDC 对每列重复 JSON 编码/反射；
  • 小事务过多导致每事务固定开销占比高。

优化顺序通常是减少发布数据、修正 Identity、批量 Sink、减少目标冗余索引，再考虑增加 Worker。盲目增加 Worker 可能把瓶颈从 CPU 移到 WAL、锁或 I/O。

9.3 内存与 shared_buffers

  • Logical Decoding 的事务缓冲不等同于 shared_buffers；logical_decoding_work_mem 和进程私有内存更直接。
  • Subscriber Apply 仍访问目标 Heap/Index Buffer；工作集进入 shared_buffers 与 OS Page Cache。
  • CDC Consumer 必须使用有界队列和事务大小上限。queue_capacity × 平均事件大小 + 最大并发事务缓存 才是实际内存预算。
  • 初始 COPY 的多个 Worker、索引构建、维护工作内存会叠加；不要按单进程参数估算总内存。

9.4 I/O、Page Cache 与 PG18 AIO

逻辑复制的主要 I/O 路径：

  1. Publisher 顺序读取 WAL；
  2. 大事务可能写/读临时 spill 文件；
  3. Subscriber 写 Heap、索引、WAL；
  4. 初始 COPY 扫描 Publisher 表并顺序写 Subscriber；
  5. Checkpoint 把 Subscriber Dirty Buffer 刷盘。

PG18 AIO 可改善部分 PostgreSQL I/O 路径，但它不会消除逻辑解码、网络、索引维护或目标约束成本。确认实际 Wait Event、pg_aios/相关监控与存储队列后再归因，不能看到 PG18 就假设复制自动更快。

9.5 网络往返与格式

持续复制是流式协议，不是每行一次 SQL 往返，但事务提交、Keepalive、Status Update 和 TCP 拥塞仍受 RTT 影响。

  • binary=true 可能减少转换成本，但跨版本/架构/类型不安全；升级默认文本。
  • TLS 有 CPU 代价，但生产不应为少量吞吐关闭传输安全。
  • 网络抖动会让 Apply Lag 和 WAL Retention 同时上升。
  • 压缩不由基础 pgoutput 自动解决；跨地域高 WAL 流量需评估带宽、专线或架构调整。

9.6 索引维护、读写放大与空间放大

每条远端 DML 在 Subscriber 再执行一次：

  • Heap 写一次；
  • 每个受影响索引再写一次；
  • Subscriber 自己生成 WAL；
  • Vacuum 处理 Subscriber 死 Tuple；
  • 物理 Standby 若存在，还会再次传输 Subscriber WAL。

因此升级期间总写放大可能是源端写 + 目标端写 + 两边副本。目标端保留全部生产索引有利于切流即用，但会拖慢初始同步。可把非必要二级索引安排到 COPY 后 CREATE INDEX CONCURRENTLY，同时评估额外扫描和磁盘峰值。

9.7 Checkpoint、WAL 与 Slot

  • Publisher Checkpoint 不只影响业务 I/O，也影响 max_slot_wal_keep_size/失活 Slot 超时检查时点。
  • Subscriber Checkpoint 过激会抬高 Apply P99；过松会增加崩溃恢复时间和内存脏页压力。
  • synchronous_commit=off 的 Subscription Apply 可降低本地提交等待；Subscriber 崩溃后远端会重发。若把 Subscription 反馈纳入 Publisher 同步提交，需重新分析持久化确认语义。

9.8 Vacuum

Publisher：Slot 的 xmin/catalog_xmin 可阻止旧 Tuple/Catalog Tuple 清理。Subscriber：持续 UPDATE/DELETE 生成死 Tuple，需要正常 Autovacuum。禁止为了“提高复制速度”关闭 Autovacuum；应控制长事务、Slot 和表级 Vacuum 参数。

9.9 吞吐与 P95/P99

复制平均 Lag 很低并不代表尾延迟安全。常见 P99 尖峰：

  • 单个超大事务提交；
  • Subscriber DDL/长事务阻塞 Apply；
  • Checkpoint I/O 峰值；
  • 唯一冲突反复重试；
  • Sink 限流触发队列满；
  • 初始 COPY 与生产流量争抢 I/O；
  • Relation Schema 变化触发 Decoder 暂停/重建缓存。

测试至少记录每秒事务数、每秒事件数、每秒字节、Commit-to-Apply 延迟分位数、队列深度、Slot retained bytes、CPU、I/O、WAL、临时文件和 Wait Event。

10. 高并发分析

10.1 四种并发不能混为一谈

指标含义
数据库连接数Backend/复制连接总数，不代表都在执行
活跃查询/事务数当前占用 CPU、锁或 I/O 的工作量
应用 goroutine 数可远高于连接数；多余请求在池或队列等待
TPS/事件吞吐单位时间完成的事务/事件，受事务大小影响
排队请求数Backpressure 是否已从数据库传到应用

CDC Reader 通常只需一个 Slot 对应一个复制连接；增加 goroutine 不能并行消费同一 Slot。并行化应放在事务后的可证明安全分区上，或由多个独立 Publication/Slot 分片，但这会失去跨 Slot 全局顺序。

10.2 MVCC 与 Snapshot

Initial COPY 使用 Snapshot，因此不阻塞大多数普通 DML，但：

  • 长时间 Snapshot 会保留旧版本并增加 Vacuum 压力；
  • 每表独立同步导致跨表视图不一定是同一全局时刻；
  • 外键关联表若不同步完成就开放读取，可能暂时看见引用不完整；
  • 自定义 Backfill 必须明确 Snapshot 与增量 LSN 的关系。

10.3 锁竞争与热点

Subscriber Apply 可被以下对象阻塞：

  • 应用误写同一热点行；
  • 长事务持有目标行锁；
  • ALTER TABLE 等 DDL 请求/持有强锁；
  • Foreign Key 查找缺少索引；
  • Replica Trigger 访问热点表；
  • 唯一索引热点页。

查 blocker：

SELECT
    a.pid AS blocked_pid,
    a.application_name AS blocked_app,
    a.wait_event_type,
    a.wait_event,
    a.query AS blocked_query,
    b.pid AS blocker_pid,
    b.application_name AS blocker_app,
    b.state AS blocker_state,
    b.xact_start AS blocker_xact_start,
    b.query AS blocker_query
FROM pg_stat_activity AS a
CROSS JOIN LATERAL unnest(pg_blocking_pids(a.pid)) AS x(blocker_pid)
JOIN pg_stat_activity AS b ON b.pid = x.blocker_pid
WHERE a.backend_type IN ('logical replication worker', 'client backend')
ORDER BY a.query_start;

backend_type/application_name 的具体显示随版本和 Worker 类型变化，排障时也应从 pg_stat_subscription.pid 反查。

10.4 事务边界与并行

Subscription 保持每个远端事务原子性。streaming=parallel 可以改善超大事务的 Apply，但仍需：

  • 足够 max_logical_replication_workers；
  • 足够 max_parallel_apply_workers_per_subscription；
  • 避免目标端相同键/索引页竞争；
  • 保持远端事务提交顺序的正确交接。

增加并行 Worker 会增加连接、内存和锁占用。max_sync_workers_per_subscription 控制初始同步并发，也不应超过磁盘和网络可承受范围。

10.5 Backpressure 与 Admission Control

正确 Backpressure 链：

Sink 变慢
 -> CDC 有界队列增长
 -> Reader 停止继续消费但仍发送已持久 LSN 的 Keepalive/Status
 -> confirmed_flush_lsn 停滞
 -> Slot retained bytes 增长
 -> 达到软阈值时降低源业务非关键写入/扩容 Sink/隔离消费者
 -> 达到硬阈值前人工决策：恢复、切换或重做基线

错误做法是无限开 goroutine 或无限内存队列，这只会把可观察延迟变成 OOM。Admission Control 应同时考虑：

  • Sink Queue 深度与处理时间；
  • Publisher 剩余 WAL 安全空间；
  • 最大允许 RPO/Lag；
  • 业务写入优先级。

10.6 幂等与重试风暴

  • Duplicate Event 是正常恢复路径，不能作为异常直接丢弃整条流水线。
  • 幂等记录必须持久化并与业务写处于同一事务。
  • 多个 Consumer 实例抢同一 Slot 只会失败或互踢，不会自动分担。
  • 故障恢复时不要让大量实例同时重连；使用指数退避、Jitter 和单实例租约。
  • Apply 冲突未解决时频繁 Enable/Disable 只会形成日志与连接风暴。

10.7 死锁

Schema 不同、Replica Trigger 或 Subscriber 本地写入可能让 Apply Worker 与业务事务形成死锁。PostgreSQL 会中止其中一个事务，Subscription 重试。排障需查看 SQLSTATE 40P01、Server Log 和锁图，而不是仅看 Lag。根治通常是统一锁顺序、移除 Apply 路径的外部副作用、禁止目标业务写和缩短 DDL/事务。

11. 高可用分析

11.1 RPO 与 RTO

模式典型 RPO典型 RTO说明
异步 Subscription，Publisher 可恢复Slot 保留范围内可为 0 数据丢失，但目标有可见延迟重连 + 重放时间Publisher 提交不等待 Subscriber
Publisher 永久丢失且无 Failover Slot最后 ACK/备份后可能丢失重建 Slot/基线Slot 状态不在新 Primary
[PG17+] 已同步 Failover Slot可延续到已同步物理位置Promotion + Fencing + 重连仍受物理复制 RPO约束
CDC Sink ACK 在提交前可能静默丢失无法从 Slot 恢复已回收事件设计错误
切流后无反向链路新库写入后无法无损回滚需人工数据合并“旧库在线”不等于回滚能力

11.2 逻辑复制不能替代 Backup/PITR

逻辑复制会快速复制误删、坏数据和应用错误；它不保留完整历史，也不覆盖所有对象。必须保留独立 Backup、WAL Archive、PITR 演练和数据校验。

11.3 [PG17+] Logical Slot Failover

一个可用拓扑：

PostgreSQL Logical Replication、CDC、Schema Evolution 与在线大版本升级 flow 4

关键配置/步骤：

  1. Subscription/Slot 使用 failover=true。
  2. Standby 配置 primary_slot_name、可用的数据库连接信息，并启用 sync_replication_slots=true。
  3. hot_standby_feedback 等配置确保 Standby 能保留 Logical Decoding 所需 Catalog 行。
  4. Primary 可用 synchronized_standby_slots 约束 Logical Sender 不超过指定 Physical Slot 的 Flush 位置，避免 Failover Slot 在 Standby 上落后到无法连续恢复。
  5. 在 Standby 的 pg_replication_slots 验证目标 Slot synced=true，而不是只验证“名字存在”。
  6. Promotion 前后 Fencing 旧 Primary；暂停 Subscription，修改 Connection 到新 Primary，再 Enable。

若 Slot 在 Standby 尚未 synced=true 就故障转移，不能假设它可用。若旧 Primary 未被隔离，两个 Publisher 可能同时提供不同历史，形成脑裂。

11.4 Planned Switchover 与 Unplanned Failover

Planned：先停止新事务、等 Physical/Logical 位置追平、确认 Slot 同步、切连接、再开放写入。RPO 可接近 0，RTO 可预测。

Unplanned：以最后同步到 Standby 的 Physical WAL 为上限。即使 Subscriber 已从旧 Primary 收到更靠后的逻辑变化，新 Primary 也可能缺少对应事务；必须防止消费者从两个分叉历史继续消费。应用要处理连接重置和 Commit Outcome Unknown。

11.5 Failback

Failback 不是把旧 Primary 直接接回：

  • 先 Fencing；
  • 使用 pg_rewind/重建物理副本恢复到新 Timeline；
  • 检查 Logical Slot 的来源和同步状态；
  • 确认 Subscription Connection 只指向当前权威；
  • 对外部 CDC Sink 验证是否出现重复或历史分叉。

11.6 同步逻辑复制

可把逻辑 WAL Sender 的反馈纳入 synchronous_standby_names，使 Publisher Commit 等待 Subscriber 的写/刷新位置，但代价是：

  • 网络 RTT 和 Subscriber 磁盘延迟进入业务 Commit P99；
  • Subscriber/网络故障可能阻塞写入；
  • 仍不自动复制 DDL、Sequence 或解决业务冲突；
  • 自定义 CDC 若报告虚假的 Flush LSN，会破坏持久性承诺。

除非业务明确要求“下游持久化后源事务才确认”，通常应保持异步并用可恢复 Slot、告警和容量保障实现可用性。

11.7 应用连接与切流

TCP 连接不能随 DNS 自动迁移。切流需要：

  • 旧入口变只读或 Fencing；
  • 关闭/轮换应用连接池；
  • 处理 DNS TTL、Service Discovery Cache 和 Secret 变更；
  • 对 read-only transaction、admin shutdown、connection reset 做分类；
  • Commit 超时视为结果未知，依靠业务幂等键查询最终状态。

11.8 数据恢复验证

任何 Failover/Cutover 后至少验证：

  • 所有关键表/分区存在且约束有效；
  • Origin/Slot LSN 关系合理；
  • 无异常 update_missing/delete_missing 增量；
  • Sequence 高于最大键；
  • 核心业务聚合与抽样 Hash 相符；
  • 应用只连接一个写权威；
  • Backup/PITR 在新权威上继续运行。

12. 三维影响矩阵

维度相关度核心收益主要风险关键指标
高性能高在线 Backfill、增量同步、按行列过滤、跨版本迁移双端 WAL/索引写放大、大事务、COPY/Checkpoint I/O、Slot 磁盘增长WAL bytes/s、Apply Lag P95/P99、CPU、I/O、temp bytes、retained WAL、events/s
高并发高源端业务可持续写入；流式大事务和并行 ApplySubscriber 锁/唯一冲突、长事务、Backpressure、重试风暴、目标误写Worker 数、blocker、queue depth、active xacts、conflict count、pool wait
高可用高低停机升级、可恢复 CDC、[PG17+] Failover SlotSlot 丢失、旧 Primary 未 Fencing、切流后无法回滚、ACK 语义错误RPO/RTO、Slot synced、Origin LSN、连接端点、Backup/PITR、校验差异

13. 实验一：Publication、Subscription 初始同步与持续复制

13.1 实验目标

验证：

  • copy_data=true 的 Initial Table Sync；
  • pg_subscription_rel 的状态变化；
  • 初始 COPY 与后续 INSERT/UPDATE/DELETE；
  • Publisher 先增加列会如何让 Apply 失败；
  • Subscriber 先补齐 Schema 后如何恢复。

13.2 版本、扩展与拓扑

  • 推荐 PostgreSQL 18；PG15+ 可完成主体实验。
  • 不需要扩展。
  • 两个不同实例：Publisher localhost:5432，Subscriber localhost:5433。
  • Publisher 已配置 wal_level=logical、足够的 Sender/Slot，并已重载 pg_hba.conf。

环境变量仅作示意：

export PUB_URL='postgres://postgres:[email protected]:5432/appdb?sslmode=disable'
export SUB_URL='postgres://postgres:[email protected]:5433/appdb?sslmode=disable'

13.3 Session A：Publisher 建表、数据与 Publication

-- 连接 Publisher
CREATE SCHEMA IF NOT EXISTS lab22;

DROP TABLE IF EXISTS lab22.orders;
CREATE TABLE lab22.orders (
    id          bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    tenant_id   bigint NOT NULL,
    order_no    text NOT NULL,
    status      text NOT NULL,
    amount      numeric(12,2) NOT NULL CHECK (amount >= 0),
    updated_at  timestamptz NOT NULL DEFAULT clock_timestamp(),
    UNIQUE (tenant_id, order_no)
);

INSERT INTO lab22.orders(tenant_id, order_no, status, amount)
SELECT
    42,
    'ORD-' || g,
    CASE WHEN g % 2 = 0 THEN 'paid' ELSE 'created' END,
    g * 10.00
FROM generate_series(1, 10000) AS g;

DROP PUBLICATION IF EXISTS lab22_pub_orders;
CREATE PUBLICATION lab22_pub_orders
FOR TABLE lab22.orders
WITH (publish = 'insert, update, delete');

SELECT count(*), min(id), max(id), sum(amount)
FROM lab22.orders;

Publication 不包含 truncate，但这不影响初始 COPY 复制现有 10000 行。

13.4 Session B：Subscriber 准备 Schema 并创建 Subscription

-- 连接 Subscriber
CREATE SCHEMA IF NOT EXISTS lab22;

DROP TABLE IF EXISTS lab22.orders;
CREATE TABLE lab22.orders (
    id          bigint PRIMARY KEY,
    tenant_id   bigint NOT NULL,
    order_no    text NOT NULL,
    status      text NOT NULL,
    amount      numeric(12,2) NOT NULL CHECK (amount >= 0),
    updated_at  timestamptz NOT NULL,
    UNIQUE (tenant_id, order_no)
);

CREATE SUBSCRIPTION 中替换真实连接参数：

DROP SUBSCRIPTION IF EXISTS lab22_sub_orders;

CREATE SUBSCRIPTION lab22_sub_orders
CONNECTION 'host=127.0.0.1 port=5432 dbname=appdb user=repl_login password=secret sslmode=disable'
PUBLICATION lab22_pub_orders
WITH (
    copy_data = true,
    binary = false,
    streaming = parallel,
    disable_on_error = true
);

CREATE SUBSCRIPTION 创建 Catalog/Slot 后即可返回，不会等待全部表 COPY 完成。

13.5 Session C：观察状态

-- 连接 Subscriber，重复执行或在 psql 中 \watch 1
SELECT
    s.subname,
    r.srrelid::regclass AS relation,
    r.srsubstate,
    r.srsublsn
FROM pg_subscription_rel AS r
JOIN pg_subscription AS s ON s.oid = r.srsubid
WHERE s.subname = 'lab22_sub_orders';
SELECT
    subname,
    worker_type,
    pid,
    relid::regclass AS relation,
    received_lsn,
    latest_end_lsn,
    last_msg_receipt_time
FROM pg_stat_subscription
WHERE subname = 'lab22_sub_orders';

初始 COPY 足够慢时可观察：

SELECT
    pid,
    datname,
    relid::regclass AS relation,
    command,
    type,
    bytes_processed,
    tuples_processed
FROM pg_stat_progress_copy;

等待点：Session C 等待 srsubstate='r'；SQL 本身不阻塞，只是轮询状态。

13.6 持续复制时间线

Session A：

BEGIN;

INSERT INTO lab22.orders(tenant_id, order_no, status, amount)
VALUES (42, 'ORD-10001', 'created', 10001.00);

UPDATE lab22.orders
SET status = 'paid', updated_at = clock_timestamp()
WHERE id = 1;

DELETE FROM lab22.orders
WHERE id = 2;

COMMIT;

Session B：

SELECT id, tenant_id, order_no, status, amount
FROM lab22.orders
WHERE id IN (1, 2, 10001)
ORDER BY id;

预期：

  • id=1 为 paid；
  • id=2 不存在；
  • 新行存在；
  • 三个变化作为一个远端事务顺序应用。

13.7 故意制造 Schema 不兼容

Session A 先改 Publisher，这是错误顺序：

ALTER TABLE lab22.orders ADD COLUMN note text;

UPDATE lab22.orders
SET note = 'publisher-only-column', updated_at = clock_timestamp()
WHERE id = 1;

该源事务在 Publisher 正常提交；异步逻辑复制不会让源事务因 Subscriber Schema 不兼容而回滚。Subscriber Apply 在接收新 Relation 定义后失败，并因 disable_on_error=true 禁用 Subscription。

Session B 诊断：

SELECT subname, subenabled
FROM pg_subscription
WHERE subname = 'lab22_sub_orders';

SELECT *
FROM pg_stat_subscription_stats
WHERE subname = 'lab22_sub_orders';

查看 Subscriber Server Log，预期看到目标 Relation 缺少发布列之类的错误。

修复必须先补 Subscriber：

ALTER TABLE lab22.orders ADD COLUMN note text;
ALTER SUBSCRIPTION lab22_sub_orders ENABLE;

验证：

SELECT id, note
FROM lab22.orders
WHERE id = 1;

远端失败事务会重放，note 最终应出现。

13.8 诊断、执行计划与统计指标

Publisher Slot：

SELECT
    slot_name,
    active,
    restart_lsn,
    confirmed_flush_lsn,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained,
    wal_status,
    safe_wal_size
FROM pg_replication_slots
WHERE slot_name = 'lab22_sub_orders';

Subscriber 验证 Replica Identity 查找路径：

EXPLAIN (
    ANALYZE,
    BUFFERS,
    WAL,
    SETTINGS,
    VERBOSE,
    SUMMARY
)
SELECT *
FROM lab22.orders
WHERE id = 1;

预期使用主键 Index Scan。该查询只是验证键查找条件，不是 Apply Worker 内部语句的逐字重放。

性能记录至少包括：

SELECT version();
SHOW shared_buffers;
SHOW max_logical_replication_workers;
SHOW max_sync_workers_per_subscription;
SHOW max_parallel_apply_workers_per_subscription;

SELECT datname, temp_files, temp_bytes, xact_commit, tup_inserted, tup_updated, tup_deleted
FROM pg_stat_database
WHERE datname = current_database();

实验不宣称固定耗时。记录行宽、10000 行规模、缓存冷热状态、COPY Worker 数、CPU、磁盘、网络、P50/P95/P99 Lag 和 Wait Event。

13.9 哪一步等待、失败和提交

步骤结果
CREATE SUBSCRIPTION连接和建 Slot 可能等待网络/锁；不等待所有初始数据完成
Session C 轮询逻辑等待状态变 r，查询本身立即返回
初始 COPYTable Sync Worker 执行，失败会重试
正常 DMLPublisher 提交；Subscriber 稍后异步提交
Publisher 先加 noteDDL 提交，不自动复制
随后的 UPDATEPublisher 提交；Subscriber Apply 因缺列失败/禁用
Subscriber 补列并 Enable原远端事务重放并提交

13.10 结果解释

本实验证明：数据流可在线追赶，但 Schema 不在数据流内。源事务和目标 Apply 是两个本地事务，默认不存在分布式原子提交。正确发布顺序应是 Subscriber Expand 在前、Publisher 使用新 Schema 在后。

13.11 清理

先在 Subscriber：

DROP SUBSCRIPTION IF EXISTS lab22_sub_orders;
DROP SCHEMA IF EXISTS lab22 CASCADE;

再在 Publisher：

DROP PUBLICATION IF EXISTS lab22_pub_orders;
DROP SCHEMA IF EXISTS lab22 CASCADE;

正常 DROP SUBSCRIPTION 会连接 Publisher 删除 Slot。若 Publisher 不可达，不要反复强制删除：先记录 Slot 名和位置，按官方流程解除 Subscription 与 Slot 关联，再在 Publisher 手工 pg_drop_replication_slot()。

13.12 生产安全警告

  • 不要在业务同名 Schema 上直接运行实验。
  • 初始 COPY 会读全表、写目标 WAL 并维护索引；先评估 I/O 和磁盘。
  • 不要在同一 Cluster 内直接使用默认 create_slot=true 建回环 Subscription，可能挂住。
  • 不要为了实验关闭 fsync、full_page_writes、Autovacuum 或数据校验。

14. 实验二：Subscriber 冲突与恢复

14.1 实验目标

复现并区分：

  1. insert_exists：唯一键冲突，Apply 停止；
  2. update_missing：目标行缺失，UPDATE 被跳过但 Apply 继续；
  3. 数据修复优先于 SKIP；
  4. [PG18] pg_stat_subscription_stats 冲突计数。

14.2 版本、扩展与配置

  • 完整冲突分类/统计要求 PostgreSQL 18。
  • 不需要扩展。
  • 可选在 Subscriber 开启 track_commit_timestamp=on 以获得更丰富的来源/提交时间细节；该参数需要重启并有额外开销，不应只为临时实验盲目在生产开启。

14.3 Session A：Publisher

CREATE SCHEMA IF NOT EXISTS lab22_conflict;
DROP TABLE IF EXISTS lab22_conflict.items;

CREATE TABLE lab22_conflict.items (
    id          bigint PRIMARY KEY,
    sku         text NOT NULL UNIQUE,
    quantity    integer NOT NULL CHECK (quantity >= 0),
    updated_at  timestamptz NOT NULL DEFAULT clock_timestamp()
);

DROP PUBLICATION IF EXISTS lab22_conflict_pub;
CREATE PUBLICATION lab22_conflict_pub
FOR TABLE lab22_conflict.items;

14.4 Session B：Subscriber

CREATE SCHEMA IF NOT EXISTS lab22_conflict;
DROP TABLE IF EXISTS lab22_conflict.items;

CREATE TABLE lab22_conflict.items (
    id          bigint PRIMARY KEY,
    sku         text NOT NULL UNIQUE,
    quantity    integer NOT NULL CHECK (quantity >= 0),
    updated_at  timestamptz NOT NULL
);

DROP SUBSCRIPTION IF EXISTS lab22_conflict_sub;
CREATE SUBSCRIPTION lab22_conflict_sub
CONNECTION 'host=127.0.0.1 port=5432 dbname=appdb user=repl_login password=secret sslmode=disable'
PUBLICATION lab22_conflict_pub
WITH (
    copy_data = false,
    streaming = on,
    disable_on_error = true
);

copy_data=false 只在本实验的空表基线中安全。生产中若目标不是已验证的一致快照，设置它会制造 Missing Row。

14.5 冲突一：insert_exists

Session B 先制造不应存在的本地写：

INSERT INTO lab22_conflict.items(id, sku, quantity, updated_at)
VALUES (1, 'LOCAL-SKU', 99, clock_timestamp());
COMMIT;

Session A：

INSERT INTO lab22_conflict.items(id, sku, quantity)
VALUES (1, 'REMOTE-SKU', 10);
COMMIT;

提交结果：Publisher INSERT 正常提交。Subscriber 重放 INSERT 时违反主键，产生 insert_exists，Apply 报错并因配置被禁用。

Session C 在 Subscriber 诊断：

SELECT subname, subenabled
FROM pg_subscription
WHERE subname = 'lab22_conflict_sub';

SELECT
    subname,
    apply_error_count,
    confl_insert_exists,
    confl_update_exists,
    confl_update_missing,
    stats_reset
FROM pg_stat_subscription_stats
WHERE subname = 'lab22_conflict_sub';

SELECT *
FROM lab22_conflict.items
WHERE id = 1;

查看 Server Log，记录：

  • Relation；
  • Conflict Type；
  • Key；
  • Existing Local Row；
  • Remote Row；
  • Replication Origin；
  • 远端事务 Finish LSN（可用时）。

14.6 恢复：Remote-Wins

先决定权威。本实验选择 Remote-Wins：

-- Subscriber
DELETE FROM lab22_conflict.items
WHERE id = 1;

ALTER SUBSCRIPTION lab22_conflict_sub ENABLE;

轮询：

SELECT *
FROM lab22_conflict.items
WHERE id = 1;

预期远端行被重放并出现。不要先执行 SKIP：它会丢弃该远端事务中的所有变化。

14.7 冲突二：update_missing

Session A：

INSERT INTO lab22_conflict.items(id, sku, quantity)
VALUES (2, 'REMOTE-2', 20);
COMMIT;

等 Session B 确认 id=2 已出现：

SELECT * FROM lab22_conflict.items WHERE id = 2;

Session B 本地误删：

DELETE FROM lab22_conflict.items WHERE id = 2;
COMMIT;

Session A 更新：

UPDATE lab22_conflict.items
SET quantity = 21, updated_at = clock_timestamp()
WHERE id = 2;
COMMIT;

INSERT INTO lab22_conflict.items(id, sku, quantity)
VALUES (3, 'REMOTE-3', 30);
COMMIT;

Session B：

SELECT *
FROM lab22_conflict.items
WHERE id IN (2, 3)
ORDER BY id;

预期：

  • id=2 仍不存在；远端 UPDATE 被跳过；
  • id=3 存在，证明 Apply 没有因 update_missing 停止；
  • confl_update_missing 增加。
SELECT
    confl_update_missing,
    confl_delete_missing,
    apply_error_count
FROM pg_stat_subscription_stats
WHERE subname = 'lab22_conflict_sub';

这类“继续复制”比停止更隐蔽：流水线绿色并不代表数据一致。

14.8 SKIP 的高风险演示说明

发生错误型冲突时，日志可能给出 Finish LSN：

ALTER SUBSCRIPTION lab22_conflict_sub
SKIP (lsn = '0/REPLACE_WITH_FINISH_LSN');

仅在以下条件全部成立时考虑：

  1. 已保留冲突事务完整审计；
  2. 已确认事务内每个变化都可以丢弃或已人工补偿；
  3. 有切片校验计划；
  4. 负责人批准并记录变更单。

streaming=parallel 下失败事务 Finish LSN 可能不在日志中；实验使用 streaming=on 以便诊断。不要通过猜 LSN 或随意调用 pg_replication_origin_advance() 越过未知范围。

14.9 Blocker、Identity 与执行计划诊断

SELECT
    a.pid AS blocked_pid,
    a.wait_event_type,
    a.wait_event,
    b.pid AS blocker_pid,
    b.xact_start,
    b.query
FROM pg_stat_activity AS a
CROSS JOIN LATERAL unnest(pg_blocking_pids(a.pid)) AS x(pid)
JOIN pg_stat_activity AS b ON b.pid = x.pid
WHERE a.pid IN (
    SELECT pid
    FROM pg_stat_subscription
    WHERE subname = 'lab22_conflict_sub'
);

验证主键查找：

EXPLAIN (ANALYZE, BUFFERS, WAL, SETTINGS, VERBOSE, SUMMARY)
SELECT *
FROM lab22_conflict.items
WHERE id = 2;

若 Subscriber 缺少等价索引，REPLICA IDENTITY FULL 或高频 UPDATE/DELETE 可能形成全表/高代价查找。实际 Apply 查询不通过普通 pg_stat_statements 总能完整还原，必要时结合 auto_explain、系统日志和实验性等价查询定位最早的估算错误。

14.10 明确时间线

时间操作等待/失败/提交
T0Subscriber 本地插入 id=1本地提交
T1Publisher 插入 id=1源提交
T2Apply 插入 id=1唯一冲突失败，Subscription 禁用
T3Subscriber 删除本地行、Enable远端事务重放并提交
T4Publisher 插入 id=2两端最终提交
T5Subscriber 本地删除 id=2本地提交，形成分叉
T6Publisher 更新 id=2源提交；目标 Missing Row，跳过但继续
T7Publisher 插入 id=3两端提交，证明流水线仍运行

14.11 清理

Subscriber：

DROP SUBSCRIPTION IF EXISTS lab22_conflict_sub;
DROP SCHEMA IF EXISTS lab22_conflict CASCADE;

Publisher：

DROP PUBLICATION IF EXISTS lab22_conflict_pub;
DROP SCHEMA IF EXISTS lab22_conflict CASCADE;

14.12 生产安全警告

  • Subscription 目标表应默认禁止业务写入。
  • update_missing/delete_missing 不停止复制，必须对冲突计数设告警。
  • SKIP 和 Origin Advance 都是数据手术，不是普通重试按钮。
  • 唯一冲突修复时先确认权威，避免删除正确的本地业务数据。

15. 实验三：模拟 PostgreSQL 大版本切流

15.1 实验目标

模拟 PG17 → PG18：

  • Schema-only 迁移；
  • Initial Copy + 持续写入；
  • 最终写冻结与增量追平；
  • 数据、对象与 Sequence 校验；
  • Cutover、Point of No Return 与 Rollback 条件。

15.2 版本、扩展与拓扑

  • Publisher：PostgreSQL 17，端口 5432。
  • Subscriber：PostgreSQL 18，端口 5433。
  • 无必要扩展；若生产库使用 Extension，必须在目标安装兼容版本后再导入 Schema。
  • 使用较新版本的 pg_dump 客户端读取旧 Server，仍需先在演练环境验证。
export OLD_URL='postgres://postgres:[email protected]:5432/appdb?sslmode=disable'
export NEW_URL='postgres://postgres:[email protected]:5433/appdb?sslmode=disable'

15.3 Session A：旧版本 Publisher 准备数据

CREATE SCHEMA IF NOT EXISTS lab22_upgrade;

DROP TABLE IF EXISTS lab22_upgrade.cutover_marker;
DROP TABLE IF EXISTS lab22_upgrade.orders;

CREATE TABLE lab22_upgrade.orders (
    id           bigint GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
    customer_id  bigint NOT NULL,
    amount       numeric(12,2) NOT NULL CHECK (amount >= 0),
    status       text NOT NULL,
    updated_at   timestamptz NOT NULL DEFAULT clock_timestamp()
);

CREATE TABLE lab22_upgrade.cutover_marker (
    marker_id    text PRIMARY KEY,
    created_at   timestamptz NOT NULL DEFAULT clock_timestamp()
);

INSERT INTO lab22_upgrade.orders(customer_id, amount, status)
SELECT
    1 + (g % 1000),
    (g % 10000) / 10.0,
    CASE WHEN g % 3 = 0 THEN 'paid' ELSE 'created' END
FROM generate_series(1, 100000) AS g;

DROP PUBLICATION IF EXISTS lab22_upgrade_pub;
CREATE PUBLICATION lab22_upgrade_pub
FOR TABLE lab22_upgrade.orders, lab22_upgrade.cutover_marker;

15.4 导出并导入 Schema-only

Shell：

pg_dump --schema-only --no-owner --no-privileges \
  --table='lab22_upgrade.orders' \
  --table='lab22_upgrade.cutover_marker' \
  "$OLD_URL" > /tmp/lab22_upgrade_schema.sql

psql "$NEW_URL" -f /tmp/lab22_upgrade_schema.sql

在目标检查：

SELECT version();
\d+ lab22_upgrade.orders
\d+ lab22_upgrade.cutover_marker

生产迁移还需单独清单：Role、Grant、Default Privilege、RLS、Function、Trigger、Extension、Collation、Job、FDW、Large Object、Materialized View、Publication、Backup 配置。不要只比较表数量。

15.5 Session B：PG18 创建 Subscription

CREATE SUBSCRIPTION lab22_upgrade_sub
CONNECTION 'host=127.0.0.1 port=5432 dbname=appdb user=repl_login password=secret sslmode=disable'
PUBLICATION lab22_upgrade_pub
WITH (
    copy_data = true,
    binary = false,
    streaming = parallel,
    disable_on_error = true
);

轮询到两张表均 r：

SELECT
    r.srrelid::regclass AS relation,
    r.srsubstate,
    r.srsublsn
FROM pg_subscription_rel AS r
JOIN pg_subscription AS s ON s.oid = r.srsubid
WHERE s.subname = 'lab22_upgrade_sub'
ORDER BY relation;

15.6 Session A：在旧库持续写入

可手工多次执行：

INSERT INTO lab22_upgrade.orders(customer_id, amount, status)
SELECT
    1 + (random() * 999)::bigint,
    round((random() * 1000)::numeric, 2),
    'created'
FROM generate_series(1, 100);

UPDATE lab22_upgrade.orders
SET status = 'paid', updated_at = clock_timestamp()
WHERE id IN (
    SELECT id
    FROM lab22_upgrade.orders
    WHERE status = 'created'
    ORDER BY id
    LIMIT 20
);
COMMIT;

或使用有明确结束时间的 pgbench 自定义脚本；不要无限后台循环而忘记停止。

15.7 Session C：迁移期间监控

Publisher：

SELECT
    slot_name,
    active,
    restart_lsn,
    confirmed_flush_lsn,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained,
    wal_status,
    safe_wal_size
FROM pg_replication_slots
WHERE slot_name = 'lab22_upgrade_sub';

Subscriber：

SELECT
    subname,
    worker_type,
    pid,
    received_lsn,
    latest_end_lsn,
    now() - last_msg_receipt_time AS receive_idle
FROM pg_stat_subscription
WHERE subname = 'lab22_upgrade_sub';
SELECT external_id, remote_lsn, local_lsn
FROM pg_replication_origin_status;

15.8 在线预校验

两端分别执行：

SELECT
    count(*) AS rows,
    min(id) AS min_id,
    max(id) AS max_id,
    sum(amount) AS amount_sum,
    count(*) FILTER (WHERE status = 'paid') AS paid_rows
FROM lab22_upgrade.orders;

在线阶段结果可能因 Lag 暂时不同。小型实验可在写冻结后做确定性 Hash：

SELECT md5(
    string_agg(
        concat_ws('|', id, customer_id, amount, status, updated_at),
        E'\n' ORDER BY id
    )
) AS table_hash
FROM lab22_upgrade.orders;

该全表 string_agg 不适合生产大表。生产应按主键范围/租户/日期分桶，比较 count/sum/min/max 和可控大小的 Hash，记录 Snapshot/LSN，并支持重跑差异桶。

15.9 最终 Cutover 时间线

步骤 1：冻结旧写入

停止写入进程，撤销旧 Writer 入口并终止旧连接。实验中至少确认：

-- Publisher
SELECT pid, usename, application_name, state, xact_start, query
FROM pg_stat_activity
WHERE datname = current_database()
  AND pid <> pg_backend_pid()
  AND (xact_start IS NOT NULL OR state = 'active');

生产中可对应用角色设置只读/撤销写权限并刷新连接池；Superuser 不受很多保护，仍需运维 Fencing。

步骤 2：写入最终 Marker 并记录 WAL

-- Publisher，确保这是冻结后的最后一个业务事务
INSERT INTO lab22_upgrade.cutover_marker(marker_id)
VALUES ('cutover-2026-06-21T00:00:00Z');
COMMIT;

SELECT pg_current_wal_flush_lsn() AS final_flush_lsn;

记录返回值，例如 0/5000ABC。

步骤 3：等待目标看到 Marker 并追平

-- Subscriber
SELECT *
FROM lab22_upgrade.cutover_marker
WHERE marker_id = 'cutover-2026-06-21T00:00:00Z';
SELECT
    subname,
    received_lsn,
    latest_end_lsn,
    last_msg_receipt_time
FROM pg_stat_subscription
WHERE subname = 'lab22_upgrade_sub';

Publisher 同时确认 Slot：

SELECT
    confirmed_flush_lsn,
    pg_wal_lsn_diff(confirmed_flush_lsn, '0/5000ABC'::pg_lsn) AS bytes_past_gate
FROM pg_replication_slots
WHERE slot_name = 'lab22_upgrade_sub';

最终 Marker 的出现证明它之前同一复制流中的事务已按序应用；LSN 检查证明反馈已越过 Gate。不要单看“Lag 显示 0”。

步骤 4：最终数据校验

  • 两端执行分桶行数/聚合/Hash；
  • pg_stat_subscription_stats 无新增错误/冲突；
  • 所有 pg_subscription_rel 为 r；
  • 无 Apply Blocker；
  • 对象清单和权限通过；
  • 核心查询在 PG18 完成 EXPLAIN (ANALYZE, BUFFERS, WAL, SETTINGS, VERBOSE, SUMMARY) 回归。只对只读查询直接使用；DML 的 EXPLAIN ANALYZE 会真实执行。

步骤 5：对齐 Sequence

Subscriber：

WITH m AS (
    SELECT max(id) AS max_id
    FROM lab22_upgrade.orders
)
SELECT setval(
    'lab22_upgrade.orders_id_seq',
    COALESCE(max_id, 1),
    max_id IS NOT NULL
)
FROM m;

验证下一值在事务回滚中也可能消耗，因此不要用“连续无空洞”作为正确性标准：

SELECT last_value, is_called
FROM lab22_upgrade.orders_id_seq;

步骤 6：停 Subscription 并切连接

-- Subscriber
ALTER SUBSCRIPTION lab22_upgrade_sub DISABLE;

应用连接从 PG17 切到 PG18，关闭旧连接池并建立新池。先执行 Smoke Test：

INSERT INTO lab22_upgrade.orders(customer_id, amount, status)
VALUES (999999, 1.00, 'cutover-smoke')
RETURNING id;

确认返回 ID 大于迁移前最大值。

15.10 Rollback 边界

时点是否可直接切回旧库条件
PG18 尚未开放写是旧库仍被 Fencing 保护，连接切回即可
PG18 只做只读 Smoke Test通常是无新权威数据
PG18 已接受业务写否必须把 PG18 新增变化复制/合并回 PG17
PG18 使用旧版本不支持的类型/DDL极难反向逻辑复制可能不兼容

生产方案应定义 Point of No Return。可在切流前建立 PG18→PG17 反向 Publication/Subscription 并保持禁用，验证兼容性后在需要时启用；但双向链路必须处理 Origin、冲突和 Fencing，不能让两端同时开放写入。

15.11 哪一步等待、失败和提交

步骤行为
Schema-only 导入目标 DDL 提交；Extension/类型缺失会失败
Initial COPY异步执行；C 轮询等待所有表 r
旧库持续写源正常提交，目标异步追赶
写冻结等待应用停流和旧事务结束；这是主要短停窗口起点
Marker源提交；目标出现后证明顺序 Gate 已通过
Sequence 对齐目标本地提交；逻辑复制不会替你完成
Disable Subscription停止后续 Apply；确认旧库已无写，否则会积压 Slot
PG18 Smoke Write新库本地提交；从此直接回滚不再安全

15.12 清理与 Slot 保留窗口

不要切流后立即删旧库 Slot。先保留明确的回滚窗口，并监控 WAL 空间。窗口结束后：

Subscriber：

DROP SUBSCRIPTION IF EXISTS lab22_upgrade_sub;
DROP SCHEMA IF EXISTS lab22_upgrade CASCADE;

Publisher：

DROP PUBLICATION IF EXISTS lab22_upgrade_pub;
DROP SCHEMA IF EXISTS lab22_upgrade CASCADE;

确认：

SELECT *
FROM pg_replication_slots
WHERE slot_name = 'lab22_upgrade_sub';

应无残留。若保留旧 Cluster 作审计，转为只读并继续 Backup/PITR，而不是让它维持不受监控的孤儿 Slot。

15.13 生产安全警告

  • 演练必须使用与生产相近的数据量、WAL 速率、Extension、Collation 和网络。
  • 低停机不等于零风险；切流前必须有恢复点和明确 Go/No-Go Gate。
  • 不要用 DNS 变更代替旧写入口 Fencing；旧 TCP 连接仍可继续写。
  • 不要在未对齐 Sequence 时开放新库写入。
  • 逻辑复制不能迁移所有对象；对象清单缺失比行数据差异更常导致上线失败。

16. Go 实战：基于 pglogrepl 的简化 CDC Consumer

16.1 目标、保证与明确边界

下面的程序直接使用 PostgreSQL Replication Protocol、pgoutput、pglogrepl、pgx/v5 和 pgxpool，用于验证一条最小但闭环的 CDC 可靠性链路：

PostgreSQL Slot
    -> pgoutput protocol v2
    -> bounded in-memory queue
    -> sink transaction: schema + events + checkpoint
    -> commit succeeds
    -> Standby Status Update confirms durable LSN

它实现：

  • IDENTIFY_SYSTEM、START_REPLICATION、Primary Keepalive 处理；
  • 周期性 StandbyStatusUpdate，且只确认目标端已经持久化的 LSN；
  • pgoutput Protocol Version 2 与 [PG14+] Large Transaction Streaming；
  • 按 XID 聚合事务，直到普通 COMMIT 或 STREAM COMMIT 后才向下游暴露；
  • 有界队列和 Backpressure；队列满时不丢数据，也不虚假推进 LSN；
  • 目标端事件、Schema Version 与 LSN Checkpoint 在同一 PostgreSQL 事务中提交；
  • 用 (consumer_id, transaction_end_lsn, event_index) 做幂等键；
  • 处理“目标端已提交、源端尚未收到 ACK”的 Commit Outcome Unknown：重放时由幂等键吸收重复；
  • Relation 元数据哈希形成 schema_version，事件携带版本标识；
  • 对单事务内存设置硬上限，超过上限就停止且不 ACK，避免静默 OOM；
  • 对目标端 Serialization Failure 和 Deadlock 做有限重试。

它不是完整生产 CDC 平台，没有实现：

  1. 与 Slot 一致点绑定的 Initial Snapshot/Backfill 协调器；
  2. 可恢复的本地磁盘 Spool、对象存储缓冲或 Kafka Transaction；
  3. pgoutput Protocol Version 3/4、Prepared Transaction 或 Parallel Streaming；
  4. 完整 PostgreSQL Type OID Registry、类型修饰符、Domain、Array、Composite、Range 和时区语义解码；
  5. DDL 捕获、Schema Registry 审批、自动重建投影与自动 Resnapshot；
  6. Logical Slot Failover、租约、单消费者 Fencing 和跨机房自动接管；
  7. 外部系统的 Exactly-Once。程序只在“同一个 PostgreSQL Sink 事务”内获得原子性，并整体提供 At-Least-Once。
  8. 对幂等键冲突后的 Payload 等值校验；生产实现应验证重复事件内容一致，防止错误复用 consumer_id 或源谱系时静默吞掉不同事件。

这里请求 proto_version '2'，是因为当前示例使用 pglogrepl.ParseV2 的公开解析接口。生产程序若要启用 Protocol Version 3/4，必须同步实现相应消息、Prepared Transaction 状态和 Parallel Streaming 语义，不能只修改一个参数。

16.2 源端准备

先创建测试表和 Publication。已有业务表应优先保留主键；无主键又需要 UPDATE/DELETE 时，必须显式评估 REPLICA IDENTITY FULL 的 WAL 与目标端查找代价。

CREATE SCHEMA IF NOT EXISTS lab22_cdc;

CREATE TABLE lab22_cdc.account (
    account_id bigint PRIMARY KEY,
    owner_name text NOT NULL,
    balance numeric(18, 2) NOT NULL,
    updated_at timestamptz NOT NULL DEFAULT clock_timestamp()
);

CREATE PUBLICATION lab22_cdc_pub
FOR TABLE lab22_cdc.account;

CREATE ROLE lab22_cdc_user
LOGIN REPLICATION PASSWORD 'replace-with-secret';

GRANT CONNECT ON DATABASE appdb TO lab22_cdc_user;
GRANT USAGE ON SCHEMA lab22_cdc TO lab22_cdc_user;
GRANT SELECT ON lab22_cdc.account TO lab22_cdc_user;

为教学程序预先创建 Slot，并记录返回的 Consistent Point：

SELECT slot_name, lsn
FROM pg_create_logical_replication_slot(
    'lab22_cdc_slot',
    'pgoutput'
);

第一次运行时，把返回的 lsn 作为 START_LSN。后续运行必须优先读取 Sink 中的持久化 Checkpoint，而不是人工改写起点。

重要边界：上述 SQL 只建立“从该一致点开始的增量流”，不会把 Slot 之前的历史行送给程序。若目标端需要完整当前状态，必须使用能够在 Slot 一致点导出 Snapshot 的 Backfill 协调流程：先固定 Slot/Snapshot 边界，再用该 Snapshot 扫描基线，最后从对应 LSN 追增量。不要先随意 pg_dump、过一段时间再创建 Slot；两者之间的变化会永久形成缺口。

检查 Slot：

SELECT slot_name,
       plugin,
       slot_type,
       active,
       restart_lsn,
       confirmed_flush_lsn,
       wal_status,
       safe_wal_size,
       invalidation_reason
FROM pg_replication_slots
WHERE slot_name = 'lab22_cdc_slot';

16.3 依赖与运行

创建项目并使用与本机 Go Toolchain 兼容的当前版本依赖；生产仓库应在验证后把具体版本写入 go.mod/go.sum：

mkdir lab22-cdc && cd lab22-cdc
go mod init example.com/lab22-cdc
go get github.com/jackc/pglogrepl github.com/jackc/pgx/v5

Replication Connection 必须带 replication=database：

export DATABASE_URL='postgres://lab22_cdc_user:replace-with-secret@publisher:5432/appdb?replication=database&sslmode=require'
export SINK_DATABASE_URL='postgres://sink_user:replace-with-secret@sink:5432/cdcdb?sslmode=require'
export PUBLICATION='lab22_cdc_pub'
export SLOT_NAME='lab22_cdc_slot'
export CONSUMER_ID='account-search-index-v1'
export START_LSN='0/16B6A90'       # 仅第一次；替换为建 Slot 返回值
export QUEUE_CAPACITY='128'
export MAX_TX_BYTES='67108864'     # 64 MiB，教学上限

go run .

Sink 用户需要创建 cdc_meta Schema 与表的权限。程序会自动建立：

  • cdc_meta.event：不可变事件与幂等键；
  • cdc_meta.schema_version：Relation 消息的结构哈希及定义；
  • cdc_meta.checkpoint：每个 consumer_id 的最后持久化 transaction_end_lsn。

16.4 完整代码

package main

import (
	"context"
	"crypto/sha256"
	"encoding/base64"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"math/rand"
	"os"
	"os/signal"
	"regexp"
	"strconv"
	"sync"
	"syscall"
	"time"

	"github.com/jackc/pglogrepl"
	"github.com/jackc/pgx/v5"
	"github.com/jackc/pgx/v5/pgconn"
	"github.com/jackc/pgx/v5/pgproto3"
	"github.com/jackc/pgx/v5/pgxpool"
)

const (
	statusInterval   = 10 * time.Second
	receiveTimeout   = 1 * time.Second
	sinkQueryTimeout = 15 * time.Second
	maxSinkRetries   = 5
)

type config struct {
	sourceURL   string
	sinkURL     string
	publication string
	slotName    string
	consumerID  string
	startLSN    string
	queueCap    int
	maxTxBytes  int
}

type columnMeta struct {
	Name         string `json:"name"`
	DataTypeOID  uint32 `json:"data_type_oid"`
	TypeModifier int32  `json:"type_modifier"`
	Key          bool   `json:"key"`
}

type relationMeta struct {
	RelationID      uint32          `json:"relation_id"`
	Namespace       string          `json:"namespace"`
	Name            string          `json:"name"`
	ReplicaIdentity uint8           `json:"replica_identity"`
	Columns         []columnMeta    `json:"columns"`
	Version         string          `json:"-"`
	Definition      json.RawMessage `json:"-"`
}

type schemaRecord struct {
	Version      string
	RelationName string
	Definition   json.RawMessage
}

type changeEvent struct {
	Operation     string         `json:"operation"`
	RelationName  string         `json:"relation"`
	SchemaVersion string         `json:"schema_version"`
	Old           map[string]any `json:"old,omitempty"`
	New           map[string]any `json:"new,omitempty"`
	Metadata      map[string]any `json:"metadata,omitempty"`
}

type transactionBuffer struct {
	XID     uint32
	Events  []changeEvent
	Schemas map[string]schemaRecord
	Bytes   int
}

type transactionEnvelope struct {
	XID        uint32
	CommitLSN  pglogrepl.LSN
	EndLSN     pglogrepl.LSN
	CommitTime time.Time
	Events     []changeEvent
	Schemas    map[string]schemaRecord
}

type durableAck struct {
	EndLSN pglogrepl.LSN
}

type sinkWorkerState struct {
	done chan struct{}
	mu   sync.RWMutex
	err  error
}

func newSinkWorkerState() *sinkWorkerState {
	return &sinkWorkerState{done: make(chan struct{})}
}

func (s *sinkWorkerState) finish(err error) {
	s.mu.Lock()
	s.err = err
	s.mu.Unlock()
	close(s.done)
}

func (s *sinkWorkerState) Err() error {
	s.mu.RLock()
	defer s.mu.RUnlock()
	return s.err
}

var (
	publicationRE = regexp.MustCompile(`^[A-Za-z_][A-Za-z0-9_]*$`)
	slotRE        = regexp.MustCompile(`^[a-z0-9_]+$`)
)

func main() {
	ctx, stop := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	defer stop()

	cfg, err := loadConfig()
	if err != nil {
		log.Fatal(err)
	}

	if err := run(ctx, cfg); err != nil && !errors.Is(err, context.Canceled) {
		log.Fatalf("cdc stopped: %v", err)
	}
}

func loadConfig() (config, error) {
	cfg := config{
		sourceURL:   os.Getenv("DATABASE_URL"),
		sinkURL:     os.Getenv("SINK_DATABASE_URL"),
		publication: os.Getenv("PUBLICATION"),
		slotName:    os.Getenv("SLOT_NAME"),
		consumerID:  os.Getenv("CONSUMER_ID"),
		startLSN:    os.Getenv("START_LSN"),
		queueCap:    envInt("QUEUE_CAPACITY", 128),
		maxTxBytes:  envInt("MAX_TX_BYTES", 64<<20),
	}

	if cfg.sourceURL == "" || cfg.sinkURL == "" || cfg.publication == "" ||
		cfg.slotName == "" || cfg.consumerID == "" {
		return config{}, errors.New("DATABASE_URL, SINK_DATABASE_URL, PUBLICATION, SLOT_NAME and CONSUMER_ID are required")
	}
	if !publicationRE.MatchString(cfg.publication) {
		return config{}, fmt.Errorf("PUBLICATION must be a simple unquoted identifier: %q", cfg.publication)
	}
	if !slotRE.MatchString(cfg.slotName) {
		return config{}, fmt.Errorf("SLOT_NAME must match [a-z0-9_]+: %q", cfg.slotName)
	}
	if cfg.queueCap < 1 || cfg.maxTxBytes < 1 {
		return config{}, errors.New("QUEUE_CAPACITY and MAX_TX_BYTES must be positive")
	}
	return cfg, nil
}

func envInt(name string, fallback int) int {
	value := os.Getenv(name)
	if value == "" {
		return fallback
	}
	n, err := strconv.Atoi(value)
	if err != nil || n <= 0 {
		log.Fatalf("%s must be a positive integer", name)
	}
	return n
}

func run(ctx context.Context, cfg config) error {
	poolConfig, err := pgxpool.ParseConfig(cfg.sinkURL)
	if err != nil {
		return fmt.Errorf("parse sink URL: %w", err)
	}
	poolConfig.MaxConns = 4
	poolConfig.MinConns = 1
	poolConfig.MaxConnLifetime = 30 * time.Minute

	sinkPool, err := pgxpool.NewWithConfig(ctx, poolConfig)
	if err != nil {
		return fmt.Errorf("connect sink: %w", err)
	}
	defer sinkPool.Close()

	if err := ensureSinkSchema(ctx, sinkPool); err != nil {
		return err
	}

	startLSN, err := loadStartLSN(ctx, sinkPool, cfg)
	if err != nil {
		return err
	}

	replConn, err := pgconn.Connect(ctx, cfg.sourceURL)
	if err != nil {
		return fmt.Errorf("connect replication source: %w", err)
	}
	defer replConn.Close(context.Background())

	system, err := pglogrepl.IdentifySystem(ctx, replConn)
	if err != nil {
		return fmt.Errorf("IDENTIFY_SYSTEM: %w", err)
	}
	log.Printf("source system_id=%s timeline=%d database=%s start_lsn=%s",
		system.SystemID, system.Timeline, system.DBName, startLSN)

	err = pglogrepl.StartReplication(ctx, replConn, cfg.slotName, startLSN,
		pglogrepl.StartReplicationOptions{
			PluginArgs: []string{
				"proto_version '2'",
				fmt.Sprintf("publication_names '%s'", cfg.publication),
				"streaming 'on'",
			},
		})
	if err != nil {
		return fmt.Errorf("START_REPLICATION: %w", err)
	}

	queue := make(chan transactionEnvelope, cfg.queueCap)
	acks := make(chan durableAck, cfg.queueCap)
	sinkState := newSinkWorkerState()
	workerCtx, cancelWorker := context.WithCancel(context.Background())

	go func() {
		sinkState.finish(sinkLoop(workerCtx, sinkPool, cfg.consumerID, queue, acks))
	}()

	defer func() {
		close(queue)

		timer := time.NewTimer(10 * time.Second)
		defer timer.Stop()
		select {
		case <-sinkState.done:
		case <-timer.C:
			cancelWorker()
		}
		cancelWorker()
	}()

	relations := make(map[uint32]relationMeta)
	txns := make(map[uint32]*transactionBuffer)
	var currentXID uint32
	var inStream bool
	durableLSN := startLSN
	lastStatus := time.Time{}

	for {
		if err := drainAcks(acks, &durableLSN); err != nil {
			return err
		}

		select {
		case <-sinkState.done:
			err := sinkState.Err()
			if err == nil {
				return errors.New("sink worker stopped unexpectedly")
			}
			return fmt.Errorf("sink worker: %w", err)
		default:
		}

		if time.Since(lastStatus) >= statusInterval {
			if err := sendStatus(ctx, replConn, durableLSN, false); err != nil {
				return fmt.Errorf("periodic status update: %w", err)
			}
			lastStatus = time.Now()
		}

		receiveCtx, cancel := context.WithTimeout(ctx, receiveTimeout)
		raw, err := replConn.ReceiveMessage(receiveCtx)
		cancel()
		if err != nil {
			if ctx.Err() != nil {
				return ctx.Err()
			}
			if pgconn.Timeout(err) {
				continue
			}
			return fmt.Errorf("receive replication message: %w", err)
		}

		switch message := raw.(type) {
		case *pgproto3.ErrorResponse:
			return pgconn.ErrorResponseToPgError(message)
		case *pgproto3.CopyData:
			if len(message.Data) == 0 {
				continue
			}

			switch message.Data[0] {
			case pglogrepl.PrimaryKeepaliveMessageByteID:
				keepalive, err := pglogrepl.ParsePrimaryKeepaliveMessage(message.Data[1:])
				if err != nil {
					return fmt.Errorf("parse keepalive: %w", err)
				}
				if keepalive.ReplyRequested {
					if err := sendStatus(ctx, replConn, durableLSN, false); err != nil {
						return fmt.Errorf("keepalive reply: %w", err)
					}
					lastStatus = time.Now()
				}

			case pglogrepl.XLogDataByteID:
				xld, err := pglogrepl.ParseXLogData(message.Data[1:])
				if err != nil {
					return fmt.Errorf("parse XLogData: %w", err)
				}

				logicalMessage, err := pglogrepl.ParseV2(xld.WALData, inStream)
				if err != nil {
					return fmt.Errorf("parse pgoutput v2 at WAL %s: %w", xld.WALStart, err)
				}

				envelope, committed, err := handleLogicalMessage(
					logicalMessage,
					&inStream,
					&currentXID,
					relations,
					txns,
					cfg.maxTxBytes,
				)
				if err != nil {
					return err
				}
				if committed {
					if err := enqueueWithBackpressure(
						ctx, replConn, queue, acks, sinkState,
						envelope, &durableLSN, &lastStatus,
					); err != nil {
						return err
					}
				}
			}
		}
	}
}

func ensureSinkSchema(ctx context.Context, pool *pgxpool.Pool) error {
	statements := []string{
		`CREATE SCHEMA IF NOT EXISTS cdc_meta`,
		`CREATE TABLE IF NOT EXISTS cdc_meta.checkpoint (
			consumer_id text PRIMARY KEY,
			last_lsn pg_lsn NOT NULL,
			updated_at timestamptz NOT NULL DEFAULT clock_timestamp()
		)`,
		`CREATE TABLE IF NOT EXISTS cdc_meta.schema_version (
			consumer_id text NOT NULL,
			schema_version text NOT NULL,
			relation_name text NOT NULL,
			definition jsonb NOT NULL,
			seen_at timestamptz NOT NULL DEFAULT clock_timestamp(),
			PRIMARY KEY (consumer_id, schema_version)
		)`,
		`CREATE TABLE IF NOT EXISTS cdc_meta.event (
			consumer_id text NOT NULL,
			transaction_end_lsn pg_lsn NOT NULL,
			event_index integer NOT NULL,
			xid bigint NOT NULL,
			commit_lsn pg_lsn NOT NULL,
			commit_time timestamptz NOT NULL,
			relation_name text NOT NULL,
			schema_version text NOT NULL,
			operation text NOT NULL,
			old_data jsonb,
			new_data jsonb,
			metadata jsonb,
			PRIMARY KEY (consumer_id, transaction_end_lsn, event_index)
		)`,
	}

	for _, statement := range statements {
		queryCtx, cancel := context.WithTimeout(ctx, sinkQueryTimeout)
		_, err := pool.Exec(queryCtx, statement)
		cancel()
		if err != nil {
			return fmt.Errorf("initialize sink schema: %w", err)
		}
	}
	return nil
}

func loadStartLSN(ctx context.Context, pool *pgxpool.Pool, cfg config) (pglogrepl.LSN, error) {
	queryCtx, cancel := context.WithTimeout(ctx, sinkQueryTimeout)
	defer cancel()

	var checkpoint string
	err := pool.QueryRow(queryCtx,
		`SELECT last_lsn::text FROM cdc_meta.checkpoint WHERE consumer_id = $1`,
		cfg.consumerID,
	).Scan(&checkpoint)

	switch {
	case err == nil:
		lsn, parseErr := pglogrepl.ParseLSN(checkpoint)
		if parseErr != nil {
			return 0, fmt.Errorf("parse stored checkpoint %q: %w", checkpoint, parseErr)
		}
		return lsn, nil
	case errors.Is(err, pgx.ErrNoRows):
		if cfg.startLSN == "" {
			return 0, errors.New("no durable checkpoint exists; START_LSN must be the slot's verified consistent start point")
		}
		lsn, parseErr := pglogrepl.ParseLSN(cfg.startLSN)
		if parseErr != nil {
			return 0, fmt.Errorf("parse START_LSN %q: %w", cfg.startLSN, parseErr)
		}
		return lsn, nil
	default:
		return 0, fmt.Errorf("load checkpoint: %w", err)
	}
}

func handleLogicalMessage(
	message pglogrepl.Message,
	inStream *bool,
	currentXID *uint32,
	relations map[uint32]relationMeta,
	txns map[uint32]*transactionBuffer,
	maxTxBytes int,
) (transactionEnvelope, bool, error) {
	switch m := message.(type) {
	case *pglogrepl.BeginMessage:
		*currentXID = m.Xid
		ensureTxn(txns, m.Xid)

	case *pglogrepl.RelationMessageV2:
		relation, err := buildRelationMeta(m)
		if err != nil {
			return transactionEnvelope{}, false, err
		}
		relations[m.RelationID] = relation

	case *pglogrepl.InsertMessageV2:
		xid := selectXID(m.Xid, *currentXID)
		relation, ok := relations[m.RelationID]
		if !ok {
			return transactionEnvelope{}, false, fmt.Errorf("insert references unknown relation id %d", m.RelationID)
		}
		event := changeEvent{
			Operation:     "INSERT",
			RelationName:  relation.QualifiedName(),
			SchemaVersion: relation.Version,
			New:           tupleToMap(relation, m.Tuple, false),
		}
		if err := appendEvent(txns, xid, event, relation, maxTxBytes); err != nil {
			return transactionEnvelope{}, false, err
		}

	case *pglogrepl.UpdateMessageV2:
		xid := selectXID(m.Xid, *currentXID)
		relation, ok := relations[m.RelationID]
		if !ok {
			return transactionEnvelope{}, false, fmt.Errorf("update references unknown relation id %d", m.RelationID)
		}
		event := changeEvent{
			Operation:     "UPDATE",
			RelationName:  relation.QualifiedName(),
			SchemaVersion: relation.Version,
			Old:           tupleToMap(relation, m.OldTuple, m.OldTupleType == 'K'),
			New:           tupleToMap(relation, m.NewTuple, false),
			Metadata:      map[string]any{"old_tuple_type": string([]byte{m.OldTupleType})},
		}
		if err := appendEvent(txns, xid, event, relation, maxTxBytes); err != nil {
			return transactionEnvelope{}, false, err
		}

	case *pglogrepl.DeleteMessageV2:
		xid := selectXID(m.Xid, *currentXID)
		relation, ok := relations[m.RelationID]
		if !ok {
			return transactionEnvelope{}, false, fmt.Errorf("delete references unknown relation id %d", m.RelationID)
		}
		event := changeEvent{
			Operation:     "DELETE",
			RelationName:  relation.QualifiedName(),
			SchemaVersion: relation.Version,
			Old:           tupleToMap(relation, m.OldTuple, m.OldTupleType == 'K'),
			Metadata:      map[string]any{"old_tuple_type": string([]byte{m.OldTupleType})},
		}
		if err := appendEvent(txns, xid, event, relation, maxTxBytes); err != nil {
			return transactionEnvelope{}, false, err
		}

	case *pglogrepl.TruncateMessageV2:
		xid := selectXID(m.Xid, *currentXID)
		for _, relationID := range m.RelationIDs {
			relation, ok := relations[relationID]
			if !ok {
				return transactionEnvelope{}, false, fmt.Errorf("truncate references unknown relation id %d", relationID)
			}
			event := changeEvent{
				Operation:     "TRUNCATE",
				RelationName:  relation.QualifiedName(),
				SchemaVersion: relation.Version,
				Metadata: map[string]any{
					"cascade":          m.Option&pglogrepl.TruncateOptionCascade != 0,
					"restart_identity": m.Option&pglogrepl.TruncateOptionRestartIdentity != 0,
				},
			}
			if err := appendEvent(txns, xid, event, relation, maxTxBytes); err != nil {
				return transactionEnvelope{}, false, err
			}
		}

	case *pglogrepl.StreamStartMessageV2:
		*inStream = true
		ensureTxn(txns, m.Xid)

	case *pglogrepl.StreamStopMessageV2:
		*inStream = false

	case *pglogrepl.StreamAbortMessageV2:
		delete(txns, m.Xid)
		*inStream = false

	case *pglogrepl.StreamCommitMessageV2:
		envelope, err := commitTxn(txns, m.Xid, m.CommitLSN, m.TransactionEndLSN, m.CommitTime)
		*inStream = false
		if err != nil {
			return transactionEnvelope{}, false, err
		}
		return envelope, true, nil

	case *pglogrepl.CommitMessage:
		xid := *currentXID
		envelope, err := commitTxn(txns, xid, m.CommitLSN, m.TransactionEndLSN, m.CommitTime)
		*currentXID = 0
		if err != nil {
			return transactionEnvelope{}, false, err
		}
		return envelope, true, nil
	}

	return transactionEnvelope{}, false, nil
}

func (r relationMeta) QualifiedName() string {
	return r.Namespace + "." + r.Name
}

func buildRelationMeta(message *pglogrepl.RelationMessageV2) (relationMeta, error) {
	relation := relationMeta{
		RelationID:      message.RelationID,
		Namespace:       message.Namespace,
		Name:            message.RelationName,
		ReplicaIdentity: message.ReplicaIdentity,
		Columns:         make([]columnMeta, 0, len(message.Columns)),
	}
	for _, column := range message.Columns {
		relation.Columns = append(relation.Columns, columnMeta{
			Name:         column.Name,
			DataTypeOID:  column.DataType,
			TypeModifier: column.TypeModifier,
			Key:          column.Flags&1 == 1,
		})
	}

	definition, err := json.Marshal(relation)
	if err != nil {
		return relationMeta{}, fmt.Errorf("marshal relation schema: %w", err)
	}
	digest := sha256.Sum256(definition)
	relation.Version = hex.EncodeToString(digest[:])
	relation.Definition = definition
	return relation, nil
}

func selectXID(messageXID, currentXID uint32) uint32 {
	if messageXID != 0 {
		return messageXID
	}
	return currentXID
}

func ensureTxn(txns map[uint32]*transactionBuffer, xid uint32) *transactionBuffer {
	if xid == 0 {
		return nil
	}
	if txns[xid] == nil {
		txns[xid] = &transactionBuffer{
			XID:     xid,
			Schemas: make(map[string]schemaRecord),
		}
	}
	return txns[xid]
}

func appendEvent(
	txns map[uint32]*transactionBuffer,
	xid uint32,
	event changeEvent,
	relation relationMeta,
	maxTxBytes int,
) error {
	buffer := ensureTxn(txns, xid)
	if buffer == nil {
		return errors.New("received row change outside a known transaction")
	}

	encoded, err := json.Marshal(event)
	if err != nil {
		return fmt.Errorf("estimate event size: %w", err)
	}
	buffer.Bytes += len(encoded)
	if buffer.Bytes > maxTxBytes {
		return fmt.Errorf(
			"transaction %d exceeded MAX_TX_BYTES=%d; stop without ACK and use a durable disk spool in production",
			xid, maxTxBytes,
		)
	}

	buffer.Events = append(buffer.Events, event)
	buffer.Schemas[relation.Version] = schemaRecord{
		Version:      relation.Version,
		RelationName: relation.QualifiedName(),
		Definition:   relation.Definition,
	}
	return nil
}

func commitTxn(
	txns map[uint32]*transactionBuffer,
	xid uint32,
	commitLSN pglogrepl.LSN,
	endLSN pglogrepl.LSN,
	commitTime time.Time,
) (transactionEnvelope, error) {
	buffer := txns[xid]
	if buffer == nil {
		return transactionEnvelope{}, fmt.Errorf("commit for unknown xid %d", xid)
	}
	delete(txns, xid)

	return transactionEnvelope{
		XID:        xid,
		CommitLSN:  commitLSN,
		EndLSN:     endLSN,
		CommitTime: commitTime,
		Events:     buffer.Events,
		Schemas:    buffer.Schemas,
	}, nil
}

func tupleToMap(relation relationMeta, tuple *pglogrepl.TupleData, keyOnly bool) map[string]any {
	if tuple == nil {
		return nil
	}

	columns := relation.Columns
	if keyOnly && len(tuple.Columns) != len(columns) {
		keyColumns := make([]columnMeta, 0, len(columns))
		for _, column := range columns {
			if column.Key {
				keyColumns = append(keyColumns, column)
			}
		}
		if len(tuple.Columns) == len(keyColumns) {
			columns = keyColumns
		}
	}

	result := make(map[string]any, len(tuple.Columns))
	for i, value := range tuple.Columns {
		name := fmt.Sprintf("column_%d", i)
		if i < len(columns) {
			name = columns[i].Name
		}

		switch value.DataType {
		case pglogrepl.TupleDataTypeNull:
			result[name] = nil
		case pglogrepl.TupleDataTypeToast:
			result[name] = map[string]string{"kind": "unchanged_toast"}
		case pglogrepl.TupleDataTypeText:
			result[name] = string(value.Data)
		case pglogrepl.TupleDataTypeBinary:
			result[name] = map[string]string{
				"kind":  "binary_base64",
				"value": base64.StdEncoding.EncodeToString(value.Data),
			}
		default:
			result[name] = map[string]any{"kind": "unknown", "code": value.DataType}
		}
	}
	return result
}

func enqueueWithBackpressure(
	ctx context.Context,
	conn *pgconn.PgConn,
	queue chan<- transactionEnvelope,
	acks <-chan durableAck,
	sinkState *sinkWorkerState,
	envelope transactionEnvelope,
	durableLSN *pglogrepl.LSN,
	lastStatus *time.Time,
) error {
	ticker := time.NewTicker(time.Second)
	defer ticker.Stop()

	for {
		select {
		case queue <- envelope:
			return nil
		case ack := <-acks:
			if ack.EndLSN > *durableLSN {
				*durableLSN = ack.EndLSN
			}
		case <-sinkState.done:
			err := sinkState.Err()
			if err == nil {
				return errors.New("sink worker stopped while queue was backpressured")
			}
			return fmt.Errorf("sink worker: %w", err)
		case <-ticker.C:
			if time.Since(*lastStatus) >= statusInterval {
				if err := sendStatus(ctx, conn, *durableLSN, false); err != nil {
					return fmt.Errorf("status update during backpressure: %w", err)
				}
				*lastStatus = time.Now()
			}
		case <-ctx.Done():
			return ctx.Err()
		}
	}
}

func drainAcks(acks <-chan durableAck, durableLSN *pglogrepl.LSN) error {
	for {
		select {
		case ack := <-acks:
			if ack.EndLSN < *durableLSN {
				return fmt.Errorf("durable LSN moved backwards: current=%s ack=%s", *durableLSN, ack.EndLSN)
			}
			*durableLSN = ack.EndLSN
		default:
			return nil
		}
	}
}

func sendStatus(ctx context.Context, conn *pgconn.PgConn, durableLSN pglogrepl.LSN, requestReply bool) error {
	return pglogrepl.SendStandbyStatusUpdate(ctx, conn, pglogrepl.StandbyStatusUpdate{
		WALWritePosition: durableLSN,
		WALFlushPosition: durableLSN,
		WALApplyPosition: durableLSN,
		ClientTime:       time.Now(),
		ReplyRequested:   requestReply,
	})
}

func sinkLoop(
	ctx context.Context,
	pool *pgxpool.Pool,
	consumerID string,
	queue <-chan transactionEnvelope,
	acks chan<- durableAck,
) error {
	for {
		select {
		case <-ctx.Done():
			return ctx.Err()
		case envelope, ok := <-queue:
			if !ok {
				return nil
			}
			if err := persistWithRetry(ctx, pool, consumerID, envelope); err != nil {
				return err
			}
			select {
			case acks <- durableAck{EndLSN: envelope.EndLSN}:
			case <-ctx.Done():
				return ctx.Err()
			}
		}
	}
}

func persistWithRetry(
	ctx context.Context,
	pool *pgxpool.Pool,
	consumerID string,
	envelope transactionEnvelope,
) error {
	var lastErr error
	for attempt := 0; attempt < maxSinkRetries; attempt++ {
		lastErr = persistOnce(ctx, pool, consumerID, envelope)
		if lastErr == nil {
			return nil
		}
		if !isRetryableTransactionError(lastErr) {
			return lastErr
		}

		base := 100 * time.Millisecond * time.Duration(1<<attempt)
		jitter := time.Duration(rand.Int63n(int64(base/2) + 1))
		delay := base + jitter
		select {
		case <-time.After(delay):
		case <-ctx.Done():
			return ctx.Err()
		}
	}
	return fmt.Errorf("sink transaction failed after %d retries: %w", maxSinkRetries, lastErr)
}

func persistOnce(
	ctx context.Context,
	pool *pgxpool.Pool,
	consumerID string,
	envelope transactionEnvelope,
) error {
	queryCtx, cancel := context.WithTimeout(ctx, sinkQueryTimeout)
	defer cancel()

	tx, err := pool.BeginTx(queryCtx, pgx.TxOptions{IsoLevel: pgx.ReadCommitted})
	if err != nil {
		return fmt.Errorf("begin sink transaction: %w", err)
	}
	defer func() {
		rollbackCtx, rollbackCancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer rollbackCancel()
		_ = tx.Rollback(rollbackCtx)
	}()

	for _, schema := range envelope.Schemas {
		_, err = tx.Exec(queryCtx, `
			INSERT INTO cdc_meta.schema_version(
				consumer_id, schema_version, relation_name, definition
			)
			VALUES ($1, $2, $3, $4::jsonb)
			ON CONFLICT (consumer_id, schema_version) DO NOTHING`,
			consumerID, schema.Version, schema.RelationName, string(schema.Definition),
		)
		if err != nil {
			return fmt.Errorf("insert schema version: %w", err)
		}
	}

	for index, event := range envelope.Events {
		oldJSON, err := marshalNullableJSON(event.Old)
		if err != nil {
			return err
		}
		newJSON, err := marshalNullableJSON(event.New)
		if err != nil {
			return err
		}
		metadataJSON, err := marshalNullableJSON(event.Metadata)
		if err != nil {
			return err
		}

		_, err = tx.Exec(queryCtx, `
			INSERT INTO cdc_meta.event(
				consumer_id,
				transaction_end_lsn,
				event_index,
				xid,
				commit_lsn,
				commit_time,
				relation_name,
				schema_version,
				operation,
				old_data,
				new_data,
				metadata
			)
			VALUES (
				$1, $2::pg_lsn, $3, $4, $5::pg_lsn, $6,
				$7, $8, $9, $10::jsonb, $11::jsonb, $12::jsonb
			)
			ON CONFLICT (consumer_id, transaction_end_lsn, event_index)
			DO NOTHING`,
			consumerID,
			envelope.EndLSN.String(),
			index,
			envelope.XID,
			envelope.CommitLSN.String(),
			envelope.CommitTime,
			event.RelationName,
			event.SchemaVersion,
			event.Operation,
			oldJSON,
			newJSON,
			metadataJSON,
		)
		if err != nil {
			return fmt.Errorf("insert CDC event: %w", err)
		}
	}

	_, err = tx.Exec(queryCtx, `
		INSERT INTO cdc_meta.checkpoint(consumer_id, last_lsn, updated_at)
		VALUES ($1, $2::pg_lsn, clock_timestamp())
		ON CONFLICT (consumer_id) DO UPDATE
		SET last_lsn = EXCLUDED.last_lsn,
			updated_at = EXCLUDED.updated_at
		WHERE cdc_meta.checkpoint.last_lsn < EXCLUDED.last_lsn`,
		consumerID, envelope.EndLSN.String(),
	)
	if err != nil {
		return fmt.Errorf("advance durable checkpoint: %w", err)
	}

	if err := tx.Commit(queryCtx); err != nil {
		// A network error here can mean the server committed but the client did not
		// receive confirmation. Do not ACK PostgreSQL. A restart safely replays the
		// envelope because event keys and checkpoint advancement are idempotent.
		return fmt.Errorf("sink commit outcome unknown: %w", err)
	}
	return nil
}

func marshalNullableJSON(value map[string]any) (any, error) {
	if value == nil {
		return nil, nil
	}
	encoded, err := json.Marshal(value)
	if err != nil {
		return nil, fmt.Errorf("marshal event JSON: %w", err)
	}
	return encoded, nil
}

func isRetryableTransactionError(err error) bool {
	var pgErr *pgconn.PgError
	if !errors.As(err, &pgErr) {
		return false
	}
	return pgErr.Code == "40001" || pgErr.Code == "40P01"
}

16.5 验证增量、Checkpoint 与重复吸收

在 Publisher 写入一个多语句事务：

BEGIN;
INSERT INTO lab22_cdc.account(account_id, owner_name, balance)
VALUES (1, 'Ada', 100.00);

UPDATE lab22_cdc.account
SET balance = balance + 25,
    updated_at = clock_timestamp()
WHERE account_id = 1;
COMMIT;

在 Sink 检查：

SELECT consumer_id,
       transaction_end_lsn,
       event_index,
       xid,
       operation,
       relation_name,
       schema_version,
       old_data,
       new_data
FROM cdc_meta.event
ORDER BY transaction_end_lsn, event_index;

SELECT *
FROM cdc_meta.checkpoint;

SELECT consumer_id,
       schema_version,
       relation_name,
       definition
FROM cdc_meta.schema_version;

同一源事务的两条变化应具有相同 transaction_end_lsn，但 event_index 不同；Checkpoint 等于该事务的 End LSN。

验证重复吸收时，可在测试环境制造以下时序：

  1. 让 Sink 事务成功提交；
  2. 在程序发送下一次 Status Update 前强制结束进程；
  3. 重新启动同一 CONSUMER_ID；
  4. 检查 cdc_meta.event 主键没有产生第二份事件，Checkpoint 仍单调前进。

不要通过删除 Checkpoint 来“测试重放”，除非同时确认 Slot 仍保留对应 WAL；也不要把旧 Checkpoint 配给另一个 Slot 或另一个 PostgreSQL system_identifier。

16.6 ACK 顺序为何不会静默丢数据

故障点Sink 状态Publisher Slot 状态重启结果
Sink 事务提交前崩溃事件与 Checkpoint 均未提交LSN 未确认整个源事务重放
Sink 提交后、ACK 前崩溃事件与 Checkpoint 已提交可能仍未确认从持久化 Checkpoint 请求；若仍重放，由幂等键吸收
ACK 已发送后崩溃事件与 Checkpoint 已提交confirmed_flush_lsn 可前进从 Checkpoint 之后继续
Sink Commit 返回网络错误提交结果未知不 ACK重启查询 Checkpoint；已提交则去重，未提交则重新落库
Queue 满未入队事务不持久化不确认该事务保留 WAL，形成可观测 Backpressure
单事务超过上限事务未持久化不确认程序停止；运维需提高上限或换磁盘 Spool

这里的关键不变量是：

acknowledged_lsn <= durable_sink_checkpoint

任何“先 ACK、后写 Sink”的实现都可能在 Sink 失败时形成不可恢复的数据缺口。

16.7 Schema Version 与 DDL 处理

pgoutput 在关系定义发生变化后发送新的 Relation 元数据。示例将 Namespace、Relation、Replica Identity、列名、类型 OID、Type Modifier 与 Key Flag 序列化后做 SHA-256，得到稳定的 schema_version。这样下游可以：

  • 按版本选择反序列化器；
  • 在新版本首次出现时阻塞并报警；
  • 将旧事件按旧 Schema 重放，而不套用当前表结构；
  • 对 Backfill 数据和 CDC 数据使用同一 Schema Version。

但 Relation 消息不是完整 DDL 审计记录。它无法替代 Migration Registry，也不能告诉你某列为什么重命名、默认值如何计算、历史数据是否已 Backfill。生产系统应把每次 DDL 的 Migration ID、兼容区间、Owner、回滚脚本和下游审批状态写入独立 Schema Registry。

推荐的事件契约是：

{
  "source": {
    "system_id": "...",
    "slot": "lab22_cdc_slot",
    "transaction_end_lsn": "0/16B7EF8",
    "xid": 7421,
    "commit_time": "2026-06-21T03:00:00Z"
  },
  "schema_version": "sha256:...",
  "operation": "UPDATE",
  "relation": "lab22_cdc.account",
  "old": {"account_id": "1"},
  "new": {"account_id": "1", "balance": "125.00"}
}

应把 transaction_end_lsn 视为源内顺序与幂等依据之一，不要把它解释成跨多个 PostgreSQL Cluster 的全局时间。

16.8 Backpressure、长事务与内存边界

协议 Version 2 允许未提交大事务分段传输，解决的是“Publisher 必须等到 Commit 才开始发送”的延迟和内存问题；它并不自动赋予任意 Consumer 磁盘持久化能力。本示例仍按 XID 把整个事务缓存在内存，直到 Commit 后才原子写 Sink，因此设置 MAX_TX_BYTES。

生产实现通常需要：

  1. 将每个 XID 的片段写入 Durable Spool；
  2. 用 XID、Segment 序号和 WAL 位置校验完整性；
  3. 只在 STREAM COMMIT 后发布事务；
  4. 在 STREAM ABORT 时删除暂存片段；
  5. 对 Spool 磁盘设置配额、告警和灾难恢复；
  6. 在 Sink 变慢时限制拉取速度，但持续处理 Keepalive；
  7. 把 Slot WAL Retention 与 Spool 容量共同纳入 SLO。

16.9 CDC 清理

确认 Consumer 永久下线、无需重放且已满足审计保留要求后，才删除 Slot：

SELECT pg_drop_replication_slot('lab22_cdc_slot');
DROP PUBLICATION IF EXISTS lab22_cdc_pub;

删除前至少核对：Consumer Owner、最后 Checkpoint、目标端完整性、回放窗口、备份状态、是否存在故障转移 Slot，以及是否还有其他服务复用该 Slot。一个 Slot 不应被多个独立消费者共享进度。

17. 生产排障 Runbook

17.1 首先确认什么

发生“复制延迟、Subscription 停止、Publisher 磁盘上涨、CDC 无事件或切流后冲突”时，前五分钟先固定事实，不要先执行 SKIP、删 Slot 或重建订阅。

  1. 拓扑：Publisher、物理 Standby、Subscriber/CDC Consumer、Publication、Subscription、Slot、DNS/代理实际指向。
  2. 版本：每个节点的精确 Major/Minor Version；特别标记 PG14–18 的 Streaming、Two-Phase、Failover Slot 与冲突统计差异。
  3. 故障边界：是无新事件、延迟增长、单表未同步、Apply Error、目标数据冲突、Slot 失效，还是源端 pg_wal 空间不足。
  4. 写入权威：此刻哪一端允许业务写；是否发生双写或旧连接未被 Fencing。
  5. 最近变化：DDL、Publication/Subscription 参数、网络、证书、权限、Extension、主备切换、批处理、大事务和版本升级时间。
  6. 数据安全：最后一个已验证业务标记、Sink Checkpoint、confirmed_flush_lsn、备份/PITR 可用性和回滚窗口。

先留存证据：

SELECT clock_timestamp() AS captured_at,
       version(),
       pg_is_in_recovery(),
       current_database(),
       current_user;

SHOW wal_level;
SHOW max_replication_slots;
SHOW max_wal_senders;
SHOW max_slot_wal_keep_size;
SHOW idle_replication_slot_timeout;  -- [PG18]

同时保存 PostgreSQL 日志中故障前后至少一个完整业务峰值周期。对冲突，应保存 CONTEXT 中的 Subscription Origin、Relation、操作和 Finish LSN。

17.2 查看哪些指标

指标计算/来源解释常见误判
Slot Retained Bytespg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)Publisher 因 Slot 不能回收的 WAL 距离近似值不等于 pg_wal 目录物理大小
Unconfirmed Bytespg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)消费确认落后于当前 WAL 的距离长事务可使两个 LSN 呈现不同滞后
WAL Runway可用磁盘字节 ÷ 峰值 WAL 字节/秒在当前增长速率下的剩余处理时间不能用日均 WAL 速率估算峰值事故
Apply LSN LagPublisher 当前 LSN 与 Subscriber latest_end_lsn 的差增量位置距离跨 Cluster LSN 只有同一复制流语境才可比较
Apply Time Lagnow() - latest_end_time最近应用远端事务的时间距离Publisher 空闲时会看起来“很旧”，需结合消息时间和 LSN
Worker Count/Statepg_stat_subscriptionApply、Table Sync、Parallel Apply Worker 是否存在Worker 重启窗口内短暂为空不一定是事故
Table Sync Statepg_subscription_rel.srsubstate每表处于初始化、COPY、Catch-up 或 Ready只看主 Apply Worker 会漏掉单表卡住
Conflict Counters[PG18] pg_stat_subscription_stats新增冲突和 Apply/Sync Error计数是累计值，应看增量和 stats_reset
CPU、I/O、Memory、TempOS/cgroup、pg_stat_io、日志解码、COPY、索引维护、排序或 Spill 是否成为瓶颈CPU 低不代表健康，可能在等锁/网络/I/O
Lock/Wait Eventpg_stat_activity、pg_locksApply Worker 被 DDL、长事务或目标端写阻塞state='active' 仍可能长期等待
Connection/Poolpg_stat_activity、Pool 指标Worker、Table Sync、业务连接是否耗尽配额只提高 max_connections 可能加剧内存和调度压力
Vacuum/XminSlot catalog_xmin、表年龄、Dead TupleLogical Decoding 与长 Snapshot 是否推迟清理WAL 保留和 Heap Bloat 是不同但可同时发生的问题

17.3 查询哪些系统视图

Publisher：Slot 健康与 WAL 预算

SELECT slot_name,
       database,
       plugin,
       active,
       active_pid,
       restart_lsn,
       confirmed_flush_lsn,
       pg_size_pretty(
           pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)
       ) AS retained_from_restart,
       pg_size_pretty(
           pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn)
       ) AS not_yet_confirmed,
       catalog_xmin,
       wal_status,
       safe_wal_size,
       invalidation_reason,
       failover,
       synced
FROM pg_replication_slots
WHERE slot_type = 'logical'
ORDER BY slot_name;

重要字段：

  • active/active_pid：是否有 Walsender 正在使用该 Slot，以及对应 PID；false 不代表不保留 WAL。
  • restart_lsn：最早仍可能被解码需要的 WAL；它决定回收下界。
  • confirmed_flush_lsn：Consumer 已确认持久化的位置；不能由“已收到”替代“已持久化”。
  • catalog_xmin：Slot 需要保留的系统 Catalog 行版本边界；长期停滞会影响 Catalog Vacuum。
  • wal_status/safe_wal_size：在 max_slot_wal_keep_size 等约束下的可用性和剩余预算。
  • invalidation_reason：非空表示 Slot 已失去继续解码所需条件，通常需要重建与重新基线。
  • failover/synced：[PG17+] 判断 Slot 是否被标记为可故障转移、Standby 副本是否完成同步。

Subscriber：Worker、接收和应用位置

SELECT subname,
       worker_type,
       pid,
       leader_pid,
       relid::regclass AS relation,
       received_lsn,
       last_msg_send_time,
       last_msg_receipt_time,
       latest_end_lsn,
       latest_end_time
FROM pg_stat_subscription
ORDER BY subname, worker_type, relation NULLS FIRST;
  • worker_type：区分主 Apply、Table Sync 与 Parallel Apply Worker。
  • relid：Table Sync Worker 正在处理的表；主 Apply 通常为空。
  • received_lsn：Subscriber 已收到的位置，不代表已提交到目标表。
  • latest_end_lsn/latest_end_time：最近完成处理的远端事务边界和时间。
  • last_msg_*：用于区分 Publisher 没发送、网络未收到、还是收到后 Apply 变慢。

每表状态：

SELECT s.subname,
       r.srrelid::regclass AS relation,
       r.srsubstate,
       r.srsublsn
FROM pg_subscription_rel AS r
JOIN pg_subscription AS s
  ON s.oid = r.srsubid
ORDER BY s.subname, relation;

任何长期未到 r（Ready）的表，都应按“Schema、权限、COPY 冲突、Table Sync Slot、网络、Worker 配额”逐项检查。

[PG18] 冲突增量：

SELECT subname,
       apply_error_count,
       sync_error_count,
       confl_insert_exists,
       confl_update_origin_differs,
       confl_update_exists,
       confl_update_missing,
       confl_delete_origin_differs,
       confl_delete_missing,
       confl_multiple_unique_conflicts,
       stats_reset
FROM pg_stat_subscription_stats
ORDER BY subname;

Replication Origin：

SELECT external_id,
       remote_lsn,
       local_lsn
FROM pg_replication_origin_status
ORDER BY external_id;

remote_lsn 属于 Publisher WAL 空间，local_lsn 属于 Subscriber WAL 空间；二者表达映射关系，不能互相做字节差。

17.4 如何找到 Blocker

先定位所有 Logical Replication Backend 及其等待：

SELECT pid,
       backend_type,
       application_name,
       state,
       wait_event_type,
       wait_event,
       xact_start,
       query_start,
       left(query, 300) AS query
FROM pg_stat_activity
WHERE backend_type ILIKE '%logical replication%'
   OR application_name IN (
       SELECT subname FROM pg_subscription
   )
ORDER BY query_start NULLS LAST;

再展开直接 Blocker：

SELECT blocked.pid AS blocked_pid,
       blocked.backend_type AS blocked_type,
       blocked.wait_event_type,
       blocked.wait_event,
       blocker.pid AS blocker_pid,
       blocker.usename AS blocker_user,
       blocker.application_name AS blocker_app,
       blocker.xact_start AS blocker_xact_start,
       left(blocker.query, 300) AS blocker_query
FROM pg_stat_activity AS blocked
CROSS JOIN LATERAL
     unnest(pg_blocking_pids(blocked.pid)) AS p(blocker_pid)
JOIN pg_stat_activity AS blocker
  ON blocker.pid = p.blocker_pid
WHERE blocked.backend_type ILIKE '%logical replication%'
ORDER BY blocker.xact_start;

常见 Blocker：Subscriber 上未提交的业务事务、ALTER TABLE、索引维护、手工数据修复、外键检查，以及大批量 DML。终止 Blocker 前必须确认事务 Owner 和业务副作用；不要因为它“挡住复制”就直接 pg_terminate_backend()。

17.5 如何找到最早出现的执行计划估算错误

逻辑复制事故不全是执行计划问题，但 Subscriber 的 UPDATE/DELETE 行定位、校验 SQL和 Backfill 查询可能因统计信息过期或缺索引退化。pg_stat_statements 只保存累计统计，不能凭空还原“最早何时出现错误估算”；要依靠持续快照或 auto_explain 日志建立时间线。

检查高成本语句：

SELECT queryid,
       calls,
       rows,
       total_exec_time,
       mean_exec_time,
       shared_blks_hit,
       shared_blks_read,
       temp_blks_written,
       wal_bytes,
       left(query, 300) AS query
FROM pg_stat_statements
ORDER BY total_exec_time DESC
LIMIT 30;

对 Replica Identity 查找做代表性验证：

EXPLAIN (
    ANALYZE,
    BUFFERS,
    WAL,
    SETTINGS,
    VERBOSE,
    SUMMARY
)
SELECT *
FROM target_schema.target_table
WHERE replica_key = 123456;

判断要点：

  • Plan Rows 与 Actual Rows 从何时开始显著偏离；
  • 是否在 DDL、数据倾斜、Backfill 或 ANALYZE 失效后出现；
  • 是否从 Index Scan 退化为 Seq Scan；
  • Buffers、Temp、I/O 和 Wait Event 是否同步变化；
  • Subscriber 上的 Replica Identity/唯一索引是否与 Publisher 语义一致。

需要“最早时间”时，应从监控系统按 queryid + plan fingerprint 回看 P95、估算误差和 Buffers 的首次跃迁，并与发布记录对齐。事后没有历史数据时，只能给出证据支持的时间区间，不能伪造精确起点。

17.6 如何区分 CPU、内存、I/O、锁、连接池、WAL、Vacuum 与复制问题

类别典型证据进一步验证处理方向
CPUSubscriber CPU 饱和，Run Queue 高，Apply Lag 增长Worker CPU、索引数量、触发器、类型转换降低附加索引/触发器成本，增加可用 CPU，拆分 Subscription
内存OOM、进程被杀、Temp/Spool 增长大事务、logical_decoding_work_mem、Consumer Buffer大事务 Streaming、磁盘 Spool、硬上限，不能无限加内存
I/ODataFileRead/Write、WALWrite/WALSync 等等待高pg_stat_io、设备延迟、Checkpoint提升存储、平滑 Checkpoint、减少目标端写放大
锁Apply Worker 有 Lock Wait Eventpg_blocking_pids()、DDL/长事务协调事务，设置 DDL Lock Timeout，必要时暂停迁移
连接池Worker/业务无法建立连接，连接数逼近上限pg_stat_activity、Pool Wait为复制保留连接预算，限制 Table Sync 并发和业务池
WALPublisher pg_wal 增长、Slot Retained Bytes 上升restart_lsn、confirmed_flush_lsn、WAL 速率恢复 Consumer、限流、扩容；经审批后重建失活链路
VacuumDead Tuple/年龄增加，catalog_xmin 长期不动Slot、长 Snapshot、Autovacuum 日志解除 Slot/Snapshot 阻塞，再调 Vacuum；不可随意关闭 Autovacuum
复制Worker 消失、冲突计数增、Table State 不前进Subscriber 日志、Subscription 参数、权限/Schema修复具体冲突/Schema/网络，不以重建订阅替代根因分析

17.7 哪些命令通常可以在线执行

以下操作通常不要求停库，但仍应走变更审批并评估负载：

  • 只读查询 pg_replication_slots、pg_stat_subscription、pg_subscription_rel、pg_replication_origin_status；
  • 查看日志、OS 指标、网络和证书状态；
  • ANALYZE 目标表，前提是评估 I/O 与锁影响；
  • 在 Subscriber 先执行兼容性 ADD COLUMN、创建缺失索引；大型索引优先 CREATE INDEX CONCURRENTLY；
  • ALTER SUBSCRIPTION ... DISABLE/ENABLE，用于受控暂停和恢复；暂停会让 Publisher Slot 继续保留 WAL；
  • 调整 Consumer 自身并发、批量和 Backpressure 阈值；
  • 临时扩容 pg_wal 所在文件系统或目标端存储，以换取修复时间。

“在线”只表示数据库仍可服务，不表示没有 WAL、CPU、I/O 或锁代价。

17.8 哪些命令具有高风险

操作风险
DROP SUBSCRIPTION / pg_drop_replication_slot()可能永久丢失尚未消费的增量，回滚窗口消失
ALTER SUBSCRIPTION ... SKIP跳过整个远端事务，不只是冲突行，可能制造静默不一致
pg_replication_origin_advance()人工推进应用位置，错误 LSN 会跳数据
pg_replication_slot_advance()人工推进 Slot，跳过的数据通常无法从该 Slot 恢复
copy_data=false若基线不一致，会把历史缺口永久合法化
Publisher 先执行破坏性 DDL新事件可能立刻不适配 Subscriber，Apply 停止并积压 WAL
Subscriber 本地修数/双写可能造成 Unique Violation、Missing Row 或来源冲突
REPLICA IDENTITY FULL 全面启用增加 WAL、网络和目标端匹配成本，且部分类型有操作类限制
关闭约束、触发器、fsync 或 full_page_writes可能牺牲正确性和崩溃恢复；不能作为常规提速方案
直接删除 pg_wal 文件会破坏 Cluster，绝对禁止
切流后重新开放旧库写入两端分叉；没有反向增量链路就不能称为安全回滚

17.9 临时止损方法

按风险从低到高选择：

  1. 冻结新的 DDL、Backfill 和非关键批处理，减少 WAL 与目标端写放大。
  2. 修复网络、证书、权限或目标端缺列/缺索引，让原 Subscription 继续追赶。
  3. 为 Publisher pg_wal 扩容并限制高 WAL 作业，赢得恢复 Consumer 的时间。
  4. 对 Subscriber Blocker 协调提交/回滚；只有获得 Owner 批准后才终止会话。
  5. 对停止型冲突，先保存远端事务、目标现状和 Finish LSN，再选择 Remote-Wins、Local-Wins 加补偿，或经数据 Owner 批准 SKIP。
  6. 若 Slot 已失效或 Consumer 无法恢复，保留证据后重建 Slot，并重新执行可验证的 Snapshot/Backfill；不要假装可以无损续传。
  7. 磁盘即将耗尽且业务写入面临中断时，可在明确接受“CDC 重新基线”的前提下删除孤儿 Slot。该决定必须由业务、数据和数据库 Owner 联合批准。

17.10 根本修复方法

  • 建立 Slot Owner、用途、数据保留 SLO、最大 WAL 预算和自动回收流程；
  • 将 DDL 纳入 Expand/Contract 和 Subscriber-First 兼容性部署；
  • 让 Subscriber 保持单一写权威，或引入明确的冲突解决协议；
  • 为所有 UPDATE/DELETE 表设置经过评估的 Replica Identity 和目标端唯一索引；
  • 对 CDC 使用持久化 Checkpoint、幂等键、Bounded Queue、Durable Spool 与 Schema Registry；
  • 对大版本迁移建立全量、增量、校验、Fencing、Sequence 对齐、Cutover Marker、反向同步/补偿和回滚期限；
  • [PG17+] 为需要随物理主备切换的链路配置并验证 Failover Slot，而不是只设置一个参数；
  • 将初始 COPY 对生产 CPU、I/O、WAL、网络和 Autovacuum 的冲击纳入容量模型。

17.11 如何验证修复

修复后至少连续观察一个高峰周期，并完成：

-- Publisher：保留量应下降或稳定在预算内
SELECT slot_name,
       active,
       restart_lsn,
       confirmed_flush_lsn,
       pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) AS retained_bytes,
       invalidation_reason
FROM pg_replication_slots
WHERE slot_type = 'logical';

-- Subscriber：Worker 存在、消息时间推进、表均 Ready
SELECT * FROM pg_stat_subscription;
SELECT * FROM pg_subscription_rel;

业务验证不能只看行数：

  • 分桶 Count、主键范围、NULL 分布、金额 Sum、状态分组；
  • 对确定性行做 Hash/Checksum；
  • 插入 Cutover/Validation Marker，确认源端提交后目标端可见；
  • 验证 UPDATE、DELETE、TRUNCATE、Row Filter 边界和 Partition Move；
  • 检查 Sequence/Identity、权限、索引、约束、Extension、Collation；
  • 重启 Consumer/Apply 链路，验证 Checkpoint 和幂等恢复；
  • 触发一次受控 Publisher 主备切换，验证 [PG17+] Slot Failover 链路。

17.12 如何设置监控和告警

至少建立以下告警：

  1. WAL Runway：free_pg_wal_bytes / peak_wal_bytes_per_second 小于组织规定的修复 Lead Time。
  2. Slot Inactive + Retention Growing：active=false 且 retained_bytes 连续增长。
  3. Slot Invalid：invalidation_reason IS NOT NULL 或 wal_status 异常。
  4. Apply Lag：LSN Lag、Time Lag 同时超阈值；空闲 Publisher 不应仅凭时间告警。
  5. Worker Missing：Subscription Enabled 但主 Apply Worker 超过多个重试周期不存在。
  6. Table Sync SLA：srsubstate <> 'r' 超过该表基线窗口。
  7. Conflict/Error Delta：[PG18] 任一冲突或 Error Counter 增量大于零。
  8. Backpressure：Consumer Queue/Spool 使用率、Sink 延迟和重试率超阈值。
  9. Schema Version Unknown：新 Relation Hash 未获下游批准。
  10. Failover Readiness：[PG17+] 必需 Slot 在候选 Standby 上不是 synced AND NOT temporary AND invalidation_reason IS NULL。
  11. Cutover Guardrail：旧写入口仍有连接或写事务；新库 Sequence 未对齐；验证 Marker 未到达。
  12. 长期趋势：WAL/秒、初始 COPY 吞吐、Apply 吞吐、峰值事务大小和目标端写放大容量趋势。

18. 常见错误与反模式

#错误或反模式为什么危险正确做法
1把逻辑复制当成整个数据库镜像DDL、Sequence、Large Object、权限和多数对象不自动复制建立对象清单，Schema/Sequence/Extension 独立迁移
2提前数周创建 Slot，Consumer 尚未上线Slot 从创建后就可能保留 WAL在可控窗口创建；设置 WAL 预算、Owner 和失效策略
3无一致基线却使用 copy_data=false历史缺口不会被增量自动补齐使用 Slot/Snapshot 协调的 Backfill，或重新初始同步
4Publisher 先做 Rename/Drop/Type Change新 WAL 解码后可能无法映射目标 SchemaSubscriber-First 的 Expand/Contract；破坏性步骤最后执行
5UPDATE/DELETE 表没有可用 Replica IdentityPublisher 不能发布或 Subscriber 无法高效定位行主键优先；必要时 USING INDEX；谨慎使用 FULL
6所有表统一设置 REPLICA IDENTITY FULLWAL、网络、内存和目标端匹配成本增加按表评估并用窄、稳定、非空唯一键
7Subscriber 同时接受业务写会产生 Unique Violation、Missing Row 和来源冲突单写权威；迁移期严格 Fencing；多主需专门冲突协议
8只比较源端和目标端总行数相同 Count 仍可能错行、漏删、金额错误分桶、聚合、Checksum、抽样和业务不变量组合验证
9忘记 Sequence/Identity 当前值切流后第一批 INSERT 可能撞主键冻结写入后复制每个 Sequence 的 last_value/is_called
10Consumer 收到事件就 ACKSink 随后失败会形成不可恢复缺口Sink 事务与 Checkpoint 原子提交后才确认 LSN
11宣称 Exactly-Once，但没有端到端事务网络重试和 Commit 结果未知必然制造重复窗口At-Least-Once + 稳定幂等键；明确原子性边界
12多个独立 Consumer 共用一个 Slot一个进度无法代表多个独立持久化状态每个独立消费进度使用独立 Slot，或先进入共享 Durable Log
13把 LSN 当成跨 Cluster 全局顺序LSN 只属于特定 PostgreSQL WAL 空间携带 system_identifier + timeline/slot + LSN；跨源用业务顺序协议
14用无限内存缓存大事务大事务或 Sink 变慢会 OOMBounded Queue、事务上限、Durable Spool 和 Backpressure
15遇到冲突直接 ALTER SUBSCRIPTION ... SKIP跳过整个事务，正常行也会丢失先还原事务内容和业务意图；优先修数据后重放
16切流只改 DNS，不隔离旧连接长连接仍能向旧库提交数据库权限、代理路由、应用开关和连接驱逐多层 Fencing
17新库开始写后仍宣称可“立即回滚”目标端独有写入不会自动回到旧库反向 CDC/Outbox、写入冻结或补偿流水；定义 Point of No Return
18假定逻辑 Slot 会自动跟随物理主备切换PG17 前无内建 Failover Slot；PG17+ 也需完整配置和验证配置 failover=true、Slot Sync、候选 Standby 与 Fencing 演练
19跨版本默认启用 binary=true类型二进制格式、架构和版本兼容性风险更高默认文本协议；只有验证全部类型后才启用 Binary
20为“加速迁移”关闭 fsync、约束或 Autovacuum损害耐久性、数据正确性和长期健康做容量规划、并发控制、索引优化和阶段化 Backfill
21Row Filter 被当成安全隔离边界Publication 组合、权限和新表变更可能扩大数据集合安全边界用权限/独立数据库；Filter 只做复制选择
22REFRESH PUBLICATION 被当成自动重抄已 Ready 表的过滤变化不会自动完成新基线对过滤变化显式设计 Resnapshot/Backfill 和校验

19. 模拟生产事故案例

19.1 模拟生产案例一：孤儿 Slot 持续保留 WAL，Publisher 磁盘即将耗尽

1. 系统背景

一个 6 TB OLTP Cluster 每小时高峰产生约 180 GB WAL。审计 CDC 使用独立 Logical Slot audit_v2。Consumer 服务在迁移到新平台后被下线，但变更单只删除了 Deployment，没有删除 Slot，也没有监控 Slot Owner 和 WAL Runway。max_slot_wal_keep_size 未设置有效预算。

2. 故障现象

48 小时后，pg_wal 文件系统使用率从 55% 升至 94%；Checkpoint 写入变慢，业务提交 P99 上升。应用团队看到 active=false，误以为 Slot 已停止且不会影响数据库。

3. 错误假设

  • “Inactive Slot 不会保留 WAL。”
  • “归档成功后可以直接删除 pg_wal 中旧文件。”
  • “重启 PostgreSQL 会清掉 Slot。”

三个假设都错误：永久 Slot 跨重启存在；只要 restart_lsn 未前进，所需 WAL 就不能正常回收；手工删除 pg_wal 文件会破坏 Cluster。

4. 排查过程

先计算每个 Slot 的保留量：

SELECT slot_name,
       active,
       active_pid,
       restart_lsn,
       confirmed_flush_lsn,
       pg_size_pretty(
           pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)
       ) AS retained,
       wal_status,
       safe_wal_size,
       invalidation_reason
FROM pg_replication_slots
ORDER BY pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) DESC;

audit_v2 为 active=false，保留量约 8.1 TB，远高于其他 Slot。CMDB 显示其 Consumer 已在两天前退役。再检查日志和部署平台，确认没有合法消费者准备从该 Slot 恢复；审计目标已由新的独立流水线完整接管。

5. 根因

根因不是 WAL Writer 或 Checkpoint 本身，而是 Slot 生命周期与应用生命周期脱钩：没有 Owner、删除审批、WAL 预算、Inactive 告警，也没有在 Consumer 退役流程中调用 Slot 清理。

6. 临时止损

  1. 立即停止非关键高 WAL 批处理和大规模索引重建。
  2. 在线扩容 pg_wal 文件系统，获得决策时间。
  3. 再次确认新审计链路的 Checkpoint、数据完整性和合规保留。
  4. 经数据库、审计和业务 Owner 联合批准后执行：
SELECT pg_drop_replication_slot('audit_v2');
  1. 观察 Checkpoint 后旧 WAL 被正常回收，禁止手工删除文件。

若 Consumer 仍需恢复，正确止损是恢复 Consumer 或重建后 Resnapshot，而不是擅自推进 Slot。

7. 最终修复

  • Slot 创建必须登记 Owner、Consumer、数据分类、最大停机时间和删除条件；
  • 将 retained_bytes、WAL 速率和磁盘余量组合为 Runway；
  • 为业务设置可接受的 max_slot_wal_keep_size，明确超过预算后 Slot 可能失效并需要重新基线；
  • [PG18] 评估 idle_replication_slot_timeout 作为孤儿 Slot Guardrail，但必须配套失效告警和 Resnapshot Runbook；
  • Consumer 删除流水线新增“先停写验证、后删除 Slot”的强制 Gate。

8. 监控补充

新增：Slot Inactive 超时、Retention 连续增长、WAL Runway 低于 6 小时/2 小时分级告警、invalidation_reason 非空告警、Slot Owner 缺失告警，以及每日 Slot 清单审计。

9. 如何防止复发

每季度做一次“Consumer 消失”演练；验证告警能在磁盘受威胁前触发，值班人员能够识别 Slot、找到 Owner、恢复 Consumer 或按审批重建基线。基础设施代码将 Slot 与 Consumer 资源绑定，禁止手工创建无标签永久 Slot。

19.2 模拟生产案例二：大版本切流遗漏 Sequence，对新库开放写入后出现主键冲突

1. 系统背景

团队使用 Logical Replication 将 PostgreSQL 16 迁移到 PostgreSQL 18。orders.order_id 是 bigserial 主键。全量 COPY、增量追平、行数和金额校验均通过。切流 Runbook 只检查表数据，没有 Sequence 清单，也没有反向复制。

2. 故障现象

切到 PG18 后，第一批新订单间歇性报：

duplicate key value violates unique constraint "orders_pkey"
Key (order_id)=(41827391) already exists.

有人建议立即把流量切回 PG16，但 PG18 已成功写入部分支付状态和新订单，直接回切会丢失这些 Target-Only Writes。

3. 错误假设

  • “主键值复制过去了，所以 Sequence 也会跟着走。”
  • “切流后保留旧库就等于随时能回滚。”
  • “把 Sequence 随意加一个很大的数即可解决。”

逻辑复制复制行中的 order_id 值，不复制 Sequence 自身状态；没有反向增量或补偿账本时，新库开始写入就是回滚边界。

4. 排查过程

先同时 Fencing 两端新写，保留失败请求和已提交订单清单。检查目标端表最大值和 Sequence：

SELECT max(order_id) AS max_order_id
FROM orders;

SELECT last_value, is_called
FROM orders_order_id_seq;

目标 Sequence 的 last_value 仍接近 Schema 初始化值，而表中已有四千多万行。再在旧库读取最终冻结时的 Sequence 状态，确认它高于表最大值，因为历史回滚和缓存产生了间隙。

同时查询 PG18 切流后事务，按 Cutover Marker、应用请求 ID 和审计表列出 Target-Only Writes，确认不能无损直接回旧库。

5. 根因

迁移对象模型只覆盖 Relation，没有覆盖 Sequence Boundary；Cutover Gate 缺少 Sequence/Identity 对齐和 Smoke Insert。回滚方案也只写了“改回连接串”，没有处理目标端独有写入。

6. 临时止损

  1. 保持旧库只读和新库写入冻结，防止继续分叉。
  2. 以旧库冻结时读取的精确状态对齐目标 Sequence：
-- 参数来自旧库最终冻结点；示例值仅用于说明
SELECT setval('orders_order_id_seq', 41839120, true);
  1. 在事务外做受控 Smoke Insert，确认生成值大于所有已存在键；再删除测试行。Sequence 的推进通常不会因事务回滚而回退，因此必须记录测试消耗的值。
  2. 校验 Target-Only Writes 完整无误后，仅重新开放 PG18 写入。
  3. 若业务决定回滚到 PG16，必须先通过反向 CDC、Outbox/审计流水或定制补偿把 PG18 独有提交送回旧库，再验证并切换；不能直接改连接串。

7. 最终修复

  • 自动发现所有 Serial/Identity 所依赖 Sequence，并生成源/目标对照清单；
  • 在最终写冻结、增量追平之后读取源 Sequence 的 last_value 与 is_called，原样设置目标；
  • 对不由表列拥有的业务 Sequence 单独登记；
  • Cutover Gate 新增：旧连接 Fencing、最终 LSN、Sequence 对齐、读写 Smoke Test、Target-Only Write 记录和 Point of No Return；
  • 为回滚窗口建立 PG18→PG16 的受控反向变更流，或明确切流后只能 Forward Fix。

8. 监控补充

新增 Sequence Headroom 指标：对关键表比较下一可用 Sequence 值与表中最大键；监控主键 Unique Violation；切流后按旧/新 Cluster 分组统计写入量；任何旧库写入都触发 Fencing 告警。

9. 如何防止复发

每次迁移演练必须在与生产同规模的 Sequence 间隙、缓存设置和并发下执行。Runbook 由工具生成对象清单而不是手写；Go/No-Go 必须有数据库、应用和业务三方签字。切流后的回滚声明必须写清“最后可无损回滚时点”和目标端独有写入的处理机制。

20. 面试题

20.1 核心概念题 1：Logical Replication 与 Physical Replication 的本质区别是什么？

题目：两者都读取 WAL，为什么一个能跨大版本，另一个通常不能？

30 秒回答：Physical Replication 复制块级 WAL，要求存储格式和系统 Catalog 高度一致，主要用于同一大版本的 HA；Logical Replication 在 Publisher 上把 WAL 解码成按 Replica Identity 描述的行变化，再由 Subscriber 映射到本地表，因此可以选择表、行、列并支持跨大版本，但不自动复制 DDL、Sequence 和整个 Cluster 状态。

深入回答：原理上，物理复制重放 Page/Block 变化，保真度高、覆盖对象广；逻辑复制经 pgoutput 发送事务和行级协议。场景上，同版本灾备优先物理复制，跨版本升级、单表迁移和 CDC 优先逻辑复制。逻辑复制优点是细粒度和异构性，缺点是 Apply 开销、Schema/Sequence 边界和冲突；物理复制优点是完整、低语义转换，缺点是版本与 Cluster 形态受限。替代方案包括 pg_upgrade、Dump/Restore、应用级 Outbox 和 ETL。生产上两者常组合：物理复制负责 Publisher HA，逻辑复制负责迁移或下游分发。

面试官真正考察什么：能否区分“WAL 来源相同”与“复制语义相同”，并按 HA、升级、CDC 场景选型。

常见错误回答：“逻辑复制是异步，物理复制是同步。”两者都可以是异步，物理复制也可配置同步；同步性不是本质差异。

追问：Logical Replication 能否替代 Backup/PITR？

追问答案：不能。错误 DELETE、恶意修改或应用 Bug 会被同步；逻辑复制不保留任意历史时间点，也不覆盖全部对象。必须保留 Base Backup、WAL Archive 和恢复演练。

20.2 核心概念题 2：Publication、Subscription、Slot 与 Replication Origin 分别负责什么？

题目：请用一次事务的路径解释四者边界。

30 秒回答：Publication 定义 Publisher 愿意发布哪些表、操作、行和列；Subscription 定义 Subscriber 如何连接、订阅哪些 Publication 以及 Apply 参数；Logical Slot 在 Publisher 保存解码与消费进度并保留所需 WAL；Replication Origin 在 Subscriber 记录远端位置到本地提交位置的映射，并支持来源追踪和环路控制。

深入回答：Publication 是选择规则，不保存 Consumer Checkpoint；Subscription 是控制面对象，会驱动 Apply/Table Sync Worker；Slot 是 Publisher 数据面状态，一个活动 Slot 同时只能由一个消费者使用；Origin 属于 Subscriber 的应用进度，不是 WAL 保留点。优点是职责分离、可组合多个 Publication；缺点是对象分散，运维容易只删 Subscription 却遗留外部 Slot，或只看 Slot 忽略 Origin。替代方案是外部 Connector 自管 Slot 和 Checkpoint。生产上必须用拓扑清单把四者与 Owner、版本、回滚窗口绑定。

面试官真正考察什么：是否真正理解控制面、数据面和进度状态，而不是只会背建链 SQL。

常见错误回答：“Publication 保存复制进度，Subscription 保存 WAL。”两者都不正确。

追问：可以让两个独立 Consumer 共用一个 Slot 吗？

追问答案：不应。一个 Slot 只有一份确认进度且一次只能有一个活动消费者，无法表达两个独立 Sink 的耐久状态。应使用两个 Slot，或先写入共享 Durable Log，再由多个下游消费。

20.3 核心概念题 3：restart_lsn 与 confirmed_flush_lsn 有什么区别？

题目：哪个决定 WAL 能否删除，哪个代表 Consumer 进度？

30 秒回答：confirmed_flush_lsn 是 Logical Consumer 已确认持久化的位置；restart_lsn 是 Slot 解码未来记录仍可能需要的最早 WAL。WAL 回收受 restart_lsn 约束，而不是简单受确认位置约束；两者之间可能因为长事务、解码状态或 Catalog 依赖产生差距。

深入回答：Consumer 发送 Standby Status Update 后，Slot 可推进确认位置；但 ReorderBuffer 可能仍需要更早的 WAL 来完成未提交事务，因此 restart_lsn 不能越过该边界。优点是即使事务跨多个 WAL Segment，解码仍完整；缺点是长事务可造成巨大 Retention。替代止损不是手工删 WAL，而是结束/拆分长事务、恢复 Consumer、提高预算或重新基线。生产上同时监控当前 LSN 到两个位置的字节差和 WAL Runway。

面试官真正考察什么：是否能解释 Slot 为什么在“Consumer 看似有进度”时仍保留大量 WAL。

常见错误回答：“两个字段应永远相等；不等就是 Bug。”

追问：confirmed_flush_lsn 已接近当前 LSN，但 restart_lsn 很旧，优先查什么？

追问答案：查长时间未提交事务、大事务 Streaming 状态、catalog_xmin、解码插件状态和版本已知问题；不要先推进或删除 Slot。

20.4 核心概念题 4：四种 Replica Identity 如何选择？

题目：解释 DEFAULT、USING INDEX、FULL、NOTHING 的语义和代价。

30 秒回答：DEFAULT 使用主键；USING INDEX 使用满足条件的唯一、非部分、非延迟且键列非空的索引；FULL 在 UPDATE/DELETE 中记录旧行所有列；NOTHING 不提供旧行身份，因此发布 UPDATE/DELETE 时通常不可用。优先主键，其次稳定窄唯一索引，FULL 只作经过测量的兜底。

深入回答：Replica Identity 决定 Publisher WAL 中携带什么旧值，以及 Subscriber 如何找到目标行。主键/专用唯一索引 WAL 小且查找快；FULL 无需键但增加 WAL/网络、比较和类型限制，宽表/TOAST 尤其昂贵；NOTHING 只适合仅 INSERT 或不发布 UPDATE/DELETE 的表。替代方案是补充代理键、只发布必要操作或在 CDC 中把变更转换为 Append-Only 事件。生产上两端键语义、Collation 和唯一约束必须兼容；Publisher 使用非 FULL 身份时，Subscriber 还必须设置由相同或更少键列组成的兼容 Replica Identity。

面试官真正考察什么：是否能把 WAL 体积、行定位、数据模型和并发性能联系起来。

常见错误回答：“没有主键就一律 FULL，几乎没有代价。”

追问：表没有主键，但有稳定的业务唯一键，怎么做？

追问答案：创建满足要求的唯一索引并 ALTER TABLE ... REPLICA IDENTITY USING INDEX ...；验证键列不变、非空、目标端有相同查找能力。若业务键会更新，应评估旧/新键语义或增加代理键。

20.5 核心概念题 5：Initial Table Sync 的完整状态机是什么？

题目：为什么 COPY 期间 Publisher 仍可写且最终不漏数据？

30 秒回答：Subscription 为每张待同步表启动 Table Sync Worker，建立一致性 Snapshot 和临时同步 Slot，按 Snapshot 执行 COPY；同时 Publisher 的后续变化由 Slot 保留。COPY 完成后 Worker 流式追赶 Snapshot 之后的变化，到达与主 Apply Worker 协调的同步点，再把表交还主 Apply Worker 持续复制。

深入回答：Snapshot 固定基线可见集合，Slot 固定增量起点，二者组合避免“先扫表后开增量”的缺口。并行 Table Sync 提高吞吐，但会增加源端扫描、网络、目标 WAL、索引和连接压力。替代方案是外部 Backfill 工具加 Slot/Snapshot 协调。生产上要监控 pg_subscription_rel、同步 Slot、Worker 配额、触发器语义和目标端冲突。Publication 的 publish 操作列表不决定 Initial Copy 是否复制已有行；Row Filter/Column List 则需按版本和真实测试验证。

面试官真正考察什么：能否解释 Snapshot、COPY、Catch-up 和 Handoff 的一致性证明。

常见错误回答：“先 COPY 全表，COPY 完才创建 Slot。”这会丢掉两者之间的变化。

追问：COPY 中断后必须从零重来吗？

追问答案：内置 Table Sync Worker 会由 Apply Worker 重新拉起并恢复该表同步流程；具体是否重做 COPY 取决于状态和失败点。运维应看 srsubstate 与日志，不应手工伪造 Ready 状态。

20.6 原理/排障题 1：Logical Slot 导致 pg_wal 暴涨时如何排查？

题目：请给出判断顺序，而不是只说“删 Slot”。

30 秒回答：先按 Slot 计算当前 LSN 到 restart_lsn 和 confirmed_flush_lsn 的字节差，检查 active、wal_status、safe_wal_size、invalidation_reason；再判断 Consumer 是否停止、Sink 是否慢、网络是否断、长事务是否阻止 restart_lsn、Slot 是否孤儿，并用磁盘余量除以峰值 WAL 速率计算 Runway。

深入回答：Consumer 慢通常让确认和重启位置一起落后；长事务可能让确认靠前但重启位置很旧；孤儿 Slot 则 active=false 且 Retention 持续增长。低风险止损是恢复 Consumer、修网络、扩容、暂停高 WAL 作业；删除 Slot 只适用于明确接受重新基线。优点是保留可恢复性，缺点是磁盘成本。替代是设置 WAL 预算和 [PG18] Idle Slot Guardrail。生产上绝不能直接删除 pg_wal 文件。

面试官真正考察什么：是否能在数据完整性与可用性之间做有证据的应急决策。

常见错误回答：“重启数据库即可释放 WAL。”永久 Slot 跨重启存在。

追问：磁盘只剩 20 分钟，但 Owner 联系不上怎么办？

追问答案：按预先批准的紧急策略执行：停止非关键 WAL、在线扩容、隔离批处理；若仍无法避免主库停写，依据数据分级和 Slot 最大保留 SLO决定是否删除并强制 Resnapshot。没有预案时应升级 Incident Commander，而不是个人静默删 Slot。

20.7 原理/排障题 2：Subscriber 冲突中哪些会停止，哪些会跳过？

题目：请以 PG18 的冲突类型说明。

30 秒回答：insert_exists、update_exists、multiple_unique_conflicts 等唯一约束冲突会报错并停止 Apply；update_missing 和 delete_missing 会记录冲突但跳过该变化继续；启用 track_commit_timestamp 后可识别 update_origin_differs、delete_origin_differs，当前仍会应用远端变化。其他约束、权限或 RLS 问题也可能停止。

深入回答：停止型冲突保护约束但积压 Slot；跳过型冲突保持流水线前进，却可能掩盖 Subscriber 已分叉。解决方式包括 Remote-Wins、Local-Wins 加补偿、修权限/Schema 后重放，或最后才 SKIP 整个事务。优点是内置日志和 [PG18] 统计提高可观测性；缺点是没有自动业务冲突合并。替代是 Subscriber 只读、应用层冲突协议或专门多主系统。生产上保存 Finish LSN、远端事务和目标现状。

面试官真正考察什么：是否知道“没报错”不代表没有数据不一致。

常见错误回答：“所有冲突都会停，所以只要 Subscription Running 数据就一致。”

追问：为什么 ALTER SUBSCRIPTION ... SKIP 很危险？

追问答案：它跳过的是整个远端事务，事务内不冲突的其他表/行也不会应用。必须先评估事务原子性影响并有补偿或校验计划；streaming=parallel 时失败日志还可能不含 Finish LSN，需要调整模式复现获取。

20.8 原理/排障题 3：如何做零/低停机 Schema Evolution？

题目：Publisher 和 Subscriber 的 DDL 顺序如何安排？

30 秒回答：使用 Expand/Contract：先在 Subscriber 增加向后兼容结构，再在 Publisher 增加；应用进入双读/双写或兼容读，执行可恢复 Backfill，验证后切换到新字段，最后停止旧写并在所有消费者退出兼容窗口后删除旧字段。Rename/Type Change 通常用“新列 + 回填 + 切换”而不是原地破坏。

深入回答：逻辑复制不传 DDL，Publisher Schema 一旦先变化，新的 Relation/行数据可能无法映射目标。Subscriber-First 的优点是不会阻断 Apply，缺点是兼容期对象增多、应用复杂；替代是短暂停写同时变更、重建 Subscription 或蓝绿表。生产上要管理默认值、Generated Column、Column List、Row Filter、Replica Identity、Backfill WAL 和下游 Schema Registry，不能只看 PostgreSQL 两端。

面试官真正考察什么：是否把数据库、应用和 CDC 契约视为同一个兼容性窗口。

常见错误回答：“DDL 很快，直接两端依次执行即可。”即使单条 DDL 快，顺序和锁仍可能让 Apply 停止。

追问：如何安全 Rename 一列？

追问答案：新增新列；Subscriber 先加；Publisher 加列并由应用双写或触发受控同步；Backfill；消费者切读新列；停止旧写；验证；最后删除旧列。原地 Rename 只适合能冻结写并原子协调所有端的窗口。

20.9 原理/排障题 4：为什么 CDC 通常是 At-Least-Once，如何实现幂等？

题目：请解释 Commit Outcome Unknown。

30 秒回答：Consumer 只有在 Sink 持久化后才应确认 LSN。若 Sink 已提交但 ACK 尚未到 Publisher 就崩溃，重启会再次收到同一事务，因此至少一次是自然语义。使用稳定事件键，例如 source_system + slot + transaction_end_lsn + event_index，让业务写和 Checkpoint 在同一 Sink 事务中提交，即可吸收重复。

深入回答：先 ACK 会丢数据，先 Sink Commit 会有重复窗口；这是网络与两个独立状态机间无法用单次消息消除的两难。幂等落库、Upsert、去重表、Outbox/Inbox 是常见方案。优点是故障恢复简单且不丢，缺点是 Sink 需要幂等状态和保留策略。替代是支持源 Offset 与消息事务原子提交的平台，但外部邮件/支付等副作用仍需业务幂等。生产上 Checkpoint 不能与事件分开提交。

面试官真正考察什么：是否能从分布式提交时序推导语义，而不是口号式宣称 Exactly-Once。

常见错误回答：“收到事件后记录一个内存布尔值即可去重。”进程重启即丢失。

追问：幂等键为什么不能只用 XID？

追问答案：XID 会循环复用，且跨 Cluster 不唯一；一个事务还有多条事件。至少组合源 Cluster 身份、事务 End LSN 和事件序号，或使用业务全局事件 ID。

20.10 原理/排障题 5：Large Transaction Streaming、Parallel Apply 与 Two-Phase Replication 有何区别？

题目：按 PG14–18 版本边界回答。

30 秒回答：[PG14+] Protocol V2 可在大事务尚未提交时分段传输；[PG15+] Two-Phase 可复制 PREPARE/COMMIT PREPARED 语义；[PG16+] Protocol V4/streaming=parallel 可由并行 Apply Worker 处理流式事务；[PG18] Subscription 的 streaming 默认改为 parallel。它们分别解决传输等待、2PC 语义和 Apply 并行，不等于提前对下游暴露未提交数据。

深入回答：Streaming 降低 Publisher 解码端等待和部分内存压力，但 Consumer 仍需按 XID 暂存并在 Commit 后发布；Parallel Apply 增加吞吐，也带来 Worker/内存/冲突诊断复杂度；Two-Phase 保留 Prepared Transaction 边界，但会把长期 Prepared 的运维风险传到下游。替代是限制事务大小、应用分批、Outbox。生产上协议版本、Library API、Subscriber 参数必须一致，不能只改 streaming。

面试官真正考察什么：是否能区分网络协议、执行并行和事务可见性。

常见错误回答：“Streaming 会把未提交行直接交给业务使用。”

追问：为什么教学 Go Consumer 选择 Protocol V2 而不是 V4？

追问答案：它使用当前 pglogrepl.ParseV2 公共接口并实现单线程事务提交语义。启用 V4 必须实现 Parallel Streaming 消息和状态管理；只改参数会造成协议解析错误或错误确认。

20.11 原理/排障题 6：Logical Slot Failover 的版本和配置边界是什么？

题目：物理 Standby Promotion 后，CDC 为什么可能断？

30 秒回答：内建 Logical Slot Failover 从 PG17 开始。需要 Publisher 有物理 Standby，Subscription/Slot 标记 failover=true，Standby 配置 Slot Synchronization，并在 Promotion 前验证必需 Slot 在候选 Standby 上 synced、非 Temporary、无 Invalidation；还要保证 Standby WAL 位置不落后于 Subscriber 已确认进度。

深入回答：普通 Logical Slot 是主库本地持久状态，物理 Promotion 若没有同步 Slot，新主不知道 Consumer 的解码点。PG17+ 的优点是可平滑继续，缺点是配置链更长、同步异步且需要 Fencing。PG17 之前的替代是外部 Slot 管理、从已知 Checkpoint 重建并可能 Resnapshot，或让 CDC 从 Durable Log 接续。生产上必须演练 Promotion，验证 synchronized_standby_slots、Slot 列表、WAL 保留和连接重定向。

面试官真正考察什么：是否把“数据库可提升”与“逻辑消费者可无缝续传”区分开。

常见错误回答：“物理复制会自动复制所有 Slot，所以任何版本都没问题。”

追问：候选 Standby 已有同名 Slot，但 synced=false，可以提升吗？

追问答案：不能把它视为 Failover Ready。同名不证明状态完整；提升后 Consumer 可能无法续接或发生缺口。应等待同步并验证，或进入明确的重建/重新基线故障方案。

20.12 架构设计题 1：设计一次 PostgreSQL 16 到 18 的低停机大版本升级

题目：给出阶段、Go/No-Go 与回滚边界。

30 秒回答：先做兼容性和对象清单，建立 PG18 Schema、Extension、权限和参数；PG16 建 Publication，PG18 建 Subscription 完成 Initial Copy；持续复制期间做分层校验和应用兼容测试；切流时 Fencing 旧写、等待最终 Marker/LSN 应用、对齐所有 Sequence、切连接并 Smoke Test；保留旧库只读和 Slot 到回滚窗口结束。新库产生独有写入后，回滚必须靠反向同步或补偿，不是简单改 DNS。

深入回答：原理是 Snapshot + Incremental 把长停机缩成最终冻结窗口。优点是可长时间验证、跨大版本；缺点是双环境成本、DDL/Sequence 边界、初始 COPY 负载和回滚复杂。替代是 pg_upgrade（停机短但同机/存储与回退约束不同）、Dump/Restore（简单但停机长）或托管升级。生产上要测试 Extension、Collation、Planner 变化、Generated Columns、Client Driver、Failover 和 Backup/PITR。

面试官真正考察什么：是否有完整迁移生命周期，而不只是 CREATE SUBSCRIPTION。

常见错误回答：“Lag 为 0 后改 DNS即可。”还缺 Fencing、Sequence、旧连接、对象校验和目标独有写入处理。

追问：如何定义 Point of No Return？

追问答案：当 PG18 接受了无法可靠反向复制/补偿的首个业务提交时，直接回旧库不再无损。应以 Cutover Marker 和写入审计记录该时点，并明确之后只允许 Forward Fix，或先启动反向链路。

20.13 架构设计题 2：设计一个生产级 PostgreSQL CDC 平台

题目：至少覆盖 Snapshot、增量、可靠性、Schema 和运维。

30 秒回答：每个源建立受管 Slot 和 Connector；用 Slot 一致点协调并行 Snapshot；增量按源事务写入 Durable Log；Offset/LSN 仅在 Durable Log 提交后确认；事件带源身份、XID、End LSN、序号和 Schema Version；Schema Registry 管理兼容性；Sink 用幂等 Inbox/Upsert，平台提供 Backpressure、磁盘 Spool、重放、DLQ、校验、Slot Failover 和端到端监控。

深入回答：控制面管理源、Publication、Slot、Owner、版本和 ACL；数据面解析 pgoutput、维护 Relation/Type Cache、事务 Buffer 与顺序；存储面将原始事件和 Checkpoint 原子持久化。优点是复用和可观测，缺点是高复杂度、WAL 风险和 Schema 治理成本。替代是 Debezium/Kafka Connect、云 DMS 或应用 Outbox。生产上要明确 At-Least-Once、数据保留、PII、加密、租户隔离、灾备和 Resnapshot SLO。

面试官真正考察什么：是否理解 Connector 代码只占平台的一小部分。

常见错误回答：“起一个无限循环读 WAL，再 HTTP POST 给下游。”缺少事务、重试、去重、Backpressure 和恢复边界。

追问：如何保证一个源事务跨多表事件不被下游看到一半？

追问答案：在 Durable Log 中保留事务边界，按 XID 暂存，只有 Commit 后才发布；或写入带 Begin/Commit Marker 的事务批次，并要求 Sink 原子消费。超大事务需要 Durable Spool，不能无限内存缓存。

20.14 架构设计题 3：能否用内建 Logical Replication 做双向多主？

题目：请给出可行边界和拒绝条件。

30 秒回答：技术上可以搭双向 Subscription，并用 Replication Origin/origin=none 等机制减少环路，但内建 Logical Replication 不提供业务冲突自动合并。只有写集合严格分区、主键全局唯一、Schema 同步且 Fencing 明确时才可考虑；同一行可在两端并发更新的通用多主不应直接这样做。

深入回答：环路过滤解决“重复转发”，不解决同键 INSERT、Last-Writer-Wins、删除与更新竞态、Sequence 冲突和跨节点约束。优点是组件少、可按表；缺点是冲突语义脆弱、故障恢复困难。替代是单主 + 物理 HA、按租户/区域 Sharding、应用层 Command Routing、专门多主扩展或分布式数据库。生产上要有全局 ID、写归属、冲突日志、隔离测试和脑裂 Fencing。

面试官真正考察什么：是否把“能形成环”与“能正确多主”区分开。

常见错误回答：“两边互相订阅就是 Active-Active，PostgreSQL 会自动选最新。”

追问：同一行两端同时更新，内建逻辑复制如何决定胜者？

追问答案：没有通用业务胜者协议；到达顺序、约束和本地状态可能导致覆盖或冲突。必须在应用定义 Version/Owner/冲突解决，或从架构上禁止同一 Key 多点写。

20.15 架构设计题 4：设计 PostgreSQL 到搜索/湖仓的跨系统同步

题目：目标端没有 PostgreSQL 事务和约束，怎么保证可恢复？

30 秒回答：先以 Slot 一致点做分片 Backfill，并记录每批 Schema Version 和范围；同时将 CDC 事务写入 Durable Log；Backfill 完成后按主键/版本合并增量。事件包含 UPSERT/DELETE 语义、源 End LSN、业务版本和 Schema Version；Sink 用幂等 Document ID 或 Merge Key，维护独立 Checkpoint、重试和 Poison Queue；切换前做分桶校验和最终 Marker。

深入回答：异构目标不一定支持跨事件原子事务，因此要把源事务边界作为元数据，并按业务容忍度选择原子批、最终一致或补偿。优点是解耦查询负载，缺点是类型映射、删除、乱序、回放成本和不可见约束。替代是查询联邦、批量 ETL、应用 Outbox。生产上要处理 Decimal/Time Zone/JSON、TOAST、TRUNCATE、Schema Evolution、重建索引、限流和 GDPR 删除。

面试官真正考察什么：是否能把 PostgreSQL 的事务语义翻译成目标系统可执行的契约。

常见错误回答：“按收到顺序写搜索引擎即可，失败就无限重试。”并发、重放和 Poison Event 会造成乱序与阻塞。

追问：Backfill 期间某行被更新两次，如何避免旧快照覆盖新 CDC？

追问答案：每个目标记录保存源顺序标记，例如同一源的 Commit/End LSN 或业务版本；Upsert 只接受不旧于当前版本的事件。更稳妥的是先把 CDC 进入 Durable Log，再按协调的一致点合并 Backfill 与增量。

21. 课后练习与完整参考答案

21.1 理论题 1：证明 Initial Snapshot 与 Slot 为什么必须协调

题目：方案 A 在 10:00 开始普通全表导出，10:30 导出完成，10:31 才创建 Logical Slot；方案 B 先在一致点 L0 创建 Slot/导出 Snapshot，再用该 Snapshot 做 Backfill，最后从 L0 消费。比较两者正确性。

参考答案：

  • 方案 A 存在 10:00—10:31 的边界问题。普通导出的 Snapshot 只看到自身可见集合，而 Slot 在 10:31 才开始保留增量；10:00 后提交、但未被导出 Snapshot 看见且发生在 Slot 创建前的变化可能永久遗漏。即使最终行数接近，也无法证明无缺口。
  • 方案 B 用同一一致点定义两部分：Snapshot 覆盖 L0 之前应见的基线，Slot 保留 L0 之后的增量。Backfill 慢只会增加待追赶 WAL，不会产生中间空窗。
  • 正确性不变量为：
目标最终状态 = Snapshot(L0) + CommitLSN > L0 的全部已提交变化
  • 生产上还要保证 Snapshot 生命周期、Slot WAL 预算、Backfill 分片、删除语义、Schema Version 和最终校验。

21.2 理论题 2：计算 Slot WAL Runway 与追赶时间

题目：Publisher 的 pg_wal 文件系统可用 600 GiB，Consumer 完全停止时源端峰值产生 80 MiB/s WAL。当前已积压 300 GiB。Consumer 恢复后可持续处理 140 MiB/s，而源端仍产生 80 MiB/s。忽略波动，回答：

  1. Consumer 不恢复时，剩余磁盘约能支撑多久？
  2. 恢复后清空 300 GiB Backlog 约需多久？
  3. 为什么生产告警不能直接使用这两个理想值？

参考答案：

  1. 600 GiB × 1024 MiB/GiB ÷ 80 MiB/s = 7680 s，约 128 分钟。
  2. 净追赶速率为 140 - 80 = 60 MiB/s；300 × 1024 ÷ 60 = 5120 s，约 85.3 分钟。
  3. 实际 WAL 速率有峰值，Consumer 吞吐受网络、Sink、长事务、Checkpoint 和锁影响；文件系统还需要安全余量，restart_lsn 也可能受长事务限制而不随确认位置立即前进。因此应使用时间序列 P95/P99、保守安全系数和“修复 Lead Time”，而非单点平均值。

21.3 理论题 3：为无主键宽表选择 Replica Identity

题目：表 customer_profile 有 80 列和多个大 JSONB 字段，无主键；业务上 (tenant_id, customer_no) 唯一且永不为空。需要发布 UPDATE/DELETE。如何设计？

参考答案：

首选建立窄、稳定的唯一索引：

CREATE UNIQUE INDEX CONCURRENTLY customer_profile_replica_uq
ON customer_profile(tenant_id, customer_no);

ALTER TABLE customer_profile
REPLICA IDENTITY USING INDEX customer_profile_replica_uq;

执行前验证无重复、列均 NOT NULL；在 Subscriber 建立语义一致的唯一索引，并把它设置为兼容 Replica Identity。这样 WAL 只需携带键而非整行，目标定位成本稳定。REPLICA IDENTITY FULL 可作为短期过渡，但宽 JSONB 会显著增加 WAL/网络/内存，且目标匹配和类型操作类存在边界。长期替代是增加不可变代理主键。

21.4 理论题 4：设计新增 NOT NULL 列的兼容顺序

题目：Publisher/Subscriber 都有 orders，要新增 region_code text NOT NULL，旧应用暂时不会提供该值。要求不中断逻辑复制。

参考答案：

采用 Expand/Contract：

  1. Subscriber 先 ADD COLUMN region_code text，保持 Nullable。
  2. Publisher 同样增加 Nullable 列；旧应用仍可写。
  3. 新应用开始写 region_code，并为旧行做限速、可恢复 Backfill。
  4. 两端验证 region_code IS NULL 为零；可先添加 CHECK (region_code IS NOT NULL) NOT VALID，再 VALIDATE CONSTRAINT，降低长时间强锁风险。
  5. 所有写路径兼容后，两端将列收紧为 NOT NULL，顺序仍应保证 Subscriber 能先接受 Publisher 新数据。
  6. 更新 Column List、CDC Schema Registry、校验任务和回滚脚本。

不要在 Publisher 一步增加没有兼容默认值的强制非空列，也不要假定 DDL 会自动复制。

21.5 理论题 5：分析 CDC 的四个崩溃点

题目：Consumer 的流程为“读事务 → 写 Sink 事件 → 写 Checkpoint → ACK LSN”。分别讨论崩溃发生在 Sink Commit 前、Commit 后 ACK 前、ACK 后、Commit 返回网络错误时的结果。

参考答案：

  • Commit 前：事件和 Checkpoint 都回滚；不 ACK；重启重放，无丢失。
  • Commit 后 ACK 前：Sink 已有事件和 Checkpoint；源端可能重发；幂等键吸收重复。
  • ACK 后：前提是事件和 Checkpoint 已同事务持久化，重启从 Checkpoint 后继续。
  • Commit 返回网络错误：结果未知，绝不能 ACK；重启先查持久化 Checkpoint/幂等键，已提交则去重，未提交则重做。

若 Checkpoint 与事件分开事务，可能出现“Checkpoint 前进但事件未落库”或“事件落库但 Checkpoint 未前进且无去重”两类错误。

21.6 实验题 1：验证 Row Filter、Column List 与 [PG18] Generated Column

题目：在 PG18 上设计 Publication，只发布 tenant_id = 42 的订单、列 order_id/tenant_id/amount/amount_with_tax，其中 amount_with_tax 为 Stored Generated Column。验证 Initial Copy 和增量行为。

参考答案：

Publisher：

CREATE TABLE lab22_filter.orders (
    order_id bigint PRIMARY KEY,
    tenant_id bigint NOT NULL,
    amount numeric(12,2) NOT NULL,
    note text,
    amount_with_tax numeric(12,2)
        GENERATED ALWAYS AS (round(amount * 1.10, 2)) STORED
);

INSERT INTO lab22_filter.orders(order_id, tenant_id, amount, note)
VALUES (1, 42, 100, 'included'),
       (2, 7, 200, 'excluded');

CREATE PUBLICATION lab22_filter_pub
FOR TABLE lab22_filter.orders
    (order_id, tenant_id, amount, amount_with_tax)
    WHERE (tenant_id = 42)
WITH (publish_generated_columns = 'stored');

Subscriber 为 amount_with_tax 使用普通列，才能接收 Publisher 发布的 Generated 值：

CREATE TABLE lab22_filter.orders (
    order_id bigint PRIMARY KEY,
    tenant_id bigint NOT NULL,
    amount numeric(12,2) NOT NULL,
    amount_with_tax numeric(12,2)
);

建立 Subscription 后，预期只复制 order_id=1，且没有 note。再执行：

UPDATE lab22_filter.orders SET tenant_id = 42 WHERE order_id = 2;
UPDATE lab22_filter.orders SET tenant_id = 7  WHERE order_id = 1;

Row Filter 边界迁移应分别表现为 INSERT 与 DELETE 语义。诊断 pg_subscription_rel、Subscriber 行、Publication Catalog 和冲突统计。还应增加一个 Subscriber Generated 列版本，验证“Generated→Generated 时使用 Subscriber 自己的表达式”，不能发布值直接覆盖 Generated 列。

21.7 实验题 2：验证大事务 Streaming 与 Backpressure

题目：使用本章 Go Consumer，执行包含 200,000 行 INSERT 的单事务，分别把 MAX_TX_BYTES 设为 1 MiB 和足够大的值，观察行为。

参考答案：

  1. 创建 Slot、Publication，启动 Consumer，设置 QUEUE_CAPACITY=1。
  2. Session A：
BEGIN;
INSERT INTO lab22_cdc.account(account_id, owner_name, balance)
SELECT g, 'bulk-' || g, g::numeric
FROM generate_series(1000000, 1199999) AS g;
COMMIT;
  1. MAX_TX_BYTES=1048576 时，Consumer 应在事务超过上限后停止，且不推进该事务 End LSN；Publisher Slot Retention 增加。重启后仍会重放，证明硬上限不会静默丢数据。
  2. 提高上限后重启，事务在 STREAM COMMIT 后作为一个 Envelope 落到 Sink，Checkpoint 才推进。
  3. 记录 PostgreSQL 版本、WAL 速率、事务大小、Consumer RSS、Queue 使用率、Sink Commit P50/P95/P99、Slot Retention 和 Wait Event。不要伪造固定耗时。
  4. 结论：Protocol V2 Streaming 能提前传输分段，但示例仍按事务内存聚合；生产应改为 Durable Spool。

21.8 实验题 3：[PG17+] 演练 Logical Slot Failover

题目：准备一个 Publisher Primary、Physical Standby 和 Subscriber，演练计划内主备切换后 Subscription 继续。

参考答案：

  1. Primary/Standby 均为 PG17+，Primary 配置 Logical WAL、物理复制和 Slot Synchronization 所需参数。
  2. 创建 Subscription 时使用 failover=true，并保证对应主 Slot 标记可故障转移。
  3. 在候选 Standby 检查：
SELECT slot_name,
       synced,
       temporary,
       invalidation_reason,
       (synced AND NOT temporary AND invalidation_reason IS NULL)
           AS failover_ready
FROM pg_replication_slots
WHERE slot_type = 'logical';
  1. 只有全部必需主 Slot和已完成 COPY 的 Table Sync Slot 都 Ready，且 Standby WAL 位置满足条件，才进入 Switchover。
  2. Fencing 旧 Primary、确认物理同步、提升 Standby、把 Subscriber 连接重定向新 Primary。
  3. 写入带唯一 Marker 的事务，验证 Subscriber 只出现一次且 LSN 持续推进。
  4. 失败预案：若 Slot 未同步，不允许把“数据库 Promotion 成功”当成 CDC 成功；应回旧主或重建 Slot并 Resnapshot。

21.9 排障题 1：Inactive Slot、确认位置靠前但重启位置很旧

题目：观察到：

active=false
current_lsn - confirmed_flush_lsn = 200 MB
current_lsn - restart_lsn = 1.8 TB
catalog_xmin 长时间不变

给出最可能的方向和处置顺序。

参考答案：

确认位置靠前说明 Consumer 曾确认大量数据，但 restart_lsn 仍受更早解码状态约束。优先检查长事务/Prepared Transaction、解码中的大事务、Catalog Snapshot 和插件/版本问题；再确认 Consumer 为什么 Inactive。保存 pg_replication_slots、pg_stat_activity、pg_prepared_xacts、长事务和日志证据。不要手工推进 Slot。若长事务合法，等待/协调结束并观察 restart_lsn；若 Consumer 已退役，走 Slot 清理审批；若 Slot 已异常且无法恢复，重建并重新基线。

21.10 排障题 2：Publisher 加列后 Subscription 停止

题目：Publisher 执行 ALTER TABLE orders ADD COLUMN risk_score integer，随后新应用开始写该列。Subscriber 日志显示目标 Relation 缺列，Slot Retention 快速上升。怎样恢复且不丢数据？

参考答案：

  1. 暂停后续不兼容 DDL/发布，保留 Slot；不要 SKIP。
  2. 在 Subscriber 增加兼容列：
ALTER TABLE orders ADD COLUMN risk_score integer;
  1. 检查 Subscription Owner 权限、Column List、默认值和目标约束。
  2. ALTER SUBSCRIPTION ... ENABLE 或让 Worker 自动重试；原失败事务会从 Slot 重放。
  3. 观察 latest_end_lsn、冲突统计和 Retention 回落；校验故障窗口内新列值。
  4. 根修：DDL Pipeline 强制 Subscriber-First Expand/Contract，Schema Registry 对未批准 Relation Version 拦截告警。

21.11 系统设计题：12 TB、跨两大版本、30 秒写冻结预算

题目：一个 PG15 Cluster 有 12 TB 数据、峰值 45,000 TPS、120 MiB/s WAL，需迁移到 PG18；允许最终写冻结 30 秒。关键表有 Sequence、分区、Stored Generated Column、Large Object；业务要求切流后 20 分钟内可回滚。设计方案。

参考答案：

架构与边界

  • PG15 继续作为权威 Publisher，PG18 为 Subscriber；逻辑复制支持跨版本，但 [PG18] Generated Column 发布语义需单独测试，PG15 不会像 PG18 那样发布 Generated 列。
  • Large Object 不由逻辑复制处理：迁移到普通表/对象存储，或安排独立一致性复制窗口。
  • PG15 的 Publisher HA 仍用物理复制；不能依赖 PG17+ Failover Slot，故 CDC/迁移 Slot 的主备切换必须有外部恢复方案。

准备阶段

  1. 读取 PG16、17、18 Release Notes，完成 SQL、Driver、Extension、Collation、Planner 和权限兼容测试。
  2. 生成所有对象清单：表、分区、索引、约束、Sequence、Identity、Extension、LO、RLS、Trigger、Publication Eligibility。
  3. PG18 预建 Schema；破坏性差异通过 Expand/Contract 消除。
  4. 容量模型按 120 MiB/s 峰值、Initial COPY 时长和故障恢复 Lead Time 配置 pg_wal 与 Slot 预算。

全量与增量

  1. 按表组建立 Publication/Subscription；大表分批，控制 Table Sync Worker 数量。
  2. publish_via_partition_root 是否开启由目标分区布局决定。
  3. Initial COPY 期间监控源读 I/O、目标 WAL、索引写放大、Lag、Worker 和 Autovacuum。
  4. 对 Large Object 使用独立工具并用业务 ID/Hash 校验。
  5. Schema 变化实行 Subscriber-First；迁移窗口内禁止未登记 DDL。

校验

  • 每表行数、主键范围、分桶 Count、金额 Sum、状态分布、NULL、Checksum；
  • UPDATE/DELETE、分区移动、TRUNCATE、Row Filter 和 Generated Column 专项测试；
  • 所有 Sequence 的源/目标状态清单；
  • 应用 Shadow Read、关键交易回放和 PG18 执行计划基线。

30 秒 Cutover

  1. T-30s：在代理、数据库权限和应用开关三层 Fencing PG15 新写，驱逐旧长连接。
  2. 提交 Final Marker，记录最终业务时间和源 LSN。
  3. 等 PG18 Apply 到 Marker；30 秒预算要求正常阶段 Lag 已接近零。
  4. 复制所有 Sequence 的最终 last_value/is_called；校验 Identity。
  5. 禁用迁移 Subscription或保持受控单向链路，切应用到 PG18，执行读写 Smoke Test。
  6. 将 PG15 保持只读，不删除 Slot、不清理旧环境。

20 分钟回滚设计

  • 切流前启动并验证 PG18→PG15 的反向变更通道，或让所有 PG18 新写同时进入可重放的业务 Outbox；必须避免复制环路并为键分配全局唯一空间。
  • 在 20 分钟窗口内，若回滚：先冻结 PG18，等待反向流/Outbox 把 Target-Only Writes 应用到 PG15并验证，再切回。
  • 若反向机制未通过演练，则 Point of No Return 是 PG18 第一笔业务写；此时只能 Forward Fix，不能承诺 20 分钟无损回滚。

Go/No-Go Gate

  • 所有表 Ready、Lag/Runway 在阈值内；
  • 无未知 Schema Version、无冲突增量；
  • Large Object 和 Sequence 校验通过；
  • 旧写 Fencing 演练通过；
  • Backup/PITR 与回滚演练通过；
  • 业务、应用、数据库和 Incident Commander 共同签字。

22. 本章检查清单

原理与状态

  • 我能够从 WAL 产生开始，解释 Logical Decoding、pgoutput、Walsender、Apply Worker 的完整路径。
  • 我能够区分 Publication、Subscription、Logical Slot 与 Replication Origin。
  • 我能够解释 Snapshot、COPY、Catch-up、Handoff 和 pg_subscription_rel 状态。
  • 我能够区分 restart_lsn、confirmed_flush_lsn、received_lsn、latest_end_lsn、remote_lsn 和 local_lsn。
  • 我能够说明 Replica Identity 四种模式及 WAL、查找和类型代价。

功能与版本

  • 我能够配置 Row Filter、Column List 和 publish_via_partition_root，并验证边界更新。
  • 我能够说明 DDL、Sequence、Large Object 和 Schema Evolution 的复制边界。
  • 我能够说明 PG14 Large Transaction Streaming、PG15 Two-Phase、PG16 Parallel Apply、PG17 Failover Slot 与 PG18 新变化。
  • 我能够解释 [PG18] Generated Column 的发布组合和 Subscriber 行为。
  • 我能够使用 [PG18] 冲突统计区分停止型、跳过型和来源差异冲突。

实验与诊断

  • 我能够独立完成 Publication/Subscription Initial Sync 与持续复制实验。
  • 我能够复现 insert_exists、update_missing 并选择正确恢复策略。
  • 我能够模拟大版本切流、Fencing、最终 Marker、Sequence 对齐和回滚边界。
  • 我能够查询 Slot Retention、WAL Runway、Worker、Table State、Origin 和 Blocker。
  • 我能够区分 CPU、内存、I/O、锁、连接池、WAL、Vacuum 与复制问题。
  • 我知道 SKIP、推进 Origin/Slot、删除 Slot 和直接回切为何高风险。

Go 与 CDC

  • 我能够用 pglogrepl 建立 Replication Connection，处理 Keepalive、XLogData 和 Status Update。
  • 我能够把事件、Schema Version 与 Checkpoint 放在同一 Sink 事务中。
  • 我能够解释 At-Least-Once、重复事件、Commit Outcome Unknown 和幂等键。
  • 我能够实现 Bounded Queue、Backpressure、重试抖动和事务大小上限。
  • 我能够指出教学 Consumer 与生产 CDC 平台之间的 Snapshot、Spool、类型、DDL、HA 和安全差距。

性能、并发与高可用

  • 我能够分析 Initial COPY 和持续 Apply 对 CPU、I/O、WAL、索引和网络的写放大。
  • 我能够评估 Subscriber 本地写、DDL 锁和大事务对并发的影响。
  • 我能够计算 Slot WAL Runway 和净追赶时间，而不是只看 Time Lag。
  • 我能够设计 [PG17+] Failover Slot 验证，并说明旧版本的替代恢复路径。
  • 我能够为单表迁移、CDC、大版本升级和跨系统同步写出 Initial Sync、Incremental、Schema、Validation、Cutover、Rollback、Cleanup 与 Failure Handling。

23. 官方资料与延伸阅读

以下资料以 PostgreSQL 18 文档和 2026-06-21 可见的 pglogrepl/pgx API 为基线。生产实施必须同时阅读实际源版本、目标版本以及跨越的每个 Major Release Notes，并固定经过验证的 Go 依赖版本。

PostgreSQL 18 Logical Replication

  1. Chapter 29. Logical Replication
  2. Publication
  3. Subscription
  4. Logical Replication Failover
  5. Row Filters
  6. Column Lists
  7. Generated Column Replication
  8. Conflicts
  9. Restrictions
  10. Architecture and Initial Snapshot
  11. Monitoring
  12. Configuration Settings
  13. Upgrade of Logical Replication Clusters

SQL、协议、Slot 与 Origin

  1. CREATE PUBLICATION
  2. CREATE SUBSCRIPTION
  3. ALTER SUBSCRIPTION
  4. pg_replication_slots
  5. Logical Decoding
  6. Logical Replication Message Formats
  7. Replication Progress Tracking / Origins
  8. pg_upgrade

版本差异

  1. PostgreSQL 14 Release Notes
  2. PostgreSQL 15 Release Notes
  3. PostgreSQL 16 Release Notes
  4. PostgreSQL 17 Release Notes
  5. PostgreSQL 18 Release Notes

Go

  1. jackc/pglogrepl GitHub Repository
  2. pglogrepl Package Documentation
  3. jackc/pgx GitHub Repository
  4. pgx/v5/pgxpool Package Documentation

本章结论：Logical Replication 的价值不在于“把数据自动复制过去”，而在于提供一个可组合的事务增量机制。生产质量取决于围绕它建立的 Snapshot 边界、Schema 契约、Sequence 处理、幂等确认、WAL 预算、故障转移、校验、Fencing 和回滚纪律。只有这些条件同时成立，CDC 才可靠，在线迁移才真正可回退，大版本升级才不只是一次高风险切换。