From 18db3c2caf4bf55a19e090dead74be2440688cf7 Mon Sep 17 00:00:00 2001 From: "russell@unturf.com" Date: Sun, 31 May 2026 11:21:57 -0400 Subject: [PATCH] qa/corpus: wire fts_title (documents_fts) + fts_phrase (FTS5 phrase MATCH) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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) --- arborist/ingest.py | 13 ++++ arborist/qa/corpus.py | 134 +++++++++++++++++++++++++++------ arborist/wallet/bucket.py | 151 ++++++++++++++++++++++++++++++++++++-- tests/test_qa_corpus.py | 62 ++++++++++++---- 4 files changed, 317 insertions(+), 43 deletions(-) diff --git a/arborist/ingest.py b/arborist/ingest.py index b2e14f4..0f248a8 100644 --- a/arborist/ingest.py +++ b/arborist/ingest.py @@ -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) " diff --git a/arborist/qa/corpus.py b/arborist/qa/corpus.py index 16563ea..26e3d61 100644 --- a/arborist/qa/corpus.py +++ b/arborist/qa/corpus.py @@ -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 diff --git a/arborist/wallet/bucket.py b/arborist/wallet/bucket.py index 8b67097..c93d63c 100644 --- a/arborist/wallet/bucket.py +++ b/arborist/wallet/bucket.py @@ -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 diff --git a/tests/test_qa_corpus.py b/tests/test_qa_corpus.py index 7eb8065..1490542 100644 --- a/tests/test_qa_corpus.py +++ b/tests/test_qa_corpus.py @@ -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 # ---------------------------------------------------------------------------