chore: add query_id

dozer-sink-clickhouse/src/client.rs

@@ -56,11 +56,14 @@ impl ClickhouseClient {
         datasource_name: &str,
         fields: &[FieldDefinition],
         table_options: Option<ClickhouseTableOptions>,
+        query_id: Option<String>,
     ) -> Result<(), QueryError> {
         let mut client = self.pool.get_handle().await?;
         let ddl = get_create_table_query(datasource_name, fields, table_options);
         info!("Creating Clickhouse Table");
         info!("{ddl}");
+        let query_id = query_id.unwrap_or("".to_string());
+        let ddl = Query::new(ddl).id(query_id);
         client.execute(ddl).await?;
         Ok(())
     }
@@ -72,15 +75,19 @@ impl ClickhouseClient {
         query_id: Option<String>,
     ) -> Result<SqlResult, QueryError> {
         let mut client = self.pool.get_handle().await?;
-        // TODO: Include query_id in RowBinary protocol.
-        // https://github.com/suharev7/clickhouse-rs/issues/176
-        // https://github.com/ClickHouse/ClickHouse/blob/master/src/Client/Connection.cpp
-        // https://www.propeldata.com/blog/how-to-check-your-clickhouse-version
+        /*
+            TODO: Include query_id in RowBinary protocol.
+            https://github.com/suharev7/clickhouse-rs/issues/176
+            https://github.com/ClickHouse/ClickHouse/blob/master/src/Client/Connection.cpp
+            https://www.propeldata.com/blog/how-to-check-your-clickhouse-version
+        */
 
         let query = Query::new(&query).id(query_id.map_or("".to_string(), |q| q.to_string()));
-        // let query = query_id.map_or(query.to_string(), |id| {
-        //     format!("{0} settings log_comment = '{1}'", query, id)
-        // });
+        /*
+        let query = query_id.map_or(query.to_string(), |id| {
+            format!("{0} settings log_comment = '{1}'", query, id)
+        });
+        */
 
         let block = client.query(query).fetch_all().await?;
 

dozer-sink-clickhouse/src/sink.rs

@@ -58,6 +58,7 @@ impl ClickhouseSinkFactory {
                 &repl_metadata.table_name,
                 &repl_metadata.schema.fields,
                 Some(create_table_options),
+                None,
             )
             .await?;
 
@@ -105,6 +106,7 @@ impl SinkFactory for ClickhouseSinkFactory {
                     &config.sink_table_name,
                     &schema.fields,
                     self.config.create_table_options.clone(),
+                    None,
                 )
                 .await?;
         }

dozer-sink-clickhouse/src/tests.rs

@@ -72,6 +72,7 @@ use clickhouse_rs::{Block, Pool};
 use std::error::Error;
 
 #[tokio::test]
+#[ignore]
 async fn clickhouse_test() -> Result<(), Box<dyn Error>> {
     let uuid = "248c40d9-d1eb-47c4-8801-943dbab34df9";
     let database_url = "tcp://default@localhost:9000/query_test";
@@ -97,14 +98,5 @@ async fn clickhouse_test() -> Result<(), Box<dyn Error>> {
 
     let table = Query::new("payment").id(uuid);
     client.insert(table, block).await?;
-
-    // let block = client.query("SELECT * FROM payment").fetch_all().await?;
-
-    // for row in block.rows() {
-    //     let id: u32 = row.get("customer_id")?;
-    //     let amount: u32 = row.get("amount")?;
-    //     let name: Option<&str> = row.get("account_name")?;
-    //     println!("Found payment {}: {} {:?}", id, amount, name);
-    // }
     Ok(())
 }