bench: enrich metrics + switch graph peer Kùzu→LadybugDB

- bench_common.py: latency percentiles (mean/std/min/p50/p90/p95/p99/max),
  on-disk size, raw-latency sidecar, timing helper
- vector/tabular/graph benches: record lifecycle phases (import, jvm_init,
  open, schema, ingest, index_build, close) + db_size, ingest throughput,
  cold-query latency, and full per-op latency stats
- graph_bench: replace Kùzu with LadybugDB (real_ladybug, Kùzu-compatible API)
- run.py: pass RUN_LABEL/LAT_DIR, sample cgroup CPU (user/system), dump raw latencies
- make_tables.py: DB-size columns + lifecycle and latency-percentile tables
- build_images.sh: ladybug image (real_ladybug==0.15.3)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: tae898

papers/scipy-2026/experiments/bench_common.py

@@ -0,0 +1,83 @@
+#!/usr/bin/env python3
+"""Shared metrics helpers for the benchmark lanes.
+
+Stdlib-only (so every backend image can import it): latency summary stats,
+on-disk size, a raw-latency sidecar dump, and a small timing context manager.
+"""
+import json
+import os
+import statistics as st
+import time
+
+
+def _pct(sorted_a, p):
+    """Linear-interpolation percentile (numpy-compatible), pure-python."""
+    n = len(sorted_a)
+    if n == 1:
+        return sorted_a[0]
+    rank = (p / 100.0) * (n - 1)
+    lo = int(rank)
+    hi = min(lo + 1, n - 1)
+    frac = rank - lo
+    return sorted_a[lo] * (1 - frac) + sorted_a[hi] * frac
+
+
+def latstats(prefix, arr_ms):
+    """Full latency summary (ms) for a list of per-op latencies:
+    n, mean, std, min, p50, p90, p95, p99, max."""
+    if not arr_ms:
+        return {}
+    a = sorted(float(x) for x in arr_ms)
+    n = len(a)
+    return {
+        f"{prefix}_n": n,
+        f"{prefix}_mean_ms": round(st.mean(a), 4),
+        f"{prefix}_std_ms": round(st.pstdev(a) if n > 1 else 0.0, 4),
+        f"{prefix}_min_ms": round(a[0], 4),
+        f"{prefix}_p50_ms": round(_pct(a, 50), 4),
+        f"{prefix}_p90_ms": round(_pct(a, 90), 4),
+        f"{prefix}_p95_ms": round(_pct(a, 95), 4),
+        f"{prefix}_p99_ms": round(_pct(a, 99), 4),
+        f"{prefix}_max_ms": round(a[-1], 4),
+    }
+
+
+def dir_size_mb(path):
+    """On-disk size of a file or directory tree (MiB)."""
+    if not path or not os.path.exists(path):
+        return None
+    if os.path.isfile(path):
+        total = os.path.getsize(path)
+    else:
+        total = 0
+        for root, _, files in os.walk(path):
+            for f in files:
+                try:
+                    total += os.path.getsize(os.path.join(root, f))
+                except OSError:
+                    pass
+    return round(total / 1048576.0, 3)
+
+
+def dump_latencies(run_label, lat_by_op):
+    """Write raw per-op latency arrays to $LAT_DIR/<label>.json (set by run.py).
+    No-op if LAT_DIR is unset. Returns the relative sidecar path or None."""
+    d = os.environ.get("LAT_DIR")
+    if not d or not run_label:
+        return None
+    os.makedirs(d, exist_ok=True)
+    with open(os.path.join(d, f"{run_label}.json"), "w") as f:
+        json.dump({k: [round(float(x), 5) for x in v] for k, v in lat_by_op.items() if v}, f)
+    return f"lat/{run_label}.json"
+
+
+class timed:
+    """`with timed() as t: ...` then read t.s (elapsed seconds)."""
+    def __enter__(self):
+        self._t0 = time.time()
+        self.s = 0.0
+        return self
+
+    def __exit__(self, *exc):
+        self.s = time.time() - self._t0
+        return False

papers/scipy-2026/experiments/build_images.sh

@@ -8,12 +8,12 @@ declare -A PKGS=(
   [arcadedb]="arcadedb-embedded==26.6.1 numpy pandas pyarrow"
   [sqlite]="pandas pyarrow"
   [duckdb]="duckdb==1.5.4 pandas pyarrow"
-  [kuzu]="kuzu==0.11.3 pandas pyarrow"
+  [ladybug]="real_ladybug==0.15.3 pandas pyarrow"
   [chroma]="chromadb==1.5.9 numpy"
 )
 
 targets=("$@")
-[ ${#targets[@]} -eq 0 ] && targets=(arcadedb sqlite duckdb kuzu chroma)
+[ ${#targets[@]} -eq 0 ] && targets=(arcadedb sqlite duckdb ladybug chroma)
 
 for be in "${targets[@]}"; do
   echo "=== building scipy-bench:$be  (${PKGS[$be]}) ==="

papers/scipy-2026/experiments/graph_bench.py

@@ -1,20 +1,20 @@
 #!/usr/bin/env python3
 """Graph lane: OLTP (point/1-hop reads + transactional node writes) and OLAP (traversals).
 
-Backends: kuzu, arcadedb. Graph: (User)-[:POSTED]->(Post), (Post)-[:ANSWERS]->(Post).
-Loads from normalized parquet; edges filtered to existing endpoints (tiny caps tables at 10k).
-Cypher queries are shared (ints embedded, no param-dialect issues). Prints RESULT {json}.
+Backends: ladybug, arcadedb. Graph: (User)-[:POSTED]->(Post), (Post)-[:ANSWERS]->(Post).
+(LadybugDB is the maintained continuation of Kùzu; package `real_ladybug`, Kùzu-compatible API.)
+Records lifecycle phase timings, on-disk size, and full per-op latency stats. RESULT {json}.
 """
 import argparse
 import json
 import os
 import random
-import statistics as st
-import tempfile
 import time
 
 import pandas as pd
 
+import bench_common as bc
+
 
 def load_graph(data_dir, limit):
     posts = pd.read_parquet(os.path.join(data_dir, "posts.parquet"), columns=["id"])
@@ -40,25 +40,29 @@ def load_graph(data_dir, limit):
 ]
 
 
-def be_kuzu(users, posts, posted, answers, workload):
-    import kuzu
-    db = kuzu.Database(tempfile.mkdtemp(prefix="gb_kuzu_") + "/db")
-    conn = kuzu.Connection(db)
-    conn.execute("CREATE NODE TABLE User(id INT64, PRIMARY KEY(id))")
-    conn.execute("CREATE NODE TABLE Post(id INT64, PRIMARY KEY(id))")
-    conn.execute("CREATE REL TABLE POSTED(FROM User TO Post)")
-    conn.execute("CREATE REL TABLE ANSWERS(FROM Post TO Post)")
-    d = tempfile.mkdtemp(prefix="gb_kuzu_data_")
+def be_ladybug(users, posts, posted, answers, workload):
+    import tempfile
+    with bc.timed() as t_imp:
+        import real_ladybug as lb  # maintained continuation of Kùzu; Kùzu-compatible API
+    path = tempfile.mkdtemp(prefix="gb_ladybug_") + "/db.lbug"
+    with bc.timed() as t_open:
+        db = lb.Database(path)
+        conn = lb.Connection(db)
+    with bc.timed() as t_schema:
+        conn.execute("CREATE NODE TABLE User(id INT64, PRIMARY KEY(id))")
+        conn.execute("CREATE NODE TABLE Post(id INT64, PRIMARY KEY(id))")
+        conn.execute("CREATE REL TABLE POSTED(FROM User TO Post)")
+        conn.execute("CREATE REL TABLE ANSWERS(FROM Post TO Post)")
+    d = tempfile.mkdtemp(prefix="gb_ladybug_data_")
     pd.DataFrame({"id": users}).to_parquet(f"{d}/u.parquet")
     pd.DataFrame({"id": posts}).to_parquet(f"{d}/p.parquet")
     pd.DataFrame(posted, columns=["f", "t"]).to_parquet(f"{d}/posted.parquet")
     pd.DataFrame(answers, columns=["f", "t"]).to_parquet(f"{d}/answers.parquet")
-    t0 = time.time()
-    conn.execute(f"COPY User FROM '{d}/u.parquet'")
-    conn.execute(f"COPY Post FROM '{d}/p.parquet'")
-    conn.execute(f"COPY POSTED FROM '{d}/posted.parquet'")
-    conn.execute(f"COPY ANSWERS FROM '{d}/answers.parquet'")
-    load_s = time.time() - t0
+    with bc.timed() as t_ing:
+        conn.execute(f"COPY User FROM '{d}/u.parquet'")
+        conn.execute(f"COPY Post FROM '{d}/p.parquet'")
+        conn.execute(f"COPY POSTED FROM '{d}/posted.parquet'")
+        conn.execute(f"COPY ANSWERS FROM '{d}/answers.parquet'")
 
     def query(q):
         r = conn.execute(q)
@@ -67,64 +71,70 @@ def query(q):
             r.get_next(); n += 1
         return n
 
-    def write(q):
-        conn.execute(q)
-
-    return dict(load_s=load_s, query=query, write=write, close=lambda: None,
-                version=kuzu.__version__)
+    return dict(import_s=t_imp.s, jvm_init_s=0.0, open_s=t_open.s, schema_s=t_schema.s,
+                ingest_s=t_ing.s, index_build_s=0.0, gav_build_s=0.0, db_path=path,
+                query=query, write=lambda q: conn.execute(q),
+                close=lambda: None, version=lb.__version__)
 
 
 GAV_NAME = "gbOlap"
 
 
 def be_arcadedb(users, posts, posted, answers, workload):
-    import arcadedb_embedded as arcadedb
-    ctx = arcadedb.create_database(tempfile.mkdtemp(prefix="gb_arcadedb_") + "/db",
-                                   jvm_kwargs={"heap_size": os.environ.get("ARCADEDB_HEAP", "4g")})
-    db = ctx.__enter__()
-    for v in ("User", "Post"):
-        db.command("sql", f"CREATE VERTEX TYPE {v}")
-        db.command("sql", f"CREATE PROPERTY {v}.id LONG")
-        db.command("sql", f"CREATE INDEX ON {v} (id) UNIQUE_HASH")  # fast point lookups (ex 09)
-    db.command("sql", "CREATE EDGE TYPE POSTED")
-    db.command("sql", "CREATE EDGE TYPE ANSWERS")
+    import tempfile
+    with bc.timed() as t_imp:
+        import arcadedb_embedded as arcadedb
+        from arcadedb_embedded import jvm
+    path = tempfile.mkdtemp(prefix="gb_arcadedb_") + "/db"
+    heap = os.environ.get("ARCADEDB_HEAP", "4g")
+    with bc.timed() as t_jvm:
+        jvm.start_jvm(heap_size=heap)  # heap must match (else medium OOMs)
+    with bc.timed() as t_open:
+        ctx = arcadedb.create_database(path, jvm_kwargs={"heap_size": heap})
+        db = ctx.__enter__()
+    with bc.timed() as t_schema:
+        for v in ("User", "Post"):
+            db.command("sql", f"CREATE VERTEX TYPE {v}")
+            db.command("sql", f"CREATE PROPERTY {v}.id LONG")
+            db.command("sql", f"CREATE INDEX ON {v} (id) UNIQUE_HASH")  # point lookups (ex 09)
+        db.command("sql", "CREATE EDGE TYPE POSTED")
+        db.command("sql", "CREATE EDGE TYPE ANSWERS")
 
     pf = db.async_executor().get_parallel_level() > 1
-    t0 = time.time()
-    # vertices + edges via the tuned graph batch loader (ex 09), not per-row SQL
-    for vtype, ids in (("User", users), ("Post", posts)):
-        with db.graph_batch(batch_size=max(1, len(ids)), expected_edge_count=0,
-                            bidirectional=False, commit_every=max(1, len(ids)),
-                            use_wal=False, parallel_flush=pf) as b:
-            b.create_vertices(vtype, [{"id": i} for i in ids])
-    urid = {int(r["id"]): r["rid"] for r in
-            db.query("sql", "SELECT id, @rid as rid FROM User").to_list()}
-    prid = {int(r["id"]): r["rid"] for r in
-            db.query("sql", "SELECT id, @rid as rid FROM Post").to_list()}
-    for etype, edges, frm, to in (("POSTED", posted, urid, prid),
-                                  ("ANSWERS", answers, prid, prid)):
-        with db.graph_batch(batch_size=max(1, len(edges)), expected_edge_count=max(1, len(edges)),
-                            bidirectional=False, commit_every=max(1, len(edges)),
-                            use_wal=False, parallel_flush=pf) as b:
-            for a, c in edges:
-                b.new_edge(frm[a], etype, to[c])
-    load_s = time.time() - t0
+    with bc.timed() as t_ing:
+        for vtype, ids in (("User", users), ("Post", posts)):
+            with db.graph_batch(batch_size=max(1, len(ids)), expected_edge_count=0,
+                                bidirectional=False, commit_every=max(1, len(ids)),
+                                use_wal=False, parallel_flush=pf) as b:
+                b.create_vertices(vtype, [{"id": i} for i in ids])
+        urid = {int(r["id"]): r["rid"] for r in
+                db.query("sql", "SELECT id, @rid as rid FROM User").to_list()}
+        prid = {int(r["id"]): r["rid"] for r in
+                db.query("sql", "SELECT id, @rid as rid FROM Post").to_list()}
+        for etype, edges, frm, to in (("POSTED", posted, urid, prid),
+                                      ("ANSWERS", answers, prid, prid)):
+            with db.graph_batch(batch_size=max(1, len(edges)), expected_edge_count=max(1, len(edges)),
+                                bidirectional=False, commit_every=max(1, len(edges)),
+                                use_wal=False, parallel_flush=pf) as b:
+                for a, c in edges:
+                    b.new_edge(frm[a], etype, to[c])
 
     gav_build_s = 0.0
-    if workload == "olap":  # GAV accelerates the SAME OpenCypher queries (ex 10)
-        g0 = time.time()
-        db.command("sql", f"CREATE GRAPH ANALYTICAL VIEW {GAV_NAME} "
-                   "VERTEX TYPES (User, Post) EDGE TYPES (POSTED, ANSWERS) "
-                   "PROPERTIES (id) UPDATE MODE OFF")
-        while True:
-            row = db.query("sql", "SELECT FROM schema:graphAnalyticalViews WHERE name = ?",
-                           GAV_NAME).first()
-            if row is not None and row.get("status") == "READY":
-                break
-            if time.time() - g0 > 1800:
-                raise RuntimeError("GAV did not reach READY")
-            time.sleep(0.25)
-        gav_build_s = time.time() - g0
+    with bc.timed() as t_idx:
+        if workload == "olap":  # GAV accelerates the SAME OpenCypher queries (ex 10)
+            g0 = time.time()
+            db.command("sql", f"CREATE GRAPH ANALYTICAL VIEW {GAV_NAME} "
+                       "VERTEX TYPES (User, Post) EDGE TYPES (POSTED, ANSWERS) "
+                       "PROPERTIES (id) UPDATE MODE OFF")
+            while True:
+                row = db.query("sql", "SELECT FROM schema:graphAnalyticalViews WHERE name = ?",
+                               GAV_NAME).first()
+                if row is not None and row.get("status") == "READY":
+                    break
+                if time.time() - g0 > 1800:
+                    raise RuntimeError("GAV did not reach READY")
+                time.sleep(0.25)
+            gav_build_s = time.time() - g0
 
     def query(q):
         return len(db.query("opencypher", q).to_list())
@@ -136,12 +146,13 @@ def write(q):
         except Exception:
             db.rollback()
 
-    return dict(load_s=load_s, gav_build_s=gav_build_s, query=query, write=write,
-                close=lambda: ctx.__exit__(None, None, None),
+    return dict(import_s=t_imp.s, jvm_init_s=t_jvm.s, open_s=t_open.s, schema_s=t_schema.s,
+                ingest_s=t_ing.s, index_build_s=t_idx.s, gav_build_s=gav_build_s, db_path=path,
+                query=query, write=write, close=lambda: ctx.__exit__(None, None, None),
                 version=getattr(arcadedb, "__version__", "?"))
 
 
-BACKENDS = {"kuzu": be_kuzu, "arcadedb": be_arcadedb}
+BACKENDS = {"ladybug": be_ladybug, "arcadedb": be_arcadedb}
 
 
 def run_oltp(be, users, posts, n_ops, seed=0):
@@ -161,15 +172,25 @@ def run_oltp(be, users, posts, n_ops, seed=0):
         lat[k].append((time.time() - s) * 1000)
     total = time.time() - t0
     out = {"oltp_total_s": round(total, 3), "oltp_ops_per_s": round(n_ops / total, 1)}
+    alllat = []
     for k, v in lat.items():
-        if v:
-            out[f"{k}_p50_ms"] = round(st.median(v), 3)
-    return out
+        out.update(bc.latstats(k, v)); alllat.extend(v)
+    out.update(bc.latstats("oltp", alllat))
+    return out, lat
 
 
-def run_olap(be, reps=3):
-    per = [round(min(_time(be, q) for _ in range(reps)), 3) for q in OLAP]
-    return {"olap_query_ms": per, "olap_total_ms": round(sum(per), 3)}
+def run_olap(be, reps=7):
+    per_mean, per_std, raw = [], [], {}
+    for idx, q in enumerate(OLAP):
+        samples = [_time(be, q) for _ in range(reps)]
+        raw[f"olap_q{idx}"] = samples
+        s = bc.latstats(f"olap_q{idx}", samples)
+        per_mean.append(round(s[f"olap_q{idx}_mean_ms"], 3))
+        per_std.append(round(s[f"olap_q{idx}_std_ms"], 3))
+    out = {"olap_query_ms": per_mean, "olap_query_std_ms": per_std,
+           "olap_total_ms": round(sum(per_mean), 3)}
+    out.update(bc.latstats("olap", [x for v in raw.values() for x in v]))
+    return out, raw
 
 
 def _time(be, q):
@@ -187,14 +208,28 @@ def main():
 
     users, posts, posted, answers = load_graph(args.data_dir, args.limit)
     be = BACKENDS[args.backend](users, posts, posted, answers, args.workload)
+    ingest_s = be["ingest_s"]
     res = {"backend": args.backend, "lib_version": be["version"], "lane": "graph",
            "workload": args.workload, "n_users": len(users), "n_posts": len(posts),
-           "n_posted": len(posted), "n_answers": len(answers), "load_s": round(be["load_s"], 3)}
+           "n_posted": len(posted), "n_answers": len(answers),
+           "import_s": round(be["import_s"], 4), "jvm_init_s": round(be["jvm_init_s"], 4),
+           "open_s": round(be["open_s"], 4), "schema_s": round(be["schema_s"], 4),
+           "ingest_s": round(ingest_s, 3), "index_build_s": round(be["index_build_s"], 3),
+           "load_s": round(ingest_s, 3),  # continuity
+           "ingest_edges_per_s": round((len(posted) + len(answers)) / ingest_s, 1) if ingest_s else None}
     if be.get("gav_build_s"):
         res["gav_build_s"] = round(be["gav_build_s"], 3)
         res["gav"] = True
-    res.update(run_oltp(be, users, posts, args.ops) if args.workload == "oltp" else run_olap(be))
-    be["close"]()
+    if args.workload == "oltp":
+        m, raw = run_oltp(be, users, posts, args.ops)
+    else:
+        m, raw = run_olap(be)
+    res.update(m)
+    with bc.timed() as t_close:
+        be["close"]()
+    res["close_s"] = round(t_close.s, 4)
+    res["db_size_mb"] = bc.dir_size_mb(be["db_path"])
+    bc.dump_latencies(os.environ.get("RUN_LABEL"), raw)
     print("RESULT " + json.dumps(res))