Skip to content
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion rqe-optimizer/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ Every table also has `workloads[0].e2e`, for running a plan end to end

| Field | Meaning |
|---|---|
| `e2e.queries` | classic: one entry per PromQL query and repeat interval: `promql`, `interval_secs`, and the `rqes` it reads (template 10's ratio reads two). Multi-grouping (no PromQL): one entry per distinct query and interval: canonical ASAPQuery `sql` over `NOW()` (the planner floors and replaces it), plus `table`, `value`, `capability`, `grouping`, `quantile` (p99: 0.99), `window_secs`, `interval_secs`, and the `rqes` it reads (sorted; templates 14, 15 and 17 share some groupings) |
| `e2e.queries` | classic: one entry per PromQL query and repeat interval: `promql`, `interval_secs`, and the `rqes` it reads (template 10's ratio reads two). Multi-grouping (no PromQL): one entry per distinct query and interval: canonical ASAPQuery `sql` over `NOW()` (the planner floors and replaces it), plus `table`, `value`, `capability`, `grouping`, `quantile` (p99: 0.99), `window_secs`, `interval_secs`, and the `rqes` it reads: only the strictest of templates 14, 15 and 17 where they share a grouping, or every one, sorted, with `--keep-duplicate-queries` |
| `e2e.metrics` | the metrics to generate. Classic: gauge `data` (sums, quantiles, top-k by sum) and counter `data_total` (rates), with `labels` in the `schemas` format (`label_0` is `child_of: job`, 1,000 per job, so 1e4 series), samples/s per series, and each stream's value shape. Multi-grouping: `type: "records"` tables `http` and `flows` (`data_1_http` for metric `data_1/http` in `metrics=m` tables) with the schema's `labels`, `records_per_sec` and `values` (shape, `scaled_by`, `anomaly`, `burst`) |

### Accuracy and cost
Expand Down
4 changes: 2 additions & 2 deletions rqe-optimizer/results/autosketch-vs-asap-hydra/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -200,8 +200,8 @@ zcat plans/synthetic-templatesmultigroup-shared1.plans.json.gz | jq '.plans[]
- The accuracy curves (`out_grid_1e7_cost/`, `out_1e9/`) are the committed
#138 study's, in `../autosketch-vs-asap-inputs/saturation/` (#194), with
#213's new KLL and DDSketch configs appended.
- `inputs/tables/`: the workloads, `export_autosketch_eval_table.py
--synthetic` (classic, multigroup and all × r ∈ {1, 8} × m ∈ {1, 8, 16}).
- `inputs/tables/`: the workloads, `export_autosketch_eval_table.py --synthetic
--keep-duplicate-queries` (classic, multigroup and all × r ∈ {1, 8} × m ∈ {1, 8, 16}).

## Reproduce

Expand Down
2 changes: 1 addition & 1 deletion rqe-optimizer/results/autosketch-vs-asap-inputs/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,7 @@ and `../autosketch-vs-asap-synthetic-sla/`), as run on 2026-10-08.
`scripts/merge_study_parts.py`, which refuses a point measured twice. Their
raw records are `raw/dense_*`.
- `tables/`: the synthetic mixed-set workloads from
`scripts/export_autosketch_eval_table.py --synthetic` (RQEs, streams,
`scripts/export_autosketch_eval_table.py --synthetic --keep-duplicate-queries` (RQEs, streams,
targets, data shapes) and `plan.tsv` (one run per table).

AutoSketch's measured benchmark times are next to each figure's results
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -74,7 +74,7 @@ approxbench accuracy benchmark, serially, alone on clnode109
```sh
# The inputs these results used are committed: TABLES=../autosketch-vs-asap-inputs/tables,
# DIR=../autosketch-vs-asap-inputs/saturation (paths from this directory).
python3 scripts/export_autosketch_eval_table.py --synthetic --out TABLES
python3 scripts/export_autosketch_eval_table.py --synthetic --keep-duplicate-queries --out TABLES
# For each row of TABLES/plan.tsv:
target/release/examples/autosketch_vs_asap synthetic --table TABLES/T --target p95 \
--saturation-dir DIR --out OUT/R --runs 3
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -165,7 +165,7 @@ python3 scripts/study_saturation.py --phase accuracy --n-max 1e9 --seeds 3 \
python3 scripts/study_saturation.py --phase optimizer-cost --out DIR/optimizer_cost

# Workload tables and plan.tsv, then the runs.
python3 scripts/export_autosketch_eval_table.py --synthetic --out TABLES
python3 scripts/export_autosketch_eval_table.py --synthetic --keep-duplicate-queries --out TABLES
# Or use the committed inputs: TABLES=rqe-optimizer/results/autosketch-vs-asap-inputs/tables,
# DIR=rqe-optimizer/results/autosketch-vs-asap-inputs/saturation.
SATURATION_DIR=DIR scripts/run_autosketch_vs_asap_synthetic.sh TABLES run
Expand Down
73 changes: 60 additions & 13 deletions scripts/export_autosketch_eval_table.py
Original file line number Diff line number Diff line change
Expand Up @@ -297,6 +297,10 @@ def rqe_entry(q, curves, curves_big, sat, sat_big, merge):
# Set by --samples-per-sec: a schema RQE whose smallest covered group falls
# below SCHEMA_MIN_COVERED_N is left out (logged) instead of failing.
SKIP_BELOW_CURVES = False
# Set by --keep-duplicate-queries: every template's copy of a query is an
# RQE. Otherwise only the strictest copy is, since the engine answers one
# query string from one aggregation.
KEEP_DUPLICATE_QUERIES = False


def set_samples_per_sec(rate, skip_below_curves):
Expand Down Expand Up @@ -534,6 +538,28 @@ def add(query_id, stream, capability, rng, s, t, groups, keys=None, promql=None)
return out


def schema_query(name, value, capability, what, grouping):
"""A schema template's query on `grouping`: the fields that make it
one query string."""
query = {"table": name, "value": value, "capability": capability, "grouping": grouping}
if capability == "quantile": # p99_latency -> 0.99
query["quantile"] = int(what[1:what.index("_")]) / 100
return query


def strictest_templates():
"""{query: (number of templates asking it, the strictest one's number)}
over SCHEMA_TEMPLATES. Stricter covers more groups: every group (None),
then the smallest `covers_share`."""
copies = {}
for n, what, name, value, capability, groupings, covers in SCHEMA_TEMPLATES:
for grouping in groupings:
key = json.dumps(schema_query(name, value, capability, what, grouping),
sort_keys=True)
copies.setdefault(key, []).append((covers is not None, covers or 0, n))
return {key: (len(c), min(c)[2]) for key, c in copies.items()}


def schema_queries(interval, dataset, range_with_interval, seen, metric=None, e2e=None):
"""SCHEMA_TEMPLATES as synthetic_queries' query-ranges, over
SCHEMA_WINDOWS, repeating every `interval` seconds. A metric's values are
Expand All @@ -545,13 +571,21 @@ def schema_queries(interval, dataset, range_with_interval, seen, metric=None, e2
ASAPQuery SQL (there is no PromQL): {sql, table, value,
capability, grouping, quantile (quantiles only), window_secs,
interval_secs, rqes}. `table` is the metric with `/` as `_`
(`data_1_http`)."""
(`data_1_http`).

A query several templates ask (14, 15 and 17 share groupings) is only
its strictest template's RQE, unless KEEP_DUPLICATE_QUERIES."""
out = []
strictest = strictest_templates()
for n, what, name, value, capability, groupings, covers in SCHEMA_TEMPLATES:
schema = SCHEMAS[name]
shape = schema["values"][value]
full = f"{metric}/{name}" if metric else name
for grouping in groupings:
copies, keep = strictest[json.dumps(
schema_query(name, value, capability, what, grouping), sort_keys=True)]
if not KEEP_DUPLICATE_QUERIES and keep != n:
continue
groups = grouping_cardinality(name, grouping)
query_id = f"t{n}_{what}_by_{','.join(grouping)}"
if metric:
Expand All @@ -571,16 +605,21 @@ def schema_queries(interval, dataset, range_with_interval, seen, metric=None, e2
# can't serve the RQE, so a lower --samples-per-sec leaves it
# (and its e2e query) out.
if SKIP_BELOW_CURVES and covered_min_n < SCHEMA_MIN_COVERED_N:
# A weaker template's copy might fit, but serving the
# query at a weaker target would change the workload.
if not KEEP_DUPLICATE_QUERIES and copies > 1:

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This turns a leave-out into a hard failure for the whole export. With --synthetic --samples-per-sec 2 (or 1) it raises on t17_distinct_users_by_service,endpoint/keys/5m (the strictest of 3 templates, covered group of 449 items), and no tables get written, not even classic. Before this PR only that RQE was dropped. Should this keep leaving the query out (all its copies) with a stderr line, instead of raising?

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Keeping the error; that was the decision in #214. A query several templates ask must keep its strictest requirement, so the export stops rather than drop or weaken it. At --samples-per-sec 2 the fix is to change the templates (or pass --keep-duplicate-queries), not to plan a workload without that query.

raise ValueError(
f"{dataset}/{query_id}/{kind}/{rng}, the strictest of {copies} "
f"templates asking its query, has a covered group of "
f"{covered_min_n:.0f} items, below the curves' first N")
print(f"left out {dataset}/{query_id}/{kind}/{rng}: smallest covered "
f"group holds {covered_min_n:.0f} items, below the curves' first N",
file=sys.stderr)
continue
assert covered_min_n >= SCHEMA_MIN_COVERED_N, (query_id, rng, covered_min_n)
if e2e is not None:
query = {"table": full.replace("/", "_"), "value": value,
"capability": capability, "grouping": grouping}
if capability == "quantile": # p99_latency -> 0.99
query["quantile"] = int(what[1:what.index("_")]) / 100
query = schema_query(full.replace("/", "_"), value, capability, what,
grouping)
query["sql"] = asapquery_sql(query, s)
e2e.append({**query, "window_secs": s, "interval_secs": interval,
"rqes": [f"{dataset}/{query_id}/{kind}/{rng}"]})
Expand Down Expand Up @@ -834,21 +873,29 @@ def main(argv=None):
help=f"--synthetic: samples/s per series (default "
f"{SYNTHETIC_SAMPLES_PER_SEC}); the schema metrics' rates scale "
"with it, and schema RQEs below the curves' first N are left out")

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Stale: without --keep-duplicate-queries, a shared query's strictest copy below the curves' first N now raises rather than being left out. The help here (and on --keep-duplicate-queries) should say so.

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Done in 5445c5c: both helps now say the strictest copy below the curves' first N fails the export without --keep-duplicate-queries.

parser.add_argument("--keep-duplicate-queries", action="store_true",
help="--synthetic: an RQE for every template asking a query, not "
"just the strictest (the section 6.3 tables)")
parser.add_argument("--out", required=True)
args = parser.parse_args(argv)
# An instant query reads one scrape, 1000 / rate ms, which the runner keeps
# in whole ms.
rate = args.samples_per_sec
if rate is not None and (rate <= 0 or 1000 % rate):
parser.error("--samples-per-sec must divide 1000")
if args.samples_per_sec is not None:
default = SYNTHETIC_SAMPLES_PER_SEC
set_samples_per_sec(args.samples_per_sec, True)
try:
return run(args, parser)
finally:
set_samples_per_sec(default, False)
return run(args, parser)
global KEEP_DUPLICATE_QUERIES
KEEP_DUPLICATE_QUERIES = args.keep_duplicate_queries
try:
if args.samples_per_sec is not None:
default = SYNTHETIC_SAMPLES_PER_SEC
set_samples_per_sec(args.samples_per_sec, True)
try:
return run(args, parser)
finally:
set_samples_per_sec(default, False)
return run(args, parser)
finally:
KEEP_DUPLICATE_QUERIES = False


def run(args, parser):
Expand Down
44 changes: 41 additions & 3 deletions scripts/test_export_autosketch_eval_table.py
Original file line number Diff line number Diff line change
Expand Up @@ -245,6 +245,7 @@ def test_schema_queries_read_every_rqe_with_their_fields(self):
"SELECT dst_subnet, COUNT(DISTINCT src_ip) FROM flows WHERE ts "
"BETWEEN DATEADD(s, -300, NOW()) AND NOW() GROUP BY dst_subnet")

@mock.patch("export_autosketch_eval_table.KEEP_DUPLICATE_QUERIES", True)
def test_schema_metrics_are_records_tables_named_by_metric(self):
e2e = []
qs = metric_queries({**SYNTHETIC_DEFAULT, "templates": "multigroup", "metrics": 2},
Expand Down Expand Up @@ -272,6 +273,38 @@ def test_schema_metrics_are_records_tables_named_by_metric(self):
[["t14_distinct_users_by_service_data_0",
"t15_distinct_users_by_service_data_0"]])

def test_each_query_is_only_its_strictest_templates_rqe(self):
# One query string can be served by one aggregation, so a query
# templates 14, 15 and 17 share keeps the template covering the most
# groups: 17 (every group), then 14 (1%), then 15 (5%).
e2e = []
synthetic_queries(templates="multigroup", dataset="d", e2e=e2e)
block = e2e_block([], e2e)
self.assertTrue(all(len(q["rqes"]) == 1 for q in block["queries"]))
users = {(tuple(q["grouping"]), q["window_secs"]): q["rqes"][0].split("/")[1]
for q in block["queries"] if q["value"] == "user_id"}
for window in [300, 900]:
self.assertTrue(users[(("service", "endpoint"), window)].startswith("t17_"))
for grouping in [("region",), ("service",), ("region", "service")]:
self.assertTrue(users[(grouping, window)].startswith("t14_"))
self.assertTrue(users[(("endpoint",), window)].startswith("t15_"))

def test_a_strictest_copy_below_the_curves_fails_loudly(self):
# Every group of the full label set: the smallest has ~350 items in 5m
# (left out) and ~1060 in 15m.
grouping = [all_subsets("http")[-1]]
every = (17, "distinct_users", "http", "user_id", "cardinality", grouping, None)
some = (15, "distinct_users", "http", "user_id", "cardinality", grouping, 0.05)
with mock.patch("export_autosketch_eval_table.SKIP_BELOW_CURVES", True), \
mock.patch("sys.stderr"):
with mock.patch("export_autosketch_eval_table.SCHEMA_TEMPLATES", [every]):
self.assertEqual([q["range"] for q in synthetic_queries(templates="multigroup")],
["15m"])
# A weaker copy would fit, but would weaken the workload's target.
with mock.patch("export_autosketch_eval_table.SCHEMA_TEMPLATES", [some, every]):
with self.assertRaisesRegex(ValueError, "strictest of 2 templates"):
synthetic_queries(templates="multigroup")


COMMITTED_TABLES = os.path.join(os.path.dirname(os.path.abspath(__file__)), "..",
"rqe-optimizer", "results", "autosketch-vs-asap-inputs",
Expand Down Expand Up @@ -317,8 +350,12 @@ def test_all_is_classic_plus_the_schema_templates(self):
every = synthetic_queries(templates="all", dataset="d")
self.assertEqual(every[:len(classic)], classic)
new = every[len(classic):]
# Per window (5m, 15m): 1 + 3 + 4 + 15 + 7 + 1 groupings.
self.assertEqual(len(new), 2 * 31)
# Per window (5m, 15m): 1 + 3 + 4 + 15 + 7 + 1 groupings, of which 5
# repeat another template's query.
self.assertEqual(len(new), 2 * 26)
with mock.patch("export_autosketch_eval_table.KEEP_DUPLICATE_QUERIES", True):
self.assertEqual(len(synthetic_queries(templates="multigroup", dataset="d")),
2 * 31)
self.assertEqual({q["stream"] for q in new},
{"flows/src_ip", "http/user_id", "http/latency"})
self.assertEqual({q["range"] for q in new}, {"5m", "15m"})
Expand Down Expand Up @@ -454,7 +491,8 @@ def test_samples_per_sec_scales_rates_and_leaves_out_rqes_below_the_curves(self)
for g in ["region,endpoint", "region,service,endpoint"]}
ids = {r["id"].split("/", 3)[3] for r in workload["rqes"]}
e2e_ids = {r.split("/", 3)[3] for q in workload["e2e"]["queries"] for r in q["rqes"]}
self.assertEqual(len(ids), 60)
# 62 RQEs, less 10 duplicate queries and the 2 left out.
self.assertEqual(len(ids), 50)
self.assertFalse(left_out & (ids | e2e_ids))
self.assertIn("t16_p99_latency_by_region,endpoint/values/15m", ids)
self.assertEqual(workload["e2e"]["metrics"]["http"]["records_per_sec"], 100_000)
Expand Down
Loading