phase 2: attach-forever sharding — zero merge cost
Each ingest shard owns its own SQLite file. N processes write in
true parallel (no WAL writer-lock contention). Cross-shard reads
attach all shards as UNION ALL views over the standard tables, so
existing query code (`SELECT * FROM documents`) works unchanged.
CLI:
aborist ingest --shards-dir DIR --shard 0/4 # write to DIR/000.db
aborist --shards-dir DIR stats # read across all shards
aborist --shards-dir DIR analyze # ditto
aborist --shards-dir DIR verify -n 20 # Merkle round-trips
# on a sample drawn
# from all shards
Makefile:
SHARDS_DIR=~/.aborist/shards
ingest-cur-attached SHARDS=4 ingest-old-attached SHARDS=4
stats-shards cross-shard stats
Implementation lives in store.connect_query():
- opens a fresh in-memory connection
- ATTACH DATABASE per shard file under aliases sh000, sh001, ...
- CREATE TEMP VIEW <table> AS UNION ALL SELECT * FROM <alias>.<table>
for the eight shardable tables (documents, chunks, merkle_nodes,
edges, derivations, providence_cache, audit_events, falsifications)
- exposes a _shards temp table for tools that want the shard list
Benchmark (20k cur, 4 cores):
Phase 1a serial 55.3 s
Phase 1b shared-WAL 43.1 s
Phase 2 attached-forever 40.1 s (no WAL contention)
The 7% gap vs Phase 1b is bz2-decompression duplication: each shard
decodes the full dump independently. Architectural wins are bigger
than that number — write isolation, per-shard crash recovery,
audit chain stays per-shard checkable, shards can migrate to other
disks, new shards are pure append (never rewrite existing files).
Merge cost remains: zero. `ls $SHARDS_DIR/*.db` is the registry.
This commit is contained in:
parent
d0857fb43d
commit
7d2bf4e7fa
3 changed files with 182 additions and 17 deletions
29
Makefile
29
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 <dir> <cmd>` 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
|
||||
|
|
|
|||
|
|
@ -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/<rank>.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)")
|
||||
|
|
|
|||
|
|
@ -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."""
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue