MaterializedPostgreSQL - ClickHouse Documentation

建议 ClickHouse Cloud 用户使用 ClickPipes 将 PostgreSQL 复制到 ClickHouse。该方案原生支持 PostgreSQL 的高性能 CDC (变更数据捕获) 。

创建一个 ClickHouse 数据库，其中包含来自 PostgreSQL 数据库的表。首先，使用 MaterializedPostgreSQL 引擎的数据库会为 PostgreSQL 数据库创建快照，并加载所需的表。所需表可以是指定数据库中任意 schema 子集下的任意表子集。创建快照的同时，数据库引擎还会获取 LSN；在完成表的初始转储后，便开始从 WAL 拉取更新。数据库创建完成后，之后新增到 PostgreSQL 数据库中的表不会自动加入复制，必须使用 ATTACH TABLE db.table 查询手动添加。复制基于 PostgreSQL Logical Replication Protocol 实现。该协议不支持复制 DDL，但能够识别是否发生了会破坏复制的变更 (如列类型变更、添加/删除列) 。检测到此类变更后，对应表将停止接收更新。此时，应使用 ATTACH/ DETACH PERMANENTLY 查询重新完整加载该表。如果 DDL 不会破坏复制 (例如重命名列) ，表仍会继续接收更新 (insert 按位置执行) 。

该数据库引擎属于 Experimental。要使用它，请在配置文件中将 allow_experimental_database_materialized_postgresql 设为 1，或使用 SET 命令：

SET allow_experimental_database_materialized_postgresql=1

创建数据库

CREATE DATABASE [IF NOT EXISTS] db_name [ON CLUSTER cluster]
ENGINE = MaterializedPostgreSQL('host:port', 'database', 'user', 'password') [SETTINGS ...]

引擎参数

host:port — PostgreSQL 服务器的端点。
database — PostgreSQL 数据库名。
user — PostgreSQL 用户。
password — 用户密码。

使用示例

CREATE DATABASE postgres_db
ENGINE = MaterializedPostgreSQL('postgres1:5432', 'postgres_database', 'postgres_user', 'postgres_password');

SHOW TABLES FROM postgres_db;

┌─name───┐
│ table1 │
└────────┘

SELECT * FROM postgres_db.postgres_table;

动态将新表添加到复制范围中

创建 MaterializedPostgreSQL 数据库后，它不会自动检测对应 PostgreSQL 数据库中的新表。此类表可以手动添加：

ATTACH TABLE postgres_database.new_table;

在 22.1 之前的版本中，将表加入复制时会遗留一个未删除的临时 replication slot (名为 {db_name}_ch_replication_slot_tmp) 。如果在 22.1 之前的 ClickHouse 版本中 Attach 表，请务必手动删除该 slot (SELECT pg_drop_replication_slot('{db_name}_ch_replication_slot_tmp')) 。否则，磁盘占用会持续增长。此问题已在 22.1 中修复。

动态将表从复制中移除

可以将特定表从复制中移除：

DETACH TABLE postgres_database.table_to_remove PERMANENTLY;

PostgreSQL schema

PostgreSQL schema 可通过 3 种方式进行配置 (自 21.12 版本起) 。

一个 MaterializedPostgreSQL 数据库引擎对应一个 schema。需要使用设置 materialized_postgresql_schema。表只通过表名访问：

CREATE DATABASE postgres_database
ENGINE = MaterializedPostgreSQL('postgres1:5432', 'postgres_database', 'postgres_user', 'postgres_password')
SETTINGS materialized_postgresql_schema = 'postgres_schema';

SELECT * FROM postgres_database.table1;

对于一个 MaterializedPostgreSQL 数据库引擎，可以指定任意数量的 schema 及其对应的一组表。需要使用设置 materialized_postgresql_tables_list。每个表都必须连同其所属 schema 一起写出。访问表时需要同时使用 schema 名称和表名：

CREATE DATABASE database1
ENGINE = MaterializedPostgreSQL('postgres1:5432', 'postgres_database', 'postgres_user', 'postgres_password')
SETTINGS materialized_postgresql_tables_list = 'schema1.table1,schema2.table2,schema1.table3',
         materialized_postgresql_tables_list_with_schema = 1;

SELECT * FROM database1.`schema1.table1`;
SELECT * FROM database1.`schema2.table2`;

但在这种情况下，materialized_postgresql_tables_list 中的所有表都必须写明其 schema 名称。需要设置 materialized_postgresql_tables_list_with_schema = 1。警告：在这种情况下，表名中不允许包含点号。

对于单个 MaterializedPostgreSQL 数据库引擎，可以指定任意数量的 schema 及其完整的表集。需要使用设置 materialized_postgresql_schema_list。

CREATE DATABASE database1
ENGINE = MaterializedPostgreSQL('postgres1:5432', 'postgres_database', 'postgres_user', 'postgres_password')
SETTINGS materialized_postgresql_schema_list = 'schema1,schema2,schema3';

SELECT * FROM database1.`schema1.table1`;
SELECT * FROM database1.`schema1.table2`;
SELECT * FROM database1.`schema2.table2`;

警告：在这种情况下，表名中不能包含点号。

要求

在 PostgreSQL 配置文件中，wal_level 设置的值必须为 logical，并且 max_replication_slots 参数的值必须至少为 2。
每个被复制的表都必须具有以下副本标识之一：

主键 (默认)
索引

postgres# CREATE TABLE postgres_table (a Integer NOT NULL, b Integer, c Integer NOT NULL, d Integer, e Integer NOT NULL);
postgres# CREATE unique INDEX postgres_table_index on postgres_table(a, c, e);
postgres# ALTER TABLE postgres_table REPLICA IDENTITY USING INDEX postgres_table_index;

始终会先检查主键。如果主键不存在，则会检查被定义为副本标识索引的索引。如果某个索引被用作副本标识，则一张表中只能有一个这样的索引。你可以使用以下命令查看特定表使用的是哪种类型：

postgres# SELECT CASE relreplident
          WHEN 'd' THEN 'default'
          WHEN 'n' THEN 'nothing'
          WHEN 'f' THEN 'full'
          WHEN 'i' THEN 'index'
       END AS replica_identity
FROM pg_class
WHERE oid = 'postgres_table'::regclass;

不支持 TOAST 值的复制。将使用该数据类型的默认值。

设置

`materialized_postgresql_tables_list`

设置一个以逗号分隔的 PostgreSQL 数据库表列表，这些表将通过 MaterializedPostgreSQL 数据库引擎进行复制。每个表都可以在括号中指定要复制的列子集。如果省略该列子集，则复制该表的所有列。

materialized_postgresql_tables_list = 'table1(co1, col2),table2,table3(co3, col5, col7)

默认值：空列表——表示将复制整个 PostgreSQL 数据库。

`materialized_postgresql_schema`

默认值：空字符串。 (使用默认 schema)

`materialized_postgresql_schema_list`

默认值：空列表。 (将使用默认 schema)

`materialized_postgresql_max_block_size`

设置在将数据刷新到 PostgreSQL 数据库表之前，可在内存中收集的行数。可能的值：

正整数。

默认值：65536。

`materialized_postgresql_replication_slot`

用户创建的 replication slot。必须与 materialized_postgresql_snapshot 搭配使用。

`materialized_postgresql_snapshot`

用于标识快照的文本字符串，PostgreSQL 表的初始转储将基于该快照执行。必须与 materialized_postgresql_replication_slot 一起使用。

CREATE DATABASE database1
ENGINE = MaterializedPostgreSQL('postgres1:5432', 'postgres_database', 'postgres_user', 'postgres_password')
SETTINGS materialized_postgresql_tables_list = 'table1,table2,table3';

SELECT * FROM database1.table1;

如有需要，可以使用 DDL 查询来更改这些设置。但设置 materialized_postgresql_tables_list 无法更改。要更新该设置中的表列表，请使用 ATTACH TABLE 查询。

ALTER DATABASE postgres_database MODIFY SETTING materialized_postgresql_max_block_size = <new_size>;

`materialized_postgresql_use_unique_replication_consumer_identifier`

在复制中使用唯一的复制消费者标识符。默认值：0。如果设置为 1，则允许将多个 MaterializedPostgreSQL 表配置为指向同一个 PostgreSQL 表。

`materialized_postgresql_use_extended_date_and_time_types`

将 PostgreSQL 的 date 和 timestamp/timestamptz 类型映射到 ClickHouse 的 Date32 和 DateTime64，以覆盖 PostgreSQL 这些类型更宽的取值范围。默认值：1。如果将其设为 0，则改用范围更窄的 Date 和 DateTime 类型 (超出其范围的值，或带有子秒级精度的值，将无法表示) 。此设置仅控制创建嵌套表时，类型推断所选择的列类型，因此必须在 CREATE DATABASE 时指定。之后无法通过 ALTER DATABASE ... MODIFY SETTING 修改它 (已创建的嵌套表会保留其固定的列类型，因此此类更改会被拒绝) ；如需更改，只能重新创建数据库。它不适用于 MaterializedPostgreSQL 表引擎，因为该引擎中的列类型是显式声明的。

注意事项

logical replication slot 的故障转移

存在于主节点上的 logical replication slot 在 standby 副本上不可用。因此，如果发生故障转移，新的主节点 (原来的物理 standby) 将不知道旧主节点上存在哪些 slot。这会导致 PostgreSQL 复制中断。解决办法之一是自行管理 replication slot，并定义一个永久 replication slot (可在这里找到一些相关信息) 。你需要通过 materialized_postgresql_replication_slot 设置传入 slot 名称，并且该 slot 必须使用 EXPORT SNAPSHOT 选项导出。快照标识符则需要通过 materialized_postgresql_snapshot 设置传入。请注意，只有在确实需要时才应使用这种方式。如果并无实际需求，或者不完全清楚原因，最好让表引擎自行创建和管理 replication slot。 示例 (来自 @bchrobot)

在 PostgreSQL 中配置 replication slot。

apiVersion: "acid.zalan.do/v1"
kind: postgresql
metadata:
  name: acid-demo-cluster
spec:
  numberOfInstances: 2
  postgresql:
    parameters:
      wal_level: logical
  patroni:
    slots:
      clickhouse_sync:
        type: logical
        database: demodb
        plugin: pgoutput

等待 replication slot 就绪，然后开始一个事务并导出该事务的快照标识符：
BEGIN; SELECT pg_export_snapshot();

在 ClickHouse 中创建数据库：

CREATE DATABASE demodb
ENGINE = MaterializedPostgreSQL('postgres1:5432', 'postgres_database', 'postgres_user', 'postgres_password')
SETTINGS
  materialized_postgresql_replication_slot = 'clickhouse_sync',
  materialized_postgresql_snapshot = '0000000A-0000023F-3',
  materialized_postgresql_tables_list = 'table1,table2,table3';

确认已复制到 ClickHouse DB 后，结束 PostgreSQL 事务。验证在故障转移后复制是否仍会继续：

kubectl exec acid-demo-cluster-0 -c postgres -- su postgres -c 'patronictl failover --candidate acid-demo-cluster-1 --force'

所需权限

CREATE PUBLICATION — CREATE 查询权限。
CREATE_REPLICATION_SLOT — 复制权限。
pg_drop_replication_slot — 复制权限或超级用户权限。
DROP PUBLICATION — publication 的所有者 (即 MaterializedPostgreSQL engine 自身中的 username) 。

也可以不执行 2 和 3 命令，从而无需具备这些权限。使用设置 materialized_postgresql_replication_slot 和 materialized_postgresql_snapshot 即可。但务必格外谨慎。对以下表的访问权限：

pg_publication
pg_replication_slots
pg_publication_tables

备份和恢复

MaterializedPostgreSQL 数据库可以备份。每个被复制的表的数据都存放在一个嵌套的 ReplacingMergeTree 表中，因此 BACKUP DATABASE 会通过委托给该嵌套表来备份这些数据。

BACKUP DATABASE postgres_db TO Disk('backups', 'postgres_db.zip');

不支持将 MaterializedPostgreSQL database 或表原地 restore。恢复后的 MaterializedPostgreSQL object 会立即开始从在线 PostgreSQL source 复制，因此如果在其上恢复 backup snapshot，就会把该 snapshot 与当前远端 state 混在一起。因此，在这种情况下 RESTORE 会以失败并终止的方式安全关闭。请改为将已 capture 的数据恢复到普通的 ReplacingMergeTree 表中：

对于 database backup，每个表存储的定义本身已经是合成的嵌套 ReplacingMergeTree (而不是 MaterializedPostgreSQL engine) ，因此每个表都可以直接恢复到一个新的、尚不存在的表中：
RESTORE TABLE postgres_db.table1 AS restored_db.table1 FROM Disk('backups', 'postgres_db.zip') SETTINGS allow_different_table_def = 1;
对于独立的 MaterializedPostgreSQL 表 backup，存储的定义是 MaterializedPostgreSQL engine 本身。请预先创建一个 ReplacingMergeTree 表，使其结构与嵌套表相同 (包括 _sign 和 _version 列) ，然后将数据恢复到该表中：
RESTORE TABLE src AS existing_replacing_mergetree FROM Disk('backups', 'table.zip') SETTINGS allow_different_table_def = 1;

​创建数据库

​使用示例

​动态将新表添加到复制范围中

​动态将表从复制中移除

​PostgreSQL schema

​要求

​设置

​materialized_postgresql_tables_list

​materialized_postgresql_schema

​materialized_postgresql_schema_list

​materialized_postgresql_max_block_size

​materialized_postgresql_replication_slot

​materialized_postgresql_snapshot

​materialized_postgresql_use_unique_replication_consumer_identifier

​materialized_postgresql_use_extended_date_and_time_types

​注意事项

​logical replication slot 的故障转移

​所需权限

​备份和恢复

创建数据库

使用示例

动态将新表添加到复制范围中

动态将表从复制中移除

PostgreSQL schema

要求

设置

`materialized_postgresql_tables_list`

`materialized_postgresql_schema`

`materialized_postgresql_schema_list`

`materialized_postgresql_max_block_size`

`materialized_postgresql_replication_slot`

`materialized_postgresql_snapshot`

`materialized_postgresql_use_unique_replication_consumer_identifier`

`materialized_postgresql_use_extended_date_and_time_types`

注意事项

logical replication slot 的故障转移

所需权限

备份和恢复