perf(core): read FTS chunk candidates from the GIN index

src/basic_memory/repository/postgres_search_query.py

@@ -13,7 +13,7 @@
 
 import logfire
 from loguru import logger
-from sqlalchemy import text
+from sqlalchemy import Row, text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 
 from basic_memory import db
@@ -560,7 +560,8 @@ def compile_fts_filter(
     #   scope, type and key filters, as their own subquery, so the text conditions
     #   run only on the candidates. The alias keeps every later reference to
     #   ``search_index`` unchanged.
-    if from_clause != "search_index":
+    candidate_join = from_clause != "search_index"
+    if candidate_join:
         from_clause = (
             f"(SELECT search_index.* FROM {from_clause} "
             f"WHERE {' AND '.join(shared_conditions)} OFFSET 0) AS search_index"
@@ -711,12 +712,57 @@ def compile_fts_filter(
         params=params,
         order_by_clause=AFTER_DATE_ORDER_BY if query.after_date else "",
         score_expression=score_expression,
+        prefer_index_scans=candidate_join,
     )
 
 
 # --- Execution ---
 
 
+async def execute_fts_statement(
+    session: AsyncSession,
+    statement: str,
+    params: dict[str, Any],
+    *,
+    prefer_index_scans: bool,
+    use_savepoint: bool,
+) -> Sequence[Row[Any]]:
+    """Run one FTS statement and fetch its rows, steering text candidates to GIN (#1791).
+
+    Trigger: the statement joins GIN text candidates (``prefer_index_scans``).
+    Why: chunk ``tsvector``s live in TOAST, which the planner does not cost. It priced
+      a sequential scan of ``search_index_fts_chunks`` at the heap's size and chose it
+      over the GIN index, then detoasted every chunk to test it: 375 MB per query on
+      a 95k-chunk project. A GIN bitmap reads only matching rows and never detoasts;
+      it won at every selectivity measured, even with all chunks matching (921
+      buffers against 481k).
+    Outcome: sequential scans are off for this statement only. Inside a savepoint the
+      prior value is restored before the savepoint is released, and a rolled-back
+      savepoint reverts it on its own, so a caller's session plans normally afterwards.
+      Without a savepoint the session is FTS's own, and the setting ends with its
+      transaction.
+    """
+
+    async def run() -> Sequence[Row[Any]]:
+        if not prefer_index_scans:
+            return (await session.execute(text(statement), params)).fetchall()
+        previous = (
+            await session.execute(text("SELECT current_setting('enable_seqscan')"))
+        ).scalar_one()
+        await session.execute(text("SELECT set_config('enable_seqscan', 'off', true)"))
+        rows = (await session.execute(text(statement), params)).fetchall()
+        await session.execute(
+            text("SELECT set_config('enable_seqscan', :previous, true)"),
+            {"previous": previous},
+        )
+        return rows
+
+    if use_savepoint:
+        async with session.begin_nested():
+            return await run()
+    return await run()
+
+
 class PostgresFts:
     """Run tsquery statements for any scope in one database."""
 
@@ -769,12 +815,13 @@ 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()
+            return await execute_fts_statement(
+                active_session,
+                sql,
+                query_params,
+                prefer_index_scans=compiled.prefer_index_scans,
+                use_savepoint=use_savepoint,
+            )
 
         async def run_search(active_session: AsyncSession):
             relaxed = relaxed_tsquery_text(search_text) if allow_relaxed else None
@@ -900,12 +947,14 @@ async def count(
         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())
+            rows = await execute_fts_statement(
+                active_session,
+                sql,
+                query_params,
+                prefer_index_scans=compiled.prefer_index_scans,
+                use_savepoint=allow_relaxed,
+            )
+            return int(rows[0][0])
 
         try:
             async with db.scoped_session(self._session_maker) as session:

src/basic_memory/repository/search_filters.py

@@ -68,6 +68,9 @@ class CompiledFilter:
     params: dict[str, Any]
     order_by_clause: str
     score_expression: str
+    # Postgres only: the statement joins GIN text candidates, and every table it reads
+    # has an index that serves it better than a sequential scan.
+    prefer_index_scans: bool = False
 
 
 class FtsBackend(Protocol):

tests/repository/test_postgres_search_repository.py

@@ -2010,3 +2010,60 @@ async def test_postgres_text_conditions_run_only_on_candidates(session_maker, te
     ]
     assert text_filtered
     assert {node["Node Type"] for node in text_filtered} == {"Subquery Scan"}
+
+
+@pytest.mark.asyncio
+async def test_postgres_chunk_candidates_use_the_gin_index(session_maker, test_project):
+    """Text candidates come from the chunk GIN index, never a chunk table scan (#1791).
+
+    Chunk tsvectors live in TOAST, which the planner does not cost, so it chose a
+    sequential scan that detoasted every chunk: 375 MB per query on a 95k-chunk project.
+    """
+    repo = PostgresSearchRepository(session_maker, project_id=test_project.id)
+    await repo.init_search_index()
+    query = PreparedSearchQuery(
+        search_text="meeting budget review", search_item_types=[SearchItemType.ENTITY]
+    )
+    compiled = postgres_search_query_module.compile_fts_filter(
+        ProjectScope.single(test_project.id), query, allow_relaxed=True
+    )
+    assert compiled.prefer_index_scans
+    sql = f"SELECT search_index.id FROM {compiled.from_clause} WHERE {compiled.where_clause}"
+    async with db.scoped_session(session_maker) as session:
+        rows = await postgres_search_query_module.execute_fts_statement(
+            session,
+            f"EXPLAIN (FORMAT JSON) {sql}",
+            compiled.params,
+            prefer_index_scans=compiled.prefer_index_scans,
+            use_savepoint=True,
+        )
+    chunk_scans = {
+        node["Node Type"]
+        for node in _filter_nodes(rows[0][0][0]["Plan"])
+        if node.get("Relation Name") == "search_index_fts_chunks"
+    }
+    assert chunk_scans
+    assert "Seq Scan" not in chunk_scans
+
+
+@pytest.mark.asyncio
+@pytest.mark.parametrize("search_text", ["meeting budget", "broken ( query"])
+async def test_postgres_fts_leaves_the_callers_session_planning_normally(
+    session_maker, test_project, search_text
+):
+    """Turning sequential scans off is local to the FTS statement, so a caller's session
+    plans normally afterwards, whether the statement succeeded or hit a tsquery error."""
+    repo = PostgresSearchRepository(session_maker, project_id=test_project.id)
+    await repo.init_search_index()
+    fts = postgres_search_query_module.PostgresFts(session_maker)
+    async with db.scoped_session(session_maker) as session:
+        await fts.search(
+            ProjectScope.single(test_project.id),
+            PreparedSearchQuery(search_text=search_text),
+            limit=5,
+            offset=0,
+            allow_relaxed=True,
+            session=session,
+        )
+        setting = (await session.execute(text("SHOW enable_seqscan"))).scalar_one()
+    assert setting == "on"