From f91e79a860d496496960f4aea2c8d9e8548d3b42 Mon Sep 17 00:00:00 2001 From: techysneha14 <343708768+shehatathe@users.noreply.github.com> Date: Sat, 24 Oct 2026 19:29:16 +0200 Subject: [PATCH 1/3] bench: DSB many-join dataset, queries and DuckDB oracle (P2-T141) Registers the DSB benchmark (TPC-DS derived, 52 query templates, up to 28 relations per query) as the dsb suite. The generator and templates are under the TPC EULA v2.2, so they are fetched from a pinned upstream commit and built under $KRISHIV_DATA at run time; the dataset directory holds Parquet, the instantiated queries, the DuckDB answers and a manifest with checksums and the licences. Signed-off-by: techysneha14 <343708768+shehatathe@users.noreply.github.com> --- bench/README.md | 30 ++ bench/src/krishiv_bench/cli.py | 2 +- bench/src/krishiv_bench/datasets.py | 32 ++- bench/src/krishiv_bench/dsb.py | 426 ++++++++++++++++++++++++++++ bench/src/krishiv_bench/runner.py | 28 +- bench/tests/test_dsb.py | 157 ++++++++++ 6 files changed, 668 insertions(+), 7 deletions(-) create mode 100644 bench/src/krishiv_bench/dsb.py create mode 100644 bench/tests/test_dsb.py diff --git a/bench/README.md b/bench/README.md index 0ed185c1d..500e6e9b0 100644 --- a/bench/README.md +++ b/bench/README.md @@ -181,6 +181,35 @@ uv run krishiv-bench report --suite h2o-groupby --size 5gb timings and the dataset digest. `reproduce` and `ab` use the same checksum comparison before their interleaved timing (`abba` supports the TPC suites and ClickBench only). +## Many-join suite (`dsb`, P2-T141) + +DSB (Microsoft; derived from TPC-DS, with correlated and skewed data) replaces JOB/IMDB for the join-enumerator decision: the IMDb data terms are +non-commercial. 52 query templates (`agg`, `mb` multi-block, `spj` classes), many with 8 or more relations. + +```bash +uv run krishiv-datasets dsb --sf 1 # needs git, gcc, make; sf is an integer (dsdgen), single process +uv run krishiv-datasets dsb-verify "$KRISHIV_DATA/dsb/sf1-dbgen-v1" +uv run krishiv-bench run --engine duckdb --suite dsb --sf 1 --dataset dsb@v1 +``` + +* **Licence.** The `microsoft/dsb` repository is MIT, but its generator (`dsdgen`/`dsqgen`, from the TPC-DS tools) and query templates are governed by the + TPC End User License Agreement v2.2 (`code/TPC_DS_EULA.txt`): use on systems you control, no redistribution of the materials except with the EULA, and + performance results must be identified as not comparable to TPC results. This repository therefore contains none of them. The harness checks out the + pinned upstream commit (`dsb.COMMIT`) under `$KRISHIV_DATA/cache/dsb//`, builds the tools there (gcc and make only), and writes the instantiated + queries into the dataset directory. Results are DSB results, not TPC-DS results. Evidence: `research/2026-10-24-join-benchmark-licences.md` in the + internal docs repository; the licences are also recorded in each dataset's `MANIFEST.json`. +* **Dataset** `$KRISHIV_DATA/dsb/sf-dbgen-v1/` (the `-dbgen-` infix is the shared versioned-dataset naming): 24 `.parquet`, `queries/.sql` (ids + `agg_q013`, `mb_q001`, `spj_q013`: class and upstream template number, PostgreSQL dialect, `--seed` 1), `oracle.rows.json.gz` and `MANIFEST.json`. Immutable; + `$KRISHIV_DATA/manifests/dsb-sf-dbgen-v1.sha256` is the sha256sum-style list. +* **Oracle.** The DuckDB answers to every query are stored with the dataset (typed rows). `krishiv-bench run --suite dsb` compares each engine's rows with them + (`answer_check`); the cross-engine comparison in `report` is unchanged. A query DuckDB cannot run is recorded with its error and has no oracle. At SF1, 16 of the 52 queries return no rows and 26 return one (the templates are selective), so the row-for-row check is strongest at larger scales. +* **Checksums.** The manifest holds the SHA-256 of every raw `dsdgen` file (identity independent of DuckDB), of every Parquet file (also tied to the DuckDB version, + recorded; Parquet is written with one thread), an order-independent row hash per table and the SHA-256 of each query and of the oracle file. The upstream C code + reads uninitialised locals, so the tools are built with `-ftrivial-auto-var-init=zero`: without it the data differs between optimisation levels. +* `relation_count_distribution` in the manifest is the number of base-table references per query (a repeated table counts each time). +* Scope: `run` supports the suite; `reproduce`, `abba` and `ab` do not yet. Full scale on the benchmark server: `uv run krishiv-datasets dsb --sf 10` (the + conversion is single-threaded and `dsdgen` is single-process: about 3.5 min per unit on a laptop). + ## Reference engines, two methods (`reproduce`, P2-T156) ```bash uv run krishiv-bench reproduce --suite tpch --sf 10 --reps 3 # tpch | tpcds | clickbench; --engines, --methods ours,theirs @@ -224,6 +253,7 @@ The report never prints "all agree" if any check was skipped: skipped checks are `tpch-dbgen` (`--dataset tpch-dbgen`); runs on `tpch` list the answer check as not performed (cross-engine comparison still runs). - **TPC-DS** SF via DuckDB `tpcds` extension (`dsdgen`) -> Parquet (SF1 took ~1 min). Harness query files for all 99 are vendored, but only TPC-H has been run end-to-end / verified; no TPC-DS expected-row files yet. +- **DSB**: `krishiv-datasets dsb --sf N` (see the many-join suite above). - **ClickBench**: `gen.sh clickbench --yes` downloads `hits.parquet` (~14 GB) or `--partitioned` (100 files). Not downloaded. - **Nexmark**: simplified Python generator (person/auction/bid, default 2/6/92% mix, hot-key skew) producing batch Parquet. E05/E08 may need the reference generator or a Rust port (planned in P2); this one is for smoke/relative use. diff --git a/bench/src/krishiv_bench/cli.py b/bench/src/krishiv_bench/cli.py index 06120b773..ca013985c 100644 --- a/bench/src/krishiv_bench/cli.py +++ b/bench/src/krishiv_bench/cli.py @@ -80,7 +80,7 @@ def main(argv=None) -> None: sub = ap.add_subparsers(dest="cmd", required=True) r = sub.add_parser("run", help="run a suite on an engine, write JSON to $KRISHIV_DATA/results") r.add_argument("--engine", required=True, choices=[*ENGINES, "spark"]) - r.add_argument("--suite", default="tpch", choices=["tpch", "tpcds", "memcap", *H2O_SUITES]) + r.add_argument("--suite", default="tpch", choices=["tpch", "tpcds", "dsb", "memcap", *H2O_SUITES]) add_h2o_args(r) r.add_argument("--sf", type=float, default=1) r.add_argument( diff --git a/bench/src/krishiv_bench/datasets.py b/bench/src/krishiv_bench/datasets.py index 861bf8d86..483b9af80 100644 --- a/bench/src/krishiv_bench/datasets.py +++ b/bench/src/krishiv_bench/datasets.py @@ -1,4 +1,4 @@ -"""Dataset generators (P0-T18). CLI: `uv run krishiv-datasets `. +"""Dataset generators (P0-T18). CLI: `uv run krishiv-datasets `. All output goes to $KRISHIV_DATA//sf/ as Parquet, plus a sha256 manifest in $KRISHIV_DATA/manifests/. Nothing here is ever written inside the repo. @@ -282,6 +282,7 @@ def main(argv=None) -> None: p.add_argument("--keys", type=int, default=1_000_000) p.add_argument("--alpha", type=float, default=1.1) p.add_argument("--seed", type=int, default=0) + from .dsb import DEFAULT_SEED as DEFAULT_QUERY_SEED from .h2o import DEFAULT_KEYS, SIZES from .h2o_data import GENERATOR_VERSION, UPSTREAM_SEED @@ -298,6 +299,16 @@ def main(argv=None) -> None: "h2o-verify", help="recompute a generated h2o dataset's content digests against its MANIFEST.json" ) p.add_argument("directory", type=Path) + p = sub.add_parser( + "dsb", + help="DSB (many-join TPC-DS derivative; TPC EULA tools fetched at run time): data, queries, DuckDB oracle", + ) + p.add_argument("--sf", type=int, default=1, help="integer scale (dsdgen), about 1 GB of raw data per unit") + p.add_argument("--seed", type=int, default=DEFAULT_QUERY_SEED, help="dsqgen seed for the query parameters") + p.add_argument("--threads", type=int, default=1, help="DuckDB threads for the Parquet conversion (1: byte-stable)") + p.add_argument("--no-oracle", action="store_true", help="skip the DuckDB answers") + p = sub.add_parser("dsb-verify", help="recompute a generated DSB dataset's checksums against its MANIFEST.json") + p.add_argument("directory", type=Path) p = sub.add_parser("memcap", help="memcap suite data (seeded star schema) under $KRISHIV_DATA/memcap/v1/") p.add_argument("--rows", nargs="+", default=["10m"], metavar="N", help="whole millions, e.g. 10m 50m 150m") p = sub.add_parser("manifest", help="(re)write sha256 manifest for a dataset dir") @@ -308,6 +319,25 @@ def main(argv=None) -> None: t0 = time.time() if a.cmd in ("h2o-groupby", "h2o-join"): print(gen_h2o(a.cmd.removeprefix("h2o-"), a)) + elif a.cmd == "dsb": + from . import dsb + + try: + d = dsb.generate(a.sf, a.seed, a.threads, oracle=not a.no_oracle) + except (ValueError, FileExistsError) as e: + sys.exit(str(e)) + print(d) + print("manifest:", write_manifest(d, f"{d.parent.name}-{d.name}")) + man = dsb.read_manifest(d) + assert man is not None + print("relations per query (count: queries):", man["relation_count_distribution"]) + elif a.cmd == "dsb-verify": + from . import dsb + + problems = dsb.verify(a.directory) + if problems: + sys.exit("\n".join(problems)) + print(f"{a.directory}: verified") elif a.cmd == "h2o-verify": from . import h2o_data diff --git a/bench/src/krishiv_bench/dsb.py b/bench/src/krishiv_bench/dsb.py new file mode 100644 index 000000000..23b83316f --- /dev/null +++ b/bench/src/krishiv_bench/dsb.py @@ -0,0 +1,426 @@ +"""DSB (Microsoft, derived from TPC-DS): a many-join, skewed, correlated decision-support benchmark (P2-T141). + +Used for the join-enumerator decision (P2-T113) in place of JOB/IMDB, whose data terms are non-commercial. + +Licences (evidence in ``krishiv-internal/research/2026-10-24-join-benchmark-licences.md``): + +* the ``microsoft/dsb`` repository is MIT, but its generator source (``code/tools``, derived from the TPC-DS tools) and + the query templates carry the TPC "Legal Notice" and ship with the TPC End User License Agreement v2.2 + (``code/TPC_DS_EULA.txt``): a restricted, revocable licence to use the software on systems under the user's control; + redistribution only with the EULA and legend; performance results must be "clearly identified as not being + comparable to TPC Benchmark Results"; +* so this repository contains **none** of it. ``fetch_source`` checks out the pinned upstream commit into + ``$KRISHIV_DATA/cache/dsb//``, ``build_tools`` compiles ``dsdgen`` / ``dsqgen`` there (gcc and make only: the + repository ships the generated parser sources), and the query texts are instantiated at generation time into the + dataset directory. Nothing is redistributed; results are DSB results, not TPC-DS results. + +Dataset: ``$KRISHIV_DATA/dsb/sf-dbgen-/`` (the ``dsb@`` dataset name, see ``paths``): +``
.parquet`` for the 24 tables, ``queries/.sql``, ``oracle.rows.json.gz`` (DuckDB's answers) and +``MANIFEST.json`` (licences, source commit, build flags, checksums). A dataset is immutable: it is written under a +temporary name and renamed into place, and an existing one is never touched. + +Determinism: ``dsdgen`` has a fixed default seed. The upstream C code reads uninitialised locals, so its output depends +on the compiler unless those are zeroed; the tools are built with ``-ftrivial-auto-var-init=zero`` (GCC >= 12), +which makes -O0 and -O2 builds produce identical bytes. The manifest records the SHA-256 of every raw ``.dat`` file (the +dataset identity, independent of DuckDB), the SHA-256 of every Parquet file (it also depends on the DuckDB version, +recorded) and an order-independent row hash per table. +""" + +from __future__ import annotations + +import gzip +import hashlib +import json +import os +import re +import shutil +import subprocess +import sys +import time +from pathlib import Path +from typing import Any + +from . import compare +from .adapters import Query +from .paths import data_root, dataset_dir + +REPO_URL = "https://github.com/microsoft/dsb" +COMMIT = "ec9a156ceee923db1114cbe388f6183b53d49787" +GENERATOR_VERSION = "v1" +DATASET_NAME = f"dsb@{GENERATOR_VERSION}" +DEFAULT_SEED = 1 +LICENCE = { + "benchmark": "DSB (Microsoft), derived from TPC-DS; not comparable to published TPC-DS results", + "repository": "MIT (https://github.com/microsoft/dsb/blob/main/LICENSE)", + "generator_and_templates": "TPC End User License Agreement v2.2 (code/TPC_DS_EULA.txt in the DSB repository)", + "redistribution": "none: generator, templates and queries are fetched at run time, never stored in this repo", + "evidence": "krishiv-internal/research/2026-10-24-join-benchmark-licences.md", +} +# gcc 15 rejects the K&R-isms of the upstream sources; the zeroing flag is what makes the output reproducible. +TOOL_CFLAGS = ( + "-O2 -w -std=gnu99 -fcommon -ftrivial-auto-var-init=zero -U_FORTIFY_SOURCE -D_FORTIFY_SOURCE=0 " + "-Wno-implicit-function-declaration -Wno-implicit-int -Wno-int-conversion -Wno-incompatible-pointer-types" +) +TEMPLATE_DIRS = {"agg": "agg_queries", "mb": "multi_block_queries", "spj": "spj_queries"} +MANIFEST = "MANIFEST.json" +ORACLE = "oracle.rows.json.gz" +DIALECT = "postgres" +_TYPES = { + "char": "VARCHAR", + "varchar": "VARCHAR", + "integer": "INTEGER", + "int": "INTEGER", + "date": "DATE", + "time": "TIME", +} + + +# ---------------------------------------------------------------- source and tools +def cache_dir() -> Path: + return data_root() / "cache" / "dsb" / COMMIT + + +def _run(cmd: list[str], cwd: Path | None = None) -> None: + subprocess.run(cmd, cwd=cwd, check=True) + + +def fetch_source() -> Path: + """The pinned upstream commit, checked out under the cache; verified by commit hash (content addressed).""" + src = cache_dir() / "src" + if (src / ".git").is_dir(): + head = subprocess.run(["git", "-C", str(src), "rev-parse", "HEAD"], capture_output=True, text=True).stdout + if head.strip() == COMMIT: + return src + shutil.rmtree(src) + src.mkdir(parents=True, exist_ok=True) + _run(["git", "init", "-q"], src) + _run(["git", "fetch", "-q", "--depth", "1", REPO_URL, COMMIT], src) + _run(["git", "checkout", "-q", "FETCH_HEAD"], src) + head = subprocess.run(["git", "-C", str(src), "rev-parse", "HEAD"], capture_output=True, text=True).stdout.strip() + if head != COMMIT: + raise RuntimeError(f"fetched {head}, expected {COMMIT}") + return src + + +def build_tools(src: Path) -> Path: + """Builds ``dsdgen`` and ``dsqgen`` from ``/code/tools``; returns the build directory holding both.""" + build = cache_dir() / "build" + if (build / "dsdgen").is_file() and (build / "dsqgen").is_file(): + return build + for tool in ("gcc", "make"): + if shutil.which(tool) is None: + sys.exit(f"`{tool}` not found: building the DSB generator needs gcc and make") + shutil.rmtree(build, ignore_errors=True) + shutil.copytree(src / "code" / "tools", build) + # The parser sources are generated and shipped; make would rerun yacc/lex (not installed) + # if the grammar looked newer. + now = time.time() + for name in ("qgen.y", "tokenizer.l"): + os.utime(build / name, (now - 3600, now - 3600)) + for name in ("y.tab.c", "y.tab.h", "tokenizer.c", "lex.yy.c"): + if (build / name).exists(): + os.utime(build / name, (now, now)) + cmd = ["make", "-f", "Makefile.suite", "OS=LINUX", f"LINUX_CFLAGS={TOOL_CFLAGS}", "dsdgen", "dsqgen"] + subprocess.run(cmd, cwd=build, check=True, stdout=subprocess.DEVNULL) + if not (build / "dsdgen").is_file() or not (build / "dsqgen").is_file(): + raise RuntimeError("DSB tool build produced no dsdgen/dsqgen") + return build + + +# ---------------------------------------------------------------- schema and conversion +def parse_schema(ddl: str) -> dict[str, list[tuple[str, str]]]: + """``create table`` statements of the upstream DDL as table -> [(column, DuckDB type)] (dbgen_version excluded).""" + schema: dict[str, list[tuple[str, str]]] = {} + for m in re.finditer(r"create\s+table\s+(\w+)\s*\((.*?)\n\)\s*;", ddl, re.S | re.I): + name, body = m.group(1).lower(), m.group(2) + if name == "dbgen_version": + continue + cols = [] + for line in body.splitlines(): + line = line.strip().rstrip(",") + if not line or line.lower().startswith("primary key"): + continue + t = re.match(r"(\w+)\s+(\w+)\s*(\(\s*\d+\s*(?:,\s*\d+\s*)?\))?", line) + if t is None: + raise ValueError(f"cannot parse column {line!r} of {name}") + col, typ, args = t.group(1).lower(), t.group(2).lower(), (t.group(3) or "").replace(" ", "") + if typ == "decimal": + duck = f"DECIMAL{args}" + elif typ in _TYPES: + duck = _TYPES[typ] + else: + raise ValueError(f"unsupported type {typ!r} in {name}.{col}") + cols.append((col, duck)) + schema[name] = cols + return schema + + +def sha256_file(path: Path) -> str: + h = hashlib.sha256() + with path.open("rb") as f: + while chunk := f.read(1 << 20): + h.update(chunk) + return h.hexdigest() + + +def _sql_str(p: Path | str) -> str: + return str(p).replace("'", "''") + + +def convert_table(dat: Path, cols: list[tuple[str, str]], out: Path, threads: int) -> dict[str, Any]: + """``dat`` (pipe separated, no trailing delimiter, Latin-1, empty = NULL) -> Parquet; returns the table's record.""" + import duckdb + + con = duckdb.connect() + try: + con.execute(f"SET threads={threads}") + spec = ", ".join(f"'{c}': '{t}'" for c, t in cols) + con.execute( + f"COPY (SELECT * FROM read_csv('{_sql_str(dat)}', delim='|', header=false, quote='', escape='', " + f"encoding='latin-1', columns={{{spec}}})) TO '{_sql_str(out)}' " + "(FORMAT PARQUET, COMPRESSION ZSTD, ROW_GROUP_SIZE 1000000)" + ) + rows, rowhash = con.execute( + f"SELECT count(*), coalesce(sum(h::HUGEINT), 0)::VARCHAR FROM " + f"(SELECT hash(t) AS h FROM read_parquet('{_sql_str(out)}') t)" + ).fetchone() # ty: ignore[not-iterable] + finally: + con.close() + return {"file": out.name, "rows": rows, "row_hash_sum": rowhash, "file_sha256": sha256_file(out)} + + +# ---------------------------------------------------------------- queries +def template_files(src: Path) -> dict[str, Path]: + """Query id -> template path, e.g. ``agg_q013``, ``mb_q001``, ``spj_q013`` (the number is upstream's).""" + out: dict[str, Path] = {} + for cls, d in TEMPLATE_DIRS.items(): + for f in sorted((src / "query_templates_pg" / d).glob("query*.tpl")): + num = re.match(r"query(\d+)", f.name) + assert num is not None, f.name + out[f"{cls}_q{num.group(1)}"] = f + return out + + +def generate_queries(build: Path, src: Path, scale: int, seed: int, out: Path) -> dict[str, str]: + """Instantiates every template once with ``dsqgen`` (PostgreSQL dialect, fixed seed); returns id -> SQL.""" + out.mkdir(parents=True, exist_ok=True) + texts: dict[str, str] = {} + for qid, tpl in template_files(src).items(): + work = out / qid + work.mkdir(exist_ok=True) + cmd = [ + str(build / "dsqgen"), + "-DIRECTORY", str(tpl.parent), + "-TEMPLATE", tpl.name, + "-DIALECT", DIALECT, + "-OUTPUT_DIR", str(work), + "-SCALE", str(scale), + "-RNGSEED", str(seed), + "-STREAMS", "1", + "-QUIET", "Y", + ] # fmt: skip + subprocess.run(cmd, cwd=build, check=True, stdout=subprocess.DEVNULL) + text = (work / "query_0.sql").read_text().strip() + if not text: + raise RuntimeError(f"dsqgen produced no text for {tpl.name}") + texts[qid] = text + "\n" + shutil.rmtree(work) + return texts + + +def relation_count(sql: str) -> int | None: + """Base-table references in the statement, subqueries and repeated aliases included (the join-order search size).""" + import duckdb + + try: + ser = duckdb.connect().execute("SELECT json_serialize_sql(?)", [sql]).fetchone()[0] # ty: ignore[not-subscriptable] + except duckdb.Error: + return None + if '"error":true' in ser: # json_serialize_sql reports parse errors in the JSON, it does not raise + return None + return len(re.findall(r'"type":"BASE_TABLE"', ser)) + + +def count_distribution(counts: dict[str, int | None]) -> dict[int, int]: + dist: dict[int, int] = {} + for n in counts.values(): + if n is not None: + dist[n] = dist.get(n, 0) + 1 + return dict(sorted(dist.items())) + + +# ---------------------------------------------------------------- oracle +def run_oracle(root: Path, queries: dict[str, str], tables: list[str], timeout_s: float = 600) -> dict[str, Any]: + """Runs every query in DuckDB over the dataset's Parquet and stores the answers as ``oracle.rows.json.gz``.""" + import duckdb + + con = duckdb.connect() + for t in tables: + con.execute(f"CREATE VIEW {t} AS SELECT * FROM read_parquet('{_sql_str(root / (t + '.parquet'))}')") + rows_by_q: dict[str, list] = {} + record: dict[str, Any] = {} + for qid, sql in queries.items(): + t0 = time.perf_counter() + timer = None + try: + import threading + + timer = threading.Timer(timeout_s, con.interrupt) + timer.start() + rows = con.execute(sql).fetchall() + except duckdb.Error as e: + record[qid] = {"error": str(e).splitlines()[0][:300]} + print(f"[dsb] oracle {qid}: ERROR {record[qid]['error'][:100]}", flush=True) + continue + finally: + if timer is not None: + timer.cancel() + rows_by_q[qid] = rows + # order-independent: DuckDB may return rows without a top-level ORDER BY, or ties, in any order + blob = "\n".join(sorted(json.dumps([compare.enc(c) for c in r]) for r in rows)).encode() + record[qid] = { + "rows": len(rows), + "rows_sha256": hashlib.sha256(blob).hexdigest(), + "seconds": round(time.perf_counter() - t0, 3), + } + con.close() + save_oracle(root / ORACLE, rows_by_q) + return record + + +def save_oracle(path: Path, rows_by_query: dict[str, list]) -> None: + """``compare.save_rows`` format, gzip header without a timestamp so the file's checksum is reproducible.""" + payload = json.dumps({q: [[compare.enc(c) for c in r] for r in rows] for q, rows in rows_by_query.items()}) + with path.open("wb") as raw, gzip.GzipFile(filename="", fileobj=raw, mode="wb", mtime=0) as f: + f.write(payload.encode()) + + +def load_oracle(root: Path) -> dict[str, list[tuple]]: + f = root / ORACLE + return compare.load_rows(f) if f.is_file() else {} + + +# ---------------------------------------------------------------- generation +def generate(scale: int, seed: int = DEFAULT_SEED, threads: int = 1, oracle: bool = True, force: bool = False) -> Path: + """Generates ``dsb@`` at ``scale`` (an integer, as ``dsdgen`` requires) and returns its directory.""" + if scale < 1: + raise ValueError("DSB scale must be an integer >= 1 (dsdgen)") + final = dataset_dir(DATASET_NAME, scale) + if final.exists() and any(final.iterdir()): + if not force: + print(f"[dsb] {final} exists, skipping (datasets are immutable)") + return final + raise FileExistsError(f"{final} already exists; datasets are immutable (use a new GENERATOR_VERSION)") + src = fetch_source() + build = build_tools(src) + schema = parse_schema((src / "scripts" / "create_tables.sql").read_text()) + final.parent.mkdir(parents=True, exist_ok=True) + tmp = final.parent / f".{final.name}.partial-{os.getpid()}" + shutil.rmtree(tmp, ignore_errors=True) + raw = tmp / "raw" + raw.mkdir(parents=True) + try: + print(f"[dsb] dsdgen -scale {scale} (single process, minutes per GB)", flush=True) + subprocess.run( + [str(build / "dsdgen"), "-scale", str(scale), "-dir", str(raw), "-terminate", "N", "-force", "-quiet", "Y"], + cwd=build, check=True, stdout=subprocess.DEVNULL, + ) # fmt: skip + tables: dict[str, Any] = {} + raw_sha: dict[str, str] = {} + for f in sorted(raw.iterdir()): + if f.name != "dbgen_version.dat": # holds the generation timestamp + raw_sha[f.name] = sha256_file(f) + for name, cols in schema.items(): + dat = raw / f"{name}.dat" + if not dat.is_file(): + raise RuntimeError(f"dsdgen did not write {dat.name}") + print(f"[dsb] parquet {name}", flush=True) + tables[name] = { + **convert_table(dat, cols, tmp / f"{name}.parquet", threads), + "raw_sha256": raw_sha[dat.name], + } + shutil.rmtree(raw) + qdir = tmp / "queries" + texts = generate_queries(build, src, scale, seed, tmp / ".qgen") + shutil.rmtree(tmp / ".qgen") + qdir.mkdir() + for qid, text in texts.items(): + (qdir / f"{qid}.sql").write_text(text) + counts = {qid: relation_count(sql) for qid, sql in texts.items()} + queries = { + qid: {"sha256": hashlib.sha256(texts[qid].encode()).hexdigest(), "relations": counts[qid]} for qid in texts + } + oracle_rec: dict[str, Any] = {} + if oracle: + print(f"[dsb] oracle: {len(texts)} queries in DuckDB", flush=True) + oracle_rec = run_oracle(tmp, texts, list(schema)) + for qid, rec in oracle_rec.items(): + queries[qid]["oracle"] = rec + import duckdb + + manifest = { + "schema": 1, + "dataset": "dsb", + "generator_version": GENERATOR_VERSION, + "scale": scale, + "query_seed": seed, + "licence": LICENCE, + "source": {"repo": REPO_URL, "commit": COMMIT, "tool_cflags": TOOL_CFLAGS, "dialect": DIALECT}, + "libs": {"duckdb": duckdb.__version__}, + "parquet_threads": threads, + "tables": tables, + "queries": queries, + "relation_count_distribution": count_distribution(counts), + "oracle_file": ORACLE if oracle else None, + "oracle_file_sha256": sha256_file(tmp / ORACLE) if oracle else None, + } + (tmp / MANIFEST).write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n") + if final.exists(): + final.rmdir() + tmp.rename(final) + except BaseException: + shutil.rmtree(tmp, ignore_errors=True) + raise + return final + + +def read_manifest(directory: Path) -> dict[str, Any] | None: + f = directory / MANIFEST + return json.loads(f.read_text()) if f.is_file() else None + + +def verify(directory: Path) -> list[str]: + """Recomputes file checksums against the manifest; returns the problems found (empty = verified).""" + man = read_manifest(directory) + if man is None: + return [f"{directory}: no {MANIFEST}"] + problems = [] + for name, rec in man["tables"].items(): + f = directory / rec["file"] + if not f.is_file(): + problems.append(f"{name}: {f} is missing") + elif sha256_file(f) != rec["file_sha256"]: + problems.append(f"{name}: Parquet sha256 differs from the manifest") + for qid, rec in man["queries"].items(): + f = directory / "queries" / f"{qid}.sql" + if not f.is_file(): + problems.append(f"{qid}: query file is missing") + elif hashlib.sha256(f.read_bytes()).hexdigest() != rec["sha256"]: + problems.append(f"{qid}: query text differs from the manifest") + if man.get("oracle_file"): + f = directory / man["oracle_file"] + if not f.is_file(): + problems.append(f"{f.name} is missing") + elif sha256_file(f) != man["oracle_file_sha256"]: + problems.append(f"{f.name}: sha256 differs from the manifest") + return problems + + +# ---------------------------------------------------------------- harness hooks +def load_queries(root: Path, only: list[str] | None = None) -> list[Query]: + """The dataset's instantiated queries (``queries/.sql``), in id order.""" + d = root / "queries" + files = sorted(d.glob("*.sql")) + if not files: + raise SystemExit(f"no DSB queries under {d}; run `uv run krishiv-datasets dsb --sf `") + return [Query(f.stem, f.read_text(), "dsb") for f in files if not only or f.stem in only] diff --git a/bench/src/krishiv_bench/runner.py b/bench/src/krishiv_bench/runner.py index b304f5928..8713b306f 100644 --- a/bench/src/krishiv_bench/runner.py +++ b/bench/src/krishiv_bench/runner.py @@ -22,7 +22,7 @@ from .errors import error_fields from .paths import bench_root, data_root, dataset_dir, fmt_sf, manifest_stem, queries_dir, results_dir, split_dataset -SUITE_DATASET = {"tpch": "tpch", "tpcds": "tpcds"} # default dataset per suite +SUITE_DATASET = {"tpch": "tpch", "tpcds": "tpcds", "dsb": "dsb@v1"} # default dataset per suite ANSWER_DATASETS = {"tpch-dbgen"} # datasets whose data reproduces the official TPC-H answers (base name; see paths) LEGACY_ANSWER_VERSION = {"tpch-dbgen": "1.5.6"} # unversioned `tpch-dbgen` was generated by DuckDB 1.5.6's dbgen @@ -99,7 +99,9 @@ def dataset_provenance(ds_name: str, sf: float) -> dict: return prov -def load_queries(suite: str, engine: str, only: list[str] | None = None) -> list[Query]: +def load_queries( + suite: str, engine: str, only: list[str] | None = None, dataset_root: Path | None = None +) -> list[Query]: """bench/queries//qNN.sql; an engine-specific override lives in bench/queries///qNN.sql. ClickBench (CC BY-NC-SA) is not stored here: its queries are fetched from a pinned upstream commit. @@ -111,6 +113,12 @@ def load_queries(suite: str, engine: str, only: list[str] | None = None) -> list from . import h2o return h2o.load_queries(suite, engine, only) + if suite == "dsb": # TPC-licensed templates: instantiated at dataset generation, stored with the dataset + from . import dsb + + if dataset_root is None: + raise SystemExit("the dsb suite needs its dataset directory (--sf N, generated by `krishiv-datasets dsb`)") + return dsb.load_queries(dataset_root, only) if suite == "clickbench": # licensed CC BY-NC-SA: fetched from a pinned upstream commit, never stored here texts = clickbench_queries.load(engine) return [Query(q, sql, suite) for q, sql in texts.items() if not only or q in only] @@ -167,6 +175,7 @@ def run_suite( eng: Engine = make_engine(engine_name, **(engine_kwargs or {})) expected = load_expected_rows(suite, sf) answers = {} + oracle_rows: dict[str, list[tuple]] = {} # dsb: DuckDB's typed answers base, version = split_dataset(ds_name) if suite == "tpch" and base in ANSWER_DATASETS: # Official answers come from the running DuckDB's tpch extension, which matches only its own dbgen output: @@ -185,7 +194,11 @@ def run_suite( except Exception as e: # e.g. extension not installable offline print(f"WARNING: official answers unavailable: {e}") ds = Dataset(ds_name, sf, root, tables) - queries = load_queries(suite, engine_name, only) + queries = load_queries(suite, engine_name, only, root) + if suite == "dsb": + from . import dsb + + oracle_rows = dsb.load_oracle(root) started = dt.datetime.now(dt.UTC) sf_s = fmt_sf(sf) @@ -207,7 +220,7 @@ def run_suite( "iterations": iterations, "warmup": warmup, "timeout_s": timeout_s, - "answers_available": bool(answers), + "answers_available": bool(answers or oracle_rows), "queries": {}, } rows_store: dict[str, list] = {} @@ -264,7 +277,10 @@ def run_suite( cpu_seconds_median=_med([m.cpu_seconds for m in meas]), spill_bytes=_max([m.spill_bytes for m in meas]), ) - if q.id in answers: + if q.id in oracle_rows: + ok, msg = compare.rows_match(rows, oracle_rows[q.id], ordered, sql=q.sql) + entry["answer_check"] = {"status": "pass" if ok else "fail", "detail": msg} + elif q.id in answers: ok, msg = compare.answer_check(rows, answers[q.id], ordered) entry["answer_check"] = {"status": "pass" if ok else "fail", "detail": msg} else: @@ -272,6 +288,8 @@ def run_suite( "status": "not_checked", "detail": f"dataset {ds_name!r} does not reproduce the official answers (use tpch-dbgen)" if suite == "tpch" + else "no oracle answer for this query" + if suite == "dsb" else "no official answer set", } if len(rows) <= MAX_STORED_ROWS: diff --git a/bench/tests/test_dsb.py b/bench/tests/test_dsb.py new file mode 100644 index 000000000..6e91e6937 --- /dev/null +++ b/bench/tests/test_dsb.py @@ -0,0 +1,157 @@ +"""DSB dataset: schema parsing, query ids, relation counts, manifest verification, harness hooks (no generator).""" + +from __future__ import annotations + +import hashlib +import json +from pathlib import Path + +import pytest + +from krishiv_bench import compare, dsb, runner + +DDL = """ +-- comment +create table dbgen_version +( + dv_version varchar(16) , + dv_create_date date +); + +create table warehouse +( + w_warehouse_sk integer not null, + w_warehouse_name varchar(20) , + w_state char(2) , + w_gmt_offset decimal(5,2) , + w_opened date , + w_at time , + primary key (w_warehouse_sk) +); +""" + + +def test_parse_schema_maps_types_and_skips_dbgen_version(): + schema = dsb.parse_schema(DDL) + assert list(schema) == ["warehouse"] + assert schema["warehouse"] == [ + ("w_warehouse_sk", "INTEGER"), + ("w_warehouse_name", "VARCHAR"), + ("w_state", "VARCHAR"), + ("w_gmt_offset", "DECIMAL(5,2)"), + ("w_opened", "DATE"), + ("w_at", "TIME"), + ] + + +def test_parse_schema_rejects_unknown_types(): + with pytest.raises(ValueError, match="unsupported type"): + dsb.parse_schema("create table t\n(\n a geometry\n);") + + +def test_template_files_ids(tmp_path: Path): + for d in dsb.TEMPLATE_DIRS.values(): + (tmp_path / "query_templates_pg" / d).mkdir(parents=True) + (tmp_path / "query_templates_pg" / d / "postgres.tpl").write_text("") + (tmp_path / "query_templates_pg" / "agg_queries" / "query013.tpl").write_text("") + (tmp_path / "query_templates_pg" / "spj_queries" / "query013_spj.tpl").write_text("") + (tmp_path / "query_templates_pg" / "multi_block_queries" / "query001.tpl").write_text("") + assert list(dsb.template_files(tmp_path)) == ["agg_q013", "mb_q001", "spj_q013"] + + +def test_relation_count_counts_every_base_table_reference(): + sql = "select 1 from a join b on a.x = b.x join a a2 on a2.x = b.x where a.y in (select y from c)" + assert dsb.relation_count(sql) == 4 + assert dsb.relation_count("select 1") == 0 + assert dsb.relation_count("this is not sql") is None + + +def test_count_distribution_ignores_unparsed(): + assert dsb.count_distribution({"a": 3, "b": 3, "c": 10, "d": None}) == {3: 2, 10: 1} + + +def make_dataset(root: Path) -> dict: + (root / "queries").mkdir(parents=True) + (root / "queries" / "agg_q013.sql").write_text("select 1;\n") + (root / "t.parquet").write_bytes(b"parquet") + dsb.save_oracle(root / dsb.ORACLE, {"agg_q013": [(1,)]}) + man = { + "tables": {"t": {"file": "t.parquet", "file_sha256": hashlib.sha256(b"parquet").hexdigest()}}, + "queries": {"agg_q013": {"sha256": hashlib.sha256(b"select 1;\n").hexdigest()}}, + "oracle_file": dsb.ORACLE, + "oracle_file_sha256": dsb.sha256_file(root / dsb.ORACLE), + } + (root / dsb.MANIFEST).write_text(json.dumps(man)) + return man + + +def test_verify_accepts_a_consistent_dataset_and_reports_each_corruption(tmp_path: Path): + make_dataset(tmp_path) + assert dsb.verify(tmp_path) == [] + (tmp_path / "t.parquet").write_bytes(b"other") + (tmp_path / "queries" / "agg_q013.sql").write_text("select 2;\n") + (tmp_path / dsb.ORACLE).unlink() + problems = dsb.verify(tmp_path) + assert len(problems) == 3 + assert any("Parquet sha256" in p for p in problems) + assert any("query text" in p for p in problems) + assert any("missing" in p for p in problems) + + +def test_verify_without_manifest(tmp_path: Path): + assert "no MANIFEST.json" in dsb.verify(tmp_path)[0] + + +def test_load_queries_and_oracle(tmp_path: Path): + make_dataset(tmp_path) + (tmp_path / "queries" / "mb_q001.sql").write_text("select 2;\n") + assert [q.id for q in dsb.load_queries(tmp_path)] == ["agg_q013", "mb_q001"] + assert [q.id for q in dsb.load_queries(tmp_path, ["mb_q001"])] == ["mb_q001"] + assert dsb.load_oracle(tmp_path) == {"agg_q013": [(1,)]} + assert dsb.load_oracle(tmp_path / "nowhere") == {} + with pytest.raises(SystemExit, match="no DSB queries"): + dsb.load_queries(tmp_path / "nowhere") + + +def test_runner_loads_dsb_queries_from_the_dataset_only(tmp_path: Path): + make_dataset(tmp_path) + assert [q.suite for q in runner.load_queries("dsb", "duckdb", None, tmp_path)] == ["dsb"] + with pytest.raises(SystemExit, match="dataset directory"): + runner.load_queries("dsb", "duckdb") + + +def test_manifest_licence_names_the_tpc_terms(): + assert "TPC End User License Agreement" in dsb.LICENCE["generator_and_templates"] + assert "not comparable" in dsb.LICENCE["benchmark"] + + +def test_convert_table_roundtrip(tmp_path: Path): + import pyarrow.parquet as pq + + dat = tmp_path / "w.dat" + dat.write_bytes("1|caf\xe9|2.50||\n2||0.00|2001-02-03|04:05:06\n".encode("latin-1")) + cols = [ + ("a", "INTEGER"), + ("b", "VARCHAR"), + ("c", "DECIMAL(5,2)"), + ("d", "DATE"), + ("e", "TIME"), + ] + rec = dsb.convert_table(dat, cols, tmp_path / "w.parquet", threads=1) + assert rec["rows"] == 2 + t = pq.read_table(tmp_path / "w.parquet").to_pylist() + assert t[0]["b"] == "café" + assert t[0]["d"] is None + assert t[1]["b"] is None + assert str(t[1]["d"]) == "2001-02-03" + assert rec["file_sha256"] == dsb.sha256_file(tmp_path / "w.parquet") + assert rec["row_hash_sum"].lstrip("-").isdigit() + + +def test_oracle_file_is_reproducible_and_loadable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + rows = {"q": [(1, "a", None)]} + dsb.save_oracle(tmp_path / "a.gz", rows) + monkeypatch.setattr("time.time", lambda: 4e9) + dsb.save_oracle(tmp_path / "b.gz", rows) + assert (tmp_path / "a.gz").read_bytes() == (tmp_path / "b.gz").read_bytes() + assert compare.load_rows(tmp_path / "a.gz") == rows From 901b75ad84daddc762380280dd11b7ae397d5d7c Mon Sep 17 00:00:00 2001 From: techysneha14 <343708768+shehatathe@users.noreply.github.com> Date: Sun, 25 Oct 2026 06:37:44 +0100 Subject: [PATCH 2/3] bench: key DSB tool cache on flags and compiler, fix typos, strengthen roundtrip test Signed-off-by: techysneha14 <343708768+shehatathe@users.noreply.github.com> --- bench/src/krishiv_bench/dsb.py | 24 +++++++++++++++++++----- bench/tests/test_dsb.py | 31 ++++++++++++++++++++++--------- 2 files changed, 41 insertions(+), 14 deletions(-) diff --git a/bench/src/krishiv_bench/dsb.py b/bench/src/krishiv_bench/dsb.py index 23b83316f..f2dbfed98 100644 --- a/bench/src/krishiv_bench/dsb.py +++ b/bench/src/krishiv_bench/dsb.py @@ -102,14 +102,27 @@ def fetch_source() -> Path: return src +def _build_stamp() -> dict[str, str]: + """Current tool-build configuration: flags plus the compiler version (cache identity).""" + out = subprocess.run(["gcc", "--version"], capture_output=True, text=True, check=True).stdout + gcc = out.splitlines()[0].strip() if out.splitlines() else "unknown" + return {"gcc": gcc, "tool_cflags": TOOL_CFLAGS} + + def build_tools(src: Path) -> Path: """Builds ``dsdgen`` and ``dsqgen`` from ``/code/tools``; returns the build directory holding both.""" build = cache_dir() / "build" - if (build / "dsdgen").is_file() and (build / "dsqgen").is_file(): - return build + stamp = build / ".krishiv-build-stamp.json" for tool in ("gcc", "make"): if shutil.which(tool) is None: sys.exit(f"`{tool}` not found: building the DSB generator needs gcc and make") + if (build / "dsdgen").is_file() and (build / "dsqgen").is_file(): + try: + recorded = json.loads(stamp.read_text()) + except (OSError, ValueError): + recorded = None + if recorded == _build_stamp(): + return build shutil.rmtree(build, ignore_errors=True) shutil.copytree(src / "code" / "tools", build) # The parser sources are generated and shipped; make would rerun yacc/lex (not installed) @@ -124,6 +137,7 @@ def build_tools(src: Path) -> Path: subprocess.run(cmd, cwd=build, check=True, stdout=subprocess.DEVNULL) if not (build / "dsdgen").is_file() or not (build / "dsqgen").is_file(): raise RuntimeError("DSB tool build produced no dsdgen/dsqgen") + stamp.write_text(json.dumps(_build_stamp(), sort_keys=True) + "\n") return build @@ -233,12 +247,12 @@ def relation_count(sql: str) -> int | None: import duckdb try: - ser = duckdb.connect().execute("SELECT json_serialize_sql(?)", [sql]).fetchone()[0] # ty: ignore[not-subscriptable] + plan_json = duckdb.connect().execute("SELECT json_serialize_sql(?)", [sql]).fetchone()[0] # ty: ignore[not-subscriptable] except duckdb.Error: return None - if '"error":true' in ser: # json_serialize_sql reports parse errors in the JSON, it does not raise + if '"error":true' in plan_json: # json_serialize_sql reports parse errors in the JSON, it does not raise return None - return len(re.findall(r'"type":"BASE_TABLE"', ser)) + return len(re.findall(r'"type":"BASE_TABLE"', plan_json)) def count_distribution(counts: dict[str, int | None]) -> dict[int, int]: diff --git a/bench/tests/test_dsb.py b/bench/tests/test_dsb.py index 6e91e6937..a3d633317 100644 --- a/bench/tests/test_dsb.py +++ b/bench/tests/test_dsb.py @@ -126,10 +126,14 @@ def test_manifest_licence_names_the_tpc_terms(): def test_convert_table_roundtrip(tmp_path: Path): + import datetime + from decimal import Decimal + + import duckdb import pyarrow.parquet as pq dat = tmp_path / "w.dat" - dat.write_bytes("1|caf\xe9|2.50||\n2||0.00|2001-02-03|04:05:06\n".encode("latin-1")) + dat.write_bytes("1|th\xe9|2.50||\n2||0.00|2001-02-03|04:05:06\n".encode("latin-1")) cols = [ ("a", "INTEGER"), ("b", "VARCHAR"), @@ -137,15 +141,24 @@ def test_convert_table_roundtrip(tmp_path: Path): ("d", "DATE"), ("e", "TIME"), ] - rec = dsb.convert_table(dat, cols, tmp_path / "w.parquet", threads=1) + out = tmp_path / "w.parquet" + rec = dsb.convert_table(dat, cols, out, threads=1) assert rec["rows"] == 2 - t = pq.read_table(tmp_path / "w.parquet").to_pylist() - assert t[0]["b"] == "café" - assert t[0]["d"] is None - assert t[1]["b"] is None - assert str(t[1]["d"]) == "2001-02-03" - assert rec["file_sha256"] == dsb.sha256_file(tmp_path / "w.parquet") - assert rec["row_hash_sum"].lstrip("-").isdigit() + assert rec["file"] == "w.parquet" + t = sorted(pq.read_table(out).to_pylist(), key=lambda r: r["a"]) + assert t == [ + {"a": 1, "b": "thé", "c": Decimal("2.50"), "d": None, "e": None}, + {"a": 2, "b": None, "c": Decimal("0.00"), "d": datetime.date(2001, 2, 3), "e": datetime.time(4, 5, 6)}, + ] + assert rec["file_sha256"] == dsb.sha256_file(out) + expected = ( + duckdb.connect() + .execute( + f"SELECT coalesce(sum(h::HUGEINT), 0)::VARCHAR FROM (SELECT hash(t) AS h FROM read_parquet('{out}') t)" + ) + .fetchone()[0] + ) + assert rec["row_hash_sum"] == expected def test_oracle_file_is_reproducible_and_loadable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): From f0deaf75e43d2e7d4dad0d3f532bd6eb245227e1 Mon Sep 17 00:00:00 2001 From: techysneha14 <343708768+shehatathe@users.noreply.github.com> Date: Sun, 25 Oct 2026 07:37:22 +0100 Subject: [PATCH 3/3] bench: cover dsb harness paths, fix ty diagnostic in test_dsb Fix the ty not-subscriptable diagnostic on the convert_table hash check by asserting the row is not None. Add tests covering fetch_source, build_tools, generate_queries, relation_count error path, run_oracle, generate (validation, reuse, missing-table failure, manifest and oracle paths), verify missing/mismatch branches, the dsb/dsb-verify CLI branches and the dsb oracle check in run_suite. Signed-off-by: techysneha14 <343708768+shehatathe@users.noreply.github.com> --- bench/tests/test_dsb.py | 491 +++++++++++++++++++++++++++++++++++++++- 1 file changed, 488 insertions(+), 3 deletions(-) diff --git a/bench/tests/test_dsb.py b/bench/tests/test_dsb.py index a3d633317..f2b5fefcf 100644 --- a/bench/tests/test_dsb.py +++ b/bench/tests/test_dsb.py @@ -151,14 +151,15 @@ def test_convert_table_roundtrip(tmp_path: Path): {"a": 2, "b": None, "c": Decimal("0.00"), "d": datetime.date(2001, 2, 3), "e": datetime.time(4, 5, 6)}, ] assert rec["file_sha256"] == dsb.sha256_file(out) - expected = ( + row = ( duckdb.connect() .execute( f"SELECT coalesce(sum(h::HUGEINT), 0)::VARCHAR FROM (SELECT hash(t) AS h FROM read_parquet('{out}') t)" ) - .fetchone()[0] + .fetchone() ) - assert rec["row_hash_sum"] == expected + assert row is not None + assert rec["row_hash_sum"] == row[0] def test_oracle_file_is_reproducible_and_loadable(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): @@ -168,3 +169,487 @@ def test_oracle_file_is_reproducible_and_loadable(tmp_path: Path, monkeypatch: p dsb.save_oracle(tmp_path / "b.gz", rows) assert (tmp_path / "a.gz").read_bytes() == (tmp_path / "b.gz").read_bytes() assert compare.load_rows(tmp_path / "a.gz") == rows + + +def test_cache_dir_is_pinned_under_data_root(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("KRISHIV_DATA", str(tmp_path)) + assert dsb.cache_dir() == tmp_path / "cache" / "dsb" / dsb.COMMIT + + +def test_run_invokes_subprocess_with_check(monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + seen: dict = {} + + def _fake(cmd, *args, **kwargs): + seen["cmd"] = cmd + seen["check"] = kwargs.get("check") + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(subprocess, "run", _fake) + dsb._run(["echo", "hi"]) + assert seen == {"cmd": ["echo", "hi"], "check": True} + + +def test_fetch_source_returns_existing_checkout_at_pinned_commit(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + monkeypatch.setattr(dsb, "cache_dir", lambda: tmp_path / "cache") + (tmp_path / "cache" / "src" / ".git").mkdir(parents=True) + seen: list = [] + + def _fake(cmd, *args, **kwargs): + seen.append(cmd) + return SimpleNamespace(stdout=dsb.COMMIT + "\n") + + monkeypatch.setattr(subprocess, "run", _fake) + assert dsb.fetch_source() == tmp_path / "cache" / "src" + assert all("init" not in cmd for cmd in seen) + + +def test_fetch_source_reclones_a_stale_checkout(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + monkeypatch.setattr(dsb, "cache_dir", lambda: tmp_path / "cache") + (tmp_path / "cache" / "src" / ".git").mkdir(parents=True) + calls = {"rev_parse": 0} + + def _fake(cmd, *args, **kwargs): + if "rev-parse" in cmd: + calls["rev_parse"] += 1 + out = "deadbeef\n" if calls["rev_parse"] == 1 else dsb.COMMIT + "\n" + return SimpleNamespace(stdout=out) + return SimpleNamespace(stdout="") + + monkeypatch.setattr(subprocess, "run", _fake) + assert dsb.fetch_source() == tmp_path / "cache" / "src" + assert calls["rev_parse"] == 2 + + +def test_fetch_source_clones_when_absent(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + monkeypatch.setattr(dsb, "cache_dir", lambda: tmp_path / "cache") + seen: list = [] + + def _fake(cmd, *args, **kwargs): + seen.append(cmd) + if "rev-parse" in cmd: + return SimpleNamespace(stdout=dsb.COMMIT + "\n") + return SimpleNamespace(stdout="") + + monkeypatch.setattr(subprocess, "run", _fake) + assert dsb.fetch_source() == tmp_path / "cache" / "src" + kinds = [cmd[1] for cmd in seen if cmd[0] == "git"] + assert "init" in kinds + assert "fetch" in kinds + assert "checkout" in kinds + + +def test_fetch_source_rejects_an_unexpected_commit(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + monkeypatch.setattr(dsb, "cache_dir", lambda: tmp_path / "cache") + monkeypatch.setattr(subprocess, "run", lambda *a, **k: SimpleNamespace(stdout="deadbeef\n")) + with pytest.raises(RuntimeError, match="expected"): + dsb.fetch_source() + + +def test_build_stamp_records_gcc_and_flags(monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + monkeypatch.setattr(subprocess, "run", lambda *a, **k: SimpleNamespace(stdout="gcc (GCC) 15.2.1 20251105\n")) + assert dsb._build_stamp() == {"gcc": "gcc (GCC) 15.2.1 20251105", "tool_cflags": dsb.TOOL_CFLAGS} + + +def test_build_stamp_without_gcc_output(monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + monkeypatch.setattr(subprocess, "run", lambda *a, **k: SimpleNamespace(stdout="")) + assert dsb._build_stamp()["gcc"] == "unknown" + + +def _write_tool_sources(src: Path) -> None: + tools = src / "code" / "tools" + tools.mkdir(parents=True) + (tools / "Makefile.suite").write_text("all:\n") + for name in ("qgen.y", "tokenizer.l", "y.tab.c", "y.tab.h", "tokenizer.c", "lex.yy.c"): + (tools / name).write_text("/* fake */\n") + + +def test_build_tools_reuses_a_matching_stamp(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import shutil + import subprocess + + monkeypatch.setattr(dsb, "cache_dir", lambda: tmp_path) + monkeypatch.setattr(dsb, "_build_stamp", lambda: {"gcc": "gcc-test", "tool_cflags": "flags"}) + monkeypatch.setattr(shutil, "which", lambda *a, **k: "/usr/bin/tool") + build = tmp_path / "build" + build.mkdir(parents=True) + (build / "dsdgen").write_bytes(b"x") + (build / "dsqgen").write_bytes(b"x") + (build / ".krishiv-build-stamp.json").write_text(json.dumps(dsb._build_stamp(), sort_keys=True) + "\n") + + def _boom(*args, **kwargs): + raise AssertionError("a matching stamp must not rebuild") + + monkeypatch.setattr(subprocess, "run", _boom) + assert dsb.build_tools(tmp_path / "src") == build + + +def test_build_tools_requires_gcc_and_make(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import shutil + + monkeypatch.setattr(dsb, "cache_dir", lambda: tmp_path) + monkeypatch.setattr(shutil, "which", lambda *a, **k: None) + with pytest.raises(SystemExit, match="gcc"): + dsb.build_tools(tmp_path / "src") + + +def test_build_tools_rebuilds_on_a_stale_stamp(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import shutil + import subprocess + from types import SimpleNamespace + + monkeypatch.setattr(dsb, "cache_dir", lambda: tmp_path) + monkeypatch.setattr(dsb, "_build_stamp", lambda: {"gcc": "gcc-test", "tool_cflags": "flags"}) + monkeypatch.setattr(shutil, "which", lambda *a, **k: "/usr/bin/tool") + src = tmp_path / "src" + _write_tool_sources(src) + build = tmp_path / "build" + build.mkdir(parents=True) + (build / "dsdgen").write_bytes(b"stale") + (build / "dsqgen").write_bytes(b"stale") + (build / ".krishiv-build-stamp.json").write_text(json.dumps({"gcc": "old", "tool_cflags": "old"}) + "\n") + + def _fake_make(cmd, *args, **kwargs): + assert cmd[0] == "make" + out = Path(kwargs["cwd"]) + (out / "dsdgen").write_bytes(b"new") + (out / "dsqgen").write_bytes(b"new") + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(subprocess, "run", _fake_make) + assert dsb.build_tools(src) == build + assert (build / "dsdgen").read_bytes() == b"new" + assert json.loads((build / ".krishiv-build-stamp.json").read_text()) == dsb._build_stamp() + + +def test_build_tools_rebuilds_on_a_corrupt_stamp(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import shutil + import subprocess + from types import SimpleNamespace + + monkeypatch.setattr(dsb, "cache_dir", lambda: tmp_path) + monkeypatch.setattr(dsb, "_build_stamp", lambda: {"gcc": "gcc-test", "tool_cflags": "flags"}) + monkeypatch.setattr(shutil, "which", lambda *a, **k: "/usr/bin/tool") + src = tmp_path / "src" + _write_tool_sources(src) + build = tmp_path / "build" + build.mkdir(parents=True) + (build / "dsdgen").write_bytes(b"x") + (build / "dsqgen").write_bytes(b"x") + (build / ".krishiv-build-stamp.json").write_text("not json") + + def _fake_make(cmd, *args, **kwargs): + out = Path(kwargs["cwd"]) + (out / "dsdgen").write_bytes(b"x") + (out / "dsqgen").write_bytes(b"x") + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(subprocess, "run", _fake_make) + assert dsb.build_tools(src) == build + + +def test_build_tools_reports_when_make_produces_no_binaries(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import shutil + import subprocess + from types import SimpleNamespace + + monkeypatch.setattr(dsb, "cache_dir", lambda: tmp_path) + monkeypatch.setattr(dsb, "_build_stamp", lambda: {"gcc": "gcc-test", "tool_cflags": "flags"}) + monkeypatch.setattr(shutil, "which", lambda *a, **k: "/usr/bin/tool") + src = tmp_path / "src" + _write_tool_sources(src) + monkeypatch.setattr(subprocess, "run", lambda *a, **k: SimpleNamespace(returncode=0)) + with pytest.raises(RuntimeError, match="produced no"): + dsb.build_tools(src) + + +def test_parse_schema_rejects_an_unparseable_column(): + with pytest.raises(ValueError, match="cannot parse column"): + dsb.parse_schema("create table t\n(\n justaname\n);") + + +def test_generate_queries_instantiates_every_template_once(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + build = tmp_path / "build" + build.mkdir() + src = tmp_path / "src" + agg = src / "query_templates_pg" / "agg_queries" + agg.mkdir(parents=True) + (agg / "query001.tpl").write_text("SELECT 1") + + def _fake_dsqgen(cmd, *args, **kwargs): + assert cmd[0].endswith("dsqgen") + work = Path(cmd[cmd.index("-OUTPUT_DIR") + 1]) + tpl = cmd[cmd.index("-TEMPLATE") + 1] + (work / "query_0.sql").write_text(f"-- {tpl}\nSELECT 1;\n") + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(subprocess, "run", _fake_dsqgen) + assert dsb.generate_queries(build, src, 1, 7, tmp_path / "queries") == {"agg_q001": "-- query001.tpl\nSELECT 1;\n"} + + +def test_generate_queries_rejects_empty_dsqgen_output(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + build = tmp_path / "build" + build.mkdir() + src = tmp_path / "src" + agg = src / "query_templates_pg" / "agg_queries" + agg.mkdir(parents=True) + (agg / "query001.tpl").write_text("SELECT 1") + + def _fake_empty(cmd, *args, **kwargs): + work = Path(cmd[cmd.index("-OUTPUT_DIR") + 1]) + (work / "query_0.sql").write_text(" \n") + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(subprocess, "run", _fake_empty) + with pytest.raises(RuntimeError, match="no text"): + dsb.generate_queries(build, src, 1, 7, tmp_path / "queries") + + +def test_relation_count_returns_none_when_serialization_raises(monkeypatch: pytest.MonkeyPatch): + import duckdb + + def _boom(*args, **kwargs): + raise duckdb.Error("boom") + + monkeypatch.setattr(duckdb, "connect", _boom) + assert dsb.relation_count("select 1") is None + + +def test_run_oracle_records_rows_and_errors(tmp_path: Path): + import duckdb + + root = tmp_path + parquet = root / "t.parquet" + con = duckdb.connect() + con.execute("CREATE TABLE t AS SELECT * FROM (VALUES (1), (2)) v(x)") + con.execute(f"COPY t TO '{parquet}' (FORMAT PARQUET)") + con.close() + record = dsb.run_oracle(root, {"ok": "SELECT x FROM t", "bad": "SELECT * FROM no_such_table_xyz"}, ["t"]) + assert record["ok"]["rows"] == 2 + assert "rows_sha256" in record["ok"] + assert "error" in record["bad"] + assert sorted(dsb.load_oracle(root)["ok"]) == [(1,), (2,)] + + +def _write_dsb_source(src: Path) -> None: + scripts = src / "scripts" + scripts.mkdir(parents=True) + (scripts / "create_tables.sql").write_text("create table t\n(\n a integer\n);") + + +def _fake_generate_queries(texts: dict[str, str]): + def _fake(build, src, scale, seed, out, *args, **kwargs): + out.mkdir(parents=True, exist_ok=True) + return texts + + return _fake + + +def test_generate_rejects_scales_below_one(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + monkeypatch.setenv("KRISHIV_DATA", str(tmp_path)) + with pytest.raises(ValueError, match="integer >= 1"): + dsb.generate(0) + + +def test_generate_reuses_an_existing_dataset_and_refuses_to_overwrite_it( + tmp_path: Path, monkeypatch: pytest.MonkeyPatch +): + monkeypatch.setenv("KRISHIV_DATA", str(tmp_path)) + final = tmp_path / "dsb" / "sf1-dbgen-v1" + final.mkdir(parents=True) + (final / "t.parquet").write_bytes(b"x") + assert dsb.generate(1) == final + with pytest.raises(FileExistsError, match="immutable"): + dsb.generate(1, force=True) + + +def test_generate_fails_when_dsdgen_writes_no_table(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + monkeypatch.setenv("KRISHIV_DATA", str(tmp_path)) + src = tmp_path / "src" + _write_dsb_source(src) + build = tmp_path / "build" + build.mkdir() + monkeypatch.setattr(dsb, "fetch_source", lambda: src) + monkeypatch.setattr(dsb, "build_tools", lambda _src: build) + monkeypatch.setattr(subprocess, "run", lambda *a, **k: SimpleNamespace(returncode=0)) + with pytest.raises(RuntimeError, match="did not write"): + dsb.generate(1) + + +def test_generate_converts_tables_queries_and_manifest(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + monkeypatch.setenv("KRISHIV_DATA", str(tmp_path)) + src = tmp_path / "src" + _write_dsb_source(src) + build = tmp_path / "build" + build.mkdir() + (build / "dsdgen").write_bytes(b"x") + monkeypatch.setattr(dsb, "fetch_source", lambda: src) + monkeypatch.setattr(dsb, "build_tools", lambda _src: build) + + def _fake_dsdgen(cmd, *args, **kwargs): + raw = Path(cmd[cmd.index("-dir") + 1]) + (raw / "t.dat").write_bytes(b"1\n2\n") + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(subprocess, "run", _fake_dsdgen) + monkeypatch.setattr(dsb, "generate_queries", _fake_generate_queries({"agg_q001": "select 1\n"})) + final = tmp_path / "dsb" / "sf1-dbgen-v1" + final.mkdir(parents=True) # empty: generate clears it instead of skipping + out = dsb.generate(1, threads=1, oracle=False) + assert out == final + assert (out / "t.parquet").is_file() + man = dsb.read_manifest(out) + assert man is not None + assert man["tables"]["t"]["rows"] == 2 + assert man["queries"] == {"agg_q001": {"sha256": hashlib.sha256(b"select 1\n").hexdigest(), "relations": 0}} + assert man["relation_count_distribution"] == {"0": 1} + assert man["oracle_file"] is None + + +def test_generate_records_oracle_answers_when_enabled(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import subprocess + from types import SimpleNamespace + + monkeypatch.setenv("KRISHIV_DATA", str(tmp_path)) + src = tmp_path / "src" + _write_dsb_source(src) + build = tmp_path / "build" + build.mkdir() + (build / "dsdgen").write_bytes(b"x") + monkeypatch.setattr(dsb, "fetch_source", lambda: src) + monkeypatch.setattr(dsb, "build_tools", lambda _src: build) + + def _fake_dsdgen(cmd, *args, **kwargs): + raw = Path(cmd[cmd.index("-dir") + 1]) + (raw / "t.dat").write_bytes(b"1\n") + return SimpleNamespace(returncode=0) + + monkeypatch.setattr(subprocess, "run", _fake_dsdgen) + monkeypatch.setattr(dsb, "generate_queries", _fake_generate_queries({"agg_q001": "select 1\n"})) + + def _fake_oracle(root, queries, tables, *args, **kwargs): + dsb.save_oracle(root / dsb.ORACLE, {"agg_q001": [(1,)]}) + return {"agg_q001": {"rows": 1, "rows_sha256": "abc", "seconds": 0.1}} + + monkeypatch.setattr(dsb, "run_oracle", _fake_oracle) + out = dsb.generate(1, threads=1, oracle=True) + man = dsb.read_manifest(out) + assert man is not None + assert man["queries"]["agg_q001"]["oracle"] == {"rows": 1, "rows_sha256": "abc", "seconds": 0.1} + assert man["oracle_file"] == dsb.ORACLE + assert man["oracle_file_sha256"] == dsb.sha256_file(out / dsb.ORACLE) + + +def test_verify_reports_each_missing_file_and_checksum(tmp_path: Path): + make_dataset(tmp_path) + (tmp_path / "t.parquet").unlink() + (tmp_path / "queries" / "agg_q013.sql").unlink() + (tmp_path / dsb.ORACLE).write_bytes(b"changed") + problems = dsb.verify(tmp_path) + assert len(problems) == 3 + assert any("t.parquet" in p and "missing" in p for p in problems) + assert any("query file is missing" in p for p in problems) + assert any("sha256 differs" in p for p in problems) + + +def test_datasets_main_dsb_generates_and_verifies(tmp_path: Path, monkeypatch: pytest.MonkeyPatch, capsys): + from krishiv_bench import datasets + + monkeypatch.setenv("KRISHIV_DATA", str(tmp_path)) + fake = tmp_path / "dsb" / "sf1-dbgen-v1" + fake.mkdir(parents=True) + seen: dict = {} + + def _fake_generate(scale, seed, threads, *args, **kwargs): + seen["args"] = (scale, seed, threads, kwargs.get("oracle")) + return fake + + monkeypatch.setattr("krishiv_bench.dsb.generate", _fake_generate) + monkeypatch.setattr(datasets, "write_manifest", lambda d, stem: tmp_path / f"{stem}.sha256") + monkeypatch.setattr("krishiv_bench.dsb.read_manifest", lambda d: {"relation_count_distribution": {0: 1}}) + datasets.main(["dsb", "--sf", "1"]) + assert seen["args"] == (1, 1, 1, True) + assert "relations per query" in capsys.readouterr().out + + def _boom(*args, **kwargs): + raise ValueError("bad scale") + + monkeypatch.setattr("krishiv_bench.dsb.generate", _boom) + with pytest.raises(SystemExit, match="bad scale"): + datasets.main(["dsb", "--sf", "1"]) + monkeypatch.setattr("krishiv_bench.dsb.verify", lambda d: []) + datasets.main(["dsb-verify", str(fake)]) + assert "verified" in capsys.readouterr().out + monkeypatch.setattr("krishiv_bench.dsb.verify", lambda d: ["boom"]) + with pytest.raises(SystemExit, match="boom"): + datasets.main(["dsb-verify", str(fake)]) + + +def test_run_suite_checks_dsb_oracle_answers(tmp_path: Path, monkeypatch: pytest.MonkeyPatch): + import json + + import duckdb + + from krishiv_bench.adapters.base import Engine + + monkeypatch.setenv("KRISHIV_DATA", str(tmp_path)) + root = tmp_path / "dsb" / "sf1-dbgen-v1" + (root / "queries").mkdir(parents=True) + parquet = root / "t.parquet" + con = duckdb.connect() + con.execute("CREATE TABLE t AS SELECT 1 AS x") + con.execute(f"COPY t TO '{parquet}' (FORMAT PARQUET)") + con.close() + (root / "queries" / "q1.sql").write_text("SELECT x FROM t") + (root / "queries" / "q2.sql").write_text("SELECT x FROM t") + dsb.save_oracle(root / dsb.ORACLE, {"q1": [(1,)]}) + + class _FakeEngine(Engine): + def version(self) -> str: + return "fake" + + def setup(self, dataset) -> None: + pass + + def run(self, query): + return [(1,)] + + monkeypatch.setattr(runner, "make_engine", lambda *a, **k: _FakeEngine()) + out = runner.run_suite("fake", "dsb", 1, iterations=1, warmup=0) + rec = json.loads(out.read_text()) + assert rec["answers_available"] is True + assert rec["queries"]["q1"]["answer_check"]["status"] == "pass" + assert rec["queries"]["q2"]["answer_check"]["status"] == "not_checked" + assert "no oracle answer" in rec["queries"]["q2"]["answer_check"]["detail"]