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

#1686 API search_in_files cannot work correctly when vectors is store… (#1701) 

* #1686 API search_in_files cannot work correctly when vectors is stored in certain non-default partition

Signed-off-by: yhmo <yihua.mo@zilliz.com>

* #1686 API search_in_files cannot work correctly when vectors is stored in certain non-default partition

Signed-off-by: groot <yihua.mo@zilliz.com>
pull/1712/head
groot 2020-03-20 12:14:07 +08:00 committed by GitHub
parent 5950deddfc
commit fe2595fa38
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
13 changed files with 281 additions and 100 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

1
CHANGELOG.md
View File

@ -11,6 +11,7 @@ Please mark all change in change log and use the issue from GitHub
- \#1648 The cache cannot be used all when the vector type is binary - \#1648 The cache cannot be used all when the vector type is binary
- \#1651 Check validity of dimension when collection metric type is binary one - \#1651 Check validity of dimension when collection metric type is binary one
- \#1663 PQ index parameter 'm' validation - \#1663 PQ index parameter 'm' validation
- \#1686 API search_in_files cannot work correctly when vectors is stored in certain non-default partition
## Feature ## Feature
- \#1603 BinaryFlat add 2 Metric: Substructure and Superstructure - \#1603 BinaryFlat add 2 Metric: Substructure and Superstructure

9
core/src/db/DB.h
View File

@ -29,7 +29,9 @@ class Env;
class DB { class DB {
public: public:
DB() = default; DB() = default;
DB(const DB&) = delete; DB(const DB&) = delete;
DB& DB&
operator=(const DB&) = delete; operator=(const DB&) = delete;
@ -37,6 +39,7 @@ class DB {
virtual Status virtual Status
Start() = 0; Start() = 0;
virtual Status virtual Status
Stop() = 0; Stop() = 0;
@ -121,9 +124,9 @@ class DB {
const VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances) = 0; const VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances) = 0;
virtual Status virtual Status
QueryByFileID(const std::shared_ptr<server::Context>& context, const std::string& table_id, QueryByFileID(const std::shared_ptr<server::Context>& context, const std::vector<std::string>& file_ids, uint64_t k,
const std::vector<std::string>& file_ids, uint64_t k, const milvus::json& extra_params, const milvus::json& extra_params, const VectorsData& vectors, ResultIds& result_ids,
const VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances) = 0; ResultDistances& result_distances) = 0;
virtual Status virtual Status
Size(uint64_t& result) = 0; Size(uint64_t& result) = 0;

43
core/src/db/DBImpl.cpp
View File

@ -339,9 +339,8 @@ DBImpl::PreloadTable(const std::string& table_id) {
} }
// step 1: get all table files from parent table // step 1: get all table files from parent table
std::vector<size_t> ids;
meta::TableFilesSchema files_array; meta::TableFilesSchema files_array;
auto status = GetFilesToSearch(table_id, ids, files_array); auto status = GetFilesToSearch(table_id, files_array);
if (!status.ok()) { if (!status.ok()) {
return status; return status;
} }
@ -350,7 +349,7 @@ DBImpl::PreloadTable(const std::string& table_id) {
std::vector<meta::TableSchema> partition_array; std::vector<meta::TableSchema> partition_array;
status = meta_ptr_->ShowPartitions(table_id, partition_array); status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& schema : partition_array) { for (auto& schema : partition_array) {
status = GetFilesToSearch(schema.table_id_, ids, files_array); status = GetFilesToSearch(schema.table_id_, files_array);
} }
int64_t size = 0; int64_t size = 0;
@ -1109,13 +1108,12 @@ DBImpl::Query(const std::shared_ptr<server::Context>& context, const std::string
} }
Status status; Status status;
std::vector<size_t> ids;
meta::TableFilesSchema files_array; meta::TableFilesSchema files_array;
if (partition_tags.empty()) { if (partition_tags.empty()) {
// no partition tag specified, means search in whole table // no partition tag specified, means search in whole table
// get all table files from parent table // get all table files from parent table
status = GetFilesToSearch(table_id, ids, files_array); status = GetFilesToSearch(table_id, files_array);
if (!status.ok()) { if (!status.ok()) {
return status; return status;
} }
@ -1123,7 +1121,7 @@ DBImpl::Query(const std::shared_ptr<server::Context>& context, const std::string
std::vector<meta::TableSchema> partition_array; std::vector<meta::TableSchema> partition_array;
status = meta_ptr_->ShowPartitions(table_id, partition_array); status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& schema : partition_array) { for (auto& schema : partition_array) {
status = GetFilesToSearch(schema.table_id_, ids, files_array); status = GetFilesToSearch(schema.table_id_, files_array);
} }
if (files_array.empty()) { if (files_array.empty()) {
@ -1135,7 +1133,7 @@ DBImpl::Query(const std::shared_ptr<server::Context>& context, const std::string
GetPartitionsByTags(table_id, partition_tags, partition_name_array); GetPartitionsByTags(table_id, partition_tags, partition_name_array);
for (auto& partition_name : partition_name_array) { for (auto& partition_name : partition_name_array) {
status = GetFilesToSearch(partition_name, ids, files_array); status = GetFilesToSearch(partition_name, files_array);
} }
if (files_array.empty()) { if (files_array.empty()) {
@ -1144,7 +1142,7 @@ DBImpl::Query(const std::shared_ptr<server::Context>& context, const std::string
} }
cache::CpuCacheMgr::GetInstance()->PrintInfo(); // print cache info before query cache::CpuCacheMgr::GetInstance()->PrintInfo(); // print cache info before query
status = QueryAsync(query_ctx, table_id, files_array, k, extra_params, vectors, result_ids, result_distances); status = QueryAsync(query_ctx, files_array, k, extra_params, vectors, result_ids, result_distances);
cache::CpuCacheMgr::GetInstance()->PrintInfo(); // print cache info after query cache::CpuCacheMgr::GetInstance()->PrintInfo(); // print cache info after query
query_ctx->GetTraceContext()->GetSpan()->Finish(); query_ctx->GetTraceContext()->GetSpan()->Finish();
@ -1153,9 +1151,9 @@ DBImpl::Query(const std::shared_ptr<server::Context>& context, const std::string
} }
Status Status
DBImpl::QueryByFileID(const std::shared_ptr<server::Context>& context, const std::string& table_id, DBImpl::QueryByFileID(const std::shared_ptr<server::Context>& context, const std::vector<std::string>& file_ids,
const std::vector<std::string>& file_ids, uint64_t k, const milvus::json& extra_params, uint64_t k, const milvus::json& extra_params, const VectorsData& vectors, ResultIds& result_ids,
const VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances) { ResultDistances& result_distances) {
auto query_ctx = context->Child("Query by file id"); auto query_ctx = context->Child("Query by file id");
if (!initialized_.load(std::memory_order_acquire)) { if (!initialized_.load(std::memory_order_acquire)) {
@ -1165,25 +1163,23 @@ DBImpl::QueryByFileID(const std::shared_ptr<server::Context>& context, const std
// get specified files // get specified files
std::vector<size_t> ids; std::vector<size_t> ids;
for (auto& id : file_ids) { for (auto& id : file_ids) {
meta::TableFileSchema table_file;
table_file.table_id_ = table_id;
std::string::size_type sz; std::string::size_type sz;
ids.push_back(std::stoul(id, &sz)); ids.push_back(std::stoul(id, &sz));
} }
meta::TableFilesSchema files_array; meta::TableFilesSchema search_files;
auto status = GetFilesToSearch(table_id, ids, files_array); auto status = meta_ptr_->FilesByID(ids, search_files);
if (!status.ok()) { if (!status.ok()) {
return status; return status;
} }
fiu_do_on("DBImpl.QueryByFileID.empty_files_array", files_array.clear()); fiu_do_on("DBImpl.QueryByFileID.empty_files_array", search_files.clear());
if (files_array.empty()) { if (search_files.empty()) {
return Status(DB_ERROR, "Invalid file id"); return Status(DB_ERROR, "Invalid file id");
} }
cache::CpuCacheMgr::GetInstance()->PrintInfo(); // print cache info before query cache::CpuCacheMgr::GetInstance()->PrintInfo(); // print cache info before query
status = QueryAsync(query_ctx, table_id, files_array, k, extra_params, vectors, result_ids, result_distances); status = QueryAsync(query_ctx, search_files, k, extra_params, vectors, result_ids, result_distances);
cache::CpuCacheMgr::GetInstance()->PrintInfo(); // print cache info after query cache::CpuCacheMgr::GetInstance()->PrintInfo(); // print cache info after query
query_ctx->GetTraceContext()->GetSpan()->Finish(); query_ctx->GetTraceContext()->GetSpan()->Finish();
@ -1204,9 +1200,9 @@ DBImpl::Size(uint64_t& result) {
// internal methods // internal methods
/////////////////////////////////////////////////////////////////////////////////////////////////////////////////// ///////////////////////////////////////////////////////////////////////////////////////////////////////////////////
Status Status
DBImpl::QueryAsync(const std::shared_ptr<server::Context>& context, const std::string& table_id, DBImpl::QueryAsync(const std::shared_ptr<server::Context>& context, const meta::TableFilesSchema& files, uint64_t k,
const meta::TableFilesSchema& files, uint64_t k, const milvus::json& extra_params, const milvus::json& extra_params, const VectorsData& vectors, ResultIds& result_ids,
const VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances) { ResultDistances& result_distances) {
auto query_async_ctx = context->Child("Query Async"); auto query_async_ctx = context->Child("Query Async");
server::CollectQueryMetrics metrics(vectors.vector_count_); server::CollectQueryMetrics metrics(vectors.vector_count_);
@ -1610,12 +1606,11 @@ DBImpl::GetFilesToBuildIndex(const std::string& table_id, const std::vector<int>
} }
Status Status
DBImpl::GetFilesToSearch(const std::string& table_id, const std::vector<size_t>& file_ids, DBImpl::GetFilesToSearch(const std::string& table_id, meta::TableFilesSchema& files) {
meta::TableFilesSchema& files) {
ENGINE_LOG_DEBUG << "Collect files from table: " << table_id; ENGINE_LOG_DEBUG << "Collect files from table: " << table_id;
meta::TableFilesSchema search_files; meta::TableFilesSchema search_files;
auto status = meta_ptr_->FilesToSearch(table_id, file_ids, search_files); auto status = meta_ptr_->FilesToSearch(table_id, search_files);
if (!status.ok()) { if (!status.ok()) {
return status; return status;
} }

22
core/src/db/DBImpl.h
View File

@ -42,12 +42,15 @@ class Meta;
class DBImpl : public DB, public server::CacheConfigHandler, public server::EngineConfigHandler { class DBImpl : public DB, public server::CacheConfigHandler, public server::EngineConfigHandler {
public: public:
explicit DBImpl(const DBOptions& options); explicit DBImpl(const DBOptions& options);
~DBImpl(); ~DBImpl();
Status Status
Start() override; Start() override;
Status Status
Stop() override; Stop() override;
Status Status
DropAll() override; DropAll() override;
@ -141,9 +144,9 @@ class DBImpl : public DB, public server::CacheConfigHandler, public server::Engi
const VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances) override; const VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances) override;
Status Status
QueryByFileID(const std::shared_ptr<server::Context>& context, const std::string& table_id, QueryByFileID(const std::shared_ptr<server::Context>& context, const std::vector<std::string>& file_ids, uint64_t k,
const std::vector<std::string>& file_ids, uint64_t k, const milvus::json& extra_params, const milvus::json& extra_params, const VectorsData& vectors, ResultIds& result_ids,
const VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances) override; ResultDistances& result_distances) override;
Status Status
Size(uint64_t& result) override; Size(uint64_t& result) override;
@ -157,9 +160,9 @@ class DBImpl : public DB, public server::CacheConfigHandler, public server::Engi
private: private:
Status Status
QueryAsync(const std::shared_ptr<server::Context>& context, const std::string& table_id, QueryAsync(const std::shared_ptr<server::Context>& context, const meta::TableFilesSchema& files, uint64_t k,
const meta::TableFilesSchema& files, uint64_t k, const milvus::json& extra_params, const milvus::json& extra_params, const VectorsData& vectors, ResultIds& result_ids,
const VectorsData& vectors, ResultIds& result_ids, ResultDistances& result_distances); ResultDistances& result_distances);
Status Status
GetVectorByIdHelper(const std::string& table_id, IDNumber vector_id, VectorsData& vector, GetVectorByIdHelper(const std::string& table_id, IDNumber vector_id, VectorsData& vector,
@ -167,8 +170,10 @@ class DBImpl : public DB, public server::CacheConfigHandler, public server::Engi
void void
BackgroundTimerTask(); BackgroundTimerTask();
void void
WaitMergeFileFinish(); WaitMergeFileFinish();
void void
WaitBuildIndexFinish(); WaitBuildIndexFinish();
@ -180,13 +185,16 @@ class DBImpl : public DB, public server::CacheConfigHandler, public server::Engi
Status Status
MergeFiles(const std::string& table_id, const meta::TableFilesSchema& files); MergeFiles(const std::string& table_id, const meta::TableFilesSchema& files);
Status Status
BackgroundMergeFiles(const std::string& table_id); BackgroundMergeFiles(const std::string& table_id);
void void
BackgroundMerge(std::set<std::string> table_ids); BackgroundMerge(std::set<std::string> table_ids);
void void
StartBuildIndexTask(bool force = false); StartBuildIndexTask(bool force = false);
void void
BackgroundBuildIndex(); BackgroundBuildIndex();
@ -204,7 +212,7 @@ class DBImpl : public DB, public server::CacheConfigHandler, public server::Engi
meta::TableFilesSchema& files); meta::TableFilesSchema& files);
Status Status
GetFilesToSearch(const std::string& table_id, const std::vector<size_t>& file_ids, meta::TableFilesSchema& files); GetFilesToSearch(const std::string& table_id, meta::TableFilesSchema& files);
Status Status
GetPartitionByTag(const std::string& table_id, const std::string& partition_tag, std::string& partition_name); GetPartitionByTag(const std::string& table_id, const std::string& partition_tag, std::string& partition_name);

7
core/src/db/meta/Meta.h
View File

@ -116,7 +116,7 @@ class Meta {
GetPartitionName(const std::string& table_name, const std::string& tag, std::string& partition_name) = 0; GetPartitionName(const std::string& table_name, const std::string& tag, std::string& partition_name) = 0;
virtual Status virtual Status
FilesToSearch(const std::string& table_id, const std::vector<size_t>& ids, TableFilesSchema& files) = 0; FilesToSearch(const std::string& table_id, TableFilesSchema& files) = 0;
virtual Status virtual Status
FilesToMerge(const std::string& table_id, TableFilesSchema& files) = 0; FilesToMerge(const std::string& table_id, TableFilesSchema& files) = 0;
@ -125,7 +125,10 @@ class Meta {
FilesToIndex(TableFilesSchema&) = 0; FilesToIndex(TableFilesSchema&) = 0;
virtual Status virtual Status
FilesByType(const std::string& table_id, const std::vector<int>& file_types, TableFilesSchema& table_files) = 0; FilesByType(const std::string& table_id, const std::vector<int>& file_types, TableFilesSchema& files) = 0;
virtual Status
FilesByID(const std::vector<size_t>& ids, TableFilesSchema& files) = 0;
virtual Status virtual Status
Size(uint64_t& result) = 0; Size(uint64_t& result) = 0;

117
core/src/db/meta/MySQLMetaImpl.cpp
View File

@ -1544,7 +1544,7 @@ MySQLMetaImpl::GetPartitionName(const std::string& table_id, const std::string&
} }
Status Status
MySQLMetaImpl::FilesToSearch(const std::string& table_id, const std::vector<size_t>& ids, TableFilesSchema& files) { MySQLMetaImpl::FilesToSearch(const std::string& table_id, TableFilesSchema& files) {
files.clear(); files.clear();
try { try {
@ -1565,16 +1565,6 @@ MySQLMetaImpl::FilesToSearch(const std::string& table_id, const std::vector<size
<< "SELECT id, table_id, segment_id, engine_type, file_id, file_type, file_size, row_count, date" << "SELECT id, table_id, segment_id, engine_type, file_id, file_type, file_size, row_count, date"
<< " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote << table_id; << " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote << table_id;
if (!ids.empty()) {
std::stringstream idSS;
for (auto& id : ids) {
idSS << "id = " << std::to_string(id) << " OR ";
}
std::string idStr = idSS.str();
idStr = idStr.substr(0, idStr.size() - 4); // remove the last " OR "
filesToSearchQuery << " AND (" << idStr << ")";
}
// End // End
filesToSearchQuery << " AND" filesToSearchQuery << " AND"
<< " (file_type = " << std::to_string(TableFileSchema::RAW) << " (file_type = " << std::to_string(TableFileSchema::RAW)
@ -1782,8 +1772,7 @@ MySQLMetaImpl::FilesToIndex(TableFilesSchema& files) {
} }
Status Status
MySQLMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>& file_types, MySQLMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>& file_types, TableFilesSchema& files) {
TableFilesSchema& table_files) {
if (file_types.empty()) { if (file_types.empty()) {
return Status(DB_ERROR, "file types array is empty"); return Status(DB_ERROR, "file types array is empty");
} }
@ -1791,7 +1780,7 @@ MySQLMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>&
Status ret = Status::OK(); Status ret = Status::OK();
try { try {
table_files.clear(); files.clear();
mysqlpp::StoreQueryResult res; mysqlpp::StoreQueryResult res;
{ {
@ -1857,7 +1846,7 @@ MySQLMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>&
ret = status; ret = status;
} }
table_files.emplace_back(file_schema); files.emplace_back(file_schema);
int32_t file_type = resRow["file_type"]; int32_t file_type = resRow["file_type"];
switch (file_type) { switch (file_type) {
@ -1924,6 +1913,104 @@ MySQLMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>&
return ret; return ret;
} }
Status
MySQLMetaImpl::FilesByID(const std::vector<size_t>& ids, TableFilesSchema& files) {
files.clear();
if (ids.empty()) {
return Status::OK();
}
try {
server::MetricCollector metric;
mysqlpp::StoreQueryResult res;
{
mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
bool is_null_connection = (connectionPtr == nullptr);
fiu_do_on("MySQLMetaImpl.FilesByID.null_connection", is_null_connection = true);
fiu_do_on("MySQLMetaImpl.FilesByID.throw_exception", throw std::exception(););
if (is_null_connection) {
return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
}
mysqlpp::Query filesToSearchQuery = connectionPtr->query();
filesToSearchQuery
<< "SELECT id, table_id, segment_id, engine_type, file_id, file_type, file_size, row_count, date"
<< " FROM " << META_TABLEFILES;
std::stringstream idSS;
for (auto& id : ids) {
idSS << "id = " << std::to_string(id) << " OR ";
}
std::string idStr = idSS.str();
idStr = idStr.substr(0, idStr.size() - 4); // remove the last " OR "
filesToSearchQuery << " WHERE (" << idStr << ")";
// End
filesToSearchQuery << " AND"
<< " (file_type = " << std::to_string(TableFileSchema::RAW)
<< " OR file_type = " << std::to_string(TableFileSchema::TO_INDEX)
<< " OR file_type = " << std::to_string(TableFileSchema::INDEX) << ");";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::FilesToSearch: " << filesToSearchQuery.str();
res = filesToSearchQuery.store();
} // Scoped Connection
std::map<std::string, meta::TableSchema> tables;
Status ret;
for (auto& resRow : res) {
TableFileSchema table_file;
table_file.id_ = resRow["id"]; // implicit conversion
resRow["table_id"].to_string(table_file.table_id_);
resRow["segment_id"].to_string(table_file.segment_id_);
table_file.engine_type_ = resRow["engine_type"];
resRow["file_id"].to_string(table_file.file_id_);
table_file.file_type_ = resRow["file_type"];
table_file.file_size_ = resRow["file_size"];
table_file.row_count_ = resRow["row_count"];
table_file.date_ = resRow["date"];
if (tables.find(table_file.table_id_) == tables.end()) {
TableSchema table_schema;
table_schema.table_id_ = table_file.table_id_;
auto status = DescribeTable(table_schema);
if (!status.ok()) {
return status;
}
tables.insert(std::make_pair(table_file.table_id_, table_schema));
}
auto status = utils::GetTableFilePath(options_, table_file);
if (!status.ok()) {
ret = status;
}
files.emplace_back(table_file);
}
for (auto& table_file : files) {
TableSchema& table_schema = tables[table_file.table_id_];
table_file.dimension_ = table_schema.dimension_;
table_file.index_file_size_ = table_schema.index_file_size_;
table_file.index_params_ = table_schema.index_params_;
table_file.metric_type_ = table_schema.metric_type_;
}
if (files.empty()) {
ENGINE_LOG_ERROR << "No file to search in file id list";
} else {
ENGINE_LOG_DEBUG << "Collect " << files.size() << " files by id";
}
return ret;
} catch (std::exception& e) {
return HandleException("GENERAL ERROR WHEN FINDING TABLE FILES BY ID", e.what());
}
}
// TODO(myh): Support swap to cloud storage // TODO(myh): Support swap to cloud storage
Status Status
MySQLMetaImpl::Archive() { MySQLMetaImpl::Archive() {

8
core/src/db/meta/MySQLMetaImpl.h
View File

@ -105,7 +105,7 @@ class MySQLMetaImpl : public Meta {
GetPartitionName(const std::string& table_id, const std::string& tag, std::string& partition_name) override; GetPartitionName(const std::string& table_id, const std::string& tag, std::string& partition_name) override;
Status Status
FilesToSearch(const std::string& table_id, const std::vector<size_t>& ids, TableFilesSchema& files) override; FilesToSearch(const std::string& table_id, TableFilesSchema& files) override;
Status Status
FilesToMerge(const std::string& table_id, TableFilesSchema& files) override; FilesToMerge(const std::string& table_id, TableFilesSchema& files) override;
@ -114,8 +114,10 @@ class MySQLMetaImpl : public Meta {
FilesToIndex(TableFilesSchema&) override; FilesToIndex(TableFilesSchema&) override;
Status Status
FilesByType(const std::string& table_id, const std::vector<int>& file_types, FilesByType(const std::string& table_id, const std::vector<int>& file_types, TableFilesSchema& files) override;
TableFilesSchema& table_files) override;
Status
FilesByID(const std::vector<size_t>& ids, TableFilesSchema& table_files) override;
Status Status
Archive() override; Archive() override;

102
core/src/db/meta/SqliteMetaImpl.cpp
View File

@ -968,11 +968,11 @@ SqliteMetaImpl::GetPartitionName(const std::string& table_id, const std::string&
} }
Status Status
SqliteMetaImpl::FilesToSearch(const std::string& table_id, const std::vector<size_t>& ids, TableFilesSchema& files) { SqliteMetaImpl::FilesToSearch(const std::string& table_id, TableFilesSchema& files) {
files.clear(); files.clear();
server::MetricCollector metric;
try { try {
server::MetricCollector metric;
fiu_do_on("SqliteMetaImpl.FilesToSearch.throw_exception", throw std::exception()); fiu_do_on("SqliteMetaImpl.FilesToSearch.throw_exception", throw std::exception());
auto select_columns = auto select_columns =
@ -995,14 +995,8 @@ SqliteMetaImpl::FilesToSearch(const std::string& table_id, const std::vector<siz
// perform query // perform query
decltype(ConnectorPtr->select(select_columns)) selected; decltype(ConnectorPtr->select(select_columns)) selected;
if (ids.empty()) { auto filter = where(match_tableid and match_type);
auto filter = where(match_tableid and match_type); selected = ConnectorPtr->select(select_columns, filter);
selected = ConnectorPtr->select(select_columns, filter);
} else {
auto match_fileid = in(&TableFileSchema::id_, ids);
auto filter = where(match_tableid and match_fileid and match_type);
selected = ConnectorPtr->select(select_columns, filter);
}
Status ret; Status ret;
for (auto& file : selected) { for (auto& file : selected) {
@ -1172,8 +1166,7 @@ SqliteMetaImpl::FilesToIndex(TableFilesSchema& files) {
} }
Status Status
SqliteMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>& file_types, SqliteMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>& file_types, TableFilesSchema& files) {
TableFilesSchema& table_files) {
if (file_types.empty()) { if (file_types.empty()) {
return Status(DB_ERROR, "file types array is empty"); return Status(DB_ERROR, "file types array is empty");
} }
@ -1190,7 +1183,7 @@ SqliteMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>&
try { try {
fiu_do_on("SqliteMetaImpl.FilesByType.throw_exception", throw std::exception()); fiu_do_on("SqliteMetaImpl.FilesByType.throw_exception", throw std::exception());
table_files.clear(); files.clear();
auto selected = ConnectorPtr->select( auto selected = ConnectorPtr->select(
columns(&TableFileSchema::id_, &TableFileSchema::segment_id_, &TableFileSchema::file_id_, columns(&TableFileSchema::id_, &TableFileSchema::segment_id_, &TableFileSchema::file_id_,
&TableFileSchema::file_type_, &TableFileSchema::file_size_, &TableFileSchema::row_count_, &TableFileSchema::file_type_, &TableFileSchema::file_size_, &TableFileSchema::row_count_,
@ -1241,7 +1234,7 @@ SqliteMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>&
ret = status; ret = status;
} }
table_files.emplace_back(file_schema); files.emplace_back(file_schema);
} }
std::string msg = "Get table files by type."; std::string msg = "Get table files by type.";
@ -1277,6 +1270,87 @@ SqliteMetaImpl::FilesByType(const std::string& table_id, const std::vector<int>&
return ret; return ret;
} }
Status
SqliteMetaImpl::FilesByID(const std::vector<size_t>& ids, TableFilesSchema& files) {
files.clear();
if (ids.empty()) {
return Status::OK();
}
try {
server::MetricCollector metric;
fiu_do_on("SqliteMetaImpl.FilesByID.throw_exception", throw std::exception());
auto select_columns =
columns(&TableFileSchema::id_, &TableFileSchema::table_id_, &TableFileSchema::segment_id_,
&TableFileSchema::file_id_, &TableFileSchema::file_type_, &TableFileSchema::file_size_,
&TableFileSchema::row_count_, &TableFileSchema::date_, &TableFileSchema::engine_type_);
std::vector<int> file_types = {(int)TableFileSchema::RAW, (int)TableFileSchema::TO_INDEX,
(int)TableFileSchema::INDEX};
auto match_type = in(&TableFileSchema::file_type_, file_types);
// perform query
decltype(ConnectorPtr->select(select_columns)) selected;
auto match_fileid = in(&TableFileSchema::id_, ids);
auto filter = where(match_fileid and match_type);
selected = ConnectorPtr->select(select_columns, filter);
std::map<std::string, meta::TableSchema> tables;
Status ret;
for (auto& file : selected) {
TableFileSchema table_file;
table_file.id_ = std::get<0>(file);
table_file.table_id_ = std::get<1>(file);
table_file.segment_id_ = std::get<2>(file);
table_file.file_id_ = std::get<3>(file);
table_file.file_type_ = std::get<4>(file);
table_file.file_size_ = std::get<5>(file);
table_file.row_count_ = std::get<6>(file);
table_file.date_ = std::get<7>(file);
table_file.engine_type_ = std::get<8>(file);
if (tables.find(table_file.table_id_) == tables.end()) {
TableSchema table_schema;
table_schema.table_id_ = table_file.table_id_;
auto status = DescribeTable(table_schema);
if (!status.ok()) {
return status;
}
tables.insert(std::make_pair(table_file.table_id_, table_schema));
}
auto status = utils::GetTableFilePath(options_, table_file);
if (!status.ok()) {
ret = status;
}
files.emplace_back(table_file);
}
for (auto& table_file : files) {
TableSchema& table_schema = tables[table_file.table_id_];
table_file.dimension_ = table_schema.dimension_;
table_file.index_file_size_ = table_schema.index_file_size_;
table_file.index_params_ = table_schema.index_params_;
table_file.metric_type_ = table_schema.metric_type_;
}
if (files.empty()) {
ENGINE_LOG_ERROR << "No file to search in file id list";
} else {
ENGINE_LOG_DEBUG << "Collect " << selected.size() << " files by id";
}
return ret;
} catch (std::exception& e) {
return HandleException("Encounter exception when iterate index files", e.what());
}
return Status::OK();
}
// TODO(myh): Support swap to cloud storage // TODO(myh): Support swap to cloud storage
Status Status
SqliteMetaImpl::Archive() { SqliteMetaImpl::Archive() {

8
core/src/db/meta/SqliteMetaImpl.h
View File

@ -104,7 +104,7 @@ class SqliteMetaImpl : public Meta {
GetPartitionName(const std::string& table_id, const std::string& tag, std::string& partition_name) override; GetPartitionName(const std::string& table_id, const std::string& tag, std::string& partition_name) override;
Status Status
FilesToSearch(const std::string& table_id, const std::vector<size_t>& ids, TableFilesSchema& files) override; FilesToSearch(const std::string& table_id, TableFilesSchema& files) override;
Status Status
FilesToMerge(const std::string& table_id, TableFilesSchema& files) override; FilesToMerge(const std::string& table_id, TableFilesSchema& files) override;
@ -113,8 +113,10 @@ class SqliteMetaImpl : public Meta {
FilesToIndex(TableFilesSchema&) override; FilesToIndex(TableFilesSchema&) override;
Status Status
FilesByType(const std::string& table_id, const std::vector<int>& file_types, FilesByType(const std::string& table_id, const std::vector<int>& file_types, TableFilesSchema& files) override;
TableFilesSchema& table_files) override;
Status
FilesByID(const std::vector<size_t>& ids, TableFilesSchema& files) override;
Status Status
Size(uint64_t& result) override; Size(uint64_t& result) override;

2
core/src/server/delivery/request/SearchRequest.cpp
View File

@ -154,7 +154,7 @@ SearchRequest::OnExecute() {
status = DBWrapper::DB()->Query(context_, table_name_, partition_list_, (size_t)topk_, extra_params_, status = DBWrapper::DB()->Query(context_, table_name_, partition_list_, (size_t)topk_, extra_params_,
vectors_data_, result_ids, result_distances); vectors_data_, result_ids, result_distances);
} else { } else {
status = DBWrapper::DB()->QueryByFileID(context_, table_name_, file_id_list_, (size_t)topk_, extra_params_, status = DBWrapper::DB()->QueryByFileID(context_, file_id_list_, (size_t)topk_, extra_params_,
vectors_data_, result_ids, result_distances); vectors_data_, result_ids, result_distances);
} }

21
core/unittest/db/test_db.cpp
View File

@ -371,24 +371,18 @@ TEST_F(DBTest, SEARCH_TEST) {
} }
milvus::engine::ResultIds result_ids; milvus::engine::ResultIds result_ids;
milvus::engine::ResultDistances result_distances; milvus::engine::ResultDistances result_distances;
stat = db_->QueryByFileID(dummy_context_, TABLE_NAME, file_ids, k, stat = db_->QueryByFileID(dummy_context_, file_ids, k,
json_params, json_params,
xq, xq,
result_ids, result_ids,
result_distances); result_distances);
ASSERT_TRUE(stat.ok()); ASSERT_TRUE(stat.ok());
FIU_ENABLE_FIU("SqliteMetaImpl.FilesToSearch.throw_exception"); // FIU_ENABLE_FIU("DBImpl.QueryByFileID.empty_files_array");
stat = // stat =
db_->QueryByFileID(dummy_context_, TABLE_NAME, file_ids, k, json_params, xq, result_ids, result_distances); // db_->QueryByFileID(dummy_context_, file_ids, k, json_params, xq, result_ids, result_distances);
ASSERT_FALSE(stat.ok()); // ASSERT_FALSE(stat.ok());
fiu_disable("SqliteMetaImpl.FilesToSearch.throw_exception"); // fiu_disable("DBImpl.QueryByFileID.empty_files_array");
FIU_ENABLE_FIU("DBImpl.QueryByFileID.empty_files_array");
stat =
db_->QueryByFileID(dummy_context_, TABLE_NAME, file_ids, k, json_params, xq, result_ids, result_distances);
ASSERT_FALSE(stat.ok());
fiu_disable("DBImpl.QueryByFileID.empty_files_array");
} }
// TODO(zhiru): PQ build takes forever // TODO(zhiru): PQ build takes forever
@ -434,7 +428,7 @@ TEST_F(DBTest, SEARCH_TEST) {
} }
result_ids.clear(); result_ids.clear();
result_dists.clear(); result_dists.clear();
stat = db_->QueryByFileID(dummy_context_, TABLE_NAME, file_ids, k, json_params, xq, result_ids, result_dists); stat = db_->QueryByFileID(dummy_context_, file_ids, k, json_params, xq, result_ids, result_dists);
ASSERT_TRUE(stat.ok()); ASSERT_TRUE(stat.ok());
} }
#endif #endif
@ -579,7 +573,6 @@ TEST_F(DBTest, SHUTDOWN_TEST) {
ASSERT_FALSE(stat.ok()); ASSERT_FALSE(stat.ok());
std::vector<std::string> file_ids; std::vector<std::string> file_ids;
stat = db_->QueryByFileID(dummy_context_, stat = db_->QueryByFileID(dummy_context_,
table_info.table_id_,
file_ids, file_ids,
1, 1,
json_params, json_params,

21
core/unittest/db/test_meta.cpp
View File

@ -260,13 +260,12 @@ TEST_F(MetaTest, FALID_TEST) {
fiu_disable("SqliteMetaImpl.GetPartitionName.throw_exception"); fiu_disable("SqliteMetaImpl.GetPartitionName.throw_exception");
} }
{ {
std::vector<size_t> ids;
milvus::engine::meta::TableFilesSchema table_files; milvus::engine::meta::TableFilesSchema table_files;
status = impl_->FilesToSearch("notexist", ids, table_files); status = impl_->FilesToSearch("notexist", table_files);
ASSERT_EQ(status.code(), milvus::DB_NOT_FOUND); ASSERT_EQ(status.code(), milvus::DB_NOT_FOUND);
FIU_ENABLE_FIU("SqliteMetaImpl.FilesToSearch.throw_exception"); FIU_ENABLE_FIU("SqliteMetaImpl.FilesToSearch.throw_exception");
status = impl_->FilesToSearch(table_id, ids, table_files); status = impl_->FilesToSearch(table_id, table_files);
ASSERT_EQ(status.code(), milvus::DB_META_TRANSACTION_FAILED); ASSERT_EQ(status.code(), milvus::DB_META_TRANSACTION_FAILED);
fiu_disable("SqliteMetaImpl.FilesToSearch.throw_exception"); fiu_disable("SqliteMetaImpl.FilesToSearch.throw_exception");
} }
@ -622,13 +621,21 @@ TEST_F(MetaTest, TABLE_FILES_TEST) {
ASSERT_EQ(files.size(), to_index_files_cnt); ASSERT_EQ(files.size(), to_index_files_cnt);
table_files.clear(); table_files.clear();
std::vector<size_t> ids; status = impl_->FilesToSearch(table_id, table_files);
status = impl_->FilesToSearch(table_id, ids, table_files);
ASSERT_EQ(table_files.size(), to_index_files_cnt + raw_files_cnt + index_files_cnt); ASSERT_EQ(table_files.size(), to_index_files_cnt + raw_files_cnt + index_files_cnt);
std::vector<size_t> ids;
for (auto& file : table_files) {
ids.push_back(file.id_);
}
size_t cnt = table_files.size();
table_files.clear(); table_files.clear();
ids.push_back(size_t(9999999999)); status = impl_->FilesByID(ids, table_files);
status = impl_->FilesToSearch(table_id, ids, table_files); ASSERT_EQ(table_files.size(), cnt);
table_files.clear();
ids = {9999999999UL};
status = impl_->FilesByID(ids, table_files);
ASSERT_EQ(table_files.size(), 0); ASSERT_EQ(table_files.size(), 0);
table_files.clear(); table_files.clear();

20
core/unittest/db/test_meta_mysql.cpp
View File

@ -249,6 +249,13 @@ TEST_F(MySqlMetaTest, TABLE_FILE_TEST) {
status = impl_->GetTableFiles(table_file.table_id_, ids, files); status = impl_->GetTableFiles(table_file.table_id_, ids, files);
ASSERT_TRUE(status.ok()); ASSERT_TRUE(status.ok());
table_file.table_id_ = table.table_id_;
table_file.file_type_ = milvus::engine::meta::TableFileSchema::RAW;
status = impl_->CreateTableFile(table_file);
ids = {table_file.id_};
status = impl_->FilesByID(ids, files);
ASSERT_EQ(files.size(), 1UL);
table_file.table_id_ = table.table_id_; table_file.table_id_ = table.table_id_;
table_file.file_type_ = milvus::engine::meta::TableFileSchema::TO_DELETE; table_file.file_type_ = milvus::engine::meta::TableFileSchema::TO_DELETE;
status = impl_->CreateTableFile(table_file); status = impl_->CreateTableFile(table_file);
@ -641,26 +648,25 @@ TEST_F(MySqlMetaTest, TABLE_FILES_TEST) {
fiu_disable("MySQLMetaImpl.FilesToIndex.throw_exception"); fiu_disable("MySQLMetaImpl.FilesToIndex.throw_exception");
table_files.clear(); table_files.clear();
std::vector<size_t> ids; status = impl_->FilesToSearch(table_id, table_files);
status = impl_->FilesToSearch(table_id, ids, table_files);
ASSERT_EQ(table_files.size(), to_index_files_cnt + raw_files_cnt + index_files_cnt); ASSERT_EQ(table_files.size(), to_index_files_cnt + raw_files_cnt + index_files_cnt);
table_files.clear(); table_files.clear();
ids.push_back(size_t(9999999999)); std::vector<size_t> ids = {9999999999UL};
status = impl_->FilesToSearch(table_id, ids, table_files); status = impl_->FilesByID(ids, table_files);
ASSERT_EQ(table_files.size(), 0); ASSERT_EQ(table_files.size(), 0);
FIU_ENABLE_FIU("MySQLMetaImpl.FilesToSearch.null_connection"); FIU_ENABLE_FIU("MySQLMetaImpl.FilesToSearch.null_connection");
status = impl_->FilesToSearch(table_id, ids, table_files); status = impl_->FilesToSearch(table_id, table_files);
ASSERT_FALSE(status.ok()); ASSERT_FALSE(status.ok());
fiu_disable("MySQLMetaImpl.FilesToSearch.null_connection"); fiu_disable("MySQLMetaImpl.FilesToSearch.null_connection");
FIU_ENABLE_FIU("MySQLMetaImpl.FilesToSearch.throw_exception"); FIU_ENABLE_FIU("MySQLMetaImpl.FilesToSearch.throw_exception");
status = impl_->FilesToSearch(table_id, ids, table_files); status = impl_->FilesToSearch(table_id, table_files);
ASSERT_FALSE(status.ok()); ASSERT_FALSE(status.ok());
fiu_disable("MySQLMetaImpl.FilesToSearch.throw_exception"); fiu_disable("MySQLMetaImpl.FilesToSearch.throw_exception");
status = impl_->FilesToSearch("notexist", ids, table_files); status = impl_->FilesToSearch("notexist", table_files);
ASSERT_EQ(status.code(), milvus::DB_NOT_FOUND); ASSERT_EQ(status.code(), milvus::DB_NOT_FOUND);
table_files.clear(); table_files.clear();