chore: fix the small nits collected from clean reviews

crates/devtools/tests/stage_pipeline.rs

@@ -9,6 +9,7 @@ use asap_devtools::planner_layering_examples::{
 use asap_devtools::planning_stages::{
     lower_promql_roots, run_planning_stages, PlanningStagesError,
 };
+use asap_plan_selection::RejectionCause;
 use asap_types::ir::flat::FlatDag;
 use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPDAGDocument};
 use asap_types::ir::QueryRoot;
@@ -135,14 +136,8 @@ fn example1_document_is_valid_and_committed_fixture_is_current() {
         assert!(!rejection["reason"].as_str().unwrap().is_empty());
         // A candidate that failed a check is not priced; one that lost
         // under the cost model is.
-        let failed_a_check = [
-            "failure_probability_resize_failed",
-            "accuracy_target_missed",
-            "raw_data_not_retained",
-            "latency_bound_exceeded",
-            "over_cluster_capacity",
-        ]
-        .contains(&rejection["cause"].as_str().unwrap());
+        let cause: RejectionCause = serde_json::from_value(rejection["cause"].clone()).unwrap();
+        let failed_a_check = cause.failed_a_check();
         assert_eq!(failed_a_check, !costs.contains_key(id), "{rejection}");
     }
     assert!(costs.contains_key(selected));

crates/executor/src/physical_planner/precompute.rs

@@ -82,8 +82,8 @@ pub fn raw_sample_row(
 }
 
 /// Input contract of a precompute boundary: raw sample rows for a raw time
-/// series scan (or a `TimeRange`, optionally over a `TimeShift`, of one),
-/// otherwise the stored population of its summary state.
+/// series scan (or a `TimeRange`, optionally over stacked `TimeShift`s, of
+/// one), otherwise the stored population of its summary state.
 pub fn boundary_schema(
     node: &PhysicalASAPDAGNode,
     nodes: &BTreeMap<u64, &PhysicalASAPDAGNode>,
@@ -93,7 +93,7 @@ pub fn boundary_schema(
         scan = nodes
             .get(&(*child as u64))
             .ok_or_else(|| invalid("missing precompute node"))?;
-        if let Payload::NonASAP(NonASAPOp::TimeShift { child, .. }) = &scan.payload {
+        while let Payload::NonASAP(NonASAPOp::TimeShift { child, .. }) = &scan.payload {
             scan = nodes
                 .get(&(*child as u64))
                 .ok_or_else(|| invalid("missing precompute node"))?;

crates/executor/tests/precompute.rs

@@ -2,7 +2,7 @@
 mod common;
 use asap_executor::physical_planner::precompute::{boundary_schema, raw_sample_schema};
 use common::compile_physical_asap_dag;
-use planner_types::ir::operator::Source;
+use planner_types::ir::operator::{Source, TimeShift};
 use planner_types::ir::schema::{DataType, Field, Schema};
 use planner_types::ir::{NonASAPOp, Operator, OperatorNode, TimeRangeKind};
 use std::{collections::BTreeMap, rc::Rc, time::Duration};
@@ -53,3 +53,35 @@ fn only_a_time_series_range_reads_raw_samples() {
     // A table range is not raw samples: it must be a stored population.
     assert_ne!(table.ok(), Some(raw_sample_schema()));
 }
+
+/// A range over stacked time shifts of a time series scan still reads raw
+/// samples: the boundary looks through every shift, not only the first.
+#[test]
+fn a_range_over_stacked_shifts_reads_raw_samples() {
+    let range = time_range_over(Source::TimeSeries { metric: "m".into() });
+    let NonASAPOp::TimeRange {
+        range: window,
+        kind,
+        child: scan,
+    } = range.non_asap().unwrap().clone()
+    else {
+        unreachable!()
+    };
+    let shifted = |offset_ms, child| {
+        OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeShift {
+            shift: TimeShift {
+                offset_ms,
+                at: None,
+            },
+            child,
+        }))
+        .unwrap()
+    };
+    let stacked = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::TimeRange {
+        range: window,
+        kind,
+        child: shifted(60_000, shifted(300_000, scan)),
+    }))
+    .unwrap();
+    assert_eq!(boundary(&stacked).unwrap(), raw_sample_schema());
+}

crates/logical-asap-optimization/Cargo.toml

@@ -5,8 +5,8 @@ edition = "2021"
 
 # #509 Stage 1, logical ASAP-aware optimization: Pass 1 per-computation
 # candidates and Pass 2 ASAP-aware CSE over asap-types' operator DAG, with
-# the analytical accuracy model. Depends on no front end and on no later
-# stage.
+# the analytical accuracy model. Depends on no later stage, and on no
+# query-language frontend except through the `test-support` feature.
 [dependencies]
 asap_sketchlib = { workspace = true }
 asap-types = { path = "../types" }

crates/logical-asap-optimization/src/accuracy/estimators/hll.rs

@@ -47,14 +47,18 @@ pub fn classic_relative_error(precision: u8, relative_error: Option<f64>) -> f64
 /// `relative_error` over `source` is left as ([`classic_guarantee`]).
 /// `relative_error` is written in Rust's shortest round-trip form, so it
 /// parses back to the same value.
-pub fn classic_failure_statistic(precision: u8, relative_error: f64, source: &str) -> String {
+pub(crate) fn classic_failure_statistic(
+    precision: u8,
+    relative_error: f64,
+    source: &str,
+) -> String {
     format!(
         "{CLASSIC_FAILURE_PROBABILITY}[p={precision},eps={relative_error}]({MAX_DISTINCT}[{source}])"
     )
 }
 
-/// The precision, relative error and source a [`classic_failure_statistic`]
-/// names.
+/// The precision, relative error and source a classic HLL failure
+/// statistic (`classic_failure_statistic`) names.
 pub fn parse_classic_failure_statistic(statistic: &str) -> Option<(u8, f64, &str)> {
     let rest = statistic
         .strip_prefix(CLASSIC_FAILURE_PROBABILITY)?
@@ -112,7 +116,8 @@ pub(crate) fn classic_guarantee(
     ))
 }
 
-/// The failure probability a [`classic_failure_statistic`] names, given the
+/// The failure probability a classic HLL failure statistic
+/// (`classic_failure_statistic`) names, given the
 /// source's distinct bound: `1` (no certificate) when the bound is outside
 /// the contract's range for this precision and relative error.
 pub fn classic_failure_probability(precision: u8, relative_error: f64, max_distinct: u64) -> f64 {

crates/logical-asap-optimization/src/asap_aware_cse.rs

@@ -393,24 +393,24 @@ pub fn capability_sharing_candidates(roots: &[QueryRoot]) -> Vec<CapabilityShari
             readers.entry(Rc::as_ptr(child)).or_default().push(node);
         }
     }
-    let mut groups: Vec<CapabilitySharingCandidate> = Vec::new();
+    let mut candidates: Vec<CapabilitySharingCandidate> = Vec::new();
     for node in &nodes {
         let Some(family) = shareable_family(node, readers.get(&Rc::as_ptr(node))) else {
             continue;
         };
-        match groups
+        match candidates
             .iter_mut()
-            .find(|g| g.family == family && same_input_data(&g.members[0], node))
+            .find(|c| c.family == family && same_input_data(&c.members[0], node))
         {
-            Some(group) => group.members.push(Rc::clone(node)),
-            None => groups.push(CapabilitySharingCandidate {
+            Some(candidate) => candidate.members.push(Rc::clone(node)),
+            None => candidates.push(CapabilitySharingCandidate {
                 family,
                 members: vec![Rc::clone(node)],
             }),
         }
     }
-    groups.retain(|g| !all_equal(g.members.iter()));
-    groups
+    candidates.retain(|c| !all_equal(c.members.iter()));
+    candidates
 }
 
 fn collect_children_first(
@@ -510,25 +510,28 @@ pub fn summary_capability_pass2_candidates(
     roots: &[QueryRoot],
     candidates: &[CapabilitySharingCandidate],
 ) -> Result<Vec<SummaryCapabilityPass2Candidate>, PerComputationCandidateError> {
-    // Every subset of a group is a candidate: 2^n of them. Past
-    // MAX_CAPABILITY_SHARING_MEMBERS that is no longer a list plan selection can price.
+    // Every subset of a capability sharing candidate is a Pass 2
+    // candidate: 2^n of them. Past MAX_CAPABILITY_SHARING_MEMBERS that is
+    // no longer a list plan selection can price.
     if candidates
         .iter()
-        .any(|group| group.members.len() > MAX_CAPABILITY_SHARING_MEMBERS)
+        .any(|candidate| candidate.members.len() > MAX_CAPABILITY_SHARING_MEMBERS)
     {
         return Err(PerComputationCandidateError::Unsupported(
             "a capability sharing candidate of more than 20 summary builds (2^n sharing subsets)",
         ));
     }
-    // Each group's options: `None` (no sharing), then its subsets.
+    // Each capability sharing candidate's options: `None` (no sharing),
+    // then its subsets.
     let options: Vec<Vec<Option<Vec<usize>>>> = candidates
         .iter()
-        .map(|group| {
-            let n = group.members.len();
+        .map(|candidate| {
+            let n = candidate.members.len();
             let mut options = vec![None];
             for mask in 1u64..(1u64 << n) {
                 let members: Vec<usize> = (0..n).filter(|&m| mask >> m & 1 == 1).collect();
-                if members.len() >= 2 && !all_equal(members.iter().map(|&m| &group.members[m])) {
+                if members.len() >= 2 && !all_equal(members.iter().map(|&m| &candidate.members[m]))
+                {
                     options.push(Some(members));
                 }
             }
@@ -538,11 +541,11 @@ pub fn summary_capability_pass2_candidates(
     let mut choice = vec![0usize; candidates.len()];
     let mut pass2_candidates = Vec::new();
     loop {
-        // Mixed-radix increment, the last group varying fastest.
+        // Mixed-radix increment, the last candidate varying fastest.
         let mut carried = true;
-        for (digit, group_options) in choice.iter_mut().zip(&options).rev() {
+        for (digit, candidate_options) in choice.iter_mut().zip(&options).rev() {
             *digit += 1;
-            if *digit < group_options.len() {
+            if *digit < candidate_options.len() {
                 carried = false;
                 break;
             }
@@ -554,12 +557,12 @@ pub fn summary_capability_pass2_candidates(
         let shared_builds: Vec<Vec<Rc<OperatorNode>>> = choice
             .iter()
             .zip(candidates.iter().zip(&options))
-            .filter_map(|(&option, (group, group_options))| {
-                let members = group_options[option].as_ref()?;
+            .filter_map(|(&option, (candidate, candidate_options))| {
+                let members = candidate_options[option].as_ref()?;
                 Some(
                     members
                         .iter()
-                        .map(|&m| Rc::clone(&group.members[m]))
+                        .map(|&m| Rc::clone(&candidate.members[m]))
                         .collect(),
                 )
             })

crates/logical-asap-optimization/src/lib.rs

@@ -7,10 +7,13 @@
 //! invisible at the logical level ([`asap_aware_cse`], Pass 2), e.g.
 //! `Quantile(x, 0.99)` and `Quantile(x, 0.95)` reading one built sketch.
 //!
-//! It depends only on `asap-types`, never on a front end or a later stage:
-//! Stage 2 (`asap-physical-asap-optimization`) and Stage 3
-//! (`asap-plan-selection`) depend on it, and Cargo enforces that one-way
-//! flow.
+//! It depends only on `asap-types`, never on a later stage: Stage 2
+//! (`asap-physical-asap-optimization`) and Stage 3 (`asap-plan-selection`)
+//! depend on it, and Cargo enforces that one-way flow. It depends on no
+//! query-language frontend either, except with the `test-support` feature,
+//! whose test helpers (`test_support`) lower PromQL test queries through
+//! `asap-frontend-promql`; the later stages enable that feature only for
+//! their tests.
 //!
 //! Post-lowering **canonicalization** is *not* here: it lives in
 //! `asap_types::ir::canonicalize`, run inside `asap_frontend_common`'s shared

crates/logical-asap-optimization/src/test_support.rs

@@ -11,6 +11,13 @@ use asap_types::workload::{
     QueryRequirements, QueryWorkload,
 };
 
+/// The statistic classic HLL's failure probability at `precision` and
+/// `relative_error` over `source` is left as, for Stage 3 tests that bind
+/// it. The estimator keeps the format crate-private.
+pub fn hll_classic_failure_statistic(precision: u8, relative_error: f64, source: &str) -> String {
+    crate::accuracy::estimators::hll::classic_failure_statistic(precision, relative_error, source)
+}
+
 pub fn lower_promql(query: &str, accuracy: AccuracyTarget) -> Rc<OperatorNode> {
     let workload = PlanningWorkload {
         query_workload: QueryWorkload {

crates/logical-asap-optimization/src/workload_candidates.rs

@@ -8,7 +8,7 @@ use std::rc::Rc;
 
 use asap_types::ir::flat::flatten;
 use asap_types::ir::{OperatorNode, QueryRoot};
-use asap_types::workload::{QueryRecurrence, RepeatedDemand};
+use asap_types::workload::QueryRecurrence;
 
 use crate::asap_aware_cse::{
     capability_sharing_candidates, identical_expression_pass2_candidates, pane_candidates,
@@ -79,7 +79,10 @@ pub fn stage1_logical_asap_optimization<'a, Id: Clone>(
     per_computation: &'a PerComputationCandidates<Id>,
     recurrence: &[QueryRecurrence],
 ) -> impl Iterator<Item = Stage1Candidate> + 'a {
-    let interval_ms: Vec<Option<u64>> = recurrence.iter().map(fixed_interval_ms).collect();
+    let interval_ms: Vec<Option<u64>> = recurrence
+        .iter()
+        .map(QueryRecurrence::fixed_interval_ms)
+        .collect();
     let mut choice = Some(vec![0; per_computation.targets.len()]);
     let mut queue = std::collections::VecDeque::new();
     let mut position = 0usize;
@@ -355,17 +358,6 @@ fn next_choice<Id>(inventory: &PerComputationCandidates<Id>, choice: &mut [usize
     false
 }
 
-/// The fixed interval at which a query repeats, in milliseconds.
-pub fn fixed_interval_ms(recurrence: &QueryRecurrence) -> Option<u64> {
-    match recurrence {
-        QueryRecurrence::Repeated(
-            RepeatedDemand::FixedInterval(interval)
-            | RepeatedDemand::FixedIntervalAt { interval, .. },
-        ) => Some(u64::from(interval.0)).filter(|&ms| ms > 0),
-        _ => None,
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

crates/physical-asap-optimization/src/lib.rs

@@ -93,7 +93,6 @@ use asap_types::workload::{
 use thiserror::Error;
 
 use asap_logical_asap_optimization::deployment_capabilities::DeploymentCapabilities;
-use asap_logical_asap_optimization::workload_candidates::fixed_interval_ms;
 
 /// One CandidatePhysicalASAPDAG (Stage 2 output), derived from exactly one
 /// CandidateLogicalASAPDAG (Stage 1 output). It holds planning data only;
@@ -241,7 +240,7 @@ pub(crate) fn builds_one_pane_per_evaluation<'a>(
     let mut any_reader = false;
     for recurrence in readers {
         any_reader = true;
-        if fixed_interval_ms(recurrence) != Some(pane_width_ms) {
+        if recurrence.fixed_interval_ms() != Some(pane_width_ms) {
             return false;
         }
         if let QueryRecurrence::Repeated(RepeatedDemand::FixedIntervalAt {

crates/plan-selection/src/lib.rs

@@ -571,7 +571,7 @@ pub struct Rejection {
 /// candidate failed a check and is invalid ([`Self::failed_a_check`]); the
 /// rest mean it was valid and lost under the cost model. JSON: the
 /// snake_case variant name, e.g. `"costlier"`.
-#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize)]
+#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
 #[serde(rename_all = "snake_case")]
 pub enum RejectionCause {
     /// Its summaries, resized for the queries' failure probabilities, do
@@ -2092,7 +2092,7 @@ fn price(
     let mut held: HashMap<PhysicalASAPNodeId, (u64, f64)> = HashMap::new();
     // The raw data query-time scans read, kept on disk: per source, the
     // largest scan's (bytes, node).
-    let mut raw_data: HashMap<String, (u64, PhysicalASAPNodeId)> = HashMap::new();
+    let mut raw_data: HashMap<&Source, (u64, PhysicalASAPNodeId)> = HashMap::new();
     // Each in-memory ingestion-time state's bytes per window (what it keeps
     // for one window, its output or an exact aggregate's values).
     let mut maintained: Vec<(PhysicalASAPNodeId, u64)> = Vec::new();
@@ -2554,9 +2554,7 @@ fn price(
         }
         if let Payload::NonASAP(NonASAPOp::Scan { source, .. }) = &node.payload {
             if !ingestion {
-                let largest = raw_data
-                    .entry(format!("{source:?}"))
-                    .or_insert((out.bytes, node.id));
+                let largest = raw_data.entry(source).or_insert((out.bytes, node.id));
                 if out.bytes > largest.0 {
                     *largest = (out.bytes, node.id);
                 }
@@ -3186,7 +3184,9 @@ mod tests {
     /// unknown.
     #[test]
     fn classic_hll_failure_probability_binds_only_its_own_sources_bound() {
-        let statistic = hll::classic_failure_statistic(15, 0.05, "a");
+        let statistic = asap_logical_asap_optimization::test_support::hll_classic_failure_statistic(
+            15, 0.05, "a",
+        );
         let bounded = |source: &str| {
             let mut data = data();
             data.sources

crates/plan-selection/src/measured_operation_costs.rs

@@ -5,11 +5,11 @@
 //! with given parameters costs (e.g. sketch-bench's `rqe_atomic_costs.json`,
 //! converted by `asap-devtools`' `convert_rqe_atomic_costs`) and puts the
 //! table in [`PlanSelectionSettings::measured_operation_costs`]. Stage 3
-//! prices a sketch node (an operator node of the physical plan DAG) from
-//! the entry for its sketch parameters and operation when there is one, and
-//! from its formulas otherwise, and reports which per node
-//! ([`NodeCost::price_source`]). The planner reads
-//! only this table, never a benchmark file.
+//! prices a sketch node's (an operator node of the physical plan DAG) CPU
+//! from the entry for its sketch parameters and operation when there is
+//! one, and from its formulas otherwise, and reports which per node
+//! ([`NodeCost::price_source`]). The planner reads only this table, never a
+//! benchmark file.
 //!
 //! A measured time enters [`ResourceUsage`] as CPU operations: `t` ns is
 //! `t · 10⁻⁶ / ms_per_cpu_op` operations of [`ResourceSpeed`], so latency

crates/types/src/ir/operator/operator_properties.rs

@@ -178,7 +178,7 @@ pub enum DataModel {
 /// The leaf data source of a `Scan`. The schema itself rides on the
 /// `Scan.schema` field (SchemaResolver-built); `Source` carries only the leaf's
 /// identity.
-#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+#[derive(Debug, Clone, PartialEq, Eq, Hash, Serialize, Deserialize)]
 pub enum Source {
     /// Time-series leaf — PromQL / DC lifecycle. Produces `(ts, value, *labels)`.
     TimeSeries { metric: String },