admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-28 22:45:26 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix proxy logic

Browse Source 
Signed-off-by: zhenshan.cao <zhenshan.cao@zilliz.com>
This commit is contained in:
zhenshan.cao 2020-08-24 14:24:06 +08:00 committed by yefu.chen
parent b8f012d8f0
commit 383da48ba1
30 changed files with 336 additions and 1124 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
proxy/src/CMakeLists.txt
View File

@ -22,10 +22,12 @@ add_subdirectory( tracing )
add_subdirectory( utils )
add_subdirectory( config )
add_subdirectory( query )
add_subdirectory( db ) # target milvus_engine
add_subdirectory( log )
add_subdirectory( server )
set(link_lib
milvus_engine
config
query
utils
@ -44,6 +46,10 @@ set( GRPC_LIB libprotobuf
grpc++
)
set( BOOST_LIB libboost_system.a
libboost_filesystem.a
libboost_serialization.a
)
set( THIRD_PARTY_LIBS
yaml-cpp
@ -55,6 +61,7 @@ target_link_libraries( server
PUBLIC ${link_lib}
tracing
${THIRD_PARTY_LIBS}
${BOOST_LIB}
)
# **************************** Get&Print Include Directories ****************************

21
proxy/src/db/CMakeLists.txt Normal file
View File

@ -0,0 +1,21 @@
#-------------------------------------------------------------------------------
# Copyright (C) 2019-2020 Zilliz. All rights reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
# with the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software distributed under the License
# is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
# or implied. See the License for the specific language governing permissions and limitations under the License.
#-------------------------------------------------------------------------------
# **************************** Engine Source Files ****************************
aux_source_directory( ${MILVUS_ENGINE_SRC}/db DB_MAIN_FILES )
set( ENGINE_FILES ${DB_MAIN_FILES})
# **************************** Add Target milvus engine ****************************
add_library( milvus_engine STATIC)
target_sources( milvus_engine PRIVATE ${ENGINE_FILES})

33
proxy/src/db/Constants.h Normal file
View File

@ -0,0 +1,33 @@
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License.
#pragma once
#include <stdint.h>
namespace milvus {
namespace engine {
constexpr int64_t KB = 1LL << 10;
constexpr int64_t MB = 1LL << 20;
constexpr int64_t GB = 1LL << 30;
constexpr int64_t TB = 1LL << 40;
constexpr int64_t MAX_TABLE_FILE_MEM = 128 * MB;
constexpr int64_t MAX_NAME_LENGTH = 255;
constexpr int64_t MAX_DIMENSION = 32768;
constexpr int32_t MAX_SEGMENT_ROW_COUNT = 4 * 1024 * 1024;
constexpr int64_t DEFAULT_SEGMENT_ROW_COUNT = 100000; // default row count per segment when creating collection
constexpr int64_t MAX_INSERT_DATA_SIZE = 256 * MB;
} // namespace engine
} // namespace milvus

36
proxy/src/db/Types.cpp Normal file
View File

@ -0,0 +1,36 @@
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License.
#include "db/Types.h"
// #include "knowhere/index/vector_index/helpers/IndexParameter.h"
namespace milvus {
namespace engine {
const char* FIELD_UID = "_id";
const char* ELEMENT_RAW_DATA = "_raw";
const char* ELEMENT_BLOOM_FILTER = "_blf";
const char* ELEMENT_DELETED_DOCS = "_del";
const char* ELEMENT_INDEX_COMPRESS = "_compress";
const char* PARAM_UID_AUTOGEN = "auto_id";
// const char* PARAM_DIMENSION = knowhere::meta::DIM;
const char* PARAM_INDEX_TYPE = "index_type";
// const char* PARAM_INDEX_METRIC_TYPE = knowhere::Metric::TYPE;
const char* PARAM_INDEX_EXTRA_PARAMS = "params";
const char* PARAM_SEGMENT_ROW_COUNT = "segment_row_count";
const char* DEFAULT_STRUCTURED_INDEX = "SORTED"; // this string should be defined in knowhere::IndexEnum
const char* DEFAULT_PARTITON_TAG = "_default";
} // namespace engine
} // namespace milvus

200
proxy/src/db/Types.h Normal file
View File

@ -0,0 +1,200 @@
// Copyright (C) 2019-2020 Zilliz. All rights reserved.
//
// Licensed under the Apache License, Version 2.0 (the "License"); you may not use this file except in compliance
// with the License. You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software distributed under the License
// is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
// or implied. See the License for the specific language governing permissions and limitations under the License.
#pragma once
// #include <faiss/Index.h>
#include <cstdint>
#include <map>
#include <memory>
#include <set>
#include <string>
#include <unordered_map>
#include <utility>
#include <vector>
// #include "cache/DataObj.h"
#include "db/Constants.h"
// #include "knowhere/index/vector_index/VecIndex.h"
#include "utils/Json.h"
namespace milvus {
namespace engine {
extern const char* FIELD_UID;
extern const char* ELEMENT_RAW_DATA;
extern const char* ELEMENT_BLOOM_FILTER;
extern const char* ELEMENT_DELETED_DOCS;
extern const char* ELEMENT_INDEX_COMPRESS;
extern const char* PARAM_UID_AUTOGEN;
extern const char* PARAM_DIMENSION;
extern const char* PARAM_INDEX_TYPE;
extern const char* PARAM_INDEX_METRIC_TYPE;
extern const char* PARAM_INDEX_EXTRA_PARAMS;
extern const char* PARAM_SEGMENT_ROW_COUNT;
extern const char* DEFAULT_STRUCTURED_INDEX;
extern const char* DEFAULT_PARTITON_TAG;
///////////////////////////////////////////////////////////////////////////////////////////////////
using id_t = int64_t;
using offset_t = int32_t;
using date_t = int32_t;
using IDNumbers = std::vector<id_t>;
// using VectorDistance = faiss::Index::distance_t;
using VectorDistance = float;
using VectorDistances = std::vector<VectorDistance>;
// using ResultIds = std::vector<faiss::Index::idx_t>;
using ResultIds = std::vector<int64_t>;
using ResultDistances = std::vector<VectorDistance>;
///////////////////////////////////////////////////////////////////////////////////////////////////
enum class DataType {
NONE = 0,
BOOL = 1,
INT8 = 2,
INT16 = 3,
INT32 = 4,
INT64 = 5,
FLOAT = 10,
DOUBLE = 11,
STRING = 20,
VECTOR_BINARY = 100,
VECTOR_FLOAT = 101,
};
///////////////////////////////////////////////////////////////////////////////////////////////////
enum class FieldElementType {
FET_NONE = 0,
FET_RAW = 1,
FET_BLOOM_FILTER = 2,
FET_DELETED_DOCS = 3,
FET_INDEX = 4,
FET_COMPRESS_SQ8 = 5,
};
///////////////////////////////////////////////////////////////////////////////////////////////////
// class BinaryData : public cache::DataObj {
// public:
// int64_t
// Size() {
// return data_.size();
// }
// public:
// std::vector<uint8_t> data_;
// };
// using BinaryDataPtr = std::shared_ptr<BinaryData>;
///////////////////////////////////////////////////////////////////////////////////////////////////
// class VaribleData : public cache::DataObj {
// public:
// int64_t
// Size() {
// return data_.size() + offset_.size() * sizeof(int64_t);
// }
// public:
// std::vector<uint8_t> data_;
// std::vector<int64_t> offset_;
// };
// using VaribleDataPtr = std::shared_ptr<VaribleData>;
///////////////////////////////////////////////////////////////////////////////////////////////////
// using FIELD_TYPE_MAP = std::unordered_map<std::string, DataType>;
// using FIELD_WIDTH_MAP = std::unordered_map<std::string, int64_t>;
// using FIXEDX_FIELD_MAP = std::unordered_map<std::string, BinaryDataPtr>;
// using VARIABLE_FIELD_MAP = std::unordered_map<std::string, VaribleDataPtr>;
// using VECTOR_INDEX_MAP = std::unordered_map<std::string, knowhere::VecIndexPtr>;
// using STRUCTURED_INDEX_MAP = std::unordered_map<std::string, knowhere::IndexPtr>;
///////////////////////////////////////////////////////////////////////////////////////////////////
struct DataChunk {
int64_t count_ = 0;
// FIXEDX_FIELD_MAP fixed_fields_;
// VARIABLE_FIELD_MAP variable_fields_;
};
using DataChunkPtr = std::shared_ptr<DataChunk>;
///////////////////////////////////////////////////////////////////////////////////////////////////
struct CollectionIndex {
std::string index_name_;
std::string index_type_;
std::string metric_name_;
milvus::json extra_params_ = {{"nlist", 2048}};
};
///////////////////////////////////////////////////////////////////////////////////////////////////
struct VectorsData {
uint64_t vector_count_ = 0;
std::vector<float> float_data_;
std::vector<uint8_t> binary_data_;
IDNumbers id_array_;
};
///////////////////////////////////////////////////////////////////////////////////////////////////
struct AttrsData {
uint64_t attr_count_ = 0;
std::unordered_map<std::string, engine::DataType> attr_type_;
std::unordered_map<std::string, std::vector<uint8_t>> attr_data_;
IDNumbers id_array_;
};
///////////////////////////////////////////////////////////////////////////////////////////////////
struct QueryResult {
uint64_t row_num_;
engine::ResultIds result_ids_;
engine::ResultDistances result_distances_;
engine::DataChunkPtr data_chunk_;
};
using QueryResultPtr = std::shared_ptr<QueryResult>;
///////////////////////////////////////////////////////////////////////////////////////////////////
struct DBMetaOptions {
std::string path_;
std::string backend_uri_;
}; // DBMetaOptions
///////////////////////////////////////////////////////////////////////////////////////////////////
struct DBOptions {
typedef enum { SINGLE = 0, CLUSTER_READONLY, CLUSTER_WRITABLE } MODE;
DBMetaOptions meta_;
int mode_ = MODE::SINGLE;
size_t insert_buffer_size_ = 4 * GB;
bool insert_cache_immediately_ = false;
int64_t auto_flush_interval_ = 1;
bool metric_enable_ = false;
// wal relative configurations
bool wal_enable_ = false;
int64_t buffer_size_ = 256;
std::string mxlog_path_ = "/tmp/milvus/wal/";
// transcript configurations
bool transcript_enable_ = false;
std::string replay_script_path_; // for replay
}; // Options
} // namespace engine
} // namespace milvus

2
proxy/src/query/GeneralQuery.h
View File

@ -18,7 +18,7 @@
#include <unordered_map>
#include <vector>
// #include "db/Types.h"
#include "db/Types.h"
#include "utils/Json.h"
namespace milvus {

2
proxy/src/server/CMakeLists.txt
View File

@ -20,12 +20,14 @@ set( GRPC_SERVICE_FILES ${MILVUS_ENGINE_SRC}/grpc/gen-milvus/milvus.grpc.pb.cc
aux_source_directory( ${MILVUS_ENGINE_SRC}/server SERVER_SERVICE_FILES )
aux_source_directory( ${MILVUS_ENGINE_SRC}/server/init SERVER_INIT_FILES )
aux_source_directory( ${MILVUS_ENGINE_SRC}/server/delivery/request DELIVERY_REQUEST_FILES )
aux_source_directory( ${MILVUS_ENGINE_SRC}/server/delivery/strategy DELIVERY_STRATEGY_FILES )
aux_source_directory( ${MILVUS_ENGINE_SRC}/server/delivery DELIVERY_FILES )
set( SERVER_FILES ${SERVER_INIT_FILES}
${SERVER_SERVICE_FILES}
${SERVER_INIT_FILES}
${DELIVERY_REQUEST_FILES}
${DELIVERY_STRATEGY_FILES}
${DELIVERY_FILES}
)

8
proxy/src/server/Server.cpp
View File

@ -19,7 +19,7 @@
#include <unordered_map>
#include "config/ServerConfig.h"
#include "tracing/TracerUtil.h"
#include "log/LogMgr.h"
// #include "scheduler/SchedInst.h"
#include "server/grpc_impl/GrpcServer.h"
@ -237,8 +237,8 @@ Server::Start() {
<< std::string(15, '*') << "Config in memory" << std::string(15, '*') << "\n\n"
<< ConfigMgr::GetInstance().Dump();
server::Metrics::GetInstance().Init();
server::SystemInfo::GetInstance().Init();
// server::Metrics::GetInstance().Init();
// server::SystemInfo::GetInstance().Init();
return StartService();
} catch (std::exception& ex) {
@ -282,7 +282,7 @@ Server::Stop() {
Status
Server::StartService() {
Status stat;
stat = engine::KnowhereResource::Initialize();
// stat = engine::KnowhereResource::Initialize();
if (!stat.ok()) {
LOG_SERVER_ERROR_ << "KnowhereResource initialize fail: " << stat.message();
goto FAIL;

10
proxy/src/server/ValidationUtil.cpp
View File

@ -10,7 +10,7 @@
// or implied. See the License for the specific language governing permissions and limitations under the License.
#include "server/ValidationUtil.h"
// #include "db/Constants.h"
#include "db/Constants.h"
// #include "db/Utils.h"
// #include "knowhere/index/vector_index/ConfAdapter.h"
// #include "knowhere/index/vector_index/helpers/IndexParameter.h"
@ -103,11 +103,11 @@ ValidatePartitionTags(const std::vector<std::string>& partition_tags) {
return Status::OK();
}
Status
ValidateInsertDataSize(const engine::DataChunkPtr& data) {
// Status
// ValidateInsertDataSize(const engine::DataChunkPtr& data) {
return Status::OK();
}
// return Status::OK();
// }
} // namespace server
} // namespace milvus

6
proxy/src/server/ValidationUtil.h
View File

@ -11,7 +11,7 @@
#pragma once
// #include "db/Types.h"
#include "db/Types.h"
#include "utils/Json.h"
#include "utils/Status.h"
@ -50,7 +50,7 @@ ValidateSearchTopk(int64_t top_k);
extern Status
ValidatePartitionTags(const std::vector<std::string>& partition_tags);
extern Status
ValidateInsertDataSize(const engine::DataChunkPtr& data);
// extern Status
// ValidateInsertDataSize(const engine::DataChunkPtr& data);
} // namespace server
} // namespace milvus

8
proxy/src/server/delivery/ReqHandler.cpp
View File

@ -159,9 +159,9 @@ ReqHandler::Insert(const ContextPtr& context, const std::string& collection_name
Status
ReqHandler::GetEntityByID(const ContextPtr& context, const std::string& collection_name, const engine::IDNumbers& ids,
std::vector<std::string>& field_names, std::vector<bool>& valid_row,
engine::snapshot::FieldElementMappings& field_mappings, engine::DataChunkPtr& data_chunk) {
engine::DataChunkPtr& data_chunk) {
BaseReqPtr req_ptr =
GetEntityByIDReq::Create(context, collection_name, ids, field_names, valid_row, field_mappings, data_chunk);
GetEntityByIDReq::Create(context, collection_name, ids, field_names, valid_row, data_chunk);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
@ -176,8 +176,8 @@ ReqHandler::DeleteEntityByID(const ContextPtr& context, const std::string& colle
Status
ReqHandler::Search(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::FieldElementMappings& collection_mappings, engine::QueryResultPtr& result) {
BaseReqPtr req_ptr = SearchReq::Create(context, query_ptr, json_params, collection_mappings, result);
engine::QueryResultPtr& result) {
BaseReqPtr req_ptr = SearchReq::Create(context, query_ptr, json_params, result);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}

4
proxy/src/server/delivery/ReqHandler.h
View File

@ -85,14 +85,14 @@ class ReqHandler {
Status
GetEntityByID(const ContextPtr& context, const std::string& collection_name, const engine::IDNumbers& ids,
std::vector<std::string>& field_names, std::vector<bool>& valid_row,
engine::snapshot::FieldElementMappings& field_mappings, engine::DataChunkPtr& data_chunk);
engine::DataChunkPtr& data_chunk);
Status
DeleteEntityByID(const ContextPtr& context, const std::string& collection_name, const engine::IDNumbers& ids);
Status
Search(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::FieldElementMappings& collection_mappings, engine::QueryResultPtr& result);
engine::QueryResultPtr& result);
Status
ListIDInSegment(const ContextPtr& context, const std::string& collection_name, int64_t segment_id,

7
proxy/src/server/delivery/request/DescribeIndexReq.cpp
View File

@ -41,13 +41,6 @@ DescribeIndexReq::Create(const std::shared_ptr<milvus::server::Context>& context
Status
DescribeIndexReq::OnExecute() {
try {
std::string hdr = "DescribeIndexReq(collection=" + collection_name_ + ")";
TimeRecorderAuto rc(hdr);
// step 1: check arguments
STATUS_CHECK(ValidateCollectionName(collection_name_));
STATUS_CHECK(ValidateFieldName(field_name_));
return Status::OK();
}

6
proxy/src/server/delivery/request/DropCollectionReq.cpp
View File

@ -28,12 +28,6 @@ DropCollectionReq::Create(const ContextPtr& context, const std::string& collecti
Status
DropCollectionReq::OnExecute() {
try {
std::string hdr = "DropCollectionReq(collection=" + collection_name_ + ")";
TimeRecorder rc(hdr);
STATUS_CHECK(ValidateCollectionName(collection_name_));
return Status::OK();
}

3
proxy/src/server/delivery/request/DropIndexReq.cpp
View File

@ -37,9 +37,6 @@ DropIndexReq::Create(const ContextPtr& context, const std::string& collection_na
Status
DropIndexReq::OnExecute() {
try {
std::string hdr = "DropIndexReq(collection=" + collection_name_ + ")";
TimeRecorderAuto rc(hdr);
return Status::OK();
}

5
proxy/src/server/delivery/request/GetCollectionInfoReq.cpp
View File

@ -37,11 +37,6 @@ GetCollectionInfoReq::Create(const ContextPtr& context, const std::string& colle
Status
GetCollectionInfoReq::OnExecute() {
try {
std::string hdr = "GetCollectionInfoReq(collection=" + collection_name_ + ")";
TimeRecorderAuto rc(hdr);
STATUS_CHECK(ValidateCollectionName(collection_name_));
return Status::OK();

6
proxy/src/server/delivery/request/GetCollectionStatsReq.cpp
View File

@ -38,12 +38,6 @@ GetCollectionStatsReq::Create(const ContextPtr& context, const std::string& coll
Status
GetCollectionStatsReq::OnExecute() {
try {
std::string hdr = "GetCollectionStatsReq(collection=" + collection_name_ + ")";
TimeRecorderAuto rc(hdr);
STATUS_CHECK(ValidateCollectionName(collection_name_));
return Status::OK();
}

27
proxy/src/server/delivery/request/GetEntityByIDReq.cpp
View File

@ -17,7 +17,7 @@
#include "server/delivery/request/GetEntityByIDReq.h"
// #include "db/Types.h"
#include "db/Types.h"
#include "server/ValidationUtil.h"
#include "utils/TimeRecorder.h"
@ -31,46 +31,27 @@ constexpr uint64_t MAX_COUNT_RETURNED = 1000;
GetEntityByIDReq::GetEntityByIDReq(const ContextPtr& context, const std::string& collection_name,
const engine::IDNumbers& id_array, std::vector<std::string>& field_names,
std::vector<bool>& valid_row, engine::snapshot::FieldElementMappings& field_mappings,
std::vector<bool>& valid_row,
engine::DataChunkPtr& data_chunk)
: BaseReq(context, ReqType::kGetEntityByID),
collection_name_(collection_name),
id_array_(id_array),
field_names_(field_names),
valid_row_(valid_row),
field_mappings_(field_mappings),
data_chunk_(data_chunk) {
}
BaseReqPtr
GetEntityByIDReq::Create(const ContextPtr& context, const std::string& collection_name,
const engine::IDNumbers& id_array, std::vector<std::string>& field_names_,
std::vector<bool>& valid_row, engine::snapshot::FieldElementMappings& field_mappings,
std::vector<bool>& valid_row,
engine::DataChunkPtr& data_chunk) {
return std::shared_ptr<BaseReq>(
new GetEntityByIDReq(context, collection_name, id_array, field_names_, valid_row, field_mappings, data_chunk));
new GetEntityByIDReq(context, collection_name, id_array, field_names_, valid_row, data_chunk));
}
Status
GetEntityByIDReq::OnExecute() {
try {
std::string hdr = "GetEntityByIDReq(collection=" + collection_name_ + ")";
TimeRecorderAuto rc(hdr);
// step 1: check arguments
if (id_array_.empty()) {
return Status(SERVER_INVALID_ARGUMENT, "No entity id specified");
}
if (id_array_.size() > MAX_COUNT_RETURNED) {
std::string msg = "Input id array size cannot exceed: " + std::to_string(MAX_COUNT_RETURNED);
return Status(SERVER_INVALID_ARGUMENT, msg);
}
STATUS_CHECK(ValidateCollectionName(collection_name_));
// TODO(yukun) ValidateFieldNames
return Status::OK();
}

7
proxy/src/server/delivery/request/GetEntityByIDReq.h
View File

@ -24,7 +24,7 @@
#include <vector>
// #include "db/snapshot/Context.h"
// #include "db/snapshot/Resources.h"
#include "segment/Segment.h"
// #include "segment/Segment.h"
namespace milvus {
namespace server {
@ -34,12 +34,12 @@ class GetEntityByIDReq : public BaseReq {
static BaseReqPtr
Create(const ContextPtr& context, const std::string& collection_name, const engine::IDNumbers& id_array,
std::vector<std::string>& field_names, std::vector<bool>& valid_row,
engine::snapshot::FieldElementMappings& field_mappings, engine::DataChunkPtr& data_chunk);
engine::DataChunkPtr& data_chunk);
protected:
GetEntityByIDReq(const ContextPtr& context, const std::string& collection_name, const engine::IDNumbers& id_array,
std::vector<std::string>& field_names, std::vector<bool>& valid_row,
engine::snapshot::FieldElementMappings& field_mappings, engine::DataChunkPtr& data_chunk);
engine::DataChunkPtr& data_chunk);
Status
OnExecute() override;
@ -48,7 +48,6 @@ class GetEntityByIDReq : public BaseReq {
std::string collection_name_;
engine::IDNumbers id_array_;
std::vector<std::string>& field_names_;
engine::snapshot::FieldElementMappings& field_mappings_;
engine::DataChunkPtr& data_chunk_;
std::vector<bool>& valid_row_;
};

5
proxy/src/server/delivery/request/HasCollectionReq.cpp
View File

@ -29,11 +29,6 @@ HasCollectionReq::Create(const ContextPtr& context, const std::string& collectio
Status
HasCollectionReq::OnExecute() {
try {
std::string hdr = "HasCollectionReq(collection=" + collection_name_ + ")";
TimeRecorderAuto rc(hdr);
STATUS_CHECK(ValidateCollectionName(collection_name_));
return Status::OK();
}

6
proxy/src/server/delivery/request/HasPartitionReq.cpp
View File

@ -36,12 +36,6 @@ HasPartitionReq::Create(const ContextPtr& context, const std::string& collection
Status
HasPartitionReq::OnExecute() {
try {
std::string hdr = "HasPartitionReq(collection=" + collection_name_ + " tag=" + partition_tag_ + ")";
TimeRecorderAuto rc(hdr);
has_partition_ = false;
return Status::OK();
}

7
proxy/src/server/delivery/request/InsertReq.cpp
View File

@ -46,14 +46,7 @@ InsertReq::Create(const ContextPtr& context, const std::string& collection_name,
Status
InsertReq::OnExecute() {
LOG_SERVER_INFO_ << LogOut("[%s][%ld] ", "insert", 0) << "Execute InsertReq.";
try {
std::string hdr = "InsertReq(table=" + collection_name_ + ", partition_name=" + partition_name_ + ")";
TimeRecorder rc(hdr);
if (chunk_data_.empty()) {
return Status{SERVER_INVALID_ARGUMENT,
"The vector field is empty, Make sure you have entered vector records"};
}
return Status::OK();
}

5
proxy/src/server/delivery/request/ListIDInSegmentReq.cpp
View File

@ -42,12 +42,7 @@ ListIDInSegmentReq::Create(const ContextPtr& context, const std::string& collect
Status
ListIDInSegmentReq::OnExecute() {
try {
std::string hdr =
"ListIDInSegmentReq(collection=" + collection_name_ + " segment=" + std::to_string(segment_id_) + ")";
TimeRecorderAuto rc(hdr);
bool exist = false;
return Status::OK();
}

4
proxy/src/server/delivery/request/LoadCollectionReq.cpp
View File

@ -31,10 +31,6 @@ LoadCollectionReq::Create(const ContextPtr& context, const std::string& collecti
Status
LoadCollectionReq::OnExecute() {
try {
std::string hdr = "LoadCollectionReq(collection=" + collection_name_ + ")";
TimeRecorderAuto rc(hdr);
return Status::OK();
}

12
proxy/src/server/delivery/request/SearchReq.cpp
View File

@ -29,28 +29,22 @@ namespace milvus {
namespace server {
SearchReq::SearchReq(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::FieldElementMappings& field_mappings, engine::QueryResultPtr& result)
engine::QueryResultPtr& result)
: BaseReq(context, ReqType::kSearch),
query_ptr_(query_ptr),
json_params_(json_params),
field_mappings_(field_mappings),
result_(result) {
}
BaseReqPtr
SearchReq::Create(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::FieldElementMappings& field_mappings, engine::QueryResultPtr& result) {
return std::shared_ptr<BaseReq>(new SearchReq(context, query_ptr, json_params, field_mappings, result));
engine::QueryResultPtr& result) {
return std::shared_ptr<BaseReq>(new SearchReq(context, query_ptr, json_params, result));
}
Status
SearchReq::OnExecute() {
try {
std::string hdr = "SearchReq(table=" + query_ptr_->collection_id;
TimeRecorder rc(hdr);
STATUS_CHECK(ValidateCollectionName(query_ptr_->collection_id));
STATUS_CHECK(ValidatePartitionTags(query_ptr_->partitions));
return Status::OK();
}

5
proxy/src/server/delivery/request/SearchReq.h
View File

@ -26,11 +26,11 @@ class SearchReq : public BaseReq {
public:
static BaseReqPtr
Create(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::FieldElementMappings& collection_mappings, engine::QueryResultPtr& result);
engine::QueryResultPtr& result);
protected:
SearchReq(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::FieldElementMappings& collection_mappings, engine::QueryResultPtr& result);
engine::QueryResultPtr& result);
Status
OnExecute() override;
@ -38,7 +38,6 @@ class SearchReq : public BaseReq {
private:
milvus::query::QueryPtr query_ptr_;
milvus::json json_params_;
engine::snapshot::FieldElementMappings& field_mappings_;
engine::QueryResultPtr& result_;
};

2
proxy/src/server/delivery/request/Types.h
View File

@ -11,7 +11,7 @@
#pragma once
// #include "db/Types.h"
#include "db/Types.h"
#include "grpc/gen-milvus/milvus.grpc.pb.h"
#include "grpc/gen-status/status.grpc.pb.h"
#include "grpc/gen-status/status.pb.h"

958
proxy/src/server/grpc_impl/GrpcRequestHandler.cpp
View File

File diff suppressed because it is too large Load Diff

4
proxy/src/version.h
View File

@ -11,5 +11,5 @@
#define MILVUS_VERSION "0.10.0"
#define BUILD_TYPE "Release"
#define BUILD_TIME "2020-08-21 17:48.25"
#define LAST_COMMIT_ID "8c4a905ce247333b5950bd2f03cf103e34533db8"
#define BUILD_TIME "2020-08-24 14:23.26"
#define LAST_COMMIT_ID "83d51fc3196e51a3057e1fb716ee4f4be10ec0c1"

54
writer/writer.go
View File

@ -1,54 +0,0 @@
package src
import (
"context"
"github.com/czs007/suvlim/pulsar/schema"
"github.com/czs007/suvlim/storage/pkg"
"github.com/czs007/suvlim/storage/pkg/types"
)
type write_node_time_sync struct {
delete_time_sync uint64
insert_time_sync uint64
}
type write_node struct {
open_segment_id int64
next_segment_id int64
next_segment_start_time uint64
stroe *types.Store
time_sync_table *write_node_time_sync
}
func NewWriteNode(ctx context.Context, open_segment_id int64, time_sync uint64) (*write_node, error) {
ctx = context.Background()
tikv_store, err := storage.NewStore(ctx, "tikv")
write_table_time_sync := &write_node_time_sync{delete_time_sync: time_sync, insert_time_sync: time_sync}
if err != nil {
return nil, err
}
return &write_node{
stroe: tikv_store,
time_sync_table: write_table_time_sync,
}, nil
}
func (s *write_node) InsertBatchData(ctx context.Context, data []schema.InsertMsg, time_sync uint64) error {
return nil
}
func (s *write_node) DeleteBatchData(ctx context.Context, data []schema.DeleteMsg, time_sync uint64) error {
return nil
}
func (s *write_node) AddNewSegment(segment_id int64, open_time uint64) error {
return nil
}
func (s *write_node) UpdateInsertTimeSync(time_sync uint64) {
s.time_sync_table.insert_time_sync = time_sync
}
func (s *write_node) UpdateDeleteTimeSync(time_sync uint64) {
s.time_sync_table.delete_time_sync = time_sync
}