[zzylol]

Why

A runtime's summary store should hold every summary the plan stores. InstallablePlan could describe only one kind: a node built directly from one pane of raw samples (StoredOutput.pane). plan_installable refused anything else with StoredOutputWithoutPane. For example, a sliding-window merge of stored panes kept across evaluations failed with Installable(StoredOutputWithoutPane { node_id: 15 }). So a runtime could store only panes, and every other summary the plan stores was refused at install time.

What

• Each StoredOutput says what the stored summary is, what its build reads, when it is built, and how long it is kept:
  • coverage is the IR's SummaryCoverage: a stable definition_id and the selection (value sets, value ranges, and time relative to the evaluation time).
  • inputs are the raw-sample windows and the stored outputs (by node id) it reads.
  • built_at and retention_ms are as before.
• plan_installable lists every stored node (MatAtIngestion or MatAtQueryKeptAcrossEvaluations) in dependency order. It refuses only what a runtime cannot build: a stored node that reads a source outside a raw-sample window, a stored node with no summary coverage, and one the executor does not compile with its inputs bound.
• pane is dropped. It was the single raw-sample window of a pane's inputs, and its time is the time part of the selection. Its consumers are this repo's tests, which are migrated here, and ASAPQuery-backend, which is migrating to the general form.
• Docs. The installable-plan docs (the installable module doc) say how a runtime builds an output from stored inputs.

How

• Coverage.

  • A pane's coverage is SummaryCoverage::derive.
  • A merge's selection is the union of its inputs' selections under their shared definition, following the SummaryMerge rules.
  • A summary built from exactly one stored output, with no selection of its own (a rollup to a coarser grouping, or a summary across series built from per-series ones), takes that input's selection. Its definition is its own sub-DAG with the input replaced by the input's definition. It therefore gets a definition of its own, and rollups of equal inputs share it.
  • Anything else keeps its own derived coverage, conservatively.

• definition_id. A 128-bit FNV-1a hash of the definition's operators, output kinds and schemas, children first. It ignores timing, guarantee and materialization, so it agrees with the equality a merge requires. A summary store can key instances by (definition id, concrete selection).

• Inputs. Walking down from a stored node, each stored node and each raw-sample TimeRange it reaches is an input. A Scan reached without a TimeRange is refused. Stored outputs are ordered children first, so a pane-only plan keeps its node order.

• Executability. For each stored output, the executor's own compile runs with its inputs bound as sources, the way a runtime runs it. A failure is refused with the executor's reason.

• Retention.

  • A pane takes Stage 3's chain retention, as before.
  • Otherwise, an output is kept as far back as its data reaches (its windows' and stored inputs' end_offset_ms + width_ms). An output built from stored outputs adds its readers' longest range-evaluation span.
  • Then each stored input is kept as long as the outputs built from it need it: their retention when they are built in a query run, since a range request may build one for its oldest step, or else as far back as their data reaches.
  • validate refuses an output kept for less than its data reaches.

• Shapes Stage 3 does not select today. With the built-in models, none of the three shapes in the issue is selected:

  • the rollup is shared within an evaluation (MatAtQuery);
  • the cross-series maximum is not materialized;
  • Stage 2 never offers a SummaryMerge as a materialization candidate.

  The tests force these shapes through UserInput::pass. A test pass runs Stages 1 to 3, re-times the selected plan with the chosen nodes stored, and splits nodes shared across timings as Stage 2 does. It then goes through plan_installable unchanged.

Key code interfaces added / modified

crates/planner/src/installable.rs

/// One stored summary of the installable DAG. The type stays; its fields change.
pub struct StoredOutput {
    /// The node whose output is stored.
    pub node_id: PhysicalASAPNodeId, // unchanged
    /// IngestionRun (once its inputs are complete) or QueryRun (first evaluation that reads it).
    pub built_at: StoredOutputBuild, // unchanged
    /// What the stored state is: its summary coverage.
    pub coverage: StoredCoverage, // new
    /// The raw-sample windows and stored outputs its build reads.
    pub inputs: StoredOutputInputs, // new
    /// How far back from the newest evaluation it stays available (rules above).
    pub retention_ms: u64, // unchanged; now also set for outputs built from stored outputs
    // pub pane: RawSampleWindow, // removed: it is inputs.raw_sample_windows[0] of a pane
}

/// A stored output's SummaryCoverage. New type.
pub struct StoredCoverage {
    /// SummaryCoverage::definition_id: equal for every node that computes the same thing.
    pub definition_id: String, // new
    /// The rows that went in; a merge's is its inputs' union.
    pub selection: Vec<SelectionBox>, // new
}

/// What a stored output's build reads, each bound as a source. New type.
pub struct StoredOutputInputs {
    /// Windows of raw samples it reads directly, in node order.
    pub raw_sample_windows: Vec<RawSampleWindow>, // new
    /// Stored outputs it reads, by node id, each listed before it.
    pub stored_outputs: Vec<PhysicalASAPNodeId>, // new
}

/// Why an installable plan could not be built. The type stays; variants change.
pub enum InstallablePlanError {
    /// Internal bug or bad plan input: timings do not hold.
    Timing(ExecutionDataStateError), // unchanged
    /// Internal bug: a pane chain does not expand.
    PaneChainExpansion(SchemaDerivationError), // unchanged
    /// Bad input: the DAG is not valid.
    InvalidDag(PhysicalASAPDAGValidationError), // unchanged
    /// Bad input: queries do not bind the roots.
    QueriesDoNotMatchRoots { roots: usize }, // unchanged
    /// Unsatisfiable: a standalone scalar query.
    UnsupportedQueryRoot { entry_index: usize }, // unchanged
    /// Bad input: stored outputs or windows do not match the DAG.
    WindowsOrStoredOutputsDoNotMatchDag, // unchanged
    /// Bad input: kept for less than its data reaches back.
    StoredOutputRetentionTooShort { node_id: PhysicalASAPNodeId }, // unchanged; was "than its pane"
    /// Internal bug: retention for an unknown chain.
    PaneRetentionOfUnknownChain { node_id: PhysicalASAPNodeId }, // unchanged
    // StoredOutputWithoutPane { node_id }, // removed
    /// Unsatisfiable: the build reads a source Scan outside a raw-sample window (e.g. a whole SQL table).
    StoredOutputReadsUnwindowedSource { node_id: PhysicalASAPNodeId }, // new
    /// Unsatisfiable: the stored node is not a summary state with a coverage.
    StoredOutputWithoutCoverage { node_id: PhysicalASAPNodeId, reason: String }, // new
    /// Unsatisfiable: the executor does not compile the build with its inputs bound.
    StoredOutputNotExecutable { node_id: PhysicalASAPNodeId, reason: String }, // new
    /// Unsatisfiable: a raw-sample window shape a runtime cannot feed.
    UnsupportedRawSampleWindow { time_range_node_id: PhysicalASAPNodeId, reason: &'static str }, // unchanged
}

crates/types/src/ir/properties/summary_coverage.rs

impl SummaryCoverage {
    /// Stable hex id of `definition`; equal definitions get equal ids on any node.
    pub fn definition_id(&self) -> String; // new
}
/// Now Serialize/Deserialize: columns as [{column, constraint}], bounds as
/// {"included"|"excluded": v} or "unbounded".
pub struct SelectionBox { pub columns: BTreeMap<ColumnIdentity, Constraint>, pub relative_time: Option<(Bound<i64>, Bound<i64>)> } // fields unchanged

crates/types/src/ir/physical_export.rs

impl PhysicalASAPDAG {
    /// Every node as untimed IR by id, the rebuild query_time_reference already used; validates first.
    pub fn operator_nodes(&self) -> Result<HashMap<PhysicalASAPNodeId, Rc<OperatorNode>>, PhysicalASAPDAGValidationError>; // new
}

Usage (the test a_kept_sliding_window_merge_is_listed_and_built):

// A dashboard panel: max_over_time(m[5m]) every minute, answered as the last hour
// at 1-minute steps, over 6 ingesting series; the deployment keeps query-time results.
let workload = exact_workload(&["max_over_time(m[5m])"],
    r#", "range_evaluation": {"step_ms": 60000, "span_ms": 3600000}"#);
// Stage 3 keeps five 1-minute panes; the test pass also keeps their merge.
let plan_json = forced_plan_json(&workload, &capabilities, merges_kept).await;
let plan: InstallablePlan = serde_json::from_str(&plan_json)?;
// stored_outputs, in dependency order:
//   5 panes: inputs.raw_sample_windows = [1-minute window at offset 0..4 min],
//            coverage = { definition_id: D, selection: time (-(i+1) min, -i min] },
//            built_at QueryRun, retention_ms 3,900,000
//   merge:   inputs.stored_outputs = [the 5 panes], no raw windows,
//            coverage = { definition_id: D, selection: time (-5 min, 0] },
//            built_at QueryRun, retention_ms 300,000 + 3,600,000
// The runtime builds each output once its inputs are complete: the executor compiles
// the DAG for the node with each window's rows bound at its TimeRange and each
// stored input's state at its node id. It keys states by (D, concrete selection).
let first = run_data_plane(&plan_json, &samples, EVALUATION_MS, &mut store);  // builds 6
let next = run_data_plane(&plan_json, &samples, EVALUATION_MS + 60_000, &mut store); // builds 2
// Both answer as plan.query_time_reference().

Before this PR

• A stored output that was not built from one raw pane was refused with StoredOutputWithoutPane, for example the kept merge above.
• A rollup materialized from finer stored sums was listed with the finer sums' raw pane as if it read raw samples, so a runtime would rebuild it from raw samples.
• The plan said nothing about what a stored state is, so a runtime had to invent its own identity to share states.

After this PR

• All three shapes in the issue are listed with their inputs, coverage, build phase and retention, and built from stored inputs. The answers equal the query-time reference.
• A runtime keys stored states by (definition id, selection made concrete).
• Pane-only plans list the same nodes in the same order, with the same windows (now inputs.raw_sample_windows[0]), built_at and retention_ms. The new fields are added and pane is gone.

Verification

Tests written first.

• Refusal on main. plan_installable of the kept-merge plan failed on main with StoredOutputWithoutPane { node_id: 15 }.
• planner/tests/runtime_install.rs. Each test checks inputs, coverage, phase and retention, then builds every output in the listed order from stored inputs with the executor. The answers equal query_time_reference.
  • a_rollup_from_finer_stored_sums_is_listed_and_built: sum by (job) from stored sum by (job, instance), both at ingestion. The selection is (−60s, 0] for both, the definitions differ, and 2 builds are needed.
  • a_summary_across_series_from_per_series_summaries_is_listed_and_built: max by (job) from stored per-series maxima.
  • a_kept_sliding_window_merge_is_listed_and_built: as in the usage above. A second evaluation one minute later builds only the newest pane and the new merge.
  • panes_of_an_hour_share_one_definition_with_distinct_times: the six 10-minute panes of quantile_over_time(0.99, latency_ms[60m]) share one definition id. Their time selections are (−(i+1)·10m, −i·10m] and their windows are unchanged.
  • The existing pane-plan tests pass unchanged on the generalized runtime harness, which keys states by (definition id, concrete selection): 6 panes built for one hour, 7 for an hour and its offset twin, and 5 then 1 for kept panes. So do per_sample_updater, columnar_batches and integration-tests/runtime_handoff.
• types/tests/summary_coverage.rs.
  • equal_definitions_get_equal_ids: equal across selections and timings, different for another column, family or source.
  • panes_of_an_hour_share_one_definition_with_distinct_times: the same property, plus their merge.
  • selections_round_trip_through_json.

Commands. cargo test --workspace --locked at the head: exit 0, 150 test binaries, 0 failures. Following the current speed-first policy, fmt and clippy were not run.

Limitations.

• Conservative coverage. It follows the existing SummaryCoverage rules:
  • An instant selector's TimeRange and its label matchers stay in the definition, with no time in the selection. Such a state's concrete identity also needs the evaluation time; the test harness adds it to the key.
  • A summary built from several stored outputs that is not a merge keeps its own derived coverage, with the inputs' windows inside the definition.
• Merges are not materialized by Stage 2. Stage 2 still never offers a SummaryMerge as a materialization candidate, and Stage 3 does not select the forced rollup shapes under the built-in prices. This PR makes them installable; making Stage 2 and Stage 3 choose them is separate work.
• Review. Not reviewed independently.

Closes #781