From 5e302edfbbcfe07e6f6017ab271b51020ec3b028 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:57:47 +0000 Subject: [PATCH 1/4] docs: map planner-layering stages and passes to code components and interfaces --- docs/develop_docs/README.md | 1 + .../develop_docs/planner-layering-code-map.md | 614 ++++++++++++++++++ 2 files changed, 615 insertions(+) create mode 100644 docs/develop_docs/planner-layering-code-map.md diff --git a/docs/develop_docs/README.md b/docs/develop_docs/README.md index 817722adf..f1aa0a2af 100644 --- a/docs/develop_docs/README.md +++ b/docs/develop_docs/README.md @@ -3,6 +3,7 @@ Developer documentation describes current code contracts, extension points, formats, evidence, and verification workflows. +- [Planner layering code map: stages and passes to code](planner-layering-code-map.md) - [Public library functions and examples](library-api.md) - [ASAP-aware mapping architecture](asap-aware-mapping-architecture.md) - [ASAP-aware mapping contracts](asap-aware-mapping-contracts.md) diff --git a/docs/develop_docs/planner-layering-code-map.md b/docs/develop_docs/planner-layering-code-map.md new file mode 100644 index 000000000..c2054c55a --- /dev/null +++ b/docs/develop_docs/planner-layering-code-map.md @@ -0,0 +1,614 @@ +# Planner layering: code map + +This page maps every stage and pass of +[ASAPPlanner Planning Stages Design](../design_docs/proposals/planner-layering.md) +(#509) onto the crates, modules and public interfaces that implement it. It +describes `main` @ `6665fcc3`. Paths are relative to the repository root. +Signatures are copied from the code, with bodies and doc comments trimmed. + +It is a reference for finding code, not a design document: for the meaning of +a stage, pass or term, read planner-layering.md. + +## Pipeline at a glance + +```mermaid +flowchart TB + W["PlanningWorkload
(QueryWorkload + DataWorkload)"] + S0["Stage 0, query-language frontends
frontend-promql / frontend-sql / frontend-metricsql"] + P1["Stage 1, logical ASAP-aware optimization
Pass 1: generate_per_computation_candidates"] + C1["Stage 1, composition and sizing
compose_logical_candidate (Cartesian product)"] + S2["Stage 2, physical ASAP-aware optimization
optimize_physical_asap"] + S3["Stage 3, plan selection
select_plan"] + S4["Stage 4, execution (deployment)
asap-physical-operators"] + W --> S0 + S0 -- "Vec#lt;QueryRoot#gt; (LogicalDAG per query)" --> P1 + P1 -- "PerComputationCandidates#lt;Id#gt;" --> C1 + C1 -- "Vec#lt;LogicalCandidate#gt; (CandidateLogicalASAPDAGs)" --> S2 + S2 -- "Vec#lt;PhysicalCandidate#gt; (CandidatePhysicalASAPDAGs)
+ Vec#lt;PrunedCandidate#gt;" --> S3 + S3 -- "Selection (one PhysicalASAPDAG)" --> S4 +``` + +Pass 2 (ASAP-aware CSE) runs after Pass 1 and before sizing in the design; it +is not implemented on `main` (see [Pass 2](#pass-2-asap-aware-common-subexpression-elimination)). + +### Entry points + +| Component | Role | +|---|---| +| `crates/planner/src/lib.rs` (`asap-planner`) | Library facade: `e2e_plan` runs Stage 0, builds a `ParsedWorkload`, then runs the optimization pass. | +| `crates/asap-aware-mapping/src/pass/mod.rs` | `OptimizationPass` trait, `optimize` (input validation + output contract check), `PlanningModels`, `PlanOutput`, `PassRegistry`. | +| `crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs` | The planner pipeline: `run_stage1_to_stage3`, `Stage1ToStage3Run`, `Stage1ToStage3Pass`. | + +```rust +// crates/planner/src/lib.rs +pub async fn e2e_plan(input: UserInput<'_>) -> Result { + +// crates/asap-aware-mapping/src/pass/mod.rs +pub fn optimize( + pass: &dyn OptimizationPass, + input: OptimizationInput<'_>, +) -> Result { + +pub struct PlanningModels<'a> { + pub cost: &'a dyn CostModel, + pub accuracy: &'a dyn AccuracyModel, + pub evidence: &'a dyn AccuracyEvidenceProvider, +} +``` + +`run_stage1_to_stage3` takes the frontends' roots and runs Stage 1 → Stage 2 +→ Stage 3. Only a Pass 1 failure is an error; non-composing or +unimplementable candidates and a failed selection are recorded in the run. + +```rust +// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +pub fn run_stage1_to_stage3( + roots: Vec<(Id, QueryRoot)>, + targets: &[Option], + data: &DataWorkload, + models: PlanningModels<'_>, +) -> Result, PerComputationCandidateError> { + +pub struct Stage1ToStage3Run { + pub stage1_pass1_per_computation: PerComputationCandidates, + pub combinations: usize, + pub stage1_logical_asap_optimization: Vec, + pub stage2_physical_asap_optimization: Vec, + pub stage2_pruned: Vec, + pub stage3_plan_selection: Result, +} + +pub struct Stage1ToStage3Pass; +``` + +`Stage1ToStage3Run::selected` returns the selected `PhysicalCandidate`. +`Stage1ToStage3Pass` is the `OptimizationPass` named `"stage1-to-stage3"`; its +`optimize` adds the PromQL series identity to each root, +reads one accuracy target per workload entry, runs `run_stage1_to_stage3` and +returns the selected physical candidate's roots as a `PlanOutput`. + +## Stage 0, query-language frontends + +Design: [0. Language-specific frontends](../design_docs/proposals/planner-layering.md#0-language-specific-frontends). + +| Component | Role | +|---|---| +| `crates/frontend-promql` (`asap-frontend-promql`) | PromQL parse → `UnresolvedOp` tree → `resolve_root`. | +| `crates/frontend-sql` (`asap-frontend-sql`) | SQL parse and plan through DataFusion → `UnresolvedOp` tree (`SqlLowerer`, `SqlCatalog`). | +| `crates/frontend-metricsql` (`asap-frontend-metricsql`) | MetricsQL AST → `UnresolvedOp` tree. | +| `crates/frontend-common` (`asap-frontend-common`) | Shared name-based tree (`UnresolvedOp`, `UnresolvedScalar`), `SchemaResolver`, and `resolve_root`, which binds names to positional `ColumnId`s and returns the IR. | +| `crates/types/src/ir/` (`asap-types`) | The unified operator IR (#511) every stage reads and writes. | +| `crates/types/src/workload.rs` | `PlanningWorkload`, `QueryWorkload`, `DataWorkload`, accuracy and latency requirements. | +| `crates/types/src/parsed_workload.rs` | `ParsedWorkload`: the workload with one lowered root per entry. | + +Frontend interfaces (workload in, one root per query out): + +```rust +// crates/frontend-promql/src/lib.rs +pub fn lower_promql_query_workload( + workload: &PlanningWorkload, + now_ms: u64, +) -> Result, PromqlError> { + +// crates/frontend-metricsql/src/lib.rs +pub fn lower_metricsql_query_workload( + workload: &PlanningWorkload, + now_ms: u64, +) -> Result, MetricsqlError> { + +// crates/frontend-sql/src/lib.rs +pub async fn lower_sql_dialect( + query: &str, + catalog: &SqlCatalog, + dialect: SqlDialect, + accuracy: AccuracyTarget, +) -> Result, SqlError> { +``` + +Output IR (`crates/types/src/ir/`): a `LogicalDAG` is a `QueryRoot` whose +operators are all `Operator::NonASAP`. + +```rust +// crates/types/src/ir/query.rs +pub enum QueryRoot> { + Operator(C), + Scalar(ScalarExpr), +} + +// crates/types/src/ir/node.rs +pub struct OperatorNode { + pub operator: Operator, + pub result_kind: OperatorResultKind, + pub schema: Schema, + pub guarantee: Option, + pub timing: Option, + pub materialization: Materialization, + // ... +} +``` + +Operators are `NonASAPOp` (`ir/non_asap.rs`: `Scan`, `Filter`, `Project`, +`Aggregate`, `Join`, …) or, from Stage 1 on, `ASAPOp` (`ir/asap.rs`: +`SummaryAgg`, `SummaryEstimate`, `FinalizeExactAccumulator`, +`MaintainPopulation`, `SummaryMerge`, …); scalars are in `ir/scalar.rs`. + +`DataWorkload` (arrival, ingestion interval, volume and rate, input +cardinality, distribution, each as `Evidence`) is read by Stage 3 pricing. + +## Stage 1, logical ASAP-aware optimization + +Design: [1. Logical ASAP-aware optimization](../design_docs/proposals/planner-layering.md#1-logical-asap-aware-optimization). + +| Component | Role | +|---|---| +| `crates/asap-aware-mapping/src/per_computation_candidates.rs` | Pass 1 (`generate_per_computation_candidates`) and the composition of whole-workload candidates with sizing and guarantees (`compose_logical_candidate`). | +| `crates/asap-aware-mapping/src/realization.rs` | Shared helpers: summary families per intent, accuracy target and `(ε, δ)` budget, default sizing. | +| `crates/asap-aware-mapping/src/accuracy/` | Accuracy model used for node guarantees (`DefaultAccuracyModel`), estimators and sizing rules. | +| `crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs` | Enumerates the Cartesian product of per-computation candidates into `LogicalCandidate`s. | + +### Pass 1: Per-computation candidate generation + +Design: [Pass 1](../design_docs/proposals/planner-layering.md#pass-1-per-computation-candidate-generation) +(Algorithm 1). Input: the frontends' logical DAGs. Output: for every eligible +computation (one measure of an `Aggregate`), its unranked, unsized candidates +and the candidates pruned with a reason. No cost or accuracy model is +consulted. + +```rust +// crates/asap-aware-mapping/src/per_computation_candidates.rs +pub fn generate_per_computation_candidates( + roots: Vec<(Id, QueryRoot)>, +) -> Result, PerComputationCandidateError> { + +pub struct PerComputationCandidates { + pub roots: Vec<(Id, QueryRoot)>, + pub targets: Vec, +} + +pub struct ComputationTarget { + pub target: Rc, + pub measure: usize, + pub choices: ComputationChoices, + pub filter: Option, + pub having: Vec, +} + +pub struct ComputationChoices { + pub candidates: Vec, + pub pruned: Vec<(ComputationCandidate, PruneReason)>, +} + +pub enum ComputationCandidate { + Exact, + ExactAccumulator { + kind: ExactKind, + params: ExactParams, + }, + Summary(SketchAlgorithm), + AvgAsSumCount, +} + +pub enum PruneReason { + RequiresExact, + NullableAvgInput, +} + +// Errors of Pass 1 and composition. +pub enum PerComputationCandidateError { + Structure(#[from] SchemaDerivationError), + AssignedPhysicalAnnotation, + ContainsAsap, + InvalidAccuracy { + root: Option, + measure: Option, + }, + InvalidChoice, + Unsupported(&'static str), +} +``` + +The summary families offered per intent come from +`realization::summary_candidates`. + +### Composition, sizing and guarantees + +Design: Algorithm 1's `SUBSTITUTE`, and sizing after Pass 2. On `main`, +`compose_logical_candidate` picks one candidate per target (`choice[i]` +indexes `targets[i].choices.candidates`) and builds one whole-workload +`CandidateLogicalASAPDAG`: + +- `ExactAccumulator`: `SummaryAgg` → `FinalizeExactAccumulator`; +- `Summary`: `SummaryAgg` → `SummaryEstimate`; +- `AvgAsSumCount`: finalized `sum` and `count` divided; +- `Exact`: a single-measure `Aggregate`. + +Each summary is **sized** here for its own measure's accuracy requirement +(the private `sized` helper calls `realization::default_size_params`, which +dispatches to `accuracy::estimators`). With no Pass 2 sharing, this is the +post-Pass-2 sizing step with nothing shared. + +Every node gets a theoretical **guarantee** (the private `composed_guarantee`): +exact over exact inputs; a `SummaryEstimate` over exact input gets its sized +family's bound from `DefaultAccuracyModel::local_guarantee`; anything computed +from an approximate value gets `None` (composition through operators is #669). + +```rust +// crates/asap-aware-mapping/src/per_computation_candidates.rs +pub fn compose_logical_candidate( + inventory: &PerComputationCandidates, + choice: &[usize], +) -> Result, PerComputationCandidateError> { + +// crates/asap-aware-mapping/src/realization.rs +pub fn default_size_params( + kind: SketchAlgorithm, + intent: &AggIntent, + eps: f64, + delta: f64, +) -> SketchParams { + +// crates/asap-aware-mapping/src/accuracy/mod.rs +pub trait AccuracyModel { + fn local_guarantee( + &self, + family: &FieldDataType, + query: &SketchStatistic, + ) -> Option; + fn propagate( + &self, + op: &CompositionOperator, + inputs: &[ResultGuarantee], + local: Option<&ResultGuarantee>, + stats: &PropagationStats, + ) -> Result; + fn satisfies(&self, guarantee: &ResultGuarantee, target: &AccuracyTarget) -> bool; +} + +pub struct DefaultAccuracyModel; +``` + +`accuracy/composition.rs` holds `DefaultAccuracyModel`'s registered +propagation rules (additive, multiplicative, …) behind `propagate`. + +The whole-workload candidate set is the Cartesian product of the targets' +candidates, enumerated in mixed-radix order (last target fastest): + +```rust +// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +pub struct LogicalCandidate { + pub id: String, + pub display_name: String, + pub choice: Vec, + pub roots: Result, PerComputationCandidateError>, +} +``` + +### Pass 2: ASAP-aware common-subexpression elimination + +Design: [Pass 2](../design_docs/proposals/planner-layering.md#pass-2-asap-aware-common-subexpression-elimination) +(Algorithm 2; identical-expression, summary-capability and window-composition +rules). + +**Not implemented on `main`.** `run_stage1_to_stage3` goes straight from +Pass 1 to composition; no rule of Algorithm 2 runs in the stage pipeline. +Building blocks that exist: + +| Component | Role | +|---|---| +| `crates/types/src/ir/cse.rs` | Structural CSE (hash-consing) over the IR: `share_common_sub_dags`. Called by the legacy search, not by the stage pipeline. | +| `crates/types/src/ir/summary_coverage.rs` | `SummaryCoverage::derive`: what a summary state covers (definition and selection), the comparison the sharing rules need. | + +```rust +// crates/types/src/ir/cse.rs +pub fn share_common_sub_dags( + roots: Vec<(Id, Rc)>, +) -> Vec<(Id, Rc)> { + +// crates/types/src/ir/summary_coverage.rs +pub fn derive(node: &OperatorNode) -> Result { +``` + +## Stage 2, physical ASAP-aware optimization + +Design: [2. Physical ASAP-aware optimization](../design_docs/proposals/planner-layering.md#2-physical-asap-aware-optimization) +(Algorithm 3). + +| Component | Role | +|---|---| +| `crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs` | `optimize_physical_asap`: physical operator implementation and materialization for one logical candidate. | +| `crates/types/src/post_asap/execution_data_state.rs` | `Materialization`, `Medium`, `ExecutionTiming`. | +| `crates/types/src/ir/timing.rs` | `MaterializationAssignment`, `apply_materialization_timings`. | +| `crates/types/src/ir/physical_export.rs` | `compile_physical_asap_workload_with_node_ids` → `PhysicalASAPDAG`. | + +Input: one `CandidateLogicalASAPDAG` (`&[QueryRoot]`). Output: one +`PhysicalCandidate` (`P` from `L`) or an error, which the caller records +as a `PrunedCandidate`. + +```rust +// crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs +pub fn optimize_physical_asap( + from_logical: &str, + roots: &[QueryRoot], +) -> Result { + +pub struct PhysicalCandidate { + pub id: String, + pub from_logical: String, + pub display_name: String, + pub roots: Vec, + pub dag: PhysicalASAPDAG, +} + +pub enum PhysicalASAPOptimizationError { + Structure(#[from] SchemaDerivationError), + Timing(#[from] ExecutionDataStateError), + ExactTopK(String), +} + +// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +pub struct PrunedCandidate { + pub from_logical: String, + pub display_name: String, + pub reason: String, +} +``` + +### Materialization + +Design: [Materialization](../design_docs/proposals/planner-layering.md#materialization). +On `main`, Stage 2 offers one option: every sub-DAG is `NotMaterialized` +(`MaterializationAssignment::with_default(Materialization::NotMaterialized)`), +so each logical candidate yields exactly one physical candidate; +`apply_materialization_timings` then sets every node's execution timing. Enumerating +`MatAtIngestion(medium)` and `MatAtQuery(medium)` per sub-DAG is #604. + +```rust +// crates/types/src/post_asap/execution_data_state.rs +pub enum Medium { + Memory, + Disk, +} + +pub enum Materialization { + NotMaterialized, + MatAtIngestion(Medium), + MatAtQuery(Medium), +} +``` + +### Physical operator implementation + +Design: [Physical operator implementation](../design_docs/proposals/planner-layering.md#physical-operator-implementation). +The private `implement` / `implement_topk` helpers rewrite exact top-k +(`Aggregate{[TopK]}`) to a per-group sort, a per-group limit and a `Project`; +every other operator has one implementation, and `SummaryAgg` → +`SummaryEstimate` already is build → estimate. A top-k without that rewrite +fails with `ExactTopK` and the candidate is pruned. + +### Pruning + +Stage 2 prunes only candidates it cannot implement; each `PrunedCandidate` +carries the error as its `reason` (`Stage1ToStage3Run::stage2_pruned`). A +Stage 1 candidate whose composition failed keeps the error in +`LogicalCandidate::roots` and never reaches Stage 2. + +## Stage 3, plan selection + +Design: [3. Plan selection](../design_docs/proposals/planner-layering.md#3-plan-selection). + +| Component | Role | +|---|---| +| `crates/asap-aware-mapping/src/stage3_plan_selection.rs` | `select_plan`: accuracy check, pricing, ranking. | +| `crates/asap-aware-mapping/src/accuracy/evidence.rs` | `AccuracyEvidenceProvider` (deployment evidence, incl. `output_guarantee`), `NoAccuracyEvidence`. | +| `crates/asap-aware-mapping/src/analytical_cost.rs` | `estimate_operator`, `PhysicalOperator`, `ResourceEstimate`, `ResourceCalibration`. | +| `crates/asap-aware-mapping/src/physical_operator_statistics.rs` | `EdgeStatistics`, `OperatorStatistics` fed to `estimate_operator`. | +| `crates/asap-aware-mapping/src/cost_model.rs` | `CostModel` trait carried in `PlanningModels::cost`. | +| `crates/asap-aware-mapping/src/pass/mod.rs` | `PlanningModels`: the cost model, accuracy model and evidence handed to the stages. | + +Input: the Stage 2 candidates, one accuracy target per query, the +`DataWorkload`, and the models. Output: a `Selection` or a `SelectionError`. + +```rust +// crates/asap-aware-mapping/src/stage3_plan_selection.rs +pub fn select_plan( + cands: &[PhysicalCandidate], + targets: &[Option], + data: &DataWorkload, + models: PlanningModels<'_>, +) -> Result { + +pub struct Selection { + pub selected: String, + pub costs: BTreeMap, + pub rejected: Vec, +} + +pub struct Rejection { + pub id: String, + pub valid: bool, + pub reason: String, +} + +pub enum SelectionError { + TargetCount { + candidate: String, + roots: usize, + targets: usize, + }, + Cost { + candidate: String, + node: PhysicalASAPNodeId, + error: AnalyticalCostError, + }, + NoValidCandidate(Vec), +} +``` + +### Accuracy check + +Each query's output meets its target when its root guarantee (composed in +Stage 1) satisfies it under `models.accuracy.satisfies`, or when the +deployment's evidence for that output does (`models.evidence.output_guarantee`, +read only in Stage 3). A miss rejects the candidate with `valid: false`. +Count-Min is admitted only over weights proven non-negative. Latency bounds +and deployment capabilities are not checked yet. + +```rust +// crates/asap-aware-mapping/src/accuracy/evidence.rs +pub trait AccuracyEvidenceProvider { + // ... estimator_contract, topk_max_distinct_items, quantile_input_domain, + // propagation_stats + fn output_guarantee(&self, _output: &QueryRoot) -> Option { +} + +pub struct NoAccuracyEvidence; +``` + +### Cost + +Every valid candidate is priced node by node over its `PhysicalASAPDAG` (the +private `price`): edge statistics are derived from the `DataWorkload` and +default constants, each node is estimated with `estimate_operator`, and +converted to one number by `ResourceEstimate::calibrated_cost` with an +illustrative `ResourceCalibration`. A node is charged once if materialized and +once per consumer otherwise (`NodeCost::multiplier`). The unit and source are +`COST_UNIT` (`"cpu_ms_per_workload_evaluation"`) and `COST_SOURCE`. On `main`, +`select_plan` does not read `PlanningModels::cost`. + +```rust +// crates/asap-aware-mapping/src/stage3_plan_selection.rs +pub struct CandidateCost { + pub total: f64, + pub unit: &'static str, + pub source: &'static str, + pub per_node: BTreeMap, +} + +pub struct NodeCost { + pub cost: f64, + pub multiplier: u64, + pub detail: String, +} + +// crates/asap-aware-mapping/src/analytical_cost.rs +pub fn estimate_operator( + operator: PhysicalOperator, + statistics: OperatorStatistics, +) -> Result { + +pub struct ResourceCalibration { + pub cost_per_cpu_op: f64, + pub cost_per_scan_byte: f64, + pub cost_per_retained_byte: f64, + pub version: String, +} +``` + +### Ranking + +`select_plan` keeps the cheapest valid candidate (the first on ties) as +`Selection::selected`; every other valid candidate is listed in `rejected` +with `valid: true` and a `"costlier: …"` reason. Every candidate is priced; +nothing is capped (a search that avoids building every candidate is #671). + +## Stage 4, execution + +Design: [4. Execution](../design_docs/proposals/planner-layering.md#4-execution). +Execution runs in the deployment. The planner hands off the selected +candidate's `PhysicalASAPDAG` (`PhysicalCandidate::dag`; through the facade, +`PlanOutput::execution_timed_dag`). + +| Component | Role | +|---|---| +| `crates/types/src/ir/physical_export.rs` | `PhysicalASAPDAG`, its nodes and edges with data state and materialization; `PhysicalASAPDAGDocument` for serialization. | +| `crates/asap-physical-operators` | Physical operator DAG runtime shared by ingestion time and query time; `physical_planner` compiles a `PhysicalASAPDAG` to native operators. | + +```rust +// crates/types/src/ir/physical_export.rs +pub struct PhysicalASAPDAG { + pub nodes: Vec, + pub edges: Vec, + pub roots: Vec, + // ... scalar roots +} + +// crates/asap-physical-operators/src/physical_planner/mod.rs +pub fn compile( + dag: &PhysicalASAPDAG, + inputs: BTreeMap, + roots: &[NodeId], +) -> Result { +``` + +## Tooling + +| Component | Role | +|---|---| +| `crates/devtools/src/planning_stages.rs` | `run_planning_stages`: Stage 0 roots plus `run_stage1_to_stage3`, as one `PlanningStagesRun`. Shared by `stage_pipeline` and the planner-layering acceptance tests. | +| `crates/devtools/src/bin/stage_pipeline.rs` | Writes an `asap-stage-pipeline/v1` JSON document with every stage's candidates (`--example planner-layering-1`, `--promql …`). | +| `crates/devtools/src/planner_layering_examples.rs` | The design doc's end-to-end examples as workloads. | +| `crates/devtools/tests/stage_pipeline.rs` | Acceptance tests over the examples. | +| `tools/dag-viewer/` | Browser viewer (`server.py`, `viewer.js`, `planner-ui.js`); on `main` its server plans through the `dag_export` binary. See its `README.md`. | + +```rust +// crates/devtools/src/planning_stages.rs +pub fn run_planning_stages( + workload: &PlanningWorkload, + roots: Vec, +) -> Result { + +pub struct PlanningStagesRun { + pub stage0_query_language_frontends: Vec, + pub stage1_logical_asap_optimization: Vec, + pub stage2_physical_asap_optimization: Vec, + pub stage2_pruned: Vec, + pub stage3_plan_selection: Result, +} +``` + +## Supporting modules + +All in `crates/asap-aware-mapping/src/` unless noted. + +| Component | Role | +|---|---| +| `accuracy/estimators/` (`cms.rs`, `count_sketch.rs`, `kll.rs`, `ddsketch.rs`, `hll.rs`, `univmon.rs`, `cardinality.rs`) | Per-family sizing rules and local error bounds (`size_params`). | +| `accuracy/allocation.rs` | `AccuracyBudgetAllocator`, `EqualSplitAllocator`: split an accuracy budget across composed parts. | +| `accuracy/evidence.rs` | Evidence types (`EstimatorContract`, `PropagationStats`, `WorkloadAccuracyEvidence`). | +| `physical_operator_statistics.rs` | Edge and operator statistics for `analytical_cost`. | +| `crates/types/src/post_asap/` | Summary families (`sketch.rs`), `ResultGuarantee` (`guarantee.rs`), data state and materialization (`execution_data_state.rs`). | + +Legacy, to be deleted once the stages implement what they do: `replacement.rs` +(the old candidate search, `search_workload`), `accuracy/reconciliation.rs`, +and the modules that serve only that search; also the `dag_export` devtool. +The stage pipeline does not call them. + +## Not yet on main + +- #682: Stage 3 prices state by its size. +- #686: the viewer plans through `stage_pipeline`. +- #687: guarantee composition through arithmetic over approximate values (#669, part a). +- #604: Stage 2 materialization options. +- G4, Pass 2 identical-expression rule (earlier #588). +- #572: crate and module layout by stages and the unified IR. +- #677: doc updates after the consolidation. From 3fa7931e11baa297752dff8268a7199bbacdf2bd Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 14 Oct 2026 15:58:26 +0000 Subject: [PATCH 2/4] docs: name the in-progress Pass 2 work by its rule --- docs/develop_docs/planner-layering-code-map.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/docs/develop_docs/planner-layering-code-map.md b/docs/develop_docs/planner-layering-code-map.md index c2054c55a..5b844be90 100644 --- a/docs/develop_docs/planner-layering-code-map.md +++ b/docs/develop_docs/planner-layering-code-map.md @@ -609,6 +609,6 @@ The stage pipeline does not call them. - #686: the viewer plans through `stage_pipeline`. - #687: guarantee composition through arithmetic over approximate values (#669, part a). - #604: Stage 2 materialization options. -- G4, Pass 2 identical-expression rule (earlier #588). +- Pass 2 identical-expression rule (in progress; supersedes #588). - #572: crate and module layout by stages and the unified IR. - #677: doc updates after the consolidation. From bdf6a92fdeb48f707639be6dfb1230769bbbd71f Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:09:49 +0000 Subject: [PATCH 3/4] docs: restructure the planner-layering code map as a design doc Move it to docs/design_docs/planner-layering-components.md and describe each stage by its core components, its algorithm as implemented, and its code interface; note where the implementation differs from the planning-stages design. Link it from the design docs README instead of the developer docs. --- docs/design_docs/README.md | 3 + .../planner-layering-components.md | 589 +++++++++++++++++ docs/develop_docs/README.md | 1 - .../develop_docs/planner-layering-code-map.md | 614 ------------------ 4 files changed, 592 insertions(+), 615 deletions(-) create mode 100644 docs/design_docs/planner-layering-components.md delete mode 100644 docs/develop_docs/planner-layering-code-map.md diff --git a/docs/design_docs/README.md b/docs/design_docs/README.md index 23a789d20..6d6c0aef1 100644 --- a/docs/design_docs/README.md +++ b/docs/design_docs/README.md @@ -10,6 +10,9 @@ decisions, and evolving proposals. - [Planner input, output, and workflow](architecture/input-output-workflow.md) defines the user-facing contract, required and optional inputs, output data structures, and initial-planning and replanning workflows. +- [Planner layering: stage components, algorithms and code interfaces](planner-layering-components.md) + describes how each stage and pass of the planning-stages design is built on + `main`: its components, its algorithm, and its code interface. - [Decisions](decisions/README.md) records accepted local design choices. - [Proposals](proposals/README.md) contains evolving designs and research directions; check each document’s status before treating it as current. diff --git a/docs/design_docs/planner-layering-components.md b/docs/design_docs/planner-layering-components.md new file mode 100644 index 000000000..fed65c216 --- /dev/null +++ b/docs/design_docs/planner-layering-components.md @@ -0,0 +1,589 @@ +# Planner Layering: Stage Components, Algorithms and Code Interfaces + +Status: describes `main` @ `6665fcc3`. Audience: developers of ASAPPlanner who +read [ASAPPlanner Planning Stages Design](proposals/planner-layering.md) +and need to know how each stage is built today. + +planner-layering.md defines the stages and passes and their algorithms. This +document follows the same order and, for every stage, describes its core +components, the algorithm as implemented on `main`, and its code interface. +Where `main` differs from the design or does not implement it yet, the stage +says so. + +## 1. Pipeline + +```mermaid +flowchart TB + W["PlanningWorkload
(QueryWorkload + DataWorkload)"] + S0["Stage 0, query-language frontends"] + P1["Stage 1, logical ASAP-aware optimization
Pass 1: per-computation candidate generation"] + C1["Stage 1: composition and sizing
(Cartesian product of per-computation candidates)"] + S2["Stage 2, physical ASAP-aware optimization"] + S3["Stage 3, plan selection"] + S4["Stage 4, execution (deployment)"] + W --> S0 + S0 -- "Vec#lt;QueryRoot#gt;: one LogicalDAG per query" --> P1 + P1 -- "PerComputationCandidates" --> C1 + C1 -- "Vec#lt;LogicalCandidate#gt;: CandidateLogicalASAPDAGs" --> S2 + S2 -- "Vec#lt;PhysicalCandidate#gt;: CandidatePhysicalASAPDAGs
+ pruned candidates" --> S3 + S3 -- "Selection: one PhysicalASAPDAG" --> S4 +``` + +Pass 2 (ASAP-aware CSE) runs after Pass 1 and before sizing in the design; it +is not wired into the pipeline on `main` (Section 3.3). + +Every stage of the pipeline is one plain function over one workload. The +`asap-planner` facade runs Stage 0 and then an `OptimizationPass`; the default +pass, `Stage1ToStage3Pass`, calls `run_stage1_to_stage3`, which runs Stage 1, +Stage 2 and Stage 3 in order and keeps every stage's candidates in a +`Stage1ToStage3Run`: + +```text +run_stage1_to_stage3(roots, targets, data, models): + inventory ← generate_per_computation_candidates(roots) // Stage 1, Pass 1 + combinations ← Π_t |inventory.targets[t].choices.candidates| // saturating + for each choice in the Cartesian product (mixed radix, last target fastest): + L_n ← compose_logical_candidate(inventory, choice) // Stage 1, composition + sizing + if L_n composed: + P_n ← optimize_physical_asap(L_n) or prune L_n with the error // Stage 2 + selection ← select_plan(all P_n, targets, data, models) // Stage 3 +``` + +Only a Pass 1 error fails the run. A candidate that does not compose keeps its +error in the run, a candidate Stage 2 cannot implement is pruned with a reason, +and a failed selection is reported as the Stage 3 result. Every combination is +enumerated and priced; nothing is capped; a search that avoids building every +candidate is future work. + +### Code interface + +```rust +// crates/planner/src/lib.rs +pub async fn e2e_plan(input: UserInput<'_>) -> Result { + +// crates/asap-aware-mapping/src/pass/mod.rs +pub struct PlanningModels<'a> { + pub cost: &'a dyn CostModel, + pub accuracy: &'a dyn AccuracyModel, + pub evidence: &'a dyn AccuracyEvidenceProvider, +} + +// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +pub fn run_stage1_to_stage3( + roots: Vec<(Id, QueryRoot)>, + targets: &[Option], + data: &DataWorkload, + models: PlanningModels<'_>, +) -> Result, PerComputationCandidateError> { + +pub struct Stage1ToStage3Run { + pub stage1_pass1_per_computation: PerComputationCandidates, + pub combinations: usize, + pub stage1_logical_asap_optimization: Vec, + pub stage2_physical_asap_optimization: Vec, + pub stage2_pruned: Vec, + pub stage3_plan_selection: Result, +} +``` + +## 2. Stage 0, query-language frontends + +Design: [0. Language-specific frontends](proposals/planner-layering.md#0-language-specific-frontends). + +### Core components + +| Component | What it does | +|---|---| +| PromQL frontend (`crates/frontend-promql`) | Parses PromQL and builds the shared name-based operator tree in canonical shape. | +| SQL frontend (`crates/frontend-sql`) | Parses and plans SQL through DataFusion, then builds the same tree. | +| MetricsQL frontend (`crates/frontend-metricsql`) | Converts the MetricsQL AST into the same tree. | +| Shared resolver (`crates/frontend-common`) | `UnresolvedOp` tree with name-based column references; `resolve_root` binds each name to a positional `ColumnId` and derives every schema. | +| Unified IR (`crates/types/src/ir/`) | One operator DAG type for every stage: `OperatorNode` with `NonASAPOp` or `ASAPOp` operators. | +| Workload (`crates/types/src/workload.rs`) | `QueryWorkload` (queries, recurrence, accuracy and latency requirements) and `DataWorkload` (arrival, ingestion interval, volume, rate, cardinality, distribution, each as `Evidence`). | + +### Core algorithm + +```text +for each entry q of the QueryWorkload, in entries() order: + tree ← LANGUAGE_FRONTEND(q.query) // name-based UnresolvedOp tree, canonical shape + root ← resolve_root(tree) // bottom-up: bind names, derive schemas + emit QueryRoot::Operator(root) or QueryRoot::Scalar(expr) +reject the workload if any query cannot be represented +``` + +The output is one `LogicalDAG` per query: a `QueryRoot` whose operators are all +`NonASAPOp` (`Scan`, `Filter`, `Project`, `Aggregate`, `Join`, `TimeRange`, …), +with no ASAP summary choice, timing or materialization. An aggregation records +its computation as an `AggIntent` per measure, carrying the query's accuracy +target. The facade fails the whole call if one query fails, because the output +is aligned with the workload's entries. The design's `CandidateLogicalDAGs` is +one candidate on `main`: the frontends produce a single logical form per query. + +### Code interface + +```rust +// crates/frontend-promql/src/lib.rs +pub fn lower_promql_query_workload( + workload: &PlanningWorkload, + now_ms: u64, +) -> Result, PromqlError> { + +// crates/frontend-sql/src/lib.rs +pub async fn lower_sql_dialect( + query: &str, + catalog: &SqlCatalog, + dialect: SqlDialect, + accuracy: AccuracyTarget, +) -> Result, SqlError> { + +// crates/frontend-metricsql/src/lib.rs +pub fn lower_metricsql_query_workload( + workload: &PlanningWorkload, + now_ms: u64, +) -> Result, MetricsqlError> { + +// crates/types/src/ir/query.rs +pub enum QueryRoot> { + Operator(C), + Scalar(ScalarExpr), +} +``` + +## 3. Stage 1, logical ASAP-aware optimization + +Design: [1. Logical ASAP-aware optimization](proposals/planner-layering.md#1-logical-asap-aware-optimization). + +### Core components + +| Component | What it does | +|---|---| +| Pass 1 (`crates/asap-aware-mapping/src/per_computation_candidates.rs`, `generate_per_computation_candidates`) | Finds every computation target and its unranked, unsized candidates, with pruned candidates and reasons. | +| Composition (same file, `compose_logical_candidate`) | Builds one whole-workload candidate DAG from one choice per target, sizes its summaries and gives every node a guarantee. | +| Summary realization (`crates/asap-aware-mapping/src/realization.rs`) | The summary families per intent (`summary_candidates`), the `(ε, δ)` budget of a target, and default sizing. | +| Accuracy model (`crates/asap-aware-mapping/src/accuracy/`) | `DefaultAccuracyModel`: local error bounds of sized families, propagation rules (`composition.rs`), and per-family sizing (`estimators/`). | +| Enumeration (`crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs`) | Walks the Cartesian product of per-computation candidates into `LogicalCandidate`s `L1, L2, …`. | + +### 3.1 Pass 1: Per-computation candidate generation + +Design: [Pass 1](proposals/planner-layering.md#pass-1-per-computation-candidate-generation), +Algorithms 1 and 1.1. + +#### Core algorithm + +On `main`, a computation target is one measure of an `Aggregate`, and +Algorithm 1.1's `MATCH` is reading that measure's `AggIntent`: the intent +binds the input column, grouping (`Reduction`), filter and parameters such as +`k` or `q`. The rule templates are fixed per candidate kind (Section 3.2), and +the families that fill a summary slot come from a static table, +`summary_candidates(intent)`: + +| Intent | Summary families | +|---|---| +| `Quantile` | KLL, DDSketch | +| `Cardinality` | HLL, Theta, KMV, UnivMon (not UnivMon over a tuple) | +| `FrequencyL2`, `FrequencyEntropy` | UnivMon | +| `TopK` | CMS with heap, Count Sketch with heap | +| `Count` | CMS, Count Sketch, UnivMon | + +```text +generate_per_computation_candidates(roots): + reject a root that contains ASAP operators, timing or a materialization + for each Aggregate node a reachable from the roots (once, shared or not): + record a's HAVING entries: + a's own having, and each consumer Filter directly above a + placement ← BelowAggregate if the predicate reads only group keys + (and, for a consumer Filter, every consumer applies it) + else AboveOutput + for each measure m of a, with intent i: + choices ← computation_candidates_for_intent(i) + skip m if its only option is plain Exact + if i is avg over a possibly-null input: prune AvgAsSumCount (NullableAvgInput) + emit target (a, m, choices, filter of m, HAVING entries) + +computation_candidates_for_intent(i): + exact option ← ExactAccumulator(kind) for Count, Sum, Min, Max, Rate, IRate, Increase + AvgAsSumCount for avg + Exact otherwise + if i has an accuracy target: + if the target is Exact: prune every summary_candidates(i) (RequiresExact) + else: check ε, δ ∈ (0, 1); add Summary(f) for every f in summary_candidates(i) +``` + +Each target thus has exactly one exact option, and pruning happens only for a +provable reason. Nested aggregates are separate targets, and their options +multiply in composition. Pass 1 uses no cost or accuracy model. + +Differences from the design: candidates are not checked against a family's +capability beyond the table, and the design's Hydra template for `TopK` +("one summary over all groups") is not in the table. + +#### Code interface + +```rust +// crates/asap-aware-mapping/src/per_computation_candidates.rs +pub fn generate_per_computation_candidates( + roots: Vec<(Id, QueryRoot)>, +) -> Result, PerComputationCandidateError> { + +pub struct PerComputationCandidates { + pub roots: Vec<(Id, QueryRoot)>, + pub targets: Vec, +} + +pub struct ComputationTarget { + pub target: Rc, + pub measure: usize, + pub choices: ComputationChoices, + pub filter: Option, + pub having: Vec, +} + +pub struct ComputationChoices { + pub candidates: Vec, + pub pruned: Vec<(ComputationCandidate, PruneReason)>, +} + +pub enum ComputationCandidate { + Exact, + ExactAccumulator { + kind: ExactKind, + params: ExactParams, + }, + Summary(SketchAlgorithm), + AvgAsSumCount, +} +``` + +### 3.2 Composition, sizing and guarantees + +Design: Algorithm 1, lines 18–21 (`SUBSTITUTE`), and sizing after Pass 2. + +#### Core algorithm + +`run_stage1_to_stage3` enumerates the Cartesian product of the targets' +candidates. `choice[t]` indexes `targets[t].choices.candidates`; the choice +vector is incremented in mixed radix, last target fastest, and `L` is the +`n`-th choice. The run records the product's size in `combinations` +(saturating at `usize::MAX`). For each choice, `compose_logical_candidate` +rewrites the workload bottom-up: + +```text +compose_logical_candidate(inventory, choice): + for each aggregate a with a chosen candidate other than Exact: + replace a measure by measure, over a's input: + ExactAccumulator → SummaryAgg(exact family) → FinalizeExactAccumulator + Summary(f) → SummaryAgg(SIZED(f, acc(m))) → SummaryEstimate + AvgAsSumCount → finalized sum and count, divided + Exact → a single-measure Aggregate + join several measures back on the group keys; keep a's column names + measure filter → SummaryAgg.filter + HAVING BelowAggregate → Filter below the state (consumer Filter dropped) + HAVING AboveOutput → Filter above the output + keep sub-DAGs shared between roots shared + for each node, bottom-up: node.guarantee ← COMPOSED_GUARANTEE(node) + +SIZED(f, acc): + (ε, δ) ← accuracy_budget(acc) // δ defaults to 0.01 + params ← default_size_params(f, intent, ε, δ) // accuracy/estimators: invert f's error bound + +COMPOSED_GUARANTEE(node): + FinalizeExactAccumulator over exact input → exact + SummaryEstimate over exact input, bound applies → DefaultAccuracyModel.local_guarantee(f, query) + Project of plain columns → the child's guarantee + non-ASAP operator over exact inputs → exact + anything else (incl. any value computed from an approximate input) → None +``` + +A summary estimate's bound "applies" unless the query is Top-K (a heap sketch +bounds item frequencies, not membership) or the family is Count-Min over +weights not proven non-negative. A `None` guarantee does not reject the +candidate here; Stage 3 decides. + +Differences from the design: with no Pass 2, each summary is sized for its own +measure's requirement, which is the post-Pass-2 sizing with nothing shared. +Guarantees are not yet propagated through operators over approximate values +(`AccuracyModel::propagate` exists but composition does not call it). + +#### Code interface + +```rust +// crates/asap-aware-mapping/src/per_computation_candidates.rs +pub fn compose_logical_candidate( + inventory: &PerComputationCandidates, + choice: &[usize], +) -> Result, PerComputationCandidateError> { + +// crates/asap-aware-mapping/src/realization.rs +pub fn default_size_params( + kind: SketchAlgorithm, + intent: &AggIntent, + eps: f64, + delta: f64, +) -> SketchParams { + +// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +pub struct LogicalCandidate { + pub id: String, + pub display_name: String, + pub choice: Vec, + pub roots: Result, PerComputationCandidateError>, +} +``` + +### 3.3 Pass 2: ASAP-aware common-subexpression elimination + +Design: [Pass 2](proposals/planner-layering.md#pass-2-asap-aware-common-subexpression-elimination), +Algorithm 2 (identical-expression, summary-capability and window-composition +rules). + +#### Core algorithm + +Not implemented on `main`: `run_stage1_to_stage3` goes from Pass 1 straight to +composition, so no candidate shares a summary across computations, and a sub-DAG +is shared only when the frontends already produced it once. Two building +blocks exist outside the pipeline: + +- **Structural CSE** (`crates/types/src/ir/cse.rs`, `share_common_sub_dags`): + bottom-up hash-consing of the IR; a hash narrows the search and typed + equality decides sharing. It is the structural core of the + identical-expression rule; only the legacy search calls it today. +- **Summary coverage** (`crates/types/src/ir/summary_coverage.rs`, + `SummaryCoverage::derive`): what a summary state computes (`definition`) and + which rows it took (`selection`), the comparison of summary input data that + the sharing rules need. + +#### Code interface + +```rust +// crates/types/src/ir/cse.rs +pub fn share_common_sub_dags( + roots: Vec<(Id, Rc)>, +) -> Vec<(Id, Rc)> { +``` + +## 4. Stage 2, physical ASAP-aware optimization + +Design: [2. Physical ASAP-aware optimization](proposals/planner-layering.md#2-physical-asap-aware-optimization), +Algorithm 3. + +### Core components + +| Component | What it does | +|---|---| +| Physical ASAP-aware optimization (`crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs`, `optimize_physical_asap`) | Implements every operator and assigns materialization for one logical candidate. | +| Materialization options (`crates/types/src/post_asap/execution_data_state.rs`) | `Materialization`: `NotMaterialized`, `MatAtIngestion(Medium)`, `MatAtQuery(Medium)`; `Medium` is `Memory` or `Disk`. | +| Timing (`crates/types/src/ir/timing.rs`) | `MaterializationAssignment` and `apply_materialization_timings`, which write every node's `ExecutionTiming` and `Materialization`. | +| Physical export (`crates/types/src/ir/physical_export.rs`) | Flattens the timed roots into a `PhysicalASAPDAG` with stable node ids. | + +### Core algorithm + +```text +optimize_physical_asap(L_n.roots): + // physical operator implementation, one memo for the whole workload + for each node, bottom-up: + if node is Aggregate with a TopK measure: + if it is a plain grouped top-k: Project(Limit(Sort per group)) restoring [keys.., topk_k] + else: fail ExactTopK // per-entity reduction, other measures, filter, HAVING + else: keep the node (SummaryAgg → SummaryEstimate already is build → estimate) + // materialization + m ← MaterializationAssignment::with_default(NotMaterialized) + timed ← apply_materialization_timings(roots, m) + dag ← compile_physical_asap_workload_with_node_ids(timed) + return PhysicalCandidate { from_logical: L_n, roots: timed, dag } +``` + +The caller names the result `P` after `L`; on error, it records a +`PrunedCandidate` with the error as its reason. These are the only candidates +Stage 2 removes, and each is provably invalid (it cannot be implemented). + +Differences from the design: materialization offers one option, so each logical +candidate yields exactly one physical candidate. Algorithm 3's enumeration of +`MatAtIngestion(medium)` and `MatAtQuery(medium)` per sub-DAG, its +ingestion-time upstream constraint and retention are not implemented. Parallelism, +partitioning and resource management are TODO in the design as well. + +### Code interface + +```rust +// crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs +pub fn optimize_physical_asap( + from_logical: &str, + roots: &[QueryRoot], +) -> Result { + +pub struct PhysicalCandidate { + pub id: String, + pub from_logical: String, + pub display_name: String, + pub roots: Vec, + pub dag: PhysicalASAPDAG, +} + +// crates/types/src/post_asap/execution_data_state.rs +pub enum Materialization { + NotMaterialized, + MatAtIngestion(Medium), + MatAtQuery(Medium), +} +``` + +## 5. Stage 3, plan selection + +Design: [3. Plan selection](proposals/planner-layering.md#3-plan-selection). + +### Core components + +| Component | What it does | +|---|---| +| Plan selection (`crates/asap-aware-mapping/src/stage3_plan_selection.rs`, `select_plan`) | Rejects candidates that miss an accuracy target, prices the rest over their physical DAG and selects the cheapest. | +| Accuracy model (`AccuracyModel::satisfies`, `crates/asap-aware-mapping/src/accuracy/mod.rs`) | Decides whether a guarantee meets an `AccuracyTarget`. | +| Accuracy evidence (`AccuracyEvidenceProvider`, `crates/asap-aware-mapping/src/accuracy/evidence.rs`) | The deployment's empirical accuracy; `output_guarantee` gives evidence for one candidate's whole query output. `NoAccuracyEvidence` supplies none. | +| Analytical cost (`crates/asap-aware-mapping/src/analytical_cost.rs`) | `estimate_operator`: resource estimate (CPU ops, scan bytes, retained bytes) of one physical operator over edge statistics (`physical_operator_statistics.rs`); `ResourceCalibration` turns it into one number. | + +### Core algorithm + +```text +select_plan(candidates, targets, data, models): + for each candidate P: + // accuracy check (output-level) + for each query q with a target T: + if q reads a Count-Min SummaryAgg whose weights are not proven non-negative: + reject P (valid = false) + g ← root guarantee of q (composed in Stage 1) + if not (g satisfies T under models.accuracy + or models.evidence.output_guarantee(q) satisfies T): + reject P (valid = false) + // cost + cost(P) ← PRICE(P.dag, data) + selected ← the valid candidate with the least cost (the first on ties) + every other valid candidate → rejected (valid = true, "costlier: …") + no valid candidate → SelectionError::NoValidCandidate + +PRICE(dag, data): + series, rows/ms ← data.input_cardinality, data.ingestion_rate (or defaults) + for each node, children first: + estimate ← estimate_operator(physical operator, edge statistics) + // Scan, HashAggregate, Sort, Limit, …; SummaryAgg: rows × sketch depth; + // SummaryEstimate: groups × k + multiplier ← 1 if materialized, else the number of consumers that reach it + (queries, and materialized nodes, without looking through one) + node cost ← estimate.calibrated_cost(calibration) × multiplier + return Σ node costs +``` + +The check is on the query output, not on the summaries inside the plan: a +certified estimate is not enough on its own, since a difference or ratio of +estimates can be unbounded. Cost is computed for the whole workload, so a +node shared by several queries is charged per computation; with every node +`NotMaterialized`, a shared node is charged once per query. + +Differences from the design: pricing uses a fixed illustrative +`ResourceCalibration` (`illustrative-v1`) and default statistics +(`DEFAULT_GROUP_COUNT` groups and default series and rates when the +`DataWorkload` declares none), so prices are not calibrated; `select_plan` does +not read `PlanningModels::cost`. A binary, join or concat node is priced from +its first input only. Latency bounds and deployment capabilities are not +checked yet. + +### Code interface + +```rust +// crates/asap-aware-mapping/src/stage3_plan_selection.rs +pub fn select_plan( + cands: &[PhysicalCandidate], + targets: &[Option], + data: &DataWorkload, + models: PlanningModels<'_>, +) -> Result { + +pub struct Selection { + pub selected: String, + pub costs: BTreeMap, + pub rejected: Vec, +} + +pub struct Rejection { + pub id: String, + pub valid: bool, + pub reason: String, +} + +// crates/asap-aware-mapping/src/accuracy/evidence.rs + fn output_guarantee(&self, _output: &QueryRoot) -> Option { + +// crates/asap-aware-mapping/src/analytical_cost.rs +pub fn estimate_operator( + operator: PhysicalOperator, + statistics: OperatorStatistics, +) -> Result { +``` + +## 6. Stage 4, execution + +Design: [4. Execution](proposals/planner-layering.md#4-execution) and +[Physical Planning, Summary Maintenance, and Deployment](physical-planning-and-deployment.md). + +### Core components + +| Component | What it does | +|---|---| +| Physical ASAP DAG (`crates/types/src/ir/physical_export.rs`) | The hand-off: nodes with operator payload, output data state and materialization; edges with schema and data state; one root per query. | +| Physical operators (`crates/asap-physical-operators`) | The operator DAG runtime shared by ingestion time and query time; `physical_planner::compile` lowers a `PhysicalASAPDAG` to native operators. | + +### Core algorithm + +The planner hands off the selected candidate's `PhysicalASAPDAG` +(`PhysicalCandidate::dag`; through the facade, `PlanOutput` and its +`execution_timed_dag`). The deployment runs it as given: it does not choose +among summaries or change materialization. Ingestion-time nodes form the +precompute DAG and the rest the query DAG; `compile` builds native operators +with typed inputs, which the deployment binds to its data sources. + +### Code interface + +```rust +// crates/types/src/ir/physical_export.rs +pub struct PhysicalASAPDAG { + pub nodes: Vec, + pub edges: Vec, + pub roots: Vec, + // ... +} + +// crates/asap-physical-operators/src/physical_planner/mod.rs +pub fn compile( + dag: &PhysicalASAPDAG, + inputs: BTreeMap, + roots: &[NodeId], +) -> Result { +``` + +## 7. Tooling and legacy code + +- `run_planning_stages` (`crates/devtools/src/planning_stages.rs`) runs Stage 0 + through Stage 3 for one workload; the `stage_pipeline` binary writes every + stage's candidates as an `asap-stage-pipeline/v1` document, and the + planner-layering acceptance tests use the same function. +- `tools/dag-viewer` is the browser viewer; on `main` its server plans through + the `dag_export` binary. +- Legacy, to be deleted once the stages implement what it does: `replacement.rs` + (the old candidate search, `search_workload`), `accuracy/reconciliation.rs`, + the modules only that search uses, and `dag_export`. The stage pipeline + does not call them. + +## 8. Not yet on main + +- **Pass 2, ASAP-aware CSE.** None of the three sharing rules runs in the + pipeline; summaries are sized per computation, never for the strictest + requirement among shared consumers. +- **Guarantee composition.** Guarantees are not propagated through arithmetic + or comparisons over approximate values; such outputs carry no guarantee and + are admitted only with deployment evidence. +- **Materialization.** Stage 2 assigns `NotMaterialized` everywhere; the + per-sub-DAG choice of ingestion time or query time and medium is not + enumerated. +- **Cost.** Stage 3 prices with fixed illustrative statistics and calibration, + does not price retained state by its size, and does not use the deployment's + cost model; latency bounds and capabilities are not checked. +- **Layout and tooling.** Crates and modules are not yet organized by stage, + and the viewer does not yet plan through `stage_pipeline`. diff --git a/docs/develop_docs/README.md b/docs/develop_docs/README.md index f1aa0a2af..817722adf 100644 --- a/docs/develop_docs/README.md +++ b/docs/develop_docs/README.md @@ -3,7 +3,6 @@ Developer documentation describes current code contracts, extension points, formats, evidence, and verification workflows. -- [Planner layering code map: stages and passes to code](planner-layering-code-map.md) - [Public library functions and examples](library-api.md) - [ASAP-aware mapping architecture](asap-aware-mapping-architecture.md) - [ASAP-aware mapping contracts](asap-aware-mapping-contracts.md) diff --git a/docs/develop_docs/planner-layering-code-map.md b/docs/develop_docs/planner-layering-code-map.md deleted file mode 100644 index 5b844be90..000000000 --- a/docs/develop_docs/planner-layering-code-map.md +++ /dev/null @@ -1,614 +0,0 @@ -# Planner layering: code map - -This page maps every stage and pass of -[ASAPPlanner Planning Stages Design](../design_docs/proposals/planner-layering.md) -(#509) onto the crates, modules and public interfaces that implement it. It -describes `main` @ `6665fcc3`. Paths are relative to the repository root. -Signatures are copied from the code, with bodies and doc comments trimmed. - -It is a reference for finding code, not a design document: for the meaning of -a stage, pass or term, read planner-layering.md. - -## Pipeline at a glance - -```mermaid -flowchart TB - W["PlanningWorkload
(QueryWorkload + DataWorkload)"] - S0["Stage 0, query-language frontends
frontend-promql / frontend-sql / frontend-metricsql"] - P1["Stage 1, logical ASAP-aware optimization
Pass 1: generate_per_computation_candidates"] - C1["Stage 1, composition and sizing
compose_logical_candidate (Cartesian product)"] - S2["Stage 2, physical ASAP-aware optimization
optimize_physical_asap"] - S3["Stage 3, plan selection
select_plan"] - S4["Stage 4, execution (deployment)
asap-physical-operators"] - W --> S0 - S0 -- "Vec#lt;QueryRoot#gt; (LogicalDAG per query)" --> P1 - P1 -- "PerComputationCandidates#lt;Id#gt;" --> C1 - C1 -- "Vec#lt;LogicalCandidate#gt; (CandidateLogicalASAPDAGs)" --> S2 - S2 -- "Vec#lt;PhysicalCandidate#gt; (CandidatePhysicalASAPDAGs)
+ Vec#lt;PrunedCandidate#gt;" --> S3 - S3 -- "Selection (one PhysicalASAPDAG)" --> S4 -``` - -Pass 2 (ASAP-aware CSE) runs after Pass 1 and before sizing in the design; it -is not implemented on `main` (see [Pass 2](#pass-2-asap-aware-common-subexpression-elimination)). - -### Entry points - -| Component | Role | -|---|---| -| `crates/planner/src/lib.rs` (`asap-planner`) | Library facade: `e2e_plan` runs Stage 0, builds a `ParsedWorkload`, then runs the optimization pass. | -| `crates/asap-aware-mapping/src/pass/mod.rs` | `OptimizationPass` trait, `optimize` (input validation + output contract check), `PlanningModels`, `PlanOutput`, `PassRegistry`. | -| `crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs` | The planner pipeline: `run_stage1_to_stage3`, `Stage1ToStage3Run`, `Stage1ToStage3Pass`. | - -```rust -// crates/planner/src/lib.rs -pub async fn e2e_plan(input: UserInput<'_>) -> Result { - -// crates/asap-aware-mapping/src/pass/mod.rs -pub fn optimize( - pass: &dyn OptimizationPass, - input: OptimizationInput<'_>, -) -> Result { - -pub struct PlanningModels<'a> { - pub cost: &'a dyn CostModel, - pub accuracy: &'a dyn AccuracyModel, - pub evidence: &'a dyn AccuracyEvidenceProvider, -} -``` - -`run_stage1_to_stage3` takes the frontends' roots and runs Stage 1 → Stage 2 -→ Stage 3. Only a Pass 1 failure is an error; non-composing or -unimplementable candidates and a failed selection are recorded in the run. - -```rust -// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs -pub fn run_stage1_to_stage3( - roots: Vec<(Id, QueryRoot)>, - targets: &[Option], - data: &DataWorkload, - models: PlanningModels<'_>, -) -> Result, PerComputationCandidateError> { - -pub struct Stage1ToStage3Run { - pub stage1_pass1_per_computation: PerComputationCandidates, - pub combinations: usize, - pub stage1_logical_asap_optimization: Vec, - pub stage2_physical_asap_optimization: Vec, - pub stage2_pruned: Vec, - pub stage3_plan_selection: Result, -} - -pub struct Stage1ToStage3Pass; -``` - -`Stage1ToStage3Run::selected` returns the selected `PhysicalCandidate`. -`Stage1ToStage3Pass` is the `OptimizationPass` named `"stage1-to-stage3"`; its -`optimize` adds the PromQL series identity to each root, -reads one accuracy target per workload entry, runs `run_stage1_to_stage3` and -returns the selected physical candidate's roots as a `PlanOutput`. - -## Stage 0, query-language frontends - -Design: [0. Language-specific frontends](../design_docs/proposals/planner-layering.md#0-language-specific-frontends). - -| Component | Role | -|---|---| -| `crates/frontend-promql` (`asap-frontend-promql`) | PromQL parse → `UnresolvedOp` tree → `resolve_root`. | -| `crates/frontend-sql` (`asap-frontend-sql`) | SQL parse and plan through DataFusion → `UnresolvedOp` tree (`SqlLowerer`, `SqlCatalog`). | -| `crates/frontend-metricsql` (`asap-frontend-metricsql`) | MetricsQL AST → `UnresolvedOp` tree. | -| `crates/frontend-common` (`asap-frontend-common`) | Shared name-based tree (`UnresolvedOp`, `UnresolvedScalar`), `SchemaResolver`, and `resolve_root`, which binds names to positional `ColumnId`s and returns the IR. | -| `crates/types/src/ir/` (`asap-types`) | The unified operator IR (#511) every stage reads and writes. | -| `crates/types/src/workload.rs` | `PlanningWorkload`, `QueryWorkload`, `DataWorkload`, accuracy and latency requirements. | -| `crates/types/src/parsed_workload.rs` | `ParsedWorkload`: the workload with one lowered root per entry. | - -Frontend interfaces (workload in, one root per query out): - -```rust -// crates/frontend-promql/src/lib.rs -pub fn lower_promql_query_workload( - workload: &PlanningWorkload, - now_ms: u64, -) -> Result, PromqlError> { - -// crates/frontend-metricsql/src/lib.rs -pub fn lower_metricsql_query_workload( - workload: &PlanningWorkload, - now_ms: u64, -) -> Result, MetricsqlError> { - -// crates/frontend-sql/src/lib.rs -pub async fn lower_sql_dialect( - query: &str, - catalog: &SqlCatalog, - dialect: SqlDialect, - accuracy: AccuracyTarget, -) -> Result, SqlError> { -``` - -Output IR (`crates/types/src/ir/`): a `LogicalDAG` is a `QueryRoot` whose -operators are all `Operator::NonASAP`. - -```rust -// crates/types/src/ir/query.rs -pub enum QueryRoot> { - Operator(C), - Scalar(ScalarExpr), -} - -// crates/types/src/ir/node.rs -pub struct OperatorNode { - pub operator: Operator, - pub result_kind: OperatorResultKind, - pub schema: Schema, - pub guarantee: Option, - pub timing: Option, - pub materialization: Materialization, - // ... -} -``` - -Operators are `NonASAPOp` (`ir/non_asap.rs`: `Scan`, `Filter`, `Project`, -`Aggregate`, `Join`, …) or, from Stage 1 on, `ASAPOp` (`ir/asap.rs`: -`SummaryAgg`, `SummaryEstimate`, `FinalizeExactAccumulator`, -`MaintainPopulation`, `SummaryMerge`, …); scalars are in `ir/scalar.rs`. - -`DataWorkload` (arrival, ingestion interval, volume and rate, input -cardinality, distribution, each as `Evidence`) is read by Stage 3 pricing. - -## Stage 1, logical ASAP-aware optimization - -Design: [1. Logical ASAP-aware optimization](../design_docs/proposals/planner-layering.md#1-logical-asap-aware-optimization). - -| Component | Role | -|---|---| -| `crates/asap-aware-mapping/src/per_computation_candidates.rs` | Pass 1 (`generate_per_computation_candidates`) and the composition of whole-workload candidates with sizing and guarantees (`compose_logical_candidate`). | -| `crates/asap-aware-mapping/src/realization.rs` | Shared helpers: summary families per intent, accuracy target and `(ε, δ)` budget, default sizing. | -| `crates/asap-aware-mapping/src/accuracy/` | Accuracy model used for node guarantees (`DefaultAccuracyModel`), estimators and sizing rules. | -| `crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs` | Enumerates the Cartesian product of per-computation candidates into `LogicalCandidate`s. | - -### Pass 1: Per-computation candidate generation - -Design: [Pass 1](../design_docs/proposals/planner-layering.md#pass-1-per-computation-candidate-generation) -(Algorithm 1). Input: the frontends' logical DAGs. Output: for every eligible -computation (one measure of an `Aggregate`), its unranked, unsized candidates -and the candidates pruned with a reason. No cost or accuracy model is -consulted. - -```rust -// crates/asap-aware-mapping/src/per_computation_candidates.rs -pub fn generate_per_computation_candidates( - roots: Vec<(Id, QueryRoot)>, -) -> Result, PerComputationCandidateError> { - -pub struct PerComputationCandidates { - pub roots: Vec<(Id, QueryRoot)>, - pub targets: Vec, -} - -pub struct ComputationTarget { - pub target: Rc, - pub measure: usize, - pub choices: ComputationChoices, - pub filter: Option, - pub having: Vec, -} - -pub struct ComputationChoices { - pub candidates: Vec, - pub pruned: Vec<(ComputationCandidate, PruneReason)>, -} - -pub enum ComputationCandidate { - Exact, - ExactAccumulator { - kind: ExactKind, - params: ExactParams, - }, - Summary(SketchAlgorithm), - AvgAsSumCount, -} - -pub enum PruneReason { - RequiresExact, - NullableAvgInput, -} - -// Errors of Pass 1 and composition. -pub enum PerComputationCandidateError { - Structure(#[from] SchemaDerivationError), - AssignedPhysicalAnnotation, - ContainsAsap, - InvalidAccuracy { - root: Option, - measure: Option, - }, - InvalidChoice, - Unsupported(&'static str), -} -``` - -The summary families offered per intent come from -`realization::summary_candidates`. - -### Composition, sizing and guarantees - -Design: Algorithm 1's `SUBSTITUTE`, and sizing after Pass 2. On `main`, -`compose_logical_candidate` picks one candidate per target (`choice[i]` -indexes `targets[i].choices.candidates`) and builds one whole-workload -`CandidateLogicalASAPDAG`: - -- `ExactAccumulator`: `SummaryAgg` → `FinalizeExactAccumulator`; -- `Summary`: `SummaryAgg` → `SummaryEstimate`; -- `AvgAsSumCount`: finalized `sum` and `count` divided; -- `Exact`: a single-measure `Aggregate`. - -Each summary is **sized** here for its own measure's accuracy requirement -(the private `sized` helper calls `realization::default_size_params`, which -dispatches to `accuracy::estimators`). With no Pass 2 sharing, this is the -post-Pass-2 sizing step with nothing shared. - -Every node gets a theoretical **guarantee** (the private `composed_guarantee`): -exact over exact inputs; a `SummaryEstimate` over exact input gets its sized -family's bound from `DefaultAccuracyModel::local_guarantee`; anything computed -from an approximate value gets `None` (composition through operators is #669). - -```rust -// crates/asap-aware-mapping/src/per_computation_candidates.rs -pub fn compose_logical_candidate( - inventory: &PerComputationCandidates, - choice: &[usize], -) -> Result, PerComputationCandidateError> { - -// crates/asap-aware-mapping/src/realization.rs -pub fn default_size_params( - kind: SketchAlgorithm, - intent: &AggIntent, - eps: f64, - delta: f64, -) -> SketchParams { - -// crates/asap-aware-mapping/src/accuracy/mod.rs -pub trait AccuracyModel { - fn local_guarantee( - &self, - family: &FieldDataType, - query: &SketchStatistic, - ) -> Option; - fn propagate( - &self, - op: &CompositionOperator, - inputs: &[ResultGuarantee], - local: Option<&ResultGuarantee>, - stats: &PropagationStats, - ) -> Result; - fn satisfies(&self, guarantee: &ResultGuarantee, target: &AccuracyTarget) -> bool; -} - -pub struct DefaultAccuracyModel; -``` - -`accuracy/composition.rs` holds `DefaultAccuracyModel`'s registered -propagation rules (additive, multiplicative, …) behind `propagate`. - -The whole-workload candidate set is the Cartesian product of the targets' -candidates, enumerated in mixed-radix order (last target fastest): - -```rust -// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs -pub struct LogicalCandidate { - pub id: String, - pub display_name: String, - pub choice: Vec, - pub roots: Result, PerComputationCandidateError>, -} -``` - -### Pass 2: ASAP-aware common-subexpression elimination - -Design: [Pass 2](../design_docs/proposals/planner-layering.md#pass-2-asap-aware-common-subexpression-elimination) -(Algorithm 2; identical-expression, summary-capability and window-composition -rules). - -**Not implemented on `main`.** `run_stage1_to_stage3` goes straight from -Pass 1 to composition; no rule of Algorithm 2 runs in the stage pipeline. -Building blocks that exist: - -| Component | Role | -|---|---| -| `crates/types/src/ir/cse.rs` | Structural CSE (hash-consing) over the IR: `share_common_sub_dags`. Called by the legacy search, not by the stage pipeline. | -| `crates/types/src/ir/summary_coverage.rs` | `SummaryCoverage::derive`: what a summary state covers (definition and selection), the comparison the sharing rules need. | - -```rust -// crates/types/src/ir/cse.rs -pub fn share_common_sub_dags( - roots: Vec<(Id, Rc)>, -) -> Vec<(Id, Rc)> { - -// crates/types/src/ir/summary_coverage.rs -pub fn derive(node: &OperatorNode) -> Result { -``` - -## Stage 2, physical ASAP-aware optimization - -Design: [2. Physical ASAP-aware optimization](../design_docs/proposals/planner-layering.md#2-physical-asap-aware-optimization) -(Algorithm 3). - -| Component | Role | -|---|---| -| `crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs` | `optimize_physical_asap`: physical operator implementation and materialization for one logical candidate. | -| `crates/types/src/post_asap/execution_data_state.rs` | `Materialization`, `Medium`, `ExecutionTiming`. | -| `crates/types/src/ir/timing.rs` | `MaterializationAssignment`, `apply_materialization_timings`. | -| `crates/types/src/ir/physical_export.rs` | `compile_physical_asap_workload_with_node_ids` → `PhysicalASAPDAG`. | - -Input: one `CandidateLogicalASAPDAG` (`&[QueryRoot]`). Output: one -`PhysicalCandidate` (`P` from `L`) or an error, which the caller records -as a `PrunedCandidate`. - -```rust -// crates/asap-aware-mapping/src/stage2_physical_asap_optimization.rs -pub fn optimize_physical_asap( - from_logical: &str, - roots: &[QueryRoot], -) -> Result { - -pub struct PhysicalCandidate { - pub id: String, - pub from_logical: String, - pub display_name: String, - pub roots: Vec, - pub dag: PhysicalASAPDAG, -} - -pub enum PhysicalASAPOptimizationError { - Structure(#[from] SchemaDerivationError), - Timing(#[from] ExecutionDataStateError), - ExactTopK(String), -} - -// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs -pub struct PrunedCandidate { - pub from_logical: String, - pub display_name: String, - pub reason: String, -} -``` - -### Materialization - -Design: [Materialization](../design_docs/proposals/planner-layering.md#materialization). -On `main`, Stage 2 offers one option: every sub-DAG is `NotMaterialized` -(`MaterializationAssignment::with_default(Materialization::NotMaterialized)`), -so each logical candidate yields exactly one physical candidate; -`apply_materialization_timings` then sets every node's execution timing. Enumerating -`MatAtIngestion(medium)` and `MatAtQuery(medium)` per sub-DAG is #604. - -```rust -// crates/types/src/post_asap/execution_data_state.rs -pub enum Medium { - Memory, - Disk, -} - -pub enum Materialization { - NotMaterialized, - MatAtIngestion(Medium), - MatAtQuery(Medium), -} -``` - -### Physical operator implementation - -Design: [Physical operator implementation](../design_docs/proposals/planner-layering.md#physical-operator-implementation). -The private `implement` / `implement_topk` helpers rewrite exact top-k -(`Aggregate{[TopK]}`) to a per-group sort, a per-group limit and a `Project`; -every other operator has one implementation, and `SummaryAgg` → -`SummaryEstimate` already is build → estimate. A top-k without that rewrite -fails with `ExactTopK` and the candidate is pruned. - -### Pruning - -Stage 2 prunes only candidates it cannot implement; each `PrunedCandidate` -carries the error as its `reason` (`Stage1ToStage3Run::stage2_pruned`). A -Stage 1 candidate whose composition failed keeps the error in -`LogicalCandidate::roots` and never reaches Stage 2. - -## Stage 3, plan selection - -Design: [3. Plan selection](../design_docs/proposals/planner-layering.md#3-plan-selection). - -| Component | Role | -|---|---| -| `crates/asap-aware-mapping/src/stage3_plan_selection.rs` | `select_plan`: accuracy check, pricing, ranking. | -| `crates/asap-aware-mapping/src/accuracy/evidence.rs` | `AccuracyEvidenceProvider` (deployment evidence, incl. `output_guarantee`), `NoAccuracyEvidence`. | -| `crates/asap-aware-mapping/src/analytical_cost.rs` | `estimate_operator`, `PhysicalOperator`, `ResourceEstimate`, `ResourceCalibration`. | -| `crates/asap-aware-mapping/src/physical_operator_statistics.rs` | `EdgeStatistics`, `OperatorStatistics` fed to `estimate_operator`. | -| `crates/asap-aware-mapping/src/cost_model.rs` | `CostModel` trait carried in `PlanningModels::cost`. | -| `crates/asap-aware-mapping/src/pass/mod.rs` | `PlanningModels`: the cost model, accuracy model and evidence handed to the stages. | - -Input: the Stage 2 candidates, one accuracy target per query, the -`DataWorkload`, and the models. Output: a `Selection` or a `SelectionError`. - -```rust -// crates/asap-aware-mapping/src/stage3_plan_selection.rs -pub fn select_plan( - cands: &[PhysicalCandidate], - targets: &[Option], - data: &DataWorkload, - models: PlanningModels<'_>, -) -> Result { - -pub struct Selection { - pub selected: String, - pub costs: BTreeMap, - pub rejected: Vec, -} - -pub struct Rejection { - pub id: String, - pub valid: bool, - pub reason: String, -} - -pub enum SelectionError { - TargetCount { - candidate: String, - roots: usize, - targets: usize, - }, - Cost { - candidate: String, - node: PhysicalASAPNodeId, - error: AnalyticalCostError, - }, - NoValidCandidate(Vec), -} -``` - -### Accuracy check - -Each query's output meets its target when its root guarantee (composed in -Stage 1) satisfies it under `models.accuracy.satisfies`, or when the -deployment's evidence for that output does (`models.evidence.output_guarantee`, -read only in Stage 3). A miss rejects the candidate with `valid: false`. -Count-Min is admitted only over weights proven non-negative. Latency bounds -and deployment capabilities are not checked yet. - -```rust -// crates/asap-aware-mapping/src/accuracy/evidence.rs -pub trait AccuracyEvidenceProvider { - // ... estimator_contract, topk_max_distinct_items, quantile_input_domain, - // propagation_stats - fn output_guarantee(&self, _output: &QueryRoot) -> Option { -} - -pub struct NoAccuracyEvidence; -``` - -### Cost - -Every valid candidate is priced node by node over its `PhysicalASAPDAG` (the -private `price`): edge statistics are derived from the `DataWorkload` and -default constants, each node is estimated with `estimate_operator`, and -converted to one number by `ResourceEstimate::calibrated_cost` with an -illustrative `ResourceCalibration`. A node is charged once if materialized and -once per consumer otherwise (`NodeCost::multiplier`). The unit and source are -`COST_UNIT` (`"cpu_ms_per_workload_evaluation"`) and `COST_SOURCE`. On `main`, -`select_plan` does not read `PlanningModels::cost`. - -```rust -// crates/asap-aware-mapping/src/stage3_plan_selection.rs -pub struct CandidateCost { - pub total: f64, - pub unit: &'static str, - pub source: &'static str, - pub per_node: BTreeMap, -} - -pub struct NodeCost { - pub cost: f64, - pub multiplier: u64, - pub detail: String, -} - -// crates/asap-aware-mapping/src/analytical_cost.rs -pub fn estimate_operator( - operator: PhysicalOperator, - statistics: OperatorStatistics, -) -> Result { - -pub struct ResourceCalibration { - pub cost_per_cpu_op: f64, - pub cost_per_scan_byte: f64, - pub cost_per_retained_byte: f64, - pub version: String, -} -``` - -### Ranking - -`select_plan` keeps the cheapest valid candidate (the first on ties) as -`Selection::selected`; every other valid candidate is listed in `rejected` -with `valid: true` and a `"costlier: …"` reason. Every candidate is priced; -nothing is capped (a search that avoids building every candidate is #671). - -## Stage 4, execution - -Design: [4. Execution](../design_docs/proposals/planner-layering.md#4-execution). -Execution runs in the deployment. The planner hands off the selected -candidate's `PhysicalASAPDAG` (`PhysicalCandidate::dag`; through the facade, -`PlanOutput::execution_timed_dag`). - -| Component | Role | -|---|---| -| `crates/types/src/ir/physical_export.rs` | `PhysicalASAPDAG`, its nodes and edges with data state and materialization; `PhysicalASAPDAGDocument` for serialization. | -| `crates/asap-physical-operators` | Physical operator DAG runtime shared by ingestion time and query time; `physical_planner` compiles a `PhysicalASAPDAG` to native operators. | - -```rust -// crates/types/src/ir/physical_export.rs -pub struct PhysicalASAPDAG { - pub nodes: Vec, - pub edges: Vec, - pub roots: Vec, - // ... scalar roots -} - -// crates/asap-physical-operators/src/physical_planner/mod.rs -pub fn compile( - dag: &PhysicalASAPDAG, - inputs: BTreeMap, - roots: &[NodeId], -) -> Result { -``` - -## Tooling - -| Component | Role | -|---|---| -| `crates/devtools/src/planning_stages.rs` | `run_planning_stages`: Stage 0 roots plus `run_stage1_to_stage3`, as one `PlanningStagesRun`. Shared by `stage_pipeline` and the planner-layering acceptance tests. | -| `crates/devtools/src/bin/stage_pipeline.rs` | Writes an `asap-stage-pipeline/v1` JSON document with every stage's candidates (`--example planner-layering-1`, `--promql …`). | -| `crates/devtools/src/planner_layering_examples.rs` | The design doc's end-to-end examples as workloads. | -| `crates/devtools/tests/stage_pipeline.rs` | Acceptance tests over the examples. | -| `tools/dag-viewer/` | Browser viewer (`server.py`, `viewer.js`, `planner-ui.js`); on `main` its server plans through the `dag_export` binary. See its `README.md`. | - -```rust -// crates/devtools/src/planning_stages.rs -pub fn run_planning_stages( - workload: &PlanningWorkload, - roots: Vec, -) -> Result { - -pub struct PlanningStagesRun { - pub stage0_query_language_frontends: Vec, - pub stage1_logical_asap_optimization: Vec, - pub stage2_physical_asap_optimization: Vec, - pub stage2_pruned: Vec, - pub stage3_plan_selection: Result, -} -``` - -## Supporting modules - -All in `crates/asap-aware-mapping/src/` unless noted. - -| Component | Role | -|---|---| -| `accuracy/estimators/` (`cms.rs`, `count_sketch.rs`, `kll.rs`, `ddsketch.rs`, `hll.rs`, `univmon.rs`, `cardinality.rs`) | Per-family sizing rules and local error bounds (`size_params`). | -| `accuracy/allocation.rs` | `AccuracyBudgetAllocator`, `EqualSplitAllocator`: split an accuracy budget across composed parts. | -| `accuracy/evidence.rs` | Evidence types (`EstimatorContract`, `PropagationStats`, `WorkloadAccuracyEvidence`). | -| `physical_operator_statistics.rs` | Edge and operator statistics for `analytical_cost`. | -| `crates/types/src/post_asap/` | Summary families (`sketch.rs`), `ResultGuarantee` (`guarantee.rs`), data state and materialization (`execution_data_state.rs`). | - -Legacy, to be deleted once the stages implement what they do: `replacement.rs` -(the old candidate search, `search_workload`), `accuracy/reconciliation.rs`, -and the modules that serve only that search; also the `dag_export` devtool. -The stage pipeline does not call them. - -## Not yet on main - -- #682: Stage 3 prices state by its size. -- #686: the viewer plans through `stage_pipeline`. -- #687: guarantee composition through arithmetic over approximate values (#669, part a). -- #604: Stage 2 materialization options. -- Pass 2 identical-expression rule (in progress; supersedes #588). -- #572: crate and module layout by stages and the unified IR. -- #677: doc updates after the consolidation. From 054c30c6e513ae83a3c2be7314743f658f18b8b2 Mon Sep 17 00:00:00 2001 From: zzylol <50204836+zzylol@users.noreply.github.com> Date: Wed, 14 Oct 2026 23:43:34 +0000 Subject: [PATCH 4/4] docs: refresh the planner-layering components doc to current main Describe Pass 2's identical-expression rule, whole-expression top-k candidates, symbolic guarantee composition bound in plan selection, counter weight confirmation, per-second pricing with retained-state memory, and the lazy Stage 1 enumeration; update the code interfaces and the notes on what is not implemented yet. --- .../planner-layering-components.md | 397 +++++++++++------- 1 file changed, 236 insertions(+), 161 deletions(-) diff --git a/docs/design_docs/planner-layering-components.md b/docs/design_docs/planner-layering-components.md index fed65c216..bc241f7d2 100644 --- a/docs/design_docs/planner-layering-components.md +++ b/docs/design_docs/planner-layering-components.md @@ -1,8 +1,8 @@ # Planner Layering: Stage Components, Algorithms and Code Interfaces -Status: describes `main` @ `6665fcc3`. Audience: developers of ASAPPlanner who -read [ASAPPlanner Planning Stages Design](proposals/planner-layering.md) -and need to know how each stage is built today. +Status: describes `main` @ `80fc72e1`. Audience: developers of ASAPPlanner who +read [ASAPPlanner Planning Stages Design](proposals/planner-layering.md) and +need to know how each stage is built today. planner-layering.md defines the stages and passes and their algorithms. This document follows the same order and, for every stage, describes its core @@ -18,60 +18,53 @@ flowchart TB S0["Stage 0, query-language frontends"] P1["Stage 1, logical ASAP-aware optimization
Pass 1: per-computation candidate generation"] C1["Stage 1: composition and sizing
(Cartesian product of per-computation candidates)"] + P2["Stage 1, Pass 2: ASAP-aware CSE
(identical-expression rule)"] S2["Stage 2, physical ASAP-aware optimization"] S3["Stage 3, plan selection"] S4["Stage 4, execution (deployment)"] W --> S0 S0 -- "Vec#lt;QueryRoot#gt;: one LogicalDAG per query" --> P1 P1 -- "PerComputationCandidates" --> C1 - C1 -- "Vec#lt;LogicalCandidate#gt;: CandidateLogicalASAPDAGs" --> S2 + C1 -- "one composed candidate at a time" --> P2 + P2 -- "Stage1Candidate stream: CandidateLogicalASAPDAGs" --> S2 S2 -- "Vec#lt;PhysicalCandidate#gt;: CandidatePhysicalASAPDAGs
+ pruned candidates" --> S3 S3 -- "Selection: one PhysicalASAPDAG" --> S4 ``` -Pass 2 (ASAP-aware CSE) runs after Pass 1 and before sizing in the design; it -is not wired into the pipeline on `main` (Section 3.3). - -Every stage of the pipeline is one plain function over one workload. The -`asap-planner` facade runs Stage 0 and then an `OptimizationPass`; the default -pass, `Stage1ToStage3Pass`, calls `run_stage1_to_stage3`, which runs Stage 1, -Stage 2 and Stage 3 in order and keeps every stage's candidates in a -`Stage1ToStage3Run`: +Every stage is one plain function over one workload. The `asap-planner` facade +runs Stage 0 and then an `OptimizationPass`; the default pass, +`Stage1ToStage3Pass`, reads one accuracy target and one recurrence per +workload entry and calls `run_stage1_to_stage3`: ```text -run_stage1_to_stage3(roots, targets, data, models): - inventory ← generate_per_computation_candidates(roots) // Stage 1, Pass 1 - combinations ← Π_t |inventory.targets[t].choices.candidates| // saturating - for each choice in the Cartesian product (mixed radix, last target fastest): - L_n ← compose_logical_candidate(inventory, choice) // Stage 1, composition + sizing - if L_n composed: - P_n ← optimize_physical_asap(L_n) or prune L_n with the error // Stage 2 - selection ← select_plan(all P_n, targets, data, models) // Stage 3 +run_stage1_to_stage3(roots, targets, recurrence, data, models): + inventory ← generate_per_computation_candidates(roots) // Stage 1, Pass 1 + stage1 ← stage1_logical_asap_optimization(inventory) // lazy stream: + for each choice in the Cartesian product (mixed radix, last target fastest): + skip choices that differ only inside a target an absorbing candidate removes + L ← compose_logical_candidate(inventory, choice) // composition + sizing + yield L, then each Pass 2 identical-expression candidate of L + for each L_n in stage1: // Stage 2 + P_n ← optimize_physical_asap(L_n) or prune L_n with the error + selection ← select_plan(all P_n, targets, recurrence, data, models) // Stage 3 ``` Only a Pass 1 error fails the run. A candidate that does not compose keeps its -error in the run, a candidate Stage 2 cannot implement is pruned with a reason, -and a failed selection is reported as the Stage 3 result. Every combination is -enumerated and priced; nothing is capped; a search that avoids building every -candidate is future work. +error, a candidate Stage 2 cannot implement is pruned with a reason, and a +failed selection is reported as the Stage 3 result. The library enumerates and +prices every candidate and never caps. Because Stage 1 is a lazy stream, a +caller can count candidates as they are produced and stop early: +`run_stage1_to_stage3_from_pass1` takes a finished Pass 1 inventory, and +`run_stage2_and_stage3` takes the Stage 1 candidates a caller enumerated. ### Code interface ```rust -// crates/planner/src/lib.rs -pub async fn e2e_plan(input: UserInput<'_>) -> Result { - -// crates/asap-aware-mapping/src/pass/mod.rs -pub struct PlanningModels<'a> { - pub cost: &'a dyn CostModel, - pub accuracy: &'a dyn AccuracyModel, - pub evidence: &'a dyn AccuracyEvidenceProvider, -} - // crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs pub fn run_stage1_to_stage3( roots: Vec<(Id, QueryRoot)>, targets: &[Option], + recurrence: &[QueryRecurrence], data: &DataWorkload, models: PlanningModels<'_>, ) -> Result, PerComputationCandidateError> { @@ -79,10 +72,19 @@ pub fn run_stage1_to_stage3( pub struct Stage1ToStage3Run { pub stage1_pass1_per_computation: PerComputationCandidates, pub combinations: usize, + pub stage1_pass2_identical_expression: Vec, pub stage1_logical_asap_optimization: Vec, pub stage2_physical_asap_optimization: Vec, pub stage2_pruned: Vec, pub stage3_plan_selection: Result, + pub stage3_selected_logical: Option>, +} + +// crates/asap-aware-mapping/src/pass/mod.rs +pub struct PlanningModels<'a> { + pub cost: &'a dyn CostModel, + pub accuracy: &'a dyn AccuracyModel, + pub evidence: &'a dyn AccuracyEvidenceProvider, } ``` @@ -99,7 +101,7 @@ Design: [0. Language-specific frontends](proposals/planner-layering.md#0-languag | MetricsQL frontend (`crates/frontend-metricsql`) | Converts the MetricsQL AST into the same tree. | | Shared resolver (`crates/frontend-common`) | `UnresolvedOp` tree with name-based column references; `resolve_root` binds each name to a positional `ColumnId` and derives every schema. | | Unified IR (`crates/types/src/ir/`) | One operator DAG type for every stage: `OperatorNode` with `NonASAPOp` or `ASAPOp` operators. | -| Workload (`crates/types/src/workload.rs`) | `QueryWorkload` (queries, recurrence, accuracy and latency requirements) and `DataWorkload` (arrival, ingestion interval, volume, rate, cardinality, distribution, each as `Evidence`). | +| Workload (`crates/types/src/workload.rs`) | `QueryWorkload` (queries, recurrence, accuracy and latency requirements) and `DataWorkload` (arrival, ingestion interval, volume, rate, cardinality, distribution, stream norms, per-source distinct bounds, declared metric types). | ### Core algorithm @@ -158,10 +160,11 @@ Design: [1. Logical ASAP-aware optimization](proposals/planner-layering.md#1-log | Component | What it does | |---|---| | Pass 1 (`crates/asap-aware-mapping/src/per_computation_candidates.rs`, `generate_per_computation_candidates`) | Finds every computation target and its unranked, unsized candidates, with pruned candidates and reasons. | -| Composition (same file, `compose_logical_candidate`) | Builds one whole-workload candidate DAG from one choice per target, sizes its summaries and gives every node a guarantee. | +| Composition (same file, `compose_logical_candidate`) | Builds one whole-workload candidate from one choice per target, sizes its summaries and composes every node's guarantee. | +| Pass 2 (`crates/asap-aware-mapping/src/asap_aware_cse.rs`) | The identical-expression rule: `sharing_groups` and `identical_expression_variants`. | +| Enumeration (`crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs`, `stage1_logical_asap_optimization`) | Streams composed candidates, each followed by its Pass 2 candidates. | | Summary realization (`crates/asap-aware-mapping/src/realization.rs`) | The summary families per intent (`summary_candidates`), the `(ε, δ)` budget of a target, and default sizing. | -| Accuracy model (`crates/asap-aware-mapping/src/accuracy/`) | `DefaultAccuracyModel`: local error bounds of sized families, propagation rules (`composition.rs`), and per-family sizing (`estimators/`). | -| Enumeration (`crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs`) | Walks the Cartesian product of per-computation candidates into `LogicalCandidate`s `L1, L2, …`. | +| Accuracy model (`crates/asap-aware-mapping/src/accuracy/`) | `DefaultAccuracyModel`: local error bounds of sized families, composition rules (`composition.rs`), and per-family sizing and confidence (`estimators/`). | ### 3.1 Pass 1: Per-computation candidate generation @@ -173,14 +176,13 @@ Algorithms 1 and 1.1. On `main`, a computation target is one measure of an `Aggregate`, and Algorithm 1.1's `MATCH` is reading that measure's `AggIntent`: the intent binds the input column, grouping (`Reduction`), filter and parameters such as -`k` or `q`. The rule templates are fixed per candidate kind (Section 3.2), and -the families that fill a summary slot come from a static table, +`k` or `q`. The families that fill a summary slot come from a static table, `summary_candidates(intent)`: | Intent | Summary families | |---|---| | `Quantile` | KLL, DDSketch | -| `Cardinality` | HLL, Theta, KMV, UnivMon (not UnivMon over a tuple) | +| `Cardinality` | HLL, KMV, UnivMon (not UnivMon over a tuple) | | `FrequencyL2`, `FrequencyEntropy` | UnivMon | | `TopK` | CMS with heap, Count Sketch with heap | | `Count` | CMS, Count Sketch, UnivMon | @@ -198,7 +200,9 @@ generate_per_computation_candidates(roots): choices ← computation_candidates_for_intent(i) skip m if its only option is plain Exact if i is avg over a possibly-null input: prune AvgAsSumCount (NullableAvgInput) - emit target (a, m, choices, filter of m, HAVING entries) + samples_of ← the metric whose raw samples (or plain sums of them) a reads + emit target (a, m, choices, filter of m, HAVING entries, samples_of) + ADD_WHOLE_EXPRESSION_CANDIDATES(targets) computation_candidates_for_intent(i): exact option ← ExactAccumulator(kind) for Count, Sum, Min, Max, Rate, IRate, Increase @@ -207,11 +211,21 @@ computation_candidates_for_intent(i): if i has an accuracy target: if the target is Exact: prune every summary_candidates(i) (RequiresExact) else: check ε, δ ∈ (0, 1); add Summary(f) for every f in summary_candidates(i) + +ADD_WHOLE_EXPRESSION_CANDIDATES(targets): + for each top-k target t directly over a per-item sum or count target u: + if u has no other consumer, neither has a filter or HAVING, + and the keyed-additive rule realizes a heap over u's input: + replace t's heap candidates by WholeExpression(same family) + t.absorbs ← u // choosing one removes u's aggregate ``` -Each target thus has exactly one exact option, and pruning happens only for a +Each target has exactly one exact option, and pruning happens only for a provable reason. Nested aggregates are separate targets, and their options -multiply in composition. Pass 1 uses no cost or accuracy model. +multiply in composition. A whole-expression top-k ranks items over the raw +input, weighted by the summed value; when those weights are the samples of a +metric, the summary carries the open requirement that the metric's samples are +non-negative, which Stage 3 confirms. Pass 1 uses no cost or accuracy model. Differences from the design: candidates are not checked against a family's capability beyond the table, and the design's Hydra template for `TopK` @@ -236,11 +250,8 @@ pub struct ComputationTarget { pub choices: ComputationChoices, pub filter: Option, pub having: Vec, -} - -pub struct ComputationChoices { - pub candidates: Vec, - pub pruned: Vec<(ComputationCandidate, PruneReason)>, + pub absorbs: Option, + pub samples_of: Option, } pub enum ComputationCandidate { @@ -251,6 +262,7 @@ pub enum ComputationCandidate { }, Summary(SketchAlgorithm), AvgAsSumCount, + WholeExpression(SketchAlgorithm), } ``` @@ -260,21 +272,24 @@ Design: Algorithm 1, lines 18–21 (`SUBSTITUTE`), and sizing after Pass 2. #### Core algorithm -`run_stage1_to_stage3` enumerates the Cartesian product of the targets' -candidates. `choice[t]` indexes `targets[t].choices.candidates`; the choice -vector is incremented in mixed radix, last target fastest, and `L` is the -`n`-th choice. The run records the product's size in `combinations` -(saturating at `usize::MAX`). For each choice, `compose_logical_candidate` -rewrites the workload bottom-up: +Composition enumerates the Cartesian product of the targets' candidates. +`choice[t]` indexes `targets[t].choices.candidates`; the choice vector is +incremented in mixed radix, last target fastest. A choice that picks a +whole-expression candidate and anything but the first candidate of the target +it absorbs gives the same candidate as with the first, so it is skipped. +`PerComputationCandidates::combinations` counts the rest without enumerating +them: an absorbing pair with `n` candidates over `m`, `w` of them +whole-expression, counts `(n − w) · m + w` (saturating at `usize::MAX`). ```text compose_logical_candidate(inventory, choice): for each aggregate a with a chosen candidate other than Exact: replace a measure by measure, over a's input: - ExactAccumulator → SummaryAgg(exact family) → FinalizeExactAccumulator - Summary(f) → SummaryAgg(SIZED(f, acc(m))) → SummaryEstimate - AvgAsSumCount → finalized sum and count, divided - Exact → a single-measure Aggregate + ExactAccumulator → SummaryAgg(exact family) → FinalizeExactAccumulator + Summary(f) → SummaryAgg(SIZED(f, acc(m))) → SummaryEstimate + WholeExpression(f) → SummaryAgg(SIZED(f)) over the absorbed aggregate's input → SummaryEstimate + AvgAsSumCount → finalized sum and count, divided + Exact → a single-measure Aggregate join several measures back on the group keys; keep a's column names measure filter → SummaryAgg.filter HAVING BelowAggregate → Filter below the state (consumer Filter dropped) @@ -283,26 +298,33 @@ compose_logical_candidate(inventory, choice): for each node, bottom-up: node.guarantee ← COMPOSED_GUARANTEE(node) SIZED(f, acc): - (ε, δ) ← accuracy_budget(acc) // δ defaults to 0.01 - params ← default_size_params(f, intent, ε, δ) // accuracy/estimators: invert f's error bound - -COMPOSED_GUARANTEE(node): - FinalizeExactAccumulator over exact input → exact - SummaryEstimate over exact input, bound applies → DefaultAccuracyModel.local_guarantee(f, query) - Project of plain columns → the child's guarantee - non-ASAP operator over exact inputs → exact - anything else (incl. any value computed from an approximate input) → None + (ε, δ) ← accuracy_budget(acc) // δ defaults to 0.01 + params ← default_size_params(f, intent, ε, δ) // accuracy/estimators: invert f's error bound + // (KLL: scaled for δ by a Gaussian quantile ratio) + +COMPOSED_GUARANTEE(node): // built-in DefaultAccuracyModel; no data or evidence + non-ASAP operator over exact inputs → exact + FinalizeExactAccumulator over exact input → exact + FinalizeExactAccumulator over approximate input → its function's rule (sum, extremum, rate, …) + SummaryEstimate whose family bound applies → local_guarantee(f, query), + composed with its input's if approximate + Aggregate / arithmetic over approximate values → the function's or arithmetic composition rule + Project of plain columns → the child's guarantee + comparison or value predicate over approximate → None ``` -A summary estimate's bound "applies" unless the query is Top-K (a heap sketch -bounds item frequencies, not membership) or the family is Count-Min over -weights not proven non-negative. A `None` guarantee does not reject the -candidate here; Stage 3 decides. +A rule that needs a statistic Stage 1 does not have (an input row count, a +source's stream norm, a metric's sign, an operand domain, classic HLL's +distinct bound) keeps it symbolic as an `Unknown` leaf of the guarantee's +`BoundExpr` or `ProbabilityExpr`; Stage 3 binds it. A family bound does not +apply to Top-K membership (a heap sketch bounds item frequencies), and a +Count-Min bound assumes non-negative weights. A `None` guarantee does not +reject the candidate here; Stage 3 decides. -Differences from the design: with no Pass 2, each summary is sized for its own -measure's requirement, which is the post-Pass-2 sizing with nothing shared. -Guarantees are not yet propagated through operators over approximate values -(`AccuracyModel::propagate` exists but composition does not call it). +Differences from the design: summaries are sized per measure, before Pass 2, +because the only Pass 2 rule on `main` shares identical nodes, which already +have the same size. A guarantee holds per output point and is not scaled by the +number of output groups. #### Code interface @@ -313,6 +335,9 @@ pub fn compose_logical_candidate( choice: &[usize], ) -> Result, PerComputationCandidateError> { +impl PerComputationCandidates { + pub fn combinations(&self) -> usize { + // crates/asap-aware-mapping/src/realization.rs pub fn default_size_params( kind: SketchAlgorithm, @@ -321,44 +346,71 @@ pub fn default_size_params( delta: f64, ) -> SketchParams { -// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs -pub struct LogicalCandidate { - pub id: String, - pub display_name: String, - pub choice: Vec, - pub roots: Result, PerComputationCandidateError>, +// crates/types/src/post_asap/guarantee.rs +pub struct AccuracyGuarantee { + pub metric: ErrorMetric, + pub bound: BoundExpr, + pub failure_probability: ProbabilityExpr, + pub provenance: Vec, } ``` ### 3.3 Pass 2: ASAP-aware common-subexpression elimination Design: [Pass 2](proposals/planner-layering.md#pass-2-asap-aware-common-subexpression-elimination), -Algorithm 2 (identical-expression, summary-capability and window-composition -rules). +Algorithm 2. #### Core algorithm -Not implemented on `main`: `run_stage1_to_stage3` goes from Pass 1 straight to -composition, so no candidate shares a summary across computations, and a sub-DAG -is shared only when the frontends already produced it once. Two building -blocks exist outside the pipeline: +Of the three rules, only the identical-expression rule is implemented. It runs +on each composed candidate: + +```text +sharing_groups(roots): + intern the roots with share_common_sub_dags // hash narrows, typed equality decides + group the original nodes by interned representative, children first + keep groups of two or more legal-to-share nodes // ASAP nodes and nodes with a unique key + +identical_expression_variants(roots, groups): + // Algorithm 2, line 31: one candidate per non-empty subset of the groups. + // Sharing a group also shares every group whose nodes all lie below it, + // so many subsets give the same candidate. The distinct candidates are the + // closed subsets; enumerate them directly with Ganter's Next Closure + // (one closure per step and group) instead of trying all 2^g subsets. + for each closed subset S, in binary-counter order: + emit roots with every group in S made one node serving all its consumers +``` + +The independent (unshared) candidate is kept beside its Pass 2 candidates, so +Stage 3 compares sharing against not sharing. Each Pass 2 candidate is recorded +as an `IdenticalExpressionCandidate` naming the candidate it derives from and +its shared groups. -- **Structural CSE** (`crates/types/src/ir/cse.rs`, `share_common_sub_dags`): - bottom-up hash-consing of the IR; a hash narrows the search and typed - equality decides sharing. It is the structural core of the - identical-expression rule; only the legacy search calls it today. -- **Summary coverage** (`crates/types/src/ir/summary_coverage.rs`, - `SummaryCoverage::derive`): what a summary state computes (`definition`) and - which rows it took (`selection`), the comparison of summary input data that - the sharing rules need. +Differences from the design: the summary-capability and window-composition +rules are not implemented, so no summary serves several different estimates or +windows. `SummaryCoverage::derive` (`crates/types/src/ir/summary_coverage.rs`), +which compares the summary input data of two states, exists for those rules. #### Code interface ```rust -// crates/types/src/ir/cse.rs -pub fn share_common_sub_dags( - roots: Vec<(Id, Rc)>, -) -> Vec<(Id, Rc)> { +// crates/asap-aware-mapping/src/asap_aware_cse.rs +pub fn sharing_groups(roots: &[QueryRoot]) -> Vec { + +pub fn identical_expression_variants( + roots: &[QueryRoot], + groups: &[SharingGroup], +) -> Vec { + +pub struct IdenticalExpressionVariant { + pub shared_groups: Vec, + pub roots: Vec, +} + +// crates/asap-aware-mapping/src/pass/stage1_to_stage3.rs +pub fn stage1_logical_asap_optimization( + per_computation: &PerComputationCandidates, +) -> impl Iterator + '_ { ``` ## 4. Stage 2, physical ASAP-aware optimization @@ -399,8 +451,9 @@ Stage 2 removes, and each is provably invalid (it cannot be implemented). Differences from the design: materialization offers one option, so each logical candidate yields exactly one physical candidate. Algorithm 3's enumeration of `MatAtIngestion(medium)` and `MatAtQuery(medium)` per sub-DAG, its -ingestion-time upstream constraint and retention are not implemented. Parallelism, -partitioning and resource management are TODO in the design as well. +ingestion-time upstream constraint and retention are not implemented. +Parallelism, partitioning and resource management are TODO in the design as +well. ### Code interface @@ -435,55 +488,64 @@ Design: [3. Plan selection](proposals/planner-layering.md#3-plan-selection). | Component | What it does | |---|---| -| Plan selection (`crates/asap-aware-mapping/src/stage3_plan_selection.rs`, `select_plan`) | Rejects candidates that miss an accuracy target, prices the rest over their physical DAG and selects the cheapest. | -| Accuracy model (`AccuracyModel::satisfies`, `crates/asap-aware-mapping/src/accuracy/mod.rs`) | Decides whether a guarantee meets an `AccuracyTarget`. | -| Accuracy evidence (`AccuracyEvidenceProvider`, `crates/asap-aware-mapping/src/accuracy/evidence.rs`) | The deployment's empirical accuracy; `output_guarantee` gives evidence for one candidate's whole query output. `NoAccuracyEvidence` supplies none. | -| Analytical cost (`crates/asap-aware-mapping/src/analytical_cost.rs`) | `estimate_operator`: resource estimate (CPU ops, scan bytes, retained bytes) of one physical operator over edge statistics (`physical_operator_statistics.rs`); `ResourceCalibration` turns it into one number. | +| Plan selection (`crates/asap-aware-mapping/src/stage3_plan_selection.rs`, `select_plan`) | Confirms weight requirements, binds symbolic statistics, rejects candidates that miss an accuracy target, prices the rest per second and selects the cheapest. | +| Accuracy model (`AccuracyModel::satisfies`, `crates/asap-aware-mapping/src/accuracy/mod.rs`) | Decides whether a bound guarantee meets an `AccuracyTarget`; an unknown statistic never meets one. | +| Accuracy evidence (`AccuracyEvidenceProvider`, `crates/asap-aware-mapping/src/accuracy/evidence.rs`) | The deployment's contracts and empirical accuracy: `operand_domain` (enforced value range of an operand), `output_guarantee` (evidence for one candidate's whole query output). `NoAccuracyEvidence` supplies none. | +| Data workload (`DataWorkload`) | Statistics Stage 3 binds: input cardinality, `stream_norms`, `max_distinct_per_evaluation`, `metric_types`. | +| Analytical cost (`crates/asap-aware-mapping/src/analytical_cost.rs`) | `estimate_operator`: CPU operations, scanned bytes and peak memory of one physical operator over edge statistics; `ResourceCalibration` turns them into a price. | ### Core algorithm ```text -select_plan(candidates, targets, data, models): +select_plan(candidates, targets, recurrence, data, models): for each candidate P: // accuracy check (output-level) for each query q with a target T: - if q reads a Count-Min SummaryAgg whose weights are not proven non-negative: - reject P (valid = false) + for each Count-Min SummaryAgg q reads: + weights ← confirmed_weights(weight domain, data) // samples of a declared counter → proven + if weights are not proven non-negative: reject P (valid = false) g ← root guarantee of q (composed in Stage 1) + g ← g.bind(statistic ↦ value) // from never-expiring DataWorkload evidence + // and models.evidence.operand_domain if not (g satisfies T under models.accuracy or models.evidence.output_guarantee(q) satisfies T): reject P (valid = false) - // cost - cost(P) ← PRICE(P.dag, data) - selected ← the valid candidate with the least cost (the first on ties) - every other valid candidate → rejected (valid = true, "costlier: …") + cost(P) ← PRICE(P.dag, recurrence, data, illustrative_calibration()) + selected ← the valid candidate with the least cost (the first on a tie) + every other valid candidate → rejected (valid = true, "costlier" or "tied") + plan ← the selected candidate's roots with confirmed weights no valid candidate → SelectionError::NoValidCandidate -PRICE(dag, data): - series, rows/ms ← data.input_cardinality, data.ingestion_rate (or defaults) +PRICE(dag, recurrence, data, calibration): // cost per second of wall time for each node, children first: estimate ← estimate_operator(physical operator, edge statistics) - // Scan, HashAggregate, Sort, Limit, …; SummaryAgg: rows × sketch depth; - // SummaryEstimate: groups × k - multiplier ← 1 if materialized, else the number of consumers that reach it - (queries, and materialized nodes, without looking through one) - node cost ← estimate.calibrated_cost(calibration) × multiplier + // SummaryAgg: rows × update operations per row (sketch depth; +1 per heap; + // UnivMon: expected layers updated); state bytes from the sketch's params + // SummaryEstimate: groups × k + per_second ← how often the node is computed: + NotMaterialized: once per consumer that reaches it, at that consumer's rate + MatAtQuery: once, at the combined rate of the queries reaching it + MatAtIngestion: once per second of arriving rows + // a query repeating every I ms runs 1000 / I times per second; + // one-off and unknown queries are spread over calibration.horizon_s + cost ← estimate.calibrated_cost(CPU ops, scanned bytes) × per_second + cost += kept state bytes × cost_per_retained_byte_second // materialized nodes only return Σ node costs ``` The check is on the query output, not on the summaries inside the plan: a certified estimate is not enough on its own, since a difference or ratio of estimates can be unbounded. Cost is computed for the whole workload, so a -node shared by several queries is charged per computation; with every node -`NotMaterialized`, a shared node is charged once per query. +shared `NotMaterialized` node is charged for each consumer that recomputes it, +and queries with different recurrences add up. -Differences from the design: pricing uses a fixed illustrative -`ResourceCalibration` (`illustrative-v1`) and default statistics -(`DEFAULT_GROUP_COUNT` groups and default series and rates when the -`DataWorkload` declares none), so prices are not calibrated; `select_plan` does -not read `PlanningModels::cost`. A binary, join or concat node is priced from -its first input only. Latency bounds and deployment capabilities are not -checked yet. +Differences from the design: pricing uses a fixed `illustrative_calibration` +(`illustrative-v2`) and default statistics (a default group count, and default +series and rates when the `DataWorkload` declares none), so prices are not +calibrated; `select_plan` does not read `PlanningModels::cost`. Memory is +charged only as state kept between computations, not as peak capacity. A +binary, join or concat node is priced from its first input only. Latency +bounds and deployment capabilities are not checked. ### Code interface @@ -492,30 +554,41 @@ checked yet. pub fn select_plan( cands: &[PhysicalCandidate], targets: &[Option], + recurrence: &[QueryRecurrence], data: &DataWorkload, models: PlanningModels<'_>, ) -> Result { pub struct Selection { pub selected: String, + pub plan: Vec, pub costs: BTreeMap, pub rejected: Vec, } -pub struct Rejection { - pub id: String, - pub valid: bool, - pub reason: String, +pub struct NodeCost { + pub cost: f64, + pub multiplier: u64, + pub per_second: f64, + pub detail: String, } // crates/asap-aware-mapping/src/accuracy/evidence.rs - fn output_guarantee(&self, _output: &QueryRoot) -> Option { + fn operand_domain(&self, _operand: &OperatorNode) -> Option { + fn output_guarantee(&self, _output: &QueryRoot) -> Option { + +// crates/types/src/post_asap/guarantee.rs + pub fn bind(&self, value: &dyn Fn(&str) -> Option) -> Self { // crates/asap-aware-mapping/src/analytical_cost.rs -pub fn estimate_operator( - operator: PhysicalOperator, - statistics: OperatorStatistics, -) -> Result { +pub struct ResourceCalibration { + pub cost_per_cpu_op: f64, + pub cost_per_scan_byte: f64, + pub cost_per_retained_byte: f64, + pub cost_per_retained_byte_second: f64, + pub horizon_s: f64, + pub version: String, +} ``` ## 6. Stage 4, execution @@ -533,11 +606,12 @@ Design: [4. Execution](proposals/planner-layering.md#4-execution) and ### Core algorithm The planner hands off the selected candidate's `PhysicalASAPDAG` -(`PhysicalCandidate::dag`; through the facade, `PlanOutput` and its -`execution_timed_dag`). The deployment runs it as given: it does not choose -among summaries or change materialization. Ingestion-time nodes form the -precompute DAG and the rest the query DAG; `compile` builds native operators -with typed inputs, which the deployment binds to its data sources. +(`PhysicalCandidate::dag`) together with `Selection::plan`, the selected roots +with every weight requirement Stage 3 confirmed, so the executor reads no data +workload. The deployment runs the plan as given: it does not choose among +summaries or change materialization. Ingestion-time nodes form the precompute +DAG and the rest the query DAG; `compile` builds native operators with typed +inputs, which the deployment binds to its data sources. ### Code interface @@ -561,29 +635,30 @@ pub fn compile( ## 7. Tooling and legacy code - `run_planning_stages` (`crates/devtools/src/planning_stages.rs`) runs Stage 0 - through Stage 3 for one workload; the `stage_pipeline` binary writes every - stage's candidates as an `asap-stage-pipeline/v1` document, and the - planner-layering acceptance tests use the same function. -- `tools/dag-viewer` is the browser viewer; on `main` its server plans through - the `dag_export` binary. + through Stage 3 for one workload with the built-in models. With + `max_combinations`, it counts Stage 1 candidates as they are enumerated and + refuses the workload once the count passes the limit; it never truncates. +- The `stage_pipeline` binary writes every stage's candidates as an + `asap-stage-pipeline/v1` document (default limit 10^4 Stage 1 candidates). + `tools/dag-viewer` plans through it and shows the stages. - Legacy, to be deleted once the stages implement what it does: `replacement.rs` (the old candidate search, `search_workload`), `accuracy/reconciliation.rs`, - the modules only that search uses, and `dag_export`. The stage pipeline - does not call them. + and the modules only that search uses. The stage pipeline does not call them. ## 8. Not yet on main -- **Pass 2, ASAP-aware CSE.** None of the three sharing rules runs in the - pipeline; summaries are sized per computation, never for the strictest - requirement among shared consumers. -- **Guarantee composition.** Guarantees are not propagated through arithmetic - or comparisons over approximate values; such outputs carry no guarantee and - are admitted only with deployment evidence. +- **Pass 2, summary sharing.** The summary-capability and window-composition + rules are not implemented; no summary serves several estimates or windows, + and no shared summary is sized for the strictest of its consumers. - **Materialization.** Stage 2 assigns `NotMaterialized` everywhere; the per-sub-DAG choice of ingestion time or query time and medium is not enumerated. -- **Cost.** Stage 3 prices with fixed illustrative statistics and calibration, - does not price retained state by its size, and does not use the deployment's - cost model; latency bounds and capabilities are not checked. -- **Layout and tooling.** Crates and modules are not yet organized by stage, - and the viewer does not yet plan through `stage_pipeline`. +- **Guarantees.** A guarantee is per output point and not scaled by the number + of groups; comparisons and value predicates over approximate values have no + composition rule. +- **Cost.** Stage 3 prices with illustrative statistics and calibration and + does not use the deployment's cost model; peak memory capacity, latency + bounds and capabilities are not checked. +- **Search.** Every candidate is built and priced; a large workload needs early + pruning or a search that avoids enumerating the whole product. +- **Layout.** Crates and modules are not yet organized by stage.