ClickHouse C client

埋め込み向けに設計された、ClickHouse ネイティブプロトコル用のヘッダーオンリーCクライアント。

clickhouse-c は、ClickHouse のネイティブプロトコル向けのヘッダーオンリーCクライアントです。 ソースコードと各ヘッダーのリファレンスは、GitHubリポジトリにあります。

上位レベルのクライアントとは異なり、このライブラリは意図的に多くのことを行いません。コアヘッダーは、ユーザーが提供する I/O コールバックを介して Native フォーマットの ブロック をデコードおよび エンコードします。socket、TLS コンテキスト、アロケータ、再試行、接続プーリングはユーザー側で管理します。そのため、埋め込みに適した 十分な小ささを実現しています。clickhouse.h をインクルードするだけで、リンク時の依存関係は libc 以外に発生しません。

このライブラリは現在も活発に開発されています。v1 では ClickHouse の主要な types をデコードします。 制限事項や不足している機能は、issue tracker から報告してください。 ただし、このライブラリでは設計上、あえて実装していない機能があることをご理解ください。

​

ライブラリが行わないこと

以下は、意図的に非目標としている項目です。これらはアプリケーション側、または関連ライブラリで対応してください。

• HTTP プロトコル。HTTP インターフェイス については、libcurl を直接ラップしてください。
• DNS 名前解決、エンドポイントのフェイルオーバー、接続プーリング、再試行、バックオフ。
• TLS コンテキストのライフサイクル。OpenSSL バックエンドは、すでに接続済みの SSL を使用します。
• スレッド化。各 chc_client は設計上シングルスレッドです。
• ライブラリ内部での非同期 I/O。ブロッキングクライアントは chc_io.read を同期的に呼び出します。ライブラリ自体では I/O を行わない イベントループクライアントが必要な場合は、I/O を行わないクライアント を使用してください。

​

ライブラリの構成

clickhouse-c はフラットなヘッダ群として提供されます。各ヘッダには、センチネルマクロで保護された宣言と実装の両方が含まれています。 ビルドに必要なヘッダを選択してください。

Header	Purpose	Link flags
clickhouse.h	コア: 型、エラー、アロケータ、I/O vtable、型名パーサー、ブロックリーダー、ライター	—
clickhouse-client.h	TCP パケットループ: Hello、Query、Data、EndOfStream、Exception、Progress、Pong	—
clickhouse-async.h	I/O を行わないクライアント: 呼び出し元からのバイト入力で駆動する同じパケットループ、ソケット不要	—
clickhouse-compression.h	圧縮フレームのレイアウト、CityHash128、codec のディスパッチ、LZ4/ZSTD アダプター	-llz4 -lzstd
clickhouse-posix-io.h	ブロッキング read(2)/write(2) 上で動作する I/O バックエンド	—
clickhouse-openssl.h	SSL_read/SSL_write 上で動作する I/O バックエンド	-lssl -lcrypto

​

必須のサーバー設定

デコーダーは wire 上の表示可能な型名を読み取るため、それらはテキストとしてエンコードされている必要があります。ClickHouse はデフォルトでそれらをテキストとして書き込みますが、この設定をバイナリにするサーバー設定やセッションプロファイル によってデコードが壊れないよう、クエリでこの設定を明示的に固定してください:

output_format_native_encode_types_in_binary_format = 0

​

プロジェクトへの追加

インストールするパッケージはないため、ヘッダーファイルは Git サブモジュールとして、またはコピーしてツリーに取り込んでください。 CHC_IMPLEMENTATION を定義して実装を取り込む翻訳単位は、必ず1つだけにしてください。 それ以外のすべての単位では、宣言のみを行うために同じヘッダーファイルをインクルードします。

/* 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_alloc_stdlib を使用するには、clickhouse.h をインクルードする前に CHC_PROVIDE_STDLIB_ALLOC を定義します。 clickhouse-compression.h で lz4/zstd への依存関係をなくすには、CHC_NO_LZ4 または CHC_NO_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 はシングルスレッドで動作し、1 つの connection を扱います。ライブラリは chc_io callbacks を同期的に呼び出します。これらの callbacks が内部で何を行うか (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 パケットとして到着し、chc_client_recv_packet から 非OKが返るわけではありません。非OKが返るのはトランスポート層の障害時だけです。結果の最初の CHC_PKT_DATA パケットは、スキーマを示す 0 行のヘッダーブロックで、その後にデータブロックが続きます。 chc_packet_clear はパケットのブロックまたは例外を解放します。代わりに所有権を取得する場合は、まずパケット上のそれらのフィールドを null にしてください。

​

カラムデータの読み取り

ブロックはカラム指向です。各カラムには chc_column_layout が返す物理レイアウトがあり、 それに応じて処理を分岐します。宣言された型は chc_block_column_type から取得します。複合レイアウトはネストされるため、 Nullable(Array(String)) を読み取る場合は、まず Nullable をアンラップし、配列の offsets をたどってから、 文字列データをスライスします。

Layout	Accessors
CHC_COL_FIXED	chc_column_fixed_data(c, &elem_size) — n_rows * elem_size 分のリトルエンディアンのバイト列
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 行あたり 1 バイト、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)；Dictionary の slot 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 データは、wire 上ではリトルエンディアンです。ビッグエンディアンのホストでは、複数バイトの 整数を自分でバイトスワップする必要があります。offsets と LowCardinality キーは、デコード時にすでにホストオーダーに変換されています。 UUIDs は 2 つのリトルエンディアン UInt64 の半分からなり、IPv4 は 4 バイトのリトルエンディアン整数、IPv6 は ネットワークバイトオーダーです。DateTime64 の tick は UTC で、型に含まれるタイムゾーンはメタデータにすぎません。

信頼できないピアから取り込む場合は、各カラムを走査する前に chc_column_validate を呼び出してください。 chc_block_read は、配列の offsets や LowCardinality キーのようなフィールド間の不変条件を検証しないため、 そうしないと偽造された block によって内部カラムの境界を越えて読み出されるおそれがあります。

​

データの挿入

chc_block_builder でブロックを作成し、それを chc_client_send_data に渡します。ビルダーは コピーするのではなくポインタを保持するため、カラムスラブは送信処理が完了するまで有効である必要があります。INSERT はクエリを送信し、 サーバーのヘッダーブロックを待機してから、1 つ以上のデータブロックを送信し、最後に空のブロックを送って ストリームを終了します。

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 バイトのリトルエンディアンデータを受け取ります。 chc_block_builder_append_string は、packed slab 上の累積的な排他的終端オフセットをホストバイトオーダーで受け取ります。 ビルダーを下位レベルの chc_block_write ではなく chc_client_send_data 経由にすることで、クライアントはネゴシエートされたリビジョンに基づいてブロックオプションを設定し、 圧縮を適用できます。

​

圧縮

chc_client_opts に、圧縮モードと設定済みの codec を渡します。クライアントは受信した データパケットを展開し、送信するパケットを圧縮します。圧縮ヘッダーには 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 は、SSL_read/SSL_write を利用する chc_io バックエンドを提供します。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 でネイティブプロトコルを使用します。どちらのバックエンドも、読み取りの合間にポーリングされる 省略可能な check_cancel コールバックと、 chc_openssl_io_set_deadline / chc_posix_io_set_deadline による読み取りデッドラインをサポートします。

​

I/O を行わないクライアント (非同期)

clickhouse-async.h は、イベントループ向け TCP クライアントの I/O を行わない版です。ソケットには一切触れません。 受信したバイト列を渡し、送信したいバイト列を取り出して、epoll、io_uring、または WaitLatchOrSocket は自分で駆動します。オプション、パケット型、ブロック builder は ブロッキング クライアントと同じです。

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 はバイトを双方向にやり取りします。Outbound では、chc_async_pending_out がキュー内のバイト列へのポインタと長さを 返します。ソケットがその一部を受け付けたら、そのバイト数を指定して chc_async_consume_out を呼び出してください。部分書き込みでも問題ありません。 Inbound では、ソケットから読み取ったデータを chc_async_submit に渡します。送信でブロックしたりバックプレッシャーがかかったりすることはないため、 pending-out の長さを監視し、大きくなりすぎたら送信の発行を止めてください。

動作する liburing ドライバーは test/test_async_uring.c にあります。

​

メモリとアロケータ

すべてのエントリポイントは chc_alloc の vtable を受け取るため、メモリ割り当てはホスト側で使われている方式に従います。

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;

clickhouse.h をインクルードする前に CHC_PROVIDE_STDLIB_ALLOC を定義し、標準の malloc ベースのアロケータを使用するには chc_alloc_stdlib() を呼び出します。

​

エラーとサーバー例外

関数は CHC_OK (0) または 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 を設定した場合にのみデコードされます。 各行は CHC_COL_STRING カラム内の 1 つの JSON ドキュメントとして渡されるため、文字列アクセサで読み取れます。 builder は chc_block_builder_append_json_string で同じ形式を書き込みます。その他の JSON シリアル化バージョンでは、設定名を示した CHC_ERR_TYPE が返されます。

Variant, Dynamic, AggregateFunction はまだデコードされず、CHC_ERR_TYPE を返します。 フォールバックとして、サーバー側でそれらを String にキャストしてください。