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

fix: Restore the MVCC functionality. (#29749) (#29802) 

When the TimeTravel functionality was previously removed, it
inadvertently affected the MVCC functionality within the system. This PR
aims to reintroduce the internal MVCC functionality as follows:

1. Add MvccTimestamp to the requests of Search/Query and the results of
Search internally.
2. When the delegator receives a Query/Search request and there is no
MVCC timestamp set in the request, set the delegator's current tsafe as
the MVCC timestamp of the request. If the request already has an MVCC
timestamp, do not modify it.
3. When the Proxy handles Search and triggers the second phase ReQuery,
divide the ReQuery into different shards and pass the MVCC timestamp to
the corresponding Query requests.

issue: #29656
pr: #29749

Signed-off-by: zhenshan.cao <zhenshan.cao@zilliz.com>
pull/29881/head
zhenshan.cao 2024-01-11 14:42:49 +08:00 committed by GitHub
parent e7e4561da8
commit 7cf2be09b5
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
38 changed files with 2568 additions and 222 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
internal/core/src/common/Channel.h
View File

@ -1,3 +1,14 @@
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License
#include <oneapi/tbb/concurrent_queue.h>
#include <atomic>

5
internal/core/src/segcore/SegmentInterface.cpp
View File

@ -67,11 +67,12 @@ SegmentInternalInterface::FillTargetEntry(const query::Plan* plan,
std::unique_ptr<SearchResult>
SegmentInternalInterface::Search(
const query::Plan* plan,
const query::PlaceholderGroup* placeholder_group) const {
const query::PlaceholderGroup* placeholder_group,
Timestamp timestamp) const {
std::shared_lock lck(mutex_);
milvus::tracer::AddEvent("obtained_segment_lock_mutex");
check_search(plan);
query::ExecPlanNodeVisitor visitor(*this, 1L << 63, placeholder_group);
query::ExecPlanNodeVisitor visitor(*this, timestamp, placeholder_group);
auto results = std::make_unique<SearchResult>();
*results = visitor.get_moved_result(*plan->plan_node_);
results->segment_ = (void*)this;

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

@ -52,7 +52,8 @@ class SegmentInterface {
virtual std::unique_ptr<SearchResult>
Search(const query::Plan* Plan,
const query::PlaceholderGroup* placeholder_group) const = 0;
const query::PlaceholderGroup* placeholder_group,
Timestamp timestamp) const = 0;
virtual std::unique_ptr<proto::segcore::RetrieveResults>
Retrieve(const query::RetrievePlan* Plan,
@ -131,7 +132,8 @@ class SegmentInternalInterface : public SegmentInterface {
std::unique_ptr<SearchResult>
Search(const query::Plan* Plan,
const query::PlaceholderGroup* placeholder_group) const override;
const query::PlaceholderGroup* placeholder_group,
Timestamp timestamp) const override;
void
FillPrimaryKeys(const query::Plan* plan,

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

@ -78,6 +78,7 @@ Search(CSegmentInterface c_segment,
CSearchPlan c_plan,
CPlaceholderGroup c_placeholder_group,
CTraceContext c_trace,
uint64_t timestamp,
CSearchResult* result) {
try {
auto segment = (milvus::segcore::SegmentInterface*)c_segment;
@ -88,7 +89,7 @@ Search(CSegmentInterface c_segment,
c_trace.traceID, c_trace.spanID, c_trace.flag};
auto span = milvus::tracer::StartSpan("SegCoreSearch", &ctx);
milvus::tracer::SetRootSpan(span);
auto search_result = segment->Search(plan, phg_ptr);
auto search_result = segment->Search(plan, phg_ptr, timestamp);
if (!milvus::PositivelyRelated(
plan->plan_node_->search_info_.metric_type_)) {
for (auto& dis : search_result->distances_) {

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

@ -45,6 +45,7 @@ Search(CSegmentInterface c_segment,
CSearchPlan c_plan,
CPlaceholderGroup c_placeholder_group,
CTraceContext c_trace,
uint64_t timestamp,
CSearchResult* result);
void

11
internal/core/src/storage/azure-blob-storage/CMakeLists.txt
View File

@ -1,3 +1,14 @@
# Copyright (C) 2019-2020 Zilliz. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License
# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
# or implied. See the License for the specific language governing permissions and limitations under the License
# Copyright (c) Microsoft Corporation. All rights reserved.
# SPDX-License-Identifier: MIT

11
internal/core/src/storage/azure-blob-storage/test/CMakeLists.txt
View File

@ -1,3 +1,14 @@
# Copyright (C) 2019-2020 Zilliz. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License
# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
# or implied. See the License for the specific language governing permissions and limitations under the License
# Copyright (c) Microsoft Corporation. All rights reserved.
# SPDX-License-Identifier: MIT

11
internal/core/src/storage/azure-blob-storage/test/test_azure_blob_chunk_manager.cpp
View File

@ -1,3 +1,14 @@
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License
#include "../AzureBlobChunkManager.h"
#include <azure/identity/workload_identity_credential.hpp>
#include <gtest/gtest.h>

9
internal/core/unittest/bench/bench_search.cpp
View File

@ -90,8 +90,10 @@ Search_GrowingIndex(benchmark::State& state) {
dataset_.timestamps_.data(),
dataset_.raw_);
Timestamp ts = 10000000;
for (auto _ : state) {
auto qr = segment->Search(plan.get(), ph_group.get());
auto qr = segment->Search(plan.get(), ph_group.get(), ts);
}
}
@ -123,8 +125,11 @@ Search_Sealed(benchmark::State& state) {
segment->DropFieldData(milvus::FieldId(100));
segment->LoadIndex(info);
}
Timestamp ts = 10000000;
for (auto _ : state) {
auto qr = segment->Search(plan.get(), ph_group.get());
auto qr = segment->Search(plan.get(), ph_group.get(), ts);
}
}

49
internal/core/unittest/test_array.cpp
View File

@ -32,10 +32,9 @@ TEST(Array, TestConstructArray) {
ASSERT_EQ(int_array.get_data<int>(i), i);
}
ASSERT_TRUE(int_array.is_same_array(field_int_array));
auto int_array_tmp = Array(
const_cast<char*>(int_array.data()),
int_array.byte_size(),
int_array.get_element_type(),
auto int_array_tmp = Array(const_cast<char*>(int_array.data()),
int_array.byte_size(),
int_array.get_element_type(),
{});
ASSERT_TRUE(int_array_tmp == int_array);
auto int_8_array = Array(const_cast<char*>(int_array.data()),
@ -48,10 +47,9 @@ TEST(Array, TestConstructArray) {
DataType::INT16,
{});
ASSERT_EQ(int_array.length(), int_16_array.length());
auto int_array_view = ArrayView(
const_cast<char*>(int_array.data()),
int_array.byte_size(),
int_array.get_element_type(),
auto int_array_view = ArrayView(const_cast<char*>(int_array.data()),
int_array.byte_size(),
int_array.get_element_type(),
{});
ASSERT_EQ(int_array.length(), int_array_view.length());
ASSERT_EQ(int_array.byte_size(), int_array_view.byte_size());
@ -76,10 +74,9 @@ TEST(Array, TestConstructArray) {
long_array.get_element_type(),
{});
ASSERT_TRUE(long_array_tmp == long_array);
auto long_array_view = ArrayView(
const_cast<char*>(long_array.data()),
long_array.byte_size(),
long_array.get_element_type(),
auto long_array_view = ArrayView(const_cast<char*>(long_array.data()),
long_array.byte_size(),
long_array.get_element_type(),
{});
ASSERT_EQ(long_array.length(), long_array_view.length());
ASSERT_EQ(long_array.byte_size(), long_array_view.byte_size());
@ -114,10 +111,9 @@ TEST(Array, TestConstructArray) {
string_array.get_element_type(),
std::move(string_element_offsets));
ASSERT_TRUE(string_array_tmp == string_array);
auto string_array_view = ArrayView(
const_cast<char*>(string_array.data()),
string_array.byte_size(),
string_array.get_element_type(),
auto string_array_view = ArrayView(const_cast<char*>(string_array.data()),
string_array.byte_size(),
string_array.get_element_type(),
std::move(string_view_element_offsets));
ASSERT_EQ(string_array.length(), string_array_view.length());
ASSERT_EQ(string_array.byte_size(), string_array_view.byte_size());
@ -143,10 +139,9 @@ TEST(Array, TestConstructArray) {
bool_array.get_element_type(),
{});
ASSERT_TRUE(bool_array_tmp == bool_array);
auto bool_array_view = ArrayView(
const_cast<char*>(bool_array.data()),
bool_array.byte_size(),
bool_array.get_element_type(),
auto bool_array_view = ArrayView(const_cast<char*>(bool_array.data()),
bool_array.byte_size(),
bool_array.get_element_type(),
{});
ASSERT_EQ(bool_array.length(), bool_array_view.length());
ASSERT_EQ(bool_array.byte_size(), bool_array_view.byte_size());
@ -172,10 +167,9 @@ TEST(Array, TestConstructArray) {
float_array.get_element_type(),
{});
ASSERT_TRUE(float_array_tmp == float_array);
auto float_array_view = ArrayView(
const_cast<char*>(float_array.data()),
float_array.byte_size(),
float_array.get_element_type(),
auto float_array_view = ArrayView(const_cast<char*>(float_array.data()),
float_array.byte_size(),
float_array.get_element_type(),
{});
ASSERT_EQ(float_array.length(), float_array_view.length());
ASSERT_EQ(float_array.byte_size(), float_array_view.byte_size());
@ -202,10 +196,9 @@ TEST(Array, TestConstructArray) {
double_array.get_element_type(),
{});
ASSERT_TRUE(double_array_tmp == double_array);
auto double_array_view = ArrayView(
const_cast<char*>(double_array.data()),
double_array.byte_size(),
double_array.get_element_type(),
auto double_array_view = ArrayView(const_cast<char*>(double_array.data()),
double_array.byte_size(),
double_array.get_element_type(),
{});
ASSERT_EQ(double_array.length(), double_array_view.length());
ASSERT_EQ(double_array.byte_size(), double_array_view.byte_size());

38
internal/core/unittest/test_array_expr.cpp
View File

@ -894,7 +894,7 @@ TEST(Expr, TestArrayContains) {
*seg_promote, seg_promote->get_row_count(), MAX_TIMESTAMP);
std::vector<ArrayTestcase<bool>> bool_testcases{{{true, true}, {}},
{{false, false}, {}}};
{{false, false}, {}}};
for (auto testcase : bool_testcases) {
auto check = [&](const std::vector<bool>& values) {
@ -1652,9 +1652,7 @@ TEST(Expr, TestArrayBinaryArith) {
value:<int64_val:10 >
>)",
"int",
[](milvus::Array& array) {
return array.length() == 10;
}},
[](milvus::Array& array) { return array.length() == 10; }},
{R"(binary_arith_op_eval_range_expr: <
column_info: <
field_id: 101
@ -1667,9 +1665,7 @@ TEST(Expr, TestArrayBinaryArith) {
value:<int64_val:8 >
>)",
"int",
[](milvus::Array& array) {
return array.length() != 8;
}},
[](milvus::Array& array) { return array.length() != 8; }},
};
std::string raw_plan_tmp = R"(vector_anns: <
@ -1847,7 +1843,7 @@ TEST(Expr, TestArrayInTerm) {
ExecExprVisitor visitor(
*seg_promote, seg_promote->get_row_count(), MAX_TIMESTAMP);
std::vector<std::tuple<std::string,
std::vector<std::tuple<std::string,
std::string,
std::function<bool(milvus::Array & array)>>>
testcases = {
@ -1860,11 +1856,11 @@ TEST(Expr, TestArrayInTerm) {
>
values:<int64_val:1 > values:<int64_val:2 > values:<int64_val:3 >
>)",
"long",
[](milvus::Array& array) {
auto val = array.get_data<int64_t>(0);
return val == 1 || val ==2 || val == 3;
}},
"long",
[](milvus::Array& array) {
auto val = array.get_data<int64_t>(0);
return val == 1 || val == 2 || val == 3;
}},
{R"(term_expr: <
column_info: <
field_id: 101
@ -1874,9 +1870,7 @@ TEST(Expr, TestArrayInTerm) {
>
>)",
"long",
[](milvus::Array& array) {
return false;
}},
[](milvus::Array& array) { return false; }},
{R"(term_expr: <
column_info: <
field_id: 102
@ -1900,9 +1894,7 @@ TEST(Expr, TestArrayInTerm) {
>
>)",
"bool",
[](milvus::Array& array) {
return false;
}},
[](milvus::Array& array) { return false; }},
{R"(term_expr: <
column_info: <
field_id: 103
@ -1926,9 +1918,7 @@ TEST(Expr, TestArrayInTerm) {
>
>)",
"float",
[](milvus::Array& array) {
return false;
}},
[](milvus::Array& array) { return false; }},
{R"(term_expr: <
column_info: <
field_id: 104
@ -1952,9 +1942,7 @@ TEST(Expr, TestArrayInTerm) {
>
>)",
"string",
[](milvus::Array& array) {
return false;
}},
[](milvus::Array& array) { return false; }},
{R"(term_expr: <
column_info: <
field_id: 104

7
internal/core/unittest/test_binlog_index.cpp
View File

@ -196,7 +196,8 @@ TEST_P(BinlogIndexTest, Accuracy) {
std::vector<const milvus::query::PlaceholderGroup*> ph_group_arr = {
ph_group.get()};
auto nlist = segcore_config.get_nlist();
auto binlog_index_sr = segment->Search(plan.get(), ph_group.get());
auto binlog_index_sr =
segment->Search(plan.get(), ph_group.get(), 1L << 63);
ASSERT_EQ(binlog_index_sr->total_nq_, num_queries);
EXPECT_EQ(binlog_index_sr->unity_topK_, topk);
EXPECT_EQ(binlog_index_sr->distances_.size(), num_queries * topk);
@ -231,7 +232,7 @@ TEST_P(BinlogIndexTest, Accuracy) {
EXPECT_TRUE(segment->HasIndex(vec_field_id));
EXPECT_EQ(segment->get_row_count(), data_n);
EXPECT_FALSE(segment->HasFieldData(vec_field_id));
auto ivf_sr = segment->Search(plan.get(), ph_group.get());
auto ivf_sr = segment->Search(plan.get(), ph_group.get(), 1L << 63);
auto similary = GetKnnSearchRecall(num_queries,
binlog_index_sr->seg_offsets_.data(),
topk,
@ -323,4 +324,4 @@ TEST_P(BinlogIndexTest, LoadBinlogWithoutIndexMeta) {
EXPECT_FALSE(segment->HasIndex(vec_field_id));
EXPECT_EQ(segment->get_row_count(), data_n);
EXPECT_TRUE(segment->HasFieldData(vec_field_id));
}
}

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

@ -298,7 +298,7 @@ TEST(CApiTest, SegmentTest) {
ASSERT_NE(status.error_code, Success);
DeleteCollection(collection);
DeleteSegment(segment);
free((char *)status.error_msg);
free((char*)status.error_msg);
}
TEST(CApiTest, CPlan) {
@ -1009,11 +1009,13 @@ TEST(CApiTest, SearchTest) {
placeholderGroups.push_back(placeholderGroup);
CSearchResult search_result;
auto res = Search(segment, plan, placeholderGroup, {}, &search_result);
auto res =
Search(segment, plan, placeholderGroup, {}, ts_offset, &search_result);
ASSERT_EQ(res.error_code, Success);
CSearchResult search_result2;
auto res2 = Search(segment, plan, placeholderGroup, {}, &search_result2);
auto res2 =
Search(segment, plan, placeholderGroup, {}, ts_offset, &search_result2);
ASSERT_EQ(res2.error_code, Success);
DeleteSearchPlan(plan);
@ -1077,7 +1079,12 @@ TEST(CApiTest, SearchTestWithExpr) {
dataset.timestamps_.push_back(1);
CSearchResult search_result;
auto res = Search(segment, plan, placeholderGroup, {}, &search_result);
auto res = Search(segment,
plan,
placeholderGroup,
{},
dataset.timestamps_[0],
&search_result);
ASSERT_EQ(res.error_code, Success);
DeleteSearchPlan(plan);
@ -1355,7 +1362,7 @@ TEST(CApiTest, ReudceNullResult) {
auto slice_topKs = std::vector<int64_t>{1};
std::vector<CSearchResult> results;
CSearchResult res;
status = Search(segment, plan, placeholderGroup, {}, &res);
status = Search(segment, plan, placeholderGroup, {}, 1L << 63, &res);
ASSERT_EQ(status.error_code, Success);
results.push_back(res);
CSearchResultDataBlobs cSearchResultData;
@ -1442,9 +1449,11 @@ TEST(CApiTest, ReduceRemoveDuplicates) {
auto slice_topKs = std::vector<int64_t>{topK / 2, topK};
std::vector<CSearchResult> results;
CSearchResult res1, res2;
status = Search(segment, plan, placeholderGroup, {}, &res1);
status = Search(
segment, plan, placeholderGroup, {}, dataset.timestamps_[0], &res1);
ASSERT_EQ(status.error_code, Success);
status = Search(segment, plan, placeholderGroup, {}, &res2);
status = Search(
segment, plan, placeholderGroup, {}, dataset.timestamps_[0], &res2);
ASSERT_EQ(status.error_code, Success);
results.push_back(res1);
results.push_back(res2);
@ -1473,11 +1482,14 @@ TEST(CApiTest, ReduceRemoveDuplicates) {
auto slice_topKs = std::vector<int64_t>{topK / 2, topK, topK};
std::vector<CSearchResult> results;
CSearchResult res1, res2, res3;
status = Search(segment, plan, placeholderGroup, {}, &res1);
status = Search(
segment, plan, placeholderGroup, {}, dataset.timestamps_[0], &res1);
ASSERT_EQ(status.error_code, Success);
status = Search(segment, plan, placeholderGroup, {}, &res2);
status = Search(
segment, plan, placeholderGroup, {}, dataset.timestamps_[0], &res2);
ASSERT_EQ(status.error_code, Success);
status = Search(segment, plan, placeholderGroup, {}, &res3);
status = Search(
segment, plan, placeholderGroup, {}, dataset.timestamps_[0], &res3);
ASSERT_EQ(status.error_code, Success);
results.push_back(res1);
results.push_back(res2);
@ -1566,9 +1578,11 @@ testReduceSearchWithExpr(int N, int topK, int num_queries) {
std::vector<CSearchResult> results;
CSearchResult res1;
CSearchResult res2;
auto res = Search(segment, plan, placeholderGroup, {}, &res1);
auto res = Search(
segment, plan, placeholderGroup, {}, dataset.timestamps_[N - 1], &res1);
ASSERT_EQ(res.error_code, Success);
res = Search(segment, plan, placeholderGroup, {}, &res2);
res = Search(
segment, plan, placeholderGroup, {}, dataset.timestamps_[N - 1], &res2);
ASSERT_EQ(res.error_code, Success);
results.push_back(res1);
results.push_back(res2);
@ -1792,9 +1806,15 @@ TEST(CApiTest, Indexing_Without_Predicate) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestmap = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestmap,
&c_search_result_on_smallIndex);
ASSERT_EQ(res_before_load_index.error_code, Success);
// load index to segment
@ -1854,6 +1874,7 @@ TEST(CApiTest, Indexing_Without_Predicate) {
plan,
placeholderGroup,
{},
timestmap,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -1936,9 +1957,15 @@ TEST(CApiTest, Indexing_Expr_Without_Predicate) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestamp = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_smallIndex);
ASSERT_EQ(res_before_load_index.error_code, Success);
// load index to segment
@ -1999,6 +2026,7 @@ TEST(CApiTest, Indexing_Expr_Without_Predicate) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -2109,10 +2137,15 @@ TEST(CApiTest, Indexing_With_float_Predicate_Range) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestamp = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_smallIndex);
ASSERT_EQ(res_before_load_index.error_code, Success);
// load index to segment
@ -2173,6 +2206,7 @@ TEST(CApiTest, Indexing_With_float_Predicate_Range) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -2201,7 +2235,7 @@ TEST(CApiTest, Indexing_Expr_With_float_Predicate_Range) {
generate_collection_schema(knowhere::metric::L2, DIM, false);
auto collection = NewCollection(schema_string.c_str());
auto schema = ((segcore::Collection*)collection)->get_schema();
CSegmentInterface segment;
CSegmentInterface segment;
auto status = NewSegment(collection, Growing, -1, &segment);
ASSERT_EQ(status.error_code, Success);
@ -2285,10 +2319,15 @@ TEST(CApiTest, Indexing_Expr_With_float_Predicate_Range) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestamp = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_smallIndex);
ASSERT_EQ(res_before_load_index.error_code, Success);
// load index to segment
@ -2349,6 +2388,7 @@ TEST(CApiTest, Indexing_Expr_With_float_Predicate_Range) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -2453,10 +2493,15 @@ TEST(CApiTest, Indexing_With_float_Predicate_Term) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestamp = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_smallIndex);
ASSERT_EQ(res_before_load_index.error_code, Success);
// load index to segment
@ -2517,6 +2562,7 @@ TEST(CApiTest, Indexing_With_float_Predicate_Term) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -2622,10 +2668,15 @@ TEST(CApiTest, Indexing_Expr_With_float_Predicate_Term) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestamp = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_smallIndex);
ASSERT_EQ(res_before_load_index.error_code, Success);
// load index to segment
@ -2686,6 +2737,7 @@ TEST(CApiTest, Indexing_Expr_With_float_Predicate_Term) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -2796,10 +2848,15 @@ TEST(CApiTest, Indexing_With_binary_Predicate_Range) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestamp = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_smallIndex);
ASSERT_EQ(res_before_load_index.error_code, Success);
// load index to segment
@ -2861,6 +2918,7 @@ TEST(CApiTest, Indexing_With_binary_Predicate_Range) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -2971,10 +3029,15 @@ TEST(CApiTest, Indexing_Expr_With_binary_Predicate_Range) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestamp = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_smallIndex);
ASSERT_TRUE(res_before_load_index.error_code == Success)
<< res_before_load_index.error_msg;
@ -3036,6 +3099,7 @@ TEST(CApiTest, Indexing_Expr_With_binary_Predicate_Range) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -3141,10 +3205,15 @@ TEST(CApiTest, Indexing_With_binary_Predicate_Term) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestamp = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_smallIndex);
ASSERT_EQ(res_before_load_index.error_code, Success);
// load index to segment
@ -3205,6 +3274,7 @@ TEST(CApiTest, Indexing_With_binary_Predicate_Term) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -3332,11 +3402,15 @@ TEST(CApiTest, Indexing_Expr_With_binary_Predicate_Term) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp time = 10000000;
Timestamp timestamp = 10000000;
CSearchResult c_search_result_on_smallIndex;
auto res_before_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_smallIndex);
auto res_before_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_smallIndex);
ASSERT_EQ(res_before_load_index.error_code, Success);
// load index to segment
@ -3397,6 +3471,7 @@ TEST(CApiTest, Indexing_Expr_With_binary_Predicate_Term) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -3535,7 +3610,7 @@ TEST(CApiTest, SealedSegment_search_float_Predicate_Range) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp time = 10000000;
Timestamp timestamp = 10000000;
// load index to segment
auto indexing = generate_index(vec_col.data(),
@ -3594,6 +3669,7 @@ TEST(CApiTest, SealedSegment_search_float_Predicate_Range) {
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
@ -3672,12 +3748,14 @@ TEST(CApiTest, SealedSegment_search_without_predicates) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
CSearchResult search_result;
auto res = Search(segment, plan, placeholderGroup, {}, &search_result);
auto res = Search(
segment, plan, placeholderGroup, {}, N + ts_offset, &search_result);
std::cout << res.error_msg << std::endl;
ASSERT_EQ(res.error_code, Success);
CSearchResult search_result2;
auto res2 = Search(segment, plan, placeholderGroup, {}, &search_result2);
auto res2 = Search(
segment, plan, placeholderGroup, {}, N + ts_offset, &search_result2);
ASSERT_EQ(res2.error_code, Success);
DeleteSearchPlan(plan);
@ -3766,6 +3844,7 @@ TEST(CApiTest, SealedSegment_search_float_With_Expr_Predicate_Range) {
std::vector<CPlaceholderGroup> placeholderGroups;
placeholderGroups.push_back(placeholderGroup);
Timestamp timestamp = 10000000;
// load index to segment
auto indexing = generate_index(vec_col.data(),
@ -3825,8 +3904,12 @@ TEST(CApiTest, SealedSegment_search_float_With_Expr_Predicate_Range) {
}
CSearchResult c_search_result_on_bigIndex;
auto res_after_load_index = Search(
segment, plan, placeholderGroup, {}, &c_search_result_on_bigIndex);
auto res_after_load_index = Search(segment,
plan,
placeholderGroup,
{},
timestamp,
&c_search_result_on_bigIndex);
ASSERT_EQ(res_after_load_index.error_code, Success);
auto search_result_on_bigIndex = (SearchResult*)c_search_result_on_bigIndex;
@ -4119,7 +4202,8 @@ TEST(CApiTest, RANGE_SEARCH_WITH_RADIUS_WHEN_IP) {
placeholderGroups.push_back(placeholderGroup);
CSearchResult search_result;
auto res = Search(segment, plan, placeholderGroup, {}, &search_result);
auto res =
Search(segment, plan, placeholderGroup, {}, ts_offset, &search_result);
ASSERT_EQ(res.error_code, Success);
DeleteSearchPlan(plan);
@ -4182,7 +4266,8 @@ TEST(CApiTest, RANGE_SEARCH_WITH_RADIUS_AND_RANGE_FILTER_WHEN_IP) {
placeholderGroups.push_back(placeholderGroup);
CSearchResult search_result;
auto res = Search(segment, plan, placeholderGroup, {}, &search_result);
auto res =
Search(segment, plan, placeholderGroup, {}, ts_offset, &search_result);
ASSERT_EQ(res.error_code, Success);
DeleteSearchPlan(plan);
@ -4245,7 +4330,8 @@ TEST(CApiTest, RANGE_SEARCH_WITH_RADIUS_WHEN_L2) {
placeholderGroups.push_back(placeholderGroup);
CSearchResult search_result;
auto res = Search(segment, plan, placeholderGroup, {}, &search_result);
auto res =
Search(segment, plan, placeholderGroup, {}, ts_offset, &search_result);
ASSERT_EQ(res.error_code, Success);
DeleteSearchPlan(plan);
@ -4308,7 +4394,8 @@ TEST(CApiTest, RANGE_SEARCH_WITH_RADIUS_AND_RANGE_FILTER_WHEN_L2) {
placeholderGroups.push_back(placeholderGroup);
CSearchResult search_result;
auto res = Search(segment, plan, placeholderGroup, {}, &search_result);
auto res =
Search(segment, plan, placeholderGroup, {}, ts_offset, &search_result);
ASSERT_EQ(res.error_code, Success);
DeleteSearchPlan(plan);

4
internal/core/unittest/test_float16.cpp
View File

@ -153,7 +153,7 @@ TEST(Float16, ExecWithoutPredicateFlat) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), 1L << 63);
int topk = 5;
query::Json json = SearchResultToJson(*sr);
@ -411,7 +411,7 @@ TEST(Float16, ExecWithPredicate) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), 1L << 63);
int topk = 5;
query::Json json = SearchResultToJson(*sr);

7
internal/core/unittest/test_growing_index.cpp
View File

@ -101,7 +101,9 @@ TEST(GrowingIndex, Correctness) {
*schema, plan_str.data(), plan_str.size());
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto sr = segment->Search(plan.get(), ph_group.get());
Timestamp timestamp = 1000000;
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
EXPECT_EQ(sr->total_nq_, num_queries);
EXPECT_EQ(sr->unity_topK_, top_k);
EXPECT_EQ(sr->distances_.size(), num_queries * top_k);
@ -111,7 +113,8 @@ TEST(GrowingIndex, Correctness) {
*schema, range_plan_str.data(), range_plan_str.size());
auto range_ph_group = ParsePlaceholderGroup(
range_plan.get(), ph_group_raw.SerializeAsString());
auto range_sr = segment->Search(range_plan.get(), range_ph_group.get());
auto range_sr =
segment->Search(range_plan.get(), range_ph_group.get(), timestamp);
ASSERT_EQ(range_sr->total_nq_, num_queries);
EXPECT_EQ(sr->unity_topK_, top_k);
EXPECT_EQ(sr->distances_.size(), num_queries * top_k);

36
internal/core/unittest/test_query.cpp
View File

@ -128,8 +128,9 @@ TEST(Query, ExecWithPredicateLoader) {
auto ph_group_raw = CreatePlaceholderGroup(num_queries, 16, 1024);
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
Timestamp timestamp = 1000000;
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
query::Json json = SearchResultToJson(*sr);
#ifdef __linux__
@ -212,7 +213,9 @@ TEST(Query, ExecWithPredicateSmallN) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto sr = segment->Search(plan.get(), ph_group.get());
Timestamp timestamp = 1000000;
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
query::Json json = SearchResultToJson(*sr);
std::cout << json.dump(2);
@ -270,8 +273,9 @@ TEST(Query, ExecWithPredicate) {
auto ph_group_raw = CreatePlaceholderGroup(num_queries, 16, 1024);
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
Timestamp timestamp = 1000000;
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
query::Json json = SearchResultToJson(*sr);
#ifdef __linux__
@ -345,9 +349,9 @@ TEST(Query, ExecTerm) {
auto ph_group_raw = CreatePlaceholderGroup(num_queries, 16, 1024);
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
Timestamp timestamp = 1000000;
auto sr = segment->Search(plan.get(), ph_group.get());
std::vector<std::vector<std::string>> results;
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
int topk = 5;
auto json = SearchResultToJson(*sr);
ASSERT_EQ(sr->total_nq_, num_queries);
@ -381,7 +385,8 @@ TEST(Query, ExecEmpty) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto sr = segment->Search(plan.get(), ph_group.get());
Timestamp timestamp = 1000000;
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
std::cout << SearchResultToJson(*sr);
for (auto i : sr->seg_offsets_) {
@ -428,8 +433,8 @@ TEST(Query, ExecWithoutPredicateFlat) {
auto ph_group_raw = CreatePlaceholderGroup(num_queries, 16, 1024);
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto sr = segment->Search(plan.get(), ph_group.get());
Timestamp timestamp = 1000000;
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
std::vector<std::vector<std::string>> results;
auto json = SearchResultToJson(*sr);
std::cout << json.dump(2);
@ -471,8 +476,9 @@ TEST(Query, ExecWithoutPredicate) {
auto ph_group_raw = CreatePlaceholderGroup(num_queries, 16, 1024);
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
Timestamp timestamp = 1000000;
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
assert_order(*sr, "l2");
std::vector<std::vector<std::string>> results;
auto json = SearchResultToJson(*sr);
@ -540,7 +546,9 @@ TEST(Query, InnerProduct) {
CreatePlaceholderGroupFromBlob(num_queries, 16, col.data());
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto sr = segment->Search(plan.get(), ph_group.get());
Timestamp ts = N * 2;
auto sr = segment->Search(plan.get(), ph_group.get(), ts);
assert_order(*sr, "ip");
}
@ -627,6 +635,8 @@ TEST(Query, FillSegment) {
CreateSearchPlanByExpr(*schema, plan_str.data(), plan_str.size());
auto ph_proto = CreatePlaceholderGroup(10, 16, 443);
auto ph = ParsePlaceholderGroup(plan.get(), ph_proto.SerializeAsString());
Timestamp ts = N * 2UL;
auto topk = 5;
auto num_queries = 10;
@ -636,7 +646,7 @@ TEST(Query, FillSegment) {
schema->get_field_id(FieldName("fakevec")));
plan->target_entries_.push_back(
schema->get_field_id(FieldName("the_value")));
auto result = segment->Search(plan.get(), ph.get());
auto result = segment->Search(plan.get(), ph.get(), ts);
result->result_offsets_.resize(topk * num_queries);
segment->FillTargetEntry(plan.get(), *result);
segment->FillPrimaryKeys(plan.get(), *result);
@ -740,7 +750,9 @@ TEST(Query, ExecWithPredicateBinary) {
num_queries, 512, vec_ptr.data() + 1024 * 512 / 8);
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto sr = segment->Search(plan.get(), ph_group.get());
Timestamp timestamp = 1000000;
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
query::Json json = SearchResultToJson(*sr);
std::cout << json.dump(2);

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

@ -80,10 +80,11 @@ TEST(Sealed, without_predicate) {
CreatePlaceholderGroupFromBlob(num_queries, 16, query_ptr);
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
Timestamp timestamp = 1000000;
std::vector<const PlaceholderGroup*> ph_group_arr = {ph_group.get()};
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
auto pre_result = SearchResultToJson(*sr);
milvus::index::CreateIndexInfo create_index_info;
create_index_info.field_type = DataType::VECTOR_FLOAT;
@ -127,7 +128,7 @@ TEST(Sealed, without_predicate) {
sealed_segment->DropFieldData(fake_id);
sealed_segment->LoadIndex(load_info);
sr = sealed_segment->Search(plan.get(), ph_group.get());
sr = sealed_segment->Search(plan.get(), ph_group.get(), timestamp);
auto post_result = SearchResultToJson(*sr);
std::cout << "ref_result" << std::endl;
@ -135,6 +136,9 @@ TEST(Sealed, without_predicate) {
std::cout << "post_result" << std::endl;
std::cout << post_result.dump(1);
// ASSERT_EQ(ref_result.dump(1), post_result.dump(1));
sr = sealed_segment->Search(plan.get(), ph_group.get(), 0);
EXPECT_EQ(sr->get_total_result_count(), 0);
}
TEST(Sealed, with_predicate) {
@ -196,10 +200,11 @@ TEST(Sealed, with_predicate) {
CreatePlaceholderGroupFromBlob(num_queries, 16, query_ptr);
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
Timestamp timestamp = 1000000;
std::vector<const PlaceholderGroup*> ph_group_arr = {ph_group.get()};
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
milvus::index::CreateIndexInfo create_index_info;
create_index_info.field_type = DataType::VECTOR_FLOAT;
create_index_info.metric_type = knowhere::metric::L2;
@ -242,7 +247,7 @@ TEST(Sealed, with_predicate) {
sealed_segment->DropFieldData(fake_id);
sealed_segment->LoadIndex(load_info);
sr = sealed_segment->Search(plan.get(), ph_group.get());
sr = sealed_segment->Search(plan.get(), ph_group.get(), timestamp);
for (int i = 0; i < num_queries; ++i) {
auto offset = i * topK;
@ -303,6 +308,7 @@ TEST(Sealed, with_predicate_filter_all) {
CreatePlaceholderGroupFromBlob(num_queries, 16, query_ptr);
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
Timestamp timestamp = 1000000;
std::vector<const PlaceholderGroup*> ph_group_arr = {ph_group.get()};
@ -337,7 +343,7 @@ TEST(Sealed, with_predicate_filter_all) {
ivf_sealed_segment->DropFieldData(fake_id);
ivf_sealed_segment->LoadIndex(load_info);
auto sr = ivf_sealed_segment->Search(plan.get(), ph_group.get());
auto sr = ivf_sealed_segment->Search(plan.get(), ph_group.get(), timestamp);
EXPECT_EQ(sr->get_total_result_count(), 0);
auto hnsw_conf =
@ -371,7 +377,8 @@ TEST(Sealed, with_predicate_filter_all) {
hnsw_sealed_segment->DropFieldData(fake_id);
hnsw_sealed_segment->LoadIndex(hnsw_load_info);
auto sr2 = hnsw_sealed_segment->Search(plan.get(), ph_group.get());
auto sr2 =
hnsw_sealed_segment->Search(plan.get(), ph_group.get(), timestamp);
EXPECT_EQ(sr2->get_total_result_count(), 0);
}
@ -454,7 +461,7 @@ TEST(Sealed, LoadFieldData) {
>
placeholder_tag: "$0"
>)";
Timestamp timestamp = 1000000;
auto plan_str = translate_text_plan_to_binary_plan(raw_plan);
auto plan =
CreateSearchPlanByExpr(*schema, plan_str.data(), plan_str.size());
@ -463,13 +470,13 @@ TEST(Sealed, LoadFieldData) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get()));
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get(), timestamp));
SealedLoadFieldData(dataset, *segment);
segment->Search(plan.get(), ph_group.get());
segment->Search(plan.get(), ph_group.get(), timestamp);
segment->DropFieldData(fakevec_id);
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get()));
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get(), timestamp));
LoadIndexInfo vec_info;
vec_info.field_id = fakevec_id.get();
@ -492,12 +499,12 @@ TEST(Sealed, LoadFieldData) {
ASSERT_EQ(chunk_span3[i], ref3[i]);
}
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
auto json = SearchResultToJson(*sr);
std::cout << json.dump(1);
segment->DropIndex(fakevec_id);
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get()));
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get(), timestamp));
}
TEST(Sealed, LoadFieldDataMmap) {
@ -552,7 +559,7 @@ TEST(Sealed, LoadFieldDataMmap) {
>
placeholder_tag: "$0"
>)";
Timestamp timestamp = 1000000;
auto plan_str = translate_text_plan_to_binary_plan(raw_plan);
auto plan =
CreateSearchPlanByExpr(*schema, plan_str.data(), plan_str.size());
@ -561,13 +568,13 @@ TEST(Sealed, LoadFieldDataMmap) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get()));
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get(), timestamp));
SealedLoadFieldData(dataset, *segment, {}, true);
segment->Search(plan.get(), ph_group.get());
segment->Search(plan.get(), ph_group.get(), timestamp);
segment->DropFieldData(fakevec_id);
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get()));
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get(), timestamp));
LoadIndexInfo vec_info;
vec_info.field_id = fakevec_id.get();
@ -590,12 +597,12 @@ TEST(Sealed, LoadFieldDataMmap) {
ASSERT_EQ(chunk_span3[i], ref3[i]);
}
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
auto json = SearchResultToJson(*sr);
std::cout << json.dump(1);
segment->DropIndex(fakevec_id);
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get()));
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get(), timestamp));
}
TEST(Sealed, LoadScalarIndex) {
@ -670,7 +677,7 @@ TEST(Sealed, LoadScalarIndex) {
>
placeholder_tag: "$0"
>)";
Timestamp timestamp = 1000000;
auto plan_str = translate_text_plan_to_binary_plan(raw_plan);
auto plan =
CreateSearchPlanByExpr(*schema, plan_str.data(), plan_str.size());
@ -732,7 +739,7 @@ TEST(Sealed, LoadScalarIndex) {
nothing_index.index = GenScalarIndexing<int32_t>(N, nothing_data.data());
segment->LoadIndex(nothing_index);
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), timestamp);
auto json = SearchResultToJson(*sr);
std::cout << json.dump(1);
}
@ -781,7 +788,7 @@ TEST(Sealed, Delete) {
>
placeholder_tag: "$0"
>)";
Timestamp timestamp = 1000000;
auto plan_str = translate_text_plan_to_binary_plan(raw_plan);
auto plan =
CreateSearchPlanByExpr(*schema, plan_str.data(), plan_str.size());
@ -790,7 +797,7 @@ TEST(Sealed, Delete) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get()));
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get(), timestamp));
SealedLoadFieldData(dataset, *segment);
@ -865,7 +872,7 @@ TEST(Sealed, OverlapDelete) {
>
placeholder_tag: "$0"
>)";
Timestamp timestamp = 1000000;
auto plan_str = translate_text_plan_to_binary_plan(raw_plan);
auto plan =
CreateSearchPlanByExpr(*schema, plan_str.data(), plan_str.size());
@ -874,7 +881,7 @@ TEST(Sealed, OverlapDelete) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get()));
ASSERT_ANY_THROW(segment->Search(plan.get(), ph_group.get(), timestamp));
SealedLoadFieldData(dataset, *segment);
@ -992,7 +999,7 @@ TEST(Sealed, BF) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto result = segment->Search(plan.get(), ph_group.get());
auto result = segment->Search(plan.get(), ph_group.get(), MAX_TIMESTAMP);
auto ves = SearchResultToVector(*result);
// first: offset, second: distance
EXPECT_GE(ves[0].first, 0);
@ -1046,7 +1053,7 @@ TEST(Sealed, BF_Overflow) {
auto ph_group =
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
auto result = segment->Search(plan.get(), ph_group.get());
auto result = segment->Search(plan.get(), ph_group.get(), MAX_TIMESTAMP);
auto ves = SearchResultToVector(*result);
for (int i = 0; i < num_queries; ++i) {
EXPECT_EQ(ves[0].first, -1);
@ -1322,7 +1329,7 @@ TEST(Sealed, LoadArrayFieldData) {
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
SealedLoadFieldData(dataset, *segment);
segment->Search(plan.get(), ph_group.get());
segment->Search(plan.get(), ph_group.get(), 1L << 63);
auto ids_ds = GenRandomIds(N);
auto s = dynamic_cast<SegmentSealedImpl*>(segment.get());
@ -1379,7 +1386,7 @@ TEST(Sealed, LoadArrayFieldDataWithMMap) {
ParsePlaceholderGroup(plan.get(), ph_group_raw.SerializeAsString());
SealedLoadFieldData(dataset, *segment, {}, true);
segment->Search(plan.get(), ph_group.get());
segment->Search(plan.get(), ph_group.get(), 1L << 63);
}
TEST(Sealed, QueryAllFields) {

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

@ -714,7 +714,7 @@ TEST(AlwaysTrueStringPlan, SearchWithOutputFields) {
auto sub_result = BruteForceSearch(
search_dataset, vec_col.data(), N, knowhere::Json(), nullptr);
auto sr = segment->Search(plan.get(), ph_group.get());
auto sr = segment->Search(plan.get(), ph_group.get(), MAX_TIMESTAMP);
segment->FillPrimaryKeys(plan.get(), *sr);
segment->FillTargetEntry(plan.get(), *sr);
ASSERT_EQ(sr->pk_type_, DataType::VARCHAR);

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

@ -144,13 +144,16 @@ struct TmpFileWrapper {
std::string filename;
TmpFileWrapper(const std::string& _filename) : filename{_filename} {
fd = open(
filename.c_str(), O_RDWR | O_CREAT | O_EXCL, S_IRUSR | S_IWUSR | S_IXUSR);
fd = open(filename.c_str(),
O_RDWR | O_CREAT | O_EXCL,
S_IRUSR | S_IWUSR | S_IXUSR);
}
TmpFileWrapper(const TmpFileWrapper&) = delete;
TmpFileWrapper(TmpFileWrapper&&) = delete;
TmpFileWrapper& operator =(const TmpFileWrapper&) = delete;
TmpFileWrapper& operator =(TmpFileWrapper&&) = delete;
TmpFileWrapper&
operator=(const TmpFileWrapper&) = delete;
TmpFileWrapper&
operator=(TmpFileWrapper&&) = delete;
~TmpFileWrapper() {
if (fd != -1) {
close(fd);
@ -181,8 +184,8 @@ TEST(Util, read_from_fd) {
tmp_file.fd, read_buf.get(), data_size * max_loop));
// On Linux, read() (and similar system calls) will transfer at most 0x7ffff000 (2,147,479,552) bytes once
EXPECT_THROW(milvus::index::ReadDataFromFD(
tmp_file.fd, read_buf.get(), data_size * max_loop, INT_MAX),
milvus::SegcoreError);
EXPECT_THROW(
milvus::index::ReadDataFromFD(
tmp_file.fd, read_buf.get(), data_size * max_loop, INT_MAX),
milvus::SegcoreError);
}

4
internal/proto/internal.proto
View File

@ -94,6 +94,7 @@ message SearchRequest {
common.DslType dsl_type = 8;
bytes serialized_expr_plan = 9;
repeated int64 output_fields_id = 10;
uint64 mvcc_timestamp = 11;
uint64 guarantee_timestamp = 12;
uint64 timeout_timestamp = 13;
int64 nq = 14;
@ -120,6 +121,7 @@ message SearchResults {
// search request cost
CostAggregation costAggregation = 13;
map<string, uint64> channels_mvcc = 14;
}
message CostAggregation {
@ -160,7 +162,7 @@ message RetrieveResults {
repeated int64 global_sealed_segmentIDs = 8;
// query request cost
CostAggregation costAggregation = 13;
CostAggregation costAggregation = 13;
}
message LoadIndex {

2132
internal/proto/internalpb/internal.pb.go~parent of ca1349708... Remove time travel ralted testcase (#26119) Normal file
View File

File diff suppressed because it is too large Load Diff

2
internal/proxy/lb_policy.go
View File

@ -32,7 +32,7 @@ import (
"github.com/milvus-io/milvus/pkg/util/typeutil"
)
type executeFunc func(context.Context, UniqueID, types.QueryNodeClient, ...string) error
type executeFunc func(context.Context, UniqueID, types.QueryNodeClient, string) error
type ChannelWorkload struct {
db string

18
internal/proxy/lb_policy_test.go
View File

@ -248,7 +248,7 @@ func (s *LBPolicySuite) TestExecuteWithRetry() {
channel: s.channels[0],
shardLeaders: s.nodes,
nq: 1,
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, s ...string) error {
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, channel string) error {
return nil
},
retryTimes: 1,
@ -265,7 +265,7 @@ func (s *LBPolicySuite) TestExecuteWithRetry() {
channel: s.channels[0],
shardLeaders: s.nodes,
nq: 1,
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, s ...string) error {
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, channel string) error {
return nil
},
retryTimes: 1,
@ -285,7 +285,7 @@ func (s *LBPolicySuite) TestExecuteWithRetry() {
channel: s.channels[0],
shardLeaders: s.nodes,
nq: 1,
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, s ...string) error {
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, channel string) error {
return nil
},
retryTimes: 1,
@ -303,7 +303,7 @@ func (s *LBPolicySuite) TestExecuteWithRetry() {
channel: s.channels[0],
shardLeaders: s.nodes,
nq: 1,
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, s ...string) error {
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, channel string) error {
return nil
},
retryTimes: 2,
@ -324,7 +324,7 @@ func (s *LBPolicySuite) TestExecuteWithRetry() {
channel: s.channels[0],
shardLeaders: s.nodes,
nq: 1,
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, s ...string) error {
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, channel string) error {
counter++
if counter == 1 {
return errors.New("fake error")
@ -349,7 +349,7 @@ func (s *LBPolicySuite) TestExecuteWithRetry() {
channel: s.channels[0],
shardLeaders: s.nodes,
nq: 1,
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, s ...string) error {
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, channel string) error {
_, err := qn.Search(ctx, nil)
return err
},
@ -370,7 +370,7 @@ func (s *LBPolicySuite) TestExecute() {
collectionName: s.collectionName,
collectionID: s.collectionID,
nq: 1,
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, s ...string) error {
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, channel string) error {
return nil
},
})
@ -383,7 +383,7 @@ func (s *LBPolicySuite) TestExecute() {
collectionName: s.collectionName,
collectionID: s.collectionID,
nq: 1,
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, s ...string) error {
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, channel string) error {
// succeed in first execute
if counter.Add(1) == 1 {
return nil
@ -404,7 +404,7 @@ func (s *LBPolicySuite) TestExecute() {
collectionName: s.collectionName,
collectionID: s.collectionID,
nq: 1,
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, s ...string) error {
exec: func(ctx context.Context, ui UniqueID, qn types.QueryNodeClient, channel string) error {
return nil
},
})

6
internal/proxy/task_delete.go
View File

@ -350,7 +350,7 @@ func (dr *deleteRunner) produce(ctx context.Context, primaryKeys *schemapb.IDs)
// getStreamingQueryAndDelteFunc return query function used by LBPolicy
// make sure it concurrent safe
func (dr *deleteRunner) getStreamingQueryAndDelteFunc(plan *planpb.PlanNode) executeFunc {
return func(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channelIDs ...string) error {
return func(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channel string) error {
var partitionIDs []int64
// optimize query when partitionKey on
@ -375,7 +375,7 @@ func (dr *deleteRunner) getStreamingQueryAndDelteFunc(plan *planpb.PlanNode) exe
log := log.Ctx(ctx).With(
zap.Int64("collectionID", dr.collectionID),
zap.Int64s("partitionIDs", partitionIDs),
zap.Strings("channels", channelIDs),
zap.String("channel", channel),
zap.Int64("nodeID", nodeID))
// set plan
@ -405,7 +405,7 @@ func (dr *deleteRunner) getStreamingQueryAndDelteFunc(plan *planpb.PlanNode) exe
OutputFieldsId: outputFieldIDs,
GuaranteeTimestamp: parseGuaranteeTsFromConsistency(dr.ts, dr.ts, dr.req.GetConsistencyLevel()),
},
DmlChannels: channelIDs,
DmlChannels: []string{channel},
Scope: querypb.DataScope_All,
}

16
internal/proxy/task_delete_test.go
View File

@ -546,7 +546,7 @@ func TestDeleteRunner_Run(t *testing.T) {
},
}
lb.EXPECT().Execute(mock.Anything, mock.Anything).Call.Return(func(ctx context.Context, workload CollectionWorkLoad) error {
return workload.exec(ctx, 1, qn)
return workload.exec(ctx, 1, qn, "")
})
qn.EXPECT().QueryStream(mock.Anything, mock.Anything).Return(nil, errors.New("mock error"))
@ -591,7 +591,7 @@ func TestDeleteRunner_Run(t *testing.T) {
stream.EXPECT().Produce(mock.Anything).Return(nil)
lb.EXPECT().Execute(mock.Anything, mock.Anything).Call.Return(func(ctx context.Context, workload CollectionWorkLoad) error {
return workload.exec(ctx, 1, qn)
return workload.exec(ctx, 1, qn, "")
})
qn.EXPECT().QueryStream(mock.Anything, mock.Anything).Call.Return(
@ -654,7 +654,7 @@ func TestDeleteRunner_Run(t *testing.T) {
mockMgr.EXPECT().getOrCreateDmlStream(mock.Anything).Return(stream, nil)
mockMgr.EXPECT().getChannels(collectionID).Return(channels, nil)
lb.EXPECT().Execute(mock.Anything, mock.Anything).Call.Return(func(ctx context.Context, workload CollectionWorkLoad) error {
return workload.exec(ctx, 1, qn)
return workload.exec(ctx, 1, qn, "")
})
qn.EXPECT().QueryStream(mock.Anything, mock.Anything).Call.Return(
@ -716,7 +716,7 @@ func TestDeleteRunner_Run(t *testing.T) {
mockMgr.EXPECT().getOrCreateDmlStream(mock.Anything).Return(stream, nil)
mockMgr.EXPECT().getChannels(collectionID).Return(channels, nil)
lb.EXPECT().Execute(mock.Anything, mock.Anything).Call.Return(func(ctx context.Context, workload CollectionWorkLoad) error {
return workload.exec(ctx, 1, qn)
return workload.exec(ctx, 1, qn, "")
})
qn.EXPECT().QueryStream(mock.Anything, mock.Anything).Call.Return(
@ -797,7 +797,7 @@ func TestDeleteRunner_Run(t *testing.T) {
mockMgr.EXPECT().getOrCreateDmlStream(mock.Anything).Return(stream, nil)
mockMgr.EXPECT().getChannels(collectionID).Return(channels, nil)
lb.EXPECT().Execute(mock.Anything, mock.Anything).Call.Return(func(ctx context.Context, workload CollectionWorkLoad) error {
return workload.exec(ctx, 1, qn)
return workload.exec(ctx, 1, qn, "")
})
qn.EXPECT().QueryStream(mock.Anything, mock.Anything).Call.Return(
@ -899,7 +899,7 @@ func TestDeleteRunner_StreamingQueryAndDelteFunc(t *testing.T) {
qn := mocks.NewMockQueryNodeClient(t)
// witho out plan
queryFunc := dr.getStreamingQueryAndDelteFunc(nil)
assert.Error(t, queryFunc(ctx, 1, qn))
assert.Error(t, queryFunc(ctx, 1, qn, ""))
})
t.Run("partitionKey mode get meta failed", func(t *testing.T) {
@ -938,7 +938,7 @@ func TestDeleteRunner_StreamingQueryAndDelteFunc(t *testing.T) {
plan, err := planparserv2.CreateRetrievePlan(dr.schema.CollectionSchema, dr.req.Expr)
assert.NoError(t, err)
queryFunc := dr.getStreamingQueryAndDelteFunc(plan)
assert.Error(t, queryFunc(ctx, 1, qn))
assert.Error(t, queryFunc(ctx, 1, qn, ""))
})
t.Run("partitionKey mode get partition ID failed", func(t *testing.T) {
@ -981,6 +981,6 @@ func TestDeleteRunner_StreamingQueryAndDelteFunc(t *testing.T) {
plan, err := planparserv2.CreateRetrievePlan(dr.schema.CollectionSchema, dr.req.Expr)
assert.NoError(t, err)
queryFunc := dr.getStreamingQueryAndDelteFunc(plan)
assert.Error(t, queryFunc(ctx, 1, qn))
assert.Error(t, queryFunc(ctx, 1, qn, ""))
})
}

22
internal/proxy/task_query.go
View File

@ -61,6 +61,8 @@ type queryTask struct {
plan *planpb.PlanNode
partitionKeyMode bool
lb LBPolicy
channelsMvcc map[string]Timestamp
fastSkip bool
}
type queryParams struct {
@ -466,19 +468,33 @@ func (t *queryTask) PostExecute(ctx context.Context) error {
return nil
}
func (t *queryTask) queryShard(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channelIDs ...string) error {
func (t *queryTask) queryShard(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channel string) error {
needOverrideMvcc := false
mvccTs := t.MvccTimestamp
if len(t.channelsMvcc) > 0 {
mvccTs, needOverrideMvcc = t.channelsMvcc[channel]
// In fast mode, if there is no corresponding channel in channelsMvcc, quickly skip this query.
if !needOverrideMvcc && t.fastSkip {
return nil
}
}
retrieveReq := typeutil.Clone(t.RetrieveRequest)
retrieveReq.GetBase().TargetID = nodeID
if needOverrideMvcc && mvccTs > 0 {
retrieveReq.MvccTimestamp = mvccTs
}
req := &querypb.QueryRequest{
Req: retrieveReq,
DmlChannels: channelIDs,
DmlChannels: []string{channel},
Scope: querypb.DataScope_All,
}
log := log.Ctx(ctx).With(zap.Int64("collection", t.GetCollectionID()),
zap.Int64s("partitionIDs", t.GetPartitionIDs()),
zap.Int64("nodeID", nodeID),
zap.Strings("channels", channelIDs))
zap.String("channel", channel))
result, err := qn.Query(ctx, req)
if err != nil {

39
internal/proxy/task_search.go
View File

@ -63,9 +63,10 @@ type searchTask struct {
offset int64
resultBuf *typeutil.ConcurrentSet[*internalpb.SearchResults]
qc types.QueryCoordClient
node types.ProxyComponent
lb LBPolicy
qc types.QueryCoordClient
node types.ProxyComponent
lb LBPolicy
queryChannelsTs map[string]Timestamp
}
func getPartitionIDs(ctx context.Context, dbName string, collectionName string, partitionNames []string) (partitionIDs []UniqueID, err error) {
@ -456,6 +457,13 @@ func (t *searchTask) PostExecute(ctx context.Context) error {
return err
}
t.queryChannelsTs = make(map[string]uint64)
for _, r := range toReduceResults {
for ch, ts := range r.GetChannelsMvcc() {
t.queryChannelsTs[ch] = ts
}
}
if len(toReduceResults) >= 1 {
MetricType = toReduceResults[0].GetMetricType()
}
@ -513,20 +521,20 @@ func (t *searchTask) PostExecute(ctx context.Context) error {
return nil
}
func (t *searchTask) searchShard(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channelIDs ...string) error {
func (t *searchTask) searchShard(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channel string) error {
searchReq := typeutil.Clone(t.SearchRequest)
searchReq.GetBase().TargetID = nodeID
req := &querypb.SearchRequest{
Req: searchReq,
DmlChannels: channelIDs,
DmlChannels: []string{channel},
Scope: querypb.DataScope_All,
TotalChannelNum: int32(len(channelIDs)),
TotalChannelNum: int32(1),
}
log := log.Ctx(ctx).With(zap.Int64("collection", t.GetCollectionID()),
zap.Int64s("partitionIDs", t.GetPartitionIDs()),
zap.Int64("nodeID", nodeID),
zap.Strings("channels", channelIDs))
zap.String("channel", channel))
var result *internalpb.SearchResults
var err error
@ -593,6 +601,10 @@ func (t *searchTask) Requery() error {
GuaranteeTimestamp: t.request.GetGuaranteeTimestamp(),
QueryParams: t.request.GetSearchParams(),
}
channelsMvcc := make(map[string]Timestamp)
for k, v := range t.queryChannelsTs {
channelsMvcc[k] = v
}
qt := &queryTask{
ctx: t.ctx,
Condition: NewTaskCondition(t.ctx),
@ -603,10 +615,12 @@ func (t *searchTask) Requery() error {
),
ReqID: paramtable.GetNodeID(),
},
request: queryReq,
plan: plan,
qc: t.node.(*Proxy).queryCoord,
lb: t.node.(*Proxy).lbPolicy,
request: queryReq,
plan: plan,
qc: t.node.(*Proxy).queryCoord,
lb: t.node.(*Proxy).lbPolicy,
channelsMvcc: channelsMvcc,
fastSkip: true,
}
queryResult, err := t.node.(*Proxy).query(t.ctx, qt)
if err != nil {
@ -823,7 +837,7 @@ func reduceSearchResultData(ctx context.Context, subSearchResultData []*schemapb
zap.Int64("nq", sData.NumQueries),
zap.Int64("topk", sData.TopK),
zap.Int("length of pks", pkLength),
zap.Any("length of FieldsData", len(sData.FieldsData)))
zap.Int("length of FieldsData", len(sData.FieldsData)))
if err := checkSearchResultData(sData, nq, topk); err != nil {
log.Ctx(ctx).Warn("invalid search results", zap.Error(err))
return ret, err
@ -850,6 +864,7 @@ func reduceSearchResultData(ctx context.Context, subSearchResultData []*schemapb
var retSize int64
maxOutputSize := paramtable.Get().QuotaConfig.MaxOutputSize.GetAsInt64()
// reducing nq * topk results
for i := int64(0); i < nq; i++ {
var (

6
internal/proxy/task_search_test.go
View File

@ -1985,7 +1985,7 @@ func TestSearchTask_Requery(t *testing.T) {
lb := NewMockLBPolicy(t)
lb.EXPECT().Execute(mock.Anything, mock.Anything).Run(func(ctx context.Context, workload CollectionWorkLoad) {
err = workload.exec(ctx, 0, qn)
err = workload.exec(ctx, 0, qn, "")
assert.NoError(t, err)
}).Return(nil)
lb.EXPECT().UpdateCostMetrics(mock.Anything, mock.Anything).Return()
@ -2065,7 +2065,7 @@ func TestSearchTask_Requery(t *testing.T) {
lb := NewMockLBPolicy(t)
lb.EXPECT().Execute(mock.Anything, mock.Anything).Run(func(ctx context.Context, workload CollectionWorkLoad) {
_ = workload.exec(ctx, 0, qn)
_ = workload.exec(ctx, 0, qn, "")
}).Return(fmt.Errorf("mock err 1"))
node.lbPolicy = lb
@ -2099,7 +2099,7 @@ func TestSearchTask_Requery(t *testing.T) {
lb := NewMockLBPolicy(t)
lb.EXPECT().Execute(mock.Anything, mock.Anything).Run(func(ctx context.Context, workload CollectionWorkLoad) {
_ = workload.exec(ctx, 0, qn)
_ = workload.exec(ctx, 0, qn, "")
}).Return(fmt.Errorf("mock err 1"))
node.lbPolicy = lb

8
internal/proxy/task_statistic.go
View File

@ -273,19 +273,19 @@ func (g *getStatisticsTask) getStatisticsFromQueryNode(ctx context.Context) erro
return nil
}
func (g *getStatisticsTask) getStatisticsShard(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channelIDs ...string) error {
func (g *getStatisticsTask) getStatisticsShard(ctx context.Context, nodeID int64, qn types.QueryNodeClient, channel string) error {
nodeReq := proto.Clone(g.GetStatisticsRequest).(*internalpb.GetStatisticsRequest)
nodeReq.Base.TargetID = nodeID
req := &querypb.GetStatisticsRequest{
Req: nodeReq,
DmlChannels: channelIDs,
DmlChannels: []string{channel},
Scope: querypb.DataScope_All,
}
result, err := qn.GetStatistics(ctx, req)
if err != nil {
log.Warn("QueryNode statistic return error",
zap.Int64("nodeID", nodeID),
zap.Strings("channels", channelIDs),
zap.String("channel", channel),
zap.Error(err))
globalMetaCache.DeprecateShardCache(g.request.GetDbName(), g.collectionName)
return err
@ -293,7 +293,7 @@ func (g *getStatisticsTask) getStatisticsShard(ctx context.Context, nodeID int64
if result.GetStatus().GetErrorCode() == commonpb.ErrorCode_NotShardLeader {
log.Warn("QueryNode is not shardLeader",
zap.Int64("nodeID", nodeID),
zap.Strings("channels", channelIDs))
zap.String("channel", channel))
globalMetaCache.DeprecateShardCache(g.request.GetDbName(), g.collectionName)
return errInvalidShardLeaders
}

34
internal/querynodev2/delegator/delegator.go
View File

@ -200,11 +200,14 @@ func (sd *shardDelegator) Search(ctx context.Context, req *querypb.SearchRequest
// wait tsafe
waitTr := timerecord.NewTimeRecorder("wait tSafe")
err := sd.waitTSafe(ctx, req.Req.GuaranteeTimestamp)
tSafe, err := sd.waitTSafe(ctx, req.Req.GuaranteeTimestamp)
if err != nil {
log.Warn("delegator search failed to wait tsafe", zap.Error(err))
return nil, err
}
if req.GetReq().GetMvccTimestamp() == 0 {
req.Req.MvccTimestamp = tSafe
}
metrics.QueryNodeSQLatencyWaitTSafe.WithLabelValues(
fmt.Sprint(paramtable.GetNodeID()), metrics.SearchLabel).
Observe(float64(waitTr.ElapseSpan().Milliseconds()))
@ -275,11 +278,14 @@ func (sd *shardDelegator) QueryStream(ctx context.Context, req *querypb.QueryReq
// wait tsafe
waitTr := timerecord.NewTimeRecorder("wait tSafe")
err := sd.waitTSafe(ctx, req.Req.GuaranteeTimestamp)
tSafe, err := sd.waitTSafe(ctx, req.Req.GuaranteeTimestamp)
if err != nil {
log.Warn("delegator query failed to wait tsafe", zap.Error(err))
return err
}
if req.GetReq().GetMvccTimestamp() == 0 {
req.Req.MvccTimestamp = tSafe
}
metrics.QueryNodeSQLatencyWaitTSafe.WithLabelValues(
fmt.Sprint(paramtable.GetNodeID()), metrics.QueryLabel).
Observe(float64(waitTr.ElapseSpan().Milliseconds()))
@ -343,11 +349,14 @@ func (sd *shardDelegator) Query(ctx context.Context, req *querypb.QueryRequest)
// wait tsafe
waitTr := timerecord.NewTimeRecorder("wait tSafe")
err := sd.waitTSafe(ctx, req.Req.GuaranteeTimestamp)
tSafe, err := sd.waitTSafe(ctx, req.Req.GuaranteeTimestamp)
if err != nil {
log.Warn("delegator query failed to wait tsafe", zap.Error(err))
return nil, err
}
if req.GetReq().GetMvccTimestamp() == 0 {
req.Req.MvccTimestamp = tSafe
}
metrics.QueryNodeSQLatencyWaitTSafe.WithLabelValues(
fmt.Sprint(paramtable.GetNodeID()), metrics.QueryLabel).
Observe(float64(waitTr.ElapseSpan().Milliseconds()))
@ -406,7 +415,7 @@ func (sd *shardDelegator) GetStatistics(ctx context.Context, req *querypb.GetSta
}
// wait tsafe
err := sd.waitTSafe(ctx, req.Req.GuaranteeTimestamp)
_, err := sd.waitTSafe(ctx, req.Req.GuaranteeTimestamp)
if err != nil {
log.Warn("delegator GetStatistics failed to wait tsafe", zap.Error(err))
return nil, err
@ -547,14 +556,15 @@ func executeSubTasks[T any, R interface {
}
// waitTSafe returns when tsafe listener notifies a timestamp which meet the guarantee ts.
func (sd *shardDelegator) waitTSafe(ctx context.Context, ts uint64) error {
func (sd *shardDelegator) waitTSafe(ctx context.Context, ts uint64) (uint64, error) {
log := sd.getLogger(ctx)
// already safe to search
if sd.latestTsafe.Load() >= ts {
return nil
latestTSafe := sd.latestTsafe.Load()
if latestTSafe >= ts {
return latestTSafe, nil
}
// check lag duration too large
st, _ := tsoutil.ParseTS(sd.latestTsafe.Load())
st, _ := tsoutil.ParseTS(latestTSafe)
gt, _ := tsoutil.ParseTS(ts)
lag := gt.Sub(st)
maxLag := paramtable.Get().QueryNodeCfg.MaxTimestampLag.GetAsDuration(time.Second)
@ -565,7 +575,7 @@ func (sd *shardDelegator) waitTSafe(ctx context.Context, ts uint64) error {
zap.Duration("lag", lag),
zap.Duration("maxTsLag", maxLag),
)
return WrapErrTsLagTooLarge(lag, maxLag)
return 0, WrapErrTsLagTooLarge(lag, maxLag)
}
ch := make(chan struct{})
@ -587,12 +597,12 @@ func (sd *shardDelegator) waitTSafe(ctx context.Context, ts uint64) error {
case <-ctx.Done():
// notify wait goroutine to quit
sd.tsCond.Broadcast()
return ctx.Err()
return 0, ctx.Err()
case <-ch:
if !sd.Serviceable() {
return merr.WrapErrChannelNotAvailable(sd.vchannelName, "delegator closed during wait tsafe")
return 0, merr.WrapErrChannelNotAvailable(sd.vchannelName, "delegator closed during wait tsafe")
}
return nil
return sd.latestTsafe.Load(), nil
}
}
}

1
internal/querynodev2/handlers.go
View File

@ -359,7 +359,6 @@ func (node *QueryNode) searchChannel(ctx context.Context, req *querypb.SearchReq
metrics.QueryNodeSQCount.WithLabelValues(fmt.Sprint(paramtable.GetNodeID()), metrics.SearchLabel, metrics.SuccessLabel, metrics.Leader).Inc()
metrics.QueryNodeSearchNQ.WithLabelValues(fmt.Sprint(paramtable.GetNodeID())).Observe(float64(req.Req.GetNq()))
metrics.QueryNodeSearchTopK.WithLabelValues(fmt.Sprint(paramtable.GetNodeID())).Observe(float64(req.Req.GetTopk()))
return resp, nil
}

2
internal/querynodev2/segments/plan.go
View File

@ -90,6 +90,7 @@ type SearchRequest struct {
cPlaceholderGroup C.CPlaceholderGroup
msgID UniqueID
searchFieldID UniqueID
mvccTimestamp Timestamp
}
func NewSearchRequest(ctx context.Context, collection *Collection, req *querypb.SearchRequest, placeholderGrp []byte) (*SearchRequest, error) {
@ -129,6 +130,7 @@ func NewSearchRequest(ctx context.Context, collection *Collection, req *querypb.
cPlaceholderGroup: cPlaceholderGroup,
msgID: req.GetReq().GetBase().GetMsgID(),
searchFieldID: int64(fieldID),
mvccTimestamp: req.GetReq().GetMvccTimestamp(),
}
return ret, nil

2
internal/querynodev2/segments/reduce_test.go
View File

@ -34,6 +34,7 @@ import (
"github.com/milvus-io/milvus/pkg/common"
"github.com/milvus-io/milvus/pkg/util/funcutil"
"github.com/milvus-io/milvus/pkg/util/paramtable"
"github.com/milvus-io/milvus/pkg/util/typeutil"
)
type ReduceSuite struct {
@ -166,6 +167,7 @@ func (suite *ReduceSuite) TestReduceAllFunc() {
plan, err := createSearchPlanByExpr(context.Background(), suite.collection, serializedPlan, "")
suite.NoError(err)
searchReq, err := parseSearchRequest(context.Background(), plan, placeGroupByte)
searchReq.mvccTimestamp = typeutil.MaxTimestamp
suite.NoError(err)
defer searchReq.Delete()

8
internal/querynodev2/segments/result.go
View File

@ -49,6 +49,12 @@ func ReduceSearchResults(ctx context.Context, results []*internalpb.SearchResult
return results[0], nil
}
channelsMvcc := make(map[string]uint64)
for _, r := range results {
for ch, ts := range r.GetChannelsMvcc() {
channelsMvcc[ch] = ts
}
}
log := log.Ctx(ctx)
searchResultData, err := DecodeSearchResults(results)
@ -88,7 +94,7 @@ func ReduceSearchResults(ctx context.Context, results []*internalpb.SearchResult
return nil, false
})
searchResults.CostAggregation = mergeRequestCost(requestCosts)
searchResults.ChannelsMvcc = channelsMvcc
return searchResults, nil
}

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

@ -372,6 +372,7 @@ func (s *LocalSegment) Search(ctx context.Context, searchReq *SearchRequest) (*S
searchReq.plan.cSearchPlan,
searchReq.cPlaceholderGroup,
traceCtx,
C.uint64_t(searchReq.mvccTimestamp),
&searchResult.cSearchResult,
)
metrics.QueryNodeSQSegmentLatencyInCore.WithLabelValues(fmt.Sprint(paramtable.GetNodeID()), metrics.SearchLabel).Observe(float64(tr.ElapseSpan().Milliseconds()))

14
internal/querynodev2/services.go
View File

@ -658,8 +658,13 @@ func (node *QueryNode) SearchSegments(ctx context.Context, req *querypb.SearchRe
zap.String("channel", channel),
zap.String("scope", req.GetScope().String()),
)
resp := &internalpb.SearchResults{}
channelsMvcc := make(map[string]uint64)
for _, ch := range req.GetDmlChannels() {
channelsMvcc[ch] = req.GetReq().GetMvccTimestamp()
}
resp := &internalpb.SearchResults{
ChannelsMvcc: channelsMvcc,
}
if err := node.lifetime.Add(merr.IsHealthy); err != nil {
resp.Status = merr.Status(err)
return resp, nil
@ -790,7 +795,6 @@ func (node *QueryNode) Search(ctx context.Context, req *querypb.SearchRequest) (
Scope: req.Scope,
TotalChannelNum: req.TotalChannelNum,
}
runningGp.Go(func() error {
ret, err := node.searchChannel(runningCtx, req, ch)
mu.Lock()
@ -889,7 +893,7 @@ func (node *QueryNode) QuerySegments(ctx context.Context, req *querypb.QueryRequ
return resp, nil
}
tr.CtxElapse(ctx, fmt.Sprintf("do query done, traceID = %s, fromSharedLeader = %t, vChannel = %s, segmentIDs = %v",
tr.CtxElapse(ctx, fmt.Sprintf("do query done, traceID = %s, fromShardLeader = %t, vChannel = %s, segmentIDs = %v",
traceID,
req.GetFromShardLeader(),
channel,
@ -1101,7 +1105,7 @@ func (node *QueryNode) QueryStreamSegments(req *querypb.QueryRequest, srv queryp
return nil
}
tr.CtxElapse(ctx, fmt.Sprintf("do query done, traceID = %s, fromSharedLeader = %t, vChannel = %s, segmentIDs = %v",
tr.CtxElapse(ctx, fmt.Sprintf("do query done, traceID = %s, fromShardLeader = %t, vChannel = %s, segmentIDs = %v",
traceID,
req.GetFromShardLeader(),
channel,

1
internal/querynodev2/services_test.go
View File

@ -1149,6 +1149,7 @@ func (suite *ServiceSuite) genCSearchRequest(nq int64, indexType string, schema
PlaceholderGroup: placeHolder,
DslType: commonpb.DslType_BoolExprV1,
Nq: nq,
MvccTimestamp: typeutil.MaxTimestamp,
}, nil
}

8
internal/querynodev2/tasks/task.go
View File

@ -257,6 +257,7 @@ func (t *SearchTask) Merge(other *SearchTask) bool {
// Check mergeable
if t.req.GetReq().GetDbID() != other.req.GetReq().GetDbID() ||
t.req.GetReq().GetCollectionID() != other.req.GetReq().GetCollectionID() ||
t.req.GetReq().GetMvccTimestamp() != other.req.GetReq().GetMvccTimestamp() ||
t.req.GetReq().GetDslType() != other.req.GetReq().GetDslType() ||
t.req.GetDmlChannels()[0] != other.req.GetDmlChannels()[0] ||
nq+otherNq > paramtable.Get().QueryNodeCfg.MaxGroupNQ.GetAsInt64() ||
@ -300,6 +301,13 @@ func (t *SearchTask) Wait() error {
}
func (t *SearchTask) Result() *internalpb.SearchResults {
if t.result != nil {
channelsMvcc := make(map[string]uint64)
for _, ch := range t.req.GetDmlChannels() {
channelsMvcc[ch] = t.req.GetReq().GetMvccTimestamp()
}
t.result.ChannelsMvcc = channelsMvcc
}
return t.result
}