WIP feat: implement native format inserts over HTTP

Cargo.toml

@@ -196,6 +196,8 @@ arrow = "58.2.0"
 # Required by `record_batch!()`
 arrow-schema = "58.2.0"
 
+insta = "1.48.0"
+
 # Only used in testing and examples if the `opentelemetry` feature is enabled,
 # but dev-dependencies cannot be optional.
 tracing-subscriber = { version = "0.3.23", features = ["env-filter", "json"] }

src/error.rs

@@ -7,7 +7,7 @@ use std::{error::Error as StdError, fmt, io, result, str::Utf8Error};
 /// A result with a specified [`Error`] type.
 pub type Result<T, E = Error> = result::Result<T, E>;
 
-type BoxedError = Box<dyn StdError + Send + Sync>;
+pub(crate) type BoxedError = Box<dyn StdError + Send + Sync>;
 
 /// Represents all possible errors.
 #[derive(Debug, thiserror::Error)]

src/insert_native.rs

@@ -0,0 +1,47 @@
+use crate::native::Block;
+use crate::native::writer::BlockWriter;
+use crate::{Client, Compression, insert_formatted, sql};
+
+pub struct InsertNative {
+    writer: BlockWriter,
+}
+
+impl InsertNative {
+    pub(crate) fn new(client: &Client, table_name: &str, escape: bool) -> Self {
+        let mut sql = "INSERT INTO ".to_string();
+
+        if escape {
+            sql::escape::identifier(table_name, &mut sql).expect("error escaping table name");
+        } else {
+            sql.push_str(table_name);
+        }
+
+        sql.push_str(" FORMAT Native");
+
+        Self {
+            writer: BlockWriter::new(insert_formatted::InsertFormatted::new(
+                // FIXME: use HTTP body compression instead of block-level compression
+                &client.clone().with_compression(Compression::None),
+                sql,
+                Some(table_name),
+            )),
+        }
+    }
+
+    /// Send a block of data.
+    ///
+    /// # NOT Cancel Safe
+    /// If this `async` method is canceled (i.e. by dropping the resulting `Future`),
+    /// the insert is automatically aborted.
+    ///
+    /// This is because the block data is not sent in a single write, since that would require
+    /// copying it into a separate buffer. There is no way to resynchronize the stream
+    /// once a block has been partially sent. Resuming a write would corrupt the stream.
+    pub async fn write(&mut self, block: &Block) -> crate::Result<()> {
+        self.writer.write(block).await
+    }
+
+    pub async fn end(self) -> crate::Result<()> {
+        self.writer.end().await
+    }
+}

src/lib.rs

@@ -21,6 +21,7 @@ use tokio::sync::RwLock;
 pub mod error;
 pub mod insert;
 pub mod insert_formatted;
+pub mod insert_native;
 #[cfg(feature = "inserter")]
 pub mod inserter;
 pub mod native;
@@ -543,6 +544,14 @@ impl Client {
         insert_formatted::InsertFormatted::new(self, sql.into(), None)
     }
 
+    pub fn insert_native(&self, table_name: &str) -> insert_native::InsertNative {
+        insert_native::InsertNative::new(self, table_name, true)
+    }
+
+    pub fn insert_native_unescaped(&self, raw_table_name: &str) -> insert_native::InsertNative {
+        insert_native::InsertNative::new(self, raw_table_name, false)
+    }
+
     /// Starts a new SELECT/DDL query.
     pub fn query(&self, query: &str) -> query::Query {
         query::Query::new(self, query)