Client de C de ClickHouse

clickhouse-c es un client de C solo de cabecera para el protocolo nativo de ClickHouse. El código fuente y la referencia de cada cabecera están en el repositorio de GitHub.

A diferencia de los clients de más alto nivel, este está diseñado deliberadamente para hacer muy poco por usted. La cabecera principal decodifica y codifica bloques en formato Native mediante un callback de E/S que usted proporciona. Usted gestiona el socket, el contexto de TLS, el asignador de memoria, los reintentos y la agrupación de conexiones. Eso hace que sea lo bastante pequeño como para integrarlo: incluir solo clickhouse.h no añade dependencias de enlazado más allá de libc.

Estos son no objetivos deliberados. Manéjelos en su aplicación o con una biblioteca complementaria:

• Protocolo HTTP. Envuelva libcurl directamente para la interfaz HTTP.
• Resolución de DNS, failover del endpoint, pooling de conexiones, reintentos y backoff.
• Ciclo de vida del contexto TLS. El backend de OpenSSL usa un SSL que usted ya ha conectado.
• Uso de hilos. Cada chc_client está diseñado para ejecutarse en un solo hilo.
• E/S asíncrona dentro de la biblioteca. El client con bloqueo llama a chc_io.read de forma síncrona. Para un client de bucle de eventos que no realiza ninguna E/S por sí mismo, use el client sin E/S.

clickhouse-c se distribuye como un conjunto plano de cabeceras. Cada cabecera incluye tanto las declaraciones como la implementación, protegidas por una macro centinela. Elija las cabeceras que necesite su compilación.

El decodificador lee nombres de tipo imprimibles del wire, por lo que deben codificarse como texto. ClickHouse los escribe como texto de forma predeterminada, pero fije esta configuración en sus consultas para que un perfil del servidor o de la sesión que la establezca en binario no haga fallar la decodificación:

output_format_native_encode_types_in_binary_format = 0

No hay ningún paquete que instalar, así que debes incorporar los archivos de cabecera a tu árbol del proyecto mediante un submódulo de Git o una copia. Exactamente una unidad de traducción define CHC_IMPLEMENTATION e incluye la implementación; todas las demás unidades incluyen los mismos archivos de cabecera solo para las declaraciones.

/* 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 o CHC_NO_ZSTD para clickhouse-compression.h a fin de eliminar las dependencias de lz4/zstd.

Para comunicarse con un servidor de ClickHouse, configure el socket usted mismo, envuélvalo en un chc_io y páselo a chc_client_init, que ejecuta el handshake Hello de forma síncrona. La biblioteca no realiza DNS, failover, reconexión ni pooling; eso es responsabilidad de quien llama.

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 funciona en un solo hilo y encapsula una conexión. La biblioteca llama a los callbacks de chc_io de forma sincrónica; lo que esos callbacks hagan por debajo (epoll, io_uring, WaitLatchOrSocket) depende de usted.

Envíe la consulta y luego lea los paquetes hasta CHC_PKT_END_OF_STREAM. Use chc_client_send_query_ex para incluir la configuración requerida del servidor; la versión básica de chc_client_send_query envía una lista de settings vacía y hereda los valores predeterminados del 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;
}

Las excepciones del servidor llegan como paquetes CHC_PKT_EXCEPTION, no como un valor de retorno distinto de OK de chc_client_recv_packet. Solo los fallos de transporte devuelven un valor distinto de OK. El primer paquete CHC_PKT_DATA de un resultado es un bloque de encabezado que describe el esquema con cero filas; después vienen los bloques de datos. chc_packet_clear libera el bloque o la excepción del paquete; establezca primero esos campos en null en el paquete para asumir la propiedad en su lugar.

Los bloques son orientados a columna. Cada columna tiene una disposición física, devuelta por chc_column_layout, sobre la que debes hacer dispatch; su tipo declarado proviene de chc_block_column_type. Las disposiciones compuestas se anidan, así que leer un Nullable(Array(String)) implica desempaquetar el Nullable, recorrer los desplazamientos del array y luego segmentar los datos de la cadena.

Un lector para columnas numéricas simples, de cadena y 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;
    }
}

Los datos CHC_COL_FIXED usan little-endian en la transmisión; en hosts big-endian debes invertir tú mismo el orden de bytes de los enteros de varios bytes. Los desplazamientos y las claves de LowCardinality ya se convierten al orden de bytes del host durante la decodificación. UUIDs se representan como dos mitades UInt64 little-endian, IPv4 es un entero little-endian de 4 bytes, e IPv6 está en orden de bytes de red. Los ticks de DateTime64 están en UTC; la zona horaria del tipo es solo metadata.

Al ingestar desde un origen no confiable, llama a chc_column_validate en cada columna antes de recorrerla. chc_block_read no valida invariantes entre campos, como los desplazamientos de arrays y las claves de LowCardinality, por lo que un bloque falsificado podría leer más allá de los límites de la columna interna.

Cree columnas con las funciones auxiliares chc_build_*, añádalas a un chc_block_builder y luego páselo a chc_client_send_data. El constructor usa almacenamiento proporcionado por quien llama y almacena punteros en lugar de copiar, por lo que el almacenamiento, los árboles de columnas, los tipos, los nombres y los slabs deben seguir siendo válidos hasta después del envío. Un INSERT envía la consulta, espera el bloque de encabezado del server, envía uno o más bloques de datos y luego envía un bloque vacío para finalizar el flujo.

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 toma n_rows * elem_size bytes little-endian; chc_build_string toma desplazamientos de fin acumulativos exclusivos en el orden de bytes del host sobre un slab compactado. Las funciones auxiliares devuelven nodos de columna por valor. Anídelos para que coincidan con el tipo: por ejemplo, pase un nodo fijo o de cadena a chc_build_nullable, pase ese resultado a chc_build_array y agregue la raíz del array. Tuple, LowCardinality, Map y las columnas geo usan el mismo árbol: Map es Array(Tuple(K, V)).

Todas las columnas de un bloque deben tener el mismo número de filas de nivel superior. El writer comprueba que el árbol coincida con el tipo de ClickHouse analizado, pero quien llama debe dimensionar el almacenamiento de chc_block_col para cada append. También puede agregar directamente una columna decodificada desde chc_block_column para volver a codificarla, o llamar a chc_block_write_cols con un array de chc_block_col para omitir el constructor. Pasar el constructor a través de chc_client_send_data en lugar de chc_block_write, de nivel más bajo, permite que el client establezca las opciones del bloque a partir de la revision negociada y aplique compresión.

Pase un modo de compresión y un codec preconfigurado en chc_client_opts. El client descomprime los paquetes Data entrantes y comprime los salientes. El encabezado de compresión incluye adaptadores LZ4 y ZSTD; cada inicialización solo completa sus propios slots, así que llame a ambas para admitir cualquiera de los dos.

#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 una biblioteca de compresión para la que el proyecto no incluye un binding, define tú mismo un chc_codec; la vtable está declarada en clickhouse-compression.h.

clickhouse-openssl.h proporciona un backend chc_io sobre SSL_read/SSL_write. Usted gestiona OpenSSL: la biblioteca nunca crea un SSL_CTX, verifica certificados, configura SNI ni llama a SSL_connect / SSL_shutdown. Cuando se activa chc_io.read, el handshake ya debe haberse completado.

#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 y otras implementaciones con TLS activado usan el protocolo nativo en el puerto 9440. Ambos backends aceptan una callback opcional check_cancel, que se consulta entre lecturas, y un tiempo límite de lectura mediante chc_openssl_io_set_deadline / chc_posix_io_set_deadline.

clickhouse-async.h es una variante sin E/S del client TCP para bucles de eventos. Nunca toca un socket: envías los bytes que has recibido y extraes los bytes que quiere enviar, gestionando epoll, io_uring o WaitLatchOrSocket por tu cuenta. Las opciones, los tipos de paquete y el constructor de bloques son los mismos que en el client con bloqueo.

chc_async_client_init no realiza ninguna E/S y no puede bloquearse. El handshake se ejecuta después como una máquina de estados reanudable, al igual que cada envío y recepción. Cuando el análisis sobrepasa los bytes que has proporcionado, la llamada devuelve CHC_WOULD_BLOCK en lugar de bloquearse: envía más bytes entrantes y vuelve a llamar, y el analizador se reanuda a mitad de bloque.

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

Tu pump mueve bytes en ambos sentidos. Para la salida, chc_async_pending_out devuelve un puntero y una longitud de los bytes en cola; después de que el socket acepte algunos, llama a chc_async_consume_out con esa cantidad; una escritura parcial está bien. Para la entrada, pasa las lecturas del socket a chc_async_submit. Los envíos nunca bloquean ni ejercen contrapresión, así que vigila la longitud de salida pendiente y deja de enviar cuando crezca demasiado.

Hay un driver funcional de liburing en test/test_async_uring.c.

Cada punto de entrada usa una chc_alloc vtable, por lo que la asignación se basa en el mecanismo que utiliza el sistema anfitrión.

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 y llame a chc_alloc_stdlib() para usar un asignador de memoria estándar basado en malloc.

Las funciones devuelven CHC_OK (0) o un código CHC_ERR_* distinto de cero. El código es el valor de retorno; un chc_err asignado en la pila del llamador contiene el mensaje legible para humanos. La biblioteca nunca asigna un error en el 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;

Los errores de consulta del servidor no son fallos chc_err. Llegan por el flujo de paquetes como CHC_PKT_EXCEPTION, con el code, display_text y stack_trace del servidor. Reserve la comprobación de chc_err para fallos de transporte, protocolo y decodificación.

El lector de bloques 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 se decodifica como su T interno
• JSON y Object('json'), como columnas String con serialización de cadenas (consulte a continuación)

JSON y Object('json') se decodifican con serialización de cadenas; establezca output_format_native_write_json_as_string=1 en la consulta. Cada fila compatible llega como un documento JSON en una columna CHC_COL_STRING. Cree la misma estructura con chc_build_string; el escritor emite el prefijo requerido por el tipo procesado.

Variant, Dynamic, AggregateFunction todavía no se decodifican y devuelven CHC_ERR_TYPE; conviértalos a String en el servidor como alternativa.