feat(clickhouse sink): add query_settings to clickhouse sink (#22764)

pm5 · pront · thomasqueirozb · web-flow · commit f83b7e0ba321 · 2025-07-11T18:04:34.000Z
* feat(sink: clickhouse): add query_settings option to clickhouse sink

* Rename query settings struct

* config UX improvements

* Update changelog.d/22764_clickhouse_query_settings.feature.md

* Fix spelling

* cargo fmt

* regen docs

---------

Co-authored-by: Pavlos Rontidis &lt;pavlos.rontidis@gmail.com&gt;
Co-authored-by: Thomas &lt;thomas.schneider@datadoghq.com&gt;
diff --git a/changelog.d/22764_clickhouse_query_settings.feature.md b/changelog.d/22764_clickhouse_query_settings.feature.md
@@ -0,0 +1,3 @@
+Add a `query_settings` and `async_insert_settings` option in the `clickhouse` sink, which allows users to configure asynchronous inserts.
+
+authors: pm5
diff --git a/src/sinks/clickhouse/config.rs b/src/sinks/clickhouse/config.rs
@@ -114,6 +114,62 @@ pub struct ClickhouseConfig {
         skip_serializing_if = "crate::serde::is_default"
     )]
     pub acknowledgements: AcknowledgementsConfig,
+
+    #[configurable(derived)]
+    #[serde(default)]
+    pub query_settings: QuerySettingsConfig,
+}
+
+/// Query settings for the `clickhouse` sink.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Default)]
+#[serde(deny_unknown_fields)]
+pub struct QuerySettingsConfig {
+    /// Async insert-related settings.
+    #[serde(default)]
+    pub async_insert_settings: AsyncInsertSettingsConfig,
+}
+
+/// Async insert related settings for the `clickhouse` sink.
+#[configurable_component]
+#[derive(Clone, Copy, Debug, Default)]
+#[serde(deny_unknown_fields)]
+pub struct AsyncInsertSettingsConfig {
+    /// Sets `async_insert`, allowing ClickHouse to queue the inserted data and later flush to table in the background.
+    ///
+    /// If left unspecified, use the default provided by the `ClickHouse` server.
+    #[serde(default)]
+    pub enabled: Option<bool>,
+
+    /// Sets `wait_for`, allowing ClickHouse to wait for processing of asynchronous insertion.
+    ///
+    /// If left unspecified, use the default provided by the `ClickHouse` server.
+    #[serde(default)]
+    pub wait_for_processing: Option<bool>,
+
+    /// Sets 'wait_for_processing_timeout`, to control the timeout for waiting for processing asynchronous insertion.
+    ///
+    /// If left unspecified, use the default provided by the `ClickHouse` server.
+    #[serde(default)]
+    pub wait_for_processing_timeout: Option<u64>,
+
+    /// Sets `async_insert_deduplicate`, allowing ClickHouse to perform deduplication when inserting blocks in the replicated table.
+    ///
+    /// If left unspecified, use the default provided by the `ClickHouse` server.
+    #[serde(default)]
+    pub deduplicate: Option<bool>,
+
+    /// Sets `async_insert_max_data_size`, the maximum size in bytes of unparsed data collected per query before being inserted.
+    ///
+    /// If left unspecified, use the default provided by the `ClickHouse` server.
+    #[serde(default)]
+    pub max_data_size: Option<u64>,
+
+    /// Sets `async_insert_max_query_number`, the maximum number of insert queries before being inserted
+    ///
+    /// If left unspecified, use the default provided by the `ClickHouse` server.
+    #[serde(default)]
+    pub max_query_number: Option<u64>,
 }
 
 impl_generate_config_from_default!(ClickhouseConfig);
@@ -137,6 +193,7 @@ impl SinkConfig for ClickhouseConfig {
             date_time_best_effort: self.date_time_best_effort,
             insert_random_shard: self.insert_random_shard,
             compression: self.compression,
+            query_settings: self.query_settings,
         };
 
         let service: HttpService<ClickhouseServiceRequestBuilder, PartitionKey> =
diff --git a/src/sinks/clickhouse/integration_tests.rs b/src/sinks/clickhouse/integration_tests.rs
@@ -19,11 +19,13 @@ use vector_lib::event::{BatchNotifier, BatchStatus, BatchStatusReceiver, Event,
 use vector_lib::lookup::PathPrefix;
 use warp::Filter;
 
-use super::*;
 use crate::{
     codecs::{TimestampFormat, Transformer},
     config::{log_schema, SinkConfig, SinkContext},
-    sinks::util::{BatchConfig, Compression, TowerRequestConfig},
+    sinks::{
+        clickhouse::config::ClickhouseConfig,
+        util::{BatchConfig, Compression, TowerRequestConfig},
+    },
     test_util::{
         components::{run_and_assert_sink_compliance, SINK_TAGS},
         random_table_name, trace_init,
diff --git a/src/sinks/clickhouse/mod.rs b/src/sinks/clickhouse/mod.rs
@@ -9,10 +9,9 @@
 //!
 //! This sink only supports logs for now but could support metrics and traces as well in the future.
 
-mod config;
+pub mod config;
 #[cfg(all(test, feature = "clickhouse-integration-tests"))]
 mod integration_tests;
 mod request_builder;
 mod service;
 mod sink;
-pub use self::config::ClickhouseConfig;
diff --git a/src/sinks/clickhouse/service.rs b/src/sinks/clickhouse/service.rs
@@ -1,5 +1,6 @@
 //! Service implementation for the `Clickhouse` sink.
 
+use super::config::QuerySettingsConfig;
 use super::sink::PartitionKey;
 use crate::{
     http::{Auth, HttpError},
@@ -67,6 +68,7 @@ pub(super) struct ClickhouseServiceRequestBuilder {
     pub(super) date_time_best_effort: bool,
     pub(super) insert_random_shard: bool,
     pub(super) compression: Compression,
+    pub(super) query_settings: QuerySettingsConfig,
 }
 
 impl HttpServiceRequestBuilder<PartitionKey> for ClickhouseServiceRequestBuilder {
@@ -84,6 +86,7 @@ impl HttpServiceRequestBuilder<PartitionKey> for ClickhouseServiceRequestBuilder
             self.skip_unknown_fields,
             self.date_time_best_effort,
             self.insert_random_shard,
+            self.query_settings,
         )?;
 
         let auth: Option<Auth> = self.auth.clone();
@@ -107,6 +110,18 @@ impl HttpServiceRequestBuilder<PartitionKey> for ClickhouseServiceRequestBuilder
     }
 }
 
+fn append_param<T: ToString>(uri: &mut String, key: &str, value: Option<T>) {
+    if let Some(val) = value {
+        uri.push_str(&format!("{}={}&", key, val.to_string()));
+    }
+}
+fn append_param_bool(uri: &mut String, key: &str, value: Option<bool>) {
+    if let Some(val) = value {
+        uri.push_str(&format!("{}={}&", key, if val { 1 } else { 0 }));
+    }
+}
+
+#[allow(clippy::too_many_arguments)]
 fn set_uri_query(
     uri: &Uri,
     database: &str,
@@ -115,6 +130,7 @@ fn set_uri_query(
     skip_unknown: Option<bool>,
     date_time_best_effort: bool,
     insert_random_shard: bool,
+    query_settings: QuerySettingsConfig,
 ) -> crate::Result<Uri> {
     let query = url::form_urlencoded::Serializer::new(String::new())
         .append_pair(
@@ -135,19 +151,45 @@ fn set_uri_query(
     }
 
     uri.push_str("?input_format_import_nested_json=1&");
-    if let Some(skip_unknown) = skip_unknown {
-        if skip_unknown {
-            uri.push_str("input_format_skip_unknown_fields=1&");
-        } else {
-            uri.push_str("input_format_skip_unknown_fields=0&")
-        }
-    }
+    append_param_bool(&mut uri, "input_format_skip_unknown_fields", skip_unknown);
     if date_time_best_effort {
         uri.push_str("date_time_input_format=best_effort&")
     }
     if insert_random_shard {
         uri.push_str("insert_distributed_one_random_shard=1&")
     }
+    append_param_bool(
+        &mut uri,
+        "async_insert",
+        query_settings.async_insert_settings.enabled,
+    );
+    append_param_bool(
+        &mut uri,
+        "wait_for_async_insert",
+        query_settings.async_insert_settings.wait_for_processing,
+    );
+    append_param(
+        &mut uri,
+        "wait_for_async_insert_timeout",
+        query_settings
+            .async_insert_settings
+            .wait_for_processing_timeout,
+    );
+    append_param_bool(
+        &mut uri,
+        "async_insert_deduplicate",
+        query_settings.async_insert_settings.deduplicate,
+    );
+    append_param(
+        &mut uri,
+        "async_insert_max_data_size",
+        query_settings.async_insert_settings.max_data_size,
+    );
+    append_param(
+        &mut uri,
+        "async_insert_max_query_number",
+        query_settings.async_insert_settings.max_query_number,
+    );
     uri.push_str(query.as_str());
 
     uri.parse::<Uri>()
@@ -158,6 +200,7 @@ fn set_uri_query(
 #[cfg(test)]
 mod tests {
     use super::*;
+    use crate::sinks::clickhouse::config::*;
 
     #[test]
     fn encode_valid() {
@@ -169,6 +212,7 @@ mod tests {
             Some(false),
             true,
             false,
+            QuerySettingsConfig::default(),
         )
         .unwrap();
         assert_eq!(uri.to_string(), "http://localhost:80/?\
@@ -185,6 +229,7 @@ mod tests {
             Some(false),
             false,
             false,
+            QuerySettingsConfig::default(),
         )
         .unwrap();
         assert_eq!(uri.to_string(), "http://localhost:80/?\
@@ -200,6 +245,7 @@ mod tests {
             Some(true),
             true,
             false,
+            QuerySettingsConfig::default(),
         )
         .unwrap();
         assert_eq!(uri.to_string(), "http://localhost:80/?\
@@ -216,11 +262,38 @@ mod tests {
             None,
             true,
             false,
+            QuerySettingsConfig::default(),
+        )
+        .unwrap();
+        assert_eq!(uri.to_string(), "http://localhost:80/?\
+                                     input_format_import_nested_json=1&\
+                                     date_time_input_format=best_effort&\
+                                     query=INSERT+INTO+%22my_database%22.%22my_%5C%22table%5C%22%22+FORMAT+JSONAsObject");
+
+        let uri = set_uri_query(
+            &"http://localhost:80".parse().unwrap(),
+            "my_database",
+            "my_\"table\"",
+            Format::JsonAsObject,
+            None,
+            true,
+            false,
+            QuerySettingsConfig {
+                async_insert_settings: AsyncInsertSettingsConfig {
+                    enabled: Some(true),
+                    wait_for_processing: Some(true),
+                    wait_for_processing_timeout: Some(500),
+                    ..AsyncInsertSettingsConfig::default()
+                },
+            },
         )
         .unwrap();
         assert_eq!(uri.to_string(), "http://localhost:80/?\
                                      input_format_import_nested_json=1&\
                                      date_time_input_format=best_effort&\
+                                     async_insert=1&\
+                                     wait_for_async_insert=1&\
+                                     wait_for_async_insert_timeout=500&\
                                      query=INSERT+INTO+%22my_database%22.%22my_%5C%22table%5C%22%22+FORMAT+JSONAsObject");
     }
 
@@ -234,6 +307,7 @@ mod tests {
             Some(false),
             false,
             false,
+            QuerySettingsConfig::default(),
         )
         .unwrap_err();
     }
diff --git a/website/cue/reference/components/sinks/generated/clickhouse.cue b/website/cue/reference/components/sinks/generated/clickhouse.cue
@@ -337,6 +337,70 @@ generated: components: sinks: clickhouse: configuration: {
 		required:    false
 		type: bool: default: false
 	}
+	query_settings: {
+		description: "Query settings for the `clickhouse` sink."
+		required:    false
+		type: object: options: async_insert_settings: {
+			description: "Async insert-related settings."
+			required:    false
+			type: object: options: {
+				deduplicate: {
+					description: """
+						Sets `async_insert_deduplicate`, allowing ClickHouse to perform deduplication when inserting blocks in the replicated table.
+
+						If left unspecified, use the default provided by the `ClickHouse` server.
+						"""
+					required: false
+					type: bool: {}
+				}
+				enabled: {
+					description: """
+						Sets `async_insert`, allowing ClickHouse to queue the inserted data and later flush to table in the background.
+
+						If left unspecified, use the default provided by the `ClickHouse` server.
+						"""
+					required: false
+					type: bool: {}
+				}
+				max_data_size: {
+					description: """
+						Sets `async_insert_max_data_size`, the maximum size in bytes of unparsed data collected per query before being inserted.
+
+						If left unspecified, use the default provided by the `ClickHouse` server.
+						"""
+					required: false
+					type: uint: {}
+				}
+				max_query_number: {
+					description: """
+						Sets `async_insert_max_query_number`, the maximum number of insert queries before being inserted
+
+						If left unspecified, use the default provided by the `ClickHouse` server.
+						"""
+					required: false
+					type: uint: {}
+				}
+				wait_for_processing: {
+					description: """
+						Sets `wait_for`, allowing ClickHouse to wait for processing of asynchronous insertion.
+
+						If left unspecified, use the default provided by the `ClickHouse` server.
+						"""
+					required: false
+					type: bool: {}
+				}
+				wait_for_processing_timeout: {
+					description: """
+						Sets 'wait_for_processing_timeout`, to control the timeout for waiting for processing asynchronous insertion.
+
+						If left unspecified, use the default provided by the `ClickHouse` server.
+						"""
+					required: false
+					type: uint: {}
+				}
+			}
+		}
+	}
 	request: {
 		description: """
 			Middleware settings for outbound requests.
diff --git a/website/cue/reference/remap/functions/replace.cue b/website/cue/reference/remap/functions/replace.cue
@@ -72,7 +72,7 @@ remap: functions: replace: {
 			title: "Replace with capture groups when not set in the configuration file (use `$$num` in config files)"
 			source: #"""
 				# Note that in the context of Vector configuration files, an extra `$` escape character is required
-				# (i.e. `$$num`) to avoid interepreting `num` as an environment variable.
+				# (i.e. `$$num`) to avoid interpreting `num` as an environment variable.
 				replace("foo123bar", r'foo(?P<num>\d+)bar', "$num")
 				"""#
 			return: "123"

Original file line number	Diff line number	Diff line change
`@@ -0,0 +1,3 @@`
	`1`	+Add a `query_settings` and `async_insert_settings` option in the `clickhouse` sink, which allows users to configure asynchronous inserts.
	`2`	`+`
	`3`	`+authors: pm5`