clickhouse.rs

A typed client for ClickHouse.

Features

Uses serde for encoding/decoding rows.
Uses RowBinary encoding.
Provides API for selecting.
Provides API for infinite transactional (see below) inserting.
Provides API for watching live views.
Decompression (lz4, gzip, zlib and brotli). lz4 by default.

Usage

To use the crate, add this to your Cargo.toml:

[dependencies]
clickhouse = "0.7"

[dev-dependencies]
clickhouse = { version = "0.7", features = ["test-util"] }

Examples

See more examples.

Create `Client`

use clickhouse::Client;

let client = Client::default()
    .with_url("http://localhost:8123")
    .with_user("name")
    .with_password("123")
    .with_database("test");

Reuse created clients or clone them in order to reuse a connection pool.

Select rows

use serde::Deserialize;
use clickhouse::Row;

#[derive(Row, Deserialize)]
struct MyRow<'a> {
    no: u32,
    name: &'a str,
}

let mut cursor = client
    .query("SELECT ?fields FROM some WHERE no BETWEEN ? AND ?")
    .bind(500)
    .bind(504)
    .fetch::<MyRow<'_>>()?;

while let Some(row) = cursor.next().await? { .. }

Placeholder ?fields is replaced with no, name (fields of Row).
Placeholder ? is replaced with values in following bind() calls.
Convenient fetch_one::<Row>() and fetch_all::<Row>() can be used to get a first row or all rows correspondingly.
sql::Identifier can be used to bind table names.
soa-derive is useful for Nested types.

Insert a batch

let mut insert = client.insert("some")?;
insert.write(&Row { no: 0, name: "foo" }).await?;
insert.write(&Row { no: 1, name: "bar" }).await?;
insert.end().await?;

If end() isn't called the insertion will be aborted.
Rows are being sent progressively to spread network load.
ClickHouse inserts batches atomically only if all rows fit in the same partition and their number is less max_insert_block_size.
ch2rs is useful to generate a row type from ClickHouse.

Infinite inserting

let mut inserter = client.inserter("some")?
    .with_max_entries(150_000) // `250_000` by default
    .with_max_duration(Duration::from_secs(15)); // `10s` by default

inserter.write(&Row { no: 0, name: "foo" }).await?;
inserter.write(&Row { no: 1, name: "bar" }).await?;
let stats = inserter.commit().await?;
if stats.entries > 0 {
    println!("{} entries ({} transactions) have been inserted",
        stats.entries, stats.transactions);
}

Inserter ends an active insert in commit() if thresholds (max_entries, max_duration) are reached.
The interval between ending active inserts is biased (±10% of max_duration) to avoid load spikes by parallel inserters.
All rows between commit() calls are inserted in the same INSERT statement.
Do not forget to flush if you want yo terminate inserting:

inserter.end().await?;

Perform DDL

client.query("DROP TABLE IF EXISTS some").execute().await?;

Live views

let mut cursor = client
    .watch("SELECT max(no), argMax(name, no) FROM some")
    .fetch::<Row<'_>>()?;

let (version, row) = cursor.next().await?.unwrap();
println!("live view updated: version={}, row={:?}", version, row);

// Use `only_events()` to iterate over versions only.
let mut cursor = client.watch("some_live_view").limit(20).only_events().fetch()?;
println!("live view updated: version={:?}", cursor.next().await?);

This code uses or creates if not exists a temporary live view named lv_{sha1(query)} to reuse the same live view by parallel watchers.
You can specify a name instead of a query.

Testing

The crate provides utils for mocking CH server, both for SELECTs and INSERTs.

The functionality can be enabled with the test-util feature. Use it only in dev-dependencies.

See the example.

let4be/clickhouse.rs