From 1ff29b0925a0f2bd38282f2c95ad5e4874df1de5 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 18 Sep 2026 09:17:40 -0400 Subject: [PATCH 1/7] feat(query-engine): add native query plan compiler --- asap-query-engine/src/engines/mod.rs | 1 + asap-query-engine/src/engines/query_plan.rs | 235 ++++++++++++++++++ .../src/engines/simple_engine/mod.rs | 10 + 3 files changed, 246 insertions(+) create mode 100644 asap-query-engine/src/engines/query_plan.rs diff --git a/asap-query-engine/src/engines/mod.rs b/asap-query-engine/src/engines/mod.rs index 65da4653..d1ad805e 100644 --- a/asap-query-engine/src/engines/mod.rs +++ b/asap-query-engine/src/engines/mod.rs @@ -1,4 +1,5 @@ pub(crate) mod merge_utils; +pub mod query_plan; pub mod query_result; pub mod simple_engine; pub(crate) mod sliding_window_composition; diff --git a/asap-query-engine/src/engines/query_plan.rs b/asap-query-engine/src/engines/query_plan.rs new file mode 100644 index 00000000..b8f5971a --- /dev/null +++ b/asap-query-engine/src/engines/query_plan.rs @@ -0,0 +1,235 @@ +//! Explicit, request-specific physical plans for native queries. + +use crate::engines::simple_engine::{RangeQueryExecutionContext, StoreQueryParams}; +use asap_types::enums::WindowType; +use promql_utilities::query_logics::enums::Statistic; + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum StoreReadStrategy { + WindowGrid, + SlidingExactCover, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum QueryPlanStep { + StoreRead { + metric: String, + aggregation_id: u64, + strategy: StoreReadStrategy, + }, + ComposeWindows { + output_count: usize, + }, + ResolveKeys, + Estimate { + statistic: Statistic, + }, + LimitTopK, + Format, +} + +#[derive(Debug, Clone)] +pub struct QueryPlan { + steps: Vec, +} + +#[derive(Debug, Clone, Copy)] +pub struct PlanOptions { + pub limit_topk: bool, + pub format_output: bool, +} + +impl QueryPlan { + pub fn compile_range(context: &RangeQueryExecutionContext, options: PlanOptions) -> Self { + let mut steps = vec![Self::store_read( + &context.base.store_plan.values_query, + context.window_type, + )]; + steps.push(QueryPlanStep::ComposeWindows { + output_count: context.output_timestamps.len(), + }); + if let Some(keys) = &context.base.store_plan.keys_query { + steps.push(Self::store_read( + keys, + context.keys_window_type.unwrap_or(context.window_type), + )); + steps.push(QueryPlanStep::ComposeWindows { + output_count: context.output_timestamps.len(), + }); + } + steps.push(QueryPlanStep::ResolveKeys); + steps.push(QueryPlanStep::Estimate { + statistic: context.base.metadata.statistic_to_compute, + }); + if options.limit_topk && context.base.metadata.statistic_to_compute == Statistic::Topk { + steps.push(QueryPlanStep::LimitTopK); + } + if options.format_output { + steps.push(QueryPlanStep::Format); + } + Self { steps } + } + + fn store_read(query: &StoreQueryParams, window_type: WindowType) -> QueryPlanStep { + let strategy = match window_type { + WindowType::Tumbling => StoreReadStrategy::WindowGrid, + WindowType::Sliding => StoreReadStrategy::SlidingExactCover, + }; + QueryPlanStep::StoreRead { + metric: query.metric.clone(), + aggregation_id: query.aggregation_id, + strategy, + } + } + + pub fn steps(&self) -> &[QueryPlanStep] { + &self.steps + } + + pub fn explain(&self) -> String { + self.steps + .iter() + .map(|step| match step { + QueryPlanStep::StoreRead { + metric, + aggregation_id, + strategy, + } => format!("StoreRead({strategy:?}, {metric}#{aggregation_id})"), + QueryPlanStep::ComposeWindows { output_count } => { + format!("ComposeWindows({output_count} outputs)") + } + QueryPlanStep::ResolveKeys => "ResolveKeys".into(), + QueryPlanStep::Estimate { statistic } => format!("Estimate({statistic})"), + QueryPlanStep::LimitTopK => "LimitTopK".into(), + QueryPlanStep::Format => "Format".into(), + }) + .collect::>() + .join(" -> ") + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::data_model::AggregationIdInfo; + use crate::engines::simple_engine::{QueryExecutionContext, QueryMetadata, StoreQueryPlan}; + use promql_utilities::data_model::KeyByLabelNames; + use promql_utilities::query_logics::enums::AggregationType; + use std::collections::HashMap; + + fn context(window_type: WindowType) -> RangeQueryExecutionContext { + let metadata = QueryMetadata { + query_output_labels: KeyByLabelNames::empty(), + statistic_to_compute: Statistic::Sum, + query_kwargs: HashMap::new(), + keep_metric_name: false, + }; + let query = StoreQueryParams { + metric: "requests".into(), + aggregation_id: 7, + start_timestamp: 0, + end_timestamp: 1_000, + }; + let base = QueryExecutionContext { + metric: "requests".into(), + metadata, + store_plan: StoreQueryPlan { + values_query: query, + keys_query: None, + }, + agg_info: AggregationIdInfo { + aggregation_id_for_key: 7, + aggregation_id_for_value: 7, + aggregation_type_for_key: AggregationType::Sum, + aggregation_type_for_value: AggregationType::Sum, + }, + value_window_type: window_type, + do_merge: false, + spatial_filter: String::new(), + query_time: 1_000, + grouping_labels: KeyByLabelNames::empty(), + aggregated_labels: KeyByLabelNames::empty(), + }; + RangeQueryExecutionContext { + base, + output_timestamps: vec![1_000], + query_range_ms: 1_000, + buckets_per_step: 1, + lookback_bucket_count: 1, + tumbling_window_ms: 1_000, + window_type, + window_size_ms: 1_000, + keys_window_type: None, + keys_window_size_ms: None, + keys_lookback_ms: None, + keys_tumbling_window_ms: None, + } + } + + #[test] + fn compiles_a_leaf_context_into_an_explainable_plan() { + let plan = QueryPlan::compile_range( + &context(WindowType::Tumbling), + PlanOptions { + limit_topk: false, + format_output: false, + }, + ); + assert_eq!(plan.explain(), "StoreRead(WindowGrid, requests#7) -> ComposeWindows(1 outputs) -> ResolveKeys -> Estimate(sum)"); + } + + #[test] + fn compiles_a_separate_keys_branch_before_key_resolution() { + let mut context = context(WindowType::Tumbling); + context.base.store_plan.keys_query = Some(StoreQueryParams { + metric: "requests".into(), + aggregation_id: 8, + start_timestamp: 0, + end_timestamp: 1_000, + }); + context.keys_window_type = Some(WindowType::Sliding); + + let plan = QueryPlan::compile_range( + &context, + PlanOptions { + limit_topk: false, + format_output: false, + }, + ); + + assert_eq!(plan.explain(), "StoreRead(WindowGrid, requests#7) -> ComposeWindows(1 outputs) -> StoreRead(SlidingExactCover, requests#8) -> ComposeWindows(1 outputs) -> ResolveKeys -> Estimate(sum)"); + } + + #[test] + fn compiles_sliding_windows_as_an_exact_cover_read() { + let plan = QueryPlan::compile_range( + &context(WindowType::Sliding), + PlanOptions { + limit_topk: false, + format_output: false, + }, + ); + + assert!(plan + .explain() + .starts_with("StoreRead(SlidingExactCover, requests#7)")); + } + + #[test] + fn adds_requested_topk_limit_and_formatting() { + let mut context = context(WindowType::Tumbling); + context.base.metadata.statistic_to_compute = Statistic::Topk; + + let plan = QueryPlan::compile_range( + &context, + PlanOptions { + limit_topk: true, + format_output: true, + }, + ); + + assert!(plan + .explain() + .ends_with("Estimate(topk) -> LimitTopK -> Format")); + } +} diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index bbb569c1..234659dc 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -6,6 +6,7 @@ use crate::data_model::{ AggregationIdInfo, InferenceConfig, KeyByLabelValues, QueryBounds, QueryConfig, QueryLanguage, StreamingConfig, }; +use crate::engines::query_plan::{PlanOptions, QueryPlan}; use crate::engines::query_result::{InstantVectorElement, QueryResult}; use crate::engines::sliding_window_composition::{plan_exact_cover, SlidingWindowSpec}; // use crate::stores::promsketch_store::{ @@ -1187,6 +1188,15 @@ impl SimpleEngine { ) })?; + let plan = QueryPlan::compile_range( + &range_context, + PlanOptions { + limit_topk: enable_topk_limiting, + format_output: enable_topk_formatting, + }, + ); + debug!(plan = %plan.explain(), "Compiled native query plan"); + let range_results = self.execute_range_query_pipeline( &range_context, enable_topk_limiting, From 8d71f93cbe6145524abc12b484bc5a8fb74ca0a7 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 18 Sep 2026 13:45:36 -0400 Subject: [PATCH 2/7] refactor(query-engine): log native query DAGs --- asap-query-engine/src/engines/mod.rs | 2 +- asap-query-engine/src/engines/query_plan.rs | 336 ++++++++---------- .../src/engines/query_plan/README.md | 16 + .../src/engines/simple_engine/mod.rs | 28 +- .../src/engines/simple_engine/promql.rs | 10 +- 5 files changed, 193 insertions(+), 199 deletions(-) create mode 100644 asap-query-engine/src/engines/query_plan/README.md diff --git a/asap-query-engine/src/engines/mod.rs b/asap-query-engine/src/engines/mod.rs index d1ad805e..9d333640 100644 --- a/asap-query-engine/src/engines/mod.rs +++ b/asap-query-engine/src/engines/mod.rs @@ -1,5 +1,5 @@ pub(crate) mod merge_utils; -pub mod query_plan; +pub(crate) mod query_plan; pub mod query_result; pub mod simple_engine; pub(crate) mod sliding_window_composition; diff --git a/asap-query-engine/src/engines/query_plan.rs b/asap-query-engine/src/engines/query_plan.rs index b8f5971a..e6d1b07e 100644 --- a/asap-query-engine/src/engines/query_plan.rs +++ b/asap-query-engine/src/engines/query_plan.rs @@ -1,235 +1,203 @@ -//! Explicit, request-specific physical plans for native queries. +//! Request-specific native query DAGs. use crate::engines::simple_engine::{RangeQueryExecutionContext, StoreQueryParams}; use asap_types::enums::WindowType; use promql_utilities::query_logics::enums::Statistic; #[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum StoreReadStrategy { +pub(crate) struct NodeId(usize); + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) enum StoreReadStrategy { WindowGrid, SlidingExactCover, } -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum QueryPlanStep { +#[derive(Debug, Clone)] +pub(crate) enum QueryPlanNode { StoreRead { - metric: String, - aggregation_id: u64, + query: StoreQueryParams, strategy: StoreReadStrategy, }, ComposeWindows { - output_count: usize, + input: NodeId, + output_timestamps: Vec, + lookback_ms: u64, + window_size_ms: u64, + bucket_step_ms: u64, + }, + ResolveKeys { + values: NodeId, + keys: Option, }, - ResolveKeys, Estimate { + input: NodeId, statistic: Statistic, + query_kwargs: std::collections::HashMap, + }, + LimitTopK { + input: NodeId, + k: String, + }, + Format { + input: NodeId, + include_metric_name: bool, }, - LimitTopK, - Format, } #[derive(Debug, Clone)] -pub struct QueryPlan { - steps: Vec, +pub(crate) struct QueryPlan { + nodes: Vec, + root: NodeId, } #[derive(Debug, Clone, Copy)] -pub struct PlanOptions { +pub(crate) struct PlanOptions { pub limit_topk: bool, pub format_output: bool, } impl QueryPlan { - pub fn compile_range(context: &RangeQueryExecutionContext, options: PlanOptions) -> Self { - let mut steps = vec![Self::store_read( + pub(crate) fn compile_range( + context: &RangeQueryExecutionContext, + options: PlanOptions, + ) -> Self { + let mut nodes = Vec::new(); + let values_read = Self::push_read( + &mut nodes, &context.base.store_plan.values_query, context.window_type, - )]; - steps.push(QueryPlanStep::ComposeWindows { - output_count: context.output_timestamps.len(), - }); - if let Some(keys) = &context.base.store_plan.keys_query { - steps.push(Self::store_read( - keys, + ); + let values = Self::push_compose( + &mut nodes, + values_read, + &context.output_timestamps, + context.query_range_ms, + context.window_size_ms, + context.tumbling_window_ms, + ); + let keys = context.base.store_plan.keys_query.as_ref().map(|query| { + let read = Self::push_read( + &mut nodes, + query, context.keys_window_type.unwrap_or(context.window_type), - )); - steps.push(QueryPlanStep::ComposeWindows { - output_count: context.output_timestamps.len(), - }); - } - steps.push(QueryPlanStep::ResolveKeys); - steps.push(QueryPlanStep::Estimate { - statistic: context.base.metadata.statistic_to_compute, + ); + Self::push_compose( + &mut nodes, + read, + &context.output_timestamps, + context.keys_lookback_ms.unwrap_or(context.query_range_ms), + context + .keys_window_size_ms + .unwrap_or(context.window_size_ms), + context + .keys_tumbling_window_ms + .unwrap_or(context.tumbling_window_ms), + ) }); + let resolved = Self::push(&mut nodes, QueryPlanNode::ResolveKeys { values, keys }); + let mut root = Self::push( + &mut nodes, + QueryPlanNode::Estimate { + input: resolved, + statistic: context.base.metadata.statistic_to_compute, + query_kwargs: context.base.metadata.query_kwargs.clone(), + }, + ); if options.limit_topk && context.base.metadata.statistic_to_compute == Statistic::Topk { - steps.push(QueryPlanStep::LimitTopK); + let k = context + .base + .metadata + .query_kwargs + .get("k") + .cloned() + .unwrap_or_else(|| "".to_string()); + root = Self::push(&mut nodes, QueryPlanNode::LimitTopK { input: root, k }); } if options.format_output { - steps.push(QueryPlanStep::Format); + root = Self::push( + &mut nodes, + QueryPlanNode::Format { + input: root, + include_metric_name: context.base.metadata.keep_metric_name, + }, + ); } - Self { steps } + Self { nodes, root } } - fn store_read(query: &StoreQueryParams, window_type: WindowType) -> QueryPlanStep { + fn push(nodes: &mut Vec, node: QueryPlanNode) -> NodeId { + let id = NodeId(nodes.len()); + nodes.push(node); + id + } + + fn push_read( + nodes: &mut Vec, + query: &StoreQueryParams, + window_type: WindowType, + ) -> NodeId { let strategy = match window_type { WindowType::Tumbling => StoreReadStrategy::WindowGrid, WindowType::Sliding => StoreReadStrategy::SlidingExactCover, }; - QueryPlanStep::StoreRead { - metric: query.metric.clone(), - aggregation_id: query.aggregation_id, - strategy, - } - } - - pub fn steps(&self) -> &[QueryPlanStep] { - &self.steps - } - - pub fn explain(&self) -> String { - self.steps - .iter() - .map(|step| match step { - QueryPlanStep::StoreRead { - metric, - aggregation_id, - strategy, - } => format!("StoreRead({strategy:?}, {metric}#{aggregation_id})"), - QueryPlanStep::ComposeWindows { output_count } => { - format!("ComposeWindows({output_count} outputs)") - } - QueryPlanStep::ResolveKeys => "ResolveKeys".into(), - QueryPlanStep::Estimate { statistic } => format!("Estimate({statistic})"), - QueryPlanStep::LimitTopK => "LimitTopK".into(), - QueryPlanStep::Format => "Format".into(), - }) - .collect::>() - .join(" -> ") - } -} - -#[cfg(test)] -mod tests { - use super::*; - use crate::data_model::AggregationIdInfo; - use crate::engines::simple_engine::{QueryExecutionContext, QueryMetadata, StoreQueryPlan}; - use promql_utilities::data_model::KeyByLabelNames; - use promql_utilities::query_logics::enums::AggregationType; - use std::collections::HashMap; - - fn context(window_type: WindowType) -> RangeQueryExecutionContext { - let metadata = QueryMetadata { - query_output_labels: KeyByLabelNames::empty(), - statistic_to_compute: Statistic::Sum, - query_kwargs: HashMap::new(), - keep_metric_name: false, - }; - let query = StoreQueryParams { - metric: "requests".into(), - aggregation_id: 7, - start_timestamp: 0, - end_timestamp: 1_000, - }; - let base = QueryExecutionContext { - metric: "requests".into(), - metadata, - store_plan: StoreQueryPlan { - values_query: query, - keys_query: None, - }, - agg_info: AggregationIdInfo { - aggregation_id_for_key: 7, - aggregation_id_for_value: 7, - aggregation_type_for_key: AggregationType::Sum, - aggregation_type_for_value: AggregationType::Sum, + Self::push( + nodes, + QueryPlanNode::StoreRead { + query: query.clone(), + strategy, }, - value_window_type: window_type, - do_merge: false, - spatial_filter: String::new(), - query_time: 1_000, - grouping_labels: KeyByLabelNames::empty(), - aggregated_labels: KeyByLabelNames::empty(), - }; - RangeQueryExecutionContext { - base, - output_timestamps: vec![1_000], - query_range_ms: 1_000, - buckets_per_step: 1, - lookback_bucket_count: 1, - tumbling_window_ms: 1_000, - window_type, - window_size_ms: 1_000, - keys_window_type: None, - keys_window_size_ms: None, - keys_lookback_ms: None, - keys_tumbling_window_ms: None, - } + ) } - #[test] - fn compiles_a_leaf_context_into_an_explainable_plan() { - let plan = QueryPlan::compile_range( - &context(WindowType::Tumbling), - PlanOptions { - limit_topk: false, - format_output: false, + fn push_compose( + nodes: &mut Vec, + input: NodeId, + output_timestamps: &[u64], + lookback_ms: u64, + window_size_ms: u64, + bucket_step_ms: u64, + ) -> NodeId { + Self::push( + nodes, + QueryPlanNode::ComposeWindows { + input, + output_timestamps: output_timestamps.to_vec(), + lookback_ms, + window_size_ms, + bucket_step_ms, }, - ); - assert_eq!(plan.explain(), "StoreRead(WindowGrid, requests#7) -> ComposeWindows(1 outputs) -> ResolveKeys -> Estimate(sum)"); + ) } - #[test] - fn compiles_a_separate_keys_branch_before_key_resolution() { - let mut context = context(WindowType::Tumbling); - context.base.store_plan.keys_query = Some(StoreQueryParams { - metric: "requests".into(), - aggregation_id: 8, - start_timestamp: 0, - end_timestamp: 1_000, - }); - context.keys_window_type = Some(WindowType::Sliding); - - let plan = QueryPlan::compile_range( - &context, - PlanOptions { - limit_topk: false, - format_output: false, - }, - ); - - assert_eq!(plan.explain(), "StoreRead(WindowGrid, requests#7) -> ComposeWindows(1 outputs) -> StoreRead(SlidingExactCover, requests#8) -> ComposeWindows(1 outputs) -> ResolveKeys -> Estimate(sum)"); - } - - #[test] - fn compiles_sliding_windows_as_an_exact_cover_read() { - let plan = QueryPlan::compile_range( - &context(WindowType::Sliding), - PlanOptions { - limit_topk: false, - format_output: false, - }, - ); - - assert!(plan - .explain() - .starts_with("StoreRead(SlidingExactCover, requests#7)")); - } - - #[test] - fn adds_requested_topk_limit_and_formatting() { - let mut context = context(WindowType::Tumbling); - context.base.metadata.statistic_to_compute = Statistic::Topk; - - let plan = QueryPlan::compile_range( - &context, - PlanOptions { - limit_topk: true, - format_output: true, - }, - ); - - assert!(plan - .explain() - .ends_with("Estimate(topk) -> LimitTopK -> Format")); + pub(crate) fn explain(&self) -> String { + let mut lines = Vec::with_capacity(self.nodes.len() + 1); + for (index, node) in self.nodes.iter().enumerate() { + let line = match node { + QueryPlanNode::StoreRead { query, strategy } => format!( + "n{index} StoreRead({strategy:?}, {}#{}, [{}, {}])", + query.metric, query.aggregation_id, query.start_timestamp, query.end_timestamp + ), + QueryPlanNode::ComposeWindows { input, output_timestamps, lookback_ms, window_size_ms, bucket_step_ms } => format!( + "n{index} ComposeWindows(n{}, outputs={:?}, lookback={lookback_ms}ms, window={window_size_ms}ms, step={bucket_step_ms}ms)", + input.0, output_timestamps + ), + QueryPlanNode::ResolveKeys { values, keys } => format!( + "n{index} ResolveKeys(values=n{}, keys={})", + values.0, + keys.map(|id| format!("n{}", id.0)).unwrap_or_else(|| "self".to_string()) + ), + QueryPlanNode::Estimate { input, statistic, query_kwargs } => format!( + "n{index} Estimate(n{}, {statistic}, {query_kwargs:?})", input.0 + ), + QueryPlanNode::LimitTopK { input, k } => format!("n{index} LimitTopK(n{}, k={k})", input.0), + QueryPlanNode::Format { input, include_metric_name } => format!( + "n{index} Format(n{}, include_metric_name={include_metric_name})", input.0 + ), + }; + lines.push(line); + } + lines.push(format!("root: n{}", self.root.0)); + lines.join("\n") } } diff --git a/asap-query-engine/src/engines/query_plan/README.md b/asap-query-engine/src/engines/query_plan/README.md new file mode 100644 index 00000000..8e1fc783 --- /dev/null +++ b/asap-query-engine/src/engines/query_plan/README.md @@ -0,0 +1,16 @@ +# Native query plan + +`query_plan` compiles a resolved native query into a request-specific DAG. It is +debug-only for now: the existing executor remains the source of execution. + +Nodes are connected by `n` inputs: + +- `StoreRead` fetches one aggregation over its requested timestamp bounds. +- `ComposeWindows` turns stored buckets into one aggregate per output timestamp. +- `ResolveKeys` combines a value branch with an optional separate keys branch. +- `Estimate` queries the accumulator statistic with its parameters. +- `LimitTopK` and `Format` are presentation nodes. + +`StoreRead` is either a tumbling grid scan or a sliding exact-cover scan. A +range plan has one read branch shared across all output timestamps. Binary +expression DAGs are intentionally deferred to the binary-execution PR. diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index 234659dc..bb2e0fd0 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1188,16 +1188,7 @@ impl SimpleEngine { ) })?; - let plan = QueryPlan::compile_range( - &range_context, - PlanOptions { - limit_topk: enable_topk_limiting, - format_output: enable_topk_formatting, - }, - ); - debug!(plan = %plan.explain(), "Compiled native query plan"); - - let range_results = self.execute_range_query_pipeline( + let range_results = self.execute_observed_range_query_pipeline( &range_context, enable_topk_limiting, enable_topk_formatting, @@ -1911,6 +1902,23 @@ impl SimpleEngine { /// (#581 stage E.3), before insertion into the final result map. /// Formatting (metric-name label prefix) is a separate, smaller pass /// afterward, once per group rather than once per timestep. + fn execute_observed_range_query_pipeline( + &self, + context: &RangeQueryExecutionContext, + enable_topk_limiting: bool, + enable_topk_formatting: bool, + ) -> Result, String> { + let plan = QueryPlan::compile_range( + context, + PlanOptions { + limit_topk: enable_topk_limiting, + format_output: enable_topk_formatting, + }, + ); + debug!(plan = %plan.explain(), "Compiled native query plan"); + self.execute_range_query_pipeline(context, enable_topk_limiting, enable_topk_formatting) + } + fn execute_range_query_pipeline( &self, context: &RangeQueryExecutionContext, diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs index d114d627..bd1d88a1 100644 --- a/asap-query-engine/src/engines/simple_engine/promql.rs +++ b/asap-query-engine/src/engines/simple_engine/promql.rs @@ -742,7 +742,9 @@ impl SimpleEngine { // Binary arms need Topk limiting, but must remain in the // unformatted intermediate label representation until after the // arithmetic operation. - let results = self.execute_range_query_pipeline(&ctx, true, false).ok()?; + let results = self + .execute_observed_range_query_pipeline(&ctx, true, false) + .ok()?; let combined: Vec = results .into_iter() .map(|mut elem| { @@ -772,10 +774,10 @@ impl SimpleEngine { } // Binary arms need Topk limiting, but not final presentation formatting. let lhs_results = self - .execute_range_query_pipeline(&lhs_ctx, true, false) + .execute_observed_range_query_pipeline(&lhs_ctx, true, false) .ok()?; let rhs_results = self - .execute_range_query_pipeline(&rhs_ctx, true, false) + .execute_observed_range_query_pipeline(&rhs_ctx, true, false) .ok()?; // Build lookup: label_key -> {timestamp -> value} for rhs @@ -1342,7 +1344,7 @@ impl SimpleEngine { // instant's handle_query_promql -- both flags are no-ops unless this // query's statistic is Topk. let results: Vec = self - .execute_range_query_pipeline(&context, true, true) + .execute_observed_range_query_pipeline(&context, true, true) .map_err(|e| { warn!("Range query execution failed: {}", e); e From d051994fe4a1de8dff5022eeed84a00fb3c74ce0 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 18 Sep 2026 16:36:42 -0400 Subject: [PATCH 3/7] test(query-engine): cover query DAG key fan-in --- asap-query-engine/src/engines/query_plan.rs | 81 +++++++++++++++++++++ 1 file changed, 81 insertions(+) diff --git a/asap-query-engine/src/engines/query_plan.rs b/asap-query-engine/src/engines/query_plan.rs index e6d1b07e..4fb28714 100644 --- a/asap-query-engine/src/engines/query_plan.rs +++ b/asap-query-engine/src/engines/query_plan.rs @@ -201,3 +201,84 @@ impl QueryPlan { lines.join("\n") } } + +#[cfg(test)] +mod tests { + use super::*; + use crate::data_model::AggregationIdInfo; + use crate::engines::simple_engine::{QueryExecutionContext, QueryMetadata, StoreQueryPlan}; + use promql_utilities::data_model::KeyByLabelNames; + use promql_utilities::query_logics::enums::AggregationType; + use std::collections::HashMap; + + fn context() -> RangeQueryExecutionContext { + RangeQueryExecutionContext { + base: QueryExecutionContext { + metric: "requests".into(), + metadata: QueryMetadata { + query_output_labels: KeyByLabelNames::empty(), + statistic_to_compute: Statistic::Sum, + query_kwargs: HashMap::new(), + keep_metric_name: false, + }, + store_plan: StoreQueryPlan { + values_query: StoreQueryParams { + metric: "requests".into(), + aggregation_id: 7, + start_timestamp: 0, + end_timestamp: 1_000, + }, + keys_query: None, + }, + agg_info: AggregationIdInfo { + aggregation_id_for_key: 7, + aggregation_id_for_value: 7, + aggregation_type_for_key: AggregationType::Sum, + aggregation_type_for_value: AggregationType::Sum, + }, + value_window_type: WindowType::Tumbling, + do_merge: false, + spatial_filter: String::new(), + query_time: 1_000, + grouping_labels: KeyByLabelNames::empty(), + aggregated_labels: KeyByLabelNames::empty(), + }, + output_timestamps: vec![1_000], + query_range_ms: 1_000, + buckets_per_step: 1, + lookback_bucket_count: 1, + tumbling_window_ms: 1_000, + window_type: WindowType::Tumbling, + window_size_ms: 1_000, + keys_window_type: None, + keys_window_size_ms: None, + keys_lookback_ms: None, + keys_tumbling_window_ms: None, + } + } + + #[test] + fn separate_key_branch_fans_into_key_resolution() { + let mut context = context(); + context.base.store_plan.keys_query = Some(StoreQueryParams { + metric: "requests".into(), + aggregation_id: 8, + start_timestamp: 0, + end_timestamp: 1_000, + }); + context.keys_window_type = Some(WindowType::Sliding); + + let explanation = QueryPlan::compile_range( + &context, + PlanOptions { + limit_topk: false, + format_output: false, + }, + ) + .explain(); + + assert!(explanation.contains("n4 ResolveKeys(values=n1, keys=n3)")); + assert!(explanation.contains("n2 StoreRead(SlidingExactCover, requests#8")); + assert!(explanation.ends_with("root: n5")); + } +} From 8dce5f5499353cd2f7101db0c2f2fe16cc4c4ce4 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 18 Sep 2026 17:27:07 -0400 Subject: [PATCH 4/7] test(query-engine): cover query DAG roots --- asap-query-engine/src/engines/query_plan.rs | 42 +++++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/asap-query-engine/src/engines/query_plan.rs b/asap-query-engine/src/engines/query_plan.rs index 4fb28714..6f6ac9fe 100644 --- a/asap-query-engine/src/engines/query_plan.rs +++ b/asap-query-engine/src/engines/query_plan.rs @@ -281,4 +281,46 @@ mod tests { assert!(explanation.contains("n2 StoreRead(SlidingExactCover, requests#8")); assert!(explanation.ends_with("root: n5")); } + + #[test] + fn range_plan_keeps_every_output_timestamp() { + let mut context = context(); + context.output_timestamps = vec![1_000, 2_000, 3_000]; + + let explanation = QueryPlan::compile_range( + &context, + PlanOptions { + limit_topk: false, + format_output: false, + }, + ) + .explain(); + + assert!(explanation.contains("outputs=[1000, 2000, 3000]")); + } + + #[test] + fn topk_formatting_is_the_plan_root() { + let mut context = context(); + context.base.metadata.statistic_to_compute = Statistic::Topk; + context + .base + .metadata + .query_kwargs + .insert("k".to_string(), "3".to_string()); + context.base.metadata.keep_metric_name = true; + + let explanation = QueryPlan::compile_range( + &context, + PlanOptions { + limit_topk: true, + format_output: true, + }, + ) + .explain(); + + assert!(explanation.contains("LimitTopK(n3, k=3)")); + assert!(explanation.contains("Format(n4, include_metric_name=true)")); + assert!(explanation.ends_with("root: n5")); + } } From 7798628e1c4d00e066af95bef939b5ab1d76bad2 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 18 Sep 2026 17:45:08 -0400 Subject: [PATCH 5/7] fix(query-engine): validate native query DAGs --- asap-query-engine/src/engines/query_plan.rs | 36 +++++++++++++++---- .../src/engines/simple_engine/mod.rs | 2 +- 2 files changed, 31 insertions(+), 7 deletions(-) diff --git a/asap-query-engine/src/engines/query_plan.rs b/asap-query-engine/src/engines/query_plan.rs index 6f6ac9fe..4499d6ad 100644 --- a/asap-query-engine/src/engines/query_plan.rs +++ b/asap-query-engine/src/engines/query_plan.rs @@ -61,7 +61,7 @@ impl QueryPlan { pub(crate) fn compile_range( context: &RangeQueryExecutionContext, options: PlanOptions, - ) -> Self { + ) -> Result { let mut nodes = Vec::new(); let values_read = Self::push_read( &mut nodes, @@ -111,7 +111,9 @@ impl QueryPlan { .query_kwargs .get("k") .cloned() - .unwrap_or_else(|| "".to_string()); + .ok_or_else(|| "Topk query is missing required `k` parameter".to_string())?; + k.parse::() + .map_err(|_| "Topk query has an invalid `k` parameter".to_string())?; root = Self::push(&mut nodes, QueryPlanNode::LimitTopK { input: root, k }); } if options.format_output { @@ -123,7 +125,7 @@ impl QueryPlan { }, ); } - Self { nodes, root } + Ok(Self { nodes, root }) } fn push(nodes: &mut Vec, node: QueryPlanNode) -> NodeId { @@ -187,9 +189,11 @@ impl QueryPlan { values.0, keys.map(|id| format!("n{}", id.0)).unwrap_or_else(|| "self".to_string()) ), - QueryPlanNode::Estimate { input, statistic, query_kwargs } => format!( - "n{index} Estimate(n{}, {statistic}, {query_kwargs:?})", input.0 - ), + QueryPlanNode::Estimate { input, statistic, query_kwargs } => { + let mut kwargs: Vec<_> = query_kwargs.iter().collect(); + kwargs.sort_unstable_by_key(|(key, _)| *key); + format!("n{index} Estimate(n{}, {statistic}, {kwargs:?})", input.0) + }, QueryPlanNode::LimitTopK { input, k } => format!("n{index} LimitTopK(n{}, k={k})", input.0), QueryPlanNode::Format { input, include_metric_name } => format!( "n{index} Format(n{}, include_metric_name={include_metric_name})", input.0 @@ -275,6 +279,7 @@ mod tests { format_output: false, }, ) + .unwrap() .explain(); assert!(explanation.contains("n4 ResolveKeys(values=n1, keys=n3)")); @@ -294,6 +299,7 @@ mod tests { format_output: false, }, ) + .unwrap() .explain(); assert!(explanation.contains("outputs=[1000, 2000, 3000]")); @@ -317,10 +323,28 @@ mod tests { format_output: true, }, ) + .unwrap() .explain(); assert!(explanation.contains("LimitTopK(n3, k=3)")); assert!(explanation.contains("Format(n4, include_metric_name=true)")); assert!(explanation.ends_with("root: n5")); } + + #[test] + fn rejects_topk_without_a_limit() { + let mut context = context(); + context.base.metadata.statistic_to_compute = Statistic::Topk; + + let error = QueryPlan::compile_range( + &context, + PlanOptions { + limit_topk: true, + format_output: false, + }, + ) + .expect_err("topk plan without k must fail loudly"); + + assert_eq!(error, "Topk query is missing required `k` parameter"); + } } diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs index bb2e0fd0..0daaa6d7 100644 --- a/asap-query-engine/src/engines/simple_engine/mod.rs +++ b/asap-query-engine/src/engines/simple_engine/mod.rs @@ -1914,7 +1914,7 @@ impl SimpleEngine { limit_topk: enable_topk_limiting, format_output: enable_topk_formatting, }, - ); + )?; debug!(plan = %plan.explain(), "Compiled native query plan"); self.execute_range_query_pipeline(context, enable_topk_limiting, enable_topk_formatting) } From e614de9c8cf6ebdbaa809a1326a19700c43969da Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Fri, 18 Sep 2026 17:54:49 -0400 Subject: [PATCH 6/7] docs(query-engine): clarify self-keyed query DAGs --- asap-query-engine/src/engines/query_plan/README.md | 1 + 1 file changed, 1 insertion(+) diff --git a/asap-query-engine/src/engines/query_plan/README.md b/asap-query-engine/src/engines/query_plan/README.md index 8e1fc783..be15b2c6 100644 --- a/asap-query-engine/src/engines/query_plan/README.md +++ b/asap-query-engine/src/engines/query_plan/README.md @@ -8,6 +8,7 @@ Nodes are connected by `n` inputs: - `StoreRead` fetches one aggregation over its requested timestamp bounds. - `ComposeWindows` turns stored buckets into one aggregate per output timestamp. - `ResolveKeys` combines a value branch with an optional separate keys branch. + Without a keys branch, the value accumulator supplies its own keys. - `Estimate` queries the accumulator statistic with its parameters. - `LimitTopK` and `Format` are presentation nodes. From 18e77548b1efc17da1103fb98a5b49de00a7cbaf Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Tue, 22 Sep 2026 19:48:24 -0400 Subject: [PATCH 7/7] docs(query-engine): describe query DAG node inputs --- asap-query-engine/src/engines/query_plan/README.md | 13 +++++++------ 1 file changed, 7 insertions(+), 6 deletions(-) diff --git a/asap-query-engine/src/engines/query_plan/README.md b/asap-query-engine/src/engines/query_plan/README.md index be15b2c6..ec0dfe2a 100644 --- a/asap-query-engine/src/engines/query_plan/README.md +++ b/asap-query-engine/src/engines/query_plan/README.md @@ -5,12 +5,13 @@ debug-only for now: the existing executor remains the source of execution. Nodes are connected by `n` inputs: -- `StoreRead` fetches one aggregation over its requested timestamp bounds. -- `ComposeWindows` turns stored buckets into one aggregate per output timestamp. -- `ResolveKeys` combines a value branch with an optional separate keys branch. - Without a keys branch, the value accumulator supplies its own keys. -- `Estimate` queries the accumulator statistic with its parameters. -- `LimitTopK` and `Format` are presentation nodes. +- `StoreRead` — input: none; fetches one aggregation over its timestamp bounds. +- `ComposeWindows` — input: one `StoreRead`; composes buckets for each output timestamp. +- `ResolveKeys` — inputs: a value composition and optional keys composition; resolves output keys. + Without a keys input, the value accumulator supplies its own keys. +- `Estimate` — input: `ResolveKeys`; computes the requested statistic with its parameters. +- `LimitTopK` — input: `Estimate`; keeps the highest-ranked `k` candidates. +- `Format` — input: `Estimate` or `LimitTopK`; applies protocol-specific output labels. `StoreRead` is either a tumbling grid scan or a sliding exact-cover scan. A range plan has one read branch shared across all output timestamps. Binary