diff --git a/src/VecSim/algorithms/brute_force/brute_force_single.h b/src/VecSim/algorithms/brute_force/brute_force_single.h index 360a28374..aa28d97aa 100644 --- a/src/VecSim/algorithms/brute_force/brute_force_single.h +++ b/src/VecSim/algorithms/brute_force/brute_force_single.h @@ -155,7 +155,8 @@ int BruteForceIndex_Single::addVector(const void *vector_dat // Check if label already exists, so it is an update operation. if (optionalID != this->labelToIdLookup.end()) { idType id = optionalID->second; - this->vectors->updateElement(id, vector_data); + auto processed_blob = this->preprocessForStorage(vector_data); + this->vectors->updateElement(id, processed_blob.get()); return 0; } diff --git a/src/VecSim/algorithms/hnsw/hnsw.h b/src/VecSim/algorithms/hnsw/hnsw.h index f78b678c9..cb71d12f1 100644 --- a/src/VecSim/algorithms/hnsw/hnsw.h +++ b/src/VecSim/algorithms/hnsw/hnsw.h @@ -251,6 +251,7 @@ class HNSWIndex : public VecSimIndexAbstract, void unlockIndexDataGuard() const; void lockSharedIndexDataGuard() const; void unlockSharedIndexDataGuard() const; + std::shared_lock acquireSharedIndexDataGuard() const; void lockNodeLinks(idType node_id) const; void unlockNodeLinks(idType node_id) const; VisitedNodesHandler *getVisitedList() const; @@ -545,6 +546,12 @@ void HNSWIndex::unlockSharedIndexDataGuard() const { indexDataGuard.unlock_shared(); } +template +std::shared_lock +HNSWIndex::acquireSharedIndexDataGuard() const { + return std::shared_lock(indexDataGuard); +} + template void HNSWIndex::lockNodeLinks(idType node_id) const { elementLocks[node_id].lock(); diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered.h b/src/VecSim/algorithms/hnsw/hnsw_tiered.h index 11f6cf76f..b82ad2343 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered.h @@ -9,7 +9,12 @@ #pragma once +#include +#include +#include + #include "VecSim/algorithms/brute_force/brute_force_single.h" +#include "VecSim/spaces/computer/preprocessors.h" #include "VecSim/vec_sim_tiered_index.h" #include "hnsw.h" #include "VecSim/index_factories/hnsw_factory.h" @@ -46,6 +51,7 @@ struct HNSWSwapJob : public VecsimBaseObject { static const size_t DEFAULT_PENDING_SWAP_JOBS_THRESHOLD = DEFAULT_BLOCK_SIZE; static const size_t MAX_PENDING_SWAP_JOBS_THRESHOLD = 100000; +static const size_t MAX_QUANT_NORMALIZATION_SET_SIZE = 100 * DEFAULT_BLOCK_SIZE; /** * Definition of a job that repairs a certain node's connection in HNSW Index after delete @@ -99,6 +105,25 @@ class TieredHNSWIndex : public VecSimTieredIndex { // Not atomic since it's only accessed from the main thread. size_t directHNSWInsertions{0}; + bool isQuantized{false}; + size_t quantNormalizationSetSize; + + struct SQAccumulationState { + vecsim_stl::vector runningSumVec; + VecSimParams backendIndexParams; + + SQAccumulationState(std::shared_ptr allocator, const VecSimParams ¶ms) + : runningSumVec(allocator), backendIndexParams(params) {} + }; + std::optional sqAccumulationState; + +#ifdef BUILD_TESTS + std::function beforeQuantizedBackendReplacement; + std::function afterBackendInsertBeforeFlatRemoval; +#endif + + void initializeQuantizedBackend(); + void executeInsertJob(HNSWInsertJob *job); void executeRepairJob(HNSWRepairJob *job); @@ -150,6 +175,24 @@ class TieredHNSWIndex : public VecSimTieredIndex { // Handle deletion of vector inplace considering that async deletion might occurred beforehand. int deleteLabelFromHNSWInplace(labelType label); + void addToSum(const DataType *vector) { + if constexpr (QuantInput) { + auto &runningSumVec = this->sqAccumulationState->runningSumVec; + for (size_t i = 0; i < runningSumVec.size(); i++) { + runningSumVec[i] += to_fp32(vector[i]); + } + } + } + + void subtractFromSum(const DataType *vector) { + if constexpr (QuantInput) { + auto &runningSumVec = this->sqAccumulationState->runningSumVec; + for (size_t i = 0; i < runningSumVec.size(); i++) { + runningSumVec[i] -= to_fp32(vector[i]); + } + } + } + #ifdef BUILD_TESTS #include "VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h" #endif @@ -158,6 +201,7 @@ class TieredHNSWIndex : public VecSimTieredIndex { class TieredHNSW_BatchIterator : public VecSimBatchIterator { private: const TieredHNSWIndex *index; + std::optional> backend_index_lock; VecSimQueryParams *queryParams; VecSimQueryResultContainer flat_results; @@ -211,11 +255,24 @@ class TieredHNSWIndex : public VecSimTieredIndex { std::shared_ptr allocator); virtual ~TieredHNSWIndex(); +#ifdef BUILD_TESTS + void setBeforeQuantizedBackendReplacementHook(std::function hook) { + beforeQuantizedBackendReplacement = std::move(hook); + } + + void setAfterBackendInsertBeforeFlatRemovalHook(std::function hook) { + afterBackendInsertBeforeFlatRemoval = std::move(hook); + } +#endif + int addVector(const void *blob, labelType label) override; int deleteVector(labelType label) override; VecSimRelabelCode relabelVector(labelType old_label, labelType new_label) override; size_t getNumMarkedDeleted() const override { - return this->getHNSWIndex()->getNumMarkedDeleted(); + return this->isBackendPublished() + ? static_cast &>(this->publishedBackend()) + .getNumMarkedDeleted() + : 0; } size_t indexSize() const override; size_t indexCapacity() const override; @@ -233,9 +290,16 @@ class TieredHNSWIndex : public VecSimTieredIndex { TieredHNSW_BatchIterator(queryBlob, this, queryParams, this->allocator); } inline void setLastSearchMode(VecSearchMode mode) override { - return this->backendIndex->setLastSearchMode(mode); + if (this->isBackendPublished()) { + this->publishedBackend().setLastSearchMode(mode); + } else { + this->frontendIndex->setLastSearchMode(mode); + } } void runGC() override { + if (!this->isBackendPublished()) { + return; + } // Run no more than pendingSwapJobsThreshold value jobs. TIERED_LOG(VecSimCommonStrings::LOG_VERBOSE_STRING, "running asynchronous GC for tiered HNSW index"); @@ -244,26 +308,33 @@ class TieredHNSWIndex : public VecSimTieredIndex { void acquireSharedLocks() override { this->flatIndexGuard.lock_shared(); this->mainIndexGuard.lock_shared(); - this->getHNSWIndex()->lockSharedIndexDataGuard(); + if (this->backendIndex) { + this->getHNSWIndex()->lockSharedIndexDataGuard(); + } } void releaseSharedLocks() override { + if (this->backendIndex) { + this->getHNSWIndex()->unlockSharedIndexDataGuard(); + } this->flatIndexGuard.unlock_shared(); this->mainIndexGuard.unlock_shared(); - this->getHNSWIndex()->unlockSharedIndexDataGuard(); } VecSimDebugCommandCode getHNSWElementNeighbors(size_t label, int ***neighborsData) { - this->mainIndexGuard.lock_shared(); - auto res = this->getHNSWIndex()->getHNSWElementNeighbors(label, neighborsData); - this->mainIndexGuard.unlock_shared(); - return res; + std::shared_lock main_index_lock(this->mainIndexGuard); + return this->backendIndex + ? this->getHNSWIndex()->getHNSWElementNeighbors(label, neighborsData) + : VecSimDebugCommandCode_LabelNotExists; } #ifdef BUILD_TESTS size_t indexMetaDataCapacity() const override { - return this->backendIndex->indexMetaDataCapacity() + - this->frontendIndex->indexMetaDataCapacity(); + size_t capacity = this->frontendIndex->indexMetaDataCapacity(); + if (this->isBackendPublished()) { + capacity += this->publishedBackend().indexMetaDataCapacity(); + } + return capacity; } #endif }; @@ -339,6 +410,8 @@ void TieredHNSWIndex::invalidateRepairJobs(idType deleted_id template HNSWIndex *TieredHNSWIndex::getHNSWIndex() const { + // The pointer itself is plain storage. Callers must run on the write thread or hold + // mainIndexGuard; lock-free readers must first observe publication and use publishedBackend(). return dynamic_cast *>(this->backendIndex); } @@ -597,6 +670,12 @@ void TieredHNSWIndex::executeInsertJob(HNSWInsertJob *job) { this->insertVectorToHNSW(hnsw_index, job->label, blob_copy.get()); +#ifdef BUILD_TESTS + if (afterBackendInsertBeforeFlatRemoval) { + afterBackendInsertBeforeFlatRemoval(); + } +#endif + // Remove the vector and the insert job from the flat buffer. this->flatIndexGuard.lock(); // The job might have been invalidated due to overwrite in the meantime. In this case, @@ -704,7 +783,7 @@ TieredHNSWIndex::TieredHNSWIndex(HNSWIndex(hnsw_index, bf_index, tiered_index_params, allocator), labelToInsertJobs(this->allocator), idToRepairJobs(this->allocator), idToSwapJob(this->allocator), invalidJobs(this->allocator), currInvalidJobId(0), - readySwapJobs(0) { + readySwapJobs(0), quantNormalizationSetSize(0) { // If the param for swapJobThreshold is 0 use the default value, if it exceeds the maximum // allowed, use the maximum value. this->pendingSwapJobsThreshold = @@ -712,6 +791,20 @@ TieredHNSWIndex::TieredHNSWIndex(HNSWIndexalgoParams.hnswParams; + if (hnswParams.quantType != VecSimQuant_NONE) { + isQuantized = true; + size_t normSize = + tiered_index_params.specificParams.tieredHnswParams.QuantNormalizationSetSize; + if (normSize > 0) { + this->quantNormalizationSetSize = std::min(normSize, MAX_QUANT_NORMALIZATION_SET_SIZE); + this->sqAccumulationState.emplace(this->allocator, + *tiered_index_params.primaryIndexParams); + this->sqAccumulationState->runningSumVec.resize(hnswParams.dim, 0.0); + } + } } template @@ -738,19 +831,55 @@ TieredHNSWIndex::~TieredHNSWIndex() { } } +template +void TieredHNSWIndex::initializeQuantizedBackend() { + assert((QuantInput && std::is_same_v)); + assert(this->sqAccumulationState); + assert(this->quantNormalizationSetSize > 0); + assert(this->frontendIndex->indexSize() == this->quantNormalizationSetSize); + + auto &accumulationState = *this->sqAccumulationState; + auto &hnswParams = accumulationState.backendIndexParams.algoParams.hnswParams; + vecsim_stl::vector mean(hnswParams.dim, this->allocator); + for (size_t i = 0; i < hnswParams.dim; i++) { + mean[i] = static_cast(accumulationState.runningSumVec[i] / + static_cast(this->quantNormalizationSetSize)); + } + + hnswParams.quantParams = mean.data(); + auto *new_backend = static_cast *>( + HNSWFactory::NewIndex(&accumulationState.backendIndexParams, true)); + +#ifdef BUILD_TESTS + if (beforeQuantizedBackendReplacement) { + beforeQuantizedBackendReplacement(); + } +#endif + + { + auto main_index_lock = this->acquireMainIndexGuard(); + this->backendIndex = new_backend; + this->backendPublished.store(true, std::memory_order_release); + } + this->sqAccumulationState.reset(); +} + template size_t TieredHNSWIndex::indexSize() const { - this->flatIndexGuard.lock_shared(); - this->getHNSWIndex()->lockSharedIndexDataGuard(); - size_t res = this->backendIndex->indexSize() + this->frontendIndex->indexSize(); - this->getHNSWIndex()->unlockSharedIndexDataGuard(); - this->flatIndexGuard.unlock_shared(); + std::shared_lock flat_index_lock(this->flatIndexGuard); + size_t res = this->frontendIndex->indexSize(); + if (this->isBackendPublished()) { + auto &hnsw_index = static_cast &>(this->publishedBackend()); + auto index_data_lock = hnsw_index.acquireSharedIndexDataGuard(); + res += hnsw_index.indexSize(); + } return res; } template size_t TieredHNSWIndex::indexCapacity() const { - return this->backendIndex->indexCapacity() + this->frontendIndex->indexCapacity(); + return (this->isBackendPublished() ? this->publishedBackend().indexCapacity() : 0) + + this->frontendIndex->indexCapacity(); } // In the tiered index, we assume that the blobs are processed by the flat buffer @@ -764,7 +893,8 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l auto hnsw_index = this->getHNSWIndex(); // writeMode is not protected since it is assumed to be called only from the "main thread" // (that is the thread that is exclusively calling add/delete vector). - if (this->getWriteMode() == VecSim_WriteInPlace) { + // VecSim_WriteInPlace is ignored during the accumulation phase + if (hnsw_index && this->getWriteMode() == VecSim_WriteInPlace) { // First, check if we need to overwrite the vector in-place for single (from both indexes). if (!this->backendIndex->isMultiValue()) { ret -= this->deleteVector(label); @@ -782,7 +912,7 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l ++this->directHNSWInsertions; return ret; } - if (this->frontendIndex->indexSize() >= this->flatBufferLimit) { + if (hnsw_index && this->frontendIndex->indexSize() >= this->flatBufferLimit) { // Handle overwrite situation. if (!this->backendIndex->isMultiValue()) { // This will do nothing (and return 0) if this label doesn't exist. Otherwise, it may @@ -809,6 +939,10 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l if (this->frontendIndex->isLabelExists(label) && !this->frontendIndex->isMultiValue()) { // Overwrite the vector and invalidate its only pending job (since we are not in MULTI). auto *old_job = this->labelToInsertJobs.at(label).at(0); + if (!hnsw_index) { + const DataType *vector_data = this->frontendIndex->getDataByInternalId(old_job->id); + this->subtractFromSum(vector_data); + } old_job->id = this->setAndSaveInvalidJob(old_job); this->labelToInsertJobs.erase(label); ret = 0; @@ -822,6 +956,9 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l } // If this label already exists, this will do overwrite. this->frontendIndex->addVector(blob, label); + if (!hnsw_index) { + this->addToSum(this->frontendIndex->getDataByInternalId(new_flat_id)); + } AsyncJob *new_insert_job = new (this->allocator) HNSWInsertJob(this->allocator, label, new_flat_id, executeInsertJobWrapper, this); @@ -829,7 +966,7 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l if (this->labelToInsertJobs.find(label) != this->labelToInsertJobs.end()) { // There's already a pending insert job for this label, add another one (without overwrite, // only possible in multi index) - assert(this->backendIndex->isMultiValue()); + assert(this->frontendIndex->isMultiValue()); this->labelToInsertJobs.at(label).push_back((HNSWInsertJob *)new_insert_job); } else { vecsim_stl::vector new_jobs_vec(1, (HNSWInsertJob *)new_insert_job, @@ -841,7 +978,7 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l // Here, a worker might ingest the previous vector that was stored under "label" // (in case of override in non-MULTI index) - so if it's there, we remove it (and create the // required repair jobs), *before* we submit the insert job. - if (!this->backendIndex->isMultiValue()) { + if (hnsw_index && !this->backendIndex->isMultiValue()) { // If we removed the previous vector from both HNSW and flat in the overwrite process, // we still return 0 (not -1). ret = std::max(ret - this->deleteLabelFromHNSW(label), 0); @@ -855,8 +992,24 @@ int TieredHNSWIndex::addVector(const void *blob, labelType l this->executeReadySwapJobs(this->pendingSwapJobsThreshold); } - // Insert job to the queue and signal the workers' updater. - this->submitSingleJob(new_insert_job); + if (hnsw_index) { + // Insert job to the queue and signal the workers' updater. + this->submitSingleJob(new_insert_job); + } else if (this->frontendIndex->indexSize() >= this->quantNormalizationSetSize) { + // If we are in the accumulation phase and we just reached the quantization set size, we + // can initalize the backend index with accumulated mean and transition to the regular mode. + this->initializeQuantizedBackend(); + + // Submit all pending insert jobs to the job queue. + vecsim_stl::vector jobs(this->allocator); + jobs.reserve(this->frontendIndex->indexSize()); + for (auto &entry : this->labelToInsertJobs) { + for (auto *job : entry.second) { + jobs.push_back(job); + } + } + this->submitJobs(jobs); + } return ret; } @@ -872,6 +1025,10 @@ int TieredHNSWIndex::deleteVector(labelType label) { // Invalidate the pending insert job(s) into HNSW associated with this label auto &insert_jobs = this->labelToInsertJobs.at(label); for (auto *job : insert_jobs) { + if (!this->backendIndex) { + const DataType *vector_data = this->frontendIndex->getDataByInternalId(job->id); + this->subtractFromSum(vector_data); + } job->id = this->setAndSaveInvalidJob(job); } num_deleted_vectors += insert_jobs.size(); @@ -893,6 +1050,10 @@ int TieredHNSWIndex::deleteVector(labelType label) { this->flatIndexGuard.unlock_shared(); } + if (!this->backendIndex) { + return num_deleted_vectors; + } + // Next, check if there vector(s) stored under the given label in HNSW and delete them as well. // Note that we may remove the same vector that has been removed from the flat index, if it was // being ingested at that time. @@ -950,11 +1111,11 @@ VecSimRelabelCode TieredHNSWIndex::relabelVector(labelType o const bool source_exists = this->frontendIndex->isLabelExists(old_label) || this->labelToInsertJobs.find(old_label) != this->labelToInsertJobs.end() || - hnsw_index->isLabelExists(old_label); + (hnsw_index && hnsw_index->isLabelExists(old_label)); const bool target_taken = this->frontendIndex->isLabelExists(new_label) || this->labelToInsertJobs.find(new_label) != this->labelToInsertJobs.end() || - hnsw_index->isLabelExists(new_label); + (hnsw_index && hnsw_index->isLabelExists(new_label)); // Each home is asked to move the label only once it reported holding it, and the checks above // ruled out every other rejection - the label is present, the target is free everywhere, and @@ -987,7 +1148,7 @@ VecSimRelabelCode TieredHNSWIndex::relabelVector(labelType o // `relabelVector` takes the HNSW index data guard internally, which is the same // main-guard-then-data-guard order that `insertVectorToHNSW` uses. - if (hnsw_index->isLabelExists(old_label)) { + if (hnsw_index && hnsw_index->isLabelExists(old_label)) { const VecSimRelabelCode hnsw_ret = hnsw_index->relabelVector(old_label, new_label); #ifdef BUILD_TESTS assert(hnsw_ret == VecSimRelabel_OK && "HNSW just reported holding this label"); @@ -1039,6 +1200,10 @@ double TieredHNSWIndex::getDistanceFrom_Unsafe(labelType lab // If the label doesn't exist, the distance will be NaN. auto flat_dist = this->frontendIndex->getDistanceFrom_Unsafe(label, blob); + if (!this->backendIndex) { + return flat_dist; + } + // Optimization. TODO: consider having different implementations for single and multi indexes, // to avoid checking the index type on every query. if (!this->backendIndex->isMultiValue() && !std::isnan(flat_dist)) { @@ -1077,8 +1242,14 @@ TieredHNSWIndex::TieredHNSW_BatchIterator::TieredHNSW_BatchI : VecSimBatchIterator(nullptr, queryParams ? queryParams->timeoutCtx : nullptr, std::move(allocator)), index(index), flat_results(this->allocator), hnsw_results(this->allocator), - flat_iterator(this->index->frontendIndex->newBatchIterator(query_vector, queryParams)), - hnsw_iterator(UNINITIALIZED), returned_results_set(this->allocator) { + flat_iterator(UNINITIALIZED), hnsw_iterator(UNINITIALIZED), + returned_results_set(this->allocator) { + { + std::shared_lock flat_index_lock(this->index->flatIndexGuard); + this->flat_iterator = + this->index->frontendIndex->newBatchIterator(query_vector, queryParams); + } + // Save a copy of the query params to initialize the HNSW iterator with (on first batch and // first batch after reset). if (queryParams) { @@ -1096,7 +1267,6 @@ TieredHNSWIndex::TieredHNSW_BatchIterator::~TieredHNSW_Batch if (this->hnsw_iterator != UNINITIALIZED && this->hnsw_iterator != DEPLETED) { delete this->hnsw_iterator; - this->index->mainIndexGuard.unlock_shared(); } this->allocator->free_allocation(this->queryParams); @@ -1108,7 +1278,8 @@ template VecSimQueryReply *TieredHNSWIndex::TieredHNSW_BatchIterator::getNextResults( size_t n_res, VecSimQueryReply_Order order) { - const bool isMulti = this->index->backendIndex->isMultiValue(); + const bool isMulti = this->index->frontendIndex->isMultiValue(); + const bool needsDedup = isMulti || this->index->isQuantized; auto hnsw_code = VecSim_QueryReply_OK; if (this->hnsw_iterator == UNINITIALIZED) { @@ -1125,17 +1296,22 @@ VecSimQueryReply *TieredHNSWIndex::TieredHNSW_BatchIterator: VecSimQueryReply_Free(cur_flat_results); // We also take the lock on the main index on the first call to getNextResults, and we hold // it until the iterator is depleted or freed. - this->index->mainIndexGuard.lock_shared(); - this->hnsw_iterator = this->index->backendIndex->newBatchIterator( - this->flat_iterator->getQueryBlob(), queryParams); - auto cur_hnsw_results = this->hnsw_iterator->getNextResults(n_res, BY_SCORE_THEN_ID); - hnsw_code = cur_hnsw_results->code; - this->hnsw_results.swap(cur_hnsw_results->results); - VecSimQueryReply_Free(cur_hnsw_results); - if (this->hnsw_iterator->isDepleted()) { - delete this->hnsw_iterator; + this->backend_index_lock.emplace(this->index->mainIndexGuard); + if (!this->index->backendIndex) { this->hnsw_iterator = DEPLETED; - this->index->mainIndexGuard.unlock_shared(); + this->backend_index_lock.reset(); + } else { + this->hnsw_iterator = this->index->backendIndex->newBatchIterator( + this->flat_iterator->getQueryBlob(), queryParams); + auto cur_hnsw_results = this->hnsw_iterator->getNextResults(n_res, BY_SCORE_THEN_ID); + hnsw_code = cur_hnsw_results->code; + this->hnsw_results.swap(cur_hnsw_results->results); + VecSimQueryReply_Free(cur_hnsw_results); + if (this->hnsw_iterator->isDepleted()) { + delete this->hnsw_iterator; + this->hnsw_iterator = DEPLETED; + this->backend_index_lock.reset(); + } } } else { while (this->flat_results.size() < n_res && !this->flat_iterator->isDepleted()) { @@ -1145,15 +1321,15 @@ VecSimQueryReply *TieredHNSWIndex::TieredHNSW_BatchIterator: tail->results.end()); VecSimQueryReply_Free(tail); - if (!isMulti) { + if (!needsDedup) { // On single-value indexes, duplicates will never appear in the hnsw results before // they appear in the flat results (at the same time or later if the approximation // misses) so we don't need to try and filter the flat results (and recheck // conditions). break; } else { - // On multi-value indexes, the flat results may contain results that are already - // returned from the hnsw index. We need to filter them out. + // On multi-value and quantized indexes, the flat results may contain results that + // were already returned from the hnsw index. We need to filter them out. filter_irrelevant_results(this->flat_results); } } @@ -1174,7 +1350,7 @@ VecSimQueryReply *TieredHNSWIndex::TieredHNSW_BatchIterator: if (this->hnsw_iterator->isDepleted()) { delete this->hnsw_iterator; this->hnsw_iterator = DEPLETED; - this->index->mainIndexGuard.unlock_shared(); + this->backend_index_lock.reset(); } } } @@ -1184,7 +1360,7 @@ VecSimQueryReply *TieredHNSWIndex::TieredHNSW_BatchIterator: } VecSimQueryReply *batch; - if (isMulti) + if (needsDedup) batch = compute_current_batch(n_res); else batch = compute_current_batch(n_res); @@ -1214,7 +1390,7 @@ template void TieredHNSWIndex::TieredHNSW_BatchIterator::reset() { if (this->hnsw_iterator != UNINITIALIZED && this->hnsw_iterator != DEPLETED) { delete this->hnsw_iterator; - this->index->mainIndexGuard.unlock_shared(); + this->backend_index_lock.reset(); } this->resetResultsCount(); this->flat_iterator->reset(); @@ -1337,7 +1513,7 @@ VecSimDebugInfoIterator *TieredHNSWIndex::debugInfoIterator( template VecSimIndexBasicInfo TieredHNSWIndex::basicInfo() const { - VecSimIndexBasicInfo info = this->backendIndex->getBasicInfo(); + VecSimIndexBasicInfo info = this->frontendIndex->getBasicInfo(); info.isTiered = true; info.algo = VecSimAlgo_HNSWLIB; return info; diff --git a/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h b/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h index 0ea70d59b..df3b6994f 100644 --- a/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h +++ b/src/VecSim/algorithms/hnsw/hnsw_tiered_tests_friends.h @@ -68,6 +68,9 @@ INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_deleteInplaceAvoidUpdatedMarked INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_switchDeleteModes_Test) INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestBasic_HNSWResize_Test) +INDEX_TEST_FRIEND_CLASS(HNSWTieredIndexTestSQ8) +INDEX_TEST_FRIEND_CLASS(SQ8TieredHNSWTest) + friend class CommonAPITest_SearchDifferentScores_Test; friend class BF16TieredTest; friend class FP16TieredTest; diff --git a/src/VecSim/index_factories/components/components_factory.h b/src/VecSim/index_factories/components/components_factory.h index f52db7c5f..4c4489d5b 100644 --- a/src/VecSim/index_factories/components/components_factory.h +++ b/src/VecSim/index_factories/components/components_factory.h @@ -31,6 +31,58 @@ CreateIndexComponents(std::shared_ptr allocator, VecSimMetric m return {indexCalculator, preprocessors}; } +// Asymmetric dispatch reports alignment for the stored operand only. Ask the query type's +// dispatcher for the query allocation alignment. +template +[[nodiscard]] unsigned char GetQueryAlignment(VecSimMetric metric, size_t dim) { + unsigned char alignment = 0; + spaces::GetDistFunc(metric, dim, &alignment); + return alignment; +} + +template +IndexComponents +CreateSQ8IndexComponents(const std::shared_ptr &allocator, size_t dim, + const float *mean_ptr) { + const bool with_mean = mean_ptr != nullptr; + unsigned char storage_alignment = 0, asym_storage_alignment = 0; + + // Graph construction compares two stored SQ8 blobs; search compares a stored blob with a + // DataType query. Both dispatchers report alignment for the stored operand. + auto sym_func = spaces::GetDistFunc(Metric, dim, &storage_alignment); + auto asym_func = spaces::GetDistFunc( + Metric, dim, &asym_storage_alignment); + storage_alignment = spaces::combineAlignments(storage_alignment, asym_storage_alignment); + const unsigned char query_alignment = GetQueryAlignment(Metric, dim); + + PreprocessorInterface *pp = nullptr; + IndexCalculatorInterface *calc = nullptr; + + if (with_mean) { + vecsim_stl::vector mean_vec(allocator); + mean_vec.assign(mean_ptr, mean_ptr + dim); + + float mean_sum_squares = 0.0f; + for (float v : mean_vec) { + mean_sum_squares += v * v; + } + + pp = new (allocator) QuantPreprocessor(allocator, dim, mean_vec); + calc = new (allocator) DistanceCalculatorWithNorm( + allocator, asym_func, sym_func, mean_sum_squares); + } else { + pp = new (allocator) QuantPreprocessor(allocator, dim); + calc = new (allocator) DistanceCalculatorCommon(allocator, sym_func, asym_func); + } + + auto *container = new (allocator) + MultiPreprocessorsContainer(allocator, query_alignment, storage_alignment); + [[maybe_unused]] const int ret = container->addPreprocessor(pp); + assert(ret != -1 && "SQ8 preprocessor was not added correctly"); + + return {calc, container}; +} + template size_t EstimateComponentsMemory(VecSimMetric metric, bool is_normalized) { size_t allocations_overhead = VecSimAllocator::getAllocationOverheadSize(); diff --git a/src/VecSim/index_factories/hnsw_factory.cpp b/src/VecSim/index_factories/hnsw_factory.cpp index d011def8f..162f60c9e 100644 --- a/src/VecSim/index_factories/hnsw_factory.cpp +++ b/src/VecSim/index_factories/hnsw_factory.cpp @@ -43,15 +43,6 @@ template : sq8::storage_bytes_count(dim); } -// Asymmetric dispatch reports alignment for the stored operand only. Ask the query type's -// dispatcher for the query allocation alignment. -template -[[nodiscard]] unsigned char GetQueryAlignment(VecSimMetric metric, size_t dim) { - unsigned char alignment = 0; - spaces::GetDistFunc(metric, dim, &alignment); - return alignment; -} - // Cosine over pre-normalized vectors is computed as inner product. [[nodiscard]] constexpr VecSimMetric ResolveSQ8Metric(VecSimMetric metric, bool is_normalized) { return (is_normalized && metric == VecSimMetric_Cosine) ? VecSimMetric_IP : metric; @@ -79,48 +70,13 @@ template template VecSimIndex *NewIndex_SQ8(const HNSWParams *hnswParams, AbstractIndexInitParams abstractInitParams, const float *mean_ptr) { - auto &allocator = abstractInitParams.allocator; - const size_t dim = abstractInitParams.dim; const bool with_norm = mean_ptr != nullptr; - unsigned char storage_alignment = 0, asym_storage_alignment = 0; - - abstractInitParams.storedDataSize = GetSQ8StoredDataSize(dim, with_norm); + abstractInitParams.storedDataSize = + GetSQ8StoredDataSize(abstractInitParams.dim, with_norm); abstractInitParams.isQuantized = true; - // Graph construction compares two stored SQ8 blobs; search compares a stored blob with a - // DataType query. Both dispatchers report alignment for the stored operand. - auto sym_func = spaces::GetDistFunc(Metric, dim, &storage_alignment); - auto asym_func = - spaces::GetDistFunc(Metric, dim, &asym_storage_alignment); - storage_alignment = spaces::combineAlignments(storage_alignment, asym_storage_alignment); - const unsigned char query_alignment = GetQueryAlignment(Metric, dim); - - PreprocessorInterface *pp = nullptr; - IndexCalculatorInterface *calc = nullptr; - - if (with_norm) { - vecsim_stl::vector mean_vec(allocator); - mean_vec.assign(mean_ptr, mean_ptr + dim); - - float mean_sum_squares = 0.0f; - for (float v : mean_vec) { - mean_sum_squares += v * v; - } - - pp = new (allocator) QuantPreprocessor(allocator, dim, mean_vec); - calc = new (allocator) DistanceCalculatorWithNorm( - allocator, asym_func, sym_func, mean_sum_squares); - } else { - pp = new (allocator) QuantPreprocessor(allocator, dim); - calc = new (allocator) DistanceCalculatorCommon(allocator, sym_func, asym_func); - } - - auto *container = new (allocator) - MultiPreprocessorsContainer(allocator, query_alignment, storage_alignment); - [[maybe_unused]] const int ret = container->addPreprocessor(pp); - assert(ret != -1 && "SQ8 preprocessor was not added correctly"); - - IndexComponents components{calc, container}; + IndexComponents components = CreateSQ8IndexComponents( + abstractInitParams.allocator, abstractInitParams.dim, mean_ptr); return NewIndex_ChooseMultiOrSingle(hnswParams, abstractInitParams, components); } diff --git a/src/VecSim/index_factories/tiered_factory.cpp b/src/VecSim/index_factories/tiered_factory.cpp index 711472d30..7ec95f4d1 100644 --- a/src/VecSim/index_factories/tiered_factory.cpp +++ b/src/VecSim/index_factories/tiered_factory.cpp @@ -37,24 +37,56 @@ static inline BFParams NewBFParams(const TieredIndexParams *params) { return bf_params; } +static inline bool RequiresSQAccumulation(const TieredIndexParams *params) { + return params->primaryIndexParams->algoParams.hnswParams.quantType != VecSimQuant_NONE && + params->specificParams.tieredHnswParams.QuantNormalizationSetSize > 0; +} + +template +static inline bool IsQuantizationSupported(const TieredIndexParams *params) { + const auto &hnsw_params = params->primaryIndexParams->algoParams.hnswParams; + if (hnsw_params.quantType == VecSimQuant_NONE) { + return true; + } + + if constexpr (!QuantInput || !std::is_same_v) { + return false; + } else { + return hnsw_params.quantType == VecSimQuant_SQ8 && + !(std::is_same_v && hnsw_params.metric == VecSimMetric_L2 && + RequiresSQAccumulation(params)); + } +} + template inline VecSimIndex *NewIndex(const TieredIndexParams *params) { + if (!IsQuantizationSupported(params)) { + return nullptr; + } - // initialize hnsw index + const auto &hnsw_params = params->primaryIndexParams->algoParams.hnswParams; + const bool requires_accumulation = RequiresSQAccumulation(params); // Normalization is done by the frontend index. - auto *hnsw_index = reinterpret_cast *>( - HNSWFactory::NewIndex(params->primaryIndexParams, true)); - // initialize brute force index + auto *hnsw_index = requires_accumulation + ? nullptr + : static_cast *>( + HNSWFactory::NewIndex(params->primaryIndexParams, true)); BFParams bf_params = NewBFParams(params); AbstractIndexInitParams abstractInitParams = VecSimFactory::NewAbstractInitParams(&bf_params, params->primaryIndexParams->logCtx, false); - assert(hnsw_index->getInputBlobSize() == abstractInitParams.storedDataSize); - assert(hnsw_index->getStoredDataSize() == abstractInitParams.storedDataSize); + assert(!hnsw_index || hnsw_index->getInputBlobSize() == abstractInitParams.storedDataSize); + assert(!hnsw_index || hnsw_params.quantType != VecSimQuant_NONE || + hnsw_index->getStoredDataSize() == abstractInitParams.storedDataSize); auto frontendIndex = static_cast *>( BruteForceFactory::NewIndex(&bf_params, abstractInitParams, false)); + if (hnsw_params.quantType == VecSimQuant_SQ8 && hnsw_params.dim < 64) { + frontendIndex->log(VecSimCommonStrings::LOG_WARNING_STRING, + "SQ8 compression is not recommended for dimensions below 64"); + } + // Create new tiered hnsw index std::shared_ptr management_layer_allocator = VecSimAllocator::newVecsimAllocator(); @@ -66,17 +98,31 @@ inline VecSimIndex *NewIndex(const TieredIndexParams *params) { inline size_t EstimateInitialSize(const TieredIndexParams *params) { HNSWParams hnsw_params = params->primaryIndexParams->algoParams.hnswParams; - // Keep size estimation consistent with NewIndex, which rejects quantized tiered indexes. - if (hnsw_params.quantType != VecSimQuant_NONE) { - throw std::invalid_argument("Quantization is not supported for tiered HNSW indexes"); + size_t est = 0; + + const bool requires_accumulation = RequiresSQAccumulation(params); + + if (requires_accumulation) { + // Set quantParams non-null to indicate HNSW SQ8 with_norm index + static char dummy; + hnsw_params.quantParams = &dummy; } - // Add size estimation of VecSimTieredIndex sub indexes. - // Normalization is done by the frontend index. - size_t est = HNSWFactory::EstimateInitialSize(&hnsw_params, true); + // HNSWFactory::EstimateInitialSize will throw if the parameters are invalid + size_t est_backend = HNSWFactory::EstimateInitialSize(&hnsw_params, true); - // Management layer allocator overhead. size_t allocations_overhead = VecSimAllocator::getAllocationOverheadSize(); + + if (requires_accumulation) { + // Add size of SQ accumulation buffer + est += allocations_overhead + hnsw_params.dim * sizeof(double); + } else { + // Add size estimation of VecSimTieredIndex sub indexes. + // Normalization is done by the frontend index. + est += est_backend; + } + + // Management layer allocator overhead. est += sizeof(VecSimAllocator) + allocations_overhead; // Size of the TieredHNSWIndex struct. @@ -100,12 +146,6 @@ inline size_t EstimateInitialSize(const TieredIndexParams *params) { } VecSimIndex *NewIndex(const TieredIndexParams *params) { - // The brute-force frontend is not quantized, so an SQ8 primary index would use an incompatible - // stored-vector layout. - if (params->primaryIndexParams->algoParams.hnswParams.quantType != VecSimQuant_NONE) { - return nullptr; - } - // Tiered index that contains HNSW index as primary index VecSimType type = params->primaryIndexParams->algoParams.hnswParams.type; if (type == VecSimType_FLOAT32) { @@ -247,7 +287,13 @@ size_t EstimateElementSize(const TieredIndexParams *params) { // Match HNSW's element estimator, which leaves validation to NewIndex. size_t est = 0; if (params->primaryIndexParams->algo == VecSimAlgo_HNSWLIB) { - est = HNSWFactory::EstimateElementSize(¶ms->primaryIndexParams->algoParams.hnswParams); + HNSWParams hnsw_params = params->primaryIndexParams->algoParams.hnswParams; + if (TieredHNSWFactory::RequiresSQAccumulation(params)) { + // Set quantParams non-null to indicate HNSW SQ8 with_norm index + static char dummy; + hnsw_params.quantParams = &dummy; + } + est = HNSWFactory::EstimateElementSize(&hnsw_params); } if (params->primaryIndexParams->algo == VecSimAlgo_SVS) { est = SVSFactory::EstimateElementSize(¶ms->primaryIndexParams->algoParams.svsParams); diff --git a/src/VecSim/vec_sim_common.h b/src/VecSim/vec_sim_common.h index 3ec945a01..13008520a 100644 --- a/src/VecSim/vec_sim_common.h +++ b/src/VecSim/vec_sim_common.h @@ -229,6 +229,9 @@ typedef struct { typedef struct { size_t swapJobThreshold; // The minimum number of swap jobs to accumulate before applying // all the ready swap jobs in a batch. + size_t QuantNormalizationSetSize; // Number of vectors to accumulate before SQ initialization. + // 0 = skip accumulation phase (naive SQ8, no mean). + // Max: 100 * DEFAULT_BLOCK_SIZE (102400). } TieredHNSWParams; // A struct that contains HNSW Disk tiered index specific params. diff --git a/src/VecSim/vec_sim_tiered_index.h b/src/VecSim/vec_sim_tiered_index.h index f67e0a649..cdbf36644 100644 --- a/src/VecSim/vec_sim_tiered_index.h +++ b/src/VecSim/vec_sim_tiered_index.h @@ -16,6 +16,8 @@ #include "VecSim/utils/query_result_utils.h" #include "VecSim/utils/alignment.h" +#include +#include #include #if HAVE_SVS @@ -23,7 +25,7 @@ #include "VecSim/algorithms/svs/svs.h" #endif -#define TIERED_LOG this->backendIndex->log +#define TIERED_LOG this->frontendIndex->log /** * Definition of generic job structure for asynchronous tiered index. @@ -49,6 +51,20 @@ class VecSimTieredIndex : public VecSimIndexInterface { VecSimIndexAbstract *backendIndex; BruteForceIndex *frontendIndex; + // Set once backendIndex points to a usable index, and never cleared while readers can access + // this object. The publisher stores backendIndex before the release store; a reader that + // observes true with an acquire load may therefore safely read the plain pointer. + std::atomic backendPublished; + + bool isBackendPublished() const { return backendPublished.load(std::memory_order_acquire); } + + // Callers must first observe isBackendPublished(). Returning a reference encodes the resulting + // non-null invariant at the call site without exposing atomic state through the index API. + VecSimIndexAbstract &publishedBackend() const { + assert(isBackendPublished()); + return *backendIndex; + } + void *jobQueue; void *jobQueueCtx; // External context to be sent to the submit callback. SubmitCB SubmitJobsToQueue; @@ -63,6 +79,14 @@ class VecSimTieredIndex : public VecSimIndexInterface { } void unlockMainIndexGuard() const { mainIndexGuard.unlock(); } + + std::unique_lock acquireMainIndexGuard() const { + std::unique_lock lock(mainIndexGuard); +#ifdef BUILD_TESTS + mainIndexGuard_write_lock_count++; +#endif + return lock; + } #ifdef BUILD_TESTS mutable std::atomic_int mainIndexGuard_write_lock_count = 0; #endif @@ -91,8 +115,11 @@ class VecSimTieredIndex : public VecSimIndexInterface { * @return index label count for debug purposes. */ vecsim_stl::vector computeUnifiedIndexLabelsSetUnsafe() const { - auto [flat_labels, backend_labels] = - std::make_pair(this->frontendIndex->getLabelsSet(), this->backendIndex->getLabelsSet()); + auto flat_labels = this->frontendIndex->getLabelsSet(); + vecsim_stl::set backend_labels(this->allocator); + if (this->backendIndex) { + backend_labels = this->backendIndex->getLabelsSet(); + } // Compute the union of the two sets. vecsim_stl::vector labels_union(this->allocator); @@ -147,13 +174,14 @@ class VecSimTieredIndex : public VecSimIndexInterface { assert(vectors_output.empty() && "getDataByLabel expects an empty output vector"); #endif - // A quantized backend cannot report its stored vectors as values -- the stored form is - // compression plus metadata, and nothing here dequantizes -- so it would append nothing. - bool backend_can_report = true; + // Take the flat lock before sampling publication. If publication wins first, read both + // tiers; if this load still sees false, migration cannot remove a flat vector until this + // method returns. + std::shared_lock flat_lock(this->flatIndexGuard); + const bool backend_published = this->isBackendPublished(); #if HAVE_SVS // TODO(MOD-17706): remove once SVSIndex::getDataByLabel reports real data. Removing it - // means deleting this block, the `backend_can_report` flag, and the guarded include of - // svs.h, then unwrapping the body below. + // means deleting this block and the guarded include of svs.h. // // Until then nothing is read at all for an SVS backend: the buffer alone would be a // partial answer for a multi-value label split across the tiers, and a caller cannot tell @@ -166,18 +194,20 @@ class VecSimTieredIndex : public VecSimIndexInterface { // that), so a derived override would simply not be found. The type test is deliberately // explicit rather than dressed up as a capability: it is a special case, not // architecture. - backend_can_report = dynamic_cast(this->backendIndex) == nullptr; + if (backend_published && + dynamic_cast(&this->publishedBackend()) != nullptr) { + return; + } #endif - if (backend_can_report) { - std::shared_lock flat_lock(this->flatIndexGuard); - const size_t before_flat = vectors_output.size(); - this->frontendIndex->getDataByLabel(label, vectors_output); - // Whether the buffer held it, measured rather than read off emptiness, so the tier - // decision does not depend on an assertion that only exists in test builds. - if (this->backendIndex->isMultiValue() || vectors_output.size() == before_flat) { - std::shared_lock main_lock(this->mainIndexGuard); - this->backendIndex->getDataByLabel(label, vectors_output); - } + const size_t before_flat = vectors_output.size(); + this->frontendIndex->getDataByLabel(label, vectors_output); + // Whether the buffer held it, measured rather than read off emptiness, so the tier + // decision does not depend on an assertion that only exists in test builds. The frontend + // and backend have the same multi-value configuration, and the former exists in Phase 0. + if (backend_published && + (this->frontendIndex->isMultiValue() || vectors_output.size() == before_flat)) { + std::shared_lock main_lock(this->mainIndexGuard); + this->publishedBackend().getDataByLabel(label, vectors_output); } } @@ -185,12 +215,14 @@ class VecSimTieredIndex : public VecSimIndexInterface { BruteForceIndex *frontendIndex_, TieredIndexParams tieredParams, std::shared_ptr allocator) : VecSimIndexInterface(allocator), backendIndex(backendIndex_), - frontendIndex(frontendIndex_), jobQueue(tieredParams.jobQueue), - jobQueueCtx(tieredParams.jobQueueCtx), SubmitJobsToQueue(tieredParams.submitCb), - flatBufferLimit(tieredParams.flatBufferLimit) {} + frontendIndex(frontendIndex_), backendPublished(backendIndex_ != nullptr), + jobQueue(tieredParams.jobQueue), jobQueueCtx(tieredParams.jobQueueCtx), + SubmitJobsToQueue(tieredParams.submitCb), flatBufferLimit(tieredParams.flatBufferLimit) {} virtual ~VecSimTieredIndex() { - VecSimIndex_Free(backendIndex); + if (backendIndex) { + VecSimIndex_Free(backendIndex); + } VecSimIndex_Free(frontendIndex); } @@ -202,8 +234,8 @@ class VecSimTieredIndex : public VecSimIndexInterface { VecSimQueryReply_Order order) const override; virtual inline uint64_t getAllocationSize() const override { - return this->allocator->getAllocationSize() + this->backendIndex->getAllocationSize() + - this->frontendIndex->getAllocationSize(); + return this->allocator->getAllocationSize() + this->frontendIndex->getAllocationSize() + + (this->isBackendPublished() ? this->publishedBackend().getAllocationSize() : 0); } virtual size_t getNumMarkedDeleted() const = 0; size_t indexLabelCount() const override; @@ -213,9 +245,11 @@ class VecSimTieredIndex : public VecSimIndexInterface { bool preferAdHocSearch(size_t subsetSize, size_t k, bool initial_check) const override { // For now, decide according to the bigger index. - return this->backendIndex->indexSize() > this->frontendIndex->indexSize() - ? this->backendIndex->preferAdHocSearch(subsetSize, k, initial_check) - : this->frontendIndex->preferAdHocSearch(subsetSize, k, initial_check); + if (this->isBackendPublished() && + this->publishedBackend().indexSize() > this->frontendIndex->indexSize()) { + return this->publishedBackend().preferAdHocSearch(subsetSize, k, initial_check); + } + return this->frontendIndex->preferAdHocSearch(subsetSize, k, initial_check); } // Return the current state of the global write mode (async/in-place). @@ -226,7 +260,9 @@ class VecSimTieredIndex : public VecSimIndexInterface { inline size_t getFlatBufferLimit() { return this->flatBufferLimit; } virtual void fitMemory() override { - this->backendIndex->fitMemory(); + if (this->isBackendPublished()) { + this->publishedBackend().fitMemory(); + } this->frontendIndex->fitMemory(); } #endif @@ -238,6 +274,13 @@ VecSimTieredIndex::topKQueryImp(const void *queryBlob, size_ VecSimQueryParams *queryParams) const { this->flatIndexGuard.lock_shared(); + // If the backend has not been published yet, every vector is still in the flat buffer. + if (!this->isBackendPublished()) { + auto res = this->frontendIndex->topKQuery(queryBlob, k, queryParams); + this->flatIndexGuard.unlock_shared(); + return res; + } + // If the flat buffer is empty, we can simply query the main index. if (this->frontendIndex->indexSize() == 0) { // Release the flat lock and acquire the main lock. @@ -304,6 +347,16 @@ VecSimTieredIndex::rangeQueryImp(const void *queryBlob, doub VecSimQueryReply_Order order) const { this->flatIndexGuard.lock_shared(); + // If the backend has not been published yet, every vector is still in the flat buffer. + if (!this->isBackendPublished()) { + auto res = this->frontendIndex->rangeQuery(queryBlob, radius, queryParams); + this->flatIndexGuard.unlock_shared(); + if (res) { + sort_results(res, order); + } + return res; + } + // If the flat buffer is empty, we can simply query the main index. if (this->frontendIndex->indexSize() == 0) { // Release the flat lock and acquire the main lock. @@ -393,7 +446,13 @@ VecSimIndexDebugInfo VecSimTieredIndex::debugInfo() const { this->mainIndexGuard.lock_shared(); VecSimIndexDebugInfo frontendInfo = this->frontendIndex->debugInfo(); - VecSimIndexDebugInfo backendInfo = this->backendIndex->debugInfo(); + VecSimIndexDebugInfo backendInfo{}; + if (this->backendIndex) { + backendInfo = this->backendIndex->debugInfo(); + } else { + backendInfo.commonInfo.basicInfo = this->basicInfo(); + backendInfo.commonInfo.lastMode = frontendInfo.commonInfo.lastMode; + } info.commonInfo.indexLabelCount = this->computeUnifiedIndexLabelsSetUnsafe().size(); @@ -409,7 +468,7 @@ VecSimIndexDebugInfo VecSimTieredIndex::debugInfo() const { .algo = backendInfo.commonInfo.basicInfo.algo, .metric = backendInfo.commonInfo.basicInfo.metric, .type = backendInfo.commonInfo.basicInfo.type, - .isMulti = this->backendIndex->isMultiValue(), + .isMulti = backendInfo.commonInfo.basicInfo.isMulti, .isTiered = true, .isDisk = backendInfo.commonInfo.basicInfo.isDisk, .blockSize = backendInfo.commonInfo.basicInfo.blockSize, @@ -455,7 +514,7 @@ VecSimDebugInfoIterator *VecSimTieredIndex::debugInfoIterato .fieldType = INFOFIELD_STRING, .fieldValue = {FieldValue{.stringValue = VecSimCommonStrings::TIERED_STRING}}}); - this->backendIndex->addCommonInfoToIterator(infoIterator, info.commonInfo); + this->frontendIndex->addCommonInfoToIterator(infoIterator, info.commonInfo); infoIterator->addInfoField(VecSim_InfoField{ .fieldName = VecSimCommonStrings::TIERED_MANAGEMENT_MEMORY_STRING, @@ -484,10 +543,13 @@ VecSimDebugInfoIterator *VecSimTieredIndex::debugInfoIterato { std::shared_lock main_lock(this->mainIndexGuard); - infoIterator->addInfoField(VecSim_InfoField{ - .fieldName = VecSimCommonStrings::BACKEND_INDEX_STRING, - .fieldType = INFOFIELD_ITERATOR, - .fieldValue = {FieldValue{.iteratorValue = this->backendIndex->debugInfoIterator()}}}); + if (this->backendIndex) { + auto *backendInfoIterator = this->backendIndex->debugInfoIterator(); + infoIterator->addInfoField( + VecSim_InfoField{.fieldName = VecSimCommonStrings::BACKEND_INDEX_STRING, + .fieldType = INFOFIELD_ITERATOR, + .fieldValue = {FieldValue{.iteratorValue = backendInfoIterator}}}); + } } return infoIterator; }; diff --git a/tests/unit/test_bruteforce.cpp b/tests/unit/test_bruteforce.cpp index f00f4d8c0..5c6f71617 100644 --- a/tests/unit/test_bruteforce.cpp +++ b/tests/unit/test_bruteforce.cpp @@ -129,6 +129,33 @@ TYPED_TEST(BruteForceTest, brute_force_vector_update_test) { VecSimIndex_Free(index); } +TYPED_TEST(BruteForceTest, brute_force_cosine_vector_overwrite_is_normalized) { + size_t dim = 4; + labelType label = 1; + + BFParams params = {.dim = dim, .metric = VecSimMetric_Cosine}; + VecSimIndex *index = this->CreateNewIndex(params); + auto *bf_single_index = this->CastToBF_Single(index); + + TEST_DATA_T initial_vector[] = {1, 1, 1, 1}; + TEST_DATA_T replacement_vector[] = {2, 2, 2, 2}; + TEST_DATA_T normalized_replacement[dim]; + memcpy(normalized_replacement, replacement_vector, sizeof(replacement_vector)); + VecSim_Normalize(normalized_replacement, dim, TypeParam::get_index_type()); + + VecSimIndex_AddVector(index, initial_vector, label); + VecSimIndex_AddVector(index, replacement_vector, label); + + ASSERT_EQ(VecSimIndex_IndexSize(index), 1); + + std::vector> stored_vectors; + bf_single_index->getDataByLabel(label, stored_vectors); + ASSERT_EQ(stored_vectors.size(), 1); + ASSERT_NO_FATAL_FAILURE(CompareVectors(stored_vectors[0].data(), normalized_replacement, dim)); + + VecSimIndex_Free(index); +} + /**** resizing cases ****/ TYPED_TEST(BruteForceTest, resize_and_align_index) { diff --git a/tests/unit/test_hnsw_sq8.cpp b/tests/unit/test_hnsw_sq8.cpp index 6f4c9b451..d588d4027 100644 --- a/tests/unit/test_hnsw_sq8.cpp +++ b/tests/unit/test_hnsw_sq8.cpp @@ -1,8 +1,6 @@ /* * Copyright (c) 2006-Present, Redis Ltd. * All rights reserved. - * SPDX-FileCopyrightText: Copyright 2026 Arm Limited and/or its affiliates - * * * Licensed under your choice of the Redis Source Available License 2.0 * (RSALv2); or (b) the Server Side Public License v1 (SSPLv1); or (c) the @@ -14,14 +12,20 @@ #include "VecSim/types/float16.h" #include "VecSim/types/sq8.h" #include "VecSim/vec_sim.h" +#include "mock_thread_pool.h" #include "unit_test_utils.h" +#include #include +#include #include #include #include +#include #include +#include #include +#include template struct HNSWSQ8IndexType : IndexType { @@ -51,7 +55,7 @@ class HNSWSQ8Test : public ::testing::Test { } } - void SetUp(HNSWParams ¶ms) { + virtual void SetUp(HNSWParams ¶ms) { params.type = index_type_t::get_index_type(); params.quantType = VecSimQuant_SQ8; if constexpr (index_type_t::with_quant_params) { @@ -70,7 +74,7 @@ class HNSWSQ8Test : public ::testing::Test { } } - HNSWIndex *CastToHNSW() { + virtual HNSWIndex *CastToHNSW() { return dynamic_cast *>(index); } @@ -290,7 +294,6 @@ void HNSWSQ8Test::search_empty_index_test() { for (size_t i = 0; i < 100; i++) { VecSimIndex_DeleteVector(index, i); } - ASSERT_EQ(VecSimIndex_IndexSize(index), 0u); reply = VecSimIndex_TopKQuery(index, query, 11, nullptr, BY_SCORE); ASSERT_EQ(VecSimQueryReply_Len(reply), 0u); @@ -515,18 +518,649 @@ TYPED_TEST(HNSWSQ8Test, GraphConstructionIP) { runTopKSearchTest(this->index, query.data(), 10, verify); } -// The tiered frontend does not propagate SQ8 settings, so creation and initial-size estimation must -// reject quantization. -TEST(HNSWSQ8TieredTest, RejectsQuantizedTieredIndex) { +/* ---------------------------- Tiered HNSW tests ---------------------------- */ + +using HNSWSQ8TieredDataTypeSet = + ::testing::Types, + HNSWSQ8IndexType, + HNSWSQ8IndexType>; + +template +class SQ8TieredHNSWTest : public HNSWSQ8Test { +public: + using data_t = typename index_type_t::data_t; + + void create_index_test(); + +protected: + static constexpr size_t normalization_set_size = 10; + + void SetUp(HNSWParams &hnsw_params) override { + hnsw_params.type = index_type_t::get_index_type(); + hnsw_params.quantType = VecSimQuant_SQ8; + VecSimParams vecsim_hnsw_params = CreateParams(hnsw_params); + TieredIndexParams tiered_params = { + .jobQueue = &mock_thread_pool.jobQ, + .jobQueueCtx = mock_thread_pool.ctx, + .submitCb = tieredIndexMock::submit_callback, + .primaryIndexParams = &vecsim_hnsw_params, + .specificParams = {TieredHNSWParams{ + .QuantNormalizationSetSize = + index_type_t::with_quant_params ? normalization_set_size : 0}}}; + VecSimParams vecsim_params = CreateParams(tiered_params); + this->index = VecSimIndex_New(&vecsim_params); + ASSERT_NE(this->index, nullptr); + this->dim = hnsw_params.dim; + mock_thread_pool.ctx->index_strong_ref.reset(this->index); + } + + void TearDown() override {} + + HNSWIndex *CastToHNSW() override { + auto *tiered_index = dynamic_cast *>(this->index); + return tiered_index ? tiered_index->getHNSWIndex() : nullptr; + } + + tieredIndexMock mock_thread_pool; +}; + +template +void SQ8TieredHNSWTest::create_index_test() { + HNSWParams params = {.dim = 40, .metric = VecSimMetric_IP, .M = 16, .efConstruction = 200}; + SetUp(params); + + ASSERT_EQ(VecSimIndex_IndexSize(this->index), 0u); + for (size_t label = 0; label < 100; label++) { + ASSERT_EQ(this->GenerateAndAddVector(label, static_cast(label), 1.0f), 1); + ASSERT_EQ(VecSimIndex_IndexSize(this->index), label + 1); + } + EXPECT_EQ(this->index->basicInfo().type, index_type_t::get_index_type()); + EXPECT_TRUE(this->index->basicInfo().isTiered); +} + +TYPED_TEST_SUITE(SQ8TieredHNSWTest, HNSWSQ8TieredDataTypeSet); + +TYPED_TEST(SQ8TieredHNSWTest, CreateIndex) { this->create_index_test(); } + +TYPED_TEST(SQ8TieredHNSWTest, SizeEstimation) { + constexpr size_t block_size = DEFAULT_BLOCK_SIZE; + HNSWParams hnsw_params = { + .dim = 16, .metric = VecSimMetric_IP, .initialCapacity = block_size, .M = 32}; + this->SetUp(hnsw_params); + + VecSimParams vecsim_hnsw_params = CreateParams(hnsw_params); + TieredIndexParams tiered_params = { + .jobQueue = &this->mock_thread_pool.jobQ, + .jobQueueCtx = this->mock_thread_pool.ctx, + .submitCb = tieredIndexMock::submit_callback, + .primaryIndexParams = &vecsim_hnsw_params, + .specificParams = {TieredHNSWParams{ + .QuantNormalizationSetSize = + TypeParam::with_quant_params ? TestFixture::normalization_set_size : 0}}}; + VecSimParams params = CreateParams(tiered_params); + + EXPECT_EQ(VecSimIndex_EstimateInitialSize(¶ms), this->index->getAllocationSize()); + + for (size_t label = 0; label < block_size; label++) { + ASSERT_EQ(this->GenerateAndAddVector(label, static_cast(label)), 1); + } + while (!this->mock_thread_pool.jobQ.empty()) { + this->mock_thread_pool.thread_iteration(); + } + + const size_t estimation = VecSimIndex_EstimateElementSize(¶ms) * block_size; + const size_t before = this->index->getAllocationSize(); + ASSERT_EQ(this->GenerateAndAddVector(block_size, static_cast(block_size)), 1); + while (!this->mock_thread_pool.jobQ.empty()) { + this->mock_thread_pool.thread_iteration(); + } + const size_t actual = this->index->getAllocationSize() - before; + + EXPECT_EQ(this->index->indexSize(), block_size + 1); + EXPECT_EQ(this->index->indexCapacity(), 2 * block_size); + EXPECT_GE(estimation, actual * 0.99); + EXPECT_LE(estimation, actual * 1.01); +} + +TYPED_TEST(SQ8TieredHNSWTest, SearchByID) { this->search_by_id_test(); } + +TYPED_TEST(SQ8TieredHNSWTest, SearchByScore) { this->search_by_score_test(); } + +TYPED_TEST(SQ8TieredHNSWTest, SearchEmptyIndex) { this->search_empty_index_test(); } + +TYPED_TEST(SQ8TieredHNSWTest, Override) { this->test_override(); } + +TYPED_TEST(SQ8TieredHNSWTest, RangeQuery) { this->test_range_query(); } + +TYPED_TEST(SQ8TieredHNSWTest, GetDistanceL2) { this->test_get_distance(VecSimMetric_L2, false); } +TYPED_TEST(SQ8TieredHNSWTest, GetDistanceIP) { this->test_get_distance(VecSimMetric_IP, false); } +TYPED_TEST(SQ8TieredHNSWTest, GetDistanceMultiL2) { + this->test_get_distance(VecSimMetric_L2, true); +} +TYPED_TEST(SQ8TieredHNSWTest, GetDistanceMultiIP) { + this->test_get_distance(VecSimMetric_IP, true); +} + +TYPED_TEST(SQ8TieredHNSWTest, BatchIteratorBasic) { this->test_batch_iterator_basic(); } + +TEST(SQ8TieredHNSWTest, PhaseZeroAccessorsAndRelabelDoNotRequireBackend) { + constexpr size_t dim = 4; + constexpr size_t normalization_set_size = 4; HNSWParams hnsw_params = {.type = VecSimType_FLOAT32, - .dim = 4, + .dim = dim, + .metric = VecSimMetric_IP, + .quantType = VecSimQuant_SQ8}; + VecSimParams primary_index_params = CreateParams(hnsw_params); + tieredIndexMock mock_thread_pool; + TieredIndexParams tiered_params = { + .jobQueue = &mock_thread_pool.jobQ, + .jobQueueCtx = mock_thread_pool.ctx, + .submitCb = tieredIndexMock::submit_callback, + .primaryIndexParams = &primary_index_params, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = normalization_set_size}}}; + VecSimParams params = CreateParams(tiered_params); + auto *index = VecSimIndex_New(¶ms); + ASSERT_NE(index, nullptr); + mock_thread_pool.ctx->index_strong_ref.reset(index); + + auto *tiered_index = dynamic_cast *>(index); + ASSERT_NE(tiered_index, nullptr); + + float vector[dim] = {1.0f, 2.0f, 3.0f, 4.0f}; + ASSERT_EQ(VecSimIndex_AddVector(index, vector, 7), 1); + + EXPECT_EQ(tiered_index->indexSize(), 1); + EXPECT_EQ(tiered_index->getNumMarkedDeleted(), 0); + EXPECT_GT(tiered_index->getAllocationSize(), 0); + EXPECT_EQ(tiered_index->indexMetaDataCapacity(), + tiered_index->getFlatBufferIndex()->indexMetaDataCapacity()); + EXPECT_EQ(tiered_index->preferAdHocSearch(1, 1, true), + tiered_index->getFlatBufferIndex()->preferAdHocSearch(1, 1, true)); + tiered_index->setLastSearchMode(STANDARD_KNN); + EXPECT_EQ(tiered_index->debugInfo().commonInfo.lastMode, STANDARD_KNN); + EXPECT_NO_THROW(tiered_index->fitMemory()); + EXPECT_NO_THROW(tiered_index->runGC()); + + std::vector> stored; + tiered_index->getDataByLabel(7, stored); + ASSERT_EQ(stored.size(), 1); + EXPECT_EQ(stored[0], std::vector(vector, vector + dim)); + + EXPECT_EQ(VecSimIndex_RelabelVector(index, 7, 70), VecSimRelabel_OK); + stored.clear(); + tiered_index->getDataByLabel(7, stored); + EXPECT_TRUE(stored.empty()); + tiered_index->getDataByLabel(70, stored); + ASSERT_EQ(stored.size(), 1); + EXPECT_EQ(stored[0], std::vector(vector, vector + dim)); + + auto allocator = index->getAllocator(); + mock_thread_pool.reset_ctx(); +} + +namespace { +struct MigrationQuerySubmitContext { + tieredIndexMock *mock_thread_pool; + VecSimIndex *index; + const void *query; + size_t queued_jobs_after_migration = 0; + bool query_succeeded = false; +}; + +int executeOneMigrationThenQuery(void *, void *index_ctx, AsyncJob **jobs, JobCallback *callbacks, + size_t jobs_len) { + auto *context = static_cast(index_ctx); + const int status = + context->mock_thread_pool->submit_callback_internal(jobs, callbacks, jobs_len); + if (status != VecSim_OK) { + return status; + } + + context->mock_thread_pool->thread_iteration(); + context->queued_jobs_after_migration = context->mock_thread_pool->jobQ.size(); + + auto *reply = VecSimIndex_TopKQuery(context->index, context->query, 2, nullptr, BY_SCORE); + context->query_succeeded = + reply && reply->code == VecSim_QueryReply_OK && VecSimQueryReply_Len(reply) == 2; + VecSimQueryReply_Free(reply); + return VecSim_OK; +} +} // namespace + +TEST(SQ8TieredHNSWTest, QueryDuringSubmissionCallbackAfterOneMigration) { + constexpr size_t dim = 4; + HNSWParams hnsw_params = {.type = VecSimType_FLOAT32, + .dim = dim, .metric = VecSimMetric_L2, .quantType = VecSimQuant_SQ8}; - VecSimParams primary_params = CreateParams(hnsw_params); - // Rejection happens before the factory needs a job queue or thread pool. - TieredIndexParams tiered_params = {.primaryIndexParams = &primary_params}; + VecSimParams primary_index_params = CreateParams(hnsw_params); + tieredIndexMock mock_thread_pool; + float first_vector[dim] = {1.0f, 1.0f, 1.0f, 1.0f}; + float second_vector[dim] = {2.0f, 2.0f, 2.0f, 2.0f}; + MigrationQuerySubmitContext submit_context = {.mock_thread_pool = &mock_thread_pool, + .query = second_vector}; + TieredIndexParams tiered_params = { + .jobQueue = &mock_thread_pool.jobQ, + .jobQueueCtx = &submit_context, + .submitCb = executeOneMigrationThenQuery, + .primaryIndexParams = &primary_index_params, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = 2}}}; VecSimParams params = CreateParams(tiered_params); + auto *index = VecSimIndex_New(¶ms); + ASSERT_NE(index, nullptr); + submit_context.index = index; + mock_thread_pool.ctx->index_strong_ref.reset(index); + + ASSERT_EQ(VecSimIndex_AddVector(index, first_vector, 0), 1); + ASSERT_EQ(VecSimIndex_AddVector(index, second_vector, 1), 1); + EXPECT_EQ(submit_context.queued_jobs_after_migration, 1); + EXPECT_TRUE(submit_context.query_succeeded); + + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } - EXPECT_EQ(VecSimIndex_New(¶ms), nullptr); + auto allocator = index->getAllocator(); + mock_thread_pool.reset_ctx(); +} + +TEST(SQ8TieredHNSWTest, BatchIteratorDoesNotRepeatLabelsDuringMigrationOverlap) { + constexpr size_t dim = 4; + constexpr size_t normalization_set_size = 4; + HNSWParams hnsw_params = {.type = VecSimType_FLOAT32, + .dim = dim, + .metric = VecSimMetric_L2, + .quantType = VecSimQuant_SQ8}; + VecSimParams primary_index_params = CreateParams(hnsw_params); + tieredIndexMock mock_thread_pool; + TieredIndexParams tiered_params = { + .jobQueue = &mock_thread_pool.jobQ, + .jobQueueCtx = mock_thread_pool.ctx, + .submitCb = tieredIndexMock::submit_callback, + .primaryIndexParams = &primary_index_params, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = normalization_set_size}}}; + VecSimParams params = CreateParams(tiered_params); + auto *index = VecSimIndex_New(¶ms); + ASSERT_NE(index, nullptr); + mock_thread_pool.ctx->index_strong_ref.reset(index); + + auto *tiered_index = dynamic_cast *>(index); + ASSERT_NE(tiered_index, nullptr); + + float vectors[normalization_set_size][dim] = { + {7.0f, 1.5f, 6.66f, 1.11f}, + {2.0f, 2.22f, 2.0f, 3.33f}, + {3.0f, 3.33f, 4.0f, 4.44f}, + {4.44f, 5.66f, 5.0f, 5.55f}, + }; + for (size_t label = 0; label < normalization_set_size - 1; label++) { + ASSERT_EQ(VecSimIndex_AddVector(index, vectors[label], label), 1); + } + + std::mutex overlap_mutex; + std::condition_variable overlap_cv; + bool backend_inserted = false; + bool allow_flat_removal = false; + tiered_index->setAfterBackendInsertBeforeFlatRemovalHook([&] { + std::unique_lock lock(overlap_mutex); + backend_inserted = true; + overlap_cv.notify_all(); + overlap_cv.wait(lock, [&] { return allow_flat_removal; }); + }); + + ASSERT_EQ(VecSimIndex_AddVector(index, vectors[normalization_set_size - 1], + normalization_set_size - 1), + 1); + std::thread migration_worker([&] { mock_thread_pool.thread_iteration(); }); + bool overlap_reached = false; + { + std::unique_lock lock(overlap_mutex); + overlap_reached = + overlap_cv.wait_for(lock, std::chrono::seconds(10), [&] { return backend_inserted; }); + } + EXPECT_TRUE(overlap_reached); + + VecSimBatchIterator *iterator = VecSimBatchIterator_New(index, vectors[0], nullptr); + EXPECT_NE(iterator, nullptr); + if (iterator) { + std::unordered_set returned_labels; + size_t batch_count = 0; + while (VecSimBatchIterator_HasNext(iterator)) { + auto *batch = VecSimBatchIterator_Next(iterator, 1, BY_SCORE); + EXPECT_NE(batch, nullptr); + if (!batch) { + break; + } + for (const auto &result : batch->results) { + EXPECT_TRUE(returned_labels.insert(VecSimQueryResult_GetId(&result)).second); + } + VecSimQueryReply_Free(batch); + if (++batch_count > normalization_set_size) { + ADD_FAILURE() << "batch iterator did not deplete"; + break; + } + } + EXPECT_EQ(batch_count, normalization_set_size); + EXPECT_EQ(returned_labels.size(), normalization_set_size); + VecSimBatchIterator_Free(iterator); + } + + { + std::lock_guard lock(overlap_mutex); + allow_flat_removal = true; + } + overlap_cv.notify_all(); + migration_worker.join(); + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + auto allocator = index->getAllocator(); + mock_thread_pool.reset_ctx(); +} + +TEST(SQ8TieredHNSWTest, BatchIteratorCreatedBeforeNormalizationSeesMigratedLabels) { + constexpr size_t dim = 4; + constexpr size_t normalization_set_size = 4; + HNSWParams hnsw_params = {.type = VecSimType_FLOAT32, + .dim = dim, + .metric = VecSimMetric_L2, + .quantType = VecSimQuant_SQ8}; + VecSimParams primary_index_params = CreateParams(hnsw_params); + tieredIndexMock mock_thread_pool; + TieredIndexParams tiered_params = { + .jobQueue = &mock_thread_pool.jobQ, + .jobQueueCtx = mock_thread_pool.ctx, + .submitCb = tieredIndexMock::submit_callback, + .primaryIndexParams = &primary_index_params, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = normalization_set_size}}}; + VecSimParams params = CreateParams(tiered_params); + auto *index = VecSimIndex_New(¶ms); + ASSERT_NE(index, nullptr); + mock_thread_pool.ctx->index_strong_ref.reset(index); + + float vectors[normalization_set_size][dim] = { + {1.0f, 1.0f, 1.0f, 1.0f}, + {2.0f, 2.0f, 2.0f, 2.0f}, + {3.0f, 3.0f, 3.0f, 3.0f}, + {4.0f, 4.0f, 4.0f, 4.0f}, + }; + for (size_t label = 0; label < normalization_set_size - 1; label++) { + ASSERT_EQ(VecSimIndex_AddVector(index, vectors[label], label), 1); + } + + VecSimBatchIterator *iterator = VecSimBatchIterator_New(index, vectors[0], nullptr); + ASSERT_NE(iterator, nullptr); + + ASSERT_EQ(VecSimIndex_AddVector(index, vectors[normalization_set_size - 1], + normalization_set_size - 1), + 1); + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + std::unordered_set returned_labels; + size_t batch_count = 0; + while (VecSimBatchIterator_HasNext(iterator)) { + auto *batch = VecSimBatchIterator_Next(iterator, 1, BY_SCORE); + ASSERT_NE(batch, nullptr); + const size_t batch_len = VecSimQueryReply_Len(batch); + if (batch_len == 0) { + VecSimQueryReply_Free(batch); + break; + } + ASSERT_EQ(batch_len, 1); + for (const auto &result : batch->results) { + EXPECT_TRUE(returned_labels.insert(VecSimQueryResult_GetId(&result)).second); + } + VecSimQueryReply_Free(batch); + ASSERT_LE(++batch_count, normalization_set_size); + } + for (labelType label = 0; label < normalization_set_size - 1; label++) { + EXPECT_NE(returned_labels.find(label), returned_labels.end()); + } + VecSimBatchIterator_Free(iterator); + + auto allocator = index->getAllocator(); + mock_thread_pool.reset_ctx(); +} + +TEST(SQ8TieredHNSWTest, ConcurrentQueriesDuringNormalizationTransition) { + constexpr size_t dim = 4; + constexpr size_t normalization_set_size = 2; + HNSWParams hnsw_params = {.type = VecSimType_FLOAT32, + .dim = dim, + .metric = VecSimMetric_L2, + .quantType = VecSimQuant_SQ8}; + VecSimParams primary_index_params = CreateParams(hnsw_params); + tieredIndexMock mock_thread_pool; + TieredIndexParams tiered_params = { + .jobQueue = &mock_thread_pool.jobQ, + .jobQueueCtx = mock_thread_pool.ctx, + .submitCb = tieredIndexMock::submit_callback, + .primaryIndexParams = &primary_index_params, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = normalization_set_size}}}; + VecSimParams params = CreateParams(tiered_params); + auto *index = VecSimIndex_New(¶ms); + ASSERT_NE(index, nullptr); + mock_thread_pool.ctx->index_strong_ref.reset(index); + + auto *tiered_index = dynamic_cast *>(index); + ASSERT_NE(tiered_index, nullptr); + + float first_vector[dim] = {1.0f, 1.0f, 1.0f, 1.0f}; + float second_vector[dim] = {2.0f, 2.0f, 2.0f, 2.0f}; + float query[dim] = {1.0f, 1.0f, 1.0f, 1.0f}; + ASSERT_EQ(VecSimIndex_AddVector(index, first_vector, 0), 1); + + std::mutex transition_mutex; + std::condition_variable transition_cv; + bool replacement_entered = false; + bool allow_replacement = false; + tiered_index->setBeforeQuantizedBackendReplacementHook([&] { + std::unique_lock lock(transition_mutex); + replacement_entered = true; + transition_cv.notify_all(); + transition_cv.wait(lock, [&] { return allow_replacement; }); + }); + + std::thread writer([&] { EXPECT_EQ(VecSimIndex_AddVector(index, second_vector, 1), 1); }); + { + std::unique_lock lock(transition_mutex); + ASSERT_TRUE(transition_cv.wait_for(lock, std::chrono::seconds(10), + [&] { return replacement_entered; })); + } + + std::atomic_bool keep_reading = true; + std::atomic_int failures = 0; + std::condition_variable readers_cv; + std::mutex readers_mutex; + bool reader_completed_iteration = false; + std::thread reader([&] { + bool announced = false; + while (keep_reading.load()) { + auto *reply = VecSimIndex_TopKQuery(index, query, 2, nullptr, BY_SCORE); + if (!reply || reply->code != VecSim_QueryReply_OK || + VecSimQueryReply_Len(reply) != normalization_set_size) { + failures.fetch_add(1); + } + if (reply) { + VecSimQueryReply_Free(reply); + } + + reply = VecSimIndex_RangeQuery(index, query, 100.0, nullptr, BY_SCORE); + if (!reply || reply->code != VecSim_QueryReply_OK || + VecSimQueryReply_Len(reply) != normalization_set_size) { + failures.fetch_add(1); + } + if (reply) { + VecSimQueryReply_Free(reply); + } + + auto *iterator = VecSimBatchIterator_New(index, query, nullptr); + if (!iterator) { + failures.fetch_add(1); + } else { + if (VecSimBatchIterator_HasNext(iterator)) { + reply = VecSimBatchIterator_Next(iterator, 2, BY_SCORE); + if (!reply || reply->code != VecSim_QueryReply_OK) { + failures.fetch_add(1); + } + if (reply) { + VecSimQueryReply_Free(reply); + } + } + VecSimBatchIterator_Free(iterator); + } + + (void)VecSimIndex_IndexSize(index); + (void)VecSimIndex_DebugInfo(index); + (void)VecSimIndex_StatsInfo(index); + (void)VecSimIndex_PreferAdHocSearch(index, normalization_set_size, 1, true); + + if (!announced) { + { + std::lock_guard lock(readers_mutex); + reader_completed_iteration = true; + } + readers_cv.notify_one(); + announced = true; + } + } + }); + + { + std::unique_lock lock(readers_mutex); + EXPECT_TRUE(readers_cv.wait_for(lock, std::chrono::seconds(10), + [&] { return reader_completed_iteration; })); + } + + mock_thread_pool.init_threads(); + + { + std::lock_guard lock(transition_mutex); + allow_replacement = true; + } + transition_cv.notify_all(); + writer.join(); + mock_thread_pool.thread_pool_wait(); + + keep_reading.store(false); + reader.join(); + EXPECT_EQ(failures.load(), 0); + + auto *reply = VecSimIndex_TopKQuery(index, query, 2, nullptr, BY_SCORE); + ASSERT_NE(reply, nullptr); + EXPECT_EQ(reply->code, VecSim_QueryReply_OK); + EXPECT_EQ(VecSimQueryReply_Len(reply), normalization_set_size); + VecSimQueryReply_Free(reply); + + // Keep the allocator alive while reset_ctx releases the index's final reference. + auto allocator = index->getAllocator(); + mock_thread_pool.reset_ctx(); +} + +TEST(SQ8TieredHNSWTest, WarnsForDimensionsBelow64) { + auto previous_log_callback = VecSimIndexInterface::logCallback; + struct LogCallbackRestorer { + logCallbackFunction callback; + ~LogCallbackRestorer() { VecSimIndexInterface::logCallback = callback; } + } restore_log_callback{previous_log_callback}; + + std::vector warnings; + VecSimIndexInterface::logCallback = [](void *ctx, const char *level, const char *message) { + if (strcmp(level, VecSimCommonStrings::LOG_WARNING_STRING) == 0) { + static_cast *>(ctx)->emplace_back(message); + } + }; + + HNSWParams hnsw_params = {.type = VecSimType_FLOAT32, + .dim = 32, + .metric = VecSimMetric_L2, + .quantType = VecSimQuant_SQ8}; + VecSimParams primary_index_params = CreateParams(hnsw_params); + primary_index_params.logCtx = &warnings; + tieredIndexMock mock_thread_pool; + TieredIndexParams tiered_params = { + .jobQueue = &mock_thread_pool.jobQ, + .jobQueueCtx = mock_thread_pool.ctx, + .submitCb = tieredIndexMock::submit_callback, + .primaryIndexParams = &primary_index_params, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = 10}}}; + VecSimParams params = CreateParams(tiered_params); + auto *index = VecSimIndex_New(¶ms); + ASSERT_NE(index, nullptr); + mock_thread_pool.ctx->index_strong_ref.reset(index); + + ASSERT_EQ(warnings.size(), 1); + EXPECT_NE(warnings.front().find("SQ8 compression is not recommended for dimensions below 64"), + std::string::npos); + + auto allocator = index->getAllocator(); + mock_thread_pool.reset_ctx(); +} + +// SQ8 kernels support only FLOAT32 and FLOAT16 input vectors. +TEST(SQ8TieredHNSWTest, RejectsUnsupportedDataType) { + for (auto type : {VecSimType_FLOAT64, VecSimType_BFLOAT16, VecSimType_INT8, VecSimType_UINT8}) { + HNSWParams hnsw_params = { + .type = type, .dim = 4, .metric = VecSimMetric_L2, .quantType = VecSimQuant_SQ8}; + VecSimParams params = CreateParams(hnsw_params); + TieredIndexParams tiered_params = { + .primaryIndexParams = ¶ms, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = 10}}}; + VecSimParams vecsim_params = CreateParams(tiered_params); + EXPECT_EQ(VecSimIndex_New(&vecsim_params), nullptr) << "data type " << type; + EXPECT_EQ(EstimateInitialSize(tiered_params), SIZE_MAX) << "data type " << type; + } +} + +// Value 3 has no VecSimMetric enumerator but is within the enum's representable range, so the +// factory must reject it before dispatch. +TEST(SQ8TieredHNSWTest, RejectsOutOfRangeMetric) { + HNSWParams hnsw_params = {.type = VecSimType_FLOAT32, + .dim = 4, + .metric = static_cast(3), + .quantType = VecSimQuant_SQ8}; + VecSimParams params = CreateParams(hnsw_params); + TieredIndexParams tiered_params = { + .primaryIndexParams = ¶ms, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = 10}}}; + VecSimParams vecsim_params = CreateParams(tiered_params); + EXPECT_EQ(VecSimIndex_New(&vecsim_params), nullptr); EXPECT_EQ(EstimateInitialSize(tiered_params), SIZE_MAX); } + +// Mean-centering a FLOAT16 L2 query can lose precision or overflow when it is narrowed back to +// FLOAT16. Inner-product queries are not centered and remain supported. +TEST(SQ8TieredHNSWTest, RejectsMeanCenteredFP16L2) { + std::vector mean(4, 1.0f); + + HNSWParams l2 = {.type = VecSimType_FLOAT16, + .dim = 4, + .metric = VecSimMetric_L2, + .quantType = VecSimQuant_SQ8, + .quantParams = mean.data()}; + VecSimParams l2_params = CreateParams(l2); + TieredIndexParams l2_tiered_params = { + .primaryIndexParams = &l2_params, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = 10}}}; + VecSimParams l2_vecsim_params = CreateParams(l2_tiered_params); + EXPECT_EQ(VecSimIndex_New(&l2_vecsim_params), nullptr); + EXPECT_EQ(EstimateInitialSize(l2_tiered_params), SIZE_MAX); + + HNSWParams ip = {.type = VecSimType_FLOAT16, + .dim = 4, + .metric = VecSimMetric_IP, + .quantType = VecSimQuant_SQ8, + .quantParams = mean.data()}; + VecSimParams ip_params = CreateParams(ip); + TieredIndexParams ip_tiered_params = { + .primaryIndexParams = &ip_params, + .specificParams = {TieredHNSWParams{.QuantNormalizationSetSize = 10}}}; + VecSimParams ip_vecsim_params = CreateParams(ip_tiered_params); + VecSimIndex *ip_index = VecSimIndex_New(&ip_vecsim_params); + ASSERT_NE(ip_index, nullptr); + VecSimIndex_Free(ip_index); + EXPECT_NE(EstimateInitialSize(ip_tiered_params), SIZE_MAX); +} diff --git a/tests/unit/test_hnsw_tiered.cpp b/tests/unit/test_hnsw_tiered.cpp index f9e2b4d2d..2c7982c5f 100644 --- a/tests/unit/test_hnsw_tiered.cpp +++ b/tests/unit/test_hnsw_tiered.cpp @@ -11,9 +11,12 @@ #include "VecSim/algorithms/hnsw/hnsw_tiered.h" #include "VecSim/algorithms/hnsw/hnsw_single.h" #include "VecSim/algorithms/hnsw/hnsw_multi.h" +#include "VecSim/types/float16.h" #include "VecSim/vec_sim_debug.h" #include #include +#include +#include #include "unit_test_utils.h" #include "mock_thread_pool.h" @@ -4990,3 +4993,1482 @@ TYPED_TEST(HNSWTieredIndexTestBasic, relabelVectorDuringIngestion) { << "label " << i + relabel_offset << " does not hold its original vector"; } } +using float16 = vecsim_types::float16; + +// ------------------------------------------------------------------- +// Type definitions for parameterized tests (float32 and float16) +// ------------------------------------------------------------------- + +template +struct SQ8IndexType { + static VecSimType get_index_type() { return type; } + static bool isMulti() { return IsMulti; } + typedef DataType data_t; + typedef DistType dist_t; +}; + +// ------------------------------------------------------------------- +// Test fixture +// ------------------------------------------------------------------- + +template +class HNSWTieredIndexTestSQ8 : public ::testing::Test { +public: + using data_t = typename index_type_t::data_t; + using dist_t = typename index_type_t::dist_t; + +protected: + VecSimWriteMode original_mode; + + void SetUp() override { original_mode = VecSimIndexInterface::asyncWriteMode; } + void TearDown() override { VecSimIndexInterface::asyncWriteMode = original_mode; } + + // Create a tiered HNSW index with SQ8 quantization and accumulation phase. + TieredHNSWIndex * + CreateSQ8TieredIndex(tieredIndexMock &mock_thread_pool, size_t dim = 16, + VecSimMetric metric = VecSimMetric_IP, size_t normSetSize = 100, + size_t flat_buffer_limit = SIZE_MAX, size_t M = 16, + size_t efConstruction = 200) { + HNSWParams hnsw_params = {.type = index_type_t::get_index_type(), + .dim = dim, + .metric = metric, + .multi = index_type_t::isMulti(), + .M = M, + .efConstruction = efConstruction, + .quantType = VecSimQuant_SQ8}; + VecSimParams vecsim_params = CreateParams(hnsw_params); + TieredIndexParams tiered_params = { + .jobQueue = &mock_thread_pool.jobQ, + .jobQueueCtx = mock_thread_pool.ctx, + .submitCb = tieredIndexMock::submit_callback, + .flatBufferLimit = flat_buffer_limit, + .primaryIndexParams = &vecsim_params, + .specificParams = { + TieredHNSWParams{.swapJobThreshold = 0, .QuantNormalizationSetSize = normSetSize}}}; + auto *tiered_index = reinterpret_cast *>( + TieredFactory::NewIndex(&tiered_params)); + mock_thread_pool.ctx->index_strong_ref.reset(tiered_index); + return tiered_index; + } + + HNSWIndex *CastToHNSW(VecSimIndex *index) { + auto tiered_index = reinterpret_cast *>(index); + return tiered_index->getHNSWIndex(); + } + + // --- Accessor helpers (HNSWTieredIndexTestSQ8 is a friend of TieredHNSWIndex) --- + + bool getIsInAccumulationPhase(TieredHNSWIndex *idx) { + return !idx->backendIndex; + } + + const vecsim_stl::vector &getRunningSumVec(TieredHNSWIndex *idx) { + return idx->sqAccumulationState->runningSumVec; + } + + bool hasSQAccumulationState(TieredHNSWIndex *idx) { + return idx->sqAccumulationState.has_value(); + } + + size_t getQuantNormalizationSetSize(TieredHNSWIndex *idx) { + return idx->quantNormalizationSetSize; + } + + BruteForceIndex *getFrontendIndex(TieredHNSWIndex *idx) { + return idx->frontendIndex; + } + + VecSimIndexAbstract *getBackendIndex(TieredHNSWIndex *idx) { + return idx->backendIndex; + } + + auto &getLabelToInsertJobs(TieredHNSWIndex *idx) { + return idx->labelToInsertJobs; + } + + void callExecuteReadySwapJobs(TieredHNSWIndex *idx) { + idx->executeReadySwapJobs(); + } + + // Generate a vector with a pattern based on label. + void GenerateVectorData(data_t *output, size_t dim, float base_value) { + const float angle = base_value * 0.01f; + for (size_t i = 0; i < dim; i++) { + float val = 0.0f; + if (i == 0) { + val = std::cos(angle); + } else if (i == 1) { + val = std::sin(angle); + } + if constexpr (std::is_same_v) { + output[i] = val; + } else if constexpr (std::is_same_v) { + output[i] = vecsim_types::FP32_to_FP16(val); + } + } + } + + // Get value as float from data type. + float ToFloat(data_t val) { + if constexpr (std::is_same_v) { + return val; + } else { + return vecsim_types::FP16_to_FP32(val); + } + } +}; + +using SQ8FP32Single = SQ8IndexType; +using SQ8FP32Multi = SQ8IndexType; +using SQ8FP16Single = SQ8IndexType; +using SQ8FP16Multi = SQ8IndexType; + +using SQ8DataTypeSet = ::testing::Types; +using SQ8SingleDataTypeSet = ::testing::Types; +using SQ8MultiDataTypeSet = ::testing::Types; + +template +class HNSWTieredIndexTestSQ8Single : public HNSWTieredIndexTestSQ8 {}; + +template +class HNSWTieredIndexTestSQ8Multi : public HNSWTieredIndexTestSQ8 {}; + +TYPED_TEST_SUITE(HNSWTieredIndexTestSQ8, SQ8DataTypeSet); +TYPED_TEST_SUITE(HNSWTieredIndexTestSQ8Single, SQ8SingleDataTypeSet); +TYPED_TEST_SUITE(HNSWTieredIndexTestSQ8Multi, SQ8MultiDataTypeSet); + +// ------------------------------------------------------------------- +// Accumulation Phase Core Tests +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, AccumulationPhaseInitialization) { + // Verify that creating an SQ8 tiered index enters accumulation phase. + size_t dim = 16; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Verify accumulation phase state. + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_TRUE(this->hasSQAccumulationState(tiered_index)); + ASSERT_EQ(this->getRunningSumVec(tiered_index).size(), dim); + ASSERT_EQ(this->getQuantNormalizationSetSize(tiered_index), normSetSize); + + // Verify running sum is zero-initialized. + for (size_t i = 0; i < dim; i++) { + ASSERT_DOUBLE_EQ(this->getRunningSumVec(tiered_index)[i], 0.0); + } + + // Backend is not published until accumulation completes. + ASSERT_EQ(this->getBackendIndex(tiered_index), nullptr); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 0); + ASSERT_EQ(tiered_index->indexSize(), 0); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, RunningSumAccuracy) { + // Verify that runningSumVec correctly accumulates vector values. + size_t dim = 8; + size_t normSetSize = 10; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add vectors and verify running sum. + std::vector expected_sum(dim, 0.0); + for (size_t i = 0; i < 5; i++) { + TEST_DATA_T vec[dim]; + float base = static_cast(i + 1); + this->GenerateVectorData(vec, dim, base); + VecSimIndex_AddVector(tiered_index, vec, i); + + // Update expected sum. + for (size_t d = 0; d < dim; d++) { + expected_sum[d] += static_cast(this->ToFloat(vec[d])); + } + } + + // Verify running sum matches expected. + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + for (size_t d = 0; d < dim; d++) { + ASSERT_DOUBLE_EQ(this->getRunningSumVec(tiered_index)[d], expected_sum[d]) + << "Mismatch at dimension " << d; + } +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, QueryDuringAccumulation) { + // Search should only return flat buffer results during accumulation. + size_t dim = 8; + size_t normSetSize = 100; // High threshold so we stay in accumulation. + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add some vectors. + size_t n = 10; + for (size_t i = 0; i < n; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), n); + ASSERT_EQ(this->getBackendIndex(tiered_index), nullptr); + + // Run TopK query. + TEST_DATA_T query[dim]; + this->GenerateVectorData(query, dim, 0.0f); + auto *results = VecSimIndex_TopKQuery(tiered_index, query, 5, nullptr, BY_SCORE); + ASSERT_NE(results, nullptr); + size_t res_count = VecSimQueryReply_Len(results); + ASSERT_GT(res_count, 0); + ASSERT_LE(res_count, 5); + VecSimQueryReply_Free(results); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, RangeQueryDuringAccumulation) { + // Range queries should only use flat buffer during accumulation. + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add identical vectors (distance 0 from each other). + size_t n = 5; + for (size_t i = 0; i < n; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, 1.0f); // Same vector + VecSimIndex_AddVector(tiered_index, vec, i); + } + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + + // Range query with large radius should find all vectors. + TEST_DATA_T query[dim]; + this->GenerateVectorData(query, dim, 1.0f); + auto *results = VecSimIndex_RangeQuery(tiered_index, query, 0.01, nullptr, BY_SCORE); + ASSERT_NE(results, nullptr); + size_t res_count = VecSimQueryReply_Len(results); + ASSERT_EQ(res_count, n); + VecSimQueryReply_Free(results); +} + +// ------------------------------------------------------------------- +// Accumulation Phase Insert/Delete Tests +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, AddVectorDuringAccumulation) { + // Vectors added during accumulation go to flat buffer; no jobs submitted to queue. + size_t dim = 8; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, 1.5f); + VecSimIndex_AddVector(tiered_index, vec, 42); + + // Vector should be in flat buffer. + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 1); + ASSERT_EQ(this->getBackendIndex(tiered_index), nullptr); + ASSERT_EQ(tiered_index->indexSize(), 1); + + // Job should be created in labelToInsertJobs but NOT submitted to queue. + ASSERT_EQ(this->getLabelToInsertJobs(tiered_index).size(), 1); + ASSERT_EQ(mock_thread_pool.jobQ.size(), 0); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, DeleteVectorDuringAccumulation) { + // Deletion from flat buffer during accumulation subtracts from running sum. + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + ASSERT_EQ(this->getBackendIndex(tiered_index), nullptr); + + // Add two vectors. + TEST_DATA_T vec1[dim], vec2[dim]; + this->GenerateVectorData(vec1, dim, 1.0f); + this->GenerateVectorData(vec2, dim, 2.0f); + VecSimIndex_AddVector(tiered_index, vec1, 1); + VecSimIndex_AddVector(tiered_index, vec2, 2); + + // Record sum before deletion. + std::vector sum_before(this->getRunningSumVec(tiered_index).begin(), + this->getRunningSumVec(tiered_index).end()); + + // Delete label 1. + VecSimIndex_DeleteVector(tiered_index, 1); + + // Verify running sum was updated (subtracted vec1's values). + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + for (size_t d = 0; d < dim; d++) { + float expected = sum_before[d] - this->ToFloat(vec1[d]); + ASSERT_NEAR(this->getRunningSumVec(tiered_index)[d], expected, 1e-3f); + } + + // Verify index size. + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 1); + ASSERT_EQ(tiered_index->indexSize(), 1); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8Single, OverwriteDuringAccumulation) { + // Vector overwrite should update running sum correctly (only for single-label). + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add vector with label 1. + TEST_DATA_T vec1[dim]; + this->GenerateVectorData(vec1, dim, 1.0f); + VecSimIndex_AddVector(tiered_index, vec1, 1); + + std::vector sum_after_first(this->getRunningSumVec(tiered_index).begin(), + this->getRunningSumVec(tiered_index).end()); + + // Overwrite with different vector. + TEST_DATA_T vec2[dim]; + this->GenerateVectorData(vec2, dim, 3.0f); + VecSimIndex_AddVector(tiered_index, vec2, 1); + + // Running sum should reflect: sum - vec1 + vec2 (overwrite subtracts old + adds new). + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + for (size_t d = 0; d < dim; d++) { + float expected = sum_after_first[d] - this->ToFloat(vec1[d]) + this->ToFloat(vec2[d]); + ASSERT_NEAR(this->getRunningSumVec(tiered_index)[d], expected, 1e-3f); + } + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 1); +} + +// ------------------------------------------------------------------- +// Backend Index Initialization Tests +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, BackendCreatedAtThreshold) { + // When accumulation reaches quantNormalizationSetSize, backend is initialized. + size_t dim = 4; + size_t normSetSize = 10; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add vectors up to threshold - 1. + for (size_t i = 0; i < normSetSize - 1; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), normSetSize - 1); + + // Add the threshold-triggering vector. + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(normSetSize - 1)); + VecSimIndex_AddVector(tiered_index, vec, normSetSize - 1); + + // Accumulation phase should be over. + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_FALSE(this->hasSQAccumulationState(tiered_index)); + // Backend should be initialized (still empty since jobs haven't run). + ASSERT_NE(this->getBackendIndex(tiered_index), nullptr); + ASSERT_EQ(this->getBackendIndex(tiered_index)->indexSize(), 0); + // Flat buffer should hold all vectors. + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), normSetSize); + // All vectors should have associated insert jobs. + ASSERT_EQ(this->getLabelToInsertJobs(tiered_index).size(), normSetSize); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, MeanComputedCorrectly) { + // Verify the mean vector computed during initializeQuantizedBackend. + size_t dim = 4; + size_t normSetSize = 5; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Track expected sum. + std::vector expected_sum(dim, 0.0f); + for (size_t i = 0; i < normSetSize - 1; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i + 1)); + VecSimIndex_AddVector(tiered_index, vec, i); + for (size_t d = 0; d < dim; d++) { + expected_sum[d] += this->ToFloat(vec[d]); + } + } +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, QueryDuringPartialMigration) { + // SQ8 scores from the flat and backend indexes are not directly comparable. Verify both + // query types find an exact-match vector that is still in the flat index while migration is + // in progress. For multi-value indexes, the vector shares a label with the migrated vector + // to verify duplicate labels are merged into one result. + + size_t dim = 8; + size_t normSetSize = 3; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + for (size_t i = 0; i < normSetSize; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i * 10)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Migrate only one threshold vector, leaving the remaining vectors in the flat index. + mock_thread_pool.thread_iteration(); + ASSERT_EQ(this->getBackendIndex(tiered_index)->indexSize(), 1); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), normSetSize - 1); + + // This post-transition vector remains in the flat index while the backend has SQ8 data. + // In multi-value indexes, reuse the migrated label to exercise deduplication across indexes. + TEST_DATA_T flat_vec[dim]; + labelType flat_label = TypeParam::isMulti() ? 0 : 100; + this->GenerateVectorData(flat_vec, dim, static_cast(flat_label)); + VecSimIndex_AddVector(tiered_index, flat_vec, flat_label); + ASSERT_EQ(this->getBackendIndex(tiered_index)->indexSize(), 1); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), normSetSize); + + auto *topk_results = VecSimIndex_TopKQuery(tiered_index, flat_vec, 1, nullptr, BY_SCORE); + ASSERT_NE(topk_results, nullptr); + ASSERT_EQ(VecSimQueryReply_Len(topk_results), 1); + auto *topk_iterator = VecSimQueryReply_GetIterator(topk_results); + auto *topk_result = VecSimQueryReply_IteratorNext(topk_iterator); + ASSERT_EQ(VecSimQueryResult_GetId(topk_result), flat_label); + VecSimQueryReply_IteratorFree(topk_iterator); + VecSimQueryReply_Free(topk_results); + + auto *range_results = VecSimIndex_RangeQuery(tiered_index, flat_vec, 0.001, nullptr, BY_SCORE); + ASSERT_NE(range_results, nullptr); + ASSERT_EQ(VecSimQueryReply_Len(range_results), 1); + auto *range_iterator = VecSimQueryReply_GetIterator(range_results); + auto *range_result = VecSimQueryReply_IteratorNext(range_iterator); + ASSERT_EQ(VecSimQueryResult_GetId(range_result), flat_label); + VecSimQueryReply_IteratorFree(range_iterator); + VecSimQueryReply_Free(range_results); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, NewVectorsAfterAccumulation) { + // Vectors added after accumulation are submitted to job queue. + size_t dim = 4; + size_t normSetSize = 5; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Trigger transition. + for (size_t i = 0; i < normSetSize; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + size_t queue_size_before = mock_thread_pool.jobQ.size(); + + // Add a new vector after accumulation. + TEST_DATA_T new_vec[dim]; + this->GenerateVectorData(new_vec, dim, 99.0f); + VecSimIndex_AddVector(tiered_index, new_vec, 99); + + // New job should be submitted to queue. + ASSERT_GT(mock_thread_pool.jobQ.size(), queue_size_before); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, DeleteFromBackendAfterAccumulation) { + // Delete operations work on SQ backend after accumulation. + size_t dim = 4; + size_t normSetSize = 5; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Trigger transition. + for (size_t i = 0; i < normSetSize; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Execute all jobs to move vectors to HNSW backend. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + ASSERT_EQ(tiered_index->indexSize(), normSetSize); + + // Delete a vector (marks it for deletion in HNSW). + int deleted = VecSimIndex_DeleteVector(tiered_index, 0); + ASSERT_EQ(deleted, 1); + + // Execute repair jobs, then run swap jobs to physically remove the vector. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + this->callExecuteReadySwapJobs(tiered_index); + + ASSERT_EQ(tiered_index->indexSize(), normSetSize - 1); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, ConcurrentSearchDuringAccumulation) { + // Parallel searches should work correctly during accumulation. + size_t dim = 8; + size_t normSetSize = 1000; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add some vectors. + size_t n = 50; + for (size_t i = 0; i < n; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + + // Launch parallel searches. + std::atomic_int successful_searches(0); + size_t n_threads = 4; + auto search_fn = [&](size_t thread_id) { + TEST_DATA_T query[dim]; + this->GenerateVectorData(query, dim, static_cast(thread_id)); + auto *results = VecSimIndex_TopKQuery(tiered_index, query, 5, nullptr, BY_SCORE); + if (results && VecSimQueryReply_Len(results) > 0) { + successful_searches++; + } + VecSimQueryReply_Free(results); + }; + + std::vector threads; + for (size_t t = 0; t < n_threads; t++) { + threads.emplace_back(search_fn, t); + } + for (auto &t : threads) { + t.join(); + } + + ASSERT_EQ(successful_searches, (int)n_threads); +} + +// ------------------------------------------------------------------- +// Memory & Size Tracking Tests +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, IndexSizeDuringAccumulation) { + // indexSize() returns flat buffer size during accumulation (backend is empty). + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + for (size_t i = 0; i < 10; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(tiered_index->indexSize(), 10); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 10); + ASSERT_EQ(this->getBackendIndex(tiered_index), nullptr); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, CapacityDuringAccumulation) { + // indexCapacity() reflects flat buffer capacity during accumulation. + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, 1.0f); + VecSimIndex_AddVector(tiered_index, vec, 0); + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + // Capacity should come from flat buffer (at least DEFAULT_BLOCK_SIZE after first insert). + ASSERT_GE(tiered_index->indexCapacity(), 1); + ASSERT_EQ(tiered_index->indexCapacity(), this->getFrontendIndex(tiered_index)->indexCapacity()); +} + +// ------------------------------------------------------------------- +// Edge Cases +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, ZeroAccumulationThreshold) { + // QuantNormalizationSetSize=0 should skip accumulation phase entirely. + size_t dim = 4; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, 0 /* normSetSize=0 */); + + // Should NOT be in accumulation phase. + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Adding a vector should immediately submit job to queue. + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, 1.0f); + VecSimIndex_AddVector(tiered_index, vec, 0); + ASSERT_GT(mock_thread_pool.jobQ.size(), 0); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, AllVectorsDeletedBeforeThreshold) { + // All vectors deleted before reaching threshold - should remain in accumulation. + size_t dim = 4; + size_t normSetSize = 10; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add and delete vectors. + for (size_t i = 0; i < 5; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + for (size_t i = 0; i < 5; i++) { + VecSimIndex_DeleteVector(tiered_index, i); + } + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 0); + ASSERT_EQ(tiered_index->indexSize(), 0); + + // Running sum should be approximately zero. + for (size_t d = 0; d < dim; d++) { + ASSERT_NEAR(this->getRunningSumVec(tiered_index)[d], 0.0f, 1e-3f); + } +} + +// ------------------------------------------------------------------- +// Multi-Label Accumulation +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8Multi, AccumulationMultiLabel) { + // Multi-label: multiple vectors per label during accumulation. + size_t dim = 4; + size_t normSetSize = 10; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add multiple vectors with the same label. + labelType shared_label = 42; + for (size_t i = 0; i < 3; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i + 1)); + VecSimIndex_AddVector(tiered_index, vec, shared_label); + } + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 3); + // labelToInsertJobs should have 3 jobs for the same label. + ASSERT_EQ(this->getLabelToInsertJobs(tiered_index).at(shared_label).size(), 3); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8Multi, DeleteMultiLabelDuringAccumulation) { + // Multi-label: deleting one label removes all its vectors and adjusts the sum. + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add vectors with different labels. + TEST_DATA_T vec1[dim], vec2[dim], vec3[dim]; + this->GenerateVectorData(vec1, dim, 1.0f); + this->GenerateVectorData(vec2, dim, 2.0f); + this->GenerateVectorData(vec3, dim, 3.0f); + + VecSimIndex_AddVector(tiered_index, vec1, 10); + VecSimIndex_AddVector(tiered_index, vec2, 10); // Same label + VecSimIndex_AddVector(tiered_index, vec3, 20); // Different label + + std::vector sum_before(this->getRunningSumVec(tiered_index).begin(), + this->getRunningSumVec(tiered_index).end()); + + // Delete label 10 (should remove both vectors). + VecSimIndex_DeleteVector(tiered_index, 10); + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 1); + + // Running sum should be adjusted by subtracting both vec1 and vec2. + for (size_t d = 0; d < dim; d++) { + float expected = sum_before[d] - this->ToFloat(vec1[d]) - this->ToFloat(vec2[d]); + ASSERT_NEAR(this->getRunningSumVec(tiered_index)[d], expected, 1e-2f); + } +} + +// ------------------------------------------------------------------- +// Batch Iterator During Accumulation +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, BatchIteratorDuringAccumulation) { + // Batch iterator works during accumulation (only flat buffer results). + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + size_t n = 20; + for (size_t i = 0; i < n; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + + TEST_DATA_T query[dim]; + this->GenerateVectorData(query, dim, 0.0f); + auto *batch_iterator = VecSimBatchIterator_New(tiered_index, query, nullptr); + ASSERT_NE(batch_iterator, nullptr); + + size_t count = 0; + size_t batches = 0; + while (VecSimBatchIterator_HasNext(batch_iterator)) { + auto *batch = VecSimBatchIterator_Next(batch_iterator, 5, BY_SCORE); + ASSERT_NE(batch, nullptr); + ASSERT_GT(VecSimQueryReply_Len(batch), 0); + ASSERT_LE(VecSimQueryReply_Len(batch), 5); + count += VecSimQueryReply_Len(batch); + VecSimQueryReply_Free(batch); + ASSERT_LE(++batches, n); + } + EXPECT_EQ(count, n); + EXPECT_FALSE(VecSimBatchIterator_HasNext(batch_iterator)); + + VecSimBatchIterator_Free(batch_iterator); +} + +// ------------------------------------------------------------------- +// Index Statistics During Accumulation +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, DebugInfoDuringAccumulation) { + // Debug info during accumulation reflects flat-only state. + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + for (size_t i = 0; i < 5; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + VecSimIndexDebugInfo info = tiered_index->debugInfo(); + ASSERT_EQ(info.commonInfo.indexSize, 5); +} + +// ------------------------------------------------------------------- +// Write Mode Interactions with Accumulation +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, WriteInPlaceDuringAccumulation) { + // WriteInPlace mode is ignored during accumulation phase. + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Switch to write-in-place mode. + VecSimIndexInterface::asyncWriteMode = VecSim_WriteInPlace; + + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, 1.0f); + VecSimIndex_AddVector(tiered_index, vec, 0); + + // During accumulation, WriteInPlace is ignored - vector goes to flat buffer. + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 1); + ASSERT_EQ(this->getBackendIndex(tiered_index), nullptr); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, WriteInPlaceAfterAccumulation) { + // After accumulation, WriteInPlace inserts directly to SQ backend. + size_t dim = 4; + size_t normSetSize = 5; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Trigger transition. + for (size_t i = 0; i < normSetSize; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Execute all pending jobs (submitted by initializeQuantizedBackend). + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + // Switch to write-in-place mode. + VecSimIndexInterface::asyncWriteMode = VecSim_WriteInPlace; + + // Add vector - should go directly to HNSW backend. + TEST_DATA_T new_vec[dim]; + this->GenerateVectorData(new_vec, dim, 99.0f); + VecSimIndex_AddVector(tiered_index, new_vec, 99); + + ASSERT_EQ(this->getBackendIndex(tiered_index)->indexSize(), normSetSize + 1); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 0); +} + +// ------------------------------------------------------------------- +// Buffer Limit Interactions +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, BufferLimitWithAccumulation) { + // Flat buffer limit is respected during accumulation. + size_t dim = 4; + size_t normSetSize = 100; // High normalization set size. + size_t buffer_limit = 10; // Small buffer limit. + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, + normSetSize, buffer_limit); + + // During accumulation, buffer limit should not trigger direct insert to backend + // (since backend is not ready). The addVector code checks isInAccumulationPhase + // before checking flatBufferLimit. + for (size_t i = 0; i < 15; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + + // Should still be in accumulation phase. + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + // All vectors should be in flat buffer (accumulation overrides buffer limit behavior). + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 15); +} + +// ------------------------------------------------------------------- +// Quantization Quality Validation +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, QuantizedSearchQuality) { + // After accumulation, SQ8 backend should produce reasonable search results. + size_t dim = 16; + size_t normSetSize = 50; + size_t n = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add enough vectors to trigger transition and more. + for (size_t i = 0; i < n; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Execute all jobs. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + // Query with the same vector as label 0 - should find label 0 as nearest. + TEST_DATA_T query[dim]; + this->GenerateVectorData(query, dim, 0.0f); + auto *results = VecSimIndex_TopKQuery(tiered_index, query, 1, nullptr, BY_SCORE); + ASSERT_NE(results, nullptr); + ASSERT_EQ(VecSimQueryReply_Len(results), 1); + + auto it = VecSimQueryReply_GetIterator(results); + auto *entry = VecSimQueryReply_IteratorNext(it); + // The closest vector should be label 0 (same as query). + ASSERT_EQ(VecSimQueryResult_GetId(entry), 0); + // Distance should be very small (quantization introduces some error). + ASSERT_LT(VecSimQueryResult_GetScore(entry), 1.0); + + VecSimQueryReply_IteratorFree(it); + VecSimQueryReply_Free(results); +} + +// ------------------------------------------------------------------- +// End-to-end flow tests +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, FullFlowAsyncInsertAndSearch) { + // Full end-to-end test: accumulation -> transition -> async insert -> search. + size_t dim = 8; + size_t normSetSize = 20; + size_t total_vectors = 50; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Phase 1: Accumulation. + for (size_t i = 0; i < normSetSize; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Phase 2: Post-accumulation inserts. + for (size_t i = normSetSize; i < total_vectors; i++) { + TEST_DATA_T vec[dim]; + this->GenerateVectorData(vec, dim, static_cast(i)); + VecSimIndex_AddVector(tiered_index, vec, i); + } + + // Phase 3: Execute all jobs (including accumulation-phase jobs submitted during transition). + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + ASSERT_EQ(this->getBackendIndex(tiered_index)->indexSize(), total_vectors); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 0); + ASSERT_EQ(tiered_index->indexSize(), total_vectors); + + // Phase 4: Search. + TEST_DATA_T query[dim]; + this->GenerateVectorData(query, dim, 0.0f); + auto *results = VecSimIndex_TopKQuery(tiered_index, query, 10, nullptr, BY_SCORE); + ASSERT_NE(results, nullptr); + ASSERT_EQ(VecSimQueryReply_Len(results), 10); + VecSimQueryReply_Free(results); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, DeleteThenReinsertDuringAccumulation) { + // Delete and re-insert a vector during accumulation. + size_t dim = 4; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + // Add vector. + TEST_DATA_T vec1[dim]; + this->GenerateVectorData(vec1, dim, 1.0f); + VecSimIndex_AddVector(tiered_index, vec1, 0); + + // Delete it. + VecSimIndex_DeleteVector(tiered_index, 0); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 0); + + // Re-insert with different data. + TEST_DATA_T vec2[dim]; + this->GenerateVectorData(vec2, dim, 5.0f); + VecSimIndex_AddVector(tiered_index, vec2, 0); + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 1); + + // Running sum should only contain vec2's values (vec1 was subtracted, vec2 was added). + for (size_t d = 0; d < dim; d++) { + ASSERT_NEAR(this->getRunningSumVec(tiered_index)[d], this->ToFloat(vec2[d]), 1e-3f); + } +} + +// ------------------------------------------------------------------- +// Accumulation precision tests +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, RunningSumPrecision) { + // Values smaller than the FP32 ULP of a large running sum must not be discarded. Both values + // are exactly representable in FP32 and FP16, as is their expected sum in FP64. + size_t dim = 4; + size_t normSetSize = 100; + constexpr size_t n_small_vectors = 32; + constexpr float large_value = 32768.0f; + constexpr float small_value = 1.0f / 1024.0f; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_IP, normSetSize); + + for (size_t i = 0; i <= n_small_vectors; i++) { + TEST_DATA_T vec[dim]; + const float value = i == 0 ? large_value : small_value; + for (size_t d = 0; d < dim; d++) { + if constexpr (std::is_same_v) { + vec[d] = value; + } else { + vec[d] = vecsim_types::FP32_to_FP16(value); + } + } + VecSimIndex_AddVector(tiered_index, vec, i); + } + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + const double expected_sum = + static_cast(large_value) + n_small_vectors * static_cast(small_value); + for (size_t d = 0; d < dim; d++) { + ASSERT_DOUBLE_EQ(this->getRunningSumVec(tiered_index)[d], expected_sum) + << "Precision loss at dim " << d; + } +} + +// ------------------------------------------------------------------- +// Cosine Metric Tests +// ------------------------------------------------------------------- + +TYPED_TEST(HNSWTieredIndexTestSQ8, CosineAccumulationPhase) { + // Verify accumulation phase works correctly with Cosine metric. + // Cosine normalizes vectors before storage, so addToSum must use stored data. + size_t dim = 8; + size_t normSetSize = 10; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_Cosine, normSetSize); + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + + // Add vectors with varying magnitudes (normalization will make them unit vectors). + for (size_t i = 0; i < normSetSize - 1; i++) { + TEST_DATA_T vec[dim]; + float scale = static_cast(i + 1); // Different magnitudes + this->GenerateVectorData(vec, dim, scale); + VecSimIndex_AddVector(tiered_index, vec, i); + } + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), normSetSize - 1); + + // Verify the running sum is computed from STORED (normalized) vectors. + // After normalization, each stored vector has unit length, so each component + // should be bounded by [-1, 1]. The sum of N unit vectors has bounded magnitude. + const auto &running_sum = this->getRunningSumVec(tiered_index); + float sum_norm_sq = 0.0f; + for (size_t d = 0; d < dim; d++) { + sum_norm_sq += running_sum[d] * running_sum[d]; + } + // The magnitude of the sum of (normSetSize-1) unit vectors is at most (normSetSize-1). + float sum_norm = std::sqrt(sum_norm_sq); + ASSERT_LE(sum_norm, static_cast(normSetSize - 1) + 0.1f); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, CosineTransitionAndQuery) { + // Full flow: accumulation -> transition -> query with Cosine metric. + size_t dim = 16; + size_t normSetSize = 20; + size_t total_vectors = 50; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_Cosine, normSetSize); + + // Phase 1: Fill up to threshold to trigger transition. + for (size_t i = 0; i < normSetSize; i++) { + TEST_DATA_T vec[dim]; + // Create vectors with distinct directions by varying the first component. + for (size_t d = 0; d < dim; d++) { + float val = (d == 0) ? static_cast(i + 1) : 1.0f; + if constexpr (std::is_same_v) { + vec[d] = val; + } else { + vec[d] = vecsim_types::FP32_to_FP16(val); + } + } + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Phase 2: Add more vectors after transition. + for (size_t i = normSetSize; i < total_vectors; i++) { + TEST_DATA_T vec[dim]; + for (size_t d = 0; d < dim; d++) { + float val = (d == 0) ? static_cast(i + 1) : 1.0f; + if constexpr (std::is_same_v) { + vec[d] = val; + } else { + vec[d] = vecsim_types::FP32_to_FP16(val); + } + } + VecSimIndex_AddVector(tiered_index, vec, i); + } + + // Phase 3: Execute all jobs. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + ASSERT_EQ(tiered_index->indexSize(), total_vectors); + + // Phase 4: Query - use same direction as highest-label vector (should be nearest). + TEST_DATA_T query[dim]; + for (size_t d = 0; d < dim; d++) { + float val = (d == 0) ? static_cast(total_vectors) : 1.0f; + if constexpr (std::is_same_v) { + query[d] = val; + } else { + query[d] = vecsim_types::FP32_to_FP16(val); + } + } + + auto *results = VecSimIndex_TopKQuery(tiered_index, query, 5, nullptr, BY_SCORE); + ASSERT_NE(results, nullptr); + ASSERT_EQ(VecSimQueryReply_Len(results), 5); + + // The nearest neighbor should be the vector with highest first-component + // (most similar direction to query). + auto it = VecSimQueryReply_GetIterator(results); + auto *entry = VecSimQueryReply_IteratorNext(it); + labelType top_label = VecSimQueryResult_GetId(entry); + double top_score = VecSimQueryResult_GetScore(entry); + + // Score for Cosine is 1 - cosine_similarity. Should be close to 0 for nearest. + ASSERT_LT(top_score, 0.01); + // The top result should be one of the vectors with the largest first component. + // With FP16+SQ8, vectors 48 and 49 are nearly indistinguishable, so allow some slack. + ASSERT_GE(top_label, total_vectors - 3) + << "Expected a high-label vector (near-identical direction to query)"; + + VecSimQueryReply_IteratorFree(it); + VecSimQueryReply_Free(results); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8Single, CosineOverwriteDuringAccumulation) { + // Overwrite during accumulation with Cosine: subtractFromSum must use stored + // (normalized) data, matching what addToSum accumulated. + size_t dim = 8; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_Cosine, normSetSize); + + // Add a vector with label 1 (magnitude = ~4). + TEST_DATA_T vec1[dim]; + for (size_t d = 0; d < dim; d++) { + float val = static_cast(d + 1) * 0.5f; + if constexpr (std::is_same_v) { + vec1[d] = val; + } else { + vec1[d] = vecsim_types::FP32_to_FP16(val); + } + } + VecSimIndex_AddVector(tiered_index, vec1, 1); + + // Record running sum after first insert. + std::vector sum_after_first(this->getRunningSumVec(tiered_index).begin(), + this->getRunningSumVec(tiered_index).end()); + + // Overwrite label 1 with a completely different vector (different direction). + TEST_DATA_T vec2[dim]; + for (size_t d = 0; d < dim; d++) { + float val = (d == 0) ? 10.0f : 0.01f; + if constexpr (std::is_same_v) { + vec2[d] = val; + } else { + vec2[d] = vecsim_types::FP32_to_FP16(val); + } + } + VecSimIndex_AddVector(tiered_index, vec2, 1); + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 1); + + // After overwrite: sum should reflect only vec2's stored (normalized) data. + // Since the running sum = 0 + stored(vec1) - stored(vec1) + stored(vec2) = stored(vec2), + // the sum should be the normalized form of vec2. + const auto &running_sum = this->getRunningSumVec(tiered_index); + float sum_norm_sq = 0.0f; + for (size_t d = 0; d < dim; d++) { + sum_norm_sq += running_sum[d] * running_sum[d]; + } + float sum_norm = std::sqrt(sum_norm_sq); + // With only one normalized vector in the sum, the norm should be ~1.0. + ASSERT_NEAR(sum_norm, 1.0f, 0.05f); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, CosineDeleteDuringAccumulation) { + // Delete during accumulation with Cosine: verify running sum is correctly updated. + size_t dim = 8; + size_t normSetSize = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_Cosine, normSetSize); + + // Add two vectors. + TEST_DATA_T vec1[dim], vec2[dim]; + for (size_t d = 0; d < dim; d++) { + float v1 = static_cast(d + 1); + float v2 = static_cast(dim - d); + if constexpr (std::is_same_v) { + vec1[d] = v1; + vec2[d] = v2; + } else { + vec1[d] = vecsim_types::FP32_to_FP16(v1); + vec2[d] = vecsim_types::FP32_to_FP16(v2); + } + } + VecSimIndex_AddVector(tiered_index, vec1, 1); + VecSimIndex_AddVector(tiered_index, vec2, 2); + + // Record sum with both vectors. + std::vector sum_with_both(this->getRunningSumVec(tiered_index).begin(), + this->getRunningSumVec(tiered_index).end()); + + // Delete label 1. + VecSimIndex_DeleteVector(tiered_index, 1); + + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + ASSERT_EQ(this->getFrontendIndex(tiered_index)->indexSize(), 1); + + // After deleting vec1, sum should equal just stored(vec2). + // stored(vec2) is the normalized version of vec2, so its norm ≈ 1. + const auto &running_sum = this->getRunningSumVec(tiered_index); + float sum_norm_sq = 0.0f; + for (size_t d = 0; d < dim; d++) { + sum_norm_sq += running_sum[d] * running_sum[d]; + } + float sum_norm = std::sqrt(sum_norm_sq); + ASSERT_NEAR(sum_norm, 1.0f, 0.05f); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, CosineSearchAccuracyAfterTransition) { + // Verify that SQ8+Cosine produces reasonable search accuracy after transition. + // Compare results ordering against brute-force on the same index. + size_t dim = 32; + size_t normSetSize = 30; + size_t n = 100; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_Cosine, + normSetSize, SIZE_MAX, 16, 200); + + // Insert vectors with different directions. + for (size_t i = 0; i < n; i++) { + TEST_DATA_T vec[dim]; + for (size_t d = 0; d < dim; d++) { + // Create vectors where the i-th vector has a strong d==i%dim component. + float val = (d == (i % dim)) ? 10.0f : 1.0f / (d + 1.0f); + if constexpr (std::is_same_v) { + vec[d] = val; + } else { + vec[d] = vecsim_types::FP32_to_FP16(val); + } + } + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Execute all jobs to move vectors to backend. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + ASSERT_EQ(tiered_index->indexSize(), n); + + // Query for a vector similar to label 0 (strong component at dim 0). + TEST_DATA_T query[dim]; + for (size_t d = 0; d < dim; d++) { + float val = (d == 0) ? 10.0f : 0.5f / (d + 1.0f); + if constexpr (std::is_same_v) { + query[d] = val; + } else { + query[d] = vecsim_types::FP32_to_FP16(val); + } + } + + size_t k = 10; + auto *results = VecSimIndex_TopKQuery(tiered_index, query, k, nullptr, BY_SCORE); + ASSERT_NE(results, nullptr); + ASSERT_EQ(VecSimQueryReply_Len(results), k); + + // Verify results are sorted by increasing score (1 - cosine_sim). + auto it = VecSimQueryReply_GetIterator(results); + double prev_score = -1.0; + while (auto *entry = VecSimQueryReply_IteratorNext(it)) { + double score = VecSimQueryResult_GetScore(entry); + ASSERT_GE(score, 0.0); + ASSERT_LE(score, 2.0); // Cosine distance is in [0, 2] + ASSERT_GE(score, prev_score); + prev_score = score; + } + VecSimQueryReply_IteratorFree(it); + + // The top-1 result should be label 0 (same strong direction at dim 0). + it = VecSimQueryReply_GetIterator(results); + auto *first = VecSimQueryReply_IteratorNext(it); + // Labels with strong component at dim 0 are: 0, 32, 64, 96 + labelType top_label = VecSimQueryResult_GetId(first); + ASSERT_TRUE(top_label % dim == 0) + << "Top result label=" << top_label << " expected a vector with strong dim-0 component"; + VecSimQueryReply_IteratorFree(it); + VecSimQueryReply_Free(results); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, CosineRangeQueryAfterTransition) { + // Verify range query with Cosine metric after accumulation transition. + size_t dim = 8; + size_t normSetSize = 10; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_Cosine, normSetSize); + + // Insert parallel vectors (identical direction, different magnitudes) - should have distance 0. + for (size_t i = 0; i < normSetSize; i++) { + TEST_DATA_T vec[dim]; + float scale = static_cast(i + 1); + for (size_t d = 0; d < dim; d++) { + float val = scale * (d + 1.0f); + if constexpr (std::is_same_v) { + vec[d] = val; + } else { + vec[d] = vecsim_types::FP32_to_FP16(val); + } + } + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Execute all jobs. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + + // Query with same direction - all vectors should be at distance ~0. + TEST_DATA_T query[dim]; + for (size_t d = 0; d < dim; d++) { + float val = static_cast(d + 1); + if constexpr (std::is_same_v) { + query[d] = val; + } else { + query[d] = vecsim_types::FP32_to_FP16(val); + } + } + + // Range query with small radius should find all parallel vectors. + auto *results = VecSimIndex_RangeQuery(tiered_index, query, 0.1, nullptr, BY_SCORE); + ASSERT_NE(results, nullptr); + // All vectors have the same direction, so cosine distance ≈ 0 for all. + ASSERT_EQ(VecSimQueryReply_Len(results), normSetSize); + VecSimQueryReply_Free(results); +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, CosineMeanCorrectness) { + // Verify the mean used for SQ8 quantization is computed from normalized vectors. + // The mean of N unit vectors with the same direction should be that unit vector itself. + size_t dim = 4; + size_t normSetSize = 5; + size_t addedVectorCount = normSetSize - 1; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_Cosine, normSetSize); + + // Add parallel vectors (same direction [1,2,3,4], different magnitudes). + for (size_t i = 0; i < addedVectorCount; i++) { + TEST_DATA_T vec[dim]; + float scale = static_cast(i + 1); + for (size_t d = 0; d < dim; d++) { + float val = scale * (d + 1.0f); + if constexpr (std::is_same_v) { + vec[d] = val; + } else { + vec[d] = vecsim_types::FP32_to_FP16(val); + } + } + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_TRUE(this->getIsInAccumulationPhase(tiered_index)); + + // The running sum / addedVectorCount should be the mean of the normalized vectors. + // Since all vectors have the same direction [1,2,3,4], after normalization they're all + // [1,2,3,4]/sqrt(1+4+9+16) = [1,2,3,4]/sqrt(30). The mean is the same unit vector. + const auto &running_sum = this->getRunningSumVec(tiered_index); + float norm_factor = std::sqrt(1.0f + 4.0f + 9.0f + 16.0f); + for (size_t d = 0; d < dim; d++) { + float expected_mean = (d + 1.0f) / norm_factor; + float actual_mean = running_sum[d] / addedVectorCount; + ASSERT_NEAR(actual_mean, expected_mean, 0.02f) << "Mean mismatch at dim " << d; + } +} + +TYPED_TEST(HNSWTieredIndexTestSQ8, CosineDeleteAndReinsertAfterTransition) { + // Delete from Cosine SQ8 backend and reinsert. + size_t dim = 16; + size_t normSetSize = 10; + auto mock_thread_pool = tieredIndexMock(); + auto *tiered_index = + this->CreateSQ8TieredIndex(mock_thread_pool, dim, VecSimMetric_Cosine, normSetSize); + + // Trigger transition. + for (size_t i = 0; i < normSetSize; i++) { + TEST_DATA_T vec[dim]; + for (size_t d = 0; d < dim; d++) { + float val = (d == (i % dim)) ? 5.0f : 0.1f; + if constexpr (std::is_same_v) { + vec[d] = val; + } else { + vec[d] = vecsim_types::FP32_to_FP16(val); + } + } + VecSimIndex_AddVector(tiered_index, vec, i); + } + ASSERT_FALSE(this->getIsInAccumulationPhase(tiered_index)); + + // Move all to backend. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + ASSERT_EQ(tiered_index->indexSize(), normSetSize); + + // Delete label 0. + VecSimIndex_DeleteVector(tiered_index, 0); + // Process repair jobs. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + this->callExecuteReadySwapJobs(tiered_index); + ASSERT_EQ(tiered_index->indexSize(), normSetSize - 1); + + // Reinsert with same label, different vector. + TEST_DATA_T new_vec[dim]; + for (size_t d = 0; d < dim; d++) { + float val = (d == 0) ? 10.0f : 0.01f; + if constexpr (std::is_same_v) { + new_vec[d] = val; + } else { + new_vec[d] = vecsim_types::FP32_to_FP16(val); + } + } + VecSimIndex_AddVector(tiered_index, new_vec, 0); + + // Move to backend. + while (!mock_thread_pool.jobQ.empty()) { + mock_thread_pool.thread_iteration(); + } + ASSERT_EQ(tiered_index->indexSize(), normSetSize); + + // Query for the reinserted vector's direction. + TEST_DATA_T query[dim]; + for (size_t d = 0; d < dim; d++) { + float val = (d == 0) ? 1.0f : 0.0f; + if constexpr (std::is_same_v) { + query[d] = val; + } else { + query[d] = vecsim_types::FP32_to_FP16(val); + } + } + + auto *results = VecSimIndex_TopKQuery(tiered_index, query, 1, nullptr, BY_SCORE); + ASSERT_NE(results, nullptr); + ASSERT_EQ(VecSimQueryReply_Len(results), 1); + auto it = VecSimQueryReply_GetIterator(results); + auto *entry = VecSimQueryReply_IteratorNext(it); + // Label 0 has the strongest dim-0 component, should be top result. + ASSERT_EQ(VecSimQueryResult_GetId(entry), 0); + VecSimQueryReply_IteratorFree(it); + VecSimQueryReply_Free(results); +}