eval: RQE expression trees in e2e tables

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

@@ -19,7 +19,9 @@ differences for the end-to-end run:
 
 - `inputs/tables/`: the tables, `export_autosketch_eval_table.py --synthetic
   --samples-per-sec 10` (classic, all, multigroup at shared = 1). Each one's
-  `workloads[0].e2e` holds the queries and metrics to generate.
+  `workloads[0].e2e` holds the queries and metrics to generate. Each query
+  but `rate` carries `expr`, its RQE expression tree (leaves name the plans'
+  RQE ids), added after planning: the RQEs and plans are unchanged.
 - `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
   `scripts/extract_autosketch_plans.py`).

scripts/export_autosketch_eval_table.py

@@ -222,6 +222,10 @@ def rqe_entry(q, curves, curves_big, sat, sat_big, merge):
 EXACT_SKETCH = {"exact-sum": "exact-sum", "exact-increase": "exact-increase"}
 WINDOW_SECS = {"1m": 60, "5m": 300, "10m": 600, "15m": 900, "1h": 3600, "6h": 21600,
                "24h": 86400}
+SERIES_LABELS = ["job", "label_0"]  # a classic metric's labels
+# Capabilities an expression-tree leaf can read (no rate); a topk leaf is
+# named topk_by_value.
+LEAF_CAPABILITIES = {"sum", "quantile", "cardinality", "topk"}
 DASHBOARD_QUANTILES = ["0.5", "0.9", "0.99"]
 DASHBOARD_FIXED_WINDOWS = ["15m", "1h"]  # D5
 # Shared replicas: each replica draws windows, quantiles and an interval.
@@ -425,6 +429,20 @@ def point_id(point):
     return base if metrics == 1 else f"{base}/metrics={metrics}"
 
 
+def rqe_leaf(rqe, metric, capability, lookback_s, grouping, phi=None, k=None, value=None):
+    """An expression-tree leaf reading RQE `rqe`, whose output is grouped by
+    `grouping`; None for a capability no leaf reads (rate)."""
+    if capability not in LEAF_CAPABILITIES:
+        return None
+    if capability == "topk":
+        capability = "topk_by_value"
+    leaf = {"id": rqe, "metric": metric, "capability": capability,
+            "lookback_ms": round(lookback_s * 1000), "grouping": list(grouping)}
+    leaf.update((key, v) for key, v in [("phi", phi), ("k", k), ("value", value)]
+                if v is not None)
+    return {"rqe": leaf}
+
+
 def synthetic_queries(windows=SYNTHETIC_WINDOWS, interval=SYNTHETIC_INTERVAL,
                       templates="dashboard", quantiles=None, dataset=None,
                       range_with_interval=False, seen=None, metric=None, e2e=None):
@@ -445,9 +463,10 @@ 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()), and the schema
-    templates their query fields (schema_queries()). Values are read from the
-    gauge `data`, increments from the counter `data_total`.
+    one {promql, interval_secs, rqes, expr} per query (e2e_block()), and the
+    schema templates their query fields (schema_queries()). `expr` is the
+    query's RQE expression tree (rqe_leaf()); a rate query has none. 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
     scrape = 1 / SYNTHETIC_SAMPLES_PER_SEC  # an instant query's lookback, in s
@@ -459,14 +478,18 @@ def synthetic_queries(windows=SYNTHETIC_WINDOWS, interval=SYNTHETIC_INTERVAL,
     def label(rng):
         return f"{rng}@{interval}s" if range_with_interval else rng
 
-    def add(query_id, stream, capability, rng, s, t, groups, keys=None, promql=None):
-        """Adds the RQE (once) and its query `promql`; returns the RQE's id."""
+    def add(query_id, stream, capability, rng, s, t, groups, keys=None, promql=None,
+            grouping=SERIES_LABELS, **leaf):
+        """Adds the RQE (once) and its query `promql`, grouped by `grouping`,
+        with `leaf`'s rqe_leaf() fields; returns the RQE's id."""
         if metric:
             query_id, stream = f"{query_id}_{metric}", f"{metric}/{stream}"
         kind = "values" if capability == "quantile" else "keys"
         rqe = f"{dataset}/{query_id}/{kind}/{rng}"
         if promql and e2e is not None:
-            e2e.append({"promql": promql, "interval_secs": t, "rqes": [rqe]})
+            query = {"promql": promql, "interval_secs": t, "rqes": [rqe]}
+            expr = rqe_leaf(rqe, gauge, capability, s, grouping, **leaf)
+            e2e.append({**query, "expr": expr} if expr else query)
         if (query_id, rng) in seen:
             return rqe
         seen.add((query_id, rng))
@@ -505,34 +528,38 @@ 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", scrape, 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})",
+        grouping=["job"])
     for q in quantiles:
         add(f"quantile_by_job_p{q}", "job/dist", "quantile", "1s", scrape, 1, J,
-            promql=f"quantile by (job) ({q}, {gauge})")
+            promql=f"quantile by (job) ({q}, {gauge})", grouping=["job"], phi=float(q))
     for rng, s in ranges:
         lbl = label(rng)
         # Template 2's PromQL omits `sum by (label_0)`: label_0 names one
         # series (child_of job, unique values), so it only drops `job`, and
         # the ranking and values are the same. ASAPQuery's planner supports
         # this two-level shape, not the three-level one (#744, #772).
         add("topk32_sum_by_label0_sum_over_time", "label_0/value", "topk", lbl, s, interval,
-            1, C, promql=f"topk(32, sum_over_time({gauge}[{rng}]))")
+            1, C, promql=f"topk(32, sum_over_time({gauge}[{rng}]))", k=32)
         add("sum_over_time", "series/value", "sum", lbl, s, interval, C,
             promql=f"sum_over_time({gauge}[{rng}])")
         for q in quantiles:
             add(f"quantile_over_time_p{q}", "series/dist", "quantile", lbl, s, interval, C,
-                promql=f"quantile_over_time({q}, {gauge}[{rng}])")
+                promql=f"quantile_over_time({q}, {gauge}[{rng}])", phi=float(q))
         add("rate", "series/increment", "rate", lbl, s, interval, C,
             promql=f"rate({counter}[{rng}])")
         add("sum_by_job_sum_over_time", "job/value", "sum", lbl, s, interval, J,
-            promql=f"sum by (job) (sum_over_time({gauge}[{rng}]))")
+            promql=f"sum by (job) (sum_over_time({gauge}[{rng}]))", grouping=["job"])
         # Template 10, the same RQEs as template 5's: one query reads both.
         rqes = [add(f"quantile_over_time_p{q}", "series/dist", "quantile", lbl, s, interval, C)
                 for q in ["0.9", "0.5"]]
         if e2e is not None:
+            lhs, rhs = (rqe_leaf(rqe, gauge, "quantile", s, SERIES_LABELS, phi=phi)
+                        for rqe, phi in zip(rqes, [0.9, 0.5]))
             e2e.append({"promql": f"quantile_over_time(0.9, {gauge}[{rng}]) / "
                                   f"quantile_over_time(0.5, {gauge}[{rng}])",
-                        "interval_secs": interval, "rqes": rqes})
+                        "interval_secs": interval, "rqes": rqes,
+                        "expr": {"op": "/", "lhs": lhs, "rhs": rhs}})
     if templates == "all":
         out += schema_queries(interval, dataset, range_with_interval, seen, metric, e2e)
     return out
@@ -570,8 +597,8 @@ def schema_queries(interval, dataset, range_with_interval, seen, metric=None, e2
     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`).
+    interval_secs, rqes, expr}. `table` is the metric with `/` as `_`
+    (`data_1_http`); `expr` is one rqe_leaf() on it.
 
     A query several templates ask (14, 15 and 17 share groupings) is only
     its strictest template's RQE, unless KEEP_DUPLICATE_QUERIES."""
@@ -621,8 +648,11 @@ def schema_queries(interval, dataset, range_with_interval, seen, metric=None, e2
                     query = schema_query(full.replace("/", "_"), value, capability, what,
                                          grouping)
                     query["sql"] = asapquery_sql(query, s)
+                    rqe = f"{dataset}/{query_id}/{kind}/{rng}"
                     e2e.append({**query, "window_secs": s, "interval_secs": interval,
-                                "rqes": [f"{dataset}/{query_id}/{kind}/{rng}"]})
+                                "rqes": [rqe],
+                                "expr": rqe_leaf(rqe, query["table"], capability, s, grouping,
+                                                 phi=query.get("quantile"), value=value)})
                 out.append({
                     "dataset": dataset, "query_id": query_id, "kind": kind,
                     "range": rng, "range_s": s, "step_s": interval,
@@ -741,7 +771,8 @@ def e2e_block(qs, queries):
     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
+    per distinct query (with every RQE it reads, sorted; one reading several
+    has no `expr`, as no single leaf names them); 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 = {}
@@ -751,9 +782,11 @@ def e2e_block(qs, queries):
             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)
+        key = json.dumps({k: v for k, v in q.items() if k not in ("rqes", "expr")},
+                         sort_keys=True)
         if key in unique:
-            unique[key] = {**unique[key], "rqes": sorted(unique[key]["rqes"] + q["rqes"])}
+            merged = {k: v for k, v in unique[key].items() if k != "expr"}
+            unique[key] = {**merged, "rqes": sorted(unique[key]["rqes"] + q["rqes"])}
         else:
             unique[key] = q
     labels = [{"name": "job", "cardinality": SYNTHETIC_JOBS},

scripts/test_export_autosketch_eval_table.py

@@ -238,13 +238,74 @@ def test_schema_queries_read_every_rqe_with_their_fields(self):
             "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"})
+                   "DATEADD(s, -300, NOW()) AND NOW() GROUP BY region",
+            "expr": {"rqe": {"id": "d/t16_p99_latency_by_region/values/5m", "metric": "http",
+                             "capability": "quantile", "lookback_ms": 300_000,
+                             "grouping": ["region"], "phi": 0.99, "value": "latency"}}})
         distinct = next(q for q in e2e if q["capability"] == "cardinality")
         self.assertNotIn("quantile", distinct)
+        self.assertEqual(distinct["expr"], {"rqe": {
+            "id": distinct["rqes"][0], "metric": "flows", "capability": "cardinality",
+            "lookback_ms": 300_000, "grouping": ["dst_subnet"], "value": "src_ip"}})
         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_classic_queries_carry_their_rqe_expression_trees(self):
+        e2e = []
+        synthetic_queries(templates="classic", dataset="d", e2e=e2e)
+        by_text = {query["promql"]: query for query in e2e}
+
+        def leaf(query_id, capability, lookback_ms, grouping, **extra):
+            return {"rqe": {"id": f"d/{query_id}", "metric": "data", "capability": capability,
+                            "lookback_ms": lookback_ms, "grouping": grouping, **extra}}
+
+        series = ["job", "label_0"]
+        # Instant RQEs read one scrape: 5 ms at 200 samples/s.
+        self.assertEqual(by_text["sum by (job) (data)"]["expr"],
+                         leaf("sum_by_job/keys/1s", "sum", 5, ["job"]))
+        self.assertEqual(by_text["quantile by (job) (0.75, data)"]["expr"],
+                         leaf("quantile_by_job_p0.75/values/1s", "quantile", 5, ["job"],
+                              phi=0.75))
+        self.assertEqual(by_text["topk(32, sum_over_time(data[15m]))"]["expr"],
+                         leaf("topk32_sum_by_label0_sum_over_time/keys/15m", "topk_by_value",
+                              900_000, series, k=32))
+        self.assertEqual(by_text["sum_over_time(data[1h])"]["expr"],
+                         leaf("sum_over_time/keys/1h", "sum", 3_600_000, series))
+        self.assertEqual(by_text["quantile_over_time(0.99, data[6h])"]["expr"],
+                         leaf("quantile_over_time_p0.99/values/6h", "quantile", 21_600_000,
+                              series, phi=0.99))
+        self.assertEqual(by_text["sum by (job) (sum_over_time(data[24h]))"]["expr"],
+                         leaf("sum_by_job_sum_over_time/keys/24h", "sum", 86_400_000, ["job"]))
+        ratio = by_text["quantile_over_time(0.9, data[1h]) / quantile_over_time(0.5, data[1h])"]
+        self.assertEqual(ratio["expr"], {
+            "op": "/",
+            "lhs": leaf("quantile_over_time_p0.9/values/1h", "quantile", 3_600_000, series,
+                        phi=0.9),
+            "rhs": leaf("quantile_over_time_p0.5/values/1h", "quantile", 3_600_000, series,
+                        phi=0.5)})
+        # A rate leaf has no capability on the RQE route.
+        self.assertNotIn("expr", by_text["rate(data_total[15m])"])
+        self.assertEqual([q["promql"] for q in e2e if "expr" not in q],
+                         [f"rate(data_total[{w}])" for w in ["15m", "1h", "6h", "24h"]])
+
+        def leaves(node):
+            return [node["rqe"]] if "rqe" in node else leaves(node["lhs"]) + leaves(node["rhs"])
+
+        # Every leaf names the query's RQEs, in order.
+        for query in e2e:
+            if "expr" in query:
+                self.assertEqual([x["id"] for x in leaves(query["expr"])], query["rqes"])
+
+    def test_metric_copies_name_their_metric_in_each_leaf(self):
+        e2e = []
+        metric_queries({**SYNTHETIC_DEFAULT, "templates": "all", "metrics": 2}, e2e)
+        leaf = next(q for q in e2e if q.get("promql") == "sum_over_time(data_1[1h])")["expr"]
+        self.assertEqual(leaf["rqe"]["metric"], "data_1")
+        self.assertTrue(leaf["rqe"]["id"].endswith("/sum_over_time_data_1/keys/1h"))
+        sql = next(q for q in e2e if q.get("table") == "data_1_http")["expr"]
+        self.assertEqual(sql["rqe"]["metric"], "data_1_http")
+
     @mock.patch("export_autosketch_eval_table.KEEP_DUPLICATE_QUERIES", True)
     def test_schema_metrics_are_records_tables_named_by_metric(self):
         e2e = []
@@ -262,9 +323,12 @@ def test_schema_metrics_are_records_tables_named_by_metric(self):
         # 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)
+        fields = [json.dumps({k: v for k, v in q.items() if k not in ("rqes", "expr")},
+                             sort_keys=True)
                   for q in block["queries"]]
         self.assertEqual(len(fields), len(set(fields)))
+        # A query reading several RQEs has no single leaf.
+        self.assertTrue(all(("expr" in q) == (len(q["rqes"]) == 1) for q in block["queries"]))
         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"]
@@ -281,6 +345,7 @@ def test_each_query_is_only_its_strictest_templates_rqe(self):
         synthetic_queries(templates="multigroup", dataset="d", e2e=e2e)
         block = e2e_block([], e2e)
         self.assertTrue(all(len(q["rqes"]) == 1 for q in block["queries"]))
+        self.assertTrue(all(q["expr"]["rqe"]["id"] == q["rqes"][0] 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]: