bench(gfql): large real-graph + bulk-OLAP + fairest-kuzu harnesses (4 engines vs kuzu)

lmeyerov · claude · lmeyerov · commit 1257dac9a94e · 2026-06-30T18:13:43.000-07:00
Adds the dgx-spark benchmark harnesses backing the CSR-index + bulk-OLAP claims,
all guarded for trust (timing only reported when the index path was actually taken
AND index result == scan result; engine parity checked via matched row counts):

- index_largegraph_bench.py: real SNAP edge lists (LiveJournal 35M, Orkut 117M,
  Friendster stretch), parquet-cached, degree-percentile + multi-seed sweeps. Shows
  seeded latency flat in N, scaling with seed degree (the O(degree) honesty sweep).
- index_bulk_olap_bench.py: BULK regime the index deliberately avoids -- seeded
  multi-hop frontier expansion via the chain API (the one GFQL surface supporting
  all 4 engines) + full-graph out-degree aggregation, 4 engines + kuzu. Answers
  "is bulk OLAP better with GFQL cudf?": yes with GFQL, but on polars/polars-gpu
  (fused lazy), not cudf (eager per-op). polars-CPU 11-47x over pandas, 6-18x over
  cudf, 3-87x over kuzu on frontier expansion at 35-117M edges.
- index_vs_kuzu_prepared.py: fairest seeded comparison -- kuzu prepared statement +
  columnar get_as_df, in-process (no bolt), matched rows.

Co-Authored-By: Claude Opus 4.8 (1M context) &lt;noreply@anthropic.com&gt;
diff --git a/benchmarks/gfql/index_bulk_olap_bench.py b/benchmarks/gfql/index_bulk_olap_bench.py
@@ -0,0 +1,194 @@
+#!/usr/bin/env python3
+"""BULK-OLAP head-to-head: GFQL 4 engines vs kuzu on REAL graphs.
+
+Answers "is bulk OLAP better with GFQL (cudf / polars-gpu)?" The seeded CSR index
+is O(degree) and wins tiny work; this bench deliberately AVOIDS that path and
+measures the BULK regime instead — large-frontier multi-hop + full-graph
+aggregation, i.e. the scan/join work where columnar GPU throughput should pay off
+and the index does NOT help. We run g0.hop (NO resident index -> engine traversal,
+the honest bulk path) so every engine does the same materialized join work.
+
+Tasks (all bulk, all materialized on both sides):
+  BULK1   1-hop forward from K seeds         (edge semijoin, frontier=K)
+  BULK2   2-hop forward from K seeds         (edge-edge join, frontier blows up)
+  DEGALL  full-graph out-degree aggregation  (group_by over ALL edges; pure OLAP)
+K frontier sweep: 1k, 10k, 100k seeds. cudf/polars-gpu should overtake pandas as K
+(hence work) grows; kuzu is the WCOJ/optimizer peer for the multi-hop join.
+
+Trust: GFQL rows reported per engine (engine parity is separately guaranteed by the
+conformance suite); kuzu rows reported alongside with a semantic note. Timing is the
+deliverable — rows are the honesty check that each system did real work.
+
+Env: PARQUET=/path/edges.parquet  KS=1000,10000,100000  ENGINES=pandas,polars,cudf,polars-gpu
+     SYSTEMS=gfql,kuzu  REPS=10  WARM=2  OUT=/tmp/bulk.jsonl  SEED=0
+"""
+from __future__ import annotations
+import json, os, statistics, time, tempfile, shutil
+import numpy as np
+import pandas as pd
+import graphistry
+from graphistry.compute.ast import n, e_forward
+
+
+def _sync(engine):
+    if engine in ("cudf", "polars-gpu"):
+        try:
+            import cupy as cp  # type: ignore
+            cp.cuda.runtime.deviceSynchronize()
+        except Exception:
+            pass
+
+
+def timeit(fn, reps, engine="cpu", warmup=2):
+    for _ in range(warmup):
+        fn(); _sync(engine)
+    ts = []
+    for _ in range(reps):
+        t0 = time.perf_counter(); fn(); _sync(engine)
+        ts.append((time.perf_counter() - t0) * 1e3)
+    ts.sort()
+    return statistics.median(ts)
+
+
+def load_graph():
+    edf = pd.read_parquet(os.environ["PARQUET"]).astype({"src": np.int64, "dst": np.int64})
+    nodes = np.unique(np.concatenate([edf["src"].values, edf["dst"].values]))
+    ndf = pd.DataFrame({"id": nodes})
+    return ndf, edf, nodes
+
+
+def gfql_trav(g0, seed_ids, hops, engine):
+    """BULK seeded multi-hop via the CHAIN API — the one GFQL surface that supports
+    ALL FOUR engines (generic hop() is pandas/cudf only; polars/polars-gpu route
+    through engine_polars). n({id:seeds}) = frontier filter, then e_forward()*hops."""
+    ops = [n({"id": seed_ids})] + [e_forward() for _ in range(hops)]
+    return g0.chain(ops, engine=engine)
+
+
+def run_gfql(ndf, edf, nodes, ks, engines, reps, warm, outf, seed):
+    N, E = len(ndf), len(edf)
+    rng = np.random.default_rng(seed)
+    seed_sets = {k: rng.choice(nodes, size=min(k, len(nodes)), replace=False).tolist() for k in ks}
+    for engine in engines:
+        try:
+            g0 = graphistry.nodes(ndf, "id").edges(edf, "src", "dst")
+            # warm/convert frames onto the engine ONCE (exclude H2D/convert from timing)
+            _ = gfql_trav(g0, seed_sets[ks[0]], 1, engine)
+        except Exception as ex:
+            print(f"  gfql {engine}: SETUP FAILED {type(ex).__name__}: {ex}"); continue
+        # frontier sweep: BULK1 (1-hop) + BULK2 (2-hop)
+        for k in ks:
+            sids = seed_sets[k]
+            for task, hops in (("BULK1", 1), ("BULK2", 2)):
+                try:
+                    res = gfql_trav(g0, sids, hops, engine)
+                    rows = int(res._edges.shape[0]); nn = int(res._nodes.shape[0])
+                    ms = timeit(lambda: gfql_trav(g0, sids, hops, engine), reps, engine, warm)
+                except Exception as ex:
+                    print(f"  gfql {engine} {task} k={k} FAILED: {type(ex).__name__}: {ex}"); continue
+                rec = dict(system="gfql", engine=engine, task=task, k=k, hops=hops,
+                           n=N, edges=E, warm_ms=ms, e_rows=rows, n_rows=nn)
+                print(f"  gfql {engine:11} {task} k={k:>7} {ms:10.3f}ms  e_rows={rows:>10}  n_rows={nn:>9}")
+                if outf: outf.write(json.dumps(rec) + "\n"); outf.flush()
+        # DEGALL: full-graph out-degree aggregation (pure columnar OLAP, no traversal)
+        try:
+            ms, rows = degall(edf, engine, reps, warm)
+            rec = dict(system="gfql", engine=engine, task="DEGALL", k=None, hops=0,
+                       n=N, edges=E, warm_ms=ms, e_rows=rows, n_rows=rows)
+            print(f"  gfql {engine:11} DEGALL{'':>13} {ms:10.3f}ms  groups={rows:>10}")
+            if outf: outf.write(json.dumps(rec) + "\n"); outf.flush()
+        except Exception as ex:
+            print(f"  gfql {engine} DEGALL FAILED: {type(ex).__name__}: {ex}")
+
+
+def degall(edf, engine, reps, warm):
+    """Full-graph out-degree = group_by(src).size() on the chosen engine."""
+    if engine == "pandas":
+        df = edf
+        fn = lambda: df.groupby("src").size()
+    elif engine == "cudf":
+        import cudf
+        df = cudf.from_pandas(edf)
+        fn = lambda: df.groupby("src").size()
+    elif engine in ("polars", "polars-gpu"):
+        import polars as pl
+        df = pl.from_pandas(edf)
+        if engine == "polars-gpu":
+            eng = pl.GPUEngine(executor="in-memory", raise_on_fail=False)
+            fn = lambda: df.lazy().group_by("src").len().collect(engine=eng)
+        else:
+            fn = lambda: df.group_by("src").len()
+    else:
+        raise ValueError(engine)
+    r = fn(); rows = int(r.shape[0])
+    ms = timeit(fn, reps, engine, warm)
+    return ms, rows
+
+
+def run_kuzu(ndf, edf, nodes, ks, reps, warm, outf, seed, tmpdir):
+    try:
+        import kuzu
+    except Exception:
+        print("  kuzu: NOT AVAILABLE (pip install kuzu)"); return
+    rng = np.random.default_rng(seed)
+    seed_sets = {k: rng.choice(nodes, size=min(k, len(nodes)), replace=False).tolist() for k in ks}
+    dbp = tempfile.mkdtemp(dir=tmpdir)
+    db = kuzu.Database(os.path.join(dbp, "kz")); conn = kuzu.Connection(db)
+    conn.execute("CREATE NODE TABLE N(id INT64, PRIMARY KEY(id))")
+    conn.execute("CREATE REL TABLE E(FROM N TO N)")
+    np_path = os.path.join(dbp, "n.parquet"); ep_path = os.path.join(dbp, "e.parquet")
+    ndf.to_parquet(np_path)
+    edf.rename(columns={"src": "from", "dst": "to"}).to_parquet(ep_path)
+    t0 = time.perf_counter()
+    conn.execute(f'COPY N FROM "{np_path}"'); conn.execute(f'COPY E FROM "{ep_path}"')
+    load_ms = (time.perf_counter() - t0) * 1e3
+    print(f"  kuzu load: {load_ms:.0f}ms")
+    # BULK1/BULK2: distinct reachable set from K seeds (materialized columnar via get_as_df)
+    q1 = conn.prepare("MATCH (a:N)-[:E]->(b:N) WHERE a.id IN $seeds RETURN b.id")
+    q2 = conn.prepare("MATCH (a:N)-[:E]->()-[:E]->(b:N) WHERE a.id IN $seeds RETURN b.id")
+    for k in ks:
+        s = seed_sets[k]
+        for task, stmt in (("BULK1", q1), ("BULK2", q2)):
+            try:
+                rows = len(conn.execute(stmt, {"seeds": s}).get_as_df())
+                ms = timeit(lambda: conn.execute(stmt, {"seeds": s}).get_as_df(), reps, "kuzu", warm)
+            except Exception as ex:
+                print(f"  kuzu {task} k={k} FAILED: {type(ex).__name__}: {ex}"); continue
+            rec = dict(system="kuzu", engine="kuzu", task=task, k=k, n=len(ndf), edges=len(edf),
+                       warm_ms=ms, e_rows=rows, n_rows=rows, load_ms=load_ms)
+            print(f"  kuzu {'':11} {task} k={k:>7} {ms:10.3f}ms  rows={rows:>10}  (b.id, not-distinct)")
+            if outf: outf.write(json.dumps(rec) + "\n"); outf.flush()
+    # DEGALL: full out-degree aggregation
+    try:
+        qd = "MATCH (a:N)-[:E]->() RETURN a.id, count(*) AS deg"
+        for _ in range(warm): conn.execute(qd).get_as_df()
+        rows = len(conn.execute(qd).get_as_df())
+        ms = timeit(lambda: conn.execute(qd).get_as_df(), reps, "kuzu", warm)
+        rec = dict(system="kuzu", engine="kuzu", task="DEGALL", k=None, n=len(ndf), edges=len(edf),
+                   warm_ms=ms, e_rows=rows, n_rows=rows, load_ms=load_ms)
+        print(f"  kuzu {'':11} DEGALL{'':>13} {ms:10.3f}ms  groups={rows:>10}")
+        if outf: outf.write(json.dumps(rec) + "\n"); outf.flush()
+    except Exception as ex:
+        print(f"  kuzu DEGALL FAILED: {type(ex).__name__}: {ex}")
+    shutil.rmtree(dbp, ignore_errors=True)
+
+
+def main():
+    ndf, edf, nodes = load_graph()
+    print(f"===== graph: {len(ndf):,} nodes  {len(edf):,} edges =====")
+    ks = [int(x) for x in os.environ.get("KS", "1000,10000,100000").split(",")]
+    engines = os.environ.get("ENGINES", "pandas,polars,cudf,polars-gpu").split(",")
+    systems = os.environ.get("SYSTEMS", "gfql,kuzu").split(",")
+    reps = int(os.environ.get("REPS", "10")); warm = int(os.environ.get("WARM", "2"))
+    seed = int(os.environ.get("SEED", "0"))
+    tmpdir = os.environ.get("TMPDIR_BENCH", "/tmp/bulkbench"); os.makedirs(tmpdir, exist_ok=True)
+    outf = open(os.environ["OUT"], "a") if os.environ.get("OUT") else None
+    if "gfql" in systems:
+        run_gfql(ndf, edf, nodes, ks, engines, reps, warm, outf, seed)
+    if "kuzu" in systems:
+        run_kuzu(ndf, edf, nodes, ks, reps, warm, outf, seed, tmpdir)
+    if outf: outf.close()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/benchmarks/gfql/index_largegraph_bench.py b/benchmarks/gfql/index_largegraph_bench.py
@@ -0,0 +1,149 @@
+#!/usr/bin/env python3
+"""Large REAL-graph CSR-index bench (Step 7). Power-law topology exposes what the
+uniform deg-8 synthetic never did: the index is O(degree), so warm latency is flat
+in N but scales with SEED DEGREE — a hub seed is the adversarial worst case.
+
+Same trust discipline as index_takeover_bench.py: every GFQL timing is GUARDED by
+(index path actually taken via index_trace) AND (index result == scan result). A
+cell failing either guard is reported INVALID, never as a speedup.
+
+Datasets (SNAP edge lists, gzipped `u v`, load once -> parquet cache):
+  com-Orkut       3.07M nodes / 117M edges   https://snap.stanford.edu/data/bigdata/communities/com-orkut.ungraph.txt.gz
+  com-LiveJournal 4.0M / 34.7M               https://snap.stanford.edu/data/bigdata/communities/com-lj.ungraph.txt.gz
+  soc-LiveJournal1 4.8M / 69M (directed)     https://snap.stanford.edu/data/soc-LiveJournal1.txt.gz
+  com-Friendster  65.6M / 1.8B (STRETCH)     https://snap.stanford.edu/data/bigdata/communities/com-friendster.ungraph.txt.gz
+  twitter-2010    41.7M / 1.47B (STRETCH)
+LDBC SNB sf10/sf100 via ~/Work/pyg-bench loader + the live snb-interactive-neo4j.
+
+Env: EDGELIST=/path/to/edges.txt.gz  (or PARQUET=/path/edges.parquet)
+     DEG_PCTLS=50,90,99,100  MULTISEED=1,10,100,1000  ENGINES=pandas,polars,cudf,polars-gpu
+     REPS=15  OUT=/tmp/lg.jsonl  MAXSCAN_REPS=3  (cap scan reps at large E)
+"""
+from __future__ import annotations
+import gzip, json, os, statistics, time
+import numpy as np
+import pandas as pd
+import graphistry
+from graphistry.compute.gfql.index import index_trace
+
+
+def load_graph(seed=0):
+    """Load a real edge list -> graphistry graph (int64 ids), parquet-cached."""
+    pq = os.environ.get("PARQUET")
+    el = os.environ.get("EDGELIST")
+    if pq and os.path.exists(pq):
+        edf = pd.read_parquet(pq)
+    elif el:
+        cache = el + ".parquet"
+        if os.path.exists(cache):
+            edf = pd.read_parquet(cache)
+        else:
+            op = gzip.open if el.endswith(".gz") else open
+            with op(el, "rt") as f:
+                edf = pd.read_csv(f, sep=r"\s+", comment="#", header=None,
+                                  names=["src", "dst"], dtype=np.int64)
+            edf.to_parquet(cache)
+            print(f"  cached parquet -> {cache}")
+    else:
+        # fallback synthetic power-law (Barabasi-ish via preferential attachment proxy)
+        rng = np.random.default_rng(seed)
+        n = int(os.environ.get("SYNTH_N", "1000000")); m = n * 8
+        deg = rng.zipf(2.2, m) % n
+        edf = pd.DataFrame({"src": rng.integers(0, n, m, dtype=np.int64), "dst": deg.astype(np.int64)})
+    nodes = np.unique(np.concatenate([edf["src"].values, edf["dst"].values]))
+    ndf = pd.DataFrame({"id": nodes})
+    return graphistry.nodes(ndf, "id").edges(edf, "src", "dst"), ndf, edf
+
+
+def degree_seeds(edf, pctls):
+    """Pick one seed id at each out-degree percentile (the O(degree) honesty sweep)."""
+    deg = edf.groupby("src").size()
+    out = {}
+    for p in pctls:
+        if p >= 100:
+            sid = int(deg.idxmax()); out["max"] = (int(deg.max()), sid)
+        else:
+            thr = np.percentile(deg.values, p)
+            cand = deg[deg >= thr]
+            sid = int(cand.index[0]); out[f"p{p}"] = (int(deg.loc[sid]), sid)
+    return out
+
+
+def _sync(engine):
+    if engine in ("cudf", "polars-gpu"):
+        try:
+            import cupy as cp; cp.cuda.runtime.deviceSynchronize()
+        except Exception:
+            pass
+
+
+def timeit(fn, reps, engine, warmup=2):
+    for _ in range(warmup):
+        fn(); _sync(engine)
+    ts = []
+    for _ in range(reps):
+        t0 = time.perf_counter(); fn(); _sync(engine); ts.append((time.perf_counter() - t0) * 1e3)
+    ts.sort(); return statistics.median(ts)
+
+
+def _sig(g):
+    n, e = g._nodes, g._edges
+    if "polars" in type(n).__module__ or "cudf" in type(n).__module__: n = n.to_pandas()
+    if "polars" in type(e).__module__ or "cudf" in type(e).__module__: e = e.to_pandas()
+    return (len(n), len(e), int(e["src"].sum()) + int(e["dst"].sum()))
+
+
+def bench(g0, ndf, edf, engines, reps):
+    maxscan = int(os.environ.get("MAXSCAN_REPS", "3"))
+    E = len(edf)
+    pctls = [int(x) for x in os.environ.get("DEG_PCTLS", "50,90,99,100").split(",")]
+    multiseed = [int(x) for x in os.environ.get("MULTISEED", "1,10,100,1000").split(",")]
+    dseeds = degree_seeds(edf, pctls)
+    print(f"  degree seeds: " + ", ".join(f"{k}=deg{d}" for k, (d, _) in dseeds.items()))
+    outf = open(os.environ["OUT"], "a") if os.environ.get("OUT") else None
+    for engine in engines:
+        try:
+            t0 = time.perf_counter(); gi = g0.gfql_index_all(engine=engine); _sync(engine)
+            build_ms = (time.perf_counter() - t0) * 1e3
+        except Exception as ex:
+            print(f"  {engine}: BUILD FAILED {type(ex).__name__}: {ex}"); continue
+        # T3: seed-degree sweep (1-hop), guarded
+        for tag, (deg, sid) in dseeds.items():
+            seeds = pd.DataFrame({"id": [sid]})
+            with index_trace() as steps:
+                gidx = gi.hop(nodes=seeds, engine=engine, hops=1, direction="forward")
+            took = any(s.get("path") == "index" for s in steps)
+            gscan = g0.hop(nodes=seeds, engine=engine, hops=1, direction="forward")
+            same = _sig(gidx) == _sig(gscan)
+            valid = took and same
+            wi = timeit(lambda: gi.hop(nodes=seeds, engine=engine, hops=1, direction="forward"), reps, engine)
+            ws = timeit(lambda: g0.hop(nodes=seeds, engine=engine, hops=1, direction="forward"),
+                        min(reps, maxscan), engine)
+            rec = dict(system="gfql", engine=engine, task="degsweep", seed_deg=deg, n=len(ndf), edges=E,
+                       valid=valid, warm_idx_ms=wi, warm_scan_ms=ws, speedup=ws / wi if wi else None, build_ms=build_ms)
+            print(f"  {engine:11} deg={deg:>8} idx={wi:9.4f}ms scan={ws:10.3f}ms x{ws/wi:7.1f}{'' if valid else '  <<INVALID'}")
+            if outf: outf.write(json.dumps(rec) + "\n"); outf.flush()
+        # T4: multi-seed frontier sweep (where the cost gate flips index->scan)
+        rng = np.random.default_rng(0)
+        allids = ndf["id"].values
+        for k in multiseed:
+            seeds = pd.DataFrame({"id": rng.choice(allids, size=min(k, len(allids)), replace=False)})
+            with index_trace() as steps:
+                gidx = gi.hop(nodes=seeds, engine=engine, hops=1, direction="forward")
+            took = any(s.get("path") == "index" for s in steps)
+            wi = timeit(lambda: gi.hop(nodes=seeds, engine=engine, hops=1, direction="forward"), reps, engine)
+            print(f"  {engine:11} kseed={k:>6} idx={wi:9.4f}ms path={'index' if took else 'scan'}")
+            if outf: outf.write(json.dumps(dict(system="gfql", engine=engine, task="multiseed", kseed=k,
+                                n=len(ndf), edges=E, took_index=took, warm_idx_ms=wi)) + "\n"); outf.flush()
+    if outf: outf.close()
+
+
+def main():
+    g0, ndf, edf = load_graph()
+    print(f"===== graph: {len(ndf):,} nodes  {len(edf):,} edges =====")
+    engines = os.environ.get("ENGINES", "pandas,polars").split(",")
+    bench(g0, ndf, edf, engines, int(os.environ.get("REPS", "15")))
+
+
+if __name__ == "__main__":
+    main()
diff --git a/benchmarks/gfql/index_vs_kuzu_prepared.py b/benchmarks/gfql/index_vs_kuzu_prepared.py
@@ -0,0 +1,54 @@
+#!/usr/bin/env python3
+"""Fairest GFQL-vs-kuzu seeded 1-hop: BOTH in-process, warm, kuzu using a PREPARED
+statement (its fast path) + result fully materialized on both sides, same seed,
+matched answer counts. Removes Cypher-parse-per-call from kuzu so the comparison is
+engine-vs-engine, not engine-vs-(parse+engine). kuzu is embedded like GFQL (no bolt
+network), so it's the cleanest peer. Env: PARQUET=/data/<edges>.parquet"""
+import os, time, statistics, tempfile, shutil
+import numpy as np, pandas as pd, graphistry, kuzu
+
+
+def med(fn, reps=25, warm=4):
+    for _ in range(warm): fn()
+    ts = []
+    for _ in range(reps):
+        t = time.perf_counter(); fn(); ts.append((time.perf_counter() - t) * 1e3)
+    ts.sort(); return statistics.median(ts)
+
+
+def main():
+    edf = pd.read_parquet(os.environ["PARQUET"]).astype({"src": np.int64, "dst": np.int64})
+    nodes = np.unique(np.concatenate([edf["src"].values, edf["dst"].values]))
+    print(f"graph: {len(nodes):,} nodes / {len(edf):,} edges")
+    g = graphistry.nodes(pd.DataFrame({"id": nodes}), "id").edges(edf, "src", "dst")
+    gi = g.gfql_index_all(engine="pandas")
+    deg = edf.groupby("src").size()
+    # typical (median-degree) and hub seeds
+    for tag, sid in [("typical", int(deg[deg >= deg.median()].index[0])), ("hub", int(deg.idxmax()))]:
+        d = int(deg.loc[sid]); seeds = pd.DataFrame({"id": [sid]})
+        gfql_rows = int(gi.hop(nodes=seeds, engine="pandas", hops=1)._edges.shape[0])
+        gfql_ms = med(lambda: gi.hop(nodes=seeds, engine="pandas", hops=1))
+
+        dbp = tempfile.mkdtemp()
+        db = kuzu.Database(os.path.join(dbp, "kz")); conn = kuzu.Connection(db)
+        conn.execute("CREATE NODE TABLE N(id INT64, PRIMARY KEY(id))")
+        conn.execute("CREATE REL TABLE E(FROM N TO N)")
+        np_path = os.path.join(dbp, "n.parquet"); ep_path = os.path.join(dbp, "e.parquet")
+        pd.DataFrame({"id": nodes}).to_parquet(np_path)
+        edf.rename(columns={"src": "from", "dst": "to"}).to_parquet(ep_path)
+        conn.execute(f'COPY N FROM "{np_path}"'); conn.execute(f'COPY E FROM "{ep_path}"')
+        stmt = conn.prepare("MATCH (a:N {id:$sid})-[:E]->(b:N) RETURN b.id")
+        # Columnar materialization (kuzu's fast result path) == GFQL's DataFrame output.
+        def kq():
+            conn.execute(stmt, {"sid": sid}).get_as_df()
+        kr = len(conn.execute(stmt, {"sid": sid}).get_as_df())
+        kuzu_ms = med(kq)
+        ratio = kuzu_ms / gfql_ms if gfql_ms else float("nan")
+        print(f"  {tag:8} deg={d:>7}  GFQL-pandas {gfql_ms:8.4f}ms (rows={gfql_rows})  "
+              f"kuzu-prepared {kuzu_ms:8.4f}ms (rows={kr})  match={gfql_rows==kr}  "
+              f"GFQL {'faster' if ratio>1 else 'SLOWER'} {ratio:.2f}x")
+        shutil.rmtree(dbp, ignore_errors=True)
+
+
+if __name__ == "__main__":
+    main()
