feat(precompute): apply spatial filters during ingest routing (#724)

* feat(planner): load profiled atomic costs

* feat(precompute): apply spatial filters during ingest routing

* fix(precompute): validate planner spatial filters

* fix(precompute): use supported aggregation type in routing tests

Author: milindsrivastava1997

.design_docs/cost-optimizer-decision-map.md

@@ -0,0 +1,147 @@
+# Cost-based optimizer decision map
+
+Goal: turn a PromQL workload with query frequencies and dataset-conditioned
+sketch measurements into an offline deployment plan, then collect evidence that
+the selected plan improves the stated objective without violating accuracy.
+
+## Progress log
+
+- 2026-09-07: Located the existing offline planner entry point:
+  `asap-optimizer-cli`. `Controller::generate()` still takes the hardcoded
+  planner path, so production wiring is intentionally not the first milestone.
+- 2026-09-07: Verified a wire-contract gap: sketch-bench emits a versioned,
+  workload-profiled atomic-cost document, while ASAPQuery currently loads a
+  legacy flat entry array. A profile-aware loader and explicit selector are the
+  first integration change.
+- 2026-09-07: Distinguished grouping labels from sketch keys. Grouping labels
+  determine how many sketch instances the planner deploys; sketch-key
+  properties belong to a measured benchmark profile.
+- 2026-09-07: Chosen first vertical slice: KLL backing PromQL
+  `quantile_over_time`. The next unresolved concrete inputs are a source
+  metric, KLL input value/key column, trace time slice, and grouping labels.
+- 2026-09-07: Candidate external workload specs already present in
+  `sketch-bench`: Alibaba microservices CPU (`cpu_utilization`, grouped by
+  `msname`, one-minute slice); Google task CPU (`cpu_rate`, grouped by
+  `machine_id`, three-minute slice); and Datadog/BOOM (`target`, scalar,
+  one-minute slice). Existing ASAPQuery Prometheus replay configurations also
+  exist for Alibaba node CPU and Google CPU. Recommended first case: Google
+  CPU, because it has both a sketch-bench workload spec and a matching
+  Prometheus replay configuration with explicit labels.
+- 2026-09-07: Verified the selected Google trace is available locally:
+  `google-cluster-data/ClusterData2011/clusterdata-2011-2/task_usage/part-00262-of-00500.csv.gz`
+  (92 MB). The external loader supports numeric f64 grouped workloads, and the
+  selected spec uses `cpu_rate`, grouped by `machine_id`, over a three-minute
+  time slice. `kll-percall` is the correct benchmark variant because the
+  ASAPQuery runtime invokes KLL `quantile()` per query. Remaining prerequisites:
+  (1) a matching `quantile_over_time` workload and series-inventory CSV,
+  (2) profile-aware atomic-cost loading/selection in ASAPQuery, and
+  (3) explicit accuracy, arrival-rate, and exact-baseline assumptions.
+- 2026-09-07: Inspected sketch-bench PR #124. It is the correct producer-side
+  contract: a versioned document, one profile per exact external workload,
+  required `query_accuracy`, and no cross-profile merging. Decision: do not
+  add an ASAPQuery-specific `logical_metric` field to PR #124. Its
+  `value_column` identifies the physical trace column measured; a Prometheus
+  metric name is a query-engine identity that may rename or transform that
+  column. ASAPQuery owns the explicit mapping from a queried metric to a
+  benchmark profile. Do not add KLL key columns for this slice: KLL consumes
+  numeric values and keyed external workloads are intentionally unsupported.
+  ASAPQuery should mirror the versioned types, select exactly one external
+  profile from an explicit selector, and pass only its entries to the existing
+  candidate resolver.
+- 2026-09-07: Corrected an earlier naming example after inspecting collector
+  code. `google_mean_cpu_usage_rate_0` is an older experiment-specific name;
+  the existing evaluation mapper emits `google_cluster_2019_cpu_rate` from
+  raw `cpu_rate`, and creates a `_q_kll` alias only to route a second copy of
+  the same samples through KLL. These are naming/routing conventions, not
+  distinct source values. For the first slice, use one canonical raw-stream
+  identity throughout (recommended Prometheus-safe name:
+  `google_task_usage_cpu_rate`), and do not require a separate logical-metric
+  field in the sketch-bench document.
+- 2026-09-07: Validated the PR-123/PR-124 dependency analysis. PR 123 changes
+  the underlying benchmark-record schema; PR 124 is the atomic-cost consumer
+  break and must be handled via its independent document schema version. Two
+  implementation corrections: `query_accuracy` is a required map of named
+  metrics, not one `f64`; and workload selection can happen once at the
+  ASAPQuery load boundary, returning the selected profile's existing flat
+  `AtomicCostTable`. This avoids threading workload identity through every
+  candidate resolver while still rejecting zero or ambiguous matches before
+  any candidate lookup.
+- 2026-09-07: Implemented the ASAPQuery consumer seam on branch
+  `feat/profiled-atomic-cost-loader`. It mirrors PR 124's versioned document,
+  external workload, required accuracy map, strict schema validation, and
+  exact single-profile selection. The two offline optimizer CLIs now require a
+  JSON `profiles[].workload` selector whenever `--atomic-costs` is supplied;
+  selected entries preserve the existing flat resolver interface. Focused
+  loader and full planner tests pass (216 tests plus doc-tests); the latter was
+  run outside the sandbox because its existing ClickHouse mock opens a local
+  listener.
+- 2026-09-07: Review follow-up: centralized CLI profile loading behind one
+  optimizer module interface and added explicit zero-match coverage. Synthetic
+  profile descriptions remain opaque JSON because sketch-bench's data-generator
+  schema is independently versioned; document, profile, entry, and external
+  workload fields remain strict at this consumer boundary.
+
+## #1: What identifies a benchmark cost profile?
+
+Type: Discuss
+
+### Question
+
+Which dataset properties are expected to change insert, merge, query, memory,
+or accuracy enough that they must select a distinct atomic-cost profile? Decide
+the first paper-scale profile matrix and the semantics for choosing a profile at
+planning time.
+
+### Answer
+
+Open. `sketch-bench` already emits `AtomicCostDocument { schema_version,
+profiles: [{ workload, entries }] }`; the optimizer currently reads the older
+flat entry array and therefore has no profile-selection rule. The available
+observability corpus is under `../benchmarks/metrics_observability/data` from
+the workspace root (Datadog/BOOM and Alibaba traces). Candidate framing from
+the user: `(dataset_name, metric_name, keying/aggregated-label names,
+time-range)`. The remaining decision is whether labels and range identify an
+atomic measurement or instead parameterize the structural cost formula.
+
+Clarification: **grouping labels** are the PromQL `GROUP BY` labels and
+partition the metric stream into separate sketch instances. A **sketch key**
+is the value or label tuple inserted into keyed sketches such as HLL, CMS, and
+Hydra. Grouping labels and their observed distinct-group count are planner
+context; the sketch-key distribution/cardinality (and, where relevant, encoded
+key size) belong in the benchmark profile.
+
+For the first vertical slice, scope this decision to one KLL
+`quantile_over_time` workload. Do not define the full cross-dataset matrix yet.
+
+## #2: What is the minimum credible empirical planning loop?
+
+Blocked by: #1
+Type: Prototype
+
+### Question
+
+What end-to-end experiment should prove that measured costs, rather than
+hand-tuned constants, change a planner decision appropriately for a fixed
+PromQL workload and series inventory?
+
+### Answer
+
+Open. The existing offline `asap-optimizer-cli` is the intended harness after
+the document-loader/profile-selection gap is closed.
+
+## #3: What feasibility evidence constrains optimization?
+
+Blocked by: #1, #2
+Type: Discuss
+
+### Question
+
+For each query/sketch/config/dataset case, which accuracy metric and threshold
+make a candidate eligible, and how will exact-query cost and arrival rate be
+measured rather than assumed?
+
+### Answer
+
+Open. `sketch-bench` already retains capability-specific `query_accuracy`, but
+the ASAPQuery greedy optimizer does not use it; arrival rate (`rho`) and exact
+query cost are currently placeholders.

.design_docs/optimizer-v1-implementation-plan.md

@@ -322,7 +322,8 @@ cargo run -p asap_planner --bin asap-optimizer-cli -- \
   --dataset <path/to/series-inventory.csv> \
   --data-ingestion-interval-ms 60000 \
   [--rho 1.0] \
-  [--atomic-costs <path/to/atomic_costs.json>]
+  [--atomic-costs <path/to/atomic_costs.json> \
+   --atomic-cost-workload <path/to/workload-selector.json>]
 ```
 
 Takes the same `ControllerConfig` YAML format as `asap-planner --input_config`.
@@ -332,7 +333,11 @@ group count; no live Prometheus connection is needed. A `metrics:` hints block,
 present, is checked against the dataset and mismatches fail loudly.
 Prints deployed streaming configs and query configs to stdout. `--rho` is the
 placeholder arrival rate (see TODOs below — not real yet). `--atomic-costs` is
-optional; omit it and ordinary unbenchmarked candidates use the flat stub, while
+optional; when supplied it requires `--atomic-cost-workload`, a JSON file
+containing the exact `profiles[].workload` value from that benchmark artifact.
+The loader validates the document schema and rejects a selector that matches
+zero or multiple profiles; it never mixes entries across workloads. Omit both
+flags and ordinary unbenchmarked candidates use the flat stub, while
 CMS-with-heap candidates warn and are dropped until a matching reference row is available.
 
 ### Running with real sketch-bench costs
@@ -352,13 +357,15 @@ CMS-with-heap candidates warn and are dropped until a matching reference row is
 # See what each candidate would cost, before selection:
  cargo run -p asap_planner --bin candidate-gen-dump -- \
   --input_config workload.yaml --data-ingestion-interval-ms 60000 \
-  --atomic-costs path/to/atomic_costs.json
+  --atomic-costs path/to/atomic_costs.json \
+  --atomic-cost-workload path/to/workload-selector.json
 
 # Run the actual optimizer:
 cargo run -p asap_planner --bin asap-optimizer-cli -- \
   --input_config workload.yaml --data-ingestion-interval-ms 60000 \
   --dataset path/to/series-inventory.csv \
-  --atomic-costs path/to/atomic_costs.json
+  --atomic-costs path/to/atomic_costs.json \
+  --atomic-cost-workload path/to/workload-selector.json
 ```
 
 `candidate-gen-dump`'s output labels each resolved params row `[real]` or `[stub]`; candidates

CONTEXT.md

@@ -0,0 +1,24 @@
+# ASAPQuery
+
+Terms used by the cost-based sketch optimizer and its benchmarking pipeline.
+
+## Language
+
+**Grouping labels**:
+PromQL `GROUP BY` labels that partition a metric stream into label groups; a
+non-subpopulation sketch is instantiated once per distinct group.
+_Avoid_: keying labels
+
+**Sketch key**:
+The value or label tuple inserted into and queried from a keyed sketch such as
+HLL, CMS, or Hydra. Its distribution, cardinality, and encoded size may affect
+the measured atomic cost and accuracy.
+_Avoid_: grouping label, partition key
+
+**Benchmark profile**:
+A reproducible description of the input trace slice used to measure an
+atomic-cost table for a sketch family and configuration.
+
+**Planner context**:
+Query- and deployment-specific structural inputs, such as label-group count
+and retained-window count, that scale atomic costs into a plan cost.

asap-planner-rs/src/bin/candidate_gen_dump.rs

@@ -6,8 +6,8 @@ use std::path::PathBuf;
 
 use asap_planner::{
     optimizer::{
-        enumerate_candidates, extract_aqes, load_atomic_cost_table, resolve_atomic_costs,
-        AtomicCostTable, AtomicCosts, CandidateConfig, RQE,
+        enumerate_candidates, extract_aqes, load_optional_selected_atomic_cost_table,
+        resolve_atomic_costs, AtomicCostTable, AtomicCosts, CandidateConfig, RQE,
     },
     ControllerConfig,
 };
@@ -28,21 +28,26 @@ struct Args {
     #[arg(long = "data-ingestion-interval-ms")]
     scrape_interval_ms: u64,
 
-    /// Path to sketch-bench's exported atomic-cost table (see ASAPQuery#524).
+    /// Path to sketch-bench's versioned atomic-cost document.
     /// When given, each params row also prints its resolved AtomicCosts --
     /// real (from the table) or the flat stub (unbenchmarked family, or this
     /// exact param point missing from the table) -- labeled which.
     #[arg(long = "atomic-costs")]
     atomic_costs: Option<PathBuf>,
+
+    /// JSON `profiles[].workload` value selecting exactly one measured profile.
+    #[arg(long = "atomic-cost-workload", requires = "atomic_costs")]
+    atomic_cost_workload: Option<PathBuf>,
 }
 
 fn main() -> anyhow::Result<()> {
     let args = Args::parse();
 
-    let atomic_cost_table = match &args.atomic_costs {
-        Some(path) => load_atomic_cost_table(path)?,
-        None => AtomicCostTable::default(),
-    };
+    let atomic_cost_table = load_optional_selected_atomic_cost_table(
+        args.atomic_costs.as_deref(),
+        args.atomic_cost_workload.as_deref(),
+    )?
+    .unwrap_or_default();
 
     let yaml_str = std::fs::read_to_string(&args.input_config)?;
     let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;

asap-planner-rs/src/bin/optimizer_cli.rs

@@ -7,7 +7,7 @@
 use std::path::PathBuf;
 
 use asap_planner::optimizer::{
-    load_atomic_cost_table, run_greedy_pipeline, AtomicCostTable, SeriesDataset,
+    load_optional_selected_atomic_cost_table, run_greedy_pipeline, AtomicCostTable, SeriesDataset,
 };
 use asap_planner::ControllerConfig;
 use clap::Parser;
@@ -35,14 +35,21 @@ struct Args {
     #[arg(long = "rho", default_value = "1.0", value_parser = parse_positive_finite)]
     rho: f64,
 
-    /// Path to the atomic-cost table sketch-bench's `atomic-costs` subcommand
-    /// exports (see ASAPQuery#524, sketch-bench#30). Omitted: every
+    /// Path to the versioned atomic-cost document sketch-bench's `atomic-costs`
+    /// subcommand exports. Requires --atomic-cost-workload to select exactly
+    /// one measured workload profile. Omitted: every
     /// benchmarked-family candidate (CMS/HLL/KLL) is dropped, since there is
     /// no data to cost it at — only trivial accumulators and EXACT remain
     /// selectable.
     #[arg(long = "atomic-costs")]
     atomic_costs: Option<PathBuf>,
 
+    /// JSON `profiles[].workload` value copied from the sketch-bench atomic-cost
+    /// document. This makes the empirical workload profile explicit and avoids
+    /// mixing costs from different traces or time windows.
+    #[arg(long = "atomic-cost-workload", requires = "atomic_costs")]
+    atomic_cost_workload: Option<PathBuf>,
+
     #[arg(short, long, action = clap::ArgAction::Count)]
     verbose: u8,
 }
@@ -70,8 +77,11 @@ fn main() -> anyhow::Result<()> {
     let config: ControllerConfig = serde_yaml::from_str(&yaml_str)?;
     let dataset = SeriesDataset::from_path(&args.dataset)?;
 
-    let atomic_cost_table = match &args.atomic_costs {
-        Some(path) => load_atomic_cost_table(path)?,
+    let atomic_cost_table = match load_optional_selected_atomic_cost_table(
+        args.atomic_costs.as_deref(),
+        args.atomic_cost_workload.as_deref(),
+    )? {
+        Some(table) => table,
         None => {
             tracing::warn!(
                 "no --atomic-costs supplied; CMS/HLL/KLL candidates will never be selected"