前言

今天，白鳝老师问了一个问题，关于LogicalRewriteCheckpointSync等待事件的

乍一看有点陌生，遂翻了一下官网，但是遗憾的是，官网的介绍不知所云，对于新手以及我这样的半熟油条，实在忍不住吐槽，着实费力！

中文手册翻译，依旧没看懂

搞不定的时候，只能看源码了

分析

在说明里面，可以看到和checkpoint有关，checkpoint我们知道，最核心的函数是CheckPointGuts()

/*
 * Flush all data in shared memory to disk, and fsync
 *
 * This is the common code shared between regular checkpoints and
 * recovery restartpoints.
 */
static void
CheckPointGuts(XLogRecPtr checkPointRedo, int flags)
{
    CheckPointRelationMap();
    CheckPointReplicationSlots();
    CheckPointSnapBuild();
    CheckPointLogicalRewriteHeap();
    CheckPointReplicationOrigin();

    /* Write out all dirty data in SLRUs and the main buffer pool */
    TRACE_POSTGRESQL_BUFFER_CHECKPOINT_START(flags);
    CheckpointStats.ckpt_write_t = GetCurrentTimestamp();
    CheckPointCLOG();
    CheckPointCommitTs();
    CheckPointSUBTRANS();
    CheckPointMultiXact();
    CheckPointPredicate();
    CheckPointBuffers(flags);

    /* Perform all queued up fsyncs */
    TRACE_POSTGRESQL_BUFFER_CHECKPOINT_SYNC_START();
    CheckpointStats.ckpt_sync_t = GetCurrentTimestamp();
    ProcessSyncRequests();
    CheckpointStats.ckpt_sync_end_t = GetCurrentTimestamp();
    TRACE_POSTGRESQL_BUFFER_CHECKPOINT_DONE();

    /* We deliberately delay 2PC checkpointing as long as possible */
    CheckPointTwoPhase(checkPointRedo);
}

CheckPointGuts()里面调用了CheckPointLogicalRewriteHeap()这个函数

/* ---
 * Perform a checkpoint for logical rewrite mappings
 *
 * This serves two tasks:
 * 1) Remove all mappings not needed anymore based on the logical restart LSN
 * 2) Flush all remaining mappings to disk, so that replay after a checkpoint
 *      only has to deal with the parts of a mapping that have been written out
 *      after the checkpoint started.
 * ---
 */
void
CheckPointLogicalRewriteHeap(void)
{
    XLogRecPtr    cutoff;
    XLogRecPtr    redo;
    DIR           *mappings_dir;
    struct dirent *mapping_de;
    char        path[MAXPGPATH + 20];

    /*
     * We start of with a minimum of the last redo pointer. No new decoding
     * slot will start before that, so that's a safe upper bound for removal.
     */
    redo = GetRedoRecPtr();

    /* now check for the restart ptrs from existing slots */
    cutoff = ReplicationSlotsComputeLogicalRestartLSN();

因为逻辑复制相关的脏数据，会写入到pg_logical目录下面的mappings和snapshots子目录下

[postgres@xiongcc ~]$ ll pgdata/pg_logical/
total 96
drwx------ 2 postgres postgres  4096 Feb  8 18:28 mappings
-rw------- 1 postgres postgres     8 Feb  8 18:28 replorigin_checkpoint
drwx------ 2 postgres postgres 90112 Feb  8 18:28 snapshots

所以，这个等待事件也就呼之欲出了，等待checkpoint将逻辑复制相关的脏数据刷到存储上

过了一会，白老师又问了我一个问题，出了DDL之外，DML会修改吗？

这不又触及到知识盲区了，在pg_logical目录，我们知道是用于配置逻辑复制时，存储逻辑解码的数据状态

[postgres@xiongcc ~]$ ll pgdata/pg_logical/
total 96
drwx------ 2 postgres postgres  4096 Feb  8 18:28 mappings
-rw------- 1 postgres postgres     8 Feb  8 18:28 replorigin_checkpoint
drwx------ 2 postgres postgres 90112 Feb  8 18:28 snapshots

最开始里面什么都没有

[postgres@xiongcc ~]$ ll pgdata/pg_logical/snapshots/
total 0
[postgres@xiongcc ~]$ ll pgdata/pg_logical/mappings/
total 0

创建一个逻辑复制，发布端

postgres=# create table test(id int,info text);
CREATE TABLE
postgres=# create publication pub1 for table test;
CREATE PUBLICATION

订阅端

postgres=# create table test(id int,info text);
CREATE TABLE
postgres=# create subscription test_sub connection 'host=localhost port=5432' publication pub1;
NOTICE:  created replication slot "test_sub" on publisher
CREATE SUBSCRIPTION

发布端查看，OK，创建没什么问题

postgres=# select slot_name,plugin,slot_type,active,active_pid from pg_replication_slots ;
 slot_name |  plugin  | slot_type | active | active_pid 
-----------+----------+-----------+--------+------------
 test_sub  | pgoutput | logical   | t      |      11603
(1 row)

这里面有一个注意点，在刚刚创建完之后，有些人可能会看到有两个复制槽

postgres=# truncate table test;
TRUNCATE TABLE
postgres=# insert into test select n,'test' from generate_series(1,10000000) as n;
INSERT 0 10000000

订阅端重新订阅一下

postgres=# drop subscription test_sub ;
NOTICE:  dropped replication slot "test_sub" on publisher
DROP SUBSCRIPTION
postgres=# truncate table test;
TRUNCATE TABLE
postgres=# create subscription test_sub connection 'host=localhost port=5432' publication pub1;
NOTICE:  created replication slot "test_sub" on publisher
CREATE SUBSCRIPTION

再次查看复制槽，便会看到有两个复制槽，虽然订阅端仅打印了一条 NOTICE: created replication slot "test_sub" on publisher，另外一个复制槽名为pg_33777_sync_33758_7038607583722269381

-[ RECORD 1 ]-------+----------------------------------------
slot_name           | test_sub
plugin              | pgoutput
slot_type           | logical
datoid              | 13890
database            | postgres
temporary           | f
active              | t
active_pid          | 12382
xmin                | 
catalog_xmin        | 1212624
restart_lsn         | 7/DD783D70
confirmed_flush_lsn | 7/DD783DA8
wal_status          | reserved
safe_wal_size       | 
two_phase           | f
-[ RECORD 2 ]-------+----------------------------------------
slot_name           | pg_33777_sync_33758_7038607583722269381
plugin              | pgoutput
slot_type           | logical
datoid              | 13890
database            | postgres
temporary           | f
active              | f
active_pid          | 
xmin                | 
catalog_xmin        | 1212624
restart_lsn         | 7/DD783D70
confirmed_flush_lsn | 7/DD783DA8
wal_status          | reserved
safe_wal_size       | 
two_phase           | f

这个其实是14版本做的增强，如果原始表的记录数特别多, 初始数据同步时间比较长, 在这个过程中可能产生了很多新事务, 会带来一系列问题：

1.We have to perform the entire copy operation again if there is any error (network breakdown, error in the database operation, etc.) while we synchronize the WAL position between tablesync worker and apply worker; this will be onerous especially for large copies,2.Using a single transaction in the synchronization-phase (where we can receive WAL from multiple transactions) will have the risk of exceeding the CID limit,3.The slot will hold the WAL till the entire sync is complete because we never commit till the end.

每个 Tablesync Worker 都有一个关联的复制槽。以前是临时的插槽，仅在每个 Tablesync Worker 的生命周期内存在。如果 Tablesync Worker 意外崩溃，则槽位丢失，当替代 Tablesync Worker 启动时，它会从新创建的临时槽位开始，然后再次重复所有内容，就像第一次看到一样。Tablesync Worker已改成使用永久的复制槽而不是临时复制槽。复制源信息会保存，用于跟踪已经复制的数据，因此通过使用永久插槽，这意味着在崩溃/重新启动后，可以从最近记录的检查点再次继续复制。

复制槽命名格式如下：pg<subscription OID>sync<table relid><system id>

回到目录结构pg_logical，那么mappings和snapshots这两个子目录有什么用呢？

snapshots

snapshots顾名思义，快照，逻辑复制相关的脏数据会写到pg_logical/snapshots，发布端插入一些数据

postgres=# insert into test select n,'test' from generate_series(1,1000000) as n;
INSERT 0 1000000
postgres=# insert into test select n,'test' from generate_series(1,1000000) as n;
INSERT 0 1000000

[postgres@xiongcc ~]$ ll pgdata/pg_logical/snapshots/
total 16
-rw------- 1 postgres postgres 132 Feb  8 16:35 7-E54791E0.snap
-rw------- 1 postgres postgres 128 Feb  8 16:37 7-E5479218.snap
-rw------- 1 postgres postgres 128 Feb  8 16:37 7-E547E478.snap
-rw------- 1 postgres postgres 128 Feb  8 16:37 7-ECEE91F8.snap

然后做个checkpoint，在checkpoint期间会回收相关数据

[postgres@xiongcc ~]$ ll pgdata/pg_logical/snapshots/
total 8
-rw------- 1 postgres postgres 128 Feb  8 16:38 7-F868C058.snap
-rw------- 1 postgres postgres 128 Feb  8 16:38 7-F86FB138.snap

具体每个字节是什么意思还要找时间看看代码

[postgres@xiongcc ~]$ hexdump -C pgdata/pg_logical/snapshots/7-F868C058.snap | more
00000000  01 e0 a1 51 5d 1b 3c e2  04 00 00 00 80 00 00 00  |...Q].<.........|
00000010  02 00 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
00000020  db 80 12 00 d6 80 12 00  e0 3d 78 dd 07 00 00 00  |.........=x.....|
00000030  e0 3d 78 dd 07 00 00 00  00 00 00 00 00 00 00 00  |.=x.............|
00000040  00 00 00 00 00 00 00 00  90 bf 68 f8 07 00 00 00  |..........h.....|
00000050  00 00 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
00000060  00 00 00 00 00 00 00 00  80 00 00 00 00 00 00 00  |................|
00000070  00 00 00 00 00 00 00 00  00 00 00 00 00 00 00 00  |................|
00000080

checkpoint回收的代码可以参照CheckPointSnapBuild

/*
 * Remove all serialized snapshots that are not required anymore because no
 * slot can need them. This doesn't actually have to run during a checkpoint,
 * but it's a convenient point to schedule this.
 *
 * NB: We run this during checkpoints even if logical decoding is disabled so
 * we cleanup old slots at some point after it got disabled.
 */
void
CheckPointSnapBuild(void)
{
    XLogRecPtr    cutoff;
    XLogRecPtr    redo;
    DIR           *snap_dir;
    struct dirent *snap_de;
    char        path[MAXPGPATH + 21];

mappings

那么mappings是什么呢，在插入数据解码的过程中，该子目录下面一直是空的，那么什么时候会有数据呢？

[postgres@xiongcc ~]$ ll pgdata/pg_logical/mappings/
total 0

翻翻代码看看

/*
 * Apply any existing logical remapping files if there are any targeted at our
 * transaction for relid.
 */
static void
UpdateLogicalMappings(HTAB *tuplecid_data, Oid relid, Snapshot snapshot)
{
    DIR           *mapping_dir;
    struct dirent *mapping_de;
    List       *files = NIL;
    ListCell   *file;
    Oid            dboid = IsSharedRelation(relid) ? InvalidOid : MyDatabaseId;

    mapping_dir = AllocateDir("pg_logical/mappings");
    ...

/*
 * Apply a single mapping file to tuplecid_data.
 *
 * The mapping file has to have been verified to be a) committed b) for our
 * transaction c) applied in LSN order.
 */
static void
ApplyLogicalMappingFile(HTAB *tuplecid_data, Oid relid, const char *fname)
{
    char        path[MAXPGPATH];
    int            fd;
    int            readBytes;
    LogicalRewriteMappingData map;

    sprintf(path, "pg_logical/mappings/%s", fname);
    fd = OpenTransientFile(path, O_RDONLY | PG_BINARY);
    if (fd < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not open file \"%s\": %m", path)));

    while (true)
    {
        ReorderBufferTupleCidKey key;
        ReorderBufferTupleCidEnt *ent;
        ReorderBufferTupleCidEnt *new_ent;

    /*
     * failed to find a mapping, check whether the table was rewritten and
     * apply mapping if so, but only do that once - there can be no new
     * mappings while we are in here since we have to hold a lock on the
     * relation.
     */
    if (ent == NULL && !updated_mapping)
    {
        UpdateLogicalMappings(tuplecid_data, htup->t_tableOid, snapshot);
        /* now check but don't update for a mapping again */
        updated_mapping = true;
        goto restart;
    }

可以看到，和逻辑映射关系以及table rewritten有关，那么给表做一个DDL试试

postgres=# alter table test add column t_time timestamp;
ALTER TABLE
postgres=# vacuum test;
VACUUM
postgres=# vacuum full test;
VACUUM

但目录下还是空的

[postgres@xiongcc mappings]$ ll 
total 0
[postgres@xiongcc mappings]$ ll 
total 0
[postgres@xiongcc mappings]$ ll 
total 0

再看一下代码，在rewriteheap.c里面也涉及到了mappings，此处提到了XLOG_HEAP2_REWRITE这个类型的wal record

/*
 * Replay XLOG_HEAP2_REWRITE records
 */
void
heap_xlog_logical_rewrite(XLogReaderState *r)
{
    char        path[MAXPGPATH];
    int            fd;
    xl_heap_rewrite_mapping *xlrec;
    uint32        len;
    char       *data;

    xlrec = (xl_heap_rewrite_mapping *) XLogRecGetData(r);

    snprintf(path, MAXPGPATH,
             "pg_logical/mappings/" LOGICAL_REWRITE_FORMAT,
             xlrec->mapped_db, xlrec->mapped_rel,
             LSN_FORMAT_ARGS(xlrec->start_lsn),
             xlrec->mapped_xid, XLogRecGetXid(r));

wal record的类型在src\include\access\heapam_xlog.h，

•heap2（RM_HEAP2_ID）是数据库对heap的修改产生的wal record（去除heap）•heap（RM_HEAP_ID）是数据库对heap进行insert、delete、update等常规操作后所产生的日志类型

其实根据名字也能看出大概的用途

/*
 * WAL record definitions for heapam.c's WAL operations
 *
 * XLOG allows to store some information in high 4 bits of log
 * record xl_info field.  We use 3 for opcode and one for init bit.
 */
#define XLOG_HEAP_INSERT          0x00
#define XLOG_HEAP_DELETE          0x10
#define XLOG_HEAP_UPDATE          0x20
#define XLOG_HEAP_TRUNCATE        0x30
#define XLOG_HEAP_HOT_UPDATE      0x40
#define XLOG_HEAP_CONFIRM         0x50
#define XLOG_HEAP_LOCK            0x60
#define XLOG_HEAP_INPLACE         0x70

#define XLOG_HEAP_OPMASK          0x70
/*
 * When we insert 1st item on new page in INSERT, UPDATE, HOT_UPDATE,
 * or MULTI_INSERT, we can (and we do) restore entire page in redo
 */
#define XLOG_HEAP_INIT_PAGE        0x80
/*
 * We ran out of opcodes, so heapam.c now has a second RmgrId.  These opcodes
 * are associated with RM_HEAP2_ID, but are not logically different from
 * the ones above associated with RM_HEAP_ID.  XLOG_HEAP_OPMASK applies to
 * these, too.
 */
#define XLOG_HEAP2_REWRITE        0x00
#define XLOG_HEAP2_PRUNE          0x10
#define XLOG_HEAP2_VACUUM         0x20
#define XLOG_HEAP2_FREEZE_PAGE    0x30
#define XLOG_HEAP2_VISIBLE        0x40
#define XLOG_HEAP2_MULTI_INSERT   0x50
#define XLOG_HEAP2_LOCK_UPDATED   0x60
#define XLOG_HEAP2_NEW_CID        0x70

回到XLOG_HEAP2_REWRITE，当开启logical日志模式时，在系统表和一些特殊用户表在执行vacuum full的过程中，会产生这个wal记录类型。这个wal日志类型主要是为了恢复pg_logical/mapping下的文件，mapping目录下的文件用于在系统表发生vacuum full后来判断系统表元组的可见性。

/*
 * Replay XLOG_HEAP2_REWRITE records
 */
void
heap_xlog_logical_rewrite(XLogReaderState *r)
{
    char        path[MAXPGPATH];
    int            fd;
    xl_heap_rewrite_mapping *xlrec;
    uint32        len;
    char       *data;

    xlrec = (xl_heap_rewrite_mapping *) XLogRecGetData(r);

    snprintf(path, MAXPGPATH,
             "pg_logical/mappings/" LOGICAL_REWRITE_FORMAT,
             xlrec->mapped_db, xlrec->mapped_rel,
             LSN_FORMAT_ARGS(xlrec->start_lsn),
             xlrec->mapped_xid, XLogRecGetXid(r));

    fd = OpenTransientFile(path,
                           O_CREAT | O_WRONLY | PG_BINARY);
    if (fd < 0)
        ereport(ERROR,
                (errcode_for_file_access(),
                 errmsg("could not create file \"%s\": %m", path)));

所以这里也可以看出，假如碰到类似的错，有迹可循，参照 https://postgrespro.com/list/thread-id/2399779

 ERROR:  53000: exceeded maxAllocatedDescs (492) while trying to open
 file "pg_logical/mappings/map-4000-4eb-1_60DE1E08-5376b5-537c6b"
 LOCATION:  OpenTransientFile, fd.c:2161
 LOG:  00000: starting logical decoding for slot "s"
 DETAIL:  streaming transactions committing after 1/6097DD48, reading
 WAL from 1/60275848
 LOCATION:  CreateDecodingContext, logical.c:414
 LOG:  00000: logical decoding found consistent point at 1/60275848
 DETAIL:  Logical decoding will begin using saved snapshot.

并且此处也提到了，运行了vacuum full pg_class

I suppose there are reasons why it's done this way, and admittedly the test that happens to trigger this is a bit extreme (essentially running pgbench concurrently with 'vacuum full pg_class' in a loop)

相关结构体

/* logical rewrite xlog record header */
typedef struct xl_heap_rewrite_mapping
{
    TransactionId mapped_xid;    /* xid that might need to see the row */
    Oid            mapped_db;        /* DbOid or InvalidOid for shared rels */
    Oid            mapped_rel;        /* Oid of the mapped relation */
    off_t        offset;            /* How far have we written so far */
    uint32        num_mappings;    /* Number of in-memory mappings */
    XLogRecPtr    start_lsn;        /* Insert LSN at begin of rewrite */
} xl_heap_rewrite_mapping;

pg_logical/mappings目录下的文件格式如下

/*
 * On-Disk data format for an individual logical rewrite mapping.
 */
typedef struct LogicalRewriteMappingData
{
    RelFileNode old_node;
    RelFileNode new_node;
    ItemPointerData old_tid;
    ItemPointerData new_tid;
} LogicalRewriteMappingData;

那么做一个vacuum full试试

postgres=# vacuum full pg_class;
VACUUM

果然，出现了mapping文件

[postgres@xiongcc mappings]$ ll
total 4
-rw------- 1 postgres postgres 36 Feb  8 18:21 map-3642-4eb-8_25DA5598-1280ed-1280ed

同理，checkpoint的时候去删除mapping文件

/* ---
 * Perform a checkpoint for logical rewrite mappings
 *
 * This serves two tasks:
 * 1) Remove all mappings not needed anymore based on the logical restart LSN
 * 2) Flush all remaining mappings to disk, so that replay after a checkpoint
 *      only has to deal with the parts of a mapping that have been written out
 *      after the checkpoint started.
 * ---
 */
void
CheckPointLogicalRewriteHeap(void)
{
    XLogRecPtr    cutoff;
    XLogRecPtr    redo;
    DIR           *mappings_dir;
    struct dirent *mapping_de;
    char        path[MAXPGPATH + 20];

    /*
     * We start of with a minimum of the last redo pointer. No new decoding
     * slot will start before that, so that's a safe upper bound for removal.
     */
    redo = GetRedoRecPtr();

postgres=# checkpoint ;
CHECKPOINT

mappings文件被删除

[postgres@xiongcc mappings]$ ll
total 0

小结

逻辑解码依赖于WAL日志，进行decoding，变更都记录在WAL日志里面，翻译出来必须要知道当时的表结构是啥样的，即需要元数据信息才能将二进制信息转化为逻辑操作信息。因为也需要类似的快照保留机制，这个就是pg_replication_slots里面的catalog_xmin字段的作用，vacuum在清理系统表的时候会加以判断，保留还需要的那些系统表元组，这个mappings也是类似的道理。

不过话说回来，PostgreSQL里面的等待事件光看官网，有些地方着实令人费解，需要实际场景的打磨。罪名+1，白鳝老师在公众号上有过一篇整理，《针对PG的等待事件不妨玩点大的》各位感兴趣的童鞋可以看一下，不过明天会放出最新的等待事件，让我们拭目以待。