diff --git a/asap-common/dependencies/rs/asap_types/src/query_requirements.rs b/asap-common/dependencies/rs/asap_types/src/query_requirements.rs
index 3dc9e799..caca7911 100644
--- a/asap-common/dependencies/rs/asap_types/src/query_requirements.rs
+++ b/asap-common/dependencies/rs/asap_types/src/query_requirements.rs
@@ -1,6 +1,6 @@
 use promql_utilities::ast_matching::PromQLMatchResult;
 use promql_utilities::data_model::KeyByLabelNames;
-use promql_utilities::query_logics::enums::{QueryPatternType, Statistic};
+use promql_utilities::query_logics::enums::Statistic;
 use promql_utilities::query_logics::parsing::{
     get_metric_and_spatial_filter, get_spatial_aggregation_output_labels, get_statistics_to_compute,
 };
@@ -46,13 +46,12 @@ pub struct QueryRequirements {
 pub fn build_query_requirements_promql(
     query: &str,
     match_result: &PromQLMatchResult,
-    pattern_type: QueryPatternType,
     metric_schema: &PromQLSchema,
     data_ingestion_interval_ms: u64,
 ) -> Option<QueryRequirements> {
     let (metric, spatial_filter) = get_metric_and_spatial_filter(match_result);
 
-    let statistics = get_statistics_to_compute(pattern_type, match_result)
+    let statistics = get_statistics_to_compute(match_result)
         .map_err(|err| {
             warn!(
                 query = %query,
@@ -63,13 +62,19 @@ pub fn build_query_requirements_promql(
         })
         .ok()?;
 
-    let data_range_ms = match pattern_type {
-        QueryPatternType::OnlySpatial => data_ingestion_interval_ms,
+    let has_temporal_function = match_result.tokens.contains_key("function");
+    let has_aggregation = match_result.tokens.contains_key("aggregation");
+
+    let data_range_ms = if has_temporal_function {
         // promql-parser supports a literal `ms` duration suffix (e.g. `[500ms]`),
         // so .num_seconds() would truncate sub-second ranges to 0.
-        _ => match_result
+        match_result
             .get_range_duration()
-            .map(|d| d.num_milliseconds() as u64)?,
+            .map(|d| d.num_milliseconds() as u64)?
+    } else {
+        // OnlySpatial (no temporal component): the query has no range of its
+        // own, so its data range is exactly one scrape interval.
+        data_ingestion_interval_ms
     };
 
     let all_labels = metric_schema
@@ -77,14 +82,13 @@ pub fn build_query_requirements_promql(
         .cloned()
         .unwrap_or_else(KeyByLabelNames::empty);
 
-    let grouping_labels = match pattern_type {
+    let grouping_labels = if has_aggregation {
+        // OnlySpatial and (collapsable, see #508) OneTemporalOneSpatial encode
+        // their output labels in the AST's `by (...)` / `without (...)` clause.
+        get_spatial_aggregation_output_labels(match_result, &all_labels)
+    } else {
         // OnlyTemporal preserves all labels.
-        QueryPatternType::OnlyTemporal => all_labels,
-        // OnlySpatial and OneTemporalOneSpatial encode their output labels in
-        // the AST's `by (...)` / `without (...)` clause.
-        QueryPatternType::OnlySpatial | QueryPatternType::OneTemporalOneSpatial => {
-            get_spatial_aggregation_output_labels(match_result, &all_labels)
-        }
+        all_labels
     };
 
     Some(QueryRequirements {
diff --git a/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs b/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs
index 977dfd47..04110285 100644
--- a/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs
+++ b/asap-common/dependencies/rs/promql_utilities/src/query_logics/enums.rs
@@ -3,24 +3,6 @@ use std::fmt;
 use std::str::FromStr;
 use tracing::debug;
 
-#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
-pub enum QueryPatternType {
-    OnlyTemporal,
-    OnlySpatial,
-    OneTemporalOneSpatial,
-}
-
-impl std::fmt::Display for QueryPatternType {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        debug!("Formatting QueryPatternType: {:?}", self);
-        match self {
-            QueryPatternType::OnlyTemporal => write!(f, "only_temporal"),
-            QueryPatternType::OnlySpatial => write!(f, "only_spatial"),
-            QueryPatternType::OneTemporalOneSpatial => write!(f, "one_temporal_one_spatial"),
-        }
-    }
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum QueryTreatmentType {
     Exact,
@@ -69,6 +51,22 @@ impl std::fmt::Display for Statistic {
 
 #[allow(clippy::should_implement_trait)]
 impl Statistic {
+    /// Returns `true` for statistics whose result requires approximate
+    /// pre-aggregation, regardless of whether they were reached via a
+    /// temporal function (`PromQLFunction::is_approximate`) or a spatial
+    /// aggregation operator (`AggregationOperator::is_approximate`) — for
+    /// every `Statistic` reachable from either origin, the two origins agree.
+    pub fn is_approximate(self) -> bool {
+        matches!(
+            self,
+            Statistic::Count
+                | Statistic::Sum
+                | Statistic::Cardinality
+                | Statistic::Quantile
+                | Statistic::Topk
+        )
+    }
+
     pub fn from_str(s: &str) -> Option<Self> {
         debug!("Parsing Statistic from string: {}", s);
         match s.to_lowercase().as_str() {
diff --git a/asap-common/dependencies/rs/promql_utilities/src/query_logics/parsing.rs b/asap-common/dependencies/rs/promql_utilities/src/query_logics/parsing.rs
index ec0dfd70..be91485a 100644
--- a/asap-common/dependencies/rs/promql_utilities/src/query_logics/parsing.rs
+++ b/asap-common/dependencies/rs/promql_utilities/src/query_logics/parsing.rs
@@ -4,21 +4,22 @@ use tracing::debug;
 use crate::ast_matching::promql_pattern::AggregationModifierType;
 use crate::ast_matching::PromQLMatchResult;
 use crate::data_model::KeyByLabelNames;
-use crate::query_logics::enums::{AggregationOperator, QueryPatternType, Statistic};
+use crate::query_logics::enums::{AggregationOperator, PromQLFunction, Statistic};
+use crate::query_logics::logics::get_is_collapsable;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum StatisticExtractionError {
-    MissingStatistic { pattern_type: QueryPatternType },
+    MissingStatistic,
     UnsupportedStatistic { statistic: String },
 }
 
 impl std::fmt::Display for StatisticExtractionError {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         match self {
-            Self::MissingStatistic { pattern_type } => {
+            Self::MissingStatistic => {
                 write!(
                     f,
-                    "No statistic found for query pattern type {pattern_type:?}"
+                    "No temporal function or aggregation operation found in match result"
                 )
             }
             Self::UnsupportedStatistic { statistic } => {
@@ -72,28 +73,66 @@ pub fn get_metric_and_spatial_filter(match_result: &PromQLMatchResult) -> (Strin
     (metric_name, spatial_filter)
 }
 
-/// Get statistics to compute based on pattern type and tokens.
+/// Get statistics to compute from a matched query's tokens.
+///
+/// Explicitly handles the three reachable shapes:
+/// - Only a temporal function (`"function"` token, no `"aggregation"`): the
+///   statistic comes from the function name.
+/// - Only a spatial aggregation (`"aggregation"` token, no `"function"`): the
+///   statistic comes from the aggregation operator.
+/// - Both (a spatial aggregation wrapping a temporal function): only ever
+///   reachable via a pattern already narrowed to a collapsable `(function,
+///   op)` pair (see `get_is_collapsable`, and #508's pattern-narrowing fix
+///   that makes non-collapsable combinations fail to match at all) — asserted
+///   below rather than silently trusted. The statistic still comes from the
+///   *function*, never the outer op: e.g. `count_over_time` + `sum` needs a
+///   `Count` accumulator, not a `Sum` one — summing per-series counts gives
+///   the group's total count, so the outer op only describes how per-series
+///   results combine, never which statistic must be precomputed.
+///
 /// Returns a typed error if the matched statistic/function name is not
 /// recognized, so callers can decide whether to skip or fail the query.
 pub fn get_statistics_to_compute(
-    pattern_type: QueryPatternType,
     match_result: &PromQLMatchResult,
 ) -> Result<Vec<Statistic>, StatisticExtractionError> {
-    debug!("Computing statistics for pattern type {:?}", pattern_type);
-    let statistic_to_compute: Option<String> = match pattern_type {
-        QueryPatternType::OnlyTemporal | QueryPatternType::OneTemporalOneSpatial => {
-            match_result.get_function_name().map(|function_name| {
-                let name = function_name.to_lowercase();
-                name.split('_').next().unwrap_or(&name).to_string()
-            })
-        }
-        QueryPatternType::OnlySpatial => match_result
+    let has_function = match_result.tokens.contains_key("function");
+    let has_aggregation = match_result.tokens.contains_key("aggregation");
+    debug!("Computing statistics (has_function={has_function}, has_aggregation={has_aggregation})");
+
+    let function_statistic = |match_result: &PromQLMatchResult| {
+        match_result.get_function_name().map(|function_name| {
+            let name = function_name.to_lowercase();
+            name.split('_').next().unwrap_or(&name).to_string()
+        })
+    };
+
+    let statistic_to_compute: Option<String> = if has_function && has_aggregation {
+        debug_assert!(
+            match_result
+                .get_function_name()
+                .and_then(|f| f.parse::<PromQLFunction>().ok())
+                .zip(
+                    match_result
+                        .get_aggregation_op()
+                        .and_then(|o| o.parse::<AggregationOperator>().ok())
+                )
+                .is_some_and(|(f, o)| get_is_collapsable(f, o)),
+            "a match with both function and aggregation tokens must be collapsable \
+             (patterns are narrowed to only collapsable pairs, see #508)"
+        );
+        function_statistic(match_result)
+    } else if has_function {
+        function_statistic(match_result)
+    } else if has_aggregation {
+        match_result
             .get_aggregation_op()
-            .map(|agg| agg.to_lowercase()),
+            .map(|agg| agg.to_lowercase())
+    } else {
+        None
     };
 
     let Some(statistic_to_compute) = statistic_to_compute else {
-        return Err(StatisticExtractionError::MissingStatistic { pattern_type });
+        return Err(StatisticExtractionError::MissingStatistic);
     };
 
     debug!("Found statistic to compute: {}", statistic_to_compute);
@@ -194,9 +233,7 @@ mod tests {
 
     #[test]
     fn unsupported_matched_temporal_statistic_returns_typed_error() {
-        let err =
-            get_statistics_to_compute(QueryPatternType::OnlyTemporal, &temporal_match("stddev"))
-                .unwrap_err();
+        let err = get_statistics_to_compute(&temporal_match("stddev")).unwrap_err();
 
         assert_eq!(
             err,
@@ -208,17 +245,9 @@ mod tests {
 
     #[test]
     fn missing_statistic_returns_typed_error() {
-        let err = get_statistics_to_compute(
-            QueryPatternType::OnlyTemporal,
-            &PromQLMatchResult::with_tokens(HashMap::new()),
-        )
-        .unwrap_err();
+        let err =
+            get_statistics_to_compute(&PromQLMatchResult::with_tokens(HashMap::new())).unwrap_err();
 
-        assert_eq!(
-            err,
-            StatisticExtractionError::MissingStatistic {
-                pattern_type: QueryPatternType::OnlyTemporal
-            }
-        );
+        assert_eq!(err, StatisticExtractionError::MissingStatistic);
     }
 }
diff --git a/asap-planner-rs/src/optimizer/aqe_extractor.rs b/asap-planner-rs/src/optimizer/aqe_extractor.rs
index 77e45dd0..52869d1b 100644
--- a/asap-planner-rs/src/optimizer/aqe_extractor.rs
+++ b/asap-planner-rs/src/optimizer/aqe_extractor.rs
@@ -170,10 +170,10 @@ fn extract_requirements(
     let ast = promql_parser::parser::parse(query).ok()?;
     let patterns = build_patterns();
 
-    let (pattern_type, match_result) = patterns.iter().find_map(|(pt, pat)| {
+    let match_result = patterns.iter().find_map(|pat| {
         let r = pat.matches(&ast);
         if r.matches {
-            Some((*pt, r))
+            Some(r)
         } else {
             None
         }
@@ -182,7 +182,6 @@ fn extract_requirements(
     build_query_requirements_promql(
         query,
         &match_result,
-        pattern_type,
         metric_schema,
         data_ingestion_interval_ms,
     )
diff --git a/asap-planner-rs/src/optimizer/greedy.rs b/asap-planner-rs/src/optimizer/greedy.rs
index bb98dcef..4cf19b11 100644
--- a/asap-planner-rs/src/optimizer/greedy.rs
+++ b/asap-planner-rs/src/optimizer/greedy.rs
@@ -126,7 +126,7 @@ mod tests {
         );
 
         let mut seen_ids: StdHashMap<u64, ()> = StdHashMap::new();
-        for (id, _) in solution.deployed_configs.iter() {
+        for id in solution.deployed_configs.keys() {
             assert!(
                 seen_ids.insert(*id, ()).is_none(),
                 "duplicate aggregation_id"
diff --git a/asap-planner-rs/src/planner/cleanup.rs b/asap-planner-rs/src/planner/cleanup.rs
index 444b977c..4f07d75c 100644
--- a/asap-planner-rs/src/planner/cleanup.rs
+++ b/asap-planner-rs/src/planner/cleanup.rs
@@ -1,13 +1,17 @@
 use asap_types::enums::{CleanupPolicy, WindowType};
-use promql_utilities::ast_matching::PromQLMatchResult;
-use promql_utilities::query_logics::enums::QueryPatternType;
 
 use super::window::get_effective_repeat;
 
+/// `data_range_ms` is the query's own requested duration (from
+/// `QueryRequirements.data_range_ms` — for a spatial-only query this equals
+/// `data_ingestion_interval_ms` by construction), used directly as the
+/// lookback: once `set_window_parameters`'s invariant holds
+/// (`data_range_ms >= t_repeat_ms >= data_ingestion_interval_ms`), this is
+/// exactly equivalent to the old pattern-type-gated `t_repeat_ms`-vs-range-duration
+/// split, with no shape check needed (see #508).
 pub fn get_cleanup_param(
     cleanup_policy: CleanupPolicy,
-    query_pattern_type: QueryPatternType,
-    match_result: &PromQLMatchResult,
+    data_range_ms: u64,
     t_repeat_ms: u64,
     window_type: WindowType,
     range_duration_ms: u64,
@@ -22,19 +26,7 @@ pub fn get_cleanup_param(
     }
 
     let is_range_query = step_ms > 0;
-
-    let t_lookback: u64 = if query_pattern_type == QueryPatternType::OnlySpatial {
-        t_repeat_ms
-    } else {
-        match_result
-            .get_range_duration()
-            .map(|d| {
-                let ms = d.num_milliseconds();
-                debug_assert!(ms >= 0, "PromQL range duration should never be negative");
-                ms as u64
-            })
-            .ok_or_else(|| "No range_vector token found".to_string())?
-    };
+    let t_lookback: u64 = data_range_ms;
 
     if window_type == WindowType::Sliding {
         let result = if is_range_query {
@@ -98,34 +90,15 @@ pub fn get_sql_cleanup_param(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::planner::patterns::build_patterns;
-
-    use promql_utilities::ast_matching::PromQLMatchResult;
-    use promql_utilities::query_logics::enums::QueryPatternType;
-
-    fn match_query(query: &str) -> (QueryPatternType, PromQLMatchResult) {
-        let ast = promql_parser::parser::parse(query).unwrap();
-        let patterns = build_patterns();
-        for (pt, pattern) in &patterns {
-            let result = pattern.matches(&ast);
-            if result.matches {
-                return (*pt, result);
-            }
-        }
-        panic!("no pattern matched query: {}", query);
-    }
 
     #[test]
     fn cleanup_param_circular_buffer_spatial_instant_query() {
-        let (pt, mr) = match_query("sum(some_metric)");
-        assert_eq!(pt, QueryPatternType::OnlySpatial);
-        // t_lookback = t_repeat_ms = 300_000 (OnlySpatial path)
+        // data_range_ms = t_lookback = 300_000 (a spatial-only query's canonical range)
         // effective_repeat = 300_000 (step_ms=0)
         // ceil((300_000 + 0) / 300_000) = 1
         let result = get_cleanup_param(
             CleanupPolicy::CircularBuffer,
-            pt,
-            &mr,
+            300_000,
             300_000,
             WindowType::Tumbling,
             0,
@@ -137,13 +110,11 @@ mod tests {
 
     #[test]
     fn cleanup_param_circular_buffer_spatial_range_query() {
-        let (pt, mr) = match_query("sum(some_metric)");
-        // t_lookback = t_repeat_ms = 300_000, effective_repeat = min(300_000, 30_000) = 30_000
+        // t_lookback = data_range_ms = 300_000, effective_repeat = min(300_000, 30_000) = 30_000
         // ceil((300_000 + 3_600_000) / 30_000) = ceil(130) = 130
         let result = get_cleanup_param(
             CleanupPolicy::CircularBuffer,
-            pt,
-            &mr,
+            300_000,
             300_000,
             WindowType::Tumbling,
             3_600_000,
@@ -155,12 +126,10 @@ mod tests {
 
     #[test]
     fn cleanup_param_read_based_spatial_instant_query() {
-        let (pt, mr) = match_query("sum(some_metric)");
         // lookback_buckets = ceil(300_000/300_000) = 1, num_steps = 1 → result = 1
         let result = get_cleanup_param(
             CleanupPolicy::ReadBased,
-            pt,
-            &mr,
+            300_000,
             300_000,
             WindowType::Tumbling,
             0,
@@ -172,13 +141,11 @@ mod tests {
 
     #[test]
     fn cleanup_param_read_based_spatial_range_query() {
-        let (pt, mr) = match_query("sum(some_metric)");
         // lookback_buckets = ceil(300_000/30_000) = 10, num_steps = 3_600_000/30_000 + 1 = 121
         // result = 10 * 121 = 1210
         let result = get_cleanup_param(
             CleanupPolicy::ReadBased,
-            pt,
-            &mr,
+            300_000,
             300_000,
             WindowType::Tumbling,
             3_600_000,
@@ -190,14 +157,11 @@ mod tests {
 
     #[test]
     fn cleanup_param_circular_buffer_temporal_instant_query() {
-        let (pt, mr) = match_query("rate(some_metric[5m])");
-        assert_eq!(pt, QueryPatternType::OnlyTemporal);
-        // t_lookback = 5m = 300_000ms (from [5m] range vector), range_duration_ms=0, step_ms=0
+        // data_range_ms = 5m = 300_000ms (from a [5m] range vector), range_duration_ms=0, step_ms=0
         // effective_repeat = 60_000, ceil((300_000 + 0) / 60_000) = 5
         let result = get_cleanup_param(
             CleanupPolicy::CircularBuffer,
-            pt,
-            &mr,
+            300_000,
             60_000,
             WindowType::Tumbling,
             0,
@@ -209,11 +173,9 @@ mod tests {
 
     #[test]
     fn cleanup_param_no_cleanup_returns_error() {
-        let (pt, mr) = match_query("sum(some_metric)");
         let result = get_cleanup_param(
             CleanupPolicy::NoCleanup,
-            pt,
-            &mr,
+            300_000,
             300_000,
             WindowType::Tumbling,
             0,
@@ -224,12 +186,10 @@ mod tests {
 
     #[test]
     fn cleanup_param_mismatched_range_and_step_returns_error() {
-        let (pt, mr) = match_query("sum(some_metric)");
         // range_duration_ms > 0 but step_ms == 0 is invalid
         let result = get_cleanup_param(
             CleanupPolicy::CircularBuffer,
-            pt,
-            &mr,
+            300_000,
             300_000,
             WindowType::Tumbling,
             3_600_000,
diff --git a/asap-planner-rs/src/planner/patterns.rs b/asap-planner-rs/src/planner/patterns.rs
index d2de5afd..5b263f18 100644
--- a/asap-planner-rs/src/planner/patterns.rs
+++ b/asap-planner-rs/src/planner/patterns.rs
@@ -1,10 +1,11 @@
 use promql_utilities::ast_matching::{PromQLPattern, PromQLPatternBuilder};
-use promql_utilities::query_logics::enums::{
-    AggregationOperator, PromQLFunction, QueryPatternType,
-};
+use promql_utilities::query_logics::enums::{AggregationOperator, PromQLFunction};
+use promql_utilities::query_logics::logics::get_is_collapsable;
 
-/// Build all 5 patterns in priority order: ONLY_TEMPORAL (2), ONLY_SPATIAL (1), ONE_TEMPORAL_ONE_SPATIAL (2)
-pub fn build_patterns() -> Vec<(QueryPatternType, PromQLPattern)> {
+/// Build all patterns in priority order: ONLY_TEMPORAL (2), ONLY_SPATIAL (1),
+/// ONE_TEMPORAL_ONE_SPATIAL (one per collapsable (function, op) pair). Tried
+/// in order until one matches.
+pub fn build_patterns() -> Vec<PromQLPattern> {
     let metric_pattern = || PromQLPatternBuilder::metric(None, None, None, Some("metric"));
     let range_vector_pattern =
         || PromQLPatternBuilder::matrix_selector(metric_pattern(), None, Some("range_vector"));
@@ -35,18 +36,6 @@ pub fn build_patterns() -> Vec<(QueryPatternType, PromQLPattern)> {
     .map(AggregationOperator::as_str)
     .to_vec();
 
-    // Spatial-of-temporal excludes topk (no topk(quantile_over_time(...)) pattern)
-    let spatial_ops_no_topk: Vec<&str> = [
-        AggregationOperator::Sum,
-        AggregationOperator::Count,
-        AggregationOperator::Avg,
-        AggregationOperator::Quantile,
-        AggregationOperator::Min,
-        AggregationOperator::Max,
-    ]
-    .map(AggregationOperator::as_str)
-    .to_vec();
-
     // ONLY_TEMPORAL pattern 1: quantile_over_time(phi, metric[range])
     let ot_quantile = PromQLPattern::new(PromQLPatternBuilder::function(
         vec![PromQLFunction::QuantileOverTime.as_str()],
@@ -76,44 +65,103 @@ pub fn build_patterns() -> Vec<(QueryPatternType, PromQLPattern)> {
         Some("aggregation"),
     ));
 
-    // ONE_TEMPORAL_ONE_SPATIAL pattern 1: agg_op(quantile_over_time(phi, metric[range]))
-    let ottos_quantile = PromQLPattern::new(PromQLPatternBuilder::aggregation(
-        spatial_ops_no_topk.clone(),
-        PromQLPatternBuilder::function(
-            vec![PromQLFunction::QuantileOverTime.as_str()],
-            vec![
-                PromQLPatternBuilder::number(None, None),
-                range_vector_pattern(),
-            ],
-            Some("function"),
-            Some("function_args"),
-        ),
-        None,
-        None,
-        None,
-        Some("aggregation"),
-    ));
+    // ONE_TEMPORAL_ONE_SPATIAL: one narrow pattern per collapsable (function, op)
+    // pair (see `get_is_collapsable`) — e.g. `sum(min_over_time(x[5m]))` cannot be
+    // served by a single precomputed statistic the way `sum(sum_over_time(x[5m]))`
+    // can. A broad any-op-wrapping-any-function pattern would structurally match
+    // non-collapsable combinations too, silently dropping the outer aggregation
+    // instead of rejecting the query (see #508).
+    let one_temporal_one_spatial_collapsable: Vec<PromQLPattern> = [
+        PromQLFunction::Rate,
+        PromQLFunction::Increase,
+        PromQLFunction::SumOverTime,
+        PromQLFunction::CountOverTime,
+        PromQLFunction::AvgOverTime,
+        PromQLFunction::MinOverTime,
+        PromQLFunction::MaxOverTime,
+        PromQLFunction::QuantileOverTime,
+    ]
+    .into_iter()
+    .flat_map(|func| {
+        [
+            AggregationOperator::Sum,
+            AggregationOperator::Count,
+            AggregationOperator::Avg,
+            AggregationOperator::Quantile,
+            AggregationOperator::Min,
+            AggregationOperator::Max,
+        ]
+        .into_iter()
+        .filter_map(move |op| {
+            if !get_is_collapsable(func, op) {
+                return None;
+            }
+            let pattern = PromQLPatternBuilder::aggregation(
+                vec![op.as_str()],
+                PromQLPatternBuilder::function(
+                    vec![func.as_str()],
+                    vec![range_vector_pattern()],
+                    Some("function"),
+                    Some("function_args"),
+                ),
+                None,
+                None,
+                None,
+                Some("aggregation"),
+            );
+            Some(PromQLPattern::new(pattern))
+        })
+    })
+    .collect();
 
-    // ONE_TEMPORAL_ONE_SPATIAL pattern 2: agg_op(temporal_func(metric[range]))
-    let ottos_temporal = PromQLPattern::new(PromQLPatternBuilder::aggregation(
-        spatial_ops_no_topk,
-        PromQLPatternBuilder::function(
-            temporal_funcs,
-            vec![range_vector_pattern()],
-            Some("function"),
-            Some("function_args"),
-        ),
-        None,
-        None,
-        None,
-        Some("aggregation"),
-    ));
+    let mut patterns = vec![ot_quantile, ot_temporal_funcs, os_spatial];
+    patterns.extend(one_temporal_one_spatial_collapsable);
+    patterns
+}
 
-    vec![
-        (QueryPatternType::OnlyTemporal, ot_quantile),
-        (QueryPatternType::OnlyTemporal, ot_temporal_funcs),
-        (QueryPatternType::OnlySpatial, os_spatial),
-        (QueryPatternType::OneTemporalOneSpatial, ottos_quantile),
-        (QueryPatternType::OneTemporalOneSpatial, ottos_temporal),
-    ]
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    fn matches_some_pattern(query: &str) -> bool {
+        let ast = promql_parser::parser::parse(query).expect("query should parse");
+        build_patterns()
+            .iter()
+            .any(|pattern| pattern.matches(&ast).matches)
+    }
+
+    #[test]
+    fn exactly_four_collapsable_one_temporal_one_spatial_patterns() {
+        // 2 ONLY_TEMPORAL + 1 ONLY_SPATIAL + 4 collapsable ONE_TEMPORAL_ONE_SPATIAL
+        // (sum+sum_over_time, sum+count_over_time, min+min_over_time, max+max_over_time).
+        assert_eq!(build_patterns().len(), 7);
+    }
+
+    #[test]
+    fn collapsable_combinations_match() {
+        assert!(matches_some_pattern("sum(sum_over_time(x[5m]))"));
+        assert!(matches_some_pattern("sum(count_over_time(x[5m]))"));
+        assert!(matches_some_pattern("min(min_over_time(x[5m]))"));
+        assert!(matches_some_pattern("max(max_over_time(x[5m]))"));
+    }
+
+    #[test]
+    fn non_collapsable_combinations_are_rejected() {
+        assert!(
+            !matches_some_pattern("sum(min_over_time(x[5m]))"),
+            "sum+min_over_time is not collapsable"
+        );
+        assert!(
+            !matches_some_pattern("avg(rate(x[5m]))"),
+            "avg+rate is not collapsable"
+        );
+        assert!(
+            !matches_some_pattern("sum(rate(x[5m]))"),
+            "sum+rate is not collapsable"
+        );
+        assert!(
+            !matches_some_pattern("min(max_over_time(x[5m]))"),
+            "min+max_over_time is not collapsable"
+        );
+    }
 }
diff --git a/asap-planner-rs/src/planner/promql.rs b/asap-planner-rs/src/planner/promql.rs
index 5127968a..d47b9a96 100644
--- a/asap-planner-rs/src/planner/promql.rs
+++ b/asap-planner-rs/src/planner/promql.rs
@@ -1,14 +1,11 @@
 use asap_types::enums::CleanupPolicy;
+use asap_types::query_requirements::build_query_requirements_promql;
 use asap_types::PromQLSchema;
 use promql_utilities::ast_matching::PromQLMatchResult;
-use promql_utilities::data_model::KeyByLabelNames;
 use promql_utilities::query_logics::enums::{
-    AggregationOperator, AggregationType, PromQLFunction, QueryPatternType, QueryTreatmentType,
-};
-use promql_utilities::query_logics::logics::get_is_collapsable;
-use promql_utilities::query_logics::parsing::{
-    get_metric_and_spatial_filter, get_spatial_aggregation_output_labels, get_statistics_to_compute,
+    AggregationType, PromQLFunction, QueryTreatmentType, Statistic,
 };
+use promql_utilities::query_logics::parsing::get_metric_and_spatial_filter;
 
 use crate::config::input::SketchParameterOverrides;
 use crate::error::ControllerError;
@@ -108,41 +105,32 @@ impl SingleQueryProcessor {
         }
     }
 
-    /// Try to match query and return (pattern_type, match_result) or None
-    fn match_pattern(
-        &self,
-        ast: &promql_parser::parser::Expr,
-    ) -> Option<(QueryPatternType, PromQLMatchResult)> {
+    /// Try to match query and return the match result, or None
+    fn match_pattern(&self, ast: &promql_parser::parser::Expr) -> Option<PromQLMatchResult> {
         let patterns = build_patterns();
-        for (pattern_type, pattern) in &patterns {
+        for pattern in &patterns {
             let result = pattern.matches(ast);
             if result.matches {
-                return Some((*pattern_type, result));
+                return Some(result);
             }
         }
         None
     }
 
-    /// Get treatment type (Exact vs Approximate) from pattern match
-    fn get_treatment_type(
-        pattern_type: QueryPatternType,
-        match_result: &PromQLMatchResult,
-    ) -> QueryTreatmentType {
-        match pattern_type {
-            QueryPatternType::OnlyTemporal | QueryPatternType::OneTemporalOneSpatial => {
-                let fn_name = match_result.get_function_name().unwrap_or_default();
-                match fn_name.parse::<PromQLFunction>() {
-                    Ok(f) if f.is_approximate() => QueryTreatmentType::Approximate,
-                    _ => QueryTreatmentType::Exact,
-                }
-            }
-            QueryPatternType::OnlySpatial => {
-                let op = match_result.get_aggregation_op().unwrap_or_default();
-                match op.parse::<AggregationOperator>() {
-                    Ok(o) if o.is_approximate() => QueryTreatmentType::Approximate,
-                    _ => QueryTreatmentType::Exact,
-                }
-            }
+    /// Get treatment type (Exact vs Approximate) for a query's statistics.
+    /// All constituent statistics for a query (e.g. avg's `[Sum, Count]`)
+    /// always agree on approximate-ness, so any one of them suffices.
+    fn get_treatment_type(statistics: &[Statistic]) -> QueryTreatmentType {
+        debug_assert!(
+            statistics
+                .iter()
+                .all(|s| s.is_approximate() == statistics[0].is_approximate()),
+            "all statistics for a query must agree on approximate-ness"
+        );
+        if statistics[0].is_approximate() {
+            QueryTreatmentType::Approximate
+        } else {
+            QueryTreatmentType::Exact
         }
     }
 
@@ -184,12 +172,19 @@ impl SingleQueryProcessor {
             Ok(a) => a,
             Err(_) => return false,
         };
-        let (pattern_type, match_result) = match self.match_pattern(&ast) {
+        let match_result = match self.match_pattern(&ast) {
             Some(x) => x,
             None => return true,
         };
 
-        if pattern_type == QueryPatternType::OnlyTemporal {
+        // OnlyTemporal only — a temporal function with no outer spatial
+        // aggregation. Deliberately not reduced to a QueryRequirements-based
+        // check: doing so would also start applying this point-count-based
+        // punting to a bare spatial `quantile()` aggregation, which has never
+        // happened before and isn't asked for here (see #508).
+        let is_only_temporal = match_result.tokens.contains_key("function")
+            && !match_result.tokens.contains_key("aggregation");
+        if is_only_temporal {
             let fn_name = match_result.get_function_name().unwrap_or_default();
             let parsed_fn = fn_name.parse::<PromQLFunction>();
             if matches!(
@@ -223,11 +218,24 @@ impl SingleQueryProcessor {
         let ast = promql_parser::parser::parse(&self.query)
             .map_err(|e| ControllerError::PromQLParse(e.to_string()))?;
 
-        let (pattern_type, match_result) = self.match_pattern(&ast).ok_or_else(|| {
+        let match_result = self.match_pattern(&ast).ok_or_else(|| {
+            ControllerError::PlannerError(format!("Unsupported query: {}", self.query))
+        })?;
+
+        // Statistics, data range, and grouping labels all come from the same
+        // canonical derivation query-engine uses (`build_query_requirements_promql`),
+        // rather than being independently re-derived here (see #508).
+        let requirements = build_query_requirements_promql(
+            &self.query,
+            &match_result,
+            &self.metric_schema,
+            self.data_ingestion_interval_ms,
+        )
+        .ok_or_else(|| {
             ControllerError::PlannerError(format!("Unsupported query: {}", self.query))
         })?;
 
-        let treatment_type = Self::get_treatment_type(pattern_type, &match_result);
+        let treatment_type = Self::get_treatment_type(&requirements.statistics);
 
         let (metric, spatial_filter) = get_metric_and_spatial_filter(&match_result);
 
@@ -237,28 +245,21 @@ impl SingleQueryProcessor {
             .ok_or_else(|| ControllerError::UnknownMetric(metric.clone()))?
             .clone();
 
-        let statistics = get_statistics_to_compute(pattern_type, &match_result).map_err(|err| {
-            ControllerError::PlannerError(format!(
-                "Unsupported statistic for query '{}': {}",
-                self.query, err
-            ))
-        })?;
-
         let mut window_cfg = IntermediateWindowConfig::default();
         set_window_parameters(
-            pattern_type,
+            requirements.data_range_ms,
             self.t_repeat_ms,
             self.data_ingestion_interval_ms,
-            "any", // aggregation_type doesn't matter (sliding always false)
             self.step_ms,
             &mut window_cfg,
-        );
+        )
+        .map_err(ControllerError::PlannerError)?;
 
-        let (rollup, subpopulation_labels) =
-            get_label_routing(pattern_type, &match_result, &all_labels);
+        let subpopulation_labels = requirements.grouping_labels;
+        let rollup = all_labels.difference(&subpopulation_labels);
 
         let configs = build_agg_configs_for_statistics(
-            &statistics,
+            &requirements.statistics,
             treatment_type,
             &subpopulation_labels,
             &rollup,
@@ -285,8 +286,7 @@ impl SingleQueryProcessor {
             Some(
                 get_cleanup_param(
                     self.cleanup_policy,
-                    pattern_type,
-                    &match_result,
+                    requirements.data_range_ms,
                     self.t_repeat_ms,
                     window_cfg.window_type,
                     self.range_duration_ms,
@@ -299,49 +299,3 @@ impl SingleQueryProcessor {
         Ok((configs, cleanup_param))
     }
 }
-
-/// Returns `(rollup, subpopulation_labels)` for a given PromQL pattern type.
-/// These are constant across all statistics in a query, so they are computed
-/// once before the per-statistic loop.
-fn get_label_routing(
-    pattern_type: QueryPatternType,
-    match_result: &PromQLMatchResult,
-    all_labels: &KeyByLabelNames,
-) -> (KeyByLabelNames, KeyByLabelNames) {
-    match pattern_type {
-        QueryPatternType::OnlyTemporal => (KeyByLabelNames::empty(), all_labels.clone()),
-        QueryPatternType::OnlySpatial => {
-            // Match Python: if no by/without modifier, spatial_output = [] (rollup gets all labels).
-            // promql_utilities::get_spatial_aggregation_output_labels has a topk patch that returns
-            // all_labels when there is no modifier, but the Python planner returns [] in that case.
-            let has_modifier = match_result
-                .tokens
-                .get("aggregation")
-                .and_then(|t| t.aggregation.as_ref())
-                .and_then(|a| a.modifier.as_ref())
-                .is_some();
-            let spatial_output = if has_modifier {
-                get_spatial_aggregation_output_labels(match_result, all_labels)
-            } else {
-                KeyByLabelNames::empty()
-            };
-            (all_labels.difference(&spatial_output), spatial_output)
-        }
-        QueryPatternType::OneTemporalOneSpatial => {
-            let fn_name = match_result.get_function_name().unwrap_or_default();
-            let agg_op = match_result.get_aggregation_op().unwrap_or_default();
-            let collapsable = fn_name
-                .parse::<PromQLFunction>()
-                .ok()
-                .zip(agg_op.parse::<AggregationOperator>().ok())
-                .is_some_and(|(f, o)| get_is_collapsable(f, o));
-            if !collapsable {
-                (KeyByLabelNames::empty(), all_labels.clone())
-            } else {
-                let spatial_output =
-                    get_spatial_aggregation_output_labels(match_result, all_labels);
-                (all_labels.difference(&spatial_output), spatial_output)
-            }
-        }
-    }
-}
diff --git a/asap-planner-rs/src/planner/sql.rs b/asap-planner-rs/src/planner/sql.rs
index 5aef496c..f7d13f2a 100644
--- a/asap-planner-rs/src/planner/sql.rs
+++ b/asap-planner-rs/src/planner/sql.rs
@@ -221,19 +221,27 @@ fn get_sql_statistics(name: &str) -> Result<Vec<Statistic>, ControllerError> {
     }
 }
 
-/// Enforces two invariants —
-/// `t_repeat_ms >= data_ingestion_interval_ms` (can't refresh faster than
-/// raw ingestion) and `duration_ms >= t_repeat_ms` (a precompute window must
-/// not outlive the query range it's sized for) — rather than silently
-/// picking `data_ingestion_interval_ms` or `t_repeat_ms` depending on query
-/// shape. Once both hold, `window_size_ms` is simply `t_repeat_ms`: at the
-/// single-scrape-interval boundary `duration_ms == data_ingestion_interval_ms`
-/// exactly, so both invariants together force `t_repeat_ms == data_ingestion_interval_ms`
-/// there, reproducing the old `Spatial` branch without a special case.
+/// Enforces `t_repeat_ms >= data_ingestion_interval_ms` (can't refresh faster
+/// than raw ingestion) and, for a genuinely multi-interval query,
+/// `duration_ms >= t_repeat_ms` (a precompute window must not outlive the
+/// query range it's sized for) — rather than silently picking
+/// `data_ingestion_interval_ms` or `t_repeat_ms` depending on query shape.
+///
+/// Relaxation, kept consistent with the PromQL side
+/// (`asap-planner-rs/src/planner/window.rs::set_window_parameters`): when
+/// `duration_ms == data_ingestion_interval_ms` exactly (the query's own range
+/// is exactly one scrape interval), the query only ever concerns a single
+/// precomputed bucket — it asks for "the current/latest bucket," not "the
+/// last N buckets" — so re-reading that one bucket less often than it's
+/// produced is always safe, and the `duration_ms >= t_repeat_ms` upper bound
+/// is skipped. `window_size_ms` is `data_ingestion_interval_ms` in that case,
+/// or `t_repeat_ms` otherwise.
 ///
 /// Old code used `t_repeat_ms` uncapped even when it exceeded the query's
 /// own duration (see `temporal_sum_t600` in `sql_integration.rs`, updated
-/// alongside this change to expect a `PlannerError` instead).
+/// alongside the original version of this change to expect a `PlannerError`
+/// instead — still correct, since that test's duration (300s) differs from
+/// `data_ingestion_interval_ms` (15s), so it isn't the relaxed case).
 fn compute_sql_window(
     time_info: &TimeInfo,
     data_ingestion_interval_ms: u64,
@@ -245,14 +253,19 @@ fn compute_sql_window(
         )));
     }
     let duration_ms = (time_info.get_duration() * 1000.0).round() as u64;
-    if duration_ms < t_repeat_ms {
-        return Err(ControllerError::PlannerError(format!(
-            "query duration ({duration_ms}ms) must be >= t_repeat_ms ({t_repeat_ms}ms)"
-        )));
-    }
+    let window_size_ms = if duration_ms == data_ingestion_interval_ms {
+        data_ingestion_interval_ms
+    } else {
+        if duration_ms < t_repeat_ms {
+            return Err(ControllerError::PlannerError(format!(
+                "query duration ({duration_ms}ms) must be >= t_repeat_ms ({t_repeat_ms}ms)"
+            )));
+        }
+        t_repeat_ms
+    };
     Ok(IntermediateWindowConfig {
-        window_size_ms: t_repeat_ms,
-        slide_interval_ms: t_repeat_ms,
+        window_size_ms,
+        slide_interval_ms: window_size_ms,
         window_type: WindowType::Tumbling,
     })
 }
diff --git a/asap-planner-rs/src/planner/window.rs b/asap-planner-rs/src/planner/window.rs
index b89a599c..7cbc9be9 100644
--- a/asap-planner-rs/src/planner/window.rs
+++ b/asap-planner-rs/src/planner/window.rs
@@ -1,5 +1,4 @@
 use asap_types::enums::WindowType;
-use promql_utilities::query_logics::enums::QueryPatternType;
 
 pub fn get_effective_repeat(t_repeat_ms: u64, step_ms: u64) -> u64 {
     if step_ms > 0 {
@@ -9,51 +8,84 @@ pub fn get_effective_repeat(t_repeat_ms: u64, step_ms: u64) -> u64 {
     }
 }
 
-pub fn should_use_sliding_window(
-    _query_pattern_type: QueryPatternType,
-    _aggregation_type: &str,
-) -> bool {
+pub fn should_use_sliding_window() -> bool {
     // HARDCODED: sliding windows crash Arroyo
     false
 }
 
+/// Sets tumbling window parameters for a query, enforcing the same invariant
+/// SQL's `compute_sql_window` does (`asap-planner-rs/src/planner/sql.rs`,
+/// kept consistent with this function — see the relaxation note there too):
+/// `t_repeat_ms >= data_ingestion_interval_ms` (can't refresh faster than raw
+/// ingestion), plus, for a genuinely multi-interval query, `data_range_ms >=
+/// t_repeat_ms` (a precompute window can't outlive the query range it's sized
+/// for). `step_ms >= data_ingestion_interval_ms` is also required for range
+/// queries (can't sample more finely than raw ingestion).
+///
+/// Relaxation: when `data_range_ms == data_ingestion_interval_ms`, the query's
+/// own range is exactly one scrape interval — always true for a spatial-only
+/// query (which has no range of its own, so its canonical range is defined as
+/// the interval), and sometimes true for a genuinely narrow temporal query
+/// (e.g. `rate(x[15s])` at a 15s scrape interval). Either way, such a query
+/// only ever concerns a single precomputed bucket — it asks for "the
+/// current/latest bucket," not "the last N buckets" — so re-reading that one
+/// bucket less often than it's produced is always safe: there's no fresher
+/// answer to give. The `t_repeat_ms <= data_range_ms` upper bound is skipped
+/// in this case, and the window is sized to exactly one interval regardless
+/// of `t_repeat_ms`. Without this relaxation, an ordinary dashboard panel
+/// running `sum(x)` refreshed once a minute over 15s-scraped data would be
+/// rejected outright, even though serving it is trivially correct (see #508).
+///
+/// Sliding windows are never used (they crash Arroyo), so this always
+/// produces a tumbling window.
 pub fn set_window_parameters(
-    query_pattern_type: QueryPatternType,
+    data_range_ms: u64,
     t_repeat_ms: u64,
     data_ingestion_interval_ms: u64,
-    aggregation_type: &str,
     step_ms: u64,
     config: &mut IntermediateWindowConfig,
-) {
-    let effective_repeat = get_effective_repeat(t_repeat_ms, step_ms);
-    let _use_sliding = should_use_sliding_window(query_pattern_type, aggregation_type);
-    // use_sliding is always false, so always tumbling
-    set_tumbling_window_parameters(
-        query_pattern_type,
-        effective_repeat,
-        data_ingestion_interval_ms,
-        config,
-    );
-}
+) -> Result<(), String> {
+    if t_repeat_ms < data_ingestion_interval_ms {
+        return Err(format!(
+            "t_repeat_ms ({t_repeat_ms}ms) must be >= data_ingestion_interval_ms ({data_ingestion_interval_ms}ms)"
+        ));
+    }
+    if step_ms > 0 && step_ms < data_ingestion_interval_ms {
+        return Err(format!(
+            "step_ms ({step_ms}ms) must be >= data_ingestion_interval_ms ({data_ingestion_interval_ms}ms)"
+        ));
+    }
 
-fn set_tumbling_window_parameters(
-    query_pattern_type: QueryPatternType,
-    effective_repeat: u64,
-    data_ingestion_interval_ms: u64,
-    config: &mut IntermediateWindowConfig,
-) {
-    match query_pattern_type {
-        QueryPatternType::OnlyTemporal | QueryPatternType::OneTemporalOneSpatial => {
-            config.window_size_ms = effective_repeat;
-            config.slide_interval_ms = effective_repeat;
-            config.window_type = WindowType::Tumbling;
-        }
-        QueryPatternType::OnlySpatial => {
-            config.window_size_ms = data_ingestion_interval_ms;
-            config.slide_interval_ms = data_ingestion_interval_ms;
-            config.window_type = WindowType::Tumbling;
+    let _use_sliding = should_use_sliding_window();
+    // use_sliding is always false, so always tumbling
+    let window_size_ms = if data_range_ms == data_ingestion_interval_ms {
+        data_ingestion_interval_ms
+    } else {
+        if data_range_ms < t_repeat_ms {
+            return Err(format!(
+                "query data range ({data_range_ms}ms) must be >= t_repeat_ms ({t_repeat_ms}ms)"
+            ));
         }
+        get_effective_repeat(t_repeat_ms, step_ms)
+    };
+
+    // A range query reads `step_ms`-spaced samples by summing whole tumbling
+    // buckets (see `validate_range_query_params` in
+    // asap-query-engine/src/engines/simple_engine/mod.rs, which rejects a
+    // range query at query time unless `step % tumbling_window_ms == 0`).
+    // Catch a window size incompatible with its own planning-time step_ms
+    // here, at planning time, instead of provisioning a window that would
+    // reject every range query using exactly the step_ms it was planned for.
+    if step_ms > 0 && !step_ms.is_multiple_of(window_size_ms) {
+        return Err(format!(
+            "step_ms ({step_ms}ms) must be a multiple of the computed window size ({window_size_ms}ms)"
+        ));
     }
+
+    config.window_size_ms = window_size_ms;
+    config.slide_interval_ms = window_size_ms;
+    config.window_type = WindowType::Tumbling;
+    Ok(())
 }
 
 /// A mutable window config holder used during planning
@@ -82,4 +114,65 @@ mod tests {
     fn effective_repeat_step_larger_than_t_repeat() {
         assert_eq!(get_effective_repeat(30_000, 300_000), 30_000);
     }
+
+    #[test]
+    fn set_window_parameters_temporal_shape() {
+        let mut config = IntermediateWindowConfig::default();
+        set_window_parameters(300_000, 60_000, 15_000, 0, &mut config).unwrap();
+        assert_eq!(config.window_size_ms, 60_000);
+        assert_eq!(config.slide_interval_ms, 60_000);
+        assert_eq!(config.window_type, WindowType::Tumbling);
+    }
+
+    #[test]
+    fn set_window_parameters_spatial_shape_reproduces_interval() {
+        // data_range_ms == data_ingestion_interval_ms (spatial-only query),
+        // t_repeat_ms also equal to the interval: unaffected by the relaxation.
+        let mut config = IntermediateWindowConfig::default();
+        set_window_parameters(15_000, 15_000, 15_000, 0, &mut config).unwrap();
+        assert_eq!(config.window_size_ms, 15_000);
+    }
+
+    #[test]
+    fn set_window_parameters_spatial_shape_allows_slower_repeat_than_interval() {
+        // A spatial-only query refreshed less often than the scrape interval
+        // (e.g. a dashboard panel polling every 60s over 15s-scraped data) is
+        // always safe to serve: re-reading the single precomputed bucket at
+        // any cadence gives the latest available answer. window_size stays
+        // exactly one interval regardless of t_repeat_ms.
+        let mut config = IntermediateWindowConfig::default();
+        set_window_parameters(15_000, 60_000, 15_000, 0, &mut config).unwrap();
+        assert_eq!(config.window_size_ms, 15_000);
+    }
+
+    #[test]
+    fn set_window_parameters_rejects_t_repeat_below_interval() {
+        let mut config = IntermediateWindowConfig::default();
+        assert!(set_window_parameters(300_000, 10_000, 15_000, 0, &mut config).is_err());
+    }
+
+    #[test]
+    fn set_window_parameters_rejects_data_range_below_t_repeat() {
+        let mut config = IntermediateWindowConfig::default();
+        assert!(set_window_parameters(30_000, 60_000, 15_000, 0, &mut config).is_err());
+    }
+
+    #[test]
+    fn set_window_parameters_rejects_step_below_interval() {
+        let mut config = IntermediateWindowConfig::default();
+        assert!(set_window_parameters(300_000, 60_000, 15_000, 10_000, &mut config).is_err());
+    }
+
+    #[test]
+    fn set_window_parameters_rejects_step_not_multiple_of_window_size() {
+        // t_repeat_ms (40_000) < step_ms (100_000), so window_size_ms =
+        // effective_repeat = t_repeat_ms = 40_000. But 100_000 is not a
+        // multiple of 40_000 (100_000 % 40_000 == 20_000) — this would
+        // otherwise provision a window that query-engine's
+        // validate_range_query_params rejects for exactly this step_ms.
+        let mut config = IntermediateWindowConfig::default();
+        let result = set_window_parameters(300_000, 40_000, 10_000, 100_000, &mut config);
+        assert!(result.is_err());
+        assert!(result.unwrap_err().contains("must be a multiple of"));
+    }
 }
diff --git a/asap-query-engine/src/bin/test_offline_precomputes.rs b/asap-query-engine/src/bin/test_offline_precomputes.rs
index dbd2b81e..632ace15 100644
--- a/asap-query-engine/src/bin/test_offline_precomputes.rs
+++ b/asap-query-engine/src/bin/test_offline_precomputes.rs
@@ -9,10 +9,22 @@ use clap::Parser;
 use serde::Deserialize;
 
 // Internal imports from QueryEngineRust
-use promql_utilities::query_logics::enums::{AggregationType, QueryPatternType, Statistic};
+use promql_utilities::query_logics::enums::{AggregationType, Statistic};
 use query_engine_rust::data_model::{AggregateCore, KeyByLabelValues, PrecomputedOutput};
 use query_engine_rust::precompute_operators::*;
 
+/// Simulated query shape for offline merge testing — this tool has no real
+/// query/tokens (it replays a dumped precompute file), so `--pattern-type`
+/// lets the caller manually pick which shape to pretend the data came from.
+/// Local to this binary rather than the shared `QueryPatternType` (deleted,
+/// see #508): nothing here derives from an actual `PromQLMatchResult`.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum PatternKind {
+    OnlyTemporal,
+    OnlySpatial,
+    OneTemporalOneSpatial,
+}
+
 /// CLI Arguments
 #[derive(Parser, Debug)]
 #[command(name = "test_offline_precomputes")]
@@ -396,7 +408,7 @@ fn group_precomputes_by_key(
 /// Reference: SimpleEngine::merge_precomputed_outputs
 fn test_merge_functionality(
     grouped_precomputes: &HashMap<Option<KeyByLabelValues>, Vec<Box<dyn AggregateCore>>>,
-    query_pattern_type: QueryPatternType,
+    query_pattern_type: PatternKind,
     aggregation_type: &str,
     window_config: Option<WindowConfig>,
 ) -> Result<MergedPrecomputes, Box<dyn std::error::Error>> {
@@ -424,7 +436,7 @@ fn test_merge_functionality(
 /// Merge all precomputes for each key (standard mode)
 fn test_merge_all(
     grouped_precomputes: &HashMap<Option<KeyByLabelValues>, Vec<Box<dyn AggregateCore>>>,
-    query_pattern_type: QueryPatternType,
+    query_pattern_type: PatternKind,
     aggregation_type: &str,
 ) -> Result<MergedPrecomputes, Box<dyn std::error::Error>> {
     let mut merged_results = HashMap::new();
@@ -477,7 +489,7 @@ fn test_merge_all(
 /// Merge precomputes using sliding windows
 fn test_merge_with_windows(
     grouped_precomputes: &HashMap<Option<KeyByLabelValues>, Vec<Box<dyn AggregateCore>>>,
-    query_pattern_type: QueryPatternType,
+    query_pattern_type: PatternKind,
     aggregation_type: &str,
     config: &WindowConfig,
 ) -> Result<MergedPrecomputes, Box<dyn std::error::Error>> {
@@ -611,10 +623,10 @@ fn print_window_merge_statistics(stats: &WindowMergeStatistics) {
 
 /// Determine if merging should happen based on pattern type and aggregation type
 /// Mirrors logic from simple_engine.rs:1360-1395
-fn should_merge(pattern_type: QueryPatternType, aggregation_type: &str) -> bool {
+fn should_merge(pattern_type: PatternKind, aggregation_type: &str) -> bool {
     match pattern_type {
-        QueryPatternType::OnlyTemporal | QueryPatternType::OneTemporalOneSpatial => true,
-        QueryPatternType::OnlySpatial => aggregation_type == "DeltaSetAggregator",
+        PatternKind::OnlyTemporal | PatternKind::OneTemporalOneSpatial => true,
+        PatternKind::OnlySpatial => aggregation_type == "DeltaSetAggregator",
     }
 }
 
@@ -882,14 +894,14 @@ fn print_load_statistics(stats: &LoadStatistics) {
 }
 
 /// Parse pattern type string to enum
-fn parse_pattern_type(s: &str) -> QueryPatternType {
+fn parse_pattern_type(s: &str) -> PatternKind {
     match s.to_lowercase().as_str() {
-        "temporal" | "only_temporal" => QueryPatternType::OnlyTemporal,
-        "spatial" | "only_spatial" => QueryPatternType::OnlySpatial,
-        "temporal_spatial" | "one_temporal_one_spatial" => QueryPatternType::OneTemporalOneSpatial,
+        "temporal" | "only_temporal" => PatternKind::OnlyTemporal,
+        "spatial" | "only_spatial" => PatternKind::OnlySpatial,
+        "temporal_spatial" | "one_temporal_one_spatial" => PatternKind::OneTemporalOneSpatial,
         _ => {
             eprintln!("Unknown pattern type '{}', defaulting to OnlyTemporal", s);
-            QueryPatternType::OnlyTemporal
+            PatternKind::OnlyTemporal
         }
     }
 }
diff --git a/asap-query-engine/src/engines/simple_engine/mod.rs b/asap-query-engine/src/engines/simple_engine/mod.rs
index b59e2138..501f7218 100644
--- a/asap-query-engine/src/engines/simple_engine/mod.rs
+++ b/asap-query-engine/src/engines/simple_engine/mod.rs
@@ -22,8 +22,9 @@ use crate::AggregateCore;
 use asap_types::enums::WindowType;
 use promql_utilities::ast_matching::{PromQLPattern, PromQLPatternBuilder};
 use promql_utilities::data_model::KeyByLabelNames;
+use promql_utilities::get_is_collapsable;
 use promql_utilities::query_logics::enums::{
-    AggregationOperator, AggregationType, PromQLFunction, QueryPatternType, Statistic,
+    AggregationOperator, AggregationType, PromQLFunction, Statistic,
 };
 use serde_json::Value;
 
@@ -137,7 +138,7 @@ pub struct SimpleEngine {
     /// clone the Arc pointer, then use without holding the lock.
     streaming_config: RwLock<Arc<StreamingConfig>>,
     data_ingestion_interval_ms: u64,
-    controller_patterns: HashMap<QueryPatternType, Vec<PromQLPattern>>,
+    controller_patterns: Vec<PromQLPattern>,
     query_language: QueryLanguage,
 }
 
@@ -211,16 +212,6 @@ impl SimpleEngine {
         ]
         .map(AggregationOperator::as_str)
         .to_vec();
-        let spatial_ops_no_topk: Vec<&str> = [
-            AggregationOperator::Sum,
-            AggregationOperator::Count,
-            AggregationOperator::Avg,
-            AggregationOperator::Quantile,
-            AggregationOperator::Min,
-            AggregationOperator::Max,
-        ]
-        .map(AggregationOperator::as_str)
-        .to_vec();
         spatial_pattern_blocks.insert(
             "generic".to_string(),
             PromQLPatternBuilder::aggregation(
@@ -248,39 +239,70 @@ impl SimpleEngine {
             PromQLPattern::new(blocks[pattern_type].clone())
         }
 
-        let spatial_of_temporal_pattern =
-            |temporal_block: &Option<HashMap<String, Value>>| -> PromQLPattern {
+        // A spatial aggregation wrapping a temporal function only collapses into a
+        // single equivalent statistic for specific (function, op) pairs (see
+        // `get_is_collapsable`) — e.g. `sum(min_over_time(x[5m]))` cannot be served
+        // by a single precomputed statistic the way `sum(sum_over_time(x[5m]))` can.
+        // Building one narrow pattern per collapsable pair (rather than a broad
+        // any-op-wrapping-any-function pattern) means a non-collapsable combination
+        // never structurally matches at all, instead of matching and then silently
+        // dropping the outer aggregation (see #508).
+        let one_temporal_one_spatial_collapsable_patterns: Vec<PromQLPattern> = [
+            PromQLFunction::Rate,
+            PromQLFunction::Increase,
+            PromQLFunction::SumOverTime,
+            PromQLFunction::CountOverTime,
+            PromQLFunction::AvgOverTime,
+            PromQLFunction::MinOverTime,
+            PromQLFunction::MaxOverTime,
+            PromQLFunction::QuantileOverTime,
+        ]
+        .into_iter()
+        .flat_map(|func| {
+            [
+                AggregationOperator::Sum,
+                AggregationOperator::Count,
+                AggregationOperator::Avg,
+                AggregationOperator::Quantile,
+                AggregationOperator::Min,
+                AggregationOperator::Max,
+            ]
+            .into_iter()
+            .filter_map(move |op| {
+                if !get_is_collapsable(func, op) {
+                    return None;
+                }
+                let range_vector = PromQLPatternBuilder::matrix_selector(
+                    PromQLPatternBuilder::metric(None, None, None, Some("metric")),
+                    None,
+                    Some("range_vector"),
+                );
+                let function_pattern = PromQLPatternBuilder::function(
+                    vec![func.as_str()],
+                    vec![range_vector],
+                    Some("function"),
+                    Some("function_args"),
+                );
                 let pattern = PromQLPatternBuilder::aggregation(
-                    spatial_ops_no_topk.clone(),
-                    temporal_block.clone(),
+                    vec![op.as_str()],
+                    function_pattern,
                     None,
                     None,
                     None,
                     Some("aggregation"),
                 );
-                PromQLPattern::new(pattern)
-            };
+                Some(PromQLPattern::new(pattern))
+            })
+        })
+        .collect();
 
-        // Create controller patterns
-        let mut controller_patterns = HashMap::new();
-        controller_patterns.insert(
-            QueryPatternType::OnlyTemporal,
-            vec![
-                temporal_pattern("quantile", &temporal_pattern_blocks),
-                temporal_pattern("generic", &temporal_pattern_blocks),
-            ],
-        );
-        controller_patterns.insert(
-            QueryPatternType::OnlySpatial,
-            vec![spatial_pattern("generic", &spatial_pattern_blocks)],
-        );
-        controller_patterns.insert(
-            QueryPatternType::OneTemporalOneSpatial,
-            vec![
-                spatial_of_temporal_pattern(&temporal_pattern_blocks["quantile"]),
-                spatial_of_temporal_pattern(&temporal_pattern_blocks["generic"]),
-            ],
-        );
+        // Create controller patterns: tried in order until one matches.
+        let mut controller_patterns = vec![
+            temporal_pattern("quantile", &temporal_pattern_blocks),
+            temporal_pattern("generic", &temporal_pattern_blocks),
+            spatial_pattern("generic", &spatial_pattern_blocks),
+        ];
+        controller_patterns.extend(one_temporal_one_spatial_collapsable_patterns);
 
         Self {
             store,
@@ -326,37 +348,6 @@ impl SimpleEngine {
             .cloned()
     }
 
-    /// Validates and potentially aligns end timestamp based on query pattern
-    fn validate_and_align_end_timestamp(
-        &self,
-        mut end_timestamp: u64,
-        query_pattern_type: QueryPatternType,
-    ) -> u64 {
-        let interval_ms = self.data_ingestion_interval_ms;
-
-        if !end_timestamp.is_multiple_of(interval_ms) {
-            warn!(
-                "Query end timestamp {} is not aligned with data ingestion interval of {} ms. \
-                 This may lead to inaccurate results.",
-                end_timestamp, self.data_ingestion_interval_ms
-            );
-        }
-
-        // For OnlySpatial, align end_timestamp to nearest scrape interval
-        if query_pattern_type == QueryPatternType::OnlySpatial
-            && !end_timestamp.is_multiple_of(interval_ms)
-        {
-            let aligned_end_timestamp = (end_timestamp / interval_ms) * interval_ms;
-            debug!(
-                "OnlySpatial query: Aligning end_timestamp from {} to {} using data ingestion interval of {} ms",
-                end_timestamp, aligned_end_timestamp, self.data_ingestion_interval_ms
-            );
-            end_timestamp = aligned_end_timestamp;
-        }
-
-        end_timestamp
-    }
-
     /// Creates query parameters for separate keys query
     fn create_keys_query_params(
         &self,
diff --git a/asap-query-engine/src/engines/simple_engine/promql.rs b/asap-query-engine/src/engines/simple_engine/promql.rs
index cda3623b..a9a2d546 100644
--- a/asap-query-engine/src/engines/simple_engine/promql.rs
+++ b/asap-query-engine/src/engines/simple_engine/promql.rs
@@ -14,13 +14,8 @@ use asap_types::query_requirements::build_query_requirements_promql;
 use asap_types::PromQLSchema;
 use promql_utilities::ast_matching::PromQLMatchResult;
 use promql_utilities::data_model::KeyByLabelNames;
-use promql_utilities::get_is_collapsable;
-use promql_utilities::query_logics::enums::{
-    AggregationOperator, PromQLFunction, QueryPatternType, Statistic,
-};
-use promql_utilities::query_logics::parsing::{
-    get_metric_and_spatial_filter, get_spatial_aggregation_output_labels, get_statistics_to_compute,
-};
+use promql_utilities::query_logics::enums::Statistic;
+use promql_utilities::query_logics::parsing::get_metric_and_spatial_filter;
 use std::collections::HashMap;
 use std::time::Instant;
 use tracing::{debug, warn};
@@ -42,33 +37,30 @@ fn detect_scalar_arm<'a>(
 }
 
 impl SimpleEngine {
-    /// Calculates start timestamp for PromQL queries
-    fn calculate_start_timestamp_promql(
-        &self,
-        end_timestamp: u64,
-        query_pattern_type: QueryPatternType,
-        match_result: &PromQLMatchResult,
-    ) -> u64 {
-        match query_pattern_type {
-            QueryPatternType::OnlyTemporal | QueryPatternType::OneTemporalOneSpatial => {
-                // promql-parser supports a literal `ms` duration suffix (e.g. `[500ms]`),
-                // so .num_seconds() would truncate genuinely sub-second range vectors to 0.
-                let range_ms = match_result
-                    .get_range_duration()
-                    .unwrap()
-                    .num_milliseconds() as u64;
-                end_timestamp - range_ms
-            }
-            QueryPatternType::OnlySpatial => end_timestamp - self.data_ingestion_interval_ms,
+    /// Aligns `end_timestamp` down to the nearest data-ingestion-interval
+    /// boundary, unconditionally — mirroring SQL's `align_end_timestamp_sql`.
+    /// A no-op in the common case (already-aligned timestamps), a safety net
+    /// otherwise, for every PromQL query shape, not just a subset.
+    fn align_end_timestamp_promql(&self, end_timestamp: u64) -> u64 {
+        let interval_ms = self.data_ingestion_interval_ms;
+        if end_timestamp.is_multiple_of(interval_ms) {
+            return end_timestamp;
         }
+        let aligned = (end_timestamp / interval_ms) * interval_ms;
+        warn!(
+            "PromQL query end timestamp {} is not aligned with data ingestion interval of {} ms; \
+             aligning down to {}.",
+            end_timestamp, interval_ms, aligned
+        );
+        aligned
     }
 
     /// Calculates and validates query timestamps for PromQL
     fn calculate_query_timestamps_promql(
         &self,
         query_time: u64,
-        query_pattern_type: QueryPatternType,
         match_result: &PromQLMatchResult,
+        data_range_ms: u64,
     ) -> QueryTimestamps {
         let mut end_timestamp = if let Some(at_modifier) = match_result
             .tokens
@@ -81,9 +73,8 @@ impl SimpleEngine {
             query_time
         };
 
-        end_timestamp = self.validate_and_align_end_timestamp(end_timestamp, query_pattern_type);
-        let start_timestamp =
-            self.calculate_start_timestamp_promql(end_timestamp, query_pattern_type, match_result);
+        end_timestamp = self.align_end_timestamp_promql(end_timestamp);
+        let start_timestamp = end_timestamp - data_range_ms;
 
         QueryTimestamps {
             start_timestamp,
@@ -91,56 +82,52 @@ impl SimpleEngine {
         }
     }
 
-    /// Extracts quantile parameter from PromQL match result
-    fn extract_quantile_param_promql(
-        &self,
-        query_pattern_type: QueryPatternType,
-        match_result: &PromQLMatchResult,
-    ) -> Option<String> {
-        let quantile_value = match query_pattern_type {
-            QueryPatternType::OnlyTemporal | QueryPatternType::OneTemporalOneSpatial => {
-                match_result
-                    .tokens
-                    .get("function_args")
-                    .and_then(|token| token.function.as_ref())
-                    .and_then(|func| func.args.first())
-            }
-            QueryPatternType::OnlySpatial => match_result
-                .tokens
-                .get("aggregation")
-                .and_then(|token| token.aggregation.as_ref())
-                .and_then(|agg| agg.param.as_ref()),
-        };
+    /// Extracts quantile parameter from PromQL match result. Quantile only
+    /// ever appears in exactly one of these two token locations for a given
+    /// match (mutually exclusive — collapsable spatial-of-temporal
+    /// combinations never involve quantile, see `get_is_collapsable`), so
+    /// trying one then the other needs no additional signal to pick between
+    /// them — asserted below rather than silently trusted.
+    fn extract_quantile_param_promql(&self, match_result: &PromQLMatchResult) -> Option<String> {
+        let function_args_quantile = match_result
+            .tokens
+            .get("function_args")
+            .and_then(|token| token.function.as_ref())
+            .and_then(|func| func.args.first());
+        let aggregation_quantile = match_result
+            .tokens
+            .get("aggregation")
+            .and_then(|token| token.aggregation.as_ref())
+            .and_then(|agg| agg.param.as_ref());
+
+        debug_assert!(
+            function_args_quantile.is_none() || aggregation_quantile.is_none(),
+            "quantile must appear in exactly one of function_args or aggregation, never both"
+        );
 
-        quantile_value.map(|s| s.to_string())
+        function_args_quantile
+            .or(aggregation_quantile)
+            .map(|s| s.to_string())
     }
 
-    /// Extracts topk k parameter from PromQL match result
-    fn extract_topk_param(
-        &self,
-        query_pattern_type: QueryPatternType,
-        match_result: &PromQLMatchResult,
-    ) -> Result<String, String> {
-        match query_pattern_type {
-            QueryPatternType::OnlySpatial => match_result
-                .tokens
-                .get("aggregation")
-                .and_then(|token| token.aggregation.as_ref())
-                .and_then(|agg| agg.param.as_ref())
-                .map(|s| s.to_string())
-                .ok_or_else(|| "Missing k parameter for top-k query".to_string()),
-            _ => Err(format!(
-                "Top-k statistic is only supported for OnlySpatial pattern, found {:?}",
-                query_pattern_type
-            )),
-        }
+    /// Extracts topk k parameter from PromQL match result. Topk is only ever
+    /// produced by a spatial `topk` aggregation (excluded from every
+    /// collapsable spatial-of-temporal pattern), so no shape check is needed
+    /// before looking at the aggregation token.
+    fn extract_topk_param(&self, match_result: &PromQLMatchResult) -> Result<String, String> {
+        match_result
+            .tokens
+            .get("aggregation")
+            .and_then(|token| token.aggregation.as_ref())
+            .and_then(|agg| agg.param.as_ref())
+            .map(|s| s.to_string())
+            .ok_or_else(|| "Missing k parameter for top-k query".to_string())
     }
 
     /// Builds query kwargs (quantile, k, etc.) for PromQL queries
     fn build_query_kwargs_promql(
         &self,
         statistic: &Statistic,
-        query_pattern_type: QueryPatternType,
         match_result: &PromQLMatchResult,
     ) -> Result<HashMap<String, String>, String> {
         let mut query_kwargs = HashMap::new();
@@ -148,13 +135,13 @@ impl SimpleEngine {
         match statistic {
             Statistic::Quantile => {
                 let quantile = self
-                    .extract_quantile_param_promql(query_pattern_type, match_result)
+                    .extract_quantile_param_promql(match_result)
                     .ok_or_else(|| "Missing quantile parameter for quantile query".to_string())?;
                 debug!("Extracted quantile value: {:?}", quantile);
                 query_kwargs.insert("quantile".to_string(), quantile);
             }
             Statistic::Topk => {
-                let k = self.extract_topk_param(query_pattern_type, match_result)?;
+                let k = self.extract_topk_param(match_result)?;
                 debug!("Extracted k value: {:?}", k);
                 query_kwargs.insert("k".to_string(), k);
             }
@@ -190,25 +177,19 @@ impl SimpleEngine {
             .cloned()
     }
 
-    /// Scans `self.controller_patterns` for the first `PromQLPattern` that matches
-    /// `ast`, returning its `QueryPatternType` and match result. `query` is used
-    /// only for debug logging.
+    /// Scans `self.controller_patterns` for the first `PromQLPattern` that
+    /// matches `ast`. `query` is used only for debug logging.
     fn find_matching_controller_pattern(
         &self,
         ast: &promql_parser::parser::Expr,
         query: &str,
-    ) -> Option<(QueryPatternType, PromQLMatchResult)> {
-        for (pattern_type, patterns) in &self.controller_patterns {
-            for pattern in patterns {
-                debug!(
-                    "Trying pattern type: {:?} for query: {}",
-                    pattern_type, query
-                );
-                let match_result = pattern.matches(ast);
-                debug!("Match result: {:?}", match_result);
-                if match_result.matches {
-                    return Some((*pattern_type, match_result));
-                }
+    ) -> Option<PromQLMatchResult> {
+        for pattern in &self.controller_patterns {
+            debug!("Trying pattern for query: {}", query);
+            let match_result = pattern.matches(ast);
+            debug!("Match result: {:?}", match_result);
+            if match_result.matches {
+                return Some(match_result);
             }
         }
         None
@@ -224,8 +205,7 @@ impl SimpleEngine {
     ) -> Option<QueryExecutionContext> {
         let query_time = Self::convert_query_time_to_data_time(time);
 
-        let (query_pattern_type, match_result) =
-            self.find_matching_controller_pattern(arm_ast, &query_config.query)?;
+        let match_result = self.find_matching_controller_pattern(arm_ast, &query_config.query)?;
 
         let agg_info = self
             .get_aggregation_id_info(query_config)
@@ -236,8 +216,8 @@ impl SimpleEngine {
             .ok()?;
 
         self.build_promql_execution_context_tail(
+            &query_config.query,
             &match_result,
-            query_pattern_type,
             query_time,
             agg_info,
         )
@@ -247,12 +227,19 @@ impl SimpleEngine {
     ///
     /// Called by `build_query_execution_context_from_ast` and
     /// `build_query_execution_context_promql` after pattern matching and
-    /// `agg_info` resolution are complete.  Computes labels, statistics,
+    /// `agg_info` resolution are complete. Computes labels, statistics,
     /// kwargs, metadata, query plan, and the final `QueryExecutionContext`.
+    ///
+    /// Labels, statistic, and data range come from `build_query_requirements_promql`
+    /// — the same canonical derivation used by the capability-matching fallback
+    /// and by `asap-planner-rs` — rather than being independently re-derived
+    /// here. Only quantile/topk kwargs (not part of `QueryRequirements`, since
+    /// they don't affect which aggregation satisfies the query) still read
+    /// `match_result` directly.
     fn build_promql_execution_context_tail(
         &self,
+        query: &str,
         match_result: &PromQLMatchResult,
-        query_pattern_type: QueryPatternType,
         query_time: u64,
         agg_info: AggregationIdInfo,
     ) -> Option<QueryExecutionContext> {
@@ -263,61 +250,43 @@ impl SimpleEngine {
             SchemaConfig::PromQL(schema) => schema,
             _ => return None,
         };
-        let all_labels = match promql_schema.get_labels(&metric).cloned() {
-            Some(labels) => labels,
-            None => {
-                warn!("No metric configuration found for '{}'", metric);
-                return None;
-            }
-        };
+        if promql_schema.get_labels(&metric).is_none() {
+            warn!("No metric configuration found for '{}'", metric);
+            return None;
+        }
 
-        let mut query_output_labels = match query_pattern_type {
-            QueryPatternType::OnlyTemporal => all_labels.clone(),
-            QueryPatternType::OnlySpatial => {
-                get_spatial_aggregation_output_labels(match_result, &all_labels)
-            }
-            QueryPatternType::OneTemporalOneSpatial => {
-                let temporal_aggregation = match_result.get_function_name().unwrap();
-                let spatial_aggregation = match_result.get_aggregation_op().unwrap();
-                let collapsable = temporal_aggregation
-                    .parse::<PromQLFunction>()
-                    .ok()
-                    .zip(spatial_aggregation.parse::<AggregationOperator>().ok())
-                    .is_some_and(|(f, o)| get_is_collapsable(f, o));
-                if collapsable {
-                    get_spatial_aggregation_output_labels(match_result, &all_labels)
-                } else {
-                    all_labels.clone()
-                }
-            }
-        };
+        let requirements = build_query_requirements_promql(
+            query,
+            match_result,
+            promql_schema,
+            self.data_ingestion_interval_ms,
+        )?;
 
-        let timestamps =
-            self.calculate_query_timestamps_promql(query_time, query_pattern_type, match_result);
+        let mut query_output_labels = requirements.grouping_labels;
 
-        let statistics_to_compute = get_statistics_to_compute(query_pattern_type, match_result)
-            .map_err(|err| {
-                warn!("{}", err);
-                err
-            })
-            .ok()?;
-        if statistics_to_compute.len() != 1 {
+        let timestamps = self.calculate_query_timestamps_promql(
+            query_time,
+            match_result,
+            requirements.data_range_ms,
+        );
+
+        if requirements.statistics.len() != 1 {
             warn!(
                 "Expected exactly one statistic to compute, found {}",
-                statistics_to_compute.len()
+                requirements.statistics.len()
             );
             return None;
         }
-        let statistic_to_compute = statistics_to_compute.first().unwrap();
+        let statistic_to_compute = requirements.statistics[0];
 
-        if *statistic_to_compute == Statistic::Topk {
+        if statistic_to_compute == Statistic::Topk {
             let mut new_labels = vec!["__name__".to_string()];
             new_labels.extend(query_output_labels.labels);
             query_output_labels = KeyByLabelNames::new(new_labels);
         }
 
         let query_kwargs = self
-            .build_query_kwargs_promql(statistic_to_compute, query_pattern_type, match_result)
+            .build_query_kwargs_promql(&statistic_to_compute, match_result)
             .map_err(|e| {
                 warn!("{}", e);
                 e
@@ -326,7 +295,7 @@ impl SimpleEngine {
 
         let metadata = QueryMetadata {
             query_output_labels: query_output_labels.clone(),
-            statistic_to_compute: *statistic_to_compute,
+            statistic_to_compute,
             query_kwargs,
         };
 
@@ -1042,14 +1011,14 @@ impl SimpleEngine {
 
         let found_match = self.find_matching_controller_pattern(ast, query);
 
-        let (query_pattern_type, match_result) = match found_match {
-            Some((pt, result)) => {
+        let match_result = match found_match {
+            Some(result) => {
                 let pattern_match_duration = pattern_match_start_time.elapsed();
                 debug!(
                     "Pattern matching took: {:.2}ms",
                     pattern_match_duration.as_secs_f64() * 1000.0
                 );
-                (pt, result)
+                result
             }
             None => {
                 warn!("No matching pattern found for query: {}", query);
@@ -1083,7 +1052,6 @@ impl SimpleEngine {
             let requirements = build_query_requirements_promql(
                 query,
                 &match_result,
-                query_pattern_type,
                 metric_schema,
                 self.data_ingestion_interval_ms,
             )?;
@@ -1094,12 +1062,8 @@ impl SimpleEngine {
                 .find_compatible_aggregation(&requirements)?
         };
 
-        let result = self.build_promql_execution_context_tail(
-            &match_result,
-            query_pattern_type,
-            query_time,
-            agg_info,
-        );
+        let result =
+            self.build_promql_execution_context_tail(query, &match_result, query_time, agg_info);
 
         let query_context_duration = query_context_start_time.elapsed();
         debug!(
diff --git a/asap-query-engine/src/engines/simple_engine/sql.rs b/asap-query-engine/src/engines/simple_engine/sql.rs
index daf7b31b..faf9df41 100644
--- a/asap-query-engine/src/engines/simple_engine/sql.rs
+++ b/asap-query-engine/src/engines/simple_engine/sql.rs
@@ -1527,12 +1527,11 @@ mod topk_pipeline_tests {
     }
 }
 
-/// `build_spatiotemporal_context`'s end_timestamp snap:
-/// unlike the old hardcoded `QueryPatternType::OnlyTemporal` call into the
-/// shared `validate_and_align_end_timestamp` (which only snaps for
-/// `OnlySpatial`), SQL now always snaps a misaligned end_timestamp down to
-/// the nearest data-ingestion-interval boundary, for every SQL query shape
-/// including genuine multi-interval SpatioTemporal queries.
+/// `build_spatiotemporal_context`'s end_timestamp snap: SQL always snaps a
+/// misaligned end_timestamp down to the nearest data-ingestion-interval
+/// boundary, for every SQL query shape including genuine multi-interval
+/// SpatioTemporal queries (PromQL's `align_end_timestamp_promql` mirrors this
+/// unconditional behavior too, see #508).
 #[cfg(test)]
 mod spatiotemporal_timestamp_alignment_tests {
     use super::SimpleEngine;
diff --git a/asap-query-engine/src/planner_client.rs b/asap-query-engine/src/planner_client.rs
index 925e15e5..b2b794d6 100644
--- a/asap-query-engine/src/planner_client.rs
+++ b/asap-query-engine/src/planner_client.rs
@@ -105,7 +105,9 @@ mod tests {
         ControllerConfig {
             query_groups: vec![QueryGroup {
                 id: Some(1),
-                queries: vec!["sum(rate(http_requests_total[5m]))".to_string()],
+                // sum+sum_over_time is collapsable (see get_is_collapsable); sum+rate is not,
+                // and non-collapsable combinations no longer match any pattern (#508).
+                queries: vec!["sum(sum_over_time(http_requests_total[5m]))".to_string()],
                 repetition_delay_ms: 60_000,
                 controller_options: ControllerOptions::default(),
                 step_ms: None,
diff --git a/asap-query-engine/src/stores/simple_map_store/legacy/global.rs b/asap-query-engine/src/stores/simple_map_store/legacy/global.rs
index 476c2665..73f36a18 100644
--- a/asap-query-engine/src/stores/simple_map_store/legacy/global.rs
+++ b/asap-query-engine/src/stores/simple_map_store/legacy/global.rs
@@ -137,7 +137,7 @@ impl LegacySimpleMapStoreGlobal {
         // Collect windows where read_count >= threshold
         let mut windows_to_remove: Vec<TimestampRange> = Vec::new();
 
-        for (timestamp_range, _) in time_map.iter() {
+        for timestamp_range in time_map.keys() {
             let read_count = read_count_map.get(timestamp_range).copied().unwrap_or(0);
 
             if read_count >= threshold {
diff --git a/asap-query-engine/src/stores/simple_map_store/legacy/per_key.rs b/asap-query-engine/src/stores/simple_map_store/legacy/per_key.rs
index b7807e1c..58cd41f8 100644
--- a/asap-query-engine/src/stores/simple_map_store/legacy/per_key.rs
+++ b/asap-query-engine/src/stores/simple_map_store/legacy/per_key.rs
@@ -122,7 +122,7 @@ impl LegacySimpleMapStorePerKey {
         // Collect windows where read_count >= threshold
         let mut windows_to_remove: Vec<TimestampRange> = Vec::new();
 
-        for (timestamp_range, _) in data.time_map.iter() {
+        for timestamp_range in data.time_map.keys() {
             let read_count = data.read_counts.get(timestamp_range).copied().unwrap_or(0);
 
             if read_count >= threshold {
diff --git a/asap-query-engine/src/tests/datafusion/plan_execution_temporal_tests.rs b/asap-query-engine/src/tests/datafusion/plan_execution_temporal_tests.rs
index ffad4085..fd081bd3 100644
--- a/asap-query-engine/src/tests/datafusion/plan_execution_temporal_tests.rs
+++ b/asap-query-engine/src/tests/datafusion/plan_execution_temporal_tests.rs
@@ -485,6 +485,59 @@ mod tests {
         );
     }
 
+    // ========================================================================
+    // Non-collapsable spatial-of-temporal combinations must be rejected (#508)
+    //
+    // Before this fix, any spatial op could wrap any temporal function
+    // structurally, and a non-collapsable combination (e.g. `sum(min_over_time(...))`)
+    // silently dropped the outer aggregation instead of being rejected — returning
+    // ungrouped per-series values as if the outer `sum(...)` had never been
+    // written. `get_is_collapsable` only allows sum+sum_over_time,
+    // sum+count_over_time, min+min_over_time, max+max_over_time.
+    // ========================================================================
+
+    #[tokio::test]
+    async fn test_non_collapsable_sum_of_min_over_time_is_rejected() {
+        let query = "sum(min_over_time(http_requests[5s]))";
+        let engine = create_engine_multi_timestamp_with_window(
+            "http_requests",
+            AggregationType::Sum,
+            vec!["host"],
+            vec![],
+            query,
+            5_000,
+            WindowType::Tumbling,
+        );
+
+        assert!(
+            engine
+                .build_query_execution_context_promql(query.to_string(), QUERY_TIME)
+                .is_none(),
+            "sum+min_over_time is not collapsable and must not match any pattern"
+        );
+    }
+
+    #[tokio::test]
+    async fn test_non_collapsable_avg_of_rate_is_rejected() {
+        let query = "avg(rate(http_requests[5s]))";
+        let engine = create_engine_multi_timestamp_with_window(
+            "http_requests",
+            AggregationType::Sum,
+            vec!["host"],
+            vec![],
+            query,
+            5_000,
+            WindowType::Tumbling,
+        );
+
+        assert!(
+            engine
+                .build_query_execution_context_promql(query.to_string(), QUERY_TIME)
+                .is_none(),
+            "avg+rate is not collapsable and must not match any pattern"
+        );
+    }
+
     // ========================================================================
     // Old-vs-New comparison tests for temporal queries
     // ========================================================================
diff --git a/asap-query-engine/src/tests/datafusion/plan_execution_tests.rs b/asap-query-engine/src/tests/datafusion/plan_execution_tests.rs
index 7f1c6135..8f942905 100644
--- a/asap-query-engine/src/tests/datafusion/plan_execution_tests.rs
+++ b/asap-query-engine/src/tests/datafusion/plan_execution_tests.rs
@@ -537,7 +537,10 @@ mod tests {
 
     #[tokio::test]
     async fn test_execute_plan_not_implemented_increase() {
-        // IncreaseAccumulator has no arroyo serde, so execute_plan should fail
+        // IncreaseAccumulator has no arroyo serde, so execute_plan should fail.
+        // Uses a bare OnlyTemporal query (not `sum(increase(...))`): Increase is
+        // never collapsable with any spatial aggregation (see get_is_collapsable),
+        // so a spatial wrapper would no longer match any pattern at all (#508).
         let inc = IncreaseAccumulator::new(Measurement::new(0.0), 0, Measurement::new(100.0), 10);
 
         let engine = create_engine_single_pop(
@@ -545,12 +548,12 @@ mod tests {
             AggregationType::Increase,
             vec!["host"],
             vec![(Some(vec!["host-a".to_string()]), Box::new(inc))],
-            "sum(increase(http_requests_total[10s])) by (host)",
+            "increase(http_requests_total[10s])",
         );
 
         let context = engine
             .build_query_execution_context_promql(
-                "sum(increase(http_requests_total[10s])) by (host)".to_string(),
+                "increase(http_requests_total[10s])".to_string(),
                 1000.0,
             )
             .expect("Should build context");