Cliente C do ClickHouse

clickhouse-c é um cliente C header-only para o protocolo nativo do ClickHouse. O código-fonte e a referência de cada cabeçalho estão no repositório GitHub.

Ao contrário dos clientes de mais alto nível, ele intencionalmente faz pouca coisa por você. O cabeçalho principal decodifica e codifica blocos no formato Native por meio de um callback de E/S fornecido por você. Ficam por sua conta o socket, o contexto TLS, o alocador, as novas tentativas e o pool de conexões. Isso o torna pequeno o suficiente para ser embutido: incluir apenas clickhouse.h não traz dependências de linkagem além da libc.

Estes são objetivos deliberadamente fora do escopo. Trate-os na sua aplicação ou com uma biblioteca complementar:

• Protocolo HTTP. Use o libcurl diretamente como wrapper para a interface HTTP.
• Resolução de DNS, failover de endpoint, pool de conexões, retry e backoff.
• Ciclo de vida do contexto TLS. O backend OpenSSL usa um SSL cuja conexão você já estabeleceu.
• Threads. Cada chc_client foi projetado para usar uma única thread.
• E/S assíncrona dentro da biblioteca. O cliente bloqueante chama chc_io.read de forma síncrona. Para um cliente de loop de eventos que não realiza nenhuma E/S por conta própria, use o ioless client.

clickhouse-c é distribuída como um conjunto simples de cabeçalhos. Cada cabeçalho reúne declarações e implementação, protegidas por uma macro sentinela. Escolha os cabeçalhos de que sua compilação precisa.

O decodificador lê nomes de tipos legíveis a partir do wire, portanto eles precisam ser codificados como texto. O ClickHouse os escreve como texto por padrão, mas fixe essa configuração nas suas consultas para que um perfil de servidor ou de sessão que a defina como binária não possa prejudicar a decodificação:

output_format_native_encode_types_in_binary_format = 0

Não há pacote para instalar, então você deve incorporar os arquivos de cabeçalho à sua árvore de código-fonte por meio de um submódulo do Git ou de uma cópia. Exatamente uma unidade de tradução define CHC_IMPLEMENTATION e inclui a implementação; todas as outras unidades incluem os mesmos arquivos de cabeçalho apenas para declarações.

/* 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"

Defina CHC_PROVIDE_STDLIB_ALLOC antes de incluir clickhouse.h para usar chc_alloc_stdlib. Defina CHC_NO_LZ4 ou CHC_NO_ZSTD em clickhouse-compression.h para remover as dependências de lz4/zstd.

Para se comunicar com um servidor ClickHouse, você mesmo configura o socket, o encapsula em um chc_io e o passa para chc_client_init, que executa o handshake Hello de forma síncrona. A biblioteca não faz DNS, failover, reconexão nem pool de conexões — isso é responsabilidade de quem faz a chamada.

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);

Cada chc_client é de thread única e encapsula uma conexão. A biblioteca chama os callbacks chc_io de forma síncrona; o que esses callbacks fazem internamente (epoll, io_uring, WaitLatchOrSocket) fica a seu critério.

Envie a consulta e, em seguida, consuma os pacotes até CHC_PKT_END_OF_STREAM. Use chc_client_send_query_ex para anexar a configuração de servidor necessária; chc_client_send_query, sem parâmetros extras, envia uma lista de configurações vazia e herda a configuração padrão do servidor.

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;
}

As exceções do servidor chegam como pacotes CHC_PKT_EXCEPTION, não como um retorno não OK de chc_client_recv_packet. Somente falhas no nível de transporte retornam não OK. O primeiro pacote CHC_PKT_DATA de um resultado é um bloco de cabeçalho que descreve o esquema com zero linhas; os blocos de dados vêm em seguida. chc_packet_clear libera o bloco ou a exceção do pacote — primeiro defina esses campos do pacote como nulos para assumir a propriedade deles.

Os blocos são orientados a colunas. Cada coluna tem um layout físico, retornado por chc_column_layout, com base no qual você faz o encaminhamento; seu tipo declarado vem de chc_block_column_type. Layouts compostos são aninhados, então ler um Nullable(Array(String)) significa desempacotar o Nullable, percorrer os offsets do array e, em seguida, fatiar os dados da string.

Um leitor para colunas numéricas simples, de string e 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;
    }
}

Os dados CHC_COL_FIXED são little-endian on the wire; em hosts big-endian, você mesmo deve fazer o byte-swap de inteiros com vários bytes. Offsets e chaves de LowCardinality já são convertidos para a ordem do host no momento da decodificação. UUIDs são duas metades UInt64 little-endian, IPv4 é um inteiro little-endian de 4 bytes e IPv6 está em network byte order. Os ticks de DateTime64 são UTC — o timezone no tipo é apenas metadata.

Ao fazer a ingestão a partir de um peer não confiável, chame chc_column_validate em cada coluna antes de percorrê- la. chc_block_read não valida invariantes entre campos, como offsets de array e chaves de LowCardinality, então um bloco forjado poderia, caso contrário, ler além dos limites da coluna interna.

Construa colunas com os helpers chc_build_*, adicione-as a um chc_block_builder e, em seguida, passe-o para chc_client_send_data. O builder usa armazenamento fornecido pelo chamador e armazena ponteiros em vez de copiar os dados, portanto o armazenamento, as árvores de colunas, os tipos, os nomes e os slabs devem permanecer válidos durante o envio. Um INSERT envia a consulta, aguarda o bloco de cabeçalho do servidor, envia um ou mais blocos de dados e, em seguida, envia um bloco vazio para encerrar o fluxo.

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 aceita n_rows * elem_size bytes em little-endian; chc_build_string aceita offsets finais cumulativos exclusivos na ordem de bytes do host sobre um slab compactado. Os helpers retornam nós de coluna por valor. Aninhe-os para corresponder ao tipo: por exemplo, passe um nó fixo ou de string para chc_build_nullable, passe esse resultado para chc_build_array e acrescente a raiz do array. Tuple, LowCardinality, Map e colunas geo usam a mesma árvore: Map é Array(Tuple(K, V)).

Todas as colunas em um bloco devem ter a mesma contagem de linhas no nível superior. A rotina de gravação verifica a árvore em relação ao tipo do ClickHouse analisado, mas o chamador deve dimensionar o armazenamento chc_block_col para cada acréscimo. Você também pode acrescentar diretamente uma coluna decodificada de chc_block_column para recodificá-la, ou chamar chc_block_write_cols com um array chc_block_col para dispensar o builder. Passar o builder por meio de chc_client_send_data, em vez do chc_block_write de nível inferior, permite que o cliente defina as opções do bloco com base na revision negociada e aplique compressão.

Passe um modo de compressão e um codec configurado em chc_client_opts. O cliente descomprime os pacotes Data recebidos e comprime os enviados. O cabeçalho de compressão inclui adaptadores LZ4 e ZSTD; cada inicialização preenche apenas seus próprios slots, então chame ambos para dar suporte a qualquer um dos dois.

#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,
};

Para usar uma biblioteca de compressão para a qual o projeto não fornece um binding, defina você mesmo um chc_codec; a vtable está declarada em clickhouse-compression.h.

clickhouse-openssl.h fornece um backend chc_io baseado em SSL_read/SSL_write. Você controla o OpenSSL: a biblioteca nunca cria um SSL_CTX, verifica certificados, define o SNI nem chama SSL_connect / SSL_shutdown. Quando chc_io.read é acionado, o handshake já deve ter sido concluído.

#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 e outras implantações com TLS ativado usam o protocolo nativo na porta 9440. Ambos os backends aceitam um callback opcional check_cancel, consultado entre leituras, e um prazo de leitura por meio de chc_openssl_io_set_deadline / chc_posix_io_set_deadline.

clickhouse-async.h é uma variante ioless do cliente TCP para loops de eventos. Ele nunca acessa um socket: você fornece os bytes que recebeu e drena os bytes que ele quer enviar, controlando epoll, io_uring ou WaitLatchOrSocket por conta própria. As opções, os tipos de pacote e o construtor de blocos são os mesmos do cliente bloqueante.

chc_async_client_init não faz E/S e não pode bloquear. O handshake é executado depois como uma máquina de estados que pode ser retomada, assim como cada envio e recebimento. Quando o parser avança além dos bytes que você forneceu, a chamada retorna CHC_WOULD_BLOCK em vez de bloquear — forneça mais bytes de entrada e chame novamente, e o parser será retomado no meio do bloco.

#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;
}

Seu pump move bytes nos dois sentidos. Na saída, chc_async_pending_out retorna um ponteiro e o tamanho dos bytes enfileirados; depois que o socket aceitar parte deles, chame chc_async_consume_out com essa quantidade; uma escrita parcial é aceitável. Na entrada, passe as leituras do socket para chc_async_submit. Os envios nunca bloqueiam nem fazem backpressure, então monitore o tamanho pendente de saída e pare de enviar quando ele ficar grande demais.

Um driver liburing funcional está em test/test_async_uring.c.

Cada ponto de entrada recebe uma vtable chc_alloc, portanto a alocação fica a cargo do mecanismo usado pelo host.

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;

Defina CHC_PROVIDE_STDLIB_ALLOC antes de incluir clickhouse.h e chame chc_alloc_stdlib() para usar um alocador padrão baseado em malloc.

As funções retornam CHC_OK (0) ou um código CHC_ERR_* diferente de zero. O código é o valor de retorno; um chc_err alocado na pilha pelo chamador contém a mensagem legível para humanos. A biblioteca nunca aloca um erro no heap.

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;

Erros de consulta no servidor não são falhas chc_err. Eles chegam pelo fluxo de pacotes como CHC_PKT_EXCEPTION, trazendo code, display_text e stack_trace do servidor. Reserve a verificação de chc_err para falhas de transporte, protocolo e decodificação.

O leitor de bloco decodifica:

• 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), que é decodificado como seu T interno
• JSON e Object('json'), como colunas String com serialização de string (veja abaixo)

JSON e Object('json') são decodificados com serialização de string; defina output_format_native_write_json_as_string=1 na consulta. Cada linha suportada chega como um documento JSON em uma coluna CHC_COL_STRING. Construa a mesma estrutura com chc_build_string; o gravador emite o prefixo exigido pelo tipo convertido.

Variant, Dynamic, AggregateFunction ainda não são decodificados e retornam CHC_ERR_TYPE; converta-os para String no servidor como alternativa.