feat!: install the Planner's InstallablePlan; store its stored outputs as summaries

Cargo.toml

@@ -7,12 +7,10 @@ edition = "2021"
 version = "0.1.0"
 
 [workspace.dependencies]
-# Keep every Planner crate on the same immutable revision. Alias upstream
-# asap-types as planner-types to name its origin at use sites.
-planner-types = { package = "asap-types", git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" }
-asap-frontend-promql = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" }
-asap-logical-asap-optimization = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" }
-asap-plan-selection = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" }
+# The backend depends on two Planner crates, at one immutable revision:
+# `asap-planner` to plan and `asap-executor` to execute. Planner IR types
+# are reached through `asap_executor::planner`.
+asap-planner = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" }
 asap-executor = { git = "https://github.com/ProjectASAP/ASAPPlanner", rev = "ecea897f78640c377ce1d80846156f330f62dee2" }
 
 serde = { version = "1.0", features = ["derive"] }

README.md

@@ -17,9 +17,10 @@ given.
 ```text
 PlanningWorkload (JSON)       DeploymentCapabilities + PlanSelectionSettings
           \                             /
-           control_plane: ASAPPlanner stages 0-3
+           control_plane: asap_planner::plan_installable
                          |
-           plan.json: selected PhysicalASAPDAG + query per root
+           plan.json: InstallablePlan (selected PhysicalASAPDAG, query per
+                      root, stored outputs with retention, raw windows)
                          |
            data_plane: Remote Write in, Prometheus query API out
 ```
@@ -36,6 +37,12 @@ The backend is being rebuilt on the Planner's physical DAG:
 | Precompute engine: build ingestion-time outputs once per tumbling pane | done |
 | Query engine: run query-time nodes over stored panes and raw data | planned |
 
+The data plane stores the plan's `stored_outputs`. Outputs that differ only
+in which pane they read share one stored summary, kept for the largest
+`retention_ms` the plan states among them (which covers range-evaluation
+spans), plus one pane width so an evaluation one pane behind the newest
+closed pane is still served; older panes are evicted.
+
 Panes close `--allowed-lateness-ms` (default 15 s) after the newest sample
 of a stored summary's metric, capped at the wall clock, passes their end; later samples for them are
 dropped. The first pane after startup is skipped because it may be partial.

control_plane/Cargo.toml

@@ -4,10 +4,8 @@ version.workspace = true
 edition.workspace = true
 
 [dependencies]
-planner-types.workspace = true
-asap-frontend-promql.workspace = true
-asap-logical-asap-optimization.workspace = true
-asap-plan-selection.workspace = true
+asap-planner.workspace = true
+futures = "0.3"
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true

control_plane/src/lib.rs

@@ -1,35 +1,19 @@
 //! Backend control plane: asks ASAPPlanner for the workload's plan and
 //! writes it for the data plane to install.
 //!
-//! The Planner runs every planning stage, including plan selection. This
-//! crate only supplies the deployment inputs (what the data plane can execute
-//! and keep, and how Stage 3 prices candidates) and records which query each
-//! root of the selected `PhysicalASAPDAG` answers.
+//! The Planner runs every planning stage, including plan selection, and
+//! returns the selected DAG ready to install ([`InstallablePlan`]): the DAG,
+//! the query each root answers, the outputs to store with how long each
+//! must stay available, and the raw-sample windows read at query time. This
+//! crate only supplies the deployment inputs: what the data plane can
+//! execute and keep, and how Stage 3 prices candidates.
 
 use std::path::Path;
 
-use asap_logical_asap_optimization::DeploymentCapabilities;
-use asap_plan_selection::{
-    run_stage1_to_stage3, PlanSelectionSettings, PlanningModels, Stage1ToStage3Run,
+use asap_planner::{
+    plan_installable, DeploymentCapabilities, FrontendInput, InstallablePlan, InstallablePlanError,
+    PlanSelectionSettings, PlanningModels, PlanningWorkload, UserInput,
 };
-use planner_types::ir::operator::pane_chain::expand_pane_chains;
-use planner_types::ir::physical_export::{compile_physical_asap_workload, PhysicalASAPDAGDocument};
-use planner_types::ir::schema::schema_support::with_promql_series_identity;
-use planner_types::ir::QueryRoot;
-use planner_types::workload::{PlanningWorkload, QueryLanguage, QueryWorkloadEntry};
-use serde::{Deserialize, Serialize};
-
-/// The plan the data plane installs: the Planner's selected physical DAG and
-/// the query each of its roots answers.
-#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
-#[serde(deny_unknown_fields)]
-pub struct InstalledPlan {
-    /// PromQL text of each root of `dag`, in root order.
-    pub queries: Vec<String>,
-    /// The selected candidate's DAG, with every `PaneChain` expanded into
-    /// the panes the executor runs.
-    pub dag: PhysicalASAPDAGDocument,
-}
 
 /// What the backend gives the Planner about its deployment. The Planner
 /// selects; these only say what can run and what it costs.
@@ -44,153 +28,63 @@ pub struct DeploymentInputs {
 
 #[derive(Debug, thiserror::Error)]
 pub enum PlanError {
-    #[error("only PromQL workloads are supported, got {0:?}")]
-    UnsupportedLanguage(QueryLanguage),
-    #[error("workload: {0}")]
-    Workload(String),
-    #[error("query {0} lowers to a scalar, which the data plane cannot answer yet")]
-    ScalarRoot(usize),
-    #[error("PromQL lowering: {0}")]
-    Lowering(String),
-    #[error("query {query}: series identity: {message}")]
-    SeriesIdentity { query: usize, message: String },
     #[error("planning: {0}")]
-    Planning(String),
-    #[error("invalid physical DAG: {0}")]
-    InvalidDag(String),
-    #[error("the DAG has {roots} roots for {queries} queries")]
-    RootCount { roots: usize, queries: usize },
+    Planning(#[from] asap_planner::PlanError),
+    #[error("invalid plan: {0}")]
+    InvalidPlan(#[from] InstallablePlanError),
     #[error("{path}: {message}")]
     File { path: String, message: String },
 }
 
-impl InstalledPlan {
-    pub fn validate(&self) -> Result<(), PlanError> {
-        self.dag
-            .validate()
-            .map_err(|e| PlanError::InvalidDag(e.to_string()))?;
-        let roots = self.dag.dag.roots.len();
-        if roots != self.queries.len() {
-            return Err(PlanError::RootCount {
-                roots,
-                queries: self.queries.len(),
-            });
-        }
-        Ok(())
-    }
-
-    /// Reads and validates a plan written by [`InstalledPlan::save`].
-    pub fn load(path: &Path) -> Result<Self, PlanError> {
-        let file_error = |message: String| PlanError::File {
-            path: path.display().to_string(),
-            message,
-        };
-        let text = std::fs::read_to_string(path).map_err(|e| file_error(e.to_string()))?;
-        let plan: Self = serde_json::from_str(&text).map_err(|e| file_error(e.to_string()))?;
-        plan.validate()?;
-        Ok(plan)
-    }
-
-    pub fn save(&self, path: &Path) -> Result<(), PlanError> {
-        let text = serde_json::to_string_pretty(self).expect("plans serialize");
-        std::fs::write(path, text).map_err(|e| PlanError::File {
-            path: path.display().to_string(),
-            message: e.to_string(),
-        })
-    }
-
-    /// The installed plan for `roots`, the timed roots of one candidate in
-    /// query order.
-    fn from_roots(queries: Vec<String>, roots: &[QueryRoot]) -> Result<Self, PlanError> {
-        let roots = roots
-            .iter()
-            .enumerate()
-            .map(|(i, root)| match root {
-                // The executor runs a pane chain as its panes.
-                QueryRoot::Operator(node) => expand_pane_chains(node)
-                    .map(QueryRoot::Operator)
-                    .map_err(|e| PlanError::InvalidDag(e.to_string())),
-                QueryRoot::Scalar(_) => Err(PlanError::ScalarRoot(i)),
-            })
-            .collect::<Result<Vec<_>, _>>()?;
-        let dag = compile_physical_asap_workload(&roots)
-            .map_err(|e| PlanError::InvalidDag(e.to_string()))?;
-        let plan = Self {
-            queries,
-            dag: PhysicalASAPDAGDocument::new(dag),
-        };
-        plan.validate()?;
-        Ok(plan)
-    }
-}
-
-/// One planned workload: its queries and what Stages 1–3 returned.
-struct Planned {
-    queries: Vec<String>,
-    run: Stage1ToStage3Run<usize>,
-}
-
-/// Numbers the lowered PromQL roots and makes their rows carry each
-/// series' full identity, the representation per-series state needs at
-/// runtime.
-fn with_series_identity(roots: Vec<QueryRoot>) -> Result<Vec<(usize, QueryRoot)>, PlanError> {
-    roots
-        .into_iter()
-        .enumerate()
-        .map(|(i, root)| match root {
-            QueryRoot::Operator(node) => with_promql_series_identity(&node)
-                .map(|node| (i, QueryRoot::Operator(node)))
-                .map_err(|message| PlanError::SeriesIdentity { query: i, message }),
-            QueryRoot::Scalar(_) => Err(PlanError::ScalarRoot(i)),
-        })
-        .collect()
-}
-
-/// Lowers `workload` (Stage 0) and runs Stages 1–3 under `inputs`.
-fn plan(workload: &PlanningWorkload, inputs: &DeploymentInputs) -> Result<Planned, PlanError> {
-    let language = &workload.query_workload.language;
-    if *language != QueryLanguage::PromQL {
-        return Err(PlanError::UnsupportedLanguage(language.clone()));
-    }
-    workload
-        .validate()
-        .map_err(|e| PlanError::Workload(e.to_string()))?;
-    let roots = asap_frontend_promql::lower_promql_query_workload(workload)
-        .map_err(|e| PlanError::Lowering(e.to_string()))?;
-    let roots = with_series_identity(roots)?;
-    let entries: Vec<QueryWorkloadEntry> = workload.query_workload.entries().collect();
-    let data = workload.data_workload.clone().unwrap_or_default();
+/// Plans `workload` with the Planner (frontend and Stages 1–3) and returns
+/// the plan Stage 3 selected, ready to install. Only PromQL workloads are
+/// planned: the Planner refuses another language for the PromQL frontend.
+/// It rejects every candidate that needs more than
+/// `inputs.deployment_capabilities`, and refuses a plan a runtime could not
+/// run through the executor.
+pub fn plan_workload(
+    workload: &PlanningWorkload,
+    inputs: &DeploymentInputs,
+) -> Result<InstallablePlan, PlanError> {
     let models = PlanningModels::builtin()
         .with_deployment_capabilities(&inputs.deployment_capabilities)
         .with_plan_selection(&inputs.plan_selection);
-    let run = run_stage1_to_stage3(roots, &entries, &data, models)
-        .map_err(|e| PlanError::Planning(e.to_string()))?;
-    Ok(Planned {
-        queries: entries.into_iter().map(|e| e.query.0).collect(),
-        run,
+    let input = UserInput::new(workload, FrontendInput::Promql { histograms: None }, models);
+    // Only the SQL frontend awaits anything; PromQL planning completes on
+    // its first poll.
+    let plan = futures::executor::block_on(plan_installable(input))?;
+    if let Some(selection) = &plan.selection {
+        tracing::info!(candidate = %selection.selected_candidate_id, "selected physical plan");
+    }
+    Ok(plan)
+}
+
+/// Writes `plan` as the plan file the data plane installs.
+pub fn save_plan(plan: &InstallablePlan, path: &Path) -> Result<(), PlanError> {
+    let text = serde_json::to_string_pretty(plan).expect("plans serialize");
+    std::fs::write(path, text).map_err(|e| PlanError::File {
+        path: path.display().to_string(),
+        message: e.to_string(),
     })
 }
 
-/// Plans `workload` with the Planner's stage pipeline (Stages 0–3) and
-/// returns the plan Stage 3 selected. The Planner rejects every candidate
-/// that needs more than `inputs.deployment_capabilities`.
-pub fn plan_workload(
-    workload: &PlanningWorkload,
-    inputs: &DeploymentInputs,
-) -> Result<InstalledPlan, PlanError> {
-    let planned = plan(workload, inputs)?;
-    let selection = planned
-        .run
-        .stage3_plan_selection
-        .map_err(|e| PlanError::Planning(e.to_string()))?;
-    tracing::info!(candidate = %selection.selected, "selected physical plan");
-    InstalledPlan::from_roots(planned.queries, &selection.plan)
+/// Reads and validates a plan written by [`save_plan`].
+pub fn load_plan(path: &Path) -> Result<InstallablePlan, PlanError> {
+    let file_error = |message: String| PlanError::File {
+        path: path.display().to_string(),
+        message,
+    };
+    let text = std::fs::read_to_string(path).map_err(|e| file_error(e.to_string()))?;
+    let plan: InstallablePlan =
+        serde_json::from_str(&text).map_err(|e| file_error(e.to_string()))?;
+    plan.validate()?;
+    Ok(plan)
 }
 
 /// What this backend's data plane can execute and keep: every summary the
 /// Planner executor builds, raw samples kept in Prometheus, and no
-/// query-time result kept across evaluations (the data plane does not
-/// implement `MatAtQueryKeptAcrossEvaluations`).
+/// query-time result kept across evaluations: the SDS describes such
+/// outputs, but nothing builds them until the query engine does.
 pub fn backend_capabilities() -> DeploymentCapabilities {
     DeploymentCapabilities {
         supported_summaries: None,
@@ -209,27 +103,3 @@ impl Default for DeploymentInputs {
         }
     }
 }
-
-#[cfg(test)]
-mod tests {
-    use super::*;
-    use planner_types::ir::schema::Schema;
-    use planner_types::ir::{NonASAPOp, Operator, OperatorNode};
-
-    /// A root whose rows cannot carry the full series identity is rejected
-    /// rather than planned without it.
-    #[test]
-    fn rejects_a_root_without_a_series_identity_realization() {
-        let values = OperatorNode::new_shared(Operator::NonASAP(NonASAPOp::Values {
-            rows: vec![],
-            schema: Schema::default(),
-        }))
-        .unwrap();
-        let error = with_series_identity(vec![QueryRoot::Operator(values)])
-            .expect_err("the identity rewrite must fail");
-        assert_eq!(
-            error.to_string(),
-            "query 0: series identity: operator has no dynamic series-identity realization"
-        );
-    }
-}

control_plane/src/main.rs

@@ -1,9 +1,9 @@
 use std::path::{Path, PathBuf};
 
 use anyhow::Context;
+use asap_planner::PlanningWorkload;
 use clap::Parser;
-use control_plane::{plan_workload, DeploymentInputs};
-use planner_types::workload::PlanningWorkload;
+use control_plane::{plan_workload, save_plan, DeploymentInputs};
 
 /// Plan a workload with ASAPPlanner and write the selected plan.
 #[derive(Parser)]
@@ -45,7 +45,7 @@ fn main() -> anyhow::Result<()> {
         inputs.plan_selection.measured_operation_costs = read_json(path)?;
     }
     let plan = plan_workload(&workload, &inputs)?;
-    plan.save(&args.out)?;
+    save_plan(&plan, &args.out)?;
     println!(
         "wrote {} ({} queries, {} nodes)",
         args.out.display(),