fix(planner): preserve maintained population selection horizons

Author: zzylol

crates/asap-aware-mapping/src/maintained_population.rs

@@ -119,11 +119,17 @@ fn recognize(root: &QueryExpr) -> Option<(MaintainedPopulation, PopulationReadou
         return Some((population, readout, Rc::clone(source)));
     }
     // A bare PromQL selector carries the declared ingestion interval as a
-    // temporal input scope. It still denotes the same current-series
-    // population; preserve the wrapper as the maintained input below.
-    let series_source = match source.as_ref() {
-        QueryExpr::TimeRange { child, .. } => child.as_ref(),
-        other => other,
+    // temporal input scope. Membership must expire at that horizon; retain
+    // the wrapper as the maintained input so validation can check agreement.
+    let (series_source, lookback_ms) = match source.as_ref() {
+        QueryExpr::TimeRange { range, child } => {
+            let ms = u64::try_from(range.as_millis()).ok()?;
+            if ms == 0 || std::time::Duration::from_millis(ms) != *range {
+                return None;
+            }
+            (child.as_ref(), ms)
+        }
+        other => (other, 300_000),
     };
     let QueryExpr::Scan {
         source: Source::TimeSeries { metric },
@@ -183,7 +189,7 @@ fn recognize(root: &QueryExpr) -> Option<(MaintainedPopulation, PopulationReadou
                 matchers,
                 grouping: labels,
                 without: grouping.is_without(),
-                lookback_ms: 300_000,
+                lookback_ms,
             }),
             max_k: 0,
             quantiles: false,

crates/frontend-promql/tests/maintained_population_horizon.rs

@@ -0,0 +1,37 @@
+mod support;
+use asap_aware_mapping::maintained_population::MaintainedPopulationStrategy;
+use asap_types::post_asap::maintained_population::PopulationInput;
+use asap_types::post_asap::{SummaryExpr, ValueOperation};
+use asap_types::types::AccuracyTarget;
+use std::rc::Rc;
+
+// A population for a one-second selector must expire members after one second.
+#[test]
+fn population_preserves_selector_horizon() {
+    let root = Rc::new(support::lower_promql("sum(a)", AccuracyTarget::Exact).unwrap());
+    let candidate = MaintainedPopulationStrategy::new(std::slice::from_ref(&root))
+        .candidate(&root)
+        .unwrap();
+    let SummaryExpr::ValueOperation { child, .. } = &candidate.expr else {
+        panic!()
+    };
+    let SummaryExpr::ValueOperation {
+        operation: ValueOperation::MaintainPopulation { population },
+        ..
+    } = &child.expr
+    else {
+        panic!()
+    };
+    let PopulationInput::CurrentSeries(spec) = &population.input else {
+        panic!()
+    };
+    assert_eq!(spec.lookback_ms, 1_000);
+    asap_types::post_asap::compile_executable_dag(&candidate).unwrap();
+    let asap_types::pre_asap::QueryExpr::Aggregate { child: source, .. } = root.as_ref() else {
+        panic!()
+    };
+    assert!(spec.matches_input(source));
+    let mut wrong = spec.clone();
+    wrong.lookback_ms = 300_000;
+    assert!(!wrong.matches_input(source));
+}

crates/types/src/post_asap/maintained_population.rs

@@ -40,11 +40,18 @@ impl CurrentSeriesInput {
     pub fn matches_input(&self, input: &crate::pre_asap::QueryExpr) -> bool {
         use crate::pre_asap::{CompareOpKind, DataType, QueryExpr, ScalarValue, Source};
         // PromQL instant selectors carry an ingestion-interval `TimeRange` as
-        // their input scope. Population membership remains the wrapped series
-        // selector; shifted and otherwise transformed inputs still fail below.
+        // their input scope. The population must use the same expiry horizon;
+        // shifted and otherwise transformed inputs still fail below.
         let input = match input {
-            QueryExpr::TimeRange { child, .. } => child.as_ref(),
-            other => other,
+            QueryExpr::TimeRange { range, child }
+                if self.lookback_ms > 0
+                    && *range == std::time::Duration::from_millis(self.lookback_ms) =>
+            {
+                child.as_ref()
+            }
+            QueryExpr::TimeRange { .. } => return false,
+            other if self.lookback_ms == 300_000 => other,
+            _ => return false,
         };
         let QueryExpr::Scan {
             source: Source::TimeSeries { metric },
@@ -58,7 +65,6 @@ impl CurrentSeriesInput {
             || *metric != self.metric
             || schema.closed
             || schema.time_index.is_none()
-            || self.lookback_ms != 300_000
         {
             return false;
         }