diff --git a/semantica/vector_store/qdrant_store.py b/semantica/vector_store/qdrant_store.py index 5b057ca5..ac55f553 100644 --- a/semantica/vector_store/qdrant_store.py +++ b/semantica/vector_store/qdrant_store.py @@ -590,20 +590,19 @@ class QdrantStore: with_payload=True, with_vectors=True, ) - results = [] - for rec in records: - results.append( - { - "id": str(rec.id), - "metadata": rec.payload or {}, - "vector": np.array(rec.vector) if rec.vector is not None else None, - } - ) - return results + return [self._record_to_result(rec) for rec in records] except Exception as e: self.logger.warning(f"Failed to scroll Qdrant points by metadata filter: {e}") return [] + @staticmethod + def _record_to_result(rec: Any) -> Dict[str, Any]: + return { + "id": str(rec.id), + "metadata": rec.payload or {}, + "vector": np.array(rec.vector) if rec.vector is not None else None, + } + def iter_all(self, batch_size: int = 500): """ Iterate over every stored point using Qdrant's scroll cursor. @@ -624,7 +623,7 @@ class QdrantStore: Raises: ProcessingError: If the collection or client is not initialized, or - if the scan cannot advance past a full page. + if the cursor stops advancing before the scan completes. """ if self.collection is None or self.client is None or not QDRANT_AVAILABLE: raise ProcessingError( @@ -632,6 +631,7 @@ class QdrantStore: ) next_offset = None + last_offset = object() while True: records, next_offset = self.client.scroll( collection_name=self.collection.collection_name, @@ -642,11 +642,7 @@ class QdrantStore: ) for rec in records: - yield { - "id": str(rec.id), - "metadata": rec.payload or {}, - "vector": np.array(rec.vector) if rec.vector is not None else None, - } + yield self._record_to_result(rec) # A final page can carry records alongside a null cursor, so they # are yielded above before stopping. Passing offset=None back to @@ -654,14 +650,16 @@ class QdrantStore: if next_offset is None: return - # A cursor without records means the scan cannot advance, which is - # truncation rather than completion. - if not records: + # An empty page with a live cursor isn't necessarily truncation — + # a batch window that lands entirely on deleted points comes back + # this way too, and there's more to scan past it. Only treat it as + # stuck if the cursor itself stops moving. + if not records and next_offset == last_offset: raise ProcessingError( - "Qdrant returned an empty page alongside a continuation " - "cursor, so the scan cannot advance. Refusing to return a " - "truncated scan." + "Qdrant scroll cursor stopped advancing without reaching " + "the end of the collection, so the scan cannot complete." ) + last_offset = next_offset def delete_vectors( self, point_ids: List[Union[str, int]], **options diff --git a/tests/vector_store/test_qdrant_store.py b/tests/vector_store/test_qdrant_store.py index 8b4ef049..8958d8c1 100644 --- a/tests/vector_store/test_qdrant_store.py +++ b/tests/vector_store/test_qdrant_store.py @@ -94,12 +94,26 @@ def test_iter_all_empty_collection_yields_nothing(): @patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True) -def test_iter_all_raises_on_empty_page_with_a_cursor(): - """An empty page with a cursor left set cannot advance, so returning here - would hand back a partial scan that reads as a complete one.""" - store = _store_with_scroll(([], "cursor-that-never-clears")) +def test_iter_all_continues_past_empty_page_with_advancing_cursor(): + store = _store_with_scroll( + ([], "cursor-1"), + ([_record(1)], None), + ) - with pytest.raises(ProcessingError, match="cannot advance"): + result = list(store.iter_all()) + + assert [item["id"] for item in result] == ["1"] + assert store.client.scroll.call_count == 2 + + +@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True) +def test_iter_all_raises_when_cursor_stops_advancing(): + store = _store_with_scroll( + ([], "stuck-cursor"), + ([], "stuck-cursor"), + ) + + with pytest.raises(ProcessingError, match="stopped advancing"): list(store.iter_all())