From 9d6727fb21affa347f53defb3752adf2d5a8631a Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 12 Oct 2026 13:09:48 -0400 Subject: [PATCH 01/10] feat(experiments): fake_exporter-equivalent dataset for ClickHouse generated2 workloads Co-Authored-By: Claude Opus 5.5 --- .../datasets/fake_metric/generate_data.py | 84 +++++++++++++++++++ .../experiments/datasets/fake_metric/init.sql | 14 ++++ 2 files changed, 98 insertions(+) create mode 100644 asap-tools/experiments/datasets/fake_metric/generate_data.py create mode 100644 asap-tools/experiments/datasets/fake_metric/init.sql diff --git a/asap-tools/experiments/datasets/fake_metric/generate_data.py b/asap-tools/experiments/datasets/fake_metric/generate_data.py new file mode 100644 index 00000000..fbbaae6d --- /dev/null +++ b/asap-tools/experiments/datasets/fake_metric/generate_data.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Generate fake_exporter-equivalent data for the ClickHouse generated2 workloads. + +Mirrors the generated2 Prometheus setup: one fake_exporter per instance +(zipf, valuescale 10000, 3 labels x 20 values, counter), scraped every 1s. +Values match fake_exporter in distribution, not sample-for-sample: +- each scrape draws one bounded Zipf(n=VALUE_SCALE, s=ZIPF_ALPHA) sample per + series and adds it to the instance's single shared counter, in series order + (fake_exporter's counter_state is per process, not per series); +- every series of every instance is stamped with the same 1s-aligned tick. + +Ticks are DATA_START + t for t in [0, duration). Writes gzipped JSON lines +with columns matching init.sql. +""" +import argparse +import gzip +import itertools +from typing import List + +import numpy as np + +DATA_START = 1767225600 # 2026-01-01 00:00:00 UTC; the SQL translator anchors T here +START_PORT = 50000 +JOB = "fake_exporter" +NUM_LABELS = 3 +NUM_VALUES_PER_LABEL = 20 +VALUE_SCALE = 10000 +ZIPF_ALPHA = 1.01 +# Last query in generated2 fires at T = starting_delay 70 + 1149 reps - offset 10. +DEFAULT_DURATION_SECONDS = 1210 + + +def zipf_cdf(n: int, alpha: float) -> np.ndarray: + weights = np.arange(1, n + 1, dtype=np.float64) ** -alpha + cdf = np.cumsum(weights) / weights.sum() + cdf[-1] = 1.0 # float rounding must not let searchsorted step past n + return cdf + + +def series_prefixes(instance: str) -> List[str]: + combos = itertools.product(range(NUM_VALUES_PER_LABEL), repeat=NUM_LABELS) + return [ + f'{{"instance":"{instance}","job":"{JOB}",' + + ",".join(f'"label_{i}":"value_{i}_value_{v}"' for i, v in enumerate(combo)) + + "," + for combo in combos + ] + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--out", required=True, help="output .jsonl.gz path") + parser.add_argument("--num-instances", type=int, default=10) + parser.add_argument("--duration", type=int, default=DEFAULT_DURATION_SECONDS) + parser.add_argument("--seed", type=int, default=0) + args = parser.parse_args() + + rng = np.random.default_rng(args.seed) + cdf = zipf_cdf(VALUE_SCALE, ZIPF_ALPHA) + prefixes = [ + series_prefixes(f"localhost:{START_PORT + i}") + for i in range(args.num_instances) + ] + num_series = len(prefixes[0]) + counters = np.zeros(args.num_instances) + + # ponytail: gzip level 1, the loader re-parses every line so CPU here is not the bottleneck + with gzip.open(args.out, "wt", compresslevel=1) as f: + for t in range(args.duration): + ts = DATA_START + t + samples = ( + np.searchsorted(cdf, rng.random((args.num_instances, num_series))) + 1.0 + ) + values = counters[:, None] + np.cumsum(samples, axis=1) + counters = values[:, -1] + f.writelines( + f'{p}"ts":{ts},"value":{v!r}}}\n' + for inst_prefixes, inst_values in zip(prefixes, values.tolist()) + for p, v in zip(inst_prefixes, inst_values) + ) + + +if __name__ == "__main__": + main() diff --git a/asap-tools/experiments/datasets/fake_metric/init.sql b/asap-tools/experiments/datasets/fake_metric/init.sql new file mode 100644 index 00000000..ea6b7b1b --- /dev/null +++ b/asap-tools/experiments/datasets/fake_metric/init.sql @@ -0,0 +1,14 @@ +DROP TABLE IF EXISTS fake_metric_total; + +CREATE TABLE fake_metric_total +( + ts DateTime, + instance LowCardinality(String), + job LowCardinality(String), + label_0 LowCardinality(String), + label_1 LowCardinality(String), + label_2 LowCardinality(String), + value Float64 +) +ENGINE = MergeTree +ORDER BY (instance, label_0, label_1, label_2, ts); From dfa8022c1ca09b0d28cf25764837c30c4a0b23e6 Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Mon, 12 Oct 2026 13:17:21 -0400 Subject: [PATCH 02/10] feat(experiments): translate generated2 PromQL workloads to ClickHouse SQL configs Adds a client_options.simulated_start_time so templated SQL can be evaluated against bulk-loaded data with fixed timestamps. Co-Authored-By: Claude Opus 5.5 --- .../fake_metric/generate_workloads.py | 206 ++++++++++++++++++ .../main_prometheus_client.py | 13 +- 2 files changed, 218 insertions(+), 1 deletion(-) create mode 100644 asap-tools/experiments/datasets/fake_metric/generate_workloads.py diff --git a/asap-tools/experiments/datasets/fake_metric/generate_workloads.py b/asap-tools/experiments/datasets/fake_metric/generate_workloads.py new file mode 100644 index 00000000..a68bb8b7 --- /dev/null +++ b/asap-tools/experiments/datasets/fake_metric/generate_workloads.py @@ -0,0 +1,206 @@ +#!/usr/bin/env python3 +"""Translate generated2 PromQL workloads into ClickHouse experiment configs. + +For each generated2 workload yaml (Prometheus variants only), writes a +baseline-only ClickHouse experiment config plus one templated SQL file per +query group under config/experiment_type/generated2_clickhouse/ (gitignored, +like its generated2 inputs). Query groups keep their +original delays, repetitions and query_time_offset; the client fills in +{{ end_time }} (= T, the PromQL evaluation time) per repetition. + +Semantics follow Prometheus 2.53 with these windows: +- instant vector at T: samples in (T - SCRAPE_INTERVAL_SECONDS, T], i.e. the + one scrape at T; +- range vector [d] at T: samples in (T - d, T]. +rate/increase reproduce extrapolatedRate. Only the query shapes that appear +in generated2 are supported; anything else raises. + +Run from asap-tools/experiments: + python3 datasets/fake_metric/generate_workloads.py \ + --src-dir config/experiment_type/generated2 --data-file +""" +import argparse +import glob +import os +import re +from typing import List + +import yaml + +from generate_data import DATA_START + +SCRAPE_INTERVAL_SECONDS = 1 # config/config.yaml prometheus.scrape_interval +OUT_CONFIG_DIR = "config/experiment_type/generated2_clickhouse" +INIT_SQL = "datasets/fake_metric/init.sql" +END = "{{ end_time }}" + +DURATION_UNITS = {"s": 1, "m": 60, "h": 3600} +RANGE = r"(\w+)\[(\d+)([smh])\]" +SHAPES = [ + ("instant_agg", re.compile(r"^(count|sum) by \((\w+)\) \((\w+)\)$")), + ("instant_quantile", re.compile(r"^quantile by \((\w+)\) \(([\d.]+), (\w+)\)$")), + ( + "agg_over_time", + re.compile(rf"^sum by \((\w+)\) \((sum|count)_over_time\({RANGE}\)\)$"), + ), + ("over_time", re.compile(rf"^(sum|count)_over_time\({RANGE}\)$")), + ("quantile_over_time", re.compile(rf"^quantile_over_time\(([\d.]+), {RANGE}\)$")), + ("rate", re.compile(rf"^(rate|increase)\({RANGE}\)$")), +] +AGG_SQL = {"sum": "sum(value)", "count": "count()"} + + +def window_seconds(amount: str, unit: str) -> int: + return int(amount) * DURATION_UNITS[unit] + + +def select(cols: str, agg: str, table: str, window: int) -> str: + return ( + f"SELECT {cols}, {agg} AS value FROM {table} " + f"WHERE ts > {END} - {window} AND ts <= {END} GROUP BY {cols}" + ) + + +def extrapolated_rate(cols: str, table: str, window: int, is_rate: bool) -> str: + """Prometheus 2.53 extrapolatedRate, isCounter=true.""" + # ponytail: no counter-reset correction, fake_exporter counters only grow + samples = ( + f"SELECT {cols}, argMin(value, ts) AS first_v, argMax(value, ts) AS last_v, " + f"toFloat64(toUnixTimestamp(min(ts))) AS first_t, " + f"toFloat64(toUnixTimestamp(max(ts))) AS last_t, count() AS n " + f"FROM {table} WHERE ts > {END} - {window} AND ts <= {END} " + f"GROUP BY {cols} HAVING n >= 2" + ) + durations = ( + f"SELECT {cols}, last_v - first_v AS delta, first_v, " + f"last_t - first_t AS sampled, sampled / (n - 1) AS avg_step, " + f"if(first_t - ({END} - {window}) >= 1.1 * avg_step, avg_step / 2, " + f"first_t - ({END} - {window})) AS to_start_clamped, " + f"if(delta > 0 AND first_v >= 0, " + f"least(to_start_clamped, sampled * first_v / delta), to_start_clamped) AS to_start, " + f"if({END} - last_t >= 1.1 * avg_step, avg_step / 2, {END} - last_t) AS to_end " + f"FROM ({samples})" + ) + per_second = f" / {window}" if is_rate else "" + return ( + f"SELECT {cols}, delta * (sampled + to_start + to_end) / sampled{per_second} " + f"AS value FROM ({durations})" + ) + + +def translate(promql: str, metric: str, series_cols: str) -> str: + query = " ".join(promql.split()) + for shape, pattern in SHAPES: + m = pattern.match(query) + if not m: + continue + g = m.groups() + if shape == "instant_agg": + fn, by, name = g + agg, cols, window = AGG_SQL[fn], by, SCRAPE_INTERVAL_SECONDS + elif shape == "instant_quantile": + by, q, name = g + agg, cols, window = ( + f"quantileExactInclusive({q})(value)", + by, + SCRAPE_INTERVAL_SECONDS, + ) + elif shape == "agg_over_time": + by, fn, name, amount, unit = g + agg, cols, window = AGG_SQL[fn], by, window_seconds(amount, unit) + elif shape == "over_time": + fn, name, amount, unit = g + agg, cols, window = AGG_SQL[fn], series_cols, window_seconds(amount, unit) + elif shape == "quantile_over_time": + q, name, amount, unit = g + agg, cols, window = ( + f"quantileExactInclusive({q})(value)", + series_cols, + window_seconds(amount, unit), + ) + else: + fn, name, amount, unit = g + if name != metric: + raise ValueError(f"unexpected metric {name!r} in {promql!r}") + return extrapolated_rate( + series_cols, metric, window_seconds(amount, unit), fn == "rate" + ) + if name != metric: + raise ValueError(f"unexpected metric {name!r} in {promql!r}") + return select(cols, agg, metric, window) + raise ValueError(f"unsupported PromQL shape: {promql!r}") + + +def convert(src_path: str, data_file: str) -> str: + with open(src_path) as f: + src = yaml.safe_load(f) + (metric_cfg,) = src["metrics"] + metric = metric_cfg["metric"] + series_cols = ", ".join(metric_cfg["labels"]) + stem = os.path.splitext(os.path.basename(src_path))[0] + sql_dir = os.path.join(OUT_CONFIG_DIR, stem) + os.makedirs(sql_dir, exist_ok=True) + + query_groups = [] + for group in src["query_groups"]: + sql_path = os.path.join(sql_dir, f"group_{group['id']}.sql") + statements: List[str] = [ + f"-- {q}\n{translate(q, metric, series_cols)}" for q in group["queries"] + ] + with open(sql_path, "w") as f: + f.write(";\n\n".join(statements) + ";\n") + query_groups.append( + { + "name": f"group_{group['id']}", + "sql_file": {"baseline": sql_path}, + "repetition_delay_ms": group["repetition_delay_ms"], + # Queries use only {{ end_time }}; the client still requires a value. + "time_window_seconds": 0, + "client_options": { + **group["client_options"], + "simulated_start_time": DATA_START, + }, + "controller_options": group["controller_options"], + } + ) + + out = { + "experiment": [{"mode": "baseline", "server": "clickhouse"}], + "servers": [{"name": "clickhouse", "url": "http://localhost:8123"}], + "dataset": { + "name": "custom", + "local_data_file": data_file, + "table": metric, + "init_sql_file": INIT_SQL, + "max_rows": 0, + }, + "query_groups": query_groups, + "data_ingestion_interval_ms": SCRAPE_INTERVAL_SECONDS * 1000, + } + out_path = os.path.join(OUT_CONFIG_DIR, f"{stem}_clickhouse.yaml") + with open(out_path, "w") as f: + f.write(f"# @package experiment_params\n# Generated from {src_path}\n") + yaml.safe_dump(out, f, sort_keys=False) + return out_path + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--src-dir", required=True, help="generated2 config dir") + parser.add_argument( + "--data-file", + required=True, + help="generate_data.py output; path as seen from asap-tools/experiments", + ) + args = parser.parse_args() + sources = sorted( + p + for p in glob.glob(os.path.join(args.src_dir, "*.yaml")) + if not p.endswith("_victoriametrics.yaml") + ) + for src in sources: + print(convert(src, args.data_file)) + + +if __name__ == "__main__": + main() diff --git a/asap-tools/queriers/prometheus-client/main_prometheus_client.py b/asap-tools/queriers/prometheus-client/main_prometheus_client.py index 104e4d68..84be87f4 100644 --- a/asap-tools/queriers/prometheus-client/main_prometheus_client.py +++ b/asap-tools/queriers/prometheus-client/main_prometheus_client.py @@ -316,8 +316,19 @@ def handle_query_group( for server_name in servers } + simulated_start_time = getattr(query_group, "simulated_start_time", None) + for repetition_idx in range(query_group.repetitions): - current_time = datetime.datetime.now() + if simulated_start_time is None: + current_time = datetime.datetime.now() + else: + # Pre-loaded data has fixed timestamps, so query time follows a + # clock anchored at the data start instead of the wall clock. + current_time = datetime.datetime.fromtimestamp( + simulated_start_time + + query_group.starting_delay + + repetition_idx * query_group.repetition_delay_ms / 1000 + ) logger.debug("Current unix time: {}", int(current_time.timestamp())) if hasattr(query_group, "query_time_offset"): From f2a24f406fca8add4c51c9a21c5a053c6b1eff5f Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:17:48 +0000 Subject: [PATCH 03/10] feat(asap-tools): translate sketch-bench e2e PromQL to ClickHouse SQL configs sketch_bench_workloads.py reads a sketch-bench table's workloads[0].e2e block and writes, per table, the metric tables' init SQL, one SQL file per query interval and a baseline experiment_run_clickhouse.py config whose query groups repeat every interval. Queries keep generate_workloads.py's Prometheus 2.53 semantics (range vectors read (T - d, T], extrapolatedRate for rate/increase, quantileExactInclusive) and add the classic template shapes: topk over sum by, sum/count/quantile by over instant vectors (latest sample per series in (T - 1s, T]) and rates, and the ratio of two quantile_over_time. Eval time is {{ end_time }} floored to the interval. Unsupported shapes and structured (non-PromQL) entries raise. generate_workloads.extrapolated_rate now builds on rate_from_samples, which the new module reuses; its SQL output is unchanged. The test compares every generated statement in clickhouse-local against a Python implementation of the Prometheus semantics at several eval times, covering empty windows and every extrapolation branch. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../fake_metric/generate_workloads.py | 15 +- .../fake_metric/sketch_bench_workloads.py | 351 ++++++++++++++ .../tests/test_sketch_bench_workloads.py | 430 ++++++++++++++++++ 3 files changed, 793 insertions(+), 3 deletions(-) create mode 100644 asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py create mode 100644 asap-tools/experiments/tests/test_sketch_bench_workloads.py diff --git a/asap-tools/experiments/datasets/fake_metric/generate_workloads.py b/asap-tools/experiments/datasets/fake_metric/generate_workloads.py index a68bb8b7..72cc0f39 100644 --- a/asap-tools/experiments/datasets/fake_metric/generate_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/generate_workloads.py @@ -71,14 +71,23 @@ def extrapolated_rate(cols: str, table: str, window: int, is_rate: bool) -> str: f"FROM {table} WHERE ts > {END} - {window} AND ts <= {END} " f"GROUP BY {cols} HAVING n >= 2" ) + return rate_from_samples(cols, samples, END, window, is_rate) + + +def rate_from_samples( + cols: str, samples: str, end: str, window: int, is_rate: bool +) -> str: + """extrapolatedRate over `samples`, a query yielding per series `cols`, + first_v, last_v, first_t, last_t (unix seconds) and n >= 2 samples in + (end - window, end].""" durations = ( f"SELECT {cols}, last_v - first_v AS delta, first_v, " f"last_t - first_t AS sampled, sampled / (n - 1) AS avg_step, " - f"if(first_t - ({END} - {window}) >= 1.1 * avg_step, avg_step / 2, " - f"first_t - ({END} - {window})) AS to_start_clamped, " + f"if(first_t - ({end} - {window}) >= 1.1 * avg_step, avg_step / 2, " + f"first_t - ({end} - {window})) AS to_start_clamped, " f"if(delta > 0 AND first_v >= 0, " f"least(to_start_clamped, sampled * first_v / delta), to_start_clamped) AS to_start, " - f"if({END} - last_t >= 1.1 * avg_step, avg_step / 2, {END} - last_t) AS to_end " + f"if({end} - last_t >= 1.1 * avg_step, avg_step / 2, {end} - last_t) AS to_end " f"FROM ({samples})" ) per_second = f" / {window}" if is_rate else "" diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py new file mode 100644 index 00000000..7dbbc917 --- /dev/null +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -0,0 +1,351 @@ +#!/usr/bin/env python3 +"""Translate a sketch-bench table's end-to-end queries into ClickHouse configs. + +A sketch-bench table's `workloads[0].e2e` block lists its queries as +`{promql, interval_secs, rqes}` and describes each metric (`metrics`). Every +metric is one ClickHouse table named after it: `ts DateTime64(3)`, then the +metric's labels in description order as String, then `value Float64`. + +For a table `.json` this writes, under OUT_CONFIG_DIR: +- /init.sql: the metric tables; +- /interval_s.sql: one statement per query that repeats every T s; +- _clickhouse.yaml: a baseline-only experiment_run_clickhouse.py config + with one query group per T, repeating every T seconds. + +Semantics are generate_workloads.py's (Prometheus 2.53, left-open windows): +range vectors [d] at T read (T - d, T]; instant vectors read the latest sample +of each series in (T - 1s, T]; quantiles use quantileExactInclusive; rate and +increase reproduce extrapolatedRate (counters never reset). T is the client's +{{ end_time }} floored to a multiple of the query's interval. Unsupported +query shapes raise. + +Run from asap-tools/experiments: + python3 datasets/fake_metric/sketch_bench_workloads.py \ + --table --data-file \ + --starting-delay-secs --duration-secs +""" +import argparse +import json +import os +import re +from dataclasses import dataclass, replace +from typing import Callable, Dict, List, Optional, Tuple + +import yaml + +from generate_workloads import ( + END, + RANGE, + SCRAPE_INTERVAL_SECONDS, + rate_from_samples, + window_seconds, +) + +OUT_CONFIG_DIR = "config/experiment_type/sketch_bench_clickhouse" +CLICKHOUSE_URL = "http://localhost:8123" +VALUE_COLUMN = "value" +# Instant vectors read the latest sample of each series in the last scrape. +INSTANT_WINDOW_SECONDS = SCRAPE_INTERVAL_SECONDS + +LABELS = r"(\w+(?:, \w+)*)" +LEVEL = r"([\d.]+)" +TOPK = re.compile(r"^topk\((\d+), (.+)\)$") +AGG_BY = re.compile(rf"^(sum|count) by \({LABELS}\) \((.+)\)$") +QUANTILE_BY = re.compile(rf"^quantile by \({LABELS}\) \({LEVEL}, (.+)\)$") +OVER_TIME = re.compile(rf"^(sum|count)_over_time\({RANGE}\)$") +QUANTILE_OVER_TIME = re.compile(rf"^quantile_over_time\({LEVEL}, {RANGE}\)$") +QUANTILE_RATIO = re.compile( + rf"^quantile_over_time\({LEVEL}, {RANGE}\) / quantile_over_time\({LEVEL}, {RANGE}\)$" +) +COUNTER = re.compile(rf"^(rate|increase)\({RANGE}\)$") +SELECTOR = re.compile(r"^(\w+)$") + + +@dataclass(frozen=True) +class Query: + """One query as the SQL renderers take it. + + With `series_fn` None, `agg` runs over the raw samples of each `grouping` + group in the window (sum_over_time, sum by (job) (sum_over_time), ...). + Otherwise `series_fn` (last, rate, increase) gives one value per series, + and `agg` combines those per group; `agg` None keeps them per series. + `ratio_quantile` divides the `quantile` by a second quantile of the same + samples. Structured (non-PromQL) query entries map onto these fields. + """ + + table: str + interval_secs: int + window_secs: int + grouping: Tuple[str, ...] + series_fn: Optional[str] + agg: Optional[str] + quantile: Optional[str] = None # level as written, e.g. "0.99" + ratio_quantile: Optional[str] = None + topk: Optional[int] = None + value: str = VALUE_COLUMN + + +def _parse_series( + text: str, interval: int, labels: Dict[str, List[str]], promql: str +) -> Query: + """A per-series expression: a selector, *_over_time, rate or increase.""" + + def per_series(name, window, series_fn, agg, quantile=None, ratio=None): + if name not in labels: + raise ValueError(f"unknown metric {name!r} in {promql!r}") + return Query( + name, + interval, + window, + tuple(labels[name]), + series_fn, + agg, + quantile, + ratio, + ) + + m = SELECTOR.match(text) + if m: + return per_series(m[1], INSTANT_WINDOW_SECONDS, "last", None) + m = OVER_TIME.match(text) + if m: + fn, name, amount, unit = m.groups() + return per_series(name, window_seconds(amount, unit), None, fn) + m = QUANTILE_OVER_TIME.match(text) + if m: + q, name, amount, unit = m.groups() + return per_series(name, window_seconds(amount, unit), None, "quantile", q) + m = QUANTILE_RATIO.match(text) + if m: + q, name, amount, unit, q2, name2, amount2, unit2 = m.groups() + if (name, amount, unit) != (name2, amount2, unit2): + raise ValueError(f"ratio of different range vectors: {promql!r}") + window = window_seconds(amount, unit) + return per_series(name, window, None, "quantile", q, q2) + m = COUNTER.match(text) + if m: + fn, name, amount, unit = m.groups() + return per_series(name, window_seconds(amount, unit), fn, None) + raise ValueError(f"unsupported PromQL shape: {promql!r}") + + +def _grouped(inner: Query, agg: str, by: str, quantile: Optional[str]) -> Query: + grouping = tuple(by.split(", ")) + unknown = set(grouping) - set(inner.grouping) + if unknown: + raise ValueError(f"unknown labels {sorted(unknown)} for {inner.table!r}") + if inner.series_fn is not None and inner.agg is None: + return replace(inner, grouping=grouping, agg=agg, quantile=quantile) + # sum by of a pooled sum/count is the pooled sum/count of the larger group. + if agg == "sum" and inner.agg in ("sum", "count") and inner.ratio_quantile is None: + return replace(inner, grouping=grouping) + raise ValueError(f"cannot aggregate {inner} by {agg}") + + +def parse_promql(promql: str, interval: int, labels: Dict[str, List[str]]) -> Query: + """`labels`: each metric's label names in description order.""" + text = " ".join(promql.split()) + m = TOPK.match(text) + if m: + inner = parse_promql(m[2], interval, labels) + if inner.topk is not None: + raise ValueError(f"nested topk: {promql!r}") + return replace(inner, topk=int(m[1])) + m = AGG_BY.match(text) + if m: + agg, by, inner = m.groups() + return _grouped(_parse_series(inner, interval, labels, promql), agg, by, None) + m = QUANTILE_BY.match(text) + if m: + by, q, inner = m.groups() + series = _parse_series(inner, interval, labels, promql) + return _grouped(series, "quantile", by, q) + return _parse_series(text, interval, labels, promql) + + +def query_from_entry(entry: dict, labels: Dict[str, List[str]]) -> Query: + """One `e2e.queries` entry. Entries are PromQL; structured entries (no + `promql`) are not supported yet and raise.""" + if "promql" not in entry: + raise ValueError(f"query entry without promql is not supported: {entry}") + return parse_promql(entry["promql"], entry["interval_secs"], labels) + + +def eval_time(interval_secs: int) -> str: + """The client's {{ end_time }} floored to a multiple of the interval.""" + return f"intDiv({END}, {interval_secs}) * {interval_secs}" + + +def render( + q: Query, series: Tuple[str, ...], source: str, time_col: str, agg: Callable +) -> str: + """SQL for `q` reading `source` rows with `time_col` in the window. + + `series` is the table's full label list; `agg(stat, quantile)` gives the + aggregate expression of a statistic over the source rows: sum, count, + quantile, last (per-series latest value) or counter (the select list + rate_from_samples reads). + """ + end = eval_time(q.interval_secs) + window = ( + f"FROM {source} WHERE {time_col} > {end} - {q.window_secs} " + f"AND {time_col} <= {end}" + ) + group = ", ".join(q.grouping) + cols = ", ".join(series) + if q.series_fn is None: + value = agg(q.agg, q.quantile) + if q.ratio_quantile is not None: + value = f"{value} / {agg('quantile', q.ratio_quantile)}" + sql = f"SELECT {group}, {value} AS value {window} GROUP BY {group}" + else: + if q.series_fn == "last": + sql = ( + f"SELECT {cols}, {agg('last', None)} AS value {window} GROUP BY {cols}" + ) + else: + samples = ( + f"SELECT {cols}, {agg('counter', None)} {window} " + f"GROUP BY {cols} HAVING n >= 2" + ) + is_rate = q.series_fn == "rate" + sql = rate_from_samples(cols, samples, end, q.window_secs, is_rate) + if q.agg is not None: + combine = raw_agg(q.agg, q.quantile) + sql = f"SELECT {group}, {combine} AS value FROM ({sql}) GROUP BY {group}" + if q.topk is not None: + sql += f" ORDER BY value DESC, {group} LIMIT {q.topk}" + return sql + + +def raw_agg(stat: str, quantile: Optional[str], value: str = VALUE_COLUMN) -> str: + if stat == "sum": + return f"sum({value})" + if stat == "count": + return "count()" + if stat == "quantile": + return f"quantileExactInclusive({quantile})({value})" + if stat == "last": + return f"argMax({value}, ts)" + if stat == "counter": + return ( + f"argMin({value}, ts) AS first_v, argMax({value}, ts) AS last_v, " + "toFloat64(min(ts)) AS first_t, toFloat64(max(ts)) AS last_t, count() AS n" + ) + raise ValueError(f"unknown statistic {stat!r}") + + +def raw_sql(q: Query, labels: Dict[str, List[str]]) -> str: + """`q` over the raw metric table.""" + return render( + q, + tuple(labels[q.table]), + q.table, + "ts", + lambda stat, quantile: raw_agg(stat, quantile, q.value), + ) + + +def init_sql(labels: Dict[str, List[str]]) -> str: + statements = [] + for table, names in labels.items(): + columns = ", ".join(f"{name} String" for name in names) + statements += [ + f"DROP TABLE IF EXISTS {table}", + f"CREATE TABLE {table} (ts DateTime64(3), {columns}, " + f"{VALUE_COLUMN} Float64) ENGINE = MergeTree " + f"ORDER BY ({', '.join(names)}, ts)", + ] + return ";\n\n".join(statements) + ";\n" + + +def metric_labels(e2e: dict) -> Dict[str, List[str]]: + return { + name: [label["name"] for label in desc["labels"]] + for name, desc in e2e["metrics"].items() + } + + +def convert( + table_path: str, data_file: str, starting_delay_secs: int, duration_secs: int +) -> str: + with open(table_path) as f: + (workload,) = json.load(f)["workloads"] + e2e = workload["e2e"] + labels = metric_labels(e2e) + stem = os.path.splitext(os.path.basename(table_path))[0] + sql_dir = os.path.join(OUT_CONFIG_DIR, stem) + os.makedirs(sql_dir, exist_ok=True) + + init_path = os.path.join(sql_dir, "init.sql") + with open(init_path, "w") as f: + f.write(init_sql(labels)) + + by_interval: Dict[int, List[str]] = {} + for entry in e2e["queries"]: + q = query_from_entry(entry, labels) + by_interval.setdefault(q.interval_secs, []).append( + f"-- {entry['promql']}\n{raw_sql(q, labels)}" + ) + + query_groups = [] + for interval, statements in sorted(by_interval.items()): + name = f"interval_{interval}s" + sql_path = os.path.join(sql_dir, f"{name}.sql") + with open(sql_path, "w") as f: + f.write(";\n\n".join(statements) + ";\n") + query_groups.append( + { + "name": name, + "sql_file": {"baseline": sql_path}, + "repetition_delay_ms": interval * 1000, + # Queries use only {{ end_time }}; the client still requires a value. + "time_window_seconds": 0, + "client_options": { + "starting_delay": starting_delay_secs, + "repetitions": duration_secs // interval, + }, + } + ) + + out = { + "experiment": [{"mode": "baseline", "server": "clickhouse"}], + "servers": [{"name": "clickhouse", "url": CLICKHOUSE_URL}], + "dataset": { + "name": "custom", + "local_data_file": data_file, + # The runner's loader fills one table; the paced feeder writes all. + "table": next(iter(labels)), + "init_sql_file": init_path, + "max_rows": 0, + }, + "query_groups": query_groups, + "data_ingestion_interval_ms": SCRAPE_INTERVAL_SECONDS * 1000, + } + out_path = os.path.join(OUT_CONFIG_DIR, f"{stem}_clickhouse.yaml") + with open(out_path, "w") as f: + f.write(f"# @package experiment_params\n# Generated from {table_path}\n") + yaml.safe_dump(out, f, sort_keys=False) + return out_path + + +def main() -> None: + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("--table", required=True, help="sketch-bench table .json") + parser.add_argument( + "--data-file", required=True, help="dataset path as seen from experiments/" + ) + parser.add_argument("--starting-delay-secs", type=int, required=True) + parser.add_argument( + "--duration-secs", type=int, required=True, help="query phase length" + ) + args = parser.parse_args() + print( + convert( + args.table, args.data_file, args.starting_delay_secs, args.duration_secs + ) + ) + + +if __name__ == "__main__": + main() diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py new file mode 100644 index 00000000..79c7f086 --- /dev/null +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -0,0 +1,430 @@ +"""Tests for datasets/fake_metric/sketch_bench_workloads.py. + +ClickHouseSemanticsTest runs every generated statement in clickhouse-local +over a small dataset and compares it with a separate Python implementation of +the Prometheus semantics. It needs a `clickhouse` binary on PATH (or +CLICKHOUSE_BINARY) and is skipped otherwise. +""" + +import json +import math +import os +import random +import shutil +import subprocess +import sys +import tempfile +import unittest +from typing import Callable, Dict, List, Tuple + +import yaml + +sys.path.insert( + 0, os.path.join(os.path.dirname(__file__), "..", "datasets", "fake_metric") +) + +import sketch_bench_workloads as sbw # noqa: E402 + +LABELS = {"data": ["job", "label_0"], "data_total": ["job", "label_0"]} +CLICKHOUSE = os.environ.get("CLICKHOUSE_BINARY") or shutil.which("clickhouse") + +# The PromQL of sketch-bench's ten classic templates (windows vary per table). +CLASSIC = [ + (1, "sum by (job) (data)"), + (60, "topk(32, sum by (label_0) (sum_over_time(data[15m])))"), + (1, "quantile by (job) (0.9, data)"), + (60, "sum_over_time(data[15m])"), + (60, "quantile_over_time(0.99, data[15m])"), + (60, "rate(data_total[15m])"), + (60, "sum by (job) (rate(data_total[15m]))"), + (60, "sum by (job) (sum_over_time(data[15m]))"), + (60, "topk(32, sum by (label_0) (rate(data_total[15m])))"), + (60, "quantile_over_time(0.9, data[15m]) / quantile_over_time(0.5, data[15m])"), +] + + +class ParseTest(unittest.TestCase): + def test_classic_templates(self): + got = [sbw.parse_promql(p, t, LABELS) for t, p in CLASSIC] + series = ("job", "label_0") + self.assertEqual(got[0], sbw.Query("data", 1, 1, ("job",), "last", "sum")) + self.assertEqual( + got[1], sbw.Query("data", 60, 900, ("label_0",), None, "sum", topk=32) + ) + self.assertEqual( + got[2], sbw.Query("data", 1, 1, ("job",), "last", "quantile", "0.9") + ) + self.assertEqual(got[3], sbw.Query("data", 60, 900, series, None, "sum")) + self.assertEqual( + got[4], sbw.Query("data", 60, 900, series, None, "quantile", "0.99") + ) + self.assertEqual(got[5], sbw.Query("data_total", 60, 900, series, "rate", None)) + self.assertEqual( + got[6], sbw.Query("data_total", 60, 900, ("job",), "rate", "sum") + ) + self.assertEqual(got[7], sbw.Query("data", 60, 900, ("job",), None, "sum")) + self.assertEqual( + got[8], + sbw.Query("data_total", 60, 900, ("label_0",), "rate", "sum", topk=32), + ) + self.assertEqual( + got[9], sbw.Query("data", 60, 900, series, None, "quantile", "0.9", "0.5") + ) + + def test_metrics_m_names(self): + labels = {"data_3": ["job", "label_0"], "data_3_total": ["job", "label_0"]} + q = sbw.parse_promql("sum by (job) (rate(data_3_total[1h]))", 60, labels) + self.assertEqual((q.table, q.window_secs), ("data_3_total", 3600)) + + def test_unsupported_shapes_raise(self): + for promql in [ + "avg by (job) (data)", + "max_over_time(data[1m])", + # Not poolable: a count of per-series sums or a sum of quantiles. + "count by (job) (sum_over_time(data[1m]))", + "sum by (job) (quantile_over_time(0.5, data[1m]))", + "sum by (job) (quantile_over_time(0.9, data[1m]) / " + "quantile_over_time(0.5, data[1m]))", + "quantile by (job) (0.5, sum_over_time(data[1m]))", + "quantile_over_time(0.9, data[1m]) / quantile_over_time(0.5, data[2m])", + "topk(3, topk(2, data))", + "sum by (instance) (data)", + "rate(other_total[1m])", + ]: + with self.subTest(promql=promql): + with self.assertRaises(ValueError): + sbw.parse_promql(promql, 60, LABELS) + + def test_structured_entry_raises(self): + entry = {"table": "http", "capability": "quantile", "interval_secs": 60} + with self.assertRaises(ValueError): + sbw.query_from_entry(entry, LABELS) + + +class ConvertTest(unittest.TestCase): + def test_writes_config_grouped_by_interval(self): + e2e = { + "metrics": { + name: {"labels": [{"name": "job"}, {"name": "label_0"}]} + for name in LABELS + }, + "queries": [ + {"promql": p, "interval_secs": t, "rqes": []} for t, p in CLASSIC + ], + } + with tempfile.TemporaryDirectory() as tmp: + table = os.path.join(tmp, "classic.json") + with open(table, "w") as f: + json.dump({"workloads": [{"e2e": e2e}]}, f) + cwd = os.getcwd() + os.chdir(tmp) + try: + with open(sbw.convert(table, "data.json", 30, 600)) as f: + cfg = yaml.safe_load(f) + groups = {g["name"]: g for g in cfg["query_groups"]} + self.assertEqual(sorted(groups), ["interval_1s", "interval_60s"]) + g60 = groups["interval_60s"] + self.assertEqual(g60["repetition_delay_ms"], 60000) + self.assertEqual( + g60["client_options"], {"starting_delay": 30, "repetitions": 10} + ) + self.assertEqual( + groups["interval_1s"]["client_options"]["repetitions"], 600 + ) + with open(g60["sql_file"]["baseline"]) as f: + statements = [s for s in f.read().split(";") if s.strip()] + self.assertEqual(len(statements), 8) + with open(cfg["dataset"]["init_sql_file"]) as f: + self.assertIn("CREATE TABLE data_total (ts DateTime64(3)", f.read()) + finally: + os.chdir(cwd) + + +# --- Python reference of the Prometheus semantics ------------------------- + +Series = Tuple[str, ...] +Samples = List[Tuple[int, float]] # (ts ms, value), ascending +Vector = Dict[Series, float] +DATA_START_MS = 1767225600 * 1000 +INSTANT_MS = 1000 +BRANCHES_HIT = set() + + +def in_window(samples: Samples, t_ms: int, w_ms: int) -> Samples: + return [(t, v) for t, v in samples if t_ms - w_ms < t <= t_ms] + + +def prom_quantile(q: float, values: List[float]) -> float: + values = sorted(values) + rank = q * (len(values) - 1) + lower = max(0, math.floor(rank)) + upper = min(len(values) - 1, lower + 1) + weight = rank - math.floor(rank) + return values[lower] * (1 - weight) + values[upper] * weight + + +def extrapolated_rate(s: Samples, t_ms: int, w_ms: int, is_rate: bool) -> float: + """Prometheus 2.53 extrapolatedRate for a counter without resets.""" + first_t, first_v = s[0] + last_t, last_v = s[-1] + result = last_v - first_v + to_start = (first_t - (t_ms - w_ms)) / 1000 + to_end = (t_ms - last_t) / 1000 + sampled = (last_t - first_t) / 1000 + avg = sampled / (len(s) - 1) + if to_start >= avg * 1.1: + BRANCHES_HIT.add("start_avg") + to_start = avg / 2 + else: + BRANCHES_HIT.add("start_boundary") + if result > 0 and first_v >= 0 and sampled * first_v / result < to_start: + BRANCHES_HIT.add("zero_clamp") + to_start = sampled * first_v / result + if to_end >= avg * 1.1: + BRANCHES_HIT.add("end_avg") + to_end = avg / 2 + else: + BRANCHES_HIT.add("end_boundary") + factor = (sampled + to_start + to_end) / sampled + return result * factor / (w_ms / 1000 if is_rate else 1) + + +def range_fn(data, t_ms: int, w_ms: int, fn: Callable, min_n: int = 1) -> Vector: + out = {} + for series, samples in data.items(): + window = in_window(samples, t_ms, w_ms) + if len(window) >= min_n: + out[series] = fn(window) + return out + + +def instant(data, t_ms: int) -> Vector: + return range_fn(data, t_ms, INSTANT_MS, lambda w: w[-1][1]) + + +def rate(data, t_ms, w_ms, is_rate=True) -> Vector: + return range_fn( + data, t_ms, w_ms, lambda w: extrapolated_rate(w, t_ms, w_ms, is_rate), 2 + ) + + +def agg_by(vec: Vector, idx: List[int], fn: Callable) -> Vector: + groups: Dict[Series, List[float]] = {} + for series, v in vec.items(): + groups.setdefault(tuple(series[i] for i in idx), []).append(v) + return {k: fn(vs) for k, vs in groups.items()} + + +def topk(k: int, vec: Vector) -> Vector: + return dict(sorted(vec.items(), key=lambda kv: (-kv[1], kv[0]))[:k]) + + +def values(w: Samples) -> List[float]: + return [v for _, v in w] + + +JOB, LABEL_0 = [0], [1] +W2, W5 = 120_000, 300_000 + + +def q_ratio(t, gauge): + return { + s: prom_quantile(0.9, values(w)) / prom_quantile(0.5, values(w)) + for s, w in range_fn(gauge, t, W2, lambda w: w).items() + } + + +# (interval secs, PromQL, reference over (gauge, counter) at eval time t ms) +CASES = [ + (1, "sum by (job) (data)", lambda t, g, c: agg_by(instant(g, t), JOB, sum)), + (1, "count by (job) (data)", lambda t, g, c: agg_by(instant(g, t), JOB, len)), + *[ + ( + 1, + f"quantile by (job) ({q}, data)", + lambda t, g, c, q=q: agg_by( + instant(g, t), JOB, lambda vs: prom_quantile(q, vs) + ), + ) + for q in (0.5, 0.99) + ], + ( + 60, + "topk(4, sum by (label_0) (sum_over_time(data[2m])))", + lambda t, g, c: topk( + 4, agg_by(range_fn(g, t, W2, lambda w: sum(values(w))), LABEL_0, sum) + ), + ), + ( + 60, + "sum_over_time(data[5m])", + lambda t, g, c: range_fn(g, t, W5, lambda w: sum(values(w))), + ), + (60, "count_over_time(data[2m])", lambda t, g, c: range_fn(g, t, W2, len)), + *[ + ( + 60, + f"quantile_over_time({q}, data[2m])", + lambda t, g, c, q=q: range_fn( + g, t, W2, lambda w: prom_quantile(q, values(w)) + ), + ) + for q in (0.5, 0.9, 0.99) + ], + (60, "rate(data_total[2m])", lambda t, g, c: rate(c, t, W2)), + (60, "increase(data_total[5m])", lambda t, g, c: rate(c, t, W5, False)), + ( + 60, + "sum by (job) (rate(data_total[2m]))", + lambda t, g, c: agg_by(rate(c, t, W2), JOB, sum), + ), + ( + 60, + "sum by (job) (sum_over_time(data[2m]))", + lambda t, g, c: agg_by(range_fn(g, t, W2, lambda w: sum(values(w))), JOB, sum), + ), + ( + 60, + "sum by (job) (count_over_time(data[2m]))", + lambda t, g, c: agg_by(range_fn(g, t, W2, len), JOB, sum), + ), + ( + 60, + "quantile by (job) (0.5, rate(data_total[2m]))", + lambda t, g, c: agg_by(rate(c, t, W2), JOB, lambda vs: prom_quantile(0.5, vs)), + ), + ( + 60, + "topk(4, sum by (label_0) (rate(data_total[2m])))", + lambda t, g, c: topk(4, agg_by(rate(c, t, W2), LABEL_0, sum)), + ), + ( + 60, + "quantile_over_time(0.9, data[2m]) / quantile_over_time(0.5, data[2m])", + lambda t, g, c: q_ratio(t, g), + ), +] +EVAL_OFFSETS_SECS = { + # Before data, data start (the only sample sits on the window's end), + # mid-run, around series 2's gap and series 1's end, and after all data. + 1: [-5, 0, 1, 37, 151, 202, 203, 301, 419, 420, 421, 500], + 60: [0, 60, 120, 180, 300, 360, 420, 540], +} +DURATION_SECS = 420 + + +def make_dataset(seed: int = 7): + """Two jobs x three label_0s, ~4 samples/s at ms timestamps, one on every + whole second. Series 0 starts late with its counter at 0 (rate's zero + clamp), series 1 stops early, series 2 has a gap in (200s, 203s].""" + rng = random.Random(seed) + gauge, counter = {}, {} + for i in range(6): + series = (f"job_{i // 3}", f"label_0_{i}") + start, stop = (150, DURATION_SECS) if i == 0 else (0, DURATION_SECS) + if i == 1: + stop = 300 + total = 0.0 if i == 0 else 1e6 * i + g, c = [], [] + for k in range(start * 4, stop * 4 + 1): + t = DATA_START_MS + k * 250 + (0 if k % 4 == 0 else rng.randint(1, 249)) + if i == 2 and DATA_START_MS + 200_000 < t <= DATA_START_MS + 203_000: + continue + g.append((t, rng.paretovariate(2.0) * 10)) + c.append((t, total)) + total += rng.randint(1, 100) + gauge[series], counter[series] = g, c + return gauge, counter + + +def run_clickhouse(script: str, tmp: str) -> List[dict]: + path = os.path.join(tmp, "script.sql") + with open(path, "w") as f: + f.write(script) + out = subprocess.run( + [CLICKHOUSE, "local", "--queries-file", path], + check=True, + capture_output=True, + text=True, + ).stdout + return [json.loads(line) for line in out.splitlines()] + + +def load_script(tmp: str, data: Dict[str, Dict[Series, Samples]]) -> str: + """Init SQL, then the data inserted through it (and any views on it).""" + parts = [sbw.init_sql(LABELS)] + for table, by_series in data.items(): + csv = os.path.join(tmp, f"{table}.csv") + with open(csv, "w") as f: + for (job, label_0), samples in by_series.items(): + f.writelines(f"{t},{job},{label_0},{v!r}\n" for t, v in samples) + parts.append( + f"INSERT INTO {table} SELECT fromUnixTimestamp64Milli(ts_ms), job, " + f"label_0, value FROM file('{csv}', 'CSV', " + "'ts_ms Int64, job String, label_0 String, value Float64');\n" + ) + return "".join(parts) + + +def query_script(statements: List[Tuple[str, str]]) -> str: + """Each (id, sql) as JSON rows tagged with the id.""" + return "".join( + f"SELECT '{qid}' AS qid, * FROM ({sql}) FORMAT JSONEachRow;\n" + for qid, sql in statements + ) + + +def rows_by_id(rows: List[dict], labels: Dict[str, Tuple[str, ...]]) -> dict: + out: Dict[str, Vector] = {qid: {} for qid in labels} + for row in rows: + qid = row["qid"] + out[qid][tuple(row[name] for name in labels[qid])] = row["value"] + return out + + +def assert_vectors_close(test, got: Vector, want: Vector, rel: float) -> None: + test.assertEqual(sorted(got), sorted(want)) + for key, value in want.items(): + test.assertTrue( + math.isclose(got[key], value, rel_tol=rel), (key, got[key], value) + ) + + +@unittest.skipUnless(CLICKHOUSE, "clickhouse binary not found") +class ClickHouseSemanticsTest(unittest.TestCase): + def test_raw_sql_matches_prometheus_reference(self): + gauge, counter = make_dataset() + queries = [(t, p, sbw.parse_promql(p, t, LABELS)) for t, p, _ in CASES] + statements, groupings = [], {} + for i, (interval, _, q) in enumerate(queries): + for offset in EVAL_OFFSETS_SECS[interval]: + end = DATA_START_MS // 1000 + offset + qid = f"{i}@{offset}" + sql = sbw.raw_sql(q, LABELS).replace(sbw.END, str(end)) + statements.append((qid, sql)) + groupings[qid] = q.grouping + with tempfile.TemporaryDirectory() as tmp: + script = load_script(tmp, {"data": gauge, "data_total": counter}) + got = rows_by_id( + run_clickhouse(script + query_script(statements), tmp), groupings + ) + + BRANCHES_HIT.clear() + nonempty = 0 + for i, (interval, promql, ref) in enumerate(CASES): + for offset in EVAL_OFFSETS_SECS[interval]: + with self.subTest(promql=promql, offset=offset): + want = ref(DATA_START_MS + offset * 1000, gauge, counter) + nonempty += bool(want) + assert_vectors_close(self, got[f"{i}@{offset}"], want, 1e-9) + self.assertEqual( + BRANCHES_HIT, + {"start_avg", "start_boundary", "zero_clamp", "end_avg", "end_boundary"}, + ) + # Both empty and non-empty windows are compared. + self.assertGreater(nonempty, len(statements) // 2) + self.assertLess(nonempty, len(statements)) + + +if __name__ == "__main__": + unittest.main() From 7107a85f1776ed391aa0d12dd58078e50ff1f7ad Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:23:58 +0000 Subject: [PATCH 04/10] fix(asap-tools): use Prometheus 2.53's closed windows in sketch-bench SQL Experiments run Prometheus 2.53.2 (installation/prometheus/install.sh), whose range selectors and instant lookback are closed, [T - d, T]; left-open windows arrived in 3.0. Aligned eval times put a sample on the left boundary, so the SQL now reads ts >= T - d. The Python reference selects [T - d, T] as well. generate_workloads.py's generated2 SQL is unchanged; only the shared rate_from_samples docstring no longer names a window shape. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../fake_metric/generate_workloads.py | 4 ++-- .../fake_metric/sketch_bench_workloads.py | 22 +++++++++---------- .../tests/test_sketch_bench_workloads.py | 3 ++- 3 files changed, 15 insertions(+), 14 deletions(-) diff --git a/asap-tools/experiments/datasets/fake_metric/generate_workloads.py b/asap-tools/experiments/datasets/fake_metric/generate_workloads.py index 72cc0f39..5b914b8c 100644 --- a/asap-tools/experiments/datasets/fake_metric/generate_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/generate_workloads.py @@ -78,8 +78,8 @@ def rate_from_samples( cols: str, samples: str, end: str, window: int, is_rate: bool ) -> str: """extrapolatedRate over `samples`, a query yielding per series `cols`, - first_v, last_v, first_t, last_t (unix seconds) and n >= 2 samples in - (end - window, end].""" + first_v, last_v, first_t, last_t (unix seconds) and n >= 2 samples in the + window of `window` seconds ending at `end`.""" durations = ( f"SELECT {cols}, last_v - first_v AS delta, first_v, " f"last_t - first_t AS sampled, sampled / (n - 1) AS avg_step, " diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py index 7dbbc917..52bf40e3 100644 --- a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -12,10 +12,12 @@ - _clickhouse.yaml: a baseline-only experiment_run_clickhouse.py config with one query group per T, repeating every T seconds. -Semantics are generate_workloads.py's (Prometheus 2.53, left-open windows): -range vectors [d] at T read (T - d, T]; instant vectors read the latest sample -of each series in (T - 1s, T]; quantiles use quantileExactInclusive; rate and -increase reproduce extrapolatedRate (counters never reset). T is the client's +Semantics are those of Prometheus 2.53 (the version installation/prometheus +installs for bare-metal runs), whose windows are closed on both ends: range +vectors [d] at T read [T - d, T]; instant vectors read the latest sample of +each series in [T - 1s, T] (a lookback of one scrape interval); quantiles use +quantileExactInclusive; rate and increase reproduce extrapolatedRate (counters +never reset). T is the client's {{ end_time }} floored to a multiple of the query's interval. Unsupported query shapes raise. @@ -177,9 +179,10 @@ def eval_time(interval_secs: int) -> str: def render( - q: Query, series: Tuple[str, ...], source: str, time_col: str, agg: Callable + q: Query, series: Tuple[str, ...], source: str, where: Callable, agg: Callable ) -> str: - """SQL for `q` reading `source` rows with `time_col` in the window. + """SQL for `q` reading the `source` rows that `where(end, window_secs)` + keeps for the window [end - window_secs, end]. `series` is the table's full label list; `agg(stat, quantile)` gives the aggregate expression of a statistic over the source rows: sum, count, @@ -187,10 +190,7 @@ def render( rate_from_samples reads). """ end = eval_time(q.interval_secs) - window = ( - f"FROM {source} WHERE {time_col} > {end} - {q.window_secs} " - f"AND {time_col} <= {end}" - ) + window = f"FROM {source} WHERE {where(end, q.window_secs)}" group = ", ".join(q.grouping) cols = ", ".join(series) if q.series_fn is None: @@ -241,7 +241,7 @@ def raw_sql(q: Query, labels: Dict[str, List[str]]) -> str: q, tuple(labels[q.table]), q.table, - "ts", + lambda end, w: f"ts >= {end} - {w} AND ts <= {end}", lambda stat, quantile: raw_agg(stat, quantile, q.value), ) diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py index 79c7f086..6cb25a96 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_workloads.py +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -151,7 +151,8 @@ def test_writes_config_grouped_by_interval(self): def in_window(samples: Samples, t_ms: int, w_ms: int) -> Samples: - return [(t, v) for t, v in samples if t_ms - w_ms < t <= t_ms] + # Prometheus 2.x selects [t - w, t]; 3.x made it left-open (t - w, t]. + return [(t, v) for t, v in samples if t_ms - w_ms <= t <= t_ms] def prom_quantile(q: float, values: List[float]) -> float: From 392ecf5103c3ac986a88d24f1a04f3187c507587 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:31:01 +0000 Subject: [PATCH 05/10] feat(asap-tools): multi-grouping sketch-bench queries in ClickHouse SQL Structured e2e entries {table, value, capability, grouping, quantile, window_secs, interval_secs} become pooled queries over a records table. Cardinality is uniqExact of the value field per group, and quantile is quantileExactInclusive over all of a group's samples. Both read the same closed [T - W, T] window, with T floored to the interval as the PromQL queries are. init.sql now creates records tables, with one Float64 column per value field. Each table's structured queries are also written in ASAPQuery's SQL frontend dialect (COUNT(DISTINCT col), QUANTILE(q, col), BETWEEN DATEADD(s, -W, NOW()) AND NOW()) to asapquery.sql. The clickhouse-local test adds a skewed records dataset in which endpoint is a child of service. It checks template 16's CUBE over (region, service, endpoint) and several cardinality groupings against a Python reference. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../fake_metric/sketch_bench_workloads.py | 128 +++++++--- .../tests/test_sketch_bench_workloads.py | 225 +++++++++++++++--- 2 files changed, 297 insertions(+), 56 deletions(-) diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py index 52bf40e3..54cd8220 100644 --- a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -1,14 +1,17 @@ #!/usr/bin/env python3 """Translate a sketch-bench table's end-to-end queries into ClickHouse configs. -A sketch-bench table's `workloads[0].e2e` block lists its queries as -`{promql, interval_secs, rqes}` and describes each metric (`metrics`). Every -metric is one ClickHouse table named after it: `ts DateTime64(3)`, then the -metric's labels in description order as String, then `value Float64`. +A sketch-bench table's `workloads[0].e2e` block lists its queries, either +`{promql, interval_secs, rqes}` or structured multi-grouping queries, and +describes each metric (`metrics`). Every metric is one ClickHouse table named +after it: `ts DateTime64(3)`, then the metric's labels in description order as +String, then `value Float64`, or for a `records` metric one Float64 column per +value field in description order. For a table `.json` this writes, under OUT_CONFIG_DIR: - /init.sql: the metric tables; - /interval_s.sql: one statement per query that repeats every T s; +- /asapquery.sql: the structured queries in ASAPQuery's SQL dialect; - _clickhouse.yaml: a baseline-only experiment_run_clickhouse.py config with one query group per T, repeating every T seconds. @@ -17,9 +20,10 @@ vectors [d] at T read [T - d, T]; instant vectors read the latest sample of each series in [T - 1s, T] (a lookback of one scrape interval); quantiles use quantileExactInclusive; rate and increase reproduce extrapolatedRate (counters -never reset). T is the client's -{{ end_time }} floored to a multiple of the query's interval. Unsupported -query shapes raise. +never reset). Structured queries read the same [T - W, T] and pool all samples +of a group: cardinality is uniqExact of the value field, quantile is +quantileExactInclusive. T is the client's {{ end_time }} floored to a multiple +of the query's interval. Unsupported query shapes raise. Run from asap-tools/experiments: python3 datasets/fake_metric/sketch_bench_workloads.py \ @@ -49,6 +53,9 @@ # Instant vectors read the latest sample of each series in the last scrape. INSTANT_WINDOW_SECONDS = SCRAPE_INTERVAL_SECONDS +# Structured entries' capabilities, all pooled over a group's samples. +STRUCTURED_AGGS = ("cardinality", "quantile") + LABELS = r"(\w+(?:, \w+)*)" LEVEL = r"([\d.]+)" TOPK = re.compile(r"^topk\((\d+), (.+)\)$") @@ -165,12 +172,42 @@ def parse_promql(promql: str, interval: int, labels: Dict[str, List[str]]) -> Qu return _parse_series(text, interval, labels, promql) -def query_from_entry(entry: dict, labels: Dict[str, List[str]]) -> Query: - """One `e2e.queries` entry. Entries are PromQL; structured entries (no - `promql`) are not supported yet and raise.""" - if "promql" not in entry: - raise ValueError(f"query entry without promql is not supported: {entry}") - return parse_promql(entry["promql"], entry["interval_secs"], labels) +def query_from_entry(entry: dict, metrics: dict) -> Query: + """One `e2e.queries` entry: PromQL, or a structured multi-grouping query + `{table, value, capability, grouping, quantile, window_secs, + interval_secs}` that pools every sample of a group over the window.""" + if "promql" in entry: + return parse_promql( + entry["promql"], entry["interval_secs"], metric_labels(metrics) + ) + table, capability = entry["table"], entry["capability"] + if capability not in STRUCTURED_AGGS: + raise ValueError(f"unsupported capability {capability!r}: {entry}") + labels = metric_labels(metrics)[table] + unknown = set(entry["grouping"]) - set(labels) + if unknown or entry["value"] not in value_columns(metrics[table]): + raise ValueError(f"unknown labels or value for {table!r}: {entry}") + return Query( + table, + entry["interval_secs"], + entry["window_secs"], + tuple(entry["grouping"]), + None, + capability, + str(entry["quantile"]) if capability == "quantile" else None, + value=entry["value"], + ) + + +def describe(entry: dict) -> str: + """A one-line comment naming the query.""" + if "promql" in entry: + return entry["promql"] + level = f"{entry['quantile']}, " if entry["capability"] == "quantile" else "" + return ( + f"{entry['capability']}({level}{entry['value']}) by " + f"({', '.join(entry['grouping'])}) over {entry['window_secs']}s" + ) def eval_time(interval_secs: int) -> str: @@ -223,6 +260,8 @@ def raw_agg(stat: str, quantile: Optional[str], value: str = VALUE_COLUMN) -> st return f"sum({value})" if stat == "count": return "count()" + if stat == "cardinality": + return f"uniqExact({value})" if stat == "quantile": return f"quantileExactInclusive({quantile})({value})" if stat == "last": @@ -246,54 +285,89 @@ def raw_sql(q: Query, labels: Dict[str, List[str]]) -> str: ) -def init_sql(labels: Dict[str, List[str]]) -> str: +def asapquery_sql(q: Query) -> str: + """A structured query in the dialect of ASAPQuery's SQL frontend. Its + window ends at the engine's NOW(); it cannot floor the eval time.""" + 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}") + return ( + f"SELECT {group}, {agg} FROM {q.table} WHERE ts BETWEEN " + f"DATEADD(s, -{q.window_secs}, NOW()) AND NOW() GROUP BY {group}" + ) + + +def value_columns(desc: dict) -> List[str]: + """A records metric has one column per value field; a gauge or counter + has one value column.""" + return list(desc["values"]) if desc["type"] == "records" else [VALUE_COLUMN] + + +def init_sql(metrics: dict) -> str: statements = [] - for table, names in labels.items(): - columns = ", ".join(f"{name} String" for name in names) + for table, desc in metrics.items(): + names = [label["name"] for label in desc["labels"]] + columns = [f"{name} String" for name in names] + [ + f"{value} Float64" for value in value_columns(desc) + ] statements += [ f"DROP TABLE IF EXISTS {table}", - f"CREATE TABLE {table} (ts DateTime64(3), {columns}, " - f"{VALUE_COLUMN} Float64) ENGINE = MergeTree " - f"ORDER BY ({', '.join(names)}, ts)", + f"CREATE TABLE {table} (ts DateTime64(3), {', '.join(columns)}) " + f"ENGINE = MergeTree ORDER BY ({', '.join(names)}, ts)", ] return ";\n\n".join(statements) + ";\n" -def metric_labels(e2e: dict) -> Dict[str, List[str]]: +def metric_labels(metrics: dict) -> Dict[str, List[str]]: return { name: [label["name"] for label in desc["labels"]] - for name, desc in e2e["metrics"].items() + for name, desc in metrics.items() } +def write_statements(path: str, statements: List[str]) -> None: + with open(path, "w") as f: + f.write(";\n\n".join(statements) + ";\n") + + def convert( table_path: str, data_file: str, starting_delay_secs: int, duration_secs: int ) -> str: with open(table_path) as f: (workload,) = json.load(f)["workloads"] e2e = workload["e2e"] - labels = metric_labels(e2e) + metrics = e2e["metrics"] + labels = metric_labels(metrics) stem = os.path.splitext(os.path.basename(table_path))[0] sql_dir = os.path.join(OUT_CONFIG_DIR, stem) os.makedirs(sql_dir, exist_ok=True) init_path = os.path.join(sql_dir, "init.sql") with open(init_path, "w") as f: - f.write(init_sql(labels)) + f.write(init_sql(metrics)) by_interval: Dict[int, List[str]] = {} + asapquery: List[str] = [] for entry in e2e["queries"]: - q = query_from_entry(entry, labels) + q = query_from_entry(entry, metrics) by_interval.setdefault(q.interval_secs, []).append( - f"-- {entry['promql']}\n{raw_sql(q, labels)}" + f"-- {describe(entry)}\n{raw_sql(q, labels)}" ) + if "promql" not in entry: + asapquery.append(f"-- {describe(entry)}\n{asapquery_sql(q)}") + if asapquery: + # For the ASAPQuery arm; the runner does not read it. + write_statements(os.path.join(sql_dir, "asapquery.sql"), asapquery) query_groups = [] for interval, statements in sorted(by_interval.items()): name = f"interval_{interval}s" sql_path = os.path.join(sql_dir, f"{name}.sql") - with open(sql_path, "w") as f: - f.write(";\n\n".join(statements) + ";\n") + write_statements(sql_path, statements) query_groups.append( { "name": name, diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py index 6cb25a96..7206b863 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_workloads.py +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -6,6 +6,7 @@ CLICKHOUSE_BINARY) and is skipped otherwise. """ +import itertools import json import math import os @@ -25,7 +26,23 @@ import sketch_bench_workloads as sbw # noqa: E402 -LABELS = {"data": ["job", "label_0"], "data_total": ["job", "label_0"]} +SERIES_LABELS = [{"name": "job"}, {"name": "label_0", "child_of": "job"}] +HTTP_LABELS = ["region", "service", "endpoint", "status"] +METRICS = { + "data": {"type": "gauge", "labels": SERIES_LABELS}, + "data_total": {"type": "counter", "labels": SERIES_LABELS}, + "http": { + "type": "records", + "labels": [ + {"name": "region"}, + {"name": "service"}, + {"name": "endpoint", "child_of": "service"}, + {"name": "status"}, + ], + "values": {"latency": {}, "user_id": {}}, + }, +} +LABELS = sbw.metric_labels(METRICS) CLICKHOUSE = os.environ.get("CLICKHOUSE_BINARY") or shutil.which("clickhouse") # The PromQL of sketch-bench's ten classic templates (windows vary per table). @@ -95,22 +112,69 @@ def test_unsupported_shapes_raise(self): with self.assertRaises(ValueError): sbw.parse_promql(promql, 60, LABELS) - def test_structured_entry_raises(self): - entry = {"table": "http", "capability": "quantile", "interval_secs": 60} - with self.assertRaises(ValueError): - sbw.query_from_entry(entry, LABELS) + def test_structured_entries(self): + cardinality = mg_entry("cardinality", ["region", "service"], 300) + quantile = mg_entry("quantile", ["service"], 900) + self.assertEqual( + sbw.query_from_entry(cardinality, METRICS), + sbw.Query( + "http", + 60, + 300, + ("region", "service"), + None, + "cardinality", + value="user_id", + ), + ) + self.assertEqual( + sbw.query_from_entry(quantile, METRICS), + sbw.Query( + "http", + 60, + 900, + ("service",), + None, + "quantile", + "0.99", + value="latency", + ), + ) + + def test_unsupported_structured_entries_raise(self): + for entry in [ + {**mg_entry("cardinality", ["region"], 300), "capability": "sum"}, + mg_entry("cardinality", ["instance"], 300), + {**mg_entry("quantile", ["region"], 300), "value": "bytes"}, + ]: + with self.subTest(entry=entry), self.assertRaises(ValueError): + sbw.query_from_entry(entry, METRICS) + + +def mg_entry(capability: str, grouping: List[str], window_secs: int) -> dict: + """A structured multi-grouping entry over `http`, as sketch-bench writes it.""" + entry = { + "table": "http", + "value": "latency" if capability == "quantile" else "user_id", + "capability": capability, + "grouping": grouping, + "window_secs": window_secs, + "interval_secs": 60, + "rqes": [], + } + if capability == "quantile": + entry["quantile"] = 0.99 + return entry class ConvertTest(unittest.TestCase): def test_writes_config_grouped_by_interval(self): e2e = { - "metrics": { - name: {"labels": [{"name": "job"}, {"name": "label_0"}]} - for name in LABELS - }, + "metrics": METRICS, "queries": [ {"promql": p, "interval_secs": t, "rqes": []} for t, p in CLASSIC - ], + ] + + [mg_entry("quantile", ["service"], 300)], } with tempfile.TemporaryDirectory() as tmp: table = os.path.join(tmp, "classic.json") @@ -133,9 +197,23 @@ def test_writes_config_grouped_by_interval(self): ) with open(g60["sql_file"]["baseline"]) as f: statements = [s for s in f.read().split(";") if s.strip()] - self.assertEqual(len(statements), 8) + self.assertEqual(len(statements), 9) with open(cfg["dataset"]["init_sql_file"]) as f: - self.assertIn("CREATE TABLE data_total (ts DateTime64(3)", f.read()) + init = f.read() + self.assertIn("CREATE TABLE data_total (ts DateTime64(3)", init) + self.assertIn( + "endpoint String, status String, latency Float64, " + "user_id Float64)", + init, + ) + sql_dir = os.path.dirname(g60["sql_file"]["baseline"]) + 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, -300, NOW()) AND NOW() GROUP BY service;\n", + ) finally: os.chdir(cwd) @@ -338,6 +416,65 @@ def make_dataset(seed: int = 7): return gauge, counter +def make_records(seed: int = 11) -> List[tuple]: + """`http` records every 50 ms (one on every whole second) with skewed + labels. Endpoint names repeat across services (endpoint is a child of + service), so grouping by endpoint alone pools several services.""" + rng = random.Random(seed) + records = [] + for k in range(DURATION_SECS * 20 + 1): + t = DATA_START_MS + k * 50 + (0 if k % 20 == 0 else rng.randint(0, 49)) + service = rng.choices(["s0", "s1", "s2", "s3"], [8, 4, 2, 1])[0] + labels = ( + rng.choices(["r0", "r1", "r2"], [6, 3, 1])[0], + service, + rng.choices(["e0", "e1", "e2"], [5, 2, 1])[0], + rng.choices(["200", "404", "500"], [20, 3, 1])[0], + ) + latency = rng.paretovariate(2.0) * (1 + "s0s1s2s3".index(service)) + records.append((t, labels, (latency, float(int(rng.paretovariate(1.2) * 10))))) + return records + + +def series_rows(by_series: Dict[Series, Samples]) -> List[tuple]: + return [(t, s, (v,)) for s, samples in by_series.items() for t, v in samples] + + +def pooled(records, t_ms: int, w_ms: int, q: "sbw.Query") -> Vector: + """A structured query's reference: every record of a group in the window.""" + idx = [HTTP_LABELS.index(name) for name in q.grouping] + col = list(METRICS["http"]["values"]).index(q.value) + groups: Dict[Series, List[float]] = {} + for t, labels, values in records: + if t_ms - w_ms <= t <= t_ms: + groups.setdefault(tuple(labels[i] for i in idx), []).append(values[col]) + if q.agg == "cardinality": + return {k: len(set(vs)) for k, vs in groups.items()} + return {k: prom_quantile(float(q.quantile), vs) for k, vs in groups.items()} + + +# Template 16's CUBE over (region, service, endpoint) and template 14/15-like +# cardinality groupings, including status and the full label set. +MULTIGROUP = [ + mg_entry("quantile", list(g), w) + for w in (120, 300) + for n in (1, 2, 3) + for g in itertools.combinations(["region", "service", "endpoint"], n) +] + [ + mg_entry("cardinality", g, w) + for w in (120, 300) + for g in [ + ["region"], + ["service"], + ["endpoint"], + ["status"], + ["region", "service"], + ["service", "endpoint"], + ["region", "service", "endpoint", "status"], + ] +] + + def run_clickhouse(script: str, tmp: str) -> List[dict]: path = os.path.join(tmp, "script.sql") with open(path, "w") as f: @@ -351,18 +488,27 @@ def run_clickhouse(script: str, tmp: str) -> List[dict]: return [json.loads(line) for line in out.splitlines()] -def load_script(tmp: str, data: Dict[str, Dict[Series, Samples]]) -> str: - """Init SQL, then the data inserted through it (and any views on it).""" - parts = [sbw.init_sql(LABELS)] - for table, by_series in data.items(): +def load_script(tmp: str, rows: Dict[str, List[tuple]]) -> str: + """Init SQL, then each table's (ts ms, labels, values) rows inserted + through it (and any views on it).""" + parts = [sbw.init_sql(METRICS)] + for table, table_rows in rows.items(): + names = LABELS[table] + values = sbw.value_columns(METRICS[table]) csv = os.path.join(tmp, f"{table}.csv") with open(csv, "w") as f: - for (job, label_0), samples in by_series.items(): - f.writelines(f"{t},{job},{label_0},{v!r}\n" for t, v in samples) + f.writelines( + ",".join([str(t), *labels, *map(repr, vs)]) + "\n" + for t, labels, vs in table_rows + ) + structure = ", ".join( + ["ts_ms Int64"] + + [f"{n} String" for n in names] + + [f"{v} Float64" for v in values] + ) parts.append( - f"INSERT INTO {table} SELECT fromUnixTimestamp64Milli(ts_ms), job, " - f"label_0, value FROM file('{csv}', 'CSV', " - "'ts_ms Int64, job String, label_0 String, value Float64');\n" + f"INSERT INTO {table} SELECT fromUnixTimestamp64Milli(ts_ms), " + f"{', '.join(names + values)} FROM file('{csv}', 'CSV', '{structure}');\n" ) return "".join(parts) @@ -391,31 +537,52 @@ def assert_vectors_close(test, got: Vector, want: Vector, rel: float) -> None: ) +def all_cases(gauge, counter, records) -> List[Tuple[str, "sbw.Query", Callable]]: + """(name, query, reference at eval time t ms) for CASES and MULTIGROUP.""" + cases = [ + (p, sbw.parse_promql(p, t, LABELS), lambda e, ref=ref: ref(e, gauge, counter)) + for t, p, ref in CASES + ] + for entry in MULTIGROUP: + q = sbw.query_from_entry(entry, METRICS) + w_ms = q.window_secs * 1000 + cases.append( + (sbw.describe(entry), q, lambda e, q=q, w=w_ms: pooled(records, e, w, q)) + ) + return cases + + @unittest.skipUnless(CLICKHOUSE, "clickhouse binary not found") class ClickHouseSemanticsTest(unittest.TestCase): def test_raw_sql_matches_prometheus_reference(self): gauge, counter = make_dataset() - queries = [(t, p, sbw.parse_promql(p, t, LABELS)) for t, p, _ in CASES] + records = make_records() + cases = all_cases(gauge, counter, records) statements, groupings = [], {} - for i, (interval, _, q) in enumerate(queries): - for offset in EVAL_OFFSETS_SECS[interval]: + for i, (_, q, _) in enumerate(cases): + for offset in EVAL_OFFSETS_SECS[q.interval_secs]: end = DATA_START_MS // 1000 + offset qid = f"{i}@{offset}" sql = sbw.raw_sql(q, LABELS).replace(sbw.END, str(end)) statements.append((qid, sql)) groupings[qid] = q.grouping + rows = { + "data": series_rows(gauge), + "data_total": series_rows(counter), + "http": records, + } with tempfile.TemporaryDirectory() as tmp: - script = load_script(tmp, {"data": gauge, "data_total": counter}) + script = load_script(tmp, rows) got = rows_by_id( run_clickhouse(script + query_script(statements), tmp), groupings ) BRANCHES_HIT.clear() nonempty = 0 - for i, (interval, promql, ref) in enumerate(CASES): - for offset in EVAL_OFFSETS_SECS[interval]: - with self.subTest(promql=promql, offset=offset): - want = ref(DATA_START_MS + offset * 1000, gauge, counter) + for i, (name, q, ref) in enumerate(cases): + for offset in EVAL_OFFSETS_SECS[q.interval_secs]: + with self.subTest(query=name, offset=offset): + want = ref(DATA_START_MS + offset * 1000) nonempty += bool(want) assert_vectors_close(self, got[f"{i}@{offset}"], want, 1e-9) self.assertEqual( From 28cc8416a8e20510f4f7f01aebbb9cbfc0d05d0b Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 01:44:24 +0000 Subject: [PATCH 06/10] feat(asap-tools): ASAPQuery's left-open window for multigroup queries Multigroup (structured) queries involve no Prometheus, so they now read ASAPQuery's left-open window (E - W, E] instead of the closed [E - W, E] that classic PromQL keeps to match Prometheus 2.53. ASAPQuery's SQL frontend and precompute windows are (start, end], so ClickHouse and ASAPQuery compute over the same samples. Query gets a `left_open` flag that structured entries set; the raw SQL uses `ts > E - W`. The ASAPQuery-dialect file ends its window at the floored eval time instead of NOW(). The frontend cannot floor a literal, so the client's {{ end_datetime }} is moved back by {{ end_time % T }} inside DATEADD: ts BETWEEN DATEADD(s, -{{ end_time % T + W }}, '{{ end_datetime }}') AND DATEADD(s, -{{ end_time % T }}, '{{ end_datetime }}') which the frontend parses to (start, duration) = (E - W, W). The Python reference for multigroup now pools records in (t - w, t]. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../fake_metric/sketch_bench_workloads.py | 36 +++++++++++++------ .../tests/test_sketch_bench_workloads.py | 11 ++++-- 2 files changed, 34 insertions(+), 13 deletions(-) diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py index 54cd8220..da2b87a1 100644 --- a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -20,10 +20,12 @@ vectors [d] at T read [T - d, T]; instant vectors read the latest sample of each series in [T - 1s, T] (a lookback of one scrape interval); quantiles use quantileExactInclusive; rate and increase reproduce extrapolatedRate (counters -never reset). Structured queries read the same [T - W, T] and pool all samples -of a group: cardinality is uniqExact of the value field, quantile is -quantileExactInclusive. T is the client's {{ end_time }} floored to a multiple -of the query's interval. Unsupported query shapes raise. +never reset). Structured queries involve no Prometheus: they read ASAPQuery's +left-open window (T - W, T], the window its SQL frontend and precompute +windows use, and pool all samples of a group: cardinality is uniqExact of +the value field, quantile is quantileExactInclusive. T is the client's +{{ end_time }} floored to a multiple of the query's interval. Unsupported +query shapes raise. Run from asap-tools/experiments: python3 datasets/fake_metric/sketch_bench_workloads.py \ @@ -79,7 +81,8 @@ class Query: Otherwise `series_fn` (last, rate, increase) gives one value per series, and `agg` combines those per group; `agg` None keeps them per series. `ratio_quantile` divides the `quantile` by a second quantile of the same - samples. Structured (non-PromQL) query entries map onto these fields. + samples. Structured (non-PromQL) query entries map onto these fields and + set `left_open`: their window is (end - W, end], not PromQL's closed one. """ table: str @@ -92,6 +95,7 @@ class Query: ratio_quantile: Optional[str] = None topk: Optional[int] = None value: str = VALUE_COLUMN + left_open: bool = False def _parse_series( @@ -196,6 +200,7 @@ def query_from_entry(entry: dict, metrics: dict) -> Query: capability, str(entry["quantile"]) if capability == "quantile" else None, value=entry["value"], + left_open=True, ) @@ -219,7 +224,8 @@ def render( q: Query, series: Tuple[str, ...], source: str, where: Callable, agg: Callable ) -> str: """SQL for `q` reading the `source` rows that `where(end, window_secs)` - keeps for the window [end - window_secs, end]. + keeps for the window [end - window_secs, end], or (end - window_secs, end] + when `q.left_open`. `series` is the table's full label list; `agg(stat, quantile)` gives the aggregate expression of a statistic over the source rows: sum, count, @@ -280,14 +286,21 @@ def raw_sql(q: Query, labels: Dict[str, List[str]]) -> str: q, tuple(labels[q.table]), q.table, - lambda end, w: f"ts >= {end} - {w} AND ts <= {end}", + lambda end, w: f"ts {'>' if q.left_open else '>='} {end} - {w} AND ts <= {end}", lambda stat, quantile: raw_agg(stat, quantile, q.value), ) def asapquery_sql(q: Query) -> str: - """A structured query in the dialect of ASAPQuery's SQL frontend. Its - window ends at the engine's NOW(); it cannot floor the eval time.""" + """A structured query in the dialect of ASAPQuery's SQL frontend. + + 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. + """ group = ", ".join(q.grouping) if q.agg == "cardinality": agg = f"COUNT(DISTINCT {q.value})" @@ -295,9 +308,12 @@ def asapquery_sql(q: Query) -> str: 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, -{q.window_secs}, NOW()) AND NOW() GROUP BY {group}" + f"DATEADD(s, -{{{{ end_time % {t} + {q.window_secs} }}}}, {end}) AND " + f"DATEADD(s, -{{{{ end_time % {t} }}}}, {end}) GROUP BY {group}" ) diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py index 7206b863..e831bf10 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_workloads.py +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -125,6 +125,7 @@ def test_structured_entries(self): None, "cardinality", value="user_id", + left_open=True, ), ) self.assertEqual( @@ -138,6 +139,7 @@ def test_structured_entries(self): "quantile", "0.99", value="latency", + left_open=True, ), ) @@ -212,7 +214,9 @@ def test_writes_config_grouped_by_interval(self): f.read(), "-- quantile(0.99, latency) by (service) over 300s\n" "SELECT service, QUANTILE(0.99, latency) FROM http WHERE ts " - "BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY service;\n", + "BETWEEN DATEADD(s, -{{ end_time % 60 + 300 }}, " + "'{{ end_datetime }}') AND DATEADD(s, -{{ end_time % 60 }}, " + "'{{ end_datetime }}') GROUP BY service;\n", ) finally: os.chdir(cwd) @@ -441,12 +445,13 @@ def series_rows(by_series: Dict[Series, Samples]) -> List[tuple]: def pooled(records, t_ms: int, w_ms: int, q: "sbw.Query") -> Vector: - """A structured query's reference: every record of a group in the window.""" + """A structured query's reference: every record of a group in ASAPQuery's + left-open window (t - w, t].""" idx = [HTTP_LABELS.index(name) for name in q.grouping] col = list(METRICS["http"]["values"]).index(q.value) groups: Dict[Series, List[float]] = {} for t, labels, values in records: - if t_ms - w_ms <= t <= t_ms: + if t_ms - w_ms < t <= t_ms: groups.setdefault(tuple(labels[i] for i in idx), []).append(values[col]) if q.agg == "cardinality": return {k: len(set(vs)) for k, vs in groups.items()} From 24310d61b19e218a768b2cdfcfd1dc9321a2d681 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 02:01:45 +0000 Subject: [PATCH 07/10] feat(asap-tools): request aligned eval times in sketch-bench ClickHouse configs The SQL floors {{ end_time }} to T, but the current query client sleeps T after each repetition's queries finish, so its eval times drift by the query latency and the arms answer different windows. Each query group now sets `align_to_interval: true` and `query_time_offset: 10` (PR #841's default). With #841's client, repetition k is evaluated at floor((start - O) / T) * T + k * T and sent O seconds later. The current client ignores `align_to_interval` (unknown client options become unused attributes) and shifts its eval time back by the offset. The module docstring records the drift until #841's client lands. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../fake_metric/sketch_bench_workloads.py | 16 ++++++++++++++++ .../tests/test_sketch_bench_workloads.py | 8 +++++++- 2 files changed, 23 insertions(+), 1 deletion(-) diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py index da2b87a1..3f483bfd 100644 --- a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -27,6 +27,16 @@ {{ end_time }} floored to a multiple of the query's interval. Unsupported query shapes raise. +Evaluation times: each query group sets `align_to_interval` and a +`query_time_offset` of QUERY_TIME_OFFSET_SECS. The query client of PR #841 +then evaluates repetition k at floor((start - O) / T) * T + k * T and sends +it O seconds later, so every run and arm answers the same windows. The +current client ignores `align_to_interval`: it subtracts O from the eval +time and sleeps T after each repetition's queries finish, so its eval times +drift by the query latency. A 1 s group whose queries take 0.3 s skips +about one T in four, and a 60 s group occasionally repeats or skips a T; +arms then answer different windows. + Run from asap-tools/experiments: python3 datasets/fake_metric/sketch_bench_workloads.py \ --table --data-file \ @@ -51,6 +61,9 @@ OUT_CONFIG_DIR = "config/experiment_type/sketch_bench_clickhouse" CLICKHOUSE_URL = "http://localhost:8123" +# Seconds after an eval time the client sends it (PR #841's default), so +# data up to the eval time has arrived. +QUERY_TIME_OFFSET_SECS = 10 VALUE_COLUMN = "value" # Instant vectors read the latest sample of each series in the last scrape. INSTANT_WINDOW_SECONDS = SCRAPE_INTERVAL_SECONDS @@ -394,6 +407,9 @@ def convert( "client_options": { "starting_delay": starting_delay_secs, "repetitions": duration_secs // interval, + # Takes effect with PR #841's query client; see above. + "align_to_interval": True, + "query_time_offset": QUERY_TIME_OFFSET_SECS, }, } ) diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py index e831bf10..68208f29 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_workloads.py +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -192,7 +192,13 @@ def test_writes_config_grouped_by_interval(self): g60 = groups["interval_60s"] self.assertEqual(g60["repetition_delay_ms"], 60000) self.assertEqual( - g60["client_options"], {"starting_delay": 30, "repetitions": 10} + g60["client_options"], + { + "starting_delay": 30, + "repetitions": 10, + "align_to_interval": True, + "query_time_offset": 10, + }, ) self.assertEqual( groups["interval_1s"]["client_options"]["repetitions"], 600 From c5d98e3941ea28738ac5804db44b79552baef574 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 03:10:20 +0000 Subject: [PATCH 08/10] fix(asap-tools): compare ts with DateTime64 bounds in sketch-bench raw SQL ClickHouse 26.10 turns `ts > ` on a DateTime64 key column into `ts in [ + 1, +Inf)` for index pruning, so the left-open multigroup SQL dropped the samples in the window's first second from the result (A6: raw vs MV relative differences of 1e-6 to 2.6e-3 on 49 of 52 queries). The raw SQL now compares ts with toDateTime64(, 3) on both ends, for classic and multigroup queries. A regression test runs a left-open and a closed query in clickhouse-local over a table with 8-row granules and samples every 7 ms around the window start. Without the cast it reproduces the loss (8435 of 8571 rows). Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../fake_metric/sketch_bench_workloads.py | 8 +++- .../tests/test_sketch_bench_workloads.py | 38 +++++++++++++++++++ 2 files changed, 45 insertions(+), 1 deletion(-) diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py index 3f483bfd..e2481fde 100644 --- a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -299,7 +299,13 @@ def raw_sql(q: Query, labels: Dict[str, List[str]]) -> str: q, tuple(labels[q.table]), q.table, - lambda end, w: f"ts {'>' if q.left_open else '>='} {end} - {w} AND ts <= {end}", + # DateTime64 bounds: ClickHouse 26.10 prunes `ts > ` on the + # DateTime64 key as `ts >= + 1`, dropping the samples in + # between from the result. + lambda end, w: ( + f"ts {'>' if q.left_open else '>='} toDateTime64({end} - {w}, 3) " + f"AND ts <= toDateTime64({end}, 3)" + ), lambda stat, quantile: raw_agg(stat, quantile, q.value), ) diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py index 68208f29..39ea6839 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_workloads.py +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -563,6 +563,44 @@ def all_cases(gauge, counter, records) -> List[Tuple[str, "sbw.Query", Callable] return cases +@unittest.skipUnless(CLICKHOUSE, "clickhouse binary not found") +class KeyPruningTest(unittest.TestCase): + """ClickHouse 26.10 prunes `ts > ` on a DateTime64 key as + `ts >= + 1`, so a left-open window lost the samples in its first + second. Small granules let the index prune them here too.""" + + def test_samples_just_after_window_start_are_kept(self): + end_s = DATA_START_MS // 1000 + 3600 + start_ms = (end_s - 60) * 1000 + # Every 7 ms from just before the window start to past its end; each + # record has its own user_id, so the distinct count is a row count. + times = range(start_ms - 50, end_s * 1000 + 50, 7) + records = [(t, ("r0", "s0", "e0", "200"), (1.0, float(t))) for t in times] + series = [(t, ("job_0", "label_0_0"), (1.0,)) for t in times] + cases = [ + ( + sbw.query_from_entry(mg_entry("cardinality", ["region"], 60), METRICS), + sum(start_ms < t <= end_s * 1000 for t in times), + ), + ( + sbw.parse_promql("count_over_time(data[1m])", 60, LABELS), + sum(start_ms <= t <= end_s * 1000 for t in times), + ), + ] + with tempfile.TemporaryDirectory() as tmp: + script = load_script(tmp, {"http": records, "data": series}).replace( + "ts);", "ts) SETTINGS index_granularity = 8;" + ) + self.assertIn("index_granularity = 8", script) + statements = [ + (str(i), sbw.raw_sql(q, LABELS).replace(sbw.END, str(end_s))) + for i, (q, _) in enumerate(cases) + ] + rows = run_clickhouse(script + query_script(statements), tmp) + got = {row["qid"]: row["value"] for row in rows} + self.assertEqual(got, {str(i): want for i, (_, want) in enumerate(cases)}) + + @unittest.skipUnless(CLICKHOUSE, "clickhouse binary not found") class ClickHouseSemanticsTest(unittest.TestCase): def test_raw_sql_matches_prometheus_reference(self): From f8a50b59c7de985a4e2b4ca76c40ca5d266ad914 Mon Sep 17 00:00:00 2001 From: zz_y Date: Tue, 13 Oct 2026 18:36:42 +0000 Subject: [PATCH 09/10] 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 Claude-Session: https://claude.ai/code/session_01AnJ7LocWxguh3q4QWkCYae --- .../fake_metric/sketch_bench_workloads.py | 57 +++++++++++-------- .../tests/test_sketch_bench_workloads.py | 42 ++++++++++++-- 2 files changed, 71 insertions(+), 28 deletions(-) diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py index e2481fde..10b31506 100644 --- a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -11,7 +11,9 @@ For a table `.json` this writes, under OUT_CONFIG_DIR: - /init.sql: the metric tables; - /interval_s.sql: one statement per query that repeats every T s; -- /asapquery.sql: the structured queries in ASAPQuery's SQL dialect; +- /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; - _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) diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py index 39ea6839..38b097f7 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_workloads.py +++ b/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) From 64188274d0d43ac911c0647ec9780e7349009d1c Mon Sep 17 00:00:00 2001 From: Milind Srivastava Date: Wed, 14 Oct 2026 10:22:58 -0400 Subject: [PATCH 10/10] fix(asap-tools): address #844 review on sketch-bench SQL rendering - Render an empty structured grouping as the grand total, no GROUP BY. - Comment PromQL on one line so multi-line PromQL stays a comment. - Reject PromQL over a records metric, and ASAPQuery sql without NOW(). - Document the 1 s instant lookback and the UTC server requirement. - Drop PR references from comments. Co-Authored-By: Claude Opus 5.5 --- .../fake_metric/sketch_bench_workloads.py | 43 +++++++++++-------- .../tests/test_sketch_bench_workloads.py | 19 ++++++-- 2 files changed, 40 insertions(+), 22 deletions(-) diff --git a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py index 10b31506..3f599f5e 100644 --- a/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -20,7 +20,10 @@ Semantics are those of Prometheus 2.53 (the version installation/prometheus installs for bare-metal runs), whose windows are closed on both ends: range vectors [d] at T read [T - d, T]; instant vectors read the latest sample of -each series in [T - 1s, T] (a lookback of one scrape interval); quantiles use +each series in [T - 1s, T] (a lookback of one scrape interval, not +Prometheus's 5m default: the feeder writes every series every scrape, so the +two agree unless a sample is missing, where Prometheus would return the +series' last value and this SQL drops the series); quantiles use quantileExactInclusive; rate and increase reproduce extrapolatedRate (counters never reset). Structured queries involve no Prometheus: they read ASAPQuery's left-open window (T - W, T], the window its SQL frontend and precompute @@ -30,14 +33,9 @@ query shapes raise. Evaluation times: each query group sets `align_to_interval` and a -`query_time_offset` of QUERY_TIME_OFFSET_SECS. The query client of PR #841 -then evaluates repetition k at floor((start - O) / T) * T + k * T and sends -it O seconds later, so every run and arm answers the same windows. The -current client ignores `align_to_interval`: it subtracts O from the eval -time and sleeps T after each repetition's queries finish, so its eval times -drift by the query latency. A 1 s group whose queries take 0.3 s skips -about one T in four, and a 60 s group occasionally repeats or skips a T; -arms then answer different windows. +`query_time_offset` of QUERY_TIME_OFFSET_SECS, so the query client evaluates +repetition k at floor((start - O) / T) * T + k * T and sends it O seconds +later; every run and arm answers the same windows. Run from asap-tools/experiments: python3 datasets/fake_metric/sketch_bench_workloads.py \ @@ -63,8 +61,8 @@ OUT_CONFIG_DIR = "config/experiment_type/sketch_bench_clickhouse" CLICKHOUSE_URL = "http://localhost:8123" -# Seconds after an eval time the client sends it (PR #841's default), so -# data up to the eval time has arrived. +# Seconds after an eval time the client sends it, so data up to the eval +# time has arrived. QUERY_TIME_OFFSET_SECS = 10 VALUE_COLUMN = "value" # Instant vectors read the latest sample of each series in the last scrape. @@ -196,9 +194,12 @@ def query_from_entry(entry: dict, metrics: dict) -> Query: `{table, value, capability, grouping, quantile, window_secs, interval_secs}` that pools every sample of a group over the window.""" if "promql" in entry: - return parse_promql( + q = parse_promql( entry["promql"], entry["interval_secs"], metric_labels(metrics) ) + if metrics[q.table]["type"] == "records": + raise ValueError(f"PromQL over records metric {q.table!r}: {entry}") + return q table, capability = entry["table"], entry["capability"] if capability not in STRUCTURED_AGGS: raise ValueError(f"unsupported capability {capability!r}: {entry}") @@ -222,7 +223,7 @@ def query_from_entry(entry: dict, metrics: dict) -> Query: def describe(entry: dict) -> str: """A one-line comment naming the query.""" if "promql" in entry: - return entry["promql"] + return " ".join(entry["promql"].split()) level = f"{entry['quantile']}, " if entry["capability"] == "quantile" else "" return ( f"{entry['capability']}({level}{entry['value']}) by " @@ -255,7 +256,11 @@ def render( value = agg(q.agg, q.quantile) if q.ratio_quantile is not None: value = f"{value} / {agg('quantile', q.ratio_quantile)}" - sql = f"SELECT {group}, {value} AS value {window} GROUP BY {group}" + if group: + sql = f"SELECT {group}, {value} AS value {window} GROUP BY {group}" + else: + # The grand total: one row, none for an empty window. + sql = f"SELECT {value} AS value {window} HAVING count() > 0" else: if q.series_fn == "last": sql = ( @@ -334,13 +339,14 @@ 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 + the literal is floored in the client's template. The literal is UTC and + the frontend reads timezone-naive literals as server-local time, so the + ASAPQuery server must run in UTC. 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}") + if "NOW()" not in entry.get("sql", ""): + raise ValueError(f"structured entry without an ASAPQuery sql on NOW(): {entry}") end = floored_end_datetime(entry["interval_secs"]) return entry["sql"].replace("NOW()", f"'{end}'") @@ -422,7 +428,6 @@ def convert( "client_options": { "starting_delay": starting_delay_secs, "repetitions": duration_secs // interval, - # Takes effect with PR #841's query client; see above. "align_to_interval": True, "query_time_offset": QUERY_TIME_OFFSET_SECS, }, diff --git a/asap-tools/experiments/tests/test_sketch_bench_workloads.py b/asap-tools/experiments/tests/test_sketch_bench_workloads.py index 38b097f7..68aa8fdc 100644 --- a/asap-tools/experiments/tests/test_sketch_bench_workloads.py +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -150,10 +150,22 @@ def test_unsupported_structured_entries_raise(self): {**mg_entry("cardinality", ["region"], 300), "capability": "sum"}, mg_entry("cardinality", ["instance"], 300), {**mg_entry("quantile", ["region"], 300), "value": "bytes"}, + # records tables have no `value` column. + {"promql": "sum_over_time(http[1m])", "interval_secs": 60, "rqes": []}, ]: with self.subTest(entry=entry), self.assertRaises(ValueError): sbw.query_from_entry(entry, METRICS) + def test_asapquery_sql_without_now_raises(self): + entry = mg_entry("quantile", ["service"], 300) + entry["sql"] = entry["sql"].replace("NOW()", "now()") + with self.assertRaises(ValueError): + sbw.asapquery_sql(entry) + + def test_describe_is_one_line(self): + entry = {"promql": "sum by (job) (\n data\n)", "interval_secs": 1} + self.assertEqual(sbw.describe(entry), "sum by (job) ( data )") + def mg_entry(capability: str, grouping: List[str], window_secs: int) -> dict: """A structured multi-grouping entry over `http`, as sketch-bench writes it.""" @@ -172,8 +184,9 @@ def mg_entry(capability: str, grouping: List[str], window_secs: int) -> dict: 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}" + f"SELECT {group + ', ' if group else ''}{agg} FROM http WHERE ts BETWEEN " + f"DATEADD(s, -{window_secs}, NOW()) AND NOW()" + + (f" GROUP BY {group}" if group else "") ) return entry @@ -503,7 +516,7 @@ def pooled(records, t_ms: int, w_ms: int, q: "sbw.Query") -> Vector: MULTIGROUP = [ mg_entry("quantile", list(g), w) for w in (120, 300) - for n in (1, 2, 3) + for n in (0, 1, 2, 3) for g in itertools.combinations(["region", "service", "endpoint"], n) ] + [ mg_entry("cardinality", g, w)