feat(query): keep query-time panes across evaluations

[zzylol]

Stack 6/6. Base: #822.

Why

#819 rejected any plan that keeps a query-time result across evaluations, and the backend declared query_time_results_retained: false, so the Planner never offered #509 Example 4's B3 (tumbling panes built at query time and kept). The design's query engine section specifies MatAtQueryKeptAcrossEvaluations, so the backend can now execute it and declare it.

What

• Kept outputs: a MatAtQueryKeptAcrossEvaluations node's output is stored after it is computed, in memory. It is keyed like a pane, by (definition, pane end). Identity follows the SDS rule: an offset-only TimeShift takes its input's identity, and the offset goes into the pane binding. A kept output must have the stored-pane shape TimeRange(width) ← TimeShift(offset)* ← Scan; any other shape is UnsupportedShape.
• Reuse: at evaluation time T, a kept output reads the kept pane ending at T − offset. A missing one is computed from its pane's raw samples (end − width, end], fetched from Prometheus.
• Kept only once closed: a computed pane is kept only when it is closed by the precompute engine's rule: its end is at or before the watermark (newest sample minus --allowed-lateness-ms; no watermark before the first sample). A pane read while still open is computed for that evaluation but not kept. A later evaluation computes it again, so it includes samples that arrived late, and keeps it once it is closed. In B3 each evaluation after the first fetches one minute of raw samples instead of five.
• Eviction: same rule as stored panes. A pane is evicted once it is older than the largest read offset plus two widths, measured from the newest evaluation. An older evaluation is still answered, but its panes are evicted right away.
• Default capabilities: backend_capabilities() sets query_time_results_retained: true.
• Removed: SdsError::KeptAcrossEvaluations.
• Unchanged: forwarding, x-asap-execution, x-asap-forward-reason.
• Example: examples/planner-layering-example4-b3.json is Example 3b's panel (quantile_over_time(0.99, latency_ms[5m]) every minute) over 10 series sampled every millisecond (10,000 rows/s). The Planner selects B3 (P3-m2) for it under the default cost model.

How

Sds::from_dag runs the frontier's per-output pass over every kept node too, into kept_definitions/kept_outputs. The pane build (Scan, then the definition's sub-DAG from its TimeRange up to its builder) moves out of the precompute engine into PaneBuild, which both engines use. The precompute engine runs it under ingestion scope. The query engine runs it under query scope at the pane's end, so a kept pane is the same whichever evaluation built it. QueryEngine compiles each root with kept outputs as bounded inputs. raw_ranges asks for the raw range of each kept pane not held yet. evaluate builds those panes, keeps the closed ones (the engine thread passes the precompute engine's watermark), reuses the rest, then evicts.

Key code interfaces added / modified

// data_plane/src/sds.rs
pub struct PaneBinding {
    /// At evaluation time `T` the output reads the pane ending at `T - offset_ms`.
    pub offset_ms: i64,
    /// The output's own `TimeRange` node: where that pane's raw samples
    /// feed when the output is computed at query time.
    pub raw_input: u64, // new
}

pub struct Sds {
    /// Definitions built at ingestion time, by identity.
    pub definitions: BTreeMap<DefinitionId, SummaryDefinition>,
    /// Every frontier output, with its definition and pane.
    pub outputs: BTreeMap<u64, (DefinitionId, PaneBinding)>,
    /// Definitions of the outputs computed at query time and kept across
    /// evaluations, by identity.
    pub kept_definitions: BTreeMap<DefinitionId, SummaryDefinition>, // new
    /// Every output kept across evaluations, with its definition and pane.
    pub kept_outputs: BTreeMap<u64, (DefinitionId, PaneBinding)>, // new
}

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

// data_plane/src/query.rs
/// What happened to the kept panes evaluations read.
pub struct KeptResultStats {
    /// Kept panes computed from raw samples, because no evaluation had kept
    /// them; kept if closed.
    pub built: u64,
    /// Kept panes read from an earlier evaluation instead of recomputed.
    pub reused: u64,
}

pub struct QueryEngine {
    /// Kept panes (`MatAtQueryKeptAcrossEvaluations`), by definition and
    /// pane end, in memory.
    pub kept: SummaryStore<Batch>,
    /// How many kept panes evaluations built and reused.
    pub kept_stats: KeptResultStats,
    /* private: compiled roots, kept pane builds, newest evaluation time */
}
impl QueryEngine {
    /// The root answering `query` at `time_ms` and the raw samples it needs:
    /// its query-time ranges and the panes of the kept outputs not kept yet.
    pub fn raw_ranges(&self, query: &str, time_ms: i64, store: &SummaryStore<Batch>)
        -> Result<(usize, Vec<RawRange>), Unanswered>;
    /// Runs `root` at `time_ms`; builds missing kept panes from `raw`, keeps
    /// those whose end is at or before `watermark_ms` (the precompute
    /// engine's watermark, `None` before any sample), reuses kept ones, then
    /// evicts. Now `&mut self`, with the new `watermark_ms` parameter.
    pub fn evaluate(&mut self, root: usize, time_ms: i64, store: &SummaryStore<Batch>,
        watermark_ms: Option<i64>, raw: &BTreeMap<u64, Vec<RawSample>>)
        -> Result<Vec<ResultSample>, Unanswered>;
}

// control_plane/src/lib.rs
/// What this backend's data plane can execute and keep; now with
/// `query_time_results_retained: true`.
pub fn backend_capabilities() -> DeploymentCapabilities;

Before this PR

The B3 workload planned with query_time_results_retained: false (the old default) gets P2, one KLL rebuilt over the whole 5-minute window at every evaluation. A plan that does contain B3 fails to install:

node N is kept across evaluations (MatAtQueryKeptAcrossEvaluations), which the data plane does not support

After this PR

The default deployment gets B3 (P3-m2) for that workload: five 1-minute KLL panes built at query time and kept. The first evaluation fetches five 1-minute raw ranges from Prometheus. Each later one fetches only the newest minute and reuses the other four panes. Every answer equals the Planner executor's.

Usage

// Example 3b's p99 panel over ten series sampled every millisecond.
let workload: PlanningWorkload = serde_json::from_str(
    &std::fs::read_to_string("examples/planner-layering-example4-b3.json")?)?;
// The default deployment keeps query-time results across evaluations.
let plan = plan_workload(&workload, &DeploymentInputs::default())?;
let dag = &plan.dag.dag;
// Nothing is built at ingestion time; five kept 1-minute panes share one definition.
let sds = Sds::from_dag(dag, &frontier_from_timing(dag)?)?;
// definitions=0 kept_definitions=1 kept_offsets_ms=[0, 60000, 120000, 180000, 240000]
let mut query = QueryEngine::new(&plan, &sds);
let store = SummaryStore::new(); // no ingestion-time panes
for minute in [10, 11] {
    let t = minute * 60_000;
    // The raw ranges of the kept panes no evaluation has kept yet.
    let (root, ranges) = query.raw_ranges(&plan.queries[0], t, &store)?;
    // Fetch them (here from memory; the data plane asks Prometheus).
    let raw = /* each range's samples, by RawRange::node */;
    // Every pane read is closed: the watermark is past 11 min.
    let result = query.evaluate(root, t, &store, Some(12 * 60_000), &raw)?;
}
// T=10min raw_ranges=5 value=95 KeptResultStats { built: 5, reused: 0 }
// T=11min raw_ranges=1 value=94 KeptResultStats { built: 6, reused: 4 }

The output comments come from running this as a scratch test on this PR's tree, with one series whose nth 15-second sample is 25n mod 101. Over HTTP, the same flow runs in data_plane/tests/e2e_examples.rs::example4_b3 against the data_plane binary.

End-to-end numbers

cargo test -p data_plane --test e2e_examples -- --nocapture, Example 4 B3 (two series, one sample every 15 s up to 13 min, so every pane read is closed; evaluated at 10, 11 and 12 min):

T	series	plan	executor	difference	raw ranges fetched
10 min	a / b	95 / 195	95 / 195	0	5
11 min	a / b	94 / 194	94 / 194	0	1
12 min	a / b	93 / 193	93 / 193	0	1

The other examples select the same plans as before (P1, P485, P2, P3-m1, P4125, P3-m1) and give the same answers. 3a/4a still differ by at most 1, as in #822.

Planner notes (no Planner change)

• Under the default cost model, the Planner selects B3 only for a few series at a high sample rate. For the Planner's own Pattern B (1M series, 66,667 rows/s), B3 costs 8,884.5/s, against 2.1/s for rebuilding the whole-window KLL, because cost model 1 charges for the four kept panes of 1M KLLs every second. No Planner test selects B3 under any setting; planner_layering_example4.rs only checks that it is offered. Swept over 1–1M series at a consistent rate (series / interval): B3 never wins at 15 s or 1 s intervals. At a 1 ms interval it wins from 10 to 100 series, which is the example added here.
• Without raw data B3 is invalid (it scans raw samples at query time), so Example 3b/4b without raw data still select B1.

Limitations

• Kept panes are in memory only and are lost on restart; the first evaluation after a restart rebuilds them.
• A pane is kept once the data plane's watermark passes its end. A sample that reaches Prometheus after that, but not the data plane (beyond --allowed-lateness-ms), is not in the kept pane, as for stored panes.
• The watermark is the newest sample over all metrics, as in the precompute engine.
• Range queries are still forwarded.

Verification

Tests written first; each failed against interface stubs before the implementation (the kept answers matched, then the stats assertion failed with built: 0; the control plane test failed on the old default).

• data_plane/tests/kept.rs (3 tests): a pane read before it closes is not kept, is computed again by the next evaluation including a late sample (answer changes, equals the reference), and is kept once the watermark reaches its end; with no watermark nothing is kept. (a) and (b), the B3 plan answers like the reference at 10 and 11 min; the second evaluation reads one raw range and reuses four panes (KeptResultStats { built: 6, reused: 4 }); repeating it recomputes nothing. (c), eviction at 11, 12 and 20 min and after an older evaluation.
• data_plane/tests/sds.rs: kept panes share one definition with offsets 0..4 min; a kept output without a pane is UnsupportedShape (replaces the rejection test).
• control_plane/tests/plan_workload.rs: the default capabilities keep query-time results, and the B3 workload selects five kept SummaryAgg panes.
• data_plane/tests/e2e_examples.rs::example4_b3: the binary over HTTP, three evaluations vs the executor, raw ranges fetched [5, 1, 1].
• cargo test -p control_plane -p data_plane (all pass, e2e 6 tests in 382 s debug), cargo fmt --all -- --check, cargo clippy --workspace --all-targets --locked -- -D warnings.