From 39a24212b0b85f9c880e0e87088a891da3d9021e Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Thu, 23 Jul 2026 14:32:12 +0200 Subject: [PATCH 1/2] fix: tolerate matched docs with missing field payload (Redis 8.8 expiry race) Redis 8.8 defaults the RediSearch worker pool to a nonzero background executor, exposing a race where a document expiring (TTL/HPEXPIRE) mid FT.SEARCH is returned as a matched id with a nil field array. redis-py collapses this to a Document with only id/payload, so RedisVL previously raised KeyError in the vector-normalize branch or leaked a partial record into CacheHit/ChatMessage/RouteMatch. process_results now skips such documents, detected only on zero-false positive signals (vector/range query missing vector_distance; JSON full-object unpack missing json), leaving legitimate id-only and INDEXMISSING results untouched. process_aggregate_results and the hybrid/SQL paths drop entirely-empty rows, and the semantic cache, message history, and router guard the paths the core rule does not cover. query() now returns a SearchResults list subclass exposing dropped_count and complete so callers can detect a race-shortened result set; it behaves exactly like a list otherwise. Skips are logged at WARNING level. Adds deterministic unit tests and a docs section. Co-Authored-By: Claude Opus 4.8 --- docs/concepts/queries.md | 34 +++ redisvl/extensions/cache/llm/semantic.py | 25 +- .../message_history/base_history.py | 21 +- redisvl/extensions/router/semantic.py | 27 +- redisvl/index/__init__.py | 4 +- redisvl/index/index.py | 166 +++++++++- tests/unit/test_extension_result_hardening.py | 104 +++++++ tests/unit/test_query_types.py | 286 +++++++++++++++++- 8 files changed, 645 insertions(+), 22 deletions(-) create mode 100644 tests/unit/test_extension_result_hardening.py diff --git a/docs/concepts/queries.md b/docs/concepts/queries.md index 285754b0..fb080c44 100644 --- a/docs/concepts/queries.md +++ b/docs/concepts/queries.md @@ -350,5 +350,39 @@ query = HybridQuery( ) ``` +## Results and expiring documents + +RedisVL returns query results as a list of dictionaries, one per matched document. On Redis 8.8+ a document can occasionally come back with its field payload missing: the RediSearch worker pool now defaults to a nonzero number of background worker threads, and if a document — or one of its indexed fields — expires (via TTL or `HPEXPIRE`) at the exact moment a background search is reading it, the server returns the matched id with an empty field payload instead of dropping it from the result set. This is most likely in workloads that combine search with short TTLs, such as a semantic cache or chat history. + +RedisVL detects and **silently skips** such a document (logging at `WARNING` level) whenever the query type guarantees a field would be present on a healthy match: + +- **`VectorQuery` / `VectorRangeQuery`** with `return_score=True` (the default) — a healthy match always carries a `vector_distance`. +- **`FilterQuery` on JSON storage** that returns the whole object — a healthy match always carries its JSON payload. + +For a plain **`FilterQuery` / `TextQuery` on hash storage**, a race victim is indistinguishable from a legitimately sparse result (for example, a document matched via an `INDEXMISSING` field, or a query that requested only the `id`). RedisVL cannot safely tell those apart, so it returns the document as an id-only `{"id": ...}` dict rather than risk dropping a valid result. If you query hash storage directly this way, use `.get()` for optional fields rather than assuming every field is present. + +### Detecting incomplete results + +`SearchIndex.query()` returns a `SearchResults` object, which is a normal list of result dictionaries with two extra attributes so you can tell a race-shortened result set apart from one that genuinely had fewer matches: + +- `results.dropped_count` — how many matched documents were skipped because their field payload was missing (`0` in the normal case). +- `results.complete` — `False` when any document was dropped. + +```python +results = index.query(query) + +if not results.complete: + # e.g. retry, reconcile against a CountQuery, or annotate the answer + logger.warning("Result set is incomplete: %d dropped", results.dropped_count) +``` + +`SearchResults` behaves exactly like a `list` everywhere else, so existing code needs no changes. (Operations that build a new list — slicing, `sorted()`, concatenation — return a plain `list` without these attributes.) + +Practical consequences: + +- A query may return **fewer results than `num_results`** (or fewer than `page_size` when paginating) when some matched documents were expiring. Treat those limits as upper bounds, not guarantees; use `results.complete` to detect when it happened. +- A `CountQuery` reports the server's match count, which still includes the expiring document, so a count can legitimately exceed the number of documents a materializing query returns at the same instant. +- The higher-level extensions build on this: the **semantic cache** drops an expiring hit (a benign cache miss), **message history** drops an expiring message from its formatted output, and the **semantic router** drops an expiring route candidate (which, in the rare case the best match is the one expiring, can shift the selected route). Message history's `raw=True` mode returns the unprocessed hash entries and may therefore still surface an id-only record. + **Learn more:** {doc}`/user_guide/11_advanced_queries` demonstrates these query types in detail. diff --git a/redisvl/extensions/cache/llm/semantic.py b/redisvl/extensions/cache/llm/semantic.py index 7b71e617..661beb0d 100644 --- a/redisvl/extensions/cache/llm/semantic.py +++ b/redisvl/extensions/cache/llm/semantic.py @@ -1,6 +1,7 @@ import asyncio from typing import Any +from pydantic import ValidationError from redis import Redis from redisvl.extensions.cache.llm.base import BaseLLMCache @@ -544,10 +545,28 @@ def _process_cache_results( for cache_search_result in cache_search_results: # Pop the redis key from the result redis_key = cache_search_result.pop("id") - redis_keys.append(redis_key) - # Create and process cache hit - cache_hit = CacheHit(**cache_search_result) + # Create and process cache hit. A matched entry whose field payload + # came back missing (the Redis 8.8+ background-search expiry race) + # would arrive as an id-only dict and fail validation; skip it and + # do not refresh its TTL rather than raise. The core parser already + # drops these upstream, so this is defense-in-depth. + # Only ValidationError is expected here: CacheHit has no + # before-validator that does keyed access on the input (unlike + # message history's ChatMessage.generate_id, whose guard must also + # catch KeyError). If CacheHit ever gains such a validator, widen + # this accordingly. + try: + cache_hit = CacheHit(**cache_search_result) + except ValidationError: + logger.warning( + "Skipping cache hit with missing field data (likely expired " + "during a background search on Redis 8.8+): key=%s", + redis_key, + ) + continue + + redis_keys.append(redis_key) cache_hit_dict = cache_hit.to_dict() # Filter down to only selected return fields if needed diff --git a/redisvl/extensions/message_history/base_history.py b/redisvl/extensions/message_history/base_history.py index 631e9003..8f49ab89 100644 --- a/redisvl/extensions/message_history/base_history.py +++ b/redisvl/extensions/message_history/base_history.py @@ -1,6 +1,8 @@ import warnings from typing import Any +from pydantic import ValidationError + from redisvl.extensions.constants import ( CONTENT_FIELD_NAME, METADATA_FIELD_NAME, @@ -8,8 +10,11 @@ TOOL_FIELD_NAME, ) from redisvl.extensions.message_history.schema import ChatMessage, ChatRole +from redisvl.utils.log import get_logger from redisvl.utils.utils import create_ulid, deserialize +logger = get_logger(__name__) + class BaseMessageHistory: @@ -167,7 +172,21 @@ def _format_context( for message in messages: - chat_message = ChatMessage(**message) + try: + chat_message = ChatMessage(**message) + except (ValidationError, KeyError): + # A matched message whose field payload came back missing (the + # Redis 8.8+ background-search expiry race) would arrive as an + # id-only dict and fail construction -- either a ValidationError + # for missing required fields, or a KeyError raised by the + # ChatMessage.generate_id validator when session_tag is absent. + # Skip it rather than raise. + logger.warning( + "Skipping message with missing field data (likely expired " + "during a background search on Redis 8.8+): id=%s", + message.get("id"), + ) + continue if as_text: context.append(chat_message.content) diff --git a/redisvl/extensions/router/semantic.py b/redisvl/extensions/router/semantic.py index ea9bbb6f..6ae7ea01 100644 --- a/redisvl/extensions/router/semantic.py +++ b/redisvl/extensions/router/semantic.py @@ -324,9 +324,19 @@ def get(self, route_name: str) -> Route | None: """ return next((route for route in self.routes if route.name == route_name), None) - def _process_route(self, result: dict[str, Any]) -> RouteMatch: - """Process resulting route objects and metadata.""" + def _process_route(self, result: dict[str, Any]) -> RouteMatch | None: + """Process resulting route objects and metadata. + + Returns None for a row missing ``route_name``/``distance`` — e.g. a doc + that expired mid-search on Redis 8.8+ (the background-search expiry + race) — so the caller can drop it rather than raise a KeyError. Note + this is a stricter rule than the core ``process_aggregate_results``, + which only drops entirely-empty rows; the router applies it because it + knows its exact aggregate schema. The caller aggregates the warning. + """ route_dict = make_dict(convert_bytes(result)) + if "route_name" not in route_dict or "distance" not in route_dict: + return None return RouteMatch( name=route_dict["route_name"], distance=float(route_dict["distance"]) ) @@ -409,10 +419,19 @@ def _get_route_matches( ) raise e - # process aggregation results into route matches - return [ + # process aggregation results into route matches, dropping any row whose + # field payload came back missing (the Redis 8.8+ expiry race) + processed = [ self._process_route(route_match) for route_match in aggregation_result.rows ] + matches = [match for match in processed if match is not None] + if len(matches) != len(processed): + logger.warning( + "Dropped %d route match(es) with missing field data (likely " + "expired during a background search on Redis 8.8+).", + len(processed) - len(matches), + ) + return matches def _classify_route( self, diff --git a/redisvl/index/__init__.py b/redisvl/index/__init__.py index 3218a06b..487454be 100644 --- a/redisvl/index/__init__.py +++ b/redisvl/index/__init__.py @@ -1,3 +1,3 @@ -from redisvl.index.index import AsyncSearchIndex, SearchIndex +from redisvl.index.index import AsyncSearchIndex, SearchIndex, SearchResults -__all__ = ["SearchIndex", "AsyncSearchIndex"] +__all__ = ["SearchIndex", "AsyncSearchIndex", "SearchResults"] diff --git a/redisvl/index/index.py b/redisvl/index/index.py index ac5e3982..b3ab8484 100644 --- a/redisvl/index/index.py +++ b/redisvl/index/index.py @@ -126,6 +126,78 @@ def _sql_executor_cache_key(sql_redis_options: dict[str, Any]) -> str: return json.dumps(sql_redis_options, sort_keys=True, default=repr) +class SearchResults(list): + """A list of result documents that also reports result completeness. + + This is a drop-in ``list`` — iteration, indexing, ``len()``, and every other + list operation behave exactly as before, so existing callers need no + changes. It additionally carries: + + - ``dropped_count``: the number of matched documents that were skipped + because their field payload came back missing (the Redis 8.8+ + background-search TTL/expiry race); ``0`` in the normal case. + - ``complete``: ``False`` when ``dropped_count > 0``, i.e. the result set may + be missing one or more matches that the server reported but could not + materialize. + + Callers that need completeness guarantees — audit/compliance queries, eval + harnesses, or agents making control-flow decisions — can inspect these to + detect and react to a partial result (e.g. retry, reconcile against a + ``CountQuery``, or annotate a downstream answer as incomplete). Everyone else + can ignore them. Note that operations returning a new list (slicing, + ``sorted()``, concatenation) yield a plain ``list`` without this metadata. + """ + + def __init__(self, iterable: Any = (), dropped_count: int = 0): + super().__init__(iterable) + self.dropped_count = dropped_count + + @property + def complete(self) -> bool: + """True when no matched documents were dropped from this result set.""" + return self.dropped_count == 0 + + +# Sentinel returned by the per-document processor to mark a matched document +# whose field payload came back missing (the Redis 8.8+ background-WORKERS +# TTL/expiry race). Such documents are filtered out of the final result list. +_SKIP_DOC = object() + + +def _has_missing_field_payload( + doc_dict: dict[str, Any], query: BaseQuery, unpack_json: bool +) -> bool: + """Detect a matched document whose field payload is missing. + + On Redis 8.8+ the RediSearch worker pool defaults to a nonzero background + executor. If a document key or an indexed hash field expires (TTL/HPEXPIRE) + exactly while a background ``FT.SEARCH`` is running, the server returns the + matched document id with a ``nil`` field array instead of dropping the + expired doc. redis-py collapses that ``nil`` to an empty field set, leaving a + ``Document`` whose ``__dict__`` is only ``{"id": ..., "payload": None}``. + + A race-victim doc is byte-for-byte identical to a *legitimate* id-only result + (e.g. ``FilterQuery(return_fields=["id"])``) or an ``INDEXMISSING`` sparse + doc, so we cannot key off "the dict has only id". We only report a missing + payload when the query type *guarantees* a field would be present on a + healthy match: + + - vector/range queries always project ``DISTANCE_ID`` (``vector_distance``) + when ``return_score=True`` (the default), so its absence is unambiguous; + - the JSON full-object unpack path always sees a ``"json"`` key on a healthy + match. + + Every other query passes through untouched. + """ + if unpack_json: + return "json" not in doc_dict + if isinstance(query, BaseVectorQuery) and query.DISTANCE_ID in getattr( + query, "_return_fields", [] + ): + return query.DISTANCE_ID not in doc_dict + return False + + def process_results( results: "Result", query: BaseQuery, schema: IndexSchema ) -> list[dict[str, Any]]: @@ -137,11 +209,28 @@ def process_results( unpacks the JSON object while retaining the document ID. The 'payload' field is also removed from all resulting documents for consistency. + Where it can be detected, a matched document whose field payload came back + missing is silently dropped (and logged at WARNING level). On Redis 8.8+ the + RediSearch worker pool defaults to a nonzero number of background worker + threads, and a document that expires (TTL/HPEXPIRE) exactly while a + background search runs is returned as a matched id with a ``nil`` field array + instead of being dropped server-side. This is only detectable for queries + that guarantee a field on a healthy match — vector/range queries with + ``return_score=True`` (missing ``vector_distance``) and JSON full-object + unpack (missing ``json``). For a plain hash ``FilterQuery``/``TextQuery`` a + race victim is indistinguishable from a legitimately sparse/id-only result, + so it is passed through as an id-only dict rather than dropped. + + Because of this, the returned list is best-effort and may contain fewer + items than ``num_results`` (or a paginated ``page_size``) when such docs are + present. ``CountQuery.total`` is unaffected — it reflects the server's match + count and can therefore exceed the number of materialized documents. + Args: results (Result): The search results from Redis. query (BaseQuery): The query object used for the search. - storage_type (StorageType): The storage type of the search - index (json or hash). + schema (IndexSchema): The index schema, used to determine storage type + and field metadata (e.g. the vector distance metric). Returns: List[Dict[str, Any]]: A list of processed document dictionaries. @@ -170,12 +259,24 @@ def process_results( else: norm_fn = None + # Collect ids of documents skipped because their field payload was missing + # (the Redis 8.8+ background-WORKERS expiry race); warned once per query. + skipped_ids: list[Any] = [] + # Process records - def _process(doc: "Document") -> dict[str, Any]: + def _process(doc: "Document") -> Any: doc_dict = doc.__dict__ - # Unpack and Project JSON fields properly - if unpack_json and "json" in doc_dict: + # Skip matched documents whose field payload came back missing rather + # than raising (KeyError below) or leaking a partial record downstream. + if _has_missing_field_payload(doc_dict, query, unpack_json): + skipped_ids.append(doc_dict.get("id")) + return _SKIP_DOC + + # Unpack and Project JSON fields properly. The skip guard above already + # dropped the unpack_json case where "json" is absent, so here it is + # guaranteed present. + if unpack_json: json_data = doc_dict.get("json", {}) if isinstance(json_data, str): json_data = json.loads(json_data) @@ -194,7 +295,21 @@ def _process(doc: "Document") -> dict[str, Any]: return doc_dict - return [_process(doc) for doc in results.docs] + processed = [_process(doc) for doc in results.docs] + if skipped_ids: + logger.warning( + "Skipped %d matched document(s) with missing field data (likely " + "expired during a background FT.SEARCH on Redis 8.8+). " + "query_type=%s index=%s ids=%s", + len(skipped_ids), + type(query).__name__, + schema.index.name, + skipped_ids[:10], + ) + return SearchResults( + (doc for doc in processed if doc is not _SKIP_DOC), + dropped_count=len(skipped_ids), + ) def process_aggregate_results( @@ -222,7 +337,36 @@ def _process(row): result.pop("__score", None) return result - return [_process(r) for r in results.rows] + processed = [_process(r) for r in results.rows] + kept = [r for r in processed if r] + dropped = len(processed) - len(kept) + if dropped: + logger.warning( + "Dropped %d empty aggregate row(s) with no field data.", + dropped, + ) + return SearchResults(kept, dropped_count=dropped) + + +def _convert_and_drop_empty_rows(rows: Any, source: str) -> list[dict[str, Any]]: + """Byte-decode result rows and drop any that came back entirely empty. + + Used by the hybrid (FT.HYBRID) and SQL result paths, which decode rows with + ``convert_bytes`` and perform no keyed field access (so they never raise). + An entirely-empty row carries no data and is dropped for behavioral + consistency with the FT.SEARCH/FT.AGGREGATE paths under the Redis 8.8+ + background-search expiry race. Non-empty rows are returned unchanged. + """ + converted = [convert_bytes(r) for r in rows] + kept = [r for r in converted if r] + dropped = len(converted) - len(kept) + if dropped: + logger.warning( + "Dropped %d empty %s row(s) with no field data.", + dropped, + source, + ) + return SearchResults(kept, dropped_count=dropped) class BaseSearchIndex: @@ -1009,7 +1153,7 @@ def _sql_query(self, sql_query: SQLQuery) -> list[dict[str, Any]]: # Execute the query with any params result = executor.execute(sql_query.sql, params=sql_query.params) - return [convert_bytes(row) for row in result.rows] + return _convert_and_drop_empty_rows(result.rows, "SQL") def aggregate(self, *args, **kwargs) -> "AggregateResult": """Perform an aggregation operation against the index. @@ -1184,7 +1328,7 @@ def _hybrid_search(self, query: HybridQuery, **kwargs) -> list[dict[str, Any]]: params_substitution=query.params, # type: ignore[arg-type] **kwargs, ) # type: ignore - return [convert_bytes(r) for r in results.results] # type: ignore[union-attr] + return _convert_and_drop_empty_rows(results.results, "hybrid") # type: ignore[union-attr] def batch_query( self, queries: Sequence[BaseQuery], batch_size: int = 10 @@ -2141,7 +2285,7 @@ async def _hybrid_search( params_substitution=query.params, # type: ignore[arg-type] **kwargs, ) # type: ignore - return [convert_bytes(r) for r in results.results] # type: ignore[union-attr] + return _convert_and_drop_empty_rows(results.results, "hybrid") # type: ignore[union-attr] async def batch_query( self, queries: list[BaseQuery], batch_size: int = 10 @@ -2214,7 +2358,7 @@ async def _sql_query(self, sql_query: SQLQuery) -> list[dict[str, Any]]: # Execute the query with any params asynchronously result = await executor.execute(sql_query.sql, params=sql_query.params) - return [convert_bytes(row) for row in result.rows] + return _convert_and_drop_empty_rows(result.rows, "SQL") async def query( self, query: BaseQuery | AggregationQuery | HybridQuery | SQLQuery diff --git a/tests/unit/test_extension_result_hardening.py b/tests/unit/test_extension_result_hardening.py new file mode 100644 index 00000000..a8b10a56 --- /dev/null +++ b/tests/unit/test_extension_result_hardening.py @@ -0,0 +1,104 @@ +"""Defense-in-depth unit tests for extension result parsing. + +On Redis 8.8+ a document that expires mid-search can be returned as a matched id +with a missing field payload. The core ``process_results`` parser already drops +such docs, but the extension consumers construct strict models from parsed +results and must independently tolerate an id-only / incomplete record rather +than raise. These tests exercise the extension processing methods directly. + +The processing methods under test do not use ``self``, so they are invoked +unbound with a dummy ``self`` to avoid standing up a real Redis / vectorizer. +""" + +import logging + +import pytest + +from redisvl.extensions.cache.llm.semantic import SemanticCache +from redisvl.extensions.message_history.base_history import BaseMessageHistory +from redisvl.extensions.router.semantic import SemanticRouter + + +def _valid_cache_hit(): + return { + "id": "llmcache:key1", + "entry_id": "e1", + "prompt": "hello", + "response": "world", + "vector_distance": 0.1, + "inserted_at": 1.0, + "updated_at": 1.0, + } + + +def test_process_cache_results_drops_id_only_hit(caplog): + """An id-only cache hit fails CacheHit validation and is skipped, not raised.""" + with caplog.at_level(logging.WARNING): + redis_keys, cache_hits = SemanticCache._process_cache_results( + None, [{"id": "llmcache:expired"}], None + ) + assert cache_hits == [] + # The expiring key is not returned for TTL refresh either. + assert redis_keys == [] + assert any("missing field data" in r.message for r in caplog.records) + + +def test_process_cache_results_keeps_valid_hit_and_drops_bad(): + redis_keys, cache_hits = SemanticCache._process_cache_results( + None, [_valid_cache_hit(), {"id": "llmcache:expired"}], None + ) + assert len(cache_hits) == 1 + assert redis_keys == ["llmcache:key1"] + assert cache_hits[0]["key"] == "llmcache:key1" + + +def test_process_cache_results_bad_then_valid_ordering(): + """A skipped hit's key must not leak into redis_keys before a later valid hit.""" + redis_keys, cache_hits = SemanticCache._process_cache_results( + None, [{"id": "llmcache:expired"}, _valid_cache_hit()], None + ) + assert len(cache_hits) == 1 + assert redis_keys == ["llmcache:key1"] # only the valid hit's key + + +@pytest.mark.parametrize("as_text", [True, False]) +def test_format_context_drops_id_only_message(as_text, caplog): + """A message with 'id' but no entry_id/session_tag hits the KeyError arm.""" + messages = [ + {"id": "mh:expired"}, # generate_id -> KeyError on session_tag + {"role": "user", "content": "hi", "session_tag": "s"}, + ] + with caplog.at_level(logging.WARNING): + out = BaseMessageHistory._format_context(None, messages, as_text=as_text) + assert len(out) == 1 # only the healthy message survives + assert any("missing field data" in r.message for r in caplog.records) + + +def test_format_context_drops_validation_error_message(): + """A message missing required role/content hits the ValidationError arm.""" + messages = [ + {"entry_id": "e1", "session_tag": "s"}, # no role/content -> ValidationError + {"role": "user", "content": "hi", "session_tag": "s"}, + ] + out = BaseMessageHistory._format_context(None, messages, as_text=False) + assert len(out) == 1 + assert out[0]["content"] == "hi" + + +def test_process_route_drops_incomplete_row(): + """A route row missing route_name/distance returns None rather than KeyError. + + _process_route returns None silently; the aggregated warning is emitted by + the caller (_get_route_matches), not per-row here. + """ + assert SemanticRouter._process_route(None, []) is None + assert SemanticRouter._process_route(None, ["route_name", "tech"]) is None + + +def test_process_route_keeps_valid_row(): + match = SemanticRouter._process_route( + None, ["route_name", "tech", "distance", "0.1"] + ) + assert match is not None + assert match.name == "tech" + assert match.distance == 0.1 diff --git a/tests/unit/test_query_types.py b/tests/unit/test_query_types.py index d04a2a63..78e6b795 100644 --- a/tests/unit/test_query_types.py +++ b/tests/unit/test_query_types.py @@ -1,13 +1,21 @@ +import logging + import pytest from redis import __version__ as redis_version +from redis.commands.search.document import Document from redis.commands.search.query import Query from redis.commands.search.result import Result -from redisvl.index.index import process_results +from redisvl.index.index import ( + _convert_and_drop_empty_rows, + process_aggregate_results, + process_results, +) from redisvl.query import CountQuery, FilterQuery, RangeQuery, TextQuery, VectorQuery from redisvl.query.filter import Tag from redisvl.query.query import VectorRangeQuery from redisvl.redis.connection import is_version_gte +from redisvl.schema import IndexSchema # Sample data for testing sample_vector = [0.1, 0.2, 0.3, 0.4] @@ -1140,3 +1148,279 @@ def test_vector_query_all_runtime_params(): assert params["SEARCH_WINDOW_SIZE"] == 40 assert params["USE_SEARCH_HISTORY"] == "ON" assert params["SEARCH_BUFFER_CAPACITY"] == 50 + + +# --------------------------------------------------------------------------- +# process_results: tolerate matched docs with a missing field payload +# +# Redis 8.8+ defaults the RediSearch worker pool to a nonzero background +# executor. A doc that expires (TTL/HPEXPIRE) mid-FT.SEARCH is returned as a +# matched id with a nil field array; redis-py collapses that to a Document whose +# __dict__ is only {"id": ..., "payload": None}. RedisVL must skip such docs +# rather than raise (KeyError in the vector-normalize branch) or leak a partial +# record -- WITHOUT over-skipping legitimate id-only / INDEXMISSING results. +# --------------------------------------------------------------------------- + + +def _hash_schema(): + return IndexSchema.from_dict( + { + "index": {"name": "test_hash", "prefix": "test", "storage_type": "hash"}, + "fields": [ + { + "name": "user_embedding", + "type": "vector", + "attrs": { + "dims": 4, + "distance_metric": "cosine", + "algorithm": "flat", + "datatype": "float32", + }, + }, + {"name": "brand", "type": "tag"}, + ], + } + ) + + +def _json_schema(): + return IndexSchema.from_dict( + { + "index": {"name": "test_json", "prefix": "test", "storage_type": "json"}, + "fields": [ + { + "name": "user_embedding", + "type": "vector", + "attrs": { + "dims": 4, + "distance_metric": "cosine", + "algorithm": "flat", + "datatype": "float32", + }, + }, + {"name": "brand", "type": "tag"}, + ], + } + ) + + +def test_vector_query_skips_nil_field_doc(): + """A vector match missing its always-present vector_distance is dropped.""" + nil_result = Result([1, "doc:1", None], True) + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ) + assert process_results(nil_result, query, _hash_schema()) == [] + + +def test_vector_query_normalize_skips_nil_field_doc(): + """No KeyError in the normalize branch when the distance field is absent.""" + nil_result = Result([1, "doc:1", None], True) + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + normalize_vector_distance=True, + ) + # Previously raised KeyError on doc_dict[query.DISTANCE_ID]; now skipped. + assert process_results(nil_result, query, _hash_schema()) == [] + + +def test_vector_query_normalize_keeps_healthy_doc(): + """A healthy doc carrying vector_distance is kept and normalized.""" + healthy = Result([1, "doc:1", ["vector_distance", "0.5", "brand", "Nike"]], True) + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + normalize_vector_distance=True, + ) + results = process_results(healthy, query, _hash_schema()) + assert len(results) == 1 + assert results[0]["id"] == "doc:1" + # cosine normalization: (2 - 0.5) / 2 = 0.75 + assert results[0]["vector_distance"] == "0.75" + + +def test_vector_range_query_skips_nil_field_doc(): + nil_result = Result([1, "doc:1", None], True) + query = VectorRangeQuery( + vector=sample_vector, + vector_field_name="user_embedding", + distance_threshold=0.8, + ) + assert process_results(nil_result, query, _hash_schema()) == [] + + +def test_filter_query_json_unpack_skips_nil_field_doc(): + """JSON full-object unpack: a match with no 'json' key is dropped.""" + nil_result = Result([1, "doc:1", None], True) + query = FilterQuery(Tag("brand") == "Nike") # no return_fields -> unpack_json + assert process_results(nil_result, query, _json_schema()) == [] + + +def test_filter_query_json_unpack_keeps_healthy_doc(): + healthy = Result([1, "doc:1", ["json", '{"brand": "Nike"}']], True) + query = FilterQuery(Tag("brand") == "Nike") + results = process_results(healthy, query, _json_schema()) + assert len(results) == 1 + assert results[0] == {"id": "doc:1", "brand": "Nike"} + + +def test_filter_query_return_fields_id_only_not_skipped(): + """Over-skip guard: a legitimate return_fields=['id'] doc must be KEPT. + + A race-victim doc and a legitimate id-only projection are byte-for-byte + identical ({'id':..., 'payload':None}); a naive key-set skip predicate would + drop this legitimate result. It must survive. + """ + idonly = Result([1, "doc:1", ["id", "doc:1"]], True) + query = FilterQuery(Tag("brand") == "Nike", return_fields=["id"]) + results = process_results(idonly, query, _hash_schema()) + assert len(results) == 1 + assert results[0]["id"] == "doc:1" + + +def test_filter_query_hash_id_only_not_skipped(): + """A plain hash FilterQuery id-only doc (e.g. INDEXMISSING) is passed through.""" + idonly = Result([1, "doc:1", None], True) + query = FilterQuery(Tag("brand") == "Nike", return_fields=["brand"]) + results = process_results(idonly, query, _hash_schema()) + # Not a vector query and not JSON unpack -> we cannot distinguish this from a + # legitimate sparse doc, so it is kept as an id-only dict rather than dropped. + assert results == [{"id": "doc:1"}] + + +def test_mixed_healthy_and_nil_doc(): + """In a mixed result, the healthy doc survives and the nil doc is dropped.""" + mixed = Result( + [2, "doc:1", ["vector_distance", "0.25", "brand", "Nike"], "doc:2", None], + True, + ) + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ) + results = process_results(mixed, query, _hash_schema()) + assert len(results) == 1 + assert results[0]["id"] == "doc:1" + assert results[0]["brand"] == "Nike" + + +def test_count_query_unaffected_by_nil_doc(): + """CountQuery still returns the server's total, skipping never touches it.""" + count_query = CountQuery(Tag("brand") == "Nike") + assert process_results(Result([5, "doc:1", None], True), count_query, "json") == 5 + + +def test_skip_emits_warning(caplog): + nil_result = Result([1, "doc:1", None], True) + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ) + with caplog.at_level(logging.WARNING): + process_results(nil_result, query, _hash_schema()) + assert any("missing field data" in r.message for r in caplog.records) + + +def test_hand_built_document_shapes(): + """Independence from the raw-list encoding: build Documents directly.""" + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ) + result = Result([0], "") + result.docs = [ + Document(id="doc:1", payload=None, vector_distance="0.25", brand="Nike"), + Document(id="doc:2", payload=None), # collapsed nil-field shape + ] + processed = process_results(result, query, _hash_schema()) + assert len(processed) == 1 + assert processed[0]["id"] == "doc:1" + + +class _FakeAggResult: + """Minimal AggregateResult stand-in: process_aggregate_results reads .rows.""" + + def __init__(self, rows): + self.rows = rows + + +def test_process_aggregate_skips_empty_row(): + agg = _FakeAggResult( + [["route_name", "tech", "distance", "0.1"], []] # one healthy, one empty + ) + processed = process_aggregate_results(agg, None, None) + assert len(processed) == 1 + assert processed[0]["route_name"] == "tech" + + +def test_process_aggregate_strips_score(): + agg = _FakeAggResult([["route_name", "tech", "distance", "0.1", "__score", "9"]]) + processed = process_aggregate_results(agg, None, None) + assert "__score" not in processed[0] + assert processed[0]["route_name"] == "tech" + + +def test_convert_and_drop_empty_rows_hybrid(): + rows = [{"content": "a"}, {}, [b"k", b"v"]] + kept = _convert_and_drop_empty_rows(rows, "hybrid") + # empty dict dropped; non-empty dict and decoded list-row kept + assert {"content": "a"} in kept + assert ["k", "v"] in kept + assert {} not in kept + assert len(kept) == 2 + + +def test_convert_and_drop_empty_rows_sql_warns(caplog): + with caplog.at_level(logging.WARNING): + kept = _convert_and_drop_empty_rows([{"col": "1"}, {}], "SQL") + assert kept == [{"col": "1"}] + assert any("Dropped" in r.message and "SQL" in r.message for r in caplog.records) + + +def test_search_results_signal_on_skip(): + """The returned result reports dropped_count / complete when docs are skipped.""" + mixed = Result( + [2, "doc:1", ["vector_distance", "0.25", "brand", "Nike"], "doc:2", None], + True, + ) + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ) + results = process_results(mixed, query, _hash_schema()) + # behaves like a list... + assert isinstance(results, list) + assert len(results) == 1 + assert results[0]["id"] == "doc:1" + # ...and carries completeness metadata + assert results.dropped_count == 1 + assert results.complete is False + + +def test_search_results_signal_healthy_is_complete(): + healthy = Result([1, "doc:1", ["vector_distance", "0.25", "brand", "Nike"]], True) + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ) + results = process_results(healthy, query, _hash_schema()) + assert results.dropped_count == 0 + assert results.complete is True + + +def test_search_results_signal_on_aggregate_and_rows(): + agg = process_aggregate_results( + _FakeAggResult([["route_name", "tech", "distance", "0.1"], []]), None, None + ) + assert agg.dropped_count == 1 and agg.complete is False + + rows = _convert_and_drop_empty_rows([{"a": 1}, {}], "hybrid") + assert rows.dropped_count == 1 and rows.complete is False From c554a076bf82efa2d18e08504180f25a14690882 Mon Sep 17 00:00:00 2001 From: Vishal Bala Date: Thu, 23 Jul 2026 17:25:42 +0200 Subject: [PATCH 2/2] test: make test_simple result comparison order-independent test_simple compared two independent FT.SEARCH result sets positionally. john and mary share the query vector (vector_distance == 0.0), so their relative order can differ between the two calls, making the test flaky (observed on Python 3.12 / redis-py 6.x / redis:8.4). Compare the result sets keyed by the unique `user` field instead. Applied to the sync and async twins. Co-Authored-By: Claude Opus 4.8 --- tests/integration/test_flow.py | 10 +++++++++- tests/integration/test_flow_async.py | 10 +++++++++- 2 files changed, 18 insertions(+), 2 deletions(-) diff --git a/tests/integration/test_flow.py b/tests/integration/test_flow.py index 25aedd8d..7cb2da6a 100644 --- a/tests/integration/test_flow.py +++ b/tests/integration/test_flow.py @@ -92,7 +92,15 @@ def hash_preprocess(item: dict) -> dict: assert float(users[1].vector_distance) == 0.0 assert float(users[2].vector_distance) > 0 - for doc1, doc2 in zip(results.docs, results_2): + # Compare the two result sets independent of ordering. john and mary share + # the query vector (vector_distance == 0.0), so their relative order can + # differ between the two separate FT.SEARCH calls above; key by the unique + # `user` field instead of comparing positionally. + docs_by_user = {doc.user: doc for doc in results.docs} + processed_by_user = {doc["user"]: doc for doc in results_2} + assert docs_by_user.keys() == processed_by_user.keys() + for user, doc1 in docs_by_user.items(): + doc2 = processed_by_user[user] for field in return_fields: assert getattr(doc1, field) == doc2[field] diff --git a/tests/integration/test_flow_async.py b/tests/integration/test_flow_async.py index 583dc0b3..d12181f9 100644 --- a/tests/integration/test_flow_async.py +++ b/tests/integration/test_flow_async.py @@ -95,7 +95,15 @@ def hash_preprocess(item: dict) -> dict: assert float(users[1].vector_distance) == 0.0 assert float(users[2].vector_distance) > 0 - for doc1, doc2 in zip(results.docs, results_2): + # Compare the two result sets independent of ordering. john and mary share + # the query vector (vector_distance == 0.0), so their relative order can + # differ between the two separate FT.SEARCH calls above; key by the unique + # `user` field instead of comparing positionally. + docs_by_user = {doc.user: doc for doc in results.docs} + processed_by_user = {doc["user"]: doc for doc in results_2} + assert docs_by_user.keys() == processed_by_user.keys() + for user, doc1 in docs_by_user.items(): + doc2 = processed_by_user[user] for field in return_fields: assert getattr(doc1, field) == doc2[field]