feat(data-plane): sharded ingestion workers, a query thread pool and a shared SDS

README.md

@@ -35,7 +35,7 @@ The backend is being rebuilt on the Planner's physical DAG:
 | Data plane: load and validate `plan.json`, accept Remote Write | done |
 | 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: run query-time nodes over stored panes and raw data | planned |
+| Query engine: answer planned instant queries from stored panes and raw samples | done |
 
 The data plane stores the plan's `stored_outputs`. Outputs that differ only
 in which pane they read share one stored summary, kept for the largest
@@ -53,9 +53,41 @@ are dropped. Remote Write gets 503 when admitting it would leave more than
 stopped, which also makes `/-/ready` fail. `/api/v1/status/asap` reports the
 engine's sample and pane counters as JSON.
 
-Until the query engine lands, the data plane forwards every query to
-Prometheus, with its end-to-end headers such as `Authorization` and
-`X-Scope-OrgID`, and marks the response `x-asap-execution: forwarded`. SQL and
+Series are sharded by their label set across `--ingestion-workers` (default
+one per core); with several workers, a pane closes once the newest sample of
+every worker that has received one has passed it. Planned queries run on
+`--query-threads` (default one per core), one query per thread, and each
+sees the panes of every write acknowledged before it. The data plane uses
+jemalloc.
+
+A producer can close panes without waiting for later samples:
+`POST /api/v1/precompute/watermark?time_ms=T` states that every sample at or
+before `T` has been written. After the writes acknowledged before it have
+been ingested, every pane ending at or before `T` (capped at the wall clock)
+is closed, so the last pane of a series that stopped sending is built and
+served. A repeated or lower `T` does nothing and input stays open. A sample
+that arrives later for a pane the barrier closed (one ending at or before
+`T`) counts as late; a later sample in a pane still open is accepted, even if
+it is at or before `T`. The response is `{"barrier_ms": B}`, the barrier in
+force: `T` capped at the wall clock, or an earlier higher barrier. With several producers,
+send the barrier only after all of them have written through `T`.
+
+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. A summary the plan builds in
+a query run and keeps (`MatAtQueryKeptAcrossEvaluations`) is built from its
+coverage's raw samples the first time an evaluation needs it, stored in the
+SDS under the same key as ingestion-built summaries once its coverage is
+closed (its end at or before the watermark when the evaluation chose its
+fetch), reused by later evaluations, and evicted like ingestion-built
+summaries: the plan's `retention_ms` plus one pane, measured from the newest
+closed coverage at or before the watermark. `/api/v1/status/asap` reports
+`query_run` counters, including summaries kept 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
 ClickHouse are not supported yet.
 
 ## Running
@@ -71,8 +103,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 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

control_plane/src/lib.rs

@@ -82,14 +82,14 @@ 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
-/// query-time result kept across evaluations: the SDS describes such
-/// outputs, but nothing builds them until the query engine does.
+/// Planner executor builds, raw samples kept in Prometheus, and query-time
+/// results kept across evaluations (the query engine builds query-run
+/// summaries on first use and keeps them 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,20 @@ fn load_rejects_a_plan_without_a_query_per_root() {
     ));
     std::fs::remove_file(path).unwrap();
 }
+
+/// 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 summaries built in a query run and kept.
+#[test]
+fn selects_summaries_kept_across_evaluations() {
+    let inputs = DeploymentInputs::default();
+    assert!(inputs.deployment_capabilities.query_time_results_retained);
+    let plan = plan_workload(&example("example4-b3"), &inputs).unwrap();
+    let kept: Vec<_> = plan
+        .stored_outputs
+        .iter()
+        .filter(|o| o.built_at == asap_planner::StoredOutputBuild::QueryRun)
+        .collect();
+    assert_eq!(kept.len(), 5, "five one-minute summaries");
+    assert!(kept.iter().all(|o| o.pane.width_ms == 60_000));
+}

data_plane/Cargo.toml

@@ -7,6 +7,7 @@ edition.workspace = true
 asap-planner.workspace = true
 asap-executor.workspace = true
 sha2 = "0.10"
+tikv-jemallocator = "0.6"
 serde.workspace = true
 serde_json.workspace = true
 thiserror.workspace = true
@@ -20,8 +21,10 @@ reqwest.workspace = true
 prost = "0.13"
 snap = "1.1"
 futures = "0.3"
+form_urlencoded = "1"
 
 [dev-dependencies]
 control_plane.workspace = true
 tower = { version = "0.5", features = ["util"] }
 http-body-util = "0.1"
+serde_json.workspace = true