MaterializedPostgreSQL - ClickHouse Documentation

Пользователям ClickHouse Cloud рекомендуется использовать ClickPipes для репликации PostgreSQL в ClickHouse. Этот способ нативно поддерживает высокопроизводительную CDC (фиксация изменений данных) для PostgreSQL.

Создает базу данных ClickHouse с таблицами из базы данных PostgreSQL. Сначала база данных с движком MaterializedPostgreSQL создает снимок базы данных PostgreSQL и загружает необходимые таблицы. Эти таблицы могут представлять собой любое подмножество таблиц из любого подмножества схем указанной базы данных. Вместе со снимком движок базы данных получает LSN, а после выполнения первоначального дампа таблиц начинает получать обновления из WAL. После создания базы данных новые таблицы, добавленные в базу данных PostgreSQL, автоматически в репликацию не включаются. Их нужно добавлять вручную запросом ATTACH TABLE db.table. Репликация реализована с использованием протокола PostgreSQL Logical Replication Protocol, который не позволяет реплицировать DDL, но позволяет определить, произошли ли изменения, нарушающие репликацию (изменение типов столбцов, добавление или удаление столбцов). Такие изменения обнаруживаются, и соответствующие таблицы перестают получать обновления. В этом случае следует использовать запросы ATTACH/DETACH PERMANENTLY, чтобы полностью перезагрузить таблицу. Если DDL не нарушает репликацию (например, при переименовании столбца), таблица все равно будет получать обновления (вставка выполняется по позиции).

Этот движок базы данных является экспериментальным. Чтобы использовать его, установите 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 при добавлении таблицы в репликацию оставался неудалённый временный слот репликации (с именем {db_name}_ch_replication_slot_tmp). Если вы выполняете ATTACH таблиц в ClickHouse версии ниже 22.1, обязательно удалите его вручную (SELECT pg_drop_replication_slot('{db_name}_ch_replication_slot_tmp')). В противном случае будет расти использование диска. Эта проблема исправлена в версии 22.1.

Динамическое удаление таблиц из репликации

Из репликации можно исключить отдельные таблицы:

DETACH TABLE postgres_database.table_to_remove PERMANENTLY;

Схема PostgreSQL

Схема схема в PostgreSQL может быть настроена тремя способами (начиная с версии 21.12).

Одна схема для одного движка базы данных MaterializedPostgreSQL. Требуется использовать настройку 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. Требуется использовать настройку materialized_postgresql_tables_list. Каждая таблица записывается вместе со своей схемой. Доступ к таблицам осуществляется одновременно по имени схемы и имени таблицы:

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 должны быть указаны вместе с именем своей схемы. Требуется materialized_postgresql_tables_list_with_schema = 1. Предупреждение: в этом случае точки в именах таблиц не допускаются.

Любое количество схем с полным набором таблиц для одного движка базы данных MaterializedPostgreSQL. Для этого требуется использовать настройку 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`;

Предупреждение: в этом случае точки в имени таблицы недопустимы.

Требования

Параметр wal_level в файле конфигурации PostgreSQL должен иметь значение 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`

Значение по умолчанию: пустая строка. (Используется схема по умолчанию)

`materialized_postgresql_schema_list`

По умолчанию: пустой список. (Используется схема по умолчанию)

`materialized_postgresql_max_block_size`

Задаёт количество строк, накапливаемых в памяти перед записью данных в таблицу базы данных PostgreSQL. Возможные значения:

Положительное целое число.

Значение по умолчанию: 65536.

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

Использовать уникальный идентификатор consumer’а репликации. Значение по умолчанию: 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, где типы столбцов задаются явно.

Примечания

Переключение слота логической репликации при сбое

Слоты логической репликации, существующие на основном узле, недоступны на резервных репликах. Поэтому при переключении при сбое новый основной узел (бывший физический standby) не будет знать о слотах, существовавших на старом основном узле. Это приведет к нарушению репликации из PostgreSQL. Решение — самостоятельно управлять слотами репликации и определить постоянный слот репликации (дополнительную информацию можно найти здесь). Имя слота нужно передать через настройку materialized_postgresql_replication_slot, а сам слот должен быть экспортирован с параметром EXPORT SNAPSHOT. Идентификатор снимка нужно передать через настройку materialized_postgresql_snapshot. Обратите внимание, что использовать это следует только при реальной необходимости. Если явной необходимости нет или вы не до конца понимаете, зачем это нужно, лучше позволить движку таблицы самостоятельно создать слот репликации и управлять им. Пример (от @bchrobot)

Настройте слот репликации в PostgreSQL.

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

Дождитесь, пока слот репликации будет готов, затем начните транзакцию и экспортируйте идентификатор снимка транзакции:
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';

Завершите транзакцию PostgreSQL, как только убедитесь, что репликация в DB ClickHouse работает. Проверьте, что после переключения при сбое репликация продолжается:
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 — владелец публикации (username в самом движке MaterializedPostgreSQL).

Можно избежать выполнения команд 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 на месте не поддерживается. Восстановленный объект MaterializedPostgreSQL сразу начинает репликацию из действующего источника PostgreSQL, поэтому восстановление поверх него снимка резервной копии привело бы к смешению данных из снимка с текущим состоянием удалённого источника. Поэтому в этом случае команда RESTORE завершается отказом. Вместо этого восстанавливайте захваченные данные в обычные таблицы ReplacingMergeTree:

В резервной копии базы данных сохранённое определение каждой таблицы уже является синтетической вложенной таблицей ReplacingMergeTree (а не движком MaterializedPostgreSQL), поэтому каждую таблицу можно восстановить напрямую в новую, ещё не существующую таблицу:
RESTORE TABLE postgres_db.table1 AS restored_db.table1 FROM Disk('backups', 'postgres_db.zip') SETTINGS allow_different_table_def = 1;
Для резервной копии автономной таблицы MaterializedPostgreSQL сохранённое определение представляет собой сам движок MaterializedPostgreSQL. Заранее создайте таблицу ReplacingMergeTree с той же структурой, что и вложенная таблица (включая столбцы _sign и _version), и восстановите данные в неё:
RESTORE TABLE src AS existing_replacing_mergetree FROM Disk('backups', 'table.zip') SETTINGS allow_different_table_def = 1;

​Создание базы данных

​Пример использования

​Динамическое добавление новых таблиц в репликацию

​Динамическое удаление таблиц из репликации

​Схема PostgreSQL

​Требования

​Настройки

​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

​Примечания

​Переключение слота логической репликации при сбое

​Необходимые разрешения

​Резервное копирование и восстановление

Создание базы данных

Пример использования

Динамическое добавление новых таблиц в репликацию

Динамическое удаление таблиц из репликации

Схема PostgreSQL

Требования

Настройки

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

Примечания

Переключение слота логической репликации при сбое

Необходимые разрешения

Резервное копирование и восстановление