mirror of
https://github.com/milvus-io/milvus.git
synced 2026-07-21 10:15:43 +00:00
enhance: add latest delete snapshot optimization (#47347)
issue: #47346 Config: N=100000, Deleted=9000, Queries=10000 Optimization OFF (slow path): 201534 us (20 us/query) Optimization ON (fast path): 5323 us (0 us/query) Speedup: 37.86x faster with optimization ON Signed-off-by: luzhang <luzhang@zilliz.com> Co-authored-by: luzhang <luzhang@zilliz.com>
This commit is contained in:
zhagnlu
@@ -24,6 +24,8 @@ std::atomic<int64_t> FILE_SLICE_SIZE(DEFAULT_INDEX_FILE_SLICE_SIZE);
|
||||
std::atomic<int64_t> EXEC_EVAL_EXPR_BATCH_SIZE(
|
||||
DEFAULT_EXEC_EVAL_EXPR_BATCH_SIZE);
|
||||
std::atomic<int64_t> DELETE_DUMP_BATCH_SIZE(DEFAULT_DELETE_DUMP_BATCH_SIZE);
|
||||
std::atomic<bool> ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION(
|
||||
DEFAULT_ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION);
|
||||
std::atomic<bool> OPTIMIZE_EXPR_ENABLED(DEFAULT_OPTIMIZE_EXPR_ENABLED);
|
||||
|
||||
std::atomic<bool> GROWING_JSON_KEY_STATS_ENABLED(
|
||||
@@ -81,6 +83,13 @@ SetDefaultEnableParquetStatsSkipIndex(bool val) {
|
||||
ENABLE_PARQUET_STATS_SKIP_INDEX.load());
|
||||
}
|
||||
|
||||
void
|
||||
SetEnableLatestDeleteSnapshotOptimization(bool val) {
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(val);
|
||||
LOG_INFO("set default enable latest delete snapshot optimization: {}",
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.load());
|
||||
}
|
||||
|
||||
void
|
||||
SetLogLevel(const char* level) {
|
||||
LOG_INFO("set log level: {}", level);
|
||||
|
||||
@@ -28,6 +28,7 @@ namespace milvus {
|
||||
extern std::atomic<int64_t> FILE_SLICE_SIZE;
|
||||
extern std::atomic<int64_t> EXEC_EVAL_EXPR_BATCH_SIZE;
|
||||
extern std::atomic<int64_t> DELETE_DUMP_BATCH_SIZE;
|
||||
extern std::atomic<bool> ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION;
|
||||
extern std::atomic<bool> OPTIMIZE_EXPR_ENABLED;
|
||||
extern std::atomic<bool> GROWING_JSON_KEY_STATS_ENABLED;
|
||||
extern std::atomic<bool> CONFIG_PARAM_TYPE_CHECK_ENABLED;
|
||||
@@ -53,6 +54,10 @@ SetDefaultConfigParamTypeCheck(bool val);
|
||||
|
||||
void
|
||||
SetDefaultEnableParquetStatsSkipIndex(bool val);
|
||||
|
||||
void
|
||||
SetEnableLatestDeleteSnapshotOptimization(bool val);
|
||||
|
||||
void
|
||||
SetLogLevel(const char* level);
|
||||
|
||||
|
||||
@@ -83,6 +83,8 @@ const int64_t DEFAULT_EXEC_EVAL_EXPR_BATCH_SIZE = 8192;
|
||||
|
||||
const int64_t DEFAULT_DELETE_DUMP_BATCH_SIZE = 10000;
|
||||
|
||||
const bool DEFAULT_ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION = true;
|
||||
|
||||
constexpr const char* COLLECTION_TTL_FIELD_KEY = "ttl_field";
|
||||
|
||||
constexpr const char* RADIUS = knowhere::meta::RADIUS;
|
||||
|
||||
@@ -80,6 +80,11 @@ SetDefaultEnableParquetStatsSkipIndex(bool val) {
|
||||
milvus::SetDefaultEnableParquetStatsSkipIndex(val);
|
||||
}
|
||||
|
||||
void
|
||||
SetEnableLatestDeleteSnapshotOptimization(bool val) {
|
||||
milvus::SetEnableLatestDeleteSnapshotOptimization(val);
|
||||
}
|
||||
|
||||
void
|
||||
SetLogLevel(const char* level) {
|
||||
milvus::SetLogLevel(level);
|
||||
|
||||
@@ -57,6 +57,9 @@ SetDefaultConfigParamTypeCheck(bool val);
|
||||
void
|
||||
SetDefaultEnableParquetStatsSkipIndex(bool val);
|
||||
|
||||
void
|
||||
SetEnableLatestDeleteSnapshotOptimization(bool val);
|
||||
|
||||
// dynamic update segcore params
|
||||
void
|
||||
SetLogLevel(const char* level);
|
||||
|
||||
@@ -21,6 +21,7 @@
|
||||
#include <folly/ConcurrentSkipList.h>
|
||||
|
||||
#include "AckResponder.h"
|
||||
#include "common/Common.h"
|
||||
#include "common/Schema.h"
|
||||
#include "common/Types.h"
|
||||
#include "segcore/Record.h"
|
||||
@@ -49,6 +50,21 @@ using SortedDeleteList =
|
||||
|
||||
static int32_t DELETE_PAIR_SIZE = sizeof(std::pair<Timestamp, Offset>);
|
||||
|
||||
// atomic snapshot for fast path query optimization
|
||||
// contains a consistent view of (max_timestamp, deleted_bitset)
|
||||
struct DeleteSnapshot {
|
||||
Timestamp max_ts{0};
|
||||
BitsetType bitset;
|
||||
|
||||
DeleteSnapshot() = default;
|
||||
DeleteSnapshot(Timestamp ts, BitsetType&& b)
|
||||
: max_ts(ts), bitset(std::move(b)) {
|
||||
}
|
||||
DeleteSnapshot(Timestamp ts, const BitsetType& b)
|
||||
: max_ts(ts), bitset(b.clone()) {
|
||||
}
|
||||
};
|
||||
|
||||
template <bool is_sealed = false>
|
||||
class DeletedRecord {
|
||||
public:
|
||||
@@ -103,7 +119,11 @@ class DeletedRecord {
|
||||
return;
|
||||
}
|
||||
|
||||
InternalPush(pks, timestamps);
|
||||
auto max_ts = InternalPush(pks, timestamps);
|
||||
|
||||
if (ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.load()) {
|
||||
UpdateLatestSnapshot(max_ts);
|
||||
}
|
||||
|
||||
bool can_dump = timestamps[0] >= max_load_timestamp_;
|
||||
if (can_dump) {
|
||||
@@ -182,6 +202,18 @@ class DeletedRecord {
|
||||
return max_timestamp;
|
||||
}
|
||||
|
||||
// update the atomic snapshot with current deleted_mask_ and max timestamp
|
||||
// this ensures a consistent view for fast path query
|
||||
void
|
||||
UpdateLatestSnapshot(Timestamp new_max_ts) {
|
||||
std::lock_guard<std::mutex> lock(snapshot_update_mutex_);
|
||||
|
||||
auto new_snapshot =
|
||||
std::make_shared<const DeleteSnapshot>(new_max_ts, deleted_mask_);
|
||||
|
||||
std::atomic_store(&latest_snapshot_, new_snapshot);
|
||||
}
|
||||
|
||||
void
|
||||
Query(BitsetTypeView& bitset,
|
||||
int64_t insert_barrier,
|
||||
@@ -193,7 +225,17 @@ class DeletedRecord {
|
||||
return;
|
||||
}
|
||||
|
||||
// try use snapshot to skip iterations
|
||||
// fast path: use atomic snapshot when query_timestamp >= max_delete_timestamp
|
||||
// this avoids traversing the SkipList entirely
|
||||
auto snapshot = std::atomic_load(&latest_snapshot_);
|
||||
if (snapshot && snapshot->max_ts > 0 &&
|
||||
query_timestamp >= snapshot->max_ts) {
|
||||
auto or_size = std::min({snapshot->bitset.size(), bitset.size()});
|
||||
bitset.inplace_or_with_count(snapshot->bitset, or_size);
|
||||
return;
|
||||
}
|
||||
|
||||
// slow path: try use snapshot to skip iterations
|
||||
bool hit_snapshot = false;
|
||||
SortedDeleteList::iterator next_iter;
|
||||
{
|
||||
@@ -361,6 +403,12 @@ class DeletedRecord {
|
||||
std::atomic<int64_t> dumped_entry_count_{0};
|
||||
// estimated memory size of DeletedRecord, only used for sealed segment
|
||||
int64_t estimated_memory_size_{0};
|
||||
|
||||
// atomic snapshot for fast path query optimization
|
||||
// when query_timestamp >= snapshot.max_ts, we can directly use the bitset
|
||||
// without traversing the SkipList
|
||||
std::shared_ptr<const DeleteSnapshot> latest_snapshot_;
|
||||
mutable std::mutex snapshot_update_mutex_;
|
||||
};
|
||||
|
||||
} // namespace milvus::segcore
|
||||
|
||||
@@ -14,6 +14,7 @@
|
||||
|
||||
#include <boost/format.hpp>
|
||||
#include <chrono>
|
||||
#include <iomanip>
|
||||
#include <iostream>
|
||||
#include <memory>
|
||||
#include <string>
|
||||
@@ -590,6 +591,177 @@ TEST(DeleteMVCC, QueryTimestampLowerThanFirstSnapshot) {
|
||||
}
|
||||
}
|
||||
|
||||
TEST(DeleteMVCC, LatestSnapshotOptimizationBenchmark) {
|
||||
using namespace milvus;
|
||||
using namespace milvus::query;
|
||||
using namespace milvus::segcore;
|
||||
|
||||
auto schema = std::make_shared<Schema>();
|
||||
auto vec_fid = schema->AddDebugField(
|
||||
"fakevec", DataType::VECTOR_FLOAT, 16, knowhere::metric::L2);
|
||||
auto i64_fid = schema->AddDebugField("age", DataType::INT64);
|
||||
schema->set_primary_field_id(i64_fid);
|
||||
|
||||
const int N = 100000; // total rows
|
||||
const int DN = 9000; // delete count (< 10000 to avoid snapshot dump)
|
||||
const int QUERY_COUNT = 10000; // number of queries
|
||||
|
||||
// Helper lambda to create search_pk function
|
||||
auto make_search_pk_func = [](InsertRecord<false>& insert_record) {
|
||||
return [&insert_record](const std::vector<PkType>& pks,
|
||||
const Timestamp* timestamps,
|
||||
std::function<void(const SegOffset offset,
|
||||
const Timestamp ts)> cb) {
|
||||
for (size_t i = 0; i < pks.size(); ++i) {
|
||||
auto timestamp = timestamps[i];
|
||||
auto offsets = insert_record.search_pk(pks[i], timestamp);
|
||||
for (auto offset : offsets) {
|
||||
cb(offset, timestamp);
|
||||
}
|
||||
}
|
||||
};
|
||||
};
|
||||
|
||||
// Helper lambda to setup insert record
|
||||
auto setup_insert_record = [&](InsertRecord<false>& insert_record,
|
||||
std::vector<int64_t>& age_data,
|
||||
std::vector<Timestamp>& tss) {
|
||||
for (int i = 0; i < N; ++i) {
|
||||
age_data[i] = i;
|
||||
tss[i] = i;
|
||||
insert_record.insert_pk(age_data[i], i);
|
||||
}
|
||||
auto insert_offset = insert_record.reserved.fetch_add(N);
|
||||
insert_record.timestamps_.set_data_raw(insert_offset, tss.data(), N);
|
||||
auto field_data = insert_record.get_data_base(i64_fid);
|
||||
field_data->set_data_raw(insert_offset, age_data.data(), N);
|
||||
insert_record.ack_responder_.AddSegment(insert_offset,
|
||||
insert_offset + N);
|
||||
};
|
||||
|
||||
// Prepare delete data
|
||||
std::vector<Timestamp> delete_ts(DN);
|
||||
std::vector<PkType> delete_pk(DN);
|
||||
for (int i = 0; i < DN; ++i) {
|
||||
delete_pk[i] = i; // delete first DN pks
|
||||
delete_ts[i] = N + i; // ts after all inserts
|
||||
}
|
||||
|
||||
Timestamp query_timestamp = N + DN + 100; // after all deletes
|
||||
int64_t insert_barrier = N;
|
||||
|
||||
// ============ Setup for Optimization OFF (no latest_snapshot_) ============
|
||||
InsertRecord<false> insert_record_off(*schema, N);
|
||||
DeletedRecord<false> delete_record_off(
|
||||
&insert_record_off, make_search_pk_func(insert_record_off), 0);
|
||||
|
||||
std::vector<int64_t> age_data_off(N);
|
||||
std::vector<Timestamp> tss_off(N);
|
||||
setup_insert_record(insert_record_off, age_data_off, tss_off);
|
||||
|
||||
// Push deletes with optimization OFF - no latest_snapshot_ created
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(false);
|
||||
delete_record_off.StreamPush(delete_pk, delete_ts.data());
|
||||
ASSERT_EQ(DN, delete_record_off.size());
|
||||
|
||||
// ============ Setup for Optimization ON (has latest_snapshot_) ============
|
||||
InsertRecord<false> insert_record_on(*schema, N);
|
||||
DeletedRecord<false> delete_record_on(
|
||||
&insert_record_on, make_search_pk_func(insert_record_on), 0);
|
||||
|
||||
std::vector<int64_t> age_data_on(N);
|
||||
std::vector<Timestamp> tss_on(N);
|
||||
setup_insert_record(insert_record_on, age_data_on, tss_on);
|
||||
|
||||
// Push deletes with optimization ON - latest_snapshot_ created
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(true);
|
||||
delete_record_on.StreamPush(delete_pk, delete_ts.data());
|
||||
ASSERT_EQ(DN, delete_record_on.size());
|
||||
|
||||
// ============ Verify correctness ============
|
||||
BitsetType ref_bitmap(insert_barrier);
|
||||
BitsetTypeView ref_view(ref_bitmap);
|
||||
delete_record_off.Query(ref_view, insert_barrier, query_timestamp);
|
||||
|
||||
BitsetType fast_bitmap(insert_barrier);
|
||||
BitsetTypeView fast_view(fast_bitmap);
|
||||
delete_record_on.Query(fast_view, insert_barrier, query_timestamp);
|
||||
|
||||
// Verify both produce same result
|
||||
for (int i = 0; i < DN; ++i) {
|
||||
ASSERT_TRUE(ref_view[i]) << "Row " << i << " should be deleted (OFF)";
|
||||
ASSERT_TRUE(fast_view[i]) << "Row " << i << " should be deleted (ON)";
|
||||
}
|
||||
for (int i = DN; i < N; ++i) {
|
||||
ASSERT_FALSE(ref_view[i])
|
||||
<< "Row " << i << " should NOT be deleted (OFF)";
|
||||
ASSERT_FALSE(fast_view[i])
|
||||
<< "Row " << i << " should NOT be deleted (ON)";
|
||||
}
|
||||
std::cout << "Correctness verified: OFF and ON produce same results"
|
||||
<< std::endl;
|
||||
|
||||
// ============ Benchmark: Optimization OFF (slow path) ============
|
||||
BitsetType res_bitmap(insert_barrier);
|
||||
auto start_off = std::chrono::steady_clock::now();
|
||||
for (int i = 0; i < QUERY_COUNT; ++i) {
|
||||
res_bitmap.reset();
|
||||
BitsetTypeView res_view(res_bitmap);
|
||||
delete_record_off.Query(res_view, insert_barrier, query_timestamp);
|
||||
}
|
||||
auto end_off = std::chrono::steady_clock::now();
|
||||
auto duration_off = std::chrono::duration_cast<std::chrono::microseconds>(
|
||||
end_off - start_off)
|
||||
.count();
|
||||
std::cout << "[OFF] result bitmap count: " << res_bitmap.count()
|
||||
<< std::endl;
|
||||
|
||||
// ============ Benchmark: Optimization ON (fast path) ============
|
||||
auto start_on = std::chrono::steady_clock::now();
|
||||
for (int i = 0; i < QUERY_COUNT; ++i) {
|
||||
res_bitmap.reset();
|
||||
BitsetTypeView res_view(res_bitmap);
|
||||
delete_record_on.Query(res_view, insert_barrier, query_timestamp);
|
||||
}
|
||||
auto end_on = std::chrono::steady_clock::now();
|
||||
auto duration_on =
|
||||
std::chrono::duration_cast<std::chrono::microseconds>(end_on - start_on)
|
||||
.count();
|
||||
std::cout << "[ON] result bitmap count: " << res_bitmap.count()
|
||||
<< std::endl;
|
||||
|
||||
// ============ Print results ============
|
||||
std::cout << "============================================" << std::endl;
|
||||
std::cout << "Latest Delete Snapshot Optimization Benchmark" << std::endl;
|
||||
std::cout << "============================================" << std::endl;
|
||||
std::cout << "Config: N=" << N << ", Deleted=" << DN
|
||||
<< ", Queries=" << QUERY_COUNT << std::endl;
|
||||
std::cout << "--------------------------------------------" << std::endl;
|
||||
std::cout << "Optimization OFF (slow path): " << duration_off << " us ("
|
||||
<< duration_off / QUERY_COUNT << " us/query)" << std::endl;
|
||||
std::cout << "Optimization ON (fast path): " << duration_on << " us ("
|
||||
<< duration_on / QUERY_COUNT << " us/query)" << std::endl;
|
||||
std::cout << "--------------------------------------------" << std::endl;
|
||||
|
||||
if (duration_on < duration_off) {
|
||||
double speedup = static_cast<double>(duration_off) / duration_on;
|
||||
std::cout << "Speedup: " << std::fixed << std::setprecision(2)
|
||||
<< speedup << "x faster with optimization ON" << std::endl;
|
||||
} else {
|
||||
double slowdown = static_cast<double>(duration_on) / duration_off;
|
||||
std::cout << "WARNING: " << std::fixed << std::setprecision(2)
|
||||
<< slowdown << "x slower with optimization ON" << std::endl;
|
||||
}
|
||||
std::cout << "============================================" << std::endl;
|
||||
|
||||
// Verify optimization is faster
|
||||
EXPECT_LT(duration_on, duration_off)
|
||||
<< "Optimization should be faster than slow path";
|
||||
|
||||
// Restore default
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(true);
|
||||
}
|
||||
|
||||
TEST(DeleteMVCC, SnapshotDumpProgress) {
|
||||
using namespace milvus;
|
||||
using namespace milvus::query;
|
||||
|
||||
@@ -13,6 +13,7 @@
|
||||
#include <gtest/gtest.h>
|
||||
|
||||
#include "cachinglayer/Utils.h"
|
||||
#include "common/Common.h"
|
||||
#include "common/Types.h"
|
||||
#include "index/IndexFactory.h"
|
||||
#include "knowhere/version.h"
|
||||
@@ -1235,6 +1236,143 @@ TEST(Sealed, RealCount) {
|
||||
ASSERT_EQ(0, segment->get_real_count());
|
||||
}
|
||||
|
||||
// Test for DeletedRecord atomic snapshot optimization fast path
|
||||
// When query_timestamp >= max_delete_timestamp, we can directly use the snapshot bitset
|
||||
TEST(Sealed, DeleteSnapshotOptimizationFastPath) {
|
||||
// Save original value and ensure optimization is enabled
|
||||
bool original_value = ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.load();
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(true);
|
||||
|
||||
auto schema = std::make_shared<Schema>();
|
||||
auto pk = schema->AddDebugField("pk", DataType::INT64);
|
||||
schema->set_primary_field_id(pk);
|
||||
|
||||
int64_t N = 100;
|
||||
auto dataset = DataGen(schema, N);
|
||||
auto pks = dataset.get_col<int64_t>(pk);
|
||||
auto segment = CreateSealedWithFieldDataLoaded(schema, dataset);
|
||||
|
||||
// Delete some records with timestamps
|
||||
int64_t delete_count = 10;
|
||||
auto del_ids = GenPKs(pks.begin(), pks.begin() + delete_count);
|
||||
Timestamp delete_ts = 1000;
|
||||
auto del_tss = GenTss(delete_count, delete_ts);
|
||||
auto status = segment->Delete(delete_count, del_ids.get(), del_tss.data());
|
||||
ASSERT_TRUE(status.ok());
|
||||
|
||||
// Query with timestamp >= max_delete_ts should use fast path
|
||||
BitsetType bitset1(N, false);
|
||||
auto bitset_view1 = BitsetTypeView(bitset1);
|
||||
Timestamp query_ts_fast =
|
||||
delete_ts + delete_count + 100; // >= max_delete_ts
|
||||
segment->mask_with_delete(bitset_view1, N, query_ts_fast);
|
||||
ASSERT_EQ(bitset1.count(), delete_count);
|
||||
|
||||
// Query with timestamp < max_delete_ts should use slow path
|
||||
BitsetType bitset2(N, false);
|
||||
auto bitset_view2 = BitsetTypeView(bitset2);
|
||||
Timestamp query_ts_slow =
|
||||
delete_ts + 5; // < max_delete_ts, covers some but not all
|
||||
segment->mask_with_delete(bitset_view2, N, query_ts_slow);
|
||||
// Should have fewer deletions visible since query_ts is in the middle
|
||||
ASSERT_LT(bitset2.count(), delete_count);
|
||||
ASSERT_GT(bitset2.count(), 0);
|
||||
|
||||
// Restore original value
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(original_value);
|
||||
}
|
||||
|
||||
// Test for DeletedRecord atomic snapshot optimization disabled
|
||||
TEST(Sealed, DeleteSnapshotOptimizationDisabled) {
|
||||
// Save original value and disable optimization
|
||||
bool original_value = ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.load();
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(false);
|
||||
|
||||
auto schema = std::make_shared<Schema>();
|
||||
auto pk = schema->AddDebugField("pk", DataType::INT64);
|
||||
schema->set_primary_field_id(pk);
|
||||
|
||||
int64_t N = 100;
|
||||
auto dataset = DataGen(schema, N);
|
||||
auto pks = dataset.get_col<int64_t>(pk);
|
||||
auto segment = CreateSealedWithFieldDataLoaded(schema, dataset);
|
||||
|
||||
// Delete some records
|
||||
int64_t delete_count = 10;
|
||||
auto del_ids = GenPKs(pks.begin(), pks.begin() + delete_count);
|
||||
Timestamp delete_ts = 1000;
|
||||
auto del_tss = GenTss(delete_count, delete_ts);
|
||||
auto status = segment->Delete(delete_count, del_ids.get(), del_tss.data());
|
||||
ASSERT_TRUE(status.ok());
|
||||
|
||||
// Query should still work correctly via slow path
|
||||
BitsetType bitset(N, false);
|
||||
auto bitset_view = BitsetTypeView(bitset);
|
||||
Timestamp query_ts = delete_ts + delete_count + 100;
|
||||
segment->mask_with_delete(bitset_view, N, query_ts);
|
||||
ASSERT_EQ(bitset.count(), delete_count);
|
||||
|
||||
// Restore original value
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(original_value);
|
||||
}
|
||||
|
||||
// Test for multiple sequential deletes updating snapshot correctly
|
||||
TEST(Sealed, DeleteSnapshotMultipleDeletes) {
|
||||
bool original_value = ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.load();
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(true);
|
||||
|
||||
auto schema = std::make_shared<Schema>();
|
||||
auto pk = schema->AddDebugField("pk", DataType::INT64);
|
||||
schema->set_primary_field_id(pk);
|
||||
|
||||
int64_t N = 100;
|
||||
auto dataset = DataGen(schema, N);
|
||||
auto pks = dataset.get_col<int64_t>(pk);
|
||||
auto segment = CreateSealedWithFieldDataLoaded(schema, dataset);
|
||||
|
||||
// First batch of deletes
|
||||
int64_t delete_count1 = 5;
|
||||
auto del_ids1 = GenPKs(pks.begin(), pks.begin() + delete_count1);
|
||||
Timestamp delete_ts1 = 1000;
|
||||
auto del_tss1 = GenTss(delete_count1, delete_ts1);
|
||||
auto status =
|
||||
segment->Delete(delete_count1, del_ids1.get(), del_tss1.data());
|
||||
ASSERT_TRUE(status.ok());
|
||||
|
||||
// Query after first delete batch
|
||||
BitsetType bitset1(N, false);
|
||||
auto bitset_view1 = BitsetTypeView(bitset1);
|
||||
Timestamp query_ts1 = delete_ts1 + delete_count1 + 100;
|
||||
segment->mask_with_delete(bitset_view1, N, query_ts1);
|
||||
ASSERT_EQ(bitset1.count(), delete_count1);
|
||||
|
||||
// Second batch of deletes
|
||||
int64_t delete_count2 = 5;
|
||||
auto del_ids2 = GenPKs(pks.begin() + delete_count1,
|
||||
pks.begin() + delete_count1 + delete_count2);
|
||||
Timestamp delete_ts2 = 2000;
|
||||
auto del_tss2 = GenTss(delete_count2, delete_ts2);
|
||||
status = segment->Delete(delete_count2, del_ids2.get(), del_tss2.data());
|
||||
ASSERT_TRUE(status.ok());
|
||||
|
||||
// Query after second delete batch - should see all deletes
|
||||
BitsetType bitset2(N, false);
|
||||
auto bitset_view2 = BitsetTypeView(bitset2);
|
||||
Timestamp query_ts2 = delete_ts2 + delete_count2 + 100;
|
||||
segment->mask_with_delete(bitset_view2, N, query_ts2);
|
||||
ASSERT_EQ(bitset2.count(), delete_count1 + delete_count2);
|
||||
|
||||
// Query with timestamp between batches - should only see first batch
|
||||
BitsetType bitset3(N, false);
|
||||
auto bitset_view3 = BitsetTypeView(bitset3);
|
||||
Timestamp query_ts_between =
|
||||
delete_ts1 + delete_count1 + 50; // Between batch1 and batch2
|
||||
segment->mask_with_delete(bitset_view3, N, query_ts_between);
|
||||
ASSERT_EQ(bitset3.count(), delete_count1);
|
||||
|
||||
ENABLE_LATEST_DELETE_SNAPSHOT_OPTIMIZATION.store(original_value);
|
||||
}
|
||||
|
||||
TEST(Sealed, GetVector) {
|
||||
auto dim = 4;
|
||||
auto N = ROW_COUNT;
|
||||
|
||||
@@ -553,6 +553,15 @@ func SetupCoreConfigChangelCallback() {
|
||||
return nil
|
||||
})
|
||||
|
||||
paramtable.Get().QueryNodeCfg.EnableLatestDeleteSnapshotOptimization.RegisterCallback(func(ctx context.Context, key, oldValue, newValue string) error {
|
||||
enable, err := strconv.ParseBool(newValue)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
UpdateEnableLatestDeleteSnapshotOptimization(enable)
|
||||
return nil
|
||||
})
|
||||
|
||||
paramtable.Get().QueryNodeCfg.ExprResCacheEnabled.RegisterCallback(func(ctx context.Context, key, oldValue, newValue string) error {
|
||||
enable, err := strconv.ParseBool(newValue)
|
||||
if err != nil {
|
||||
|
||||
@@ -117,6 +117,9 @@ func doInitQueryNodeOnce(ctx context.Context) error {
|
||||
cDeleteDumpBatchSize := C.int64_t(paramtable.Get().QueryNodeCfg.DeleteDumpBatchSize.GetAsInt64())
|
||||
C.SetDefaultDeleteDumpBatchSize(cDeleteDumpBatchSize)
|
||||
|
||||
cEnableLatestDeleteSnapshotOptimization := C.bool(paramtable.Get().QueryNodeCfg.EnableLatestDeleteSnapshotOptimization.GetAsBool())
|
||||
C.SetEnableLatestDeleteSnapshotOptimization(cEnableLatestDeleteSnapshotOptimization)
|
||||
|
||||
cOptimizeExprEnabled := C.bool(paramtable.Get().CommonCfg.EnabledOptimizeExpr.GetAsBool())
|
||||
C.SetDefaultOptimizeExprEnable(cOptimizeExprEnabled)
|
||||
|
||||
|
||||
@@ -87,3 +87,7 @@ func UpdateDefaultConfigParamTypeCheck(enable bool) {
|
||||
func UpdateDefaultEnableParquetStatsSkipIndex(enable bool) {
|
||||
C.SetDefaultEnableParquetStatsSkipIndex(C.bool(enable))
|
||||
}
|
||||
|
||||
func UpdateEnableLatestDeleteSnapshotOptimization(enable bool) {
|
||||
C.SetEnableLatestDeleteSnapshotOptimization(C.bool(enable))
|
||||
}
|
||||
|
||||
@@ -3372,6 +3372,9 @@ type queryNodeConfig struct {
|
||||
// delete snapshot dump batch size
|
||||
DeleteDumpBatchSize ParamItem `refreshable:"false"`
|
||||
|
||||
// delete snapshot optimization
|
||||
EnableLatestDeleteSnapshotOptimization ParamItem `refreshable:"true"`
|
||||
|
||||
// expr cache
|
||||
ExprResCacheEnabled ParamItem `refreshable:"false"`
|
||||
ExprResCacheCapacityBytes ParamItem `refreshable:"false"`
|
||||
@@ -4484,6 +4487,15 @@ user-task-polling:
|
||||
}
|
||||
p.DeleteDumpBatchSize.Init(base.mgr)
|
||||
|
||||
p.EnableLatestDeleteSnapshotOptimization = ParamItem{
|
||||
Key: "queryNode.segcore.enableLatestDeleteSnapshotOptimization",
|
||||
Version: "2.6.11",
|
||||
DefaultValue: "true",
|
||||
Doc: "Enable latest delete snapshot optimization for fast path query when query_timestamp >= max_delete_timestamp.",
|
||||
Export: false,
|
||||
}
|
||||
p.EnableLatestDeleteSnapshotOptimization.Init(base.mgr)
|
||||
|
||||
// expr cache
|
||||
p.ExprResCacheEnabled = ParamItem{
|
||||
Key: "queryNode.exprCache.enabled",
|
||||
|
||||
Reference in New Issue
Block a user