feat: refac entity and impl Vamana. (#371)

This commit is contained in:
luoxiaojian
2026-04-30 10:22:25 +08:00
committed by GitHub
parent 005680522b
commit efab064676
64 changed files with 8669 additions and 201 deletions
+415
View File
@@ -0,0 +1,415 @@
# Copyright 2025-present the zvec project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Tests for the ``use_contiguous_memory`` optimization on ``HnswIndexParam``.
The HNSW streamer supports two allocation strategies for graph nodes:
* ``use_contiguous_memory=False`` (default): each node allocates its own
linked buffer. Lower peak memory usage, worse cache locality.
* ``use_contiguous_memory=True``: a single contiguous arena holds every
node. Higher peak memory usage, better cache locality and search
throughput.
These tests exercise the Python surface end-to-end and make sure that
when a collection is created / reopened with ``use_contiguous_memory=True``
the underlying HNSW streamer entity is constructed correctly and serves
search traffic.
"""
from __future__ import annotations
import pickle
import sys
import numpy as np
import pytest
import zvec
from zvec import (
Collection,
CollectionOption,
CollectionSchema,
Doc,
FieldSchema,
HnswIndexParam,
HnswQueryParam,
InvertIndexParam,
VectorQuery,
VectorSchema,
)
from zvec.typing import DataType, IndexType, MetricType, QuantizeType
DIMENSION = 32
NUM_DOCS = 128
TOPK = 5
# ---------------------------------------------------------------------------
def _debug_hnsw_storage_mode(coll: Collection, column: str = "dense") -> str:
"""Return the internal HNSW entity storage mode for ``column``.
Exposes the debug-only introspection hook on the pybind11 ``_Collection``.
Only meaningful after ``optimize()`` has built a persisted HNSW index; on
a pure writing segment it will raise ``KeyError``.
"""
underlying = coll._obj # type: ignore[attr-defined]
return underlying._debug_hnsw_storage_mode(column)
def _build_schema(name: str, *, use_contiguous_memory: bool) -> CollectionSchema:
"""Create a simple schema with a single FP32 HNSW vector column."""
return CollectionSchema(
name=name,
fields=[
FieldSchema(
"id",
DataType.INT64,
nullable=False,
index_param=InvertIndexParam(enable_range_optimization=True),
),
],
vectors=[
VectorSchema(
"dense",
DataType.VECTOR_FP32,
dimension=DIMENSION,
index_param=HnswIndexParam(
metric_type=MetricType.IP,
m=16,
ef_construction=100,
use_contiguous_memory=use_contiguous_memory,
),
),
],
)
def _generate_docs(rng: np.random.Generator, num: int = NUM_DOCS) -> list[Doc]:
"""Produce deterministic documents for insertion."""
docs: list[Doc] = []
for i in range(num):
vec = rng.standard_normal(DIMENSION).astype(np.float32)
docs.append(
Doc(
id=str(i),
fields={"id": i},
vectors={"dense": vec.tolist()},
)
)
return docs
def _assert_query_matches(coll: Collection, query_vec: list[float]) -> list[str]:
"""Run a top-k vector query and return the returned ids in order."""
vector_query = VectorQuery(
field_name="dense",
vector=query_vec,
param=HnswQueryParam(ef=128),
)
hits = coll.query(vectors=vector_query, topk=TOPK)
# Expect a single result group for the single vector query.
assert hits is not None, "query returned None"
assert len(hits) >= 1, f"expected at least one hit, got {hits!r}"
return [doc.id for doc in hits]
# ---------------------------------------------------------------------------
# 1) Pure Python surface: construction / property / to_dict / repr / pickle
# ---------------------------------------------------------------------------
class TestHnswIndexParamContiguousMemorySurface:
"""Verify the Python binding exposes ``use_contiguous_memory`` correctly."""
def test_default_is_false(self):
param = HnswIndexParam()
assert param.use_contiguous_memory is False
def test_custom_true(self):
param = HnswIndexParam(use_contiguous_memory=True)
assert param.use_contiguous_memory is True
assert param.type == IndexType.HNSW
# other fields keep their default values
assert param.m == 50
assert param.ef_construction == 500
def test_to_dict_includes_use_contiguous_memory(self):
param = HnswIndexParam(
metric_type=MetricType.L2,
m=16,
ef_construction=100,
quantize_type=QuantizeType.FP16,
use_contiguous_memory=True,
)
data = param.to_dict()
assert data["use_contiguous_memory"] is True
# Make sure existing fields are still present.
assert data["metric_type"] == "L2"
assert data["m"] == 16
assert data["ef_construction"] == 100
assert data["quantize_type"] == "FP16"
def test_repr_contains_flag(self):
on = repr(HnswIndexParam(use_contiguous_memory=True))
off = repr(HnswIndexParam(use_contiguous_memory=False))
assert "use_contiguous_memory" in on
assert "use_contiguous_memory" in off
assert "true" in on
assert "false" in off
def test_readonly_property(self):
param = HnswIndexParam(use_contiguous_memory=True)
if sys.version_info >= (3, 11):
match_pattern = r"(can't set attribute|has no setter|readonly attribute)"
else:
match_pattern = r"can't set attribute"
with pytest.raises(AttributeError, match=match_pattern):
param.use_contiguous_memory = False # type: ignore[misc]
def test_pickle_roundtrip(self):
original = HnswIndexParam(
metric_type=MetricType.COSINE,
m=24,
ef_construction=150,
quantize_type=QuantizeType.INT8,
use_contiguous_memory=True,
)
restored = pickle.loads(pickle.dumps(original))
assert restored.use_contiguous_memory is True
assert restored.metric_type == MetricType.COSINE
assert restored.m == 24
assert restored.ef_construction == 150
assert restored.quantize_type == QuantizeType.INT8
# ---------------------------------------------------------------------------
# 2) End-to-end: create collection, insert, query with contiguous memory on
# ---------------------------------------------------------------------------
@pytest.fixture
def rng() -> np.random.Generator:
return np.random.default_rng(seed=42)
# NOTE: the ``enable_mmap=False`` (BufferPool) variant is intentionally
# omitted from this fixture. Building a persisted HNSW index via
# ``optimize()`` / ``create_vector_index`` / ``drop_vector_index``
# currently requires mmap-backed storage, because the BufferPool backend
# has not implemented the ``create_new`` semantics yet and the guard in
# ``SegmentImpl::merge_vector_indexer`` rejects that combination. Once
# BufferPool gains write support, re-add ``False`` to ``params`` (and
# drop the guard in segment.cc) so these end-to-end tests cover both
# storage modes again.
@pytest.fixture(params=[True], ids=["mmap_on"])
def collection_option(request) -> CollectionOption:
return CollectionOption(read_only=False, enable_mmap=request.param)
# Building a new persisted HNSW index currently requires mmap-backed storage
# because the BufferPool backend has not implemented `create_new` semantics
# yet. Collections opened with ``enable_mmap=False`` therefore cannot run
# optimize()/create_vector_index/drop_vector_index. Tests use this fixture
# to know which behaviour to assert, and once BufferPool gains write support
# the guard in segment.cc (and these branches) can be removed together.
@pytest.fixture
def build_index_supported(collection_option: CollectionOption) -> bool:
return bool(collection_option.enable_mmap)
# Error message fragments emitted by the NotSupported guard in
# SegmentImpl::merge_vector_indexer / drop_vector_index. If the C++ message
# changes, update these together.
_BUILD_NOT_SUPPORTED_FRAGMENTS = ("not yet supported", "enable_mmap=false")
class TestHnswContiguousMemoryEndToEnd:
"""End-to-end: schema -> create_and_open -> insert -> query works."""
def test_create_with_contiguous_memory_and_query(
self,
tmp_path_factory,
collection_option,
rng,
):
"""With the flag on, the schema round-trips and search works end-to-end.
After ``optimize()`` the writing segment is compacted into a persisted
segment backed by the configured HNSW entity. We assert both the
user-observable behaviour (schema + search) and, via the debug hook,
that the entity type actually honours ``use_contiguous_memory``.
"""
schema = _build_schema("hnsw_contig_create", use_contiguous_memory=True)
path = tmp_path_factory.mktemp("zvec") / "hnsw_contig_create"
coll = zvec.create_and_open(
path=str(path), schema=schema, option=collection_option
)
try:
# Schema round-trips with the flag set.
vec_schema = coll.schema.vectors[0]
assert vec_schema.index_param.use_contiguous_memory is True
docs = _generate_docs(rng)
insert_result = coll.insert(docs=docs)
for r in insert_result:
assert r.ok(), f"insert failed: code={r.code()}"
assert coll.stats.doc_count == NUM_DOCS
# Build persisted HNSW index; this is where the contiguous entity
# is actually instantiated.
coll.optimize()
assert _debug_hnsw_storage_mode(coll) == "contiguous", (
"use_contiguous_memory=True should produce a contiguous entity"
)
# Pick an existing vector as the query; top-1 must be itself.
query_vec = docs[0].vector("dense")
ids = _assert_query_matches(coll, query_vec)
assert ids[0] == "0", f"expected self-recall, got top-1 id={ids[0]}"
finally:
coll.destroy()
def test_create_without_contiguous_memory_uses_mmap_entity(
self,
tmp_path_factory,
collection_option,
rng,
):
"""Baseline: when the flag is omitted the default (mmap) entity is used."""
schema = _build_schema("hnsw_contig_default", use_contiguous_memory=False)
path = tmp_path_factory.mktemp("zvec") / "hnsw_contig_default"
coll = zvec.create_and_open(
path=str(path), schema=schema, option=collection_option
)
try:
vec_schema = coll.schema.vectors[0]
assert vec_schema.index_param.use_contiguous_memory is False
docs = _generate_docs(rng)
for r in coll.insert(docs=docs):
assert r.ok()
assert coll.stats.doc_count == NUM_DOCS
coll.optimize()
# With the flag off and mmap on, the persisted entity must be the
# default mmap layout — specifically, not the contiguous arena.
assert _debug_hnsw_storage_mode(coll) == "mmap", (
"use_contiguous_memory=False + enable_mmap=True should "
"produce the mmap entity"
)
# Search still functions with the default entity backing.
query_vec = docs[0].vector("dense")
ids = _assert_query_matches(coll, query_vec)
assert ids[0] == "0"
finally:
coll.destroy()
def test_close_and_reopen_with_contiguous_memory(
self,
tmp_path_factory,
collection_option,
rng,
):
"""Reopening a collection must preserve the ``use_contiguous_memory`` flag.
The core property: the flag survives the schema persist/reload
round-trip so the HNSW streamer entity — constructed lazily on first
persisted-segment build — honours the user's choice. We run
``optimize()`` after reopen and confirm the contiguous entity was
materialized.
"""
schema = _build_schema("hnsw_contig_reopen", use_contiguous_memory=True)
path = tmp_path_factory.mktemp("zvec") / "hnsw_contig_reopen"
path_str = str(path)
created = zvec.create_and_open(
path=path_str, schema=schema, option=collection_option
)
docs = _generate_docs(rng)
for r in created.insert(docs=docs):
assert r.ok()
assert created.stats.doc_count == NUM_DOCS
# Persist pending writes so that reopen reconstructs state from disk.
created.flush()
del created # close the handle
reopened = zvec.open(path=path_str, option=collection_option)
try:
assert reopened is not None
assert reopened.stats.doc_count == NUM_DOCS
# Schema persisted the flag across the reopen boundary.
vec_schema = reopened.schema.vectors[0]
assert vec_schema.index_param.use_contiguous_memory is True
reopened.optimize()
assert _debug_hnsw_storage_mode(reopened) == "contiguous"
# Entity actually works: exact self-recall + fetch parity.
query_vec = docs[7].vector("dense")
ids = _assert_query_matches(reopened, query_vec)
assert ids[0] == "7"
fetched = reopened.fetch([d.id for d in docs[:10]])
assert len(fetched) == 10
finally:
reopened.destroy()
def test_result_parity_with_and_without_contiguous_memory(
self,
tmp_path_factory,
rng,
):
"""
Two collections built from the same documents must return the same
top-k neighbors regardless of whether contiguous memory is enabled:
the flag is a memory-layout optimization and must not alter recall
for identical graph construction parameters on the same data.
"""
docs = _generate_docs(rng)
query_vec = docs[3].vector("dense")
def _build_and_query(tag: str, flag: bool) -> list[str]:
schema = _build_schema(f"hnsw_parity_{tag}", use_contiguous_memory=flag)
option = CollectionOption(read_only=False, enable_mmap=True)
path = tmp_path_factory.mktemp("zvec") / f"hnsw_parity_{tag}"
coll = zvec.create_and_open(path=str(path), schema=schema, option=option)
try:
for r in coll.insert(docs=docs):
assert r.ok()
coll.optimize()
expected_mode = "contiguous" if flag else "mmap"
assert _debug_hnsw_storage_mode(coll) == expected_mode, (
f"{tag}: unexpected entity type"
)
return _assert_query_matches(coll, query_vec)
finally:
coll.destroy()
ids_off = _build_and_query("off", flag=False)
ids_on = _build_and_query("on", flag=True)
# The graph is built with the same (m, ef_construction, data, order),
# so top-k results must match exactly.
assert ids_on == ids_off, (
f"top-{TOPK} results diverged between use_contiguous_memory modes: "
f"on={ids_on}, off={ids_off}"
)
# Sanity: self-recall is still perfect.
assert ids_on[0] == "3"
+481
View File
@@ -0,0 +1,481 @@
# Copyright 2025-present the zvec project
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Tests for the Python entry point of the Vamana (DiskANN) dense vector index.
Mirrors the structure of ``test_hnsw_contiguous_memory.py`` (the closest
hnsw dense reference), and is split into two parts:
1. **Surface tests** — verify that ``VamanaIndexParam`` / ``VamanaQueryParam``
are correctly bound: construction defaults, readonly properties,
``to_dict``, ``__repr__``, pickle round-trip, and that they appear in the
public ``zvec`` namespace with the expected ``IndexType.VAMANA`` value.
2. **End-to-end tests** — build a collection that uses Vamana on a dense
FP32 column, insert deterministic documents, then run a top-k query
through ``VamanaQueryParam`` on both the writer segment and the
persisted (post-``optimize()``) segment.
"""
from __future__ import annotations
import pickle
import sys
import numpy as np
import pytest
import zvec
from zvec import (
Collection,
CollectionOption,
CollectionSchema,
Doc,
FieldSchema,
InvertIndexParam,
VamanaIndexParam,
VamanaQueryParam,
VectorQuery,
VectorSchema,
)
from zvec.typing import DataType, IndexType, MetricType, QuantizeType
DIMENSION = 32
NUM_DOCS = 128
TOPK = 5
# Defaults pulled from src/include/zvec/core/interface/constants.h. Keep
# in sync with kDefaultVamana* if the engine defaults ever change.
DEFAULT_MAX_DEGREE = 64
DEFAULT_SEARCH_LIST_SIZE = 100
DEFAULT_ALPHA = 1.2
DEFAULT_EF_SEARCH = 200
DEFAULT_SATURATE_GRAPH = False
# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def _build_schema(
name: str,
*,
metric_type: MetricType = MetricType.IP,
max_degree: int = 32,
search_list_size: int = 64,
alpha: float = 1.2,
use_contiguous_memory: bool = False,
) -> CollectionSchema:
"""Create a simple schema with a single FP32 Vamana vector column."""
return CollectionSchema(
name=name,
fields=[
FieldSchema(
"id",
DataType.INT64,
nullable=False,
index_param=InvertIndexParam(enable_range_optimization=True),
),
],
vectors=[
VectorSchema(
"dense",
DataType.VECTOR_FP32,
dimension=DIMENSION,
index_param=VamanaIndexParam(
metric_type=metric_type,
max_degree=max_degree,
search_list_size=search_list_size,
alpha=alpha,
use_contiguous_memory=use_contiguous_memory,
),
),
],
)
def _generate_docs(rng: np.random.Generator, num: int = NUM_DOCS) -> list[Doc]:
"""Produce deterministic documents for insertion."""
docs: list[Doc] = []
for i in range(num):
vec = rng.standard_normal(DIMENSION).astype(np.float32)
docs.append(
Doc(
id=str(i),
fields={"id": i},
vectors={"dense": vec.tolist()},
)
)
return docs
def _query_topk(
coll: Collection, query_vec: list[float], *, ef_search: int = 64
) -> list[str]:
"""Run a top-k vector query and return the returned ids in order."""
vector_query = VectorQuery(
field_name="dense",
vector=query_vec,
param=VamanaQueryParam(ef_search=ef_search),
)
hits = coll.query(vectors=vector_query, topk=TOPK)
assert hits is not None, "query returned None"
assert len(hits) >= 1, f"expected at least one hit, got {hits!r}"
return [doc.id for doc in hits]
# ---------------------------------------------------------------------------
# 1) Surface: construction / property / to_dict / repr / pickle / namespace
# ---------------------------------------------------------------------------
class TestVamanaIndexParamSurface:
"""Verify the Python binding for ``VamanaIndexParam``."""
def test_defaults(self):
param = VamanaIndexParam()
assert param.type == IndexType.VAMANA
assert param.metric_type == MetricType.IP
assert param.max_degree == DEFAULT_MAX_DEGREE
assert param.search_list_size == DEFAULT_SEARCH_LIST_SIZE
assert param.alpha == pytest.approx(DEFAULT_ALPHA)
assert param.saturate_graph is DEFAULT_SATURATE_GRAPH
assert param.use_contiguous_memory is False
assert param.use_id_map is False
assert param.quantize_type == QuantizeType.UNDEFINED
def test_custom_construction(self):
param = VamanaIndexParam(
metric_type=MetricType.COSINE,
max_degree=48,
search_list_size=128,
alpha=1.5,
saturate_graph=True,
use_contiguous_memory=True,
use_id_map=False,
quantize_type=QuantizeType.INT8,
)
assert param.type == IndexType.VAMANA
assert param.metric_type == MetricType.COSINE
assert param.max_degree == 48
assert param.search_list_size == 128
assert param.alpha == pytest.approx(1.5)
assert param.saturate_graph is True
assert param.use_contiguous_memory is True
assert param.use_id_map is False
assert param.quantize_type == QuantizeType.INT8
def test_to_dict_includes_all_fields(self):
param = VamanaIndexParam(
metric_type=MetricType.L2,
max_degree=32,
search_list_size=80,
alpha=1.3,
saturate_graph=True,
use_contiguous_memory=True,
use_id_map=False,
quantize_type=QuantizeType.FP16,
)
data = param.to_dict()
assert data["type"] == "VAMANA"
assert data["metric_type"] == "L2"
assert data["max_degree"] == 32
assert data["search_list_size"] == 80
assert data["alpha"] == pytest.approx(1.3)
assert data["saturate_graph"] is True
assert data["use_contiguous_memory"] is True
assert data["use_id_map"] is False
assert data["quantize_type"] == "FP16"
def test_repr_contains_key_fields(self):
text = repr(
VamanaIndexParam(
metric_type=MetricType.COSINE,
max_degree=24,
search_list_size=72,
alpha=1.4,
saturate_graph=True,
use_contiguous_memory=True,
)
)
# Spot-check the most diagnostic fields are rendered.
assert "VAMANA" in text
assert "COSINE" in text
assert "max_degree" in text and "24" in text
assert "search_list_size" in text and "72" in text
assert "alpha" in text
assert "saturate_graph" in text and "true" in text
assert "use_contiguous_memory" in text and "true" in text
@pytest.mark.parametrize(
"field, kwargs",
[
("max_degree", dict(max_degree=99)),
("search_list_size", dict(search_list_size=99)),
("alpha", dict(alpha=1.7)),
("saturate_graph", dict(saturate_graph=True)),
("use_contiguous_memory", dict(use_contiguous_memory=True)),
("use_id_map", dict(use_id_map=True)),
],
)
def test_readonly_properties(self, field, kwargs):
param = VamanaIndexParam(**kwargs)
if sys.version_info >= (3, 11):
match_pattern = r"(can't set attribute|has no setter|readonly attribute)"
else:
match_pattern = r"can't set attribute"
with pytest.raises(AttributeError, match=match_pattern):
setattr(param, field, getattr(param, field))
def test_pickle_roundtrip(self):
original = VamanaIndexParam(
metric_type=MetricType.COSINE,
max_degree=48,
search_list_size=120,
alpha=1.4,
saturate_graph=True,
use_contiguous_memory=True,
use_id_map=False,
quantize_type=QuantizeType.INT8,
)
restored = pickle.loads(pickle.dumps(original))
assert restored.type == IndexType.VAMANA
assert restored.metric_type == MetricType.COSINE
assert restored.max_degree == 48
assert restored.search_list_size == 120
assert restored.alpha == pytest.approx(1.4)
assert restored.saturate_graph is True
assert restored.use_contiguous_memory is True
assert restored.use_id_map is False
assert restored.quantize_type == QuantizeType.INT8
# to_dict equality is the strongest end-to-end equivalence we have.
assert restored.to_dict() == original.to_dict()
class TestVamanaQueryParamSurface:
"""Verify the Python binding for ``VamanaQueryParam``."""
def test_defaults(self):
q = VamanaQueryParam()
assert q.type == IndexType.VAMANA
assert q.ef_search == DEFAULT_EF_SEARCH
assert q.radius == pytest.approx(0.0)
assert q.is_linear is False
assert q.is_using_refiner is False
def test_custom_construction(self):
q = VamanaQueryParam(
ef_search=300, radius=0.5, is_linear=True, is_using_refiner=True
)
assert q.type == IndexType.VAMANA
assert q.ef_search == 300
assert q.radius == pytest.approx(0.5)
assert q.is_linear is True
assert q.is_using_refiner is True
def test_repr_contains_key_fields(self):
text = repr(VamanaQueryParam(ef_search=128, radius=0.25))
assert "VAMANA" in text
assert "ef_search" in text and "128" in text
assert "radius" in text
def test_readonly_ef_search(self):
q = VamanaQueryParam(ef_search=100)
if sys.version_info >= (3, 11):
match_pattern = r"(can't set attribute|has no setter|readonly attribute)"
else:
match_pattern = r"can't set attribute"
with pytest.raises(AttributeError, match=match_pattern):
q.ef_search = 200 # type: ignore[misc]
def test_pickle_roundtrip(self):
original = VamanaQueryParam(
ef_search=256, radius=0.3, is_linear=False, is_using_refiner=True
)
restored = pickle.loads(pickle.dumps(original))
assert restored.type == IndexType.VAMANA
assert restored.ef_search == 256
assert restored.radius == pytest.approx(0.3)
assert restored.is_linear is False
assert restored.is_using_refiner is True
class TestVamanaPublicNamespace:
"""The Vamana entry points must be importable from the top-level ``zvec``."""
def test_top_level_exports(self):
assert zvec.VamanaIndexParam is VamanaIndexParam
assert zvec.VamanaQueryParam is VamanaQueryParam
assert "VamanaIndexParam" in zvec.__all__
assert "VamanaQueryParam" in zvec.__all__
def test_index_type_enum_member(self):
# Sanity: the IndexType enum exposes VAMANA and it is what the
# bound params advertise.
assert IndexType.VAMANA is not None
assert VamanaIndexParam().type == IndexType.VAMANA
assert VamanaQueryParam().type == IndexType.VAMANA
# ---------------------------------------------------------------------------
# 2) End-to-end: create collection, insert, query through the writer segment
# ---------------------------------------------------------------------------
@pytest.fixture
def rng() -> np.random.Generator:
return np.random.default_rng(seed=42)
# Mirror the hnsw dense test fixture: only the mmap-backed variant is
# currently usable for vector index construction. BufferPool (enable_mmap=
# False) is intentionally omitted because the same write-path guard in
# ``SegmentImpl::merge_vector_indexer`` rejects that combination.
@pytest.fixture(params=[True], ids=["mmap_on"])
def collection_option(request) -> CollectionOption:
return CollectionOption(read_only=False, enable_mmap=request.param)
class TestVamanaEndToEnd:
"""End-to-end: schema -> create_and_open -> insert -> query works."""
def test_schema_round_trip(self, tmp_path_factory, collection_option):
"""The Vamana index params survive the schema persist path."""
schema = _build_schema(
"vamana_schema_rt",
metric_type=MetricType.COSINE,
max_degree=32,
search_list_size=80,
alpha=1.3,
use_contiguous_memory=True,
)
path = tmp_path_factory.mktemp("zvec") / "vamana_schema_rt"
coll = zvec.create_and_open(
path=str(path), schema=schema, option=collection_option
)
try:
vec_schema = coll.schema.vectors[0]
ip = vec_schema.index_param
assert ip.type == IndexType.VAMANA
assert ip.metric_type == MetricType.COSINE
assert ip.max_degree == 32
assert ip.search_list_size == 80
assert ip.alpha == pytest.approx(1.3)
assert ip.use_contiguous_memory is True
finally:
coll.destroy()
def test_insert_and_query_self_recall(
self, tmp_path_factory, collection_option, rng
):
"""Top-1 of a query equal to an inserted vector must be that vector.
Exercises the writer-segment Vamana streamer end-to-end through the
Python entry point: ``VamanaIndexParam`` for build and
``VamanaQueryParam`` for search.
"""
schema = _build_schema("vamana_e2e_recall")
path = tmp_path_factory.mktemp("zvec") / "vamana_e2e_recall"
coll = zvec.create_and_open(
path=str(path), schema=schema, option=collection_option
)
try:
docs = _generate_docs(rng)
for r in coll.insert(docs=docs):
assert r.ok(), f"insert failed: code={r.code()}"
assert coll.stats.doc_count == NUM_DOCS
# Self-recall: query with the i-th inserted vector, expect id i
# to be the top result.
for probe in (0, 7, 42, NUM_DOCS - 1):
query_vec = docs[probe].vector("dense")
ids = _query_topk(coll, query_vec)
assert ids[0] == str(probe), (
f"expected self-recall at probe={probe}, got top-1 id={ids[0]} "
f"(top-{TOPK}={ids})"
)
finally:
coll.destroy()
def test_query_param_ef_search_affects_only_quality(
self, tmp_path_factory, collection_option, rng
):
"""``ef_search`` is a search-time knob and must not crash for any
sensible value. Larger ``ef_search`` should be at least as good as
smaller for self-recall."""
schema = _build_schema("vamana_e2e_ef")
path = tmp_path_factory.mktemp("zvec") / "vamana_e2e_ef"
coll = zvec.create_and_open(
path=str(path), schema=schema, option=collection_option
)
try:
docs = _generate_docs(rng)
for r in coll.insert(docs=docs):
assert r.ok()
query_vec = docs[3].vector("dense")
ids_small = _query_topk(coll, query_vec, ef_search=16)
ids_large = _query_topk(coll, query_vec, ef_search=256)
# Both should self-recall the probe vector at top-1.
assert ids_small[0] == "3"
assert ids_large[0] == "3"
assert len(ids_small) == TOPK
assert len(ids_large) == TOPK
finally:
coll.destroy()
def test_optimize_then_query(self, tmp_path_factory, collection_option, rng):
"""The persisted Vamana segment built by ``optimize()`` must serve
queries correctly.
Until the cmake fix to force-load ``core_knn_vamana_static`` into the
``_zvec`` pybind module, this path failed at ``VamanaStreamer``
creation because the global factory registration in
``vamana_streamer.cc`` was never linked in. This test pins down the
regression.
"""
schema = _build_schema("vamana_e2e_optimize")
path = tmp_path_factory.mktemp("zvec") / "vamana_e2e_optimize"
coll = zvec.create_and_open(
path=str(path), schema=schema, option=collection_option
)
try:
docs = _generate_docs(rng)
for r in coll.insert(docs=docs):
assert r.ok()
assert coll.stats.doc_count == NUM_DOCS
# Snapshot the writer-segment top-k for a probe vector.
query_vec = docs[5].vector("dense")
ids_pre = _query_topk(coll, query_vec)
assert ids_pre[0] == "5"
# Trigger persisted segment build. Pre-fix this raised
# RuntimeError("Failed to create index").
coll.optimize()
# Persisted segment must still serve queries with the same
# top-1 self-recall guarantee. We do not assert full top-k
# equality with the writer segment because the persisted
# streamer may visit nodes in a different order; top-1 self-
# recall is the strong invariant.
ids_post = _query_topk(coll, query_vec)
assert ids_post[0] == "5", (
f"post-optimize top-1 should still be probe id, got {ids_post}"
)
assert len(ids_post) == TOPK
finally:
coll.destroy()
+4
View File
@@ -70,6 +70,8 @@ from .model.param import (
IVFIndexParam,
IVFQueryParam,
OptimizeOption,
VamanaIndexParam,
VamanaQueryParam,
)
from .model.param.vector_query import VectorQuery
@@ -122,6 +124,8 @@ __all__ = [
"HnswQueryParam",
"HnswRabitqQueryParam",
"IVFQueryParam",
"VamanaIndexParam",
"VamanaQueryParam",
# Extensions
"DenseEmbeddingFunction",
"SparseEmbeddingFunction",
+11
View File
@@ -26,6 +26,8 @@ from .model.param import (
IVFIndexParam,
IVFQueryParam,
OptimizeOption,
VamanaIndexParam,
VamanaQueryParam,
)
from .model.param.vector_query import VectorQuery
from .model.schema import CollectionSchema, CollectionStats, FieldSchema, VectorSchema
@@ -73,6 +75,8 @@ __all__: list = [
"RrfReRanker",
"Status",
"StatusCode",
"VamanaIndexParam",
"VamanaQueryParam",
"VectorQuery",
"VectorSchema",
"WeightedReRanker",
@@ -122,6 +126,13 @@ class _Collection:
def Stats(self) -> schema.CollectionStats: ...
def Update(self, arg0: collections.abc.Sequence[_Doc]) -> list[typing.Status]: ...
def Upsert(self, arg0: collections.abc.Sequence[_Doc]) -> list[typing.Status]: ...
def _debug_hnsw_storage_mode(self, column_name: str) -> str:
"""Debug-only: returns the storage mode of the HNSW entity on the
given vector column. One of 'mmap', 'buffer_pool', 'contiguous'.
Raises KeyError if no HNSW index exists on the column, or
ValueError if the column's index is not an HNSW index. Intended
for introspection and testing only; not part of the stable API."""
def __getstate__(self) -> tuple: ...
def __setstate__(self, arg0: tuple) -> None: ...
+4
View File
@@ -27,6 +27,8 @@ from _zvec.param import (
IVFIndexParam,
IVFQueryParam,
OptimizeOption,
VamanaIndexParam,
VamanaQueryParam,
)
__all__ = [
@@ -43,4 +45,6 @@ __all__ = [
"IndexOption",
"InvertIndexParam",
"OptimizeOption",
"VamanaIndexParam",
"VamanaQueryParam",
]
+16 -2
View File
@@ -198,6 +198,10 @@ class HnswIndexParam(VectorIndexParam):
quantize_type (QuantizeType): Optional quantization type for vector
compression (e.g., FP16, INT8). Default is `QuantizeType.UNDEFINED` to
disable quantization.
use_contiguous_memory (bool): If True, the HNSW streamer allocates a
single contiguous memory arena for all graph nodes, improving cache
locality and search throughput at the cost of peak memory usage.
Default is False.
Examples:
>>> from zvec.typing import MetricType, QuantizeType
@@ -205,10 +209,11 @@ class HnswIndexParam(VectorIndexParam):
... metric_type=MetricType.COSINE,
... m=16,
... ef_construction=200,
... quantize_type=QuantizeType.INT8
... quantize_type=QuantizeType.INT8,
... use_contiguous_memory=True,
... )
>>> print(params)
{'metric_type': 'IP', 'm': 16, 'ef_construction': 200, 'quantize_type': 'INT8'}
{'metric_type': 'IP', 'm': 16, 'ef_construction': 200, 'quantize_type': 'INT8', 'use_contiguous_memory': True}
"""
def __getstate__(self) -> tuple: ...
@@ -218,6 +223,7 @@ class HnswIndexParam(VectorIndexParam):
m: typing.SupportsInt = 50,
ef_construction: typing.SupportsInt = 500,
quantize_type: _zvec.typing.QuantizeType = ...,
use_contiguous_memory: bool = False,
) -> None: ...
def __repr__(self) -> str: ...
def __setstate__(self, arg0: tuple) -> None: ...
@@ -238,6 +244,14 @@ class HnswIndexParam(VectorIndexParam):
int: Maximum number of neighbors per node in upper layers.
"""
@property
def use_contiguous_memory(self) -> bool:
"""
bool: Whether to allocate a single contiguous memory arena for all
HNSW graph nodes. Improves cache locality and search throughput at
the cost of peak memory usage. Defaults to False.
"""
class HnswQueryParam(QueryParam):
"""