From 3cf0740b5ebdf6bd16914b1d1f950e62e4603b84 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 18:50:53 -0600
Subject: [PATCH 1/7] feat: measure MetricsQL candidates through the shared
 compiler

---
 .../examples/calibration_candidates.rs        |  10 +-
 .../examples/compile_workload_artifact.rs     |   8 +-
 control_plane/src/physical/compiler.rs        |  43 +++++-
 data_plane/examples/univmon_erp_artifact.rs   | 140 ++++++++++++++++++
 tools/o11y-execution/CALIBRATION.md           |  35 +++++
 tools/o11y-execution/calibrate_runtime.py     |  85 ++++++++---
 .../o11y-execution/test_calibrate_runtime.py  |  26 ++++
 7 files changed, 319 insertions(+), 28 deletions(-)
 create mode 100644 data_plane/examples/univmon_erp_artifact.rs

diff --git a/control_plane/examples/calibration_candidates.rs b/control_plane/examples/calibration_candidates.rs
index 607bc1501..3b119bb70 100644
--- a/control_plane/examples/calibration_candidates.rs
+++ b/control_plane/examples/calibration_candidates.rs
@@ -120,7 +120,8 @@ fn planner_forest(queries: &[control_plane::physical::compiler::PlanningQuery])
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     let path = std::env::args()
         .nth(1)
-        .ok_or("usage: calibration_candidates SNAPSHOT.json")?;
+        .ok_or("usage: calibration_candidates SNAPSHOT.json [--metricsql]")?;
+    let metricsql = std::env::args().skip(2).any(|arg| arg == "--metricsql");
     let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?;
     let (request, environment) = snapshot.planning_request()?;
     let mut results = Vec::new();
@@ -131,7 +132,12 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         let queries = candidate.queries.clone();
         let materialization_policy = candidate.materialization_policy.clone();
         let planner_selected_queries = planner_forest(&queries);
-        let plan = match PhysicalCompiler.compile(candidate, environment.clone()) {
+        let compiled = if metricsql {
+            PhysicalCompiler.compile_metricsql(candidate, environment.clone())
+        } else {
+            PhysicalCompiler.compile(candidate, environment.clone())
+        };
+        let plan = match compiled {
             Ok(plan) => plan,
             Err(error) => {
                 results.push(
diff --git a/control_plane/examples/compile_workload_artifact.rs b/control_plane/examples/compile_workload_artifact.rs
index 54fd24fb7..b9c3c3480 100644
--- a/control_plane/examples/compile_workload_artifact.rs
+++ b/control_plane/examples/compile_workload_artifact.rs
@@ -5,7 +5,7 @@ use serde_json::json;
 fn main() -> Result<(), Box<dyn std::error::Error>> {
     let path = std::env::args()
         .nth(1)
-        .ok_or("usage: compile_workload_artifact SNAPSHOT.json")?;
+        .ok_or("usage: compile_workload_artifact SNAPSHOT.json [--metricsql]")?;
     let snapshot: BackendLocalPlanningSnapshot = serde_json::from_slice(&std::fs::read(path)?)?;
     if snapshot.snapshot_version != 2 {
         return Err(
@@ -13,7 +13,11 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
         );
     }
     let start = std::time::Instant::now();
-    let plan = snapshot.compile()?;
+    let plan = if std::env::args().skip(2).any(|arg| arg == "--metricsql") {
+        snapshot.compile_metricsql()?
+    } else {
+        snapshot.compile()?
+    };
     let elapsed = start.elapsed().as_nanos();
     let comparison = plan
         .cost_comparison
diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index f382c0c66..687c8cdf4 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -1276,6 +1276,15 @@ impl BackendLocalPlanningSnapshot {
     /// one backend-local PhysicalPlan. No CollectorPlan is produced and no
     /// precompiled serving artifact is accepted at this boundary.
     pub fn compile(self) -> Result<PhysicalPlan, CompileError> {
+        self.compile_frontend(false)
+    }
+
+    /// Use the shared parser subset with MetricsQL serving and exact routing.
+    pub fn compile_metricsql(self) -> Result<PhysicalPlan, CompileError> {
+        self.compile_frontend(true)
+    }
+
+    fn compile_frontend(self, metricsql: bool) -> Result<PhysicalPlan, CompileError> {
         let evidence = self.workload_cost_evidence.clone();
         if self.snapshot_version == 2 && evidence.is_none() {
             return Err(CompileError::Snapshot(
@@ -1284,18 +1293,25 @@ impl BackendLocalPlanningSnapshot {
         }
         let (request, environment) = self.planning_request()?;
         match evidence {
-            Some(evidence) => super::workload_cost::select(
-                super::workload_cost::with_exact_alternative(request)?,
-                environment,
-                &evidence,
-            ),
+            Some(evidence) => {
+                let candidates = super::workload_cost::with_exact_alternative(request)?;
+                if metricsql {
+                    super::workload_cost::select_metricsql(candidates, environment, &evidence)
+                } else {
+                    super::workload_cost::select(candidates, environment, &evidence)
+                }
+            }
             None => {
                 // Unquoted v1 startup snapshots keep the established summary/native
                 // compatibility policy. Local residual candidates are enumerated by
                 // planning_request and admitted through measured workload selection.
                 let mut request = request;
                 request.hybrid_execution = false;
-                PhysicalCompiler.compile(request, environment)
+                if metricsql {
+                    PhysicalCompiler.compile_metricsql(request, environment)
+                } else {
+                    PhysicalCompiler.compile(request, environment)
+                }
             }
         }
     }
@@ -4267,6 +4283,21 @@ mod tests {
         assert!(matches!(result, Err(CompileError::QueryPlan(_))));
     }
 
+    #[test]
+    fn snapshot_metricsql_entry_uses_the_shared_serving_language_contract() {
+        let snapshot: BackendLocalPlanningSnapshot = serde_json::from_str(include_str!(
+            "../../../docs/examples/asapquery-compatibility-demo-snapshot.json"
+        ))
+        .unwrap();
+        let plan = snapshot.compile_metricsql().unwrap();
+        assert!(!plan.query_plan.entries.is_empty());
+        assert!(plan
+            .query_plan
+            .entries
+            .values()
+            .all(|entry| entry.language == crate::query_plan::QueryLanguage::MetricsQl));
+    }
+
     #[test]
     fn duplicate_query_ids_cannot_overwrite_installed_dag_documents() {
         let mut workload = request("shared-id", "max_over_time(a[1m])");
diff --git a/data_plane/examples/univmon_erp_artifact.rs b/data_plane/examples/univmon_erp_artifact.rs
new file mode 100644
index 000000000..c48c95a92
--- /dev/null
+++ b/data_plane/examples/univmon_erp_artifact.rs
@@ -0,0 +1,140 @@
+//! Measure readout-specific ERP evidence from finite JSONL evaluation data.
+//! This offline tool retains samples; the production backend does not.
+use data_plane::precompute_engine::operators::univmon_accumulator::UnivMonAccumulator;
+use data_plane::storage_engines::types::{AggregateCore, SerializableToSink};
+use serde_json::{json, Value};
+use std::collections::{BTreeMap, HashMap};
+use std::io::{BufRead, BufReader};
+
+fn main() -> Result<(), Box<dyn std::error::Error>> {
+    let path = std::env::args()
+        .nth(1)
+        .ok_or("usage: univmon_erp_artifact samples.jsonl")?;
+    let mut series = BTreeMap::<String, Vec<(u64, f64)>>::new();
+    for line in BufReader::new(std::fs::File::open(&path)?).lines() {
+        let row: Value = serde_json::from_str(&line?)?;
+        let key = serde_json::to_string(&row["labels"])?;
+        series.entry(key).or_default().push((
+            row["ts_ms"].as_u64().ok_or("timestamp")?,
+            row["value"].as_f64().ok_or("value")?,
+        ));
+    }
+    if series
+        .values()
+        .any(|samples| samples.windows(2).any(|pair| pair[0].0 >= pair[1].0))
+    {
+        return Err("each series must have strictly increasing timestamps".into());
+    }
+    let first = series.values().next().ok_or("empty dataset")?;
+    let start = first.first().ok_or("empty series")?.0;
+    let end = first.last().ok_or("empty series")?.0;
+    let populations: Vec<_> = series
+        .iter()
+        .take(10)
+        .map(|(key, samples)| {
+            (
+                key,
+                samples
+                    .iter()
+                    .filter(|(ts, _)| *ts > start && *ts <= end)
+                    .map(|(_, v)| *v)
+                    .collect::<Vec<_>>(),
+            )
+        })
+        .collect();
+    if populations.len() < 2 {
+        return Err("need at least two independent populations".into());
+    }
+    let mut records = Vec::new();
+    for (heap, cols, layers) in [(64, 512, 8), (512, 2048, 8), (4096, 8192, 16)] {
+        let mut max_errors = [0.0f64; 3];
+        let mut bytes = 0;
+        let mut observed = None;
+        for (_, raw) in &populations {
+            let mut counts = HashMap::<u64, usize>::new();
+            for v in raw {
+                *counts
+                    .entry(if *v == 0.0 { 0 } else { v.to_bits() })
+                    .or_default() += 1;
+            }
+            let truths = [
+                counts.len() as f64,
+                counts
+                    .values()
+                    .map(|c| (*c as f64).powi(2))
+                    .sum::<f64>()
+                    .sqrt(),
+                counts
+                    .values()
+                    .map(|c| {
+                        let p = *c as f64 / raw.len() as f64;
+                        -p * p.log2()
+                    })
+                    .sum(),
+            ];
+            let mut observer = control_plane::physical::erp::ErpShapeObserver::new(raw.len())?;
+            for (i, v) in raw.iter().enumerate() {
+                observer.observe(&v.to_string(), i / 1000)?;
+            }
+            let next = observer.snapshot().ok_or("invalid observation")?;
+            if observed.as_ref().is_some_and(
+                |prior: &control_plane::physical::erp::ErpObservedShape| {
+                    prior.observation != next.observation
+                },
+            ) {
+                return Err(
+                    "training populations have different shapes; calibrate separate shape strata"
+                        .into(),
+                );
+            }
+            observed = Some(next);
+            let mut left =
+                UnivMonAccumulator::new(heap, 5, cols, layers).map_err(|e| e.to_string())?;
+            let mut right = left.clone();
+            for (i, v) in raw.iter().enumerate() {
+                if i % 2 == 0 { &mut left } else { &mut right }
+                    .insert_sample(*v)
+                    .map_err(|e| e.to_string())?;
+            }
+            left.merge_in_place(&right).map_err(|e| e.to_string())?;
+            bytes = bytes.max(left.serialize_to_bytes().len());
+            for (i, stat) in [
+                asap_types::Statistic::Cardinality,
+                asap_types::Statistic::FrequencyL2,
+                asap_types::Statistic::FrequencyEntropy,
+            ]
+            .into_iter()
+            .enumerate()
+            {
+                let answer = left
+                    .query_statistic(stat, &None, &Default::default())
+                    .map_err(|e| e.to_string())?;
+                if !answer.is_finite() {
+                    return Err("nonfinite estimate".into());
+                }
+                max_errors[i] = max_errors[i]
+                    .max((answer - truths[i]).abs() / if i == 2 { 1.0 } else { truths[i] });
+            }
+        }
+        let observed = observed.unwrap();
+        let fit = observed
+            .observation
+            .fits
+            .iter()
+            .min_by(|a, b| a.goodness_of_fit.total_cmp(&b.goodness_of_fit))
+            .ok_or("no shape")?;
+        records.push(json!({"id":format!("univmon-h{heap}-c{cols}-l{layers}"),"sketch":"univmon","implementation":"asap-sketchlib-univmon-standard-v1",
+            "parameters":{"heap_size":heap,"sketch_rows":5,"sketch_cols":cols,"layers":layers},"trials":populations.len(),
+            "distribution":{"erp_shape":{"family":fit.family,"parameters":fit.parameters,"cardinality":observed.observation.cardinality,"benchmark_events":observed.observation.observed_events}},
+            "error_metrics":{"max_cardinality_relative_error":max_errors[0],"max_frequency_l2_relative_error":max_errors[1],"max_frequency_entropy_absolute_bits_error":max_errors[2]},
+            "resources":{"memory_bytes":bytes,"update_cpu_seconds":0.0,"merge_cpu_seconds":0.0,"query_cpu_seconds":0.0}}));
+        eprintln!("measured heap={heap} cols={cols} layers={layers}: {max_errors:?}");
+    }
+    println!(
+        "{}",
+        serde_json::to_string_pretty(
+            &json!({"schema_version":1,"producer_version":"backend-standard-unit-frequency-merged-two-pane-data-calibration-cpu-excluded","records":records})
+        )?
+    );
+    Ok(())
+}
diff --git a/tools/o11y-execution/CALIBRATION.md b/tools/o11y-execution/CALIBRATION.md
index 5217d19dc..f986e5138 100644
--- a/tools/o11y-execution/CALIBRATION.md
+++ b/tools/o11y-execution/CALIBRATION.md
@@ -63,3 +63,38 @@ The backend currently exposes its Planner-selected forest and a whole-workload
 exact alternative. This workflow does not claim exhaustive algorithm or lifecycle
 search, and its unit discovery seed can influence which forest becomes available.
 A calibrated comparison is only between the candidates actually exported.
+
+## VictoriaMetrics and readout-specific checks
+
+Use `calibration_candidates SNAPSHOT --metricsql` and
+`compile_workload_artifact SNAPSHOT --metricsql` to compile the shared parser
+subset through the existing MetricsQL serving path. This preserves language-tagged
+query entries and exact edges; do not change an emitted install artifact by hand.
+`BackendLocalPlanningSnapshot::compile_metricsql()` exposes the same operation.
+
+`calibrate_runtime.py --victoriametrics BINARY` starts a fresh VictoriaMetrics
+fallback instead of Prometheus. Ingestion and drain use the normal backend port;
+queries use `--metricsql-port`. `--exact-cache-bytes` controls VictoriaMetrics cache
+allocation, not process RSS. Record the binary version, cache budget and CPU set.
+`--disable-result-cache` sets VictoriaMetrics `-search.disableCache` (including
+its use through backend exact edges) and sends `nocache=1` to both query endpoints, so repeated
+identical timestamps measure query execution instead of the exact server's result
+cache. Declare that policy in the comparison report.
+
+A corpus occurrence may contain `accuracy_validation` with a `metric` and
+`bound`. Supported validation units are `relative`, `absolute_bits`, and `exact`
+(the last requires zero bound). In particular, entropy absolute bits cannot
+borrow a relative tolerance. A rank-error contract needs a rank oracle and is
+intentionally rejected by this scalar comparison helper; do not use value error
+to certify KLL rank accuracy. Native unsupported exact functions remain failed
+comparisons and cannot obtain an executable cost quote.
+
+For offline UnivMon error evidence, `cargo run --release -p data_plane --example
+univmon_erp_artifact -- samples.jsonl` measures two-pane merged readouts on the
+first ten source populations. It requires equal observed shapes; differing
+populations must be calibrated separately. The tool retains samples offline,
+records observed maxima and serialized state bytes, and does not measure CPU or
+calibrate a failure probability. Use a zero CPU objective weight for that artifact;
+whole-candidate CPU comes from the independent runtime calibration above. Keep
+held-out source populations and performance runs separate from these training
+populations. This tool does not select a plan.
diff --git a/tools/o11y-execution/calibrate_runtime.py b/tools/o11y-execution/calibrate_runtime.py
index 905053fb6..9e8923a8a 100644
--- a/tools/o11y-execution/calibrate_runtime.py
+++ b/tools/o11y-execution/calibrate_runtime.py
@@ -7,6 +7,7 @@
 import argparse
 import hashlib
 import json
+import math
 import os
 from pathlib import Path
 import resource
@@ -46,6 +47,47 @@ def file_bytes(root):
     return sum(path.stat().st_size for path in root.rglob("*") if path.is_file())
 
 
+def exact_service_command(args, folder, port):
+    if getattr(args, "victoriametrics", None):
+        command = [str(args.victoriametrics.resolve()),
+                f"-storageDataPath={(folder / 'exact-data').resolve()}",
+                f"-httpListenAddr=127.0.0.1:{port}", "-retentionPeriod=1y",
+                f"-memory.allowedBytes={args.exact_cache_bytes}"]
+        if args.disable_result_cache:
+            command.append("-search.disableCache")
+        return command
+    config = folder / "prometheus.yml"
+    config.write_text('global:\n  scrape_interval: 1h\nscrape_configs: []\n')
+    return [str(args.prometheus.resolve()), f"--config.file={config.resolve()}",
+            f"--storage.tsdb.path={(folder / 'exact-data').resolve()}",
+            f"--web.listen-address=127.0.0.1:{port}", "--web.enable-remote-write-receiver",
+            "--storage.tsdb.retention.time=1000000h"]
+
+
+def query_parameters(occurrence, disable_cache):
+    params = {"query": occurrence["query"], "time": f'{occurrence["eval_timestamp_ms"] / 1000:.3f}'}
+    if disable_cache:
+        params["nocache"] = "1"
+    return urllib.parse.urlencode(params)
+
+
+def comparison_tolerances(occurrence, args):
+    contract = occurrence.get("accuracy_validation")
+    if contract is None:
+        return args.relative_tolerance, args.absolute_tolerance
+    bound = contract.get("bound")
+    if not isinstance(bound, (float, int)) or not math.isfinite(bound) or bound < 0:
+        raise ValueError("invalid per-query accuracy bound")
+    metric = contract.get("metric")
+    if metric == "relative":
+        return bound, 0.0
+    if metric == "absolute_bits":
+        return 0.0, bound
+    if metric == "exact" and bound == 0:
+        return 0.0, 0.0
+    raise ValueError("unsupported per-query accuracy metric")
+
+
 def measure(args, artifact, corpus, snapshot, folder):
     folder.mkdir()
     manifest = artifact["manifest"]
@@ -65,21 +107,19 @@ def launch(name, command):
     backend = f"http://127.0.0.1:{args.backend_port}"
     install = folder / "install.json"
     runner.write_json(install, artifact["install_request"])
-    config = folder / "prometheus.yml"
-    config.write_text('global:\n  scrape_interval: 1h\nscrape_configs: []\n')
+    query_backend = f"http://127.0.0.1:{args.metricsql_port}" if args.victoriametrics else backend
     children_cpu_before = resource.getrusage(resource.RUSAGE_CHILDREN)
     try:
         start = time.perf_counter_ns()
-        prom = launch("fallback", [str(args.prometheus.resolve()), f"--config.file={config.resolve()}",
-                     f"--storage.tsdb.path={(folder / 'prometheus-data').resolve()}",
-                     f"--web.listen-address=127.0.0.1:{args.fallback_port}", "--web.enable-remote-write-receiver",
-                     "--storage.tsdb.retention.time=1000000h"])
-        # Backend startup validates its exact service immediately.
-        wait_ready(fallback + "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/-/ready", prom)
-        dp = launch("backend", [str(args.data_plane.resolve()), "--profile", "asapquery", "--physical-plan", str(install.resolve()),
-                    "--prometheus-server", fallback, "--forward-unsupported-queries", "--http-port", str(args.backend_port),
-                    "--output-dir", str((folder / "backend-data").resolve()), "--precompute-allowed-lateness-ms", "0",
-                    "--precompute-flush-interval-ms", "25"])
+        prom = launch("fallback", exact_service_command(args, folder, args.fallback_port))
+        wait_ready(fallback + ("/health" if args.victoriametrics else "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/-/ready"), prom)
+        command = [str(args.data_plane.resolve()), "--profile", "asapquery", "--physical-plan", str(install.resolve()),
+                   "--prometheus-server", fallback, "--forward-unsupported-queries", "--http-port", str(args.backend_port),
+                   "--output-dir", str((folder / "backend-data").resolve()), "--precompute-allowed-lateness-ms", "0",
+                   "--precompute-flush-interval-ms", "25"]
+        if args.victoriametrics:
+            command += ["--victoriametrics-url", fallback, "--victoriametrics-http-port", str(args.metricsql_port)]
+        dp = launch("backend", command)
         wait_ready(backend + "/api/v1/health", dp)
         after = snapshots(children)
         # Fresh processes: cumulative CPU from exec includes all install/startup work.
@@ -107,18 +147,18 @@ def launch(name, command):
                 raise RuntimeError(f"no original corpus occurrences for {qid}")
             exact = {}
             for occurrence in occurrences:
-                params = urllib.parse.urlencode({"query": occurrence["query"], "time": f'{occurrence["eval_timestamp_ms"] / 1000:.3f}'})
+                params = query_parameters(occurrence, args.disable_result_cache)
                 exact[occurrence["id"]] = runner._http_request(args.reference_url.rstrip("/") + "/api/v1/query?" + params)
             records, before, start = [], snapshots(children), time.perf_counter_ns()
             repeat = 0
             measured_cpu = 0
             while repeat < args.repetitions or (measured_cpu < args.minimum_query_cpu_ns and repeat < args.max_repetitions):
                 for occurrence in occurrences:
-                    params = urllib.parse.urlencode({"query": occurrence["query"], "time": f'{occurrence["eval_timestamp_ms"] / 1000:.3f}'})
-                    answer = runner._http_request(backend + "/api/v1/query?" + params)
+                    params = query_parameters(occurrence, args.disable_result_cache)
+                    answer = runner._http_request(query_backend + "/api/v1/query?" + params)
                     reference = exact[occurrence["id"]]
                     route = runner.classify(answer["response"], answer["headers"]) if answer["http_status"] == 200 else "failed"
-                    comparison = compare_results(answer["response"], reference["response"], args.relative_tolerance, args.absolute_tolerance)
+                    comparison = compare_results(answer["response"], reference["response"], *comparison_tolerances(occurrence, args))
                     records.append({**occurrence, "repetition": repeat, "execution": route,
                                     "execution_provenance": runner.execution_provenance(answer["response"], answer["headers"]), **answer,
                                     "exact": reference, "comparison": comparison})
@@ -141,7 +181,7 @@ def launch(name, command):
         runner.write_json(folder / "store.json", state)
         final = snapshots(children)
         row["resources"] = {"peak_memory_bytes": sum(v["process_lifetime_peak_rss_bytes"] for v in final.values()),
-                            "storage_bytes": file_bytes(folder / "prometheus-data") + file_bytes(folder / "backend-data"),
+                            "storage_bytes": file_bytes(folder / "exact-data") + file_bytes(folder / "backend-data"),
                             "backend_state": state, "processes": final, "source_scan_bytes": None, "network_bytes": None,
                             "residency_wall_seconds": args.residency_seconds,
                             "scope": "accelerated finite-input replay; no extrapolation of short idle residency to logical data horizon; process HWM sum is conservative"}
@@ -283,8 +323,14 @@ def validate_candidate_topk_execution(artifact, records):
 
 def main():
     parser = argparse.ArgumentParser(description=__doc__)
-    for name in ["candidates", "metrics", "queries", "snapshot", "data-plane", "prometheus", "output"]:
+    for name in ["candidates", "metrics", "queries", "snapshot", "data-plane", "output"]:
         parser.add_argument("--" + name, type=Path, required=True)
+    engines = parser.add_mutually_exclusive_group(required=True)
+    engines.add_argument("--prometheus", type=Path)
+    engines.add_argument("--victoriametrics", type=Path)
+    parser.add_argument("--metricsql-port", type=int, default=19212)
+    parser.add_argument("--exact-cache-bytes", type=int, default=268435456, help="VictoriaMetrics cache budget, not an RSS limit")
+    parser.add_argument("--disable-result-cache", action="store_true", help="send nocache=1 to both query endpoints")
     parser.add_argument("--reference-url", required=True, help="separate already-loaded exact Prometheus using identical input")
     parser.add_argument("--cpu-affinity", required=True)
     parser.add_argument("--backend-port", type=int, default=19210)
@@ -298,12 +344,15 @@ def main():
     args = parser.parse_args()
     if args.repetitions < 1 or args.max_repetitions < args.repetitions or args.minimum_query_cpu_ns <= 0 or args.residency_seconds < 0:
         parser.error("positive repetitions and nonnegative residency required")
+    if len({args.backend_port, args.fallback_port, args.metricsql_port}) != 3 or args.exact_cache_bytes <= 0:
+        parser.error("distinct listener ports and positive exact cache budget required")
     args.output.mkdir(parents=True, exist_ok=False)
     sample_count = runner.validate_sample_file(args.metrics)
     corpus, snapshot = json.loads(args.queries.read_text()), json.loads(args.snapshot.read_text())
     runner.validate_workload(snapshot, corpus)
     candidate_document = json.loads(args.candidates.read_text())
     result = {"units": "cpu_ns", "compiler_identity": candidate_document.get("compiler_identity"), "data_snapshot_id": "sha256:" + hashlib.sha256(args.metrics.read_bytes()).hexdigest(),
+              "exact_engine": "victoriametrics" if args.victoriametrics else "prometheus", "result_cache_disabled": args.disable_result_cache,
               "scope": "accelerated finite-input calibration; measured wall residency is not full logical-horizon residency", "validated_sample_count": sample_count, "candidates": []}
     candidates = candidate_document["candidates"]
     for candidate in candidates:
diff --git a/tools/o11y-execution/test_calibrate_runtime.py b/tools/o11y-execution/test_calibrate_runtime.py
index f8d3533d8..e638a8c38 100644
--- a/tools/o11y-execution/test_calibrate_runtime.py
+++ b/tools/o11y-execution/test_calibrate_runtime.py
@@ -113,3 +113,29 @@ def test_candidate_filtered_execution_requires_hybrid_one_rpc_and_one_summary_re
 
 if __name__ == "__main__":
     unittest.main()
+
+class PerQueryAccuracyTests(unittest.TestCase):
+    def test_entropy_absolute_error_does_not_accept_relative_tolerance(self):
+        from types import SimpleNamespace
+        from calibrate_runtime import comparison_tolerances
+        from compare import compare_results
+        args = SimpleNamespace(relative_tolerance=.9, absolute_tolerance=.9)
+        relative, absolute = comparison_tolerances({"accuracy_validation": {"metric": "absolute_bits", "bound": .1}}, args)
+        actual = {"status": "success", "data": {"resultType": "vector", "result": [{"metric": {}, "value": [0, "10.2"]}]}}
+        exact = {"status": "success", "data": {"resultType": "vector", "result": [{"metric": {}, "value": [0, "10"]}]}}
+        self.assertEqual((relative, absolute), (0, .1))
+        self.assertFalse(compare_results(actual, exact, relative, absolute)["equal"])
+
+    def test_unknown_metric_and_invalid_bound_are_rejected(self):
+        from types import SimpleNamespace
+        from calibrate_runtime import comparison_tolerances
+        for metric, bound in [("rank", .1), ("relative", float("nan")), ("absolute_bits", -1), ("exact", .1)]:
+            with self.assertRaises(ValueError):
+                comparison_tolerances({"accuracy_validation": {"metric": metric, "bound": bound}}, SimpleNamespace())
+
+    def test_result_cache_policy_is_explicit_on_both_query_endpoints(self):
+        from urllib.parse import parse_qs
+        from calibrate_runtime import query_parameters
+        row = {"query": "count_over_time(m[1h])", "eval_timestamp_ms": 1234}
+        self.assertNotIn("nocache", parse_qs(query_parameters(row, False)))
+        self.assertEqual(parse_qs(query_parameters(row, True))["nocache"], ["1"])

From ef757f1d85a5a10fa07d757d89bf6019ffe8041b Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 19:08:40 -0600
Subject: [PATCH 2/7] Calibrate HLL cardinality with readout-specific evidence

---
 control_plane/src/physical/compiler.rs |  2 ++
 control_plane/src/physical/erp.rs      | 38 ++++++++++++++++++++++++--
 2 files changed, 38 insertions(+), 2 deletions(-)

diff --git a/control_plane/src/physical/compiler.rs b/control_plane/src/physical/compiler.rs
index 687c8cdf4..01ceb30c8 100644
--- a/control_plane/src/physical/compiler.rs
+++ b/control_plane/src/physical/compiler.rs
@@ -2532,6 +2532,8 @@ pub fn select_workload_roots_with_erp(
             // for the collector's sketchlib KLL, even with the same k.
             policy.artifact.records.retain(|row| {
                 (row.sketch == "kll-percall" && row.implementation == "lib")
+                    || (row.sketch == "hll"
+                        && row.implementation == "asap-sketchlib-hll-regular-v1")
                     || (row.sketch == "univmon"
                         && row.implementation == "asap-sketchlib-univmon-standard-v1")
             });
diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs
index 8a1bc7ac2..aed6c9806 100644
--- a/control_plane/src/physical/erp.rs
+++ b/control_plane/src/physical/erp.rs
@@ -183,7 +183,7 @@ impl ReadoutEvidence {
         use planner_types::pre_asap::AggIntent;
         match (algorithm, intent) {
             (SketchAlgorithm::Kll, AggIntent::Quantile { .. }) => Some(Self::QuantileRank),
-            (SketchAlgorithm::UnivMon, AggIntent::Cardinality { .. }) => {
+            (SketchAlgorithm::Hll | SketchAlgorithm::UnivMon, AggIntent::Cardinality { .. }) => {
                 Some(Self::DistinctRelative)
             }
             (SketchAlgorithm::UnivMon, AggIntent::FrequencyL2 { .. }) => {
@@ -203,7 +203,9 @@ impl ReadoutEvidence {
         use planner_types::post_asap::SketchQuery;
         match (algorithm, query) {
             (SketchAlgorithm::Kll, SketchQuery::Quantile { .. }) => Some(Self::QuantileRank),
-            (SketchAlgorithm::UnivMon, SketchQuery::Cardinality) => Some(Self::DistinctRelative),
+            (SketchAlgorithm::Hll | SketchAlgorithm::UnivMon, SketchQuery::Cardinality) => {
+                Some(Self::DistinctRelative)
+            }
             (SketchAlgorithm::UnivMon, SketchQuery::FrequencyL2) => Some(Self::FrequencyL2Relative),
             (SketchAlgorithm::UnivMon, SketchQuery::FrequencyEntropy) => {
                 Some(Self::EntropyAbsoluteBits)
@@ -809,6 +811,38 @@ mod tests {
         }
     }
 
+    #[test]
+    fn hll_cardinality_uses_measured_relative_error_not_rse() {
+        use asap_aware_mapping::AccuracyModel;
+        use planner_types::post_asap::*;
+        let mut policy = input(ErpAccuracyMode::Hybrid);
+        let row = &mut policy.artifact.records[0];
+        row.sketch = "hll".into();
+        row.parameters = serde_json::json!({"precision":12});
+        row.error_metrics = BTreeMap::from([("max_cardinality_relative_error".into(), 0.04)]);
+        let family = SummaryFamilyType::Sketch(
+            SketchKind::new(SketchAlgorithm::Hll, SketchParams::Hll { precision: 12 }),
+            GroupingStrategy::PerSubpopulationInstance,
+        );
+        let model = ErpAccuracyModel {
+            policy: Some(&policy),
+            max_error: 0.05,
+        };
+        let guarantee = model
+            .local_guarantee(&family, &SketchQuery::Cardinality)
+            .unwrap();
+        assert_eq!(guarantee.metric, ErrorMetric::Cardinality);
+        assert_eq!(guarantee.bound, BoundExpr::Constant { value: 0.04 });
+        assert!(matches!(
+            guarantee.failure_probability,
+            ProbabilityExpr::Unknown { .. }
+        ));
+        assert_eq!(
+            ReadoutEvidence::for_query(SketchAlgorithm::Hll, &SketchQuery::FrequencyEntropy),
+            None
+        );
+    }
+
     /// Contract fixture only; the process test measures real sketch errors.
     #[test]
     fn readout_evidence_keeps_units_and_missing_metrics_fail_closed() {

From 0e6c663a2919e068514b76a189733beb75263871 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 19:10:05 -0600
Subject: [PATCH 3/7] Measure disjoint HLL calibration populations and preserve
 replay phase

---
 data_plane/examples/univmon_erp_artifact.rs   | 66 ++++++++++++++++++-
 tools/o11y-execution/CALIBRATION.md           | 11 ++++
 tools/o11y-execution/calibrate_runtime.py     |  6 ++
 tools/o11y-execution/discover_snapshot.py     |  8 ++-
 .../o11y-execution/test_discover_snapshot.py  |  1 +
 5 files changed, 89 insertions(+), 3 deletions(-)

diff --git a/data_plane/examples/univmon_erp_artifact.rs b/data_plane/examples/univmon_erp_artifact.rs
index c48c95a92..aa37ec137 100644
--- a/data_plane/examples/univmon_erp_artifact.rs
+++ b/data_plane/examples/univmon_erp_artifact.rs
@@ -1,5 +1,6 @@
 //! Measure readout-specific ERP evidence from finite JSONL evaluation data.
 //! This offline tool retains samples; the production backend does not.
+use data_plane::precompute_engine::operators::hll_sketch_accumulator::HllSketchAccumulator;
 use data_plane::precompute_engine::operators::univmon_accumulator::UnivMonAccumulator;
 use data_plane::storage_engines::types::{AggregateCore, SerializableToSink};
 use serde_json::{json, Value};
@@ -37,7 +38,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
                 samples
                     .iter()
                     .filter(|(ts, _)| *ts > start && *ts <= end)
-                    .map(|(_, v)| *v)
+                    .map(|(_, v)| *v + 1_000_000_000_000.0)
                     .collect::<Vec<_>>(),
             )
         })
@@ -45,7 +46,18 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     if populations.len() < 2 {
         return Err("need at least two independent populations".into());
     }
+    // Calibration hashes a disjoint numeric key namespace. The shift preserves
+    // frequencies/cardinality for this integer-valued finite fixture; reject
+    // inputs for which floating-point rounding could silently merge values.
+    if series.values().flatten().any(|(_, value)| {
+        !value.is_finite() || value.fract() != 0.0 || !(0.0..1_000_000_000_000.0).contains(value)
+    }) {
+        return Err(
+            "calibration namespace requires finite nonnegative integer values below 1e12".into(),
+        );
+    }
     let mut records = Vec::new();
+    let mut observed_shape = None;
     for (heap, cols, layers) in [(64, 512, 8), (512, 2048, 8), (4096, 8192, 16)] {
         let mut max_errors = [0.0f64; 3];
         let mut bytes = 0;
@@ -117,6 +129,7 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
             }
         }
         let observed = observed.unwrap();
+        observed_shape = Some(observed.observation.clone());
         let fit = observed
             .observation
             .fits
@@ -130,6 +143,57 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
             "resources":{"memory_bytes":bytes,"update_cpu_seconds":0.0,"merge_cpu_seconds":0.0,"query_cpu_seconds":0.0}}));
         eprintln!("measured heap={heap} cols={cols} layers={layers}: {max_errors:?}");
     }
+    // Predeclared precision grid; validation uses disjoint source populations.
+    let observation = observed_shape.as_ref().ok_or("no observation")?;
+    let fit = observation
+        .fits
+        .iter()
+        .min_by(|a, b| a.goodness_of_fit.total_cmp(&b.goodness_of_fit))
+        .ok_or("no fit")?;
+    for precision in [10, 12, 14] {
+        let mut max_error = 0.0f64;
+        let mut bytes = 0;
+        for (_, raw) in &populations {
+            let mut left =
+                HllSketchAccumulator::new(asap_sketchlib::HllVariant::Regular, precision);
+            let mut right = left.clone();
+            let mut distinct = std::collections::HashSet::new();
+            for (i, value) in raw.iter().enumerate() {
+                let bits = if *value == 0.0 { 0 } else { value.to_bits() };
+                distinct.insert(bits);
+                if i % 2 == 0 { &mut left } else { &mut right }
+                    .inner
+                    .update(&bits.to_le_bytes());
+            }
+            let merged = left.merge_with(&right).map_err(|e| e.to_string())?;
+            let merged = merged
+                .as_any()
+                .downcast_ref::<HllSketchAccumulator>()
+                .ok_or("HLL merge type")?;
+            let estimate = merged
+                .query_statistic(
+                    asap_types::Statistic::Cardinality,
+                    &None,
+                    &Default::default(),
+                )
+                .map_err(|e| e.to_string())?;
+            max_error =
+                max_error.max((estimate - distinct.len() as f64).abs() / distinct.len() as f64);
+            bytes = bytes.max(merged.serialize_to_bytes().len());
+        }
+        records.push(json!({"id":format!("hll-p{precision}"),"sketch":"hll","implementation":"asap-sketchlib-hll-regular-v1",
+            "parameters":{"precision":precision},"trials":populations.len(),
+            "distribution":{"erp_shape":{"family":fit.family,"parameters":fit.parameters,"cardinality":observation.cardinality,"benchmark_events":observation.observed_events}},
+            "error_metrics":{"max_cardinality_relative_error":max_error},
+            "resources":{"memory_bytes":bytes,"update_cpu_seconds":0.0,"merge_cpu_seconds":0.0,"query_cpu_seconds":0.0}}));
+        eprintln!("measured HLL precision={precision}: max relative cardinality error={max_error}");
+    }
+    if let Some(path) = std::env::args().nth(2) {
+        std::fs::write(
+            path,
+            serde_json::to_vec_pretty(&observed_shape.ok_or("no observation")?)?,
+        )?;
+    }
     println!(
         "{}",
         serde_json::to_string_pretty(
diff --git a/tools/o11y-execution/CALIBRATION.md b/tools/o11y-execution/CALIBRATION.md
index f986e5138..10d652e5f 100644
--- a/tools/o11y-execution/CALIBRATION.md
+++ b/tools/o11y-execution/CALIBRATION.md
@@ -98,3 +98,14 @@ calibrate a failure probability. Use a zero CPU objective weight for that artifa
 whole-candidate CPU comes from the independent runtime calibration above. Keep
 held-out source populations and performance runs separate from these training
 populations. This tool does not select a plan.
+
+For the finite integer-valued distinct study, the offline artifact tool measures
+HLL precisions 10, 12, and 14 as a predeclared grid, alongside the UnivMon grid.
+It uses the first ten source series and shifts their values by 1e12 into a disjoint
+hash namespace; inputs outside its documented integer domain are rejected. The
+held-out query uses groups 3–9 with original values. Neither a training maximum
+nor HLL's theoretical relative standard error is a probabilistic per-query error
+bound. The held-out 5% target remains fixed, and failed configurations stay in the
+measurement history. A second optional argument writes the actual observed shape
+for bounded ERP matching. CPU fields in this offline artifact remain excluded;
+production process calibration measures CPU separately.
diff --git a/tools/o11y-execution/calibrate_runtime.py b/tools/o11y-execution/calibrate_runtime.py
index 9e8923a8a..5b4760a5e 100644
--- a/tools/o11y-execution/calibrate_runtime.py
+++ b/tools/o11y-execution/calibrate_runtime.py
@@ -190,6 +190,12 @@ def launch(name, command):
             child.terminate()
         for child in children.values():
             child.wait(timeout=30)
+        row["resources"]["storage_before_shutdown_bytes"] = row["resources"]["storage_bytes"]
+        row["resources"]["storage_after_shutdown"] = {
+            "exact_bytes": file_bytes(folder / "exact-data"),
+            "backend_bytes": file_bytes(folder / "backend-data"),
+        }
+        row["resources"]["storage_bytes"] = sum(row["resources"]["storage_after_shutdown"].values())
         usage = resource.getrusage(resource.RUSAGE_CHILDREN)
         total_cpu = int((usage.ru_utime + usage.ru_stime - children_cpu_before.ru_utime - children_cpu_before.ru_stime) * 1e9)
         retirement = folder / "phase-retirement.json"
diff --git a/tools/o11y-execution/discover_snapshot.py b/tools/o11y-execution/discover_snapshot.py
index 7a1d60d33..7c640065e 100644
--- a/tools/o11y-execution/discover_snapshot.py
+++ b/tools/o11y-execution/discover_snapshot.py
@@ -72,13 +72,17 @@ def evidence(value):
         if args.interval_ms % frequencies[query]:
             raise ValueError("base interval must divide exactly by query occurrence frequency")
         interval = args.interval_ms // frequencies[query]
-        registrations.append({"query": query, "demand": {"fixed_interval": interval},
+        phases = {row["eval_timestamp_ms"] % interval for row in corpus["queries"] if row["query"] == query}
+        if len(phases) != 1:
+            raise ValueError("query occurrences require one explicit evaluation phase")
+        phase = phases.pop()
+        registrations.append({"query": query, "demand": {"fixed_interval_at": {"interval": interval, "evaluation_phase": phase}},
                               "requirements": {"accuracy": {"explicit": "Exact"}, "response_latency": "unspecified"},
                               "predictability": {"predictable": {"known_at": None}},
                               "time_selection": {"scope": "real_time", "lookback": lookback, "as_of": None}})
         query_audit.append({"query": query, "window_lookback_ms": lookback,
                             "occurrence_count": frequencies[query], "expected_evaluations": frequencies[query] * args.repetitions,
-                            "declared_interval_ms": interval,
+                            "declared_interval_ms": interval, "evaluation_phase_ms": phase,
                             "lookback_method": "largest explicit range; instant selector defaults to Prometheus 5m; original offsets/subqueries preserved in query"})
     snapshot["query_workload"].update(repeating_queries=registrations, data_workload=data, query_batch=None)
     snapshot["snapshot_version"] = 2
diff --git a/tools/o11y-execution/test_discover_snapshot.py b/tools/o11y-execution/test_discover_snapshot.py
index 9150d81ac..94497ec8b 100644
--- a/tools/o11y-execution/test_discover_snapshot.py
+++ b/tools/o11y-execution/test_discover_snapshot.py
@@ -24,6 +24,7 @@ def test_historical_input_does_not_backdate_plan_activation(self):
                 "--template", str(template), "--output", str(output), "--repetitions", "1",
             ], check=True)
             snapshot = json.loads(output.read_text())
+            self.assertEqual(snapshot["query_workload"]["repeating_queries"][0]["demand"], {"fixed_interval_at": {"interval": 60000, "evaluation_phase": 0}})
             environment = snapshot["environment"]
             self.assertEqual(environment["activation_unix_ms"], environment["observed_at_unix_ms"])
             provenance = json.loads(output.with_suffix(".provenance.json").read_text())

From 3e4803acecd2b697a0c3c3caa9103d402dd30abc Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 19:22:18 -0600
Subject: [PATCH 4/7] Verify VictoriaMetrics input visibility before measuring
 queries

---
 control_plane/src/physical/erp.rs             |  2 +-
 tools/o11y-execution/CALIBRATION.md           |  5 ++
 tools/o11y-execution/calibrate_runtime.py     | 46 ++++++++++++++++++-
 .../o11y-execution/test_calibrate_runtime.py  | 17 +++++++
 4 files changed, 68 insertions(+), 2 deletions(-)

diff --git a/control_plane/src/physical/erp.rs b/control_plane/src/physical/erp.rs
index aed6c9806..646162e21 100644
--- a/control_plane/src/physical/erp.rs
+++ b/control_plane/src/physical/erp.rs
@@ -838,7 +838,7 @@ mod tests {
             ProbabilityExpr::Unknown { .. }
         ));
         assert_eq!(
-            ReadoutEvidence::for_query(SketchAlgorithm::Hll, &SketchQuery::FrequencyEntropy),
+            ReadoutEvidence::for_query(&SketchAlgorithm::Hll, &SketchQuery::FrequencyEntropy),
             None
         );
     }
diff --git a/tools/o11y-execution/CALIBRATION.md b/tools/o11y-execution/CALIBRATION.md
index 10d652e5f..3a8c7ad85 100644
--- a/tools/o11y-execution/CALIBRATION.md
+++ b/tools/o11y-execution/CALIBRATION.md
@@ -109,3 +109,8 @@ bound. The held-out 5% target remains fixed, and failed configurations stay in t
 measurement history. A second optional argument writes the actual observed shape
 for bounded ERP matching. CPU fields in this offline artifact remain excluded;
 production process calibration measures CPU separately.
+
+Finite VictoriaMetrics replays call `/internal/force_flush` once after ingestion
+and charge it to build CPU. Accepted imports may otherwise remain invisible to
+queries for several seconds. This is a test barrier, not a production ingestion
+policy; see the [VictoriaMetrics forced-flush contract](https://docs.victoriametrics.com/victoriametrics/#forced-flush).
diff --git a/tools/o11y-execution/calibrate_runtime.py b/tools/o11y-execution/calibrate_runtime.py
index 5b4760a5e..e991b0eca 100644
--- a/tools/o11y-execution/calibrate_runtime.py
+++ b/tools/o11y-execution/calibrate_runtime.py
@@ -19,6 +19,41 @@
 from compare import compare_results, process_delta, process_snapshot
 
 
+def input_inventory(path):
+    counts, last = {}, {}
+    first = None
+    with path.open() as stream:
+        for labels, _, timestamp in runner.iter_samples(stream):
+            key = tuple(sorted(labels.items()))
+            counts[key] = counts.get(key, 0) + 1
+            last[key] = max(last.get(key, timestamp), timestamp)
+            first = timestamp if first is None else min(first, timestamp)
+    return counts, last, first, max(last.values())
+
+
+def verify_vm_visibility(url, inventory, folder):
+    counts, last, first, end = inventory
+    expected = [counts, {key: timestamp / 1000 for key, timestamp in last.items()}]
+    attempts = []
+    deadline = time.monotonic() + 60
+    while True:
+        valid = True
+        for index, function in enumerate(("count_over_time", "tlast_over_time")):
+            query = function + '({__name__!=""}[' + str(end - first + 1) + 'ms]) keep_metric_names'
+            response = runner._http_request(url + "/api/v1/query?" + urllib.parse.urlencode({"query":query,"time":end/1000,"nocache":1}))
+            rows = response["response"].get("data", {}).get("result", [])
+            actual = {tuple(sorted(row["metric"].items())): float(row["value"][1]) for row in rows}
+            matched = response["http_status"] == 200 and actual == expected[index]
+            attempts.append({"query":query,"matched":matched,**response})
+            valid = valid and matched
+        if valid or time.monotonic() >= deadline:
+            runner.write_json(folder / "exact-visibility.json", {"complete":valid,"attempts":attempts})
+            if not valid:
+                raise RuntimeError("VictoriaMetrics input counts/last timestamps are incomplete")
+            return
+        time.sleep(.1)
+
+
 def wait_ready(url, child):
     deadline = time.monotonic() + 60
     while time.monotonic() < deadline:
@@ -134,6 +169,14 @@ def launch(name, command):
         runner.write_json(folder / "drain.json", drained)
         if drained["http_status"] != 200 or drained["response"].get("complete") is not True:
             raise RuntimeError("precompute drain did not complete")
+        if args.victoriametrics:
+            # Finite replay barrier, charged to build: imported samples may still
+            # be buffered and invisible to queries after the write is accepted.
+            flushed = runner._http_request(fallback + "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/internal/force_flush")
+            runner.write_json(folder / "exact-flush.json", flushed)
+            if flushed["http_status"] != 200:
+                raise RuntimeError("VictoriaMetrics finite replay flush failed")
+            verify_vm_visibility(fallback, args.input_inventory, folder)
         after = snapshots(children)
         row["horizon_phases"]["ingest_and_build"] = phase(folder, "ingest_and_build", before, after, time.perf_counter_ns() - start)
         before, start = after, time.perf_counter_ns()
@@ -353,7 +396,8 @@ def main():
     if len({args.backend_port, args.fallback_port, args.metricsql_port}) != 3 or args.exact_cache_bytes <= 0:
         parser.error("distinct listener ports and positive exact cache budget required")
     args.output.mkdir(parents=True, exist_ok=False)
-    sample_count = runner.validate_sample_file(args.metrics)
+    args.input_inventory = input_inventory(args.metrics) if args.victoriametrics else None
+    sample_count = sum(args.input_inventory[0].values()) if args.input_inventory else runner.validate_sample_file(args.metrics)
     corpus, snapshot = json.loads(args.queries.read_text()), json.loads(args.snapshot.read_text())
     runner.validate_workload(snapshot, corpus)
     candidate_document = json.loads(args.candidates.read_text())
diff --git a/tools/o11y-execution/test_calibrate_runtime.py b/tools/o11y-execution/test_calibrate_runtime.py
index e638a8c38..74a0931dc 100644
--- a/tools/o11y-execution/test_calibrate_runtime.py
+++ b/tools/o11y-execution/test_calibrate_runtime.py
@@ -139,3 +139,20 @@ def test_result_cache_policy_is_explicit_on_both_query_endpoints(self):
         row = {"query": "count_over_time(m[1h])", "eval_timestamp_ms": 1234}
         self.assertNotIn("nocache", parse_qs(query_parameters(row, False)))
         self.assertEqual(parse_qs(query_parameters(row, True))["nocache"], ["1"])
+
+class VictoriaMetricsVisibilityTests(unittest.TestCase):
+    def test_incomplete_counts_are_retried_before_queries(self):
+        import tempfile
+        from pathlib import Path
+        from unittest.mock import patch
+        from calibrate_runtime import verify_vm_visibility
+        key = (("__name__", "m"),)
+        def answer(value):
+            return {"http_status":200,"response":{"data":{"result":[{"metric":dict(key),"value":[2,str(value)]}]}},"headers":{},"elapsed_ns":1}
+        with tempfile.TemporaryDirectory() as folder, patch("calibrate_runtime.runner._http_request", side_effect=[answer(1),answer(2),answer(2),answer(2)]) as request, patch("calibrate_runtime.time.sleep"):
+            verify_vm_visibility("http://test", ({key:2},{key:2000},1000,2000), Path(folder))
+            self.assertEqual(request.call_count, 4)
+            import json
+            report = json.loads((Path(folder)/"exact-visibility.json").read_text())
+            self.assertTrue(report["complete"])
+            self.assertFalse(report["attempts"][0]["matched"])

From 0b7afdbc505abac1b4a16737229f825ae6ac30b1 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 19:31:45 -0600
Subject: [PATCH 5/7] Measure the native exact engine without a backend proxy

---
 tools/o11y-execution/CALIBRATION.md          |  9 ++
 tools/o11y-execution/calibrate_runtime.py    |  2 +
 tools/o11y-execution/measure_native_exact.py | 91 ++++++++++++++++++++
 3 files changed, 102 insertions(+)
 create mode 100644 tools/o11y-execution/measure_native_exact.py

diff --git a/tools/o11y-execution/CALIBRATION.md b/tools/o11y-execution/CALIBRATION.md
index 3a8c7ad85..b95ec262b 100644
--- a/tools/o11y-execution/CALIBRATION.md
+++ b/tools/o11y-execution/CALIBRATION.md
@@ -114,3 +114,12 @@ Finite VictoriaMetrics replays call `/internal/force_flush` once after ingestion
 and charge it to build CPU. Accepted imports may otherwise remain invisible to
 queries for several seconds. This is a test barrier, not a production ingestion
 policy; see the [VictoriaMetrics forced-flush contract](https://docs.victoriametrics.com/victoriametrics/#forced-flush).
+
+`measure_native_exact.py` runs a separate fresh VictoriaMetrics process with no
+backend proxy. Use the same input, query corpus, CPU affinity, cache budget and
+cache policy as the candidate run. It records native query latencies, process
+CPU/RSS, lifecycle CPU and storage after shutdown. Report backend-only analytical
+resources separately from the candidate's combined backend + fallback service;
+retaining an exact service does not make its raw storage or memory disappear.
+Visibility validation scans can warm native data caches, so the first reported
+request is a first workload query after validation, not a cold-storage query.
diff --git a/tools/o11y-execution/calibrate_runtime.py b/tools/o11y-execution/calibrate_runtime.py
index e991b0eca..330f0b9b2 100644
--- a/tools/o11y-execution/calibrate_runtime.py
+++ b/tools/o11y-execution/calibrate_runtime.py
@@ -402,6 +402,8 @@ def main():
     runner.validate_workload(snapshot, corpus)
     candidate_document = json.loads(args.candidates.read_text())
     result = {"units": "cpu_ns", "compiler_identity": candidate_document.get("compiler_identity"), "data_snapshot_id": "sha256:" + hashlib.sha256(args.metrics.read_bytes()).hexdigest(),
+              "runtime_binary_sha256": hashlib.sha256(args.data_plane.read_bytes()).hexdigest(),
+              "exact_binary_sha256": hashlib.sha256((args.victoriametrics or args.prometheus).read_bytes()).hexdigest(),
               "exact_engine": "victoriametrics" if args.victoriametrics else "prometheus", "result_cache_disabled": args.disable_result_cache,
               "scope": "accelerated finite-input calibration; measured wall residency is not full logical-horizon residency", "validated_sample_count": sample_count, "candidates": []}
     candidates = candidate_document["candidates"]
diff --git a/tools/o11y-execution/measure_native_exact.py b/tools/o11y-execution/measure_native_exact.py
new file mode 100644
index 000000000..acfa57c3c
--- /dev/null
+++ b/tools/o11y-execution/measure_native_exact.py
@@ -0,0 +1,91 @@
+#!/usr/bin/env python3
+"""Measure a fresh native VictoriaMetrics baseline without a backend proxy."""
+import argparse
+import hashlib
+import json
+import os
+from pathlib import Path
+import resource
+import subprocess
+import time
+
+import calibrate_runtime as calibration
+import replay as runner
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('--victoriametrics', type=Path, required=True)
+    parser.add_argument('--metrics', type=Path, required=True)
+    parser.add_argument('--queries', type=Path, required=True)
+    parser.add_argument('--output', type=Path, required=True)
+    parser.add_argument('--cpu-affinity', required=True)
+    parser.add_argument('--port', type=int, default=19450)
+    parser.add_argument('--repetitions', type=int, default=60)
+    parser.add_argument('--exact-cache-bytes', type=int, default=268435456)
+    parser.add_argument('--disable-result-cache', action='store_true')
+    args = parser.parse_args()
+    if args.repetitions < 1 or args.exact_cache_bytes <= 0:
+        parser.error('positive repetitions and cache budget required')
+    cpus = {int(value) for value in args.cpu_affinity.split(',')}
+    inventory = calibration.input_inventory(args.metrics)
+    corpus = json.loads(args.queries.read_text())
+    if not corpus.get('upstream_revision') or not corpus.get('queries'):
+        raise ValueError('versioned nonempty corpus required')
+    args.output.mkdir(parents=True, exist_ok=False)
+    url = f'http://127.0.0.1:{args.port}'
+    command = calibration.exact_service_command(args, args.output, args.port)
+    report = {'engine':'victoriametrics','command':command,'result_cache_disabled':args.disable_result_cache,
+              'data_sha256':hashlib.sha256(args.metrics.read_bytes()).hexdigest(),
+              'binary_sha256':hashlib.sha256(args.victoriametrics.read_bytes()).hexdigest(),
+              'samples':sum(inventory[0].values()),'queries':{},'phases':{},
+              'scope':'native engine only; first workload query follows full input visibility validation'}
+    child = None
+    usage_before = resource.getrusage(resource.RUSAGE_CHILDREN)
+    try:
+        with (args.output/'native.log').open('w') as log:
+            child = subprocess.Popen(command, stdout=log, stderr=subprocess.STDOUT,
+                                     preexec_fn=lambda: os.sched_setaffinity(0, cpus))
+        children = {'native_exact':child}
+        calibration.wait_ready(url+'/health',child)
+        before = calibration.snapshots(children)
+        report['startup'] = before
+        started = time.perf_counter_ns()
+        runner.PROCESS_IDS.clear()
+        runner.PROCESS_IDS.update({'native_exact':child.pid})
+        runner.ingest_sample_file(args.metrics,[url],args.output)
+        flush = runner._http_request(url+'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/internal/force_flush')
+        runner.write_json(args.output/'exact-flush.json',flush)
+        if flush['http_status'] != 200:
+            raise RuntimeError('native flush failed')
+        calibration.verify_vm_visibility(url,inventory,args.output)
+        after = calibration.snapshots(children)
+        report['phases']['ingest_and_build'] = calibration.phase(args.output,'ingest_and_build',before,after,time.perf_counter_ns()-started)
+        for query in corpus['queries']:
+            before, started = calibration.snapshots(children), time.perf_counter_ns()
+            records = []
+            for repeat in range(args.repetitions):
+                result = runner._http_request(url+'/api/v1/query?'+calibration.query_parameters(query,args.disable_result_cache))
+                records.append({**query,'repetition':repeat,**result})
+                if result['http_status'] != 200 or result['response'].get('status') != 'success':
+                    raise RuntimeError('native query failed: '+query['query'])
+            after = calibration.snapshots(children)
+            measurement = calibration.phase(args.output,'query-'+query['id'],before,after,time.perf_counter_ns()-started)
+            records_path = args.output/('queries-'+query['id']+'.json')
+            runner.write_json(records_path,records)
+            report['queries'][query['id']] = {**measurement,'evaluations':len(records),'raw_measurement_file':str(records_path.resolve())}
+        report['final_processes'] = calibration.snapshots(children)
+        child.terminate()
+        child.wait(timeout=30)
+        usage_after = resource.getrusage(resource.RUSAGE_CHILDREN)
+        report['lifecycle_cpu_ns'] = int((usage_after.ru_utime+usage_after.ru_stime-usage_before.ru_utime-usage_before.ru_stime)*1e9)
+        report['storage_after_shutdown_bytes'] = calibration.file_bytes(args.output/'exact-data')
+        runner.write_json(args.output/'measurement.json',report)
+    finally:
+        if child is not None and child.poll() is None:
+            child.terminate()
+            child.wait(timeout=30)
+
+
+if __name__ == '__main__':
+    main()

From 44cd1a5d4e6bfd301236f2f2033c8f1890a8b229 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 19:37:31 -0600
Subject: [PATCH 6/7] fix(metricsql): preserve metric identity through value
 rollups

---
 .../asap_query_engine/logical_dag.rs          | 106 ++++++++++++-
 .../asap_query_engine/post_asap_readout.rs    | 141 +++++++++++++++++-
 .../storage_engines/sketch_db/index/mod.rs    |   7 +
 3 files changed, 250 insertions(+), 4 deletions(-)

diff --git a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs
index b07ecf64a..f2c392724 100644
--- a/data_plane/src/query_engines/asap_query_engine/logical_dag.rs
+++ b/data_plane/src/query_engines/asap_query_engine/logical_dag.rs
@@ -320,7 +320,24 @@ impl<F: FnMut(QueryNodeId, u64) -> Result<QueryResult, EngineError>> Evaluator<'
                                     }))
                                 }
                             };
-                            value.map(|v| (no_name(labels), v))
+                            value.map(|v| {
+                                let preserve_name = self.entry.language
+                                    == control_plane::query_plan::QueryLanguage::MetricsQl
+                                    && matches!(
+                                        operation,
+                                        TemporalOperation::Min
+                                            | TemporalOperation::Max
+                                            | TemporalOperation::Avg
+                                    );
+                                (
+                                    if preserve_name {
+                                        labels
+                                    } else {
+                                        no_name(labels)
+                                    },
+                                    v,
+                                )
+                            })
                         })
                         .collect(),
                 ))
@@ -851,6 +868,93 @@ mod topk_tests {
         assert_eq!(stats.raw_scan_evaluations, 0);
     }
 
+    // A temporal operator over an external subquery follows the same language
+    // policy as a summary readout; changing execution placement cannot drop names.
+    #[test]
+    fn metricsql_temporal_subdag_preserves_names_only_for_value_rollups() {
+        use control_plane::query_plan::QueryLanguage;
+        for language in [QueryLanguage::PromQl, QueryLanguage::MetricsQl] {
+            for operation in [
+                TemporalOperation::Max,
+                TemporalOperation::Min,
+                TemporalOperation::Avg,
+                TemporalOperation::Sum,
+                TemporalOperation::Count,
+                TemporalOperation::Rate,
+            ] {
+                let entry = QueryPlanEntry {
+                    language,
+                    query_id: "labels".into(),
+                    canonical_query: "test".into(),
+                    fixed_evaluation: None,
+                    root: QueryNodeId(1),
+                    nodes: BTreeMap::from([
+                        (
+                            QueryNodeId(0),
+                            QueryPlanNode::Logical {
+                                operator: LogicalOperator::ExactSubquery {
+                                    query: "m[1s]".into(),
+                                },
+                                inputs: vec![],
+                            },
+                        ),
+                        (
+                            QueryNodeId(1),
+                            QueryPlanNode::Logical {
+                                operator: LogicalOperator::Temporal { operation },
+                                inputs: vec![QueryNodeId(0)],
+                            },
+                        ),
+                    ]),
+                    instant: InstantExecution {
+                        lookback_ms: 1000,
+                        full_history: false,
+                        cumulative_readout: true,
+                    },
+                    fallback: FallbackPolicy::ExactBackend,
+                };
+                let leaves = BTreeMap::from([(
+                    (QueryNodeId(0), 1000),
+                    PreparedLeaf {
+                        value: Value::Matrix(
+                            vec![(
+                                labels(&[("__name__", "m"), ("job", "api")]),
+                                vec![(100, 1.), (900, 3.)],
+                            )],
+                            0,
+                            1000,
+                        ),
+                        remote: true,
+                        remote_evaluations: 1,
+                        remote_rpcs: 1,
+                    },
+                )]);
+                let (result, _) = execute_installed(&entry, &leaves, 1000, |_, _| {
+                    panic!("external child supplied")
+                })
+                .unwrap();
+                let QueryResult::Vector(result) = result else {
+                    panic!("vector required")
+                };
+                let expected = language == QueryLanguage::MetricsQl
+                    && matches!(
+                        operation,
+                        TemporalOperation::Max | TemporalOperation::Min | TemporalOperation::Avg
+                    );
+                assert_eq!(
+                    result.values[0]
+                        .label_keys_override
+                        .as_ref()
+                        .unwrap()
+                        .iter()
+                        .any(|name| name == "__name__"),
+                    expected,
+                    "{language:?} {operation:?}"
+                );
+            }
+        }
+    }
+
     #[test]
     fn installed_topk_ranks_exact_rate_summary_values() {
         let summary = QueryNodeId(0);
diff --git a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs
index a908595dd..4b0808e2a 100644
--- a/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs
+++ b/data_plane/src/query_engines/asap_query_engine/post_asap_readout.rs
@@ -154,6 +154,8 @@ enum PhysicalNodeError {
 }
 
 struct PhysicalQueryRuntime<'a> {
+    language: control_plane::query_plan::QueryLanguage,
+    catalog: Option<std::sync::Arc<asap_types::summary_catalog::SummaryCatalog>>,
     context: QueryExecutionContext<'a>,
 }
 
@@ -182,10 +184,37 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> {
                 reduce_sum_values(grouping, values, *coverage)
             }
             QueryPlanNode::ReadMaterialization { binding } => {
-                let groups = self
+                let mut groups = self
                     .context
                     .read_bound_materialization(binding)
                     .map_err(PhysicalNodeError::Store)?;
+                // Metric identity belongs to the shared DataDescriptor, not to
+                // the population labels or a reconstructed query string.
+                if self.language == control_plane::query_plan::QueryLanguage::MetricsQl
+                    && binding.output_grouping
+                        == control_plane::query_plan::PhysicalGrouping::PerEntity
+                {
+                    let metric = self
+                        .catalog
+                        .as_ref()
+                        .and_then(|catalog| {
+                            let definition =
+                                catalog.materializations.get(&binding.materialization)?;
+                            catalog
+                                .data_descriptors
+                                .get(&definition.data_descriptor_id)?
+                                .time_series_metric()
+                        })
+                        .ok_or_else(|| {
+                            PhysicalNodeError::Fallback(
+                                "MetricsQL per-series readout requires catalog metric identity"
+                                    .into(),
+                            )
+                        })?;
+                    for (labels, _) in &mut groups {
+                        labels.insert("__name__".into(), metric.into());
+                    }
+                }
                 Ok(PhysicalQueryOutput::State {
                     groups,
                     item_labels: binding.item_labels.clone(),
@@ -207,7 +236,17 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> {
                         .context
                         .readout_bound(state, &query)
                         .map_err(PhysicalNodeError::Store)?;
-                    let (rows, row_coverage) = expand_item_readout(key, value, item_labels)?;
+                    let (mut rows, row_coverage) = expand_item_readout(key, value, item_labels)?;
+                    if self.language == control_plane::query_plan::QueryLanguage::MetricsQl
+                        && !matches!(
+                            query,
+                            planner_types::post_asap::SketchQuery::Quantile { .. }
+                        )
+                    {
+                        for (labels, _) in &mut rows {
+                            labels.remove("__name__");
+                        }
+                    }
                     fold_coverage(&mut coverage, row_coverage);
                     values.extend(rows);
                 }
@@ -231,7 +270,17 @@ impl QueryNodeRuntime for PhysicalQueryRuntime<'_> {
                             )
                             .map(|value| {
                                 (
-                                    key.clone(),
+                                    {
+                                        let mut labels = key.clone();
+                                        if self.language
+                                            == control_plane::query_plan::QueryLanguage::MetricsQl
+                                            && *readout
+                                                != control_plane::query_plan::ExactReadout::Max
+                                        {
+                                            labels.remove("__name__");
+                                        }
+                                        labels
+                                    },
                                     SummaryValue::Points(
                                         vec![(self.context.t1_ms as i64, value)],
                                         state.exact_coverage(),
@@ -536,6 +585,8 @@ fn execute_physical_query_payload(
     is_cumulative: bool,
 ) -> Result<PostAsapReadoutOutcome, LoweringSkip> {
     let runtime = PhysicalQueryRuntime {
+        language: entry.language,
+        catalog: index.summary_catalog_snapshot(),
         context: QueryExecutionContext {
             index,
             t0_ms,
@@ -891,6 +942,90 @@ mod tests {
         idx
     }
 
+    // The same installed summary follows each language's metric-name semantics;
+    // spatial reduction must not invent a source metric on the aggregate.
+    #[test]
+    fn metricsql_quantile_preserves_catalog_metric_name_only_per_entity() {
+        use control_plane::query_plan::*;
+        let config: asap_types::PrecomputeMaterialization =
+            serde_json::from_value(serde_json::json!({
+                "aggregation_type": "DDSketch", "aggregation_sub_type": "",
+                "metric": "latency_ms", "window_size": 1, "slide_interval": 1,
+                "window_type": "tumbling", "num_aggregates_to_retain": 3,
+                "parameters": {"alpha": 0.01}, "pane_origin_ms": 0,
+                "partitioning": "per_entity", "window_layout": {"kind": "pane", "pane_secs": 1},
+                "grouping_labels": {"labels": []}, "aggregated_labels": {"labels": []},
+                "rollup_labels": {"labels": []}, "spatial_filter": "",
+                "spatial_filter_normalized": "", "original_yaml": ""
+            }))
+            .unwrap();
+        let idx = ddsketch_fixture();
+        let mut metadata = (*idx.instance(1).unwrap()).clone();
+        metadata.policy_fp = config.policy_fingerprint();
+        idx.install_summary_catalog(std::sync::Arc::new(
+            asap_types::summary_catalog::SummaryCatalog::from_materializations(
+                1,
+                1,
+                &[config.clone()],
+            )
+            .unwrap(),
+        ))
+        .unwrap();
+        idx.register(metadata);
+        let mut entry = QueryPlanEntry {
+            language: QueryLanguage::MetricsQl,
+            query_id: "quantile".into(),
+            canonical_query: "quantile_over_time(0.9, latency_ms[1s])".into(),
+            fixed_evaluation: None,
+            root: QueryNodeId(0),
+            nodes: BTreeMap::from([
+                (
+                    QueryNodeId(0),
+                    QueryPlanNode::SummaryEstimate {
+                        input: QueryNodeId(1),
+                        query: QueryReadout::Quantile { q: 0.9 },
+                    },
+                ),
+                (
+                    QueryNodeId(1),
+                    QueryPlanNode::ReadMaterialization {
+                        binding: MaterializationBinding {
+                            materialization: config.policy_fingerprint().into(),
+                            output_grouping: PhysicalGrouping::PerEntity,
+                            item_labels: vec![],
+                            window_ms: 1000,
+                            pane_origin_ms: Some(0),
+                            readout_lookback_ms: Some(1000),
+                        },
+                    },
+                ),
+            ]),
+            instant: InstantExecution {
+                lookback_ms: 1000,
+                full_history: false,
+                cumulative_readout: true,
+            },
+            fallback: FallbackPolicy::ExactBackend,
+        };
+        let result = execute_query_plan_readout(&idx, &entry, 1000, 2000, true).unwrap();
+        assert_eq!(
+            result.series[0].0.get("__name__").map(String::as_str),
+            Some("latency_ms")
+        );
+        entry.language = QueryLanguage::PromQl;
+        let result = execute_query_plan_readout(&idx, &entry, 1000, 2000, true).unwrap();
+        assert!(!result.series[0].0.contains_key("__name__"));
+        entry.language = QueryLanguage::MetricsQl;
+        let QueryPlanNode::ReadMaterialization { binding } =
+            entry.nodes.get_mut(&QueryNodeId(1)).unwrap()
+        else {
+            unreachable!()
+        };
+        binding.output_grouping = PhysicalGrouping::Reduce(vec![]);
+        let result = execute_query_plan_readout(&idx, &entry, 1000, 2000, true).unwrap();
+        assert!(!result.series[0].0.contains_key("__name__"));
+    }
+
     #[test]
     fn formal_query_plan_executes_only_its_bound_policy() {
         let idx = SketchStore::new();
diff --git a/data_plane/src/storage_engines/sketch_db/index/mod.rs b/data_plane/src/storage_engines/sketch_db/index/mod.rs
index 8353d646c..cafb6eb8a 100644
--- a/data_plane/src/storage_engines/sketch_db/index/mod.rs
+++ b/data_plane/src/storage_engines/sketch_db/index/mod.rs
@@ -752,6 +752,13 @@ impl SketchStore {
             .map_err(|error| error.to_string())
     }
 
+    /// Share the installed metadata snapshot without copying descriptors or state.
+    pub(crate) fn summary_catalog_snapshot(
+        &self,
+    ) -> Option<Arc<asap_types::summary_catalog::SummaryCatalog>> {
+        self.descriptors.authoritative_catalog()
+    }
+
     /// Record that `sid` is a per-item (item_label-mode) frequency sketch
     /// keyed by the data-point attribute `label` (e.g. "service"). The
     /// query engine consults this to decide whether a keyed selector like

From 467ebcb8bec5ad6b7ac5ff4ad5d910a452ef1d09 Mon Sep 17 00:00:00 2001
From: zz_y <zeyingz@umd.edu>
Date: Thu, 10 Sep 2026 21:25:57 -0600
Subject: [PATCH 7/7] docs(eval): report independent cost-selected HLL results

---
 ...ost-selected-victoriametrics-evaluation.md | 74 +++++++++++++++++++
 1 file changed, 74 insertions(+)
 create mode 100644 docs/developer_docs/hll-cost-selected-victoriametrics-evaluation.md

diff --git a/docs/developer_docs/hll-cost-selected-victoriametrics-evaluation.md b/docs/developer_docs/hll-cost-selected-victoriametrics-evaluation.md
new file mode 100644
index 000000000..32319e8fa
--- /dev/null
+++ b/docs/developer_docs/hll-cost-selected-victoriametrics-evaluation.md
@@ -0,0 +1,74 @@
+# Cost-selected HLL evaluation against VictoriaMetrics
+
+A fresh evaluation of the control plane's selected HLL plan reduced query latency
+and CPU for one generated `distinct_over_time` workload. It did **not** reduce
+memory. This is a sensitivity experiment, not coverage of the original o11ybench
+queries.
+
+The input contains 1,440,040 samples: 40 series, one hour, 100 ms sampling. The
+query selects 28 held-out series:
+
+```promql
+distinct_over_time(data{label_0=~"g00000[3-9]"}[1h])
+```
+
+Calibration used separate hash populations and a predeclared HLL precision grid
+of 10, 12, and 14. The measured accuracy evidence admitted precision 12 for an
+empirical 5% relative-error target. This is not an epsilon-delta guarantee.
+Complete candidate CPU calibration included the backend and its external exact
+service. The ordinary control-plane compiler selected HLL precision 12 with an
+estimated cost of 14.478 CPU seconds versus 20.920 seconds for its exact
+alternative, for a demand of 60 query evaluations. The benchmark did not select
+the winner.
+
+An independent run installed that selected plan in fresh processes and compared
+it with a fresh native VictoriaMetrics v1.126.0 process:
+
+| Metric | ASAP backend + external VM | Native VM |
+| --- | ---: | ---: |
+| Median query latency | 13.174 ms | 62.564 ms |
+| Query CPU, 60 evaluations | 0.690 s | 13.790 s |
+| Full child-process lifecycle CPU (`wait4`) | 15.264 s | 16.735 s |
+| Query-phase RSS | 138.6 MB combined; 84.1 MB backend | 63.7 MB |
+| Observed process peak RSS | 182.2 MB backend; 72.1 MB external VM | 85.9 MB |
+| Result completeness | 28/28 series, 60/60 evaluations | Reference |
+| Maximum relative result error | 4.557% | Reference |
+
+All 60 selected-plan evaluations reported warm execution, without exact subtree
+RPCs. The latency improvement was 4.75×, query CPU about 20× lower, and combined
+finite-run lifecycle CPU 8.79% lower. Sum of separate process peaks is not a
+simultaneous deployment peak. SummaryStore reported 274,466 approximate resident
+bytes; that number is not a durable storage footprint.
+
+Both deployments used the same four CPU affinity cores in sequential quiet
+measurement windows. Result caches were disabled. Input visibility validation
+ran before workload queries and warmed data caches. All repetitions used the
+same evaluation timestamp; this does not measure moving windows, concurrent
+throughput, cold storage, or one hour of wall-clock residency. The ASAP
+configuration retained and ingested a complete external VM, whose CPU and memory
+are included above. Driver and separate correctness-oracle resources are
+excluded. The VM cache budget was 256 MiB, not a process RSS limit.
+
+## Reproduction artifacts
+
+The dataset is at
+`/mydata/univmon-benefit-study/datasets/100ms-1h-10g4m/`.
+The artifact root is `/mydata/univmon-benefit-study/`:
+
+- `distinct-revision-aligned-costed.json`: input to normal compiler selection.
+- `distinct-revision-aligned-selected.json`: selected plan, alternatives, costs,
+  catalog, and executable installation request.
+- `calibration-revision-aligned-cache-off/`: fresh candidate calibration.
+- `independent-cost-selected-cache-off/`: selected-plan independent execution.
+- `independent-native-cache-off/`: adjacent native baseline.
+- `HLL_SELECTED_RESULT.json`: machine-readable results and limitations.
+
+Export and runtime revision:
+`dd102e07fb274725edf9819ff202864878633d2d`.
+Planner revision: `7e7931b581f4941ad02e4c6d90d41f2e95b2d7fc`.
+Sketch library revision: `079ba44936bb74f7e0f4374936e8846266b35243`.
+The result artifact also records both executable digests and the input digest.
+Earlier measurements with mismatched candidate-export provenance remain in the
+artifact directory; they were not reused as cost quotes for this result.
+
+See [the calibration workflow](../../tools/o11y-execution/CALIBRATION.md).