Skip to content
Closed
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
70 changes: 52 additions & 18 deletions include/svs/index/vamana/dynamic_index.h
Original file line number Diff line number Diff line change
Expand Up @@ -94,19 +94,32 @@ inline constexpr std::string_view name(SlotMetadata metadata) {

class ValidBuilder {
public:
ValidBuilder(const std::vector<SlotMetadata>& status)
: status_{status} {}
// Optional `pred` (with `translator`): an entry is also invalid if `pred(external_id)`
// is false. Invalid entries are navigated through but never returned.
ValidBuilder(
const std::vector<SlotMetadata>& status,
const IDTranslator* translator = nullptr,
const std::function<bool(size_t)>* pred = nullptr
)
: status_{status}
, translator_{translator}
, pred_{pred} {}

template <typename I>
constexpr PredicatedSearchNeighbor<I> operator()(I i, float distance) const {
bool invalid = getindex(status_, i) == SlotMetadata::Deleted;
// This neighbor should be skipped if the metadata corresponding to the given index
// marks this slot as deleted.
if (!invalid && pred_ != nullptr) {
invalid = !(*pred_)(translator_->get_external(i));
}
return PredicatedSearchNeighbor<I>(i, distance, !invalid);
}

private:
const std::vector<SlotMetadata>& status_;
const IDTranslator* translator_;
const std::function<bool(size_t)>* pred_;
};

template <graphs::MemoryGraph Graph, typename Data, typename Dist>
Expand Down Expand Up @@ -532,6 +545,9 @@ class MutableVamanaIndex {
[&](const auto is, uint64_t /*tid*/) {
for (auto i : is) {
for (size_t j = 0, jmax = getsize<1>(ids); j < jmax; ++j) {
if (ids.at(i, j) == static_cast<size_t>(-1)) {
continue; // Padding from a filtered search.
}
auto internal = lib::narrow_cast<Idx>(ids.at(i, j));
ids.at(i, j) = translate_internal_id(internal);
}
Expand Down Expand Up @@ -573,13 +589,17 @@ class MutableVamanaIndex {

// Return a `greedy_search` compatible builder for this index.
// This is an internal method, mostly used to help implement the batch iterator.
ValidBuilder internal_search_builder() const { return ValidBuilder{status_}; }
ValidBuilder internal_search_builder(const std::function<bool(size_t)>* pred = nullptr
) const {
return ValidBuilder{status_, &translator_, pred};
}

auto greedy_search_closure(
GreedySearchPrefetchParameters prefetch_parameters,
const lib::DefaultPredicate& cancel = lib::Returns(lib::Const<false>())
const lib::DefaultPredicate& cancel = lib::Returns(lib::Const<false>()),
const std::function<bool(size_t)>* pred = nullptr
) const {
return [&, prefetch_parameters](
return [&, prefetch_parameters, pred](
const auto& query, auto& accessor, auto& distance, auto& buffer
) {
// Perform the greedy search using the provided resources.
Expand All @@ -591,7 +611,7 @@ class MutableVamanaIndex {
distance,
buffer,
vamana::EntryPointInitializer<Idx>{lib::as_const_span(entry_point_)},
internal_search_builder(),
internal_search_builder(pred),
prefetch_parameters,
cancel
);
Expand Down Expand Up @@ -623,8 +643,12 @@ class MutableVamanaIndex {
QueryResultView<I> results,
const Queries& queries,
const search_parameters_type& sp,
const lib::DefaultPredicate& cancel = lib::Returns(lib::Const<false>())
const lib::DefaultPredicate& cancel = lib::Returns(lib::Const<false>()),
const std::function<bool(size_t)>* pred = nullptr
) {
// Optional `pred`: only entries whose external id passes are returned. It is called
// from all search threads. Rows with fewer passing entries than requested are
// padded with id `size_t(-1)` and the worst score.
threads::parallel_for(
threadpool_,
threads::StaticPartition{queries.size()},
Expand All @@ -637,23 +661,33 @@ class MutableVamanaIndex {
: sp.buffer_config_,
distance::comparator(distance_),
sp.search_buffer_visited_set_};
// Filtered: visit entries from a min-heap during the search.
buffer.enable_heap(pred != nullptr);

auto prefetch_parameters = GreedySearchPrefetchParameters{
sp.prefetch_lookahead_, sp.prefetch_step_};

auto scratch = extensions::per_thread_batch_search_setup(data_, distance_);

extensions::per_thread_batch_search(
data_,
buffer,
scratch,
queries,
results,
threads::UnitRange{is},
greedy_search_closure(prefetch_parameters, cancel),
*this,
cancel
);
auto run = [&](const auto& search) {
extensions::per_thread_batch_search(
data_,
buffer,
scratch,
queries,
results,
threads::UnitRange{is},
search,
*this,
cancel
);
};
auto search = greedy_search_closure(prefetch_parameters, cancel, pred);
if (pred == nullptr) {
run(search);
} else {
run(extensions::FilteredSearch<decltype(search)>{search, pred});
}
}
);

Expand Down
81 changes: 81 additions & 0 deletions include/svs/index/vamana/dynamic_search_buffer.h
Original file line number Diff line number Diff line change
Expand Up @@ -95,6 +95,14 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {
vector_type candidates_{};
// An optional visited filter.
std::optional<set_type> visited_{std::nullopt};
// Optional (`enable_heap`): while searching, entries to visit are taken from this
// min-heap (best on top), and `candidates_` keeps only the best valid entries, sorted.
// `seen_`: ids already queued in this search (exact duplicate check).
bool use_heap_ = false;
bool heap_done_ = false;
vector_type heap_{};
std::unordered_set<Idx> seen_{};
value_type popped_{};

public:
MutableBuffer() = default;
Expand Down Expand Up @@ -197,6 +205,9 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {
/// @brief Prepare the buffer for a new search operation.
void clear() {
candidates_.clear();
heap_.clear();
seen_.clear();
heap_done_ = false;
best_unvisited_ = 0;
roi_end_ = 0;
valid_ = 0;
Expand All @@ -223,6 +234,9 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {

size_t capacity() const { return candidates_.capacity(); }

/// @brief Visit entries from a min-heap during search (search window = capacity).
void enable_heap(bool enable = true) { use_heap_ = enable; }

/// @brief Return the number of valid elements currently contained in the buffer.
size_t valid() const { return valid_; }

Expand Down Expand Up @@ -267,6 +281,12 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {
/// N.B.: If `done()` evaluates to `true`, do not try to extract further candidates
/// from it using `next()`.
bool done() const {
if (heap_mode()) {
// Nothing left to visit, or the best one is worse than the worst result.
return heap_.empty() ||
(full() &&
compare_(candidates_.back().distance(), heap_.front().distance()));
}
// Until we've reached the target number of valid elements, we have to ignore
// the state of `roi_end_`.
return best_unvisited_ == ((valid() < target_valid_) ? size() : roi_end_);
Expand All @@ -282,6 +302,13 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {
/// Post-conditions:
/// * The returned neighbor will be marked as visited.
const_reference next() {
if (heap_mode()) {
std::pop_heap(heap_.begin(), heap_.end(), heap_compare());
popped_ = heap_.back();
heap_.pop_back();
popped_.set_visited();
return popped_;
}
// Get the best unvisited node
reference node = getindex(candidates_, best_unvisited_);
node.set_visited();
Expand All @@ -302,6 +329,10 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {
///
/// Therefore, sequences of `push_back` must always be followed by a call to `sort`.
void push_back(value_type neighbor) {
if (heap_mode()) {
heap_insert(neighbor);
return;
}
// Follow the contract of not appending more valid elements than the capacity
// allows.
bool valid = neighbor.valid();
Expand Down Expand Up @@ -453,6 +484,9 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {

/// Insert the given neighbor into the search buffer.
size_t insert(value_type neighbor) {
if (heap_mode()) {
return heap_insert(neighbor);
}
if (can_skip(neighbor.distance())) {
return size();
}
Expand Down Expand Up @@ -505,6 +539,9 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {

// Sort all stored elements in the buffer.
void sort() {
if (heap_mode()) {
return; // `candidates_` is kept sorted on every insert.
}
assert(valid() <= valid_capacity_);

// Put the elements in ascending order.
Expand Down Expand Up @@ -554,6 +591,13 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {

// TODO: Switch over to using iterators for the return values to avoid this.
void cleanup() {
if (heap_mode()) {
// `candidates_` already holds only valid entries; continue as a sorted list
// (e.g. for the top-up).
heap_.clear();
seen_.clear();
heap_done_ = true;
}
auto new_end =
std::remove_if(begin(), end(), [](const auto& x) { return !x.valid(); });
candidates_.resize(new_end - begin());
Expand Down Expand Up @@ -619,6 +663,43 @@ template <typename Idx, typename Cmp = std::less<>> class MutableBuffer {
}

private:
bool heap_mode() const { return use_heap_ && !heap_done_; }

// Min-heap order: the best entry on top.
auto heap_compare() const {
return [this](const value_type& a, const value_type& b) {
return compare_(b.distance(), a.distance());
};
}

size_t heap_insert(value_type neighbor) {
// Skip entries worse than the worst result when full, and ids already queued.
if (can_skip(neighbor.distance()) || !seen_.insert(neighbor.id()).second) {
return size();
}
heap_.push_back(neighbor);
std::push_heap(heap_.begin(), heap_.end(), heap_compare());
if (!neighbor.valid()) {
return size();
}
// Valid: keep it among the best `valid_capacity_` results.
auto pos = std::upper_bound(
begin(),
end(),
neighbor.distance(),
[&](const float& d, const value_type& other) {
return compare_(d, other.distance());
}
);
size_t i = pos - begin();
candidates_.insert(pos, neighbor);
if (candidates_.size() > valid_capacity_) {
candidates_.pop_back();
}
valid_ = candidates_.size();
return i;
}

/// Return how many more valid candidates exist than required.
/// If the number of valid candidates is *less* than the target, a negative number
/// is returned.
Expand Down
36 changes: 34 additions & 2 deletions include/svs/index/vamana/extensions.h
Original file line number Diff line number Diff line change
Expand Up @@ -27,6 +27,9 @@
#include "svs/lib/preprocessor.h"
#include "svs/lib/threads.h"

#include <functional>
#include <limits>

namespace svs::index::vamana::extensions {

/////
Expand Down Expand Up @@ -432,12 +435,18 @@ struct VamanaSingleSearchType {
/// with additional results.
template <typename Index, typename SearchBuffer, typename Query>
void check_and_supplement_search_buffer(
const Index& index, SearchBuffer& search_buffer, const Query& query
const Index& index,
SearchBuffer& search_buffer,
const Query& query,
const std::function<bool(size_t)>* pred = nullptr
) {
if (search_buffer.valid() < search_buffer.target_window() &&
search_buffer.valid() < index.size()) {
search_buffer.sort();
for (auto external_id : index.external_ids()) {
if (pred != nullptr && !(*pred)(external_id)) {
continue;
}
auto internal_id = index.translate_external_id(external_id);
auto dist = index.get_distance(external_id, query);
auto builder = index.internal_search_builder();
Expand All @@ -449,6 +458,15 @@ void check_and_supplement_search_buffer(
}
}

/// A search closure that also carries the filter of a filtered search.
template <typename Search> struct FilteredSearch {
Search search;
const std::function<bool(size_t)>* pred;
template <typename... Args> void operator()(Args&&... args) const {
search(std::forward<Args>(args)...);
}
};

/// Customization point object for processing single queries.
inline constexpr VamanaSingleSearchType single_search{};

Expand Down Expand Up @@ -481,7 +499,11 @@ SVS_FORCE_INLINE void svs_invoke(
search(query, accessor, distance, search_buffer);

if constexpr (Index::needs_id_translation) {
check_and_supplement_search_buffer(index, search_buffer, query);
const std::function<bool(size_t)>* pred = nullptr;
if constexpr (requires { search.pred; }) {
pred = search.pred; // Filtered search: keep filtered-out ids out of the top-up.
}
check_and_supplement_search_buffer(index, search_buffer, query, pred);
}
}

Expand Down Expand Up @@ -586,6 +608,16 @@ void svs_invoke(

// Copy back results.
for (size_t j = 0; j < num_neighbors; ++j) {
if constexpr (requires { search.pred; }) {
if (j >= search_buffer.size()) { // Filtered: fewer than k passed.
result.index(i, j) = static_cast<I>(-1);
result.distance(i, j) =
distance::comparator(distance)(0.0f, 1.0f)
? std::numeric_limits<float>::infinity()
: -std::numeric_limits<float>::infinity();
continue;
}
}
result.set(search_buffer[j], i, j);
}
}
Expand Down
Loading
Loading