clickhouse-rs

A Block represents a collection of columns in ClickHouse, organized by rows. It is the primary container for data being sent to or received from the server. You can construct blocks manually, add columns, or push rows using a builder pattern.

Key capabilities:

• Column Access: Retrieve columns by index (usize) or by name (&str or String).
• Cell Access: Use .get::<T>(row, col) to retrieve a specific value from a cell, where T must implement FromSql.
• Row Iteration: Use .rows() to get an iterator over the rows in the block.
• Data Insertion: Use .push(row) with a RowBuilder (like the row! macro) to append data.
• Chunking: Use .chunks(n) to split a large block into smaller blocks of size n.

// Creating a block and adding columns
let block = Block::<Simple>::new()
    .column("hello_id", vec![5_u32, 6_u32])
    .column("value", vec!["lol", "zuz"]);

// Accessing a specific cell
let val: u32 = block.get(0, "hello_id").unwrap();

// Iterating over rows
for row in block.rows() {
    let id: u32 = row.get("hello_id").unwrap();
}

The client uses a URL schema to define connection parameters. The format is:

schema://user:password@host[:port]/database?param1=value1&...&paramN=valueN

Connection Parameters

Example URL: tcp://user:password@host:9000/clicks?compression=lz4&ping_timeout=42ms

The recommended way to interact with ClickHouse is to create a Pool and obtain a ClientHandle.

Workflow

1. Create a Pool: Initialize a Pool with a connection URL.
2. Get a Handle: Use pool.get_handle().await? to get a ClientHandle.
3. Execute DDL: Use client.execute(sql).await? for statements like CREATE TABLE.
4. Insert Data: Use client.insert(table, block).await? to insert a Block of data.
5. Query Data: Use client.query(sql).fetch_all().await? to retrieve results as a Block.
6. Iterate Rows: Use block.rows() to iterate through the returned rows and row.get(column_name)? to extract values.

Example

use clickhouse_rs::{Block, Pool, errors::Error};

#[tokio::main]
async fn main() -> Result<(), Error> {
    let ddl = "CREATE TABLE IF NOT EXISTS payment (customer_id UInt32, amount UInt32, account_name Nullable(FixedString(3))) Engine=Memory";

    let block = Block::new()
        .column("customer_id", vec![1_u32, 3, 5, 7, 9])
        .column("amount", vec![2_u32, 4, 6, 8, 10])
        .column("account_name", vec![Some("foo"), None, None, None, Some("bar")]);

    let pool = Pool::new("tcp://localhost:9000?compression=lz4");
    let mut client = pool.get_handle().await?;

    client.execute(ddl).await?;
    client.insert("payment", block).await?;

    let block = client.query("SELECT * FROM payment").fetch_all().await?;

    for row in block.rows() {
        let id: u32 = row.get("customer_id")?;
        let amount: u32 = row.get("amount")?;
        let name: Option<&str> = row.get("account_name")?;
        println!("Found payment {id}: {amount} {name:?}");
    }
    Ok(())
}

# use std::env;
# use clickhouse_rs::{Block, Pool, errors::Error};
#
# #[tokio::main]
# async fn main() -> Result<(), Error> {
#     let ddl = r"
#         CREATE TABLE IF NOT EXISTS payment (
#             customer_id  UInt32,
#             amount       UInt32,
#             account_name Nullable(FixedString(3))
#         ) Engine=Memory";
#
#     let block = Block::new()
#         .column("customer_id",  vec![1_u32,  3,  5,  7,  9])
#         .column("amount",       vec![2_u32,  4,  6,  8, 10])
#         .column("account_name", vec![Some("foo"), None, None, None, Some("bar")]);
#
#     let pool = Pool::new("tcp://localhost:9000?compression=lz4".into());
#
#     let mut client = pool.get_handle().await?;
#     client.execute(ddl).await?;
#     client.insert("payment", block).await?;
#     let block = client.query("SELECT * FROM payment").fetch_all().await?;
#
#     for row in block.rows() {
#         let id: u32             = row.get("customer_id")?;
#         let amount: u32         = row.get("amount")?;
#         let name: Option<&str>  = row.get("account_name")?;
#         println!("Found payment {id}: {amount} {name:?}");
#     }
#     Ok(())
# }

To interact with ClickHouse, use a Pool to manage connections. You can create a Block to prepare data for insertion, execute DDL statements, and fetch query results as blocks.

use clickhouse_rs::{Block, Pool};
use std::error::Error;

#[tokio::main]
async fn main() -> Result<(), Box<dyn Error>> {
    let database_url = "tcp://user:password@host:9000/db";
    let ddl = r"
        CREATE TABLE IF NOT EXISTS payment (
            customer_id  UInt32,
            amount       UInt32,
            account_name Nullable(FixedString(3))
        ) Engine=Memory";

    // Prepare data using Block
    let block = Block::new()
        .column("customer_id",  vec![1_u32,  3,  5,  7,  9])
        .column("amount",       vec![2_u32,  4,  6,  8, 10])
        .column("account_name", vec![Some("foo"), None, None, None, Some("bar")]);

    let pool = Pool::new(database_url);

    // Get a handle from the pool and perform operations
    let mut client = pool.get_handle().await?;
    client.execute(ddl).await?;
    client.insert("payment", block).await?;
    
    // Query data
    let block = client.query("SELECT * FROM payment").fetch_all().await?;

    // Iterate through rows and extract values
    for row in block.rows() {
        let id: u32             = row.get("customer_id")?;
        let amount: u32         = row.get("amount")?;
        let name: Option<&str>  = row.get("account_name")?;
        println!("Found payment {}: {} {:?}", id, amount, name);
    }

    Ok(())
}

The Pool configuration is derived from the Options object passed to Pool::new(). You can control the connection lifecycle using the following settings:

• pool_min: The minimum number of idle connections to maintain in the pool (defaults to 5 if not specified).
• pool_max: The maximum number of total connections allowed (defaults to 10 if not specified).
• addr: The primary connection address.
• alt_hosts: A list of alternative hosts for load balancing/failover. The pool uses a round-robin approach via get_addr() to distribute connections across these hosts.

Note: Ensure you are using the binary protocol port (typically 9000). If you attempt to connect via port 8123 (the HTTP/text protocol), a warning will be issued as clickhouse-rs is designed for the binary protocol.

use clickhouse_rs::{Pool, Options};
use std::str::FromStr;

// Example with custom pool limits and multiple hosts
let options = Options::from_str("tcp://host1:9000?alt_hosts=host2:9000,host3:9000&pool_min=6&pool_max=12")
    .unwrap();
let pool = Pool::new(options);

The client connects using a URL with the following schema:

schema://user:password@host[:port]/database?param1=value1&...&paramN=valueN

Available Parameters

Example URL: tcp://user:password@host:9000/clicks?compression=lz4&ping_timeout=42ms