eval: one RQE per schema query, the strictest template's

rqe-optimizer/README.md

@@ -33,8 +33,8 @@ Each workload is one file, `synthetic-templates<set>-shared<r>[-metrics<m>]-tp95
 | Name part | Meaning |
 |---|---|
 | `templatesclassic` | the mixed set: 8 PromQL templates on one metric (42 RQEs); templates 7 and 9 (`sum by (job) (rate(...))`, `topk(32, sum by (label_0) (rate(...)))`) are dropped, as ASAPQuery would need per-series counter deltas to serve them |
-| `templatesmultigroup` | the multi-grouping templates alone: distinct counts and p99s over several groupings of `http` and `flows`, including p99 `GROUP BY CUBE(region, service, endpoint)` (62 RQEs) |
-| `templatesall` | classic + multi-grouping (104 RQEs) |
+| `templatesmultigroup` | the multi-grouping templates alone: distinct counts and p99s over several groupings of `http` and `flows`, including p99 `GROUP BY CUBE(region, service, endpoint)` (62 RQEs, exported with `--keep-duplicate-queries`; 52 without) |
+| `templatesall` | classic + multi-grouping (104 RQEs, exported with `--keep-duplicate-queries`; 94 without) |
 | `shared8` | 8 users query the same streams with their own windows, quantiles and repeat intervals |
 | `metrics8`, `metrics16` | 8 or 16 independent copies of the workload, each on its own metric |
 | `tp95` | every RQE must meet 95% accuracy in its sketch's own metric |
@@ -146,7 +146,7 @@ Every table also has `workloads[0].e2e`, for running a plan end to end
 
 | Field | Meaning |
 |---|---|
-| `e2e.queries` | classic: one entry per PromQL query and repeat interval: `promql`, `interval_secs`, and the `rqes` it reads (template 10's ratio reads two). Multi-grouping (no PromQL): one entry per distinct query and interval: canonical ASAPQuery `sql` over `NOW()` (the planner floors and replaces it), plus `table`, `value`, `capability`, `grouping`, `quantile` (p99: 0.99), `window_secs`, `interval_secs`, and the `rqes` it reads (sorted; templates 14, 15 and 17 share some groupings) |
+| `e2e.queries` | classic: one entry per PromQL query and repeat interval: `promql`, `interval_secs`, and the `rqes` it reads (template 10's ratio reads two). Multi-grouping (no PromQL): one entry per distinct query and interval: canonical ASAPQuery `sql` over `NOW()` (the planner floors and replaces it), plus `table`, `value`, `capability`, `grouping`, `quantile` (p99: 0.99), `window_secs`, `interval_secs`, and the `rqes` it reads: only the strictest of templates 14, 15 and 17 where they share a grouping, or every one, sorted, with `--keep-duplicate-queries` |
 | `e2e.metrics` | the metrics to generate. Classic: gauge `data` (sums, quantiles, top-k by sum) and counter `data_total` (rates), with `labels` in the `schemas` format (`label_0` is `child_of: job`, 1,000 per job, so 1e4 series), samples/s per series, and each stream's value shape. Multi-grouping: `type: "records"` tables `http` and `flows` (`data_1_http` for metric `data_1/http` in `metrics=m` tables) with the schema's `labels`, `records_per_sec` and `values` (shape, `scaled_by`, `anomaly`, `burst`) |
 
 ### Accuracy and cost

rqe-optimizer/results/autosketch-vs-asap-hydra/README.md

@@ -200,8 +200,8 @@ zcat plans/synthetic-templatesmultigroup-shared1.plans.json.gz | jq '.plans[]
 - The accuracy curves (`out_grid_1e7_cost/`, `out_1e9/`) are the committed
   #138 study's, in `../autosketch-vs-asap-inputs/saturation/` (#194), with
   #213's new KLL and DDSketch configs appended.
-- `inputs/tables/`: the workloads, `export_autosketch_eval_table.py
-  --synthetic` (classic, multigroup and all × r ∈ {1, 8} × m ∈ {1, 8, 16}).
+- `inputs/tables/`: the workloads, `export_autosketch_eval_table.py --synthetic
+  --keep-duplicate-queries` (classic, multigroup and all × r ∈ {1, 8} × m ∈ {1, 8, 16}).
 
 ## Reproduce
 

rqe-optimizer/results/autosketch-vs-asap-inputs/README.md

@@ -21,7 +21,7 @@ and `../autosketch-vs-asap-synthetic-sla/`), as run on 2026-10-08.
   `scripts/merge_study_parts.py`, which refuses a point measured twice. Their
   raw records are `raw/dense_*`.
 - `tables/`: the synthetic mixed-set workloads from
-  `scripts/export_autosketch_eval_table.py --synthetic` (RQEs, streams,
+  `scripts/export_autosketch_eval_table.py --synthetic --keep-duplicate-queries` (RQEs, streams,
   targets, data shapes) and `plan.tsv` (one run per table).
 
 AutoSketch's measured benchmark times are next to each figure's results

rqe-optimizer/results/autosketch-vs-asap-synthetic-sla/README.md

@@ -74,7 +74,7 @@ approxbench accuracy benchmark, serially, alone on clnode109
 ```sh
 # The inputs these results used are committed: TABLES=../autosketch-vs-asap-inputs/tables,
 # DIR=../autosketch-vs-asap-inputs/saturation (paths from this directory).
-python3 scripts/export_autosketch_eval_table.py --synthetic --out TABLES
+python3 scripts/export_autosketch_eval_table.py --synthetic --keep-duplicate-queries --out TABLES
 # For each row of TABLES/plan.tsv:
 target/release/examples/autosketch_vs_asap synthetic --table TABLES/T --target p95 \
     --saturation-dir DIR --out OUT/R --runs 3

rqe-optimizer/results/autosketch-vs-asap-synthetic/README.md

@@ -165,7 +165,7 @@ python3 scripts/study_saturation.py --phase accuracy --n-max 1e9 --seeds 3 \
 python3 scripts/study_saturation.py --phase optimizer-cost --out DIR/optimizer_cost
 
 # Workload tables and plan.tsv, then the runs.
-python3 scripts/export_autosketch_eval_table.py --synthetic --out TABLES
+python3 scripts/export_autosketch_eval_table.py --synthetic --keep-duplicate-queries --out TABLES
 # Or use the committed inputs: TABLES=rqe-optimizer/results/autosketch-vs-asap-inputs/tables,
 # DIR=rqe-optimizer/results/autosketch-vs-asap-inputs/saturation.
 SATURATION_DIR=DIR scripts/run_autosketch_vs_asap_synthetic.sh TABLES run

rqe-optimizer/results/e2e-rate10/README.md

@@ -10,7 +10,7 @@ table, p95 targets, runner) is the same as `../autosketch-vs-asap-hydra/`.
 ## Files
 
 - `inputs/tables/`: the tables, `export_autosketch_eval_table.py --synthetic
-  --samples-per-sec 10` (classic, all, multigroup at shared = 1). Each one's
+  --samples-per-sec 10 --keep-duplicate-queries` (classic, all, multigroup at shared = 1). Each one's
   `workloads[0].e2e` holds the queries and metrics to generate.
 - `plans/<table>.plans.json.gz`: each method's chosen plans, in the format of
   `../autosketch-vs-asap-hydra/README.md`, "Plans" (`--runs 1`, kept, then
@@ -88,9 +88,9 @@ bound stays 490 ms (SLAs from 1,000 ms); classic's is 21 ms.
 ```sh
 cargo build --release -p rqe-optimizer --examples
 DIR=rqe-optimizer/results/e2e-rate10; TMP=$(mktemp -d)
-python3 scripts/export_autosketch_eval_table.py --synthetic --samples-per-sec 10 \
+python3 scripts/export_autosketch_eval_table.py --synthetic --samples-per-sec 10 --keep-duplicate-queries \
     --revision 44ac160dd350f695a09224fd124f0dcfe3dd825a --out $TMP/classic   # classic tables
-python3 scripts/export_autosketch_eval_table.py --synthetic --samples-per-sec 10 \
+python3 scripts/export_autosketch_eval_table.py --synthetic --samples-per-sec 10 --keep-duplicate-queries \
     --revision beacf68 --out $TMP/rest                                        # all, multigroup
 # copy synthetic-templates{classic,all,multigroup}-shared1.json to $DIR/inputs/tables/
 # SAT: the saturation dir of ../autosketch-vs-asap-hydra/README.md, "Reproduce"

scripts/export_autosketch_eval_table.py

@@ -297,6 +297,10 @@ def rqe_entry(q, curves, curves_big, sat, sat_big, merge):
 # Set by --samples-per-sec: a schema RQE whose smallest covered group falls
 # below SCHEMA_MIN_COVERED_N is left out (logged) instead of failing.
 SKIP_BELOW_CURVES = False
+# Set by --keep-duplicate-queries: every template's copy of a query is an
+# RQE. Otherwise only the strictest copy is, since the engine answers one
+# query string from one aggregation.
+KEEP_DUPLICATE_QUERIES = False
 
 
 def set_samples_per_sec(rate, skip_below_curves):
@@ -534,6 +538,28 @@ def add(query_id, stream, capability, rng, s, t, groups, keys=None, promql=None)
     return out
 
 
+def schema_query(name, value, capability, what, grouping):
+    """A schema template's query on `grouping`: the fields that make it
+    one query string."""
+    query = {"table": name, "value": value, "capability": capability, "grouping": grouping}
+    if capability == "quantile":  # p99_latency -> 0.99
+        query["quantile"] = int(what[1:what.index("_")]) / 100
+    return query
+
+
+def strictest_templates():
+    """{query: (number of templates asking it, the strictest one's number)}
+    over SCHEMA_TEMPLATES. Stricter covers more groups: every group (None),
+    then the smallest `covers_share`."""
+    copies = {}
+    for n, what, name, value, capability, groupings, covers in SCHEMA_TEMPLATES:
+        for grouping in groupings:
+            key = json.dumps(schema_query(name, value, capability, what, grouping),
+                             sort_keys=True)
+            copies.setdefault(key, []).append((covers is not None, covers or 0, n))
+    return {key: (len(c), min(c)[2]) for key, c in copies.items()}
+
+
 def schema_queries(interval, dataset, range_with_interval, seen, metric=None, e2e=None):
     """SCHEMA_TEMPLATES as synthetic_queries' query-ranges, over
     SCHEMA_WINDOWS, repeating every `interval` seconds. A metric's values are
@@ -545,13 +571,21 @@ def schema_queries(interval, dataset, range_with_interval, seen, metric=None, e2
     ASAPQuery SQL (there is no PromQL): {sql, table, value,
     capability, grouping, quantile (quantiles only), window_secs,
     interval_secs, rqes}. `table` is the metric with `/` as `_`
-    (`data_1_http`)."""
+    (`data_1_http`).
+
+    A query several templates ask (14, 15 and 17 share groupings) is only
+    its strictest template's RQE, unless KEEP_DUPLICATE_QUERIES."""
     out = []
+    strictest = strictest_templates()
     for n, what, name, value, capability, groupings, covers in SCHEMA_TEMPLATES:
         schema = SCHEMAS[name]
         shape = schema["values"][value]
         full = f"{metric}/{name}" if metric else name
         for grouping in groupings:
+            copies, keep = strictest[json.dumps(
+                schema_query(name, value, capability, what, grouping), sort_keys=True)]
+            if not KEEP_DUPLICATE_QUERIES and keep != n:
+                continue
             groups = grouping_cardinality(name, grouping)
             query_id = f"t{n}_{what}_by_{','.join(grouping)}"
             if metric:
@@ -571,16 +605,21 @@ def schema_queries(interval, dataset, range_with_interval, seen, metric=None, e2
                 # can't serve the RQE, so a lower --samples-per-sec leaves it
                 # (and its e2e query) out.
                 if SKIP_BELOW_CURVES and covered_min_n < SCHEMA_MIN_COVERED_N:
+                    # A weaker template's copy might fit, but serving the
+                    # query at a weaker target would change the workload.
+                    if not KEEP_DUPLICATE_QUERIES and copies > 1:
+                        raise ValueError(
+                            f"{dataset}/{query_id}/{kind}/{rng}, the strictest of {copies} "
+                            f"templates asking its query, has a covered group of "
+                            f"{covered_min_n:.0f} items, below the curves' first N")
                     print(f"left out {dataset}/{query_id}/{kind}/{rng}: smallest covered "
                           f"group holds {covered_min_n:.0f} items, below the curves' first N",
                           file=sys.stderr)
                     continue
                 assert covered_min_n >= SCHEMA_MIN_COVERED_N, (query_id, rng, covered_min_n)
                 if e2e is not None:
-                    query = {"table": full.replace("/", "_"), "value": value,
-                             "capability": capability, "grouping": grouping}
-                    if capability == "quantile":  # p99_latency -> 0.99
-                        query["quantile"] = int(what[1:what.index("_")]) / 100
+                    query = schema_query(full.replace("/", "_"), value, capability, what,
+                                         grouping)
                     query["sql"] = asapquery_sql(query, s)
                     e2e.append({**query, "window_secs": s, "interval_secs": interval,
                                 "rqes": [f"{dataset}/{query_id}/{kind}/{rng}"]})
@@ -833,22 +872,33 @@ def main(argv=None):
     parser.add_argument("--samples-per-sec", type=int,
                         help=f"--synthetic: samples/s per series (default "
                              f"{SYNTHETIC_SAMPLES_PER_SEC}); the schema metrics' rates scale "
-                             "with it, and schema RQEs below the curves' first N are left out")
+                             "with it, and schema RQEs below the curves' first N are left out; "
+                             "without --keep-duplicate-queries, the export fails instead if "
+                             "one is the strictest copy of a query several templates ask")
+    parser.add_argument("--keep-duplicate-queries", action="store_true",
+                        help="--synthetic: an RQE for every template asking a query, not "
+                             "just the strictest (the section 6.3 tables). Without it, a strictest "
+                             "copy below the curves' first N fails the export")
     parser.add_argument("--out", required=True)
     args = parser.parse_args(argv)
     # An instant query reads one scrape, 1000 / rate ms, which the runner keeps
     # in whole ms.
     rate = args.samples_per_sec
     if rate is not None and (rate <= 0 or 1000 % rate):
         parser.error("--samples-per-sec must divide 1000")
-    if args.samples_per_sec is not None:
-        default = SYNTHETIC_SAMPLES_PER_SEC
-        set_samples_per_sec(args.samples_per_sec, True)
-        try:
-            return run(args, parser)
-        finally:
-            set_samples_per_sec(default, False)
-    return run(args, parser)
+    global KEEP_DUPLICATE_QUERIES
+    KEEP_DUPLICATE_QUERIES = args.keep_duplicate_queries
+    try:
+        if args.samples_per_sec is not None:
+            default = SYNTHETIC_SAMPLES_PER_SEC
+            set_samples_per_sec(args.samples_per_sec, True)
+            try:
+                return run(args, parser)
+            finally:
+                set_samples_per_sec(default, False)
+        return run(args, parser)
+    finally:
+        KEEP_DUPLICATE_QUERIES = False
 
 
 def run(args, parser):

scripts/test_export_autosketch_eval_table.py

@@ -245,6 +245,7 @@ def test_schema_queries_read_every_rqe_with_their_fields(self):
                          "SELECT dst_subnet, COUNT(DISTINCT src_ip) FROM flows WHERE ts "
                          "BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY dst_subnet")
 
+    @mock.patch("export_autosketch_eval_table.KEEP_DUPLICATE_QUERIES", True)
     def test_schema_metrics_are_records_tables_named_by_metric(self):
         e2e = []
         qs = metric_queries({**SYNTHETIC_DEFAULT, "templates": "multigroup", "metrics": 2},
@@ -272,6 +273,38 @@ def test_schema_metrics_are_records_tables_named_by_metric(self):
                          [["t14_distinct_users_by_service_data_0",
                            "t15_distinct_users_by_service_data_0"]])
 
+    def test_each_query_is_only_its_strictest_templates_rqe(self):
+        # One query string can be served by one aggregation, so a query
+        # templates 14, 15 and 17 share keeps the template covering the most
+        # groups: 17 (every group), then 14 (1%), then 15 (5%).
+        e2e = []
+        synthetic_queries(templates="multigroup", dataset="d", e2e=e2e)
+        block = e2e_block([], e2e)
+        self.assertTrue(all(len(q["rqes"]) == 1 for q in block["queries"]))
+        users = {(tuple(q["grouping"]), q["window_secs"]): q["rqes"][0].split("/")[1]
+                 for q in block["queries"] if q["value"] == "user_id"}
+        for window in [300, 900]:
+            self.assertTrue(users[(("service", "endpoint"), window)].startswith("t17_"))
+            for grouping in [("region",), ("service",), ("region", "service")]:
+                self.assertTrue(users[(grouping, window)].startswith("t14_"))
+            self.assertTrue(users[(("endpoint",), window)].startswith("t15_"))
+
+    def test_a_strictest_copy_below_the_curves_fails_loudly(self):
+        # Every group of the full label set: the smallest has ~350 items in 5m
+        # (left out) and ~1060 in 15m.
+        grouping = [all_subsets("http")[-1]]
+        every = (17, "distinct_users", "http", "user_id", "cardinality", grouping, None)
+        some = (15, "distinct_users", "http", "user_id", "cardinality", grouping, 0.05)
+        with mock.patch("export_autosketch_eval_table.SKIP_BELOW_CURVES", True), \
+                mock.patch("sys.stderr"):
+            with mock.patch("export_autosketch_eval_table.SCHEMA_TEMPLATES", [every]):
+                self.assertEqual([q["range"] for q in synthetic_queries(templates="multigroup")],
+                                 ["15m"])
+            # A weaker copy would fit, but would weaken the workload's target.
+            with mock.patch("export_autosketch_eval_table.SCHEMA_TEMPLATES", [some, every]):
+                with self.assertRaisesRegex(ValueError, "strictest of 2 templates"):
+                    synthetic_queries(templates="multigroup")
+
 
 COMMITTED_TABLES = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..",
                                 "rqe-optimizer", "results", "autosketch-vs-asap-inputs",
@@ -317,8 +350,12 @@ def test_all_is_classic_plus_the_schema_templates(self):
         every = synthetic_queries(templates="all", dataset="d")
         self.assertEqual(every[:len(classic)], classic)
         new = every[len(classic):]
-        # Per window (5m, 15m): 1 + 3 + 4 + 15 + 7 + 1 groupings.
-        self.assertEqual(len(new), 2 * 31)
+        # Per window (5m, 15m): 1 + 3 + 4 + 15 + 7 + 1 groupings, of which 5
+        # repeat another template's query.
+        self.assertEqual(len(new), 2 * 26)
+        with mock.patch("export_autosketch_eval_table.KEEP_DUPLICATE_QUERIES", True):
+            self.assertEqual(len(synthetic_queries(templates="multigroup", dataset="d")),
+                             2 * 31)
         self.assertEqual({q["stream"] for q in new},
                          {"flows/src_ip", "http/user_id", "http/latency"})
         self.assertEqual({q["range"] for q in new}, {"5m", "15m"})
@@ -454,7 +491,8 @@ def test_samples_per_sec_scales_rates_and_leaves_out_rqes_below_the_curves(self)
                     for g in ["region,endpoint", "region,service,endpoint"]}
         ids = {r["id"].split("/", 3)[3] for r in workload["rqes"]}
         e2e_ids = {r.split("/", 3)[3] for q in workload["e2e"]["queries"] for r in q["rqes"]}
-        self.assertEqual(len(ids), 60)
+        # 62 RQEs, less 10 duplicate queries and the 2 left out.
+        self.assertEqual(len(ids), 50)
         self.assertFalse(left_out & (ids | e2e_ids))
         self.assertIn("t16_p99_latency_by_region,endpoint/values/15m", ids)
         self.assertEqual(workload["e2e"]["metrics"]["http"]["records_per_sec"], 100_000)