Saturation curves at label cardinalities 1e1–1e6 for the synthetic planner workload

scripts/check_cost_size.py

@@ -0,0 +1,76 @@
+#!/usr/bin/env python3
+"""Check that cost measured at a smaller N matches cost measured at the full N.
+
+Re-measures the given points of a serial saturation_cost.jsonl one at a time
+at --n-max (the cost phase's approxbench command otherwise) and prints, per
+point, the small-N / full-N ratio of:
+- insert CPU per item (each run's insert CPU divided by its own N);
+- merge CPU per fold (MERGE_SHARDS shards folded into one, same at both N);
+- the whole query phase's CPU, as the evaluation tables use it;
+- CPU per query (the query phase divided by its query count).
+
+Usage: check_cost_size.py SERIAL_JSONL OUT_JSONL --n-max 1e7 [--seed 1]
+           [--binary ./target/release/approxbench] < keys.jsonl
+       One JSON key per line on stdin:
+       ["cms-fastpath-vector2d", "rows=3 cols=1024", "zipf", 1.0, 10000]
+"""
+
+import argparse
+import json
+import sys
+
+from complete_saturation_costs import dataset_args
+from study_saturation import MERGE_SHARDS, cpu_secs, load_saved, point_key, query_secs, run
+
+
+def size(record):
+    return record["workload"]["synthetic"]["description"]["row_num"]
+
+
+def per_op(record):
+    return {
+        "insert per item": cpu_secs(record, "insert") / size(record),
+        "merge per fold": cpu_secs(record, "merge") / (MERGE_SHARDS - 1),
+        "query phase": cpu_secs(record, "query"),
+        "per query": query_secs(record),
+    }
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("serial")
+    parser.add_argument("out")
+    parser.add_argument("--binary", default="./target/release/approxbench")
+    parser.add_argument("--n-max", type=float, required=True)
+    parser.add_argument("--seed", type=int, default=1)
+    args = parser.parse_args()
+
+    serial = load_saved(args.serial, point_key)
+    keys = [json.loads(line) for line in sys.stdin if line.strip()]
+    ops = list(per_op(next(iter(serial.values()))))
+    print("| sketch | config | θ | K | " + " | ".join(f"{op} small/full N" for op in ops) + " |")
+    print("|---|---|---|---|" + "---|" * len(ops))
+    worst = 0.0
+    with open(args.out, "w") as out:
+        for key in keys:
+            sketch, config, dist, param, card = key
+            row = {"dist": dist, "param": str(param), "cardinality": str(card)}
+            record = run(args.binary, [
+                "--variant", sketch, "--library", "lib", "--config", config,
+                "--operations", "insert,query,merge",
+                "--metrics", "throughput,cpu,memory",
+                "--merge-shards", str(MERGE_SHARDS), "--runs", "3", "--warmup-runs", "1",
+                "--size", str(int(args.n_max)), "--seed", str(args.seed), "--flat",
+            ] + dataset_args(row))
+            out.write(json.dumps({"key": key, **record}) + "\n")
+            out.flush()
+            small, full = per_op(record), per_op(serial[point_key(*key)])
+            ratios = [small[op] / full[op] if full[op] else float("nan") for op in ops]
+            worst = max(worst, *(abs(r - 1) for r in ratios))
+            print(f"| {sketch} | {config} | {param:g} | {card} | "
+                  + " | ".join(f"{r:.3f}" for r in ratios) + " |", flush=True)
+    print(f"\nLargest deviation from the full-N measurement: {worst:.1%}")
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/check_parallel_costs.py

@@ -0,0 +1,64 @@
+#!/usr/bin/env python3
+"""Check that cost measured --jobs at a time matches serial measurement.
+
+Re-measures the given points of a serial saturation_cost.jsonl, all at once
+on --jobs workers, with the cost phase's approxbench command, and prints the
+parallel/serial ratio of insert, merge and query CPU seconds per point.
+
+Usage: check_parallel_costs.py SERIAL_JSONL OUT_JSONL [--jobs 12]
+           [--n-max 1e8] [--seed 1] [--binary ./target/release/approxbench]
+       Points are read from stdin, one JSON key per line:
+       ["cms-fastpath-vector2d", "rows=3 cols=1024", "zipf", 1.0, 10000]
+"""
+
+import argparse
+import json
+import sys
+from concurrent.futures import ThreadPoolExecutor
+
+from complete_saturation_costs import dataset_args
+from study_saturation import MERGE_SHARDS, cpu_secs, load_saved, point_key, run
+
+OPS = ("insert", "merge", "query")
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("serial")
+    parser.add_argument("out")
+    parser.add_argument("--binary", default="./target/release/approxbench")
+    parser.add_argument("--jobs", type=int, default=12)
+    parser.add_argument("--n-max", type=float, default=1e8)
+    parser.add_argument("--seed", type=int, default=1)
+    args = parser.parse_args()
+
+    serial = load_saved(args.serial, point_key)
+    keys = [json.loads(line) for line in sys.stdin if line.strip()]
+
+    def measure(key):
+        sketch, config, dist, param, card = key
+        row = {"dist": dist, "param": str(param), "cardinality": str(card)}
+        return key, run(args.binary, [
+            "--variant", sketch, "--library", "lib", "--config", config,
+            "--operations", "insert,query,merge",
+            "--metrics", "throughput,cpu,memory",
+            "--merge-shards", str(MERGE_SHARDS), "--runs", "3", "--warmup-runs", "1",
+            "--size", str(int(args.n_max)), "--seed", str(args.seed), "--flat",
+        ] + dataset_args(row))
+
+    print("| sketch | config | θ | K | " + " | ".join(f"{op} parallel/serial" for op in OPS) + " |")
+    print("|---|---|---|---|" + "---|" * len(OPS))
+    worst = 0.0
+    with ThreadPoolExecutor(max_workers=args.jobs) as pool, open(args.out, "w") as out:
+        for key, record in pool.map(measure, keys):
+            out.write(json.dumps({"key": key, **record}) + "\n")
+            base = serial[point_key(*key)]
+            ratios = [cpu_secs(record, op) / cpu_secs(base, op) for op in OPS]
+            worst = max(worst, *(abs(r - 1) for r in ratios))
+            print(f"| {key[0]} | {key[1]} | {key[3]:g} | {key[4]} | "
+                  + " | ".join(f"{r:.3f}" for r in ratios) + " |")
+    print(f"\nLargest deviation from serial: {worst:.1%}")
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/complete_saturation_costs.py

@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+"""Finish an interrupted `study_saturation.py --phase cost` run.
+
+The cost phase measures one point at a time and rewrites
+saturation_cost.jsonl from scratch, so an interrupted run cannot resume.
+This runs the points still missing from saturation_cost.jsonl (the same
+approxbench command as the cost phase, --jobs at a time), appends them, and
+fills the cost columns of saturation.csv from every record. It does not run
+the exact-baseline crossover.
+
+--n-max is the stream length of the new measurements. Readers of
+saturation.csv divide insert CPU by the point's last curve N, so each
+record's insert CPU is written scaled to --normalize-n (default: --n-max),
+keeping insert CPU per item equal to the record's own CPU / its own N.
+
+Usage: complete_saturation_costs.py GRID_DIR [--jobs 12] [--cost-rows 3]
+           [--n-max 1e8] [--normalize-n 1e8] [--seed 1]
+           [--binary ./target/release/approxbench]
+"""
+
+import argparse
+import csv
+import json
+import os
+import sys
+from concurrent.futures import ThreadPoolExecutor
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from study_saturation import (  # noqa: E402
+    MERGE_SHARDS, SUMMARY_COLUMNS, cpu_secs, load_saved, point_key, run)
+
+
+def dataset_args(row):
+    """The cost phase's dataset arguments for one saturation.csv row."""
+    if row["dist"] == "pareto":
+        return ["--dataset", "pareto", "--pareto-alpha", row["param"],
+                "--pareto-scale", "1000", "--cardinality", "1", "--dtype", "i64"]
+    return ["--dataset", "zipf", "--zipf-s", row["param"],
+            "--cardinality", row["cardinality"], "--dtype", "i64"]
+
+
+def main():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("grid_dir")
+    parser.add_argument("--binary", default="./target/release/approxbench")
+    parser.add_argument("--jobs", type=int, default=12)
+    parser.add_argument("--cost-rows", type=int, default=3)
+    parser.add_argument("--n-max", type=float, default=1e8)
+    parser.add_argument("--normalize-n", type=float,
+                        help="N the insert CPU is scaled to (default: --n-max)")
+    parser.add_argument("--seed", type=int, default=1)
+    args = parser.parse_args()
+    normalize_n = args.normalize_n or args.n_max
+
+    csv_path = os.path.join(args.grid_dir, "saturation.csv")
+    cost_path = os.path.join(args.grid_dir, "saturation_cost.jsonl")
+    rows = list(csv.DictReader(open(csv_path)))
+    saved = load_saved(cost_path, point_key) if os.path.exists(cost_path) else {}
+
+    def key(row):
+        return point_key(row["sketch"], row["config"], row["dist"], row["param"],
+                         row["cardinality"])
+
+    def measured(row):
+        config = row["config"]
+        return not (config.startswith("rows=")
+                    and not config.startswith(f"rows={args.cost_rows} "))
+
+    missing = [r for r in rows if measured(r) and key(r) not in saved]
+    print(f"{len(saved)} saved, {len(missing)} to run", file=sys.stderr)
+
+    def measure(row):
+        return row, run(args.binary, [
+            "--variant", row["sketch"], "--library", "lib", "--config", row["config"],
+            "--operations", "insert,query,merge",
+            "--metrics", "throughput,cpu,memory",
+            "--merge-shards", str(MERGE_SHARDS), "--runs", "3", "--warmup-runs", "1",
+            "--size", str(int(args.n_max)), "--seed", str(args.seed), "--flat",
+        ] + dataset_args(row))
+
+    with ThreadPoolExecutor(max_workers=args.jobs) as pool, open(cost_path, "a") as out:
+        for row, record in pool.map(measure, missing):
+            k = key(row)
+            out.write(json.dumps({"key": [row["sketch"], row["config"], row["dist"],
+                                          float(row["param"]), row["cardinality"]],
+                                  **record}) + "\n")
+            out.flush()
+            saved[k] = record
+            print(f"  {row['sketch']} ({row['config']}) {row['dist']}={row['param']} "
+                  f"K={row['cardinality']}", file=sys.stderr)
+
+    for row in rows:
+        record = saved.get(key(row), {}) if measured(row) else {}
+        insert = cpu_secs(record, "insert")
+        size = record.get("workload", {}).get("synthetic", {}).get("description", {}).get("row_num")
+        row["insert_cpu_secs"] = insert * normalize_n / size if insert != "" and size else insert
+        row["merge_cpu_secs"] = cpu_secs(record, "merge")
+        row["query_cpu_secs"] = cpu_secs(record, "query")
+        row["memory_bytes"] = record.get("memory_bytes", "")
+    with open(csv_path, "w", newline="") as f:
+        writer = csv.DictWriter(f, fieldnames=SUMMARY_COLUMNS)
+        writer.writeheader()
+        writer.writerows(rows)
+    print(f"Done. {csv_path}", file=sys.stderr)
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/summarize_synthetic_curves.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""Summarize the saturation runs at the synthetic workload's label
+cardinalities into the tables in
+docs/figures/saturation/synthetic_cardinalities/README.md.
+
+Usage: summarize_synthetic_curves.py GRID_DIR MERGE_DIR
+  GRID_DIR/saturation.csv            single-sketch N_sat per point
+  MERGE_DIR/saturation_merge_curve.csv  top-k precision per shard count
+Prints Markdown: N_sat per (sketch, config) x (theta, K), counts of points
+that never saturate, and the top-k merged/single precision ratio at the
+largest N of the merge run.
+"""
+
+import csv
+import math
+import sys
+from collections import defaultdict
+
+
+def nsat_table(path):
+    rows = list(csv.DictReader(open(path)))
+    thetas = sorted({float(r["param"]) for r in rows})
+    ks = sorted({int(r["cardinality"]) for r in rows})
+    cells = defaultdict(dict)
+    for r in rows:
+        cells[(r["sketch"], r["config"])][(float(r["param"]), int(r["cardinality"]))] = r["n_sat"]
+    head = ["sketch", "config"] + [f"θ={t:g} K={k:.0e}" for t in thetas for k in ks]
+    print("| " + " | ".join(head) + " |")
+    print("|" + "---|" * len(head))
+    for (sketch, config), c in sorted(cells.items()):
+        vals = []
+        for t in thetas:
+            for k in ks:
+                v = c.get((t, k), "")
+                vals.append("never" if v == "not_saturated" else (f"{float(v):.0e}" if v else ""))
+        print(f"| {sketch} | {config} | " + " | ".join(vals) + " |")
+    never = defaultdict(int)
+    total = defaultdict(int)
+    for r in rows:
+        key = (r["sketch"], int(r["cardinality"]))
+        total[key] += 1
+        never[key] += r["n_sat"] == "not_saturated"
+    print("\nPoints that never saturate by the largest N, per sketch and K:\n")
+    print("| sketch | K | never saturated / points |")
+    print("|---|---|---|")
+    for key in sorted(total):
+        print(f"| {key[0]} | {key[1]:.0e} | {never[key]} / {total[key]} |")
+
+
+def merge_table(path):
+    rows = list(csv.DictReader(open(path)))
+    nmax = max(int(r["n"]) for r in rows)
+    single = {}
+    merged = defaultdict(dict)
+    for r in rows:
+        if int(r["n"]) != nmax:
+            continue
+        key = (r["config"], float(r["param"]), int(r["cardinality"]))
+        m = int(r["shards"])
+        if m == 1:
+            single[key] = float(r["seed_mean_error"])
+        else:
+            merged[key][m] = float(r["seed_mean_error"])
+    print(f"\nTop-k precision@k merged / single at N = {nmax:.0e} (1.00 = no loss):\n")
+    shards = sorted({m for v in merged.values() for m in v})
+    print("| config | θ | K | " + " | ".join(f"m={m}" for m in shards) + " |")
+    print("|---|---|---|" + "---|" * len(shards))
+    worst = []
+    for key in sorted(merged):
+        s = single.get(key)
+        ratios = [merged[key].get(m, float("nan")) / s if s else float("nan") for m in shards]
+        if not any(math.isnan(r) for r in ratios):  # single-sketch precision 0: no ratio
+            worst.append((min(ratios), key))
+        print(f"| {key[0]} | {key[1]:g} | {key[2]:.0e} | " + " | ".join(f"{x:.2f}" for x in ratios) + " |")
+    worst.sort()
+    undefined = sum(1 for key in merged if not single.get(key))
+    print(f"\n{undefined} points have single-sketch precision 0 (no ratio, shown as nan).")
+    print("\nLargest losses:", ", ".join(f"{k[0]} θ={k[1]:g} K={k[2]:.0e}: {r:.2f}" for r, k in worst[:5]))
+
+
+if __name__ == "__main__":
+    nsat_table(f"{sys.argv[1]}/saturation.csv")
+    merge_table(f"{sys.argv[2]}/saturation_merge_curve.csv")