逻辑复制不为熟知的点
前言
长事务，俗称DB killer，在PostgreSQL中，危害更加明显。
我们比较熟悉的几点是，假如某个长事务持有了xmin，会导致表膨胀、年龄回收失败。还有一点可能对于不少人比较陌生的，是在Logical decoding的场景下。
长事务阻止逻辑解码
首先第一点是，长事务会阻止logical decoding，假如现在有这样一个操作流：
T1时刻：开启一个长事务，做了相关更改，但不提交
👇
T2时刻：开启逻辑解码
👇
T3时刻：T1事务提交
那么是否可以将T1 ~ T3这段时间的操作decoding出来？
实际上这是一个伪命题，逻辑解码这个动作会被阻塞。见如下样例：
postgres=# create publication pub1 for table test;CREATE PUBLICATIONpostgres=# begin;BEGINpostgres=*# insert into t1 values(1);INSERT 0 1---事务不提交，并且持有了xmin
新开一个会话，进行订阅
[postgres@xiongcc ~]$ psql -p 5439psql (13.2)Type "help" for help.postgres=# create subscription sub1 connection 'host=localhost port=5432' publication pub1;---此处卡住
第一个会话提交或者回滚之后，订阅才能正常创建。
同理，当有2pc没有提交或回滚的话，也是会阻塞
postgres=# begin;BEGINpostgres=*# insert into t1 values(1);INSERT 0 1postgres=*# prepare transaction 't1';PREPARE TRANSACTIONpostgres=# \q[postgres@xiongcc ~]$ psqlpsql (13.2)Type "help" for help.postgres=# select * from pg_prepared_xacts ;transaction | gid | prepared | owner | database-------------+-----+-------------------------------+----------+----------326114 | t1 | 2021-09-21 17:35:19.564798+08 | postgres | postgres(1 row)postgres=# create publication pub1 for table test;CREATE PUBLICATION
订阅端
[postgres@xiongcc ~]$ psql -p 5439psql (13.2)Type "help" for help.postgres=# create subscription sub1 connection 'host=localhost port=5432' publication pub1;---此处卡住
但是这二者还是有一点细微的差别，对于普通的长事务，假如没有持有xmin，即是普通的查询，是不会阻塞的
postgres=# create publication pub1 for table test;CREATE PUBLICATIONpostgres=# begin;BEGINpostgres=*# select * from t1 limit 1;info------haha(1 row)
订阅端可以创建成功
postgres=# create subscription sub1 connection 'host=localhost port=5432' publication pub1;NOTICE: created replication slot "sub1" on publisherCREATE SUBSCRIPTION
而2pc不管你是查询还是DML，都会阻塞
postgres=# create publication pub1 for table test;CREATE PUBLICATIONpostgres=# begin;BEGINpostgres=*# select * from t1 limit 1;info------haha(1 row)postgres=*# prepare transaction 't1';PREPARE TRANSACTION
订阅端会卡住
postgres=# create subscription sub1 connection 'host=localhost port=5432' publication pub1;---此处卡住
所以2pc的危害十分大。不仅会导致表膨胀、年龄回收失败，还会持有锁，锁的具体强度取决于你做了些什么动作。再到此例中，还会阻塞逻辑复制。
postgres=# select locktype,granted from pg_locks where relation='t1'::regclass;locktype | granted----------+---------relation | t(1 row)
那么可能有人会问了，这个是逻辑复制，逻辑解码会如何呢？稍等，现象更加"奇妙"
[postgres@xiongcc ~]$ psql -p 5432psql (13.2)Type "help" for help.postgres=# begin;BEGINpostgres=*# insert into t1 values(2);INSERT 0 1
使用pg_recvlogical创建一个复制槽
[postgres@xiongcc ~]$ pg_recvlogical --create-slot -S myslot -P wal2json -U postgres -d postgres -p 5432---此处卡住
但在数据库里却已经可以看到这个复制槽了，不过处于等待中，并且在pg_stat_replication里面的state状态显示是startup，意味着walsender还在启动中
postgres=# select * from pg_replication_slots ;-[ RECORD 1 ]-------+-----------slot_name | myslotplugin | wal2jsonslot_type | logicaldatoid | 13578database | postgrestemporary | factive | tactive_pid | 23469xmin |catalog_xmin | 326131restart_lsn | 0/9801E510confirmed_flush_lsn |wal_status | reservedsafe_wal_size |
正在等待事务的结束，可以看到被前面的长事务阻塞
postgres=# SELECT blocked_locks.pid AS blocked_pid,blocked_activity.usename AS blocked_user,blocking_locks.pid AS blocking_pid,blocking_activity.usename AS blocking_user,blocked_activity.query AS blocked_statement,blocking_activity.query AS current_statement_in_blocking_processFROM pg_catalog.pg_locks blocked_locksJOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pidJOIN pg_catalog.pg_locks blocking_locksON blocking_locks.locktype = blocked_locks.locktypeAND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.databaseAND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relationAND blocking_locks.page IS NOT DISTINCT FROM blocked_locks.pageAND blocking_locks.tuple IS NOT DISTINCT FROM blocked_locks.tupleAND blocking_locks.virtualxid IS NOT DISTINCT FROM blocked_locks.virtualxidAND blocking_locks.transactionid IS NOT DISTINCT FROM blocked_locks.transactionidAND blocking_locks.classid IS NOT DISTINCT FROM blocked_locks.classidAND blocking_locks.objid IS NOT DISTINCT FROM blocked_locks.objidAND blocking_locks.objsubid IS NOT DISTINCT FROM blocked_locks.objsubidAND blocking_locks.pid != blocked_locks.pidJOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pidWHERE NOT blocked_locks.granted;-[ RECORD 1 ]-------------------------+--------------------------------------------------------blocked_pid | 23469blocked_user | postgresblocking_pid | 23454blocking_user | postgresblocked_statement | SELECT pg_catalog.set_config('search_path', '', false);current_statement_in_blocking_process | insert into t1 values(1);[postgres@xiongcc ~]$ ps -ef | grep 23469postgres 23469 21197 0 20:03 ? 00:00:00 postgres: walsender postgres [local] idle waitingpostgres 23525 23413 0 20:07 pts/9 00:00:00 grep --color=auto 23469
假如这个时候以一个"不健康"的复制槽去消费，还好会提示复制槽不存在
[postgres@xiongcc ~]$ pg_recvlogical -d postgres --slot myslot --start -f -pg_recvlogical: error: could not send replication command "START_REPLICATION SLOT "myslot" LOGICAL 0/0": ERROR: replication slot "myslot" is active for PID 21012pg_recvlogical: disconnected; waiting 5 seconds to try again
并且更为尴尬的是，CTRL + C的信号它是不会鸟你的
[postgres@xiongcc ~]$ pg_recvlogical --create-slot -S myslot -P wal2json -U postgres -d postgres -p 5432^C^C^C^C^C^C
并且手动kill了会留一下一个坚挺的复制槽
[postgres@xiongcc ~]$ ps -ef | grep pg_recvpostgres 23468 23391 0 20:03 pts/8 00:00:00 pg_recvlogical --create-slot -S myslot -P wal2json -U postgres -d postgres -p 5432postgres 23693 23413 0 20:16 pts/9 00:00:00 grep --color=auto pg_recv[postgres@xiongcc ~]$ kill -9 23468[postgres@xiongcc ~]$ ps -ef | grep pg_recvpostgres 23697 23413 0 20:16 pts/9 00:00:00 grep --color=auto pg_recv原窗口pg_recvlogical提示被Kill[postgres@xiongcc ~]$ pg_recvlogical --create-slot -S myslot -P wal2json -U postgres -d postgres -p 5432^C^C^C^C^C^CKilled
坚挺的复制槽，锁等待也还存在
[postgres@xiongcc ~]$ psql -p 5432psql (13.2)Type "help" for help.postgres=# select * from pg_replication_slots ;-[ RECORD 1 ]-------+-----------slot_name | myslotplugin | wal2jsonslot_type | logicaldatoid | 13578database | postgrestemporary | factive | tactive_pid | 23469xmin |catalog_xmin | 326131restart_lsn | 0/9801E510confirmed_flush_lsn |wal_status | reservedsafe_wal_size |postgres=# SELECT blocked_locks.pid AS blocked_pid,blocked_activity.usename AS blocked_user,blocking_locks.pid AS blocking_pid,blocking_activity.usename AS blocking_user,blocked_activity.query AS blocked_statement,blocking_activity.query AS current_statement_in_blocking_processFROM pg_catalog.pg_locks blocked_locksJOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pidJOIN pg_catalog.pg_locks blocking_locksON blocking_locks.locktype = blocked_locks.locktypeAND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.databaseAND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relationAND blocking_locks.page IS NOT DISTINCT FROM blocked_locks.pageAND blocking_locks.tuple IS NOT DISTINCT FROM blocked_locks.tupleAND blocking_locks.virtualxid IS NOT DISTINCT FROM blocked_locks.virtualxidAND blocking_locks.transactionid IS NOT DISTINCT FROM blocked_locks.transactionidAND blocking_locks.classid IS NOT DISTINCT FROM blocked_locks.classidAND blocking_locks.objid IS NOT DISTINCT FROM blocked_locks.objidAND blocking_locks.objsubid IS NOT DISTINCT FROM blocked_locks.objsubidAND blocking_locks.pid != blocked_locks.pidJOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pidWHERE NOT blocked_locks.granted;-[ RECORD 1 ]-------------------------+--------------------------------------------------------blocked_pid | 23469blocked_user | postgresblocking_pid | 23454blocking_user | postgresblocked_statement | SELECT pg_catalog.set_config('search_path', '', false);current_statement_in_blocking_process | insert into t1 values(1);
即使将事务提交或回滚，这个坚挺的复制槽还遗留着，同时也无效了
postgres=# begin;BEGINpostgres=*# insert into t1 values(1);INSERT 0 1postgres=*# rollback ;ROLLBACK
postgres=# select * from pg_replication_slots ;-[ RECORD 1 ]-------+-----------slot_name | myslotplugin | wal2jsonslot_type | logicaldatoid | 13578database | postgrestemporary | factive | tactive_pid | 23469xmin |catalog_xmin | 326131restart_lsn | 0/9801E510confirmed_flush_lsn |wal_status | reservedsafe_wal_size |postgres=# SELECT blocked_locks.pid AS blocked_pid,blocked_activity.usename AS blocked_user,blocking_locks.pid AS blocking_pid,blocking_activity.usename AS blocking_user,blocked_activity.query AS blocked_statement,blocking_activity.query AS current_statement_in_blocking_processFROM pg_catalog.pg_locks blocked_locksJOIN pg_catalog.pg_stat_activity blocked_activity ON blocked_activity.pid = blocked_locks.pidJOIN pg_catalog.pg_locks blocking_locksON blocking_locks.locktype = blocked_locks.locktypeAND blocking_locks.database IS NOT DISTINCT FROM blocked_locks.databaseAND blocking_locks.relation IS NOT DISTINCT FROM blocked_locks.relationAND blocking_locks.page IS NOT DISTINCT FROM blocked_locks.pageAND blocking_locks.tuple IS NOT DISTINCT FROM blocked_locks.tupleAND blocking_locks.virtualxid IS NOT DISTINCT FROM blocked_locks.virtualxidAND blocking_locks.transactionid IS NOT DISTINCT FROM blocked_locks.transactionidAND blocking_locks.classid IS NOT DISTINCT FROM blocked_locks.classidAND blocking_locks.objid IS NOT DISTINCT FROM blocked_locks.objidAND blocking_locks.objsubid IS NOT DISTINCT FROM blocked_locks.objsubidAND blocking_locks.pid != blocked_locks.pidJOIN pg_catalog.pg_stat_activity blocking_activity ON blocking_activity.pid = blocking_locks.pidWHERE NOT blocked_locks.granted;-[ RECORD 1 ]-------------------------+--------------------------------------------------------blocked_pid | 23469blocked_user | postgresblocking_pid | 21207blocking_user | postgresblocked_statement | SELECT pg_catalog.set_config('search_path', '', false);current_statement_in_blocking_process | insert into t1 values(2);
消费也同样报错
[postgres@xiongcc ~]$ pg_recvlogical -d postgres --slot myslot --start -p 5432 -o pretty-print=1 -o add-msg-prefixes=wal2json -f -pg_recvlogical: error: could not send replication command "START_REPLICATION SLOT "myslot" LOGICAL 0/0 ("pretty-print" '1', "add-msg-prefixes" 'wal2json')": ERROR: replication slot "myslot" is active for PID 23469pg_recvlogical: disconnected; waiting 5 seconds to try again
所以还是老老实实terminate掉相关进程，重新操作。
postgres=# select pg_terminate_backend(23469);-[ RECORD 1 ]--------+--pg_terminate_backend | tpostgres=# select * from pg_replication_slots ;slot_name | plugin | slot_type | datoid | database | temporary | active | active_pid | xmin | catalog_xmin | restart_lsn | confirmed_flush_lsn | wal_status | safe_wal_size-----------+--------+-----------+--------+----------+-----------+--------+------------+------+--------------+-------------+---------------------+------------+---------------(0 rows)
长事务导致的WAL堆积
另外一个情形时，长事务导致的WAL日志堆积。WAL日志堆积或增长较快，较为常见的有如下几种情形：
1.归档失败，只有归档成功的WAL日志才会被回收，对应的现象就是pg_wal目录下大量的xxxx.ready2.过小的archive_timeout，到达 archive_timeout 会触发 WAL 文件的强制切换3.过大的wal_keep_segments，v13对应 wal_keep_size参数4.失效的复制槽，复制槽会一直保留restart_lsn之后的WAL文件，好在v13提供了max_slot_wal_keep_size，控制最大为复制槽保留多少WAL日志
今天讨论的情况是，长事务导致的WAL日志堆积。
首先需要知道的背景知识是，WAL record是一条条顺序记录在WAL文件中的，同时还要提一下被德哥吐槽过的《逻辑日志只有全局开关》，即
•开启wal_level = logical后，日志量会有较大增加。如果订阅的表比较少，而实际有用的logical日志占比较少，造成较大浪费。•如果订阅的表比较少, 在wal sender端解析时依旧需要解析并过滤不需要的wal，所以会造成wal的读浪费和CPU的解析浪费。
另外需要加深理解的是
逻辑复制也通过walsender实现，walsender不停地读取WAL日志，会对每一条WAL日志记录都进行解析，将解析出的元组按照事务进行分组（保存进RecorderBuffer)，并按照事务的开始时间进行排序。在事务提交时，会对提交事务在RecorderBuffer中的所有信息进行解码，并将解码后的逻辑日志发送给备机或者接收工具（例如pg_recvlogical）
先创建发布和订阅
[postgres@xiongcc ~]$ psql -p 5432psql (13.2)Type "help" for help.postgres=# create publication pub1 for table test;CREATE PUBLICATION
[postgres@xiongcc ~]$ psql -p 5439psql (13.2)Type "help" for help.postgres=# create subscription sub1 connection 'host=localhost port=5432' publication pub1;NOTICE: created replication slot "sub1" on publisherCREATE SUBSCRIPTION
WAL日志从0000000100000000000000DC开始保留
postgres=# select slot_name,slot_type,active_pid,active,pg_walfile_name(restart_lsn) from pg_replication_slots ;slot_name | slot_type | active_pid | active | pg_walfile_name-----------+-----------+------------+--------+--------------------------sub1 | logical | 27205 | t | 0000000100000000000000DC(1 row)postgres=# select pg_walfile_name(sent_lsn) as sent_wal,pg_walfile_name(write_lsn) as write_wal,pg_walfile_name(flush_lsn) as flush_wal,pg_walfile_name(replay_lsn) as replay_wal,sent_lsn,write_lsn,flush_lsn,replay_lsn from pg_stat_replication ;-[ RECORD 1 ]------------------------sent_wal | 0000000100000000000000DCwrite_wal | 0000000100000000000000DCflush_wal | 0000000100000000000000DCreplay_wal | 0000000100000000000000DCsent_lsn | 0/DC54FAC8write_lsn | 0/DC54FAC8flush_lsn | 0/DC54FAC8replay_lsn | 0/DC54FAC8[postgres@xiongcc ~]$ ll pgdata/pg_wal/total 70984-rw------- 1 postgres postgres 16777216 Sep 21 23:02 0000000100000000000000DC-rw------- 1 postgres postgres 16777216 Sep 21 21:16 0000000100000000000000DD-rw------- 1 postgres postgres 16777216 Sep 21 21:18 0000000100000000000000DE-rw------- 1 postgres postgres 16777216 Sep 21 21:18 0000000100000000000000DF-rw------- 1 postgres postgres 16777216 Sep 21 21:16 0000000100000000000000E0drwx------ 2 postgres postgres 4096 Sep 5 20:40 archive_status[postgres@xiongcc ~]$ ll pgdata/pg_wal/ | wc -l7
模拟一下，开启长事务，同时进行压测，不断产生WAL日志
[postgres@xiongcc ~]$ ll pgdata/pg_wal/ | wc -l7[postgres@xiongcc ~]$ ll pgdata/pg_wal/ | wc -l10[postgres@xiongcc ~]$ ll pgdata/pg_wal/ | wc -l12
还是从0000000100000000000000DC开始保留，但发送的相关lsn已经及时推进了
postgres=# select slot_name,slot_type,active_pid,active,pg_walfile_name(restart_lsn) from pg_replication_slots ;slot_name | slot_type | active_pid | active | pg_walfile_name-----------+-----------+------------+--------+--------------------------sub1 | logical | 27205 | t | 0000000100000000000000DC(1 row)postgres=# select pg_walfile_name(sent_lsn) as sent_wal,pg_walfile_name(write_lsn) as write_wal,pg_walfile_name(flush_lsn) as flush_wal,pg_walfile_name(replay_lsn) as replay_wal,sent_lsn,write_lsn,flush_lsn,replay_lsn from pg_stat_replication ;-[ RECORD 1 ]------------------------sent_wal | 000000010000000100000009write_wal | 000000010000000100000009flush_wal | 000000010000000100000009replay_wal | 000000010000000100000009sent_lsn | 1/9F1F260write_lsn | 1/9F1F260flush_lsn | 1/9F1F260replay_lsn | 1/9F1F260postgres=# select pg_walfile_name(sent_lsn) as sent_wal,pg_walfile_name(write_lsn) as write_wal,pg_walfile_name(flush_lsn) as flush_wal,pg_walfile_name(replay_lsn) as replay_wal,sent_lsn,write_lsn,flush_lsn,replay_lsn from pg_stat_replication ;-[ RECORD 1 ]------------------------sent_wal | 00000001000000010000000Awrite_wal | 00000001000000010000000Aflush_wal | 00000001000000010000000Areplay_wal | 00000001000000010000000Asent_lsn | 1/A26D338write_lsn | 1/A26D338flush_lsn | 1/A26D338replay_lsn | 1/A26D338
不一会儿，WAL日志的量就上来了
[postgres@xiongcc ~]$ ll pgdata/pg_wal/ | wc -l53[postgres@xiongcc ~]$ ll pgdata/pg_wal/ | head -n 5total 858608-rw------- 1 postgres postgres 16777216 Sep 21 23:04 0000000100000000000000DC-rw------- 1 postgres postgres 16777216 Sep 21 23:04 0000000100000000000000DD-rw------- 1 postgres postgres 16777216 Sep 21 23:04 0000000100000000000000DE-rw------- 1 postgres postgres 16777216 Sep 21 23:04 0000000100000000000000DF
所以，长事务也是会导致WAL日志不断堆积的。
如何溯源长事务
相比不少DBA肯定和开发撕逼过，开发会振振有词地说道：长事务？肯定不可能的，我们的应用代码肯定会及时提交的，那么假如遇到了WAL日志持续上涨堆积，但是发现复制槽的状态可能在我们去查看前变成了active，同时没有归档失败和较大的wal_keep_segments的时候，该怎么怼回去呢？该如何找到业务确实跑了长事务，反将一军呢？
我们可以从WAL日志下手，比如目前我这个环境
postgres=# select * from pg_replication_slots;-[ RECORD 1 ]-------+-----------slot_name | sub1plugin | pgoutputslot_type | logicaldatoid | 13578database | postgrestemporary | factive | tactive_pid | 27205xmin |catalog_xmin | 497297restart_lsn | 0/DC54FA90confirmed_flush_lsn | 1/14A9DB80wal_status | extendedsafe_wal_size |
可以看到，wal_status是extended，意味着虽然超过了max_wal_size但是仍然因为复制槽或者wal_keep_size需要保留
extended means that max_wal_size is exceeded but the files are still retained, either by the replication slot or by wal_keep_size.
另外，还是引用一下张树杰老师中的原话：
在WAL日志中已经包含了所有的用户数据，即常规的DML操作的数据和操作类型都包含在WAL日志中，但是WAL日志已经将这些信息保存成二进制形式，根据WAL日志本身无法将这些信息反向解析成逻辑日志。在decoding的过程中，需要借助元数据才能将二进制信息转换成逻辑操作信息。因此，需要构建访问元数据信息的事务快照，这个事务快照是在WAL日志记录生成时对应事务状态的一个切面，所以必须保证解析WAL日志记录所需的元数据尚未被清理。
PostgreSQL在复制槽中会保留catalog_xmin作为最小事务ID，只有小于该事务ID的元组才会被清理。ReplicationSlotsComputeRequiredXmin会计算所有复制槽的xmin，然后保存到ProcArrayStruct（proc全局变量）中，在vacuum时会根据这个事务ID防止清理掉需要的数据。在进行逻辑解码的过程中，要获得快照，最主要的就是要获得快照中的xmin、xmax和xip等信息。这些信息代表了当前活跃事务的范围及清单，但这些信息没有办法通过当前的procArray获得。因此我们要获得的是生成WAL日志记录时的快照，而WAL日志记录可能是在很久之前生成的，通过procArray只能获得当前的快照，和WAL日志记录产生时对应的快照可能不一致，因此元数据在这期间可能被修改。
幸运的时，Hot Standby功能通过LogStandbySnapshot函数记录事务的运行信息。在逻辑日志解码的过程中，可以借助XLOG_RUNNING_XACTS（xl_running_xacts结构体）类型的日志记录进行历史快照（Snapshot）的构建。
通过上述的引用，我们可以简单得出两点：
1.在逻辑解码中，必须要知道当时的表结构是啥样的。由于表结构可能变化，或者表可能被删除，即元数据发生了变化，catalog会发生变化，那么这些dead tuple必须要保留到翻译对应的WAL还需要的时候。这个也就是pg_replication_slots视图里面的catalog_xmin要做的事2.XLOG_RUNNING_XACTS类型的日志记录了历史快照
那么我们看一下XLOG_RUNNING_XACTS类型的日志，使用pg_waldump看一下
[postgres@xiongcc pg_wal]$ pg_waldump -s 0/DC54FA90 | grep -i 'XLOG_RUNNING_XACTS' | more---解析需要点时间
很不幸，等了半天，报了个错，这是因为解析到了末尾，在做recovery的时候，回放完成的时候也会打印如下日志
[postgres@xiongcc pg_wal]$ pg_waldump -s 0/DC54FA90 | grep -i 'XLOG_RUNNING_XACTS' | morepg_waldump: fatal: error in WAL record at 1/18D658F0: invalid record length at 1/18D65928: wanted 24, got 0
那换个方式
[postgres@xiongcc pg_wal]$ pg_waldump -s 0/DC54FA90 -n 1000 | more | grep RUNNING_XACTSrmgr: Standby len (rec/tot): 50/ 50, tx: 0, lsn: 0/DC54FA90, prev 0/DC54FA58, desc: RUNNING_XACTS nextXid 497297 latestCompletedXid 497296 oldestRunningXid 497297rmgr: Standby len (rec/tot): 54/ 54, tx: 0, lsn: 0/DC5508E0, prev 0/DC54FAC8, desc: RUNNING_XACTS nextXid 497298 latestCompletedXid 497296 oldestRunningXid 497297; 1 xacts: 497297
这里可以看到，
nextXid 497297 latestCompletedXid 497296 oldestRunningXid 497297下一个事务ID为497297，最近完成的事务ID是latestCompletedXid 497296，最老正在运行的是497297，第一个长事务还未提交，事务ID正是497297
postgres=# begin;BEGINpostgres=*# insert into t1 values(1);INSERT 0 1postgres=*# select txid_current();-[ RECORD 1 ]+-------txid_current | 497297
postgres=# select pg_xact_commit_timestamp('497296'::xid); ---已经提交的最老事务是497296，完成时间是23：02pg_xact_commit_timestamp-------------------------------2021-09-21 23:02:08.526366+08(1 row)postgres=# select pg_walfile_name('0/DC54FA90'); ---根据restart_lsn找到相关的WAL日志，获取时间点pg_walfile_name--------------------------0000000100000000000000DC(1 row)postgres=# \! ls -l pgdata/pg_wal | grep 0000000100000000000000DC-rw------- 1 postgres postgres 16777216 Sep 21 23:04 0000000100000000000000DC
所以，至此我们可以看到，这个长事务是23：04的时候左右开启的，现在提交一下
postgres=# begin;BEGINpostgres=*# insert into t1 values(1);INSERT 0 1postgres=*# select txid_current();-[ RECORD 1 ]+-------txid_current | 497297postgres=*# commit ;COMMIT
提交时间变了，变成了23:22:23，所以这个长事务在这个案例中就跑了18分钟！
postgres=# select pg_walfile_name('0/DC54FA90');pg_walfile_name--------------------------0000000100000000000000DC(1 row)postgres=# \! ls -l pgdata/pg_wal | grep 0000000100000000000000DC-rw------- 1 postgres postgres 16777216 Sep 21 23:04 0000000100000000000000DCpostgres=# select pg_xact_commit_timestamp('497297'::xid);pg_xact_commit_timestamp-------------------------------2021-09-21 23:22:23.011386+08(1 row)postgres=# select slot_name,slot_type,active_pid,active,pg_walfile_name(restart_lsn) from pg_replication_slots ;slot_name | slot_type | active_pid | active | pg_walfile_name-----------+-----------+------------+--------+--------------------------sub1 | logical | 27205 | t | 000000010000000100000018(1 row)
复制槽的保留位置也变成了000000010000000100000018，即最新的
[postgres@xiongcc pg_wal]$ ls -lrt | tail -n 5-rw------- 1 postgres postgres 16777216 Sep 21 23:08 000000010000000100000015-rw------- 1 postgres postgres 16777216 Sep 21 23:08 000000010000000100000016-rw------- 1 postgres postgres 16777216 Sep 21 23:08 000000010000000100000017-rw------- 1 postgres postgres 16777216 Sep 21 23:15 000000010000000100000019-rw------- 1 postgres postgres 16777216 Sep 21 23:22 000000010000000100000018
再做个checkpoint，WAL日志也正常被回收，数量下来了。
[postgres@xiongcc pg_wal]$ ls -lrt | tail -n 5-rw------- 1 postgres postgres 16777216 Sep 21 23:07 000000010000000100000026-rw------- 1 postgres postgres 16777216 Sep 21 23:08 00000001000000010000001E-rw------- 1 postgres postgres 16777216 Sep 21 23:08 00000001000000010000002A-rw------- 1 postgres postgres 16777216 Sep 21 23:15 000000010000000100000019-rw------- 1 postgres postgres 16777216 Sep 21 23:24 000000010000000100000018[postgres@xiongcc pg_wal]$ ls -lrt | wc -l25
所以，碰到大事务，和开发扯皮的时候，我们是有据可依的。不过前提是需要开启track_commit_timestamp，这个会追踪事务的提交时间。
v14对于逻辑复制的优化
大事务的流式传输
前面主要介绍了长事务的另外几个危害：
1.阻止逻辑解码的创建2.导致WAL日志堆积
其实可以根据原理推敲一下，同一条事务的WAL日志可能不是连续的（比如该例中的大事务，N久之后再做个写入动作），这个期间其他的WAL record继续连续顺序存放，使得WAL日志不断填充增长。而逻辑解码要求同一事务的逻辑日志要按顺序相邻排列并传送给订阅端，在RecordBuffer里面有一个TXN的唯一标识，用于标识事务。所以，假如开启了一个大事务，得从事务最开始的位点所在WAL日志，一直保留到事务提交时的WAL日志。
传统的OutputPluginCallbacks定义了三种类型（本质就是回调函数，通过函数指针调用）：
1.普通的事务解码接口，也就是目前已经存在的，普通的事务会使用该接口并结合快照对RecordBufferChange逐个解析2.两阶段提交事务的解码接口，v14已经支持了部分，test_decoding已经支持，但是原生的逻辑复制还不支持（pgoutput），预计v15支持，可以参照https://git.postgresql.org/gitweb/?p=postgresql.git;a=commit;h=a8fd13cab0ba815e9925dc9676e6309f699b5f72，**Add support for prepared transactions to built-in logical replication.**3.长事务的解码接口，(stream系列)，v14已经支持，逻辑复制支持流式decoding和发送, 无需等待事务结束, 大幅度降低大事务、 长事务的复制延迟，主要是下面几个接口：stream_start_cb、stream_stop_cb、stream_abort_cb、stream_commit_cb 和 stream_change_cb）和两个可选的回调（stream_message_cb 和 stream_truncate_cb）
在v14以前，处理大型事务通常会导致明显的应用延迟，因为只有在提交时，所有更改才会同时发送到下游。主要目的是为了避免复制可能最终被中止的事务。
对于大型交易，进行网络传输的时间可能很长，这会导致应用延迟。另外，假如解码的数量超过了 **logical_decoding_work_mem **参数，会溢出到磁盘，进一步降低性能。
在v14里面，允许流式传输正在进行的事务，而不是溢出到磁盘上。同时在应用方面，流式接收更改并把它们写入到文件中。这样在提交时，更改就会在本地重放，而无需等待大量的数据传输。优点如下：
1.提交时无需通过网络发送大量数据。2.如果在不相关的表（非发布表）上发生了一些重大更改，那么可以提前丢弃这些更改，而不是将它们溢出到本地文件。
同时，在v14里面，每当 walsender 进程中的更改超过了 logical_decoding_work_mem 大小时，会根据其内存大小选择最大的事务并流式传输其所有更改。然后一直重复这个过程，直到达到 logical_decoding_work_mem 的内存限制。
新增了一个视图来监控溢出到磁盘的数量
[postgres@xiongcc ~]$ psql -p 5555psql (14beta2)Type "help" for help.postgres=#postgres=# \d pg_stat_replication_slotsView "pg_catalog.pg_stat_replication_slots"Column | Type | Collation | Nullable | Default--------------+--------------------------+-----------+----------+---------slot_name | text | | |spill_txns | bigint | | |spill_count | bigint | | |spill_bytes | bigint | | |stream_txns | bigint | | |stream_count | bigint | | |stream_bytes | bigint | | |total_txns | bigint | | |total_bytes | bigint | | |stats_reset | timestamp with time zone | | |
见如下样例
[postgres@xiongcc ~]$ psql -p 5555psql (14beta2)Type "help" for help.postgres=# set logical_decoding_work_mem to '64kB';SETpostgres=# CREATE TABLE stream_test(data text);CREATE TABLEpostgres=# SELECT 'init' FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding');?column?----------init(1 row)postgres=# INSERT INTO stream_test SELECT repeat('a', 6000) || g.i FROM generate_series(1, 500) g(i);INSERT 0 500postgres=# SELECT data FROM pg_logical_slot_get_changes('regression_slot', NULL,NULL, 'include-xids', '1', 'skip-empty-xacts', '1', 'stream-changes', '1');data-----------------------------------------------------opening a streamed block for transaction TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492streaming change for TXN 131492......
可以看到，已经有一些溢出了，然后进行了流式传输
postgres=# select * from pg_stat_replication_slots where slot_name = 'regression_slot';-[ RECORD 1 ]+----------------slot_name | regression_slotspill_txns | 0spill_count | 0spill_bytes | 0stream_txns | 1stream_count | 2stream_bytes | 104883total_txns | 1total_bytes | 104883stats_reset |
而发布订阅的例子可以参照如下
Publisher node:Set logical_decoding_work_mem = '64kB';# Set up publication with some initial dataCREATE TABLE test_tab (a int primary key, b varchar);INSERT INTO test_tab VALUES (1, 'foo'), (2, 'bar');CREATE PUBLICATION tap_pub FOR TABLE test_tab;Subscriber node:CREATE TABLE test_tab (a int primary key, b varchar);CREATE SUBSCRIPTION tap_sub CONNECTION 'host=localhost port=5432 dbname=postgres' PUBLICATION tap_pub WITH (streaming = on);Publisher Node:# Ensure the corresponding replication slot is created on publisher nodeselect slot_name, plugin, slot_type from pg_replication_slots;slot_name | plugin | slot_type-----------+----------+-----------tap_sub | pgoutput | logical(1 row)# Confirm there is no streamed bytes yetpostgres=# SELECT slot_name, stream_txns, stream_count, stream_bytes FROM pg_stat_replication_slots;slot_name | stream_txns | stream_count | stream_bytes-----------+-------------+--------------+--------------tap_sub | 0 | 0 | 0(1 row)# Insert, update and delete enough rows to exceed the logical_decoding_work_mem (64kB) limit.BEGIN;INSERT INTO test_tab SELECT i, md5(i::text) FROM generate_series(3, 5000) s(i);UPDATE test_tab SET b = md5(b) WHERE mod(a,2) = 0;DELETE FROM test_tab WHERE mod(a,3) = 0;# Confirm that streaming happenedSELECT slot_name, stream_txns, stream_count, stream_bytes FROM pg_stat_replication_slots;slot_name | stream_txns | stream_count | stream_bytes-----------+-------------+--------------+--------------tap_sub | 1 | 22 | 1444410(1 row)Subscriber Node:# The streamed data is still not visible.select * from test_tab;a | b---+-----1 | foo2 | bar(2 rows)Publisher Node:# Commit the large transactionsCommit;Subscriber Node:# The data must be visible on the subscriberselect count(*) from test_tab;count-------3334(1 row)
预备事务
在PostgreSQL14里面，已经支持了2pc的解码，不过目前仅有test_decoding支持，wal2json测试了一下不行。这也就意味着，允许通过逻辑复制跨多个节点的两阶段分布式事务，骚操作不断。
相关的钩子
| Callback name | Description |
| filter_prepare_cb | Allows plugins to filter transactions that do not need to be decoded at prepare time based on the GID used in the PREPARE TRANSACTION command. |
| begin_prepare_cb | The start of a prepared transaction. |
| prepare_cb | Called when the PREPARE TRANSACTION command is decoded. |
| commit_prepared_cb | Called when the COMMIT PREPARED command is decoded. |
| rollback_prepared_cb | Called when the ROLLBACK PREPARED command is decoded. |
[postgres@xiongcc ~]$ psql -p 5555psql (14beta2)Type "help" for help.postgres=# SELECT * FROM pg_create_logical_replication_slot('regression_slot', 'wal2json', false, true );slot_name | lsn-----------------+------------regression_slot | 0/38A53F20(1 row)postgres=# CREATE TABLE data(id serial primary key, data text);CREATE TABLEpostgres=# BEGIN;BEGINpostgres=*# INSERT INTO data(data) VALUES('5');INSERT 0 1postgres=*# PREPARE TRANSACTION 'test_prepared1';PREPARE TRANSACTIONpostgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL)postgres-# ;lsn | xid | data------------+--------+---------------0/38A7A250 | 131503 | {"change":[]} ---可以看到，wal2json不支持解析2pc(1 row)postgres=# rollback ;WARNING: there is no transaction in progressROLLBACKpostgres=# select pg_drop_replication_slot('regression_slot');pg_drop_replication_slot--------------------------(1 row)postgres=# rollback prepared 'test_prepared1';ROLLBACK PREPARED
再看一下test_decoding
postgres=# SELECT * FROM pg_create_logical_replication_slot('regression_slot', 'test_decoding', false, true );slot_name | lsn-----------------+------------regression_slot | 0/38A7A650(1 row)postgres=# BEGIN;BEGINpostgres=*# INSERT INTO data(data) VALUES('5');INSERT 0 1postgres=*# PREPARE TRANSACTION 'test_prepared1';PREPARE TRANSACTIONpostgres=# SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);lsn | xid | data------------+--------+---------------------------------------------------------0/38A7C538 | 131505 | BEGIN 1315050/38A7C5A0 | 131505 | table public.data: INSERT: id[integer]:2 data[text]:'5'0/38A7C7E8 | 131505 | PREPARE TRANSACTION 'test_prepared1', txid 131505(3 rows)
可以看到，在v14中已经有了基础的infrastructure，允许解码两阶段提交。下一步就是将两阶段的支持实现到PostgreSQL内部最大的逻辑解码插件——pgoutput中。富士通 OSS 团队正在与开源社区合作，预估可以在PostgreSQL 15 看到此功能。
其他增强
逻辑解码能力的增强
Reduced the CPU usage and improve decoding performance of transactions having a lot of DDLs. It has been observed that decoding of a transaction containing truncation of a table with 1000 partitions would be finished in 1s whereas before this work it used to take 4-5 minutes.
允许pgoutput获得消息了
意味着可以使用pgoutput来进行CDC了
Provide a “messages” option to the pgoutput plugin. This allows logical decoding messages (i.e. generated via pg_logical_emit_message) to be sent to the slot consumer. This is useful for pgoutput plugin users that use it for Change Data Capture.
postgres=# SELECT pg_create_logical_replication_slot('pgout_slot','pgoutput');pg_create_logical_replication_slot------------------------------------(pgout_slot,0/38A93C68)(1 row)postgres=# SELECT pg_logical_emit_message(true, 'pgoutput', 'a transactional message');pg_logical_emit_message-------------------------0/38A96510(1 row)postgres=# SELECT get_byte(data, 1), encode(substr(data, 24, 23), 'escape') FROM pg_logical_slot_peek_binary_changes('pgout_slot', NULL, NULL, 'proto_version', '1', 'publication_names', 'pgout_slot', 'messages', 'true') OFFSET 1 LIMIT 1;get_byte | encode----------+-------------------------1 | a transactional message(1 row)postgres=# SELECT pg_drop_replication_slot('pgout_slot');pg_drop_replication_slot--------------------------(1 row)
允许二进制传输
速度会更快，可以使用在跨版本升级中，不过要求两端都要有对应的二进制接收函数
When doing cross-version replication, if the subscriber lacks a binary receive function for the type, the data transfer will fail, and this option can't be used.
postgres=# CREATE SUBSCRIPTION mysub CONNECTION 'host=localhost port=5432 dbname=postgres' PUBLICATION mypub1 WITH (binary = true);NOTICE: created replication slot "mysub" on publisherCREATE SUBSCRIPTION
初始同步阶段的速度增强
如果原始表的记录数特别多, 初始数据同步时间比较长, 在这个过程中可能产生了很多新事务, 会带来一系列问题：
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已改成使用永久的复制槽而不是临时复制槽。复制源信息会保存，用于跟踪已经复制的数据，因此通过使用永久插槽，这意味着在崩溃/重新启动后，可以从最近记录的检查点再次继续复制。详情参考https://www.postgresql.fastware.com/blog/logical-replication-tablesync-workers
命名格式如下：pg<subscription OID>sync<table relid><system id>
postgres=# select * from pg_replication_slots where slot_name like '%pg%';-[ RECORD 1 ]-------+----------------------------------------slot_name | pg_16388_sync_16384_7010596180506970511plugin | pgoutputslot_type | logicaldatoid | 13890database | postgrestemporary | factive | factive_pid |xmin |catalog_xmin | 738restart_lsn | 0/3B271498confirmed_flush_lsn | 0/3B2714D0wal_status | reservedsafe_wal_size |two_phase | f
小结
1.逻辑复制的功能不断增强，v14针对2pc已经有了infrastructure，在v15中应该可以在pgoutput中得到支持，目前仅支持test_decoding解析2pc，这意味着，允许通过逻辑复制跨多个节点进行两阶段分布式事务，骚操作不断，意味着以后可以做的操作越来越骚，天马行空。2.v14针对逻辑复制最开始的table sync阶段也进行了增强，可以断点续传，使用到了永久复制槽3.v14允许流式传输事务，延迟可以得到显著降低4.长事务老生常谈的DB killer了，在此篇中新增两项罪名5.碰大事务不用怕，有了pg_waldump，有据可依，不怕撕逼
参考
http://amitkapila16.blogspot.com/2021/09/logical-replication-improvements-in.html
https://amitkapila16.blogspot.com/2021/07/logical-replication-of-in-progress.html
https://www.postgresql.fastware.com/blog/logical-decoding-of-two-phase-commits
https://www.postgresql.org/message-id/CAFPTHDZv4XiK2VJG_fpjuvq7rv2-o4185uzJJvmzgAW%3D64FY8Q%40mail.gmail.com
https://www.enterprisedb.com/blog/logical-decoding-large-progress-transactions-postgresql
https://www.postgresql.org/docs/14/monitoring-stats.html#MONITORING-PG-STAT-REPLICATION-SLOTS-VIEW
https://github.com/digoal/blog
https://www.postgresql.fastware.com/blog/logical-replication-tablesync-workers
原文作者：PostgreSQL学徒；如有侵权，请联系管理员删除。
