engineering/arborist
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

ETL throughput: 4x parser, 2.5x full ingest

Browse source 
Profile of the prior implementation (cProfile on 2000-doc ingest):
  total       23.6s
  parser      20.1s   (85% of time, 27.7M list.append calls)
  DB writes    2.9s

Three changes compound:

1. PRAGMA pack in store.connect():
     synchronous=NORMAL    skip per-commit fsync (durable under WAL replay)
     cache_size=-65536     64 MB page cache
     temp_store=MEMORY     temp tables in RAM
     mmap_size=256 MB      page-cache reads bypass read() syscall

2. Batched transactions in ingest_source and distill_existing:
     200 docs/tx default, --batch-size CLI flag.
     One BEGIN/COMMIT per batch instead of per doc.
     Audit chain SELECT-then-INSERT still chains correctly because
     SQLite reads its own uncommitted writes within a transaction.
     Idempotent on collision (INSERT OR IGNORE / pre-existence check)
     so partial-batch re-runs are safe.

3. Parser rewrite (the actual bottleneck): replaced char-by-char
   list-append loops with str.find / string slicing.
     _decode_mysql_string fast-path: if no '\\\\' in input, return as-is.
     _split_values_tuples uses _find_string_end (str.find with
       backslash-parity check) instead of per-char scanning.
     _iter_insert_statements buffered scan (str.find for ';' and "'")
       crossing read-chunk boundaries via in_string flag.
     ~13x fewer Python function calls, semantics identical.
   _upsert_edges now batches via executemany.

After: 2000 docs in 6.0s; 5000 old revisions in 12.8s wall (was 32.3s).
Throughput 85 -> 333 docs/sec. 49 tests still passing.

Cheaper to fail = cheaper to experiment.
This commit is contained in:
russell@unturf.com 2026-04-27 08:51:52 -04:00
parent f623250135
commit 85ae761df8
No known key found for this signature in database
5 changed files with 501 additions and 317 deletions
Download patch file Download diff file Expand all files Collapse all files

23
aborist/cli.py
View file

@ -49,7 +49,13 @@ def _cmd_ingest(args: argparse.Namespace) -> int:
conn = connect(args.db)
try:
result = ingest_source(conn, src, chunker_name=args.chunker, limit=args.limit)
result = ingest_source(
conn,
src,
chunker_name=args.chunker,
limit=args.limit,
batch_size=args.batch_size,
)
finally:
conn.close()
print(json.dumps(result.__dict__, indent=2))
@ -118,6 +124,7 @@ def _cmd_distill(args: argparse.Namespace) -> int:
source_type=args.source_type,
limit=args.limit,
chunker_name=args.chunker,
batch_size=args.batch_size,
)
finally:
conn.close()
@ -419,6 +426,13 @@ def build_parser() -> argparse.ArgumentParser:
ingest.add_argument(
"--limit", type=int, default=None, help="cap number of documents"
)
ingest.add_argument(
"--batch-size",
dest="batch_size",
type=int,
default=200,
help="documents per SQLite transaction (default 200)",
)
ingest.set_defaults(func=_cmd_ingest)
search = sub.add_parser("search", help="keyword search (VISUAL audit mode)")
@ -458,6 +472,13 @@ def build_parser() -> argparse.ArgumentParser:
distill.add_argument(
"--limit", type=int, default=None, help="cap number of docs scanned"
)
distill.add_argument(
"--batch-size",
dest="batch_size",
type=int,
default=200,
help="cores written per SQLite transaction (default 200)",
)
distill.set_defaults(func=_cmd_distill)
ask_cmd = sub.add_parser(

314
aborist/distill/runner.py
View file

@ -1,14 +1,18 @@
"""Distill existing surface docs into core docs.
"""Distill existing surface (or core) docs into deeper cores.
For each surface row in `documents`:
1. Load its chunks (must be hot tier — content present).
2. Run the Distiller.
3. Compute the core's own Merkle tree (it's a normal Document).
4. Generate a Merkle proof for every contributing source chunk against the
source document_root. Pack into proof_blob (JSON).
5. Insert the core Document, its chunks, FTS rows, interior merkle nodes,
the derivation row, a `derived_from` edge, and an audit event.
6. Idempotent: re-running with the same distiller skips existing core_root.
Per source row, the runner:
1. Loads the source's chunks (must be hot tier — content present).
2. Runs the Distiller (pure function: source -> core artifact).
3. Computes the core's own Merkle tree.
4. Generates a Merkle proof for every contributing source chunk against the
source's document_root and packs them into proof_blob (JSON).
These are read-only / pure steps. Writes are accumulated into batches and
flushed in a single transaction per batch — same fsync-amortization approach
as ingest. With WAL+synchronous=NORMAL this gives ~10x throughput.
Idempotent: re-running with the same distiller skips existing core_root.
"""
from __future__ import annotations
@ -16,6 +20,7 @@ from __future__ import annotations
import json
import sqlite3
import time
from dataclasses import dataclass
from aborist import (
CANONICALIZATION_VERSION,
@ -27,6 +32,29 @@ from aborist.merkle import MerkleTree, hash_leaf, proof_to_dict
from aborist.store import append_audit, transaction
DEFAULT_BATCH_SIZE = 200
@dataclass
class _Pending:
"""Pre-computed distillation artifacts ready for a batched DB write."""
src_root: str
src_uri: str
src_compression_depth: int
core_uri: str
core_source_type: str
core_title: str | None
core_root: str
core_chunk_strs: list[str]
core_leaves: list[bytes]
core_tree: MerkleTree
proof_blob: str
src_chars: int
core_chars: int
contributing_count: int
def distill_existing(
conn: sqlite3.Connection,
distiller: Distiller,
@ -35,12 +63,13 @@ def distill_existing(
source_type: str | None = None,
limit: int | None = None,
chunker_name: str | None = None,
batch_size: int = DEFAULT_BATCH_SIZE,
) -> dict:
"""Distill documents of `kind` already in the store. Returns counts.
`kind='surface'` (default) compresses ingested docs into depth=1 cores.
`kind='core'` runs recursive distillation: a depth=N core compresses
into a depth=N+1 core. Each round tightens the planet toward its center.
into a depth=N+1 core.
"""
if kind not in ("surface", "core"):
raise ValueError("kind must be 'surface' or 'core'")
@ -57,14 +86,27 @@ def distill_existing(
)
if limit:
sql += f" LIMIT {int(limit)}"
surface_rows = conn.execute(sql, params).fetchall()
src_rows = conn.execute(sql, params).fetchall()
distilled = 0
skipped_cold = 0
skipped_existing = 0
skipped_empty = 0
counters = {
"scanned": len(src_rows),
"distilled": 0,
"skipped_existing": 0,
"skipped_cold": 0,
"skipped_empty": 0,
}
batch: list[_Pending] = []
for s in surface_rows:
def flush() -> None:
if not batch:
return
with transaction(conn):
for entry in batch:
outcome = _persist_no_tx(conn, entry, distiller.name, chunker.name, kind)
counters[outcome] += 1
batch.clear()
for s in src_rows:
src_root = s["document_root"]
chunk_rows = conn.execute(
"SELECT idx, leaf_hash, content FROM chunks "
@ -72,13 +114,13 @@ def distill_existing(
(src_root,),
).fetchall()
if any(r["content"] is None for r in chunk_rows):
skipped_cold += 1
counters["skipped_cold"] += 1
continue
chunk_strs = [r["content"] for r in chunk_rows]
src_doc = Document(
uri=s["document_uri"],
content="", # not used by current distillers
content="",
source_type=s["source_type"],
title=s["title"],
)
@ -87,14 +129,13 @@ def distill_existing(
core_text = canonicalize(result.core.content)
core_chunk_strs = chunker.split(core_text)
if not core_chunk_strs:
skipped_empty += 1
counters["skipped_empty"] += 1
continue
core_leaves = [hash_leaf(c.encode("utf-8")) for c in core_chunk_strs]
core_tree = MerkleTree.build(core_leaves)
core_root = core_tree.root.hex()
# Build source tree from the stored leaves to generate per-chunk proofs.
src_leaves = [bytes.fromhex(r["leaf_hash"]) for r in chunk_rows]
src_tree = MerkleTree.build(src_leaves)
contributing: list[dict] = []
@ -104,7 +145,6 @@ def distill_existing(
contributing.append(
{"src_chunk_idx": idx, "proof": proof_to_dict(p)}
)
proof_blob = json.dumps(
{
"process_id": distiller.name,
@ -116,113 +156,131 @@ def distill_existing(
sort_keys=True,
)
with transaction(conn):
existing = conn.execute(
"SELECT 1 FROM documents WHERE document_root = ?", (core_root,)
).fetchone()
if existing:
# Ensure derivation row is present even if core was made earlier.
conn.execute(
"INSERT OR IGNORE INTO derivations "
"(core_root, src_root, proof_blob, process_id, distilled_at) "
"VALUES (?, ?, ?, ?, ?)",
(core_root, src_root, proof_blob, distiller.name, int(time.time())),
)
conn.execute(
"INSERT OR IGNORE INTO edges "
"(src_root, dst_root, dst_uri, edge_type, anchor) "
"VALUES (?, ?, ?, 'derived_from', '')",
(core_root, src_root, s["document_uri"]),
)
skipped_existing += 1
continue
batch.append(
_Pending(
src_root=src_root,
src_uri=s["document_uri"],
src_compression_depth=s["compression_depth"] or 0,
core_uri=result.core.uri,
core_source_type=result.core.source_type,
core_title=result.core.title,
core_root=core_root,
core_chunk_strs=core_chunk_strs,
core_leaves=core_leaves,
core_tree=core_tree,
proof_blob=proof_blob,
src_chars=sum(len(c) for c in chunk_strs),
core_chars=sum(len(c) for c in core_chunk_strs),
contributing_count=len(contributing),
)
)
if len(batch) >= batch_size:
flush()
flush()
return counters
now = int(time.time())
new_depth = (s["compression_depth"] or 0) + 1
conn.execute(
"INSERT INTO documents "
"(document_root, document_uri, source_type, kind, compression_depth, "
" title, chunking_version, canonicalization_version, schema_version, "
" ingest_ts) "
"VALUES (?, ?, ?, 'core', ?, ?, ?, ?, ?, ?)",
(
core_root,
result.core.uri,
result.core.source_type,
new_depth,
result.core.title,
chunker.name,
CANONICALIZATION_VERSION,
SCHEMA_VERSION,
now,
),
)
conn.executemany(
"INSERT INTO chunks (document_root, idx, leaf_hash, content) "
"VALUES (?, ?, ?, ?)",
[
(core_root, i, core_leaves[i].hex(), core_chunk_strs[i])
for i in range(len(core_chunk_strs))
],
)
conn.executemany(
"INSERT INTO chunks_fts (document_root, idx, content) VALUES (?, ?, ?)",
[
(core_root, i, core_chunk_strs[i])
for i in range(len(core_chunk_strs))
],
)
interior: list[tuple] = []
for layer_idx in range(1, len(core_tree.layers)):
for node_idx, h in enumerate(core_tree.layers[layer_idx]):
interior.append((core_root, layer_idx, node_idx, h.hex()))
if interior:
conn.executemany(
"INSERT INTO merkle_nodes (document_root, layer, idx, hash) "
"VALUES (?, ?, ?, ?)",
interior,
)
conn.execute(
"INSERT INTO derivations "
"(core_root, src_root, proof_blob, process_id, distilled_at) "
"VALUES (?, ?, ?, ?, ?)",
(core_root, src_root, proof_blob, distiller.name, now),
)
conn.execute(
"INSERT OR IGNORE INTO edges "
"(src_root, dst_root, dst_uri, edge_type, anchor) "
"VALUES (?, ?, ?, 'derived_from', '')",
(core_root, src_root, s["document_uri"]),
)
src_chars = sum(len(c) for c in chunk_strs)
core_chars = sum(len(c) for c in core_chunk_strs)
append_audit(
conn,
event_type="derive",
subject_root=core_root,
body={
"src_root": src_root,
"process_id": distiller.name,
"src_kind": kind,
"compression_depth": new_depth,
"core_chunks": len(core_chunk_strs),
"src_chunks_used": len(contributing),
"src_chars": src_chars,
"core_chars": core_chars,
"compression_ratio": (
round(core_chars / src_chars, 4) if src_chars else 0.0
),
},
ts=now,
)
def _persist_no_tx(
conn: sqlite3.Connection,
p: _Pending,
process_id: str,
chunker_name: str,
src_kind: str,
) -> str:
"""Write one pending distillation. Caller wraps the batch in a transaction.
distilled += 1
Returns the counter key to bump: 'distilled' or 'skipped_existing'.
"""
existing = conn.execute(
"SELECT 1 FROM documents WHERE document_root = ?", (p.core_root,)
).fetchone()
if existing:
# Same core_root from another source — record the (core, src) edge
# and the alternative derivation row, but don't re-insert the doc.
now = int(time.time())
conn.execute(
"INSERT OR IGNORE INTO derivations "
"(core_root, src_root, proof_blob, process_id, distilled_at) "
"VALUES (?, ?, ?, ?, ?)",
(p.core_root, p.src_root, p.proof_blob, process_id, now),
)
conn.execute(
"INSERT OR IGNORE INTO edges "
"(src_root, dst_root, dst_uri, edge_type, anchor) "
"VALUES (?, ?, ?, 'derived_from', '')",
(p.core_root, p.src_root, p.src_uri),
)
return "skipped_existing"
return {
"scanned": len(surface_rows),
"distilled": distilled,
"skipped_existing": skipped_existing,
"skipped_cold": skipped_cold,
"skipped_empty": skipped_empty,
}
now = int(time.time())
new_depth = p.src_compression_depth + 1
conn.execute(
"INSERT INTO documents "
"(document_root, document_uri, source_type, kind, compression_depth, "
" title, chunking_version, canonicalization_version, schema_version, "
" ingest_ts) "
"VALUES (?, ?, ?, 'core', ?, ?, ?, ?, ?, ?)",
(
p.core_root,
p.core_uri,
p.core_source_type,
new_depth,
p.core_title,
chunker_name,
CANONICALIZATION_VERSION,
SCHEMA_VERSION,
now,
),
)
conn.executemany(
"INSERT INTO chunks (document_root, idx, leaf_hash, content) VALUES (?, ?, ?, ?)",
[
(p.core_root, i, p.core_leaves[i].hex(), p.core_chunk_strs[i])
for i in range(len(p.core_chunk_strs))
],
)
conn.executemany(
"INSERT INTO chunks_fts (document_root, idx, content) VALUES (?, ?, ?)",
[(p.core_root, i, p.core_chunk_strs[i]) for i in range(len(p.core_chunk_strs))],
)
interior: list[tuple] = []
for layer_idx in range(1, len(p.core_tree.layers)):
for node_idx, h in enumerate(p.core_tree.layers[layer_idx]):
interior.append((p.core_root, layer_idx, node_idx, h.hex()))
if interior:
conn.executemany(
"INSERT INTO merkle_nodes (document_root, layer, idx, hash) VALUES (?, ?, ?, ?)",
interior,
)
conn.execute(
"INSERT INTO derivations "
"(core_root, src_root, proof_blob, process_id, distilled_at) "
"VALUES (?, ?, ?, ?, ?)",
(p.core_root, p.src_root, p.proof_blob, process_id, now),
)
conn.execute(
"INSERT OR IGNORE INTO edges "
"(src_root, dst_root, dst_uri, edge_type, anchor) "
"VALUES (?, ?, ?, 'derived_from', '')",
(p.core_root, p.src_root, p.src_uri),
)
append_audit(
conn,
event_type="derive",
subject_root=p.core_root,
body={
"src_root": p.src_root,
"process_id": process_id,
"src_kind": src_kind,
"compression_depth": new_depth,
"core_chunks": len(p.core_chunk_strs),
"src_chunks_used": p.contributing_count,
"src_chars": p.src_chars,
"core_chars": p.core_chars,
"compression_ratio": (
round(p.core_chars / p.src_chars, 4) if p.src_chars else 0.0
),
},
ts=now,
)
return "distilled"

260
aborist/ingest.py
View file

@ -1,8 +1,12 @@
"""Ingest pipeline: Source -> normalize -> chunk -> merkle -> upsert.
Idempotent: re-ingesting the same Document is a no-op (document_root collision
is the upsert key). Edges with unresolved dst_root are stored with NULL and
is the upsert key). Edges with unresolved dst_root are stored as '' and are
backfilled later when the target document is ingested.
Performance: documents are accumulated and flushed in batches under a single
SQLite transaction. With WAL+synchronous=NORMAL this trades per-doc fsync
for per-batch fsync, typically 5-20x throughput on SSD.
"""
from __future__ import annotations
@ -13,7 +17,6 @@ from dataclasses import dataclass
from aborist import (
CANONICALIZATION_VERSION,
CHUNKING_VERSION,
SCHEMA_VERSION,
)
from aborist.document import Document, canonicalize, get_chunker
@ -22,6 +25,17 @@ from aborist.source import Source
from aborist.store import append_audit, transaction
DEFAULT_BATCH_SIZE = 200
@dataclass
class _DocArtifacts:
document_root: str
leaves: list[bytes]
chunk_strs: list[str]
tree: MerkleTree
@dataclass
class IngestStats:
seen: int = 0
@ -36,136 +50,152 @@ def ingest_source(
source: Source,
chunker_name: str | None = None,
limit: int | None = None,
batch_size: int = DEFAULT_BATCH_SIZE,
) -> IngestStats:
"""Ingest every document the source yields. Returns counts."""
chunker = get_chunker(chunker_name)
stats = IngestStats()
batch: list[tuple[Document, _DocArtifacts]] = []
def flush() -> None:
if not batch:
return
with transaction(conn):
for doc, art in batch:
if _insert_document_no_tx(conn, doc, art, chunker.name):
stats.inserted += 1
else:
stats.skipped_duplicate += 1
batch.clear()
for doc in source.iter_documents():
stats.seen += 1
if limit is not None and stats.seen > limit:
break
if _ingest_document(conn, doc, chunker):
stats.inserted += 1
else:
art = _compute_artifacts(doc, chunker)
if art is None:
# empty content: count as skipped duplicate so totals stay sensible
stats.skipped_duplicate += 1
continue
batch.append((doc, art))
if len(batch) >= batch_size:
flush()
flush()
# Recompute chunk/edge totals once at end.
stats.chunks_total = conn.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
stats.edges_total = conn.execute("SELECT COUNT(*) FROM edges").fetchone()[0]
return stats
def _ingest_document(
conn: sqlite3.Connection,
doc: Document,
chunker,
) -> bool:
"""Returns True if newly inserted, False if document_root already exists."""
def _compute_artifacts(doc: Document, chunker) -> _DocArtifacts | None:
"""Pure: canonicalize, chunk, hash leaves, build the tree. No DB access."""
text = canonicalize(doc.content)
chunk_strs = chunker.split(text)
if not chunk_strs:
return False
return None
leaves = [hash_leaf(c.encode("utf-8")) for c in chunk_strs]
tree = MerkleTree.build(leaves)
document_root = tree.root.hex()
return _DocArtifacts(
document_root=tree.root.hex(),
leaves=leaves,
chunk_strs=chunk_strs,
tree=tree,
)
with transaction(conn):
existing = conn.execute(
"SELECT 1 FROM documents WHERE document_root = ?", (document_root,)
).fetchone()
if existing:
# Same content → same root → idempotent skip. Still backfill edges
# in case URI is new, but don't re-hash.
_upsert_edges(conn, document_root, doc)
return False
# Versioned re-ingest: if a prior doc exists for this URI with a
# DIFFERENT root, the source content has changed since last ingest.
# Keep the prior version (lossless history); link the new doc back
# to it via a 'supersedes' edge so the lineage stays addressable.
prior = conn.execute(
"SELECT document_root FROM documents "
"WHERE document_uri = ? AND document_root != ? "
"ORDER BY ingest_ts DESC LIMIT 1",
(doc.uri, document_root),
).fetchone()
def _insert_document_no_tx(
conn: sqlite3.Connection,
doc: Document,
art: _DocArtifacts,
chunker_name: str,
) -> bool:
"""Insert one document's rows. Caller wraps the batch in a transaction.
ingest_ts = int(time.time())
conn.execute(
"INSERT INTO documents "
"(document_root, document_uri, source_type, kind, compression_depth, "
" title, chunking_version, canonicalization_version, schema_version, "
" ingest_ts) "
"VALUES (?, ?, ?, 'surface', 0, ?, ?, ?, ?, ?)",
(
document_root,
doc.uri,
doc.source_type,
doc.title,
chunker.name,
CANONICALIZATION_VERSION,
SCHEMA_VERSION,
ingest_ts,
),
)
Returns True if newly inserted, False if document_root already exists.
"""
existing = conn.execute(
"SELECT 1 FROM documents WHERE document_root = ?", (art.document_root,)
).fetchone()
if existing:
# Same content -> same root -> idempotent skip. Still backfill edges
# in case a new caller surfaces fresh outbound links.
_upsert_edges(conn, art.document_root, doc)
return False
# Insert chunks (tier='hot' default).
# Versioned re-ingest: same URI + different root -> link new -> old.
prior = conn.execute(
"SELECT document_root FROM documents "
"WHERE document_uri = ? AND document_root != ? "
"ORDER BY ingest_ts DESC LIMIT 1",
(doc.uri, art.document_root),
).fetchone()
ingest_ts = int(time.time())
conn.execute(
"INSERT INTO documents "
"(document_root, document_uri, source_type, kind, compression_depth, "
" title, chunking_version, canonicalization_version, schema_version, "
" ingest_ts) "
"VALUES (?, ?, ?, 'surface', 0, ?, ?, ?, ?, ?)",
(
art.document_root,
doc.uri,
doc.source_type,
doc.title,
chunker_name,
CANONICALIZATION_VERSION,
SCHEMA_VERSION,
ingest_ts,
),
)
conn.executemany(
"INSERT INTO chunks (document_root, idx, leaf_hash, content) "
"VALUES (?, ?, ?, ?)",
[
(art.document_root, i, art.leaves[i].hex(), art.chunk_strs[i])
for i in range(len(art.chunk_strs))
],
)
conn.executemany(
"INSERT INTO chunks_fts (document_root, idx, content) VALUES (?, ?, ?)",
[(art.document_root, i, art.chunk_strs[i]) for i in range(len(art.chunk_strs))],
)
interior: list[tuple] = []
for layer_idx in range(1, len(art.tree.layers)):
for node_idx, h in enumerate(art.tree.layers[layer_idx]):
interior.append((art.document_root, layer_idx, node_idx, h.hex()))
if interior:
conn.executemany(
"INSERT INTO chunks (document_root, idx, leaf_hash, content) "
"INSERT INTO merkle_nodes (document_root, layer, idx, hash) "
"VALUES (?, ?, ?, ?)",
[
(document_root, i, leaves[i].hex(), chunk_strs[i])
for i in range(len(chunk_strs))
],
interior,
)
# Insert FTS rows.
conn.executemany(
"INSERT INTO chunks_fts (document_root, idx, content) VALUES (?, ?, ?)",
[(document_root, i, chunk_strs[i]) for i in range(len(chunk_strs))],
)
# Insert interior Merkle layers (skip layer 0 — it's chunks.leaf_hash).
rows: list[tuple] = []
for layer_idx in range(1, len(tree.layers)):
for node_idx, h in enumerate(tree.layers[layer_idx]):
rows.append((document_root, layer_idx, node_idx, h.hex()))
if rows:
conn.executemany(
"INSERT INTO merkle_nodes (document_root, layer, idx, hash) "
"VALUES (?, ?, ?, ?)",
rows,
)
_upsert_edges(conn, document_root, doc)
# If we found a prior version for this URI, link new -> old.
if prior is not None:
conn.execute(
"INSERT OR IGNORE INTO edges "
"(src_root, dst_root, dst_uri, edge_type, anchor) "
"VALUES (?, ?, ?, 'supersedes', '')",
(document_root, prior["document_root"], doc.uri),
)
append_audit(
conn,
event_type="ingest",
subject_root=document_root,
body={
"document_uri": doc.uri,
"source_type": doc.source_type,
"chunks": len(chunk_strs),
"chunking_version": chunker.name,
"canonicalization_version": CANONICALIZATION_VERSION,
"schema_version": SCHEMA_VERSION,
"supersedes": prior["document_root"] if prior is not None else None,
},
ts=ingest_ts,
_upsert_edges(conn, art.document_root, doc)
if prior is not None:
conn.execute(
"INSERT OR IGNORE INTO edges "
"(src_root, dst_root, dst_uri, edge_type, anchor) "
"VALUES (?, ?, ?, 'supersedes', '')",
(art.document_root, prior["document_root"], doc.uri),
)
append_audit(
conn,
event_type="ingest",
subject_root=art.document_root,
body={
"document_uri": doc.uri,
"source_type": doc.source_type,
"chunks": len(art.chunk_strs),
"chunking_version": chunker_name,
"canonicalization_version": CANONICALIZATION_VERSION,
"schema_version": SCHEMA_VERSION,
"supersedes": prior["document_root"] if prior is not None else None,
},
ts=ingest_ts,
)
return True
@ -173,7 +203,11 @@ def _upsert_edges(conn: sqlite3.Connection, src_root: str, doc: Document) -> Non
"""Insert edges; backfill dst_root from documents table if URI matches.
Unresolved edges store dst_root=''; resolved edges store the target's root.
Batched via executemany — single SQLite call instead of N.
"""
if not doc.edges:
return
rows: list[tuple] = []
for e in doc.edges:
dst_root = e.dst_root or ""
if not dst_root:
@ -184,16 +218,19 @@ def _upsert_edges(conn: sqlite3.Connection, src_root: str, doc: Document) -> Non
).fetchone()
if row:
dst_root = row["document_root"]
conn.execute(
"INSERT OR IGNORE INTO edges "
"(src_root, dst_root, dst_uri, edge_type, anchor) VALUES (?, ?, ?, ?, ?)",
(src_root, dst_root, e.dst_uri or "", e.edge_type, e.anchor or ""),
rows.append(
(src_root, dst_root, e.dst_uri or "", e.edge_type, e.anchor or "")
)
conn.executemany(
"INSERT OR IGNORE INTO edges "
"(src_root, dst_root, dst_uri, edge_type, anchor) VALUES (?, ?, ?, ?, ?)",
rows,
)
def verify_random_sample(conn: sqlite3.Connection, n: int = 10) -> dict:
"""Sample N documents, regenerate Merkle proof for chunk 0, verify."""
from aborist.merkle import MerkleProof, ProofNode, hash_leaf, verify_proof
from aborist.merkle import hash_leaf, verify_proof
rows = conn.execute(
"SELECT document_root FROM documents ORDER BY RANDOM() LIMIT ?", (n,)
@ -213,21 +250,13 @@ def verify_random_sample(conn: sqlite3.Connection, n: int = 10) -> dict:
if not chunk_rows:
failed += 1
continue
# Rebuild the tree from leaves and verify chunk 0's proof against root.
leaves = [bytes.fromhex(r["leaf_hash"]) for r in chunk_rows]
# First content -> recompute leaf_hash to confirm content matches stored hash.
c0 = chunk_rows[0]
if c0["content"] is None:
# Cold tier: skip content-roundtrip but verify structural proof.
recomputed_leaf = leaves[0]
else:
if c0["content"] is not None:
recomputed_leaf = hash_leaf(c0["content"].encode("utf-8"))
if recomputed_leaf != leaves[0]:
failed += 1
continue
from aborist.merkle import MerkleTree
tree = MerkleTree.build(leaves)
if tree.root.hex() != document_root:
failed += 1
@ -237,5 +266,4 @@ def verify_random_sample(conn: sqlite3.Connection, n: int = 10) -> dict:
passed += 1
else:
failed += 1
return {"sampled": len(rows), "passed": passed, "failed": failed}

207
aborist/sources/wikipedia.py
View file

@ -25,75 +25,107 @@ WIKILINK_RE = re.compile(r"\[\[([^\]\|#\n\r]+)(?:#[^\]\|\n\r]*)?(?:\|[^\]\n\r]*)
NAMESPACE_MAIN = 0
_MYSQL_ESCAPE_MAP = {
"n": "\n",
"r": "\r",
"t": "\t",
"0": "\0",
"\\": "\\",
"'": "'",
'"': '"',
"Z": "\x1a",
}
def _decode_mysql_string(s: str) -> str:
"""Decode MySQL-escaped string content (outer quotes already stripped)."""
"""Decode MySQL-escaped string content (outer quotes already stripped).
Hot fast path: the vast majority of string fields contain no backslashes
at all. Returning the input unchanged for those skips the slow loop.
"""
if "\\" not in s:
return s
out: list[str] = []
i = 0
n = len(s)
# Cache the mapping locally for tight-loop dispatch.
while i < n:
c = s[i]
if c == "\\" and i + 1 < n:
nxt = s[i + 1]
mapped = {
"n": "\n",
"r": "\r",
"t": "\t",
"0": "\0",
"\\": "\\",
"'": "'",
'"': '"',
"Z": "\x1a",
}.get(nxt, nxt)
out.append(mapped)
i += 2
else:
out.append(c)
i += 1
# Jump forward to the next backslash with C-level find.
bs = s.find("\\", i)
if bs < 0:
out.append(s[i:])
break
if bs > i:
out.append(s[i:bs])
if bs + 1 >= n:
out.append("\\")
i = bs + 1
continue
nxt = s[bs + 1]
out.append(_MYSQL_ESCAPE_MAP.get(nxt, nxt))
i = bs + 2
return "".join(out)
def _find_string_end(payload: str, start: int) -> int:
"""Return position of the closing `'` for a string opened just before `start`.
Backslash escapes considered: even count of preceding backslashes => the
quote is a real terminator; odd count => the quote is escaped.
"""
n = len(payload)
i = start
while i < n:
pos = payload.find("'", i)
if pos < 0:
return -1
bs = 0
p = pos - 1
while p >= start and payload[p] == "\\":
bs += 1
p -= 1
if bs % 2 == 0:
return pos
i = pos + 1
return -1
def _split_values_tuples(payload: str) -> list[list[str | None]]:
"""Parse `(v1,v2,...),(...),...` into tuples of decoded strings or None."""
"""Parse `(v1,v2,...),(...),...` into tuples of decoded strings or None.
String slicing + str.find based. No per-char list.append. Roughly 5-10x
faster than the char-by-char version on real Wikipedia dumps.
"""
rows: list[list[str | None]] = []
i = 0
n = len(payload)
while i < n:
while i < n and payload[i] in " \t\n\r,":
i += 1
if i >= n or payload[i] != "(":
# Find the next '(' starting a tuple.
open_paren = payload.find("(", i)
if open_paren < 0:
break
i += 1
i = open_paren + 1
values: list[str | None] = []
while True:
while i < n:
# Skip leading whitespace inside the tuple.
while i < n and payload[i] in " \t":
i += 1
if i >= n:
break
if payload[i] == "'":
# Quoted string
i += 1
buf: list[str] = []
while i < n:
c = payload[i]
if c == "\\" and i + 1 < n:
# Preserve escape pair; decode later in one pass.
buf.append(c)
buf.append(payload[i + 1])
i += 2
elif c == "'":
i += 1
break
else:
buf.append(c)
i += 1
values.append(_decode_mysql_string("".join(buf)))
end = _find_string_end(payload, i + 1)
if end < 0:
return rows # malformed; bail
values.append(_decode_mysql_string(payload[i + 1 : end]))
i = end + 1
else:
# NULL or number
start = i
while i < n and payload[i] not in ",)":
i += 1
v = payload[start:i].strip()
# NULL or unquoted scalar; ends at , or )
end = i
while end < n and payload[end] not in ",)":
end += 1
v = payload[i:end].strip()
values.append(None if v.upper() == "NULL" else v)
i = end
while i < n and payload[i] in " \t":
i += 1
if i < n and payload[i] == ",":
@ -108,34 +140,67 @@ def _split_values_tuples(payload: str) -> list[list[str | None]]:
def _iter_insert_statements(file_obj: IO[str]) -> Iterator[str]:
"""Yield complete SQL statements (text up to ; outside a quoted string)."""
buf: list[str] = []
"""Yield complete SQL statements using buffered string find.
Reads the file in 512 KB chunks, accumulates pending text, and uses
str.find for ';' and "'" to scan. String state crosses chunk boundaries
via the `in_string` flag. Avoids the per-char Python loop that dominated
the hot path before.
"""
pending = ""
in_string = False
escape = False
def flush_complete():
"""Yield as many complete statements as exist in `pending`."""
nonlocal pending, in_string
i = 0
last_yield = 0
n = len(pending)
while i < n:
if in_string:
# Find string terminator (handle escapes).
end = i
while True:
pos = pending.find("'", end)
if pos < 0:
i = n # need more data
break
bs = 0
p = pos - 1
while p >= 0 and pending[p] == "\\":
bs += 1
p -= 1
if bs % 2 == 0:
in_string = False
i = pos + 1
break
end = pos + 1
if in_string:
break
continue
sc = pending.find(";", i)
qt = pending.find("'", i)
if sc < 0 and qt < 0:
i = n
break
if qt < 0 or (sc >= 0 and sc < qt):
yield pending[last_yield : sc + 1]
last_yield = sc + 1
i = sc + 1
else:
in_string = True
i = qt + 1
if last_yield > 0:
pending = pending[last_yield:]
while True:
chunk = file_obj.read(1 << 17)
chunk = file_obj.read(1 << 19)
if not chunk:
break
for c in chunk:
buf.append(c)
if escape:
escape = False
continue
if in_string:
if c == "\\":
escape = True
elif c == "'":
in_string = False
continue
if c == "'":
in_string = True
elif c == ";":
yield "".join(buf)
buf = []
if buf:
tail = "".join(buf).strip()
if tail:
yield tail
pending += chunk
yield from flush_complete()
if pending.strip():
yield pending
def _extract_wikilinks(text: str, base_uri: str) -> list[Edge]:

14
aborist/store.py
View file

@ -167,12 +167,24 @@ CREATE VIRTUAL TABLE IF NOT EXISTS chunks_fts USING fts5(
def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection:
"""Open a connection, creating the parent dir and applying schema if needed."""
"""Open a connection, creating the parent dir and applying schema if needed.
Performance pragmas applied per-connection. Under WAL (set in the schema):
- synchronous=NORMAL skips the per-commit fsync; durable up to the last
checkpoint (SQLite auto-checkpoints at WAL ~1000 frames).
- cache_size=-65536 = 64 MB page cache (reduces re-reads).
- temp_store=MEMORY keeps temp tables in RAM (no /tmp churn).
- mmap_size=256 MB lets reads come from page-cache without read() syscalls.
"""
p = Path(db_path)
p.parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(p, isolation_level=None) # autocommit; we'll BEGIN manually
conn.row_factory = sqlite3.Row
conn.executescript(SCHEMA_SQL)
conn.execute("PRAGMA synchronous = NORMAL")
conn.execute("PRAGMA cache_size = -65536")
conn.execute("PRAGMA temp_store = MEMORY")
conn.execute("PRAGMA mmap_size = 268435456")
return conn