#liuhuo23/clickhouse-driver

#Overview

• Uses the native ClickHouse HTTP interface (default port 8123).
• Responses parsed as TabSeparatedWithNamesAndTypes so column names and types come back automatically — no manual type mapping on the client side.
• Single unified API: execute_query(sql, params?) covers SELECT, DDL, and inline-VALUES INSERT.
• Named-parameter substitution via ClickHouse's {name: Type} placeholder syntax — values are passed as param_<key>=<value> URL parameters and substituted server-side (quoted and escaped automatically).
• Proper error handling via MoonBit try-catch with a DbError suberror (ServerError / ConnectionError).
• Simple per-request connections by default — no persistent state to leak; opt into keep-alive reuse with conn.pool(size) for high-QPS workloads.
• Compatible with ClickHouse 22.x+ (the HTTP interface has been stable since 21.x).

#Installation

import {
  "moonbitlang/async@0.20.1",
}

import {
  "moonbitlang/async/http",
  "moonbitlang/core/buffer",
  "moonbitlang/core/encoding/base64",
  "moonbitlang/core/encoding/utf8",
  "liuhuo23/clickhouse-driver" @lib,
}

#Quick start

///|
async fn main {
  let conn = @lib.connect(
    host="127.0.0.1",
    port=8123,
    user="default",
    password="",
    database="default",
    client_name="my-app",
  )
  defer conn.close()

  // 1. Health check
  conn.ping()

  // 2. SELECT — columns come back automatically
  let result = conn.execute_query(
    "SELECT id, name FROM users WHERE created_at > {lo: DateTime} LIMIT {n: UInt32}",
    params=Map::from_array([("lo", "2024-01-01 00:00:00"), ("n", "10")]),
  )

  // 3. Inspect schema
  for col in result.columns {
    println(col.name + " : " + col.type_)
  }

  // 4. Iterate rows (each cell is a string)
  for row in result.rows {
    println(row.values)
  }

  // 5. Or convert to row maps keyed by column name
  for m in result.to_map() {
    println(m["name"])
  }

  // 6. INSERT via inline VALUES — all values come from params
  ignore(
    conn.execute_query(
      "INSERT INTO users (id, name) VALUES " +
      "({id: UInt32}, {name: String}), ({id2: UInt32}, {name2: String})",
      params=Map::from_array([
        ("id", "1"),
        ("name", "alice"),
        ("id2", "2"),
        ("name2", "bob"),
      ]),
    ),
  )
}

#API reference

#connect

pub fn connect(
  host~ : String,
  port~ : Int,
  user~ : String,
  password~ : String,
  database~ : String,
  client_name~ : String,
  timeout_ms? : Int = 0,
  https? : Bool = false,
  skip_verify? : Bool = false,
  compress? : Bool = false,
) -> Connection

#Connection

///|
pub struct Connection {
  host : String
  port : Int
  user : String
  password : String
  database : String
  client_name : String
  timeout_ms : Int
  https : Bool
  skip_verify : Bool
  compress : Bool
}

#Connection::ping

pub async fn ping(self : Connection) -> Unit raise

#Connection::execute_query

pub async fn execute_query(
  self : Connection,
  sql : String,
  params? : Map[String, String] = {},
) -> ResultSet raise

// No params
let r = conn.execute_query("SELECT version()")

// Single param
let r = conn.execute_query(
  "SELECT * FROM events WHERE id = {id: UInt64}",
  params=Map::from_array([("id", "42")]),
)

// Multiple params, used in INSERT VALUES
ignore(conn.execute_query(
  "INSERT INTO events (id, ts, msg) VALUES " +
  "({id: UInt64}, {ts: DateTime}, {msg: String})",
  params=Map::from_array([
    ("id", "1"),
    ("ts", "2024-01-01 00:00:00"),
    ("msg", "hello"),
  ]),
))

• DbError::ServerError(code, name, message) — the server returned a non-2xx HTTP response (syntax error, unknown table, permission denied, …).
• DbError::ConnectionError(String) — network / I/O error.

#Connection::insert

pub async fn insert(
  self : Connection,
  table : String,
  columns : Array[String],
  rows : Array[Array[String]],
  format? : InsertFormat = Tsv,
) -> Unit raise

conn.insert("users", ["id", "name"], [["1", "alice"], ["2", "bob"]])

#Connection::execute_stream

pub async fn execute_stream(
  self : Connection,
  sql : String,
  params? : Map[String, String] = {},
) -> ResultSetCursor raise

let cur = conn.execute_stream("SELECT * FROM events")
defer cur.close()
while true {
  match cur.next() {
    None => break
    Some(row) => println(row.values)
  }
}

#Connection pooling

let pool = conn.pool(size=4)   // up to 4 concurrent connections
defer pool.close()

pool.ping()                                    // same API as Connection
let r = pool.execute_query("SELECT count() FROM events")
pool.insert("events", ["id", "ts"], [["1", "2024-01-01 00:00:00"]])
let cur = pool.execute_stream("SELECT * FROM events")  // cursor returns the
defer cur.close()                                        // connection on close

• size bounds concurrent connections; extra requests wait for a free one.
• Connections that fail are discarded and replaced, never reused.
• pool.execute_stream returns its connection when the stream is fully drained; an abandoned stream drops the connection instead.
• pool.close() closes idle connections; in-flight ones are closed on return.

pub async fn ConnectionPool::ping(Self) -> Unit
pub async fn ConnectionPool::execute_query(Self, String, params? : Map[String, String]) -> ResultSet
pub async fn ConnectionPool::execute(Self, String, Array[String]) -> ResultSet
pub async fn ConnectionPool::insert(Self, String, Array[String], Array[Array[String]], format? : InsertFormat) -> Unit
pub async fn ConnectionPool::execute_stream(Self, String, params? : Map[String, String]) -> ResultSetCursor
pub fn ConnectionPool::max_size(Self) -> Int
pub fn ConnectionPool::close(Self) -> Unit

#Connection::cancel

pub async fn cancel(self : Connection) -> Unit

#Connection::close

pub fn close(self : Connection) -> Unit

let conn = @lib.connect(...)
defer conn.close()

#ResultSet

///|
pub struct ResultSet {
  columns : Array[Column]
  rows : Array[Row]
}

#ResultSet::to_map

pub fn to_map(self : ResultSet) -> Array[Map[String, String]]

for m in result.to_map() {
  let name = m.get_or_default("name", "")
  println(name)
}

#Row

///|
pub struct Row {
  values : Array[String]
}

#Column

///|
pub struct Column {
  name : String
  type_ : String
}

#Error handling

try {
  conn.execute_query("SELECT * FROM no_such_table")
} catch {
  @lib.DbError::ServerError(code~, name=_, message~) =>
    println("server error: code=" + code.to_string() + " " + message)
  @lib.DbError::ConnectionError(msg) =>
    println("connection error: " + msg)
  _ => println("other error")
}

#DbError suberror

///|
pub suberror DbError {
  ServerError(code~ : Int, name~ : String, message~ : String)
  ConnectionError(String)
} derive(Show)

#How it works

POST /?database=<db>&default_format=TabSeparatedWithNamesAndTypes
    &query=<url-encoded SQL>
    [&param_<key>=<url-encoded value>...]
HTTP/1.1
Host: <host>:<port>
Authorization: Basic <base64(user:password)>
X-ClickHouse-Client-Name: <client_name>
Content-Length: 0

<col1>\t<col2>\t<col3>
<Type1>\t<Type2>\t<Type3>
<val1>\t<val2>\t<val3>
<val4>\t<val5>\t<val6>
...

#ClickHouse transactions

• ReplacingMergeTree(version_column) — keeps the row with the largest version_column after merge.
• CollapsingMergeTree(sign_column) — uses a sign column (+1 insert, -1 cancel) to collapse pairs of rows on merge.
• VersionedCollapsingMergeTree(version, sign) — like CollapsingMergeTree but order-independent.
• SummingMergeTree / AggregatingMergeTree — for state-aggregation patterns.

#Limitations

1. No mid-query cancel — once a request is sent, the driver has no handle to cancel it. Drop the connection if you must abort.
2. Streaming reads are cursor-only — execute_stream() streams rows lazily, but execute_query() still buffers the full result (capped at 256 MB) because the driver needs the complete body for LZ4 decompression and text parsing.
3. Compression + streaming don't mix yet — execute_stream() raises ConnectionError when compress=true; use execute_query() for compressed responses.
4. Response body size limit — buffered responses are capped at 256 MB to avoid runaway memory. Queries returning more should use filters, aggregation, or execute_stream().
5. No insert progress / retries — insert() streams the body in one shot; on failure the whole request must be retried.

#Run the example CLI

# Default connection (127.0.0.1:8123, user `default`)
moon run cmd/main

# Point it at any ClickHouse via environment variables
CLICKHOUSE_HOST=127.0.0.1 CLICKHOUSE_PORT=38123 \
CLICKHOUSE_USER=default CLICKHOUSE_PASSWORD=barn \
moon run cmd/main

• ping() health check and close() / cancel() semantics
• execute_query() — SELECT / DDL with named {name: Type} params
• execute() — positional ? binding (inline VALUES)
• insert() — streamed bulk insert as the HTTP body, TSV and JSONEachRow formats
• execute_stream() — row-by-row cursor with immediate schema access
• to_map() — result rows as column-name maps
• Error handling — ServerError (missing table, syntax error) and ConnectionError caught with try-catch
• compress=true — LZ4-compressed responses decoded automatically
• timeout_ms — a slow query (SELECT sleep(3)) is cancelled after the configured timeout
• https=true — TLS connection (fails gracefully against a plain-HTTP server, succeeds against a TLS-enabled port)
• conn.pool(size) — pooled keep-alive connections with concurrent queries