clickhouse.rs

A typed client for ClickHouse.

• Uses serde for encoding/decoding rows.
• Supports serde attributes: skip_serializing, skip_deserializing, rename.
• Uses RowBinary encoding over HTTP transport.
  • There are plans to switch to Native over TCP.
• Supports TLS.
• Supports compression and decompression (LZ4 and LZ4HC).
• Provides API for selecting.
• Provides API for inserting.
• Provides API for infinite transactional (see below) inserting.
• Provides API for watching live views.
• Provides mocks for unit testing.

Note: ch2rs is useful to generate a row type from ClickHouse.

Usage

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

[dependencies]
clickhouse = "0.11.6"

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

use clickhouse::Client;

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

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? { .. }

use serde::Serialize;
use clickhouse::Row;

#[derive(Row, Serialize)]
struct MyRow {
    no: u32,
    name: String,
}

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

let mut inserter = client.inserter("some")?
    .with_timeouts(Some(Duration::from_secs(5)), Some(Duration::from_secs(20)))
    .with_max_bytes(50_000_000)
    .with_max_rows(750_000)
    .with_period(Some(Duration::from_secs(15)));

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

inserter.end().await?;

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

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

See examples.

Feature Flags

• lz4 (enabled by default) — enables Compression::Lz4 and Compression::Lz4Hc(_) variants. If enabled, Compression::Lz4 is used by default for all queries except for WATCH.
• tls (enabled by default) — supports urls with the HTTPS schema.
• inserter — enables client.inserter().
• test-util — adds mocks. See the example. Use it only in dev-dependencies.
• watch — enables client.watch functionality. See the corresponding section for details.
• uuid — adds serde::uuid to work with uuid crate.
• time — adds serde::time to work with time crate.

Data Types

• (U)Int(8|16|32|64|128) maps to/from corresponding (u|i)(8|16|32|64|128) types or newtypes around them.

• (U)Int256 aren't supported directly, but there is a workaround for it.

• Float(32|64) maps to/from corresponding f(32|64) or newtypes around them.

• Decimal(32|64|128) maps to/from corresponding i(32|64|128) or newtypes around them. It's more convenient to use fixnum or another implementation of signed fixed-point numbers.

• Boolean maps to/from bool or newtypes around it.

• String maps to/from any string or bytes types, e.g. &str, &[u8], String, Vec<u8> or SmartString. Newtypes are also supported. To store bytes, consider using serde_bytes, because it's more efficient.

  Example

  #[derive(Row, Debug, Serialize, Deserialize)]
  struct MyRow<'a> {
      str: &'a str,
      string: String,
      #[serde(with = "serde_bytes")]
      bytes: Vec<u8>,
      #[serde(with = "serde_bytes")]
      byte_slice: &'a [u8],
  }
  

• FixedString(_) isn't supported yet.

• Enum(8|16) are supported using serde_repr.

  Example

  use serde_repr::{Deserialize_repr, Serialize_repr};
  
  #[derive(Row, Serialize, Deserialize)]
  struct MyRow {
      level: Level,
  }
  
  #[derive(Debug, Serialize_repr, Deserialize_repr)]
  #[repr(u8)]
  enum Level {
      Debug = 1,
      Info = 2,
      Warn = 3,
      Error = 4,
  }
  

• UUID maps to/from uuid::Uuid by using serde::uuid. Requires the uuid feature.

  Example

  #[derive(Row, Serialize, Deserialize)]
  struct MyRow {
      #[serde(with = "clickhouse::serde::uuid")]
      uuid: uuid::Uuid,
  }
  

• IPv6 maps to/from std::net::Ipv6Addr.

• IPv4 maps to/from std::net::Ipv4Addr by using serde::ipv4.

  Example

  #[derive(Row, Serialize, Deserialize)]
  struct MyRow {
      #[serde(with = "clickhouse::serde::ipv4")]
      ipv4: std::net::Ipv4Addr,
  }
  

• Date maps to/from u16 or a newtype around it and represents a number of days elapsed since 1970-01-01. Also, time::Date is supported by using serde::time::date, that requires the time feature.

  Example

  #[derive(Row, Serialize, Deserialize)]
  struct MyRow {
      days: u16,
      #[serde(with = "clickhouse::serde::time::date")]
      date: Date,
  }
  

• Date32 maps to/from i32 or a newtype around it and represents a number of days elapsed since 1970-01-01. Also, time::Date is supported by using serde::time::date32, that requires the time feature.

  Example

  #[derive(Row, Serialize, Deserialize)]
  struct MyRow {
      days: i32,
      #[serde(with = "clickhouse::serde::time::date32")]
      date: Date,
  }
  

• DateTime maps to/from u32 or a newtype around it and represents a number of seconds elapsed since UNIX epoch. Also, time::OffsetDateTime is supported by using serde::time::datetime, that requires the time feature.

  Example

  #[derive(Row, Serialize, Deserialize)]
  struct MyRow {
      ts: u32,
      #[serde(with = "clickhouse::serde::time::datetime")]
      dt: OffsetDateTime,
  }
  

• DateTime64(_) maps to/from i32 or a newtype around it and represents a time elapsed since UNIX epoch. Also, time::OffsetDateTime is supported by using serde::time::datetime64::*, that requires the time feature.

  Example

  #[derive(Row, Serialize, Deserialize)]
  struct MyRow {
      ts: i64, // elapsed s/us/ms/ns depending on `DateTime64(X)`
      #[serde(with = "clickhouse::serde::time::datetime64::secs")]
      dt64s: OffsetDateTime,  // `DateTime64(0)`
      #[serde(with = "clickhouse::serde::time::datetime64::millis")]
      dt64ms: OffsetDateTime, // `DateTime64(3)`
      #[serde(with = "clickhouse::serde::time::datetime64::micros")]
      dt64us: OffsetDateTime, // `DateTime64(6)`
      #[serde(with = "clickhouse::serde::time::datetime64::nanos")]
      dt64ns: OffsetDateTime, // `DateTime64(9)`
  }
  

• Typle(A, B, ...) maps to/from (A, B, ...) or a newtype around it.

• Array(_) maps to/from any slice, e.g. Vec<_>, &[_]. Newtypes are also supported.

• Map(K, V) behaves like Array((K, V)).

• LowCardinality(_) is supported seamlessly.

• Nullable(_) maps to/from Option<_>. For clickhouse::serde::* helpers add ::option.

  Example

  #[derive(Row, Serialize, Deserialize)]
  struct MyRow {
      #[serde(with = "clickhouse::serde::ipv4::option")]
      ipv4_opt: Option<Ipv4Addr>,
  }
  

• Nested is supported by providing multiple arrays with renaming.

  Example

  // CREATE TABLE test(items Nested(name String, count UInt32))
  #[derive(Row, Serialize, Deserialize)]
  struct MyRow {
      #[serde(rename = "items.name")]
      items_name: Vec<String>,
      #[serde(rename = "items.count")]
      items_count: Vec<u32>,
  }
  

• JSON and Geo aren't supported for now.

Mocking

The crate provides utils for mocking CH server and testing DDL, SELECT, INSERT and WATCH queries.

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

See the example.