[zzylol]

Base: main (#820 merged). Next: #821 (query engine), then query-run summaries, then the #814 watermark-barrier port.

Part of #837 (SDS stores summaries in general). This PR covers:

• the SDS model for every stored output the plan lists, whether built in an ingestion run or in a query run and kept;
• summary instances keyed by SummaryInstanceKey { definition, selection }, with the Planner's retention. This is ASAPPlanner's summary coverage (definition plus selection) restricted to time-only selections. Once StoredOutput carries the coverage (ASAPPlanner fix(control-plane): match a Planner fragment against its own subtree #781), the SDS uses the plan's definition id and selection instead of its own hash and coverage end;
• ingestion builds in the plan's dependency order.

Still to come under #837: building and reusing query-run summaries (with the query engine, #821 and the kept-results PR after it), and summaries built from other stored outputs (after ASAPPlanner #781).

Why

The data plane worked out for itself what to store and for how long. It found the materialization frontier from node timings, walked each output's pane shape, and kept a definition's panes for "largest offset plus two pane widths". The Planner already returns all of this in its deployment handoff (InstallablePlan): the outputs to store, how each one is built, the pane window it reads, and how long it must stay available, range-evaluation spans included. The design boundary says the Planner decides what to compute and keep, and the backend decides how.

What

• Control plane: plan_workload calls asap_planner::plan_installable once and returns its InstallablePlan. save_plan/load_plan write it as plan.json and read and validate it. The backend's Stage 0–3 plumbing, pane-chain expansion and series-identity rewrite are deleted, because the Planner does them. A non-PromQL workload is refused by the Planner's PromQL frontend.
• SDS: Sds::from_plan(&InstallablePlan) reads stored_outputs.
  • Outputs that differ only in their pane's offset share one SummaryDefinition, using the SDS's own identity (sub-DAG hash, offset-only TimeShifts transparent).
  • Each definition keeps its builder's pane window (RawSampleWindow).
  • Every definition records which run builds it (built_at). An output built in a query run and kept (StoredOutputBuild::QueryRun) is described like any other; the precompute engine leaves it to the query engine. Until the query engine builds such summaries, the backend still declares query_time_results_retained: false.
  • Each definition lists the stored outputs its builder reads (stored_inputs). Sds::ingestion_build_order orders ingestion-run definitions after their inputs and fails on a cycle. Today every stored output reads raw samples, so the lists are empty. The precompute engine refuses a definition with stored inputs until it can build one once its inputs are sealed (ASAPPlanner fix(control-plane): match a Planner fragment against its own subtree #781).
  • ingestion_frontier and the pane-shape walk are deleted.
• Retention source: each definition is kept for the largest retention_ms among its outputs in stored_outputs. That is the Planner's retention per node: its input's offset plus its window plus the longest range-evaluation span of its readers. Retention is measured from the newest evaluation the stored panes can serve: the end E of the newest closed pane at or before the watermark. Panes are kept one pane width beyond the plan's retention, so an evaluation at E − pane width is still served. Panes ending before oldest_kept_end_ms(E − pane width) are evicted, and when the watermark jumps they are skipped rather than built. This replaces "largest offset plus two pane widths". Example 4b without raw data keeps its 10-minute panes for the plan's 60 minutes plus one pane.
• Precompute: builds ingestion-run definitions in ingestion_build_order. Each definition's RawSampleSelector is built from the plan's coverage window, and the backend's own Scan search is gone.
• Dependencies: among Planner crates the workspace depends on asap-planner and asap-executor only, at ASAPPlanner main ecea897f; IR types come through asap_executor::planner. scripts/sync-asapplanner-main.sh, its test and scripts/ci/README.md update both. asap-planner currently pulls in the SQL and MetricsQL frontends (DataFusion), so Cargo.lock grows by about 1,800 lines. ASAPPlanner docs: design offline ERP planning from physical selection to deployment #779 makes those frontends optional features; after it lands the backend can depend on asap-planner with default-features = false, features = ["promql"].

How

• Control plane: plan_workload builds a UserInput (PromQL frontend, built-in models with the backend's capabilities and plan selection settings) and blocks on plan_installable. PromQL planning completes on its first poll. save_plan writes the result as plan.json.
• SDS (Sds::from_plan): one pass over stored_outputs.
  • Hash each output's sub-DAG, memoized per node, with input ids blanked and offset-only TimeShifts transparent. The hash is the definition id.
  • Group outputs by definition id. The lowest-numbered output is the builder, and its pane is the definition's window.
  • Take the largest retention_ms and record built_at, with ingestion winning.
  • Walk each builder's inputs to find the stored outputs it reads (stored_inputs), then topologically sort ingestion-run definitions (ingestion_build_order).
• Precompute: per ingestion-run definition, in build order, compile the builder from its TimeRange and build a RawSampleSelector from the window. Panes close as in feat(precompute): build stored summary panes from Remote Write samples #820.
  • Eviction and the expired-pane skip use oldest_kept_end_ms(E − pane width), where E is the end of the newest closed pane at or before the watermark. Panes are therefore kept one pane width beyond the plan's retention, so an evaluation one pane behind E is still served.

Key code interfaces added / modified

// control_plane/src/lib.rs
/// Plans `workload` with the Planner and returns the selected plan, ready to install.
pub fn plan_workload(workload: &PlanningWorkload, inputs: &DeploymentInputs) -> Result<InstallablePlan, PlanError>;
/// Writes `plan` as the plan file the data plane installs.
pub fn save_plan(plan: &InstallablePlan, path: &Path) -> Result<(), PlanError>;
/// Reads and validates a plan written by `save_plan`.
pub fn load_plan(path: &Path) -> Result<InstallablePlan, PlanError>;
pub enum PlanError {
    Planning(asap_planner::PlanError),
    InvalidPlan(InstallablePlanError),
    File { path: String, message: String },
}

// data_plane/src/sds.rs
pub struct SummaryDefinition {
    pub id: DefinitionId,
    /// Which run builds its summaries; ingestion when any output says so.
    pub built_at: StoredOutputBuild,
    /// Pane width: each stored summary covers `(end - pane_ms, end]`.
    pub pane_ms: i64,
    /// The output that builds a pane (the first in node order).
    pub builder: u64,
    /// The builder's pane: its `TimeRange` and the `Scan` below.
    pub window: RawSampleWindow,
    pub output_schema: Schema,
    /// Definitions of the stored outputs the builder reads.
    pub stored_inputs: Vec<DefinitionId>,
    /// The largest `retention_ms` the plan states for the definition's outputs.
    pub retention_ms: i64,
}
impl SummaryDefinition {
    /// The builder's `TimeRange`, which one pane's samples feed.
    pub fn raw_input(&self) -> u64;
    /// End of the oldest pane still read when the newest evaluation is at `newest_ms`.
    pub fn oldest_kept_end_ms(&self, newest_ms: i64) -> i64;
}
impl Sds {
    /// The definitions behind `plan`'s stored outputs.
    pub fn from_plan(plan: &InstallablePlan) -> Result<Self, SdsError>;
    /// Ingestion-run definitions, each after the definitions it reads.
    pub fn ingestion_build_order(&self) -> Result<Vec<DefinitionId>, SdsError>;
}
pub enum SdsError {
    UnknownNode(u64),
    RetentionOverflow(u64),
    Cycle(DefinitionId),
    /// An ingestion-run definition reads a query-run one.
    IngestionReadsQueryRun { definition: DefinitionId, input: DefinitionId },
}
/// Which data one summary instance covers; time-only for now: (end - width, end].
pub struct Selection { pub end_ms: i64 }
/// One stored summary: its definition and the data it covers.
pub struct SummaryInstanceKey { pub definition: DefinitionId, pub selection: Selection }
impl<S> SummaryStore<S> {
    pub fn insert(&mut self, key: SummaryInstanceKey, state: S);
    /// Borrows the definition id; a lookup allocates nothing.
    pub fn get(&self, definition: &DefinitionId, selection: Selection) -> Option<&S>;
    /// Drops the instances of `definition` whose coverage ends before `oldest`'s.
    pub fn evict_before(&mut self, definition: &DefinitionId, oldest: Selection);
}

Before this PR

plan.json was the backend's own InstalledPlan {queries, dag}. The data plane derived the frontier from node timings, walked each output down to its Scan to find its pane, and kept Example 4b's panes for 70 minutes (50-minute offset plus two 10-minute panes).

After this PR

plan.json is the Planner's InstallablePlan, and the data plane reads what to store and for how long from it. Example 4b without raw data has one definition of six 10-minute KLL panes, kept for the plan's 60 minutes.

Usage

// Plan #509 Example 4b for a deployment that keeps no raw data.
let workload: PlanningWorkload = serde_json::from_str(
    &std::fs::read_to_string("examples/planner-layering-example4b.json")?)?;
let mut inputs = DeploymentInputs::default();
inputs.deployment_capabilities.raw_data_retained = false;
let plan = control_plane::plan_workload(&workload, &inputs)?;
// stored_outputs=6, largest retention_ms=3600000, selected P3-m1

// Install: the stored summaries behind the plan's stored outputs.
let sds = Sds::from_plan(&plan)?;
// definitions=1, pane_ms=600000, retention_ms=3600000, built_at=IngestionRun

// Ingest one series, a sample every 15 s from 1 min to 75 min.
let options = PrecomputeOptions { allowed_lateness_ms: 0, ..PrecomputeOptions::default() };
let mut precompute = Precompute::new(plan.dag.dag.clone(), &sds, options)?;
precompute.ingest(&samples);
// samples_accepted: 259, samples_skipped_first_pane: 37, panes_built: 6
// stored summaries: Selection::ending_at(20, 30, 40, 50, 60, 70 min)

Over HTTP the same flow is control_plane --workload … --deployment-capabilities caps.json --out plan.json (with caps.json containing {"raw_data_retained": false}), then data_plane --plan plan.json, then Remote Write. /api/v1/status/asap reports the same counters.

Verification

• control_plane/tests/plan_workload.rs: round trip through save_plan/load_plan; Example 4b's six stored outputs (10-minute panes, offsets 0..50 min, largest retention 60 min); SQL refused by the Planner; a plan with a missing query refused by InstallablePlan::validate.
• data_plane/tests/sds.rs: shared definition and offsets, the largest retention among outputs, no definitions for a query-time plan, identity independent of node numbering, identity dependent on the source, QueryRun outputs as query-run definitions, ingestion winning when outputs share a definition, build order and cycle detection, build_order_puts_inputs_first (three definitions), build_order_refuses_ingestion_reading_query_run, and the store.
• data_plane/tests/precompute.rs (19 tests): feat(precompute): build stored summary panes from Remote Write samples #820's tests with the plan's retention, plus does_not_build_query_run_summaries and keeps_every_pane_the_newest_served_evaluation_reads, and keeps_one_pane_of_margin_beyond_the_retention. The last one is the review repro (watermark 84:45, evaluation at 70 min): it failed on fae43fe and passes now, and it also checks that the store stays at 7 panes. keeps_every_pane_the_newest_served_evaluation_reads failed on bc988c7: with the watermark at 74:45 the pane ending at 20 min was skipped, though the evaluation at 70 min reads it. The feat(query): answer planned instant queries by running the plan's roots #821 restack's HTTP test found this. A 30-day watermark jump builds the seven retained panes, and eviction at a 100-minute watermark keeps panes ending at 40..100 minutes.
• Ran the MVP CI steps locally (sync-script tests, the node merge test, fmt, check, clippy -D warnings on Rust 1.99, cargo test --workspace, all --locked). All passed.