feat(planner): plan_installable, the one call a deployment runtime makes (#749)

[zzylol]

Why

A deployment runtime (ASAPQuery-backend or any other) should call ASAPPlanner once, get back the single selected DAG, and install it. It needs three things from the plan: which DAG to install, which outputs to store, and how to feed samples in. Today the backend links Planner internals for each of these (#749):

• it re-runs run_stage1_to_stage3;
• it expands pane chains on the IR;
• it declares summary families by hand;
• it finds what to store with frontier_from_timing and walks the DAG for each pane's window;
• it encodes Remote Write samples with promql_rows::series_row;
• it rebuilds its correctness reference by matching candidate ids (L<n> to P<n>);
• it can only set cost-model prices in Rust.

What

• One facade entry point. asap_planner::plan_installable(UserInput) -> InstallablePlan plans the workload as e2e_plan does and returns the runtime's plan file:

  • dag: a PhysicalASAPDAGDocument, the versioned envelope checked by validate. It holds the selected DAG with Stage 3's timings and materialization, pane chains already expanded into the panes the executor runs.
  • queries: each query's text and root_node_id, the DAG root that answers it. A standalone scalar query (e.g. scalar(sum(m))) has no node that answers it, so it is refused with UnsupportedQueryRoot { entry_index }.
  • stored_outputs: each node stored across evaluations, with built_at, a computation_id, its pane and its retention_ms. Every output of one computation_id carries the largest retention among them, since a runtime keeps their panes as one. built_at is either IngestionRun, or QueryRun for a query-time result kept across evaluations: the evaluation that first needs it builds it, then the runtime stores it. The materialization is read from the node itself. Panes that differ only in offset share a computation_id, and it is equal across plans of one wire version. The pane is a RawSampleWindow: TimeRange width, summed TimeShift offset, Scan node. A stored node without a pane is refused (StoredOutputWithoutPane).
  • query_time_raw_windows: the raw-sample windows read at query time, other than stored outputs' panes. A kept query-time result's range is fed only through its build, never twice.
  • selection: selected_candidate_id and cost_per_second, for reporting.

• Refused shapes. A raw-sample window whose position is not a fixed offset from the evaluation time is refused with UnsupportedRawSampleWindow { time_range_node_id, reason }. That covers a PromqlSubquery, TimeShift or TimeRange above the range, an @ anchor below it, a non-TimeShift operator between the range and its Scan, or a negative total offset below it. A negative offset gives a window that ends after the evaluation, which no ingestion run can build in time. Such windows are no longer listed with a wrong window or silently dropped. I chose refusal over computing windows from ancestors: it is the simpler correct option, and those shapes can be supported later.

• Pane-chain expansion is internal, and panes are memoized by (chain, index). Two queries on one chain (p99 and p50 over the same hour) now store 6 panes, not 12. The stage export (PlanOutput::physical_asap_dag, stage_pipeline, the viewer) keeps the chains for Stage 2/3 costing and display.

• Supported summaries default to the executor's. When DeploymentCapabilities.supported_summaries is None, plan_installable offers only what the Planner's executor runs per instance (asap_executor::supported_summary_families(), cross-checked against the factory and the evaluations). Count-Min lists Total only. Its item frequency comes only from a Hydra grid, which Stage 2 does not produce yet; SupportedSummary gains a grouping dimension when it does. Raw-data and query-time-result retention stay deployment inputs.

• Samples in, labels out. asap_executor::RawSampleSelector::new(&dag, &window) prepares one raw-sample window once:

  • .metric() says which metric's samples to route to it.
  • .rows(window_end_ms, samples) keeps the samples in (window_end_ms - width_ms, window_end_ms] that the window's Scan predicates (its label matchers) keep. For an instant selector, it keeps only each series' latest sample in the lookback. Rows come back in the TimeRange's schema.

  series_labels_of_output_row reads an answer's labels, and Sample is the input. promql_rows is a pub(crate) module, and frontier_from_timing is compiled only under cfg(any(test, feature = "test-support")). The test-support feature makes both public for the Planner's own executor and integration tests.

• Correctness reference. InstallablePlan::query_time_reference() validates first, then returns the same summaries and parameters with nothing materialized. It is for a runtime's tests only.

• Building each pane once. StoredOutput.computation_id exists for this: a runtime builds each distinct (computation_id, pane.end_offset_ms) once. For example, [60m] and [60m] offset 10m store 12 panes of one computation, and 7 are built.

• One dependency pair. asap_planner re-exports what a runtime needs to build UserInput and its deployment inputs, and to read the plan. That includes SketchAlgorithm, StatisticKind, SketchParams, SummaryOperation and HistogramKind. A runtime depends on asap_planner + asap_executor only. A new CI step, cargo check -p asap-planner --tests --locked, enforces this: the runtime test (crates/planner/tests/runtime_install.rs) builds there with the executor linked without test-support.

• Cost-model settings load from JSON: PlanSelectionSettings and the types under it, plus stage_pipeline --plan-selection-settings <file>, which cannot be combined with --cost-model, the --cluster-* flags or --measured-operation-costs.

• Pane retention. Each stored pane carries retention_ms: how far back from the newest evaluation the runtime keeps it. The value is Stage 3's Selection::pane_retention_ms for its chain (end offset + N·w + range span), or how far the pane reaches back when Stage 3 records none. A separate Stage 3 commit fixes the offset part: retention and the kept-memory price used to ignore a chain's input offset, so [60m] offset 10m was retained for 3,600,000 ms instead of 4,200,000, and its kept memory was priced one pane short. No pinned number changes for workloads without offsets.

• Doc fix (Runtime handoff: gaps in consuming Stage 3's selected plan outside the Planner #749 item 1). PlanOutput::physical_asap_dag already keeps Stage 3's timings; its doc said otherwise. The doc is corrected and the behavior pinned by a test.

Names

• plan_installable / InstallablePlan: beside e2e_plan / PlanOutput (every query's selected IR plus the stage record), the name says what this one returns: the plan a runtime installs.
• queries: each entry holds the query text and its root, so it is not called roots.
• RawSampleWindow { width_ms, end_offset_ms }: a window of raw samples, as opposed to a node input. RawSampleSelector prepares one for rows. query_time_reference(): the plan a runtime's correctness test compares against.

Changes since review of 3246976

Finding	Change
F1 matchers dropped	input_rows_from_samples takes (dag, &RawSampleWindow, evaluation_time_ms, samples) and runs the plan's Scan (replaced by RawSampleSelector in round 2); matcher e2e case added
F2 duplicated panes	panes memoized by (chain, index) in expand; types and integration regression tests
F3 wrong windows	refused with UnsupportedRawSampleWindow (subquery, TimeShift above, @ below); a plain offset 5m range is listed correctly
F4 optional pane	StoredOutput.pane: RawSampleWindow, StoredOutputWithoutPane { node_id }; RawSampleInput renamed RawSampleWindow and moved to asap-types, since the executor takes it
F5 Count-Min	ItemFrequency dropped; the cross-check is per-instance
F6 version	dag: PhysicalASAPDAGDocument, checked in validate
F7 panic	with_every_node_at_query_time validates first and returns InstallablePlanError
F8 not enforced	runtime e2e test moved to crates/planner/tests/runtime_install.rs (executor without test-support), no internal imports
F9/F10 report and unread fields	SelectedPlanReport { selected_candidate_id, cost_per_second: Option<f64> }; removed resource_usage, accepted_accuracy_per_query, StoredOutput.materialization, RawSampleInput.scan_source (the Scan node holds it); the ResourceUsage/QueryLatency serde derives they needed are reverted
F11 body	visibility claim fixed; Usage below is the test's code, run at this head
re-exports, rename	asap_planner re-exports; InstallablePlanError::Export renamed Timing

Changes since review of c021915

Item	Change
1 instant selectors	RawSampleSelector::rows keeps each series' latest sample in an instant lookback; sum(latency_ms) with a@T-900=1, a@T-100=10, b@T-200=100 now gives 110 (111 before, test written first)
2 selector	input_rows_from_samples is replaced by RawSampleSelector::new(&dag, &window)? with .metric() and .rows(window_end_ms, samples); the Scan predicates are compiled once per window; ingestion passes the pane end. No label-matcher accessor: matchers stay compiled predicates (exposing them as data would mean decoding the predicate IR)
3 F7 test	adds nodes.remove(0) → InvalidDag; without the guards it panicked (confirmed)
4 enforcement	CI format-and-lint runs cargo check -p asap-planner --tests --locked; importing frontier_from_timing there fails to compile (confirmed); body claims fixed
5 doc	PlanError::Installable reworded
6 computation_id	Usage builds each (computation_id, end_offset_ms) once; test: [60m] + [60m] offset 10m → 12 stored outputs, 1 computation_id, 7 builds, answers equal the reference
7 tests	unit test of the TimeShift-above-the-range refusal on a hand-built DAG; executor unit tests of the selector (window bound edges, instant latest rule, metric routing and matchers, a table Scan refused)
8 re-exports	SketchAlgorithm, StatisticKind, SketchParams, SummaryOperation, HistogramKind
9 renames	width_ms / end_offset_ms; query_time_reference()
10 efficiency	validate compares without cloning; of_dag keeps validation as a debug_assert; raw-sample reachability and "re-timed from above" are computed once per DAG in O(nodes + edges) passes (bottom-up and top-down over one postorder)

Changes since review of bf1bb85

Item	Change
F1 window position	sample values now name their pane (1000 × pane + base), and the e2e tests assert which pane each p99 comes from (5 for the hour, 6 for the offset hour, 0 for the 5-minute max); changing bounds_ms to ignore end_offset_ms fails 3 of them (confirmed). Unit test: width 600,000, offset 300,000, T 3,600,000 → (2,700,000, 3,300,000)
F2 scalar roots	refused with UnsupportedQueryRoot { entry_index }; InstalledQueryRoot is removed and each query carries root_node_id
F3 kept query-time results	StoredOutput.built_at: StoredOutputBuild { IngestionRun, QueryRun }. A stored output's pane range is not listed in query_time_raw_windows; dropping that filter fails the new test (confirmed). The Usage branches on built_at. Test: a query_time_results_retained dashboard panel stores 5 query-run panes, builds each once, and equals the reference
F4 body	planner test count; --plan-selection-settings cannot be combined with the other flags
E1	RawSampleSelector::new compiles the Scan's predicates straight from its payload (no DAG rebuild); restore is private again
E2	not done: see Limitations
#759 follow-up	StoredOutput.retention_ms, from Stage 3; Stage 3's retention and price include the chain's offset (own commit, test first)
rebase	onto 5a0318d1 (#759); one conflict, both PRs appending a stage_pipeline test

Changes since review of f7f4254

Item	Change
1 shared retention	every output of one computation_id carries the largest retention_ms among them; the offset-twin test asserts all 12 carry 4,200,000 (failed first: half carried 3,600,000)
2 negative offsets	UnsupportedRawSampleWindow { reason: "a negative offset below the range" }; max_over_time(latency_ms[60m] offset -10m) added to the refused-shapes test (failed first: it planned)
3 test comments, reuse	stale comments in runtime_install.rs fixed. The data plane keeps a pane store across evaluations, keyed by (computation, pane end), and evicts by retention_ms. The kept-results test evaluates again at T + 60 s and builds 1 pane, reusing 4
4 body	runtime_handoff has 8 tests
5 names	expanded and ExpandedWithRetention.retention_ms_by_node

How

• plan_installable:
  1. Copy the caller's capabilities, filling supported_summaries from the executor when it is None.
  2. Run e2e_plan.
  3. Time the output roots with the default assignment, which keeps Stage 3's timings (apply_materialization_timings_to_roots).
  4. Expand pane chains with expand_pane_chains_in_workload, one memo for nodes and panes, and export.
  5. Derive queries, stored_outputs and query_time_raw_windows from the DAG and validate.
• Raw-sample windows: every TimeRange that reaches a Scan without another TimeRange is checked. Only offset TimeShifts may sit between it and the Scan, and no TimeShift/TimeRange/PromqlSubquery may sit above it. Query-time ones are listed; an ingestion-time one is fed through its stored output's pane, found by following single inputs down to the TimeRange.
• computation_id: SHA-256 over each node's payload (child ids zeroed) and output schema, then its inputs' hashes; an offset TimeShift takes its input's hash. This is the backend's DefinitionId scheme.
• RawSampleSelector::new checks the window is a time-series Scan under a TimeRange with the same columns, and compiles the Scan's predicates over its schema (CompiledExpression). rows then:
  1. Filters by window and metric.
  2. For TimeRangeKind::Instant, keeps each label set's latest sample.
  3. Encodes the rows and applies the compiled predicates.
• IndexedDag derives two sets in one pass each, from one children-first postorder: which nodes read raw samples (bottom-up), and which have a TimeShift/TimeRange/PromqlSubquery above them (top-down).
• Visibility uses #[cfg(feature = "test-support")] pub mod promql_rows / pub(crate) mod, and frontier_from_timing under cfg(any(test, feature = "test-support")). This avoids rewriting the 11 Planner test files that use them.

Key code interfaces added / modified

crates/planner/src/installable.rs (new; re-exported from asap_planner)

/// Plan `input` for a deployment runtime as e2e_plan does, returning the single selected DAG ready to
/// install. Unrestricted supported summaries default to the executor's families.
pub async fn plan_installable(input: UserInput<'_>) -> Result<InstallablePlan, PlanError>; // new

/// The selected plan, ready to install and execute: what a runtime writes as its plan file. New type.
pub struct InstallablePlan {
    /// The selected DAG in its versioned envelope: Stage 3's timings and materialization, chains expanded.
    pub dag: PhysicalASAPDAGDocument, // new
    /// One per workload query, in entries() order: its text and the root that answers it.
    pub queries: Vec<InstalledQuery>, // new
    /// Every node stored across evaluations, in node order, with the pane it is built from.
    pub stored_outputs: Vec<StoredOutput>, // new
    /// Every raw-sample window read at query time, in node order; fed when a query evaluates.
    pub query_time_raw_windows: Vec<RawSampleWindow>, // new
    /// Stage 3's record of the selection, for reporting; None for the query-time reference.
    pub selection: Option<SelectedPlanReport>, // new
}
impl InstallablePlan {
    /// Checks a plan file: wire version, DAG validity, and queries/stored outputs/windows match the DAG.
    pub fn validate(&self) -> Result<(), InstallablePlanError>; // new
    /// For correctness tests only: validates, then this plan with nothing materialized, all at query time.
    pub fn query_time_reference(&self) -> Result<InstallablePlan, InstallablePlanError>; // new
}

/// One workload query and the root that answers it. New type.
pub struct InstalledQuery {
    /// The query text, as the workload gave it.
    pub query: String, // new
    /// The node whose output answers it, one of dag.dag.roots.
    pub root_node_id: PhysicalASAPNodeId, // new
}

/// One node whose output the runtime stores, and the pane it is built from. New type.
pub struct StoredOutput {
    /// The node whose output is stored.
    pub node_id: PhysicalASAPNodeId, // new
    /// Which run builds it: an ingestion run, or the query run that first needs it.
    pub built_at: StoredOutputBuild, // new
    /// Hex SHA-256 of what it computes; shared by outputs differing only in pane offset.
    pub computation_id: String, // new
    /// The pane of raw samples it is built from.
    pub pane: RawSampleWindow, // new
    /// How far back from the newest evaluation the pane is kept: Stage 3's pane_retention_ms of its
    /// chain, or pane.reach_ms() when Stage 3 records none; the largest among the outputs sharing
    /// its computation_id, on each of them.
    pub retention_ms: u64, // new
}

/// Which run builds a stored output. New type.
pub enum StoredOutputBuild {
    /// MatAtIngestion: an ingestion run over the pane's window, once its samples have arrived.
    IngestionRun, // new
    /// MatAtQueryKeptAcrossEvaluations: the query run that first needs the pane builds it from the
    /// pane's samples; the runtime stores it and binds it later like a stored pane.
    QueryRun, // new
}

/// Stage 3's record of the selected plan, for reporting. New type.
pub struct SelectedPlanReport {
    /// The selected physical candidate's id, e.g. "P3-m1".
    pub selected_candidate_id: String, // new
    /// Its price per second; None under cost model 3 or when Stage 3 recorded no cost.
    pub cost_per_second: Option<f64>, // new
}

/// Why an installable plan could not be built or is inconsistent. New type.
pub enum InstallablePlanError {
    /// Internal bug: the selected plan's timings do not hold or it does not export
    /// (for the reference: bad input, the plan cannot run at query time).
    Timing(ExecutionDataStateError), // new
    /// Internal bug: a pane chain of the selected plan does not expand.
    PaneChainExpansion(SchemaDerivationError), // new
    /// Bad input (a plan file): the DAG is invalid or of another wire version.
    InvalidDag(PhysicalASAPDAGValidationError), // new
    /// Bad input (a plan file): queries do not bind each query, in order, to the roots.
    QueriesDoNotMatchRoots { roots: usize }, // new
    /// Unsatisfiable request: query entry_index is a standalone scalar query, which no node answers.
    UnsupportedQueryRoot { entry_index: usize }, // new
    /// Bad input (a plan file): a stored output is kept for less than its pane reaches back.
    StoredOutputRetentionTooShort { node_id: PhysicalASAPNodeId }, // new
    /// Internal bug: Stage 3 records a pane retention for a node that is not a selected pane chain.
    PaneRetentionOfUnknownChain { node_id: PhysicalASAPNodeId }, // new
    /// Bad input (a plan file): stored_outputs or query_time_raw_windows do not match the DAG.
    WindowsOrStoredOutputsDoNotMatchDag, // new
    /// Unsatisfiable request: a stored node is not built from one pane of raw samples.
    StoredOutputWithoutPane { node_id: PhysicalASAPNodeId }, // new
    /// Unsatisfiable request: a raw-sample window is not a fixed offset from the evaluation time,
    /// or ends after it (a negative offset below the range).
    UnsupportedRawSampleWindow { time_range_node_id: PhysicalASAPNodeId, reason: &'static str }, // new
}

crates/planner/src/lib.rs

/// Why e2e_plan or plan_installable failed. The type stays; it gains one variant.
pub enum PlanError {
    /// Bad input: the workload or frontend input is invalid.
    Input(UserInputError), // unchanged
    /// Bad input: a query does not lower; entry_index None when the frontend fails the batch.
    Lowering { entry_index: Option<usize>, source: LoweringError }, // unchanged
    /// Bad input: the lowered workload is inconsistent.
    Parsed(ParsedWorkloadError), // unchanged
    /// The optimization pass failed.
    Optimize(OptimizeError), // unchanged
    /// The selected plan cannot be installed: an unsupported raw-sample window or stored output
    /// (unsatisfiable request), or an internal bug; see InstallablePlanError.
    Installable(InstallablePlanError), // new
}
// new re-exports, so a runtime depends on asap_planner + asap_executor alone:
pub use asap_frontend_promql::{HistogramCatalog, HistogramKind};
pub use asap_frontend_sql::SqlCatalog;
pub use asap_logical_asap_optimization::{DeploymentCapabilities, SupportedSummary};
pub use asap_plan_selection::{ClusterLimits, ClusterObjective, CostModel, MeasuredOperationCost,
    PlanSelectionSettings, ResourceSpeed, ResourceUnitCost, SummaryOperation};
pub use asap_types::ir::schema::{SketchAlgorithm, SketchParams, StatisticKind};
pub use asap_types::ir::physical_export::{PhysicalASAPDAG, PhysicalASAPDAGDocument, PhysicalASAPNodeId, RawSampleWindow};
pub use asap_types::workload::PlanningWorkload;

crates/types/src/ir/physical_export.rs

/// A window of raw samples a runtime feeds: at evaluation time T, the samples in
/// (T - end_offset_ms - width_ms, T - end_offset_ms] the Scan selects,
/// bound to the TimeRange. Shape: TimeRange(width) <- TimeShift(offset)* <- Scan; for an instant
/// selector width_ms is its lookback. New type.
pub struct RawSampleWindow {
    /// The TimeRange node the window's input rows are bound to.
    pub time_range_node_id: PhysicalASAPNodeId, // new
    /// Width of the window, from the TimeRange (an instant selector's lookback).
    pub width_ms: i64, // new
    /// How long before T the window ends: the summed offset TimeShifts down to the Scan.
    pub end_offset_ms: i64, // new
    /// The Scan below, whose source and label matchers select the samples.
    pub scan_node_id: PhysicalASAPNodeId, // new
}
impl RawSampleWindow {
    /// (start, end] of the window at evaluation time `evaluation_time_ms`.
    pub fn bounds_ms(&self, evaluation_time_ms: i64) -> (i64, i64); // new
    /// How far back from the evaluation time the window reaches: end_offset_ms + width_ms.
    pub fn reach_ms(&self) -> u64; // new
}
impl PhysicalASAPDAG {
    /// This DAG's plan with nothing materialized: same operators, summaries and parameters, all at query time.
    pub fn with_every_node_at_query_time(&self) -> Result<PhysicalASAPDAG, ReferencePlanError>; // new
}
/// Why with_every_node_at_query_time failed. New type.
pub enum ReferencePlanError {
    /// Bad input: the DAG is not valid.
    InvalidDag(PhysicalASAPDAGValidationError), // new
    /// Bad input: the plan cannot run at query time.
    Timing(ExecutionDataStateError), // new
}

crates/types/src/ir/properties/timing.rs, crates/types/src/ir/operator/pane_chain.rs

/// apply_materialization_timings over a workload's roots with one memo; scalar roots time what they read.
pub fn apply_materialization_timings_to_roots(roots: &[QueryRoot], assignment: &MaterializationAssignment)
    -> Result<Vec<QueryRoot>, ExecutionDataStateError>; // new (was plan-selection's private time_query_root)
/// expand_pane_chains over a workload's roots with one memo for nodes and for panes (by chain and
/// index), so shared nodes and the panes of a shared chain stay one; also returns each chain's panes.
pub fn expand_pane_chains_in_workload(roots: &[QueryRoot]) -> Result<PaneChainExpansion, SchemaDerivationError>; // new
/// A workload with its pane chains expanded. New type.
pub struct PaneChainExpansion {
    /// The workload's roots, in order, with every chain expanded.
    pub roots: Vec<QueryRoot>, // new
    /// Each expanded chain with its panes, in the order first reached.
    pub chains: Vec<ExpandedPaneChain>, // new
}
/// One expanded chain and the panes that replace it. New type.
pub struct ExpandedPaneChain {
    /// The chain node, as the input workload holds it.
    pub chain: Rc<OperatorNode>, // new
    /// Its panes in the expanded workload, newest first.
    pub panes: Vec<Rc<OperatorNode>>, // new
}
/// One root with its chains expanded; now also shares a chain's panes between the merges that read it.
pub fn expand_pane_chains(root: &Rc<OperatorNode>) -> Result<Rc<OperatorNode>, SchemaDerivationError>; // unchanged signature

crates/types/src/ir/schema/family_capability.rs (also re-exported at its old path from asap_logical_asap_optimization)

/// One sketch family an executor builds with the readouts it answers. The type stays; moved here from logical-asap-optimization.
pub struct SupportedSummary {
    /// The family the executor can build.
    pub family: SketchAlgorithm, // unchanged
    /// The statistics it answers from one state.
    pub readouts: Vec<StatisticKind>, // unchanged
}

crates/executor/src/samples.rs (new; re-exported from asap_executor)

/// One raw sample, e.g. one Prometheus Remote Write sample. New type.
pub struct Sample {
    /// Every label of the series, __name__ included.
    pub labels: BTreeMap<String, String>, // new
    /// When it was taken, in Unix milliseconds.
    pub timestamp_ms: i64, // new
    /// The sampled value.
    pub value: f64, // new
}
/// One raw-sample window of an installable DAG, prepared once to turn samples into its input rows. New type.
pub struct RawSampleSelector { /* private: metric, schemas, compiled Scan predicates, width, instant flag */ }
impl RawSampleSelector {
    /// The selector of \`window\`; bad input unless it is a time-series Scan under a TimeRange of \`dag\`.
    pub fn new(dag: &PhysicalASAPDAG, window: &RawSampleWindow) -> Result<Self, Error>; // new; compiles the Scan's predicates from its payload
    /// The metric whose samples the window reads; a runtime routes only those to rows().
    pub fn metric(&self) -> &str; // new
    /// The window's rows for the window ending at window_end_ms: samples in (end - width, end] the Scan keeps,
    /// only each series' latest for an instant selector, in the TimeRange's schema.
    pub fn rows(&self, window_end_ms: i64, samples: &[Sample]) -> Result<Batch, Error>; // new
}
/// The series labels of a PromQL answer row: its decoded series identity plus non-empty label columns.
pub fn series_labels_of_output_row(schema: &Schema, row: &[Value]) -> Result<BTreeMap<String, String>, Error>; // new

crates/executor/src/supported_summaries.rs (new; re-exported from asap_executor)

/// Every sketch family the executor builds per instance, with its readouts: UnivMon [Total, Distinct,
/// FrequencyL2, FrequencyEntropy], Kll [Quantile], Cms [Total], Hll [Total, Distinct], DDSketch [Total,
/// Quantile], CmsWithHeap [TopK], Kmv [Total, Distinct], CountSketchWithHeap [TopK].
/// plan_installable's default for unrestricted supported summaries.
pub fn supported_summary_families() -> Vec<SupportedSummary>; // new

crates/executor/src/physical_planner/mod.rs, candidates.rs, Cargo.toml

/// The executor's PromQL row encoding (series_row, series identity, test helpers).
pub(crate) mod promql_rows; // now private; pub with feature "test-support"
/// Ingestion-time nodes read at query time.
#[cfg(any(test, feature = "test-support"))]
pub fn frontier_from_timing(dag: &PhysicalASAPDAG) -> Result<Vec<NodeId>, Error>; // now only under cfg(any(test, feature = "test-support")); re-exported with "test-support"

[features]
test-support = []   # new: internals the Planner's own tests use

.github/workflows/rust.yml

# new step in format-and-lint: the runtime tests against the public APIs alone
- name: Check the runtime tests against the public APIs alone
  run: cargo check -p asap-planner --tests --locked

crates/plan-selection/src/lib.rs

/// Stage 3's deployment settings. The type stays; now (de)serializes, #[serde(default, deny_unknown_fields)].
pub struct PlanSelectionSettings {
    /// How candidates are compared; default cost model 1 with illustrative unit costs.
    pub cost_model: CostModel, // unchanged
    /// Time per operation and scanned byte; default ResourceSpeed::illustrative().
    pub resource_speed: ResourceSpeed, // unchanged
    /// Seconds non-repeating queries are spread over; default 3,600.
    pub non_repeating_query_horizon_seconds: f64, // unchanged
    /// Measured CPU time per sketch operation; default empty.
    pub measured_operation_costs: Vec<MeasuredOperationCost>, // unchanged
}
/// How Stage 3 compares candidates. The type stays; serializes snake_case: {"usage_over_time": {...}}.
pub enum CostModel {
    /// Cost model 1: unit cost × usage per second.
    UsageOverTime(ResourceUnitCost), // unchanged
    /// Cost model 2: unit cost × peak memory, cores, disk.
    ProvisionedCapacity(ResourceUnitCost), // unchanged
    /// Cost model 3: hard cluster limits and an objective.
    ClusterCapacity(ClusterLimits), // unchanged
}
/// Unit prices per second. The type stays; now (de)serializes, all fields required, unknown refused.
pub struct ResourceUnitCost {
    /// Price of one CPU operation.
    pub cost_per_cpu_op: f64, // unchanged
    /// Price of scanning one byte.
    pub cost_per_scanned_byte: f64, // unchanged
    /// Price of holding one byte of memory for one second (cost model 1).
    pub cost_per_memory_byte_second: f64, // unchanged
    /// Price of provisioning one byte of memory for one second (cost model 2).
    pub cost_per_provisioned_memory_byte_second: f64, // unchanged
    /// Price of provisioning one CPU core for one second (cost model 2).
    pub cost_per_provisioned_cpu_core_second: f64, // unchanged
    /// Price of keeping one byte on disk for one second.
    pub cost_per_disk_byte_second: f64, // unchanged
}
/// How fast one core works. The type stays; now (de)serializes.
pub struct ResourceSpeed {
    /// Milliseconds per CPU operation on one core.
    pub ms_per_cpu_op: f64, // unchanged
    /// Milliseconds per scanned byte on one core.
    pub ms_per_scanned_byte: f64, // unchanged
}
/// A fixed cluster's capacity (cost model 3). The type stays; now (de)serializes.
pub struct ClusterLimits {
    /// Memory available.
    pub memory_bytes: u64, // unchanged
    /// Disk available.
    pub disk_bytes: u64, // unchanged
    /// CPU cores available; None (or omitted) leaves cores unchecked.
    pub cpu_cores: Option<u32>, // unchanged
    /// How to choose among the candidates that fit.
    pub objective: ClusterObjective, // unchanged
}
/// What cost model 3 optimizes. The type stays; serializes snake_case.
pub enum ClusterObjective {
    /// "lowest_latency".
    LowestLatency, // unchanged
    /// "least_memory".
    LeastMemory, // unchanged
    /// "pareto_front".
    ParetoFront, // unchanged
}

crates/plan-selection/src/lib.rs (Stage 3 retention fix)

/// How Stage 3 chose the plan. The type stays; one field's value changes.
pub struct Selection {
    /// The selected physical candidate's id.
    pub selected: String, // unchanged
    /// The selected candidate's query roots, with confirmed weight proofs: the plan to run.
    pub plan: Vec<QueryRoot>, // unchanged
    /// Each valid candidate's cost.
    pub costs: BTreeMap<String, CandidateCost>, // unchanged
    /// For each valid candidate, per query, the accuracy Stage 3 accepted.
    pub accepted: BTreeMap<String, Vec<AcceptedAccuracy>>, // unchanged
    /// Every candidate not selected, with its cause.
    pub rejected: Vec<Rejection>, // unchanged
    /// Under cost model 3's pareto-front objective, the front; empty otherwise.
    pub front: Vec<String>, // unchanged
    /// For each pane chain kept beyond one evaluation, by node id in its physical DAG: how far back
    /// the runtime keeps its completed panes, end offset + N·w + longest reader span (was N·w + span).
    pub pane_retention_ms: BTreeMap<PhysicalASAPNodeId, u64>, // unchanged type; now includes the chain's input offset
}

crates/plan-selection/src/pass/mod.rs: PlanOutput::physical_asap_dag is unchanged; only its doc changes (each node keeps the timing its pass gave it).

crates/devtools/src/bin/stage_pipeline.rs

--plan-selection-settings <file>   # new: PlanSelectionSettings JSON; cannot be combined with --cost-model, --cluster-*, --measured-operation-costs

Usage

This is the code of crates/planner/tests/runtime_install.rs, run at this head. It imports only asap_planner and asap_executor. CI checks it with cargo check -p asap-planner --tests --locked, where the executor links without test-support, so the compiler enforces that no internal is used.

The workload is #509 Example 4b (quantile_over_time(0.99, latency_ms[60m]) every 10 minutes, rank error 0.01 at δ 0.01, 1,000 ingesting series sampled every second), plus its offset 10m twin. The samples are series a and b, every 10 s over the 70 minutes before T = 3,600,000 ms, plus one sample of a metric the plan does not read. Each value is 1000 × pane + base + 0..=100: pane 0 is the newest 10 minutes, and base is 100 for a, 300 for b. A wrong window therefore gives a visibly wrong answer.

Control plane: the workload and the deployment inputs, one plan_installable call, plan.json.

async fn plan_json(queries: &[&str]) -> String {
    // What the deployment keeps: no raw samples, no query-time results
    // across evaluations. Supported summaries are left unrestricted, so the
    // Planner offers only what its own executor runs.
    let capabilities = DeploymentCapabilities {
        raw_data_retained: false,
        ..DeploymentCapabilities::default()
    };
    plan_json_for(workload(queries), &capabilities).await
}

async fn plan_json_for(
    workload: PlanningWorkload,
    capabilities: &DeploymentCapabilities,
) -> String {
    // The deployment's Stage 3 prices, from its JSON file: cost model 1.
    let settings: PlanSelectionSettings = serde_json::from_str(
        r#"{"cost_model": {"usage_over_time": {
              "cost_per_cpu_op": 1e-6, "cost_per_scanned_byte": 1e-7,
              "cost_per_memory_byte_second": 1.25e-7,
              "cost_per_provisioned_memory_byte_second": 1.25e-7,
              "cost_per_provisioned_cpu_core_second": 1000.0,
              "cost_per_disk_byte_second": 0.0}}}"#,
    )
    .unwrap();
    // The models Stages 1-3 read: built-in accuracy, the deployment inputs.
    let models = PlanningModels::builtin()
        .with_deployment_capabilities(capabilities)
        .with_plan_selection(&settings);
    // The one Planner call: the selected plan, ready to install.
    let plan = plan_installable(UserInput::new(
        &workload,
        FrontendInput::Promql { histograms: None },
        models,
    ))
    .await
    .unwrap();
    // `plan.json`, as the control plane writes it.
    serde_json::to_string(&plan).unwrap()
}

Data plane, at one evaluation: from plan.json, the samples, and the panes earlier evaluations left in the store. Each distinct (computation_id, pane end) the store lacks is built once, in the run its built_at names. Then the queries run, panes past their retention_ms are dropped, and the query-time reference runs.

struct DataPlaneRun {
    /// Per query, in workload order, the installed plan's answers.
    answers_per_query: Vec<Vec<Answer>>,
    /// Per query, the query-time reference's answers.
    reference_answers_per_query: Vec<Vec<Answer>>,
    /// How many panes this evaluation built (ingestion or query runs);
    /// the others it read came from the store.
    panes_built: usize,
}

/// One pane the data plane stores: its built state, and how long to keep it.
struct StoredPane {
    /// The built state, bound at each stored output of its computation.
    state: Batch,
    /// When the pane's window starts, in Unix milliseconds.
    start_ms: i64,
    /// How far back from the newest evaluation the pane is kept.
    retention_ms: u64,
}

/// The data plane's pane store, by computation and pane end.
type PaneStore = BTreeMap<(String, i64), StoredPane>;

fn query_roots(plan: &InstallablePlan) -> Vec<u64> {
    plan.queries
        .iter()
        .map(|query| query.root_node_id as u64)
        .collect()
}

fn run_data_plane(
    plan_json: &str,
    samples: &[Sample],
    evaluation_ms: i64,
    store: &mut PaneStore,
) -> DataPlaneRun {
    // The installed plan, read back from its file and checked.
    let installed: InstallablePlan = serde_json::from_str(plan_json).unwrap();
    installed.validate().unwrap();
    // The DAG the executor runs.
    let dag = &installed.dag.dag;
    // Every input of the query run, keyed by the node it is bound to.
    let mut inputs = BTreeMap::new();
    // How many panes this evaluation builds.
    let mut panes_built = 0;
    for output in &installed.stored_outputs {
        // The pane's window at this evaluation, `(start_ms, end_ms]`.
        let (start_ms, end_ms) = output.pane.bounds_ms(evaluation_ms);
        // The pane's identity: its computation and where it ends.
        let pane_key = (output.computation_id.clone(), end_ms);
        if !store.contains_key(&pane_key) {
            // Prepared once per window: routes by metric, applies the Scan.
            let selector = RawSampleSelector::new(dag, &output.pane).unwrap();
            // Only the samples of the selector's metric reach it.
            let routed: Vec<Sample> = samples
                .iter()
                .filter(|s| s.labels["__name__"] == selector.metric())
                .cloned()
                .collect();
            // The pane's input rows: the samples its Scan selects in its window.
            let rows = selector.rows(end_ms, &routed).unwrap();
            // The run that builds the pane: an ingestion run over its
            // window, or, for a query-time result kept across evaluations,
            // the query run of this evaluation.
            let scope = match output.built_at {
                StoredOutputBuild::IngestionRun => Scope::Ingestion {
                    window_start_ms: start_ms,
                    window_end_ms: end_ms,
                    revision: 1,
                },
                StoredOutputBuild::QueryRun => Scope::Query {
                    evaluation_time_ms: evaluation_ms,
                    revision: 1,
                },
            };
            // The rows bound to the pane's TimeRange.
            let pane_input = BTreeMap::from([(output.pane.time_range_node_id as u64, rows)]);
            // The batches that run built.
            let built = run(dag, pane_input, &[output.node_id as u64], scope).remove(0);
            // The pane's state, as the store keeps it, in one batch.
            let schema = built[0].schema().clone();
            let state_rows = built.iter().flat_map(|b| b.rows().to_vec()).collect();
            let state = Batch::try_new(schema, state_rows).unwrap();
            store.insert(
                pane_key.clone(),
                StoredPane {
                    state,
                    start_ms,
                    retention_ms: output.retention_ms,
                },
            );
            panes_built += 1;
        }
        inputs.insert(output.node_id as u64, store[&pane_key].state.clone());
    }
    // Raw samples read at query time, other than stored outputs' panes.
    for window in &installed.query_time_raw_windows {
        // The window's selector and its end at this evaluation.
        let selector = RawSampleSelector::new(dag, window).unwrap();
        let (_, end_ms) = window.bounds_ms(evaluation_ms);
        inputs.insert(
            window.time_range_node_id as u64,
            selector.rows(end_ms, samples).unwrap(),
        );
    }
    // The query evaluation, which merges the stored panes.
    let query = Scope::Query {
        evaluation_time_ms: evaluation_ms,
        revision: 1,
    };
    // Each query's answers: (series labels, p99) per series.
    let answers_per_query = run(dag, inputs, &query_roots(&installed), query.clone())
        .iter()
        .map(|batches| answers(batches))
        .collect();
    // The store drops each pane whose window starts more than its
    // retention before this evaluation.
    store.retain(|_, pane| pane.start_ms >= evaluation_ms - pane.retention_ms as i64);

    // The correctness reference: nothing stored, every window fed raw
    // samples at query time.
    let reference = installed.query_time_reference().unwrap();
    // The reference's DAG.
    let reference_dag = &reference.dag.dag;
    // Each query-time window's input rows, keyed by its TimeRange.
    let raw: BTreeMap<u64, Batch> = reference
        .query_time_raw_windows
        .iter()
        .map(|window| {
            // The window's selector and its end at this evaluation.
            let selector = RawSampleSelector::new(reference_dag, window).unwrap();
            let (_, end_ms) = window.bounds_ms(evaluation_ms);
            (
                window.time_range_node_id as u64,
                selector.rows(end_ms, samples).unwrap(),
            )
        })
        .collect();
    // The reference's answers, which the installed plan's must equal.
    let reference_answers_per_query = run(reference_dag, raw, &query_roots(&reference), query)
        .iter()
        .map(|batches| answers(batches))
        .collect();
    DataPlaneRun {
        answers_per_query,
        reference_answers_per_query,
        panes_built,
    }
}

The runtime's compile-and-run with the executor:

fn run(
    dag: &PhysicalASAPDAG,
    inputs: BTreeMap<u64, Batch>,
    ids: &[u64],
    scope: Scope,
) -> Vec<Vec<Batch>> {
    // What each bound input holds: its rows' schema.
    let contracts = inputs
        .iter()
        .map(|(&id, batch)| (id, InputContract::bounded(batch.schema().clone())))
        .collect();
    // The executor's plan for `ids`, cut at the bound inputs.
    let compiled = compile(dag, contracts, ids).expect("the executor compiles the plan");
    // Each bound input as an in-memory source of its rows.
    let sources = inputs
        .into_iter()
        .map(|(id, batch)| {
            let source = ExecutorOperator::source(batch.schema().clone(), vec![batch]).unwrap();
            (id, Box::new(source) as Source<'_>)
        })
        .collect();
    // The plan bound to its sources, ready to run.
    let instance = compiled.instantiate(sources).unwrap();
    block_on(async {
        // The run's scope (ingestion window or query time) and limits.
        let context = RunContext::new(scope, Limits::default()).unwrap();
        // One output stream per id.
        let outputs = instance.execute(compiled.roots(), context).unwrap();
        futures::future::join_all(outputs.into_iter().map(|mut output| async move {
            // Its batches, collected.
            let mut batches = Vec::new();
            while let Some(batch) = output.next().await {
                batches.push((*batch.unwrap()).clone());
            }
            batches
        }))
        .await
    })
}

What the runtime gets back:

• Example 4b builds 6 panes, and each series' p99 comes from the hour's oldest pane, equal to the reference.
• With its offset 10m twin there are 12 stored outputs of one computation, each kept 4,200,000 ms, and 7 builds; the twin's p99 comes from pane 6.
• The kept-results panel builds all 5 panes at T and only 1 at T + 60 s, reusing 4 from the store.

async fn a_runtime_installs_and_runs_the_plan_from_one_facade_call() {
    let plan_json = plan_json(&["quantile_over_time(0.99, latency_ms[60m])"]).await;
    let run = run_data_plane(
        &plan_json,
        &remote_write_samples(),
        EVALUATION_MS,
        &mut PaneStore::new(),
    );
    assert_eq!(run.panes_built, 6, "six ten-minute panes");
    // The p99 of the hour comes from its oldest pane, pane 5.
    assert_answers_from_pane(&run.answers_per_query[0], 5);
    assert_eq!(run.answers_per_query[0][0].0["instance"], "a");
    assert_eq!(run.answers_per_query, run.reference_answers_per_query);
}

async fn panes_of_one_computation_build_once_across_queries() {
    let plan_json = plan_json(&[
        "quantile_over_time(0.99, latency_ms[60m])",
        "quantile_over_time(0.99, latency_ms[60m] offset 10m)",
    ])
    .await;
    let installed: InstallablePlan = serde_json::from_str(&plan_json).unwrap();
    assert_eq!(installed.stored_outputs.len(), 12);
    let computations: BTreeSet<&str> = installed
        .stored_outputs
        .iter()
        .map(|o| o.computation_id.as_str())
        .collect();
    assert_eq!(computations.len(), 1);
    let run = run_data_plane(
        &plan_json,
        &remote_write_samples(),
        EVALUATION_MS,
        &mut PaneStore::new(),
    );
    assert_eq!(run.panes_built, 7);
    assert_eq!(run.answers_per_query.len(), 2);
    // The last hour's p99 comes from pane 5, the offset hour's from pane 6.
    assert_answers_from_pane(&run.answers_per_query[0], 5);
    assert_answers_from_pane(&run.answers_per_query[1], 6);
    assert_eq!(run.answers_per_query, run.reference_answers_per_query);
}

Before this PR

• A runtime called run_stage1_to_stage3 and expanded pane chains itself; the executor rejects an unexpanded chain.
• supported_summaries: None admitted families the executor cannot run.
• What to store came from frontier_from_timing, and each pane's window from walking the DAG (backend sds.rs pane_of / identity).
• Samples were encoded with promql_rows::series_row, and answer labels read with decode_series_identity.
• The reference plan was rebuilt by candidate-id matching and re-sizing.
• Cost-model prices could only be set in Rust.

After this PR

A runtime builds the workload and its deployment inputs, calls plan_installable, writes the InstallablePlan as its plan file, and runs it with RawSampleSelector and the executor's compile/instantiate/execute, with no other Planner call:

• stored_outputs replaces frontier_from_timing plus the SDS DAG walk.
• RawSampleSelector replaces promql_rows and applies the plan's matchers and the instant-selector rule.
• InstallablePlan::query_time_reference replaces the id-matching reference.

Verification

Only the tests covering the change, plus fmt, the new CI check and clippy on the touched crates:

• cargo nextest run -p asap-planner: 29 passed.
  • runtime_install (6 tests):
    • Example 4b: 6 builds, p99s from pane 5, equal to the reference.
    • Label matcher: 1 answer.
    • [60m] + [60m] offset 10m: 12 stored outputs, 1 computation_id, 7 builds, p99s from panes 5 and 6.
    • Tampered plan files: refused with errors.
    • Instant selector: 110.
    • Query-time results kept across evaluations: 5 QueryRun panes, none also listed as query-time windows. 5 builds at T, then 1 build at T + 60 s (4 reused); both evaluations equal the reference.
  • Unit tests: a_time_shift_above_the_range_is_refused, and the default to the executor's families.
• With bounds_ms changed to ignore end_offset_ms, 3 runtime_install tests fail; with the stored-pane filter removed, the kept-results test fails.
• cargo check -p asap-planner --tests --locked (the CI step): passes.
• cargo nextest run -p asap-integration-tests --test runtime_handoff: 8 passed.
  • New: a scalar query is refused (UnsupportedQueryRoot).
  • New: retention under a range evaluation is 7,200,000 ms, and a too-short retention is refused.
  • New: an offset chain keeps its panes 4,200,000 ms, and its unshifted twin's shared panes carry the same value.
  • The refused-shapes test now includes offset -10m.
• --test range_evaluation: passed, including the new Stage 3 test. It failed first: the offset chain reported 3,600,000 ms. It now reports 4,200,000 ms, and the kept memory is priced higher.
• Pinned numbers after the Stage 3 fix: full asap-plan-selection (171), asap-integration-tests (273) and asap-devtools (42) suites passed with no number changed.
• --test tumbling_pane_merge --test materialization_e2e --test frequency_execution: passed.
• cargo nextest run -p asap-types --test physical_export --test pane_chain: 19 passed, including the bounds_ms offset test.
• cargo nextest run -p asap-executor -E 'test(samples::) | test(supported_summary) | test(frontier)': 12 passed.
• cargo fmt --all -- --check: clean.
• cargo clippy on the 7 touched crates with --all-targets --all-features --locked -- -D warnings: clean.
• cargo clippy -p asap-planner --all-targets --locked -- -D warnings (no test-support): clean.
• cargo clippy -p asap-executor --lib --locked -- -D warnings: clean.

Limitations and follow-ups

• Refused shapes (UnsupportedRawSampleWindow): windows under a subquery, under a TimeShift above the range, or with an @ anchor below it. Supporting them means computing the window from the ancestors and carrying an absolute at.
• RawSampleSelector exposes the metric, but not the label matchers as data. Matchers run as the Scan's compiled predicates.
• Standalone scalar queries (scalar(...), a constant) are refused (UnsupportedQueryRoot). Answering them needs a public executor entry for scalar roots.
• RawSampleSelector::rows encodes every in-window sample of its metric before applying the Scan's label matchers. With K windows over one metric that is K encodings per sample in their windows (a pane's samples fall in one window per computation). Matching label-only matchers on Sample.labels first would avoid that. It needs a decoder for the predicate IR's column comparisons, so it is not done here.
• SupportedSummary has no grouping dimension, so Count-Min's Hydra-only item frequency is not listed. It is added when Stage 2 produces Hydra.
• computation_id hashes the serialized DAG payload, so it is stable within one PHYSICAL_ASAP_DAG_WIRE_VERSION. validate refuses other versions.
• promql_rows stays reachable through test-support for the Planner's own tests. Moving those tests onto the public sample API would let it become fully private.
• The backend changes that use plan_installable belong to ASAPQuery-backend.

Closes #749