clickhouse-rs

ClickHouse 数据库官方纯 Rust 类型客户端。

• 使用 serde 进行行编码/解码。
• 支持 serde 属性： skip_serializing, skip_deserializing, rename。
• 通过 HTTP 传输使用 RowBinary 编码。
  • 计划切换到 TCP 上的 Native。
• 支持 TLS（见下面的 native-tls 和 rustls-tls 功能）。
• 支持压缩和解压缩（LZ4 和 LZ4HC）。
• 提供选择 API。
• 提供插入 API。
• 提供无限事务性插入 API（见下文）。
• 提供实时视图监控 API。
• 提供单元测试的模拟。

注意： ch2rs 可以用来从 ClickHouse 生成行类型。

用法

要使用该 crate，请将以下内容添加到您的 Cargo.toml

[dependencies]
clickhouse = "0.12.2"

[dev-dependencies]
clickhouse = { version = "0.12.2", 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?);

请参阅 示例。

功能标志

• lz4（默认启用）——启用 Compression::Lz4 和 Compression::Lz4Hc(_) 变体。如果启用，除了 WATCH 之外的所有查询都将默认使用 Compression::Lz4。
• native-tls——通过 hyper-tls 支持使用 HTTPS 方案的 URL，该方案链接到 OpenSSL。
• rustls-tls——通过 hyper-rustls 支持使用 HTTPS 方案的 URL，该方案不链接到 OpenSSL。
• inserter——启用 client.inserter()。
• test-util——添加模拟。请参阅 示例。仅在 dev-dependencies 中使用。
• watch——启用 client.watch 功能。请参阅相应的部分以获取详细信息。
• uuid——将 serde::uuid 添加到与 uuid 包一起使用。
• time——将 serde::time 添加到与 time 包一起使用。

注意：通过 HTTPS URL 连接到 ClickHouse 时，必须启用 native-tls 或 rustls-tls 功能之一。如果两者都启用，则 rustls-tls 功能将具有优先级。

数据类型

• (U)Int(8|16|32|64|128) 映射到/从对应的 (u|i)(8|16|32|64|128) 类型或围绕它们的新类型。

• (U)Int256 不直接支持，但有 一个解决方案。

• Float(32|64) 映射到/从对应的 f(32|64) 或围绕它们的新类型。

• Decimal(32|64|128) 与相应的 i(32|64|128) 或其周围的 newtypes 进行映射。使用 fixnum 或其他有符号定点数的实现更为方便。

• Boolean 与 bool 或其周围的 newtypes 进行映射。

• String 与任何字符串或字节类型进行映射，例如 &str、&[u8]、String、Vec<u8> 或 SmartString。也支持 newtypes。为了存储字节，考虑使用 serde_bytes，因为它更高效。

  示例

  #[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(_) 目前还不支持。

• Enum(8|16) 使用 serde_repr 支持映射。

  示例

  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 通过使用 serde::uuid 映射到/从 uuid::Uuid。需要 uuid 功能。

  示例

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

• IPv6 映射到/从 std::net::Ipv6Addr。

• IPv4 通过使用 serde::ipv4 映射到/从 std::net::Ipv4Addr。

  示例

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

• Date 映射到/从 u16 或其周围的 newtypes，并代表自 1970-01-01 以来经过的天数。此外，通过使用 serde::time::date 支持使用 time::Date，这需要 time 功能。

  示例

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

• Date32 与 i32 或其新类型映射，表示自 1970-01-01 以来经过的天数。此外，使用 serde::time::date32 支持 time::Date，需要 time 功能。

  示例

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

• DateTime 与 u32 或其新类型映射，表示自 UNIX 纪元以来经过的秒数。此外，使用 serde::time::datetime 支持 time::OffsetDateTime，需要 time 功能。

  示例

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

• DateTime64(_) 与 i32 或其新类型映射，表示自 UNIX 纪元以来经过的时间。此外，使用 serde::time::datetime64::* 支持 time::OffsetDateTime，需要 time 功能。

  示例

  #[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, ...) 与 (A, B, ...) 或其新类型映射。

• Array(_) 与任何切片映射，例如 Vec<_>，&[_]。也支持新类型。

• Map(K, V) 的行为类似于 Array((K, V))。

• LowCardinality(_) 支持无缝映射。

• Nullable(_) 与 Option<_> 映射。对于 clickhouse::serde::* 辅助器添加 ::option。

  示例

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

• Nested 通过提供具有重命名的多个数组来支持。

  示例

  // 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 和 Geo。