ClickHouse Rust客户端

基于tokio的异步纯Rust ClickHouse客户端库

开发状态： alpha

在Linux x86-64 (ubuntu 20.04 LTS)，Windows 10上测试过。

原因

• 为了创建一个轻量级且健壮的ClickHouse驱动程序，快速的开源列式数据库
• 为了学习Rust并发和零成本抽象

支持的功能

• 基于tokio的异步引擎
• 原生ClickHouse协议
• LZ4压缩
• 持久连接池
• 简单的行到对象映射器
• Date | DateTime | DateTime64- 读写
• (U)Int(8|16|32|64) - 读写
• Float32 | Float64 - 读写
• UUID - 读写
• String | FixedString- 读写
• Ipv4 | Ipv6 - 读写
• Nullable(*) - 读写
• Decimal - 读写
• Enum8, Enum16 - 读写

用例

• 使用ClickHouse服务器支持的SQL语法进行查询
• 执行任意DDL命令
• 查询服务器状态
• 向ClickHouse服务器插入大量（可能持续）数据流
• 使用轮询方法进行负载均衡

快速入门

需要rust 1.42。

该包尚未在crates.io上发布。从主页git下载源代码

git module update --init --recursive

构建需要rust 1.41稳定版或nightly版，tokio-0.2.x。

• 将以下行添加到您的 Cargo.toml 文件的 dependencies 部分

 clickhouse-driver = { version="0.1.0-alpha.3", path="../path_to_package/clickhouse-driver"}
 clickhouse-driver-lz4 = { version="0.1.0", path="../path_to_package/lz4a"}
 clickhouse-driver-cthrs = { version="0.1.0", path="../path_to_package/cityhash-rs"}

• 在main.rs中添加使用说明

  extern crate clickhouse_driver;   
  use clickhouse_driver::prelude::*;

要连接到服务器，提供连接URL

tcp://username:password@localhost/database?paramname=paramvalue&...

例如

tcp://user:default@localhost/log?ping_timout=200ms&execute_timeout=5s&query_timeout=20s&pool_max=4&compression=lz4

支持的URL参数

• compression - 接受 'lz4' 或 'none'。lz4 - 快速高效的压缩方法。如果用于大数据块，可以显著减少传输数据的大小和时间。对于小数据，最好选择无压缩；

• connection_timeout - 建立连接的超时时间。默认为500ms；

• execute_timeout - 等待 execute 方法调用结果的超时时间。如果执行用于修改大表，可能需要很长时间才能完成。在这种情况下，设置适当的参数。在其他情况下，保留默认值（180秒）；

• query_timout - 在 查询 调用中等待服务器返回下一块数据的超时时间。注意：大数据查询可能需要很长时间。此超时要求仅在超时结束前接收一个数据块。默认值是 180 秒；

• insert_timeout - 等待 insert 调用结果的超时时间。如果服务器在 insert_timeout 结束前未收到消息，则 insert 方法调用将返回错误。由于插入数据处理是异步的，因此它不包括服务器块处理时间。默认值是 180 秒；

• ping_timout - 在 ping 响应前的等待时间。如果服务器在 ping_timeout 结束前不返回 pong 响应，则将认为主机不可用；
  ping_timeout 结束；

• retry_timeout - 如果服务器未返回，则在发送下一个 ping 之前等待的秒数；

• ping_before_query - 1（默认）或 0。如果设置此选项，则要求驱动程序在从连接池返回连接后检查 Clickhouse 服务器的责任；

• pool_min - 最小连接池大小。可以在池中保持的空闲连接数；默认值是 2；

• pool_max - 池可以发出的最大已建立连接数。如果在池达到最大且没有空闲连接的情况下需要新的连接，则此任务将被放入等待队列。默认值是 10；

• readonly - 0（默认）|1|2。0 - 允许所有命令。2 - select 查询和更改设置，1 - 只允许 select 查询；

• keepalive - keepalive TCP 选项；

• host - 代替主机(s)

所有超时参数接受整数 - 秒数。要在末尾指定毫秒超时，请添加 ms。示例

• 200ms（200 毫秒）
• 20（20 秒）
• 10s（10 秒）

示例

struct Blob {
    id: u64,
    url: String,
    date: ServerDate,
    client: Uuid,
    ip: Ipv4Addr,
    value: Decimal32,
}

impl Deserialize for Blob {
    fn deserialize(row: Row) -> errors::Result<Self> {
        let err = || errors::ConversionError::UnsupportedConversion;

        let id: u64 = row.value(0)?.ok_or_else(err)?;
        let url: &str = row.value(1)?.ok_or_else(err)?;
        let date: ServerDate = row.value(2)?.ok_or_else(err)?;
        let client: Uuid = row.value(3)?.ok_or_else(err)?;
        let ip = row.value(4)?.ok_or_else(err)?;
        let value: Decimal32 = row.value(5)?.ok_or_else(err)?;

        Ok(Blob {
            id,
            date,
            client,
            value,
            url: url.to_string(),
            ip,
        })
    }
}

#[tokio::main]
async fn main() -> Result<(), io::Error> {

    //    CREATE TABLE IF NOT EXISTS blob (
    //        id          UInt64,
    //        url         String,
    //        date        DateTime,
    //        client      UUID,
    //        ip          IPv4
    //    ) ENGINE=MergeTree PARTITION BY id ORDER BY date

    let database_url =
        env::var("DATABASE_URL").unwrap_or_else(|_| "tcp://localhost:9000?compression=lz4".into());

    let pool = Pool::create(database_url.as_str())?;
    {
        let mut conn = pool.connection().await?;
        conn.ping().await?;

        let mut result = conn
            .query("SELECT id, url, date, client, ip FROM blob WHERE id=150  ORDER BY date LIMIT 30000")
            .await?;

        while let Some(block) = result.next().await? {
            for blob in block.iter::<Blob>() {
                ...
            }
        }
    }

    Ok(())
}

• 有关更多示例，请参阅 clickhouse-driver/examples/ 目录*

已知问题和限制

• 不支持多维数组；
• 数组数据类型只读；
• LowCardinality - 只读且仅基于 String 类型；
• Insert 方法只支持有限的数据类型 insert 要求插入的数据与表列类型完全匹配
  • Int8(16|32|64) - i8(16|32|64)
  • UInt8(16|32|64) - u8(16|32|64)
  • Float32 - f32
  • Float64 - f64
  • Date - chrono::Date
  • DateTime - chrono::DateTime
  • UUID - Uuid
  • IPv4 - AddrIpv4
  • IPv6 - AddrIpv6
  • String - &str,String,或 &[u8]
  • Enum8|16 - &str 或 String。此外，还可以检索枚举索引的 i16 值。

路线图

• Array 列数据类型 - 读写；
• Tuple - 没有计划支持；
• AggregateFunction - 没有计划支持；
• LowCardinality - 添加写支持，扩展到 Date，DateTime 类型；
• Serde - 除了临时的序列化/反序列化接口外，还提供 Row 序列化/反序列化接口；
• TLS
• C-API？
• async_std 运行时；