逻辑复制并行 Apply 用户接口文档

文档版本:v1.0
适配版本:PostgreSQL 18(含社区已合入的并行 Apply 增强补丁)
源码基线:src/backend/replication/logical/{worker,applyparallelworker,launcher,tablesync}.c
读者:DBA、应用开发者、运维工程师

一、文档目的

本文档定义 TWPA(Transaction Window Parallel Apply,事务窗口并行 Apply) 用户可见的接口规范,包括:

• CREATE/ALTER SUBSCRIPTION 的新参数
• 相关 GUC(Grand Unified Configuration)参数
• 监控视图与统计字段
• 错误码与异常处理
• 兼容性矩阵

不涉及内部数据结构、调度算法(见《概要设计文档》)与依赖冲突分析(见《依赖冲突检测设计文档》)。

二、术语表

三、功能定位与边界

3.1 三套 Apply 机制并存

3.2 TWPA 目标

3.3 非目标(N1~N4)

四、DDL 接口

4.1 CREATE SUBSCRIPTION 新增选项

CREATE SUBSCRIPTION name
    CONNECTION 'conninfo'
    PUBLICATION publication_names
    [ WITH ... ]          -- 现有选项保留
    -- ↓↓↓ 新增选项 ↓↓↓
    [ TRANSACTION_PARALLEL { off | on | adaptive } ]
    [ NTRANSACTIONS_PARALLEL_WORKERS integer ]
    [ TRANSACTION_PARALLEL_WINDOW_SIZE integer ]
    [ TRANSACTION_PARALLEL_WINDOW_TIME interval ]
    [ TRANSACTION_PARALLEL_WINDOW_BYTES integer ]
    [ TRANSACTION_PARALLEL_MIN_TRANSACTIONS integer ]
    [ CONFLICT_DETECTION_MODE { conservative | primary_key | full } ]
;

4.2 ALTER SUBSCRIPTION

-- 启用
ALTER SUBSCRIPTION name ENABLE;
ALTER SUBSCRIPTION name SET (transaction_parallel = on);

-- 关闭
ALTER SUBSCRIPTION name SET (transaction_parallel = off);

-- 动态调参
ALTER SUBSCRIPTION name SET (transaction_parallel_window_size = 2000);

4.3 兼容性矩阵

历史 / 已弃用选项保留原 pg_doc 描述,本节不重复。

五、GUC 参数一览(用户接口)

5.1 参数总览

5.2 transaction_parallel(主开关)

5.2.1 取值含义

5.2.2 切换时的状态机

5.2.3 SQL 示例

-- 方式 1:CREATE 时指定
CREATE SUBSCRIPTION sub_oltp_to_olap
    CONNECTION 'host=pub port=5432 dbname=oltp user=repl'
    PUBLICATION pub_orders
    WITH (transaction_parallel = adaptive);

-- 方式 2:动态修改
ALTER SUBSCRIPTION sub_oltp_to_olap SET (transaction_parallel = on);

-- 方式 3:全局默认(对所有新 subscription)
-- 注:此 GUC 不在 postgresql.conf 中,仅作为 subscription option

5.3 transaction_parallel_workers

ALTER SUBSCRIPTION sub SET (transaction_parallel_workers = 4);

约束:

1 ≤ transaction_parallel_workers
  ≤ max_parallel_apply_workers_per_subscription (默认 2)

调优建议:

5.4 transaction_parallel_window_size

控制窗口最多包含多少个已接收事务。

ALTER SUBSCRIPTION sub SET (transaction_parallel_window_size = 2000);

推荐默认 1000,过大导致内存压力大,过小导致频繁调度。

5.5 transaction_parallel_window_time

窗口最大等待时间。

ALTER SUBSCRIPTION sub SET (transaction_parallel_window_time = '30s');

T0: window 接收第一个事务
T0 + transaction_parallel_window_time: 强制触发分析

与 window_size / window_bytes 形成”三选一”的 OR 关系,任一达到即触发调度。

5.6 transaction_parallel_window_bytes(建议)

窗口内变更总字节数限制。

ALTER SUBSCRIPTION sub SET (transaction_parallel_window_bytes = '2GB');

5.7 transaction_parallel_min_transactions

只有当窗口内事务数 ≥ 该值时才尝试并行;否则直接串行 apply。

ALTER SUBSCRIPTION sub SET (transaction_parallel_min_transactions = 10);

推荐默认 5,避免单事务也走并行调度反而更慢。

5.8 conflict_detection_mode(冲突检测粒度)

5.9 推荐最终 CREATE SUBSCRIPTION 模板

CREATE SUBSCRIPTION sub_from_pubdb
    CONNECTION 'host=10.0.0.1 port=5432 dbname=oltp user=repl password=xxx'
    PUBLICATION pub_orders, pub_users
    WITH (
        -- 现有选项
        streaming = on,
        binary = on,
        two_phase = off,
        disable_on_error = off,
        run_as_owner = off,

        -- 新增 TWPA 选项
        transaction_parallel = adaptive,
        transaction_parallel_workers = 4,
        transaction_parallel_window_size = 1000,
        transaction_parallel_window_time = '10s',
        transaction_parallel_window_bytes = '1GB',
        transaction_parallel_min_transactions = 5,
        conflict_detection_mode = primary_key
    );

六、监控接口

6.1 新增视图:pg_stat_subscription_worker

CREATE VIEW pg_stat_subscription_worker AS
SELECT
    w.subid,
    w.worker_pid,
    w.worker_type,
    w.xid,
    w.state,
    w.begin_lsn,
    w.commit_lsn,
    w.change_count,
    w.bytes_processed,
    w.start_time,
    w.duration_ms,
    w.assigned_window_id
FROM pg_stat_subscription_worker_internal w;

字段说明:

6.2 worker_type 分类

6.3 监控 SQL 模板

6.3.1 总体状态

SELECT worker_type, state, count(*), sum(bytes_processed) AS total_bytes
  FROM pg_stat_subscription_worker
 GROUP BY worker_type, state
 ORDER BY 1, 2;

6.3.2 找出卡住的事务

SELECT worker_pid, xid, state, duration_ms, change_count
  FROM pg_stat_subscription_worker
 WHERE state NOT IN ('COMMITTED')
   AND duration_ms > 30000   -- 超过 30s
 ORDER BY duration_ms DESC;

6.3.3 窗口 / 吞吐统计

SELECT subid,
       count(*) AS worker_count,
       avg(duration_ms) AS avg_duration,
       sum(bytes_processed) / (extract(epoch FROM now() - min(start_time)) * 1024 * 1024) AS mb_per_sec
  FROM pg_stat_subscription_worker
 WHERE state IN ('RUNNING', 'COMMIT_WAIT')
 GROUP BY subid;

6.4 GUC 视图补充

-- 查看某个 subscription 的全部并行 apply 相关参数
SELECT s.subname,
       s.subtransaction_parallel,
       s.subparallel_workers,
       s.subwindow_size,
       s.subwindow_time,
       s.subwindow_bytes,
       s.submin_transactions,
       s.subconflict_mode
  FROM pg_subscription s
 WHERE s.subname = 'sub_from_pubdb';

七、错误码与异常处理

7.1 错误分类

7.2 Fallback 行为

7.3 SQL 异常输出

ERROR:  TWPA fallback: deadlock detected between workers (xid=12345, 12346)
DETAIL:  Automatically falling back to serial apply for this window.
HINT:    Consider reducing transaction_parallel_window_size or adding primary keys.

八、兼容性

8.1 与现有选项的兼容

8.2 与 DDL 同步的兼容

TWPA 窗口内包含 DDL 时强制串行,避免 DDL 与 DML 竞争。

九、典型示例

9.1 启用 TWPA(从单事务串行升级)

-- 步骤 1:查看当前状态
SELECT * FROM pg_stat_subscription WHERE subname = 'sub_from_pubdb';

-- 步骤 2:开启自适应模式
ALTER SUBSCRIPTION sub_from_pubdb SET (transaction_parallel = adaptive);

-- 步骤 3:监控
SELECT * FROM pg_stat_subscription_worker WHERE subid = <sub_oid>;

9.2 关闭 TWPA(临时)

ALTER SUBSCRIPTION sub_from_pubdb SET (transaction_parallel = off);

9.3 调高并行度

-- 步骤 1:确认 subscription 级 GUC
SHOW max_parallel_apply_workers_per_subscription;  -- 默认 2

-- 步骤 2:同步 GUC
ALTER SYSTEM SET max_parallel_apply_workers_per_subscription = 8;
SELECT pg_reload_conf();

-- 步骤 3:subscription 侧匹配
ALTER SUBSCRIPTION sub_from_pubdb SET (transaction_parallel_workers = 8);

十、FAQ

Q1:开启 TWPA 后性能反而下降?

Q2:TWPA 与 streaming=parallel 谁优先?

不冲突,分工不同:

Q3:能否只对某些表启用 TWPA?

不支持表级开关;要么订阅级开,要么不开。

十一、版本历史

十二、参考

• PG 18 源码:
  • src/backend/replication/logical/applyparallelworker.c(并行 apply worker)
  • src/backend/replication/logical/worker.c(leader apply worker)
  • src/backend/replication/logical/launcher.c(apply launcher)
  • src/include/replication/{logicalworker,worker_internal}.h(API)
• 配套文档:
  • 《TWPA 概要设计文档》
  • 《TWPA 依赖冲突检测设计文档》
• 标准参考:Documentation/subscription 与 pg_subscription 系统视图