bench: measure ClickHouse and MetricsQL o11y paths

control_plane/examples/audit_metricsql_compile.rs

@@ -0,0 +1,81 @@
+use control_plane::physical::{
+    compiler::{BackendLocalPlanningSnapshot, PhysicalCompiler},
+    post_asap::cost_model::ForcedFamilyCostModel,
+};
+use planner_types::post_asap::SketchAlgorithm;
+use serde::Deserialize;
+use serde_json::{json, Value};
+use std::fs;
+#[derive(Deserialize)]
+struct C {
+    queries: Vec<R>,
+}
+#[derive(Deserialize)]
+struct R {
+    id: String,
+    metricsql: String,
+}
+fn main() {
+    let c: C =
+        serde_json::from_slice(&fs::read(std::env::args().nth(1).unwrap()).unwrap()).unwrap();
+    let template: Value = serde_json::from_str(include_str!(
+        "../../docs/examples/asapquery-compatibility-demo-snapshot.json"
+    ))
+    .unwrap();
+    let mut out = vec![];
+    for row in c.queries {
+        let mut fixture = template.clone();
+        let mut demand = fixture["query_workload"]["repeating_queries"][0].clone();
+        demand["query"] = row.metricsql.clone().into();
+        fixture["query_workload"]["repeating_queries"] = json!([demand]);
+        fixture["implementation"]["topk_evidence"] = json!({});
+        let snapshot: BackendLocalPlanningSnapshot = serde_json::from_value(fixture).unwrap();
+        let (mut req, env) = snapshot.planning_request().unwrap();
+        req.hybrid_execution = false;
+        let accuracy = req.queries[0].accuracy.clone();
+        let expr = match asap_frontend_metricsql::lower_metricsql(&row.metricsql, accuracy.clone())
+        {
+            Ok(x) => x,
+            Err(e) => {
+                out.push(
+                    json!({"id":row.id,"compile":{"status":"not_reached","reason":e.to_string()}}),
+                );
+                continue;
+            }
+        };
+        if let Err(e) =
+            control_plane::physical::compiler::validate_metricsql_acceleration_shape(&expr)
+        {
+            out.push(json!({"id":row.id,"compile":{"status":"not_reached","reason":e}}));
+            continue;
+        }
+        req.queries[0].query_string = row.metricsql.clone();
+        req.queries[0].post_asap = match control_plane::planner_selection::select_summary(
+            &expr,
+            &ForcedFamilyCostModel::new(accuracy, SketchAlgorithm::Kll),
+        ) {
+            Ok(x) => x,
+            Err(e) => {
+                out.push(json!({"id":row.id,"compile":{"status":"typed_fallback","reason":e.to_string()}}));
+                continue;
+            }
+        };
+        match PhysicalCompiler.compile_metricsql(req, env) {
+            Ok(plan) => {
+                let pub_ok = plan.publication().and_then(|p| {
+                    serde_json::to_vec(&p)
+                        .map(|_| ())
+                        .map_err(|e| e.to_string())
+                });
+                out.push(json!({"id":row.id,"compile":{"status":"pass"},"publication":{"status":if pub_ok.is_ok(){"pass"}else{"failed"},"reason":pub_ok.err()}}))
+            }
+            Err(e) => out.push(
+                json!({"id":row.id,"compile":{"status":"typed_fallback","reason":e.to_string()}}),
+            ),
+        }
+    }
+    println!(
+        "{}",
+        serde_json::to_string_pretty(&json!({"queries":out})).unwrap()
+    )
+}

control_plane/examples/audit_multilang_corpus.rs

@@ -0,0 +1,79 @@
+use asap_frontend_sql::SqlCatalog;
+use control_plane::physical::post_asap::bind_query_expr;
+use planner_types::pre_asap::{Column, DataType, Schema};
+use planner_types::types::AccuracyTarget;
+use serde::Deserialize;
+use serde_json::json;
+use std::{fs, path::PathBuf};
+#[derive(Deserialize)]
+struct Corpus {
+    queries: Vec<Row>,
+}
+#[derive(Deserialize)]
+struct Row {
+    id: String,
+    metricsql: String,
+    clickhouse_sql: Option<String>,
+}
+#[tokio::main]
+async fn main() {
+    let path = std::env::args_os()
+        .nth(1)
+        .map(PathBuf::from)
+        .expect("corpus path");
+    let corpus: Corpus = serde_json::from_slice(&fs::read(path).unwrap()).unwrap();
+    let schema = Schema::with_time_index(
+        vec![
+            Column::new("metric", DataType::Utf8, false),
+            Column::new("labels", DataType::Utf8, false),
+            Column::new("ts_ms", DataType::Timestamp, false),
+            Column::new("value", DataType::Float64, false),
+        ],
+        2,
+        vec![],
+    );
+    let catalog = SqlCatalog::new().with_table("raw_samples", schema);
+    let accuracy = AccuracyTarget::Epsilon(0.01);
+    let mut rows = Vec::new();
+    for row in corpus.queries {
+        let ml = asap_frontend_metricsql::lower_metricsql(&row.metricsql, accuracy.clone());
+        let (mcanon, mplan) = match ml {
+            Ok(expr) => (
+                json!({"status":"pass"}),
+                match control_plane::physical::compiler::validate_metricsql_acceleration_shape(
+                    &expr,
+                )
+                .and_then(|_| {
+                    bind_query_expr(&expr, accuracy.clone())
+                        .map(|_| ())
+                        .map_err(|e| Box::leak(e.to_string().into_boxed_str()) as &str)
+                }) {
+                    Ok(_) => json!({"status":"pass"}),
+                    Err(e) => json!({"status":"typed_fallback","reason":e}),
+                },
+            ),
+            Err(e) => (
+                json!({"status":"failed","reason":e.to_string()}),
+                json!({"status":"not_reached"}),
+            ),
+        };
+        let sql = match row.clickhouse_sql {
+            Some(sql) => match control_plane::clickhouse::plan_clickhouse_sql(
+                &sql.replace("{eval_ms}", "1788891296000"),
+                &catalog,
+                accuracy.clone(),
+            )
+            .await
+            {
+                Ok(_) => json!({"parser_canonical":"pass","planner":"pass"}),
+                Err(e) => json!({"status":"typed_fallback","reason":e.to_string()}),
+            },
+            None => json!({"status":"missing_mapping"}),
+        };
+        rows.push(json!({"id":row.id,"metricsql":{"parser_canonical":mcanon,"planner":mplan},"clickhouse_sql":sql}));
+    }
+    println!(
+        "{}",
+        serde_json::to_string_pretty(&json!({"queries":rows})).unwrap()
+    );
+}

control_plane/examples/emit_benchmark_physical_plan.rs

@@ -0,0 +1,22 @@
+use serde_json::json;
+
+fn main() {
+    let snapshot: control_plane::physical::compiler::BackendLocalPlanningSnapshot =
+        serde_json::from_str(include_str!(
+            "../../docs/examples/asapquery-compatibility-demo-snapshot.json"
+        ))
+        .expect("checked-in planning snapshot");
+    let plan = snapshot.compile().expect("compile physical plan");
+    let artifact = json!({
+        "summary_catalog": plan.summary_catalog,
+        "collector_plans": plan.collector_plans,
+        "precompute_plan": plan.precompute_plan,
+        "transmission_plan": plan.transmission_plan,
+        "query_plan": plan.query_plan,
+        "metricsql_plan": plan.metricsql_plan,
+        "clickhouse_sql": null,
+        "storage_routing": null,
+        "adaptation_evidence": [],
+    });
+    serde_json::to_writer(std::io::stdout(), &artifact).expect("write artifact");
+}

docs/benchmarks/o11y-multilang-benefit.md

@@ -0,0 +1,48 @@
+# O11y multi-language fallback experiment
+
+This benchmark keeps all 27 source occurrences in the denominator. `corpus.json` preserves each PromQL expression byte-for-byte as MetricsQL and contains its exact ClickHouse SQL over `raw_samples(metric, labels, ts_ms, value)`. The SQL implements counter reset and boundary extrapolation, offset, subquery grids, and classic histogram interpolation rather than importing answers from another engine.
+
+## Reproduce
+
+The command below starts with an empty output directory. It deterministically generates the OpenMetrics fixture, Prometheus configuration and TSDB, and physical plan; records their SHA-256 hashes; builds the data-plane binary from that same clean HEAD; runs both production frontend/compiler auditors; provisions fresh Prometheus, VictoriaMetrics, ClickHouse, and data-plane state; then writes raw requests, structured comparisons, phase resources, and terminal-stage coverage.
+
+```bash
+CARGO_TARGET_DIR=/path/to/target python3 tools/o11y-multilang/reproduce.py \
+  --backend-source "$PWD" \
+  --output-dir /tmp/o11y-repro-fresh \
+  --trials 1 --repetitions 3 --seed 20260910
+```
+
+The command never changes tracked evidence. After review, promote only the designated JSON artifacts explicitly:
+
+```bash
+cp /tmp/o11y-repro-fresh/{manifest,source-provenance,frontend-planner,metricsql-production,stage-pre-runtime,stage-coverage,trial-0-raw,trial-0-comparisons,trial-0-latency-summary}.json \
+  tools/o11y-multilang/repro-fresh/
+```
+
+The seed randomizes query order independently for each repetition. Engine order alternates. Each trial creates new storage and process state and removes it afterward. Container images are launched by immutable digest and the manifest also records their actual image IDs. Resource snapshots are totals for the mixed alternating query phase and cannot be attributed to native or ASAP mode independently. The manifest records the clean source HEAD, binary hash, generated-input hashes, lifecycle and ingest duration, and process CPU ticks, RSS/HWM, and storage at start, after ingest, and after queries.
+
+## Observed result
+
+The checked-in evidence is one fresh trial with three repetitions and 27 queries, or 81 requests per engine. All five endpoints completed 81/81 requests. Both ASAP listeners reported `exact_fallback` for 81/81, so this result measures fallback overhead and does not establish acceleration benefit.
+
+- VictoriaMetrics median/p95: 2.36/9.12 ms; ASAP MetricsQL fallback: 5.03/6.34 ms.
+- ClickHouse median/p95: 49.92/169.56 ms; ASAP ClickHouse fallback: 56.25/179.34 ms.
+- Native ClickHouse versus ASAP ClickHouse: 81/81 structured matches.
+- Native VictoriaMetrics versus ASAP MetricsQL: 81/81 structured matches.
+- Prometheus versus VictoriaMetrics: 48/81 strict matches and 33/81 mismatches.
+- Prometheus versus ClickHouse label/value SQL oracle: 81/81 matches; timestamps and result type are reported as protocol-noncomparable.
+
+The offline fail-closed matrix reports that the parser/canonical stage accepts 12/27 MetricsQL expressions, the early planner accepts 7/27, and the production compiler/publication validator accepts 2/27. The SQL acceleration frontend accepts 0/27 of the exact ClickHouse dialect mappings. Every row records its offline terminal stage plus the HTTP-observed exact fallback. Although q03 and q16 pass offline publication validation, the benchmark physical plan deliberately contains no corpus sidecars; their catalog-miss classification is inferred from that artifact, while only exact fallback is observed on the wire. Binder, validator, and executor are not reached in this fallback-only experiment.
+
+## Evidence files
+
+- `repro-fresh/stage-coverage.json`: per-query parser, planner, compiler, publication, terminal, adapter, and fallback result.
+- `repro-fresh/manifest.json`: hashes, immutable images, lifecycle timing, and phase resource snapshots.
+- `repro-fresh/trial-0-raw.json`: every timed response and `x-asap-execution` value.
+- `repro-fresh/trial-0-comparisons.json`: labels, value, timestamp, result type, and warnings for each required engine pair.
+- `repro-fresh/trial-0-latency-summary.json`: query-only latency summary.
+
+## Limitations
+
+This is one fresh trial and does not estimate variance across trials. CPU is process scheduler ticks rather than normalized CPU time, and the process totals cover mixed alternating native and ASAP requests rather than either mode alone. Container writable-layer size is an operational proxy for VM and ClickHouse storage. The production corpus has no warm executions, so it cannot quantify acceleration benefit. Native-histogram exponential interpolation is outside q21, which consumes classic `_bucket` series.

tools/o11y-multilang/build_counter_sql.py

@@ -0,0 +1,35 @@
+#!/usr/bin/env python3
+import json
+from pathlib import Path
+E='{eval_ms}'
+def series(metric,window,offset=0,rate=False):
+ end=f'({E}-{offset})'; start=f'({end}-{window})'
+ divisor=f'/({window}/1000)' if rate else ''
+ return f'''SELECT labels, corrected*(sampled+least(start_extra,if(corrected>0,sampled*(first_value/corrected),start_extra))+end_extra)/sampled{divisor} AS value FROM (SELECT labels,samples,length(samples) n,samples[1].1 first_ts,samples[n].1 last_ts,samples[1].2 first_value,samples[n].2 last_value,(last_ts-first_ts)/1000 sampled,(last_value-first_value)+arraySum(i -> if(samples[i].2<samples[i-1].2,samples[i-1].2,0.),range(2,n+1)) corrected,if((first_ts-{start})/1000<sampled/(n-1)*1.1,(first_ts-{start})/1000,sampled/(n-1)/2) start_extra,if(({end}-last_ts)/1000<sampled/(n-1)*1.1,({end}-last_ts)/1000,sampled/(n-1)/2) end_extra FROM (SELECT labels,arraySort(x -> x.1,groupArray((ts_ms,value))) samples FROM raw_samples WHERE metric='{metric}' AND ts_ms>={start} AND ts_ms<={end} GROUP BY labels) WHERE n>=2)'''
+def agg(src,by=None):
+ if by:return f"SELECT map('{by}',job) labels,sum(value) value FROM (SELECT labels['{by}'] job,value FROM ({src})) GROUP BY job"
+ return f"SELECT map() labels,sum(value) value FROM ({src})"
+def ratio(a,b):return f"SELECT a.labels,a.value/b.value value FROM ({a}) a INNER JOIN ({b}) b ON a.labels=b.labels"
+def top(src,k):return f"SELECT * FROM ({src}) ORDER BY value DESC,labels LIMIT {k}"
+def filt(src):return f"SELECT * FROM ({src}) WHERE value>0"
+inc=lambda m,w,o=0:series(m,w,o,False); rate=lambda m,w,o=0:series(m,w,o,True)
+job=lambda x:agg(x,'job'); glob=lambda x:agg(x)
+ratio6=ratio(job(inc('backend_http_5xx_total',21600000)),job(inc('backend_http_requests_total',21600000)))
+M={
+'q01':f'SELECT * FROM ({ratio6}) ORDER BY value DESC,labels','q02':top(ratio6,1),
+'q03':ratio(glob(inc('payment_service_http_5xx_total',3600000)),glob(inc('payment_service_http_requests_total',3600000))),
+'q04':ratio(glob(inc('payment_service_http_5xx_total',3600000,21600000)),glob(inc('payment_service_http_requests_total',3600000,21600000))),
+'q08':glob(rate('backend_process_cpu_seconds_total',3600000)),
+'q11':top(job(rate('backend_process_cpu_seconds_total',3600000)),2),
+'q14':glob(rate('backend_http_requests_total',300000)),'q15':job(rate('backend_http_requests_total',300000)),
+'q16':ratio(glob(inc('backend_http_5xx_total',3600000)),glob(inc('backend_http_requests_total',3600000))),
+'q17':top(ratio6,1),'q18':filt(job(inc('backend_http_5xx_total',86400000))),
+'q19':glob(rate('order_service_http_requests_total',300000)),'q20':glob(rate('order_service_http_requests_total',300000,3600000)),
+'q22':glob(rate('order_service_http_requests_total',300000)),
+'q25':top(f"SELECT a.labels,a.value/b.value value FROM ({job(inc('backend_http_5xx_total',86400000))}) a CROSS JOIN ({glob(inc('backend_http_5xx_total',86400000))}) b",1),
+'q26':top(job(rate('backend_process_cpu_seconds_total',21600000)),1),
+}
+p=Path(__file__).with_name('corpus.json');x=json.loads(p.read_text())
+for r in x['queries']:
+ if r['id'] in M:r['clickhouse_sql']=M[r['id']];r['sql_mapping_status']='oracle_pending'
+p.write_text(json.dumps(x,indent=2)+'\n')

tools/o11y-multilang/build_exact_sql.py

@@ -0,0 +1,18 @@
+#!/usr/bin/env python3
+"""Attach native ClickHouse SQL for the first exact-semantics pattern batch."""
+import json
+from pathlib import Path
+E='{eval_ms}'
+SQL={
+'q05': f"""SELECT labels, max(value) AS value FROM raw_samples WHERE metric='cache_refresh_lag_seconds' AND ts_ms>{E}-43200000 AND ts_ms<={E} GROUP BY labels ORDER BY labels""",
+'q06': f"""SELECT labels, max(value) AS value FROM raw_samples WHERE metric='user_service_cache_refresh_lag_seconds' AND ts_ms>{E}-43200000 AND ts_ms<={E} GROUP BY labels ORDER BY labels""",
+'q07': f"""SELECT mapConcat(labels,map('__name__','user_service_cache_refresh_lag_seconds')) AS labels, argMax(value,ts_ms) AS value FROM raw_samples WHERE metric='user_service_cache_refresh_lag_seconds' AND ts_ms>{E}-300000 AND ts_ms<={E} GROUP BY labels ORDER BY labels""",
+'q09': f"""SELECT map() AS labels, sum(value) AS value FROM (SELECT labels,argMax(value,ts_ms) AS value FROM raw_samples WHERE metric='backend_process_resident_memory_bytes' AND ts_ms>{E}-300000 AND ts_ms<={E} GROUP BY labels)""",
+'q12': f"""SELECT map('job',job) AS labels,sum(value) AS value FROM (SELECT labels['job'] AS job,labels,argMax(value,ts_ms) AS value FROM raw_samples WHERE metric='backend_process_resident_memory_bytes' AND ts_ms>{E}-300000 AND ts_ms<={E} GROUP BY job,labels) GROUP BY job ORDER BY value DESC,job LIMIT 2""",
+'q23': f"""SELECT labels,max(value) AS value FROM raw_samples WHERE metric='backend_retry_backlog_depth' AND ts_ms>{E}-21600000 AND ts_ms<={E} GROUP BY labels ORDER BY value DESC,labels LIMIT 2""",
+}
+p=Path(__file__).with_name('corpus.json'); corpus=json.loads(p.read_text())
+for row in corpus['queries']:
+ row['clickhouse_sql']=SQL.get(row['id'])
+ row['sql_mapping_status']='oracle_pending' if row['id'] in SQL else 'later_pattern_batch'
+p.write_text(json.dumps(corpus,indent=2)+'\n')

tools/o11y-multilang/build_histogram_sql.py

@@ -0,0 +1,11 @@
+#!/usr/bin/env python3
+import json
+from pathlib import Path
+# Reuse the reviewed counter template without importing side-effect script.
+ns={}; exec(Path(__file__).with_name('build_counter_sql.py').read_text().split("p=Path(__file__)")[0],ns)
+src=ns['rate']('order_service_http_request_duration_seconds_bucket',300000)
+sql=f'''SELECT map() labels,if(idx=length(buckets),buckets[idx-1].1,if(idx=1 AND buckets[idx].1<=0,buckets[idx].1,(if(idx=1,0.,buckets[idx-1].1)+(buckets[idx].1-if(idx=1,0.,buckets[idx-1].1))*(rank-if(idx=1,0.,buckets[idx-1].2))/(buckets[idx].2-if(idx=1,0.,buckets[idx-1].2))))) value FROM (SELECT buckets,0.95*buckets[length(buckets)].2 rank,arrayFirstIndex(x->x.2>=rank,buckets) idx FROM (SELECT arrayMap(i->(raw[i].1,arrayMax(arrayMap(x->x.2,arraySlice(raw,1,i)))),range(1,length(raw)+1)) buckets FROM (SELECT arraySort(x->x.1,groupArray((if(le='+Inf',inf,toFloat64(le)),value))) raw FROM (SELECT labels['le'] le,sum(value) value FROM ({src}) GROUP BY le)))) WHERE idx>0'''
+p=Path(__file__).with_name('corpus.json');x=json.loads(p.read_text())
+for r in x['queries']:
+ if r['id']=='q21':r['clickhouse_sql']=sql;r['sql_mapping_status']='oracle_pending'
+p.write_text(json.dumps(x,indent=2)+'\n')

tools/o11y-multilang/build_stage_report.py

@@ -0,0 +1,9 @@
+#!/usr/bin/env python3
+import argparse,json
+from pathlib import Path
+p=argparse.ArgumentParser();p.add_argument('--corpus',type=Path,required=True);p.add_argument('--frontend',type=Path,required=True);p.add_argument('--production',type=Path,required=True);p.add_argument('--output',type=Path,required=True);a=p.parse_args()
+corpus=json.loads(a.corpus.read_text());front={r['id']:r for r in json.loads(a.frontend.read_text())['queries']};production={r['id']:r for r in json.loads(a.production.read_text())['queries']};out=[]
+for q in corpus['queries']:
+ f=front[q['id']];m=f['metricsql'];pc=production[q['id']];sf=f['clickhouse_sql'];sp=sf.get('status','pass')
+ langs={'metricsql':{'parser_canonical':m['parser_canonical'],'planner':m['planner'],'compiler':pc['compile'],'publication':pc.get('publication',{'status':'not_reached'}),'binder':{'status':'not_reached'},'validator':{'status':'not_reached'},'executor':{'status':'not_reached'}},'clickhouse_sql':{'exact_sql_oracle':{'status':q['sql_mapping_status']},'parser_canonical':sf,'planner':{'status':'not_reached' if sp!='pass' else 'pass'},'publication':{'status':'not_reached'},'binder':{'status':'not_reached'},'validator':{'status':'not_reached'},'executor':{'status':'not_reached'}}};out.append({'id':q['id'],'operators':q['operators'],'languages':langs})
+report={'schema_version':2,'denominator':27,'evidence':{'metricsql_parser_pass':sum(r['languages']['metricsql']['parser_canonical']['status']=='pass' for r in out),'metricsql_planner_pass':sum(r['languages']['metricsql']['planner']['status']=='pass' for r in out),'metricsql_compiler_pass':sum(r['languages']['metricsql']['compiler']['status']=='pass' for r in out),'metricsql_publication_pass':sum(r['languages']['metricsql']['publication']['status']=='pass' for r in out),'sql_acceleration_frontend_pass':sum(r['languages']['clickhouse_sql']['parser_canonical'].get('status','pass')=='pass' for r in out)},'queries':out};a.output.write_text(json.dumps(report,indent=2)+'\n')