PostgreSQL 15 逻辑复制模式中所有的表

开源软件联盟PostgreSQL分会 2022-02-19

1730

翻译：魏波

中国PG分会培训认证执行总监、资深数据库工程师，十多年的数据库运维管理及培训经验，掌握PostgreSQL架构部署、性能优化等，致力于推动PostgreSQL在中国的发展。

作者：维涅什瓦兰 C（Vigneshwaran C）

2022 年 2 月 10 日

在这篇文章中，我想讨论富士通 OSS 团队与 PostgreSQL 开源社区合作为即将推出的 PostgreSQL 15 添加的新功能，以实现模式中所有表的逻辑复制。

背景

PostgreSQL 14 支持数据库中特定表或所有表从发布者到订阅者的逻辑复制。如果用户想要发布存在于一个模式或多个模式中的表，他们必须通过查询数据库手动整理表列表，然后使用列表创建发布。

当模式中只有几个表时，整理列表是快速和容易的，但如果有数百或数千个表，那么这将是一项繁琐的任务。

为了克服这个问题，即将推出的 PostgreSQL 15 将添加选项ALL TABLES IN SCHEMA，这将允许指定一个或多个模式，其表由发布者选择用于向订阅者发送数据。

注意：在这篇文章中，我将把这个新特性称为模式发布，它允许对模式中的所有表进行逻辑复制。

功能概述

下图说明了模式发布的逻辑复制工作：

架构发布的逻辑复制

让我们逐步完成上图中的各个阶段：

用户对表执行各种 DML 操作并执行prepare/commit。
后端进程会为用户执行的操作生成WAL日志。
后端向 WAL 发送者进程发送 SIGUSR1 信号，通知 WAL 记录可供处理。
walsender 进程开始对 WAL 记录进行逻辑解码 - pgoutput 插件转换从 WAL 读取的更改。
pgoutput 插件检查数据是否是模式发布的一部分。如果是，则使用流复制协议将数据连续传输到 Apply Worker。
Apply Worker 将数据映射到本地表，并在收到更改时以正确的事务顺序应用各个更改。
您可以在PostgreSQL git 存储库中查看实现的更多详细信息。

https://git.postgresql.org/gitweb/p=postgresql.git;a=commitdiff;h=5a2832465fd8984d089e8c44c094e6900d987fcd

语法的变化

新语法允许在创建或更改发布时包括指定模式中的所有表。

CREATE PUBLICATION pub1 FOR ALL TABLES IN SCHEMA sch1,sch2;

或

ALTER PUBLICATION pub1 ADD ALL TABLES IN SCHEMA sch3,sch4;

除了指定模式之外，新语法还允许在创建或更改发布时指定单个表。

CREATE PUBLICATION pub1 FOR ALL TABLES IN SCHEMA sch1,sch2, TABLE tbl1, tbl2;

或

ALTER PUBLICATION pub1 ADD ALL TABLES IN SCHEMA sch3,sch4, TABLE tbl3, tbl4;

请注意，将模式添加到已订阅的发布将需要在订阅端执行ALTER SUBSCRIPTION ... REFRESH PUBLICATION操作才能生效。

新系统表 pg_publication_namespace

将添加一个新的系统表 pg_publication_namespace，以维护用户为发布指定的模式。

用户可以使用 pg_publication_namespace 和 pg_publication 来获取与发布的模式映射，如下例所示。

postgres=# CREATE PUBLICATION pub1 FOR ALL TABLES IN SCHEMA sch1,sch2;
CREATE PUBLICATION
postgres=# SELECT pubname, pnnspid::regnamespace
postgres-# FROM pg_publication_namespace pn, pg_publication p
postgres-# WHERE pn.pnpubid = p.oid;
pubname | pnnspid
---------+------
pub1 | sch1
pub1 | sch2
（2 rows）

对 pgoutput 的更改

修改了 pgoutput 插件以检查表是否是模式发布的一部分并将更改发布给订阅者。

发布描述

\d命令系列已更新为显示模式发布，而\dRp+ 命令将显示与发布关联的模式。

postgres=# CREATE PUBLICATION pub1 FOR ALL TABLES IN SCHEMA sch1,sch2;
CREATE PUBLICATION
postgres=# \dRp+ pub1

Publication pub1
Owner | All tables | Inserts | Updates | Deletes | Truncates | Via root
------+------------+---------+---------+---------+-----------+----------
 Dba  | f          | t       | t       | t       | t         | f

Tables from schemas:

对 pg_dump 的更改

pg_dump 客户端工具已更新，用以识别是否创建发布以发布模式中的所有表并转储发布的 ddl，包括ALL TABLES IN SCHEMA选项。

pg_dump 生成的 DDL 示例：

-- 
-- Name: pub1; Type: PUBLICATION; Schema: -; Owner: dba
--
CREATE PUBLICATION pub1 WITH (publish = 'insert, update, delete, truncate');
ALTER PUBLICATION pub1 OWNER TO dba;
--
-- Name: pub1 sch1; Type: PUBLICATION TABLES IN SCHEMA; Schema: sch1; Owner: dba
--
ALTER PUBLICATION pub1 ADD ALL TABLES IN SCHEMA sch1;
--
-- Name: pub1 sch2; Type: PUBLICATION TABLES IN SCHEMA; Schema: sch2; Owner: dba
--
ALTER PUBLICATION pub1 ADD ALL TABLES IN SCHEMA sch2;

对 psql 的更改

psql 客户端工具已更新，可以支持ALL TABLES IN SCHEMA选项的制表符补全。

如何在 SCHEMA数据中发布所有表？

以下是使用新选项ALL TABLES IN SCHEMA的步骤。

在发布者和订阅者中创建一些模式和表：

postgres=# CREATE SCHEMA sch1;
CREATE SCHEMA
postgres=# CREATE TABLE sch1.tbl1 (col1 int);
CREATE TABLE
postgres=# CREATE TABLE sch1.tbl2 (col1 int);
CREATE TABLE
postgres=# CREATE SCHEMA sch2;
CREATE SCHEMA
postgres=# CREATE TABLE sch2.tbl3 (col1 int);
CREATE TABLE
postgres=# CREATE TABLE sch2.tbl4 (col1 int);
CREATE TABLE

2.在发布者中创建发布：

postgres=# CREATE PUBLICATION pub1 FOR ALL TABLES IN SCHEMA sch1,sch2;
CREATE PUBLICATION

3.通过指定发布者主机和发布者端口 6666 在订阅者中创建订阅：

postgres=# CREATE SUBSCRIPTION sub1 CONNECTION 'host=host1 
dbname=postgres port=6666' PUBLICATION pub1;
NOTICE: created replication slot "sub1" on publisher
CREATE SUBSCRIPTION

4.将一些数据插入发布者：

postgres=#INSERT INTO sch1.tbl1 VALUES(11);     
INSERT 0 1 
postgres=# INSERT INTO sch1.tbl2 VALUES(12); 
INSERT 0 1 
postgres=# INSERT INTO sch2.tbl3 VALUES(23); 
INSERT 0 1 
postgres=# INSERT INTO sch2.tbl4 VALUES(24); 
INSERT 0 1

5.检查发布者发布的数据是否在逻辑上复制到订阅者：

postgres=# SELECT * FROM sch1.tbl1;      
col1 
---- 
11 
(1 rows) 
postgres=# SELECT * FROM sch1.tbl2; 
col1 
---- 
12 
(1 rows) 
postgres=# SELECT * FROM sch2.tbl3; 
col1 
---- 
23 
(1 rows) 
postgres=# SELECT * FROM sch2.tbl4; 
col1 
---- 
24 
(1 rows)