feat(query): build stored outputs built in a QueryRun on first use

README.md

@@ -54,9 +54,19 @@ stopped, which also makes `/-/ready` fail. `/api/v1/status/asap` reports the
 engine's sample and pane counters as JSON.
 
 An instant query whose text is one of the plan's queries is answered by the
-plan (`x-asap-execution: plan`). Its stored inputs read the panes ending at
-`T - offset`, so `T` must be on the pane grid and those panes stored; its
-query-time bound raw-sample windows are read from Prometheus. Anything else is forwarded
+plan (`x-asap-execution: plan`) at its evaluation time `T`. Stored outputs
+built in an `IngestionRun` are read from the SDS as the panes ending at
+`T - offset`, so `T` must be on the pane grid and those panes stored. The
+plan's query-time `RawSampleWindow`s are read from Prometheus. A stored output
+built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`) is built from its
+pane's raw samples the first time an evaluation needs it. It is stored in the
+SDS under the same key once its pane is closed, that is, once its end is at or
+before the watermark at the time the evaluation chose its fetch. Later
+evaluations reuse it. It is evicted like an `IngestionRun` output: the plan's
+`retention_ms` plus one pane, measured from the newest closed pane at or
+before the watermark. `/api/v1/status/asap` reports `query_run` counters,
+including summary states stored while their own metric lagged behind the
+global watermark. Anything else is forwarded
 to Prometheus with its end-to-end headers, such as `Authorization` and
 `X-Scope-OrgID` (`x-asap-execution: forwarded`); a planned query that could not
 be answered also says why in `x-asap-forward-reason`. Range queries, SQL and
@@ -75,8 +85,7 @@ cargo run -p data_plane -- --plan plan.json --prometheus-url http://localhost:90
 
 `--deployment-capabilities caps.json` replaces the default
 `DeploymentCapabilities`: every summary the Planner can build, raw data kept
-in Prometheus, and no stored outputs built in a `QueryRun`
-(`MatAtQueryKeptAcrossEvaluations`). For example
+in Prometheus, and stored outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`). For example
 `{"raw_data_retained": false}` makes the Planner build every input at
 ingestion time. `--measured-operation-costs costs.json` gives Stage 3 the
 deployment's measured sketch operation costs; otherwise it prices every node

control_plane/src/lib.rs

@@ -82,14 +82,15 @@ pub fn load_plan(path: &Path) -> Result<InstallablePlan, PlanError> {
 }
 
 /// What this backend's data plane can execute and keep: every summary the
-/// Planner executor builds, raw samples kept in Prometheus, and no
-/// stored outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`): the SDS describes such
-/// outputs, but nothing builds them until the query engine does.
+/// Planner executor builds, raw samples kept in Prometheus, and stored
+/// outputs built in a `QueryRun` (`MatAtQueryKeptAcrossEvaluations`): the
+/// query engine builds their summary states on first use and stores them in
+/// the SDS once their coverage is closed.
 pub fn backend_capabilities() -> DeploymentCapabilities {
     DeploymentCapabilities {
         supported_summaries: None,
         raw_data_retained: true,
-        query_time_results_retained: false,
+        query_time_results_retained: true,
     }
 }
 

control_plane/tests/plan_workload.rs

@@ -88,3 +88,21 @@ fn load_rejects_a_plan_without_a_query_per_root() {
     ));
     std::fs::remove_file(path).unwrap();
 }
+
+/// The backend retains query-time results (`query_time_results_retained`),
+/// so for a few series sampled every millisecond the Planner selects #509
+/// Example 4's B3: five 1-minute KLL panes, stored outputs built in a
+/// `QueryRun` (`MatAtQueryKeptAcrossEvaluations`).
+#[test]
+fn selects_stored_outputs_built_in_a_query_run() {
+    let inputs = DeploymentInputs::default();
+    assert!(inputs.deployment_capabilities.query_time_results_retained);
+    let plan = plan_workload(&example("example4-b3"), &inputs).unwrap();
+    let query_run: Vec<_> = plan
+        .stored_outputs
+        .iter()
+        .filter(|o| o.built_at == asap_planner::StoredOutputBuild::QueryRun)
+        .collect();
+    assert_eq!(query_run.len(), 5, "five one-minute summaries");
+    assert!(query_run.iter().all(|o| o.pane.width_ms == 60_000));
+}

data_plane/src/engine.rs

@@ -12,7 +12,9 @@ use asap_planner::InstallablePlan;
 use tokio::sync::oneshot;
 
 use crate::precompute::{Precompute, PrecomputeOptions, PrecomputeStats, RawSample};
-use crate::query::{BoundRawSampleWindow, QueryEngine, ResultSample, Unanswered};
+use crate::query::{
+    IngestionProgress, Prepared, QueryEngine, QueryRunStats, ResultSample, Unanswered,
+};
 use crate::sds::Sds;
 
 /// How the engine thread ingests samples.
@@ -42,11 +44,12 @@ enum Command {
     BoundRawSampleWindows {
         query: String,
         time_ms: i64,
-        reply: Reply<(usize, Vec<BoundRawSampleWindow>)>,
+        reply: Reply<Prepared>,
     },
     Evaluate {
         root: usize,
         time_ms: i64,
+        progress: IngestionProgress,
         raw: BTreeMap<u64, Vec<RawSample>>,
         reply: Reply<Vec<ResultSample>>,
     },
@@ -75,6 +78,8 @@ pub struct EngineHandle {
     running: Arc<AtomicBool>,
     /// The counters as of the last command the engine ran.
     stats: Arc<Mutex<PrecomputeStats>>,
+    /// The stored output built in a `QueryRun` counters as of the last evaluation.
+    query_run_stats: Arc<Mutex<QueryRunStats>>,
 }
 
 /// Marks the engine stopped when its thread exits, panics included.
@@ -96,9 +101,11 @@ impl EngineHandle {
         let running = Arc::new(AtomicBool::new(true));
         let queued_samples = Arc::new(AtomicU64::new(0));
         let stats = Arc::new(Mutex::new(PrecomputeStats::default()));
+        let query_run_stats = Arc::new(Mutex::new(QueryRunStats::default()));
         let shared = Shared {
             queued_samples: queued_samples.clone(),
             stats: stats.clone(),
+            query_run_stats: query_run_stats.clone(),
         };
         let guard = RunningGuard(running.clone());
         std::thread::Builder::new()
@@ -122,6 +129,7 @@ impl EngineHandle {
             max_queued_samples: options.max_queued_samples,
             running,
             stats,
+            query_run_stats,
         };
         Ok((handle, queries))
     }
@@ -158,7 +166,7 @@ impl EngineHandle {
         &self,
         query: String,
         time_ms: i64,
-    ) -> Result<(usize, Vec<BoundRawSampleWindow>), Unanswered> {
+    ) -> Result<Prepared, Unanswered> {
         self.ask(|reply| Command::BoundRawSampleWindows {
             query,
             time_ms,
@@ -167,22 +175,32 @@ impl EngineHandle {
         .await
     }
 
-    /// Runs a root over stored panes and `raw`; see [`QueryEngine::evaluate`].
+    /// Runs a prepared root over stored summaries and `raw`; see
+    /// [`QueryEngine::evaluate`].
     pub async fn evaluate(
         &self,
-        root: usize,
+        prepared: Prepared,
         time_ms: i64,
         raw: BTreeMap<u64, Vec<RawSample>>,
     ) -> Result<Vec<ResultSample>, Unanswered> {
         self.ask(|reply| Command::Evaluate {
-            root,
+            root: prepared.root,
             time_ms,
+            progress: prepared.progress,
             raw,
             reply,
         })
         .await
     }
 
+    /// The stored output built in a `QueryRun` counters as of the last evaluation.
+    pub fn query_run_stats(&self) -> QueryRunStats {
+        *self
+            .query_run_stats
+            .lock()
+            .unwrap_or_else(|e| e.into_inner())
+    }
+
     async fn ask<T>(&self, command: impl FnOnce(Reply<T>) -> Command) -> Result<T, Unanswered> {
         let stopped = || Unanswered::Execution("the engine has stopped".into());
         let (reply, answer) = oneshot::channel();
@@ -232,6 +250,7 @@ fn admit(queued: u64, count: u64, max_queued: u64) -> Result<u64, IngestError> {
 struct Shared {
     queued_samples: Arc<AtomicU64>,
     stats: Arc<Mutex<PrecomputeStats>>,
+    query_run_stats: Arc<Mutex<QueryRunStats>>,
 }
 
 struct Engine {
@@ -249,18 +268,6 @@ impl Engine {
             outputs = sds.outputs.len(),
             "stored summaries"
         );
-        let query_run = sds
-            .definitions
-            .values()
-            .filter(|d| d.built_at == asap_planner::StoredOutputBuild::QueryRun)
-            .count();
-        if query_run > 0 {
-            tracing::warn!(
-                definitions = query_run,
-                "the plan has stored outputs built in a QueryRun, which this data plane does not build yet; \
-                 queries that read them are forwarded"
-            );
-        }
         let queries = plan.queries.iter().map(|q| q.query.clone()).collect();
         let query = QueryEngine::new(&plan, &sds);
         let engine = Self {
@@ -289,20 +296,30 @@ impl Engine {
                         &query,
                         time_ms,
                         &self.precompute.store,
+                        self.precompute.progress(),
                     ));
                 }
                 Command::Evaluate {
                     root,
                     time_ms,
+                    progress,
                     raw,
                     reply,
                 } => {
-                    let _ = reply.send(self.query.evaluate(
+                    let answer = self.query.evaluate(
                         root,
                         time_ms,
-                        &self.precompute.store,
+                        &mut self.precompute.store,
+                        &progress,
                         &raw,
-                    ));
+                    );
+                    // Published before the reply, so a caller that awaited
+                    // the evaluation reads its counters.
+                    *shared
+                        .query_run_stats
+                        .lock()
+                        .unwrap_or_else(|e| e.into_inner()) = self.query.query_run_stats;
+                    let _ = reply.send(answer);
                 }
                 Command::Stop => break,
             }

data_plane/src/lib.rs

@@ -176,6 +176,7 @@ async fn status(State(state): State<Arc<DataPlane>>) -> Response {
         "engine_running": state.engine.is_running(),
         "samples_queued": state.engine.queued_samples(),
         "precompute": state.engine.stats(),
+        "query_run": state.engine.query_run_stats(),
     });
     (
         [(header::CONTENT_TYPE, "application/json")],
@@ -249,10 +250,11 @@ async fn answer(
             .duration_since(std::time::UNIX_EPOCH)
             .map_or(0, |d| d.as_millis() as i64)
     });
-    let (root, ranges) = state
+    let prepared = state
         .engine
         .bound_raw_sample_windows(query.query, time_ms)
         .await?;
+    let ranges = &prepared.windows;
     let mut fetch_headers = forwarded_headers(headers);
     // The raw read is a GET without the caller's body.
     fetch_headers.remove(header::CONTENT_TYPE);
@@ -267,7 +269,7 @@ async fn answer(
         .map_err(|e| Unanswered::Execution(format!("fetching raw samples: {e}")))?
         .into_iter()
         .collect();
-    let samples = state.engine.evaluate(root, time_ms, raw).await?;
+    let samples = state.engine.evaluate(prepared, time_ms, raw).await?;
     Ok(prometheus::vector_response(time_ms, &samples))
 }