diff --git a/controller/docs/design.md b/controller/docs/design.md index 28f35ec0..80bab374 100644 --- a/controller/docs/design.md +++ b/controller/docs/design.md @@ -43,11 +43,11 @@ DataCollector/controller already documents its query→sketch translation as a 5 | # | Layer | What it does | Today's locations | |---|-------|--------------|-------------------| -| 1 | **Query Language** | Parse raw strings (PromQL, SQL, DataFusion, ElasticDSL, …) into a language-specific AST | DC `controller/src/query_parser/{promql,sql,mod}.rs`; asap-planner-rs pulls `promql-parser` + `sqlparser` directly; asap-fusion consumes a pre-built DataFusion `LogicalPlan` (its L1 happens upstream) | -| 2 | **Language Logical Plan** | Per-language algebra tree (`Aggregate` / `Window` / `Filter` / `Sort` / `Limit`) preserving language semantics, **no sketch names, no sketch binding** | DC `controller/src/algebra/lower.rs`; asap-fusion inherits DataFusion's `LogicalPlan` as its L2; asap-planner-rs has no L2 today (uses a template-pattern catalogue) — **Phase 4 builds one** | -| 3 | **Intent algebra** | Language- and deployment-independent IR: `QueryExpr` + `AggIntent`. Describes **intent only** — *what* to compute, with accuracy target. **No sketch type, no sketch parameters, no sketch-bound nodes** (`SketchAgg` / `SketchJoin` / `SketchSubtract` etc. live in the L4 IR `SketchExpr`). **No language-shaped operators** (no `HistogramQuantile`, no `PromQLSubquery` — those are PromQL L2 nodes that lower to data-model-agnostic shapes here). **One canonical form per plan** — no `WindowedAgg` (use `Window` over `Aggregate`). Heavy-hitter intents are first-class (`AggIntent::TopK`) so heavy-hitter sketches bind directly on the intent rather than on a generic `Sort + Limit` shape; generic `Sort + Limit` survives in `QueryExpr` for non-heavy-hitter cases (e.g. `ORDER BY name LIMIT 10`). Every edge carries a typed `Schema`. Data-model-agnostic — `QueryExpr::Scan` wraps a `Source` sum with `TimeSeries` / `Table` / `Join` variants so the same L3 IR covers ASAPQuery's time-series queries and asap-fusion's tabular queries. | DC `controller/src/algebra/{expr,directory}.rs`; asap-fusion's 3-variant `SubPopulationAnalyticsType` maps to a subset; asap-planner-rs's 9-variant `Statistic` maps to a subset — **Phase 4 splits sketch binding out of planner's current fused L3+L4** | -| 4 | **Sketch algebra + optimizer** | Cost-aware algebraic rewrite rules under deployment constraints. **This is where sketch binding happens** — L4 rules take intent-only L3 (`QueryExpr`) and emit the sketch-bound IR (`SketchExpr`). ~12 rules in DC; a smaller targeted subset in planner; `SketchConfigRule` + `HashModeRule` in fusion. | Core provides the **rule engine driver** + `OptimizerRule` trait + a shared rule library + the sketch-bound IR `core::sketch_algebra::SketchExpr`; deployment models **pick** which rules to enable + supply their own deployment constraints. DC `controller/src/algebra/optimizer.rs` (rules); fusion `src/optimizer/rules/`; planner's `map_statistic_to_precompute_operator` | -| 5 | **Physical Execution Plan** | Assign ops to pipeline stages (edge / gateway / backend / object store); produce the deployment-specific artifact (OpAMP YAML, `streaming_config.yaml`, rewritten DataFusion `LogicalPlan`). **Sketch binding is already committed by L4**; L5 is about stage allocation + emission. | Core provides the **stage allocator framework** + `PhysicalPlanner` trait + the sketch catalogue; deployment models supply their own **topology** (3-stage / 1-stage / 0-stage) + their own **emitter** for the output format. DC `controller/src/algebra/{physical,allocator,plan}.rs`; asap-planner-rs `output/generator.rs`; asap-fusion `src/executor/` | +| 1 | **Query Language** | Parse raw strings (PromQL, SQL, DataFusion, ElasticDSL, …) into a language-specific AST | DC `controller/src/query_parser/{promql,sql}.rs` + the per-language façade `controller/src/query_parser/language/{promql,sql,elastic_dsl}/`; asap-planner-rs pulls `promql-parser` + `sqlparser` directly; asap-fusion consumes a pre-built DataFusion `LogicalPlan` (its L1 happens upstream). *Refactor 2026-05 absorbed `controller/src/query_language/` into `query_parser::language/`.* | +| 2 | **Language Logical Plan** | Per-language algebra tree (`Aggregate` / `Window` / `Filter` / `Sort` / `Limit`) preserving language semantics, **no sketch names, no sketch binding** | DC `controller/src/language_logical_plan/{lower,plan}.rs` + `controller/src/intent_algebra/legacy_lower.rs` (legacy L2→L3 lowering pending unification with `intent_algebra/lower.rs`); asap-fusion inherits DataFusion's `LogicalPlan` as its L2; asap-planner-rs has no L2 today (uses a template-pattern catalogue) — **Phase 4 builds one** | +| 3 | **Intent algebra** | Language- and deployment-independent IR: `QueryExpr` + `AggIntent`. Describes **intent only** — *what* to compute, with accuracy target. **No sketch type, no sketch parameters, no sketch-bound nodes** (`SketchAgg` / `SketchJoin` / `SketchSubtract` etc. live in the L4 IR `SketchExpr`). **No language-shaped operators** (no `HistogramQuantile`, no `PromQLSubquery` — those are PromQL L2 nodes that lower to data-model-agnostic shapes here). **One canonical form per plan** — no `WindowedAgg` (use `Window` over `Aggregate`). Heavy-hitter intents are first-class (`AggIntent::TopK`) so heavy-hitter sketches bind directly on the intent rather than on a generic `Sort + Limit` shape; generic `Sort + Limit` survives in `QueryExpr` for non-heavy-hitter cases (e.g. `ORDER BY name LIMIT 10`). Every edge carries a typed `Schema`. Data-model-agnostic — `QueryExpr::Scan` wraps a `Source` sum with `TimeSeries` / `Table` / `Join` variants so the same L3 IR covers ASAPQuery's time-series queries and asap-fusion's tabular queries. | DC `controller/src/intent_algebra/{agg_intent,query_expr,schema,lower,cse}.rs` (canonical L3) + `controller/src/intent_algebra/legacy_expr.rs` (legacy heavy-IR pending dedup with the canonical types); asap-fusion's 3-variant `SubPopulationAnalyticsType` maps to a subset; asap-planner-rs's 9-variant `Statistic` maps to a subset — **Phase 4 splits sketch binding out of planner's current fused L3+L4**. *Refactor 2026-05 absorbed `controller/src/algebra/expr.rs` into `intent_algebra/legacy_expr.rs`.* | +| 4 | **Sketch algebra + optimizer** | Cost-aware algebraic rewrite rules under deployment constraints. **This is where sketch binding happens** — L4 rules take intent-only L3 (`QueryExpr`) and emit the sketch-bound IR (`SketchExpr`). ~12 rules in DC; a smaller targeted subset in planner; `SketchConfigRule` + `HashModeRule` in fusion. | Core provides the **rule engine driver** + `OptimizerRule` trait + a shared rule library + the sketch-bound IR `core::sketch_algebra::SketchExpr`; deployment models **pick** which rules to enable + supply their own deployment constraints. DC `controller/src/sketch_algebra/` (IR + binding rules) + `controller/src/optimizer/{engine,trait_def,baseline,rules/,cost/}.rs` (rule engine + cost models); fusion `src/optimizer/rules/`; planner's `map_statistic_to_precompute_operator`. *Refactor 2026-05 absorbed `controller/src/algebra/optimizer.rs` (→ `optimizer/engine.rs`) and `controller/src/planner/{cost_model,delta_cost_model,online_cost_model,pareto,tco,wire_cost,rules,baseline_planner}.rs` (→ `optimizer/{cost/,rules/,baseline.rs}`).* | +| 5 | **Physical Execution Plan** | Assign ops to pipeline stages (edge / gateway / backend / object store); produce the deployment-specific artifact (OpAMP YAML, `streaming_config.yaml`, rewritten DataFusion `LogicalPlan`). **Sketch binding is already committed by L4**; L5 is about stage allocation + emission. | Core provides the **stage allocator framework** + `PhysicalPlanner` trait + the sketch catalogue; deployment models supply their own **topology** (3-stage / 1-stage / 0-stage) + their own **emitter** for the output format. DC `controller/src/physical/{allocator,planner,plan,sketch_catalog,stage_split,topology,colored_dag/}.rs` (allocator framework + sketch catalogue + typed three-stage colouring) + `controller/src/emit/{stage_config,otap,telegraf,agent,backend,asapquery_backend,precompute,trait_def}.rs` (per-deployment-model emitters + `PlanEmitter` trait) + `controller/src/pipeline.rs` (L1→…→L5 driver, formerly `analyzer.rs`); asap-planner-rs `output/generator.rs`; asap-fusion `src/executor/`. *Refactor 2026-05 absorbed `controller/src/algebra/{physical,allocator,plan,directory}.rs` and `controller/src/planner/stage_split.rs` and the legacy `controller/src/stage_split/` framework into `physical/`; absorbed `controller/src/config/` into `emit/` + `workload.rs`; renamed `analyzer.rs` to `pipeline.rs`.* | **The doc's key claim: layers 1–3 are query-language-independent and workload-independent.** That makes them the natural **common core**. Every deployment model reads PromQL (or SQL, or …) the same way, lowers it to the same per-language algebra, and lowers THAT to the same intent-algebra IR (intent only, no sketch binding). @@ -159,6 +159,32 @@ If something must be optional (e.g. OpAMP for deployments that don't run OTel co ## 5. Target repo layout +> **Refactor 2026-05 — current state vs target.** The per-deployment-model crate +> split below (`crates/deployment-model-{asaplifecycle,asapquery,asapfusion}/`) +> is **deferred** until ≥2 deployment models actually ship. Today's +> implementation is a single `controller/` crate whose internal module +> structure mirrors design.md §5's modular split: +> +> | Target §5 path | Current single-crate path | +> |---|---| +> | `crates/core/query_language/` | `controller/src/query_parser/language/` | +> | `crates/core/logical_plan/` | `controller/src/language_logical_plan/` | +> | `crates/core/intent_algebra/` | `controller/src/intent_algebra/` (with `legacy_expr` / `legacy_lower` carrying the older `algebra::expr`-flavored IR pending unification) | +> | `crates/core/sketch_algebra/` | `controller/src/sketch_algebra/` | +> | `crates/core/optimizer/{engine,trait,rules,cost}/` | `controller/src/optimizer/{engine.rs,trait_def.rs,rules/,cost/,baseline.rs}` | +> | `crates/core/physical/{planner_trait,stage_allocator,topology,executor,sketch_catalog}/` | `controller/src/physical/{planner,allocator,plan,stage_split,sketch_catalog,topology,colored_dag/}.rs` | +> | `crates/core/pipeline/` | `controller/src/pipeline.rs` (single-file driver — was `analyzer.rs`) | +> | `crates/core/workload/` | `controller/src/workload.rs` | +> | `crates/core/emit/` | `controller/src/emit/` | +> | `crates/core/registry/` | `controller/src/deployment_model.rs` | +> | `crates/runtime/{http,opamp,monitor,replan,store,backend_client}/` | `controller/src/{main.rs,opamp/,monitor/,replan.rs,store/,backend_client.rs,metrics_exposer.rs}` | +> | `crates/deployment-model-asaplifecycle/` | folded into `controller/src/emit/{agent,backend,asapquery_backend,otap,telegraf,stage_config,precompute}.rs` + the three-stage topology in `controller/src/physical/colored_dag/` | +> +> The target multi-crate layout below remains the long-term goal but is **not a +> current migration**. The single-crate is structurally aligned to the §5 +> module boundaries so future extraction is largely `git mv` + `Cargo.toml` +> edits. + ``` ASAPController/ ├── Cargo.toml # workspace @@ -251,6 +277,40 @@ Mashing them into one crate means the union of all their dependencies (sqlparser Core is not a trait-stubs library. It ships real L1/L2/L3 code lifted from DC's `controller/src/query_parser/` + `controller/src/algebra/` and exposes a small set of traits for L4/L5 plugin points. +> **§6 / `core::*` paths vs current single-crate paths.** §6 below names +> targets like `core::query_language`, `core::sketch_algebra`, +> `core::optimizer` from the design.md §5 target layout. Today's +> single-crate (refactor 2026-05) is structurally aligned to those names +> via a 1:1 mapping. The lookup table: +> +> | `core::*` reference in §6 | Current single-crate path | +> |---|---| +> | `core::query_language` | `controller/src/query_parser/language/` | +> | `core::logical_plan` | `controller/src/language_logical_plan/` | +> | `core::intent_algebra` | `controller/src/intent_algebra/` (canonical) + `controller/src/intent_algebra/legacy_{expr,lower}.rs` (legacy IR pending unification) | +> | `core::sketch_algebra` | `controller/src/sketch_algebra/` | +> | `core::lower` | per-layer: `controller/src/language_logical_plan/lower.rs` + `controller/src/intent_algebra/{lower,legacy_lower}.rs` + `controller/src/sketch_algebra/lower.rs` | +> | `core::optimizer::engine` | `controller/src/optimizer/engine.rs` | +> | `core::optimizer::trait` | `controller/src/optimizer/trait_def.rs` (placeholder) | +> | `core::optimizer::rules` | `controller/src/optimizer/rules/` | +> | `core::optimizer::cost` | `controller/src/optimizer/cost/` | +> | `core::physical::planner_trait` | `controller/src/physical/planner.rs` (legacy `physical_plan_to_staged` impl pending trait extraction) | +> | `core::physical::stage_allocator` | `controller/src/physical/{allocator,stage_split,colored_dag/allocator}.rs` | +> | `core::physical::topology` | `controller/src/physical/topology.rs` | +> | `core::physical::executor` | not yet materialised (placeholder absent) | +> | `core::physical::sketch_catalog` | `controller/src/physical/sketch_catalog.rs` | +> | `core::pipeline` | `controller/src/pipeline.rs` (single file — was `analyzer.rs`) | +> | `core::workload` | `controller/src/workload.rs` | +> | `core::emit` | `controller/src/emit/` | +> | `core::registry` | `controller/src/deployment_model.rs` | +> | `core::telemetry` | `controller/src/metrics_exposer.rs` (out-of-core today — drives Prometheus metrics) | +> +> Every §6 reference below should be read with this table as the +> concrete pointer. Where §6 says "lifted from DC's +> `controller/src/algebra/`", read "structurally aligned to +> `intent_algebra/legacy_expr.rs` + `physical/` + `optimizer/engine.rs` +> per the §16 ADR." + ### `core::query_language` — Layer 1 Per-language parsers, one module each: @@ -1720,3 +1780,65 @@ The migration is done when: 4. `asap-fusion`'s microbenchmarks still run under `deployment-model-asapfusion` with identical numbers. 5. The `DataCollector/controller/`, `ASAPQuery/asap-planner-rs/`, `ASAPQuery-backend/asap-planner-rs/`, and `asap-fusion/` directories are deletable (or already deleted) without breaking any currently-running deployment. 6. A new hypothetical deployment model can be added with zero changes outside its crate + one line in `bin/asap-controller/main.rs`. + +## 16. ADR — Capability consolidation + `algebra/`/`planner/`/`config/` retirement (2026-05) + +This section records two migration steps that brought the controller into its current shape. + +### 16.1 PR #129 — Four-merged-then-cleaned capability tables + +Before PR #129 there were **four** overlapping capability tables in the controller: + +| Source | Location | Shape | +|---|---|---| +| YAML at `controller/sketch_capabilities.yml` | filesystem | per-sketch perf profile, runtime-loaded | +| Compiled-in defaults | `algebra/optimizer.rs::sketch_capability` | per-sketch perf profile, hard-coded | +| `SketchKind` enum | `sketch_algebra/params.rs` | sketch type tag | +| `Capability` / `SketchKindHandle` | `warm_tier_analysis.rs` (PR #128) | query-side dispatch tag invented per-query | + +All four collapsed into one module: `controller/src/sketch_algebra/capability.rs`. The four-way map is now: + +- `SketchCapability` + `SupportedIntent` — per-sketch perf profile, read by L4 cost model + L5 physical planner (formerly the YAML + compiled-in copies). +- `Capability` + `SketchKindHandle` — query-side capability tag, used by `asap-query-engine`'s warm-tier reducer (formerly the per-query invention in PR #128). +- `capability_for(intent: &AggIntent) -> Option` — the **semantic** intent → warm-tier dispatch bridge. The new signature replaces the older `capability_for(query_func: &str)` string-keyed lookup. PromQL → `intent_algebra::lower` → `AggIntent` → (this fn) → `Capability`. The warm-tier analyzer at `controller/src/warm_tier_analysis.rs` is now a thin facade around this single function. +- `default_capability_table()` / `load_capability_overrides()` — compiled-in defaults + YAML override loader. + +The four-table fragmentation reflected partial consolidations that never finished; once `Capability` was the canonical query-side tag (PR #128) the perf-profile + override-loader story had a natural home next to it (PR #129). + +### 16.2 Refactor 2026-05 (`refactor/controller-layered-cleanup`) — `algebra/` / `planner/` / `config/` retirement + +The pre-refactor layout grew organically as the controller absorbed three legacy code paths (the DC `algebra/` IR, the `planner/` cost models, the `config/` per-deployment emitters). Each had its own naming convention and module structure. This refactor restructures `controller/src/` to mirror design.md §5's target module split without splitting into multiple crates. The retirements: + +| Retired path | Replacement | +|---|---| +| `controller/src/algebra/expr.rs` | `controller/src/intent_algebra/legacy_expr.rs` | +| `controller/src/algebra/lower.rs` | `controller/src/intent_algebra/legacy_lower.rs` | +| `controller/src/algebra/directory.rs` | `controller/src/physical/sketch_catalog.rs` | +| `controller/src/algebra/physical.rs` | `controller/src/physical/planner.rs` | +| `controller/src/algebra/allocator.rs` | `controller/src/physical/allocator.rs` | +| `controller/src/algebra/plan.rs` | `controller/src/physical/plan.rs` | +| `controller/src/algebra/optimizer.rs` | `controller/src/optimizer/engine.rs` | +| `controller/src/planner/cost_model.rs` | `controller/src/optimizer/cost/mod.rs` | +| `controller/src/planner/{delta,online}_cost_model.rs` | `controller/src/optimizer/cost/{delta,online}.rs` | +| `controller/src/planner/{pareto,tco,wire_cost}.rs` | `controller/src/optimizer/cost/{pareto,tco,wire}.rs` | +| `controller/src/planner/rules.rs` | `controller/src/optimizer/rules/mod.rs` | +| `controller/src/planner/baseline_planner.rs` | `controller/src/optimizer/baseline.rs` | +| `controller/src/planner/stage_split.rs` | `controller/src/physical/stage_split.rs` | +| `controller/src/analyzer.rs` | `controller/src/pipeline.rs` | +| `controller/src/stage_split/` | `controller/src/physical/colored_dag/` | +| `controller/src/query_language/` | `controller/src/query_parser/language/` | +| `controller/src/config/workloads.rs` | `controller/src/workload.rs` | +| `controller/src/config/{stage_config*,otap,telegraf,agent,backend,asapquery_backend,precompute}.rs` | `controller/src/emit/{stage_config,otap,telegraf,agent,backend,asapquery_backend,precompute}.rs` | +| (new) | `controller/src/emit/trait_def.rs` — `PlanEmitter` trait placeholder | +| (new) | `controller/src/optimizer/trait_def.rs` — `OptimizerRule` trait placeholder | +| (new) | `controller/src/deployment_model.rs` — `DeploymentModelRegistry` + `DeploymentModelId` placeholder | +| (new) | `controller/src/physical/topology.rs` — `Topology` descriptor re-export | + +`controller/src/lib.rs` keeps thin `pub use` aliases (`pub use pipeline as analyzer;`, `pub use emit as config;`, `pub mod algebra { ... }`, `pub mod planner { ... }`, `pub use physical::colored_dag as stage_split;`) so the historical paths `controller::analyzer::*`, `controller::config::*`, `controller::algebra::*`, `controller::planner::*`, `controller::stage_split::*` keep resolving for external consumers (the `controller` bin's `use controller::algebra;` etc.) without further source churn. Internal source code is migrated to the new paths. + +**TODOs left from the refactor:** + +- `controller/src/emit/stage_config.rs` (3,020 lines, formerly `config/stage_config.rs`) was moved whole rather than split into `emit/opamp.rs` + `emit/streaming_config.rs` + `emit/inference_config.rs` per design.md §5. The monolith mixes OTel-collector YAML emit, ASAPQuery-backend JSON emit, storage-routing JSON emit, and shared internals; a clean split needs ownership reorganisation, not file renames. Tracked for a follow-up. +- `controller/src/intent_algebra/{legacy_expr,legacy_lower}.rs` carry the older `algebra::expr`-flavored `QueryExpr` / `AggIntent` / `WindowSpec` IR alongside the canonical `intent_algebra::{query_expr,agg_intent,lower}` types. Most of the controller still consumes the legacy types (query_parser, language_logical_plan, physical/, optimizer/, emit/); migrating each call site onto the canonical types is a follow-up. +- `controller/src/optimizer/trait_def.rs` (`OptimizerRule`), `controller/src/emit/trait_def.rs` (`PlanEmitter`), `controller/src/deployment_model.rs` (`DeploymentModelRegistry`) ship as placeholders — the existing free-function emitters and concrete rule loops still drive behaviour. Migrating them onto the trait surfaces lands when the per-deployment-model crate split lands. + diff --git a/controller/src/algebra/mod.rs b/controller/src/algebra/mod.rs deleted file mode 100644 index 6cbfc080..00000000 --- a/controller/src/algebra/mod.rs +++ /dev/null @@ -1,44 +0,0 @@ -//! Sketch algebra — the 5-layer query translation pipeline. -//! -//! # Layer architecture -//! -//! | Layer | Module | Role | -//! |-------|--------|------| -//! | **3. Sketch Logical Plan** | [`expr`] | `QueryExpr` + `AggIntent` — implementation-independent algebra | -//! | **3. Sketch Logical Plan** | [`directory`] | Candidate sketch types per `AggIntent`, memory estimation | -//! | **4. Sketch Optimizer** | [`optimizer`] | 12 algebraic rewrite rules (fixed-point iteration) | -//! | **5. Physical Plan** | [`physical`] | `PhysicalAggOp` — resolves `AggIntent` → concrete `SketchType` + `SketchParams` | -//! | **5. Physical Plan** | [`allocator`] | `SketchAllocator` — assigns physical ops to pipeline stages | -//! | **5. Physical Plan** | [`plan`] | `PlanNode` tree with cost estimates and stage annotations | -//! -//! Layers 1–2 (language parsing) live in `query_parser/`. -//! -//! # Typical usage -//! -//! ```rust,ignore -//! use controller::query_parser; -//! use controller::algebra::{expr::QueryExpr, optimizer::QueryOptimizer, physical}; -//! -//! // Layers 1–3: parse query string → sketch logical plan. -//! let query_expr = query_parser::parse_query_expr("quantile_over_time(0.99, latency[5m])")?; -//! -//! // Layer 4: optimise (algebraic rewrite rules). -//! let (opt_expr, _iters) = QueryOptimizer::new(raw_bps).optimize(query_expr); -//! -//! // Layer 5: resolve logical AggIntent → physical SketchType + SketchParams. -//! // (done automatically by stage_split / allocator via physical::resolve) -//! ``` - -pub mod allocator; -pub mod directory; -pub mod expr; -pub mod lower; -pub mod optimizer; -pub mod physical; -pub mod plan; - -// Convenience re-exports. -pub use allocator::SketchAllocator; -pub use expr::{AggFunc, AggIntent, BinaryOpKind, QueryExpr, ScalarExpr, WindowKind, WindowSpec}; -pub use optimizer::QueryOptimizer; -pub use plan::{CostEstimate, ExecutionMode, PipelineStage, PlanNode, PlanSummary}; diff --git a/controller/src/deployment_model.rs b/controller/src/deployment_model.rs new file mode 100644 index 00000000..00b67bbe --- /dev/null +++ b/controller/src/deployment_model.rs @@ -0,0 +1,75 @@ +//! `DeploymentModelRegistry` + `DeploymentModelId` — placeholder +//! scaffolding for the design.md §5 `core::registry` surface. +//! +//! Refactor 2026-05 created this module so the design.md target layout +//! is materialised in code. The single-crate today (no separate +//! `deployment-model-asaplifecycle` / `deployment-model-asapquery` / +//! `deployment-model-asapfusion` crates) means there is exactly one +//! deployment model and the registry is degenerate. The shape is +//! defined here so the future multi-crate migration is purely additive. + +#![allow(dead_code)] + +use std::collections::HashMap; + +/// Stable identifier for a deployment model — `asaplifecycle`, +/// `asapquery`, `asapfusion`, etc. The string is the wire identifier +/// used in `QuerySpec::deployment_model` (see `crate::pipeline::QuerySpec`) +/// and in the per-deployment-model crate selection on a future `bin/` +/// build. +#[derive(Debug, Clone, PartialEq, Eq, Hash)] +pub struct DeploymentModelId(pub String); + +impl DeploymentModelId { + /// Default DC deployment id — the controller's historical name for + /// the lifecycle deployment model (edge / gateway / backend OTel + /// collectors). + pub fn asaplifecycle() -> Self { + Self("asaplifecycle".to_string()) + } + + /// `asapquery` — the ASAPQuery-backend in-process planner for + /// `streaming_config.yaml` + `inference_config.yaml` emission. + pub fn asapquery() -> Self { + Self("asapquery".to_string()) + } + + /// `asapfusion` — the DataFusion `LogicalPlan` rewriter. + pub fn asapfusion() -> Self { + Self("asapfusion".to_string()) + } +} + +/// Registry of available deployment models. +/// +/// Today this is a stub — the single-crate ships only the +/// `asaplifecycle` deployment model and the registry contains one +/// entry. The shape is here so the future multi-crate migration can +/// populate it with concrete `Box` entries without +/// reworking the controller's startup flow. +#[derive(Default)] +pub struct DeploymentModelRegistry { + /// Map of deployment model id → opaque marker. Reserved for the + /// `dyn DeploymentModel` trait object that will land with the + /// multi-crate split. + entries: HashMap, +} + +impl DeploymentModelRegistry { + /// Register a deployment model id. Returns `true` when the id was + /// added; `false` when it was already present (the registry is a + /// set today, not a map). + pub fn register(&mut self, id: DeploymentModelId) -> bool { + self.entries.insert(id, ()).is_none() + } + + /// Whether the registry knows about a deployment model id. + pub fn contains(&self, id: &DeploymentModelId) -> bool { + self.entries.contains_key(id) + } + + /// Iterate over the known deployment model ids. + pub fn ids(&self) -> impl Iterator { + self.entries.keys() + } +} diff --git a/controller/src/config/agent.rs b/controller/src/emit/agent.rs similarity index 99% rename from controller/src/config/agent.rs rename to controller/src/emit/agent.rs index 699112e8..a2cdd988 100644 --- a/controller/src/config/agent.rs +++ b/controller/src/emit/agent.rs @@ -3,7 +3,7 @@ use serde::Serialize; use serde_yaml::{Mapping, Value}; use std::collections::HashMap; -use crate::analyzer::format_duration; +use crate::pipeline::format_duration; use crate::types::*; // ── YAML structural types ───────────────────────────────────────────────────── diff --git a/controller/src/config/asapquery_backend.rs b/controller/src/emit/asapquery_backend.rs similarity index 100% rename from controller/src/config/asapquery_backend.rs rename to controller/src/emit/asapquery_backend.rs diff --git a/controller/src/config/backend.rs b/controller/src/emit/backend.rs similarity index 100% rename from controller/src/config/backend.rs rename to controller/src/emit/backend.rs diff --git a/controller/src/config/mod.rs b/controller/src/emit/mod.rs similarity index 90% rename from controller/src/config/mod.rs rename to controller/src/emit/mod.rs index e6939d52..3edab6fb 100644 --- a/controller/src/config/mod.rs +++ b/controller/src/emit/mod.rs @@ -1,11 +1,28 @@ +//! `emit/` — per-deployment-model plan emitters (L5 output side). +//! +//! Per `controller/docs/design.md` §5 `core::emit`. The 2026-05 +//! layered-cleanup refactor consolidated the former +//! `controller/src/config/` directory here. Mapping: +//! +//! | Old path | New path | +//! |---|---| +//! | `config/agent.rs` | [`agent`] | +//! | `config/backend.rs` | [`backend`] | +//! | `config/asapquery_backend.rs` | [`asapquery_backend`] | +//! | `config/precompute.rs` | [`precompute`] | +//! | `config/stage_config.rs` | [`stage_config`] (TODO: split into `opamp` + `streaming_config` + `inference_config` per design.md §5; deferred from refactor 2026-05 because the 3,020-line monolith mixes OTel-collector YAML emit, ASAPQuery-backend JSON emit, and shared internals — clean split needs ownership reorganisation, not file renames) | +//! | `config/stage_config_otap.rs` | [`otap`] | +//! | `config/stage_config_telegraf.rs` | [`telegraf`] | +//! | `config/workloads.rs` | [`crate::workload`] (top-level — design.md §5 puts `workload` next to `emit`, not inside it) | + pub mod agent; pub mod asapquery_backend; pub mod backend; pub mod precompute; pub mod stage_config; -pub mod stage_config_otap; -pub mod stage_config_telegraf; -pub mod workloads; +pub mod otap; +pub mod telegraf; +pub mod trait_def; pub use agent::generate_agent_config; pub use asapquery_backend::generate_streaming_config_yaml; @@ -17,11 +34,18 @@ pub use stage_config::{ emit_backend_storage_routing_with_prometheus_for_tenant, emit_edge_yaml, emit_gateway_yaml, DEFAULT_TENANT, }; -pub use stage_config_otap::emit_otap_dag_yaml; -pub use stage_config_telegraf::emit_telegraf_toml; -pub use workloads::WorkloadRegistry; - -use crate::stage_split::emitter::EdgeStageConfig; +pub use otap::emit_otap_dag_yaml; +pub use telegraf::emit_telegraf_toml; +pub use trait_def::PlanEmitter; + +// Refactor 2026-05: design.md §5 puts `WorkloadRegistry` next to +// `emit`, not inside it. The new home is `crate::workload`; we re-export +// here so historical `crate::config::WorkloadRegistry` and +// `controller::config::WorkloadRegistry` references via the `config` +// back-compat alias keep working without churn. +pub use crate::workload::WorkloadRegistry; + +use crate::physical::colored_dag::emitter::EdgeStageConfig; use crate::sketch_algebra::SketchExpr; use crate::sketch_algebra::params::SketchKind; use crate::store::WorkloadStore; @@ -148,7 +172,7 @@ pub fn extend_edge_with_demo_plumbing( edge_cfg: &mut EdgeStageConfig, workload_registry_metrics: impl IntoIterator, ) { - use crate::stage_split::emitter::ArchiveTierMetric; + use crate::physical::colored_dag::emitter::ArchiveTierMetric; // 1. Freshness probes → archive tier with the tight 10s window. for m in FRESHNESS_PROBE_METRICS.iter() { @@ -300,8 +324,8 @@ mod runtime_tests { #[test] fn emit_for_runtime_default_matches_emit_edge_yaml() { use crate::sketch_algebra::params::{DDSketchParams, SketchParams}; - use crate::stage_split::emitter::{EdgeSketchProcessor, ExportTarget}; - use crate::stage_split::stage_id::StageId; + use crate::physical::colored_dag::emitter::{EdgeSketchProcessor, ExportTarget}; + use crate::physical::colored_dag::stage_id::StageId; use crate::sketch_algebra::params::SketchKind; let cfg = EdgeStageConfig { @@ -330,8 +354,8 @@ mod runtime_tests { #[test] fn emit_for_runtime_otap_yields_dag_yaml() { - use crate::stage_split::emitter::ExportTarget; - use crate::stage_split::stage_id::StageId; + use crate::physical::colored_dag::emitter::ExportTarget; + use crate::physical::colored_dag::stage_id::StageId; let cfg = EdgeStageConfig { source_metric: Some("m".to_string()), @@ -353,8 +377,8 @@ mod runtime_tests { #[test] fn emit_for_runtime_telegraf_yields_toml() { - use crate::stage_split::emitter::ExportTarget; - use crate::stage_split::stage_id::StageId; + use crate::physical::colored_dag::emitter::ExportTarget; + use crate::physical::colored_dag::stage_id::StageId; let cfg = EdgeStageConfig { source_metric: Some("m".to_string()), @@ -392,7 +416,7 @@ mod runtime_tests { registry: &WorkloadRegistry, store: &WorkloadStore, ) { - use crate::analyzer::{Analyzer, QuerySpec}; + use crate::pipeline::{Analyzer, QuerySpec}; use crate::types; use crate::types_v2; let analyzer = Analyzer::new(); @@ -464,11 +488,11 @@ mod runtime_tests { assign_to_role: agent sketch_family_override: CountMinSketch "#; - let entries: Vec = + let entries: Vec = serde_yaml::from_str(yaml).expect("parse workload yaml"); assert_eq!(entries.len(), 6, "all 6 contract metrics must deserialize"); - let registry = crate::config::workloads::WorkloadRegistry::from_entries(entries); + let registry = crate::workload::WorkloadRegistry::from_entries(entries); let store = WorkloadStore::new(); populate_store_from_registry(®istry, &store); diff --git a/controller/src/config/stage_config_otap.rs b/controller/src/emit/otap.rs similarity index 98% rename from controller/src/config/stage_config_otap.rs rename to controller/src/emit/otap.rs index 85fa47c5..f3e9e24e 100644 --- a/controller/src/config/stage_config_otap.rs +++ b/controller/src/emit/otap.rs @@ -39,8 +39,8 @@ use serde_yaml::{Mapping, Value}; use std::collections::BTreeMap; use crate::sketch_algebra::params::{SketchKind, SketchParams}; -use crate::stage_split::emitter::{EdgeStageConfig, EdgeSketchProcessor, ExportTarget}; -use crate::stage_split::stage_id::StageId; +use crate::physical::colored_dag::emitter::{EdgeStageConfig, EdgeSketchProcessor, ExportTarget}; +use crate::physical::colored_dag::stage_id::StageId; /// Default URL for Prometheus's native OTLP HTTP receiver. /// Matches `super::stage_config::emit_edge_yaml`'s placeholder so the @@ -337,7 +337,7 @@ fn sketch_kind_tag(kind: &SketchKind) -> &'static str { mod tests { use super::*; use crate::sketch_algebra::params::DDSketchParams; - use crate::stage_split::emitter::{EdgeSketchProcessor, PrometheusArchiveMetric}; + use crate::physical::colored_dag::emitter::{EdgeSketchProcessor, PrometheusArchiveMetric}; /// Minimal struct-stub used to validate the emitted DAG parses as the /// otap-dataflow schema. We don't pull in the otap-df-config crate diff --git a/controller/src/config/precompute.rs b/controller/src/emit/precompute.rs similarity index 99% rename from controller/src/config/precompute.rs rename to controller/src/emit/precompute.rs index bd074e60..2462daff 100644 --- a/controller/src/config/precompute.rs +++ b/controller/src/emit/precompute.rs @@ -3,7 +3,7 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use std::time::Duration; -use crate::analyzer::format_duration; +use crate::pipeline::format_duration; use crate::types::*; // ── Scheduling rule ─────────────────────────────────────────────────────────── diff --git a/controller/src/config/stage_config.rs b/controller/src/emit/stage_config.rs similarity index 99% rename from controller/src/config/stage_config.rs rename to controller/src/emit/stage_config.rs index 5fde9e35..91fdba70 100644 --- a/controller/src/config/stage_config.rs +++ b/controller/src/emit/stage_config.rs @@ -1,5 +1,5 @@ //! Phase B (MVP v6) — turn a typed L5 [`StageConfig`] map (produced by -//! [`crate::stage_split::ThreeStageEmitter`]) into the **wire bytes** the +//! [`crate::physical::colored_dag::ThreeStageEmitter`]) into the **wire bytes** the //! three executors actually consume: //! //! - [`emit_edge_yaml`] → OTel-collector YAML for the edge agent (OTLP @@ -40,12 +40,12 @@ use std::collections::HashMap; use crate::sketch_algebra::params::{SketchKind, SketchParams}; use crate::sketch_algebra::sketch_expr::EstimateOp; -use crate::stage_split::emitter::{ +use crate::physical::colored_dag::emitter::{ AggregationInput, ArchiveTierMetric, BackendAggregation, BackendReadout, BackendStageConfig, EdgeSketchProcessor, EdgeStageConfig, ExportTarget, GatewayMergeProcessor, GatewayStageConfig, PrometheusArchiveMetric, }; -use crate::stage_split::stage_id::StageId; +use crate::physical::colored_dag::stage_id::StageId; // ── YAML structural types ───────────────────────────────────────────────────── // @@ -1134,7 +1134,7 @@ tsdb_block_duration: {window_secs}s\n", /// Map a `SketchKind` to the OTel processor name registered by the /// patched contrib build's factory. Keep in sync with -/// `crate::stage_split::emitter::edge_processor_name`. +/// `crate::physical::colored_dag::emitter::edge_processor_name`. fn sketch_kind_to_processor_name(kind: &SketchKind) -> &'static str { match kind { SketchKind::DDSketch => "ddsketch", diff --git a/controller/src/config/stage_config_telegraf.rs b/controller/src/emit/telegraf.rs similarity index 98% rename from controller/src/config/stage_config_telegraf.rs rename to controller/src/emit/telegraf.rs index 5542525b..5d346646 100644 --- a/controller/src/config/stage_config_telegraf.rs +++ b/controller/src/emit/telegraf.rs @@ -38,8 +38,8 @@ use anyhow::{Context, Result}; use crate::sketch_algebra::params::{SketchKind, SketchParams}; -use crate::stage_split::emitter::{EdgeStageConfig, EdgeSketchProcessor, ExportTarget}; -use crate::stage_split::stage_id::StageId; +use crate::physical::colored_dag::emitter::{EdgeStageConfig, EdgeSketchProcessor, ExportTarget}; +use crate::physical::colored_dag::stage_id::StageId; /// Default Prometheus remote-write URL for Mode 3 — Telegraf doesn't /// support OTLP-HTTP egress, so we land in the same Prometheus archive @@ -296,7 +296,7 @@ mod toml_minimal { mod tests { use super::*; use crate::sketch_algebra::params::{DDSketchParams, KllParams}; - use crate::stage_split::emitter::{EdgeSketchProcessor, PrometheusArchiveMetric}; + use crate::physical::colored_dag::emitter::{EdgeSketchProcessor, PrometheusArchiveMetric}; fn ddsketch_edge_cfg_mode1() -> EdgeStageConfig { EdgeStageConfig { diff --git a/controller/src/emit/trait_def.rs b/controller/src/emit/trait_def.rs new file mode 100644 index 00000000..bebda867 --- /dev/null +++ b/controller/src/emit/trait_def.rs @@ -0,0 +1,37 @@ +//! `PlanEmitter` trait — placeholder scaffolding for the L5 output +//! surface declared in `controller/docs/design.md` §5 / §6 +//! `core::emit::PlanEmitter`. +//! +//! Refactor 2026-05 created this module so the design.md target layout +//! is materialised in code. The existing per-runtime emitters +//! ([`super::stage_config::emit_edge_yaml`], [`super::otap::emit_otap_dag_yaml`], +//! [`super::telegraf::emit_telegraf_toml`], [`super::asapquery_backend::generate_streaming_config_yaml`], +//! [`super::agent::generate_agent_config`], [`super::backend::generate_backend_config`]) +//! still operate as free functions with deployment-specific signatures. +//! Migrating each onto this trait is a follow-up — until then this +//! module is intentionally minimal so it has zero behavior impact. + +#![allow(dead_code)] + +use anyhow::Result; + +/// `PlanEmitter` — every per-deployment-model plan emitter implements +/// this so a future controller pipeline can call them polymorphically. +/// +/// The `Input` associated type is the typed L5 stage config the emitter +/// consumes (e.g. `EdgeStageConfig` for OTel YAML emitters, +/// `BackendStageConfig` for the ASAPQuery-backend `StreamingConfig` +/// emitter). `Output` is the wire-format string / JSON the deployment +/// model's transport expects. +pub trait PlanEmitter: Send + Sync { + /// Typed L5 stage config this emitter consumes. + type Input; + /// Wire-format output produced for the deployment model's transport. + type Output; + + /// Stable name for diagnostics + the emitter registry. + fn name(&self) -> &'static str; + + /// Emit the wire-format payload for the given stage config. + fn emit(&self, input: &Self::Input) -> Result; +} diff --git a/controller/src/algebra/expr.rs b/controller/src/intent_algebra/legacy_expr.rs similarity index 98% rename from controller/src/algebra/expr.rs rename to controller/src/intent_algebra/legacy_expr.rs index 26b2144a..35eb433a 100644 --- a/controller/src/algebra/expr.rs +++ b/controller/src/intent_algebra/legacy_expr.rs @@ -12,8 +12,8 @@ //! //! # Stage vocabulary //! -//! Once the [`crate::algebra::allocator::SketchAllocator`] annotates the tree, -//! every node carries a [`PipelineStage`](crate::algebra::plan::PipelineStage) +//! Once the [`crate::physical::allocator::SketchAllocator`] annotates the tree, +//! every node carries a [`PipelineStage`](crate::physical::plan::PipelineStage) //! tag that says where the work executes: //! //! | Stage | Component | @@ -175,16 +175,16 @@ impl AggIntent { } // ── Accuracy helpers ───────────────────────────────────────────────────────── - -/// HLL accuracy from register count: `1.04 / sqrt(2^registers)`. -pub fn hll_accuracy(registers: u8) -> f64 { - 1.04 / (2.0f64.powi(registers as i32)).sqrt() -} - -/// CountMin accuracy from width: `e / width`. -pub fn countmin_accuracy(width: u32) -> f64 { - std::f64::consts::E / width as f64 -} +// +// The 2026-05 layered-cleanup refactor moved these helpers to +// `sketch_algebra::capability` (their structural home — sketch-family +// error bounds). The thin re-exports below keep `legacy_expr::hll_accuracy` / +// `legacy_expr::countmin_accuracy` available so the in-file +// `default_cardinality` call site (and any external `algebra::expr::hll_accuracy` +// reference resolved via the back-compat `algebra` alias in `lib.rs`) keep +// compiling. + +pub use crate::sketch_algebra::capability::{countmin_accuracy, hll_accuracy}; /// Unified window specification — captures all language-level window semantics. #[derive(Debug, Clone, PartialEq)] diff --git a/controller/src/algebra/lower.rs b/controller/src/intent_algebra/legacy_lower.rs similarity index 99% rename from controller/src/algebra/lower.rs rename to controller/src/intent_algebra/legacy_lower.rs index 760c220d..4c37e4a0 100644 --- a/controller/src/algebra/lower.rs +++ b/controller/src/intent_algebra/legacy_lower.rs @@ -13,7 +13,7 @@ //! Multi-agg `Aggregate` nodes or those with HAVING clauses pass through //! unchanged — the physical planner handles them. -use crate::algebra::expr::*; +use crate::intent_algebra::legacy_expr::*; /// Lower a Layer 2 `QueryExpr` (relational operators only) to Layer 3 /// (sketch algebra with `AggIntent`). diff --git a/controller/src/intent_algebra/mod.rs b/controller/src/intent_algebra/mod.rs index 0927ef77..c88d94a2 100644 --- a/controller/src/intent_algebra/mod.rs +++ b/controller/src/intent_algebra/mod.rs @@ -79,6 +79,17 @@ pub mod lower; pub mod query_expr; pub mod schema; +// Refactor 2026-05 (`refactor/controller-layered-cleanup`): the +// pre-existing legacy L3+ IR formerly at `controller/src/algebra/expr.rs` +// + `controller/src/algebra/lower.rs` lives here while a separate +// follow-up unifies it with the canonical `query_expr` / `lower` +// modules above. These two `legacy_*` modules carry the heavy +// `QueryExpr` / `AggIntent` types used by the planner, allocator, +// physical planner, query_parser, and language_logical_plan modules +// today. +pub mod legacy_expr; +pub mod legacy_lower; + // Re-exports for the canonical surface — `crate::intent_algebra::*` for // downstream callers that don't want to chase sub-module paths. pub use agg_intent::AggIntent; diff --git a/controller/src/language_logical_plan/lower.rs b/controller/src/language_logical_plan/lower.rs index d7316126..7b10e5ab 100644 --- a/controller/src/language_logical_plan/lower.rs +++ b/controller/src/language_logical_plan/lower.rs @@ -1,12 +1,12 @@ //! L1 → L2 lowering — produces a [`LanguageLogicalPlan`] from a -//! [`crate::query_language::LanguageAst`]. +//! [`crate::query_parser::language::LanguageAst`]. //! //! Today only the PromQL backend has a working L1, so only the PromQL //! variant of `LanguageAst` lowers to a real L2 tree. Other variants //! return [`LoweringError::UnsupportedLanguage`] cleanly so callers //! can surface a uniform error. -use crate::query_language::{language_ast::LanguageAst, promql::PromQLAst}; +use crate::query_parser::language::{language_ast::LanguageAst, promql::PromQLAst}; use crate::types_v2::QueryLanguage; use super::plan::{LanguageLogicalPlan, LanguageLogicalPlanSummary}; diff --git a/controller/src/language_logical_plan/mod.rs b/controller/src/language_logical_plan/mod.rs index c7f502f0..982f9469 100644 --- a/controller/src/language_logical_plan/mod.rs +++ b/controller/src/language_logical_plan/mod.rs @@ -10,7 +10,7 @@ //! per-language algebra tree that preserves language-specific //! semantics (PromQL instant vs range vector, SQL window frames, //! Elastic buckets) before the L3 normalisation pass collapses -//! everything into the language-orthogonal [`crate::algebra::expr::QueryExpr`]. +//! everything into the language-orthogonal [`crate::intent_algebra::legacy_expr::QueryExpr`]. //! //! # DC deployment scope //! @@ -20,11 +20,11 @@ //! //! # Layer split //! -//! - L1 ([`crate::query_language`]) — `&str` → [`crate::query_language::LanguageAst`]. -//! - L2 ([`Self`]) — [`crate::query_language::LanguageAst`] → +//! - L1 ([`crate::query_parser::language`]) — `&str` → [`crate::query_parser::language::LanguageAst`]. +//! - L2 ([`Self`]) — [`crate::query_parser::language::LanguageAst`] → //! [`LanguageLogicalPlan`]. //! - L3 (`crate::intent_algebra`, Phase B) — [`LanguageLogicalPlan`] -//! → language-orthogonal [`crate::algebra::expr::QueryExpr`]. +//! → language-orthogonal [`crate::intent_algebra::legacy_expr::QueryExpr`]. pub mod lower; pub mod plan; diff --git a/controller/src/language_logical_plan/plan.rs b/controller/src/language_logical_plan/plan.rs index 97d9c86b..3063ff0b 100644 --- a/controller/src/language_logical_plan/plan.rs +++ b/controller/src/language_logical_plan/plan.rs @@ -7,7 +7,7 @@ //! `Sort`, `Limit`. **No sketch names yet**. //! //! In DC's existing tree the equivalent representation is the -//! [`crate::algebra::expr::QueryExpr`] tree produced by +//! [`crate::intent_algebra::legacy_expr::QueryExpr`] tree produced by //! `query_parser::parse_query_expr`. To stay consistent with the //! design.md L2 contract while not duplicating the algebra: //! @@ -25,7 +25,7 @@ use std::collections::HashMap; use std::time::Duration; -use crate::algebra::expr::QueryExpr; +use crate::intent_algebra::legacy_expr::QueryExpr; use crate::query_parser::{ParsedQuery, QueryHint}; use crate::types::AggType; use crate::types_v2::QueryLanguage; diff --git a/controller/src/language_logical_plan/tests.rs b/controller/src/language_logical_plan/tests.rs index d3e30311..4e0f9644 100644 --- a/controller/src/language_logical_plan/tests.rs +++ b/controller/src/language_logical_plan/tests.rs @@ -1,8 +1,8 @@ //! Tests for L2 lowering (`language_logical_plan`). use super::*; -use crate::algebra::expr::{AggIntent, QueryExpr}; -use crate::query_language::{Language, LanguageAst, PromQLLanguage}; +use crate::intent_algebra::legacy_expr::{AggIntent, QueryExpr}; +use crate::query_parser::language::{Language, LanguageAst, PromQLLanguage}; use crate::types::AggType; use crate::types_v2::QueryLanguage; @@ -68,10 +68,10 @@ fn lower_unsupported_language_errors_cleanly() { // smoke test is: stub backends fail at L1 with `Unimplemented`, and // the type system rules out passing them to L2 lowering. We assert // that contract here. - let err = crate::query_language::SqlLanguage.parse("SELECT 1").unwrap_err(); + let err = crate::query_parser::language::SqlLanguage.parse("SELECT 1").unwrap_err(); assert!(matches!( err, - crate::query_language::ParseError::Unimplemented(_) + crate::query_parser::language::ParseError::Unimplemented(_) )); } diff --git a/controller/src/lib.rs b/controller/src/lib.rs index 130a5280..5812cf3d 100644 --- a/controller/src/lib.rs +++ b/controller/src/lib.rs @@ -1,23 +1,56 @@ //! Controller crate — library surface for the ASAPQuery-backend host. //! -//! Refactor-2026-05 (Phase 9): the controller previously ran as a -//! standalone binary with its own OpAMP server and HTTP API. After the -//! controller crate moved into ASAPQuery-backend, the same modules are -//! exposed as a Rust library so `asap-query-engine` can call them -//! in-process — capability mapping, plan emission, OpAMP push from the -//! backend host. This `lib.rs` declares the public module surface; the -//! existing `main.rs` continues to provide the standalone binary -//! entrypoint for any deployments that still want to run controller -//! out-of-process. +//! Refactor-2026-05 (Phase 9 / `refactor/controller-layered-cleanup`): +//! the controller previously ran as a standalone binary with its own +//! OpAMP server and HTTP API. After the controller crate moved into +//! ASAPQuery-backend, the same modules are exposed as a Rust library so +//! `asap-query-engine` can call them in-process — capability mapping, +//! plan emission, OpAMP push from the backend host. This `lib.rs` +//! declares the public module surface; the existing `main.rs` continues +//! to provide the standalone binary entrypoint for any deployments that +//! still want to run controller out-of-process. +//! +//! ## 2026-05 layered-cleanup refactor — old → new module mapping +//! +//! The internal module layout was restructured to mirror +//! `controller/docs/design.md` §5 target layout without splitting into +//! multiple crates: +//! +//! | Old path | New path | +//! |---|---| +//! | `controller/src/algebra/expr.rs` | `controller/src/intent_algebra/legacy_expr.rs` | +//! | `controller/src/algebra/lower.rs` | `controller/src/intent_algebra/legacy_lower.rs` | +//! | `controller/src/algebra/directory.rs` | `controller/src/physical/sketch_catalog.rs` | +//! | `controller/src/algebra/physical.rs` | `controller/src/physical/planner.rs` | +//! | `controller/src/algebra/allocator.rs` | `controller/src/physical/allocator.rs` | +//! | `controller/src/algebra/plan.rs` | `controller/src/physical/plan.rs` | +//! | `controller/src/algebra/optimizer.rs` | `controller/src/optimizer/engine.rs` | +//! | `controller/src/planner/cost_model.rs` | `controller/src/optimizer/cost/mod.rs` | +//! | `controller/src/planner/{delta,online}_cost_model.rs` | `controller/src/optimizer/cost/{delta,online}.rs` | +//! | `controller/src/planner/{pareto,tco,wire_cost}.rs` | `controller/src/optimizer/cost/{pareto,tco,wire}.rs` | +//! | `controller/src/planner/rules.rs` | `controller/src/optimizer/rules/mod.rs` | +//! | `controller/src/planner/baseline_planner.rs` | `controller/src/optimizer/baseline.rs` | +//! | `controller/src/planner/stage_split.rs` | `controller/src/physical/stage_split.rs` | +//! | `controller/src/analyzer.rs` | `controller/src/pipeline.rs` | +//! | `controller/src/stage_split/` | `controller/src/physical/colored_dag/` | +//! | `controller/src/query_language/` | `controller/src/query_parser/language/` | +//! | `controller/src/config/workloads.rs` | `controller/src/workload.rs` | +//! | `controller/src/config/{stage_config*,agent,backend,asapquery_backend,precompute}.rs` | `controller/src/emit/{...}.rs` | //! //! Public modules to consume from `asap-query-engine`: -//! - `sketch_algebra` — Capability enum + `capability_for(query_func)` +//! - `sketch_algebra` — `Capability` enum + `capability_for(agg_intent: &AggIntent)` //! lookup table (Phase 4 / 5 use this to route raw-name PromQL). -//! - `intent_algebra` — `AggIntent` + `QueryExpr` DAG. +//! - `intent_algebra` — `AggIntent` + `QueryExpr` DAG (canonical L3 IR; +//! `legacy_expr` / `legacy_lower` carry the older `algebra::expr`-flavored +//! IR pending full migration). //! - `language_logical_plan` — PromQL → AST → logical plan. -//! - `query_parser` / `query_language` — front-end parsers. -//! - `planner` — full L1→L5 pipeline runner. -//! - `stage_split` — per-stage YAML emission. +//! - `query_parser` — front-end parsers (Layer 1). +//! The former top-level `query_language/` module was folded into +//! `query_parser::language` by this refactor. +//! - `physical` — L5 framework (allocator, planner, plan, sketch_catalog, +//! colored_dag, stage_split, topology). +//! - `optimizer` — L4 rule engine + cost model traits/impls + baseline +//! planner. //! - `opamp` — OpAMP server (will be invoked from the backend's //! service startup once Phase 4 wires the in-process integration). //! - `types`, `types_v2` — controller-internal data model. @@ -27,27 +60,124 @@ //! part of any wire/protocol contract. pub mod accuracy; -pub mod algebra; -pub mod analyzer; pub mod backend_client; -pub mod config; +pub mod deployment_model; +pub mod emit; pub mod intent_algebra; pub mod language_logical_plan; pub mod metrics_exposer; pub mod monitor; pub mod opamp; -pub mod planner; -pub mod query_language; +pub mod optimizer; +pub mod physical; +pub mod pipeline; pub mod query_parser; pub mod replan; pub mod runtime_samples; pub mod sketch_algebra; -pub mod stage_split; pub mod store; pub mod types; pub mod types_v2; +pub mod workload; + +/// Back-compat alias — the legacy `crate::config` module surface, +/// re-exported from its new homes ([`emit`] for the per-deployment-model +/// emitters and [`workload`] for `WorkloadRegistry`). Refactor 2026-05 +/// introduced this so `main.rs` keeps using `controller::config::*` +/// without source churn. +pub use emit as config; + +// Refactor 2026-05: the legacy `analyzer` and `planner` module paths +// resolve into `pipeline` and the new `optimizer` + `physical` split +// respectively. Keeping `analyzer` as a module alias preserves the +// historical name on the `crate::analyzer::*` path for downstream +// callers (`controller::analyzer::QuerySpec` is the JSON-facing type +// in `main.rs` and the HTTP route handlers). +pub use pipeline as analyzer; + +/// Back-compat shim — the legacy `crate::algebra` module surface, +/// re-exported from its new homes (`intent_algebra::legacy_expr`, +/// `intent_algebra::legacy_lower`, `optimizer::engine`, +/// `physical::sketch_catalog`, `physical::planner`, +/// `physical::allocator`, `physical::plan`). +/// +/// Refactor 2026-05 introduced this shim so `main.rs` and other +/// consumers can keep using `algebra::QueryOptimizer`, +/// `algebra::SketchAllocator`, `algebra::physical::*`, +/// `algebra::optimizer::DeploymentConstraints`, etc. without source +/// churn. Future cleanup should migrate call sites to the new paths +/// and delete this shim. +pub mod algebra { + pub use crate::intent_algebra::legacy_expr as expr; + pub use crate::intent_algebra::legacy_lower as lower; + pub use crate::physical::sketch_catalog as directory; + pub use crate::physical::planner as physical; + pub use crate::physical::allocator; + pub use crate::physical::plan; + pub use crate::optimizer::engine as optimizer; + + pub use crate::physical::allocator::SketchAllocator; + pub use crate::physical::plan::{ + CostEstimate, ExecutionMode, PipelineStage, PlanNode, PlanSummary, + }; + pub use crate::optimizer::engine::QueryOptimizer; + pub use crate::intent_algebra::legacy_expr::{ + AggFunc, AggIntent, BinaryOpKind, QueryExpr, ScalarExpr, WindowKind, WindowSpec, + }; +} + +/// Back-compat shim — the legacy `crate::stage_split` module surface, +/// re-exported from its new home [`physical::colored_dag`]. +/// +/// Refactor 2026-05 moved `controller/src/stage_split/` into +/// `controller/src/physical/colored_dag/` so the L5 typed colouring +/// framework sits beneath the `physical` umbrella. The alias preserves +/// `controller::stage_split::*` and `crate::stage_split::*` paths. +pub use physical::colored_dag as stage_split; + +/// Back-compat shim — the legacy `crate::planner` module surface, +/// re-exported from its new homes: +/// +/// - `planner::cost_model`, `planner::delta_cost_model`, +/// `planner::online_cost_model`, `planner::pareto`, `planner::tco`, +/// `planner::wire_cost` → [`optimizer::cost`] (+ submodules). +/// - `planner::rules` → [`optimizer::rules`]. +/// - `planner::baseline_planner` → [`optimizer::baseline`]. +/// - `planner::stage_split` → [`physical::stage_split`]. +/// +/// Refactor 2026-05 introduced this shim to avoid touching ~40 +/// `crate::planner::*` sites in `main.rs` / `replan.rs` / tests. +/// Future cleanup should migrate call sites to the new paths and +/// delete this shim. +pub mod planner { + pub use crate::optimizer::baseline as baseline_planner; + pub use crate::optimizer::cost as cost_model; + pub use crate::optimizer::cost::delta as delta_cost_model; + pub use crate::optimizer::cost::online as online_cost_model; + pub use crate::optimizer::cost::pareto; + pub use crate::optimizer::cost::tco; + pub use crate::optimizer::cost::wire as wire_cost; + pub use crate::optimizer::rules; + pub use crate::physical::stage_split; + + // Top-level convenience re-exports that historically lived at + // `crate::planner::*`. The legacy `planner/mod.rs` was 19 lines — + // these are the symbols it surfaced. + pub use crate::optimizer::baseline::BaselinePlanner; + pub use crate::optimizer::cost::CostModelPlanner; + pub use crate::optimizer::cost::online::{init_store as init_online_store, OnlineMetricsStore}; + pub use crate::optimizer::cost::pareto::{ + pareto_frontier, select_best, ObjectiveWeights, ParetoPoint, + }; + pub use crate::optimizer::cost::wire::{ + break_even_samples, est_wire_bytes_per_window_per_series, select_bind_mode, BindMode, + SketchWireCost, WireCostTable, WireWorkload, + }; + pub use crate::optimizer::rules::RulesPlanner; +} /// PromQL → warm-tier candidate analyzer. Phase-9 unification of the /// per-`Capability` dispatch knowledge that previously lived in /// `asap-query-engine/src/engines/warm_tier/promql_extract.rs`. See /// the module docs for the full PromQL shape coverage matrix. pub mod warm_tier_analysis; + diff --git a/controller/src/main.rs b/controller/src/main.rs index 8735355d..265493d1 100644 --- a/controller/src/main.rs +++ b/controller/src/main.rs @@ -9,7 +9,6 @@ use controller::metrics_exposer; use controller::monitor; use controller::opamp; use controller::planner; -use controller::query_language; use controller::query_parser; use controller::replan; use controller::runtime_samples; @@ -1392,7 +1391,7 @@ mod api_tests { async fn rollback_no_previous_returns_400() { let (st, app) = test_app(); // Seed one plan directly. - use crate::planner::rules::RulesPlanner; + use crate::optimizer::rules::RulesPlanner; let wl = crate::types::QueryWorkload { metric_name: "m".into(), label_filters: std::collections::HashMap::new(), diff --git a/controller/src/planner/baseline_planner.rs b/controller/src/optimizer/baseline.rs similarity index 99% rename from controller/src/planner/baseline_planner.rs rename to controller/src/optimizer/baseline.rs index 2b56bb4c..50ff4648 100644 --- a/controller/src/planner/baseline_planner.rs +++ b/controller/src/optimizer/baseline.rs @@ -17,7 +17,7 @@ use std::collections::HashMap; use std::sync::{Arc, RwLock}; use crate::types::{CollectionPlan, QueryWorkload, WorkloadCharacteristics}; -use super::cost_model::CostModelPlanner; +use crate::optimizer::cost::CostModelPlanner; pub struct BaselinePlanner { inner: CostModelPlanner, diff --git a/controller/src/planner/delta_cost_model.rs b/controller/src/optimizer/cost/delta.rs similarity index 99% rename from controller/src/planner/delta_cost_model.rs rename to controller/src/optimizer/cost/delta.rs index f08c3a1b..0a31f411 100644 --- a/controller/src/planner/delta_cost_model.rs +++ b/controller/src/optimizer/cost/delta.rs @@ -453,7 +453,7 @@ pub fn decide_delta( #[cfg(test)] mod tests { use super::*; - use crate::planner::rules::default_sketch_params; + use crate::optimizer::rules::default_sketch_params; use chrono::Utc; use std::collections::HashMap; diff --git a/controller/src/planner/cost_model.rs b/controller/src/optimizer/cost/mod.rs similarity index 98% rename from controller/src/planner/cost_model.rs rename to controller/src/optimizer/cost/mod.rs index 980accf2..4ee7c550 100644 --- a/controller/src/planner/cost_model.rs +++ b/controller/src/optimizer/cost/mod.rs @@ -1,9 +1,17 @@ use std::collections::HashMap; use std::time::Duration; -use super::delta_cost_model::decide_delta; -use super::online_cost_model; -use super::rules::{default_sketch_params, select_window_strategy, RulesPlanner}; +// Sub-modules — formerly siblings of `planner/cost_model.rs` (now +// `optimizer/cost/mod.rs`); the 2026-05 refactor pulled each into a +// dedicated `optimizer/cost/.rs`. +pub mod delta; +pub mod online; +pub mod pareto; +pub mod tco; +pub mod wire; + +use self::delta::decide_delta; +use crate::optimizer::rules::{default_sketch_params, select_window_strategy, RulesPlanner}; use crate::types::*; // ── Benchmark-derived cost table ────────────────────────────────────────────── @@ -170,7 +178,7 @@ fn estimate_error(_st: &SketchType, p: &SketchParams, costs: SketchCosts) -> f64 /// so that real-world behaviour gradually supersedes the static benchmark defaults. pub struct CostModelPlanner { inner: RulesPlanner, - online_store: Option, + online_store: Option, } impl CostModelPlanner { @@ -184,7 +192,7 @@ impl CostModelPlanner { } /// Attach a live EMA store so scoring uses blended benchmark + observed costs. - pub fn with_online_store(mut self, store: online_cost_model::OnlineMetricsStore) -> Self { + pub fn with_online_store(mut self, store: online::OnlineMetricsStore) -> Self { self.online_store = Some(store); self } @@ -192,7 +200,7 @@ impl CostModelPlanner { /// Returns the effective cost table: online-blended when available, benchmark otherwise. fn cost_table(&self) -> HashMap { match &self.online_store { - Some(s) => online_cost_model::effective_table(s), + Some(s) => online::effective_table(s), None => benchmark_table_pub(), } } @@ -234,7 +242,7 @@ impl CostModelPlanner { return plan; } - let candidates = crate::algebra::directory::candidates_for_workload(&w.aggregations); + let candidates = crate::physical::sketch_catalog::candidates_for_workload(&w.aggregations); // Start with the rule-based plan as the baseline. let baseline = self.inner.plan(w); diff --git a/controller/src/planner/online_cost_model.rs b/controller/src/optimizer/cost/online.rs similarity index 99% rename from controller/src/planner/online_cost_model.rs rename to controller/src/optimizer/cost/online.rs index 63c4cf32..ffcfd9e6 100644 --- a/controller/src/planner/online_cost_model.rs +++ b/controller/src/optimizer/cost/online.rs @@ -21,7 +21,7 @@ use std::sync::Arc; use tokio::sync::RwLock; -use super::cost_model::{benchmark_table_pub, SketchCosts}; +use crate::optimizer::cost::{benchmark_table_pub, SketchCosts}; use crate::types::SketchType; // ── Tuning constants ────────────────────────────────────────────────────────── diff --git a/controller/src/planner/pareto.rs b/controller/src/optimizer/cost/pareto.rs similarity index 97% rename from controller/src/planner/pareto.rs rename to controller/src/optimizer/cost/pareto.rs index 7dcfd6ea..20501faf 100644 --- a/controller/src/planner/pareto.rs +++ b/controller/src/optimizer/cost/pareto.rs @@ -23,10 +23,10 @@ use std::collections::HashMap; -use super::cost_model::{benchmark_table_pub, score_with, SketchCosts}; -use super::delta_cost_model::decide_delta; -use super::online_cost_model; -use super::rules::{default_sketch_params, select_window_strategy, RulesPlanner}; +use crate::optimizer::cost::{benchmark_table_pub, score_with, SketchCosts}; +use crate::optimizer::cost::delta::decide_delta; +use crate::optimizer::cost::online; +use crate::optimizer::rules::{default_sketch_params, select_window_strategy, RulesPlanner}; use crate::types::*; // ── Types ───────────────────────────────────────────────────────────────────── @@ -88,10 +88,10 @@ pub fn pareto_frontier( workload: &QueryWorkload, wc: &WorkloadCharacteristics, weights: ObjectiveWeights, - online_store: Option<&online_cost_model::OnlineMetricsStore>, + online_store: Option<&online::OnlineMetricsStore>, ) -> Vec { let table: HashMap = match online_store { - Some(s) => online_cost_model::effective_table(s), + Some(s) => online::effective_table(s), None => benchmark_table_pub(), }; diff --git a/controller/src/planner/tco.rs b/controller/src/optimizer/cost/tco.rs similarity index 100% rename from controller/src/planner/tco.rs rename to controller/src/optimizer/cost/tco.rs diff --git a/controller/src/planner/wire_cost.rs b/controller/src/optimizer/cost/wire.rs similarity index 100% rename from controller/src/planner/wire_cost.rs rename to controller/src/optimizer/cost/wire.rs diff --git a/controller/src/algebra/optimizer.rs b/controller/src/optimizer/engine.rs similarity index 98% rename from controller/src/algebra/optimizer.rs rename to controller/src/optimizer/engine.rs index f31c7705..ff266bbf 100644 --- a/controller/src/algebra/optimizer.rs +++ b/controller/src/optimizer/engine.rs @@ -1,7 +1,7 @@ //! Cost-based fixed-point query optimizer. //! //! The optimizer applies a set of algebraic rewrite rules to a -//! [`QueryExpr`](super::expr::QueryExpr) tree until no rule fires (fixed +//! [`QueryExpr`](crate::intent_algebra::legacy_expr::QueryExpr) tree until no rule fires (fixed //! point). Each rule is a pure function `QueryExpr → Option`: //! returning `None` means "this rule does not apply here". //! @@ -30,8 +30,8 @@ use std::collections::HashMap; -use super::expr::{QueryExpr, ScalarExpr, SetOpKind, SortKey}; -use super::expr::{AggIntent, PartitionKeys, SourceSpec}; +use crate::intent_algebra::legacy_expr::{QueryExpr, ScalarExpr, SetOpKind, SortKey}; +use crate::intent_algebra::legacy_expr::{AggIntent, PartitionKeys, SourceSpec}; use crate::sketch_algebra::capability::{ default_capability_table, load_capability_overrides, SketchCapability, }; @@ -208,7 +208,7 @@ impl CostModel for DefaultCostModel { let memory = match expr { QueryExpr::SketchAgg { op, .. } | QueryExpr::WindowedAgg { agg: op, .. } => - crate::algebra::directory::estimated_sketch_memory_bytes(op) as f64, + crate::physical::sketch_catalog::estimated_sketch_memory_bytes(op) as f64, _ => self.raw_bytes_per_sec * factor * 0.01, }; @@ -235,7 +235,7 @@ impl CostModel for DefaultCostModel { // For sketch nodes, check sketch capability against stage budget. if let QueryExpr::SketchAgg { op, .. } | QueryExpr::WindowedAgg { agg: op, .. } = expr { - let sketch_type = crate::algebra::directory::sketch_type_for_op(op); + let sketch_type = crate::physical::sketch_catalog::sketch_type_for_op(op); let cap = sketch_capability(&sketch_type); if !stage_budget.fits(&cap) { // Sketch doesn't fit — apply 10× penalty across all dimensions. @@ -247,7 +247,7 @@ impl CostModel for DefaultCostModel { } // Sliding window: penalise if sketch doesn't support it natively. if let QueryExpr::WindowedAgg { window, .. } = expr { - if matches!(window.kind, crate::algebra::expr::WindowKind::Sliding { .. }) + if matches!(window.kind, crate::intent_algebra::legacy_expr::WindowKind::Sliding { .. }) && !cap.supports_sliding_window { return NodeCost { @@ -997,8 +997,8 @@ fn default_rules() -> Vec> { #[cfg(test)] mod tests { use super::*; - use crate::algebra::expr::{LiteralValue, ScalarExpr}; - use crate::algebra::expr::{AggIntent, ColumnRef, PartitionKeys, SourceSpec}; + use crate::intent_algebra::legacy_expr::{LiteralValue, ScalarExpr}; + use crate::intent_algebra::legacy_expr::{AggIntent, ColumnRef, PartitionKeys, SourceSpec}; use std::time::Duration; fn src(name: &str) -> QueryExpr { diff --git a/controller/src/optimizer/mod.rs b/controller/src/optimizer/mod.rs new file mode 100644 index 00000000..185c226a --- /dev/null +++ b/controller/src/optimizer/mod.rs @@ -0,0 +1,36 @@ +//! L4 — **sketch-binding optimizer** framework. +//! +//! Per `controller/docs/design.md` §3/§5/§6 `core::optimizer`: this is +//! the rule engine driver + cost-model trait + rule library that takes +//! L3 [`crate::intent_algebra::legacy_expr::QueryExpr`] / canonical +//! [`crate::intent_algebra::QueryExpr`] inputs and produces L4 +//! sketch-bound output (in the legacy path: an annotated `QueryExpr` +//! with `SketchAgg` nodes; in the canonical path: a +//! [`crate::sketch_algebra::SketchExpr`] DAG). +//! +//! Refactor 2026-05 (`refactor/controller-layered-cleanup`) absorbed +//! the former `controller/src/algebra/optimizer.rs` and the entire +//! former `controller/src/planner/` directory here. Mapping: +//! +//! | Old path | New path | +//! |---|---| +//! | `algebra/optimizer.rs` | [`engine`] (rule driver + `QueryOptimizer` + `DeploymentConstraints`) | +//! | `planner/cost_model.rs` | [`cost`] (cost model trait + per-strategy impls) | +//! | `planner/delta_cost_model.rs` | [`cost::delta`] | +//! | `planner/online_cost_model.rs` | [`cost::online`] | +//! | `planner/pareto.rs` | [`cost::pareto`] | +//! | `planner/tco.rs` | [`cost::tco`] | +//! | `planner/wire_cost.rs` | [`cost::wire`] | +//! | `planner/rules.rs` | [`rules`] (shared rule library) | +//! | `planner/baseline_planner.rs` | [`baseline`] | + +pub mod engine; +pub mod cost; +pub mod rules; +pub mod baseline; +pub mod trait_def; + +// Re-exports — preserve the surface that `crate::algebra::QueryOptimizer` +// and `crate::planner::*` consumers historically relied on. +pub use engine::{DeploymentConstraints, QueryOptimizer}; +pub use trait_def::{OptimizerRule, RuleCategory}; diff --git a/controller/src/planner/rules.rs b/controller/src/optimizer/rules/mod.rs similarity index 99% rename from controller/src/planner/rules.rs rename to controller/src/optimizer/rules/mod.rs index fa55400e..60621ad8 100644 --- a/controller/src/planner/rules.rs +++ b/controller/src/optimizer/rules/mod.rs @@ -327,8 +327,8 @@ impl RulesPlanner { return self.raw_passthrough_plan(w); } - let sketch_type = crate::algebra::directory::sketch_type_for_agg(&w.aggregations); - let sketch_params = crate::algebra::directory::build_sketch_params(&self.sketch_defaults, &sketch_type, w.accuracy_sla, &w.quantiles); + let sketch_type = crate::physical::sketch_catalog::sketch_type_for_agg(&w.aggregations); + let sketch_params = crate::physical::sketch_catalog::build_sketch_params(&self.sketch_defaults, &sketch_type, w.accuracy_sla, &w.quantiles); let (mode, window_duration) = select_window_strategy(w); let mut aggregate_by = w.group_by_labels.clone(); @@ -428,7 +428,7 @@ impl RulesPlanner { // ── Sketch selection (delegated to algebra::directory) ─────────────────────── -pub use crate::algebra::directory::{default_sketch_params, build_sketch_params}; +pub use crate::physical::sketch_catalog::{default_sketch_params, build_sketch_params}; // ── Window strategy ─────────────────────────────────────────────────────────── diff --git a/controller/src/optimizer/trait_def.rs b/controller/src/optimizer/trait_def.rs new file mode 100644 index 00000000..7236c2ec --- /dev/null +++ b/controller/src/optimizer/trait_def.rs @@ -0,0 +1,47 @@ +//! `OptimizerRule` trait + `RuleCategory` enum — placeholder scaffolding +//! for the L4 rule-library surface declared in +//! `controller/docs/design.md` §6 `core::optimizer::trait`. +//! +//! The 2026-05 layered-cleanup refactor created this module so the +//! design.md target layout is materialised in code, but the existing +//! optimizer (`crate::optimizer::engine`) and rule library +//! (`crate::optimizer::rules`) still consume their own concrete types +//! defined in those files. Migrating callers onto this trait is a +//! follow-up — until then this module is intentionally minimal so it +//! has zero behavior impact. + +#![allow(dead_code)] + +/// Category tag for a rule, so callers can enable / disable groups of +/// rules together. +/// +/// Reserved for the future deployment-model registry surface (`crates/ +/// deployment-model-*/src/rules.rs` per design.md §5). +#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)] +pub enum RuleCategory { + /// Push down predicates / projections. + PushDown, + /// Fuse adjacent operators into one. + Fusion, + /// Eliminate redundant operators. + Elim, + /// Bind a logical intent to a sketch family. + Bind, +} + +/// `OptimizerRule` trait — every L4 rule implements this so the engine +/// driver can iterate over them uniformly. +/// +/// **Status:** placeholder shape only. The existing +/// [`crate::sketch_algebra::rules::Rule`] trait covers the L4 binding +/// dispatch on the `intent_algebra` canonical path; the legacy +/// `algebra::optimizer` rule loop runs against `QueryExpr` directly +/// without a polymorphic trait. Unifying both paths under this trait is +/// a follow-up. +pub trait OptimizerRule: Send + Sync { + /// Stable name for diagnostics + rule selection. + fn name(&self) -> &'static str; + + /// Category tag — see [`RuleCategory`]. + fn category(&self) -> RuleCategory; +} diff --git a/controller/src/algebra/allocator.rs b/controller/src/physical/allocator.rs similarity index 98% rename from controller/src/algebra/allocator.rs rename to controller/src/physical/allocator.rs index fe57bead..9649c8aa 100644 --- a/controller/src/algebra/allocator.rs +++ b/controller/src/physical/allocator.rs @@ -27,11 +27,11 @@ //! | PromQLSubquery, BinaryOp | Precompute | has sketch children | //! | LetBinding, Subquery | same as body/inner | propagated | -use super::expr::QueryExpr; +use crate::intent_algebra::legacy_expr::QueryExpr; use super::plan::{ CostEstimate, ExecutionMode, NodeAnnotation, PipelineStage, PlanNode, }; -use super::expr::{AggIntent, ExactAgg}; +use crate::intent_algebra::legacy_expr::{AggIntent, ExactAgg}; use crate::types::{SketchType, StageResourceBudgets}; // ── Resource budget tracker ─────────────────────────────────────────────────── @@ -539,7 +539,7 @@ impl SketchAllocator { fn alloc_sketch_agg( &self, op: AggIntent, - col: super::expr::ColumnRef, + col: crate::intent_algebra::legacy_expr::ColumnRef, child: PlanNode, budget: &mut BudgetState, ) -> PlanNode { @@ -588,7 +588,7 @@ impl SketchAllocator { } // Sketch operators: resolve to physical, then try Agent → Backend → Precompute. - let physical = super::physical::resolve(&op); + let physical = super::planner::resolve(&op); let mem = estimated_sketch_memory(&op); let (sketch_type, params) = (physical.sketch_type, physical.sketch_params); @@ -676,7 +676,7 @@ impl SketchAllocator { /// Estimate the memory footprint of a sketch in bytes. fn estimated_sketch_memory(op: &AggIntent) -> f64 { - super::directory::estimated_sketch_memory_bytes(op) as f64 + super::sketch_catalog::estimated_sketch_memory_bytes(op) as f64 } // sketch_type_for_op delegated to algebra::directory::sketch_type_and_params. @@ -686,9 +686,9 @@ fn estimated_sketch_memory(op: &AggIntent) -> f64 { #[cfg(test)] mod tests { use super::*; - use crate::algebra::expr::QueryExpr; - use crate::algebra::plan::{ExecutionMode, PipelineStage}; - use crate::algebra::expr::{AggIntent, ColumnRef, PartitionKeys, SourceSpec}; + use crate::intent_algebra::legacy_expr::QueryExpr; + use crate::physical::plan::{ExecutionMode, PipelineStage}; + use crate::intent_algebra::legacy_expr::{AggIntent, ColumnRef, PartitionKeys, SourceSpec}; use crate::types::{SketchType, StageResourceBudgets}; use std::time::Duration; @@ -732,7 +732,7 @@ mod tests { #[test] fn filter_at_agent() { - use crate::algebra::expr::{LiteralValue, ScalarExpr}; + use crate::intent_algebra::legacy_expr::{LiteralValue, ScalarExpr}; let expr = QueryExpr::Filter { pred: ScalarExpr::Literal(LiteralValue::Bool(true)), input: Box::new(src("m")), @@ -869,7 +869,7 @@ mod tests { #[test] fn join_goes_to_db() { - use crate::algebra::expr::JoinKind; + use crate::intent_algebra::legacy_expr::JoinKind; let expr = QueryExpr::Join { kind: JoinKind::Inner, pred: None, diff --git a/controller/src/stage_split/allocator.rs b/controller/src/physical/colored_dag/allocator.rs similarity index 98% rename from controller/src/stage_split/allocator.rs rename to controller/src/physical/colored_dag/allocator.rs index 9113228c..d4c6ea4c 100644 --- a/controller/src/stage_split/allocator.rs +++ b/controller/src/physical/colored_dag/allocator.rs @@ -39,8 +39,8 @@ use std::collections::HashMap; use crate::sketch_algebra::SketchExpr; -use crate::stage_split::colored_dag::{ColoredDag, ColoredNode, NodeId}; -use crate::stage_split::stage_id::{StageId, Topology}; +use crate::physical::colored_dag::dag::{ColoredDag, ColoredNode, NodeId}; +use crate::physical::colored_dag::stage_id::{StageId, Topology}; use crate::types_v2::BindingName; /// Errors surfaced by [`StageAllocator::allocate`]. diff --git a/controller/src/stage_split/colored_dag.rs b/controller/src/physical/colored_dag/dag.rs similarity index 96% rename from controller/src/stage_split/colored_dag.rs rename to controller/src/physical/colored_dag/dag.rs index eff1167f..671c8c93 100644 --- a/controller/src/stage_split/colored_dag.rs +++ b/controller/src/physical/colored_dag/dag.rs @@ -4,7 +4,7 @@ //! is a colouring of the L4-bound `SketchExpr` DAG by `StageId`, with //! sketch-merge / data-shipping nodes inserted on the cut edges." //! -//! [`ColoredDag`] is the IR the [`crate::stage_split::Emitter`] +//! [`ColoredDag`] is the IR the [`crate::physical::colored_dag::Emitter`] //! consumes. It carries: //! //! - the `nodes` vector (every visited `SketchExpr` node keyed by @@ -23,7 +23,7 @@ use serde::{Deserialize, Serialize}; use crate::sketch_algebra::SketchExpr; -use crate::stage_split::stage_id::{StageId, Topology}; +use crate::physical::colored_dag::stage_id::{StageId, Topology}; /// Stable position-based identifier for a node within a `ColoredDag`. /// `NodeId(0)` is the root; depth-first walk order otherwise. @@ -57,7 +57,7 @@ pub struct ColoredNode { pub stage: StageId, } -/// L5 colored DAG. The output of [`crate::stage_split::StageAllocator::allocate`]. +/// L5 colored DAG. The output of [`crate::physical::colored_dag::StageAllocator::allocate`]. /// /// Per design.md §6: "L5 colors the DAG by `StageId` and emits /// per-executor configs. Same `SketchExpr` input; topology and emitter diff --git a/controller/src/stage_split/emitter.rs b/controller/src/physical/colored_dag/emitter.rs similarity index 98% rename from controller/src/stage_split/emitter.rs rename to controller/src/physical/colored_dag/emitter.rs index e6d357c6..22e8edf8 100644 --- a/controller/src/stage_split/emitter.rs +++ b/controller/src/physical/colored_dag/emitter.rs @@ -33,8 +33,8 @@ use serde::{Deserialize, Serialize}; use crate::sketch_algebra::params::{SketchKind, SketchParams}; use crate::sketch_algebra::sketch_expr::{EstimateOp, SketchExpr}; -use crate::stage_split::colored_dag::ColoredDag; -use crate::stage_split::stage_id::{StageId, Topology}; +use crate::physical::colored_dag::dag::ColoredDag; +use crate::physical::colored_dag::stage_id::{StageId, Topology}; /// Errors surfaced by [`Emitter::emit_per_stage`]. #[derive(Debug, thiserror::Error, PartialEq)] @@ -615,8 +615,8 @@ fn extract_edge_facts(qe: &crate::intent_algebra::QueryExpr, edge: &mut EdgeStag /// deterministic. fn children_of<'a>( dag: &'a ColoredDag, - parent: crate::stage_split::colored_dag::NodeId, -) -> impl Iterator + 'a { + parent: crate::physical::colored_dag::dag::NodeId, +) -> impl Iterator + 'a { dag.edges .iter() .filter(move |(p, _)| *p == parent) @@ -628,7 +628,7 @@ fn children_of<'a>( /// pull the right aggregation_id from `sketch_agg_ids`. fn first_sketch_child_via_edges( dag: &ColoredDag, - parent: crate::stage_split::colored_dag::NodeId, + parent: crate::physical::colored_dag::dag::NodeId, sketch_agg_ids: &HashMap, ) -> Option<(SketchKind, String)> { for cid in children_of(dag, parent) { @@ -655,7 +655,7 @@ fn first_sketch_child_via_edges( _ => None, }) { - let bnode_id = crate::stage_split::colored_dag::NodeId(bid); + let bnode_id = crate::physical::colored_dag::dag::NodeId(bid); if let Some(found) = first_sketch_child_via_edges(dag, bnode_id, sketch_agg_ids) { return Some(found); @@ -673,7 +673,7 @@ fn first_sketch_child_via_edges( /// path (we only need the id, not the kind). fn resolve_descendant_agg_id_via_edges( dag: &ColoredDag, - parent: crate::stage_split::colored_dag::NodeId, + parent: crate::physical::colored_dag::dag::NodeId, sketch_agg_ids: &HashMap, ) -> Option { first_sketch_child_via_edges(dag, parent, sketch_agg_ids).map(|(_, aid)| aid) diff --git a/controller/src/stage_split/mod.rs b/controller/src/physical/colored_dag/mod.rs similarity index 88% rename from controller/src/stage_split/mod.rs rename to controller/src/physical/colored_dag/mod.rs index 368d3338..d60d9bad 100644 --- a/controller/src/stage_split/mod.rs +++ b/controller/src/physical/colored_dag/mod.rs @@ -36,16 +36,18 @@ #![allow(dead_code, unused_imports)] pub mod allocator; -pub mod colored_dag; +pub mod dag; pub mod emitter; pub mod stage_id; #[cfg(test)] mod tests; -// Re-exports — `crate::stage_split::*` for downstream callers. +// Re-exports — `crate::physical::colored_dag::*` for downstream callers. +// The historical `crate::physical::colored_dag::*` path is preserved via a +// re-export in `lib.rs` so existing callers see no source-level break. pub use allocator::{AllocateError, StageAllocator}; -pub use colored_dag::{ColoredDag, ColoredNode, NodeId}; +pub use dag::{ColoredDag, ColoredNode, NodeId}; pub use emitter::{ ArchiveTierMetric, BackendAggregation, BackendReadout, BackendStageConfig, EdgeSketchProcessor, EdgeStageConfig, EmitError, Emitter, ExportTarget, GatewayMergeProcessor, GatewayStageConfig, diff --git a/controller/src/stage_split/stage_id.rs b/controller/src/physical/colored_dag/stage_id.rs similarity index 100% rename from controller/src/stage_split/stage_id.rs rename to controller/src/physical/colored_dag/stage_id.rs diff --git a/controller/src/stage_split/tests.rs b/controller/src/physical/colored_dag/tests.rs similarity index 98% rename from controller/src/stage_split/tests.rs rename to controller/src/physical/colored_dag/tests.rs index 4331b377..fef621e6 100644 --- a/controller/src/stage_split/tests.rs +++ b/controller/src/physical/colored_dag/tests.rs @@ -15,9 +15,9 @@ use crate::sketch_algebra::params::{ DDSketchParams, HllParams, KllParams, SketchKind, SketchParams, }; use crate::sketch_algebra::sketch_expr::{EstimateOp, MergeAlgebra, SketchExpr}; -use crate::stage_split::allocator::StageAllocator; -use crate::stage_split::emitter::{EmitError, Emitter, StageConfig, ThreeStageEmitter}; -use crate::stage_split::stage_id::{StageId, Topology}; +use crate::physical::colored_dag::allocator::StageAllocator; +use crate::physical::colored_dag::emitter::{EmitError, Emitter, StageConfig, ThreeStageEmitter}; +use crate::physical::colored_dag::stage_id::{StageId, Topology}; use crate::types_v2::{AccuracyTarget, BindingName}; // ── Test fixtures ───────────────────────────────────────────────────────────── diff --git a/controller/src/physical/mod.rs b/controller/src/physical/mod.rs new file mode 100644 index 00000000..24649729 --- /dev/null +++ b/controller/src/physical/mod.rs @@ -0,0 +1,34 @@ +//! L5 — **physical execution plan** framework. +//! +//! Per `controller/docs/design.md` §3/§5/§6 `core::physical`: this is +//! the stage allocator + sketch catalogue + physical planner surface. +//! Refactor 2026-05 (`refactor/controller-layered-cleanup`) absorbed +//! the former `controller/src/algebra/{allocator,physical,plan}.rs` +//! + `controller/src/algebra/directory.rs` (renamed to +//! [`sketch_catalog`]) + the L5 colored-DAG framework formerly at +//! `controller/src/stage_split/` ([`colored_dag`] subdir) here. +//! +//! Module layout: +//! +//! | File | Role | +//! |---|---| +//! | [`allocator`] | `SketchAllocator` — assigns physical ops to pipeline stages over the legacy [`crate::intent_algebra::legacy_expr::QueryExpr`] IR | +//! | [`planner`] | `PhysicalPlanner` — `QueryExpr` → staged sub-plans (DC three-stage emission) | +//! | [`plan`] | `PlanNode` / `PlanSummary` — annotated plan tree with cost estimates + stage colouring | +//! | [`sketch_catalog`] | Candidate sketch types per `AggIntent`, default `SketchParams`, memory estimation | +//! | [`stage_split`] | AST-aware hierarchical stage assignment (legacy `QueryExpr` splitter) | +//! | [`colored_dag`] | L5 typed colouring framework over the canonical [`crate::sketch_algebra::SketchExpr`] DAG (`StageId` + `Topology` + per-stage emitter) | +//! | [`topology`] | Re-exports `StageId` + `Topology` from [`colored_dag::stage_id`] — design.md §5 entry point for deployment-topology descriptors | + +pub mod allocator; +pub mod plan; +pub mod planner; +pub mod sketch_catalog; +pub mod stage_split; +pub mod colored_dag; +pub mod topology; + +// Convenience re-exports — preserve the surface that consumers of the +// former `algebra` module relied on. +pub use allocator::SketchAllocator; +pub use plan::{CostEstimate, ExecutionMode, PipelineStage, PlanNode, PlanSummary}; diff --git a/controller/src/algebra/plan.rs b/controller/src/physical/plan.rs similarity index 98% rename from controller/src/algebra/plan.rs rename to controller/src/physical/plan.rs index 4cf237ae..1c30e044 100644 --- a/controller/src/algebra/plan.rs +++ b/controller/src/physical/plan.rs @@ -1,6 +1,6 @@ //! Annotated plan nodes — the output of the [`super::allocator::SketchAllocator`]. //! -//! After the optimizer rewrites a [`QueryExpr`](super::expr::QueryExpr) tree, +//! After the optimizer rewrites a [`QueryExpr`](crate::intent_algebra::legacy_expr::QueryExpr) tree, //! the allocator wraps every node in a [`PlanNode`] that carries: //! //! * **`stage`** — which pipeline component executes this operator. @@ -15,7 +15,7 @@ use serde::{Deserialize, Serialize}; -use super::expr::QueryExpr; +use crate::intent_algebra::legacy_expr::QueryExpr; // ── Pipeline stages ─────────────────────────────────────────────────────────── @@ -271,8 +271,8 @@ impl PlanNode { #[cfg(test)] mod tests { use super::*; - use crate::algebra::expr::QueryExpr; - use crate::algebra::expr::SourceSpec; + use crate::intent_algebra::legacy_expr::QueryExpr; + use crate::intent_algebra::legacy_expr::SourceSpec; fn source_node(name: &str, stage: PipelineStage) -> PlanNode { PlanNode::leaf( diff --git a/controller/src/algebra/physical.rs b/controller/src/physical/planner.rs similarity index 98% rename from controller/src/algebra/physical.rs rename to controller/src/physical/planner.rs index 8d53cf70..ffb945c0 100644 --- a/controller/src/algebra/physical.rs +++ b/controller/src/physical/planner.rs @@ -12,8 +12,8 @@ use std::time::Duration; -use crate::algebra::directory; -use crate::algebra::expr::{AggIntent, WindowKind, WindowSpec}; +use crate::physical::sketch_catalog; +use crate::intent_algebra::legacy_expr::{AggIntent, WindowKind, WindowSpec}; use crate::types::{SketchParams, SketchType}; // ── PhysicalAggOp (resolved sketch intent) ────────────────────────────────── @@ -40,9 +40,9 @@ pub struct PhysicalAggOp { pub fn resolve(intent: &AggIntent) -> PhysicalAggOp { PhysicalAggOp { intent: intent.clone(), - sketch_type: directory::sketch_type_for_op(intent), - sketch_params: directory::sketch_params_for_op(intent), - estimated_memory_bytes: directory::estimated_sketch_memory_bytes(intent), + sketch_type: sketch_catalog::sketch_type_for_op(intent), + sketch_params: sketch_catalog::sketch_params_for_op(intent), + estimated_memory_bytes: sketch_catalog::estimated_sketch_memory_bytes(intent), } } @@ -214,8 +214,8 @@ pub fn resolve_window(window: &WindowSpec, placement: &Placement) -> PhysicalWin // ── Physical planner ──────────────────────────────────────────────────────── -use crate::algebra::expr::*; -use crate::algebra::optimizer::DeploymentConstraints; +use crate::intent_algebra::legacy_expr::*; +use crate::optimizer::engine::DeploymentConstraints; use crate::types::StageResourceBudgets; /// Physical planner configuration. @@ -466,7 +466,7 @@ fn plan_node(expr: &QueryExpr, config: &PhysicalPlannerConfig) -> PhysicalNode { /// Uses `StageBudget::fits(SketchCapability)` to check each stage in order: /// Agent → BackendCollector → QueryEngine. fn decide_sketch_placement(resolved: &PhysicalAggOp, config: &PhysicalPlannerConfig) -> Placement { - use crate::algebra::optimizer::sketch_capability; + use crate::optimizer::engine::sketch_capability; let cap = sketch_capability(&resolved.sketch_type); diff --git a/controller/src/algebra/directory.rs b/controller/src/physical/sketch_catalog.rs similarity index 99% rename from controller/src/algebra/directory.rs rename to controller/src/physical/sketch_catalog.rs index 925ab15d..382ab9e9 100644 --- a/controller/src/algebra/directory.rs +++ b/controller/src/physical/sketch_catalog.rs @@ -9,7 +9,7 @@ //! //! All callers now go through this module. -use crate::algebra::expr::{AggIntent, ExactAgg}; +use crate::intent_algebra::legacy_expr::{AggIntent, ExactAgg}; use crate::types::{ AggType, SketchDefaults, SketchParams, SketchType, diff --git a/controller/src/planner/stage_split.rs b/controller/src/physical/stage_split.rs similarity index 97% rename from controller/src/planner/stage_split.rs rename to controller/src/physical/stage_split.rs index 3a891b32..7ae16115 100644 --- a/controller/src/planner/stage_split.rs +++ b/controller/src/physical/stage_split.rs @@ -37,10 +37,10 @@ use std::time::Duration; -use crate::algebra::expr::{AggFunc, BinaryOpKind, LiteralValue, QueryExpr, ScalarExpr}; -use crate::analyzer::format_duration; -use crate::algebra::expr::{AggIntent, ExactAgg}; -use crate::algebra::directory; +use crate::intent_algebra::legacy_expr::{AggFunc, BinaryOpKind, LiteralValue, QueryExpr, ScalarExpr}; +use crate::pipeline::format_duration; +use crate::intent_algebra::legacy_expr::{AggIntent, ExactAgg}; +use crate::physical::sketch_catalog; use crate::types::{ AgentSubPlan, BackendSubPlan, DbSubPlan, PrecomputeSubPlan, SketchParams, SketchType, @@ -114,7 +114,7 @@ fn walk(expr: &QueryExpr, plan: &mut StagedPlan, budgets: &StageResourceBudgets) // WindowedAgg — bundles window + sketch agg intent. QueryExpr::WindowedAgg { agg, window, input, .. } => { - if let crate::algebra::expr::WindowKind::Tumbling { size } = &window.kind { + if let crate::intent_algebra::legacy_expr::WindowKind::Tumbling { size } = &window.kind { plan.agent.window_secs = Some(size.as_secs()); } assign_sketch_agg(agg, plan, budgets); @@ -246,7 +246,7 @@ fn assign_sketch_agg(op: &AggIntent, plan: &mut StagedPlan, budgets: &StageResou } // Sketch ops: resolve to physical, assign to Agent, defer if budget exceeded. sketch_op => { - let physical = crate::algebra::physical::resolve(sketch_op); + let physical = crate::physical::planner::resolve(sketch_op); let stage = resolve_sketch_stage(physical.estimated_memory_bytes, budgets, &mut plan.deferral_log, sketch_op); match stage { SketchStage::Agent => { @@ -381,7 +381,7 @@ fn promql_from_qe(expr: &QueryExpr, ctx: &mut PromQLCtx) -> String { // ── WindowedAgg — bundled window + sketch agg ──────────────────────── QueryExpr::WindowedAgg { agg, window, input, .. } => { if ctx.window.is_none() { - if let crate::algebra::expr::WindowKind::Tumbling { size } = &window.kind { + if let crate::intent_algebra::legacy_expr::WindowKind::Tumbling { size } = &window.kind { ctx.window = Some(*size); } } @@ -455,7 +455,7 @@ fn promql_from_qe(expr: &QueryExpr, ctx: &mut PromQLCtx) -> String { let match_str = vector_match .as_ref() .map(|m| { - use crate::algebra::expr::{GroupSide, VectorMatchKind}; + use crate::intent_algebra::legacy_expr::{GroupSide, VectorMatchKind}; let kw = match m.kind { VectorMatchKind::On => "on", VectorMatchKind::Ignoring => "ignoring", @@ -654,7 +654,7 @@ fn collect_label_filters_into(pred: &ScalarExpr, out: &mut Vec) { // ── Phase E: typed L5 call path (opt-in, additive) ──────────────────────────── /// Env-var that opts the planner into the typed L5 stage_split path -/// (`crate::stage_split::StageAllocator` + `ThreeStageEmitter`). +/// (`crate::physical::colored_dag::StageAllocator` + `ThreeStageEmitter`). /// Additive — when unset, the existing untyped `split_expr_by_stage` /// flow runs unchanged. Mirror of `ENV_USE_TYPED_SKETCH_ALGEBRA` from /// Phase C. @@ -667,7 +667,7 @@ pub const ENV_USE_TYPED_STAGE_SPLIT: &str = "USE_TYPED_STAGE_SPLIT"; /// Reads the env var once per call (cheap; called per `plan()` /// invocation at most). Phase E is additive — both code paths produce /// per-stage descriptions, but the typed path's structural output is -/// `crate::stage_split::StageConfig` (sketched against design.md §6), +/// `crate::physical::colored_dag::StageConfig` (sketched against design.md §6), /// while the legacy path is the existing `StagedPlan` shape. /// /// Phase B (MVP v6) wires `main::handle_plan` to consult this gate. @@ -679,7 +679,7 @@ pub fn typed_stage_split_enabled() -> bool { } /// Run the typed L5 path on a Phase-C-bound `SketchExpr` DAG. Returns -/// the per-stage [`crate::stage_split::StageConfig`] map for the DC +/// the per-stage [`crate::physical::colored_dag::StageConfig`] map for the DC /// lifecycle topology. /// /// Returns `None` when the typed path errors out (unsupported topology @@ -696,9 +696,9 @@ pub fn typed_stage_split_enabled() -> bool { /// `Backend`. Phase C plumbs deployment-aware endpoint resolution. pub fn split_typed_three_stage( expr: &crate::sketch_algebra::SketchExpr, -) -> Option> +) -> Option> { - use crate::stage_split::{ + use crate::physical::colored_dag::{ Emitter, StageAllocator, ThreeStageEmitter, Topology, }; let dag = StageAllocator.allocate(expr, Topology::ThreeStage).ok()?; @@ -710,8 +710,8 @@ pub fn split_typed_three_stage( #[cfg(test)] mod tests { use super::*; - use crate::algebra::expr::{AggItem, AggIntent, BinaryOpKind, LiteralValue, QueryExpr, ScalarExpr}; - use crate::algebra::expr::{ColumnRef, PartitionKeys, SourceSpec}; + use crate::intent_algebra::legacy_expr::{AggItem, AggIntent, BinaryOpKind, LiteralValue, QueryExpr, ScalarExpr}; + use crate::intent_algebra::legacy_expr::{ColumnRef, PartitionKeys, SourceSpec}; fn source(name: &str) -> QueryExpr { QueryExpr::Source(SourceSpec { name: name.into() }) @@ -1010,7 +1010,7 @@ mod tests { }; use crate::sketch_algebra::params::{KllParams, SketchKind, SketchParams as L4Params}; use crate::sketch_algebra::sketch_expr::{EstimateOp, SketchExpr}; - use crate::stage_split::StageId; + use crate::physical::colored_dag::StageId; let scan = L3QE::Scan { source: L3Source::TimeSeries { metric: "http_request_duration_seconds".into(), diff --git a/controller/src/physical/topology.rs b/controller/src/physical/topology.rs new file mode 100644 index 00000000..6980499b --- /dev/null +++ b/controller/src/physical/topology.rs @@ -0,0 +1,16 @@ +//! L5 deployment-topology descriptors. +//! +//! Per `controller/docs/design.md` §5 / §6 `core::physical::topology`: +//! the topology declares which stages exist (3-stage / 1-stage / +//! 0-stage). Stages are roles, not instances — see also +//! [`crate::physical::colored_dag::stage_id`] for the underlying +//! `StageId` + `Topology` enums. +//! +//! Refactor 2026-05: the `StageId` + `Topology` types currently live in +//! [`super::colored_dag::stage_id`] (phase E delivered the typed-L5 +//! framework there). This module re-exports them so the design.md §5 +//! `core::physical::topology` entry point exists in code; future +//! deployment-topology descriptor extensions land here without +//! disturbing the colored-DAG framework. + +pub use super::colored_dag::stage_id::{StageId, Topology}; diff --git a/controller/src/analyzer.rs b/controller/src/pipeline.rs similarity index 100% rename from controller/src/analyzer.rs rename to controller/src/pipeline.rs diff --git a/controller/src/planner/mod.rs b/controller/src/planner/mod.rs deleted file mode 100644 index ef07aef7..00000000 --- a/controller/src/planner/mod.rs +++ /dev/null @@ -1,19 +0,0 @@ -pub mod rules; -pub mod cost_model; -pub mod delta_cost_model; -pub mod online_cost_model; -pub mod pareto; -pub mod baseline_planner; -pub mod stage_split; -pub mod tco; -pub mod wire_cost; - -pub use rules::RulesPlanner; -pub use cost_model::CostModelPlanner; -pub use baseline_planner::BaselinePlanner; -pub use online_cost_model::{OnlineMetricsStore, init_store as init_online_store}; -pub use pareto::{ObjectiveWeights, ParetoPoint, pareto_frontier, select_best}; -pub use wire_cost::{ - break_even_samples, est_wire_bytes_per_window_per_series, select_bind_mode, BindMode, - SketchWireCost, WireCostTable, WireWorkload, -}; diff --git a/controller/src/query_language/mod.rs b/controller/src/query_language/mod.rs deleted file mode 100644 index 1dd7e460..00000000 --- a/controller/src/query_language/mod.rs +++ /dev/null @@ -1,45 +0,0 @@ -// Layer 1 scaffolding ships ahead of any in-tree call site (the -// downstream `pipeline` driver lands in a later phase). Dead-code -// warnings are silenced here, not on individual items, so the public -// surface is uncluttered. -#![allow(dead_code, unused_imports)] - -//! Layer 1 — `query_language`. -//! -//! Per-language parser façade. See `controller/docs/design.md` §6 -//! `core::query_language` for the design contract. -//! -//! # Module layout -//! -//! ```text -//! query_language/ -//! ├── language.rs — Language trait + ParseError -//! ├── language_ast.rs — LanguageAst sum type -//! ├── promql/ — PromQL backend (active; wraps query_parser::promql) -//! ├── sql/ — SQL backend (stub; Unimplemented) -//! └── elastic_dsl/ — ElasticDSL backend (stub; Unimplemented) -//! ``` -//! -//! # DC deployment scope -//! -//! Only [`promql::PromQLLanguage`] is implemented. The other backends -//! return [`language::ParseError::Unimplemented`] so the type system is -//! uniform; adding a real backend later is a localised change to the -//! relevant sub-module. - -pub mod language; -pub mod language_ast; - -pub mod promql; -pub mod sql; -pub mod elastic_dsl; - -pub use language::{Language, ParseError}; -pub use language_ast::LanguageAst; - -pub use promql::PromQLLanguage; -pub use sql::SqlLanguage; -pub use elastic_dsl::ElasticDslLanguage; - -#[cfg(test)] -mod tests; diff --git a/controller/src/query_language/elastic_dsl/mod.rs b/controller/src/query_parser/language/elastic_dsl/mod.rs similarity index 94% rename from controller/src/query_language/elastic_dsl/mod.rs rename to controller/src/query_parser/language/elastic_dsl/mod.rs index 836224c9..8b20f3f5 100644 --- a/controller/src/query_language/elastic_dsl/mod.rs +++ b/controller/src/query_parser/language/elastic_dsl/mod.rs @@ -4,7 +4,7 @@ //! (`controller/docs/design.md` §3 row 1). No L1 parser is shipped in //! the DC deployment build. -use super::language::{Language, ParseError}; +use super::{Language, ParseError}; use super::language_ast::LanguageAst; use crate::types_v2::QueryLanguage; diff --git a/controller/src/query_language/language_ast.rs b/controller/src/query_parser/language/language_ast.rs similarity index 100% rename from controller/src/query_language/language_ast.rs rename to controller/src/query_parser/language/language_ast.rs diff --git a/controller/src/query_language/language.rs b/controller/src/query_parser/language/mod.rs similarity index 56% rename from controller/src/query_language/language.rs rename to controller/src/query_parser/language/mod.rs index 3c324cce..bdee934a 100644 --- a/controller/src/query_language/language.rs +++ b/controller/src/query_parser/language/mod.rs @@ -1,14 +1,49 @@ -//! L1 `Language` trait — abstract over the per-language parser. +// L1 scaffolding ships ahead of any in-tree call site (the +// downstream `pipeline` driver lands in a later phase). Dead-code +// warnings are silenced here, not on individual items, so the public +// surface is uncluttered. +#![allow(dead_code, unused_imports)] + +//! Layer 1 — `query_parser::language`. +//! +//! Per-language parser façade. See `controller/docs/design.md` §6 +//! `core::query_language` for the design contract. Refactor 2026-05: +//! the former top-level `query_language/` module was folded into +//! `query_parser/language/` so the L1 parser frontend lives under one +//! module path. +//! +//! # Module layout +//! +//! ```text +//! query_parser/language/ +//! ├── mod.rs — Language trait + ParseError (this file) +//! ├── language_ast.rs — LanguageAst sum type +//! ├── promql/ — PromQL backend (active; wraps query_parser::promql) +//! ├── sql/ — SQL backend (stub; Unimplemented) +//! └── elastic_dsl/ — ElasticDSL backend (stub; Unimplemented) +//! ``` //! -//! See `controller/docs/design.md` §6 `core::query_language`. +//! # DC deployment scope //! -//! Every backend (PromQL, SQL, ElasticDSL) implements -//! [`Language`] and returns a [`LanguageAst`] variant tagged with the -//! same [`QueryLanguage`] enum the public `QuerySpec` carries. +//! Only [`promql::PromQLLanguage`] is implemented. The other backends +//! return [`ParseError::Unimplemented`] so the type system is uniform; +//! adding a real backend later is a localised change to the relevant +//! sub-module. -use crate::types_v2::QueryLanguage; +pub mod language_ast; +pub mod promql; +pub mod sql; +pub mod elastic_dsl; + +pub use language_ast::LanguageAst; +pub use promql::PromQLLanguage; +pub use sql::SqlLanguage; +pub use elastic_dsl::ElasticDslLanguage; -use super::language_ast::LanguageAst; +#[cfg(test)] +mod tests; + +use crate::types_v2::QueryLanguage; // ── Errors ──────────────────────────────────────────────────────────────────── diff --git a/controller/src/query_language/promql/ast.rs b/controller/src/query_parser/language/promql/ast.rs similarity index 96% rename from controller/src/query_language/promql/ast.rs rename to controller/src/query_parser/language/promql/ast.rs index 86ccfe19..5e3993fc 100644 --- a/controller/src/query_language/promql/ast.rs +++ b/controller/src/query_parser/language/promql/ast.rs @@ -5,7 +5,7 @@ //! downstream L2 lowering can pick whichever shape it needs without //! re-parsing. -use crate::algebra::expr::QueryExpr; +use crate::intent_algebra::legacy_expr::QueryExpr; use crate::query_parser::ParsedQuery; /// PromQL parse output. Carries both the algebra tree and the flat diff --git a/controller/src/query_language/promql/mod.rs b/controller/src/query_parser/language/promql/mod.rs similarity index 96% rename from controller/src/query_language/promql/mod.rs rename to controller/src/query_parser/language/promql/mod.rs index 9f099317..3f293930 100644 --- a/controller/src/query_language/promql/mod.rs +++ b/controller/src/query_parser/language/promql/mod.rs @@ -7,7 +7,7 @@ pub mod ast; -use super::language::{Language, ParseError}; +use super::{Language, ParseError}; use super::language_ast::LanguageAst; use crate::query_parser::{parse_query, parse_query_expr}; use crate::types_v2::QueryLanguage; diff --git a/controller/src/query_language/sql/mod.rs b/controller/src/query_parser/language/sql/mod.rs similarity index 95% rename from controller/src/query_language/sql/mod.rs rename to controller/src/query_parser/language/sql/mod.rs index ac7ee4d1..2e4b184d 100644 --- a/controller/src/query_language/sql/mod.rs +++ b/controller/src/query_parser/language/sql/mod.rs @@ -5,7 +5,7 @@ //! is not exposed via the new [`Language`] trait yet. A real impl wraps //! `sqlparser` and emits a `SqlAst` analogous to [`super::promql::PromQLAst`]. -use super::language::{Language, ParseError}; +use super::{Language, ParseError}; use super::language_ast::LanguageAst; use crate::types_v2::QueryLanguage; diff --git a/controller/src/query_language/tests.rs b/controller/src/query_parser/language/tests.rs similarity index 100% rename from controller/src/query_language/tests.rs rename to controller/src/query_parser/language/tests.rs diff --git a/controller/src/query_parser/mod.rs b/controller/src/query_parser/mod.rs index fe4acdc3..c2ccbf4c 100644 --- a/controller/src/query_parser/mod.rs +++ b/controller/src/query_parser/mod.rs @@ -29,11 +29,17 @@ pub mod promql; pub mod sql; +pub mod language; + +// Re-export the L1 language façade at this module's top so call sites +// that historically used `crate::query_language::*` (folded in by the +// 2026-05 refactor) keep a one-level import. +pub use language::{Language, LanguageAst, ParseError, PromQLLanguage, SqlLanguage, ElasticDslLanguage}; use std::collections::HashMap; use std::time::Duration; -use crate::algebra::expr::{AggIntent, QueryExpr}; +use crate::intent_algebra::legacy_expr::{AggIntent, QueryExpr}; use crate::types::AggType; // ── Output types (legacy — consumed by analyzer and planner) ────────────────── @@ -100,7 +106,7 @@ pub fn parse_query_expr(query: &str) -> anyhow::Result { promql::parse_promql_expr(q)? }; // Layer 2 → Layer 3 lowering (shared by both languages). - Ok(crate::algebra::lower::lower_to_sketch_algebra(layer2)) + Ok(crate::intent_algebra::legacy_lower::lower_to_sketch_algebra(layer2)) } /// Parse a raw query string (PromQL or SQL) into a [`ParsedQuery`]. @@ -136,7 +142,7 @@ struct QeCollector { impl QeCollector { fn visit(&mut self, expr: &QueryExpr) { - use crate::algebra::expr::{FilterOp, FilterVal, LiteralValue, ScalarExpr}; + use crate::intent_algebra::legacy_expr::{FilterOp, FilterVal, LiteralValue, ScalarExpr}; match expr { QueryExpr::Source(s) => { if self.metric_name.is_none() { @@ -168,7 +174,7 @@ impl QeCollector { } QueryExpr::WindowedAgg { agg, window, input, .. } => { if self.time_window.is_none() { - if let crate::algebra::expr::WindowKind::Tumbling { size } = &window.kind { + if let crate::intent_algebra::legacy_expr::WindowKind::Tumbling { size } = &window.kind { self.time_window = Some(*size); } } @@ -223,8 +229,8 @@ impl QeCollector { } } - fn collect_agg_func_with_group(&mut self, func: &crate::algebra::expr::AggFunc, has_group_by: bool) { - use crate::algebra::expr::AggFunc; + fn collect_agg_func_with_group(&mut self, func: &crate::intent_algebra::legacy_expr::AggFunc, has_group_by: bool) { + use crate::intent_algebra::legacy_expr::AggFunc; // COUNT(*) without GROUP BY → exact (no sketch benefit), unless inside topk // where Count means frequency counting. if matches!(func, AggFunc::Count) && !has_group_by && !self.inside_topk { @@ -234,8 +240,8 @@ impl QeCollector { self.collect_agg_func(func); } - fn collect_agg_func(&mut self, func: &crate::algebra::expr::AggFunc) { - use crate::algebra::expr::AggFunc; + fn collect_agg_func(&mut self, func: &crate::intent_algebra::legacy_expr::AggFunc) { + use crate::intent_algebra::legacy_expr::AggFunc; match func { AggFunc::CountDistinct => { if !self.agg_types.contains(&AggType::Cardinality) { @@ -353,10 +359,10 @@ impl QeCollector { } fn collect_filters_from_scalar( - pred: &crate::algebra::expr::ScalarExpr, + pred: &crate::intent_algebra::legacy_expr::ScalarExpr, out: &mut HashMap, ) { - use crate::algebra::expr::{BinaryOpKind, LiteralValue, ScalarExpr}; + use crate::intent_algebra::legacy_expr::{BinaryOpKind, LiteralValue, ScalarExpr}; match pred { ScalarExpr::BinaryOp { op: BinaryOpKind::Eq, lhs, rhs } => { if let (ScalarExpr::Column(col), ScalarExpr::Literal(LiteralValue::Str(v))) = @@ -456,7 +462,7 @@ mod tests { #[cfg(test)] mod doc_verify_all { use super::*; - use crate::algebra::expr::*; + use crate::intent_algebra::legacy_expr::*; #[test] fn example4_promql_quantile() { diff --git a/controller/src/query_parser/promql.rs b/controller/src/query_parser/promql.rs index 2c61bdae..c6658c0b 100644 --- a/controller/src/query_parser/promql.rs +++ b/controller/src/query_parser/promql.rs @@ -35,7 +35,7 @@ use std::time::Duration; use anyhow::anyhow; use promql_parser::parser::{self, AggregateExpr, Call, Expr, LabelModifier, MatrixSelector, VectorSelector}; -use crate::algebra::expr::{FilterOp, FilterVal, PartitionKeys, Predicate}; +use crate::intent_algebra::legacy_expr::{FilterOp, FilterVal, PartitionKeys, Predicate}; // ── Walk context ────────────────────────────────────────────────────────────── @@ -153,7 +153,7 @@ fn modifier_to_partition(modifier: &LabelModifier) -> PartitionKeys { // | `m[5m:1m]` subquery | PromQLSubquery { 5m, Some(1m) } | // | `a op b` binary | BinaryOp { VectorMatch } | -use crate::algebra::expr::{ +use crate::intent_algebra::legacy_expr::{ AggFunc, AggItem, BinaryOpKind, ColumnRef as QeColumnRef, GroupSide, PartitionKeys as QePartitionKeys, QueryExpr, @@ -479,7 +479,7 @@ fn apply_qe_filters( if filters.is_empty() { input } else { - use crate::algebra::expr::{BinaryOpKind, LiteralValue, ScalarExpr}; + use crate::intent_algebra::legacy_expr::{BinaryOpKind, LiteralValue, ScalarExpr}; let pred = filters.iter().fold( ScalarExpr::Literal(LiteralValue::Bool(true)), |acc, p| { diff --git a/controller/src/query_parser/sql.rs b/controller/src/query_parser/sql.rs index 3ef1440d..7eb7cecc 100644 --- a/controller/src/query_parser/sql.rs +++ b/controller/src/query_parser/sql.rs @@ -48,7 +48,7 @@ use sqlparser::ast::{ }; use sqlparser::dialect::GenericDialect; -use crate::algebra::expr::{ +use crate::intent_algebra::legacy_expr::{ AggFunc, AggItem as AlgAggItem, BinaryOpKind, ColumnRef, JoinKind, LiteralValue, ProjectItem, QueryExpr, ScalarExpr, SetOpKind, SortKey, SourceSpec, }; @@ -644,7 +644,7 @@ fn expr_to_col_name(expr: &Expr) -> Option { #[cfg(test)] mod tests { use super::parse_sql_expr; - use crate::algebra::expr::QueryExpr; + use crate::intent_algebra::legacy_expr::QueryExpr; use crate::types::AggType; fn parse(sql: &str) -> QueryExpr { diff --git a/controller/src/replan.rs b/controller/src/replan.rs index 1ebb55d0..10ebe4ee 100644 --- a/controller/src/replan.rs +++ b/controller/src/replan.rs @@ -21,14 +21,16 @@ use tokio::sync::RwLock; use tracing::{info, warn}; use crate::backend_client::{push_or_log, BackendClient}; -use crate::config::{ +use crate::emit::{ build_precompute_jobs, collect_metric_to_family, emit_for_runtime, extend_edge_with_demo_plumbing, generate_agent_config, generate_backend_config, generate_streaming_config_yaml, AgentRuntime, WorkloadRegistry, }; use crate::monitor::Scraper; use crate::opamp::{AgentRole, OpampServer, RemoteConfig}; -use crate::planner::{self, BaselinePlanner}; +use crate::optimizer::baseline::BaselinePlanner; +use crate::optimizer::{cost as cost_model, rules}; +use crate::physical::stage_split; use crate::store::{PlanStore, WorkloadStore}; use crate::types::QueryWorkload; @@ -169,10 +171,10 @@ impl Replanner { /// `QueryWorkload` directly. Used by `replan_metric` which already /// has the workload in scope. fn try_emit_typed_edge_yaml_for_workload(&self, workload: &QueryWorkload) -> Option { - let sketch_expr = planner::rules::bind_workload_typed(workload)?; - let configs = planner::stage_split::split_typed_three_stage(&sketch_expr)?; + let sketch_expr = rules::bind_workload_typed(workload)?; + let configs = stage_split::split_typed_three_stage(&sketch_expr)?; let mut edge_cfg = configs.into_iter().find_map(|(_, cfg)| match cfg { - crate::stage_split::StageConfig::Edge(edge) => Some(edge), + crate::physical::colored_dag::StageConfig::Edge(edge) => Some(edge), _ => None, })?; @@ -237,7 +239,7 @@ impl Replanner { let Ok(plan) = self.plan_store.get(&metric) else { return false }; - let yaml = if planner::stage_split::typed_stage_split_enabled() { + let yaml = if stage_split::typed_stage_split_enabled() { match self.try_emit_typed_edge_yaml(&metric) { Some(y) => { info!( @@ -303,7 +305,7 @@ impl Replanner { // rather than broadcasting to all agent-role collectors. Same gate // as `push_config_to_agent` — typed path on, legacy fallback on // emit failure or when the gate is off. - let agent_yaml: Option = if planner::stage_split::typed_stage_split_enabled() { + let agent_yaml: Option = if stage_split::typed_stage_split_enabled() { match self.try_emit_typed_edge_yaml_for_workload(&workload) { Some(y) => { info!( @@ -421,7 +423,8 @@ mod tests { use chrono::Utc; - use crate::planner::{CostModelPlanner, BaselinePlanner}; + use crate::optimizer::baseline::BaselinePlanner; + use crate::optimizer::cost::CostModelPlanner; use crate::store::{PlanStore, WorkloadStore}; use crate::types::*; diff --git a/controller/src/sketch_algebra/capability.rs b/controller/src/sketch_algebra/capability.rs index 00d9b25d..aef42d0f 100644 --- a/controller/src/sketch_algebra/capability.rs +++ b/controller/src/sketch_algebra/capability.rs @@ -430,6 +430,33 @@ pub fn load_capability_overrides(path: &str) -> HashMap f64 { + 1.04 / (2.0f64.powi(registers as i32)).sqrt() +} + +/// Count-Min Sketch accuracy from width: `e / width`. +/// +/// Source: Cormode & Muthukrishnan, "An improved data stream summary: +/// the count-min sketch and its applications" (2005). +pub fn countmin_accuracy(width: u32) -> f64 { + std::f64::consts::E / width as f64 +} + // ── Tests ──────────────────────────────────────────────────────────────────── #[cfg(test)] diff --git a/controller/src/config/workloads.rs b/controller/src/workload.rs similarity index 100% rename from controller/src/config/workloads.rs rename to controller/src/workload.rs