Select dashboard tumbling pane sizes and enforce window correctness

control_plane/src/emit/stage_config.rs

@@ -2991,6 +2991,11 @@ pub(crate) fn build_backend_aggregation_json(agg: &BackendAggregation) -> JsonVa
             obj.insert("weight_mode".into(), JsonValue::String(mode.into()));
         }
     }
+    // PromQL range selectors are (start, end]. Encode the boundary convention
+    // in state identity so legacy half-open panes cannot satisfy this binding.
+    if matches!(agg.aggregation_input, AggregationInput::Raw) {
+        parameters["promql_right_closed"] = json!(true);
+    }
     let aggregation_input = match agg.aggregation_input {
         AggregationInput::SketchEnvelope => "sketch_envelope",
         AggregationInput::Raw => "raw",

control_plane/src/main.rs

@@ -826,6 +826,7 @@ fn compile_physical_plan_request(
     }
 
     let planning_request = physical::compiler::PlanningRequest {
+        query_workload: None,
         queries,
         evidence: request.evidence,
         planner_revision: request.planner_revision,

control_plane/src/physical/post_asap/cost_model.rs

@@ -63,7 +63,7 @@ pub struct ControlPlaneCostModel {
     pub workload_accuracy: AccuracyTarget,
     lifecycle_costs: SummaryMaintenanceLifecycleCostInputs,
     summary_maintenance: SummaryMaintenanceCapabilities,
-    window_framework_costs: Vec<(SummaryWindowFramework, Cost)>,
+    window_framework_costs: Vec<(Option<String>, SummaryWindowFramework, Cost)>,
     offline_evidence: Option<EmpiricalEvidenceProvider>,
     offline_frequency_comparison: Option<(OfflineComparisonEvidence, OfflineComparisonRequest)>,
 }
@@ -213,15 +213,27 @@ impl ControlPlaneCostModel {
         costs.into_iter().map(|(algorithm, _)| algorithm).collect()
     }
 
-    /// Bind the cheapest complete, workload-scoped physical realization for
-    /// each Planner-owned abstract window framework. Concrete implementation
-    /// identities stay in the physical compiler; only framework and cost
-    /// cross into Planner's candidate comparison.
+    /// Keep concrete physical identities in Planner's complete-candidate estimate,
+    /// including distinct pane sizes using the same abstract window framework.
+    pub fn with_window_implementation_costs(
+        mut self,
+        costs: Vec<(String, SummaryWindowFramework, Cost)>,
+    ) -> Self {
+        self.window_framework_costs = costs
+            .into_iter()
+            .map(|(id, framework, cost)| (Some(id), framework, cost))
+            .collect();
+        self
+    }
+
     pub fn with_window_framework_costs(
         mut self,
         costs: Vec<(SummaryWindowFramework, Cost)>,
     ) -> Self {
-        self.window_framework_costs = costs;
+        self.window_framework_costs = costs
+            .into_iter()
+            .map(|(framework, cost)| (None, framework, cost))
+            .collect();
         self
     }
 
@@ -387,14 +399,19 @@ impl CostModel for ControlPlaneCostModel {
             // GOS/error propagation is introduced by the later adaptation
             // slice. Until then, do not claim an approximate exponential
             // histogram window is exact.
-            .filter(|(framework, _)| {
+            .filter(|(_, framework, _)| {
                 !matches!(framework, SummaryWindowFramework::ExponentialHistogram)
             })
-            .filter(|(_, cost)| cost.0.is_finite() && cost.0 >= 0.0)
-            .min_by(|left, right| left.1 .0.total_cmp(&right.1 .0))
+            .filter(|(_, _, cost)| cost.0.is_finite() && cost.0 >= 0.0)
+            .min_by(|left, right| {
+                left.2
+                     .0
+                    .total_cmp(&right.2 .0)
+                    .then_with(|| left.0.cmp(&right.0))
+            })
             .map(
-                |(framework, physical_cost)| CompleteSummaryCandidateEstimate {
-                    physical_plan_id: None,
+                |(id, framework, physical_cost)| CompleteSummaryCandidateEstimate {
+                    physical_plan_id: id.clone(),
                     cost: Cost(lifecycle_cost + physical_cost.0),
                     window_frameworks: vec![Some(framework.clone()); deployments.len()],
                     window_accuracy_guarantee: Some(

control_plane/src/physical/workload_cost.rs

@@ -518,6 +518,14 @@ mod tests {
         ] {
             let (mut request, env) = fixture().planning_request().unwrap();
             request.queries[0].query_string = query.into();
+            request
+                .query_workload
+                .as_mut()
+                .unwrap()
+                .repeating_queries
+                .as_mut()
+                .unwrap()[0]
+                .query = planner_types::workload::Query(query.into());
             let exact = with_exact_alternative(request).unwrap().pop().unwrap();
             let plan = PhysicalCompiler
                 .compile(exact.clone(), env.clone())
@@ -659,6 +667,16 @@ mod tests {
         let mut second = shared.queries[0].clone();
         second.query_id = "second-consumer".into();
         second.query_string = "sum(count_over_time(m[1m]))".into();
+        let entries = shared
+            .query_workload
+            .as_mut()
+            .unwrap()
+            .repeating_queries
+            .as_mut()
+            .unwrap();
+        let mut demand = entries[0].clone();
+        demand.query = planner_types::workload::Query(second.query_string.clone());
+        entries.push(demand);
         shared.queries.push(second);
         let roots = shared
             .queries

data_plane/src/precompute_engine/worker.rs

@@ -407,18 +407,33 @@ impl Worker {
         }
         let state = self.group_states.get_mut(&sid).unwrap();
 
+        // Keep original timestamps inside accumulators (notably rate/increase),
+        // shifting only pane membership and closure watermark for PromQL (a,b].
+        let right_closed = state
+            .config
+            .parameters
+            .get("promql_right_closed")
+            .and_then(serde_json::Value::as_bool)
+            .unwrap_or(false);
+        let pane_timestamp = |ts: i64| {
+            if right_closed {
+                ts.saturating_sub(1)
+            } else {
+                ts
+            }
+        };
         // Find the timestamp span in this batch. A first batch may contain
         // several windows (Prometheus commonly sends catch-up samples after
         // startup), so its minimum timestamp is also the initial closure
         // scan boundary.
         let batch_min_ts = samples
             .iter()
-            .map(|(_, ts, _)| *ts)
+            .map(|(_, ts, _)| pane_timestamp(*ts))
             .min()
             .unwrap_or(i64::MIN);
         let batch_max_ts = samples
             .iter()
-            .map(|(_, ts, _)| *ts)
+            .map(|(_, ts, _)| pane_timestamp(*ts))
             .max()
             .unwrap_or(i64::MIN);
         let previous_event_time = state.max_event_time_ms;
@@ -435,8 +450,9 @@ impl Worker {
         // Route each sample to its pane
         for (series_key, ts, val) in &samples {
             let too_late = previous_event_time != i64::MIN
-                && *ts < watermark_for_event_time(previous_event_time, allowed_lateness_ms);
-            let pane_start = state.window_manager.pane_start_for(*ts);
+                && pane_timestamp(*ts)
+                    < watermark_for_event_time(previous_event_time, allowed_lateness_ms);
+            let pane_start = state.window_manager.pane_start_for(pane_timestamp(*ts));
             let pane_end = pane_start + state.window_manager.slide_interval_ms();
             let pane_closed = !state.active_panes.contains_key(&pane_start)
                 && previous_closure_watermark >= pane_start + state.window_manager.window_size_ms();

data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs

@@ -110,6 +110,16 @@ pub fn execute_query_plan_instant(
     } else {
         now_ms.saturating_sub(entry.instant.lookback_ms)
     };
+    for binding in entry.materialization_bindings() {
+        if binding.window_ms == 0
+            || t0_ms % binding.window_ms != 0
+            || now_ms % binding.window_ms != 0
+        {
+            return Err(LoweringSkip::MaterializationNotReady(
+                "evaluation interval cuts a materialized pane".into(),
+            ));
+        }
+    }
     let outcome = execute_physical_query_plan(
         index,
         entry,
@@ -896,6 +906,101 @@ mod tests {
         assert_eq!(outcome.coverage, Some((2_000, 2_000)));
     }
 
+    #[test]
+    fn repeated_multi_pane_reads_exclude_expired_state_and_reject_gaps() {
+        let idx = SketchStore::new();
+        let policy = asap_types::PolicyFingerprint(777);
+        idx.register(SketchInstanceMetadata {
+            sid: 7,
+            metric_name: "requests_total".into(),
+            group_by_keys: std::collections::BTreeSet::new(),
+            capability: Some(Capability::ExactAgg(asap_types::AggregationType::Sum)),
+            agg_kind: AggKind::ExactAgg {
+                agg_type: asap_types::AggregationType::Sum,
+                parameters_canonical: String::new(),
+                spatial_filter_canonical: String::new(),
+            },
+            accuracy: None,
+            first_seen_unix_ms: 0,
+            retired_at_ms: None,
+            expires_at_ms: None,
+            policy_fp: policy,
+        });
+        // Old panes remain stored; each advancing query must select only its lookback.
+        for pane in 0..8 {
+            idx.append_precompute(
+                7,
+                BTreeMap::new(),
+                (pane * 10_000, (pane + 1) * 10_000),
+                Box::new(
+                    crate::precompute_engine::operators::SumAccumulator::with_sum(
+                        (pane + 1) as f64,
+                    ),
+                ),
+            );
+        }
+
+        let entry = control_plane::query_plan::QueryPlanEntry {
+            query_id: "q-rate".into(),
+            canonical_promql: "rate(requests_total[1m])".into(),
+            root: control_plane::query_plan::QueryNodeId(0),
+            nodes: BTreeMap::from([
+                (
+                    control_plane::query_plan::QueryNodeId(0),
+                    QueryPlanNode::ExactReadout {
+                        input: control_plane::query_plan::QueryNodeId(1),
+                        readout: control_plane::query_plan::ExactReadout::Sum,
+                    },
+                ),
+                (
+                    control_plane::query_plan::QueryNodeId(1),
+                    QueryPlanNode::ReadMaterialization {
+                        binding: control_plane::query_plan::MaterializationBinding {
+                            materialization: policy,
+                            metric: "requests_total".into(),
+                            sid_grouping: vec![],
+                            output_grouping: control_plane::query_plan::PhysicalGrouping::PerEntity,
+                            window_ms: 10_000,
+                        },
+                    },
+                ),
+            ]),
+            instant: control_plane::query_plan::InstantExecution {
+                lookback_ms: 60_000,
+                full_history: false,
+                cumulative_readout: true,
+            },
+            fallback: control_plane::query_plan::FallbackPolicy::ExactBackend,
+        };
+        for (now, expected) in [(60_000, 21.0), (70_000, 27.0), (80_000, 33.0)] {
+            let (outcome, _) = execute_query_plan_instant(&idx, &entry, now).unwrap();
+            assert_eq!(outcome.series[0].1[0].1, expected);
+        }
+        assert!(
+            execute_query_plan_instant(&idx, &entry, 70_001).is_err(),
+            "partial pane must fall back"
+        );
+        assert!(
+            execute_query_plan_instant(&idx, &entry, 90_000).is_err(),
+            "open/missing trailing pane must fall back"
+        );
+        // Both endpoints exist, but the missing interior pane is not evidence of zero samples.
+        let gap_idx = SketchStore::new();
+        idx.with_instance(7, |meta| gap_idx.register(meta.clone()));
+        for pane in [0, 1, 3, 4, 5] {
+            gap_idx.append_precompute(
+                7,
+                BTreeMap::new(),
+                (pane * 10_000, (pane + 1) * 10_000),
+                Box::new(crate::precompute_engine::operators::SumAccumulator::with_sum(1.0)),
+            );
+        }
+        assert!(
+            execute_query_plan_instant(&gap_idx, &entry, 60_000).is_err(),
+            "interior gap must fall back"
+        );
+    }
+
     #[test]
     fn exact_query_plan_rate_uses_reset_aware_readout() {
         let idx = SketchStore::new();

data_plane/src/query_engines/asap_query_engine/summary_executor.rs

@@ -372,6 +372,42 @@ impl QueryExecutionContext<'_> {
             ExactAgg(AggregationType),
         }
 
+        // A missing pane can mean delayed ingestion, not an empty interval.
+        // Until explicit empty-pane completion exists, multi-pane reads require
+        // every pane for each stored series, before any cross-series merge.
+        let check_panes = |ends: Vec<i64>| -> Result<(), SummaryExecutorError> {
+            let width = binding.window_ms;
+            if width == 0 {
+                return Err(SummaryExecutorError::Unsupported("zero pane width"));
+            }
+            if self.t1_ms.saturating_sub(self.t0_ms) <= width {
+                return Ok(());
+            }
+            if self.t0_ms % width != 0 || self.t1_ms % width != 0 {
+                return Err(SummaryExecutorError::Unsupported("partial pane interval"));
+            }
+            let mut expected = self.t0_ms.checked_add(width);
+            for end in ends {
+                let Ok(end) = u64::try_from(end) else {
+                    continue;
+                };
+                if end <= self.t0_ms {
+                    continue;
+                } // delta decoding carry-in is not an answer pane
+                if Some(end) != expected {
+                    return Err(SummaryExecutorError::Unsupported(
+                        "missing materialized pane",
+                    ));
+                }
+                expected = end.checked_add(width);
+            }
+            if expected != self.t1_ms.checked_add(width) {
+                return Err(SummaryExecutorError::Unsupported(
+                    "incomplete materialized panes",
+                ));
+            }
+            Ok(())
+        };
         let required_keys: BTreeSet<_> = binding.sid_grouping.iter().cloned().collect();
         let mut sids = self.index.sids_for_policy(binding.materialization);
         sids.sort_unstable();
@@ -407,8 +443,10 @@ impl QueryExecutionContext<'_> {
                         .into_iter()
                         .next()
                     else {
+                        check_panes(Vec::new())?;
                         continue;
                     };
+                    check_panes(series.samples.keys().copied().collect())?;
                     let key = match &binding.output_grouping {
                         PhysicalGrouping::PerEntity => series.series_label_values.clone(),
                         PhysicalGrouping::Reduce(keys) => {
@@ -427,8 +465,10 @@ impl QueryExecutionContext<'_> {
                         .into_iter()
                         .next()
                     else {
+                        check_panes(Vec::new())?;
                         continue;
                     };
+                    check_panes(windows.keys().copied().collect())?;
                     let key = match &binding.output_grouping {
                         PhysicalGrouping::PerEntity => labels,
                         PhysicalGrouping::Reduce(keys) => project_group_key(keys, &labels),