Clickhouse add protos

nexus/analyzer/src/lib.rs

@@ -11,7 +11,7 @@ use pt::{
     flow_model::{FlowJob, FlowJobTableMapping, QRepFlowJob},
     peerdb_peers::{
         peer::Config, BigqueryConfig, DbType, EventHubConfig, MongoConfig, Peer, PostgresConfig,
-        S3Config, SnowflakeConfig, SqlServerConfig,
+        S3Config, SnowflakeConfig, SqlServerConfig,ClickhouseConfig
     },
 };
 use qrep::process_options;
@@ -803,6 +803,37 @@ fn parse_db_options(
             let config = Config::EventhubGroupConfig(eventhub_group_config);
             Some(config)
         }
+        DbType::Clickhouse => {
+            let s3_int = opts
+                .get("s3_integration")
+                .map(|s| s.to_string())
+                .unwrap_or_default();
+
+            let clickhouse_config = ClickhouseConfig {
+                host: opts.get("host").context("no host specified")?.to_string(),
+                port: opts
+                    .get("port")
+                    .context("no port specified")?
+                    .parse::<u32>()
+                    .context("unable to parse port as valid int")?,
+                user: opts
+                    .get("user")
+                    .context("no username specified")?
+                    .to_string(),
+                password: opts
+                    .get("password")
+                    .context("no password specified")?
+                    .to_string(),
+                database: opts
+                    .get("database")
+                    .context("no default database specified")?
+                    .to_string(),
+                metadata_schema: opts.get("metadata_schema").map(|s| s.to_string()),
+                s3_integration: s3_int,
+            };
+            let config = Config::ClickhouseConfig(clickhouse_config);
+            Some(config)
+        }        
     };
 
     Ok(config)

nexus/catalog/src/lib.rs

@@ -139,6 +139,11 @@ impl Catalog {
                     buf.reserve(config_len);
                     eventhub_group_config.encode(&mut buf)?;
                 }
+                Config::ClickhouseConfig(clickhouse_config) => {
+                    let config_len = clickhouse_config.encoded_len();
+                    buf.reserve(config_len);
+                    clickhouse_config.encode(&mut buf)?;
+                }                
             };
 
             buf
@@ -334,6 +339,12 @@ impl Catalog {
                     pt::peerdb_peers::EventHubGroupConfig::decode(options).context(err)?;
                 Ok(Some(Config::EventhubGroupConfig(eventhub_group_config)))
             }
+            Some(DbType::Clickhouse) => {
+                let err = format!("unable to decode {} options for peer {}", "clickhouse", name);
+                let clickhouse_config =
+                    pt::peerdb_peers::ClickhouseConfig::decode(options).context(err)?;
+                Ok(Some(Config::ClickhouseConfig(clickhouse_config)))
+            }            
             None => Ok(None),
         }
     }

protos/peers.proto

@@ -89,6 +89,16 @@ message S3Config {
   PostgresConfig metadata_db = 7;
 }
 
+message ClickhouseConfig{
+  string host = 1;
+  uint32 port = 2;
+  string user = 3;
+  string password = 4;
+  string database = 5;
+  optional string metadata_schema = 6;
+  string s3_integration = 7;
+}
+
 message SqlServerConfig {
   string server = 1;
   uint32 port = 2;
@@ -106,6 +116,7 @@ enum DBType {
   S3 = 5;
   SQLSERVER = 6;
   EVENTHUB_GROUP = 7;
+  CLICKHOUSE = 8;  
 }
 
 message Peer {
@@ -120,5 +131,6 @@ message Peer {
     S3Config s3_config = 8;
     SqlServerConfig sqlserver_config = 9;
     EventHubGroupConfig eventhub_group_config = 10;
+    ClickhouseConfig clickhouse_config = 11;
   }
 }

ui/components/PeerTypeComponent.tsx

@@ -22,6 +22,8 @@ export const DBTypeToGoodText = (ptype: DBType) => {
       return 'SQL Server';
     case DBType.MONGO:
       return 'MongoDB';
+    case DBType.CLICKHOUSE:
+      return 'Clickhouse';
     case DBType.UNRECOGNIZED:
       return 'Unrecognised';
   }