chore!: accumulated review nits (column-metric and viewer fixes, timing names, lost tests, stale comments)

crates/asap-aware-mapping/src/accuracy/composition.rs

@@ -263,7 +263,7 @@ impl DefaultAccuracyModel {
                 operator: op.clone(),
                 input_metrics: inputs.iter().map(|g| g.spec.accuracy_metric).collect(),
                 local_metric: None,
-                reason: "exact average requires AbsoluteValue input guarantees".into(),
+                reason: "exact average requires absolute-error input guarantees".into(),
             });
         }
         let mut provenance = Vec::new();
@@ -315,7 +315,7 @@ impl DefaultAccuracyModel {
                 operator: op.clone(),
                 input_metrics: inputs.iter().map(|g| g.spec.accuracy_metric).collect(),
                 local_metric: None,
-                reason: "exact max/min requires AbsoluteValue input guarantees".into(),
+                reason: "exact max/min requires absolute-error input guarantees".into(),
             });
         }
         let mut provenance = Vec::new();
@@ -869,12 +869,12 @@ pub(super) fn propagate(
             }
             if inputs.len() != 1 || !same_metric(AccuracyMetric::ABSOLUTE) {
                 return Err(unsupported(
-                    "Lipschitz rule is registered for exactly one AbsoluteValue input".into(),
+                    "Lipschitz rule is registered for exactly one absolute-error input".into(),
                 ));
             }
             if local.is_some_and(|g| g.spec.accuracy_metric != AccuracyMetric::ABSOLUTE) {
                 return Err(unsupported(
-                    "Lipschitz rule needs an AbsoluteValue local guarantee".into(),
+                    "Lipschitz rule needs an absolute-error local guarantee".into(),
                 ));
             }
             Ok(DefaultAccuracyModel::lipschitz(

crates/asap-aware-mapping/src/accuracy/estimators/univmon.rs

@@ -415,7 +415,7 @@ mod tests {
         (0.2, 0.2),
     ];
 
-    /// Each L2 shape meets its target as `RelativeValue`, and one fewer
+    /// Each L2 shape meets its target as a relative error, and one fewer
     /// row or column does not.
     #[test]
     fn l2_shape_is_the_smallest_meeting_its_target() {

crates/asap-aware-mapping/src/pass/mod.rs

@@ -281,7 +281,7 @@ impl PlanOutput {
     pub fn execution_timed_dag(
         &self,
     ) -> Result<asap_types::ir::physical_export::PhysicalASAPDAG, ExecutionDataStateError> {
-        // One memo, so a node shared by several roots is timed and exported once.
+        // One memo, so a node several roots share by `Rc` is timed and exported once.
         let mut memo = TimingMemo::new();
         let assignment = MaterializationAssignment::default();
         let timed = self
@@ -619,6 +619,28 @@ mod tests {
         assert!(check_contract(&output, "stub", 0).is_ok());
     }
 
+    /// An empty workload has no DAG: the export's `NoRoots`, wrapped in
+    /// `InvalidPhysicalDAG`. A workload of only literal scalars exports a DAG
+    /// with no nodes, just its `scalar_roots`.
+    #[test]
+    fn execution_timed_dag_of_an_empty_and_a_literal_only_workload() {
+        use asap_types::ir::physical_export::PhysicalASAPDAGValidationError;
+        use asap_types::ir::ScalarExpr;
+        assert_eq!(
+            PlanOutput::new(Vec::new()).execution_timed_dag().err(),
+            Some(ExecutionDataStateError::InvalidPhysicalDAG(
+                PhysicalASAPDAGValidationError::NoRoots
+            ))
+        );
+        let mut literal = PlanOutput::new(Vec::new());
+        literal.scalar_roots = vec![(0, ScalarExpr::literal_f64(1.0))];
+        let dag = literal.execution_timed_dag().unwrap();
+        assert!(dag.nodes.is_empty() && dag.roots.is_empty());
+        let [(0, ScalarExpr::Literal(_))] = dag.scalar_roots.as_slice() else {
+            panic!("one literal scalar root: {:?}", dag.scalar_roots)
+        };
+    }
+
     /// Registering a name twice fails instead of overwriting, so a comparison
     /// run cannot silently measure one pass twice.
     #[test]

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

@@ -29,7 +29,7 @@
 //! other combination of options is one physical candidate, the
 //! all-`NotMaterialized` one first: nothing is capped. A node that one
 //! query reads at ingestion time and another at query time is copied, one
-//! copy per phase ([`split_shared_by_phase`]).
+//! copy per timing ([`split_shared_by_timing`]).
 //!
 //! A build may run at ingestion time (S4) only when:
 //!
@@ -76,7 +76,7 @@ use asap_types::ir::physical_export::{
     compile_physical_asap_workload_with_node_ids, PhysicalASAPDAG, PhysicalASAPNodeId,
 };
 use asap_types::ir::{
-    apply_materialization_timings, is_filtered_build, split_shared_by_phase, ASAPOp,
+    apply_materialization_timings, is_filtered_build, split_shared_by_timing, ASAPOp,
     MaterializationAssignment, Medium, NonASAPOp, Operator, OperatorNode, ProjectItem, QueryRoot,
     ScalarExpr, SchemaDerivationError, SortKey, TimingMemo,
 };
@@ -301,7 +301,7 @@ pub fn optimize_physical_asap(
             .iter()
             .flat_map(|root| root_operators(root).into_iter().cloned())
             .collect();
-        let (split, options) = split_shared_by_phase(&operators, &options);
+        let (split, options) = split_shared_by_timing(&operators, &options);
         let replaced: HashMap<_, _> = operators
             .iter()
             .map(Rc::as_ptr)
@@ -1047,11 +1047,12 @@ mod tests {
     /// `root` with its summary build in 1-min tumbling panes.
     fn in_panes(root: QueryRoot) -> QueryRoot {
         let roots = [root];
-        let sites = crate::asap_aware_cse::pane_candidates(&roots, &[Some(MINUTE)]);
-        assert_eq!(sites.len(), 1, "{sites:?}");
-        let mut variants =
-            crate::asap_aware_cse::window_composition_pass2_candidates(&roots, &sites).unwrap();
-        variants.pop().unwrap().roots.unwrap().remove(0)
+        let rule_candidates = crate::asap_aware_cse::pane_candidates(&roots, &[Some(MINUTE)]);
+        assert_eq!(rule_candidates.len(), 1, "{rule_candidates:?}");
+        let mut pass2_candidates =
+            crate::asap_aware_cse::window_composition_pass2_candidates(&roots, &rule_candidates)
+                .unwrap();
+        pass2_candidates.pop().unwrap().roots.unwrap().remove(0)
     }
 
     const AT_INGESTION: Materialization = Materialization::MatAtIngestion(Medium::Memory);
@@ -1399,9 +1400,9 @@ mod tests {
 
     /// Two queries share one scan: one reads it through panes maintained at
     /// ingestion time, the other at query time. The scan is split, one copy
-    /// per phase; with nothing materialized it stays one node.
+    /// per timing; with nothing materialized it stays one node.
     #[test]
-    fn a_scan_two_queries_read_at_two_phases_is_split() {
+    fn a_scan_two_queries_read_at_two_timings_is_split() {
         let QueryRoot::Operator(panes) = in_panes(summary_root("quantile_over_time(0.99, m[5m])"))
         else {
             unreachable!()

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

@@ -776,13 +776,20 @@ pub fn select_plan(
 /// order; `None` without candidates.
 fn choose(ranking: Ranking, costs: &[&CandidateCost]) -> Option<(usize, Vec<usize>)> {
     let front: Vec<usize> = match ranking {
-        Ranking::Front => (0..costs.len())
-            .filter(|&i| {
-                !costs
-                    .iter()
-                    .any(|other| dominates(&other.usage, &costs[i].usage))
-            })
-            .collect(),
+        Ranking::Front => {
+            // Each candidate's worst latency once, not once per comparison.
+            let latencies: Vec<f64> = costs.iter().map(|c| worst_latency_ms(&c.usage)).collect();
+            (0..costs.len())
+                .filter(|&i| {
+                    !(0..costs.len()).any(|j| {
+                        dominates_at(
+                            (&costs[j].usage, latencies[j]),
+                            (&costs[i].usage, latencies[i]),
+                        )
+                    })
+                })
+                .collect()
+        }
         _ => Vec::new(),
     };
     // Under the Pareto front, only a front member can be selected.
@@ -843,7 +850,11 @@ fn capacity_violation(usage: &ResourceUsage, limits: &ClusterLimits) -> Option<S
 /// Whether `a` is at least as good as `b` on peak memory, disk and
 /// worst-query latency, and better on one of them.
 fn dominates(a: &ResourceUsage, b: &ResourceUsage) -> bool {
-    let (la, lb) = (worst_latency_ms(a), worst_latency_ms(b));
+    dominates_at((a, worst_latency_ms(a)), (b, worst_latency_ms(b)))
+}
+
+/// [`dominates`] with each usage's worst-query latency already computed.
+fn dominates_at((a, la): (&ResourceUsage, f64), (b, lb): (&ResourceUsage, f64)) -> bool {
     a.peak_memory_bytes <= b.peak_memory_bytes
         && a.disk_bytes <= b.disk_bytes
         && !cheaper(lb, la)
@@ -3712,6 +3723,65 @@ mod tests {
         assert_eq!(kll_params(&select(&per_value).plan), stage1);
     }
 
+    /// With its metric declared a counter, the whole-expression CMS+heap
+    /// candidate (P2) proves its weights non-negative, is valid and is
+    /// priced: its per-node costs cover its DAG and sum to its total.
+    #[test]
+    fn a_declared_counter_prices_the_count_min_heap_candidate() {
+        use asap_types::workload::MetricType;
+        let candidates = candidates();
+        let target = AccuracyTarget::max_error(0.01)
+            .failure_probability(0.001)
+            .accuracy_metric(AccuracyMetric::ABSOLUTE_STREAM_L1);
+        let counter = with_metric_types(data(), &[("m", MetricType::Counter)]);
+        let selection = select_plan(
+            &candidates,
+            &[entry_with(target, repeating(10_000))],
+            &counter,
+            PlanningModels::builtin().with_evidence(&EveryOutputMeasuredExact),
+        )
+        .unwrap();
+        let p2 = candidates.iter().find(|c| c.id == "P2").unwrap();
+        let cost = &selection.costs["P2"];
+        let mut ids: Vec<_> = p2.dag.nodes.iter().map(|n| n.id).collect();
+        ids.sort();
+        let mut keys: Vec<_> = cost.per_node.keys().copied().collect();
+        keys.sort();
+        assert_eq!(keys, ids);
+        let sum: f64 = cost
+            .per_node
+            .values()
+            .map(|n| n.cost_per_second.unwrap())
+            .sum();
+        assert_eq!(cost.cost_per_second.unwrap(), sum);
+    }
+
+    /// One summary two queries read is sized for the stricter one: its
+    /// failure probability is the smallest share either query needs (the
+    /// largest `n · G` over its readers).
+    #[test]
+    fn a_shared_summary_takes_the_smallest_failure_probability_share() {
+        let (candidates, lenient) = per_series_kll(AccuracyScope::PerValue);
+        let root = candidates[0].roots[0].clone();
+        let strict = lenient.clone().failure_probability(0.001);
+        let roots = [root.clone(), root];
+        for targets in [[lenient.clone(), strict.clone()], [strict, lenient]] {
+            let resized = resize_for_failure_probability(&roots, &targets, &data()).unwrap();
+            let kll = kll_params(&resized);
+            assert_eq!(kll.len(), 2, "{kll:?}");
+            for params in kll {
+                let SketchParams::Kll {
+                    failure_probability,
+                    ..
+                } = params
+                else {
+                    panic!("{params:?}")
+                };
+                assert_eq!(failure_probability, Some(0.001));
+            }
+        }
+    }
+
     /// Without a bound on the values one evaluation returns, a per-value
     /// guarantee does not meet a whole-result target: a PromQL vector whose
     /// source declares no series count, and a SQL relation (until a bound on
@@ -5363,13 +5433,15 @@ mod tests {
                 .into_iter()
                 .map(|(_, root)| root)
                 .collect();
-        let sites = crate::asap_aware_cse::pane_candidates(&roots, &[Some(u64::from(MINUTE))]);
-        let roots = crate::asap_aware_cse::window_composition_pass2_candidates(&roots, &sites)
-            .unwrap()
-            .pop()
-            .unwrap()
-            .roots
-            .unwrap();
+        let rule_candidates =
+            crate::asap_aware_cse::pane_candidates(&roots, &[Some(u64::from(MINUTE))]);
+        let roots =
+            crate::asap_aware_cse::window_composition_pass2_candidates(&roots, &rule_candidates)
+                .unwrap()
+                .pop()
+                .unwrap()
+                .roots
+                .unwrap();
         let mut candidates = optimize_physical_asap(
             "L1",
             &roots,

crates/asap-aware-mapping/tests/per_computation_candidates.rs

@@ -621,6 +621,63 @@ fn having_on_one_consumer_of_a_shared_aggregate_stays_above() {
     assert_eq!(targets[0].having.len(), 2);
 }
 
+/// A scalar reference to a grouped aggregate (`Filter(scan, EXISTS G)` as a
+/// second root) is a consumer that reads it without the `HAVING`, so the
+/// first root's key-only `HAVING` stays above the output.
+#[test]
+fn a_scalar_reference_consumer_keeps_a_key_only_having_above() {
+    let node = grouped_count(None);
+    let key = equals(0, ScalarValue::Utf8("a".into()));
+    let filtered = having_filter(node.clone(), key.clone());
+    let exists = having_filter(
+        scan(false),
+        Predicate(ScalarExpr::Exists {
+            subquery: node.clone(),
+            negated: false,
+        }),
+    );
+    let targets = generate_per_computation_candidates(vec![
+        (0, QueryRoot::Operator(filtered.clone())),
+        (1, QueryRoot::Operator(exists)),
+    ])
+    .unwrap()
+    .targets;
+    let target = targets
+        .iter()
+        .find(|target| Rc::ptr_eq(&target.aggregate, &node))
+        .expect("the grouped count's target");
+    assert_eq!(
+        target.having,
+        vec![HavingEntry {
+            filter: Some(filtered),
+            predicate: key,
+            placement: HavingPlacement::AboveOutput,
+        }]
+    );
+}
+
+/// Of two stacked `Filter`s above an aggregate, only the adjacent one is
+/// its `HAVING`; the outer one filters the inner `Filter`'s output.
+#[test]
+fn only_the_adjacent_of_stacked_filters_is_having() {
+    let node = grouped_count(None);
+    let inner_key = equals(0, ScalarValue::Utf8("a".into()));
+    let inner = having_filter(node.clone(), inner_key.clone());
+    let outer = having_filter(inner.clone(), equals(0, ScalarValue::Utf8("b".into())));
+    let targets = generate_per_computation_candidates(vec![(0, QueryRoot::Operator(outer))])
+        .unwrap()
+        .targets;
+    assert_eq!(targets.len(), 1);
+    assert_eq!(
+        targets[0].having,
+        vec![HavingEntry {
+            filter: Some(inner),
+            predicate: inner_key,
+            placement: HavingPlacement::BelowAggregate,
+        }]
+    );
+}
+
 /// A global aggregate's `HAVING` always stays above the output, even a
 /// constant one: the one output row exists with no input rows.
 #[test]

crates/asap-physical-operators/src/physical_planner/mod.rs

@@ -519,7 +519,7 @@ fn compile_internal(
                 }
             }
             if let Payload::ASAP(ASAPOp::FinalizeExactAccumulator { .. }) = &node.payload {
-                // Exact counts read out as Int64; PromQL declares a Float64 sample.
+                // Exact counts evaluate to Int64; PromQL declares a Float64 sample.
                 let evaluation = bind_operation(node, &schemas)
                     .map_err(|error| invalid(format!("node {id}: {error}")))?;
                 let actual = evaluation.schema();
@@ -678,8 +678,6 @@ fn temporal_evaluation_drops_name(node: &PhysicalASAPDAGNode) -> bool {
         )
 }
 
-/// Bind a Planner node against the schemas supplied by its deployment edges.
-/// This is the same checked path used by complete DAG binding.
 /// The IR accepts a `SummaryMerge` over any state family; the executor runs
 /// one only when the family-capability table gives the family a sound merge.
 pub(crate) fn require_sound_merge(input: &SchemaRef) -> Result<(), Error> {
@@ -692,6 +690,8 @@ pub(crate) fn require_sound_merge(input: &SchemaRef) -> Result<(), Error> {
     Ok(())
 }
 
+/// Bind a Planner node against the schemas supplied by its deployment edges.
+/// This is the same checked path used by complete DAG binding.
 pub fn compile_node(node: &PhysicalASAPDAGNode, inputs: &[SchemaRef]) -> Result<Operator, Error> {
     for schema in inputs {
         crate::values::validate_schema(schema)?;

crates/asap-physical-operators/src/physical_planner/promql_values.rs

@@ -221,6 +221,9 @@ pub fn exact_state_schema(family: SummaryFamilyType) -> Result<SchemaRef, Error>
 }
 
 /// Retain exact evaluation semantics before any deployment state is opened.
+/// A counter (`Rate`, `Increase`) is evaluated over the `lookback_ms`
+/// window ending at the evaluation time: this entry point takes no
+/// offset, so its counter window has offset 0.
 pub fn compile_exact_evaluation(
     family: SummaryFamilyType,
     lookback_ms: u64,

crates/asap-physical-operators/src/summary_kernels/factory.rs

@@ -245,11 +245,12 @@ impl AccumulatorUpdater for CmsAccumulatorUpdater {
 // ---------------------------------------------------------------------------
 
 /// Keyed frequency updater backed by sketchlib's per-row-seeded weighted
-/// frequency kernel, the state the DAG's keyed summary build produces: plain
-/// CountSketch (no candidate heap), Count-Min with a top-k heap, or
-/// CountSketch with a top-k heap. Each row hashes with its own seed, so rows
-/// stay independent at any depth. Each label value of the key is one `Utf8`
-/// identity, as in the DAG's keyed build.
+/// frequency kernel: plain CountSketch (no candidate heap), Count-Min with a
+/// top-k heap, or CountSketch with a top-k heap. For the two heap families
+/// this is the state the DAG's keyed summary build produces; plain
+/// CountSketch uses the same kernel without a heap. Each row hashes with
+/// its own seed, so rows stay independent at any depth. Each label value of
+/// the key is one `Utf8` identity, as in the DAG's keyed build.
 ///
 /// Each raw Prometheus sample contributes its `value` as a Float64 weight,
 /// kept exactly (no rounding). CountSketch accepts signed weights; Count-Min