From b1e81d7e6411ff4d1fbd1417cf91d22828b02b12 Mon Sep 17 00:00:00 2001 From: zz_y Date: Fri, 17 Jul 2026 22:08:39 -0600 Subject: [PATCH] feat(control_plane): Phase 1 -- merge AggIntent vocabulary from ASAPController Adopts ASAPController's current AggIntent as the base (~40 variants vs control_plane's pre-merge 25), per docs/migration-plan-backend-plan.md Phase 1. Two deliberate deviations from a byte-for-byte port, both documented in agg_intent.rs's module docs: - `window: Duration` stays on the time-series-derivative variants instead of moving to QueryExpr::TimeRange -- that requires verifying query_expr.rs/lower.rs thread it correctly everywhere, which is Phase 2 scope. - `AggIntent::Frequency` is kept, not folded into ASAPController's RankingMeasure::Frequency -- they're unrelated concepts that share a name (standalone point-frequency-via-CMS query vs. a TopK ranking-measure classifier). This corrects the decision recorded in PR #389/#390; RankingMeasure is still adopted additively. Also folds `irate` into `Rate` per the agreed tie-break (ASAPController's approach): `AggIntent::Irate` is removed. This was already a no-op on real query behavior -- the PromQL walk already lowered "rate"|"irate" to the same AggFunc::Rate with no AggFunc::Irate ever existing, so AggIntent::Irate was unreachable from real parsing; only unit tests constructed it directly. New intent families added (histogram accessors, math/trig transforms, time/calendar accessors, presence functions, Group/CountValues, extended range-vector reducers) are all archive-only for now -- none has a Bind* rule yet, consistent with existing policy for intents with no ASAP-tier sketch binding. Renames to match ASAPController: Idelta -> IDelta, HoltWinters -> DoubleExpSmoothing, Absent -> {Absent, AbsentOverTime}, Present -> PresentOverTime. Adds col: Option to the single-column reducers (Sum/Min/Max/Avg/Quantile/Cardinality/StdDev/Variance), defaulting to None everywhere today (PromQL sample-value convention) -- plumbing for the Phase 0 "control_plane gains SQL support" decision, populated once frontend-sql lands in Phase 2. Verified: full workspace builds clean (cargo build --workspace); control_plane's 837-test suite passes unchanged (1 pre-existing failure, confirmed identical on unmodified main via git stash, unrelated to this change). The analyzer-parity-matrix.md corpus this phase's testing bar was meant to gate on no longer exists under its documented name (data_plane/.../engine.rs::analyzer_parity_tests) -- flagging for Phase 3, since that phase's plan explicitly depends on it as the acceptance test for the capability_for() fix. --- control_plane/src/asap_tier_analysis.rs | 36 +- .../src/intent_algebra/agg_intent.rs | 885 +++++++++++++----- .../src/intent_algebra/column_resolution.rs | 6 +- control_plane/src/intent_algebra/cse.rs | 7 +- control_plane/src/intent_algebra/lower.rs | 13 +- .../src/intent_algebra/query_expr.rs | 8 +- .../src/intent_algebra/relational.rs | 16 +- control_plane/src/optimizer/cost/mod.rs | 8 +- control_plane/src/optimizer/rules/mod.rs | 2 + control_plane/src/physical/allocator.rs | 44 +- control_plane/src/physical/planner.rs | 3 +- control_plane/src/physical/sketch_catalog.rs | 6 +- control_plane/src/physical/window_fusion.rs | 4 +- control_plane/src/query_parser/mod.rs | 4 +- control_plane/src/query_parser/promql.rs | 4 +- .../src/sketch_algebra/capability.rs | 83 +- .../src/sketch_algebra/physical_expr.rs | 1 + .../sketch_algebra/rules/bind_archive_only.rs | 22 +- .../rules/bind_ddsketch_quantile.rs | 2 +- .../sketch_algebra/rules/bind_exact_agg.rs | 11 +- .../rules/bind_hll_cardinality.rs | 2 +- .../sketch_algebra/rules/bind_kll_quantile.rs | 2 +- control_plane/src/sketch_algebra/tests.rs | 40 +- 23 files changed, 845 insertions(+), 364 deletions(-) diff --git a/control_plane/src/asap_tier_analysis.rs b/control_plane/src/asap_tier_analysis.rs index df2e861e..ea3079de 100644 --- a/control_plane/src/asap_tier_analysis.rs +++ b/control_plane/src/asap_tier_analysis.rs @@ -355,10 +355,12 @@ fn collect_agg_intents(expr: &QueryExpr, out: &mut Vec) { fn intent_kind_label(intent: &AggIntent) -> &'static str { match intent { AggIntent::Count { .. } => "count", - AggIntent::Sum => "sum", - AggIntent::Min => "min", - AggIntent::Max => "max", - AggIntent::Avg => "avg", + AggIntent::Sum { .. } => "sum", + AggIntent::Min { .. } => "min", + AggIntent::Max { .. } => "max", + AggIntent::Avg { .. } => "avg", + AggIntent::StdDev { .. } => "stddev", + AggIntent::Variance { .. } => "variance", AggIntent::Quantile { .. } => "quantile", AggIntent::TopK { .. } => "topk", AggIntent::Cardinality { .. } => "cardinality", @@ -366,15 +368,33 @@ fn intent_kind_label(intent: &AggIntent) -> &'static str { AggIntent::Rate { .. } => "rate", AggIntent::Increase { .. } => "increase", AggIntent::Absent => "absent", - AggIntent::Present => "present", + AggIntent::AbsentOverTime => "absent_over_time", + AggIntent::PresentOverTime => "present_over_time", AggIntent::Delta { .. } => "delta", AggIntent::Deriv { .. } => "deriv", AggIntent::PredictLinear { .. } => "predict_linear", - AggIntent::HoltWinters { .. } => "holt_winters", - AggIntent::Idelta { .. } => "idelta", - AggIntent::Irate { .. } => "irate", + AggIntent::DoubleExpSmoothing { .. } => "double_exponential_smoothing", + AggIntent::IDelta { .. } => "idelta", AggIntent::Resets { .. } => "resets", AggIntent::Changes { .. } => "changes", + AggIntent::HistogramCount => "histogram_count", + AggIntent::HistogramSum => "histogram_sum", + AggIntent::HistogramAvg => "histogram_avg", + AggIntent::HistogramStdDev => "histogram_stddev", + AggIntent::HistogramStdVar => "histogram_stdvar", + AggIntent::HistogramFraction { .. } => "histogram_fraction", + AggIntent::HistogramQuantile { .. } => "histogram_quantile", + AggIntent::Math(_) => "math", + AggIntent::TimeFn(_) => "time_fn", + AggIntent::Group => "group", + AggIntent::CountValues { .. } => "count_values", + AggIntent::LastOverTime => "last_over_time", + AggIntent::FirstOverTime => "first_over_time", + AggIntent::MadOverTime => "mad_over_time", + AggIntent::TsOfMinOverTime => "ts_of_min_over_time", + AggIntent::TsOfMaxOverTime => "ts_of_max_over_time", + AggIntent::TsOfFirstOverTime => "ts_of_first_over_time", + AggIntent::TsOfLastOverTime => "ts_of_last_over_time", } } diff --git a/control_plane/src/intent_algebra/agg_intent.rs b/control_plane/src/intent_algebra/agg_intent.rs index 81c55fef..7aa87215 100644 --- a/control_plane/src/intent_algebra/agg_intent.rs +++ b/control_plane/src/intent_algebra/agg_intent.rs @@ -18,9 +18,48 @@ //! same regardless. PromQL `quantile_over_time(0.99, m[5m])` lowers to //! `Window{size=5m} → Aggregate{aggs:[Quantile{q=0.99}]}`. //! -//! `Rate` and `Increase` survive that argument because they include -//! PromQL's counter-reset adjustment, a non-trivial transformation that -//! exact `Sum` does not perform. +//! ## Phase 1 IR merge (ASAPController base) +//! +//! Per `control_plane/docs/migration-plan-backend-plan.md` Phase 1: this +//! file adopts ASAPController's current `AggIntent` vocabulary as the base +//! (its ~40-variant set is now richer than this file's pre-merge 25 +//! variants — see `control_plane/docs/design-backend-plan-wire-format.md` +//! §4). Two deliberate deviations from a byte-for-byte port, both +//! recorded in the migration plan's decision log: +//! +//! 1. **`window: Duration` stays on the time-series-derivative variants** +//! (`Rate` / `Increase` / `Changes` / `Delta` / `IDelta` / `Deriv` / +//! `Resets` / `PredictLinear` / `DoubleExpSmoothing`), instead of +//! ASAPController's design of threading the window only through the +//! enclosing `QueryExpr::TimeRange` node. Moving the window off the +//! intent requires verifying `query_expr.rs`/`lower.rs` thread it +//! correctly everywhere `AggIntent` is currently constructed or +//! consulted — that's Phase 2 scope (those are Phase 2 files), not +//! this one. Reconcile in Phase 2 once `query_expr.rs` lands. +//! 2. **`AggIntent::Frequency { accuracy }` is kept**, not folded into +//! ASAPController's `RankingMeasure::Frequency`. These are different +//! concepts that happen to share a name: control_plane's `Frequency` +//! is a standalone, independently-bindable point-frequency query +//! (`count(*) WHERE key = k`, bound to CMS in +//! `sketch_algebra/rules/bind_cms_count.rs`); ASAPController's +//! `RankingMeasure::Frequency` classifies what a `TopK` ranks by, and +//! has no relation to point-frequency queries at all. This is exactly +//! the "genuinely control_plane-only, no ASAPController equivalent" +//! exception to the merge's tie-break rule. `RankingMeasure` is still +//! adopted below, additively — it's a real capability control_plane +//! lacked (nothing today classifies whether a `TopK` is a sketchable +//! heavy-hitter), it just isn't a replacement for `Frequency`. +//! +//! `irate`/`rate` fold: per the decided tie-break, `AggIntent::Irate` is +//! removed; `irate(...)` and `rate(...)` both lower to `AggIntent::Rate` +//! (unchanged from today — the PromQL L1→L2 walk already maps `"rate" | +//! "irate" => AggFunc::Rate`, and no `AggFunc::Irate` variant exists, so +//! `AggIntent::Irate` was already unreachable from real query parsing; +//! only unit tests constructed it directly. The estimation-method +//! distinction (windowed-average vs last-two-samples) is deferred to L4, +//! per ASAPController's design — no L4 rule makes that distinction today, +//! which is fine, since nothing reachable exercised it before this change +//! either). #![allow(dead_code)] @@ -28,15 +67,20 @@ use std::time::Duration; use serde::{Deserialize, Serialize}; -use crate::intent_algebra::schema::{Column, DataType}; +use crate::intent_algebra::schema::{Column, ColumnId, DataType}; use crate::types_v2::AccuracyTarget; -/// "What to compute" at L3 — vocabulary the planner pivots on. See module -/// doc for the intent vs operator distinction. +/// "What to compute" at L3 — the vocabulary the planner pivots on. /// -/// Variants intentionally mirror `design.md` §6 line ~468; data-model- -/// agnostic intents come first, time-series-streaming derivatives -/// (`Rate` / `Increase`) come last. +/// The single-column reducers (`Sum` / `Min` / `Max` / `Avg` / `StdDev` / +/// `Variance` / `Quantile` / `Cardinality`) carry `col: Option` — +/// the positional input column they reduce. `None` is the PromQL +/// convention "the time-series sample value"; SQL `SUM(bytes), +/// AVG(latency)` sets distinct `Some(id)`s. `TopK`'s grouping rides on the +/// enclosing `QueryExpr::Aggregate.by`, like every other aggregate; the +/// intent itself carries only `k` + the accuracy target, no `col` (it +/// ranks by the aggregate output, not a base column — see +/// [`RankingMeasure`] below). #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum AggIntent { @@ -47,17 +91,49 @@ pub enum AggIntent { Count { accuracy: AccuracyTarget }, /// SUM(col). Always exact at L3 — no approximation intent for `Sum` /// in the catalog (`design.md` §6 line ~485). - Sum, + Sum { + #[serde(default)] + col: Option, + }, /// Per-group minimum — exact at L3. - Min, + Min { + #[serde(default)] + col: Option, + }, /// Per-group maximum — exact at L3. - Max, + Max { + #[serde(default)] + col: Option, + }, /// Arithmetic mean. Exact at L3; sketch backends fold this onto a /// `Quantile{q=0.5}` only when the cost model allows the relaxation. - Avg, + Avg { + #[serde(default)] + col: Option, + }, + /// Sample standard deviation when `population == false`; population + /// stddev otherwise. PromQL `stddev` / `stddev_over_time`; SQL + /// `STDDEV(col)`. + StdDev { + #[serde(default)] + col: Option, + population: bool, + }, + /// Variance — PromQL `stdvar` / `stdvar_over_time`; SQL + /// `VARIANCE(col)`. + Variance { + #[serde(default)] + col: Option, + population: bool, + }, /// Compute the φ-th quantile (0 ≤ q ≤ 1) to the given accuracy. /// Sketch families: KLL, DDSketch, t-digest. - Quantile { q: f64, accuracy: AccuracyTarget }, + Quantile { + #[serde(default)] + col: Option, + q: f64, + accuracy: AccuracyTarget, + }, /// Heavy-hitter top-k. Distinct from generic `Sort + Limit` because /// a dedicated sketch primitive (SpaceSaving, CMS-with-heap, /// Misra-Gries) computes it as a single operation. L1→L2→L3 lowering @@ -66,70 +142,201 @@ pub enum AggIntent { TopK { k: usize, accuracy: AccuracyTarget }, /// COUNT DISTINCT — number of distinct values in the input column, /// to the given accuracy. Sketch families: HLL, theta-sketch. - Cardinality { accuracy: AccuracyTarget }, - /// Frequency of a key in the input — `count(*) WHERE key = k` modeled - /// as a sketch query. Sketch families: CMS, count-min-log. + Cardinality { + #[serde(default)] + col: Option, + accuracy: AccuracyTarget, + }, + /// control_plane-only (no ASAPController equivalent — see module docs). + /// Point-frequency of a key in the input — `count(*) WHERE key = k` + /// modeled as a sketch query. Sketch families: CMS, count-min-log. + /// **Not** the same concept as [`RankingMeasure::Frequency`] below, + /// which classifies what a `TopK` ranks by. Frequency { accuracy: AccuracyTarget }, // ── Time-series streaming derivatives ──────────────────────────────── /// Per-second average derivative with PromQL's counter-reset - /// adjustment. Specialized — exact `Sum / Count over Window` does - /// NOT serve this intent. + /// adjustment. Also serves `irate(...)` post the rate/irate fold (see + /// module docs) — the windowed-average vs last-two-samples estimation + /// choice is an L4 concern, not encoded here. Rate { window: Duration }, /// Cumulative increase over the given window with counter-reset /// adjustment. Specialized — see `Rate` above. Increase { window: Duration }, - // ── Archive-only intents (Phase β migration) ───────────────────────── - // Intents below have no ASAP-tier sketch family today; the L4 binder - // emits a `PhysicalExpr::Logical` pass-through and the L5 emitter routes - // them to the cold archive tier (Gorilla / Thanos). Adding a streaming - // sketch family for any of these is a follow-up — the L3 vocabulary - // captures the intent so the routing decision is layered above intent. - // - // Note: `histogram_quantile(φ, …)` is NOT an L3 intent — it's a PromQL - // /MetricsQL language-level operator. Per Step γ5 of the relational - // migration, the PromQL parser substitutes it directly into a plain - // `Aggregate { Quantile(φ) }` (which lowers to - // `AggIntent::Quantile { q, accuracy }`); bucket-aware handling is a - // physical-planner concern, not an L3 intent. - // - /// `absent(vector_selector)` — 1 iff the selector matched no series in - /// the evaluation window, no value otherwise. Routed to archive: the - /// engine answers it directly off the index. - Absent, - /// `present_over_time(m[range])` — 1 iff the selector had at least one - /// sample in the window. Inverse of `Absent`. Archive-routed. - Present, - /// `delta(m[range])` — last − first sample within the window, NO - /// counter-reset adjustment. Distinct from [`AggIntent::Increase`]. + // ── Counter-derivative / range-vector functions ────────────────────── + // All per-series, label-preserving reductions of a single series' + // range window to one value. Each has distinct semantics and is + // deliberately NOT aliased to `Rate`/`Increase`/`Count`. + /// `changes(m[range])` — number of times the value changed in the + /// window. + Changes { window: Duration }, + /// `delta(m[range])` — difference between the first and last sample + /// (gauge semantics; not counter-reset-adjusted). Distinct from + /// [`AggIntent::Increase`]. Delta { window: Duration }, + /// `idelta(m[range])` — difference between the last two samples. + IDelta { window: Duration }, /// `deriv(m[range])` — per-second derivative via simple linear - /// regression. Archive-routed (no streaming sketch). + /// regression over the window (gauges). Deriv { window: Duration }, - /// `predict_linear(m[range], t)` — linear-regression prediction `t` - /// seconds into the future. Archive-routed. - PredictLinear { window: Duration, ahead: Duration }, - /// `holt_winters(m[range], sf, tf)` — exponential-smoothing forecast. - /// Archive-routed. - HoltWinters { - window: Duration, - smoothing_factor: f64, - trend_factor: f64, - }, - /// `idelta(m[range])` — `last − second_to_last`, instant delta. No - /// streaming sketch. - Idelta { window: Duration }, - /// `irate(m[range])` — instant per-second rate computed from the last - /// two samples. Counter-reset adjusted but evaluated point-wise; not - /// the same as the streaming [`AggIntent::Rate`]. - Irate { window: Duration }, /// `resets(m[range])` — count of counter resets over the window. - /// Archive-routed. Resets { window: Duration }, - /// `changes(m[range])` — count of value changes over the window. - /// Archive-routed. - Changes { window: Duration }, + /// `predict_linear(m[range], t)` — linear-regression extrapolation of + /// the value `t` seconds into the future. + PredictLinear { + window: Duration, + /// The prediction horizon in seconds (the 2nd, scalar argument). + seconds: f64, + }, + /// `double_exponential_smoothing(v[w], sf, tf)` (a.k.a. the legacy + /// `holt_winters`) — Holt-Winters double-exponential smoothing. + DoubleExpSmoothing { + window: Duration, + /// Data (level) smoothing factor `sf` ∈ (0, 1). + smoothing: f64, + /// Trend smoothing factor `tf` ∈ (0, 1). + trend: f64, + }, + + // ── Native-histogram accessors ──────────────────────────────────────── + // Per-series extractions from a native-histogram instant vector — one + // float per series, label-preserving. (Classic `le`-bucket + // `histogram_quantile` stays a `Quantile` over the bucketed vector — + // per Step γ5, the PromQL parser substitutes it directly into a plain + // `Aggregate { Quantile(φ) }`; bucket-aware handling is a + // physical-planner concern, not this L3 intent.) + /// `histogram_count(v)` — observation count of each native histogram. + HistogramCount, + /// `histogram_sum(v)` — sum of observations. + HistogramSum, + /// `histogram_avg(v)` — mean (`sum/count`). + HistogramAvg, + /// `histogram_stddev(v)` — standard deviation of observations. + HistogramStdDev, + /// `histogram_stdvar(v)` — variance of observations. + HistogramStdVar, + /// `histogram_fraction(lower, upper, v)` — fraction of observations in + /// `[lower, upper]`. + HistogramFraction { lower: f64, upper: f64 }, + /// `histogram_quantile(φ, )` over a *native* + /// histogram — exact bucket interpolation, not a sketch-able + /// quantile; distinct from [`AggIntent::Quantile`]. + HistogramQuantile { q: f64 }, + + /// A per-sample element-wise math / trig transform — `abs`, `ceil`, + /// `sqrt`, `ln`, `clamp_max`, the trig family, … Label-preserving. + Math(MathFunc), + + // ── Presence functions ──────────────────────────────────────────────── + /// `absent(v)` — a 1-sample vector when the instant vector `v` has no + /// matching series, else empty. + Absent, + /// `absent_over_time(v[w])` — `absent` over a range vector. + AbsentOverTime, + /// `present_over_time(v[w])` — value 1 per series that has any sample + /// in the range (per-series). + PresentOverTime, + + /// A time / calendar accessor — `timestamp`, `minute`, `hour`, + /// `day_of_week`, … over each sample's timestamp. Label-preserving. + TimeFn(TimeFunc), + + // ── Extended aggregation operators ──────────────────────────────────── + /// `group(v)` — a constant `1` per group ("group presence"). The + /// grouping keys ride on the enclosing `Aggregate.by`. Deliberately + /// NOT aliased to `Sum`/`Count`: the output value is always 1, + /// independent of the input values. + Group, + /// `count_values("l", v)` — group the input series by their sample + /// *value* and count each distinct value, emitting that value as a + /// new label `l`. Unlike every other reducer this adds a synthesized + /// `Utf8` label column, so `Aggregate` schema derivation special-cases + /// it (two output columns, not one). + CountValues { label: String }, + + // ── Additional range-vector reducers ────────────────────────────────── + // All per-series, label-preserving reductions of a single series' + // range window to one value. + /// `last_over_time(v[w])` — the most recent sample in the window. + LastOverTime, + /// `first_over_time(v[w])` — the oldest sample in the window. + FirstOverTime, + /// `mad_over_time(v[w])` — median absolute deviation over the window. + MadOverTime, + /// `ts_of_min_over_time(v[w])` — timestamp of the minimum sample. + TsOfMinOverTime, + /// `ts_of_max_over_time(v[w])` — timestamp of the maximum sample. + TsOfMaxOverTime, + /// `ts_of_first_over_time(v[w])` — timestamp of the first sample. + TsOfFirstOverTime, + /// `ts_of_last_over_time(v[w])` — timestamp of the last sample. + TsOfLastOverTime, +} + +/// Time / calendar accessor functions, evaluated over a timestamp. +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "fn", rename_all = "snake_case")] +pub enum TimeFunc { + /// `timestamp(v)` — the sample's own timestamp as a value. + Timestamp, + Minute, + Hour, + DayOfWeek, + DayOfMonth, + DayOfYear, + Month, + Year, + DaysInMonth, +} + +/// The element-wise math / trig functions. Unary over the sample value +/// unless a variant carries scalar params (`clamp*`, `round`). +#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] +#[serde(tag = "fn", rename_all = "snake_case")] +pub enum MathFunc { + Abs, + Ceil, + Floor, + Exp, + Ln, + Log2, + Log10, + Sqrt, + Sgn, + Sin, + Cos, + Tan, + Asin, + Acos, + Atan, + Sinh, + Cosh, + Tanh, + Asinh, + Acosh, + Atanh, + /// `deg(v)` — radians → degrees. + Deg, + /// `rad(v)` — degrees → radians. + Rad, + /// `round(v, to_nearest)` — nearest multiple of `to_nearest` (default 1). + Round { + to_nearest: f64, + }, + /// `clamp(v, min, max)`. + Clamp { + min: f64, + max: f64, + }, + /// `clamp_min(v, min)`. + ClampMin { + min: f64, + }, + /// `clamp_max(v, max)`. + ClampMax { + max: f64, + }, } impl AggIntent { @@ -137,162 +344,176 @@ impl AggIntent { /// today. `false` means a `Bind*` rule may match. `true` means the /// L5 emitter routes the intent to the cold-store / archive tier. /// - /// Per Phase β orchestrator spec, the PromQL functions that were - /// previously refused outright by `asap-planner-rs::single_query:: - /// is_supported()` (everything outside the 5 patterns) all return - /// `true` here. Adding a sketch family for any of them is a future - /// PR — flipping the flag to `false` is the single point of change. - /// - /// Note: `histogram_quantile(...)` was previously listed as - /// archive-only here but is no longer an `AggIntent` variant — - /// it's a PromQL/MetricsQL language-level operator. Per Step γ5, - /// the PromQL parser substitutes it into a plain - /// `Aggregate { Quantile(φ) }`, which lowers to - /// `AggIntent::Quantile { q, .. }`. + /// Every intent newly added by the Phase 1 IR merge (histogram + /// accessors, math/trig, time accessors, presence functions, `Group` + /// / `CountValues`, the extended range-vector reducers) is + /// archive-only today — none has a `Bind*` rule yet. Adding a sketch + /// family for any of them is a future PR; flipping the flag here is + /// the single point of change (same policy as before the merge). pub fn archive_only(&self) -> bool { matches!( self, AggIntent::Absent - | AggIntent::Present + | AggIntent::AbsentOverTime + | AggIntent::PresentOverTime | AggIntent::Delta { .. } | AggIntent::Deriv { .. } | AggIntent::PredictLinear { .. } - | AggIntent::HoltWinters { .. } - | AggIntent::Idelta { .. } - | AggIntent::Irate { .. } + | AggIntent::DoubleExpSmoothing { .. } + | AggIntent::IDelta { .. } | AggIntent::Resets { .. } | AggIntent::Changes { .. } + | AggIntent::HistogramCount + | AggIntent::HistogramSum + | AggIntent::HistogramAvg + | AggIntent::HistogramStdDev + | AggIntent::HistogramStdVar + | AggIntent::HistogramFraction { .. } + | AggIntent::HistogramQuantile { .. } + | AggIntent::Math(_) + | AggIntent::TimeFn(_) + | AggIntent::Group + | AggIntent::CountValues { .. } + | AggIntent::LastOverTime + | AggIntent::FirstOverTime + | AggIntent::MadOverTime + | AggIntent::TsOfMinOverTime + | AggIntent::TsOfMaxOverTime + | AggIntent::TsOfFirstOverTime + | AggIntent::TsOfLastOverTime ) } -} -impl AggIntent { + /// Whether this is a *per-series* reduction — it reduces a single + /// series' samples over its range window (one value out per series), + /// so it does **not** collapse across series and every label column + /// is preserved. (Cross-series reductions like `sum`/`avg` over a + /// series set, and control_plane's point-query `Frequency`, return + /// `false`.) + pub fn is_per_series(&self) -> bool { + matches!( + self, + Self::Rate { .. } + | Self::Increase { .. } + | Self::Changes { .. } + | Self::Delta { .. } + | Self::IDelta { .. } + | Self::Deriv { .. } + | Self::Resets { .. } + | Self::PredictLinear { .. } + | Self::DoubleExpSmoothing { .. } + | Self::HistogramCount + | Self::HistogramSum + | Self::HistogramAvg + | Self::HistogramStdDev + | Self::HistogramStdVar + | Self::HistogramFraction { .. } + | Self::Math(_) + | Self::Absent + | Self::AbsentOverTime + | Self::PresentOverTime + | Self::TimeFn(_) + | Self::LastOverTime + | Self::FirstOverTime + | Self::MadOverTime + | Self::TsOfMinOverTime + | Self::TsOfMaxOverTime + | Self::TsOfFirstOverTime + | Self::TsOfLastOverTime + ) + } + + /// The positional input column this intent reduces, if it carries + /// one. `None` = the synthetic time-series sample value (PromQL), an + /// argument-less aggregate (`Count` / `TopK`), or control_plane's + /// point-query `Frequency` (keyed, not column-reducing). + pub fn input_col(&self) -> Option { + match self { + AggIntent::Sum { col } + | AggIntent::Min { col } + | AggIntent::Max { col } + | AggIntent::Avg { col } + | AggIntent::Quantile { col, .. } + | AggIntent::Cardinality { col, .. } + | AggIntent::StdDev { col, .. } + | AggIntent::Variance { col, .. } => *col, + _ => None, + } + } + /// Output column name + type produced by this intent when applied to - /// `input`. Used by `QueryExpr::Aggregate`'s schema-derivation rule - /// (`design.md` §6 schema-flow table: "one new column per entry in - /// `aggs`, each named and typed by `AggIntent::output_type(input_field)`"). + /// `input`. Used by `QueryExpr::Aggregate`'s schema-derivation rule. /// /// PromQL convention: aggregate column name = intent kind (`count`, /// `quantile_0_99`, …) so consumers can locate it without an alias /// lookup. pub fn output_column(&self, input: &Column) -> Column { match self { - AggIntent::Count { .. } => Column { - name: "count".into(), - dtype: DataType::Int64, - nullable: false, - }, - AggIntent::Sum => Column { - name: "sum".into(), - dtype: input.dtype.clone(), - nullable: false, - }, - AggIntent::Min => Column { - name: "min".into(), - dtype: input.dtype.clone(), - nullable: input.nullable, - }, - AggIntent::Max => Column { - name: "max".into(), - dtype: input.dtype.clone(), - nullable: input.nullable, - }, - AggIntent::Avg => Column { - name: "avg".into(), - dtype: DataType::Float64, - nullable: false, - }, - AggIntent::Quantile { q, .. } => Column { - name: format!("quantile_{}", quantile_suffix(*q)), - dtype: DataType::Float64, - nullable: false, - }, - AggIntent::TopK { k, .. } => Column { - name: format!("topk_{k}"), - // TopK output is a struct/list per row; modeled as Utf8 - // for L3 (the L4 sketch-bound IR upgrades the dtype). - dtype: DataType::Utf8, - nullable: false, - }, - AggIntent::Cardinality { .. } => Column { - name: "cardinality".into(), - dtype: DataType::Int64, - nullable: false, - }, - AggIntent::Frequency { .. } => Column { - name: "frequency".into(), - dtype: DataType::Int64, - nullable: false, - }, - AggIntent::Rate { .. } => Column { - name: "rate".into(), - dtype: DataType::Float64, - nullable: false, - }, - AggIntent::Increase { .. } => Column { - name: "increase".into(), - dtype: DataType::Float64, - nullable: false, - }, - // ── Archive-only intents (Phase β) ──────────────────────────── - // Each carries a stable column name keyed on the intent kind so - // the StreamingConfig emitter and Phase α routing entry can - // locate them. All are Float64 except the boolean Absent / - // Present, which surface as Int64 (1 / 0) per PromQL convention. - AggIntent::Absent => Column { - name: "absent".into(), - dtype: DataType::Int64, - nullable: false, - }, - AggIntent::Present => Column { - name: "present".into(), - dtype: DataType::Int64, - nullable: false, - }, - AggIntent::Delta { .. } => Column { - name: "delta".into(), - dtype: DataType::Float64, - nullable: false, - }, - AggIntent::Deriv { .. } => Column { - name: "deriv".into(), - dtype: DataType::Float64, - nullable: false, - }, - AggIntent::PredictLinear { .. } => Column { - name: "predict_linear".into(), - dtype: DataType::Float64, - nullable: false, - }, - AggIntent::HoltWinters { .. } => Column { - name: "holt_winters".into(), - dtype: DataType::Float64, - nullable: false, - }, - AggIntent::Idelta { .. } => Column { - name: "idelta".into(), - dtype: DataType::Float64, - nullable: false, - }, - AggIntent::Irate { .. } => Column { - name: "irate".into(), - dtype: DataType::Float64, - nullable: false, - }, - AggIntent::Resets { .. } => Column { - name: "resets".into(), - dtype: DataType::Int64, - nullable: false, - }, - AggIntent::Changes { .. } => Column { - name: "changes".into(), - dtype: DataType::Int64, - nullable: false, - }, + AggIntent::Count { .. } => col("count", DataType::Int64, false), + AggIntent::Sum { .. } => col("sum", input.dtype.clone(), false), + AggIntent::Min { .. } => col("min", input.dtype.clone(), input.nullable), + AggIntent::Max { .. } => col("max", input.dtype.clone(), input.nullable), + AggIntent::Avg { .. } => col("avg", DataType::Float64, false), + AggIntent::StdDev { .. } => col("stddev", DataType::Float64, false), + AggIntent::Variance { .. } => col("variance", DataType::Float64, false), + AggIntent::Quantile { q, .. } => col( + &format!("quantile_{}", quantile_suffix(*q)), + DataType::Float64, + false, + ), + // TopK output is a struct/list per row; modeled as Utf8 for + // L3 (the L4 sketch-bound IR upgrades the dtype). + AggIntent::TopK { k, .. } => col(&format!("topk_{k}"), DataType::Utf8, false), + AggIntent::Cardinality { .. } => col("cardinality", DataType::Int64, false), + AggIntent::Frequency { .. } => col("frequency", DataType::Int64, false), + AggIntent::Rate { .. } => col("rate", DataType::Float64, false), + AggIntent::Increase { .. } => col("increase", DataType::Float64, false), + AggIntent::Changes { .. } => col("changes", DataType::Int64, false), + AggIntent::Delta { .. } => col("delta", DataType::Float64, false), + AggIntent::IDelta { .. } => col("idelta", DataType::Float64, false), + AggIntent::Deriv { .. } => col("deriv", DataType::Float64, false), + AggIntent::Resets { .. } => col("resets", DataType::Int64, false), + AggIntent::PredictLinear { .. } => col("predict_linear", DataType::Float64, false), + AggIntent::DoubleExpSmoothing { .. } => { + col("double_exponential_smoothing", DataType::Float64, false) + } + AggIntent::HistogramCount => col("histogram_count", DataType::Float64, false), + AggIntent::HistogramSum => col("histogram_sum", DataType::Float64, false), + AggIntent::HistogramAvg => col("histogram_avg", DataType::Float64, false), + AggIntent::HistogramStdDev => col("histogram_stddev", DataType::Float64, false), + AggIntent::HistogramStdVar => col("histogram_stdvar", DataType::Float64, false), + AggIntent::HistogramFraction { .. } => { + col("histogram_fraction", DataType::Float64, false) + } + AggIntent::HistogramQuantile { .. } => { + col("histogram_quantile", DataType::Float64, false) + } + AggIntent::Math(_) => col("value", DataType::Float64, false), + AggIntent::Absent => col("absent", DataType::Int64, false), + AggIntent::AbsentOverTime => col("absent", DataType::Int64, false), + AggIntent::PresentOverTime => col("present", DataType::Int64, false), + AggIntent::TimeFn(_) => col("value", DataType::Float64, false), + AggIntent::Group => col("group", DataType::Float64, false), + AggIntent::CountValues { .. } => col("count", DataType::Int64, false), + AggIntent::LastOverTime => col("last_over_time", DataType::Float64, false), + AggIntent::FirstOverTime => col("first_over_time", DataType::Float64, false), + AggIntent::MadOverTime => col("mad_over_time", DataType::Float64, false), + AggIntent::TsOfMinOverTime => col("ts_of_min_over_time", DataType::Float64, false), + AggIntent::TsOfMaxOverTime => col("ts_of_max_over_time", DataType::Float64, false), + AggIntent::TsOfFirstOverTime => col("ts_of_first_over_time", DataType::Float64, false), + AggIntent::TsOfLastOverTime => col("ts_of_last_over_time", DataType::Float64, false), } } } +fn col(name: &str, dtype: DataType, nullable: bool) -> Column { + Column { + name: name.into(), + dtype, + nullable, + } +} + /// `0.99` → `"0_99"`, `0.5` → `"0_5"`. Used by `Quantile` output naming /// so `quantile_0_99` is a valid identifier downstream. fn quantile_suffix(q: f64) -> String { @@ -310,19 +531,85 @@ fn quantile_suffix(q: f64) -> String { // re-exports them during the legacy-IR retirement; consumers migrate to // `intent_algebra::*` paths and the re-exports drop with `relational`. +/// What a top-k ranks its groups by — the axis that decides whether the +/// ranking is a sketchable **heavy-hitter** or a generic order-by-value +/// `Sort + Limit`. Adopted from ASAPController (Phase 1 IR merge) — this +/// is a real capability control_plane lacked before this merge, not a +/// replacement for [`AggIntent::Frequency`] (see module docs). +/// +/// Sketchability follows the *additivity* of the ranking measure, not +/// "count" per se: an additive per-key aggregate admits a single-pass +/// heavy-hitter sketch (CMS-with-heap / SpaceSaving), a non-additive one +/// does not. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum RankingMeasure { + /// Unweighted frequency — `count` of rows/samples per key. Additive, + /// and the one heavy-hitter measure **realised today** (→ + /// `AggIntent::TopK`). + Frequency, + /// Weighted frequency — an additive `sum` of a per-row weight per + /// key. Sketchable in principle (weighted SpaceSaving), but no + /// weighted heavy-hitter sketch is realised yet, so a `sum`-ranked + /// top-k currently stays a generic `Sort + Limit`. + WeightedSum, + /// A non-additive measure (`avg` / `quantile` / `min` / `max`) or a + /// raw, un-aggregated value. Never a heavy-hitter — always generic. + NonAdditive, +} + +impl RankingMeasure { + /// Whether a top-k ranked by this measure can be a heavy-hitter + /// **with a sketch that exists today**. Only [`Frequency`](Self::Frequency) + /// is realised; [`WeightedSum`](Self::WeightedSum) is additive + /// (sketchable in principle) but has no implemented sketch yet, so + /// it stays generic until one lands. + pub fn is_realised_heavy_hitter(self) -> bool { + matches!(self, RankingMeasure::Frequency) + } +} + +/// Classify the aggregate a top-k ranks by into its [`RankingMeasure`]. +pub fn ranking_measure(agg: &AggIntent) -> RankingMeasure { + match agg { + AggIntent::Count { .. } => RankingMeasure::Frequency, + AggIntent::Sum { .. } => RankingMeasure::WeightedSum, + _ => RankingMeasure::NonAdditive, + } +} + +/// The single rule that decides whether a top-k ranking is the frequency +/// **heavy-hitter** that [`AggIntent::TopK`] represents, as opposed to a +/// generic order-by-value `Sort + Limit`. A ranking qualifies iff it +/// takes the **top** k (`descending`) **and** ranks by a measure with a +/// realised heavy-hitter sketch. +pub fn is_frequency_heavy_hitter(descending: bool, measure: RankingMeasure) -> bool { + descending && measure.is_realised_heavy_hitter() +} + /// Two instances of this aggregation can be merged -/// (`agg(A ∪ B) = combine(agg(A), agg(B))`). `Avg` is the only -/// non-mergeable case (needs `(sum, count)`, not a single value). +/// (`agg(A ∪ B) = combine(agg(A), agg(B))`). `Avg` / `StdDev` / +/// `Variance` need richer partial state than a single value, so they are +/// not mergeable. pub fn agg_is_mergeable(op: &AggIntent) -> bool { - !matches!(op, AggIntent::Avg) + !matches!( + op, + AggIntent::Avg { .. } | AggIntent::StdDev { .. } | AggIntent::Variance { .. } + ) } /// Whether this op implies `exact_required` — no sketch benefit. The -/// exact intents are `Sum / Count / Avg / Min / Max`. +/// exact intents are `Sum / Count / Avg / Min / Max / Group / +/// CountValues`. pub fn agg_is_exact(op: &AggIntent) -> bool { matches!( op, - AggIntent::Sum | AggIntent::Count { .. } | AggIntent::Avg | AggIntent::Min | AggIntent::Max + AggIntent::Sum { .. } + | AggIntent::Count { .. } + | AggIntent::Avg { .. } + | AggIntent::Min { .. } + | AggIntent::Max { .. } + | AggIntent::Group + | AggIntent::CountValues { .. } ) } @@ -332,7 +619,7 @@ pub fn agg_is_exact(op: &AggIntent) -> bool { pub fn agg_accuracy(op: &AggIntent) -> f64 { match op { AggIntent::Quantile { accuracy, .. } - | AggIntent::Cardinality { accuracy } + | AggIntent::Cardinality { accuracy, .. } | AggIntent::Frequency { accuracy } | AggIntent::Count { accuracy } | AggIntent::TopK { accuracy, .. } => accuracy_target_to_f64(accuracy), @@ -347,24 +634,29 @@ fn accuracy_target_to_f64(t: &AccuracyTarget) -> f64 { } } -/// Default `Frequency` intent — `accuracy = e / 2000`. +/// Default `Frequency` intent — `accuracy = e / 2000`. control_plane-only +/// (see module docs); not touched by the ASAPController merge. pub fn default_frequency() -> AggIntent { AggIntent::Frequency { accuracy: AccuracyTarget::Epsilon(std::f64::consts::E / 2000.0), } } -/// Default `Cardinality` intent — `accuracy = hll_accuracy(14)`. +/// Default `Cardinality` intent over the sample value — `accuracy = +/// hll_accuracy(14)`. pub fn default_cardinality() -> AggIntent { AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::Epsilon(crate::sketch_algebra::capability::hll_accuracy(14)), } } -/// Default `Quantile` intent at φ = `q`, `accuracy = ε 0.01`. Canonical -/// `Quantile` is single-φ; multi-φ callers invoke this once per φ. +/// Default `Quantile` intent over the sample value at φ = `q`, `accuracy +/// = ε 0.01`. Canonical `Quantile` is single-φ; multi-φ callers invoke +/// this once per φ. pub fn default_quantile(q: f64) -> AggIntent { AggIntent::Quantile { + col: None, q, accuracy: AccuracyTarget::Epsilon(0.01), } @@ -377,7 +669,7 @@ mod tests { use super::*; use crate::intent_algebra::schema::{Column, DataType}; - fn col(name: &str, dtype: DataType) -> Column { + fn c(name: &str, dtype: DataType) -> Column { Column { name: name.into(), dtype, @@ -391,11 +683,21 @@ mod tests { AggIntent::Count { accuracy: AccuracyTarget::Exact, }, - AggIntent::Sum, - AggIntent::Min, - AggIntent::Max, - AggIntent::Avg, + AggIntent::Sum { col: None }, + AggIntent::Sum { col: Some(3) }, + AggIntent::Min { col: None }, + AggIntent::Max { col: None }, + AggIntent::Avg { col: None }, + AggIntent::StdDev { + col: None, + population: false, + }, + AggIntent::Variance { + col: Some(1), + population: true, + }, AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }, @@ -404,6 +706,7 @@ mod tests { accuracy: AccuracyTarget::Epsilon(0.05), }, AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::EpsilonDelta { eps: 0.01, delta: 0.001, @@ -418,6 +721,23 @@ mod tests { AggIntent::Increase { window: Duration::from_secs(300), }, + AggIntent::HistogramCount, + AggIntent::HistogramFraction { + lower: 0.0, + upper: 1.0, + }, + AggIntent::Math(MathFunc::Abs), + AggIntent::Math(MathFunc::Clamp { min: 0.0, max: 1.0 }), + AggIntent::Absent, + AggIntent::AbsentOverTime, + AggIntent::PresentOverTime, + AggIntent::TimeFn(TimeFunc::DayOfWeek), + AggIntent::Group, + AggIntent::CountValues { + label: "value".into(), + }, + AggIntent::LastOverTime, + AggIntent::TsOfMaxOverTime, ]; for variant in cases { let json = serde_json::to_string(&variant).unwrap(); @@ -428,7 +748,7 @@ mod tests { #[test] fn output_column_names_are_intent_keyed() { - let v = col("value", DataType::Float64); + let v = c("value", DataType::Float64); assert_eq!( AggIntent::Count { accuracy: AccuracyTarget::Exact, @@ -437,9 +757,10 @@ mod tests { .name, "count" ); - assert_eq!(AggIntent::Sum.output_column(&v).name, "sum"); + assert_eq!(AggIntent::Sum { col: None }.output_column(&v).name, "sum"); assert_eq!( AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), } @@ -460,8 +781,9 @@ mod tests { #[test] fn quantile_output_is_float64() { - let v = col("value", DataType::Int64); + let v = c("value", DataType::Int64); let out = AggIntent::Quantile { + col: None, q: 0.5, accuracy: AccuracyTarget::Epsilon(0.01), } @@ -471,31 +793,32 @@ mod tests { #[test] fn sum_preserves_input_dtype() { - let int_col = col("c", DataType::Int64); - let float_col = col("c", DataType::Float64); + let int_col = c("c", DataType::Int64); + let float_col = c("c", DataType::Float64); assert!(matches!( - AggIntent::Sum.output_column(&int_col).dtype, + AggIntent::Sum { col: None }.output_column(&int_col).dtype, DataType::Int64 )); assert!(matches!( - AggIntent::Sum.output_column(&float_col).dtype, + AggIntent::Sum { col: None }.output_column(&float_col).dtype, DataType::Float64 )); } - // ── Phase β archive-only intent tests ───────────────────────────────── + // ── Archive-only intent tests ────────────────────────────────────────── - /// Every intent the legacy `asap-planner-rs::single_query::is_supported` - /// previously refused now lifts to L3 with `archive_only() == true`. - /// The negative cases are the ASAP-tier-bound intents — they must - /// continue to return false, otherwise the L4 binder would short-circuit - /// them to the cold tier. + /// Every intent with no ASAP-tier sketch binding today — the Phase β + /// migration targets plus everything newly added by the Phase 1 IR + /// merge — must return `archive_only() == true`. The negative cases + /// are the ASAP-tier-bound intents — they must continue to return + /// false, otherwise the L4 binder would short-circuit them to the + /// cold tier. #[test] fn archive_only_flag_partitions_intents() { - // Archive-only — every Phase β migration target. let archive: Vec = vec![ AggIntent::Absent, - AggIntent::Present, + AggIntent::AbsentOverTime, + AggIntent::PresentOverTime, AggIntent::Delta { window: Duration::from_secs(60), }, @@ -504,17 +827,14 @@ mod tests { }, AggIntent::PredictLinear { window: Duration::from_secs(300), - ahead: Duration::from_secs(60), + seconds: 60.0, }, - AggIntent::HoltWinters { + AggIntent::DoubleExpSmoothing { window: Duration::from_secs(300), - smoothing_factor: 0.3, - trend_factor: 0.3, - }, - AggIntent::Idelta { - window: Duration::from_secs(60), + smoothing: 0.3, + trend: 0.3, }, - AggIntent::Irate { + AggIntent::IDelta { window: Duration::from_secs(60), }, AggIntent::Resets { @@ -523,25 +843,33 @@ mod tests { AggIntent::Changes { window: Duration::from_secs(300), }, + AggIntent::HistogramCount, + AggIntent::Math(MathFunc::Abs), + AggIntent::TimeFn(TimeFunc::Hour), + AggIntent::Group, + AggIntent::CountValues { label: "v".into() }, + AggIntent::LastOverTime, ]; for v in archive { assert!( v.archive_only(), - "{v:?} should be archive-only after Phase β migration" + "{v:?} should be archive-only (no ASAP-tier binding yet)" ); } // Warm-tier — must NOT be flagged archive-only or the L4 binder - // breaks. + // breaks. `Irate` is intentionally absent from this list (removed + // by the rate/irate fold — see module docs). let warm: Vec = vec![ AggIntent::Count { accuracy: AccuracyTarget::Exact, }, - AggIntent::Sum, - AggIntent::Min, - AggIntent::Max, - AggIntent::Avg, + AggIntent::Sum { col: None }, + AggIntent::Min { col: None }, + AggIntent::Max { col: None }, + AggIntent::Avg { col: None }, AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }, @@ -550,6 +878,7 @@ mod tests { accuracy: AccuracyTarget::Epsilon(0.05), }, AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::EpsilonDelta { eps: 0.01, delta: 0.001, @@ -577,7 +906,8 @@ mod tests { fn archive_only_intent_serde_roundtrip() { let cases = vec![ AggIntent::Absent, - AggIntent::Present, + AggIntent::AbsentOverTime, + AggIntent::PresentOverTime, AggIntent::Delta { window: Duration::from_secs(60), }, @@ -586,17 +916,14 @@ mod tests { }, AggIntent::PredictLinear { window: Duration::from_secs(300), - ahead: Duration::from_secs(60), + seconds: 60.0, }, - AggIntent::HoltWinters { + AggIntent::DoubleExpSmoothing { window: Duration::from_secs(300), - smoothing_factor: 0.3, - trend_factor: 0.3, + smoothing: 0.3, + trend: 0.3, }, - AggIntent::Idelta { - window: Duration::from_secs(60), - }, - AggIntent::Irate { + AggIntent::IDelta { window: Duration::from_secs(60), }, AggIntent::Resets { @@ -615,9 +942,10 @@ mod tests { #[test] fn archive_only_output_column_names() { - let v = col("value", DataType::Float64); + let v = c("value", DataType::Float64); assert_eq!(AggIntent::Absent.output_column(&v).name, "absent"); - assert_eq!(AggIntent::Present.output_column(&v).name, "present"); + assert_eq!(AggIntent::AbsentOverTime.output_column(&v).name, "absent"); + assert_eq!(AggIntent::PresentOverTime.output_column(&v).name, "present"); assert_eq!( AggIntent::Delta { window: Duration::from_secs(60) @@ -635,4 +963,57 @@ mod tests { "resets" ); } + + // ── RankingMeasure tests (Phase 1 IR merge, adopted from ASAPController) ── + + #[test] + fn frequency_heavy_hitter_rule() { + use RankingMeasure::*; + assert!(is_frequency_heavy_hitter(true, Frequency)); + assert!(!is_frequency_heavy_hitter(false, Frequency)); + assert!(!is_frequency_heavy_hitter(true, NonAdditive)); + assert!(!is_frequency_heavy_hitter(true, WeightedSum)); + } + + #[test] + fn ranking_measure_classifies_by_additivity() { + use RankingMeasure::*; + assert_eq!( + ranking_measure(&AggIntent::Count { + accuracy: AccuracyTarget::Exact + }), + Frequency + ); + assert_eq!(ranking_measure(&AggIntent::Sum { col: None }), WeightedSum); + assert_eq!(ranking_measure(&AggIntent::Avg { col: None }), NonAdditive); + assert_eq!(ranking_measure(&AggIntent::Max { col: None }), NonAdditive); + assert!(Frequency.is_realised_heavy_hitter()); + assert!(!WeightedSum.is_realised_heavy_hitter()); + assert!(!NonAdditive.is_realised_heavy_hitter()); + } + + #[test] + fn input_col_tracks_only_reducers() { + assert_eq!(AggIntent::Sum { col: Some(3) }.input_col(), Some(3)); + assert_eq!( + AggIntent::Avg { col: None }.input_col(), + None, + "None = PromQL sample value" + ); + assert_eq!( + AggIntent::Count { + accuracy: AccuracyTarget::Exact + } + .input_col(), + None + ); + assert_eq!( + AggIntent::Frequency { + accuracy: AccuracyTarget::Exact + } + .input_col(), + None, + "Frequency is a keyed point-query, not a column reducer" + ); + } } diff --git a/control_plane/src/intent_algebra/column_resolution.rs b/control_plane/src/intent_algebra/column_resolution.rs index 6ece5c1b..0c93e80c 100644 --- a/control_plane/src/intent_algebra/column_resolution.rs +++ b/control_plane/src/intent_algebra/column_resolution.rs @@ -236,7 +236,7 @@ pub fn resolve_named_keys(keys: &[String], schema: &Schema) -> Result = vec![]; /// let aggs = vec![ /// AggIntent::Count { accuracy: types_v2::AccuracyTarget::Exact }, -/// AggIntent::Sum, +/// AggIntent::Sum { col: None }, /// ]; /// let output = output_schema_for_aggregate(&input, &by, &aggs); /// assert_eq!(output.columns.len(), 2); // count + sum @@ -410,7 +410,7 @@ mod tests { }); // Group by host, region (positions 2 and 3). let by = vec![2usize, 3usize]; - let aggs = vec![AggIntent::Sum]; + let aggs = vec![AggIntent::Sum { col: None }]; let out = output_schema_for_aggregate(&input, &by, &aggs); // Output columns: host, region, sum. assert_eq!(out.columns.len(), 3); @@ -425,7 +425,7 @@ mod tests { fn output_schema_for_aggregate_drops_out_of_range_by_ids() { let input = infer_source_schema("m"); // schema only has columns 0..=1; ask for by=[5] which is out of range. - let aggs = vec![AggIntent::Sum]; + let aggs = vec![AggIntent::Sum { col: None }]; let out = output_schema_for_aggregate(&input, &[5usize], &aggs); // The out-of-range by id is silently dropped; output has only the agg. assert_eq!(out.columns.len(), 1); diff --git a/control_plane/src/intent_algebra/cse.rs b/control_plane/src/intent_algebra/cse.rs index 008157bc..0863d6ac 100644 --- a/control_plane/src/intent_algebra/cse.rs +++ b/control_plane/src/intent_algebra/cse.rs @@ -217,6 +217,7 @@ mod tests { let q = QueryExpr::Aggregate { by: vec![1], aggs: vec![AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }], @@ -237,6 +238,7 @@ mod tests { let q1 = QueryExpr::Aggregate { by: vec![1], aggs: vec![AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }], @@ -246,6 +248,7 @@ mod tests { let q2 = QueryExpr::Aggregate { by: vec![1], aggs: vec![AggIntent::Quantile { + col: None, q: 0.95, accuracy: AccuracyTarget::Epsilon(0.01), }], @@ -280,7 +283,7 @@ mod tests { fn dedupe_subtrees_no_shared_subexpr() { let q1 = QueryExpr::Aggregate { by: vec![], - aggs: vec![AggIntent::Sum], + aggs: vec![AggIntent::Sum { col: None }], having: None, child: Box::new(windowed_scan()), }; @@ -303,7 +306,7 @@ mod tests { }; let q2 = QueryExpr::Aggregate { by: vec![], - aggs: vec![AggIntent::Max], + aggs: vec![AggIntent::Max { col: None }], having: None, child: Box::new(QueryExpr::Window { kind: WindowKind::Sliding, diff --git a/control_plane/src/intent_algebra/lower.rs b/control_plane/src/intent_algebra/lower.rs index c0eee13a..76333c85 100644 --- a/control_plane/src/intent_algebra/lower.rs +++ b/control_plane/src/intent_algebra/lower.rs @@ -488,25 +488,28 @@ fn agg_func_to_intents(func: &AggFunc) -> Vec { AggFunc::Frequency => vec![default_frequency()], AggFunc::Count => vec![default_frequency()], AggFunc::Avg => vec![AggIntent::Quantile { + col: None, q: 0.5, accuracy: AccuracyTarget::Epsilon(0.01), }], - AggFunc::Min => vec![AggIntent::Min], - AggFunc::Max => vec![AggIntent::Max], + AggFunc::Min => vec![AggIntent::Min { col: None }], + AggFunc::Max => vec![AggIntent::Max { col: None }], // StdDev / Variance: legacy carried two quantiles in a single // Quantile intent; Step α F1 fans them out into two siblings. AggFunc::StdDev { .. } | AggFunc::Variance { .. } => vec![ AggIntent::Quantile { + col: None, q: 0.25, accuracy: AccuracyTarget::Epsilon(0.01), }, AggIntent::Quantile { + col: None, q: 0.75, accuracy: AccuracyTarget::Epsilon(0.01), }, ], AggFunc::Sum | AggFunc::Rate | AggFunc::Increase | AggFunc::Delta => { - vec![AggIntent::Sum] + vec![AggIntent::Sum { col: None }] } AggFunc::Custom(_) => vec![], } @@ -590,7 +593,7 @@ mod tests { assert_eq!(size, Duration::from_secs(300)); match *child { CQueryExpr::Aggregate { aggs, child, .. } => { - assert!(matches!(aggs.as_slice(), [AggIntent::Sum])); + assert!(matches!(aggs.as_slice(), [AggIntent::Sum { col: None }])); assert!(matches!(*child, CQueryExpr::Scan { .. })); } other => panic!("expected Aggregate, got {other:?}"), @@ -629,7 +632,7 @@ mod tests { match convert_root(&legacy).unwrap() { CQueryExpr::Aggregate { by, aggs, .. } => { assert!(by.is_empty(), "no GROUP BY → empty `by`: {by:?}"); - assert!(matches!(aggs.as_slice(), [AggIntent::Sum])); + assert!(matches!(aggs.as_slice(), [AggIntent::Sum { col: None }])); } other => panic!("expected Aggregate, got {other:?}"), } diff --git a/control_plane/src/intent_algebra/query_expr.rs b/control_plane/src/intent_algebra/query_expr.rs index 30e1bd84..1662aab4 100644 --- a/control_plane/src/intent_algebra/query_expr.rs +++ b/control_plane/src/intent_algebra/query_expr.rs @@ -823,6 +823,7 @@ mod tests { let expr = QueryExpr::Aggregate { by: vec![1], // service aggs: vec![AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }], @@ -860,7 +861,7 @@ mod tests { }), child: Box::new(QueryExpr::Aggregate { by: vec![1], - aggs: vec![AggIntent::Max], + aggs: vec![AggIntent::Max { col: None }], having: None, child: Box::new(QueryExpr::Ref { name: BindingName::new("w"), @@ -906,7 +907,7 @@ mod tests { fn query_expr_aggregate_invalid_by_column() { let expr = QueryExpr::Aggregate { by: vec![99], - aggs: vec![AggIntent::Sum], + aggs: vec![AggIntent::Sum { col: None }], having: None, child: Box::new(ts_scan()), }; @@ -919,6 +920,7 @@ mod tests { let expr = QueryExpr::Aggregate { by: vec![1], aggs: vec![AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }], @@ -942,7 +944,7 @@ mod tests { // consumers can legally share this. let producer_with_uk = QueryExpr::Aggregate { by: vec![1], - aggs: vec![AggIntent::Sum], + aggs: vec![AggIntent::Sum { col: None }], having: None, child: Box::new(QueryExpr::Window { kind: WindowKind::Sliding, diff --git a/control_plane/src/intent_algebra/relational.rs b/control_plane/src/intent_algebra/relational.rs index b418fef3..ad1ee66e 100644 --- a/control_plane/src/intent_algebra/relational.rs +++ b/control_plane/src/intent_algebra/relational.rs @@ -515,10 +515,10 @@ impl AggFunc { AggFunc::Count => Some(AggIntent::Count { accuracy: AccuracyTarget::Exact, }), - AggFunc::Sum => Some(AggIntent::Sum), - AggFunc::Avg => Some(AggIntent::Avg), - AggFunc::Min => Some(AggIntent::Min), - AggFunc::Max => Some(AggIntent::Max), + AggFunc::Sum => Some(AggIntent::Sum { col: None }), + AggFunc::Avg => Some(AggIntent::Avg { col: None }), + AggFunc::Min => Some(AggIntent::Min { col: None }), + AggFunc::Max => Some(AggIntent::Max { col: None }), _ => None, } } @@ -914,7 +914,7 @@ mod tests { #[test] fn agg_intent_avg_not_mergeable() { - assert!(!agg_is_mergeable(&AggIntent::Avg)); + assert!(!agg_is_mergeable(&AggIntent::Avg { col: None })); } #[test] @@ -936,9 +936,9 @@ mod tests { #[test] fn agg_intent_is_exact() { - assert!(agg_is_exact(&AggIntent::Sum)); - assert!(agg_is_exact(&AggIntent::Min)); - assert!(agg_is_exact(&AggIntent::Max)); + assert!(agg_is_exact(&AggIntent::Sum { col: None })); + assert!(agg_is_exact(&AggIntent::Min { col: None })); + assert!(agg_is_exact(&AggIntent::Max { col: None })); assert!(!agg_is_exact(&default_cardinality())); } } diff --git a/control_plane/src/optimizer/cost/mod.rs b/control_plane/src/optimizer/cost/mod.rs index 8368966a..a13f6cb1 100644 --- a/control_plane/src/optimizer/cost/mod.rs +++ b/control_plane/src/optimizer/cost/mod.rs @@ -704,7 +704,10 @@ fn node_cost_aggregate(by: &[usize], aggs: &[AggIntent], _input: &Schema) -> f64 #[allow(dead_code)] fn intent_cost(intent: &AggIntent) -> f64 { match intent { - AggIntent::Sum | AggIntent::Min | AggIntent::Max | AggIntent::Avg => 5.0, + AggIntent::Sum { .. } + | AggIntent::Min { .. } + | AggIntent::Max { .. } + | AggIntent::Avg { .. } => 5.0, AggIntent::Count { .. } => 5.0, AggIntent::Quantile { .. } => 20.0, AggIntent::Cardinality { .. } => 15.0, @@ -777,6 +780,7 @@ mod workload_cost_tests { QueryExpr::Aggregate { by: vec![], aggs: vec![AggIntent::Quantile { + col: None, q, accuracy: AccuracyTarget::Epsilon(0.01), }], @@ -789,7 +793,7 @@ mod workload_cost_tests { fn max_root(child: QueryExpr) -> QueryExpr { QueryExpr::Aggregate { by: vec![], - aggs: vec![AggIntent::Max], + aggs: vec![AggIntent::Max { col: None }], having: None, child: Box::new(child), } diff --git a/control_plane/src/optimizer/rules/mod.rs b/control_plane/src/optimizer/rules/mod.rs index ad277398..6082dbba 100644 --- a/control_plane/src/optimizer/rules/mod.rs +++ b/control_plane/src/optimizer/rules/mod.rs @@ -177,10 +177,12 @@ pub fn bind_workload_typed(w: &QueryWorkload) -> Option L3AggIntent::Quantile { + col: None, q: w.quantiles.first().copied().unwrap_or(0.99), accuracy: intent_accuracy, }, StatisticClass::Cardinality => L3AggIntent::Cardinality { + col: None, accuracy: intent_accuracy, }, StatisticClass::Frequency => L3AggIntent::Frequency { diff --git a/control_plane/src/physical/allocator.rs b/control_plane/src/physical/allocator.rs index d99e5865..3eaa1f95 100644 --- a/control_plane/src/physical/allocator.rs +++ b/control_plane/src/physical/allocator.rs @@ -527,7 +527,7 @@ impl SketchAllocator { let intent = aggs[0].clone(); // Exact non-mergeable (Avg) → always Db. - if matches!(intent, AggIntent::Avg) { + if matches!(intent, AggIntent::Avg { .. }) { return PlanNode { expr: QueryExpr::Aggregate { by, @@ -673,10 +673,12 @@ fn estimated_sketch_memory(op: &AggIntent) -> f64 { fn canonical_intent_kind_str(intent: &AggIntent) -> &'static str { match intent { AggIntent::Count { .. } => "count", - AggIntent::Sum => "sum", - AggIntent::Min => "min", - AggIntent::Max => "max", - AggIntent::Avg => "avg", + AggIntent::Sum { .. } => "sum", + AggIntent::Min { .. } => "min", + AggIntent::Max { .. } => "max", + AggIntent::Avg { .. } => "avg", + AggIntent::StdDev { .. } => "stddev", + AggIntent::Variance { .. } => "variance", AggIntent::Quantile { .. } => "quantile", AggIntent::TopK { .. } => "topk", AggIntent::Cardinality { .. } => "cardinality", @@ -684,15 +686,33 @@ fn canonical_intent_kind_str(intent: &AggIntent) -> &'static str { AggIntent::Rate { .. } => "rate", AggIntent::Increase { .. } => "increase", AggIntent::Absent => "absent", - AggIntent::Present => "present", + AggIntent::AbsentOverTime => "absent_over_time", + AggIntent::PresentOverTime => "present_over_time", AggIntent::Delta { .. } => "delta", AggIntent::Deriv { .. } => "deriv", AggIntent::PredictLinear { .. } => "predict_linear", - AggIntent::HoltWinters { .. } => "holt_winters", - AggIntent::Idelta { .. } => "idelta", - AggIntent::Irate { .. } => "irate", + AggIntent::DoubleExpSmoothing { .. } => "double_exponential_smoothing", + AggIntent::IDelta { .. } => "idelta", AggIntent::Resets { .. } => "resets", AggIntent::Changes { .. } => "changes", + AggIntent::HistogramCount => "histogram_count", + AggIntent::HistogramSum => "histogram_sum", + AggIntent::HistogramAvg => "histogram_avg", + AggIntent::HistogramStdDev => "histogram_stddev", + AggIntent::HistogramStdVar => "histogram_stdvar", + AggIntent::HistogramFraction { .. } => "histogram_fraction", + AggIntent::HistogramQuantile { .. } => "histogram_quantile", + AggIntent::Math(_) => "math", + AggIntent::TimeFn(_) => "time_fn", + AggIntent::Group => "group", + AggIntent::CountValues { .. } => "count_values", + AggIntent::LastOverTime => "last_over_time", + AggIntent::FirstOverTime => "first_over_time", + AggIntent::MadOverTime => "mad_over_time", + AggIntent::TsOfMinOverTime => "ts_of_min_over_time", + AggIntent::TsOfMaxOverTime => "ts_of_max_over_time", + AggIntent::TsOfFirstOverTime => "ts_of_first_over_time", + AggIntent::TsOfLastOverTime => "ts_of_last_over_time", } } @@ -808,7 +828,7 @@ mod tests { #[test] fn exact_avg_goes_to_db() { - let node = alloc(unlimited(), agg(AggIntent::Avg)); + let node = alloc(unlimited(), agg(AggIntent::Avg { col: None })); assert_eq!(node.stage, PipelineStage::Db); assert_eq!(node.mode, ExecutionMode::Exact); } @@ -817,7 +837,7 @@ mod tests { #[test] fn exact_sum_goes_to_backend() { - let node = alloc(unlimited(), agg(AggIntent::Sum)); + let node = alloc(unlimited(), agg(AggIntent::Sum { col: None })); assert_eq!(node.stage, PipelineStage::Backend); assert_eq!(node.mode, ExecutionMode::Exact); } @@ -886,7 +906,7 @@ mod tests { fn multi_intent_aggregate_goes_to_db() { let expr = QueryExpr::Aggregate { by: vec![], - aggs: vec![AggIntent::Sum, AggIntent::Min], + aggs: vec![AggIntent::Sum { col: None }, AggIntent::Min { col: None }], having: None, child: Box::new(scan("m")), }; diff --git a/control_plane/src/physical/planner.rs b/control_plane/src/physical/planner.rs index 91802d0a..8c757752 100644 --- a/control_plane/src/physical/planner.rs +++ b/control_plane/src/physical/planner.rs @@ -635,6 +635,7 @@ mod tests { #[test] fn resolve_preserves_intent() { let intent = AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.005), }; @@ -721,7 +722,7 @@ mod tests { // (no single sketch serves multiple intents). let expr = QueryExpr::Aggregate { by: vec![], - aggs: vec![AggIntent::Sum, AggIntent::Min], + aggs: vec![AggIntent::Sum { col: None }, AggIntent::Min { col: None }], having: None, child: Box::new(scan("trades")), }; diff --git a/control_plane/src/physical/sketch_catalog.rs b/control_plane/src/physical/sketch_catalog.rs index fd464cdf..441e9f39 100644 --- a/control_plane/src/physical/sketch_catalog.rs +++ b/control_plane/src/physical/sketch_catalog.rs @@ -115,7 +115,7 @@ pub fn sketch_params_for_op(op: &AggIntent) -> SketchParams { } // Min / Max — preserve the legacy `Extrema` mapping (DDSketch over // the 0.0 / 1.0 boundary quantiles). - AggIntent::Min | AggIntent::Max => SketchParams::DDSketch { + AggIntent::Min { .. } | AggIntent::Max { .. } => SketchParams::DDSketch { relative_accuracy: 0.01, quantiles: vec![0.0, 1.0], }, @@ -160,7 +160,7 @@ pub fn estimated_sketch_memory_bytes(op: &AggIntent) -> u64 { width * 5 * 8 } // Legacy `Extrema { .. }` (now canonical Min / Max) — 16 bytes. - AggIntent::Min | AggIntent::Max => 16, + AggIntent::Min { .. } | AggIntent::Max { .. } => 16, // Sum / Count / Avg / TopK / Rate / Increase / archive-only — no // sketch state; preserve the legacy `Exact(_) → 8` mapping. _ => 8, @@ -270,6 +270,7 @@ mod tests { fn op_quantile_yields_ddsketch_type_and_params() { use crate::types_v2::AccuracyTarget; let op = AggIntent::Quantile { + col: None, q: 0.5, accuracy: AccuracyTarget::Epsilon(0.01), }; @@ -303,6 +304,7 @@ mod tests { fn memory_quantile() { use crate::types_v2::AccuracyTarget; let op = AggIntent::Quantile { + col: None, q: 0.5, accuracy: AccuracyTarget::Epsilon(0.01), }; diff --git a/control_plane/src/physical/window_fusion.rs b/control_plane/src/physical/window_fusion.rs index 7d6b9b04..46758857 100644 --- a/control_plane/src/physical/window_fusion.rs +++ b/control_plane/src/physical/window_fusion.rs @@ -321,7 +321,7 @@ mod tests { #[test] fn equivalence_sum_intent() { assert_equivalent( - AggIntent::Sum, + AggIntent::Sum { col: None }, WindowKind::Tumbling, Duration::from_secs(300), None, @@ -334,7 +334,7 @@ mod tests { // sketch case — not a windowed sketch. let canonical = QueryExpr::Aggregate { by: Vec::new(), - aggs: vec![AggIntent::Sum], + aggs: vec![AggIntent::Sum { col: None }], having: None, child: Box::new(canonical_scan("m")), }; diff --git a/control_plane/src/query_parser/mod.rs b/control_plane/src/query_parser/mod.rs index bcd425ce..563df476 100644 --- a/control_plane/src/query_parser/mod.rs +++ b/control_plane/src/query_parser/mod.rs @@ -324,7 +324,7 @@ impl QeCollector { self.quantiles.push(*q); } } - AggIntent::Min => { + AggIntent::Min { .. } => { if !self.agg_types.contains(&AggType::Quantile) { self.agg_types.push(AggType::Quantile); } @@ -332,7 +332,7 @@ impl QeCollector { self.quantiles.push(0.0); } } - AggIntent::Max => { + AggIntent::Max { .. } => { if !self.agg_types.contains(&AggType::Quantile) { self.agg_types.push(AggType::Quantile); } diff --git a/control_plane/src/query_parser/promql.rs b/control_plane/src/query_parser/promql.rs index 49e39e21..8aa1eb69 100644 --- a/control_plane/src/query_parser/promql.rs +++ b/control_plane/src/query_parser/promql.rs @@ -33,9 +33,7 @@ use std::time::Duration; use anyhow::anyhow; -use promql_parser::parser::{ - self, AggregateExpr, Call, Expr, LabelModifier, VectorSelector, -}; +use promql_parser::parser::{self, AggregateExpr, Call, Expr, LabelModifier, VectorSelector}; use crate::intent_algebra::relational::{FilterOp, FilterVal, PartitionKeys, Predicate}; diff --git a/control_plane/src/sketch_algebra/capability.rs b/control_plane/src/sketch_algebra/capability.rs index bbb27560..45f164dc 100644 --- a/control_plane/src/sketch_algebra/capability.rs +++ b/control_plane/src/sketch_algebra/capability.rs @@ -483,7 +483,7 @@ pub fn capability_for(intent: &AggIntent) -> Option { Some(Capability::QuantileApprox(SketchKindHandle::Any)) } } - AggIntent::Cardinality { accuracy } => { + AggIntent::Cardinality { accuracy, .. } => { if is_exact(accuracy) { None } else { @@ -552,7 +552,9 @@ pub fn capability_for(intent: &AggIntent) -> Option { // DDSketch / KLL answer min = quantile(0) and max = quantile(1) // out of the box. No dedicated extrema sketch is needed; route // these through the quantile-family handler. - AggIntent::Min | AggIntent::Max => Some(Capability::QuantileApprox(SketchKindHandle::Any)), + AggIntent::Min { .. } | AggIntent::Max { .. } => { + Some(Capability::QuantileApprox(SketchKindHandle::Any)) + } // ── ExactAgg (PR-6 follow-up) ──────────────────────────────── // These intents previously returned `None` and routed to the // archive engine. Now that the data plane carries @@ -561,28 +563,50 @@ pub fn capability_for(intent: &AggIntent) -> Option { // state instead. `is_satisfied_by` checks `agg_type` equality // structurally — a sid registered as `ExactAgg(Sum)` only // satisfies a required `ExactAgg(Sum)`. - AggIntent::Sum => Some(Capability::ExactAgg(AggregationType::Sum)), + AggIntent::Sum { .. } => Some(Capability::ExactAgg(AggregationType::Sum)), AggIntent::Rate { .. } | AggIntent::Increase { .. } => { Some(Capability::ExactAgg(AggregationType::Increase)) } - // ── Avg: still no ASAP-tier substitute ─────────────────────── + // ── Avg / StdDev / Variance: still no ASAP-tier substitute ──── // Avg = Sum / Count, which needs two separate ExactAgg policies // (one for Sum, one for Count) joined at query time. The L4 // binder doesn't yet emit that pattern, so capability_for keeps - // Avg on the archive path for now. Follow-up. - AggIntent::Avg => None, + // these on the archive path for now. Follow-up. + AggIntent::Avg { .. } | AggIntent::StdDev { .. } | AggIntent::Variance { .. } => None, // Archive-only intents — never bind to a ASAP-tier capability; - // routed to the cold tier (Gorilla / Thanos). + // routed to the cold tier (Gorilla / Thanos). Includes every + // intent added by the Phase 1 IR merge (none has a `Bind*` rule + // yet) plus the pre-existing archive-only set. `Irate` is + // intentionally absent — folded into `Rate` above (see + // `agg_intent.rs` module docs). AggIntent::Absent - | AggIntent::Present + | AggIntent::AbsentOverTime + | AggIntent::PresentOverTime | AggIntent::Delta { .. } | AggIntent::Deriv { .. } | AggIntent::PredictLinear { .. } - | AggIntent::HoltWinters { .. } - | AggIntent::Idelta { .. } - | AggIntent::Irate { .. } + | AggIntent::DoubleExpSmoothing { .. } + | AggIntent::IDelta { .. } | AggIntent::Resets { .. } - | AggIntent::Changes { .. } => None, + | AggIntent::Changes { .. } + | AggIntent::HistogramCount + | AggIntent::HistogramSum + | AggIntent::HistogramAvg + | AggIntent::HistogramStdDev + | AggIntent::HistogramStdVar + | AggIntent::HistogramFraction { .. } + | AggIntent::HistogramQuantile { .. } + | AggIntent::Math(_) + | AggIntent::TimeFn(_) + | AggIntent::Group + | AggIntent::CountValues { .. } + | AggIntent::LastOverTime + | AggIntent::FirstOverTime + | AggIntent::MadOverTime + | AggIntent::TsOfMinOverTime + | AggIntent::TsOfMaxOverTime + | AggIntent::TsOfFirstOverTime + | AggIntent::TsOfLastOverTime => None, } } @@ -838,6 +862,7 @@ mod tests { #[test] fn capability_for_quantile_returns_quantile_approx() { let intent = AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }; @@ -850,6 +875,7 @@ mod tests { #[test] fn capability_for_quantile_exact_returns_none() { let intent = AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Exact, }; @@ -861,6 +887,7 @@ mod tests { #[test] fn capability_for_cardinality_with_epsilon_returns_cardinality_approx() { let intent = AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::Epsilon(0.01), }; assert_eq!(capability_for(&intent), Some(Capability::CardinalityApprox)); @@ -869,6 +896,7 @@ mod tests { #[test] fn capability_for_cardinality_with_epsilon_delta_returns_cardinality_approx() { let intent = AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::EpsilonDelta { eps: 0.01, delta: 0.001, @@ -880,6 +908,7 @@ mod tests { #[test] fn capability_for_cardinality_with_exact_returns_none() { let intent = AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::Exact, }; assert_eq!(capability_for(&intent), None); @@ -913,7 +942,7 @@ mod tests { // PR-6 follow-up: Sum routes to ASAP-tier ExactAgg(Sum) state. // Pre-follow-up this returned `None`. assert_eq!( - capability_for(&AggIntent::Sum), + capability_for(&AggIntent::Sum { col: None }), Some(Capability::ExactAgg(AggregationType::Sum)) ); } @@ -924,14 +953,14 @@ mod tests { // today (a sketch-bound `Avg` would fold onto `Quantile{q=0.5}` // only when the cost model allows the relaxation, which is a // follow-up). - assert_eq!(capability_for(&AggIntent::Avg), None); + assert_eq!(capability_for(&AggIntent::Avg { col: None }), None); } #[test] fn capability_for_min_returns_quantile_approx() { // Min = quantile(0); DDSketch / KLL answer it directly. assert_eq!( - capability_for(&AggIntent::Min), + capability_for(&AggIntent::Min { col: None }), Some(Capability::QuantileApprox(SketchKindHandle::Any)) ); } @@ -940,7 +969,7 @@ mod tests { fn capability_for_max_returns_quantile_approx() { // Max = quantile(1); DDSketch / KLL answer it directly. assert_eq!( - capability_for(&AggIntent::Max), + capability_for(&AggIntent::Max { col: None }), Some(Capability::QuantileApprox(SketchKindHandle::Any)) ); } @@ -1029,9 +1058,13 @@ mod tests { #[test] fn capability_for_archive_only_intents_return_none() { - // Spot-check each archive-only variant. + // Spot-check each archive-only variant, including a few added by + // the Phase 1 IR merge. `Irate` is intentionally absent — folded + // into `Rate` (see `agg_intent.rs` module docs); `rate`/`irate` + // now share `capability_for`'s `Rate` arm. assert_eq!(capability_for(&AggIntent::Absent), None); - assert_eq!(capability_for(&AggIntent::Present), None); + assert_eq!(capability_for(&AggIntent::AbsentOverTime), None); + assert_eq!(capability_for(&AggIntent::PresentOverTime), None); assert_eq!( capability_for(&AggIntent::Delta { window: Duration::from_secs(60) @@ -1039,17 +1072,13 @@ mod tests { None ); assert_eq!( - capability_for(&AggIntent::Idelta { - window: Duration::from_secs(60) - }), - None - ); - assert_eq!( - capability_for(&AggIntent::Irate { + capability_for(&AggIntent::IDelta { window: Duration::from_secs(60) }), None ); + assert_eq!(capability_for(&AggIntent::HistogramCount), None); + assert_eq!(capability_for(&AggIntent::Group), None); } // ── Capability::is_satisfied_by ────────────────────────────────────── @@ -1238,7 +1267,7 @@ mod tests { // data plane has a real accumulator for: `Sum` (SumAccumulator) // and `Rate` / `Increase` (IncreaseAccumulator). assert_eq!( - capability_for(&AggIntent::Sum), + capability_for(&AggIntent::Sum { col: None }), Some(Capability::ExactAgg(AggregationType::Sum)) ); assert_eq!( @@ -1262,7 +1291,7 @@ mod tests { }), None ); - assert_eq!(capability_for(&AggIntent::Avg), None); + assert_eq!(capability_for(&AggIntent::Avg { col: None }), None); } #[test] diff --git a/control_plane/src/sketch_algebra/physical_expr.rs b/control_plane/src/sketch_algebra/physical_expr.rs index 593d868b..90a24b03 100644 --- a/control_plane/src/sketch_algebra/physical_expr.rs +++ b/control_plane/src/sketch_algebra/physical_expr.rs @@ -325,6 +325,7 @@ mod tests { QueryExpr::Aggregate { by: vec![], aggs: vec![AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }], diff --git a/control_plane/src/sketch_algebra/rules/bind_archive_only.rs b/control_plane/src/sketch_algebra/rules/bind_archive_only.rs index b6dbfd57..a1ed7a8a 100644 --- a/control_plane/src/sketch_algebra/rules/bind_archive_only.rs +++ b/control_plane/src/sketch_algebra/rules/bind_archive_only.rs @@ -143,7 +143,8 @@ mod tests { fn binds_each_archive_only_intent() { let intents = vec![ AggIntent::Absent, - AggIntent::Present, + AggIntent::AbsentOverTime, + AggIntent::PresentOverTime, AggIntent::Delta { window: Duration::from_secs(60), }, @@ -152,17 +153,14 @@ mod tests { }, AggIntent::PredictLinear { window: Duration::from_secs(300), - ahead: Duration::from_secs(60), + seconds: 60.0, }, - AggIntent::HoltWinters { + AggIntent::DoubleExpSmoothing { window: Duration::from_secs(300), - smoothing_factor: 0.3, - trend_factor: 0.3, + smoothing: 0.3, + trend: 0.3, }, - AggIntent::Idelta { - window: Duration::from_secs(60), - }, - AggIntent::Irate { + AggIntent::IDelta { window: Duration::from_secs(60), }, AggIntent::Resets { @@ -171,6 +169,8 @@ mod tests { AggIntent::Changes { window: Duration::from_secs(300), }, + AggIntent::HistogramCount, + AggIntent::Group, ]; for intent in intents { let expr = agg_with(intent.clone()); @@ -187,12 +187,14 @@ mod tests { // Sum / Quantile / Cardinality / TopK are NOT archive-only — they // must NOT trigger BindArchiveOnly (the ASAP-tier rules own them). for intent in [ - AggIntent::Sum, + AggIntent::Sum { col: None }, AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }, AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::Epsilon(0.01), }, AggIntent::TopK { diff --git a/control_plane/src/sketch_algebra/rules/bind_ddsketch_quantile.rs b/control_plane/src/sketch_algebra/rules/bind_ddsketch_quantile.rs index 9ec73ba7..a86b0c8c 100644 --- a/control_plane/src/sketch_algebra/rules/bind_ddsketch_quantile.rs +++ b/control_plane/src/sketch_algebra/rules/bind_ddsketch_quantile.rs @@ -46,7 +46,7 @@ impl Rule for BindDDSketchOnQuantile { QueryExpr::Aggregate { aggs, child, by, .. } if aggs.len() == 1 && by.is_empty() => match &aggs[0] { - AggIntent::Quantile { q, accuracy } => (*q, accuracy.clone(), child), + AggIntent::Quantile { q, accuracy, .. } => (*q, accuracy.clone(), child), _ => return None, }, _ => return None, diff --git a/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs b/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs index 3d5819cd..e7e867ef 100644 --- a/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs +++ b/control_plane/src/sketch_algebra/rules/bind_exact_agg.rs @@ -90,7 +90,7 @@ impl Rule for BindExactAgg { // the query can fan results out over the surviving labels. // Unkeyed aggregations stay on the single-pop variant. let agg_type = match intent { - AggIntent::Sum => { + AggIntent::Sum { .. } => { if keyed { AggregationType::MultipleSum } else { @@ -166,7 +166,7 @@ mod tests { #[test] fn binds_sum_to_exact_agg_sum() { - check_binds(AggIntent::Sum, AggregationType::Sum); + check_binds(AggIntent::Sum { col: None }, AggregationType::Sum); } #[test] @@ -217,7 +217,7 @@ mod tests { #[test] fn does_not_bind_avg() { - let expr = agg_over(AggIntent::Avg, "test_metric"); + let expr = agg_over(AggIntent::Avg { col: None }, "test_metric"); assert!(BindExactAgg.apply(&expr, &AccuracyTarget::Exact).is_none()); } @@ -225,6 +225,7 @@ mod tests { fn does_not_bind_quantile() { let expr = agg_over( AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }, @@ -255,7 +256,7 @@ mod tests { #[test] fn keyed_sum_binds_to_multiple_sum() { - check_keyed_binds(AggIntent::Sum, AggregationType::MultipleSum); + check_keyed_binds(AggIntent::Sum { col: None }, AggregationType::MultipleSum); } #[test] @@ -296,7 +297,7 @@ mod tests { fn unkeyed_sum_still_binds_to_single_pop_sum() { // Regression guard: the keyed/unkeyed branch must still // dispatch correctly on `by.is_empty()`. - check_binds(AggIntent::Sum, AggregationType::Sum); + check_binds(AggIntent::Sum { col: None }, AggregationType::Sum); } #[test] diff --git a/control_plane/src/sketch_algebra/rules/bind_hll_cardinality.rs b/control_plane/src/sketch_algebra/rules/bind_hll_cardinality.rs index 17beb822..89d68a15 100644 --- a/control_plane/src/sketch_algebra/rules/bind_hll_cardinality.rs +++ b/control_plane/src/sketch_algebra/rules/bind_hll_cardinality.rs @@ -36,7 +36,7 @@ impl Rule for BindHllOnCardinality { QueryExpr::Aggregate { aggs, child, by, .. } if aggs.len() == 1 && by.is_empty() => match &aggs[0] { - AggIntent::Cardinality { accuracy } => (accuracy.clone(), child), + AggIntent::Cardinality { accuracy, .. } => (accuracy.clone(), child), _ => return None, }, _ => return None, diff --git a/control_plane/src/sketch_algebra/rules/bind_kll_quantile.rs b/control_plane/src/sketch_algebra/rules/bind_kll_quantile.rs index 29deb094..400dae29 100644 --- a/control_plane/src/sketch_algebra/rules/bind_kll_quantile.rs +++ b/control_plane/src/sketch_algebra/rules/bind_kll_quantile.rs @@ -48,7 +48,7 @@ impl Rule for BindKllOnQuantile { QueryExpr::Aggregate { aggs, child, by, .. } if aggs.len() == 1 && by.is_empty() => match &aggs[0] { - AggIntent::Quantile { q, accuracy } => (*q, accuracy.clone(), child), + AggIntent::Quantile { q, accuracy, .. } => (*q, accuracy.clone(), child), _ => return None, }, _ => return None, diff --git a/control_plane/src/sketch_algebra/tests.rs b/control_plane/src/sketch_algebra/tests.rs index 44b9e83b..efe5ebe6 100644 --- a/control_plane/src/sketch_algebra/tests.rs +++ b/control_plane/src/sketch_algebra/tests.rs @@ -57,7 +57,11 @@ fn windowed_scan() -> QueryExpr { fn agg_quantile(q: f64, accuracy: AccuracyTarget) -> QueryExpr { QueryExpr::Aggregate { by: vec![], - aggs: vec![AggIntent::Quantile { q, accuracy }], + aggs: vec![AggIntent::Quantile { + col: None, + q, + accuracy, + }], having: None, child: Box::new(windowed_scan()), } @@ -324,7 +328,11 @@ fn bind_cms_topk_picks_cost_min_meeting_sla() { let bound = bind_query_expr(&agg_topk(10, acc.clone()), acc).unwrap(); let (kind, ..) = topk_binding_family(&bound); let chosen = table.for_kind(&kind).per_flush(); - assert_eq!(chosen, cms.min(cs), "must pick the cost-min family that meets the SLA"); + assert_eq!( + chosen, + cms.min(cs), + "must pick the cost-min family that meets the SLA" + ); assert_eq!(kind, SketchKind::Cms); } @@ -333,6 +341,7 @@ fn bind_hll_cardinality_basic() { let expr = QueryExpr::Aggregate { by: vec![], aggs: vec![AggIntent::Cardinality { + col: None, accuracy: AccuracyTarget::Epsilon(0.01), }], having: None, @@ -375,7 +384,7 @@ fn sum_now_binds_to_exact_agg_after_pr_6_followup() { // exact-aggregation path can serve the intent. let expr = QueryExpr::Aggregate { by: vec![], - aggs: vec![AggIntent::Sum], + aggs: vec![AggIntent::Sum { col: None }], having: None, child: Box::new(windowed_scan()), }; @@ -462,6 +471,7 @@ fn phase_b_pattern_only_temporal_quantile_binds_to_sketch() { let expr = QueryExpr::Aggregate { by: vec![], aggs: vec![AggIntent::Quantile { + col: None, q: 0.99, accuracy: AccuracyTarget::Epsilon(0.01), }], @@ -497,7 +507,7 @@ fn phase_b_pattern_only_temporal_quantile_binds_to_sketch() { fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() { let expr = QueryExpr::Aggregate { by: vec![], - aggs: vec![AggIntent::Sum], + aggs: vec![AggIntent::Sum { col: None }], having: None, child: Box::new(windowed_scan()), }; @@ -521,7 +531,7 @@ fn phase_b_pattern_only_temporal_sum_binds_to_exact_agg() { fn phase_b_pattern_only_spatial_aggregate_binds_to_multiple_sum() { let expr = QueryExpr::Aggregate { by: vec![1], // service column - aggs: vec![AggIntent::Sum], + aggs: vec![AggIntent::Sum { col: None }], having: None, child: Box::new(ts_scan()), }; @@ -843,7 +853,8 @@ fn phase_b_e2e_archive_only_e2e_binding() { fn phase_b_archive_only_intents_round_trip_through_binder() { let intents = vec![ AggIntent::Absent, - AggIntent::Present, + AggIntent::AbsentOverTime, + AggIntent::PresentOverTime, AggIntent::Delta { window: Duration::from_secs(60), }, @@ -852,17 +863,14 @@ fn phase_b_archive_only_intents_round_trip_through_binder() { }, AggIntent::PredictLinear { window: Duration::from_secs(300), - ahead: Duration::from_secs(60), + seconds: 60.0, }, - AggIntent::HoltWinters { + AggIntent::DoubleExpSmoothing { window: Duration::from_secs(300), - smoothing_factor: 0.3, - trend_factor: 0.3, - }, - AggIntent::Idelta { - window: Duration::from_secs(60), + smoothing: 0.3, + trend: 0.3, }, - AggIntent::Irate { + AggIntent::IDelta { window: Duration::from_secs(60), }, AggIntent::Resets { @@ -871,6 +879,10 @@ fn phase_b_archive_only_intents_round_trip_through_binder() { AggIntent::Changes { window: Duration::from_secs(300), }, + // Spot-check a couple of the Phase 1 IR merge's new archive-only + // intents through the same round-trip. + AggIntent::HistogramCount, + AggIntent::Group, ]; for intent in intents { let expr = QueryExpr::Aggregate {