From d0857fb43d3aac72e4c7fdbde9833b1f1de91741 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Mon, 27 Apr 2026 11:10:25 -0400 Subject: [PATCH] phase 1b: --shard rank/total + parallel Makefile targets MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Workers each parse the full SQL stream but only emit rows whose index satisfies (idx % total == rank). N processes cooperate against a single shared SQLite file; WAL+synchronous=NORMAL handles writer contention via BEGIN IMMEDIATE serialization. Implementation chose OS-level processes over in-process mp.Queue: the gain ceiling is identical (writer is the bottleneck either way) and the surface area is one new flag plus stride-filter logic instead of a queue-based pipeline with picklable workers. Makefile gains: ingest-cur-parallel SHARDS=4 ingest-old-parallel SHARDS=4 Benchmark on 20k cur articles: serial 55.3 s 361 docs/s 1.0x CPU usage 4-way parallel 43.1 s 464 docs/s ~3.0x CPU usage 1.28x wall Both runs landed 19,999 unique documents — correctness verified. The write-lock is now the bottleneck; further gains require Phase 2 attach-forever sharding (each shard owns its own SQLite file, queries fan out via ATTACH DATABASE, zero merge cost). --- Makefile | 17 ++++++++++++++++- aborist/cli.py | 15 ++++++++++++++- aborist/sources/wikipedia.py | 25 +++++++++++++++++++++++-- 3 files changed, 53 insertions(+), 4 deletions(-) diff --git a/Makefile b/Makefile index 2f055aa..164e5bd 100644 --- a/Makefile +++ b/Makefile @@ -74,7 +74,22 @@ ingest-cur: bootstrap fetch-cur ## ingest INGEST_LIMIT cur articles ingest-old: bootstrap fetch-old ## ingest INGEST_LIMIT old (history) revisions $(ABORIST) --db $(DB) ingest --source wikipedia_old --path $(WP_OLD) --limit $(INGEST_LIMIT) -ingest: ingest-cur ## default ingest = cur (use ingest-old or both for full) +# Spawn N parallel shard processes. They parallelize parser CPU and +# serialize SQLite writes through WAL. Override SHARDS=N (default 4). +SHARDS ?= 4 +ingest-cur-parallel: bootstrap fetch-cur ## ingest cur with SHARDS=N (default 4) parallel processes + @for i in $$(seq 0 $$(($(SHARDS) - 1))); do \ + $(ABORIST) --db $(DB) ingest --source wikipedia_cur --path $(WP_CUR) \ + --shard $$i/$(SHARDS) & \ + done; wait + +ingest-old-parallel: bootstrap fetch-old ## ingest old history with SHARDS=N parallel processes + @for i in $$(seq 0 $$(($(SHARDS) - 1))); do \ + $(ABORIST) --db $(DB) ingest --source wikipedia_old --path $(WP_OLD) \ + --shard $$i/$(SHARDS) & \ + done; wait + +ingest: ingest-cur ## default ingest = cur (use ingest-old or *-parallel for full) verify: bootstrap ## round-trip Merkle proofs for VERIFY_N random documents $(ABORIST) --db $(DB) verify -n $(VERIFY_N) diff --git a/aborist/cli.py b/aborist/cli.py index 126bb3e..ea7a060 100644 --- a/aborist/cli.py +++ b/aborist/cli.py @@ -22,7 +22,11 @@ def _cmd_ingest(args: argparse.Namespace) -> int: from aborist.sources import WikipediaSqlDump table = "cur" if args.source == "wikipedia_cur" else "old" - src = WikipediaSqlDump(path=args.path, table=table) + shard = None + if args.shard: + rank_str, total_str = args.shard.split("/", 1) + shard = (int(rank_str), int(total_str)) + src = WikipediaSqlDump(path=args.path, table=table, shard=shard) elif args.source == "html": try: from aborist.sources import HtmlPageSource @@ -433,6 +437,15 @@ def build_parser() -> argparse.ArgumentParser: default=200, help="documents per SQLite transaction (default 200)", ) + ingest.add_argument( + "--shard", + default=None, + help=( + "rank/total — yield only every N-th doc for parallel ingest. " + "spawn N processes, each with --shard 0/N, 1/N, ... they " + "parallelize parser CPU and serialize writes via WAL" + ), + ) ingest.set_defaults(func=_cmd_ingest) search = sub.add_parser("search", help="keyword search (VISUAL audit mode)") diff --git a/aborist/sources/wikipedia.py b/aborist/sources/wikipedia.py index bace0a6..8e85bc8 100644 --- a/aborist/sources/wikipedia.py +++ b/aborist/sources/wikipedia.py @@ -228,6 +228,11 @@ class WikipediaSqlDump(Source): the first four column positions: id, namespace, title, text. The `cur` table has `cur_is_redirect` at position 10 (we skip redirects); `old` has no redirect flag (every revision is real). + + Shard support: pass `shard=(rank, total)` and the source yields only + docs whose 0-based index satisfies `index % total == rank`. Useful for + spawning N parallel ingest processes against the same dump file — + parser CPU runs in parallel, writes serialize at the WAL writer lock. """ def __init__( @@ -237,6 +242,7 @@ class WikipediaSqlDump(Source): table: str = "cur", namespace: int = NAMESPACE_MAIN, base_uri: str = "https://en.wikipedia.org/wiki/", + shard: tuple[int, int] | None = None, ): if table not in ("cur", "old"): raise ValueError("table must be 'cur' or 'old'") @@ -245,9 +251,18 @@ class WikipediaSqlDump(Source): self.namespace = namespace self.base_uri = base_uri self.source_type = f"wikipedia_{table}" + if shard is not None: + rank, total = shard + if not (0 <= rank < total) or total < 1: + raise ValueError( + f"invalid shard {shard}: need 0 <= rank < total, total >= 1" + ) + self.shard = shard def iter_documents(self) -> Iterator[Document]: opener = bz2.open if str(self.path).endswith(".bz2") else open + rank, total = (0, 1) if self.shard is None else self.shard + idx = 0 with opener(self.path, "rt", encoding="utf-8", errors="replace") as f: for stmt in _iter_insert_statements(f): head = stmt.lstrip() @@ -258,14 +273,20 @@ class WikipediaSqlDump(Source): if vidx < 0: continue table_clause = head[:vidx].lower() - # Whitespace-bounded match: "INSERT INTO cur" must not also - # accept "INSERT INTO cursor" or similar. token = f" {self.table} " if token not in (table_clause + " "): continue payload = head[vidx + len("VALUES"):] payload = payload.rstrip().rstrip(";").rstrip() for row in _split_values_tuples(payload): + # Stride filter applied at the row index — every shard sees + # the full SQL stream but only emits its share. Wasted + # parser work is acceptable; bz2 decompression in parallel + # is faster than serializing through one decompressor. + if idx % total != rank: + idx += 1 + continue + idx += 1 yield from self._row_to_doc(row) def _row_to_doc(self, row: list[str | None]) -> Iterator[Document]: