feat(query): answer planned instant queries by running the plan's roots

[zzylol]

Stack 4/5. Base: #820. Next: #822.

Why

Every query was forwarded to Prometheus, so the selected plan never answered anything.

What

A planned instant query is answered by running its root of the DAG at evaluation time T:

• Stored inputs: each reads the pane of its definition ending at T − offset.
• Query-time raw inputs (TimeRange ← TimeShift* ← Scan): the data plane fetches (T − offset − range, T − offset] from Prometheus and filters it with the plan's Scan.
• Responses: an answered query returns a Prometheus vector with x-asap-execution: plan. Any other query, including range queries, is forwarded with x-asap-execution: forwarded. A planned query that can't be answered gives its reason in x-asap-forward-reason: a missing pane, T off the pane grid, or an unsupported shape.
• MatAtQuery nodes run like NotMaterialized ones: each instant query is its own evaluation, so there is nothing to share within it.

How

Each root is compiled once at install time with asap_executor::physical_planner::compile, its stored outputs and raw TimeRanges as bounded input contracts. At query time the engine thread checks the panes, the HTTP side fetches the raw ranges, and the engine instantiates the compiled root over both and runs it under query scope.

Key code interfaces added / modified

// data_plane/src/query.rs
pub struct RawRange {
    /// The `TimeRange` node the samples feed.
    pub node: u64,
    /// The `Scan`'s metric.
    pub metric: String,
    /// Exclusive start.
    pub start_ms: i64,
    /// Inclusive end.
    pub end_ms: i64,
}

pub struct ResultSample {
    /// Series labels, from the series identity and the other text columns.
    pub labels: BTreeMap<String, String>,
    /// The one `Float64` column.
    pub value: f64,
}

/// Why a query is not answered; it is then forwarded.
pub enum Unanswered {
    /// The query text is not one of the plan's.
    NotPlanned(String),
    /// The root does not compile or its result rows are not a vector.
    Unsupported(String),
    /// A stored pane is missing or `T` is off the pane grid.
    MissingPane(String),
    /// Fetching raw samples or execution failed.
    Execution(String),
}

pub struct QueryEngine { /* private: each query with its compiled root */ }
impl QueryEngine {
    /// Compiles every root of `plan` against the stored outputs of `sds`.
    pub fn new(plan: &InstalledPlan, sds: &Sds) -> Self;
    /// The root answering `query` at `time_ms` and the raw samples it needs.
    pub fn raw_ranges(&self, query: &str, time_ms: i64, store: &SummaryStore<Batch>)
        -> Result<(usize, Vec<RawRange>), Unanswered>;
    /// Runs `root` at `time_ms` over stored panes and `raw` (samples by `RawRange::node`).
    pub fn evaluate(&self, root: usize, time_ms: i64, store: &SummaryStore<Batch>,
        raw: &BTreeMap<u64, Vec<RawSample>>) -> Result<Vec<ResultSample>, Unanswered>;
}

// data_plane/src/lib.rs
/// `plan` or `forwarded`.
pub const EXECUTION_HEADER: &str = "x-asap-execution";
/// Why a planned query was forwarded.
pub const FORWARD_REASON_HEADER: &str = "x-asap-forward-reason";

Before this PR

Every query was forwarded to Prometheus.

After this PR

Example 4b without raw data answers quantile_over_time(0.99, latency_ms[60m]) at 70 min from six stored panes, reading no raw range; Example 1 answers both queries from raw ranges fetched from Prometheus.

Usage

// The plan, SDS and precompute engine of #819/#820's usage (Example 4b, no raw data).
let query = QueryEngine::new(&plan, &sds);
// Evaluation time: 70 min, on the 10-minute pane grid.
let t = 70 * 60_000;
// The root answering the query and the raw ranges it needs (none: all panes are stored).
let (root, ranges) = query.raw_ranges(&plan.queries[0], t, &precompute.store)?;
// Run the root over the stored panes.
let result = query.evaluate(root, t, &precompute.store, &BTreeMap::new())?;
// raw_ranges=0 result=[ResultSample { labels: {"service": "a"}, value: 237.0 }]

Over HTTP the same query returns a Prometheus vector with x-asap-execution: plan (data_plane/tests/answers.rs).

Limitations

• Only instant queries are answered.
• Stored panes need T on the pane grid.
• Panes are kept in memory only.
• Standalone scalar roots aren't supported (the control plane rejects them).

Verification

• data_plane/tests/query.rs (3 tests): stored panes answer like one query-time build, query-time plans read raw ranges, unanswerable queries report why.
• data_plane/tests/answers.rs (2 tests) over HTTP against a stand-in Prometheus: Example 1 from raw samples, including the left-open range start, and Example 4b from panes built through Remote Write.
• cargo test -p control_plane -p data_plane, fmt, clippy on this PR's tree; Usage run on this PR's tree.