C-клиент ClickHouse

clickhouse-c — это C-клиент, состоящий только из заголовочных файлов, для ClickHouse собственного протокола. Исходный код и справочная информация по каждому заголовочному файлу доступны в репозитории GitHub.

В отличие от более высокоуровневых клиентов, он намеренно берёт на себя минимум. Основной заголовочный файл декодирует и кодирует блоки формата Native через предоставленный вами callback ввода-вывода. Управление сокетом, TLS-контекстом, аллокатором, retries и пулом соединений остаётся на вашей стороне. Благодаря этому библиотека достаточно мала для встраивания: подключение одного лишь clickhouse.h не добавляет зависимостей на этапе линковки, кроме libc.

Это намеренно не входит в её задачи. Реализуйте это в своём приложении или с помощью родственной библиотеки:

• HTTP-протокол. Напрямую оберните libcurl для HTTP-интерфейса.
• DNS-разрешение, переключение конечной точки при отказе, пул соединений, повторные попытки и задержка.
• Жизненный цикл TLS-контекста. Backend OpenSSL использует уже подключённый вами SSL.
• Потоки. Каждый chc_client по своей архитектуре однопоточный.
• Асинхронный I/O внутри библиотеки. Блокирующий клиент вызывает chc_io.read синхронно. Для клиента на основе цикла событий, который сам не выполняет I/O, используйте клиент без I/O.

clickhouse-c поставляется как единый набор заголовков. Каждый заголовочный файл содержит и объявления, и реализацию, защищённые сторожевым макросом. Выберите заголовки, необходимые для вашей сборки.

Декодер считывает печатные имена типов из передаваемых по сети данных, поэтому они должны быть закодированы как текст. ClickHouse по умолчанию записывает их в текстовом виде, но явно задавайте эту настройку в своих запросах, чтобы профиль сервера или сеанса, который переключает её в двоичный формат, не нарушил декодирование:

output_format_native_encode_types_in_binary_format = 0

Устанавливать пакет не нужно, поэтому включите заголовочные файлы в дерево проекта через Git-подмодуль или просто скопируйте их. Ровно в одной единице трансляции определяется CHC_IMPLEMENTATION и подключается реализация; во всех остальных единицах подключаются те же заголовочные файлы только с объявлениями.

/* clickhouse_impl.c */
#define CHC_IMPLEMENTATION
#include "clickhouse.h"
#include "clickhouse-posix-io.h"
#include "clickhouse-client.h"
#include "clickhouse-compression.h"

/* every other TU */
#include "clickhouse.h"
#include "clickhouse-client.h"

Определите CHC_PROVIDE_STDLIB_ALLOC перед подключением clickhouse.h, чтобы использовать chc_alloc_stdlib. Определите CHC_NO_LZ4 или CHC_NO_ZSTD для clickhouse-compression.h, чтобы исключить зависимости от lz4/zstd.

Чтобы подключиться к серверу ClickHouse, вы самостоятельно настраиваете сокет, оборачиваете его в chc_io и передаёте в chc_client_init, который синхронно выполняет рукопожатие Hello. Библиотека не занимается DNS, failover, повторным подключением или pooling — это задача вызывающей стороны.

int fd = socket(AF_INET, SOCK_STREAM, 0);
int one = 1;
setsockopt(fd, IPPROTO_TCP, TCP_NODELAY, &one, sizeof one);

struct sockaddr_in sa = {};
sa.sin_family      = AF_INET;
sa.sin_port        = htons(9000);
sa.sin_addr.s_addr = htonl(INADDR_LOOPBACK);
connect(fd, (struct sockaddr *) &sa, sizeof sa);

chc_alloc al = chc_alloc_stdlib();
chc_posix_io state;
chc_io io;
chc_posix_io_init(&state, &io, fd, NULL, NULL);

chc_client *client = NULL;
chc_client_opts opts = {
    .user     = "default",
    .password = "",
    .database = "default",
};
chc_err err = {};
if (chc_client_init(&client, &opts, &al, &io, &err) != CHC_OK) {
    fprintf(stderr, "connect: %s\n", err.msg);
    chc_client_close(client);   /* safe to call on the NULL-on-failure handle */
    return 1;
}

const chc_server_info *info = chc_client_server_info(client);
printf("connected to %s %llu.%llu.%llu\n", info->display_name,
       (unsigned long long) info->version_major,
       (unsigned long long) info->version_minor,
       (unsigned long long) info->version_patch);

Каждый chc_client однопоточный и оборачивает одно соединение. Библиотека синхронно вызывает обратные вызовы chc_io; что именно эти обратные вызовы делают на нижнем уровне (epoll, io_uring, WaitLatchOrSocket), зависит от вас.

Отправьте запрос, затем считывайте пакеты до CHC_PKT_END_OF_STREAM. Используйте chc_client_send_query_ex, чтобы передать обязательную настройку сервера; chc_client_send_query без дополнительных аргументов отправляет пустой список настроек и наследует серверные значения по умолчанию.

chc_query_setting settings[] = {
    { .name = "output_format_native_encode_types_in_binary_format", .value = "0" },
};
chc_query_opts qopts = { .settings = settings, .n_settings = 1 };

const char *sql = "SELECT number, toString(number * number) FROM numbers(5)";
if (chc_client_send_query_ex(client, sql, strlen(sql), &qopts, &err) != CHC_OK) {
    fprintf(stderr, "query: %s\n", err.msg);
    return 1;
}

for (;;) {
    chc_packet pkt = {};
    if (chc_client_recv_packet(client, &pkt, &err) != CHC_OK) {
        fprintf(stderr, "recv: %s\n", err.msg);
        break;
    }

    if (pkt.kind == CHC_PKT_DATA) {
        for (size_t r = 0; r < chc_block_n_rows(pkt.block); r++)
            for (size_t c = 0; c < chc_block_n_columns(pkt.block); c++)
                print_value(chc_block_column_type(pkt.block, c),
                            chc_block_column(pkt.block, c), r);
    } else if (pkt.kind == CHC_PKT_EXCEPTION) {
        fprintf(stderr, "server: %s\n", pkt.exception->display_text);
    }

    bool done = pkt.kind == CHC_PKT_END_OF_STREAM;
    chc_packet_clear(client, &pkt);
    if (done) break;
}

Исключения сервера поступают в виде пакетов CHC_PKT_EXCEPTION, а не как возврат со статусом не-OK из chc_client_recv_packet. Статус не-OK возвращается только при сбоях транспортного уровня. Первый пакет CHC_PKT_DATA в результате — это заголовочный блок, описывающий схему с нулевым числом строк; далее следуют блоки данных. chc_packet_clear освобождает блок или исключение пакета'а — чтобы вместо этого забрать владение, сначала обнулите эти поля в пакете.

Блоки имеют столбцовую организацию. У каждого столбца есть физическая структура, которую возвращает chc_column_layout; по ней выбирается способ обработки, а объявленный тип возвращает chc_block_column_type. Составные структуры могут быть вложенными, поэтому чтение Nullable(Array(String)) означает, что нужно снять обёртку Nullable, обойти смещения массива, а затем выделить срезы строковых данных.

Пример чтения обычных числовых, строковых столбцов и столбцов с типом Nullable:

void print_value(const chc_type *t, const chc_column *c, size_t row)
{
    if (chc_column_layout(c) == CHC_COL_NULLABLE) {
        if (chc_column_null_map(c)[row]) { fputs("\\N", stdout); return; }
        print_value(chc_type_child(t, 0), chc_column_nullable_inner(c), row);
        return;
    }

    switch (chc_column_layout(c)) {
    case CHC_COL_FIXED: {
        /* fixed_data is a raw little-endian byte slab. memcpy into a typed
           local to avoid unaligned loads and strict-aliasing UB, then
           byte-swap on big-endian hosts. */
        size_t es;
        const uint8_t *p = chc_column_fixed_data(c, &es) + row * es;
        switch (chc_type_kind(t)) {
        case CHC_UINT64: { uint64_t v; memcpy(&v, p, sizeof v); printf("%" PRIu64, v); break; }
        case CHC_INT32:  { int32_t  v; memcpy(&v, p, sizeof v); printf("%" PRId32, v); break; }
        case CHC_FLOAT64: { double  v; memcpy(&v, p, sizeof v); printf("%g", v); break; }
        /* ... remaining numeric kinds ... */
        default: break;
        }
        break;
    }
    case CHC_COL_STRING: {
        const uint8_t  *bytes   = chc_column_string_data(c);
        const uint64_t *offsets = chc_column_string_offsets(c);
        uint64_t start = row == 0 ? 0 : offsets[row - 1];
        fwrite(bytes + start, 1, (size_t) (offsets[row] - start), stdout);
        break;
    }
    default: break;
    }
}

Данные CHC_COL_FIXED при передаче имеют порядок байтов little-endian; на хостах с порядком байтов big-endian вам нужно самостоятельно переставлять байты в многобайтовых целых числах. Смещения и ключи LowCardinality уже приводятся к порядку байтов хоста при декодировании. UUID представляются как две половины UInt64 в little-endian, IPv4 — как 4-байтовое little-endian-целое число, а IPv6 — в сетевом порядке байтов. Тики DateTime64 — в UTC; часовой пояс в типе — это лишь метаданные.

При приёме данных от недоверенного peer вызывайте chc_column_validate для каждого столбца, прежде чем обходить его. chc_block_read не проверяет межполевые инварианты, такие как смещения массивов и ключи LowCardinality, поэтому без такой проверки поддельный блок может прочитать данные за пределами границ внутренних столбцов.

Соберите столбцы с помощью вспомогательных функций chc_build_*, добавьте их в chc_block_builder, а затем передайте его в chc_client_send_data. chc_block_builder использует хранилище, предоставленное вызывающей стороной, и сохраняет указатели, а не копирует данные, поэтому хранилище, деревья столбцов, типы, имена и slabs должны существовать дольше, чем длится отправка. INSERT отправляет запрос, ожидает заголовочный блок от сервера, отправляет один или несколько блоков данных, а затем отправляет пустой блок, чтобы завершить поток.

const char *sql = "INSERT INTO greetings (id, message) VALUES";
chc_client_send_query(client, sql, strlen(sql), "", 0, &err);

/* Wait for the server's header block (schema, 0 rows). */
bool got_header = false;
while (!got_header) {
    chc_packet pkt = {};
    if (chc_client_recv_packet(client, &pkt, &err) != CHC_OK) {
        fprintf(stderr, "recv: %s\n", err.msg);
        return 1;
    }
    chc_packet_kind kind = pkt.kind;
    if (kind == CHC_PKT_DATA) got_header = true;
    else if (kind == CHC_PKT_EXCEPTION && pkt.exception)
        fprintf(stderr, "server: %s\n", pkt.exception->display_text);
    chc_packet_clear(client, &pkt);
    if (kind == CHC_PKT_EXCEPTION || kind == CHC_PKT_END_OF_STREAM) return 1;  /* no header coming */
}

chc_block_col columns[2];
chc_block_builder bb;
chc_block_builder_init(&bb, columns);

uint64_t ids[3] = { 1, 2, 3 };
chc_type *u64 = NULL;
chc_type_parse("UInt64", 6, &al, &u64, &err);
chc_column id = chc_build_fixed(ids, sizeof ids[0], 3);
chc_block_builder_append(&bb, "id", 2, u64, &id);

/* String columns: cumulative exclusive end offsets + a packed byte slab. */
uint64_t offsets[3] = { 5, 11, 20 };   /* "hello", "buenas", "goedendag" */
const uint8_t bytes[] = "hellobuenasgoedendag";
chc_type *string = NULL;
chc_type_parse("String", 6, &al, &string, &err);
chc_column message = chc_build_string(offsets, bytes, 3);
chc_block_builder_append(&bb, "message", 7, string, &message);

chc_client_send_data(client, &bb, &err);  /* the populated block */
chc_client_send_data(client, NULL, &err); /* empty block ends the INSERT */

/* Drain to EndOfStream. */
for (;;) {
    chc_packet pkt = {};
    chc_client_recv_packet(client, &pkt, &err);
    bool done = pkt.kind == CHC_PKT_END_OF_STREAM;
    chc_packet_clear(client, &pkt);
    if (done) break;
}

chc_type_destroy(u64, &al);
chc_type_destroy(string, &al);

chc_build_fixed принимает n_rows * elem_size байт в формате little-endian; chc_build_string принимает накопительные исключающие конечные смещения в порядке байт хоста для упакованного slab. Вспомогательные функции возвращают узлы столбцов по значению. Вкладывайте их в соответствии с типом: например, передайте фиксированный или строковый узел в chc_build_nullable, затем передайте результат в chc_build_array и добавьте корневой узел массива. Tuple, LowCardinality, Map и Geo-столбцы используют одно и то же дерево: Map — это Array(Tuple(K, V)).

Все столбцы в блоке должны иметь одинаковое количество строк на верхнем уровне. writer проверяет дерево на соответствие разобранному типу ClickHouse, но вызывающая сторона должна выделить хранилище chc_block_col для каждого append. Вы также можете напрямую добавить декодированный столбец из chc_block_column, чтобы повторно закодировать его, или вызвать chc_block_write_cols с массивом chc_block_col, чтобы обойтись без builder. Использование builder через chc_client_send_data, а не через низкоуровневый chc_block_write, позволяет client задавать параметры блока на основе согласованной ревизии и применять сжатие.

Передайте режим сжатия и настроенный кодек в chc_client_opts. Клиент распаковывает входящие пакеты Data и сжимает исходящие. В заголовочном файле сжатия есть адаптеры LZ4 и ZSTD; каждая функция инициализации заполняет только свои слоты, поэтому вызовите обе, чтобы поддерживать любой из этих вариантов.

#include "clickhouse-compression.h"

chc_codec codec = {};
chc_lz4_codec_init(&codec);
chc_zstd_codec_init(&codec);

chc_client_opts opts = {
    .user        = "default",
    .compression = CHC_COMP_LZ4,   /* or CHC_COMP_ZSTD */
    .codec       = &codec,
};

Чтобы использовать библиотеку сжатия, для которой в проекте нет готового биндинга, самостоятельно заполните структуру chc_codec; vtable объявлена в clickhouse-compression.h.

clickhouse-openssl.h предоставляет реализацию chc_io поверх SSL_read/SSL_write. OpenSSL настраиваете вы: библиотека сама не создает SSL_CTX, не проверяет сертификаты, не задает SNI и не вызывает SSL_connect / SSL_shutdown. К моменту срабатывания chc_io.read рукопожатие уже должно быть завершено.

#include "clickhouse-openssl.h"

SSL *ssl = /* connected, handshake complete */;
chc_openssl_io state;
chc_io io;
chc_openssl_io_init(&state, &io, ssl, NULL, NULL);
/* hand &io to chc_client_init, same as the POSIX backend */

ClickHouse Cloud и другие развертывания с поддержкой TLS используют собственный протокол на порту 9440. Оба backend-соединения поддерживают необязательную функцию обратного вызова check_cancel, которая опрашивается между операциями чтения, а также дедлайн чтения через chc_openssl_io_set_deadline / chc_posix_io_set_deadline.

clickhouse-async.h — это ioless-вариант TCP-клиента для циклов событий. Он вообще не работает с сокетом: вы передаёте полученные байты и считываете байты, которые клиент хочет отправить, самостоятельно управляя epoll, io_uring или WaitLatchOrSocket. Параметры, типы пакетов и построитель блоков здесь те же, что и у блокирующего клиента.

chc_async_client_init не выполняет I/O и не может блокироваться. После этого рукопожатие работает как возобновляемая машина состояний; то же самое относится к каждой операции отправки и получения. Когда разбор выходит за пределы переданных вами байтов, вызов возвращает CHC_WOULD_BLOCK вместо блокировки — передайте больше входящих байтов и вызовите функцию снова, и парсер продолжит работу с середины блока.

#include "clickhouse-async.h"

chc_async_client *c = NULL;
chc_client_opts opts = { .user = "default" };
chc_async_client_init(&c, &opts, &al, &err);

for (;;) {
    int rc = chc_async_handshake(c, &err);
    if (rc == CHC_OK) break;
    if (rc != CHC_WOULD_BLOCK) break;   /* hard error */
    pump(c);   /* drain pending_out to the socket; feed received bytes to chc_async_submit */
}

chc_async_send_query(c, sql, strlen(sql), "", 0, &err);

for (;;) {
    chc_packet pkt = {};
    int rc = chc_async_recv_packet(c, &pkt, &err);
    if (rc == CHC_WOULD_BLOCK) { pump(c); continue; }
    if (rc != CHC_OK) break;

    bool done = pkt.kind == CHC_PKT_END_OF_STREAM;
    if (pkt.kind == CHC_PKT_DATA && pkt.block) { /* read columns as above */ }
    chc_async_packet_clear(c, &pkt);
    if (done) break;
}

Ваш pump перемещает байты в обоих направлениях. Для исходящего потока chc_async_pending_out возвращает указатель и длину для байтов в очереди; после того как сокет примет часть данных, вызовите chc_async_consume_out с этим значением — частичная запись допустима. Для входящего потока передавайте в chc_async_submit данные, прочитанные из сокета. Отправка никогда не блокируется и не создает обратного давления, поэтому следите за длиной pending-out и прекращайте отправку, когда она становится слишком большой.

Рабочий драйвер liburing находится в test/test_async_uring.c.

В каждой точке входа используется vtable chc_alloc, поэтому выделение памяти работает по той же схеме, что и у хоста.

typedef struct chc_alloc {
    void *ud;
    void *(*alloc)  (void *ud, size_t bytes);
    void *(*realloc)(void *ud, void *p, size_t old_bytes, size_t new_bytes);
    void  (*free)   (void *ud, void *p, size_t bytes);
} chc_alloc;

Определите CHC_PROVIDE_STDLIB_ALLOC перед включением clickhouse.h и вызовите chc_alloc_stdlib() для стандартного аллокатора на базе malloc.

Функции возвращают CHC_OK (0) или ненулевой код CHC_ERR_*. Код является возвращаемым значением, а chc_err, выделяемый вызывающей стороной на стеке, содержит понятное человеку сообщение. Библиотека никогда не выделяет память в куче для ошибки.

typedef struct chc_err {
    int  server_code;           /* set when the return code is CHC_ERR_SERVER */
    char msg[CHC_ERR_MSG_LEN];  /* NUL-terminated, default 256 bytes */
    char server_name[64];       /* ClickHouse exception class, if SERVER */
} chc_err;

Ошибки запросов на стороне сервера — это не ошибки chc_err. Они приходят в потоке пакетов как CHC_PKT_EXCEPTION и содержат серверные code, display_text и stack_trace. Проверку chc_err оставьте для сбоев транспорта, протокола и декодирования.

Читатель блока декодирует:

• Int8–Int256, UInt8–UInt256
• Float32, Float64, BFloat16
• Bool
• Decimal32, Decimal64, Decimal128, Decimal256
• Date, Date32, DateTime, DateTime64, Time, Time64
• String, FixedString(N)
• UUID, IPv4, IPv6
• Enum8, Enum16
• Nullable(T), Array(T), Tuple(...), Map(K, V), Nested(...)
• LowCardinality(T)
• Interval
• QBit(...)
• Point, Ring, Polygon, MultiPolygon
• SimpleAggregateFunction(f, T), который декодируется как внутренний T
• JSON и Object('json') как столбцы String при строковой сериализации (см. ниже)

JSON и Object('json') декодируются при строковой сериализации; задайте output_format_native_write_json_as_string=1 в запросе. Каждая поддерживаемая строка передаётся как отдельный JSON-документ в столбце CHC_COL_STRING. Соберите ту же структуру с помощью chc_build_string; модуль записи добавляет префикс, необходимый для разобранного типа.

Variant, Dynamic, AggregateFunction пока не декодируются и возвращают CHC_ERR_TYPE; в качестве обходного варианта приводите их к String на стороне server.