diff --git a/docs/api/searchindex.rst b/docs/api/searchindex.rst index 6080d2744..826becf38 100644 --- a/docs/api/searchindex.rst +++ b/docs/api/searchindex.rst @@ -12,6 +12,8 @@ Search Index Classes - Primary class to write, read, and search across data structures in Redis. * - :ref:`asyncsearchindex_api` - Async version of the SearchIndex to write, read, and search across data structures in Redis. + * - :ref:`searchresults_api` + - List of result documents returned by a query, which also reports result completeness. .. _searchindex_api: @@ -34,3 +36,13 @@ AsyncSearchIndex .. autoclass:: AsyncSearchIndex :inherited-members: :members: + +.. _searchresults_api: + +SearchResults +============= + +.. currentmodule:: redisvl.index + +.. autoclass:: SearchResults + :members: diff --git a/docs/concepts/queries.md b/docs/concepts/queries.md index fb080c442..9f6354241 100644 --- a/docs/concepts/queries.md +++ b/docs/concepts/queries.md @@ -382,6 +382,7 @@ 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. +- **`paginate()` keeps going past a page it could not materialize.** If every match on one page was expiring, that page is skipped and iteration continues to the end of the offset range rather than stopping early. An empty batch is never yielded, and a skipped page's `dropped_count` is added to the next batch you receive, so `results.complete` stays meaningful across the whole stream. Two caveats: a result set whose *trailing* pages were entirely dropped has no later batch to report on, so those drops appear only in the logged warning; and paging is only stable if the query sorts on a unique field — Redis documents `LIMIT` without sorting as non-deterministic, which can duplicate or miss documents independently of expiry. - 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/index/index.py b/redisvl/index/index.py index 1c690d475..171d76ad4 100644 --- a/redisvl/index/index.py +++ b/redisvl/index/index.py @@ -330,7 +330,16 @@ def _has_missing_field_payload( match. Every other query passes through untouched. + + A ``NOCONTENT`` query breaks both guarantees: the server then returns "the + document ids and not the content" for *every* healthy match (``RETURN 0`` acts + the same way), so no field payload is expected and its absence says nothing. + redis-py's ``Query.no_content()`` sets only ``_no_content`` and leaves + ``_return_fields`` untouched, so neither predicate above notices -- without + this short-circuit every healthy document would be reported as missing. """ + if getattr(query, "_no_content", False): + return False if unpack_json: return "json" not in doc_dict if isinstance(query, BaseVectorQuery) and query.DISTANCE_ID in getattr( @@ -426,7 +435,10 @@ def _process(doc: "Document") -> Any: return {"id": doc_dict.get("id"), **json_data} raise ValueError(f"Unable to parse json data from Redis {json_data}") - if norm_fn: + # The skip guard above guarantees the distance is present for a normal + # vector query, but not for a NOCONTENT one, where the server returns ids + # only and there is no distance to normalize. + if norm_fn and query.DISTANCE_ID in doc_dict: # type: ignore # convert float back to string to be consistent doc_dict[query.DISTANCE_ID] = str( # type: ignore norm_fn(float(doc_dict[query.DISTANCE_ID])) # type: ignore @@ -515,6 +527,49 @@ def _convert_and_drop_empty_rows(rows: Any, source: str) -> list[dict[str, Any]] return SearchResults(kept, dropped_count=dropped) +def _page_had_matches(results: Any) -> bool: + """Whether the server reported any match for one page of paginated results. + + ``paginate`` must not treat "empty page" as "result set exhausted". A page + comes back empty for two very different reasons: + + 1. the server reported no more matches — iteration is genuinely done; + 2. the server reported matches whose field payload could not be + materialized, so ``process_results`` dropped them -- a key that expires or + is updated mid-query is returned as a matched id with a ``nil`` field + array, and is still counted in the server's total (see + ``_has_missing_field_payload``). + + Stopping on case 2 silently truncates iteration and discards every remaining + page. ``SearchResults.dropped_count`` tells the two apart: a page had matches + when it either yielded documents or dropped some. + + Truthiness (not ``len()``) is deliberate, and ``getattr`` is defensive: every + query path returns ``SearchResults`` today, but a subclass or test double may + substitute a plain ``list``, and ``process_results`` returns a bare ``int`` + for a ``CountQuery``. Neither should raise here. + """ + return bool(results) or bool(getattr(results, "dropped_count", 0)) + + +def _fold_carried_drops(results: Any, carried: int) -> None: + """Add drops carried over from skipped pages to this page's ``dropped_count``. + + ``paginate`` never yields an empty batch, so a page whose matches were *all* + dropped would otherwise carry its ``dropped_count`` out of the stream and + leave the remaining batches reporting ``complete is True`` — the same silent + incompleteness the drop accounting exists to surface. Folding the count into + the next yielded batch keeps the signal reachable from the batches a caller + actually sees, without breaking the non-empty-batch guarantee. + + Note the residual gap: if the *trailing* pages of a result set are entirely + dropped there is no subsequent batch to fold into, and those drops are + reported only by the ``process_results`` warning. + """ + if carried and isinstance(results, SearchResults): + results.dropped_count += carried + + class BaseSearchIndex: """Base search engine class""" @@ -1914,7 +1969,7 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: batch. Defaults to 30. Yields: - A generator yielding batches of search results. + A generator yielding non-empty ``SearchResults`` batches. Raises: TypeError: If the page_size argument is not of type int. @@ -1933,8 +1988,21 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: considerations and the expected volume of search results. Note: - For stable pagination, the query must have a `sort_by` clause. + For stable pagination, the query must have a `sort_by` clause on a + **unique** field. Redis documents that ``LIMIT`` without sorting is + non-deterministic, so pages may otherwise repeat or miss documents. + Very deep pagination is also bounded server-side by + ``search-max-search-results`` (1,000,000 by default, but 10,000 on + some managed tiers), past which the search errors rather than ending. + Note: + A yielded batch may contain fewer than ``page_size`` documents, and an + empty batch is never yielded. On Redis 8+ a matched document whose + data expires while the search is running is skipped; each batch + reports how many were skipped via ``dropped_count`` (and + ``complete``), including skips inherited from a page that was dropped + in its entirety. Pagination still runs to the end of the result set in + that case. """ if not isinstance(page_size, int): raise TypeError("page_size must be an integer") @@ -1942,14 +2010,29 @@ def paginate(self, query: BaseQuery, page_size: int = 30) -> Generator: if page_size <= 0: raise ValueError("page_size must be greater than 0") + if isinstance(query, CountQuery): + raise TypeError( + "CountQuery cannot be paginated: it returns a match count rather " + "than documents. Use index.query(query) instead." + ) + offset = 0 + carried_drops = 0 while True: query.paging(offset, page_size) results = self._query(query) - if not results: + if not _page_had_matches(results): break - yield results - # Increment the offset for the next batch of pagination + if results: + _fold_carried_drops(results, carried_drops) + carried_drops = 0 + yield results + else: + # Whole page dropped: keep its count so it reaches the caller on + # the next yielded batch instead of vanishing with the page. + carried_drops += getattr(results, "dropped_count", 0) + # Advance unconditionally, so a page we cannot materialize can never + # wedge the loop. offset += page_size def listall(self) -> list[str]: @@ -3133,7 +3216,7 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato batch. Defaults to 30. Yields: - An async generator yielding batches of search results. + An async generator yielding non-empty ``SearchResults`` batches. Raises: TypeError: If the page_size argument is not of type int. @@ -3152,8 +3235,21 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato considerations and the expected volume of search results. Note: - For stable pagination, the query must have a `sort_by` clause. + For stable pagination, the query must have a `sort_by` clause on a + **unique** field. Redis documents that ``LIMIT`` without sorting is + non-deterministic, so pages may otherwise repeat or miss documents. + Very deep pagination is also bounded server-side by + ``search-max-search-results`` (1,000,000 by default, but 10,000 on + some managed tiers), past which the search errors rather than ending. + Note: + A yielded batch may contain fewer than ``page_size`` documents, and an + empty batch is never yielded. On Redis 8+ a matched document whose + data expires while the search is running is skipped; each batch + reports how many were skipped via ``dropped_count`` (and + ``complete``), including skips inherited from a page that was dropped + in its entirety. Pagination still runs to the end of the result set in + that case. """ if not isinstance(page_size, int): raise TypeError("page_size must be of type int") @@ -3161,13 +3257,29 @@ async def paginate(self, query: BaseQuery, page_size: int = 30) -> AsyncGenerato if page_size <= 0: raise ValueError("page_size must be greater than 0") + if isinstance(query, CountQuery): + raise TypeError( + "CountQuery cannot be paginated: it returns a match count rather " + "than documents. Use index.query(query) instead." + ) + first = 0 + carried_drops = 0 while True: query.paging(first, page_size) results = await self._query(query) - if not results: + if not _page_had_matches(results): break - yield results + if results: + _fold_carried_drops(results, carried_drops) + carried_drops = 0 + yield results + else: + # Whole page dropped: keep its count so it reaches the caller on + # the next yielded batch instead of vanishing with the page. + carried_drops += getattr(results, "dropped_count", 0) + # Advance unconditionally, so a page we cannot materialize can never + # wedge the loop. first += page_size async def listall(self) -> list[str]: diff --git a/tests/integration/test_query.py b/tests/integration/test_query.py index 335a37bc0..776ce25f5 100644 --- a/tests/integration/test_query.py +++ b/tests/integration/test_query.py @@ -605,6 +605,57 @@ def test_paginate_range_query(index, range_query): assert all(float(item["vector_distance"]) <= 0.2 for item in all_results) +def test_no_content_query_returns_ids(index, sample_data): + """NOCONTENT results must survive the missing-field-payload filter. + + Unlike the expiry race this needs no timing: under ``NOCONTENT`` the server + returns ids and no field data for *every* healthy match, so the drop + heuristic in ``process_results`` sees the same shape it uses to detect a + race victim. Before the ``_no_content`` short-circuit this returned an empty + list for every NOCONTENT query -- including through ``paginate``. + """ + total = len(sample_data) + high_credit = sum(1 for d in sample_data if d["credit_score"] == "high") + + # Plain filter, vector, and normalize-distance shapes. The last one used to + # raise KeyError on the absent vector_distance rather than return ids. + for query, expected in [ + (FilterQuery(filter_expression=Tag("credit_score") == "high"), high_credit), + ( + VectorQuery( + vector=[0.1, 0.1, 0.5], + vector_field_name="user_embedding", + num_results=total, + ), + total, + ), + ( + VectorQuery( + vector=[0.1, 0.1, 0.5], + vector_field_name="user_embedding", + num_results=total, + normalize_vector_distance=True, + ), + total, + ), + ]: + results = index.query(query.no_content()) + assert [doc["id"] for doc in results] and len(results) == expected + assert results.dropped_count == 0 + assert results.complete + + # paginate must reach the end of a NOCONTENT result set, not stop on page 1. + paged = [ + doc + for batch in index.paginate( + FilterQuery(filter_expression=Tag("credit_score") == "high").no_content(), + page_size=2, + ) + for doc in batch + ] + assert len(paged) == high_credit + + def test_sort_filter_query(index, sorted_filter_query): t = Text("job") % "" search(sorted_filter_query, index, t, 7, sort=True) diff --git a/tests/unit/test_paginate_dropped_page.py b/tests/unit/test_paginate_dropped_page.py new file mode 100644 index 000000000..700f9f719 --- /dev/null +++ b/tests/unit/test_paginate_dropped_page.py @@ -0,0 +1,191 @@ +"""Regression tests: ``paginate`` must not stop on a page whose matches were dropped. + +``process_results`` deliberately drops matched documents whose field payload came +back missing (a key that expires or is updated mid-query is returned as a matched +id with a ``nil`` field array, and is still counted in the server's total), so a +page can be empty while the server still has matches to report. ``paginate`` +previously terminated on ``if not results: break``, conflating "no more matches" +with "matches we could not materialize" and silently discarding every remaining +page. + +The live race is not reproducible on demand, so these tests serve canned +``FT.SEARCH`` replies from a fake ``index.search``, keyed by the paging offset the +query carried. That seam runs the real ``_query`` -> ``process_results`` -> +``SearchResults`` chain, so the ``dropped_count`` the fix depends on is produced by +production code rather than hand-constructed. Every fake bounds its request count, +so a termination regression fails loudly instead of hanging the suite. +""" + +import pytest +from redis.commands.search.result import Result + +from redisvl.index import AsyncSearchIndex, SearchIndex, SearchResults +from redisvl.query import CountQuery, VectorQuery +from redisvl.schema import IndexSchema + +sample_vector = [0.1, 0.1, 0.5, 0.15] + + +def _schema(): + return IndexSchema.from_dict( + { + "index": { + "name": "paginate_test", + "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 _query(): + return VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ) + + +def _healthy(doc_id): + """A raw FT.SEARCH reply fragment for a match that carries its fields.""" + return [doc_id, ["vector_distance", "0.1", "brand", "Nike"]] + + +def _race_victim(doc_id): + """A matched id whose field array came back nil -- the race victim. + + redis-py collapses the nil to an empty field set, leaving a ``Document`` whose + ``__dict__`` is only ``{"id": ..., "payload": None}``. For a vector query + (which always projects ``vector_distance``) ``process_results`` detects the + missing payload and drops the doc, incrementing ``dropped_count``. + """ + return [doc_id, None] + + +def _page(*fragments): + flat = [item for fragment in fragments for item in fragment] + return Result([len(fragments), *flat], True) + + +EMPTY_PAGE = Result([0], True) + +# One pass over every drop shape that matters: +# +# offset 0 both matches dropped -> a LEADING dropped page must not terminate +# offset 2 both healthy -> carries offset 0's 2 drops (dropped_count 2) +# offset 4 both matches dropped -> a second carry episode, forcing the reset +# offset 6 1 healthy + 1 dropped -> own 1 drop plus carried 2 (dropped_count 3) +# offset 8 empty -> the genuine end of the result set +# +# The dropped_count sequence [2, 3] is what pins the accounting: losing the fold +# gives [0, 1], losing the reset gives [2, 5], and assigning instead of adding +# gives [2, 2]. +DROP_SHAPES = { + 0: _page(_race_victim("doc:1"), _race_victim("doc:2")), + 2: _page(_healthy("doc:3"), _healthy("doc:4")), + 4: _page(_race_victim("doc:5"), _race_victim("doc:6")), + 6: _page(_healthy("doc:7"), _race_victim("doc:8")), + 8: EMPTY_PAGE, +} + + +def _fake_search(pages, seen): + """Serve canned FT.SEARCH replies keyed by the query's paging offset. + + Keying by offset (rather than popping a queue) means a failure to advance + shows up as repeated documents or a wedge, not as a silently different page. + ``seen`` records the request order and bounds runaway loops. It reads + redis-py's private ``Query._offset``, which ``paging()`` sets. + """ + + def _search(query, query_params=None): + seen.append(query._offset) + assert len(seen) <= 20, f"paginate made {len(seen)} requests; likely wedged" + return pages.get(query._offset, EMPTY_PAGE) + + return _search + + +def test_paginate_continues_past_dropped_pages_and_accounts_for_their_drops(): + index = SearchIndex(_schema()) + seen: list[int] = [] + index.search = _fake_search(DROP_SHAPES, seen) # type: ignore[method-assign] + + batches = list(index.paginate(_query(), page_size=2)) + + # Iteration reached the end rather than stopping at offset 0's dropped page. + assert [[doc["id"] for doc in batch] for batch in batches] == [ + ["doc:3", "doc:4"], + ["doc:7"], + ] + # Never an empty batch, and the completeness metadata survives the generator. + assert all(batch for batch in batches) + assert all(isinstance(batch, SearchResults) for batch in batches) + # Drops from skipped pages ride along on the next batch the caller sees. + assert [batch.dropped_count for batch in batches] == [2, 3] + assert [batch.complete for batch in batches] == [False, False] + assert seen == [0, 2, 4, 6, 8] + + +@pytest.mark.asyncio +async def test_async_paginate_continues_past_dropped_pages_and_accounts_for_their_drops(): + """Async mirror: ``AsyncSearchIndex.paginate`` is a hand-duplicated copy.""" + index = AsyncSearchIndex(_schema()) + seen: list[int] = [] + sync_search = _fake_search(DROP_SHAPES, seen) + + async def _search(query, query_params=None): + return sync_search(query, query_params) + + index.search = _search # type: ignore[method-assign] + + batches = [batch async for batch in index.paginate(_query(), page_size=2)] + + assert [[doc["id"] for doc in batch] for batch in batches] == [ + ["doc:3", "doc:4"], + ["doc:7"], + ] + assert all(batch for batch in batches) + assert all(isinstance(batch, SearchResults) for batch in batches) + assert [batch.dropped_count for batch in batches] == [2, 3] + assert [batch.complete for batch in batches] == [False, False] + assert seen == [0, 2, 4, 6, 8] + + +def test_paginate_rejects_count_query(): + """CountQuery returns a match count, not documents, so it cannot paginate. + + ``process_results`` returns a bare ``int`` for it, which used to make + ``paginate`` yield that integer forever. + """ + index = SearchIndex(_schema()) + with pytest.raises(TypeError, match="CountQuery cannot be paginated"): + list(index.paginate(CountQuery("*"), page_size=2)) + + +@pytest.mark.asyncio +async def test_async_paginate_rejects_count_query(): + index = AsyncSearchIndex(_schema()) + with pytest.raises(TypeError, match="CountQuery cannot be paginated"): + [batch async for batch in index.paginate(CountQuery("*"), page_size=2)] + + +def test_paginate_validates_page_size(): + index = SearchIndex(_schema()) + with pytest.raises(TypeError, match="page_size must be an integer"): + list(index.paginate(_query(), page_size="2")) # type: ignore[arg-type] + with pytest.raises(ValueError, match="page_size must be greater than 0"): + list(index.paginate(_query(), page_size=0)) diff --git a/tests/unit/test_query_types.py b/tests/unit/test_query_types.py index ff969e11f..9575c1318 100644 --- a/tests/unit/test_query_types.py +++ b/tests/unit/test_query_types.py @@ -1291,6 +1291,44 @@ def test_filter_query_hash_id_only_not_skipped(): assert results == [{"id": "doc:1"}] +def test_no_content_query_not_skipped(): + """Over-skip guard: a NOCONTENT query legitimately returns ids only. + + The server "returns the document ids and not the content" for every healthy + match, so a missing field payload carries no information and must not be read + as the expiry race. redis-py's ``no_content()`` leaves ``_return_fields`` + untouched, so both detection branches would otherwise fire on every document. + """ + nocontent = Result([2, "doc:1", "doc:2"], False) + + vector_query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + return_fields=["brand"], + ).no_content() + results = process_results(nocontent, vector_query, _hash_schema()) + assert [doc["id"] for doc in results] == ["doc:1", "doc:2"] + assert results.dropped_count == 0 + + # JSON full-object unpack: no "json" key is expected under NOCONTENT either. + json_query = FilterQuery(Tag("brand") == "Nike").no_content() + results = process_results(nocontent, json_query, _json_schema()) + assert [doc["id"] for doc in results] == ["doc:1", "doc:2"] + assert results.dropped_count == 0 + + +def test_no_content_query_with_normalize_does_not_raise(): + """NOCONTENT leaves no distance to normalize; the branch must skip it.""" + nocontent = Result([1, "doc:1"], False) + query = VectorQuery( + vector=sample_vector, + vector_field_name="user_embedding", + normalize_vector_distance=True, + ).no_content() + # Previously raised KeyError on doc_dict[query.DISTANCE_ID]. + assert process_results(nocontent, query, _hash_schema()) == [{"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(