feat!(insert): RowBinaryWithNamesAndTypes

Cargo.toml

@@ -158,6 +158,9 @@ criterion = "0.6"
 serde = { version = "1.0.106", features = ["derive"] }
 tokio = { version = "1.0.1", features = ["full", "test-util"] }
 hyper = { version = "1.1", features = ["server"] }
+indexmap = { version = "2.10.0", features = ["serde"] }
+linked-hash-map = { version = "0.5.6", features = ["serde_impl"] }
+fxhash = { version = "0.2.1" }
 serde_bytes = "0.11.4"
 serde_json = "1"
 serde_repr = "0.1.7"

README.md

@@ -137,7 +137,7 @@ struct MyRow {
     name: String,
 }
 
-let mut insert = client.insert("some")?;
+let mut insert = client.insert::<MyRow>("some").await?;
 insert.write(&MyRow { no: 0, name: "foo".into() }).await?;
 insert.write(&MyRow { no: 1, name: "bar".into() }).await?;
 insert.end().await?;
@@ -158,14 +158,14 @@ insert.end().await?;
 Requires the `inserter` feature.
 
 ```rust,ignore
-let mut inserter = client.inserter("some")?
+let mut inserter = client.inserter::<MyRow>("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() })?;
+inserter.write(&MyRow { no: 0, name: "foo".into() }).await?;
+inserter.write(&MyRow { no: 1, name: "bar".into() }).await?;
 let stats = inserter.commit().await?;
 if stats.rows > 0 {
     println!(

benches/mocked_insert.rs

@@ -1,23 +1,57 @@
 use bytes::Bytes;
+use clickhouse::{error::Result, Client, Compression, Row};
+use clickhouse_types::{Column, DataTypeNode};
 use criterion::{criterion_group, criterion_main, Criterion, Throughput};
-use http_body_util::Empty;
+use futures_util::stream;
+use http_body_util::StreamBody;
+use hyper::body::{Body, Frame};
 use hyper::{body::Incoming, Request, Response};
 use serde::Serialize;
+use std::convert::Infallible;
 use std::hint::black_box;
-use std::net::SocketAddr;
+use std::net::{Ipv4Addr, SocketAddr, SocketAddrV4};
 use std::{
     future::Future,
     mem,
     time::{Duration, Instant},
 };
 
-use clickhouse::{error::Result, Client, Compression, Row};
-
 mod common;
 
-async fn serve(request: Request<Incoming>) -> Response<Empty<Bytes>> {
+async fn serve(
+    request: Request<Incoming>,
+    compression: Compression,
+    with_validation: bool,
+) -> Response<impl Body<Data = Bytes, Error = Infallible>> {
     common::skip_incoming(request).await;
-    Response::new(Empty::new())
+
+    let bytes = if with_validation {
+        let schema = vec![
+            Column::new("a".to_string(), DataTypeNode::UInt64),
+            Column::new("b".to_string(), DataTypeNode::Int64),
+            Column::new("c".to_string(), DataTypeNode::Int32),
+            Column::new("d".to_string(), DataTypeNode::UInt32),
+            Column::new("e".to_string(), DataTypeNode::UInt64),
+            Column::new("f".to_string(), DataTypeNode::UInt32),
+            Column::new("g".to_string(), DataTypeNode::UInt64),
+            Column::new("h".to_string(), DataTypeNode::Int64),
+        ];
+
+        let mut buffer = Vec::new();
+        clickhouse_types::put_rbwnat_columns_header(&schema, &mut buffer).unwrap();
+
+        match compression {
+            Compression::None => Bytes::from(buffer),
+            #[cfg(feature = "lz4")]
+            Compression::Lz4 => clickhouse::_priv::lz4_compress(&buffer).unwrap(),
+            _ => unreachable!(),
+        }
+    } else {
+        Bytes::new()
+    };
+
+    let stream = StreamBody::new(stream::once(async { Ok(Frame::data(bytes)) }));
+    Response::new(stream)
 }
 
 #[derive(Row, Serialize)]
@@ -47,11 +81,18 @@ impl SomeRow {
     }
 }
 
-async fn run_insert(client: Client, addr: SocketAddr, iters: u64) -> Result<Duration> {
-    let _server = common::start_server(addr, serve).await;
+const ADDR: SocketAddr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 6524));
+
+async fn run_insert(
+    client: Client,
+    iters: u64,
+    compression: Compression,
+    validation: bool,
+) -> Result<Duration> {
+    let _server = common::start_server(ADDR, move |req| serve(req, compression, validation)).await;
 
     let start = Instant::now();
-    let mut insert = client.insert::<SomeRow>("table")?;
+    let mut insert = client.insert::<SomeRow>("table").await?;
 
     for _ in 0..iters {
         insert.write(&SomeRow::sample()).await?;
@@ -64,65 +105,68 @@ async fn run_insert(client: Client, addr: SocketAddr, iters: u64) -> Result<Dura
 #[cfg(feature = "inserter")]
 async fn run_inserter<const WITH_PERIOD: bool>(
     client: Client,
-    addr: SocketAddr,
     iters: u64,
+    compression: Compression,
+    validation: bool,
 ) -> Result<Duration> {
-    let _server = common::start_server(addr, serve).await;
+    let _server = common::start_server(ADDR, move |req| serve(req, compression, validation)).await;
 
     let start = Instant::now();
-    let mut inserter = client.inserter::<SomeRow>("table")?.with_max_rows(iters);
+    let mut inserter = client.inserter::<SomeRow>("table").with_max_rows(iters);
 
     if WITH_PERIOD {
         // Just to measure overhead, not to actually use it.
         inserter = inserter.with_period(Some(Duration::from_secs(1000)));
     }
 
     for _ in 0..iters {
-        inserter.write(&SomeRow::sample())?;
+        inserter.write(&SomeRow::sample()).await?;
         inserter.commit().await?;
     }
 
     inserter.end().await?;
     Ok(start.elapsed())
 }
 
-fn run<F>(c: &mut Criterion, name: &str, port: u16, f: impl Fn(Client, SocketAddr, u64) -> F)
+fn run<F>(c: &mut Criterion, name: &str, f: impl Fn(Client, u64, Compression, bool) -> F)
 where
     F: Future<Output = Result<Duration>> + Send + 'static,
 {
-    let addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap();
     let runner = common::start_runner();
-
     let mut group = c.benchmark_group(name);
     group.throughput(Throughput::Bytes(mem::size_of::<SomeRow>() as u64));
-    group.bench_function("uncompressed", |b| {
-        b.iter_custom(|iters| {
-            let client = Client::default()
-                .with_url(format!("http://{addr}"))
-                .with_compression(Compression::None);
-            runner.run((f)(client, addr, iters))
-        })
-    });
-    #[cfg(feature = "lz4")]
-    group.bench_function("lz4", |b| {
-        b.iter_custom(|iters| {
-            let client = Client::default()
-                .with_url(format!("http://{addr}"))
-                .with_compression(Compression::Lz4);
-            runner.run((f)(client, addr, iters))
-        })
-    });
+    for validation in [true, false] {
+        #[allow(clippy::single_element_loop)]
+        for compression in [
+            Compression::None,
+            #[cfg(feature = "lz4")]
+            Compression::Lz4,
+        ] {
+            group.bench_function(
+                format!("validation={validation}/compression={compression:?}"),
+                |b| {
+                    b.iter_custom(|iters| {
+                        let client = Client::default()
+                            .with_url(format!("http://{ADDR}"))
+                            .with_compression(compression)
+                            .with_validation(validation);
+                        runner.run((f)(client, iters, compression, validation))
+                    })
+                },
+            );
+        }
+    }
     group.finish();
 }
 
 fn insert(c: &mut Criterion) {
-    run(c, "insert", 6543, run_insert);
+    run(c, "insert", run_insert);
 }
 
 #[cfg(feature = "inserter")]
 fn inserter(c: &mut Criterion) {
-    run(c, "inserter", 6544, run_inserter::<false>);
-    run(c, "inserter-period", 6545, run_inserter::<true>);
+    run(c, "inserter", run_inserter::<false>);
+    run(c, "inserter-period", run_inserter::<true>);
 }
 
 #[cfg(not(feature = "inserter"))]

benches/mocked_select.rs

@@ -21,11 +21,11 @@ mod common;
 async fn serve(
     request: Request<Incoming>,
     compression: Compression,
-    use_rbwnat: bool,
+    with_validation: bool,
 ) -> Response<impl Body<Data = Bytes, Error = Infallible>> {
     common::skip_incoming(request).await;
 
-    let maybe_schema = if use_rbwnat {
+    let maybe_schema = if with_validation {
         let schema = vec![
             Column::new("a".to_string(), DataTypeNode::UInt64),
             Column::new("b".to_string(), DataTypeNode::Int64),
@@ -79,8 +79,8 @@ fn prepare_chunk() -> Bytes {
 const ADDR: SocketAddr = SocketAddr::V4(SocketAddrV4::new(Ipv4Addr::LOCALHOST, 6523));
 
 fn select(c: &mut Criterion) {
-    async fn start_server(compression: Compression, use_rbwnat: bool) -> common::ServerHandle {
-        common::start_server(ADDR, move |req| serve(req, compression, use_rbwnat)).await
+    async fn start_server(compression: Compression, with_validation: bool) -> common::ServerHandle {
+        common::start_server(ADDR, move |req| serve(req, compression, with_validation)).await
     }
 
     let runner = common::start_runner();

benches/select_market_data.rs

@@ -41,21 +41,21 @@ async fn prepare_data() {
     client
         .query(
             r#"
-            CREATE TABLE IF NOT EXISTS l2_book_log
-            (
-                `instrument_id` UInt32 CODEC(T64, Default),
-                `received_time` DateTime64(9) CODEC(DoubleDelta, Default),
-                `exchange_time` Nullable(DateTime64(9)) CODEC(DoubleDelta, Default),
-                `sequence_no` Nullable(UInt64) CODEC(DoubleDelta, Default),
-                `trace_id` UInt64 CODEC(DoubleDelta, Default),
-                `side` Enum8('Bid' = 0, 'Ask' = 1),
-                `price` Float64,
-                `amount` Float64,
-                `is_eot` Bool
-            )
-            ENGINE = MergeTree
-            PRIMARY KEY (instrument_id, received_time)
-        "#,
+                CREATE TABLE IF NOT EXISTS l2_book_log
+                (
+                    `instrument_id` UInt32                      CODEC(T64,         Default),
+                    `received_time` DateTime64(9)               CODEC(DoubleDelta, Default),
+                    `exchange_time` Nullable(DateTime64(9))     CODEC(DoubleDelta, Default),
+                    `sequence_no`   Nullable(UInt64)            CODEC(DoubleDelta, Default),
+                    `trace_id`      UInt64                      CODEC(DoubleDelta, Default),
+                    `side`          Enum8('Bid' = 0, 'Ask' = 1),
+                    `price`         Float64,
+                    `amount`        Float64,
+                    `is_eot`        Bool
+                )
+                ENGINE = MergeTree
+                PRIMARY KEY (instrument_id, received_time)
+            "#,
         )
         .execute()
         .await
@@ -71,7 +71,7 @@ async fn prepare_data() {
         return;
     }
 
-    let mut insert = client.insert::<L2Update>("l2_book_log").unwrap();
+    let mut insert = client.insert::<L2Update>("l2_book_log").await.unwrap();
 
     for i in 0..10_000_000 {
         insert

examples/async_insert.rs

@@ -40,7 +40,7 @@ async fn main() -> Result<()> {
         .execute()
         .await?;
 
-    let mut insert = client.insert::<Event>(table_name)?;
+    let mut insert = client.insert::<Event>(table_name).await?;
     insert
         .write(&Event {
             timestamp: now(),

examples/clickhouse_cloud.rs

@@ -42,7 +42,7 @@ async fn main() -> clickhouse::error::Result<()> {
         .execute()
         .await?;
 
-    let mut insert = client.insert::<Data>(table_name)?;
+    let mut insert = client.insert::<Data>(table_name).await?;
     insert
         .write(&Data {
             id: 42,

examples/data_types_derive_containers.rs

@@ -37,7 +37,7 @@ async fn main() -> Result<()> {
         .execute()
         .await?;
 
-    let mut insert = client.insert::<Row>(table_name)?;
+    let mut insert = client.insert::<Row>(table_name).await?;
     insert.write(&Row::new()).await?;
     insert.end().await?;
 

examples/data_types_derive_simple.rs

@@ -80,7 +80,7 @@ async fn main() -> Result<()> {
         .execute()
         .await?;
 
-    let mut insert = client.insert::<Row>(table_name)?;
+    let mut insert = client.insert::<Row>(table_name).await?;
     insert.write(&Row::new()).await?;
     insert.end().await?;
 

examples/data_types_new_json.rs

@@ -49,7 +49,7 @@ async fn main() -> Result<()> {
         .to_string(),
     };
 
-    let mut insert = client.insert::<Row>(table_name)?;
+    let mut insert = client.insert::<Row>(table_name).await?;
     insert.write(&row).await?;
     insert.end().await?;
 

examples/data_types_variant.rs

@@ -42,7 +42,7 @@ async fn main() -> Result<()> {
         .execute()
         .await?;
 
-    let mut insert = client.insert::<MyRow>(table_name)?;
+    let mut insert = client.insert::<MyRow>(table_name).await?;
     let rows_to_insert = get_rows();
     for row in rows_to_insert {
         insert.write(&row).await?;

examples/enums.rs

@@ -50,7 +50,7 @@ async fn main() -> Result<()> {
         Error = 4,
     }
 
-    let mut insert = client.insert::<Event>("event_log")?;
+    let mut insert = client.insert::<Event>("event_log").await?;
     insert
         .write(&Event {
             timestamp: now(),

examples/inserter.rs

@@ -22,7 +22,7 @@ struct MyRow {
 // In other words, this pattern is applicable for ETL-like tasks.
 async fn dense(client: &Client, mut rx: Receiver<u32>) -> Result<()> {
     let mut inserter = client
-        .inserter::<MyRow>(TABLE_NAME)?
+        .inserter::<MyRow>(TABLE_NAME)
         // We limit the number of rows to be inserted in a single `INSERT` statement.
         // We use small value (100) for the example only.
         // See documentation of `with_max_rows` for details.
@@ -32,7 +32,7 @@ async fn dense(client: &Client, mut rx: Receiver<u32>) -> Result<()> {
         .with_max_bytes(1_048_576);
 
     while let Some(no) = rx.recv().await {
-        inserter.write(&MyRow { no })?;
+        inserter.write(&MyRow { no }).await?;
         inserter.commit().await?;
     }
 
@@ -47,7 +47,7 @@ async fn dense(client: &Client, mut rx: Receiver<u32>) -> Result<()> {
 // Some rows are arriving one by one with delay, some batched.
 async fn sparse(client: &Client, mut rx: Receiver<u32>) -> Result<()> {
     let mut inserter = client
-        .inserter::<MyRow>(TABLE_NAME)?
+        .inserter::<MyRow>(TABLE_NAME)
         // Slice the stream into chunks (one `INSERT` per chunk) by time.
         // See documentation of `with_period` for details.
         .with_period(Some(Duration::from_millis(100)))
@@ -85,7 +85,7 @@ async fn sparse(client: &Client, mut rx: Receiver<u32>) -> Result<()> {
             Err(TryRecvError::Disconnected) => break,
         };
 
-        inserter.write(&MyRow { no })?;
+        inserter.write(&MyRow { no }).await?;
         inserter.commit().await?;
 
         // You can use result of `commit()` to get the number of rows inserted.

examples/mock.rs

@@ -19,7 +19,7 @@ async fn make_select(client: &Client) -> Result<Vec<SomeRow>> {
 }
 
 async fn make_insert(client: &Client, data: &[SomeRow]) -> Result<()> {
-    let mut insert = client.insert::<SomeRow>("who cares")?;
+    let mut insert = client.insert::<SomeRow>("who cares").await?;
     for row in data {
         insert.write(row).await?;
     }