在“逻辑复制入门”中，我们建立了发布者和订阅者之间的基本工作复制，并了解了基本概念。在本文中，我们将扩展逻辑复制的运营管理、监控等实际操作，并深入探讨逻辑解码的基础知识。
初始数据副本
正如我们在第一部分中演示的那样，在设置订阅服务器时，你可以使用 选项选择（或不选择）依赖初始数据副本WITH (copy_data = false)。虽然默认副本是一个非常有用的行为，但在生产环境中使用它之前，你应该了解此默认行为的一些特性。
该机制有效地要求发布者通过拍摄快照（由 MVCC 提供）来复制表数据，并将其发送给订阅者，并且由于复制槽“书签”，可以从拍摄快照的点无缝地继续流式传输更改。
简单是这里的关键特性，因为单个命令可以处理快照、传输和过渡到持续流。
您所做的权衡是在性能方面，这完全是由于它每个表使用单个进程这一事实。虽然它对于测试表几乎可以立即生效，但在处理包含 GB 数据的表时，您会遇到明显的延迟和开销。
虽然可以通过配置参数控制并行度max_sync_workers_per_subscription，但实际数据库的复制过程可能仍需要等待数小时（甚至数天）才能完成。您可以使用pg_subscription_rel目录来监控表是否已同步，或者仍在等待/进行中。
SELECT srrelid::regclass AS table_name, srsubstate
FROM pg_subscription_rel;
每个表将具有以下状态之一：
- i尚未开始
- d复制正在进行中
- s同步（或等待确认）
- r完成并复制
幸运的是，即使并非所有表都已同步，状态r指示更改流也可以开始。然而，复制槽会保留 LSN 位置，这意味着发布者将保留 WAL 文件，直到订阅者跟上进度。
生产工作负载的手动同步
如上所述，隐式复制可能会带来性能上的牺牲，这对于生产环境来说过于严重，除非您在设计逻辑复制拓扑时考虑到这一点。在所有其他情况下，手动同步才是最佳选择。
整个过程有一个主要原则：手动加载的数据必须与逻辑复制流开始的时间点保持一致。实现这一点的方法是，在数据恢复点 (PITR) 之前创建一个逻辑复制槽（或与之同步），并在数据传输完成后启用该槽。只有这样，您才能在订阅服务器上正确应用所有后续更改。
有几种方法可以实现这一点，您必须根据特定用例中的可用约束来评估机制：
- 如果需要同步的数据量接近备份数据的总大小（即大多数表），请使用支持时间点恢复（PITR）的备份和恢复机制。
- 使用备份来协调发布服务器上的数据变更提取，方法是停止传入的变更（例如，在计划维护时段内），并且仅在已知 LSN 有可用的一致快照时才恢复这些变更。这适用于您可以控制表提取并且传输的数据量适合维护时段的情况。
- 如果以上方法均不可用，您可以手动将复制槽推进到预定义的 LSN。请理解，这应该是最后的手段，因为它可能被视为专家领域。
最可靠的备份和恢复处理推荐使用pgBackRest，因为它允许您提前恢复特定备份，并仅应用稍后需要的更改以推进到选定的时间点，从而显著减少初始同步所需的时间。另一方面，如果您在订阅服务器上仅使用备份中的一小部分数据，则可能会造成设置不可接受的资源限制。
虽然pg_dump 不能被视为可靠的备份工具，但如果您正在讨论使用创建的“同步快照”对一小部分表进行一致的备份，它可能会对您有所帮助pg_export_snapshot。
pg_dump -d publisher_db --snapshot="000004A2-1" --data-only -Fc -f initial_data.dump
请注意，一致性备份和时间点恢复的主题远远超出了本逻辑复制指南的范围。
监控逻辑复制
一旦您设置了发布者和订阅者并准备好初始数据，就该开始考虑如何保持该流程的运行。
对于日常运营来说，监控的基本构建块是pg_replication_slots目录。示例查询：
SELECT
slot_name,
plugin,
slot_type,
active,
pg_size_pretty(
pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)
) AS wal_retained_size
FROM pg_replication_slots
WHERE
slot_type = 'logical';
结果如下：
-[ RECORD 1 ]-----+-----------------------------
slot_name | my_sample_subscription
plugin | pgoutput
slot_type | logical
active | t
wal_retained_size | 49 MB
您应该寻找两件事：
- active指示该槽位是否正在被任何连接使用。虽然这在较短的持续时间内是可以接受的，但它可能表示用户已关闭或配置不正确。
- 计算值wal_retained_size是一项关键指标。值增加表示订阅服务器使用更改时出现问题，并且您的集群可能面临磁盘空间不足的风险。
这两个值都很重要，因为它们可能指示需要注意的问题。active标志可能指示悬空槽，即订阅者被销毁（或被擦除）而没有正确删除订阅，从而留下了复制槽。在这种情况下，唯一的直接响应是按槽名删除该槽：
SELECT pg_drop_replication_slot('my_dangling_slot');
如果您想深入挖掘，您还可以使用pg_stat_replication它来提供有关复制状态的更多（尽管不稳定）数据。
为了完成监控部分，您可以设计自定义检查或使用上述查询的修改版本：
SELECT
slot_name,
CASE WHEN active
THEN 1
ELSE 0
END as active_status,
pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn) as wal_bytes_retained,
extract(epoch from now()) as time
FROM pg_replication_slots
WHERE slot_type = 'logical';
作为 Grafana 警报规则的一部分：
Alert Rule: "Inactive Replication Slot"
Condition:
- Query: active_status
- Reducer: last()
- Evaluator: below 1
- For: 30m # if inactive for 30+ minutes
Additional conditions (AND):
- Query: wal_bytes_retained
- Reducer: last()
- Evaluator: above 1073741824 # 1GB limit
尽管如此，您仍需要评估适用于您特定用例的监控细节。可能还有其他选项，例如，如果您的架构允许，可以使用max_slot_wal_keep_size并将复制槽标记为无效，并在特定槽的大小低于配置值（例如 100GB）时释放 WAL 文件。虽然这听起来合理，但您必须考虑您的应用程序或用例能够从此类数据丢失中恢复的能力。
动态变化的发布配置
回到我们逻辑复制之旅的第一部分，需要注意一个重要的区别。虽然我们描述了初始用例，但在实际应用中，发布者的架构会随着时间的推移而发生变化。表将被隐式创建和添加（通过FOR ALL TABLES），或显式添加/删除。
ALTER PUBLICATION my_publication ADD TABLE public.new_table;
ALTER PUBLICATION my_publication DROP TABLE public.old_table;
但是，当您添加新表时会发生什么？对于现有的订阅，答案是几乎不会发生。订阅者反映底层配置更改的唯一方法是刷新订阅。在此过程中，PostgreSQL 会将当前订阅表列表与发布表列表进行比较（如果已配置），然后启动初始同步过程，并在完成后应用流式更改。
如上所述，您可以刷新订阅，无论是否复制初始数据。此copy_data选项控制此行为，并且是目前唯一支持的设置。
ALTER SUBSCRIPTION my_subscription
REFRESH PUBLICATION;
ALTER SUBSCRIPTION my_subscription
REFRESH PUBLICATION WITH (copy_data = false);
当您从发布者中删除一个表时，情况会变得简单得多，因为该特定表的更改不再被视为逻辑解码。
但是，如果您在不刷新订阅的情况下向发布中添加新表，会发生什么情况？尽管 WAL 文件包含所有表的更改，并且逻辑解码过程将处理所有已发布的表更改，但无需考虑额外的 WAL 保留。订阅者将confirmed_flush_lsn像以前一样继续操作，因为新添加表的初始状态尚未记录，并且流式传输的更改（暂时）将被忽略。
逻辑解码
如果您在整个过程中一直关注，或者已经对逻辑复制有所了解，那么您可能已经遇到过一个叫做pgoutput插件的东西。这是一个内置的默认机制，负责逻辑解码——一种遍历 WAL 流并将其转换为更高级格式的机制。
例如，逻辑解码不是进行物理复制（即页面 Z 中偏移量 Y 处的字节 X），而是将 WAL 流转换为行级更改。它利用源数据库的上下文感知来理解更改，并从对象标识符到模式、表和列名，以及相应的值（包括新旧值）。它还提供了一种将更改组装到正确的事务流中的方法。
该插件的作用是以特定方式格式化输出，并且本身并不知道：
- 任何传输层细节
- 复制槽
- 任何重试逻辑
PostgreSQL 提供了一个强大的逻辑解码框架。内置插件只是一个开始，开发人员可以创建自定义输出插件。不仅如此，我们还可以手动查看逻辑解码流的内部结构。
让我们考虑这个例子：
-- create new publication
CREATE PUBLICATION all_data FOR ALL TABLES;
-- create new replication slot with pgoutput plugin
SELECT pg_create_logical_replication_slot('plugin_demo', 'pgoutput');
-- sample table
CREATE TABLE IF NOT EXISTS demo_table (
id INT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
name TEXT,
email TEXT
);
-- generate some changes with insert/update/delete
INSERT INTO demo_table (name, email) VALUES ('John Doe', 'john@example.com');
UPDATE demo_table SET email = 'doe@example.com' WHERE id = 1;
DELETE FROM demo_table WHERE id = 1;
-- get binary changes
SELECT * FROM pg_logical_slot_get_binary_changes('plugin_demo', NULL, NULL, 'proto_version', '1', 'publication_names', 'all_data');
-- drop the replication slot and the publication
SELECT pg_drop_replication_slot('plugin_demo');
DROP TABLE demo_table;
DROP PUBLICATION all_data;
让我们看一下最近执行的更改（输出缩短）：
lsn | xid | data
------------+------+---------------------------------
1/D6997380 | 1038 | \x4200000001d69974180
1/D6997380 | 1038 | \x52000042057075626c696300646...
1/D6997380 | 1038 | \x49000042054e000374000...
1/D6997448 | 1038 | \x430000000001d6997...
1/D6997448 | 1039 | \x4200000001d6997...
1/D6997448 | 1039 | \x49000042054e0....
1/D6997510 | 1039 | \x430000000001d6...
这没错，但并不能帮助我们演示流程。别忘了pgoutput，实际使用的插件是逻辑解码的，而且是二进制的。不过别担心，PostgreSQL 提供了一个test_decoding插件，可以将更改解码成人类可读的文本表示。
-- setup the publication
CREATE PUBLICATION all_data FOR ALL TABLES;
-- create replication slot with test_decoding plugin
SELECT pg_create_logical_replication_slot('plugin_demo', 'test_decoding');
CREATE TABLE IF NOT EXISTS demo_table (
id INT GENERATED ALWAYS AS IDENTITY PRIMARY KEY,
name TEXT,
email TEXT
);
-- generate some changes with insert/update/delete
INSERT INTO demo_table (name, email) VALUES ('John Doe', 'john@example.com');
UPDATE demo_table SET email = 'doe@example.com' WHERE id = 1;
DELETE FROM demo_table WHERE id = 1;
-- get human readable changes
SELECT * FROM pg_logical_slot_get_changes('plugin_demo', NULL, NULL);
-- drop the replication slot and the publication
SELECT pg_drop_replication_slot('plugin_demo');
DROP TABLE demo_table;
DROP PUBLICATION all_data;
这次的输出实际上是有帮助的。
lsn | xid | data
------------+------+-----------------------------------------------------------------------------------------------------
1/D69CCC30 | 1048 | BEGIN 1048
1/D69CCC98 | 1048 | table public.demo_table: INSERT: id[integer]:1 name[text]:'John Doe' email[text]:'john@example.com'
1/D69CCDC0 | 1048 | COMMIT 1048
1/D69CCDC0 | 1049 | BEGIN 1049
1/D69CCDC0 | 1049 | table public.demo_table: UPDATE: id[integer]:1 name[text]:'John Doe' email[text]:'doe@example.com'
1/D69CCE50 | 1049 | COMMIT 1049
1/D69CCE50 | 1050 | BEGIN 1050
1/D69CCE50 | 1050 | table public.demo_table: DELETE: id[integer]:1
1/D69CCEC0 | 1050 | COMMIT 1050
只要阅读它，您就可以轻松地跟踪事务的顺序（psql 中的自动提交）并跟踪单个更改。
我们还可以通过此示例重申 的重要性REPLICA IDENTITY。如果您仔细检查 的 DDL demo_table，您会看到IDENTITYused ，根据其定义，它将创建一个主键。使用默认副本标识，您实际上依赖于它作为识别数据的主要来源。您可以使用该test_decoding插件来演示它将创建的详细程度。
（在此示例中，保留设置和拆卸。）
ALTER TABLE demo_table REPLICA IDENTITY FULL;
INSERT INTO demo_table (id, name, email) OVERRIDING SYSTEM VALUE VALUES (999, 'John Doe', 'john@example.com');
UPDATE demo_table SET email = 'doe@example.com' WHERE id = 999;
DELETE FROM demo_table WHERE id = 999;
SELECT * FROM pg_logical_slot_get_changes('plugin_demo', NULL, NULL);
您可以观察到变化。虽然本质上INSERT输出相同（所有值均已发送），UPDATE但DELETE不再依赖主键。相反，它们必须提供所有值，包括UPDATE新旧数据：
lsn | 1/D69F8770
xid | 1062
data | table public.demo_table: UPDATE: old-key: id[integer]:999 name[text]:'John Doe' email[text]:'john@example.com' new-tuple: id[integer]:999 name[text]:'John Doe' email[text]:'doe@exa
mple.com'
以及DELETE旧数据。
lsn | 1/D69F8828
xid | 1063
data | table public.demo_table: DELETE: id[integer]:999 name[text]:'John Doe' email[text]:'doe@example.com'
显而易见的是，错误配置的副本身份可能会增加解码和复制的数据量。
细粒度的发布控制
在本系列后续部分介绍高级复制拓扑之前，我们先来了解如何精确控制要复制到订阅服务器的数据。细粒度的控制有助于控制复制开销、减少网络流量、增强安全性，并确保订阅服务器只接收所需的数据。
第一个过滤数据的选项是显式过滤column_list，允许您排除敏感数据（PII 或类似信息）或不必要的数据。选择列时，您应该包含主键或副本身份后面的列；如果没有，PostgreSQL 会自动添加它们。
CREATE PUBLICATION hr_analytics
FOR TABLE hr.employees (employee_id, first_name, last_name, department, start_date);
与列选择类似，您可以控制发布复制的操作。可用选项包括“插入”、“更新”、“删除”和“截断”。
CREATE PUBLICATION hr_analytics
FOR TABLE hr.employees (employee_id, first_name, last_name, department, start_date)
WITH (publish = 'insert,delete');
确保逻辑复制的安全
在上一节中，我们讨论了逻辑复制的一个关键要素——限制对发布的访问权限。否则，如果其他发布仍然可以不受限制地提供字段，那么限制已发布的字段又有何意义呢？到目前为止，在本系列文章中，我们一直依靠访问权限来简化SUPERUSER发布内容的创建和管理。要设计用于实际生产环境的逻辑复制，您需要一个最小权限模型。
幸运的是，PostgreSQL 逻辑复制的安全模型遵循与常规查询相同的规则。如果用户无法从特定表、特定列或特定行执行 SELECT 操作，那么他们也无法将其包含在发布中。因此，本节假设您已经熟悉 PostgreSQL 权限模型。
所需权限的总体概述是，复制用户必须拥有CONNECT数据库、USAGE架构以及SELECT特定表（列）的权限。您还需要确保（如果适用）行级安全性 (RLS) 得到正确管理。
使用上面使用的细粒度示例，我们可以演示设置过程。首先，创建一个带有LOGIN和REPLICATION子句的角色，并授予其数据库和架构访问权限：
CREATE ROLE hr_analytics_role WITH LOGIN REPLICATION PASSWORD 'a_strong_password';
GRANT CONNECT ON DATABASE my_publisher_db TO hr_analytics_role;
GRANT USAGE ON SCHEMA hr TO hr_analytics_role;
关键的一步是正确定义GRANT选择数据。
GRANT SELECT
(employee_id, first_name, last_name, department, hire_date)
ON hr.employees TO hr_analytics_role;
这种方法确保即使有人获得了复制角色的访问权限，他们也无法暴露超出明确授予权限的数据。发布将遵循这些列级限制，为复制数据创建安全边界。
结论
今天，我们已经超越了逻辑复制的基础知识，涵盖了生产环境的基本实践，并扩展了对其实际工作原理的理解。
本文是即将发布的《精通 PostgreSQL 中的逻辑复制》指南的第二部分。如果您对该主题感兴趣，请考虑订阅以获取最新发布的文章。
原文作者：Radim Marek
原文地址：https://boringsql.com/posts/logical-replication-beyond-the-basics/
