Measure MetricsQL candidates against VictoriaMetrics

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(

control_plane/examples/compile_workload_artifact.rs

@@ -5,15 +5,19 @@ 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(
             "execution evaluation requires version 2 complete workload cost evidence".into(),
         );
     }
     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

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)
+                }
             }
         }
     }
@@ -4253,6 +4269,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])");

data_plane/examples/univmon_erp_artifact.rs

@@ -0,0 +1,204 @@
+//! 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};
+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 + 1_000_000_000_000.0)
+                    .collect::<Vec<_>>(),
+            )
+        })
+        .collect();
+    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;
+        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();
+        observed_shape = Some(observed.observation.clone());
+        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:?}");
+    }
+    // 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(
+            &json!({"schema_version":1,"producer_version":"backend-standard-unit-frequency-merged-two-pane-data-calibration-cpu-excluded","records":records})
+        )?
+    );
+    Ok(())
+}