diff --git a/cmake/svs.cmake b/cmake/svs.cmake index 6174d827c..fa28df32c 100644 --- a/cmake/svs.cmake +++ b/cmake/svs.cmake @@ -123,6 +123,29 @@ if(USE_SVS) message("SVS LVQ implementation not found") add_compile_definitions(VectorSimilarity PUBLIC "HAVE_SVS_LVQ=0") endif() + + # `replace_external_id` is newer than the pre-built SVS releases that SVS_SHARED_LIB downloads, + # so whether it is available depends on which SVS this build ended up with rather than on the + # platform. Detect it in the header instead of assuming: it is a method, not a file, so this + # greps the header the LVQ check would have tested for existence. + set(SVS_DYNAMIC_INDEX_HEADER "svs/index/vamana/dynamic_index.h") + set(SVS_HAS_REPLACE_EXTERNAL_ID 0) + if(EXISTS "${svs_SOURCE_DIR}/include/${SVS_DYNAMIC_INDEX_HEADER}") + file(READ "${svs_SOURCE_DIR}/include/${SVS_DYNAMIC_INDEX_HEADER}" SVS_DYNAMIC_INDEX_SRC) + string(FIND "${SVS_DYNAMIC_INDEX_SRC}" "replace_external_id" SVS_REPLACE_EXTERNAL_ID_POS) + if(NOT SVS_REPLACE_EXTERNAL_ID_POS EQUAL -1) + set(SVS_HAS_REPLACE_EXTERNAL_ID 1) + endif() + unset(SVS_DYNAMIC_INDEX_SRC) + endif() + + if(SVS_HAS_REPLACE_EXTERNAL_ID) + message("SVS replace_external_id found - SVS relabeling enabled") + add_compile_definitions(VectorSimilarity PUBLIC "HAVE_SVS_REPLACE_EXTERNAL_ID=1") + else() + message("SVS replace_external_id not found - SVS relabeling reports unsupported") + add_compile_definitions(VectorSimilarity PUBLIC "HAVE_SVS_REPLACE_EXTERNAL_ID=0") + endif() else() message(STATUS "SVS support disabled") add_compile_definitions("HAVE_SVS=0") diff --git a/deps/ScalableVectorSearch b/deps/ScalableVectorSearch index 7786d43b9..a7e3494fa 160000 --- a/deps/ScalableVectorSearch +++ b/deps/ScalableVectorSearch @@ -1 +1 @@ -Subproject commit 7786d43b98ac9769ad7668d0d4896143cfb2f167 +Subproject commit a7e3494faab9454f577b87ac934a96c78c45dd7b diff --git a/src/VecSim/algorithms/svs/svs.h b/src/VecSim/algorithms/svs/svs.h index 1e124519f..d6c4566cb 100644 --- a/src/VecSim/algorithms/svs/svs.h +++ b/src/VecSim/algorithms/svs/svs.h @@ -549,6 +549,29 @@ class SVSIndex : public VecSimIndexAbstract, fl return deleteVectorsImpl(labels, n); } +#if HAVE_SVS_REPLACE_EXTERNAL_ID + // Only declared when the SVS this was built against offers `replace_external_id`. The + // pre-built SVS releases predate it, so where it is missing this override is left out and the + // interface default reports `VecSimRelabel_Unsupported` - which tells a caller to fall back to + // delete + insert rather than read it as a no-op. + VecSimRelabelCode relabelVector(labelType old_label, labelType new_label) override { + if (old_label == new_label) { + return VecSimRelabel_SameLabel; + } + // `isLabelExists` also covers the index that never held a vector, where `impl_` has not + // been created yet and so trivially holds nothing. + if (!isLabelExists(old_label)) { + return VecSimRelabel_OldLabelMissing; + } + if (isLabelExists(new_label)) { + return VecSimRelabel_NewLabelTaken; + } + + impl_->replace_external_id(old_label, new_label); + return VecSimRelabel_OK; + } +#endif // HAVE_SVS_REPLACE_EXTERNAL_ID + bool isLabelExists(labelType label) const override { return impl_ ? impl_->has_id(label) : false; } diff --git a/src/VecSim/algorithms/svs/svs_tiered.h b/src/VecSim/algorithms/svs/svs_tiered.h index edb950121..1aef30749 100644 --- a/src/VecSim/algorithms/svs/svs_tiered.h +++ b/src/VecSim/algorithms/svs/svs_tiered.h @@ -964,6 +964,61 @@ class TieredSVSIndex : public VecSimTieredIndex { } return ret; } + /** + * Move `old_label` onto `new_label`, leaving the vector where it is in whichever tier holds + * it. `new_label` must be unused in both tiers, not just the one holding `old_label`: a + * multi-value label routinely has copies in each, and a target taken in either would collide + * once the buffer drains. + * + * Reports `Unsupported` when the backend holds the label and cannot move it, which is the + * case when built against an SVS without `replace_external_id`. All-or-nothing: on any code + * other than `VecSimRelabel_OK` both tiers are untouched. + * + * `updateJobMutex` is taken first, in the order `updateSVSIndex` takes its own locks. Holding + * it is what makes this correct rather than merely serialised: an update job snapshots the + * buffer's labels *by value* and afterwards reconciles only id swaps and deletions, so a + * rename landing inside its window would be invisible to it and the vector would reach the + * backend under the old label. + */ + VecSimRelabelCode relabelVector(labelType old_label, labelType new_label) override { + if (old_label == new_label) { + return VecSimRelabel_SameLabel; + } + auto *svs_index = GetSVSIndex(); + + std::lock_guard update_lock(this->updateJobMutex); + std::unique_lock flat_lock(this->flatIndexGuard); + std::unique_lock main_lock(this->mainIndexGuard); + + const bool in_flat = this->frontendIndex->isLabelExists(old_label); + const bool in_backend = svs_index->isLabelExists(old_label); + if (!in_flat && !in_backend) { + return VecSimRelabel_OldLabelMissing; + } + if (this->frontendIndex->isLabelExists(new_label) || svs_index->isLabelExists(new_label)) { + return VecSimRelabel_NewLabelTaken; + } + + // The backend goes first because it is the tier that can refuse; refusing after the + // buffer had already moved would leave the label half applied. + if (in_backend) { + const VecSimRelabelCode backend_ret = + this->backendIndex->relabelVector(old_label, new_label); + if (backend_ret != VecSimRelabel_OK) { + return backend_ret; + } + } + if (in_flat) { + const VecSimRelabelCode flat_ret = + this->frontendIndex->relabelVector(old_label, new_label); +#ifdef BUILD_TESTS + assert(flat_ret == VecSimRelabel_OK && "the buffer just reported holding this label"); +#endif + UNUSED(flat_ret); + } + return VecSimRelabel_OK; + } + size_t getNumMarkedDeleted() const override { return this->GetSVSIndex()->getNumMarkedDeleted(); } diff --git a/src/python_bindings/bindings.cpp b/src/python_bindings/bindings.cpp index b68c14653..e1b2c6a81 100644 --- a/src/python_bindings/bindings.cpp +++ b/src/python_bindings/bindings.cpp @@ -106,12 +106,11 @@ class PyVecSimIndex { template inline py::object rawVectorsAsNumpy(labelType label, size_t dim) { std::vector> vectors; - if (index->basicInfo().algo == VecSimAlgo_BF) { - dynamic_cast *>(this->index.get()) - ->getDataByLabel(label, vectors); + if (auto *tiered = + dynamic_cast *>(this->index.get())) { + tiered->getDataByLabel(label, vectors); } else { - // index is HNSW - dynamic_cast *>(this->index.get()) + dynamic_cast *>(this->index.get()) ->getDataByLabel(label, vectors); } size_t n_vectors = vectors.size(); @@ -216,6 +215,11 @@ class PyVecSimIndex { void runGC() { VecSimTieredIndex_GC(index.get()); } + VecSimRelabelCode relabelVector(labelType old_label, labelType new_label) { + py::gil_scoped_release py_gil; + return VecSimIndex_RelabelVector(index.get(), old_label, new_label); + } + py::object getVector(labelType label) { VecSimIndexBasicInfo info = index->basicInfo(); size_t dim = info.dim; @@ -713,6 +717,14 @@ PYBIND11_MODULE(VecSim, m) { .def_readwrite("initialCapacity", &BFParams::initialCapacity) .def_readwrite("blockSize", &BFParams::blockSize); + py::enum_(m, "VecSimRelabelCode") + .value("VecSimRelabel_OK", VecSimRelabel_OK) + .value("VecSimRelabel_OldLabelMissing", VecSimRelabel_OldLabelMissing) + .value("VecSimRelabel_NewLabelTaken", VecSimRelabel_NewLabelTaken) + .value("VecSimRelabel_SameLabel", VecSimRelabel_SameLabel) + .value("VecSimRelabel_Unsupported", VecSimRelabel_Unsupported) + .export_values(); + py::enum_(m, "VecSimSvsQuantBits") .value("VecSimSvsQuant_NONE", VecSimSvsQuant_NONE) .value("VecSimSvsQuant_Scalar", VecSimSvsQuant_Scalar) @@ -799,6 +811,8 @@ PYBIND11_MODULE(VecSim, m) { .def("create_batch_iterator", &PyVecSimIndex::createBatchIterator, py::arg("query_blob"), py::arg("query_param") = nullptr) .def("get_vector", &PyVecSimIndex::getVector) + .def("relabel_vector", &PyVecSimIndex::relabelVector, py::arg("old_label"), + py::arg("new_label")) .def("run_gc", &PyVecSimIndex::runGC); py::class_(m, "HNSWIndex") diff --git a/tests/flow/test_bruteforce.py b/tests/flow/test_bruteforce.py index becbc7481..36bc330ee 100644 --- a/tests/flow/test_bruteforce.py +++ b/tests/flow/test_bruteforce.py @@ -747,3 +747,61 @@ def test_range_query(self): def test_multi_value(self): self.multi_value(create_uint8_vectors) + + +def test_relabel_vector(test_logger): + dim = 16 + num_elements = 100 + index = create_flat_index(dim, VecSimMetric_L2, VecSimType_FLOAT32) + + data = np.float32(np.random.random((num_elements, dim))) + for label, vector in enumerate(data): + index.add_vector(vector, label) + + old_label, new_label = 7, num_elements + 500 + assert index.relabel_vector(old_label, new_label) == VecSimRelabel_OK + + # The label moved without the data moving, and the index neither grew nor shrank. + assert index.index_size() == num_elements + assert_allclose(index.get_vector(new_label)[0], data[old_label], rtol=1e-6) + assert index.get_vector(old_label).shape == (0, dim) + + # Still searchable, and answering under the new label. + labels, distances = index.knn_query(data[old_label], 1) + assert labels[0][0] == new_label + assert distances[0][0] < 1e-6 + + # Each rejection is reported distinctly, and none of them modifies the index. + assert index.relabel_vector(num_elements + 1, 0) == VecSimRelabel_OldLabelMissing + assert index.relabel_vector(0, 1) == VecSimRelabel_NewLabelTaken + assert index.relabel_vector(0, 0) == VecSimRelabel_SameLabel + assert index.index_size() == num_elements + test_logger.info("flat index relabel_vector moved the label, keeping the vector data") + + +def test_relabel_vector_multi(test_logger): + dim = 16 + num_labels = 20 + per_label = 3 + index = create_flat_index(dim, VecSimMetric_L2, VecSimType_FLOAT32, is_multi=True) + + data = np.float32(np.random.random((num_labels, per_label, dim))) + for label in range(num_labels): + for vector in data[label]: + index.add_vector(vector, label) + + old_label, new_label = 7, num_labels + 500 + assert index.relabel_vector(old_label, new_label) == VecSimRelabel_OK + + # Every vector under the label moves as a unit, keeping its data and insertion order. + assert index.index_size() == num_labels * per_label + assert_allclose(index.get_vector(new_label), data[old_label], rtol=1e-6) + assert index.get_vector(old_label).shape == (0, dim) + + # Moving onto an occupied label must be rejected: accepting it would silently merge two + # labels' vectors, which is the failure mode unique to a multi index. + assert index.relabel_vector(new_label, 0) == VecSimRelabel_NewLabelTaken + assert index.get_vector(new_label).shape == (per_label, dim) + assert index.get_vector(0).shape == (per_label, dim) + assert index.index_size() == num_labels * per_label + test_logger.info("flat multi relabel_vector moved every vector under the label") diff --git a/tests/flow/test_hnsw.py b/tests/flow/test_hnsw.py index 9094ff1be..ee00e71d4 100644 --- a/tests/flow/test_hnsw.py +++ b/tests/flow/test_hnsw.py @@ -1156,3 +1156,64 @@ def test_range_query(self, test_logger): def test_multi_value(self, test_logger): self.multi_value(create_uint8_vectors, test_logger) + + +def test_relabel_vector(test_logger): + dim = 16 + num_elements = 100 + index = create_hnsw_index(dim, num_elements, VecSimMetric_L2, VecSimType_FLOAT32) + + data = np.float32(np.random.random((num_elements, dim))) + for label, vector in enumerate(data): + index.add_vector(vector, label) + + old_label, new_label = 7, num_elements + 500 + assert index.relabel_vector(old_label, new_label) == VecSimRelabel_OK + + # A relabel is bookkeeping only: the graph is untouched, so the vector keeps its data and its + # place in the index rather than being reinserted. + assert index.index_size() == num_elements + assert_allclose(index.get_vector(new_label)[0], data[old_label], rtol=1e-6) + assert index.get_vector(old_label).shape == (0, dim) + assert index.check_integrity() + + labels, distances = index.knn_query(data[old_label], 1) + assert labels[0][0] == new_label + assert distances[0][0] < 1e-6 + + # Each rejection is reported distinctly, and none of them modifies the index. + assert index.relabel_vector(num_elements + 1, 0) == VecSimRelabel_OldLabelMissing + assert index.relabel_vector(0, 1) == VecSimRelabel_NewLabelTaken + assert index.relabel_vector(0, 0) == VecSimRelabel_SameLabel + assert index.index_size() == num_elements + assert index.check_integrity() + test_logger.info("HNSW relabel_vector moved the label, leaving the graph intact") + + +def test_relabel_vector_multi(test_logger): + dim = 16 + num_labels = 20 + per_label = 3 + index = create_hnsw_index(dim, num_labels * per_label, VecSimMetric_L2, VecSimType_FLOAT32, + is_multi=True) + + data = np.float32(np.random.random((num_labels, per_label, dim))) + for label in range(num_labels): + for vector in data[label]: + index.add_vector(vector, label) + + old_label, new_label = 7, num_labels + 500 + assert index.relabel_vector(old_label, new_label) == VecSimRelabel_OK + + # All the label's internal ids are re-pointed together, and the graph is left alone. + assert index.index_size() == num_labels * per_label + assert_allclose(index.get_vector(new_label), data[old_label], rtol=1e-6) + assert index.get_vector(old_label).shape == (0, dim) + assert index.check_integrity() + + # Accepting a move onto an occupied label would merge two labels' vectors. + assert index.relabel_vector(new_label, 0) == VecSimRelabel_NewLabelTaken + assert index.get_vector(new_label).shape == (per_label, dim) + assert index.get_vector(0).shape == (per_label, dim) + assert index.check_integrity() + test_logger.info("HNSW multi relabel_vector moved every vector under the label") diff --git a/tests/flow/test_hnsw_tiered.py b/tests/flow/test_hnsw_tiered.py index f6f692adb..cf84c96ab 100644 --- a/tests/flow/test_hnsw_tiered.py +++ b/tests/flow/test_hnsw_tiered.py @@ -5,6 +5,7 @@ # (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the # GNU Affero General Public License v3 (AGPLv3). import time +import pytest from common import * @@ -621,3 +622,152 @@ def test_multi_range_query(test_logger): # Expect zero results for radius==0 tiered_labels, tiered_distances = index.range_query(query_data, radius=0) assert len(tiered_labels[0]) == 0 + + +def test_relabel_vector(test_logger): + dim = 16 + num_elements = 1000 + hnsw_params = create_hnsw_params(dim, num_elements, VecSimMetric_L2, VecSimType_FLOAT32) + # A flat buffer large enough to hold everything, so the relabel below has a real chance of + # landing while the vector is still buffered with a pending ingest job. + index = Tiered_HNSWIndex(hnsw_params, create_tiered_hnsw_params(), num_elements) + + data = np.float32(np.random.random((num_elements, dim))) + for label, vector in enumerate(data): + index.add_vector(vector, label) + + # Relabel one early and one late label. The workers ingest in insertion order, so by now the + # early one is most likely already in HNSW while the late one is most likely still buffered + # with a pending ingest job - between them the two tiers both get covered. The buffered case is + # the delicate one: a job left holding the old label would either ingest the vector under it or + # throw out of the worker thread, and neither would survive the assertions below. + buffered = index.get_curr_bf_size() + test_logger.info(f"relabeling with {buffered} of {num_elements} vectors still buffered") + moved = {7: num_elements + 500, num_elements - 1: num_elements + 501} + for old_label, new_label in moved.items(): + assert index.relabel_vector(old_label, new_label) == VecSimRelabel_OK + + index.wait_for_index() + + # Once ingestion has drained, the vector sits in HNSW under the new label and under no other. + assert index.index_size() == num_elements + assert index.hnsw_label_count() == num_elements + for old_label, new_label in moved.items(): + assert_allclose(index.get_vector(new_label)[0], data[old_label], rtol=1e-6) + assert index.get_vector(old_label).shape == (0, dim) + + labels, distances = index.knn_query(data[old_label], 1) + assert labels[0][0] == new_label + assert distances[0][0] < 1e-6 + + # Each rejection is reported distinctly, and none of them modifies the index. + assert index.relabel_vector(num_elements + 1, 0) == VecSimRelabel_OldLabelMissing + assert index.relabel_vector(0, 1) == VecSimRelabel_NewLabelTaken + assert index.relabel_vector(0, 0) == VecSimRelabel_SameLabel + assert index.index_size() == num_elements + test_logger.info("tiered relabel_vector moved the label across both tiers") + + +# The relabel counterpart to `test_parallel_insert_search`: the operation running on one thread +# while queries run on another, with the relabels aimed at labels that are still being ingested so +# a label is briefly held by both tiers. +# +# Unlike the insert test there is no approximate assertion to fall back on. Inserting concurrently +# makes a query legitimately miss vectors, so that test can only check recall did not regress; a +# relabel adds and removes nothing, so the invariant here is exact: a reply must never list one +# label twice. `merge_result_lists` collapses a vector both tiers report by matching labels, and a +# relabel inside a query's window moves that key, so the two copies survive the merge as one +# vector under two labels. +# +# Skipped on this branch, not because the assertion is unsound -- a duplicate label in a reply is +# always wrong -- but because the fix is on another branch (VecSim #1047, MOD-18494) and the +# failure is intermittent, so leaving it live would redden this PR's CI for a defect it did not +# introduce. Remove the marker once that lands. +@pytest.mark.skip(reason="needs the cross-tier relabel fix from #1047 / MOD-18494") +def test_relabel_vector_during_query(test_logger): + import threading + + dim = 16 + num_elements = 20000 + k = 10 + hnsw_params = create_hnsw_params(dim, num_elements, VecSimMetric_L2, VecSimType_FLOAT32) + # A flat buffer big enough to hold everything, so ingestion is still draining while the two + # threads run and the relabelled labels really are in both tiers for a while. + index = Tiered_HNSWIndex(hnsw_params, create_tiered_hnsw_params(), num_elements) + + data = np.float32(np.random.random((num_elements, dim))) + for label, vector in enumerate(data): + index.add_vector(vector, label) + + offset = num_elements + 1000 + query_data = np.float32(np.random.random((200, dim))) + duplicates = [] + relabel_failures = [] + + def relabel_labels(): + for label in range(num_elements): + code = index.relabel_vector(label, label + offset) + if code != VecSimRelabel_OK: + relabel_failures.append((label, code)) + break + + def run_queries(): + # Keep querying for as long as the relabel thread is working, so the windows overlap + # many times rather than once. + while relabel_thread.is_alive(): + labels, _ = index.knn_query(query_data, k) + for row in labels: + if len(set(row)) != len(row): + duplicates.append(row.tolist()) + + relabel_thread = threading.Thread(target=relabel_labels) + query_thread = threading.Thread(target=run_queries) + relabel_thread.start() + query_thread.start() + for t in (relabel_thread, query_thread): + t.join() + + assert not relabel_failures, f"relabel refused a live label: {relabel_failures[:3]}" + assert not duplicates, f"one vector reported under two labels: {duplicates[:3]}" + + # The settled state, which holds whatever the interleaving was: every label moved, nothing + # gained or lost. + index.wait_for_index() + assert index.index_size() == num_elements + assert index.hnsw_label_count() == num_elements + for label in (0, num_elements // 2, num_elements - 1): + assert index.get_vector(label).shape == (0, dim) + assert_allclose(index.get_vector(label + offset)[0], data[label], rtol=1e-6) + + test_logger.info("tiered relabel_vector kept queries duplicate-free") + + +def test_relabel_vector_multi(test_logger): + dim = 16 + num_labels = 200 + per_label = 5 + hnsw_params = create_hnsw_params(dim, num_labels * per_label, VecSimMetric_L2, + VecSimType_FLOAT32, is_multi=True) + index = Tiered_HNSWIndex(hnsw_params, create_tiered_hnsw_params(), num_labels * per_label) + + data = np.float32(np.random.random((num_labels, per_label, dim))) + for label in range(num_labels): + for vector in data[label]: + index.add_vector(vector, label) + + # In a multi index a label can hold several pending ingest jobs at once, so a late label + # exercises re-keying all of them together while an early one is most likely already in HNSW. + buffered = index.get_curr_bf_size() + test_logger.info(f"relabeling with {buffered} of {num_labels * per_label} vectors buffered") + moved = {7: num_labels + 500, num_labels - 1: num_labels + 501} + for old_label, new_label in moved.items(): + assert index.relabel_vector(old_label, new_label) == VecSimRelabel_OK + + index.wait_for_index() + + assert index.index_size() == num_labels * per_label + assert index.hnsw_label_count() == num_labels + for old_label, new_label in moved.items(): + assert index.get_vector(new_label).shape == (per_label, dim) + assert index.get_vector(old_label).shape == (0, dim) + test_logger.info("tiered multi relabel_vector moved every vector under the label") diff --git a/tests/flow/test_svs.py b/tests/flow/test_svs.py index 9b6d96d74..830d23337 100644 --- a/tests/flow/test_svs.py +++ b/tests/flow/test_svs.py @@ -11,6 +11,7 @@ import time from VecSim import * from common import * +import pytest import hnswlib def create_svs_index(dim, num_elements, data_type, metric = VecSimMetric_L2, @@ -456,3 +457,157 @@ def test_multi_range_query(test_logger): # Expect zero results for radius==0 svs_labels, svs_distances = index.range_query(query_data, radius=0) assert len(svs_labels[0]) == 0 + +# The pre-built SVS releases predate `replace_external_id`, so a build that picked one up reports +# relabeling as unsupported rather than implementing it. There is no python-visible build flag for +# that, so ask the index itself on a throwaway label - a real implementation cannot answer +# Unsupported, so this cannot mask a broken one. +def svs_relabel_supported(index, label): + return index.relabel_vector(label, label) != VecSimRelabel_Unsupported + +def test_get_vector(test_logger): + dim = 16 + num_elements = 100 + index = create_svs_index(dim, num_elements, VecSimType_FLOAT32, VecSimMetric_L2) + + data = np.float32(np.random.random((num_elements, dim))) + for label, vector in enumerate(data): + index.add_vector(vector, label) + + # L2 applies no insert-time preprocessing, so the stored vector is the one that was handed to + # add_vector. Under cosine it would come back normalized instead. + for label in [0, num_elements // 2, num_elements - 1]: + stored = index.get_vector(label) + assert stored.shape == (1, dim) + assert_allclose(stored[0], data[label], rtol=1e-6) + + # An absent label appends nothing, so the caller gets zero rows rather than an error. + assert index.get_vector(num_elements + 1).shape == (0, dim) + test_logger.info("SVS get_vector returned the stored vectors") + + +def test_get_vector_multi(test_logger): + dim = 16 + num_labels = 20 + per_label = 3 + index = create_svs_index(dim, num_labels * per_label, VecSimType_FLOAT32, VecSimMetric_L2, + is_multi=True) + + data = np.float32(np.random.random((num_labels, per_label, dim))) + for label in range(num_labels): + for vector in data[label]: + index.add_vector(vector, label) + + # Every vector grouped under the label comes back, in insertion order. + for label in [0, num_labels - 1]: + stored = index.get_vector(label) + assert stored.shape == (per_label, dim) + assert_allclose(stored, data[label], rtol=1e-6) + test_logger.info("SVS get_vector returned every vector under a multi label") + + +def test_get_vector_quantized(test_logger): + dim = 16 + num_elements = 100 + svs_params = create_svs_params(dim, num_elements, VecSimType_FLOAT32, VecSimMetric_L2, + quantBits=VecSimSvsQuant_Scalar) + index = SVSIndex(svs_params) + + data = np.float32(np.random.random((num_elements, dim))) + for label, vector in enumerate(data): + index.add_vector(vector, label) + + # A quantized index cannot report its stored vectors as values, and nothing here dequantizes. + # It therefore reports no rows - "cannot tell" - rather than reinterpreting the compressed + # form and its metadata as vector elements. + assert index.get_vector(0).shape == (0, dim) + test_logger.info("SVS get_vector reported nothing for a quantized index") + + +def test_relabel_vector(test_logger): + dim = 16 + num_elements = 100 + index = create_svs_index(dim, num_elements, VecSimType_FLOAT32, VecSimMetric_L2) + + data = np.float32(np.random.random((num_elements, dim))) + for label, vector in enumerate(data): + index.add_vector(vector, label) + + if not svs_relabel_supported(index, 0): + pytest.skip("this SVS build has no replace_external_id") + + old_label, new_label = 7, num_elements + 500 + assert index.relabel_vector(old_label, new_label) == VecSimRelabel_OK + + # The vector moved label without moving data: same contents, and the index neither grew nor + # shrank. + assert index.index_size() == num_elements + assert_allclose(index.get_vector(new_label)[0], data[old_label], rtol=1e-6) + assert index.get_vector(old_label).shape == (0, dim) + + # A relabeled vector is still searchable, and answers under the new label. + labels, distances = index.knn_query(data[old_label], 1) + assert labels[0][0] == new_label + # Querying with the stored vector itself, so the distance to it is zero. + assert distances[0][0] < 1e-6 + test_logger.info("SVS relabel_vector moved the label, keeping the vector data") + + +def test_relabel_vector_rejects(test_logger): + dim = 16 + num_elements = 10 + index = create_svs_index(dim, num_elements, VecSimType_FLOAT32, VecSimMetric_L2) + + if not svs_relabel_supported(index, 0): + pytest.skip("this SVS build has no replace_external_id") + + # An index that never held a vector has no SVS impl yet - still a clean rejection. + assert index.relabel_vector(1, 2) == VecSimRelabel_OldLabelMissing + + data = np.float32(np.random.random((num_elements, dim))) + for label, vector in enumerate(data): + index.add_vector(vector, label) + + # Each rejection is reported distinctly, so a caller can tell a conflict it may resolve from a + # label that simply is not there. + assert index.relabel_vector(num_elements + 1, 0) == VecSimRelabel_OldLabelMissing + assert index.relabel_vector(0, 1) == VecSimRelabel_NewLabelTaken + assert index.relabel_vector(0, 0) == VecSimRelabel_SameLabel + + # None of the rejections touched the index. + assert index.index_size() == num_elements + for label in range(num_elements): + assert_allclose(index.get_vector(label)[0], data[label], rtol=1e-6) + test_logger.info("SVS relabel_vector reported each rejection without modifying the index") + + +def test_relabel_vector_multi(test_logger): + dim = 16 + num_labels = 20 + per_label = 3 + index = create_svs_index(dim, num_labels * per_label, VecSimType_FLOAT32, VecSimMetric_L2, + is_multi=True) + + data = np.float32(np.random.random((num_labels, per_label, dim))) + for label in range(num_labels): + for vector in data[label]: + index.add_vector(vector, label) + + if not svs_relabel_supported(index, 0): + pytest.skip("this SVS build has no replace_external_id") + + old_label, new_label = 7, num_labels + 500 + assert index.relabel_vector(old_label, new_label) == VecSimRelabel_OK + + # SVS renames the label itself, so every vector grouped under it stays grouped - and keeps its + # own internal id, which is why the data comes back unchanged and in order. + assert index.index_size() == num_labels * per_label + assert_allclose(index.get_vector(new_label), data[old_label], rtol=1e-6) + assert index.get_vector(old_label).shape == (0, dim) + + # Accepting a move onto an occupied label would merge two labels' vectors. + assert index.relabel_vector(new_label, 0) == VecSimRelabel_NewLabelTaken + assert index.get_vector(new_label).shape == (per_label, dim) + assert index.get_vector(0).shape == (per_label, dim) + assert index.index_size() == num_labels * per_label + test_logger.info("SVS multi relabel_vector moved every vector under the label") diff --git a/tests/flow/test_svs_tiered.py b/tests/flow/test_svs_tiered.py index 746bab43f..8531984ff 100644 --- a/tests/flow/test_svs_tiered.py +++ b/tests/flow/test_svs_tiered.py @@ -278,6 +278,71 @@ def test_create_fp16(test_logger): test_logger.info("Test create FLOAT16 tiered svs index") create_tiered_index(test_logger, is_multi=False, data_type=VecSimType_FLOAT16) +def relabel_vector(test_logger, is_multi: bool, num_per_label=1, data_type=VecSimType_FLOAT32, + quantBits=VecSimSvsQuant_NONE): + data_size = 2000 + # Small thresholds so ingestion is under way by the time the relabels run, leaving some labels + # in the backend and some still buffered. The tier has to move the label in either state. + indices_ctx = IndexCtx(dim=32, data_size=data_size, is_multi=is_multi, num_per_label=num_per_label, + flat_buffer_size=data_size, graph_degree=32, data_type=data_type, + quantBits=quantBits, trainingThreshold=1024, updateThreshold=128) + index = indices_ctx.tiered_index + num_labels = indices_ctx.num_labels + + indices_ctx.populate_index(index) + + # Moved into a range past the populated one, so a move can never land on a label that is + # still waiting to be ingested. + offset = num_labels + 500 + moved = {7: 7 + offset, num_labels - 1: num_labels - 1 + offset} + buffered = index.get_curr_bf_size() + test_logger.info(f"relabeling with {buffered} of {num_labels} labels still buffered") + for old_label, new_label in moved.items(): + assert index.relabel_vector(old_label, new_label) == VecSimRelabel_OK + + # Rejections, from a tier: the target has to be free in both of them. + assert index.relabel_vector(7, 7) == VecSimRelabel_SameLabel + assert index.relabel_vector(num_labels + 10000, offset) == VecSimRelabel_OldLabelMissing + assert index.relabel_vector(8, 9) == VecSimRelabel_NewLabelTaken + + index.wait_for_index() + + # Nothing gained or lost a label, which is what fails if an in-flight update job ingested a + # vector under the label it snapshotted rather than the one it now has. + assert index.index_size() == num_labels * num_per_label + assert index.svs_label_count() == num_labels + + for old_label, new_label in moved.items(): + assert index.get_vector(old_label).shape == (0, indices_ctx.dim) + if quantBits == VecSimSvsQuant_NONE: + # A compressed backend reports no stored values, so only the unquantized + # configurations can be checked by reading the vector back. + assert index.get_vector(new_label).shape == (num_per_label, indices_ctx.dim) + + # Searchable under the new label either way. + labels, _ = index.knn_query(np.array([indices_ctx.data[old_label]]), 1) + assert labels[0][0] == new_label + + test_logger.info("tiered svs relabel_vector moved the label in both tiers") + + +def test_relabel_vector(test_logger): + test_logger.info("Start tiered svs relabel test") + relabel_vector(test_logger, is_multi=False) + +def test_relabel_vector_q8(test_logger): + test_logger.info("Start tiered svs relabel test, 8-bit quantized") + relabel_vector(test_logger, is_multi=False, quantBits=VecSimSvsQuant_8) + +def test_relabel_vector_leanvec_8x8(test_logger): + test_logger.info("Start tiered svs relabel test, LeanVec 8x8") + relabel_vector(test_logger, is_multi=False, quantBits=VecSimSvsQuant_8x8_LeanVec) + +def test_relabel_vector_fp16(test_logger): + test_logger.info("Start tiered svs relabel test, FLOAT16") + relabel_vector(test_logger, is_multi=False, data_type=VecSimType_FLOAT16) + + def test_search_insert(test_logger): test_logger.info("Start insert & search test") search_insert(test_logger, is_multi=False) diff --git a/tests/unit/test_hnsw_tiered.cpp b/tests/unit/test_hnsw_tiered.cpp index f9e2b4d2d..ed0f900e9 100644 --- a/tests/unit/test_hnsw_tiered.cpp +++ b/tests/unit/test_hnsw_tiered.cpp @@ -531,6 +531,94 @@ TYPED_TEST(HNSWTieredIndexTestBasic, getDataByLabelSpansBothTiers) { ASSERT_TRUE(stored.empty()); } +// The buffered case is the one that regressed before: reading only the backend reports nothing +// for a vector written recently enough to still be queued for ingestion, which is exactly when a +// document is most likely to be written again. Determinism comes from never running the job. +TYPED_TEST(HNSWTieredIndexTestBasic, getDataByLabelWhileStillBuffered) { + size_t dim = 4; + HNSWParams params = { + .type = TypeParam::get_index_type(), .dim = dim, .metric = VecSimMetric_L2, .multi = false}; + VecSimParams hnsw_params = CreateParams(params); + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = this->CreateTieredHNSWIndex(hnsw_params, mock_thread_pool); + + TEST_DATA_T vector[dim]; + GenerateVector(vector, dim, 1); + VecSimIndex_AddVector(tiered_index, vector, 0); + // Nothing ingested: the vector exists only in the buffer, with its job still pending. + ASSERT_EQ(this->GetFlatIndex(tiered_index)->indexSize(), 1); + ASSERT_EQ(this->CastToHNSW(tiered_index)->indexSize(), 0); + + std::vector> stored; + tiered_index->getDataByLabel(0, stored); + ASSERT_EQ(stored.size(), 1) << "the buffered tier was not read"; + ASSERT_NO_FATAL_FAILURE(CompareVectors(stored[0].data(), vector, dim)); + + // Draining moves it to the backend, which must report it exactly once -- not once per tier. + mock_thread_pool.thread_iteration(); + ASSERT_EQ(this->GetFlatIndex(tiered_index)->indexSize(), 0); + ASSERT_EQ(this->CastToHNSW(tiered_index)->indexSize(), 1); + stored.clear(); + tiered_index->getDataByLabel(0, stored); + ASSERT_EQ(stored.size(), 1); + ASSERT_NO_FATAL_FAILURE(CompareVectors(stored[0].data(), vector, dim)); +} + +// An ingest job inserts into the backend before dropping the buffered copy, so a label is briefly +// held by both tiers. What that means for `getDataByLabel` differs by index kind, and both halves +// are asserted here because each is a deliberate consequence of one condition in the dispatch: +// a single-value label short-circuits on the buffer hit and never reads the backend, while a +// multi-value label always reads it and so reports the caught vector twice. +TYPED_TEST(HNSWTieredIndexTestBasic, getDataByLabelInTheIngestWindow) { + size_t dim = 4; + TEST_DATA_T vector[dim]; + GenerateVector(vector, dim, 1); + + { + HNSWParams single = {.type = TypeParam::get_index_type(), + .dim = dim, + .metric = VecSimMetric_L2, + .multi = false}; + VecSimParams hnsw_params = CreateParams(single); + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = this->CreateTieredHNSWIndex(hnsw_params, mock_thread_pool); + auto *hnsw_index = this->CastToHNSW(tiered_index); + + // Build the window directly rather than racing a worker into it. + VecSimIndex_AddVector(tiered_index, vector, 0); + hnsw_index->addVector(vector, 0); + ASSERT_EQ(this->GetFlatIndex(tiered_index)->indexSize(), 1); + ASSERT_EQ(hnsw_index->indexSize(), 1); + + std::vector> stored; + tiered_index->getDataByLabel(0, stored); + ASSERT_EQ(stored.size(), 1) << "a single-value label must not report its two tier copies"; + ASSERT_NO_FATAL_FAILURE(CompareVectors(stored[0].data(), vector, dim)); + } + + { + HNSWParams multi = {.type = TypeParam::get_index_type(), + .dim = dim, + .metric = VecSimMetric_L2, + .multi = true}; + VecSimParams hnsw_params = CreateParams(multi); + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = this->CreateTieredHNSWIndex(hnsw_params, mock_thread_pool); + auto *hnsw_index = this->CastToHNSW(tiered_index); + + VecSimIndex_AddVector(tiered_index, vector, 0); + hnsw_index->addVector(vector, 0); + + std::vector> stored; + tiered_index->getDataByLabel(0, stored); + // Documented on the declaration: a multi-value label always reads both tiers, so a vector + // caught mid-ingest is reported by each. Pinned here so the caveat cannot change unnoticed. + ASSERT_EQ(stored.size(), 2); + ASSERT_NO_FATAL_FAILURE(CompareVectors(stored[0].data(), vector, dim)); + ASSERT_NO_FATAL_FAILURE(CompareVectors(stored[1].data(), vector, dim)); + } +} + TYPED_TEST(HNSWTieredIndexTestBasic, insertJobAsyncMulti) { // Create TieredHNSW index instance with a mock queue. size_t dim = 4; diff --git a/tests/unit/test_svs.cpp b/tests/unit/test_svs.cpp index 144e1562f..59ee2798b 100644 --- a/tests/unit/test_svs.cpp +++ b/tests/unit/test_svs.cpp @@ -3539,11 +3539,115 @@ TEST(SVSTest, ThreadPoolLazyInit) { VecSimSVSThreadPoolImpl::instance()->resetForTest(); } -// SVS delegates label management to the external library, so it does not implement relabelVector -// and inherits the VecSimIndexInterface default that reports "unsupported". The source label exists -// and the target is free here, so a 0 return can only come from that default - which is the -// contract callers must handle, and the reason the interface provides a default instead of a pure -// virtual. +#if HAVE_SVS_REPLACE_EXTERNAL_ID + +TYPED_TEST(SVSTest, relabelVector) { + size_t dim = 4; + SVSParams params = {.dim = dim, .metric = VecSimMetric_L2}; + VecSimIndex *index = this->CreateNewIndex(params); + ASSERT_INDEX(index); + + GenerateAndAddVector(index, dim, 1, 1); + GenerateAndAddVector(index, dim, 2, 2); + ASSERT_EQ(VecSimIndex_IndexSize(index), 2); + + TEST_DATA_T query[dim]; + GenerateVector(query, dim, 1); + // The distance to the label's own vector, captured before the move. Asserting it is unchanged + // afterwards says the move did not disturb the stored vector without assuming what that + // distance is -- a quantized index does not answer 0 for a vector's own query. + const double before = VecSimIndex_GetDistanceFrom_Unsafe(index, 1, query); + ASSERT_FALSE(std::isnan(before)); + + ASSERT_EQ(VecSimIndex_RelabelVector(index, 1, 100), VecSimRelabel_OK); + + // Nothing was added or removed, and the vector answers to the new label only. + ASSERT_EQ(VecSimIndex_IndexSize(index), 2); + ASSERT_EQ(index->indexLabelCount(), 2); + ASSERT_EQ(VecSimIndex_GetDistanceFrom_Unsafe(index, 100, query), before); + ASSERT_TRUE(std::isnan(VecSimIndex_GetDistanceFrom_Unsafe(index, 1, query))); + + auto verify_res = [&](size_t id, double score, size_t rank) { + ASSERT_EQ(id, 100); + ASSERT_EQ(score, before); + }; + runTopKSearchTest(index, query, 1, verify_res); + + VecSimIndex_Free(index); +} + +// SVS deletes softly -- the entry is marked and only dropped by a later consolidation, so it is +// still occupying an id when this runs. `has_id` excludes it, which is what makes the move report +// the label absent rather than renaming a tombstone: the same contract HNSW states for its own +// marked-deleted elements, and worth pinning here because the two arrive at it by different +// means. +TYPED_TEST(SVSTest, relabelVectorMarkedDeleted) { + size_t dim = 4; + SVSParams params = {.dim = dim, .metric = VecSimMetric_L2}; + VecSimIndex *index = this->CreateNewIndex(params); + ASSERT_INDEX(index); + + GenerateAndAddVector(index, dim, 0, 0); + GenerateAndAddVector(index, dim, 1, 1); + ASSERT_EQ(VecSimIndex_DeleteVector(index, 0), 1); + + auto *svs_index = dynamic_cast(index); + ASSERT_NE(svs_index, nullptr); + // Soft, not gone: the assertions below are about a marked entry, not an absent one. + ASSERT_GT(svs_index->getNumMarkedDeleted(), 0); + + ASSERT_EQ(VecSimIndex_RelabelVector(index, 0, 100), VecSimRelabel_OldLabelMissing); + ASSERT_FALSE(svs_index->isLabelExists(100)) << "a tombstone was renamed"; + ASSERT_EQ(index->indexLabelCount(), 1); + + // A live label in the same index still relabels fine. + ASSERT_EQ(VecSimIndex_RelabelVector(index, 1, 101), VecSimRelabel_OK); + ASSERT_TRUE(svs_index->isLabelExists(101)); + ASSERT_EQ(index->indexLabelCount(), 1); + + // And the deleted label stays free for reuse rather than being half-claimed by the refusal. + GenerateAndAddVector(index, dim, 0, 7); + ASSERT_TRUE(svs_index->isLabelExists(0)); + ASSERT_EQ(index->indexLabelCount(), 2); + + VecSimIndex_Free(index); +} + +TYPED_TEST(SVSTest, relabelVectorRejects) { + size_t dim = 4; + SVSParams params = {.dim = dim, .metric = VecSimMetric_L2}; + VecSimIndex *index = this->CreateNewIndex(params); + ASSERT_INDEX(index); + + // An index that never held a vector has no SVS impl yet - still a clean rejection, not a crash. + ASSERT_EQ(VecSimIndex_RelabelVector(index, 1, 2), VecSimRelabel_OldLabelMissing); + + GenerateAndAddVector(index, dim, 1, 1); + GenerateAndAddVector(index, dim, 2, 2); + + ASSERT_EQ(VecSimIndex_RelabelVector(index, 42, 100), VecSimRelabel_OldLabelMissing); + ASSERT_EQ(VecSimIndex_RelabelVector(index, 1, 2), VecSimRelabel_NewLabelTaken); + ASSERT_EQ(VecSimIndex_RelabelVector(index, 1, 1), VecSimRelabel_SameLabel); + + ASSERT_EQ(VecSimIndex_IndexSize(index), 2); + ASSERT_EQ(index->indexLabelCount(), 2); + for (labelType label : {1, 2}) { + TEST_DATA_T v[dim]; + GenerateVector(v, dim, label); + // Present and answering for its own vector; the value itself depends on the encoding. + ASSERT_FALSE(std::isnan(VecSimIndex_GetDistanceFrom_Unsafe(index, label, v))) + << "label " << label << " was modified"; + } + + VecSimIndex_Free(index); +} + +#else // HAVE_SVS_REPLACE_EXTERNAL_ID + +// Built against an SVS without `replace_external_id`, so SVSIndex leaves relabelVector to the +// interface default. Asserting the code here rather than skipping keeps the contract covered in +// this configuration too: a caller has to be able to tell "this index never relabels" from a +// rejection it could resolve itself. TEST(SVSTest, relabelVectorUnsupported) { size_t dim = 4; SVSParams params = {.type = VecSimType_FLOAT32, .dim = dim, .metric = VecSimMetric_L2}; @@ -3551,16 +3655,18 @@ TEST(SVSTest, relabelVectorUnsupported) { VecSimIndex *index = VecSimIndex_New(&index_params); ASSERT_NE(index, nullptr); - GenerateAndAddVector(index, dim, 1); + GenerateAndAddVector(index, dim, 1, 1); ASSERT_EQ(VecSimIndex_IndexSize(index), 1); + // The label exists and the target is free, so only the unsupported default can produce this. ASSERT_EQ(VecSimIndex_RelabelVector(index, 1, 2), VecSimRelabel_Unsupported); - // The rejected call left the index untouched. ASSERT_EQ(VecSimIndex_IndexSize(index), 1); VecSimIndex_Free(index); } +#endif // HAVE_SVS_REPLACE_EXTERNAL_ID + #else // HAVE_SVS TEST(SVSTest, svs_not_supported) { diff --git a/tests/unit/test_svs_multi.cpp b/tests/unit/test_svs_multi.cpp index 9c75ab8c3..eb87379d9 100644 --- a/tests/unit/test_svs_multi.cpp +++ b/tests/unit/test_svs_multi.cpp @@ -1257,3 +1257,93 @@ TYPED_TEST(SVSMultiTest, rangeQuery) { } #endif // HAVE_SVS + +#if HAVE_SVS_REPLACE_EXTERNAL_ID + +// A multi-value label is where a move that handles only the label's first id still looks correct +// in the single-value tests: every one of its vectors has to answer to the new label, and none to +// the old. Runs across the type set, so it also covers moving a label whose vectors are quantized. +TYPED_TEST(SVSMultiTest, relabelVectorMulti) { + const size_t dim = 4; + const size_t per_label = 5; + + SVSParams params = {.dim = dim, .metric = VecSimMetric_L2}; + VecSimIndex *index = this->CreateNewIndex(params); + ASSERT_INDEX(index); + + // Label 7 gets several vectors, and label 8 is an untouched neighbour with its own, so a move + // that is too broad shows up as well. + for (size_t i = 0; i < per_label; i++) { + GenerateAndAddVector(index, dim, 7, i); + GenerateAndAddVector(index, dim, 8, i + 100); + } + ASSERT_EQ(VecSimIndex_IndexSize(index), per_label * 2); + ASSERT_EQ(index->indexLabelCount(), 2); + + // Captured per vector before the move: asserting these are unchanged afterwards says every + // copy still holds its own data, without assuming what the distances are under quantization. + std::vector before(per_label); + for (size_t i = 0; i < per_label; i++) { + TEST_DATA_T v[dim]; + GenerateVector(v, dim, i); + before[i] = VecSimIndex_GetDistanceFrom_Unsafe(index, 7, v); + ASSERT_FALSE(std::isnan(before[i])); + } + + ASSERT_EQ(VecSimIndex_RelabelVector(index, 7, 70), VecSimRelabel_OK); + + // Every copy moved, and only the label changed. + ASSERT_EQ(VecSimIndex_IndexSize(index), per_label * 2); + ASSERT_EQ(index->indexLabelCount(), 2); + for (size_t i = 0; i < per_label; i++) { + TEST_DATA_T v[dim]; + GenerateVector(v, dim, i); + ASSERT_EQ(VecSimIndex_GetDistanceFrom_Unsafe(index, 70, v), before[i]) + << "copy " << i << " was disturbed by the move"; + } + TEST_DATA_T probe[dim]; + GenerateVector(probe, dim, 0); + ASSERT_TRUE(std::isnan(VecSimIndex_GetDistanceFrom_Unsafe(index, 7, probe))) + << "the old label still answers"; + + // The neighbour kept all of its own copies. + TEST_DATA_T neighbour[dim]; + GenerateVector(neighbour, dim, 100); + ASSERT_FALSE(std::isnan(VecSimIndex_GetDistanceFrom_Unsafe(index, 8, neighbour))); + + // A search for one of the moved vectors reports the new label. + TEST_DATA_T query[dim]; + GenerateVector(query, dim, 0); + auto verify_res = [&](size_t id, double score, size_t rank) { ASSERT_EQ(id, 70); }; + runTopKSearchTest(index, query, 1, verify_res); + + VecSimIndex_Free(index); +} + +// Rejections on a multi index: a target that already holds vectors of its own is taken, however +// many copies either label has. +TYPED_TEST(SVSMultiTest, relabelVectorMultiRejects) { + const size_t dim = 4; + const size_t per_label = 3; + + SVSParams params = {.dim = dim, .metric = VecSimMetric_L2}; + VecSimIndex *index = this->CreateNewIndex(params); + ASSERT_INDEX(index); + + for (size_t i = 0; i < per_label; i++) { + GenerateAndAddVector(index, dim, 1, i); + GenerateAndAddVector(index, dim, 2, i + 100); + } + + ASSERT_EQ(VecSimIndex_RelabelVector(index, 1, 2), VecSimRelabel_NewLabelTaken); + ASSERT_EQ(VecSimIndex_RelabelVector(index, 1, 1), VecSimRelabel_SameLabel); + ASSERT_EQ(VecSimIndex_RelabelVector(index, 42, 43), VecSimRelabel_OldLabelMissing); + + // A rejection leaves every copy of both labels in place. + ASSERT_EQ(VecSimIndex_IndexSize(index), per_label * 2); + ASSERT_EQ(index->indexLabelCount(), 2); + + VecSimIndex_Free(index); +} + +#endif // HAVE_SVS_REPLACE_EXTERNAL_ID diff --git a/tests/unit/test_svs_tiered.cpp b/tests/unit/test_svs_tiered.cpp index 8ff9d7d38..f6e387558 100644 --- a/tests/unit/test_svs_tiered.cpp +++ b/tests/unit/test_svs_tiered.cpp @@ -9,6 +9,7 @@ #if HAVE_SVS #include +#include // For getAvailableCPUs(): #include @@ -269,6 +270,247 @@ TYPED_TEST(SVSTieredIndexTest, ThreadsReservation) { mock_thread_pool.thread_pool_join(); } +// Relabel on a tier, in the two write states a vector can be in: buffered with its update job +// still pending, and moved to the backend. These are the states `addVector` and `insertJob` cover +// for insertion, and the tier has to move the label in whichever one holds it. +// +// The fixture's type set spans single-value, multi-value and Quant_8, so this also answers whether +// a compressed backend can move a label -- `replace_external_id` renames an id and never touches +// the stored vector, so it can, even where the backend cannot report values. +TYPED_TEST(SVSTieredIndexTest, relabelVectorMovesTheLabelInBothWriteStates) { + size_t dim = 4; + SVSParams params = {.type = TypeParam::get_index_type(), + .dim = dim, + .metric = VecSimMetric_L2, + .multi = TypeParam::isMulti(), + .quantBits = TypeParam::get_quant_bits()}; + VecSimParams svs_params = CreateParams(params); + auto mock_thread_pool = tieredIndexMock(); + // Thresholds of 1, as `insertJob` uses, so one vector is enough to trigger the update job. + auto *tiered_index = this->CreateTieredSVSIndex(svs_params, mock_thread_pool, 1, 1); + ASSERT_INDEX(tiered_index); + + TEST_DATA_T vector[dim]; + GenerateVector(vector, dim, 7); + VecSimIndex_AddVector(tiered_index, vector, 7); + + // Buffered: nothing has run, so the label lives only in the flat buffer. + ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 1); + ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 0); + ASSERT_EQ(VecSimIndex_RelabelVector(tiered_index, 7, 70), VecSimRelabel_OK); + ASSERT_TRUE(tiered_index->GetFlatIndex()->isLabelExists(70)); + ASSERT_FALSE(tiered_index->GetFlatIndex()->isLabelExists(7)); + ASSERT_EQ(tiered_index->indexLabelCount(), 1) << "the move must not add a label"; + + // Drain: the update job carries the vector to the backend, and must carry the *new* label, + // not the one it had when the job was queued. + mock_thread_pool.init_threads(); + mock_thread_pool.thread_pool_join(); + ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 0); + ASSERT_EQ(tiered_index->GetBackendIndex()->indexSize(), 1); + ASSERT_TRUE(tiered_index->GetSVSIndex()->isLabelExists(70)); + ASSERT_FALSE(tiered_index->GetSVSIndex()->isLabelExists(7)); + + // Backend only: the same move again, now served by the backend tier. + ASSERT_EQ(VecSimIndex_RelabelVector(tiered_index, 70, 700), VecSimRelabel_OK); + ASSERT_TRUE(tiered_index->GetSVSIndex()->isLabelExists(700)); + ASSERT_FALSE(tiered_index->GetSVSIndex()->isLabelExists(70)); + + // Still the same vector, and findable under the label it ended up with. Asserted by search + // rather than by reading values back, because a compressed backend reports no values. + auto verify = [&](size_t label, double score, size_t rank) { + ASSERT_EQ(label, 700); + ASSERT_NEAR(score, 0, 1e-4); + }; + runTopKSearchTest(tiered_index, vector, 1, verify); +} + +// Each rejection, asked of a tier: the target has to be free in *both* of them, so a target taken +// in the buffer and a target taken in the backend are separate cases. +TYPED_TEST(SVSTieredIndexTest, relabelVectorRejectsOnATier) { + size_t dim = 4; + SVSParams params = {.type = TypeParam::get_index_type(), + .dim = dim, + .metric = VecSimMetric_L2, + .multi = TypeParam::isMulti(), + .quantBits = TypeParam::get_quant_bits()}; + VecSimParams svs_params = CreateParams(params); + auto mock_thread_pool = tieredIndexMock(); + // Thresholds of 1, as `insertJob` uses, so one vector is enough to trigger the update job. + auto *tiered_index = this->CreateTieredSVSIndex(svs_params, mock_thread_pool, 1, 1); + ASSERT_INDEX(tiered_index); + + TEST_DATA_T vector[dim]; + GenerateVector(vector, dim, 1); + VecSimIndex_AddVector(tiered_index, vector, 1); + GenerateVector(vector, dim, 2); + VecSimIndex_AddVector(tiered_index, vector, 2); + + ASSERT_EQ(VecSimIndex_RelabelVector(tiered_index, 1, 1), VecSimRelabel_SameLabel); + ASSERT_EQ(VecSimIndex_RelabelVector(tiered_index, 42, 43), VecSimRelabel_OldLabelMissing); + ASSERT_EQ(VecSimIndex_RelabelVector(tiered_index, 1, 2), VecSimRelabel_NewLabelTaken) + << "target held by the buffer"; + + // Move label 2 to the backend, so the next rejection comes from the other tier. + mock_thread_pool.init_threads(); + mock_thread_pool.thread_pool_join(); + ASSERT_TRUE(tiered_index->GetSVSIndex()->isLabelExists(2)); + GenerateVector(vector, dim, 3); + VecSimIndex_AddVector(tiered_index, vector, 3); + ASSERT_EQ(VecSimIndex_RelabelVector(tiered_index, 3, 2), VecSimRelabel_NewLabelTaken) + << "target held by the backend"; + + // A rejection leaves both tiers as they were. + ASSERT_TRUE(tiered_index->GetFlatIndex()->isLabelExists(3)); + ASSERT_TRUE(tiered_index->GetSVSIndex()->isLabelExists(2)); +} + +// The analogue of `insertJobAsync` for relabel: workers running while the label moves. This is +// the case `updateJobMutex` is held for -- an update job snapshots the buffer's labels by value +// and afterwards reconciles only id swaps and deletions, so a rename that landed inside its +// window would be invisible to it and the vector would reach the backend under the old label. +// The tier's version of the deleted-label contract, which has an extra dimension the plain index +// does not: the delete can land while the vector is still buffered or after it reached the +// backend, and a move must report the label absent either way rather than resurrecting it in the +// tier the delete missed. +// +// Unlike a plain SVS index, no tombstone is observable here -- `getNumMarkedDeleted()` reads 0 +// after the delete -- so this asserts the refusal and the absence rather than the marking. The +// plain index covers the soft-delete state itself. +TYPED_TEST(SVSTieredIndexTest, relabelVectorAfterDeleteOnATier) { + size_t dim = 4; + SVSParams params = {.type = TypeParam::get_index_type(), + .dim = dim, + .metric = VecSimMetric_L2, + .multi = TypeParam::isMulti(), + .quantBits = TypeParam::get_quant_bits()}; + VecSimParams svs_params = CreateParams(params); + auto mock_thread_pool = tieredIndexMock(); + // Thresholds of 1 so a single vector reaches the backend, as `insertJob` does. + auto *tiered_index = this->CreateTieredSVSIndex(svs_params, mock_thread_pool, 1, 1); + ASSERT_INDEX(tiered_index); + + // Buffered: deleted before any job ran, so the flat buffer is the tier that held it. + TEST_DATA_T vector[dim]; + GenerateVector(vector, dim, 7); + VecSimIndex_AddVector(tiered_index, vector, 7); + ASSERT_EQ(tiered_index->GetFlatIndex()->indexSize(), 1); + ASSERT_EQ(VecSimIndex_DeleteVector(tiered_index, 7), 1); + ASSERT_EQ(VecSimIndex_RelabelVector(tiered_index, 7, 70), VecSimRelabel_OldLabelMissing); + ASSERT_FALSE(tiered_index->GetFlatIndex()->isLabelExists(70)); + ASSERT_FALSE(tiered_index->GetSVSIndex()->isLabelExists(70)); + + // Ingested, then deleted: now the backend is the tier that held it, and its delete is soft. + GenerateVector(vector, dim, 8); + VecSimIndex_AddVector(tiered_index, vector, 8); + mock_thread_pool.init_threads(); + mock_thread_pool.thread_pool_join(); + ASSERT_TRUE(tiered_index->GetSVSIndex()->isLabelExists(8)); + ASSERT_EQ(VecSimIndex_DeleteVector(tiered_index, 8), 1); + ASSERT_FALSE(tiered_index->GetSVSIndex()->isLabelExists(8)); + + ASSERT_EQ(VecSimIndex_RelabelVector(tiered_index, 8, 80), VecSimRelabel_OldLabelMissing); + ASSERT_FALSE(tiered_index->GetSVSIndex()->isLabelExists(80)) << "a tombstone was renamed"; + ASSERT_FALSE(tiered_index->GetFlatIndex()->isLabelExists(80)); + ASSERT_EQ(tiered_index->indexLabelCount(), 0); +} + +TYPED_TEST(SVSTieredIndexTest, relabelVectorDuringUpdateJob) { + size_t dim = 4; + size_t n = 200; + SVSParams params = {.type = TypeParam::get_index_type(), + .dim = dim, + .metric = VecSimMetric_L2, + .multi = TypeParam::isMulti(), + .quantBits = TypeParam::get_quant_bits()}; + VecSimParams svs_params = CreateParams(params); + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = this->CreateTieredSVSIndex(svs_params, mock_thread_pool); + ASSERT_INDEX(tiered_index); + + // Labels are moved into a range past `n`, so a move can never collide with a label that is + // still waiting to be ingested. + const labelType offset = n + 500; + mock_thread_pool.init_threads(); + for (size_t i = 0; i < n; i++) { + GenerateAndAddVector(tiered_index, dim, i, i); + } + for (size_t i = 0; i < n; i++) { + // Whichever tier holds it by now, and whether or not a job is mid-flight, the move must + // be accepted and must be the only thing that changes. + ASSERT_EQ(VecSimIndex_RelabelVector(tiered_index, i, i + offset), VecSimRelabel_OK) + << "label " << i; + } + mock_thread_pool.thread_pool_join(); + + // Every vector survived under its new label and under no other, which is what fails if a job + // ingested one under the label it was queued with. + ASSERT_EQ(tiered_index->indexLabelCount(), n); + for (size_t i = 0; i < n; i++) { + ASSERT_FALSE(tiered_index->GetFlatIndex()->isLabelExists(i)) << "stale label " << i; + ASSERT_FALSE(tiered_index->GetSVSIndex()->isLabelExists(i)) << "stale label " << i; + ASSERT_TRUE(tiered_index->GetFlatIndex()->isLabelExists(i + offset) || + tiered_index->GetSVSIndex()->isLabelExists(i + offset)) + << "lost label " << i + offset; + } +} + +// What `updateJobMutex` is actually for, pinned. An update job snapshots the buffer's labels by +// value, then adds to the backend; the tracing hook sits between those two steps, which is the +// only window where a rename is invisible to the job. Land one there and, unheld, the job would +// carry the vector into the backend under the label it snapshotted -- the old one. +// +// The relabel has to come from another thread: the job holds `updateJobMutex` for its whole +// duration, so relabeling from inside the hook would block on a lock this thread already holds. +// Blocking is the expected outcome here, and is what makes the result correct. +TYPED_TEST(SVSTieredIndexTest, relabelVectorCannotLandInsideAnUpdateJobsWindow) { + size_t dim = 4; + SVSParams params = {.type = TypeParam::get_index_type(), + .dim = dim, + .metric = VecSimMetric_L2, + .multi = TypeParam::isMulti(), + .quantBits = TypeParam::get_quant_bits()}; + VecSimParams svs_params = CreateParams(params); + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = this->CreateTieredSVSIndex(svs_params, mock_thread_pool, 1, 1); + ASSERT_INDEX(tiered_index); + + std::thread relabel_thread; + std::atomic relabel_entered{false}; + bool hooked = false; + tiered_index->registerTracingCallback("UpdateJob::before_add_to_svs", [&]() { + if (hooked) { + return; + } + hooked = true; + relabel_thread = std::thread([&]() { + relabel_entered = true; + VecSimIndex_RelabelVector(tiered_index, 7, 70); + }); + while (!relabel_entered) { + std::this_thread::yield(); + } + // Long enough that the move would have completed if it were free to. + std::this_thread::sleep_for(std::chrono::milliseconds(50)); + }); + + TEST_DATA_T vector[dim]; + GenerateVector(vector, dim, 7); + VecSimIndex_AddVector(tiered_index, vector, 7); + mock_thread_pool.init_threads(); + mock_thread_pool.thread_pool_join(); + ASSERT_TRUE(hooked) << "the hook never ran, so the window was not exercised"; + relabel_thread.join(); + + // The move happened, and it happened to the vector the job had already ingested -- so the + // backend holds the new label and nothing holds the old one. + ASSERT_EQ(tiered_index->indexLabelCount(), 1); + ASSERT_FALSE(tiered_index->GetSVSIndex()->isLabelExists(7)) << "ingested under the stale label"; + ASSERT_FALSE(tiered_index->GetFlatIndex()->isLabelExists(7)); + ASSERT_TRUE(tiered_index->GetSVSIndex()->isLabelExists(70) || + tiered_index->GetFlatIndex()->isLabelExists(70)); +} + TYPED_TEST(SVSTieredIndexTest, TestDebugInfoThreadCount) { // Set thread_pool_size to 4 or actual number of available CPUs const auto num_threads = std::min(4U, getAvailableCPUs());