Compare commits

..
Author SHA1 Message Date
Zohaib Hassnain 9b74838974 making it clean 2026-08-31 13:42:24 +05:00
Zohaib Hassnain dfca24694d fix(vector_store): raise instead of truncating when a qdrant scan cannot advance 2026-08-31 13:39:46 +05:00
Zohaib Hassnain ec9e63e16f fix(vector_store): drop qdrant migrate wiring, keep iter_all only
VectorStore cannot actually migrate to or from qdrant yet. _init_backend_store constructs QdrantStore without connecting or selecting a collection, so reads raise a Collection not initialized error, and the facade store_vectors dispatches only to add/add_vectors while QdrantStore exposes insert_vectors, so writes raise NotImplementedError.

Both are pre-existing facade gaps that nothing had exposed, since migrate previously only allowed faiss/sqlite/pgvector. Adding qdrant to the allowlist claimed support that does not work end to end, so it is removed along with the dimension inference that only fires for backends missing a .dimension attribute. Tracked separately; this PR keeps just the iter_all primitive.
2026-08-31 03:00:56 +05:00
Zohaib Hassnain 274d5d1195 feat(vector_store): add iter_all enumeration and wire up qdrant migration 2026-08-31 02:34:26 +05:00
11 changed files with 323 additions and 34 deletions
+1 -1
View File
@@ -72,7 +72,7 @@ jobs:
diff \
<(grep -E '^[a-zA-Z0-9._-]+==' requirements-ci.txt | sed 's/ \\$//') \
<(grep -E '^[a-zA-Z0-9._-]+==' /tmp/requirements-ci-check.txt)
- run: pip install build==1.6.0
- run: pip install build
# wheel is build-time only (not in requirements-ci.txt) — install the
# same pinned version [build-system] declares so --no-isolation works.
- run: pip install wheel==0.48.0
+3 -17
View File
@@ -16,7 +16,7 @@ jobs:
cancel-in-progress: false
permissions:
contents: write # for the GitHub Release
id-token: write # for PyPI Trusted Publishing (OIDC), attestation signing, and Sigstore
id-token: write # for PyPI Trusted Publishing (OIDC) and attestation signing
attestations: write # for SLSA build provenance
# If you add another job to this workflow, give it its own explicit
# `permissions:` block rather than relying on the workflow-level default
@@ -40,7 +40,7 @@ jobs:
# build runs against the same versions CI tests against.
- name: Install pinned build dependencies
run: pip install -r requirements-ci.txt
- run: pip install build==1.6.0
- run: pip install build
# wheel is build-time only (not in requirements-ci.txt) — install the
# same pinned version [build-system] declares so --no-isolation works.
- run: pip install wheel==0.48.0
@@ -71,21 +71,7 @@ jobs:
uses: actions/attest-build-provenance@4d101475d8b20a2381f78447822ac1eab6504dd8 # v4
with:
subject-path: 'dist/*'
# attest-build-provenance publishes to the GH attestations API only, which
# OpenSSF Scorecard's Signed-Releases check does not inspect - it looks for
# signature files attached as release assets. Sign here too so
# `dist/*.sigstore.json` bundles ship alongside the wheel/sdist on the
# GitHub Release itself.
- name: Sign artifacts with Sigstore
uses: sigstore/gh-action-sigstore-python@790bc6befb9d733738f18d8f895854b453640ec9 # v3.5.0
with:
inputs: |
dist/*.whl
dist/*.tar.gz
- uses: softprops/action-gh-release@efb35369e0ad2afab669f228072c1b0d510eae64 # v3.0.3
with:
files: |
dist/*.whl
dist/*.tar.gz
dist/*.sigstore.json
files: dist/*
- uses: pypa/gh-action-pypi-publish@dc37677b2e1c63e2034f94d8a5b11f265b73ba33 # release/v1
+1 -1
View File
@@ -52,7 +52,7 @@ jobs:
# Tooling AFTER the pinned set: installing safety/bandit/semgrep/jq
# first lets the pinned requirements overwrite their transitive deps
# (e.g. rich), which breaks the safety CLI at runtime.
pip install safety==3.8.1 bandit==1.9.4 semgrep==1.175.0 jq==1.12.0
pip install safety bandit semgrep jq
- name: Run Safety Check (Package Vulnerabilities)
run: |
+1 -1
View File
@@ -37,6 +37,6 @@ jobs:
# pyproject.toml changes under review. The schedule/workflow_dispatch
# runs stay non-blocking until a full pass over pre-existing findings
# across the whole [all] tree has been done.
- run: pip install pip-audit==2.10.1
- run: pip install pip-audit
- run: pip-audit -r requirements-ci.txt
continue-on-error: ${{ github.event_name != 'pull_request' }}
+2 -2
View File
@@ -1,5 +1,5 @@
# syntax=docker/dockerfile:1
FROM node:26-alpine@sha256:2d984a15c9b54fd0aeb608b8e0d0d83529eb34d2966db27a1fb4f1edc3d298a3 AS frontend-builder
FROM node:26-alpine AS frontend-builder
WORKDIR /app
COPY explorer/package*.json ./explorer/
@@ -9,7 +9,7 @@ RUN npm ci
COPY explorer/ ./
RUN mkdir -p /app/semantica && npm run build
FROM python:3.13-slim@sha256:7ce4b6dfe35e55397b7cda544f8a13f191b7ae28dc5aad71fe664dbc9bc2623f AS runtime
FROM python:3.13-slim AS runtime
ENV PYTHONDONTWRITEBYTECODE=1 \
PYTHONUNBUFFERED=1 \
+1 -8
View File
@@ -49,14 +49,7 @@ dependencies = [
"scipy>=1.13.1",
"scikit-learn>=1.7.2",
"umap-learn>=0.5.12",
# thinc (spacy's core dep) dropped Python 3.9 wheels at 8.3.10, and later
# spacy patch releases (3.8.8+) require thinc>=8.3.9-only-on-3.10+ ranges,
# which forces a source build that fails outright on 3.9 (see Install
# Matrix run history). Capping both keeps 3.9 on the last wheel-compatible
# pair; 3.10+ is left unconstrained to always get the latest spacy/thinc.
"spacy>=3.4.0,<3.8.8; python_version < '3.10'",
"spacy>=3.4.0; python_version >= '3.10'",
"thinc<8.3.5; python_version < '3.10'",
"spacy>=3.4.0",
"transformers>=4.20.0",
"torch>=1.13.1",
"sentence-transformers>=2.2.0",
+3 -3
View File
@@ -782,9 +782,9 @@ click-repl==0.3.0 \
--hash=sha256:17849c23dba3d667247dc4defe1757fff98694e90fe37474f3feebb69ced26a9 \
--hash=sha256:fb7e06deb8da8de86180a33a9da97ac316751c094c6899382da7feeeeb51b812
# via celery
cloudpathlib==0.25.0 \
--hash=sha256:63612e17778c5e3a51b472def8d785d0aaaf347486d6b6786dc7be627556d4c6 \
--hash=sha256:8faef3ed3a0dd71d134e8617b4fdc5ce56a12a6b485c080cfe80106e5f1d1f5d
cloudpathlib==0.24.0 \
--hash=sha256:b1c51e2d2ec7dc4fed6538991f4aea849d6cf11a7e6b9069f86e461aa1f9b5b4 \
--hash=sha256:c521a984e77b47e656fe78e20a7e3e260e0ab45fc69e33ac01094227c979e34a
# via weasel
colorlog==6.12.0 \
--hash=sha256:2a7924c1dadf18b22a0eb8b06d1c7b01d5341707ec1641eb6fcc4fde0c3e8e5f \
+59
View File
@@ -604,6 +604,65 @@ 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 scroll cursor.
Paginates by point-ID cursor rather than row offset, which is why this
exists instead of scan_vectors(offset, limit). An integer offset is a
point ID, not a rank.
Assumes a single unnamed vector per point, as insert_vectors() and
get_vector() already do. Named and multi-vector collections are not
handled.
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, or
if the scan cannot advance past a full page.
"""
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,
}
# A final page can carry records alongside a null cursor, so they
# are yielded above before stopping. Passing offset=None back to
# scroll() would restart from the beginning, not continue.
if next_offset is None:
return
# A cursor without records means the scan cannot advance, which is
# truncation rather than completion.
if not records:
raise ProcessingError(
"Qdrant returned an empty page alongside a continuation "
"cursor, so the scan cannot advance. Refusing to return a "
"truncated scan."
)
def delete_vectors(
self, point_ids: List[Union[str, int]], **options
) -> Dict[str, Any]:
+11 -1
View File
@@ -867,12 +867,22 @@ class VectorStore:
"""
Iterate over every stored vector, one page at a time.
Cursor-based backends expose iter_all() because they cannot support a
positional offset; it takes precedence when present. Everything else
falls through to the scan_vectors() offset loop.
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)
+146
View File
@@ -0,0 +1,146 @@
"""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 continues from the previous page's cursor."""
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():
"""Records and a null cursor can arrive together; those records must still
be yielded or every scan loses its tail."""
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_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"))
with pytest.raises(ProcessingError, match="cannot advance"):
list(store.iter_all())
@patch("semantica.vector_store.qdrant_store.QDRANT_AVAILABLE", True)
def test_iter_all_raises_when_collection_not_initialized():
"""Must fail loudly: an empty scan reads the same as an empty source."""
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,34 @@ class _NonScanningBackendStore:
"""Fake persistent backend store without any scan capability."""
class _IterAllBackendStore:
"""Fake cursor-based store: iter_all() only, no usable scan_vectors()."""
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 +221,72 @@ 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."""
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):
# 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):
# Silently yielding nothing would read as an empty source (#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
# ---------------------------------------------------------------------------