diff --git a/asap-tools/experiments/datasets/fake_metric/generate_workloads.py b/asap-tools/experiments/datasets/fake_metric/generate_workloads.py index a68bb8b7..5b914b8c 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 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, " - 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..2abfd2ad --- /dev/null +++ b/asap-tools/experiments/datasets/fake_metric/sketch_bench_workloads.py @@ -0,0 +1,498 @@ +#!/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, 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: 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. + +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, 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 +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. + +Evaluation times: each query group sets `align_to_interval` and a +`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 \ + --table --data-file \ + --starting-delay-secs --duration-secs +""" +import argparse +import json +import os +import re +import warnings +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" +# 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. +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+), (.+)\)$") +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 and + set `left_open`: their window is (end - W, end], not PromQL's closed one. + """ + + 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 + left_open: bool = False + + +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, metrics: dict, labels: Dict[str, List[str]]) -> 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. + `labels` is metric_labels(metrics).""" + if "promql" in entry: + q = parse_promql(entry["promql"], entry["interval_secs"], labels) + 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}") + if table not in labels: + raise ValueError(f"unknown table {table!r}: {entry}") + unknown = set(entry["grouping"]) - set(labels[table]) + 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"], + left_open=True, + ) + + +def describe(entry: dict) -> str: + """A one-line comment naming the query.""" + if "promql" in entry: + return " ".join(entry["promql"].split()) + 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: + """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, 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], 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, + 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 {where(end, q.window_secs)}" + 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)}" + 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 = ( + 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 == "cardinality": + return f"uniqExact({value})" + 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, + # 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), + ) + + +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 }}. + + 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). + """ + 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. 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 "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"]) + sql = entry["sql"].rstrip().rstrip(";") + return sql.replace("NOW()", f"'{end}'") + + +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, desc in metrics.items(): + names = [label["name"] for label in desc["labels"]] + if not names: + raise ValueError(f"metric {table!r} has no 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), {', '.join(columns)}) " + f"ENGINE = MergeTree ORDER BY ({', '.join(names)}, ts)", + ] + return ";\n\n".join(statements) + ";\n" + + +def metric_labels(metrics: dict) -> Dict[str, List[str]]: + return { + name: [label["name"] for label in desc["labels"]] + 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: + workloads = json.load(f)["workloads"] + if len(workloads) != 1: + raise ValueError(f"{table_path}: expected 1 workload, got {len(workloads)}") + e2e = workloads[0]["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(metrics)) + + by_interval: Dict[int, List[str]] = {} + asapquery: List[str] = [] + max_window = 0 + for entry in e2e["queries"]: + q = query_from_entry(entry, metrics, labels) + if q.interval_secs > duration_secs: + raise ValueError( + f"interval {q.interval_secs}s exceeds --duration-secs " + f"{duration_secs}: {describe(entry)}" + ) + max_window = max(max_window, q.window_secs) + by_interval.setdefault(q.interval_secs, []).append( + f"-- {describe(entry)}\n{raw_sql(q, labels)}" + ) + if "promql" not in entry: + 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) + if starting_delay_secs < QUERY_TIME_OFFSET_SECS + max_window: + # Same data for every arm, so arms stay comparable; absolutes are biased. + warnings.warn( + f"--starting-delay-secs {starting_delay_secs} < " + f"{QUERY_TIME_OFFSET_SECS + max_window}: early windows start " + "before ingestion" + ) + + 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") + write_statements(sql_path, statements) + 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, + "align_to_interval": True, + "query_time_offset": QUERY_TIME_OFFSET_SECS, + }, + } + ) + + 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 only this table; nothing fills the + # other metrics' tables yet. + "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..55f11fd0 --- /dev/null +++ b/asap-tools/experiments/tests/test_sketch_bench_workloads.py @@ -0,0 +1,734 @@ +"""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 datetime +import itertools +import json +import math +import os +import random +import shutil +import subprocess +import sys +import tempfile +import unittest +import warnings +from typing import Callable, Dict, List, Tuple + +import jinja2 +import yaml + +sys.path.insert( + 0, os.path.join(os.path.dirname(__file__), "..", "datasets", "fake_metric") +) + +import sketch_bench_workloads as sbw # noqa: E402 + +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). +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_entries(self): + cardinality = mg_entry("cardinality", ["region", "service"], 300) + quantile = mg_entry("quantile", ["service"], 900) + self.assertEqual( + sbw.query_from_entry(cardinality, METRICS, LABELS), + sbw.Query( + "http", + 60, + 300, + ("region", "service"), + None, + "cardinality", + value="user_id", + left_open=True, + ), + ) + self.assertEqual( + sbw.query_from_entry(quantile, METRICS, LABELS), + sbw.Query( + "http", + 60, + 900, + ("service",), + None, + "quantile", + "0.99", + value="latency", + left_open=True, + ), + ) + + 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"}, + {**mg_entry("quantile", ["region"], 300), "table": "nope"}, + # 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, LABELS) + + 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_asapquery_sql_strips_trailing_semicolon(self): + entry = mg_entry("quantile", ["service"], 300) + plain = sbw.asapquery_sql(entry) + entry["sql"] += "; \n" + self.assertEqual(sbw.asapquery_sql(entry), plain) + + def test_metric_without_labels_raises(self): + with self.assertRaises(ValueError): + sbw.init_sql({"data": {"type": "gauge", "labels": []}}) + + 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.""" + entry = { + "table": "http", + "value": "latency" if capability == "quantile" else "user_id", + "capability": capability, + "grouping": grouping, + "window_secs": window_secs, + "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 + ', ' 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 + + +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 = { + "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") + with open(table, "w") as f: + json.dump({"workloads": [{"e2e": e2e}]}, f) + cwd = os.getcwd() + os.chdir(tmp) + try: + # 30 s < offset + the 15m windows: early windows predate data. + with self.assertWarns(UserWarning): + out_path = sbw.convert(table, "data.json", 30, 600) + with open(out_path) 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, + "align_to_interval": True, + "query_time_offset": 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), 9) + with open(cfg["dataset"]["init_sql_file"]) as f: + 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"]) + # 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" + + e2e["queries"][-1]["sql"].replace("NOW()", end) + + ";\n", + ) + finally: + os.chdir(cwd) + + def test_checks_workloads_duration_and_delay(self): + e2e = {"metrics": METRICS, "queries": [mg_entry("quantile", ["service"], 300)]} + with tempfile.TemporaryDirectory() as tmp: + table = os.path.join(tmp, "t.json") + cwd = os.getcwd() + os.chdir(tmp) + try: + with open(table, "w") as f: + json.dump({"workloads": [{"e2e": e2e}, {"e2e": e2e}]}, f) + with self.assertRaises(ValueError): + sbw.convert(table, "data.json", 310, 600) + with open(table, "w") as f: + json.dump({"workloads": [{"e2e": e2e}]}, f) + # Interval 60 s, so 30 s would yield 0 repetitions. + with self.assertRaises(ValueError): + sbw.convert(table, "data.json", 310, 30) + # Delay covers offset + window: no warning. + with warnings.catch_warnings(): + warnings.simplefilter("error") + sbw.convert(table, "data.json", 310, 600) + 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: + # 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: + 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 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 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: + 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 (0, 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: + 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, 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: + 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), " + f"{', '.join(names + values)} FROM file('{csv}', 'CSV', '{structure}');\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) + ) + + +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, LABELS) + 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 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, LABELS + ), + 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): + gauge, counter = make_dataset() + records = make_records() + cases = all_cases(gauge, counter, records) + statements, groupings = [], {} + 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, rows) + got = rows_by_id( + run_clickhouse(script + query_script(statements), tmp), groupings + ) + + BRANCHES_HIT.clear() + nonempty = 0 + 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( + 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()