From 929a8598a196c99b12dc6324fc9f7eedd85f0ee8 Mon Sep 17 00:00:00 2001 From: zz_y Date: Mon, 20 Jul 2026 11:52:20 -0600 Subject: [PATCH] chore(sketch_algebra): re-layer three cost/planning modules out of L4 IR MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Stage 4 of the sketch_algebra re-layering, scoped down after a fresh audit found two of the originally-planned four items were based on a stale premise (sketch_params.rs has 14 real production consumers, not "one"; PhysicalExpr::RawAtEdgeSketchAtBackend/RawAtEdgePrometheusArchive are actively constructed across optimizer/rules, emit/*, and physical/colored_dag/*, not dead) — both dropped from this stage pending a separate, properly-scoped investigation. Three confirmed-safe moves: - Deleted sketch_algebra/schema.rs. SketchStateSchema/SketchStateMetadata had zero real callers — every hit was a stale doc-comment cross-reference, never constructed or consumed anywhere. - Moved SketchCapability/SupportedIntent/default_capability_table/ load_capability_overrides from sketch_algebra::capability into optimizer::cost::sketch_capability. This is the perf/cost-model half of the "four overlapping capability tables" Step 2a originally consolidated — read by the optimizer and physical planner for cost-based plan rewriting, not L4 IR. sketch_algebra::capability keeps the query-side Capability/SketchKindHandle tag and the capability_for semantic bridge, which are genuinely L4-adjacent. - Moved sketch_selection.rs to a top-level control_plane::sketch_selection module. Its one external caller is query_planning.rs; naming the concrete sketch families that satisfy a Capability is a query-planning concern, not L4 IR. No behavior change — all three moves are verbatim relocations with updated doc comments and import paths. - cargo build --workspace: clean - cargo test -p control_plane: 824 passed, 1 pre-existing unrelated failure (invalid_sketch_type_override_falls_back_to_default) Co-Authored-By: Claude Sonnet 5 --- control_plane/src/lib.rs | 3 +- control_plane/src/optimizer/cost/mod.rs | 1 + .../src/optimizer/cost/sketch_capability.rs | 265 ++++++++++++++++ control_plane/src/optimizer/engine.rs | 12 +- control_plane/src/query_planning.rs | 21 +- .../src/sketch_algebra/capability.rs | 293 +----------------- control_plane/src/sketch_algebra/mod.rs | 14 +- control_plane/src/sketch_algebra/schema.rs | 181 ----------- .../{sketch_algebra => }/sketch_selection.rs | 34 +- 9 files changed, 316 insertions(+), 508 deletions(-) create mode 100644 control_plane/src/optimizer/cost/sketch_capability.rs delete mode 100644 control_plane/src/sketch_algebra/schema.rs rename control_plane/src/{sketch_algebra => }/sketch_selection.rs (87%) diff --git a/control_plane/src/lib.rs b/control_plane/src/lib.rs index b6545887..8a07d261 100644 --- a/control_plane/src/lib.rs +++ b/control_plane/src/lib.rs @@ -60,8 +60,8 @@ pub mod accuracy; pub mod backend_client; pub mod deployment_model; -pub mod epsilon_alloc; pub mod emit; +pub mod epsilon_alloc; pub mod intent_algebra; pub mod metrics_exposer; pub mod monitor; @@ -74,6 +74,7 @@ pub mod query_planning; pub mod replan; pub mod runtime_samples; pub mod sketch_algebra; +pub mod sketch_selection; pub mod store; pub mod threshold_alloc; pub mod types; diff --git a/control_plane/src/optimizer/cost/mod.rs b/control_plane/src/optimizer/cost/mod.rs index c60a5be1..ad96858d 100644 --- a/control_plane/src/optimizer/cost/mod.rs +++ b/control_plane/src/optimizer/cost/mod.rs @@ -7,6 +7,7 @@ use std::time::Duration; pub mod delta; pub mod online; pub mod pareto; +pub mod sketch_capability; pub mod tco; pub mod wire; diff --git a/control_plane/src/optimizer/cost/sketch_capability.rs b/control_plane/src/optimizer/cost/sketch_capability.rs new file mode 100644 index 00000000..30f8bbaa --- /dev/null +++ b/control_plane/src/optimizer/cost/sketch_capability.rs @@ -0,0 +1,265 @@ +//! Per-sketch performance / capability profile — the optimizer's cost-model +//! surface. +//! +//! Moved out of `sketch_algebra::capability` (Stage 4 of the +//! `promql_utilities` retirement / `sketch_algebra` re-layering): this is a +//! cost-model concern (insert/query throughput, memory, CPU, transmission +//! size, which intents each sketch family serves), read by the optimizer +//! for cost-based plan rewriting and by the physical planner to check +//! whether a sketch fits within a stage's budget — it was never L4 IR, just +//! filed alongside it because both modules touched `SketchKind`. +//! +//! Distinct from [`crate::sketch_algebra::schema::SketchStateMetadata`] — +//! that struct carries the **L4 type-system flags** (`mergeable` / +//! `subtractable` / `deletable`) that gate `SketchMerge` / `SketchSubtract` +//! / `SketchDelete` at plan-time. `SketchCapability` here is the +//! **perf / feasibility / intent-routing** profile consumed by the cost +//! model and the optimizer's binding rules — read at every plan-rewrite +//! call site, whereas `SketchStateMetadata` is sealed onto each +//! `PhysicalExpr` edge once the binding rule fires. + +use std::collections::HashMap; + +use serde::{Deserialize, Serialize}; + +use crate::sketch_algebra::params::SketchKind; + +/// Performance and capability profile for a single sketch family. +/// +/// Used by the optimizer to compare candidates and by the physical +/// planner to check whether a sketch fits within a stage's budget. +/// Populated from compiled-in defaults via [`default_capability_table`] +/// or overridden at runtime via [`load_capability_overrides`]. +#[derive(Debug, Clone)] +pub struct SketchCapability { + /// Insertion throughput (samples/sec at 1 core). + pub insert_throughput: f64, + /// Query throughput (queries/sec at 1 core). + pub query_throughput: f64, + /// Memory footprint per series (bytes). + pub memory_bytes_per_series: u64, + /// CPU cost per insert (µs/sample). + pub cpu_micros_per_insert: f64, + /// Transmission size per flush (bytes). + pub transmission_bytes: u64, + /// Which logical aggregation intents this sketch supports. + pub supported_intents: Vec, + /// Whether the sketch supports merge (`sketch(A∪B) = merge(sketch(A), sketch(B))`). + pub mergeable: bool, + /// Whether the sketch supports delta encoding. + pub supports_delta: bool, + /// Whether the sketch supports sliding windows natively. + pub supports_sliding_window: bool, +} + +/// A logical aggregation intent that a sketch can serve. Used in +/// [`SketchCapability::supported_intents`] to declare per-sketch +/// coverage; the optimizer reads this when deciding which family to +/// bind to an `AggIntent`. +#[derive(Debug, Clone, PartialEq, Eq)] +pub enum SupportedIntent { + Quantile, + Cardinality, + Frequency, + Extrema, +} + +// ── YAML override loader ───────────────────────────────────────────────────── + +/// YAML-serialisable capability profile (matches `sketch_capabilities.yml`). +#[derive(Debug, Clone, Deserialize, Serialize)] +struct SketchCapabilityYaml { + insert_throughput: f64, + query_throughput: f64, + memory_bytes_per_series: u64, + cpu_micros_per_insert: f64, + transmission_bytes: u64, + supported_intents: Vec, + mergeable: bool, + supports_delta: bool, + supports_sliding_window: bool, +} + +impl SketchCapabilityYaml { + fn to_capability(&self) -> SketchCapability { + let intents = self + .supported_intents + .iter() + .filter_map(|s| match s.as_str() { + "quantile" => Some(SupportedIntent::Quantile), + "cardinality" => Some(SupportedIntent::Cardinality), + "frequency" => Some(SupportedIntent::Frequency), + "extrema" => Some(SupportedIntent::Extrema), + _ => None, + }) + .collect(); + SketchCapability { + insert_throughput: self.insert_throughput, + query_throughput: self.query_throughput, + memory_bytes_per_series: self.memory_bytes_per_series, + cpu_micros_per_insert: self.cpu_micros_per_insert, + transmission_bytes: self.transmission_bytes, + supported_intents: intents, + mergeable: self.mergeable, + supports_delta: self.supports_delta, + supports_sliding_window: self.supports_sliding_window, + } + } +} + +/// YAML file structure for all sketch capabilities. Mirrors +/// `control_plane/sketch_capabilities.yml` 1:1. +#[derive(Debug, Clone, Deserialize, Serialize)] +struct SketchCapabilitiesFile { + ddsketch: SketchCapabilityYaml, + kll: SketchCapabilityYaml, + hll: SketchCapabilityYaml, + count_sketch: SketchCapabilityYaml, + count_min_sketch: SketchCapabilityYaml, +} + +/// Compiled-in capability defaults — one entry per [`SketchKind`]. +/// Replaces the per-variant `sketch_capability(SketchType)` function +/// that previously lived in `algebra/optimizer.rs`. Numerical values +/// are mirrored from the YAML so the in-process defaults match the +/// reference deployment file. +pub fn default_capability_table() -> HashMap { + let mut map = HashMap::new(); + map.insert( + SketchKind::DDSketch, + SketchCapability { + insert_throughput: 10_000_000.0, + query_throughput: 50_000_000.0, + memory_bytes_per_series: 4_096, + cpu_micros_per_insert: 0.1, + transmission_bytes: 4_096, + supported_intents: vec![SupportedIntent::Quantile, SupportedIntent::Extrema], + mergeable: true, + supports_delta: true, + supports_sliding_window: false, + }, + ); + map.insert( + SketchKind::Kll, + SketchCapability { + insert_throughput: 5_000_000.0, + query_throughput: 20_000_000.0, + memory_bytes_per_series: 8_192, + cpu_micros_per_insert: 0.2, + transmission_bytes: 8_192, + supported_intents: vec![SupportedIntent::Quantile, SupportedIntent::Extrema], + mergeable: true, + supports_delta: false, + supports_sliding_window: false, + }, + ); + map.insert( + SketchKind::Hll, + SketchCapability { + insert_throughput: 20_000_000.0, + query_throughput: 100_000_000.0, + memory_bytes_per_series: 16_384, + cpu_micros_per_insert: 0.05, + transmission_bytes: 16_384, + supported_intents: vec![SupportedIntent::Cardinality], + mergeable: true, + supports_delta: true, + supports_sliding_window: false, + }, + ); + map.insert( + SketchKind::CountSketch, + SketchCapability { + insert_throughput: 8_000_000.0, + query_throughput: 10_000_000.0, + memory_bytes_per_series: 80_000, + cpu_micros_per_insert: 0.5, + transmission_bytes: 80_000, + supported_intents: vec![SupportedIntent::Frequency], + mergeable: true, + supports_delta: true, + supports_sliding_window: false, + }, + ); + map.insert( + SketchKind::Cms, + SketchCapability { + insert_throughput: 8_000_000.0, + query_throughput: 10_000_000.0, + memory_bytes_per_series: 80_000, + cpu_micros_per_insert: 0.5, + transmission_bytes: 80_000, + supported_intents: vec![SupportedIntent::Frequency], + mergeable: true, + supports_delta: true, + supports_sliding_window: false, + }, + ); + map +} + +/// Load sketch capability overrides from a YAML file. Falls back to +/// [`default_capability_table`] if the file is missing or malformed. +/// Replaces `algebra::optimizer::load_sketch_capabilities`. +/// +/// Env var: `CONTROLLER_SKETCH_CAPABILITIES=path/to/this/file.yml`. +pub fn load_capability_overrides(path: &str) -> HashMap { + if let Ok(contents) = std::fs::read_to_string(path) { + if let Ok(file) = serde_yaml::from_str::(&contents) { + let mut map = HashMap::new(); + map.insert(SketchKind::DDSketch, file.ddsketch.to_capability()); + map.insert(SketchKind::Kll, file.kll.to_capability()); + map.insert(SketchKind::Hll, file.hll.to_capability()); + map.insert(SketchKind::CountSketch, file.count_sketch.to_capability()); + map.insert(SketchKind::Cms, file.count_min_sketch.to_capability()); + return map; + } + } + default_capability_table() +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn default_table_carries_all_five_sketch_kinds() { + let t = default_capability_table(); + assert!(t.contains_key(&SketchKind::DDSketch)); + assert!(t.contains_key(&SketchKind::Kll)); + assert!(t.contains_key(&SketchKind::Hll)); + assert!(t.contains_key(&SketchKind::Cms)); + assert!(t.contains_key(&SketchKind::CountSketch)); + } + + #[test] + fn default_table_ddsketch_serves_quantile_intent() { + let t = default_capability_table(); + let cap = t.get(&SketchKind::DDSketch).unwrap(); + assert!(cap.supported_intents.contains(&SupportedIntent::Quantile)); + assert!(cap.mergeable); + } + + #[test] + fn default_table_hll_serves_cardinality_intent() { + let t = default_capability_table(); + let cap = t.get(&SketchKind::Hll).unwrap(); + assert!(cap + .supported_intents + .contains(&SupportedIntent::Cardinality)); + } + + #[test] + fn load_overrides_missing_path_returns_defaults() { + let loaded = load_capability_overrides("/nonexistent/path/sketch_capabilities.yml"); + let defaults = default_capability_table(); + // Same set of keys, same defaults — we don't assert byte + // equality on the SketchCapability values because they don't + // impl PartialEq, but they share the same supported_intents + // set per kind. + assert_eq!(loaded.len(), defaults.len()); + for k in defaults.keys() { + assert!(loaded.contains_key(k)); + } + } +} diff --git a/control_plane/src/optimizer/engine.rs b/control_plane/src/optimizer/engine.rs index 4db922d5..e24ddaf1 100644 --- a/control_plane/src/optimizer/engine.rs +++ b/control_plane/src/optimizer/engine.rs @@ -33,7 +33,7 @@ use std::collections::HashMap; use crate::intent_algebra::agg_intent::AggIntent; use crate::intent_algebra::query_expr::{ColumnRef, Predicate, QueryExpr, SetOpKind, Source}; use crate::intent_algebra::relational::{agg_is_exact, agg_is_mergeable}; -use crate::sketch_algebra::capability::{ +use crate::optimizer::cost::sketch_capability::{ default_capability_table, load_capability_overrides, SketchCapability, }; use crate::types_v2::{AccuracyTarget, BindingName}; @@ -62,15 +62,17 @@ pub trait CostModel: Send + Sync { // ── Sketch capabilities ───────────────────────────────────────────────────── // -// Per the Step 2a consolidation, `SketchCapability` / `SupportedIntent` and -// the YAML-loader logic now live in `crate::sketch_algebra::capability`. +// Per the Step 2a consolidation (later re-homed to `optimizer::cost` in +// Stage 4 of the `sketch_algebra` re-layering — this is a cost-model +// concern, not L4 IR), `SketchCapability` / `SupportedIntent` and the +// YAML-loader logic live in `crate::optimizer::cost::sketch_capability`. // The optimizer re-exports `sketch_capability(SketchType)` and // `load_sketch_capabilities(path)` as thin shims so existing callers // (`algebra::physical`, `main.rs`) keep building while the legacy // `crate::types::SketchType` key continues to be the lookup key. /// Load sketch capabilities from a YAML file. Thin shim — the real -/// loader lives in `sketch_algebra::capability::load_capability_overrides` +/// loader lives in `optimizer::cost::sketch_capability::load_capability_overrides` /// and is keyed by `SketchKind`. This shim translates the result to the /// legacy `SketchType` key used by call sites that haven't migrated. /// @@ -88,7 +90,7 @@ pub fn load_sketch_capabilities( } /// Built-in capability profile for a known sketch type. Thin shim — -/// the real defaults live in `sketch_algebra::capability::default_capability_table`. +/// the real defaults live in `optimizer::cost::sketch_capability::default_capability_table`. pub fn sketch_capability(st: &crate::types::SketchType) -> SketchCapability { use crate::sketch_algebra::params::SketchKind; use crate::types::SketchType; diff --git a/control_plane/src/query_planning.rs b/control_plane/src/query_planning.rs index 759e4c6b..fa5ff847 100644 --- a/control_plane/src/query_planning.rs +++ b/control_plane/src/query_planning.rs @@ -6,7 +6,7 @@ //! `ASAPTierCandidate`s, each already carrying a `metric_name` and the //! `required_capability` the query needs (and an `unsupported` reason for //! the parts that only the cold tier can answer). -//! 2. [`required_sketches_for_capabilities`](crate::sketch_algebra::required_sketches_for_capabilities) +//! 2. [`required_sketches_for_capabilities`](crate::sketch_selection::required_sketches_for_capabilities) //! (slice 1) maps a capability set to the `SketchType` families that satisfy it. //! //! The result, [`QuerySetPlan`], is the warm-tier half of autonomous @@ -17,7 +17,8 @@ use std::collections::BTreeMap; use crate::asap_tier_analysis::{analyze_promql_for_asap_tier, UnsupportedReason}; -use crate::sketch_algebra::{required_sketches_for_capabilities, Capability}; +use crate::sketch_algebra::Capability; +use crate::sketch_selection::required_sketches_for_capabilities; use crate::types::SketchType; /// The sketch allocation derived for one metric across the whole query set. @@ -142,7 +143,9 @@ mod tests { .find(|m| m.metric_name == "http_requests_total_latency_ms") .expect("metric present"); // a quantile demands a quantile-family sketch - assert!(m.sketches.contains(&SketchType::DDSketch) || m.sketches.contains(&SketchType::KLL)); + assert!( + m.sketches.contains(&SketchType::DDSketch) || m.sketches.contains(&SketchType::KLL) + ); } #[test] @@ -152,7 +155,10 @@ mod tests { "count(http_requests_total)", ]); // at least the quantile metric must appear with a quantile sketch - let latency = plan.per_metric.iter().find(|m| m.metric_name == "latency_ms"); + let latency = plan + .per_metric + .iter() + .find(|m| m.metric_name == "latency_ms"); assert!(latency.is_some(), "latency_ms should be planned: {plan:?}"); } @@ -177,7 +183,12 @@ mod tests { .find(|m| m.metric_name == "m_latency") .expect("metric present"); // capabilities are de-duplicated (both queries need the same quantile cap) - assert_eq!(m.capabilities.len(), 1, "caps deduped: {:?}", m.capabilities); + assert_eq!( + m.capabilities.len(), + 1, + "caps deduped: {:?}", + m.capabilities + ); } #[test] diff --git a/control_plane/src/sketch_algebra/capability.rs b/control_plane/src/sketch_algebra/capability.rs index 2477bf5a..9f8042d7 100644 --- a/control_plane/src/sketch_algebra/capability.rs +++ b/control_plane/src/sketch_algebra/capability.rs @@ -1,21 +1,14 @@ -//! Single source of truth for capability state. +//! Query-side capability tag + the semantic `AggIntent` → `Capability` +//! dispatch bridge. //! -//! Step 2a of the architectural refactor consolidates the four overlapping -//! capability tables that previously existed in the controller — the YAML -//! at `control_plane/sketch_capabilities.yml`, the compiled-in defaults in -//! `algebra/optimizer.rs::sketch_capability`, the `SketchKind` enum in -//! `sketch_algebra/params.rs`, and the per-query `Capability` / -//! `SketchKindHandle` invented inside `asap_tier_analysis.rs` (PR #128). -//! All four collapse into this module: +//! Step 2a of the architectural refactor originally consolidated four +//! overlapping capability tables into this module. The performance / +//! cost-model half of that consolidation — [`SketchCapability`] / +//! `SupportedIntent` / `default_capability_table` / `load_capability_overrides` +//! — moved to `crate::optimizer::cost::sketch_capability` (Stage 4 of the +//! `sketch_algebra` re-layering): it's a cost-model concern read by the +//! optimizer and physical planner, not L4 IR. What's left here: //! -//! - [`SketchCapability`] / [`SupportedIntent`] — per-sketch performance -//! profile (insert / memory / CPU / transmission costs + the logical -//! intents the sketch can serve). Read by `algebra/optimizer.rs` for -//! cost-based plan rewriting and by `algebra/physical.rs` for stage -//! placement. Disambiguation: distinct from `schema.rs::SketchStateMetadata` -//! (which carries L4 type-system flags `mergeable` / `subtractable` / -//! `deletable`) — `SketchCapability` here is the perf / cost-model surface, -//! `SketchStateMetadata` is the L4 catalog-flag surface. //! - [`Capability`] / [`SketchKindHandle`] — query-side capability tag, //! used by the ASAP-tier reducer in `asap-query-engine` to dispatch //! PromQL → per-Capability sketch evaluation. @@ -24,25 +17,9 @@ //! `Capability`. The ASAP-tier analyzer is now a thin facade around //! this single function; PromQL function-name string matching lives //! only inside the lowerer. -//! - [`default_capability_table`] / [`load_capability_overrides`] — -//! compiled-in defaults + YAML override loader. Replaces the -//! `sketch_capability()` and `load_sketch_capabilities()` functions -//! that previously lived in `algebra/optimizer.rs`. -//! -//! ## Why one module -//! -//! Before Step 2a, "what can a sketch do" was duplicated four times. -//! Adding a new sketch family meant touching `params.rs`, -//! `optimizer.rs`, the YAML, and `asap_tier_analysis.rs`. After Step 2a -//! every capability fact has exactly one home — `params.rs` declares -//! the sketch families, this module declares everything else. #![allow(dead_code)] -use std::collections::HashMap; - -use serde::{Deserialize, Serialize}; - use crate::intent_algebra::agg_intent::AggIntent; use crate::sketch_algebra::params::SketchKind; use crate::types_v2::AccuracyTarget; @@ -619,213 +596,6 @@ fn is_exact(accuracy: &AccuracyTarget) -> bool { matches!(accuracy, AccuracyTarget::Exact) } -// ── Per-sketch performance / capability profile ────────────────────────────── - -/// Performance and capability profile for a single sketch family. -/// -/// Used by the optimizer to compare candidates and by the physical -/// planner to check whether a sketch fits within a stage's budget. -/// Populated from compiled-in defaults via [`default_capability_table`] -/// or overridden at runtime via [`load_capability_overrides`]. -/// -/// Distinct from -/// [`crate::sketch_algebra::schema::SketchStateMetadata`] — this struct -/// is the **perf / feasibility / intent-routing** profile consumed by -/// the cost model and the optimizer's binding rules. The schema-side -/// `SketchStateMetadata` carries the **L4 type-system flags** -/// (`mergeable` / `subtractable` / `deletable`) that gate `SketchMerge` / -/// `SketchSubtract` / `SketchDelete` at plan-time. The two have -/// different consumers and different lifecycles — `SketchCapability` -/// is read at every plan-rewrite call site; `SketchStateMetadata` -/// is sealed onto each `PhysicalExpr` edge once the binding rule fires. -#[derive(Debug, Clone)] -pub struct SketchCapability { - /// Insertion throughput (samples/sec at 1 core). - pub insert_throughput: f64, - /// Query throughput (queries/sec at 1 core). - pub query_throughput: f64, - /// Memory footprint per series (bytes). - pub memory_bytes_per_series: u64, - /// CPU cost per insert (µs/sample). - pub cpu_micros_per_insert: f64, - /// Transmission size per flush (bytes). - pub transmission_bytes: u64, - /// Which logical aggregation intents this sketch supports. - pub supported_intents: Vec, - /// Whether the sketch supports merge (`sketch(A∪B) = merge(sketch(A), sketch(B))`). - pub mergeable: bool, - /// Whether the sketch supports delta encoding. - pub supports_delta: bool, - /// Whether the sketch supports sliding windows natively. - pub supports_sliding_window: bool, -} - -/// A logical aggregation intent that a sketch can serve. Used in -/// [`SketchCapability::supported_intents`] to declare per-sketch -/// coverage; the optimizer reads this when deciding which family to -/// bind to an [`AggIntent`]. -#[derive(Debug, Clone, PartialEq, Eq)] -pub enum SupportedIntent { - Quantile, - Cardinality, - Frequency, - Extrema, -} - -// ── YAML override loader ───────────────────────────────────────────────────── - -/// YAML-serialisable capability profile (matches `sketch_capabilities.yml`). -#[derive(Debug, Clone, Deserialize, Serialize)] -struct SketchCapabilityYaml { - insert_throughput: f64, - query_throughput: f64, - memory_bytes_per_series: u64, - cpu_micros_per_insert: f64, - transmission_bytes: u64, - supported_intents: Vec, - mergeable: bool, - supports_delta: bool, - supports_sliding_window: bool, -} - -impl SketchCapabilityYaml { - fn to_capability(&self) -> SketchCapability { - let intents = self - .supported_intents - .iter() - .filter_map(|s| match s.as_str() { - "quantile" => Some(SupportedIntent::Quantile), - "cardinality" => Some(SupportedIntent::Cardinality), - "frequency" => Some(SupportedIntent::Frequency), - "extrema" => Some(SupportedIntent::Extrema), - _ => None, - }) - .collect(); - SketchCapability { - insert_throughput: self.insert_throughput, - query_throughput: self.query_throughput, - memory_bytes_per_series: self.memory_bytes_per_series, - cpu_micros_per_insert: self.cpu_micros_per_insert, - transmission_bytes: self.transmission_bytes, - supported_intents: intents, - mergeable: self.mergeable, - supports_delta: self.supports_delta, - supports_sliding_window: self.supports_sliding_window, - } - } -} - -/// YAML file structure for all sketch capabilities. Mirrors -/// `control_plane/sketch_capabilities.yml` 1:1. -#[derive(Debug, Clone, Deserialize, Serialize)] -struct SketchCapabilitiesFile { - ddsketch: SketchCapabilityYaml, - kll: SketchCapabilityYaml, - hll: SketchCapabilityYaml, - count_sketch: SketchCapabilityYaml, - count_min_sketch: SketchCapabilityYaml, -} - -/// Compiled-in capability defaults — one entry per [`SketchKind`]. -/// Replaces the per-variant `sketch_capability(SketchType)` function -/// that previously lived in `algebra/optimizer.rs`. Numerical values -/// are mirrored from the YAML so the in-process defaults match the -/// reference deployment file. -pub fn default_capability_table() -> HashMap { - let mut map = HashMap::new(); - map.insert( - SketchKind::DDSketch, - SketchCapability { - insert_throughput: 10_000_000.0, - query_throughput: 50_000_000.0, - memory_bytes_per_series: 4_096, - cpu_micros_per_insert: 0.1, - transmission_bytes: 4_096, - supported_intents: vec![SupportedIntent::Quantile, SupportedIntent::Extrema], - mergeable: true, - supports_delta: true, - supports_sliding_window: false, - }, - ); - map.insert( - SketchKind::Kll, - SketchCapability { - insert_throughput: 5_000_000.0, - query_throughput: 20_000_000.0, - memory_bytes_per_series: 8_192, - cpu_micros_per_insert: 0.2, - transmission_bytes: 8_192, - supported_intents: vec![SupportedIntent::Quantile, SupportedIntent::Extrema], - mergeable: true, - supports_delta: false, - supports_sliding_window: false, - }, - ); - map.insert( - SketchKind::Hll, - SketchCapability { - insert_throughput: 20_000_000.0, - query_throughput: 100_000_000.0, - memory_bytes_per_series: 16_384, - cpu_micros_per_insert: 0.05, - transmission_bytes: 16_384, - supported_intents: vec![SupportedIntent::Cardinality], - mergeable: true, - supports_delta: true, - supports_sliding_window: false, - }, - ); - map.insert( - SketchKind::CountSketch, - SketchCapability { - insert_throughput: 8_000_000.0, - query_throughput: 10_000_000.0, - memory_bytes_per_series: 80_000, - cpu_micros_per_insert: 0.5, - transmission_bytes: 80_000, - supported_intents: vec![SupportedIntent::Frequency], - mergeable: true, - supports_delta: true, - supports_sliding_window: false, - }, - ); - map.insert( - SketchKind::Cms, - SketchCapability { - insert_throughput: 8_000_000.0, - query_throughput: 10_000_000.0, - memory_bytes_per_series: 80_000, - cpu_micros_per_insert: 0.5, - transmission_bytes: 80_000, - supported_intents: vec![SupportedIntent::Frequency], - mergeable: true, - supports_delta: true, - supports_sliding_window: false, - }, - ); - map -} - -/// Load sketch capability overrides from a YAML file. Falls back to -/// [`default_capability_table`] if the file is missing or malformed. -/// Replaces `algebra::optimizer::load_sketch_capabilities`. -/// -/// Env var: `CONTROLLER_SKETCH_CAPABILITIES=path/to/this/file.yml`. -pub fn load_capability_overrides(path: &str) -> HashMap { - if let Ok(contents) = std::fs::read_to_string(path) { - if let Ok(file) = serde_yaml::from_str::(&contents) { - let mut map = HashMap::new(); - map.insert(SketchKind::DDSketch, file.ddsketch.to_capability()); - map.insert(SketchKind::Kll, file.kll.to_capability()); - map.insert(SketchKind::Hll, file.hll.to_capability()); - map.insert(SketchKind::CountSketch, file.count_sketch.to_capability()); - map.insert(SketchKind::Cms, file.count_min_sketch.to_capability()); - return map; - } - } - default_capability_table() -} - // ── Sketch-family error bounds ─────────────────────────────────────────────── // // These two helpers were originally defined in `controller/src/algebra/expr.rs` @@ -1283,35 +1053,6 @@ mod tests { assert!(!required_cms.is_satisfied_by(&cap)); } - // ── default_capability_table ───────────────────────────────────────── - - #[test] - fn default_table_carries_all_five_sketch_kinds() { - let t = default_capability_table(); - assert!(t.contains_key(&SketchKind::DDSketch)); - assert!(t.contains_key(&SketchKind::Kll)); - assert!(t.contains_key(&SketchKind::Hll)); - assert!(t.contains_key(&SketchKind::Cms)); - assert!(t.contains_key(&SketchKind::CountSketch)); - } - - #[test] - fn default_table_ddsketch_serves_quantile_intent() { - let t = default_capability_table(); - let cap = t.get(&SketchKind::DDSketch).unwrap(); - assert!(cap.supported_intents.contains(&SupportedIntent::Quantile)); - assert!(cap.mergeable); - } - - #[test] - fn default_table_hll_serves_cardinality_intent() { - let t = default_capability_table(); - let cap = t.get(&SketchKind::Hll).unwrap(); - assert!(cap - .supported_intents - .contains(&SupportedIntent::Cardinality)); - } - // ── OuterAgg fold semantics ────────────────────────────────────────── // // Pin the fold-by-operator dispatch shape the engine reads off the @@ -1411,20 +1152,4 @@ mod tests { assert_eq!(OuterAgg::Avg(vec![]).fold(&[]), None); assert_eq!(OuterAgg::Count(vec![]).fold(&[]), None); } - - // ── load_capability_overrides ──────────────────────────────────────── - - #[test] - fn load_overrides_missing_path_returns_defaults() { - let loaded = load_capability_overrides("/nonexistent/path/sketch_capabilities.yml"); - let defaults = default_capability_table(); - // Same set of keys, same defaults — we don't assert byte - // equality on the SketchCapability values because they don't - // impl PartialEq, but they share the same supported_intents - // set per kind. - assert_eq!(loaded.len(), defaults.len()); - for k in defaults.keys() { - assert!(loaded.contains_key(k)); - } - } } diff --git a/control_plane/src/sketch_algebra/mod.rs b/control_plane/src/sketch_algebra/mod.rs index 2641ca40..d086e9c1 100644 --- a/control_plane/src/sketch_algebra/mod.rs +++ b/control_plane/src/sketch_algebra/mod.rs @@ -9,9 +9,6 @@ //! - [`SketchKind`] / [`SketchParams`] — typed sketch-family selector + //! parameter payload. Convertible to the legacy `crate::types` //! shape via [`SketchParams::to_legacy`] for the L5 emitter side. -//! - [`SketchStateSchema`] — the L4 type-system primitive that mirrors -//! the §6.4 input/output spec table (`Sketch(SketchKind, SketchParams)` -//! field type + catalog-capability flags). //! - [`bind_query_expr`] — the L3→L4 lowering driver: bottom-up walk //! that fires `Bind*` rules. //! - [`rules`] — the `Bind*` rule family. Each rule pattern-matches on a @@ -36,9 +33,7 @@ pub mod capability_matching; pub mod lower; pub mod physical_expr; pub mod rules; -pub mod schema; pub mod sketch_params; -pub mod sketch_selection; // Back-compat alias. External call sites that imported // `control_plane::sketch_algebra::params::*` (and the in-tree @@ -50,19 +45,12 @@ pub use sketch_params as params; mod tests; // Re-exports — `crate::sketch_algebra::*` for downstream callers. -pub use capability::{ - capability_for, default_capability_table, load_capability_overrides, Capability, - SketchCapability, SketchKindHandle, SupportedIntent, -}; +pub use capability::{capability_for, Capability, SketchKindHandle}; pub use capability_matching::{ classify_demo_metric, is_valid_pair, pick_family, AccuracyPreference, StatisticClass, }; pub use lower::{bind_query_expr, BindingError}; pub use physical_expr::{EstimateOp, MergeAlgebra, PhysicalExpr}; -pub use schema::{SketchStateMetadata, SketchStateSchema}; -pub use sketch_selection::{ - required_sketches_for_capabilities, sketch_families_for_capability, sketch_type_for_handle, -}; pub use sketch_params::{ CmsParams, CountSketchParams, DDSketchParams, HllParams, KllParams, SketchKind, SketchParams, }; diff --git a/control_plane/src/sketch_algebra/schema.rs b/control_plane/src/sketch_algebra/schema.rs deleted file mode 100644 index d35e3d88..00000000 --- a/control_plane/src/sketch_algebra/schema.rs +++ /dev/null @@ -1,181 +0,0 @@ -//! Layer 4 sketch-state schema. -//! -//! Per `control_plane/docs/design.md` §6.4 ("Per-node input/output spec for -//! `PhysicalExpr`", around line ~618). L4 introduces a new field type into -//! `Schema`: `DataType::Sketch(SketchKind, SketchParams)`. This module -//! defines that extension as a parallel typed layer that the L4 type -//! checker consults. -//! -//! Keeping this in a parallel struct (rather than mutating the L3 -//! `intent_algebra::DataType`) avoids touching the recently-shipped L3 -//! IR. The L4 type checker asks: "for this `PhysicalExpr` node, what is -//! the input sketch-state schema, and what is the output?" Each `Bind*` -//! rule populates an [`SketchStateSchema`] when it produces a -//! sketch-state-bearing node. -//! -//! Local-checkable invariants (per design.md §6.4 line ~643): -//! 1. **Sketch-family mismatch is a plan-time error.** A `SketchMerge` -//! over inputs with mismatched `(kind, params)` fails at L4. -//! 2. **Catalog capability flags gate which nodes can fire.** The -//! capability flags live on the [`SketchStateSchema`] so the type -//! checker has them locally without re-consulting the catalog. - -#![allow(dead_code)] - -use serde::{Deserialize, Serialize}; - -use crate::sketch_algebra::params::{SketchKind, SketchParams}; - -/// Sketch-state schema annotation on a [`crate::sketch_algebra::PhysicalExpr`] -/// edge. Carries the family + params + the catalog-capability flags so -/// the L4 type checker can decide locally whether a downstream node may -/// merge / subtract / delete from the state. -/// -/// Mirrors the per-node input/output spec table in design.md §6.4: a -/// `SketchAgg` produces a sketch-state schema; a `SketchEstimate` -/// consumes one and emits a regular row schema; a `SketchMerge` consumes -/// N matching sketch-state schemas and emits one of the same family. -#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)] -pub struct SketchStateSchema { - /// Sketch family of the state. - pub kind: SketchKind, - /// Parameter payload — must match across all inputs to a `SketchMerge`. - pub params: SketchParams, - /// Capability flags from the sketch catalog. - pub caps: SketchStateMetadata, -} - -/// L4 type-system catalog flags for a sketch state. Populated from the -/// sketch catalog at `Bind*`-rule time. `mergeable` gates `SketchMerge`; -/// `subtractable` gates `SketchSubtract`; `deletable` gates -/// `SketchDelete`. See design.md §6 line ~646 ("catalog is the single -/// source of truth for these flags; binding rules consult it before -/// producing the node"). -/// -/// Renamed from `SketchCapabilities` in May 2026 to disambiguate from -/// [`crate::sketch_algebra::capability::SketchCapability`] (perf / -/// cost-model profile). The two structs live side-by-side: this one is -/// the **L4 type-system / plan-time** surface — sealed onto every -/// `PhysicalExpr` edge by the binding rule and consulted by the type -/// checker. `SketchCapability` is the **perf / feasibility / intent- -/// routing** surface — consumed by the optimizer and cost model. -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -pub struct SketchStateMetadata { - /// Whether two states of this family + params can be unioned — - /// catalog default for KLL / HLL / DDSketch / CMS / CountSketch. - pub mergeable: bool, - /// Whether `SketchSubtract` is meaningful for this family — true for - /// CMS / count-based sketches, false for KLL / HLL / DDSketch. - pub subtractable: bool, - /// Whether `SketchDelete` is meaningful — true for deletable Bloom - /// filters and CMS (with -1 update); false for KLL / HLL / DDSketch. - pub deletable: bool, -} - -impl SketchStateSchema { - /// Construct the catalog-default schema for a given family + params. - /// Capability flags follow the design.md §6 catalog defaults; if a - /// future catalog reorder changes them, this is the single point of - /// truth that downstream rules must consult. - pub fn for_kind(kind: SketchKind, params: SketchParams) -> Self { - let caps = match kind { - SketchKind::Kll => SketchStateMetadata { - mergeable: true, - subtractable: false, - deletable: false, - }, - SketchKind::DDSketch => SketchStateMetadata { - mergeable: true, - subtractable: false, - deletable: false, - }, - SketchKind::Hll => SketchStateMetadata { - mergeable: true, - subtractable: false, - deletable: false, - }, - SketchKind::Cms => SketchStateMetadata { - mergeable: true, - subtractable: true, - deletable: true, - }, - SketchKind::CountSketch => SketchStateMetadata { - mergeable: true, - subtractable: true, - deletable: false, - }, - }; - SketchStateSchema { kind, params, caps } - } - - /// Whether two sketch-state schemas can be `SketchMerge`-d. Per - /// design.md §6.4 invariant 1: the family + params must match - /// exactly, and the family must be `mergeable`. - pub fn is_compatible_for_merge(&self, other: &Self) -> bool { - self.kind == other.kind && self.params == other.params && self.caps.mergeable - } - - /// Whether two sketch-state schemas can be `SketchSubtract`-ed. - pub fn is_compatible_for_subtract(&self, other: &Self) -> bool { - self.kind == other.kind && self.params == other.params && self.caps.subtractable - } -} - -// ── Tests ───────────────────────────────────────────────────────────────────── - -#[cfg(test)] -mod tests { - use super::*; - use crate::sketch_algebra::params::{CmsParams, KllParams}; - - #[test] - fn kll_default_caps() { - let s = - SketchStateSchema::for_kind(SketchKind::Kll, SketchParams::Kll(KllParams { k: 200 })); - assert!(s.caps.mergeable); - assert!(!s.caps.subtractable); - assert!(!s.caps.deletable); - } - - #[test] - fn cms_supports_subtract_and_delete() { - let s = SketchStateSchema::for_kind( - SketchKind::Cms, - SketchParams::Cms(CmsParams { - w: 2048, - d: 5, - with_heap: false, - }), - ); - assert!(s.caps.mergeable); - assert!(s.caps.subtractable); - assert!(s.caps.deletable); - } - - #[test] - fn merge_compatibility_requires_matching_params() { - let a = - SketchStateSchema::for_kind(SketchKind::Kll, SketchParams::Kll(KllParams { k: 200 })); - let b = - SketchStateSchema::for_kind(SketchKind::Kll, SketchParams::Kll(KllParams { k: 200 })); - let c = - SketchStateSchema::for_kind(SketchKind::Kll, SketchParams::Kll(KllParams { k: 400 })); - assert!(a.is_compatible_for_merge(&b)); - assert!(!a.is_compatible_for_merge(&c)); // different k - } - - #[test] - fn merge_compatibility_rejects_family_mismatch() { - let kll = - SketchStateSchema::for_kind(SketchKind::Kll, SketchParams::Kll(KllParams { k: 200 })); - let cms = SketchStateSchema::for_kind( - SketchKind::Cms, - SketchParams::Cms(CmsParams { - w: 2048, - d: 5, - with_heap: false, - }), - ); - assert!(!kll.is_compatible_for_merge(&cms)); - } -} diff --git a/control_plane/src/sketch_algebra/sketch_selection.rs b/control_plane/src/sketch_selection.rs similarity index 87% rename from control_plane/src/sketch_algebra/sketch_selection.rs rename to control_plane/src/sketch_selection.rs index 93ca2571..c7a7df3f 100644 --- a/control_plane/src/sketch_algebra/sketch_selection.rs +++ b/control_plane/src/sketch_selection.rs @@ -1,6 +1,6 @@ -//! Inverse of [`capability_for`](super::capability::capability_for): given a -//! required [`Capability`], enumerate the concrete [`SketchType`] families that -//! can satisfy it. +//! Inverse of [`capability_for`](crate::sketch_algebra::capability::capability_for): +//! given a required [`Capability`], enumerate the concrete [`SketchType`] +//! families that can satisfy it. //! //! This is the keystone of autonomous allocation. `capability_for` lowers an //! `AggIntent` (parsed from a query) to the *capability* it needs; this module @@ -9,7 +9,7 @@ //! without a hand-written workload YAML. //! //! Policy encoded here (matches `Capability::is_satisfied_by` on the matching -//! side, capability.rs): +//! side, `sketch_algebra::capability`): //! * `QuantileApprox` → DDSketch | KLL //! * `CardinalityApprox`→ HLL //! * `FrequencyEstimate`→ CountSketch | CountMinSketch (heap-less point query) @@ -18,8 +18,12 @@ //! //! When a capability binds a *concrete* `SketchKindHandle` (not `Any`), the //! result is exactly that one family; `Any` expands to the full candidate set. +//! +//! Moved out of `sketch_algebra` (Stage 4 of the `sketch_algebra` +//! re-layering) — this is a query-planning concern (its one caller is +//! [`crate::query_planning`]), not L4 IR. -use super::capability::{Capability, SketchKindHandle}; +use crate::sketch_algebra::capability::{Capability, SketchKindHandle}; use crate::types::SketchType; /// Map a sketch handle to the allocatable control-plane [`SketchType`]. @@ -52,9 +56,7 @@ pub fn sketch_type_for_handle(h: SketchKindHandle) -> Option { /// no-op (the caller routes it to cold/exact instead). pub fn sketch_families_for_capability(cap: &Capability) -> Vec { match cap { - Capability::QuantileApprox(h) => { - concrete_or(*h, &[SketchType::DDSketch, SketchType::KLL]) - } + Capability::QuantileApprox(h) => concrete_or(*h, &[SketchType::DDSketch, SketchType::KLL]), Capability::CardinalityApprox => vec![SketchType::HLL], Capability::FrequencyEstimate(h) | Capability::FrequencyTopk(h) => { concrete_or(*h, &[SketchType::CountSketch, SketchType::CountMinSketch]) @@ -96,18 +98,15 @@ mod tests { #[test] fn quantile_any_expands_to_ddsketch_and_kll() { - let fams = sketch_families_for_capability(&Capability::QuantileApprox( - SketchKindHandle::Any, - )); + let fams = + sketch_families_for_capability(&Capability::QuantileApprox(SketchKindHandle::Any)); assert_eq!(fams, vec![SketchType::DDSketch, SketchType::KLL]); } #[test] fn quantile_concrete_handle_pins_one_family() { assert_eq!( - sketch_families_for_capability(&Capability::QuantileApprox( - SketchKindHandle::DDSketch - )), + sketch_families_for_capability(&Capability::QuantileApprox(SketchKindHandle::DDSketch)), vec![SketchType::DDSketch] ); assert_eq!( @@ -128,9 +127,7 @@ mod tests { fn frequency_families_and_heap_collapse() { // bare frequency: Any -> both matrix families assert_eq!( - sketch_families_for_capability(&Capability::FrequencyEstimate( - SketchKindHandle::Any - )), + sketch_families_for_capability(&Capability::FrequencyEstimate(SketchKindHandle::Any)), vec![SketchType::CountSketch, SketchType::CountMinSketch] ); // heap-bearing handles collapse to their matrix family @@ -151,8 +148,7 @@ mod tests { #[test] fn exact_agg_allocates_no_sketch() { assert!( - sketch_families_for_capability(&Capability::ExactAgg(AggregationType::Sum)) - .is_empty() + sketch_families_for_capability(&Capability::ExactAgg(AggregationType::Sum)).is_empty() ); }