implemented plan

Author: milindsrivastava1997

Cargo.lock

@@ -1,4 +1,6 @@
 use asap_types::enums::CleanupPolicy;
+use asap_types::inference_config::InferenceConfig;
+use asap_types::streaming_config::StreamingConfig;
 use asap_types::PromQLSchema;
 use promql_utilities::data_model::KeyByLabelNames;
 use serde::Deserialize;
@@ -13,6 +15,18 @@ pub struct ControllerConfig {
     /// returns no series for a metric. Prometheus-inferred labels take priority.
     #[serde(default)]
     pub metrics: Option<Vec<MetricDefinition>>,
+    /// Current streaming config, passed as context for repeated reconfiguration.
+    /// NOTE: reserved for future use — the planner does not yet act on these fields.
+    /// They are wired through now so that repeated-reconfig support can be added
+    /// without a second round of type-signature changes.
+    #[serde(default)]
+    pub existing_streaming_config: Option<StreamingConfig>,
+    /// Current inference config, passed as context for repeated reconfiguration.
+    /// NOTE: see existing_streaming_config — same future-use caveat applies.
+    /// Not serializable via serde (InferenceConfig does not impl Deserialize);
+    /// set programmatically only.
+    #[serde(skip)]
+    pub existing_inference_config: Option<InferenceConfig>,
 }
 
 impl ControllerConfig {

asap-planner-rs/src/config/input.rs

@@ -42,5 +42,7 @@ pub fn to_controller_config(
             policy: Some(CleanupPolicy::ReadBased),
         }),
         metrics: None,
+        existing_streaming_config: None,
+        existing_inference_config: None,
     }
 }

asap-planner-rs/src/query_log/converter.rs

@@ -57,6 +57,7 @@ lazy_static = "1.4"
 zstd = "0.13"
 reqwest = { version = "0.11", features = ["json"] }
 tracing-appender = "0.2"
+arc-swap = "1"
 elastic_dsl_utilities.workspace = true
 asap_sketchlib = { git = "https://github.com/ProjectASAP/asap_sketchlib" }
 

asap-query-engine/Cargo.toml

@@ -52,7 +52,7 @@ impl SimpleEngine {
         // TODO: Figure out how to handle query configuration for ElasticSearch queries.
         let query_config = self.find_query_config(&query)?;
         let agg_info = self
-            .get_aggregation_id_info(query_config)
+            .get_aggregation_id_info(&query_config)
             .map_err(|e| {
                 warn!("{}", e);
                 e
@@ -76,14 +76,13 @@ impl SimpleEngine {
             })
             .ok()?;
 
-        let grouping_labels = self
-            .streaming_config
+        let sc = self.streaming_config.read().unwrap().clone();
+        let grouping_labels = sc
             .get_aggregation_config(agg_info.aggregation_id_for_value)
             .map(|config| config.grouping_labels.clone())
             .unwrap_or_else(|| query_metadata.query_output_labels.clone());
 
-        let aggregated_labels = self
-            .streaming_config
+        let aggregated_labels = sc
             .get_aggregation_config(agg_info.aggregation_id_for_key)
             .map(|config| config.aggregated_labels.clone())
             .unwrap_or_else(KeyByLabelNames::empty);

asap-query-engine/src/engines/simple_engine/elastic.rs

@@ -12,7 +12,7 @@ use crate::engines::query_result::{InstantVectorElement, QueryResult};
 // };
 use crate::stores::{Store, TimestampedBucketsMap};
 use std::collections::HashMap;
-use std::sync::Arc;
+use std::sync::{Arc, RwLock};
 use std::time::Instant;
 use tracing::{debug, warn};
 
@@ -125,8 +125,12 @@ pub struct RangeQueryExecutionContext {
 pub struct SimpleEngine {
     store: Arc<dyn Store>,
     // promsketch_store: Option<Arc<PromSketchStore>>,
-    inference_config: InferenceConfig,
-    streaming_config: Arc<StreamingConfig>,
+    /// Updated at runtime via update_inference_config(). RwLock provides interior
+    /// mutability since SimpleEngine is shared behind Arc<SimpleEngine>.
+    inference_config: RwLock<InferenceConfig>,
+    /// Updated at runtime via update_streaming_config(). Readers briefly lock to
+    /// clone the Arc pointer, then use without holding the lock.
+    streaming_config: RwLock<Arc<StreamingConfig>>,
     prometheus_scrape_interval: u64,
     controller_patterns: HashMap<QueryPatternType, Vec<PromQLPattern>>,
     query_language: QueryLanguage,
@@ -276,25 +280,45 @@ impl SimpleEngine {
         Self {
             store,
             // promsketch_store,
-            inference_config,
-            streaming_config,
+            inference_config: RwLock::new(inference_config),
+            streaming_config: RwLock::new(streaming_config),
             prometheus_scrape_interval,
             controller_patterns,
             query_language,
         }
     }
 
+    /// Replace the inference config at runtime. Called by the applier task after
+    /// the planner fires.
+    ///
+    /// NOTE: streaming_config and inference_config are applied to their respective
+    /// components independently (not atomically). A brief window may exist where
+    /// the precompute engine has a new streaming_config but this engine still uses
+    /// the old inference_config, causing query misses that fall back to Prometheus.
+    pub fn update_inference_config(&self, new_config: InferenceConfig) {
+        *self.inference_config.write().unwrap() = new_config;
+    }
+
+    /// Replace the streaming config at runtime. Called by the applier task after
+    /// the planner fires.
+    pub fn update_streaming_config(&self, new_config: Arc<StreamingConfig>) {
+        *self.streaming_config.write().unwrap() = new_config;
+    }
+
     /// Convert query timestamp (seconds) to data timestamp (milliseconds)
     pub fn convert_query_time_to_data_time(query_time: f64) -> u64 {
         (query_time * 1000.0) as u64
     }
 
     /// Finds the query configuration for a given query string
-    fn find_query_config(&self, query: &str) -> Option<&QueryConfig> {
+    fn find_query_config(&self, query: &str) -> Option<QueryConfig> {
         self.inference_config
+            .read()
+            .unwrap()
             .query_configs
             .iter()
             .find(|config| config.query == query)
+            .cloned()
     }
 
     /// Validates and potentially aligns end timestamp based on query pattern
@@ -344,6 +368,8 @@ impl SimpleEngine {
                 // Latest window only
                 let window_size = self
                     .streaming_config
+                    .read()
+                    .unwrap()
                     .get_aggregation_config(agg_info.aggregation_id_for_key)
                     .map(|config| config.window_size * 1000)
                     .ok_or_else(|| {
@@ -375,9 +401,9 @@ impl SimpleEngine {
         timestamps: &QueryTimestamps,
         agg_info: &AggregationIdInfo,
     ) -> Result<StoreQueryPlan, String> {
+        let sc = self.streaming_config.read().unwrap().clone();
         // Get aggregation config for value to determine window type
-        let aggregation_config_for_value = self
-            .streaming_config
+        let aggregation_config_for_value = sc
             .get_aggregation_config(agg_info.aggregation_id_for_value)
             .ok_or_else(|| {
                 format!(
@@ -896,10 +922,10 @@ impl SimpleEngine {
         let mut aggregation_type_for_key: Option<AggregationType> = None;
         let mut aggregation_type_for_value: Option<AggregationType> = None;
 
+        let sc = self.streaming_config.read().unwrap().clone();
         if query_config_aggregations.len() == 2 {
             for aggregation in query_config_aggregations {
-                let aggregation_type = self
-                    .streaming_config
+                let aggregation_type = sc
                     .get_aggregation_config(aggregation.aggregation_id)
                     .map(|config| config.aggregation_type)
                     .ok_or_else(|| {
@@ -935,8 +961,7 @@ impl SimpleEngine {
         } else {
             // Single aggregation: key and value share the same aggregation
             let id = query_config_aggregations[0].aggregation_id;
-            let agg_type = self
-                .streaming_config
+            let agg_type = sc
                 .get_aggregation_config(id)
                 .map(|config| config.aggregation_type)
                 .ok_or_else(|| format!("No streaming config for aggregation_id {id}"))?;

asap-query-engine/src/engines/simple_engine/mod.rs

@@ -155,14 +155,20 @@ impl SimpleEngine {
     pub fn find_query_config_promql_structural(
         &self,
         arm_ast: &promql_parser::parser::Expr,
-    ) -> Option<&QueryConfig> {
+    ) -> Option<QueryConfig> {
         let arm_canonical = format!("{}", arm_ast);
-        self.inference_config.query_configs.iter().find(|config| {
-            let config_canonical = promql_parser::parser::parse(&config.query)
-                .map(|ast| format!("{}", ast))
-                .unwrap_or_default();
-            config_canonical == arm_canonical
-        })
+        self.inference_config
+            .read()
+            .unwrap()
+            .query_configs
+            .iter()
+            .find(|config| {
+                let config_canonical = promql_parser::parser::parse(&config.query)
+                    .map(|ast| format!("{}", ast))
+                    .unwrap_or_default();
+                config_canonical == arm_canonical
+            })
+            .cloned()
     }
 
     /// Variant of `build_query_execution_context_promql` that accepts a pre-parsed
@@ -222,7 +228,8 @@ impl SimpleEngine {
     ) -> Option<QueryExecutionContext> {
         let (metric, spatial_filter) = get_metric_and_spatial_filter(match_result);
 
-        let promql_schema = match &self.inference_config.schema {
+        let ic = self.inference_config.read().unwrap();
+        let promql_schema = match &ic.schema {
             SchemaConfig::PromQL(schema) => schema,
             _ => return None,
         };
@@ -299,14 +306,13 @@ impl SimpleEngine {
         let do_merge = query_pattern_type == QueryPatternType::OnlyTemporal
             || query_pattern_type == QueryPatternType::OneTemporalOneSpatial;
 
-        let grouping_labels = self
-            .streaming_config
+        let sc = self.streaming_config.read().unwrap().clone();
+        let grouping_labels = sc
             .get_aggregation_config(agg_info.aggregation_id_for_value)
             .map(|config| config.grouping_labels.clone())
             .unwrap_or_else(|| query_output_labels.clone());
 
-        let aggregated_labels = self
-            .streaming_config
+        let aggregated_labels = sc
             .get_aggregation_config(agg_info.aggregation_id_for_key)
             .map(|config| config.aggregated_labels.clone())
             .unwrap_or_else(KeyByLabelNames::empty);
@@ -355,7 +361,7 @@ impl SimpleEngine {
             other => {
                 // Leaf pattern: structural config lookup + context + plan
                 let config = self.find_query_config_promql_structural(other)?;
-                let ctx = self.build_query_execution_context_from_ast(other, config, time)?;
+                let ctx = self.build_query_execution_context_from_ast(other, &config, time)?;
                 let label_names = ctx.metadata.query_output_labels.labels.clone();
                 let plan = ctx.to_logical_plan().ok()?;
                 Some((plan, label_names))
@@ -463,7 +469,7 @@ impl SimpleEngine {
             other => {
                 let config = self.find_query_config_promql_structural(other)?;
                 let base_context =
-                    self.build_query_execution_context_from_ast(other, config, end)?;
+                    self.build_query_execution_context_from_ast(other, &config, end)?;
                 let label_names = base_context.metadata.query_output_labels.labels.clone();
 
                 let start_ms = Self::convert_query_time_to_data_time(start);
@@ -472,6 +478,8 @@ impl SimpleEngine {
 
                 let tumbling_window_ms = self
                     .streaming_config
+                    .read()
+                    .unwrap()
                     .get_aggregation_config(base_context.agg_info.aggregation_id_for_value)
                     .map(|c| c.window_size * 1000)?;
 
@@ -619,7 +627,7 @@ impl SimpleEngine {
                 .map(|d| d.num_seconds() as u64 * 1000),
         };
 
-        let all_labels = match &self.inference_config.schema {
+        let all_labels = match &self.inference_config.read().unwrap().schema {
             SchemaConfig::PromQL(schema) => schema
                 .get_labels(&metric)
                 .cloned()
@@ -1053,7 +1061,7 @@ impl SimpleEngine {
 
         // Resolve aggregation: try pre-configured query_configs first, fall back to capability matching.
         let agg_info: AggregationIdInfo = if let Some(config) = self.find_query_config(&query) {
-            self.get_aggregation_id_info(config)
+            self.get_aggregation_id_info(&config)
                 .map_err(|e| {
                     warn!("{}", e);
                     e
@@ -1067,6 +1075,9 @@ impl SimpleEngine {
             let requirements =
                 self.build_query_requirements_promql(&match_result, query_pattern_type);
             self.streaming_config
+                .read()
+                .unwrap()
+                .clone()
                 .find_compatible_aggregation(&requirements)?
         };
 
@@ -1106,6 +1117,8 @@ impl SimpleEngine {
         // Get window size
         let tumbling_window_ms = self
             .streaming_config
+            .read()
+            .unwrap()
             .get_aggregation_config(base_context.agg_info.aggregation_id_for_value)
             .map(|config| config.window_size * 1000)?;