10k Node Timing Test¶

Actions happening in 100ms or less appears instant to a human. This is what we are aiming for.

Spawns 10,000 nodes as a branching forest and times every operation. From 5 ingest roots, each new node usually continues the most recent line of work, sometimes backs up a few steps to try an alternative, and occasionally splinters off a much older node — so parents have varying numbers of children, like a real exploratory workflow. Nodes are metadata-only so the timings are dominated by ancestree itself, not artifact I/O. Re-run top to bottom; the store is wiped at the start of each run.

In [ ]:

import shutil
import time
import random
import io
import contextlib
from pathlib import Path
import ancestree

ROOT = Path("benchmark_10k_store")
if ROOT.exists():
    shutil.rmtree(ROOT)

rules = {"ingest": [None], "transform": ["ingest", "transform"]}
store = ancestree.LineageStore(
    ROOT, rules=rules, gen_triggers=["ingest"], dedupe=True, chunk=True
)

import shutil import time import random import io import contextlib from pathlib import Path import ancestree ROOT = Path("benchmark_10k_store") if ROOT.exists(): shutil.rmtree(ROOT) rules = {"ingest": [None], "transform": ["ingest", "transform"]} store = ancestree.LineageStore( ROOT, rules=rules, gen_triggers=["ingest"], dedupe=True, chunk=True )

In [2]:

import numpy as np


def timed(label, fn, reps=100):
    times, result = [], None
    for _ in range(reps):
        t = time.perf_counter()
        result = fn()
        times.append((time.perf_counter() - t) * 1000)
    hits = f"{len(result)} hits" if isinstance(result, list) else type(result).__name__
    # print(f"{label:<50} best of {reps}: {min(times):8.2f} ms   ({hits})")
    # print(f"{label:<50} worst of {reps}: {max(times):8.2f} ms   ({hits})")
    print(f"{label:<50} median of {reps} trials: {np.median(times):8.2f} ms   ({hits})")
    return result

import numpy as np def timed(label, fn, reps=100): times, result = [], None for _ in range(reps): t = time.perf_counter() result = fn() times.append((time.perf_counter() - t) * 1000) hits = f"{len(result)} hits" if isinstance(result, list) else type(result).__name__ # print(f"{label:<50} best of {reps}: {min(times):8.2f} ms ({hits})") # print(f"{label:<50} worst of {reps}: {max(times):8.2f} ms ({hits})") print(f"{label:<50} median of {reps} trials: {np.median(times):8.2f} ms ({hits})") return result

1. Create 10,000 nodes¶

In [3]:

N = 10_000
N_ROOTS = 5

random.seed(0)
nodes, ids = [], []
depth, children = {}, {}
t0 = time.perf_counter()
for i in range(N):
    if i < N_ROOTS:
        step, parent = "ingest", None
    else:
        step = "transform"
        # Exploratory branching: mostly continue the current line of work,
        # sometimes back up a few steps and try an alternative, occasionally
        # splinter off something much older.
        r = random.random()
        if r < 0.6:
            parent = nodes[-1]
        elif r < 0.9:
            parent = random.choice(nodes[-50:])
        else:
            parent = random.choice(nodes)
    with store.create_node(step_type=step, parent=parent) as node:
        node.add_meta("accuracy", round(random.random(), 4))
        node.add_meta("batch", i // 100)
    nodes.append(node)
    ids.append(node.node_id)
    depth[node.node_id] = 0 if parent is None else depth[parent.node_id] + 1
    if parent is not None:
        children.setdefault(parent.node_id, []).append(node.node_id)
    if (i + 1) % 1000 == 0:
        print(f"{i + 1:>6} nodes   {time.perf_counter() - t0:6.1f}s elapsed")

elapsed = time.perf_counter() - t0
fanouts = [len(c) for c in children.values()]
print(f"\ncreated {N} nodes in {elapsed:.1f}s  ({elapsed / N * 1000:.2f} ms/node)")
print(
    f"tree shape: {N_ROOTS} roots, max depth {max(depth.values())}, "
    f"max fan-out {max(fanouts)}, "
    f"{sum(1 for f in fanouts if f > 1)} nodes with multiple children"
)

N = 10_000 N_ROOTS = 5 random.seed(0) nodes, ids = [], [] depth, children = {}, {} t0 = time.perf_counter() for i in range(N): if i < N_ROOTS: step, parent = "ingest", None else: step = "transform" # Exploratory branching: mostly continue the current line of work, # sometimes back up a few steps and try an alternative, occasionally # splinter off something much older. r = random.random() if r < 0.6: parent = nodes[-1] elif r < 0.9: parent = random.choice(nodes[-50:]) else: parent = random.choice(nodes) with store.create_node(step_type=step, parent=parent) as node: node.add_meta("accuracy", round(random.random(), 4)) node.add_meta("batch", i // 100) nodes.append(node) ids.append(node.node_id) depth[node.node_id] = 0 if parent is None else depth[parent.node_id] + 1 if parent is not None: children.setdefault(parent.node_id, []).append(node.node_id) if (i + 1) % 1000 == 0: print(f"{i + 1:>6} nodes {time.perf_counter() - t0:6.1f}s elapsed") elapsed = time.perf_counter() - t0 fanouts = [len(c) for c in children.values()] print(f"\ncreated {N} nodes in {elapsed:.1f}s ({elapsed / N * 1000:.2f} ms/node)") print( f"tree shape: {N_ROOTS} roots, max depth {max(depth.values())}, " f"max fan-out {max(fanouts)}, " f"{sum(1 for f in fanouts if f > 1)} nodes with multiple children" )

  1000 nodes     19.1s elapsed
  2000 nodes     37.2s elapsed
  3000 nodes     55.0s elapsed
  4000 nodes     73.7s elapsed
  5000 nodes     92.4s elapsed
  6000 nodes    112.3s elapsed
  7000 nodes    133.5s elapsed
  8000 nodes    152.9s elapsed
  9000 nodes    172.0s elapsed
 10000 nodes    191.8s elapsed

created 10000 nodes in 191.8s  (19.18 ms/node)
tree shape: 5 roots, max depth 153, max fan-out 5, 2196 nodes with multiple children

2. Searching and querying (warm, in-memory)¶

In [4]:

_ = timed("find_node(step_type='ingest')", lambda: store.find_node(step_type="ingest"))
_ = timed("find_node(batch=20)", lambda: store.find_node(batch=20))
_ = timed(
    "find_node(lambda accuracy > 0.99)",
    lambda: store.find_node(accuracy=lambda a: a is not None and a > 0.99),
)
_ = timed(
    "get_most_recent_node(step_type='transform')",
    lambda: store.get_most_recent_node(step_type="transform"),
)
_ = timed("get_child_nodes(first root)", lambda: store.get_child_nodes(ids[1]))
_ = timed("get_most_recent_node()", lambda: store.get_most_recent_node())

_ = timed("find_node(step_type='ingest')", lambda: store.find_node(step_type="ingest")) _ = timed("find_node(batch=20)", lambda: store.find_node(batch=20)) _ = timed( "find_node(lambda accuracy > 0.99)", lambda: store.find_node(accuracy=lambda a: a is not None and a > 0.99), ) _ = timed( "get_most_recent_node(step_type='transform')", lambda: store.get_most_recent_node(step_type="transform"), ) _ = timed("get_child_nodes(first root)", lambda: store.get_child_nodes(ids[1])) _ = timed("get_most_recent_node()", lambda: store.get_most_recent_node())

find_node(step_type='ingest')                      median of 100 trials:     1.89 ms   (5 hits)
find_node(batch=20)                                median of 100 trials:     1.97 ms   (100 hits)
find_node(lambda accuracy > 0.99)                  median of 100 trials:     2.25 ms   (97 hits)
get_most_recent_node(step_type='transform')        median of 100 trials:     4.23 ms   (Node)
get_child_nodes(first root)                        median of 100 trials:     2.40 ms   (2 hits)
get_most_recent_node()                             median of 100 trials:     3.52 ms   (Node)

3. Lineage traversal¶

In [11]:

deepest = max(depth, key=depth.get)  # tip of the longest exploration line
_ = timed(f"get_lineage (depth {depth[deepest]})", lambda: store.get_lineage(deepest))
_ = timed(
    "find_in_lineage(lambda accuracy > 0.5)",
    lambda: store.find_in_lineage(
        deepest, accuracy=lambda a: a is not None and a > 0.5
    ),
)

deepest = max(depth, key=depth.get) # tip of the longest exploration line _ = timed(f"get_lineage (depth {depth[deepest]})", lambda: store.get_lineage(deepest)) _ = timed( "find_in_lineage(lambda accuracy > 0.5)", lambda: store.find_in_lineage( deepest, accuracy=lambda a: a is not None and a > 0.5 ), )

get_lineage (depth 153)                            median of 100 trials:     0.28 ms   (154 hits)
find_in_lineage(lambda accuracy > 0.5)             median of 100 trials:     0.19 ms   (80 hits)

4. Cold start (fresh process simulation: new store instance loads the snapshot)¶

In [17]:

t = time.perf_counter()
store2 = ancestree.LineageStore(ROOT)
first = store2.find_node(batch=0)
print(
    f"cold start (init + load index + first query):    {(time.perf_counter() - t) * 1000:8.2f} ms"
)
_ = timed("same query again (warm)", lambda: store2.find_node(batch=0))

t = time.perf_counter() store2 = ancestree.LineageStore(ROOT) first = store2.find_node(batch=0) print( f"cold start (init + load index + first query): {(time.perf_counter() - t) * 1000:8.2f} ms" ) _ = timed("same query again (warm)", lambda: store2.find_node(batch=0))

cold start (init + load index + first query):       92.14 ms
same query again (warm)                            median of 100 trials:     2.00 ms   (100 hits)

5. Full index rebuild (the only disk crawl — recovery path)¶

In [18]:

t = time.perf_counter()
store.rebuild_db_from_disk()
print(f"rebuild_db_from_disk ({N} meta.json files): {time.perf_counter() - t:.2f} s")

t = time.perf_counter() store.rebuild_db_from_disk() print(f"rebuild_db_from_disk ({N} meta.json files): {time.perf_counter() - t:.2f} s")

rebuild_db_from_disk (10000 meta.json files): 1.06 s

6. Prune the first root's subtree¶

In [24]:

def subtree_size(node_id):
    size, stack = 0, [node_id]
    while stack:
        nid = stack.pop()
        size += 1
        stack.extend(children.get(nid, []))
    return size


n_sub = subtree_size(ids[0])
t = time.perf_counter()
with contextlib.redirect_stdout(io.StringIO()):
    store.prune(ids[0], dry_run=True)
print(
    f"prune dry-run (subtree of {n_sub}):  {(time.perf_counter() - t) * 1000:8.2f} ms"
)

t = time.perf_counter()
store.prune(ids[0], dry_run=False)
print(
    f"prune for real (subtree of {n_sub}): {(time.perf_counter() - t) * 1000:8.2f} ms"
)

def subtree_size(node_id): size, stack = 0, [node_id] while stack: nid = stack.pop() size += 1 stack.extend(children.get(nid, [])) return size n_sub = subtree_size(ids[0]) t = time.perf_counter() with contextlib.redirect_stdout(io.StringIO()): store.prune(ids[0], dry_run=True) print( f"prune dry-run (subtree of {n_sub}): {(time.perf_counter() - t) * 1000:8.2f} ms" ) t = time.perf_counter() store.prune(ids[0], dry_run=False) print( f"prune for real (subtree of {n_sub}): {(time.perf_counter() - t) * 1000:8.2f} ms" )

prune dry-run (subtree of 2):      0.34 ms
prune for real (subtree of 2):     0.19 ms

7. Web graph generation¶

Timing the HTML generation only — don't actually open a 10k node graph in a browser unless you enjoy fan noise.

In [25]:

t = time.perf_counter()
with contextlib.redirect_stdout(io.StringIO()):
    store.generate_web_graph()
print(f"generate_web_graph: {time.perf_counter() - t:.2f} s")
html = ROOT / "interactive_pipeline.html"
print(f"HTML size: {html.stat().st_size / 1e6:.1f} MB")

t = time.perf_counter() with contextlib.redirect_stdout(io.StringIO()): store.generate_web_graph() print(f"generate_web_graph: {time.perf_counter() - t:.2f} s") html = ROOT / "interactive_pipeline.html" print(f"HTML size: {html.stat().st_size / 1e6:.1f} MB")

generate_web_graph: 1.50 s
HTML size: 19.4 MB

8. What's on disk¶

In [28]:

files = [f for f in ROOT.rglob("*") if f.is_file()]
snap = ROOT / ".index.json"
print(f"snapshot (.index.json): {snap.stat().st_size / 1e6:.2f} MB")
print(
    f"store total:            {sum(f.stat().st_size for f in files) / 1e6:.1f} MB across {len(files)} files"
)

files = [f for f in ROOT.rglob("*") if f.is_file()] snap = ROOT / ".index.json" print(f"snapshot (.index.json): {snap.stat().st_size / 1e6:.2f} MB") print( f"store total: {sum(f.stat().st_size for f in files) / 1e6:.1f} MB across {len(files)} files" )

snapshot (.index.json): 2.39 MB
store total:            42.7 MB across 10002 files