diff --git a/README.md b/README.md index 798f0c5..952eb95 100644 --- a/README.md +++ b/README.md @@ -65,7 +65,7 @@ machine (no parallel work). Every task writes a completion marker | 06 Convert to PostgreSQL | `convert_postgresql.py` | live PostgreSQL DB (+ dump when pg_dump is available) | implemented | | 07 Convert to RDF | `convert_rdf.py` | `out/rdf/dataset.nt.gz`, `dataset.ttl`, oxigraph store | implemented | | 08 Benchmark queries | `make_queries.py` | `out/bench/queries/`, `out/bench/expected/` | implemented | -| 09 Benchmark execution | `bench_runner.py` | `out/bench/results_raw.csv`, `results_median.csv` | planned | +| 09 Benchmark execution | `bench_runner.py` | `out/bench/results_raw.csv`, `results_median.csv` | implemented | | 10 Extrapolation | `extrapolate.py` | `out/bench/extrapolation.csv`, `hardware_sizing.csv` | planned | | 11 Reporting | `report.py` | `out/report/REPORT.md`, charts, tables | planned | @@ -156,6 +156,7 @@ convert_sqlite.py task 05 entry script convert_postgresql.py task 06 entry script (TRIBO_PG_DSN, TRIBO_PROM_URL) convert_rdf.py task 07 entry script make_queries.py task 08 entry script (canonical results need TRIBO_PG_DSN) +bench_runner.py task 09 entry script (TRIBO_PG_DSN, TRIBO_PROM_URL) requirements.txt closed dependency list (docs/rules/code-python-style.md) ``` diff --git a/bench_runner.py b/bench_runner.py new file mode 100644 index 0000000..eeeea65 --- /dev/null +++ b/bench_runner.py @@ -0,0 +1,527 @@ +"""Task 09 - Benchmark execution and resource measurement. + +Runs the 35-cell matrix (Q1-Q7 x 5 formats) in randomized order. Every +cell: 1 warm-up run (recorded as run 0, excluded from statistics) + 3 +measured runs, each in its OWN subprocess sampled with psutil every 50 ms +for peak RSS / CPU time / read bytes. Result rows of every run are +validated against out/bench/expected/ (tolerance 1e-9) and recorded as +checksum_ok. Writes results_raw.csv, results_median.csv, PG EXPLAIN plans, +PG host/server metrics, the archival storage sizes, an environment record, +and the 09.ok marker. Spec: docs/specs/09_benchmark_execution.md; protocol: +docs/rules/bench-methodology.md. + +Environment adaptations (bench-methodology section 6): +- the cold-cache matrix pass is SKIPPED (Windows host, no page-cache drop); +- PostgreSQL runs on a remote host: no backend-PID psutil sampling; + instead best-effort Prometheus host CPU/RAM windows per PG cell and per + session, plus best-effort pg_stat_statements per-query session deltas; +- on Windows the 50 ms sample reads the OS-maintained peak working set + (peak_wset), so sub-interval children still report a true peak RSS. + +After a run times out, the cell's remaining runs are SKIPPED (never retry +a timeout - bench-methodology section 3); only executed runs appear in the +raw results and the cell reports TIMEOUT. +""" + +from __future__ import annotations + +import csv +import gzip +import logging +import random +import re +import shutil +import statistics +import subprocess +import sys +import tarfile +import time +from dataclasses import dataclass, field +from pathlib import Path + +import psutil +import psycopg + +from common.monitoring import get_prom_url, host_window_metrics +from common.pg import get_dsn +from common.pipeline import ( + REPO_ROOT, + check_dependencies, + load_lab_config, + parse_task_args, + remove_stale_marker, + write_marker, +) +from common.results import compare, parse_rows +from common.storage_sizes import update_storage_sizes + +logger = logging.getLogger(__name__) + +TASK_ID = "09" +DEPENDS_ON = ["08"] + +CELL_TIMEOUT_S = 30 * 60 # hard per-run timeout, spec 09 +SAMPLE_INTERVAL_S = 0.05 # psutil sampling cadence, spec 09 +MEASURED_RUNS = 3 # 1 warm-up + 3 measured per cell, spec 09 + +FORMATS = ("csv", "json", "sqlite", "pg", "rdf") +QUERY_NUMBERS = tuple(range(1, 8)) + +RAW_HEADER = [ + "format", "query", "run", "wall_s", "peak_rss_mb", "cpu_s", + "cpu_util_pct", "read_mb", "rows", "checksum_ok", "status", "exec_order", +] +MEDIAN_HEADER = [ + "format", "query", "wall_s_median", "wall_s_min", "wall_s_max", + "peak_rss_mb_median", "cpu_s_median", "cpu_util_pct_median", + "read_mb_median", "rows", "checksum_ok", "status", +] + +# storage_sizes.csv completeness gate: every (format, variant) that must +# exist after consolidation (working + archival footprints, spec 09) +REQUIRED_SIZE_ROWS = { + ("csv", "tree"), ("csv", "targz"), + ("json", "full"), ("json", "hybrid"), ("json", "targz"), + ("sqlite", "db"), ("sqlite", "db_gz"), + ("pg", "live"), ("pg", "dump"), + ("rdf", "nt_gz"), ("rdf", "ttl"), ("rdf", "store"), +} + + +class CellTimeoutError(Exception): + pass + + +@dataclass +class RunResult: + status: str + wall_s: float = 0.0 + peak_rss_mb: float = 0.0 + cpu_s: float = 0.0 + cpu_util_pct: float = 0.0 + read_mb: float = 0.0 + rows: int | None = None + checksum_ok: bool | None = None + + def raw_row(self, fmt: str, qn: int, run: int, order: int) -> list: + ok = "" if self.checksum_ok is None else str(self.checksum_ok).lower() + return [ + fmt, f"q{qn}", run, round(self.wall_s, 4), round(self.peak_rss_mb, 1), + round(self.cpu_s, 3), round(self.cpu_util_pct, 1), round(self.read_mb, 2), + "" if self.rows is None else self.rows, ok, self.status, order, + ] + + +def _kill_tree(proc: subprocess.Popen) -> None: + """Kill the child and every descendant (the stub's real interpreter).""" + try: + for child in psutil.Process(proc.pid).children(recursive=True): + try: + child.kill() + except psutil.NoSuchProcess: + logger.debug("kill_tree: pid %d already exited", child.pid) + except psutil.Error: + logger.warning("kill_tree: descendants of pid %d not enumerable", proc.pid, exc_info=True) + proc.kill() + proc.wait() + + +def _run_child(cmd: list[str], stdout_path: Path, stderr_path: Path) -> tuple[int, RunResult]: + """Spawn one query subprocess and sample its process TREE until exit. + + The venv python.exe on Windows is a launcher stub: the real interpreter + runs as its child, so sampling only the spawned PID measures the stub + (~4 MB flat, ~0 CPU). Every sample therefore walks the descendant tree + and keeps per-PID last-known CPU / read bytes and the OS peak working + set; totals are sums over all PIDs ever seen. The flat stub adds ~4 MB + to the peak-RSS sum - negligible against real query footprints. + """ + cores = psutil.cpu_count(logical=True) + seen: dict[int, list] = {} # pid -> [cpu_s, read_bytes, peak_rss_bytes] + timed_out = False + start = time.perf_counter() + with open(stdout_path, "wb") as out_fh, open(stderr_path, "wb") as err_fh: + proc = subprocess.Popen(cmd, stdout=out_fh, stderr=err_fh, cwd=REPO_ROOT) + try: + try: + root = psutil.Process(proc.pid) + except psutil.Error: + root = None # child already gone: exited faster than one sample + while root is not None and proc.poll() is None: + try: + procs = [root, *root.children(recursive=True)] + except psutil.Error: + break + for p in procs: + try: + mem = p.memory_info() + cpu = p.cpu_times() + io = p.io_counters() + except psutil.Error: + continue # that process exited between listing and sampling + entry = seen.setdefault(p.pid, [0.0, 0, 0]) + entry[0] = cpu.user + cpu.system + entry[1] = io.read_bytes + entry[2] = max(entry[2], getattr(mem, "peak_wset", 0) or mem.rss) + if time.perf_counter() - start > CELL_TIMEOUT_S: + timed_out = True + break + time.sleep(SAMPLE_INTERVAL_S) + if timed_out: + _kill_tree(proc) + raise CellTimeoutError(f"exceeded {CELL_TIMEOUT_S} s") + returncode = proc.wait() + finally: + if proc.poll() is None: + _kill_tree(proc) + wall = time.perf_counter() - start + cpu_s = sum(entry[0] for entry in seen.values()) + result = RunResult( + status="", wall_s=wall, + peak_rss_mb=sum(entry[2] for entry in seen.values()) / 1048576, + cpu_s=cpu_s, + cpu_util_pct=100.0 * cpu_s / wall / cores if wall > 0 else 0.0, + read_mb=sum(entry[1] for entry in seen.values()) / 1048576, + ) + return returncode, result + + +def _run_cell_once(fmt: str, qn: int, cmd: list[str], tmp_dir: Path, expected: list[tuple]) -> RunResult: + """One benchmark run with pattern C isolation (code-error-handling).""" + stdout_path = tmp_dir / "stdout.txt" + stderr_path = tmp_dir / "stderr.txt" + try: + returncode, result = _run_child(cmd, stdout_path, stderr_path) + except CellTimeoutError: + logger.error("cell %s/q%d timed out after %d s", fmt, qn, CELL_TIMEOUT_S) + return RunResult(status="TIMEOUT", wall_s=CELL_TIMEOUT_S) + except Exception: + logger.error("cell %s/q%d failed to execute", fmt, qn, exc_info=True) + return RunResult(status="ERROR") + if returncode != 0: + tail = stderr_path.read_text(encoding="utf-8", errors="replace")[-2000:] + logger.error("cell %s/q%d exited %d; stderr tail: %s", fmt, qn, returncode, tail) + result.status = "ERROR" + return result + actual = parse_rows(stdout_path.read_text(encoding="ascii")) + diff = compare(expected, actual) + if diff: + logger.error("cell %s/q%d checksum mismatch: %s", fmt, qn, diff) + result.status = "OK" + result.rows = len(actual) + result.checksum_ok = diff is None + return result + + +@dataclass +class CellOutcome: + fmt: str + qn: int + measured: list[RunResult] = field(default_factory=list) + timed_out: bool = False # any run incl. warm-up: remaining runs skipped + window: tuple[float, float] | None = None + + def status(self) -> str: + if self.timed_out or any(r.status == "TIMEOUT" for r in self.measured): + return "TIMEOUT" + if len(self.measured) < MEASURED_RUNS or any(r.status == "ERROR" for r in self.measured): + return "ERROR" + if not all(r.checksum_ok for r in self.measured): + return "INVALID" + return "OK" + + +def _median_row(outcome: CellOutcome) -> list: + status = outcome.status() + fmt, qn, measured = outcome.fmt, outcome.qn, outcome.measured + if status != "OK": + ok = "false" if status == "INVALID" else "" + return [fmt, f"q{qn}", "", "", "", "", "", "", "", "", ok, status] + walls = [r.wall_s for r in measured] + return [ + fmt, f"q{qn}", + round(statistics.median(walls), 4), round(min(walls), 4), round(max(walls), 4), + round(statistics.median(r.peak_rss_mb for r in measured), 1), + round(statistics.median(r.cpu_s for r in measured), 3), + round(statistics.median(r.cpu_util_pct for r in measured), 1), + round(statistics.median(r.read_mb for r in measured), 2), + measured[0].rows, "true", status, + ] + + +def _capture_explain(dsn: str, queries_dir: Path, explain_dir: Path) -> None: + """EXPLAIN (ANALYZE, BUFFERS) per PG query; verify Q5 avoids track_summary.""" + explain_dir.mkdir(parents=True, exist_ok=True) + with psycopg.connect(dsn) as conn: + for qn in QUERY_NUMBERS: + sql = (queries_dir / "pg" / f"q{qn}.sql").read_text(encoding="ascii") + plan = "\n".join(r[0] for r in conn.execute("EXPLAIN (ANALYZE, BUFFERS) " + sql).fetchall()) + (explain_dir / f"q{qn}.txt").write_text(plan + "\n", encoding="utf-8", newline="\n") + if qn == 5 and "track_summary" in plan: + raise RuntimeError("q5 plan touches track_summary - the forced full scan is violated (bench-methodology section 4)") + logger.info("EXPLAIN plans captured to %s", explain_dir) + + +PG_STAT_QUERY = """ +SELECT query, calls, total_exec_time, rows, + shared_blks_hit, shared_blks_read, temp_blks_read, temp_blks_written +FROM pg_stat_statements +""" +PG_STAT_COLUMNS = [ + "calls", "total_exec_time_ms", "rows", + "shared_blks_hit", "shared_blks_read", "temp_blks_read", "temp_blks_written", +] +_QTAG = re.compile(r"^-- Q(\d):") + + +def _pg_statements_snapshot(dsn: str) -> dict[int, list[float]] | None: + """Sum pg_stat_statements counters per benchmark query (matched by the + leading '-- Q:' comment). Best-effort: None when the extension is + unavailable - server-side stats are auxiliary and never fail the task. + """ + try: + sums: dict[int, list[float]] = {} + with psycopg.connect(dsn) as conn: + for query_text, *values in conn.execute(PG_STAT_QUERY): + match = _QTAG.match(query_text) + if not match: + continue + acc = sums.setdefault(int(match.group(1)), [0.0] * len(values)) + for i, v in enumerate(values): + acc[i] += float(v) + return sums + except Exception: + logger.warning("pg_stat_statements snapshot skipped (extension unavailable or not readable)", exc_info=True) + return None + + +def _write_pg_server_stats(path: Path, start: dict[int, list[float]], end: dict[int, list[float]]) -> None: + with open(path, "w", encoding="ascii", newline="") as fh: + writer = csv.writer(fh) + writer.writerow(["query"] + PG_STAT_COLUMNS) + for qn in QUERY_NUMBERS: + before = start.get(qn, [0.0] * len(PG_STAT_COLUMNS)) + after = end.get(qn) + if after is None: + logger.warning("pg_stat_statements: q%d missing in the end snapshot (entry evicted?)", qn) + writer.writerow([f"q{qn}"] + [""] * len(PG_STAT_COLUMNS)) + continue + delta = [after[i] - before[i] for i in range(len(PG_STAT_COLUMNS))] + writer.writerow([f"q{qn}"] + [round(v, 3) for v in delta]) + logger.info("pg server-side statement deltas written to %s", path) + + +def _tree_bytes(root: Path) -> int: + return sum(p.stat().st_size for p in root.rglob("*") if p.is_file()) + + +def _consolidate_storage_sizes(out_root: Path, bench_dir: Path) -> None: + """Build archival variants, record all sizes, verify completeness.""" + archives = bench_dir / "archives" + archives.mkdir(parents=True, exist_ok=True) + + csv_tgz = archives / "csv_tree.tar.gz" + logger.info("archiving out/csv -> %s", csv_tgz) + with tarfile.open(csv_tgz, "w:gz") as tar: + tar.add(out_root / "csv", arcname="csv") + update_storage_sizes(bench_dir, "csv", [ + ("tree", _tree_bytes(out_root / "csv")), ("targz", csv_tgz.stat().st_size), + ]) + + json_tgz = archives / "json.tar.gz" + logger.info("archiving out/json -> %s", json_tgz) + with tarfile.open(json_tgz, "w:gz") as tar: + tar.add(out_root / "json", arcname="json") + update_storage_sizes(bench_dir, "json", [("targz", json_tgz.stat().st_size)]) + + db_gz = archives / "tribo.db.gz" + logger.info("compressing out/sqlite/tribo.db -> %s", db_gz) + with open(out_root / "sqlite" / "tribo.db", "rb") as src, gzip.open(db_gz, "wb") as dst: + shutil.copyfileobj(src, dst, 1048576) + update_storage_sizes(bench_dir, "sqlite", [("db_gz", db_gz.stat().st_size)]) + + # pg_dump -Fc is internally gzip-compressed: it IS the pg archival variant + update_storage_sizes(bench_dir, "pg", [("dump", (out_root / "pg" / "tribo.dump").stat().st_size)]) + + present = set() + with open(bench_dir / "storage_sizes.csv", encoding="ascii") as fh: + for row in csv.reader(fh): + present.add((row[0], row[1])) + missing = REQUIRED_SIZE_ROWS - present + if missing: + raise ValueError(f"storage_sizes.csv incomplete after consolidation: missing {sorted(missing)}") + logger.info("storage_sizes.csv consolidated: all %d required rows present", len(REQUIRED_SIZE_ROWS)) + + +def _environment_record(path: Path, dsn: str) -> None: + import platform + + lines = [ + f"benchmark_host_os: {platform.platform()}", + f"benchmark_host_cpu: {platform.processor()}", + f"benchmark_host_cores_logical: {psutil.cpu_count(logical=True)}", + f"benchmark_host_cores_physical: {psutil.cpu_count(logical=False)}", + f"benchmark_host_ram_gib: {round(psutil.virtual_memory().total / 1073741824, 1)}", + f"python: {platform.python_version()}", + ] + try: + disks = subprocess.run( + ["powershell", "-NoProfile", "-Command", + "Get-PhysicalDisk | ForEach-Object { $_.FriendlyName + ' / ' + $_.MediaType + ' / ' + [math]::Round($_.Size / 1GB) + ' GB' }"], + capture_output=True, text=True, timeout=60, check=True) + lines.append("benchmark_host_disks: " + "; ".join(ln.strip() for ln in disks.stdout.splitlines() if ln.strip())) + except Exception: + logger.warning("disk info skipped (Get-PhysicalDisk query failed - non-Windows host or no permission)", exc_info=True) + import sqlite3 + + import ijson + import pyoxigraph + + lines += [ + f"sqlite: {sqlite3.sqlite_version}", + f"psycopg: {psycopg.__version__}", + f"pyoxigraph: {pyoxigraph.__version__}", + f"ijson: {ijson.__version__}", + f"psutil: {psutil.__version__}", + ] + with psycopg.connect(dsn) as conn: + pg_version = conn.execute("SELECT version()").fetchone()[0] + lines.append(f"postgresql_server: {pg_version}") + lines.append(f"postgresql_host: {conn.info.host} (remote; 4 cores / 3.8 GiB, see .done/06.ok)") + lines += [ + "cold_cache_pass: skipped (Windows host, no page-cache drop - bench-methodology section 6)", + "pg_backend_sampling: not possible for a remote server; Prometheus windows + pg_stat_statements deltas instead", + ] + path.write_text("\n".join(lines) + "\n", encoding="utf-8", newline="\n") + logger.info("environment record written to %s", path) + + +def _run_matrix(bench_dir: Path, expected: dict[int, list[tuple]], seed: int) -> list[CellOutcome]: + cells = [(fmt, qn) for fmt in FORMATS for qn in QUERY_NUMBERS] + random.Random(f"{seed}-bench-order").shuffle(cells) + tmp_dir = bench_dir / "tmp" + tmp_dir.mkdir(parents=True, exist_ok=True) + outcomes: list[CellOutcome] = [] + with open(bench_dir / "results_raw.csv", "w", encoding="ascii", newline="") as raw_fh: + writer = csv.writer(raw_fh) + writer.writerow(RAW_HEADER) + for order, (fmt, qn) in enumerate(cells, start=1): + cmd = [sys.executable, str(bench_dir / "queries" / fmt / f"q{qn}.py")] + outcome = CellOutcome(fmt, qn) + window_start = time.time() + for run in range(MEASURED_RUNS + 1): # run 0 = warm-up + result = _run_cell_once(fmt, qn, cmd, tmp_dir, expected[qn]) + writer.writerow(result.raw_row(fmt, qn, run, order)) + raw_fh.flush() + if run > 0: + outcome.measured.append(result) + logger.info("cell %s/q%d run %d: %s wall=%.3fs rss=%.1fMB", fmt, qn, run, result.status, result.wall_s, result.peak_rss_mb) + if result.status == "TIMEOUT": + outcome.timed_out = True + logger.error("cell %s/q%d: skipping remaining runs after timeout", fmt, qn) + break + outcome.window = (window_start, time.time()) + outcomes.append(outcome) + status = outcome.status() + walls = [r.wall_s for r in outcome.measured if r.status == "OK"] + detail = f"median {statistics.median(walls):.3f} s" if status == "OK" else status + print(f"[{order}/{len(cells)}] {fmt}/q{qn}: {detail}") + shutil.rmtree(tmp_dir) + return outcomes + + +def _write_pg_host_metrics(path: Path, prom_url: str, outcomes: list[CellOutcome]) -> None: + with open(path, "w", encoding="ascii", newline="") as fh: + writer = csv.writer(fh) + writer.writerow([ + "query", "window_s", "host_cpu_avg_pct", "host_cpu_max_pct", + "host_ram_used_baseline_mb", "host_ram_used_peak_mb", "host_ram_used_delta_mb", + ]) + for outcome in sorted((o for o in outcomes if o.fmt == "pg"), key=lambda o: o.qn): + start, end = outcome.window + metrics = host_window_metrics(prom_url, start, end) + row = [f"q{outcome.qn}", round(end - start, 1)] + if metrics is None: + row += [""] * 5 + else: + row += [ + metrics["host_cpu_avg_pct"], metrics["host_cpu_max_pct"], + metrics["host_ram_used_baseline_mb"], metrics["host_ram_used_peak_mb"], + metrics["host_ram_used_delta_mb"], + ] + writer.writerow(row) + logger.info("per-cell PG host metrics written to %s", path) + + +def main() -> int: + args = parse_task_args("Task 09: execute the Q1-Q7 x 5 formats benchmark matrix") + out_root: Path = args.out_root + bench_dir = out_root / "bench" + try: + check_dependencies(out_root, DEPENDS_ON) + remove_stale_marker(out_root, TASK_ID) + dump_path = out_root / "pg" / "tribo.dump" + if not dump_path.exists(): + raise FileNotFoundError(f"{dump_path} missing - produce it with pg_dump -Fc before task 09 (storage consolidation needs it)") + dsn = get_dsn() + expected = {qn: parse_rows((bench_dir / "expected" / f"q{qn}.rows.csv").read_text(encoding="ascii")) for qn in QUERY_NUMBERS} + seed = load_lab_config(out_root)["seed"] + logger.warning("cold-cache pass skipped (Windows host - bench-methodology section 6)") + _environment_record(bench_dir / "environment.txt", dsn) + + prom_url = get_prom_url() + stats_start = _pg_statements_snapshot(dsn) + session_start = time.time() + outcomes = _run_matrix(bench_dir, expected, seed) + session_end = time.time() + + with open(bench_dir / "results_median.csv", "w", encoding="ascii", newline="") as fh: + writer = csv.writer(fh) + writer.writerow(MEDIAN_HEADER) + for outcome in sorted(outcomes, key=lambda o: (o.fmt, o.qn)): + writer.writerow(_median_row(outcome)) + + # snapshot BEFORE the EXPLAIN pass so its executions cannot pollute the deltas + if stats_start is not None: + stats_end = _pg_statements_snapshot(dsn) + if stats_end is not None: + _write_pg_server_stats(bench_dir / "pg_server_stats.csv", stats_start, stats_end) + _capture_explain(dsn, bench_dir / "queries", bench_dir / "explain") + session_host = None + if prom_url: + _write_pg_host_metrics(bench_dir / "pg_host_metrics.csv", prom_url, outcomes) + session_host = host_window_metrics(prom_url, session_start, session_end) + else: + logger.warning("TRIBO_PROM_URL unset - PG host metrics skipped") + + _consolidate_storage_sizes(out_root, bench_dir) + + by_status: dict[str, int] = {} + for outcome in outcomes: + by_status[outcome.status()] = by_status.get(outcome.status(), 0) + 1 + entries = { + "cells_total": len(outcomes), + "cells_ok": by_status.get("OK", 0), + "cells_invalid": by_status.get("INVALID", 0), + "cells_timeout": by_status.get("TIMEOUT", 0), + "cells_error": by_status.get("ERROR", 0), + "session_wall_s": round(session_end - session_start, 1), + "cold_cache_pass": "skipped (Windows host)", + } + if session_host: + entries.update({f"session_{k}": v for k, v in session_host.items()}) + marker_path = write_marker(out_root, TASK_ID, entries) + except Exception: + logger.critical("task %s failed", TASK_ID, exc_info=True) + return 1 + + print(f"task 09 ok: {entries['cells_ok']}/{entries['cells_total']} cells OK " + f"(invalid={entries['cells_invalid']}, timeout={entries['cells_timeout']}, error={entries['cells_error']}), " + f"session {entries['session_wall_s']} s") + print(f"results: {bench_dir / 'results_raw.csv'}, {bench_dir / 'results_median.csv'}") + print(f"marker: {marker_path}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/common/storage_sizes.py b/common/storage_sizes.py index d01ebe4..ab777b7 100644 --- a/common/storage_sizes.py +++ b/common/storage_sizes.py @@ -1,8 +1,10 @@ -"""Idempotent per-format updates to out/bench/storage_sizes.csv. +"""Idempotent per-(format, variant) updates to out/bench/storage_sizes.csv. Contract (docs/rules/build-pipeline-tasks.md section 4): columns -format,variant,bytes; a task re-run REPLACES its own format's rows and never -touches other formats' rows. +format,variant,bytes; a task re-run REPLACES its own (format, variant) rows +and never touches other rows. Replacement is per variant, not per format: +converters 04-07 own the working-footprint variants while task 09 appends +archival variants of the same formats. """ from __future__ import annotations @@ -25,12 +27,13 @@ def update_storage_sizes(bench_dir: Path, fmt: str, variants: list[tuple[str, in lock = bench_dir / "storage_sizes.lock" fd = _acquire(lock) try: + replaced = {(fmt, variant) for variant, _ in variants} kept: list[str] = [] if path.exists(): lines = path.read_text(encoding="ascii").splitlines() if lines and lines[0] != HEADER: raise ValueError(f"{path}: unexpected header {lines[0]!r}") - kept = [ln for ln in lines[1:] if ln and ln.split(",", 1)[0] != fmt] + kept = [ln for ln in lines[1:] if ln and tuple(ln.split(",")[:2]) not in replaced] rows = kept + [f"{fmt},{variant},{nbytes}" for variant, nbytes in variants] with open(path, "w", encoding="ascii", newline="\n") as fh: fh.write(HEADER + "\n" + "\n".join(rows) + "\n") diff --git a/docs/Initial_Prompt.md b/docs/Initial_Prompt.md new file mode 100644 index 0000000..af10ecf --- /dev/null +++ b/docs/Initial_Prompt.md @@ -0,0 +1,90 @@ +# Prompt: Generate a Multi-File Specification for a Tribology Data Storage Evaluation Pipeline + +You are an AI assistant generating a specification for execution by Anthropic Claude Code +in a Cursor development environment (local filesystem, on-premise only, no cloud). + +## Objective +Produce 12 Markdown files — `00_PLAN.md` plus task files `01`–`11` — that specify a complete, +reproducible pipeline: simulate a tribology laboratory, generate ~150 MB of realistic +measurement data in CSV, convert it into 4 target storage formats, benchmark 7 retrieval +scenarios across all 5 formats with resource metering, extrapolate results to 600 GB+, +and produce a decision report with tables, charts, and weighted scoring. + +## File structure requirements +Each task file must contain: Context/Goal, Dependencies, Input, Processing instructions, +Output (artifacts + completion marker `./out/.done/.ok`). +`00_PLAN.md` must contain: goals, execution environment, a Mermaid dependency flowchart, +parallelism notes, execution order, and global conventions (ISO-8601 UTC timestamps, +SI-explicit column names, fixed random seed 20260711, Python with tabs for indentation). + +## Laboratory configuration (task 01) +- Material: Ti-6Al-4V coupons 10×10×3 mm, Cr adhesion layer, Pt-Au coating sputtered in a + composition gradient; film thickness 0.3–1.1 µm. +- Instruments: RAPID custom high-throughput parallelized 6-probe tribometer (friction); + Bruker TI980 TriboIndenter (nanoindentation); Bruker M4 Tornado micro-XRF (composition + mapping); Kurt J. Lesker PVD 200 sputter-down (deposition); AFM (roughness); optical + profilometry (thickness); SIMTRA (sputter-transport Monte-Carlo simulation). +- Deposition matrix, 4 batches (B721–B724) with gun tilt (20°:0°, 20°:20°, 20°:20°, 0°:20°), + Pt/Au power (150/50, 100/100, 150/50, 50/150 W) and discharge voltages. +- Hierarchy and volumes: batch(×4) → wafer(×3) → coupon(×49, 7×7 grid) = 588 coupons; + 480 friction coupons (4 RAPID runs × 5 plates × 6 probes × 4 coupons/square), + 108 reserve; 2 environments (Lab Air / Dry N2, 2 runs each); 3 replicate tracks per + coupon with counterface rotation → 1,440 tracks × 1,000 cycles = 1.44 M cycle rows; + friction loops (200 points every 100th cycle) → 2.88 M loop points. +- Physical models for simulation: exponential COF run-in with steady-state depending on + Au% and environment; hardness/modulus linear in Au% with noise; log-normal roughness; + Archard wear; Au gradient across wafer position (batch-dependent center: 8/25/10/60 wt%). + +## Process flow diagrams (task 02) — Mermaid +Coupon assembly (clean → sonicate → Cr adhesive → PVD sputter); characterization & +batch assembly (SIMTRA, profilometry, ×49/wafer, ×3/batch, batches 721–724); testing tree +(friction µ_normal/µ_dry_nit, nanoindentation, AFM, XRF); tribometer session sequence +(request → samples → ball holders, 3 counterfaces/holder → Excel test plan → setup → +run → software stop → save avg files → teardown); execution loop (5 plates × 6 probes × +4 coupons × 3 tracks with counterface rotation); data-hierarchy ERD. + +## Data generation (task 03) +Hierarchical CSV file tree mirroring the lab structure (batches, simtra, per-coupon +xrf_map/profilometry/nanoindentation/afm, per-track track_info/cof_vs_cycle/ +friction_loops/wear), streaming generation, MANIFEST.csv with row/byte/sha256, +size guard 120–200 MB. + +## Conversions (tasks 04–07), each recording sizes into storage_sizes.csv +- JSON-LD: vocab + QUDT + PROV-O; two variants — FULL (bulk embedded, per-coupon files, + compact) and HYBRID (metadata only, sourceFile links to CSV). +- SQLite: single file, integer surrogate keys, WITHOUT ROWID bulk tables, precomputed + track_summary, indexes for the benchmark queries, load PRAGMAs, VACUUM/ANALYZE. +- PostgreSQL 16: same logical schema, RANGE partitioning of bulk tables, COPY loading, + b-tree + BRIN indexes, materialized track_summary, pg_dump archive. +- RDF: N-Triples.gz + Turtle serializations; queryable embedded store (oxigraph preferred, + Fuseki/TDB2 fallback); document triple-modeling decisions; expect 25–45 M triples. + +## Benchmarks (tasks 08–09) +Seven scenarios: Q1 single-track COF curve; Q2 steady-state COF at Au=10±0.5 wt%; +Q3 hardness vs COF join; Q4 condition filter (Dry N2, 100 mN, cof>0.20); Q5 mean run-in +per batch aggregating ALL raw cycles (summary tables forbidden); Q6 wear volume vs load +join; Q7 Stribeck-style grouped aggregation. Implement per format (direct path read / +ijson streaming / SQL / SQL+EXPLAIN / SPARQL); identical result sets validated by checksum. +Harness: each cell in a separate subprocess; psutil metering (wall time, peak RSS, CPU +time/utilization, read bytes); 1 warm-up + 3 measured runs, medians; 30-min timeout; +outputs results_raw.csv / results_median.csv. + +## Extrapolation (task 10) +Project storage linearly and query time by access-pattern complexity laws +(O(1)/O(log n)/O(n)/single-threaded vs parallel partitioned scan) from the 150 MB +calibration point to 600 GB, 1.2 TB, 6 TB; flag >1 h IMPRACTICAL, >24 h FAIL; RAM models +per format; hardware sizing and cost table driven by an editable hw_prices.yaml +(June 2026 defaults: DDR5 RDIMM ~$14/GB, enterprise NVMe ~$250/TB, 16-core 1U ~$6k). + +## Reporting (task 11) +Tables (footprint + coefficients vs CSV, measured metrics, projected times with flags, +hardware costs, scoring matrix); matplotlib charts (log scales): footprint, RAM, +per-query bars, degradation curves 150 MB→6 TB, stacked weighted score, cost-vs-performance +scatter. Weighted scoring: search speed ×5 (0–50), RAM economy ×2 (0–20), disk economy +×1 (0–10), max 80; search score from inverse geometric mean of Q1–Q7 projected at 600 GB. +Concise English narrative: executive summary and use-case mapping (analysis, reports, +search/filter, archiving, inter-lab exchange), limitations. + +## Style +Specification prose in English, concise and directive. No implementation code in the +spec — only precise instructions an autonomous coding agent can execute. \ No newline at end of file diff --git a/docs/rules/bench-methodology.md b/docs/rules/bench-methodology.md index 4cf8338..a61d512 100644 --- a/docs/rules/bench-methodology.md +++ b/docs/rules/bench-methodology.md @@ -82,6 +82,15 @@ invalidates everything downstream. pass is therefore SKIPPED and marked as such in results and report - never approximated by a fake (e.g. file re-copy tricks) without a rules update describing the method. +- **PostgreSQL is remote on this setup** (dedicated DB host on the LAN), + so spec 09's backend-PID sampling via `pg_stat_activity` + psutil is + impossible - psutil cannot attribute a remote process. Instead the + runner records (a) client-side child metrics exactly like every other + format (wall times therefore include LAN round-trips - stated in the + report), (b) best-effort Prometheus / node_exporter host CPU/RAM + windows per PG cell and for the whole session (`TRIBO_PROM_URL`), and + (c) best-effort `pg_stat_statements` per-query session deltas + (server-side execution time and buffer traffic). - Quiesce the machine during measured runs (no parallel pipeline tasks, no background loads); the runner runs cells sequentially by design. diff --git a/make_queries.py b/make_queries.py index c039904..43ca2e9 100644 --- a/make_queries.py +++ b/make_queries.py @@ -182,6 +182,55 @@ PY_FORMATS = [ ("rdf", "rdf_queries", '"rdf" / "oxigraph_store"'), ] +SQLITE_WRAPPER = '''"""Benchmark query q{n} for the sqlite format (generated by task 08). + +Executes the adjacent q{n}.sql against out/sqlite/tribo.db (read-only) and +prints the canonical result rows to stdout at full float precision; the +benchmark runner (task 09) captures and validates them against +out/bench/expected/q{n}.rows.csv. +""" +import sqlite3 +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[4] +sys.path.insert(0, str(REPO_ROOT)) + +from common.results import print_rows + +if __name__ == "__main__": + sql = Path(__file__).with_suffix(".sql").read_text(encoding="ascii") + db = REPO_ROOT / "out" / "sqlite" / "tribo.db" + conn = sqlite3.connect(f"file:{{db.as_posix()}}?mode=ro", uri=True) + print_rows([tuple(r) for r in conn.execute(sql).fetchall()]) +''' + +PG_WRAPPER = '''"""Benchmark query q{n} for the pg format (generated by task 08). + +Executes the adjacent q{n}.sql on the PostgreSQL instance addressed by +TRIBO_PG_DSN and prints the canonical result rows to stdout at full float +precision; the benchmark runner (task 09) captures and validates them +against out/bench/expected/q{n}.rows.csv. +""" +import sys +from pathlib import Path + +REPO_ROOT = Path(__file__).resolve().parents[4] +sys.path.insert(0, str(REPO_ROOT)) + +import psycopg + +from common.pg import get_dsn +from common.results import print_rows + +if __name__ == "__main__": + sql = Path(__file__).with_suffix(".sql").read_text(encoding="ascii") + with psycopg.connect(get_dsn()) as conn: + print_rows([tuple(r) for r in conn.execute(sql).fetchall()]) +''' + +SQL_WRAPPERS = {"sqlite": SQLITE_WRAPPER, "pg": PG_WRAPPER} + def write_query_files(bench_dir: Path) -> int: queries_dir = bench_dir / "queries" @@ -198,7 +247,8 @@ def write_query_files(bench_dir: Path) -> int: fmt_dir.mkdir(parents=True, exist_ok=True) for n, text in sql.items(): (fmt_dir / f"q{n}.sql").write_text(text, encoding="ascii", newline="\n") - written += 1 + (fmt_dir / f"q{n}.py").write_text(SQL_WRAPPERS[fmt].format(n=n), encoding="ascii", newline="\n") + written += 2 rdf_dir = queries_dir / "rdf" for n, keys in RDF_SPARQL_FILES.items(): text = "\n\n".join(f"# retrieval query: {key}\n{rdf_queries.SPARQL[key]}" for key in keys)