[WIP] enhancement(sinks clickhouse): support native protocol

Cargo.toml

@@ -236,13 +236,16 @@ bollard = { version = "0.13.0", default-features = false, features = ["ssl", "ch
 bytes = { version = "1.2.1", default-features = false, features = ["serde"] }
 bytesize = { version = "1.1.0", default-features = false }
 chrono = { version = "0.4.22", default-features = false, features = ["serde"] }
+chrono-tz = { version = "0.6", default-features = false, features = ["serde"], optional = true }
 cidr-utils = { version = "0.5.8", default-features = false }
 clap = { version = "4.0.18", default-features = false, features = ["derive", "error-context", "env", "help", "std", "string", "usage", "wrap_help"] }
+clickhouse-rs = { git = "https://github.com/suharev7/clickhouse-rs", rev="e40016b", features = ["tokio_io", "tls"], optional = true }
 colored = { version = "2.0.0", default-features = false }
 csv = { version = "1.1", default-features = false }
 derivative = { version = "2.2.0", default-features = false }
 dirs-next = { version = "2.0.0", default-features = false, optional = true }
 dyn-clone = { version = "1.0.9", default-features = false }
+either = { version = "1.6.1", default-features = false, optional = true }
 encoding_rs = { version = "0.8.31", default-features = false, features = ["serde"] }
 enum_dispatch = { version = "0.3.8", default-features = false }
 exitcode = { version = "1.1.2", default-features = false }
@@ -662,7 +665,7 @@ sinks-azure_blob = ["dep:azure_core", "dep:azure_identity", "dep:azure_storage",
 sinks-azure_monitor_logs = []
 sinks-blackhole = []
 sinks-chronicle = []
-sinks-clickhouse = []
+sinks-clickhouse = ["dep:clickhouse-rs", "dep:chrono-tz", "dep:either", "dep:nom"]
 sinks-console = []
 sinks-datadog_archives = ["sinks-aws_s3", "sinks-azure_blob", "sinks-gcp"]
 sinks-datadog_events = []

src/sinks/clickhouse/config.rs

@@ -1,5 +1,3 @@
-use vector_config::configurable_component;
-
 use crate::{
     codecs::Transformer,
     config::{AcknowledgementsConfig, Input, SinkConfig, SinkContext},
@@ -13,8 +11,10 @@ use crate::{
     },
     tls::TlsConfig,
 };
+use std::collections::BTreeMap;
+use vector_config::configurable_component;
 
-use super::http_sink::build_http_sink;
+use super::{http_sink::build_http_sink, native::build_native_sink};
 
 /// Configuration for the `clickhouse` sink.
 #[configurable_component(sink("clickhouse"))]
@@ -30,10 +30,16 @@ pub struct ClickhouseConfig {
 
     /// The database that contains the table that data will be inserted into.
     pub database: Option<String>,
+    /// If true`, ClickHouse Native Protocol is used. Defaults to `false`, using `JSONEachRow` over HTTP.
+    #[serde(default)]
+    pub use_native_proto: bool,
 
     /// Sets `input_format_skip_unknown_fields`, allowing Clickhouse to discard fields not present in the table schema.
     #[serde(default)]
     pub skip_unknown_fields: bool,
+    /// SQL table column definition. For example: {"col1":"String", "col_2":"Nullable(UInt16)", ...}
+    #[serde(default)]
+    pub sql_table_col_def: BTreeMap<String, String>,
 
     #[configurable(derived)]
     #[serde(default = "Compression::gzip_default")]
@@ -74,9 +80,12 @@ impl_generate_config_from_default!(ClickhouseConfig);
 #[async_trait::async_trait]
 impl SinkConfig for ClickhouseConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
-        // later we can build different sink(http, native) here
-        // according to the clickhouseConfig
-        build_http_sink(self, cx).await
+        // Determine the sink to build from the config.
+        if !self.use_native_proto {
+            build_http_sink(self, cx).await
+        } else {
+            build_native_sink(self, cx).await
+        }
     }
 
     fn input(&self) -> Input {
@@ -87,3 +96,13 @@ impl SinkConfig for ClickhouseConfig {
         &self.acknowledgements
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::ClickhouseConfig;
+
+    #[test]
+    fn generate_config() {
+        crate::test_util::test_generate_config::<ClickhouseConfig>();
+    }
+}

src/sinks/clickhouse/mod.rs

@@ -2,4 +2,6 @@ mod config;
 mod http_sink;
 #[cfg(all(test, feature = "clickhouse-integration-tests"))]
 mod integration_tests;
-pub use self::config::ClickhouseConfig;
+mod native;
+
+pub use self::config::*;