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

fix the merge conflict 

Former-commit-id: 187d167cc412129e93cd9eec333e33c74368f370
pull/191/head
Heisenberg 2019-09-10 18:28:52 +08:00
parent 63d7157d52 fe966d81cb
commit fc98a3a2d2
23 changed files with 333 additions and 220 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
cpp/CMakeLists.txt
View File

@ -31,7 +31,6 @@ if(CMAKE_BUILD_TYPE STREQUAL "Release")
set(BUILD_TYPE "release")
else()
set(BUILD_TYPE "debug")
SET(CMAKE_VERBOSE_MAKEFILE on)
endif()
message(STATUS "Build type = ${BUILD_TYPE}")

2
cpp/conf/server_config.template
View File

@ -8,7 +8,6 @@ server_config:
db_config:
db_path: @MILVUS_DB_PATH@ # milvus data storage path
db_slave_path: # secondry data storage path, split by semicolon
parallel_reduce: false # use multi-threads to reduce topk result
# URI format: dialect://username:password@host:port/database
# All parts except dialect are optional, but you MUST include the delimiters
@ -40,7 +39,6 @@ cache_config:
engine_config:
use_blas_threshold: 20
omp_thread_num: 0 # how many compute threads be used by engine, 0 means use all cpu core to compute
resource_config:
# resource list, length: 0~N

12
cpp/src/core/include/knowhere/index/vector_index/idmap.h
View File

@ -32,6 +32,12 @@ class IDMAP : public VectorIndex, public BasicIndex {
virtual int64_t *GetRawIds();
protected:
virtual void search_impl(int64_t n,
const float *data,
int64_t k,
float *distances,
int64_t *labels,
const Config &cfg);
std::mutex mutex_;
};
@ -49,6 +55,12 @@ class GPUIDMAP : public IDMAP, public GPUIndex {
VectorIndexPtr CopyGpuToGpu(const int64_t &device_id, const Config &config) override;
protected:
void search_impl(int64_t n,
const float *data,
int64_t k,
float *distances,
int64_t *labels,
const Config &cfg) override;
BinarySet SerializeImpl() override;
void LoadImpl(const BinarySet &index_binary) override;
};

44
cpp/src/core/src/knowhere/index/preprocessor/normalize.cpp
View File

@ -9,32 +9,32 @@ namespace knowhere {
DatasetPtr
NormalizePreprocessor::Preprocess(const DatasetPtr &dataset) {
// TODO: wrap dataset->tensor
auto tensor = dataset->tensor()[0];
auto p_data = (float *)tensor->raw_mutable_data();
auto dimension = tensor->shape()[1];
auto rows = tensor->shape()[0];
#pragma omp parallel for
for (auto i = 0; i < rows; ++i) {
Normalize(&(p_data[i * dimension]), dimension);
}
// // TODO: wrap dataset->tensor
// auto tensor = dataset->tensor()[0];
// auto p_data = (float *)tensor->raw_mutable_data();
// auto dimension = tensor->shape()[1];
// auto rows = tensor->shape()[0];
//
//#pragma omp parallel for
// for (auto i = 0; i < rows; ++i) {
// Normalize(&(p_data[i * dimension]), dimension);
// }
}
void
NormalizePreprocessor::Normalize(float *arr, int64_t dimension) {
double vector_length = 0;
for (auto j = 0; j < dimension; j++) {
double val = arr[j];
vector_length += val * val;
}
vector_length = std::sqrt(vector_length);
if (vector_length < 1e-6) {
auto val = (float) (1.0 / std::sqrt((double) dimension));
for (int j = 0; j < dimension; j++) arr[j] = val;
} else {
for (int j = 0; j < dimension; j++) arr[j] = (float) (arr[j] / vector_length);
}
//double vector_length = 0;
//for (auto j = 0; j < dimension; j++) {
// double val = arr[j];
// vector_length += val * val;
//}
//vector_length = std::sqrt(vector_length);
//if (vector_length < 1e-6) {
// auto val = (float) (1.0 / std::sqrt((double) dimension));
// for (int j = 0; j < dimension; j++) arr[j] = val;
//} else {
// for (int j = 0; j < dimension; j++) arr[j] = (float) (arr[j] / vector_length);
//}
}
} // namespace knowhere

12
cpp/src/core/src/knowhere/index/vector_index/cpu_kdt_rng.cpp
View File

@ -70,10 +70,10 @@ CPUKDTRNG::Train(const DatasetPtr &origin, const Config &train_config) {
SetParameters(train_config);
DatasetPtr dataset = origin->Clone();
if (index_ptr_->GetDistCalcMethod() == SPTAG::DistCalcMethod::Cosine
&& preprocessor_) {
//if (index_ptr_->GetDistCalcMethod() == SPTAG::DistCalcMethod::Cosine
// && preprocessor_) {
preprocessor_->Preprocess(dataset);
}
//}
auto vectorset = ConvertToVectorSet(dataset);
auto metaset = ConvertToMetadataSet(dataset);
@ -88,10 +88,10 @@ CPUKDTRNG::Add(const DatasetPtr &origin, const Config &add_config) {
SetParameters(add_config);
DatasetPtr dataset = origin->Clone();
if (index_ptr_->GetDistCalcMethod() == SPTAG::DistCalcMethod::Cosine
&& preprocessor_) {
//if (index_ptr_->GetDistCalcMethod() == SPTAG::DistCalcMethod::Cosine
// && preprocessor_) {
preprocessor_->Preprocess(dataset);
}
//}
auto vectorset = ConvertToVectorSet(dataset);
auto metaset = ConvertToMetadataSet(dataset);

14
cpp/src/core/src/knowhere/index/vector_index/gpu_ivf.cpp
View File

@ -280,12 +280,15 @@ void FaissGpuResourceMgr::InitResource() {
is_init = true;
//std::cout << "InitResource" << std::endl;
for(auto& device : devices_params_) {
auto& device_id = device.first;
//std::cout << "Device Id: " << device_id << std::endl;
auto& device_param = device.second;
auto& bq = idle_map[device_id];
for (int64_t i = 0; i < device_param.resource_num; ++i) {
//std::cout << "Resource Id: " << i << std::endl;
auto raw_resource = std::make_shared<faiss::gpu::StandardGpuResources>();
// TODO(linxj): enable set pinned memory
@ -295,6 +298,7 @@ void FaissGpuResourceMgr::InitResource() {
bq.Put(res_wrapper);
}
}
//std::cout << "End initResource" << std::endl;
}
ResPtr FaissGpuResourceMgr::GetRes(const int64_t &device_id,
@ -311,16 +315,6 @@ ResPtr FaissGpuResourceMgr::GetRes(const int64_t &device_id,
return nullptr;
}
//bool FaissGpuResourceMgr::GetRes(const int64_t &device_id,
// ResPtr &res,
// const int64_t &alloc_size) {
// InitResource();
//
// std::lock_guard<std::mutex> lk(res->mutex);
// AllocateTempMem(res, device_id, alloc_size);
// return true;
//}
void FaissGpuResourceMgr::MoveToIdle(const int64_t &device_id, const ResPtr &res) {
auto finder = idle_map.find(device_id);
if (finder != idle_map.end()) {

18
cpp/src/core/src/knowhere/index/vector_index/idmap.cpp
View File

@ -50,7 +50,7 @@ DatasetPtr IDMAP::Search(const DatasetPtr &dataset, const Config &config) {
auto res_ids = (int64_t *) malloc(sizeof(int64_t) * elems);
auto res_dis = (float *) malloc(sizeof(float) * elems);
index_->search(rows, (float *) p_data, k, res_dis, res_ids);
search_impl(rows, (float *) p_data, k, res_dis, res_ids, Config());
auto id_buf = MakeMutableBufferSmart((uint8_t *) res_ids, sizeof(int64_t) * elems);
auto dist_buf = MakeMutableBufferSmart((uint8_t *) res_dis, sizeof(float) * elems);
@ -72,6 +72,11 @@ DatasetPtr IDMAP::Search(const DatasetPtr &dataset, const Config &config) {
return std::make_shared<Dataset>(array, nullptr);
}
void IDMAP::search_impl(int64_t n, const float *data, int64_t k, float *distances, int64_t *labels, const Config &cfg) {
index_->search(n, (float *) data, k, distances, labels);
}
void IDMAP::Add(const DatasetPtr &dataset, const Config &config) {
if (!index_) {
KNOWHERE_THROW_MSG("index not initialize");
@ -207,6 +212,7 @@ void GPUIDMAP::LoadImpl(const BinarySet &index_binary) {
if (auto res = FaissGpuResourceMgr::GetInstance().GetRes(gpu_id_) ){
ResScope rs(gpu_id_, res);
res_ = res;
auto device_index = faiss::gpu::index_cpu_to_gpu(res->faiss_res.get(), gpu_id_, index);
index_.reset(device_index);
} else {
@ -230,5 +236,15 @@ int64_t *GPUIDMAP::GetRawIds() {
KNOWHERE_THROW_MSG("Not support");
}
void GPUIDMAP::search_impl(int64_t n,
const float *data,
int64_t k,
float *distances,
int64_t *labels,
const Config &cfg) {
ResScope rs(res_);
index_->search(n, (float *) data, k, distances, labels);
}
}
}

35
cpp/src/core/test/test_idmap.cpp
View File

@ -12,6 +12,7 @@
#include "knowhere/index/vector_index/idmap.h"
#include "knowhere/adapter/structure.h"
#include "knowhere/index/vector_index/cloner.h"
#include "knowhere/common/exception.h"
#include "utils.h"
@ -65,19 +66,20 @@ void PrintResult(const DatasetPtr &result,
}
TEST_F(IDMAPTest, idmap_basic) {
assert(!xb.empty());
ASSERT_TRUE(!xb.empty());
Config Default_cfg;
index_->Train(Config::object{{"dim", dim}, {"metric_type", "L2"}});
index_->Add(base_dataset, Default_cfg);
EXPECT_EQ(index_->Count(), nb);
EXPECT_EQ(index_->Dimension(), dim);
assert(index_->GetRawVectors() != nullptr);
assert(index_->GetRawIds() != nullptr);
ASSERT_TRUE(index_->GetRawVectors() != nullptr);
ASSERT_TRUE(index_->GetRawIds() != nullptr);
auto result = index_->Search(query_dataset, Config::object{{"k", k}});
AssertAnns(result, nq, k);
PrintResult(result, nq, k);
index_->Seal();
auto binaryset = index_->Serialize();
auto new_index = std::make_shared<IDMAP>();
new_index->Load(binaryset);
@ -126,15 +128,15 @@ TEST_F(IDMAPTest, idmap_serialize) {
}
TEST_F(IDMAPTest, copy_test) {
assert(!xb.empty());
ASSERT_TRUE(!xb.empty());
Config Default_cfg;
index_->Train(Config::object{{"dim", dim}, {"metric_type", "L2"}});
index_->Add(base_dataset, Default_cfg);
EXPECT_EQ(index_->Count(), nb);
EXPECT_EQ(index_->Dimension(), dim);
assert(index_->GetRawVectors() != nullptr);
assert(index_->GetRawIds() != nullptr);
ASSERT_TRUE(index_->GetRawVectors() != nullptr);
ASSERT_TRUE(index_->GetRawIds() != nullptr);
auto result = index_->Search(query_dataset, Config::object{{"k", k}});
AssertAnns(result, nq, k);
//PrintResult(result, nq, k);
@ -151,8 +153,16 @@ TEST_F(IDMAPTest, copy_test) {
auto clone_index = CopyCpuToGpu(index_, device_id, Config());
auto clone_result = clone_index->Search(query_dataset, Config::object{{"k", k}});
AssertAnns(clone_result, nq, k);
//assert(std::static_pointer_cast<GPUIDMAP>(clone_index)->GetRawVectors() != nullptr);
//assert(std::static_pointer_cast<GPUIDMAP>(clone_index)->GetRawIds() != nullptr);
ASSERT_THROW({ std::static_pointer_cast<GPUIDMAP>(clone_index)->GetRawVectors(); },
zilliz::knowhere::KnowhereException);
ASSERT_THROW({ std::static_pointer_cast<GPUIDMAP>(clone_index)->GetRawIds(); },
zilliz::knowhere::KnowhereException);
auto binary = clone_index->Serialize();
clone_index->Load(binary);
auto new_result = clone_index->Search(query_dataset, Config::object{{"k", k}});
AssertAnns(new_result, nq, k);
auto clone_gpu_idx = clone_index->Clone();
auto clone_gpu_res = clone_gpu_idx->Search(query_dataset, Config::object{{"k", k}});
AssertAnns(clone_gpu_res, nq, k);
@ -161,14 +171,13 @@ TEST_F(IDMAPTest, copy_test) {
auto host_index = CopyGpuToCpu(clone_index, Config());
auto host_result = host_index->Search(query_dataset, Config::object{{"k", k}});
AssertAnns(host_result, nq, k);
assert(std::static_pointer_cast<IDMAP>(host_index)->GetRawVectors() != nullptr);
assert(std::static_pointer_cast<IDMAP>(host_index)->GetRawIds() != nullptr);
ASSERT_TRUE(std::static_pointer_cast<IDMAP>(host_index)->GetRawVectors() != nullptr);
ASSERT_TRUE(std::static_pointer_cast<IDMAP>(host_index)->GetRawIds() != nullptr);
// gpu to gpu
auto device_index = CopyCpuToGpu(index_, device_id, Config());
auto device_result = device_index->Search(query_dataset, Config::object{{"k", k}});
auto new_device_index = std::static_pointer_cast<GPUIDMAP>(device_index)->CopyGpuToGpu(device_id, Config());
auto device_result = new_device_index->Search(query_dataset, Config::object{{"k", k}});
AssertAnns(device_result, nq, k);
//assert(std::static_pointer_cast<GPUIDMAP>(device_index)->GetRawVectors() != nullptr);
//assert(std::static_pointer_cast<GPUIDMAP>(device_index)->GetRawIds() != nullptr);
}
}

16
cpp/src/core/test/test_ivf.cpp
View File

@ -394,8 +394,11 @@ TEST_F(GPURESTEST, gpu_ivf_resource_test) {
{
index_type = "GPUIVF";
index_ = IndexFactory(index_type);
index_ = std::make_shared<GPUIVF>(-1);
ASSERT_EQ(std::dynamic_pointer_cast<GPUIVF>(index_)->GetGpuDevice(), -1);
std::dynamic_pointer_cast<GPUIVF>(index_)->SetGpuDevice(device_id);
ASSERT_EQ(std::dynamic_pointer_cast<GPUIVF>(index_)->GetGpuDevice(), device_id);
auto preprocessor = index_->BuildPreprocessor(base_dataset, preprocess_cfg);
index_->set_preprocessor(preprocessor);
train_cfg = Config::object{{"nlist", 1638}, {"gpu_id", device_id}, {"metric_type", "L2"}};
@ -412,8 +415,9 @@ TEST_F(GPURESTEST, gpu_ivf_resource_test) {
if (i > search_count - 6 || i < 5)
tc.RecordSection("search once");
}
tc.RecordSection("search all");
tc.ElapseFromBegin("search all");
}
FaissGpuResourceMgr::GetInstance().Dump();
{
// IVF-Search
@ -430,7 +434,7 @@ TEST_F(GPURESTEST, gpu_ivf_resource_test) {
if (i > search_count - 6 || i < 5)
tc.RecordSection("search once");
}
tc.RecordSection("search all");
tc.ElapseFromBegin("search all");
}
}
@ -461,7 +465,7 @@ TEST_F(GPURESTEST, gpuivfsq) {
if (i > search_count - 6 || i < 5)
tc.RecordSection("search once");
}
tc.RecordSection("search all");
tc.ElapseFromBegin("search all");
}
{
@ -493,7 +497,7 @@ TEST_F(GPURESTEST, gpuivfsq) {
if (i > search_count - 6 || i < 5)
tc.RecordSection("search once");
}
tc.RecordSection("search all");
tc.ElapseFromBegin("search all");
delete cpu_index;
delete search_idx;
}

9
cpp/src/core/test/test_nsg/test_nsg.cpp
View File

@ -7,6 +7,7 @@
#include <gtest/gtest.h>
#include <memory>
#include "knowhere/common/exception.h"
#include "knowhere/index/vector_index/gpu_ivf.h"
#include "knowhere/index/vector_index/nsg_index.h"
#include "knowhere/index/vector_index/nsg/nsg_io.h"
@ -71,6 +72,14 @@ TEST_P(NSGInterfaceTest, basic_test) {
auto new_result = new_index->Search(query_dataset, Config::object{{"k", k}});
AssertAnns(result, nq, k);
ASSERT_EQ(index_->Count(), nb);
ASSERT_EQ(index_->Dimension(), dim);
ASSERT_THROW({index_->Clone();}, zilliz::knowhere::KnowhereException);
ASSERT_NO_THROW({
index_->Add(base_dataset, Config());
index_->Seal();
});
{
//std::cout << "k = 1" << std::endl;
//new_index->Search(GenQuery(1), Config::object{{"k", 1}});

41
cpp/src/db/meta/MySQLMetaImpl.cpp
View File

@ -881,6 +881,7 @@ Status MySQLMetaImpl::FilesToIndex(TableFilesSchema &files) {
res = filesToIndexQuery.store();
} //Scoped Connection
Status ret;
std::map<std::string, TableSchema> groups;
TableFileSchema table_file;
for (auto &resRow : res) {
@ -925,16 +926,17 @@ Status MySQLMetaImpl::FilesToIndex(TableFilesSchema &files) {
auto status = utils::GetTableFilePath(options_, table_file);
if(!status.ok()) {
return status;
ret = status;
}
files.push_back(table_file);
}
return ret;
} catch (std::exception &e) {
return HandleException("GENERAL ERROR WHEN FINDING TABLE FILES TO INDEX", e.what());
}
return Status::OK();
}
Status MySQLMetaImpl::FilesToSearch(const std::string &table_id,
@ -998,6 +1000,7 @@ Status MySQLMetaImpl::FilesToSearch(const std::string &table_id,
return status;
}
Status ret;
TableFileSchema table_file;
for (auto &resRow : res) {
@ -1031,7 +1034,7 @@ Status MySQLMetaImpl::FilesToSearch(const std::string &table_id,
auto status = utils::GetTableFilePath(options_, table_file);
if(!status.ok()) {
return status;
ret = status;
}
auto dateItr = files.find(table_file.date_);
@ -1041,11 +1044,11 @@ Status MySQLMetaImpl::FilesToSearch(const std::string &table_id,
files[table_file.date_].push_back(table_file);
}
return ret;
} catch (std::exception &e) {
return HandleException("GENERAL ERROR WHEN FINDING TABLE FILES TO SEARCH", e.what());
}
return Status::OK();
}
Status MySQLMetaImpl::FilesToMerge(const std::string &table_id,
@ -1083,6 +1086,7 @@ Status MySQLMetaImpl::FilesToMerge(const std::string &table_id,
res = filesToMergeQuery.store();
} //Scoped Connection
Status ret;
for (auto &resRow : res) {
TableFileSchema table_file;
table_file.file_size_ = resRow["file_size"];
@ -1120,7 +1124,7 @@ Status MySQLMetaImpl::FilesToMerge(const std::string &table_id,
auto status = utils::GetTableFilePath(options_, table_file);
if(!status.ok()) {
return status;
ret = status;
}
auto dateItr = files.find(table_file.date_);
@ -1131,11 +1135,11 @@ Status MySQLMetaImpl::FilesToMerge(const std::string &table_id,
files[table_file.date_].push_back(table_file);
}
return ret;
} catch (std::exception &e) {
return HandleException("GENERAL ERROR WHEN FINDING TABLE FILES TO MERGE", e.what());
}
return Status::OK();
}
Status MySQLMetaImpl::GetTableFiles(const std::string &table_id,
@ -1165,7 +1169,8 @@ Status MySQLMetaImpl::GetTableFiles(const std::string &table_id,
getTableFileQuery << "SELECT id, engine_type, file_id, file_type, file_size, row_count, date, created_on " <<
"FROM TableFiles " <<
"WHERE table_id = " << quote << table_id << " AND " <<
"(" << idStr << ");";
"(" << idStr << ") AND " <<
"file_type <> " << std::to_string(TableFileSchema::TO_DELETE) << ";";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::GetTableFiles: " << getTableFileQuery.str();
@ -1174,11 +1179,9 @@ Status MySQLMetaImpl::GetTableFiles(const std::string &table_id,
TableSchema table_schema;
table_schema.table_id_ = table_id;
auto status = DescribeTable(table_schema);
if (!status.ok()) {
return status;
}
DescribeTable(table_schema);
Status ret;
for (auto &resRow : res) {
TableFileSchema file_schema;
@ -1211,18 +1214,16 @@ Status MySQLMetaImpl::GetTableFiles(const std::string &table_id,
file_schema.dimension_ = table_schema.dimension_;
auto status = utils::GetTableFilePath(options_, file_schema);
if(!status.ok()) {
return status;
}
utils::GetTableFilePath(options_, file_schema);
table_files.emplace_back(file_schema);
}
return ret;
} catch (std::exception &e) {
return HandleException("GENERAL ERROR WHEN RETRIEVING TABLE FILES", e.what());
}
return Status::OK();
}
// PXU TODO: Support Swap

38
cpp/src/db/meta/SqliteMetaImpl.cpp
View File

@ -603,6 +603,7 @@ Status SqliteMetaImpl::FilesToIndex(TableFilesSchema &files) {
std::map<std::string, TableSchema> groups;
TableFileSchema table_file;
Status ret;
for (auto &file : selected) {
table_file.id_ = std::get<0>(file);
table_file.table_id_ = std::get<1>(file);
@ -616,7 +617,7 @@ Status SqliteMetaImpl::FilesToIndex(TableFilesSchema &files) {
auto status = utils::GetTableFilePath(options_, table_file);
if(!status.ok()) {
return status;
ret = status;
}
auto groupItr = groups.find(table_file.table_id_);
if (groupItr == groups.end()) {
@ -635,11 +636,11 @@ Status SqliteMetaImpl::FilesToIndex(TableFilesSchema &files) {
files.push_back(table_file);
}
return ret;
} catch (std::exception &e) {
return HandleException("Encounter exception when iterate raw files", e.what());
}
return Status::OK();
}
Status SqliteMetaImpl::FilesToSearch(const std::string &table_id,
@ -695,6 +696,7 @@ Status SqliteMetaImpl::FilesToSearch(const std::string &table_id,
result = ConnectorPtr->select(select_columns, filter);
}
Status ret;
TableFileSchema table_file;
for (auto &file : result) {
table_file.id_ = std::get<0>(file);
@ -712,7 +714,7 @@ Status SqliteMetaImpl::FilesToSearch(const std::string &table_id,
auto status = utils::GetTableFilePath(options_, table_file);
if(!status.ok()) {
return status;
ret = status;
}
auto dateItr = files.find(table_file.date_);
@ -724,13 +726,12 @@ Status SqliteMetaImpl::FilesToSearch(const std::string &table_id,
if(files.empty()) {
ENGINE_LOG_ERROR << "No file to search for table: " << table_id;
}
return ret;
} catch (std::exception &e) {
return HandleException("Encounter exception when iterate index files", e.what());
}
return Status::OK();
}
Status SqliteMetaImpl::FilesToMerge(const std::string &table_id,
@ -761,6 +762,7 @@ Status SqliteMetaImpl::FilesToMerge(const std::string &table_id,
c(&TableFileSchema::table_id_) == table_id),
order_by(&TableFileSchema::file_size_).desc());
Status result;
for (auto &file : selected) {
TableFileSchema table_file;
table_file.file_size_ = std::get<4>(file);
@ -782,7 +784,7 @@ Status SqliteMetaImpl::FilesToMerge(const std::string &table_id,
auto status = utils::GetTableFilePath(options_, table_file);
if(!status.ok()) {
return status;
result = status;
}
auto dateItr = files.find(table_file.date_);
@ -791,11 +793,12 @@ Status SqliteMetaImpl::FilesToMerge(const std::string &table_id,
}
files[table_file.date_].push_back(table_file);
}
return result;
} catch (std::exception &e) {
return HandleException("Encounter exception when iterate merge files", e.what());
}
return Status::OK();
}
Status SqliteMetaImpl::GetTableFiles(const std::string& table_id,
@ -812,7 +815,8 @@ Status SqliteMetaImpl::GetTableFiles(const std::string& table_id,
&TableFileSchema::engine_type_,
&TableFileSchema::created_on_),
where(c(&TableFileSchema::table_id_) == table_id and
in(&TableFileSchema::id_, ids)
in(&TableFileSchema::id_, ids) and
c(&TableFileSchema::file_type_) != (int) TableFileSchema::TO_DELETE
));
TableSchema table_schema;
@ -822,6 +826,7 @@ Status SqliteMetaImpl::GetTableFiles(const std::string& table_id,
return status;
}
Status result;
for (auto &file : files) {
TableFileSchema file_schema;
file_schema.table_id_ = table_id;
@ -838,18 +843,15 @@ Status SqliteMetaImpl::GetTableFiles(const std::string& table_id,
file_schema.nlist_ = table_schema.nlist_;
file_schema.metric_type_ = table_schema.metric_type_;
auto status = utils::GetTableFilePath(options_, file_schema);
if(!status.ok()) {
return status;
}
utils::GetTableFilePath(options_, file_schema);
table_files.emplace_back(file_schema);
}
return result;
} catch (std::exception &e) {
return HandleException("Encounter exception when lookup table files", e.what());
}
return Status::OK();
}
// PXU TODO: Support Swap

151
cpp/src/server/ServerConfig.cpp
View File

@ -97,30 +97,30 @@ ServerConfig::CheckServerConfig() {
std::string ip_address = server_config.GetValue(CONFIG_SERVER_ADDRESS, "127.0.0.1");
if (ValidationUtil::ValidateIpAddress(ip_address) != SERVER_SUCCESS) {
std::cerr << "Error: invalid server IP address: " << ip_address << std::endl;
std::cerr << "ERROR: invalid server IP address: " << ip_address << std::endl;
okay = false;
}
std::string port_str = server_config.GetValue(CONFIG_SERVER_PORT, "19530");
if (ValidationUtil::ValidateStringIsNumber(port_str) != SERVER_SUCCESS) {
std::cerr << "Error: port " << port_str << " is not a number" << std::endl;
std::cerr << "ERROR: port " << port_str << " is not a number" << std::endl;
okay = false;
} else {
int32_t port = std::stol(port_str);
if (port < 1025 | port > 65534) {
std::cerr << "Error: port " << port_str << " out of range [1025, 65534]" << std::endl;
std::cerr << "ERROR: port " << port_str << " out of range [1025, 65534]" << std::endl;
okay = false;
}
}
std::string gpu_index_str = server_config.GetValue(CONFIG_GPU_INDEX, "0");
if (ValidationUtil::ValidateStringIsNumber(gpu_index_str) != SERVER_SUCCESS) {
std::cerr << "Error: gpu_index " << gpu_index_str << " is not a number" << std::endl;
std::cerr << "ERROR: gpu_index " << gpu_index_str << " is not a number" << std::endl;
okay = false;
} else {
int32_t gpu_index = std::stol(gpu_index_str);
if (ValidationUtil::ValidateGpuIndex(gpu_index) != SERVER_SUCCESS) {
std::cerr << "Error: invalid gpu_index " << gpu_index_str << std::endl;
std::cerr << "ERROR: invalid gpu_index " << gpu_index_str << std::endl;
okay = false;
}
}
@ -180,41 +180,44 @@ ServerConfig::CheckDBConfig() {
okay = false;
}
std::string parallel_reduce_str = db_config.GetValue(CONFIG_DB_PARALLEL_REDUCE, "false");
if (ValidationUtil::ValidateStringIsBool(parallel_reduce_str) != SERVER_SUCCESS) {
std::cerr << "Error: invalid parallel_reduce config: " << parallel_reduce_str << std::endl;
okay = false;
}
std::string db_backend_url = db_config.GetValue(CONFIG_DB_URL);
if (ValidationUtil::ValidateDbURI(db_backend_url) != SERVER_SUCCESS) {
std::cerr << "Error: invalid db_backend_url " << db_backend_url << std::endl;
std::cerr << "ERROR: invalid db_backend_url: " << db_backend_url << std::endl;
okay = false;
}
std::string archive_disk_threshold_str = db_config.GetValue(CONFIG_DB_INSERT_BUFFER_SIZE, "0");
if (ValidationUtil::ValidateStringIsNumber(archive_disk_threshold_str) != SERVER_SUCCESS) {
std::cerr << "Error: archive_disk_threshold " << archive_disk_threshold_str << " is not a number" << std::endl;
std::cerr << "ERROR: archive_disk_threshold " << archive_disk_threshold_str << " is not a number" << std::endl;
okay = false;
}
std::string archive_days_threshold_str = db_config.GetValue(CONFIG_DB_INSERT_BUFFER_SIZE, "0");
if (ValidationUtil::ValidateStringIsNumber(archive_days_threshold_str) != SERVER_SUCCESS) {
std::cerr << "Error: archive_days_threshold " << archive_days_threshold_str << " is not a number" << std::endl;
std::cerr << "ERROR: archive_days_threshold " << archive_days_threshold_str << " is not a number" << std::endl;
okay = false;
}
std::string insert_buffer_size_str = db_config.GetValue(CONFIG_DB_INSERT_BUFFER_SIZE, "4");
if (ValidationUtil::ValidateStringIsNumber(insert_buffer_size_str) != SERVER_SUCCESS) {
std::cerr << "Error: insert_buffer_size " << insert_buffer_size_str << " is not a number" << std::endl;
std::cerr << "ERROR: insert_buffer_size " << insert_buffer_size_str << " is not a number" << std::endl;
okay = false;
<<<<<<< HEAD
} else {
=======
}
else {
>>>>>>> branch-0.4.0
uint64_t insert_buffer_size = (uint64_t) std::stol(insert_buffer_size_str);
insert_buffer_size *= GB;
unsigned long total_mem = 0, free_mem = 0;
CommonUtil::GetSystemMemInfo(total_mem, free_mem);
if (insert_buffer_size >= total_mem) {
<<<<<<< HEAD
std::cerr << "Error: insert_buffer_size exceed system memory" << std::endl;
=======
std::cerr << "ERROR: insert_buffer_size exceed system memory" << std::endl;
>>>>>>> branch-0.4.0
okay = false;
}
}
@ -238,13 +241,13 @@ ServerConfig::CheckMetricConfig() {
std::string is_startup_str = metric_config.GetValue(CONFIG_METRIC_IS_STARTUP, "off");
if (ValidationUtil::ValidateStringIsBool(is_startup_str) != SERVER_SUCCESS) {
std::cerr << "Error: invalid is_startup config: " << is_startup_str << std::endl;
std::cerr << "ERROR: invalid is_startup config: " << is_startup_str << std::endl;
okay = false;
}
std::string port_str = metric_config.GetChild(CONFIG_PROMETHEUS).GetValue(CONFIG_METRIC_PROMETHEUS_PORT, "8080");
if (ValidationUtil::ValidateStringIsNumber(port_str) != SERVER_SUCCESS) {
std::cerr << "Error: port specified in prometheus_config " << port_str << " is not a number" << std::endl;
std::cerr << "ERROR: port specified in prometheus_config " << port_str << " is not a number" << std::endl;
okay = false;
}
@ -269,24 +272,34 @@ ServerConfig::CheckCacheConfig() {
std::string cpu_cache_capacity_str = cache_config.GetValue(CONFIG_CPU_CACHE_CAPACITY, "16");
if (ValidationUtil::ValidateStringIsNumber(cpu_cache_capacity_str) != SERVER_SUCCESS) {
std::cerr << "Error: cpu_cache_capacity " << cpu_cache_capacity_str << " is not a number" << std::endl;
std::cerr << "ERROR: cpu_cache_capacity " << cpu_cache_capacity_str << " is not a number" << std::endl;
okay = false;
<<<<<<< HEAD
} else {
=======
}
else {
>>>>>>> branch-0.4.0
uint64_t cpu_cache_capacity = (uint64_t) std::stol(cpu_cache_capacity_str);
cpu_cache_capacity *= GB;
unsigned long total_mem = 0, free_mem = 0;
CommonUtil::GetSystemMemInfo(total_mem, free_mem);
if (cpu_cache_capacity >= total_mem) {
std::cerr << "Error: cpu_cache_capacity exceed system memory" << std::endl;
std::cerr << "ERROR: cpu_cache_capacity exceed system memory" << std::endl;
okay = false;
<<<<<<< HEAD
} else if (cpu_cache_capacity > (double) total_mem * 0.9) {
=======
}
else if (cpu_cache_capacity > (double) total_mem * 0.9) {
>>>>>>> branch-0.4.0
std::cerr << "Warning: cpu_cache_capacity value is too aggressive" << std::endl;
}
uint64_t insert_buffer_size = (uint64_t) GetConfig(CONFIG_DB).GetInt32Value(CONFIG_DB_INSERT_BUFFER_SIZE, 4);
insert_buffer_size *= GB;
if (insert_buffer_size + cpu_cache_capacity >= total_mem) {
std::cerr << "Error: sum of cpu_cache_capacity and insert_buffer_size exceed system memory" << std::endl;
std::cerr << "ERROR: sum of cpu_cache_capacity and insert_buffer_size exceed system memory" << std::endl;
okay = false;
}
}
@ -294,36 +307,57 @@ ServerConfig::CheckCacheConfig() {
std::string cpu_cache_free_percent_str = cache_config.GetValue(CACHE_FREE_PERCENT, "0.85");
double cpu_cache_free_percent;
if (ValidationUtil::ValidateStringIsDouble(cpu_cache_free_percent_str, cpu_cache_free_percent) != SERVER_SUCCESS) {
std::cerr << "Error: cpu_cache_free_percent " << cpu_cache_free_percent_str << " is not a double" << std::endl;
std::cerr << "ERROR: cpu_cache_free_percent " << cpu_cache_free_percent_str << " is not a double" << std::endl;
okay = false;
<<<<<<< HEAD
} else if (cpu_cache_free_percent < std::numeric_limits<double>::epsilon() || cpu_cache_free_percent > 1.0) {
std::cerr << "Error: invalid cpu_cache_free_percent " << cpu_cache_free_percent_str << std::endl;
=======
}
else if (cpu_cache_free_percent < std::numeric_limits<double>::epsilon() || cpu_cache_free_percent > 1.0) {
std::cerr << "ERROR: invalid cpu_cache_free_percent " << cpu_cache_free_percent_str << std::endl;
>>>>>>> branch-0.4.0
okay = false;
}
std::string insert_cache_immediately_str = cache_config.GetValue(CONFIG_INSERT_CACHE_IMMEDIATELY, "false");
if (ValidationUtil::ValidateStringIsBool(insert_cache_immediately_str) != SERVER_SUCCESS) {
std::cerr << "Error: invalid insert_cache_immediately config: " << insert_cache_immediately_str << std::endl;
std::cerr << "ERROR: invalid insert_cache_immediately config: " << insert_cache_immediately_str << std::endl;
okay = false;
}
std::string gpu_cache_capacity_str = cache_config.GetValue(CONFIG_GPU_CACHE_CAPACITY, "5");
if (ValidationUtil::ValidateStringIsNumber(gpu_cache_capacity_str) != SERVER_SUCCESS) {
std::cerr << "Error: gpu_cache_capacity " << gpu_cache_capacity_str << " is not a number" << std::endl;
std::cerr << "ERROR: gpu_cache_capacity " << gpu_cache_capacity_str << " is not a number" << std::endl;
okay = false;
<<<<<<< HEAD
} else {
=======
}
else {
>>>>>>> branch-0.4.0
uint64_t gpu_cache_capacity = (uint64_t) std::stol(gpu_cache_capacity_str);
gpu_cache_capacity *= GB;
int gpu_index = GetConfig(CONFIG_SERVER).GetInt32Value(CONFIG_GPU_INDEX, 0);
size_t gpu_memory;
if (ValidationUtil::GetGpuMemory(gpu_index, gpu_memory) != SERVER_SUCCESS) {
std::cerr << "Error: could not get gpu memory for device " << gpu_index << std::endl;
std::cerr << "ERROR: could not get gpu memory for device " << gpu_index << std::endl;
okay = false;
<<<<<<< HEAD
} else if (gpu_cache_capacity >= gpu_memory) {
std::cerr << "Error: gpu_cache_capacity " << gpu_cache_capacity
<< " exceed total gpu memory " << gpu_memory << std::endl;
okay = false;
} else if (gpu_cache_capacity > (double) gpu_memory * 0.9) {
=======
}
else if (gpu_cache_capacity >= gpu_memory) {
std::cerr << "ERROR: gpu_cache_capacity " << gpu_cache_capacity
<< " exceed total gpu memory " << gpu_memory << std::endl;
okay = false;
}
else if (gpu_cache_capacity > (double) gpu_memory * 0.9) {
>>>>>>> branch-0.4.0
std::cerr << "Warning: gpu_cache_capacity value is too aggressive" << std::endl;
}
}
@ -331,10 +365,16 @@ ServerConfig::CheckCacheConfig() {
std::string gpu_cache_free_percent_str = cache_config.GetValue(GPU_CACHE_FREE_PERCENT, "0.85");
double gpu_cache_free_percent;
if (ValidationUtil::ValidateStringIsDouble(gpu_cache_free_percent_str, gpu_cache_free_percent) != SERVER_SUCCESS) {
std::cerr << "Error: gpu_cache_free_percent " << gpu_cache_free_percent_str << " is not a double" << std::endl;
std::cerr << "ERROR: gpu_cache_free_percent " << gpu_cache_free_percent_str << " is not a double" << std::endl;
okay = false;
<<<<<<< HEAD
} else if (gpu_cache_free_percent < std::numeric_limits<double>::epsilon() || gpu_cache_free_percent > 1.0) {
std::cerr << "Error: invalid gpu_cache_free_percent " << gpu_cache_free_percent << std::endl;
=======
}
else if (gpu_cache_free_percent < std::numeric_limits<double>::epsilon() || gpu_cache_free_percent > 1.0) {
std::cerr << "ERROR: invalid gpu_cache_free_percent " << gpu_cache_free_percent << std::endl;
>>>>>>> branch-0.4.0
okay = false;
}
@ -342,10 +382,16 @@ ServerConfig::CheckCacheConfig() {
for (std::string &gpu_id : conf_gpu_ids) {
if (ValidationUtil::ValidateStringIsNumber(gpu_id) != SERVER_SUCCESS) {
std::cerr << "Error: gpu_id " << gpu_id << " is not a number" << std::endl;
std::cerr << "ERROR: gpu_id " << gpu_id << " is not a number" << std::endl;
okay = false;
<<<<<<< HEAD
} else if (ValidationUtil::ValidateGpuIndex(std::stol(gpu_id)) != SERVER_SUCCESS) {
std::cerr << "Error: gpu_id " << gpu_id << " is valid" << std::endl;
=======
}
else if (ValidationUtil::ValidateGpuIndex(std::stol(gpu_id)) != SERVER_SUCCESS) {
std::cerr << "ERROR: gpu_id " << gpu_id << " is invalid" << std::endl;
>>>>>>> branch-0.4.0
okay = false;
}
}
@ -365,19 +411,23 @@ ServerConfig::CheckEngineConfig() {
std::string use_blas_threshold_str = engine_config.GetValue(CONFIG_DCBT, "20");
if (ValidationUtil::ValidateStringIsNumber(use_blas_threshold_str) != SERVER_SUCCESS) {
std::cerr << "Error: use_blas_threshold " << use_blas_threshold_str << " is not a number" << std::endl;
std::cerr << "ERROR: use_blas_threshold " << use_blas_threshold_str << " is not a number" << std::endl;
okay = false;
}
std::string omp_thread_num_str = engine_config.GetValue(CONFIG_OMP_THREAD_NUM, "0");
if (ValidationUtil::ValidateStringIsNumber(omp_thread_num_str) != SERVER_SUCCESS) {
std::cerr << "Error: omp_thread_num " << omp_thread_num_str << " is not a number" << std::endl;
std::cerr << "ERROR: omp_thread_num " << omp_thread_num_str << " is not a number" << std::endl;
okay = false;
} else {
int32_t omp_thread = std::stol(omp_thread_num_str);
uint32_t sys_thread_cnt = 8;
if (omp_thread > CommonUtil::GetSystemAvailableThreads(sys_thread_cnt)) {
<<<<<<< HEAD
std::cerr << "Error: omp_thread_num " << omp_thread_num_str << " > system available thread "
=======
std::cerr << "ERROR: omp_thread_num " << omp_thread_num_str << " > system available thread "
>>>>>>> branch-0.4.0
<< sys_thread_cnt << std::endl;
okay = false;
}
@ -435,7 +485,7 @@ ServerConfig::CheckResourceConfig() {
bool okay = true;
server::ConfigNode resource_config = GetConfig(CONFIG_RESOURCE);
if (resource_config.GetChildren().empty()) {
std::cerr << "Error: no context under resource" << std::endl;
std::cerr << "ERROR: no context under resource" << std::endl;
okay = false;
}
@ -457,9 +507,9 @@ ServerConfig::CheckResourceConfig() {
auto type = resource_conf.GetValue(CONFIG_RESOURCE_TYPE);
std::string device_id_str = resource_conf.GetValue(CONFIG_RESOURCE_DEVICE_ID, "0");
int32_t device_id;
int32_t device_id = -1;
if (ValidationUtil::ValidateStringIsNumber(device_id_str) != SERVER_SUCCESS) {
std::cerr << "Error: device_id " << device_id_str << " is not a number" << std::endl;
std::cerr << "ERROR: device_id " << device_id_str << " is not a number" << std::endl;
okay = false;
} else {
device_id = std::stol(device_id_str);
@ -467,7 +517,7 @@ ServerConfig::CheckResourceConfig() {
std::string enable_executor_str = resource_conf.GetValue(CONFIG_RESOURCE_ENABLE_EXECUTOR, "off");
if (ValidationUtil::ValidateStringIsBool(enable_executor_str) != SERVER_SUCCESS) {
std::cerr << "Error: invalid enable_executor config: " << enable_executor_str << std::endl;
std::cerr << "ERROR: invalid enable_executor config: " << enable_executor_str << std::endl;
okay = false;
}
@ -478,7 +528,12 @@ ServerConfig::CheckResourceConfig() {
if (resource_conf.GetBoolValue(CONFIG_RESOURCE_ENABLE_EXECUTOR, false)) {
hasExecutor = true;
}
<<<<<<< HEAD
} else if (type == "GPU") {
=======
}
else if (type == "GPU") {
>>>>>>> branch-0.4.0
int build_index_gpu_index = GetConfig(CONFIG_SERVER).GetInt32Value(CONFIG_GPU_INDEX, 0);
if (device_id == build_index_gpu_index) {
resource_valid_flag = true;
@ -488,18 +543,36 @@ ServerConfig::CheckResourceConfig() {
}
std::string gpu_resource_num_str = resource_conf.GetValue(CONFIG_RESOURCE_NUM, "2");
if (ValidationUtil::ValidateStringIsNumber(gpu_resource_num_str) != SERVER_SUCCESS) {
std::cerr << "Error: gpu_resource_num " << gpu_resource_num_str << " is not a number" << std::endl;
std::cerr << "ERROR: gpu_resource_num " << gpu_resource_num_str << " is not a number" << std::endl;
okay = false;
}
bool mem_valid = true;
std::string pinned_memory_str = resource_conf.GetValue(CONFIG_RESOURCE_PIN_MEMORY, "300");
if (ValidationUtil::ValidateStringIsNumber(pinned_memory_str) != SERVER_SUCCESS) {
std::cerr << "Error: pinned_memory " << pinned_memory_str << " is not a number" << std::endl;
std::cerr << "ERROR: pinned_memory " << pinned_memory_str << " is not a number" << std::endl;
okay = false;
mem_valid = false;
}
std::string temp_memory_str = resource_conf.GetValue(CONFIG_RESOURCE_TEMP_MEMORY, "300");
if (ValidationUtil::ValidateStringIsNumber(temp_memory_str) != SERVER_SUCCESS) {
std::cerr << "Error: temp_memory " << temp_memory_str << " is not a number" << std::endl;
std::cerr << "ERROR: temp_memory " << temp_memory_str << " is not a number" << std::endl;
okay = false;
mem_valid = false;
}
if (mem_valid) {
size_t gpu_memory;
if (ValidationUtil::GetGpuMemory(device_id, gpu_memory) != SERVER_SUCCESS) {
std::cerr << "ERROR: could not get gpu memory for device " << device_id << std::endl;
okay = false;
}
else {
size_t prealoc_mem = std::stol(pinned_memory_str) + std::stol(temp_memory_str);
if (prealoc_mem >= gpu_memory) {
std::cerr << "ERROR: sum of pinned_memory and temp_memory " << prealoc_mem
<< " exceeds total gpu memory " << gpu_memory << " for device " << device_id << std::endl;
okay = false;
}
}
}
}
}
@ -523,7 +596,7 @@ ServerConfig::CheckResourceConfig() {
std::string speed_str = connection_conf.GetValue(CONFIG_SPEED_CONNECTIONS);
if (ValidationUtil::ValidateStringIsNumber(speed_str) != SERVER_SUCCESS) {
std::cerr << "Error: speed " << speed_str << " is not a number" << std::endl;
std::cerr << "ERROR: speed " << speed_str << " is not a number" << std::endl;
okay = false;
}
@ -531,17 +604,17 @@ ServerConfig::CheckResourceConfig() {
std::string delimiter = "===";
auto delimiter_pos = endpoint_str.find(delimiter);
if (delimiter_pos == std::string::npos) {
std::cerr << "Error: invalid endpoint format: " << endpoint_str << std::endl;
std::cerr << "ERROR: invalid endpoint format: " << endpoint_str << std::endl;
okay = false;
} else {
std::string left_resource = endpoint_str.substr(0, delimiter_pos);
if (resource_list.find(left_resource) == resource_list.end()) {
std::cerr << "Error: left resource " << left_resource << " does not exist" << std::endl;
std::cerr << "ERROR: left resource " << left_resource << " does not exist" << std::endl;
okay = false;
}
std::string right_resource = endpoint_str.substr(delimiter_pos + delimiter.length(), endpoint_str.length());
if (resource_list.find(right_resource) == resource_list.end()) {
std::cerr << "Error: right resource " << right_resource << " does not exist" << std::endl;
std::cerr << "ERROR: right resource " << right_resource << " does not exist" << std::endl;
okay = false;
}
}

2
cpp/src/server/ServerConfig.h
View File

@ -53,9 +53,7 @@ static const char* CONFIG_OMP_THREAD_NUM = "omp_thread_num";
static const char* CONFIG_RESOURCE = "resource_config";
static const char* CONFIG_RESOURCES = "resources";
static const char* CONFIG_RESOURCE_TYPE = "type";
static const char* CONFIG_RESOURCE_MEMORY = "memory";
static const char* CONFIG_RESOURCE_DEVICE_ID = "device_id";
static const char* CONFIG_RESOURCE_ENABLE_LOADER = "enable_loader";
static const char* CONFIG_RESOURCE_ENABLE_EXECUTOR = "enable_executor";
static const char* CONFIG_RESOURCE_NUM = "gpu_resource_num";
static const char* CONFIG_RESOURCE_PIN_MEMORY = "pinned_memory";

15
cpp/src/server/grpc_impl/GrpcRequestScheduler.cpp
View File

@ -16,7 +16,7 @@ namespace grpc {
using namespace ::milvus;
namespace {
const std::map<ErrorCode, ::milvus::grpc::ErrorCode> &ErrorMap() {
::milvus::grpc::ErrorCode ErrorMap(ErrorCode code) {
static const std::map<ErrorCode, ::milvus::grpc::ErrorCode> code_map = {
{SERVER_UNEXPECTED_ERROR, ::milvus::grpc::ErrorCode::UNEXPECTED_ERROR},
{SERVER_UNSUPPORTED_ERROR, ::milvus::grpc::ErrorCode::UNEXPECTED_ERROR},
@ -40,8 +40,9 @@ namespace {
{SERVER_INVALID_ROWRECORD_ARRAY, ::milvus::grpc::ErrorCode::ILLEGAL_ROWRECORD},
{SERVER_INVALID_TOPK, ::milvus::grpc::ErrorCode::ILLEGAL_TOPK},
{SERVER_INVALID_NPROBE, ::milvus::grpc::ErrorCode::ILLEGAL_ARGUMENT},
{SERVER_INVALID_INDEX_NLIST, ::milvus::grpc::ErrorCode::ILLEGAL_ARGUMENT},
{SERVER_INVALID_INDEX_METRIC_TYPE,::milvus::grpc::ErrorCode::ILLEGAL_ARGUMENT},
{SERVER_INVALID_INDEX_NLIST, ::milvus::grpc::ErrorCode::ILLEGAL_NLIST},
{SERVER_INVALID_INDEX_METRIC_TYPE,::milvus::grpc::ErrorCode::ILLEGAL_METRIC_TYPE},
{SERVER_INVALID_INDEX_FILE_SIZE, ::milvus::grpc::ErrorCode::ILLEGAL_ARGUMENT},
{SERVER_ILLEGAL_VECTOR_ID, ::milvus::grpc::ErrorCode::ILLEGAL_VECTOR_ID},
{SERVER_ILLEGAL_SEARCH_RESULT, ::milvus::grpc::ErrorCode::ILLEGAL_SEARCH_RESULT},
{SERVER_CACHE_ERROR, ::milvus::grpc::ErrorCode::CACHE_FAILED},
@ -49,7 +50,11 @@ namespace {
{SERVER_BUILD_INDEX_ERROR, ::milvus::grpc::ErrorCode::BUILD_INDEX_ERROR},
};
return code_map;
if(code_map.find(code) != code_map.end()) {
return code_map.at(code);
} else {
return ::milvus::grpc::ErrorCode::UNEXPECTED_ERROR;
}
}
}
@ -115,7 +120,7 @@ void GrpcRequestScheduler::ExecTask(BaseTaskPtr &task_ptr, ::milvus::grpc::Statu
ErrorCode err = task_ptr->ErrorID();
if (err != SERVER_SUCCESS) {
grpc_status->set_reason(task_ptr->ErrorMsg());
grpc_status->set_error_code(ErrorMap().at(err));
grpc_status->set_error_code(ErrorMap(err));
}
}
}

12
cpp/src/server/grpc_impl/GrpcRequestTask.cpp
View File

@ -463,12 +463,12 @@ InsertTask::OnExecute() {
bool user_provide_ids = !insert_param_->row_id_array().empty();
//user already provided id before, all insert action require user id
if((table_info.flag_ & engine::meta::FLAG_MASK_HAS_USERID) && !user_provide_ids) {
return SetError(SERVER_INVALID_ARGUMENT, "Table vector ids are user defined, please provide id for this batch");
return SetError(SERVER_ILLEGAL_VECTOR_ID, "Table vector ids are user defined, please provide id for this batch");
}
//user didn't provided id before, no need to provide user id
if((table_info.flag_ & engine::meta::FLAG_MASK_NO_USERID) && user_provide_ids) {
return SetError(SERVER_INVALID_ARGUMENT, "Table vector ids are auto generated, no need to provide id for this batch");
return SetError(SERVER_ILLEGAL_VECTOR_ID, "Table vector ids are auto generated, no need to provide id for this batch");
}
rc.RecordSection("check validation");
@ -485,12 +485,12 @@ InsertTask::OnExecute() {
// TODO: change to one dimension array in protobuf or use multiple-thread to copy the data
for (size_t i = 0; i < insert_param_->row_record_array_size(); i++) {
if (insert_param_->row_record_array(i).vector_data().empty()) {
return SetError(SERVER_INVALID_ROWRECORD_ARRAY, "Row record float array is empty");
return SetError(SERVER_INVALID_ROWRECORD_ARRAY, "Row record array data is empty");
}
uint64_t vec_dim = insert_param_->row_record_array(i).vector_data().size();
if (vec_dim != table_info.dimension_) {
ErrorCode error_code = SERVER_INVALID_VECTOR_DIMENSION;
std::string error_msg = "Invalid rowrecord dimension: " + std::to_string(vec_dim)
std::string error_msg = "Invalid row record dimension: " + std::to_string(vec_dim)
+ " vs. table dimension:" +
std::to_string(table_info.dimension_);
return SetError(error_code, error_msg);
@ -638,12 +638,12 @@ SearchTask::OnExecute() {
std::vector<float> vec_f(record_array_size * table_info.dimension_, 0);
for (size_t i = 0; i < record_array_size; i++) {
if (search_param_->query_record_array(i).vector_data().empty()) {
return SetError(SERVER_INVALID_ROWRECORD_ARRAY, "Query record float array is empty");
return SetError(SERVER_INVALID_ROWRECORD_ARRAY, "Row record array data is empty");
}
uint64_t query_vec_dim = search_param_->query_record_array(i).vector_data().size();
if (query_vec_dim != table_info.dimension_) {
ErrorCode error_code = SERVER_INVALID_VECTOR_DIMENSION;
std::string error_msg = "Invalid rowrecord dimension: " + std::to_string(query_vec_dim)
std::string error_msg = "Invalid row record dimension: " + std::to_string(query_vec_dim)
+ " vs. table dimension:" + std::to_string(table_info.dimension_);
return SetError(error_code, error_msg);
}

40
cpp/src/utils/ValidationUtil.cpp
View File

@ -9,6 +9,7 @@
#include <regex>
#include <algorithm>
namespace zilliz {
namespace milvus {
namespace server {
@ -56,15 +57,16 @@ ValidationUtil::ValidateTableDimension(int64_t dimension) {
if (dimension <= 0 || dimension > TABLE_DIMENSION_LIMIT) {
SERVER_LOG_ERROR << "Table dimension excceed the limitation: " << TABLE_DIMENSION_LIMIT;
return SERVER_INVALID_VECTOR_DIMENSION;
} else {
}
else {
return SERVER_SUCCESS;
}
}
ErrorCode
ValidationUtil::ValidateTableIndexType(int32_t index_type) {
int engine_type = (int)engine::EngineType(index_type);
if(engine_type <= 0 || engine_type > (int)engine::EngineType::MAX_VALUE) {
int engine_type = (int) engine::EngineType(index_type);
if (engine_type <= 0 || engine_type > (int) engine::EngineType::MAX_VALUE) {
return SERVER_INVALID_INDEX_TYPE;
}
@ -73,7 +75,7 @@ ValidationUtil::ValidateTableIndexType(int32_t index_type) {
ErrorCode
ValidationUtil::ValidateTableIndexNlist(int32_t nlist) {
if(nlist <= 0) {
if (nlist <= 0) {
return SERVER_INVALID_INDEX_NLIST;
}
@ -82,7 +84,7 @@ ValidationUtil::ValidateTableIndexNlist(int32_t nlist) {
ErrorCode
ValidationUtil::ValidateTableIndexFileSize(int64_t index_file_size) {
if(index_file_size <= 0 || index_file_size > INDEX_FILE_SIZE_LIMIT) {
if (index_file_size <= 0 || index_file_size > INDEX_FILE_SIZE_LIMIT) {
return SERVER_INVALID_INDEX_FILE_SIZE;
}
@ -91,14 +93,14 @@ ValidationUtil::ValidateTableIndexFileSize(int64_t index_file_size) {
ErrorCode
ValidationUtil::ValidateTableIndexMetricType(int32_t metric_type) {
if(metric_type != (int32_t)engine::MetricType::L2 && metric_type != (int32_t)engine::MetricType::IP) {
if (metric_type != (int32_t) engine::MetricType::L2 && metric_type != (int32_t) engine::MetricType::IP) {
return SERVER_INVALID_INDEX_METRIC_TYPE;
}
return SERVER_SUCCESS;
}
ErrorCode
ValidationUtil::ValidateSearchTopk(int64_t top_k, const engine::meta::TableSchema& table_schema) {
ValidationUtil::ValidateSearchTopk(int64_t top_k, const engine::meta::TableSchema &table_schema) {
if (top_k <= 0 || top_k > 2048) {
return SERVER_INVALID_TOPK;
}
@ -107,7 +109,7 @@ ValidationUtil::ValidateSearchTopk(int64_t top_k, const engine::meta::TableSchem
}
ErrorCode
ValidationUtil::ValidateSearchNprobe(int64_t nprobe, const engine::meta::TableSchema& table_schema) {
ValidationUtil::ValidateSearchNprobe(int64_t nprobe, const engine::meta::TableSchema &table_schema) {
if (nprobe <= 0 || nprobe > table_schema.nlist_) {
return SERVER_INVALID_NPROBE;
}
@ -124,7 +126,7 @@ ValidationUtil::ValidateGpuIndex(uint32_t gpu_index) {
return SERVER_UNEXPECTED_ERROR;
}
if(gpu_index >= num_devices) {
if (gpu_index >= num_devices) {
return SERVER_INVALID_ARGUMENT;
}
@ -132,7 +134,7 @@ ValidationUtil::ValidateGpuIndex(uint32_t gpu_index) {
}
ErrorCode
ValidationUtil::GetGpuMemory(uint32_t gpu_index, size_t& memory) {
ValidationUtil::GetGpuMemory(uint32_t gpu_index, size_t &memory) {
cudaDeviceProp deviceProp;
auto cuda_err = cudaGetDeviceProperties(&deviceProp, gpu_index);
if (cuda_err) {
@ -145,20 +147,17 @@ ValidationUtil::GetGpuMemory(uint32_t gpu_index, size_t& memory) {
}
ErrorCode
ValidationUtil::ValidateIpAddress(const std::string &ip_address) {
ValidationUtil::ValidateIpAddress(const std::string &ip_address) {
struct in_addr address;
int result = inet_pton(AF_INET, ip_address.c_str(), &address);
switch(result) {
case 1:
return SERVER_SUCCESS;
case 0:
SERVER_LOG_ERROR << "Invalid IP address: " << ip_address;
switch (result) {
case 1:return SERVER_SUCCESS;
case 0:SERVER_LOG_ERROR << "Invalid IP address: " << ip_address;
return SERVER_INVALID_ARGUMENT;
default:
SERVER_LOG_ERROR << "inet_pton conversion error";
default:SERVER_LOG_ERROR << "inet_pton conversion error";
return SERVER_UNEXPECTED_ERROR;
}
}
@ -188,7 +187,7 @@ ValidationUtil::ValidateStringIsBool(std::string &str) {
ErrorCode
ValidationUtil::ValidateStringIsDouble(const std::string &str, double &val) {
char* end = nullptr;
char *end = nullptr;
val = std::strtod(str.c_str(), &end);
if (end != str.c_str() && *end == '\0' && val != HUGE_VAL) {
return SERVER_SUCCESS;
@ -240,7 +239,8 @@ ValidationUtil::ValidateDbURI(const std::string &uri) {
okay = false;
}
}
} else {
}
else {
SERVER_LOG_ERROR << "Wrong URI format: URI = " << uri;
okay = false;
}

4
cpp/unittest/db/db_tests.cpp
View File

@ -378,11 +378,11 @@ TEST_F(DBTest, INDEX_TEST) {
ASSERT_TRUE(stat.ok());
engine::TableIndex index_out;
stat = db_->DescribeIndex(table_info.table_id_, index);
stat = db_->DescribeIndex(table_info.table_id_, index_out);
ASSERT_TRUE(stat.ok());
ASSERT_EQ(index.engine_type_, index_out.engine_type_);
ASSERT_EQ(index.nlist_, index_out.nlist_);
ASSERT_EQ(index.metric_type_, index_out.metric_type_);
ASSERT_EQ(table_info.metric_type_, index_out.metric_type_);
stat = db_->DropIndex(table_info.table_id_);
ASSERT_TRUE(stat.ok());

36
cpp/unittest/db/engine_test.cpp
View File

@ -5,16 +5,16 @@
////////////////////////////////////////////////////////////////////////////////
#include <gtest/gtest.h>
#include <boost/filesystem.hpp>
#include <vector>
#include "db/engine/EngineFactory.h"
#include "db/engine/ExecutionEngineImpl.h"
#include "server/ServerConfig.h"
#include <vector>
#include "utils.h"
using namespace zilliz::milvus;
TEST(EngineTest, FACTORY_TEST) {
TEST_F(EngineTest, FACTORY_TEST) {
{
auto engine_ptr = engine::EngineFactory::Build(
512,
@ -76,7 +76,7 @@ TEST(EngineTest, FACTORY_TEST) {
}
}
TEST(EngineTest, ENGINE_IMPL_TEST) {
TEST_F(EngineTest, ENGINE_IMPL_TEST) {
uint16_t dimension = 64;
std::string file_path = "/tmp/milvus_index_1";
auto engine_ptr = engine::EngineFactory::Build(
@ -105,19 +105,19 @@ TEST(EngineTest, ENGINE_IMPL_TEST) {
ASSERT_EQ(engine_ptr->Dimension(), dimension);
ASSERT_EQ(engine_ptr->Count(), ids.size());
server::ConfigNode& config = server::ServerConfig::GetInstance().GetConfig(server::CONFIG_CACHE);
config.AddSequenceItem(server::CONFIG_GPU_IDS, "0");
status = engine_ptr->CopyToGpu(0);
//ASSERT_TRUE(status.ok());
auto new_engine = engine_ptr->Clone();
ASSERT_EQ(new_engine->Dimension(), dimension);
ASSERT_EQ(new_engine->Count(), ids.size());
status = new_engine->CopyToCpu();
//ASSERT_TRUE(status.ok());
auto engine_build = new_engine->BuildIndex("/tmp/milvus_index_2", engine::EngineType::FAISS_IVFSQ8);
//ASSERT_TRUE(status.ok());
// server::ConfigNode& config = server::ServerConfig::GetInstance().GetConfig(server::CONFIG_CACHE);
// config.AddSequenceItem(server::CONFIG_GPU_IDS, "0");
//
// status = engine_ptr->CopyToGpu(0);
// //ASSERT_TRUE(status.ok());
//
// auto new_engine = engine_ptr->Clone();
// ASSERT_EQ(new_engine->Dimension(), dimension);
// ASSERT_EQ(new_engine->Count(), ids.size());
// status = new_engine->CopyToCpu();
// //ASSERT_TRUE(status.ok());
//
// auto engine_build = new_engine->BuildIndex("/tmp/milvus_index_2", engine::EngineType::FAISS_IVFSQ8);
// //ASSERT_TRUE(status.ok());
}

16
cpp/unittest/db/meta_tests.cpp
View File

@ -101,8 +101,7 @@ TEST_F(MetaTest, TABLE_FILE_TEST) {
meta::TableFilesSchema files;
status = impl_->GetTableFiles(table_file.table_id_, ids, files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(files.size(), 1UL);
ASSERT_TRUE(files[0].file_type_ == meta::TableFileSchema::TO_DELETE);
ASSERT_EQ(files.size(), 0UL);
}
TEST_F(MetaTest, ARCHIVE_TEST_DAYS) {
@ -150,8 +149,6 @@ TEST_F(MetaTest, ARCHIVE_TEST_DAYS) {
for(auto& file : files_get) {
if (days[i] < days_num) {
ASSERT_EQ(file.file_type_, meta::TableFileSchema::NEW);
} else {
ASSERT_EQ(file.file_type_, meta::TableFileSchema::TO_DELETE);
}
i++;
}
@ -195,9 +192,7 @@ TEST_F(MetaTest, ARCHIVE_TEST_DISK) {
ASSERT_TRUE(status.ok());
for(auto& file : files_get) {
if (i < 5) {
ASSERT_TRUE(file.file_type_ == meta::TableFileSchema::TO_DELETE);
} else {
if (i >= 5) {
ASSERT_EQ(file.file_type_, meta::TableFileSchema::NEW);
}
++i;
@ -277,38 +272,31 @@ TEST_F(MetaTest, TABLE_FILES_TEST) {
meta::TableFilesSchema files;
status = impl_->FilesToIndex(files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(files.size(), to_index_files_cnt);
meta::DatePartionedTableFilesSchema dated_files;
status = impl_->FilesToMerge(table.table_id_, dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(), raw_files_cnt);
status = impl_->FilesToIndex(files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(files.size(), to_index_files_cnt);
meta::DatesT dates = {table_file.date_};
std::vector<size_t> ids;
status = impl_->FilesToSearch(table_id, ids, dates, dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(),
to_index_files_cnt+raw_files_cnt+index_files_cnt);
status = impl_->FilesToSearch(table_id, ids, meta::DatesT(), dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(),
to_index_files_cnt+raw_files_cnt+index_files_cnt);
status = impl_->FilesToSearch(table_id, ids, meta::DatesT(), dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(),
to_index_files_cnt+raw_files_cnt+index_files_cnt);
ids.push_back(size_t(9999999999));
status = impl_->FilesToSearch(table_id, ids, dates, dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(),0);
std::vector<int> file_types;

17
cpp/unittest/db/mysql_meta_test.cpp
View File

@ -108,9 +108,7 @@ TEST_F(MySqlMetaTest, TABLE_FILE_TEST) {
std::vector<size_t> ids = {table_file.id_};
meta::TableFilesSchema files;
status = impl_->GetTableFiles(table_file.table_id_, ids, files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(files.size(), 1UL);
ASSERT_TRUE(files[0].file_type_ == meta::TableFileSchema::TO_DELETE);
ASSERT_EQ(files.size(), 0UL);
}
TEST_F(MySqlMetaTest, ARCHIVE_TEST_DAYS) {
@ -159,8 +157,6 @@ TEST_F(MySqlMetaTest, ARCHIVE_TEST_DAYS) {
for(auto& file : files_get) {
if (days[i] < days_num) {
ASSERT_EQ(file.file_type_, meta::TableFileSchema::NEW);
} else {
ASSERT_EQ(file.file_type_, meta::TableFileSchema::TO_DELETE);
}
i++;
}
@ -219,9 +215,7 @@ TEST_F(MySqlMetaTest, ARCHIVE_TEST_DISK) {
ASSERT_TRUE(status.ok());
for(auto& file : files_get) {
if (i < 5) {
ASSERT_TRUE(file.file_type_ == meta::TableFileSchema::TO_DELETE);
} else {
if (i >= 5) {
ASSERT_EQ(file.file_type_, meta::TableFileSchema::NEW);
}
++i;
@ -302,38 +296,31 @@ TEST_F(MySqlMetaTest, TABLE_FILES_TEST) {
meta::TableFilesSchema files;
status = impl_->FilesToIndex(files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(files.size(), to_index_files_cnt);
meta::DatePartionedTableFilesSchema dated_files;
status = impl_->FilesToMerge(table.table_id_, dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(), raw_files_cnt);
status = impl_->FilesToIndex(files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(files.size(), to_index_files_cnt);
meta::DatesT dates = {table_file.date_};
std::vector<size_t> ids;
status = impl_->FilesToSearch(table_id, ids, dates, dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(),
to_index_files_cnt+raw_files_cnt+index_files_cnt);
status = impl_->FilesToSearch(table_id, ids, meta::DatesT(), dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(),
to_index_files_cnt+raw_files_cnt+index_files_cnt);
status = impl_->FilesToSearch(table_id, ids, meta::DatesT(), dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(),
to_index_files_cnt+raw_files_cnt+index_files_cnt);
ids.push_back(size_t(9999999999));
status = impl_->FilesToSearch(table_id, ids, dates, dated_files);
ASSERT_TRUE(status.ok());
ASSERT_EQ(dated_files[table_file.date_].size(),0);
std::vector<int> file_types;

13
cpp/unittest/db/utils.cpp
View File

@ -13,6 +13,7 @@
#include "db/Factories.h"
#include "db/Options.h"
#include "server/ServerConfig.h"
#include "knowhere/index/vector_index/gpu_ivf.h"
INITIALIZE_EASYLOGGINGPP
@ -46,6 +47,12 @@ void BaseTest::InitLog() {
void BaseTest::SetUp() {
InitLog();
zilliz::knowhere::FaissGpuResourceMgr::GetInstance().InitDevice(0, 1024*1024*200, 1024*1024*300, 2);
}
void BaseTest::TearDown() {
zilliz::knowhere::FaissGpuResourceMgr::GetInstance().Free();
}
engine::Options BaseTest::GetOptions() {
@ -84,6 +91,8 @@ void DBTest::TearDown() {
db_->DropAll();
delete db_;
BaseTest::TearDown();
engine::ResMgrInst::GetInstance()->Stop();
engine::SchedInst::GetInstance()->Stop();
@ -111,6 +120,8 @@ void MetaTest::SetUp() {
void MetaTest::TearDown() {
impl_->DropAll();
BaseTest::TearDown();
auto options = GetOptions();
boost::filesystem::remove_all(options.meta.path);
}
@ -139,6 +150,8 @@ void MySqlMetaTest::SetUp() {
void MySqlMetaTest::TearDown() {
impl_->DropAll();
BaseTest::TearDown();
auto options = GetOptions();
boost::filesystem::remove_all(options.meta.path);
}

5
cpp/unittest/db/utils.h
View File

@ -37,6 +37,7 @@ protected:
void InitLog();
virtual void SetUp() override;
virtual void TearDown() override;
virtual zilliz::milvus::engine::Options GetOptions();
};
@ -55,6 +56,10 @@ class DBTest2 : public DBTest {
virtual zilliz::milvus::engine::Options GetOptions() override;
};
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class EngineTest : public DBTest {
};
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class MetaTest : public BaseTest {
protected: