A 库有数据
    ↓
把数据同步到 B 库

A 库通过 Publication 声明“我要发布哪些表的数据变化”，
B 库通过 Subscription 声明“我要订阅哪个发布”。

业务写入 A 库
    ↓
A 库生成 WAL
    ↓
逻辑解码把 WAL 转换成行级变化
    ↓
Publication 过滤需要发布的表和操作
    ↓
Replication Slot 记录 B 库消费位置
    ↓
B 库 Subscription 接收变化
    ↓
Apply Worker 将变化应用到同名目标表

IP：192.168.1.1
端口：5432
数据库：test_db
schema：test_sync
Publication：test_sync_pub
复制用户：repl_test

IP：192.168.1.2
端口：5432
数据库：test_db
schema：test_sync
Subscription：test_sync_sub

test_customer
test_order
test_order_item
test_task
test_metric

SELECT *
FROM "test-order";

A 库某个数据页发生变化
    ↓
WAL 记录这个底层变化
    ↓
B 库按相同物理方式重放

我只想复制 test_order，
但不复制 test_customer。

表：test_sync.test_order
操作：INSERT
主键：10001
字段：
  order_no = TEST-10001
  amount   = 128.50

INSERT INTO test_sync.test_order(
    id,
    order_no,
    amount,
    created_at
)
VALUES (
    10001,
    'TEST-10001',
    128.50,
    now()
);

wal_level = logical

pgoutput

CREATE PUBLICATION test_sync_pub
FOR TABLES IN SCHEMA test_sync
WITH (
    publish = 'insert, update, delete, truncate'
);

只关注 test_sync schema 中的表；
发布 INSERT、UPDATE、DELETE、TRUNCATE。

test_sync_sub

A 库当前 WAL：0/A000000
B 库已确认：  0/8000000

0/8000000 之后的 WAL 暂时不能被清理，
因为 B 库还没有消费完成。

test_sync.test_order

INSERT
  ↓
WAL
  ↓
Logical Decoding
  ↓
Publication
  ↓
Replication Slot
  ↓
Subscription
  ↓
Apply Worker
  ↓
B 库目标表

192.168.1.1

发布哪些表？
发布哪些操作？

CREATE PUBLICATION test_sync_pub
FOR TABLES IN SCHEMA test_sync
WITH (
    publish = 'insert, update, delete, truncate'
);

192.168.1.2

没有复制槽：
B 库断开后可能丢失所需 WAL。

有复制槽但不监控：
B 库长期离线可能导致 A 库 WAL 撑满磁盘。

copy_data = true

A 库现有数据
    ↓
COPY
    ↓
B 库目标表

max_sync_workers_per_subscription = 2

2 张表处于 d
其他表处于 i

存量数据复制完成
+
COPY 期间发生的增量变化

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

r

r | 5

主 Apply Worker
test_sync.test_order
test_sync.test_order_item

state = idle
wait_event = LogicalApplyMain

既然数据能同步，表结构应该也会同步。

id bigserial

SELECT
    n.nspname,
    c.relname,
    c.relpersistence
FROM pg_class c
JOIN pg_namespace n
  ON n.oid = c.relnamespace
WHERE n.nspname = 'test_sync'
  AND c.relkind IN ('r', 'p');

p = 持久化表
u = UNLOGGED 表
t = 临时表

FOR TABLES IN SCHEMA test_sync

B 库复制表只读。

SELECT
    n.nspname AS schema_name,
    c.relname AS table_name,
    EXISTS (
        SELECT 1
        FROM pg_index i
        WHERE i.indrelid = c.oid
          AND i.indisprimary
    ) AS has_primary_key,
    c.relreplident
FROM pg_class c
JOIN pg_namespace n
  ON n.oid = c.relnamespace
WHERE n.nspname = 'test_sync'
  AND c.relkind IN ('r', 'p')
ORDER BY c.relname;

ALTER TABLE test_sync.test_xxx
REPLICA IDENTITY FULL;

copy_data = true

copy_data = false

max_slot_wal_keep_size = -1

新增字段：B 先，A 后。

missing replicated columns

ERROR: schema "test_sync" does not exist

SELECT
    current_database(),
    current_user;

\c test_db

SELECT schema_name
FROM information_schema.schemata
WHERE schema_name = 'test_sync';

SELECT
    n.nspname AS schema_name,
    c.relname AS table_name,
    CASE c.relkind
        WHEN 'r' THEN '普通表'
        WHEN 'p' THEN '分区表'
    END AS table_type,
    CASE c.relpersistence
        WHEN 'p' THEN '持久化表'
        WHEN 'u' THEN 'UNLOGGED表'
        WHEN 't' THEN '临时表'
    END AS persistence,
    EXISTS (
        SELECT 1
        FROM pg_index i
        WHERE i.indrelid = c.oid
          AND i.indisprimary
    ) AS has_primary_key,
    c.relreplident
FROM pg_class c
JOIN pg_namespace n
  ON n.oid = c.relnamespace
WHERE n.nspname = 'test_sync'
  AND c.relkind IN ('r', 'p')
ORDER BY c.relname;

表类型正确
持久化属性为 p
需要 UPDATE/DELETE 的表具备主键
表数量符合预期

SHOW wal_level;

replica

ALTER SYSTEM SET wal_level = 'logical';

SELECT
    name,
    setting,
    source,
    sourcefile,
    pending_restart
FROM pg_settings
WHERE name = 'wal_level';

SELECT pg_reload_conf();

sudo systemctl restart postgresql

SHOW wal_level;

logical

SHOW max_replication_slots;
SHOW max_wal_senders;
SHOW max_slot_wal_keep_size;
SHOW wal_sender_timeout;

max_replication_slots = 10

CREATE ROLE repl_test
WITH
    LOGIN
    REPLICATION
    PASSWORD '<REPLICATION_PASSWORD>';

GRANT CONNECT
ON DATABASE test_db
TO repl_test;

GRANT USAGE
ON SCHEMA test_sync
TO repl_test;

GRANT SELECT
ON ALL TABLES IN SCHEMA test_sync
TO repl_test;

ALTER DEFAULT PRIVILEGES
FOR ROLE test_owner
IN SCHEMA test_sync
GRANT SELECT ON TABLES TO repl_test;

FOR ROLE test_owner

SHOW hba_file;
SHOW listen_addresses;

host    test_db    repl_test    192.168.1.2/32    scram-sha-256

SELECT pg_reload_conf();

0.0.0.0/0

CREATE PUBLICATION test_sync_pub
FOR TABLES IN SCHEMA test_sync
WITH (
    publish = 'insert, update, delete, truncate'
);

SELECT
    pubname,
    pubinsert,
    pubupdate,
    pubdelete,
    pubtruncate
FROM pg_publication
WHERE pubname = 'test_sync_pub';

SELECT
    pubname,
    schemaname,
    tablename
FROM pg_publication_tables
WHERE pubname = 'test_sync_pub'
ORDER BY schemaname, tablename;

CREATE PUBLICATION

psql \
  -h 192.168.1.1 \
  -p 5432 \
  -U repl_test \
  -d test_db

SELECT
    current_database(),
    current_user;

SELECT count(*)
FROM test_sync.test_order;

SHOW max_replication_slots;
SHOW max_logical_replication_workers;
SHOW max_worker_processes;
SHOW max_sync_workers_per_subscription;
SHOW wal_receiver_timeout;

max_sync_workers_per_subscription

max_sync_workers_per_subscription = 2

export PGPASSWORD='<A_DATABASE_PASSWORD>'

pg_dump \
  -h 192.168.1.1 \
  -p 5432 \
  -U postgres \
  -d test_db \
  --format=custom \
  --schema-only \
  --schema=test_sync \
  --no-owner \
  --no-privileges \
  --file=/tmp/test_sync_schema.dump

pg_restore \
  --list \
  /tmp/test_sync_schema.dump \
  >/tmp/test_sync_schema.list

echo $?

0

test_sync.test_order.created_by
    → public.test_user.id

SELECT to_regclass('public.test_user');

停业务写入
    ↓
停用或删除旧 Subscription
    ↓
确认复制 Worker 退出
    ↓
终止长事务和阻塞连接
    ↓
删除旧 schema
    ↓
恢复最新结构

DROP SCHEMA IF EXISTS test_sync CASCADE;

SET lock_timeout = '15s';

export PGPASSWORD='<B_DATABASE_PASSWORD>'

pg_restore \
  -h 192.168.1.2 \
  -p 5432 \
  -U postgres \
  --role=test_owner \
  -d test_db \
  --no-owner \
  --no-privileges \
  --exit-on-error \
  --single-transaction \
  --verbose \
  /tmp/test_sync_schema.dump

--role=test_owner

--exit-on-error

--single-transaction

SELECT
    tablename,
    tableowner
FROM pg_tables
WHERE schemaname = 'test_sync'
ORDER BY tablename;

SELECT
    table_name,
    ordinal_position,
    column_name,
    data_type,
    udt_name,
    is_nullable,
    column_default
FROM information_schema.columns
WHERE table_schema = 'test_sync'
ORDER BY table_name, ordinal_position;

SELECT format(
    'SELECT %L AS table_name, count(*) AS row_count FROM %I.%I;',
    schemaname || '.' || tablename,
    schemaname,
    tablename
)
FROM pg_tables
WHERE schemaname = 'test_sync'
ORDER BY tablename
\gexec

CREATE SUBSCRIPTION test_sync_sub
CONNECTION '
    host=192.168.1.1
    port=5432
    dbname=test_db
    user=repl_test
    password=<REPLICATION_PASSWORD>
    application_name=test_sync_sub
    connect_timeout=10
'
PUBLICATION test_sync_pub
WITH (
    copy_data = true,
    create_slot = true,
    enabled = true,
    disable_on_error = true
);

SELECT
    subname,
    subenabled,
    subslotname,
    subdisableonerr,
    subpublications
FROM pg_subscription
WHERE subname = 'test_sync_sub';

subenabled = t

SELECT
    r.srrelid::regclass AS table_name,
    r.srsubstate,
    CASE r.srsubstate
        WHEN 'i' THEN '初始化排队'
        WHEN 'd' THEN '复制存量数据'
        WHEN 'f' THEN '存量复制完成'
        WHEN 's' THEN '已同步，等待切换'
        WHEN 'r' THEN '正常实时复制'
        ELSE '未知状态'
    END AS sync_state
FROM pg_subscription_rel r
JOIN pg_subscription s
  ON s.oid = r.srsubid
WHERE s.subname = 'test_sync_sub'
ORDER BY table_name;

SELECT
    r.srsubstate,
    count(*) AS table_count
FROM pg_subscription_rel r
JOIN pg_subscription s
  ON s.oid = r.srsubid
WHERE s.subname = 'test_sync_sub'
GROUP BY r.srsubstate
ORDER BY r.srsubstate;

r | 5

SELECT
    ps.pid,
    CASE
        WHEN ps.relid IS NULL THEN '主 Apply Worker'
        ELSE ps.relid::regclass::text
    END AS worker,
    a.state,
    a.wait_event_type,
    a.wait_event,
    ps.latest_end_time
FROM pg_stat_subscription ps
LEFT JOIN pg_stat_activity a
  ON a.pid = ps.pid
WHERE ps.subname = 'test_sync_sub'
ORDER BY ps.relid NULLS FIRST;

SELECT
    pid,
    relid::regclass AS table_name,
    command,
    type,
    pg_size_pretty(bytes_processed) AS processed,
    tuples_processed
FROM pg_stat_progress_copy;

SELECT
    clock_timestamp(),
    pid,
    relid::regclass AS table_name,
    bytes_processed,
    tuples_processed
FROM pg_stat_progress_copy
\watch 10

INSERT INTO test_sync.test_order(
    id,
    order_no,
    amount,
    created_at
)
VALUES (
    900001,
    'TEST-ORDER-900001',
    100.00,
    now()
);

SELECT *
FROM test_sync.test_order
WHERE id = 900001;

UPDATE test_sync.test_order
SET amount = 200.00
WHERE id = 900001;

SELECT
    id,
    order_no,
    amount
FROM test_sync.test_order
WHERE id = 900001;

DELETE FROM test_sync.test_order
WHERE id = 900001;

SELECT *
FROM test_sync.test_order
WHERE id = 900001;

0 rows

ERROR: schema "test_sync" does not exist

SELECT current_database();

ALTER SYSTEM SET wal_level = 'logical';

SELECT pg_reload_conf();

SELECT
    setting,
    pending_restart
FROM pg_settings
WHERE name = 'wal_level';

一个文件里出现多个 dump complete
SQL 从半行开始
约束定义被截断

pg_dump --format=custom

pg_restore --list

logical replication target relation
is missing replicated columns

逻辑复制不同步 DDL。

duplicate key value violates unique constraint

停止写入
→ 清理目标表
→ 重建结构
→ 重新初始同步

SELECT
    subenabled,
    subdisableonerr
FROM pg_subscription
WHERE subname = 'test_sync_sub';

subenabled = f

subdisableonerr = t

ALTER SUBSCRIPTION test_sync_sub ENABLE;

SELECT
    pid,
    relid::regclass AS table_name,
    bytes_processed,
    tuples_processed
FROM pg_stat_progress_copy;

SELECT
    ps.pid,
    ps.relid::regclass AS table_name,
    a.wait_event_type,
    a.wait_event
FROM pg_stat_subscription ps
LEFT JOIN pg_stat_activity a
  ON a.pid = ps.pid
WHERE ps.subname = 'test_sync_sub';

SELECT pg_terminate_backend(<TABLE_SYNC_WORKER_PID>);

terminating logical replication worker due to timeout

A 库：wal_sender_timeout
B 库：wal_receiver_timeout

-- A 库
ALTER SYSTEM SET wal_sender_timeout = '15min';
SELECT pg_reload_conf();

-- B 库
ALTER SYSTEM SET wal_receiver_timeout = '15min';
SELECT pg_reload_conf();

-- B 库先执行
ALTER TABLE test_sync.test_order
ADD COLUMN test_remark varchar(500);

-- A 库后执行
ALTER TABLE test_sync.test_order
ADD COLUMN test_remark varchar(500);

B 库结构兼容
    ↓
A 库新增字段
    ↓
业务开始写新字段

业务停止读写字段
    ↓
确认 A 库不再使用
    ↓
A 库删除字段
    ↓
验证复制正常
    ↓
B 库删除字段

FOR TABLES IN SCHEMA test_sync

B 库先创建同名表
    ↓
设置正确 owner
    ↓
A 库创建表
    ↓
给复制用户 SELECT
    ↓
B 库 REFRESH PUBLICATION
    ↓
观察新表进入 r

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

SELECT
    s.subname,
    s.subenabled,
    s.subdisableonerr,
    st.apply_error_count,
    st.sync_error_count
FROM pg_subscription s
LEFT JOIN pg_stat_subscription_stats st
  ON st.subid = s.oid
WHERE s.subname = 'test_sync_sub';

SELECT
    ps.pid,
    ps.relid::regclass AS syncing_table,
    a.state,
    a.wait_event_type,
    a.wait_event,
    ps.received_lsn,
    ps.latest_end_lsn,
    ps.latest_end_time
FROM pg_stat_subscription ps
LEFT JOIN pg_stat_activity a
  ON a.pid = ps.pid
WHERE ps.subname = 'test_sync_sub';

SELECT
    slot_name,
    active,
    wal_status,
    pg_size_pretty(
        pg_wal_lsn_diff(
            pg_current_wal_lsn(),
            restart_lsn
        )
    ) AS retained_wal
FROM pg_replication_slots
WHERE slot_name = 'test_sync_sub';

ALTER SUBSCRIPTION test_sync_sub DISABLE;

SELECT
    pid,
    relid::regclass AS syncing_table
FROM pg_stat_subscription
WHERE subname = 'test_sync_sub';

ALTER SUBSCRIPTION test_sync_sub ENABLE;

DROP SUBSCRIPTION test_sync_sub;

ALTER SUBSCRIPTION test_sync_sub DISABLE;

ALTER SUBSCRIPTION test_sync_sub
SET (
    slot_name = NONE
);

DROP SUBSCRIPTION test_sync_sub;

ALTER SUBSCRIPTION test_sync_sub
CONNECTION '
    host=192.168.1.1
    port=5432
    dbname=test_db
    user=repl_test
    password=<NEW_REPLICATION_PASSWORD>
    application_name=test_sync_sub
    connect_timeout=10
';

srsubstate = r

[ ] 当前数据库正确
[ ] wal_level = logical
[ ] max_replication_slots 足够
[ ] max_wal_senders 足够
[ ] 复制用户具备 REPLICATION
[ ] 复制用户具备 CONNECT、USAGE、SELECT
[ ] pg_hba.conf 仅允许 B 库来源
[ ] Publication 表清单正确
[ ] 发布表有主键或合理 Replica Identity
[ ] 复制槽纳入 WAL 和磁盘监控

[ ] 目标 schema 已创建
[ ] 目标表结构与 A 库兼容
[ ] 目标表 owner 正确
[ ] 初次 copy_data=true 时目标表为空
[ ] Worker 参数足够
[ ] wal_receiver_timeout 合理
[ ] Subscription 已创建
[ ] subenabled = true
[ ] 所有表 srsubstate = r
[ ] 主 Apply Worker 正常
[ ] B 库业务账号无同步表写权限

[ ] INSERT 已验证
[ ] UPDATE 已验证
[ ] DELETE 已验证
[ ] A/B 抽样数据一致
[ ] 错误统计不再增长
[ ] 复制槽 active=true
[ ] retained_wal 在合理范围
[ ] 告警已配置

A 库创建 Publication
B 库创建 Subscription

复制范围设计
+
结构变更规范
+
故障排查手册
+
监控告警
+
安全权限
+
验收标准