From a0d31ec366e0df43e8362f151464724c721e2d5e Mon Sep 17 00:00:00 2001 From: phernandez Date: Tue, 15 Sep 2026 12:06:44 -0500 Subject: [PATCH] refactor(core): move FTS execution behind FtsBackend and bind the read path to scope Second step of #1558. The project-bound repositories no longer own full-text execution, and every statement the shared read path issues binds the repository's ProjectScope instead of assuming a single :project_id. No query behavior changes. - search_filters.FtsBackend: the narrow contract for running a compiled full-text statement on one engine. SQLiteFts and PostgresFts implement it in the backend query modules, carrying the bodies that used to live in each repository's search()/count(): FTS5 syntax errors answer empty; Postgres retries a malformed strict tsquery relaxed inside a savepoint. - SearchRepositoryBase.search and count are concrete: shared vector/hybrid dispatch, then self._fts. The two repositories lose their search(), count(), pass-through _run_vector_query overrides, and the SQLite entity-column cache (now inside SQLiteFts). - Manifest hydration, candidate row fetch, readiness, and hydration-drop classification take the scope. current_vector_manifest_predicate replaces the :project_id constant; project_readiness uses it for its one project. - PreparedSearchQuery moves to repository.search_query with defaults so the repository layer can build and pass it. compile_fts_filter takes it. - The filter helpers both backends share, and the SearchIndexKey alias, move out of the base class into search_filters and search_index_row. - Test fakes that bypass __init__ set scope beside project_id, and their search() overrides carry the session parameter the base method has. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_019YW9ysxugGGBCNEGzsxtFV Signed-off-by: phernandez --- CHANGELOG.md | 10 + .../repository/postgres_search_query.py | 276 ++++++++++++--- .../repository/postgres_search_repository.py | 329 +----------------- src/basic_memory/repository/search_filters.py | 230 ++++++++++-- .../repository/search_index_row.py | 6 + src/basic_memory/repository/search_query.py | 30 ++ .../repository/search_repository_base.py | 311 ++++++----------- src/basic_memory/repository/search_trace.py | 28 +- .../repository/sqlite_search_query.py | 280 ++++++++++++--- .../repository/sqlite_search_repository.py | 326 +---------------- .../services/project_readiness.py | 25 +- src/basic_memory/services/search_service.py | 23 +- tests/repository/test_hybrid_fusion.py | 4 + .../test_postgres_search_quoted_queries.py | 8 +- .../test_postgres_search_repository.py | 17 +- .../test_search_file_path_prefix.py | 2 +- tests/repository/test_search_trace.py | 9 +- tests/repository/test_semantic_search_base.py | 4 + tests/repository/test_semantic_vector_sync.py | 4 + ...est_vector_filter_candidate_restriction.py | 2 +- tests/repository/test_vector_pagination.py | 4 + tests/repository/test_vector_threshold.py | 4 + 22 files changed, 892 insertions(+), 1040 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4262bb491..a523a2777 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -121,6 +121,16 @@ `(project_id, ...)` identity. Project repositories call the compilers with a scope of one; no query behavior changes. First step of the shared single/multi-project reader. +- **#1558**: Full-text execution leaves the project repositories. `SQLiteFts` and + `PostgresFts` run a compiled statement for any `ProjectScope` and own their engine's + failure semantics (FTS5 syntax errors answer empty, Postgres retries a malformed strict + tsquery relaxed inside a savepoint). `SearchRepositoryBase.search` and `count` are + concrete: shared vector/hybrid dispatch, then the engine's `FtsBackend`. The base read + path binds every statement to the repository's scope (manifest hydration, candidate row + fetch, readiness and drop classification). `PreparedSearchQuery` moves to the repository + layer with defaults, and the filter helpers both backends share move from the base into + `search_filters`. No query behavior changes. + ## v0.23.2 (2026-08-25) diff --git a/src/basic_memory/repository/postgres_search_query.py b/src/basic_memory/repository/postgres_search_query.py index ea30070b7..0fa2fc339 100644 --- a/src/basic_memory/repository/postgres_search_query.py +++ b/src/basic_memory/repository/postgres_search_query.py @@ -1,32 +1,40 @@ -"""PostgreSQL tsquery preparation: term syntax and filter compilation. +"""PostgreSQL tsquery preparation and execution. -Pure functions over a ``ProjectScope`` and the caller's filters. Nothing here opens -a session or owns an index. +Term preparation and filter compilation are pure functions over a ``ProjectScope`` and +a ``PreparedSearchQuery``. ``PostgresFts`` runs the compiled statement and owns +tsquery's failure semantics. Nothing here initializes or mutates an index. """ import json import re +import time from collections.abc import Sequence -from datetime import datetime from typing import Any +import logfire +from loguru import logger +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from basic_memory import db from basic_memory.repository.metadata_filters import parse_metadata_filters from basic_memory.repository.script_ngrams import analyze_script_query from basic_memory.repository.search_filters import ( AFTER_DATE_ORDER_BY, POSTGRES_FILTER_DIALECT, CompiledFilter, - shared_filter_conditions, -) -from basic_memory.repository.search_query import relaxation_word_tokens, relaxed_query_words -from basic_memory.repository.search_repository_base import ( - SearchIndexKey, metadata_contains_like_condition, metadata_filter_content_type_condition, + shared_filter_conditions, +) +from basic_memory.repository.search_index_row import SearchIndexKey, SearchIndexRow +from basic_memory.repository.search_query import ( + PreparedSearchQuery, + relaxation_word_tokens, + relaxed_query_words, ) from basic_memory.repository.search_scope import ProjectScope -from basic_memory.schemas.search import SearchItemType -from basic_memory.temporal import TemporalFilter +from basic_memory.repository.search_trace import SearchTraceCollector, build_fts_page_stage _TSQUERY_OPERAND_PATTERN = re.compile(r"'(?:''|[^'])*'(?::\*)?|[^\s&|!()]+") _TSQUERY_WORD_PATTERN = re.compile(r"[^\W_]+(?:'[^\W_]+)?", re.UNICODE) @@ -36,6 +44,24 @@ # tsquery special characters that must not reach the parser as text. _TSQUERY_SPECIAL_CHARS = ("&", "|", "!", "(", ")", ":") +# Every FTS statement returns these columns plus a score. +_RESULT_COLUMNS = """ + search_index.project_id, + search_index.id, + search_index.title, + search_index.permalink, + search_index.file_path, + search_index.type, + search_index.metadata, + search_index.from_id, + search_index.to_id, + search_index.relation_type, + search_index.entity_id, + search_index.content_snippet, + search_index.category, + search_index.created_at, + search_index.updated_at""" + # --- Term preparation --- @@ -436,18 +462,8 @@ def _script_candidate_from_clause(scope: ProjectScope, params: dict[str, Any]) - def compile_fts_filter( scope: ProjectScope, + query: PreparedSearchQuery, *, - search_text: str | None = None, - permalink: str | None = None, - permalink_match: str | None = None, - title: str | None = None, - note_types: Sequence[str] | None = None, - after_date: datetime | None = None, - search_item_types: Sequence[SearchItemType] | None = None, - categories: Sequence[str] | None = None, - metadata_filters: dict[str, Any] | None = None, - file_path_prefix: str | None = None, - temporal: TemporalFilter | None = None, allow_relaxed: bool = False, candidate_keys: Sequence[SearchIndexKey] | None = None, ) -> CompiledFilter: @@ -458,21 +474,12 @@ def compile_fts_filter( """ params: dict[str, Any] = {} conditions = shared_filter_conditions( - scope, - params, - dialect=POSTGRES_FILTER_DIALECT, - permalink=permalink, - file_path_prefix=file_path_prefix, - candidate_keys=candidate_keys, - search_item_types=search_item_types, - categories=categories, - note_types=note_types, - after_date=after_date, - temporal=temporal, + scope, params, dialect=POSTGRES_FILTER_DIALECT, query=query, candidate_keys=candidate_keys ) from_clause = "search_index" document_vector: str | None = None script_tsqueries: list[str] = [] + search_text = query.search_text # Wildcard-only and blank text add no text condition: every row matches. if search_text and search_text.strip() not in ("", "*"): @@ -525,15 +532,15 @@ def compile_fts_filter( for index in range(len(script_tsqueries)) ) - if title: - params["title_text"] = prepare_search_term(title.strip(), is_prefix=False) + if query.title: + params["title_text"] = prepare_search_term(query.title.strip(), is_prefix=False) conditions.append( "to_tsvector('english', search_index.title) @@ to_tsquery('english', :title_text)" ) - if permalink_match: - permalink_text = permalink_match.lower().strip() - if "*" in permalink_match: + if query.permalink_match: + permalink_text = query.permalink_match.lower().strip() + if "*" in query.permalink_match: # ``*`` becomes the LIKE wildcard. params["permalink"] = permalink_text.replace("*", "%") conditions.append("search_index.permalink LIKE :permalink") @@ -543,8 +550,8 @@ def compile_fts_filter( # Structured metadata filters use jsonb_extract_path_text() / jsonb_extract_path() # with parameterized path parts instead of #>> / #> with interpolated paths. - if metadata_filters: - parsed_filters = parse_metadata_filters(metadata_filters) + if query.metadata_filters: + parsed_filters = parse_metadata_filters(query.metadata_filters) from_clause = f"{from_clause} JOIN entity ON search_index.entity_id = entity.id" # Frontmatter filters answer for notes only; see # metadata_filter_content_type_condition for why every regular file would @@ -662,6 +669,193 @@ def compile_fts_filter( from_clause=from_clause, where_clause=" AND ".join(conditions), params=params, - order_by_clause=AFTER_DATE_ORDER_BY if after_date else "", + order_by_clause=AFTER_DATE_ORDER_BY if query.after_date else "", score_expression=score_expression, ) + + +# --- Execution --- + + +class PostgresFts: + """Run tsquery statements for any scope in one database.""" + + def __init__(self, session_maker: async_sessionmaker[AsyncSession]) -> None: + self._session_maker = session_maker + + async def search( + self, + scope: ProjectScope, + query: PreparedSearchQuery, + *, + limit: int, + offset: int, + allow_relaxed: bool = False, + session: AsyncSession | None = None, + candidate_keys: Sequence[SearchIndexKey] | None = None, + trace: SearchTraceCollector | None = None, + ) -> list[SearchIndexRow]: + """Run one tsquery page, retrying a malformed or empty strict query relaxed.""" + search_text = query.search_text + compiled = compile_fts_filter( + scope, query, allow_relaxed=allow_relaxed, candidate_keys=candidate_keys + ) + params = compiled.params + params["limit"] = limit + params["offset"] = offset + + sql = f""" + SELECT{_RESULT_COLUMNS}, + {compiled.score_expression} as score + FROM {compiled.from_clause} + WHERE {compiled.where_clause} + ORDER BY score DESC {compiled.order_by_clause}, search_index.id ASC + LIMIT :limit + OFFSET :offset + """ + + logger.trace(f"Search {sql} params: {params}") + fts_started_at = time.perf_counter() if trace is not None else None + + use_savepoint = session is not None or allow_relaxed + + async def execute_rows(active_session: AsyncSession, query_params: dict[str, Any]): + # PostgreSQL leaves a transaction unusable after invalid tsquery syntax. + # Scope retryable or caller-owned attempts to a savepoint so a relaxed + # retry, and any caller continuing to use its session, starts healthy. + if use_savepoint: + async with active_session.begin_nested(): + result = await active_session.execute(text(sql), query_params) + return result.fetchall() + result = await active_session.execute(text(sql), query_params) + return result.fetchall() + + async def run_search(active_session: AsyncSession): + relaxed = relaxed_tsquery_text(search_text) if allow_relaxed else None + strict_syntax_error = False + relaxed_fallback_used = False + try: + rows = await execute_rows(active_session, params) + except Exception as exc: + if not (is_tsquery_syntax_error(exc) and relaxed and params.get("text")): + raise + strict_syntax_error = True + rows = [] + + # Trigger: multi-word natural-language query matched nothing under the + # default all-terms-AND tsquery semantics, or its punctuation produced + # invalid strict tsquery syntax. + # Why: questions rarely have every word in one document; without + # relaxation the FTS half of hybrid search contributes zero candidates. + # The relaxed renderer also tokenizes punctuation safely. + # Outcome: one retry with OR-joined prefix lexemes; ts_rank still ranks + # multi-term matches first. + if relaxed and not rows and params.get("text"): + relaxed_fallback_used = True + retry_reason = "invalid syntax" if strict_syntax_error else "0 results" + logger.debug( + f"Strict Postgres FTS returned {retry_reason}; retrying relaxed FTS query " + f"strict='{search_text}' relaxed='{relaxed}'" + ) + with logfire.span( + "search.relaxed_fts_retry", + backend="postgres", + reason="syntax_error" if strict_syntax_error else "empty_result", + token_count=len(relaxed_query_words(search_text) or ()), + limit=limit, + offset=offset, + ): + rows = await execute_rows(active_session, {**params, "text": relaxed}) + return rows, relaxed_fallback_used + + try: + if session is not None: + rows, relaxed_fallback_used = await run_search(session) + else: + async with db.scoped_session(self._session_maker) as owned_session: + rows, relaxed_fallback_used = await run_search(owned_session) + except Exception as e: + if is_tsquery_syntax_error(e): + logger.warning(f"tsquery syntax error for search term: {search_text}, error: {e}") + if trace is not None: + trace.fts = build_fts_page_stage( + [], + relaxed_fallback_used=False, + fts_ms=( + (time.perf_counter() - fts_started_at) * 1000 + if fts_started_at is not None + else None + ), + ) + return [] + logger.error(f"Database error during search: {e}") + raise + + results = [SearchIndexRow.from_mapping(row._asdict()) for row in rows] + if trace is not None: + trace.fts = build_fts_page_stage( + [((row.type, row.id), row.score or 0.0) for row in results], + relaxed_fallback_used=relaxed_fallback_used, + fts_ms=( + (time.perf_counter() - fts_started_at) * 1000 + if fts_started_at is not None + else None + ), + ) + + logger.trace(f"Found {len(results)} search results") + for r in results: + logger.trace( + f"Search result: project_id: {r.project_id} type:{r.type} title: {r.title} permalink: {r.permalink} score: {r.score}" + ) + return results + + async def count( + self, + scope: ProjectScope, + query: PreparedSearchQuery, + *, + allow_relaxed: bool = False, + ) -> int: + """Count rows matching the tsquery, with the same relaxed retry as search.""" + search_text = query.search_text + compiled = compile_fts_filter(scope, query, allow_relaxed=allow_relaxed) + params = compiled.params + sql = f"SELECT COUNT(*) FROM {compiled.from_clause} WHERE {compiled.where_clause}" + logger.trace(f"Count {sql} params: {params}") + + async def execute_count(active_session: AsyncSession, query_params: dict[str, Any]) -> int: + if allow_relaxed: + async with active_session.begin_nested(): + result = await active_session.execute(text(sql), query_params) + return int(result.scalar_one()) + result = await active_session.execute(text(sql), query_params) + return int(result.scalar_one()) + + try: + async with db.scoped_session(self._session_maker) as session: + relaxed = relaxed_tsquery_text(search_text) if allow_relaxed else None + strict_syntax_error = False + try: + total = await execute_count(session, params) + except Exception as exc: + if not (is_tsquery_syntax_error(exc) and relaxed and params.get("text")): + raise + strict_syntax_error = True + total = 0 + + if relaxed and total == 0 and params.get("text"): + with logfire.span( + "search.count.relaxed_fts_retry", + backend="postgres", + reason="syntax_error" if strict_syntax_error else "empty_result", + token_count=len(relaxed_query_words(search_text) or ()), + ): + total = await execute_count(session, {**params, "text": relaxed}) + return total + except Exception as e: + if is_tsquery_syntax_error(e): + logger.warning(f"tsquery syntax error for search term: {search_text}, error: {e}") + return 0 + logger.error(f"Database error during search count: {e}") + raise diff --git a/src/basic_memory/repository/postgres_search_repository.py b/src/basic_memory/repository/postgres_search_repository.py index fc89e18da..0cb17fcb0 100644 --- a/src/basic_memory/repository/postgres_search_repository.py +++ b/src/basic_memory/repository/postgres_search_repository.py @@ -2,12 +2,9 @@ import asyncio import json -import time from collections.abc import Sequence -from datetime import datetime -from typing import Any, override, List, Optional +from typing import Any, override, List -import logfire from loguru import logger from sqlalchemy import text from sqlalchemy.ext.asyncio import AsyncSession @@ -20,23 +17,13 @@ from basic_memory.repository.rerank_provider import RerankProvider from basic_memory.repository.rerank_provider_factory import create_rerank_provider from basic_memory.repository.search_index_row import SearchIndexRow -from basic_memory.repository.search_query import relaxed_query_words from basic_memory.repository.script_ngrams import build_script_ngrams from basic_memory.repository.semantic_chunking import VectorChunkRecord from basic_memory.repository.search_repository_base import ( - SearchIndexKey, SearchRepositoryBase, VectorChunkState, ) -from basic_memory.repository.search_trace import ( - SearchTraceCollector, - build_fts_page_stage, -) -from basic_memory.repository.postgres_search_query import ( - compile_fts_filter, - is_tsquery_syntax_error, - relaxed_tsquery_text, -) +from basic_memory.repository.postgres_search_query import PostgresFts from basic_memory.repository.semantic_errors import SemanticDependenciesMissingError from basic_memory.repository.semantic_vector_index import SemanticVectorIndex from basic_memory.repository.semantic_vector_sync import ( @@ -49,8 +36,6 @@ ) from basic_memory.repository.pgvector_index import PgVectorIndex from basic_memory.repository.postgres_fts_chunks import split_postgres_fts_chunks -from basic_memory.schemas.search import SearchItemType, SearchRetrievalMode -from basic_memory.temporal import TemporalFilter def _strip_nul_from_row(row_data: dict[str, Any]) -> dict[str, Any]: @@ -89,6 +74,7 @@ def __init__( rerank_provider: RerankProvider | None = None, ): super().__init__(session_maker, project_id) + self._fts = PostgresFts(session_maker) self._app_config = app_config or ConfigManager().config self._semantic_enabled = self._app_config.semantic_search_enabled self._semantic_vector_k = self._app_config.semantic_vector_k @@ -386,22 +372,6 @@ async def _ensure_vector_tables(self) -> None: logger.debug(f"Postgres vector tables ready (dimensions={self._vector_dimensions})") self._vector_tables_initialized = True - @override - async def _run_vector_query( - self, - session: AsyncSession, - query_embedding: list[float], - candidate_limit: int, - *, - trace: SearchTraceCollector | None = None, - ) -> list[dict[str, Any]]: - return await super()._run_vector_query( - session, - query_embedding, - candidate_limit, - trace=trace, - ) - @override def _vector_prepare_window_size(self) -> int: """Use a bounded config-driven prepare window for Postgres vector sync.""" @@ -611,296 +581,3 @@ async def bulk_index_items(self, search_index_rows: List[SearchIndexRow]) -> Non await self._replace_fts_chunks(session, search_index_rows) logger.debug(f"Bulk indexed {len(search_index_rows)} rows") await session.commit() - - # ------------------------------------------------------------------ - # FTS search (Postgres-specific) - # ------------------------------------------------------------------ - - @override - async def search( - self, - search_text: Optional[str] = None, - permalink: Optional[str] = None, - permalink_match: Optional[str] = None, - title: Optional[str] = None, - note_types: Optional[List[str]] = None, - after_date: Optional[datetime] = None, - search_item_types: Optional[List[SearchItemType]] = None, - categories: Optional[List[str]] = None, - metadata_filters: Optional[dict[str, Any]] = None, - file_path_prefix: Optional[str] = None, - temporal: Optional[TemporalFilter] = None, - retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS, - min_similarity: Optional[float] = None, - limit: int = 10, - offset: int = 0, - allow_relaxed: bool = False, - session: AsyncSession | None = None, - *, - candidate_keys: Sequence[SearchIndexKey] | None = None, - trace: SearchTraceCollector | None = None, - ) -> List[SearchIndexRow]: - """Search across all indexed content using PostgreSQL tsvector.""" - # --- Dispatch vector / hybrid modes (shared logic) --- - dispatched = await self._dispatch_retrieval_mode( - search_text=search_text, - permalink=permalink, - permalink_match=permalink_match, - title=title, - note_types=note_types, - after_date=after_date, - search_item_types=search_item_types, - categories=categories, - metadata_filters=metadata_filters, - file_path_prefix=file_path_prefix, - temporal=temporal, - retrieval_mode=retrieval_mode, - min_similarity=min_similarity, - limit=limit, - offset=offset, - trace=trace, - ) - if dispatched is not None: - return dispatched - - # --- FTS mode (Postgres-specific) --- - compiled = compile_fts_filter( - self.scope, - search_text=search_text, - permalink=permalink, - permalink_match=permalink_match, - title=title, - note_types=note_types, - after_date=after_date, - search_item_types=search_item_types, - categories=categories, - metadata_filters=metadata_filters, - file_path_prefix=file_path_prefix, - temporal=temporal, - allow_relaxed=allow_relaxed, - candidate_keys=candidate_keys, - ) - params = compiled.params - params["limit"] = limit - params["offset"] = offset - - sql = f""" - SELECT - search_index.project_id, - search_index.id, - search_index.title, - search_index.permalink, - search_index.file_path, - search_index.type, - search_index.metadata, - search_index.from_id, - search_index.to_id, - search_index.relation_type, - search_index.entity_id, - search_index.content_snippet, - search_index.category, - search_index.created_at, - search_index.updated_at, - {compiled.score_expression} as score - FROM {compiled.from_clause} - WHERE {compiled.where_clause} - ORDER BY score DESC {compiled.order_by_clause}, search_index.id ASC - LIMIT :limit - OFFSET :offset - """ - - logger.trace(f"Search {sql} params: {params}") - fts_started_at = time.perf_counter() if trace is not None else None - - use_savepoint = session is not None or allow_relaxed - - async def execute_rows(active_session: AsyncSession, query_params: dict[str, Any]): - # PostgreSQL leaves a transaction unusable after invalid tsquery syntax. - # Scope retryable or caller-owned attempts to a savepoint so a relaxed - # retry—and any caller continuing to use its session—starts healthy. - if use_savepoint: - async with active_session.begin_nested(): - result = await active_session.execute(text(sql), query_params) - return result.fetchall() - result = await active_session.execute(text(sql), query_params) - return result.fetchall() - - async def run_search(active_session: AsyncSession): - relaxed = relaxed_tsquery_text(search_text) if allow_relaxed else None - strict_syntax_error = False - relaxed_fallback_used = False - try: - rows = await execute_rows(active_session, params) - except Exception as exc: - if not (is_tsquery_syntax_error(exc) and relaxed and params.get("text")): - raise - strict_syntax_error = True - rows = [] - - # Trigger: multi-word natural-language query matched nothing - # under the default all-terms-AND tsquery semantics, or its punctuation - # produced invalid strict tsquery syntax. - # Why: questions rarely have every word in one document; - # without relaxation the FTS half of hybrid search contributes zero - # candidates. The relaxed renderer also tokenizes punctuation safely. - # Outcome: one retry with OR-joined prefix lexemes; ts_rank - # still ranks multi-term matches first. - if relaxed and not rows and params.get("text"): - relaxed_fallback_used = True - retry_reason = "invalid syntax" if strict_syntax_error else "0 results" - logger.debug( - f"Strict Postgres FTS returned {retry_reason}; retrying relaxed FTS query " - f"strict='{search_text}' relaxed='{relaxed}'" - ) - with logfire.span( - "search.relaxed_fts_retry", - backend="postgres", - reason="syntax_error" if strict_syntax_error else "empty_result", - token_count=len(relaxed_query_words(search_text) or ()), - limit=limit, - offset=offset, - ): - rows = await execute_rows( - active_session, - {**params, "text": relaxed}, - ) - return rows, relaxed_fallback_used - - try: - if session is not None: - rows, relaxed_fallback_used = await run_search(session) - else: - async with db.scoped_session(self.session_maker) as owned_session: - rows, relaxed_fallback_used = await run_search(owned_session) - except Exception as e: - if is_tsquery_syntax_error(e): - logger.warning(f"tsquery syntax error for search term: {search_text}, error: {e}") - if trace is not None: - trace.fts = build_fts_page_stage( - [], - relaxed_fallback_used=False, - fts_ms=( - (time.perf_counter() - fts_started_at) * 1000 - if fts_started_at is not None - else None - ), - ) - return [] - - # Re-raise other database errors - logger.error(f"Database error during search: {e}") - raise - - results = [SearchIndexRow.from_mapping(row._asdict()) for row in rows] - if trace is not None: - trace.fts = build_fts_page_stage( - [((row.type, row.id), row.score or 0.0) for row in results], - relaxed_fallback_used=relaxed_fallback_used, - fts_ms=( - (time.perf_counter() - fts_started_at) * 1000 - if fts_started_at is not None - else None - ), - ) - - logger.trace(f"Found {len(results)} search results") - for r in results: - logger.trace( - f"Search result: project_id: {r.project_id} type:{r.type} title: {r.title} permalink: {r.permalink} score: {r.score}" - ) - - return results - - @override - async def count( - self, - search_text: Optional[str] = None, - permalink: Optional[str] = None, - permalink_match: Optional[str] = None, - title: Optional[str] = None, - note_types: Optional[List[str]] = None, - after_date: Optional[datetime] = None, - search_item_types: Optional[List[SearchItemType]] = None, - categories: Optional[List[str]] = None, - metadata_filters: Optional[dict[str, Any]] = None, - file_path_prefix: Optional[str] = None, - temporal: Optional[TemporalFilter] = None, - retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS, - min_similarity: Optional[float] = None, - allow_relaxed: bool = False, - ) -> int: - """Count indexed content matching the Postgres FTS query.""" - if retrieval_mode != SearchRetrievalMode.FTS: - return await super().count( - search_text=search_text, - permalink=permalink, - permalink_match=permalink_match, - title=title, - note_types=note_types, - after_date=after_date, - search_item_types=search_item_types, - categories=categories, - metadata_filters=metadata_filters, - file_path_prefix=file_path_prefix, - temporal=temporal, - retrieval_mode=retrieval_mode, - min_similarity=min_similarity, - ) - - compiled = compile_fts_filter( - self.scope, - search_text=search_text, - permalink=permalink, - permalink_match=permalink_match, - title=title, - note_types=note_types, - after_date=after_date, - search_item_types=search_item_types, - categories=categories, - metadata_filters=metadata_filters, - file_path_prefix=file_path_prefix, - temporal=temporal, - allow_relaxed=allow_relaxed, - ) - params = compiled.params - sql = f"SELECT COUNT(*) FROM {compiled.from_clause} WHERE {compiled.where_clause}" - logger.trace(f"Count {sql} params: {params}") - - async def execute_count(active_session: AsyncSession, query_params: dict[str, Any]) -> int: - if allow_relaxed: - async with active_session.begin_nested(): - result = await active_session.execute(text(sql), query_params) - return int(result.scalar_one()) - result = await active_session.execute(text(sql), query_params) - return int(result.scalar_one()) - - try: - async with db.scoped_session(self.session_maker) as session: - relaxed = relaxed_tsquery_text(search_text) if allow_relaxed else None - strict_syntax_error = False - try: - total = await execute_count(session, params) - except Exception as exc: - if not (is_tsquery_syntax_error(exc) and relaxed and params.get("text")): - raise - strict_syntax_error = True - total = 0 - - if relaxed and total == 0 and params.get("text"): - with logfire.span( - "search.count.relaxed_fts_retry", - backend="postgres", - reason="syntax_error" if strict_syntax_error else "empty_result", - token_count=len(relaxed_query_words(search_text) or ()), - ): - total = await execute_count( - session, - {**params, "text": relaxed}, - ) - return total - except Exception as e: - if is_tsquery_syntax_error(e): - logger.warning(f"tsquery syntax error for search term: {search_text}, error: {e}") - return 0 - logger.error(f"Database error during search count: {e}") - raise diff --git a/src/basic_memory/repository/search_filters.py b/src/basic_memory/repository/search_filters.py index 0a3233bca..2431d11cb 100644 --- a/src/basic_memory/repository/search_filters.py +++ b/src/basic_memory/repository/search_filters.py @@ -1,35 +1,40 @@ -"""WHERE-clause pieces both search backends share. +"""WHERE-clause pieces both search backends share, and the FTS execution contract. The two FTS engines differ in how they match text and read JSON. Every other filter a search accepts asks the same question of the same columns on both, so it is compiled once here. A backend supplies the two spellings that differ through ``FilterDialect`` and appends its own text, title, permalink-pattern, and metadata predicates around the -shared ones. +shared ones. ``FtsBackend`` is the narrow contract through which the shared read path +runs a compiled full-text statement on one engine. """ from collections.abc import Sequence from dataclasses import dataclass -from datetime import datetime -from typing import Any +from typing import Any, Protocol + +from sqlalchemy.ext.asyncio import AsyncSession from basic_memory.repository.note_type_filters import ( POSTGRES_NOTE_TYPE_VALUE, SQLITE_NOTE_TYPE_VALUE, build_note_type_predicate, ) -from basic_memory.repository.search_repository_base import ( - SearchIndexKey, - candidate_key_restriction_condition, - file_path_prefix_condition, -) +from basic_memory.repository.search_index_row import SearchIndexKey, SearchIndexRow +from basic_memory.repository.search_query import PreparedSearchQuery from basic_memory.repository.search_scope import ProjectScope +from basic_memory.repository.search_trace import SearchTraceCollector from basic_memory.repository.temporal_filters import build_temporal_predicate -from basic_memory.schemas.search import SearchItemType -from basic_memory.temporal import TemporalFilter +from basic_memory.runtime.storage import RUNTIME_MARKDOWN_CONTENT_TYPE +from basic_memory.schemas.search import normalize_file_path_prefix # Newest edits first whenever the caller filtered on ``after_date``. AFTER_DATE_ORDER_BY = ", search_index.updated_at DESC" +# SQLite's LIKE has no default escape character, and Postgres's is already the +# backslash, so naming this one explicitly in every pattern is what lets a single +# escaped pattern mean the same thing on both backends. +_LIKE_ESCAPE_CHARACTER = "\\" + @dataclass(frozen=True, slots=True) class FilterDialect: @@ -65,19 +70,179 @@ class CompiledFilter: score_expression: str +class FtsBackend(Protocol): + """Run one engine's full-text statement for a scope and a prepared query. + + Both implementations compile through their own ``compile_fts_filter``, execute, and + own the engine's failure semantics: FTS5 syntax errors answer with no rows, while + Postgres retries a malformed strict tsquery through the relaxed renderer inside a + savepoint so the caller's transaction survives. + """ + + async def search( + self, + scope: ProjectScope, + query: PreparedSearchQuery, + *, + limit: int, + offset: int, + allow_relaxed: bool = False, + session: AsyncSession | None = None, + candidate_keys: Sequence[SearchIndexKey] | None = None, + trace: SearchTraceCollector | None = None, + ) -> list[SearchIndexRow]: ... + + async def count( + self, + scope: ProjectScope, + query: PreparedSearchQuery, + *, + allow_relaxed: bool = False, + ) -> int: ... + + +# --- Filters both backends compile identically --- + + +def file_path_prefix_condition( + file_path_prefix: str | None, + params: dict[str, Any], +) -> str | None: + """Build the SQL scoping search rows to one directory subtree of the project. + + One implementation, shared verbatim by both backends: a subtree scope that + means different things on SQLite and Postgres would report an exact total + for a match set the other dialect never produces. + + Boundary: the compared prefix carries its trailing separator, so "specs" + admits "specs/api.md" and never "specs-archive/api.md". + + Why an explicit-length comparison rather than ``file_path LIKE 'specs/%'``: + LIKE reads "_" and "%" as wildcards and both are ordinary characters in a + directory name, so "my_notes" would silently also admit "my-notes"; and + LIKE case-folds differently per backend, so one filter would answer two + different questions. SUBSTR equality has no pattern language to escape and + compares under each backend's deterministic default text collation, which is + byte equality on both, so the dialects match exactly the same rows. + """ + normalized = normalize_file_path_prefix(file_path_prefix) + if normalized is None: + return None + prefix = f"{normalized}/" + params["file_path_prefix"] = prefix + params["file_path_prefix_length"] = len(prefix) + return "SUBSTR(search_index.file_path, 1, :file_path_prefix_length) = :file_path_prefix" + + +def metadata_filter_content_type_condition(params: dict[str, Any]) -> str: + """Build the SQL restricting a metadata-filtered query to Markdown notes. + + Frontmatter is a Markdown-only construct, but every indexed file (PDF, image, + binary) gets its own ENTITY row whose ``entity_metadata`` carries no keys at all. + A positive predicate can never match one, so this constraint was invisible until + ``{"key": None}`` arrived: ``IS NULL`` is satisfied by the *absence* of a key, + which is exactly the state every regular file is in, and the whole non-note half + of a project counted into an exact total. + + Applied to any metadata filter, not just the null one, so the frontmatter-only + contract is a property of the clause rather than of which operator happened to + be used. + """ + params["metadata_filter_content_type"] = RUNTIME_MARKDOWN_CONTENT_TYPE + return "entity.content_type = :metadata_filter_content_type" + + +def metadata_contains_like_condition( + extract_expr: str, + value: Any, + *, + param_prefix: str, + params: dict[str, Any], +) -> str: + """Build the compatibility half of an array-contains metadata filter. + + The primary half of a ``{"tags": ["security"]}`` filter asks JSON whether the + array holds the element (``json_each`` on SQLite, ``@>`` on Postgres) and answers + only when the stored value really is a JSON array. Frontmatter written before + tags were normalized can hold the array's *text* instead, either JSON-quoted + ('["security", "auth"]') or as a Python repr ("['security', 'auth']"), and only a + substring match finds an element inside those. Hence a pattern per quote style. + + LIKE reads "%" and "_" in the searched-for value as wildcards, so interpolating + the value raw turned `tags has 100%` into a pattern that also matched + "100-percent". Escaping both wildcards and the escape character itself makes the + value literal again. + """ + escaped = ( + str(value) + .replace(_LIKE_ESCAPE_CHARACTER, _LIKE_ESCAPE_CHARACTER * 2) + .replace("%", f"{_LIKE_ESCAPE_CHARACTER}%") + .replace("_", f"{_LIKE_ESCAPE_CHARACTER}_") + ) + double_quoted_param = f"{param_prefix}_like" + single_quoted_param = f"{param_prefix}_like_single" + params[double_quoted_param] = f'%"{escaped}"%' + params[single_quoted_param] = f"%'{escaped}'%" + escape_clause = f" ESCAPE '{_LIKE_ESCAPE_CHARACTER}'" + return ( + f"{extract_expr} LIKE :{double_quoted_param}{escape_clause} " + f"OR {extract_expr} LIKE :{single_quoted_param}{escape_clause}" + ) + + +def candidate_key_restriction_condition( + candidate_keys: Sequence[SearchIndexKey], + params: dict[str, Any], +) -> str: + """Build the SQL restricting a filter query to an explicit set of search rows. + + This is what turns the vector/hybrid filter pass from "give me a page of everything + the filter admits" into "of *these* candidates, which does the filter admit". The + first question has an answer the size of the project and had to be capped, and every + candidate outside the cap was then read as disallowed (#1431). The second question's + answer is bounded by the candidate set itself, so no cap is needed and none of the + candidates can fall off the end. + + Keys are grouped by row type rather than emitted as one ``(type, id)`` pair per + branch: entity, observation, and relation ids come from independent sequences, so the + type is part of the identity, but a handful of type-scoped ``IN`` lists binds one + parameter per key instead of two and leaves the id list in the shape both planners + can drive an index from. PostgreSQL's ``search_index`` primary key is + ``(id, type, project_id)``. + + An empty candidate set is a real state, not a caller error (a vector search whose + every hit was already dropped), and it admits nothing, so it yields a false + predicate rather than the vacuous truth an empty ``OR`` would collapse to. + """ + ids_by_type: dict[str, list[int]] = {} + for row_type, row_id in candidate_keys: + ids_by_type.setdefault(row_type, []).append(row_id) + + branches: list[str] = [] + for type_index, (row_type, row_ids) in enumerate(ids_by_type.items()): + type_param = f"candidate_type_{type_index}" + params[type_param] = row_type + id_params: list[str] = [] + for id_index, row_id in enumerate(dict.fromkeys(row_ids)): + id_param = f"candidate_id_{type_index}_{id_index}" + params[id_param] = row_id + id_params.append(f":{id_param}") + branches.append( + f"(search_index.type = :{type_param} AND search_index.id IN ({', '.join(id_params)}))" + ) + + if not branches: + return "1 = 0" + return f"({' OR '.join(branches)})" + + def shared_filter_conditions( scope: ProjectScope, params: dict[str, Any], *, dialect: FilterDialect, - permalink: str | None, - file_path_prefix: str | None, + query: PreparedSearchQuery, candidate_keys: Sequence[SearchIndexKey] | None, - search_item_types: Sequence[SearchItemType] | None, - categories: Sequence[str] | None, - note_types: Sequence[str] | None, - after_date: datetime | None, - temporal: TemporalFilter | None, ) -> list[str]: """Compile the filters whose SQL is identical on both backends. @@ -87,23 +252,20 @@ def shared_filter_conditions( """ conditions = [scope.predicate("search_index.project_id", params)] - if permalink: - params["permalink"] = permalink + if query.permalink: + params["permalink"] = query.permalink conditions.append("search_index.permalink = :permalink") - # See file_path_prefix_condition for the subtree boundary and escaping rules. - subtree_condition = file_path_prefix_condition(file_path_prefix, params) + subtree_condition = file_path_prefix_condition(query.file_path_prefix, params) if subtree_condition is not None: conditions.append(subtree_condition) - # See candidate_key_restriction_condition for why the vector filter pass asks - # about its candidates rather than paging the filter's whole match set (#1431). if candidate_keys is not None: conditions.append(candidate_key_restriction_condition(candidate_keys, params)) - if search_item_types: + if query.search_item_types: type_placeholders: list[str] = [] - for index, item_type in enumerate(search_item_types): + for index, item_type in enumerate(query.search_item_types): name = f"search_type_{index}" params[name] = item_type.value type_placeholders.append(f":{name}") @@ -114,9 +276,9 @@ def shared_filter_conditions( # callers expect exact-category matching, not incidental text matches. # Outcome: only rows whose indexed category exactly equals a requested value # survive (entities/relations have NULL category and are excluded). - if categories: + if query.categories: category_placeholders: list[str] = [] - for index, category in enumerate(categories): + for index, category in enumerate(query.categories): name = f"category_{index}" params[name] = category category_placeholders.append(f":{name}") @@ -125,23 +287,23 @@ def shared_filter_conditions( # The note type belongs to the note, but only its entity row carries the # frontmatter, so the predicate resolves through the owning note. See # note_type_filters for why reading it off each row excluded every non-entity row. - if note_types: + if query.note_types: conditions.append( build_note_type_predicate( - note_types, params, scope=scope, note_type_value=dialect.note_type_value + query.note_types, params, scope=scope, note_type_value=dialect.note_type_value ) ) # Filter on updated_at so recently edited notes are included even when created_at # is old. The matching ORDER BY lives in AFTER_DATE_ORDER_BY. - if after_date: - params["after_date"] = after_date + if query.after_date: + params["after_date"] = query.after_date conditions.append(dialect.after_date_condition) # Authored valid time (SPEC-82) is independent of ``after_date``: that one is # bookkeeping about the file, this one is a claim about the world. See # temporal_filters for the overlap rule and why the subquery is non-correlated. - if temporal is not None: - conditions.append(build_temporal_predicate(temporal, params, scope=scope)) + if query.temporal is not None: + conditions.append(build_temporal_predicate(query.temporal, params, scope=scope)) return conditions diff --git a/src/basic_memory/repository/search_index_row.py b/src/basic_memory/repository/search_index_row.py index ce8dc8d55..e9835b6e0 100644 --- a/src/basic_memory/repository/search_index_row.py +++ b/src/basic_memory/repository/search_index_row.py @@ -156,3 +156,9 @@ def to_insert(self, serialize_json: bool = True): "updated_at": self.updated_at if self.updated_at else None, "project_id": self.project_id, } + + +# Entity, observation, and relation rows carry ids from independent auto-increment +# sequences, so a bare id is ambiguous across row types. Every map in the retrieval +# path keys rows by (type, id) to avoid collisions. +type SearchIndexKey = tuple[str, int] diff --git a/src/basic_memory/repository/search_query.py b/src/basic_memory/repository/search_query.py index d73d70c0c..9cd2cb2b4 100644 --- a/src/basic_memory/repository/search_query.py +++ b/src/basic_memory/repository/search_query.py @@ -2,6 +2,36 @@ import re import unicodedata +from dataclasses import dataclass +from datetime import datetime +from typing import Any + +from basic_memory.schemas.search import SearchItemType, SearchRetrievalMode +from basic_memory.temporal import TemporalFilter + + +@dataclass(frozen=True) +class PreparedSearchQuery: + """Normalized query inputs shared by search and count. + + Built once at the service boundary from the API's ``SearchQuery``; every layer + below reads the same value instead of threading thirteen keyword arguments. + """ + + search_text: str | None = None + permalink: str | None = None + permalink_match: str | None = None + title: str | None = None + note_types: list[str] | None = None + search_item_types: list[SearchItemType] | None = None + categories: list[str] | None = None + after_date: datetime | None = None + metadata_filters: dict[str, Any] | None = None + file_path_prefix: str | None = None + temporal: TemporalFilter | None = None + retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS + min_similarity: float | None = None + # Interrogative/function words contribute lexical noise when a strict # full-text query is relaxed: "when OR did OR a" matches loud wrong documents diff --git a/src/basic_memory/repository/search_repository_base.py b/src/basic_memory/repository/search_repository_base.py index 99f8b23ae..31f97096a 100644 --- a/src/basic_memory/repository/search_repository_base.py +++ b/src/basic_memory/repository/search_repository_base.py @@ -32,7 +32,9 @@ demote_tail_scores, validate_rerank_scores, ) -from basic_memory.repository.search_index_row import SearchIndexRow +from basic_memory.repository.search_filters import FtsBackend +from basic_memory.repository.search_index_row import SearchIndexKey, SearchIndexRow +from basic_memory.repository.search_query import PreparedSearchQuery from basic_memory.repository.search_scope import ProjectScope from basic_memory.repository.script_ngrams import build_script_ngrams from basic_memory.repository.search_trace import ( @@ -78,12 +80,10 @@ StagedVectorDeletion as _StagedVectorDeletion, VectorChunkState, ) -from basic_memory.runtime.storage import RUNTIME_MARKDOWN_CONTENT_TYPE from basic_memory.runtime.vector_sync import VectorSyncBatchResult from basic_memory.schemas.search import ( SearchItemType, SearchRetrievalMode, - normalize_file_path_prefix, ) from basic_memory.temporal import TemporalFilter from basic_memory.utils import ensure_timezone_aware @@ -102,13 +102,19 @@ # vector-index change, or one still pending. Readiness reporting must apply the same # predicate — calling such a row "embedded" would report an index settled that # retrieval cannot answer from, which is the class of lie #1414 exists to remove. -# Both callers bind :project_id, :vector_index, and :embedding_model. -CURRENT_VECTOR_MANIFEST_PREDICATE = ( - "project_id = :project_id " - "AND vector_index = :vector_index " - "AND embedding_model = :embedding_model " - "AND embedding_status = 'ready'" -) +# Callers bind :vector_index and :embedding_model; the scope binds its own IDs. + + +def current_vector_manifest_predicate(scope: ProjectScope, params: dict[str, Any]) -> str: + """SQL admitting only manifest rows retrieval can answer from, within ``scope``.""" + return ( + f"{scope.predicate('project_id', params)} " + "AND vector_index = :vector_index " + "AND embedding_model = :embedding_model " + "AND embedding_status = 'ready'" + ) + + # Over-fetch factor for the rerank candidate chunk pool: chunks collapse to unique # (type, id) rows before reranking, so fetch several times reranker_candidates chunks # to keep enough unique documents in the rerank window. @@ -122,10 +128,6 @@ _SQLITE_MAX_PREPARE_WINDOW = semantic_vector_sync.SQLITE_MAX_PREPARE_WINDOW _BUILT_IN_VECTOR_INDEX_NAMES = frozenset({"pgvector", "sqlite-vec"}) -# Entity, observation, and relation rows in search_index carry ids from independent -# auto-increment sequences, so a bare id is ambiguous across row types. Every map in -# the vector/hybrid retrieval path must key rows by (type, id) to avoid collisions. -type SearchIndexKey = tuple[str, int] type StoredEmbeddingStatus = Literal["pending", "ready"] @@ -175,158 +177,6 @@ def __post_init__(self) -> None: object.__setattr__(self, "updated_at", ensure_timezone_aware(updated_at)) -def file_path_prefix_condition( - file_path_prefix: Optional[str], - params: Dict[str, Any], -) -> Optional[str]: - """Build the SQL scoping search rows to one directory subtree of the project. - - One implementation, shared verbatim by both backends: a subtree scope that - means different things on SQLite and Postgres would report an exact total - for a match set the other dialect never produces. - - Boundary: the compared prefix carries its trailing separator, so "specs" - admits "specs/api.md" and never "specs-archive/api.md". - - Why an explicit-length comparison rather than ``file_path LIKE 'specs/%'``: - LIKE reads "_" and "%" as wildcards and both are ordinary characters in a - directory name, so "my_notes" would silently also admit "my-notes"; and - LIKE case-folds differently per backend — SQLite's is ASCII-case-insensitive - while Postgres's is case-sensitive — so one filter would answer two - different questions. SUBSTR equality has no pattern language to escape and - compares under each backend's deterministic default text collation, which is - byte equality on both, so the dialects match exactly the same rows. - """ - normalized = normalize_file_path_prefix(file_path_prefix) - if normalized is None: - return None - prefix = f"{normalized}/" - params["file_path_prefix"] = prefix - params["file_path_prefix_length"] = len(prefix) - return "SUBSTR(search_index.file_path, 1, :file_path_prefix_length) = :file_path_prefix" - - -def metadata_filter_content_type_condition(params: Dict[str, Any]) -> str: - """Build the SQL restricting a metadata-filtered query to Markdown notes. - - Frontmatter is a Markdown-only construct, but every indexed file — PDF, - image, binary — gets its own ENTITY row whose ``entity_metadata`` carries no - keys at all. A positive predicate can never match one, so this constraint - was invisible until ``{"key": None}`` arrived: ``IS NULL`` is satisfied by - the *absence* of a key, which is exactly the state every regular file is in, - and the whole non-note half of a project counted into an exact total. - - Applied to any metadata filter, not just the null one, so the - frontmatter-only contract is a property of the clause rather than of which - operator happened to be used. Shared by both backends for the same reason - the subtree scope is: a filter that admits different rows per dialect would - report an exact total for a match set the other never produces. - """ - params["metadata_filter_content_type"] = RUNTIME_MARKDOWN_CONTENT_TYPE - return "entity.content_type = :metadata_filter_content_type" - - -# SQLite's LIKE has no default escape character, and Postgres's is already the -# backslash, so naming this one explicitly in every pattern is what lets a single -# escaped pattern mean the same thing on both backends. -_LIKE_ESCAPE_CHARACTER = "\\" - - -def metadata_contains_like_condition( - extract_expr: str, - value: Any, - *, - param_prefix: str, - params: Dict[str, Any], -) -> str: - """Build the compatibility half of an array-contains metadata filter. - - The primary half of a ``{"tags": ["security"]}`` filter asks JSON whether the - array holds the element — ``json_each`` on SQLite, ``@>`` on Postgres — and - answers only when the stored value really is a JSON array. Frontmatter - written before tags were normalized can hold the array's *text* instead, - either JSON-quoted ('["security", "auth"]') or as a Python repr - ("['security', 'auth']"), and only a substring match finds an element inside - those. Hence a pattern per quote style, and hence the pattern-language - problem this function exists to solve. - - LIKE reads "%" and "_" in the searched-for value as wildcards, so - interpolating the value raw turned `tags has 100%` into a pattern that also - matched "100-percent" — a wrong hit and an inflated exact total, produced by - the branch the caller only meant as a fallback. Escaping both wildcards and - the escape character itself makes the value literal again. - - Shared by both backends for the same reason the subtree scope is: a filter - that admits different rows per dialect would report an exact total for a - match set the other never produces. - """ - escaped = ( - str(value) - .replace(_LIKE_ESCAPE_CHARACTER, _LIKE_ESCAPE_CHARACTER * 2) - .replace("%", f"{_LIKE_ESCAPE_CHARACTER}%") - .replace("_", f"{_LIKE_ESCAPE_CHARACTER}_") - ) - double_quoted_param = f"{param_prefix}_like" - single_quoted_param = f"{param_prefix}_like_single" - params[double_quoted_param] = f'%"{escaped}"%' - params[single_quoted_param] = f"%'{escaped}'%" - escape_clause = f" ESCAPE '{_LIKE_ESCAPE_CHARACTER}'" - return ( - f"{extract_expr} LIKE :{double_quoted_param}{escape_clause} " - f"OR {extract_expr} LIKE :{single_quoted_param}{escape_clause}" - ) - - -def candidate_key_restriction_condition( - candidate_keys: Sequence[SearchIndexKey], - params: Dict[str, Any], -) -> str: - """Build the SQL restricting a filter query to an explicit set of search rows. - - This is what turns the vector/hybrid filter pass from "give me a page of everything - the filter admits" into "of *these* candidates, which does the filter admit". The - first question has an answer the size of the project and had to be capped, and every - candidate outside the cap was then read as disallowed (#1431). The second question's - answer is bounded by the candidate set itself, so no cap is needed and none of the - candidates can fall off the end. - - Keys are grouped by row type rather than emitted as one ``(type, id)`` pair per - branch: entity, observation, and relation ids come from independent sequences, so the - type is part of the identity, but a handful of type-scoped ``IN`` lists binds one - parameter per key instead of two and leaves the id list in the shape both planners - can drive an index from. PostgreSQL's ``search_index`` primary key is - ``(id, type, project_id)``. - - An empty candidate set is a real state, not a caller error — a vector search whose - every hit was already dropped — and it admits nothing, so it yields a false - predicate rather than the vacuous truth an empty ``OR`` would collapse to. - - Shared verbatim by both backends for the same reason the subtree scope is: a - restriction that admitted different rows per dialect would give semantic search a - different candidate set depending on which database happened to be underneath. - """ - ids_by_type: dict[str, list[int]] = {} - for row_type, row_id in candidate_keys: - ids_by_type.setdefault(row_type, []).append(row_id) - - branches: list[str] = [] - for type_index, (row_type, row_ids) in enumerate(ids_by_type.items()): - type_param = f"candidate_type_{type_index}" - params[type_param] = row_type - id_params: list[str] = [] - for id_index, row_id in enumerate(dict.fromkeys(row_ids)): - id_param = f"candidate_id_{type_index}_{id_index}" - params[id_param] = row_id - id_params.append(f":{id_param}") - branches.append( - f"(search_index.type = :{type_param} AND search_index.id IN ({', '.join(id_params)}))" - ) - - if not branches: - return "1 = 0" - return f"({' OR '.join(branches)})" - - async def purge_stale_search_index_rows( session_maker: async_sessionmaker[AsyncSession], project_id: int, @@ -390,6 +240,8 @@ class SearchRepositoryBase(ABC): _vector_tables_initialized: bool _semantic_vector_index: SemanticVectorIndex _semantic_vector_index_name: str = "" + # Runs compiled full-text statements for this repository's engine. + _fts: FtsBackend def __init__(self, session_maker: async_sessionmaker[AsyncSession], project_id: int): """Initialize with session maker and project_id filter. @@ -458,7 +310,6 @@ async def init_search_index(self) -> None: """ pass - @abstractmethod async def search( self, search_text: Optional[str] = None, @@ -477,42 +328,71 @@ async def search( limit: int = 10, offset: int = 0, allow_relaxed: bool = False, + session: AsyncSession | None = None, *, candidate_keys: Sequence[SearchIndexKey] | None = None, trace: SearchTraceCollector | None = None, ) -> List[SearchIndexRow]: - """Search across all indexed content. + """Search this repository's project. - Args: - search_text: Full-text search across title and content - permalink: Exact permalink match - permalink_match: Permalink pattern match (supports *) - title: Title search - note_types: Filter by note types (from metadata.note_type) - after_date: Filter by created_at > after_date - search_item_types: Filter by SearchItemType (ENTITY, OBSERVATION, RELATION) - categories: Filter observations by exact category (e.g. "requirement") - metadata_filters: Structured frontmatter metadata filters - file_path_prefix: Directory subtree scope, matched against file_path - temporal: Authored valid-time filter. Unlike after_date, which reads the - note's edit bookkeeping, this reads the time an observation claims to - be true of the world. Sources without such a claim are excluded. - limit: Maximum results to return - offset: Number of results to skip - candidate_keys: Restrict results to these ``(type, id)`` search rows. ``None`` - searches the whole project; an empty sequence matches nothing. Honored by - the full-text pass, which is where vector and hybrid retrieval evaluate - their structured filters: that pass asks which of a known candidate set a - filter admits instead of paging the filter's whole match set (#1431). - - Returns: - List of SearchIndexRow results with relevance scores + ``candidate_keys`` restricts results to those ``(type, id)`` search rows. + ``None`` searches the whole scope; an empty sequence matches nothing. Honored by + the full-text pass, which is where vector and hybrid retrieval evaluate their + structured filters: that pass asks which of a known candidate set a filter + admits instead of paging the filter's whole match set (#1431). - Backend-specific implementations: - - SQLite: Uses MATCH operator and bm25() for scoring - - Postgres: Uses @@ operator and ts_rank() for scoring + ``allow_relaxed=True`` retries a zero-result strict multi-word query with + OR-joined content terms. Only the hybrid path opts in: its FTS branch otherwise + contributes nothing for question-form queries. """ - pass + # --- Vector and hybrid: shared retrieval over this repository's scope --- + dispatched = await self._dispatch_retrieval_mode( + search_text=search_text, + permalink=permalink, + permalink_match=permalink_match, + title=title, + note_types=note_types, + after_date=after_date, + search_item_types=search_item_types, + categories=categories, + metadata_filters=metadata_filters, + file_path_prefix=file_path_prefix, + temporal=temporal, + retrieval_mode=retrieval_mode, + min_similarity=min_similarity, + limit=limit, + offset=offset, + trace=trace, + ) + if dispatched is not None: + return dispatched + + # --- Full text: the engine runs the compiled statement --- + query = PreparedSearchQuery( + search_text=search_text, + permalink=permalink, + permalink_match=permalink_match, + title=title, + note_types=note_types, + search_item_types=search_item_types, + categories=categories, + after_date=after_date, + metadata_filters=metadata_filters, + file_path_prefix=file_path_prefix, + temporal=temporal, + retrieval_mode=retrieval_mode, + min_similarity=min_similarity, + ) + return await self._fts.search( + self.scope, + query, + limit=limit, + offset=offset, + allow_relaxed=allow_relaxed, + session=session, + candidate_keys=candidate_keys, + trace=trace, + ) async def count( self, @@ -531,10 +411,25 @@ async def count( min_similarity: Optional[float] = None, allow_relaxed: bool = False, ) -> int: - """Count results when a backend-specific COUNT query is available.""" + """Count full-text matches with the same filters as ``search``.""" if retrieval_mode != SearchRetrievalMode.FTS: raise ValueError("Exact counts are only supported for full-text search retrieval.") - raise NotImplementedError("Backend search repositories must implement full-text counts.") + query = PreparedSearchQuery( + search_text=search_text, + permalink=permalink, + permalink_match=permalink_match, + title=title, + note_types=note_types, + search_item_types=search_item_types, + categories=categories, + after_date=after_date, + metadata_filters=metadata_filters, + file_path_prefix=file_path_prefix, + temporal=temporal, + retrieval_mode=retrieval_mode, + min_similarity=min_similarity, + ) + return await self._fts.count(self.scope, query, allow_relaxed=allow_relaxed) # ------------------------------------------------------------------ # Abstract methods — semantic search (backend-specific DB operations) @@ -573,7 +468,7 @@ async def _run_vector_query( if trace is not None: trace.readiness = await read_manifest_readiness( session, - self.project_id, + self.scope, self._semantic_vector_index_name, self._embedding_model_key(), ) @@ -588,7 +483,7 @@ async def _run_vector_query( if trace is not None and trace.readiness is None: trace.readiness = await read_manifest_readiness( session, - self.project_id, + self.scope, self._semantic_vector_index_name, self._embedding_model_key(), ) @@ -715,11 +610,11 @@ async def _hydrate_vector_matches( chunks_by_key: dict[VectorKey, str] = {} for batch_start in range(0, len(matches), VECTOR_HYDRATION_BATCH_SIZE): batch = matches[batch_start : batch_start + VECTOR_HYDRATION_BATCH_SIZE] - params: dict[str, object] = { - "project_id": self.project_id, + params: dict[str, Any] = { "vector_index": self._semantic_vector_index_name, "embedding_model": self._embedding_model_key(), } + manifest_predicate = current_vector_manifest_predicate(self.scope, params) predicates: list[str] = [] for index, match in enumerate(batch): params[f"entity_id_{index}"] = match.key.entity_id @@ -734,7 +629,7 @@ async def _hydrate_vector_matches( result = await session.execute( text( "SELECT entity_id, chunk_key, chunk_text FROM search_vector_chunks " - "WHERE " + CURRENT_VECTOR_MANIFEST_PREDICATE + " " + "WHERE " + manifest_predicate + " " "AND (" + " OR ".join(predicates) + ")" ), params, @@ -770,7 +665,7 @@ async def _hydrate_vector_matches( for match in matches if match.key not in chunks_by_key ] - drops = await classify_hydration_drops(session, self.project_id, dropped_keys) + drops = await classify_hydration_drops(session, self.scope, dropped_keys) chunk_matches: dict[SearchIndexKey, list[tuple[str, float, int | None]]] = {} malformed_drops: list[HydrationDropped] = [] for row in hydrated: @@ -2796,17 +2691,15 @@ async def _fetch_search_index_rows_by_ids( if not row_ids: return {} placeholders = ",".join(f":id_{idx}" for idx in range(len(row_ids))) - params: dict[str, Any] = { - **{f"id_{idx}": rid for idx, rid in enumerate(row_ids)}, - "project_id": self.project_id, - } + params: dict[str, Any] = {f"id_{idx}": rid for idx, rid in enumerate(row_ids)} + scope_predicate = self.scope.predicate("project_id", params) sql = f""" SELECT project_id, id, title, permalink, file_path, type, metadata, from_id, to_id, relation_type, entity_id, content_snippet, category, created_at, updated_at, 0 as score FROM search_index - WHERE project_id = :project_id + WHERE {scope_predicate} AND id IN ({placeholders}) """ result: dict[SearchIndexKey, SearchIndexRow] = {} diff --git a/src/basic_memory/repository/search_trace.py b/src/basic_memory/repository/search_trace.py index eb7f53ffa..c090b08b0 100644 --- a/src/basic_memory/repository/search_trace.py +++ b/src/basic_memory/repository/search_trace.py @@ -1,5 +1,6 @@ """Typed, execution-native trace values for the search retrieval pipeline.""" +from basic_memory.repository.search_scope import ProjectScope from collections.abc import Mapping, Sequence from dataclasses import dataclass from typing import Any, Literal @@ -508,25 +509,23 @@ class HydrationDropKey: async def read_manifest_readiness( session: Any, - project_id: int, + scope: ProjectScope, vector_index: str, embedding_model: str, ) -> ManifestReadiness: """Count configured readiness and rows stored under another vector identity.""" from sqlalchemy import text + params: dict[str, Any] = {"vector_index": vector_index, "embedding_model": embedding_model} + scope_predicate = scope.predicate("project_id", params) readiness_result = await session.execute( text( "SELECT embedding_status, COUNT(*) AS row_count " - "FROM search_vector_chunks WHERE project_id = :project_id " + f"FROM search_vector_chunks WHERE {scope_predicate} " "AND vector_index = :vector_index AND embedding_model = :embedding_model " "GROUP BY embedding_status" ), - { - "project_id": project_id, - "vector_index": vector_index, - "embedding_model": embedding_model, - }, + params, ) counts = { str(row["embedding_status"]): int(row["row_count"]) @@ -534,14 +533,10 @@ async def read_manifest_readiness( } other_result = await session.execute( text( - "SELECT COUNT(*) FROM search_vector_chunks WHERE project_id = :project_id " + f"SELECT COUNT(*) FROM search_vector_chunks WHERE {scope_predicate} " "AND (vector_index <> :vector_index OR embedding_model <> :embedding_model)" ), - { - "project_id": project_id, - "vector_index": vector_index, - "embedding_model": embedding_model, - }, + params, ) return ManifestReadiness( configured_index=vector_index, @@ -554,7 +549,7 @@ async def read_manifest_readiness( async def classify_hydration_drops( session: Any, - project_id: int, + scope: ProjectScope, dropped_keys: Sequence[HydrationDropKey], ) -> tuple[HydrationDropped, ...]: """Classify adapter hits rejected by authoritative manifest hydration.""" @@ -570,7 +565,8 @@ async def classify_hydration_drops( HYDRATION_DROP_CLASSIFICATION_BATCH_SIZE, ): batch = dropped_keys[batch_start : batch_start + HYDRATION_DROP_CLASSIFICATION_BATCH_SIZE] - params: dict[str, object] = {"project_id": project_id} + params: dict[str, Any] = {} + scope_predicate = scope.predicate("project_id", params) predicates: list[str] = [] for index, dropped in enumerate(batch): params[f"entity_id_{index}"] = dropped.entity_id @@ -585,7 +581,7 @@ async def classify_hydration_drops( result = await session.execute( text( "SELECT entity_id, chunk_key, embedding_model, vector_index, embedding_status " - "FROM search_vector_chunks WHERE project_id = :project_id AND (" + f"FROM search_vector_chunks WHERE {scope_predicate} AND (" + " OR ".join(predicates) + ")" ), diff --git a/src/basic_memory/repository/sqlite_search_query.py b/src/basic_memory/repository/sqlite_search_query.py index 9fa161bab..42dfdbd67 100644 --- a/src/basic_memory/repository/sqlite_search_query.py +++ b/src/basic_memory/repository/sqlite_search_query.py @@ -1,32 +1,35 @@ -"""SQLite FTS5 query preparation: term syntax and filter compilation. +"""SQLite FTS5 query preparation and execution. -Pure functions over a ``ProjectScope`` and the caller's filters. Nothing here opens -a session or owns an index; the repository resolves the one piece of live schema -state the compiler needs (the entity table's columns) and passes it in. +Term preparation and filter compilation are pure functions over a ``ProjectScope`` and +a ``PreparedSearchQuery``. ``SQLiteFts`` runs the compiled statement and owns FTS5's +failure semantics. Nothing here initializes or mutates an index. """ import re +import time from collections.abc import Collection, Sequence -from datetime import datetime from typing import Any +import logfire +from loguru import logger +from sqlalchemy import text +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from basic_memory import db from basic_memory.repository.metadata_filters import build_sqlite_json_path, parse_metadata_filters from basic_memory.repository.script_ngrams import analyze_script_query from basic_memory.repository.search_filters import ( AFTER_DATE_ORDER_BY, SQLITE_FILTER_DIALECT, CompiledFilter, - shared_filter_conditions, -) -from basic_memory.repository.search_query import relaxed_query_words -from basic_memory.repository.search_repository_base import ( - SearchIndexKey, metadata_contains_like_condition, metadata_filter_content_type_condition, + shared_filter_conditions, ) +from basic_memory.repository.search_index_row import SearchIndexKey, SearchIndexRow +from basic_memory.repository.search_query import PreparedSearchQuery, relaxed_query_words from basic_memory.repository.search_scope import ProjectScope -from basic_memory.schemas.search import SearchItemType -from basic_memory.temporal import TemporalFilter +from basic_memory.repository.search_trace import SearchTraceCollector, build_fts_page_stage SQLITE_WORD_COLUMNS = "{title content_stems content_snippet}" @@ -39,6 +42,24 @@ _SPACE_OR_SPECIAL_CHARS = frozenset(" .:;,<>?/-") _BOOLEAN_OPERATOR_PATTERN = r"(\bAND\b|\bOR\b|\bNOT\b)" +# Every FTS statement returns these columns plus a score. +_RESULT_COLUMNS = """ + search_index.project_id, + search_index.id, + search_index.title, + search_index.permalink, + search_index.file_path, + search_index.type, + search_index.metadata, + search_index.from_id, + search_index.to_id, + search_index.relation_type, + search_index.entity_id, + search_index.content_snippet, + search_index.category, + search_index.created_at, + search_index.updated_at""" + # --- Term preparation --- @@ -188,19 +209,9 @@ def is_fts5_syntax_error(exc: Exception) -> bool: def compile_fts_filter( scope: ProjectScope, + query: PreparedSearchQuery, *, entity_columns: Collection[str], - search_text: str | None = None, - permalink: str | None = None, - permalink_match: str | None = None, - title: str | None = None, - note_types: Sequence[str] | None = None, - after_date: datetime | None = None, - search_item_types: Sequence[SearchItemType] | None = None, - categories: Sequence[str] | None = None, - metadata_filters: dict[str, Any] | None = None, - file_path_prefix: str | None = None, - temporal: TemporalFilter | None = None, candidate_keys: Sequence[SearchIndexKey] | None = None, ) -> CompiledFilter: """Compile SQLite FTS FROM/WHERE/score shared by search and count. @@ -210,22 +221,13 @@ def compile_fts_filter( """ params: dict[str, Any] = {} conditions = shared_filter_conditions( - scope, - params, - dialect=SQLITE_FILTER_DIALECT, - permalink=permalink, - file_path_prefix=file_path_prefix, - candidate_keys=candidate_keys, - search_item_types=search_item_types, - categories=categories, - note_types=note_types, - after_date=after_date, - temporal=temporal, + scope, params, dialect=SQLITE_FILTER_DIALECT, query=query, candidate_keys=candidate_keys ) match_conditions: list[str] = [] from_clause = "search_index" score_expression = "bm25(search_index)" preserve_match_score = False + search_text = query.search_text # Wildcard-only and blank text add no text condition: every row matches. if search_text and search_text.strip() not in ("", "*"): @@ -268,15 +270,15 @@ def compile_fts_filter( "search_index.content_snippet MATCH :text)" ) - if title: - params["title_text"] = prepare_search_term(title.strip(), is_prefix=False) + if query.title: + params["title_text"] = prepare_search_term(query.title.strip(), is_prefix=False) match_conditions.append("search_index.title MATCH :title_text") - if permalink_match: + if query.permalink_match: # GLOB patterns keep their syntax; prepare_search_term would quote the slashes. - permalink_text = permalink_match.lower().strip() + permalink_text = query.permalink_match.lower().strip() params["permalink"] = permalink_text - if "*" in permalink_match: + if "*" in query.permalink_match: conditions.append("search_index.permalink GLOB :permalink") elif "/" in permalink_text: conditions.append("search_index.permalink = :permalink") @@ -285,8 +287,8 @@ def compile_fts_filter( params["permalink"] = prepare_search_term(permalink_text, is_prefix=False) match_conditions.append("search_index.permalink MATCH :permalink") - if metadata_filters: - parsed_filters = parse_metadata_filters(metadata_filters) + if query.metadata_filters: + parsed_filters = parse_metadata_filters(query.metadata_filters) from_clause = "search_index JOIN entity ON search_index.entity_id = entity.id" # Frontmatter filters answer for notes only; see # metadata_filter_content_type_condition for why every regular file would @@ -397,7 +399,7 @@ def compile_fts_filter( # word-column OR predicates cannot evaluate bm25 in the same derived query. # Outcome: rank script matches before joining metadata; retain the established # rowid-filter path for word-only searches. - if metadata_filters and match_conditions: + if query.metadata_filters and match_conditions: match_where = " AND ".join(match_conditions) if preserve_match_score: from_clause = ( @@ -418,6 +420,198 @@ def compile_fts_filter( from_clause=from_clause, where_clause=" AND ".join(conditions), params=params, - order_by_clause=AFTER_DATE_ORDER_BY if after_date else "", + order_by_clause=AFTER_DATE_ORDER_BY if query.after_date else "", score_expression=score_expression, ) + + +# --- Execution --- + + +class SQLiteFts: + """Run FTS5 statements for any scope in one database. + + Holds the one piece of live schema state compilation needs: the entity table's + columns, read once per instance so generated frontmatter columns are used when the + database has them. + """ + + def __init__(self, session_maker: async_sessionmaker[AsyncSession]) -> None: + self._session_maker = session_maker + self._entity_columns: frozenset[str] | None = None + + async def _entity_column_names(self) -> frozenset[str]: + if self._entity_columns is None: + async with db.scoped_session(self._session_maker) as session: + result = await session.execute(text("PRAGMA table_info(entity)")) + self._entity_columns = frozenset(row[1] for row in result.fetchall()) + return self._entity_columns + + async def search( + self, + scope: ProjectScope, + query: PreparedSearchQuery, + *, + limit: int, + offset: int, + allow_relaxed: bool = False, + session: AsyncSession | None = None, + candidate_keys: Sequence[SearchIndexKey] | None = None, + trace: SearchTraceCollector | None = None, + ) -> list[SearchIndexRow]: + """Run one FTS5 page. + + ``allow_relaxed=True`` retries a zero-result strict multi-word query with + OR-joined content terms. Only the hybrid path opts in: its FTS branch otherwise + contributes nothing for question-form queries. Service-level FTS searches keep + their own conservative fallback. + """ + search_text = query.search_text + # Generated frontmatter columns are read only when a metadata filter needs them. + entity_columns = ( + await self._entity_column_names() if query.metadata_filters else frozenset() + ) + compiled = compile_fts_filter( + scope, query, entity_columns=entity_columns, candidate_keys=candidate_keys + ) + params = compiled.params + params["limit"] = limit + params["offset"] = offset + relaxed_search_text = search_text + if search_text and "script_text" in params: + relaxed_search_text = analyze_script_query(search_text.strip()).word_text + + sql = f""" + SELECT{_RESULT_COLUMNS}, + {compiled.score_expression} as score + FROM {compiled.from_clause} + WHERE {compiled.where_clause} + ORDER BY score ASC {compiled.order_by_clause} + LIMIT :limit + OFFSET :offset + """ + + logger.trace(f"Search {sql} params: {params}") + fts_started_at = time.perf_counter() if trace is not None else None + + async def run_search(active_session: AsyncSession): + result = await active_session.execute(text(sql), params) + rows = result.fetchall() + relaxed_fallback_used = False + # Trigger: multi-word natural-language query matched nothing under the + # default all-terms-AND semantics. + # Why: questions ("when did X do Y") rarely have every word in one + # document; without relaxation the FTS half of hybrid search contributes + # zero candidates and ranking degrades to vector-only. + # Outcome: one retry with OR-joined prefix terms; bm25 still ranks + # multi-term matches first. + relaxed = relaxed_fts_text(relaxed_search_text) if allow_relaxed and not rows else None + if relaxed and params.get("text"): + relaxed_fallback_used = True + params["text"] = ( + f"{SQLITE_WORD_COLUMNS}: ({relaxed})" if "script_text" in params else relaxed + ) + logger.debug( + "Strict SQLite FTS returned 0 results; retrying relaxed FTS query " + f"strict='{search_text}' relaxed='{relaxed}'" + ) + with logfire.span( + "search.relaxed_fts_retry", + backend="sqlite", + token_count=len(relaxed_query_words(relaxed_search_text) or ()), + limit=limit, + offset=offset, + ): + result = await active_session.execute(text(sql), params) + rows = result.fetchall() + return rows, relaxed_fallback_used + + try: + if session is not None: + rows, relaxed_fallback_used = await run_search(session) + else: + async with db.scoped_session(self._session_maker) as owned_session: + rows, relaxed_fallback_used = await run_search(owned_session) + except Exception as e: + # An FTS5 syntax error answers with no rows rather than failing the request. + if is_fts5_syntax_error(e): # pragma: no cover + logger.warning(f"FTS5 syntax error for search term: {search_text}, error: {e}") + if trace is not None: + trace.fts = build_fts_page_stage( + [], + relaxed_fallback_used=False, + fts_ms=( + (time.perf_counter() - fts_started_at) * 1000 + if fts_started_at is not None + else None + ), + ) + return [] + logger.error(f"Database error during search: {e}") + raise + + results = [SearchIndexRow.from_mapping(row._asdict()) for row in rows] + if trace is not None: + trace.fts = build_fts_page_stage( + [((row.type, row.id), row.score or 0.0) for row in results], + relaxed_fallback_used=relaxed_fallback_used, + fts_ms=( + (time.perf_counter() - fts_started_at) * 1000 + if fts_started_at is not None + else None + ), + ) + + logger.trace(f"Found {len(results)} search results") + for r in results: + logger.trace( + f"Search result: project_id: {r.project_id} type:{r.type} title: {r.title} permalink: {r.permalink} score: {r.score}" + ) + return results + + async def count( + self, + scope: ProjectScope, + query: PreparedSearchQuery, + *, + allow_relaxed: bool = False, + ) -> int: + """Count rows matching the FTS5 query, with the same relaxed retry as search.""" + search_text = query.search_text + entity_columns = ( + await self._entity_column_names() if query.metadata_filters else frozenset() + ) + compiled = compile_fts_filter(scope, query, entity_columns=entity_columns) + params = compiled.params + sql = f"SELECT COUNT(*) FROM {compiled.from_clause} WHERE {compiled.where_clause}" + logger.trace(f"Count {sql} params: {params}") + relaxed_search_text = search_text + if search_text and "script_text" in params: + relaxed_search_text = analyze_script_query(search_text.strip()).word_text + try: + async with db.scoped_session(self._session_maker) as session: + result = await session.execute(text(sql), params) + total = int(result.scalar_one()) + relaxed = ( + relaxed_fts_text(relaxed_search_text) if allow_relaxed and total == 0 else None + ) + if relaxed and params.get("text"): + params["text"] = ( + f"{SQLITE_WORD_COLUMNS}: ({relaxed})" + if "script_text" in params + else relaxed + ) + with logfire.span( + "search.count.relaxed_fts_retry", + backend="sqlite", + token_count=len(relaxed_query_words(relaxed_search_text) or ()), + ): + result = await session.execute(text(sql), params) + total = int(result.scalar_one()) + return total + except Exception as e: + if is_fts5_syntax_error(e): # pragma: no cover + logger.warning(f"FTS5 syntax error for search term: {search_text}, error: {e}") + return 0 + logger.error(f"Database error during search count: {e}") + raise diff --git a/src/basic_memory/repository/sqlite_search_repository.py b/src/basic_memory/repository/sqlite_search_repository.py index 11bb39ff0..aa75461e7 100644 --- a/src/basic_memory/repository/sqlite_search_repository.py +++ b/src/basic_memory/repository/sqlite_search_repository.py @@ -1,13 +1,10 @@ """SQLite FTS5-based search repository implementation.""" import asyncio -import time from collections.abc import Sequence from contextlib import asynccontextmanager -from datetime import datetime -from typing import Any, override, List, Optional +from typing import override, List -import logfire from loguru import logger from sqlalchemy import text from sqlalchemy.exc import OperationalError as SAOperationalError @@ -26,29 +23,15 @@ from basic_memory.repository.rerank_provider import RerankProvider from basic_memory.repository.rerank_provider_factory import create_rerank_provider from basic_memory.repository.search_index_row import SearchIndexRow -from basic_memory.repository.search_query import relaxed_query_words from basic_memory.repository.search_repository_base import ( - SearchIndexKey, SearchRepositoryBase, ) -from basic_memory.repository.script_ngrams import analyze_script_query -from basic_memory.repository.search_trace import ( - SearchTraceCollector, - build_fts_page_stage, -) -from basic_memory.repository.sqlite_search_query import ( - SQLITE_WORD_COLUMNS, - compile_fts_filter, - is_fts5_syntax_error, - relaxed_fts_text, -) +from basic_memory.repository.sqlite_search_query import SQLiteFts from basic_memory.repository.semantic_errors import SemanticDependenciesMissingError from basic_memory.repository.semantic_vector_index import SemanticVectorIndex from basic_memory.repository.semantic_vector_sync import StagedVectorDeletion from basic_memory.repository.semantic_vector_index_factory import build_vector_index_scope from basic_memory.repository.sqlite_vec_index import SQLiteVecIndex -from basic_memory.schemas.search import SearchItemType, SearchRetrievalMode -from basic_memory.temporal import TemporalFilter class SQLiteSearchRepository(SearchRepositoryBase): @@ -72,7 +55,7 @@ def __init__( rerank_provider: RerankProvider | None = None, ): super().__init__(session_maker, project_id) - self._entity_columns: set[str] | None = None + self._fts = SQLiteFts(session_maker) self._app_config = app_config or ConfigManager().config self._semantic_enabled = self._app_config.semantic_search_enabled self._semantic_vector_k = self._app_config.semantic_vector_k @@ -109,13 +92,6 @@ def __init__( ), ) - async def _get_entity_columns(self) -> set[str]: - if self._entity_columns is None: - async with db.scoped_session(self.session_maker) as session: - result = await session.execute(text("PRAGMA table_info(entity)")) - self._entity_columns = {row[1] for row in result.fetchall()} - return self._entity_columns - @override async def init_search_index(self): """Create FTS5 virtual table for search if it doesn't exist. @@ -358,25 +334,6 @@ async def _prepare_vector_session(self, session: AsyncSession) -> None: """Load sqlite-vec extension for the session.""" await self._ensure_sqlite_vec_loaded(session) - # sqlite-vec hard limit for knn k parameter - SQLITE_VEC_MAX_K = 4096 - - @override - async def _run_vector_query( - self, - session: AsyncSession, - query_embedding: list[float], - candidate_limit: int, - *, - trace: SearchTraceCollector | None = None, - ) -> list[dict[str, Any]]: - return await super()._run_vector_query( - session, - query_embedding, - candidate_limit, - trace=trace, - ) - @override async def _delete_entity_chunks( self, @@ -492,280 +449,3 @@ async def index_item(self, search_index_row: SearchIndexRow) -> None: async def bulk_index_items(self, search_index_rows: List[SearchIndexRow]) -> None: """Index multiple rows in FTS only.""" await super().bulk_index_items(search_index_rows) - - # ------------------------------------------------------------------ - # FTS search (backend-specific) - # ------------------------------------------------------------------ - - @override - async def search( - self, - search_text: Optional[str] = None, - permalink: Optional[str] = None, - permalink_match: Optional[str] = None, - title: Optional[str] = None, - note_types: Optional[List[str]] = None, - after_date: Optional[datetime] = None, - search_item_types: Optional[List[SearchItemType]] = None, - categories: Optional[List[str]] = None, - metadata_filters: Optional[dict[str, Any]] = None, - file_path_prefix: Optional[str] = None, - temporal: Optional[TemporalFilter] = None, - retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS, - min_similarity: Optional[float] = None, - limit: int = 10, - offset: int = 0, - allow_relaxed: bool = False, - session: AsyncSession | None = None, - *, - candidate_keys: Sequence[SearchIndexKey] | None = None, - trace: SearchTraceCollector | None = None, - ) -> List[SearchIndexRow]: - """Search across all indexed content using SQLite FTS5. - - ``allow_relaxed=True`` retries a zero-result strict multi-word query - with OR-joined content terms. Only the hybrid path opts in: its FTS - branch otherwise contributes nothing for question-form queries. - Service-level FTS searches keep their own conservative fallback. - """ - # --- Dispatch vector / hybrid modes (shared logic) --- - dispatched = await self._dispatch_retrieval_mode( - search_text=search_text, - permalink=permalink, - permalink_match=permalink_match, - title=title, - note_types=note_types, - after_date=after_date, - search_item_types=search_item_types, - categories=categories, - metadata_filters=metadata_filters, - file_path_prefix=file_path_prefix, - temporal=temporal, - retrieval_mode=retrieval_mode, - min_similarity=min_similarity, - limit=limit, - offset=offset, - trace=trace, - ) - if dispatched is not None: - return dispatched - - # --- FTS mode (SQLite-specific) --- - # Generated frontmatter columns are read only when a metadata filter needs them. - entity_columns = await self._get_entity_columns() if metadata_filters else frozenset() - compiled = compile_fts_filter( - self.scope, - entity_columns=entity_columns, - search_text=search_text, - permalink=permalink, - permalink_match=permalink_match, - title=title, - note_types=note_types, - after_date=after_date, - search_item_types=search_item_types, - categories=categories, - metadata_filters=metadata_filters, - file_path_prefix=file_path_prefix, - temporal=temporal, - candidate_keys=candidate_keys, - ) - params = compiled.params - params["limit"] = limit - params["offset"] = offset - relaxed_search_text = search_text - if search_text and "script_text" in params: - relaxed_search_text = analyze_script_query(search_text.strip()).word_text - - sql = f""" - SELECT - search_index.project_id, - search_index.id, - search_index.title, - search_index.permalink, - search_index.file_path, - search_index.type, - search_index.metadata, - search_index.from_id, - search_index.to_id, - search_index.relation_type, - search_index.entity_id, - search_index.content_snippet, - search_index.category, - search_index.created_at, - search_index.updated_at, - {compiled.score_expression} as score - FROM {compiled.from_clause} - WHERE {compiled.where_clause} - ORDER BY score ASC {compiled.order_by_clause} - LIMIT :limit - OFFSET :offset - """ - - logger.trace(f"Search {sql} params: {params}") - fts_started_at = time.perf_counter() if trace is not None else None - - async def run_search(active_session: AsyncSession): - result = await active_session.execute(text(sql), params) - rows = result.fetchall() - relaxed_fallback_used = False - # Trigger: multi-word natural-language query matched nothing - # under the default all-terms-AND semantics. - # Why: questions ("when did X do Y") rarely have every word in - # one document; without relaxation the FTS half of hybrid - # search contributes zero candidates and ranking degrades to - # vector-only. - # Outcome: one retry with OR-joined prefix terms; bm25 still - # ranks multi-term matches first. - relaxed = relaxed_fts_text(relaxed_search_text) if allow_relaxed and not rows else None - if relaxed and params.get("text"): - relaxed_fallback_used = True - params["text"] = ( - f"{SQLITE_WORD_COLUMNS}: ({relaxed})" if "script_text" in params else relaxed - ) - logger.debug( - "Strict SQLite FTS returned 0 results; retrying relaxed FTS query " - f"strict='{search_text}' relaxed='{relaxed}'" - ) - with logfire.span( - "search.relaxed_fts_retry", - backend="sqlite", - token_count=len(relaxed_query_words(relaxed_search_text) or ()), - limit=limit, - offset=offset, - ): - result = await active_session.execute(text(sql), params) - rows = result.fetchall() - return rows, relaxed_fallback_used - - try: - if session is not None: - rows, relaxed_fallback_used = await run_search(session) - else: - async with db.scoped_session(self.session_maker) as owned_session: - rows, relaxed_fallback_used = await run_search(owned_session) - except Exception as e: - # Handle FTS5 syntax errors and provide user-friendly feedback - if is_fts5_syntax_error(e): # pragma: no cover - logger.warning(f"FTS5 syntax error for search term: {search_text}, error: {e}") - # Return empty results rather than crashing - if trace is not None: - trace.fts = build_fts_page_stage( - [], - relaxed_fallback_used=False, - fts_ms=( - (time.perf_counter() - fts_started_at) * 1000 - if fts_started_at is not None - else None - ), - ) - return [] - else: - # Re-raise other database errors - logger.error(f"Database error during search: {e}") - raise - - results = [SearchIndexRow.from_mapping(row._asdict()) for row in rows] - if trace is not None: - trace.fts = build_fts_page_stage( - [((row.type, row.id), row.score or 0.0) for row in results], - relaxed_fallback_used=relaxed_fallback_used, - fts_ms=( - (time.perf_counter() - fts_started_at) * 1000 - if fts_started_at is not None - else None - ), - ) - - logger.trace(f"Found {len(results)} search results") - for r in results: - logger.trace( - f"Search result: project_id: {r.project_id} type:{r.type} title: {r.title} permalink: {r.permalink} score: {r.score}" - ) - - return results - - @override - async def count( - self, - search_text: Optional[str] = None, - permalink: Optional[str] = None, - permalink_match: Optional[str] = None, - title: Optional[str] = None, - note_types: Optional[List[str]] = None, - after_date: Optional[datetime] = None, - search_item_types: Optional[List[SearchItemType]] = None, - categories: Optional[List[str]] = None, - metadata_filters: Optional[dict[str, Any]] = None, - file_path_prefix: Optional[str] = None, - temporal: Optional[TemporalFilter] = None, - retrieval_mode: SearchRetrievalMode = SearchRetrievalMode.FTS, - min_similarity: Optional[float] = None, - allow_relaxed: bool = False, - ) -> int: - """Count indexed content matching the SQLite FTS query.""" - if retrieval_mode != SearchRetrievalMode.FTS: - return await super().count( - search_text=search_text, - permalink=permalink, - permalink_match=permalink_match, - title=title, - note_types=note_types, - after_date=after_date, - search_item_types=search_item_types, - categories=categories, - metadata_filters=metadata_filters, - file_path_prefix=file_path_prefix, - temporal=temporal, - retrieval_mode=retrieval_mode, - min_similarity=min_similarity, - ) - - entity_columns = await self._get_entity_columns() if metadata_filters else frozenset() - compiled = compile_fts_filter( - self.scope, - entity_columns=entity_columns, - search_text=search_text, - permalink=permalink, - permalink_match=permalink_match, - title=title, - note_types=note_types, - after_date=after_date, - search_item_types=search_item_types, - categories=categories, - metadata_filters=metadata_filters, - file_path_prefix=file_path_prefix, - temporal=temporal, - ) - params = compiled.params - sql = f"SELECT COUNT(*) FROM {compiled.from_clause} WHERE {compiled.where_clause}" - logger.trace(f"Count {sql} params: {params}") - relaxed_search_text = search_text - if search_text and "script_text" in params: - relaxed_search_text = analyze_script_query(search_text.strip()).word_text - try: - async with db.scoped_session(self.session_maker) as session: - result = await session.execute(text(sql), params) - total = int(result.scalar_one()) - relaxed = ( - relaxed_fts_text(relaxed_search_text) if allow_relaxed and total == 0 else None - ) - if relaxed and params.get("text"): - params["text"] = ( - f"{SQLITE_WORD_COLUMNS}: ({relaxed})" - if "script_text" in params - else relaxed - ) - with logfire.span( - "search.count.relaxed_fts_retry", - backend="sqlite", - token_count=len(relaxed_query_words(relaxed_search_text) or ()), - ): - result = await session.execute(text(sql), params) - total = int(result.scalar_one()) - return total - except Exception as e: - if is_fts5_syntax_error(e): # pragma: no cover - logger.warning(f"FTS5 syntax error for search term: {search_text}, error: {e}") - return 0 - logger.error(f"Database error during search count: {e}") - raise diff --git a/src/basic_memory/services/project_readiness.py b/src/basic_memory/services/project_readiness.py index f37a14fc2..ed7bc1ca8 100644 --- a/src/basic_memory/services/project_readiness.py +++ b/src/basic_memory/services/project_readiness.py @@ -24,7 +24,8 @@ from basic_memory.repository.embedding_provider_factory import ( configured_embedding_provider_identity, ) -from basic_memory.repository.search_repository_base import CURRENT_VECTOR_MANIFEST_PREDICATE +from basic_memory.repository.search_repository_base import current_vector_manifest_predicate +from basic_memory.repository.search_scope import ProjectScope from basic_memory.repository.semantic_vector_index_factory import ( resolve_semantic_vector_index_name, ) @@ -310,10 +311,21 @@ async def _embedding_counts( # embedded -- the third time a count and the thing it measures disagreed # (#1440 review). The marker is written by the sharded sync itself, in # `record_entity_vector_deferrals`, so the two cannot drift. + manifest_params: dict[str, object] = { + "project_id": project_id, + "vector_index": resolve_semantic_vector_index_name( + self.app_config, + self.app_config.database_backend, + ), + "embedding_model": configured_embedding_provider_identity(self.app_config), + } + manifest_predicate = current_vector_manifest_predicate( + ProjectScope.single(project_id), manifest_params + ) usable_result = await session.execute( text( "SELECT DISTINCT entity_id FROM search_vector_chunks " - "WHERE " + CURRENT_VECTOR_MANIFEST_PREDICATE + " " + "WHERE " + manifest_predicate + " " # Applied as a subquery so the shared predicate is used verbatim # rather than rewritten to carry a table alias. "AND entity_id NOT IN (" @@ -321,14 +333,7 @@ async def _embedding_counts( " AND vector_sync_deferred_at IS NOT NULL" ")" ), - { - "project_id": project_id, - "vector_index": resolve_semantic_vector_index_name( - self.app_config, - self.app_config.database_backend, - ), - "embedding_model": configured_embedding_provider_identity(self.app_config), - }, + manifest_params, ) usable_entity_ids = {int(entity_id) for entity_id in usable_result.scalars().all()} return len(owed_entity_ids), len(owed_entity_ids & usable_entity_ids) diff --git a/src/basic_memory/services/search_service.py b/src/basic_memory/services/search_service.py index 301e643f2..0a0f70317 100644 --- a/src/basic_memory/services/search_service.py +++ b/src/basic_memory/services/search_service.py @@ -4,7 +4,7 @@ import ast import re from collections.abc import Mapping, Sequence -from dataclasses import dataclass, replace +from dataclasses import replace from datetime import datetime from typing import Any, List, Optional, Set, Dict @@ -23,7 +23,7 @@ SearchIndexRow, SearchRepository, ) -from basic_memory.repository.search_query import relaxed_query_words +from basic_memory.repository.search_query import PreparedSearchQuery, relaxed_query_words from basic_memory.repository.search_trace import SearchTraceCollector from basic_memory.schemas.base import normalize_note_type from basic_memory.schemas.search import SearchQuery, SearchItemType, SearchRetrievalMode @@ -42,25 +42,6 @@ MAX_CONTENT_STEMS_SIZE = 6000 -@dataclass(frozen=True) -class PreparedSearchQuery: - """Normalized query inputs shared by search and count.""" - - search_text: str | None - permalink: str | None - permalink_match: str | None - title: str | None - note_types: list[str] | None - search_item_types: list[SearchItemType] | None - categories: list[str] | None - after_date: datetime | None - metadata_filters: dict[str, Any] | None - file_path_prefix: str | None - temporal: TemporalFilter | None - retrieval_mode: SearchRetrievalMode - min_similarity: float | None - - def entity_embeddings_enabled(entity: Entity) -> bool: """Return whether semantic embeddings should be generated for this entity. diff --git a/tests/repository/test_hybrid_fusion.py b/tests/repository/test_hybrid_fusion.py index 4bf7135b1..0ad16896b 100644 --- a/tests/repository/test_hybrid_fusion.py +++ b/tests/repository/test_hybrid_fusion.py @@ -6,6 +6,8 @@ 3. Produces zero fused score when the source score is zero """ +from sqlalchemy.ext.asyncio import AsyncSession +from basic_memory.repository.search_scope import ProjectScope from collections.abc import Sequence from dataclasses import dataclass from datetime import datetime @@ -62,6 +64,7 @@ def __init__(self): self._vector_tables_initialized = True self.session_maker = None self.project_id = 1 + self.scope = ProjectScope.single(1) @override async def init_search_index(self): @@ -90,6 +93,7 @@ async def search( limit: int = 10, offset: int = 0, allow_relaxed: bool = False, + session: AsyncSession | None = None, *, candidate_keys: Sequence[SearchIndexKey] | None = None, trace: SearchTraceCollector | None = None, diff --git a/tests/repository/test_postgres_search_quoted_queries.py b/tests/repository/test_postgres_search_quoted_queries.py index 40a412619..3f59f562b 100644 --- a/tests/repository/test_postgres_search_quoted_queries.py +++ b/tests/repository/test_postgres_search_quoted_queries.py @@ -5,7 +5,7 @@ import pytest -import basic_memory.repository.postgres_search_repository as postgres_search_repository_module +import basic_memory.repository.postgres_search_query as postgres_search_query_module from basic_memory.repository.postgres_search_repository import PostgresSearchRepository from basic_memory.repository.search_index_row import SearchIndexRow @@ -61,7 +61,7 @@ async def test_quoted_or_phrases_complete_without_tsquery_recovery( ) syntax_errors: list[Exception] = [] - real_is_syntax_error = postgres_search_repository_module.is_tsquery_syntax_error + real_is_syntax_error = postgres_search_query_module.is_tsquery_syntax_error def record_syntax_error(exception: Exception) -> bool: is_syntax_error = real_is_syntax_error(exception) @@ -69,9 +69,9 @@ def record_syntax_error(exception: Exception) -> bool: syntax_errors.append(exception) return is_syntax_error - # The repository module binds the classifier at import; patch it where it is read. + # PostgresFts binds the classifier at import; patch it where it is read. monkeypatch.setattr( - postgres_search_repository_module, "is_tsquery_syntax_error", record_syntax_error + postgres_search_query_module, "is_tsquery_syntax_error", record_syntax_error ) query = '"incident response" OR "database recovery"' diff --git a/tests/repository/test_postgres_search_repository.py b/tests/repository/test_postgres_search_repository.py index 308509ac7..197571fd6 100644 --- a/tests/repository/test_postgres_search_repository.py +++ b/tests/repository/test_postgres_search_repository.py @@ -14,13 +14,13 @@ import basic_memory.repository.search_repository_base as search_repository_base_module from basic_memory.repository.litellm_provider import LiteLLMEmbeddingProvider import basic_memory.repository.postgres_search_query as postgres_search_query_module -import basic_memory.repository.postgres_search_repository as postgres_search_repository_module from basic_memory.repository.postgres_search_query import ( compile_fts_filter, prepare_search_term, prepare_single_term, relaxed_tsquery_text, ) +from basic_memory.repository.search_query import PreparedSearchQuery from basic_memory.repository.postgres_search_repository import ( PostgresSearchRepository, _strip_nul_from_row, @@ -276,19 +276,22 @@ async def test_postgres_search_repository_bulk_index_items_and_prepare_terms( assert prepare_single_term(" ") == " " assert prepare_single_term("coffee", is_prefix=False) == "coffee" - indexed = compile_fts_filter(repo.scope, search_text="coffee brewing", allow_relaxed=True) + indexed = compile_fts_filter( + repo.scope, PreparedSearchQuery(search_text="coffee brewing"), allow_relaxed=True + ) assert "FROM search_index AS candidate_parent" in indexed.from_clause assert "FROM search_index_fts_chunks AS candidate_chunk" in indexed.from_clause assert "querytree(to_tsquery('english', :text))" in indexed.from_clause assert indexed.params["text_candidate"] == "coffee:* | brewing:*" filtered = compile_fts_filter( - repo.scope, search_text="coffee brewing", metadata_filters={"status": "active"} + repo.scope, + PreparedSearchQuery(search_text="coffee brewing", metadata_filters={"status": "active"}), ) assert "AS fts_candidate" in filtered.from_clause assert "JOIN entity ON search_index.entity_id = entity.id" in filtered.from_clause - negated = compile_fts_filter(repo.scope, search_text="coffee NOT brewing") + negated = compile_fts_filter(repo.scope, PreparedSearchQuery(search_text="coffee NOT brewing")) assert "AS fts_candidate" in negated.from_clause assert "FROM search_index AS candidate_all" in negated.from_clause assert negated.params["text_candidate"] == "coffee | brewing" @@ -1273,7 +1276,7 @@ async def test_postgres_relaxes_after_strict_tsquery_syntax_error( ) syntax_errors: list[Exception] = [] - real_is_syntax_error = postgres_search_repository_module.is_tsquery_syntax_error + real_is_syntax_error = postgres_search_query_module.is_tsquery_syntax_error def record_syntax_error(exc: Exception) -> bool: is_syntax_error = real_is_syntax_error(exc) @@ -1281,9 +1284,9 @@ def record_syntax_error(exc: Exception) -> bool: syntax_errors.append(exc) return is_syntax_error - # The repository module binds the classifier at import; patch it where it is read. + # PostgresFts binds the classifier at import; patch it where it is read. monkeypatch.setattr( - postgres_search_repository_module, "is_tsquery_syntax_error", record_syntax_error + postgres_search_query_module, "is_tsquery_syntax_error", record_syntax_error ) query = "foo