# 逻辑复制

> PostgreSQL 逻辑复制的发布订阅模型、ReorderBuffer、pgoutput、消息传输、应用与恢复机制。

---

LLMS 索引： [llms.txt](/llms.txt)

---

> **Beta版：仍在编写中。** 英文原著明确把第12章标记为Beta/WIP。本译文按当前上游内容同步，章节结构和技术细节仍可能继续变化。

PostgreSQL从10（2017年）开始原生支持逻辑复制。流复制中的节点称为**主库（Primary）**和**备库（Standby）**；逻辑复制则使用**发布者（Publisher）**与**订阅者（Subscriber）**。

**图 12.1.** 流复制与逻辑复制的数据流及节点角色

![图12\.1 流复制使用主库和备库，逻辑复制使用发布者和订阅者](/img/fig-12-01.png)

流复制复制整个数据库集簇，而逻辑复制可以只复制一个数据库中的指定表。流复制备库通常只能处理只读查询；逻辑订阅者则可以一边接收发布者的数据，一边独立执行本地写入。

| 特性 | 逻辑复制 | 流复制 |
|:-----|:---------|:-------|
| WAL发送时机 | 通常在事务提交后；大事务可在提交前分段传输 | WAL生成后立即传输 |
| 发送进程 | walsender | walsender |
| 接收进程 | logical replication launcher、apply worker | walreceiver、startup |
| 复制单位 | 表、数据库 | 整个数据库集簇 |
| 不复制的操作 | DDL | 无 |
| 不复制的对象 | 视图、索引、大对象等非表对象；序列自PG 19起支持 | UNLOGGED表 |
| 跨大版本 | 支持 | 不支持 |
| 跨操作系统 | 支持 | 不支持 |
| 订阅端/备库上的DML | `INSERT`、`UPDATE`、`DELETE`、`SELECT`、`TRUNCATE` | `SELECT` |
| 订阅端/备库触发器 | apply worker默认不触发普通触发器；可启用`REPLICA`或`ALWAYS`行级触发器 | 禁用 |

本章依次介绍逻辑复制的关键概念与启动流程、ReorderBuffer结构、WAL过滤与缓冲、`pgoutput`输出插件、事务消息构造、订阅端应用，以及重启和崩溃恢复。

## 12.1 概述与关键概念

### 12.1.1 相关进程

逻辑复制由四类进程协同完成：

* 发布者上的**walsender**向订阅者发送逻辑变更，并执行各种协调工作。
* 订阅者上的**logical replication launcher**负责启动apply worker。
* **apply worker**连接发布者的walsender，接收逻辑变更流、解析消息，并更新目标表。
* **table sync worker**负责某张表的初始数据同步；追上主复制流后，再把后续更新交给apply worker。

### 12.1.2 逻辑复制概要

PostgreSQL逻辑复制是**基于行（row-based）**的：它捕获各行的变化，并以解码后的格式传输。逻辑消息携带操作**已经求值的结果**，因此不会因为`random()`、`now()`等非确定性函数在两端重复求值而产生不一致。

**图 12.2.** 流复制与逻辑复制的数据发送方式

![图12\.2 流复制按WAL物理顺序发送，逻辑复制按事务重组并过滤变更](/img/fig-12-02.png)

假设两个并发事务（`txid=99`和`100`）修改`tbl_a`与`tbl_b`：

* **流复制**在每条SQL执行时依次写入WAL记录$w_1$至$w_5$，walsender不考虑事务是否提交，直接把记录发送给备库。walreceiver按相同顺序写入本地WAL，形成主库的物理镜像。
* **逻辑复制**按事务边界与发布范围处理数据。walsender一边读取WAL，一边把记录解码为逻辑变更，并在**ReorderBuffer**中按事务重组，而不是立即发送。

ReorderBuffer相关处理分为三步：

1. **过滤：** 按发布范围过滤变更。如果订阅者只订阅`tbl_a`，那么来自`tbl_b`的变更会被丢弃。
2. **解码与缓冲：** 把每条WAL记录$w_n$解码为逻辑变更$c_n$，存入ReorderBuffer。
3. **发送：** 读到`txid=100`的提交记录$w_5$后，收集属于该事务且在发布范围内的$c_1$、$c_4$，连同提交消息$c_5$一起发送。尚未提交的`txid=99`变更继续保留在缓冲区中。

标准逻辑复制默认使用`pgoutput`插件进行逻辑解码。订阅端apply worker根据收到的消息重建事务并执行相应修改。

#### 12.1.2.1 异步与同步

PostgreSQL逻辑复制同时支持异步和同步模式，默认异步。

**图 12.3.** 异步与同步逻辑复制的事务流程

![图12\.3 异步提交只等待本地WAL，同步remote\_apply还等待订阅端应用完成](/img/fig-12-03.png)

* **异步模式：** 发布者的`COMMIT`在本地WAL刷盘后立即完成，不等待订阅者响应。
* **同步模式：** 当`synchronous_commit='remote_apply'`时，发布者提交会等待订阅者ACK；apply worker只有在订阅端完成应用后才发送确认。

同步模式相较异步模式多出的提交耗时称为**端到端延迟（End-to-End Latency）**。事务越大，解码、传输和应用越耗时：异步模式下体现为更大的复制延迟和更长的不一致窗口；同步模式下则直接增加发布者的提交延迟。

#### 12.1.2.2 大事务的管理与优化

##### 1. 标准事务处理（`streaming=off`）

变更量超过[`logical_decoding_work_mem`](https://www.postgresql.org/docs/current/runtime-config-resource.html#GUC-LOGICAL-DECODING-WORK-MEM)后，walsender会把ReorderBuffer中的数据写入磁盘**溢出文件（spill file）**。

**图 12.4.** 关闭流式传输时的大事务

![图12\.4 大事务在发布端溢出到磁盘，提交后才整体发送和串行应用](/img/fig-12-04.png)

处理流程为：变更进入ReorderBuffer；超出内存限制后写入溢出文件；事务在发布者提交；walsender从内存和溢出文件收集全部变更并发送；apply worker按顺序应用。每条连接一次只发送一个事务，因此大事务容易造成明显复制延迟。

##### 2. 大事务提交前流式发送（PostgreSQL 14及以后）

PostgreSQL 14允许在事务提交前分段发送变更，从而隐藏提交后的传输开销。

**图 12.5.** 大事务的进行中流式传输

![图12\.5 发布端达到内存阈值后提前发送，订阅端缓存并在提交后应用](/img/fig-12-05.png)

当ReorderBuffer达到内存阈值时，walsender立即开始发送；apply worker把收到的原始二进制消息缓存在`StringInfoData original_msg`中，超过订阅端内存限制后写入临时文件。发布者提交后，apply worker从内存或临时文件取出消息并应用。

使用`CREATE SUBSCRIPTION ... WITH (streaming = on)`启用该模式。apply worker的消息缓存同样受`logical_decoding_work_mem`控制。

##### 3. 流式变更的并行应用（PostgreSQL 16及以后）

PostgreSQL 16进一步允许在事务提交前就并行应用已流式传输的变更，使传输与应用都能和发布者上的事务执行重叠。

**图 12.6.** 并行apply worker处理流式变更

![图12\.6 leader apply worker把流式消息交给并行worker并在提交时完成本地事务](/img/fig-12-06.png)

leader apply worker接收消息并分派给parallel apply worker；后者立即修改目标表。发布者提交后，leader把提交消息转交给parallel worker，由其完成本地事务。使用`streaming='parallel'`启用。若活跃worker达到`max_parallel_apply_workers_per_subscription`上限，则会退化为提交后串行应用。

### 12.1.3 副本标识

逻辑复制处理逻辑行，而不是块号、偏移量等物理位置。订阅者执行`UPDATE`或`DELETE`时，需要一个“搜索键”准确定位要修改的行，这就是**副本标识（Replica Identity）**。

如果发布表没有合适的副本标识，PostgreSQL会拒绝发布`UPDATE`或`DELETE`：

```sql
testdb=# UPDATE tbl SET data = 'updated_value' WHERE id = 1;
ERROR:  cannot update table "tbl" because it does not have a replica identity and publishes updates
HINT:  To enable updating the table, set REPLICA IDENTITY using ALTER TABLE.
```

#### 12.1.3.1 副本标识类型

| 模式 | 含义 | 特点 |
|:-----|:-----|:-----|
| `DEFAULT` | 使用主键列 | 表有主键时自动使用 |
| `USING INDEX` | 使用指定的唯一、非空索引 | 适用于无主键但有合适唯一索引的表 |
| `FULL` | 记录行中所有列的旧值 | 无唯一约束时使用，但会增加消息体积 |
| `NOTHING` | 不记录标识信息 | 无主键表的默认值；可复制`INSERT`，不能复制`UPDATE`与`DELETE` |

#### 12.1.3.2 配置与验证

使用`ALTER TABLE ... REPLICA IDENTITY`配置：

```sql
testdb=# CREATE TABLE tbl_ri (id int NOT NULL, name text, data int NOT NULL);
testdb=# CREATE UNIQUE INDEX tbl_ri_idx ON tbl_ri (id, data);
testdb=# ALTER TABLE tbl_ri REPLICA IDENTITY USING INDEX tbl_ri_idx;

testdb=# CREATE TABLE tbl_ri_full (id int, name text, data int);
testdb=# ALTER TABLE tbl_ri_full REPLICA IDENTITY FULL;
```

模式保存在`pg_class.relreplident`中，取值为`d`（default）、`n`（nothing）、`f`（full）或`i`（index）：

```sql
testdb=# SELECT relname, relreplident
           FROM pg_class WHERE relname IN ('tbl_ri', 'tbl_ri_full');
   relname   | relreplident
-------------+--------------
 tbl_ri      | i
 tbl_ri_full | f
(2 rows)
```

当使用`USING INDEX`时，具体索引并不存于`pg_class.relreplident`，而是在`pg_index.indisreplident`中标记。

### 12.1.4 复制来源

**复制来源（Replication Origin）**标识一次数据变更来自哪里，主要有两个用途：

1. **跟踪复制进度：** 订阅者应用外部变更时，把发布者的提交LSN记录到自己的WAL中，使中断后能从正确位置继续。
2. **阻止无限复制环：** 双向复制中，节点A发送到B的变更可能又被B发回A。给变更盖上来源标记，可以区分本地变更与复制得到的变更。

`origin_id`是订阅者本地分配的标识，不是集簇级或全局唯一键，其具体数字对其他节点没有意义：`0`表示事务最初在本节点执行；非零表示本节点正在重放上游变更。

#### 12.1.4.1 使用复制来源阻止无限循环

订阅端表仍可写，因此两个节点可以同时作为发布者与订阅者，形成双活复制。

**图 12.7.** 双活复制与无限复制环

![图12\.7 双向逻辑复制若不筛选来源，会在两个节点间反复发送同一变更](/img/fig-12-07.png)

PostgreSQL 16以前会无条件转发所有解码变更，双向配置中同一变更会在节点之间不断往返。PostgreSQL 16增加了订阅参数`origin`：

* `origin='any'`（默认）：无论WAL来自本地事务还是外部消息应用，都继续发送。
* `origin='none'`：过滤由外部节点消息应用所生成的WAL。

apply worker提交时，会在`COMMIT`或`ABORT` WAL记录上附加非零`origin_id`；walsender在ReorderBuffer中解码事务时读取来源。如果配置为`none`且发现来源标识，就跳过整个事务。

**图 12.8.** origin设置决定级联传播行为

![图12\.8 origin any允许跨节点传播，origin none在中间节点停止转发](/img/fig-12-08.png)

在Node 1 → Node 2 → Node 3的级联场景中，`any`允许Node 2继续转发带来源标记的变更；`none`则让Node 2应用变更但不再发往Node 3。所有参与节点都配置`origin='none'`即可避免双向拓扑中的无限循环。

### 12.1.5 逻辑复制槽

逻辑复制槽比物理槽多出若干属性，后续最重要的是：

* **`plugin`：** 逻辑解码输出插件名，例如`pgoutput`。
* **`database`：** 复制槽所属数据库。物理槽作用于整个实例，逻辑槽严格限定在单个数据库。
* **`confirmed_flush_lsn`：** 订阅端apply worker已经确认收到数据的位置。发布者不再需要保留该LSN之前已提交事务的解码数据。

逻辑槽还包含`catalog_xmin`、`two_phase`等字段，详见`pg_replication_slots`官方文档。

### 12.1.6 冲突

逻辑复制不复制DDL，也不会像物理流复制那样受到发布者VACUUM的影响，因此不会出现流复制的那些冲突。逻辑复制冲突主要来自订阅端应用程序的并发写入。例如，一行先在订阅端被本地删除，随后发布者又尝试更新同一行，就会产生`update_missing`冲突。

PostgreSQL能够检测的完整冲突列表参见[逻辑复制冲突官方文档](https://www.postgresql.org/docs/current/logical-replication-conflicts.html)。


## 12.2 启动逻辑复制

下面假设发布者已经存在需要复制的表，订阅者也已创建结构相同但尚无数据的表：

```sql
CREATE TABLE tbl_1 (id int PRIMARY KEY, name text, data int);
CREATE TABLE tbl_2 (id int NOT NULL UNIQUE, name text, data int);
CREATE INDEX tbl_2_idx ON tbl_2 (id, name);
CREATE TABLE tbl_3 (id int, name text, data int);
ALTER TABLE tbl_3 REPLICA IDENTITY FULL;
```

建立逻辑复制需要两条命令：在发布者执行`CREATE PUBLICATION`，在订阅者执行`CREATE SUBSCRIPTION`。

### 12.2.1 创建发布

[`CREATE PUBLICATION`](https://www.postgresql.org/docs/current/sql-createpublication.html)把发布信息写入`pg_publication`、`pg_publication_rel`和`pg_publication_namespace`，并在需要时更新`pg_class`：

```sql
testdb=# -- Publisher
testdb=# CREATE PUBLICATION mypub FOR TABLE tbl_1, tbl_2, tbl_3;
CREATE PUBLICATION

testdb=# SELECT pubname, puballtables, pubinsert, pubupdate, pubdelete, pubtruncate
           FROM pg_publication;
 pubname | puballtables | pubinsert | pubupdate | pubdelete | pubtruncate
---------+--------------+-----------+-----------+-----------+-------------
 mypub   | f            | t         | t         | t         | t
(1 row)

testdb=# SELECT * FROM pg_publication_tables;
 pubname | schemaname | tablename |    attnames    | rowfilter
---------+------------+-----------+----------------+-----------
 mypub   | public     | tbl_1     | {id,name,data} |
 mypub   | public     | tbl_2     | {id,name,data} |
 mypub   | public     | tbl_3     | {id,name,data} |
(3 rows)
```

系统视图`pg_publication_tables`展示发布与所属表之间的映射。

### 12.2.2 创建订阅

订阅者使用[`CREATE SUBSCRIPTION`](https://www.postgresql.org/docs/current/sql-createsubscription.html)创建订阅：

```sql
testdb=# -- Subscriber
testdb=# CREATE SUBSCRIPTION mysub
    CONNECTION 'host=192.168.3.10 port=5432 dbname=testdb'
    PUBLICATION mypub
    WITH (enabled = true, binary = false);
```

整个过程分为三个阶段：创建订阅并默认请求发布者上的逻辑复制槽；连接apply worker与发布者walsender；同步发布中的表。

**图 12.9.** 逻辑复制初始化序列

![图12\.9 创建订阅、启动apply worker并通过table sync worker复制初始表数据](/img/fig-12-09.png)

#### 12.2.2.1 阶段一：创建元数据和复制槽

订阅端首先完成两项工作：把订阅元数据写入`pg_subscription`与`pg_subscription_rel`；默认请求发布者创建逻辑复制槽。

1. 执行`CREATE SUBSCRIPTION`的postgres进程连接发布者walsender。
2. 它检查指定发布是否存在，取得目标表的模式和OID并验证结构兼容性；如果没有指定复用现有槽，则请求创建新槽。
3. 把订阅信息写入`pg_subscription`和`pg_subscription_rel`。
4. 发布者终止这条初始化连接所用的walsender。

```sql
testdb=# -- Subscriber
testdb=# SELECT subname, subenabled, substream, subslotname,
                subpublications, suborigin, subconninfo
           FROM pg_subscription;
 subname | subenabled | substream | subslotname | subpublications | suborigin | subconninfo
---------+------------+-----------+-------------+-----------------+-----------+-----------------------------------------------
 mysub   | t          | p         | mysub       | {mypub}         | any       | host=192.168.3.10 port=5432 dbname=testdb
(1 row)

testdb=# SELECT s.srsubid, s.srrelid, c.relname, s.srsublsn
           FROM pg_subscription_rel AS s, pg_class AS c
          WHERE c.oid = s.srrelid;
 srsubid | srrelid | relname | srsublsn
---------+---------+---------+-----------
   16436 |   16414 | tbl_1   | 0/1BF35E0
   16436 |   16422 | tbl_2   | 0/1BF35E0
   16436 |   16431 | tbl_3   | 0/1BF5320
(3 rows)
```

发布者上会出现使用`pgoutput`的逻辑槽：

```sql
testdb=# -- Publisher
testdb=# SELECT slot_name, plugin, slot_type, database, active,
                restart_lsn, confirmed_flush_lsn
           FROM pg_replication_slots;
 slot_name |  plugin  | slot_type | database | active | restart_lsn | confirmed_flush_lsn
-----------+----------+-----------+----------+--------+-------------+---------------------
 mysub     | pgoutput | logical   | testdb   | t      | 0/1BF5320  | 0/1BF5358
(1 row)
```

#### 12.2.2.2 阶段二：启动后台进程

5. Logical Replication Launcher启动apply worker。
6. apply worker连接发布者walsender，并初始化复制来源；若`pg_replication_origin`中还没有对应项，则创建一个。

此后，发布者walsender与订阅者apply worker会持续运行，负责发送和应用增量变更。

#### 12.2.2.3 阶段三：同步表

Logical Replication Launcher启动专门的**table sync worker**，从发布者取得目标表的现有行。为了提高效率，launcher会在配置上限内尽可能并发启动多个同步worker。

7. launcher启动table sync worker。
8. table sync worker连接发布者上的walsender。
9. walsender发送表数据，table sync worker把数据插入订阅端目标表。
10. 表追上主复制流后，table sync worker及其walsender终止，后续变更交给主apply worker。

walsender与table sync worker使用`COPY ... TO STDOUT`和`COPY ... FROM STDIN`协议高效传输并插入表数据，内部实现见[`tablesync.c`](https://github.com/postgres/postgres/blob/master/src/backend/replication/logical/tablesync.c)。

> PostgreSQL 17还提供[`pg_createsubscriber`](https://www.postgresql.org/docs/current/app-pgcreatesubscriber.html)，可以把物理备库转换为逻辑订阅者。


## 12.3 ReorderBuffer结构

每个walsender都会分配一块**ReorderBuffer**区域，其内存使用受[`logical_decoding_work_mem`](https://www.postgresql.org/docs/current/runtime-config-resource.html#GUC-LOGICAL-DECODING-WORK-MEM)限制，默认64 MB。

ReorderBuffer由三个核心组件组成：`ReorderBuffer`、`ReorderBufferTXN`与`ReorderBufferChange`。

**图 12.11.** ReorderBuffer结构

![图12\.11 ReorderBuffer按事务散列表关联ReorderBufferTXN及其变更链表](/img/fig-12-11.png)

`ReorderBuffer`最重要的成员是以事务ID为键的`by_txn`散列表。每个表项都是一个`ReorderBufferTXN`，保存某个事务的元数据和实际WAL变更。每次`INSERT`、`UPDATE`、`DELETE`等修改则表示为`ReorderBufferChange`，按LSN顺序追加到对应事务的`changes`双向链表。

### 12.3.1 `ReorderBuffer`

该结构保存逻辑解码的主上下文：

| 字段 | 类型 | 含义 |
|:-----|:-----|:-----|
| `by_txn` | `HTAB *` | 从`txid`映射到`ReorderBufferTXN`的散列表，用于快速查找活动事务。 |

### 12.3.2 `ReorderBufferTXN`

该结构管理单个事务及其变更：

| 字段 | 类型 | 含义 |
|:-----|:-----|:-----|
| `first_lsn` | `XLogRecPtr` | 此事务第一条变更记录的LSN。 |
| `final_lsn` | `XLogRecPtr` | 此事务提交或中止记录的LSN。 |
| `origin_id` | `RepOriginId` | 事务最初产生处的复制来源ID。 |
| `origin_lsn` | `XLogRecPtr` | 事务在来源发布者上的提交LSN。 |
| `base_snapshot` | `Snapshot` | 解码事务时使用的历史快照，确保扫描系统目录时可见性正确。 |
| `changes` | `dlist_head` | 按LSN顺序保存`ReorderBufferChange`的双向链表。 |

`base_snapshot`对于刚创建复制槽或事务修改系统目录时尤其重要；后文主要关注稳定运行阶段的数据流，因此不再展开。

### 12.3.3 `ReorderBufferChange`

该结构表示一次独立数据修改，主要字段如下：

| 字段 | 类型 | 含义 |
|:-----|:-----|:-----|
| `lsn` | `XLogRecPtr` | 产生该变更的WAL记录LSN。 |
| `action` | `ReorderBufferChangeType` | `INSERT`、`UPDATE`、`DELETE`、`TRUNCATE`等操作类型。 |
| `data` | `union` | 与操作类型对应的数据。 |
| `data.tp.rlocator` | `RelFileLocator` | 被修改关系的表空间OID、数据库OID和relfilenode编号。 |
| `data.tp.oldtuple` | `HeapTuple` | `UPDATE`或`DELETE`需要时保存的旧元组，取决于副本标识。 |
| `data.tp.newtuple` | `HeapTuple` | `INSERT`或`UPDATE`后的新元组。 |

核心结构可以简化为：

```c
typedef enum ReorderBufferChangeType
{
    REORDER_BUFFER_CHANGE_INSERT,
    REORDER_BUFFER_CHANGE_UPDATE,
    REORDER_BUFFER_CHANGE_DELETE,
    REORDER_BUFFER_CHANGE_MESSAGE,
    REORDER_BUFFER_CHANGE_INVALIDATION,
    REORDER_BUFFER_CHANGE_TRUNCATE
} ReorderBufferChangeType;

typedef struct ReorderBufferChange
{
    XLogRecPtr  lsn;
    ReorderBufferChangeType action;
    struct ReorderBufferTXN *txn;
    RepOriginId origin_id;

    union
    {
        struct
        {
            RelFileLocator rlocator;
            bool           clear_toast_afterwards;
            HeapTuple      oldtuple;
            HeapTuple      newtuple;
        } tp;

        struct
        {
            Size nrelids;
            bool cascade;
            bool restart_seqs;
            Oid *relids;
        } truncate;
    } data;

    dlist_node node;
} ReorderBufferChange;
```


## 12.4 WAL数据过滤与缓冲

本节介绍walsender如何通过多级过滤决定哪些WAL记录可以进入ReorderBuffer，并用一个并发事务示例展示ReorderBuffer如何累积变更；随后说明副本标识如何决定`UPDATE`、`DELETE`记录中的元组数据，以及超过内存上限后的磁盘溢出机制。

### 12.4.1 过滤

walsender通过多道过滤门决定是否解码一条WAL记录并把它加入ReorderBuffer的`changes`链表：

1. **数据库OID：** 立即丢弃不属于目标数据库的WAL记录。
2. **来源ID/回调：** PostgreSQL 16起，在数据进入ReorderBuffer之前检查来源。配置`origin='none'`时，丢弃来自其他节点（`origin_id > 0`）的变更，阻止无限复制环。
3. **发布缓存：** PostgreSQL 15起，在缓冲前按表过滤；不属于发布的表，以及被行过滤器排除的行，都不会进入缓冲区。
4. **关系类型/物理结构：** 过滤索引更新。逻辑复制只关心行级变化；订阅端修改元组后，本地索引机制会自动更新索引。

只有通过全部过滤门的变更才会存入ReorderBuffer。

> **PostgreSQL 14及以前：** 旧版本先检查数据库OID，再解码并缓冲所有变更，直到`COMMIT`时才按发布定义过滤。这种惰性过滤会让ReorderBuffer暂时保存最终必然丢弃的数据，浪费内存。

### 12.4.2 WAL数据缓冲

下面用两个并发事务说明walsender如何从原始WAL重建逻辑变更。

#### 12.4.2.1 目标关系

`tbl_a`与`tbl_b`都使用主键作为默认副本标识：

```sql
CREATE TABLE tbl_a (id int PRIMARY KEY, name text, data int);
CREATE TABLE tbl_b (id int PRIMARY KEY, name text, data int);

INSERT INTO tbl_a VALUES (1, 'Alice', 100);
INSERT INTO tbl_b VALUES (10, 'Ken', 100);
```

#### 12.4.2.2 交错事务场景

两个事务`txid=840`与`841`按以下时间线交错执行：

```text
时刻  txid 840                                  txid 841
T0    BEGIN                                     BEGIN
T1    INSERT INTO tbl_a VALUES(2,'Bob',200)
T2                                              INSERT INTO tbl_a VALUES(3,'Candy',3)
T3    INSERT INTO tbl_b VALUES(11,'Luke',110)
T4                                              UPDATE tbl_a SET data=data+1 WHERE id=1
T5                                              UPDATE tbl_a SET data=data+1 WHERE id=1
T6    DELETE FROM tbl_b WHERE id=10
T7    COMMIT
T8                                              COMMIT
```

事务隔离级别不会影响逻辑解码。隔离级别在事务运行时控制元组可见性；逻辑解码处理的则是已经完成可见性检查、写入WAL的历史结果，它只负责按事务重组这些结果。

#### 12.4.2.3 ReorderBuffer状态变化

##### T1：`txid=840`向`tbl_a`插入

系统为`txid=840`创建`ReorderBufferTXN`，把`first_lsn`设为该`INSERT`记录的LSN，并把变更加入`changes`链表。

**图 12.12.** T1后的ReorderBuffer

![图12\.12 txid 840创建事务缓冲并加入第一条INSERT变更](/img/fig-12-12.png)

`ReorderBufferChange`保存LSN、`INSERT`等动作类型、目标表空间/数据库/关系OID和实际元组数据。

> 当`wal_level=logical`时，整页写入的main data部分包含实际被修改的元组。walsender可以直接从main data取得元组，而无需解析整个页面块，从而降低逻辑解码开销。

##### T2：`txid=841`向`tbl_a`插入

系统为`txid=841`创建另一份`ReorderBufferTXN`。两个事务的变更存入相互独立的子链表。

**图 12.13.** T2后的ReorderBuffer

![图12\.13 txid 840和841分别拥有独立变更链表](/img/fig-12-13.png)

##### T3：`txid=840`向`tbl_b`插入

新的`INSERT`变更追加到`txid=840`自己的`changes`链表。

**图 12.14.** T3后的ReorderBuffer

![图12\.14 txid 840的第二条INSERT按LSN顺序追加](/img/fig-12-14.png)

##### T4与T5：`txid=841`两次更新`tbl_a`

第一次与第二次`UPDATE`依次追加到`txid=841`的变更链表。

**图 12.15.** T4后的ReorderBuffer

![图12\.15 txid 841追加第一条UPDATE](/img/fig-12-15.png)

**图 12.16.** T5后的ReorderBuffer

![图12\.16 txid 841再次更新同一元组](/img/fig-12-16.png)

在`REPLICA IDENTITY DEFAULT`下，只有主键列被修改时才需要`oldtuple`。本例的主键`id`未变，订阅者可使用`newtuple`中的`id=1`找到目标元组，因此省略`oldtuple`。

##### T6：`txid=840`删除`tbl_b`中的行

`DELETE`记录追加到`txid=840`的链表，只保存作为副本标识的主键数据。

**图 12.17.** T6后的ReorderBuffer

![图12\.17 txid 840追加DELETE变更](/img/fig-12-17.png)

**图 12.18.** DELETE变更只保存主键

![图12\.18 DELETE的oldtuple只包含定位目标行所需的主键](/img/fig-12-18.png)

订阅者删除目标元组只需要键值；省略`name`、`data`等非键列既减少ReorderBuffer内存，也降低网络流量。

##### T7与T8：事务提交

`txid=840`提交时，ReorderBuffer从链表头开始遍历其变更。输出插件把每条变更编码为复制消息并发送；完成后移除该`ReorderBufferTXN`。

**图 12.19.** txid 840提交后的ReorderBuffer

![图12\.19 txid 840的变更被发送并从缓冲区移除，只剩txid 841](/img/fig-12-19.png)

`txid=841`在T8提交时也按同样方式重组并发送。

### 12.4.3 副本标识与解码元组

`UPDATE`和`DELETE`写入WAL的元组内容由目标表的副本标识决定。

#### 12.4.3.1 `UPDATE`

当副本标识为主键或指定索引时：

| 更新类型 | `oldtuple` | `newtuple` |
|:---------|:-----------|:-----------|
| 只更新非键列 | 无 | 完整新元组 |
| 更新键列 | 旧键值 | 完整新元组 |

**图 12.20.** 键式副本标识下的UPDATE载荷

![图12\.20 修改非键列时省略oldtuple，修改键列时保存旧键](/img/fig-12-20.png)

如果更新副本标识中的键列，订阅者需要用`oldtuple`中的旧键定位原行：

```sql
-- 伪SQL：使用旧键定位
UPDATE tbl_a SET id = id + 100 WHERE id = 1;
```

若只修改非键列，`oldtuple`为空，订阅者可使用`newtuple`中仍未改变的键：

```sql
-- 伪SQL：使用新元组中的当前键定位
UPDATE tbl_a SET data = data + 100 WHERE id = 1;
```

apply worker并不会拼接或执行文本SQL；它复用执行器基础设施，但绕过解析与计划阶段，直接使用索引扫描或顺序扫描查找并修改目标元组。

`REPLICA IDENTITY FULL`则让`oldtuple`保存更新前的完整元组，`newtuple`保存更新后的完整元组：

**图 12.21.** FULL副本标识下的UPDATE载荷

![图12\.21 REPLICA IDENTITY FULL同时携带完整旧元组与新元组](/img/fig-12-21.png)

没有正式唯一键时，订阅者需要比较所有列来定位目标行：

```sql
-- 伪SQL：用所有旧列值匹配
UPDATE tbl_a SET data = data + 100
 WHERE id = 1 AND name = 'Alice' AND data = 100;
```

#### 12.4.3.2 `DELETE`

`DELETE`没有后续状态，因此只记录`oldtuple`：

| 副本标识 | `oldtuple` | `newtuple` |
|:---------|:-----------|:-----------|
| 主键/索引 | 键列 | 无 |
| `FULL` | 完整旧元组 | 无 |

**图 12.22.** DELETE通过oldtuple定位元组

![图12\.22 键式标识只携带键列，FULL携带完整旧元组](/img/fig-12-22.png)

### 12.4.4 ReorderBuffer内存管理与磁盘溢出

为了避免大事务耗尽内存，当全部缓冲事务的累计内存超过`logical_decoding_work_mem`时，ReorderBuffer会把部分变更**溢出到磁盘（spill-to-disk）**。

#### 12.4.4.1 溢出算法

1. 每加入一条`ReorderBufferChange`，累计内存增加相应大小。
2. 将当前用量与`logical_decoding_work_mem`比较。
3. 超出阈值时，找到当前积累变更最多的“最重”事务。
4. 把该事务的全部已缓冲变更序列化到`$PGDATA/pg_replslot/<slot_name>`中的`.spill`文件。
5. 释放这些变更占用的内存，但保留`ReorderBufferTXN`元数据。
6. 把事务标为`serialized=true`，提交解码时再从磁盘读回。

**图 12.23.** ReorderBuffer内存管理与磁盘溢出

![图12\.23 缓冲区满时选择变更最多的事务写入spill文件并释放空间](/img/fig-12-23.png)

PostgreSQL只溢出最重事务，而不是同时刷新所有事务，以较少的磁盘I/O把内存维持在上限内。

#### 12.4.4.2 溢出文件

溢出文件位于复制槽目录：

```text
$PGDATA/pg_replslot/<slot_name>/
    xid-856-lsn-0-6000000.spill
    xid-856-lsn-0-7000000.spill
    xid-856-lsn-0-8000000.spill
```

命名格式为`xid-[XID]-lsn-[LSN_HIGH]-[LSN_LOW].spill`：XID使用十进制；LSN高32位和低32位使用十六进制。

```bash
$ ls -lh $PGDATA/pg_replslot/myslot/
total 68M
-rw------- 1 postgres postgres  200 Mar 24 08:12 state
-rw------- 1 postgres postgres  30M Mar 24 08:12 xid-856-lsn-0-6000000.spill
-rw------- 1 postgres postgres  34M Mar 24 08:12 xid-856-lsn-0-7000000.spill
-rw------- 1 postgres postgres 4.5M Mar 24 08:12 xid-856-lsn-0-8000000.spill
```


## 12.5 逻辑解码输出插件：`pgoutput`

输出插件把ReorderBuffer已经组织、排序好的变更数据序列化为订阅者能够理解的格式，主要完成两项工作：

* **最终过滤与转换：** 再次根据发布配置检查目标表、操作类型、行过滤器和列列表，丢弃无关数据，并按协议要求编码为文本或二进制。
* **生成消息：** 构造`BEGIN`、DML、`TRUNCATE`、`COMMIT`等消息序列，交给walsender发送。

这一机制是可插拔的；自定义插件可以输出不同格式或实现特殊逻辑。标准PostgreSQL逻辑复制使用官方插件`pgoutput`。下文列出核心消息结构；完整的协议版本差异参见[逻辑复制消息格式](https://www.postgresql.org/docs/current/protocol-logicalrep-message-formats.html)。

### 12.5.1 `TupleData`公共子消息

`TupleData`在DML消息中表示元组内容：

```text
[num_cols: Int16]
  对每一列：
    [kind: Byte1('n'|'u'|'t'|'b')]
    如果kind为't'或'b'：
      [length: Int32] [value: Byte*n*]
```

| 字段 | 类型 | 含义 |
|:-----|:-----|:-----|
| `num_cols` | `Int16` | 元组列数。 |
| `kind` | `Byte1` | `n`=NULL；`u`=未改变的TOAST值；`t`=文本格式；`b`=二进制格式。 |
| `length` | `Int32` | `t`或`b`值的字节长度。 |
| `value` | `Byte*n*` | 实际列值。 |

### 12.5.2 事务控制

#### 12.5.2.1 Begin（`B`）

标识事务开始，并携带最终LSN、提交时间与事务ID：

```text
[Byte1('B')] [final_lsn: Int64] [commit_timestamp: Int64] [txid: Int32]
```

`final_lsn`通常是提交LSN；两阶段提交时则是prepare LSN。`commit_timestamp`是从PostgreSQL纪元（2000-01-01）开始计算的微秒时间戳。

#### 12.5.2.2 Commit（`C`）

标识事务结束。订阅者收到后，把此前累积的变更作为一个原子单元在本地提交：

```text
[Byte1('C')] [flags: Int8] [commit_lsn: Int64]
[end_lsn: Int64] [commit_timestamp: Int64]
```

`flags`当前保留且为0；`commit_lsn`是提交记录LSN，`end_lsn`是事务末尾LSN。

### 12.5.3 DML消息

#### 12.5.3.1 Origin（`O`）

多节点复制中用于说明事务最初产生的位置：

```text
[Byte1('O')] [origin_lsn: Int64] [origin_name: String]
```

一个事务中可以出现多个Origin消息；`origin_lsn`是来源服务器上的提交LSN。

#### 12.5.3.2 Relation（`R`）

提供表的元数据，通常在会话中第一次发送某表的DML消息前发送，把关系OID映射到模式、表名和列定义：

```text
[Byte1('R')] [rel_id: Int32] [namespace: String] [relname: String]
             [replica_identity: Int8] [num_columns: Int16]
  对每一列：
    [flags: Int8] [name: String] [type_oid: Int32] [atttypmod: Int32]
```

`replica_identity`对应`pg_class.relreplident`的`d`、`n`、`f`、`i`；列`flags=1`表示该列属于副本标识键。

#### 12.5.3.3 Insert（`I`）

```text
[Byte1('I')] [rel_id: Int32] [Byte1('N')] [new_tuple: TupleData]
```

`N`表示随后是新元组，`rel_id`对应此前Relation消息中的关系OID。

#### 12.5.3.4 Update（`U`）

消息形态取决于副本标识以及键列是否改变：

```text
-- DEFAULT或INDEX，键列发生变化：
[Byte1('U')] [rel_id: Int32]
[Byte1('K')] [old_tuple: TupleData]
[Byte1('N')] [new_tuple: TupleData]

-- REPLICA IDENTITY FULL：
[Byte1('U')] [rel_id: Int32]
[Byte1('O')] [old_tuple: TupleData]
[Byte1('N')] [new_tuple: TupleData]

-- DEFAULT或INDEX，键列未变化：
[Byte1('U')] [rel_id: Int32]
[Byte1('N')] [new_tuple: TupleData]
```

`K`表示随后是旧键，`O`表示完整旧元组，两者互斥；`N`始终标记新元组。

#### 12.5.3.5 Delete（`D`）

```text
-- DEFAULT或INDEX：
[Byte1('D')] [rel_id: Int32]
[Byte1('K')] [old_key_tuple: TupleData]

-- REPLICA IDENTITY FULL：
[Byte1('D')] [rel_id: Int32]
[Byte1('O')] [old_tuple: TupleData]
```

订阅者使用给出的旧键或完整旧元组定位并删除记录。

#### 12.5.3.6 Truncate（`T`）

表示批量清空一张或多张表：

```text
[Byte1('T')] [num_relations: Int32] [options: Int8]
             [rel_id: Int32] ...
```

`options`的位0表示`CASCADE`，位1表示`RESTART IDENTITY`；后续重复`num_relations`个关系OID。

### 12.5.4 流控制消息

发布者为支持大事务流式传输而分段发送数据时，使用以下消息。

#### 12.5.4.1 Stream Start（`S`）

```text
[Byte1('S')] [txid: Int32] [first_segment: Int8]
```

标识一个流片段开始；`first_segment=1`表示这是该XID的首段。

#### 12.5.4.2 Stream Stop（`E`）

```text
[Byte1('E')]
```

标识当前流片段结束。

#### 12.5.4.3 Stream Commit（`c`）

```text
[Byte1('c')] [txid: Int32] [flags: Int8]
[commit_lsn: Int64] [end_lsn: Int64] [commit_timestamp: Int64]
```

标识流式事务提交，字段与普通Commit消息类似，但额外显式携带`txid`。

#### 12.5.4.4 Stream Abort（`A`）

```text
[Byte1('A')] [txid: Int32] [subxid: Int32]
[abort_lsn: Int64]? [abort_timestamp: Int64]?
```

标识流式事务或子事务中止。`subxid`在顶层事务中等于`txid`；`abort_lsn`与`abort_timestamp`只在支持并行流式应用的协议v4及以后出现。


## 12.6 事务重组与消息发送

本节说明ReorderBuffer如何重组累积的变更，并通过输出插件发送给订阅者apply worker。

### 12.6.1 事务重组概要

walsender在WAL中发现`COMMIT`记录后，ReorderBuffer开始重组事务：

1. 找到已提交事务的`ReorderBufferTXN`。
2. 如果存在子事务，把子事务变更链表合并到顶层事务。
3. 按LSN排序最终变更列表，保证时间顺序。
4. 检查提交或中止记录中的复制来源；存在时把来源信息加入复制消息。
5. 依次遍历变更并交给`pgoutput`等输出插件。
6. 释放`ReorderBufferTXN`与所有`ReorderBufferChange`对象。

不含子事务时，单个事务链表天然按LSN排序，只需正向遍历。若事务中止，ReorderBuffer会立即丢弃该事务及全部变更，不交给输出插件。

#### 12.6.1.1 示例

**图 12.24.** txid 840的ReorderBufferTXN与变更

![图12\.24 txid 840包含两条INSERT和一条DELETE](/img/fig-12-24.png)

`txid=840`经`pgoutput`生成的消息序列为：

```text
[B] lsn=0/1CAA128  ts=2026-03-28T16:55:00  txid=840
[R] oid=16456  "public"."tbl_a"  'd'  3cols  [id:int4(key)][name:text][data:int4]
[I] oid=16456  N  [t"2"][t"Bob"][t"2"]
[R] oid=16464  "public"."tbl_b"  'd'  3cols  [id:int4(key)][name:text][data:int4]
[I] oid=16464  N  [t"11"][t"Luke"][t"110"]
[D] oid=16464  K  [t"10"][u][u]
[C] flags=0  commit=0/1CAA128  end=0/1CAA200  ts=2026-03-28T16:55:10
```

**图 12.25.** txid 841的ReorderBufferTXN与变更

![图12\.25 txid 841包含一条INSERT和两条UPDATE](/img/fig-12-25.png)

`txid=841`的消息序列为：

```text
[B] lsn=0/1CA9E00  ts=2026-03-28T16:55:05  txid=841
[R] oid=16456  "public"."tbl_a"  'd'  3cols  [id:int4(key)][name:text][data:int4]
[I] oid=16456  N  [t"3"][t"Candy"][t"3"]
[U] oid=16456  N  [t"1"][t"Alice"][t"2"]
[U] oid=16456  N  [t"1"][t"Alice"][t"3"]
[C] flags=0  commit=0/1CA9E00  end=0/1CA9F00  ts=2026-03-28T16:55:15
```

Relation（`R`）消息只在当前walsender会话尚未发送过该表元数据，或表定义发生变化时发送，以避免重复传输。

### 12.6.2 发送消息

输出插件完成序列化后，消息会封装进逻辑复制协议并经网络发送。

**图 12.26.** 普通非流式事务的消息序列

![图12\.26 发布者提交后，订阅端一次连续接收Begin到Commit的完整消息流](/img/fig-12-26.png)

普通事务只有在发布者提交后，apply worker才会收到从`Begin`到`Commit`的完整连续序列。

`pgoutput`消息是二进制格式，不能直接作为文本查看。调试时可以把[`pg_recvlogical`](https://www.postgresql.org/docs/current/app-pgrecvlogical.html)输出交给`od`：

```bash
$ pg_recvlogical -d testdb --slot myslot --create-slot -P pgoutput
$ pg_recvlogical -d testdb --slot=myslot --start -f - \
    -o proto_version=1 -o publication_names='my_publication' | od -c -A n
```

若希望查看可读输出，可以使用`test_decoding`插件：

```bash
$ pg_recvlogical -d testdb --slot myslot --create-slot -P test_decoding
$ pg_recvlogical -d testdb --slot=myslot --start -f -

BEGIN 840
table public.tbl_a: INSERT: id[integer]:2 name[text]:'Bob' data[integer]:2
table public.tbl_b: INSERT: id[integer]:11 name[text]:'Luke' data[integer]:110
table public.tbl_b: DELETE: id[integer]:10
COMMIT 840
```

### 12.6.3 大事务流式传输

当`streaming`为`on`或`parallel`且ReorderBuffer达到内存上限时，PostgreSQL不再把变更写入发布端本地spill文件，而是立即发送给订阅端apply worker或leader apply worker。

**图 12.27.** 流式事务的分段消息序列

![图12\.27 大事务以Stream Start和Stream Stop分段，最后发送Stream Commit](/img/fig-12-27.png)

假设溢出时`txid=842`占用内存最多，发布者先重排它的变更，再用Stream Start（`S`）与Stream Stop（`E`）包住形成一个片段。首段的`S`取代普通事务的Begin消息。

#### 12.6.3.1 第一段

```text
[S] txid=842  first_segment=1
[R] oid=16456  "public"."tbl_a"  'd'  3cols  [id:int4(key)][name:text][data:int4]
[I] txid=842  oid=16456  N  [t"1"][t"Data1"][t"100"]
[I] txid=842  oid=16456  N  [t"2"][t"Data2"][t"200"]
... 数千条INSERT ...
[I] txid=842  oid=16456  N  [t"50000"][t"Data50000"][t"5000000"]
[E]
```

后续再次达到阈值时继续发送新片段，`first_segment`改为0。

#### 12.6.3.2 第二段

```text
[S] txid=842  first_segment=0
[I] txid=842  oid=16456  N  [t"50001"][t"Data50001"][t"5000100"]
... 后续INSERT ...
[I] txid=842  oid=16456  N  [t"100000"][t"Data100000"][t"10000000"]
[E]
```

#### 12.6.3.3 流式提交

发布者最终提交`txid=842`时发送Stream Commit：

```text
[c] txid=842  flags=0  commit=0/1CB0000  end=0/1CB0100  ts=2026-03-28T17:10:00
```

若事务中止，则发送Stream Abort（`A`），通知订阅端丢弃此前收到的全部片段。


## 12.7 Apply Worker与事务重放

apply worker负责把发布者发送的逻辑变更重放到订阅者本地表。本节先介绍普通模式，再说明大事务流式处理与并行apply worker。

### 12.7.1 普通模式（`streaming=off`）

订阅端apply worker执行以下主要工作：

* **来源过滤：** 检查消息来源，决定应用还是跳过，防止复制环。
* **LSN跳过检查（幂等性）：** 把事务LSN与`pg_replication_origin_status`中持久化的`remote_lsn`比较；已经应用过的事务直接跳过。
* **分派：** 按消息类型交给`INSERT`、`UPDATE`、`DELETE`等处理器。
* **冲突检测：** 检测更新目标行不存在、唯一约束冲突等问题。会产生错误的冲突将停止复制；`update_missing`、`delete_missing`等缺行冲突则记录后跳过相应操作。

#### 12.7.1.1 来源过滤与LSN跟踪

是否应用事务取决于订阅参数`origin`及消息中的`origin_id`。当前实现中，如果配置`origin='none'`且`origin_id>0`，apply worker跳过重放；否则继续。

#### 12.7.1.2 消息分派与重放

假设发布者向`tbl_a`插入一行，消息流为：

```text
[B] lsn=0/1CA96C0  ts=2026-03-29T16:55:05  txid=845
[R] oid=16456  "public"."tbl_a"  'd'  3cols  [id:int4(key)][name:text][data:int4]
[I] oid=16456  N  [t"3"][t"Candy"][t"3"]
[C] flags=0  commit=0/1CA9E00  end=0/1CA9F00  ts=2026-03-29T16:55:15
```

处理流程如下：

1. **Begin：** 开启本地事务，记录发布者提交LSN和时间戳，并把`session_replication_role`设为`replica`，使本地触发器和约束按其复制角色运行。
2. **Relation：** 接收表定义，更新`RelationSyncCache`，依据模式名与表名把发布端OID映射到本地表OID。
3. **Insert：** 把二进制元组转为本地表格式，调用内部插入逻辑直接写表并更新索引；该操作也会在订阅者生成自己的WAL。
4. **Commit：** 提交本地事务，把对应发布者的`origin_id`附加到提交WAL；更新复制来源的已应用LSN，并向walsender返回已刷盘ACK。

订阅端提交WAL会附带发布者元数据：`origin_id`、作为`origin_lsn`的`final_lsn`，以及作为`origin_timestamp`的提交时间。`UPDATE`与`DELETE`还需要使用索引或顺序扫描直接查找目标元组。

#### 12.7.1.3 `REPLICA IDENTITY FULL`优化

PostgreSQL 15及以前，`REPLICA IDENTITY FULL`表在应用`UPDATE`或`DELETE`时必须顺序扫描定位目标元组，大表开销很高。PostgreSQL 16起，apply worker可以使用现有的合适索引；`FindLogicalRepLocalIndex()`会寻找能够定位行的非部分B-tree索引，显著减少顺序扫描。

### 12.7.2 流式模式（`streaming=on`）

此模式下，apply worker在最终提交前就会分段收到尚在进行中的事务。假设`txid=842`是分两段发送的大事务，而`txid=843`是普通小事务：

**图 12.28.** streaming\=on时的逻辑变更重放

![图12\.28 大事务片段先缓冲，小事务可立即应用，大事务提交时再顺序重放](/img/fig-12-28.png)

1. 收到`txid=842`首段后，用`StringInfoData original_msg`在内存中累积消息。
2. 普通`txid=843`到达后，立即解码并应用。
3. `txid=842`第二段追加到原有缓冲区。
4. 收到Stream Commit后，解码`txid=842`的全部累积变更并顺序重放。

因此，普通消息立即处理，流式片段则先缓冲到提交消息到达。

#### 12.7.2.1 内存压力与临时文件

流式变更缓存同样受`logical_decoding_work_mem`限制。超过上限后，apply worker把数据写入`$PGDATA/base/pgsql_tmp/`下的临时文件：

```text
目录：$PGDATA/base/pgsql_tmp/pgsql_tmp[PID].[FilesetSerial].fileset/
文件：[SubID]-[txid].changes.[SegmentID]
```

例如订阅OID为`16403`、事务ID为`767`、apply worker PID为`2212`：

```text
base/pgsql_tmp/pgsql_tmp2212.0.fileset/16403-767.changes.0
```

事务提交或中止后，PostgreSQL自动删除这些临时文件。

### 12.7.3 并行Apply Worker（`streaming=parallel`）

PostgreSQL 16引入并行应用，允许订阅端并发重放多个事务。与`streaming=on`先缓冲到提交不同，被分配到parallel apply worker的变更会在到达时立即应用。

* **leader apply worker：** 维持walsender连接，并为收到的变更分派parallel worker。
* **parallel apply worker：** 由leader启动并并发应用事务。
* **worker上限：** 由`max_parallel_apply_workers_per_subscription`控制。
* **退化机制：** 活跃流式事务超过上限时，leader对额外事务退回`streaming=on`，把片段缓存在内存或磁盘直到提交。

#### 12.7.3.1 并行重放序列

假设`txid=845`是分两段发送的大事务，`txid=846`是单元组事务：

**图 12.29.** 并行apply模式下的事务重放

![图12\.29 leader把不同事务分派给parallel workers并在收到时立即应用](/img/fig-12-29.png)

1. leader收到`txid=845`首段，启动或分配一个parallel worker。
2. 把首段交给该worker，立即开始应用。
3. 收到`txid=846`后，为它启动或分配另一个worker并立即应用。
4. `txid=845`第二段到达后，转给原先分配的worker继续应用。
5. leader收到`txid=845`的Stream Commit，通知该worker提交本地事务并回到就绪状态。

如果事务已经分配worker，后续片段始终转给同一worker；没有分配且仍低于上限时启动新worker；达到上限时则由leader缓冲并在提交后串行应用。


## 12.8 重启与崩溃恢复的内部机制

本节先说明逻辑复制如何管理WAL位置，再介绍订阅者正常重启与崩溃恢复的序列。

### 12.8.1 WAL数据管理

复制进度以WAL的LSN为基准。物理流复制的主备共享同一套WAL空间；逻辑复制的发布者与订阅者各自使用独立LSN，因此必须维护两套WAL空间之间的**映射**。

PostgreSQL由订阅者负责这项映射：发布者始终使用自己的LSN，订阅者记录本地LSN与从发布者收到的远端LSN之间的对应关系。

#### 12.8.1.1 发布者与订阅者的LSN状态

发布者在逻辑复制槽中维护`confirmed_flush_lsn`：它表示复制槽消费者已经确认接收的位置；在此位置之前提交的事务不再需要提供给复制。复制槽状态保存在内存中，通常在检查点时持久化。

订阅者提交或中止由复制产生的本地事务时，会在自己的WAL中记录发布者元数据：`origin_id`、发布者提交LSN与提交时间。同时在内存中维护：

* **`local_lsn`：** 订阅者本地提交记录的LSN。
* **`remote_lsn`：** 与该提交对应的发布者`final_lsn`。

这些值可通过`pg_replication_origin_status`查看，并在每次检查点写入`$PGDATA/pg_logical/replorigin_checkpoint`。订阅者意外崩溃时，该文件中的最新映射可能落后；恢复过程会扫描WAL自动重建。

#### 12.8.1.2 正常事务中的LSN流

**图 12.30.** 正常事务中的LSN映射

![图12\.30 发布者LSN、订阅者本地LSN、复制来源状态和confirmed\_flush\_lsn的更新](/img/fig-12-30.png)

1. 复制槽当前`confirmed_flush_lsn`为$LSN^P_0$。发布者执行`txid=100`的`INSERT`，提交WAL从$LSN^P_1$开始，到$LSN^P_2$结束；`pgoutput`在Begin中发送`final_lsn=LSN^P_1`，在Commit中发送`commit_lsn=LSN^P_1`、`end_lsn=LSN^P_2`。
2. 订阅者内存最初保存`local_lsn=LSN^S_0`、`remote_lsn=LSN^P_0`。apply worker应用变更，并在订阅者WAL的$LSN^S_1$写入提交记录，其中包含`origin_id`和发布者$LSN^P_1$；内存映射随即更新为$LSN^S_1 \leftrightarrow LSN^P_1$，但`replorigin_checkpoint`要到下次检查点才更新。
3. 事务完成后，订阅者返回包含`write_lsn`、`flush_lsn`与`apply_lsn`的ACK；这些值使用发布者WAL空间，本例均为$LSN^P_1$。发布者据此把复制槽的`confirmed_flush_lsn`推进到$LSN^P_1$。

### 12.8.2 重启序列

任一端正常重启导致复制停止后，逻辑复制会自动恢复。

**图 12.31.** 逻辑复制重启序列

![图12\.31 订阅者读取来源检查点、启动apply worker并与walsender协商remote\_lsn](/img/fig-12-31.png)

1. 订阅者读取`replorigin_checkpoint`，初始化`local_lsn`和`remote_lsn`。
2. logical replication launcher启动apply worker。
3. apply worker向发布者请求连接。
4. 发布者postmaster启动walsender。
5. apply worker与walsender建立连接。
6. 双方使用`remote_lsn`协商起点；订阅者发送远端位置，walsender从发布者WAL中的该位置开始解码。
7. 复制从指定LSN继续。

### 12.8.3 崩溃恢复序列

订阅者崩溃时，`replorigin_checkpoint`中的映射只反映上一次检查点，可能已经过期。因此连接发布者前必须扫描本地WAL，重建最新映射。

**图 12.32.** 逻辑复制崩溃恢复序列

![图12\.32 崩溃恢复扫描订阅者WAL重建复制来源LSN，再按正常序列重连](/img/fig-12-32.png)

1. **扫描WAL段：** 在常规崩溃恢复过程中，从源自发布者的提交记录中提取`origin_id`与`origin_lsn`。
2. **恢复来源状态：** 恢复完成后，把最新`origin_lsn`及其对应的订阅者提交LSN恢复到内存，分别作为新的`remote_lsn`与`local_lsn`，并重建`replorigin_checkpoint`。
3. 随后进入普通重启序列，与发布者重新建立连接。