[экспериментальный] MaterializedPostgreSQL

Создает базу данных ClickHouse с исходным дампом данных таблиц PostgreSQL и запускает процесс репликации, т.е. выполняется применение новых изменений в фоне, как эти изменения происходят в таблице PostgreSQL в удаленной базе данных PostgreSQL.

Сервер ClickHouse работает как реплика PostgreSQL. Он читает WAL и выполняет DML запросы. Данные, полученные в результате DDL запросов, не реплицируются, но сами запросы могут быть обработаны (описано ниже).

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

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

Параметры движка

• host:port — адрес сервера PostgreSQL.
• database — имя базы данных на удалённом сервере.
• user — пользователь PostgreSQL.
• password — пароль пользователя.

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

ATTACH TABLE postgres_database.new_table;

При указании конкретного списка таблиц в базе с помощью настройки materialized_postgresql_tables_list, он будет обновлен (в .sql метаданных) на актуальный с учетом таблиц, добавленных с помощью запроса ATTACH TABLE.

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

DETACH TABLE postgres_database.table_to_remove;

Настройки

• materialized_postgresql_max_block_size

• materialized_postgresql_tables_list

• materialized_postgresql_allow_automatic_update

• materialized_postgresql_replication_slot

• materialized_postgresql_snapshot

CREATE DATABASE database1
ENGINE = MaterializedPostgreSQL('postgres1:5432', 'postgres_database', 'postgres_user', 'postgres_password')
SETTINGS materialized_postgresql_max_block_size = 65536,
         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>;

Требования

1. Настройка wal_level должна иметь значение logical, параметр max_replication_slots должен быть равен по меньшей мере 2 в конфигурационном файле в PostgreSQL.

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;

Первичный ключ всегда проверяется первым. Если он отсутствует, то проверяется индекс, определенный как replica identity 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-значений не поддерживается. Для типа данных будет использоваться значение по умолчанию.

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

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

SELECT * FROM postgresql_db.postgres_table;

Примечания

Сбой слота логической репликации

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

Имейте в виду, что это стоит делать только если есть реальная необходимость. Если такой необходимости нет, или если нет полного понимания того, как это работает, то самостоятельно слот репликации конфигурировать не стоит, он будет создан таблицей.

Пример (от @bchrobot)

1. Сконфигурируйте слот репликации в 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

2. Дождитесь готовности слота репликации, затем инициируйте транзакцию и экспортируйте идентификатор снэпшота этой транзакции:

BEGIN;
SELECT pg_export_snapshot();

3. Создайте базу данных в 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';

4. Когда начнет выполняться репликация БД в ClickHouse, прервите транзакцию в PostgreSQL. Убедитесь, что репликация продолжается после сбоя:

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