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

Merge pull request #361 from yhmo/0.6.0 

#266 Rpc request source code refactor
pull/398/head
Jin Hai 2019-11-18 17:45:13 +08:00 committed by GitHub
parent ef5256d4a9 661a60374e
commit 4744c0cc77
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
52 changed files with 2604 additions and 1612 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
CHANGELOG.md
View File

@ -10,6 +10,7 @@ Please mark all change in change log and use the ticket from JIRA.
- \#316 - Some files not merged after vectors added
- \#327 - Search does not use GPU when index type is FLAT
- \#340 - Test cases run failed on 0.6.0
- \#353 - Rename config.h.in to version.h.in
- \#374 - sdk_simple return empty result
## Feature
@ -22,6 +23,7 @@ Please mark all change in change log and use the ticket from JIRA.
- \#275 - Rename C++ SDK IndexType
- \#284 - Change C++ SDK to shared library
- \#260 - C++ SDK README
- \#266 - Rpc request source code refactor
- \#314 - add Find FAISS in CMake
- \#310 - Add Q&A for 'protocol https not supported or disable in libcurl' issue
- \#322 - Add option to enable / disable prometheus

2
core/CMakeLists.txt
View File

@ -70,7 +70,7 @@ if (MILVUS_VERSION_MAJOR STREQUAL ""
endif()
message(STATUS "Build version = ${MILVUS_VERSION}")
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/src/config.h.in ${CMAKE_CURRENT_SOURCE_DIR}/src/config.h @ONLY)
configure_file(${CMAKE_CURRENT_SOURCE_DIR}/src/version.h.in ${CMAKE_CURRENT_SOURCE_DIR}/src/version.h @ONLY)
message(STATUS "Milvus version: "
"${MILVUS_VERSION_MAJOR}.${MILVUS_VERSION_MINOR}.${MILVUS_VERSION_PATCH} "

8
core/src/CMakeLists.txt
View File

@ -75,7 +75,13 @@ set(thirdparty_files
)
aux_source_directory(${MILVUS_ENGINE_SRC}/server server_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/server/grpc_impl grpc_server_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/server/grpc_impl/request grpc_request_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/server/grpc_impl grpc_impl_files)
set(grpc_server_files
${grpc_request_files}
${grpc_impl_files}
)
aux_source_directory(${MILVUS_ENGINE_SRC}/utils utils_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/wrapper wrapper_files)

2
core/src/db/meta/MetaTypes.h
View File

@ -19,7 +19,7 @@
#include "db/Constants.h"
#include "db/engine/ExecutionEngine.h"
#include "src/config.h"
#include "src/version.h"
#include <map>
#include <memory>

57
core/src/grpc/gen-milvus/milvus.pb.h
View File

@ -1321,8 +1321,7 @@ class RowRecord :
void clear_vector_data();
float vector_data(int index) const;
void set_vector_data(int index, float value);
// void add_vector_data(float value);
void add_vector_data(std::vector<float>::const_iterator begin, std::vector<float>::const_iterator end);
void add_vector_data(float value);
const ::PROTOBUF_NAMESPACE_ID::RepeatedField< float >&
vector_data() const;
::PROTOBUF_NAMESPACE_ID::RepeatedField< float >*
@ -1474,9 +1473,7 @@ class InsertParam :
void clear_row_id_array();
::PROTOBUF_NAMESPACE_ID::int64 row_id_array(int index) const;
void set_row_id_array(int index, ::PROTOBUF_NAMESPACE_ID::int64 value);
// void add_row_id_array(::PROTOBUF_NAMESPACE_ID::int64 value);
void add_row_id_array(std::vector<::PROTOBUF_NAMESPACE_ID::int64>::const_iterator begin,
std::vector<::PROTOBUF_NAMESPACE_ID::int64>::const_iterator end);
void add_row_id_array(::PROTOBUF_NAMESPACE_ID::int64 value);
const ::PROTOBUF_NAMESPACE_ID::RepeatedField< ::PROTOBUF_NAMESPACE_ID::int64 >&
row_id_array() const;
::PROTOBUF_NAMESPACE_ID::RepeatedField< ::PROTOBUF_NAMESPACE_ID::int64 >*
@ -2139,9 +2136,7 @@ class TopKQueryResult :
void clear_ids();
::PROTOBUF_NAMESPACE_ID::int64 ids(int index) const;
void set_ids(int index, ::PROTOBUF_NAMESPACE_ID::int64 value);
// void add_ids(::PROTOBUF_NAMESPACE_ID::int64 value);
void add_ids(std::vector<::PROTOBUF_NAMESPACE_ID::int64>::const_iterator begin,
std::vector<::PROTOBUF_NAMESPACE_ID::int64>::const_iterator end);
void add_ids(::PROTOBUF_NAMESPACE_ID::int64 value);
const ::PROTOBUF_NAMESPACE_ID::RepeatedField< ::PROTOBUF_NAMESPACE_ID::int64 >&
ids() const;
::PROTOBUF_NAMESPACE_ID::RepeatedField< ::PROTOBUF_NAMESPACE_ID::int64 >*
@ -2152,8 +2147,7 @@ class TopKQueryResult :
void clear_distances();
float distances(int index) const;
void set_distances(int index, float value);
// void add_distances(float value);
void add_distances(std::vector<float>::const_iterator begin, std::vector<float>::const_iterator end);
void add_distances(float value);
const ::PROTOBUF_NAMESPACE_ID::RepeatedField< float >&
distances() const;
::PROTOBUF_NAMESPACE_ID::RepeatedField< float >*
@ -3928,14 +3922,9 @@ inline void RowRecord::set_vector_data(int index, float value) {
vector_data_.Set(index, value);
// @@protoc_insertion_point(field_set:milvus.grpc.RowRecord.vector_data)
}
//inline void RowRecord::add_vector_data(float value) {
// vector_data_.Add(value);
// // @@protoc_insertion_point(field_add:milvus.grpc.RowRecord.vector_data)
//}
inline void RowRecord::add_vector_data(std::vector<float>::const_iterator begin,
std::vector<float>::const_iterator end) {
vector_data_.Add(begin, end);
// @@protoc_insertion_point(field_add:milvus.grpc.RowRecord.vector_data)
inline void RowRecord::add_vector_data(float value) {
vector_data_.Add(value);
// @@protoc_insertion_point(field_add:milvus.grpc.RowRecord.vector_data)
}
inline const ::PROTOBUF_NAMESPACE_ID::RepeatedField< float >&
RowRecord::vector_data() const {
@ -4048,14 +4037,9 @@ inline void InsertParam::set_row_id_array(int index, ::PROTOBUF_NAMESPACE_ID::in
row_id_array_.Set(index, value);
// @@protoc_insertion_point(field_set:milvus.grpc.InsertParam.row_id_array)
}
//inline void InsertParam::add_row_id_array(::PROTOBUF_NAMESPACE_ID::int64 value) {
// row_id_array_.Add(value);
// // @@protoc_insertion_point(field_add:milvus.grpc.InsertParam.row_id_array)
//}
inline void InsertParam::add_row_id_array(std::vector<::PROTOBUF_NAMESPACE_ID::int64>::const_iterator begin,
std::vector<::PROTOBUF_NAMESPACE_ID::int64>::const_iterator end) {
row_id_array_.Add(begin, end);
// @@protoc_insertion_point(field_add:milvus.grpc.InsertParam.row_id_array)
inline void InsertParam::add_row_id_array(::PROTOBUF_NAMESPACE_ID::int64 value) {
row_id_array_.Add(value);
// @@protoc_insertion_point(field_add:milvus.grpc.InsertParam.row_id_array)
}
inline const ::PROTOBUF_NAMESPACE_ID::RepeatedField< ::PROTOBUF_NAMESPACE_ID::int64 >&
InsertParam::row_id_array() const {
@ -4604,14 +4588,9 @@ inline void TopKQueryResult::set_ids(int index, ::PROTOBUF_NAMESPACE_ID::int64 v
ids_.Set(index, value);
// @@protoc_insertion_point(field_set:milvus.grpc.TopKQueryResult.ids)
}
//inline void TopKQueryResult::add_ids(::PROTOBUF_NAMESPACE_ID::int64 value) {
// ids_.Add(value);
// // @@protoc_insertion_point(field_add:milvus.grpc.TopKQueryResult.ids)
//}
inline void TopKQueryResult::add_ids(std::vector<::PROTOBUF_NAMESPACE_ID::int64>::const_iterator begin,
std::vector<::PROTOBUF_NAMESPACE_ID::int64>::const_iterator end) {
ids_.Add(begin,end);
// @@protoc_insertion_point(field_add:milvus.grpc.TopKQueryResult.ids)
inline void TopKQueryResult::add_ids(::PROTOBUF_NAMESPACE_ID::int64 value) {
ids_.Add(value);
// @@protoc_insertion_point(field_add:milvus.grpc.TopKQueryResult.ids)
}
inline const ::PROTOBUF_NAMESPACE_ID::RepeatedField< ::PROTOBUF_NAMESPACE_ID::int64 >&
TopKQueryResult::ids() const {
@ -4639,13 +4618,9 @@ inline void TopKQueryResult::set_distances(int index, float value) {
distances_.Set(index, value);
// @@protoc_insertion_point(field_set:milvus.grpc.TopKQueryResult.distances)
}
//inline void TopKQueryResult::add_distances(float value) {
// distances_.Add(value);
// // @@protoc_insertion_point(field_add:milvus.grpc.TopKQueryResult.distances)
//}
inline void TopKQueryResult::add_distances(std::vector<float>::const_iterator begin, std::vector<float>::const_iterator end) {
distances_.Add(begin, end);
// @@protoc_insertion_point(field_add:milvus.grpc.TopKQueryResult.distances)
inline void TopKQueryResult::add_distances(float value) {
distances_.Add(value);
// @@protoc_insertion_point(field_add:milvus.grpc.TopKQueryResult.distances)
}
inline const ::PROTOBUF_NAMESPACE_ID::RepeatedField< float >&
TopKQueryResult::distances() const {

2
core/src/main.cpp
View File

@ -25,7 +25,7 @@
#include "easyloggingpp/easylogging++.h"
#include "metrics/Metrics.h"
#include "server/Server.h"
#include "src/config.h"
#include "src/version.h"
#include "utils/CommonUtil.h"
#include "utils/SignalUtil.h"

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

@ -17,7 +17,7 @@
#include "sdk/grpc/ClientProxy.h"
#include "grpc/gen-milvus/milvus.grpc.pb.h"
#include "src/config.h"
#include "src/version.h"
#include <memory>
#include <string>
@ -32,6 +32,13 @@ UriCheck(const std::string& uri) {
return (index != std::string::npos);
}
void
CopyRowRecord(::milvus::grpc::RowRecord* target, const RowRecord& src) {
auto vector_data = target->mutable_vector_data();
vector_data->Resize(static_cast<int>(src.data.size()), 0.0);
memcpy(vector_data->mutable_data(), src.data.data(), src.data.size() * sizeof(float));
}
Status
ClientProxy::Connect(const ConnectParam& param) {
std::string uri = param.ip_address + ":" + param.port;
@ -189,14 +196,16 @@ ClientProxy::Insert(const std::string& table_name, const std::string& partition_
for (auto& record : record_array) {
::milvus::grpc::RowRecord* grpc_record = insert_param.add_row_record_array();
grpc_record->add_vector_data(record.data.begin(), record.data.end());
CopyRowRecord(grpc_record, record);
}
// Single thread
::milvus::grpc::VectorIds vector_ids;
if (!id_array.empty()) {
/* set user's ids */
insert_param.add_row_id_array(id_array.begin(), id_array.end());
auto row_ids = insert_param.mutable_row_id_array();
row_ids->Resize(static_cast<int>(id_array.size()), -1);
memcpy(row_ids->mutable_data(), id_array.data(), id_array.size() * sizeof(int64_t));
client_ptr_->Insert(vector_ids, insert_param, status);
} else {
client_ptr_->Insert(vector_ids, insert_param, status);
@ -226,7 +235,7 @@ ClientProxy::Search(const std::string& table_name, const std::vector<std::string
}
for (auto& record : query_record_array) {
::milvus::grpc::RowRecord* row_record = search_param.add_query_record_array();
row_record->add_vector_data(record.data.begin(), record.data.end());
CopyRowRecord(row_record, record);
}
// step 2: convert range array

2
core/src/server/Server.cpp
View File

@ -25,7 +25,7 @@
#include "server/DBWrapper.h"
#include "server/Server.h"
#include "server/grpc_impl/GrpcServer.h"
#include "src/config.h"
#include "src/version.h"
#include "utils/Log.h"
#include "utils/LogUtil.h"
#include "utils/SignalUtil.h"

92
core/src/server/grpc_impl/GrpcRequestHandler.cpp
View File

@ -16,7 +16,24 @@
// under the License.
#include "server/grpc_impl/GrpcRequestHandler.h"
#include "server/grpc_impl/GrpcRequestTask.h"
#include "server/grpc_impl/GrpcRequestScheduler.h"
#include "server/grpc_impl/request/CmdRequest.h"
#include "server/grpc_impl/request/CountTableRequest.h"
#include "server/grpc_impl/request/CreateIndexRequest.h"
#include "server/grpc_impl/request/CreatePartitionRequest.h"
#include "server/grpc_impl/request/CreateTableRequest.h"
#include "server/grpc_impl/request/DeleteByDateRequest.h"
#include "server/grpc_impl/request/DescribeIndexRequest.h"
#include "server/grpc_impl/request/DescribeTableRequest.h"
#include "server/grpc_impl/request/DropIndexRequest.h"
#include "server/grpc_impl/request/DropPartitionRequest.h"
#include "server/grpc_impl/request/DropTableRequest.h"
#include "server/grpc_impl/request/HasTableRequest.h"
#include "server/grpc_impl/request/InsertRequest.h"
#include "server/grpc_impl/request/PreloadTableRequest.h"
#include "server/grpc_impl/request/SearchRequest.h"
#include "server/grpc_impl/request/ShowPartitionsRequest.h"
#include "server/grpc_impl/request/ShowTablesRequest.h"
#include "utils/TimeRecorder.h"
#include <vector>
@ -28,8 +45,8 @@ namespace grpc {
::grpc::Status
GrpcRequestHandler::CreateTable(::grpc::ServerContext* context, const ::milvus::grpc::TableSchema* request,
::milvus::grpc::Status* response) {
BaseTaskPtr task_ptr = CreateTableTask::Create(request);
GrpcRequestScheduler::ExecTask(task_ptr, response);
BaseRequestPtr request_ptr = CreateTableRequest::Create(request);
GrpcRequestScheduler::ExecRequest(request_ptr, response);
return ::grpc::Status::OK;
}
@ -37,9 +54,9 @@ GrpcRequestHandler::CreateTable(::grpc::ServerContext* context, const ::milvus::
GrpcRequestHandler::HasTable(::grpc::ServerContext* context, const ::milvus::grpc::TableName* request,
::milvus::grpc::BoolReply* response) {
bool has_table = false;
BaseTaskPtr task_ptr = HasTableTask::Create(request->table_name(), has_table);
BaseRequestPtr request_ptr = HasTableRequest::Create(request->table_name(), has_table);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->set_bool_reply(has_table);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
@ -49,25 +66,25 @@ GrpcRequestHandler::HasTable(::grpc::ServerContext* context, const ::milvus::grp
::grpc::Status
GrpcRequestHandler::DropTable(::grpc::ServerContext* context, const ::milvus::grpc::TableName* request,
::milvus::grpc::Status* response) {
BaseTaskPtr task_ptr = DropTableTask::Create(request->table_name());
GrpcRequestScheduler::ExecTask(task_ptr, response);
BaseRequestPtr request_ptr = DropTableRequest::Create(request->table_name());
GrpcRequestScheduler::ExecRequest(request_ptr, response);
return ::grpc::Status::OK;
}
::grpc::Status
GrpcRequestHandler::CreateIndex(::grpc::ServerContext* context, const ::milvus::grpc::IndexParam* request,
::milvus::grpc::Status* response) {
BaseTaskPtr task_ptr = CreateIndexTask::Create(request);
GrpcRequestScheduler::ExecTask(task_ptr, response);
BaseRequestPtr request_ptr = CreateIndexRequest::Create(request);
GrpcRequestScheduler::ExecRequest(request_ptr, response);
return ::grpc::Status::OK;
}
::grpc::Status
GrpcRequestHandler::Insert(::grpc::ServerContext* context, const ::milvus::grpc::InsertParam* request,
::milvus::grpc::VectorIds* response) {
BaseTaskPtr task_ptr = InsertTask::Create(request, response);
BaseRequestPtr request_ptr = InsertRequest::Create(request, response);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
return ::grpc::Status::OK;
@ -77,9 +94,9 @@ GrpcRequestHandler::Insert(::grpc::ServerContext* context, const ::milvus::grpc:
GrpcRequestHandler::Search(::grpc::ServerContext* context, const ::milvus::grpc::SearchParam* request,
::milvus::grpc::TopKQueryResult* response) {
std::vector<std::string> file_id_array;
BaseTaskPtr task_ptr = SearchTask::Create(request, file_id_array, response);
BaseRequestPtr request_ptr = SearchRequest::Create(request, file_id_array, response);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->mutable_status()->set_error_code(grpc_status.error_code());
response->mutable_status()->set_reason(grpc_status.reason());
return ::grpc::Status::OK;
@ -93,9 +110,10 @@ GrpcRequestHandler::SearchInFiles(::grpc::ServerContext* context, const ::milvus
file_id_array.push_back(request->file_id_array(i));
}
::milvus::grpc::SearchInFilesParam* request_mutable = const_cast<::milvus::grpc::SearchInFilesParam*>(request);
BaseTaskPtr task_ptr = SearchTask::Create(request_mutable->mutable_search_param(), file_id_array, response);
BaseRequestPtr request_ptr =
SearchRequest::Create(request_mutable->mutable_search_param(), file_id_array, response);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->mutable_status()->set_error_code(grpc_status.error_code());
response->mutable_status()->set_reason(grpc_status.reason());
return ::grpc::Status::OK;
@ -104,9 +122,9 @@ GrpcRequestHandler::SearchInFiles(::grpc::ServerContext* context, const ::milvus
::grpc::Status
GrpcRequestHandler::DescribeTable(::grpc::ServerContext* context, const ::milvus::grpc::TableName* request,
::milvus::grpc::TableSchema* response) {
BaseTaskPtr task_ptr = DescribeTableTask::Create(request->table_name(), response);
BaseRequestPtr request_ptr = DescribeTableRequest::Create(request->table_name(), response);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->mutable_status()->set_error_code(grpc_status.error_code());
response->mutable_status()->set_reason(grpc_status.reason());
return ::grpc::Status::OK;
@ -116,9 +134,9 @@ GrpcRequestHandler::DescribeTable(::grpc::ServerContext* context, const ::milvus
GrpcRequestHandler::CountTable(::grpc::ServerContext* context, const ::milvus::grpc::TableName* request,
::milvus::grpc::TableRowCount* response) {
int64_t row_count = 0;
BaseTaskPtr task_ptr = CountTableTask::Create(request->table_name(), row_count);
BaseRequestPtr request_ptr = CountTableRequest::Create(request->table_name(), row_count);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->set_table_row_count(row_count);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
@ -128,9 +146,9 @@ GrpcRequestHandler::CountTable(::grpc::ServerContext* context, const ::milvus::g
::grpc::Status
GrpcRequestHandler::ShowTables(::grpc::ServerContext* context, const ::milvus::grpc::Command* request,
::milvus::grpc::TableNameList* response) {
BaseTaskPtr task_ptr = ShowTablesTask::Create(response);
BaseRequestPtr request_ptr = ShowTablesRequest::Create(response);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->mutable_status()->set_error_code(grpc_status.error_code());
response->mutable_status()->set_reason(grpc_status.reason());
return ::grpc::Status::OK;
@ -140,9 +158,9 @@ GrpcRequestHandler::ShowTables(::grpc::ServerContext* context, const ::milvus::g
GrpcRequestHandler::Cmd(::grpc::ServerContext* context, const ::milvus::grpc::Command* request,
::milvus::grpc::StringReply* response) {
std::string result;
BaseTaskPtr task_ptr = CmdTask::Create(request->cmd(), result);
BaseRequestPtr request_ptr = CmdRequest::Create(request->cmd(), result);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->set_string_reply(result);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
@ -152,9 +170,9 @@ GrpcRequestHandler::Cmd(::grpc::ServerContext* context, const ::milvus::grpc::Co
::grpc::Status
GrpcRequestHandler::DeleteByDate(::grpc::ServerContext* context, const ::milvus::grpc::DeleteByDateParam* request,
::milvus::grpc::Status* response) {
BaseTaskPtr task_ptr = DeleteByDateTask::Create(request);
BaseRequestPtr request_ptr = DeleteByDateRequest::Create(request);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->set_error_code(grpc_status.error_code());
response->set_reason(grpc_status.reason());
return ::grpc::Status::OK;
@ -163,9 +181,9 @@ GrpcRequestHandler::DeleteByDate(::grpc::ServerContext* context, const ::milvus:
::grpc::Status
GrpcRequestHandler::PreloadTable(::grpc::ServerContext* context, const ::milvus::grpc::TableName* request,
::milvus::grpc::Status* response) {
BaseTaskPtr task_ptr = PreloadTableTask::Create(request->table_name());
BaseRequestPtr request_ptr = PreloadTableRequest::Create(request->table_name());
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->set_reason(grpc_status.reason());
response->set_error_code(grpc_status.error_code());
return ::grpc::Status::OK;
@ -174,9 +192,9 @@ GrpcRequestHandler::PreloadTable(::grpc::ServerContext* context, const ::milvus:
::grpc::Status
GrpcRequestHandler::DescribeIndex(::grpc::ServerContext* context, const ::milvus::grpc::TableName* request,
::milvus::grpc::IndexParam* response) {
BaseTaskPtr task_ptr = DescribeIndexTask::Create(request->table_name(), response);
BaseRequestPtr request_ptr = DescribeIndexRequest::Create(request->table_name(), response);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
return ::grpc::Status::OK;
@ -185,9 +203,9 @@ GrpcRequestHandler::DescribeIndex(::grpc::ServerContext* context, const ::milvus
::grpc::Status
GrpcRequestHandler::DropIndex(::grpc::ServerContext* context, const ::milvus::grpc::TableName* request,
::milvus::grpc::Status* response) {
BaseTaskPtr task_ptr = DropIndexTask::Create(request->table_name());
BaseRequestPtr request_ptr = DropIndexRequest::Create(request->table_name());
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->set_reason(grpc_status.reason());
response->set_error_code(grpc_status.error_code());
return ::grpc::Status::OK;
@ -196,17 +214,17 @@ GrpcRequestHandler::DropIndex(::grpc::ServerContext* context, const ::milvus::gr
::grpc::Status
GrpcRequestHandler::CreatePartition(::grpc::ServerContext* context, const ::milvus::grpc::PartitionParam* request,
::milvus::grpc::Status* response) {
BaseTaskPtr task_ptr = CreatePartitionTask::Create(request);
GrpcRequestScheduler::ExecTask(task_ptr, response);
BaseRequestPtr request_ptr = CreatePartitionRequest::Create(request);
GrpcRequestScheduler::ExecRequest(request_ptr, response);
return ::grpc::Status::OK;
}
::grpc::Status
GrpcRequestHandler::ShowPartitions(::grpc::ServerContext* context, const ::milvus::grpc::TableName* request,
::milvus::grpc::PartitionList* response) {
BaseTaskPtr task_ptr = ShowPartitionsTask::Create(request->table_name(), response);
BaseRequestPtr request_ptr = ShowPartitionsRequest::Create(request->table_name(), response);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->mutable_status()->set_reason(grpc_status.reason());
response->mutable_status()->set_error_code(grpc_status.error_code());
return ::grpc::Status::OK;
@ -215,9 +233,9 @@ GrpcRequestHandler::ShowPartitions(::grpc::ServerContext* context, const ::milvu
::grpc::Status
GrpcRequestHandler::DropPartition(::grpc::ServerContext* context, const ::milvus::grpc::PartitionParam* request,
::milvus::grpc::Status* response) {
BaseTaskPtr task_ptr = DropPartitionTask::Create(request);
BaseRequestPtr request_ptr = DropPartitionRequest::Create(request);
::milvus::grpc::Status grpc_status;
GrpcRequestScheduler::ExecTask(task_ptr, &grpc_status);
GrpcRequestScheduler::ExecRequest(request_ptr, &grpc_status);
response->set_reason(grpc_status.reason());
response->set_error_code(grpc_status.error_code());
return ::grpc::Status::OK;

97
core/src/server/grpc_impl/GrpcRequestScheduler.cpp
View File

@ -70,43 +70,6 @@ ErrorMap(ErrorCode code) {
}
} // namespace
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
GrpcBaseTask::GrpcBaseTask(const std::string& task_group, bool async)
: task_group_(task_group), async_(async), done_(false) {
}
GrpcBaseTask::~GrpcBaseTask() {
WaitToFinish();
}
Status
GrpcBaseTask::Execute() {
status_ = OnExecute();
Done();
return status_;
}
void
GrpcBaseTask::Done() {
done_ = true;
finish_cond_.notify_all();
}
Status
GrpcBaseTask::SetStatus(ErrorCode error_code, const std::string& error_msg) {
status_ = Status(error_code, error_msg);
SERVER_LOG_ERROR << error_msg;
return status_;
}
Status
GrpcBaseTask::WaitToFinish() {
std::unique_lock<std::mutex> lock(finish_mtx_);
finish_cond_.wait(lock, [this] { return done_; });
return status_;
}
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
GrpcRequestScheduler::GrpcRequestScheduler() : stopped_(false) {
Start();
@ -117,17 +80,17 @@ GrpcRequestScheduler::~GrpcRequestScheduler() {
}
void
GrpcRequestScheduler::ExecTask(BaseTaskPtr& task_ptr, ::milvus::grpc::Status* grpc_status) {
if (task_ptr == nullptr) {
GrpcRequestScheduler::ExecRequest(BaseRequestPtr& request_ptr, ::milvus::grpc::Status* grpc_status) {
if (request_ptr == nullptr) {
return;
}
GrpcRequestScheduler& scheduler = GrpcRequestScheduler::GetInstance();
scheduler.ExecuteTask(task_ptr);
scheduler.ExecuteRequest(request_ptr);
if (!task_ptr->IsAsync()) {
task_ptr->WaitToFinish();
const Status& status = task_ptr->status();
if (!request_ptr->IsAsync()) {
request_ptr->WaitToFinish();
const Status& status = request_ptr->status();
if (!status.ok()) {
grpc_status->set_reason(status.message());
grpc_status->set_error_code(ErrorMap(status.code()));
@ -153,7 +116,7 @@ GrpcRequestScheduler::Stop() {
SERVER_LOG_INFO << "Scheduler gonna stop...";
{
std::lock_guard<std::mutex> lock(queue_mtx_);
for (auto iter : task_groups_) {
for (auto iter : request_groups_) {
if (iter.second != nullptr) {
iter.second->Put(nullptr);
}
@ -171,64 +134,64 @@ GrpcRequestScheduler::Stop() {
}
Status
GrpcRequestScheduler::ExecuteTask(const BaseTaskPtr& task_ptr) {
if (task_ptr == nullptr) {
GrpcRequestScheduler::ExecuteRequest(const BaseRequestPtr& request_ptr) {
if (request_ptr == nullptr) {
return Status::OK();
}
auto status = PutTaskToQueue(task_ptr);
auto status = PutToQueue(request_ptr);
if (!status.ok()) {
SERVER_LOG_ERROR << "Put task to queue failed with code: " << status.ToString();
SERVER_LOG_ERROR << "Put request to queue failed with code: " << status.ToString();
return status;
}
if (task_ptr->IsAsync()) {
if (request_ptr->IsAsync()) {
return Status::OK(); // async execution, caller need to call WaitToFinish at somewhere
}
return task_ptr->WaitToFinish(); // sync execution
return request_ptr->WaitToFinish(); // sync execution
}
void
GrpcRequestScheduler::TakeTaskToExecute(TaskQueuePtr task_queue) {
if (task_queue == nullptr) {
GrpcRequestScheduler::TakeToExecute(RequestQueuePtr request_queue) {
if (request_queue == nullptr) {
return;
}
while (true) {
BaseTaskPtr task = task_queue->Take();
if (task == nullptr) {
SERVER_LOG_ERROR << "Take null from task queue, stop thread";
BaseRequestPtr request = request_queue->Take();
if (request == nullptr) {
SERVER_LOG_ERROR << "Take null from request queue, stop thread";
break; // stop the thread
}
try {
auto status = task->Execute();
auto status = request->Execute();
if (!status.ok()) {
SERVER_LOG_ERROR << "Task failed with code: " << status.ToString();
SERVER_LOG_ERROR << "Request failed with code: " << status.ToString();
}
} catch (std::exception& ex) {
SERVER_LOG_ERROR << "Task failed to execute: " << ex.what();
SERVER_LOG_ERROR << "Request failed to execute: " << ex.what();
}
}
}
Status
GrpcRequestScheduler::PutTaskToQueue(const BaseTaskPtr& task_ptr) {
GrpcRequestScheduler::PutToQueue(const BaseRequestPtr& request_ptr) {
std::lock_guard<std::mutex> lock(queue_mtx_);
std::string group_name = task_ptr->TaskGroup();
if (task_groups_.count(group_name) > 0) {
task_groups_[group_name]->Put(task_ptr);
std::string group_name = request_ptr->RequestGroup();
if (request_groups_.count(group_name) > 0) {
request_groups_[group_name]->Put(request_ptr);
} else {
TaskQueuePtr queue = std::make_shared<TaskQueue>();
queue->Put(task_ptr);
task_groups_.insert(std::make_pair(group_name, queue));
RequestQueuePtr queue = std::make_shared<RequestQueue>();
queue->Put(request_ptr);
request_groups_.insert(std::make_pair(group_name, queue));
// start a thread
ThreadPtr thread = std::make_shared<std::thread>(&GrpcRequestScheduler::TakeTaskToExecute, this, queue);
ThreadPtr thread = std::make_shared<std::thread>(&GrpcRequestScheduler::TakeToExecute, this, queue);
execute_threads_.push_back(thread);
SERVER_LOG_INFO << "Create new thread for task group: " << group_name;
SERVER_LOG_INFO << "Create new thread for request group: " << group_name;
}
return Status::OK();

64
core/src/server/grpc_impl/GrpcRequestScheduler.h
View File

@ -19,6 +19,7 @@
#include "grpc/gen-status/status.grpc.pb.h"
#include "grpc/gen-status/status.pb.h"
#include "server/grpc_impl/request/GrpcBaseRequest.h"
#include "utils/BlockingQueue.h"
#include "utils/Status.h"
@ -32,57 +33,8 @@ namespace milvus {
namespace server {
namespace grpc {
class GrpcBaseTask {
protected:
explicit GrpcBaseTask(const std::string& task_group, bool async = false);
virtual ~GrpcBaseTask();
public:
Status
Execute();
void
Done();
Status
WaitToFinish();
std::string
TaskGroup() const {
return task_group_;
}
const Status&
status() const {
return status_;
}
bool
IsAsync() const {
return async_;
}
protected:
virtual Status
OnExecute() = 0;
Status
SetStatus(ErrorCode error_code, const std::string& error_msg);
protected:
mutable std::mutex finish_mtx_;
std::condition_variable finish_cond_;
std::string task_group_;
bool async_;
bool done_;
Status status_;
};
using BaseTaskPtr = std::shared_ptr<GrpcBaseTask>;
using TaskQueue = BlockingQueue<BaseTaskPtr>;
using TaskQueuePtr = std::shared_ptr<TaskQueue>;
using RequestQueue = BlockingQueue<BaseRequestPtr>;
using RequestQueuePtr = std::shared_ptr<RequestQueue>;
using ThreadPtr = std::shared_ptr<std::thread>;
class GrpcRequestScheduler {
@ -100,10 +52,10 @@ class GrpcRequestScheduler {
Stop();
Status
ExecuteTask(const BaseTaskPtr& task_ptr);
ExecuteRequest(const BaseRequestPtr& request_ptr);
static void
ExecTask(BaseTaskPtr& task_ptr, ::milvus::grpc::Status* grpc_status);
ExecRequest(BaseRequestPtr& request_ptr, ::milvus::grpc::Status* grpc_status);
protected:
GrpcRequestScheduler();
@ -111,15 +63,15 @@ class GrpcRequestScheduler {
virtual ~GrpcRequestScheduler();
void
TakeTaskToExecute(TaskQueuePtr task_queue);
TakeToExecute(RequestQueuePtr request_queue);
Status
PutTaskToQueue(const BaseTaskPtr& task_ptr);
PutToQueue(const BaseRequestPtr& request_ptr);
private:
mutable std::mutex queue_mtx_;
std::map<std::string, TaskQueuePtr> task_groups_;
std::map<std::string, RequestQueuePtr> request_groups_;
std::vector<ThreadPtr> execute_threads_;

1066
core/src/server/grpc_impl/GrpcRequestTask.cpp
View File

File diff suppressed because it is too large Load Diff

321
core/src/server/grpc_impl/GrpcRequestTask.h
View File

@ -1,321 +0,0 @@
// 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 "server/grpc_impl/GrpcRequestScheduler.h"
#include "utils/Status.h"
#include "grpc/gen-milvus/milvus.grpc.pb.h"
#include "grpc/gen-status/status.pb.h"
#include <condition_variable>
#include <memory>
#include <string>
#include <vector>
namespace milvus {
namespace server {
namespace grpc {
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class CreateTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::TableSchema* schema);
protected:
explicit CreateTableTask(const ::milvus::grpc::TableSchema* schema);
Status
OnExecute() override;
private:
const ::milvus::grpc::TableSchema* schema_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class HasTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name, bool& has_table);
protected:
HasTableTask(const std::string& table_name, bool& has_table);
Status
OnExecute() override;
private:
std::string table_name_;
bool& has_table_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DescribeTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name, ::milvus::grpc::TableSchema* schema);
protected:
DescribeTableTask(const std::string& table_name, ::milvus::grpc::TableSchema* schema);
Status
OnExecute() override;
private:
std::string table_name_;
::milvus::grpc::TableSchema* schema_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DropTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name);
protected:
explicit DropTableTask(const std::string& table_name);
Status
OnExecute() override;
private:
std::string table_name_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class CreateIndexTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::IndexParam* index_param);
protected:
explicit CreateIndexTask(const ::milvus::grpc::IndexParam* index_param);
Status
OnExecute() override;
private:
const ::milvus::grpc::IndexParam* index_param_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class ShowTablesTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(::milvus::grpc::TableNameList* table_name_list);
protected:
explicit ShowTablesTask(::milvus::grpc::TableNameList* table_name_list);
Status
OnExecute() override;
private:
::milvus::grpc::TableNameList* table_name_list_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class InsertTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::InsertParam* insert_param, ::milvus::grpc::VectorIds* record_ids);
protected:
InsertTask(const ::milvus::grpc::InsertParam* insert_param, ::milvus::grpc::VectorIds* record_ids);
Status
OnExecute() override;
private:
const ::milvus::grpc::InsertParam* insert_param_;
::milvus::grpc::VectorIds* record_ids_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class SearchTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::SearchParam* search_param, const std::vector<std::string>& file_id_array,
::milvus::grpc::TopKQueryResult* response);
protected:
SearchTask(const ::milvus::grpc::SearchParam* search_param, const std::vector<std::string>& file_id_array,
::milvus::grpc::TopKQueryResult* response);
Status
OnExecute() override;
private:
const ::milvus::grpc::SearchParam* search_param_;
std::vector<std::string> file_id_array_;
::milvus::grpc::TopKQueryResult* topk_result_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class CountTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name, int64_t& row_count);
protected:
CountTableTask(const std::string& table_name, int64_t& row_count);
Status
OnExecute() override;
private:
std::string table_name_;
int64_t& row_count_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class CmdTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& cmd, std::string& result);
protected:
CmdTask(const std::string& cmd, std::string& result);
Status
OnExecute() override;
private:
std::string cmd_;
std::string& result_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DeleteByDateTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::DeleteByDateParam* delete_by_range_param);
protected:
explicit DeleteByDateTask(const ::milvus::grpc::DeleteByDateParam* delete_by_range_param);
Status
OnExecute() override;
private:
const ::milvus::grpc::DeleteByDateParam* delete_by_range_param_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class PreloadTableTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name);
protected:
explicit PreloadTableTask(const std::string& table_name);
Status
OnExecute() override;
private:
std::string table_name_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DescribeIndexTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name, ::milvus::grpc::IndexParam* index_param);
protected:
DescribeIndexTask(const std::string& table_name, ::milvus::grpc::IndexParam* index_param);
Status
OnExecute() override;
private:
std::string table_name_;
::milvus::grpc::IndexParam* index_param_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DropIndexTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name);
protected:
explicit DropIndexTask(const std::string& table_name);
Status
OnExecute() override;
private:
std::string table_name_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class CreatePartitionTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::PartitionParam* partition_param);
protected:
explicit CreatePartitionTask(const ::milvus::grpc::PartitionParam* partition_param);
Status
OnExecute() override;
private:
const ::milvus::grpc::PartitionParam* partition_param_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class ShowPartitionsTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const std::string& table_name, ::milvus::grpc::PartitionList* partition_list);
protected:
ShowPartitionsTask(const std::string& table_name, ::milvus::grpc::PartitionList* partition_list);
Status
OnExecute() override;
private:
std::string table_name_;
::milvus::grpc::PartitionList* partition_list_;
};
////////////////////////////////////////////////////////////////////////////////////////////////////////////////////////
class DropPartitionTask : public GrpcBaseTask {
public:
static BaseTaskPtr
Create(const ::milvus::grpc::PartitionParam* partition_param);
protected:
explicit DropPartitionTask(const ::milvus::grpc::PartitionParam* partition_param);
Status
OnExecute() override;
private:
const ::milvus::grpc::PartitionParam* partition_param_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

51
core/src/server/grpc_impl/request/CmdRequest.cpp Normal file
View File

@ -0,0 +1,51 @@
// 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 "server/grpc_impl/request/CmdRequest.h"
#include "scheduler/SchedInst.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
CmdRequest::CmdRequest(const std::string& cmd, std::string& result)
: GrpcBaseRequest(INFO_REQUEST_GROUP), cmd_(cmd), result_(result) {
}
BaseRequestPtr
CmdRequest::Create(const std::string& cmd, std::string& result) {
return std::shared_ptr<GrpcBaseRequest>(new CmdRequest(cmd, result));
}
Status
CmdRequest::OnExecute() {
if (cmd_ == "version") {
result_ = MILVUS_VERSION;
} else if (cmd_ == "tasktable") {
result_ = scheduler::ResMgrInst::GetInstance()->DumpTaskTables();
} else {
result_ = "OK";
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

46
core/src/server/grpc_impl/request/CmdRequest.h Normal file
View File

@ -0,0 +1,46 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
namespace milvus {
namespace server {
namespace grpc {
class CmdRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const std::string& cmd, std::string& result);
protected:
CmdRequest(const std::string& cmd, std::string& result);
Status
OnExecute() override;
private:
std::string cmd_;
std::string& result_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

73
core/src/server/grpc_impl/request/CountTableRequest.cpp Normal file
View File

@ -0,0 +1,73 @@
// 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 "server/grpc_impl/request/CountTableRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
CountTableRequest::CountTableRequest(const std::string& table_name, int64_t& row_count)
: GrpcBaseRequest(INFO_REQUEST_GROUP), table_name_(table_name), row_count_(row_count) {
}
BaseRequestPtr
CountTableRequest::Create(const std::string& table_name, int64_t& row_count) {
return std::shared_ptr<GrpcBaseRequest>(new CountTableRequest(table_name, row_count));
}
Status
CountTableRequest::OnExecute() {
try {
TimeRecorder rc("CountTableRequest");
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
// step 2: get row count
uint64_t row_count = 0;
status = DBWrapper::DB()->GetTableRowCount(table_name_, row_count);
if (!status.ok()) {
if (status.code(), DB_NOT_FOUND) {
return Status(SERVER_TABLE_NOT_EXIST, TableNotExistMsg(table_name_));
} else {
return status;
}
}
row_count_ = static_cast<int64_t>(row_count);
rc.ElapseFromBegin("total cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

46
core/src/server/grpc_impl/request/CountTableRequest.h Normal file
View File

@ -0,0 +1,46 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
namespace milvus {
namespace server {
namespace grpc {
class CountTableRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const std::string& table_name, int64_t& row_count);
protected:
CountTableRequest(const std::string& table_name, int64_t& row_count);
Status
OnExecute() override;
private:
std::string table_name_;
int64_t& row_count_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

96
core/src/server/grpc_impl/request/CreateIndexRequest.cpp Normal file
View File

@ -0,0 +1,96 @@
// 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 "server/grpc_impl/request/CreateIndexRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
#include <string>
namespace milvus {
namespace server {
namespace grpc {
CreateIndexRequest::CreateIndexRequest(const ::milvus::grpc::IndexParam* index_param)
: GrpcBaseRequest(DDL_DML_REQUEST_GROUP), index_param_(index_param) {
}
BaseRequestPtr
CreateIndexRequest::Create(const ::milvus::grpc::IndexParam* index_param) {
if (index_param == nullptr) {
SERVER_LOG_ERROR << "grpc input is null!";
return nullptr;
}
return std::shared_ptr<GrpcBaseRequest>(new CreateIndexRequest(index_param));
}
Status
CreateIndexRequest::OnExecute() {
try {
TimeRecorder rc("CreateIndexRequest");
// step 1: check arguments
std::string table_name_ = index_param_->table_name();
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
bool has_table = false;
status = DBWrapper::DB()->HasTable(table_name_, has_table);
if (!status.ok()) {
return status;
}
if (!has_table) {
return Status(SERVER_TABLE_NOT_EXIST, TableNotExistMsg(table_name_));
}
auto& grpc_index = index_param_->index();
status = ValidationUtil::ValidateTableIndexType(grpc_index.index_type());
if (!status.ok()) {
return status;
}
status = ValidationUtil::ValidateTableIndexNlist(grpc_index.nlist());
if (!status.ok()) {
return status;
}
// step 2: check table existence
engine::TableIndex index;
index.engine_type_ = grpc_index.index_type();
index.nlist_ = grpc_index.nlist();
status = DBWrapper::DB()->CreateIndex(table_name_, index);
if (!status.ok()) {
return status;
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

43
core/src/server/grpc_impl/request/CreateIndexRequest.h Normal file
View File

@ -0,0 +1,43 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
namespace milvus {
namespace server {
namespace grpc {
class CreateIndexRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const ::milvus::grpc::IndexParam* index_param);
protected:
explicit CreateIndexRequest(const ::milvus::grpc::IndexParam* index_param);
Status
OnExecute() override;
private:
const ::milvus::grpc::IndexParam* index_param_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

85
core/src/server/grpc_impl/request/CreatePartitionRequest.cpp Normal file
View File

@ -0,0 +1,85 @@
// 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 "server/grpc_impl/request/CreatePartitionRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
CreatePartitionRequest::CreatePartitionRequest(const ::milvus::grpc::PartitionParam* partition_param)
: GrpcBaseRequest(DDL_DML_REQUEST_GROUP), partition_param_(partition_param) {
}
BaseRequestPtr
CreatePartitionRequest::Create(const ::milvus::grpc::PartitionParam* partition_param) {
if (partition_param == nullptr) {
SERVER_LOG_ERROR << "grpc input is null!";
return nullptr;
}
return std::shared_ptr<GrpcBaseRequest>(new CreatePartitionRequest(partition_param));
}
Status
CreatePartitionRequest::OnExecute() {
TimeRecorder rc("CreatePartitionRequest");
try {
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(partition_param_->table_name());
if (!status.ok()) {
return status;
}
status = ValidationUtil::ValidateTableName(partition_param_->partition_name());
if (!status.ok()) {
return status;
}
status = ValidationUtil::ValidatePartitionTags({partition_param_->tag()});
if (!status.ok()) {
return status;
}
// step 2: create partition
status = DBWrapper::DB()->CreatePartition(partition_param_->table_name(), partition_param_->partition_name(),
partition_param_->tag());
if (!status.ok()) {
// partition could exist
if (status.code() == DB_ALREADY_EXIST) {
return Status(SERVER_INVALID_TABLE_NAME, status.message());
}
return status;
}
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
rc.ElapseFromBegin("totally cost");
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

43
core/src/server/grpc_impl/request/CreatePartitionRequest.h Normal file
View File

@ -0,0 +1,43 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
namespace milvus {
namespace server {
namespace grpc {
class CreatePartitionRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const ::milvus::grpc::PartitionParam* partition_param);
protected:
explicit CreatePartitionRequest(const ::milvus::grpc::PartitionParam* partition_param);
Status
OnExecute() override;
private:
const ::milvus::grpc::PartitionParam* partition_param_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

96
core/src/server/grpc_impl/request/CreateTableRequest.cpp Normal file
View File

@ -0,0 +1,96 @@
// 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 "server/grpc_impl/request/CreateTableRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
CreateTableRequest::CreateTableRequest(const ::milvus::grpc::TableSchema* schema)
: GrpcBaseRequest(DDL_DML_REQUEST_GROUP), schema_(schema) {
}
BaseRequestPtr
CreateTableRequest::Create(const ::milvus::grpc::TableSchema* schema) {
if (schema == nullptr) {
SERVER_LOG_ERROR << "grpc input is null!";
return nullptr;
}
return std::shared_ptr<GrpcBaseRequest>(new CreateTableRequest(schema));
}
Status
CreateTableRequest::OnExecute() {
TimeRecorder rc("CreateTableRequest");
try {
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(schema_->table_name());
if (!status.ok()) {
return status;
}
status = ValidationUtil::ValidateTableDimension(schema_->dimension());
if (!status.ok()) {
return status;
}
status = ValidationUtil::ValidateTableIndexFileSize(schema_->index_file_size());
if (!status.ok()) {
return status;
}
status = ValidationUtil::ValidateTableIndexMetricType(schema_->metric_type());
if (!status.ok()) {
return status;
}
// step 2: construct table schema
engine::meta::TableSchema table_info;
table_info.table_id_ = schema_->table_name();
table_info.dimension_ = static_cast<uint16_t>(schema_->dimension());
table_info.index_file_size_ = schema_->index_file_size();
table_info.metric_type_ = schema_->metric_type();
// step 3: create table
status = DBWrapper::DB()->CreateTable(table_info);
if (!status.ok()) {
// table could exist
if (status.code() == DB_ALREADY_EXIST) {
return Status(SERVER_INVALID_TABLE_NAME, status.message());
}
return status;
}
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
rc.ElapseFromBegin("totally cost");
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

43
core/src/server/grpc_impl/request/CreateTableRequest.h Normal file
View File

@ -0,0 +1,43 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
namespace milvus {
namespace server {
namespace grpc {
class CreateTableRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const ::milvus::grpc::TableSchema* schema);
protected:
explicit CreateTableRequest(const ::milvus::grpc::TableSchema* schema);
Status
OnExecute() override;
private:
const ::milvus::grpc::TableSchema* schema_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

101
core/src/server/grpc_impl/request/DeleteByDateRequest.cpp Normal file
View File

@ -0,0 +1,101 @@
// 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 "server/grpc_impl/request/DeleteByDateRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
#include <string>
#include <vector>
namespace milvus {
namespace server {
namespace grpc {
DeleteByDateRequest::DeleteByDateRequest(const ::milvus::grpc::DeleteByDateParam* delete_by_range_param)
: GrpcBaseRequest(DDL_DML_REQUEST_GROUP), delete_by_range_param_(delete_by_range_param) {
}
BaseRequestPtr
DeleteByDateRequest::Create(const ::milvus::grpc::DeleteByDateParam* delete_by_range_param) {
if (delete_by_range_param == nullptr) {
SERVER_LOG_ERROR << "grpc input is null!";
return nullptr;
}
return std::shared_ptr<GrpcBaseRequest>(new DeleteByDateRequest(delete_by_range_param));
}
Status
DeleteByDateRequest::OnExecute() {
try {
TimeRecorder rc("DeleteByDateRequest");
// step 1: check arguments
std::string table_name = delete_by_range_param_->table_name();
auto status = ValidationUtil::ValidateTableName(table_name);
if (!status.ok()) {
return status;
}
// step 2: check table existence
engine::meta::TableSchema table_info;
table_info.table_id_ = table_name;
status = DBWrapper::DB()->DescribeTable(table_info);
if (!status.ok()) {
if (status.code(), DB_NOT_FOUND) {
return Status(SERVER_TABLE_NOT_EXIST, TableNotExistMsg(table_name));
} else {
return status;
}
}
rc.ElapseFromBegin("check validation");
// step 3: check date range, and convert to db dates
std::vector<DB_DATE> dates;
ErrorCode error_code = SERVER_SUCCESS;
std::string error_msg;
std::vector<::milvus::grpc::Range> range_array;
range_array.emplace_back(delete_by_range_param_->range());
status = ConvertTimeRangeToDBDates(range_array, dates);
if (!status.ok()) {
return status;
}
#ifdef MILVUS_ENABLE_PROFILING
std::string fname = "/tmp/search_nq_" + this->delete_by_range_param_->table_name() + ".profiling";
ProfilerStart(fname.c_str());
#endif
status = DBWrapper::DB()->DropTable(table_name, dates);
if (!status.ok()) {
return status;
}
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

43
core/src/server/grpc_impl/request/DeleteByDateRequest.h Normal file
View File

@ -0,0 +1,43 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
namespace milvus {
namespace server {
namespace grpc {
class DeleteByDateRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const ::milvus::grpc::DeleteByDateParam* delete_by_range_param);
protected:
explicit DeleteByDateRequest(const ::milvus::grpc::DeleteByDateParam* delete_by_range_param);
Status
OnExecute() override;
private:
const ::milvus::grpc::DeleteByDateParam* delete_by_range_param_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

71
core/src/server/grpc_impl/request/DescribeIndexRequest.cpp Normal file
View File

@ -0,0 +1,71 @@
// 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 "server/grpc_impl/request/DescribeIndexRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
DescribeIndexRequest::DescribeIndexRequest(const std::string& table_name, ::milvus::grpc::IndexParam* index_param)
: GrpcBaseRequest(INFO_REQUEST_GROUP), table_name_(table_name), index_param_(index_param) {
}
BaseRequestPtr
DescribeIndexRequest::Create(const std::string& table_name, ::milvus::grpc::IndexParam* index_param) {
return std::shared_ptr<GrpcBaseRequest>(new DescribeIndexRequest(table_name, index_param));
}
Status
DescribeIndexRequest::OnExecute() {
try {
TimeRecorder rc("DescribeIndexRequest");
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
// step 2: check table existence
engine::TableIndex index;
status = DBWrapper::DB()->DescribeIndex(table_name_, index);
if (!status.ok()) {
return status;
}
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());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

46
core/src/server/grpc_impl/request/DescribeIndexRequest.h Normal file
View File

@ -0,0 +1,46 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
namespace milvus {
namespace server {
namespace grpc {
class DescribeIndexRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const std::string& table_name, ::milvus::grpc::IndexParam* index_param);
protected:
DescribeIndexRequest(const std::string& table_name, ::milvus::grpc::IndexParam* index_param);
Status
OnExecute() override;
private:
std::string table_name_;
::milvus::grpc::IndexParam* index_param_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

73
core/src/server/grpc_impl/request/DescribeTableRequest.cpp Normal file
View File

@ -0,0 +1,73 @@
// 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 "server/grpc_impl/request/DescribeTableRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
DescribeTableRequest::DescribeTableRequest(const std::string& table_name, ::milvus::grpc::TableSchema* schema)
: GrpcBaseRequest(INFO_REQUEST_GROUP), table_name_(table_name), schema_(schema) {
}
BaseRequestPtr
DescribeTableRequest::Create(const std::string& table_name, ::milvus::grpc::TableSchema* schema) {
return std::shared_ptr<GrpcBaseRequest>(new DescribeTableRequest(table_name, schema));
}
Status
DescribeTableRequest::OnExecute() {
TimeRecorder rc("DescribeTableRequest");
try {
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
// step 2: get table info
engine::meta::TableSchema table_info;
table_info.table_id_ = table_name_;
status = DBWrapper::DB()->DescribeTable(table_info);
if (!status.ok()) {
return status;
}
schema_->set_table_name(table_info.table_id_);
schema_->set_dimension(table_info.dimension_);
schema_->set_index_file_size(table_info.index_file_size_);
schema_->set_metric_type(table_info.metric_type_);
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
rc.ElapseFromBegin("totally cost");
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

46
core/src/server/grpc_impl/request/DescribeTableRequest.h Normal file
View File

@ -0,0 +1,46 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
namespace milvus {
namespace server {
namespace grpc {
class DescribeTableRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const std::string& table_name, ::milvus::grpc::TableSchema* schema);
protected:
DescribeTableRequest(const std::string& table_name, ::milvus::grpc::TableSchema* schema);
Status
OnExecute() override;
private:
std::string table_name_;
::milvus::grpc::TableSchema* schema_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

76
core/src/server/grpc_impl/request/DropIndexRequest.cpp Normal file
View File

@ -0,0 +1,76 @@
// 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 "server/grpc_impl/request/DropIndexRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
DropIndexRequest::DropIndexRequest(const std::string& table_name)
: GrpcBaseRequest(DDL_DML_REQUEST_GROUP), table_name_(table_name) {
}
BaseRequestPtr
DropIndexRequest::Create(const std::string& table_name) {
return std::shared_ptr<GrpcBaseRequest>(new DropIndexRequest(table_name));
}
Status
DropIndexRequest::OnExecute() {
try {
TimeRecorder rc("DropIndexRequest");
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
bool has_table = false;
status = DBWrapper::DB()->HasTable(table_name_, has_table);
if (!status.ok()) {
return status;
}
if (!has_table) {
return Status(SERVER_TABLE_NOT_EXIST, TableNotExistMsg(table_name_));
}
// step 2: check table existence
status = DBWrapper::DB()->DropIndex(table_name_);
if (!status.ok()) {
return status;
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

45
core/src/server/grpc_impl/request/DropIndexRequest.h Normal file
View File

@ -0,0 +1,45 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
namespace milvus {
namespace server {
namespace grpc {
class DropIndexRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const std::string& table_name);
protected:
explicit DropIndexRequest(const std::string& table_name);
Status
OnExecute() override;
private:
std::string table_name_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

63
core/src/server/grpc_impl/request/DropPartitionRequest.cpp Normal file
View File

@ -0,0 +1,63 @@
// 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 "server/grpc_impl/request/DropPartitionRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
DropPartitionRequest::DropPartitionRequest(const ::milvus::grpc::PartitionParam* partition_param)
: GrpcBaseRequest(DDL_DML_REQUEST_GROUP), partition_param_(partition_param) {
}
BaseRequestPtr
DropPartitionRequest::Create(const ::milvus::grpc::PartitionParam* partition_param) {
return std::shared_ptr<GrpcBaseRequest>(new DropPartitionRequest(partition_param));
}
Status
DropPartitionRequest::OnExecute() {
if (!partition_param_->partition_name().empty()) {
auto status = ValidationUtil::ValidateTableName(partition_param_->partition_name());
if (!status.ok()) {
return status;
}
return DBWrapper::DB()->DropPartition(partition_param_->partition_name());
} else {
auto status = ValidationUtil::ValidateTableName(partition_param_->table_name());
if (!status.ok()) {
return status;
}
status = ValidationUtil::ValidatePartitionTags({partition_param_->tag()});
if (!status.ok()) {
return status;
}
return DBWrapper::DB()->DropPartitionByTag(partition_param_->table_name(), partition_param_->tag());
}
}
} // namespace grpc
} // namespace server
} // namespace milvus

43
core/src/server/grpc_impl/request/DropPartitionRequest.h Normal file
View File

@ -0,0 +1,43 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
namespace milvus {
namespace server {
namespace grpc {
class DropPartitionRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const ::milvus::grpc::PartitionParam* partition_param);
protected:
explicit DropPartitionRequest(const ::milvus::grpc::PartitionParam* partition_param);
Status
OnExecute() override;
private:
const ::milvus::grpc::PartitionParam* partition_param_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

82
core/src/server/grpc_impl/request/DropTableRequest.cpp Normal file
View File

@ -0,0 +1,82 @@
// 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 "server/grpc_impl/request/DropTableRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
#include <vector>
namespace milvus {
namespace server {
namespace grpc {
DropTableRequest::DropTableRequest(const std::string& table_name)
: GrpcBaseRequest(DDL_DML_REQUEST_GROUP), table_name_(table_name) {
}
BaseRequestPtr
DropTableRequest::Create(const std::string& table_name) {
return std::shared_ptr<GrpcBaseRequest>(new DropTableRequest(table_name));
}
Status
DropTableRequest::OnExecute() {
try {
TimeRecorder rc("DropTableRequest");
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
// step 2: check table existence
engine::meta::TableSchema table_info;
table_info.table_id_ = table_name_;
status = DBWrapper::DB()->DescribeTable(table_info);
if (!status.ok()) {
if (status.code() == DB_NOT_FOUND) {
return Status(SERVER_TABLE_NOT_EXIST, TableNotExistMsg(table_name_));
} else {
return status;
}
}
rc.ElapseFromBegin("check validation");
// step 3: Drop table
std::vector<DB_DATE> dates;
status = DBWrapper::DB()->DropTable(table_name_, dates);
if (!status.ok()) {
return status;
}
rc.ElapseFromBegin("total cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

45
core/src/server/grpc_impl/request/DropTableRequest.h Normal file
View File

@ -0,0 +1,45 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
namespace milvus {
namespace server {
namespace grpc {
class DropTableRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const std::string& table_name);
protected:
explicit DropTableRequest(const std::string& table_name);
Status
OnExecute() override;
private:
std::string table_name_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

107
core/src/server/grpc_impl/request/GrpcBaseRequest.cpp Normal file
View File

@ -0,0 +1,107 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include "utils/CommonUtil.h"
#include "utils/Log.h"
namespace milvus {
namespace server {
namespace grpc {
constexpr int64_t DAY_SECONDS = 24 * 60 * 60;
Status
ConvertTimeRangeToDBDates(const std::vector<::milvus::grpc::Range>& range_array, std::vector<DB_DATE>& dates) {
dates.clear();
for (auto& range : range_array) {
time_t tt_start, tt_end;
tm tm_start, tm_end;
if (!CommonUtil::TimeStrToTime(range.start_value(), tt_start, tm_start)) {
return Status(SERVER_INVALID_TIME_RANGE, "Invalid time range: " + range.start_value());
}
if (!CommonUtil::TimeStrToTime(range.end_value(), tt_end, tm_end)) {
return Status(SERVER_INVALID_TIME_RANGE, "Invalid time range: " + range.start_value());
}
int64_t days = (tt_end - tt_start) / DAY_SECONDS;
if (days <= 0) {
return Status(SERVER_INVALID_TIME_RANGE,
"Invalid time range: The start-date should be smaller than end-date!");
}
// range: [start_day, end_day)
for (int64_t i = 0; i < days; i++) {
time_t tt_day = tt_start + DAY_SECONDS * i;
tm tm_day;
CommonUtil::ConvertTime(tt_day, tm_day);
int64_t date = tm_day.tm_year * 10000 + tm_day.tm_mon * 100 + tm_day.tm_mday; // according to db logic
dates.push_back(date);
}
}
return Status::OK();
}
GrpcBaseRequest::GrpcBaseRequest(const std::string& request_group, bool async)
: request_group_(request_group), async_(async), done_(false) {
}
GrpcBaseRequest::~GrpcBaseRequest() {
WaitToFinish();
}
Status
GrpcBaseRequest::Execute() {
status_ = OnExecute();
Done();
return status_;
}
void
GrpcBaseRequest::Done() {
done_ = true;
finish_cond_.notify_all();
}
Status
GrpcBaseRequest::SetStatus(ErrorCode error_code, const std::string& error_msg) {
status_ = Status(error_code, error_msg);
SERVER_LOG_ERROR << error_msg;
return status_;
}
std::string
GrpcBaseRequest::TableNotExistMsg(const std::string& table_name) {
return "Table " + table_name +
" not exist. Use milvus.has_table to verify whether the table exists. You also can check if the table name "
"exists.";
}
Status
GrpcBaseRequest::WaitToFinish() {
std::unique_lock<std::mutex> lock(finish_mtx_);
finish_cond_.wait(lock, [this] { return done_; });
return status_;
}
} // namespace grpc
} // namespace server
} // namespace milvus

101
core/src/server/grpc_impl/request/GrpcBaseRequest.h Normal file
View File

@ -0,0 +1,101 @@
// 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/meta/MetaTypes.h"
#include "grpc/gen-milvus/milvus.grpc.pb.h"
#include "grpc/gen-status/status.grpc.pb.h"
#include "grpc/gen-status/status.pb.h"
#include "utils/Status.h"
#include <condition_variable>
//#include <gperftools/profiler.h>
#include <memory>
#include <string>
#include <thread>
#include <vector>
namespace milvus {
namespace server {
namespace grpc {
static const char* DQL_REQUEST_GROUP = "dql";
static const char* DDL_DML_REQUEST_GROUP = "ddl_dml";
static const char* INFO_REQUEST_GROUP = "info";
using DB_DATE = milvus::engine::meta::DateT;
Status
ConvertTimeRangeToDBDates(const std::vector<::milvus::grpc::Range>& range_array, std::vector<DB_DATE>& dates);
class GrpcBaseRequest {
protected:
explicit GrpcBaseRequest(const std::string& request_group, bool async = false);
virtual ~GrpcBaseRequest();
public:
Status
Execute();
void
Done();
Status
WaitToFinish();
std::string
RequestGroup() const {
return request_group_;
}
const Status&
status() const {
return status_;
}
bool
IsAsync() const {
return async_;
}
protected:
virtual Status
OnExecute() = 0;
Status
SetStatus(ErrorCode error_code, const std::string& error_msg);
std::string
TableNotExistMsg(const std::string& table_name);
protected:
mutable std::mutex finish_mtx_;
std::condition_variable finish_cond_;
std::string request_group_;
bool async_;
bool done_;
Status status_;
};
using BaseRequestPtr = std::shared_ptr<GrpcBaseRequest>;
} // namespace grpc
} // namespace server
} // namespace milvus

66
core/src/server/grpc_impl/request/HasTableRequest.cpp Normal file
View File

@ -0,0 +1,66 @@
// 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 "server/grpc_impl/request/HasTableRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
HasTableRequest::HasTableRequest(const std::string& table_name, bool& has_table)
: GrpcBaseRequest(INFO_REQUEST_GROUP), table_name_(table_name), has_table_(has_table) {
}
BaseRequestPtr
HasTableRequest::Create(const std::string& table_name, bool& has_table) {
return std::shared_ptr<GrpcBaseRequest>(new HasTableRequest(table_name, has_table));
}
Status
HasTableRequest::OnExecute() {
try {
TimeRecorder rc("HasTableRequest");
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
// step 2: check table existence
status = DBWrapper::DB()->HasTable(table_name_, has_table_);
if (!status.ok()) {
return status;
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

46
core/src/server/grpc_impl/request/HasTableRequest.h Normal file
View File

@ -0,0 +1,46 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
namespace milvus {
namespace server {
namespace grpc {
class HasTableRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const std::string& table_name, bool& has_table);
protected:
HasTableRequest(const std::string& table_name, bool& has_table);
Status
OnExecute() override;
private:
std::string table_name_;
bool& has_table_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

170
core/src/server/grpc_impl/request/InsertRequest.cpp Normal file
View File

@ -0,0 +1,170 @@
// 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 "server/grpc_impl/request/InsertRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
#include <string>
#include <vector>
namespace milvus {
namespace server {
namespace grpc {
InsertRequest::InsertRequest(const ::milvus::grpc::InsertParam* insert_param, ::milvus::grpc::VectorIds* record_ids)
: GrpcBaseRequest(DDL_DML_REQUEST_GROUP), insert_param_(insert_param), record_ids_(record_ids) {
}
BaseRequestPtr
InsertRequest::Create(const ::milvus::grpc::InsertParam* insert_param, ::milvus::grpc::VectorIds* record_ids) {
if (insert_param == nullptr) {
SERVER_LOG_ERROR << "grpc input is null!";
return nullptr;
}
return std::shared_ptr<GrpcBaseRequest>(new InsertRequest(insert_param, record_ids));
}
Status
InsertRequest::OnExecute() {
try {
TimeRecorder rc("InsertRequest");
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(insert_param_->table_name());
if (!status.ok()) {
return status;
}
if (insert_param_->row_record_array().empty()) {
return Status(SERVER_INVALID_ROWRECORD_ARRAY,
"The vector array is empty. Make sure you have entered vector records.");
}
if (!insert_param_->row_id_array().empty()) {
if (insert_param_->row_id_array().size() != insert_param_->row_record_array_size()) {
return Status(SERVER_ILLEGAL_VECTOR_ID,
"The size of vector ID array must be equal to the size of the vector.");
}
}
// step 2: check table existence
engine::meta::TableSchema table_info;
table_info.table_id_ = insert_param_->table_name();
status = DBWrapper::DB()->DescribeTable(table_info);
if (!status.ok()) {
if (status.code() == DB_NOT_FOUND) {
return Status(SERVER_TABLE_NOT_EXIST, TableNotExistMsg(insert_param_->table_name()));
} else {
return status;
}
}
// step 3: check table flag
// all user provide id, or all internal id
bool user_provide_ids = !insert_param_->row_id_array().empty();
// user already provided id before, all insert action require user id
if ((table_info.flag_ & engine::meta::FLAG_MASK_HAS_USERID) != 0 && !user_provide_ids) {
return Status(SERVER_ILLEGAL_VECTOR_ID,
"Table vector IDs are user-defined. Please provide IDs for all vectors of this table.");
}
// user didn't provided id before, no need to provide user id
if ((table_info.flag_ & engine::meta::FLAG_MASK_NO_USERID) != 0 && user_provide_ids) {
return Status(
SERVER_ILLEGAL_VECTOR_ID,
"Table vector IDs are auto-generated. All vectors of this table must use auto-generated IDs.");
}
rc.RecordSection("check validation");
#ifdef MILVUS_ENABLE_PROFILING
std::string fname =
"/tmp/insert_" + std::to_string(this->insert_param_->row_record_array_size()) + ".profiling";
ProfilerStart(fname.c_str());
#endif
// step 4: prepare float data
std::vector<float> vec_f(insert_param_->row_record_array_size() * table_info.dimension_, 0);
// TODO(yk): change to one dimension array or use multiple-thread to copy the data
for (size_t i = 0; i < insert_param_->row_record_array_size(); i++) {
if (insert_param_->row_record_array(i).vector_data().empty()) {
return Status(SERVER_INVALID_ROWRECORD_ARRAY,
"The vector dimension must be equal to the table dimension.");
}
uint64_t vec_dim = insert_param_->row_record_array(i).vector_data().size();
if (vec_dim != table_info.dimension_) {
ErrorCode error_code = SERVER_INVALID_VECTOR_DIMENSION;
std::string error_msg = "The vector dimension must be equal to the table dimension.";
return Status(error_code, error_msg);
}
memcpy(&vec_f[i * table_info.dimension_], insert_param_->row_record_array(i).vector_data().data(),
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);
if (!insert_param_->row_id_array().empty()) {
const int64_t* src_data = insert_param_->row_id_array().data();
int64_t* target_data = vec_ids.data();
memcpy(target_data, src_data, static_cast<size_t>(sizeof(int64_t) * insert_param_->row_id_array_size()));
}
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;
}
for (int64_t id : vec_ids) {
record_ids_->add_vector_id_array(id);
}
auto ids_size = record_ids_->vector_id_array_size();
if (ids_size != vec_count) {
std::string msg =
"Add " + std::to_string(vec_count) + " vectors but only return " + std::to_string(ids_size) + " id";
return Status(SERVER_ILLEGAL_VECTOR_ID, msg);
}
// step 6: update table flag
user_provide_ids ? table_info.flag_ |= engine::meta::FLAG_MASK_HAS_USERID
: table_info.flag_ |= engine::meta::FLAG_MASK_NO_USERID;
status = DBWrapper::DB()->UpdateTableFlag(insert_param_->table_name(), table_info.flag_);
#ifdef MILVUS_ENABLE_PROFILING
ProfilerStop();
#endif
rc.RecordSection("add vectors to engine");
rc.ElapseFromBegin("total cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

44
core/src/server/grpc_impl/request/InsertRequest.h Normal file
View File

@ -0,0 +1,44 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
namespace milvus {
namespace server {
namespace grpc {
class InsertRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const ::milvus::grpc::InsertParam* insert_param, ::milvus::grpc::VectorIds* record_ids);
protected:
InsertRequest(const ::milvus::grpc::InsertParam* insert_param, ::milvus::grpc::VectorIds* record_ids);
Status
OnExecute() override;
private:
const ::milvus::grpc::InsertParam* insert_param_;
::milvus::grpc::VectorIds* record_ids_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

66
core/src/server/grpc_impl/request/PreloadTableRequest.cpp Normal file
View File

@ -0,0 +1,66 @@
// 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 "server/grpc_impl/request/PreloadTableRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
PreloadTableRequest::PreloadTableRequest(const std::string& table_name)
: GrpcBaseRequest(DQL_REQUEST_GROUP), table_name_(table_name) {
}
BaseRequestPtr
PreloadTableRequest::Create(const std::string& table_name) {
return std::shared_ptr<GrpcBaseRequest>(new PreloadTableRequest(table_name));
}
Status
PreloadTableRequest::OnExecute() {
try {
TimeRecorder rc("PreloadTableRequest");
// step 1: check arguments
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
// step 2: check table existence
status = DBWrapper::DB()->PreloadTable(table_name_);
if (!status.ok()) {
return status;
}
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

45
core/src/server/grpc_impl/request/PreloadTableRequest.h Normal file
View File

@ -0,0 +1,45 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
namespace milvus {
namespace server {
namespace grpc {
class PreloadTableRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const std::string& table_name);
protected:
explicit PreloadTableRequest(const std::string& table_name);
Status
OnExecute() override;
private:
std::string table_name_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

188
core/src/server/grpc_impl/request/SearchRequest.cpp Normal file
View File

@ -0,0 +1,188 @@
// 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 "server/grpc_impl/request/SearchRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
namespace milvus {
namespace server {
namespace grpc {
SearchRequest::SearchRequest(const ::milvus::grpc::SearchParam* search_vector_infos,
const std::vector<std::string>& file_id_array, ::milvus::grpc::TopKQueryResult* response)
: GrpcBaseRequest(DQL_REQUEST_GROUP),
search_param_(search_vector_infos),
file_id_array_(file_id_array),
topk_result_(response) {
}
BaseRequestPtr
SearchRequest::Create(const ::milvus::grpc::SearchParam* search_vector_infos,
const std::vector<std::string>& file_id_array, ::milvus::grpc::TopKQueryResult* response) {
if (search_vector_infos == nullptr) {
SERVER_LOG_ERROR << "grpc input is null!";
return nullptr;
}
return std::shared_ptr<GrpcBaseRequest>(new SearchRequest(search_vector_infos, file_id_array, response));
}
Status
SearchRequest::OnExecute() {
try {
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) + ")";
TimeRecorder rc(hdr);
// step 1: check table name
std::string table_name_ = search_param_->table_name();
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
// step 2: check table existence
engine::meta::TableSchema table_info;
table_info.table_id_ = table_name_;
status = DBWrapper::DB()->DescribeTable(table_info);
if (!status.ok()) {
if (status.code() == DB_NOT_FOUND) {
return Status(SERVER_TABLE_NOT_EXIST, TableNotExistMsg(table_name_));
} else {
return status;
}
}
// step 3: check search parameter
status = ValidationUtil::ValidateSearchTopk(top_k, table_info);
if (!status.ok()) {
return status;
}
status = ValidationUtil::ValidateSearchNprobe(nprobe, table_info);
if (!status.ok()) {
return status;
}
if (search_param_->query_record_array().empty()) {
return Status(SERVER_INVALID_ROWRECORD_ARRAY,
"The vector array is empty. Make sure you have entered vector records.");
}
// step 4: check date range, and convert to db dates
std::vector<DB_DATE> dates;
std::vector<::milvus::grpc::Range> range_array;
for (size_t i = 0; i < search_param_->query_range_array_size(); i++) {
range_array.emplace_back(search_param_->query_range_array(i));
}
status = ConvertTimeRangeToDBDates(range_array, dates);
if (!status.ok()) {
return status;
}
rc.RecordSection("check validation");
// step 5: prepare float data
auto record_array_size = search_param_->query_record_array_size();
std::vector<float> vec_f(record_array_size * table_info.dimension_, 0);
for (size_t i = 0; i < record_array_size; i++) {
if (search_param_->query_record_array(i).vector_data().empty()) {
return Status(SERVER_INVALID_ROWRECORD_ARRAY,
"The vector dimension must be equal to the table dimension.");
}
uint64_t query_vec_dim = search_param_->query_record_array(i).vector_data().size();
if (query_vec_dim != table_info.dimension_) {
ErrorCode error_code = SERVER_INVALID_VECTOR_DIMENSION;
std::string error_msg = "The vector dimension must be equal to the table dimension.";
return Status(error_code, error_msg);
}
memcpy(&vec_f[i * table_info.dimension_], search_param_->query_record_array(i).vector_data().data(),
table_info.dimension_ * sizeof(float));
}
rc.RecordSection("prepare vector data");
// step 6: search vectors
engine::ResultIds result_ids;
engine::ResultDistances result_distances;
auto record_count = (uint64_t)search_param_->query_record_array().size();
#ifdef MILVUS_ENABLE_PROFILING
std::string fname =
"/tmp/search_nq_" + std::to_string(this->search_param_->query_record_array_size()) + ".profiling";
ProfilerStart(fname.c_str());
#endif
if (file_id_array_.empty()) {
std::vector<std::string> partition_tags;
for (size_t i = 0; i < search_param_->partition_tag_array_size(); i++) {
partition_tags.emplace_back(search_param_->partition_tag_array(i));
}
status = ValidationUtil::ValidatePartitionTags(partition_tags);
if (!status.ok()) {
return status;
}
status = DBWrapper::DB()->Query(table_name_, partition_tags, (size_t)top_k, record_count, nprobe,
vec_f.data(), dates, result_ids, result_distances);
} else {
status = DBWrapper::DB()->QueryByFileID(table_name_, file_id_array_, (size_t)top_k, record_count, nprobe,
vec_f.data(), dates, result_ids, result_distances);
}
#ifdef MILVUS_ENABLE_PROFILING
ProfilerStop();
#endif
rc.RecordSection("search vectors from engine");
if (!status.ok()) {
return status;
}
if (result_ids.empty()) {
return Status::OK(); // empty table
}
// step 7: construct result array
topk_result_->set_row_num(record_count);
topk_result_->mutable_ids()->Resize(static_cast<int>(result_ids.size()), -1);
memcpy(topk_result_->mutable_ids()->mutable_data(), result_ids.data(), result_ids.size() * sizeof(int64_t));
topk_result_->mutable_distances()->Resize(static_cast<int>(result_distances.size()), 0.0);
memcpy(topk_result_->mutable_distances()->mutable_data(), result_distances.data(),
result_distances.size() * sizeof(float));
// step 8: print time cost percent
rc.RecordSection("construct result and send");
rc.ElapseFromBegin("totally cost");
} catch (std::exception& ex) {
return Status(SERVER_UNEXPECTED_ERROR, ex.what());
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

50
core/src/server/grpc_impl/request/SearchRequest.h Normal file
View File

@ -0,0 +1,50 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
#include <vector>
namespace milvus {
namespace server {
namespace grpc {
class SearchRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const ::milvus::grpc::SearchParam* search_param, const std::vector<std::string>& file_id_array,
::milvus::grpc::TopKQueryResult* response);
protected:
SearchRequest(const ::milvus::grpc::SearchParam* search_param, const std::vector<std::string>& file_id_array,
::milvus::grpc::TopKQueryResult* response);
Status
OnExecute() override;
private:
const ::milvus::grpc::SearchParam* search_param_;
std::vector<std::string> file_id_array_;
::milvus::grpc::TopKQueryResult* topk_result_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

65
core/src/server/grpc_impl/request/ShowPartitionsRequest.cpp Normal file
View File

@ -0,0 +1,65 @@
// 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 "server/grpc_impl/request/ShowPartitionsRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include "utils/ValidationUtil.h"
#include <memory>
#include <vector>
namespace milvus {
namespace server {
namespace grpc {
ShowPartitionsRequest::ShowPartitionsRequest(const std::string& table_name,
::milvus::grpc::PartitionList* partition_list)
: GrpcBaseRequest(INFO_REQUEST_GROUP), table_name_(table_name), partition_list_(partition_list) {
}
BaseRequestPtr
ShowPartitionsRequest::Create(const std::string& table_name, ::milvus::grpc::PartitionList* partition_list) {
return std::shared_ptr<GrpcBaseRequest>(new ShowPartitionsRequest(table_name, partition_list));
}
Status
ShowPartitionsRequest::OnExecute() {
auto status = ValidationUtil::ValidateTableName(table_name_);
if (!status.ok()) {
return status;
}
std::vector<engine::meta::TableSchema> schema_array;
auto statuts = DBWrapper::DB()->ShowPartitions(table_name_, schema_array);
if (!statuts.ok()) {
return statuts;
}
for (auto& schema : schema_array) {
::milvus::grpc::PartitionParam* param = partition_list_->add_partition_array();
param->set_table_name(schema.owner_table_);
param->set_partition_name(schema.table_id_);
param->set_tag(schema.partition_tag_);
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

46
core/src/server/grpc_impl/request/ShowPartitionsRequest.h Normal file
View File

@ -0,0 +1,46 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
#include <string>
namespace milvus {
namespace server {
namespace grpc {
class ShowPartitionsRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(const std::string& table_name, ::milvus::grpc::PartitionList* partition_list);
protected:
ShowPartitionsRequest(const std::string& table_name, ::milvus::grpc::PartitionList* partition_list);
Status
OnExecute() override;
private:
std::string table_name_;
::milvus::grpc::PartitionList* partition_list_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

55
core/src/server/grpc_impl/request/ShowTablesRequest.cpp 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.
#include "server/grpc_impl/request/ShowTablesRequest.h"
#include "server/DBWrapper.h"
#include "utils/Log.h"
#include "utils/TimeRecorder.h"
#include <memory>
#include <vector>
namespace milvus {
namespace server {
namespace grpc {
ShowTablesRequest::ShowTablesRequest(::milvus::grpc::TableNameList* table_name_list)
: GrpcBaseRequest(INFO_REQUEST_GROUP), table_name_list_(table_name_list) {
}
BaseRequestPtr
ShowTablesRequest::Create(::milvus::grpc::TableNameList* table_name_list) {
return std::shared_ptr<GrpcBaseRequest>(new ShowTablesRequest(table_name_list));
}
Status
ShowTablesRequest::OnExecute() {
std::vector<engine::meta::TableSchema> schema_array;
auto statuts = DBWrapper::DB()->AllTables(schema_array);
if (!statuts.ok()) {
return statuts;
}
for (auto& schema : schema_array) {
table_name_list_->add_table_names(schema.table_id_);
}
return Status::OK();
}
} // namespace grpc
} // namespace server
} // namespace milvus

43
core/src/server/grpc_impl/request/ShowTablesRequest.h Normal file
View File

@ -0,0 +1,43 @@
// 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 "server/grpc_impl/request/GrpcBaseRequest.h"
namespace milvus {
namespace server {
namespace grpc {
class ShowTablesRequest : public GrpcBaseRequest {
public:
static BaseRequestPtr
Create(::milvus::grpc::TableNameList* table_name_list);
protected:
explicit ShowTablesRequest(::milvus::grpc::TableNameList* table_name_list);
Status
OnExecute() override;
private:
::milvus::grpc::TableNameList* table_name_list_;
};
} // namespace grpc
} // namespace server
} // namespace milvus

0
core/src/config.h.in → core/src/version.h.in
View File

8
core/unittest/CMakeLists.txt
View File

@ -66,7 +66,13 @@ set(thirdparty_files
)
aux_source_directory(${MILVUS_ENGINE_SRC}/server server_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/server/grpc_impl grpc_server_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/server/grpc_impl/request grpc_request_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/server/grpc_impl grpc_impl_files)
set(grpc_server_files
${grpc_request_files}
${grpc_impl_files}
)
aux_source_directory(${MILVUS_ENGINE_SRC}/utils utils_files)
aux_source_directory(${MILVUS_ENGINE_SRC}/wrapper wrapper_files)

28
core/unittest/server/test_rpc.cpp
View File

@ -22,8 +22,8 @@
#include "server/Server.h"
#include "server/grpc_impl/GrpcRequestHandler.h"
#include "server/grpc_impl/GrpcRequestScheduler.h"
#include "server/grpc_impl/GrpcRequestTask.h"
#include "src/config.h"
#include "server/grpc_impl/request/GrpcBaseRequest.h"
#include "src/version.h"
#include "grpc/gen-milvus/milvus.grpc.pb.h"
#include "grpc/gen-status/status.pb.h"
@ -453,20 +453,20 @@ TEST_F(RpcHandlerTest, DELETE_BY_RANGE_TEST) {
//////////////////////////////////////////////////////////////////////
namespace {
class DummyTask : public milvus::server::grpc::GrpcBaseTask {
class DummyRequest : public milvus::server::grpc::GrpcBaseRequest {
public:
milvus::Status
OnExecute() override {
return milvus::Status::OK();
}
static milvus::server::grpc::BaseTaskPtr
static milvus::server::grpc::BaseRequestPtr
Create(std::string& dummy) {
return std::shared_ptr<milvus::server::grpc::GrpcBaseTask>(new DummyTask(dummy));
return std::shared_ptr<milvus::server::grpc::GrpcBaseRequest>(new DummyRequest(dummy));
}
public:
explicit DummyTask(std::string& dummy) : GrpcBaseTask(dummy) {
explicit DummyRequest(std::string& dummy) : GrpcBaseRequest(dummy) {
}
};
@ -475,27 +475,27 @@ class RpcSchedulerTest : public testing::Test {
void
SetUp() override {
std::string dummy = "dql";
task_ptr = std::make_shared<DummyTask>(dummy);
request_ptr = std::make_shared<DummyRequest>(dummy);
}
std::shared_ptr<DummyTask> task_ptr;
std::shared_ptr<DummyRequest> request_ptr;
};
} // namespace
TEST_F(RpcSchedulerTest, BASE_TASK_TEST) {
auto status = task_ptr->Execute();
auto status = request_ptr->Execute();
ASSERT_TRUE(status.ok());
milvus::server::grpc::GrpcRequestScheduler::GetInstance().Start();
::milvus::grpc::Status grpc_status;
std::string dummy = "dql";
milvus::server::grpc::BaseTaskPtr base_task_ptr = DummyTask::Create(dummy);
milvus::server::grpc::GrpcRequestScheduler::GetInstance().ExecTask(base_task_ptr, &grpc_status);
milvus::server::grpc::BaseRequestPtr base_task_ptr = DummyRequest::Create(dummy);
milvus::server::grpc::GrpcRequestScheduler::GetInstance().ExecRequest(base_task_ptr, &grpc_status);
milvus::server::grpc::GrpcRequestScheduler::GetInstance().ExecuteTask(task_ptr);
task_ptr = nullptr;
milvus::server::grpc::GrpcRequestScheduler::GetInstance().ExecuteTask(task_ptr);
milvus::server::grpc::GrpcRequestScheduler::GetInstance().ExecuteRequest(request_ptr);
request_ptr = nullptr;
milvus::server::grpc::GrpcRequestScheduler::GetInstance().ExecuteRequest(request_ptr);
milvus::server::grpc::GrpcRequestScheduler::GetInstance().Stop();
}