feat!(insert): RowBinaryWithNamesAndTypes

benches/mocked_insert.rs

@@ -51,7 +51,7 @@ async fn run_insert(client: Client, addr: SocketAddr, iters: u64) -> Result<Dura
     let _server = common::start_server(addr, serve).await;
 
     let start = Instant::now();
-    let mut insert = client.insert("table")?;
+    let mut insert = client.insert("table").await?;
 
     for _ in 0..iters {
         insert.write(&SomeRow::sample()).await?;
@@ -70,15 +70,15 @@ async fn run_inserter<const WITH_PERIOD: bool>(
     let _server = common::start_server(addr, serve).await;
 
     let start = Instant::now();
-    let mut inserter = client.inserter("table")?.with_max_rows(iters);
+    let mut inserter = client.inserter("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?;
     }
 

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("l2_book_log").unwrap();
+    let mut insert = client.insert("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(table_name)?;
+    let mut insert = client.insert(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(table_name)?;
+    let mut insert = client.insert(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(table_name)?;
+    let mut insert = client.insert(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(table_name)?;
+    let mut insert = client.insert(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(table_name)?;
+    let mut insert = client.insert(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(table_name)?;
+    let mut insert = client.insert(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_log")?;
+    let mut insert = client.insert("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(TABLE_NAME)?
+        .inserter(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(TABLE_NAME)?
+        .inserter(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("who cares")?;
+    let mut insert = client.insert("who cares").await?;
     for row in data {
         insert.write(row).await?;
     }

examples/session_id.rs

@@ -41,7 +41,7 @@ async fn main() -> Result<()> {
         i: i32,
     }
 
-    let mut insert = client.insert(table_name)?;
+    let mut insert = client.insert(table_name).await?;
     insert.write(&MyRow { i: 42 }).await?;
     insert.end().await?;
 

examples/usage.rs

@@ -29,7 +29,7 @@ async fn ddl(client: &Client) -> Result<()> {
 }
 
 async fn insert(client: &Client) -> Result<()> {
-    let mut insert = client.insert("some")?;
+    let mut insert = client.insert("some").await?;
     for i in 0..1000 {
         insert.write(&MyRow { no: i, name: "foo" }).await?;
     }
@@ -42,12 +42,12 @@ async fn insert(client: &Client) -> Result<()> {
 #[cfg(feature = "inserter")]
 async fn inserter(client: &Client) -> Result<()> {
     let mut inserter = client
-        .inserter("some")?
+        .inserter("some")
         .with_max_rows(100_000)
         .with_period(Some(std::time::Duration::from_secs(15)));
 
     for i in 0..1000 {
-        inserter.write(&MyRow { no: i, name: "foo" })?;
+        inserter.write(&MyRow { no: i, name: "foo" }).await?;
         inserter.commit().await?;
     }
 

src/insert.rs

@@ -1,24 +1,26 @@
-use std::{future::Future, marker::PhantomData, mem, panic, pin::Pin, time::Duration};
-
+use crate::headers::{with_authentication, with_request_headers};
+use crate::row_metadata::RowMetadata;
+use crate::rowbinary::{serialize_row_binary, serialize_with_validation};
+use crate::{
+    error::{Error, Result},
+    request_body::{ChunkSender, RequestBody},
+    response::Response,
+    row::{self, Row},
+    Client, Compression,
+};
 use bytes::{Bytes, BytesMut};
+use clickhouse_types::put_rbwnat_columns_header;
 use hyper::{self, Request};
 use replace_with::replace_with_or_abort;
 use serde::Serialize;
+use std::sync::Arc;
+use std::{future::Future, marker::PhantomData, mem, panic, pin::Pin, time::Duration};
 use tokio::{
     task::JoinHandle,
     time::{Instant, Sleep},
 };
 use url::Url;
 
-use crate::headers::{with_authentication, with_request_headers};
-use crate::{
-    error::{Error, Result},
-    request_body::{ChunkSender, RequestBody},
-    response::Response,
-    row::{self, Row},
-    rowbinary, Client, Compression,
-};
-
 // The desired max frame size.
 const BUFFER_SIZE: usize = 256 * 1024;
 // Threshold to send a chunk. Should be slightly less than `BUFFER_SIZE`
@@ -37,6 +39,7 @@ const_assert!(BUFFER_SIZE.is_power_of_two()); // to use the whole buffer's capac
 pub struct Insert<T> {
     state: InsertState,
     buffer: BytesMut,
+    row_metadata: Option<Arc<RowMetadata>>,
     #[cfg(feature = "lz4")]
     compression: Compression,
     send_timeout: Option<Duration>,
@@ -119,8 +122,7 @@ macro_rules! timeout {
 }
 
 impl<T> Insert<T> {
-    // TODO: remove Result
-    pub(crate) fn new(client: &Client, table: &str) -> Result<Self>
+    pub(crate) fn new(client: &Client, table: &str, row_metadata: Option<Arc<RowMetadata>>) -> Self
     where
         T: Row,
     {
@@ -129,9 +131,14 @@ impl<T> Insert<T> {
 
         // TODO: what about escaping a table name?
         // https://clickhouse.com/docs/en/sql-reference/syntax#identifiers
-        let sql = format!("INSERT INTO {}({}) FORMAT RowBinary", table, fields);
+        let format = if row_metadata.is_some() {
+            "RowBinaryWithNamesAndTypes"
+        } else {
+            "RowBinary"
+        };
+        let sql = format!("INSERT INTO {table}({fields}) FORMAT {format}");
 
-        Ok(Self {
+        Self {
             state: InsertState::NotStarted {
                 client: Box::new(client.clone()),
                 sql,
@@ -143,7 +150,8 @@ impl<T> Insert<T> {
             end_timeout: None,
             sleep: Box::pin(tokio::time::sleep(Duration::new(0, 0))),
             _marker: PhantomData,
-        })
+            row_metadata,
+        }
     }
 
     /// Sets timeouts for different operations.
@@ -194,6 +202,7 @@ impl<T> Insert<T> {
     /// socket.
     ///
     /// Close to:
+    ///
     /// ```ignore
     /// async fn write<T>(&self, row: &T) -> Result<usize>;
     /// ```
@@ -205,10 +214,11 @@ impl<T> Insert<T> {
     /// used anymore.
     ///
     /// # Panics
+    ///
     /// If called after the previous call that returned an error.
     pub fn write<'a>(&'a mut self, row: &T) -> impl Future<Output = Result<()>> + 'a + Send
     where
-        T: Serialize,
+        T: Serialize + Row,
     {
         let result = self.do_write(row);
 
@@ -224,7 +234,7 @@ impl<T> Insert<T> {
     #[inline(always)]
     pub(crate) fn do_write(&mut self, row: &T) -> Result<usize>
     where
-        T: Serialize,
+        T: Serialize + Row,
     {
         match self.state {
             InsertState::NotStarted { .. } => self.init_request(),
@@ -233,7 +243,10 @@ impl<T> Insert<T> {
         }?;
 
         let old_buf_size = self.buffer.len();
-        let result = rowbinary::serialize_into(&mut self.buffer, row);
+        let result = match &self.row_metadata {
+            Some(metadata) => serialize_with_validation(&mut self.buffer, row, metadata),
+            None => serialize_row_binary(&mut self.buffer, row),
+        };
         let written = self.buffer.len() - old_buf_size;
 
         if result.is_err() {
@@ -249,7 +262,7 @@ impl<T> Insert<T> {
     /// successfully, including all materialized views and quorum writes.
     ///
     /// NOTE: If it isn't called, the whole `INSERT` is aborted.
-    pub async fn end(mut self) -> Result<()> {
+    pub async fn end(&mut self) -> Result<()> {
         if !self.buffer.is_empty() {
             self.send_chunk().await?;
         }
@@ -264,7 +277,6 @@ impl<T> Insert<T> {
         // It's difficult to determine when allocations occur.
         // So, instead we control it manually here and rely on the system allocator.
         let chunk = self.take_and_prepare_chunk()?;
-
         let sender = self.state.sender().unwrap(); // checked above
 
         let is_timed_out = match timeout!(self, send_timeout, sender.send(chunk)) {
@@ -366,6 +378,11 @@ impl<T> Insert<T> {
         let handle =
             tokio::spawn(async move { Response::new(future, Compression::None).finish().await });
 
+        put_rbwnat_columns_header(
+            &self.row_metadata.as_ref().unwrap().columns,
+            &mut self.buffer,
+        )?;
+
         self.state = InsertState::Active { handle, sender };
         Ok(())
     }