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

fix ci (#3097) 

* rename CollectionMappings to FieldElementMappings

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* code opt

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>

* code opt

Signed-off-by: yudong.cai <yudong.cai@zilliz.com>
pull/3098/head^2
Cai Yudong 2020-08-01 17:03:05 +08:00 committed by GitHub
parent c9d7d54df5
commit 4024e5cc03
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
17 changed files with 116 additions and 136 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -61,7 +61,7 @@ class DB {
virtual Status
GetCollectionInfo(const std::string& collection_name, snapshot::CollectionPtr& collection,
snapshot::CollectionMappings& fields_schema) = 0;
snapshot::FieldElementMappings& fields_schema) = 0;
virtual Status
GetCollectionStats(const std::string& collection_name, milvus::json& collection_stats) = 0;

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

@ -277,7 +277,7 @@ DBImpl::ListCollections(std::vector<std::string>& names) {
Status
DBImpl::GetCollectionInfo(const std::string& collection_name, snapshot::CollectionPtr& collection,
snapshot::CollectionMappings& fields_schema) {
snapshot::FieldElementMappings& fields_schema) {
CHECK_INITIALIZED;
snapshot::ScopedSnapshotT ss;

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

@ -56,7 +56,7 @@ class DBImpl : public DB, public ConfigObserver {
Status
GetCollectionInfo(const std::string& collection_name, snapshot::CollectionPtr& collection,
snapshot::CollectionMappings& fields_schema) override;
snapshot::FieldElementMappings& fields_schema) override;
Status
GetCollectionStats(const std::string& collection_name, milvus::json& collection_stats) override;

13
core/src/db/SnapshotHandlers.cpp
View File

@ -135,14 +135,15 @@ GetEntityByIdSegmentHandler::Handle(const snapshot::SegmentPtr& segment) {
}
segment::SegmentReader segment_reader(dir_root_, segment_visitor);
auto uid_field_visitor = segment_visitor->GetFieldVisitor(DEFAULT_UID_NAME);
std::vector<int64_t> uids;
STATUS_CHECK(segment_reader.LoadUids(uids));
// load UID's bloom filter file
segment::IdBloomFilterPtr id_bloom_filter_ptr;
STATUS_CHECK(segment_reader.LoadBloomFilter(id_bloom_filter_ptr));
std::vector<int64_t> uids;
segment::DeletedDocsPtr deleted_docs_ptr;
STATUS_CHECK(segment_reader.LoadDeletedDocs(deleted_docs_ptr));
std::vector<int64_t> offsets;
for (auto id : ids_) {
// fast check using bloom filter
@ -152,9 +153,6 @@ GetEntityByIdSegmentHandler::Handle(const snapshot::SegmentPtr& segment) {
}
// check if id really exists in uids
if (uids.empty()) {
STATUS_CHECK(segment_reader.LoadUids(uids)); // lazy load
}
auto found = std::find(uids.begin(), uids.end(), id);
if (found == uids.end()) {
valid_row_.push_back(false);
@ -163,9 +161,6 @@ GetEntityByIdSegmentHandler::Handle(const snapshot::SegmentPtr& segment) {
// check if this id is deleted
auto offset = std::distance(uids.begin(), found);
if (deleted_docs_ptr == nullptr) {
STATUS_CHECK(segment_reader.LoadDeletedDocs(deleted_docs_ptr)); // lazy load
}
if (deleted_docs_ptr) {
auto& deleted_docs = deleted_docs_ptr->GetDeletedDocs();
auto deleted = std::find(deleted_docs.begin(), deleted_docs.end(), offset);

4
core/src/db/snapshot/Context.h
View File

@ -88,12 +88,12 @@ struct OperationContext {
ToString() const;
};
using CollectionMappings = std::unordered_map<FieldPtr, std::vector<FieldElementPtr>>;
using FieldElementMappings = std::unordered_map<FieldPtr, std::vector<FieldElementPtr>>;
struct CreateCollectionContext {
CollectionPtr collection = nullptr;
CollectionMappings fields_schema;
CollectionCommitPtr collection_commit = nullptr;
FieldElementMappings fields_schema;
LSN_TYPE lsn = 0;
std::string

9
core/src/segment/SegmentReader.cpp
View File

@ -157,7 +157,7 @@ SegmentReader::LoadEntities(const std::string& field_name, const std::vector<int
engine::snapshot::GetResPath<engine::snapshot::SegmentFile>(dir_collections_, raw_visitor->GetFile());
int64_t field_width = 0;
segment_ptr_->GetFixedFieldWidth(field_name, field_width);
STATUS_CHECK(segment_ptr_->GetFixedFieldWidth(field_name, field_width));
if (field_width <= 0) {
return Status(DB_ERROR, "Invalid field width");
}
@ -184,14 +184,9 @@ SegmentReader::LoadFieldsEntities(const std::vector<std::string>& fields_name, c
data_chunk->count_ = offsets.size();
for (auto& name : fields_name) {
engine::FIXED_FIELD_DATA raw_data;
auto status = LoadEntities(name, offsets, raw_data);
if (!status.ok()) {
return status;
}
STATUS_CHECK(LoadEntities(name, offsets, raw_data));
data_chunk->fixed_fields_[name] = raw_data;
}
return Status::OK();
}

73
core/src/server/delivery/ReqHandler.cpp
View File

@ -43,37 +43,36 @@ namespace milvus {
namespace server {
Status
ReqHandler::CreateCollection(const std::shared_ptr<Context>& context, const std::string& collection_name,
std::unordered_map<std::string, FieldSchema>& fields, milvus::json& json_param) {
ReqHandler::CreateCollection(const ContextPtr& context, const std::string& collection_name, FieldsType& fields,
milvus::json& json_param) {
BaseReqPtr req_ptr = CreateCollectionReq::Create(context, collection_name, fields, json_param);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::DropCollection(const std::shared_ptr<Context>& context, const std::string& collection_name) {
ReqHandler::DropCollection(const ContextPtr& context, const std::string& collection_name) {
BaseReqPtr req_ptr = DropCollectionReq::Create(context, collection_name);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::HasCollection(const std::shared_ptr<Context>& context, const std::string& collection_name,
bool& has_collection) {
ReqHandler::HasCollection(const ContextPtr& context, const std::string& collection_name, bool& has_collection) {
BaseReqPtr req_ptr = HasCollectionReq::Create(context, collection_name, has_collection);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::ListCollections(const std::shared_ptr<Context>& context, std::vector<std::string>& collections) {
ReqHandler::ListCollections(const ContextPtr& context, std::vector<std::string>& collections) {
BaseReqPtr req_ptr = ListCollectionsReq::Create(context, collections);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::GetCollectionInfo(const std::shared_ptr<Context>& context, const std::string& collection_name,
ReqHandler::GetCollectionInfo(const ContextPtr& context, const std::string& collection_name,
CollectionSchema& collection_schema) {
BaseReqPtr req_ptr = GetCollectionInfoReq::Create(context, collection_name, collection_schema);
ReqScheduler::ExecReq(req_ptr);
@ -81,7 +80,7 @@ ReqHandler::GetCollectionInfo(const std::shared_ptr<Context>& context, const std
}
Status
ReqHandler::GetCollectionStats(const std::shared_ptr<Context>& context, const std::string& collection_name,
ReqHandler::GetCollectionStats(const ContextPtr& context, const std::string& collection_name,
std::string& collection_stats) {
BaseReqPtr req_ptr = GetCollectionStatsReq::Create(context, collection_name, collection_stats);
ReqScheduler::ExecReq(req_ptr);
@ -89,38 +88,36 @@ ReqHandler::GetCollectionStats(const std::shared_ptr<Context>& context, const st
}
Status
ReqHandler::CountEntities(const std::shared_ptr<Context>& context, const std::string& collection_name, int64_t& count) {
ReqHandler::CountEntities(const ContextPtr& context, const std::string& collection_name, int64_t& count) {
BaseReqPtr req_ptr = CountEntitiesReq::Create(context, collection_name, count);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::CreatePartition(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& tag) {
ReqHandler::CreatePartition(const ContextPtr& context, const std::string& collection_name, const std::string& tag) {
BaseReqPtr req_ptr = CreatePartitionReq::Create(context, collection_name, tag);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::DropPartition(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& tag) {
ReqHandler::DropPartition(const ContextPtr& context, const std::string& collection_name, const std::string& tag) {
BaseReqPtr req_ptr = DropPartitionReq::Create(context, collection_name, tag);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::HasPartition(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& tag, bool& has_partition) {
ReqHandler::HasPartition(const ContextPtr& context, const std::string& collection_name, const std::string& tag,
bool& has_partition) {
BaseReqPtr req_ptr = HasPartitionReq::Create(context, collection_name, tag, has_partition);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::ListPartitions(const std::shared_ptr<Context>& context, const std::string& collection_name,
ReqHandler::ListPartitions(const ContextPtr& context, const std::string& collection_name,
std::vector<std::string>& partitions) {
BaseReqPtr req_ptr = ListPartitionsReq::Create(context, collection_name, partitions);
ReqScheduler::ExecReq(req_ptr);
@ -128,43 +125,41 @@ ReqHandler::ListPartitions(const std::shared_ptr<Context>& context, const std::s
}
Status
ReqHandler::CreateIndex(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& field_name, const std::string& index_name, const milvus::json& json_params) {
ReqHandler::CreateIndex(const ContextPtr& context, const std::string& collection_name, const std::string& field_name,
const std::string& index_name, const milvus::json& json_params) {
BaseReqPtr req_ptr = CreateIndexReq::Create(context, collection_name, field_name, index_name, json_params);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::DescribeIndex(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& field_name, std::string& index_name, milvus::json& json_params) {
ReqHandler::DescribeIndex(const ContextPtr& context, const std::string& collection_name, const std::string& field_name,
std::string& index_name, milvus::json& json_params) {
BaseReqPtr req_ptr = DescribeIndexReq::Create(context, collection_name, field_name, index_name, json_params);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::DropIndex(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& field_name, const std::string& index_name) {
ReqHandler::DropIndex(const ContextPtr& context, const std::string& collection_name, const std::string& field_name,
const std::string& index_name) {
BaseReqPtr req_ptr = DropIndexReq::Create(context, collection_name, index_name, field_name);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::Insert(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& partition_name, const int64_t& row_count,
std::unordered_map<std::string, std::vector<uint8_t>>& chunk_data) {
ReqHandler::Insert(const ContextPtr& context, const std::string& collection_name, const std::string& partition_name,
const int64_t& row_count, std::unordered_map<std::string, std::vector<uint8_t>>& chunk_data) {
BaseReqPtr req_ptr = InsertReq::Create(context, collection_name, partition_name, row_count, chunk_data);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::GetEntityByID(const std::shared_ptr<Context>& context, const std::string& collection_name,
const engine::IDNumbers& ids, std::vector<std::string>& field_names,
std::vector<bool>& valid_row, engine::snapshot::CollectionMappings& field_mappings,
engine::DataChunkPtr& data_chunk) {
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) {
BaseReqPtr req_ptr =
GetEntityByIDReq::Create(context, collection_name, ids, field_names, valid_row, field_mappings, data_chunk);
ReqScheduler::ExecReq(req_ptr);
@ -172,7 +167,7 @@ ReqHandler::GetEntityByID(const std::shared_ptr<Context>& context, const std::st
}
Status
ReqHandler::DeleteEntityByID(const std::shared_ptr<Context>& context, const std::string& collection_name,
ReqHandler::DeleteEntityByID(const ContextPtr& context, const std::string& collection_name,
const engine::IDNumbers& ids) {
BaseReqPtr req_ptr = DeleteEntityByIDReq::Create(context, collection_name, ids);
ReqScheduler::ExecReq(req_ptr);
@ -180,46 +175,44 @@ ReqHandler::DeleteEntityByID(const std::shared_ptr<Context>& context, const std:
}
Status
ReqHandler::Search(const std::shared_ptr<milvus::server::Context>& context, const query::QueryPtr& query_ptr,
const milvus::json& json_params, engine::snapshot::CollectionMappings& collection_mappings,
engine::QueryResultPtr& result) {
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);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::ListIDInSegment(const std::shared_ptr<Context>& context, const std::string& collection_name,
int64_t segment_id, engine::IDNumbers& ids) {
ReqHandler::ListIDInSegment(const ContextPtr& context, const std::string& collection_name, int64_t segment_id,
engine::IDNumbers& ids) {
BaseReqPtr req_ptr = ListIDInSegmentReq::Create(context, collection_name, segment_id, ids);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::LoadCollection(const std::shared_ptr<Context>& context, const std::string& collection_name) {
ReqHandler::LoadCollection(const ContextPtr& context, const std::string& collection_name) {
BaseReqPtr req_ptr = LoadCollectionReq::Create(context, collection_name);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::Flush(const std::shared_ptr<Context>& context, const std::vector<std::string>& collection_names) {
ReqHandler::Flush(const ContextPtr& context, const std::vector<std::string>& collection_names) {
BaseReqPtr req_ptr = FlushReq::Create(context, collection_names);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::Compact(const std::shared_ptr<Context>& context, const std::string& collection_name,
double compact_threshold) {
ReqHandler::Compact(const ContextPtr& context, const std::string& collection_name, double compact_threshold) {
BaseReqPtr req_ptr = CompactReq::Create(context, collection_name, compact_threshold);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();
}
Status
ReqHandler::Cmd(const std::shared_ptr<Context>& context, const std::string& cmd, std::string& reply) {
ReqHandler::Cmd(const ContextPtr& context, const std::string& cmd, std::string& reply) {
BaseReqPtr req_ptr = CmdReq::Create(context, cmd, reply);
ReqScheduler::ExecReq(req_ptr);
return req_ptr->status();

68
core/src/server/delivery/ReqHandler.h
View File

@ -31,90 +31,84 @@ class ReqHandler {
ReqHandler() = default;
Status
CreateCollection(const std::shared_ptr<Context>& context, const std::string& collection_name,
std::unordered_map<std::string, FieldSchema>& fields, milvus::json& json_params);
CreateCollection(const ContextPtr& context, const std::string& collection_name, FieldsType& fields,
milvus::json& json_params);
Status
DropCollection(const std::shared_ptr<Context>& context, const std::string& collection_name);
DropCollection(const ContextPtr& context, const std::string& collection_name);
Status
HasCollection(const std::shared_ptr<Context>& context, const std::string& collection_name, bool& has_collection);
HasCollection(const ContextPtr& context, const std::string& collection_name, bool& has_collection);
Status
ListCollections(const std::shared_ptr<Context>& context, std::vector<std::string>& collections);
ListCollections(const ContextPtr& context, std::vector<std::string>& collections);
Status
GetCollectionInfo(const std::shared_ptr<Context>& context, const std::string& collection_name,
GetCollectionInfo(const ContextPtr& context, const std::string& collection_name,
CollectionSchema& collection_schema);
Status
GetCollectionStats(const std::shared_ptr<Context>& context, const std::string& collection_name,
std::string& collection_stats);
GetCollectionStats(const ContextPtr& context, const std::string& collection_name, std::string& collection_stats);
Status
CountEntities(const std::shared_ptr<Context>& context, const std::string& collection_name, int64_t& count);
CountEntities(const ContextPtr& context, const std::string& collection_name, int64_t& count);
Status
CreatePartition(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& tag);
CreatePartition(const ContextPtr& context, const std::string& collection_name, const std::string& tag);
Status
DropPartition(const std::shared_ptr<Context>& context, const std::string& collection_name, const std::string& tag);
DropPartition(const ContextPtr& context, const std::string& collection_name, const std::string& tag);
Status
HasPartition(const std::shared_ptr<Context>& context, const std::string& collection_name, const std::string& tag,
HasPartition(const ContextPtr& context, const std::string& collection_name, const std::string& tag,
bool& has_partition);
Status
ListPartitions(const std::shared_ptr<Context>& context, const std::string& collection_name,
std::vector<std::string>& partitions);
ListPartitions(const ContextPtr& context, const std::string& collection_name, std::vector<std::string>& partitions);
Status
CreateIndex(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& field_name, const std::string& index_name, const milvus::json& json_params);
CreateIndex(const ContextPtr& context, const std::string& collection_name, const std::string& field_name,
const std::string& index_name, const milvus::json& json_params);
Status
DescribeIndex(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& field_name, std::string& index_name, milvus::json& json_params);
DescribeIndex(const ContextPtr& context, const std::string& collection_name, const std::string& field_name,
std::string& index_name, milvus::json& json_params);
Status
DropIndex(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& field_name, const std::string& index_name);
DropIndex(const ContextPtr& context, const std::string& collection_name, const std::string& field_name,
const std::string& index_name);
Status
Insert(const std::shared_ptr<Context>& context, const std::string& collection_name,
const std::string& partition_name, const int64_t& row_count,
std::unordered_map<std::string, std::vector<uint8_t>>& chunk_data);
Insert(const ContextPtr& context, const std::string& collection_name, const std::string& partition_name,
const int64_t& row_count, std::unordered_map<std::string, std::vector<uint8_t>>& chunk_data);
Status
GetEntityByID(const std::shared_ptr<Context>& context, const std::string& collection_name,
const engine::IDNumbers& ids, std::vector<std::string>& field_names, std::vector<bool>& valid_row,
engine::snapshot::CollectionMappings& field_mappings, engine::DataChunkPtr& data_chunk);
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);
Status
DeleteEntityByID(const std::shared_ptr<Context>& context, const std::string& collection_name,
const engine::IDNumbers& ids);
DeleteEntityByID(const ContextPtr& context, const std::string& collection_name, const engine::IDNumbers& ids);
Status
Search(const std::shared_ptr<milvus::server::Context>& context, const query::QueryPtr& query_ptr,
const milvus::json& json_params, engine::snapshot::CollectionMappings& collection_mappings,
engine::QueryResultPtr& result);
Search(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::FieldElementMappings& collection_mappings, engine::QueryResultPtr& result);
Status
ListIDInSegment(const std::shared_ptr<Context>& context, const std::string& collection_name, int64_t segment_id,
ListIDInSegment(const ContextPtr& context, const std::string& collection_name, int64_t segment_id,
engine::IDNumbers& ids);
Status
LoadCollection(const std::shared_ptr<Context>& context, const std::string& collection_name);
LoadCollection(const ContextPtr& context, const std::string& collection_name);
Status
Flush(const std::shared_ptr<Context>& context, const std::vector<std::string>& collection_names);
Flush(const ContextPtr& context, const std::vector<std::string>& collection_names);
Status
Compact(const std::shared_ptr<Context>& context, const std::string& collection_name, double compact_threshold);
Compact(const ContextPtr& context, const std::string& collection_name, double compact_threshold);
Status
Cmd(const std::shared_ptr<Context>& context, const std::string& cmd, std::string& reply);
Cmd(const ContextPtr& context, const std::string& cmd, std::string& reply);
};
} // namespace server

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

@ -57,7 +57,7 @@ CreateIndexReq::OnExecute() {
// only process root collection, ignore partition collection
engine::snapshot::CollectionPtr collection;
engine::snapshot::CollectionMappings fields_schema;
engine::snapshot::FieldElementMappings fields_schema;
auto status = DBWrapper::DB()->GetCollectionInfo(collection_name_, collection, fields_schema);
if (!status.ok()) {
if (status.code() == DB_NOT_FOUND) {

6
core/src/server/delivery/request/GetCollectionInfoReq.cpp
View File

@ -44,12 +44,12 @@ GetCollectionInfoReq::OnExecute() {
STATUS_CHECK(ValidateCollectionName(collection_name_));
engine::snapshot::CollectionPtr collection;
engine::snapshot::CollectionMappings collection_mappings;
STATUS_CHECK(DBWrapper::DB()->GetCollectionInfo(collection_name_, collection, collection_mappings));
engine::snapshot::FieldElementMappings field_mappings;
STATUS_CHECK(DBWrapper::DB()->GetCollectionInfo(collection_name_, collection, field_mappings));
collection_schema_.collection_name_ = collection_name_;
collection_schema_.extra_params_ = collection->GetParams();
for (auto& field_kv : collection_mappings) {
for (auto& field_kv : field_mappings) {
auto field = field_kv.first;
if (field->GetFtype() == (engine::snapshot::FTYPE_TYPE)engine::DataType::UID) {
continue;

18
core/src/server/delivery/request/GetEntityByIDReq.cpp
View File

@ -32,7 +32,7 @@ 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::CollectionMappings& field_mappings,
std::vector<bool>& valid_row, engine::snapshot::FieldElementMappings& field_mappings,
engine::DataChunkPtr& data_chunk)
: BaseReq(context, ReqType::kGetEntityByID),
collection_name_(collection_name),
@ -46,7 +46,7 @@ GetEntityByIDReq::GetEntityByIDReq(const ContextPtr& context, const std::string&
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::CollectionMappings& field_mappings,
std::vector<bool>& valid_row, engine::snapshot::FieldElementMappings& field_mappings,
engine::DataChunkPtr& data_chunk) {
return std::shared_ptr<BaseReq>(
new GetEntityByIDReq(context, collection_name, id_array, field_names_, valid_row, field_mappings, data_chunk));
@ -74,24 +74,24 @@ GetEntityByIDReq::OnExecute() {
// only process root collection, ignore partition collection
engine::snapshot::CollectionPtr collectionPtr;
engine::snapshot::CollectionMappings collection_mappings;
STATUS_CHECK(DBWrapper::DB()->GetCollectionInfo(collection_name_, collectionPtr, collection_mappings));
engine::snapshot::FieldElementMappings field_mappings;
STATUS_CHECK(DBWrapper::DB()->GetCollectionInfo(collection_name_, collectionPtr, field_mappings));
if (collectionPtr == nullptr) {
return Status(SERVER_INVALID_COLLECTION_NAME, "Collection not exist: " + collection_name_);
}
if (field_names_.empty()) {
for (const auto& schema : collection_mappings) {
for (const auto& schema : field_mappings) {
field_names_.emplace_back(schema.first->GetName());
}
field_mappings_ = collection_mappings;
field_mappings_ = field_mappings;
} else {
for (const auto& name : field_names_) {
bool find_field_name = false;
for (const auto& schema : collection_mappings) {
if (name == schema.first->GetName()) {
for (const auto& kv : field_mappings) {
if (name == kv.first->GetName()) {
find_field_name = true;
field_mappings_.insert(schema);
field_mappings_.insert(kv);
break;
}
}

6
core/src/server/delivery/request/GetEntityByIDReq.h
View File

@ -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::CollectionMappings& field_mappings, engine::DataChunkPtr& data_chunk);
engine::snapshot::FieldElementMappings& field_mappings, 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::CollectionMappings& field_mappings, engine::DataChunkPtr& data_chunk);
engine::snapshot::FieldElementMappings& field_mappings, engine::DataChunkPtr& data_chunk);
Status
OnExecute() override;
@ -48,7 +48,7 @@ class GetEntityByIDReq : public BaseReq {
std::string collection_name_;
engine::IDNumbers id_array_;
std::vector<std::string>& field_names_;
engine::snapshot::CollectionMappings& field_mappings_;
engine::snapshot::FieldElementMappings& field_mappings_;
engine::DataChunkPtr& data_chunk_;
std::vector<bool>& valid_row_;
};

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

@ -38,7 +38,7 @@ LoadCollectionReq::OnExecute() {
TimeRecorderAuto rc(hdr);
engine::snapshot::CollectionPtr collection;
engine::snapshot::CollectionMappings fields_schema;
engine::snapshot::FieldElementMappings fields_schema;
auto status = DBWrapper::DB()->GetCollectionInfo(collection_name_, collection, fields_schema);
if (!status.ok()) {
if (status.code() == DB_NOT_FOUND) {

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

@ -31,18 +31,18 @@ namespace milvus {
namespace server {
SearchReq::SearchReq(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::CollectionMappings& collection_mappings, engine::QueryResultPtr& result)
engine::snapshot::FieldElementMappings& field_mappings, engine::QueryResultPtr& result)
: BaseReq(context, ReqType::kSearch),
query_ptr_(query_ptr),
json_params_(json_params),
collection_mappings_(collection_mappings),
field_mappings_(field_mappings),
result_(result) {
}
BaseReqPtr
SearchReq::Create(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::CollectionMappings& collection_mappings, engine::QueryResultPtr& result) {
return std::shared_ptr<BaseReq>(new SearchReq(context, query_ptr, json_params, collection_mappings, result));
engine::snapshot::FieldElementMappings& field_mappings, engine::QueryResultPtr& result) {
return std::shared_ptr<BaseReq>(new SearchReq(context, query_ptr, json_params, field_mappings, result));
}
Status
@ -55,7 +55,7 @@ SearchReq::OnExecute() {
// step 2: check table existence
// only process root table, ignore partition table
engine::snapshot::CollectionPtr collection;
engine::snapshot::CollectionMappings fields_schema;
engine::snapshot::FieldElementMappings fields_schema;
auto status = DBWrapper::DB()->GetCollectionInfo(query_ptr_->collection_id, collection, fields_schema);
fiu_do_on("SearchReq.OnExecute.describe_table_fail", status = Status(milvus::SERVER_UNEXPECTED_ERROR, ""));
if (!status.ok()) {
@ -91,7 +91,7 @@ SearchReq::OnExecute() {
for (const auto& schema : fields_schema) {
if (name.get<std::string>() == schema.first->GetName()) {
find_field_name = true;
collection_mappings_.insert(schema);
field_mappings_.insert(schema);
break;
}
}

6
core/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::CollectionMappings& collection_mappings, engine::QueryResultPtr& result);
engine::snapshot::FieldElementMappings& collection_mappings, engine::QueryResultPtr& result);
protected:
SearchReq(const ContextPtr& context, const query::QueryPtr& query_ptr, const milvus::json& json_params,
engine::snapshot::CollectionMappings& collection_mappings, engine::QueryResultPtr& result);
engine::snapshot::FieldElementMappings& collection_mappings, engine::QueryResultPtr& result);
Status
OnExecute() override;
@ -38,7 +38,7 @@ class SearchReq : public BaseReq {
private:
milvus::query::QueryPtr query_ptr_;
milvus::json json_params_;
engine::snapshot::CollectionMappings& collection_mappings_;
engine::snapshot::FieldElementMappings& field_mappings_;
engine::QueryResultPtr& result_;
};

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

@ -264,7 +264,7 @@ ConstructResults(const TopKQueryResult& result, ::milvus::grpc::QueryResult* res
void
CopyDataChunkToEntity(const engine::DataChunkPtr& data_chunk,
const engine::snapshot::CollectionMappings& field_mappings, int64_t id_size,
const engine::snapshot::FieldElementMappings& field_mappings, int64_t id_size,
::milvus::grpc::Entities* response) {
for (const auto& it : field_mappings) {
auto type = it.first->GetFtype();
@ -876,7 +876,7 @@ GrpcRequestHandler::GetEntityByID(::grpc::ServerContext* context, const ::milvus
}
engine::DataChunkPtr data_chunk;
engine::snapshot::CollectionMappings field_mappings;
engine::snapshot::FieldElementMappings field_mappings;
std::vector<bool> valid_row;
@ -886,11 +886,14 @@ GrpcRequestHandler::GetEntityByID(::grpc::ServerContext* context, const ::milvus
int valid_size = 0;
for (auto it : valid_row) {
response->add_valid_row(it);
if (it)
if (it) {
valid_size++;
}
}
CopyDataChunkToEntity(data_chunk, field_mappings, valid_size, response);
if (valid_size > 0) {
CopyDataChunkToEntity(data_chunk, field_mappings, valid_size, response);
}
LOG_SERVER_INFO_ << LogOut("Request [%s] %s end.", GetContext(context)->ReqID().c_str(), __func__);
SET_RESPONSE(response->mutable_status(), status, context);
@ -1416,8 +1419,8 @@ GrpcRequestHandler::SearchPB(::grpc::ServerContext* context, const ::milvus::grp
engine::QueryResultPtr result = std::make_shared<engine::QueryResult>();
std::vector<std::string> field_names;
engine::snapshot::CollectionMappings collection_mappings;
status = req_handler_.Search(GetContext(context), query_ptr, json_params, collection_mappings, result);
engine::snapshot::FieldElementMappings field_mappings;
status = req_handler_.Search(GetContext(context), query_ptr, json_params, field_mappings, result);
// step 6: construct and return result
response->set_row_num(result->row_num_);
@ -1775,9 +1778,9 @@ GrpcRequestHandler::Search(::grpc::ServerContext* context, const ::milvus::grpc:
}
engine::QueryResultPtr result = std::make_shared<engine::QueryResult>();
engine::snapshot::CollectionMappings collection_mappings;
engine::snapshot::FieldElementMappings field_mappings;
status = req_handler_.Search(GetContext(context), query_ptr, json_params, collection_mappings, result);
status = req_handler_.Search(GetContext(context), query_ptr, json_params, field_mappings, result);
if (!status.ok()) {
SET_RESPONSE(response->mutable_status(), status, context);
@ -1789,7 +1792,7 @@ GrpcRequestHandler::Search(::grpc::ServerContext* context, const ::milvus::grpc:
int64_t id_size = result->result_ids_.size();
grpc_entity->mutable_valid_row()->Resize(id_size, true);
CopyDataChunkToEntity(result->data_chunk_, collection_mappings, id_size, grpc_entity);
CopyDataChunkToEntity(result->data_chunk_, field_mappings, id_size, grpc_entity);
grpc_entity->mutable_ids()->Resize(static_cast<int>(result->result_ids_.size()), 0);
memcpy(grpc_entity->mutable_ids()->mutable_data(), result->result_ids_.data(),

6
core/src/server/web_impl/handler/WebRequestHandler.cpp
View File

@ -734,8 +734,8 @@ WebRequestHandler::Search(const std::string& collection_name, const nlohmann::js
query_ptr_->root = general_query;
engine::QueryResultPtr result = std::make_shared<engine::QueryResult>();
engine::snapshot::CollectionMappings collection_mappings;
status = req_handler_.Search(context_ptr_, query_ptr_, extra_params, collection_mappings, result);
engine::snapshot::FieldElementMappings field_mappings;
status = req_handler_.Search(context_ptr_, query_ptr_, extra_params, field_mappings, result);
if (!status.ok()) {
return status;
@ -805,7 +805,7 @@ WebRequestHandler::GetEntityByIDs(const std::string& collection_name, const std:
std::vector<std::string>& field_names, nlohmann::json& json_out) {
std::vector<bool> valid_row;
engine::DataChunkPtr data_chunk;
engine::snapshot::CollectionMappings field_mappings;
engine::snapshot::FieldElementMappings field_mappings;
std::vector<engine::AttrsData> attr_batch;
std::vector<engine::VectorsData> vector_batch;