mirror of
https://github.com/semantica-agi/semantica.git
synced 2026-09-01 04:00:28 +00:00
Compare commits
4
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
bd1ba24b24 | ||
|
|
e8ff36f088 | ||
|
|
ec9e63e16f | ||
|
|
274d5d1195 |
@@ -604,6 +604,62 @@ class QdrantStore:
|
||||
self.logger.warning(f"Failed to scroll Qdrant points by metadata filter: {e}")
|
||||
return []
|
||||
|
||||
def iter_all(self, batch_size: int = 500):
|
||||
"""
|
||||
Iterate over every stored point using Qdrant's native scroll cursor.
|
||||
|
||||
Qdrant paginates by point-ID cursor, not by row offset, so this is
|
||||
exposed instead of scan_vectors(offset, limit). An integer passed to
|
||||
scroll()'s offset is a point ID rather than a rank, so there is no way
|
||||
to seek to "the Nth record" without walking from the start.
|
||||
VectorStore.iter_vectors() prefers this method when it is present.
|
||||
|
||||
Assumes a single unnamed vector per point, matching how insert_vectors()
|
||||
writes them and how get_vector() reads them back. Collections configured
|
||||
with named or multi-vectors are not handled here.
|
||||
|
||||
Args:
|
||||
batch_size: Points to request per scroll call
|
||||
|
||||
Yields:
|
||||
Result dicts with 'id', 'metadata', and 'vector', in scroll order
|
||||
|
||||
Raises:
|
||||
ProcessingError: If the collection or client is not initialized.
|
||||
Errors are raised rather than swallowed because a scan that
|
||||
silently yields nothing is indistinguishable from an empty
|
||||
source, which would let a caller such as `store migrate`
|
||||
report success having copied nothing (issue #1083).
|
||||
"""
|
||||
if self.collection is None or self.client is None or not QDRANT_AVAILABLE:
|
||||
raise ProcessingError(
|
||||
"Collection not initialized. Call create_collection() or get_collection() first."
|
||||
)
|
||||
|
||||
next_offset = None
|
||||
while True:
|
||||
records, next_offset = self.client.scroll(
|
||||
collection_name=self.collection.collection_name,
|
||||
limit=batch_size,
|
||||
offset=next_offset,
|
||||
with_payload=True,
|
||||
with_vectors=True,
|
||||
)
|
||||
|
||||
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,
|
||||
}
|
||||
|
||||
# The final page can carry records while already reporting no next
|
||||
# cursor, so those records are yielded above before stopping here.
|
||||
# Calling scroll() again with offset=None would restart from the
|
||||
# beginning rather than continue past the end.
|
||||
if next_offset is None or not records:
|
||||
return
|
||||
|
||||
def delete_vectors(
|
||||
self, point_ids: List[Union[str, int]], **options
|
||||
) -> Dict[str, Any]:
|
||||
|
||||
@@ -867,12 +867,25 @@ class VectorStore:
|
||||
"""
|
||||
Iterate over every stored vector, one page at a time.
|
||||
|
||||
Backends whose native pagination is cursor based (Qdrant, Pinecone,
|
||||
Milvus, Weaviate) cannot honestly implement the positional
|
||||
scan_vectors(offset, limit) contract, so they expose iter_all()
|
||||
instead and it is preferred here when present. Backends with real
|
||||
positional access (inmemory, FAISS, SQLite-vec, PgVector) fall
|
||||
through to the offset loop below.
|
||||
|
||||
Args:
|
||||
batch_size: Number of vectors to fetch per underlying scan_vectors() call
|
||||
batch_size: Number of vectors to fetch per underlying call
|
||||
|
||||
Yields:
|
||||
Result dicts with 'id', 'metadata', and 'vector', in scan order
|
||||
"""
|
||||
if self.backend != "inmemory" and self._backend_store is not None:
|
||||
iter_all = getattr(self._backend_store, "iter_all", None)
|
||||
if callable(iter_all):
|
||||
yield from iter_all(batch_size=batch_size)
|
||||
return
|
||||
|
||||
offset = 0
|
||||
while True:
|
||||
page = self.scan_vectors(offset=offset, limit=batch_size)
|
||||
|
||||
@@ -611,6 +611,115 @@ class WeaviateStore:
|
||||
self.logger.warning(f"Failed to fetch Weaviate objects by metadata filter: {e}")
|
||||
return results if results else []
|
||||
|
||||
def iter_all(self, batch_size: int = 500):
|
||||
"""
|
||||
Iterate over every stored object using Weaviate's UUID cursor.
|
||||
|
||||
Paginates by the last object's UUID rather than a row offset, which is
|
||||
why this exists instead of scan_vectors(offset, limit).
|
||||
|
||||
Assumes a single unnamed vector per object, as get_vector() and
|
||||
filter_by_metadata() already do. Named-vector collections return a
|
||||
mapping and are not handled.
|
||||
|
||||
Args:
|
||||
batch_size: Objects to request per fetch_objects() call
|
||||
|
||||
Yields:
|
||||
Result dicts with 'id', 'metadata', and 'vector', in cursor order
|
||||
|
||||
Raises:
|
||||
ProcessingError: If the collection is not initialized, or if the
|
||||
scan cannot advance past a full page.
|
||||
"""
|
||||
if self.collection is None or not WEAVIATE_AVAILABLE:
|
||||
raise ProcessingError(
|
||||
"Collection not initialized. Call get_collection() first."
|
||||
)
|
||||
|
||||
after_cursor = None
|
||||
scanned_count = 0
|
||||
# Degrades cursor -> offset -> single_page as the client rejects each
|
||||
# form. Tracked across iterations, not just inside the except branch,
|
||||
# or later pages go out with no pagination argument at all.
|
||||
mode = "cursor"
|
||||
|
||||
while True:
|
||||
kwargs = {"limit": batch_size, "include_vector": True}
|
||||
if mode == "cursor" and after_cursor is not None:
|
||||
kwargs["after"] = after_cursor
|
||||
elif mode == "offset":
|
||||
kwargs["offset"] = scanned_count
|
||||
|
||||
try:
|
||||
objs = self.collection.query.fetch_objects(**kwargs)
|
||||
except TypeError:
|
||||
if mode == "cursor" and "after" in kwargs:
|
||||
mode = "offset"
|
||||
kwargs.pop("after", None)
|
||||
kwargs["offset"] = scanned_count
|
||||
try:
|
||||
objs = self.collection.query.fetch_objects(**kwargs)
|
||||
except TypeError:
|
||||
mode = "single_page"
|
||||
kwargs.pop("offset", None)
|
||||
objs = self.collection.query.fetch_objects(**kwargs)
|
||||
elif mode == "offset":
|
||||
mode = "single_page"
|
||||
kwargs.pop("offset", None)
|
||||
objs = self.collection.query.fetch_objects(**kwargs)
|
||||
else:
|
||||
raise
|
||||
|
||||
batch_objects = getattr(objs, "objects", None) if objs else None
|
||||
if not batch_objects:
|
||||
return
|
||||
|
||||
for obj in batch_objects:
|
||||
obj_uuid = getattr(obj, "uuid", None)
|
||||
raw_vector = getattr(obj, "vector", None)
|
||||
yield {
|
||||
"id": str(obj_uuid) if obj_uuid is not None else None,
|
||||
"metadata": getattr(obj, "properties", None) or {},
|
||||
"vector": (
|
||||
np.array(raw_vector)
|
||||
if raw_vector is not None and len(raw_vector) > 0
|
||||
else None
|
||||
),
|
||||
}
|
||||
|
||||
scanned_count += len(batch_objects)
|
||||
|
||||
# Past this point the page was full, so failing to advance is
|
||||
# truncation rather than completion.
|
||||
if len(batch_objects) < batch_size:
|
||||
return
|
||||
|
||||
if mode == "single_page":
|
||||
raise ProcessingError(
|
||||
"This Weaviate client accepts neither an `after` cursor nor a "
|
||||
"numeric offset, so the scan cannot advance past the first "
|
||||
"page. Refusing to return a truncated scan."
|
||||
)
|
||||
|
||||
if mode == "offset":
|
||||
continue
|
||||
|
||||
last_uuid = getattr(batch_objects[-1], "uuid", None)
|
||||
if last_uuid is None:
|
||||
raise ProcessingError(
|
||||
"The last object of a full Weaviate page has no uuid, so the "
|
||||
"cursor cannot advance. Refusing to return a truncated scan."
|
||||
)
|
||||
|
||||
next_cursor = str(last_uuid)
|
||||
if next_cursor == after_cursor:
|
||||
raise ProcessingError(
|
||||
"The Weaviate cursor stopped advancing, so the listing is "
|
||||
"repeating a page. Refusing to return a truncated scan."
|
||||
)
|
||||
after_cursor = next_cursor
|
||||
|
||||
|
||||
def query_vectors(
|
||||
self,
|
||||
|
||||
@@ -0,0 +1,152 @@
|
||||
"""Tests for QdrantStore.iter_all() cursor enumeration.
|
||||
|
||||
Qdrant is not installed in this environment, so these drive the real
|
||||
QdrantStore against a MagicMock standing in for the qdrant_client, following
|
||||
the pattern already used for qdrant in test_backend_metadata_filtering.py.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from semantica.utils.exceptions import ProcessingError
|
||||
from semantica.vector_store.qdrant_store import QdrantStore
|
||||
|
||||
|
||||
def _record(point_id, payload=None, vector=None):
|
||||
"""Build a stand-in for a qdrant_client Record."""
|
||||
rec = MagicMock()
|
||||
rec.id = point_id
|
||||
rec.payload = payload
|
||||
rec.vector = vector
|
||||
return rec
|
||||
|
||||
|
||||
def _store_with_scroll(*pages):
|
||||
"""QdrantStore whose client.scroll() returns the given (records, cursor) pages."""
|
||||
store = QdrantStore()
|
||||
store.client = MagicMock()
|
||||
store.client.scroll.side_effect = list(pages)
|
||||
store.collection = MagicMock()
|
||||
store.collection.collection_name = "test_collection"
|
||||
return store
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
|
||||
def test_iter_all_threads_cursor_across_pages():
|
||||
"""The next call must continue from the previous page's next_page_offset."""
|
||||
store = _store_with_scroll(
|
||||
([_record(1), _record(2)], "cursor-1"),
|
||||
([_record(3)], None),
|
||||
)
|
||||
|
||||
result = list(store.iter_all(batch_size=2))
|
||||
|
||||
assert [item["id"] for item in result] == ["1", "2", "3"]
|
||||
calls = store.client.scroll.call_args_list
|
||||
assert len(calls) == 2
|
||||
assert calls[0][1]["offset"] is None
|
||||
assert calls[0][1]["limit"] == 2
|
||||
assert calls[1][1]["offset"] == "cursor-1"
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
|
||||
def test_iter_all_yields_final_page_that_reports_no_next_cursor():
|
||||
"""Qdrant can return records and a null cursor on the same page.
|
||||
|
||||
Those records must still be yielded. Treating a null cursor as "stop
|
||||
before this page" would silently drop the tail of every scan.
|
||||
"""
|
||||
store = _store_with_scroll(([_record(1), _record(2)], None))
|
||||
|
||||
result = list(store.iter_all(batch_size=10))
|
||||
|
||||
assert [item["id"] for item in result] == ["1", "2"]
|
||||
assert store.client.scroll.call_count == 1
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
|
||||
def test_iter_all_converts_records_to_the_shared_result_shape():
|
||||
store = _store_with_scroll(
|
||||
([_record(7, payload={"tag": "x"}, vector=[0.1, 0.2, 0.3])], None),
|
||||
)
|
||||
|
||||
item = list(store.iter_all())[0]
|
||||
|
||||
assert item["id"] == "7"
|
||||
assert item["metadata"] == {"tag": "x"}
|
||||
np.testing.assert_allclose(item["vector"], np.array([0.1, 0.2, 0.3]))
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
|
||||
def test_iter_all_handles_missing_payload_and_vector():
|
||||
store = _store_with_scroll(([_record(1, payload=None, vector=None)], None))
|
||||
|
||||
item = list(store.iter_all())[0]
|
||||
|
||||
assert item["metadata"] == {}
|
||||
assert item["vector"] is None
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
|
||||
def test_iter_all_empty_collection_yields_nothing():
|
||||
store = _store_with_scroll(([], None))
|
||||
|
||||
assert list(store.iter_all()) == []
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
|
||||
def test_iter_all_stops_on_empty_page_even_with_a_cursor():
|
||||
"""Defensive: an empty page ends the scan rather than looping forever."""
|
||||
store = _store_with_scroll(([], "cursor-that-never-clears"))
|
||||
|
||||
assert list(store.iter_all()) == []
|
||||
assert store.client.scroll.call_count == 1
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
|
||||
def test_iter_all_raises_when_collection_not_initialized():
|
||||
"""Must fail loudly, not yield nothing.
|
||||
|
||||
An empty scan is indistinguishable from an empty source, which would let
|
||||
`store migrate` report success having copied nothing (issue #1083).
|
||||
"""
|
||||
store = QdrantStore()
|
||||
|
||||
with pytest.raises(ProcessingError, match="Collection not initialized"):
|
||||
list(store.iter_all())
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", False)
|
||||
def test_iter_all_raises_when_qdrant_unavailable():
|
||||
store = QdrantStore()
|
||||
store.client = MagicMock()
|
||||
store.collection = MagicMock()
|
||||
|
||||
with pytest.raises(ProcessingError):
|
||||
list(store.iter_all())
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
|
||||
def test_iter_all_propagates_scroll_errors():
|
||||
store = QdrantStore()
|
||||
store.client = MagicMock()
|
||||
store.client.scroll.side_effect = RuntimeError("connection reset")
|
||||
store.collection = MagicMock()
|
||||
store.collection.collection_name = "test_collection"
|
||||
|
||||
with pytest.raises(RuntimeError, match="connection reset"):
|
||||
list(store.iter_all())
|
||||
|
||||
|
||||
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
|
||||
def test_iter_all_requests_payload_and_vectors():
|
||||
store = _store_with_scroll(([], None))
|
||||
|
||||
list(store.iter_all())
|
||||
|
||||
kwargs = store.client.scroll.call_args[1]
|
||||
assert kwargs["with_payload"] is True
|
||||
assert kwargs["with_vectors"] is True
|
||||
assert kwargs["collection_name"] == "test_collection"
|
||||
@@ -26,6 +26,7 @@ from unittest.mock import MagicMock, patch
|
||||
|
||||
import numpy as np
|
||||
|
||||
from semantica.utils.exceptions import ProcessingError
|
||||
from semantica.vector_store.vector_store import VectorStore, VectorManager
|
||||
|
||||
|
||||
@@ -138,6 +139,38 @@ class _NonScanningBackendStore:
|
||||
"""Fake persistent backend store without any scan capability."""
|
||||
|
||||
|
||||
class _IterAllBackendStore:
|
||||
"""Fake cursor-based backend store exposing iter_all() but not scan_vectors().
|
||||
|
||||
Mirrors qdrant/pinecone/milvus/weaviate, which cannot honour a positional
|
||||
offset and therefore expose native iteration instead.
|
||||
"""
|
||||
|
||||
def __init__(self, items):
|
||||
self._items = items
|
||||
self.batch_sizes = []
|
||||
|
||||
def iter_all(self, batch_size=500):
|
||||
self.batch_sizes.append(batch_size)
|
||||
for item in self._items:
|
||||
yield item
|
||||
|
||||
def scan_vectors(self, offset=0, limit=100):
|
||||
raise AssertionError("scan_vectors() must not be called when iter_all() exists")
|
||||
|
||||
|
||||
class _MisShapedIterAllBackendStore:
|
||||
"""Backend store whose ``iter_all`` attribute is not callable."""
|
||||
|
||||
iter_all = 42 # plain attribute, not a method
|
||||
|
||||
def __init__(self, items):
|
||||
self._items = items
|
||||
|
||||
def scan_vectors(self, offset=0, limit=100):
|
||||
return self._items[offset:offset + limit]
|
||||
|
||||
|
||||
class VectorStoreScanVectorsTests(unittest.TestCase):
|
||||
"""VectorStore.scan_vectors() / iter_vectors() backend-agnostic accessors."""
|
||||
|
||||
@@ -192,6 +225,81 @@ class VectorStoreScanVectorsTests(unittest.TestCase):
|
||||
self.assertEqual(list(store.iter_vectors(batch_size=2)), [])
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# VectorStore.iter_vectors() preference for a native iter_all()
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
class VectorStoreIterAllDispatchTests(unittest.TestCase):
|
||||
"""iter_vectors() prefers a backend's native iter_all() when present.
|
||||
|
||||
Cursor-based backends cannot implement scan_vectors(offset, limit)
|
||||
honestly, so they expose iter_all() instead and iter_vectors() routes to
|
||||
it rather than walking offsets.
|
||||
"""
|
||||
|
||||
def _persistent_store(self, backend_store, backend_name="qdrant"):
|
||||
store = VectorStore(backend="inmemory", dimension=2)
|
||||
store.backend = backend_name
|
||||
store._backend_store = backend_store
|
||||
return store
|
||||
|
||||
def test_iter_vectors_uses_iter_all_when_available(self):
|
||||
items = [
|
||||
{"id": "a", "vector": None, "metadata": {"n": 1}},
|
||||
{"id": "b", "vector": None, "metadata": {"n": 2}},
|
||||
]
|
||||
backend = _IterAllBackendStore(items)
|
||||
store = self._persistent_store(backend)
|
||||
|
||||
self.assertEqual(list(store.iter_vectors(batch_size=7)), items)
|
||||
|
||||
def test_iter_vectors_forwards_batch_size_to_iter_all(self):
|
||||
backend = _IterAllBackendStore([])
|
||||
store = self._persistent_store(backend)
|
||||
|
||||
list(store.iter_vectors(batch_size=32))
|
||||
|
||||
self.assertEqual(backend.batch_sizes, [32])
|
||||
|
||||
def test_iter_vectors_falls_back_to_scan_vectors_without_iter_all(self):
|
||||
items = [{"id": "a", "vector": None, "metadata": {}}]
|
||||
store = self._persistent_store(_ScanningBackendStore(items))
|
||||
|
||||
self.assertEqual(list(store.iter_vectors(batch_size=2)), items)
|
||||
|
||||
def test_iter_vectors_falls_back_when_iter_all_not_callable(self):
|
||||
# A mis-shaped adapter exposing a non-callable ``iter_all`` must not be
|
||||
# invoked; the offset path still has to work. Mirrors the count()
|
||||
# precedent in _MisShapedBackendStore.
|
||||
items = [{"id": "a", "vector": None, "metadata": {}}]
|
||||
store = self._persistent_store(_MisShapedIterAllBackendStore(items))
|
||||
|
||||
self.assertEqual(list(store.iter_vectors(batch_size=2)), items)
|
||||
|
||||
def test_iter_vectors_inmemory_ignores_iter_all(self):
|
||||
store = VectorStore(backend="inmemory", dimension=2)
|
||||
store.store_vectors([np.array([1.0, 0.0])], [{"type": "a"}])
|
||||
store._backend_store = _IterAllBackendStore([{"id": "wrong"}])
|
||||
|
||||
collected = list(store.iter_vectors(batch_size=2))
|
||||
|
||||
self.assertEqual([item["metadata"] for item in collected], [{"type": "a"}])
|
||||
|
||||
def test_iter_vectors_propagates_iter_all_errors(self):
|
||||
# A scan that silently yields nothing is indistinguishable from an
|
||||
# empty source, which would let `store migrate` report success having
|
||||
# copied nothing (issue #1083).
|
||||
class _FailingIterAll:
|
||||
def iter_all(self, batch_size=500):
|
||||
raise ProcessingError("backend unreachable")
|
||||
yield # pragma: no cover - makes this a generator
|
||||
|
||||
store = self._persistent_store(_FailingIterAll())
|
||||
|
||||
with self.assertRaises(ProcessingError):
|
||||
list(store.iter_vectors(batch_size=2))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# VectorManager tests — inmemory backend
|
||||
# ---------------------------------------------------------------------------
|
||||
|
||||
@@ -0,0 +1,231 @@
|
||||
"""Tests for WeaviateStore.iter_all() cursor enumeration.
|
||||
|
||||
weaviate-client is not installed in this environment, so these drive the real
|
||||
WeaviateStore against MagicMocks, following the pattern already used for
|
||||
weaviate in test_backend_metadata_filtering.py.
|
||||
"""
|
||||
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import numpy as np
|
||||
import pytest
|
||||
|
||||
from semantica.utils.exceptions import ProcessingError
|
||||
from semantica.vector_store.weaviate_store import WeaviateStore
|
||||
|
||||
|
||||
def _obj(uuid, properties=None, vector=None):
|
||||
"""Stand-in for a weaviate v4 returned object."""
|
||||
obj = MagicMock()
|
||||
obj.uuid = uuid
|
||||
obj.properties = properties
|
||||
obj.vector = vector
|
||||
return obj
|
||||
|
||||
|
||||
def _page(objects):
|
||||
"""Stand-in for a fetch_objects() response."""
|
||||
response = MagicMock()
|
||||
response.objects = objects
|
||||
return response
|
||||
|
||||
|
||||
def _store_with_pages(*pages):
|
||||
store = WeaviateStore()
|
||||
store.collection = MagicMock()
|
||||
store.collection.query.fetch_objects.side_effect = list(pages)
|
||||
return store
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_threads_uuid_cursor_across_pages():
|
||||
"""The next page must continue after the last object's UUID."""
|
||||
store = _store_with_pages(
|
||||
_page([_obj("uuid-1"), _obj("uuid-2")]),
|
||||
_page([_obj("uuid-3")]),
|
||||
)
|
||||
|
||||
result = list(store.iter_all(batch_size=2))
|
||||
|
||||
assert [item["id"] for item in result] == ["uuid-1", "uuid-2", "uuid-3"]
|
||||
calls = store.collection.query.fetch_objects.call_args_list
|
||||
assert "after" not in calls[0][1]
|
||||
assert calls[1][1]["after"] == "uuid-2"
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_stops_on_short_page():
|
||||
"""A page smaller than batch_size means the collection is exhausted."""
|
||||
store = _store_with_pages(_page([_obj("uuid-1")]))
|
||||
|
||||
result = list(store.iter_all(batch_size=5))
|
||||
|
||||
assert [item["id"] for item in result] == ["uuid-1"]
|
||||
assert store.collection.query.fetch_objects.call_count == 1
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_raises_when_cursor_stops_advancing():
|
||||
"""A stalled cursor must terminate, but not quietly: a partial scan reads
|
||||
as a complete one."""
|
||||
store = WeaviateStore()
|
||||
store.collection = MagicMock()
|
||||
store.collection.query.fetch_objects.return_value = _page(
|
||||
[_obj("same-uuid"), _obj("same-uuid")]
|
||||
)
|
||||
|
||||
with pytest.raises(ProcessingError, match="stopped advancing"):
|
||||
list(store.iter_all(batch_size=2))
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_offset_fallback_advances_across_pages():
|
||||
"""Regression: the offset was only set inside the except branch, so pages
|
||||
after the fallback went out with no pagination at all and the scan
|
||||
restarted from page one."""
|
||||
store = WeaviateStore()
|
||||
store.collection = MagicMock()
|
||||
calls = []
|
||||
|
||||
def _fetch(**kwargs):
|
||||
calls.append(dict(kwargs))
|
||||
if "after" in kwargs:
|
||||
raise TypeError("unexpected keyword argument 'after'")
|
||||
page_number = len(calls)
|
||||
if page_number < 4:
|
||||
return _page([_obj(f"u{page_number}a"), _obj(f"u{page_number}b")])
|
||||
return _page([_obj("last")])
|
||||
|
||||
store.collection.query.fetch_objects.side_effect = _fetch
|
||||
|
||||
ids = [item["id"] for item in store.iter_all(batch_size=2)]
|
||||
|
||||
assert len(set(ids)) == len(ids), f"duplicate ids means the scan restarted: {ids}"
|
||||
assert [c.get("offset") for c in calls] == [None, None, 2, 4]
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_raises_when_no_pagination_is_supported():
|
||||
"""A client rejecting both `after` and `offset` cannot page past the first
|
||||
result."""
|
||||
store = WeaviateStore()
|
||||
store.collection = MagicMock()
|
||||
|
||||
def _fetch(**kwargs):
|
||||
if "after" in kwargs or "offset" in kwargs:
|
||||
raise TypeError("unsupported")
|
||||
return _page([_obj("a"), _obj("b")])
|
||||
|
||||
store.collection.query.fetch_objects.side_effect = _fetch
|
||||
|
||||
with pytest.raises(ProcessingError, match="neither an .after. cursor nor a"):
|
||||
list(store.iter_all(batch_size=2))
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_empty_collection_yields_nothing():
|
||||
store = _store_with_pages(_page([]))
|
||||
|
||||
assert list(store.iter_all()) == []
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_converts_objects_to_the_shared_result_shape():
|
||||
store = _store_with_pages(
|
||||
_page([_obj("uuid-7", properties={"tag": "x"}, vector=[0.1, 0.2, 0.3])]),
|
||||
)
|
||||
|
||||
item = list(store.iter_all())[0]
|
||||
|
||||
assert item["id"] == "uuid-7"
|
||||
assert item["metadata"] == {"tag": "x"}
|
||||
np.testing.assert_allclose(item["vector"], np.array([0.1, 0.2, 0.3]))
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_handles_missing_properties_and_vector():
|
||||
store = _store_with_pages(_page([_obj("uuid-1", properties=None, vector=None)]))
|
||||
|
||||
item = list(store.iter_all())[0]
|
||||
|
||||
assert item["metadata"] == {}
|
||||
assert item["vector"] is None
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_treats_empty_vector_as_none():
|
||||
store = _store_with_pages(_page([_obj("uuid-1", vector=[])]))
|
||||
|
||||
assert list(store.iter_all())[0]["vector"] is None
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_requests_vectors():
|
||||
"""Weaviate omits vectors unless include_vector is set."""
|
||||
store = _store_with_pages(_page([]))
|
||||
|
||||
list(store.iter_all(batch_size=64))
|
||||
|
||||
kwargs = store.collection.query.fetch_objects.call_args[1]
|
||||
assert kwargs["include_vector"] is True
|
||||
assert kwargs["limit"] == 64
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_falls_back_to_offset_when_after_unsupported():
|
||||
"""Older clients reject `after`; the scan degrades to numeric offset."""
|
||||
store = WeaviateStore()
|
||||
store.collection = MagicMock()
|
||||
seen = {"calls": 0}
|
||||
|
||||
def _fetch(**kwargs):
|
||||
if "after" in kwargs:
|
||||
raise TypeError("unexpected keyword argument 'after'")
|
||||
seen["calls"] += 1
|
||||
if seen["calls"] == 1:
|
||||
return _page([_obj("uuid-1"), _obj("uuid-2")])
|
||||
return _page([_obj("uuid-3")])
|
||||
|
||||
store.collection.query.fetch_objects.side_effect = _fetch
|
||||
|
||||
result = list(store.iter_all(batch_size=2))
|
||||
|
||||
assert [item["id"] for item in result] == ["uuid-1", "uuid-2", "uuid-3"]
|
||||
offsets = [
|
||||
c[1]["offset"]
|
||||
for c in store.collection.query.fetch_objects.call_args_list
|
||||
if "offset" in c[1]
|
||||
]
|
||||
assert offsets == [2]
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_raises_when_collection_not_initialized():
|
||||
"""Must fail loudly, not yield nothing.
|
||||
|
||||
An empty scan is indistinguishable from an empty source, which would let
|
||||
`store migrate` report success having copied nothing (issue #1083).
|
||||
"""
|
||||
store = WeaviateStore()
|
||||
|
||||
with pytest.raises(ProcessingError, match="Collection not initialized"):
|
||||
list(store.iter_all())
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", False)
|
||||
def test_iter_all_raises_when_weaviate_unavailable():
|
||||
store = WeaviateStore()
|
||||
store.collection = MagicMock()
|
||||
|
||||
with pytest.raises(ProcessingError):
|
||||
list(store.iter_all())
|
||||
|
||||
|
||||
@patch("semantica.vector_store.weaviate_store.WEAVIATE_AVAILABLE", True)
|
||||
def test_iter_all_propagates_fetch_errors():
|
||||
store = WeaviateStore()
|
||||
store.collection = MagicMock()
|
||||
store.collection.query.fetch_objects.side_effect = RuntimeError("connection reset")
|
||||
|
||||
with pytest.raises(RuntimeError, match="connection reset"):
|
||||
list(store.iter_all())
|
||||
Reference in New Issue
Block a user