diff --git a/docs/en/guides/01-configuration.md b/docs/en/guides/01-configuration.md
index 46cf0d77f6..2a9ccac670 100644
--- a/docs/en/guides/01-configuration.md
+++ b/docs/en/guides/01-configuration.md
@@ -1386,7 +1386,7 @@ Vector database storage configuration
| Parameter | Type | Description | Default |
|-----------|------|-------------|---------|
-| `backend` | str | VectorDB backend type: 'local' (file-based), 'http' (remote service), 'volcengine' (cloud VikingDB), 'vikingdb' (private deployment), or 'cuvs' (local storage + GPU dense search) | "local" |
+| `backend` | str | VectorDB backend type: 'local' (file-based), 'http' (remote service), 'volcengine' (cloud VikingDB), 'vikingdb' (private deployment), 'qdrant' (REST), or 'cuvs' (local storage + GPU dense search) | "local" |
| `name` | str | VectorDB collection name | "context" |
| `url` | str | Remote service URL for 'http' type (e.g., 'http://localhost:5000') | null |
| `project_name` | str | Project name (alias project) | "default" |
@@ -1395,6 +1395,7 @@ Vector database storage configuration
| `sparse_weight` | float | Sparse weight for hybrid vector search, only effective when using hybrid index | 0.0 |
| `volcengine` | object | 'volcengine' type VikingDB configuration | - |
| `vikingdb` | object | 'vikingdb' type private deployment configuration | - |
+| `qdrant` | object | Qdrant REST URL, API key, timeout, named vector names, and optional metadata collection name | - |
| `cuvs` | object | NVIDIA cuVS configuration for the 'cuvs' backend and the opt-in memory-aware auto mode on 'local'; see the [cuVS guide](./16-cuvs.md) | - |
Default local mode
@@ -1440,9 +1441,52 @@ acl_inherited_grants
Each element uses `{mask}:{principal}`: `1` means `read`, `3` means `write`, and `7` means `manage`.
-Local backends add the fields to an existing collection and rebuild the scalar index during startup. Existing records are not rewritten; missing ACL fields read as `acl_enabled=false` and empty lists.
+Local, cuVS, and Qdrant backends add the fields to an existing collection and rebuild/update the scalar index during startup. Existing records are not rewritten; missing ACL fields read as `acl_enabled=false` and empty lists.
-For existing remote collections, including Volcengine VikingDB, provision these fields and scalar indexes before startup; OpenViking validates but does not alter the remote schema. Volcengine API-key data-plane mode also requires the context collection and configured index to exist. See [Resource Access Control (ACL)](../concepts/15-acl.md) for permission semantics.
+For other existing remote collections, including Volcengine VikingDB, provision these fields and scalar indexes before startup; OpenViking validates but does not alter the remote schema. Volcengine API-key data-plane mode also requires the context collection and configured index to exist. See [Resource Access Control (ACL)](../concepts/15-acl.md) for permission semantics.
+
+
+Qdrant REST
+
+Qdrant uses the standard-library REST transport; no `qdrant-client` dependency is
+required. Set `sparse_weight` to `0` for dense-only mode, or to a value in
+`(0, 1]` to enable named sparse vectors and client-side weighted RRF hybrid
+search:
+
+```json
+{
+ "storage": {
+ "vectordb": {
+ "backend": "qdrant",
+ "url": "http://127.0.0.1:6333",
+ "project": "default",
+ "name": "context",
+ "dimension": 1536,
+ "sparse_weight": 0.5,
+ "qdrant": {
+ "api_key": "optional-key",
+ "timeout_seconds": 10,
+ "dense_vector_name": "vector",
+ "sparse_vector_name": "sparse_vector"
+ }
+ }
+ }
+}
+```
+
+OpenViking stores a metadata marker and sparse term dictionary in a deterministic
+sidecar collection. Existing Qdrant collections without that marker fail closed
+instead of being adopted. URI scope metadata and account/tag filters are retained;
+`Contains` and server-side content grep are unsupported, so grep uses the
+filesystem fallback (`USE_CONTENT_FIELD=False`).
+
+For live coverage, set `QDRANT_URL` and optionally `QDRANT_API_KEY`, then run:
+
+```bash
+QDRANT_URL=http://127.0.0.1:6333 \
+ pytest --confcutdir=tests/storage -q tests/storage/test_qdrant_integration.py
+```
+
openGauss
@@ -1477,7 +1521,6 @@ In the official container, the initial `omm` user may be restricted for remote l
Set `mode` to `"distributed"` for openGauss distributed deployments; OpenViking will attempt to mark metadata tables as reference tables and distribute collection tables by `id`.
-
## Config Files
OpenViking uses two config files:
diff --git a/docs/superpowers/plans/2026-08-20-qdrant-integration.md b/docs/superpowers/plans/2026-08-20-qdrant-integration.md
new file mode 100644
index 0000000000..011dee26b9
--- /dev/null
+++ b/docs/superpowers/plans/2026-08-20-qdrant-integration.md
@@ -0,0 +1,211 @@
+# Qdrant Integration Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Add phase 1 dense and phase 2 sparse Qdrant support behind the current OpenViking `CollectionAdapter` contract.
+
+**Architecture:** Keep all Qdrant-specific behavior in a REST collection implementation and a thin adapter. Store path metadata and stable IDs in the data collection, and store the OpenViking marker plus sparse term dictionary in a deterministic sidecar collection. Reuse the existing Filter AST and tenant-aware upper layer without changing its API.
+
+**Tech Stack:** Python standard library (`urllib`, `json`, `hashlib`, `uuid`, `threading`), OpenViking `Collection`/`ICollection`, Qdrant REST API, pytest.
+
+**Spec:** `docs/superpowers/specs/2026-08-20-qdrant-integration-design.md`
+
+## Global Constraints
+
+- Do not modify or reset the user's pre-existing `.gitignore` change.
+- Do not reintroduce `qdrant-client` or use lossy sparse hashing.
+- Do not silently drop filters or sparse vectors.
+- Preserve `CollectionAdapter` and upper-layer search APIs.
+- Every production behavior is introduced by a failing test first.
+- Keep `USE_CONTENT_FIELD=False`.
+
+### Task 1: Add failing pure conversion and filter tests
+
+**Files:**
+- Create: `tests/storage/test_qdrant_adapter.py`
+
+**Interfaces:**
+- Tests will import `QdrantCollectionAdapter`, `QdrantCollection`,
+ `compile_qdrant_filter`, `build_qdrant_payload`, `to_qdrant_point_id`,
+ and `SparseTermDictionary`.
+
+- [x] **Step 1: Write failing tests**
+
+Cover:
+
+```python
+def test_path_payload_includes_self_and_ancestors(): ...
+def test_path_scope_depth_mapping_is_segment_aware(): ...
+def test_multi_tag_eq_is_qdrant_must_and_in_is_match_any(): ...
+def test_account_filter_is_preserved(): ...
+def test_point_id_is_deterministic_and_original_id_round_trips(): ...
+def test_sparse_term_collision_raises_instead_of_merging(): ...
+```
+
+- [x] **Step 2: Run the focused tests**
+
+Run: `uv run --project . pytest -q tests/storage/test_qdrant_adapter.py`
+
+Expected: collection/import failure because the Qdrant implementation does not
+exist yet. If dependency installation is unavailable, run the pure module
+through the repository's available Python environment and record the blocker.
+
+### Task 2: Implement Qdrant REST transport and collection lifecycle
+
+**Files:**
+- Create: `openviking/storage/vectordb/collection/qdrant_rest.py`
+- Create: `openviking/storage/vectordb/collection/qdrant_collection.py`
+- Modify: `openviking/storage/vectordb/collection/__init__.py`
+
+**Interfaces:**
+- `QdrantRestClient.request(method, path, body=None, params=None) -> dict`
+- `QdrantCollection(ICollection)`
+- `QdrantCollection.create_remote_collection(schema, sparse_enabled)`
+- `QdrantCollection.has_openviking_metadata() -> bool`
+
+- [x] **Step 1: Add the minimal HTTP client**
+
+Use `urllib.request`, JSON, optional `api-key`, bounded error bodies, and a
+single configurable timeout. Keep the transport injectable for tests.
+
+- [x] **Step 2: Implement collection existence and metadata**
+
+Implement Qdrant collection discovery, create/delete, metadata sidecar creation,
+schema marker validation, and index creation.
+
+- [x] **Step 3: Implement data methods**
+
+Implement `upsert_data`, `update_data`, `fetch_data`, `delete_data`,
+`delete_all_data`, `aggregate_data`, and `get_meta_data` with the existing
+`SearchResult`, `DataItem`, and `AggregateResult` dataclasses.
+
+- [x] **Step 4: Run transport/lifecycle tests**
+
+Run: `uv run --project . pytest -q tests/storage/test_qdrant_adapter.py -k 'transport or lifecycle'`
+
+### Task 3: Implement phase 1 adapter and filter mapping
+
+**Files:**
+- Create: `openviking/storage/vectordb_adapters/qdrant_adapter.py`
+- Modify: `openviking/storage/vectordb_adapters/factory.py`
+- Modify: `openviking_cli/utils/config/vectordb_config.py`
+
+**Interfaces:**
+- `QdrantCollectionAdapter.from_config(config)`
+- `compile_qdrant_filter(expr)`
+- `build_qdrant_payload(record)`
+- `to_qdrant_point_id(value)`
+
+- [x] **Step 1: Implement deterministic IDs and path payloads**
+
+Normalize `viking://` URIs to `/...`, include self and ancestors in
+`scope_roots`, and reject malformed path scope input.
+
+- [x] **Step 2: Implement AST mapping**
+
+Map `And`, `Or`, `Eq`, `In`, `Range`, `TimeRange`, and `PathScope` to Qdrant
+filters. Reject unsupported `Contains` with `NotImplementedError` rather than
+changing its substring semantics.
+
+- [x] **Step 3: Implement adapter factory/config wiring**
+
+Add standard backend name `qdrant`, a nested config object for URL/auth/vector
+names/timeout, and registry wiring. Preserve dotted custom class paths.
+
+- [x] **Step 4: Run phase 1 tests**
+
+Run: `uv run --project . pytest -q tests/storage/test_qdrant_adapter.py`
+
+### Task 4: Implement phase 2 durable sparse term dictionary
+
+**Files:**
+- Modify: `openviking/storage/vectordb/collection/qdrant_collection.py`
+- Modify: `openviking/storage/vectordb_adapters/qdrant_adapter.py`
+- Modify: `tests/storage/test_qdrant_adapter.py`
+
+**Interfaces:**
+- `SparseTermDictionary.index_for(term) -> int`
+- `SparseTermDictionary.encode(vector) -> dict[str, list[...]]`
+- `QdrantCollection.encode_sparse_vector(vector)`
+
+- [x] **Step 1: Add failing sparse tests**
+
+Verify stable term indices, sidecar persistence, dense+sparse conversion, and
+collision rejection.
+
+- [x] **Step 2: Implement deterministic collision-checked Qdrant-compatible indices**
+
+Derive a positive uint32 candidate from SHA-256 because Qdrant sparse indices
+are uint32 at the REST/protobuf boundary. Persist each term/index pair in the
+sidecar collection. Query existing records for the candidate index; raise a
+collision error when a different term owns it.
+
+- [x] **Step 3: Wire sparse upsert and search**
+
+Convert OpenViking `{term: weight}` dictionaries to named Qdrant sparse vectors,
+and reject sparse operations when the collection was created dense-only.
+
+- [x] **Step 4: Run phase 2 tests**
+
+Run: `uv run --project . pytest -q tests/storage/test_qdrant_adapter.py -k sparse`
+
+### Task 5: Add optional live Qdrant integration coverage
+
+**Files:**
+- Create: `tests/storage/test_qdrant_integration.py`
+- Modify: `tests/README.md`
+
+- [x] **Step 1: Write environment-gated tests**
+
+Use `QDRANT_URL` and optional `QDRANT_API_KEY`; skip cleanly without a live
+server. Exercise create, upsert, path/tag filters, count, sparse search, and
+drop.
+
+- [x] **Step 2: Run the integration tests when available**
+
+Run: `QDRANT_URL=http://127.0.0.1:6333 uv run --project . pytest -q tests/storage/test_qdrant_integration.py`
+
+### Task 6: Verify, document, and commit
+
+**Files:**
+- Modify: `openviking/storage/vectordb_adapters/README.md`
+- Modify: `docs/en/guides/01-configuration.md`
+- Modify: `docs/zh/guides/01-configuration.md`
+
+- [x] **Step 1: Add configuration and capability docs**
+
+Document Qdrant URL/auth, vector names, dense-only versus sparse-enabled mode,
+metadata sidecar behavior, and the explicit unsupported-filter policy.
+
+- [x] **Step 2: Run final verification**
+
+Run:
+
+```bash
+uv run --project . pytest -q tests/storage/test_qdrant_adapter.py
+uv run --project . pytest -q tests/storage tests/unit
+uv run --project . ruff check openviking/storage/vectordb openviking/storage/vectordb_adapters tests/storage/test_qdrant_adapter.py
+git diff --check
+```
+
+- [x] **Step 3: Review the diff and commit**
+
+```bash
+git status --short
+git diff --stat
+git commit -am "feat: add qdrant vector backend"
+```
+
+## Verification notes (2026-08-21)
+
+- Focused adapter coverage: `45 passed`; transport/lifecycle: `2 passed`;
+ sparse: `9 passed`.
+- The reconstructed pre-implementation test run on `main` failed at collection
+ with the expected missing Qdrant module import.
+- Disposable live Qdrant coverage passed: `1 passed` against a local Qdrant
+ container at `http://127.0.0.1:6333`; the container was removed afterward.
+- The integration test path is `tests/storage/test_qdrant_integration.py`.
+- The repository-wide `tests/storage tests/unit` run remains red: 19 known
+ failures reproduce on `main`, with additional environment/order-dependent
+ failures still unclassified. This is recorded as a repository-health
+ boundary, not a Qdrant implementation blocker.
diff --git a/docs/superpowers/specs/2026-08-20-qdrant-integration-design.md b/docs/superpowers/specs/2026-08-20-qdrant-integration-design.md
new file mode 100644
index 0000000000..77282d8529
--- /dev/null
+++ b/docs/superpowers/specs/2026-08-20-qdrant-integration-design.md
@@ -0,0 +1,100 @@
+# OpenViking Qdrant Integration Design
+
+## Goal
+
+Add a first-party Qdrant vector backend that preserves the current
+`CollectionAdapter` contract, OpenViking path/tag/tenant semantics, and
+hybrid sparse-vector correctness without reverting the removed implementation.
+
+## Scope
+
+### Phase 1
+
+- Qdrant REST transport implemented with the Python standard library.
+- Collection create/load/drop and index lifecycle.
+- Dense vector upsert, fetch, delete, count, scalar ordering, and search.
+- OpenViking filter AST mapping:
+ - `And`, `Or`, `Eq`, `In`, `Range`, `TimeRange`, `PathScope`.
+ - `Contains` is rejected unless a caller uses a supported Qdrant text field
+ explicitly; the adapter never silently changes substring semantics.
+- Segment-aware path scopes using `uri`, `uri_depth`, and `scope_roots`.
+- Exact multi-tag AND and `In` semantics for `search_tags`.
+- Account isolation through the existing `_SingleAccountBackend` filter path.
+- Deterministic UUID point IDs with the original OpenViking ID in payload.
+- Durable OpenViking metadata marker; existing unmarked collections fail closed.
+- `USE_CONTENT_FIELD = False`; grep remains filesystem-backed.
+
+### Phase 2
+
+- Named Qdrant sparse vector support.
+- Durable sparse-term dictionary stored in a sidecar Qdrant collection.
+- Stable SHA-256-derived positive uint32 term IDs (the Qdrant REST/protobuf
+ sparse-index limit) with collision detection and an explicit error on
+ collision; no lossy hash merging.
+- Dense-only configurations continue to work without creating sparse metadata.
+- A non-zero sparse query against a backend configured without sparse support
+ fails explicitly.
+
+## Non-goals
+
+- No revert of PR #3872.
+- No Qdrant Python SDK dependency; REST is sufficient for the contract.
+- No migration or implicit takeover of existing unmarked Qdrant collections.
+- No server-side full-text content index in Qdrant.
+- No change to upper-layer search APIs or Filter AST.
+- No distributed metadata transaction protocol beyond Qdrant point-level
+ idempotence and collision detection.
+
+## Data model
+
+The data collection stores:
+
+- Dense vector under a configurable named vector (default `vector`).
+- Sparse vector under a configurable named vector (default `sparse_vector`).
+- OpenViking fields from the collection schema.
+- `uri_depth` as an integer payload.
+- `scope_roots` as an array of exact normalized path strings.
+- `_openviking_original_id` as the original string ID.
+
+The metadata collection stores:
+
+- A fixed OpenViking schema marker point.
+- One point per sparse term containing `term` and `index`.
+- The deterministic point ID is derived from the term; a matching index with a
+ different term is treated as a collision and raises.
+
+## Filter mapping
+
+Qdrant `must`/`should` clauses carry the OpenViking AST structure:
+
+| OpenViking expression | Qdrant representation |
+| --- | --- |
+| `And` | `must` |
+| `Or` | `should` |
+| `Eq` | `match.value` |
+| `In` | `match.any` for multiple values |
+| `Range` / `TimeRange` | `range` |
+| `PathScope(depth=0)` | exact `uri` match |
+| `PathScope(depth<0)` | `scope_roots` exact element match |
+| finite `PathScope` | `scope_roots` match plus `uri_depth` upper bound |
+
+Multiple `Eq("search_tags", ...)` expressions remain separate `must` clauses,
+so tag filters are AND rather than OR.
+
+## Error handling
+
+- HTTP failures become a backend-specific `QdrantError` carrying method, path,
+ status, and a bounded response body.
+- Missing OpenViking metadata refuses collection loading.
+- Unsupported filters and sparse requests fail before sending a lossy query.
+- Invalid vector dimensions and malformed sparse values fail at conversion.
+- Qdrant response shape mismatches fail loudly rather than returning empty
+ results.
+
+## Testing
+
+- Pure unit tests for path payload construction and filter compilation.
+- Pure unit tests for deterministic IDs and sparse term collision handling.
+- Fake-transport tests for collection CRUD, search, count, and index creation.
+- Optional integration tests enabled by `QDRANT_URL`, skipped otherwise.
+- Existing OpenViking unit tests remain unchanged.
diff --git a/docs/zh/guides/01-configuration.md b/docs/zh/guides/01-configuration.md
index 6a2a23ad4b..933d6d6418 100644
--- a/docs/zh/guides/01-configuration.md
+++ b/docs/zh/guides/01-configuration.md
@@ -1472,7 +1472,7 @@ Redis Sentinel 分别配置数据节点和 Sentinel 的 ACL：
| 参数 | 类型 | 说明 | 默认值 |
|------|------|------|--------|
-| `backend` | str | VectorDB 后端类型: 'local'（基于文件）, 'http'（远程服务）, 'volcengine'（云上 VikingDB）, 'vikingdb'（私有部署）或 'cuvs'（本地存储 + GPU dense search） | "local" |
+| `backend` | str | VectorDB 后端类型: 'local'（基于文件）, 'http'（远程服务）, 'volcengine'（云上 VikingDB）, 'vikingdb'（私有部署）, 'qdrant'（REST）或 'cuvs'（本地存储 + GPU dense search） | "local" |
| `name` | str | VectorDB 的集合名称 | "context" |
| `url` | str | 'http' 类型的远程服务 URL（例如 'http://localhost:5000'） | null |
| `project_name` | str | 项目名称（别名 project） | "default" |
@@ -1481,6 +1481,7 @@ Redis Sentinel 分别配置数据节点和 Sentinel 的 ACL：
| `sparse_weight` | float | 混合向量搜索的稀疏权重，仅在使用混合索引时生效 | 0.0 |
| `volcengine` | object | 'volcengine' 类型的 VikingDB 配置 | - |
| `vikingdb` | object | 'vikingdb' 类型的私有部署配置 | - |
+| `qdrant` | object | Qdrant REST 地址、API key、超时、named vector 名称和可选 metadata collection 名称 | - |
| `cuvs` | object | NVIDIA cuVS 配置，也用于在 'local' 下显式开启显存感知自动模式，参见 [cuVS 使用指南](./16-cuvs.md) | - |
默认使用本地模式
@@ -1526,9 +1527,51 @@ acl_inherited_grants
每个元素使用 `{mask}:{principal}` 格式，其中 `1` 表示 `read`、`3` 表示 `write`、`7` 表示 `manage`。
-本地 backend 会在启动时为存量 collection 增加字段并重建标量索引。旧记录不做全量回填；缺失 ACL 字段按 `acl_enabled=false` 和空列表读取。
+本地、cuVS 和 Qdrant backend 会在启动时为存量 collection 增加字段并重建或更新标量索引。旧记录不做全量回填；缺失 ACL 字段按 `acl_enabled=false` 和空列表读取。
-火山向量库等远端 backend 的存量 collection 需要由部署方预先添加这些字段和 scalar index，OpenViking 只校验 schema。`volcengine` API key 数据面模式还要求 context collection 和配置的 index 已存在。权限模型详见 [资源访问控制（ACL）](../concepts/15-acl.md)。
+其他远端 backend（包括火山向量库）的存量 collection 需要由部署方预先添加这些字段和 scalar index，OpenViking 只校验 schema。`volcengine` API key 数据面模式还要求 context collection 和配置的 index 已存在。权限模型详见 [资源访问控制（ACL）](../concepts/15-acl.md)。
+
+
+Qdrant REST
+
+Qdrant 使用 Python 标准库 REST transport，不需要新增 `qdrant-client`
+依赖。`sparse_weight=0` 表示 dense-only；设置为 `(0, 1]` 内的值会启用
+named sparse vector 和客户端 weighted RRF hybrid search：
+
+```json
+{
+ "storage": {
+ "vectordb": {
+ "backend": "qdrant",
+ "url": "http://127.0.0.1:6333",
+ "project": "default",
+ "name": "context",
+ "dimension": 1536,
+ "sparse_weight": 0.5,
+ "qdrant": {
+ "api_key": "optional-key",
+ "timeout_seconds": 10,
+ "dense_vector_name": "vector",
+ "sparse_vector_name": "sparse_vector"
+ }
+ }
+ }
+}
+```
+
+OpenViking 会在确定性的 sidecar collection 中保存 metadata marker 和 sparse
+term dictionary。没有 marker 的既有 Qdrant collection 会 fail closed，不会被
+隐式接管。URI scope metadata、account 隔离和多 tag 过滤会保留；`Contains`
+及服务端 content grep 不支持，因此 grep 继续使用 filesystem fallback
+（`USE_CONTENT_FIELD=False`）。
+
+要运行 live coverage，请设置 `QDRANT_URL`（可选 `QDRANT_API_KEY`）：
+
+```bash
+QDRANT_URL=http://127.0.0.1:6333 \
+ pytest --confcutdir=tests/storage -q tests/storage/test_qdrant_integration.py
+```
+
openGauss
@@ -1563,8 +1606,6 @@ acl_inherited_grants
分布式 openGauss 部署可将 `mode` 设为 `"distributed"`；OpenViking 会尝试把元数据表标记为 reference table，并按 `id` 分布集合表。
-
-
## 配置文件
OpenViking 使用两个配置文件：
diff --git a/openviking/storage/collection_schemas.py b/openviking/storage/collection_schemas.py
index 33499acffc..516a92c4e4 100644
--- a/openviking/storage/collection_schemas.py
+++ b/openviking/storage/collection_schemas.py
@@ -324,9 +324,13 @@ async def init_context_collection(storage) -> bool:
missing_acl_indexes = sorted(ACL_CONTEXT_FIELDS - existing_scalar_indexes)
async def _migrate_acl_schema() -> None:
- if not missing_acl_fields and not missing_acl_indexes:
+ if (
+ not missing_acl_fields
+ and not missing_acl_indexes
+ and vectordb_cfg.backend != "qdrant"
+ ):
return
- if vectordb_cfg.backend not in {"local", "cuvs"}:
+ if vectordb_cfg.backend not in {"local", "cuvs", "qdrant"}:
raise EmbeddingConfigurationError(
"Context collection is missing ACL schema: "
f"fields={missing_acl_fields}, scalar_indexes={missing_acl_indexes}. "
diff --git a/openviking/storage/vectordb/collection/__init__.py b/openviking/storage/vectordb/collection/__init__.py
index 88bb52e7cd..9d80a6943b 100644
--- a/openviking/storage/vectordb/collection/__init__.py
+++ b/openviking/storage/vectordb/collection/__init__.py
@@ -11,6 +11,8 @@
LocalCollection,
get_or_create_local_collection,
)
+from openviking.storage.vectordb.collection.qdrant_collection import QdrantCollection
+from openviking.storage.vectordb.collection.qdrant_rest import QdrantError, QdrantRestClient
from openviking.storage.vectordb.collection.volcengine_collection import (
VolcengineCollection,
get_or_create_volcengine_collection,
@@ -25,4 +27,7 @@
"get_or_create_http_collection",
"LocalCollection",
"get_or_create_local_collection",
+ "QdrantCollection",
+ "QdrantRestClient",
+ "QdrantError",
]
diff --git a/openviking/storage/vectordb/collection/qdrant_collection.py b/openviking/storage/vectordb/collection/qdrant_collection.py
new file mode 100644
index 0000000000..ea5be8f915
--- /dev/null
+++ b/openviking/storage/vectordb/collection/qdrant_collection.py
@@ -0,0 +1,973 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: Apache-2.0
+"""Qdrant collection implementation for the OpenViking Collection contract."""
+
+from __future__ import annotations
+
+import math
+import random
+from dataclasses import dataclass
+from typing import Any
+from urllib.parse import quote
+
+from openviking.storage.vectordb.collection.collection import ICollection
+from openviking.storage.vectordb.collection.qdrant_rest import QdrantError, QdrantRestClient
+from openviking.storage.vectordb.collection.result import (
+ AggregateResult,
+ DataItem,
+ FetchDataInCollectionResult,
+ SearchItemResult,
+ SearchResult,
+)
+from openviking.storage.vectordb.qdrant_sparse import SparseTermDictionary
+from openviking.storage.vectordb.qdrant_utils import (
+ build_qdrant_payload,
+ to_qdrant_point_id,
+)
+
+_META_VERSION = 1
+_META_MARKER_ID = to_qdrant_point_id("openviking:metadata")
+_META_VECTOR_NAME = "meta"
+_INTERNAL_PAYLOAD_FIELDS = {
+ "uri_depth",
+ "scope_roots",
+}
+
+
+@dataclass
+class _Hit:
+ point_id: str
+ item: SearchItemResult
+
+
+class QdrantCollection(ICollection):
+ """REST-backed implementation of :class:`ICollection`."""
+
+ def __init__(
+ self,
+ *,
+ client: QdrantRestClient,
+ collection_name: str,
+ metadata_collection_name: str,
+ dense_vector_name: str,
+ sparse_vector_name: str,
+ vector_dim: int,
+ distance: str,
+ sparse_enabled: bool,
+ sparse_weight: float,
+ ) -> None:
+ self._client = client
+ self._collection_name = collection_name
+ self._metadata_collection_name = metadata_collection_name
+ self._dense_vector_name = dense_vector_name
+ self._sparse_vector_name = sparse_vector_name
+ self._vector_dim = int(vector_dim)
+ self._distance = self._normalize_distance(distance)
+ self._sparse_enabled = bool(sparse_enabled)
+ self._sparse_weight = float(sparse_weight)
+ self._schema: dict[str, Any] = {}
+ self._indexes: dict[str, dict[str, Any]] = {}
+ self._sparse_dictionary: SparseTermDictionary | None = None
+
+ @staticmethod
+ def _normalize_distance(distance: str) -> str:
+ value = str(distance or "cosine").strip().lower()
+ mapping = {"cosine": "Cosine", "ip": "Dot", "dot": "Dot", "l2": "Euclid", "euclid": "Euclid"}
+ if value not in mapping:
+ raise ValueError(f"Unsupported Qdrant distance metric: {distance!r}")
+ return mapping[value]
+
+ @staticmethod
+ def _result(response: dict[str, Any]) -> Any:
+ return response.get("result", response)
+
+ def _path(self, name: str, suffix: str = "") -> str:
+ return f"/collections/{quote(name, safe='')}{suffix}"
+
+ def _exists(self, name: str) -> bool:
+ try:
+ self._client.request("GET", self._path(name))
+ except QdrantError as exc:
+ if exc.status == 404:
+ return False
+ raise
+ return True
+
+ def collection_exists(self) -> bool:
+ return self._exists(self._collection_name)
+
+ def _create_collection(self, name: str, *, metadata: bool = False) -> None:
+ if self._exists(name):
+ return
+ if metadata:
+ vectors = {_META_VECTOR_NAME: {"size": 1, "distance": "Dot"}}
+ body: dict[str, Any] = {"vectors": vectors}
+ else:
+ body = {
+ "vectors": {
+ self._dense_vector_name: {
+ "size": self._vector_dim,
+ "distance": self._distance,
+ }
+ }
+ }
+ if self._sparse_enabled:
+ body["sparse_vectors"] = {self._sparse_vector_name: {}}
+ try:
+ self._client.request("PUT", self._path(name), body, params={"wait": "true"})
+ except QdrantError as exc:
+ if exc.status != 409:
+ raise
+
+ def create_remote_collection(self, metadata: dict[str, Any]) -> None:
+ self._schema = dict(metadata)
+ if self._vector_dim <= 0:
+ for field in self._schema.get("Fields", []):
+ if str(field.get("FieldType", "")).lower() == "vector" and field.get("Dim"):
+ self._vector_dim = int(field["Dim"])
+ break
+ if self._vector_dim <= 0:
+ raise ValueError("Qdrant backend requires a positive dense vector dimension")
+ self._create_collection(self._collection_name)
+ self._create_collection(self._metadata_collection_name, metadata=True)
+ self._write_metadata_marker()
+
+ def has_openviking_metadata(self) -> bool:
+ payload = self._load_metadata_marker()
+ return bool(
+ payload
+ and payload.get("_openviking_meta_version") == _META_VERSION
+ and payload.get("collection_name") == self._collection_name
+ and isinstance(payload.get("schema"), dict)
+ )
+
+ def _metadata_payload(self) -> dict[str, Any]:
+ return {
+ "_openviking_meta_version": _META_VERSION,
+ "collection_name": self._collection_name,
+ "schema": self._schema,
+ "dense_vector_name": self._dense_vector_name,
+ "sparse_vector_name": self._sparse_vector_name,
+ "vector_dim": self._vector_dim,
+ "distance": self._distance,
+ "sparse_enabled": self._sparse_enabled,
+ "sparse_weight": self._sparse_weight,
+ "indexes": self._indexes,
+ }
+
+ def _write_metadata_marker(self) -> None:
+ self._upsert_points(
+ self._metadata_collection_name,
+ [
+ {
+ "id": _META_MARKER_ID,
+ "vector": {_META_VECTOR_NAME: [0.0]},
+ "payload": self._metadata_payload(),
+ }
+ ],
+ )
+
+ def _load_metadata_marker(self) -> dict[str, Any] | None:
+ if not self._exists(self._metadata_collection_name):
+ return None
+ points = self._retrieve_points(
+ self._metadata_collection_name,
+ [_META_MARKER_ID],
+ with_vectors=False,
+ )
+ if not points:
+ return None
+ payload = points[0].get("payload")
+ return payload if isinstance(payload, dict) else None
+
+ def _ensure_loaded(self) -> None:
+ if self._schema:
+ return
+ marker = self._load_metadata_marker()
+ if not marker:
+ raise RuntimeError(
+ f"Qdrant collection {self._collection_name!r} is missing OpenViking metadata"
+ )
+ if marker.get("collection_name") != self._collection_name:
+ raise RuntimeError(
+ f"Qdrant metadata collection does not belong to {self._collection_name!r}"
+ )
+ self._schema = dict(marker["schema"])
+ self._vector_dim = int(marker.get("vector_dim") or self._vector_dim)
+ self._dense_vector_name = str(
+ marker.get("dense_vector_name") or self._dense_vector_name
+ )
+ self._sparse_vector_name = str(
+ marker.get("sparse_vector_name") or self._sparse_vector_name
+ )
+ self._distance = str(marker.get("distance") or self._distance)
+ self._sparse_enabled = bool(marker.get("sparse_enabled", self._sparse_enabled))
+ if "sparse_weight" in marker:
+ self._sparse_weight = float(marker["sparse_weight"])
+ indexes = marker.get("indexes")
+ self._indexes = (
+ {str(name): dict(meta) for name, meta in indexes.items()}
+ if isinstance(indexes, dict)
+ else {}
+ )
+
+ def get_meta_data(self) -> dict[str, Any]:
+ self._ensure_loaded()
+ return dict(self._schema)
+
+ def update(
+ self,
+ fields: dict[str, Any] | list[dict[str, Any]] | None = None,
+ description: str | None = None,
+ ):
+ self._ensure_loaded()
+ if fields:
+ if isinstance(fields, list):
+ field_updates = fields
+ schema_updates = {}
+ else:
+ schema_updates = dict(fields)
+ field_updates = schema_updates.pop("Fields", None)
+
+ if isinstance(field_updates, list):
+ existing_fields = {
+ field.get("FieldName"): field
+ for field in self._schema.get("Fields", [])
+ if isinstance(field, dict) and field.get("FieldName")
+ }
+ for field in field_updates:
+ if isinstance(field, dict) and field.get("FieldName"):
+ existing_fields.setdefault(field["FieldName"], field)
+ self._schema["Fields"] = list(existing_fields.values())
+ self._schema.update(schema_updates)
+ else:
+ self._schema.update(fields)
+ if description is not None:
+ self._schema["Description"] = description
+ self._write_metadata_marker()
+ return self._schema
+
+ def close(self) -> None:
+ return None
+
+ def drop(self):
+ for name in (self._collection_name, self._metadata_collection_name):
+ if self._exists(name):
+ self._client.request("DELETE", self._path(name), params={"timeout": 30})
+ self._schema.clear()
+ return True
+
+ def _field_schema(self, field: str) -> str:
+ fields = self._schema.get("Fields", [])
+ for item in fields:
+ if item.get("FieldName") != field:
+ continue
+ field_type = str(item.get("FieldType") or "").lower()
+ if field_type.startswith("list<") and field_type.endswith(">"):
+ field_type = field_type[5:-1]
+ if field_type in {
+ "int",
+ "int8",
+ "int16",
+ "int32",
+ "int64",
+ "uint",
+ "uint8",
+ "uint16",
+ "uint32",
+ "uint64",
+ }:
+ return "integer"
+ if field_type in {"float", "float16", "float32", "float64", "double"}:
+ return "float"
+ if field_type in {"bool", "boolean"}:
+ return "bool"
+ if field_type in {"date_time", "datetime"}:
+ return "datetime"
+ return "keyword"
+ return "keyword"
+
+ @staticmethod
+ def _index_fields(meta: dict[str, Any]) -> list[str]:
+ scalar_index = meta.get("ScalarIndex")
+ if isinstance(scalar_index, dict):
+ return [str(field) for field in scalar_index]
+ if isinstance(scalar_index, (list, tuple, set)):
+ return [str(field) for field in scalar_index]
+ return []
+
+ def _ensure_remote_indexes(self, meta_data: dict[str, Any]) -> None:
+ scalar_fields = list(meta_data.get("ScalarIndex") or [])
+ scalar_fields.extend(["uri_depth", "scope_roots"])
+ for field in dict.fromkeys(scalar_fields):
+ body = {
+ "field_name": field,
+ "field_schema": (
+ "integer" if field == "uri_depth" else self._field_schema(field)
+ ),
+ }
+ try:
+ self._client.request(
+ "PUT",
+ self._path(self._collection_name, "/index"),
+ body,
+ params={"wait": "true"},
+ )
+ except QdrantError as exc:
+ if exc.status != 409:
+ raise
+
+ def _delete_remote_indexes(self, fields: list[str]) -> None:
+ for field in dict.fromkeys(fields):
+ try:
+ self._client.request(
+ "DELETE",
+ self._path(
+ self._collection_name,
+ f"/index/{quote(field, safe='')}",
+ ),
+ params={"wait": "true"},
+ )
+ except QdrantError as exc:
+ if exc.status != 404:
+ raise
+
+ def create_index(self, index_name: str, meta_data: dict[str, Any]):
+ self._ensure_remote_indexes(meta_data)
+ previous_indexes = self._indexes
+ self._indexes = dict(previous_indexes)
+ self._indexes[index_name] = dict(meta_data)
+ try:
+ self._write_metadata_marker()
+ except Exception:
+ self._indexes = previous_indexes
+ raise
+ return meta_data
+
+ def has_index(self, index_name: str) -> bool:
+ return index_name in self._indexes
+
+ def get_index(self, index_name: str):
+ return self._indexes.get(index_name)
+
+ def get_index_meta_data(self, index_name: str) -> dict[str, Any]:
+ return dict(self._indexes.get(index_name, {}))
+
+ def list_indexes(self) -> list[str]:
+ return list(self._indexes)
+
+ def update_index(
+ self,
+ index_name: str,
+ scalar_index: dict[str, Any] | list[str] | None = None,
+ description: str | None = None,
+ ):
+ if index_name not in self._indexes:
+ return None
+ meta = dict(self._indexes.get(index_name, {}))
+ if scalar_index is not None:
+ meta["ScalarIndex"] = scalar_index
+ if description is not None:
+ meta["Description"] = description
+
+ old_fields = set(self._index_fields(self._indexes.get(index_name, {})))
+ other_fields = {
+ field
+ for name, item in self._indexes.items()
+ if name != index_name
+ for field in self._index_fields(item)
+ }
+ self._ensure_remote_indexes(meta)
+ self._delete_remote_indexes(
+ [
+ field
+ for field in old_fields - set(self._index_fields(meta))
+ if field not in other_fields
+ ]
+ )
+ previous_indexes = self._indexes
+ self._indexes = dict(previous_indexes)
+ self._indexes[index_name] = meta
+ try:
+ self._write_metadata_marker()
+ except Exception:
+ self._indexes = previous_indexes
+ raise
+ return meta
+
+ def drop_index(self, index_name: str):
+ removed = self._indexes.get(index_name)
+ if removed is None:
+ return True
+
+ remaining_fields = {
+ field
+ for name, meta in self._indexes.items()
+ if name != index_name
+ for field in self._index_fields(meta)
+ }
+ if any(name != index_name for name in self._indexes):
+ remaining_fields.update({"uri_depth", "scope_roots"})
+ fields_to_remove = list(
+ dict.fromkeys([*self._index_fields(removed), "uri_depth", "scope_roots"])
+ )
+ self._delete_remote_indexes(
+ [field for field in fields_to_remove if field not in remaining_fields]
+ )
+ previous_indexes = self._indexes
+ self._indexes = dict(previous_indexes)
+ self._indexes.pop(index_name, None)
+ try:
+ self._write_metadata_marker()
+ except Exception:
+ self._indexes = previous_indexes
+ raise
+ return True
+
+ def _upsert_points(self, collection_name: str, points: list[dict[str, Any]]) -> None:
+ self._client.request(
+ "PUT",
+ self._path(collection_name, "/points"),
+ {"points": points},
+ params={"wait": "true"},
+ )
+
+ def _retrieve_points(
+ self,
+ collection_name: str,
+ ids: list[str],
+ *,
+ with_vectors: bool,
+ ) -> list[dict[str, Any]]:
+ response = self._client.request(
+ "POST",
+ self._path(collection_name, "/points"),
+ {
+ "ids": ids,
+ "with_payload": True,
+ "with_vector": with_vectors,
+ },
+ )
+ result = self._result(response)
+ return result if isinstance(result, list) else []
+
+ def _scroll(
+ self,
+ collection_name: str,
+ *,
+ filter: dict[str, Any] | None,
+ limit: int = 1,
+ with_vectors: bool = False,
+ order_by: dict[str, Any] | None = None,
+ output_fields: list[str] | None = None,
+ ) -> list[dict[str, Any]]:
+ if limit <= 0:
+ return []
+
+ points: list[dict[str, Any]] = []
+ offset: Any = None
+ while len(points) < limit:
+ body: dict[str, Any] = {
+ "limit": limit - len(points),
+ "with_payload": self._payload_selector(output_fields),
+ "with_vector": with_vectors,
+ }
+ if filter:
+ body["filter"] = filter
+ if order_by:
+ body["order_by"] = order_by
+ if offset is not None:
+ body["offset"] = offset
+ response = self._client.request(
+ "POST",
+ self._path(collection_name, "/points/scroll"),
+ body,
+ )
+ result = self._result(response)
+ if not isinstance(result, dict):
+ break
+ page = result.get("points")
+ if not isinstance(page, list) or not page:
+ break
+ points.extend(point for point in page if isinstance(point, dict))
+ if len(points) >= limit:
+ break
+ offset = result.get("next_page_offset")
+ if offset is None:
+ break
+ return points[:limit]
+
+ def _point_from_record(self, record: dict[str, Any]) -> dict[str, Any]:
+ original_id = record.get("id")
+ if original_id is None:
+ raise ValueError("Qdrant upsert requires an OpenViking record id")
+ dense = record.get("vector")
+ if dense is not None:
+ if not isinstance(dense, list) or len(dense) != self._vector_dim:
+ raise ValueError(
+ f"Qdrant dense vector dimension must be {self._vector_dim}, got {len(dense) if isinstance(dense, list) else type(dense).__name__}"
+ )
+ sparse = record.get("sparse_vector")
+ vectors: dict[str, Any] = {}
+ if dense is not None:
+ dense_values = [float(value) for value in dense]
+ if not all(math.isfinite(value) for value in dense_values):
+ raise ValueError("Qdrant dense vector values must be finite")
+ vectors[self._dense_vector_name] = dense_values
+ if sparse:
+ vectors[self._sparse_vector_name] = self.encode_sparse_vector(sparse)
+ if not vectors:
+ raise ValueError("Qdrant record requires a dense or sparse vector")
+ payload = build_qdrant_payload(record)
+ return {
+ "id": to_qdrant_point_id(original_id),
+ "vector": vectors,
+ "payload": payload,
+ }
+
+ def upsert_data(self, data_list: list[dict[str, Any]], ttl: int = 0):
+ del ttl
+ if not data_list:
+ return {"status": "ok"}
+ self._upsert_points(self._collection_name, [self._point_from_record(item) for item in data_list])
+ return {"status": "ok"}
+
+ def _payload_to_record(self, point: dict[str, Any]) -> dict[str, Any]:
+ payload = dict(point.get("payload") or {})
+ original_id = payload.pop("_openviking_original_id", None)
+ if original_id is None:
+ original_id = point.get("id")
+ for field in _INTERNAL_PAYLOAD_FIELDS:
+ payload.pop(field, None)
+ payload["id"] = original_id
+ return payload
+
+ def _vectors_to_record(self, point: dict[str, Any]) -> dict[str, Any]:
+ record = self._payload_to_record(point)
+ vectors = point.get("vector") or point.get("vectors") or {}
+ if isinstance(vectors, dict):
+ dense = vectors.get(self._dense_vector_name)
+ sparse = vectors.get(self._sparse_vector_name)
+ if dense is not None:
+ record["vector"] = dense
+ if sparse is not None:
+ record["sparse_vector"] = self._decode_sparse_vector(sparse)
+ return record
+
+ def _decode_sparse_vector(self, value: Any) -> dict[str, float]:
+ if not isinstance(value, dict):
+ raise ValueError("Qdrant sparse vector must be a mapping")
+ indices = value.get("indices")
+ values = value.get("values")
+ if (
+ not isinstance(indices, list)
+ or not isinstance(values, list)
+ or len(indices) != len(values)
+ ):
+ raise ValueError(
+ "Qdrant sparse vector indices and values must be lists of equal length"
+ )
+ self._get_sparse_dictionary()
+ result: dict[str, float] = {}
+ for index, weight in zip(indices, values, strict=True):
+ term = self._resolve_sparse_index(int(index))
+ if term is None:
+ raise ValueError(f"unknown sparse term index: {index}")
+ result[term] = float(weight)
+ return result
+
+ def update_data(self, data_list: list[dict[str, Any]]):
+ pending: list[tuple[str, dict[str, Any]]] = []
+ missing: list[Any] = []
+ for item in data_list:
+ if "id" not in item:
+ raise ValueError("primary key 'id' is required for update")
+ record_id = item["id"]
+ points = self._retrieve_points(
+ self._collection_name,
+ [to_qdrant_point_id(record_id)],
+ with_vectors=True,
+ )
+ if not points:
+ missing.append(record_id)
+ continue
+ merged = self._vectors_to_record(points[0])
+ merged.update(item)
+ merged["id"] = record_id
+ pending.append((str(record_id), merged))
+
+ if missing:
+ raise ValueError(f"record not found for primary key(s): {missing}")
+
+ updated: list[str] = []
+ for record_id, merged in pending:
+ self.upsert_data([merged])
+ updated.append(record_id)
+ return updated
+
+ def fetch_data(self, primary_keys: list[Any]) -> FetchDataInCollectionResult:
+ if not primary_keys:
+ return FetchDataInCollectionResult()
+ points = self._retrieve_points(
+ self._collection_name,
+ [to_qdrant_point_id(value) for value in primary_keys],
+ with_vectors=False,
+ )
+ items = [
+ DataItem(
+ id=self._payload_to_record(point).get("id"),
+ fields=self._payload_to_record(point),
+ )
+ for point in points
+ ]
+ found = {item.id for item in items}
+ missing = [key for key in primary_keys if str(key) not in {str(item) for item in found}]
+ return FetchDataInCollectionResult(items=items, ids_not_exist=missing)
+
+ def delete_data(self, primary_keys: list[Any]):
+ if not primary_keys:
+ return {"status": "ok"}
+ self._client.request(
+ "POST",
+ self._path(self._collection_name, "/points/delete"),
+ {"points": [to_qdrant_point_id(value) for value in primary_keys]},
+ params={"wait": "true"},
+ )
+ return {"status": "ok"}
+
+ def delete_all_data(self):
+ self._client.request(
+ "POST",
+ self._path(self._collection_name, "/points/delete"),
+ {"filter": {}},
+ params={"wait": "true"},
+ )
+ return True
+
+ def aggregate_data(
+ self,
+ index_name: str,
+ op: str = "count",
+ field: str | None = None,
+ filters: dict[str, Any] | None = None,
+ cond: dict[str, Any] | None = None,
+ ) -> AggregateResult:
+ del index_name, field, cond
+ if op != "count":
+ raise NotImplementedError(f"Qdrant aggregate operation is unsupported: {op}")
+ response = self._client.request(
+ "POST",
+ self._path(self._collection_name, "/points/count"),
+ {"filter": filters or {}, "exact": True},
+ )
+ result = self._result(response)
+ count = result.get("count", 0) if isinstance(result, dict) else 0
+ return AggregateResult(agg={"_total": int(count)}, op="count")
+
+ def _payload_selector(self, output_fields: list[str] | None) -> bool | dict[str, list[str]]:
+ if output_fields is None:
+ return True
+ fields = list(dict.fromkeys([*output_fields, "_openviking_original_id"]))
+ return {"include": fields}
+
+ def _search_one(
+ self,
+ *,
+ vector: Any,
+ using: str,
+ filter: dict[str, Any],
+ limit: int,
+ offset: int,
+ output_fields: list[str] | None,
+ ) -> list[_Hit]:
+ query = vector.get("vector") if isinstance(vector, dict) and "vector" in vector else vector
+ body = {
+ "query": query,
+ "using": using,
+ "filter": filter,
+ "limit": limit,
+ "offset": offset,
+ "with_payload": self._payload_selector(output_fields),
+ "with_vector": False,
+ }
+ response = self._client.request(
+ "POST",
+ self._path(self._collection_name, "/points/query"),
+ body,
+ )
+ result = self._result(response)
+ if isinstance(result, dict):
+ result = result.get("points")
+ if not isinstance(result, list):
+ raise QdrantError("Qdrant search response did not contain a result list")
+ hits: list[_Hit] = []
+ for point in result:
+ if not isinstance(point, dict):
+ continue
+ record = self._payload_to_record(point)
+ hits.append(
+ _Hit(
+ point_id=str(point.get("id")),
+ item=SearchItemResult(
+ id=record.get("id"),
+ fields=record,
+ score=float(point.get("score") or 0.0),
+ ),
+ )
+ )
+ return hits
+
+ def search_by_vector(
+ self,
+ index_name: str,
+ dense_vector: list[float] | None = None,
+ limit: int = 10,
+ offset: int = 0,
+ filters: dict[str, Any] | None = None,
+ sparse_vector: dict[str, float] | None = None,
+ output_fields: list[str] | None = None,
+ ) -> SearchResult:
+ del index_name
+ if sparse_vector and not self._sparse_enabled:
+ raise ValueError("Qdrant collection was created without sparse-vector support")
+ if dense_vector is not None and len(dense_vector) != self._vector_dim:
+ raise ValueError(
+ "Qdrant dense query vector dimension must be "
+ f"{self._vector_dim}, got {len(dense_vector)}"
+ )
+ if dense_vector is not None and not all(
+ math.isfinite(float(value)) for value in dense_vector
+ ):
+ raise ValueError("Qdrant dense query vector values must be finite")
+ if sparse_vector is not None and not sparse_vector:
+ raise ValueError("Qdrant sparse query vector must not be empty")
+ qdrant_filter = filters or {}
+ if dense_vector is None and sparse_vector is None:
+ dense_vector = [random.uniform(-1, 1) for _ in range(self._vector_dim)]
+ if dense_vector is not None and sparse_vector is not None:
+ candidate_limit = max(limit + offset, limit * 2)
+ dense_hits = self._search_one(
+ vector={"vector": [float(value) for value in dense_vector]},
+ using=self._dense_vector_name,
+ filter=qdrant_filter,
+ limit=candidate_limit,
+ offset=0,
+ output_fields=output_fields,
+ )
+ sparse = self.encode_sparse_vector(sparse_vector)
+ sparse_hits = self._search_one(
+ vector={"indices": sparse["indices"], "values": sparse["values"]},
+ using=self._sparse_vector_name,
+ filter=qdrant_filter,
+ limit=candidate_limit,
+ offset=0,
+ output_fields=output_fields,
+ )
+ merged = self._weighted_rank_fusion(dense_hits, sparse_hits)
+ return SearchResult(data=[hit.item for hit in merged[offset : offset + limit]])
+ if dense_vector is not None:
+ hits = self._search_one(
+ vector={"vector": [float(value) for value in dense_vector]},
+ using=self._dense_vector_name,
+ filter=qdrant_filter,
+ limit=limit,
+ offset=offset,
+ output_fields=output_fields,
+ )
+ else:
+ sparse = self.encode_sparse_vector(sparse_vector)
+ hits = self._search_one(
+ vector={"indices": sparse["indices"], "values": sparse["values"]},
+ using=self._sparse_vector_name,
+ filter=qdrant_filter,
+ limit=limit,
+ offset=offset,
+ output_fields=output_fields,
+ )
+ return SearchResult(data=[hit.item for hit in hits])
+
+ def _weighted_rank_fusion(self, dense: list[_Hit], sparse: list[_Hit]) -> list[_Hit]:
+ alpha = min(max(self._sparse_weight, 0.0), 1.0)
+ scores: dict[str, float] = {}
+ hits: dict[str, _Hit] = {}
+ for rank, hit in enumerate(dense, start=1):
+ scores[hit.point_id] = scores.get(hit.point_id, 0.0) + (1.0 - alpha) / (60 + rank)
+ hits[hit.point_id] = hit
+ for rank, hit in enumerate(sparse, start=1):
+ scores[hit.point_id] = scores.get(hit.point_id, 0.0) + alpha / (60 + rank)
+ hits.setdefault(hit.point_id, hit)
+ ordered = sorted(hits, key=lambda point_id: scores[point_id], reverse=True)
+ return [
+ _Hit(
+ point_id=point_id,
+ item=SearchItemResult(
+ id=hits[point_id].item.id,
+ fields=hits[point_id].item.fields,
+ score=scores[point_id],
+ ),
+ )
+ for point_id in ordered
+ ]
+
+ def search_by_keywords(
+ self,
+ index_name: str,
+ keywords: list[str] | None = None,
+ query: str | None = None,
+ limit: int = 10,
+ offset: int = 0,
+ filters: dict[str, Any] | None = None,
+ output_fields: list[str] | None = None,
+ ) -> SearchResult:
+ del index_name, keywords, query, limit, offset, filters, output_fields
+ raise NotImplementedError(
+ "Qdrant does not provide OpenViking content grep; use the filesystem fallback"
+ )
+
+ def search_by_id(
+ self,
+ index_name: str,
+ id: Any,
+ limit: int = 10,
+ offset: int = 0,
+ filters: dict[str, Any] | None = None,
+ output_fields: list[str] | None = None,
+ ) -> SearchResult:
+ points = self._retrieve_points(self._collection_name, [to_qdrant_point_id(id)], with_vectors=True)
+ if not points:
+ return SearchResult()
+ record = self._vectors_to_record(points[0])
+ return self.search_by_vector(
+ index_name,
+ dense_vector=record.get("vector"),
+ sparse_vector=record.get("sparse_vector"),
+ limit=limit,
+ offset=offset,
+ filters=filters,
+ output_fields=output_fields,
+ )
+
+ def search_by_multimodal(self, *args: Any, **kwargs: Any) -> SearchResult:
+ del args, kwargs
+ raise NotImplementedError("Qdrant multimodal search is not supported")
+
+ def search_by_random(
+ self,
+ index_name: str,
+ limit: int = 10,
+ offset: int = 0,
+ filters: dict[str, Any] | None = None,
+ output_fields: list[str] | None = None,
+ ) -> SearchResult:
+ return self.search_by_vector(
+ index_name,
+ dense_vector=None,
+ sparse_vector=None,
+ limit=limit,
+ offset=offset,
+ filters=filters,
+ output_fields=output_fields,
+ )
+
+ def search_by_scalar(
+ self,
+ index_name: str,
+ field: str,
+ order: str | None = "desc",
+ limit: int = 10,
+ offset: int = 0,
+ filters: dict[str, Any] | None = None,
+ output_fields: list[str] | None = None,
+ ) -> SearchResult:
+ del index_name
+ points = self._scroll(
+ self._collection_name,
+ filter=filters,
+ limit=limit + offset,
+ with_vectors=False,
+ order_by={"key": field, "direction": "desc" if order == "desc" else "asc"},
+ output_fields=output_fields,
+ )
+ items = []
+ for point in points[offset : offset + limit]:
+ record = self._payload_to_record(point)
+ items.append(
+ SearchItemResult(
+ id=record.get("id"),
+ fields=record,
+ score=float(record.get(field) or 0.0)
+ if isinstance(record.get(field), (int, float))
+ else 0.0,
+ )
+ )
+ return SearchResult(data=items)
+
+ def _get_sparse_dictionary(self) -> SparseTermDictionary:
+ if not self._sparse_enabled:
+ raise ValueError("Qdrant collection was created without sparse-vector support")
+ if self._sparse_dictionary is None:
+ self._sparse_dictionary = SparseTermDictionary(
+ resolve_term=self._resolve_sparse_term,
+ resolve_index=self._resolve_sparse_index,
+ persist=self._persist_sparse_term,
+ )
+ return self._sparse_dictionary
+
+ def encode_sparse_vector(self, vector: dict[str, float]) -> dict[str, list[Any]]:
+ return self._get_sparse_dictionary().encode(vector) or {"indices": [], "values": []}
+
+ def _resolve_sparse_term(self, term: str) -> int | None:
+ points = self._scroll(
+ self._metadata_collection_name,
+ filter={"must": [{"key": "term", "match": {"value": term}}]},
+ limit=1,
+ )
+ if not points:
+ return None
+ value = points[0].get("payload", {}).get("index")
+ if value is None:
+ return None
+ index = int(value)
+ owner = self._resolve_sparse_index(index)
+ if owner is not None and owner != term:
+ raise ValueError(
+ "sparse term index collision: "
+ f"index={index} existing_term={owner!r} new_term={term!r}"
+ )
+ return index
+
+ def _resolve_sparse_index(self, index: int) -> str | None:
+ points = self._scroll(
+ self._metadata_collection_name,
+ filter={"must": [{"key": "index", "match": {"value": int(index)}}]},
+ limit=2,
+ )
+ if not points:
+ return None
+ terms = {
+ str(value)
+ for point in points
+ if (value := point.get("payload", {}).get("term")) is not None
+ }
+ if len(terms) > 1:
+ raise ValueError(
+ "sparse term index collision: "
+ f"index={index} existing_terms={sorted(terms)!r}"
+ )
+ return next(iter(terms), None)
+
+ def _persist_sparse_term(self, term: str, index: int) -> None:
+ self._upsert_points(
+ self._metadata_collection_name,
+ [
+ {
+ "id": to_qdrant_point_id(f"openviking:sparse:{term}"),
+ "vector": {_META_VECTOR_NAME: [0.0]},
+ "payload": {
+ "_openviking_sparse_term": True,
+ "term": term,
+ "index": int(index),
+ },
+ }
+ ],
+ )
diff --git a/openviking/storage/vectordb/collection/qdrant_rest.py b/openviking/storage/vectordb/collection/qdrant_rest.py
new file mode 100644
index 0000000000..d0febf81ac
--- /dev/null
+++ b/openviking/storage/vectordb/collection/qdrant_rest.py
@@ -0,0 +1,117 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: Apache-2.0
+"""Small dependency-free Qdrant REST client."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Callable
+from typing import Any
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+
+class QdrantError(RuntimeError):
+ """A bounded Qdrant REST failure."""
+
+ def __init__(
+ self,
+ message: str,
+ *,
+ method: str | None = None,
+ path: str | None = None,
+ status: int | None = None,
+ ) -> None:
+ super().__init__(message)
+ self.method = method
+ self.path = path
+ self.status = status
+
+
+class QdrantRestClient:
+ """Minimal JSON REST transport with injectable opener for tests."""
+
+ def __init__(
+ self,
+ base_url: str,
+ *,
+ api_key: str | None = None,
+ timeout_seconds: float = 10.0,
+ opener: Callable[..., Any] | None = None,
+ ) -> None:
+ normalized = str(base_url).strip().rstrip("/")
+ if not normalized:
+ raise ValueError("Qdrant URL must not be empty")
+ self._base_url = normalized
+ self._api_key = api_key
+ self._timeout_seconds = float(timeout_seconds)
+ self._opener = opener or urlopen
+
+ @property
+ def base_url(self) -> str:
+ return self._base_url
+
+ def request(
+ self,
+ method: str,
+ path: str,
+ body: dict[str, Any] | None = None,
+ *,
+ params: dict[str, Any] | None = None,
+ ) -> dict[str, Any]:
+ if not path.startswith("/"):
+ path = f"/{path}"
+ query = f"?{urlencode(params)}" if params else ""
+ request = Request(
+ f"{self._base_url}{path}{query}",
+ data=json.dumps(body).encode("utf-8") if body is not None else None,
+ headers={
+ "Accept": "application/json",
+ "Content-Type": "application/json",
+ **({"api-key": self._api_key} if self._api_key else {}),
+ },
+ method=method.upper(),
+ )
+ try:
+ with self._opener(request, timeout=self._timeout_seconds) as response:
+ raw = response.read()
+ except HTTPError as exc:
+ raw = exc.read()
+ detail = raw.decode("utf-8", errors="replace")[:1000]
+ raise QdrantError(
+ f"Qdrant HTTP {exc.code} {method.upper()} {path}: {detail}",
+ method=method.upper(),
+ path=path,
+ status=exc.code,
+ ) from exc
+ except URLError as exc:
+ raise QdrantError(
+ f"Qdrant transport error {method.upper()} {path}: {exc.reason}",
+ method=method.upper(),
+ path=path,
+ ) from exc
+ except TimeoutError as exc:
+ raise QdrantError(
+ f"Qdrant request timed out {method.upper()} {path}",
+ method=method.upper(),
+ path=path,
+ ) from exc
+
+ if not raw:
+ return {}
+ try:
+ decoded = json.loads(raw.decode("utf-8"))
+ except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+ raise QdrantError(
+ f"Qdrant returned invalid JSON for {method.upper()} {path}",
+ method=method.upper(),
+ path=path,
+ ) from exc
+ if not isinstance(decoded, dict):
+ raise QdrantError(
+ f"Qdrant returned a non-object response for {method.upper()} {path}",
+ method=method.upper(),
+ path=path,
+ )
+ return decoded
diff --git a/openviking/storage/vectordb/qdrant_sparse.py b/openviking/storage/vectordb/qdrant_sparse.py
new file mode 100644
index 0000000000..504d9cc785
--- /dev/null
+++ b/openviking/storage/vectordb/qdrant_sparse.py
@@ -0,0 +1,82 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: Apache-2.0
+"""Pure sparse-term mapping primitives used by the Qdrant backend."""
+
+from __future__ import annotations
+
+import hashlib
+import math
+from collections.abc import Callable
+from threading import RLock
+from typing import Any
+
+SparseResolveTerm = Callable[[str], int | None]
+SparseResolveIndex = Callable[[int], str | None]
+SparsePersist = Callable[[str, int], Any]
+_MAX_QDRANT_SPARSE_INDEX = 0x7FFF_FFFF
+_SPARSE_TERM_LOCK = RLock()
+
+
+def stable_sparse_index(term: str) -> int:
+ """Return a positive Qdrant-compatible sparse index derived from a term."""
+ digest = hashlib.sha256(term.encode("utf-8")).digest()
+ # Qdrant's sparse indices are uint32 on the REST/protobuf boundary.
+ value = int.from_bytes(digest[:4], "big") & _MAX_QDRANT_SPARSE_INDEX
+ return value or 1
+
+
+class SparseTermDictionary:
+ """Resolve string sparse terms to durable numeric Qdrant indices."""
+
+ def __init__(
+ self,
+ *,
+ resolve_term: SparseResolveTerm,
+ resolve_index: SparseResolveIndex,
+ persist: SparsePersist,
+ hash_term: Callable[[str], int] = stable_sparse_index,
+ ) -> None:
+ self._resolve_term = resolve_term
+ self._resolve_index = resolve_index
+ self._persist = persist
+ self._hash_term = hash_term
+
+ def index_for(self, term: str) -> int:
+ with _SPARSE_TERM_LOCK:
+ normalized = str(term)
+ existing = self._resolve_term(normalized)
+ if existing is not None:
+ return int(existing)
+
+ candidate = int(self._hash_term(normalized))
+ if not 0 < candidate <= _MAX_QDRANT_SPARSE_INDEX:
+ raise ValueError(
+ "sparse term index must be a positive Qdrant-compatible uint32"
+ )
+ owner = self._resolve_index(candidate)
+ if owner is not None and owner != normalized:
+ raise ValueError(
+ "sparse term index collision: "
+ f"index={candidate} existing_term={owner!r} new_term={normalized!r}"
+ )
+ self._persist(normalized, candidate)
+ owner = self._resolve_index(candidate)
+ if owner is not None and owner != normalized:
+ raise ValueError(
+ "sparse term index collision after persistence: "
+ f"index={candidate} existing_term={owner!r} new_term={normalized!r}"
+ )
+ return candidate
+
+ def encode(self, vector: dict[str, float] | None) -> dict[str, list[Any]] | None:
+ if not vector:
+ return None
+ indices: list[int] = []
+ values: list[float] = []
+ for term, weight in vector.items():
+ value = float(weight)
+ if not math.isfinite(value):
+ raise ValueError("sparse vector weights must be finite")
+ indices.append(self.index_for(str(term)))
+ values.append(value)
+ return {"indices": indices, "values": values}
diff --git a/openviking/storage/vectordb/qdrant_utils.py b/openviking/storage/vectordb/qdrant_utils.py
new file mode 100644
index 0000000000..69167359d1
--- /dev/null
+++ b/openviking/storage/vectordb/qdrant_utils.py
@@ -0,0 +1,272 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: Apache-2.0
+"""Pure Qdrant/OpenViking conversion helpers."""
+
+from __future__ import annotations
+
+import uuid
+from datetime import date, datetime
+from typing import Any
+
+from openviking.storage.expr import (
+ And,
+ Contains,
+ Eq,
+ FilterExpr,
+ In,
+ Or,
+ PathScope,
+ Range,
+ RawDSL,
+ TimeRange,
+)
+
+_OPENVIKING_QDRANT_ID_NAMESPACE = uuid.UUID("4b6bb5a8-7f1f-5b1a-9d4c-b93f29b1d67c")
+_URI_FIELDS = {"uri", "parent_uri"}
+
+
+def _normalize_path(value: Any) -> Any:
+ if not isinstance(value, str):
+ return value
+ stripped = value.strip()
+ if stripped.startswith("viking://"):
+ stripped = stripped[len("viking://") :]
+ if not stripped:
+ return "/"
+ normalized = "/" + stripped.lstrip("/")
+ return normalized.rstrip("/") or "/"
+
+
+def _path_depth(path: str) -> int:
+ return len([part for part in path.split("/") if part])
+
+
+def _scope_roots(path: str) -> list[str]:
+ parts = [part for part in path.split("/") if part]
+ roots = ["/"]
+ for index in range(1, len(parts) + 1):
+ roots.append("/" + "/".join(parts[:index]))
+ return roots
+
+
+def build_qdrant_payload(record: dict[str, Any]) -> dict[str, Any]:
+ """Normalize an OpenViking record into Qdrant payload fields."""
+ payload = dict(record)
+ original_id = payload.pop("id", None)
+ payload.pop("vector", None)
+ payload.pop("sparse_vector", None)
+ if original_id is not None:
+ payload["_openviking_original_id"] = str(original_id)
+
+ for field in _URI_FIELDS:
+ value = payload.get(field)
+ if isinstance(value, str):
+ payload[field] = _normalize_path(value)
+
+ uri = payload.get("uri")
+ if isinstance(uri, str):
+ payload["uri_depth"] = _path_depth(uri)
+ payload["scope_roots"] = _scope_roots(uri)
+ return payload
+
+
+def to_qdrant_point_id(value: Any) -> str:
+ """Return a stable UUID point ID for an arbitrary OpenViking ID."""
+ return str(uuid.uuid5(_OPENVIKING_QDRANT_ID_NAMESPACE, str(value)))
+
+
+def _value(value: Any) -> Any:
+ if isinstance(value, (datetime, date)):
+ return value.isoformat()
+ return value
+
+
+def _match(field: str, value: Any) -> dict[str, Any]:
+ return {"key": field, "match": {"value": _value(value)}}
+
+
+def _match_any(field: str, values: list[Any]) -> dict[str, Any]:
+ return {"key": field, "match": {"any": [_value(value) for value in values]}}
+
+
+def _legacy_depth(payload: dict[str, Any]) -> int | None:
+ marker = payload.get("para")
+ if not isinstance(marker, str) or not marker.startswith("-d="):
+ return None
+ try:
+ return int(marker[3:])
+ except ValueError:
+ raise ValueError(f"Invalid legacy path depth: {marker!r}") from None
+
+
+def _compile_legacy(payload: dict[str, Any]) -> dict[str, Any]:
+ op = str(payload.get("op") or "").lower()
+ if op == "and":
+ return _compile(And([item for item in payload.get("conds", []) if item]))
+ if op == "or":
+ return _compile(Or([item for item in payload.get("conds", []) if item]))
+ if op in {"must", "must_not"}:
+ field = payload.get("field")
+ values = list(payload.get("conds") or [])
+ if not isinstance(field, str) or not values:
+ return {}
+ values = [_normalize_path(value) if field in _URI_FIELDS else _value(value) for value in values]
+ depth = _legacy_depth(payload)
+ if depth is not None and field in _URI_FIELDS:
+ return _compile(PathScope(field, values[0], depth))
+ condition = _match(field, values[0]) if len(values) == 1 else _match_any(field, values)
+ return {op: [condition]}
+ if op in {"range", "time_range"}:
+ field = payload.get("field")
+ bounds = {
+ key: _value(payload[key])
+ for key in ("gte", "gt", "lte", "lt")
+ if payload.get(key) is not None
+ }
+ return {"must": [{"key": field, "range": bounds}]}
+ if op == "range_out":
+ field = payload.get("field")
+ branches = []
+ if payload.get("gte") is not None:
+ branches.append({"must": [{"key": field, "range": {"lt": _value(payload["gte"])}}]})
+ if payload.get("lte") is not None:
+ branches.append({"must": [{"key": field, "range": {"gt": _value(payload["lte"])}}]})
+ return _compile(Or(branches))
+ if op == "prefix":
+ field = payload.get("field")
+ prefix = payload.get("prefix", "")
+ if field in _URI_FIELDS:
+ return _compile(PathScope(field, prefix, depth=-1))
+ raise NotImplementedError("Qdrant adapter only supports prefix filters for URI fields")
+ if op == "contains":
+ raise NotImplementedError("Contains is not supported by the Qdrant adapter")
+ if op:
+ raise NotImplementedError(f"Unsupported legacy Qdrant filter operation: {op}")
+ return payload
+
+
+def _compile(expr: FilterExpr | dict[str, Any]) -> dict[str, Any]:
+ if isinstance(expr, dict):
+ if "op" in expr:
+ return _compile_legacy(expr)
+ return expr
+ if isinstance(expr, RawDSL):
+ return _compile(expr.payload)
+ if isinstance(expr, And):
+ clauses = [_compile(item) for item in expr.conds if item is not None]
+ clauses = [item for item in clauses if item]
+ if not clauses:
+ return {}
+ if len(clauses) == 1:
+ return clauses[0]
+ simple_keys = {"must", "must_not"}
+ complex_count = 0
+ for item in clauses:
+ keys = [key for key, value in item.items() if value]
+ if not (
+ len(keys) == 1
+ and keys[0] in simple_keys
+ and isinstance(item[keys[0]], list)
+ ):
+ complex_count += 1
+
+ result: dict[str, Any] = {}
+ for item in clauses:
+ keys = [key for key, value in item.items() if value]
+ if (
+ len(keys) == 1
+ and keys[0] in simple_keys
+ and isinstance(item[keys[0]], list)
+ ):
+ result.setdefault(keys[0], []).extend(item[keys[0]])
+ continue
+ if complex_count == 1:
+ for key, value in item.items():
+ if not value:
+ continue
+ if key in simple_keys and isinstance(value, list):
+ result.setdefault(key, []).extend(value)
+ else:
+ result[key] = value
+ else:
+ result.setdefault("must", []).append(item)
+ return result
+ if isinstance(expr, Or):
+ clauses = [_compile(item) for item in expr.conds if item is not None]
+ clauses = [item for item in clauses if item]
+ if not clauses:
+ return {}
+ if len(clauses) == 1:
+ return clauses[0]
+ conditions: list[Any] = []
+ for item in clauses:
+ keys = [key for key, value in item.items() if value]
+ if (
+ len(keys) == 1
+ and keys[0] in {"must", "should"}
+ and isinstance(item[keys[0]], list)
+ and len(item[keys[0]]) == 1
+ ):
+ conditions.append(item[keys[0]][0])
+ else:
+ conditions.append(item)
+ return {"should": conditions}
+ if isinstance(expr, Eq):
+ field = expr.field
+ value = _normalize_path(expr.value) if field in _URI_FIELDS else expr.value
+ return {"must": [_match(field, value)]}
+ if isinstance(expr, In):
+ field = expr.field
+ values = [
+ _normalize_path(value) if field in _URI_FIELDS else value for value in expr.values
+ ]
+ if len(values) == 1:
+ return {"must": [_match(field, values[0])]}
+ return {"must": [_match_any(field, values)]}
+ if isinstance(expr, (Range, TimeRange)):
+ payload: dict[str, Any] = {}
+ if isinstance(expr, Range):
+ for key in ("gte", "gt", "lte", "lt"):
+ value = getattr(expr, key)
+ if value is not None:
+ payload[key] = _value(value)
+ else:
+ if expr.start is not None:
+ payload["gte"] = _value(expr.start)
+ if expr.end is not None:
+ payload["lt"] = _value(expr.end)
+ return {"must": [{"key": expr.field, "range": payload}]}
+ if isinstance(expr, PathScope):
+ if expr.field in _URI_FIELDS and not isinstance(expr.path, str):
+ raise ValueError("Qdrant URI path scope requires a string URI path")
+ if expr.field != "uri":
+ raise NotImplementedError(
+ "Qdrant PathScope supports uri only; the field has no scope payload"
+ )
+ path = _normalize_path(expr.path) if expr.field in _URI_FIELDS else expr.path
+ if expr.depth == 0:
+ return {"must": [_match(expr.field, path)]}
+ scope_match = _match("scope_roots", path)
+ if expr.depth < 0:
+ return {"must": [scope_match]}
+ return {
+ "must": [
+ scope_match,
+ {
+ "key": "uri_depth",
+ "range": {"lte": _path_depth(path) + expr.depth},
+ },
+ ]
+ }
+ if isinstance(expr, Contains):
+ raise NotImplementedError("Contains is not supported by the Qdrant adapter")
+ raise TypeError(f"Unsupported filter expression: {type(expr)!r}")
+
+
+def compile_qdrant_filter(
+ expr: FilterExpr | dict[str, Any] | None,
+) -> dict[str, Any]:
+ """Compile OpenViking's filter AST into Qdrant's filter JSON."""
+ if expr is None:
+ return {}
+ return _compile(expr)
diff --git a/openviking/storage/vectordb_adapters/README.md b/openviking/storage/vectordb_adapters/README.md
index 40d6acbbd3..9eac1afbf4 100644
--- a/openviking/storage/vectordb_adapters/README.md
+++ b/openviking/storage/vectordb_adapters/README.md
@@ -244,4 +244,49 @@ class ThirdPartyCollectionAdapter(CollectionAdapter):
- `backend=thirdparty` 可正常初始化。
- create 后可完成 upsert/get/query/delete/count 全流程。
- 不改上层业务调用方式即可参与 `find/search` 检索链路。
-- 后端差异全部封装在 adapter 层。
\ No newline at end of file
+- 后端差异全部封装在 adapter 层。
+
+---
+
+## 10. Qdrant REST backend
+
+内置 `backend: qdrant` 使用 Python 标准库通过 Qdrant REST API 访问远端
+collection，不新增 `qdrant-client` 依赖：
+
+```json
+{
+ "storage": {
+ "vectordb": {
+ "backend": "qdrant",
+ "url": "http://127.0.0.1:6333",
+ "project": "default",
+ "name": "context",
+ "dimension": 1536,
+ "sparse_weight": 0.5,
+ "qdrant": {
+ "api_key": "optional-key",
+ "dense_vector_name": "vector",
+ "sparse_vector_name": "sparse_vector",
+ "timeout_seconds": 10
+ }
+ }
+ }
+}
+```
+
+- `sparse_weight: 0` 是 dense-only；`0 < sparse_weight <= 1` 启用 named sparse
+ vector 与 hybrid weighted-RRF。
+- Qdrant data collection 会附带一个 OpenViking metadata sidecar，保存 schema、
+ index metadata 与 sparse term dictionary；term ID 使用 Qdrant 兼容的
+ positive uint32 SHA-256 candidate，并在碰撞时 fail closed。没有 marker
+ 的既有 collection 会 fail closed，不会被隐式接管。
+- `uri` 会保存标准化路径、深度与 ancestor scope roots；上层 account filter
+ 与多 `search_tags` AND 语义会保留。
+- `Contains` 与 Qdrant content grep 不支持；`USE_CONTENT_FIELD=False`，grep
+ 继续走 filesystem fallback。
+- 可用 `QDRANT_URL`（以及可选的 `QDRANT_API_KEY`）运行 live coverage：
+
+```bash
+QDRANT_URL=http://127.0.0.1:6333 \
+ pytest --confcutdir=tests/storage -q tests/storage/test_qdrant_integration.py
+```
diff --git a/openviking/storage/vectordb_adapters/__init__.py b/openviking/storage/vectordb_adapters/__init__.py
index 1d19e1bac3..cea491a6cc 100644
--- a/openviking/storage/vectordb_adapters/__init__.py
+++ b/openviking/storage/vectordb_adapters/__init__.py
@@ -6,6 +6,7 @@
from .factory import create_collection_adapter
from .http_adapter import HttpCollectionAdapter
from .local_adapter import CuVSCollectionAdapter, LocalCollectionAdapter
+from .qdrant_adapter import QdrantCollectionAdapter
from .vikingdb_private_adapter import VikingDBPrivateCollectionAdapter
from .volcengine_adapter import VolcengineCollectionAdapter
@@ -16,5 +17,6 @@
"HttpCollectionAdapter",
"VolcengineCollectionAdapter",
"VikingDBPrivateCollectionAdapter",
+ "QdrantCollectionAdapter",
"create_collection_adapter",
]
diff --git a/openviking/storage/vectordb_adapters/factory.py b/openviking/storage/vectordb_adapters/factory.py
index bc7ec0e124..1c0b91c224 100644
--- a/openviking/storage/vectordb_adapters/factory.py
+++ b/openviking/storage/vectordb_adapters/factory.py
@@ -9,6 +9,7 @@
from .base import CollectionAdapter
from .http_adapter import HttpCollectionAdapter
from .local_adapter import CuVSCollectionAdapter, LocalCollectionAdapter
+from .qdrant_adapter import QdrantCollectionAdapter
from .vikingdb_private_adapter import VikingDBPrivateCollectionAdapter
from .volcengine_adapter import VolcengineCollectionAdapter
@@ -16,6 +17,7 @@
"local": LocalCollectionAdapter,
"cuvs": CuVSCollectionAdapter,
"http": HttpCollectionAdapter,
+ "qdrant": QdrantCollectionAdapter,
"volcengine": VolcengineCollectionAdapter,
"vikingdb": VikingDBPrivateCollectionAdapter,
}
diff --git a/openviking/storage/vectordb_adapters/qdrant_adapter.py b/openviking/storage/vectordb_adapters/qdrant_adapter.py
new file mode 100644
index 0000000000..be811b9111
--- /dev/null
+++ b/openviking/storage/vectordb_adapters/qdrant_adapter.py
@@ -0,0 +1,173 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: Apache-2.0
+"""Qdrant backend adapter for OpenViking."""
+
+from __future__ import annotations
+
+import math
+from typing import Any
+
+from openviking.storage.vectordb.collection.collection import Collection
+from openviking.storage.vectordb.collection.qdrant_collection import QdrantCollection
+from openviking.storage.vectordb.collection.qdrant_rest import QdrantRestClient
+from openviking.storage.vectordb.qdrant_utils import compile_qdrant_filter
+
+from .base import CollectionAdapter
+
+
+class QdrantCollectionAdapter(CollectionAdapter):
+ """Adapter that maps OpenViking collection operations to Qdrant REST."""
+
+ _DATA_BATCH_SIZE = 100
+ USE_CONTENT_FIELD = False
+
+ def __init__(
+ self,
+ *,
+ url: str,
+ api_key: str | None,
+ timeout_seconds: float,
+ project_name: str,
+ collection_name: str,
+ index_name: str,
+ distance_metric: str,
+ dimension: int,
+ sparse_weight: float,
+ dense_vector_name: str,
+ sparse_vector_name: str,
+ metadata_collection_name: str | None,
+ ) -> None:
+ super().__init__(collection_name=collection_name, index_name=index_name)
+ self.mode = "qdrant"
+ self._client = QdrantRestClient(
+ url,
+ api_key=api_key,
+ timeout_seconds=timeout_seconds,
+ )
+ self._project_name = project_name or "default"
+ self._physical_collection_name = f"{self._project_name}__{collection_name}"
+ self._metadata_collection_name_override = metadata_collection_name
+ self._metadata_collection_name = metadata_collection_name or (
+ f"{self._physical_collection_name}__openviking_meta"
+ )
+ self._distance_metric = distance_metric
+ self._dimension = int(dimension)
+ self._sparse_weight = float(sparse_weight)
+ if not math.isfinite(self._sparse_weight) or not 0.0 <= self._sparse_weight <= 1.0:
+ raise ValueError("Qdrant sparse_weight must be a finite number between 0 and 1")
+ self._dense_vector_name = dense_vector_name
+ self._sparse_vector_name = sparse_vector_name
+
+ @classmethod
+ def from_config(cls, config: Any) -> "QdrantCollectionAdapter":
+ qdrant_cfg = getattr(config, "qdrant", None)
+ custom = dict(getattr(config, "custom_params", {}) or {})
+ url = (
+ getattr(qdrant_cfg, "url", None)
+ or getattr(config, "url", None)
+ or custom.get("url")
+ )
+ if not url:
+ raise ValueError("Qdrant backend requires qdrant.url or vectordb.url")
+ return cls(
+ url=str(url).strip().rstrip("/"),
+ api_key=getattr(qdrant_cfg, "api_key", None) or custom.get("api_key"),
+ timeout_seconds=float(
+ getattr(qdrant_cfg, "timeout_seconds", None)
+ or custom.get("timeout_seconds")
+ or 10.0
+ ),
+ project_name=str(config.project_name or "default"),
+ collection_name=str(config.name or "context"),
+ index_name=str(config.index_name or "default"),
+ distance_metric=str(config.distance_metric or "cosine"),
+ dimension=int(config.dimension or 0),
+ sparse_weight=float(config.sparse_weight or 0.0),
+ dense_vector_name=str(
+ getattr(qdrant_cfg, "dense_vector_name", None)
+ or custom.get("dense_vector_name")
+ or "vector"
+ ),
+ sparse_vector_name=str(
+ getattr(qdrant_cfg, "sparse_vector_name", None)
+ or custom.get("sparse_vector_name")
+ or "sparse_vector"
+ ),
+ metadata_collection_name=getattr(qdrant_cfg, "metadata_collection_name", None)
+ or custom.get("metadata_collection_name"),
+ )
+
+ def _new_collection(self) -> QdrantCollection:
+ self._physical_collection_name = f"{self._project_name}__{self._collection_name}"
+ self._metadata_collection_name = self._metadata_collection_name_override or (
+ f"{self._physical_collection_name}__openviking_meta"
+ )
+ return QdrantCollection(
+ client=self._client,
+ collection_name=self._physical_collection_name,
+ metadata_collection_name=self._metadata_collection_name,
+ dense_vector_name=self._dense_vector_name,
+ sparse_vector_name=self._sparse_vector_name,
+ vector_dim=self._dimension,
+ distance=self._distance_metric,
+ sparse_enabled=self._sparse_weight > 0.0,
+ sparse_weight=self._sparse_weight,
+ )
+
+ def _load_existing_collection_if_needed(self) -> None:
+ if self._collection is not None:
+ return
+ candidate = self._new_collection()
+ if not candidate.collection_exists():
+ return
+ if not candidate.has_openviking_metadata():
+ candidate.close()
+ raise RuntimeError(
+ "Qdrant collection exists but OpenViking metadata is missing: "
+ f"{self._physical_collection_name}. "
+ "Use a different project/name, restore metadata, or drop the stale collection."
+ )
+ candidate.get_meta_data()
+ self._collection = Collection(candidate)
+
+ def _create_backend_collection(self, meta: dict[str, Any]) -> Collection:
+ candidate = self._new_collection()
+ candidate.create_remote_collection(meta)
+ return Collection(candidate)
+
+ def _sanitize_scalar_index_fields(
+ self,
+ scalar_index_fields: list[str],
+ fields_meta: list[dict[str, Any]],
+ ) -> list[str]:
+ del fields_meta
+ return list(dict.fromkeys([*scalar_index_fields, "uri_depth", "scope_roots"]))
+
+ def _build_default_index_meta(
+ self,
+ *,
+ index_name: str,
+ distance: str,
+ use_sparse: bool,
+ sparse_weight: float,
+ scalar_index_fields: list[str],
+ ) -> dict[str, Any]:
+ return {
+ "IndexName": index_name,
+ "VectorIndex": {
+ "IndexType": "hnsw_hybrid" if use_sparse else "hnsw",
+ "Distance": distance,
+ },
+ "ScalarIndex": scalar_index_fields,
+ "SparseWeight": sparse_weight,
+ }
+
+ def _compile_filter(self, expr: Any) -> dict[str, Any]:
+ return compile_qdrant_filter(expr)
+
+ def update_data(self, data_list: list[dict[str, Any]]) -> list[str]:
+ result = self.get_collection().update_data(data_list)
+ return [str(item) for item in result]
+
+
+__all__ = ["QdrantCollectionAdapter"]
diff --git a/openviking/storage/viking_vector_index_backend.py b/openviking/storage/viking_vector_index_backend.py
index 2c3ae592a0..5a3ab4a2aa 100644
--- a/openviking/storage/viking_vector_index_backend.py
+++ b/openviking/storage/viking_vector_index_backend.py
@@ -133,12 +133,41 @@ async def update_collection_schema(
) -> None:
def _update() -> None:
collection = self._adapter.get_collection()
- collection.update(fields=fields)
+ if self._adapter.mode == "qdrant":
+ current_schema = collection.get_meta_data()
+ schema_scalar_index = list(
+ dict.fromkeys(
+ [*(current_schema.get("ScalarIndex") or []), *scalar_index]
+ )
+ )
+ current_index = collection.get_index_meta_data(index_name)
+ index_scalar_index = list(
+ dict.fromkeys(
+ [*(current_index.get("ScalarIndex") or []), *scalar_index]
+ )
+ )
+ collection.update(
+ fields={"Fields": fields, "ScalarIndex": schema_scalar_index}
+ )
+ else:
+ collection.update(fields=fields)
if self._adapter.mode in {"local", "cuvs"}:
index_meta = collection.get_index_meta_data(index_name)
index_meta["ScalarIndex"] = scalar_index
collection.drop_index(index_name)
collection.create_index(index_name, index_meta)
+ elif self._adapter.mode == "qdrant":
+ if collection.has_index(index_name):
+ collection.update_index(index_name, scalar_index=index_scalar_index)
+ else:
+ index_meta = self._adapter._build_default_index_meta(
+ index_name=index_name,
+ distance=self._adapter._distance_metric,
+ use_sparse=self._adapter._sparse_weight > 0.0,
+ sparse_weight=self._adapter._sparse_weight,
+ scalar_index_fields=schema_scalar_index,
+ )
+ collection.create_index(index_name, index_meta)
else:
collection.update_index(index_name, scalar_index=scalar_index)
diff --git a/openviking_cli/utils/config/vectordb_config.py b/openviking_cli/utils/config/vectordb_config.py
index bccc25cded..ee41d9c5f0 100644
--- a/openviking_cli/utils/config/vectordb_config.py
+++ b/openviking_cli/utils/config/vectordb_config.py
@@ -50,6 +50,22 @@ class VikingDBConfig(BaseModel):
model_config = {"extra": "forbid"}
+class QdrantConfig(BaseModel):
+ """Configuration for the Qdrant REST backend."""
+
+ url: Optional[str] = Field(default=None, description="Qdrant REST endpoint")
+ api_key: Optional[str] = Field(default=None, description="Optional Qdrant API key")
+ timeout_seconds: float = Field(default=10.0, gt=0)
+ dense_vector_name: str = Field(default="vector", min_length=1)
+ sparse_vector_name: str = Field(default="sparse_vector", min_length=1)
+ metadata_collection_name: Optional[str] = Field(
+ default=None,
+ description="Optional explicit OpenViking metadata collection name",
+ )
+
+ model_config = {"extra": "forbid"}
+
+
class CuVSConfig(BaseModel):
"""Configuration for GPU dense-vector search through NVIDIA cuVS."""
@@ -201,7 +217,7 @@ class VectorDBBackendConfig(BaseModel):
description=(
"VectorDB backend type: 'local', 'cuvs', 'http', "
"'volcengine' (AK/SK signed or API key data-plane only), "
- "or 'vikingdb' (private deployment)"
+ "'vikingdb' (private deployment), or 'qdrant' (REST)"
),
)
@@ -214,7 +230,10 @@ class VectorDBBackendConfig(BaseModel):
url: Optional[str] = Field(
default=None,
- description="Remote service URL for 'http' type (e.g., 'http://localhost:5000')",
+ description=(
+ "Remote service URL for 'http' or 'qdrant' backends "
+ "(e.g., 'http://localhost:5000')"
+ ),
)
project_name: Optional[str] = Field(
@@ -255,6 +274,11 @@ class VectorDBBackendConfig(BaseModel):
description="VikingDB private deployment configuration for 'vikingdb' type",
)
+ qdrant: Optional[QdrantConfig] = Field(
+ default_factory=QdrantConfig,
+ description="Qdrant REST configuration for the 'qdrant' backend",
+ )
+
cuvs: Optional[CuVSConfig] = Field(
default_factory=CuVSConfig,
description="NVIDIA cuVS dense-vector search configuration for the 'cuvs' backend",
@@ -276,6 +300,7 @@ def validate_config(self):
"http",
"volcengine",
"vikingdb",
+ "qdrant",
]
# Allow custom backend classes (containing dot) without standard validation
@@ -325,4 +350,8 @@ def validate_config(self):
if not self.vikingdb or not self.vikingdb.host:
raise ValueError("VectorDB vikingdb backend requires 'host' to be set")
+ elif self.backend == "qdrant":
+ if not (self.qdrant and (self.qdrant.url or self.url)):
+ raise ValueError("VectorDB qdrant backend requires qdrant.url or url to be set")
+
return self
diff --git a/tests/README.md b/tests/README.md
index 4c8f431e30..8365a0cdc2 100644
--- a/tests/README.md
+++ b/tests/README.md
@@ -66,6 +66,17 @@ pytest tests/ -k "initialize" -v
pytest tests/client/test_http_client_config.py -v -s
```
+### Qdrant integration
+
+The Qdrant REST backend has environment-gated live coverage. It is skipped
+when `QDRANT_URL` is unset; set `QDRANT_API_KEY` when the server requires
+authentication:
+
+```bash
+QDRANT_URL=http://127.0.0.1:6333 \
+ pytest --confcutdir=tests/storage -q tests/storage/test_qdrant_integration.py
+```
+
### Common Test Scenarios
```bash
diff --git a/tests/misc/test_config_validation.py b/tests/misc/test_config_validation.py
index cd685f80f9..02dc2a1a78 100644
--- a/tests/misc/test_config_validation.py
+++ b/tests/misc/test_config_validation.py
@@ -878,7 +878,7 @@ def test_removed_volcengine_api_key_backend_name_is_rejected():
assert "volcengine_api_key" in str(e)
-@pytest.mark.parametrize("backend", ["qdrant", "opengauss"])
+@pytest.mark.parametrize("backend", ["opengauss"])
def test_removed_third_party_vectordb_backends_are_rejected(backend):
with pytest.raises(ValueError) as exc_info:
VectorDBBackendConfig(backend=backend)
@@ -890,6 +890,17 @@ def test_removed_third_party_vectordb_backends_are_rejected(backend):
assert "vikingdb" in message
+def test_qdrant_vectordb_backend_is_accepted():
+ config = VectorDBBackendConfig(
+ backend="qdrant",
+ qdrant={"url": "http://127.0.0.1:6333"},
+ )
+
+ assert config.backend == "qdrant"
+ assert config.qdrant is not None
+ assert config.qdrant.url == "http://127.0.0.1:6333"
+
+
def test_vectordb_volcengine_api_key_auth_requires_host_or_region():
try:
VectorDBBackendConfig(
diff --git a/tests/storage/test_collection_schemas.py b/tests/storage/test_collection_schemas.py
index 4e6355d099..f1477e7682 100644
--- a/tests/storage/test_collection_schemas.py
+++ b/tests/storage/test_collection_schemas.py
@@ -239,6 +239,99 @@ async def update_collection_schema(self, fields, scalar_index):
assert ACL_CONTEXT_FIELDS <= set(scalar_index)
+@pytest.mark.asyncio
+async def test_init_context_collection_migrates_qdrant_legacy_schema(monkeypatch):
+ schema_updates = []
+
+ class _FakeStorage:
+ async def create_collection(self, name, schema):
+ del name, schema
+ return False
+
+ async def get_collection_meta(self):
+ schema = CollectionSchemas.context_collection("context", 2)
+ return {
+ "Description": "Unified context collection",
+ "Fields": [
+ field
+ for field in schema["Fields"]
+ if field["FieldName"] not in ACL_CONTEXT_FIELDS
+ ],
+ "ScalarIndex": [
+ field
+ for field in schema["ScalarIndex"]
+ if field not in ACL_CONTEXT_FIELDS
+ ],
+ }
+
+ async def count(self):
+ return 0
+
+ async def update_collection_description(self, description):
+ del description
+
+ async def update_collection_schema(self, fields, scalar_index):
+ schema_updates.append((fields, scalar_index))
+
+ config = _DummyConfig(_DummyEmbedder(), backend="qdrant")
+ monkeypatch.setattr(
+ "openviking_cli.utils.config.get_openviking_config",
+ lambda: config,
+ )
+
+ created = await init_context_collection(_FakeStorage())
+
+ assert created is False
+ assert len(schema_updates) == 1
+ fields, scalar_index = schema_updates[0]
+ fields_by_name = {field["FieldName"]: field for field in fields}
+ assert fields_by_name["acl_enabled"]["FieldType"] == "bool"
+ assert all(fields_by_name[field]["FieldType"] == "list" for field in ACL_GRANT_FIELDS)
+ assert ACL_CONTEXT_FIELDS <= set(scalar_index)
+
+
+@pytest.mark.asyncio
+async def test_init_context_collection_rechecks_qdrant_schema_when_acl_metadata_is_complete(
+ monkeypatch,
+):
+ schema_updates = []
+ attempts = 0
+
+ class _FakeStorage:
+ async def create_collection(self, name, schema):
+ del name, schema
+ return False
+
+ async def get_collection_meta(self):
+ return CollectionSchemas.context_collection("context", 2)
+
+ async def count(self):
+ return 0
+
+ async def update_collection_description(self, description):
+ del description
+
+ async def update_collection_schema(self, fields, scalar_index):
+ nonlocal attempts
+ attempts += 1
+ schema_updates.append((fields, scalar_index))
+ if attempts == 1:
+ raise RuntimeError("transient index failure")
+
+ config = _DummyConfig(_DummyEmbedder(), backend="qdrant")
+ monkeypatch.setattr(
+ "openviking_cli.utils.config.get_openviking_config",
+ lambda: config,
+ )
+
+ with pytest.raises(RuntimeError, match="transient index failure"):
+ await init_context_collection(_FakeStorage())
+ created = await init_context_collection(_FakeStorage())
+
+ assert created is False
+ assert len(schema_updates) == 2
+
+
@pytest.mark.asyncio
async def test_init_context_collection_rejects_mismatched_nonempty_collection(monkeypatch):
"""When embedding dimension mismatches for a non-empty collection, vectors are
diff --git a/tests/storage/test_qdrant_adapter.py b/tests/storage/test_qdrant_adapter.py
new file mode 100644
index 0000000000..50516ec7a7
--- /dev/null
+++ b/tests/storage/test_qdrant_adapter.py
@@ -0,0 +1,1497 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import io
+import json
+import math
+from urllib.error import HTTPError
+from urllib.parse import parse_qs, urlsplit
+
+import pytest
+
+from openviking.storage.expr import And, Contains, Eq, In, Or, PathScope, RawDSL
+from openviking.storage.vectordb.collection.qdrant_collection import QdrantCollection
+from openviking.storage.vectordb.collection.qdrant_rest import QdrantError, QdrantRestClient
+from openviking.storage.vectordb.qdrant_sparse import SparseTermDictionary, stable_sparse_index
+from openviking.storage.vectordb.qdrant_utils import (
+ build_qdrant_payload,
+ compile_qdrant_filter,
+ to_qdrant_point_id,
+)
+from openviking.storage.vectordb_adapters.factory import create_collection_adapter
+from openviking.storage.vectordb_adapters.qdrant_adapter import QdrantCollectionAdapter
+from openviking.storage.viking_vector_index_backend import _AsyncVectorAdapter
+from openviking_cli.utils.config.vectordb_config import VectorDBBackendConfig
+
+
+class _Response:
+ def __init__(self, payload: dict):
+ self._body = json.dumps(payload).encode("utf-8")
+
+ def read(self):
+ return self._body
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *_args):
+ return False
+
+
+class _ScriptedTransport:
+ def __init__(self, *responses):
+ self.responses = list(responses)
+ self.requests = []
+
+ def __call__(self, request, timeout):
+ body = json.loads(request.data.decode("utf-8")) if request.data else None
+ self.requests.append(
+ {
+ "method": request.method,
+ "url": request.full_url,
+ "body": body,
+ "timeout": timeout,
+ "headers": dict(request.header_items()),
+ }
+ )
+ status, payload = self.responses.pop(0)
+ if status >= 400:
+ raise HTTPError(
+ request.full_url,
+ status,
+ "qdrant error",
+ {},
+ io.BytesIO(json.dumps(payload).encode("utf-8")),
+ )
+ return _Response(payload)
+
+
+def test_path_payload_includes_self_and_ancestors() -> None:
+ payload = build_qdrant_payload(
+ {
+ "id": "doc-1",
+ "uri": "viking://resources/wiki/physics/doc.md",
+ "parent_uri": "viking://resources/wiki/physics",
+ "account_id": "acct",
+ }
+ )
+
+ assert payload["uri"] == "/resources/wiki/physics/doc.md"
+ assert payload["uri_depth"] == 4
+ assert payload["scope_roots"] == [
+ "/",
+ "/resources",
+ "/resources/wiki",
+ "/resources/wiki/physics",
+ "/resources/wiki/physics/doc.md",
+ ]
+ assert payload["parent_uri"] == "/resources/wiki/physics"
+
+
+def test_path_scope_depth_mapping_is_segment_aware() -> None:
+ subtree = compile_qdrant_filter(PathScope("uri", "viking://resources", depth=-1))
+ finite = compile_qdrant_filter(PathScope("uri", "viking://resources/wiki", depth=2))
+
+ assert subtree == {
+ "must": [
+ {
+ "key": "scope_roots",
+ "match": {"value": "/resources"},
+ }
+ ]
+ }
+ assert finite == {
+ "must": [
+ {
+ "key": "scope_roots",
+ "match": {"value": "/resources/wiki"},
+ },
+ {
+ "key": "uri_depth",
+ "range": {"lte": 4},
+ },
+ ]
+ }
+
+
+def test_path_scope_rejects_non_string_uri_paths() -> None:
+ with pytest.raises(ValueError, match="URI path"):
+ compile_qdrant_filter(PathScope("uri", 123, depth=-1)) # type: ignore[arg-type]
+
+
+def test_parent_uri_path_scope_is_rejected_without_scope_payload() -> None:
+ with pytest.raises(NotImplementedError, match="scope payload"):
+ compile_qdrant_filter(PathScope("parent_uri", "viking://resources", depth=-1))
+
+
+def test_multi_tag_eq_is_qdrant_must_and_in_is_match_any() -> None:
+ result = compile_qdrant_filter(
+ And(
+ [
+ Eq("search_tags", "team=search"),
+ Eq("search_tags", "env=prod"),
+ ]
+ )
+ )
+
+ assert result == {
+ "must": [
+ {"key": "search_tags", "match": {"value": "team=search"}},
+ {"key": "search_tags", "match": {"value": "env=prod"}},
+ ]
+ }
+ assert compile_qdrant_filter(
+ In("search_tags", ["team=search", "team=infra"])
+ ) == {
+ "must": [
+ {
+ "key": "search_tags",
+ "match": {"any": ["team=search", "team=infra"]},
+ }
+ ]
+ }
+
+
+def test_account_filter_is_preserved() -> None:
+ result = compile_qdrant_filter(
+ And(
+ [
+ Eq("account_id", "acct"),
+ PathScope("uri", "viking://resources", depth=-1),
+ ]
+ )
+ )
+
+ assert result["must"][0] == {
+ "key": "account_id",
+ "match": {"value": "acct"},
+ }
+ assert result["must"][1]["key"] == "scope_roots"
+
+
+def test_composed_raw_filter_preserves_all_boolean_clauses() -> None:
+ result = compile_qdrant_filter(
+ And(
+ [
+ RawDSL(
+ {
+ "must": [{"key": "account_id", "match": {"value": "acct"}}],
+ "should": [{"key": "kind", "match": {"value": "doc"}}],
+ }
+ ),
+ Eq("name", "README.md"),
+ ]
+ )
+ )
+
+ assert result == {
+ "must": [
+ {"key": "account_id", "match": {"value": "acct"}},
+ {"key": "name", "match": {"value": "README.md"}},
+ ],
+ "should": [{"key": "kind", "match": {"value": "doc"}}],
+ }
+
+
+def test_legacy_raw_filter_is_compiled_when_combined_with_account_filter() -> None:
+ result = compile_qdrant_filter(
+ And(
+ [
+ Eq("account_id", "acct"),
+ RawDSL(
+ {
+ "op": "and",
+ "conds": [
+ {
+ "op": "must",
+ "field": "search_tags",
+ "conds": ["team=search"],
+ },
+ {
+ "op": "must",
+ "field": "search_tags",
+ "conds": ["env=prod"],
+ },
+ ],
+ }
+ ),
+ ]
+ )
+ )
+
+ assert result == {
+ "must": [
+ {"key": "account_id", "match": {"value": "acct"}},
+ {"key": "search_tags", "match": {"value": "team=search"}},
+ {"key": "search_tags", "match": {"value": "env=prod"}},
+ ]
+ }
+
+
+def test_or_does_not_flatten_must_not_into_should() -> None:
+ result = compile_qdrant_filter(
+ Or(
+ [
+ RawDSL(
+ {
+ "must_not": [
+ {"key": "kind", "match": {"value": "draft"}},
+ ]
+ }
+ ),
+ Eq("account_id", "acct"),
+ ]
+ )
+ )
+
+ assert result == {
+ "should": [
+ {
+ "must_not": [
+ {"key": "kind", "match": {"value": "draft"}},
+ ]
+ },
+ {"key": "account_id", "match": {"value": "acct"}},
+ ]
+ }
+
+
+def test_point_id_is_deterministic_and_original_id_round_trips() -> None:
+ first = to_qdrant_point_id("viking://resources/doc.md")
+ second = to_qdrant_point_id("viking://resources/doc.md")
+ payload = build_qdrant_payload({"id": "viking://resources/doc.md", "uri": "viking://resources/doc.md"})
+
+ assert first == second
+ assert first != "viking://resources/doc.md"
+ assert payload["_openviking_original_id"] == "viking://resources/doc.md"
+
+
+def test_parent_uri_round_trips_on_read() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+
+ record = collection._payload_to_record(
+ {
+ "id": to_qdrant_point_id("doc-1"),
+ "payload": {
+ "_openviking_original_id": "doc-1",
+ "uri": "/resources/doc.md",
+ "parent_uri": "/resources",
+ },
+ }
+ )
+
+ assert record["parent_uri"] == "/resources"
+
+
+def test_numeric_scalar_field_types_map_to_qdrant_numeric_schemas() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._schema = {
+ "Fields": [
+ {"FieldName": "score", "FieldType": "float32"},
+ {"FieldName": "counts", "FieldType": "list"},
+ ]
+ }
+
+ assert collection._field_schema("score") == "float"
+ assert collection._field_schema("counts") == "integer"
+
+
+@pytest.mark.asyncio
+async def test_update_collection_schema_accepts_openviking_field_list() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._schema = {
+ "CollectionName": "docs",
+ "Fields": [{"FieldName": "legacy", "FieldType": "string"}],
+ }
+ collection._indexes = {"default": {"ScalarIndex": ["account_id"]}}
+ marker_writes: list[dict[str, object]] = []
+ collection._write_metadata_marker = lambda: marker_writes.append( # type: ignore[method-assign]
+ dict(collection._schema)
+ )
+ collection._ensure_remote_indexes = lambda _meta: None # type: ignore[method-assign]
+
+ adapter = type(
+ "_Adapter",
+ (),
+ {"mode": "qdrant", "get_collection": lambda self: collection},
+ )()
+
+ await _AsyncVectorAdapter(adapter).update_collection_schema(
+ [
+ {"FieldName": "acl_enabled", "FieldType": "bool"},
+ ],
+ ["account_id", "acl_enabled"],
+ "default",
+ )
+
+ assert collection.get_meta_data()["Fields"] == [
+ {"FieldName": "legacy", "FieldType": "string"},
+ {"FieldName": "acl_enabled", "FieldType": "bool"},
+ ]
+ assert collection.get_meta_data()["ScalarIndex"] == [
+ "account_id",
+ "acl_enabled",
+ ]
+ assert collection.get_index_meta_data("default")["ScalarIndex"] == [
+ "account_id",
+ "acl_enabled",
+ ]
+ assert marker_writes == [
+ {
+ "CollectionName": "docs",
+ "ScalarIndex": [
+ "account_id",
+ "acl_enabled",
+ ],
+ "Fields": [
+ {"FieldName": "legacy", "FieldType": "string"},
+ {"FieldName": "acl_enabled", "FieldType": "bool"},
+ ],
+ },
+ {
+ "CollectionName": "docs",
+ "ScalarIndex": [
+ "account_id",
+ "acl_enabled",
+ ],
+ "Fields": [
+ {"FieldName": "legacy", "FieldType": "string"},
+ {"FieldName": "acl_enabled", "FieldType": "bool"},
+ ],
+ },
+ ]
+
+
+def test_update_preserves_existing_same_name_field_metadata() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._schema = {
+ "CollectionName": "docs",
+ "Fields": [{"FieldName": "legacy", "FieldType": "string", "DefaultValue": "keep"}],
+ }
+ collection._write_metadata_marker = lambda: None # type: ignore[method-assign]
+
+ collection.update(
+ fields=[
+ {"FieldName": "legacy", "FieldType": "int64", "DefaultValue": 0},
+ {"FieldName": "acl_enabled", "FieldType": "bool"},
+ ]
+ )
+
+ assert collection.get_meta_data()["Fields"] == [
+ {"FieldName": "legacy", "FieldType": "string", "DefaultValue": "keep"},
+ {"FieldName": "acl_enabled", "FieldType": "bool"},
+ ]
+
+
+@pytest.mark.asyncio
+async def test_update_collection_schema_creates_missing_qdrant_index() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._schema = {
+ "CollectionName": "docs",
+ "Fields": [{"FieldName": "account_id", "FieldType": "string"}],
+ "ScalarIndex": ["account_id", "tenant_custom"],
+ }
+ collection._indexes = {}
+ collection._write_metadata_marker = lambda: None # type: ignore[method-assign]
+ requests: list[tuple[str, str, dict[str, object], dict[str, object]]] = []
+
+ def request(method: str, path: str, body=None, *, params=None):
+ requests.append((method, path, body or {}, params or {}))
+ return {}
+
+ collection._client.request = request # type: ignore[method-assign]
+ adapter = type(
+ "_Adapter",
+ (),
+ {
+ "mode": "qdrant",
+ "_distance_metric": "cosine",
+ "_sparse_weight": 0.0,
+ "get_collection": lambda self: collection,
+ "_build_default_index_meta": lambda self, **kwargs: {
+ "IndexName": kwargs["index_name"],
+ "ScalarIndex": kwargs["scalar_index_fields"],
+ },
+ },
+ )()
+
+ await _AsyncVectorAdapter(adapter).update_collection_schema(
+ [
+ {"FieldName": "account_id", "FieldType": "string"},
+ {"FieldName": "acl_enabled", "FieldType": "bool"},
+ ],
+ ["account_id", "acl_enabled"],
+ "default",
+ )
+
+ assert collection.get_index_meta_data("default") == {
+ "IndexName": "default",
+ "ScalarIndex": ["account_id", "tenant_custom", "acl_enabled"],
+ }
+ assert (
+ "PUT",
+ "/collections/docs/index",
+ {"field_name": "tenant_custom", "field_schema": "keyword"},
+ {"wait": "true"},
+ ) in requests
+ assert (
+ "PUT",
+ "/collections/docs/index",
+ {"field_name": "acl_enabled", "field_schema": "bool"},
+ {"wait": "true"},
+ ) in requests
+
+
+@pytest.mark.asyncio
+async def test_update_collection_schema_preserves_custom_qdrant_indexes() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._schema = {
+ "CollectionName": "docs",
+ "Fields": [
+ {"FieldName": "account_id", "FieldType": "string"},
+ {"FieldName": "tenant_custom", "FieldType": "string"},
+ ],
+ "ScalarIndex": ["account_id", "tenant_custom"],
+ }
+ collection._indexes = {
+ "default": {"ScalarIndex": ["account_id", "tenant_custom"]},
+ }
+ collection._write_metadata_marker = lambda: None # type: ignore[method-assign]
+ requests: list[tuple[str, str, dict[str, object], dict[str, object]]] = []
+
+ def request(method: str, path: str, body=None, *, params=None):
+ requests.append((method, path, body or {}, params or {}))
+ return {}
+
+ collection._client.request = request # type: ignore[method-assign]
+ adapter = type(
+ "_Adapter",
+ (),
+ {"mode": "qdrant", "get_collection": lambda self: collection},
+ )()
+
+ await _AsyncVectorAdapter(adapter).update_collection_schema(
+ [
+ {"FieldName": "account_id", "FieldType": "string"},
+ {"FieldName": "acl_enabled", "FieldType": "bool"},
+ ],
+ ["account_id", "acl_enabled"],
+ "default",
+ )
+
+ assert collection.get_meta_data()["ScalarIndex"] == [
+ "account_id",
+ "tenant_custom",
+ "acl_enabled",
+ ]
+ assert collection.get_index_meta_data("default")["ScalarIndex"] == [
+ "account_id",
+ "tenant_custom",
+ "acl_enabled",
+ ]
+ assert not any(
+ method == "DELETE" and path.endswith("/tenant_custom")
+ for method, path, _body, _params in requests
+ )
+
+
+def test_drop_index_removes_remote_payload_indexes_and_metadata() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._indexes = {"default": {"ScalarIndex": ["account_id"]}}
+ marker_writes: list[dict[str, object]] = []
+ collection._write_metadata_marker = lambda: marker_writes.append( # type: ignore[method-assign]
+ dict(collection._indexes)
+ )
+
+ requests: list[tuple[str, str, dict[str, object], dict[str, object]]] = []
+
+ def request(method: str, path: str, body=None, *, params=None):
+ requests.append((method, path, body or {}, params or {}))
+ return {}
+
+ collection._client.request = request # type: ignore[method-assign]
+
+ assert collection.drop_index("default") is True
+ assert requests == [
+ (
+ "DELETE",
+ "/collections/docs/index/account_id",
+ {},
+ {"wait": "true"},
+ ),
+ (
+ "DELETE",
+ "/collections/docs/index/uri_depth",
+ {},
+ {"wait": "true"},
+ ),
+ (
+ "DELETE",
+ "/collections/docs/index/scope_roots",
+ {},
+ {"wait": "true"},
+ ),
+ ]
+ assert collection.list_indexes() == []
+ assert marker_writes == [{}]
+
+
+def test_drop_index_keeps_shared_uri_indexes_for_remaining_indexes() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._indexes = {
+ "one": {"ScalarIndex": ["account_id"]},
+ "two": {"ScalarIndex": ["kind"]},
+ }
+ collection._write_metadata_marker = lambda: None # type: ignore[method-assign]
+ requests: list[tuple[str, str, dict[str, object], dict[str, object]]] = []
+
+ def request(method: str, path: str, body=None, *, params=None):
+ requests.append((method, path, body or {}, params or {}))
+ return {}
+
+ collection._client.request = request # type: ignore[method-assign]
+
+ assert collection.drop_index("one") is True
+ assert requests == [
+ (
+ "DELETE",
+ "/collections/docs/index/account_id",
+ {},
+ {"wait": "true"},
+ )
+ ]
+
+
+def test_update_index_removes_remote_fields_removed_from_metadata() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._indexes = {"default": {"ScalarIndex": ["account_id", "kind"]}}
+ collection._write_metadata_marker = lambda: None # type: ignore[method-assign]
+ requests: list[tuple[str, str, dict[str, object], dict[str, object]]] = []
+
+ def request(method: str, path: str, body=None, *, params=None):
+ requests.append((method, path, body or {}, params or {}))
+ return {}
+
+ collection._client.request = request # type: ignore[method-assign]
+
+ assert collection.update_index("default", scalar_index=["account_id"]) == {
+ "ScalarIndex": ["account_id"]
+ }
+ assert (
+ "DELETE",
+ "/collections/docs/index/kind",
+ {},
+ {"wait": "true"},
+ ) in requests
+ assert collection.get_index_meta_data("default") == {
+ "ScalarIndex": ["account_id"]
+ }
+
+
+def test_update_index_ignores_missing_indexes() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ requests: list[tuple[str, str, dict[str, object], dict[str, object]]] = []
+
+ def request(method: str, path: str, body=None, *, params=None):
+ requests.append((method, path, body or {}, params or {}))
+ return {}
+
+ collection._client.request = request # type: ignore[method-assign]
+ collection._write_metadata_marker = lambda: pytest.fail( # type: ignore[method-assign]
+ "missing index must not publish metadata"
+ )
+
+ assert collection.update_index("missing", scalar_index=["foo"]) is None
+ assert collection.list_indexes() == []
+ assert requests == []
+
+
+def test_drop_index_keeps_metadata_when_remote_delete_fails() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._indexes = {"default": {"ScalarIndex": ["account_id"]}}
+ marker_writes: list[dict[str, object]] = []
+ collection._write_metadata_marker = lambda: marker_writes.append( # type: ignore[method-assign]
+ dict(collection._indexes)
+ )
+
+ def request(*_args, **_kwargs):
+ raise QdrantError("delete failed", status=503)
+
+ collection._client.request = request # type: ignore[method-assign]
+
+ with pytest.raises(QdrantError, match="delete failed"):
+ collection.drop_index("default")
+ assert collection.get_index_meta_data("default") == {
+ "ScalarIndex": ["account_id"]
+ }
+ assert marker_writes == []
+
+
+def test_drop_index_keeps_retryable_metadata_when_marker_write_fails() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._indexes = {"default": {"ScalarIndex": ["account_id"]}}
+ marker_attempts = 0
+
+ def write_marker():
+ nonlocal marker_attempts
+ marker_attempts += 1
+ if marker_attempts == 1:
+ raise QdrantError("marker failed", status=503)
+
+ collection._write_metadata_marker = write_marker # type: ignore[method-assign]
+ collection._client.request = lambda *_args, **_kwargs: {} # type: ignore[method-assign]
+
+ with pytest.raises(QdrantError, match="marker failed"):
+ collection.drop_index("default")
+ assert collection.has_index("default")
+
+ assert collection.drop_index("default") is True
+ assert not collection.has_index("default")
+ assert marker_attempts == 2
+
+
+def test_update_index_keeps_retryable_metadata_when_marker_write_fails() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._indexes = {"default": {"ScalarIndex": ["account_id", "kind"]}}
+ marker_attempts = 0
+
+ def write_marker():
+ nonlocal marker_attempts
+ marker_attempts += 1
+ if marker_attempts == 1:
+ raise QdrantError("marker failed", status=503)
+
+ collection._write_metadata_marker = write_marker # type: ignore[method-assign]
+ collection._client.request = lambda *_args, **_kwargs: {} # type: ignore[method-assign]
+
+ with pytest.raises(QdrantError, match="marker failed"):
+ collection.update_index("default", scalar_index=["account_id"])
+ assert collection.get_index_meta_data("default") == {
+ "ScalarIndex": ["account_id", "kind"]
+ }
+
+ assert collection.update_index("default", scalar_index=["account_id"]) == {
+ "ScalarIndex": ["account_id"]
+ }
+ assert collection.get_index_meta_data("default") == {
+ "ScalarIndex": ["account_id"]
+ }
+ assert marker_attempts == 2
+
+
+def test_create_index_keeps_retryable_metadata_when_marker_write_fails() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ marker_attempts = 0
+
+ def write_marker():
+ nonlocal marker_attempts
+ marker_attempts += 1
+ if marker_attempts == 1:
+ raise QdrantError("marker failed", status=503)
+
+ collection._write_metadata_marker = write_marker # type: ignore[method-assign]
+ collection._client.request = lambda *_args, **_kwargs: {} # type: ignore[method-assign]
+
+ with pytest.raises(QdrantError, match="marker failed"):
+ collection.create_index("default", {"ScalarIndex": ["account_id"]})
+ assert not collection.has_index("default")
+
+ assert collection.create_index("default", {"ScalarIndex": ["account_id"]}) == {
+ "ScalarIndex": ["account_id"]
+ }
+ assert collection.has_index("default")
+ assert marker_attempts == 2
+
+
+def test_create_index_does_not_publish_metadata_after_remote_400() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+
+ def request(*_args, **_kwargs):
+ raise QdrantError("bad schema", status=400)
+
+ collection._client.request = request # type: ignore[method-assign]
+ with pytest.raises(QdrantError, match="bad schema"):
+ collection.create_index("default", {"ScalarIndex": ["account_id"]})
+ assert collection.list_indexes() == []
+
+
+def test_sparse_term_collision_raises_instead_of_merging() -> None:
+ persisted: dict[str, int] = {}
+ by_index = {7: "existing-term"}
+
+ dictionary = SparseTermDictionary(
+ resolve_term=lambda term: persisted.get(term),
+ resolve_index=lambda index: by_index.get(index),
+ persist=lambda term, index: persisted.__setitem__(term, index),
+ hash_term=lambda _term: 7,
+ )
+
+ with pytest.raises(ValueError, match="sparse term index collision"):
+ dictionary.index_for("new-term")
+
+
+def test_sparse_term_index_fits_qdrant_uint32() -> None:
+ index = stable_sparse_index("qdrant")
+
+ assert 0 < index <= 0x7FFF_FFFF
+
+
+def test_sparse_term_index_rejects_values_outside_qdrant_range() -> None:
+ dictionary = SparseTermDictionary(
+ resolve_term=lambda _term: None,
+ resolve_index=lambda _index: None,
+ persist=lambda _term, _index: None,
+ hash_term=lambda _term: 0x1_0000_0000,
+ )
+
+ with pytest.raises(ValueError, match="Qdrant-compatible uint32"):
+ dictionary.index_for("token")
+
+
+def test_sparse_encoding_rejects_non_finite_weights() -> None:
+ dictionary = SparseTermDictionary(
+ resolve_term=lambda _term: None,
+ resolve_index=lambda _index: None,
+ persist=lambda _term, _index: None,
+ )
+
+ with pytest.raises(ValueError, match="finite"):
+ dictionary.encode({"token": math.nan})
+
+
+def test_contains_is_rejected_until_substring_semantics_are_defined() -> None:
+ with pytest.raises(NotImplementedError, match="Contains"):
+ compile_qdrant_filter(Contains("name", "partial"))
+
+
+def test_legacy_raw_filter_is_compiled_instead_of_sent_to_qdrant_unchanged() -> None:
+ assert compile_qdrant_filter(
+ {
+ "op": "and",
+ "conds": [
+ {"op": "must", "field": "account_id", "conds": ["acct"]},
+ {
+ "op": "must",
+ "field": "uri",
+ "conds": ["viking://resources/doc.md"],
+ "para": "-d=0",
+ },
+ ],
+ }
+ ) == {
+ "must": [
+ {"key": "account_id", "match": {"value": "acct"}},
+ {"key": "uri", "match": {"value": "/resources/doc.md"}},
+ ]
+ }
+
+
+def test_rest_client_sends_json_and_api_key() -> None:
+ transport = _ScriptedTransport((200, {"result": {"ok": True}}))
+ client = QdrantRestClient(
+ "http://qdrant.local/",
+ api_key="secret",
+ timeout_seconds=3,
+ opener=transport,
+ )
+
+ assert client.request("post", "collections/demo", {"hello": "world"}, params={"wait": True}) == {
+ "result": {"ok": True}
+ }
+ request = transport.requests[0]
+ assert request["method"] == "POST"
+ assert urlsplit(request["url"]).path == "/collections/demo"
+ assert parse_qs(urlsplit(request["url"]).query) == {"wait": ["True"]}
+ assert request["body"] == {"hello": "world"}
+ assert request["timeout"] == 3.0
+ assert request["headers"]["Api-key"] == "secret"
+
+
+def test_collection_lifecycle_writes_marker_and_payload_indexes() -> None:
+ transport = _ScriptedTransport(
+ (404, {}),
+ (200, {"result": True}),
+ (404, {}),
+ (200, {"result": True}),
+ (200, {"result": True}),
+ (200, {"result": True}),
+ (200, {"result": True}),
+ (200, {"result": True}),
+ (200, {"result": True}),
+ (200, {"result": True}),
+ )
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=transport),
+ collection_name="project__docs",
+ metadata_collection_name="project__docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=3,
+ distance="cosine",
+ sparse_enabled=True,
+ sparse_weight=0.25,
+ )
+
+ collection.create_remote_collection(
+ {
+ "CollectionName": "docs",
+ "Fields": [{"FieldName": "vector", "Dim": 3}],
+ }
+ )
+ collection.create_index(
+ "default",
+ {"ScalarIndex": ["account_id", "search_tags"]},
+ )
+
+ assert [request["method"] for request in transport.requests] == [
+ "GET",
+ "PUT",
+ "GET",
+ "PUT",
+ "PUT",
+ "PUT",
+ "PUT",
+ "PUT",
+ "PUT",
+ "PUT",
+ ]
+ assert urlsplit(transport.requests[1]["url"]).path == "/collections/project__docs"
+ assert transport.requests[1]["body"] == {
+ "vectors": {"dense": {"size": 3, "distance": "Cosine"}},
+ "sparse_vectors": {"sparse": {}},
+ }
+ assert urlsplit(transport.requests[3]["url"]).path == "/collections/project__docs__meta"
+ marker = transport.requests[4]["body"]["points"][0]
+ assert marker["vector"] == {"meta": [0.0]}
+ assert marker["payload"]["_openviking_meta_version"] == 1
+ index_requests = transport.requests[5:9]
+ assert [urlsplit(request["url"]).path for request in index_requests] == [
+ "/collections/project__docs/index",
+ "/collections/project__docs/index",
+ "/collections/project__docs/index",
+ "/collections/project__docs/index",
+ ]
+ assert [request["body"]["field_name"] for request in index_requests] == [
+ "account_id",
+ "search_tags",
+ "uri_depth",
+ "scope_roots",
+ ]
+
+
+def test_collection_lifecycle_infers_dimension_from_vector_field_type() -> None:
+ transport = _ScriptedTransport(
+ (404, {}),
+ (200, {"result": True}),
+ (404, {}),
+ (200, {"result": True}),
+ (200, {"result": True}),
+ )
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=transport),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=0,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+
+ collection.create_remote_collection(
+ {
+ "CollectionName": "docs",
+ "Fields": [{"FieldName": "embedding", "FieldType": "vector", "Dim": 3}],
+ }
+ )
+
+ assert transport.requests[1]["body"]["vectors"] == {
+ "dense": {"size": 3, "distance": "Cosine"}
+ }
+
+
+def test_metadata_marker_round_trips_index_metadata() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ marker: dict[str, object] = {}
+ collection._client.request = lambda *args, **kwargs: {} # type: ignore[method-assign]
+ collection._upsert_points = lambda _name, points: marker.update(points[0]["payload"]) # type: ignore[method-assign]
+ collection._schema = {"CollectionName": "docs", "Fields": []}
+ collection.create_index("default", {"ScalarIndex": ["account_id"]})
+
+ reloaded = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ reloaded._load_metadata_marker = lambda: marker # type: ignore[method-assign]
+
+ assert reloaded.get_meta_data() == {"CollectionName": "docs", "Fields": []}
+ assert reloaded.has_index("default")
+ assert reloaded.get_index_meta_data("default") == {"ScalarIndex": ["account_id"]}
+
+
+def test_collection_crud_search_count_and_scalar_scroll_use_qdrant_shapes() -> None:
+ point_id = to_qdrant_point_id("doc-1")
+ transport = _ScriptedTransport(
+ (200, {"result": True}),
+ (
+ 200,
+ {
+ "result": [
+ {
+ "id": point_id,
+ "payload": {
+ "_openviking_original_id": "doc-1",
+ "uri": "/resources/doc.md",
+ "name": "doc.md",
+ },
+ }
+ ],
+ },
+ ),
+ (200, {"result": True}),
+ (200, {"result": {"count": 1}}),
+ (
+ 200,
+ {
+ "result": [
+ {
+ "id": point_id,
+ "score": 0.9,
+ "payload": {
+ "_openviking_original_id": "doc-1",
+ "name": "doc.md",
+ },
+ }
+ ],
+ },
+ ),
+ (
+ 200,
+ {
+ "result": {
+ "points": [
+ {
+ "id": point_id,
+ "payload": {
+ "_openviking_original_id": "doc-1",
+ "updated_at": 7,
+ },
+ }
+ ]
+ }
+ },
+ ),
+ )
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=transport),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="dot",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+
+ collection.upsert_data(
+ [{"id": "doc-1", "uri": "viking://resources/doc.md", "vector": [0.1, 0.2]}]
+ )
+ fetched = collection.fetch_data(["doc-1"])
+ collection.delete_data(["doc-1"])
+ counted = collection.aggregate_data("default")
+ searched = collection.search_by_vector(
+ "default",
+ dense_vector=[0.1, 0.2],
+ limit=1,
+ offset=0,
+ filters={"must": [{"key": "account_id", "match": {"value": "acct"}}]},
+ output_fields=["name"],
+ )
+ scalar = collection.search_by_scalar(
+ "default",
+ "updated_at",
+ order="desc",
+ limit=1,
+ output_fields=["updated_at"],
+ )
+
+ assert fetched.items[0].id == "doc-1"
+ assert counted.agg == {"_total": 1}
+ assert searched.data[0].id == "doc-1"
+ assert scalar.data[0].fields["updated_at"] == 7
+
+ upsert_body = transport.requests[0]["body"]["points"][0]
+ assert upsert_body["vector"] == {"dense": [0.1, 0.2]}
+ assert upsert_body["payload"]["uri"] == "/resources/doc.md"
+ search_request = transport.requests[4]
+ assert urlsplit(search_request["url"]).path == "/collections/docs/points/query"
+ assert search_request["body"] == {
+ "query": [0.1, 0.2],
+ "using": "dense",
+ "filter": {"must": [{"key": "account_id", "match": {"value": "acct"}}]},
+ "limit": 1,
+ "offset": 0,
+ "with_payload": {"include": ["name", "_openviking_original_id"]},
+ "with_vector": False,
+ }
+ assert transport.requests[5]["body"]["order_by"] == {
+ "key": "updated_at",
+ "direction": "desc",
+ }
+ assert transport.requests[5]["body"]["with_payload"] == {
+ "include": ["updated_at", "_openviking_original_id"]
+ }
+
+
+def test_sparse_query_uses_named_qdrant_sparse_vector_shape() -> None:
+ transport = _ScriptedTransport(
+ (
+ 200,
+ {
+ "result": [
+ {
+ "id": to_qdrant_point_id("doc-1"),
+ "score": 0.4,
+ "payload": {"_openviking_original_id": "doc-1"},
+ }
+ ],
+ },
+ )
+ )
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=transport),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=True,
+ sparse_weight=0.5,
+ )
+
+ collection.encode_sparse_vector = lambda vector: {"indices": [7], "values": [1.5]} # type: ignore[method-assign]
+ collection.search_by_vector(
+ "default",
+ sparse_vector={"token": 1.5},
+ limit=1,
+ )
+
+ request = transport.requests[0]
+ assert urlsplit(request["url"]).path == "/collections/docs/points/query"
+ assert request["body"]["query"] == {"indices": [7], "values": [1.5]}
+ assert request["body"]["using"] == "sparse"
+
+
+def test_sparse_decode_rejects_unknown_term_index() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=True,
+ sparse_weight=0.5,
+ )
+ collection._resolve_sparse_index = lambda _index: None # type: ignore[method-assign]
+
+ with pytest.raises(ValueError, match="unknown sparse term index"):
+ collection._decode_sparse_vector({"indices": [7], "values": [1.5]})
+
+
+def test_sparse_decode_rejects_malformed_vectors() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=True,
+ sparse_weight=0.5,
+ )
+
+ with pytest.raises(ValueError, match="indices and values"):
+ collection._decode_sparse_vector({"indices": [7], "values": []})
+ with pytest.raises(ValueError, match="sparse vector"):
+ collection._decode_sparse_vector([])
+
+
+def test_update_data_rejects_missing_records_before_upsert() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ collection._retrieve_points = lambda *_args, **_kwargs: [] # type: ignore[method-assign]
+ upserts: list[list[dict[str, object]]] = []
+ collection.upsert_data = lambda data: upserts.append(data) # type: ignore[method-assign]
+
+ with pytest.raises(ValueError, match="record not found"):
+ collection.update_data(
+ [{"id": "missing", "name": "new", "vector": [0.1, 0.2]}]
+ )
+ assert upserts == []
+
+
+def test_update_data_requires_primary_key() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+ upserts: list[list[dict[str, object]]] = []
+ collection.upsert_data = lambda data: upserts.append(data) # type: ignore[method-assign]
+
+ with pytest.raises(ValueError, match="primary key 'id' is required for update"):
+ collection.update_data([{"name": "missing-id"}])
+ assert upserts == []
+
+
+def test_scroll_follows_qdrant_next_page_offset() -> None:
+ transport = _ScriptedTransport(
+ (
+ 200,
+ {
+ "result": {
+ "points": [{"id": "first", "payload": {}}],
+ "next_page_offset": "cursor-2",
+ }
+ },
+ ),
+ (
+ 200,
+ {
+ "result": {
+ "points": [{"id": "second", "payload": {}}],
+ "next_page_offset": None,
+ }
+ },
+ ),
+ )
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=transport),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+
+ points = collection._scroll(
+ "docs",
+ filter=None,
+ limit=2,
+ )
+
+ assert [point["id"] for point in points] == ["first", "second"]
+ assert transport.requests[1]["body"]["offset"] == "cursor-2"
+
+
+def test_dense_query_rejects_wrong_dimension() -> None:
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=_ScriptedTransport()),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=False,
+ sparse_weight=0.0,
+ )
+
+ with pytest.raises(ValueError, match="dense query vector dimension"):
+ collection.search_by_vector("default", dense_vector=[1.0])
+
+
+def test_existing_unmarked_collection_fails_closed() -> None:
+ transport = _ScriptedTransport(
+ (200, {"result": True}),
+ (404, {}),
+ )
+ config = VectorDBBackendConfig(
+ backend="qdrant",
+ qdrant={"url": "http://qdrant.local"},
+ project="project",
+ name="docs",
+ dimension=2,
+ )
+ adapter = QdrantCollectionAdapter.from_config(config)
+
+ adapter._client = QdrantRestClient("http://qdrant.local", opener=transport)
+
+ with pytest.raises(RuntimeError, match="metadata is missing"):
+ adapter.get_collection()
+
+
+def test_qdrant_rejects_sparse_weight_outside_rrf_range() -> None:
+ config = VectorDBBackendConfig(
+ backend="qdrant",
+ qdrant={"url": "http://qdrant.local"},
+ sparse_weight=1.1,
+ dimension=2,
+ )
+
+ with pytest.raises(ValueError, match="sparse_weight"):
+ QdrantCollectionAdapter.from_config(config)
+
+
+def test_sparse_encoding_persists_terms_in_metadata_sidecar() -> None:
+ transport = _ScriptedTransport(
+ (200, {"result": {"points": []}}),
+ (200, {"result": {"points": []}}),
+ (200, {"result": True}),
+ (200, {"result": {"points": []}}),
+ )
+ collection = QdrantCollection(
+ client=QdrantRestClient("http://qdrant.local", opener=transport),
+ collection_name="docs",
+ metadata_collection_name="docs__meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=True,
+ sparse_weight=0.5,
+ )
+
+ encoded = collection.encode_sparse_vector({"token": 1.5})
+
+ assert encoded["indices"] and encoded["values"] == [1.5]
+ assert transport.requests[0]["body"]["filter"]["must"][0]["key"] == "term"
+ assert transport.requests[1]["body"]["filter"]["must"][0]["key"] == "index"
+ persisted = transport.requests[2]["body"]["points"][0]
+ assert persisted["payload"] == {
+ "_openviking_sparse_term": True,
+ "term": "token",
+ "index": encoded["indices"][0],
+ }
+
+
+def test_adapter_recomputes_physical_collection_name_when_logical_name_changes() -> None:
+ config = VectorDBBackendConfig(
+ backend="qdrant",
+ qdrant={"url": "http://qdrant.local"},
+ project="project",
+ name="initial",
+ dimension=2,
+ )
+ adapter = QdrantCollectionAdapter.from_config(config)
+
+ adapter._collection_name = "created"
+
+ assert adapter._new_collection()._collection_name == "project__created"
+
+
+def test_qdrant_config_accepts_nested_url_and_keeps_content_disabled() -> None:
+ config = VectorDBBackendConfig(
+ backend="qdrant",
+ qdrant={"url": "http://qdrant.local", "dense_vector_name": "dense"},
+ dimension=2,
+ )
+
+ adapter = QdrantCollectionAdapter.from_config(config)
+
+ assert adapter._client.base_url == "http://qdrant.local"
+ assert adapter._dense_vector_name == "dense"
+ assert adapter.USE_CONTENT_FIELD is False
+
+
+def test_qdrant_factory_registry_returns_qdrant_adapter() -> None:
+ config = VectorDBBackendConfig(
+ backend="qdrant",
+ qdrant={"url": "http://qdrant.local"},
+ dimension=2,
+ )
+
+ assert isinstance(create_collection_adapter(config), QdrantCollectionAdapter)
diff --git a/tests/storage/test_qdrant_integration.py b/tests/storage/test_qdrant_integration.py
new file mode 100644
index 0000000000..08aac39432
--- /dev/null
+++ b/tests/storage/test_qdrant_integration.py
@@ -0,0 +1,106 @@
+# Copyright (c) 2026 Beijing Volcano Engine Technology Co., Ltd.
+# SPDX-License-Identifier: Apache-2.0
+
+from __future__ import annotations
+
+import os
+import uuid
+
+import pytest
+
+from openviking.storage.expr import And, Eq, PathScope
+from openviking.storage.vectordb.collection.qdrant_collection import QdrantCollection
+from openviking.storage.vectordb.collection.qdrant_rest import QdrantRestClient
+from openviking.storage.vectordb.qdrant_utils import compile_qdrant_filter
+
+QDRANT_URL = os.environ.get("QDRANT_URL")
+requires_qdrant = pytest.mark.skipif(not QDRANT_URL, reason="QDRANT_URL not set")
+
+
+@requires_qdrant
+@pytest.mark.integration
+def test_qdrant_phase_1_and_phase_2_round_trip() -> None:
+ assert QDRANT_URL is not None
+ suffix = uuid.uuid4().hex[:12]
+ collection = QdrantCollection(
+ client=QdrantRestClient(
+ QDRANT_URL,
+ api_key=os.environ.get("QDRANT_API_KEY"),
+ ),
+ collection_name=f"openviking_integration_{suffix}",
+ metadata_collection_name=f"openviking_integration_{suffix}_meta",
+ dense_vector_name="dense",
+ sparse_vector_name="sparse",
+ vector_dim=2,
+ distance="cosine",
+ sparse_enabled=True,
+ sparse_weight=0.5,
+ )
+ try:
+ collection.create_remote_collection(
+ {
+ "CollectionName": collection._collection_name,
+ "Fields": [
+ {"FieldName": "account_id", "FieldType": "string"},
+ {"FieldName": "search_tags", "FieldType": "list"},
+ ],
+ }
+ )
+ collection.create_index(
+ "default",
+ {"ScalarIndex": ["account_id", "search_tags"]},
+ )
+ collection.upsert_data(
+ [
+ {
+ "id": "doc-a",
+ "account_id": "acct-a",
+ "search_tags": ["team=search", "env=prod"],
+ "uri": "viking://resources/wiki/a.md",
+ "vector": [1.0, 0.0],
+ "sparse_vector": {"qdrant": 1.0},
+ },
+ {
+ "id": "doc-b",
+ "account_id": "acct-b",
+ "search_tags": ["team=search"],
+ "uri": "viking://resources/wiki/b.md",
+ "vector": [0.0, 1.0],
+ "sparse_vector": {"other": 1.0},
+ },
+ ]
+ )
+
+ path_and_account = compile_qdrant_filter(
+ And(
+ [
+ Eq("account_id", "acct-a"),
+ PathScope("uri", "viking://resources/wiki", depth=-1),
+ ]
+ )
+ )
+ assert collection.aggregate_data("default", filters=path_and_account).agg == {"_total": 1}
+ assert collection.search_by_vector(
+ "default",
+ dense_vector=[1.0, 0.0],
+ filters=path_and_account,
+ limit=2,
+ ).data[0].id == "doc-a"
+ assert collection.search_by_vector(
+ "default",
+ sparse_vector={"qdrant": 1.0},
+ filters=path_and_account,
+ limit=2,
+ ).data[0].id == "doc-a"
+ assert collection.search_by_vector(
+ "default",
+ dense_vector=[1.0, 0.0],
+ sparse_vector={"qdrant": 1.0},
+ filters=path_and_account,
+ limit=2,
+ ).data[0].id == "doc-a"
+ finally:
+ try:
+ collection.drop()
+ except Exception:
+ pass