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

Cargo.toml

@@ -237,8 +237,8 @@ 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.7", default-features = false }
-clap = { version = "4.0.12", default-features = false, features = ["derive", "error-context", "env", "help", "std", "string", "usage", "wrap_help"] }
+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 }

src/sinks/clickhouse/config.rs

@@ -1,5 +1,3 @@
-use std::collections::BTreeMap;
-use vector_config::configurable_component;
 use crate::{
     codecs::Transformer,
     config::{AcknowledgementsConfig, Input, SinkConfig, SinkContext},
@@ -13,11 +11,10 @@ use crate::{
     },
     tls::TlsConfig,
 };
+use std::collections::BTreeMap;
+use vector_config::configurable_component;
 
-use super::{
-    http_sink::build_http_sink,
-    native_sink::build_native_sink,
-};
+use super::{http_sink::build_http_sink, native::build_native_sink};
 
 /// Configuration for the `clickhouse` sink.
 #[configurable_component(sink("clickhouse"))]
@@ -33,16 +30,16 @@ pub struct ClickhouseConfig {
 
     /// The database that contains the table that data will be inserted into.
     pub database: Option<String>,
-    /// Enable using clickhouse native protocol rather than the default jsoneachrow over http.
+    /// 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,
-    /// table column names like {"col1":"String", "col_2":"Nullable(UInt16)", ...}
+    /// SQL table column definition. For example: {"col1":"String", "col_2":"Nullable(UInt16)", ...}
     #[serde(default)]
-    pub table_def: BTreeMap<String, String>,
+    pub sql_table_col_def: BTreeMap<String, String>,
 
     #[configurable(derived)]
     #[serde(default = "Compression::gzip_default")]
@@ -83,8 +80,7 @@ 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
+        // Determine the sink to build from the config.
         if !self.use_native_proto {
             build_http_sink(self, cx).await
         } else {
@@ -99,4 +95,14 @@ impl SinkConfig for ClickhouseConfig {
     fn acknowledgements(&self) -> &AcknowledgementsConfig {
         &self.acknowledgements
     }
-}
+}
+
+#[cfg(test)]
+mod tests {
+    use super::ClickhouseConfig;
+
+    #[test]
+    fn generate_config() {
+        crate::test_util::test_generate_config::<ClickhouseConfig>();
+    }
+}