added option to insert to random shard (#20336)

* added option to insert to random shard

* added changelog and updated cue file

* fix spelling

* fix fragment type on changelog

* changelog suggestion

* added authors

* fix missing new line (please work already)

* fix missing parameters for testing

changelog.d/20335_added_support_for_insert_distributed_one_random_shard.feature.md

@@ -0,0 +1,3 @@
+The `clickhouse` sink now has a new configuration option, `insert_random_shard`, to tell Clickhouse to insert into a random shard (by setting `insert_distributed_one_random_shard`). See the Clickhouse [Distributed Table Engine docs](https://clickhouse.com/docs/en/engines/table-engines/special/distributed) for details.
+
+authors: rguleryuz

src/sinks/clickhouse/config.rs

@@ -79,6 +79,10 @@ pub struct ClickhouseConfig {
     #[serde(default)]
     pub date_time_best_effort: bool,
 
+    /// Sets `insert_distributed_one_random_shard`, allowing ClickHouse to insert data into a random shard when using Distributed Table Engine.
+    #[serde(default)]
+    pub insert_random_shard: bool,
+
     #[configurable(derived)]
     #[serde(default = "Compression::gzip_default")]
     pub compression: Compression,
@@ -129,6 +133,7 @@ impl SinkConfig for ClickhouseConfig {
             endpoint: endpoint.clone(),
             skip_unknown_fields: self.skip_unknown_fields,
             date_time_best_effort: self.date_time_best_effort,
+            insert_random_shard: self.insert_random_shard,
             compression: self.compression,
         };
 

src/sinks/clickhouse/service.rs

@@ -65,6 +65,7 @@ pub(super) struct ClickhouseServiceRequestBuilder {
     pub(super) endpoint: Uri,
     pub(super) skip_unknown_fields: bool,
     pub(super) date_time_best_effort: bool,
+    pub(super) insert_random_shard: bool,
     pub(super) compression: Compression,
 }
 
@@ -82,6 +83,7 @@ impl HttpServiceRequestBuilder<PartitionKey> for ClickhouseServiceRequestBuilder
             metadata.format,
             self.skip_unknown_fields,
             self.date_time_best_effort,
+            self.insert_random_shard,
         )?;
 
         let auth: Option<Auth> = self.auth.clone();
@@ -112,6 +114,7 @@ fn set_uri_query(
     format: Format,
     skip_unknown: bool,
     date_time_best_effort: bool,
+    insert_random_shard: bool,
 ) -> crate::Result<Uri> {
     let query = url::form_urlencoded::Serializer::new(String::new())
         .append_pair(
@@ -138,6 +141,9 @@ fn set_uri_query(
     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&")
+    }
     uri.push_str(query.as_str());
 
     uri.parse::<Uri>()
@@ -158,6 +164,7 @@ mod tests {
             Format::JsonEachRow,
             false,
             true,
+            false,
         )
         .unwrap();
         assert_eq!(uri.to_string(), "http://localhost:80/?\
@@ -172,6 +179,7 @@ mod tests {
             Format::JsonEachRow,
             false,
             false,
+            false,
         )
         .unwrap();
         assert_eq!(uri.to_string(), "http://localhost:80/?\
@@ -185,6 +193,7 @@ mod tests {
             Format::JsonAsObject,
             true,
             true,
+            false,
         )
         .unwrap();
         assert_eq!(uri.to_string(), "http://localhost:80/?\
@@ -203,6 +212,7 @@ mod tests {
             Format::JsonEachRow,
             false,
             false,
+            false,
         )
         .unwrap_err();
     }

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

@@ -202,6 +202,11 @@ base: components: sinks: clickhouse: configuration: {
 			}
 		}
 	}
+	insert_random_shard: {
+		description: "Sets `insert_distributed_one_random_shard`, allowing ClickHouse to insert data into a random shard when using Distributed Table Engine."
+		required:    false
+		type: bool: default: false
+	}
 	request: {
 		description: """
 			Middleware settings for outbound requests.