[zzylol]

Closes #970. Part of #968 (PR 6 of 6).

Why

The plan file a runtime installs kept two copies of most facts: the DAG, and fields derived from it (queries[i].root_node_id, stored_outputs, query_time_raw_windows). InstallablePlan::validate mostly re-derived those fields and compared them. The file also carried Stage 3's selection report, which a runtime never reads. The version sat one level below the file it versioned (plan.dag.schema_version, plan.dag.dag). And a runtime could not tell how long to keep raw data, although Stage 3 prices exactly that.

What

• InstallablePlan is { wire_version, dag, query_texts }. INSTALLABLE_PLAN_WIRE_VERSION = 11 versions the whole file. PhysicalASAPDAGDocument and PHYSICAL_ASAP_DAG_WIRE_VERSION are removed.

• PhysicalASAPDAGNode gains retention_ms, set at installation on two kinds of node:

  • each materialized-view node (MatAtIngestion or MatAtQueryKeptAcrossEvaluations);
  • each query-time Scan, holding how long its source's rows are kept, or None when it reads them unbounded.

  PhysicalASAPDAG::validate rejects it on any other node.

• materialized_views() and raw_data_retention() are derived from the DAG and those node retentions. They live in asap-types (asap_types::ir::materialized_view), so a runtime can call them without the planner crate.

• Renames: StoredOutput → MaterializedView, StoredOutputBuild::{IngestionRun, QueryRun} → MaterializedViewRefresh::{AtIngestion, OnFirstRead}, StoredOutputInputs → MaterializedViewInputs { raw_data_windows, materialized_views }, StoredCoverage → ViewCoverage. RawDataSampleWindow becomes RawDataWindow { source, end_offset_ms, width_ms }. It names the Source the view scans instead of two node ids.

• plan_installable and installable_plan_of return (InstallablePlan, Option<SelectedPlanReport>).

• query_time_reference moves to asap_planner::test_support (feature test-support).

• InstallablePlan::from_json reads wire_version first. An older file, such as a v10 file that has no wire_version, is refused as UnsupportedWireVersion, not as a missing field.

• The DAG viewer document's stored_outputs becomes materialized_views, with the new entry shape.

How

• PhysicalASAPDAG::set_retention(pane_retention_ms_by_node, reader_span_ms_by_node, deadline) sets two kinds of retention on the expanded DAG. installable_plan_of calls it, then compiles each view's refresh with the executor, as before.
  • Views: the computation runtime_inputs_of used to do. That is Stage 3's pane retention, or the reach plus the readers' range span, then propagated to the views each one reads.
  • Query-time scans: scan_window(scan).lookback_ms() plus the readers' longest range-evaluation span, and at least the retention of every view refreshed on first read from that scan. It is None when the window is unbounded.
  • It stops with DeadlineExceeded when the compile check's deadline passes.
• PhysicalASAPDAG::materialized_views() makes the same DAG walk that runtime_inputs_of made: the inputs, the shape check of every raw-data window, the ingestion offsets check and the coverage. It takes retention from the nodes and checks that each view's retention covers its reach.
• PhysicalASAPDAG::raw_data_retention() takes, per Source read by a query-time Scan, the max of those scans' retention_ms, or None when one is unbounded. It checks that each bounded scan has a retention covering its window.
  • In an installed DAG, a chain kept across evaluations is its expanded panes, so the scan keeps the reach of all of them. Reusing a pane refreshed earlier is optional, and a restart or an older range step refreshes older panes from raw data.
  • Stage 3's newest-pane rule applies to the unexpanded DAG.
• InstallablePlan::validate checks only non-derived facts: the wire version, one text per root with no scalar root, and that view and scan retention cover their reach. The DAG's validity is checked inside materialized_views().
• PhysicalASAPDAG::scan_window now indexes the DAG once per call: O(nodes + edges) instead of a linear search per edge. Its result is unchanged.
• The derivation errors (UnsupportedRawDataWindow, ViewReadsUnwindowedSource, ViewWithoutCoverage, IngestionBuildReadsWindowsOfDifferentOffsets, ViewWithoutRetention, ViewRetentionTooShort, ScanWithoutRetention, ScanRetentionTooShort, DeadlineExceeded) move to InstallableDagError in asap-types. InstallablePlanError::Uninstallable wraps them, and Stage 3's compile check still rejects a candidate for any of them.

Key code interfaces added / modified

crates/types/src/ir/materialized_view.rs (new module)

/// One materialized view of a DAG; derived, not in the plan file (new, was planner's `StoredOutput`).
pub struct MaterializedView {
    /// The node whose output is kept. // unchanged
    pub node_id: PhysicalASAPNodeId,
    /// Which run refreshes it. // was `built_at: StoredOutputBuild`
    pub refresh: MaterializedViewRefresh,
    /// The node's summary coverage. // was `StoredCoverage`
    pub coverage: ViewCoverage,
    /// What its refresh reads. // was `StoredOutputInputs`
    pub inputs: MaterializedViewInputs,
    /// The node's `retention_ms`. // unchanged meaning; now read from the node
    pub retention_ms: u64,
}

/// What a view's stored state is (new name, was `StoredCoverage`).
pub struct ViewCoverage {
    /// `SummaryCoverage::definition_id`. // unchanged
    pub definition_id: String,
    /// The rows that went in, time relative to the evaluation. // unchanged
    pub selection: Vec<SelectionBox>,
}

/// What a view's refresh reads (new name, was `StoredOutputInputs`).
pub struct MaterializedViewInputs {
    /// The raw-data windows it scans directly, one per TimeRange, node order. // was `raw_data_sample_windows` (serialized `raw_sample_windows`)
    pub raw_data_windows: Vec<RawDataWindow>,
    /// The views it reads, node order. // was `stored_outputs`
    pub materialized_views: Vec<PhysicalASAPNodeId>,
}

/// A window of a raw-data source a view scans (new, replaces `physical_export::RawDataSampleWindow`, removed).
pub struct RawDataWindow {
    /// The `Scan`'s source. // new, replaces `time_range_node_id` and `scan_node_id` (removed)
    pub source: Source,
    /// Sum of the offset `TimeShift`s below the `TimeRange`. // unchanged
    pub end_offset_ms: i64,
    /// The `TimeRange`'s width. // unchanged
    pub width_ms: i64,
}
impl RawDataWindow {
    /// `end_offset_ms + width_ms`. // was `RawDataSampleWindow::reach_ms`
    pub fn lookback_ms(&self) -> u64;
    /// `(start, end]` at an evaluation time. // unchanged
    pub fn bounds_ms(&self, evaluation_time_ms: i64) -> (i64, i64);
}

/// Which run refreshes a view (new name, was `StoredOutputBuild`).
pub enum MaterializedViewRefresh {
    /// `MatAtIngestion`: an ingestion run once its inputs are complete. // was `IngestionRun` (wire `ingestion_run` → `at_ingestion`)
    AtIngestion,
    /// `MatAtQueryKeptAcrossEvaluations`: the query run that first reads it. // was `QueryRun` (wire `query_run` → `on_first_read`)
    OnFirstRead,
}

/// How far back a runtime keeps one raw-data source for query-time scans (new).
pub struct RawDataRetention {
    /// The source. // new
    pub source: Source,
    /// Longest `retention_ms` of a query-time Scan of it; None = unbounded. // new
    pub lookback_ms: Option<u64>,
}

/// Why a runtime cannot install a DAG (new; variants moved from `InstallablePlanError`).
pub enum InstallableDagError {
    /// Bad input: invalid DAG. // new
    InvalidDag(PhysicalASAPDAGValidationError),
    /// Unsatisfiable request: a raw-data window not at a fixed offset from T. // was `InstallablePlanError::UnsupportedRawDataSampleWindow`
    UnsupportedRawDataWindow { time_range_node_id: PhysicalASAPNodeId, reason: &'static str },
    /// Unsatisfiable request: a view reads a Scan outside a window. // was `StoredOutputReadsUnwindowedSource`
    ViewReadsUnwindowedSource { node_id: PhysicalASAPNodeId },
    /// Unsatisfiable request: a view with no summary coverage. // was `StoredOutputWithoutCoverage`
    ViewWithoutCoverage { node_id: PhysicalASAPNodeId, reason: String },
    /// Unsatisfiable request: an ingestion view over windows of different offsets. // was `InstallablePlanError::IngestionBuildReadsWindowsOfDifferentOffsets`
    IngestionBuildReadsWindowsOfDifferentOffsets { node_id: PhysicalASAPNodeId, end_offsets_ms: Vec<i64> },
    /// Bad input (plan file): a view node without `retention_ms`. // new
    ViewWithoutRetention { node_id: PhysicalASAPNodeId },
    /// Bad input (plan file): retention shorter than the view's reach. // was `StoredOutputRetentionTooShort`
    ViewRetentionTooShort { node_id: PhysicalASAPNodeId },
    /// Bad input (plan file): a query-time Scan with a bounded window and no `retention_ms`. // new
    ScanWithoutRetention { node_id: PhysicalASAPNodeId },
    /// Bad input (plan file): a query-time Scan kept for less than its window. // new
    ScanRetentionTooShort { node_id: PhysicalASAPNodeId },
    /// The caller's deadline passed during the derivation (compile check). // new
    DeadlineExceeded,
}

impl PhysicalASAPDAG {
    /// Every view in dependency order, retention from the nodes; validates. // new (was `runtime_inputs_of` in the planner)
    pub fn materialized_views(&self) -> Result<Vec<MaterializedView>, InstallableDagError>;
    /// Sets `retention_ms` on each view (Stage 3's pane retention, else reach + readers' span, propagated to inputs) and each query-time Scan (window + readers' span, ≥ on-first-read views it feeds). // new (view part was inside `runtime_inputs_of`)
    pub fn set_retention(
        &mut self,
        pane_retention_ms_by_node: &HashMap<PhysicalASAPNodeId, u64>,
        reader_span_ms_by_node: &HashMap<PhysicalASAPNodeId, u64>,
        deadline: Option<Instant>,
    ) -> Result<Vec<MaterializedView>, InstallableDagError>;
    /// Per source read by a query-time Scan, the longest Scan `retention_ms`; checks it covers each window. // new
    pub fn raw_data_retention(&self) -> Result<Vec<RawDataRetention>, InstallableDagError>;
    /// Every raw-data window (TimeRange or PaneChain pane) has a supported shape. // new (was planner `IndexedDag::window_at` checks)
    pub fn check_raw_data_windows(&self) -> Result<(), InstallableDagError>;
}

crates/types/src/ir/physical_export.rs, crates/types/src/physical/materialization.rs

pub const PHYSICAL_ASAP_DAG_WIRE_VERSION: u32 = 10; // removed (see INSTALLABLE_PLAN_WIRE_VERSION)

/// Versioned transport envelope (removed; `InstallablePlan.wire_version` versions the file).
pub struct PhysicalASAPDAGDocument { pub schema_version: u32, pub dag: PhysicalASAPDAG } // removed

/// A raw-sample window (removed; replaced by `materialized_view::RawDataWindow`).
pub struct RawDataSampleWindow { /* time_range_node_id, width_ms, end_offset_ms, scan_node_id */ } // removed

/// One node of the physical ASAP DAG (type stays).
pub struct PhysicalASAPDAGNode {
    /// Node id. // unchanged
    pub id: PhysicalASAPNodeId,
    /// Operator with children as ids. // unchanged
    pub payload: PhysicalASAPOperatorPayload,
    /// Timing and primitive of its output. // unchanged
    pub output_state: ExecutionDataState,
    /// Stage 2's materialization of the sub-DAG it produces. // unchanged
    pub materialization: Materialization,
    /// Output schema. // unchanged
    pub output_schema: Schema,
    /// Exported accuracy guarantee. // unchanged
    pub guarantee: Option<AccuracyGuarantee>,
    /// A view's retention, or a query-time Scan's raw-data retention (None = unbounded); None elsewhere and before installation sets it; omitted from JSON when None. // new
    pub retention_ms: Option<u64>,
}

pub enum PhysicalASAPDAGValidationError {
    /// Bad input: `retention_ms` on a node that is neither a materialized view nor a query-time Scan. // new
    UnexpectedRetention(PhysicalASAPNodeId),
    /// Bad input: unsupported schema version. // removed (now `InstallablePlanError::UnsupportedWireVersion`)
    UnsupportedVersion(u32),
    // every other variant unchanged:
    MaterializationTimingMismatch { node: PhysicalASAPNodeId, timing: ExecutionTiming, materialization: Materialization }, // unchanged
    QueryDependencyInIngestion { producer: PhysicalASAPNodeId, consumer: PhysicalASAPNodeId }, // unchanged
    DuplicateNodeId(PhysicalASAPNodeId), // unchanged
    MissingRoot(PhysicalASAPNodeId), // unchanged
    MissingEdgeEndpoint(PhysicalASAPNodeId), // unchanged
    EdgeSchemaMismatch { producer: PhysicalASAPNodeId, consumer: PhysicalASAPNodeId }, // unchanged
    EdgeDataStateMismatch { producer: PhysicalASAPNodeId, consumer: PhysicalASAPNodeId }, // unchanged
    EdgeRoleMismatch { consumer: PhysicalASAPNodeId }, // unchanged
    NoRoots, // unchanged
    ScalarRootPosition(usize), // unchanged
    Cycle, // unchanged
    UnreachableNode(PhysicalASAPNodeId), // unchanged
    SummaryFamilySchemaMismatch { node: PhysicalASAPNodeId }, // unchanged
    SummaryGroupingMismatch { node: PhysicalASAPNodeId }, // unchanged
}

impl Materialization {
    /// `MatAtIngestion` or `MatAtQueryKeptAcrossEvaluations`: the node is a materialized view. // new
    pub fn is_materialized_view(self) -> bool;
}

crates/planner/src/installable.rs, crates/planner/src/lib.rs

/// Version of the whole plan file (new; replaces PHYSICAL_ASAP_DAG_WIRE_VERSION = 10).
pub const INSTALLABLE_PLAN_WIRE_VERSION: u32 = 11;

/// The plan file a runtime installs (type stays).
pub struct InstallablePlan {
    /// INSTALLABLE_PLAN_WIRE_VERSION of the writer. // new
    pub wire_version: u32,
    /// The selected expanded DAG; view nodes carry `retention_ms`. // was `PhysicalASAPDAGDocument`
    pub dag: PhysicalASAPDAG,
    /// Query i's text; answered by `dag.roots[i]`. // was `queries: Vec<InstalledQuery>`
    pub query_texts: Vec<String>,
    pub stored_outputs: Vec<StoredOutput>, // removed (derived: `materialized_views()`)
    pub query_time_raw_windows: Vec<RawDataSampleWindow>, // removed (derived: `raw_data_retention()`)
    pub selection: Option<SelectedPlanReport>, // removed (returned beside the plan)
}
pub struct InstalledQuery { pub query: String, pub root_node_id: PhysicalASAPNodeId } // removed

impl InstallablePlan {
    /// The DAG's views (delegates to `PhysicalASAPDAG::materialized_views`). // new
    pub fn materialized_views(&self) -> Result<Vec<MaterializedView>, InstallablePlanError>;
    /// The DAG's raw-data retention. // new
    pub fn raw_data_retention(&self) -> Result<Vec<RawDataRetention>, InstallablePlanError>;
    /// Reads a plan file, checking `wire_version` before the full parse. // new
    pub fn from_json(text: &str) -> Result<InstallablePlan, InstallablePlanError>;
    /// Only non-derived facts: version, texts per root, no scalar root, view and scan retention ≥ reach (validates the DAG). // changed
    pub fn validate(&self) -> Result<(), InstallablePlanError>;
    /// Executor compiles each refresh and the query run. // unchanged signature
    pub fn check_compiles(&self) -> Result<(), String>;
    pub fn query_time_reference(&self) -> Result<InstallablePlan, InstallablePlanError>; // removed (moved to test_support)
}

/// Plan for a runtime; Stage 3's report is beside the plan. // was `-> Result<InstallablePlan, PlanError>`
pub async fn plan_installable(input: UserInput<'_>)
    -> Result<(InstallablePlan, Option<SelectedPlanReport>), PlanError>;
/// The plan of selected roots, with the report. // was `-> Result<InstallablePlan, InstallablePlanError>`
pub fn installable_plan_of(roots: &[QueryRoot], selection: Option<&Selection>, entries: &[QueryWorkloadEntry])
    -> Result<(InstallablePlan, Option<SelectedPlanReport>), InstallablePlanError>;

/// Correctness-test helper (feature `test-support`): the plan with every node at query time. // was `InstallablePlan::query_time_reference`
pub mod test_support { pub fn query_time_reference(plan: &InstallablePlan) -> Result<InstallablePlan, InstallablePlanError>; }

pub enum InstallablePlanError {
    /// Deadline expired in the compile check. // unchanged
    DeadlineExceeded,
    /// Internal bug or reference cannot run at query time. // unchanged
    Timing(ExecutionDataStateError),
    /// Internal bug: a pane chain does not expand. // unchanged
    PaneChainExpansion(SchemaDerivationError),
    /// Bad input (plan file): another wire version; None = no `wire_version` (written before 11). // new
    UnsupportedWireVersion(Option<u32>),
    /// Bad input (plan file): not JSON of an InstallablePlan (`from_json`). // new
    Json(serde_json::Error),
    /// Bad input (plan file): invalid DAG. // unchanged
    InvalidDag(PhysicalASAPDAGValidationError),
    /// Bad input (plan file): not one query text per root. // unchanged name, now about `query_texts`
    QueriesDoNotMatchRoots { roots: usize },
    /// Unsatisfiable request: a standalone scalar query. // unchanged
    UnsupportedQueryRoot { entry_index: usize },
    /// Unsatisfiable request or bad input: views or raw windows not derivable. // new (wraps InstallableDagError)
    Uninstallable(InstallableDagError),
    /// Internal bug: Stage 3 pane retention for a non-chain node. // unchanged
    PaneRetentionOfUnknownChain { node_id: PhysicalASAPNodeId },
    /// Unsatisfiable request: the executor cannot refresh a view. // was `StoredOutputNotExecutable`
    MaterializedViewNotExecutable { node_id: PhysicalASAPNodeId, reason: String },
    WindowsOrStoredOutputsDoNotMatchDag, // removed (no derived copy left to mismatch)
    StoredOutputRetentionTooShort { node_id: PhysicalASAPNodeId }, // removed (InstallableDagError::ViewRetentionTooShort)
    StoredOutputReadsUnwindowedSource { node_id: PhysicalASAPNodeId }, // removed (InstallableDagError::ViewReadsUnwindowedSource)
    StoredOutputWithoutCoverage { node_id: PhysicalASAPNodeId, reason: String }, // removed (InstallableDagError::ViewWithoutCoverage)
    UnsupportedRawDataSampleWindow { time_range_node_id: PhysicalASAPNodeId, reason: &'static str }, // removed (InstallableDagError::UnsupportedRawDataWindow)
    IngestionBuildReadsWindowsOfDifferentOffsets { node_id: PhysicalASAPNodeId, end_offsets_ms: Vec<i64> }, // removed (moved to InstallableDagError)
}

SelectedPlanReport, PredictedUsage, PredictedNodeUsage, PredictedQueryUsage and PlannerExecutorCompileCheck are unchanged. The planner re-exports MaterializedView, MaterializedViewInputs, MaterializedViewRefresh, RawDataRetention, RawDataWindow, ViewCoverage and InstallableDagError. It no longer re-exports PhysicalASAPDAGDocument, RawDataSampleWindow, InstalledQuery, StoredOutput* or StoredCoverage.

Plan file and viewer document formats

• Plan file (InstallablePlan JSON): top-level keys are wire_version, dag and query_texts. In dag.nodes[], a view node and a bounded query-time Scan have "retention_ms": <u64>; other nodes omit it.
• tools/dag-viewer/stages.js document contract: stage3_plan_selection.stored_outputs / stored_outputs_error → materialized_views / materialized_views_error. Each entry is a MaterializedView (refresh: at_ingestion | on_first_read; inputs: raw_data_windows[{source, end_offset_ms, width_ms}], materialized_views) plus stage2_node_id, installed_position and installed_count.

Usage

Compiled and run as a test in crates/integration-tests at this head (not committed; the committed equivalents are crates/integration-tests/tests/materialized_views.rs and runtime_handoff.rs):

// The user's workload: `rate(m[5m])` every minute over 1,000 series of `m`
// sampled every second, at 1% error.
let workload: PlanningWorkload = serde_json::from_str(r#"{
  "query_workload": {"language": "promql", "repeating_queries": [
    {"query": "rate(m[5m])", "demand": {"fixed_interval": 60000},
     "requirements": {"accuracy": {"max_error": 0.01, "failure_probability": 0.01}},
     "predictability": {"predictable": {"known_at": null}}}]},
  "data_workload": {"arrival": {"ingesting": {"rows_per_second": 1000.0}},
    "sources": {"m": {"series_count": 1000, "sample_interval": 1000}}}}"#).unwrap();
// The deployment keeps no raw data, so Stage 2/3 must materialize panes at ingestion.
let capabilities = DeploymentCapabilities { raw_data_retained: false, ..DeploymentCapabilities::default() };
// Frontends, Stages 1-3 (with the executor compile check), pane-chain expansion, then
// `set_retention` writes each pane's retention onto its node.
// `plan` is the plan file; `report` is Stage 3's record, kept out of the file.
let (plan, report) = plan_installable(UserInput::new(&workload, FrontendInput::Promql { histograms: None },
    PlanningModels::builtin().with_deployment_capabilities(&capabilities))).await.unwrap();
println!("selected {}", report.unwrap().selected_candidate_id);   // selected pane-round-15-joint-plan-1
// What crosses the process boundary: the serialized plan file.
let file = serde_json::to_value(&plan).unwrap();
println!("{:?}", file.as_object().unwrap().keys().collect::<Vec<_>>()); // ["dag", "query_texts", "wire_version"]
// The runtime reads the file back and checks the non-derived facts.
let plan = InstallablePlan::from_json(&file.to_string()).unwrap();
plan.validate().unwrap();
// The views to maintain, derived from the DAG (asap-types): five 1-minute panes refreshed at
// ingestion, each read from `m`, each kept 5 minutes.
let views = plan.materialized_views().unwrap();
// 5 views, e.g. [(2, AtIngestion, [RawDataWindow { source: TimeSeries { metric: "m" }, end_offset_ms: 0, width_ms: 60000 }], [], 300000),
//                (5, AtIngestion, [RawDataWindow { source: TimeSeries { metric: "m" }, end_offset_ms: 60000, width_ms: 60000 }], [], 300000)]
// Raw data read at query time: none, since every scan runs at ingestion.
println!("{:?}", plan.raw_data_retention().unwrap());              // []
plan.check_compiles().unwrap();                                     // the executor compiles each refresh and the query run

With raw data kept (the default capabilities), rate(m[5m]) plus rate(m[5m] offset 1h) plan at query time. plan.raw_data_retention() is then [RawDataRetention { source: TimeSeries { metric: "m" }, lookback_ms: Some(3_900_000) }], which is 65 minutes and equals Stage 3's priced lookback (raw_data_retention_is_the_lookback_stage3_prices).

The same query, max_over_time(m[5m]), range-evaluated over the last hour at query time also keeps 65 minutes of m: the 5-minute window plus the hour of steps (a_range_evaluated_query_time_scan_keeps_its_window_plus_the_span).

Major algorithm changes

Raw-data retention (new).

• Problem: a runtime could not tell how long to keep raw data. A first version read only scan_window, which misses range evaluation. For max_over_time(m[5m]) over the last hour with query-time results kept, the panes are kept 3,900,000 ms, but that version said m: 300_000. Refreshing a pane for the oldest step reads 65 minutes back.
• After: set_retention gives each query-time Scan the following, and raw_data_retention takes the per-source max:
  1. scan_window(scan).lookback_ms();
  2. plus the longest range-evaluation span of the queries that read the scan (reader_span_ms_by_node);
  3. at least the retention_ms of each view refreshed on first read from the scan;
  4. None when the window is unbounded.
• Correctness: each scan's retention covers its window (checked in raw_data_retention). Its oldest read is a range query's oldest step, or the oldest refresh of a view refreshed on first read. Tests: query_time_results_kept_across_evaluations_are_built_in_query_runs (now 65 min), a_range_evaluated_query_time_scan_keeps_its_window_plus_the_span, raw_data_retention_is_the_lookback_stage3_prices.
• Complexity: O(query-time scans · (nodes + edges)), since scan_window is now indexed once per call.
• Limits: Stage 3 prices one evaluation's window, without the range span, and a kept chain as its newest pane, so the plan's value can exceed Stage 3's. Stage 3 is unchanged here; a follow-up issue covers pricing the span.

Retention of views, coverage, view inputs and the compile check compute what they did before; they are moved into asap-types and read from the DAG.

Before this PR

InstallablePlan {
    dag: PhysicalASAPDAGDocument { schema_version: 10, dag },   // plan.dag.dag
    queries: vec![InstalledQuery { query: "rate(...)", root_node_id: 7 }],
    stored_outputs: vec![StoredOutput { node_id: 4, built_at: IngestionRun, coverage, inputs, retention_ms: 360_000 }],
    query_time_raw_windows: vec![],
    selection: Some(SelectedPlanReport { .. }),
}

validate recomputed queries, stored_outputs and query_time_raw_windows and rejected a mismatch. A runtime had no raw-data retention to read.

After this PR

let (plan, report) = plan_installable(input).await?;
InstallablePlan { wire_version: 11, dag /* node 4: retention_ms: Some(360_000) */, query_texts: vec!["rate(...)".into()] }
plan.materialized_views()?   // [MaterializedView { node_id: 4, refresh: AtIngestion, coverage, inputs, retention_ms: 360_000 }]
plan.raw_data_retention()    // per Source, the longest query-time lookback

Every fact has one source. A runtime that depends only on asap-types derives the views and the raw-data retention from the plan file's dag.

Verification

• Equality with main: plans captured on main (f06a555) are in crates/integration-tests/tests/materialized_view_fixtures/stored_outputs_on_main.json. There are eight: four repeating PromQL workloads without raw data, and four runtime_install forced plans covering rollups read by views and a kept sliding-window merge refreshed on first read. Each holds the plan file's DAG and its stored_outputs. materialized_views_equal_the_stored_outputs_main_listed checks that materialized_views() of each DAG, with the captured retention on the nodes, equals stored_outputs field for field after the renames. It passes.
• One-time check (not committed): re-planning the repeating workloads with this branch gives the same DAG and retention as main for p99-60m, p99-p50-60m and offset-chain. The rollup capture differs, but the main capture was a release build and the comparison a debug build. A debug build of main selects the same candidate (seed-joint-plan-2, 0.08500 cost/s) as this branch, so the difference is not from this PR.
• raw_data_retention_is_the_lookback_stage3_prices: m: Some(3_900_000) (65 min), equal to the per-source max of scan_window(..).lookback_ms() over the query-time scans of Stage 3's selected DAG. Without raw data kept, it is empty.
  • This equality holds only without range evaluation and kept chains. Stage 3 does not price the range span, so a range-evaluated plan reports more than Stage 3 prices, as a_range_evaluated_query_time_scan_keeps_its_window_plus_the_span (65 min against Stage 3's 5) shows.
• New crates/types/tests/materialized_view.rs covers ViewWithoutRetention, ViewRetentionTooShort, retention propagation from a view refreshed on first read to its inputs (panes kept 2 min + 1 h), and the derivation deadline.
• a_plan_file_of_an_older_wire_version_is_refused_by_version: a v10 plan file is refused by from_json as UnsupportedWireVersion(None). The file is materialized_view_fixtures/plan_wire_version_10.json, assembled from the DAG and stored_outputs of a plan written on main, in the v10 layout {dag: {schema_version: 10, dag}, queries, stored_outputs, query_time_raw_windows, selection}. A newer version is refused as Some(v), and a current file round-trips.
• Commands, from the worktree (rebased on main 249ebd2), with LIBCLANG_PATH set:
  • cargo test -p asap-types --test physical_export --test materialized_view: 18 / 4 passed.
  • cargo test -p asap-planner --lib: 6 passed.
  • cargo test -p asap-planner --test runtime_install --test per_sample_updater --test columnar_batches --test sql_per_group_topk --test sql_table_scans: 41 / 2 / 1 / 6 / 11 passed.
  • cargo test -p asap-integration-tests --test materialization_e2e --test per_series_exact_execution --test runtime_handoff --test tumbling_pane_merge --test kll_pane_execution --test materialized_views: 4 (1 ignored) / 8 / 8 / 13 (1 ignored) / 2 / 4 passed.
  • cargo test -p asap-plan-selection --lib raw_data: 7 passed.
  • cargo test -p asap-devtools --test viewer_contract: 2 passed.
  • cargo test -p asap-devtools --test stage_pipeline: 45 passed, 2 failed. Each failure gives the same value when the same test runs on main, so neither is from this PR:
    • workload_file_with_a_range_evaluation_is_planned: cost 719154.73, also on main f06a555.
    • scans_projecting_different_labels_are_shared: cost 0.21003, also on main 249ebd2.
  • tools/dag-viewer: python3 -m unittest test_viewer (74 OK), python3 -m unittest test_server (28 OK), node --test stages.test.mjs (14 pass).
  • cargo fmt --all --check: clean.
  • cargo clippy -p asap-types -p asap-planner -p asap-devtools -p asap-integration-tests -p asap-executor -p asap-plan-selection --all-targets -- -D warnings: clean.
  • cargo check --workspace --all-targets --all-features: clean.
• I did not run the plan-selection test producer_reuse_rules_produce_validated_shared_plans, which is known to fail on main (from test(optimizer): read the reuse and sharing tests from the memo, not the whole-workload Pass 2 #983/fix(optimizer): an exact rollup reads the finer query's own result #984).

Limitations and follow-ups

• Stage 3 prices raw data without the range-evaluation span, and a kept chain as its newest pane. The plan file's raw-data retention includes both: the all-panes reach is what a deployment needs. Pricing the span in Stage 3 is a separate follow-up issue.
• InstallablePlan stays in the planner crate, because check_compiles needs the executor and the planning errors live there. The derivation and its types are in asap-types, so a runtime that depends only on asap-types can call PhysicalASAPDAG::materialized_views() and raw_data_retention() on the plan file's dag.
• materialized_views() and raw_data_retention() return a Result, where the issue sketch returns a Vec. A plan file read from disk may be invalid: a missing or too-short retention_ms, or a raw window of an unsupported shape.
• set_retention is public, because the planner crate calls it. A runtime does not need it.
• ASAPQuery-backend, which pins Planner f46cbf6c, must move to the new plan-file format (wire version 11).
• docs/design_docs/physical-planning-and-deployment.md still names stored_outputs. docs(design): physical DAG to deployment interface (raw data and materialized view slots, DataStore) #972 owns that document.