admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-29 06:55:27 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge branch '0.6.0' into spell_error

Browse Source
This commit is contained in:
Jin Hai 2019-12-03 14:35:04 +08:00 committed by GitHub
commit be357f5154
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
70 changed files with 1452 additions and 709 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
CHANGELOG.md
View File

@ -36,12 +36,17 @@ Please mark all change in change log and use the ticket from JIRA.
- \#543 - client raise exception in shards when search results is empty
- \#545 - Avoid dead circle of build index thread when error occurs
- \#547 - NSG build failed using GPU-edition if set gpu_enable false
- \#548 - NSG search accuracy is too low
- \#552 - Server down during building index_type: IVF_PQ using GPU-edition
- \#561 - Milvus server should report exception/error message or terminate on mysql metadata backend error
- \#579 - Build index hang in GPU version when gpu_resources disabled
- \#596 - Frequently insert operation cost too much disk space
- \#599 - Build index log is incorrect
- \#602 - Optimizer specify wrong gpu_id
- \#606 - No log generated during building index with CPU
- \#631 - FAISS isn't compiled with O3 option
- \#649 - Typo "partiton" should be "partition"
- \#654 - Random crash when frequently insert vector one by one
## Feature
- \#12 - Pure CPU version for Milvus
@ -55,6 +60,7 @@ Please mark all change in change log and use the ticket from JIRA.
- \#502 - C++ SDK support IVFPQ and SPTAG
- \#560 - Add version in server config file
- \#605 - Print more messages when server start
- \#644 - Add a new rpc command to get milvus build version whether cpu or gpu
## Improvement
- \#255 - Add ivfsq8 test report detailed version
@ -76,6 +82,7 @@ Please mark all change in change log and use the ticket from JIRA.
- \#470 - Small raw files should not be build index
- \#584 - Intergrate internal FAISS
- \#611 - Remove MILVUS_CPU_VERSION
- \#634 - FAISS GPU version is compiled with O0
## Task

5
core/src/cache/Cache.inl vendored
View File

@ -176,6 +176,11 @@ Cache<ItemObj>::print() {
{
std::lock_guard<std::mutex> lock(mutex_);
cache_count = lru_.size();
#if 0
for (auto it = lru_.begin(); it != lru_.end(); ++it) {
SERVER_LOG_DEBUG << it->first;
}
#endif
}
SERVER_LOG_DEBUG << "[Cache item count]: " << cache_count;

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

@ -80,7 +80,7 @@ class DB {
DropPartitionByTag(const std::string& table_id, const std::string& partition_tag) = 0;
virtual Status
ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partiton_schema_array) = 0;
ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partition_schema_array) = 0;
virtual Status
InsertVectors(const std::string& table_id, const std::string& partition_tag, uint64_t n, const float* vectors,

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

@ -52,9 +52,7 @@ constexpr uint64_t METRIC_ACTION_INTERVAL = 1;
constexpr uint64_t COMPACT_ACTION_INTERVAL = 1;
constexpr uint64_t INDEX_ACTION_INTERVAL = 1;
constexpr uint64_t INDEX_FAILED_RETRY_TIME = 1;
static const Status SHUTDOWN_ERROR = Status(DB_ERROR, "Milsvus server is shutdown!");
static const Status SHUTDOWN_ERROR = Status(DB_ERROR, "Milvus server is shutdown!");
void
TraverseFiles(const meta::DatePartionedTableFilesSchema& date_files, meta::TableFilesSchema& files_array) {
@ -192,9 +190,9 @@ DBImpl::PreloadTable(const std::string& table_id) {
}
// step 2: get files from partition tables
std::vector<meta::TableSchema> partiton_array;
status = meta_ptr_->ShowPartitions(table_id, partiton_array);
for (auto& schema : partiton_array) {
std::vector<meta::TableSchema> partition_array;
status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& schema : partition_array) {
status = GetFilesToSearch(schema.table_id_, ids, dates, files_array);
}
@ -298,12 +296,12 @@ DBImpl::DropPartitionByTag(const std::string& table_id, const std::string& parti
}
Status
DBImpl::ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partiton_schema_array) {
DBImpl::ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partition_schema_array) {
if (shutting_down_.load(std::memory_order_acquire)) {
return SHUTDOWN_ERROR;
}
return meta_ptr_->ShowPartitions(table_id, partiton_schema_array);
return meta_ptr_->ShowPartitions(table_id, partition_schema_array);
}
Status
@ -370,7 +368,7 @@ DBImpl::CreateIndex(const std::string& table_id, const TableIndex& index) {
WaitMergeFileFinish();
// step 4: wait and build index
status = CleanFailedIndexFileOfTable(table_id);
status = index_failed_checker_.CleanFailedIndexFileOfTable(table_id);
status = BuildTableIndexRecursively(table_id, index);
return status;
@ -429,9 +427,9 @@ DBImpl::Query(const std::string& table_id, const std::vector<std::string>& parti
return status;
}
std::vector<meta::TableSchema> partiton_array;
status = meta_ptr_->ShowPartitions(table_id, partiton_array);
for (auto& schema : partiton_array) {
std::vector<meta::TableSchema> partition_array;
status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& schema : partition_array) {
status = GetFilesToSearch(schema.table_id_, ids, dates, files_array);
}
} else {
@ -504,7 +502,9 @@ DBImpl::QueryAsync(const std::string& table_id, const meta::TableFilesSchema& fi
TimeRecorder rc("");
// step 1: get files to search
// step 1: construct search job
auto status = ongoing_files_checker_.MarkOngoingFiles(files);
ENGINE_LOG_DEBUG << "Engine query begin, index file count: " << files.size();
scheduler::SearchJobPtr job = std::make_shared<scheduler::SearchJob>(k, nq, nprobe, vectors);
for (auto& file : files) {
@ -512,9 +512,11 @@ DBImpl::QueryAsync(const std::string& table_id, const meta::TableFilesSchema& fi
job->AddIndexFile(file_ptr);
}
// step 2: put search task to scheduler
// step 2: put search job to scheduler and wait result
scheduler::JobMgrInst::GetInstance()->Put(job);
job->WaitResult();
status = ongoing_files_checker_.UnmarkOngoingFiles(files);
if (!job->GetStatus().ok()) {
return job->GetStatus();
}
@ -693,7 +695,6 @@ DBImpl::MergeFiles(const std::string& table_id, const meta::DateT& date, const m
auto file_schema = file;
file_schema.file_type_ = meta::TableFileSchema::TO_DELETE;
updated.push_back(file_schema);
ENGINE_LOG_DEBUG << "Merging file " << file_schema.file_id_;
index_size = index->Size();
if (index_size >= file_schema.index_file_size_) {
@ -703,20 +704,27 @@ DBImpl::MergeFiles(const std::string& table_id, const meta::DateT& date, const m
// step 3: serialize to disk
try {
index->Serialize();
status = index->Serialize();
if (!status.ok()) {
ENGINE_LOG_ERROR << status.message();
}
} catch (std::exception& ex) {
// typical error: out of disk space or permition denied
std::string msg = "Serialize merged index encounter exception: " + std::string(ex.what());
ENGINE_LOG_ERROR << msg;
status = Status(DB_ERROR, msg);
}
if (!status.ok()) {
// if failed to serialize merge file to disk
// typical error: out of disk space, out of memory or permition denied
table_file.file_type_ = meta::TableFileSchema::TO_DELETE;
status = meta_ptr_->UpdateTableFile(table_file);
ENGINE_LOG_DEBUG << "Failed to update file to index, mark file: " << table_file.file_id_ << " to to_delete";
std::cout << "ERROR: failed to persist merged index file: " << table_file.location_
<< ", possible out of disk space" << std::endl;
ENGINE_LOG_ERROR << "Failed to persist merged file: " << table_file.location_
<< ", possible out of disk space or memory";
return Status(DB_ERROR, msg);
return status;
}
// step 4: update table files state
@ -751,13 +759,15 @@ DBImpl::BackgroundMergeFiles(const std::string& table_id) {
}
for (auto& kv : raw_files) {
auto files = kv.second;
meta::TableFilesSchema& files = kv.second;
if (files.size() < options_.merge_trigger_number_) {
ENGINE_LOG_TRACE << "Files number not greater equal than merge trigger number, skip merge action";
continue;
}
status = ongoing_files_checker_.MarkOngoingFiles(files);
MergeFiles(table_id, kv.first, kv.second);
status = ongoing_files_checker_.UnmarkOngoingFiles(files);
if (shutting_down_.load(std::memory_order_acquire)) {
ENGINE_LOG_DEBUG << "Server will shutdown, skip merge action for table: " << table_id;
@ -788,16 +798,12 @@ DBImpl::BackgroundCompaction(std::set<std::string> table_ids) {
meta_ptr_->Archive();
{
uint64_t ttl = 10 * meta::SECOND; // default: file data will be erase from cache after few seconds
meta_ptr_->CleanUpCacheWithTTL(ttl);
}
{
uint64_t ttl = 5 * meta::MINUTE; // default: file will be deleted after few minutes
uint64_t ttl = 10 * meta::SECOND; // default: file will be hard-deleted few seconds after soft-deleted
if (options_.mode_ == DBOptions::MODE::CLUSTER_WRITABLE) {
ttl = meta::DAY;
ttl = meta::HOUR;
}
meta_ptr_->CleanUpFilesWithTTL(ttl);
meta_ptr_->CleanUpFilesWithTTL(ttl, &ongoing_files_checker_);
}
// ENGINE_LOG_TRACE << " Background compaction thread exit";
@ -833,14 +839,15 @@ DBImpl::StartBuildIndexTask(bool force) {
void
DBImpl::BackgroundBuildIndex() {
// ENGINE_LOG_TRACE << "Background build index thread start";
std::unique_lock<std::mutex> lock(build_index_mutex_);
meta::TableFilesSchema to_index_files;
meta_ptr_->FilesToIndex(to_index_files);
Status status = IgnoreFailedIndexFiles(to_index_files);
Status status = index_failed_checker_.IgnoreFailedIndexFiles(to_index_files);
if (!to_index_files.empty()) {
ENGINE_LOG_DEBUG << "Background build index thread begin";
status = ongoing_files_checker_.MarkOngoingFiles(to_index_files);
// step 2: put build index task to scheduler
std::vector<std::pair<scheduler::BuildIndexJobPtr, scheduler::TableFileSchemaPtr>> job2file_map;
for (auto& file : to_index_files) {
@ -851,6 +858,7 @@ DBImpl::BackgroundBuildIndex() {
job2file_map.push_back(std::make_pair(job, file_ptr));
}
// step 3: wait build index finished and mark failed files
for (auto iter = job2file_map.begin(); iter != job2file_map.end(); ++iter) {
scheduler::BuildIndexJobPtr job = iter->first;
meta::TableFileSchema& file_schema = *(iter->second.get());
@ -859,17 +867,17 @@ DBImpl::BackgroundBuildIndex() {
Status status = job->GetStatus();
ENGINE_LOG_ERROR << "Building index job " << job->id() << " failed: " << status.ToString();
MarkFailedIndexFile(file_schema);
index_failed_checker_.MarkFailedIndexFile(file_schema);
} else {
MarkSucceedIndexFile(file_schema);
ENGINE_LOG_DEBUG << "Building index job " << job->id() << " succeed.";
index_failed_checker_.MarkSucceedIndexFile(file_schema);
}
status = ongoing_files_checker_.UnmarkOngoingFile(file_schema);
}
ENGINE_LOG_DEBUG << "Background build index thread finished";
}
// ENGINE_LOG_TRACE << "Background build index thread exit";
}
Status
@ -894,6 +902,8 @@ DBImpl::GetFilesToBuildIndex(const std::string& table_id, const std::vector<int>
Status
DBImpl::GetFilesToSearch(const std::string& table_id, const std::vector<size_t>& file_ids, const meta::DatesT& dates,
meta::TableFilesSchema& files) {
ENGINE_LOG_DEBUG << "Collect files from table: " << table_id;
meta::DatePartionedTableFilesSchema date_files;
auto status = meta_ptr_->FilesToSearch(table_id, file_ids, dates, date_files);
if (!status.ok()) {
@ -907,15 +917,15 @@ DBImpl::GetFilesToSearch(const std::string& table_id, const std::vector<size_t>&
Status
DBImpl::GetPartitionsByTags(const std::string& table_id, const std::vector<std::string>& partition_tags,
std::set<std::string>& partition_name_array) {
std::vector<meta::TableSchema> partiton_array;
auto status = meta_ptr_->ShowPartitions(table_id, partiton_array);
std::vector<meta::TableSchema> partition_array;
auto status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& tag : partition_tags) {
// trim side-blank of tag, only compare valid characters
// for example: " ab cd " is treated as "ab cd"
std::string valid_tag = tag;
server::StringHelpFunctions::TrimStringBlank(valid_tag);
for (auto& schema : partiton_array) {
for (auto& schema : partition_array) {
if (server::StringHelpFunctions::IsRegexMatch(schema.partition_tag_, valid_tag)) {
partition_name_array.insert(schema.table_id_);
}
@ -934,7 +944,7 @@ DBImpl::DropTableRecursively(const std::string& table_id, const meta::DatesT& da
if (dates.empty()) {
status = mem_mgr_->EraseMemVector(table_id); // not allow insert
status = meta_ptr_->DropTable(table_id); // soft delete table
CleanFailedIndexFileOfTable(table_id);
index_failed_checker_.CleanFailedIndexFileOfTable(table_id);
// scheduler will determine when to delete table files
auto nres = scheduler::ResMgrInst::GetInstance()->GetNumOfComputeResource();
@ -945,9 +955,9 @@ DBImpl::DropTableRecursively(const std::string& table_id, const meta::DatesT& da
status = meta_ptr_->DropDataByDate(table_id, dates);
}
std::vector<meta::TableSchema> partiton_array;
status = meta_ptr_->ShowPartitions(table_id, partiton_array);
for (auto& schema : partiton_array) {
std::vector<meta::TableSchema> partition_array;
status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& schema : partition_array) {
status = DropTableRecursively(schema.table_id_, dates);
if (!status.ok()) {
return status;
@ -967,9 +977,9 @@ DBImpl::UpdateTableIndexRecursively(const std::string& table_id, const TableInde
return status;
}
std::vector<meta::TableSchema> partiton_array;
status = meta_ptr_->ShowPartitions(table_id, partiton_array);
for (auto& schema : partiton_array) {
std::vector<meta::TableSchema> partition_array;
status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& schema : partition_array) {
status = UpdateTableIndexRecursively(schema.table_id_, index);
if (!status.ok()) {
return status;
@ -1014,13 +1024,13 @@ DBImpl::BuildTableIndexRecursively(const std::string& table_id, const TableIndex
GetFilesToBuildIndex(table_id, file_types, table_files);
times++;
IgnoreFailedIndexFiles(table_files);
index_failed_checker_.IgnoreFailedIndexFiles(table_files);
}
// build index for partition
std::vector<meta::TableSchema> partiton_array;
status = meta_ptr_->ShowPartitions(table_id, partiton_array);
for (auto& schema : partiton_array) {
std::vector<meta::TableSchema> partition_array;
status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& schema : partition_array) {
status = BuildTableIndexRecursively(schema.table_id_, index);
if (!status.ok()) {
return status;
@ -1029,7 +1039,7 @@ DBImpl::BuildTableIndexRecursively(const std::string& table_id, const TableIndex
// failed to build index for some files, return error
std::vector<std::string> failed_files;
GetFailedIndexFileOfTable(table_id, failed_files);
index_failed_checker_.GetFailedIndexFileOfTable(table_id, failed_files);
if (!failed_files.empty()) {
std::string msg = "Failed to build index for " + std::to_string(failed_files.size()) +
((failed_files.size() == 1) ? " file" : " files");
@ -1043,16 +1053,16 @@ DBImpl::BuildTableIndexRecursively(const std::string& table_id, const TableIndex
Status
DBImpl::DropTableIndexRecursively(const std::string& table_id) {
ENGINE_LOG_DEBUG << "Drop index for table: " << table_id;
CleanFailedIndexFileOfTable(table_id);
index_failed_checker_.CleanFailedIndexFileOfTable(table_id);
auto status = meta_ptr_->DropTableIndex(table_id);
if (!status.ok()) {
return status;
}
// drop partition index
std::vector<meta::TableSchema> partiton_array;
status = meta_ptr_->ShowPartitions(table_id, partiton_array);
for (auto& schema : partiton_array) {
std::vector<meta::TableSchema> partition_array;
status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& schema : partition_array) {
status = DropTableIndexRecursively(schema.table_id_);
if (!status.ok()) {
return status;
@ -1071,9 +1081,9 @@ DBImpl::GetTableRowCountRecursively(const std::string& table_id, uint64_t& row_c
}
// get partition row count
std::vector<meta::TableSchema> partiton_array;
status = meta_ptr_->ShowPartitions(table_id, partiton_array);
for (auto& schema : partiton_array) {
std::vector<meta::TableSchema> partition_array;
status = meta_ptr_->ShowPartitions(table_id, partition_array);
for (auto& schema : partition_array) {
uint64_t partition_row_count = 0;
status = GetTableRowCountRecursively(schema.table_id_, partition_row_count);
if (!status.ok()) {
@ -1086,86 +1096,5 @@ DBImpl::GetTableRowCountRecursively(const std::string& table_id, uint64_t& row_c
return Status::OK();
}
Status
DBImpl::CleanFailedIndexFileOfTable(const std::string& table_id) {
std::lock_guard<std::mutex> lck(index_failed_mutex_);
index_failed_files_.erase(table_id); // rebuild failed index files for this table
return Status::OK();
}
Status
DBImpl::GetFailedIndexFileOfTable(const std::string& table_id, std::vector<std::string>& failed_files) {
failed_files.clear();
std::lock_guard<std::mutex> lck(index_failed_mutex_);
auto iter = index_failed_files_.find(table_id);
if (iter != index_failed_files_.end()) {
FileID2FailedTimes& failed_map = iter->second;
for (auto it_file = failed_map.begin(); it_file != failed_map.end(); ++it_file) {
failed_files.push_back(it_file->first);
}
}
return Status::OK();
}
Status
DBImpl::MarkFailedIndexFile(const meta::TableFileSchema& file) {
std::lock_guard<std::mutex> lck(index_failed_mutex_);
auto iter = index_failed_files_.find(file.table_id_);
if (iter == index_failed_files_.end()) {
FileID2FailedTimes failed_files;
failed_files.insert(std::make_pair(file.file_id_, 1));
index_failed_files_.insert(std::make_pair(file.table_id_, failed_files));
} else {
auto it_failed_files = iter->second.find(file.file_id_);
if (it_failed_files != iter->second.end()) {
it_failed_files->second++;
} else {
iter->second.insert(std::make_pair(file.file_id_, 1));
}
}
return Status::OK();
}
Status
DBImpl::MarkSucceedIndexFile(const meta::TableFileSchema& file) {
std::lock_guard<std::mutex> lck(index_failed_mutex_);
auto iter = index_failed_files_.find(file.table_id_);
if (iter != index_failed_files_.end()) {
iter->second.erase(file.file_id_);
}
return Status::OK();
}
Status
DBImpl::IgnoreFailedIndexFiles(meta::TableFilesSchema& table_files) {
std::lock_guard<std::mutex> lck(index_failed_mutex_);
// there could be some failed files belong to different table.
// some files may has failed for several times, no need to build index for these files.
// thus we can avoid dead circle for build index operation
for (auto it_file = table_files.begin(); it_file != table_files.end();) {
auto it_failed_files = index_failed_files_.find((*it_file).table_id_);
if (it_failed_files != index_failed_files_.end()) {
auto it_failed_file = it_failed_files->second.find((*it_file).file_id_);
if (it_failed_file != it_failed_files->second.end()) {
if (it_failed_file->second >= INDEX_FAILED_RETRY_TIME) {
it_file = table_files.erase(it_file);
continue;
}
}
}
++it_file;
}
return Status::OK();
}
} // namespace engine
} // namespace milvus

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

@ -18,8 +18,10 @@
#pragma once
#include "DB.h"
#include "Types.h"
#include "src/db/insert/MemManager.h"
#include "db/IndexFailedChecker.h"
#include "db/OngoingFileChecker.h"
#include "db/Types.h"
#include "db/insert/MemManager.h"
#include "utils/ThreadPool.h"
#include <atomic>
@ -87,7 +89,7 @@ class DBImpl : public DB {
DropPartitionByTag(const std::string& table_id, const std::string& partition_tag) override;
Status
ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partiton_schema_array) override;
ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partition_schema_array) override;
Status
InsertVectors(const std::string& table_id, const std::string& partition_tag, uint64_t n, const float* vectors,
@ -178,21 +180,6 @@ class DBImpl : public DB {
Status
GetTableRowCountRecursively(const std::string& table_id, uint64_t& row_count);
Status
CleanFailedIndexFileOfTable(const std::string& table_id);
Status
GetFailedIndexFileOfTable(const std::string& table_id, std::vector<std::string>& failed_files);
Status
MarkFailedIndexFile(const meta::TableFileSchema& file);
Status
MarkSucceedIndexFile(const meta::TableFileSchema& file);
Status
IgnoreFailedIndexFiles(meta::TableFilesSchema& table_files);
private:
const DBOptions options_;
@ -214,11 +201,10 @@ class DBImpl : public DB {
std::list<std::future<void>> index_thread_results_;
std::mutex build_index_mutex_;
std::mutex index_failed_mutex_;
using FileID2FailedTimes = std::map<std::string, uint64_t>;
using Table2FailedFiles = std::map<std::string, FileID2FailedTimes>;
Table2FailedFiles index_failed_files_; // file id mapping to failed times
}; // DBImpl
IndexFailedChecker index_failed_checker_;
OngoingFileChecker ongoing_files_checker_;
}; // DBImpl
} // namespace engine
} // namespace milvus

112
core/src/db/IndexFailedChecker.cpp Normal file
View File

@ -0,0 +1,112 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include "db/IndexFailedChecker.h"
#include <utility>
namespace milvus {
namespace engine {
constexpr uint64_t INDEX_FAILED_RETRY_TIME = 1;
Status
IndexFailedChecker::CleanFailedIndexFileOfTable(const std::string& table_id) {
std::lock_guard<std::mutex> lck(mutex_);
index_failed_files_.erase(table_id); // rebuild failed index files for this table
return Status::OK();
}
Status
IndexFailedChecker::GetFailedIndexFileOfTable(const std::string& table_id, std::vector<std::string>& failed_files) {
failed_files.clear();
std::lock_guard<std::mutex> lck(mutex_);
auto iter = index_failed_files_.find(table_id);
if (iter != index_failed_files_.end()) {
File2RefCount& failed_map = iter->second;
for (auto it_file = failed_map.begin(); it_file != failed_map.end(); ++it_file) {
failed_files.push_back(it_file->first);
}
}
return Status::OK();
}
Status
IndexFailedChecker::MarkFailedIndexFile(const meta::TableFileSchema& file) {
std::lock_guard<std::mutex> lck(mutex_);
auto iter = index_failed_files_.find(file.table_id_);
if (iter == index_failed_files_.end()) {
File2RefCount failed_files;
failed_files.insert(std::make_pair(file.file_id_, 1));
index_failed_files_.insert(std::make_pair(file.table_id_, failed_files));
} else {
auto it_failed_files = iter->second.find(file.file_id_);
if (it_failed_files != iter->second.end()) {
it_failed_files->second++;
} else {
iter->second.insert(std::make_pair(file.file_id_, 1));
}
}
return Status::OK();
}
Status
IndexFailedChecker::MarkSucceedIndexFile(const meta::TableFileSchema& file) {
std::lock_guard<std::mutex> lck(mutex_);
auto iter = index_failed_files_.find(file.table_id_);
if (iter != index_failed_files_.end()) {
iter->second.erase(file.file_id_);
if (iter->second.empty()) {
index_failed_files_.erase(file.table_id_);
}
}
return Status::OK();
}
Status
IndexFailedChecker::IgnoreFailedIndexFiles(meta::TableFilesSchema& table_files) {
std::lock_guard<std::mutex> lck(mutex_);
// there could be some failed files belong to different table.
// some files may has failed for several times, no need to build index for these files.
// thus we can avoid dead circle for build index operation
for (auto it_file = table_files.begin(); it_file != table_files.end();) {
auto it_failed_files = index_failed_files_.find((*it_file).table_id_);
if (it_failed_files != index_failed_files_.end()) {
auto it_failed_file = it_failed_files->second.find((*it_file).file_id_);
if (it_failed_file != it_failed_files->second.end()) {
if (it_failed_file->second >= INDEX_FAILED_RETRY_TIME) {
it_file = table_files.erase(it_file);
continue;
}
}
}
++it_file;
}
return Status::OK();
}
} // namespace engine
} // namespace milvus

55
core/src/db/IndexFailedChecker.h Normal file
View File

@ -0,0 +1,55 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#pragma once
#include "db/Types.h"
#include "meta/Meta.h"
#include "utils/Status.h"
#include <map>
#include <mutex>
#include <string>
#include <vector>
namespace milvus {
namespace engine {
class IndexFailedChecker {
public:
Status
CleanFailedIndexFileOfTable(const std::string& table_id);
Status
GetFailedIndexFileOfTable(const std::string& table_id, std::vector<std::string>& failed_files);
Status
MarkFailedIndexFile(const meta::TableFileSchema& file);
Status
MarkSucceedIndexFile(const meta::TableFileSchema& file);
Status
IgnoreFailedIndexFiles(meta::TableFilesSchema& table_files);
private:
std::mutex mutex_;
Table2Files index_failed_files_; // table id mapping to (file id mapping to failed times)
};
} // namespace engine
} // namespace milvus

130
core/src/db/OngoingFileChecker.cpp Normal file
View File

@ -0,0 +1,130 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#include "db/OngoingFileChecker.h"
#include "utils/Log.h"
#include <utility>
namespace milvus {
namespace engine {
Status
OngoingFileChecker::MarkOngoingFile(const meta::TableFileSchema& table_file) {
std::lock_guard<std::mutex> lck(mutex_);
return MarkOngoingFileNoLock(table_file);
}
Status
OngoingFileChecker::MarkOngoingFiles(const meta::TableFilesSchema& table_files) {
std::lock_guard<std::mutex> lck(mutex_);
for (auto& table_file : table_files) {
MarkOngoingFileNoLock(table_file);
}
return Status::OK();
}
Status
OngoingFileChecker::UnmarkOngoingFile(const meta::TableFileSchema& table_file) {
std::lock_guard<std::mutex> lck(mutex_);
return UnmarkOngoingFileNoLock(table_file);
}
Status
OngoingFileChecker::UnmarkOngoingFiles(const meta::TableFilesSchema& table_files) {
std::lock_guard<std::mutex> lck(mutex_);
for (auto& table_file : table_files) {
UnmarkOngoingFileNoLock(table_file);
}
return Status::OK();
}
bool
OngoingFileChecker::IsIgnored(const meta::TableFileSchema& schema) {
std::lock_guard<std::mutex> lck(mutex_);
auto iter = ongoing_files_.find(schema.table_id_);
if (iter == ongoing_files_.end()) {
return false;
} else {
auto it_file = iter->second.find(schema.file_id_);
if (it_file == iter->second.end()) {
return false;
} else {
return (it_file->second > 0);
}
}
}
Status
OngoingFileChecker::MarkOngoingFileNoLock(const meta::TableFileSchema& table_file) {
if (table_file.table_id_.empty() || table_file.file_id_.empty()) {
return Status(DB_ERROR, "Invalid table files");
}
auto iter = ongoing_files_.find(table_file.table_id_);
if (iter == ongoing_files_.end()) {
File2RefCount files_refcount;
files_refcount.insert(std::make_pair(table_file.file_id_, 1));
ongoing_files_.insert(std::make_pair(table_file.table_id_, files_refcount));
} else {
auto it_file = iter->second.find(table_file.file_id_);
if (it_file == iter->second.end()) {
iter->second[table_file.file_id_] = 1;
} else {
it_file->second++;
}
}
ENGINE_LOG_DEBUG << "Mark ongoing file:" << table_file.file_id_
<< " refcount:" << ongoing_files_[table_file.table_id_][table_file.file_id_];
return Status::OK();
}
Status
OngoingFileChecker::UnmarkOngoingFileNoLock(const meta::TableFileSchema& table_file) {
if (table_file.table_id_.empty() || table_file.file_id_.empty()) {
return Status(DB_ERROR, "Invalid table files");
}
auto iter = ongoing_files_.find(table_file.table_id_);
if (iter != ongoing_files_.end()) {
auto it_file = iter->second.find(table_file.file_id_);
if (it_file != iter->second.end()) {
it_file->second--;
ENGINE_LOG_DEBUG << "Unmark ongoing file:" << table_file.file_id_ << " refcount:" << it_file->second;
if (it_file->second <= 0) {
iter->second.erase(table_file.file_id_);
if (iter->second.empty()) {
ongoing_files_.erase(table_file.table_id_);
}
}
}
}
return Status::OK();
}
} // namespace engine
} // namespace milvus

62
core/src/db/OngoingFileChecker.h Normal file
View File

@ -0,0 +1,62 @@
// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements. See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership. The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied. See the License for the
// specific language governing permissions and limitations
// under the License.
#pragma once
#include "db/Types.h"
#include "meta/Meta.h"
#include "utils/Status.h"
#include <map>
#include <mutex>
#include <set>
#include <string>
namespace milvus {
namespace engine {
class OngoingFileChecker : public meta::Meta::CleanUpFilter {
public:
Status
MarkOngoingFile(const meta::TableFileSchema& table_file);
Status
MarkOngoingFiles(const meta::TableFilesSchema& table_files);
Status
UnmarkOngoingFile(const meta::TableFileSchema& table_file);
Status
UnmarkOngoingFiles(const meta::TableFilesSchema& table_files);
bool
IsIgnored(const meta::TableFileSchema& schema) override;
private:
Status
MarkOngoingFileNoLock(const meta::TableFileSchema& table_file);
Status
UnmarkOngoingFileNoLock(const meta::TableFileSchema& table_file);
private:
std::mutex mutex_;
Table2Files ongoing_files_; // table id mapping to (file id mapping to ongoing ref-count)
};
} // namespace engine
} // namespace milvus

6
core/src/db/Types.h
View File

@ -21,6 +21,9 @@
#include <faiss/Index.h>
#include <stdint.h>
#include <map>
#include <set>
#include <string>
#include <utility>
#include <vector>
@ -40,5 +43,8 @@ struct TableIndex {
int32_t metric_type_ = (int)MetricType::L2;
};
using File2RefCount = std::map<std::string, int64_t>;
using Table2Files = std::map<std::string, File2RefCount>;
} // namespace engine
} // namespace milvus

11
core/src/db/engine/ExecutionEngineImpl.cpp
View File

@ -271,6 +271,12 @@ ExecutionEngineImpl::Serialize() {
// here we reset index size by file size,
// since some index type(such as SQ8) data size become smaller after serialized
index_->set_size(PhysicalSize());
ENGINE_LOG_DEBUG << "Finish serialize index file: " << location_ << " size: " << index_->Size();
if (index_->Size() == 0) {
std::string msg = "Failed to serialize file: " + location_ + " reason: out of disk space or memory";
status = Status(DB_ERROR, msg);
}
return status;
}
@ -465,7 +471,9 @@ ExecutionEngineImpl::Merge(const std::string& location) {
if (auto file_index = std::dynamic_pointer_cast<BFIndex>(to_merge)) {
auto status = index_->Add(file_index->Count(), file_index->GetRawVectors(), file_index->GetRawIds());
if (!status.ok()) {
ENGINE_LOG_ERROR << "Merge: Add Error";
ENGINE_LOG_ERROR << "Failed to merge: " << location << " to: " << location_;
} else {
ENGINE_LOG_DEBUG << "Finish merge index file: " << location;
}
return status;
} else {
@ -503,6 +511,7 @@ ExecutionEngineImpl::BuildIndex(const std::string& location, EngineType engine_t
throw Exception(DB_ERROR, status.message());
}
ENGINE_LOG_DEBUG << "Finish build index file: " << location << " size: " << to_index->Size();
return std::make_shared<ExecutionEngineImpl>(to_index, location, engine_type, metric_type_, nlist_);
}

2
core/src/db/insert/MemManagerImpl.cpp
View File

@ -116,6 +116,7 @@ MemManagerImpl::EraseMemVector(const std::string& table_id) {
size_t
MemManagerImpl::GetCurrentMutableMem() {
size_t total_mem = 0;
std::unique_lock<std::mutex> lock(mutex_);
for (auto& kv : mem_id_map_) {
auto memTable = kv.second;
total_mem += memTable->GetCurrentMem();
@ -126,6 +127,7 @@ MemManagerImpl::GetCurrentMutableMem() {
size_t
MemManagerImpl::GetCurrentImmutableMem() {
size_t total_mem = 0;
std::unique_lock<std::mutex> lock(serialization_mtx_);
for (auto& mem_table : immu_mem_list_) {
total_mem += mem_table->GetCurrentMem();
}

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

@ -35,6 +35,13 @@ static const char* META_TABLES = "Tables";
static const char* META_TABLEFILES = "TableFiles";
class Meta {
public:
class CleanUpFilter {
public:
virtual bool
IsIgnored(const TableFileSchema& schema) = 0;
};
public:
virtual ~Meta() = default;
@ -93,7 +100,7 @@ class Meta {
DropPartition(const std::string& partition_name) = 0;
virtual Status
ShowPartitions(const std::string& table_name, std::vector<meta::TableSchema>& partiton_schema_array) = 0;
ShowPartitions(const std::string& table_name, std::vector<meta::TableSchema>& partition_schema_array) = 0;
virtual Status
GetPartitionName(const std::string& table_name, const std::string& tag, std::string& partition_name) = 0;
@ -121,10 +128,7 @@ class Meta {
CleanUpShadowFiles() = 0;
virtual Status
CleanUpCacheWithTTL(uint64_t seconds) = 0;
virtual Status
CleanUpFilesWithTTL(uint64_t seconds) = 0;
CleanUpFilesWithTTL(uint64_t seconds, CleanUpFilter* filter = nullptr) = 0;
virtual Status
DropAll() = 0;

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

@ -1212,7 +1212,7 @@ MySQLMetaImpl::DropPartition(const std::string& partition_name) {
}
Status
MySQLMetaImpl::ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partiton_schema_array) {
MySQLMetaImpl::ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partition_schema_array) {
try {
server::MetricCollector metric;
mysqlpp::StoreQueryResult res;
@ -1236,7 +1236,7 @@ MySQLMetaImpl::ShowPartitions(const std::string& table_id, std::vector<meta::Tab
meta::TableSchema partition_schema;
resRow["table_id"].to_string(partition_schema.table_id_);
DescribeTable(partition_schema);
partiton_schema_array.emplace_back(partition_schema);
partition_schema_array.emplace_back(partition_schema);
}
} catch (std::exception& e) {
return HandleException("GENERAL ERROR WHEN SHOW PARTITIONS", e.what());
@ -1783,49 +1783,7 @@ MySQLMetaImpl::CleanUpShadowFiles() {
}
Status
MySQLMetaImpl::CleanUpCacheWithTTL(uint64_t seconds) {
auto now = utils::GetMicroSecTimeStamp();
// erase deleted/backup files from cache
try {
server::MetricCollector metric;
mysqlpp::ScopedConnection connectionPtr(*mysql_connection_pool_, safe_grab_);
if (connectionPtr == nullptr) {
return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
}
mysqlpp::Query cleanUpFilesWithTTLQuery = connectionPtr->query();
cleanUpFilesWithTTLQuery << "SELECT id, table_id, file_id, date"
<< " FROM " << META_TABLEFILES << " WHERE file_type IN ("
<< std::to_string(TableFileSchema::TO_DELETE) << ","
<< std::to_string(TableFileSchema::BACKUP) << ")"
<< " AND updated_time < " << std::to_string(now - seconds * US_PS) << ";";
mysqlpp::StoreQueryResult res = cleanUpFilesWithTTLQuery.store();
TableFileSchema table_file;
std::vector<std::string> idsToDelete;
for (auto& resRow : res) {
table_file.id_ = resRow["id"]; // implicit conversion
resRow["table_id"].to_string(table_file.table_id_);
resRow["file_id"].to_string(table_file.file_id_);
table_file.date_ = resRow["date"];
utils::GetTableFilePath(options_, table_file);
server::CommonUtil::EraseFromCache(table_file.location_);
}
} catch (std::exception& e) {
return HandleException("GENERAL ERROR WHEN CLEANING UP FILES WITH TTL", e.what());
}
return Status::OK();
}
Status
MySQLMetaImpl::CleanUpFilesWithTTL(uint64_t seconds) {
MySQLMetaImpl::CleanUpFilesWithTTL(uint64_t seconds, CleanUpFilter* filter) {
auto now = utils::GetMicroSecTimeStamp();
std::set<std::string> table_ids;
@ -1840,33 +1798,52 @@ MySQLMetaImpl::CleanUpFilesWithTTL(uint64_t seconds) {
return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
}
mysqlpp::Query cleanUpFilesWithTTLQuery = connectionPtr->query();
cleanUpFilesWithTTLQuery << "SELECT id, table_id, file_id, date"
<< " FROM " << META_TABLEFILES
<< " WHERE file_type = " << std::to_string(TableFileSchema::TO_DELETE)
<< " AND updated_time < " << std::to_string(now - seconds * US_PS) << ";";
mysqlpp::Query query = connectionPtr->query();
query << "SELECT id, table_id, file_id, file_type, date"
<< " FROM " << META_TABLEFILES << " WHERE file_type IN ("
<< std::to_string(TableFileSchema::TO_DELETE) << "," << std::to_string(TableFileSchema::BACKUP) << ")"
<< " AND updated_time < " << std::to_string(now - seconds * US_PS) << ";";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << query.str();
mysqlpp::StoreQueryResult res = cleanUpFilesWithTTLQuery.store();
mysqlpp::StoreQueryResult res = query.store();
TableFileSchema table_file;
std::vector<std::string> idsToDelete;
int64_t clean_files = 0;
for (auto& resRow : res) {
table_file.id_ = resRow["id"]; // implicit conversion
resRow["table_id"].to_string(table_file.table_id_);
resRow["file_id"].to_string(table_file.file_id_);
table_file.date_ = resRow["date"];
table_file.file_type_ = resRow["file_type"];
utils::DeleteTableFilePath(options_, table_file);
// check if the file can be deleted
if (filter && filter->IsIgnored(table_file)) {
ENGINE_LOG_DEBUG << "File:" << table_file.file_id_
<< " currently is in use, not able to delete now";
continue; // ignore this file, don't delete it
}
ENGINE_LOG_DEBUG << "Removing file id:" << table_file.id_ << " location:" << table_file.location_;
// erase file data from cache
// because GetTableFilePath won't able to generate file path after the file is deleted
utils::GetTableFilePath(options_, table_file);
server::CommonUtil::EraseFromCache(table_file.location_);
idsToDelete.emplace_back(std::to_string(table_file.id_));
table_ids.insert(table_file.table_id_);
if (table_file.file_type_ == (int)TableFileSchema::TO_DELETE) {
// delete file from disk storage
utils::DeleteTableFilePath(options_, table_file);
ENGINE_LOG_DEBUG << "Remove file id:" << table_file.id_ << " location:" << table_file.location_;
idsToDelete.emplace_back(std::to_string(table_file.id_));
table_ids.insert(table_file.table_id_);
clean_files++;
}
}
// delete file from meta
if (!idsToDelete.empty()) {
std::stringstream idsToDeleteSS;
for (auto& id : idsToDelete) {
@ -1875,18 +1852,17 @@ MySQLMetaImpl::CleanUpFilesWithTTL(uint64_t seconds) {
std::string idsToDeleteStr = idsToDeleteSS.str();
idsToDeleteStr = idsToDeleteStr.substr(0, idsToDeleteStr.size() - 4); // remove the last " OR "
cleanUpFilesWithTTLQuery << "DELETE FROM " << META_TABLEFILES << " WHERE " << idsToDeleteStr << ";";
query << "DELETE FROM " << META_TABLEFILES << " WHERE " << idsToDeleteStr << ";";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << query.str();
if (!cleanUpFilesWithTTLQuery.exec()) {
return HandleException("QUERY ERROR WHEN CLEANING UP FILES WITH TTL",
cleanUpFilesWithTTLQuery.error());
if (!query.exec()) {
return HandleException("QUERY ERROR WHEN CLEANING UP FILES WITH TTL", query.error());
}
}
if (res.size() > 0) {
ENGINE_LOG_DEBUG << "Clean " << res.size() << " files deleted in " << seconds << " seconds";
if (clean_files > 0) {
ENGINE_LOG_DEBUG << "Clean " << clean_files << " files expired in " << seconds << " seconds";
}
} // Scoped Connection
} catch (std::exception& e) {
@ -1904,14 +1880,13 @@ MySQLMetaImpl::CleanUpFilesWithTTL(uint64_t seconds) {
return Status(DB_ERROR, "Failed to connect to meta server(mysql)");
}
mysqlpp::Query cleanUpFilesWithTTLQuery = connectionPtr->query();
cleanUpFilesWithTTLQuery << "SELECT id, table_id"
<< " FROM " << META_TABLES
<< " WHERE state = " << std::to_string(TableSchema::TO_DELETE) << ";";
mysqlpp::Query query = connectionPtr->query();
query << "SELECT id, table_id"
<< " FROM " << META_TABLES << " WHERE state = " << std::to_string(TableSchema::TO_DELETE) << ";";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << query.str();
mysqlpp::StoreQueryResult res = cleanUpFilesWithTTLQuery.store();
mysqlpp::StoreQueryResult res = query.store();
int64_t remove_tables = 0;
if (!res.empty()) {
@ -1927,13 +1902,12 @@ MySQLMetaImpl::CleanUpFilesWithTTL(uint64_t seconds) {
}
std::string idsToDeleteStr = idsToDeleteSS.str();
idsToDeleteStr = idsToDeleteStr.substr(0, idsToDeleteStr.size() - 4); // remove the last " OR "
cleanUpFilesWithTTLQuery << "DELETE FROM " << META_TABLES << " WHERE " << idsToDeleteStr << ";";
query << "DELETE FROM " << META_TABLES << " WHERE " << idsToDeleteStr << ";";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << query.str();
if (!cleanUpFilesWithTTLQuery.exec()) {
return HandleException("QUERY ERROR WHEN CLEANING UP TABLES WITH TTL",
cleanUpFilesWithTTLQuery.error());
if (!query.exec()) {
return HandleException("QUERY ERROR WHEN CLEANING UP TABLES WITH TTL", query.error());
}
}
@ -1958,14 +1932,13 @@ MySQLMetaImpl::CleanUpFilesWithTTL(uint64_t seconds) {
}
for (auto& table_id : table_ids) {
mysqlpp::Query cleanUpFilesWithTTLQuery = connectionPtr->query();
cleanUpFilesWithTTLQuery << "SELECT file_id"
<< " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote
<< table_id << ";";
mysqlpp::Query query = connectionPtr->query();
query << "SELECT file_id"
<< " FROM " << META_TABLEFILES << " WHERE table_id = " << mysqlpp::quote << table_id << ";";
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << cleanUpFilesWithTTLQuery.str();
ENGINE_LOG_DEBUG << "MySQLMetaImpl::CleanUpFilesWithTTL: " << query.str();
mysqlpp::StoreQueryResult res = cleanUpFilesWithTTLQuery.store();
mysqlpp::StoreQueryResult res = query.store();
if (res.empty()) {
utils::DeleteTablePath(options_, table_id);

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

@ -91,7 +91,7 @@ class MySQLMetaImpl : public Meta {
DropPartition(const std::string& partition_name) override;
Status
ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partiton_schema_array) override;
ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partition_schema_array) override;
Status
GetPartitionName(const std::string& table_id, const std::string& tag, std::string& partition_name) override;
@ -120,10 +120,7 @@ class MySQLMetaImpl : public Meta {
CleanUpShadowFiles() override;
Status
CleanUpCacheWithTTL(uint64_t seconds) override;
Status
CleanUpFilesWithTTL(uint64_t seconds) override;
CleanUpFilesWithTTL(uint64_t seconds, CleanUpFilter* filter = nullptr) override;
Status
DropAll() override;

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

@ -804,7 +804,7 @@ SqliteMetaImpl::DropPartition(const std::string& partition_name) {
}
Status
SqliteMetaImpl::ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partiton_schema_array) {
SqliteMetaImpl::ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partition_schema_array) {
try {
server::MetricCollector metric;
@ -816,7 +816,7 @@ SqliteMetaImpl::ShowPartitions(const std::string& table_id, std::vector<meta::Ta
meta::TableSchema partition_schema;
partition_schema.table_id_ = partition_name;
DescribeTable(partition_schema);
partiton_schema_array.emplace_back(partition_schema);
partition_schema_array.emplace_back(partition_schema);
}
} catch (std::exception& e) {
return HandleException("Encounter exception when show partitions", e.what());
@ -1294,51 +1294,7 @@ SqliteMetaImpl::CleanUpShadowFiles() {
}
Status
SqliteMetaImpl::CleanUpCacheWithTTL(uint64_t seconds) {
auto now = utils::GetMicroSecTimeStamp();
// erase deleted/backup files from cache
try {
server::MetricCollector metric;
// multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
std::lock_guard<std::mutex> meta_lock(meta_mutex_);
std::vector<int> file_types = {
(int)TableFileSchema::TO_DELETE,
(int)TableFileSchema::BACKUP,
};
auto files = ConnectorPtr->select(columns(&TableFileSchema::id_,
&TableFileSchema::table_id_,
&TableFileSchema::file_id_,
&TableFileSchema::date_),
where(
in(&TableFileSchema::file_type_, file_types)
and
c(&TableFileSchema::updated_time_)
< now - seconds * US_PS));
for (auto& file : files) {
TableFileSchema table_file;
table_file.id_ = std::get<0>(file);
table_file.table_id_ = std::get<1>(file);
table_file.file_id_ = std::get<2>(file);
table_file.date_ = std::get<3>(file);
utils::GetTableFilePath(options_, table_file);
server::CommonUtil::EraseFromCache(table_file.location_);
}
} catch (std::exception& e) {
return HandleException("Encounter exception when clean cache", e.what());
}
return Status::OK();
}
Status
SqliteMetaImpl::CleanUpFilesWithTTL(uint64_t seconds) {
SqliteMetaImpl::CleanUpFilesWithTTL(uint64_t seconds, CleanUpFilter* filter) {
auto now = utils::GetMicroSecTimeStamp();
std::set<std::string> table_ids;
@ -1346,33 +1302,60 @@ SqliteMetaImpl::CleanUpFilesWithTTL(uint64_t seconds) {
try {
server::MetricCollector metric;
std::vector<int> file_types = {
(int)TableFileSchema::TO_DELETE,
(int)TableFileSchema::BACKUP,
};
// multi-threads call sqlite update may get exception('bad logic', etc), so we add a lock here
std::lock_guard<std::mutex> meta_lock(meta_mutex_);
// collect files to be deleted
auto files = ConnectorPtr->select(columns(&TableFileSchema::id_,
&TableFileSchema::table_id_,
&TableFileSchema::file_id_,
&TableFileSchema::file_type_,
&TableFileSchema::date_),
where(
c(&TableFileSchema::file_type_) ==
(int)TableFileSchema::TO_DELETE
in(&TableFileSchema::file_type_, file_types)
and
c(&TableFileSchema::updated_time_)
< now - seconds * US_PS));
int64_t clean_files = 0;
auto commited = ConnectorPtr->transaction([&]() mutable {
TableFileSchema table_file;
for (auto& file : files) {
table_file.id_ = std::get<0>(file);
table_file.table_id_ = std::get<1>(file);
table_file.file_id_ = std::get<2>(file);
table_file.date_ = std::get<3>(file);
table_file.file_type_ = std::get<3>(file);
table_file.date_ = std::get<4>(file);
utils::DeleteTableFilePath(options_, table_file);
ENGINE_LOG_DEBUG << "Removing file id:" << table_file.file_id_ << " location:" << table_file.location_;
ConnectorPtr->remove<TableFileSchema>(table_file.id_);
// check if the file can be deleted
if (filter && filter->IsIgnored(table_file)) {
ENGINE_LOG_DEBUG << "File:" << table_file.file_id_
<< " currently is in use, not able to delete now";
continue; // ignore this file, don't delete it
}
table_ids.insert(table_file.table_id_);
// erase from cache, must do this before file deleted,
// because GetTableFilePath won't able to generate file path after the file is deleted
utils::GetTableFilePath(options_, table_file);
server::CommonUtil::EraseFromCache(table_file.location_);
if (table_file.file_type_ == (int)TableFileSchema::TO_DELETE) {
// delete file from meta
ConnectorPtr->remove<TableFileSchema>(table_file.id_);
// delete file from disk storage
utils::DeleteTableFilePath(options_, table_file);
ENGINE_LOG_DEBUG << "Remove file id:" << table_file.file_id_ << " location:" << table_file.location_;
table_ids.insert(table_file.table_id_);
clean_files++;
}
}
return true;
});
@ -1381,8 +1364,8 @@ SqliteMetaImpl::CleanUpFilesWithTTL(uint64_t seconds) {
return HandleException("CleanUpFilesWithTTL error: sqlite transaction failed");
}
if (files.size() > 0) {
ENGINE_LOG_DEBUG << "Clean " << files.size() << " files deleted in " << seconds << " seconds";
if (clean_files > 0) {
ENGINE_LOG_DEBUG << "Clean " << clean_files << " files expired in " << seconds << " seconds";
}
} catch (std::exception& e) {
return HandleException("Encounter exception when clean table files", e.what());

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

@ -91,7 +91,7 @@ class SqliteMetaImpl : public Meta {
DropPartition(const std::string& partition_name) override;
Status
ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partiton_schema_array) override;
ShowPartitions(const std::string& table_id, std::vector<meta::TableSchema>& partition_schema_array) override;
Status
GetPartitionName(const std::string& table_id, const std::string& tag, std::string& partition_name) override;
@ -120,10 +120,7 @@ class SqliteMetaImpl : public Meta {
CleanUpShadowFiles() override;
Status
CleanUpCacheWithTTL(uint64_t seconds) override;
Status
CleanUpFilesWithTTL(uint64_t seconds) override;
CleanUpFilesWithTTL(uint64_t seconds, CleanUpFilter* filter = nullptr) override;
Status
DropAll() override;

2
core/src/index/cmake/ThirdPartyPackagesCore.cmake
View File

@ -708,7 +708,7 @@ macro(build_faiss)
set(FAISS_CONFIGURE_ARGS
"--prefix=${FAISS_PREFIX}"
"CFLAGS=${EP_C_FLAGS}"
"CXXFLAGS=${EP_CXX_FLAGS} -mavx2 -mf16c"
"CXXFLAGS=${EP_CXX_FLAGS} -mavx2 -mf16c -O3"
--without-python)
if (FAISS_WITH_MKL)

34
core/src/index/knowhere/knowhere/index/vector_index/IndexGPUIDMAP.cpp
View File

@ -126,4 +126,38 @@ GPUIDMAP::search_impl(int64_t n, const float* data, int64_t k, float* distances,
index_->search(n, (float*)data, k, distances, labels);
}
void
GPUIDMAP::GenGraph(float* data, const int64_t& k, Graph& graph, const Config& config) {
int64_t K = k + 1;
auto ntotal = Count();
size_t dim = config->d;
auto batch_size = 1000;
auto tail_batch_size = ntotal % batch_size;
auto batch_search_count = ntotal / batch_size;
auto total_search_count = tail_batch_size == 0 ? batch_search_count : batch_search_count + 1;
std::vector<float> res_dis(K * batch_size);
graph.resize(ntotal);
Graph res_vec(total_search_count);
for (int i = 0; i < total_search_count; ++i) {
auto b_size = (i == (total_search_count - 1)) && tail_batch_size != 0 ? tail_batch_size : batch_size;
auto& res = res_vec[i];
res.resize(K * b_size);
auto xq = data + batch_size * dim * i;
search_impl(b_size, (float*)xq, K, res_dis.data(), res.data(), config);
for (int j = 0; j < b_size; ++j) {
auto& node = graph[batch_size * i + j];
node.resize(k);
auto start_pos = j * K + 1;
for (int m = 0, cursor = start_pos; m < k && cursor < start_pos + k; ++m, ++cursor) {
node[m] = res[cursor];
}
}
}
}
} // namespace knowhere

4
core/src/index/knowhere/knowhere/index/vector_index/IndexGPUIDMAP.h
View File

@ -23,6 +23,7 @@
#include <memory>
#include <utility>
#include <vector>
namespace knowhere {
@ -47,6 +48,9 @@ class GPUIDMAP : public IDMAP, public GPUIndex {
VectorIndexPtr
CopyGpuToGpu(const int64_t& device_id, const Config& config) override;
void
GenGraph(float* data, const int64_t& k, Graph& graph, const Config& config);
protected:
void
search_impl(int64_t n, const float* data, int64_t k, float* distances, int64_t* labels, const Config& cfg) override;

20
core/src/index/knowhere/knowhere/index/vector_index/IndexIDMAP.cpp
View File

@ -121,6 +121,26 @@ IDMAP::Add(const DatasetPtr& dataset, const Config& config) {
index_->add_with_ids(rows, (float*)p_data, p_ids);
}
void
IDMAP::AddWithoutId(const DatasetPtr& dataset, const Config& config) {
if (!index_) {
KNOWHERE_THROW_MSG("index not initialize");
}
std::lock_guard<std::mutex> lk(mutex_);
GETTENSOR(dataset)
// TODO: magic here.
auto array = dataset->array()[0];
std::vector<int64_t> new_ids(rows);
for (int i = 0; i < rows; ++i) {
new_ids[i] = i;
}
index_->add_with_ids(rows, (float*)p_data, new_ids.data());
}
int64_t
IDMAP::Count() {
return index_->ntotal;

14
core/src/index/knowhere/knowhere/index/vector_index/IndexIDMAP.h
View File

@ -34,20 +34,31 @@ class IDMAP : public VectorIndex, public FaissBaseIndex {
BinarySet
Serialize() override;
void
Load(const BinarySet& index_binary) override;
void
Train(const Config& config);
DatasetPtr
Search(const DatasetPtr& dataset, const Config& config) override;
int64_t
Count() override;
// VectorIndexPtr
// Clone() override;
int64_t
Dimension() override;
void
Add(const DatasetPtr& dataset, const Config& config) override;
void
AddWithoutId(const DatasetPtr& dataset, const Config& config);
VectorIndexPtr
CopyCpuToGpu(const int64_t& device_id, const Config& config);
void
@ -55,12 +66,15 @@ class IDMAP : public VectorIndex, public FaissBaseIndex {
virtual float*
GetRawVectors();
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);
protected:
std::mutex mutex_;
};

25
core/src/index/knowhere/knowhere/index/vector_index/IndexIVF.cpp
View File

@ -195,35 +195,34 @@ IVF::Dimension() {
}
void
IVF::GenGraph(const int64_t& k, Graph& graph, const DatasetPtr& dataset, const Config& config) {
GETTENSOR(dataset)
IVF::GenGraph(float* data, const int64_t& k, Graph& graph, const Config& config) {
int64_t K = k + 1;
auto ntotal = Count();
auto batch_size = 100;
size_t dim = config->d;
auto batch_size = 1000;
auto tail_batch_size = ntotal % batch_size;
auto batch_search_count = ntotal / batch_size;
auto total_search_count = tail_batch_size == 0 ? batch_search_count : batch_search_count + 1;
std::vector<float> res_dis(k * batch_size);
std::vector<float> res_dis(K * batch_size);
graph.resize(ntotal);
Graph res_vec(total_search_count);
for (int i = 0; i < total_search_count; ++i) {
auto b_size = i == total_search_count - 1 && tail_batch_size != 0 ? tail_batch_size : batch_size;
auto b_size = (i == (total_search_count - 1)) && tail_batch_size != 0 ? tail_batch_size : batch_size;
auto& res = res_vec[i];
res.resize(k * b_size);
res.resize(K * b_size);
auto xq = p_data + batch_size * dim * i;
search_impl(b_size, (float*)xq, k, res_dis.data(), res.data(), config);
auto xq = data + batch_size * dim * i;
search_impl(b_size, (float*)xq, K, res_dis.data(), res.data(), config);
int tmp = 0;
for (int j = 0; j < b_size; ++j) {
auto& node = graph[batch_size * i + j];
node.resize(k);
for (int m = 0; m < k && tmp < k * b_size; ++m, ++tmp) {
// TODO(linxj): avoid memcopy here.
node[m] = res[tmp];
auto start_pos = j * K + 1;
for (int m = 0, cursor = start_pos; m < k && cursor < start_pos + k; ++m, ++cursor) {
node[m] = res[cursor];
}
}
}

2
core/src/index/knowhere/knowhere/index/vector_index/IndexIVF.h
View File

@ -57,7 +57,7 @@ class IVF : public VectorIndex, public FaissBaseIndex {
Search(const DatasetPtr& dataset, const Config& config) override;
void
GenGraph(const int64_t& k, Graph& graph, const DatasetPtr& dataset, const Config& config);
GenGraph(float* data, const int64_t& k, Graph& graph, const Config& config);
BinarySet
Serialize() override;

26
core/src/index/knowhere/knowhere/index/vector_index/IndexNSG.cpp
View File

@ -20,9 +20,12 @@
#include "knowhere/common/Exception.h"
#include "knowhere/common/Timer.h"
#ifdef MILVUS_GPU_VERSION
#include "knowhere/index/vector_index/IndexGPUIDMAP.h"
#include "knowhere/index/vector_index/IndexGPUIVF.h"
#include "knowhere/index/vector_index/helpers/Cloner.h"
#endif
#include "knowhere/index/vector_index/IndexIDMAP.h"
#include "knowhere/index/vector_index/IndexIVF.h"
#include "knowhere/index/vector_index/nsg/NSG.h"
#include "knowhere/index/vector_index/nsg/NSGIO.h"
@ -110,33 +113,36 @@ NSG::Search(const DatasetPtr& dataset, const Config& config) {
IndexModelPtr
NSG::Train(const DatasetPtr& dataset, const Config& config) {
config->Dump();
auto build_cfg = std::dynamic_pointer_cast<NSGCfg>(config);
if (build_cfg != nullptr) {
build_cfg->CheckValid(); // throw exception
}
// TODO(linxj): dev IndexFactory, support more IndexType
auto idmap = std::make_shared<IDMAP>();
idmap->Train(config);
idmap->AddWithoutId(dataset, config);
Graph knng;
float* raw_data = idmap->GetRawVectors();
#ifdef MILVUS_GPU_VERSION
if (build_cfg->gpu_id == knowhere::INVALID_VALUE) {
auto preprocess_index = std::make_shared<IVF>();
auto model = preprocess_index->Train(dataset, config);
preprocess_index->set_index_model(model);
preprocess_index->AddWithoutIds(dataset, config);
preprocess_index->GenGraph(build_cfg->knng, knng, dataset, config);
preprocess_index->Add(dataset, config);
preprocess_index->GenGraph(raw_data, build_cfg->knng, knng, config);
} else {
auto preprocess_index = std::make_shared<GPUIVF>(build_cfg->gpu_id);
auto model = preprocess_index->Train(dataset, config);
preprocess_index->set_index_model(model);
preprocess_index->AddWithoutIds(dataset, config);
preprocess_index->GenGraph(build_cfg->knng, knng, dataset, config);
// TODO(linxj): use ivf instead?
auto gpu_idx = cloner::CopyCpuToGpu(idmap, build_cfg->gpu_id, config);
auto gpu_idmap = std::dynamic_pointer_cast<GPUIDMAP>(gpu_idx);
gpu_idmap->GenGraph(raw_data, build_cfg->knng, knng, config);
}
#else
auto preprocess_index = std::make_shared<IVF>();
auto model = preprocess_index->Train(dataset, config);
preprocess_index->set_index_model(model);
preprocess_index->AddWithoutIds(dataset, config);
preprocess_index->GenGraph(build_cfg->knng, knng, dataset, config);
preprocess_index->GenGraph(raw_data, build_cfg->knng, knng, config);
#endif
algo::BuildParams b_params;
@ -144,10 +150,10 @@ NSG::Train(const DatasetPtr& dataset, const Config& config) {
b_params.out_degree = build_cfg->out_degree;
b_params.search_length = build_cfg->search_length;
GETTENSOR(dataset)
auto array = dataset->array()[0];
auto p_ids = array->data()->GetValues<int64_t>(1, 0);
GETTENSOR(dataset)
index_ = std::make_shared<algo::NsgIndex>(dim, rows);
index_->SetKnnGraph(knng);
index_->Build_with_ids(rows, (float*)p_data, (int64_t*)p_ids, b_params);

212
core/src/index/knowhere/knowhere/index/vector_index/nsg/NSG.cpp
View File

@ -18,7 +18,6 @@
#include <algorithm>
#include <cstdlib>
#include <cstring>
#include <fstream>
#include <iostream>
#include <stack>
#include <utility>
@ -29,12 +28,13 @@
#include "knowhere/index/vector_index/nsg/NSG.h"
#include "knowhere/index/vector_index/nsg/NSGHelper.h"
// TODO: enable macro
//#include <gperftools/profiler.h>
namespace knowhere {
namespace algo {
unsigned int seed = 100;
NsgIndex::NsgIndex(const size_t& dimension, const size_t& n, METRICTYPE metric)
: dimension(dimension), ntotal(n), metric_type(metric) {
switch (metric) {
@ -55,8 +55,6 @@ NsgIndex::~NsgIndex() {
void
NsgIndex::Build_with_ids(size_t nb, const float* data, const int64_t* ids, const BuildParams& parameters) {
TimeRecorder rc("NSG");
ntotal = nb;
ori_data_ = new float[ntotal * dimension];
ids_ = new int64_t[ntotal];
@ -67,25 +65,17 @@ NsgIndex::Build_with_ids(size_t nb, const float* data, const int64_t* ids, const
out_degree = parameters.out_degree;
candidate_pool_size = parameters.candidate_pool_size;
TimeRecorder rc("NSG", 1);
InitNavigationPoint();
rc.RecordSection("init");
Link();
rc.RecordSection("Link");
//>> Debug code
/////
// int count = 0;
// for (int i = 0; i < ntotal; ++i) {
// count += nsg[i].size();
//}
/////
CheckConnectivity();
rc.RecordSection("Connect");
//>> Debug code
///
int total_degree = 0;
for (size_t i = 0; i < ntotal; ++i) {
total_degree += nsg[i].size();
@ -93,9 +83,17 @@ NsgIndex::Build_with_ids(size_t nb, const float* data, const int64_t* ids, const
KNOWHERE_LOG_DEBUG << "Graph physical size: " << total_degree * sizeof(node_t) / 1024 / 1024 << "m";
KNOWHERE_LOG_DEBUG << "Average degree: " << total_degree / ntotal;
/////
is_trained = true;
// Debug code
// for (size_t i = 0; i < ntotal; i++) {
// auto& x = nsg[i];
// for (size_t j = 0; j < x.size(); j++) {
// std::cout << "id: " << x[j] << std::endl;
// }
// std::cout << std::endl;
// }
}
void
@ -114,28 +112,22 @@ NsgIndex::InitNavigationPoint() {
}
// select navigation point
std::vector<Neighbor> resset, fullset;
unsigned int seed = 100;
std::vector<Neighbor> resset;
navigation_point = rand_r(&seed) % ntotal; // random initialize navigating point
//>> Debug code
/////
// navigation_point = drand48();
/////
GetNeighbors(center, resset, knng);
navigation_point = resset[0].id;
//>> Debug code
/////
// Debug code
// std::cout << "ep: " << navigation_point << std::endl;
/////
//>> Debug code
/////
// for (int k = 0; k < resset.size(); ++k) {
// std::cout << "id: " << resset[k].id << ", dis: " << resset[k].distance << std::endl;
// }
// std::cout << std::endl;
//
// std::cout << "ep: " << navigation_point << std::endl;
//
// float r1 = distance_->Compare(center, ori_data_ + navigation_point * dimension, dimension);
// assert(r1 == resset[0].distance);
/////
}
// Specify Link
@ -149,7 +141,9 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, std::v
// TODO: throw exception here.
}
std::vector<node_t> init_ids;
resset.resize(search_length);
std::vector<node_t> init_ids(buffer_size);
// std::vector<node_t> init_ids;
{
/*
@ -158,25 +152,26 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, std::v
size_t count = 0;
// Get all neighbors
for (size_t i = 0; i < graph[navigation_point].size(); ++i) {
init_ids.push_back(graph[navigation_point][i]);
for (size_t i = 0; i < init_ids.size() && i < graph[navigation_point].size(); ++i) {
// for (size_t i = 0; i < graph[navigation_point].size(); ++i) {
// init_ids.push_back(graph[navigation_point][i]);
init_ids[i] = graph[navigation_point][i];
has_calculated_dist[init_ids[i]] = true;
++count;
}
unsigned int seed = 100;
while (count < buffer_size) {
node_t id = rand_r(&seed) % ntotal;
if (has_calculated_dist[id])
continue; // duplicate id
init_ids.push_back(id);
// init_ids.push_back(id);
init_ids[count] = id;
++count;
has_calculated_dist[id] = true;
}
}
{
resset.resize(init_ids.size());
// resset.resize(init_ids.size());
// init resset and sort by distance
for (size_t i = 0; i < init_ids.size(); ++i) {
@ -190,7 +185,7 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, std::v
float dist = distance_->Compare(ori_data_ + dimension * id, query, dimension);
resset[i] = Neighbor(id, dist, false);
///////////// difference from other GetNeighbors ///////////////
//// difference from other GetNeighbors
fullset.push_back(resset[i]);
///////////////////////////////////////
}
@ -247,8 +242,10 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, std::v
// TODO: throw exception here.
}
std::vector<node_t> init_ids;
boost::dynamic_bitset<> has_calculated_dist{ntotal, 0}; // TODO: ?
// std::vector<node_t> init_ids;
std::vector<node_t> init_ids(buffer_size);
resset.resize(buffer_size);
boost::dynamic_bitset<> has_calculated_dist{ntotal, 0};
{
/*
@ -257,24 +254,26 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, std::v
size_t count = 0;
// Get all neighbors
for (size_t i = 0; i < graph[navigation_point].size(); ++i) {
init_ids.push_back(graph[navigation_point][i]);
for (size_t i = 0; i < init_ids.size() && i < graph[navigation_point].size(); ++i) {
// for (size_t i = 0; i < graph[navigation_point].size(); ++i) {
// init_ids.push_back(graph[navigation_point][i]);
init_ids[i] = graph[navigation_point][i];
has_calculated_dist[init_ids[i]] = true;
++count;
}
unsigned int seed = 100;
while (count < buffer_size) {
node_t id = rand_r(&seed) % ntotal;
if (has_calculated_dist[id])
continue; // duplicate id
init_ids.push_back(id);
// init_ids.push_back(id);
init_ids[count] = id;
++count;
has_calculated_dist[id] = true;
}
}
{
resset.resize(init_ids.size());
// resset.resize(init_ids.size());
// init resset and sort by distance
for (size_t i = 0; i < init_ids.size(); ++i) {
@ -333,13 +332,15 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, std::v
void
NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, Graph& graph, SearchParams* params) {
size_t& buffer_size = params ? params->search_length : search_length;
size_t buffer_size = params ? params->search_length : search_length;
if (buffer_size > ntotal) {
// TODO: throw exception here.
}
std::vector<node_t> init_ids;
// std::vector<node_t> init_ids;
std::vector<node_t> init_ids(buffer_size);
resset.resize(buffer_size);
boost::dynamic_bitset<> has_calculated_dist{ntotal, 0};
{
@ -349,33 +350,33 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, Graph&
size_t count = 0;
// Get all neighbors
for (size_t i = 0; i < graph[navigation_point].size(); ++i) {
init_ids.push_back(graph[navigation_point][i]);
for (size_t i = 0; i < init_ids.size() && i < graph[navigation_point].size(); ++i) {
// for (size_t i = 0; i < graph[navigation_point].size(); ++i) {
// init_ids.push_back(graph[navigation_point][i]);
init_ids[i] = graph[navigation_point][i];
has_calculated_dist[init_ids[i]] = true;
++count;
}
unsigned int seed = 100;
while (count < buffer_size) {
node_t id = rand_r(&seed) % ntotal;
if (has_calculated_dist[id])
continue; // duplicate id
init_ids.push_back(id);
// init_ids.push_back(id);
init_ids[count] = id;
++count;
has_calculated_dist[id] = true;
}
}
{
resset.resize(init_ids.size());
// resset.resize(init_ids.size());
// init resset and sort by distance
for (size_t i = 0; i < init_ids.size(); ++i) {
node_t id = init_ids[i];
// assert(id < ntotal);
if (id >= static_cast<node_t>(ntotal)) {
KNOWHERE_THROW_MSG("Build Index Error, id > ntotal");
continue;
}
float dist = distance_->Compare(ori_data_ + id * dimension, query, dimension);
@ -383,13 +384,6 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, Graph&
}
std::sort(resset.begin(), resset.end()); // sort by distance
//>> Debug code
/////
// for (int j = 0; j < buffer_size; ++j) {
// std::cout << "resset_id: " << resset[j].id << ", resset_dist: " << resset[j].distance << std::endl;
//}
/////
// search nearest neighbor
size_t cursor = 0;
while (cursor < buffer_size) {
@ -410,7 +404,8 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, Graph&
if (dist >= resset[buffer_size - 1].distance)
continue;
///////////// difference from other GetNeighbors ///////////////
//// difference from other GetNeighbors
Neighbor nn(id, dist, false);
///////////////////////////////////////
@ -440,59 +435,50 @@ NsgIndex::GetNeighbors(const float* query, std::vector<Neighbor>& resset, Graph&
void
NsgIndex::Link() {
auto cut_graph_dist = new float[ntotal * out_degree];
float* cut_graph_dist = new float[ntotal * out_degree];
nsg.resize(ntotal);
#pragma omp parallel
{
std::vector<Neighbor> fullset;
std::vector<Neighbor> temp;
boost::dynamic_bitset<> flags{ntotal, 0}; // TODO: ?
boost::dynamic_bitset<> flags{ntotal, 0};
#pragma omp for schedule(dynamic, 100)
for (size_t n = 0; n < ntotal; ++n) {
fullset.clear();
temp.clear();
flags.reset();
GetNeighbors(ori_data_ + dimension * n, temp, fullset, flags);
//>> Debug code
/////
// float r1 = distance_->Compare(ori_data_ + n * dimension, ori_data_ + temp[0].id * dimension, dimension);
// assert(r1 == temp[0].distance);
/////
SyncPrune(n, fullset, flags, cut_graph_dist);
}
// Debug code
// std::cout << "ep: " << 0 << std::endl;
// for (int k = 0; k < fullset.size(); ++k) {
// std::cout << "id: " << fullset[k].id << ", dis: " << fullset[k].distance << std::endl;
// }
}
//>> Debug code
/////
// auto bak_nsg = nsg;
/////
// Debug code
// for (size_t i = 0; i < ntotal; i++)
// {
// auto& x = nsg[i];
// for (size_t j=0; j < x.size(); j++)
// {
// std::cout << "id: " << x[j] << std::endl;
// }
// std::cout << std::endl;
// }
knng.clear();
knng.shrink_to_fit();
std::vector<std::mutex> mutex_vec(ntotal);
#pragma omp for schedule(dynamic, 100)
for (unsigned n = 0; n < ntotal; ++n) {
InterInsert(n, mutex_vec, cut_graph_dist);
}
delete[] cut_graph_dist;
//>> Debug code
/////
// int count = 0;
// for (int i = 0; i < ntotal; ++i) {
// if (bak_nsg[i].size() != nsg[i].size()) {
// //count += nsg[i].size() - bak_nsg[i].size();
// count += nsg[i].size();
// }
//}
/////
for (size_t i = 0; i < ntotal; ++i) {
nsg[i].shrink_to_fit();
}
}
void
@ -654,9 +640,9 @@ NsgIndex::DFS(size_t root, boost::dynamic_bitset<>& has_linked, int64_t& linked_
std::stack<size_t> s;
s.push(root);
if (!has_linked[root]) {
linked_count++; // not link
has_linked[root] = true; // link start...
linked_count++; // not link
}
has_linked[root] = true; // link start...
while (!s.empty()) {
size_t next = ntotal + 1;
@ -709,7 +695,6 @@ NsgIndex::FindUnconnectedNode(boost::dynamic_bitset<>& has_linked, int64_t& root
}
}
if (found == 0) {
unsigned int seed = 100;
while (true) { // random a linked-node and add unlinked-node as its neighbor
size_t rid = rand_r(&seed) % ntotal;
if (has_linked[rid]) {
@ -726,7 +711,10 @@ NsgIndex::Search(const float* query, const unsigned& nq, const unsigned& dim, co
int64_t* ids, SearchParams& params) {
std::vector<std::vector<Neighbor>> resset(nq);
params.search_length = k;
if (k >= 45) {
params.search_length = k;
}
TimeRecorder rc("NsgIndex::search", 1);
// TODO(linxj): when to use openmp
if (nq <= 4) {
@ -734,7 +722,7 @@ NsgIndex::Search(const float* query, const unsigned& nq, const unsigned& dim, co
} else {
#pragma omp parallel for
for (unsigned int i = 0; i < nq; ++i) {
auto single_query = query + i * dim;
const float* single_query = query + i * dim;
GetNeighbors(single_query, resset[i], nsg, &params);
}
}
@ -759,13 +747,6 @@ NsgIndex::Search(const float* query, const unsigned& nq, const unsigned& dim, co
}
rc.RecordSection("merge");
//>> Debug: test single insert
// int x_0 = resset[0].size();
// for (int l = 0; l < resset[0].size(); ++l) {
// resset[0].pop_back();
//}
// resset.clear();
// ProfilerStart("xx.prof");
// std::vector<Neighbor> resset;
// GetNeighbors(query, resset, nsg, &params);
@ -781,30 +762,5 @@ NsgIndex::SetKnnGraph(Graph& g) {
knng = std::move(g);
}
// void NsgIndex::GetKnnGraphFromFile() {
// //std::string filename = "sift.1M.50NN.graph";
// std::string filename = "sift.50NN.graph";
//
// std::ifstream in(filename, std::ios::binary);
// unsigned k;
// in.read((char *) &k, sizeof(unsigned));
// in.seekg(0, std::ios::end);
// std::ios::pos_type ss = in.tellg();
// size_t fsize = (size_t) ss;
// size_t num = (unsigned) (fsize / (k + 1) / 4);
// in.seekg(0, std::ios::beg);
//
// knng.resize(num);
// knng.reserve(num);
// unsigned kk = (k + 3) / 4 * 4;
// for (size_t i = 0; i < num; i++) {
// in.seekg(4, std::ios::cur);
// knng[i].resize(k);
// knng[i].reserve(kk);
// in.read((char *) knng[i].data(), k * sizeof(unsigned));
// }
// in.close();
//}
} // namespace algo
} // namespace knowhere

9
core/src/index/knowhere/knowhere/index/vector_index/nsg/NSG.h
View File

@ -52,9 +52,9 @@ class NsgIndex {
Distance* distance_;
float* ori_data_;
int64_t* ids_; // TODO: support different type
Graph nsg; // final graph
Graph knng; // reset after build
int64_t* ids_;
Graph nsg; // final graph
Graph knng; // reset after build
node_t navigation_point; // offset of node in origin data
@ -134,9 +134,6 @@ class NsgIndex {
void
FindUnconnectedNode(boost::dynamic_bitset<>& flags, int64_t& root);
// private:
// void GetKnnGraphFromFile();
};
} // namespace algo

6
core/src/index/thirdparty/faiss/InvertedLists.cpp vendored
View File

@ -20,6 +20,12 @@
namespace faiss {
/*
* Use pin memory to build Readonly Inverted list will accelerate cuda memory copy, but it will downgrade cpu ivf search
* performance. read only inverted list structure will also make ivf search performance not stable. ISSUE 500 mention
* this problem. Best performance is the original inverted list with non pin memory.
*/
PageLockMemory::PageLockMemory(size_t size) : nbytes(size) {
CUDA_VERIFY(cudaHostAlloc(&data, size, 0));
}

2
core/src/index/thirdparty/faiss/makefile.inc.in vendored
View File

@ -17,7 +17,7 @@ NVCC = @NVCC@
CUDA_ROOT = @CUDA_PREFIX@
CUDA_ARCH = @CUDA_ARCH@
NVCCFLAGS = -I $(CUDA_ROOT)/targets/x86_64-linux/include/ \
-O0 -g \
-O3 \
-Xcompiler -fPIC \
-Xcudafe --diag_suppress=unrecognized_attribute \
$(CUDA_ARCH) \

174
core/src/index/unittest/test_nsg/test_nsg.cpp
View File

@ -22,6 +22,8 @@
#include "knowhere/index/vector_index/FaissBaseIndex.h"
#include "knowhere/index/vector_index/IndexNSG.h"
#ifdef MILVUS_GPU_VERSION
#include "knowhere/index/vector_index/IndexGPUIDMAP.h"
#include "knowhere/index/vector_index/helpers/Cloner.h"
#include "knowhere/index/vector_index/helpers/FaissGpuResourceMgr.h"
#endif
@ -50,6 +52,7 @@ class NSGInterfaceTest : public DataGen, public ::testing::Test {
auto tmp_conf = std::make_shared<knowhere::NSGCfg>();
tmp_conf->gpu_id = DEVICEID;
tmp_conf->d = 256;
tmp_conf->knng = 20;
tmp_conf->nprobe = 8;
tmp_conf->nlist = 163;
@ -116,3 +119,174 @@ TEST_F(NSGInterfaceTest, comparetest) {
}
tc.RecordSection("IP");
}
//#include <src/index/knowhere/knowhere/index/vector_index/nsg/OriNSG.h>
// TEST(test, ori_nsg) {
// // float* p_data = nullptr;
// size_t rows, dim;
// char* filename = "/mnt/112d53a6-5592-4360-a33b-7fd789456fce/workspace/Data/sift/sift_base.fvecs";
// // loads_data(filename, p_data, rows, dim);
// float* p_data = fvecs_read(filename, &dim, &rows);
//
// std::string knng_filename =
// "/mnt/112d53a6-5592-4360-a33b-7fd789456fce/workspace/Cellar/anns/efanna_graph/tests/sift.1M.50NN.graph";
// std::vector<std::vector<int64_t>> knng;
// Load_nns_graph(knng, knng_filename.c_str());
//
// // float* search_data = nullptr;
// size_t nq, search_dim;
// char* searchfile = "/mnt/112d53a6-5592-4360-a33b-7fd789456fce/workspace/Data/sift/sift_query.fvecs";
// // loads_data(searchfile, search_data, nq, search_dim);
// float* search_data = fvecs_read(searchfile, &search_dim, &nq);
// assert(search_dim == dim);
//
// size_t k, nq2;
// char* gtfile = "/mnt/112d53a6-5592-4360-a33b-7fd789456fce/workspace/Data/sift/sift_groundtruth.ivecs";
// int* gt_int = ivecs_read(gtfile, &k, &nq2);
// int64_t* gt = new int64_t[k * nq2];
// for (int i = 0; i < k * nq2; i++) {
// gt[i] = gt_int[i];
// }
// delete[] gt_int;
//
// std::vector<int64_t> store_ids(rows);
// for (int i = 0; i < rows; ++i) {
// store_ids[i] = i;
// }
//
// int64_t* I = new int64_t[nq * k];
// float* D = new float[nq * k];
//#if 0
// efanna2e::Parameters params;
// params.Set<int64_t>("L", 50);
// params.Set<int64_t>("R", 55);
// params.Set<int64_t>("C", 300);
// auto orinsg = std::make_shared<efanna2e::IndexNSG>(dim, rows, efanna2e::Metric::L2, nullptr);
// orinsg->Load_nn_graph(knng);
// orinsg->Build(rows, (float*)p_data, params);
//
// efanna2e::Parameters paras;
// paras.Set<unsigned>("L_search", 45);
// paras.Set<unsigned>("P_search",100);
// k = 10;
// std::vector<std::vector<int64_t> > res;
// for (unsigned i = 0; i < nq; i++) {
// std::vector<int64_t> tmp(k);
// orinsg->Search(search_data + i * dim, p_data, k, paras, tmp.data());
// res.push_back(tmp);
// }
// }
//#else
// knowhere::algo::BuildParams params;
// params.search_length = 50;
// params.out_degree = 55;
// params.candidate_pool_size = 300;
// auto nsg = std::make_shared<knowhere::algo::NsgIndex>(dim, rows);
//#if 1
// knowhere::FaissGpuResourceMgr::GetInstance().InitDevice(DEVICEID, 1024 * 1024 * 200, 1024 * 1024 * 600, 2);
// auto dataset = generate_dataset(int64_t(rows), int64_t(dim), p_data, store_ids.data());
// auto config = std::make_shared<knowhere::IVFCfg>();
// config->d = dim;
// config->gpu_id = 0;
// config->metric_type = knowhere::METRICTYPE::L2;
// auto preprocess_index = std::make_shared<knowhere::IDMAP>();
// preprocess_index->Train(config);
// preprocess_index->AddWithoutId(dataset, config);
// auto xx = knowhere::cloner::CopyCpuToGpu(preprocess_index, 0, config);
// auto ss = std::dynamic_pointer_cast<knowhere::GPUIDMAP>(xx);
//
// std::vector<std::vector<int64_t>> kng;
// ss->GenGraph(p_data, 50, kng, config);
// nsg->SetKnnGraph(kng);
// knowhere::FaissGpuResourceMgr::GetInstance().Free();
//#else
// nsg->SetKnnGraph(knng);
//#endif
// nsg->Build_with_ids(rows, (float*)p_data, store_ids.data(), params);
// knowhere::algo::SearchParams s_params;
// s_params.search_length = 45;
// nsg->Search(search_data, nq, dim, k, D, I, s_params);
//#endif
//
// int n_1 = 0, n_10 = 0, n_100 = 0;
// for (int i = 0; i < nq; i++) {
// int gt_nn = gt[i * k];
// for (int j = 0; j < k; j++) {
// if (I[i * k + j] == gt_nn) {
// if (j < 1)
// n_1++;
// if (j < 10)
// n_10++;
// if (j < 100)
// n_100++;
// }
// }
// }
// printf("R@1 = %.4f\n", n_1 / float(nq));
// printf("R@10 = %.4f\n", n_10 / float(nq));
// printf("R@100 = %.4f\n", n_100 / float(nq));
//}
//
// TEST(testxx, test_idmap){
// int k = 50;
// std::string knng_filename =
// "/mnt/112d53a6-5592-4360-a33b-7fd789456fce/workspace/Cellar/anns/efanna_graph/tests/sift.50NN.graph";
// std::vector<std::vector<int64_t>> gt_knng;
// Load_nns_graph(gt_knng, knng_filename.c_str());
//
// size_t rows, dim;
// char* filename =
// "/mnt/112d53a6-5592-4360-a33b-7fd789456fce/workspace/Cellar/anns/efanna_graph/tests/siftsmall/siftsmall_base.fvecs";
// float* p_data = fvecs_read(filename, &dim, &rows);
//
// std::vector<int64_t> store_ids(rows);
// for (int i = 0; i < rows; ++i) {
// store_ids[i] = i;
// }
//
// knowhere::FaissGpuResourceMgr::GetInstance().InitDevice(DEVICEID, 1024 * 1024 * 200, 1024 * 1024 * 600, 2);
// auto dataset = generate_dataset(int64_t(rows), int64_t(dim), p_data, store_ids.data());
// auto config = std::make_shared<knowhere::IVFCfg>();
// config->d = dim;
// config->gpu_id = 0;
// config->metric_type = knowhere::METRICTYPE::L2;
// auto preprocess_index = std::make_shared<knowhere::IDMAP>();
// preprocess_index->Train(config);
// preprocess_index->AddWithoutId(dataset, config);
// auto xx = knowhere::cloner::CopyCpuToGpu(preprocess_index, 0, config);
// auto ss = std::dynamic_pointer_cast<knowhere::GPUIDMAP>(xx);
// std::vector<std::vector<int64_t>> idmap_knng;
// ss->GenGraph(p_data, k, idmap_knng,config);
// knowhere::FaissGpuResourceMgr::GetInstance().Free();
//
// int n_1 = 0, n_10 = 0, n_100 = 0;
// for (int i = 0; i < rows; i++) {
// int gt_nn = gt_knng[i][0];
// int l_n_1 = 0;
// int l_n_10 = 0;
// int l_n_100 = 0;
// for (int j = 0; j < k; j++) {
// if (idmap_knng[i][j] == gt_nn) {
// if (j < 1){
// n_1++;
// l_n_1++;
// }
// if (j < 10){
// n_10++;
// l_n_10++;
// }
// if (j < 100){
// n_100++;
// l_n_100++;
// }
//
// }
// if ((j == k-1) && (l_n_100 == 0)){
// std::cout << "error id: " << i << std::endl;
// }
// }
// }
// printf("R@1 = %.4f\n", n_1 / float(rows));
// printf("R@10 = %.4f\n", n_10 / float(rows));
// printf("R@100 = %.4f\n", n_100 / float(rows));
//}

69
core/src/index/unittest/utils.cpp
View File

@ -178,3 +178,72 @@ PrintResult(const knowhere::DatasetPtr& result, const int& nq, const int& k) {
std::cout << "id\n" << ss_id.str() << std::endl;
std::cout << "dist\n" << ss_dist.str() << std::endl;
}
void
Load_nns_graph(std::vector<std::vector<int64_t>>& final_graph, const char* filename) {
std::vector<std::vector<unsigned>> knng;
std::ifstream in(filename, std::ios::binary);
unsigned k;
in.read((char*)&k, sizeof(unsigned));
in.seekg(0, std::ios::end);
std::ios::pos_type ss = in.tellg();
size_t fsize = (size_t)ss;
size_t num = (size_t)(fsize / (k + 1) / 4);
in.seekg(0, std::ios::beg);
knng.resize(num);
knng.reserve(num);
int64_t kk = (k + 3) / 4 * 4;
for (size_t i = 0; i < num; i++) {
in.seekg(4, std::ios::cur);
knng[i].resize(k);
knng[i].reserve(kk);
in.read((char*)knng[i].data(), k * sizeof(unsigned));
}
in.close();
final_graph.resize(knng.size());
for (int i = 0; i < knng.size(); ++i) {
final_graph[i].resize(knng[i].size());
for (int j = 0; j < knng[i].size(); ++j) {
final_graph[i][j] = knng[i][j];
}
}
}
float*
fvecs_read(const char* fname, size_t* d_out, size_t* n_out) {
FILE* f = fopen(fname, "r");
if (!f) {
fprintf(stderr, "could not open %s\n", fname);
perror("");
abort();
}
int d;
fread(&d, 1, sizeof(int), f);
assert((d > 0 && d < 1000000) || !"unreasonable dimension");
fseek(f, 0, SEEK_SET);
struct stat st;
fstat(fileno(f), &st);
size_t sz = st.st_size;
assert(sz % ((d + 1) * 4) == 0 || !"weird file size");
size_t n = sz / ((d + 1) * 4);
*d_out = d;
*n_out = n;
float* x = new float[n * (d + 1)];
size_t nr = fread(x, sizeof(float), n * (d + 1), f);
assert(nr == n * (d + 1) || !"could not read whole file");
// shift array to remove row headers
for (size_t i = 0; i < n; i++) memmove(x + i * d, x + 1 + i * (d + 1), d * sizeof(*x));
fclose(f);
return x;
}
int* // not very clean, but works as long as sizeof(int) == sizeof(float)
ivecs_read(const char* fname, size_t* d_out, size_t* n_out) {
return (int*)fvecs_read(fname, d_out, n_out);
}

9
core/src/index/unittest/utils.h
View File

@ -93,3 +93,12 @@ struct FileIOReader {
size_t
operator()(void* ptr, size_t size);
};
void
Load_nns_graph(std::vector<std::vector<int64_t>>& final_graph_, const char* filename);
float*
fvecs_read(const char* fname, size_t* d_out, size_t* n_out);
int*
ivecs_read(const char* fname, size_t* d_out, size_t* n_out);

21
core/src/scheduler/task/BuildIndexTask.cpp
View File

@ -168,21 +168,28 @@ XBuildIndexTask::Execute() {
// step 5: save index file
try {
index->Serialize();
status = index->Serialize();
if (!status.ok()) {
ENGINE_LOG_ERROR << status.message();
}
} catch (std::exception& ex) {
// typical error: out of disk space or permition denied
std::string msg = "Serialize index encounter exception: " + std::string(ex.what());
ENGINE_LOG_ERROR << msg;
status = Status(DB_ERROR, msg);
}
if (!status.ok()) {
// if failed to serialize index file to disk
// typical error: out of disk space, out of memory or permition denied
table_file.file_type_ = engine::meta::TableFileSchema::TO_DELETE;
status = meta_ptr->UpdateTableFile(table_file);
ENGINE_LOG_DEBUG << "Failed to update file to index, mark file: " << table_file.file_id_ << " to to_delete";
ENGINE_LOG_ERROR << "Failed to persist index file: " << table_file.location_
<< ", possible out of disk space";
<< ", possible out of disk space or memory";
build_index_job->BuildIndexDone(to_index_id_);
build_index_job->GetStatus() = Status(DB_ERROR, msg);
build_index_job->GetStatus() = status;
to_index_engine_ = nullptr;
return;
}
@ -196,7 +203,11 @@ XBuildIndexTask::Execute() {
origin_file.file_type_ = engine::meta::TableFileSchema::BACKUP;
engine::meta::TableFilesSchema update_files = {table_file, origin_file};
status = meta_ptr->UpdateTableFiles(update_files);
if (status.ok()) { // makesure index file is sucessfully serialized to disk
status = meta_ptr->UpdateTableFiles(update_files);
}
if (status.ok()) {
ENGINE_LOG_DEBUG << "New index file " << table_file.file_id_ << " of size " << index->PhysicalSize()
<< " bytes"

16
core/src/sdk/examples/partition/src/ClientTest.cpp
View File

@ -141,18 +141,18 @@ ClientTest::Test(const std::string& address, const std::string& port) {
{ // search vectors
std::cout << "Search in correct partition" << std::endl;
std::vector<std::string> partiton_tags = {std::to_string(TARGET_PARTITION)};
std::vector<std::string> partition_tags = {std::to_string(TARGET_PARTITION)};
milvus::TopKQueryResult topk_query_result;
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partiton_tags, TOP_K, NPROBE, search_record_array,
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partition_tags, TOP_K, NPROBE, search_record_array,
topk_query_result);
std::cout << "Search in wrong partition" << std::endl;
partiton_tags = {"0"};
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partiton_tags, TOP_K, NPROBE, search_record_array,
partition_tags = {"0"};
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partition_tags, TOP_K, NPROBE, search_record_array,
topk_query_result);
std::cout << "Search by regex matched partition tag" << std::endl;
partiton_tags = {"\\d"};
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partiton_tags, TOP_K, NPROBE, search_record_array,
partition_tags = {"\\d"};
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partition_tags, TOP_K, NPROBE, search_record_array,
topk_query_result);
}
@ -191,9 +191,9 @@ ClientTest::Test(const std::string& address, const std::string& port) {
{ // search vectors
std::cout << "Search in whole table" << std::endl;
std::vector<std::string> partiton_tags;
std::vector<std::string> partition_tags;
milvus::TopKQueryResult topk_query_result;
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partiton_tags, TOP_K, NPROBE, search_record_array,
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partition_tags, TOP_K, NPROBE, search_record_array,
topk_query_result);
}

8
core/src/sdk/examples/simple/src/ClientTest.cpp
View File

@ -143,9 +143,9 @@ ClientTest::Test(const std::string& address, const std::string& port) {
milvus_sdk::Utils::Sleep(3);
{ // search vectors
std::vector<std::string> partiton_tags;
std::vector<std::string> partition_tags;
milvus::TopKQueryResult topk_query_result;
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partiton_tags, TOP_K, NPROBE, search_record_array,
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partition_tags, TOP_K, NPROBE, search_record_array,
topk_query_result);
}
@ -169,9 +169,9 @@ ClientTest::Test(const std::string& address, const std::string& port) {
}
{ // search vectors
std::vector<std::string> partiton_tags;
std::vector<std::string> partition_tags;
milvus::TopKQueryResult topk_query_result;
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partiton_tags, TOP_K, NPROBE, search_record_array,
milvus_sdk::Utils::DoSearch(conn, TABLE_NAME, partition_tags, TOP_K, NPROBE, search_record_array,
topk_query_result);
}

4
core/src/sdk/examples/utils/Utils.cpp
View File

@ -202,7 +202,7 @@ Utils::CheckSearchResult(const std::vector<std::pair<int64_t, milvus::RowRecord>
void
Utils::DoSearch(std::shared_ptr<milvus::Connection> conn, const std::string& table_name,
const std::vector<std::string>& partiton_tags, int64_t top_k, int64_t nprobe,
const std::vector<std::string>& partition_tags, int64_t top_k, int64_t nprobe,
const std::vector<std::pair<int64_t, milvus::RowRecord>>& search_record_array,
milvus::TopKQueryResult& topk_query_result) {
topk_query_result.clear();
@ -222,7 +222,7 @@ Utils::DoSearch(std::shared_ptr<milvus::Connection> conn, const std::string& tab
BLOCK_SPLITER
milvus_sdk::TimeRecorder rc("search");
milvus::Status stat =
conn->Search(table_name, partiton_tags, record_array, query_range_array, top_k, nprobe, topk_query_result);
conn->Search(table_name, partition_tags, record_array, query_range_array, top_k, nprobe, topk_query_result);
std::cout << "SearchVector function call status: " << stat.message() << std::endl;
BLOCK_SPLITER
}

2
core/src/sdk/examples/utils/Utils.h
View File

@ -69,7 +69,7 @@ class Utils {
static void
DoSearch(std::shared_ptr<milvus::Connection> conn, const std::string& table_name,
const std::vector<std::string>& partiton_tags, int64_t top_k, int64_t nprobe,
const std::vector<std::string>& partition_tags, int64_t top_k, int64_t nprobe,
const std::vector<std::pair<int64_t, milvus::RowRecord>>& search_record_array,
milvus::TopKQueryResult& topk_query_result);
};

4
core/src/sdk/grpc/ClientProxy.cpp
View File

@ -221,7 +221,7 @@ ClientProxy::Insert(const std::string& table_name, const std::string& partition_
}
Status
ClientProxy::Search(const std::string& table_name, const std::vector<std::string>& partiton_tags,
ClientProxy::Search(const std::string& table_name, const std::vector<std::string>& partition_tags,
const std::vector<RowRecord>& query_record_array, const std::vector<Range>& query_range_array,
int64_t topk, int64_t nprobe, TopKQueryResult& topk_query_result) {
try {
@ -230,7 +230,7 @@ ClientProxy::Search(const std::string& table_name, const std::vector<std::string
search_param.set_table_name(table_name);
search_param.set_topk(topk);
search_param.set_nprobe(nprobe);
for (auto& tag : partiton_tags) {
for (auto& tag : partition_tags) {
search_param.add_partition_tag_array(tag);
}
for (auto& record : query_record_array) {

2
core/src/sdk/grpc/ClientProxy.h
View File

@ -58,7 +58,7 @@ class ClientProxy : public Connection {
std::vector<int64_t>& id_array) override;
Status
Search(const std::string& table_name, const std::vector<std::string>& partiton_tags,
Search(const std::string& table_name, const std::vector<std::string>& partition_tags,
const std::vector<RowRecord>& query_record_array, const std::vector<Range>& query_range_array, int64_t topk,
int64_t nprobe, TopKQueryResult& topk_query_result) override;

2
core/src/sdk/include/MilvusApi.h
View File

@ -273,7 +273,7 @@ class Connection {
* @return Indicate if query is successful.
*/
virtual Status
Search(const std::string& table_name, const std::vector<std::string>& partiton_tags,
Search(const std::string& table_name, const std::vector<std::string>& partition_tags,
const std::vector<RowRecord>& query_record_array, const std::vector<Range>& query_range_array, int64_t topk,
int64_t nprobe, TopKQueryResult& topk_query_result) = 0;

4
core/src/sdk/interface/ConnectionImpl.cpp
View File

@ -89,10 +89,10 @@ ConnectionImpl::Insert(const std::string& table_name, const std::string& partiti
}
Status
ConnectionImpl::Search(const std::string& table_name, const std::vector<std::string>& partiton_tags,
ConnectionImpl::Search(const std::string& table_name, const std::vector<std::string>& partition_tags,
const std::vector<RowRecord>& query_record_array, const std::vector<Range>& query_range_array,
int64_t topk, int64_t nprobe, TopKQueryResult& topk_query_result) {
return client_proxy_->Search(table_name, partiton_tags, query_record_array, query_range_array, topk, nprobe,
return client_proxy_->Search(table_name, partition_tags, query_record_array, query_range_array, topk, nprobe,
topk_query_result);
}

2
core/src/sdk/interface/ConnectionImpl.h
View File

@ -60,7 +60,7 @@ class ConnectionImpl : public Connection {
std::vector<int64_t>& id_array) override;
Status
Search(const std::string& table_name, const std::vector<std::string>& partiton_tags,
Search(const std::string& table_name, const std::vector<std::string>& partition_tags,
const std::vector<RowRecord>& query_record_array, const std::vector<Range>& query_range_array, int64_t topk,
int64_t nprobe, TopKQueryResult& topk_query_result) override;

11
core/src/server/grpc_impl/request/CmdRequest.cpp
View File

@ -17,6 +17,8 @@
#include "server/grpc_impl/request/CmdRequest.h"
#include "scheduler/SchedInst.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include <memory>
@ -35,10 +37,19 @@ CmdRequest::Create(const std::string& cmd, std::string& result) {
Status
CmdRequest::OnExecute() {
std::string hdr = "CmdRequest(cmd=" + cmd_ + ")";
TimeRecorderAuto rc(hdr);
if (cmd_ == "version") {
result_ = MILVUS_VERSION;
} else if (cmd_ == "tasktable") {
result_ = scheduler::ResMgrInst::GetInstance()->DumpTaskTables();
} else if (cmd_ == "mode") {
#ifdef MILVUS_GPU_VERSION
result_ = "GPU";
#else
result_ = "CPU";
#endif
} else {
result_ = "OK";
}

5
core/src/server/grpc_impl/request/CountTableRequest.cpp
View File

@ -39,7 +39,8 @@ CountTableRequest::Create(const std::string& table_name, int64_t& row_count) {
Status
CountTableRequest::OnExecute() {
try {
TimeRecorder rc("CountTableRequest");
std::string hdr = "CountTableRequest(table=" + table_name_ + ")";
TimeRecorderAuto rc(hdr);
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
@ -59,8 +60,6 @@ CountTableRequest::OnExecute() {
}
row_count_ = static_cast<int64_t>(row_count);
rc.ElapseFromBegin("total cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}

5
core/src/server/grpc_impl/request/CreateIndexRequest.cpp
View File

@ -44,7 +44,8 @@ CreateIndexRequest::Create(const ::milvus::grpc::IndexParam* index_param) {
Status
CreateIndexRequest::OnExecute() {
try {
TimeRecorder rc("CreateIndexRequest");
std::string hdr = "CreateIndexRequest(table=" + index_param_->table_name() + ")";
TimeRecorderAuto rc(hdr);
// step 1: check arguments
std::string table_name_ = index_param_->table_name();
@ -82,8 +83,6 @@ CreateIndexRequest::OnExecute() {
if (!status.ok()) {
return status;
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}

8
core/src/server/grpc_impl/request/CreatePartitionRequest.cpp
View File

@ -22,6 +22,7 @@
#include "utils/ValidationUtil.h"
#include <memory>
#include <string>
namespace milvus {
namespace server {
@ -42,7 +43,10 @@ CreatePartitionRequest::Create(const ::milvus::grpc::PartitionParam* partition_p
Status
CreatePartitionRequest::OnExecute() {
TimeRecorder rc("CreatePartitionRequest");
std::string hdr = "CreatePartitionRequest(table=" + partition_param_->table_name() +
", partition_name=" + partition_param_->partition_name() +
", partition_tag=" + partition_param_->tag() + ")";
TimeRecorderAuto rc(hdr);
try {
// step 1: check arguments
@ -75,8 +79,6 @@ CreatePartitionRequest::OnExecute() {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
rc.ElapseFromBegin("totally cost");
return Status::OK();
}

7
core/src/server/grpc_impl/request/CreateTableRequest.cpp
View File

@ -22,6 +22,7 @@
#include "utils/ValidationUtil.h"
#include <memory>
#include <string>
namespace milvus {
namespace server {
@ -42,7 +43,9 @@ CreateTableRequest::Create(const ::milvus::grpc::TableSchema* schema) {
Status
CreateTableRequest::OnExecute() {
TimeRecorder rc("CreateTableRequest");
std::string hdr = "CreateTableRequest(table=" + schema_->table_name() +
", dimension=" + std::to_string(schema_->dimension()) + ")";
TimeRecorderAuto rc(hdr);
try {
// step 1: check arguments
@ -86,8 +89,6 @@ CreateTableRequest::OnExecute() {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
rc.ElapseFromBegin("totally cost");
return Status::OK();
}

4
core/src/server/grpc_impl/request/DeleteByDateRequest.cpp
View File

@ -46,7 +46,7 @@ DeleteByDateRequest::Create(const ::milvus::grpc::DeleteByDateParam* delete_by_r
Status
DeleteByDateRequest::OnExecute() {
try {
TimeRecorder rc("DeleteByDateRequest");
TimeRecorderAuto rc("DeleteByDateRequest");
// step 1: check arguments
std::string table_name = delete_by_range_param_->table_name();
@ -67,7 +67,7 @@ DeleteByDateRequest::OnExecute() {
}
}
rc.ElapseFromBegin("check validation");
rc.RecordSection("check validation");
// step 3: check date range, and convert to db dates
std::vector<DB_DATE> dates;

5
core/src/server/grpc_impl/request/DescribeIndexRequest.cpp
View File

@ -39,7 +39,8 @@ DescribeIndexRequest::Create(const std::string& table_name, ::milvus::grpc::Inde
Status
DescribeIndexRequest::OnExecute() {
try {
TimeRecorder rc("DescribeIndexRequest");
std::string hdr = "DescribeIndexRequest(table=" + table_name_ + ")";
TimeRecorderAuto rc(hdr);
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
@ -57,8 +58,6 @@ DescribeIndexRequest::OnExecute() {
index_param_->set_table_name(table_name_);
index_param_->mutable_index()->set_index_type(index.engine_type_);
index_param_->mutable_index()->set_nlist(index.nlist_);
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}

5
core/src/server/grpc_impl/request/DescribeTableRequest.cpp
View File

@ -38,7 +38,8 @@ DescribeTableRequest::Create(const std::string& table_name, ::milvus::grpc::Tabl
Status
DescribeTableRequest::OnExecute() {
TimeRecorder rc("DescribeTableRequest");
std::string hdr = "DescribeTableRequest(table=" + table_name_ + ")";
TimeRecorderAuto rc(hdr);
try {
// step 1: check arguments
@ -63,8 +64,6 @@ DescribeTableRequest::OnExecute() {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
rc.ElapseFromBegin("totally cost");
return Status::OK();
}

5
core/src/server/grpc_impl/request/DropIndexRequest.cpp
View File

@ -39,7 +39,8 @@ DropIndexRequest::Create(const std::string& table_name) {
Status
DropIndexRequest::OnExecute() {
try {
TimeRecorder rc("DropIndexRequest");
std::string hdr = "DropIndexRequest(table=" + table_name_ + ")";
TimeRecorderAuto rc(hdr);
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
@ -62,8 +63,6 @@ DropIndexRequest::OnExecute() {
if (!status.ok()) {
return status;
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}

5
core/src/server/grpc_impl/request/DropPartitionRequest.cpp
View File

@ -39,6 +39,11 @@ DropPartitionRequest::Create(const ::milvus::grpc::PartitionParam* partition_par
Status
DropPartitionRequest::OnExecute() {
std::string hdr = "DropPartitionRequest(table=" + partition_param_->table_name() +
", partition_name=" + partition_param_->partition_name() +
", partition_tag=" + partition_param_->tag() + ")";
TimeRecorderAuto rc(hdr);
std::string table_name = partition_param_->table_name();
std::string partition_name = partition_param_->partition_name();
std::string partition_tag = partition_param_->tag();

5
core/src/server/grpc_impl/request/DropTableRequest.cpp
View File

@ -40,7 +40,8 @@ DropTableRequest::Create(const std::string& table_name) {
Status
DropTableRequest::OnExecute() {
try {
TimeRecorder rc("DropTableRequest");
std::string hdr = "DropTableRequest(table=" + table_name_ + ")";
TimeRecorder rc(hdr);
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
@ -60,7 +61,7 @@ DropTableRequest::OnExecute() {
}
}
rc.ElapseFromBegin("check validation");
rc.RecordSection("check validation");
// step 3: Drop table
std::vector<DB_DATE> dates;

5
core/src/server/grpc_impl/request/HasTableRequest.cpp
View File

@ -39,7 +39,8 @@ HasTableRequest::Create(const std::string& table_name, bool& has_table) {
Status
HasTableRequest::OnExecute() {
try {
TimeRecorder rc("HasTableRequest");
std::string hdr = "HasTableRequest(table=" + table_name_ + ")";
TimeRecorderAuto rc(hdr);
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
@ -52,8 +53,6 @@ HasTableRequest::OnExecute() {
if (!status.ok()) {
return status;
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}

9
core/src/server/grpc_impl/request/InsertRequest.cpp
View File

@ -45,7 +45,10 @@ InsertRequest::Create(const ::milvus::grpc::InsertParam* insert_param, ::milvus:
Status
InsertRequest::OnExecute() {
try {
TimeRecorder rc("InsertRequest");
std::string hdr = "InsertRequest(table=" + insert_param_->table_name() +
", n=" + std::to_string(insert_param_->row_record_array_size()) +
", partition_tag=" + insert_param_->partition_tag() + ")";
TimeRecorder rc(hdr);
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(insert_param_->table_name());
@ -119,8 +122,6 @@ InsertRequest::OnExecute() {
table_info.dimension_ * sizeof(float));
}
rc.ElapseFromBegin("prepare vectors data");
// step 5: insert vectors
auto vec_count = static_cast<uint64_t>(insert_param_->row_record_array_size());
std::vector<int64_t> vec_ids(insert_param_->row_id_array_size(), 0);
@ -130,9 +131,9 @@ InsertRequest::OnExecute() {
memcpy(target_data, src_data, static_cast<size_t>(sizeof(int64_t) * insert_param_->row_id_array_size()));
}
rc.RecordSection("prepare vectors data");
status = DBWrapper::DB()->InsertVectors(insert_param_->table_name(), insert_param_->partition_tag(), vec_count,
vec_f.data(), vec_ids);
rc.ElapseFromBegin("add vectors to engine");
if (!status.ok()) {
return status;
}

5
core/src/server/grpc_impl/request/PreloadTableRequest.cpp
View File

@ -39,7 +39,8 @@ PreloadTableRequest::Create(const std::string& table_name) {
Status
PreloadTableRequest::OnExecute() {
try {
TimeRecorder rc("PreloadTableRequest");
std::string hdr = "PreloadTableRequest(table=" + table_name_ + ")";
TimeRecorderAuto rc(hdr);
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
@ -52,8 +53,6 @@ PreloadTableRequest::OnExecute() {
if (!status.ok()) {
return status;
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}

4
core/src/server/grpc_impl/request/SearchRequest.cpp
View File

@ -51,7 +51,9 @@ SearchRequest::OnExecute() {
int64_t top_k = search_param_->topk();
int64_t nprobe = search_param_->nprobe();
std::string hdr = "SearchRequest(k=" + std::to_string(top_k) + ", nprob=" + std::to_string(nprobe) + ")";
std::string hdr = "SearchRequest(table=" + search_param_->table_name() +
", nq=" + std::to_string(search_param_->query_record_array_size()) +
", k=" + std::to_string(top_k) + ", nprob=" + std::to_string(nprobe) + ")";
TimeRecorder rc(hdr);
// step 1: check table name

3
core/src/server/grpc_impl/request/ShowPartitionsRequest.cpp
View File

@ -40,6 +40,9 @@ ShowPartitionsRequest::Create(const std::string& table_name, ::milvus::grpc::Par
Status
ShowPartitionsRequest::OnExecute() {
std::string hdr = "ShowPartitionsRequest(table=" + table_name_ + ")";
TimeRecorderAuto rc(hdr);
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;

2
core/src/server/grpc_impl/request/ShowTablesRequest.cpp
View File

@ -38,6 +38,8 @@ ShowTablesRequest::Create(::milvus::grpc::TableNameList* table_name_list) {
Status
ShowTablesRequest::OnExecute() {
TimeRecorderAuto rc("ShowTablesRequest");
std::vector<engine::meta::TableSchema> schema_array;
auto statuts = DBWrapper::DB()->AllTables(schema_array);
if (!statuts.ok()) {

2
core/src/utils/CommonUtil.cpp
View File

@ -229,7 +229,7 @@ CommonUtil::ConvertTime(tm time_struct, time_t& time_integer) {
void
CommonUtil::EraseFromCache(const std::string& item_key) {
if (item_key.empty()) {
// SERVER_LOG_ERROR << "Empty key cannot be erased from cache";
SERVER_LOG_ERROR << "Empty key cannot be erased from cache";
return;
}

7
core/src/wrapper/ConfAdapter.cpp
View File

@ -204,10 +204,11 @@ NSGConfAdapter::Match(const TempMetaConf& metaconf) {
auto scale_factor = round(metaconf.dim / 128.0);
scale_factor = scale_factor >= 4 ? 4 : scale_factor;
conf->nprobe = int64_t(conf->nlist * 0.01);
conf->knng = 40 + 10 * scale_factor; // the size of knng
conf->search_length = 40 + 5 * scale_factor;
// conf->knng = 40 + 10 * scale_factor; // the size of knng
conf->knng = 50;
conf->search_length = 50 + 5 * scale_factor;
conf->out_degree = 50 + 5 * scale_factor;
conf->candidate_pool_size = 200 + 100 * scale_factor;
conf->candidate_pool_size = 300;
MatchBase(conf);
return conf;
}

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

@ -536,12 +536,12 @@ TEST_F(DBTest, PARTITION_TEST) {
stat = db_->CreatePartition(table_name, "", "0");
ASSERT_FALSE(stat.ok());
std::vector<milvus::engine::meta::TableSchema> partiton_schema_array;
stat = db_->ShowPartitions(table_name, partiton_schema_array);
std::vector<milvus::engine::meta::TableSchema> partition_schema_array;
stat = db_->ShowPartitions(table_name, partition_schema_array);
ASSERT_TRUE(stat.ok());
ASSERT_EQ(partiton_schema_array.size(), PARTITION_COUNT);
ASSERT_EQ(partition_schema_array.size(), PARTITION_COUNT);
for (int64_t i = 0; i < PARTITION_COUNT; i++) {
ASSERT_EQ(partiton_schema_array[i].table_id_, table_name + "_" + std::to_string(i));
ASSERT_EQ(partition_schema_array[i].table_id_, table_name + "_" + std::to_string(i));
}
{ // build index

8
core/unittest/db/test_db_mysql.cpp
View File

@ -323,12 +323,12 @@ TEST_F(MySqlDBTest, PARTITION_TEST) {
stat = db_->CreatePartition(table_name, "", "0");
ASSERT_FALSE(stat.ok());
std::vector<milvus::engine::meta::TableSchema> partiton_schema_array;
stat = db_->ShowPartitions(table_name, partiton_schema_array);
std::vector<milvus::engine::meta::TableSchema> partition_schema_array;
stat = db_->ShowPartitions(table_name, partition_schema_array);
ASSERT_TRUE(stat.ok());
ASSERT_EQ(partiton_schema_array.size(), PARTITION_COUNT);
ASSERT_EQ(partition_schema_array.size(), PARTITION_COUNT);
for (int64_t i = 0; i < PARTITION_COUNT; i++) {
ASSERT_EQ(partiton_schema_array[i].table_id_, table_name + "_" + std::to_string(i));
ASSERT_EQ(partition_schema_array[i].table_id_, table_name + "_" + std::to_string(i));
}
{ // build index

60
core/unittest/db/test_misc.cpp
View File

@ -15,6 +15,8 @@
// specific language governing permissions and limitations
// under the License.
#include "db/IndexFailedChecker.h"
#include "db/OngoingFileChecker.h"
#include "db/Options.h"
#include "db/Utils.h"
#include "db/engine/EngineFactory.h"
@ -119,3 +121,61 @@ TEST(DBMiscTest, UTILS_TEST) {
status = milvus::engine::utils::DeleteTableFilePath(options, file);
ASSERT_TRUE(status.ok());
}
TEST(DBMiscTest, CHECKER_TEST) {
{
milvus::engine::IndexFailedChecker checker;
milvus::engine::meta::TableFileSchema schema;
schema.table_id_ = "aaa";
schema.file_id_ = "5000";
checker.MarkFailedIndexFile(schema);
schema.table_id_ = "bbb";
schema.file_id_ = "5001";
checker.MarkFailedIndexFile(schema);
std::vector<std::string> failed_files;
checker.GetFailedIndexFileOfTable("aaa", failed_files);
ASSERT_EQ(failed_files.size(), 1UL);
schema.table_id_ = "bbb";
schema.file_id_ = "5002";
checker.MarkFailedIndexFile(schema);
checker.MarkFailedIndexFile(schema);
milvus::engine::meta::TableFilesSchema table_files = {schema};
checker.IgnoreFailedIndexFiles(table_files);
ASSERT_TRUE(table_files.empty());
checker.GetFailedIndexFileOfTable("bbb", failed_files);
ASSERT_EQ(failed_files.size(), 2UL);
checker.MarkSucceedIndexFile(schema);
checker.GetFailedIndexFileOfTable("bbb", failed_files);
ASSERT_EQ(failed_files.size(), 1UL);
}
{
milvus::engine::OngoingFileChecker checker;
milvus::engine::meta::TableFileSchema schema;
schema.table_id_ = "aaa";
schema.file_id_ = "5000";
checker.MarkOngoingFile(schema);
ASSERT_TRUE(checker.IsIgnored(schema));
schema.table_id_ = "bbb";
schema.file_id_ = "5001";
milvus::engine::meta::TableFilesSchema table_files = {schema};
checker.MarkOngoingFiles(table_files);
ASSERT_TRUE(checker.IsIgnored(schema));
checker.UnmarkOngoingFile(schema);
ASSERT_FALSE(checker.IsIgnored(schema));
schema.table_id_ = "aaa";
schema.file_id_ = "5000";
checker.UnmarkOngoingFile(schema);
ASSERT_FALSE(checker.IsIgnored(schema));
}
}

106
install.md
View File

@ -1,6 +1,13 @@
# Install Milvus from Source Code
## Software requirements
- [Build from source](#build-from-source)
- [Compile Milvus on Docker](#compile-milvus-on-docker)
If you encounter any problems/issues compiling Milvus from source, please refer to [Troubleshooting](#troubleshooting).
## Build from source
### Requirements
- Ubuntu 18.04 or higher
@ -8,21 +15,21 @@
- CMake 3.12 or higher
##### For GPU version, you will also need:
##### For GPU-enabled version, you will also need:
- CUDA 10.0 or higher
- NVIDIA driver 418 or higher
## Compilation
### Compilation
### Step 1 Install dependencies
#### Step 1 Install dependencies
```shell
$ cd [Milvus root path]/core
$ ./ubuntu_build_deps.sh
```
### Step 2 Build
#### Step 2 Build
```shell
$ cd [Milvus root path]/core
@ -31,7 +38,7 @@ or
$ ./build.sh -t Release
```
By default, it will build CPU version. To build GPU version, add `-g` option
By default, it will build CPU-only version. To build GPU version, add `-g` option
```shell
$ ./build.sh -g
```
@ -43,7 +50,7 @@ $./build.sh -h
When the build is completed, all the stuff that you need in order to run Milvus will be installed under `[Milvus root path]/core/milvus`.
## Launch Milvus server
### Launch Milvus server
```shell
$ cd [Milvus root path]/core/milvus
@ -68,14 +75,87 @@ To stop Milvus server, run:
$ ./stop_server.sh
```
## Compile Milvus on Docker
With the following Docker images, you should be able to compile Milvus on any Linux platform that run Docker. To build a GPU supported Milvus, you neeed to install [NVIDIA Docker](https://github.com/NVIDIA/nvidia-docker/) first.
### Step 1 Pull Milvus Docker images
Pull CPU-only image:
```shell
$ docker pull milvusdb/milvus-cpu-build-env:v0.6.0-ubuntu18.04
```
Pull GPU-enabled image:
```shell
$ docker pull milvusdb/milvus-gpu-build-env:v0.6.0-ubuntu18.04
```
### Step 2 Start the Docker container
Start a CPU-only container:
```shell
$ docker run -it -p 19530:19530 -d milvusdb/milvus-cpu-build-env:v0.6.0-ubuntu18.04
```
Start a GPU container:
```shell
$ docker run — runtime=nvidia -it -p 19530:19530 -d milvusdb/milvus-gpu-build-env:v0.6.0-ubuntu18.04
```
To enter the container:
```shell
$ docker exec -it [container_id] bash
```
### Step 3 Download Milvus source code
Download Milvus source code:
```shell
$ cd /home
$ wget https://github.com/milvus-io/milvus/archive/0.6.0.tar.gz
```
Extract the source package:
```shell
$ tar xvf ./v0.6.0.tar.gz
```
The source code is extracted into a folder called `milvus-0.6.0`. To enter its core directory:
```shell
$ cd ./milvus-0.6.0/core
```
### Step 4 Compile Milvus in the container
If you are using a CPU-only image, compile it like this:
```shell
$ ./build.sh -t Release
```
If you are using a GPU-enabled image, you need to add a `-g` parameter:
```shell
$ ./build.sh -g -t Release
```
Then start Milvus server：
```shell
$ ./start_server.sh
```
## Troubleshooting
1. If you encounter the following error when compiling:
`protocol https not supported or disabled in libcurl`.
First, make sure you have `libcurl4-openssl-dev` installed in your system.
Then try reinstall CMake from source with `--system-curl` option:
```shell
$ ./bootstrap --system-curl
$ make
$ sudo make install
```
Then try reinstalling the latest CMake from source with `--system-curl` option:
```shell
$ ./bootstrap --system-curl
$ make
$ sudo make install
```
If the `--system-curl` command doesn't work, you can also reinstall CMake in **Ubuntu Software** on your local computer.

2
tests/milvus_python_test/test_add_vectors.py
View File

@ -31,6 +31,8 @@ class TestAddBase:
if "internal" not in args:
if request.param["index_type"] == IndexType.IVF_SQ8H:
pytest.skip("sq8h not support in open source")
if request.param["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip PQ Temporary")
return request.param
def test_add_vector_create_table(self, connect, table):

299
tests/milvus_python_test/test_index.py
View File

@ -14,7 +14,7 @@ from utils import *
nb = 10000
dim = 128
index_file_size = 10
index_file_size = 20
vectors = gen_vectors(nb, dim)
vectors = sklearn.preprocessing.normalize(vectors, axis=1, norm='l2')
vectors = vectors.tolist()
@ -63,6 +63,18 @@ class TestIndexBase:
status = connect.create_index(table, index_params)
assert status.OK()
@pytest.mark.timeout(BUILD_TIMEOUT)
def test_create_index_no_vectors(self, connect, table, get_simple_index_params):
'''
target: test create index interface
method: create table and add vectors in it, create index
expected: return code equals to 0, and search success
'''
index_params = get_simple_index_params
logging.getLogger().info(index_params)
status = connect.create_index(table, index_params)
assert status.OK()
@pytest.mark.timeout(BUILD_TIMEOUT)
def test_create_index_partition(self, connect, table, get_simple_index_params):
'''
@ -72,6 +84,8 @@ class TestIndexBase:
'''
partition_name = gen_unique_str()
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
logging.getLogger().info(index_params)
status = connect.create_partition(table, partition_name, tag)
status, ids = connect.add_vectors(table, vectors, partition_tag=tag)
@ -242,6 +256,8 @@ class TestIndexBase:
expected: return code equals to 0
'''
index_param = get_simple_index_params
if index_param["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
status = connect.create_index(table, index_param)
status, ids = connect.add_vectors(table, vectors)
assert status.OK()
@ -255,6 +271,8 @@ class TestIndexBase:
'''
status, ids = connect.add_vectors(table, vectors)
index_param = get_simple_index_params
if index_param["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
status = connect.create_index(table, index_param)
status = connect.create_index(table, index_param)
assert status.OK()
@ -291,15 +309,15 @@ class TestIndexBase:
******************************************************************
"""
def test_describe_index(self, connect, table, get_simple_index_params):
def test_describe_index(self, connect, table, get_index_params):
'''
target: test describe index interface
method: create table and add vectors in it, create index, call describe index
expected: return code 0, and index instructure
'''
index_params = get_simple_index_params
index_params = get_index_params
logging.getLogger().info(index_params)
status, ids = connect.add_vectors(table, vectors)
# status, ids = connect.add_vectors(table, vectors)
status = connect.create_index(table, index_params)
status, result = connect.describe_index(table)
logging.getLogger().info(result)
@ -325,6 +343,8 @@ class TestIndexBase:
'metric_type': MetricType.L2}
connect.create_table(param)
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
logging.getLogger().info(index_params)
status, ids = connect.add_vectors(table_name=table_name, records=vectors)
status = connect.create_index(table_name, index_params)
@ -405,7 +425,7 @@ class TestIndexBase:
expected: return code 0, and default index param
'''
index_param = get_simple_index_params
status, ids = connect.add_vectors(table, vectors)
# status, ids = connect.add_vectors(table, vectors)
status = connect.create_index(table, index_param)
assert status.OK()
status, result = connect.describe_index(table)
@ -425,7 +445,7 @@ class TestIndexBase:
expected: return code 0
'''
index_param = get_simple_index_params
status, ids = connect.add_vectors(table, vectors)
# status, ids = connect.add_vectors(table, vectors)
status = connect.create_index(table, index_param)
assert status.OK()
status, result = connect.describe_index(table)
@ -494,10 +514,9 @@ class TestIndexBase:
expected: return code 0
'''
index_params = get_simple_index_params
status, ids = connect.add_vectors(table, vectors)
# status, ids = connect.add_vectors(table, vectors)
for i in range(2):
status = connect.create_index(table, index_params)
assert status.OK()
status, result = connect.describe_index(table)
logging.getLogger().info(result)
@ -517,7 +536,7 @@ class TestIndexBase:
'''
nlist = 16384
index_params = [{"index_type": IndexType.IVFLAT, "nlist": nlist}, {"index_type": IndexType.IVF_SQ8, "nlist": nlist}]
status, ids = connect.add_vectors(table, vectors)
# status, ids = connect.add_vectors(table, vectors)
for i in range(2):
status = connect.create_index(table, index_params[i])
assert status.OK()
@ -570,10 +589,7 @@ class TestIndexIP:
logging.getLogger().info(index_params)
status, ids = connect.add_vectors(ip_table, vectors)
status = connect.create_index(ip_table, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
assert status.OK()
@pytest.mark.timeout(BUILD_TIMEOUT)
def test_create_index_partition(self, connect, ip_table, get_simple_index_params):
@ -584,14 +600,13 @@ class TestIndexIP:
'''
partition_name = gen_unique_str()
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
logging.getLogger().info(index_params)
status = connect.create_partition(ip_table, partition_name, tag)
status, ids = connect.add_vectors(ip_table, vectors, partition_tag=tag)
status = connect.create_index(partition_name, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
assert status.OK()
@pytest.mark.level(2)
def test_create_index_without_connect(self, dis_connect, ip_table):
@ -616,17 +631,13 @@ class TestIndexIP:
logging.getLogger().info(index_params)
status, ids = connect.add_vectors(ip_table, vectors)
status = connect.create_index(ip_table, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
logging.getLogger().info(connect.describe_index(ip_table))
query_vecs = [vectors[0], vectors[1], vectors[2]]
top_k = 5
status, result = connect.search_vectors(ip_table, top_k, nprobe, query_vecs)
logging.getLogger().info(result)
assert status.OK()
assert len(result) == len(query_vecs)
logging.getLogger().info(connect.describe_index(ip_table))
query_vecs = [vectors[0], vectors[1], vectors[2]]
top_k = 5
status, result = connect.search_vectors(ip_table, top_k, nprobe, query_vecs)
logging.getLogger().info(result)
assert status.OK()
assert len(result) == len(query_vecs)
# TODO: enable
@pytest.mark.timeout(BUILD_TIMEOUT)
@ -734,6 +745,8 @@ class TestIndexIP:
expected: return code equals to 0
'''
index_param = get_simple_index_params
if index_param["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
status = connect.create_index(ip_table, index_param)
status, ids = connect.add_vectors(ip_table, vectors)
assert status.OK()
@ -792,7 +805,7 @@ class TestIndexIP:
'''
index_params = get_simple_index_params
logging.getLogger().info(index_params)
status, ids = connect.add_vectors(ip_table, vectors)
# status, ids = connect.add_vectors(ip_table, vectors[:5000])
status = connect.create_index(ip_table, index_params)
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
@ -808,6 +821,8 @@ class TestIndexIP:
'''
partition_name = gen_unique_str()
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
logging.getLogger().info(index_params)
status = connect.create_partition(ip_table, partition_name, tag)
status, ids = connect.add_vectors(ip_table, vectors, partition_tag=tag)
@ -831,6 +846,8 @@ class TestIndexIP:
'''
partition_name = gen_unique_str()
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
logging.getLogger().info(index_params)
status = connect.create_partition(ip_table, partition_name, tag)
status, ids = connect.add_vectors(ip_table, vectors, partition_tag=tag)
@ -856,6 +873,8 @@ class TestIndexIP:
new_partition_name = gen_unique_str()
new_tag = "new_tag"
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
logging.getLogger().info(index_params)
status = connect.create_partition(ip_table, partition_name, tag)
status = connect.create_partition(ip_table, new_partition_name, new_tag)
@ -892,6 +911,8 @@ class TestIndexIP:
'metric_type': MetricType.IP}
connect.create_table(param)
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
logging.getLogger().info(index_params)
status, ids = connect.add_vectors(table_name=table_name, records=vectors)
status = connect.create_index(table_name, index_params)
@ -944,28 +965,30 @@ class TestIndexIP:
******************************************************************
"""
def test_drop_index(self, connect, ip_table, get_index_params):
def test_drop_index(self, connect, ip_table, get_simple_index_params):
'''
target: test drop index interface
method: create table and add vectors in it, create index, call drop index
expected: return code 0, and default index param
'''
index_params = get_index_params
status, ids = connect.add_vectors(ip_table, vectors)
index_params = get_simple_index_params
status, mode = connect._cmd("mode")
assert status.OK()
# status, ids = connect.add_vectors(ip_table, vectors)
status = connect.create_index(ip_table, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
if str(mode) == "GPU" and index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
def test_drop_index_partition(self, connect, ip_table, get_simple_index_params):
'''
@ -975,22 +998,21 @@ class TestIndexIP:
'''
partition_name = gen_unique_str()
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
status = connect.create_partition(ip_table, partition_name, tag)
status, ids = connect.add_vectors(ip_table, vectors, partition_tag=tag)
status = connect.create_index(ip_table, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
def test_drop_index_partition_A(self, connect, ip_table, get_simple_index_params):
'''
@ -1000,25 +1022,24 @@ class TestIndexIP:
'''
partition_name = gen_unique_str()
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
status = connect.create_partition(ip_table, partition_name, tag)
status, ids = connect.add_vectors(ip_table, vectors, partition_tag=tag)
status = connect.create_index(partition_name, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
status, result = connect.describe_index(partition_name)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == partition_name
assert result._index_type == IndexType.FLAT
assert status.OK()
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
status, result = connect.describe_index(partition_name)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == partition_name
assert result._index_type == IndexType.FLAT
def test_drop_index_partition_B(self, connect, ip_table, get_simple_index_params):
'''
@ -1028,25 +1049,24 @@ class TestIndexIP:
'''
partition_name = gen_unique_str()
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
status = connect.create_partition(ip_table, partition_name, tag)
status, ids = connect.add_vectors(ip_table, vectors, partition_tag=tag)
status = connect.create_index(partition_name, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
status = connect.drop_index(partition_name)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
status, result = connect.describe_index(partition_name)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == partition_name
assert result._index_type == IndexType.FLAT
assert status.OK()
status = connect.drop_index(partition_name)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
status, result = connect.describe_index(partition_name)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == partition_name
assert result._index_type == IndexType.FLAT
def test_drop_index_partition_C(self, connect, ip_table, get_simple_index_params):
'''
@ -1058,31 +1078,30 @@ class TestIndexIP:
new_partition_name = gen_unique_str()
new_tag = "new_tag"
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
status = connect.create_partition(ip_table, partition_name, tag)
status = connect.create_partition(ip_table, new_partition_name, new_tag)
status, ids = connect.add_vectors(ip_table, vectors)
status = connect.create_index(ip_table, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
status = connect.drop_index(new_partition_name)
assert status.OK()
status, result = connect.describe_index(new_partition_name)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == new_partition_name
assert result._index_type == IndexType.FLAT
status, result = connect.describe_index(partition_name)
logging.getLogger().info(result)
assert result._nlist == index_params["nlist"]
assert result._table_name == partition_name
assert result._index_type == index_params["index_type"]
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == index_params["nlist"]
assert result._table_name == ip_table
assert result._index_type == index_params["index_type"]
assert status.OK()
status = connect.drop_index(new_partition_name)
assert status.OK()
status, result = connect.describe_index(new_partition_name)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == new_partition_name
assert result._index_type == IndexType.FLAT
status, result = connect.describe_index(partition_name)
logging.getLogger().info(result)
assert result._nlist == index_params["nlist"]
assert result._table_name == partition_name
assert result._index_type == index_params["index_type"]
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == index_params["nlist"]
assert result._table_name == ip_table
assert result._index_type == index_params["index_type"]
def test_drop_index_repeatly(self, connect, ip_table, get_simple_index_params):
'''
@ -1091,23 +1110,20 @@ class TestIndexIP:
expected: return code 0
'''
index_params = get_simple_index_params
status, ids = connect.add_vectors(ip_table, vectors)
# status, ids = connect.add_vectors(ip_table, vectors)
status = connect.create_index(ip_table, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
status = connect.drop_index(ip_table)
assert status.OK()
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
status = connect.drop_index(ip_table)
assert status.OK()
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
@pytest.mark.level(2)
def test_drop_index_without_connect(self, dis_connect, ip_table):
@ -1145,22 +1161,21 @@ class TestIndexIP:
expected: return code 0
'''
index_params = get_simple_index_params
if index_params["index_type"] == IndexType.IVF_PQ:
pytest.skip("Skip some PQ cases")
status, ids = connect.add_vectors(ip_table, vectors)
for i in range(2):
status = connect.create_index(ip_table, index_params)
if index_params["index_type"] == IndexType.IVF_PQ:
assert not status.OK()
else:
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
status = connect.drop_index(ip_table)
assert status.OK()
status, result = connect.describe_index(ip_table)
logging.getLogger().info(result)
assert result._nlist == 16384
assert result._table_name == ip_table
assert result._index_type == IndexType.FLAT
def test_create_drop_index_repeatly_different_index_params(self, connect, ip_table):
'''
@ -1200,7 +1215,7 @@ class TestIndexTableInvalid(object):
def get_table_name(self, request):
yield request.param
@pytest.mark.level(2)
@pytest.mark.level(1)
def test_create_index_with_invalid_tablename(self, connect, get_table_name):
table_name = get_table_name
nlist = 16384
@ -1208,13 +1223,13 @@ class TestIndexTableInvalid(object):
status = connect.create_index(table_name, index_param)
assert not status.OK()
@pytest.mark.level(2)
@pytest.mark.level(1)
def test_describe_index_with_invalid_tablename(self, connect, get_table_name):
table_name = get_table_name
status, result = connect.describe_index(table_name)
assert not status.OK()
@pytest.mark.level(2)
@pytest.mark.level(1)
def test_drop_index_with_invalid_tablename(self, connect, get_table_name):
table_name = get_table_name
status = connect.drop_index(table_name)
@ -1232,13 +1247,13 @@ class TestCreateIndexParamsInvalid(object):
def get_index_params(self, request):
yield request.param
@pytest.mark.level(2)
@pytest.mark.level(1)
def test_create_index_with_invalid_index_params(self, connect, table, get_index_params):
index_params = get_index_params
index_type = index_params["index_type"]
nlist = index_params["nlist"]
logging.getLogger().info(index_params)
status, ids = connect.add_vectors(table, vectors)
# status, ids = connect.add_vectors(table, vectors)
if (not index_type) or (not nlist) or (not isinstance(index_type, IndexType)) or (not isinstance(nlist, int)):
with pytest.raises(Exception) as e:
status = connect.create_index(table, index_params)

12
tests/milvus_python_test/test_search_vectors.py
View File

@ -48,6 +48,8 @@ class TestSearchBase:
if "internal" not in args:
if request.param["index_type"] == IndexType.IVF_SQ8H:
pytest.skip("sq8h not support in open source")
if request.param["index_type"] == IndexType.IVF_PQ:
pytest.skip("skip pq case temporary")
return request.param
@pytest.fixture(
@ -58,6 +60,8 @@ class TestSearchBase:
if "internal" not in args:
if request.param["index_type"] == IndexType.IVF_SQ8H:
pytest.skip("sq8h not support in open source")
if request.param["index_type"] == IndexType.IVF_PQ:
pytest.skip("skip pq case temporary")
return request.param
"""
generate top-k params
@ -89,13 +93,13 @@ class TestSearchBase:
else:
assert not status.OK()
def test_search_l2_index_params(self, connect, table, get_index_params):
def test_search_l2_index_params(self, connect, table, get_simple_index_params):
'''
target: test basic search fuction, all the search params is corrent, test all index params, and build
method: search with the given vectors, check the result
expected: search status ok, and the length of the result is top_k
'''
index_params = get_index_params
index_params = get_simple_index_params
logging.getLogger().info(index_params)
vectors, ids = self.init_data(connect, table)
status = connect.create_index(table, index_params)
@ -297,14 +301,14 @@ class TestSearchBase:
assert result[0][0].distance <= epsilon
assert result[1][0].distance <= epsilon
def test_search_ip_index_params(self, connect, ip_table, get_index_params):
def test_search_ip_index_params(self, connect, ip_table, get_simple_index_params):
'''
target: test basic search fuction, all the search params is corrent, test all index params, and build
method: search with the given vectors, check the result
expected: search status ok, and the length of the result is top_k
'''
index_params = get_index_params
index_params = get_simple_index_params
logging.getLogger().info(index_params)
vectors, ids = self.init_data(connect, ip_table)
status = connect.create_index(ip_table, index_params)

2
tests/milvus_python_test/test_table.py
View File

@ -594,6 +594,8 @@ class TestTable:
if "internal" not in args:
if request.param["index_type"] == IndexType.IVF_SQ8H:
pytest.skip("sq8h not support in open source")
# if request.param["index_type"] == IndexType.IVF_PQ:
# pytest.skip("sq8h not support in open source")
return request.param
@pytest.mark.level(1)

2
tests/milvus_python_test/test_table_count.py
View File

@ -270,6 +270,8 @@ class TestTableCountIP:
if "internal" not in args:
if request.param["index_type"] == IndexType.IVF_SQ8H:
pytest.skip("sq8h not support in open source")
if request.param["index_type"] == IndexType.IVF_PQ:
pytest.skip("skip pq case temporary")
return request.param
def test_table_rows_count(self, connect, ip_table, add_vectors_nb):