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

qa/corpus: wire fts_title (documents_fts) + fts_phrase (FTS5 phrase MATCH)

Browse source 
Completes the Corpus protocol surface. Both routes already worked at the
storage layer (every shard has chunks_fts AND documents_fts virtual
tables; FTS5 supports MATCH '"phrase"' natively); they just weren't
plumbed through. Now SqliteShardCorpus, MultiShardSqliteCorpus,
BucketClient, FtsSidecarShardClient, MultiShardSidecarCorpus, and
SidecarBucketCorpus all expose fts_title + fts_phrase end-to-end.

Bonus fix in arborist/ingest.py: every shard's INSERT INTO documents
now also INSERTs into documents_fts in the same transaction. Without
this, fts_title returned empty on freshly-ingested shards — only
migrated genesis shards had documents_fts populated. The cost is one
FTS5 row per new doc, negligible vs the chunk inserts.

Also includes operational cleanup (E):
  - Deleted s3://arborist/clones/manifest-sidecar.json
  - Deleted s3://arborist/clones/full-bench-64k/00[0-3].sidecar.bin
    (~3.1 GB reclaimed; the slim FTS5 manifest is now the only cloud
    surface)
  - Fixed MultiShardSidecarCorpus.stats() — was calling .bucket on
    FtsSidecarShardClient (attribute went away when SidecarShardClient
    was deleted); now calls .stats() directly on whichever client.

Validation
----------
- Local fts_title("dinosaur"): "Dinosaur" main article ranks #1
- Local fts_phrase(["always been at war"]): "Nineteen Eighty-Four" ranks
  #1 (the canonical phrase-route test from CLAUDE.md)
- Cloud (slim FTS5 sidecar) fts_title / fts_phrase: IDENTICAL ranking to
  local on the same query (bit-for-bit FTS5 parity preserved)
- Full pytest suite: 2737 passed, 28 skipped, 1 xfailed (the same two
  pre-existing failures from main HEAD)
This commit is contained in:
russell@unturf.com 2026-05-31 11:21:57 -04:00
parent bc3f94efeb
commit 18db3c2caf
No known key found for this signature in database
4 changed files with 317 additions and 43 deletions
Download patch file Download diff file Expand all files Collapse all files

13
arborist/ingest.py
View file

@ -343,6 +343,19 @@ def _flush_batch(
"VALUES (?, ?, ?, 'surface', 0, ?, ?, ?, ?, ?)",
documents_rows,
)
# Keep documents_fts (the title FTS5 index) in sync with the
# documents table on every ingest. Without this the fts_title
# route on a freshly ingested shard returns empty — only
# genesis/migrated shards had it populated. The cost is one
# FTS5 INSERT per new doc, negligible vs the chunk inserts
# below. Skip rows with NULL title (FTS5 indexes the column
# value; NULL would yield a row with no terms — wasted).
conn.executemany(
"INSERT INTO documents_fts (rowid, title) "
"SELECT rowid, title FROM documents "
"WHERE document_root = ? AND title IS NOT NULL",
[(r[0],) for r in documents_rows],
)
if chunk_rows:
conn.executemany(
"INSERT INTO chunks (chunk_id, document_root, idx, leaf_hash, content) "

134
arborist/qa/corpus.py
View file

@ -292,16 +292,66 @@ class SqliteShardCorpus:
]
def fts_title(self, query: str, *, limit: int = 8) -> list[Hit]:
raise NotSupportedError(
"fts_title not yet routed through SqliteShardCorpus — "
"query.py owns title-LIKE today; will move in the next refactor."
"""Title-only bm25() via the documents_fts virtual table that
lives alongside chunks_fts in every shard. Uses the SAME
sanitizer as fts_body so a user query with punctuation /
stopwords gets the same shape on both routes."""
from arborist.wallet.bucket import _to_fts5
match_expr = _to_fts5(query)
if not match_expr.strip():
return []
sql = (
"SELECT d.document_root, d.document_uri, d.title, "
" bm25(documents_fts) AS score "
"FROM documents_fts "
"JOIN documents d ON d.rowid = documents_fts.rowid "
"WHERE documents_fts MATCH ? "
"ORDER BY score LIMIT ?"
)
rows = list(self._conn.execute(sql, (match_expr, limit)))
return [
Hit(
document_root=r["document_root"],
document_uri=r["document_uri"] or "",
title=r["title"] or "",
score=r["score"],
shard_id=None,
)
for r in rows
]
def fts_phrase(self, ngrams, *, limit: int = 8) -> list[Hit]:
raise NotSupportedError(
"fts_phrase not yet routed through SqliteShardCorpus — "
"query.py owns phrase-pattern today; will move in the next refactor."
"""Verbatim phrase search via FTS5's native MATCH '"phrase"'
operator. Builds ``"ng1" OR "ng2" OR …`` so any matching
phrase contributes. Internal double-quotes in n-grams are
stripped (FTS5 has no phrase-escape — drop rather than fail)."""
cleaned = [
ng.replace('"', '').strip()
for ng in ngrams if ng and ng.strip()
]
cleaned = [c for c in cleaned if c]
if not cleaned:
return []
match_expr = " OR ".join(f'"{c}"' for c in cleaned)
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, (match_expr, limit)))
return [
Hit(
document_root=r["document_root"],
document_uri=r["document_uri"] or "",
title=r["title"] or "",
score=r["score"],
shard_id=None,
)
for r in rows
]
def chunks_for_doc(
self, document_root: str, *, limit: int | None = None
@ -418,14 +468,37 @@ class MultiShardSqliteCorpus:
return merged[:limit]
def fts_title(self, query: str, *, limit: int = 8) -> list[Hit]:
raise NotSupportedError(
"fts_title not yet routed through MultiShardSqliteCorpus."
)
"""Per-shard fts_title merged by raw bm25 score (ASC)."""
return self._fanout("fts_title", limit, query=query)
def fts_phrase(self, ngrams, *, limit: int = 8) -> list[Hit]:
raise NotSupportedError(
"fts_phrase not yet routed through MultiShardSqliteCorpus."
)
"""Per-shard fts_phrase merged by raw bm25 score (ASC).
Materializes ``ngrams`` to a list — Iterable might be a
one-shot generator and each shard needs to re-walk it.
"""
ngs = list(ngrams)
return self._fanout("fts_phrase", limit, ngrams=ngs)
def _fanout(self, method: str, limit: int, **kwargs) -> list[Hit]:
"""Common per-shard fan-out + merge for title/phrase routes.
Same shape as fts_body's merge: each shard returns its top-K,
we annotate shard_id, then sort ASC by bm25 (lower=better)."""
merged: list[Hit] = []
for sh_path, sc in self._shards:
hits = getattr(sc, method)(limit=limit * 2, **kwargs)
for h in hits:
merged.append(Hit(
document_root=h.document_root,
document_uri=h.document_uri,
title=h.title,
score=h.score,
shard_id=sh_path,
extras=dict(h.extras),
))
merged.sort(key=lambda h: h.score)
return merged[:limit]
def chunks_for_doc(
self, document_root: str, *, limit: int | None = None
@ -501,18 +574,35 @@ class SidecarBucketCorpus:
]
def fts_title(self, query: str, *, limit: int = 8) -> list[Hit]:
raise NotSupportedError(
"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."
)
"""Title-only bm25 via documents_fts on each shard."""
hits = self._mscorpus.fts_title_search(query, limit=limit)
return [
Hit(
document_root=h["document_root"],
document_uri=h.get("document_uri", ""),
title=h.get("title", ""),
score=h.get("score", 0.0),
shard_id=h.get("_shard_url"),
extras={"merge_score": h.get("merge_score")} if "merge_score" in h else {},
)
for h in hits
]
def fts_phrase(self, ngrams, *, limit: int = 8) -> list[Hit]:
raise NotSupportedError(
"fts_phrase not implemented for SidecarBucketCorpus — "
"FTS5 supports MATCH '\"phrase here\"' directly; the route "
"just isn't surfaced here yet."
)
"""Verbatim phrase search via FTS5 MATCH '"phrase"' on each
shard's chunks_fts."""
hits = self._mscorpus.fts_phrase_search(list(ngrams), limit=limit)
return [
Hit(
document_root=h["document_root"],
document_uri=h.get("document_uri", ""),
title=h.get("title", ""),
score=h.get("score", 0.0),
shard_id=h.get("_shard_url"),
extras={"merge_score": h.get("merge_score")} if "merge_score" in h else {},
)
for h in hits
]
def chunks_for_doc(
self, document_root: str, *, limit: int | None = None

151
arborist/wallet/bucket.py
View file

@ -494,6 +494,56 @@ class BucketClient:
for r in rows
]
def fts_title_search(
self, query: str, *, limit: int = 8, raw: bool = False,
) -> list[dict]:
"""Title-only bm25 via documents_fts. Same sanitizer as
fts_search so callers don't think about query shape."""
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(documents_fts) AS score "
"FROM documents_fts "
"JOIN documents d ON d.rowid = documents_fts.rowid "
"WHERE documents_fts MATCH ? "
"ORDER BY score LIMIT ?"
)
rows = list(self.conn.execute(sql, (fts_query, limit)))
return [
{"document_root": r[0], "document_uri": r[1],
"title": r[2], "score": r[3]}
for r in rows
]
def fts_phrase_search(
self, ngrams, *, limit: int = 8,
) -> list[dict]:
"""Verbatim phrase search via FTS5's native MATCH '"phrase"'."""
cleaned = [
ng.replace('"', '').strip()
for ng in ngrams if ng and ng.strip()
]
cleaned = [c for c in cleaned if c]
if not cleaned:
return []
match_expr = " OR ".join(f'"{c}"' for c in cleaned)
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, (match_expr, limit)))
return [
{"document_root": r[0], "document_uri": r[1],
"title": r[2], "score": r[3]}
for r in rows
]
def chunks_for_doc(self, document_root: str) -> list[dict]:
"""Return {idx, leaf_hash} for every chunk of a document."""
sql = (
@ -828,6 +878,60 @@ class FtsSidecarShardClient:
for r in rows
]
def fts_title_search(
self, query: str, *, limit: int = 8, raw: bool = False,
) -> list[dict]:
"""Title-only bm25 via the slim sidecar's documents_fts.
Same FTS5 query shape as BucketClient.fts_title_search — the
slim sidecar contains the same documents_fts shadow tables
the source shard does.
"""
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(documents_fts) AS score "
"FROM documents_fts "
"JOIN documents d ON d.rowid = documents_fts.rowid "
"WHERE documents_fts MATCH ? "
"ORDER BY score LIMIT ?"
)
rows = list(self._conn.execute(sql, (fts_query, limit)))
return [
{"document_root": r[0], "document_uri": r[1],
"title": r[2], "score": r[3]}
for r in rows
]
def fts_phrase_search(
self, ngrams, *, limit: int = 8,
) -> list[dict]:
"""Verbatim phrase search against the slim sidecar's chunks_fts."""
cleaned = [
ng.replace('"', '').strip()
for ng in ngrams if ng and ng.strip()
]
cleaned = [c for c in cleaned if c]
if not cleaned:
return []
match_expr = " OR ".join(f'"{c}"' for c in cleaned)
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, (match_expr, limit)))
return [
{"document_root": r[0], "document_uri": r[1],
"title": r[2], "score": r[3]}
for r in rows
]
def chunks_for_doc(self, document_root: str) -> list[dict]:
"""Return chunk metadata (idx, leaf_hash) for a document.
@ -966,21 +1070,51 @@ class MultiShardSidecarCorpus:
def fts_search(
self, query: str, *, limit: int = 8, raw: bool = False,
) -> list[dict]:
"""Run each shard's FTS in parallel, merge by raw bm25 score.
"""Body-FTS5 across all shards, merged by MIN raw bm25."""
return self._fanout(
"fts_search", limit, query=query, raw=raw,
)
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.
def fts_title_search(
self, query: str, *, limit: int = 8, raw: bool = False,
) -> list[dict]:
"""Title-FTS5 across all shards. Backends without the method
(e.g. BucketClient on a shard with no documents_fts) contribute
nothing — fall-open."""
return self._fanout(
"fts_title_search", limit, query=query, raw=raw,
)
def fts_phrase_search(
self, ngrams, *, limit: int = 8,
) -> list[dict]:
"""Verbatim phrase MATCH across all shards. Materialize the
ngrams iterable so each shard's lambda re-walk works."""
ngs = list(ngrams)
return self._fanout(
"fts_phrase_search", limit, ngrams=ngs,
)
def _fanout(self, method: str, limit: int, **kwargs) -> list[dict]:
"""Shared per-shard fan-out + MIN-score merge.
Every backend returns FTS5's negative bm25 (lower = better),
so MIN per document_root + sort ASC = correct global ranking.
Per-shard failures (missing method on a backend, query error)
are swallowed to keep the fan-out fail-open.
"""
from concurrent.futures import ThreadPoolExecutor
def _one(item):
sh_url, sh_client = item
fn = getattr(sh_client, method, None)
if fn is None:
return sh_url, []
try:
# Oversample 4× per shard so post-merge ranking has more
# Oversample 4× per shard so the merge 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)
return sh_url, fn(limit=limit * 4, **kwargs)
except Exception:
return sh_url, []
@ -1021,8 +1155,9 @@ class MultiShardSidecarCorpus:
for sh, c, has_sc in zip(
self.manifest.shards, self.shard_clients, self._has_sidecar,
):
bucket = c.bucket if has_sc else c # type: ignore[attr-defined]
s = bucket.stats()
# Both FtsSidecarShardClient and BucketClient expose .stats()
# directly now; no .bucket attribute on the slim client.
s = c.stats()
merged["total_http_requests"] += s.get("http_requests") or 0
merged["total_bytes_fetched"] += (
(s.get("cache") or {}).get("bytes_fetched") or 0

62
tests/test_qa_corpus.py
View file

@ -159,40 +159,76 @@ def test_sqlite_fts_body_returns_hit_shape(shard_conn):
# ---------------------------------------------------------------------------
# Routes that should NotSupportedError
# fts_title + fts_phrase routes (documents_fts + FTS5 phrase MATCH)
# ---------------------------------------------------------------------------
def test_sqlite_fts_title_not_supported(shard_conn):
def test_sqlite_fts_title_via_documents_fts(shard_conn):
"""fts_title returns title-bm25 ranked hits from documents_fts."""
corpus = SqliteShardCorpus(shard_conn)
with pytest.raises(NotSupportedError):
corpus.fts_title("anarchism")
hits = corpus.fts_title("anarchism", limit=5)
# Should rank the Anarchism doc highly via title MATCH.
assert any("Anarchism" in h.title for h in hits)
def test_sqlite_fts_phrase_not_supported(shard_conn):
def test_sqlite_fts_title_stopword_only_returns_empty(shard_conn):
"""All-stopword queries reduce to empty FTS5 — short-circuit."""
corpus = SqliteShardCorpus(shard_conn)
with pytest.raises(NotSupportedError):
corpus.fts_phrase(["anarchism is a"])
assert corpus.fts_title("the of and") == []
def test_sidecar_fts_title_not_supported():
def test_sqlite_fts_phrase_matches_verbatim(shard_conn):
"""FTS5 phrase MATCH '"phrase"' returns chunks containing the
verbatim phrase. Use a phrase known to appear in the fixture doc."""
corpus = SqliteShardCorpus(shard_conn)
# The Anarchism fixture has "political philosophy" verbatim.
hits = corpus.fts_phrase(["political philosophy"], limit=5)
assert hits, "expected at least one hit for verbatim phrase"
assert any("Anarchism" in h.title for h in hits)
# Phrase tokens that DON'T appear together should return nothing.
misses = corpus.fts_phrase(["stateless capital"], limit=5)
assert misses == []
def test_sqlite_fts_phrase_empty_input_returns_empty(shard_conn):
corpus = SqliteShardCorpus(shard_conn)
assert corpus.fts_phrase([]) == []
assert corpus.fts_phrase(["", " "]) == []
def test_sidecar_fts_title_delegates_to_multishard():
"""SidecarBucketCorpus.fts_title delegates to the multi-shard
fts_title_search method. Stub the multi-shard and confirm shape."""
class _StubMulti:
manifest = type("M", (), {"shards": [], "snapshot_root": None})()
def fts_search(self, q, *, limit=8): return []
def fts_title_search(self, q, *, limit=8): return [
{"document_root": "abc", "document_uri": "test://x",
"title": "X", "score": -12.0, "_shard_url": "s0"},
]
def fts_phrase_search(self, ngs, *, limit=8): return []
def conn_for_shard(self, url): raise KeyError(url)
corpus = SidecarBucketCorpus(_StubMulti())
with pytest.raises(NotSupportedError):
corpus.fts_title("anarchism")
hits = corpus.fts_title("anything")
assert len(hits) == 1
assert hits[0].title == "X"
assert hits[0].shard_id == "s0"
def test_sidecar_fts_phrase_not_supported():
def test_sidecar_fts_phrase_delegates_to_multishard():
class _StubMulti:
manifest = type("M", (), {"shards": [], "snapshot_root": None})()
def fts_search(self, q, *, limit=8): return []
def fts_title_search(self, q, *, limit=8): return []
def fts_phrase_search(self, ngs, *, limit=8): return [
{"document_root": "abc", "document_uri": "test://x",
"title": "X", "score": -23.5},
]
def conn_for_shard(self, url): raise KeyError(url)
corpus = SidecarBucketCorpus(_StubMulti())
with pytest.raises(NotSupportedError):
corpus.fts_phrase(["a b c"])
hits = corpus.fts_phrase(["any phrase"])
assert len(hits) == 1
assert hits[0].score == -23.5
# ---------------------------------------------------------------------------