perf: add revision-pinned retrieval serving index

CONTEXT.md

@@ -173,6 +173,76 @@ The query workflow that returns cited evidence from published documents.
 The typed retrieval request that owns cache-shaping fields and route policy:
 scope, filters, data type, channels, ranking options, and agentic toggle.
 
+### Retrieval Run
+
+One execution of a Retrieval Query, recorded across classic, map-nav,
+small-corpus, cache-hit, failed, and cancelled outcomes with route, timing,
+and terminal-status metadata.
+
+### Retrieval Progress Event
+
+A safe, user-facing update about the current phase of a Retrieval run. It uses
+the fixed phases `started`, `planning`, `searching`, `reviewing_sources`, and
+`finalizing`; it never contains chain-of-thought or raw planner output.
+
+### Retrieval Stream
+
+The server-to-client event stream for a Retrieval Query. It carries
+Retrieval Progress Events during execution and one authoritative final result
+or terminal failure, while leaving answer generation to downstream clients.
+
+### Retrieval Duration
+
+The end-to-end server time for a Retrieval Query, measured from retrieval
+execution start through final public-result assembly. It includes cache lookup
+and excludes authentication, network transfer, SSE delivery time, and
+downstream answer generation.
+
+### Retrieval Non-LLM Work
+
+The database and retrieval-engine work for a Retrieval Query: snapshot or
+serving-index loading, lexical scoring, ranking, result hydration, citation
+assembly, and asset-reference resolution. It excludes planner, harvest,
+control, and answer-generation model time, which are measured separately.
+
+### Retrieval Serving Index
+
+The publication-derived read model used to load retrieval structure and
+scoring inputs without rebuilding them from the full document corpus for each
+query. It is revision-pinned and complete before its document revision becomes
+active.
+
+### Retrieval Serving Fallback
+
+The exact legacy retrieval path used when a serving index is missing,
+incomplete, or inconsistent. It preserves retrieval quality while sacrificing
+the serving-index latency target until the derived data is repaired.
+
+### Retrieval Serving Generation
+
+The namespace-scoped version that identifies one coherent set of active
+document revisions and their serving-index statistics. Retrieval captures one
+generation and retries or falls back if publication changes it during capture.
+
+### Retrieval Semantic Parity
+
+The compatibility requirement that a serving-index retrieval returns the same
+selected chunk IDs, ordering, rounded scores, citations, and asset references
+as the legacy retrieval path for the same request.
+
+### Retrieval Revision Pin
+
+The set of document revision IDs captured at retrieval start and used for the
+entire retrieval run, including lazy content and asset resolution. A later
+publication affects subsequent runs, not the run already in progress.
+
+### Online Retrieval Serving Rollout
+
+The additive rollout of retrieval-serving schema and derived data while
+retrieval and document publication remain available. Incomplete or
+inconsistent revisions use the exact legacy retrieval path until backfill and
+validation finish.
+
 ### Workflow Run Request
 
 The agentic Retrieval request passed through planning and step execution. It

apps/api/alembic/versions/2e3f4a5b6c7d_add_map_unit_token_covering_index.py

@@ -0,0 +1,40 @@
+"""Add a covering index for persisted map-nav token lookups."""
+
+from __future__ import annotations
+
+from alembic import op
+
+
+revision = "2e3f4a5b6c7d"
+down_revision = "1d2e3f4a5b6c"
+branch_labels = None
+depends_on = None
+
+_INDEX_NAME = "idx_document_map_unit_tokens_unit_lookup"
+
+
+def upgrade() -> None:
+    external_transaction = bool(
+        op.get_context().opts.get("knowhere_external_transaction", False)
+    )
+    statement = (
+        f"CREATE INDEX {{concurrently}}IF NOT EXISTS {_INDEX_NAME} "
+        "ON document_map_unit_tokens (map_unit_id, channel, token_hash) "
+        "INCLUDE (token, frequency)"
+    )
+    if external_transaction:
+        op.execute(statement.format(concurrently=""))
+        return
+    with op.get_context().autocommit_block():
+        op.execute(statement.format(concurrently="CONCURRENTLY "))
+
+
+def downgrade() -> None:
+    external_transaction = bool(
+        op.get_context().opts.get("knowhere_external_transaction", False)
+    )
+    if external_transaction:
+        op.execute(f"DROP INDEX IF EXISTS {_INDEX_NAME}")
+        return
+    with op.get_context().autocommit_block():
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {_INDEX_NAME}")

apps/api/alembic/versions/3f4a5b6c7d8e_add_section_snapshot_order_index.py

@@ -0,0 +1,41 @@
+"""Add the index used by lazy map-nav section pagination."""
+
+from __future__ import annotations
+
+from alembic import op
+
+
+revision = "3f4a5b6c7d8e"
+down_revision = "2e3f4a5b6c7d"
+branch_labels = None
+depends_on = None
+
+_INDEX_NAME = "idx_document_sections_revision_snapshot_order"
+
+
+def upgrade() -> None:
+    external_transaction = bool(
+        op.get_context().opts.get("knowhere_external_transaction", False)
+    )
+    statement = (
+        f"CREATE INDEX {{concurrently}}IF NOT EXISTS {_INDEX_NAME} "
+        "ON document_sections "
+        "(document_id, job_result_id, sort_order, section_id)"
+    )
+    if external_transaction:
+        op.execute(statement.format(concurrently=""))
+        return
+    with op.get_context().autocommit_block():
+        op.execute(statement.format(concurrently="CONCURRENTLY "))
+
+
+def downgrade() -> None:
+    external_transaction = bool(
+        op.get_context().opts.get("knowhere_external_transaction", False)
+    )
+    statement = f"DROP INDEX {{concurrently}}IF EXISTS {_INDEX_NAME}"
+    if external_transaction:
+        op.execute(statement.format(concurrently=""))
+        return
+    with op.get_context().autocommit_block():
+        op.execute(statement.format(concurrently="CONCURRENTLY "))

apps/api/alembic/versions/4a5b6c7d8e9f_add_retrieval_serving_generations.py

@@ -0,0 +1,171 @@
+"""Add revision-pinned serving manifests and namespace statistics."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "4a5b6c7d8e9f"
+down_revision: str | None = "3f4a5b6c7d8e"
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    if not sa.inspect(op.get_bind()).has_table("retrieval_namespace_generations"):
+        op.create_table(
+            "retrieval_namespace_generations",
+            sa.Column("id", sa.String(length=100), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("namespace", sa.String(length=255), nullable=False),
+            sa.Column(
+                "generation", sa.BigInteger(), nullable=False, server_default="0"
+            ),
+            sa.Column("updated_at", sa.DateTime(), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "user_id",
+                "namespace",
+                name="uq_retrieval_namespace_generations_scope",
+            ),
+        )
+    if not sa.inspect(op.get_bind()).has_table("retrieval_serving_revision_manifests"):
+        op.create_table(
+            "retrieval_serving_revision_manifests",
+            sa.Column("id", sa.String(length=100), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("namespace", sa.String(length=255), nullable=False),
+            sa.Column("document_id", sa.String(length=36), nullable=False),
+            sa.Column("job_result_id", sa.String(length=36), nullable=False),
+            sa.Column("format_version", sa.Integer(), nullable=False),
+            sa.Column("payload_zlib", sa.LargeBinary(), nullable=False),
+            sa.Column("checksum", sa.String(length=64), nullable=False),
+            sa.Column("created_at", sa.DateTime(), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["document_id"], ["documents.document_id"], ondelete="CASCADE"
+            ),
+            sa.ForeignKeyConstraint(
+                ["job_result_id"], ["job_results.id"], ondelete="CASCADE"
+            ),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "document_id",
+                "job_result_id",
+                name="uq_retrieval_serving_revision_manifests_revision",
+            ),
+        )
+    if not _index_exists("idx_retrieval_serving_revision_manifests_scope"):
+        op.create_index(
+            "idx_retrieval_serving_revision_manifests_scope",
+            "retrieval_serving_revision_manifests",
+            ["user_id", "namespace", "document_id", "job_result_id"],
+        )
+    if not sa.inspect(op.get_bind()).has_table("retrieval_serving_revision_stats"):
+        op.create_table(
+            "retrieval_serving_revision_stats",
+            sa.Column("id", sa.String(length=100), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("namespace", sa.String(length=255), nullable=False),
+            sa.Column("document_id", sa.String(length=36), nullable=False),
+            sa.Column("job_result_id", sa.String(length=36), nullable=False),
+            sa.Column("format_version", sa.Integer(), nullable=False),
+            sa.Column("payload_zlib", sa.LargeBinary(), nullable=False),
+            sa.Column("checksum", sa.String(length=64), nullable=False),
+            sa.Column("created_at", sa.DateTime(), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["document_id"], ["documents.document_id"], ondelete="CASCADE"
+            ),
+            sa.ForeignKeyConstraint(
+                ["job_result_id"], ["job_results.id"], ondelete="CASCADE"
+            ),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "document_id",
+                "job_result_id",
+                name="uq_retrieval_serving_revision_stats_revision",
+            ),
+        )
+    if not _index_exists("idx_retrieval_serving_revision_stats_scope"):
+        op.create_index(
+            "idx_retrieval_serving_revision_stats_scope",
+            "retrieval_serving_revision_stats",
+            ["user_id", "namespace", "document_id", "job_result_id"],
+        )
+    if not sa.inspect(op.get_bind()).has_table("retrieval_namespace_stats"):
+        op.create_table(
+            "retrieval_namespace_stats",
+            sa.Column("id", sa.String(length=100), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("namespace", sa.String(length=255), nullable=False),
+            sa.Column("generation", sa.BigInteger(), nullable=False),
+            sa.Column("payload_zlib", sa.LargeBinary(), nullable=False),
+            sa.Column("checksum", sa.String(length=64), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "user_id", "namespace", name="uq_retrieval_namespace_stats_scope"
+            ),
+        )
+    if not sa.inspect(op.get_bind()).has_table("retrieval_namespace_token_stats"):
+        op.create_table(
+            "retrieval_namespace_token_stats",
+            sa.Column("id", sa.String(length=100), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("namespace", sa.String(length=255), nullable=False),
+            sa.Column("generation", sa.BigInteger(), nullable=False),
+            sa.Column("channel", sa.String(length=32), nullable=False),
+            sa.Column("token_hash", sa.String(length=64), nullable=False),
+            sa.Column("document_frequency", sa.Integer(), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "user_id",
+                "namespace",
+                "channel",
+                "token_hash",
+                name="uq_retrieval_namespace_token_stats_key",
+            ),
+        )
+    if not _index_exists("idx_retrieval_namespace_token_stats_lookup"):
+        op.create_index(
+            "idx_retrieval_namespace_token_stats_lookup",
+            "retrieval_namespace_token_stats",
+            ["user_id", "namespace", "generation", "channel", "token_hash"],
+        )
+
+
+def _index_exists(index_name: str) -> bool:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    for table_name in inspector.get_table_names():
+        if any(
+            index.get("name") == index_name
+            for index in inspector.get_indexes(table_name)
+        ):
+            return True
+    return False
+
+
+def downgrade() -> None:
+    op.drop_index(
+        "idx_retrieval_namespace_token_stats_lookup",
+        table_name="retrieval_namespace_token_stats",
+        if_exists=True,
+    )
+    op.drop_table("retrieval_namespace_token_stats", if_exists=True)
+    op.drop_table("retrieval_namespace_stats", if_exists=True)
+    op.drop_index(
+        "idx_retrieval_serving_revision_stats_scope",
+        table_name="retrieval_serving_revision_stats",
+        if_exists=True,
+    )
+    op.drop_table("retrieval_serving_revision_stats", if_exists=True)
+    op.drop_index(
+        "idx_retrieval_serving_revision_manifests_scope",
+        table_name="retrieval_serving_revision_manifests",
+        if_exists=True,
+    )
+    op.drop_table("retrieval_serving_revision_manifests", if_exists=True)
+    op.drop_table("retrieval_namespace_generations", if_exists=True)

apps/api/alembic/versions/5b6c7d8e9f0a_add_term_trigram_indexes.py

@@ -0,0 +1,35 @@
+"""Add trigram acceleration for exact term-channel candidate discovery."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+
+revision: str = "5b6c7d8e9f0a"
+down_revision: str | None = "4a5b6c7d8e9f"
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+_MAP_UNIT_INDEX = "idx_document_map_units_term_trgm"
+_CHUNK_INDEX = "idx_document_chunks_term_trgm"
+
+
+def upgrade() -> None:
+    op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
+    op.execute(
+        f"CREATE INDEX IF NOT EXISTS {_MAP_UNIT_INDEX} "
+        "ON document_map_units USING gin "
+        "(term_search_text_lower gin_trgm_ops)"
+    )
+    op.execute(
+        f"CREATE INDEX IF NOT EXISTS {_CHUNK_INDEX} "
+        "ON document_chunks USING gin "
+        "(lower(COALESCE(term_search_text, '')) gin_trgm_ops)"
+    )
+
+
+def downgrade() -> None:
+    op.execute(f"DROP INDEX IF EXISTS {_CHUNK_INDEX}")
+    op.execute(f"DROP INDEX IF EXISTS {_MAP_UNIT_INDEX}")