format; clickhouse can run

Author: STWang

asap-common/.gitignore

@@ -5,6 +5,7 @@
 .vscode/
 
 dependencies/py/promql_utilities/promql_utilities.egg-info/
+dependencies/py/promql_utilities/build/
 dependencies/rs/**/target/
 
 tests/**/*.json

asap-query-engine/.gitignore

@@ -1 +1,2 @@
 target
+output/

asap-query-engine/src/main.rs

@@ -7,7 +7,7 @@ use tracing::{error, info};
 
 use sketch_core::config::{self, ImplMode};
 
-use query_engine_rust::data_model::enums::{InputFormat, LockStrategy, StreamingEngine};
+use query_engine_rust::data_model::enums::{InputFormat, LockStrategy, QueryProtocol, StreamingEngine};
 use query_engine_rust::drivers::AdapterConfig;
 use query_engine_rust::utils::file_io::{read_inference_config, read_streaming_config};
 use query_engine_rust::{
@@ -233,11 +233,24 @@ async fn main() -> Result<()> {
     ));
 
     // Setup Kafka consumer (equivalent to Python's kafka_thread)
+    // When deleting the DB, use a unique group ID so no prior committed offset exists
+    // and the consumer replays from the beginning of the topic.
+    let kafka_group_id = if args.delete_existing_db {
+        format!(
+            "query-engine-rust-{}",
+            std::time::SystemTime::now()
+                .duration_since(std::time::UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_secs()
+        )
+    } else {
+        "query-engine-rust".to_string()
+    };
     let kafka_config = KafkaConsumerConfig {
         broker: args.kafka_broker.clone(),
         topic: args.kafka_topic.clone(),
-        group_id: "query-engine-rust".to_string(),
-        auto_offset_reset: "beginning".to_string(),
+        group_id: kafka_group_id,
+        auto_offset_reset: "earliest".to_string(),
         input_format: args.input_format,
         decompress_json: args.decompress_json,
         batch_size: 1000,
@@ -328,11 +341,22 @@ async fn main() -> Result<()> {
     //    true,                                // Always forward (fallback for every query)
     //);
 
-    // Original Prometheus config (commented out temporarily):
-    let adapter_config = AdapterConfig::prometheus_promql(
-        args.prometheus_server.clone(),
-        args.forward_unsupported_queries,
-    );
+    let adapter_config = match args.query_language {
+        QueryLanguage::sql => AdapterConfig::new(
+            QueryProtocol::ClickHouseHttp,
+            QueryLanguage::sql,
+            None,
+        ),
+        QueryLanguage::elastic_sql => AdapterConfig::new(
+            QueryProtocol::ElasticHttp,
+            QueryLanguage::elastic_sql,
+            None,
+        ),
+        _ => AdapterConfig::prometheus_promql(
+            args.prometheus_server.clone(),
+            args.forward_unsupported_queries,
+        ),
+    };
 
     let http_config = HttpServerConfig {
         port: args.http_port,

asap-summary-ingest/.gitignore

@@ -1,3 +1,4 @@
 __pycache__
 **/*.pyc
 **/*.swp
+outputs/

asap-summary-ingest/templates/udfs/countminsketch_count.rs.j2

@@ -3,7 +3,7 @@
 rmp-serde = "1.1"
 serde = { version = "1.0", features = ["derive"] }
 twox-hash = "2.1.0"
-sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust" }
+sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust", rev = "440427438fdaf3ac2298b53ee148f9e12a64ffcc" }
 */
 
 use arroyo_udf_plugin::udf;

asap-summary-ingest/templates/udfs/countminsketch_sum.rs.j2

@@ -3,7 +3,7 @@
 rmp-serde = "1.1"
 serde = { version = "1.0", features = ["derive"] }
 twox-hash = "2.1.0"
-sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust" }
+sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust", rev = "440427438fdaf3ac2298b53ee148f9e12a64ffcc" }
 */
 
 use arroyo_udf_plugin::udf;

asap-summary-ingest/templates/udfs/countminsketchwithheap_topk.rs.j2

@@ -3,7 +3,7 @@
 rmp-serde = "1.1"
 serde = { version = "1.0", features = ["derive"] }
 twox-hash = "2.1.0"
-sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust" }
+sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust", rev = "440427438fdaf3ac2298b53ee148f9e12a64ffcc" }
 */
 
 use std::cmp::Ordering;

asap-summary-ingest/templates/udfs/datasketcheskll_.rs.j2

@@ -1,6 +1,6 @@
 /*
 [dependencies]
-sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust" }
+sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust", rev = "440427438fdaf3ac2298b53ee148f9e12a64ffcc" }
 arroyo-udf-plugin = "0.1"
 rmp-serde = "1.1"
 serde = { version = "1.0", features = ["derive"] }

asap-summary-ingest/templates/udfs/hydrakll_.rs.j2

@@ -1,6 +1,6 @@
 /*
 [dependencies]
-sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust" }
+sketchlib-rust = { git = "https://github.com/ProjectASAP/sketchlib-rust", rev = "440427438fdaf3ac2298b53ee148f9e12a64ffcc" }
 arroyo-udf-plugin = "0.1"
 rmp-serde = "1.1"
 serde = { version = "1.0", features = ["derive"] }

asap-tools/execution-utilities/.gitignore

@@ -7,6 +7,9 @@
 
 clickhouse-benchmark-pipeline/benchmark_results/
 **/data/
+benchmark/arroyo_outputs/
+benchmark/queries/
+benchmark/results/
 
 **/*.csv
 **/*.png