eval: instant RQEs read one scrape, not 1 s

rqe-optimizer/examples/autosketch_vs_asap.rs

@@ -98,9 +98,19 @@ const TRACES_TABLE: &str = "rqe-optimizer/data/autosketch-eval/table.json";
 /// Objective weights of #777 §4: CPU only, then AWS Fargate (us-east-1,
 /// Linux/x86) $/vCPU-hour and $/GB-hour, so the objective reads in $/hour.
 const WEIGHTS: [(&str, f64, f64); 2] = [("cpu", 1.0, 0.0), ("fargate", 0.0405, 0.00445)];
-/// Every stream is scraped once a second: windows and slides are whole
-/// seconds, and the stream's samples/s become its series count.
+/// A stream's scrape interval when its table gives no per-series rate: windows
+/// and slides are whole seconds, and the stream's samples/s become its series
+/// count.
 const SCRAPE_MS: Millis = 1_000;
+/// `ms` in seconds: an integer when whole, so second-grained plans read as
+/// before, else fractional (an instant RQE's one-scrape window).
+fn json_secs(ms: Millis) -> Value {
+    if ms.is_multiple_of(1000) {
+        json!(ms / 1000)
+    } else {
+        json!(ms as f64 / 1000.0)
+    }
+}
 const BYTES_PER_GIB: f64 = (1u64 << 30) as f64;
 const SEED: u64 = 7;
 /// AutoSketch's benchmark time per evaluated configuration, at the paper's
@@ -387,6 +397,12 @@ fn from_table(
         };
         let capability = capability_of(r);
         let rate = r["label_set"]["arrival_rate_per_sec"].as_f64().unwrap();
+        // A metric with a per-series rate (the synthetic gauge and counter) is
+        // scraped at it, so an instant RQE's one-scrape lookback holds one
+        // sample per series. Record metrics (http, flows) keep SCRAPE_MS.
+        let scrape_ms = workload["e2e"]["metrics"][metric]["samples_per_sec_per_series"]
+            .as_u64()
+            .map_or(SCRAPE_MS, |per_sec| 1_000 / per_sec);
         let groups = groups_of(r, capability);
         // A schema metric's RQE names its grouping (its `groups` are that
         // grouping's); otherwise the stream has one label `g`, the grouping.
@@ -406,11 +422,10 @@ fn from_table(
             max_covered_share.insert(id.clone(), r["max_covered_share"].as_f64().unwrap());
         }
         // One metric per stream, labels {g, x} (or the schema's and x), and
-        // with one scrape a second the full label set has one series per
-        // sample/s. A stream's RQEs differ only in range and grouping, so
+        // the full label set has one series per sample per scrape interval. A stream's RQEs differ only in range and grouping, so
         // take the largest per-RQE rate (conservative). Rates below `groups`
         // samples/s are rounded up so every group has a series.
-        let series = (rate * SCRAPE_MS as f64 / 1000.0).round() as u64;
+        let series = (rate * scrape_ms as f64 / 1000.0).round() as u64;
         let most = stream_series.entry(stream.clone()).or_default();
         *most = (*most).max(series);
         let most = *most;
@@ -436,7 +451,7 @@ fn from_table(
             MetricFacts {
                 cardinality: [(labels.clone(), 0), (LabelSet::new(), 1)].into(),
                 labels,
-                scrape_interval_ms: SCRAPE_MS,
+                scrape_interval_ms: scrape_ms,
                 value_range: None,
                 data_shape: BTreeMap::new(),
                 // A stream is one value of its metric, so one capability.
@@ -741,8 +756,8 @@ fn summarize(
                 "rqe": r.id,
                 "sketch": d.config.sketch,
                 "params": d.config.sketch_config["params"],
-                "window_secs": d.window_ms / 1000,
-                "slide_secs": d.slide_ms / 1000,
+                "window_secs": json_secs(d.window_ms),
+                "slide_secs": json_secs(d.slide_ms),
                 "merged_shards": r.lookback_ms / d.window_ms,
                 "ingest_workers": rqe_optimizer::analytical_cost_model::ingest_workers(d, &w.facts),
                 "latency_ms": latency,

scripts/export_autosketch_eval_table.py

@@ -429,8 +429,9 @@ def synthetic_queries(windows=SYNTHETIC_WINDOWS, interval=SYNTHETIC_INTERVAL,
     Streams: `series/*` hold one instance per series (C groups), `job/*` one per
     job (J groups), and `label_0/*` one top-k sketch over all C series as keys.
     `*/dist` streams feed quantile sketches, `*/value` sums, `*/increment`
-    increases. Spatial templates repeat every 1 s and read the last second
-    (S = T = 1 s); temporal ones repeat every `interval` seconds over each
+    increases. Spatial templates repeat every 1 s and, like a PromQL instant
+    query, read one sample per series: one scrape interval (S = 1 /
+    SYNTHETIC_SAMPLES_PER_SEC s, T = 1 s); temporal ones repeat every `interval` seconds over each
     lookback in `windows`. RQEs that repeat another's query and range (template
     10, D5) are kept once. Templates 7, `sum by (job) (rate(data_total[w]))`,
     and 9, `topk(32, sum by (label_0) (rate(data_total[w])))`, are dropped
@@ -445,6 +446,7 @@ def synthetic_queries(windows=SYNTHETIC_WINDOWS, interval=SYNTHETIC_INTERVAL,
     gauge `data`, increments from the counter `data_total`.
     """
     rate = SYNTHETIC_SERIES * SYNTHETIC_SAMPLES_PER_SEC  # samples/s over all series
+    scrape = 1 / SYNTHETIC_SAMPLES_PER_SEC  # an instant query's lookback, in s
     dataset = dataset or "synthetic/" + point_id({"templates": templates, "shared": 1})
     ranges = [(w, WINDOW_SECS[w]) for w in windows.split(",")]
     seen = set() if seen is None else seen
@@ -487,7 +489,7 @@ def add(query_id, stream, capability, rng, s, t, groups, keys=None, promql=None)
                 add(f"quantile_over_time_p{q}", "series/dist", "quantile", label(rng), s,
                     interval, C)
         for q in quantiles:
-            add(f"quantile_by_job_p{q}", "job/dist", "quantile", "1s", 1, 1, J)
+            add(f"quantile_by_job_p{q}", "job/dist", "quantile", "1s", scrape, 1, J)
         for rng, s in [(w, WINDOW_SECS[w]) for w in DASHBOARD_FIXED_WINDOWS]:
             if (rng, s) not in ranges:
                 continue
@@ -499,9 +501,9 @@ def add(query_id, stream, capability, rng, s, t, groups, keys=None, promql=None)
         return schema_queries(interval, dataset, range_with_interval, seen, metric, e2e)
     assert templates in ("classic", "all"), templates
     quantiles = quantiles or SYNTHETIC_QUANTILES
-    add("sum_by_job", "job/value", "sum", "1s", 1, 1, J, promql=f"sum by (job) ({gauge})")
+    add("sum_by_job", "job/value", "sum", "1s", scrape, 1, J, promql=f"sum by (job) ({gauge})")
     for q in quantiles:
-        add(f"quantile_by_job_p{q}", "job/dist", "quantile", "1s", 1, 1, J,
+        add(f"quantile_by_job_p{q}", "job/dist", "quantile", "1s", scrape, 1, J,
             promql=f"quantile by (job) ({q}, {gauge})")
     for rng, s in ranges:
         lbl = label(rng)

scripts/test_export_autosketch_eval_table.py

@@ -184,8 +184,10 @@ def test_classic_templates_give_42_distinct_rqes(self):
 
     def test_items_per_instance_follow_the_data_model(self):
         qs = {(q["query_id"], q["range"]): q for q in synthetic_queries(templates="classic")}
-        # 1e4 series x 200 samples/s = 2e6 samples/s.
-        self.assertEqual(qs[("quantile_by_job_p0.5", "1s")]["max_N"], 2e5)
+        # 1e4 series x 200 samples/s = 2e6 samples/s; an instant query reads
+        # one scrape, one sample from each of a job's 1e3 series.
+        q = qs[("quantile_by_job_p0.5", "1s")]
+        self.assertEqual((q["range_s"], q["max_N"]), (0.005, 1e3))
         self.assertEqual(qs[("quantile_over_time_p0.5", "15m")]["max_N"], 200 * 900)
         top = qs[("topk32_sum_by_label0_sum_over_time", "24h")]
         self.assertEqual((top["K"], top["groups"], top["max_N"]), (10_000, 1, 2e6 * 86400))
@@ -284,8 +286,8 @@ class SchemaTemplatesTest(unittest.TestCase):
     def test_classic_tables_are_main_s_all_tables_byte_for_byte(self):
         # The committed tables are the 10-template set's, from before "all"
         # grew: renamed, today's "classic" tables match them byte for byte
-        # without templates 7 and 9's RQEs, apart from the `e2e` block, which
-        # the runner does not read.
+        # without templates 7 and 9's RQEs and with instant RQEs reading one
+        # scrape, apart from the `e2e` block, which the runner does not read.
         with tempfile.TemporaryDirectory() as out:
             main(["--synthetic", "--revision", "test", "--out", out])
             classic = [n for n in sorted(os.listdir(out)) if "templatesclassic" in n]
@@ -298,6 +300,9 @@ def test_classic_tables_are_main_s_all_tables_byte_for_byte(self):
                     committed = json.load(f)
                 rqes = committed["workloads"][0]["rqes"]
                 rqes[:] = [r for r in rqes if not r["query_id"].startswith(DROPPED)]
+                for r in rqes:  # instant RQEs now read one scrape (5 ms)
+                    if r["range"] == "1s":
+                        r.update(lookback_secs=0.005, min_N=1e3, max_N=1e3)
                 del ours["sketch_bench_revision"], committed["sketch_bench_revision"]
                 del ours["workloads"][0]["e2e"]
                 self.assertEqual(json.dumps(ours, sort_keys=True),