eval: e2e block for multi-grouping and all tables

rqe-optimizer/README.md

@@ -137,13 +137,13 @@ multi-grouping metric's labels, cardinalities and skew). An RQE:
 | `min_N`, `max_N` | items per group per window (mean group) |
 | `covers_share`, `covered_min_N`, `covered_max_N` | the coverage rule (only groups holding at least this share of records, plus the largest; `null`: every group) and the items of its smallest and largest covered group, where accuracy is read |
 
-Classic tables also have `workloads[0].e2e`, for running a plan end to end
+Every table also has `workloads[0].e2e`, for running a plan end to end
 (sketch-bench#204). The runner does not read it, so the plans are unchanged.
 
 | Field | Meaning |
 |---|---|
-| `e2e.queries` | one entry per PromQL query and repeat interval: `promql`, `interval_secs`, and the `rqes` it reads (template 10's ratio reads two) |
-| `e2e.metrics` | the metrics to generate: 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 |
+| `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.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
 

scripts/export_autosketch_eval_table.py

@@ -423,8 +423,9 @@ def synthetic_queries(windows=SYNTHETIC_WINDOWS, interval=SYNTHETIC_INTERVAL,
     schema_queries() to the `classic` set.
 
     With `e2e` (a list), the `classic` templates append their PromQL to it,
-    one {promql, interval_secs, rqes} per query (e2e_block()). Values are read
-    from the gauge `data`, increments from the counter `data_total`.
+    one {promql, interval_secs, rqes} per query (e2e_block()), and the schema
+    templates their query fields (schema_queries()). Values are read from the
+    gauge `data`, increments from the counter `data_total`.
     """
     rate = SYNTHETIC_SERIES * SYNTHETIC_SAMPLES_PER_SEC  # samples/s over all series
     dataset = dataset or "synthetic/" + point_id({"templates": templates, "shared": 1})
@@ -481,7 +482,7 @@ def add(query_id, stream, capability, rng, s, t, groups, keys=None, promql=None)
                     interval, C)
         return out
     if templates == "multigroup":
-        return schema_queries(interval, dataset, range_with_interval, seen, metric)
+        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})")
@@ -513,16 +514,22 @@ def add(query_id, stream, capability, rng, s, t, groups, keys=None, promql=None)
                                   f"quantile_over_time(0.5, {gauge}[{rng}])",
                         "interval_secs": interval, "rqes": rqes})
     if templates == "all":
-        out += schema_queries(interval, dataset, range_with_interval, seen, metric)
+        out += schema_queries(interval, dataset, range_with_interval, seen, metric, e2e)
     return out
 
 
-def schema_queries(interval, dataset, range_with_interval, seen, metric=None):
+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
     its streams (`http/user_id`); each RQE names its `grouping` and
     `covers_share`, and carries its metric's schema. With `metric`, the
-    schema metrics are named under it (`data_1/http`)."""
+    schema metrics are named under it (`data_1/http`).
+
+    With `e2e` (a list), each RQE appends its query's fields and canonical
+    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`)."""
     out = []
     for n, what, name, value, capability, groupings, covers in SCHEMA_TEMPLATES:
         schema = SCHEMAS[name]
@@ -539,15 +546,23 @@ def schema_queries(interval, dataset, range_with_interval, seen, metric=None):
                 if (query_id, rng) in seen:
                     continue
                 seen.add((query_id, rng))
+                kind = "values" if capability == "quantile" else "keys"
+                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["sql"] = asapquery_sql(query, s)
+                    e2e.append({**query, "window_secs": s, "interval_secs": interval,
+                                "rqes": [f"{dataset}/{query_id}/{kind}/{rng}"]})
                 n_items = schema["rate"] * s / groups
                 covered, min_share, max_share = group_coverage(name, grouping, covers)
                 covered_min_n = schema["rate"] * s * min_share
                 # Accuracy is read at the smallest covered group: keep it on
                 # the curves (their first N is 1e3).
                 assert covered_min_n >= SCHEMA_MIN_COVERED_N, (query_id, rng, covered_min_n)
                 out.append({
-                    "dataset": dataset, "query_id": query_id,
-                    "kind": "values" if capability == "quantile" else "keys",
+                    "dataset": dataset, "query_id": query_id, "kind": kind,
                     "range": rng, "range_s": s, "step_s": interval,
                     "theta": shape.get("zipf"), "K": shape.get("keys"),
                     "min_N": n_items, "max_N": n_items,
@@ -566,6 +581,22 @@ def schema_queries(interval, dataset, range_with_interval, seen, metric=None):
     return out
 
 
+def asapquery_sql(query, window_secs):
+    """Canonical ASAPQuery SQL for a structured schema query.
+
+    The end-to-end planner replaces NOW() with its floored evaluation time,
+    leaving this exporter as the single renderer for these queries.
+    """
+    group = ", ".join(query["grouping"])
+    if query["capability"] == "cardinality":
+        aggregate = f"COUNT(DISTINCT {query['value']})"
+    else:
+        assert query["capability"] == "quantile", query["capability"]
+        aggregate = f"QUANTILE({query['quantile']}, {query['value']})"
+    return (f"SELECT {group}, {aggregate} FROM {query['table']} WHERE ts BETWEEN "
+            f"DATEADD(s, -{window_secs}, NOW()) AND NOW() GROUP BY {group}")
+
+
 def shared_queries(point, e2e=None):
     """`point["shared"]` replicas of the point's template set on the same streams.
     Replica i draws SHARED_WINDOWS of the windows, SHARED_QUANTILES of the
@@ -639,27 +670,46 @@ def synthetic_entry(q):
 
 
 def e2e_block(qs, queries):
-    """What an end-to-end run of a `classic` table needs and the runner does
-    not read: the PromQL `queries` (one entry per distinct query and interval,
-    with the RQEs it reads; a binary query reads two) and the `metrics` to
-    generate. Each metric is a gauge (`value` and `dist` streams) and a
-    counter `<metric>_total` (`increment` streams), with labels in SCHEMAS'
-    format: `label_0` has SYNTHETIC_SERIES / SYNTHETIC_JOBS values under each
-    `job` (`child_of`), so there are SYNTHETIC_SERIES series."""
+    """What an end-to-end run of a table needs and the runner does not read:
+    the `queries` (one entry per distinct query and interval, with the RQEs
+    it reads; a binary query reads two) and the `metrics` to generate.
+    Classic queries are PromQL; each classic metric is a gauge (`value` and
+    `dist` streams) and a counter `<metric>_total` (`increment` streams), with
+    labels in SCHEMAS' format: `label_0` has SYNTHETIC_SERIES /
+    SYNTHETIC_JOBS values under each `job` (`child_of`), so there are
+    SYNTHETIC_SERIES series. Schema queries are schema_queries()' fields, one
+    per distinct query (with every RQE it reads, sorted); each schema metric
+    is a `records` table (`http`, or `data_1_http` under metric `data_1`)
+    with SCHEMAS' labels, `rate` as `records_per_sec`, and its `values`."""
     unique = {}
     for q in queries:
-        unique.setdefault((q["promql"], q["interval_secs"]), q)
+        if "promql" in q:
+            unique.setdefault((q["promql"], q["interval_secs"]), q)
+            continue
+        # Schema queries identical but for their RQEs (templates 14, 15, 17
+        # share groupings) are one query reading them all.
+        key = json.dumps({k: v for k, v in q.items() if k != "rqes"}, sort_keys=True)
+        if key in unique:
+            unique[key] = {**unique[key], "rqes": sorted(unique[key]["rqes"] + q["rqes"])}
+        else:
+            unique[key] = q
     labels = [{"name": "job", "cardinality": SYNTHETIC_JOBS},
               {"name": "label_0", "cardinality": SYNTHETIC_SERIES // SYNTHETIC_JOBS,
                "child_of": "job"}]
     keys = {"zipf": SYNTHETIC_THETA, "keys": SYNTHETIC_SERIES}
     metrics = {}
-    for gauge in dict.fromkeys(q["metric"] for q in qs):
+    for gauge in dict.fromkeys(q["metric"] for q in qs if "schema" not in q):
         common = {"labels": labels, "samples_per_sec_per_series": SYNTHETIC_SAMPLES_PER_SEC}
         metrics[gauge] = {"type": "gauge", **common,
                           "values": {"value": keys, "dist": {"pareto": SYNTHETIC_ALPHA}}}
         metrics[f"{gauge}_total"] = {"type": "counter", **common,
                                      "values": {"increment": keys}}
+    for q in qs:
+        if "schema" in q:
+            schema = q["schema"]
+            metrics[q["metric"].replace("/", "_")] = {
+                "type": "records", "labels": schema["labels"],
+                "records_per_sec": schema["rate"], "values": schema["values"]}
     return {"queries": list(unique.values()), "metrics": metrics}
 
 
@@ -696,9 +746,7 @@ def build_synthetic(args):
                 qs = synthetic_queries(templates=point["templates"],
                                        dataset="synthetic/" + point_id(point), e2e=queries)
             workload = {"dataset": qs[0]["dataset"], "rqes": [synthetic_entry(q) for q in qs]}
-            # Classic only: the multi-grouping templates have no PromQL yet (#204).
-            if point["templates"] == "classic":
-                workload["e2e"] = e2e_block(qs, queries)
+            workload["e2e"] = e2e_block(qs, queries)
             # Each schema metric's label schema, for the runner's facts.
             schemas = {q["metric"]: q["schema"] for q in qs if "schema" in q}
             if schemas:

scripts/test_export_autosketch_eval_table.py

@@ -18,7 +18,8 @@
 from study_saturation import COST_KEYS, COST_PARETO_ALPHA, COST_THETA  # noqa: E402
 from export_autosketch_eval_table import (  # noqa: E402
     SCHEMA_TEMPLATES, SCHEMAS, SYNTHETIC_DEFAULT, SYNTHETIC_GRID, SYNTHETIC_TEMPLATE_SETS,
-    all_subsets, boom_query, group_coverage, grouping_cardinality, label_shares, main, metric_queries, queries_per_instance,
+    all_subsets, boom_query, e2e_block, group_coverage, grouping_cardinality, label_shares, main,
+    metric_queries, queries_per_instance,
     shared_queries, synthetic_entry, synthetic_plan, synthetic_queries)
 from study_saturation import CURVE_COLUMNS, SUMMARY_COLUMNS  # noqa: E402
 
@@ -135,9 +136,12 @@ def test_synthetic_mode_writes_one_table_per_grid_point_and_a_plan(self):
         self.assertEqual((kll["sketch"], kll["grid_param"], kll["grid_K"]),
                          ("kll-percall", COST_PARETO_ALPHA, None))
         self.assertEqual(dd["sketch"], "dd")
-        # Only classic tables carry the e2e block (multi-grouping has no PromQL yet).
+        # Every table carries the e2e block; `all` has classic and schema metrics.
         with open(os.path.join(out, default)) as f:
-            self.assertNotIn("e2e", json.load(f)["workloads"][0])
+            self.assertEqual({m: v["type"] for m, v in
+                              json.load(f)["workloads"][0]["e2e"]["metrics"].items()},
+                             {"data": "gauge", "data_total": "counter",
+                              "http": "records", "flows": "records"})
         with open(os.path.join(out, "synthetic-templatesclassic-shared1.json")) as f:
             metrics = json.load(f)["workloads"][0]["e2e"]["metrics"]
         self.assertEqual({m: metrics[m]["type"] for m in metrics},
@@ -205,6 +209,59 @@ def test_classic_promql_reads_every_rqe_and_leaves_the_rqes_unchanged(self):
         self.assertEqual(by_text["rate(data_total[15m])"]["interval_secs"], 60)
         self.assertEqual(by_text["sum by (job) (data)"]["interval_secs"], 1)
 
+    def test_schema_queries_read_every_rqe_with_their_fields(self):
+        e2e = []
+        qs = synthetic_queries(templates="multigroup", dataset="d", e2e=e2e)
+        self.assertEqual(qs, synthetic_queries(templates="multigroup", dataset="d"))
+        self.assertEqual([r for query in e2e for r in query["rqes"]],
+                         [synthetic_entry(q)["id"] for q in qs])
+        self.assertTrue(all("promql" not in query for query in e2e))
+        # Template 16, p99 latency GROUP BY CUBE(region, service, endpoint):
+        # 7 groupings per window.
+        cube = [query for query in e2e if query["rqes"][0].startswith("d/t16_")]
+        self.assertEqual(len(cube), 2 * 7)
+        self.assertEqual(sorted(tuple(q["grouping"]) for q in cube if q["window_secs"] == 300),
+                         sorted(tuple(g) for g in all_subsets(
+                             "http", ["region", "service", "endpoint"])))
+        self.assertEqual(cube[0], {
+            "table": "http", "value": "latency", "capability": "quantile",
+            "grouping": ["region"], "quantile": 0.99, "window_secs": 300,
+            "interval_secs": 60, "rqes": ["d/t16_p99_latency_by_region/values/5m"],
+            "sql": "SELECT region, QUANTILE(0.99, latency) FROM http WHERE ts BETWEEN "
+                   "DATEADD(s, -300, NOW()) AND NOW() GROUP BY region"})
+        distinct = next(q for q in e2e if q["capability"] == "cardinality")
+        self.assertNotIn("quantile", distinct)
+        self.assertEqual(distinct["sql"],
+                         "SELECT dst_subnet, COUNT(DISTINCT src_ip) FROM flows WHERE ts "
+                         "BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY dst_subnet")
+
+    def test_schema_metrics_are_records_tables_named_by_metric(self):
+        e2e = []
+        qs = metric_queries({**SYNTHETIC_DEFAULT, "templates": "multigroup", "metrics": 2},
+                            e2e)
+        block = e2e_block(qs, e2e)
+        self.assertEqual(sorted(block["metrics"]),
+                         ["data_0_flows", "data_0_http", "data_1_flows", "data_1_http"])
+        http = block["metrics"]["data_1_http"]
+        self.assertEqual(http, {"type": "records", "labels": SCHEMAS["http"]["labels"],
+                                "records_per_sec": SCHEMAS["http"]["rate"],
+                                "values": SCHEMAS["http"]["values"]})
+        self.assertEqual({q["table"] for q in block["queries"]}, set(block["metrics"]))
+        # Each distinct query appears once and reads every RQE asking it:
+        # templates 14, 15 and 17 share some groupings.
+        read = [r for q in block["queries"] for r in q["rqes"]]
+        self.assertEqual(sorted(read), sorted(synthetic_entry(q)["id"] for q in qs))
+        fields = [json.dumps({k: v for k, v in q.items() if k != "rqes"}, sort_keys=True)
+                  for q in block["queries"]]
+        self.assertEqual(len(fields), len(set(fields)))
+        self.assertTrue(all(q["rqes"] == sorted(q["rqes"]) for q in block["queries"]))
+        by_service = [q for q in block["queries"] if q["table"] == "data_0_http"
+                      and q["capability"] == "cardinality" and q["grouping"] == ["service"]
+                      and q["window_secs"] == 300]
+        self.assertEqual([[r.split("/")[-3] for r in q["rqes"]] for q in by_service],
+                         [["t14_distinct_users_by_service_data_0",
+                           "t15_distinct_users_by_service_data_0"]])
+
 
 COMMITTED_TABLES = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..",
                                 "rqe-optimizer", "results", "autosketch-vs-asap-inputs",