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

README.md

@@ -35,6 +35,7 @@ The backend is being rebuilt on the Planner's physical DAG:
 | Stored summaries (SDS): one definition per computation, panes by end time | done |
 | Precompute engine: build ingestion-time outputs once per tumbling pane | done |
 | Query engine: answer planned instant queries from stored panes and raw samples | done |
+| Query engine: keep query-time panes across evaluations (`MatAtQueryKeptAcrossEvaluations`) | done |
 
 Panes close `--allowed-lateness-ms` (default 15 s) after the newest sample
 passes their end; later samples for them are dropped. The first pane after
@@ -43,7 +44,14 @@ startup is skipped because it may be partial.
 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 raw ranges are read from Prometheus. Anything else is forwarded
+query-time raw ranges are read from Prometheus. An input the plan keeps
+across evaluations (`MatAtQueryKeptAcrossEvaluations`) is computed from its
+pane's raw samples when an evaluation reads it and no evaluation has kept
+it. It is kept in memory, by definition and pane end, once the pane is
+closed by the precompute engine's rule (its end at or before the newest
+sample minus `--allowed-lateness-ms`), and later evaluations reuse it; a pane
+read before it closes is computed again by later evaluations. Kept panes are
+evicted like stored ones, measured from the newest evaluation. Anything else is forwarded
 to Prometheus (`x-asap-execution: forwarded`); a planned query that could not
 be answered also says why in `x-asap-forward-reason`. Range queries, SQL and
 ClickHouse are not supported yet.
@@ -61,17 +69,18 @@ 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 query-time result kept across evaluations
-(`MatAtQueryKeptAcrossEvaluations` is not implemented). For example
+in Prometheus, and query-time results kept across evaluations. 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
 with cost model 1 and illustrative unit costs.
 
 ## Examples
 
-`examples/` holds #509's PromQL example workloads (1, 3a, 3b, 4a, 4b) as
-Planner `PlanningWorkload`s. `data_plane/tests/e2e_examples.rs` plans each
+`examples/` holds #509's PromQL example workloads (1, 3a, 3b, 4a, 4b, and
+`4-b3`: Example 3b's panel over ten series sampled every millisecond, for
+which the Planner keeps query-time panes across evaluations) as Planner
+`PlanningWorkload`s. `data_plane/tests/e2e_examples.rs` plans each
 one, runs the `data_plane` binary against a Prometheus holding generated
 samples, Remote Writes the same samples, and checks every planned query
 against the Planner executor running the selected plan with nothing

control_plane/src/lib.rs

@@ -224,14 +224,14 @@ pub fn plan_selected_at_query_time(
 }
 
 /// What this backend's data plane can execute and keep: every summary the
-/// Planner executor builds, raw samples kept in Prometheus, and no
-/// query-time result kept across evaluations (the data plane does not
-/// implement `MatAtQueryKeptAcrossEvaluations`).
+/// Planner executor builds, raw samples kept in Prometheus, and query-time
+/// results kept across evaluations in memory
+/// (`MatAtQueryKeptAcrossEvaluations`).
 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

@@ -99,3 +99,37 @@ fn reference_plan_materializes_nothing() {
     assert_eq!(materialized(&reference), 0);
     assert_eq!(reference.queries, installed.queries);
 }
+
+/// The backend keeps query-time results across evaluations, so for a few
+/// series sampled every millisecond the Planner selects #509 Example 4's
+/// B3: five 1-minute KLL panes built at query time and kept, installed as
+/// plain panes.
+#[test]
+fn selects_panes_kept_across_evaluations() {
+    let inputs = DeploymentInputs::default();
+    assert!(inputs.deployment_capabilities.query_time_results_retained);
+    let path =
+        Path::new(env!("CARGO_MANIFEST_DIR")).join("../examples/planner-layering-example4-b3.json");
+    let workload: PlanningWorkload =
+        serde_json::from_str(&std::fs::read_to_string(path).unwrap()).unwrap();
+    let plan = plan_workload(&workload, &inputs).unwrap();
+    let kept = plan
+        .dag
+        .dag
+        .nodes
+        .iter()
+        .filter(|n| {
+            matches!(
+                n.materialization,
+                Materialization::MatAtQueryKeptAcrossEvaluations(_)
+            )
+        })
+        .inspect(|n| {
+            assert!(matches!(
+                n.payload,
+                Operator::ASAP(ASAPOp::SummaryAgg { .. })
+            ))
+        })
+        .count();
+    assert_eq!(kept, 5, "five one-minute panes");
+}

data_plane/src/engine.rs

@@ -135,6 +135,8 @@ impl Engine {
         tracing::info!(
             definitions = sds.definitions.len(),
             outputs = sds.outputs.len(),
+            kept_definitions = sds.kept_definitions.len(),
+            kept_outputs = sds.kept_outputs.len(),
             "stored summaries"
         );
         let query = QueryEngine::new(&plan, &sds);
@@ -172,6 +174,7 @@ impl Engine {
                         root,
                         time_ms,
                         &self.precompute.store,
+                        self.precompute.watermark_ms(),
                         &raw,
                     ));
                 }

data_plane/src/precompute.rs

@@ -29,7 +29,7 @@ use planner_types::ir::operator::operator_properties::Source as ScanSource;
 use planner_types::ir::physical_export::PhysicalASAPDAG;
 use planner_types::ir::{NonASAPOp, Operator as IrOperator};
 
-use crate::sds::{DefinitionId, Sds, SummaryDefinition, SummaryStore};
+use crate::sds::{DefinitionId, PaneBinding, Sds, SummaryDefinition, SummaryStore};
 
 /// One raw sample: its series' full label set (including `__name__`), time
 /// and value.
@@ -69,13 +69,75 @@ pub struct ScanInput {
     pub schema: SchemaRef,
 }
 
-struct Builder {
-    definition: SummaryDefinition,
-    scan: ScanInput,
+/// How the panes of one definition are built from raw samples. The
+/// precompute engine builds stored panes with it, and the query engine kept
+/// ones.
+pub(crate) struct PaneBuild {
+    pub(crate) definition: SummaryDefinition,
+    pub(crate) scan: ScanInput,
     build: CompiledPhysicalDAG,
-    /// Oldest pane offset any output reads, plus one pane: panes ending
-    /// before `watermark - retention_ms` are no longer read.
-    retention_ms: i64,
+    /// Largest offset any output of the definition reads, plus two panes:
+    /// panes ending at or before `newest - retention_ms` are no longer read.
+    pub(crate) retention_ms: i64,
+}
+
+impl PaneBuild {
+    /// Compiles `definition`'s build from its `TimeRange` up to its builder;
+    /// `outputs` are the outputs that read it.
+    pub(crate) fn new(
+        dag: &PhysicalASAPDAG,
+        definition: &SummaryDefinition,
+        outputs: &BTreeMap<u64, (DefinitionId, PaneBinding)>,
+    ) -> Result<Self, PrecomputeError> {
+        let error =
+            |message: &str| PrecomputeError::Definition(definition.id.clone(), message.into());
+        let scan = scan_below(dag, definition.raw_input)
+            .ok_or_else(|| error("no time-series Scan below its TimeRange"))?;
+        let raw_schema = Arc::new(output_schema(dag, definition.raw_input).clone());
+        if raw_schema != scan.schema {
+            return Err(error("TimeRange rows differ from its Scan's"));
+        }
+        let build = compile(
+            dag,
+            BTreeMap::from([(definition.raw_input, InputContract::bounded(raw_schema))]),
+            &[definition.builder],
+        )?;
+        let max_offset = outputs
+            .values()
+            .filter(|(id, _)| id == &definition.id)
+            .map(|(_, binding)| binding.offset_ms)
+            .max()
+            .unwrap_or(0);
+        Ok(Self {
+            definition: definition.clone(),
+            scan,
+            build,
+            retention_ms: max_offset + 2 * definition.pane_ms,
+        })
+    }
+
+    /// Runs the plan's `Scan` over `samples`, then the build over the rows
+    /// it selects, under `scope`.
+    pub(crate) fn run(
+        &self,
+        dag: &PhysicalASAPDAG,
+        samples: &[RawSample],
+        scope: Scope,
+    ) -> Result<Batch, PrecomputeError> {
+        let selected = select(dag, &self.scan, samples, scope.clone())?;
+        let input: Source<'_> =
+            Box::new(Operator::source(self.scan.schema.clone(), vec![selected])?);
+        let build = self
+            .build
+            .instantiate(BTreeMap::from([(self.definition.raw_input, input)]))?;
+        let rows = run(&build, self.build.roots(), scope)?;
+        let schema = Arc::new(self.definition.output_schema.clone());
+        Ok(Batch::try_new(schema, rows)?)
+    }
+}
+
+struct Builder {
+    pane: PaneBuild,
     /// Samples of open panes, by pane end.
     open: BTreeMap<i64, Vec<RawSample>>,
     /// End of the newest pane built or skipped; samples in it or before it
@@ -99,37 +161,17 @@ impl Precompute {
         sds: &Sds,
         allowed_lateness_ms: i64,
     ) -> Result<Self, PrecomputeError> {
-        let mut builders = Vec::new();
-        for definition in sds.definitions.values() {
-            let error =
-                |message: String| PrecomputeError::Definition(definition.id.clone(), message);
-            let scan = scan_below(&dag, definition.raw_input)
-                .ok_or_else(|| error("no time-series Scan below its TimeRange".into()))?;
-            let raw_schema = Arc::new(output_schema(&dag, definition.raw_input).clone());
-            if raw_schema != scan.schema {
-                return Err(error("TimeRange rows differ from its Scan's".into()));
-            }
-            let build = compile(
-                &dag,
-                BTreeMap::from([(definition.raw_input, InputContract::bounded(raw_schema))]),
-                &[definition.builder],
-            )?;
-            let max_offset = sds
-                .outputs
-                .values()
-                .filter(|(id, _)| id == &definition.id)
-                .map(|(_, binding)| binding.offset_ms)
-                .max()
-                .unwrap_or(0);
-            builders.push(Builder {
-                definition: definition.clone(),
-                scan,
-                build,
-                retention_ms: max_offset + 2 * definition.pane_ms,
-                open: BTreeMap::new(),
-                built_through: None,
-            });
-        }
+        let builders = sds
+            .definitions
+            .values()
+            .map(|definition| {
+                Ok(Builder {
+                    pane: PaneBuild::new(&dag, definition, &sds.outputs)?,
+                    open: BTreeMap::new(),
+                    built_through: None,
+                })
+            })
+            .collect::<Result<_, PrecomputeError>>()?;
         Ok(Self {
             dag,
             builders,
@@ -153,10 +195,10 @@ impl Precompute {
             let metric = sample.labels.get("__name__").map(String::as_str);
             let mut accepted = false;
             for builder in &mut self.builders {
-                if metric != Some(builder.scan.metric.as_str()) {
+                if metric != Some(builder.pane.scan.metric.as_str()) {
                     continue;
                 }
-                let end = pane_end(sample.timestamp_ms, builder.definition.pane_ms);
+                let end = pane_end(sample.timestamp_ms, builder.pane.definition.pane_ms);
                 // The first pane seen may be partial: skip it.
                 let built = *builder.built_through.get_or_insert(end);
                 if end <= built {
@@ -182,27 +224,33 @@ impl Precompute {
             let Some(built) = builder.built_through else {
                 continue;
             };
-            let pane_ms = builder.definition.pane_ms;
+            let pane_ms = builder.pane.definition.pane_ms;
             // Every pane after the first through the watermark is built, so
             // a pane without samples is stored as empty, not missing.
             let mut end = built + pane_ms;
             while end <= watermark {
                 let samples = builder.open.remove(&end).unwrap_or_default();
-                match build_pane(&self.dag, builder, end, &samples) {
+                let scope = Scope::Ingestion {
+                    window_start_ms: end - pane_ms,
+                    window_end_ms: end,
+                    revision: 1,
+                };
+                match builder.pane.run(&self.dag, &samples, scope) {
                     Ok(state) => {
-                        self.store.insert(&builder.definition.id, end, state);
+                        self.store.insert(&builder.pane.definition.id, end, state);
                         self.stats.panes_built += 1;
                     }
                     Err(error) => {
-                        tracing::error!(definition = ?builder.definition.id, end, %error, "pane build failed");
+                        tracing::error!(definition = ?builder.pane.definition.id, end, %error, "pane build failed");
                         self.stats.panes_failed += 1;
                     }
                 }
                 builder.built_through = Some(end);
                 end += pane_ms;
             }
+            let definition = &builder.pane.definition.id;
             self.store
-                .evict_through(&builder.definition.id, watermark - builder.retention_ms);
+                .evict_through(definition, watermark - builder.pane.retention_ms);
         }
     }
 }
@@ -214,33 +262,6 @@ pub fn pane_end(timestamp_ms: i64, pane_ms: i64) -> i64 {
     -(-timestamp_ms).div_euclid(pane_ms) * pane_ms
 }
 
-/// Runs the plan's `Scan` over `samples`, then the definition's build over
-/// the rows it selects, for the pane ending at `end`.
-fn build_pane(
-    dag: &PhysicalASAPDAG,
-    builder: &Builder,
-    end: i64,
-    samples: &[RawSample],
-) -> Result<Batch, PrecomputeError> {
-    let start = end - builder.definition.pane_ms;
-    let scope = Scope::Ingestion {
-        window_start_ms: start,
-        window_end_ms: end,
-        revision: 1,
-    };
-    let selected = select(dag, &builder.scan, samples, scope.clone())?;
-    let input: Source<'_> = Box::new(Operator::source(
-        builder.scan.schema.clone(),
-        vec![selected],
-    )?);
-    let build = builder
-        .build
-        .instantiate(BTreeMap::from([(builder.definition.raw_input, input)]))?;
-    let rows = run(&build, builder.build.roots(), scope)?;
-    let schema = Arc::new(builder.definition.output_schema.clone());
-    Ok(Batch::try_new(schema, rows)?)
-}
-
 /// The rows of `samples` the plan's `scan` selects: its label matchers
 /// applied by the executor's `Scan`.
 pub fn select(