feat(asap-tools): take ASAPQuery SQL from the table's structured entries

sketch-bench's exporter now writes each structured e2e entry's ASAPQuery
SQL (`sql`, e.g. `... WHERE ts BETWEEN DATEADD(s, -300, NOW()) AND NOW()
...`), so it is the single ASAPQuery SQL renderer. asapquery.sql now takes
that SQL as written and only replaces NOW() with the floored eval time; this
module's own dialect renderer is gone. The ClickHouse raw SQL still comes
from the structured fields.

The frontend's DATEADD takes only NOW(), a literal or a CAST as its base, so
NOW() becomes a datetime literal floored in the client's Jinja template:
{{ end_datetime }} (UTC) with its seconds floored for T dividing 60, or its
minutes for 60 | T | 3600. Other intervals raise.

All 936 distinct structured queries in sketch-bench cd51be9's tables
(T = 60 s and 300 s), rendered through the client's QueryTemplate at three
end times, parse in ASAPQuery's SQL frontend to (start, duration) =
(floor(E) - W, W) and pass its matcher.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae

Author: zzylol

asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py

@@ -11,7 +11,9 @@
 For a table `<stem>.json` this writes, under OUT_CONFIG_DIR:
 - <stem>/init.sql: the metric tables;
 - <stem>/interval_<T>s.sql: one statement per query that repeats every T s;
-- <stem>/asapquery.sql: the structured queries in ASAPQuery's SQL dialect;
+- <stem>/asapquery.sql: each structured entry's `sql` (ASAPQuery's SQL, as
+  sketch-bench's exporter writes it) with NOW() replaced by the floored eval
+  time;
 - <stem>_clickhouse.yaml: a baseline-only experiment_run_clickhouse.py config
   with one query group per T, repeating every T seconds.
 
@@ -310,30 +312,37 @@ def raw_sql(q: Query, labels: Dict[str, List[str]]) -> str:
     )
 
 
-def asapquery_sql(q: Query) -> str:
-    """A structured query in the dialect of ASAPQuery's SQL frontend.
+def floored_end_datetime(interval_secs: int) -> str:
+    """A Jinja template for the client's {{ end_datetime }} floored to a
+    multiple of the interval T, as the ClickHouse SQL floors {{ end_time }}.
 
-    The frontend takes a datetime literal but cannot floor it, so the window
-    ends `{{ end_time }} % T` seconds before the client's {{ end_datetime }}
-    (UTC; the frontend reads timezone-naive literals as server-local time):
-    the same floored T as the ClickHouse SQL. The frontend maps the BETWEEN to
-    (start, duration) = (T - W, W) and answers over (T - W, T], as the
-    ClickHouse SQL does for structured queries.
+    end_datetime is UTC 'YYYY-MM-DD HH:MM:SS', and UTC timestamps are
+    multiples of 60 exactly at whole minutes, so flooring rewrites the seconds
+    (T divides 60) or the minutes (T divides 3600 and is a multiple of 60).
     """
-    group = ", ".join(q.grouping)
-    if q.agg == "cardinality":
-        agg = f"COUNT(DISTINCT {q.value})"
-    elif q.agg == "quantile":
-        agg = f"QUANTILE({q.quantile}, {q.value})"
-    else:
-        raise ValueError(f"no ASAPQuery SQL for {q}")
-    t = q.interval_secs
-    end = "'{{ end_datetime }}'"
-    return (
-        f"SELECT {group}, {agg} FROM {q.table} WHERE ts BETWEEN "
-        f"DATEADD(s, -{{{{ end_time % {t} + {q.window_secs} }}}}, {end}) AND "
-        f"DATEADD(s, -{{{{ end_time % {t} }}}}, {end}) GROUP BY {group}"
-    )
+    t = interval_secs
+    if 60 % t == 0:
+        floored = f"end_time % 60 // {t} * {t}"
+        return f'{{{{ end_datetime[:17] }}}}{{{{ "%02d" | format({floored}) }}}}'
+    if t % 60 == 0 and 3600 % t == 0:
+        floored = f"end_time % 3600 // {t} * {t} // 60"
+        return f'{{{{ end_datetime[:14] }}}}{{{{ "%02d" | format({floored}) }}}}:00'
+    raise ValueError(f"cannot floor a datetime to {t}s: T must divide 60 or 3600")
+
+
+def asapquery_sql(entry: dict) -> str:
+    """A structured entry's `sql` with NOW() replaced by the floored eval time.
+
+    ASAPQuery's SQL frontend takes a datetime literal but cannot floor it, so
+    the literal is floored in the client's template (UTC; the frontend reads
+    timezone-naive literals as server-local time). The frontend maps
+    `ts BETWEEN DATEADD(s, -W, E) AND E` to (start, duration) = (E - W, W) and
+    answers over (E - W, E], as the ClickHouse SQL does for structured queries.
+    """
+    if "sql" not in entry:
+        raise ValueError(f"structured entry without its ASAPQuery sql: {entry}")
+    end = floored_end_datetime(entry["interval_secs"])
+    return entry["sql"].replace("NOW()", f"'{end}'")
 
 
 def value_columns(desc: dict) -> List[str]:
@@ -393,7 +402,7 @@ def convert(
             f"-- {describe(entry)}\n{raw_sql(q, labels)}"
         )
         if "promql" not in entry:
-            asapquery.append(f"-- {describe(entry)}\n{asapquery_sql(q)}")
+            asapquery.append(f"-- {describe(entry)}\n{asapquery_sql(entry)}")
     if asapquery:
         # For the ASAPQuery arm; the runner does not read it.
         write_statements(os.path.join(sql_dir, "asapquery.sql"), asapquery)

asap-tools/experiments/tests/test_sketch_bench_workloads.py

@@ -6,6 +6,7 @@
 CLICKHOUSE_BINARY) and is skipped otherwise.
 """
 
+import datetime
 import itertools
 import json
 import math
@@ -18,6 +19,7 @@
 import unittest
 from typing import Callable, Dict, List, Tuple
 
+import jinja2
 import yaml
 
 sys.path.insert(
@@ -164,11 +166,43 @@ def mg_entry(capability: str, grouping: List[str], window_secs: int) -> dict:
         "interval_secs": 60,
         "rqes": [],
     }
+    agg = "COUNT(DISTINCT user_id)"
     if capability == "quantile":
         entry["quantile"] = 0.99
+        agg = "QUANTILE(0.99, latency)"
+    group = ", ".join(grouping)
+    entry["sql"] = (
+        f"SELECT {group}, {agg} FROM http WHERE ts BETWEEN "
+        f"DATEADD(s, -{window_secs}, NOW()) AND NOW() GROUP BY {group}"
+    )
     return entry
 
 
+class FlooredEndDatetimeTest(unittest.TestCase):
+    def test_renders_floored_utc_datetime(self):
+        env = jinja2.Environment()
+        for t in (1, 15, 60, 300, 3600):
+            template = env.from_string(sbw.floored_end_datetime(t))
+            for end in (1767229237, 1767229200, 1767225599, 1767232799):
+                with self.subTest(t=t, end=end):
+                    # As the client's TimeRange.end_datetime formats it.
+                    got = template.render(
+                        end_time=end,
+                        end_datetime=datetime.datetime.fromtimestamp(
+                            end, tz=datetime.timezone.utc
+                        ).strftime("%Y-%m-%d %H:%M:%S"),
+                    )
+                    want = datetime.datetime.fromtimestamp(
+                        end // t * t, tz=datetime.timezone.utc
+                    )
+                    self.assertEqual(got, want.strftime("%Y-%m-%d %H:%M:%S"))
+
+    def test_unfloorable_interval_raises(self):
+        for t in (7, 420, 7200):
+            with self.subTest(t=t), self.assertRaises(ValueError):
+                sbw.floored_end_datetime(t)
+
+
 class ConvertTest(unittest.TestCase):
     def test_writes_config_grouped_by_interval(self):
         e2e = {
@@ -215,14 +249,14 @@ def test_writes_config_grouped_by_interval(self):
                     init,
                 )
                 sql_dir = os.path.dirname(g60["sql_file"]["baseline"])
+                # The table's sql as written, with only NOW() substituted.
+                end = "'" + sbw.floored_end_datetime(60) + "'"
                 with open(os.path.join(sql_dir, "asapquery.sql")) as f:
                     self.assertEqual(
                         f.read(),
                         "-- quantile(0.99, latency) by (service) over 300s\n"
-                        "SELECT service, QUANTILE(0.99, latency) FROM http WHERE ts "
-                        "BETWEEN DATEADD(s, -{{ end_time % 60 + 300 }}, "
-                        "'{{ end_datetime }}') AND DATEADD(s, -{{ end_time % 60 }}, "
-                        "'{{ end_datetime }}') GROUP BY service;\n",
+                        + e2e["queries"][-1]["sql"].replace("NOW()", end)
+                        + ";\n",
                     )
             finally:
                 os.chdir(cwd)