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

phase 1b: --shard rank/total + parallel Makefile targets

Browse source 
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).
This commit is contained in:
russell@unturf.com 2026-04-27 11:10:25 -04:00
parent c81c1c0ff0
commit d0857fb43d
No known key found for this signature in database
3 changed files with 53 additions and 4 deletions
Download patch file Download diff file Expand all files Collapse all files

17
Makefile
View file

@ -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)

15
aborist/cli.py
View file

@ -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)")

25
aborist/sources/wikipedia.py
View file

@ -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]: