Использование

Ниже - весь путь от создания сервера до записи, чтения и транзакционного staging, одним сквозным примером. Пароль и адреса хостов подставьте свои.

1. Расширения

Один раз на базу данных, после установки коннектора на кластер (см. Установка):

CREATE EXTENSION pxf_fdw;
CREATE EXTENSION clickhouse_fdw;

2. Сервер

SERVER описывает, где находится ClickHouse:

CREATE SERVER clickhouse_server
    FOREIGN DATA WRAPPER tkh_fdw            -- или clickhouse_fdw - оба работают одинаково
    OPTIONS (database 'default', hosts 'ch1.example.com:8123,ch2.example.com:8123');

tkh_fdw и clickhouse_fdw - два имени одного и того же обработчика: используйте tkh_fdw, если переносите DDL из ADB как есть, или clickhouse_fdw, если пишете DDL с нуля.

3. Учётные данные

CREATE USER MAPPING FOR analyst
    SERVER clickhouse_server
    OPTIONS (clickhouse_username 'gp_writer', clickhouse_password 'secret');

Другой способ хранения пароля - в файле clickhouse-site.xml на стороне PXF, без USER MAPPING - описан на странице Конфигурация.

4. Внешняя таблица

CREATE FOREIGN TABLE ft_orders (
    id          int8,
    created_at  timestamp,
    amount      numeric(15,2),
    comment     text
)
SERVER clickhouse_server
OPTIONS (resource 'orders');   -- таблица default.orders в ClickHouse

resource - обязательная опция, имя целевой таблицы: table (использует database с уровня сервера) или db.table (переопределяет базу для конкретной таблицы). Колонки в DDL должны совпадать с ClickHouse по имени, не обязательно по порядку - коннектор всегда собирает INSERT с явным списком колонок. Если имена не совпадают, используйте column_map.

5. Запись

INSERT INTO ft_orders VALUES (1, now(), 199.90, 'first order');

INSERT INTO ft_orders
SELECT id, created_at, amount, comment
FROM local_orders
WHERE created_at >= current_date - interval '1 day';

INSERT ... SELECT - основной способ переливки данных: каждый сегмент читает свою часть источника и пишет её напрямую в ClickHouse через локальный PXF, параллельно с остальными сегментами.

Формат обмена между Greenplum и PXF - всегда TEXT. Не задавайте опцию format в DDL - FDW-объекты уже несут её по умолчанию, а любое другое значение не сработает.

6. Чтение

SELECT * FROM ft_orders WHERE created_at >= current_date - interval '1 day';

SELECT id, amount FROM ft_orders WHERE amount > 1000;

Каждый сегмент получает свой срез таблицы ClickHouse и читает его через свой локальный PXF - так же параллельно, как и при записи. Условия WHERE, которые Greenplum умеет передать удалённому источнику, уходят в ClickHouse как есть.

7. Режим выполнения

По умолчанию и запись, и чтение идут со всех сегментов параллельно (mpp_execute 'all segments'). Для небольших или административных запросов, где параллелизм не нужен, можно понизить до одного координатора:

OPTIONS (mpp_execute 'master')

8. Транзакционная запись (staging)

Без дополнительных настроек запись в ClickHouse не связана с транзакцией Greenplum: если INSERT прерван ошибкой или отменой, строки, которые ClickHouse уже принял, останутся в целевой таблице - откат Greenplum их не уберёт.

Опция use_staging решает это. Каждый сегмент пишет не в целевую таблицу, а во временную staging-таблицу, и только на COMMIT транзакции Greenplum данные переносятся в целевую таблицу - самим ClickHouse, присоединением партиций или через INSERT ... SELECT. На ROLLBACK временная таблица удаляется без переноса.

CREATE FOREIGN TABLE ft_orders_txn (
    id          int8,
    created_at  timestamp,
    amount      numeric(15,2),
    comment     text
)
SERVER clickhouse_server
OPTIONS (resource 'orders', use_staging 'true', staging_table_name_format 'stg_$');

INSERT INTO ft_orders_txn VALUES (2, now(), 50.00, 'staged order');

staging_table_name_format обязателен вместе с use_staging 'true' и должен быть задан на том же уровне DDL.

Один нюанс: SAVEPOINT внутри транзакции со staging не защищает от потери данных - строки уже могли уйти в staging-таблицу к моменту отката к savepoint, а откатить их оттуда нечем. Если такое происходит, коннектор помечает всю транзакцию как повреждённую, и следующий COMMIT завершается ошибкой. Если нужен частичный откат, используйте отдельные транзакции вместо SAVEPOINT вокруг вставок в такие таблицы.

Осиротевшие staging-таблицы

Если координатор Greenplum падает между тем, как staging-таблица заполнена, и переносом данных на COMMIT, эта таблица остаётся в ClickHouse навсегда - он ничего не знает о судьбе транзакции Greenplum. Для очистки таких сирот:

SELECT clickhouse.cleanup_staging('clickhouse_server');                -- старше 1 дня по умолчанию
SELECT clickhouse.cleanup_staging('clickhouse_server', '6 hours');

Функция удаляет таблицы, чьё имя совпадает с шаблоном staging_table_name_format этого сервера и которые не менялись дольше заданного срока. Более свежие таблицы не трогает - это защита от гонки с ещё активной транзакцией. Хороший вариант - вызывать её по расписанию (cron/pg_cron) раз в несколько часов.

9. Метрики

curl -s localhost:5888/actuator/prometheus | grep pxf_clickhouse

Полный список метрик - на странице Конфигурация.

DDL заказчика ADB (совместимость)

Ниже - пример DDL, написанного под ADB ClickHouse Connector 2.0 (tkh_fdw), без единой правки:

CREATE SERVER clickhouse_server
    FOREIGN DATA WRAPPER tkh_fdw
    OPTIONS (database 'default', hosts 'ch1.example.com:8123');

CREATE USER MAPPING FOR gpadmin
    SERVER clickhouse_server
    OPTIONS (clickhouse_username 'default', clickhouse_password '');

CREATE FOREIGN TABLE rds.tkh_operations (
    uid varchar NULL,
    operation_date date NULL,
    operation_time varchar NULL,
    point_id int4 NULL,
    total_amount numeric(15, 2) NULL,
    comment varchar NULL,
    ord_num int4 NULL,
    last_modified_date timestamp NOT NULL
)
SERVER clickhouse_server
OPTIONS (resource 'operations');

INSERT INTO rds.tkh_operations
SELECT uid, operation_date, operation_time, point_id,
       total_amount, comment, ord_num, last_modified_date
FROM staging.operations_source;

Единственное, на что стоит обратить внимание при переносе: если исходная колонка Greenplum несёт микросекунды, а в ClickHouse она объявлена как DateTime (секундная точность), дробная часть будет отброшена. Для сохранения микросекунд используйте DateTime64(6).