postgresql - 手动创建用于在 PostgreSQL 10 中发布的复制槽

Question

我正在尝试从我的 PostgreSQL 数据库中获取某些表的更新流。获取所有更新的常规方式如下所示:

您创建一个逻辑复制槽

pg_create_logical_replication_slot('my_slot', 'wal2json');

然后使用 pg_recvlogical 连接到它或进行特殊的 SQL 查询。这允许您从 json 中的数据库中获取所有操作(如果您使用了 wal2json 插件或类似插件)，然后使用该数据做任何您想做的事情。

但是在 PostgreSQL 10 中，我们有发布/订阅机制，允许我们只复制选定的表。这非常方便，因为没有发送很多无用的数据。该过程如下所示:

首先，您创建一个出版物

CREATE PUBLICATION foo FOR TABLE herp, derp;

然后您从另一个数据库订阅该出版物

CREATE SUBSCRIPTION mysub CONNECTION <connection stuff> PUBLICATION foo;

这在引擎盖下的主数据库上创建了一个复制槽，并开始监听更新并将它们提交到第二个数据库上的相同表。如果您的工作是复制一些表，但想为我的东西获取原始流，这很好。

正如我提到的，CREATE SUBSCRIPTION 查询在幕后在 master 数据库上创建了一个复制槽，但是我如何在没有订阅和第二个数据库的情况下手动创建一个复制槽？ Here文档说:

To make this work, create the replication slot separately (using the function pg_create_logical_replication_slot with the plugin name pgoutput)

根据文档，这是可能的，但 pg_create_logical_replication_slot 仅创建一个常规复制槽。 pgoutput 插件负责所有的魔法吗？如果是，那么就不可能将其他插件(如 wal2json)与出版物一起使用。

我在这里错过了什么？

Answer 1

我对 Postgres 中的逻辑复制和逻辑解码经验有限，所以如果以下内容有误，请指正。话虽这么说，但这是我的发现:

1. 发布支持由 pgoutput 插件提供。您通过 plugin-specific options 使用它.可能其他插件有可能添加支持，但我不知道逻辑解码插件接口(interface)是否暴露了足够的细节。我针对 wal2json 插件版本 9e962ba 对其进行了测试，但它无法识别此选项。
2. 复制槽的创建独立于发布。在获取更改流时指定要用作过滤器的发布。可以查看一份出版物的更改，然后查看另一份出版物的更改，并观察不同的更改集，尽管使用相同的复制槽(我没有找到它的文档，我正在测试具有 Postgres 兼容性的 Aurora，因此行为可能会有所不同).
3. 插件输出似乎包括开始和提交的所有条目，即使事务没有触及感兴趣的发布中包含的任何表。但是，它不包括对出版物中包含的其他表格的更改。

这是一个如何在 Postgres 10+ 中使用它的例子:

-- Create publication
CREATE PUBLICATION cdc;

-- Create slot
SELECT pg_create_logical_replication_slot('test_slot_v1', 'pgoutput');

-- Create example table
CREATE TABLE replication_test_v1
(
  id integer NOT NULL PRIMARY KEY,
  name text
);

-- Add table to publication
ALTER PUBLICATION cdc ADD TABLE replication_test_v1;

-- Insert example data
INSERT INTO replication_test_v1(id, name) VALUES
  (1, 'Number 1')
;

-- Peak changes (does not consume changes)
SELECT pg_logical_slot_peek_binary_changes('test_slot_v1', NULL, NULL, 'publication_names', 'cdc', 'proto_version', '1');

-- Get changes (consumes changes)
SELECT pg_logical_slot_get_binary_changes('test_slot_v1', NULL, NULL, 'publication_names', 'cdc', 'proto_version', '1');

要将更改从 Postgres 流式传输到其他系统，您可以考虑使用 Debezium项目。它是一个用于更改数据捕获的开源分布式平台，其中包括一个 PostgreSQL 连接器。在 0.10 版中，他们添加了对 pgoutput 插件的支持。即使您的用例与项目提供的非常不同，您也可以查看他们的代码以了解他们如何与复制 API 交互。