wallet/fts-sidecar: real FTS5 in the cloud sidecar; delete custom BM25
Replaces the hand-rolled BM25 sidecar (arborist/wallet/sidecar.py, ~690 LOC) with a slim SQLite file that just COPIES the source shard's FTS5 shadow tables verbatim + minimal doc/chunk metadata. Cloud retrieval then runs SQLite FTS5 bm25() on the same bytes the local shard uses — bit-for-bit parity by construction. 5/5 source + audit_mode agreement on the smoke fixture between local corpus-query and cloud-query against the new manifest-fts.json. Why --- Custom binary sidecar was per-document BM25; main encyclopedia articles got length-normalized so hard that on "why did the dinosaurs go extinct?" "Edwina, the Dinosaur Who Didn't Know She Was Extinct" beat "Dinosaur" (measured cloud-vs-local divergence). Local FTS5 indexes per-chunk so each chunk is a moderate-length doc and the main article wins multiply. Different granularity, not a tuning knob — fix is to use the same indexer cloud-side. What ships ---------- - arborist/wallet/fts_sidecar_build.py — builder. ATTACH source shard, copy documents (root/uri/title only), copy chunks (id/root/idx/leaf only, NO content), CREATE VIRTUAL TABLE chunks_fts/documents_fts with same DDL as source, bulk-copy the four shadow tables verbatim, VACUUM. 8.78 GB shard → 2.15 GB sidecar (24.5%) in ~45 s; full 4-shard wiki corpus 37.4 GB → 8.1 GB (21.7%) in ~3 min. - arborist/wallet/bucket.py: FtsSidecarShardClient — downloads slim sidecar once into ~/.arborist/sidecar-fts-cache/<hash>.idx.db, opens read-only sqlite3 (check_same_thread=False for parallel shard fan- out), runs FTS5 MATCH locally. Chunk content fetches via blobs/<hash> with HTTP-range big-shard fallback when blobs aren't published. MultiShardSidecarCorpus simplified to fts_sidecar_url ∨ bucket-direct (both are FTS5 backends; merge by raw bm25 MIN ascending). - arborist/qa/corpus.py: SidecarBucketCorpus.higher_is_better=False (FTS5 bm25 is negative, lower=better). chunks_for_doc dispatches on fetch_chunk_body attr for the slim-FTS5 client. apply_title_boost imports tokenizer helpers from new arborist/qa/_text_norm.py. - arborist/qa/_text_norm.py — fold_accents, numeral_expand, tokenize_text, STOPWORDS — extracted from the deleted sidecar.py so apply_title_boost keeps its lexical shape. - arborist/cli.py: `arborist sidecar build-fts` subcommand; old `sidecar build`/`sidecar search` removed. cloud_query recognizes fts_sidecar_url + sidecar_url alike. - Makefile: `sidecar-build-fts` + `sidecar-build-fts-all` targets; `sidecar-build` + `sidecar-search` removed. - scripts/upload_fts_sidecars.py — boto3 producer: uploads slim sidecars to clones/sidecars-fts/<n>.idx.db, publishes clones/manifest-fts.json (4 wikipedia shards inherit existing shard_url for content fallback; ACL public-read; idempotent on size match). Existing manifest-sidecar.json untouched. - tests/test_qa_corpus_functional.py + test_qa_corpus_integration.py converted from build_sidecar → build_fts_sidecar; 6 fixtures pass. - bench/slim_fts_parity_bench.py — local 3-way bench (legacy/corpus/slim_fts) over the smoke fixture. Validation ---------- - Per-shard FTS5 parity: slim sidecar returns IDENTICAL rowids + bm25 scores to the source shard for top-10 of "dinosaurs extinct". - 3-way bench (legacy local / corpus local / slim-FTS5 over real bucket fallback): 5/5 source agreement AND 5/5 audit_mode agreement between corpus and slim_fts. Q5 legacy disagreement (Edwina vs Dinosaur) is the pre-existing 2000-line query() retrieval quirk, unrelated. - End-to-end cloud query against published manifest-fts.json (cold- start, ~149 s sidecar download once): STRICT · Dinosaur, every quote verified (2/2). - Full pytest suite: 2737 passed, 28 skipped, 1 xfailed. One pre- existing failure (tests/test_doc_counts.py — claim_pack docs row- count drift) and one pre-existing cold_object failure, both reproduce on main HEAD. Bucket state ------------ - s3://arborist/clones/sidecars-fts/00[0-3].idx.db (8.1 GB) — new - s3://arborist/clones/manifest-fts.json — new - s3://arborist/clones/manifest-sidecar.json — kept live (deprecated but still readable; downstream callers should switch to manifest-fts.json)
This commit is contained in:
parent
2eea5b5655
commit
d9fb6a9b69
13 changed files with 1087 additions and 984 deletions
10
Makefile
10
Makefile
|
|
@ -1604,11 +1604,6 @@ cloud-snapshot-root: bootstrap ## compute snapshot_root of the bucket-resident s
|
|||
@echo "# shard: $(SHARD_URL)" >&2
|
||||
@$(ARBORIST) cloud snapshot-root --shard-url "$(SHARD_URL)" --cache-mb $(or $(CACHE_MB),32)
|
||||
|
||||
sidecar-build: bootstrap ## build HTTP-optimized inverted-index sidecar [SHARD=path OUT=path]
|
||||
@test -n "$(SHARD)" || { echo 'usage: make sidecar-build SHARD=/path/to/shard.db OUT=/path/to/sidecar.bin'; exit 2; }
|
||||
@test -n "$(OUT)" || { echo 'OUT required'; exit 2; }
|
||||
@$(ARBORIST) sidecar build --shard "$(SHARD)" --out "$(OUT)"
|
||||
|
||||
sidecar-build-fts: bootstrap ## build slim FTS5 sidecar (.idx.db) from a shard [SHARD=path OUT=path]
|
||||
@test -n "$(SHARD)" || { echo 'usage: make sidecar-build-fts SHARD=/path/to/shard.db OUT=/path/to/shard.idx.db'; exit 2; }
|
||||
@test -n "$(OUT)" || { echo 'OUT required'; exit 2; }
|
||||
|
|
@ -1626,11 +1621,6 @@ sidecar-build-fts-all: bootstrap ## build slim FTS5 sidecar for every shard in S
|
|||
done
|
||||
@ls -lh "$$OUT_DIR"/*.idx.db 2>/dev/null || true
|
||||
|
||||
sidecar-search: bootstrap ## query a local sidecar.bin [Q="..." SIDECAR=/path/sidecar.bin LIMIT=N MODE=or|and]
|
||||
@test -n "$(Q)" || { echo 'usage: make sidecar-search Q="your question" SIDECAR=/path/sidecar.bin [LIMIT=N] [MODE=or|and]'; exit 2; }
|
||||
@test -n "$(SIDECAR)" || { echo 'SIDECAR=/path/sidecar.bin required'; exit 2; }
|
||||
@$(ARBORIST) sidecar search '$(Q)' --sidecar "$(SIDECAR)" --limit $(or $(LIMIT),8) --mode $(or $(MODE),or)
|
||||
|
||||
cloud-fetch-chunk: bootstrap ## fetch + hash-verify one chunk from a bucket [LEAF_HASH=hex BLOB_BASE=https://...]
|
||||
@test -n "$(LEAF_HASH)" || { echo 'usage: make cloud-fetch-chunk LEAF_HASH=hex BLOB_BASE=https://.../blobs'; exit 2; }
|
||||
@test -n "$(BLOB_BASE)" || { echo 'BLOB_BASE required (per-chunk blobs/<hash> base URL)'; exit 2; }
|
||||
|
|
|
|||
110
arborist/cli.py
110
arborist/cli.py
|
|
@ -7337,34 +7337,30 @@ def build_parser() -> argparse.ArgumentParser:
|
|||
sidecar_cmd = sub.add_parser(
|
||||
"sidecar",
|
||||
help=(
|
||||
"build/query an HTTP-optimized inverted-index sidecar — one "
|
||||
"small file per shard that the client downloads once and "
|
||||
"queries locally (sub-ms after warmup). Replaces bucket-"
|
||||
"direct FTS5 over WAN, which is prohibitive on multi-GB shards."
|
||||
"build a slim FTS5 sidecar — one small SQLite file per shard "
|
||||
"with the FTS5 index + minimal metadata (no chunk content). "
|
||||
"The client downloads it once and queries via real FTS5 bm25, "
|
||||
"bit-for-bit identical to a local shard."
|
||||
),
|
||||
)
|
||||
sidecar_sub = sidecar_cmd.add_subparsers(dest="sidecar_cmd", required=True)
|
||||
|
||||
sidecar_build = sidecar_sub.add_parser(
|
||||
"build",
|
||||
help="build sidecar.bin from a shard .db file",
|
||||
sidecar_build_fts = sidecar_sub.add_parser(
|
||||
"build-fts",
|
||||
help=(
|
||||
"build a slim FTS5 sidecar from a shard .db — copies the "
|
||||
"shard's FTS5 shadow tables verbatim + minimal doc/chunk "
|
||||
"metadata (no content)."
|
||||
),
|
||||
)
|
||||
sidecar_build.add_argument("--shard", required=True, help="path to shard .db")
|
||||
sidecar_build.add_argument("--out", required=True, help="output sidecar.bin path")
|
||||
sidecar_build.set_defaults(func=_cmd_sidecar_build)
|
||||
|
||||
sidecar_search = sidecar_sub.add_parser(
|
||||
"search",
|
||||
help="query a local sidecar.bin file",
|
||||
sidecar_build_fts.add_argument(
|
||||
"--shard", required=True, help="path to shard .db",
|
||||
)
|
||||
sidecar_search.add_argument("query", type=str)
|
||||
sidecar_search.add_argument("--sidecar", required=True, help="path to sidecar.bin")
|
||||
sidecar_search.add_argument("--limit", type=int, default=8)
|
||||
sidecar_search.add_argument(
|
||||
"--mode", choices=["or", "and"], default="or",
|
||||
help="OR (default) ranks by BM25; AND requires every query term",
|
||||
sidecar_build_fts.add_argument(
|
||||
"--out", required=True,
|
||||
help="output slim FTS5 sidecar path (recommended: <shard>.idx.db)",
|
||||
)
|
||||
sidecar_search.set_defaults(func=_cmd_sidecar_search)
|
||||
sidecar_build_fts.set_defaults(func=_cmd_sidecar_build_fts)
|
||||
|
||||
return p
|
||||
|
||||
|
|
@ -7639,16 +7635,40 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
ts = _time.time()
|
||||
manifest = load_bucket_manifest(args.bucket_url)
|
||||
extra_timings["manifest"] = _time.time() - ts
|
||||
any_sidecar = any(sh.get("sidecar_url") for sh in manifest.shards)
|
||||
# Detect what kind of sidecars (if any) the manifest publishes.
|
||||
# MultiShardSidecarCorpus internally dispatches per shard:
|
||||
# fts_sidecar_url → FtsSidecarShardClient (slim FTS5, real bm25)
|
||||
# sidecar_url → SidecarShardClient (legacy custom BM25)
|
||||
# neither → BucketClient (HttpRangeVFS on big shard)
|
||||
# So we can always go through MultiShardSidecarCorpus; the
|
||||
# MultiShardBucketCorpus path is only kept for legacy bucket-only
|
||||
# manifests with no sidecar of either kind (rare).
|
||||
any_sidecar = any(
|
||||
sh.get("sidecar_url") or sh.get("fts_sidecar_url")
|
||||
for sh in manifest.shards
|
||||
)
|
||||
n_fts_sidecars = sum(
|
||||
1 for sh in manifest.shards if sh.get("fts_sidecar_url")
|
||||
)
|
||||
n_custom_sidecars = sum(
|
||||
1 for sh in manifest.shards
|
||||
if sh.get("sidecar_url") and not sh.get("fts_sidecar_url")
|
||||
)
|
||||
progress.emit(
|
||||
"manifest.done", shards=len(manifest.shards),
|
||||
sidecars=sum(1 for sh in manifest.shards if sh.get("sidecar_url")),
|
||||
sidecars=n_fts_sidecars + n_custom_sidecars,
|
||||
fts_sidecars=n_fts_sidecars,
|
||||
ms=int(extra_timings["manifest"] * 1000),
|
||||
)
|
||||
progress.emit(
|
||||
"corpus.open.start",
|
||||
mode="sidecar" if any_sidecar else "bucket-direct",
|
||||
)
|
||||
if n_fts_sidecars and not n_custom_sidecars:
|
||||
mode = "fts-slim"
|
||||
elif n_custom_sidecars and not n_fts_sidecars:
|
||||
mode = "custom-sidecar"
|
||||
elif any_sidecar:
|
||||
mode = "sidecar-mixed"
|
||||
else:
|
||||
mode = "bucket-direct"
|
||||
progress.emit("corpus.open.start", mode=mode)
|
||||
ts = _time.time()
|
||||
if any_sidecar:
|
||||
multi = MultiShardSidecarCorpus(
|
||||
|
|
@ -7676,7 +7696,7 @@ def _cmd_cloud_query(args: argparse.Namespace) -> int:
|
|||
class _Single:
|
||||
manifest = type("M", (), {"shards": [
|
||||
{"url": client.endpoint.shard_url, "shard_idx": 0,
|
||||
"label": "single", "sidecar_url": None}
|
||||
"label": "single"}
|
||||
], "blob_base": client.endpoint.blob_base, "snapshot_root": None})()
|
||||
def fts_search(self, q, *, limit=8, raw=False):
|
||||
return client.fts_search(q, limit=limit, raw=raw)
|
||||
|
|
@ -7842,41 +7862,13 @@ def _render_cloud_query_human(result: dict, question: str) -> str:
|
|||
return "\n".join(lines)
|
||||
|
||||
|
||||
def _cmd_sidecar_build(args: argparse.Namespace) -> int:
|
||||
from arborist.wallet.sidecar import build_sidecar
|
||||
stats = build_sidecar(args.shard, args.out)
|
||||
def _cmd_sidecar_build_fts(args: argparse.Namespace) -> int:
|
||||
from arborist.wallet.fts_sidecar_build import build
|
||||
stats = build(args.shard, args.out)
|
||||
print(json.dumps(stats, indent=2))
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_sidecar_search(args: argparse.Namespace) -> int:
|
||||
from arborist.wallet.sidecar import SidecarReader
|
||||
import time as _t
|
||||
t0 = _t.time()
|
||||
sc = SidecarReader(args.sidecar)
|
||||
load_s = _t.time() - t0
|
||||
t0 = _t.time()
|
||||
hits = sc.search(args.query, limit=args.limit, mode=args.mode)
|
||||
query_s = _t.time() - t0
|
||||
print(json.dumps({
|
||||
"query": args.query,
|
||||
"mode": args.mode,
|
||||
"load_secs": round(load_s, 3),
|
||||
"query_secs": round(query_s, 4),
|
||||
"stats": sc.stats(),
|
||||
"hits": [
|
||||
{
|
||||
"document_root": h.document_root,
|
||||
"document_uri": h.document_uri,
|
||||
"title": h.title,
|
||||
"score": h.score,
|
||||
}
|
||||
for h in hits
|
||||
],
|
||||
}, indent=2, ensure_ascii=False))
|
||||
return 0
|
||||
|
||||
|
||||
def _cmd_cloud_fetch_chunk(args: argparse.Namespace) -> int:
|
||||
from arborist.merkle import hash_leaf
|
||||
client = _make_bucket_client(args)
|
||||
|
|
|
|||
84
arborist/qa/_text_norm.py
Normal file
84
arborist/qa/_text_norm.py
Normal file
|
|
@ -0,0 +1,84 @@
|
|||
"""Tokenization helpers shared across query paths.
|
||||
|
||||
Lifted from the (deleted) custom-sidecar binary format module so the
|
||||
title-boost reranker in arborist.qa.corpus keeps the same lexical
|
||||
shape: NFKD accent fold, lowercase, drop stopwords + 1-char tokens
|
||||
(numeric singletons preserved), Roman-Arabic numeral equivalence.
|
||||
|
||||
Stopword set is the single source of truth in
|
||||
arborist.wallet.bucket._FTS5_STOPWORDS (imported here so any change
|
||||
to the FTS5 sanitizer carries through to title-boost matching too).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import unicodedata
|
||||
|
||||
from arborist.wallet.bucket import _FTS5_STOPWORDS as STOPWORDS
|
||||
|
||||
_WORD_RE = re.compile(r"[A-Za-z0-9_]+")
|
||||
|
||||
|
||||
def fold_accents(text: str) -> str:
|
||||
"""NFKD-normalize + strip combining marks → ASCII-equivalent.
|
||||
|
||||
'pokémon' → 'pokemon' so the ASCII-only word regex below doesn't
|
||||
split it into ['pok', 'mon'].
|
||||
"""
|
||||
return "".join(
|
||||
c for c in unicodedata.normalize("NFKD", text)
|
||||
if not unicodedata.combining(c)
|
||||
)
|
||||
|
||||
|
||||
# Roman ↔ Arabic numeral equivalence (1..20). A user typing 'final
|
||||
# fantasy 7' should overlap a title 'Final Fantasy VII'. Beyond 20
|
||||
# titles use Arabic digits in practice.
|
||||
_NUMERAL_PAIRS: dict[str, str] = {
|
||||
"1": "i", "i": "1",
|
||||
"2": "ii", "ii": "2",
|
||||
"3": "iii", "iii": "3",
|
||||
"4": "iv", "iv": "4",
|
||||
"5": "v", "v": "5",
|
||||
"6": "vi", "vi": "6",
|
||||
"7": "vii", "vii": "7",
|
||||
"8": "viii", "viii": "8",
|
||||
"9": "ix", "ix": "9",
|
||||
"10": "x", "x": "10",
|
||||
"11": "xi", "xi": "11",
|
||||
"12": "xii", "xii": "12",
|
||||
"13": "xiii", "xiii": "13",
|
||||
"14": "xiv", "xiv": "14",
|
||||
"15": "xv", "xv": "15",
|
||||
"16": "xvi", "xvi": "16",
|
||||
"17": "xvii", "xvii": "17",
|
||||
"18": "xviii", "xviii": "18",
|
||||
"19": "xix", "xix": "19",
|
||||
"20": "xx", "xx": "20",
|
||||
}
|
||||
|
||||
|
||||
def numeral_expand(tokens) -> set:
|
||||
"""Return a set with each token's numeral-equivalent form added."""
|
||||
out = set(tokens)
|
||||
for t in tokens:
|
||||
alt = _NUMERAL_PAIRS.get(t)
|
||||
if alt is not None:
|
||||
out.add(alt)
|
||||
return out
|
||||
|
||||
|
||||
def tokenize_text(text: str) -> list[str]:
|
||||
"""Lowercase word tokens; drop stopwords + 1-char tokens (except
|
||||
numeric singletons like '7' — needed so numeral_expand can later
|
||||
pair '7' with 'VII'). Accent-fold first."""
|
||||
folded = fold_accents(text)
|
||||
out: list[str] = []
|
||||
for raw in _WORD_RE.findall(folded):
|
||||
t = raw.lower()
|
||||
if t in STOPWORDS:
|
||||
continue
|
||||
if len(t) <= 1 and not t.isdigit():
|
||||
continue
|
||||
out.append(t)
|
||||
return out
|
||||
|
|
@ -68,18 +68,26 @@ def apply_title_boost(
|
|||
- higher_is_better=False (FTS5 BM25): boost SUBTRACTED from score
|
||||
|
||||
Effective boost magnitude:
|
||||
max(0, overlap - extras/2) * boost
|
||||
max(0, overlap - extras) * boost
|
||||
|
||||
where overlap = |query_stems ∩ title_stems|
|
||||
and extras = |title_stems - query_stems|
|
||||
|
||||
The full-strength penalty (extras counted 1:1, not halved) keeps
|
||||
the boost concentrated on titles whose tokens are a SUBSET of the
|
||||
query. The dinosaur-extinction bench (2026-05-31) showed that
|
||||
extras/2 was too lenient: "Edwina the Dinosaur Who Didn't Know
|
||||
She Was Extinct" (overlap 2 with {dinosaur, extinct}, extras 3
|
||||
for {edwina, didn, know}) was getting +4 boost — enough to beat
|
||||
"Dinosaur" main article's +8 boost when the underlying BM25
|
||||
favored Edwina's short article by >4. Full penalty zeroes
|
||||
Edwina's boost, lets Dinosaur win.
|
||||
Stems strip possessive apostrophes + trailing-s plurals; both
|
||||
sides are numeral-expanded (7↔VII) and accent-folded (é→e).
|
||||
"""
|
||||
if not hits or not query.strip() or boost <= 0:
|
||||
return hits
|
||||
# Lazy-import: sidecar carries the tokenizer + numeral_expand;
|
||||
# importing at module top would loop on the from-sidecar imports.
|
||||
from arborist.wallet.sidecar import (
|
||||
from arborist.qa._text_norm import (
|
||||
_WORD_RE, STOPWORDS,
|
||||
fold_accents, numeral_expand, tokenize_text,
|
||||
)
|
||||
|
|
@ -110,7 +118,7 @@ def apply_title_boost(
|
|||
rescored.append(h)
|
||||
continue
|
||||
extras = len(title_tokens - query_stems)
|
||||
effective = max(0.0, overlap - extras / 2.0)
|
||||
effective = max(0.0, overlap - extras)
|
||||
if effective <= 0:
|
||||
rescored.append(h)
|
||||
continue
|
||||
|
|
@ -460,22 +468,22 @@ class MultiShardSqliteCorpus:
|
|||
|
||||
|
||||
class SidecarBucketCorpus:
|
||||
"""Adapter over MultiShardSidecarCorpus (sidecar BM25 + bucket-direct
|
||||
chunk reads). Default for `arborist cloud query`.
|
||||
"""Adapter over MultiShardSidecarCorpus (slim FTS5 sidecar +
|
||||
bucket-served chunk content). Default for `arborist cloud query`.
|
||||
|
||||
fts_body delegates to the sidecar's BM25 + title boost + extras
|
||||
penalty. fts_title / fts_phrase raise NotSupportedError until the
|
||||
sidecar format ships those indexes.
|
||||
fts_body delegates to FTS5 bm25(). fts_title / fts_phrase raise
|
||||
NotSupportedError — the slim sidecar carries a documents_fts
|
||||
table that could power fts_title, but the wiring isn't done yet.
|
||||
"""
|
||||
|
||||
name = "sidecar-bucket"
|
||||
higher_is_better = True # sidecar BM25 + boost is positive, larger = stronger
|
||||
# FTS5 bm25 is always negative; lower (more negative) = better.
|
||||
higher_is_better = False
|
||||
|
||||
def __init__(self, multi_shard):
|
||||
"""``multi_shard`` is a MultiShardSidecarCorpus (or any object
|
||||
with fts_search(query, limit) → [{document_root, document_uri,
|
||||
title, score, _shard_url}] and conn_for_shard(shard_url) for
|
||||
chunk reads)."""
|
||||
title, score, _shard_url}])."""
|
||||
self._mscorpus = multi_shard
|
||||
|
||||
def fts_body(self, query: str, *, limit: int = 8) -> list[Hit]:
|
||||
|
|
@ -494,29 +502,60 @@ class SidecarBucketCorpus:
|
|||
|
||||
def fts_title(self, query: str, *, limit: int = 8) -> list[Hit]:
|
||||
raise NotSupportedError(
|
||||
"fts_title not implemented for SidecarBucketCorpus — "
|
||||
"sidecar today fuses title-boost into fts_body. A separate "
|
||||
"title-only route would need a sidecar v3 with a title index."
|
||||
"fts_title not yet wired for SidecarBucketCorpus — the slim "
|
||||
"FTS5 sidecar already carries documents_fts (title index); "
|
||||
"exposing it just needs a per-shard MATCH route here."
|
||||
)
|
||||
|
||||
def fts_phrase(self, ngrams, *, limit: int = 8) -> list[Hit]:
|
||||
raise NotSupportedError(
|
||||
"fts_phrase not implemented for SidecarBucketCorpus — "
|
||||
"phrase index is sidecar v3 work."
|
||||
"FTS5 supports MATCH '\"phrase here\"' directly; the route "
|
||||
"just isn't surfaced here yet."
|
||||
)
|
||||
|
||||
def chunks_for_doc(
|
||||
self, document_root: str, *, limit: int | None = None
|
||||
) -> list[ChunkRow]:
|
||||
# We don't know which shard owns the doc without a lookup. The
|
||||
# caller of fts_body already has Hit.shard_id; the contract here
|
||||
# accepts a bare document_root and scans every shard's bucket
|
||||
# connection until one returns rows. Cheap because BucketClient
|
||||
# caches the .db pages it touched for the FTS search.
|
||||
# Bare document_root lookup → scan shards until one returns rows.
|
||||
# Two backend shapes, both routed through hasattr checks:
|
||||
# 1) FtsSidecarShardClient: chunk metadata local, content fetched
|
||||
# over HTTP from bucket blobs/<hash> (or shard fallback).
|
||||
# 2) BucketClient (no sidecar): content inline in chunks.content
|
||||
# of the big shard .db via apsw HttpRangeVFS.
|
||||
from arborist.compress import unpack_chunk
|
||||
|
||||
# Try each shard's conn.
|
||||
for sh in getattr(self._mscorpus, "manifest", None).shards:
|
||||
try:
|
||||
client = self._mscorpus._by_url[sh["url"]]
|
||||
except (KeyError, AttributeError):
|
||||
continue
|
||||
# Slim FTS5 client path.
|
||||
if hasattr(client, "fetch_chunk_body") and hasattr(
|
||||
client, "chunks_for_doc"
|
||||
):
|
||||
meta = client.chunks_for_doc(document_root)
|
||||
if not meta:
|
||||
continue
|
||||
if limit is not None:
|
||||
meta = meta[: int(limit)]
|
||||
out: list[ChunkRow] = []
|
||||
for m in meta:
|
||||
try:
|
||||
body_bytes = client.fetch_chunk_body(m["leaf_hash"])
|
||||
except Exception:
|
||||
continue
|
||||
text = unpack_chunk(body_bytes) or ""
|
||||
if not text:
|
||||
continue
|
||||
out.append(ChunkRow(
|
||||
document_root=document_root,
|
||||
idx=m["idx"], leaf_hash=m["leaf_hash"], content=text,
|
||||
))
|
||||
if out:
|
||||
return out
|
||||
continue
|
||||
# Bucket-direct path: apsw connection on the big shard.
|
||||
try:
|
||||
conn = self._mscorpus.conn_for_shard(sh["url"])
|
||||
except (KeyError, AttributeError):
|
||||
|
|
|
|||
|
|
@ -1,31 +0,0 @@
|
|||
"""CLI-callable sidecar build runner.
|
||||
|
||||
Tiny shim so shell scripts can do
|
||||
python3 -m arborist.wallet._sidecar_build_runner SRC DST
|
||||
without inlining multiline Python that gets eaten by shell quoting.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
import time
|
||||
|
||||
from arborist.wallet.sidecar import build_sidecar
|
||||
|
||||
|
||||
def main(src: str, dst: str) -> int:
|
||||
t0 = time.time()
|
||||
stats = build_sidecar(src, dst)
|
||||
elapsed = time.time() - t0
|
||||
print(
|
||||
f"build_secs={elapsed:.1f} "
|
||||
f"file_size_mb={stats['file_bytes']/1024**2:.1f} "
|
||||
f"terms={stats['terms']} docs={stats['docs']}"
|
||||
)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
if len(sys.argv) != 3:
|
||||
print("usage: python -m arborist.wallet._sidecar_build_runner SRC DST", file=sys.stderr)
|
||||
sys.exit(2)
|
||||
sys.exit(main(sys.argv[1], sys.argv[2]))
|
||||
|
|
@ -690,22 +690,6 @@ class MultiShardBucketCorpus:
|
|||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def _sidecar_cache_path(url: str, *, base_dir: str | None = None) -> str:
|
||||
"""Stable local cache path for a sidecar URL.
|
||||
|
||||
Hashes the URL → places under ~/.arborist/sidecar-cache/<hex>.bin so
|
||||
re-runs reuse the download. base_dir overrides for tests.
|
||||
"""
|
||||
import hashlib
|
||||
if base_dir is None:
|
||||
base_dir = os.path.join(
|
||||
os.path.expanduser("~"), ".arborist", "sidecar-cache"
|
||||
)
|
||||
os.makedirs(base_dir, exist_ok=True)
|
||||
h = hashlib.sha256(url.encode("utf-8")).hexdigest()[:16]
|
||||
return os.path.join(base_dir, f"{h}.bin")
|
||||
|
||||
|
||||
def _download_sidecar(url: str, dest: str, *, timeout_s: float = 300.0) -> int:
|
||||
"""Stream a sidecar URL into `dest`. Returns bytes downloaded.
|
||||
|
||||
|
|
@ -729,78 +713,209 @@ def _download_sidecar(url: str, dest: str, *, timeout_s: float = 300.0) -> int:
|
|||
return n
|
||||
|
||||
|
||||
class SidecarShardClient:
|
||||
"""One shard's view: a local sidecar (download-then-query-locally)
|
||||
plus a BucketClient against the shard .db for chunk content reads.
|
||||
def _fts_sidecar_cache_path(url: str, *, base_dir: str | None = None) -> str:
|
||||
"""Stable local cache path for a slim FTS5 sidecar URL.
|
||||
|
||||
fts_search uses the sidecar (sub-second BM25); chunks_for_doc and
|
||||
SQL go through the bucket VFS so chunk content fetches benefit
|
||||
from the same on-disk page cache."""
|
||||
Separate cache directory from the legacy custom binary sidecar so
|
||||
a manifest swap doesn't pull from the wrong cache. Hash of URL keeps
|
||||
the filename stable; ``.idx.db`` extension makes it recognizable as
|
||||
SQLite.
|
||||
"""
|
||||
import hashlib
|
||||
if base_dir is None:
|
||||
base_dir = os.path.join(
|
||||
os.path.expanduser("~"), ".arborist", "sidecar-fts-cache"
|
||||
)
|
||||
os.makedirs(base_dir, exist_ok=True)
|
||||
h = hashlib.sha256(url.encode("utf-8")).hexdigest()[:16]
|
||||
return os.path.join(base_dir, f"{h}.idx.db")
|
||||
|
||||
|
||||
class FtsSidecarShardClient:
|
||||
"""One shard's view using a slim FTS5 sidecar.
|
||||
|
||||
The slim sidecar (typically ``<shard>.idx.db`` in the bucket) is a
|
||||
standalone SQLite file containing the source shard's contentless
|
||||
FTS5 index plus minimal doc/chunk metadata — no chunk content. We
|
||||
download it once into the local cache and query via plain sqlite3
|
||||
(no VFS, no HTTP-range): FTS5 MATCH + bm25() answer locally at
|
||||
full-speed parity with a local shard.
|
||||
|
||||
Chunk CONTENT is fetched on demand from ``blob_base`` (per-chunk
|
||||
``blobs/<hash[:2]>/<hash[2:]>`` storage), same shape as
|
||||
BucketClient.fetch_chunk_body. The big shard .db is NOT downloaded.
|
||||
|
||||
This is the "real FTS5 in the sidecar" path that replaces the
|
||||
hand-rolled BM25 in SidecarReader. Identical scoring to a local
|
||||
shard by construction (same FTS5 shadow tables, same porter
|
||||
unicode61 tokenizer).
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
shard_url: str,
|
||||
sidecar_url: str,
|
||||
fts_sidecar_url: str,
|
||||
blob_base: str,
|
||||
*,
|
||||
cache_bytes: int = DEFAULT_CACHE_BYTES,
|
||||
shard_url: str | None = None,
|
||||
sidecar_cache_dir: str | None = None,
|
||||
cache_bytes: int = DEFAULT_CACHE_BYTES,
|
||||
timeout_s: float = 30.0,
|
||||
):
|
||||
from arborist.wallet.sidecar import SidecarReader
|
||||
|
||||
self.fts_sidecar_url = fts_sidecar_url
|
||||
self.blob_base = blob_base
|
||||
self.shard_url = shard_url
|
||||
self.sidecar_url = sidecar_url
|
||||
# 1. Download or reuse cached sidecar.
|
||||
cache_path = _sidecar_cache_path(sidecar_url, base_dir=sidecar_cache_dir)
|
||||
if not os.path.exists(cache_path):
|
||||
_download_sidecar(sidecar_url, cache_path)
|
||||
self._sidecar_path = cache_path
|
||||
self.sidecar = SidecarReader(cache_path)
|
||||
# 2. Bucket client for chunk content reads.
|
||||
self.bucket = BucketClient(
|
||||
BucketEndpoint(shard_url=shard_url, blob_base=blob_base),
|
||||
cache_bytes=cache_bytes,
|
||||
cache_path = _fts_sidecar_cache_path(
|
||||
fts_sidecar_url, base_dir=sidecar_cache_dir
|
||||
)
|
||||
if not os.path.exists(cache_path):
|
||||
_download_sidecar(fts_sidecar_url, cache_path)
|
||||
self._sidecar_path = cache_path
|
||||
import sqlite3 as _sql
|
||||
# check_same_thread=False so MultiShardSidecarCorpus can call
|
||||
# fts_search from a ThreadPoolExecutor worker. The sqlite3 lib
|
||||
# itself is built thread-safe (threadsafety=1 or 3 on every
|
||||
# mainstream distro) — the per-connection check is what blocks
|
||||
# cross-thread use by default. Read-only PRAGMA + URI ro flag
|
||||
# keep this safe: no writes, no transactions, no statement
|
||||
# interleaving worth worrying about.
|
||||
self._conn = _sql.connect(
|
||||
f"file:{cache_path}?mode=ro", uri=True,
|
||||
check_same_thread=False,
|
||||
)
|
||||
self._conn.execute("PRAGMA query_only = 1")
|
||||
self._timeout_s = timeout_s
|
||||
# Per-client HTTP stats so MultiShard can aggregate.
|
||||
self._http_requests = 0
|
||||
self._bytes_fetched = 0
|
||||
# Optional fallback for buckets without published per-chunk blobs:
|
||||
# when blob fetch 404s, fall back to HTTP-range on the big shard
|
||||
# .db (BucketClient). Only constructs the BucketClient lazily on
|
||||
# first miss so the common case (blobs published) stays free.
|
||||
self._fallback_bucket: BucketClient | None = None
|
||||
self._fallback_cache_bytes = cache_bytes
|
||||
|
||||
def close(self):
|
||||
self.bucket.close()
|
||||
try:
|
||||
self._conn.close()
|
||||
except Exception:
|
||||
pass
|
||||
|
||||
def fts_search(self, query: str, *, limit: int = 8, raw: bool = False) -> list[dict]:
|
||||
"""Sidecar-driven FTS. ``raw`` is accepted for API parity with
|
||||
BucketClient.fts_search but ignored — sidecar tokenizes via
|
||||
tokenize_text (matches the OR-sanitizer behavior)."""
|
||||
hits = self.sidecar.search(query, limit=limit, mode="or")
|
||||
def fts_search(
|
||||
self, query: str, *, limit: int = 8, raw: bool = False,
|
||||
) -> list[dict]:
|
||||
"""Local FTS5 MATCH against the slim sidecar.
|
||||
|
||||
Mirrors BucketClient.fts_search shape exactly — same SQL, same
|
||||
bm25 ordering — so the merge layer in MultiShardSidecarCorpus
|
||||
treats slim-sidecar shards as homogeneous with bucket-direct
|
||||
FTS5 shards (both return negative bm25 scores, lower = better).
|
||||
"""
|
||||
fts_query = query if raw else _to_fts5(query)
|
||||
if not fts_query.strip():
|
||||
return []
|
||||
sql = (
|
||||
"SELECT d.document_root, d.document_uri, d.title, "
|
||||
" bm25(chunks_fts) AS score "
|
||||
"FROM chunks_fts JOIN chunks c ON c.rowid = chunks_fts.rowid "
|
||||
"JOIN documents d ON d.document_root = c.document_root "
|
||||
"WHERE chunks_fts MATCH ? "
|
||||
"ORDER BY score LIMIT ?"
|
||||
)
|
||||
rows = list(self._conn.execute(sql, (fts_query, limit)))
|
||||
return [
|
||||
{
|
||||
"document_root": h.document_root,
|
||||
"document_uri": h.document_uri,
|
||||
"title": h.title,
|
||||
"score": h.score,
|
||||
}
|
||||
for h in hits
|
||||
{"document_root": r[0], "document_uri": r[1],
|
||||
"title": r[2], "score": r[3]}
|
||||
for r in rows
|
||||
]
|
||||
|
||||
@property
|
||||
def conn(self):
|
||||
"""apsw connection for SQL pulls (chunk content). Goes through
|
||||
the bucket VFS — chunk reads are cached."""
|
||||
return self.bucket.conn
|
||||
def chunks_for_doc(self, document_root: str) -> list[dict]:
|
||||
"""Return chunk metadata (idx, leaf_hash) for a document.
|
||||
|
||||
The slim sidecar carries this without needing the big shard.
|
||||
"""
|
||||
sql = (
|
||||
"SELECT idx, leaf_hash FROM chunks "
|
||||
"WHERE document_root = ? ORDER BY idx ASC"
|
||||
)
|
||||
rows = list(self._conn.execute(sql, (document_root,)))
|
||||
return [{"idx": r[0], "leaf_hash": r[1]} for r in rows]
|
||||
|
||||
def fetch_chunk_body(self, leaf_hash: str) -> bytes:
|
||||
"""GET blobs/<hash[:2]>/<hash[2:]> from the bucket.
|
||||
|
||||
Per-chunk content lives in ``blob_base`` (the JUST_ENOUGH=1
|
||||
bucket layout). On 403/404 (blob missing) AND when ``shard_url``
|
||||
is set we fall back to an HTTP-range read of chunks.content from
|
||||
the big shard .db — keeps the slim FTS5 client useful in
|
||||
transitional buckets that haven't published per-chunk blobs yet.
|
||||
Hash-verification belongs to the caller.
|
||||
"""
|
||||
url = (
|
||||
f"{self.blob_base.rstrip('/')}/"
|
||||
f"{leaf_hash[:2]}/{leaf_hash[2:]}"
|
||||
)
|
||||
req = urllib.request.Request(url)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=self._timeout_s) as resp:
|
||||
data = resp.read()
|
||||
self._http_requests += 1
|
||||
self._bytes_fetched += len(data)
|
||||
return data
|
||||
except urllib.error.HTTPError as e:
|
||||
if e.code not in (403, 404) or not self.shard_url:
|
||||
raise
|
||||
# Fallback path — open BucketClient lazily, read chunks.content
|
||||
# by leaf_hash. This is slower (page-level HTTP range reads on
|
||||
# the big shard) but works without per-chunk blob storage.
|
||||
if self._fallback_bucket is None:
|
||||
self._fallback_bucket = BucketClient(
|
||||
BucketEndpoint(
|
||||
shard_url=self.shard_url, blob_base=self.blob_base,
|
||||
),
|
||||
cache_bytes=self._fallback_cache_bytes,
|
||||
)
|
||||
rows = list(self._fallback_bucket.conn.execute(
|
||||
"SELECT content FROM chunks WHERE leaf_hash = ? "
|
||||
"AND content IS NOT NULL LIMIT 1",
|
||||
(leaf_hash,),
|
||||
))
|
||||
if not rows:
|
||||
raise FileNotFoundError(
|
||||
f"leaf_hash {leaf_hash[:12]}… not found via blobs/* OR "
|
||||
f"big-shard chunks.content fallback"
|
||||
)
|
||||
self._http_requests += 1
|
||||
data = rows[0][0]
|
||||
self._bytes_fetched += len(data) if data else 0
|
||||
return data
|
||||
|
||||
def stats(self) -> dict:
|
||||
return {
|
||||
"http_requests": self._http_requests,
|
||||
"cache": {"bytes_fetched": self._bytes_fetched},
|
||||
"sidecar_path": self._sidecar_path,
|
||||
"sidecar_bytes": (
|
||||
os.path.getsize(self._sidecar_path)
|
||||
if os.path.exists(self._sidecar_path) else 0
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
|
||||
|
||||
class MultiShardSidecarCorpus:
|
||||
"""Manifest-driven corpus where each shard MAY have a sidecar.
|
||||
"""Manifest-driven corpus. Per shard, two shapes:
|
||||
- ``fts_sidecar_url`` → FtsSidecarShardClient (download the slim
|
||||
FTS5 sidecar locally + bucket-served chunk content). Real FTS5
|
||||
bm25() with porter+unicode61 tokenizer; bit-for-bit parity with
|
||||
a local shard by construction.
|
||||
- no sidecar → BucketClient (HttpRangeVFS on the big shard .db).
|
||||
Slow on multi-GB shards but works without prepublished sidecars.
|
||||
|
||||
Shards with a `sidecar_url` use SidecarShardClient (download-then-
|
||||
local FTS). Shards without one fall back to BucketClient (in-place
|
||||
bucket-direct FTS5 — slow on multi-GB shards, but works for small
|
||||
shards or when no sidecar is published yet).
|
||||
|
||||
fts_search runs per-shard in parallel, merges by score, picks top-K.
|
||||
Mixed-mode results are score-comparable in practice: sidecar BM25
|
||||
is on a similar magnitude (~10-30) as FTS5's BM25 (~-15 to -1) but
|
||||
sign-flipped (sidecar is positive, FTS5 stored as negative). We
|
||||
normalize sidecar to negative before merging so a single ORDER BY
|
||||
DESC works across both.
|
||||
fts_search runs per-shard in parallel and merges raw bm25 scores
|
||||
when every contributing shard uses the same backend. Mixed
|
||||
(slim-FTS5 + bucket-direct) is score-comparable in practice — both
|
||||
return negative FTS5 bm25 — so the merge uses MIN across either.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
|
|
@ -808,29 +923,28 @@ class MultiShardSidecarCorpus:
|
|||
manifest: "BucketManifest",
|
||||
*,
|
||||
cache_bytes_per_shard: int = DEFAULT_CACHE_BYTES,
|
||||
sidecar_cache_dir: str | None = None,
|
||||
fts_sidecar_cache_dir: str | None = None,
|
||||
):
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
|
||||
self.manifest = manifest
|
||||
self._has_sidecar: list[bool] = [
|
||||
bool(sh.get("sidecar_url")) for sh in manifest.shards
|
||||
bool(sh.get("fts_sidecar_url")) for sh in manifest.shards
|
||||
]
|
||||
|
||||
# Pre-download every sidecar in parallel so a fresh consumer
|
||||
# pays max(per-sidecar download) instead of sum. SidecarShardClient
|
||||
# is idempotent — if the cache file already exists, it just reads
|
||||
# from disk (no re-download). Bucket-only shards (no sidecar)
|
||||
# construct fast; they don't pre-pull the .db.
|
||||
# pays max(per-sidecar download) instead of sum. Idempotent: if
|
||||
# the cache file already exists, the client just reads from disk.
|
||||
# Bucket-only shards (no sidecar) construct fast.
|
||||
def _build(item):
|
||||
i, sh = item
|
||||
if sh.get("sidecar_url"):
|
||||
return i, SidecarShardClient(
|
||||
shard_url=sh["url"],
|
||||
sidecar_url=sh["sidecar_url"],
|
||||
if sh.get("fts_sidecar_url"):
|
||||
return i, FtsSidecarShardClient(
|
||||
fts_sidecar_url=sh["fts_sidecar_url"],
|
||||
blob_base=manifest.blob_base,
|
||||
shard_url=sh.get("url"), # content fallback
|
||||
sidecar_cache_dir=fts_sidecar_cache_dir,
|
||||
cache_bytes=cache_bytes_per_shard,
|
||||
sidecar_cache_dir=sidecar_cache_dir,
|
||||
)
|
||||
return i, BucketClient(
|
||||
BucketEndpoint(shard_url=sh["url"], blob_base=manifest.blob_base),
|
||||
|
|
@ -851,115 +965,48 @@ class MultiShardSidecarCorpus:
|
|||
|
||||
def fts_search(
|
||||
self, query: str, *, limit: int = 8, raw: bool = False,
|
||||
rrf_k: int = 60,
|
||||
) -> list[dict]:
|
||||
"""Run each shard's FTS in parallel, merge via reciprocal-rank-fusion.
|
||||
"""Run each shard's FTS in parallel, merge by raw bm25 score.
|
||||
|
||||
BM25 scores aren't comparable across heterogeneous backends
|
||||
(sidecar BM25 is positive in the [5, 50] range; FTS5 BM25 is
|
||||
negative in [-20, 0]). RRF sidesteps the scale mismatch
|
||||
entirely: each shard ranks its own hits, the global rank is
|
||||
the sum of reciprocal ranks across shards. Standard k=60 is
|
||||
TREC-recommended.
|
||||
|
||||
Post-merge title-relevance filter: drop hits whose titles
|
||||
share zero query tokens with the query. Mirrors local
|
||||
query.py's ``search.title_filter`` step — without it,
|
||||
small shards (e.g. a 200-doc personal blog) hand back rank-1
|
||||
hits for any matched token, ties with rank-1 from a 1M-doc
|
||||
shard under RRF, and ranks above topically correct docs. With
|
||||
the filter, off-topic docs are dropped before the merge.
|
||||
Every backend now returns FTS5's negative bm25 (lower = better),
|
||||
so merge takes the MIN per document_root and sorts ascending —
|
||||
homogeneous scoring, no RRF needed.
|
||||
"""
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from arborist.wallet.sidecar import fold_accents, tokenize_text
|
||||
|
||||
def _one(item):
|
||||
sh_url, sh_client, _ = item
|
||||
sh_url, sh_client = item
|
||||
try:
|
||||
# Pull per-shard top-K (oversample so RRF has more to
|
||||
# work with — a doc that's barely in top-K on one
|
||||
# shard but #1 on another should still bubble up).
|
||||
# Oversample 4× per shard so post-merge ranking has more
|
||||
# candidates — a doc barely in top-K on one shard but #1
|
||||
# on another should still bubble up.
|
||||
return sh_url, sh_client.fts_search(query, limit=limit * 4, raw=raw)
|
||||
except Exception:
|
||||
return sh_url, []
|
||||
|
||||
triples = [
|
||||
(sh["url"], self._by_url[sh["url"]], self._has_sidecar[i])
|
||||
for i, sh in enumerate(self.manifest.shards)
|
||||
pairs = [
|
||||
(sh["url"], self._by_url[sh["url"]])
|
||||
for sh in self.manifest.shards
|
||||
]
|
||||
per_shard: list[tuple[str, list[dict]]] = []
|
||||
with ThreadPoolExecutor(max_workers=max(1, len(triples))) as ex:
|
||||
for sh_url, hits in ex.map(_one, triples):
|
||||
with ThreadPoolExecutor(max_workers=max(1, len(pairs))) as ex:
|
||||
for sh_url, hits in ex.map(_one, pairs):
|
||||
per_shard.append((sh_url, hits))
|
||||
|
||||
# Title-relevance filter: drop hits whose title shares zero
|
||||
# query tokens. tokenize_text already folds accents + drops
|
||||
# stopwords, so "Pokémon Red and Blue" → {pokemon, red, blue}
|
||||
# overlaps query {starter, pokemon, red}; Russell Ballestrini
|
||||
# blog root → {russell, ballestrini} overlaps NEITHER → dropped.
|
||||
# Falls open (no filter) if the query has no content tokens
|
||||
# after sanitization.
|
||||
query_tokens = set(tokenize_text(query))
|
||||
if query_tokens:
|
||||
def _title_relevant(h: dict) -> bool:
|
||||
title = h.get("title") or ""
|
||||
if not title:
|
||||
return False
|
||||
title_tokens = set(tokenize_text(title))
|
||||
# Fall-open when the title has no content tokens after
|
||||
# stopword/length filtering — single-char or all-stopword
|
||||
# titles ("A", "I", "Of Mice and Men") shouldn't be
|
||||
# filtered out just because the FILTER side has nothing
|
||||
# to match on. Keep the hit; let BM25 + title-boost rank.
|
||||
if not title_tokens:
|
||||
return True
|
||||
return bool(query_tokens & title_tokens)
|
||||
per_shard = [
|
||||
(sh_url, [h for h in hits if _title_relevant(h)])
|
||||
for sh_url, hits in per_shard
|
||||
]
|
||||
|
||||
# Merge strategy: when every shard CONTRIBUTING HITS is a
|
||||
# sidecar, BM25 scores are directly comparable (same constants,
|
||||
# same title-boost formula), so merge by raw score — preserves
|
||||
# per-shard rank discrimination instead of squashing the top-N
|
||||
# into a tie-at-1/(k+1) under RRF. When the surviving hits mix
|
||||
# sidecar + bucket-direct FTS5 (incomparable scales: BM25
|
||||
# +25..+60 vs FTS5 −20..0), fall back to RRF.
|
||||
url_to_sidecar = {
|
||||
sh["url"]: self._has_sidecar[i]
|
||||
for i, sh in enumerate(self.manifest.shards)
|
||||
}
|
||||
contributing_shards = {
|
||||
sh_url for sh_url, hits in per_shard if hits
|
||||
}
|
||||
use_raw_score = bool(contributing_shards) and all(
|
||||
url_to_sidecar[u] for u in contributing_shards
|
||||
)
|
||||
|
||||
hit_by_root: dict[str, dict] = {}
|
||||
score_by_root: dict[str, float] = {}
|
||||
for sh_url, hits in per_shard:
|
||||
for rank, h in enumerate(hits, start=1):
|
||||
for h in hits:
|
||||
droot = h["document_root"]
|
||||
if use_raw_score:
|
||||
# Take the MAX raw score across shards.
|
||||
raw = h.get("score") or 0.0
|
||||
if droot not in score_by_root or raw > score_by_root[droot]:
|
||||
score_by_root[droot] = raw
|
||||
else:
|
||||
score_by_root[droot] = (
|
||||
score_by_root.get(droot, 0.0) + 1.0 / (rrf_k + rank)
|
||||
)
|
||||
raw_score = h.get("score") or 0.0
|
||||
if droot not in score_by_root or raw_score < score_by_root[droot]:
|
||||
score_by_root[droot] = raw_score
|
||||
if droot not in hit_by_root:
|
||||
h["_shard_url"] = sh_url
|
||||
hit_by_root[droot] = h
|
||||
for droot, h in hit_by_root.items():
|
||||
h["merge_score"] = score_by_root[droot]
|
||||
ranked = sorted(
|
||||
hit_by_root.values(),
|
||||
key=lambda h: -h["merge_score"],
|
||||
)
|
||||
ranked = sorted(hit_by_root.values(), key=lambda h: h["merge_score"])
|
||||
return ranked[:limit]
|
||||
|
||||
def conn_for_shard(self, shard_url: str):
|
||||
|
|
|
|||
209
arborist/wallet/fts_sidecar_build.py
Normal file
209
arborist/wallet/fts_sidecar_build.py
Normal file
|
|
@ -0,0 +1,209 @@
|
|||
"""Build a slim FTS5 sidecar from an arborist shard.
|
||||
|
||||
A slim sidecar contains:
|
||||
- documents (document_root, document_uri, title) — for title-boost + display
|
||||
- chunks (chunk_id, document_root, idx, leaf_hash) — for evidence map
|
||||
- chunks_fts + documents_fts (contentless FTS5; copied verbatim from source
|
||||
shadow tables so the index identity is preserved bit-for-bit)
|
||||
|
||||
It does NOT contain chunk content — that lives in the big shard and is
|
||||
fetched over HTTP-range via the existing chunk-fetch path when evidence
|
||||
is needed.
|
||||
|
||||
Idea behind this layout: cloud retrieval scoring must equal local
|
||||
retrieval scoring. The source shard already uses contentless FTS5 with
|
||||
'porter unicode61' tokenizer; copying its shadow tables means cloud
|
||||
FTS5.bm25() == local FTS5.bm25() by construction. No custom BM25
|
||||
implementation, no hand-rolled tokenizer, no granularity skew.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
import sqlite3
|
||||
import sys
|
||||
import time
|
||||
|
||||
|
||||
SHADOW_TABLES = {
|
||||
"chunks_fts": [
|
||||
"chunks_fts_data",
|
||||
"chunks_fts_idx",
|
||||
"chunks_fts_docsize",
|
||||
"chunks_fts_config",
|
||||
],
|
||||
"documents_fts": [
|
||||
"documents_fts_data",
|
||||
"documents_fts_idx",
|
||||
"documents_fts_docsize",
|
||||
"documents_fts_config",
|
||||
],
|
||||
}
|
||||
|
||||
# DDL must EXACTLY match what the source shard uses or the shadow-table
|
||||
# bulk-copy will land in differently-organized internal storage and
|
||||
# MATCH lookups will fail.
|
||||
DDL_CHUNKS_FTS = (
|
||||
"CREATE VIRTUAL TABLE chunks_fts USING fts5("
|
||||
"content, content='', contentless_delete=1, "
|
||||
"tokenize='porter unicode61')"
|
||||
)
|
||||
DDL_DOCUMENTS_FTS = (
|
||||
"CREATE VIRTUAL TABLE documents_fts USING fts5("
|
||||
"title, content='', contentless_delete=1, "
|
||||
"tokenize='porter unicode61')"
|
||||
)
|
||||
|
||||
|
||||
def _copy_shadow(src: sqlite3.Connection, dst: sqlite3.Connection,
|
||||
table: str) -> int:
|
||||
"""Bulk-copy one FTS5 shadow table src→dst.
|
||||
|
||||
The destination table was just created (empty) by CREATE VIRTUAL
|
||||
TABLE; we delete it and re-insert the source bytes verbatim so the
|
||||
bm25() scorer sees the same posting-list layout.
|
||||
"""
|
||||
dst.execute(f"DELETE FROM {table}")
|
||||
cols_info = dst.execute(f"PRAGMA table_info({table})").fetchall()
|
||||
cols = [r[1] for r in cols_info]
|
||||
placeholders = ", ".join("?" * len(cols))
|
||||
col_list = ", ".join(cols)
|
||||
n = 0
|
||||
cur = src.execute(f"SELECT {col_list} FROM {table}")
|
||||
batch: list = []
|
||||
for row in cur:
|
||||
batch.append(row)
|
||||
if len(batch) >= 1000:
|
||||
dst.executemany(
|
||||
f"INSERT INTO {table} ({col_list}) VALUES ({placeholders})",
|
||||
batch,
|
||||
)
|
||||
n += len(batch)
|
||||
batch.clear()
|
||||
if batch:
|
||||
dst.executemany(
|
||||
f"INSERT INTO {table} ({col_list}) VALUES ({placeholders})",
|
||||
batch,
|
||||
)
|
||||
n += len(batch)
|
||||
return n
|
||||
|
||||
|
||||
def build(source_path: str, dest_path: str, *, verbose: bool = True) -> dict:
|
||||
"""Build a slim FTS5 sidecar at ``dest_path`` from ``source_path``.
|
||||
|
||||
Returns a stats dict (sizes, rowcounts, elapsed).
|
||||
"""
|
||||
if os.path.exists(dest_path):
|
||||
raise FileExistsError(
|
||||
f"refusing to overwrite {dest_path} — delete first if you mean it"
|
||||
)
|
||||
t0 = time.time()
|
||||
src_size = os.path.getsize(source_path)
|
||||
|
||||
src = sqlite3.connect(f"file:{source_path}?mode=ro", uri=True)
|
||||
dst = sqlite3.connect(dest_path)
|
||||
try:
|
||||
dst.execute("PRAGMA journal_mode=OFF")
|
||||
dst.execute("PRAGMA synchronous=OFF")
|
||||
dst.execute("PRAGMA cache_size=-2000000") # 2 GB cache during build
|
||||
|
||||
# 1. Slim documents.
|
||||
dst.execute(
|
||||
"CREATE TABLE documents ("
|
||||
"document_root TEXT PRIMARY KEY, "
|
||||
"document_uri TEXT, "
|
||||
"title TEXT)"
|
||||
)
|
||||
t = time.time()
|
||||
dst.executemany(
|
||||
"INSERT INTO documents VALUES (?, ?, ?)",
|
||||
src.execute(
|
||||
"SELECT document_root, document_uri, title FROM documents"
|
||||
),
|
||||
)
|
||||
dst.commit()
|
||||
n_docs = dst.execute("SELECT COUNT(*) FROM documents").fetchone()[0]
|
||||
if verbose:
|
||||
print(f" documents: {n_docs:,} rows in {time.time()-t:.1f}s",
|
||||
file=sys.stderr)
|
||||
|
||||
# 2. Slim chunks (no content).
|
||||
dst.execute(
|
||||
"CREATE TABLE chunks ("
|
||||
"chunk_id INTEGER PRIMARY KEY, "
|
||||
"document_root TEXT NOT NULL, "
|
||||
"idx INTEGER NOT NULL, "
|
||||
"leaf_hash TEXT NOT NULL)"
|
||||
)
|
||||
t = time.time()
|
||||
dst.executemany(
|
||||
"INSERT INTO chunks VALUES (?, ?, ?, ?)",
|
||||
src.execute(
|
||||
"SELECT chunk_id, document_root, idx, leaf_hash FROM chunks"
|
||||
),
|
||||
)
|
||||
dst.execute(
|
||||
"CREATE INDEX idx_chunks_document_root ON chunks(document_root)"
|
||||
)
|
||||
dst.commit()
|
||||
n_chunks = dst.execute("SELECT COUNT(*) FROM chunks").fetchone()[0]
|
||||
if verbose:
|
||||
print(f" chunks: {n_chunks:,} rows in {time.time()-t:.1f}s",
|
||||
file=sys.stderr)
|
||||
|
||||
# 3. FTS5 virtual tables — copy shadow tables verbatim.
|
||||
for vt_name, ddl in (("chunks_fts", DDL_CHUNKS_FTS),
|
||||
("documents_fts", DDL_DOCUMENTS_FTS)):
|
||||
t = time.time()
|
||||
dst.execute(ddl)
|
||||
for shadow in SHADOW_TABLES[vt_name]:
|
||||
n = _copy_shadow(src, dst, shadow)
|
||||
if verbose:
|
||||
print(f" {shadow}: {n:,} rows",
|
||||
file=sys.stderr)
|
||||
dst.commit()
|
||||
if verbose:
|
||||
print(f" {vt_name}: built in {time.time()-t:.1f}s",
|
||||
file=sys.stderr)
|
||||
|
||||
# 4. Compact.
|
||||
if verbose:
|
||||
print(" VACUUM ...", file=sys.stderr)
|
||||
dst.execute("VACUUM")
|
||||
dst.commit()
|
||||
finally:
|
||||
src.close()
|
||||
dst.close()
|
||||
|
||||
dest_size = os.path.getsize(dest_path)
|
||||
elapsed = time.time() - t0
|
||||
return {
|
||||
"source_path": source_path,
|
||||
"dest_path": dest_path,
|
||||
"source_bytes": src_size,
|
||||
"dest_bytes": dest_size,
|
||||
"ratio": dest_size / src_size if src_size else 0.0,
|
||||
"n_documents": n_docs,
|
||||
"n_chunks": n_chunks,
|
||||
"elapsed_s": round(elapsed, 2),
|
||||
}
|
||||
|
||||
|
||||
def main():
|
||||
import argparse
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument("source", help="path to source shard .db")
|
||||
p.add_argument("dest", help="path to write slim FTS5 sidecar")
|
||||
args = p.parse_args()
|
||||
stats = build(args.source, args.dest)
|
||||
print()
|
||||
print(f"source: {stats['source_bytes']/1e9:.2f} GB → "
|
||||
f"sidecar: {stats['dest_bytes']/1e9:.2f} GB "
|
||||
f"({stats['ratio']*100:.1f}%)")
|
||||
print(f"docs: {stats['n_documents']:,}")
|
||||
print(f"chunks: {stats['n_chunks']:,}")
|
||||
print(f"time: {stats['elapsed_s']:.0f}s")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -1,685 +0,0 @@
|
|||
"""HTTP-optimized inverted-index sidecar for bucket-direct queries.
|
||||
|
||||
The bucket-direct cloud-query path runs FTS5 against a SQLite .db file
|
||||
served over HTTP RANGE. That's correct but slow on multi-GB shards:
|
||||
FTS5's b-tree walk touches hundreds of random pages per query, each
|
||||
costing a WAN round-trip. We measured 4+ minutes per query on a 9 GB
|
||||
genesis shard.
|
||||
|
||||
This sidecar trades query latency for a one-time-per-corpus download:
|
||||
|
||||
Producer (server) Consumer (wallet client)
|
||||
───────────────── ────────────────────────
|
||||
read shard.db GET sidecar.bin (one HTTP request)
|
||||
tokenize chunk text load bytes into memory
|
||||
build inverted index binary-search dictionary
|
||||
write sidecar.bin decode posting list (sub-ms)
|
||||
fetch chunks from blobs/<hash>
|
||||
|
||||
Format (binary, little-endian):
|
||||
|
||||
+───────────────────────────────────────────────────+
|
||||
| MAGIC "ARBORIST-SIDECAR-V1\0" | 20 B
|
||||
| dict_count uint64 | 8 B
|
||||
| docs_count uint64 | 8 B
|
||||
| dict_offset uint64 | 8 B
|
||||
| postings_offset uint64 | 8 B
|
||||
| docs_offset uint64 | 8 B
|
||||
+───────────────────────────────────────────────────+ <- 60 B header
|
||||
| Dictionary (sorted by term) |
|
||||
| per entry: |
|
||||
| u16 term_len |
|
||||
| ... term bytes (utf-8) |
|
||||
| u32 doc_freq |
|
||||
| u64 posting_offset (absolute) |
|
||||
| u32 posting_len |
|
||||
+───────────────────────────────────────────────────+
|
||||
| Posting lists (concatenated) |
|
||||
| per list: |
|
||||
| varint num_docs |
|
||||
| varint deltas (sorted-ascending doc_ids) |
|
||||
+───────────────────────────────────────────────────+
|
||||
| Doc table (doc_id-indexed) |
|
||||
| per entry: |
|
||||
| 32 B document_root (hex-decoded) |
|
||||
| u16 uri_len |
|
||||
| ... uri bytes (utf-8) |
|
||||
| u16 title_len |
|
||||
| ... title bytes (utf-8) |
|
||||
+───────────────────────────────────────────────────+
|
||||
| Doc offset table (doc_id → byte offset in docs) |
|
||||
| docs_count u64 entries |
|
||||
+───────────────────────────────────────────────────+
|
||||
|
||||
Conventions:
|
||||
- term comparison: case-folded UTF-8 byte order (matches tokenize_text)
|
||||
- doc_ids are dense [0, docs_count) per sidecar
|
||||
- varint = LEB128 unsigned (per-byte 0..127; high bit = more bytes)
|
||||
- multi-term AND via posting-list intersection (default cloud-query
|
||||
semantics is OR; sidecar supports both via SidecarReader API)
|
||||
|
||||
This is NOT a Merkle-bound artifact. It's a soft index — the doc_root
|
||||
hashes inside ARE Merkle-bound (chunk content → document_root chain
|
||||
verifiable via the existing wallet path). A malicious sidecar can DOS
|
||||
(refuse to surface a hit) but cannot forge content because the wallet
|
||||
still verifies the chunk body's leaf_hash on retrieval.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import bisect
|
||||
import io
|
||||
import os
|
||||
import re
|
||||
import sqlite3
|
||||
import struct
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import Iterable, Sequence
|
||||
|
||||
import unicodedata
|
||||
|
||||
from arborist.compress import unpack_chunk
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tokenization. Matches the bucket-direct sanitizer (arborist.wallet.bucket
|
||||
# ._to_fts5) so that sidecar-build and sidecar-query agree on what's a term.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
_WORD_RE = re.compile(r"[A-Za-z0-9_]+")
|
||||
|
||||
# Stopword set mirrors arborist.wallet.bucket._FTS5_STOPWORDS exactly;
|
||||
# importing avoids drift across the two query paths.
|
||||
from arborist.wallet.bucket import _FTS5_STOPWORDS as STOPWORDS
|
||||
|
||||
|
||||
def fold_accents(text: str) -> str:
|
||||
"""NFKD-normalize + strip combining marks → ASCII-equivalent.
|
||||
|
||||
Critical for retrieval: a user typing 'pokemon' should match
|
||||
Wikipedia titles like 'Pokémon Red and Blue', and our ASCII-only
|
||||
word regex was splitting 'Pokémon' into ['Pok', 'mon'] (because é
|
||||
is not in [A-Za-z]), so the term 'pokemon' never landed in the
|
||||
dictionary. NFKD folds 'é'→'e' before tokenization fires.
|
||||
"""
|
||||
return "".join(
|
||||
c for c in unicodedata.normalize("NFKD", text)
|
||||
if not unicodedata.combining(c)
|
||||
)
|
||||
|
||||
|
||||
# Roman numeral <-> Arabic digit equivalence table for 1..20.
|
||||
# A user typing 'final fantasy 7' should overlap a title 'Final
|
||||
# Fantasy VII'. This is the same gap local query.py closes with its
|
||||
# numeral-fold step. We only need 1..20 — beyond that, titles use
|
||||
# Arabic digits in practice.
|
||||
_NUMERAL_PAIRS: dict[str, str] = {
|
||||
"1": "i", "i": "1",
|
||||
"2": "ii", "ii": "2",
|
||||
"3": "iii", "iii": "3",
|
||||
"4": "iv", "iv": "4",
|
||||
"5": "v", "v": "5",
|
||||
"6": "vi", "vi": "6",
|
||||
"7": "vii", "vii": "7",
|
||||
"8": "viii","viii": "8",
|
||||
"9": "ix", "ix": "9",
|
||||
"10": "x", "x": "10",
|
||||
"11": "xi", "xi": "11",
|
||||
"12": "xii","xii": "12",
|
||||
"13": "xiii","xiii": "13",
|
||||
"14": "xiv","xiv": "14",
|
||||
"15": "xv", "xv": "15",
|
||||
"16": "xvi","xvi": "16",
|
||||
"17": "xvii","xvii": "17",
|
||||
"18": "xviii","xviii": "18",
|
||||
"19": "xix","xix": "19",
|
||||
"20": "xx", "xx": "20",
|
||||
}
|
||||
|
||||
|
||||
def numeral_expand(tokens) -> set:
|
||||
"""Return a set with each token's numeral-equivalent form added.
|
||||
|
||||
'final fantasy 7' tokens → {final, fantasy, 7, vii}
|
||||
'Final Fantasy VII' tokens → {final, fantasy, vii, 7}
|
||||
Overlap of the two sets is 4 (or 3 if the duplicate folds), vs 2
|
||||
without the fold — enough to lift the right title above siblings
|
||||
that share only 'final' + 'fantasy'.
|
||||
"""
|
||||
out = set(tokens)
|
||||
for t in tokens:
|
||||
alt = _NUMERAL_PAIRS.get(t)
|
||||
if alt is not None:
|
||||
out.add(alt)
|
||||
return out
|
||||
|
||||
|
||||
def tokenize_text(text: str) -> list[str]:
|
||||
"""Lowercase word tokens, drop stopwords + 1-char tokens (except
|
||||
numeric singletons like '7' which carry meaning — 'Final Fantasy 7'
|
||||
needs '7' to survive sanitization so the numeral-fold can later
|
||||
pair it with 'VII'). Accent-fold first so 'pokémon' indexes/queries
|
||||
as 'pokemon'."""
|
||||
folded = fold_accents(text)
|
||||
out: list[str] = []
|
||||
for raw in _WORD_RE.findall(folded):
|
||||
t = raw.lower()
|
||||
if t in STOPWORDS:
|
||||
continue
|
||||
if len(t) <= 1 and not t.isdigit():
|
||||
continue
|
||||
out.append(t)
|
||||
return out
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Binary primitives.
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
MAGIC = b"ARBORIST-SIDECAR-V2\0"
|
||||
HEADER_FMT = "<20sQQQQQ"
|
||||
HEADER_LEN = struct.calcsize(HEADER_FMT)
|
||||
|
||||
|
||||
def _write_varint(buf: io.BytesIO, n: int) -> None:
|
||||
"""LEB128 unsigned varint."""
|
||||
if n < 0:
|
||||
raise ValueError(f"varint expects unsigned, got {n}")
|
||||
while n >= 0x80:
|
||||
buf.write(bytes([(n & 0x7F) | 0x80]))
|
||||
n >>= 7
|
||||
buf.write(bytes([n]))
|
||||
|
||||
|
||||
def _read_varint(data: bytes, off: int) -> tuple[int, int]:
|
||||
"""Read LEB128 from ``data`` at ``off``. Returns (value, new_offset)."""
|
||||
n = 0
|
||||
shift = 0
|
||||
while True:
|
||||
b = data[off]
|
||||
off += 1
|
||||
n |= (b & 0x7F) << shift
|
||||
if not (b & 0x80):
|
||||
return n, off
|
||||
shift += 7
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Producer: build_sidecar(shard_db, out_path).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def build_sidecar(
|
||||
shard_db: str | Path,
|
||||
out_path: str | Path,
|
||||
*,
|
||||
progress_every: int = 50_000,
|
||||
) -> dict:
|
||||
"""Read a shard's documents + chunks, tokenize, write sidecar.bin.
|
||||
|
||||
Returns a stats dict: {docs, terms, postings_bytes, file_bytes}.
|
||||
"""
|
||||
shard_db = str(shard_db)
|
||||
out_path = Path(out_path)
|
||||
|
||||
# Phase 1: scan shard, assign dense doc_ids, build term → set(doc_id) map.
|
||||
print(f"sidecar: scanning {shard_db}")
|
||||
src = sqlite3.connect(f"file:{shard_db}?mode=ro", uri=True)
|
||||
src.row_factory = sqlite3.Row
|
||||
try:
|
||||
# Per-shard doc table.
|
||||
doc_rows = list(src.execute(
|
||||
"SELECT document_root, document_uri, title FROM documents "
|
||||
"ORDER BY document_root ASC"
|
||||
))
|
||||
docs_count = len(doc_rows)
|
||||
print(f" docs: {docs_count:,}")
|
||||
|
||||
root_to_id: dict[str, int] = {
|
||||
r["document_root"]: i for i, r in enumerate(doc_rows)
|
||||
}
|
||||
|
||||
# Build inverted index with term-frequencies.
|
||||
# index[term] = list of (doc_id, tf) — appended in doc order
|
||||
# doc_lens[doc_id] = total term count for the doc (BM25 norm)
|
||||
# tf is capped at 255 (1-byte varint will use 2 bytes for >127
|
||||
# but most chunks-per-doc have tf < 100; ceiling protects against
|
||||
# adversarial repeating-token docs without changing the math).
|
||||
index: dict[str, list[tuple[int, int]]] = {}
|
||||
doc_lens: list[int] = [0] * docs_count
|
||||
seen = 0
|
||||
cursor = src.execute(
|
||||
"SELECT document_root, content FROM chunks "
|
||||
"WHERE content IS NOT NULL"
|
||||
)
|
||||
for row in cursor:
|
||||
doc_id = root_to_id.get(row["document_root"])
|
||||
if doc_id is None:
|
||||
continue
|
||||
text = unpack_chunk(row["content"])
|
||||
if not text:
|
||||
continue
|
||||
tf_local: dict[str, int] = {}
|
||||
for tok in tokenize_text(text):
|
||||
tf_local[tok] = tf_local.get(tok, 0) + 1
|
||||
doc_lens[doc_id] += 1
|
||||
for term, tf in tf_local.items():
|
||||
lst = index.get(term)
|
||||
# Cap at 65535 (u16) — easily fits in varint; well above
|
||||
# natural language tf distribution.
|
||||
tf_capped = min(tf, 65535)
|
||||
if lst is None:
|
||||
index[term] = [(doc_id, tf_capped)]
|
||||
elif lst[-1][0] != doc_id:
|
||||
lst.append((doc_id, tf_capped))
|
||||
else:
|
||||
# Same doc, second chunk — accumulate tf.
|
||||
prev_did, prev_tf = lst[-1]
|
||||
lst[-1] = (prev_did, min(prev_tf + tf, 65535))
|
||||
seen += 1
|
||||
if seen % progress_every == 0:
|
||||
print(
|
||||
f" scanned {seen:,} chunks "
|
||||
f"distinct terms: {len(index):,}"
|
||||
)
|
||||
finally:
|
||||
src.close()
|
||||
print(f" total terms: {len(index):,} (one entry per word)")
|
||||
|
||||
# Phase 2: serialize. Two passes — first compute offsets, second write.
|
||||
# Posting lists carry monotonically-increasing doc_ids → delta-encode.
|
||||
# Sort posting-list doc_ids so deltas stay positive.
|
||||
sorted_terms = sorted(index.keys())
|
||||
|
||||
# Build postings section in memory; track per-term (offset, len).
|
||||
# v2 format: per-doc varint pair (delta, tf).
|
||||
postings_buf = io.BytesIO()
|
||||
posting_meta: list[tuple[str, int, int, int]] = [] # term, off, length, doc_freq
|
||||
for term in sorted_terms:
|
||||
entries = sorted(index[term]) # by doc_id ascending
|
||||
start_off = postings_buf.tell()
|
||||
_write_varint(postings_buf, len(entries))
|
||||
prev = -1
|
||||
for did, tf in entries:
|
||||
_write_varint(postings_buf, did - prev - 1)
|
||||
_write_varint(postings_buf, tf)
|
||||
prev = did
|
||||
end_off = postings_buf.tell()
|
||||
posting_meta.append((term, start_off, end_off - start_off, len(entries)))
|
||||
postings_bytes = postings_buf.getvalue()
|
||||
|
||||
# Build doc-table bytes + doc-offset-table. v2: doc_len after title.
|
||||
docs_buf = io.BytesIO()
|
||||
doc_offsets: list[int] = []
|
||||
for i, r in enumerate(doc_rows):
|
||||
doc_offsets.append(docs_buf.tell())
|
||||
droot_hex = r["document_root"]
|
||||
docs_buf.write(bytes.fromhex(droot_hex))
|
||||
uri = (r["document_uri"] or "").encode("utf-8")
|
||||
title = (r["title"] or "").encode("utf-8")
|
||||
docs_buf.write(struct.pack("<H", min(len(uri), 0xFFFF)))
|
||||
docs_buf.write(uri[:0xFFFF])
|
||||
docs_buf.write(struct.pack("<H", min(len(title), 0xFFFF)))
|
||||
docs_buf.write(title[:0xFFFF])
|
||||
_write_varint(docs_buf, doc_lens[i])
|
||||
docs_bytes = docs_buf.getvalue()
|
||||
|
||||
# Build dict bytes. posting_offset is absolute file offset; compute
|
||||
# after we know where postings start. Layout:
|
||||
# HEADER | DICT | POSTINGS | DOCS | DOC_OFFSETS
|
||||
dict_offset = HEADER_LEN
|
||||
# First pass to size the dict — entries are variable-length so we
|
||||
# walk once to compute total dict_bytes, then once again to write
|
||||
# with correct absolute posting_offsets.
|
||||
dict_bytes_len = 0
|
||||
for term, _, plen, _ in posting_meta:
|
||||
tb = term.encode("utf-8")
|
||||
dict_bytes_len += 2 + len(tb) + 4 + 8 + 4 # term_len, term, df, off, plen
|
||||
postings_offset = dict_offset + dict_bytes_len
|
||||
docs_offset = postings_offset + len(postings_bytes)
|
||||
doc_offsets_offset = docs_offset + len(docs_bytes)
|
||||
|
||||
# Write dict.
|
||||
dict_buf = io.BytesIO()
|
||||
for term, prel_off, plen, df in posting_meta:
|
||||
tb = term.encode("utf-8")
|
||||
dict_buf.write(struct.pack("<H", len(tb)))
|
||||
dict_buf.write(tb)
|
||||
dict_buf.write(struct.pack("<I", df))
|
||||
dict_buf.write(struct.pack("<Q", postings_offset + prel_off))
|
||||
dict_buf.write(struct.pack("<I", plen))
|
||||
dict_bytes = dict_buf.getvalue()
|
||||
assert len(dict_bytes) == dict_bytes_len, (
|
||||
f"dict size mismatch: predicted {dict_bytes_len} wrote {len(dict_bytes)}"
|
||||
)
|
||||
|
||||
# Write doc-offsets table (one u64 per doc).
|
||||
doc_off_table = struct.pack(f"<{docs_count}Q", *doc_offsets)
|
||||
|
||||
# Phase 3: write file.
|
||||
with open(out_path, "wb") as out:
|
||||
out.write(struct.pack(
|
||||
HEADER_FMT, MAGIC, len(sorted_terms), docs_count,
|
||||
dict_offset, postings_offset, docs_offset,
|
||||
))
|
||||
out.write(dict_bytes)
|
||||
out.write(postings_bytes)
|
||||
out.write(docs_bytes)
|
||||
out.write(doc_off_table)
|
||||
|
||||
file_bytes = os.path.getsize(out_path)
|
||||
print(
|
||||
f"sidecar: wrote {file_bytes:,} bytes "
|
||||
f"(dict={len(dict_bytes):,}, postings={len(postings_bytes):,}, "
|
||||
f"docs={len(docs_bytes):,}, doc_offsets={len(doc_off_table):,})"
|
||||
)
|
||||
return {
|
||||
"docs": docs_count,
|
||||
"terms": len(sorted_terms),
|
||||
"postings_bytes": len(postings_bytes),
|
||||
"file_bytes": file_bytes,
|
||||
}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Consumer: SidecarReader(path_or_bytes).
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class SidecarHit:
|
||||
doc_id: int
|
||||
document_root: str
|
||||
document_uri: str
|
||||
title: str
|
||||
score: float # higher = stronger match
|
||||
|
||||
|
||||
class SidecarReader:
|
||||
"""In-memory inverted-index reader.
|
||||
|
||||
Construct from a local file path or in-memory bytes. The dictionary
|
||||
is read once at startup into a sorted list of terms (binary
|
||||
searchable); posting lists are decoded lazily on each query.
|
||||
"""
|
||||
|
||||
def __init__(self, path_or_bytes):
|
||||
if isinstance(path_or_bytes, (bytes, bytearray, memoryview)):
|
||||
self._data: bytes = bytes(path_or_bytes)
|
||||
else:
|
||||
with open(path_or_bytes, "rb") as f:
|
||||
self._data = f.read()
|
||||
magic, dict_count, docs_count, dict_off, postings_off, docs_off = struct.unpack_from(
|
||||
HEADER_FMT, self._data, 0
|
||||
)
|
||||
if magic != MAGIC:
|
||||
raise ValueError(f"not a sidecar file: magic={magic!r}")
|
||||
self.dict_count = dict_count
|
||||
self.docs_count = docs_count
|
||||
self.dict_offset = dict_off
|
||||
self.postings_offset = postings_off
|
||||
self.docs_offset = docs_off
|
||||
|
||||
# Parse dict into parallel arrays for fast binary search.
|
||||
# terms[i] is a UTF-8 str; offs[i] / lens[i] / freqs[i] are
|
||||
# the per-term posting metadata.
|
||||
self.terms: list[str] = []
|
||||
self.posting_offs: list[int] = []
|
||||
self.posting_lens: list[int] = []
|
||||
self.doc_freqs: list[int] = []
|
||||
off = dict_off
|
||||
end = postings_off
|
||||
for _ in range(dict_count):
|
||||
(term_len,) = struct.unpack_from("<H", self._data, off)
|
||||
off += 2
|
||||
term = self._data[off:off + term_len].decode("utf-8")
|
||||
off += term_len
|
||||
(df,) = struct.unpack_from("<I", self._data, off); off += 4
|
||||
(po,) = struct.unpack_from("<Q", self._data, off); off += 8
|
||||
(pl,) = struct.unpack_from("<I", self._data, off); off += 4
|
||||
self.terms.append(term)
|
||||
self.posting_offs.append(po)
|
||||
self.posting_lens.append(pl)
|
||||
self.doc_freqs.append(df)
|
||||
if off != end:
|
||||
raise ValueError(
|
||||
f"dict parse mismatch: ended at {off}, expected {end}"
|
||||
)
|
||||
|
||||
# Doc-offset table at end of file.
|
||||
doc_off_table_off = docs_off + (
|
||||
len(self._data) - docs_off - docs_count * 8
|
||||
)
|
||||
self._doc_offsets: list[int] = list(struct.unpack_from(
|
||||
f"<{docs_count}Q", self._data, doc_off_table_off,
|
||||
))
|
||||
|
||||
# --- term lookup ---
|
||||
|
||||
def _term_index(self, term: str) -> int | None:
|
||||
i = bisect.bisect_left(self.terms, term)
|
||||
if i < len(self.terms) and self.terms[i] == term:
|
||||
return i
|
||||
return None
|
||||
|
||||
def _posting_list(self, term_index: int) -> list[tuple[int, int]]:
|
||||
"""Decode a posting list. Returns [(doc_id, tf), ...]."""
|
||||
po = self.posting_offs[term_index]
|
||||
pl = self.posting_lens[term_index]
|
||||
data = self._data
|
||||
n, off = _read_varint(data, po)
|
||||
out: list[tuple[int, int]] = []
|
||||
prev = -1
|
||||
end = po + pl
|
||||
for _ in range(n):
|
||||
d, off = _read_varint(data, off)
|
||||
tf, off = _read_varint(data, off)
|
||||
cur = prev + 1 + d
|
||||
out.append((cur, tf))
|
||||
prev = cur
|
||||
if off != end:
|
||||
raise ValueError(
|
||||
f"posting parse mismatch for term[{term_index}]: ended at {off}, expected {end}"
|
||||
)
|
||||
return out
|
||||
|
||||
# --- query: AND / OR with score = matched-term count ---
|
||||
|
||||
def search(
|
||||
self,
|
||||
query: str,
|
||||
*,
|
||||
limit: int = 8,
|
||||
mode: str = "or", # or | and
|
||||
k1: float = 1.5,
|
||||
b: float = 0.75,
|
||||
title_boost: float = 8.0,
|
||||
) -> list[SidecarHit]:
|
||||
"""Tokenize ``query`` (matching tokenize_text), look each term
|
||||
up in the dictionary, score with BM25, return top ``limit`` hits.
|
||||
|
||||
BM25 per (doc, term):
|
||||
idf(t) = log((N - df(t) + 0.5) / (df(t) + 0.5) + 1)
|
||||
tf_norm(d, t) = tf(d, t) * (k1 + 1)
|
||||
/ (tf(d, t) + k1 * (1 - b + b * len(d) / avg_len))
|
||||
score(d) = sum over query terms of idf(t) * tf_norm(d, t)
|
||||
|
||||
For ``mode='and'``, posting lists are intersected first; only
|
||||
docs containing every query term get scored.
|
||||
"""
|
||||
import math
|
||||
|
||||
terms = tokenize_text(query)
|
||||
if not terms:
|
||||
return []
|
||||
# Numeral expand so 'final fantasy 7' query also looks up 'vii'
|
||||
# in the dict (built from chunk text where 'Final Fantasy VII'
|
||||
# tokenizes to ['final', 'fantasy', 'vii']). The expansion is
|
||||
# additive — original tokens stay, equivalents joined as OR
|
||||
# so docs matching either form contribute to BM25.
|
||||
terms = sorted(numeral_expand(terms))
|
||||
N = self.docs_count
|
||||
if not hasattr(self, "_avg_doc_len"):
|
||||
# Compute once + cache; needs the full doc_lens. Lazy because
|
||||
# reading every doc's varint up-front costs ~0.5s on a 1M-doc
|
||||
# sidecar.
|
||||
self._doc_len_cache: dict[int, int] = {}
|
||||
self._avg_doc_len = self._compute_avg_doc_len()
|
||||
|
||||
# Gather posting lists for terms that exist.
|
||||
per_term: list[tuple[list[tuple[int, int]], float]] = []
|
||||
for t in terms:
|
||||
ti = self._term_index(t)
|
||||
if ti is None:
|
||||
if mode == "and":
|
||||
return []
|
||||
continue
|
||||
df = self.doc_freqs[ti]
|
||||
idf = math.log((N - df + 0.5) / (df + 0.5) + 1.0)
|
||||
per_term.append((self._posting_list(ti), idf))
|
||||
if not per_term:
|
||||
return []
|
||||
|
||||
if mode == "and":
|
||||
# Intersect doc_id sets.
|
||||
doc_sets = [set(d for d, _ in pl) for pl, _ in per_term]
|
||||
common = doc_sets[0]
|
||||
for s in doc_sets[1:]:
|
||||
common &= s
|
||||
else:
|
||||
common = None # union path below
|
||||
|
||||
scores: dict[int, float] = {}
|
||||
for pl, idf in per_term:
|
||||
for did, tf in pl:
|
||||
if common is not None and did not in common:
|
||||
continue
|
||||
dl = self._doc_length(did)
|
||||
# Avoid div-by-zero on empty docs (shouldn't happen post-
|
||||
# ingest, but be defensive).
|
||||
denom = tf + k1 * (1.0 - b + b * dl / max(self._avg_doc_len, 1.0))
|
||||
tf_norm = tf * (k1 + 1.0) / denom if denom > 0 else 0.0
|
||||
scores[did] = scores.get(did, 0.0) + idf * tf_norm
|
||||
|
||||
# Title-token boost: docs whose TITLE contains a query term get
|
||||
# a fixed bonus per overlap. BM25-only ranking penalizes long
|
||||
# main articles ("Homer Simpson" Wikipedia article) vs short
|
||||
# episode articles even when the main article is the right
|
||||
# primary source. The local query.py compensates via a
|
||||
# separate title-LIKE retrieval route; sidecar emulates with
|
||||
# a render-time boost using the title we already store in the
|
||||
# doc table. Stems both sides (possessive + plural collapse —
|
||||
# matches the title-mismatch verifier's _stem helper) so
|
||||
# "simpsons" (query) overlaps "simpson" (title). title_boost=0
|
||||
# disables; default 8 is enough to lift a #3-by-BM25 long-doc
|
||||
# match above #1-by-BM25 short-doc matches on the homer query.
|
||||
if title_boost > 0 and scores:
|
||||
def _t_stem(t: str) -> str:
|
||||
t = t.replace("'", "").replace("’", "")
|
||||
if len(t) > 4 and t.endswith("s") and not t.endswith("ss"):
|
||||
return t[:-1]
|
||||
return t
|
||||
query_stems = numeral_expand({_t_stem(t) for t in terms})
|
||||
for did in scores:
|
||||
# Same accent-fold + tokenize + numeral-expand on the
|
||||
# title side so 'pokemon'/'pokémon' AND '7'/'VII' both
|
||||
# overlap correctly. Stopword + length filter (mirrors
|
||||
# tokenize_text) drops 's' from "Mona Lisa's", 'of'/'the'
|
||||
# from titles, etc. — otherwise they'd inflate the
|
||||
# extras count below.
|
||||
title = fold_accents(self._doc_title(did).lower())
|
||||
raw_title_tokens: set[str] = set()
|
||||
for tok in _WORD_RE.findall(title):
|
||||
if tok in STOPWORDS:
|
||||
continue
|
||||
if len(tok) <= 1 and not tok.isdigit():
|
||||
continue
|
||||
raw_title_tokens.add(_t_stem(tok))
|
||||
title_tokens = numeral_expand(raw_title_tokens)
|
||||
overlap = len(query_stems & title_tokens)
|
||||
if not overlap:
|
||||
continue
|
||||
# Penalty for extra title tokens not in the query —
|
||||
# without this, "Mona Lisa" (extras 0) and "Mona Lisa's
|
||||
# Revenge" (extras 1: revenge) tied at the same boost
|
||||
# and BM25 picked the shorter movie article over the
|
||||
# painting. extras/2 means each extra halves the per-
|
||||
# overlap-token bonus.
|
||||
extras = len(title_tokens - query_stems)
|
||||
effective = max(0.0, overlap - extras / 2.0)
|
||||
if effective:
|
||||
scores[did] += title_boost * effective
|
||||
|
||||
ranked = sorted(scores.items(), key=lambda x: -x[1])
|
||||
out: list[SidecarHit] = []
|
||||
for did, score in ranked[:limit]:
|
||||
out.append(self._doc_record(did, score=score))
|
||||
return out
|
||||
|
||||
def _doc_title(self, doc_id: int) -> str:
|
||||
"""Read just the title field from the doc table (no allocation
|
||||
of the rest of the doc record)."""
|
||||
off = self.docs_offset + self._doc_offsets[doc_id]
|
||||
off += 32 # document_root
|
||||
(uri_len,) = struct.unpack_from("<H", self._data, off); off += 2
|
||||
off += uri_len
|
||||
(title_len,) = struct.unpack_from("<H", self._data, off); off += 2
|
||||
return self._data[off:off + title_len].decode("utf-8", errors="replace")
|
||||
|
||||
def _doc_length(self, doc_id: int) -> int:
|
||||
"""Read doc_len from the doc table (varint after title)."""
|
||||
cached = getattr(self, "_doc_len_cache", None)
|
||||
if cached is not None and doc_id in cached:
|
||||
return cached[doc_id]
|
||||
off = self.docs_offset + self._doc_offsets[doc_id]
|
||||
off += 32 # document_root
|
||||
(uri_len,) = struct.unpack_from("<H", self._data, off); off += 2
|
||||
off += uri_len
|
||||
(title_len,) = struct.unpack_from("<H", self._data, off); off += 2
|
||||
off += title_len
|
||||
dl, _ = _read_varint(self._data, off)
|
||||
if cached is not None:
|
||||
cached[doc_id] = dl
|
||||
return dl
|
||||
|
||||
def _compute_avg_doc_len(self) -> float:
|
||||
"""Walk every doc-table entry to compute the corpus average.
|
||||
Cost ~0.5s for 1M docs (called once, cached)."""
|
||||
total = 0
|
||||
N = self.docs_count
|
||||
for i in range(N):
|
||||
total += self._doc_length(i)
|
||||
return (total / N) if N else 0.0
|
||||
|
||||
# --- doc-table lookup ---
|
||||
|
||||
def _doc_record(self, doc_id: int, *, score: float = 0.0) -> SidecarHit:
|
||||
if doc_id < 0 or doc_id >= self.docs_count:
|
||||
raise IndexError(f"doc_id {doc_id} out of range")
|
||||
off = self.docs_offset + self._doc_offsets[doc_id]
|
||||
droot = self._data[off:off + 32].hex()
|
||||
off += 32
|
||||
(uri_len,) = struct.unpack_from("<H", self._data, off); off += 2
|
||||
uri = self._data[off:off + uri_len].decode("utf-8")
|
||||
off += uri_len
|
||||
(title_len,) = struct.unpack_from("<H", self._data, off); off += 2
|
||||
title = self._data[off:off + title_len].decode("utf-8")
|
||||
return SidecarHit(
|
||||
doc_id=doc_id, document_root=droot, document_uri=uri,
|
||||
title=title, score=score,
|
||||
)
|
||||
|
||||
def doc_count(self) -> int:
|
||||
return self.docs_count
|
||||
|
||||
def term_count(self) -> int:
|
||||
return self.dict_count
|
||||
|
||||
def stats(self) -> dict:
|
||||
return {
|
||||
"file_bytes": len(self._data),
|
||||
"dict_count": self.dict_count,
|
||||
"docs_count": self.docs_count,
|
||||
"dict_size_bytes": self.postings_offset - self.dict_offset,
|
||||
"postings_size_bytes": self.docs_offset - self.postings_offset,
|
||||
}
|
||||
263
bench/slim_fts_parity_bench.py
Normal file
263
bench/slim_fts_parity_bench.py
Normal file
|
|
@ -0,0 +1,263 @@
|
|||
"""Slim FTS5 sidecar parity bench — does the new cloud path agree with local?
|
||||
|
||||
For each smoke fixture question, run:
|
||||
legacy: arborist query (current local path)
|
||||
corpus: arborist corpus-query (local shards via Corpus protocol)
|
||||
slim-fts: FtsSidecarShardClient + run_query (NEW cloud path, exercised
|
||||
locally by mocking the manifest with file:// sidecars + HTTP
|
||||
chunk-content fallback to the real bucket .db)
|
||||
|
||||
This proves the slim FTS5 cloud path produces the same primary-source
|
||||
pick as the local paths BEFORE we upload anything. If slim-fts matches
|
||||
corpus bit-for-bit (or better, matches legacy too), the bucket upload is
|
||||
safe to do.
|
||||
|
||||
Sidecars are expected at ~/.arborist/sidecar-fts/00*.idx.db. Build with:
|
||||
make sidecar-build-fts-all
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import datetime as _dt
|
||||
import json
|
||||
import os
|
||||
import subprocess
|
||||
import sys
|
||||
import time
|
||||
from pathlib import Path
|
||||
|
||||
REPO = Path(__file__).resolve().parent.parent
|
||||
ARBORIST = REPO / ".venv" / "bin" / "arborist"
|
||||
DEFAULT_SHARDS = Path.home() / ".arborist" / "shards"
|
||||
DEFAULT_SIDECAR_FTS = Path.home() / ".arborist" / "sidecar-fts"
|
||||
DEFAULT_FIXTURE = REPO / "bench" / "qa_questions_smoke.txt"
|
||||
DEFAULT_QWEN_ENDPOINT = "https://qwen.ai.unturf.com/v1"
|
||||
DEFAULT_QWEN_MODEL = "Qwen3.6-27B-UD-Q4_K_XL.gguf"
|
||||
DEFAULT_BUCKET_BLOB_BASE = "https://nyc3.digitaloceanspaces.com/arborist/blobs"
|
||||
DEFAULT_LIVE_MANIFEST = (
|
||||
"https://nyc3.digitaloceanspaces.com/arborist/clones/manifest-sidecar.json"
|
||||
)
|
||||
|
||||
|
||||
def _load_fixture(path: Path) -> list[str]:
|
||||
out = []
|
||||
for line in path.read_text().splitlines():
|
||||
s = line.strip()
|
||||
if s and not s.startswith("#"):
|
||||
out.append(s)
|
||||
return out
|
||||
|
||||
|
||||
def _run_cli(cmd: list[str], *, timeout_s: int) -> dict:
|
||||
t0 = time.time()
|
||||
try:
|
||||
proc = subprocess.run(
|
||||
cmd, capture_output=True, text=True, timeout=timeout_s,
|
||||
env={**os.environ, "ARBORIST_PROGRESS": "0"},
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
return {"_error": f"timeout {timeout_s}s", "_elapsed_s": timeout_s}
|
||||
dt = time.time() - t0
|
||||
if proc.returncode != 0:
|
||||
return {"_error": f"exit {proc.returncode}",
|
||||
"_stderr": proc.stderr[:300], "_elapsed_s": round(dt, 2)}
|
||||
try:
|
||||
d = json.loads(proc.stdout)
|
||||
except json.JSONDecodeError as e:
|
||||
return {"_error": f"parse: {e}", "_elapsed_s": round(dt, 2)}
|
||||
d["_elapsed_s"] = round(dt, 2)
|
||||
return d
|
||||
|
||||
|
||||
def _legacy(q, *, shards_dir):
|
||||
return _run_cli([
|
||||
str(ARBORIST), "--shards-dir", str(shards_dir),
|
||||
"query", "--top-k", "8", "--json", "--burn",
|
||||
"--answer-mode", "claim_lattice", q,
|
||||
], timeout_s=120)
|
||||
|
||||
|
||||
def _corpus(q, *, shards_dir, endpoint, model):
|
||||
return _run_cli([
|
||||
str(ARBORIST), "--shards-dir", str(shards_dir),
|
||||
"corpus-query", q,
|
||||
"--top-k", "4", "--json",
|
||||
"--endpoint", endpoint, "--model", model,
|
||||
], timeout_s=120)
|
||||
|
||||
|
||||
def _slim_fts(
|
||||
q, *, sidecar_dir, live_manifest_url, endpoint, model,
|
||||
):
|
||||
"""Slim FTS5 path: load the LIVE manifest (so chunk-content fallback
|
||||
points at the real big-shard URLs) then splice in
|
||||
``fts_sidecar_url=file://...`` per shard so the slim FTS5 client
|
||||
opens locally without a download.
|
||||
|
||||
Pre-symlinks the file:// URL into the FTS-sidecar cache to avoid a
|
||||
duplicate 2 GB copy through urllib.
|
||||
"""
|
||||
from arborist.wallet.bucket import (
|
||||
load_bucket_manifest, MultiShardSidecarCorpus,
|
||||
_fts_sidecar_cache_path,
|
||||
)
|
||||
from arborist.qa.corpus import SidecarBucketCorpus
|
||||
from arborist.qa.client import OpenAICompatibleClient
|
||||
from arborist.qa.corpus_query import run_query
|
||||
|
||||
manifest = load_bucket_manifest(live_manifest_url)
|
||||
# Splice fts_sidecar_url per shard by matching basename (e.g. 000.db
|
||||
# → 000.idx.db); leave shard ``url`` + ``blob_base`` as the live
|
||||
# values so the missing-blob fallback hits the real big shard.
|
||||
for sh in manifest.shards:
|
||||
basename = sh["url"].rsplit("/", 1)[-1] # "000.db"
|
||||
n = basename.rsplit(".", 1)[0] # "000"
|
||||
sidecar_path = sidecar_dir / f"{n}.idx.db"
|
||||
if not sidecar_path.exists():
|
||||
continue
|
||||
fts_sidecar_url = f"file://{sidecar_path}"
|
||||
cache_path = _fts_sidecar_cache_path(fts_sidecar_url)
|
||||
if not os.path.exists(cache_path):
|
||||
os.symlink(sidecar_path, cache_path)
|
||||
sh["fts_sidecar_url"] = fts_sidecar_url
|
||||
# Drop shards without a slim sidecar (e.g. virtback) — parity bench
|
||||
# only exercises the four wikipedia shards we built.
|
||||
manifest.shards = [
|
||||
sh for sh in manifest.shards if sh.get("fts_sidecar_url")
|
||||
]
|
||||
|
||||
t0 = time.time()
|
||||
multi = MultiShardSidecarCorpus(manifest)
|
||||
corpus = SidecarBucketCorpus(multi)
|
||||
|
||||
client = OpenAICompatibleClient(base_url=endpoint)
|
||||
result = run_query(
|
||||
corpus, q, client,
|
||||
model_id=model, top_k=4, max_context_chars=24_000,
|
||||
)
|
||||
multi.close()
|
||||
elapsed = time.time() - t0
|
||||
|
||||
out = dict(result)
|
||||
out["_elapsed_s"] = round(elapsed, 2)
|
||||
return out
|
||||
|
||||
|
||||
def _primary(r: dict) -> dict:
|
||||
srcs = r.get("sources") or []
|
||||
if not srcs:
|
||||
return {"title": "", "uri": ""}
|
||||
pri = next(
|
||||
(s for s in srcs if s.get("source_role") == "primary_answer_source"),
|
||||
srcs[0],
|
||||
)
|
||||
return {"title": (pri.get("title") or "")[:55],
|
||||
"uri": pri.get("document_uri") or ""}
|
||||
|
||||
|
||||
def main():
|
||||
p = argparse.ArgumentParser()
|
||||
p.add_argument("--fixture", type=Path, default=DEFAULT_FIXTURE)
|
||||
p.add_argument("--shards-dir", type=Path, default=DEFAULT_SHARDS)
|
||||
p.add_argument("--sidecar-dir", type=Path, default=DEFAULT_SIDECAR_FTS)
|
||||
p.add_argument("--endpoint", default=DEFAULT_QWEN_ENDPOINT)
|
||||
p.add_argument("--model", default=DEFAULT_QWEN_MODEL)
|
||||
p.add_argument("--live-manifest", default=DEFAULT_LIVE_MANIFEST,
|
||||
help="real bucket manifest URL — provides the shard "
|
||||
"URLs used for the chunk-content fallback")
|
||||
p.add_argument(
|
||||
"--out-dir", type=Path,
|
||||
default=REPO / "bench" / "slim_fts_parity_results",
|
||||
)
|
||||
p.add_argument("--skip-legacy", action="store_true")
|
||||
p.add_argument("--skip-corpus", action="store_true")
|
||||
args = p.parse_args()
|
||||
|
||||
questions = _load_fixture(args.fixture)
|
||||
args.out_dir.mkdir(parents=True, exist_ok=True)
|
||||
stamp = _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H-%M-%SZ")
|
||||
out_path = args.out_dir / f"{stamp}.jsonl"
|
||||
|
||||
print(f"# {len(questions)} questions from {args.fixture}", file=sys.stderr)
|
||||
print(f"# shards: {args.shards_dir}", file=sys.stderr)
|
||||
print(f"# sidecars-fts: {args.sidecar_dir}", file=sys.stderr)
|
||||
print(f"# live manifest: {args.live_manifest}", file=sys.stderr)
|
||||
print(f"# llm: {args.endpoint} / {args.model}", file=sys.stderr)
|
||||
print(file=sys.stderr)
|
||||
|
||||
n_agree = 0
|
||||
n_disagree = 0
|
||||
|
||||
with open(out_path, "w") as f:
|
||||
for i, q in enumerate(questions, 1):
|
||||
print(f"[{i}/{len(questions)}] {q}", file=sys.stderr)
|
||||
legacy = {} if args.skip_legacy else _legacy(
|
||||
q, shards_dir=args.shards_dir)
|
||||
corpus = {} if args.skip_corpus else _corpus(
|
||||
q, shards_dir=args.shards_dir,
|
||||
endpoint=args.endpoint, model=args.model)
|
||||
slim = _slim_fts(
|
||||
q,
|
||||
sidecar_dir=args.sidecar_dir,
|
||||
live_manifest_url=args.live_manifest,
|
||||
endpoint=args.endpoint, model=args.model,
|
||||
)
|
||||
|
||||
row = {"question": q}
|
||||
for label, r in (
|
||||
("legacy", legacy), ("corpus", corpus), ("slim_fts", slim),
|
||||
):
|
||||
row[label] = {
|
||||
"audit_mode": r.get("audit_mode"),
|
||||
"n_verified": r.get("n_verified"),
|
||||
"n_quotes": r.get("n_quotes"),
|
||||
"primary": _primary(r),
|
||||
"elapsed_s": r.get("_elapsed_s") or (
|
||||
r.get("timings") or {}).get("total"),
|
||||
"error": r.get("_error") or r.get("error"),
|
||||
}
|
||||
uris = {
|
||||
lbl: row[lbl]["primary"]["uri"]
|
||||
for lbl in ("legacy", "corpus", "slim_fts")
|
||||
if row[lbl]["primary"]["uri"]
|
||||
}
|
||||
distinct = set(uris.values())
|
||||
row["agreement"] = {
|
||||
"n_paths_with_source": len(uris),
|
||||
"distinct_sources": len(distinct),
|
||||
"all_agree": len(distinct) <= 1,
|
||||
}
|
||||
if row["agreement"]["all_agree"]:
|
||||
n_agree += 1
|
||||
else:
|
||||
n_disagree += 1
|
||||
|
||||
f.write(json.dumps(row, ensure_ascii=False) + "\n")
|
||||
f.flush()
|
||||
|
||||
for lbl in ("legacy", "corpus", "slim_fts"):
|
||||
r = row[lbl]
|
||||
badge = (
|
||||
" ←" if r["primary"]["uri"]
|
||||
and len(distinct) > 1
|
||||
and r["primary"]["uri"] != min(uris.values()) else ""
|
||||
)
|
||||
err = f" ERR={r['error']}" if r["error"] else ""
|
||||
print(f" {lbl:9s} {(r['audit_mode'] or '?'):<22} "
|
||||
f"{r['primary']['title']}{badge}{err}",
|
||||
file=sys.stderr)
|
||||
if not row["agreement"]["all_agree"]:
|
||||
print(f" SOURCE DISAGREEMENT: {len(distinct)} distinct primaries",
|
||||
file=sys.stderr)
|
||||
print(file=sys.stderr)
|
||||
|
||||
print(f"=== SUMMARY ===", file=sys.stderr)
|
||||
print(f" {len(questions)} questions", file=sys.stderr)
|
||||
print(f" {n_agree} all-paths agree on primary source", file=sys.stderr)
|
||||
print(f" {n_disagree} paths-disagree", file=sys.stderr)
|
||||
print(f" results: {out_path}", file=sys.stderr)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
194
scripts/upload_fts_sidecars.py
Normal file
194
scripts/upload_fts_sidecars.py
Normal file
|
|
@ -0,0 +1,194 @@
|
|||
"""Upload slim FTS5 sidecars to the bucket + publish manifest-fts.json.
|
||||
|
||||
Reads slim FTS5 sidecars from ~/.arborist/sidecar-fts/00*.idx.db, uploads
|
||||
them to s3://${ARBORIST_COLD_BUCKET}/clones/sidecars-fts/<n>.idx.db, and
|
||||
publishes a new manifest at clones/manifest-fts.json that references both:
|
||||
- the new fts_sidecar_url (slim FTS5)
|
||||
- the existing shard URL from manifest-sidecar.json (chunk-content fallback)
|
||||
|
||||
Idempotent: skips uploads whose remote size matches local. Existing
|
||||
manifest-sidecar.json is NOT modified.
|
||||
|
||||
Credentials via boto3 standard discovery (env vars / ~/.aws/credentials).
|
||||
Operation Voyeur: credentials never appear in argv, logs, or this file.
|
||||
|
||||
Usage:
|
||||
.venv/bin/python3 scripts/upload_fts_sidecars.py [--dry-run]
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import urllib.request
|
||||
from pathlib import Path
|
||||
|
||||
SIDECAR_DIR = Path.home() / ".arborist" / "sidecar-fts"
|
||||
PREFIX = "clones/sidecars-fts"
|
||||
NEW_MANIFEST_KEY = "clones/manifest-fts.json"
|
||||
EXISTING_MANIFEST_URL = (
|
||||
"https://nyc3.digitaloceanspaces.com/arborist/clones/manifest-sidecar.json"
|
||||
)
|
||||
|
||||
|
||||
def main():
|
||||
ap = argparse.ArgumentParser()
|
||||
ap.add_argument("--dry-run", action="store_true",
|
||||
help="print what would happen; upload nothing")
|
||||
args = ap.parse_args()
|
||||
|
||||
bucket = os.environ.get("ARBORIST_COLD_BUCKET")
|
||||
endpoint = os.environ.get("ARBORIST_COLD_ENDPOINT_URL")
|
||||
if not bucket or not endpoint:
|
||||
print("ERROR: set ARBORIST_COLD_BUCKET + ARBORIST_COLD_ENDPOINT_URL",
|
||||
file=sys.stderr)
|
||||
return 2
|
||||
|
||||
sidecars = sorted(SIDECAR_DIR.glob("[0-9][0-9][0-9].idx.db"))
|
||||
if not sidecars:
|
||||
print(f"ERROR: no sidecars at {SIDECAR_DIR}/[0-9][0-9][0-9].idx.db",
|
||||
file=sys.stderr)
|
||||
return 2
|
||||
|
||||
print(f"bucket: s3://{bucket}/", file=sys.stderr)
|
||||
print(f"endpoint: {endpoint}", file=sys.stderr)
|
||||
print(f"prefix: {PREFIX}/", file=sys.stderr)
|
||||
print(f"sidecars: {len(sidecars)} files ({sum(p.stat().st_size for p in sidecars)/1e9:.2f} GB)",
|
||||
file=sys.stderr)
|
||||
print(file=sys.stderr)
|
||||
|
||||
# Load the existing manifest's shard URLs so the new manifest reuses
|
||||
# them for the chunk-content fallback. We don't touch the existing
|
||||
# manifest itself — just borrow the URLs.
|
||||
with urllib.request.urlopen(EXISTING_MANIFEST_URL, timeout=30) as resp:
|
||||
existing = json.loads(resp.read().decode())
|
||||
# Map by trailing 3-digit shard index parsed from label
|
||||
# ("genesis-shard-000" → "000", "genesis-shard-002-64k" → "002").
|
||||
# Drops non-genesis shards (virtback, etc.) so the basename collision
|
||||
# virtback/000.db vs full-bench/000.db doesn't matter.
|
||||
import re as _re
|
||||
_GENESIS_IDX = _re.compile(r"genesis-shard-(\d{3})")
|
||||
by_idx = {}
|
||||
for sh in existing.get("shards", []):
|
||||
m = _GENESIS_IDX.search(sh.get("label", ""))
|
||||
if m:
|
||||
by_idx[m.group(1)] = sh
|
||||
|
||||
try:
|
||||
import boto3
|
||||
from boto3.s3.transfer import TransferConfig
|
||||
except ImportError:
|
||||
print("ERROR: boto3 not installed. run: make bootstrap-object-store",
|
||||
file=sys.stderr)
|
||||
return 2
|
||||
|
||||
s3 = boto3.client("s3", endpoint_url=endpoint, region_name=os.environ.get(
|
||||
"ARBORIST_COLD_REGION", "us-east-1"))
|
||||
# 64 MB parts; parallel transfer.
|
||||
cfg = TransferConfig(
|
||||
multipart_threshold=64 * 1024 * 1024,
|
||||
multipart_chunksize=64 * 1024 * 1024,
|
||||
max_concurrency=4,
|
||||
)
|
||||
|
||||
new_shards = []
|
||||
for sp in sidecars:
|
||||
n = sp.stem.split(".")[0] # 000.idx → 000
|
||||
key = f"{PREFIX}/{n}.idx.db"
|
||||
local_size = sp.stat().st_size
|
||||
|
||||
# Idempotency: skip if remote size matches.
|
||||
try:
|
||||
head = s3.head_object(Bucket=bucket, Key=key)
|
||||
remote_size = head["ContentLength"]
|
||||
if remote_size == local_size:
|
||||
print(f" skip {key} (already {local_size/1e9:.2f} GB)",
|
||||
file=sys.stderr)
|
||||
else:
|
||||
print(f" remote size mismatch on {key} "
|
||||
f"(local {local_size}, remote {remote_size}) — re-uploading",
|
||||
file=sys.stderr)
|
||||
if not args.dry_run:
|
||||
s3.upload_file(str(sp), bucket, key, Config=cfg)
|
||||
else:
|
||||
print(f" [dry-run] would re-upload {key}",
|
||||
file=sys.stderr)
|
||||
except s3.exceptions.ClientError as e:
|
||||
code = e.response.get("Error", {}).get("Code")
|
||||
if code in ("404", "NoSuchKey", "NotFound"):
|
||||
if args.dry_run:
|
||||
print(f" [dry-run] would upload {key} ({local_size/1e9:.2f} GB)",
|
||||
file=sys.stderr)
|
||||
else:
|
||||
print(f" upload {key} ({local_size/1e9:.2f} GB) ...",
|
||||
file=sys.stderr)
|
||||
s3.upload_file(str(sp), bucket, key, Config=cfg)
|
||||
s3.put_object_acl(Bucket=bucket, Key=key, ACL="public-read")
|
||||
print(f" done", file=sys.stderr)
|
||||
else:
|
||||
raise
|
||||
|
||||
# Ensure ACL is public-read (idempotent for ACL-aware backends).
|
||||
if not args.dry_run:
|
||||
try:
|
||||
s3.put_object_acl(Bucket=bucket, Key=key, ACL="public-read")
|
||||
except Exception:
|
||||
pass # endpoint may not honor ACLs
|
||||
|
||||
# Build new manifest entry — reuse existing shard url + blob_base.
|
||||
ex = by_idx.get(n)
|
||||
if ex is None:
|
||||
print(f" WARNING: no existing genesis-shard-{n} in live manifest; "
|
||||
f"the new manifest will lack a content-fallback URL "
|
||||
f"(slim FTS5 retrieval still works; blob fetch falls back "
|
||||
f"to bucket blobs/<hash> which 403s today)",
|
||||
file=sys.stderr)
|
||||
new_shard = {
|
||||
"url": None,
|
||||
"shard_idx": int(n),
|
||||
"label": f"genesis-shard-{n}",
|
||||
"fts_sidecar_url": f"{endpoint}/{bucket}/{key}",
|
||||
}
|
||||
else:
|
||||
new_shard = dict(ex)
|
||||
new_shard["fts_sidecar_url"] = f"{endpoint}/{bucket}/{key}"
|
||||
new_shard.pop("sidecar_url", None) # this manifest is FTS5-only
|
||||
new_shards.append(new_shard)
|
||||
|
||||
new_manifest = {
|
||||
"shards": new_shards,
|
||||
"blob_base": existing.get("blob_base"),
|
||||
"snapshot_root": existing.get("snapshot_root"),
|
||||
"snapshot_ts": existing.get("snapshot_ts"),
|
||||
"format": "fts5-slim-v1",
|
||||
}
|
||||
manifest_bytes = json.dumps(new_manifest, indent=2).encode()
|
||||
print(file=sys.stderr)
|
||||
print(f"new manifest: {NEW_MANIFEST_KEY} ({len(manifest_bytes)} bytes)",
|
||||
file=sys.stderr)
|
||||
print(f" shards: {len(new_shards)}", file=sys.stderr)
|
||||
for sh in new_shards:
|
||||
print(f" fts_sidecar={sh['fts_sidecar_url'].rsplit('/',1)[-1]:14s} "
|
||||
f"shard_url={sh.get('url','(none)').rsplit('/',1)[-1]}",
|
||||
file=sys.stderr)
|
||||
|
||||
if args.dry_run:
|
||||
print(file=sys.stderr)
|
||||
print("[dry-run] manifest content:", file=sys.stderr)
|
||||
print(manifest_bytes.decode(), file=sys.stderr)
|
||||
return 0
|
||||
|
||||
s3.put_object(
|
||||
Bucket=bucket, Key=NEW_MANIFEST_KEY,
|
||||
Body=manifest_bytes,
|
||||
ContentType="application/json",
|
||||
ACL="public-read",
|
||||
)
|
||||
print(f"published: {endpoint}/{bucket}/{NEW_MANIFEST_KEY}",
|
||||
file=sys.stderr)
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
sys.exit(main())
|
||||
|
|
@ -122,7 +122,8 @@ def test_sidecar_bucket_corpus_satisfies_protocol():
|
|||
corpus = SidecarBucketCorpus(_StubMulti())
|
||||
assert isinstance(corpus, Corpus)
|
||||
assert corpus.name == "sidecar-bucket"
|
||||
assert corpus.higher_is_better is True
|
||||
# FTS5 bm25 is negative; lower (more negative) = better.
|
||||
assert corpus.higher_is_better is False
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -53,7 +53,7 @@ from arborist.qa.verify import verify_claim_lattice
|
|||
from arborist.source import Source
|
||||
from arborist.store import connect
|
||||
from arborist.wallet.bucket import BucketManifest, MultiShardSidecarCorpus
|
||||
from arborist.wallet.sidecar import build_sidecar
|
||||
from arborist.wallet.fts_sidecar_build import build as build_fts_sidecar
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -101,14 +101,14 @@ def both_adapters(tmp_path: Path):
|
|||
c0.execute("PRAGMA wal_checkpoint(FULL)")
|
||||
c0.close()
|
||||
|
||||
# bucket layout + sidecar
|
||||
# bucket layout + slim FTS5 sidecar
|
||||
bucket = tmp_path / "bucket"
|
||||
clones = bucket / "clones" / "snap-1"
|
||||
clones.mkdir(parents=True)
|
||||
bucket_shard = clones / "000.db"
|
||||
shutil.copy(db, bucket_shard)
|
||||
sidecar_path = clones / "000.sidecar.bin"
|
||||
build_sidecar(bucket_shard, sidecar_path)
|
||||
sidecar_path = clones / "000.idx.db"
|
||||
build_fts_sidecar(str(bucket_shard), str(sidecar_path), verbose=False)
|
||||
|
||||
# in-process Range-aware server
|
||||
cwd_before = os.getcwd()
|
||||
|
|
@ -123,14 +123,14 @@ def both_adapters(tmp_path: Path):
|
|||
"url": f"{base_url}/clones/snap-1/000.db",
|
||||
"shard_idx": 0,
|
||||
"label": "test-shard",
|
||||
"sidecar_url": f"{base_url}/clones/snap-1/000.sidecar.bin",
|
||||
"fts_sidecar_url": f"{base_url}/clones/snap-1/000.idx.db",
|
||||
}],
|
||||
blob_base=f"{base_url}/blobs",
|
||||
snapshot_root=None,
|
||||
snapshot_ts=None,
|
||||
)
|
||||
multi = MultiShardSidecarCorpus(
|
||||
manifest, sidecar_cache_dir=str(tmp_path / "cache"),
|
||||
manifest, fts_sidecar_cache_dir=str(tmp_path / "cache"),
|
||||
)
|
||||
conn = connect(db)
|
||||
|
||||
|
|
|
|||
|
|
@ -45,7 +45,7 @@ from arborist.wallet.bucket import (
|
|||
BucketManifest,
|
||||
MultiShardSidecarCorpus,
|
||||
)
|
||||
from arborist.wallet.sidecar import build_sidecar
|
||||
from arborist.wallet.fts_sidecar_build import build as build_fts_sidecar
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
@ -90,8 +90,8 @@ def served_corpus(tmp_path: Path):
|
|||
import shutil
|
||||
bucket_shard = clones / "000.db"
|
||||
shutil.copy(db, bucket_shard)
|
||||
sidecar_path = clones / "000.sidecar.bin"
|
||||
build_sidecar(bucket_shard, sidecar_path)
|
||||
sidecar_path = clones / "000.idx.db"
|
||||
build_fts_sidecar(str(bucket_shard), str(sidecar_path), verbose=False)
|
||||
|
||||
# Re-open the original for SqliteShardCorpus tests.
|
||||
conn = connect(db)
|
||||
|
|
@ -110,13 +110,13 @@ def served_corpus(tmp_path: Path):
|
|||
"url": f"{base_url}/clones/snap-1/000.db",
|
||||
"shard_idx": 0,
|
||||
"label": "test-shard",
|
||||
"sidecar_url": f"{base_url}/clones/snap-1/000.sidecar.bin",
|
||||
"fts_sidecar_url": f"{base_url}/clones/snap-1/000.idx.db",
|
||||
}],
|
||||
blob_base=f"{base_url}/blobs",
|
||||
snapshot_root=None,
|
||||
snapshot_ts=None,
|
||||
)
|
||||
multi = MultiShardSidecarCorpus(manifest, sidecar_cache_dir=str(tmp_path / "cache"))
|
||||
multi = MultiShardSidecarCorpus(manifest, fts_sidecar_cache_dir=str(tmp_path / "cache"))
|
||||
|
||||
try:
|
||||
yield conn, multi, base_url
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue