Okasaki style time traveling database experiment.

okasaki_sqlite.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = []
# ///
"""Time travel: SQLite patterns vs Okasaki-style persistence. Value-checked.

Three ways to get "query the table AS OF transaction T":

  SQLITE-AUDIT   history table kvh(k, tx, v) PK(k, tx DESC), in-memory,
                 WITHOUT ROWID. Snapshot = remember the tx counter (O(1)).
                 AS-OF point = index seek; AS-OF range = correlated
                 per-key MAX(tx<=T) — the practitioner pattern.
  PERSISTENT     path-copying treap (immutable nodes, priority = hash(k)).
                 Snapshot = capture root (O(1)). Everything O(log n)
                 expected, at the price of ALLOC + pointer chasing —
                 exactly the PtrRec cost profile the emitter prices.
  DICT-COPY      dict + full copy per snapshot: O(n) snapshots, O(1)
                 queries. The rigid-shape corner of the triangle (LIST-like:
                 one shape per size, no sharing possible).

VALUE GATE: for sampled (snapshot, key) pairs and sampled ranges, all three
must agree with replay-from-log ground truth before any timing is reported.

Cost-model predictions (printed, then measured):
  P1 snapshot: persistent/audit O(1) vs dict-copy O(n) — ratio grows with n
  P2 AS-OF point: both O(log); SQLite's C constants beat Python's treap
     (an honest predicted LOSS for the persistent side)
  P3 AS-OF range: treap O(k + log n) vs audit O(k·log h) — the correlated
     dedup is where the audit pattern pays
Workload sized small: ≤ ~40 MB, target < 15 s total.
"""
from __future__ import annotations

import random
import sqlite3
import sys
import time

N_KEYS = 8192
OPS = 24_000
SNAP_EVERY = 240          # -> 100 snapshots
POINT_Q = 6_000
RANGE_Q = 60
RANGE_W = 128
SEED = 1234


# ------------------------------------------------------- persistent treap --
def t_insert(node, k, v):
    if node is None:
        return (k, v, hash((k, 0x9E37)) & 0xFFFFFFFF, None, None)
    nk, nv, np_, l, r = node
    if k == nk:
        return (nk, v, np_, l, r)
    if k < nk:
        l2 = t_insert(l, k, v)
        if l2[2] > np_:                       # rotate right
            lk, lv, lp, ll, lr = l2
            return (lk, lv, lp, ll, (nk, nv, np_, lr, r))
        return (nk, nv, np_, l2, r)
    r2 = t_insert(r, k, v)
    if r2[2] > np_:                           # rotate left
        rk, rv, rp, rl, rr = r2
        return (rk, rv, rp, (nk, nv, np_, l, rl), rr)
    return (nk, nv, np_, l, r2)


def t_get(node, k):
    while node is not None:
        nk = node[0]
        if k == nk:
            return node[1]
        node = node[3] if k < nk else node[4]
    return None


def t_range(node, lo, hi, out):
    if node is None:
        return
    nk = node[0]
    if nk > lo:
        t_range(node[3], lo, hi, out)
    if lo <= nk <= hi:
        out.append((nk, node[1]))
    if nk < hi:
        t_range(node[4], lo, hi, out)


# ---------------------------------------------------------------- driver ---
def main() -> int:
    rng = random.Random(SEED)
    ops = [(i + 1, rng.randrange(N_KEYS), rng.randrange(1 << 30))
           for i in range(OPS)]
    snap_txs = [tx for tx, _, _ in ops if tx % SNAP_EVERY == 0]

    # ---- SQLite audit pattern -------------------------------------------
    db = sqlite3.connect(":memory:")
    db.executescript("""
        PRAGMA synchronous=OFF;
        CREATE TABLE kvh(k INTEGER, tx INTEGER, v INTEGER,
                         PRIMARY KEY (k, tx DESC)) WITHOUT ROWID;""")
    t0 = time.time()
    db.executemany("INSERT INTO kvh VALUES (?,?,?)",
                   [(k, tx, v) for tx, k, v in ops])
    db.commit()
    t_sql_ingest = time.time() - t0

    # ---- persistent treap, capturing roots at snapshots -----------------
    roots = {}
    root = None
    t0 = time.time()
    for tx, k, v in ops:
        root = t_insert(root, k, v)
        if tx % SNAP_EVERY == 0:
            roots[tx] = root                   # O(1) snapshot
    t_treap_ingest = time.time() - t0

    # ---- dict-copy baseline ---------------------------------------------
    copies = {}
    d = {}
    t0 = time.time()
    t_copy_snap = 0.0
    for tx, k, v in ops:
        d[k] = v
        if tx % SNAP_EVERY == 0:
            s0 = time.time()
            copies[tx] = d.copy()              # O(n) snapshot
            t_copy_snap += time.time() - s0
    t_dict_ingest = time.time() - t0

    # ---- VALUE GATE: replay ground truth on samples ----------------------
    def replay(T):
        gt = {}
        for tx, k, v in ops:
            if tx > T:
                break
            gt[k] = v
        return gt

    q_point = [(rng.choice(snap_txs), rng.randrange(N_KEYS))
               for _ in range(POINT_Q)]
    q_range = [(rng.choice(snap_txs), rng.randrange(N_KEYS - RANGE_W))
               for _ in range(RANGE_Q)]
    cur = db.cursor()
    for T, k in rng.sample(q_point, 60):
        gt = replay(T).get(k)
        sq = cur.execute("SELECT v FROM kvh WHERE k=? AND tx<=? "
                         "ORDER BY tx DESC LIMIT 1", (k, T)).fetchone()
        assert (sq[0] if sq else None) == gt == t_get(roots[T], k) \
            == copies[T].get(k), f"VALUE MISMATCH at (T={T}, k={k})"
    for T, lo in q_range[:8]:
        gt = sorted((k, v) for k, v in replay(T).items()
                    if lo <= k <= lo + RANGE_W)
        out = []
        t_range(roots[T], lo, lo + RANGE_W, out)
        sq = cur.execute(
            "SELECT k, v FROM kvh h WHERE k BETWEEN ? AND ? AND tx = "
            "(SELECT MAX(tx) FROM kvh WHERE k=h.k AND tx<=?) ORDER BY k",
            (lo, lo + RANGE_W, T)).fetchall()
        assert out == gt == [tuple(r) for r in sq], "RANGE MISMATCH"
    print("VALUE GATE: all three implementations agree with replay — OK\n")

    # ---- timings ----------------------------------------------------------
    print(f"ingest ({OPS:,} updates, {len(snap_txs)} snapshots):")
    print(f"  sqlite-audit {t_sql_ingest*1e3:7.1f} ms   "
          f"treap {t_treap_ingest*1e3:7.1f} ms   "
          f"dict {t_dict_ingest*1e3:7.1f} ms "
          f"(of which copy-snapshots {t_copy_snap*1e3:.1f} ms)")
    per_snap_copy = t_copy_snap / len(snap_txs) * 1e6
    print(f"P1 snapshot cost: persistent/audit ≈ O(1); "
          f"dict-copy measured {per_snap_copy:,.0f} µs/snapshot "
          f"(≈ n·copy — grows with table size)\n")

    t0 = time.time()
    for T, k in q_point:
        cur.execute("SELECT v FROM kvh WHERE k=? AND tx<=? "
                    "ORDER BY tx DESC LIMIT 1", (k, T)).fetchone()
    t_sql_pt = (time.time() - t0) / POINT_Q * 1e6
    t0 = time.time()
    for T, k in q_point:
        t_get(roots[T], k)
    t_tr_pt = (time.time() - t0) / POINT_Q * 1e6
    t0 = time.time()
    for T, k in q_point:
        copies[T].get(k)
    t_dc_pt = (time.time() - t0) / POINT_Q * 1e6
    print(f"P2 AS-OF point query (µs/query): sqlite {t_sql_pt:6.2f}   "
          f"treap {t_tr_pt:6.2f}   dict-copy {t_dc_pt:6.2f}")

    t0 = time.time()
    for T, lo in q_range:
        cur.execute(
            "SELECT k, v FROM kvh h WHERE k BETWEEN ? AND ? AND tx = "
            "(SELECT MAX(tx) FROM kvh WHERE k=h.k AND tx<=?) ORDER BY k",
            (lo, lo + RANGE_W, T)).fetchall()
    t_sql_rg = (time.time() - t0) / RANGE_Q * 1e3
    t0 = time.time()
    for T, lo in q_range:
        out = []
        t_range(roots[T], lo, lo + RANGE_W, out)
    t_tr_rg = (time.time() - t0) / RANGE_Q * 1e3
    print(f"P3 AS-OF range scan, width {RANGE_W} (ms/scan): "
          f"sqlite-audit {t_sql_rg:6.3f}   treap {t_tr_rg:6.3f}   "
          f"-> audit/persistent = {t_sql_rg/t_tr_rg:4.1f}x\n")

    mem_nodes = OPS * 14   # ~ops·log2(n) path-copied nodes, order estimate
    print(f"space (order estimates): audit rows = {OPS:,}; "
          f"treap path-copied nodes ≈ {mem_nodes:,}; "
          f"dict copies = {len(snap_txs)} × {N_KEYS:,} entries")
    print("triangle: dict-copy buys O(1) queries with O(n) snapshots; "
          "audit buys cheap ingest with the range-dedup tax; persistence "
          "buys O(1) snapshots + O(log) everything at ALLOC/pointer cost.")
    return 0


if __name__ == "__main__":
    sys.exit(main())