admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2026-01-07 19:31:51 +08:00
Code Issues Packages Projects Releases Wiki Activity

Refactor cgo collection, partition and segment

Browse Source 
Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
This commit is contained in:
bigsheeper 2020-11-09 18:07:15 +08:00 committed by yefu.chen
parent 87a4495c59
commit be409b29c8
27 changed files with 60 additions and 217 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/core/src/segcore/CMakeLists.txt
View File

@ -5,9 +5,7 @@ set(SEGCORE_FILES
IndexMeta.cpp
ConcurrentVector.cpp
Collection.cpp
Partition.cpp
collection_c.cpp
partition_c.cpp
segment_c.cpp
SegmentBase.cpp
IndexingEntry.cpp

13
internal/core/src/segcore/Collection.cpp
View File

@ -8,8 +8,8 @@
namespace milvus::segcore {
Collection::Collection(std::string& collection_name, std::string& schema)
: collection_name_(collection_name), schema_json_(schema) {
Collection::Collection(const std::string& collection_proto)
: collection_proto_(collection_proto) {
parse();
index_ = nullptr;
}
@ -112,7 +112,7 @@ Collection::CreateIndex(std::string& index_config) {
void
Collection::parse() {
if (schema_json_.empty()) {
if (collection_proto_.empty()) {
std::cout << "WARN: Use default schema" << std::endl;
auto schema = std::make_shared<Schema>();
schema->AddField("fakevec", DataType::VECTOR_FLOAT, 16);
@ -122,11 +122,16 @@ Collection::parse() {
}
milvus::proto::etcd::CollectionMeta collection_meta;
auto suc = google::protobuf::TextFormat::ParseFromString(schema_json_, &collection_meta);
auto suc = google::protobuf::TextFormat::ParseFromString(collection_proto_, &collection_meta);
if (!suc) {
std::cerr << "unmarshal schema string failed" << std::endl;
}
collection_name_ = collection_meta.schema().name();
// TODO: delete print
std::cout << "create collection " << collection_meta.schema().name() << std::endl;
auto schema = std::make_shared<Schema>();
for (const milvus::proto::schema::FieldSchema& child : collection_meta.schema().fields()) {
const auto& type_params = child.type_params();

8
internal/core/src/segcore/Collection.h
View File

@ -1,13 +1,13 @@
#pragma once
#include "segcore/Partition.h"
#include "SegmentDefs.h"
#include "IndexMeta.h"
namespace milvus::segcore {
class Collection {
public:
explicit Collection(std::string& collection_name, std::string& schema);
explicit Collection(const std::string& collection_proto);
void
parse();
@ -23,7 +23,7 @@ class Collection {
return index_;
}
std::string&
const std::string&
get_collection_name() {
return collection_name_;
}
@ -31,7 +31,7 @@ class Collection {
private:
IndexMetaPtr index_;
std::string collection_name_;
std::string schema_json_;
std::string collection_proto_;
SchemaPtr schema_;
};

9
internal/core/src/segcore/Partition.cpp
View File

@ -1,9 +0,0 @@
#include "Partition.h"
namespace milvus::segcore {
Partition::Partition(std::string& partition_name, SchemaPtr& schema, IndexMetaPtr& index)
: partition_name_(partition_name), schema_(schema), index_(index) {
}
} // namespace milvus::segcore

35
internal/core/src/segcore/Partition.h
View File

@ -1,35 +0,0 @@
#pragma once
#include "SegmentBase.h"
namespace milvus::segcore {
class Partition {
public:
explicit Partition(std::string& partition_name, SchemaPtr& schema, IndexMetaPtr& index);
public:
SchemaPtr&
get_schema() {
return schema_;
}
IndexMetaPtr&
get_index() {
return index_;
}
std::string&
get_partition_name() {
return partition_name_;
}
private:
std::string partition_name_;
SchemaPtr schema_;
IndexMetaPtr index_;
};
using PartitionPtr = std::unique_ptr<Partition>;
} // namespace milvus::segcore

10
internal/core/src/segcore/collection_c.cpp
View File

@ -1,15 +1,13 @@
#include <iostream>
#include "collection_c.h"
#include "Collection.h"
CCollection
NewCollection(const char* collection_name, const char* schema_conf) {
auto name = std::string(collection_name);
auto conf = std::string(schema_conf);
NewCollection(const char* collection_proto) {
auto proto = std::string(collection_proto);
auto collection = std::make_unique<milvus::segcore::Collection>(name, conf);
auto collection = std::make_unique<milvus::segcore::Collection>(proto);
// TODO: delete print
std::cout << "create collection " << collection_name << std::endl;
return (void*)collection.release();
}

2
internal/core/src/segcore/collection_c.h
View File

@ -5,7 +5,7 @@ extern "C" {
typedef void* CCollection;
CCollection
NewCollection(const char* collection_name, const char* schema_conf);
NewCollection(const char* collection_proto);
void
DeleteCollection(CCollection collection);

29
internal/core/src/segcore/partition_c.cpp
View File

@ -1,29 +0,0 @@
#include "partition_c.h"
#include "Partition.h"
#include "Collection.h"
CPartition
NewPartition(CCollection collection, const char* partition_name) {
auto c = (milvus::segcore::Collection*)collection;
auto name = std::string(partition_name);
auto schema = c->get_schema();
auto index = c->get_index();
auto partition = std::make_unique<milvus::segcore::Partition>(name, schema, index);
// TODO: delete print
std::cout << "create partition " << name << std::endl;
return (void*)partition.release();
}
void
DeletePartition(CPartition partition) {
auto p = (milvus::segcore::Partition*)partition;
// TODO: delete print
std::cout << "delete partition " << p->get_partition_name() << std::endl;
delete p;
}

17
internal/core/src/segcore/partition_c.h
View File

@ -1,17 +0,0 @@
#ifdef __cplusplus
extern "C" {
#endif
#include "collection_c.h"
typedef void* CPartition;
CPartition
NewPartition(CCollection collection, const char* partition_name);
void
DeletePartition(CPartition partition);
#ifdef __cplusplus
}
#endif

7
internal/core/src/segcore/segment_c.cpp
View File

@ -3,16 +3,15 @@
#include "SegmentBase.h"
#include "Collection.h"
#include "segment_c.h"
#include "Partition.h"
#include <knowhere/index/vector_index/VecIndex.h>
#include <knowhere/index/vector_index/adapter/VectorAdapter.h>
#include <knowhere/index/vector_index/VecIndexFactory.h>
CSegmentBase
NewSegment(CPartition partition, unsigned long segment_id) {
auto p = (milvus::segcore::Partition*)partition;
NewSegment(CCollection collection, unsigned long segment_id) {
auto col = (milvus::segcore::Collection*)collection;
auto segment = milvus::segcore::CreateSegment(p->get_schema());
auto segment = milvus::segcore::CreateSegment(col->get_schema());
// TODO: delete print
std::cout << "create segment " << segment_id << std::endl;

4
internal/core/src/segcore/segment_c.h
View File

@ -3,7 +3,7 @@ extern "C" {
#endif
#include <stdbool.h>
#include "partition_c.h"
#include "collection_c.h"
typedef void* CSegmentBase;
@ -14,7 +14,7 @@ typedef struct CQueryInfo {
} CQueryInfo;
CSegmentBase
NewSegment(CPartition partition, unsigned long segment_id);
NewSegment(CCollection collection, unsigned long segment_id);
void
DeleteSegment(CSegmentBase segment);

129
internal/core/unittest/test_c_api.cpp
View File

@ -10,41 +10,23 @@
namespace chrono = std::chrono;
TEST(CApiTest, CollectionTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto collection = NewCollection(schema_tmp_conf);
DeleteCollection(collection);
}
TEST(CApiTest, PartitonTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
DeleteCollection(collection);
DeletePartition(partition);
}
TEST(CApiTest, SegmentTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, InsertTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
std::vector<char> raw_data;
std::vector<uint64_t> timestamps;
@ -73,17 +55,13 @@ TEST(CApiTest, InsertTest) {
assert(res == 0);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, DeleteTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
long delete_row_ids[] = {100000, 100001, 100002};
unsigned long delete_timestamps[] = {0, 0, 0};
@ -94,17 +72,13 @@ TEST(CApiTest, DeleteTest) {
assert(del_res == 0);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, SearchTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
std::vector<char> raw_data;
std::vector<uint64_t> timestamps;
@ -146,17 +120,13 @@ TEST(CApiTest, SearchTest) {
assert(sea_res == 0);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, BuildIndexTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
std::vector<char> raw_data;
std::vector<uint64_t> timestamps;
@ -205,49 +175,37 @@ TEST(CApiTest, BuildIndexTest) {
assert(sea_res == 0);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, IsOpenedTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
auto is_opened = IsOpened(segment);
assert(is_opened);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, CloseTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
auto status = Close(segment);
assert(status == 0);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, GetMemoryUsageInBytesTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
auto old_memory_usage_size = GetMemoryUsageInBytes(segment);
std::cout << "old_memory_usage_size = " << old_memory_usage_size << std::endl;
@ -285,7 +243,6 @@ TEST(CApiTest, GetMemoryUsageInBytesTest) {
assert(memory_usage_size == 2785280);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
@ -315,12 +272,9 @@ generate_data(int N) {
} // namespace
TEST(CApiTest, TestSearchPreference) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
auto beg = chrono::high_resolution_clock::now();
auto next = beg;
@ -432,17 +386,13 @@ TEST(CApiTest, TestSearchPreference) {
// }
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, GetDeletedCountTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
long delete_row_ids[] = {100000, 100001, 100002};
unsigned long delete_timestamps[] = {0, 0, 0};
@ -457,17 +407,13 @@ TEST(CApiTest, GetDeletedCountTest) {
assert(deleted_count == 0);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, GetRowCountTest) {
auto collection_name = "collection0";
auto schema_tmp_conf = "";
auto collection = NewCollection(collection_name, schema_tmp_conf);
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
auto collection = NewCollection(schema_tmp_conf);
auto segment = NewSegment(collection, 0);
int N = 10000;
auto [raw_data, timestamps, uids] = generate_data(N);
@ -480,23 +426,18 @@ TEST(CApiTest, GetRowCountTest) {
assert(row_count == N);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
TEST(CApiTest, SchemaTest) {
std::string schema_string =
"id: 6873737669791618215\nname: \"collection0\"\nschema: \u003c\n "
"field_metas: \u003c\n field_name: \"age\"\n type: INT32\n dim: 1\n \u003e\n "
"field_metas: \u003c\n field_name: \"field_1\"\n type: VECTOR_FLOAT\n dim: 16\n \u003e\n"
"\u003e\ncreate_time: 1600416765\nsegment_ids: 6873737669791618215\npartition_tags: \"default\"\n";
auto collection_name = "collection0";
auto collection = NewCollection(collection_name, schema_string.data());
auto partition_name = "partition0";
auto partition = NewPartition(collection, partition_name);
auto segment = NewSegment(partition, 0);
DeleteCollection(collection);
DeletePartition(partition);
DeleteSegment(segment);
}
//TEST(CApiTest, SchemaTest) {
// std::string schema_string =
// "id: 6873737669791618215\nname: \"collection0\"\nschema: \u003c\n "
// "field_metas: \u003c\n field_name: \"age\"\n type: INT32\n dim: 1\n \u003e\n "
// "field_metas: \u003c\n field_name: \"field_1\"\n type: VECTOR_FLOAT\n dim: 16\n \u003e\n"
// "\u003e\ncreate_time: 1600416765\nsegment_ids: 6873737669791618215\npartition_tags: \"default\"\n";
//
// auto collection = NewCollection(schema_string.data());
// auto segment = NewSegment(collection, 0);
// DeleteCollection(collection);
// DeleteSegment(segment);
//}

1
internal/reader/col_seg_container.go
View File

@ -7,7 +7,6 @@ package reader
#cgo LDFLAGS: -L${SRCDIR}/../core/output/lib -lmilvus_segcore -Wl,-rpath=${SRCDIR}/../core/output/lib
#include "collection_c.h"
#include "partition_c.h"
#include "segment_c.h"
*/

4
internal/reader/collection.go
View File

@ -7,7 +7,6 @@ package reader
#cgo LDFLAGS: -L${SRCDIR}/../core/output/lib -lmilvus_segcore -Wl,-rpath=${SRCDIR}/../core/output/lib
#include "collection_c.h"
#include "partition_c.h"
#include "segment_c.h"
*/
@ -39,9 +38,8 @@ func newCollection(collMeta *etcdpb.CollectionMeta, collMetaBlob string) *Collec
CCollection
newCollection(const char* schema_conf);
*/
cTmp := C.CString("tmp")
cCollMetaBlob := C.CString(collMetaBlob)
collection := C.NewCollection(cTmp, cCollMetaBlob)
collection := C.NewCollection(cCollMetaBlob)
var newCollection = &Collection{collectionPtr: collection, meta: collMeta}

0
internal/reader/fg_delete_node.go → internal/reader/flow_graph_delete_node.go
View File

0
internal/reader/fg_filter_dm_node.go → internal/reader/flow_graph_filter_dm_node.go
View File

0
internal/reader/fg_insert_node.go → internal/reader/flow_graph_insert_node.go
View File

0
internal/reader/fg_key2seg_node.go → internal/reader/flow_graph_key2seg_node.go
View File

0
internal/reader/fg_message.go → internal/reader/flow_graph_message.go
View File

0
internal/reader/fg_msg_stream_input_nodes.go → internal/reader/flow_graph_msg_stream_input_nodes.go
View File

0
internal/reader/fg_node.go → internal/reader/flow_graph_node.go
View File

0
internal/reader/fg_schema_update_node.go → internal/reader/flow_graph_schema_update_node.go
View File

0
internal/reader/fg_service_time_node.go → internal/reader/flow_graph_service_time_node.go
View File

1
internal/reader/index.go
View File

@ -7,7 +7,6 @@ package reader
#cgo LDFLAGS: -L../core/output/lib -lmilvus_segcore -Wl,-rpath=../core/output/lib
#include "collection_c.h"
#include "partition_c.h"
#include "segment_c.h"
*/

1
internal/reader/partition.go
View File

@ -7,7 +7,6 @@ package reader
#cgo LDFLAGS: -L${SRCDIR}/../core/output/lib -lmilvus_segcore -Wl,-rpath=${SRCDIR}/../core/output/lib
#include "collection_c.h"
#include "partition_c.h"
#include "segment_c.h"
*/

1
internal/reader/query_node.go
View File

@ -7,7 +7,6 @@ package reader
#cgo LDFLAGS: -L${SRCDIR}/../core/output/lib -lmilvus_segcore -Wl,-rpath=${SRCDIR}/../core/output/lib
#include "collection_c.h"
#include "partition_c.h"
#include "segment_c.h"
*/

4
internal/reader/segment.go
View File

@ -7,7 +7,6 @@ package reader
#cgo LDFLAGS: -L${SRCDIR}/../core/output/lib -lmilvus_segcore -Wl,-rpath=${SRCDIR}/../core/output/lib
#include "collection_c.h"
#include "partition_c.h"
#include "segment_c.h"
*/
@ -41,8 +40,7 @@ func newSegment(collection *Collection, segmentID int64) *Segment {
CSegmentBase
newSegment(CPartition partition, unsigned long segment_id);
*/
var tmp C.CPartition
segmentPtr := C.NewSegment(tmp, C.ulong(segmentID))
segmentPtr := C.NewSegment(collection.collectionPtr, C.ulong(segmentID))
var newSegment = &Segment{segmentPtr: segmentPtr, segmentID: segmentID}
return newSegment