Add support of clickhouse as a destination

.gitignore

@@ -23,8 +23,8 @@ target/
 site/
 
 # venv files
-bin/
-lib/
+/bin/
+/lib/
 pyvenv.cfg
 
 # Log files

Cargo.toml

@@ -39,6 +39,7 @@ base64 = { version = "0.22.1", default-features = false }
 byteorder = { version = "1.5.0", default-features = false }
 bytes = { version = "1.10.1" }
 chrono = { version = "0.4.41", default-features = false }
+clickhouse = { version = "0.14", default-features = false }
 clap = { version = "4.5.42", default-features = false }
 config = { version = "0.14", default-features = false }
 configcat = { version = "0.1.3", default-features = false }
@@ -54,6 +55,7 @@ k8s-openapi = { version = "0.25.0", default-features = false }
 kube = { version = "1.1.0", default-features = false }
 metrics = { version = "0.24.2", default-features = false }
 metrics-exporter-prometheus = { version = "0.17.2", default-features = false }
+parking_lot = "0.12.5"
 parquet = { version = "57.0", default-features = false }
 pg_escape = { version = "0.1.1", default-features = false }
 pin-project-lite = { version = "0.2.16", default-features = false }

etl-api/Cargo.toml

@@ -17,7 +17,7 @@ name = "etl-api"
 [dependencies]
 etl = { workspace = true }
 etl-config = { workspace = true, features = ["utoipa", "supabase"] }
-etl-destinations = { workspace = true, features = ["bigquery", "iceberg"] }
+etl-destinations = { workspace = true, features = ["bigquery", "iceberg", "clickhouse"] }
 etl-postgres = { workspace = true, features = ["replication"] }
 etl-telemetry = { workspace = true }
 

etl-api/src/configs/destination.rs

@@ -34,6 +34,22 @@ pub enum FullApiDestinationConfig {
         #[serde(skip_serializing_if = "Option::is_none")]
         connection_pool_size: Option<usize>,
     },
+    ClickHouse {
+        /// ClickHouse HTTP(S) endpoint URL
+        #[schema(example = "http://test:8123")]
+        #[serde(deserialize_with = "crate::utils::trim_string")]
+        url: String, //TODO: use url type instead
+        /// ClickHouse user name
+        #[schema(example = "foo")]
+        #[serde(deserialize_with = "crate::utils::trim_string")]
+        user: String,
+        /// ClickHouse password (omit for passwordless access)
+        password: Option<SerializableSecretString>,
+        /// ClickHouse target database
+        #[schema(example = "my_db")]
+        #[serde(deserialize_with = "crate::utils::trim_string")]
+        database: String,
+    },
     Iceberg {
         #[serde(flatten)]
         config: FullApiIcebergConfig,
@@ -56,6 +72,17 @@ impl From<StoredDestinationConfig> for FullApiDestinationConfig {
                 max_staleness_mins,
                 connection_pool_size: Some(connection_pool_size),
             },
+            StoredDestinationConfig::ClickHouse {
+                url,
+                user,
+                password,
+                database,
+            } => Self::ClickHouse {
+                url,
+                user,
+                password,
+                database,
+            },
             StoredDestinationConfig::Iceberg { config } => match config {
                 StoredIcebergConfig::Supabase {
                     project_ref,
@@ -107,6 +134,12 @@ pub enum StoredDestinationConfig {
         max_staleness_mins: Option<u16>,
         connection_pool_size: usize,
     },
+    ClickHouse {
+        url: String, //TODO: use url type instead
+        user: String,
+        password: Option<SerializableSecretString>,
+        database: String,
+    },
     Iceberg {
         config: StoredIcebergConfig,
     },
@@ -128,6 +161,17 @@ impl StoredDestinationConfig {
                 max_staleness_mins,
                 connection_pool_size,
             },
+            Self::ClickHouse {
+                url,
+                user,
+                password,
+                database,
+            } => DestinationConfig::ClickHouse {
+                url,
+                user,
+                password: password.map(|s| s.into()),
+                database,
+            },
             Self::Iceberg { config } => match config {
                 StoredIcebergConfig::Supabase {
                     project_ref,
@@ -187,6 +231,17 @@ impl From<FullApiDestinationConfig> for StoredDestinationConfig {
                 connection_pool_size: connection_pool_size
                     .unwrap_or(DestinationConfig::DEFAULT_CONNECTION_POOL_SIZE),
             },
+            FullApiDestinationConfig::ClickHouse {
+                url,
+                user,
+                password,
+                database,
+            } => Self::ClickHouse {
+                url,
+                user,
+                password,
+                database,
+            },
             FullApiDestinationConfig::Iceberg { config } => match config {
                 FullApiIcebergConfig::Supabase {
                     project_ref,
@@ -255,6 +310,24 @@ impl Encrypt<EncryptedStoredDestinationConfig> for StoredDestinationConfig {
                     connection_pool_size,
                 })
             }
+            Self::ClickHouse {
+                url,
+                user,
+                password,
+                database,
+            } => {
+                let encrypted_password = match password {
+                    Some(p) => Some(encrypt_text(p.expose_secret().to_owned(), encryption_key)?),
+                    None => None,
+                };
+
+                Ok(EncryptedStoredDestinationConfig::ClickHouse {
+                    url,
+                    user,
+                    password: encrypted_password,
+                    database,
+                })
+            }
             Self::Iceberg { config } => match config {
                 StoredIcebergConfig::Supabase {
                     project_ref,
@@ -326,6 +399,12 @@ pub enum EncryptedStoredDestinationConfig {
         #[serde(default = "default_connection_pool_size")]
         connection_pool_size: usize,
     },
+    ClickHouse {
+        url: String, //TODO: use url type instead
+        user: String,
+        password: Option<EncryptedValue>,
+        database: String,
+    },
     Iceberg {
         #[serde(flatten)]
         config: EncryptedStoredIcebergConfig,
@@ -427,6 +506,27 @@ impl Decrypt<StoredDestinationConfig> for EncryptedStoredDestinationConfig {
                     })
                 }
             },
+            EncryptedStoredDestinationConfig::ClickHouse {
+                url,
+                user,
+                password,
+                database,
+            } => {
+                let password = match password {
+                    Some(p) => Some(SerializableSecretString::from(decrypt_text(
+                        p,
+                        encryption_key,
+                    )?)),
+                    None => None,
+                };
+
+                Ok(StoredDestinationConfig::ClickHouse {
+                    url,
+                    user,
+                    password,
+                    database,
+                })
+            }
         }
     }
 }