Skip to content

perf(sindi): batch posting normalization for incremental add #2693

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
CharlesXu-HQ wants to merge 1 commit into
base: main
Choose a base branch
from
Open

perf(sindi): batch posting normalization for incremental add #2693

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
2 changes: 1 addition & 1 deletion src/algorithm/inner_index_interface.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -698,7 +698,7 @@ class InnerIndexInterface {
protected:
std::atomic<uint64_t> total_count_{0};

std::atomic<uint64_t> current_memory_usage_{0};
mutable std::atomic<uint64_t> current_memory_usage_{0};
mutable std::shared_mutex memory_usage_mutex_{};

bool has_raw_vector_{false};
Expand Down
52 changes: 36 additions & 16 deletions src/algorithm/sindi/sindi.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -425,12 +425,7 @@ SINDI::Add(const DatasetPtr& base) {
throw VsagException(ErrorType::UNSUPPORTED_INDEX_OPERATION,
"SINDI DMQ rerank does not support incremental Add");
}
const auto previous_window_count = mutable_term_datacell_->GetWindowCount();
auto failed_ids = this->add(base, true);
if (mutable_term_datacell_->GetWindowCount() != previous_window_count) {
this->cal_memory_usage();
}
return failed_ids;
return this->add(base, true);
}

std::vector<int64_t>
Expand All @@ -455,9 +450,9 @@ SINDI::add(const DatasetPtr& base, bool sort_affected_windows) {
Vector<uint32_t> pruned_ids(allocator_);
Vector<float> pruned_vals(allocator_);
Vector<uint32_t> remapped_ids(allocator_);
const auto first_affected_window = cur_element_count_ / window_size_;
// This remains -1 when every input vector is rejected, so post-insert loops are no-ops.
int64_t last_affected_window = -1;
const auto first_affected_window =
static_cast<uint32_t>(cur_element_count_.load(std::memory_order_relaxed) / window_size_);
std::optional<uint32_t> last_affected_window;
std::vector<SparseVector> rerank_vectors;
if (use_reorder_) {
rerank_vectors.reserve(data_num);
Expand Down Expand Up @@ -507,16 +502,32 @@ SINDI::add(const DatasetPtr& base, bool sort_affected_windows) {
if (use_reorder_) {
rerank_vectors.push_back(sparse_vectors[i]);
}
last_affected_window = cur_element_count_ / window_size_;
last_affected_window = static_cast<uint32_t>(
cur_element_count_.load(std::memory_order_relaxed) / window_size_);
cur_element_count_++;
}
if (not rerank_vectors.empty()) {
rerank_flat_->BatchInsertVector(rerank_vectors.data(),
static_cast<InnerIdType>(rerank_vectors.size()));
}
if (sort_affected_windows) {
for (int64_t window = first_affected_window; window <= last_affected_window; ++window) {
mutable_term_datacell_->SortByValue(static_cast<uint32_t>(window));
if (sort_affected_windows && last_affected_window.has_value()) {
bool posting_state_changed = false;
const auto current_count =
static_cast<uint64_t>(cur_element_count_.load(std::memory_order_relaxed));
for (uint32_t window = first_affected_window;; ++window) {
const auto window_end =
(static_cast<uint64_t>(window) + 1) * static_cast<uint64_t>(window_size_);
if (window_end <= current_count) {
posting_state_changed |= mutable_term_datacell_->NormalizeDirtyPostings(window);
} else {
posting_state_changed |= mutable_term_datacell_->FinalizeInsertBatch(window);
}
if (window == last_affected_window.value()) {
break;
}
}
if (posting_state_changed) {
this->cal_memory_usage();
}
}
return failed_ids;
Expand Down Expand Up @@ -1144,7 +1155,7 @@ SINDI::UseTermListsHeapInsert(const SINDISearchParameter& search_param,
}

void
SINDI::cal_memory_usage() {
SINDI::cal_memory_usage() const {
auto memory = sizeof(SINDI);
if (term_datacell_ != nullptr) {
memory += term_datacell_->GetMemoryUsage();
Expand All @@ -1163,9 +1174,17 @@ SINDI::cal_memory_usage() {
this->current_memory_usage_.store(static_cast<int64_t>(memory));
}

void
Comment thread
CharlesXu-HQ marked this conversation as resolved.
SINDI::normalize_dirty_postings_for_serialization() const {
if (mutable_term_datacell_ != nullptr && mutable_term_datacell_->NormalizeDirtyPostings()) {
this->cal_memory_usage();
}
Comment thread
CharlesXu-HQ marked this conversation as resolved.
}

void
SINDI::Serialize(StreamWriter& writer) const {
std::shared_lock rlock(this->global_mutex_);
std::scoped_lock wlock(this->global_mutex_);
this->normalize_dirty_postings_for_serialization();

if (cur_element_count_ == 0) {
const auto cur_element_count = cur_element_count_.load();
Expand Down Expand Up @@ -1303,9 +1322,10 @@ SINDI::serialize_windows(StreamWriter& writer) const {

void
SINDI::serialize_streaming_body(StreamWriter& writer) const {
std::shared_lock rlock(this->global_mutex_);
std::scoped_lock wlock(this->global_mutex_);
CHECK_ARGUMENT(not immutable_enabled_,
"immutable SINDI runtime does not support SerializeStreaming");
this->normalize_dirty_postings_for_serialization();

auto windows_tag = static_cast<uint32_t>(StreamSerializationTag::SINDI_WINDOWS);
auto label_tag = static_cast<uint32_t>(StreamSerializationTag::LABEL_TABLE);
Expand Down
7 changes: 6 additions & 1 deletion src/algorithm/sindi/sindi.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -269,7 +269,12 @@ class SINDI : public InnerIndexInterface {

/// Recalculate and cache the memory-usage counter.
void
cal_memory_usage();
cal_memory_usage() const;

/// Canonicalize mutable posting runs before writing the sorted-posting format.
/// The caller must hold global_mutex_ exclusively.
void
normalize_dirty_postings_for_serialization() const;

/**
* @brief Compact a sparse vector's dim-ids into the remapped space
Expand Down
9 changes: 7 additions & 2 deletions src/algorithm/sindi/sindi_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -500,7 +500,7 @@ TEST_CASE("SINDI term prune keeps highest stored values after build", "[ut][SIND
}
}

TEST_CASE("SINDI sorts incremental partial windows", "[ut][SINDI]") {
TEST_CASE("SINDI defers incremental partial window normalization", "[ut][SINDI]") {
auto allocator = SafeAllocator::FactoryDefaultAllocator();
IndexCommonParam common_param;
common_param.allocator_ = allocator;
Expand Down Expand Up @@ -536,7 +536,7 @@ TEST_CASE("SINDI sorts incremental partial windows", "[ut][SINDI]") {
auto appended = Dataset::Make();
appended->NumElements(1)->SparseVectors(&appended_vector)->Ids(&appended_label)->Owner(false);
REQUIRE(index.Add(appended).empty());
REQUIRE(SINDITestAccess::MutableTermIsSorted(index, 0, term));
REQUIRE_FALSE(SINDITestAccess::MutableTermIsSorted(index, 0, term));

float query_value = 1.0F;
SparseVector query_vector{1, &term, &query_value};
Expand All @@ -545,6 +545,11 @@ TEST_CASE("SINDI sorts incremental partial windows", "[ut][SINDI]") {
const auto search_parameters = R"({"sindi": {"n_candidate": 1, "term_retain_threshold": 1}})";
REQUIRE(index.KnnSearch(query, 1, search_parameters, nullptr)->GetIds()[0] == appended_label);

std::stringstream stream;
IOStreamWriter writer(stream);
REQUIRE_NOTHROW(index.Serialize(writer));
REQUIRE(SINDITestAccess::MutableTermIsSorted(index, 0, term));

appended_value = 3.0F;
appended_label = 13;
REQUIRE(index.Add(appended).empty());
Expand Down
29 changes: 20 additions & 9 deletions src/algorithm/sindi_v2/sindi_v2.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -514,8 +514,8 @@ SINDIV2::Add(const DatasetPtr& base) {
Vector<uint32_t> pruned_ids(allocator_);
Vector<float> pruned_vals(allocator_);
Vector<uint32_t> remapped_ids(allocator_);
const auto first_affected_window = cur_element_count_ / window_size_;
int64_t last_affected_window = -1;
const auto first_affected_window = static_cast<uint32_t>(cur_element_count_ / window_size_);
std::optional<uint32_t> last_affected_window;
std::vector<SparseVector> dmq_rerank_vectors;
if (use_reorder_ && rerank_type_ == SPARSE_RERANK_TYPE_DMQ8) {
dmq_rerank_vectors.reserve(data_num);
Expand Down Expand Up @@ -581,7 +581,7 @@ SINDIV2::Add(const DatasetPtr& base) {
{sparse_vectors + i, static_cast<InnerIdType>(cur_element_count_), {}});
}

last_affected_window = cur_element_count_ / window_size_;
last_affected_window = static_cast<uint32_t>(cur_element_count_ / window_size_);
cur_element_count_++;
}

Expand All @@ -593,8 +593,20 @@ SINDIV2::Add(const DatasetPtr& base) {
write_rerank_flat_with_layout(rerank_flat_, rerank_layout_records, rerank_layout_);
}

for (int64_t window = first_affected_window; window <= last_affected_window; ++window) {
mutable_term_datacell->SortByValue(static_cast<uint32_t>(window));
if (last_affected_window.has_value()) {
const auto current_count = static_cast<uint64_t>(cur_element_count_);
for (uint32_t window = first_affected_window;; ++window) {
const auto window_end =
(static_cast<uint64_t>(window) + 1) * static_cast<uint64_t>(window_size_);
if (window_end <= current_count) {
mutable_term_datacell->NormalizeDirtyPostings(window);
} else {
mutable_term_datacell->FinalizeInsertBatch(window);
}
if (window == last_affected_window.value()) {
break;
}
}
}
this->cal_memory_usage();
return failed_ids;
Expand Down Expand Up @@ -1063,7 +1075,7 @@ SINDIV2::UseTermListsHeapInsert(const SINDIV2SearchParameter& search_param) cons
}

void
SINDIV2::cal_memory_usage() {
SINDIV2::cal_memory_usage() const {
auto memory = sizeof(SINDIV2);
if (term_datacell_ != nullptr) {
memory += term_datacell_->GetMemoryUsage();
Expand Down Expand Up @@ -1106,9 +1118,8 @@ SINDIV2::Serialize(StreamWriter& writer) const {

if (term_datacell_ != nullptr &&
std::dynamic_pointer_cast<MutableSindiTermDataCell>(term_datacell_) != nullptr) {
const auto mutable_datacell = this->get_mutable_term_datacell();
for (uint32_t window = 0; window < mutable_datacell->GetWindowCount(); ++window) {
mutable_datacell->SortByValue(window);
if (this->get_mutable_term_datacell()->NormalizeDirtyPostings()) {
this->cal_memory_usage();
}
}

Expand Down
2 changes: 1 addition & 1 deletion src/algorithm/sindi_v2/sindi_v2.h
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,7 @@ class SINDIV2 : public InnerIndexInterface {
get_min_max_window_id(const FilterPtr& filter) const;

void
cal_memory_usage();
cal_memory_usage() const;

SparseVector
sort_and_prune_sparse_vector_for_build(const SparseVector& input,
Expand Down
4 changes: 2 additions & 2 deletions src/algorithm/sindi_v2/sindi_v2_test.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -267,7 +267,7 @@ TEST_CASE("SINDIV2 term prune keeps highest stored values after build", "[ut][SI
}
}

TEST_CASE("SINDIV2 sorts incremental partial windows", "[ut][SINDIV2]") {
TEST_CASE("SINDIV2 defers incremental partial window normalization", "[ut][SINDIV2]") {
auto allocator = SafeAllocator::FactoryDefaultAllocator();
IndexCommonParam common_param;
common_param.allocator_ = allocator;
Expand Down Expand Up @@ -305,7 +305,7 @@ TEST_CASE("SINDIV2 sorts incremental partial windows", "[ut][SINDIV2]") {
auto appended = Dataset::Make();
appended->NumElements(1)->SparseVectors(&appended_vector)->Ids(&appended_label)->Owner(false);
REQUIRE(index.Add(appended).empty());
REQUIRE(SINDIV2TestAccess::MutableTermIsSorted(index, 0, term));
REQUIRE_FALSE(SINDIV2TestAccess::MutableTermIsSorted(index, 0, term));

float query_value = 1.0F;
SparseVector query_vector{1, &term, &query_value};
Expand Down
Loading
Loading