milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Fix deleted data is still visible (#24849) 

Signed-off-by: yah01 <yang.cen@zilliz.com>
pull/24921/head
yah01 2023-06-16 17:16:41 +08:00 committed by GitHub
parent 50b1dcfe25
commit a413842e38
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
19 changed files with 112 additions and 144 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

55
internal/core/src/segcore/DeletedRecord.h
View File

@ -12,10 +12,15 @@
#pragma once
#include <memory>
#include <mutex>
#include <shared_mutex>
#include <tuple>
#include <utility>
#include <vector>
#include "AckResponder.h"
#include "common/Schema.h"
#include "common/Types.h"
#include "segcore/Record.h"
#include "ConcurrentVector.h"
@ -76,15 +81,55 @@ struct DeletedRecord {
lru_ = std::move(new_entry);
}
public:
std::atomic<int64_t> reserved = 0;
AckResponder ack_responder_;
ConcurrentVector<Timestamp> timestamps_;
ConcurrentVector<PkType> pks_;
void
push(const std::vector<PkType>& pks, const Timestamp* timestamps) {
std::lock_guard lck(buffer_mutex_);
auto size = pks.size();
ssize_t divide_point = 0;
auto n = n_.load();
// Truncate the overlapping prefix
if (n > 0) {
auto last = timestamps_[n - 1];
divide_point =
std::lower_bound(timestamps, timestamps + size, last + 1) -
timestamps;
}
// All these delete records have been applied
if (divide_point == size) {
return;
}
size -= divide_point;
pks_.set_data_raw(n, pks.data() + divide_point, size);
timestamps_.set_data_raw(n, timestamps + divide_point, size);
n_ += size;
}
const ConcurrentVector<Timestamp>&
timestamps() const {
return timestamps_;
}
const ConcurrentVector<PkType>&
pks() const {
return pks_;
}
int64_t
size() const {
return n_.load();
}
private:
std::shared_ptr<TmpBitmap> lru_;
std::shared_mutex shared_mutex_;
std::shared_mutex buffer_mutex_;
std::atomic<int64_t> n_ = 0;
ConcurrentVector<Timestamp> timestamps_;
ConcurrentVector<PkType> pks_;
};
inline auto

10
internal/core/src/segcore/InsertRecord.h
View File

@ -332,6 +332,16 @@ struct InsertRecord {
fields_data_.erase(field_id);
}
const ConcurrentVector<Timestamp>&
timestamps() const {
return timestamps_;
}
int64_t
size() const {
return ack_responder_.GetAck();
}
private:
// std::vector<std::unique_ptr<VectorBase>> fields_data_;
std::unordered_map<FieldId, std::unique_ptr<VectorBase>> fields_data_{};

4
internal/core/src/segcore/Record.h
View File

@ -16,9 +16,9 @@ namespace milvus::segcore {
template <typename RecordType>
inline int64_t
get_barrier(const RecordType& record, Timestamp timestamp) {
auto& vec = record.timestamps_;
auto& vec = record.timestamps();
int64_t beg = 0;
int64_t end = record.ack_responder_.GetAck();
int64_t end = record.size();
while (beg < end) {
auto mid = (beg + end) / 2;
if (vec[mid] <= timestamp) {

20
internal/core/src/segcore/SegmentGrowingImpl.cpp
View File

@ -32,12 +32,6 @@ SegmentGrowingImpl::PreInsert(int64_t size) {
return reserved_begin;
}
int64_t
SegmentGrowingImpl::PreDelete(int64_t size) {
auto reserved_begin = deleted_record_.reserved.fetch_add(size);
return reserved_begin;
}
void
SegmentGrowingImpl::mask_with_delete(BitsetType& bitset,
int64_t ins_barrier,
@ -145,11 +139,7 @@ SegmentGrowingImpl::Delete(int64_t reserved_begin,
}
// step 2: fill delete record
deleted_record_.timestamps_.set_data_raw(
reserved_begin, sort_timestamps.data(), size);
deleted_record_.pks_.set_data_raw(reserved_begin, sort_pks.data(), size);
deleted_record_.ack_responder_.AddSegment(reserved_begin,
reserved_begin + size);
deleted_record_.push(sort_pks, sort_timestamps.data());
return Status::OK();
}
@ -159,7 +149,7 @@ SegmentGrowingImpl::GetMemoryUsageInBytes() const {
auto chunk_rows = segcore_config_.get_chunk_rows();
int64_t ins_n = upper_align(insert_record_.reserved, chunk_rows);
total_bytes += ins_n * (schema_->get_total_sizeof() + 16 + 1);
int64_t del_n = upper_align(deleted_record_.reserved, chunk_rows);
int64_t del_n = upper_align(deleted_record_.size(), chunk_rows);
total_bytes += del_n * (16 * 2);
return total_bytes;
}
@ -181,11 +171,7 @@ SegmentGrowingImpl::LoadDeletedRecord(const LoadDeletedRecordInfo& info) {
auto timestamps = reinterpret_cast<const Timestamp*>(info.timestamps);
// step 2: fill pks and timestamps
auto reserved_begin = deleted_record_.reserved.fetch_add(size);
deleted_record_.pks_.set_data_raw(reserved_begin, pks.data(), size);
deleted_record_.timestamps_.set_data_raw(reserved_begin, timestamps, size);
deleted_record_.ack_responder_.AddSegment(reserved_begin,
reserved_begin + size);
deleted_record_.push(pks, timestamps);
}
SpanBase

5
internal/core/src/segcore/SegmentGrowingImpl.h
View File

@ -49,9 +49,6 @@ class SegmentGrowingImpl : public SegmentGrowing {
const Timestamp* timestamps,
const InsertData* insert_data) override;
int64_t
PreDelete(int64_t size) override;
// TODO: add id into delete log, possibly bitmap
Status
Delete(int64_t reserved_offset,
@ -132,7 +129,7 @@ class SegmentGrowingImpl : public SegmentGrowing {
int64_t
get_deleted_count() const override {
return deleted_record_.ack_responder_.GetAck();
return deleted_record_.size();
}
int64_t

4
internal/core/src/segcore/SegmentInterface.h
View File

@ -71,8 +71,8 @@ class SegmentInterface {
virtual int64_t
get_real_count() const = 0;
virtual int64_t
PreDelete(int64_t size) = 0;
// virtual int64_t
// PreDelete(int64_t size) = 0;
virtual Status
Delete(int64_t reserved_offset,

41
internal/core/src/segcore/SegmentSealedImpl.cpp
View File

@ -48,12 +48,6 @@ get_bit(const BitsetType& bitset, FieldId field_id) {
return bitset[pos];
}
int64_t
SegmentSealedImpl::PreDelete(int64_t size) {
auto reserved_begin = deleted_record_.reserved.fetch_add(size);
return reserved_begin;
}
void
SegmentSealedImpl::LoadIndex(const LoadIndexInfo& info) {
// print(info);
@ -294,29 +288,7 @@ SegmentSealedImpl::LoadDeletedRecord(const LoadDeletedRecordInfo& info) {
auto timestamps = reinterpret_cast<const Timestamp*>(info.timestamps);
// step 2: fill pks and timestamps
ssize_t n = deleted_record_.ack_responder_.GetAck();
ssize_t divide_point = 0;
// Truncate the overlapping prefix
if (n > 0) {
auto last = deleted_record_.timestamps_[n - 1];
divide_point =
std::lower_bound(timestamps, timestamps + size, last + 1) -
timestamps;
}
// All these delete records have been loaded
if (divide_point == size) {
return;
}
size -= divide_point;
auto reserved_begin = deleted_record_.reserved.fetch_add(size);
deleted_record_.pks_.set_data_raw(
reserved_begin, pks.data() + divide_point, size);
deleted_record_.timestamps_.set_data_raw(
reserved_begin, timestamps + divide_point, size);
deleted_record_.ack_responder_.AddSegment(reserved_begin,
reserved_begin + size);
deleted_record_.push(pks, timestamps);
}
// internal API: support scalar index only
@ -393,7 +365,7 @@ SegmentSealedImpl::get_row_count() const {
int64_t
SegmentSealedImpl::get_deleted_count() const {
std::shared_lock lck(mutex_);
return deleted_record_.ack_responder_.GetAck();
return deleted_record_.size();
}
const Schema&
@ -855,7 +827,7 @@ SegmentSealedImpl::search_ids(const IdArray& id_array,
}
Status
SegmentSealedImpl::Delete(int64_t reserved_offset,
SegmentSealedImpl::Delete(int64_t reserved_offset, // deprecated
int64_t size,
const IdArray* ids,
const Timestamp* timestamps_raw) {
@ -879,11 +851,8 @@ SegmentSealedImpl::Delete(int64_t reserved_offset,
sort_timestamps[i] = t;
sort_pks[i] = pk;
}
deleted_record_.timestamps_.set_data_raw(
reserved_offset, sort_timestamps.data(), size);
deleted_record_.pks_.set_data_raw(reserved_offset, sort_pks.data(), size);
deleted_record_.ack_responder_.AddSegment(reserved_offset,
reserved_offset + size);
deleted_record_.push(sort_pks, sort_timestamps.data());
return Status::OK();
}

3
internal/core/src/segcore/SegmentSealedImpl.h
View File

@ -98,9 +98,6 @@ class SegmentSealedImpl : public SegmentSealed {
std::string
debug() const override;
int64_t
PreDelete(int64_t size) override;
Status
Delete(int64_t reserved_offset,
int64_t size,

4
internal/core/src/segcore/Utils.h
View File

@ -101,8 +101,8 @@ get_deleted_bitmap(int64_t del_barrier,
// Avoid invalid calculations when there are a lot of repeated delete pks
std::unordered_map<PkType, Timestamp> delete_timestamps;
for (auto del_index = start; del_index < end; ++del_index) {
auto pk = delete_record.pks_[del_index];
auto timestamp = delete_record.timestamps_[del_index];
auto pk = delete_record.pks()[del_index];
auto timestamp = delete_record.timestamps()[del_index];
delete_timestamps[pk] = timestamp > delete_timestamps[pk]
? timestamp

9
internal/core/src/segcore/segment_c.cpp
View File

@ -207,7 +207,7 @@ PreInsert(CSegmentInterface c_segment, int64_t size, int64_t* offset) {
CStatus
Delete(CSegmentInterface c_segment,
int64_t reserved_offset,
int64_t reserved_offset, // deprecated
int64_t size,
const uint8_t* ids,
const uint64_t ids_size,
@ -225,13 +225,6 @@ Delete(CSegmentInterface c_segment,
}
}
int64_t
PreDelete(CSegmentInterface c_segment, int64_t size) {
auto segment = static_cast<milvus::segcore::SegmentInterface*>(c_segment);
return segment->PreDelete(size);
}
////////////////////////////// interfaces for sealed segment //////////////////////////////
CStatus
LoadFieldData(CSegmentInterface c_segment,

2
internal/core/src/segcore/segment_c.h
View File

@ -111,8 +111,6 @@ Delete(CSegmentInterface c_segment,
const uint64_t ids_size,
const uint64_t* timestamps);
int64_t
PreDelete(CSegmentInterface c_segment, int64_t size);
#ifdef __cplusplus
}
#endif

45
internal/core/unittest/test_c_api.cpp
View File

@ -356,8 +356,7 @@ TEST(CApiTest, DeleteTest) {
auto delete_data = serialize(ids.get());
uint64_t delete_timestamps[] = {0, 0, 0};
auto offset = PreDelete(segment, 3);
auto offset = 0;
auto del_res = Delete(segment,
offset,
3,
@ -398,7 +397,7 @@ TEST(CApiTest, MultiDeleteGrowingSegment) {
delete_pks.end());
auto delete_data = serialize(ids.get());
std::vector<uint64_t> delete_timestamps(1, dataset.timestamps_[N - 1]);
offset = PreDelete(segment, 1);
offset = 0;
auto del_res = Delete(segment,
offset,
1,
@ -420,10 +419,10 @@ TEST(CApiTest, MultiDeleteGrowingSegment) {
plan->plan_node_->predicate_ = std::move(term_expr);
std::vector<FieldId> target_field_ids{FieldId(100), FieldId(101)};
plan->field_ids_ = target_field_ids;
auto max_ts = dataset.timestamps_[N - 1] + 10;
CRetrieveResult retrieve_result;
res = Retrieve(
segment, plan.get(), {}, dataset.timestamps_[N - 1], &retrieve_result);
res = Retrieve(segment, plan.get(), {}, max_ts, &retrieve_result);
ASSERT_EQ(res.error_code, Success);
auto query_result = std::make_unique<proto::segcore::RetrieveResults>();
auto suc = query_result->ParseFromArray(retrieve_result.proto_blob,
@ -440,8 +439,7 @@ TEST(CApiTest, MultiDeleteGrowingSegment) {
retrive_pks,
proto::plan::GenericValue::kInt64Val);
plan->plan_node_->predicate_ = std::move(term_expr);
res = Retrieve(
segment, plan.get(), {}, dataset.timestamps_[N - 1], &retrieve_result);
res = Retrieve(segment, plan.get(), {}, max_ts, &retrieve_result);
ASSERT_EQ(res.error_code, Success);
suc = query_result->ParseFromArray(retrieve_result.proto_blob,
retrieve_result.proto_size);
@ -455,7 +453,8 @@ TEST(CApiTest, MultiDeleteGrowingSegment) {
ids->mutable_int_id()->mutable_data()->Add(delete_pks.begin(),
delete_pks.end());
delete_data = serialize(ids.get());
offset = PreDelete(segment, 1);
delete_timestamps[0]++;
offset = 0;
del_res = Delete(segment,
offset,
1,
@ -465,8 +464,7 @@ TEST(CApiTest, MultiDeleteGrowingSegment) {
ASSERT_EQ(del_res.error_code, Success);
// retrieve pks in {2}
res = Retrieve(
segment, plan.get(), {}, dataset.timestamps_[N - 1], &retrieve_result);
res = Retrieve(segment, plan.get(), {}, max_ts, &retrieve_result);
ASSERT_EQ(res.error_code, Success);
suc = query_result->ParseFromArray(retrieve_result.proto_blob,
retrieve_result.proto_size);
@ -535,7 +533,7 @@ TEST(CApiTest, MultiDeleteSealedSegment) {
delete_pks.end());
auto delete_data = serialize(ids.get());
std::vector<uint64_t> delete_timestamps(1, dataset.timestamps_[N - 1]);
auto offset = PreDelete(segment, 1);
auto offset = 0;
auto del_res = Delete(segment,
offset,
1,
@ -557,10 +555,10 @@ TEST(CApiTest, MultiDeleteSealedSegment) {
plan->plan_node_->predicate_ = std::move(term_expr);
std::vector<FieldId> target_field_ids{FieldId(100), FieldId(101)};
plan->field_ids_ = target_field_ids;
auto max_ts = dataset.timestamps_[N - 1] + 10;
CRetrieveResult retrieve_result;
res = Retrieve(
segment, plan.get(), {}, dataset.timestamps_[N - 1], &retrieve_result);
res = Retrieve(segment, plan.get(), {}, max_ts, &retrieve_result);
ASSERT_EQ(res.error_code, Success);
auto query_result = std::make_unique<proto::segcore::RetrieveResults>();
auto suc = query_result->ParseFromArray(retrieve_result.proto_blob,
@ -577,8 +575,7 @@ TEST(CApiTest, MultiDeleteSealedSegment) {
retrive_pks,
proto::plan::GenericValue::kInt64Val);
plan->plan_node_->predicate_ = std::move(term_expr);
res = Retrieve(
segment, plan.get(), {}, dataset.timestamps_[N - 1], &retrieve_result);
res = Retrieve(segment, plan.get(), {}, max_ts, &retrieve_result);
ASSERT_EQ(res.error_code, Success);
suc = query_result->ParseFromArray(retrieve_result.proto_blob,
retrieve_result.proto_size);
@ -592,7 +589,8 @@ TEST(CApiTest, MultiDeleteSealedSegment) {
ids->mutable_int_id()->mutable_data()->Add(delete_pks.begin(),
delete_pks.end());
delete_data = serialize(ids.get());
offset = PreDelete(segment, 1);
delete_timestamps[0]++;
offset = 0;
del_res = Delete(segment,
offset,
1,
@ -602,8 +600,7 @@ TEST(CApiTest, MultiDeleteSealedSegment) {
ASSERT_EQ(del_res.error_code, Success);
// retrieve pks in {2}
res = Retrieve(
segment, plan.get(), {}, dataset.timestamps_[N - 1], &retrieve_result);
res = Retrieve(segment, plan.get(), {}, max_ts, &retrieve_result);
ASSERT_EQ(res.error_code, Success);
suc = query_result->ParseFromArray(retrieve_result.proto_blob,
retrieve_result.proto_size);
@ -683,7 +680,7 @@ TEST(CApiTest, DeleteRepeatedPksFromGrowingSegment) {
auto delete_data = serialize(ids.get());
std::vector<uint64_t> delete_timestamps(3, dataset.timestamps_[N - 1]);
offset = PreDelete(segment, 3);
offset = 0;
auto del_res = Delete(segment,
offset,
3,
@ -788,7 +785,7 @@ TEST(CApiTest, DeleteRepeatedPksFromSealedSegment) {
auto delete_data = serialize(ids.get());
std::vector<uint64_t> delete_timestamps(3, dataset.timestamps_[N - 1]);
auto offset = PreDelete(segment, 3);
auto offset = 0;
auto del_res = Delete(segment,
offset,
@ -846,7 +843,7 @@ TEST(CApiTest, InsertSamePkAfterDeleteOnGrowingSegment) {
auto delete_data = serialize(ids.get());
std::vector<uint64_t> delete_timestamps(3, dataset.timestamps_[N - 1]);
offset = PreDelete(segment, 3);
offset = 0;
auto del_res = Delete(segment,
offset,
@ -967,7 +964,7 @@ TEST(CApiTest, InsertSamePkAfterDeleteOnSealedSegment) {
auto delete_data = serialize(ids.get());
std::vector<uint64_t> delete_timestamps(3, dataset.timestamps_[4]);
auto offset = PreDelete(segment, 3);
auto offset = 0;
auto del_res = Delete(segment,
offset,
@ -1232,7 +1229,7 @@ TEST(CApiTest, GetDeletedCountTest) {
auto delete_data = serialize(ids.get());
uint64_t delete_timestamps[] = {0, 0, 0};
auto offset = PreDelete(segment, 3);
auto offset = 0;
auto del_res = Delete(segment,
offset,
@ -1309,7 +1306,7 @@ TEST(CApiTest, GetRealCount) {
dataset.timestamps_[N - 1] + 2,
dataset.timestamps_[N - 1] + 3};
auto del_offset = PreDelete(segment, 3);
auto del_offset = 0;
auto del_res = Delete(segment,
del_offset,

13
internal/core/unittest/test_growing.cpp
View File

@ -27,8 +27,7 @@ TEST(Growing, DeleteCount) {
auto segment = CreateGrowingSegment(schema, empty_index_meta);
int64_t c = 10;
auto offset = segment->PreDelete(c);
ASSERT_EQ(offset, 0);
auto offset = 0;
Timestamp begin_ts = 100;
auto tss = GenTss(c, begin_ts);
@ -47,8 +46,7 @@ TEST(Growing, RealCount) {
auto segment = CreateGrowingSegment(schema, empty_index_meta);
int64_t c = 10;
auto offset = segment->PreInsert(c);
ASSERT_EQ(offset, 0);
auto offset = 0;
auto dataset = DataGen(schema, c);
auto pks = dataset.get_col<int64_t>(pk);
segment->Insert(offset,
@ -62,8 +60,7 @@ TEST(Growing, RealCount) {
// delete half.
auto half = c / 2;
auto del_offset1 = segment->PreDelete(half);
ASSERT_EQ(del_offset1, 0);
auto del_offset1 = 0;
auto del_ids1 = GenPKs(pks.begin(), pks.begin() + half);
auto del_tss1 = GenTss(half, c);
auto status =
@ -72,7 +69,7 @@ TEST(Growing, RealCount) {
ASSERT_EQ(c - half, segment->get_real_count());
// delete duplicate.
auto del_offset2 = segment->PreDelete(half);
auto del_offset2 = segment->get_deleted_count();
ASSERT_EQ(del_offset2, half);
auto del_tss2 = GenTss(half, c + half);
status =
@ -81,7 +78,7 @@ TEST(Growing, RealCount) {
ASSERT_EQ(c - half, segment->get_real_count());
// delete all.
auto del_offset3 = segment->PreDelete(c);
auto del_offset3 = segment->get_deleted_count();
ASSERT_EQ(del_offset3, half * 2);
auto del_ids3 = GenPKs(pks.begin(), pks.end());
auto del_tss3 = GenTss(c, c + half * 2);

2
internal/core/unittest/test_retrieve.cpp
View File

@ -389,7 +389,7 @@ TEST(Retrieve, Delete) {
auto ids = std::make_unique<IdArray>();
ids->mutable_int_id()->mutable_data()->Add(new_pks.begin(), new_pks.end());
std::vector<idx_t> new_timestamps{10, 10, 10, 10, 10, 10};
auto reserved_offset = segment->PreDelete(new_count);
auto reserved_offset = segment->get_deleted_count();
ASSERT_EQ(reserved_offset, row_count);
segment->Delete(reserved_offset,
new_count,

10
internal/core/unittest/test_sealed.cpp
View File

@ -753,7 +753,7 @@ TEST(Sealed, Delete) {
new_ids->mutable_int_id()->mutable_data()->Add(new_pks.begin(),
new_pks.end());
std::vector<idx_t> new_timestamps{10, 10, 10};
auto reserved_offset = segment->PreDelete(new_count);
auto reserved_offset = segment->get_deleted_count();
ASSERT_EQ(reserved_offset, row_count);
segment->Delete(reserved_offset,
new_count,
@ -1009,7 +1009,7 @@ TEST(Sealed, DeleteCount) {
auto segment = CreateSealedSegment(schema);
int64_t c = 10;
auto offset = segment->PreDelete(c);
auto offset = segment->get_deleted_count();
ASSERT_EQ(offset, 0);
Timestamp begin_ts = 100;
@ -1040,7 +1040,7 @@ TEST(Sealed, RealCount) {
// delete half.
auto half = c / 2;
auto del_offset1 = segment->PreDelete(half);
auto del_offset1 = segment->get_deleted_count();
ASSERT_EQ(del_offset1, 0);
auto del_ids1 = GenPKs(pks.begin(), pks.begin() + half);
auto del_tss1 = GenTss(half, c);
@ -1050,7 +1050,7 @@ TEST(Sealed, RealCount) {
ASSERT_EQ(c - half, segment->get_real_count());
// delete duplicate.
auto del_offset2 = segment->PreDelete(half);
auto del_offset2 = segment->get_deleted_count();
ASSERT_EQ(del_offset2, half);
auto del_tss2 = GenTss(half, c + half);
status =
@ -1059,7 +1059,7 @@ TEST(Sealed, RealCount) {
ASSERT_EQ(c - half, segment->get_real_count());
// delete all.
auto del_offset3 = segment->PreDelete(c);
auto del_offset3 = segment->get_deleted_count();
ASSERT_EQ(del_offset3, half * 2);
auto del_ids3 = GenPKs(pks.begin(), pks.end());
auto del_tss3 = GenTss(c, c + half * 2);

1
internal/core/unittest/test_segcore.cpp
View File

@ -59,7 +59,6 @@ TEST(SegmentCoreTest, NormalDistributionTest) {
auto [raw_data, timestamps, uids] = generate_data(N);
auto segment = CreateGrowingSegment(schema, empty_index_meta);
segment->PreInsert(N);
segment->PreDelete(N);
}
// Test insert column-based data

10
internal/core/unittest/test_utils.cpp
View File

@ -71,10 +71,7 @@ TEST(Util, GetDeleteBitmap) {
// test case delete pk1(ts = 0) -> insert repeated pk1 (ts = {1 ... N}) -> query (ts = N)
std::vector<Timestamp> delete_ts = {0};
std::vector<PkType> delete_pk = {1};
auto offset = delete_record.reserved.fetch_add(1);
delete_record.timestamps_.set_data_raw(offset, delete_ts.data(), 1);
delete_record.pks_.set_data_raw(offset, delete_pk.data(), 1);
delete_record.ack_responder_.AddSegment(offset, offset + 1);
delete_record.push(delete_pk, delete_ts.data());
auto query_timestamp = tss[N - 1];
auto del_barrier = get_barrier(delete_record, query_timestamp);
@ -89,10 +86,7 @@ TEST(Util, GetDeleteBitmap) {
// test case insert repeated pk1 (ts = {1 ... N}) -> delete pk1 (ts = N) -> query (ts = N)
delete_ts = {uint64_t(N)};
delete_pk = {1};
offset = delete_record.reserved.fetch_add(1);
delete_record.timestamps_.set_data_raw(offset, delete_ts.data(), 1);
delete_record.pks_.set_data_raw(offset, delete_pk.data(), 1);
delete_record.ack_responder_.AddSegment(offset, offset + 1);
delete_record.push(delete_pk, delete_ts.data());
del_barrier = get_barrier(delete_record, query_timestamp);
res_bitmap = get_deleted_bitmap(del_barrier,

17
internal/querynodev2/segments/segment.go
View File

@ -524,19 +524,6 @@ func (s *LocalSegment) preInsert(numOfRecords int) (int64, error) {
return offset, nil
}
func (s *LocalSegment) preDelete(numOfRecords int) int64 {
/*
long int
PreDelete(CSegmentInterface c_segment, long int size);
*/
var offset C.int64_t
GetPool().Submit(func() (any, error) {
offset = C.PreDelete(s.ptr, C.int64_t(int64(numOfRecords)))
return nil, nil
}).Await()
return int64(offset)
}
func (s *LocalSegment) Insert(rowIDs []int64, timestamps []typeutil.Timestamp, record *segcorepb.InsertRecord) error {
if s.Type() != SegmentTypeGrowing {
return fmt.Errorf("unexpected segmentType when segmentInsert, segmentType = %s", s.typ.String())
@ -608,9 +595,7 @@ func (s *LocalSegment) Delete(primaryKeys []storage.PrimaryKey, timestamps []typ
return WrapSegmentReleased(s.segmentID)
}
offset := s.preDelete(len(primaryKeys))
var cOffset = C.int64_t(offset)
var cOffset = C.int64_t(0) // depre
var cSize = C.int64_t(len(primaryKeys))
var cTimestampsPtr = (*C.uint64_t)(&(timestamps)[0])

1
internal/querynodev2/services.go
View File

@ -1284,6 +1284,7 @@ func (node *QueryNode) SyncDistribution(ctx context.Context, req *querypb.SyncDi
DstNodeID: nodeID,
Version: req.GetVersion(),
NeedTransfer: false,
LoadScope: querypb.LoadScope_Delta,
})
if err != nil {
return util.WrapStatus(commonpb.ErrorCode_UnexpectedError, "failed to sync(load) segment", err), nil