Add examples/online: matchlab vs Splink, batch cost and online lookups

examples/online/.gitignore

@@ -0,0 +1,7 @@
+# Generated at run time — regenerated by the scripts, never committed.
+warehouse.sqlite
+model.json
+.online_tmp/
+.mem_*.bin
+.mem_*.json
+__pycache__/

examples/online/README.md

@@ -0,0 +1,130 @@
+# matchlab vs pure Splink: expression, cost, and a single online match
+
+This example runs one pipeline two ways over ~100k entities scattered across four sources
+as ~390k records (generated by the neighbouring `examples/companies/benchmark.py`). The
+pipeline is the one from the Ministry of Justice's [*Running Splink in Production*](https://moj-analytical-services.github.io/splink/blog/2026/01/29/running-splink-in-production.html):
+standardise each source, deduplicate each in isolation, link every pair with a shared
+Splink model, and cluster once at the end.
+
+Both sides run that *identical* algorithm. `pipeline_matchlab.py` expresses it as a
+matchlab plan. `pipeline_splink.py` hand-rolls it with Splink and a union-find, the way
+`examples/companies/by_hand.py` is the fair opponent to `with_matchlab.py`. So the
+comparison isolates one thing: what matchlab adds on top of the same work.
+
+Run `python benchmark.py` for the whole story in one table. Figures are from one run on
+the dev container, so read them as orders of magnitude.
+
+## Expression
+
+matchlab maps onto the blog's shape one-to-one: `Source.view(cleaning=)` standardises,
+`.dedupe()` deduplicates each source, a `SplinkLinker` per pair generates cross-dataset
+edges, and one `Resolver` consolidates every edge and clusters once. Both pipelines
+recover the ground truth exactly, 100,000 clusters with no precision or recall loss.
+
+## Time and memory
+
+| batch build | Splink | matchlab (memory) | matchlab (on disk) |
+| --- | ---: | ---: | ---: |
+| time | 5.3 s | 11.7 s | 12.7 s |
+| peak memory | 1,675 MiB | 2,307 MiB | 2,228 MiB |
+
+matchlab takes about 2.2 times the wall clock at about 1.4 times the peak memory. Pointing
+the store at a file rather than memory barely moves either, which the next section
+explains.
+
+## Why matchlab is slower, and where the memory goes
+
+Both sides do the same matching, so the difference is not the matching. Timing the collect
+by phase makes it concrete:
+
+    1.7 s   read each source and content-hash every row into a leaf id, then store
+    2.1 s   dedupe each source, resolve, materialise
+   10.1 s   entity views, the six Splink links, and the final materialise
+    ——
+    4.9 s   of which: the six Splink predicts, shared with the hand-rolled baseline
+
+So about 5 s is the shared matching. matchlab's extra ~6 s is content-addressing and
+writing: hashing every row to a leaf, minting content-addressed cluster ids, building the
+complete `(root, leaf, key, source)` resolution and persisting it. That is the cache and
+the queryable store being built, not the match being computed.
+
+The memory tells the same story from the other side. Putting the store on disk barely
+lowers the peak, because the peak is the transient working set, Splink's own in-memory
+engine on each of the six links plus the intermediate frames, not the persisted tables.
+The store's own footprint is only the ~80 MiB gap between the two matchlab columns. A row
+store helps online serving, below, not the batch build's memory.
+
+## DAG shape, and what the cost buys
+
+The blog shapes its DAG around compute cost and recovery. It slices cross-dataset linking
+into pairwise pieces, clusters once at the end, and keeps training as a separate stage,
+because each run is expensive and has to be restartable. Both pipelines here follow that
+shape. matchlab makes it cheaper to live with.
+
+- Every step is a content-addressed artefact, so re-running an unchanged slice does no
+  work. Recovery and iteration are structural, not something you hand-build with
+  checkpoints and run identifiers.
+- The shape is open. Spine, all-pairs, or coherent subsets are just which `.link()` calls
+  you make, so you tune the topology to the data.
+- The backend is a choice. The blog already splits storage by access pattern, DuckDB for
+  the batch and a row store for the published lookup table teams join against, because
+  point selects are slow on a column store. matchlab makes that split explicit, which is
+  the next section.
+
+## A single online match
+
+Online serving is the blog's published lookup table: join a record's `(source, key)` to
+get its entity, one record at a time. That is a point select, the opposite of the batch
+scan Splink is built for. We time the same single lookup against matchlab's own DuckDB
+store and against the identical resolution in SQLite.
+
+| single online match | latency | throughput |
+| --- | ---: | ---: |
+| DuckDB (matchlab store) | 6.13 ms | 163 /s |
+| SQLite (row store) | 0.011 ms | 93,000 /s |
+
+**A single match in the pure Splink world costs about 6 ms.** Splink returns a frame and
+keeps no lookup table, so serving one match means a point select against wherever you
+persist the result. Its engine is DuckDB, and matchlab's store is that same DuckDB today,
+so both land at ~6 ms.
+
+**A single match in a SQLite-backed matchlab costs about 0.011 ms**, roughly 500 times
+faster. A row store answers a point select with a couple of B-tree reads, where a column
+store pays a per-query scan overhead. This is why Matchbox proper is backed by PostgreSQL:
+a column store for the batch build, a row store for online serving.
+
+## Is it worth it
+
+Clear-eyed, it depends on whether you look at the result again.
+
+- **Pure Splink** gives you the matcher, and a frame. You build the DAG, the checkpoints,
+  the storage, the lookup, and the caching around it, which is what `pipeline_splink.py`
+  is.
+- **matchlab** gives you the same matcher wrapped in a DAG of content-addressed artefacts,
+  so caching, recovery, and iteration come for free, plus a materialised queryable
+  resolution, a `lookup_key` API, and a backend you pick per access pattern.
+
+So it is worth it if you iterate, re-resolve, or serve matches online. The ~2.2× batch
+cost is paid once and buys near-free re-runs and a single online match that is hundreds of
+times cheaper. It is not worth it if you resolve once and never look again, where the
+multiplier buys nothing pure Splink does not already give you.
+
+## Caveats
+
+- Four sources, so linking every pair is six models. At the blog's scale (eight sources,
+  28 pairs) you would group sources into coherent subsets instead.
+- Synthetic data gives perfect quality, so the numbers show the two pipelines agreeing,
+  not a claim about linkage accuracy.
+- We block on name only. The generator derives postcodes from the entity id, so blocking
+  on them explodes the candidate set, the blog's candidate-explosion warning made concrete.
+- The SQLite figure is measured here, not just estimated.
+
+## Files
+
+| file | role |
+| --- | --- |
+| `data.py` | build the warehouse via the companies generator, plus hidden ground truth |
+| `model.py` | shared standardisation and the train-once model artefact |
+| `pipeline_matchlab.py` | the pipeline as a matchlab plan |
+| `pipeline_splink.py` | the same pipeline hand-rolled with Splink and a union-find |
+| `benchmark.py` | expression, batch time and memory, single-match latency |

examples/online/benchmark.py

@@ -0,0 +1,195 @@
+"""matchlab vs plain Splink: expression, batch cost, and a single online match.
+
+Both pipelines run the identical algorithm (dedupe each source, link every pair with the
+shared Splink model, cluster once), so the comparison isolates what matchlab adds: a
+plan, and a materialised store. We check they agree, time and memory-profile the batch
+build, then time a single online lookup against matchlab's own DuckDB store versus the
+same resolution in SQLite.
+
+    python benchmark.py                # the whole story, one table
+    python benchmark.py _one matchlab  # run one pipeline once (used by memray)
+"""
+
+from __future__ import annotations
+
+import json
+import sqlite3
+import subprocess
+import sys
+import time
+import timeit
+import warnings
+from pathlib import Path
+
+import polars as pl
+
+warnings.filterwarnings("ignore")
+import logging  # noqa: E402
+
+logging.disable(logging.WARNING)
+
+HERE = Path(__file__).parent
+sys.path.insert(0, str(HERE))
+
+import data  # noqa: E402
+import model  # noqa: E402
+import pipeline_matchlab  # noqa: E402
+import pipeline_splink  # noqa: E402
+
+from matchlab.adapters import DuckDBAdapter  # noqa: E402
+
+N_ROWS = 300_000
+REPEATS = 3
+N_PROBES = 2_000
+
+
+def build_matchlab(trained: dict, on_disk: bool = False) -> None:
+    """Collect the matchlab plan into a fresh store. Times/measures the build only.
+
+    A fresh store each time, so content-addressed caching doesn't turn a repeat into a
+    no-op. On disk, the materialised tables live in the DuckDB file rather than in RAM.
+    """
+    path = HERE / f".ml_{time.time_ns()}.duckdb" if on_disk else None
+    store = DuckDBAdapter(path) if path else DuckDBAdapter()
+    pipeline_matchlab.build_plan(trained).collect(store)
+    store.close()
+    if path:
+        path.unlink(missing_ok=True)
+
+
+def matchlab_resolution(trained: dict) -> pl.DataFrame:
+    """The matchlab resolution, for the correctness check."""
+    return pipeline_matchlab.build_plan(trained).collect(DuckDBAdapter()).resolution()
+
+
+def _mixed(res: pl.DataFrame, truth: pl.DataFrame) -> int:
+    """Clusters that mix more than one true entity (0 means perfect)."""
+    return (
+        res.join(truth, on=["source", "key"])
+        .group_by("root")
+        .agg(pl.col("entity").n_unique().alias("n"))
+        .filter(pl.col("n") > 1)
+        .height
+    )
+
+
+def _peak_mib(which: str) -> float:
+    """Peak heap of one pipeline, measured by memray in a subprocess."""
+    binf, jsonf = HERE / f".mem_{which}.bin", HERE / f".mem_{which}.json"
+    run = ["uv", "run", "memray"]
+    kw = {"check": True, "capture_output": True, "cwd": HERE}
+    subprocess.run(
+        [*run, "run", "-q", "-f", "-o", str(binf), __file__, "_one", which], **kw
+    )
+    subprocess.run([*run, "stats", "--json", "-fo", str(jsonf), str(binf)], **kw)
+    peak = json.loads(jsonf.read_text())["metadata"]["peak_memory"]
+    binf.unlink(), jsonf.unlink()
+    return peak / 1024**2
+
+
+def _time(lookup, probes: list) -> float:
+    """Mean latency of a single lookup, in milliseconds."""
+    lookup(*probes[0])  # warm up
+    start = time.perf_counter()
+    for source, key in probes:
+        lookup(source, key)
+    return (time.perf_counter() - start) / len(probes) * 1000
+
+
+def _one(which: str) -> None:
+    data.ensure_warehouse(N_ROWS)
+    trained = model.load_or_train(pipeline_splink.all_nodes())
+    if which == "splink":
+        pipeline_splink.resolution(trained)
+    else:
+        build_matchlab(trained, on_disk=which == "matchlab_disk")
+
+
+def main() -> None:
+    data.ensure_warehouse(N_ROWS)
+    truth = data.truth()
+    trained = model.load_or_train(pipeline_splink.all_nodes())
+    print(f"{truth.height:,} records, {truth['entity'].n_unique():,} true entities\n")
+
+    # Expression: both pipelines must agree with the ground truth.
+    for name, res in (
+        ("matchlab", matchlab_resolution(trained)),
+        ("splink", pipeline_splink.resolution(trained)),
+    ):
+        good = res["root"].n_unique() == truth["entity"].n_unique() and not _mixed(
+            res, truth
+        )
+        print(f"expression       {name:<9} {'ok' if good else 'FAIL'}")
+    print()
+
+    # Batch build: time (fresh store each repeat) and peak memory. matchlab is timed
+    # for the build only (the store is the product), both in memory and on disk.
+    def best(fn):
+        return min(timeit.repeat(fn, number=1, repeat=REPEATS))
+
+    ts = best(lambda: pipeline_splink.resolution(trained))
+    tm = best(lambda: build_matchlab(trained))
+    td = best(lambda: build_matchlab(trained, on_disk=True))
+    ms, mm, md = _peak_mib("splink"), _peak_mib("matchlab"), _peak_mib("matchlab_disk")
+    print(
+        f"batch build      {'Splink':>10}{'matchlab (mem)':>16}{'matchlab (disk)':>17}"
+    )
+    print(f"  time         {ts:>8.1f} s{tm:>14.1f} s{td:>15.1f} s")
+    print(f"  peak memory  {ms:>6,.0f} MiB{mm:>12,.0f} MiB{md:>13,.0f} MiB\n")
+
+    # A single online match, against matchlab's own DuckDB store and the same
+    # resolution in SQLite. No extra copies: the store is the one the plan built.
+    store_path = HERE / ".online_store.duckdb"
+    store_path.unlink(missing_ok=True)
+    store = DuckDBAdapter(store_path)
+    apex = pipeline_matchlab.build_plan(trained)
+    apex.collect(store)
+    fp = apex._fp
+    con = store.conn
+
+    def duck(source, key):
+        root = con.execute(
+            "SELECT root FROM resolution WHERE fp=? AND source=? AND key=?",
+            [fp, source, key],
+        ).fetchone()[0]
+        return con.execute(
+            "SELECT source, key FROM resolution WHERE fp=? AND root=?", [fp, root]
+        ).fetchall()
+
+    res = con.execute("SELECT root, key, source FROM resolution WHERE fp=?", [fp]).pl()
+    sqlite_path = HERE / ".online_store.sqlite"
+    sqlite_path.unlink(missing_ok=True)
+    scon = sqlite3.connect(sqlite_path)
+    scon.execute("CREATE TABLE resolution (root TEXT, key TEXT, source TEXT)")
+    scon.executemany(
+        "INSERT INTO resolution VALUES (?,?,?)",
+        res.select(pl.col("root").cast(pl.Utf8), "key", "source").iter_rows(),
+    )
+    scon.execute("CREATE INDEX i1 ON resolution (source, key)")
+    scon.execute("CREATE INDEX i2 ON resolution (root)")
+
+    def row(source, key):
+        root = scon.execute(
+            "SELECT root FROM resolution WHERE source=? AND key=?", [source, key]
+        ).fetchone()[0]
+        return scon.execute(
+            "SELECT source, key FROM resolution WHERE root=?", [root]
+        ).fetchall()
+
+    probes = list(res.select("source", "key").sample(N_PROBES, seed=1).iter_rows())
+    duck_ms, row_ms = _time(duck, probes), _time(row, probes)
+    print("single online match      latency     throughput")
+    print(f"  DuckDB (matchlab store) {duck_ms:7.3f} ms   {1000 / duck_ms:>8,.0f} /s")
+    print(f"  SQLite (row store)      {row_ms:7.3f} ms   {1000 / row_ms:>8,.0f} /s")
+
+    scon.close()
+    store.close()
+    store_path.unlink(missing_ok=True)
+    sqlite_path.unlink(missing_ok=True)
+
+
+if __name__ == "__main__":
+    if len(sys.argv) >= 3 and sys.argv[1] == "_one":
+        _one(sys.argv[2])
+    else:
+        main()