From a983d2cb72e85de48f28eeb147fb0083e6a21bec Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Thu, 1 Oct 2026 11:04:45 -0400 Subject: [PATCH 1/2] test(python): speed-of-light e2e tests for the article's use cases Five e2e tests, one per use case plus the embeddings one in the plugin suite. Each builds its corpus in a temp dir, grows it by doubling until the fastest native pipeline takes a second, runs native and dirsql three times each, and asserts identical rows and dirsql wall time minus measured launcher startup within 1.1x native. All five are red today at the timing assertion, rows identical: dossier 17x (three CTEs walk the tree three times, one walk ~4x find at 400k files), notebook 1.6x (on-file JSON ingestion), join timed out at 60 s (#1196), transcripts 37x (one parser process per file), embeddings 3.4x (one worker round trip per value). --- packages/python/tests/e2e/speed_of_light.py | 124 +++++++++++ .../python/tests/e2e/use_case_dossier_test.py | 149 +++++++++++++ .../python/tests/e2e/use_case_join_test.py | 114 ++++++++++ .../tests/e2e/use_case_notebook_test.py | 123 +++++++++++ .../tests/e2e/use_case_transcripts_test.py | 199 ++++++++++++++++++ .../tests/e2e/speed_of_light.py | 124 +++++++++++ .../tests/e2e/speed_of_light_test.py | 127 +++++++++++ 7 files changed, 960 insertions(+) create mode 100644 packages/python/tests/e2e/speed_of_light.py create mode 100644 packages/python/tests/e2e/use_case_dossier_test.py create mode 100644 packages/python/tests/e2e/use_case_join_test.py create mode 100644 packages/python/tests/e2e/use_case_notebook_test.py create mode 100644 packages/python/tests/e2e/use_case_transcripts_test.py create mode 100644 plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light.py create mode 100644 plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light_test.py diff --git a/packages/python/tests/e2e/speed_of_light.py b/packages/python/tests/e2e/speed_of_light.py new file mode 100644 index 000000000..66173c7f1 --- /dev/null +++ b/packages/python/tests/e2e/speed_of_light.py @@ -0,0 +1,124 @@ +"""Harness for the use-case speed-of-light tests. + +Each use-case test asks one question of a generated corpus twice: once through +the fastest native pipeline (find, sort, join, awk, a single parser process) +and once through the real `dirsql` launcher. The rows must be identical and +dirsql's wall time, after subtracting the launcher's measured startup, must be +within ten percent of native. The bar is a ratio, never a millisecond number, +so it holds on any machine. + +Startup is the mean of ten `dirsql --help` runs; `--help` and `SELECT 1` cost +the same, so it stands in for the fixed per-invocation cost. A corpus too +small to time is grown by doubling until the native pipeline takes at least a +second, so the comparison is above the noise floor. Each side is run three +times and the fastest run counts. +""" + +from __future__ import annotations + +import json +import shutil +import subprocess +import time + +ONE_SECOND = 1.0 +TOLERANCE = 1.1 +STARTUP_PROBES = 10 +BEST_OF = 3 + + +def cli() -> str: + dirsql = shutil.which("dirsql") + assert dirsql is not None, ( + "`dirsql` console script not on PATH -- run `uv run maturin develop`" + ) + return dirsql + + +def timed(argv, cwd, timeout, env=None): + started = time.perf_counter() + proc = subprocess.run( + argv, + cwd=str(cwd), + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=timeout, + env=env, + check=False, + ) + return proc, time.perf_counter() - started + + +def startup_seconds(argv=None, cwd=None) -> float: + argv = [cli(), "--help"] if argv is None else argv + samples = [] + for _ in range(STARTUP_PROBES): + proc, seconds = timed(argv, cwd or ".", timeout=60) + assert proc.returncode == 0, proc.stderr + samples.append(seconds) + return sum(samples) / len(samples) + + +def dirsql_rows(proc, columns): + assert proc.returncode == 0, ( + f"dirsql failed: stdout={proc.stdout!r} stderr={proc.stderr!r}" + ) + return [tuple(row[column] for column in columns) for row in json.loads(proc.stdout)] + + +def grow_until_native_takes_a_second(build, native, start, ceiling): + """Grow the corpus by doubling until `native` runs for a second. + + `build(lo, hi)` adds items `lo` through `hi - 1` to the corpus already + holding `lo` items, so each doubling writes only the new half. + """ + built, size = 0, start + while True: + build(built, size) + built = size + _, seconds = native() + if seconds >= ONE_SECOND or size >= ceiling: + return size + size *= 2 + + +def best_of(run, name, hopeless=None): + """Run `run` three times; return the rows and wall time of the fastest. + + `run` returns `(rows, seconds)`. A run that raises `TimeoutExpired` fails + the test immediately with the timeout it blew through. A run slower than + `hopeless` seconds ends the series early: noise does not account for an + order of magnitude, so no retry could bring the best under the bar. + """ + fastest = None + for _ in range(BEST_OF): + try: + rows, seconds = run() + except subprocess.TimeoutExpired as expired: + raise AssertionError( + f"{name} did not finish within {expired.timeout:.0f}s" + ) from None + if fastest is None or seconds < fastest[1]: + fastest = (rows, seconds) + if hopeless is not None and seconds > hopeless: + break + return fastest + + +def hopeless_seconds(native_seconds: float, startup: float) -> float: + return startup + 10 * TOLERANCE * native_seconds + + +def dirsql_timeout(native_seconds: float) -> float: + return max(60.0, 30 * native_seconds) + + +def assert_speed_of_light(name, native_seconds, dirsql_seconds, startup): + measured = dirsql_seconds - startup + budget = TOLERANCE * native_seconds + assert measured <= budget, ( + f"{name}: dirsql {dirsql_seconds:.3f}s minus startup {startup:.3f}s" + f" = {measured:.3f}s, over {TOLERANCE}x native {native_seconds:.3f}s" + f" = {budget:.3f}s" + ) diff --git a/packages/python/tests/e2e/use_case_dossier_test.py b/packages/python/tests/e2e/use_case_dossier_test.py new file mode 100644 index 000000000..83ee71f04 --- /dev/null +++ b/packages/python/tests/e2e/use_case_dossier_test.py @@ -0,0 +1,149 @@ +"""Use case 1, the dossier: three selections over a note tree, UNIONed. + +A standing set of background files, the five newest dailies by name, and the +ten most recently modified markdown files outside the dailies, in one ranked +list. Native is three `find` runs with `-printf`, each sorted and cut, then +one final sort. dirsql is the article's query through the real launcher. +No mocks: real console script, real process, real filesystem. +""" + +from __future__ import annotations + +import os +import shutil +import time + +import pytest + +from .speed_of_light import ( + assert_speed_of_light, + best_of, + cli, + dirsql_rows, + dirsql_timeout, + grow_until_native_takes_a_second, + hopeless_seconds, + startup_seconds, + timed, +) + +DAY = 86400 +WINDOW_DAYS = 14 +STANDING = ("situation.md", "projects.md", "profile.md") +COLUMNS = ("rank", "kind", "path", "mtime") + +QUERY = """ +WITH standing AS ( + SELECT path, mtime FROM './' + WHERE dir = 'planner/background' + AND basename IN ('situation.md', 'projects.md', 'profile.md') +), +dailies AS ( + SELECT path, mtime FROM './' + WHERE path LIKE 'vault/Dailies/Daily %' + ORDER BY basename DESC LIMIT 5 +), +recent AS ( + SELECT path, mtime FROM './' + WHERE ext = 'md' + AND path NOT LIKE 'vault/Dailies/%' + AND mtime > strftime('%s', 'now') - 14 * 86400 + ORDER BY mtime DESC LIMIT 10 +) +SELECT 1 AS rank, 'standing' AS kind, path, mtime FROM standing +UNION ALL SELECT 2, 'daily', path, mtime FROM dailies +UNION ALL SELECT 3, 'recent', path, mtime FROM recent +ORDER BY rank, mtime DESC, path +""" + +NATIVE = r""" +set -e +export LC_ALL=C +cutoff=$(( $(date +%s) - 14 * 86400 )) +{ + find planner/background -maxdepth 1 \ + \( -name situation.md -o -name projects.md -o -name profile.md \) \ + -printf '1\tstanding\t%p\t%T@\n' + find vault/Dailies -name 'Daily *' -printf '%f\t%p\t%T@\n' \ + | sort -t "$(printf '\t')" -k1,1r | head -5 \ + | awk -F '\t' -v OFS='\t' '{ print 2, "daily", $2, $3 }' + find . -path ./vault/Dailies -prune -o -name '*.md' -newermt "@$cutoff" \ + -printf '%T@\t%P\n' \ + | sort -rn | head -10 \ + | awk -F '\t' -v OFS='\t' '{ print 3, "recent", $2, $1 }' +} | sort -t "$(printf '\t')" -k1,1n -k4,4rn -k3,3 +""" + + +def _write(path, mtime, body="# note\nbody\n"): + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(body) + os.utime(path, (mtime, mtime)) + + +def build_tree(root, lo, hi): + """Bulk notes `lo` through `hi - 1`: the first thousand inside the recency + window, the rest older, every tenth one a `.txt` the ext filter must + drop. Mtimes are distinct integers nowhere near the window's edge.""" + now = int(time.time()) + if lo == 0: + for name in STANDING: + _write(root / "planner" / "background" / name, now - 70 * DAY) + for day in range(1, 41): + stamp = now - day * DAY + date = time.strftime("%Y.%m.%d", time.gmtime(stamp)) + _write(root / "vault" / "Dailies" / f"Daily {date}.md", stamp) + for i in range(lo, hi): + ext = "txt" if i % 10 == 0 else "md" + mtime = now - 3600 - i * 600 if i < 1000 else now - 15 * DAY - i * 7 + _write( + root / "planner" / "internal" / "notes" / str(i % 1000) / f"note-{i}.{ext}", + mtime, + ) + + +def native_rows(proc): + assert proc.returncode == 0, proc.stderr + rows = [] + for line in proc.stdout.splitlines(): + rank, kind, path, mtime = line.split("\t") + rows.append((int(rank), kind, path, int(float(mtime)))) + return rows + + +def describe_dossier_speed_of_light(): + @pytest.fixture + def root(tmp_path): + tree = tmp_path / "notes" + try: + yield tree + finally: + shutil.rmtree(tree, ignore_errors=True) + + def it_matches_native_rows_within_the_bar(root): + startup = startup_seconds() + + def native(): + proc, seconds = timed(["bash", "-c", NATIVE], root, timeout=600) + return native_rows(proc), seconds + + grow_until_native_takes_a_second( + lambda lo, hi: build_tree(root, lo, hi), native, start=50_000, ceiling=2**21 + ) + expected, native_seconds = best_of(native, "native") + assert len(expected) == 18, expected + + def dirsql(): + proc, seconds = timed( + [cli(), "query", QUERY], + root, + timeout=dirsql_timeout(native_seconds), + ) + return dirsql_rows(proc, COLUMNS), seconds + + actual, dirsql_seconds = best_of( + dirsql, "dirsql", hopeless_seconds(native_seconds, startup) + ) + + assert actual == expected + assert_speed_of_light("dossier", native_seconds, dirsql_seconds, startup) diff --git a/packages/python/tests/e2e/use_case_join_test.py b/packages/python/tests/e2e/use_case_join_test.py new file mode 100644 index 000000000..2698878b0 --- /dev/null +++ b/packages/python/tests/e2e/use_case_join_test.py @@ -0,0 +1,114 @@ +"""Use case 2, the arXiv sidecar layout: a join of two path tables. + +One folder per paper holding `abstract.md` and `title.md`; the question is +every paper that has both, with its title and the abstract's size. Native is +two `find` listings, one through `awk` for the title text, sorted and joined +with `join(1)`. dirsql joins the two path tables on `dir` through the real +launcher. No mocks: real console script, real process, real filesystem. +""" + +from __future__ import annotations + +import shutil + +import pytest + +from .speed_of_light import ( + assert_speed_of_light, + best_of, + cli, + dirsql_rows, + dirsql_timeout, + grow_until_native_takes_a_second, + hopeless_seconds, + startup_seconds, + timed, +) + +COLUMNS = ("dir", "title", "abstract_bytes") +QUERY = """ +SELECT a.dir, t.content AS title, a.size AS abstract_bytes +FROM './papers/*/abstract.md' a +JOIN './papers/*/title.md' t ON t.dir = a.dir +ORDER BY a.dir +""" + +NATIVE = r""" +set -e +export LC_ALL=C +tab="$(printf '\t')" +join -t "$tab" \ + <(find papers -name title.md \ + -exec awk -v OFS='\t' '{ split(FILENAME, p, "/"); print p[1] "/" p[2], $0 }' {} + \ + | sort) \ + <(find papers -name abstract.md -printf '%h\t%s\n' | sort) +""" + +SENTENCE = ( + "We study the problem from first principles and report a method that " + "improves on the baseline across every setting we tried. " +) + + +def build_papers(root, lo, hi): + """Paper folders `lo` through `hi - 1`. Every seventh lacks a title and + every eleventh an abstract, so the join has to drop rows rather than + pair everything.""" + papers = root / "papers" + for i in range(lo, hi): + folder = papers / f"2609.{i:06d}" + folder.mkdir(parents=True) + if i % 7: + (folder / "title.md").write_text(f"Paper {i}: on the {i % 13}th method") + if i % 11: + (folder / "abstract.md").write_text(SENTENCE * (1 + i % 5)) + + +def native_rows(proc): + assert proc.returncode == 0, proc.stderr + rows = [] + for line in proc.stdout.splitlines(): + folder, title, size = line.split("\t") + rows.append((folder, title, int(size))) + return rows + + +def describe_join_speed_of_light(): + @pytest.fixture + def root(tmp_path): + tree = tmp_path / "arxiv" + try: + yield tree + finally: + shutil.rmtree(tree, ignore_errors=True) + + def it_matches_native_rows_within_the_bar(root): + startup = startup_seconds() + + def native(): + proc, seconds = timed(["bash", "-c", NATIVE], root, timeout=600) + return native_rows(proc), seconds + + n = grow_until_native_takes_a_second( + lambda lo, hi: build_papers(root, lo, hi), + native, + start=10_000, + ceiling=2**20, + ) + expected, native_seconds = best_of(native, "native") + assert len(expected) == sum(1 for i in range(n) if i % 7 and i % 11) + + def dirsql(): + proc, seconds = timed( + [cli(), "query", QUERY], + root, + timeout=dirsql_timeout(native_seconds), + ) + return dirsql_rows(proc, COLUMNS), seconds + + actual, dirsql_seconds = best_of( + dirsql, "dirsql", hopeless_seconds(native_seconds, startup) + ) + + assert actual == expected + assert_speed_of_light("join", native_seconds, dirsql_seconds, startup) diff --git a/packages/python/tests/e2e/use_case_notebook_test.py b/packages/python/tests/e2e/use_case_notebook_test.py new file mode 100644 index 000000000..2e9a04ac0 --- /dev/null +++ b/packages/python/tests/e2e/use_case_notebook_test.py @@ -0,0 +1,123 @@ +"""Use case 3, the lab notebook: one file, many rows. + +One markdown file of dated `## YYYY-MM-DD` sections, split into rows by a +parser script. Native is the parser alone piped into a sort-and-head. dirsql +runs the same parser once through `--on-file` and sorts in SQLite. No mocks: +real console script, real process, real filesystem, real parser spawn. +""" + +from __future__ import annotations + +import datetime +import shutil + +import pytest + +from .speed_of_light import ( + assert_speed_of_light, + best_of, + cli, + dirsql_rows, + dirsql_timeout, + grow_until_native_takes_a_second, + hopeless_seconds, + startup_seconds, + timed, +) + +COLUMNS = ("date", "words") +QUERY = "SELECT date, words FROM './notebook.md' ORDER BY date DESC LIMIT 5" + +EXTRACT = """\ +import json, re, sys + +text = open(sys.argv[1], encoding="utf-8").read() +rows = [ + {"date": date, "body": body.strip(), "words": len(body.split())} + for date, body in re.findall(r"^## (\\d{4}-\\d\\d-\\d\\d)\\n(.*?)(?=^## |\\Z)", text, re.S | re.M) +] +print(json.dumps(rows)) +""" + +TOP_FIVE = ( + "import json, sys;" + " rows = json.load(sys.stdin);" + " rows.sort(key=lambda r: r['date'], reverse=True);" + " [print(r['date'], r['words'], sep='\\t') for r in rows[:5]]" +) + +NATIVE = f'python3 extract.py notebook.md | python3 -c "{TOP_FIVE}"' + +WORDS = [ + "ran", + "the", + "cells", + "again", + "and", + "logged", + "every", + "reading", + "before", + "lunch", +] + + +def build_notebook(root, n): + """`n` sections, newest first, one per day back from a fixed date, each + with a distinct word count so the rows carry something to compare.""" + root.mkdir(parents=True, exist_ok=True) + (root / "extract.py").write_text(EXTRACT) + newest = datetime.date(2026, 9, 23) + chunks = ["# Lab notebook\n\nOne file. Newest entry at the top.\n"] + for i in range(n): + day = newest - datetime.timedelta(days=i) + body = " ".join(WORDS[j % len(WORDS)] for j in range(3 + i % 37)) + chunks.append(f"\n## {day.isoformat()}\n\n{body}\n") + (root / "notebook.md").write_text("".join(chunks)) + + +def native_rows(proc): + assert proc.returncode == 0, proc.stderr + rows = [] + for line in proc.stdout.splitlines(): + date, words = line.split("\t") + rows.append((date, int(words))) + return rows + + +def describe_notebook_speed_of_light(): + @pytest.fixture + def root(tmp_path): + tree = tmp_path / "notebook" + try: + yield tree + finally: + shutil.rmtree(tree, ignore_errors=True) + + def it_matches_native_rows_within_the_bar(root): + startup = startup_seconds() + + def native(): + proc, seconds = timed(["sh", "-c", NATIVE], root, timeout=600) + return native_rows(proc), seconds + + grow_until_native_takes_a_second( + lambda _, n: build_notebook(root, n), native, start=20_000, ceiling=2**22 + ) + expected, native_seconds = best_of(native, "native") + assert len(expected) == 5, expected + + def dirsql(): + proc, seconds = timed( + [cli(), "query", QUERY, "--on-file", "python3 extract.py {path}"], + root, + timeout=dirsql_timeout(native_seconds), + ) + return dirsql_rows(proc, COLUMNS), seconds + + actual, dirsql_seconds = best_of( + dirsql, "dirsql", hopeless_seconds(native_seconds, startup) + ) + + assert actual == expected + assert_speed_of_light("notebook", native_seconds, dirsql_seconds, startup) diff --git a/packages/python/tests/e2e/use_case_transcripts_test.py b/packages/python/tests/e2e/use_case_transcripts_test.py new file mode 100644 index 000000000..9f57e2bc4 --- /dev/null +++ b/packages/python/tests/e2e/use_case_transcripts_test.py @@ -0,0 +1,199 @@ +"""Use case 4, the transcripts: many files, many rows. + +A tree of JSONL session logs, each turned into message rows by a parser +named in `.dirsql.toml`; the question is how many messages each project has +per role. Native is one parser process over every file. dirsql spawns the +same parser per file through the real launcher. No mocks: real console +script, real process, real filesystem, real parser spawns. +""" + +from __future__ import annotations + +import json +import shutil + +import pytest + +from .speed_of_light import ( + assert_speed_of_light, + best_of, + cli, + dirsql_rows, + dirsql_timeout, + grow_until_native_takes_a_second, + hopeless_seconds, + startup_seconds, + timed, +) + +COLUMNS = ("project", "role", "n") +QUERY = ( + "SELECT project, role, count(*) AS n FROM messages" + " GROUP BY project, role ORDER BY project, role" +) + +CONFIG = """\ +[[table]] +name = "messages" +glob = "projects/**/*.jsonl" +ddl = "CREATE TABLE messages (project TEXT, kind TEXT, ts TEXT, role TEXT, text TEXT)" +on-file = "python3 messages.py {path}" +""" + +MESSAGES = """\ +import json, sys + +NOISE = ("<", "Another Claude session", "Caveat:", "This session is being continued") + + +def rows(path): + parts = path.split("/") + project = parts[parts.index("projects") + 1] + kind = "subagent" if "subagents" in parts else "main" + out = [] + for line in open(path, encoding="utf-8"): + if not line.strip(): + continue + try: + entry = json.loads(line) + except ValueError: + continue + if entry.get("type") not in ("user", "assistant"): + continue + content = (entry.get("message") or {}).get("content") + if isinstance(content, str): + text = content + else: + text = "\\n".join(b["text"] for b in (content or []) if b.get("type") == "text") + if not text or text.startswith(NOISE): + continue + out.append({"project": project, "kind": kind, "ts": entry.get("timestamp"), "role": entry["type"], "text": text}) + return out + + +if __name__ == "__main__": + print(json.dumps(rows(sys.argv[1]))) +""" + +COUNT = """\ +import os, sys +from collections import Counter +from messages import rows + +counts = Counter() +for folder, _, names in os.walk("projects"): + for name in names: + if name.endswith(".jsonl"): + for row in rows(os.path.join(folder, name)): + counts[(row["project"], row["role"])] += 1 +for (project, role), n in sorted(counts.items()): + print(project, role, n, sep="\\t") +""" + +NATIVE = "python3 count.py" + +LINE_KINDS = ( + { + "type": "user", + "message": {"content": "please run the tests again and show me the output"}, + }, + { + "type": "assistant", + "message": { + "content": [ + {"type": "text", "text": "Running them now."}, + {"type": "tool_use", "name": "Bash"}, + ] + }, + }, + { + "type": "assistant", + "message": {"content": [{"type": "tool_use", "name": "Read"}]}, + }, + { + "type": "user", + "message": {"content": "ignored"}, + }, + {"type": "progress", "data": {"step": 3}}, + { + "type": "assistant", + "message": {"content": [{"type": "text", "text": "Caveat: this is noise"}]}, + }, +) +PADDING = " ".join(["filler"] * 60) + + +def build_transcripts(root, lo, hi): + """Session files `lo` through `hi - 1`, two hundred lines each across + eight projects, one in four under a `subagents/` folder, each line a few + hundred bytes.""" + if lo == 0: + root.mkdir(parents=True) + (root / ".dirsql.toml").write_text(CONFIG) + (root / "messages.py").write_text(MESSAGES) + (root / "count.py").write_text(COUNT) + for i in range(lo, hi): + project = root / "projects" / f"project-{i % 8}" + if i % 4 == 0: + path = project / f"session-{i}" / "subagents" / "agent.jsonl" + else: + path = project / f"session-{i}.jsonl" + path.parent.mkdir(parents=True, exist_ok=True) + lines = [] + for j in range(200): + entry = dict(LINE_KINDS[(i + j) % len(LINE_KINDS)]) + entry["timestamp"] = f"2026-09-{1 + j % 28:02d}T10:{j % 60:02d}:00Z" + entry["uuid"] = f"{i}-{j}" + entry["padding"] = PADDING + lines.append(json.dumps(entry)) + path.write_text("\n".join(lines) + "\n") + + +def native_rows(proc): + assert proc.returncode == 0, proc.stderr + rows = [] + for line in proc.stdout.splitlines(): + project, role, n = line.split("\t") + rows.append((project, role, int(n))) + return rows + + +def describe_transcripts_speed_of_light(): + @pytest.fixture + def root(tmp_path): + tree = tmp_path / "transcripts" + try: + yield tree + finally: + shutil.rmtree(tree, ignore_errors=True) + + def it_matches_native_rows_within_the_bar(root): + startup = startup_seconds() + + def native(): + proc, seconds = timed(["sh", "-c", NATIVE], root, timeout=600) + return native_rows(proc), seconds + + grow_until_native_takes_a_second( + lambda lo, hi: build_transcripts(root, lo, hi), + native, + start=128, + ceiling=2**16, + ) + expected, native_seconds = best_of(native, "native") + assert len(expected) == 16, expected + + def dirsql(): + proc, seconds = timed( + [cli(), "query", QUERY, "-c", ".dirsql.toml"], + root, + timeout=dirsql_timeout(native_seconds), + ) + return dirsql_rows(proc, COLUMNS), seconds + + actual, dirsql_seconds = best_of( + dirsql, "dirsql", hopeless_seconds(native_seconds, startup) + ) + + assert actual == expected + assert_speed_of_light("transcripts", native_seconds, dirsql_seconds, startup) diff --git a/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light.py b/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light.py new file mode 100644 index 000000000..66173c7f1 --- /dev/null +++ b/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light.py @@ -0,0 +1,124 @@ +"""Harness for the use-case speed-of-light tests. + +Each use-case test asks one question of a generated corpus twice: once through +the fastest native pipeline (find, sort, join, awk, a single parser process) +and once through the real `dirsql` launcher. The rows must be identical and +dirsql's wall time, after subtracting the launcher's measured startup, must be +within ten percent of native. The bar is a ratio, never a millisecond number, +so it holds on any machine. + +Startup is the mean of ten `dirsql --help` runs; `--help` and `SELECT 1` cost +the same, so it stands in for the fixed per-invocation cost. A corpus too +small to time is grown by doubling until the native pipeline takes at least a +second, so the comparison is above the noise floor. Each side is run three +times and the fastest run counts. +""" + +from __future__ import annotations + +import json +import shutil +import subprocess +import time + +ONE_SECOND = 1.0 +TOLERANCE = 1.1 +STARTUP_PROBES = 10 +BEST_OF = 3 + + +def cli() -> str: + dirsql = shutil.which("dirsql") + assert dirsql is not None, ( + "`dirsql` console script not on PATH -- run `uv run maturin develop`" + ) + return dirsql + + +def timed(argv, cwd, timeout, env=None): + started = time.perf_counter() + proc = subprocess.run( + argv, + cwd=str(cwd), + stdin=subprocess.DEVNULL, + capture_output=True, + text=True, + timeout=timeout, + env=env, + check=False, + ) + return proc, time.perf_counter() - started + + +def startup_seconds(argv=None, cwd=None) -> float: + argv = [cli(), "--help"] if argv is None else argv + samples = [] + for _ in range(STARTUP_PROBES): + proc, seconds = timed(argv, cwd or ".", timeout=60) + assert proc.returncode == 0, proc.stderr + samples.append(seconds) + return sum(samples) / len(samples) + + +def dirsql_rows(proc, columns): + assert proc.returncode == 0, ( + f"dirsql failed: stdout={proc.stdout!r} stderr={proc.stderr!r}" + ) + return [tuple(row[column] for column in columns) for row in json.loads(proc.stdout)] + + +def grow_until_native_takes_a_second(build, native, start, ceiling): + """Grow the corpus by doubling until `native` runs for a second. + + `build(lo, hi)` adds items `lo` through `hi - 1` to the corpus already + holding `lo` items, so each doubling writes only the new half. + """ + built, size = 0, start + while True: + build(built, size) + built = size + _, seconds = native() + if seconds >= ONE_SECOND or size >= ceiling: + return size + size *= 2 + + +def best_of(run, name, hopeless=None): + """Run `run` three times; return the rows and wall time of the fastest. + + `run` returns `(rows, seconds)`. A run that raises `TimeoutExpired` fails + the test immediately with the timeout it blew through. A run slower than + `hopeless` seconds ends the series early: noise does not account for an + order of magnitude, so no retry could bring the best under the bar. + """ + fastest = None + for _ in range(BEST_OF): + try: + rows, seconds = run() + except subprocess.TimeoutExpired as expired: + raise AssertionError( + f"{name} did not finish within {expired.timeout:.0f}s" + ) from None + if fastest is None or seconds < fastest[1]: + fastest = (rows, seconds) + if hopeless is not None and seconds > hopeless: + break + return fastest + + +def hopeless_seconds(native_seconds: float, startup: float) -> float: + return startup + 10 * TOLERANCE * native_seconds + + +def dirsql_timeout(native_seconds: float) -> float: + return max(60.0, 30 * native_seconds) + + +def assert_speed_of_light(name, native_seconds, dirsql_seconds, startup): + measured = dirsql_seconds - startup + budget = TOLERANCE * native_seconds + assert measured <= budget, ( + f"{name}: dirsql {dirsql_seconds:.3f}s minus startup {startup:.3f}s" + f" = {measured:.3f}s, over {TOLERANCE}x native {native_seconds:.3f}s" + f" = {budget:.3f}s" + ) diff --git a/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light_test.py b/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light_test.py new file mode 100644 index 000000000..c2e4c2636 --- /dev/null +++ b/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light_test.py @@ -0,0 +1,127 @@ +"""Use case 2 through the plugin: embedding every abstract in a corpus. + +The native reference is the model itself: one Python process that loads the +same model2vec model and batch-encodes every file's text. dirsql embeds each +file through `embed()` and the real worker process, cold cache every run. +The vectors must be identical and dirsql's wall time, after subtracting the +launcher's startup, within ten percent of the batch encode. Nothing mocked: +real launcher, real worker spawn, real model2vec inference on the on-disk +model from conftest. +""" + +from __future__ import annotations + +import json +import os +import shutil +import sys +from importlib import resources + +import pytest + +from .speed_of_light import ( + assert_speed_of_light, + best_of, + dirsql_rows, + dirsql_timeout, + grow_until_native_takes_a_second, + hopeless_seconds, + startup_seconds, + timed, +) + +_FRAGMENT = str(resources.files("dirsql_plugin_embeddings").joinpath("dirsql.toml")) +LAUNCHER = [ + sys.executable, + "-c", + "import sys; from dirsql.cli.main import main; sys.exit(main())", + "--no-plugin", +] +COLUMNS = ("path", "emb") + +ENCODE = """\ +import json, os, sys +from model2vec import StaticModel + +model = StaticModel.from_pretrained(sys.argv[1]) +names = sorted(os.listdir("abstracts")) +texts = [open(os.path.join("abstracts", name), encoding="utf-8").read() for name in names] +for name, vector in zip(names, model.encode(texts, show_progress_bar=False)): + print(name, json.dumps([float(c) for c in vector]), sep="\\t") +""" + +VOCABULARY = ("hello", "world") + + +def build_abstracts(root, lo, hi): + """Abstracts `lo` through `hi - 1`, each a different run of the model's + two known words so no two neighbours embed alike.""" + folder = root / "abstracts" + if lo == 0: + folder.mkdir(parents=True) + (root / "encode.py").write_text(ENCODE) + for i in range(lo, hi): + words = [VOCABULARY[(i >> bit) & 1] for bit in range(3 + i % 9)] + (folder / f"{i:07d}.md").write_text(" ".join(words), encoding="utf-8") + + +def native_rows(proc): + assert proc.returncode == 0, proc.stderr + rows = [] + for line in proc.stdout.splitlines(): + name, vector = line.split("\t") + rows.append((name, json.loads(vector))) + return rows + + +def describe_embeddings_speed_of_light(): + @pytest.fixture + def root(tmp_path): + tree = tmp_path / "corpus" + try: + yield tree + finally: + shutil.rmtree(tree, ignore_errors=True) + + def it_matches_the_batch_encode_within_the_bar(root, tiny_model, tmp_path): + startup = startup_seconds([*LAUNCHER, "--help"], cwd=tmp_path) + + def native(): + proc, seconds = timed( + [sys.executable, "encode.py", tiny_model], root, timeout=600 + ) + return native_rows(proc), seconds + + grow_until_native_takes_a_second( + lambda lo, hi: build_abstracts(root, lo, hi), + native, + start=4_096, + ceiling=2**18, + ) + expected, native_seconds = best_of(native, "native") + + query = ( + "SELECT path, embed(content, '" + tiny_model + "') AS emb" + " FROM './abstracts/*.md' ORDER BY path" + ) + cold_runs = iter(range(10)) + + def dirsql(): + cache_home = tmp_path / f"xdg-cache-{next(cold_runs)}" + proc, seconds = timed( + [*LAUNCHER, "query", query, "--config", _FRAGMENT], + root, + timeout=dirsql_timeout(native_seconds), + env={**os.environ, "XDG_CACHE_HOME": str(cache_home)}, + ) + rows = dirsql_rows(proc, COLUMNS) + return [ + (os.path.basename(path), json.loads(emb)) for path, emb in rows + ], seconds + + actual, dirsql_seconds = best_of( + dirsql, "dirsql", hopeless_seconds(native_seconds, startup) + ) + + assert actual == expected + assert_speed_of_light("embeddings", native_seconds, dirsql_seconds, startup) From d58596d21b4358bb7b7ce379dbd1e6683a4d93d0 Mon Sep 17 00:00:00 2001 From: thekevinbot <248217931+thekevinbot@users.noreply.github.com> Date: Thu, 1 Oct 2026 20:13:19 -0400 Subject: [PATCH 2/2] test(python): mark the five speed-of-light tests strict xfail until each gap is closed --- packages/python/tests/e2e/use_case_dossier_test.py | 4 ++++ packages/python/tests/e2e/use_case_join_test.py | 3 +++ packages/python/tests/e2e/use_case_notebook_test.py | 4 ++++ packages/python/tests/e2e/use_case_transcripts_test.py | 1 + .../dirsql-plugin-embeddings/tests/e2e/speed_of_light_test.py | 4 ++++ 5 files changed, 16 insertions(+) diff --git a/packages/python/tests/e2e/use_case_dossier_test.py b/packages/python/tests/e2e/use_case_dossier_test.py index 83ee71f04..c413e0409 100644 --- a/packages/python/tests/e2e/use_case_dossier_test.py +++ b/packages/python/tests/e2e/use_case_dossier_test.py @@ -120,6 +120,10 @@ def root(tmp_path): finally: shutil.rmtree(tree, ignore_errors=True) + @pytest.mark.xfail( + strict=True, + reason="each CTE's path-table reference walks the tree again, and one walk is slower than find", + ) def it_matches_native_rows_within_the_bar(root): startup = startup_seconds() diff --git a/packages/python/tests/e2e/use_case_join_test.py b/packages/python/tests/e2e/use_case_join_test.py index 2698878b0..415269955 100644 --- a/packages/python/tests/e2e/use_case_join_test.py +++ b/packages/python/tests/e2e/use_case_join_test.py @@ -82,6 +82,9 @@ def root(tmp_path): finally: shutil.rmtree(tree, ignore_errors=True) + @pytest.mark.xfail( + strict=True, reason="path-table join is an unindexed nested loop" + ) def it_matches_native_rows_within_the_bar(root): startup = startup_seconds() diff --git a/packages/python/tests/e2e/use_case_notebook_test.py b/packages/python/tests/e2e/use_case_notebook_test.py index 2e9a04ac0..fa394480f 100644 --- a/packages/python/tests/e2e/use_case_notebook_test.py +++ b/packages/python/tests/e2e/use_case_notebook_test.py @@ -94,6 +94,10 @@ def root(tmp_path): finally: shutil.rmtree(tree, ignore_errors=True) + @pytest.mark.xfail( + strict=True, + reason="ingesting the parser's JSON rows is slower than a second interpreter's json.load and sort", + ) def it_matches_native_rows_within_the_bar(root): startup = startup_seconds() diff --git a/packages/python/tests/e2e/use_case_transcripts_test.py b/packages/python/tests/e2e/use_case_transcripts_test.py index 9f57e2bc4..21058d5b2 100644 --- a/packages/python/tests/e2e/use_case_transcripts_test.py +++ b/packages/python/tests/e2e/use_case_transcripts_test.py @@ -167,6 +167,7 @@ def root(tmp_path): finally: shutil.rmtree(tree, ignore_errors=True) + @pytest.mark.xfail(strict=True, reason="on-file spawns one parser process per file") def it_matches_native_rows_within_the_bar(root): startup = startup_seconds() diff --git a/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light_test.py b/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light_test.py index c2e4c2636..5fce8cd4f 100644 --- a/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light_test.py +++ b/plugins/dirsql-plugin-embeddings/tests/e2e/speed_of_light_test.py @@ -83,6 +83,10 @@ def root(tmp_path): finally: shutil.rmtree(tree, ignore_errors=True) + @pytest.mark.xfail( + strict=True, + reason="embed() makes one worker round trip per value instead of a batch", + ) def it_matches_the_batch_encode_within_the_bar(root, tiny_model, tmp_path): startup = startup_seconds([*LAUNCHER, "--help"], cwd=tmp_path)