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 表。

注意事项

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

​创建数据库

​使用示例

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

​动态将表从复制中移除

​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

​注意事项

​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`

注意事项

logical replication slot 的故障转移

所需权限