From 8d3aedae689896acf3c346446c66832c47f50697 Mon Sep 17 00:00:00 2001 From: zz_y Date: Wed, 13 May 2026 20:35:03 -0600 Subject: [PATCH] feat(sid): L4 binder rule emits PhysicalExpr::ExactAgg for Sum/Rate/Increase/Count MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PR 6 follow-up part 2. The PR 6 follow-up (PR #200) flipped `capability_for` to return `Some(Capability::ExactAgg(...))` for Sum / Rate / Increase / Count{Exact} so the analyzer surfaces those intents as warm-tier candidates. But the L4 binder had no rule to lower them into the algebra — they kept passing through as `PhysicalExpr::Logical`, which the L5 emitter routes to archive. This PR adds the missing binder so the optimizer's plan output finally reflects the analyzer flip. ## What 1. New variant `PhysicalExpr::ExactAgg { agg_type, child }` (`sketch_algebra/physical_expr.rs`). Counterpart to the data-plane `AggKind::ExactAgg` / `AggPayload::ExactAgg` shape. No `SketchEstimate` wrapper because ExactAgg produces the answer directly — the accumulator's value IS the result. 2. New convenience constructor `PhysicalExpr::exact_agg_over_logical`, mirroring the existing `estimate_over_agg` for sketches. 3. New binder rule `bind_exact_agg::BindExactAgg` (`sketch_algebra/rules/bind_exact_agg.rs`) with priority `2` (above `bind_archive_only`'s 1, below the sketch families at 4–6). Maps: - `AggIntent::Sum` → `ExactAgg(Sum)` - `AggIntent::Rate { window }` → `ExactAgg(Increase)` - `AggIntent::Increase { window }` → `ExactAgg(Increase)` - `AggIntent::Count{Exact}` → `ExactAgg(Sum)` (count = sum-of-1s) 4. Wired into `dispatch` in `sketch_algebra/rules/mod.rs`. 5. Exhaustive-match coverage added in three downstream walkers: - `emit::extract_root_sketch_kind` → `None` (no sketch family) - `optimizer::rules::extract_family` → `None` - `physical::colored_dag::allocator::visit` → `StageId::Edge` (same locality as `SketchAgg`; accumulator runs in the precompute pipeline at edge) - `sketch_algebra::tests::collect_sketch_kinds` → recurse into child - `sketch_algebra::tests::binding_is_archive` → `false` (warm path, not cold) 6. Updated two existing tests that asserted Sum's old behaviour: - `bind_no_match_passes_through_logical` → `sum_now_binds_to_exact_agg_after_pr_6_followup` - `phase_b_pattern_only_temporal_sum_falls_through_to_logical` → `phase_b_pattern_only_temporal_sum_binds_to_exact_agg` ## What's not in scope - `bind_exact_agg` rejects `Aggregate{by: [..]}` inputs. A keyed-group ExactAgg follow-up would emit `MultipleSum` / `MultipleIncrease` instead. The dispatch / allocator / emit walkers already handle ExactAgg structurally; only the binder logic gates on `by.is_empty()`. - `AggIntent::Avg` still routes to archive — needs cross-policy Sum+Count join, separate concern. - `AggIntent::Min` / `Max` stay on the quantile-sketch path (DDSketch/KLL answer them via quantile(0)/quantile(1)); adding a MinMax ExactAgg binding would compete with that path unprofiled. Decision deferred. ## Test plan - [x] `cargo check --workspace` clean - [x] `cargo test --workspace --lib --bins` green (one pre-existing `avg_finds_sum_and_count` HashMap-iteration-order flake is unchanged; passes on retry) - [x] 9 new unit tests on `BindExactAgg` covering each intent + negative cases (by-clause / Avg / approximate-Count / zero-window) + priority gate 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude Opus 4.7 (1M context) --- control_plane/src/emit/mod.rs | 8 +- control_plane/src/optimizer/rules/mod.rs | 3 + .../src/physical/colored_dag/allocator.rs | 12 + .../src/sketch_algebra/physical_expr.rs | 41 +++ .../sketch_algebra/rules/bind_exact_agg.rs | 251 ++++++++++++++++++ control_plane/src/sketch_algebra/rules/mod.rs | 7 + control_plane/src/sketch_algebra/tests.rs | 48 +++- 7 files changed, 355 insertions(+), 15 deletions(-) create mode 100644 control_plane/src/sketch_algebra/rules/bind_exact_agg.rs diff --git a/control_plane/src/emit/mod.rs b/control_plane/src/emit/mod.rs index 741231da..5a542bd8 100644 --- a/control_plane/src/emit/mod.rs +++ b/control_plane/src/emit/mod.rs @@ -257,7 +257,13 @@ pub fn extract_root_sketch_kind(expr: &PhysicalExpr) -> Option { } PhysicalExpr::Logical(_) | PhysicalExpr::Ref { .. } - | PhysicalExpr::RawAtEdgePrometheusArchive { .. } => None, + | PhysicalExpr::RawAtEdgePrometheusArchive { .. } + // ExactAgg has no sketch family — it produces an exact + // aggregation accumulator, not a sketch state. The + // routing emitter routes these to the + // `metrics/raw_passthrough` / exact-precompute pipeline + // alongside Logical pass-throughs. + | PhysicalExpr::ExactAgg { .. } => None, } } diff --git a/control_plane/src/optimizer/rules/mod.rs b/control_plane/src/optimizer/rules/mod.rs index 9f476ff0..6c3fe2ae 100644 --- a/control_plane/src/optimizer/rules/mod.rs +++ b/control_plane/src/optimizer/rules/mod.rs @@ -632,6 +632,9 @@ mod tests { PhysicalExpr::Logical(_) | PhysicalExpr::Ref { .. } => None, PhysicalExpr::RawAtEdgeSketchAtBackend { family, .. } => Some(family.clone()), PhysicalExpr::RawAtEdgePrometheusArchive { .. } => None, + // ExactAgg has no sketch family — it produces an exact + // aggregation accumulator, not a sketch state. + PhysicalExpr::ExactAgg { .. } => None, } } diff --git a/control_plane/src/physical/colored_dag/allocator.rs b/control_plane/src/physical/colored_dag/allocator.rs index 23530f38..f6d5ac04 100644 --- a/control_plane/src/physical/colored_dag/allocator.rs +++ b/control_plane/src/physical/colored_dag/allocator.rs @@ -180,6 +180,18 @@ impl ThreeStageWalker { // Prometheus's native OTLP receiver. The agent pipeline picks // this up via `asap.mode=prometheus_archive` routing. PhysicalExpr::RawAtEdgePrometheusArchive { .. } => StageId::Edge, + + // ── ExactAgg (PR-6 follow-up): same shape as SketchAgg — + // produces typed state at the edge. The accumulator runs on + // the edge precompute pipeline; the backend's + // `SketchStoreSink::append_to_index` writes the final + // (sid, window, accumulator) tuples it ships. Coloured + // Edge to match the sketch path's locality. + PhysicalExpr::ExactAgg { child, .. } => { + let (cid, _) = self.visit(child)?; + self.dag.edges.push((id, cid)); + StageId::Edge + } }; // Patch in the resolved stage now that children have been visited. diff --git a/control_plane/src/sketch_algebra/physical_expr.rs b/control_plane/src/sketch_algebra/physical_expr.rs index dce20296..3b268f91 100644 --- a/control_plane/src/sketch_algebra/physical_expr.rs +++ b/control_plane/src/sketch_algebra/physical_expr.rs @@ -22,6 +22,7 @@ use serde::{Deserialize, Serialize}; use crate::intent_algebra::QueryExpr; use crate::sketch_algebra::params::{SketchKind, SketchParams}; use crate::types_v2::BindingName; +use promql_utilities::query_logics::enums::AggregationType; /// Readout operation extracted from a built sketch state. Inverse of /// `SketchAgg`. Mirrors design.md §6 line ~607 — `SketchEstimate` plus @@ -176,6 +177,36 @@ pub enum PhysicalExpr { /// — see `deploy/configs/prometheus-otlp-receiver.yml`. label_proj: Vec, }, + + /// Exact-aggregation node — produces the exact aggregation result + /// (Sum / Count-as-Sum / Increase / MinMax / …) directly. The L4 + /// counterpart to the data-plane `AggPayload::ExactAgg` shape: + /// state at the warm-tier sid is a typed accumulator (not a + /// sketch byte buffer), and there's no separate readout step — + /// the accumulator's value IS the answer. + /// + /// Distinct from `SketchAgg` + `SketchEstimate` in two ways: + /// 1. No `EstimateOp` wrapper. ExactAgg is its own answer. + /// 2. No `SketchParams`. The `AggregationType` enum captures the + /// parameterization (DDSketch's α etc. don't apply — exact + /// aggregations are parameter-free up to the accumulator + /// family choice). + /// + /// Emitted by `bind_exact_agg` for `AggIntent::Sum` / + /// `AggIntent::Rate` / `AggIntent::Increase` / + /// `AggIntent::Count{Exact}` once the analyzer flips them (PR 6 + /// follow-up). Until that PR, the variant existed dormant in the + /// data plane's `AggKind::ExactAgg`; this brings the L4 algebra + /// in line. + ExactAgg { + /// Which exact-aggregation family (Sum / Increase / MinMax / + /// SetAggregator / DeltaSetAggregator / HLL — the same enum + /// the data plane keys on at `AggKind::ExactAgg.agg_type`). + agg_type: AggregationType, + /// Input sub-tree — typically `Logical(Window{...})` or + /// `Logical(Scan{...})`. Same shape as `SketchAgg::child`. + child: Box, + }, } impl PhysicalExpr { @@ -197,6 +228,16 @@ impl PhysicalExpr { }), } } + + /// Convenience constructor for `ExactAgg{Logical(qe)}` — the exact + /// counterpart to [`Self::estimate_over_agg`]. No estimate wrapper + /// because ExactAgg produces the answer directly. + pub fn exact_agg_over_logical(agg_type: AggregationType, logical: QueryExpr) -> Self { + PhysicalExpr::ExactAgg { + agg_type, + child: Box::new(PhysicalExpr::Logical(logical)), + } + } } // ── Tests ───────────────────────────────────────────────────────────────────── diff --git a/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs b/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs new file mode 100644 index 00000000..8c13911a --- /dev/null +++ b/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs @@ -0,0 +1,251 @@ +//! `BindExactAgg` — emits `PhysicalExpr::ExactAgg` for the four exact-aggregation +//! intents that the PR 6 follow-up flipped to warm-tier routing. +//! +//! ## Coverage +//! +//! | L3 `AggIntent` | L4 `PhysicalExpr::ExactAgg` | +//! |---|---| +//! | `Sum` | `agg_type: AggregationType::Sum` | +//! | `Rate { .. }` | `agg_type: AggregationType::Increase` | +//! | `Increase { .. }` | `agg_type: AggregationType::Increase` | +//! | `Count { accuracy: Exact }` | `agg_type: AggregationType::Sum` | +//! +//! `Rate` and `Increase` share the `Increase` accumulator because rate is +//! computed as `increase / window_seconds` — a scalar division on the +//! accumulator's output, not a separate accumulator family. The wrapping +//! division (when needed) is the L5 emitter's responsibility, not the +//! L4 binder's. +//! +//! `Count{accuracy:Exact}` maps to `Sum` because `count_over_time` is +//! exactly the sum of presence indicators (each sample contributes 1). +//! There's no dedicated `Count` `AggregationType` variant; `Sum` +//! covers the shape. +//! +//! ## What this rule does NOT bind +//! +//! - `AggIntent::Avg` — needs cross-policy join (Sum / Count), no single +//! `AggregationType` covers it. Stays on archive until the L4 binder +//! gains a join rule. +//! - `AggIntent::Quantile { Exact }` / `Cardinality { Exact }` / +//! `TopK { Exact }` / `Frequency { Exact }` — the exact-accuracy +//! variants of approximate-by-default intents. No exact-precompute +//! shape exists in `AggregationType` for these; they need +//! `HashAgg` / `SortAgg` / `SortMerge` from the deployment's +//! exact-physical-operator family (none of which run at the warm tier +//! today). +//! - `AggIntent::Min` / `AggIntent::Max` — `MinMax` is already covered +//! by the quantile-sketch path (`quantile(0)` / `quantile(1)` via +//! DDSketch / KLL). Adding a separate `MinMax` ExactAgg binding +//! would create a competing rule; not worth the cost-model churn +//! until profile data shows MinMax-precompute is meaningfully +//! cheaper than the quantile-sketch path for some workloads. +//! +//! ## Priority +//! +//! `2` — above `bind_archive_only` (which has the lowest priority for a +//! catch-all on Sum / Rate / Increase etc.) but below the sketch +//! family rules (priorities 4–6). When this rule and `bind_archive_only` +//! both match an archive-routable intent (Sum / Rate / Increase), the +//! ExactAgg path wins — same direction as the analyzer flip in +//! `capability_for`. + +#![allow(dead_code)] + +use std::time::Duration; + +use promql_utilities::query_logics::enums::AggregationType; + +use crate::intent_algebra::{AggIntent, QueryExpr}; +use crate::sketch_algebra::physical_expr::PhysicalExpr; +use crate::sketch_algebra::rules::Rule; +use crate::types_v2::AccuracyTarget; + +/// Bind exact-aggregation intents (Sum / Rate / Increase / Count{Exact}) +/// to `PhysicalExpr::ExactAgg`. See module doc for the mapping table. +pub struct BindExactAgg; + +impl Rule for BindExactAgg { + fn name(&self) -> &'static str { + "bind_exact_agg" + } + + fn priority(&self) -> u16 { + // Above bind_archive_only (priority 1) but below sketch families + // (4-6). Sum / Rate / Increase are archive-only intents in the + // sketch-family world; this rule overrides that routing. + 2 + } + + fn apply(&self, expr: &QueryExpr, _accuracy: &AccuracyTarget) -> Option { + let (intent, child) = match expr { + QueryExpr::Aggregate { + aggs, child, by, .. + } if aggs.len() == 1 && by.is_empty() => (&aggs[0], child), + _ => return None, + }; + + let agg_type = match intent { + AggIntent::Sum => AggregationType::Sum, + AggIntent::Rate { window } | AggIntent::Increase { window } => { + // The window is informational here — the data plane keys + // the policy on (metric, attrs, agg_kind, filter) plus + // the per-policy `window_size`. Empty windows are + // semantically meaningless; reject them so the rule + // doesn't fire on a malformed L3 input. + if *window == Duration::ZERO { + return None; + } + AggregationType::Increase + } + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + } => AggregationType::Sum, + _ => return None, + }; + + Some(PhysicalExpr::exact_agg_over_logical( + agg_type, + (**child).clone(), + )) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::intent_algebra::{LabelFilter, Schema, Source}; + + fn scan(metric: &str) -> QueryExpr { + QueryExpr::Scan { + source: Source::TimeSeries { + metric: metric.into(), + }, + label_filters: vec![LabelFilter { + label: "service".into(), + equals: "api".into(), + }], + schema: Schema::default(), + } + } + + fn agg_over(intent: AggIntent, metric: &str) -> QueryExpr { + QueryExpr::Aggregate { + aggs: vec![intent], + child: Box::new(scan(metric)), + by: vec![], + having: None, + } + } + + fn check_binds(intent: AggIntent, expected: AggregationType) { + let expr = agg_over(intent, "test_metric"); + let bound = BindExactAgg + .apply(&expr, &AccuracyTarget::Exact) + .unwrap_or_else(|| panic!("rule didn't fire on {expected:?}")); + match bound { + PhysicalExpr::ExactAgg { agg_type, .. } => assert_eq!(agg_type, expected), + other => panic!("expected ExactAgg, got {other:?}"), + } + } + + #[test] + fn binds_sum_to_exact_agg_sum() { + check_binds(AggIntent::Sum, AggregationType::Sum); + } + + #[test] + fn binds_rate_to_exact_agg_increase() { + check_binds( + AggIntent::Rate { + window: Duration::from_secs(60), + }, + AggregationType::Increase, + ); + } + + #[test] + fn binds_increase_to_exact_agg_increase() { + check_binds( + AggIntent::Increase { + window: Duration::from_secs(300), + }, + AggregationType::Increase, + ); + } + + #[test] + fn binds_count_exact_to_exact_agg_sum() { + check_binds( + AggIntent::Count { + accuracy: AccuracyTarget::Exact, + }, + AggregationType::Sum, + ); + } + + #[test] + fn does_not_bind_count_approximate() { + // Approximate count is the cardinality-sketch path's domain. + let expr = agg_over( + AggIntent::Count { + accuracy: AccuracyTarget::Epsilon(0.01), + }, + "test_metric", + ); + assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none()); + } + + #[test] + fn does_not_bind_avg() { + let expr = agg_over(AggIntent::Avg, "test_metric"); + assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none()); + } + + #[test] + fn does_not_bind_quantile() { + let expr = agg_over( + AggIntent::Quantile { + q: 0.99, + accuracy: AccuracyTarget::Epsilon(0.01), + }, + "test_metric", + ); + assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none()); + } + + #[test] + fn does_not_bind_when_by_clause_present() { + // Group-by-bearing intents are L3-canonical-form-only here; + // this rule mirrors the existing bind_ddsketch_quantile shape + // and rejects `by`-bearing inputs. A keyed ExactAgg follow-up + // would emit `MultipleSum` / `MultipleIncrease` instead — see + // the AggregationType enum. + let expr = QueryExpr::Aggregate { + aggs: vec![AggIntent::Sum], + child: Box::new(scan("test_metric")), + by: vec![0], + having: None, + }; + assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none()); + } + + #[test] + fn rejects_zero_window_rate() { + // Defensive — a zero-window Rate is semantically meaningless. + let expr = agg_over( + AggIntent::Rate { + window: Duration::ZERO, + }, + "test_metric", + ); + assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none()); + } + + #[test] + fn priority_above_archive_only() { + // Sanity check: this rule wins against bind_archive_only when + // both would fire on a Sum / Rate / Increase intent. + use crate::sketch_algebra::rules::bind_archive_only::BindArchiveOnly; + assert!(BindExactAgg.priority() > BindArchiveOnly.priority()); + } +} diff --git a/control_plane/src/sketch_algebra/rules/mod.rs b/control_plane/src/sketch_algebra/rules/mod.rs index 6a25f9af..4f55d388 100644 --- a/control_plane/src/sketch_algebra/rules/mod.rs +++ b/control_plane/src/sketch_algebra/rules/mod.rs @@ -27,6 +27,7 @@ pub mod bind_archive_only; pub mod bind_cms_count; pub mod bind_cms_topk; pub mod bind_ddsketch_quantile; +pub mod bind_exact_agg; pub mod bind_hll_cardinality; pub mod bind_kll_quantile; @@ -64,6 +65,12 @@ pub fn dispatch(expr: &QueryExpr, accuracy: &AccuracyTarget) -> Option assert_eq!( + agg_type, + promql_utilities::query_logics::enums::AggregationType::Sum, + "Sum should bind to ExactAgg(Sum)" + ), + other => panic!("expected ExactAgg, got {other:?}"), + } } #[test] @@ -418,12 +425,13 @@ fn phase_b_pattern_only_temporal_quantile_binds_to_sketch() { /// `ONLY_TEMPORAL` — `sum_over_time(m[5m])` (and the count/avg/min/max /// variants that legacy `single_query.rs` accepts). -/// Control plane path: `Aggregate{Sum}` over `Window` → no warm-tier rule -/// fires (no streaming sum sketch); falls through to `Logical`. The -/// existing `algebra::directory` / `algebra::physical` engine handles the -/// exact aggregate. +/// +/// Control plane path (post PR-6 follow-up): `Aggregate{Sum}` over +/// `Window` → `BindExactAgg` fires → `PhysicalExpr::ExactAgg{Sum}`. +/// Pre-follow-up this fell through to `Logical` because no rule +/// matched Sum; the L5 emitter routed it to the archive engine. #[test] -fn phase_b_pattern_only_temporal_sum_falls_through_to_logical() { +fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() { let expr = QueryExpr::Aggregate { by: vec![], aggs: vec![AggIntent::Sum], @@ -431,8 +439,13 @@ fn phase_b_pattern_only_temporal_sum_falls_through_to_logical() { child: Box::new(windowed_scan()), }; let bound = bind_query_expr(&expr, AccuracyTarget::Epsilon(0.01)).unwrap(); - // Sum is exact → no SketchAgg, just a Logical pass-through. - assert!(matches!(bound, PhysicalExpr::Logical(_))); + match bound { + PhysicalExpr::ExactAgg { agg_type, .. } => assert_eq!( + agg_type, + promql_utilities::query_logics::enums::AggregationType::Sum, + ), + other => panic!("expected ExactAgg(Sum), got {other:?}"), + } } /// `ONLY_SPATIAL` — `sum by (host) (m)`. @@ -567,6 +580,9 @@ fn collect_sketch_kinds(expr: &PhysicalExpr) -> Vec { walk(child, out); } PhysicalExpr::RawAtEdgePrometheusArchive { .. } => {} + // ExactAgg has no SketchKind to collect; its child may + // carry one transitively (rare but possible if nested). + PhysicalExpr::ExactAgg { child, .. } => walk(child, out), } } walk(expr, &mut out); @@ -597,6 +613,10 @@ fn binding_is_archive(expr: &PhysicalExpr) -> bool { // about cold-tier scan-vs-warm-tier-sketch decisions, not Mode 3. PhysicalExpr::RawAtEdgeSketchAtBackend { child, .. } => binding_is_archive(child), PhysicalExpr::RawAtEdgePrometheusArchive { .. } => false, + // ExactAgg is a warm-tier exact-aggregation accumulator, NOT + // an archive route. The L5 emitter writes the result through + // the precompute output sink, same path as SketchAgg. + PhysicalExpr::ExactAgg { .. } => false, } }