diff --git a/changelog.d/25939_clickhouse_extra_query_params.enhancement.md b/changelog.d/25939_clickhouse_extra_query_params.enhancement.md new file mode 100644 index 0000000000000..f095ef47d057a --- /dev/null +++ b/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 diff --git a/src/sinks/clickhouse/config.rs b/src/sinks/clickhouse/config.rs index 419c5ea9e2a77..9773e05a92789 100644 --- a/src/sinks/clickhouse/config.rs +++ b/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, + #[configurable(derived)] #[serde(flatten)] pub confinement: ConfinementConfig, } +fn extra_query_params_examples() -> HashMap { + [ + ( + "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, } +/// 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 = diff --git a/src/sinks/clickhouse/service.rs b/src/sinks/clickhouse/service.rs index 3c8b034715396..3e82cdbdc4b6f 100644 --- a/src/sinks/clickhouse/service.rs +++ b/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, } impl HttpServiceRequestBuilder for ClickhouseServiceRequestBuilder { @@ -88,6 +91,7 @@ impl HttpServiceRequestBuilder for ClickhouseServiceRequestBuilder self.date_time_best_effort, self.insert_random_shard, self.query_settings, + &self.extra_query_params, )?; let auth: Option = 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, ) -> crate::Result { // 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())); + } } diff --git a/website/cue/reference/components/sinks/generated/clickhouse.cue b/website/cue/reference/components/sinks/generated/clickhouse.cue index 8f2f7e976c52e..4fe0dce850476 100644 --- a/website/cue/reference/components/sinks/generated/clickhouse.cue +++ b/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.