perf(core): run FTS text conditions only on the GIN candidates

src/basic_memory/repository/postgres_search_query.py

@@ -490,9 +490,10 @@ def compile_fts_filter(
     operands too, so the strict statement and its relaxed retry read the same rows.
     """
     params: dict[str, Any] = {}
-    conditions = shared_filter_conditions(
+    shared_conditions = shared_filter_conditions(
         scope, params, dialect=POSTGRES_FILTER_DIALECT, query=query, candidate_keys=candidate_keys
     )
+    conditions: list[str] = []
     from_clause = "search_index"
     document_vector: str | None = None
     script_tsqueries: list[str] = []
@@ -549,6 +550,24 @@ def compile_fts_filter(
                 for index in range(len(script_tsqueries))
             )
 
+    # --- Fence the text candidates (#1791) ---
+    # Trigger: a GIN candidate join narrows the rows text conditions must check.
+    # Why: the planner estimates one row from the candidate UNION, so it applied the
+    #   text conditions, a per-operand EXISTS into the chunk table, to every row in
+    #   scope before joining the candidates. On an 8k-note project that was about
+    #   17k chunk probes and 700k buffer touches per query.
+    # Outcome: OFFSET 0 keeps the joined candidates, already narrowed by the cheap
+    #   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":
+        from_clause = (
+            f"(SELECT search_index.* FROM {from_clause} "
+            f"WHERE {' AND '.join(shared_conditions)} OFFSET 0) AS search_index"
+        )
+    else:
+        conditions = [*shared_conditions, *conditions]
+
     if query.title:
         params["title_text"] = prepare_search_term(query.title.strip(), is_prefix=False)
         conditions.append(

tests/repository/test_postgres_search_repository.py

@@ -1969,3 +1969,44 @@ async def test_postgres_admission_count_reads_only_the_chunk_manifest(session_ma
 
     assert _plan_relations(plan) == {"search_vector_chunks"}
     assert counted == manifest_rows > 0
+
+
+def _filter_nodes(plan: dict[str, Any]) -> list[dict[str, Any]]:
+    """Every plan node, depth first."""
+    nodes = [plan]
+    for child in plan.get("Plans", []):
+        nodes.extend(_filter_nodes(child))
+    return nodes
+
+
+@pytest.mark.asyncio
+async def test_postgres_text_conditions_run_only_on_candidates(session_maker, test_project):
+    """The costly text conditions apply to the GIN candidates, not the whole scope (#1791).
+
+    The document-vector condition probes the chunk table once per query operand. Postgres
+    pushes any condition on a plain table down into that table's scan, so without a fence
+    it ran for every row in scope before the candidate join: about 17k chunk probes per
+    query on an 8k-note project. Behind the fence it can only run above the join.
+    """
+    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
+    )
+    sql = (
+        f"SELECT search_index.id, {compiled.score_expression} AS score "
+        f"FROM {compiled.from_clause} WHERE {compiled.where_clause} "
+        "ORDER BY score DESC LIMIT 10"
+    )
+    async with db.scoped_session(session_maker) as session:
+        explained = await session.execute(text(f"EXPLAIN (FORMAT JSON) {sql}"), compiled.params)
+        plan = explained.scalar_one()[0]["Plan"]
+
+    text_filtered = [
+        node for node in _filter_nodes(plan) if "to_tsvector" in node.get("Filter", "")
+    ]
+    assert text_filtered
+    assert {node["Node Type"] for node in text_filtered} == {"Subquery Scan"}