相关: 《Postgresql源码(66)insert on conflict语法介绍与内核执行流程解析》) 《Postgresql源码(70)逻辑复制DecodeXLOG主要流程和数据结构》
本篇只简单介绍逻辑复制解析过程,这里面的比较有意思的历史快照还没有看,后面在补一篇。
逻辑复制数据结构稍复杂,流程逻辑很简单。(主要是眼花了把reorder看成了record,困惑了很久😭)
walsender进程不断的从自己的复制槽中获取新产生的wal record,通过LogicalDecodingProcessRecord()函数进行wal record的初步过滤和解析,解析结果为一个ReorderBufferChange(解析半成品)(对于DML语句而言这个结构里面主要的信息为oldtuple和newtuple),并将这个解析结果放置到当前事务ID对应的解析buf中去。
当某个事务发生了提交,先通过ReorderBuffer维护的哈希表,通过XID找到ReorderBufferTXN,ReorderBufferTXN维护的链中的ReorderBufferChange(半成品)会被指定的plugin做二次处理(成品),并将处理结果按照指定的途径输出。
分解具体分成几步:
XLogReadRecord
模块拿到一条完整的的Xlog record
。LogicalDecodingProcessRecord
收到record
,将record
数据内容识别出来,封装到ReorderBufferChange
结构中。ReorderBufferChange
对应一次修改。ReorderBufferChange
串成链表,挂在ReorderBufferTXN
结构上。ReorderBuffer
维护一个哈希表,key为xid
,values为ReorderBufferTXN
。ReorderBuffer
的哈希表中,用xid找到ReorderBufferTXN
,在把维护的链表上的所有ReorderBufferChange
拿到,把半成品ReorderBufferChange
传入解析插件的回调函数,完成一个事务的处理。拿一个具体用例来看
drop table decoding_test;
CREATE TABLE decoding_test(x integer, y text);
SELECT pg_create_logical_replication_slot('test_slot', 'test_decoding');
-- 测试开始
SELECT pg_logical_slot_get_changes('test_slot', NULL, NULL);
pg_logical_slot_get_changes
-----------------------------
(0 rows)
INSERT INTO decoding_test(x,y) SELECT s, s::text FROM generate_series(1,2) s;
-- 调试位置
SELECT pg_logical_slot_get_changes('test_slot', NULL, NULL);
pg_logical_slot_get_changes
-------------------------------------------------------------------------------------
(7/80F087D0,4002996,"BEGIN 4002996")
(7/80F087D0,4002996,"table public.decoding_test: INSERT: x[integer]:1 y[text]:'1'")
(7/80F08A00,4002996,"table public.decoding_test: INSERT: x[integer]:2 y[text]:'2'")
(7/80F08A70,4002996,"COMMIT 4002996")