feat(clickhouse sink): add extra_query_params for arbitrary ClickHouse settings

changelog.d/25939_clickhouse_extra_query_params.enhancement.md

@@ -0,0 +1,3 @@
+Added `extra_query_params` configuration field to the ClickHouse sink, allowing arbitrary ClickHouse session settings to be forwarded as HTTP query parameters on every INSERT request (e.g. `deduplicate_blocks_in_dependent_materialized_views`, `insert_quorum`, `max_insert_block_size`).
+
+authors: valerypetrov

src/sinks/clickhouse/config.rs

@@ -1,5 +1,6 @@
 //! Configuration for the `Clickhouse` sink.
 
+use std::collections::HashMap;
 use std::fmt;
 
 use http::{Request, StatusCode, Uri};
@@ -152,11 +153,37 @@ pub struct ClickhouseConfig {
     #[serde(default)]
     pub query_settings: QuerySettingsConfig,
 
+    /// Additional ClickHouse query parameters appended to every INSERT request URL.
+    ///
+    /// Each key-value pair is percent-encoded and forwarded as a URL query parameter.
+    /// Use this to pass any ClickHouse session setting not directly exposed by this sink,
+    /// for example `deduplicate_blocks_in_dependent_materialized_views` or `insert_quorum`.
+    ///
+    /// Keys must not conflict with parameters already managed by Vector
+    /// (`query`, `param_database`, `param_table`, `input_format_import_nested_json`,
+    /// `input_format_skip_unknown_fields`, `date_time_input_format`,
+    /// `insert_distributed_one_random_shard`, and the `async_insert_*` family).
+    #[configurable(metadata(docs::additional_props_description = "A ClickHouse query parameter name-value pair."))]
+    #[configurable(metadata(docs::examples = "extra_query_params_examples()"))]
+    #[serde(default)]
+    pub extra_query_params: HashMap<String, String>,
+
     #[configurable(derived)]
     #[serde(flatten)]
     pub confinement: ConfinementConfig,
 }
 
+fn extra_query_params_examples() -> HashMap<String, String> {
+    [
+        (
+            "deduplicate_blocks_in_dependent_materialized_views".to_string(),
+            "0".to_string(),
+        ),
+        ("insert_quorum".to_string(), "2".to_string()),
+    ]
+    .into()
+}
+
 /// Query settings for the `clickhouse` sink.
 #[configurable_component]
 #[derive(Clone, Copy, Debug, Default)]
@@ -209,12 +236,40 @@ pub struct AsyncInsertSettingsConfig {
     pub max_query_number: Option<u64>,
 }
 
+/// Keys in `extra_query_params` that conflict with parameters Vector manages internally.
+const RESERVED_QUERY_PARAMS: &[&str] = &[
+    "query",
+    "param_database",
+    "param_table",
+    "input_format_import_nested_json",
+    "input_format_skip_unknown_fields",
+    "date_time_input_format",
+    "insert_distributed_one_random_shard",
+    "async_insert",
+    "wait_for_async_insert",
+    "wait_for_async_insert_timeout",
+    "async_insert_deduplicate",
+    "async_insert_max_data_size",
+    "async_insert_max_query_number",
+];
+
 impl_generate_config_from_default!(ClickhouseConfig);
 
 #[async_trait::async_trait]
 #[typetag::serde(name = "clickhouse")]
 impl SinkConfig for ClickhouseConfig {
     async fn build(&self, cx: SinkContext) -> crate::Result<(VectorSink, Healthcheck)> {
+        for key in self.extra_query_params.keys() {
+            if RESERVED_QUERY_PARAMS.contains(&key.as_str()) {
+                return Err(format!(
+                    "extra_query_params key `{}` is reserved and managed by Vector; \
+                     use the dedicated configuration field instead",
+                    key
+                )
+                .into());
+            }
+        }
+
         let mut config = self.clone();
         config.table = config
             .table
@@ -241,6 +296,7 @@ impl SinkConfig for ClickhouseConfig {
             insert_random_shard: this.insert_random_shard,
             compression: this.compression,
             query_settings: this.query_settings,
+            extra_query_params: this.extra_query_params.clone(),
         };
 
         let service: HttpService<ClickhouseServiceRequestBuilder, PartitionKey> =

src/sinks/clickhouse/service.rs

@@ -1,5 +1,7 @@
 //! Service implementation for the `Clickhouse` sink.
 
+use std::collections::HashMap;
+
 use bytes::Bytes;
 use http::{
     Request, StatusCode, Uri,
@@ -70,6 +72,7 @@ pub(super) struct ClickhouseServiceRequestBuilder {
     pub(super) insert_random_shard: bool,
     pub(super) compression: Compression,
     pub(super) query_settings: QuerySettingsConfig,
+    pub(super) extra_query_params: HashMap<String, String>,
 }
 
 impl HttpServiceRequestBuilder<PartitionKey> for ClickhouseServiceRequestBuilder {
@@ -88,6 +91,7 @@ impl HttpServiceRequestBuilder<PartitionKey> for ClickhouseServiceRequestBuilder
             self.date_time_best_effort,
             self.insert_random_shard,
             self.query_settings,
+            &self.extra_query_params,
         )?;
 
         let auth: Option<Auth> = self.auth.clone();
@@ -140,10 +144,12 @@ fn set_uri_query(
     date_time_best_effort: bool,
     insert_random_shard: bool,
     query_settings: QuerySettingsConfig,
+    extra_query_params: &HashMap<String, String>,
 ) -> crate::Result<Uri> {
     // Use ClickHouse query parameters with the Identifier type (introduced in 21.12) so
     // the server handles identifier quoting — no client-side escaping required.
-    let query = url::form_urlencoded::Serializer::new(String::new())
+    let mut serializer = url::form_urlencoded::Serializer::new(String::new());
+    serializer
         .append_pair(
             "query",
             &format!(
@@ -152,8 +158,11 @@ fn set_uri_query(
             ),
         )
         .append_pair("param_database", database)
-        .append_pair("param_table", table)
-        .finish();
+        .append_pair("param_table", table);
+    for (k, v) in extra_query_params {
+        serializer.append_pair(k, v);
+    }
+    let query = serializer.finish();
 
     let mut uri = uri.to_string();
     if !uri.ends_with('/') {
@@ -229,6 +238,7 @@ mod tests {
             true,
             false,
             QuerySettingsConfig::default(),
+            &HashMap::new(),
         )
         .unwrap();
         assert_eq!(
@@ -251,6 +261,7 @@ mod tests {
             false,
             false,
             QuerySettingsConfig::default(),
+            &HashMap::new(),
         )
         .unwrap();
         assert_eq!(
@@ -272,6 +283,7 @@ mod tests {
             true,
             false,
             QuerySettingsConfig::default(),
+            &HashMap::new(),
         )
         .unwrap();
         assert_eq!(
@@ -294,6 +306,7 @@ mod tests {
             true,
             false,
             QuerySettingsConfig::default(),
+            &HashMap::new(),
         )
         .unwrap();
         assert_eq!(
@@ -322,6 +335,7 @@ mod tests {
                     ..AsyncInsertSettingsConfig::default()
                 },
             },
+            &HashMap::new(),
         )
         .unwrap();
         assert_eq!(
@@ -350,6 +364,7 @@ mod tests {
                 false,
                 false,
                 QuerySettingsConfig::default(),
+                &HashMap::new(),
             )
             .unwrap();
             let p = parse_query_params(&uri);
@@ -400,7 +415,41 @@ mod tests {
             false,
             false,
             QuerySettingsConfig::default(),
+            &HashMap::new(),
         )
         .unwrap_err();
     }
+
+    #[test]
+    fn extra_query_params_are_appended() {
+        let mut extra = HashMap::new();
+        extra.insert(
+            "deduplicate_blocks_in_dependent_materialized_views".to_string(),
+            "0".to_string(),
+        );
+        extra.insert("insert_quorum".to_string(), "2".to_string());
+
+        let uri = set_uri_query(
+            &"http://localhost:80".parse().unwrap(),
+            "my_database",
+            "my_table",
+            Format::JsonEachRow,
+            None,
+            false,
+            false,
+            QuerySettingsConfig::default(),
+            &extra,
+        )
+        .unwrap();
+
+        let params = parse_query_params(&uri);
+        assert_eq!(
+            params.get("deduplicate_blocks_in_dependent_materialized_views"),
+            Some(&"0".to_string())
+        );
+        assert_eq!(params.get("insert_quorum"), Some(&"2".to_string()));
+        // Existing params are still present
+        assert!(params.contains_key("query"));
+        assert_eq!(params.get("param_database"), Some(&"my_database".to_string()));
+    }
 }

website/cue/reference/components/sinks/generated/clickhouse.cue

@@ -383,6 +383,32 @@ generated: components: sinks: clickhouse: configuration: {
 		required:    true
 		type: string: examples: ["http://localhost:8123"]
 	}
+	extra_query_params: {
+		description: """
+			Additional ClickHouse query parameters appended to every INSERT request URL.
+
+			Each key-value pair is percent-encoded and forwarded as a URL query parameter.
+			Use this to pass any ClickHouse session setting not directly exposed by this sink,
+			for example `deduplicate_blocks_in_dependent_materialized_views` or `insert_quorum`.
+
+			Keys must not conflict with parameters already managed by Vector
+			(`query`, `param_database`, `param_table`, `input_format_import_nested_json`,
+			`input_format_skip_unknown_fields`, `date_time_input_format`,
+			`insert_distributed_one_random_shard`, and the `async_insert_*` family).
+			"""
+		required: false
+		type: object: {
+			examples: [{
+				deduplicate_blocks_in_dependent_materialized_views: "0"
+				insert_quorum:                                      "2"
+			}]
+			options: "*": {
+				description: "A ClickHouse query parameter name-value pair."
+				required:    true
+				type: string: {}
+			}
+		}
+	}
 	format: {
 		description: """
 			Data format.