reorderbuffer.c 源码深度解析:PostgreSQL 逻辑复制的"事务重组引擎

引言:为什么 reorder buffer 是逻辑复制的核心

PostgreSQL 的逻辑复制链路:

reorderbuffer.c 位于整条链路的 中心位置。它承担三个职责:

1. 事务重组:把 WAL 顺序记录按 xid 分组,处理 subxact 嵌套,等待事务结束再按提交顺序输出
2. 变更缓存:在内存中暂存所有 DML,直到事务 commit/prepare 后再 dispatch
3. 背压管理:logical_decoding_work_mem 触发 spill-to-disk + streaming,避免 OOM

如果把逻辑复制类比成”消息队列”:reorder buffer 就是 那个 broker,负责按事务维度把无序的 WAL 流重组为有序的变更流。

1. 文件定位与规模

核心定位:reorder buffer 是逻辑解码的 **”in-memory + on-disk 事务仓库”**。

2. 核心数据结构全景

2.1 ReorderBuffer 顶层结构(reorderbuffer.h:574-695)

struct ReorderBuffer
{
    HTAB *by_txn;                    // xid -> ReorderBufferTXN 的主索引
    dlist_head toplevel_by_lsn;      // toplevel xact 按 first_lsn 排序
    dlist_head txns_by_base_snapshot_lsn;  // 有 base snapshot 的事务
    dclist_head catchange_txns;      // 修改了 catalog 的事务

    TransactionId by_txn_last_xid;   // 1-entry cache for by_txn
    ReorderBufferTXN *by_txn_last_txn;

    // 16 个 callback 函数指针
    ReorderBufferBeginCB begin;
    ReorderBufferApplyChangeCB apply_change;
    ReorderBufferApplyTruncateCB apply_truncate;
    ReorderBufferCommitCB commit;
    ReorderBufferMessageCB message;
    ReorderBufferBeginCB begin_prepare;
    ReorderBufferPrepareCB prepare;
    ReorderBufferCommitPreparedCB commit_prepared;
    ReorderBufferRollbackPreparedCB rollback_prepared;
    ReorderBufferStreamStartCB stream_start;
    ReorderBufferStreamStopCB stream_stop;
    ReorderBufferStreamAbortCB stream_abort;
    ReorderBufferStreamPrepareCB stream_prepare;
    ReorderBufferStreamCommitCB stream_commit;
    ReorderBufferStreamChangeCB stream_change;
    ReorderBufferStreamMessageCB stream_message;
    ReorderBufferStreamTruncateCB stream_truncate;
    ReorderBufferUpdateProgressTxnCB update_progress_txn;

    void *private_data;              // 指向 LogicalDecodingContext

    MemoryContext context;           // 顶层 context
    MemoryContext change_context;    // Slab: ReorderBufferChange
    MemoryContext txn_context;       // Slab: ReorderBufferTXN
    MemoryContext tup_context;       // Generation: tuple data

    char *outbuf;                    // disk<->memory 转换 buffer
    Size outbufsize;
    Size size;                       // 当前总内存使用

    pairingheap *txn_heap;           // 按 total_size 排序的 max-heap

    // 6 个统计计数器
    int64 spillTxns, spillCount, spillBytes;
    int64 streamTxns, streamCount, streamBytes;
    int64 totalTxns, totalBytes;
};

2.2 ReorderBufferTXN 事务结构(reorderbuffer.h:293-470)

关键字段分组:

2.3 ReorderBufferChange 变更结构(reorderbuffer.h:76-191)

13 种 action 决定了 union 的语义:

typedef enum ReorderBufferChangeType {
    REORDER_BUFFER_CHANGE_INSERT,              // INSERT
    REORDER_BUFFER_CHANGE_UPDATE,              // UPDATE (old+new)
    REORDER_BUFFER_CHANGE_DELETE,              // DELETE (old)
    REORDER_BUFFER_CHANGE_MESSAGE,             // 消息(pg_logical_emit_message)
    REORDER_BUFFER_CHANGE_INVALIDATION,        // cache invalidation
    REORDER_BUFFER_CHANGE_INTERNAL_SNAPSHOT,   // 内部:新 snapshot
    REORDER_BUFFER_CHANGE_INTERNAL_COMMAND_ID, // 内部:新 CID
    REORDER_BUFFER_CHANGE_INTERNAL_TUPLECID,   // 内部:新 cmin/cmap
    REORDER_BUFFER_CHANGE_INTERNAL_SPEC_INSERT,   // 内部:INSERT ON CONFLICT
    REORDER_BUFFER_CHANGE_INTERNAL_SPEC_CONFIRM,
    REORDER_BUFFER_CHANGE_INTERNAL_SPEC_ABORT,
    REORDER_BUFFER_CHANGE_TRUNCATE,            // TRUNCATE
} ReorderBufferChangeType;

关键洞察:INTERNAL_* 类型是 reorder buffer 自己产生的元数据变更,不是用户 DML。output plugin 看不到这些。

3. 4 套索引与 3 套内存管理

3.1 4 套索引

reorder buffer 为事务维护 4 套索引,服务于不同场景:

1-entry cache 的设计意图 (reorderbuffer.c:651-665):

// by_txn_last_xid + by_txn_last_txn
// 99% 的连续操作都是同一个 xid 的变更(同事务内)
if (rb->by_txn_last_xid == xid) {
    if (rb->by_txn_last_txn != NULL) {
        if (is_new) *is_new = false;
        return rb->by_txn_last_txn;       // 命中 cache
    }
    if (!create) return NULL;              // 之前查过不存在
}

这个优化能省掉绝大多数 hash 查表,在 streaming workload 下收益巨大。

3.2 3 套 MemoryContext

ReorderBufferAllocate (reorderbuffer.c:324-413) 创建 4 个 context:

为什么 Tuple 用 Generation 而不是 Slab?

reorderbuffer.c:381-390 的注释:

To minimize memory fragmentation caused by long-running transactions with changes spanning multiple memory blocks, we use a single fixed-size memory block for decoded tuple storage. The performance testing showed that the default memory block size maintains logical decoding performance without causing fragmentation due to concurrent transactions.

Generation 上下文允许在事务 commit 时一次性 reset 所有 tuple 内存,而不需要逐个 free。这对 长事务 特别重要。

4. 事务完整生命周期:从 WAL 到 commit callback

4.1 阶段 1:记录 WAL 类型

WAL 由 XLogReader 读取,decode.c 的 DecodeXLogOp 是入口。它根据 XLogRecord->xl_info 分发到 xl_heap_insert / xl_heap_update / xl_heap_delete 等处理函数。

关键调用:

• ReorderBufferProcessXid (decode.c:120) — 几乎每个 WAL record 都会调,触发”lazy”的事务结构创建
• ReorderBufferAssignChild (decode.c:107) — 处理子事务声明

4.2 阶段 2:变更入队

ReorderBufferQueueChange(reorderbuffer.c:809-870) 是 最热路径,每个 DML 都会走这里:

void ReorderBufferQueueChange(ReorderBuffer *rb, TransactionId xid,
                              XLogRecPtr lsn, ReorderBufferChange *change)
{
    ReorderBufferTXN *txn;

    // (1) 查 or 创建 TXN
    txn = ReorderBufferTXNByXid(rb, xid, true, &is_new, lsn, true);
    change->txn = txn;

    // (2) 加入 changes dlist 末尾
    dlist_push_tail(&txn->changes, &change->node);

    // (3) 维护统计
    txn->nentries++;
    txn->nentries_mem++;
    txn->size += ReorderBufferChangeSize(change);
    rb->size += ReorderBufferChangeSize(change);

    // (4) 维护 pairheap
    txn->total_size = ...;
    if (is_new)
        pairingheap_add(txn_heap, &txn->txn_node);
    else
        pairingheap_update(txn_heap, &txn->txn_node);

    // (5) 内存压力检查
    ReorderBufferCheckMemoryLimit(rb);
}

5 步骤的因果链:

1. 查 or 创建 TXN:用 1-entry cache 避免 hash 查表
2. 加入 dlist 末尾:保证 LSN 顺序(与 WAL 顺序一致)
3. 更新 entries/size 计数
4. 更新 pairheap:新 xid 入堆,旧 xid 重新堆化
5. 检查内存:超限则触发 spill/stream

4.3 阶段 3:commit 触发 replay

ReorderBufferCommit(reorderbuffer.c:2874-2896):

void ReorderBufferCommit(ReorderBuffer *rb, TransactionId xid,
                          XLogRecPtr commit_lsn, XLogRecPtr end_lsn,
                          TimestampTz commit_time,
                          RepOriginId origin_id, XLogRecPtr origin_lsn)
{
    ReorderBufferTXN *txn;
    txn = ReorderBufferTXNByXid(rb, xid, false, NULL, InvalidXLogRecPtr, false);
    if (txn == NULL) return;                    // 未知 xid,无需 replay
    ReorderBufferReplay(txn, rb, xid, commit_lsn, end_lsn, commit_time, ...);
}

ReorderBufferReplay(reorderbuffer.c:2813-2871) 是核心 dispatcher:

void ReorderBufferReplay(ReorderBufferTXN *txn, ReorderBuffer *rb, ...)
{
    txn->final_lsn = commit_lsn;
    txn->end_lsn = end_lsn;
    ...

    // (1) 已 streaming 过的走流式 commit
    if (rbtxn_is_streamed(txn)) {
        ReorderBufferStreamCommit(rb, txn);
        return;
    }

    // (2) 没有 base snapshot 说明没改过表,跳过
    if (txn->base_snapshot == NULL) {
        Assert(txn->ninvalidations == 0);
        if (!rbtxn_is_prepared(txn))
            ReorderBufferCleanupTXN(rb, txn);
        return;
    }

    // (3) 走正常 replay
    snapshot_now = txn->base_snapshot;
    ReorderBufferProcessTXN(rb, txn, commit_lsn, snapshot_now,
                            FirstCommandId, false /* streaming */);
}

3 个分支的设计意图:

• 流式:已 streaming 的事务 commit 时只需发 STREAM_COMMIT
• 无 base snapshot:说明只做 DDL 或 no-op,直接清理
• 正常:进入 ReorderBufferProcessTXN 的复杂路径

4.4 阶段 4:ReorderBufferProcessTXN 全景

reorderbuffer.c:2210-2810(约 600 行)是 整个文件最复杂的函数。

关键设计:

• k-way heap 合并 (ReorderBufferIterTXNInit/Next/Finish):合并 toplevel + 多个 subxact 的 changes,按 LSN 顺序遍历
• CHECK_FOR_INTERRUPTS() 在每条变更前:防止长事务解码中无法响应取消
• SetupCheckXidLive:在 streaming/prepare 模式下检测 publisher 端并发 abort
• PG_TRY/PG_CATCH:保证 subtransaction 失败时能 rollback 不污染主事务

5. 内存压力管理:ReorderBufferCheckMemoryLimit

这是 reorder buffer 最重要的”自我保护”机制。reorderbuffer.c:3883-3962:

static void ReorderBufferCheckMemoryLimit(ReorderBuffer *rb)
{
    if (debug_logical_replication_streaming == DEBUG_LOGICAL_REP_STREAMING_BUFFERED &&
        rb->size < logical_decoding_work_mem * (Size) 1024)
        return;                                // 未超限,不做事

    while (rb->size >= logical_decoding_work_mem * (Size) 1024 ||
           debug_streaming_immediate) {
        // 路径 1: 优先 streaming(给 output plugin)
        if (ReorderBufferCanStartStreaming(rb) &&
            (txn = ReorderBufferLargestStreamableTopTXN(rb)) != NULL) {
            ReorderBufferStreamTXN(rb, txn);
        }
        // 路径 2: 否则 spill to disk
        else {
            txn = ReorderBufferLargestTXN(rb);
            ReorderBufferSerializeTXN(rb, txn);
        }
    }
}

5.1 决策流程图

5.2 Spill-to-Disk 详解:ReorderBufferSerializeTXN

reorderbuffer.c:3963-4057:

static void ReorderBufferSerializeTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
{
    // (1) 递归序列化所有 subxact
    dlist_foreach(subtxn_i, &txn->subtxns)
        ReorderBufferSerializeTXN(rb, subtxn);

    // (2) 按 LSN 所在 WAL segment 打开文件
    dlist_foreach_modify(change_i, &txn->changes) {
        if (fd == -1 || !XLByteInSeg(change->lsn, curOpenSegNo, wal_segment_size)) {
            // 切换 segment,关旧开新
            ReorderBufferSerializedPath(path, MyReplicationSlot, txn->xid, curOpenSegNo);
            fd = OpenTransientFile(path, O_CREAT|O_WRONLY|O_APPEND|PG_BINARY);
        }
        ReorderBufferSerializeChange(rb, txn, fd, change);
        dlist_delete(&change->node);
        ReorderBufferFreeChange(rb, change, false);
    }

    // (3) 更新统计
    rb->spillCount += 1;
    rb->spillBytes += size;
    rb->spillTxns += is_new_spill ? 1 : 0;

    txn->txn_flags |= RBTXN_IS_SERIALIZED;
}

Spill 文件路径(reorderbuffer.c:279-280):

#define PG_LOGICAL_DIR "pg_logical"
ReorderBufferSerializedPath(path, MyReplicationSlot, txn->xid, segno);
// → $PGDATA/pg_logical/slotname/xid-segno

关键设计:

• 递归序列化 subxact:toplevel 串行化时,所有 subxact 一起
• 按 WAL segment 切文件:便于 truncate 时回收
• O_APPEND 模式:保证同一 segment 多 transaction 都能写入
• 删除内存中的 change:释放 rb->size 计数

5.3 恢复:从 spill 文件回填

ReorderBufferRestoreChanges(reorderbuffer.c:4510-4652):

static Size ReorderBufferRestoreChanges(ReorderBuffer *rb, ReorderBufferTXN *txn,
                                        TXNEntryFile *file, XLogSegNo *segno)
{
    // (1) 遍历 spill 文件,逐 change 反序列化
    while ((change = ReorderBufferRestoreChange(...)) != NULL) {
        dlist_push_tail(&txn->changes, &change->node);
        ...
    }
    // (2) 删除已读完的文件
    ReorderBufferSerializedPath(path, slot, xid, segno);
    unlink(path);
}

触发恢复的时机:

• 事务 commit 时,如果已 spill,从磁盘回填
• 事务 abort 时,直接删除 spill 文件

6. Streaming 路径:ReorderBufferStreamTXN

reorderbuffer.c:4308-4403(约 100 行)是 streaming 的核心:

static void ReorderBufferStreamTXN(ReorderBuffer *rb, ReorderBufferTXN *txn)
{
    // (1) 第一次 stream:转移 subxact 的 base snapshot
    if (txn->snapshot_now == NULL) {
        dlist_foreach(subxact_i, &txn->subtxns)
            ReorderBufferTransferSnapToParent(txn, subtxn);
        if (txn->base_snapshot == NULL) return;   // 无变更,直接返回

        command_id = FirstCommandId;
        snapshot_now = ReorderBufferCopySnap(rb, txn->base_snapshot, txn, command_id);
    }
    // (2) 后续 stream:复用上次 snapshot
    else {
        command_id = txn->command_id;
        snapshot_now = ReorderBufferCopySnap(rb, txn->snapshot_now, txn, command_id);
        ReorderBufferFreeSnap(rb, txn->snapshot_now);
        txn->snapshot_now = NULL;
    }

    // (3) 记录统计数据
    txn_is_streamed = rbtxn_is_streamed(txn);
    stream_bytes = txn->total_size;

    // (4) 走 streaming 路径(不是正常 commit 路径)
    ReorderBufferProcessTXN(rb, txn, InvalidXLogRecPtr, snapshot_now,
                            command_id, true /* streaming */);

    // (5) 更新统计
    rb->streamCount += 1;
    rb->streamBytes += stream_bytes;
    rb->streamTxns += txn_is_streamed ? 0 : 1;
    UpdateDecodingStats(...);

    // (6) Assert: streaming 后 changes 全部输出
    Assert(dlist_is_empty(&txn->changes));
    Assert(txn->nentries == 0);
    Assert(txn->nentries_mem == 0);
}

6.1 Streaming vs Commit 的差异

6.2 Streaming 之后的变化

streaming 一次后,事务回到内存但 changes 列表清空(Assert(dlist_is_empty(&txn->changes)))。如果同一事务继续接收 WAL:

• 新的变更继续入队到 changes
• ReorderBufferCheckMemoryLimit 再次触发
• 走 ReorderBufferStreamTXN 的”已 stream”分支(txn->snapshot_now != NULL),复用之前的 snapshot

这就是 streaming 多次 的设计:output plugin 不需要为同一事务维护整个状态。

7. Subtransaction 处理:AssignChild / CommitChild

7.1 ReorderBufferAssignChild(reorderbuffer.c:1098-1163)

void ReorderBufferAssignChild(ReorderBuffer *rb, TransactionId xid,
                              TransactionId subxid, XLogRecPtr lsn)
{
    ReorderBufferTXN *txn = ReorderBufferTXNByXid(rb, xid, false, NULL, lsn, false);
    ReorderBufferTXN *subtxn = ReorderBufferTXNByXid(rb, subxid, true, &is_new, lsn, false);

    if (is_new) {
        subtxn->toptxn = txn;
        subtxn->toplevel_xid = xid;
    } else if (subtxn->toptxn == NULL) {
        subtxn->toptxn = txn;
    }

    // 如果 subtxn 当前在 toplevel 列表(因为之前没认主),移到 subtxns 列表
    if (subtxn->toplevel_xid != xid) {
        dlist_delete(&subtxn->node);
        ...
    }
}

关键设计:

• 先把所有 xid 当 toplevel(create_as_top=true)
• 看到 ASSIGN 时回填父子关系
• 已认主的 subtxn 从 toplevel_by_lsn 移到 toptxn->subtxns

7.2 ReorderBufferCommitChild(reorderbuffer.c:1218-1282)

void ReorderBufferCommitChild(ReorderBuffer *rb, TransactionId xid,
                              TransactionId subxid, XLogRecPtr commit_lsn,
                              XLogRecPtr end_lsn, ...)
{
    ReorderBufferTXN *subtxn = ReorderBufferTXNByXid(rb, subxid, false, ...);
    ReorderBufferTXN *toptxn = ReorderBufferTXNByXid(rb, xid, false, ...);

    // 1. 转移 base snapshot:子事务的变更要按父事务的快照解读
    if (subtxn->base_snapshot != NULL) {
        ReorderBufferTransferSnapToParent(toptxn, subtxn);
    }

    // 2. 转移 invalidation messages
    if (subtxn->ninvalidations > 0) {
        toptxn->invalidations = repalloc(toptxn->invalidations, ...);
        memcpy(toptxn->invalidations + toptxn->ninvalidations, ...);
        toptxn->ninvalidations += subtxn->ninvalidations;
    }

    // 3. 把 subxact 的 changes 转移到 toptxn
    dlist_splice(&toptxn->changes, &subtxn->changes);
    toptxn->nentries += subtxn->nentries;
    toptxn->nentries_mem += subtxn->nentries_mem;

    // 4. 删除 subtxn
    ReorderBufferCleanupTXN(rb, subtxn);
}

关键洞察:

• subxact commit 时,所有变更 合并到 toptxn(dlist_splice)
• 这样 toptxn 最终 commit 时,会按 LSN 顺序输出所有 subxact 的变更
• base snapshot 只能从子事务转移到父事务,不能反向(这是 snapshot 的语义决定)

8. Abort 处理:3 种路径

8.1 三种 abort 场景

8.2 ReorderBufferAbort 详解(reorderbuffer.c:3077-3122)

void ReorderBufferAbort(ReorderBuffer *rb, TransactionId xid, XLogRecPtr lsn,
                        TimestampTz abort_time)
{
    ReorderBufferTXN *txn = ReorderBufferTXNByXid(rb, xid, false, ...);
    if (txn == NULL) return;                   // 未知 xid

    txn->xact_time.abort_time = abort_time;

    // (1) 已 streaming 的事务:通知 output plugin
    if (rbtxn_is_streamed(txn)) {
        rb->stream_abort(rb, txn, lsn);
        if (txn->ninvalidations > 0)
            ReorderBufferImmediateInvalidation(rb, ...);  // 立即执行 invalidations
    }

    // (2) 清理内存 + 磁盘
    ReorderBufferCleanupTXN(rb, txn);
}

关键点:

• 显式 abort 一定要 执行 invalidation(因为 catalog 改动可能要恢复)
• 跟 ReorderBufferAbortOld 区别:old 不执行 invalidation(因为 server 已经 crash 过)

8.3 ReorderBufferAbortOld 详解(reorderbuffer.c:3123-3168)

void ReorderBufferAbortOld(ReorderBuffer *rb, TransactionId oldestRunningXid)
{
    // (1) 按 LSN 顺序遍历 toplevel_by_lsn
    while ((txn = ReorderBufferGetOldestTXN(rb)) != NULL) {
        if (TransactionIdFollows(txn->xid, oldestRunningXid))
            break;                              // 还有运行的,停止
        ReorderBufferCheckAndTruncateAbortedTXN(rb, txn);
    }
}

应用场景:

• server restart 后,decoding 启动时调一次
• oldestRunningXid 来自 pg_replication_slot_advance 推进后的 RunningXacts
• 凡是 first_lsn < 推进点 且 xid 已不在 running 中的,都认为已 abort

9. 与其他模块的依赖关系

9.1 模块依赖图

9.2 调用方对 reorderbuffer 的 4 类操作

9.3 完整调用栈:一条 INSERT 的旅程

10. 16 个 Callback 的全景

reorder buffer 用 16 个 callback 解耦”事务调度”和”具体数据输出”:

调用入口:

• begin / apply_change / commit:在 ReorderBufferProcessTXN 中,流式=false 时
• stream_*:在 ReorderBufferProcessTXN 中,流式=true 时
• prepare / commit_prepared / rollback_prepared:在 ReorderBufferPrepare / FinishPrepared 中

11. 关键设计洞察

11.1 为什么需要 1-entry cache?

reorderbuffer.c:651-665 的注释揭示:

// 99% 的连续操作都是同一个 xid 的变更(同事务内)
if (rb->by_txn_last_xid == xid) {
    if (rb->by_txn_last_txn != NULL) {
        if (is_new) *is_new = false;
        return rb->by_txn_last_txn;
    }
}

在一次事务的连续 INSERT/UPDATE 中,xid 保持不变。by_txn_last_xid 充当 micro-cache:

• 命中:直接返回,跳过 hash 查表
• 未命中且之前查过不存在(没创建):也跳过
• 未命中且未查过:走 hash 查表

11.2 为什么 LSN 顺序是 dlist 而不是排序结构?

dlist_push_tail(&txn->changes, &change->node);

变更按 WAL 顺序到达,直接追加到 dlist 末尾即可:

• 维护成本 O(1)
• 遍历时天然 LSN 顺序
• 唯一的反序场景是 streaming 多次 + 中间 spill(AssertTXNLsnOrder 校验)

11.3 pairheap 为什么用 total_size 而不是 size?

ReorderBufferTXNSizeCompare(...)
{
    // 父事务 + 所有子事务的 size
    return ... total_size ...
}

理由:驱逐时,选”驱逐它就能释放最多内存的事务”。父事务的 size 可能小,但加上 subxact 后很大。total_size 才能反映。

11.4 为什么 Tuple 用 Generation 而 Change 用 Slab?

11.5 为什么 streaming 时复用 snapshot?

if (txn->snapshot_now == NULL) {
    // 第一次 stream
    snapshot_now = ReorderBufferCopySnap(rb, txn->base_snapshot, ...);
} else {
    // 后续 stream,复用上次 snapshot
    snapshot_now = ReorderBufferCopySnap(rb, txn->snapshot_now, ...);
    ReorderBufferFreeSnap(rb, txn->snapshot_now);
}

每次 stream 完,output plugin 那边已经收到了当时的 snapshot 表示的视图。后续新变更可以基于上次 snapshot 继续解码,不必重新分配。

12. 关键源码速查表

12.1 文件级速查

12.2 reorderbuffer 与 decode.c 的调用关系

12.3 reorderbuffer 与 snapbuild.c 的调用关系

13. 实战场景分析

13.1 场景 1:tpcc-100wh 产生数百万 spill 文件

为什么?

// logical_decoding_work_mem 默认 128MB
// tpcc-100wh 单事务可能涉及数百个 warehouse 的新订单
// 一次 INSERT 可能涉及数十万行

reorder buffer 的应对:

1. ReorderBufferQueueChange 每次都检查 ReorderBufferCheckMemoryLimit
2. 128MB 触发 ReorderBufferSerializeTXN
3. 选最大的 TXN(可能就是刚入队的),spill 到 $PGDATA/pg_logical/slotname/xid-segno
4. 提交后再回填、清理

为什么会有 500w+ spill 文件?

• 100wh TPC-C 长时间运行,即使 streaming 模式,大事务也容易多次 spill
• 每次 spill 切 segment 时创建新文件
• 事务结束清理(删除),但短时间内堆积

13.2 场景 2:streaming 模式下的”反复 stream”

触发条件:

• sub 创建时设置 streaming = parallel 或 streaming = on
• output plugin 注册了 stream_start_cb 等

流程:

1. LA 第一次 stream:ReorderBufferStreamTXN → ProcessTXN(streaming=true)
2. 调用 stream_start / stream_change / stream_stop → output plugin
3. 事务回到内存但 changes 清空
4. 后续 WAL 继续进 changes 列表
5. 内存超限时再次 stream

关键观察:streaming 之后,事务的 changes 列表已空。如果再次 stream,ReorderBufferStreamTXN 中 txn->snapshot_now != NULL 走 else 分支,复用上次的 snapshot。

13.3 场景 3:subxact commit 合并到父

BEGIN;
  INSERT INTO t1 VALUES (1);   -- subxact 1
  SAVEPOINT s1;
  INSERT INTO t1 VALUES (2);
  RELEASE SAVEPOINT s1;
COMMIT;

reorder buffer 视角:

1. 看到 XLOG_XACT_ASSIGNMENT:ReorderBufferAssignChild 标记父子
2. 看到 xl_heap_insert xid=subxact1:ReorderBufferQueueChange 创建 subtxn,变更入 subtxn->changes
3. 看到 XLOG_XACT_COMMIT xid=subxact1:ReorderBufferCommitChild:
   • 转移 base snapshot(如果有)
   • dlist_splice 把 subtxn->changes 合并到 toptxn->changes
   • 删除 subtxn
4. 看到 XLOG_XACT_COMMIT xid=toplevel:ReorderBufferCommit → ReorderBufferReplay

最终 output plugin 看到的:两次 INSERT 按 LSN 顺序输出(可能是 subxact 的 1 在前,toplevel 的 2 在后)。

14. 与其他模块的接口契约

14.1 reorder buffer 期望的”环境”

14.2 reorder buffer 保证的不变量

15. 总结

15.1 reorder buffer 的本质

reorder buffer = 内存 + 磁盘的”事务仓库”,负责把无序的 WAL 重组为有序的、按事务维度的变更流。

它实现了 4 个核心能力:

1. 事务重组:从 WAL 流中按 xid 分组,subxact 合并到父事务
2. 变更缓存:在内存中保留所有 DML,直到 commit
3. 背压管理:logical_decoding_work_mem 触发 spill-to-disk + streaming
4. 快照集成:与 snapbuild 协作,保证变更解码的一致性快照

15.2 关键洞察清单

1. 1-entry cache 是热路径的 micro-optimization,99% 命中
2. 4 套索引 各司其职,by_txn 主查,3 个 dlist 用于遍历,pairheap 用于驱逐
3. 3 套 MemoryContext 区分等大/变长/短生命周期
4. ReorderBufferProcessTXN 是 600 行的”巨无霸”,所有 replay 都走这里
5. streaming 复用 snapshot 是 streaming 多次的基石
6. subxact commit 合并到父 让 toplevel 的 commit 输出包含所有 subxact
7. spill 文件按 WAL segment 切,便于 truncate 时回收

15.3 推荐关联阅读

• PostgreSQL 逻辑复制 streaming 机制 — streaming 业务层
• PostgreSQL 逻辑复制 spill 文件分析 — spill 详细分析
• PostgreSQL 逻辑复制 MVCC 与事务机制 — 事务一致性与 reorder buffer 的关系
• PostgreSQL 逻辑复制 reorder buffer 与快照 — 快照管理
• PostgreSQL 逻辑复制并行解码 — parallel apply 的源数据

参考

• src/backend/replication/logical/reorderbuffer.c (5615 行) — 本篇分析对象
• src/include/replication/reorderbuffer.h (783 行) — 公共头文件
• src/backend/replication/logical/decode.c (1316 行) — 主要调用方(WAL 解析)
• src/backend/replication/logical/snapbuild.c (2076 行) — 协作方(snapshot 管理)
• src/backend/replication/logical/logical.c (2198 行) — 装配层(设置 callback)
• src/backend/replication/slotsync.c — slot 同步(部分使用 ReorderBuffer)
• src/backend/storage/file/fd.c — OpenTransientFile 实现
• src/backend/utils/mmgr/slab.c — SlabContext 实现
• src/backend/utils/mmgr/generation.c — GenerationContext 实现
• src/backend/lib/pairingheap.c — pairheap 实现
• PostgreSQL 18 源码:git log --follow src/backend/replication/logical/reorderbuffer.c 可查看提交历史