C-клиент ClickHouse

Header-only C-клиент для собственного протокола ClickHouse, предназначенный для встраивания.

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

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

Эта библиотека находится в активной разработке. Версия v1 декодирует основные типы ClickHouse. Сообщайте об ограничениях или недостающей функциональности через issue tracker. Однако имейте в виду, что часть функциональности в этой библиотеке отсутствует намеренно.

​

Чего библиотека не делает

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

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

​

Как организована библиотека

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

Header	Назначение	Флаги линковки
clickhouse.h	Ядро: типы, ошибки, аллокатор, vtable I/O, парсер имён типов, средство чтения блоков и средство записи	—
clickhouse-client.h	Цикл обработки TCP-пакетов: Hello, Query, Data, EndOfStream, Exception, Progress, Pong	—
clickhouse-async.h	Клиент без I/O: тот же цикл пакетов, управляемый вызывающей стороной через передачу байтов, без socket	—
clickhouse-compression.h	Структура сжатого фрейма, CityHash128, диспетчеризация кодеков и адаптеры LZ4/ZSTD	-llz4 -lzstd
clickhouse-posix-io.h	I/O backend поверх блокирующих read(2)/write(2)	—
clickhouse-openssl.h	I/O backend поверх SSL_read/SSL_write	-lssl -lcrypto

​

Обязательная настройка сервера

Декодер считывает из wire печатные имена типов, поэтому они должны кодироваться как текст. 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.

​

Подключение по TCP

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

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, пройти по смещениям массива, а затем вырезать строковые данные.

Структура	Аксессоры
CHC_COL_FIXED	chc_column_fixed_data(c, &elem_size) — n_rows * elem_size байт в формате little-endian
CHC_COL_STRING	chc_column_string_data(c), chc_column_string_offsets(c) — offsets[i] — это позиция конца строки i (не включая) в порядке байтов хоста; строка 0 начинается с 0
CHC_COL_NULLABLE	chc_column_null_map(c) (один байт на строку, 1 = NULL), chc_column_nullable_inner(c)
CHC_COL_ARRAY	chc_column_array_offsets(c) (накопительные конечные смещения), chc_column_array_values(c); Map декодируется как Array(Tuple(K, V))
CHC_COL_TUPLE	chc_column_tuple_arity(c), chc_column_tuple_child(c, i) — у каждого дочернего элемента одинаковое число строк
CHC_COL_LOW_CARDINALITY	chc_column_lc_key_size(c) (1/2/4/8), chc_column_lc_keys(c), chc_column_lc_dict(c); слот 0 в словаре — значение по умолчанию

Пример чтения обычных числовых, строковых и столбцов с типом 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 — это две little-endian половины UInt64, IPv4 — 4-байтовое little-endian целое число, а IPv6 использует network byte order. Тики DateTime64 отсчитываются в UTC — timezone в типе служит лишь как metadata.

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

​

Вставка данных

Соберите блок с помощью chc_block_builder, затем передайте его в chc_client_send_data. Построитель блоков сохраняет указатели вместо копирования, поэтому срезы столбцов должны жить дольше, чем длится отправка. 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_builder *bb = NULL;
chc_block_builder_init(&bb, &al, &err);

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

/* 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_block_builder_append_string(bb, "message", 7, offsets, bytes, 3, &err);

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_block_builder_destroy(bb);
chc_type_destroy(u64, &al);

chc_block_builder_append_fixed принимает n_rows * elem_size байт в формате little-endian; chc_block_builder_append_string принимает накопительные конечные смещения exclusive в порядке байтов хоста для упакованного slab. Передача построителя блоков через chc_client_send_data, а не через более низкоуровневый chc_block_write, позволяет клиенту задавать параметры блока в соответствии с согласованной ревизией и применять сжатие.

​

Сжатие

Передайте режим сжатия и настроенный кодек в chc_client_opts. Клиент распаковывает входящие пакеты данных и сжимает исходящие. В заголовочном файле сжатия предусмотрены адаптеры 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.

​

TLS

clickhouse-openssl.h предоставляет chc_io backend-соединение поверх 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-соединения поддерживают необязательный callback check_cancel, который опрашивается между операциями чтения, а также deadline чтения через chc_openssl_io_set_deadline / chc_posix_io_set_deadline.

​

Клиент без I/O (async)

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

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

#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. Отправка никогда не блокируется и не создаёт обратного давления, поэтому следите за длиной очереди исходящих данных и прекращайте отправку, когда она становится слишком большой.

Рабочий драйвер 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_block_builder_append_json_string. Любая другая версия JSON- сериализации возвращает CHC_ERR_TYPE с указанием настройки.

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