benchmarks/collection_query_engine/baseline.py
Rama main · réplica de lectura del repositorio original.
"""Baseline latency, WAL bytes and RSS of the dark Records store (invented data, not a gate).
python benchmarks/collection_query_engine/baseline.py --out baseline.json
"""
from __future__ import annotations
import argparse
import json
import os
import platform
import random
import resource
import sqlite3
import statistics
import subprocess
import sys
import tempfile
import time
import uuid
from pathlib import Path
REPO = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(REPO / "src"))
from exomem.collection_store import connection # noqa: E402
from exomem.collection_store.writer import CollectionWriter # noqa: E402
from exomem.query_engine import legacy, runtime # noqa: E402
from exomem.query_engine import sqlite as qsqlite # noqa: E402
from exomem.query_engine.typed_rows import execute_rows # noqa: E402
from exomem.query_engine.validation import normalize_query # noqa: E402
CID = "2db90f18-70df-4e41-986e-2d7d7db1caca"
PATH = "Knowledge Base/Records/Work/_collection.md"
EVIDENCE = "Knowledge Base/Evidence/import-a.md"
SEED, SAMPLES, BULK_SAMPLES, BATCH = 20261005, 30, 5, 500
STATUSES = ("open", "closed", "blocked", "review")
FIELDS = {"title": "string", "count": "integer", "status": "string", "amount": "integer"}
INDEXES = "indexes:\n" + "".join(f" by_{f}:\n keys: [{{field: {f}}}]\n" for f in FIELDS)
def manifest(indexed: bool) -> str:
flags = ", filterable: true, sortable: true" if indexed else ""
fields = "".join(f" {f}: {{type: {t}{', required: true' if f == 'title' else ''}{flags}}}\n"
for f, t in FIELDS.items())
return (f"---\ntype: collection\nexomem_id: {CID}\ntitle: Work\nsemantic_profile: records\n"
"collection_version: 1\nschema_version: 1\nlifecycle: active\nstorage:\n"
" strategy: markdown-items\n source: Items\n format_version: 1\n"
f"item_schema:\n natural_key: [title]\n fields:\n{fields}{INDEXES if indexed else ''}---\n")
def pct(samples):
ordered = sorted(samples)
return {"p50_ms": round(statistics.median(ordered) * 1e3, 3),
"p95_ms": round(ordered[int(0.95 * (len(ordered) - 1) + 0.5)] * 1e3, 3), "samples": len(ordered)}
def timed(fn, n=SAMPLES):
out = []
for _ in range(n):
start = time.perf_counter()
try:
fn()
except runtime.QueryError as error: # the product's own refusal (200 ms deadline, no ready projection)
return {"refused": str(error).split(":")[0], "after_samples": len(out)}
out.append(time.perf_counter() - start)
return pct(out)
def run(size: int, indexed: bool, scratch: Path) -> dict:
rng = random.Random(SEED + size)
counter = iter(range(10**9))
def row():
n = next(counter)
return {"title": f"Record {n}", "count": rng.randrange(10**6),
"status": rng.choice(STATUSES), "amount": rng.randrange(10**4)}
root = scratch / "vault"
(root / "Knowledge Base").mkdir(parents=True)
(root / "Knowledge Base/log.md").write_text("# Log\n")
(root / EVIDENCE).parent.mkdir(parents=True)
(root / EVIDENCE).write_text("---\ntype: evidence\n---\n\nInvented import.\n")
with connection.open_writer(scratch / "collections.sqlite", lease_check=lambda: True) as handle:
store = CollectionWriter(root, handle)
wal = Path(f"{handle.path}-wal")
store.create_collection(PATH, manifest(indexed), why="baseline", scaffold=False)
guard = {"hash": None}
def container():
return store.inspect_collection(CID)["lifecycle_guards"]["expected_container_hash"]
def bulk(rows):
result = store.bulk_upsert_records(CID, rows=rows, why="baseline", expected_container_hash=guard["hash"],
source=EVIDENCE)
guard["hash"] = result["after_container_hash"]
assert result["counts"]["inserted"] == len(rows), (result["counts"], result["rows"][0])
def append():
result = store.append_record(CID, item=row(), item_key=str(uuid.UUID(int=rng.getrandbits(128), version=4)),
expected_container_hash=guard["hash"], why="baseline")
guard["hash"] = result["after_container_hash"]
guard["hash"] = container()
for _ in range(size // BATCH):
bulk([{"item": row()} for _ in range(BATCH)])
# Autocheckpoint off so the -wal file only grows; its size delta is the bytes the mutation logged.
handle.connection.execute("PRAGMA wal_autocheckpoint=0")
def wal_per_mutation(fn, n=SAMPLES):
deltas = []
for _ in range(n):
before = wal.stat().st_size if wal.exists() else 0
fn()
deltas.append((wal.stat().st_size if wal.exists() else 0) - before)
return {"p50": statistics.median(deltas), "max": max(deltas)}
result = {"append": timed(append), "bulk_500": timed(lambda: bulk([{"item": row()} for _ in range(BATCH)]), BULK_SAMPLES)}
result["wal_bytes_per_append"] = wal_per_mutation(append)
result["wal_bytes_per_bulk_500"] = wal_per_mutation(lambda: bulk([{"item": row()} for _ in range(BATCH)]), 5)
declarations = {CID: {"domain": "collections", "type": "records", "vault": "baseline", "fields": {
"item_key": {"type": "string"}, **{f: {"type": t} for f, t in FIELDS.items()}}}}
typed = normalize_query({"version": 1, "select": ["title"], "order_by": [{"field": "count"}],
"page": {"limit": 50}}, collection=CID, declarations=declarations)
assert not typed.findings, typed.findings
# A read session carries a 200 ms deadline from its opening, so each sample opens its own.
def legacy_run(**kw):
with runtime.read_session(root, handle.path) as session:
admitted = session.admit(CID)
plan = legacy.normalize(columns_available=admitted.fields, **kw)
return qsqlite.execute_legacy(admitted, plan, path="source", format="markdown-items")
def typed_run():
with runtime.read_session(root, handle.path) as session:
return execute_rows(session.admit_query(typed.query, as_of="2026-10-05T00:00:00+00:00"))
result["legacy_page50_sorted"] = timed(lambda: legacy_run(sort_by="count", limit=50))
result["legacy_sum"] = timed(lambda: legacy_run(aggregate="sum:count"))
result["typed_page50_sorted"] = timed(typed_run)
result["rows_at_start"] = size
return result
def host() -> dict:
return {"cpu_count": os.cpu_count(), "os_family": platform.system(), "kernel_release_family": platform.release().split("-")[0],
"machine": platform.machine(), "filesystem": subprocess.run(
["stat", "-f", "-c", "%T", tempfile.gettempdir()], capture_output=True, text=True).stdout.strip()}
def main() -> None:
parser = argparse.ArgumentParser()
parser.add_argument("--out", required=True)
parser.add_argument("--sizes", default="1000,10000", help="comma-separated N")
parser.add_argument("--variants", default="no_index,4_indexes")
args = parser.parse_args()
sha = subprocess.run(["git", "-C", str(REPO), "rev-parse", "--short=9", "HEAD"], capture_output=True, text=True).stdout.strip()
report = {"exomem_git_sha": sha, "python": platform.python_version(), "sqlite": sqlite3.sqlite_version,
"host": host(), "pragmas": {"journal_mode": "wal", "synchronous": "FULL", "foreign_keys": "ON", "wal_autocheckpoint": "0 during runs (set by harness)"}, "seed": SEED, "samples_per_metric": SAMPLES, "bulk_samples": BULK_SAMPLES,
"scope": "dark store, invented data, single run, not a gate",
"wal_method": "wal_autocheckpoint=0; -wal file size delta around one mutation (frames x (page+24) bytes)",
"peak_rss_note": "process high-water mark after each run (cumulative, KiB)", "runs": {}}
for size in map(int, args.sizes.split(",")):
for indexed in (v == "4_indexes" for v in args.variants.split(",")):
started = time.perf_counter()
with tempfile.TemporaryDirectory() as tmp:
data = run(size, indexed, Path(tmp))
data["peak_rss_kib"] = resource.getrusage(resource.RUSAGE_SELF).ru_maxrss
report["runs"][f"N={size},{'4_indexes' if indexed else 'no_index'}"] = data
print(size, indexed, f"{time.perf_counter() - started:.0f}s", json.dumps(data), flush=True)
Path(args.out).write_text(json.dumps(report, indent=2) + "\n")
if __name__ == "__main__":
main()