[Refactor] InstallablePlan holds only a versioned DAG and query texts; remove PhysicalASAPDAGDocument; StoredOutput becomes MaterializedView #970

Example

For rate(http_requests_total{job="api"}[5m]) with 1-minute panes kept at ingestion, plan_installable returns (crates/planner/src/installable.rs:74):

InstallablePlan {
    dag,                         // the selected physical ASAP 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 { .. }),
}

Only three values here come from outside the DAG: the query text, retention_ms, and the selection report. InstallablePlan::validate (installable.rs:650) recomputes everything else from dag and rejects a plan whose copy differs (WindowsOrStoredOutputsDoNotMatchDag, QueriesDoNotMatchRoots):

After this issue:

pub struct InstallablePlan {
    pub wire_version: u32,            // INSTALLABLE_PLAN_WIRE_VERSION
    pub dag: PhysicalASAPDAG,         // node 4 carries retention_ms: Some(360_000)
    pub query_texts: Vec<String>,     // query i is answered by dag.roots[i]
}
impl InstallablePlan {
    pub fn materialized_views(&self) -> Vec<MaterializedView>;
    pub fn raw_data_retention(&self) -> Vec<RawDataRetention>; // m: keep 5m
}
pub struct RawDataRetention {
    pub source: Source,
    pub lookback_ms: Option<u64>, // None = unbounded
}

Concrete maintenance problem

• Every derived field has two sources: the DAG and the plan file's copy. validate exists mostly to check that they agree.
• selection is a report the runtime never reads, but it is part of the plan file a runtime installs.
• query_time_reference() is documented "for correctness tests only, never for installing", yet it is a public method of the installed type.
• After [Refactor] Time-series TimeRange as a native operator; bind raw data only at Scan; remove the raw-sample selectors #969, raw data binds at Scan through TableProvider. Then RawDataSampleWindow (StoredOutputInputs::raw_data_sample_windows, query_time_raw_windows) no longer tells a runtime where to bind anything.
• A runtime cannot tell how long to keep raw data. Stage 3 computes it as a per-source maximum (raw_data in price_at_scale), but the value is not in the plan, so a deployment has to re-derive it or keep everything.
• Naming: a node whose output is kept across evaluations is a materialized view (decided in [Tracking] Physical DAG → deployment boundary: InstallablePlan, input slots, TableProvider, BindingDataSources #968, following DataFusion's materialized-views extension), but the code calls it StoredOutput, StoredOutputBuild and "stored input".

Behavior that must remain unchanged

• A runtime learns the same facts: which nodes to keep, when each is refreshed, its coverage (definition_id, selection), what it is built from, how long to keep it, and which root answers which query.
• Stage 3's retention values are unchanged.
• The compile check still runs inside plan_installable.

Before and after structure

• PhysicalASAPDAGNode gains retention_ms: Option<u64>. It is set on exactly the materialized-view nodes, from Stage 3's pane retention and the readers' range spans as runtime_inputs_of computes today. Export validation rejects it on any other node.
  • It is a node field, not a payload of Materialization, because Stage 2 enumerates Materialization options before Stage 3 knows retention. Materialization stays Copy and comparable.
• Renames: StoredOutput → MaterializedView, StoredOutputBuild::{IngestionRun, QueryRun} → MaterializedViewRefresh::{AtIngestion, OnFirstRead}, StoredOutputInputs::stored_outputs → views, StoredCoverage → ViewCoverage.
• PhysicalASAPDAGDocument { schema_version, dag } is removed. Its only use is InstallablePlan.dag, which today puts the version one level below the file it versions (plan.dag.dag, 37 call sites, almost all tests). The plan file is what crosses the process boundary, so the version moves to InstallablePlan.wire_version. PHYSICAL_ASAP_DAG_WIRE_VERSION becomes INSTALLABLE_PLAN_WIRE_VERSION and covers the whole file: DAG, node fields and query texts. PhysicalASAPDAG::validate keeps every structural check.
• InstallablePlan keeps wire_version, dag and query_texts. materialized_views() derives the list in dependency order.
  • MaterializedView.inputs lists the views it reads, plus each raw-data Source it scans with that scan's time reach. The time reach is the retention lower bound that validate checks today, which replaces RawDataSampleWindow.
  • query_time_raw_windows is removed.
• raw_data_retention() gives, for each raw-data Source that some Scan reads at query time, the longest lookback among those scans. The lookback is end_offset_ms + width_ms of PhysicalASAPDAG::scan_window ([Refactor] Time-series TimeRange as a native operator; bind raw data only at Scan; remove the raw-sample selectors #969), and None when a scan is Unbounded. It is the same quantity Stage 3 prices, from the same function, so the plan file and the cost agree. A source read only at ingestion is absent. When Retained raw data as concrete state groups per lookback: offsets, @ and a retention limit #955 adds raw window(S, H) groups, this returns the selected H. It is the raw-data counterpart of each view's retention_ms.
• validate checks only what is not derived: the plan's wire version, the DAG's validity, query_texts.len() == dag.roots.len(), no scalar roots, and each view's retention covering its reach.
• plan_installable returns (InstallablePlan, Option<SelectedPlanReport>), so the report is not in the plan file.
• query_time_reference moves to a test-support function.
• The derivation lives where a runtime can call it without depending on the whole planner crate, in asap-types next to PhysicalASAPDAG.

Why the proposed structure is minimally complex

Four fields and one validation path are removed, and every fact has one source. One Option<u64> is added to the node, which is where the value belongs.

Existing unit and end-to-end protection

• crates/planner/tests/{runtime_install,per_sample_updater,columnar_batches,sql_per_group_topk,sql_table_scans}.rs.
• crates/integration-tests/tests/{materialization_e2e,per_series_exact_execution}.rs.
• installable.rs unit tests (retention, executor_check_rejects_projection_below_raw_window, a_time_shift_above_the_range_is_refused).
• Add a test: raw_data_retention() for rate(m[5m]) plus rate(m[5m] offset 1h), both at query time, is m: 65m. It equals Stage 3's priced raw lookback for the same plan.
• Add a test: for every workload in the integration suite, materialized_views() of the new plan equals today's stored_outputs field for field, after the renames.

Compatibility or migration impact

The plan-file format changes: fields are removed, retention_ms moves onto nodes, and types are renamed. PhysicalASAPDAGDocument is removed and the version moves to InstallablePlan.wire_version, bumped once for all of these changes. ASAPQuery-backend main pins an older Planner revision (f46cbf6c). Its executor integration reads the new format.

Parent: #968 (PR 6 of 6; after #969). Related: #955 (raw_data_retention() reports the raw window it selects).