fix(recall): make the recall budget reach the vector index (#3541)
The semantic arm asked the index for `max(limit * 5, 100)` rows, but every pool connection runs with a fixed `hnsw.ef_search = 200`. In pgvector the candidate list is the result set — the ground-layer search runs once and the scan ends when that list drains — so a scan returned at most ~200 rows however large the LIMIT above it, and the recall budget moved the SQL and nothing else (low/mid/high all got ~200). Enable hnsw.iterative_scan (strict_order) on recall connections. The drained list is refilled in ef_search-sized rounds until the query's LIMIT is met, so depth follows each query's budget with no per-query statement — which matters behind a transaction-mode pooler, where a session GUC issued between statements can land on a different backend. Retain-side link probing pins it off; it is tuned for latency, not depth. Measured on a 40k-row bank, EXPLAIN confirming an ANN index scan: MID unfiltered 200 -> 300 rows, 5.0 -> 4.9ms HIGH unfiltered 200 -> 1000 rows, 4.9 -> 6.8ms HIGH + filter 200 -> 324 rows, 5.9 -> 14.2ms +8ms worst case against a ~2.6s recall; the perf suite puts it at +1.3% mean latency / -1.3% throughput end to end. Memory is not the constraint it looks like: pgvector caps a resumed scan at work_mem * hnsw.scan_mem_multiplier, but max_scan_tuples binds first — squeezing work_mem from 64MB to 256kB changes neither rows nor latency. Two controls, both static server-level config: - HINDSIGHT_API_ANN_ITERATIVE_SCAN (default true) — the kill switch. False drops the resume GUCs rather than sending iterative_scan=off, so a connection is left exactly as it was before this existed and the revert lands on the behaviour already in production. - HINDSIGHT_API_ANN_MAX_SCAN_TUPLES (default 4000, pgvector's own is 20000) — the dial that governs the cost. The initial scan is not counted, so even 1 leaves the pre-existing depth intact; it interpolates rather than switches. Separately, the row over-fetch is deleted rather than tuned. It never did anything: each arm's rows arrive already ordered by distance, so keeping the first N of 5N returns precisely what LIMIT N would have. Invisible on pgvector, real on backends with no such bound. The LIMIT is now max(limit, GRAPH_SEED_LIMIT), since the graph arm reads its entry points from the same rows. Also fixed: a GUC the server rejects as unknown is remembered and dropped from later batches, instead of costing a failed batch plus one statement per setting on every acquire — reachable via pg_trgm on a cluster without it, and via hnsw.iterative_scan on a pgvector older than 0.8, which reserves the "hnsw." prefix and rejects it outright. Retain's link probing skips such a GUC too: it applies these with SET LOCAL inside its own transaction, where an erroring statement would abort the link computation. Not measured: whether the extra candidates improve answers. Everything above is cost.
This commit is contained in:
Nicolò Boschi
@@ -188,6 +188,17 @@ HINDSIGHT_API_LOG_LEVEL=info
|
||||
# Vector Extension (Optional - uses pgvector by default)
|
||||
# Options: "pgvector" (default), "vchord", "pgvectorscale" (DiskANN)
|
||||
# HINDSIGHT_API_VECTOR_EXTENSION=pgvector
|
||||
|
||||
# Let a vector index scan resume until the query's LIMIT is satisfied, instead of
|
||||
# stopping when its first candidate list drains (pgvector: hnsw.ef_search, 200) — with
|
||||
# it off, a larger recall budget cannot retrieve more rows. Needs pgvector 0.8.0+;
|
||||
# older servers reject it and it is dropped automatically. Set false and restart as a
|
||||
# quick revert to the previous retrieval depth, with no code change.
|
||||
# HINDSIGHT_API_ANN_ITERATIVE_SCAN=true
|
||||
# Ceiling on tuples one resumed scan may visit. Bounds the CPU and memory a selective
|
||||
# query can spend resuming (filters are applied after the scan, so it resumes often).
|
||||
# Lower it to trade depth back for latency. pgvector's own default is 20000.
|
||||
# HINDSIGHT_API_ANN_MAX_SCAN_TUPLES=4000
|
||||
# For Azure PostgreSQL with DiskANN:
|
||||
# HINDSIGHT_API_VECTOR_EXTENSION=pgvectorscale # Auto-detects pg_diskann on Azure
|
||||
|
||||
|
||||
@@ -47,13 +47,64 @@ _INDEX_TYPE_KEYWORDS = {
|
||||
"scann": "scann",
|
||||
}
|
||||
|
||||
# Ceiling on how many tuples one resumed ANN scan may visit (hnsw.max_scan_tuples).
|
||||
# Only iterative scans consult it, and it is approximate — the initial round is not
|
||||
# counted. pgvector defaults to 20000; this is deliberately lower.
|
||||
#
|
||||
# The filters that thin a semantic arm (the similarity floor, tags, date ranges) are
|
||||
# applied *after* the index scan, so a selective query resumes repeatedly to fill its
|
||||
# LIMIT. Unbounded, that turns the cheapest queries today into the most expensive:
|
||||
# ~20x the standing batch is enough to fill even a large recall budget on an
|
||||
# unfiltered query, and caps the pathological filtered case at a scan that returns
|
||||
# short — which is exactly what those queries did before iterative scans were on.
|
||||
# The GUCs that make a scan resumable — dropped wholesale when the operator turns the
|
||||
# behaviour off, so a connection is left exactly as it was before it existed (and a
|
||||
# pgvector too old to define them is never sent them either).
|
||||
_ITERATIVE_SCAN_GUCS = frozenset({"hnsw.iterative_scan", "hnsw.max_scan_tuples"})
|
||||
|
||||
|
||||
def iterative_scan_enabled() -> bool:
|
||||
"""Whether ANN scans may resume to satisfy a query's LIMIT.
|
||||
|
||||
Turning it off restores the previous depth exactly: a scan stops when its first
|
||||
candidate list drains, so no recall retrieves more rows than that list holds,
|
||||
whatever its budget.
|
||||
|
||||
Resolved through the config object rather than read from the environment, so a
|
||||
value set any other way — a CLI override applied with dataclasses.replace, a
|
||||
programmatically built config — is honoured, and the parsing and validation live
|
||||
in one place. Imported inside the function because config imports this module.
|
||||
"""
|
||||
from .config import get_config
|
||||
|
||||
return get_config().ann_iterative_scan
|
||||
|
||||
|
||||
def ann_max_scan_tuples() -> int:
|
||||
"""Ceiling on tuples one resumed scan may visit (hnsw.max_scan_tuples).
|
||||
|
||||
This is the knob that governs the cost of the behaviour. It bounds the CPU a
|
||||
selective query can spend resuming, and with it the scan's memory — pgvector
|
||||
otherwise caps that at ``work_mem * hnsw.scan_mem_multiplier``, but at this
|
||||
default the memory ceiling is never approached: squeezing work_mem to 256kB
|
||||
changes neither the rows returned nor the latency.
|
||||
|
||||
Approximate, and the initial scan is not counted, so even 1 leaves intact the
|
||||
depth a query had before scans could resume.
|
||||
"""
|
||||
from .config import get_config
|
||||
|
||||
return get_config().ann_max_scan_tuples
|
||||
|
||||
|
||||
# Per-backend ANN search-time tuning GUCs. Each entry is a tuple of
|
||||
# (guc_name, value) pairs the caller can apply with SET or SET LOCAL.
|
||||
#
|
||||
# - pgvector exposes hnsw.ef_search. The 60 / 200 pair is unchanged from the
|
||||
# pre-dispatcher code (internal benchmarks tuned around our embedding count
|
||||
# and recall floor; see the link_utils / pool init call sites for the
|
||||
# latency-vs-recall framing).
|
||||
# latency-vs-recall framing). With iterative scans on (below) the ef value is a
|
||||
# batch size rather than a ceiling, so a query's own LIMIT decides its depth.
|
||||
# - vchord exposes vchordrq.probes, but its shape must match the index's
|
||||
# build.internal.lists hierarchy. VectorChord 1.1 added per-index fallback
|
||||
# parameters for this reason: a session GUC overrides every vchordrq index,
|
||||
@@ -63,11 +114,30 @@ _INDEX_TYPE_KEYWORDS = {
|
||||
# indexes should attach probes to the index storage parameters instead.
|
||||
# - pgvectorscale / pg_diskann / scann do not expose an equivalent per-statement
|
||||
# knob in the engine today, so the dispatcher returns no statements for them.
|
||||
#
|
||||
# hnsw.iterative_scan is what makes ef_search a *batch* size rather than a ceiling.
|
||||
# With it off (pgvector's default, and what Hindsight ran until now) the ground-layer
|
||||
# search runs once and the scan ends when its list drains, so a query could never get
|
||||
# more rows than ef_search however large its LIMIT — the recall budget moved the SQL
|
||||
# and nothing else. With it on, the scan resumes in ef_search-sized rounds until the
|
||||
# LIMIT is met, so each query gets the depth it asks for with no per-query setting.
|
||||
# strict_order, not relaxed_order: the arms are trimmed in Python on the assumption
|
||||
# that rows arrive ordered by distance.
|
||||
#
|
||||
# Retain-side link probing wants the opposite — it is tuned for latency, not depth,
|
||||
# and resuming past its small candidate list would defeat that — so the low-latency
|
||||
# profile pins it off. Both profiles set it explicitly rather than relying on the
|
||||
# server default, so neither depends on what the other last left on the connection.
|
||||
_ANN_TUNING_LOW_LATENCY: dict[str, tuple[tuple[str, str], ...]] = {
|
||||
"pgvector": (("hnsw.ef_search", "60"),),
|
||||
"pgvector": (("hnsw.ef_search", "60"), ("hnsw.iterative_scan", "off")),
|
||||
}
|
||||
_ANN_TUNING_HIGH_RECALL: dict[str, tuple[tuple[str, str], ...]] = {
|
||||
"pgvector": (("hnsw.ef_search", "200"),),
|
||||
"pgvector": (
|
||||
("hnsw.ef_search", "200"),
|
||||
("hnsw.iterative_scan", "strict_order"),
|
||||
# Value filled in per call by ann_search_tuning_settings().
|
||||
("hnsw.max_scan_tuples", ""),
|
||||
),
|
||||
}
|
||||
|
||||
_EXTENSION_INSTALL_SQL = {
|
||||
@@ -167,7 +237,12 @@ def ann_search_tuning_settings(ext: str, *, kind: str) -> tuple[tuple[str, str],
|
||||
table = _ANN_TUNING_HIGH_RECALL
|
||||
else:
|
||||
raise ValueError(f"Unknown ANN tuning kind: {kind!r}")
|
||||
return table.get(_normalize_resolved(ext), ())
|
||||
settings = table.get(_normalize_resolved(ext), ())
|
||||
if not iterative_scan_enabled():
|
||||
return tuple(pair for pair in settings if pair[0] not in _ITERATIVE_SCAN_GUCS)
|
||||
return tuple(
|
||||
(name, str(ann_max_scan_tuples()) if name == "hnsw.max_scan_tuples" else value) for name, value in settings
|
||||
)
|
||||
|
||||
|
||||
def uses_per_bank_vector_indexes(ext: str) -> bool:
|
||||
|
||||
@@ -533,6 +533,8 @@ ENV_RERANKER_GOOGLE_PROJECT_ID = "HINDSIGHT_API_RERANKER_GOOGLE_PROJECT_ID"
|
||||
ENV_RERANKER_GOOGLE_SERVICE_ACCOUNT_KEY = "HINDSIGHT_API_RERANKER_GOOGLE_SERVICE_ACCOUNT_KEY"
|
||||
|
||||
ENV_VECTOR_EXTENSION = "HINDSIGHT_API_VECTOR_EXTENSION"
|
||||
ENV_ANN_ITERATIVE_SCAN = "HINDSIGHT_API_ANN_ITERATIVE_SCAN"
|
||||
ENV_ANN_MAX_SCAN_TUPLES = "HINDSIGHT_API_ANN_MAX_SCAN_TUPLES"
|
||||
ENV_TEXT_SEARCH_EXTENSION = "HINDSIGHT_API_TEXT_SEARCH_EXTENSION"
|
||||
ENV_TEXT_SEARCH_EXTENSION_NATIVE_LANGUAGE = "HINDSIGHT_API_TEXT_SEARCH_EXTENSION_NATIVE_LANGUAGE"
|
||||
ENV_TEXT_SEARCH_EXTENSION_PG_SEARCH_TOKENIZER = "HINDSIGHT_API_TEXT_SEARCH_EXTENSION_PG_SEARCH_TOKENIZER"
|
||||
@@ -1144,6 +1146,22 @@ DEFAULT_RERANKER_GOOGLE_MODEL = "semantic-ranker-default-004"
|
||||
|
||||
# Vector extension (pgvector, vchord, pgvectorscale, or AlloyDB ScaNN)
|
||||
DEFAULT_VECTOR_EXTENSION = "pgvector" # Options: "pgvector", "vchord", "pgvectorscale", "scann"
|
||||
# Let an ANN scan resume until the query's LIMIT is met, instead of stopping when its
|
||||
# first candidate list drains. Off, a recall can never retrieve more rows than the
|
||||
# candidate list holds (pgvector: hnsw.ef_search, 200), so a larger recall budget
|
||||
# widens the SQL and changes nothing. On is the intended behaviour; this exists as an
|
||||
# operational kill switch, because turning it off restores exactly the previous
|
||||
# retrieval depth without a deploy.
|
||||
DEFAULT_ANN_ITERATIVE_SCAN = True
|
||||
# Ceiling on how many tuples one resumed scan may visit. Bounds both the CPU a
|
||||
# selective query can spend resuming (the filters that thin a result are applied after
|
||||
# the index scan, so a selective one resumes repeatedly) and the scan's memory, which
|
||||
# pgvector otherwise caps at work_mem * hnsw.scan_mem_multiplier. Measured at this
|
||||
# value the memory ceiling is never approached — squeezing work_mem to 256kB changes
|
||||
# neither rows nor latency — so this is the knob that governs the cost, not work_mem.
|
||||
# Lower it to trade retrieval depth back for latency; the initial scan is not counted,
|
||||
# so even 1 leaves the pre-existing behaviour intact. pgvector's own default is 20000.
|
||||
DEFAULT_ANN_MAX_SCAN_TUPLES = 4000
|
||||
|
||||
# Text search extension (native PostgreSQL, vchord BM25, Timescale pg_textsearch,
|
||||
# pgroonga, or ParadeDB pg_search)
|
||||
@@ -2190,6 +2208,8 @@ class HindsightConfig:
|
||||
migration_database_url: str | None
|
||||
database_schema: str
|
||||
vector_extension: str # "pgvector", "vchord", "pgvectorscale", or "scann"
|
||||
ann_iterative_scan: bool
|
||||
ann_max_scan_tuples: int
|
||||
text_search_extension: str # "native", "vchord", "pg_textsearch", "pgroonga", or "pg_search"
|
||||
# PostgreSQL text search dictionary for the "native" backend (ignored by
|
||||
# other backends). Only the "native" backend reads this field; pgroonga
|
||||
@@ -3007,6 +3027,12 @@ class HindsightConfig:
|
||||
# Validate vector_extension
|
||||
validate_extension(self.vector_extension)
|
||||
|
||||
if self.ann_iterative_scan and self.ann_max_scan_tuples < 1:
|
||||
raise ValueError(
|
||||
f"Invalid ann_max_scan_tuples: {self.ann_max_scan_tuples}. Must be >= 1 when "
|
||||
f"iterative ANN scans are enabled (set {ENV_ANN_ITERATIVE_SCAN}=false to disable them)"
|
||||
)
|
||||
|
||||
# pg_trgm requires the similarity threshold in (0, 1]. Fail fast here
|
||||
# rather than let an out-of-range value raise on every pool connection's
|
||||
# setup (which would leave the API unable to serve any request).
|
||||
@@ -3175,6 +3201,10 @@ class HindsightConfig:
|
||||
migration_database_url=os.getenv(ENV_MIGRATION_DATABASE_URL) or None,
|
||||
database_schema=os.getenv(ENV_DATABASE_SCHEMA, DEFAULT_DATABASE_SCHEMA),
|
||||
vector_extension=os.getenv(ENV_VECTOR_EXTENSION, DEFAULT_VECTOR_EXTENSION).lower(),
|
||||
ann_iterative_scan=_parse_boolean_env(ENV_ANN_ITERATIVE_SCAN, DEFAULT_ANN_ITERATIVE_SCAN),
|
||||
ann_max_scan_tuples=_parse_non_negative_int(
|
||||
ENV_ANN_MAX_SCAN_TUPLES, os.getenv(ENV_ANN_MAX_SCAN_TUPLES), DEFAULT_ANN_MAX_SCAN_TUPLES
|
||||
),
|
||||
text_search_extension=os.getenv(ENV_TEXT_SEARCH_EXTENSION, DEFAULT_TEXT_SEARCH_EXTENSION).lower(),
|
||||
text_search_extension_native_language=os.getenv(
|
||||
ENV_TEXT_SEARCH_EXTENSION_NATIVE_LANGUAGE,
|
||||
|
||||
@@ -20,6 +20,24 @@ from .pool_instrumentation import PoolStats, instrument_acquire
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
# GUC names this server rejected as unknown. Process-wide and never cleared: the
|
||||
# server's extension set does not change under a running process, and re-probing
|
||||
# would reintroduce the per-acquire cost this exists to avoid.
|
||||
_unsupported_settings: set[str] = set()
|
||||
|
||||
|
||||
def setting_rejected_by_server(name: str) -> bool:
|
||||
"""Whether this server has already rejected ``name`` as an unknown GUC.
|
||||
|
||||
For callers that apply a setting outside this helper — notably retain's link
|
||||
probing, which uses SET LOCAL inside its own transaction so the value cannot leak
|
||||
onto a pooled backend. Such a caller cannot simply let the statement fail: an error
|
||||
inside a transaction poisons it, so an unknown GUC would abort its work rather than
|
||||
merely fail to apply. The pool's setup runs on acquire and names the same GUCs, so
|
||||
by the time one of those callers runs, an unknown one is already recorded here.
|
||||
"""
|
||||
return name in _unsupported_settings
|
||||
|
||||
|
||||
async def apply_session_settings(conn: asyncpg.Connection, settings: list[tuple[str, str]]) -> None:
|
||||
"""Apply session-scoped GUCs to ``conn`` in a single round trip.
|
||||
@@ -37,6 +55,7 @@ async def apply_session_settings(conn: asyncpg.Connection, settings: list[tuple[
|
||||
statement fails as a whole, so on error fall back to applying them one by
|
||||
one, skipping only the ones the server rejects.
|
||||
"""
|
||||
settings = [pair for pair in settings if pair[0] not in _unsupported_settings]
|
||||
if not settings:
|
||||
return
|
||||
|
||||
@@ -53,8 +72,19 @@ async def apply_session_settings(conn: asyncpg.Connection, settings: list[tuple[
|
||||
for name, value in settings:
|
||||
try:
|
||||
await conn.execute("SELECT set_config($1, $2, false)", name, value)
|
||||
except asyncpg.exceptions.UndefinedObjectError:
|
||||
# The server does not define this GUC — an extension we tune for is absent
|
||||
# or predates it (hnsw.iterative_scan needs pgvector 0.8+, and pgvector
|
||||
# reserves the "hnsw." prefix, so an older one rejects it rather than
|
||||
# accepting a placeholder). Remember it: otherwise every acquire from here
|
||||
# on re-pays a failed batch plus one statement per setting, which behind a
|
||||
# transaction-mode pooler is a server-side transaction each — the burn
|
||||
# #3499 removed. Narrow to UndefinedObjectError so a transient failure
|
||||
# does not disable a setting the server does support.
|
||||
logger.info("Server does not know %s — not sending it again on this process", name)
|
||||
_unsupported_settings.add(name)
|
||||
except asyncpg.exceptions.PostgresError:
|
||||
logger.debug("Could not set %s — the server may not know this setting", name)
|
||||
logger.debug("Could not set %s — retrying it on the next acquire", name)
|
||||
|
||||
|
||||
class PostgresConnection(DatabaseConnection):
|
||||
|
||||
@@ -107,8 +107,8 @@ class PostgresMemories(MemoriesExtension):
|
||||
The per-arm split is Postgres's own business, kept off the interface: this reproduces the
|
||||
exact orchestration recall used before it was unified — one dense+BM25 UNION query and the
|
||||
temporal query share a single connection, then the graph retriever runs per fact_type on the
|
||||
pool in parallel, seeded by the same dense over-fetch. Result is byte-identical to running
|
||||
the arms separately; fusion/rerank still happen downstream.
|
||||
pool in parallel, seeded by the same dense results. Result is byte-identical to running the
|
||||
arms separately; fusion/rerank still happen downstream.
|
||||
"""
|
||||
import asyncio
|
||||
|
||||
@@ -172,7 +172,7 @@ class PostgresMemories(MemoriesExtension):
|
||||
)
|
||||
|
||||
# Graph per fact_type in parallel, on the pool, after the dense connection is released —
|
||||
# seeded by the dense over-fetch (preselected_semantic_seeds), matching the prior path.
|
||||
# seeded by the dense results (preselected_semantic_seeds), matching the prior path.
|
||||
graph_by_ft: dict[str, list] = {ft: [] for ft in fact_types}
|
||||
if enable_graph:
|
||||
assert retriever is not None # only resolved when the arm is on
|
||||
@@ -228,6 +228,14 @@ class PostgresMemories(MemoriesExtension):
|
||||
min_keyword: float | None = None,
|
||||
graph_seed_min_similarity: float | None = None,
|
||||
) -> "dict[str, SemanticBm25Result]":
|
||||
"""The dense + keyword arms, as one UNION query.
|
||||
|
||||
How deep the ANN scan goes is not decided here: the connection carries
|
||||
``hnsw.iterative_scan``, which lets the scan resume until this query's own LIMIT
|
||||
is met (see ``_ANN_TUNING_HIGH_RECALL``). Before that was enabled the scan
|
||||
stopped at ``hnsw.ef_search`` rows — a fixed 200 — so a larger recall budget
|
||||
widened the SQL and changed nothing.
|
||||
"""
|
||||
# Imported here: retrieval imports this package, so a module-level import
|
||||
# would close the cycle.
|
||||
from ..search.retrieval import retrieve_semantic_bm25_combined_sql
|
||||
|
||||
@@ -17,6 +17,7 @@ from ..causal_links import (
|
||||
)
|
||||
from ..db.base import DatabaseConnection
|
||||
from ..db.ops import DataAccessOps
|
||||
from ..db.postgresql import setting_rejected_by_server
|
||||
from ..memory_engine import fq_table
|
||||
from .types import CausalRelation, EntityResolutionResult
|
||||
|
||||
@@ -593,7 +594,14 @@ async def compute_semantic_links_ann(
|
||||
# are safe to apply at session/transaction scope for the configured
|
||||
# backend. VectorChord probe values are index-shaped, so vchordrq uses
|
||||
# index storage fallback parameters instead of a blanket SET LOCAL.
|
||||
#
|
||||
# A GUC the server has already rejected is skipped rather than attempted:
|
||||
# hnsw.iterative_scan needs pgvector 0.8+, and pgvector reserves the "hnsw."
|
||||
# prefix, so an older server errors on it — which inside this transaction would
|
||||
# abort the whole link computation rather than merely fail to apply.
|
||||
for guc, value in ann_search_tuning_settings(configured_vector_extension(), kind="low_latency"):
|
||||
if setting_rejected_by_server(guc):
|
||||
continue
|
||||
await conn.execute(f"SET LOCAL {guc} = {value}")
|
||||
|
||||
t_setup = time_mod.time()
|
||||
|
||||
@@ -147,10 +147,14 @@ async def retrieve_semantic_bm25_combined_sql(
|
||||
idx_mu_emb_observation, idx_mu_emb_experience), created automatically by
|
||||
Alembic migration a3b4c5d6e7f8_add_partial_hnsw_indexes.py.
|
||||
|
||||
HNSW is approximate — semantic arms over-fetch by 5x (min 100) and trim to
|
||||
limit in Python to compensate. ef_search=200 is set globally on pool
|
||||
connections at init time (see memory_engine.py) to improve recall on sparse
|
||||
graphs.
|
||||
Each semantic arm asks for exactly ``limit`` rows. It used to ask for ``limit * 5``
|
||||
and trim back to ``limit`` in Python "to compensate for HNSW approximation", but that
|
||||
could never work: the rows arrive already ordered by distance within their arm, so
|
||||
keeping the first ``limit`` of ``limit * 5`` returns precisely what ``LIMIT limit``
|
||||
would have — the extra rows were fetched, decoded and dropped, unread. What actually
|
||||
governs ANN quality is the size of the candidate list the scan explores, which is a
|
||||
connection setting, not a row count; the caller sizes it for this query (see
|
||||
``PostgresMemories.search``) rather than over-fetching rows here.
|
||||
|
||||
fact_type values are inlined as literals (safe: they come from a controlled
|
||||
internal enum, never from user input).
|
||||
@@ -180,8 +184,17 @@ async def retrieve_semantic_bm25_combined_sql(
|
||||
sem_min = min_semantic if min_semantic is not None else config.semantic_min_similarity
|
||||
bm25_min = min_keyword if min_keyword is not None else config.bm25_min_score
|
||||
|
||||
# Over-fetch for HNSW approximation; semantic results trimmed to limit in Python.
|
||||
hnsw_fetch = max(limit * 5, 100)
|
||||
# How many semantic rows each arm must return. Two consumers read them: the semantic
|
||||
# list itself (``limit``), and — when the dense rows also clear the graph arm's
|
||||
# threshold — its entry points (``GRAPH_SEED_LIMIT``), derived from the same ordered
|
||||
# rows instead of a duplicate ANN query per fact type. A budget below GRAPH_SEED_LIMIT
|
||||
# would otherwise starve the graph arm of seeds.
|
||||
graph_seed_threshold = (
|
||||
graph_seed_min_similarity
|
||||
if graph_seed_min_similarity is not None and sem_min <= graph_seed_min_similarity
|
||||
else None
|
||||
)
|
||||
semantic_fetch = max(limit, GRAPH_SEED_LIMIT if graph_seed_threshold is not None else 0)
|
||||
|
||||
cols = (
|
||||
"id, text, context, event_date, occurred_start, occurred_end, mentioned_at, "
|
||||
@@ -199,7 +212,7 @@ async def retrieve_semantic_bm25_combined_sql(
|
||||
# $1 = query_emb_str (semantic arms)
|
||||
# $2 = bank_id
|
||||
# When tokens present:
|
||||
# $3 = limit (BM25 LIMIT; semantic uses inlined hnsw_fetch literal)
|
||||
# $3 = limit (BM25 LIMIT; semantic inlines the same limit as a literal)
|
||||
# $4 = bm25_text
|
||||
# $5 = tags (if present)
|
||||
# $6+ = tag_groups params (one per leaf)
|
||||
@@ -242,7 +255,7 @@ async def retrieve_semantic_bm25_combined_sql(
|
||||
fact_type=ft,
|
||||
embedding_param="$1",
|
||||
bank_id_param="$2",
|
||||
fetch_limit=hnsw_fetch,
|
||||
fetch_limit=semantic_fetch,
|
||||
min_similarity=sem_min,
|
||||
tags_clause=tags_clause,
|
||||
groups_clause=groups_clause,
|
||||
@@ -346,7 +359,7 @@ async def retrieve_semantic_bm25_combined_sql(
|
||||
fact_type=ft,
|
||||
embedding_param="$1",
|
||||
bank_id_param="$2",
|
||||
fetch_limit=hnsw_fetch,
|
||||
fetch_limit=semantic_fetch,
|
||||
min_similarity=sem_min,
|
||||
tags_clause=fb_tags_clause,
|
||||
groups_clause=fb_groups_clause,
|
||||
@@ -364,17 +377,7 @@ async def retrieve_semantic_bm25_combined_sql(
|
||||
else:
|
||||
raise
|
||||
|
||||
# Group results. The semantic SQL deliberately over-fetches for HNSW recall;
|
||||
# when that pool also covers the graph threshold, derive graph entry points
|
||||
# from the same ordered rows instead of issuing one duplicate ANN query per
|
||||
# fact type. Convert only the prefix either consumer can observe, not the
|
||||
# entire HNSW over-fetch pool.
|
||||
graph_seed_threshold = (
|
||||
graph_seed_min_similarity
|
||||
if graph_seed_min_similarity is not None and sem_min <= graph_seed_min_similarity
|
||||
else None
|
||||
)
|
||||
semantic_candidate_limit = max(limit, GRAPH_SEED_LIMIT if graph_seed_threshold is not None else 0)
|
||||
# Group results, converting only the prefix either consumer can observe.
|
||||
semantic_candidates: dict[str, list[RetrievalResult]] = {ft: [] for ft in fact_types}
|
||||
for r in rows:
|
||||
row = dict(r)
|
||||
@@ -383,7 +386,7 @@ async def retrieve_semantic_bm25_combined_sql(
|
||||
if ft not in result_dict:
|
||||
continue
|
||||
if source == "semantic":
|
||||
if len(semantic_candidates[ft]) < semantic_candidate_limit:
|
||||
if len(semantic_candidates[ft]) < semantic_fetch:
|
||||
semantic_candidates[ft].append(RetrievalResult.from_db_row(row))
|
||||
else:
|
||||
result_dict[ft].bm25.append(RetrievalResult.from_db_row(row))
|
||||
|
||||
@@ -374,7 +374,8 @@ class SQLDialect(ABC):
|
||||
fact_type: Fact type literal (inlined, not parameterized).
|
||||
embedding_param: Parameter placeholder for query embedding.
|
||||
bank_id_param: Parameter placeholder for bank_id.
|
||||
fetch_limit: Max rows to fetch (over-fetched for HNSW approximation).
|
||||
fetch_limit: Max rows the arm returns. Its ANN candidate list must be at
|
||||
least this wide or the scan cannot fill it — see PostgresMemories.search.
|
||||
min_similarity: Minimum cosine similarity to include.
|
||||
tags_clause: Optional WHERE clause fragment for tag filtering.
|
||||
groups_clause: Optional WHERE clause fragment for tag group filtering.
|
||||
|
||||
@@ -0,0 +1,319 @@
|
||||
"""ANN scan depth follows each query's LIMIT, instead of a fixed candidate list.
|
||||
|
||||
An ANN scan explores a bounded candidate list and returns what it found, so that list
|
||||
— not the SQL LIMIT — decided how many rows a recall arm could come back with. On
|
||||
pgvector it is ``hnsw.ef_search``, pinned at 200 for the connection's lifetime by the
|
||||
pool's init callback, which silently capped every recall at ~200 dense candidates
|
||||
however large the budget: the budget moved the SQL and nothing else.
|
||||
|
||||
``hnsw.iterative_scan`` (pgvector 0.8+) resolves that without a per-query statement.
|
||||
With it on, a drained candidate list is refilled in ``ef_search``-sized rounds until
|
||||
the query's LIMIT is met, so depth follows the budget on the connection settings the
|
||||
pool already applies — which matters behind a transaction-mode pooler, where a
|
||||
session GUC issued between statements can land on a different backend.
|
||||
|
||||
Covers:
|
||||
- Both tuning profiles: recall resumes, retain-side link probing explicitly does not.
|
||||
- That the arms fetch what both their consumers read (the semantic list, and the graph
|
||||
arm's seeds) and that ``search`` issues no session statement of its own.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import random
|
||||
import uuid
|
||||
from types import SimpleNamespace
|
||||
|
||||
import pytest
|
||||
|
||||
from hindsight_api._vector_index import ann_max_scan_tuples, ann_search_tuning_settings
|
||||
from hindsight_api.engine.memories.postgres import PostgresMemories
|
||||
from hindsight_api.engine.search import retrieval as retrieval_mod
|
||||
from hindsight_api.engine.search.link_expansion_retrieval import GRAPH_SEED_LIMIT
|
||||
|
||||
BUDGET_MID = 300
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Tuning profiles
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
def test_recall_connections_resume_the_scan():
|
||||
"""Without this the scan stops at ef_search rows and the budget cannot reach the index."""
|
||||
settings = dict(ann_search_tuning_settings("pgvector", kind="high_recall"))
|
||||
|
||||
assert settings["hnsw.iterative_scan"] == "strict_order"
|
||||
# relaxed_order would return rows out of distance order, which the Python-side
|
||||
# trim in retrieve_semantic_bm25_combined_sql assumes it can rely on.
|
||||
assert settings["hnsw.ef_search"] == "200"
|
||||
# Bounded, so a heavily filtered query cannot resume its way into a huge scan.
|
||||
assert settings["hnsw.max_scan_tuples"] == str(ann_max_scan_tuples())
|
||||
assert ann_max_scan_tuples() < 20000 # pgvector's default
|
||||
|
||||
|
||||
def test_retain_link_probing_does_not_resume():
|
||||
"""Link probing is tuned for latency; resuming past its small list would defeat that."""
|
||||
settings = dict(ann_search_tuning_settings("pgvector", kind="low_latency"))
|
||||
|
||||
assert settings["hnsw.iterative_scan"] == "off"
|
||||
assert settings["hnsw.ef_search"] == "60"
|
||||
|
||||
|
||||
def test_backends_without_the_knobs_get_no_settings():
|
||||
for ext in ("vchord", "pgvectorscale", "pg_diskann", "scann"):
|
||||
assert ann_search_tuning_settings(ext, kind="high_recall") == ()
|
||||
assert ann_search_tuning_settings(ext, kind="low_latency") == ()
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# What the arms ask for
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
class FakeDialect:
|
||||
"""Captures what each semantic arm asks the index for."""
|
||||
|
||||
def __init__(self):
|
||||
self.fetch_limits: list[int] = []
|
||||
|
||||
def build_semantic_arm(self, *, fetch_limit, **kwargs):
|
||||
self.fetch_limits.append(fetch_limit)
|
||||
return "SELECT 'semantic' AS source"
|
||||
|
||||
def build_bm25_arm(self, **kwargs):
|
||||
return "SELECT 'bm25' AS source"
|
||||
|
||||
def prepare_bm25_text(self, tokens, query_text, **kwargs):
|
||||
return " | ".join(tokens)
|
||||
|
||||
|
||||
class FakeConn:
|
||||
"""Fails the test if recall issues a session setting or opens a transaction."""
|
||||
|
||||
backend_type = "postgresql"
|
||||
|
||||
def transaction(self):
|
||||
raise AssertionError("recall must not open a transaction to tune the scan")
|
||||
|
||||
async def execute(self, sql, *params):
|
||||
raise AssertionError(f"recall must not issue session settings per query: {sql!r}")
|
||||
|
||||
async def fetch(self, query, *params):
|
||||
return []
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def search_path(monkeypatch):
|
||||
dialect = FakeDialect()
|
||||
config = SimpleNamespace(
|
||||
semantic_min_similarity=0.0,
|
||||
bm25_min_score=0.0,
|
||||
text_search_extension="native",
|
||||
text_search_extension_native_language="english",
|
||||
)
|
||||
monkeypatch.setattr(retrieval_mod, "create_sql_dialect", lambda backend: dialect)
|
||||
monkeypatch.setattr(retrieval_mod, "get_config", lambda: config)
|
||||
monkeypatch.setattr(retrieval_mod, "fq_table", lambda name: name)
|
||||
monkeypatch.setattr(retrieval_mod, "get_current_schema", lambda: None)
|
||||
return dialect
|
||||
|
||||
|
||||
async def _search(conn, limit: int, **kwargs):
|
||||
await PostgresMemories({}).search(
|
||||
conn=conn,
|
||||
bank_id="bank-1",
|
||||
fact_types=["world", "experience"],
|
||||
query_embedding="[0.0]",
|
||||
query_text="alpha beta",
|
||||
limit=limit,
|
||||
**kwargs,
|
||||
)
|
||||
|
||||
|
||||
async def test_arms_ask_for_exactly_the_rows_they_keep(search_path):
|
||||
"""No row over-fetch: the arms are ordered by distance, so trimming N of 5N in
|
||||
Python returned precisely what LIMIT N would have — the surplus was fetched,
|
||||
decoded and dropped unread."""
|
||||
await _search(FakeConn(), BUDGET_MID)
|
||||
|
||||
assert search_path.fetch_limits == [BUDGET_MID, BUDGET_MID] # one arm per fact_type
|
||||
|
||||
|
||||
async def test_small_budget_still_covers_the_graph_arms_seeds(search_path):
|
||||
"""The graph arm reads its entry points from these same rows, so a budget below
|
||||
GRAPH_SEED_LIMIT must not starve it."""
|
||||
await _search(FakeConn(), 5, graph_seed_min_similarity=0.3)
|
||||
|
||||
assert search_path.fetch_limits == [GRAPH_SEED_LIMIT, GRAPH_SEED_LIMIT]
|
||||
|
||||
|
||||
async def test_no_seed_threshold_means_no_seed_floor(search_path):
|
||||
"""With the graph arm off, nothing reads past the semantic list itself."""
|
||||
await _search(FakeConn(), 5)
|
||||
|
||||
assert search_path.fetch_limits == [5, 5]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# The property the change delivers, against a real index
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
EMBED_DIM = 384
|
||||
# Enough to exceed the 200-row candidate list at a budget of 400, and no more: this
|
||||
# runs alongside timing-sensitive tests, and a bulk load large enough to saturate the
|
||||
# database starves them.
|
||||
_ROWS = 600
|
||||
|
||||
|
||||
def _near_query_vector(seed: int) -> str:
|
||||
"""A distinct vector from one tight cluster.
|
||||
|
||||
Clustered rather than uniformly random on purpose: an HNSW graph over scattered
|
||||
vectors is sparsely connected, so a resumed scan exhausts the reachable set before
|
||||
it reaches the requested budget and the test measures graph connectivity instead of
|
||||
the setting under test.
|
||||
"""
|
||||
rng = random.Random(seed)
|
||||
values = [1.0] + [rng.uniform(-0.05, 0.05) for _ in range(EMBED_DIM - 1)]
|
||||
norm = sum(v * v for v in values) ** 0.5
|
||||
return "[" + ",".join(f"{v / norm:.5f}" for v in values) + "]"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_the_kill_switch_flips_real_retrieval_depth(memory, request_context, ann_config):
|
||||
"""End to end, through the pool: on, the budget reaches the index; off, it does not.
|
||||
|
||||
Both halves matter. On is the fix — with iterative scans off the ground-layer search
|
||||
runs once and the scan ends when its ef_search-sized list drains, so the arm cannot
|
||||
return more than ~200 rows however large the recall budget. Off is the operational
|
||||
revert, and it has to land on exactly that pre-existing behaviour rather than some
|
||||
third state nobody runs.
|
||||
|
||||
Driven by the environment variable through the pool's own session setup, not by
|
||||
setting the GUCs by hand, so it covers the path production actually takes. Rows and
|
||||
index are built directly: the property belongs to the index scan, and going through
|
||||
retain would drag in extraction and consolidation.
|
||||
"""
|
||||
from hindsight_api.engine.search.retrieval import retrieve_semantic_bm25_combined_sql
|
||||
from hindsight_api.engine.retain.bank_utils import get_or_create_bank_profile
|
||||
from hindsight_api.engine.task_backend import fq_table
|
||||
|
||||
bank_id = f"test_iter_scan_{uuid.uuid4().hex[:8]}"
|
||||
budget = 400 # deliberately above the standing ef_search of 200
|
||||
# Creating the bank also builds its per-(bank, fact_type) partial vector index —
|
||||
# the same one recall uses — so this exercises the production index, not a stand-in.
|
||||
await get_or_create_bank_profile(memory._backend, bank_id)
|
||||
pool = await memory._get_pool()
|
||||
probe = _near_query_vector(0)
|
||||
table = fq_table("memory_units")
|
||||
try:
|
||||
async with pool.acquire() as conn:
|
||||
await conn.executemany(
|
||||
f"INSERT INTO {table} (bank_id, text, fact_type, embedding) VALUES ($1, $2, 'world', $3::vector)",
|
||||
[(bank_id, f"filler fact {i}", _near_query_vector(i)) for i in range(_ROWS)],
|
||||
)
|
||||
await conn.execute(f"ANALYZE {table}")
|
||||
|
||||
async def semantic_rows(iterative: bool) -> int:
|
||||
ann_config("ann_iterative_scan", iterative)
|
||||
# The pool re-applies its session settings on every acquire, so a fresh
|
||||
# connection resolves the flag again rather than inheriting the value the
|
||||
# process started with.
|
||||
async with pool.acquire() as conn:
|
||||
# The property under test belongs to the ANN scan, not to the planner's
|
||||
# choice: on a table this size a full scan plus a sort is genuinely
|
||||
# cheaper, and inflating the fixture until ANN wins would only make the
|
||||
# test slow. Discourage both alternatives so the ordered path is taken.
|
||||
await conn.execute("SET enable_seqscan = off")
|
||||
await conn.execute("SET enable_sort = off")
|
||||
plan = "\n".join(
|
||||
r[0]
|
||||
for r in await conn.fetch(
|
||||
f"EXPLAIN SELECT id FROM {table} WHERE bank_id = $1 AND fact_type = 'world' "
|
||||
f"AND embedding IS NOT NULL ORDER BY embedding <=> $2::vector LIMIT {budget}",
|
||||
bank_id,
|
||||
probe,
|
||||
)
|
||||
)
|
||||
# "Index Scan" alone is not enough — a btree scan plus a Sort also
|
||||
# matches, and it returns every row regardless of the candidate list,
|
||||
# which would make this quietly measure nothing. An ANN scan emits rows
|
||||
# already ordered, so the giveaway is the absence of a Sort node.
|
||||
assert "Index Scan" in plan and "Sort" not in plan, f"expected an ANN scan, got:\n{plan}"
|
||||
result = await retrieve_semantic_bm25_combined_sql(
|
||||
conn, probe, "", bank_id, ["world"], budget, min_semantic=0.0
|
||||
)
|
||||
return len(result["world"].semantic)
|
||||
|
||||
with_resume = await semantic_rows(True)
|
||||
without_resume = await semantic_rows(False)
|
||||
|
||||
# On: the budget reaches the index.
|
||||
assert with_resume == budget, f"expected the full budget, got {with_resume}"
|
||||
# Off: capped by the candidate list, exactly as before the fix existed.
|
||||
assert without_resume <= 250, f"expected the scan to stop at ~ef_search, got {without_resume}"
|
||||
assert without_resume < with_resume
|
||||
finally:
|
||||
await memory.delete_bank(bank_id, request_context=request_context)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Operational controls
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def ann_config(monkeypatch):
|
||||
"""Override an ANN config field for one test, without disturbing anything else.
|
||||
|
||||
Set on the cached config instance rather than by setting the env var and clearing
|
||||
the cache: clearing it is process-wide, so every engine built earlier in the
|
||||
session would silently start resolving a config rebuilt from the current
|
||||
environment. monkeypatch restores the attribute at teardown.
|
||||
"""
|
||||
from hindsight_api.config import _get_raw_config
|
||||
|
||||
def _set(field: str, value) -> None:
|
||||
monkeypatch.setattr(_get_raw_config(), field, value)
|
||||
|
||||
return _set
|
||||
|
||||
|
||||
def test_the_kill_switch_removes_the_resume_settings(ann_config):
|
||||
"""Turning it off must leave a connection exactly as it was before the feature.
|
||||
|
||||
Dropping the GUCs rather than sending iterative_scan=off matters for two reasons:
|
||||
a pgvector older than 0.8 rejects them outright (it reserves the "hnsw." prefix),
|
||||
and an operator who pinned values server-side keeps them.
|
||||
"""
|
||||
ann_config("ann_iterative_scan", False)
|
||||
settings = ann_search_tuning_settings("pgvector", kind="high_recall")
|
||||
|
||||
assert settings == (("hnsw.ef_search", "200"),)
|
||||
|
||||
|
||||
def test_the_scan_ceiling_is_tunable(ann_config):
|
||||
"""The dial between the previous behaviour and full budget depth."""
|
||||
ann_config("ann_max_scan_tuples", 1500)
|
||||
settings = dict(ann_search_tuning_settings("pgvector", kind="high_recall"))
|
||||
|
||||
assert settings["hnsw.max_scan_tuples"] == "1500"
|
||||
assert settings["hnsw.iterative_scan"] == "strict_order"
|
||||
|
||||
|
||||
def test_an_unreadable_ceiling_is_rejected_at_config_load(monkeypatch):
|
||||
"""Parsing and validation belong to HindsightConfig, not to this module."""
|
||||
from hindsight_api.config import HindsightConfig
|
||||
|
||||
monkeypatch.setenv("HINDSIGHT_API_ANN_MAX_SCAN_TUPLES", "not-a-number")
|
||||
with pytest.raises(ValueError):
|
||||
HindsightConfig.from_env()
|
||||
|
||||
|
||||
def test_retain_probing_is_unaffected_by_the_switch(ann_config):
|
||||
"""Link probing never resumed; the switch has nothing to take from it."""
|
||||
ann_config("ann_iterative_scan", False)
|
||||
settings = dict(ann_search_tuning_settings("pgvector", kind="low_latency"))
|
||||
|
||||
assert settings == {"hnsw.ef_search": "60"}
|
||||
@@ -13,6 +13,7 @@ import pytest
|
||||
|
||||
from hindsight_api.engine.db import DatabaseBackend, DatabaseConnection, create_database_backend
|
||||
from hindsight_api.engine.db.ops import UpdatedWindow
|
||||
from hindsight_api.engine.db import postgresql as pg_backend
|
||||
from hindsight_api.engine.db.postgresql import PostgreSQLBackend, apply_session_settings
|
||||
from hindsight_api.engine.db.result import DictResultRow as ResultRow
|
||||
from hindsight_api.engine.sql import SQLDialect, create_sql_dialect
|
||||
@@ -645,6 +646,13 @@ class _RecordingConnection:
|
||||
class TestApplySessionSettings:
|
||||
"""The pool's setup callback runs on every acquire — it must be one round trip (#3499)."""
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _forget_rejected_settings(self):
|
||||
"""The rejected-GUC memo is process-wide, so it must not leak between tests."""
|
||||
pg_backend._unsupported_settings.clear()
|
||||
yield
|
||||
pg_backend._unsupported_settings.clear()
|
||||
|
||||
_SETTINGS = [
|
||||
("hnsw.ef_search", "200"),
|
||||
("statement_timeout", "600s"),
|
||||
@@ -684,6 +692,44 @@ class TestApplySessionSettings:
|
||||
# The rejected one raised and was skipped rather than aborting setup.
|
||||
assert len(conn.calls) == 1 + len(self._SETTINGS)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_setting_the_server_rejects_is_not_sent_again(self):
|
||||
"""Otherwise every acquire re-pays a failed batch plus one statement per setting.
|
||||
|
||||
Reached by any GUC the cluster does not define — pg_trgm when the extension is
|
||||
absent, or hnsw.iterative_scan on a pgvector older than 0.8, which reserves the
|
||||
"hnsw." prefix and so rejects it rather than accepting a placeholder.
|
||||
"""
|
||||
conn = _RecordingConnection(fail_batched=True, reject="pg_trgm.similarity_threshold")
|
||||
await apply_session_settings(conn, self._SETTINGS)
|
||||
|
||||
# Next acquire: one batched statement again, carrying only what the server took.
|
||||
conn = _RecordingConnection()
|
||||
await apply_session_settings(conn, self._SETTINGS)
|
||||
|
||||
assert len(conn.calls) == 1
|
||||
_, args = conn.calls[0]
|
||||
assert "pg_trgm.similarity_threshold" not in args
|
||||
assert args == ("hnsw.ef_search", "200", "statement_timeout", "600s")
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_a_transient_failure_does_not_disable_a_setting(self):
|
||||
"""Only "unrecognized configuration parameter" is permanent; anything else retries."""
|
||||
|
||||
class _FlakyConnection(_RecordingConnection):
|
||||
async def execute(self, query: str, *args) -> None:
|
||||
self.calls.append((query, args))
|
||||
if len(self.calls) == 1:
|
||||
raise asyncpg.exceptions.UndefinedObjectError("unrecognized configuration parameter")
|
||||
if "hnsw.ef_search" in args:
|
||||
raise asyncpg.exceptions.DeadlockDetectedError("transient")
|
||||
|
||||
await apply_session_settings(_FlakyConnection(), self._SETTINGS)
|
||||
|
||||
conn = _RecordingConnection()
|
||||
await apply_session_settings(conn, self._SETTINGS)
|
||||
assert "hnsw.ef_search" in conn.calls[0][1]
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Config integration test
|
||||
|
||||
@@ -444,6 +444,35 @@ class TestComputeSemanticLinksAnnPgBouncerSafety:
|
||||
# And there must not be a RESET — SET LOCAL handles it at commit.
|
||||
assert not any(f"RESET {guc}" in s for s in executed_sql)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_skips_a_guc_the_server_has_rejected(self, mock_conn, monkeypatch):
|
||||
"""An unknown GUC must not be attempted inside this transaction.
|
||||
|
||||
hnsw.iterative_scan needs pgvector 0.8+, and pgvector reserves the "hnsw."
|
||||
prefix, so an older server errors on it rather than accepting a placeholder —
|
||||
and an error inside an open transaction aborts the whole link computation, not
|
||||
just the setting. The pool's session setup names the same GUCs on acquire, so
|
||||
by the time this runs an unknown one is already recorded.
|
||||
"""
|
||||
from hindsight_api.engine.db import postgresql as pg_backend
|
||||
|
||||
monkeypatch.setenv("HINDSIGHT_API_VECTOR_EXTENSION", "pgvector")
|
||||
monkeypatch.setattr(pg_backend, "_unsupported_settings", {"hnsw.iterative_scan"})
|
||||
|
||||
await compute_semantic_links_ann(
|
||||
conn=mock_conn,
|
||||
bank_id="bank-1",
|
||||
unit_ids=["u1"],
|
||||
embeddings=[[0.1] * 384],
|
||||
fact_types=["world"],
|
||||
threshold=DEFAULT_SEMANTIC_LINK_MIN_SIMILARITY,
|
||||
)
|
||||
|
||||
executed_sql = [call.args[0] for call in mock_conn.execute.call_args_list]
|
||||
assert not any("hnsw.iterative_scan" in s for s in executed_sql)
|
||||
# The supported one is still applied.
|
||||
assert any("hnsw.ef_search" in s for s in executed_sql)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_vchord_ann_does_not_set_fixed_probe_count(self, mock_conn, monkeypatch):
|
||||
"""VectorChord probe counts must come from index/default config.
|
||||
|
||||
@@ -98,8 +98,12 @@ def test_scann_index_creation_defers_until_table_is_large_enough():
|
||||
|
||||
|
||||
def test_ann_search_tuning_settings_pgvector_dispatches_hnsw_ef_search():
|
||||
assert ann_search_tuning_settings("pgvector", kind="low_latency") == (("hnsw.ef_search", "60"),)
|
||||
assert ann_search_tuning_settings("pgvector", kind="high_recall") == (("hnsw.ef_search", "200"),)
|
||||
# Looked up rather than compared as a whole tuple: the profiles also carry the
|
||||
# iterative-scan settings, which test_ann_iterative_scan.py covers. Pinning the
|
||||
# exact tuple here would make every future addition to a profile fail this test
|
||||
# for a reason it is not about.
|
||||
assert dict(ann_search_tuning_settings("pgvector", kind="low_latency"))["hnsw.ef_search"] == "60"
|
||||
assert dict(ann_search_tuning_settings("pgvector", kind="high_recall"))["hnsw.ef_search"] == "200"
|
||||
|
||||
|
||||
def test_ann_search_tuning_settings_vchord_leaves_probes_to_index_defaults():
|
||||
|
||||
@@ -77,6 +77,8 @@ hindsight-admin run-db-migration --schema tenant_acme
|
||||
| Variable | Description | Default |
|
||||
|----------|-------------|---------|
|
||||
| `HINDSIGHT_API_VECTOR_EXTENSION` | Vector index algorithm: `pgvector`, `vchord`, `pgvectorscale`, or `scann` | `pgvector` |
|
||||
| `HINDSIGHT_API_ANN_ITERATIVE_SCAN` | Let a vector index scan resume until the query's `LIMIT` is satisfied, instead of stopping when its first candidate list drains. With it off, a recall can never retrieve more rows than that list holds — on pgvector, `hnsw.ef_search` (200) — so a larger recall budget widens the SQL and retrieves nothing extra. Requires pgvector 0.8.0+; older servers reject the setting and it is dropped automatically after the first attempt. This is the operational kill switch: setting it to `false` and restarting restores the previous retrieval depth exactly, with no code change. | `true` |
|
||||
| `HINDSIGHT_API_ANN_MAX_SCAN_TUPLES` | Ceiling on how many tuples a single resumed scan may visit (`hnsw.max_scan_tuples`). This is the knob that governs what iterative scans cost: the filters that thin a result — the similarity floor, tags, date ranges — are applied *after* the index scan, so a selective query resumes repeatedly, and this bounds both the CPU it can spend and the memory it can hold (pgvector otherwise caps the latter at `work_mem × hnsw.scan_mem_multiplier`, which at this default is never approached). Lower it to trade retrieval depth back for latency; the initial scan is not counted, so even `1` leaves the pre-existing depth intact. pgvector's own default is `20000`. Ignored when iterative scans are off. | `4000` |
|
||||
|
||||
Hindsight supports four PostgreSQL vector extensions:
|
||||
|
||||
|
||||
@@ -374,7 +374,7 @@ This recall budget flows through the pipeline as follows:
|
||||
|
||||
| Pipeline stage | How the recall budget is used |
|
||||
|----------------|-------------------------------|
|
||||
| **Semantic search** | Over-fetches max(recall_budget × 5, 100) from HNSW, trims to recall_budget |
|
||||
| **Semantic search** | `LIMIT recall_budget` in SQL, with the ANN candidate list sized to match for the query (on pgvector, `hnsw.ef_search`, capped at that setting's maximum of 1000) |
|
||||
| **BM25 search** | `LIMIT recall_budget` in SQL |
|
||||
| **Graph traversal** | Explores up to recall_budget nodes |
|
||||
| **Temporal spreading** | Activates up to recall_budget nodes via links |
|
||||
|
||||
@@ -188,6 +188,17 @@ HINDSIGHT_API_LOG_LEVEL=info
|
||||
# Vector Extension (Optional - uses pgvector by default)
|
||||
# Options: "pgvector" (default), "vchord", "pgvectorscale" (DiskANN)
|
||||
# HINDSIGHT_API_VECTOR_EXTENSION=pgvector
|
||||
|
||||
# Let a vector index scan resume until the query's LIMIT is satisfied, instead of
|
||||
# stopping when its first candidate list drains (pgvector: hnsw.ef_search, 200) — with
|
||||
# it off, a larger recall budget cannot retrieve more rows. Needs pgvector 0.8.0+;
|
||||
# older servers reject it and it is dropped automatically. Set false and restart as a
|
||||
# quick revert to the previous retrieval depth, with no code change.
|
||||
# HINDSIGHT_API_ANN_ITERATIVE_SCAN=true
|
||||
# Ceiling on tuples one resumed scan may visit. Bounds the CPU and memory a selective
|
||||
# query can spend resuming (filters are applied after the scan, so it resumes often).
|
||||
# Lower it to trade depth back for latency. pgvector's own default is 20000.
|
||||
# HINDSIGHT_API_ANN_MAX_SCAN_TUPLES=4000
|
||||
# For Azure PostgreSQL with DiskANN:
|
||||
# HINDSIGHT_API_VECTOR_EXTENSION=pgvectorscale # Auto-detects pg_diskann on Azure
|
||||
|
||||
|
||||
@@ -77,6 +77,8 @@ hindsight-admin run-db-migration --schema tenant_acme
|
||||
| Variable | Description | Default |
|
||||
|----------|-------------|---------|
|
||||
| `HINDSIGHT_API_VECTOR_EXTENSION` | Vector index algorithm: `pgvector`, `vchord`, `pgvectorscale`, or `scann` | `pgvector` |
|
||||
| `HINDSIGHT_API_ANN_ITERATIVE_SCAN` | Let a vector index scan resume until the query's `LIMIT` is satisfied, instead of stopping when its first candidate list drains. With it off, a recall can never retrieve more rows than that list holds — on pgvector, `hnsw.ef_search` (200) — so a larger recall budget widens the SQL and retrieves nothing extra. Requires pgvector 0.8.0+; older servers reject the setting and it is dropped automatically after the first attempt. This is the operational kill switch: setting it to `false` and restarting restores the previous retrieval depth exactly, with no code change. | `true` |
|
||||
| `HINDSIGHT_API_ANN_MAX_SCAN_TUPLES` | Ceiling on how many tuples a single resumed scan may visit (`hnsw.max_scan_tuples`). This is the knob that governs what iterative scans cost: the filters that thin a result — the similarity floor, tags, date ranges — are applied *after* the index scan, so a selective query resumes repeatedly, and this bounds both the CPU it can spend and the memory it can hold (pgvector otherwise caps the latter at `work_mem × hnsw.scan_mem_multiplier`, which at this default is never approached). Lower it to trade retrieval depth back for latency; the initial scan is not counted, so even `1` leaves the pre-existing depth intact. pgvector's own default is `20000`. Ignored when iterative scans are off. | `4000` |
|
||||
|
||||
Hindsight supports four PostgreSQL vector extensions:
|
||||
|
||||
|
||||
@@ -374,7 +374,7 @@ This recall budget flows through the pipeline as follows:
|
||||
|
||||
| Pipeline stage | How the recall budget is used |
|
||||
|----------------|-------------------------------|
|
||||
| **Semantic search** | Over-fetches max(recall_budget × 5, 100) from HNSW, trims to recall_budget |
|
||||
| **Semantic search** | `LIMIT recall_budget` in SQL, with the ANN candidate list sized to match for the query (on pgvector, `hnsw.ef_search`, capped at that setting's maximum of 1000) |
|
||||
| **BM25 search** | `LIMIT recall_budget` in SQL |
|
||||
| **Graph traversal** | Explores up to recall_budget nodes |
|
||||
| **Temporal spreading** | Activates up to recall_budget nodes via links |
|
||||
|
||||
Reference in New Issue
Block a user