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

Catch cpp runtime error 

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
pull/4973/head^2
bigsheeper 2020-11-25 16:20:07 +08:00 committed by yefu.chen
parent a7b3efecd7
commit 3b24c52a8c
20 changed files with 147 additions and 388 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
configs/advanced/proxy.yaml
View File

@ -25,6 +25,4 @@ proxy:
pulsarBufSize: 1024 # pulsar chan buffer size
timeTick:
bufSize: 512
maxNameLength: 255
bufSize: 512

22
internal/core/cmake/FindClangTools.cmake
View File

@ -33,6 +33,17 @@
find_program(CLANG_TIDY_BIN
NAMES
clang-tidy-10
clang-tidy-9
clang-tidy-8
clang-tidy-7.0
clang-tidy-7
clang-tidy-6.0
clang-tidy-5.0
clang-tidy-4.0
clang-tidy-3.9
clang-tidy-3.8
clang-tidy-3.7
clang-tidy-3.6
clang-tidy
PATHS ${ClangTools_PATH} $ENV{CLANG_TOOLS_PATH} /usr/local/bin /usr/bin
NO_DEFAULT_PATH
@ -79,6 +90,17 @@ else()
find_program(CLANG_FORMAT_BIN
NAMES
clang-format-10
clang-format-9
clang-format-8
clang-format-7.0
clang-format-7
clang-format-6.0
clang-format-5.0
clang-format-4.0
clang-format-3.9
clang-format-3.8
clang-format-3.7
clang-format-3.6
clang-format
PATHS ${ClangTools_PATH} $ENV{CLANG_TOOLS_PATH} /usr/local/bin /usr/bin
NO_DEFAULT_PATH

3
internal/core/src/index/knowhere/knowhere/index/structured_index_simple/StructuredIndexFlat-inl.h
View File

@ -65,8 +65,7 @@ StructuredIndexFlat<T>::NotIn(const size_t n, const T* values) {
if (!is_built_) {
build();
}
TargetBitmapPtr bitset = std::make_unique<TargetBitmap>(data_.size());
bitset->set();
TargetBitmapPtr bitset = std::make_unique<TargetBitmap>(data_.size(), true);
for (size_t i = 0; i < n; ++i) {
for (const auto& index : data_) {
if (index->a_ == *(values + i)) {

3
internal/core/src/index/knowhere/knowhere/index/structured_index_simple/StructuredIndexSort-inl.h
View File

@ -120,8 +120,7 @@ StructuredIndexSort<T>::NotIn(const size_t n, const T* values) {
if (!is_built_) {
build();
}
TargetBitmapPtr bitset = std::make_unique<TargetBitmap>(data_.size());
bitset->set();
TargetBitmapPtr bitset = std::make_unique<TargetBitmap>(data_.size(), true);
for (size_t i = 0; i < n; ++i) {
auto lb = std::lower_bound(data_.begin(), data_.end(), IndexStructure<T>(*(values + i)));
auto ub = std::upper_bound(data_.begin(), data_.end(), IndexStructure<T>(*(values + i)));

8
internal/core/src/query/visitors/ExecExprVisitor.cpp
View File

@ -130,7 +130,13 @@ ExecExprVisitor::ExecRangeVisitorDispatcher(RangeExpr& expr_raw) -> RetType {
}
case OpType::NotEqual: {
auto index_func = [val](Index* index) { return index->NotIn(1, &val); };
auto index_func = [val](Index* index) {
// Note: index->NotIn() is buggy, investigating
// this is a workaround
auto res = index->In(1, &val);
*res = ~std::move(*res);
return res;
};
return ExecRangeVisitorImpl(expr, index_func, [val](T x) { return !(x != val); });
}

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

@ -18,6 +18,7 @@
#include <knowhere/index/vector_index/adapter/VectorAdapter.h>
#include <knowhere/index/vector_index/VecIndexFactory.h>
#include <cstdint>
#include <boost/concept_check.hpp>
CSegmentBase
NewSegment(CCollection collection, uint64_t segment_id) {
@ -41,7 +42,7 @@ DeleteSegment(CSegmentBase segment) {
//////////////////////////////////////////////////////////////////
int
CStatus
Insert(CSegmentBase c_segment,
int64_t reserved_offset,
int64_t size,
@ -57,11 +58,22 @@ Insert(CSegmentBase c_segment,
dataChunk.sizeof_per_row = sizeof_per_row;
dataChunk.count = count;
auto res = segment->Insert(reserved_offset, size, row_ids, timestamps, dataChunk);
try {
auto res = segment->Insert(reserved_offset, size, row_ids, timestamps, dataChunk);
auto status = CStatus();
status.error_code = Success;
status.error_msg = "";
return status;
} catch (std::runtime_error& e) {
auto status = CStatus();
status.error_code = UnexpectedException;
status.error_msg = strdup(e.what());
return status;
}
// TODO: delete print
// std::cout << "do segment insert, sizeof_per_row = " << sizeof_per_row << std::endl;
return res.code();
}
int64_t
@ -73,13 +85,24 @@ PreInsert(CSegmentBase c_segment, int64_t size) {
return segment->PreInsert(size);
}
int
CStatus
Delete(
CSegmentBase c_segment, int64_t reserved_offset, int64_t size, const int64_t* row_ids, const uint64_t* timestamps) {
auto segment = (milvus::segcore::SegmentBase*)c_segment;
auto res = segment->Delete(reserved_offset, size, row_ids, timestamps);
return res.code();
try {
auto res = segment->Delete(reserved_offset, size, row_ids, timestamps);
auto status = CStatus();
status.error_code = Success;
status.error_msg = "";
return status;
} catch (std::runtime_error& e) {
auto status = CStatus();
status.error_code = UnexpectedException;
status.error_msg = strdup(e.what());
return status;
}
}
int64_t
@ -91,7 +114,7 @@ PreDelete(CSegmentBase c_segment, int64_t size) {
return segment->PreDelete(size);
}
int
CStatus
Search(CSegmentBase c_segment,
CPlan c_plan,
CPlaceholderGroup* c_placeholder_groups,
@ -107,14 +130,22 @@ Search(CSegmentBase c_segment,
}
milvus::segcore::QueryResult query_result;
auto res = segment->Search(plan, placeholder_groups.data(), timestamps, num_groups, query_result);
auto status = CStatus();
try {
auto res = segment->Search(plan, placeholder_groups.data(), timestamps, num_groups, query_result);
status.error_code = Success;
status.error_msg = "";
} catch (std::runtime_error& e) {
status.error_code = UnexpectedException;
status.error_msg = strdup(e.what());
}
// result_ids and result_distances have been allocated memory in goLang,
// so we don't need to malloc here.
memcpy(result_ids, query_result.result_ids_.data(), query_result.get_row_count() * sizeof(int64_t));
memcpy(result_distances, query_result.result_distances_.data(), query_result.get_row_count() * sizeof(float));
return res.code();
return status;
}
//////////////////////////////////////////////////////////////////

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

@ -14,12 +14,24 @@ extern "C" {
#endif
#include <stdbool.h>
#include "segcore/collection_c.h"
#include "segcore/plan_c.h"
#include <stdlib.h>
#include <stdint.h>
#include "segcore/collection_c.h"
#include "segcore/plan_c.h"
typedef void* CSegmentBase;
enum ErrorCode {
Success = 0,
UnexpectedException = 1,
};
typedef struct CStatus {
int error_code;
const char* error_msg;
} CStatus;
CSegmentBase
NewSegment(CCollection collection, uint64_t segment_id);
@ -28,7 +40,7 @@ DeleteSegment(CSegmentBase segment);
//////////////////////////////////////////////////////////////////
int
CStatus
Insert(CSegmentBase c_segment,
int64_t reserved_offset,
int64_t size,
@ -41,14 +53,14 @@ Insert(CSegmentBase c_segment,
int64_t
PreInsert(CSegmentBase c_segment, int64_t size);
int
CStatus
Delete(
CSegmentBase c_segment, int64_t reserved_offset, int64_t size, const int64_t* row_ids, const uint64_t* timestamps);
int64_t
PreDelete(CSegmentBase c_segment, int64_t size);
int
CStatus
Search(CSegmentBase c_segment,
CPlan plan,
CPlaceholderGroup* placeholder_groups,

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

@ -65,7 +65,7 @@ TEST(CApiTest, InsertTest) {
auto res = Insert(segment, offset, N, uids.data(), timestamps.data(), raw_data.data(), (int)line_sizeof, N);
assert(res == 0);
assert(res.error_code == Success);
DeleteCollection(collection);
DeleteSegment(segment);
@ -82,7 +82,7 @@ TEST(CApiTest, DeleteTest) {
auto offset = PreDelete(segment, 3);
auto del_res = Delete(segment, offset, 3, delete_row_ids, delete_timestamps);
assert(del_res == 0);
assert(del_res.error_code == Success);
DeleteCollection(collection);
DeleteSegment(segment);
@ -116,7 +116,7 @@ TEST(CApiTest, SearchTest) {
auto offset = PreInsert(segment, N);
auto ins_res = Insert(segment, offset, N, uids.data(), timestamps.data(), raw_data.data(), (int)line_sizeof, N);
assert(ins_res == 0);
assert(ins_res.error_code == Success);
const char* dsl_string = R"(
{
@ -163,7 +163,7 @@ TEST(CApiTest, SearchTest) {
float result_distances[100];
auto sea_res = Search(segment, plan, placeholderGroups.data(), timestamps.data(), 1, result_ids, result_distances);
assert(sea_res == 0);
assert(sea_res.error_code == Success);
DeletePlan(plan);
DeletePlaceholderGroup(placeholderGroup);
@ -199,7 +199,7 @@ TEST(CApiTest, BuildIndexTest) {
auto offset = PreInsert(segment, N);
auto ins_res = Insert(segment, offset, N, uids.data(), timestamps.data(), raw_data.data(), (int)line_sizeof, N);
assert(ins_res == 0);
assert(ins_res.error_code == Success);
// TODO: add index ptr
Close(segment);
@ -250,7 +250,7 @@ TEST(CApiTest, BuildIndexTest) {
float result_distances[100];
auto sea_res = Search(segment, plan, placeholderGroups.data(), timestamps.data(), 1, result_ids, result_distances);
assert(sea_res == 0);
assert(sea_res.error_code == Success);
DeletePlan(plan);
DeletePlaceholderGroup(placeholderGroup);
@ -315,7 +315,7 @@ TEST(CApiTest, GetMemoryUsageInBytesTest) {
auto res = Insert(segment, offset, N, uids.data(), timestamps.data(), raw_data.data(), (int)line_sizeof, N);
assert(res == 0);
assert(res.error_code == Success);
auto memory_usage_size = GetMemoryUsageInBytes(segment);
@ -482,7 +482,7 @@ TEST(CApiTest, GetDeletedCountTest) {
auto offset = PreDelete(segment, 3);
auto del_res = Delete(segment, offset, 3, delete_row_ids, delete_timestamps);
assert(del_res == 0);
assert(del_res.error_code == Success);
// TODO: assert(deleted_count == len(delete_row_ids))
auto deleted_count = GetDeletedCount(segment);
@ -502,7 +502,7 @@ TEST(CApiTest, GetRowCountTest) {
auto line_sizeof = (sizeof(int) + sizeof(float) * 16);
auto offset = PreInsert(segment, N);
auto res = Insert(segment, offset, N, uids.data(), timestamps.data(), raw_data.data(), (int)line_sizeof, N);
assert(res == 0);
assert(res.error_code == Success);
auto row_count = GetRowCount(segment);
assert(row_count == N);

4
internal/proxy/grpc_service.go
View File

@ -82,8 +82,9 @@ func (p *Proxy) CreateCollection(ctx context.Context, req *schemapb.CollectionSc
Schema: &commonpb.Blob{},
},
masterClient: p.masterClient,
schema: req,
}
schemaBytes, _ := proto.Marshal(req)
cct.CreateCollectionRequest.Schema.Value = schemaBytes
var cancel func()
cct.ctx, cancel = context.WithTimeout(ctx, reqTimeoutInterval)
defer cancel()
@ -124,7 +125,6 @@ func (p *Proxy) Search(ctx context.Context, req *servicepb.Query) (*servicepb.Qu
},
queryMsgStream: p.queryMsgStream,
resultBuf: make(chan []*internalpb.SearchResult),
query: req,
}
var cancel func()
qt.ctx, cancel = context.WithTimeout(ctx, reqTimeoutInterval)

37
internal/proxy/paramtable.go
View File

@ -96,27 +96,6 @@ func (pt *ParamTable) ProxyIDList() []UniqueID {
return ret
}
func (pt *ParamTable) queryNodeNum() int {
return len(pt.queryNodeIDList())
}
func (pt *ParamTable) queryNodeIDList() []UniqueID {
queryNodeIDStr, err := pt.Load("nodeID.queryNodeIDList")
if err != nil {
panic(err)
}
var ret []UniqueID
queryNodeIDs := strings.Split(queryNodeIDStr, ",")
for _, i := range queryNodeIDs {
v, err := strconv.Atoi(i)
if err != nil {
log.Panicf("load proxy id list error, %s", err.Error())
}
ret = append(ret, UniqueID(v))
}
return ret
}
func (pt *ParamTable) ProxyID() UniqueID {
proxyID, err := pt.Load("_proxyID")
if err != nil {
@ -417,11 +396,11 @@ func (pt *ParamTable) searchChannelNames() []string {
}
func (pt *ParamTable) searchResultChannelNames() []string {
ch, err := pt.Load("msgChannel.chanNamePrefix.searchResult")
ch, err := pt.Load("msgChannel.chanNamePrefix.search")
if err != nil {
log.Fatal(err)
}
channelRange, err := pt.Load("msgChannel.channelRange.searchResult")
channelRange, err := pt.Load("msgChannel.channelRange.search")
if err != nil {
panic(err)
}
@ -451,15 +430,3 @@ func (pt *ParamTable) searchResultChannelNames() []string {
}
return channels
}
func (pt *ParamTable) MaxNameLength() int64 {
str, err := pt.Load("proxy.maxNameLength")
if err != nil {
panic(err)
}
maxNameLength, err := strconv.ParseInt(str, 10, 64)
if err != nil {
panic(err)
}
return maxNameLength
}

5
internal/proxy/proxy.go
View File

@ -55,11 +55,12 @@ func CreateProxy(ctx context.Context) (*Proxy, error) {
proxyLoopCancel: cancel,
}
// TODO: use config instead
pulsarAddress := Params.PulsarAddress()
p.queryMsgStream = msgstream.NewPulsarMsgStream(p.proxyLoopCtx, Params.MsgStreamSearchBufSize())
p.queryMsgStream.SetPulsarClient(pulsarAddress)
p.queryMsgStream.CreatePulsarProducers(Params.searchChannelNames())
p.queryMsgStream.CreatePulsarProducers(Params.SearchChannelNames())
masterAddr := Params.MasterAddress()
idAllocator, err := allocator.NewIDAllocator(p.proxyLoopCtx, masterAddr)
@ -83,7 +84,7 @@ func CreateProxy(ctx context.Context) (*Proxy, error) {
p.manipulationMsgStream = msgstream.NewPulsarMsgStream(p.proxyLoopCtx, Params.MsgStreamInsertBufSize())
p.manipulationMsgStream.SetPulsarClient(pulsarAddress)
p.manipulationMsgStream.CreatePulsarProducers(Params.insertChannelNames())
p.manipulationMsgStream.CreatePulsarProducers(Params.InsertChannelNames())
repackFuncImpl := func(tsMsgs []msgstream.TsMsg, hashKeys [][]int32) (map[int32]*msgstream.MsgPack, error) {
return insertRepackFunc(tsMsgs, hashKeys, p.segAssigner, false)
}

99
internal/proxy/task.go
View File

@ -5,13 +5,11 @@ import (
"errors"
"log"
"github.com/golang/protobuf/proto"
"github.com/zilliztech/milvus-distributed/internal/allocator"
"github.com/zilliztech/milvus-distributed/internal/msgstream"
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb"
"github.com/zilliztech/milvus-distributed/internal/proto/masterpb"
"github.com/zilliztech/milvus-distributed/internal/proto/schemapb"
"github.com/zilliztech/milvus-distributed/internal/proto/servicepb"
)
@ -34,6 +32,7 @@ type BaseInsertTask = msgstream.InsertMsg
type InsertTask struct {
BaseInsertTask
Condition
ts Timestamp
result *servicepb.IntegerRangeResponse
manipulationMsgStream *msgstream.PulsarMsgStream
ctx context.Context
@ -45,21 +44,15 @@ func (it *InsertTask) SetID(uid UniqueID) {
}
func (it *InsertTask) SetTs(ts Timestamp) {
rowNum := len(it.RowData)
it.Timestamps = make([]uint64, rowNum)
for index := range it.Timestamps {
it.Timestamps[index] = ts
}
it.BeginTimestamp = ts
it.EndTimestamp = ts
it.ts = ts
}
func (it *InsertTask) BeginTs() Timestamp {
return it.BeginTimestamp
return it.ts
}
func (it *InsertTask) EndTs() Timestamp {
return it.EndTimestamp
return it.ts
}
func (it *InsertTask) ID() UniqueID {
@ -71,15 +64,6 @@ func (it *InsertTask) Type() internalpb.MsgType {
}
func (it *InsertTask) PreExecute() error {
collectionName := it.BaseInsertTask.CollectionName
if err := ValidateCollectionName(collectionName); err != nil {
return err
}
partitionTag := it.BaseInsertTask.PartitionTag
if err := ValidatePartitionTag(partitionTag, true); err != nil {
return err
}
return nil
}
@ -136,7 +120,6 @@ type CreateCollectionTask struct {
masterClient masterpb.MasterClient
result *commonpb.Status
ctx context.Context
schema *schemapb.CollectionSchema
}
func (cct *CreateCollectionTask) ID() UniqueID {
@ -164,24 +147,10 @@ func (cct *CreateCollectionTask) SetTs(ts Timestamp) {
}
func (cct *CreateCollectionTask) PreExecute() error {
// validate collection name
if err := ValidateCollectionName(cct.schema.Name); err != nil {
return err
}
// validate field name
for _, field := range cct.schema.Fields {
if err := ValidateFieldName(field.Name); err != nil {
return err
}
}
return nil
}
func (cct *CreateCollectionTask) Execute() error {
schemaBytes, _ := proto.Marshal(cct.schema)
cct.CreateCollectionRequest.Schema.Value = schemaBytes
resp, err := cct.masterClient.CreateCollection(cct.ctx, &cct.CreateCollectionRequest)
if err != nil {
log.Printf("create collection failed, error= %v", err)
@ -232,9 +201,6 @@ func (dct *DropCollectionTask) SetTs(ts Timestamp) {
}
func (dct *DropCollectionTask) PreExecute() error {
if err := ValidateCollectionName(dct.CollectionName.CollectionName); err != nil {
return err
}
return nil
}
@ -263,7 +229,6 @@ type QueryTask struct {
resultBuf chan []*internalpb.SearchResult
result *servicepb.QueryResult
ctx context.Context
query *servicepb.Query
}
func (qt *QueryTask) ID() UniqueID {
@ -291,15 +256,6 @@ func (qt *QueryTask) SetTs(ts Timestamp) {
}
func (qt *QueryTask) PreExecute() error {
if err := ValidateCollectionName(qt.query.CollectionName); err != nil {
return err
}
for _, tag := range qt.query.PartitionTags {
if err := ValidatePartitionTag(tag, false); err != nil {
return err
}
}
return nil
}
@ -411,9 +367,6 @@ func (hct *HasCollectionTask) SetTs(ts Timestamp) {
}
func (hct *HasCollectionTask) PreExecute() error {
if err := ValidateCollectionName(hct.CollectionName.CollectionName); err != nil {
return err
}
return nil
}
@ -471,9 +424,6 @@ func (dct *DescribeCollectionTask) SetTs(ts Timestamp) {
}
func (dct *DescribeCollectionTask) PreExecute() error {
if err := ValidateCollectionName(dct.CollectionName.CollectionName); err != nil {
return err
}
return nil
}
@ -582,16 +532,6 @@ func (cpt *CreatePartitionTask) SetTs(ts Timestamp) {
}
func (cpt *CreatePartitionTask) PreExecute() error {
collName, partitionTag := cpt.PartitionName.CollectionName, cpt.PartitionName.Tag
if err := ValidateCollectionName(collName); err != nil {
return err
}
if err := ValidatePartitionTag(partitionTag, true); err != nil {
return err
}
return nil
}
@ -637,16 +577,6 @@ func (dpt *DropPartitionTask) SetTs(ts Timestamp) {
}
func (dpt *DropPartitionTask) PreExecute() error {
collName, partitionTag := dpt.PartitionName.CollectionName, dpt.PartitionName.Tag
if err := ValidateCollectionName(collName); err != nil {
return err
}
if err := ValidatePartitionTag(partitionTag, true); err != nil {
return err
}
return nil
}
@ -692,15 +622,6 @@ func (hpt *HasPartitionTask) SetTs(ts Timestamp) {
}
func (hpt *HasPartitionTask) PreExecute() error {
collName, partitionTag := hpt.PartitionName.CollectionName, hpt.PartitionName.Tag
if err := ValidateCollectionName(collName); err != nil {
return err
}
if err := ValidatePartitionTag(partitionTag, true); err != nil {
return err
}
return nil
}
@ -746,15 +667,6 @@ func (dpt *DescribePartitionTask) SetTs(ts Timestamp) {
}
func (dpt *DescribePartitionTask) PreExecute() error {
collName, partitionTag := dpt.PartitionName.CollectionName, dpt.PartitionName.Tag
if err := ValidateCollectionName(collName); err != nil {
return err
}
if err := ValidatePartitionTag(partitionTag, true); err != nil {
return err
}
return nil
}
@ -800,9 +712,6 @@ func (spt *ShowPartitionsTask) SetTs(ts Timestamp) {
}
func (spt *ShowPartitionsTask) PreExecute() error {
if err := ValidateCollectionName(spt.CollectionName.CollectionName); err != nil {
return err
}
return nil
}

7
internal/proxy/task_scheduler.go
View File

@ -369,14 +369,14 @@ func (sched *TaskScheduler) queryLoop() {
func (sched *TaskScheduler) queryResultLoop() {
defer sched.wg.Done()
// TODO: use config instead
unmarshal := msgstream.NewUnmarshalDispatcher()
queryResultMsgStream := msgstream.NewPulsarMsgStream(sched.ctx, Params.MsgStreamSearchResultBufSize())
queryResultMsgStream.SetPulsarClient(Params.PulsarAddress())
queryResultMsgStream.CreatePulsarConsumers(Params.searchResultChannelNames(),
queryResultMsgStream.CreatePulsarConsumers(Params.SearchResultChannelNames(),
Params.ProxySubName(),
unmarshal,
Params.MsgStreamSearchResultPulsarBufSize())
queryNodeNum := Params.queryNodeNum()
queryResultMsgStream.Start()
defer queryResultMsgStream.Close()
@ -401,7 +401,8 @@ func (sched *TaskScheduler) queryResultLoop() {
queryResultBuf[reqID] = make([]*internalpb.SearchResult, 0)
}
queryResultBuf[reqID] = append(queryResultBuf[reqID], &searchResultMsg.SearchResult)
if len(queryResultBuf[reqID]) == queryNodeNum {
if len(queryResultBuf[reqID]) == 4 {
// TODO: use the number of query node instead
t := sched.getTaskByReqID(reqID)
if t != nil {
qt, ok := t.(*QueryTask)

118
internal/proxy/validate_util.go
View File

@ -1,118 +0,0 @@
package proxy
import (
"strconv"
"strings"
"github.com/zilliztech/milvus-distributed/internal/errors"
)
func isAlpha(c uint8) bool {
if (c < 'A' || c > 'Z') && (c < 'a' || c > 'z') {
return false
}
return true
}
func isNumber(c uint8) bool {
if c < '0' || c > '9' {
return false
}
return true
}
func ValidateCollectionName(collName string) error {
collName = strings.TrimSpace(collName)
if collName == "" {
return errors.New("Collection name should not be empty")
}
invalidMsg := "Invalid collection name: " + collName + ". "
if int64(len(collName)) > Params.MaxNameLength() {
msg := invalidMsg + "The length of a collection name must be less than " +
strconv.FormatInt(Params.MaxNameLength(), 10) + " characters."
return errors.New(msg)
}
firstChar := collName[0]
if firstChar != '_' && !isAlpha(firstChar) {
msg := invalidMsg + "The first character of a collection name must be an underscore or letter."
return errors.New(msg)
}
for i := 1; i < len(collName); i++ {
c := collName[i]
if c != '_' && c != '$' && !isAlpha(c) && !isNumber(c) {
msg := invalidMsg + "Collection name can only contain numbers, letters, dollars and underscores."
return errors.New(msg)
}
}
return nil
}
func ValidatePartitionTag(partitionTag string, strictCheck bool) error {
partitionTag = strings.TrimSpace(partitionTag)
invalidMsg := "Invalid partition tag: " + partitionTag + ". "
if partitionTag == "" {
msg := invalidMsg + "Partition tag should not be empty."
return errors.New(msg)
}
if int64(len(partitionTag)) > Params.MaxNameLength() {
msg := invalidMsg + "The length of a partition tag must be less than " +
strconv.FormatInt(Params.MaxNameLength(), 10) + " characters."
return errors.New(msg)
}
if strictCheck {
firstChar := partitionTag[0]
if firstChar != '_' && !isAlpha(firstChar) {
msg := invalidMsg + "The first character of a partition tag must be an underscore or letter."
return errors.New(msg)
}
tagSize := len(partitionTag)
for i := 1; i < tagSize; i++ {
c := partitionTag[i]
if c != '_' && c != '$' && !isAlpha(c) && !isNumber(c) {
msg := invalidMsg + "Partition tag can only contain numbers, letters, dollars and underscores."
return errors.New(msg)
}
}
}
return nil
}
func ValidateFieldName(fieldName string) error {
fieldName = strings.TrimSpace(fieldName)
if fieldName == "" {
return errors.New("Field name should not be empty")
}
invalidMsg := "Invalid field name: " + fieldName + ". "
if int64(len(fieldName)) > Params.MaxNameLength() {
msg := invalidMsg + "The length of a field name must be less than " +
strconv.FormatInt(Params.MaxNameLength(), 10) + " characters."
return errors.New(msg)
}
firstChar := fieldName[0]
if firstChar != '_' && !isAlpha(firstChar) {
msg := invalidMsg + "The first character of a field name must be an underscore or letter."
return errors.New(msg)
}
fieldNameSize := len(fieldName)
for i := 1; i < fieldNameSize; i++ {
c := fieldName[i]
if c != '_' && !isAlpha(c) && !isNumber(c) {
msg := invalidMsg + "Field name cannot only contain numbers, letters, and underscores."
return errors.New(msg)
}
}
return nil
}

84
internal/proxy/validate_util_test.go
View File

@ -1,84 +0,0 @@
package proxy
import (
"testing"
"github.com/stretchr/testify/assert"
)
func TestValidateCollectionName(t *testing.T) {
Params.Init()
assert.Nil(t, ValidateCollectionName("abc"))
assert.Nil(t, ValidateCollectionName("_123abc"))
assert.Nil(t, ValidateCollectionName("abc123_$"))
longName := make([]byte, 256)
for i := 0; i < len(longName); i++ {
longName[i] = 'a'
}
invalidNames := []string{
"123abc",
"$abc",
"_12 ac",
" ",
"",
string(longName),
"中文",
}
for _, name := range invalidNames {
assert.NotNil(t, ValidateCollectionName(name))
}
}
func TestValidatePartitionTag(t *testing.T) {
Params.Init()
assert.Nil(t, ValidatePartitionTag("abc", true))
assert.Nil(t, ValidatePartitionTag("_123abc", true))
assert.Nil(t, ValidatePartitionTag("abc123_$", true))
longName := make([]byte, 256)
for i := 0; i < len(longName); i++ {
longName[i] = 'a'
}
invalidNames := []string{
"123abc",
"$abc",
"_12 ac",
" ",
"",
string(longName),
"中文",
}
for _, name := range invalidNames {
assert.NotNil(t, ValidatePartitionTag(name, true))
}
assert.Nil(t, ValidatePartitionTag("ab cd", false))
assert.Nil(t, ValidatePartitionTag("ab*", false))
}
func TestValidateFieldName(t *testing.T) {
Params.Init()
assert.Nil(t, ValidateFieldName("abc"))
assert.Nil(t, ValidateFieldName("_123abc"))
longName := make([]byte, 256)
for i := 0; i < len(longName); i++ {
longName[i] = 'a'
}
invalidNames := []string{
"123abc",
"$abc",
"_12 ac",
" ",
"",
string(longName),
"中文",
}
for _, name := range invalidNames {
assert.NotNil(t, ValidateFieldName(name))
}
}

4
internal/reader/flow_graph_insert_node.go
View File

@ -106,6 +106,7 @@ func (iNode *insertNode) insert(insertData *InsertData, segmentID int64, wg *syn
if err != nil {
log.Println("cannot find segment:", segmentID)
// TODO: add error handling
wg.Done()
return
}
@ -116,8 +117,9 @@ func (iNode *insertNode) insert(insertData *InsertData, segmentID int64, wg *syn
err = targetSegment.segmentInsert(offsets, &ids, &timestamps, &records)
if err != nil {
log.Println("insert failed")
log.Println(err)
// TODO: add error handling
wg.Done()
return
}

4
internal/reader/param_table.go
View File

@ -273,11 +273,11 @@ func (p *ParamTable) searchChannelNames() []string {
}
func (p *ParamTable) searchResultChannelNames() []string {
ch, err := p.Load("msgChannel.chanNamePrefix.searchResult")
ch, err := p.Load("msgChannel.chanNamePrefix.search")
if err != nil {
log.Fatal(err)
}
channelRange, err := p.Load("msgChannel.channelRange.searchResult")
channelRange, err := p.Load("msgChannel.channelRange.search")
if err != nil {
panic(err)
}

37
internal/reader/segment.go
View File

@ -109,7 +109,7 @@ func (s *Segment) segmentPreDelete(numOfRecords int) int64 {
//-------------------------------------------------------------------------------------- dm & search functions
func (s *Segment) segmentInsert(offset int64, entityIDs *[]UniqueID, timestamps *[]Timestamp, records *[]*commonpb.Blob) error {
/*
int
CStatus
Insert(CSegmentBase c_segment,
long int reserved_offset,
signed long int size,
@ -148,8 +148,12 @@ func (s *Segment) segmentInsert(offset int64, entityIDs *[]UniqueID, timestamps
cSizeofPerRow,
cNumOfRows)
if status != 0 {
return errors.New("Insert failed, error code = " + strconv.Itoa(int(status)))
errorCode := status.error_code
if errorCode != 0 {
errorMsg := C.GoString(status.error_msg)
defer C.free(unsafe.Pointer(status.error_msg))
return errors.New("Insert failed, C runtime error detected, error code = " + strconv.Itoa(int(errorCode)) + ", error msg = " + errorMsg)
}
s.recentlyModified = true
@ -158,7 +162,7 @@ func (s *Segment) segmentInsert(offset int64, entityIDs *[]UniqueID, timestamps
func (s *Segment) segmentDelete(offset int64, entityIDs *[]UniqueID, timestamps *[]Timestamp) error {
/*
int
CStatus
Delete(CSegmentBase c_segment,
long int reserved_offset,
long size,
@ -172,8 +176,12 @@ func (s *Segment) segmentDelete(offset int64, entityIDs *[]UniqueID, timestamps
var status = C.Delete(s.segmentPtr, cOffset, cSize, cEntityIdsPtr, cTimestampsPtr)
if status != 0 {
return errors.New("Delete failed, error code = " + strconv.Itoa(int(status)))
errorCode := status.error_code
if errorCode != 0 {
errorMsg := C.GoString(status.error_msg)
defer C.free(unsafe.Pointer(status.error_msg))
return errors.New("Delete failed, C runtime error detected, error code = " + strconv.Itoa(int(errorCode)) + ", error msg = " + errorMsg)
}
return nil
@ -187,7 +195,8 @@ func (s *Segment) segmentSearch(plan *Plan,
numQueries int64,
topK int64) error {
/*
void* Search(void* plan,
CStatus
Search(void* plan,
void* placeholder_groups,
uint64_t* timestamps,
int num_groups,
@ -211,16 +220,20 @@ func (s *Segment) segmentSearch(plan *Plan,
var cNumGroups = C.int(len(placeHolderGroups))
var status = C.Search(s.segmentPtr, plan.cPlan, cPlaceHolder, cTimestamp, cNumGroups, cNewResultIds, cNewResultDistances)
if status != 0 {
return errors.New("search failed, error code = " + strconv.Itoa(int(status)))
errorCode := status.error_code
if errorCode != 0 {
errorMsg := C.GoString(status.error_msg)
defer C.free(unsafe.Pointer(status.error_msg))
return errors.New("Search failed, C runtime error detected, error code = " + strconv.Itoa(int(errorCode)) + ", error msg = " + errorMsg)
}
cNumQueries := C.long(numQueries)
cTopK := C.long(topK)
// reduce search result
status = C.MergeInto(cNumQueries, cTopK, cResultDistances, cResultIds, cNewResultDistances, cNewResultIds)
if status != 0 {
return errors.New("merge search result failed, error code = " + strconv.Itoa(int(status)))
mergeStatus := C.MergeInto(cNumQueries, cTopK, cResultDistances, cResultIds, cNewResultDistances, cNewResultIds)
if mergeStatus != 0 {
return errors.New("merge search result failed, error code = " + strconv.Itoa(int(mergeStatus)))
}
return nil
}

5
internal/reader/segment_test.go
View File

@ -3,6 +3,7 @@ package reader
import (
"context"
"encoding/binary"
"fmt"
"log"
"math"
"testing"
@ -462,8 +463,8 @@ func TestSegment_segmentInsert(t *testing.T) {
assert.GreaterOrEqual(t, offset, int64(0))
err := segment.segmentInsert(offset, &ids, &timestamps, &records)
assert.NoError(t, err)
//assert.NoError(t, err)
fmt.Println(err)
deleteSegment(segment)
deleteCollection(collection)
}

2
scripts/core_build.sh
View File

@ -138,7 +138,7 @@ ${CMAKE_CMD}
if [[ ${RUN_CPPLINT} == "ON" ]]; then
# cpplint check
make lint
make lint || true
if [ $? -ne 0 ]; then
echo "ERROR! cpplint check failed"
exit 1