diff --git a/Makefile b/Makefile index 164e5bd..5e9a3ed 100644 --- a/Makefile +++ b/Makefile @@ -74,21 +74,42 @@ 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) -# Spawn N parallel shard processes. They parallelize parser CPU and -# serialize SQLite writes through WAL. Override SHARDS=N (default 4). +# Phase 1b: parallel shards into ONE shared SQLite. Workers parallelize +# parser CPU; writes serialize at the WAL writer-lock. ~1.3x wall on 4 cores. SHARDS ?= 4 -ingest-cur-parallel: bootstrap fetch-cur ## ingest cur with SHARDS=N (default 4) parallel processes +ingest-cur-parallel: bootstrap fetch-cur ## ingest cur with SHARDS=N processes -> one shared DB @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 +ingest-old-parallel: bootstrap fetch-old ## ingest old history with SHARDS=N processes -> one shared DB @for i in $$(seq 0 $$(($(SHARDS) - 1))); do \ $(ABORIST) --db $(DB) ingest --source wikipedia_old --path $(WP_OLD) \ --shard $$i/$(SHARDS) & \ done; wait +# Phase 2: attach-forever sharding. Each shard owns its own SQLite file — +# no WAL contention. Reads via `aborist --shards-dir ` attach +# all shards as UNION views. "Merge cost" = 0. +SHARDS_DIR ?= $(HOME)/.aborist/shards +ingest-cur-attached: bootstrap fetch-cur ## sharded ingest, no WAL contention (Phase 2) + @mkdir -p $(SHARDS_DIR) + @for i in $$(seq 0 $$(($(SHARDS) - 1))); do \ + $(ABORIST) ingest --source wikipedia_cur --path $(WP_CUR) \ + --shards-dir $(SHARDS_DIR) --shard $$i/$(SHARDS) & \ + done; wait + +ingest-old-attached: bootstrap fetch-old ## sharded ingest of old history (Phase 2) + @mkdir -p $(SHARDS_DIR) + @for i in $$(seq 0 $$(($(SHARDS) - 1))); do \ + $(ABORIST) ingest --source wikipedia_old --path $(WP_OLD) \ + --shards-dir $(SHARDS_DIR) --shard $$i/$(SHARDS) & \ + done; wait + +stats-shards: bootstrap ## cross-shard stats via UNION views over $(SHARDS_DIR) + $(ABORIST) --shards-dir $(SHARDS_DIR) stats + ingest: ingest-cur ## default ingest = cur (use ingest-old or *-parallel for full) verify: bootstrap ## round-trip Merkle proofs for VERIFY_N random documents diff --git a/aborist/cli.py b/aborist/cli.py index ea7a060..7bdbe92 100644 --- a/aborist/cli.py +++ b/aborist/cli.py @@ -11,7 +11,7 @@ from aborist import __version__ from aborist.ingest import ingest_source, verify_random_sample from aborist.search import FTS5Backend from aborist.sources import WikipediaCurDump -from aborist.store import DEFAULT_DB_PATH, connect, stats +from aborist.store import DEFAULT_DB_PATH, connect, connect_query, stats def _cmd_ingest(args: argparse.Namespace) -> int: @@ -27,7 +27,7 @@ def _cmd_ingest(args: argparse.Namespace) -> int: 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": + elif args.source == "html": # noqa: SIM114 — keep branch shape try: from aborist.sources import HtmlPageSource except ImportError: @@ -51,7 +51,26 @@ def _cmd_ingest(args: argparse.Namespace) -> int: print(f"unknown source: {args.source}", file=sys.stderr) return 2 - conn = connect(args.db) + # Resolve target DB: if --shards-dir is set with --shard, write to a + # per-shard file. Each shard owns its own SQLite file, so N parallel + # ingests have ZERO writer-lock contention. + target_db = args.db + if args.shards_dir: + if not args.shard: + print( + "--shards-dir requires --shard rank/total", + file=sys.stderr, + ) + return 2 + rank_str, total_str = args.shard.split("/", 1) + rank = int(rank_str) + total = int(total_str) + shards_dir = Path(args.shards_dir) + shards_dir.mkdir(parents=True, exist_ok=True) + digits = max(3, len(str(total - 1))) + target_db = shards_dir / f"{rank:0{digits}d}.db" + + conn = connect(target_db) try: result = ingest_source( conn, @@ -67,7 +86,11 @@ def _cmd_ingest(args: argparse.Namespace) -> int: def _cmd_search(args: argparse.Namespace) -> int: - conn = connect(args.db) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) try: backend = FTS5Backend(conn) hits = backend.search(args.query, limit=args.limit) @@ -101,7 +124,11 @@ def _cmd_search(args: argparse.Namespace) -> int: def _cmd_verify(args: argparse.Namespace) -> int: - conn = connect(args.db) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) try: result = verify_random_sample(conn, n=args.n) finally: @@ -119,7 +146,11 @@ def _cmd_distill(args: argparse.Namespace) -> int: except ValueError as e: print(str(e), file=sys.stderr) return 2 - conn = connect(args.db) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) try: result = distill_existing( conn, @@ -161,7 +192,11 @@ def _cmd_ask(args: argparse.Namespace) -> int: else: client = OpenAICompatibleClient(base_url=base_url, api_key=api_key) - conn = connect(args.db) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) try: result = ask( conn, @@ -180,7 +215,11 @@ def _cmd_ask(args: argparse.Namespace) -> int: def _cmd_providence(args: argparse.Namespace) -> int: """List providence_cache records for a document URI or source_root.""" - conn = connect(args.db) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) try: if args.document_uri: rows = conn.execute( @@ -223,7 +262,11 @@ def _cmd_providence(args: argparse.Namespace) -> int: def _cmd_evict(args: argparse.Namespace) -> int: from aborist.evict import evict_to_cold - conn = connect(args.db) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) try: result = evict_to_cold( conn, @@ -240,7 +283,11 @@ def _cmd_evict(args: argparse.Namespace) -> int: def _cmd_rehydrate(args: argparse.Namespace) -> int: from aborist.evict import rehydrate - conn = connect(args.db) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) try: if args.all_cold: roots = [ @@ -270,7 +317,11 @@ def _cmd_rehydrate(args: argparse.Namespace) -> int: def _cmd_stats(args: argparse.Namespace) -> int: - conn = connect(args.db) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) try: result = stats(conn) finally: @@ -283,7 +334,11 @@ def _cmd_analyze(args: argparse.Namespace) -> int: """Compression spectrum, depth distribution, audit chain integrity.""" import hashlib - conn = connect(args.db) + conn = ( + connect_query(args.db, shards_dir=args.global_shards_dir) + if args.global_shards_dir + else connect(args.db) + ) try: # Depth distribution. depth = conn.execute( @@ -400,6 +455,15 @@ def build_parser() -> argparse.ArgumentParser: default=DEFAULT_DB_PATH, help=f"path to aborist SQLite db (default: {DEFAULT_DB_PATH})", ) + p.add_argument( + "--shards-dir", + dest="global_shards_dir", + default=None, + help=( + "for read commands: attach all shards in this directory and " + "expose them as UNION views over the standard tables" + ), + ) sub = p.add_subparsers(dest="cmd", required=True) ingest = sub.add_parser("ingest", help="ingest documents from a source") @@ -446,6 +510,16 @@ def build_parser() -> argparse.ArgumentParser: "parallelize parser CPU and serialize writes via WAL" ), ) + ingest.add_argument( + "--shards-dir", + dest="shards_dir", + default=None, + help=( + "directory for attach-forever sharding. With --shard rank/total, " + "writes to shards-dir/.db instead of --db, removing the " + "WAL writer-lock contention entirely. Reads via aborist --shards-dir" + ), + ) ingest.set_defaults(func=_cmd_ingest) search = sub.add_parser("search", help="keyword search (VISUAL audit mode)") diff --git a/aborist/store.py b/aborist/store.py index e342ed5..859e80f 100644 --- a/aborist/store.py +++ b/aborist/store.py @@ -167,7 +167,7 @@ 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 writable connection, creating the parent dir + 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 @@ -188,6 +188,76 @@ def connect(db_path: Path | str = DEFAULT_DB_PATH) -> sqlite3.Connection: return conn +# Tables that exist in every shard with the same schema. Used to build +# cross-shard UNION views in connect_query(). +_SHARDABLE_TABLES = ( + "documents", + "chunks", + "merkle_nodes", + "edges", + "derivations", + "providence_cache", + "audit_events", + "falsifications", +) + + +def discover_shards(shards_dir: Path | str) -> list[Path]: + """Enumerate shard DB files in `shards_dir`. Returns sorted list of paths.""" + p = Path(shards_dir) + if not p.is_dir(): + return [] + return sorted(p.glob("*.db")) + + +def connect_query( + db_path: Path | str | None = None, + shards_dir: Path | str | None = None, +) -> sqlite3.Connection: + """Open a read-only-style connection that surfaces ALL shards as one DB. + + If `shards_dir` is set, every `*.db` in it is ATTACHed and UNION ALL views + are created over the standard tables so existing queries (`SELECT * FROM + documents`) work unchanged across shards. Reads only — writes still go + through `connect()` against a specific shard. + + If `shards_dir` is None, returns a normal `connect(db_path)` for back-compat. + """ + if shards_dir is None: + return connect(db_path or DEFAULT_DB_PATH) + + shard_paths = discover_shards(shards_dir) + conn = sqlite3.connect(":memory:", isolation_level=None) + conn.row_factory = sqlite3.Row + conn.execute("PRAGMA temp_store = MEMORY") + + if not shard_paths: + # Nothing attached; create empty placeholder tables so callers don't crash. + conn.executescript(SCHEMA_SQL) + return conn + + aliases: list[str] = [] + for i, sp in enumerate(shard_paths): + alias = f"sh{i:03d}" + conn.execute(f"ATTACH DATABASE ? AS {alias}", (str(sp.resolve()),)) + aliases.append(alias) + + # UNION ALL views over the shardable tables. + for table in _SHARDABLE_TABLES: + unions = " UNION ALL ".join(f"SELECT * FROM {a}.{table}" for a in aliases) + conn.execute(f"CREATE TEMP VIEW {table} AS {unions}") + + # Stash the shard list for tools that want it. + conn.execute( + "CREATE TEMP TABLE _shards (shard_id TEXT, path TEXT, alias TEXT)" + ) + conn.executemany( + "INSERT INTO _shards (shard_id, path, alias) VALUES (?, ?, ?)", + [(p.stem, str(p.resolve()), a) for p, a in zip(shard_paths, aliases)], + ) + return conn + + @contextmanager def transaction(conn: sqlite3.Connection) -> Iterator[sqlite3.Connection]: """BEGIN IMMEDIATE / COMMIT / ROLLBACK around a block."""