bench: DSB many-join dataset, queries and DuckDB oracle (P2-T141)

bench/README.md

@@ -196,6 +196,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/<commit>/`, 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<N>-dbgen-v1/` (the `-dbgen-` infix is the shared versioned-dataset naming): 24 `<table>.parquet`, `queries/<id>.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<N>-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
@@ -239,6 +268,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.

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(

bench/src/krishiv_bench/datasets.py

@@ -1,4 +1,4 @@
-"""Dataset generators (P0-T18). CLI: `uv run krishiv-datasets <tpch|tpcds|clickbench|nexmark|zipf|h2o-*|manifest>`.
+"""Dataset generators (P0-T18). CLI: `uv run krishiv-datasets <tpch|tpcds|clickbench|nexmark|zipf|h2o-*|dsb|manifest>`.
 
 All output goes to $KRISHIV_DATA/<dataset>/sf<N>/ 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/<rows>")
     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