feat(asap-tools): add accelerated random E2E workload artifacts

.design_docs/accelerated-random-workload-eval.md

@@ -5,7 +5,7 @@
 Run a practical, end-to-end version of the paper's §6.2 experiment before the
 full PR 862 workload.  The experiment should retain the important workload
 shape—many input series and few returned series—and show the benefit of
-planning a *workload* of related RQEs rather than isolated queries.
+planning a *workload* of related repeating query expressions (RQEs) rather than isolated queries.
 
 The experiment has ten deterministic random workloads.  Each workload contains
 exactly ten atomic user queries; binary arithmetic queries are out of scope for
@@ -51,20 +51,21 @@ unbounded.
 
 | Lookback `W` | Allowed repeat intervals `T` |
 | ---: | ---: |
-| 100 ms | 1 s |
+| 100 s | 1 s |
 | 1 min | 1 s, 2 s |
 | 5 min | 5 s, 10 s |
 | 10 min | 10 s, 20 s |
 | 20 min | 30 s, 60 s |
 
 The maximum accelerated backfill is therefore only 20 minutes of event-time
-history.  The first 100 ms class is for spatial queries; temporal queries use
-the 1--20 minute classes.  The spatial class targets a 100 ms repetition
-interval.  Current exact-MV support is whole-second only, so this class is
-gated on [#876](https://github.com/ProjectASAP/ASAPQuery/issues/876), which
-adds millisecond interval metadata, `DateTime64` bucket flooring and edge
-handling, a sub-second query schedule, and raw-versus-MV equivalence tests.
-This is a limitation of the current baseline generator, not of ClickHouse.
+history.  The first 100 s class is for quantile queries; the other temporal
+queries use the 1--20 minute classes.  Scraping remains every 100 ms, so each
+quantile query consumes 1,000 samples per leaf series: the first N that
+sketch-bench's accuracy curves cover, below which no saved plan can serve it.
+Its whole-second repetition and lookback are compatible with the current
+exact-MV arm; it is not gated on
+[#876](https://github.com/ProjectASAP/ASAPQuery/issues/876).  That issue is
+needed only for a future sub-second query/MV schedule.
 
 ## Query-template catalogue
 
@@ -119,12 +120,6 @@ first pilot workload.
 
 ### Quantile
 
-For the 100 ms spatial class, PromQL has exact grouped semantics:
-
-```promql
-quantile by (G) (phi, data)
-```
-
 For a temporal grouped quantile, PromQL cannot directly express a pooled
 quantile over every raw sample in each group.  The PromQL-compatible rendering
 is instead the average of each leaf series' temporal quantile:
@@ -182,6 +177,13 @@ The manifest also contains the label hierarchy, cardinalities of every legal
 grouping, value/key distribution facts, feeder seed, and support class
 (`common-promql`, `counter-gated-promql`, or `sql-only`).
 
+The `data_facts` contract fixes a 100 ms scrape cadence and 20-minute
+backfill, a deterministic Zipf-like ranking of the 10,000 leaf series, a
+rank-scaled Pareto gauge and monotonic counter (no resets), plus a 10,000-ID user
+universe.  Each leaf rotates through 1,000 deterministically assigned IDs;
+the pools overlap across leaves.  The eventual feeder and every baseline
+adapter must implement these facts rather than choose their own distributions.
+
 Sampling is constrained rather than purely independent:
 
 - exactly ten atomic queries per workload and no binary queries;

asap-planner-rs/src/optimizer/aqe_extractor.rs

@@ -10,7 +10,7 @@ use promql_utilities::query_logics::enums::Statistic;
 
 use crate::config::input::TableDefinition;
 use crate::planner::patterns::build_patterns;
-use crate::planner::promql::{parse_binary_arms, BinaryArm};
+use crate::planner::promql::{parse_binary_arms, strip_parens, BinaryArm};
 use crate::planner::sql::{parse_sql_query, sql_data_range_ms, sql_statistics};
 
 use super::error::OptimizerError;
@@ -261,21 +261,39 @@ pub(crate) fn decompose_to_leaves(query: &str) -> Vec<String> {
     leaves
 }
 
-/// Whether any leaf of `query` is an `avg` that `decompose_to_leaves` rewrites.
+/// Whether `query` has an `avg` the engine can't answer: one that
+/// `decompose_to_leaves` splits into sum and count, or a per-series avg in a
+/// binary arm, which runs without its query-time avg.
 pub(crate) fn contains_avg(query: &str) -> bool {
     match parse_binary_arms(query) {
-        Some((lhs, rhs)) => [lhs, rhs]
-            .into_iter()
-            .any(|arm| matches!(arm, BinaryArm::Query(q) if contains_avg(&q))),
+        Some((lhs, rhs)) => [lhs, rhs].into_iter().any(|arm| {
+            matches!(arm, BinaryArm::Query(q) if contains_avg(&q) || per_series_avg(&q).is_some())
+        }),
         None => rewrite_avg(query).len() > 1,
     }
 }
 
+/// `avg by (l) (f(x[w]))` of a per-series function `f`: the leaf
+/// `f(x[w])`, and the avg that averages its series at query time.
+pub(crate) fn per_series_avg(query: &str) -> Option<(String, AggregateExpr)> {
+    match promql_parser::parser::parse(query).ok()? {
+        Expr::Aggregate(agg) if agg.op.id() == token::T_AVG => match strip_parens(&agg.expr) {
+            call @ Expr::Call(_) => Some((call.to_string(), agg)),
+            _ => None,
+        },
+        _ => None,
+    }
+}
+
 /// Rewrite an `avg` leaf into the sum and count leaves it is computed from:
 /// `avg by (l) (x)` → `sum by (l) (x)`, `count by (l) (x)`, and
 /// `avg_over_time(x[5m])` → `sum_over_time(x[5m])`, `count_over_time(x[5m])`.
-/// Any other query is returned unchanged.
+/// A per-series avg is its leaf (see `per_series_avg`). Any other query is
+/// returned unchanged.
 fn rewrite_avg(query: &str) -> Vec<String> {
+    if let Some((leaf, _)) = per_series_avg(query) {
+        return vec![leaf];
+    }
     match promql_parser::parser::parse(query) {
         Ok(Expr::Aggregate(agg)) if agg.op.id() == token::T_AVG => [token::T_SUM, token::T_COUNT]
             .into_iter()
@@ -608,6 +626,23 @@ mod tests {
         assert!(items.iter().all(|i| i.requirements.statistics.len() == 1));
     }
 
+    /// An avg of per-series quantiles averages the quantiles at query time;
+    /// sum and count items would ask for sums of quantiles.
+    #[test]
+    fn per_series_avg_is_its_per_series_leaf() {
+        let query = "avg by (job) (quantile_over_time(0.99, metric[5m]))";
+        assert_eq!(
+            decompose_to_leaves(query),
+            vec!["quantile_over_time(0.99, metric[5m])"]
+        );
+        let items = extract_aqes(&[rqe(query, 60_000)], &empty_schema(), 15_000).unwrap();
+        let [item] = items.as_slice() else {
+            panic!("{items:?}");
+        };
+        assert_eq!(item.requirements.statistics, vec![Statistic::Quantile]);
+        assert!(item.requirements.grouping_labels.labels.is_empty());
+    }
+
     #[test]
     fn spatial_only_query_sets_range_to_scrape_interval() {
         let rqes = vec![rqe("sum(metric)", 60_000)];

asap-planner-rs/src/optimizer/milp_output.rs

@@ -24,13 +24,16 @@ use crate::planner_output::PlannerOutput;
 use crate::promql::generator::{build_inference_yaml, build_streaming_yaml};
 use crate::sql::generator::{build_sql_inference_yaml, build_sql_streaming_yaml};
 
-use super::aqe_extractor::{contains_avg, decompose_to_leaves};
+use super::aqe_extractor::{contains_avg, decompose_to_leaves, per_series_avg};
 use super::milp::{MilpConfig, MilpWorkload};
 use super::solution::OptimizerItem;
 
 #[derive(Debug, Error)]
 pub enum MilpOutputError {
-    #[error("query {0:?} uses avg, which the engine can't answer from sum and count yet")]
+    #[error(
+        "query {0:?} uses an avg the engine can't answer yet: from sum and count, or in a \
+         binary arm"
+    )]
     AvgQuery(String),
     #[error(
         "query groups set step_ms or range_duration_ms, whose extra retention MILP \
@@ -227,6 +230,7 @@ pub fn deployments_to_planner_output(
             }
         }
     }
+    plan_per_series_avgs(config, &binary_leaves, &mut queries)?;
 
     let id_map: HashMap<String, u32> = aggregations
         .keys()
@@ -342,6 +346,50 @@ fn query_plan_entry(
     })
 }
 
+/// Plans each per-series avg query (`per_series_avg`) as its leaf's entry
+/// plus a query-time avg, replacing the leaf's entry unless the leaf is
+/// itself a query or a binary query's arm.
+fn plan_per_series_avgs(
+    config: MilpConfig,
+    binary_leaves: &HashSet<String>,
+    queries: &mut IndexMap<String, QueryPlanEntry>,
+) -> Result<(), MilpOutputError> {
+    let MilpConfig::Promql(config) = config else {
+        return Ok(());
+    };
+    let asked: Vec<&String> = config
+        .query_groups
+        .iter()
+        .flat_map(|group| &group.queries)
+        .collect();
+    let mut leaves = Vec::new();
+    for &query in &asked {
+        let Some((leaf, avg)) = per_series_avg(query) else {
+            continue;
+        };
+        let unregroupable = || MilpOutputError::UnregroupableQuery(query.clone());
+        let entry = queries.get(&leaf).ok_or_else(unregroupable)?;
+        // The leaf keeps every label, so its aggregation is never regrouped.
+        if !entry.query_time_aggregations.is_empty() {
+            return Err(unregroupable());
+        }
+        let stage = aggregation_stage(&avg).ok_or_else(unregroupable)?;
+        let planned = QueryPlanEntry {
+            aggregation_keys: entry.aggregation_keys.clone(),
+            planned_subquery: leaf.clone(),
+            query_time_aggregations: vec![stage],
+        };
+        queries.insert(query.clone(), planned);
+        leaves.push(leaf);
+    }
+    for leaf in leaves {
+        if !asked.contains(&&leaf) && !binary_leaves.contains(&leaf) {
+            queries.shift_remove(&leaf);
+        }
+    }
+    Ok(())
+}
+
 /// The leaves of the workload's binary queries. The engine runs a binary
 /// query's arms without query-time stages.
 fn binary_query_leaves(config: MilpConfig) -> HashSet<String> {
@@ -1091,6 +1139,45 @@ mod tests {
         assert!(matches!(err, MilpOutputError::AvgQuery(q) if q == query));
     }
 
+    /// An avg of a per-series function reads that function's aggregation
+    /// and averages its series at query time; the bare leaf isn't a query.
+    #[test]
+    fn per_series_avg_averages_its_leaf_at_query_time() {
+        let query = "avg by (job) (quantile_over_time(0.99, http_requests_total[5m]))";
+        let leaf = "quantile_over_time(0.99, http_requests_total[5m])";
+        let (config, workload, solution) = plan(&group(query, 0.99));
+        let output =
+            plan_to_planner_output(MilpConfig::Promql(&config), &workload, &solution).unwrap();
+
+        let a = aggregation_for(&output, query);
+        assert_eq!(a.aggregation_type, AggregationType::DatasketchesKLL);
+        let query_config = query_config_for(&output, query);
+        assert_eq!(query_config.planned_subquery, leaf);
+        let [stage] = query_config.query_time_aggregations.as_slice() else {
+            panic!("{:?}", query_config.query_time_aggregations);
+        };
+        assert_eq!(
+            serde_json::to_value(stage).unwrap(),
+            json!({"operator": "avg", "grouping": {"mode": "by", "labels": ["job"]},
+                   "parameter": null})
+        );
+        stage.validate().unwrap();
+        let inference = output.to_inference_config(QueryLanguage::promql).unwrap();
+        assert!(inference.query_configs.iter().all(|q| q.query != leaf));
+    }
+
+    /// A binary query's arms run without query-time stages, so a per-series
+    /// avg arm can't be answered.
+    #[test]
+    fn per_series_avg_in_a_binary_arm_is_an_error() {
+        let query = "avg by (job) (quantile_over_time(0.99, http_requests_total[5m])) * 2";
+        let (config, workload, solution) = plan(&group(query, 0.99));
+        let err = plan_to_planner_output(MilpConfig::Promql(&config), &workload, &solution)
+            .err()
+            .unwrap();
+        assert!(matches!(err, MilpOutputError::AvgQuery(q) if q == query));
+    }
+
     /// Range overrides only size retention, so they block writing configs
     /// but not planning.
     #[test]

asap-planner-rs/src/optimizer/saved_plan.rs

@@ -108,7 +108,7 @@ struct Chosen {
     /// Index of the plan's deployment.
     deployment: usize,
     /// Present only when the deployment's grouping differs from the RQE's
-    /// (a roll-up), which the converter refuses.
+    /// (a roll-up).
     deployment_grouping: Option<LabelSet>,
 }
 
@@ -583,6 +583,41 @@ mod tests {
         );
     }
 
+    /// An avg of per-series quantiles is one leaf, the per-series quantile
+    /// its one RQE names, averaged by `job` at query time.
+    #[test]
+    fn writes_a_per_series_avg_as_a_query_time_avg() {
+        let query = "avg by (job) (quantile_over_time(0.5, data[5m]))";
+        let table = TABLE.replace(
+            r#""quantile_over_time(0.5, data[5m])", "interval_secs": 60"#,
+            &format!(r#""{query}", "interval_secs": 60"#),
+        );
+        let inference = run_on(&table, plans(shared_p50()).to_string().as_bytes(), None)
+            .unwrap()
+            .to_inference_config(asap_types::enums::QueryLanguage::promql)
+            .unwrap();
+        let query_config = inference
+            .query_configs
+            .iter()
+            .find(|q| q.query == query)
+            .expect("a query_config for the avg");
+        assert_eq!(
+            query_config.planned_subquery,
+            "quantile_over_time(0.5, data[5m])"
+        );
+        let [stage] = query_config.query_time_aggregations.as_slice() else {
+            panic!("{:?}", query_config.query_time_aggregations);
+        };
+        assert_eq!(
+            serde_json::to_value(stage).unwrap(),
+            json!({
+                "operator": "avg",
+                "grouping": {"mode": "by", "labels": ["job"]},
+                "parameter": null,
+            })
+        );
+    }
+
     #[test]
     fn sla_ms_selects_the_plan_at_that_sla() {
         let yaml = run(plans(shared_p50()), Some(100.0))