MogDB逻辑解码与pg_recvlogical
概述
谈到逻辑解码需要先从流复制开始说起。流复制的最重要的一个用途就是实现数据库的热备，数据库的主备同步方式是物理级别的数据同步。但实际应用场景中仅仅通过物理赋值是无法满足业务需求的，因此提供了逻辑复制的功能。
逻辑复制主要解决了以物理赋值无法解决的一些问题，例如：
-
指定库或部分表的复制需求
-
将多个数据库实例的数据汇聚到同一个目标库
-
将一个库的数据分发到多个不同的库
-
不同的版本之间的复制
-
不同库名之间的表同步
逻辑复制的关键是将WAL日志的内容进行逻辑解码成特定的格式，如json，SQL等。pg_recvlogical 客户端工具就是逻辑解码的一种典型应用，它将WAL日志解码为json格式，保存在指定文件或标准输出stdout中。
逻辑复制约束
需要修改wal_level，在MogDB中wal_level有如下取值：
- minimal
- 优点：一些重要操作（包括创建表、创建索引、簇操作和表的复制）都能安全的跳过，这样就可以使操作变得更快。
- 缺点：WAL仅提供从数据库服务器崩溃或者紧急关闭状态恢复时所需要的基本信息，无法用WAL归档日志恢复数据。
- archive
- 这个参数增加了WAL归档需要的日志信息，从而可以支持数据库的归档恢复。
- hot_standby
- 这个参数进一步增加了在备机上运行的SQL查询的信息，这个参数只能在数据库服务重新启动后生效。
- 为了在备机上开启只读查询，wal_level必须在主机上设置成hot_standby ，并且备机必须打开hot_standby参数。
- logical
- 这个参数表示WAL日志支持逻辑复制。
默认值: minimal
- 这个参数表示WAL日志支持逻辑复制。
须知：
- 如果需要启用WAL日志归档和主备机的数据流复制，必须将此参数设置为archive或者hot_standby。
- 如果此参数设置为minimal，archive_mode必须设置为off，hot_standby必须设置为off，max_wal_senders参数设置为0，且需为单机环境，否则将导致数据库无法启动。
- 如果此参数设置为archive，hot_standby必须设置为off，否则将导致数据库无法启动。但是，hot_standby在双机环境中不能设置为off，具体参见hot_standby参数说明。
注意：
- 与PostgreSQL的wal_level略有不同，PostgreSQL只包括3种级别：minimal, replica, or logical
- PostgreSQL中wal_level的默认值是
replica，它会写入足够的数据以支持WAL归档和复制，包括在后备服务器上运行只读查询。minimal会去掉除从崩溃或者立即关机中进行恢复所需的信息之外的所有记录。最后，logical会增加支持逻辑解码所需的信息。每个层次包括所有更低层次记录的信息。这个参数只能在服务器启动时设置。
postgres.conf配置
wal_level = logical # minimal, archive, hot_standby or logical
# (change requires restart)
max_wal_senders = 10 # max number of walsender processes
# (change requires restart)
wal_keep_segments = 16 # in logfile segments, 16MB each; 0 disables
max_replication_slots = 8 # max number of replication slots.i
注：修改配置后需要重启MogDB
创建replication用户
MogDB=#CREATE ROLE pub_sub_user WITH SYSADMIN REPLICATION LOGIN PASSWORD 'pub_sub@123';
NOTICE: The encrypted password contains MD5 ciphertext, which is not secure.
CREATE ROLE
MogDB=#
注：需要有SYSADMIN 和REPLICATION 权限
pg_hba.conf配置
# replication privilege.
#local replication omm trust
#host replication omm 127.0.0.1/32 trust
#host replication omm ::1/128 trust
host all all 0.0.0.0/0 md5
host replication all 0.0.0.0/0 md5
场景限制
- 目前不支持DDL解析，只能解析DML（INSERT、UPDATE、DELETE，TRUNCATE）；
- TEMPORARY表和UNLOGGED表不会被复制；
- 表必须有主键或唯一约束，否则像update或delete操作无法被复制；
- 序列不被复制；
- 大对象不被复制；
- 新增加的表，不会自动加入订阅，需要在订阅端进行刷新；
pg_recvlogical工作原理
pg_recvlogical可以作为观察wal日志具体变化的工具。
整体架构
如上图：
- pg_recvlogical通过libpq与MogDB server建立链接
- 通过
CREATE_REPLICATION_SLOT建立逻辑复制槽 - 运行
START_REPLICATION,循环处理wal流，并进行解码 - 将解码数据写入文件或者在stdout屏幕数据
- pg_logical_slot_get_changes、pg_logical_slot_peek_changes可以查看修改情况
DROP_REPLICATION_SLOT销毁复制槽
pg_recvlogical主流程
- 设置插件
- pg_recvlogical默认使用
mppdb_decoding
- pg_recvlogical默认使用
plugin = pg_strdup("mppdb_decoding");
- pg_recvlogical也可以通过命令行参数
--plugin设置插件 如：–plugin = wal2json
通过ouput/decode插件处理pg_recvlogical可以将wal变更内容输出为特定格式（TXT、JISON）。
- 命令行处理
pg_recvlogical receives logical change stream.
Usage:
pg_recvlogical [OPTION]...
Options:
-f, --file=FILE receive log into this file. - for stdout
-n, --no-loop do not loop on connection lost
-v, --verbose output verbose messages
-V, --version output version information, then exit
-?, --help show this help, then exit
Connection options:
-d, --dbname=DBNAME database to connect to
-h, --host=HOSTNAME database server host or socket directory
-p, --port=PORT database server port number
-U, --username=NAME connect as specified database user
-w, --no-password never prompt for password
-W, --password force password prompt (should happen automatically)
Replication options:
-F --fsync-interval=INTERVAL
frequency of syncs to the output file (in seconds, defaults to 10)
-o, --option=NAME[=VALUE]
Specify option NAME with optional value VAL, to be passed
to the output plugin
-P, --plugin=PLUGIN use output plugin PLUGIN (defaults to mppdb_decoding)
-s, --status-interval=INTERVAL
time between status packets sent to server (in seconds, defaults to 10)
-S, --slot=SLOT use existing replication slot SLOT instead of starting a new one
-I, --startpos=PTR Where in an existing slot should the streaming start
-r, --raw parallel decoding output raw results without converting to text format
Action to be performed:
--create create a new replication slot (for the slotname see --slot)
--start start streaming in a replication slot (for the slotname see --slot)
--drop drop the replication slot (for the slotname see --slot)
- CREATE_REPLICATION_SLOT
- DROP_REPLICATION_SLOT
- START_REPLICATION
以上3个命令通过词法和语法解析src/backend/replication/repl_scanner.l、src/gausskernel/storage/replication/repl_gram.y
/* CREATE_REPLICATION_SLOT SLOT slot [%X/%X] */
create_replication_slot:
/* CREATE_REPLICATION_SLOT SLOT slot PHYSICAL [init_slot_lsn] */
K_CREATE_REPLICATION_SLOT IDENT K_PHYSICAL RECPTR
{
CreateReplicationSlotCmd *cmd;
cmd = makeNode(CreateReplicationSlotCmd);
cmd->kind = REPLICATION_KIND_PHYSICAL;
cmd->slotname = $2;
cmd->init_slot_lsn = $4;
$$ = (Node *) cmd;
}
/* CREATE_REPLICATION_SLOT slot LOGICAL plugin */
| K_CREATE_REPLICATION_SLOT IDENT K_LOGICAL IDENT
{
CreateReplicationSlotCmd *cmd;
cmd = makeNode(CreateReplicationSlotCmd);
cmd->kind = REPLICATION_KIND_LOGICAL;
cmd->slotname = $2;
cmd->plugin = $4;
$$ = (Node *) cmd;
}
;
/* DROP_REPLICATION_SLOT slot */
drop_replication_slot:
K_DROP_REPLICATION_SLOT IDENT
{
DropReplicationSlotCmd *cmd;
cmd = makeNode(DropReplicationSlotCmd);
cmd->slotname = $2;
$$ = (Node *) cmd;
}
;
- 加载插件
output插件都要实现_PG_output_plugin_init 函数，和必要的回调函数。
回调函数集合：
/*
* Output plugin callbacks
*/
typedef struct OutputPluginCallbacks {
LogicalDecodeStartupCB startup_cb;
LogicalDecodeBeginCB begin_cb;
LogicalDecodeChangeCB change_cb;
LogicalDecodeCommitCB commit_cb;
LogicalDecodeAbortCB abort_cb;
LogicalDecodePrepareCB prepare_cb;
LogicalDecodeShutdownCB shutdown_cb;
LogicalDecodeFilterByOriginCB filter_by_origin_cb;
} OutputPluginCallbacks;
主要的几个回调函数包括（mppdb_decoding为例）：
- pg_decode_begin_txn --开始事务
- pg_decode_change --增、删、改、truncate
- pg_decode_commit_txn --提交事务
演示
创建测试表
CREATE TABLE pub_sub
(
id SERIAL primary key ,
name character varying
);
创建复制槽
pg_recvlogical --create -S test_slot -d postgres
启动recvlogical
pg_recvlogical --start -S test_slot -d postgres -f - -- “-f -” 表示输出到stdout（屏幕）
注：pg_recvlogical会不停地每个5s检查复制槽是否有更新
插入数据
MogDB=#select * from pg_logical_slot_get_changes('test_slot', NULL, NULL);
location | xid | data
----------+-----+------
(0 rows)
MogDB=#insert into pub_sub (name) values('test pg_recvlogical 01');
INSERT 0 1
MogDB=#select * from pg_logical_slot_get_changes('test_slot', NULL, NULL);
location | xid | data
-----------+-------+-------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------
0/48C60D8 | 63062 | BEGIN 63062
0/48C60D8 | 63062 | {"table_name":"public.pub_sub","op_type":"INSERT","columns_name":["i","name"],"columns_type":["integer","character varying"],"columns_val":["23"
,"'test pg_recvlogical 01'"],"old_keys_name":[],"old_keys_type":[],"old_keys_val":[]}
0/48C62B8 | 63062 | COMMIT 63062 (at 2022-09-01 11:14:24.539429+00) CSN 47645
(3 rows)
MogDB=#select * from pg_logical_slot_peek_changes('test_slot', NULL, NULL);
location | xid | data
----------+-----+------
(0 rows)
MogDB=#insert into pub_sub (name) values('test pg_recvlogical 02');
INSERT 0 1
MogDB=#select * from pg_logical_slot_peek_changes('test_slot', NULL, NULL);
ERROR: replication slot "test_slot" is already active
MogDB=#select * from pg_logical_slot_peek_changes('test_slot', NULL, NULL);
location | xid | data
-----------+-------+-------------------------------------------------------------------------------------------------------------------------------------------------
--------------------------------------------------------------------------------------
0/48C6548 | 63063 | BEGIN 63063
0/48C6548 | 63063 | {"table_name":"public.pub_sub","op_type":"INSERT","columns_name":["i","name"],"columns_type":["integer","character varying"],"columns_val":["24"
,"'test pg_recvlogical 02'"],"old_keys_name":[],"old_keys_type":[],"old_keys_val":[]}
0/48C6728 | 63063 | COMMIT 63063 (at 2022-09-01 11:15:09.292569+00) CSN 47646
(3 rows)
pg_recvlogical输出如下
BEGIN 63063
{"table_name":"public.pub_sub","op_type":"INSERT","columns_name":["i","name"],"columns_type":["integer","character varying"],"columns_val":["24","'test pg_recvlogical 02'"],"old_keys_name":[],"old_keys_type":[],"old_keys_val":[]}
COMMIT 63063 (at 2022-09-01 11:15:09.292569+00) CSN 47646
因为pg_recvlogical默认使用mppdb_decoding插件，mppdb_decoding输出格式为json。
格式化后的json：
销毁复制槽
pg_recvlogical --drop -S test_slot -d postgres
换插件
- wal2json
$ pg_recvlogical --create -S test_slot -d postgres --plugin=wal2json $ pg_recvlogical --start -S test_slot -d postgres -f -
wal2json 没有显示BEGIN和COMMIT
警告：有概率会报一下错误，疑似BUG。
- pgoutput
$ pg_recvlogical --create -S test_slot -d postgres --plugin=pgoutput $ pg_recvlogical --start -S test_slot -d postgres -f -
更换为pgoutput 后回报版本错误：FATAL: client sent proto_version=0 but we only support protocol 1 or higher
SQL方式演示
MogDB=#SELECT * FROM pg_create_logical_replication_slot('regression_slot', 'mppdb_decoding');
slotname | xlog_position
-----------------+---------------
regression_slot | 0/48D3488
(1 row)
MogDB=#select * from pg_replication_slots;
slot_name | plugin | slot_type | datoid | database | active | xmin | catalog_xmin | restart_lsn | dummy_standby
-----------------+----------------+-----------+--------+----------+--------+------+--------------+-------------+---------------
user_sub | pgoutput | logical | 15016 | postgres | t | | 63071 | 0/48D3438 | f
wal2json | wal2json | logical | 15016 | postgres | f | | 61212 | 0/40EBA88 | f
test_slot | mppdb_decoding | logical | 15016 | postgres | f | | 61212 | 0/48C9A90 | f
regression_slot | mppdb_decoding | logical | 15016 | postgres | f | | 63071 | 0/48D3408 | f
(4 rows)
MogDB=#SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
location | xid | data
----------+-----+------
(0 rows)
MogDB=#CREATE TABLE data(id serial primary key, data text);
NOTICE: CREATE TABLE will create implicit sequence "data_id_seq" for serial column "data.id"
NOTICE: CREATE TABLE / PRIMARY KEY will create implicit index "data_pkey" for table "data"
CREATE TABLE
MogDB=#SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
location | xid | data
-----------+-------+-----------------------------------------------------------
0/48C9D30 | 63070 | BEGIN 63070
0/48D32E8 | 63070 | COMMIT 63070 (at 2022-09-01 11:51:40.602993+00) CSN 47650
(2 rows)
MogDB=#BEGIN;
BEGIN
MogDB=#INSERT INTO data(data) VALUES('1');
INSERT 0 1
MogDB=#INSERT INTO data(data) VALUES('2');
INSERT 0 1
MogDB=#COMMIT
#;
COMMIT
MogDB=#SELECT * FROM pg_logical_slot_get_changes('regression_slot', NULL, NULL);
location | xid | data
-----------+-------+-------------------------------------------------------------------------------------------------------------------------------------------------
-------------------------------------------------
0/48E0CD0 | 63073 | BEGIN 63073
0/48E0CD0 | 63073 | {"table_name":"public.data","op_type":"INSERT","columns_name":["id","data"],"columns_type":["integer","text"],"columns_val":["1","'1'"],"old_key
s_name":[],"old_keys_type":[],"old_keys_val":[]}
0/48E1090 | 63073 | {"table_name":"public.data","op_type":"INSERT","columns_name":["id","data"],"columns_type":["integer","text"],"columns_val":["2","'2'"],"old_key
s_name":[],"old_keys_type":[],"old_keys_val":[]}
0/48E11E0 | 63073 | COMMIT 63073 (at 2022-09-01 11:58:19.799037+00) CSN 47653
(4 rows)
MogDB=#SELECT pg_drop_replication_slot('regression_slot');
pg_drop_replication_slot
--------------------------
(1 row)
总结
逻辑复制/解码相对于物理复制更加灵活，也可以根据实际业务需要开发对应的逻辑解码插件，甚至可以当做ETL来使用。总的来说逻辑解码是对用户非常友好的接口。以上通过对MogDB逻辑复制、逻辑解码的原理和部分代码进行分析，利用pg_recvlogical和SQL演示逻辑解码的过程，希望对大家理解MogDB逻辑解码有所帮助。
最后修改时间：2022-11-16 14:23:46
「喜欢这篇文章，您的关注和赞赏是给作者最好的鼓励」
关注作者
【版权声明】本文为墨天轮用户原创内容，转载时必须标注文章的来源（墨天轮），文章链接，文章作者等基本信息，否则作者和墨天轮有权追究责任。如果您发现墨天轮中有涉嫌抄袭或者侵权的内容，欢迎发送邮件至：contact@modb.pro进行举报，并提供相关证据，一经查实，墨天轮将立刻删除相关内容。
