feat(sds): define stored summaries from the plan's ingestion-time outputs

[zzylol]

Stack 2/5. Base: #818. Next: #820.

Why

The data plane loads the plan but doesn't know which outputs it must store. Each MatAtIngestion output the query side reads needs one stored summary definition that the precompute engine (#820) fills and the query engine (#821) reads.

What

Sds::from_dag turns the plan's ingestion-time frontier (asap_executor::physical_planner::frontier_from_timing) into stored summary definitions:

• Pane shape: each frontier output must read TimeRange(width) ← TimeShift(offset)* ← Scan: a pane of width ending at T − offset. Any other shape is rejected as UnsupportedShape. The control plane expands each PaneChain (refactor!: plan with ASAPPlanner main's stage pipeline and drop the old execution stack #818), so every pane of a chain has this shape.
• Shared definitions: outputs that differ only in their offset share one definition. A definition is identified by a SHA-256 over its sub-DAG: each node's payload with its input ids blanked and its output schema, then its inputs' identities. An offset-only TimeShift takes its input's identity, so the newest pane (no shift) and the older panes (shifted) share one. In Example 4b without raw data, the six 10-minute KLL panes become one definition with offsets 0, 10, …, 50 min.
• Unsupported materialization: a plan with a MatAtQueryKeptAcrossEvaluations node is rejected (KeptAcrossEvaluations). The backend declares query_time_results_retained: false, so the Planner does not select one.
• Storage: SummaryStore keeps panes by (definition, end) and evicts them by end time.
• Examples: examples/planner-layering-example3b.json (Pattern B) and -example4b.json (p99 over 60 min every 10 min, 1,000 series sampled every second) in the current workload format.

How

One pass over the frontier: walk single inputs down to the TimeRange (pane width), then sum offset TimeShifts down to the Scan (offset); hash the output's sub-DAG, memoized per node; group outputs by hash. The definition's builder is its lowest-numbered output.

Key code interfaces added / modified

// data_plane/src/sds.rs
/// What a stored summary computes, independent of node ids and of which
/// pane an output reads (hex SHA-256).
pub struct DefinitionId(pub String);

pub struct SummaryDefinition {
    /// Its identity.
    pub id: DefinitionId,
    /// Pane width: each stored summary covers `(end - pane_ms, end]`.
    pub pane_ms: i64,
    /// The frontier output that builds a pane (the first in node order).
    pub builder: u64,
    /// The `TimeRange` node fed one pane's samples.
    pub raw_input: u64,
    /// Schema of a stored pane.
    pub output_schema: Schema,
}

pub struct PaneBinding {
    /// At evaluation time `T` the output reads the pane ending at `T - offset_ms`.
    pub offset_ms: i64,
}

pub struct Sds {
    /// Every definition, by identity.
    pub definitions: BTreeMap<DefinitionId, SummaryDefinition>,
    /// Every frontier output, with its definition and pane.
    pub outputs: BTreeMap<u64, (DefinitionId, PaneBinding)>,
}
impl Sds {
    /// The definitions behind `frontier`, the outputs computed at ingestion
    /// time and read at query time.
    pub fn from_dag(dag: &PhysicalASAPDAG, frontier: &[u64]) -> Result<Self, SdsError>;
}

pub enum SdsError {
    /// A frontier node is not in the DAG.
    UnknownNode(u64),
    /// A frontier output does not read one pane of one `Scan`.
    UnsupportedShape { node: u64, reason: String },
    /// A node is `MatAtQueryKeptAcrossEvaluations`.
    KeptAcrossEvaluations(u64),
}

/// Built panes, by definition and pane end; a pane never changes once stored.
pub struct SummaryStore<S> { /* private */ }
impl<S> SummaryStore<S> {
    /// An empty store.
    pub fn new() -> Self;
    /// Stores the pane of `definition` ending at `end_ms`.
    pub fn insert(&mut self, definition: &DefinitionId, end_ms: i64, state: S);
    /// The pane of `definition` ending at `end_ms`, if stored.
    pub fn get(&self, definition: &DefinitionId, end_ms: i64) -> Option<&S>;
    /// Drops the panes of `definition` ending at or before `end_ms`.
    pub fn evict_through(&mut self, definition: &DefinitionId, end_ms: i64);
    /// Number of stored panes.
    pub fn len(&self) -> usize;
    /// Whether no pane is stored.
    pub fn is_empty(&self) -> bool;
}

Before this PR

The data plane installed the plan without knowing what to store.

After this PR

Example 4b on a deployment without raw data: one definition, six outputs reading it at offsets 0..50 min.

Usage

// #509 Example 4b: p99 over the last 60 min, every 10 min.
let workload: PlanningWorkload = serde_json::from_str(
    &std::fs::read_to_string("examples/planner-layering-example4b.json")?)?;
// A deployment that keeps no raw samples: every input is built at ingestion time.
let mut inputs = DeploymentInputs::default();
inputs.deployment_capabilities.raw_data_retained = false;
// The plan Stage 3 selected (P3-m1), its PaneChain expanded into six 10-minute panes.
let plan = plan_workload(&workload, &inputs)?;
let dag = &plan.dag.dag;
// The stored summaries behind the ingestion-time outputs.
let sds = Sds::from_dag(dag, &frontier_from_timing(dag)?)?;
// definitions=1 pane_ms=600000 offsets_ms=[0, 600000, 1200000, 1800000, 2400000, 3000000]

Verification

• data_plane/tests/sds.rs (7 tests): sharing and offsets, plans without definitions, identity independent of node numbering, identity depends on the source, rejected shapes, rejected MatAtQueryKeptAcrossEvaluations, store lookup and eviction.
• cargo test -p control_plane -p data_plane, fmt, clippy on this PR's tree.
• Usage output above is from running it (as a scratch test) on the feat(query): answer planned instant queries by running the plan's roots #821 tree, whose SDS API is this PR's.