From a98d6eea82c943abffc245a7df68039beb480ae3 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Mon, 4 May 2026 15:41:58 -0400
Subject: [PATCH 1/6] feat: changed asap-query-engine cmd-line args to config
 file

---
 Cargo.lock                                    |  55 +-
 .../dependencies/rs/asap_types/src/enums.rs   |   2 +-
 asap-query-engine/Cargo.toml                  |   1 +
 asap-query-engine/examples/engine_config.yaml | 116 ++++
 asap-query-engine/src/data_model/enums.rs     |  10 +-
 asap-query-engine/src/engine_config.rs        | 352 ++++++++++++
 asap-query-engine/src/main.rs                 | 506 ++++++------------
 7 files changed, 683 insertions(+), 359 deletions(-)
 create mode 100644 asap-query-engine/examples/engine_config.yaml
 create mode 100644 asap-query-engine/src/engine_config.rs

diff --git a/Cargo.lock b/Cargo.lock
index 3187da8e..e66817c9 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -501,6 +501,15 @@ dependencies = [
  "num-traits",
 ]
 
+[[package]]
+name = "atomic"
+version = "0.6.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a89cbf775b137e9b968e67227ef7f775587cde3fd31b0d8599dbd0f598a48340"
+dependencies = [
+ "bytemuck",
+]
+
 [[package]]
 name = "atomic-waker"
 version = "1.1.2"
@@ -671,6 +680,12 @@ version = "3.20.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5d20789868f4b01b2f2caec9f5c4e0213b41e3e5702a50157d699ae31ced2fcb"
 
+[[package]]
+name = "bytemuck"
+version = "1.25.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c8efb64bd706a16a1bdde310ae86b351e4d21550d98d056f22f8a7f7a2183fec"
+
 [[package]]
 name = "byteorder"
 version = "1.5.0"
@@ -1603,7 +1618,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "39cab71617ae0d63f51a36d69f866391735b51691dbda63cf6f96d042b63efeb"
 dependencies = [
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -1624,6 +1639,19 @@ version = "2.4.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9f1f227452a390804cdb637b74a86990f2a7d7ba4b7d5693aac9b4dd6defd8d6"
 
+[[package]]
+name = "figment"
+version = "0.10.19"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "8cb01cd46b0cf372153850f4c6c272d9cbea2da513e07538405148f95bd789f3"
+dependencies = [
+ "atomic",
+ "serde",
+ "serde_yaml",
+ "uncased",
+ "version_check",
+]
+
 [[package]]
 name = "filetime"
 version = "0.2.27"
@@ -2339,7 +2367,7 @@ checksum = "3640c1c38b8e4e43584d8df18be5fc6b0aa314ce6ebf51b53313d4306cca8e46"
 dependencies = [
  "hermit-abi 0.5.2",
  "libc",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -2384,7 +2412,7 @@ dependencies = [
  "portable-atomic",
  "portable-atomic-util",
  "serde_core",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -3372,7 +3400,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
 dependencies = [
  "heck 0.5.0",
- "itertools 0.10.5",
+ "itertools 0.13.0",
  "log",
  "multimap",
  "once_cell",
@@ -3392,7 +3420,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
 dependencies = [
  "anyhow",
- "itertools 0.10.5",
+ "itertools 0.13.0",
  "proc-macro2",
  "quote",
  "syn 2.0.117",
@@ -3509,6 +3537,7 @@ dependencies = [
  "datafusion",
  "datafusion_summary_library",
  "elastic_dsl_utilities",
+ "figment",
  "flate2",
  "form_urlencoded",
  "futures",
@@ -3834,7 +3863,7 @@ dependencies = [
  "errno",
  "libc",
  "linux-raw-sys",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -4181,7 +4210,6 @@ dependencies = [
  "cfg-if",
  "libc",
  "psm",
- "windows-sys 0.52.0",
  "windows-sys 0.59.0",
 ]
 
@@ -4331,7 +4359,7 @@ dependencies = [
  "getrandom 0.4.2",
  "once_cell",
  "rustix",
- "windows-sys 0.52.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -4751,6 +4779,15 @@ version = "1.19.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb"
 
+[[package]]
+name = "uncased"
+version = "0.9.10"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "e1b88fcfe09e89d3866a5c11019378088af2d24c3fbd4f0543f96b479ec90697"
+dependencies = [
+ "version_check",
+]
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.24"
@@ -5037,7 +5074,7 @@ version = "0.1.11"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "c2a7b1c03c876122aa43f3020e6c3c3ee5c05081c9a00739faf7503aeba10d22"
 dependencies = [
- "windows-sys 0.48.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
diff --git a/asap-common/dependencies/rs/asap_types/src/enums.rs b/asap-common/dependencies/rs/asap_types/src/enums.rs
index 7d511c3a..91c233c5 100644
--- a/asap-common/dependencies/rs/asap_types/src/enums.rs
+++ b/asap-common/dependencies/rs/asap_types/src/enums.rs
@@ -4,7 +4,7 @@ use std::str::FromStr;
 // Re-export AggregationType from promql_utilities (defined there to avoid circular deps).
 pub use promql_utilities::query_logics::enums::AggregationType;
 
-#[derive(clap::ValueEnum, Clone, Copy, Debug, PartialEq)]
+#[derive(clap::ValueEnum, Clone, Copy, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
 #[allow(non_camel_case_types)]
 pub enum QueryLanguage {
     #[value(alias = "SQL")]
diff --git a/asap-query-engine/Cargo.toml b/asap-query-engine/Cargo.toml
index f5fa9868..d2886c3c 100644
--- a/asap-query-engine/Cargo.toml
+++ b/asap-query-engine/Cargo.toml
@@ -55,6 +55,7 @@ lazy_static = "1.4"
 zstd = "0.13"
 reqwest = { version = "0.11", features = ["json"] }
 tracing-appender = "0.2"
+figment = { version = "0.10", features = ["yaml"] }
 arc-swap = "1"
 csv = "1"
 elastic_dsl_utilities.workspace = true
diff --git a/asap-query-engine/examples/engine_config.yaml b/asap-query-engine/examples/engine_config.yaml
new file mode 100644
index 00000000..eee867a3
--- /dev/null
+++ b/asap-query-engine/examples/engine_config.yaml
@@ -0,0 +1,116 @@
+# ASAPQuery Engine Configuration
+#
+# Usage:
+#   query-engine-rust --config-file engine_config.yaml
+#   query-engine-rust --config-file engine_config.yaml precompute_engine.num_workers=8
+#   query-engine-rust --config-file engine_config.yaml http_server.port=9000 ingest.topic=new-topic
+
+# ---------------------------------------------------------------------------
+# Core
+# ---------------------------------------------------------------------------
+
+output_dir: "./output"
+log_level: "INFO"                # DEBUG | INFO | WARN | ERROR (also respects RUST_LOG)
+
+# Query language used for both ingest and query parsing.
+# Values are case-sensitive: use lowercase exactly as shown.
+query_language: "promql"         # promql | sql | elastic_querydsl | elastic_sql
+
+# Prometheus scrape interval in seconds. Used by the query tracker and planner.
+prometheus_scrape_interval: 15
+
+# Processing backend. Must be consistent with the ingest source chosen below:
+#   arroyo     → requires ingest.type=kafka
+#   precompute → requires ingest.type=http_remote_write or csv
+streaming_engine: "precompute"   # arroyo | precompute
+
+do_profiling: false
+
+# ---------------------------------------------------------------------------
+# HTTP query server
+# ---------------------------------------------------------------------------
+
+http_server:
+  port: 8088
+
+  # Prometheus server used for query forwarding and planner context.
+  prometheus_server: "http://localhost:9090"
+
+  # When true, queries not answerable from sketches are forwarded to prometheus_server.
+  # The server must be reachable at startup when this is enabled.
+  forward_unsupported_queries: false
+
+# ---------------------------------------------------------------------------
+# Store
+# ---------------------------------------------------------------------------
+
+store:
+  lock_strategy: "per-key"       # global | per-key
+
+# ---------------------------------------------------------------------------
+# Ingest source — choose exactly one type.
+# streaming_engine must match the chosen type (see above).
+# ---------------------------------------------------------------------------
+
+# HTTP remote write (Prometheus remote write protocol):
+ingest:
+  type: "http_remote_write"
+  port: 9090
+
+# Kafka:
+# ingest:
+#   type: "kafka"
+#   broker: "localhost:9092"
+#   topic: "my-topic"
+#   input_format: "json"         # json | byte
+#   decompress_json: false
+
+# CSV file:
+# ingest:
+#   type: "csv"
+#   path: "./data.csv"
+#   metric_name: "my_metric"
+#   value_col: "value"
+#   label_cols: ["id1", "id2"]   # empty list if no label columns
+#   timestamp_col: null          # column name, or null to synthesize timestamps
+#   start_ts_ms: 0               # synthesized timestamp start (ms); used when timestamp_col is null
+#   ts_step_ms: 15000            # ms between rows; required when timestamp_col is null
+#   batch_size: 1000
+
+# OTLP (OpenTelemetry metrics — gRPC + HTTP):
+# Requires streaming_engine: "arroyo". No precompute or Kafka path is used for OTLP;
+# the receiver runs as a standalone gRPC/HTTP server.
+# ingest:
+#   type: "otlp"
+#   grpc_port: 4317
+#   http_port: 4318
+
+# ---------------------------------------------------------------------------
+# Precompute engine (active when streaming_engine=precompute)
+# ---------------------------------------------------------------------------
+
+precompute_engine:
+  num_workers: 4
+  allowed_lateness_ms: 5000      # max out-of-order sample delay before drop
+  max_buffer_per_series: 10000   # max buffered samples per series before eviction
+  flush_interval_ms: 1000        # how often the flush timer fires
+  channel_buffer_size: 10000     # capacity of router-to-worker channels
+  dump_precomputes: false        # dump received precomputes to output_dir for debugging
+
+# ---------------------------------------------------------------------------
+# Query tracker / planner (optional)
+# ---------------------------------------------------------------------------
+
+query_tracker:
+  enabled: false
+
+  # How long to observe query patterns before triggering planning (seconds).
+  observation_window_secs: 100
+
+# ---------------------------------------------------------------------------
+# External config file paths (optional — engine starts with empty configs if omitted)
+# ---------------------------------------------------------------------------
+
+inference_config: null           # path to inference_config.yaml
+streaming_config: null           # path to streaming_config.yaml
+promsketch_config: null          # path to promsketch_config.yaml
diff --git a/asap-query-engine/src/data_model/enums.rs b/asap-query-engine/src/data_model/enums.rs
index d8c44596..602cb064 100644
--- a/asap-query-engine/src/data_model/enums.rs
+++ b/asap-query-engine/src/data_model/enums.rs
@@ -1,10 +1,12 @@
-#[derive(clap::ValueEnum, Clone, Debug)]
+#[derive(clap::ValueEnum, Clone, Debug, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "snake_case")]
 pub enum InputFormat {
     Json,
     Byte,
 }
 
-#[derive(clap::ValueEnum, Clone, Debug, PartialEq)]
+#[derive(clap::ValueEnum, Clone, Debug, PartialEq, serde::Serialize, serde::Deserialize)]
+#[serde(rename_all = "snake_case")]
 pub enum StreamingEngine {
     Arroyo,
     Precompute,
@@ -24,10 +26,12 @@ pub enum QueryProtocol {
     // Future: DuckDbHttp, etc.
 }
 
-#[derive(clap::ValueEnum, Clone, Debug, Copy, PartialEq)]
+#[derive(clap::ValueEnum, Clone, Debug, Copy, PartialEq, serde::Serialize, serde::Deserialize)]
 pub enum LockStrategy {
     #[value(name = "global")]
+    #[serde(rename = "global")]
     Global,
     #[value(name = "per-key")]
+    #[serde(rename = "per-key")]
     PerKey,
 }
diff --git a/asap-query-engine/src/engine_config.rs b/asap-query-engine/src/engine_config.rs
new file mode 100644
index 00000000..5c26d93a
--- /dev/null
+++ b/asap-query-engine/src/engine_config.rs
@@ -0,0 +1,352 @@
+use asap_types::enums::QueryLanguage;
+use query_engine_rust::data_model::enums::{InputFormat, LockStrategy, StreamingEngine};
+
+pub fn check_config(config: &EngineConfig) -> Result<(), String> {
+    match (&config.ingest, &config.streaming_engine) {
+        (IngestConfig::Kafka { .. }, StreamingEngine::Arroyo) => {}
+        (
+            IngestConfig::HttpRemoteWrite { .. } | IngestConfig::Csv { .. },
+            StreamingEngine::Precompute,
+        ) => {}
+        (IngestConfig::Otlp { .. }, StreamingEngine::Arroyo) => {}
+        (IngestConfig::Otlp { .. }, StreamingEngine::Precompute) => {
+            return Err("ingest.type=otlp requires streaming_engine=arroyo (precompute engine does not apply to OTLP)".into());
+        }
+        (IngestConfig::Kafka { .. }, StreamingEngine::Precompute) => {
+            return Err("ingest.type=kafka requires streaming_engine=arroyo".into());
+        }
+        (_, StreamingEngine::Arroyo) => {
+            return Err("streaming_engine=arroyo requires ingest.type=kafka".into());
+        }
+    }
+
+    if let IngestConfig::Csv {
+        timestamp_col: None,
+        ts_step_ms: None,
+        ..
+    } = &config.ingest
+    {
+        return Err("ingest.ts_step_ms is required when ingest.timestamp_col is not set".into());
+    }
+
+    if config.prometheus_scrape_interval == 0 {
+        return Err("prometheus_scrape_interval must be greater than 0".into());
+    }
+
+    Ok(())
+}
+
+#[derive(Debug, serde::Deserialize)]
+#[serde(default)]
+pub struct EngineConfig {
+    pub output_dir: String,
+    pub log_level: String,
+    pub query_language: QueryLanguage,
+    pub prometheus_scrape_interval: u64,
+    pub streaming_engine: StreamingEngine,
+    pub do_profiling: bool,
+    pub http_server: HttpServerSettings,
+    pub store: StoreSettings,
+    pub ingest: IngestConfig,
+    pub precompute_engine: PrecomputeSettings,
+    pub query_tracker: QueryTrackerSettings,
+    pub inference_config: Option<String>,
+    pub streaming_config: Option<String>,
+    pub promsketch_config: Option<String>,
+}
+
+impl Default for EngineConfig {
+    fn default() -> Self {
+        Self {
+            output_dir: "./output".to_string(),
+            log_level: "INFO".to_string(),
+            query_language: QueryLanguage::promql,
+            prometheus_scrape_interval: 15,
+            streaming_engine: StreamingEngine::Precompute,
+            do_profiling: false,
+            http_server: HttpServerSettings::default(),
+            store: StoreSettings::default(),
+            ingest: IngestConfig::default(),
+            precompute_engine: PrecomputeSettings::default(),
+            query_tracker: QueryTrackerSettings::default(),
+            inference_config: None,
+            streaming_config: None,
+            promsketch_config: None,
+        }
+    }
+}
+
+#[derive(Debug, serde::Deserialize)]
+#[serde(default)]
+pub struct HttpServerSettings {
+    pub port: u16,
+    pub prometheus_server: String,
+    pub forward_unsupported_queries: bool,
+}
+
+impl Default for HttpServerSettings {
+    fn default() -> Self {
+        Self {
+            port: 8088,
+            prometheus_server: "http://localhost:9090".to_string(),
+            forward_unsupported_queries: false,
+        }
+    }
+}
+
+#[derive(Debug, serde::Deserialize)]
+#[serde(default)]
+pub struct StoreSettings {
+    pub lock_strategy: LockStrategy,
+}
+
+impl Default for StoreSettings {
+    fn default() -> Self {
+        Self {
+            lock_strategy: LockStrategy::PerKey,
+        }
+    }
+}
+
+#[derive(Debug, serde::Deserialize)]
+#[serde(tag = "type", rename_all = "snake_case")]
+pub enum IngestConfig {
+    HttpRemoteWrite {
+        #[serde(default = "default_http_remote_write_port")]
+        port: u16,
+    },
+    Kafka {
+        #[serde(default = "default_kafka_broker")]
+        broker: String,
+        topic: String,
+        input_format: InputFormat,
+        #[serde(default)]
+        decompress_json: bool,
+    },
+    Csv {
+        path: String,
+        metric_name: String,
+        value_col: String,
+        #[serde(default)]
+        label_cols: Vec<String>,
+        timestamp_col: Option<String>,
+        #[serde(default)]
+        start_ts_ms: i64,
+        ts_step_ms: Option<i64>,
+        #[serde(default = "default_csv_batch_size")]
+        batch_size: usize,
+    },
+    Otlp {
+        #[serde(default = "default_otlp_grpc_port")]
+        grpc_port: u16,
+        #[serde(default = "default_otlp_http_port")]
+        http_port: u16,
+    },
+}
+
+impl Default for IngestConfig {
+    fn default() -> Self {
+        IngestConfig::HttpRemoteWrite {
+            port: default_http_remote_write_port(),
+        }
+    }
+}
+
+fn default_http_remote_write_port() -> u16 {
+    9090
+}
+
+fn default_kafka_broker() -> String {
+    "localhost:9092".to_string()
+}
+
+fn default_csv_batch_size() -> usize {
+    1000
+}
+
+fn default_otlp_grpc_port() -> u16 {
+    4317
+}
+
+fn default_otlp_http_port() -> u16 {
+    4318
+}
+
+#[derive(Debug, serde::Deserialize)]
+#[serde(default)]
+pub struct PrecomputeSettings {
+    pub num_workers: usize,
+    pub allowed_lateness_ms: i64,
+    pub max_buffer_per_series: usize,
+    pub flush_interval_ms: u64,
+    pub channel_buffer_size: usize,
+    pub dump_precomputes: bool,
+}
+
+impl Default for PrecomputeSettings {
+    fn default() -> Self {
+        Self {
+            num_workers: 4,
+            allowed_lateness_ms: 5000,
+            max_buffer_per_series: 10000,
+            flush_interval_ms: 1000,
+            channel_buffer_size: 10000,
+            dump_precomputes: false,
+        }
+    }
+}
+
+#[derive(Debug, serde::Deserialize)]
+#[serde(default)]
+pub struct QueryTrackerSettings {
+    pub enabled: bool,
+    pub observation_window_secs: u64,
+}
+
+impl Default for QueryTrackerSettings {
+    fn default() -> Self {
+        Self {
+            enabled: false,
+            observation_window_secs: 100,
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use figment::{
+        providers::{Format, Yaml},
+        Figment,
+    };
+
+    const MINIMAL_YAML: &str = r#"
+streaming_engine: "precompute"
+ingest:
+  type: "http_remote_write"
+  port: 9090
+output_dir: "./output"
+"#;
+
+    // Verifies that figment's tuple provider ("a.b", val) resolves dotted paths
+    // into nested structs rather than treating the key as a flat string.
+    #[test]
+    fn dotted_key_override_sets_nested_field() {
+        let config: EngineConfig = Figment::new()
+            .merge(Yaml::string(MINIMAL_YAML))
+            .merge(("precompute_engine.num_workers", 8usize))
+            .merge(("http_server.port", 9000u16))
+            .extract()
+            .expect("config should deserialize");
+
+        assert_eq!(config.precompute_engine.num_workers, 8);
+        assert_eq!(config.http_server.port, 9000);
+    }
+
+    #[test]
+    fn check_config_valid_precompute_http() {
+        let config: EngineConfig = Figment::new()
+            .merge(Yaml::string(MINIMAL_YAML))
+            .extract()
+            .unwrap();
+        assert!(check_config(&config).is_ok());
+    }
+
+    #[test]
+    fn check_config_valid_kafka_arroyo() {
+        let yaml = r#"
+streaming_engine: "arroyo"
+ingest:
+  type: "kafka"
+  topic: "my-topic"
+  input_format: "json"
+output_dir: "./output"
+"#;
+        let config: EngineConfig = Figment::new().merge(Yaml::string(yaml)).extract().unwrap();
+        assert!(check_config(&config).is_ok());
+    }
+
+    #[test]
+    fn check_config_rejects_kafka_with_precompute() {
+        let yaml = r#"
+streaming_engine: "precompute"
+ingest:
+  type: "kafka"
+  topic: "t"
+  input_format: "json"
+output_dir: "./output"
+"#;
+        let config: EngineConfig = Figment::new().merge(Yaml::string(yaml)).extract().unwrap();
+        assert!(check_config(&config).is_err());
+    }
+
+    #[test]
+    fn check_config_rejects_http_with_arroyo() {
+        let yaml = r#"
+streaming_engine: "arroyo"
+ingest:
+  type: "http_remote_write"
+  port: 9090
+output_dir: "./output"
+"#;
+        let config: EngineConfig = Figment::new().merge(Yaml::string(yaml)).extract().unwrap();
+        assert!(check_config(&config).is_err());
+    }
+
+    #[test]
+    fn check_config_rejects_otlp_with_precompute() {
+        let yaml = r#"
+streaming_engine: "precompute"
+ingest:
+  type: "otlp"
+output_dir: "./output"
+"#;
+        let config: EngineConfig = Figment::new().merge(Yaml::string(yaml)).extract().unwrap();
+        assert!(check_config(&config).is_err());
+    }
+
+    #[test]
+    fn check_config_rejects_csv_without_ts_step_ms() {
+        let yaml = r#"
+streaming_engine: "precompute"
+ingest:
+  type: "csv"
+  path: "data.csv"
+  metric_name: "m"
+  value_col: "v"
+output_dir: "./output"
+"#;
+        let config: EngineConfig = Figment::new().merge(Yaml::string(yaml)).extract().unwrap();
+        assert!(check_config(&config).is_err());
+    }
+
+    #[test]
+    fn check_config_csv_with_timestamp_col_does_not_require_ts_step_ms() {
+        let yaml = r#"
+streaming_engine: "precompute"
+ingest:
+  type: "csv"
+  path: "data.csv"
+  metric_name: "m"
+  value_col: "v"
+  timestamp_col: "ts"
+output_dir: "./output"
+"#;
+        let config: EngineConfig = Figment::new().merge(Yaml::string(yaml)).extract().unwrap();
+        assert!(check_config(&config).is_ok());
+    }
+
+    #[test]
+    fn check_config_rejects_zero_scrape_interval() {
+        let yaml = r#"
+streaming_engine: "precompute"
+ingest:
+  type: "http_remote_write"
+  port: 9090
+prometheus_scrape_interval: 0
+output_dir: "./output"
+"#;
+        let config: EngineConfig = Figment::new().merge(Yaml::string(yaml)).extract().unwrap();
+        assert!(check_config(&config).is_err());
+    }
+}
diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs
index a96ec4b1..8c302535 100644
--- a/asap-query-engine/src/main.rs
+++ b/asap-query-engine/src/main.rs
@@ -1,14 +1,18 @@
+mod engine_config;
+
 use clap::Parser;
-use query_engine_rust::data_model::QueryLanguage;
+use engine_config::{EngineConfig, IngestConfig};
+use figment::{
+    providers::{Format, Yaml},
+    Figment,
+};
 use std::fs;
 use std::sync::{Arc, RwLock};
 use tokio::signal;
 use tracing::{debug, error, info, warn};
 
 use asap_types::streaming_config::StreamingConfig;
-use query_engine_rust::data_model::enums::{
-    CleanupPolicy, InputFormat, LockStrategy, StreamingEngine,
-};
+use query_engine_rust::data_model::enums::{CleanupPolicy, StreamingEngine};
 use query_engine_rust::drivers::AdapterConfig;
 use query_engine_rust::precompute_engine::config::LateDataPolicy;
 use query_engine_rust::precompute_engine::csv_ingest::{CsvFileIngestConfig, CsvFileIngestSource};
@@ -25,191 +29,51 @@ use query_engine_rust::{
 #[derive(Parser, Debug)]
 #[command(author, version, about, long_about = None)]
 struct Args {
-    /// Kafka topic to consume from (required when streaming-engine=arroyo)
-    #[arg(long)]
-    kafka_topic: Option<String>,
-
-    /// Input format for Kafka messages (required when streaming-engine=arroyo)
-    #[arg(long, value_enum)]
-    input_format: Option<InputFormat>,
-
-    /// Inference config file path (optional; starts with empty config when omitted, requires --enable-query-tracker)
-    #[arg(long)]
-    config: Option<String>,
-
-    /// Streaming config file path (optional; starts with empty config when omitted, requires --enable-query-tracker)
-    #[arg(long)]
-    streaming_config: Option<String>,
-
-    /// Streaming engine to use
-    #[arg(long, value_enum, default_value = "arroyo")]
-    streaming_engine: StreamingEngine,
-
-    /// Prometheus scrape interval in seconds
-    #[arg(long)]
-    prometheus_scrape_interval: u64,
-
-    /// HTTP server port
-    #[arg(long, default_value = "8088")]
-    http_port: u16,
-
-    /// Prometheus server URL
-    #[arg(long, default_value = "http://localhost:9090")]
-    prometheus_server: String,
-
-    /// Forward unsupported queries to Prometheus
-    #[arg(long)]
-    forward_unsupported_queries: bool,
-
-    /// Kafka broker address
-    #[arg(long, default_value = "localhost:9092")]
-    kafka_broker: String,
-
-    /// Database path (currently unused, kept for compatibility)
-    #[arg(long, default_value = "sketchdb.db")]
-    db_path: String,
-
-    /// Delete existing database (currently unused, kept for compatibility)
-    #[arg(long)]
-    delete_existing_db: bool,
-
-    /// Output directory for logs
-    #[arg(long)]
-    output_dir: String,
-
-    /// Log level
-    #[arg(long, default_value = "INFO")]
-    log_level: String,
-
-    /// Enable profiling (currently unused, kept for compatibility)
-    #[arg(long)]
-    do_profiling: bool,
-
-    /// Decompress JSON messages
-    #[arg(long)]
-    decompress_json: bool,
-
-    /// Enable dumping received precomputes to files for debugging
-    #[arg(long)]
-    dump_precomputes: bool,
-
-    /// Differentiate between query languages of input query
-    #[arg(long, value_enum)]
-    query_language: QueryLanguage,
-
-    /// Lock strategy for SimpleMapStore: "global" for single mutex, "per-key" for fine-grained locking
-    #[arg(long, value_enum)]
-    lock_strategy: LockStrategy,
-
-    /// Enable Prometheus remote write ingest endpoint
-    #[arg(long)]
-    enable_prometheus_remote_write: bool,
-
-    /// Port for the Prometheus remote write endpoint
-    #[arg(long, default_value = "9090")]
-    prometheus_remote_write_port: u16,
-
-    /// Path to promsketch configuration YAML file (optional; uses defaults if omitted)
-    #[arg(long)]
-    promsketch_config: Option<String>,
-
-    /// Enable OTLP metrics ingest (gRPC + HTTP)
-    #[arg(long)]
-    enable_otel_ingest: bool,
-
-    /// OTLP gRPC listen port
-    #[arg(long, default_value = "4317")]
-    otel_grpc_port: u16,
-
-    /// OTLP HTTP listen port
-    #[arg(long, default_value = "4318")]
-    otel_http_port: u16,
-
-    /// Number of precompute engine worker threads
-    #[arg(long, default_value = "4")]
-    precompute_num_workers: usize,
-
-    /// Maximum allowed lateness for out-of-order samples (milliseconds)
-    #[arg(long, default_value = "5000")]
-    precompute_allowed_lateness_ms: i64,
-
-    /// Maximum buffered samples per series before eviction
-    #[arg(long, default_value = "10000")]
-    precompute_max_buffer_per_series: usize,
-
-    /// Interval at which the flush timer fires (milliseconds)
-    #[arg(long, default_value = "1000")]
-    precompute_flush_interval_ms: u64,
-
-    /// Capacity of the channel between router and each worker
-    #[arg(long, default_value = "10000")]
-    precompute_channel_buffer_size: usize,
-
-    /// Enable automatic query tracking and planning
-    #[arg(long)]
-    enable_query_tracker: bool,
-
-    /// Query tracker: observation window in seconds before triggering planning
-    #[arg(long, default_value = "100")]
-    tracker_observation_window_secs: u64,
-
-    // --- CSV file ingest (alternative to HTTP remote write) ---
-    /// Path to a local CSV file to ingest instead of listening for HTTP writes
-    #[arg(long)]
-    input_file: Option<String>,
-
-    /// Metric name to assign to every row (required with --input-file)
-    #[arg(long)]
-    csv_metric_name: Option<String>,
-
-    /// CSV column to use as the float value (required with --input-file)
-    #[arg(long)]
-    csv_value_col: Option<String>,
-
-    /// Comma-separated CSV columns to include as labels (e.g. "id1,id2,id3")
-    #[arg(long, default_value = "")]
-    csv_label_cols: String,
-
-    /// CSV column containing timestamps in milliseconds; omit to synthesize timestamps
-    #[arg(long)]
-    csv_timestamp_col: Option<String>,
-
-    /// Start timestamp (ms) for synthesized timestamps when --csv-timestamp-col is absent
-    #[arg(long, default_value_t = 0)]
-    csv_start_ts_ms: i64,
-
-    /// Milliseconds between consecutive rows for synthesized timestamps
-    /// (required when --csv-timestamp-col is absent)
+    /// Path to the engine YAML configuration file
     #[arg(long)]
-    csv_ts_step_ms: Option<i64>,
+    config_file: String,
 
-    /// Number of CSV rows per batch sent to workers
-    #[arg(long, default_value_t = 1000)]
-    csv_batch_size: usize,
+    /// KEY=VALUE overrides applied on top of the config file (e.g. http_server.port=9000)
+    overrides: Vec<String>,
 }
 
 #[tokio::main]
 async fn main() -> Result<()> {
     let args = Args::parse();
 
+    let mut figment = Figment::new().merge(Yaml::file(&args.config_file));
+    for kv in &args.overrides {
+        let (key, val_str) = kv
+            .split_once('=')
+            .ok_or_else(|| format!("invalid override '{kv}': expected KEY=VALUE"))?;
+        // Parse as JSON so booleans and numbers get the right type; fall back to string.
+        let val: serde_json::Value = serde_json::from_str(val_str)
+            .unwrap_or_else(|_| serde_json::Value::String(val_str.to_string()));
+        figment = figment.merge((key, val));
+    }
+    let config: EngineConfig = figment
+        .extract()
+        .map_err(|e| format!("Config error in {}: {e}", args.config_file))?;
+
+    engine_config::check_config(&config).map_err(|e| format!("Invalid config: {e}"))?;
+
     // Create output directory
-    fs::create_dir_all(&args.output_dir)?;
+    fs::create_dir_all(&config.output_dir)?;
 
-    // Initialize logging similar to Python's create_loggers function
     // Keep the guard alive for the entire lifetime of the application
-    let _log_guard = setup_logging(&args.output_dir, &args.log_level)?;
+    let _log_guard = setup_logging(&config.output_dir, &config.log_level)?;
 
     info!("Starting Query Engine Rust");
-    info!("Output directory: {}", args.output_dir);
+    info!("Output directory: {}", config.output_dir);
 
-    let inference_config = match &args.config {
+    let inference_config = match &config.inference_config {
         Some(path) => {
             info!("Config file: {}", path);
-            read_inference_config(path, args.query_language)?
+            read_inference_config(path, config.query_language)?
         }
         None => {
             info!("No config file provided; starting with empty inference config");
-            InferenceConfig::new(args.query_language, CleanupPolicy::NoCleanup)
+            InferenceConfig::new(config.query_language, CleanupPolicy::NoCleanup)
         }
     };
     info!(
@@ -217,7 +81,7 @@ async fn main() -> Result<()> {
         inference_config.query_configs.len()
     );
 
-    let streaming_config = Arc::new(match &args.streaming_config {
+    let streaming_config = Arc::new(match &config.streaming_config {
         Some(path) => read_streaming_config(path, &inference_config)?,
         None => {
             info!("No streaming config file provided; starting with empty streaming config");
@@ -230,111 +94,89 @@ async fn main() -> Result<()> {
     );
 
     // Shared config refs — passed to QueryTracker so it can populate ControllerConfig
-    // with the current configs as context for the planner.  The applier task updates
+    // with the current configs as context for the planner. The applier task updates
     // them after applying a new plan so that subsequent windows see the latest state.
     let streaming_config_ref = Arc::new(RwLock::new(streaming_config.clone()));
     let inference_config_ref = Arc::new(RwLock::new(Arc::new(inference_config.clone())));
 
-    // Setup store (equivalent to Python's SimpleMapStore())
-    // Get cleanup policy from inference config
     let cleanup_policy = inference_config.cleanup_policy;
     info!("Using cleanup policy: {:?}", cleanup_policy);
     let store = Arc::new(SimpleMapStore::new_with_strategy(
         streaming_config.clone(),
         cleanup_policy,
-        args.lock_strategy,
+        config.store.lock_strategy,
     ));
 
-    // // Setup PromSketchStore (shared between engine and remote write server)
-    // let promsketch_store = if args.enable_prometheus_remote_write {
-    //     let promsketch_config = match &args.promsketch_config {
-    //         Some(path) => {
-    //             let cfg = read_promsketch_config(path)?;
-    //             info!("Loaded promsketch config from {}: {:?}", path, cfg);
-    //             cfg
-    //         }
-    //         None => {
-    //             info!("Using default promsketch config");
-    //             PromSketchConfig::default()
-    //         }
-    //     };
-    //     info!("Prometheus remote write enabled: creating PromSketchStore");
-    //     Some(Arc::new(PromSketchStore::new(promsketch_config)))
-    // } else {
-    //     None
-    // };
-
-    // Setup query engine
     let engine = Arc::new(SimpleEngine::new(
         store.clone(),
-        // promsketch_store.clone(),
         inference_config,
         streaming_config.clone(),
-        args.prometheus_scrape_interval,
-        args.query_language,
+        config.prometheus_scrape_interval,
+        config.query_language,
     ));
 
-    // Setup Kafka consumer (only when not using precompute engine as the streaming backend)
-    let kafka_handle = if args.streaming_engine == StreamingEngine::Precompute {
-        info!("Using precompute engine as streaming backend — skipping Kafka consumer");
-        None
-    } else {
-        let kafka_topic = args.kafka_topic.clone().unwrap_or_else(|| {
-            error!("--kafka-topic is required when --streaming-engine is not precompute");
-            std::process::exit(1);
-        });
-        let input_format = args.input_format.unwrap_or_else(|| {
-            error!("--input-format is required when --streaming-engine is not precompute");
-            std::process::exit(1);
-        });
-        let kafka_config = KafkaConsumerConfig {
-            broker: args.kafka_broker.clone(),
-            topic: kafka_topic.clone(),
-            group_id: "query-engine-rust".to_string(),
-            auto_offset_reset: "beginning".to_string(),
-            input_format,
-            decompress_json: args.decompress_json,
-            batch_size: 1000,
-            poll_timeout_ms: 1000,
-            streaming_engine: args.streaming_engine.clone(),
-            dump_precomputes: args.dump_precomputes,
-            dump_output_dir: if args.dump_precomputes {
-                Some(args.output_dir.clone())
-            } else {
-                None
-            },
-        };
-
-        let store_for_kafka = store.clone();
-        let kafka_consumer_result =
-            KafkaConsumer::new(kafka_config, store_for_kafka, streaming_config.clone());
-        match kafka_consumer_result {
-            Ok(mut consumer) => {
-                info!("Starting Kafka consumer for topic: {}", kafka_topic);
-                Some(tokio::spawn(async move {
-                    if let Err(e) = consumer.run().await {
-                        error!("Kafka consumer error: {}", e);
+    // Kafka consumer — only when streaming_engine=arroyo and ingest.type=kafka.
+    let kafka_handle = if config.streaming_engine == StreamingEngine::Arroyo {
+        match &config.ingest {
+            IngestConfig::Kafka {
+                broker,
+                topic,
+                input_format,
+                decompress_json,
+            } => {
+                let kafka_config = KafkaConsumerConfig {
+                    broker: broker.clone(),
+                    topic: topic.clone(),
+                    group_id: "query-engine-rust".to_string(),
+                    auto_offset_reset: "beginning".to_string(),
+                    input_format: input_format.clone(),
+                    decompress_json: *decompress_json,
+                    batch_size: 1000,
+                    poll_timeout_ms: 1000,
+                    streaming_engine: config.streaming_engine.clone(),
+                    dump_precomputes: config.precompute_engine.dump_precomputes,
+                    dump_output_dir: if config.precompute_engine.dump_precomputes {
+                        Some(config.output_dir.clone())
+                    } else {
+                        None
+                    },
+                };
+                match KafkaConsumer::new(kafka_config, store.clone(), streaming_config.clone()) {
+                    Ok(mut consumer) => {
+                        info!("Starting Kafka consumer for topic: {}", topic);
+                        Some(tokio::spawn(async move {
+                            if let Err(e) = consumer.run().await {
+                                error!("Kafka consumer error: {}", e);
+                            }
+                        }))
                     }
-                }))
-            }
-            Err(e) => {
-                error!("Failed to create Kafka consumer: {}", e);
-                info!("Continuing without Kafka consumer");
-                None
+                    Err(e) => {
+                        error!("Failed to create Kafka consumer: {}", e);
+                        info!("Continuing without Kafka consumer");
+                        None
+                    }
+                }
             }
+            _ => unreachable!("check_config enforces arroyo requires kafka"),
         }
+    } else {
+        info!("Using precompute engine as streaming backend — skipping Kafka consumer");
+        None
     };
 
-    // Setup OTLP receiver
-    let otel_handle = if args.enable_otel_ingest {
-        let otel_config = OtlpReceiverConfig {
-            grpc_port: args.otel_grpc_port,
-            http_port: args.otel_http_port,
-        };
-        let receiver = OtlpReceiver::new(otel_config);
+    // OTLP receiver — only when ingest.type=otlp.
+    let otel_handle = if let IngestConfig::Otlp {
+        grpc_port,
+        http_port,
+    } = &config.ingest
+    {
+        let receiver = OtlpReceiver::new(OtlpReceiverConfig {
+            grpc_port: *grpc_port,
+            http_port: *http_port,
+        });
         info!(
             "Starting OTLP receiver (gRPC port {}, HTTP port {})",
-            args.otel_grpc_port, args.otel_http_port
+            grpc_port, http_port
         );
         Some(tokio::spawn(async move {
             if let Err(e) = receiver.run().await {
@@ -345,66 +187,59 @@ async fn main() -> Result<()> {
         None
     };
 
-    // Setup precompute engine (replaces standalone Prometheus remote write server)
-    // Automatically enable when using precompute streaming engine or ingesting from a file
-    let enable_precompute = args.enable_prometheus_remote_write
-        || args.streaming_engine == StreamingEngine::Precompute
-        || args.input_file.is_some();
-
-    // Handle extracted before run() so the applier task can call update_streaming_config.
+    // Precompute engine — driven by streaming_engine=precompute.
+    // check_config() already enforces the ingest source is compatible (http_remote_write or csv).
     let mut pe_engine_handle: Option<PrecomputeEngineHandle> = None;
 
-    let precompute_handle = if enable_precompute {
+    let precompute_handle = if config.streaming_engine == StreamingEngine::Precompute {
         let precompute_config = PrecomputeEngineConfig {
-            num_workers: args.precompute_num_workers,
-            allowed_lateness_ms: args.precompute_allowed_lateness_ms,
-            max_buffer_per_series: args.precompute_max_buffer_per_series,
-            flush_interval_ms: args.precompute_flush_interval_ms,
-            channel_buffer_size: args.precompute_channel_buffer_size,
+            num_workers: config.precompute_engine.num_workers,
+            allowed_lateness_ms: config.precompute_engine.allowed_lateness_ms,
+            max_buffer_per_series: config.precompute_engine.max_buffer_per_series,
+            flush_interval_ms: config.precompute_engine.flush_interval_ms,
+            channel_buffer_size: config.precompute_engine.channel_buffer_size,
             pass_raw_samples: false,
             raw_mode_aggregation_id: 0,
             late_data_policy: LateDataPolicy::Drop,
         };
         let output_sink = Arc::new(StoreOutputSink::new(store.clone()));
-        let sources: Vec<Box<dyn IngestSource>> = if let Some(ref path) = args.input_file {
-            let metric_name = args
-                .csv_metric_name
-                .clone()
-                .ok_or("--csv-metric-name is required with --input-file")?;
-            let value_col = args
-                .csv_value_col
-                .clone()
-                .ok_or("--csv-value-col is required with --input-file")?;
-            let ts_step_ms = if args.csv_timestamp_col.is_none() {
-                args.csv_ts_step_ms.ok_or(
-                    "--csv-ts-step-ms is required when --csv-timestamp-col is not specified",
-                )?
-            } else {
-                args.csv_ts_step_ms.unwrap_or(0)
-            };
-            let label_cols = if args.csv_label_cols.is_empty() {
-                vec![]
-            } else {
-                args.csv_label_cols
-                    .split(',')
-                    .map(|s| s.trim().to_string())
-                    .collect()
-            };
-            info!("File ingest mode: {}", path);
-            vec![Box::new(CsvFileIngestSource::new(CsvFileIngestConfig {
-                path: path.clone(),
+        let sources: Vec<Box<dyn IngestSource>> = match &config.ingest {
+            IngestConfig::Csv {
+                path,
                 metric_name,
                 value_col,
                 label_cols,
-                timestamp_col: args.csv_timestamp_col.clone(),
-                start_ts_ms: args.csv_start_ts_ms,
+                timestamp_col,
+                start_ts_ms,
                 ts_step_ms,
-                batch_size: args.csv_batch_size,
-            }))]
-        } else {
-            vec![Box::new(HttpIngestSource::new(HttpIngestConfig {
-                port: args.prometheus_remote_write_port,
-            }))]
+                batch_size,
+            } => {
+                // ts_step_ms is only used for timestamp synthesis (when timestamp_col is absent).
+                // check_config ensures it is present in that case.
+                let ts_step = if timestamp_col.is_none() {
+                    ts_step_ms.unwrap()
+                } else {
+                    0
+                };
+                info!("File ingest mode: {}", path);
+                vec![Box::new(CsvFileIngestSource::new(CsvFileIngestConfig {
+                    path: path.clone(),
+                    metric_name: metric_name.clone(),
+                    value_col: value_col.clone(),
+                    label_cols: label_cols.clone(),
+                    timestamp_col: timestamp_col.clone(),
+                    start_ts_ms: *start_ts_ms,
+                    ts_step_ms: ts_step,
+                    batch_size: *batch_size,
+                }))]
+            }
+            IngestConfig::HttpRemoteWrite { port } => {
+                info!("Starting precompute engine on port {}", port);
+                vec![Box::new(HttpIngestSource::new(HttpIngestConfig {
+                    port: *port,
+                }))]
+            }
+            _ => unreachable!("check_config enforces precompute requires http_remote_write or csv"),
         };
         let pe = PrecomputeEngine::new(
             precompute_config,
@@ -413,14 +248,8 @@ async fn main() -> Result<()> {
             sources,
         );
         let worker_diagnostics = pe.diagnostics();
-        // Extract the handle before run() consumes the engine.
         pe_engine_handle = Some(pe.handle());
-        info!(
-            "Starting precompute engine on port {}",
-            args.prometheus_remote_write_port
-        );
 
-        // Spawn periodic memory diagnostics logger
         let diag_store = store.clone();
         tokio::spawn(async move {
             spawn_memory_diagnostics(diag_store, Some(worker_diagnostics)).await;
@@ -432,7 +261,6 @@ async fn main() -> Result<()> {
             }
         }))
     } else {
-        // Even without precompute, log store diagnostics
         let diag_store = store.clone();
         tokio::spawn(async move {
             spawn_memory_diagnostics(diag_store, None).await;
@@ -440,34 +268,22 @@ async fn main() -> Result<()> {
         None
     };
 
-    //info!("=== TEMPORARY: Using ClickHouse HTTP adapter ===");
-    //info!("ClickHouse endpoint will be available at: /clickhouse/query");
-    //info!("ClickHouse fallback URL: http://localhost:8123/?database=default");
-
-    //let adapter_config = AdapterConfig::clickhouse_sql(
-    //    "http://localhost:8123".to_string(), // ClickHouse server URL
-    //    "default".to_string(),               // Database name
-    //    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,
+        config.http_server.prometheus_server.clone(),
+        config.http_server.forward_unsupported_queries,
     );
 
     let http_config = HttpServerConfig {
-        port: args.http_port,
+        port: config.http_server.port,
         handle_http_requests: true,
         adapter_config,
     };
 
-    // Verify Prometheus is reachable before starting (only needed when forwarding queries)
-    if args.forward_unsupported_queries {
+    if config.http_server.forward_unsupported_queries {
         let client = reqwest::Client::new();
         let health_url = format!(
             "{}/api/v1/status/runtimeinfo",
-            args.prometheus_server.trim_end_matches('/')
+            config.http_server.prometheus_server.trim_end_matches('/')
         );
         match client
             .get(&health_url)
@@ -476,12 +292,15 @@ async fn main() -> Result<()> {
             .await
         {
             Ok(resp) if resp.status().is_success() => {
-                info!("Prometheus reachable at {}", args.prometheus_server);
+                info!(
+                    "Prometheus reachable at {}",
+                    config.http_server.prometheus_server
+                );
             }
             Ok(resp) => {
                 error!(
                     "Prometheus at {} returned HTTP {} — cannot start",
-                    args.prometheus_server,
+                    config.http_server.prometheus_server,
                     resp.status()
                 );
                 std::process::exit(1);
@@ -489,32 +308,32 @@ async fn main() -> Result<()> {
             Err(e) => {
                 error!(
                     "Cannot reach Prometheus at {}: {}",
-                    args.prometheus_server, e
+                    config.http_server.prometheus_server, e
                 );
                 std::process::exit(1);
             }
         }
     }
 
-    let query_tracker = if args.enable_query_tracker {
+    let query_tracker = if config.query_tracker.enabled {
         use query_engine_rust::planner_client::{LocalPlannerClient, PlannerResult};
         use query_engine_rust::QueryTrackerConfig;
 
         let tracker_config = QueryTrackerConfig {
-            observation_window_secs: args.tracker_observation_window_secs,
-            prometheus_scrape_interval: args.prometheus_scrape_interval,
+            observation_window_secs: config.query_tracker.observation_window_secs,
+            prometheus_scrape_interval: config.prometheus_scrape_interval,
         };
         let runtime_options = asap_planner::RuntimeOptions {
-            prometheus_scrape_interval: args.prometheus_scrape_interval,
+            prometheus_scrape_interval: config.prometheus_scrape_interval,
             streaming_engine: asap_planner::StreamingEngine::Precompute,
             enable_punting: false,
             range_duration: 300,
-            step: args.prometheus_scrape_interval,
+            step: config.prometheus_scrape_interval,
         };
         let planner_client = Arc::new(LocalPlannerClient::new(
             runtime_options,
-            args.query_language,
-            args.prometheus_server.clone(),
+            config.query_language,
+            config.http_server.prometheus_server.clone(),
         ));
 
         let (plan_tx, plan_rx) = tokio::sync::watch::channel(None::<PlannerResult>);
@@ -526,12 +345,11 @@ async fn main() -> Result<()> {
         ));
         let _tracker_handle = tracker.start_background_loop(planner_client, plan_tx);
 
-        // Applier task: watches for the first plan result and applies it to all
-        // running components.
-        // NOTE: streaming_config and inference_config are not applied atomically
-        // across components. A brief window may exist where one component has the
-        // new config and another still has the old one, causing query misses that
-        // fall back to Prometheus. This is acceptable for a one-shot first-plan apply.
+        // Applier task: watches for plan results and applies them to all running components.
+        // NOTE: streaming_config and inference_config are not applied atomically across
+        // components. A brief window may exist where one component has the new config and
+        // another still has the old one, causing query misses that fall back to Prometheus.
+        // This is acceptable for a one-shot first-plan apply.
         let engine_for_applier = engine.clone();
         let store_for_applier = store.clone();
         let streaming_config_ref_for_applier = streaming_config_ref.clone();
@@ -544,7 +362,6 @@ async fn main() -> Result<()> {
                 }
                 let result = rx.borrow().clone();
                 if let Some(result) = result {
-                    // 1. Apply to precompute engine (lock-free ArcSwap + worker broadcast).
                     if let Some(ref handle) = pe_engine_handle {
                         if let Err(e) = handle
                             .update_streaming_config(&result.streaming_config)
@@ -553,13 +370,10 @@ async fn main() -> Result<()> {
                             warn!("Applier: failed to update precompute engine: {}", e);
                         }
                     }
-                    // 2. Apply to query engine.
                     engine_for_applier
                         .update_streaming_config(Arc::new(result.streaming_config.clone()));
                     engine_for_applier.update_inference_config(result.inference_config.clone());
-                    // 3. Apply to store.
                     store_for_applier.update_streaming_config(result.streaming_config.clone());
-                    // 4. Update shared config refs so future tracker windows see the new state.
                     *streaming_config_ref_for_applier.write().unwrap() =
                         Arc::new(result.streaming_config);
                     *inference_config_ref_for_applier.write().unwrap() =
@@ -571,7 +385,7 @@ async fn main() -> Result<()> {
 
         info!(
             "Query tracker enabled (observation window: {}s)",
-            args.tracker_observation_window_secs
+            config.query_tracker.observation_window_secs
         );
         Some(tracker)
     } else {
@@ -579,7 +393,7 @@ async fn main() -> Result<()> {
     };
 
     let server = HttpServer::new(http_config, engine, store, query_tracker);
-    info!("Starting HTTP server on port {}", args.http_port);
+    info!("Starting HTTP server on port {}", config.http_server.port);
 
     // Wait for shutdown signal
     tokio::select! {

From 401a9896a53e08e004608cb2e3ff0b5b5d6d8f8a Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Mon, 4 May 2026 16:00:02 -0400
Subject: [PATCH 2/6] added test

---
 asap-query-engine/src/engine_config.rs | 25 +++++++++++++++++++++++++
 asap-query-engine/src/main.rs          |  4 +++-
 2 files changed, 28 insertions(+), 1 deletion(-)

diff --git a/asap-query-engine/src/engine_config.rs b/asap-query-engine/src/engine_config.rs
index 5c26d93a..09b51d5f 100644
--- a/asap-query-engine/src/engine_config.rs
+++ b/asap-query-engine/src/engine_config.rs
@@ -336,6 +336,31 @@ output_dir: "./output"
         assert!(check_config(&config).is_ok());
     }
 
+    #[test]
+    fn check_config_valid_otlp_arroyo() {
+        let yaml = r#"
+streaming_engine: "arroyo"
+ingest:
+  type: "otlp"
+output_dir: "./output"
+"#;
+        let config: EngineConfig = Figment::new().merge(Yaml::string(yaml)).extract().unwrap();
+        assert!(check_config(&config).is_ok());
+    }
+
+    #[test]
+    fn enum_override_sets_lock_strategy() {
+        let config: EngineConfig = Figment::new()
+            .merge(Yaml::string(MINIMAL_YAML))
+            .merge(("store.lock_strategy", "global"))
+            .extract()
+            .expect("config should deserialize");
+        assert_eq!(
+            config.store.lock_strategy,
+            query_engine_rust::data_model::enums::LockStrategy::Global
+        );
+    }
+
     #[test]
     fn check_config_rejects_zero_scrape_interval() {
         let yaml = r#"
diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs
index 8c302535..f610c9de 100644
--- a/asap-query-engine/src/main.rs
+++ b/asap-query-engine/src/main.rs
@@ -41,7 +41,7 @@ struct Args {
 async fn main() -> Result<()> {
     let args = Args::parse();
 
-    let mut figment = Figment::new().merge(Yaml::file(&args.config_file));
+    let mut figment = Figment::new().merge(Yaml::file_exact(&args.config_file));
     for kv in &args.overrides {
         let (key, val_str) = kv
             .split_once('=')
@@ -157,6 +157,8 @@ async fn main() -> Result<()> {
                     }
                 }
             }
+            // OTLP uses its own receiver started below; kafka_handle is not needed.
+            IngestConfig::Otlp { .. } => None,
             _ => unreachable!("check_config enforces arroyo requires kafka"),
         }
     } else {

From f6b1e9924fe88f7cef2a72b65f6d736710257865 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Mon, 4 May 2026 17:59:27 -0400
Subject: [PATCH 3/6] added debug log statement

---
 asap-query-engine/src/main.rs | 1 +
 1 file changed, 1 insertion(+)

diff --git a/asap-query-engine/src/main.rs b/asap-query-engine/src/main.rs
index f610c9de..49b51b4d 100644
--- a/asap-query-engine/src/main.rs
+++ b/asap-query-engine/src/main.rs
@@ -63,6 +63,7 @@ async fn main() -> Result<()> {
     // Keep the guard alive for the entire lifetime of the application
     let _log_guard = setup_logging(&config.output_dir, &config.log_level)?;
 
+    debug!("Loaded config:\n{:#?}", config);
     info!("Starting Query Engine Rust");
     info!("Output directory: {}", config.output_dir);
 

From 38e95333e3f05650292b56a5add2bbc240751bb6 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Mon, 4 May 2026 18:22:10 -0400
Subject: [PATCH 4/6] fixed CI

---
 benchmarks/config/engine_config.yaml | 34 ++++++++++++++++++++++++++++
 benchmarks/docker-compose.yml        |  4 ++++
 2 files changed, 38 insertions(+)
 create mode 100644 benchmarks/config/engine_config.yaml

diff --git a/benchmarks/config/engine_config.yaml b/benchmarks/config/engine_config.yaml
new file mode 100644
index 00000000..eb2e5a60
--- /dev/null
+++ b/benchmarks/config/engine_config.yaml
@@ -0,0 +1,34 @@
+output_dir: "/app/outputs"
+log_level: "INFO"
+query_language: "promql"
+prometheus_scrape_interval: 1
+streaming_engine: "precompute"
+do_profiling: false
+
+http_server:
+  port: 8088
+  prometheus_server: "http://prometheus:9090"
+  forward_unsupported_queries: true
+
+store:
+  lock_strategy: "per-key"
+
+ingest:
+  type: "http_remote_write"
+  port: 9091
+
+precompute_engine:
+  num_workers: 4
+  allowed_lateness_ms: 5000
+  max_buffer_per_series: 10000
+  flush_interval_ms: 1000
+  channel_buffer_size: 10000
+  dump_precomputes: false
+
+query_tracker:
+  enabled: false
+  observation_window_secs: 100
+
+inference_config: "/asap-planner-output/inference_config.yaml"
+streaming_config: "/asap-planner-output/streaming_config.yaml"
+promsketch_config: null
diff --git a/benchmarks/docker-compose.yml b/benchmarks/docker-compose.yml
index 11203649..e22e72d0 100644
--- a/benchmarks/docker-compose.yml
+++ b/benchmarks/docker-compose.yml
@@ -15,3 +15,7 @@ services:
 
   queryengine:
     image: ghcr.io/projectasap/asap-query-engine:${ASAP_IMAGE_TAG}
+    volumes:
+      - ./config/engine_config.yaml:/config/engine_config.yaml:ro
+    command:
+      - "--config-file=/config/engine_config.yaml"

From b886f9d753c09dde8355a52f0d7856e83f13712b Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Mon, 4 May 2026 18:25:45 -0400
Subject: [PATCH 5/6] updated cargo.lock

---
 Cargo.lock | 314 +++++++++++++++++++++++++++++------------------------
 1 file changed, 171 insertions(+), 143 deletions(-)

diff --git a/Cargo.lock b/Cargo.lock
index e66817c9..e6f69253 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -312,7 +312,7 @@ dependencies = [
  "arrow-schema",
  "chrono",
  "half",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "lexical-core",
  "num",
  "serde",
@@ -392,8 +392,8 @@ dependencies = [
  "anyhow",
  "asap_types",
  "chrono",
- "clap 4.6.0",
- "indexmap 2.13.1",
+ "clap 4.6.1",
+ "indexmap 2.14.0",
  "pretty_assertions",
  "promql-parser",
  "promql_utilities",
@@ -412,7 +412,7 @@ dependencies = [
 [[package]]
 name = "asap_sketchlib"
 version = "0.1.0"
-source = "git+https://github.com/ProjectASAP/asap_sketchlib#d22a9abadd48ea05e219558f6612a4e4e52be45b"
+source = "git+https://github.com/ProjectASAP/asap_sketchlib#81c3436dde44cc587c098d42bf42db77acdb4fa5"
 dependencies = [
  "bytes",
  "prost",
@@ -432,7 +432,7 @@ name = "asap_types"
 version = "0.3.0"
 dependencies = [
  "anyhow",
- "clap 4.6.0",
+ "clap 4.6.1",
  "promql_utilities",
  "serde",
  "serde_json",
@@ -617,9 +617,9 @@ checksum = "bef38d45163c2f1dde094a7dfd33ccf595c92905c8f8f4fdc18d06fb1037718a"
 
 [[package]]
 name = "bitflags"
-version = "2.11.0"
+version = "2.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "843867be96c8daad0d758b57df9392b6d8d271134fce549de6ce169ff98a92af"
+checksum = "c4512299f36f043ab09a583e57bceb5a5aab7a73db1805848e8fef3c9e8c78b3"
 
 [[package]]
 name = "blake2"
@@ -632,9 +632,9 @@ dependencies = [
 
 [[package]]
 name = "blake3"
-version = "1.8.4"
+version = "1.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "4d2d5991425dfd0785aed03aedcf0b321d61975c9b5b3689c774a2610ae0b51e"
+checksum = "0aa83c34e62843d924f905e0f5c866eb1dd6545fc4d719e803d9ba6030371fce"
 dependencies = [
  "arrayref",
  "arrayvec",
@@ -741,9 +741,9 @@ checksum = "37b2a672a2cb129a2e41c10b1224bb368f9f37a2b16b612598138befd7b37eb5"
 
 [[package]]
 name = "cc"
-version = "1.2.59"
+version = "1.2.61"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b7a4d3ec6524d28a329fc53654bbadc9bdd7b0431f5d65f1a56ffb28a1ee5283"
+checksum = "d16d90359e986641506914ba71350897565610e87ce0ad9e6f28569db3dd5c6d"
 dependencies = [
  "find-msvc-tools",
  "jobserver",
@@ -763,7 +763,7 @@ version = "0.13.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "7fe45e18904af7af10e4312df7c97251e98af98c70f42f1f2587aecfcbee56bf"
 dependencies = [
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "lazy_static",
  "num-traits",
  "regex",
@@ -840,9 +840,9 @@ dependencies = [
 
 [[package]]
 name = "clap"
-version = "4.6.0"
+version = "4.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b193af5b67834b676abd72466a96c1024e6a6ad978a1f484bd90b85c94041351"
+checksum = "1ddb117e43bbf7dacf0a4190fef4d345b9bad68dfc649cb349e7d17d28428e51"
 dependencies = [
  "clap_builder",
  "clap_derive",
@@ -862,9 +862,9 @@ dependencies = [
 
 [[package]]
 name = "clap_derive"
-version = "4.6.0"
+version = "4.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1110bd8a634a1ab8cb04345d8d878267d57c3cf1b38d91b71af6686408bbca6a"
+checksum = "f2ce8604710f6733aa641a2b3731eaa1e8b3d9973d5e3565da11800813f997a9"
 dependencies = [
  "heck 0.5.0",
  "proc-macro2",
@@ -991,7 +991,7 @@ dependencies = [
  "anes",
  "cast",
  "ciborium",
- "clap 4.6.0",
+ "clap 4.6.1",
  "criterion-plot",
  "is-terminal",
  "itertools 0.10.5",
@@ -1153,7 +1153,7 @@ dependencies = [
  "glob",
  "half",
  "hashbrown 0.14.5",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "itertools 0.13.0",
  "log",
  "num_cpus",
@@ -1162,7 +1162,7 @@ dependencies = [
  "parquet",
  "paste",
  "pin-project-lite",
- "rand 0.8.5",
+ "rand 0.8.6",
  "sqlparser 0.51.0",
  "tempfile",
  "tokio",
@@ -1202,7 +1202,7 @@ dependencies = [
  "chrono",
  "half",
  "hashbrown 0.14.5",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "instant",
  "libc",
  "num_cpus",
@@ -1239,7 +1239,7 @@ dependencies = [
  "log",
  "object_store",
  "parking_lot",
- "rand 0.8.5",
+ "rand 0.8.6",
  "tempfile",
  "url",
 ]
@@ -1260,7 +1260,7 @@ dependencies = [
  "datafusion-functions-aggregate-common",
  "datafusion-functions-window-common",
  "datafusion-physical-expr-common",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "paste",
  "serde_json",
  "sqlparser 0.51.0",
@@ -1300,7 +1300,7 @@ dependencies = [
  "itertools 0.13.0",
  "log",
  "md-5",
- "rand 0.8.5",
+ "rand 0.8.6",
  "regex",
  "sha2",
  "unicode-segmentation",
@@ -1323,7 +1323,7 @@ dependencies = [
  "datafusion-physical-expr",
  "datafusion-physical-expr-common",
  "half",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "log",
  "paste",
 ]
@@ -1339,7 +1339,7 @@ dependencies = [
  "datafusion-common",
  "datafusion-expr-common",
  "datafusion-physical-expr-common",
- "rand 0.8.5",
+ "rand 0.8.6",
 ]
 
 [[package]]
@@ -1362,7 +1362,7 @@ dependencies = [
  "itertools 0.13.0",
  "log",
  "paste",
- "rand 0.8.5",
+ "rand 0.8.6",
 ]
 
 [[package]]
@@ -1403,7 +1403,7 @@ dependencies = [
  "datafusion-expr",
  "datafusion-physical-expr",
  "hashbrown 0.14.5",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "itertools 0.13.0",
  "log",
  "paste",
@@ -1431,11 +1431,11 @@ dependencies = [
  "datafusion-physical-expr-common",
  "half",
  "hashbrown 0.14.5",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "itertools 0.13.0",
  "log",
  "paste",
- "petgraph",
+ "petgraph 0.6.5",
 ]
 
 [[package]]
@@ -1449,7 +1449,7 @@ dependencies = [
  "datafusion-common",
  "datafusion-expr-common",
  "hashbrown 0.14.5",
- "rand 0.8.5",
+ "rand 0.8.6",
 ]
 
 [[package]]
@@ -1493,13 +1493,13 @@ dependencies = [
  "futures",
  "half",
  "hashbrown 0.14.5",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "itertools 0.13.0",
  "log",
  "once_cell",
  "parking_lot",
  "pin-project-lite",
- "rand 0.8.5",
+ "rand 0.8.6",
  "tokio",
 ]
 
@@ -1514,7 +1514,7 @@ dependencies = [
  "arrow-schema",
  "datafusion-common",
  "datafusion-expr",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "log",
  "regex",
  "sqlparser 0.51.0",
@@ -1675,6 +1675,12 @@ version = "0.4.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "0ce7134b9999ecaf8bcd65542e436736ef32ddca1b3e06094cb6ec5755203b80"
 
+[[package]]
+name = "fixedbitset"
+version = "0.5.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1d674e81391d1e1ab681a28d99df07927c6d4aa5b027d7da16ba32d1d21ecd99"
+
 [[package]]
 name = "flatbuffers"
 version = "24.12.23"
@@ -1892,7 +1898,7 @@ dependencies = [
  "futures-sink",
  "futures-util",
  "http 0.2.12",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "slab",
  "tokio",
  "tokio-util",
@@ -1901,9 +1907,9 @@ dependencies = [
 
 [[package]]
 name = "h2"
-version = "0.4.13"
+version = "0.4.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2f44da3a8150a6703ed5d34e164b875fd14c2cdab9af1252a9a1020bde2bdc54"
+checksum = "171fefbc92fe4a4de27e0698d6a5b392d6a0e333506bc49133760b3bcf948733"
 dependencies = [
  "atomic-waker",
  "bytes",
@@ -1911,7 +1917,7 @@ dependencies = [
  "futures-core",
  "futures-sink",
  "http 1.4.0",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "slab",
  "tokio",
  "tokio-util",
@@ -1957,9 +1963,9 @@ dependencies = [
 
 [[package]]
 name = "hashbrown"
-version = "0.16.1"
+version = "0.17.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "841d1cc9bed7f9236f321df977030373f4a4163ae1a7dbfe1a51a2c1a51d9100"
+checksum = "4f467dd6dccf739c208452f8014c75c18bb8301b050ad1cfb27153803edb0f51"
 
 [[package]]
 name = "hashlink"
@@ -2113,7 +2119,7 @@ dependencies = [
  "bytes",
  "futures-channel",
  "futures-core",
- "h2 0.4.13",
+ "h2 0.4.14",
  "http 1.4.0",
  "http-body 1.0.1",
  "httparse",
@@ -2305,9 +2311,9 @@ dependencies = [
 
 [[package]]
 name = "idna_adapter"
-version = "1.2.1"
+version = "1.2.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3acae9609540aa318d1bc588455225fb2085b9ed0c4f6bd0d9d5bcd86f1a0344"
+checksum = "cb68373c0d6620ef8105e855e7745e18b0d00d3bdb07fb532e434244cdb9a714"
 dependencies = [
  "icu_normalizer",
  "icu_properties",
@@ -2325,12 +2331,12 @@ dependencies = [
 
 [[package]]
 name = "indexmap"
-version = "2.13.1"
+version = "2.14.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "45a8a2b9cb3e0b0c1803dbb0758ffac5de2f425b23c28f518faabd9d805342ff"
+checksum = "d466e9454f08e4a911e14806c24e16fba1b4c121d1ea474396f396069cf949d9"
 dependencies = [
  "equivalent",
- "hashbrown 0.16.1",
+ "hashbrown 0.17.0",
  "serde",
  "serde_core",
 ]
@@ -2394,6 +2400,15 @@ dependencies = [
  "either",
 ]
 
+[[package]]
+name = "itertools"
+version = "0.14.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "2b192c782037fadd9cfa75548310488aabdbf3d2da73885b31bd0abd03351285"
+dependencies = [
+ "either",
+]
+
 [[package]]
 name = "itoa"
 version = "1.0.18"
@@ -2402,9 +2417,9 @@ checksum = "8f42a60cbdf9a97f5d2305f08a87dc4e09308d1276d28c869c684d7777685682"
 
 [[package]]
 name = "jiff"
-version = "0.2.23"
+version = "0.2.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1a3546dc96b6d42c5f24902af9e2538e82e39ad350b0c766eb3fbf2d8f3d8359"
+checksum = "f00b5dbd620d61dfdcb6007c9c1f6054ebd75319f163d886a9055cec1155073d"
 dependencies = [
  "jiff-static",
  "jiff-tzdb-platform",
@@ -2417,9 +2432,9 @@ dependencies = [
 
 [[package]]
 name = "jiff-static"
-version = "0.2.23"
+version = "0.2.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2a8c8b344124222efd714b73bb41f8b5120b27a7cc1c75593a6ff768d9d05aa4"
+checksum = "e000de030ff8022ea1da3f466fbb0f3a809f5e51ed31f6dd931c35181ad8e6d7"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -2453,9 +2468,9 @@ dependencies = [
 
 [[package]]
 name = "js-sys"
-version = "0.3.94"
+version = "0.3.97"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2e04e2ef80ce82e13552136fabeef8a5ed1f985a96805761cbb9a2c34e7664d9"
+checksum = "a1840c94c045fbcf8ba2812c95db44499f7c64910a912551aaaa541decebcacf"
 dependencies = [
  "cfg-if",
  "futures-util",
@@ -2534,9 +2549,9 @@ dependencies = [
 
 [[package]]
 name = "libc"
-version = "0.2.184"
+version = "0.2.186"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "48f5d2a454e16a5ea0f4ced81bd44e4cfc7bd3a507b61887c99fd3538b28e4af"
+checksum = "68ab91017fe16c622486840e4c83c9a37afeff978bd239b5293d61ece587de66"
 
 [[package]]
 name = "libm"
@@ -2546,14 +2561,14 @@ checksum = "b6d2cec3eae94f9f509c767b45932f1ada8350c4bdb85af2fcab4a3c14807981"
 
 [[package]]
 name = "libredox"
-version = "0.1.15"
+version = "0.1.16"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7ddbf48fd451246b1f8c2610bd3b4ac0cc6e149d89832867093ab69a17194f08"
+checksum = "e02f3bb43d335493c96bf3fd3a321600bf6bd07ed34bc64118e9293bdffea46c"
 dependencies = [
- "bitflags 2.11.0",
+ "bitflags 2.11.1",
  "libc",
  "plain",
- "redox_syscall 0.7.3",
+ "redox_syscall 0.7.5",
 ]
 
 [[package]]
@@ -2634,7 +2649,7 @@ dependencies = [
  "cactus",
  "cfgrammar",
  "filetime",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "lazy_static",
  "lrtable",
  "num-traits",
@@ -2941,15 +2956,14 @@ checksum = "d6790f58c7ff633d8771f42965289203411a5e5c68388703c06e14f24770b41e"
 
 [[package]]
 name = "openssl"
-version = "0.10.76"
+version = "0.10.79"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "951c002c75e16ea2c65b8c7e4d3d51d5530d8dfa7d060b4776828c88cfb18ecf"
+checksum = "bf0b434746ee2832f4f0baf10137e1cabb18cbe6912c69e2e33263c45250f542"
 dependencies = [
- "bitflags 2.11.0",
+ "bitflags 2.11.1",
  "cfg-if",
  "foreign-types",
  "libc",
- "once_cell",
  "openssl-macros",
  "openssl-sys",
 ]
@@ -2973,9 +2987,9 @@ checksum = "7c87def4c32ab89d880effc9e097653c8da5d6ef28e6b539d313baaacfbafcbe"
 
 [[package]]
 name = "openssl-sys"
-version = "0.9.112"
+version = "0.9.115"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "57d55af3b3e226502be1526dfdba67ab0e9c96fc293004e79576b2b9edb0dbdb"
+checksum = "158fe5b292746440aa6e7a7e690e55aeb72d41505e2804c23c6973ad0e9c9781"
 dependencies = [
  "cc",
  "libc",
@@ -3026,7 +3040,7 @@ dependencies = [
  "glob",
  "opentelemetry",
  "percent-encoding",
- "rand 0.8.5",
+ "rand 0.8.6",
  "serde_json",
  "thiserror 2.0.18",
 ]
@@ -3138,8 +3152,18 @@ version = "0.6.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b4c5cc86750666a3ed20bdaf5ca2a0344f9c67674cae0515bec2da16fbaa47db"
 dependencies = [
- "fixedbitset",
- "indexmap 2.13.1",
+ "fixedbitset 0.4.2",
+ "indexmap 2.14.0",
+]
+
+[[package]]
+name = "petgraph"
+version = "0.7.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "3672b37090dbd86368a4145bc067582552b29c27377cad4e0a306c97f9bd7772"
+dependencies = [
+ "fixedbitset 0.5.7",
+ "indexmap 2.14.0",
 ]
 
 [[package]]
@@ -3188,9 +3212,9 @@ checksum = "a89322df9ebe1c1578d689c92318e070967d1042b512afbe49518723f4e6d5cd"
 
 [[package]]
 name = "pkg-config"
-version = "0.3.32"
+version = "0.3.33"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7edddbd0b52d732b21ad9a5fab5c704c14cd949e5e9a1ec5929a24fded1b904c"
+checksum = "19f132c84eca552bf34cab8ec81f1c1dcc229b811638f9d283dceabe58c5569e"
 
 [[package]]
 name = "plain"
@@ -3234,9 +3258,9 @@ checksum = "c33a9471896f1c69cecef8d20cbe2f7accd12527ce60845ff44c153bb2a21b49"
 
 [[package]]
 name = "portable-atomic-util"
-version = "0.2.6"
+version = "0.2.7"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "091397be61a01d4be58e7841595bd4bfedb15f1cd54977d79b8271e94ed799a3"
+checksum = "c2a106d1259c23fac8e543272398ae0e3c0b8d33c88ed73d0cc71b0f1d902618"
 dependencies = [
  "portable-atomic",
 ]
@@ -3400,11 +3424,11 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "be769465445e8c1474e9c5dac2018218498557af32d9ed057325ec9a41ae81bf"
 dependencies = [
  "heck 0.5.0",
- "itertools 0.13.0",
+ "itertools 0.14.0",
  "log",
  "multimap",
  "once_cell",
- "petgraph",
+ "petgraph 0.7.1",
  "prettyplease",
  "prost",
  "prost-types",
@@ -3420,7 +3444,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "8a56d757972c98b346a9b766e3f02746cde6dd1cd1d1d563472929fdd74bec4d"
 dependencies = [
  "anyhow",
- "itertools 0.13.0",
+ "itertools 0.14.0",
  "proc-macro2",
  "quote",
  "syn 2.0.117",
@@ -3507,9 +3531,9 @@ checksum = "95067976aca6421a523e491fce939a3e65249bac4b977adee0ee9771568e8aa3"
 
 [[package]]
 name = "psm"
-version = "0.1.30"
+version = "0.1.31"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3852766467df634d74f0b2d7819bf8dc483a0eb2e3b0f50f756f9cfe8b0d18d8"
+checksum = "645dbe486e346d9b5de3ef16ede18c26e6c70ad97418f4874b8b1889d6e761ea"
 dependencies = [
  "ar_archive_writer",
  "cc",
@@ -3530,7 +3554,7 @@ dependencies = [
  "base64 0.21.7",
  "bincode",
  "chrono",
- "clap 4.6.0",
+ "clap 4.6.1",
  "criterion",
  "csv",
  "dashmap 5.5.3",
@@ -3597,9 +3621,9 @@ checksum = "f8dcc9c7d52a811697d2151c701e0d08956f92b0e24136cf4cf27b57a6a0d9bf"
 
 [[package]]
 name = "rand"
-version = "0.8.5"
+version = "0.8.6"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "34af8d1a0e25924bc5b7c43c079c942339d8f0a8b57c39049bef581b46327404"
+checksum = "5ca0ecfa931c29007047d1bc58e623ab12e5590e8c7cc53200d5202b69266d8a"
 dependencies = [
  "libc",
  "rand_chacha 0.3.1",
@@ -3656,9 +3680,9 @@ dependencies = [
 
 [[package]]
 name = "rayon"
-version = "1.11.0"
+version = "1.12.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "368f01d005bf8fd9b1206fb6fa653e6c4a81ceb1466406b81792d87c5677a58f"
+checksum = "fb39b166781f92d482534ef4b4b1b2568f42613b53e5b6c160e24cfbfa30926d"
 dependencies = [
  "either",
  "rayon-core",
@@ -3730,16 +3754,16 @@ version = "0.5.18"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "ed2bf2547551a7053d6fdfafda3f938979645c44812fbfcda098faae3f1a362d"
 dependencies = [
- "bitflags 2.11.0",
+ "bitflags 2.11.1",
 ]
 
 [[package]]
 name = "redox_syscall"
-version = "0.7.3"
+version = "0.7.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6ce70a74e890531977d37e532c34d45e9055d2409ed08ddba14529471ed0be16"
+checksum = "4666a1a60d8412eab19d94f6d13dcc9cea0a5ef4fdf6a5db306537413c661b1b"
 dependencies = [
- "bitflags 2.11.0",
+ "bitflags 2.11.1",
 ]
 
 [[package]]
@@ -3836,7 +3860,7 @@ version = "0.31.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b838eba278d213a8beaf485bd313fd580ca4505a00d5871caeb1457c55322cae"
 dependencies = [
- "bitflags 2.11.0",
+ "bitflags 2.11.1",
  "fallible-iterator",
  "fallible-streaming-iterator",
  "hashlink",
@@ -3859,7 +3883,7 @@ version = "1.1.4"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b6fe4565b9518b83ef4f91bb47ce29620ca828bd32cb7e408f0062e9930ba190"
 dependencies = [
- "bitflags 2.11.0",
+ "bitflags 2.11.1",
  "errno",
  "libc",
  "linux-raw-sys",
@@ -3917,7 +3941,7 @@ version = "3.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "b7f4bc775c73d9a02cde8bf7b2ec4c9d12743edf609006c7facc23998404cd1d"
 dependencies = [
- "bitflags 2.11.0",
+ "bitflags 2.11.1",
  "core-foundation 0.10.1",
  "core-foundation-sys",
  "libc",
@@ -4027,7 +4051,7 @@ version = "0.9.34+deprecated"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47"
 dependencies = [
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "itoa",
  "ryu",
  "serde",
@@ -4078,9 +4102,9 @@ checksum = "703d5c7ef118737c72f1af64ad2f6f8c5e1921f818cdcb97b8fe6fc69bf66214"
 
 [[package]]
 name = "siphasher"
-version = "1.0.2"
+version = "1.0.3"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b2aa850e253778c88a04c3d7323b043aeda9d3e30d5971937c1855769763678e"
+checksum = "8ee5873ec9cce0195efcb7a4e9507a04cd49aec9c83d0389df45b1ef7ba2e649"
 
 [[package]]
 name = "slab"
@@ -4202,15 +4226,15 @@ checksum = "6ce2be8dc25455e1f91df71bfa12ad37d7af1092ae736f3a6cd0e37bc7810596"
 
 [[package]]
 name = "stacker"
-version = "0.1.23"
+version = "0.1.24"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "08d74a23609d509411d10e2176dc2a4346e3b4aea2e7b1869f19fdedbc71c013"
+checksum = "640c8cdd92b6b12f5bcb1803ca3bbf5ab96e5e6b6b96b9ab77dabe9e880b3190"
 dependencies = [
  "cc",
  "cfg-if",
  "libc",
  "psm",
- "windows-sys 0.59.0",
+ "windows-sys 0.61.2",
 ]
 
 [[package]]
@@ -4283,6 +4307,12 @@ version = "2.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "13c2bddecc57b384dee18652358fb23172facb8a2c51ccc10d74c157bdea3292"
 
+[[package]]
+name = "symlink"
+version = "0.1.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "a7973cce6668464ea31f176d85b13c7ab3bba2cb3b77a2ed26abd7801688010a"
+
 [[package]]
 name = "syn"
 version = "1.0.109"
@@ -4495,9 +4525,9 @@ dependencies = [
 
 [[package]]
 name = "tokio"
-version = "1.51.0"
+version = "1.52.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2bd1c4c0fc4a7ab90fc15ef6daaa3ec3b893f004f915f2392557ed23237820cd"
+checksum = "110a78583f19d5cdb2c5ccf321d1290344e71313c6c37d43520d386027d18386"
 dependencies = [
  "bytes",
  "libc",
@@ -4577,14 +4607,14 @@ dependencies = [
 
 [[package]]
 name = "toml_edit"
-version = "0.25.10+spec-1.1.0"
+version = "0.25.11+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "a82418ca169e235e6c399a84e395ab6debeb3bc90edc959bf0f48647c6a32d1b"
+checksum = "0b59c4d22ed448339746c59b905d24568fcbb3ab65a500494f7b8c3e97739f2b"
 dependencies = [
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "toml_datetime",
  "toml_parser",
- "winnow 1.0.1",
+ "winnow 1.0.2",
 ]
 
 [[package]]
@@ -4593,7 +4623,7 @@ version = "1.1.2+spec-1.1.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a2abe9b86193656635d2411dc43050282ca48aa31c2451210f4202550afb7526"
 dependencies = [
- "winnow 1.0.1",
+ "winnow 1.0.2",
 ]
 
 [[package]]
@@ -4607,7 +4637,7 @@ dependencies = [
  "axum",
  "base64 0.22.1",
  "bytes",
- "h2 0.4.13",
+ "h2 0.4.14",
  "http 1.4.0",
  "http-body 1.0.1",
  "http-body-util",
@@ -4637,7 +4667,7 @@ dependencies = [
  "indexmap 1.9.3",
  "pin-project",
  "pin-project-lite",
- "rand 0.8.5",
+ "rand 0.8.6",
  "slab",
  "tokio",
  "tokio-util",
@@ -4688,11 +4718,12 @@ dependencies = [
 
 [[package]]
 name = "tracing-appender"
-version = "0.2.4"
+version = "0.2.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "786d480bce6247ab75f005b14ae1624ad978d3029d9113f0a22fa1ac773faeaf"
+checksum = "050686193eb999b4bb3bc2acfa891a13da00f79734704c4b8b4ef1a10b368a3c"
 dependencies = [
  "crossbeam-channel",
+ "symlink",
  "thiserror 2.0.18",
  "time",
  "tracing-subscriber",
@@ -4775,9 +4806,9 @@ dependencies = [
 
 [[package]]
 name = "typenum"
-version = "1.19.0"
+version = "1.20.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "562d481066bde0658276a35467c4af00bdc6ee726305698a55b86e61d7ad82bb"
+checksum = "40ce102ab67701b8526c123c1bab5cbe42d7040ccfd0f64af1a385808d2f43de"
 
 [[package]]
 name = "uncased"
@@ -4856,9 +4887,9 @@ checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821"
 
 [[package]]
 name = "uuid"
-version = "1.23.0"
+version = "1.23.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5ac8b6f42ead25368cf5b098aeb3dc8a1a2c05a3eee8a9a1a68c640edbfc79d9"
+checksum = "ddd74a9687298c6858e9b88ec8935ec45d22e8fd5e6394fa1bd4e99a87789c76"
 dependencies = [
  "getrandom 0.4.2",
  "js-sys",
@@ -4937,11 +4968,11 @@ checksum = "ccf3ec651a847eb01de73ccad15eb7d99f80485de043efb2f370cd654f4ea44b"
 
 [[package]]
 name = "wasip2"
-version = "1.0.2+wasi-0.2.9"
+version = "1.0.3+wasi-0.2.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "9517f9239f02c069db75e65f174b3da828fe5f5b945c4dd26bd25d89c03ebcf5"
+checksum = "20064672db26d7cdc89c7798c48a0fdfac8213434a1186e5ef29fd560ae223d6"
 dependencies = [
- "wit-bindgen",
+ "wit-bindgen 0.57.1",
 ]
 
 [[package]]
@@ -4950,14 +4981,14 @@ version = "0.4.0+wasi-0.3.0-rc-2026-01-06"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5428f8bf88ea5ddc08faddef2ac4a67e390b88186c703ce6dbd955e1c145aca5"
 dependencies = [
- "wit-bindgen",
+ "wit-bindgen 0.51.0",
 ]
 
 [[package]]
 name = "wasm-bindgen"
-version = "0.2.117"
+version = "0.2.120"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0551fc1bb415591e3372d0bc4780db7e587d84e2a7e79da121051c5c4b89d0b0"
+checksum = "df52b6d9b87e0c74c9edfa1eb2d9bf85e5d63515474513aa50fa181b3c4f5db1"
 dependencies = [
  "cfg-if",
  "once_cell",
@@ -4968,9 +4999,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-futures"
-version = "0.4.67"
+version = "0.4.70"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "03623de6905b7206edd0a75f69f747f134b7f0a2323392d664448bf2d3c5d87e"
+checksum = "af934872acec734c2d80e6617bbb5ff4f12b052dd8e6332b0817bce889516084"
 dependencies = [
  "js-sys",
  "wasm-bindgen",
@@ -4978,9 +5009,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro"
-version = "0.2.117"
+version = "0.2.120"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "7fbdf9a35adf44786aecd5ff89b4563a90325f9da0923236f6104e603c7e86be"
+checksum = "78b1041f495fb322e64aca85f5756b2172e35cd459376e67f2a6c9dffcedb103"
 dependencies = [
  "quote",
  "wasm-bindgen-macro-support",
@@ -4988,9 +5019,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-macro-support"
-version = "0.2.117"
+version = "0.2.120"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "dca9693ef2bab6d4e6707234500350d8dad079eb508dca05530c85dc3a529ff2"
+checksum = "9dcd0ff20416988a18ac686d4d4d0f6aae9ebf08a389ff5d29012b05af2a1b41"
 dependencies = [
  "bumpalo",
  "proc-macro2",
@@ -5001,9 +5032,9 @@ dependencies = [
 
 [[package]]
 name = "wasm-bindgen-shared"
-version = "0.2.117"
+version = "0.2.120"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "39129a682a6d2d841b6c429d0c51e5cb0ed1a03829d8b3d1e69a011e62cb3d3b"
+checksum = "49757b3c82ebf16c57d69365a142940b384176c24df52a087fb748e2085359ea"
 dependencies = [
  "unicode-ident",
 ]
@@ -5025,7 +5056,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "bb0e353e6a2fbdc176932bbaab493762eb1255a7900fe0fea1a2f96c296cc909"
 dependencies = [
  "anyhow",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "wasm-encoder",
  "wasmparser",
 ]
@@ -5036,17 +5067,17 @@ version = "0.244.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "47b807c72e1bac69382b3a6fb3dbe8ea4c0ed87ff5629b8685ae6b9a611028fe"
 dependencies = [
- "bitflags 2.11.0",
+ "bitflags 2.11.1",
  "hashbrown 0.15.5",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "semver",
 ]
 
 [[package]]
 name = "web-sys"
-version = "0.3.94"
+version = "0.3.97"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "cd70027e39b12f0849461e08ffc50b9cd7688d942c1c8e3c7b22273236b4dd0a"
+checksum = "2eadbac71025cd7b0834f20d1fe8472e8495821b4e9801eb0a60bd1f19827602"
 dependencies = [
  "js-sys",
  "wasm-bindgen",
@@ -5160,15 +5191,6 @@ dependencies = [
  "windows-targets 0.52.6",
 ]
 
-[[package]]
-name = "windows-sys"
-version = "0.59.0"
-source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "1e38bc4d79ed67fd075bcc251a1c39b32a1776bbe92e5bef1f0bf1f8c531853b"
-dependencies = [
- "windows-targets 0.52.6",
-]
-
 [[package]]
 name = "windows-sys"
 version = "0.61.2"
@@ -5310,9 +5332,9 @@ dependencies = [
 
 [[package]]
 name = "winnow"
-version = "1.0.1"
+version = "1.0.2"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "09dac053f1cd375980747450bfc7250c264eaae0583872e845c0c7cd578872b5"
+checksum = "2ee1708bef14716a11bae175f579062d4554d95be2c6829f518df847b7b3fdd0"
 dependencies = [
  "memchr",
 ]
@@ -5336,6 +5358,12 @@ dependencies = [
  "wit-bindgen-rust-macro",
 ]
 
+[[package]]
+name = "wit-bindgen"
+version = "0.57.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1ebf944e87a7c253233ad6766e082e3cd714b5d03812acc24c318f549614536e"
+
 [[package]]
 name = "wit-bindgen-core"
 version = "0.51.0"
@@ -5355,7 +5383,7 @@ checksum = "b7c566e0f4b284dd6561c786d9cb0142da491f46a9fbed79ea69cdad5db17f21"
 dependencies = [
  "anyhow",
  "heck 0.5.0",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "prettyplease",
  "syn 2.0.117",
  "wasm-metadata",
@@ -5385,8 +5413,8 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9d66ea20e9553b30172b5e831994e35fbde2d165325bec84fc43dbf6f4eb9cb2"
 dependencies = [
  "anyhow",
- "bitflags 2.11.0",
- "indexmap 2.13.1",
+ "bitflags 2.11.1",
+ "indexmap 2.14.0",
  "log",
  "serde",
  "serde_derive",
@@ -5405,7 +5433,7 @@ checksum = "ecc8ac4bc1dc3381b7f59c34f00b67e18f910c2c0f50015669dde7def656a736"
 dependencies = [
  "anyhow",
  "id-arena",
- "indexmap 2.13.1",
+ "indexmap 2.14.0",
  "log",
  "semver",
  "serde",

From 553df42492f5aa8389368cfe3fe302822861aa48 Mon Sep 17 00:00:00 2001
From: Milind Srivastava <milindsrivastava1997@gmail.com>
Date: Mon, 4 May 2026 20:28:33 -0400
Subject: [PATCH 6/6] fixed CI

---
 benchmarks/docker-compose.yml | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/benchmarks/docker-compose.yml b/benchmarks/docker-compose.yml
index e22e72d0..bab96d9b 100644
--- a/benchmarks/docker-compose.yml
+++ b/benchmarks/docker-compose.yml
@@ -16,6 +16,6 @@ services:
   queryengine:
     image: ghcr.io/projectasap/asap-query-engine:${ASAP_IMAGE_TAG}
     volumes:
-      - ./config/engine_config.yaml:/config/engine_config.yaml:ro
+      - ../benchmarks/config/engine_config.yaml:/config/engine_config.yaml:ro
     command:
       - "--config-file=/config/engine_config.yaml"