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

Refactor master scheduler: describe_collection 

Signed-off-by: XuanYang-cn <xuan.yang@zilliz.com>
pull/4973/head^2
XuanYang-cn 2020-11-14 13:06:41 +08:00 committed by yefu.chen
parent 74f8c45a23
commit 27032cc126
5 changed files with 183 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
internal/master/collection_task.go
View File

@ -185,14 +185,7 @@ func (t *describeCollectionTask) Execute() error {
return err
}
description := servicepb.CollectionDescription{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_SUCCESS,
},
Schema: collection.Schema,
}
t.description = &description
t.description.Schema = collection.Schema
return nil
}

57
internal/master/collection_task_test.go
View File

@ -149,6 +149,7 @@ func TestMaster_CreateCollectionTask(t *testing.T) {
assert.Nil(t, err)
assert.NotEqual(t, boolResp.Status.ErrorCode, commonpb.ErrorCode_SUCCESS)
// CreateCollection Test
collMeta, err := svr.mt.GetCollectionByName(sch.Name)
assert.Nil(t, err)
t.Logf("collection id = %d", collMeta.ID)
@ -182,6 +183,62 @@ func TestMaster_CreateCollectionTask(t *testing.T) {
assert.Equal(t, collMeta.Schema.Fields[1].IndexParams[0].Value, "col1_f2_iv1")
assert.Equal(t, collMeta.Schema.Fields[1].IndexParams[1].Value, "col1_f2_iv2")
// DescribeCollection
reqDescribe := &internalpb.DescribeCollectionRequest{
MsgType: internalpb.MsgType_kDescribeCollection,
ReqID: 1,
Timestamp: 11,
ProxyID: 1,
CollectionName: &servicepb.CollectionName{
CollectionName: "col1",
},
}
des, err := cli.DescribeCollection(ctx, reqDescribe)
assert.Nil(t, err)
assert.Equal(t, commonpb.ErrorCode_SUCCESS, des.Status.ErrorCode)
assert.Equal(t, "col1", des.Schema.Name)
assert.Equal(t, false, des.Schema.AutoID)
assert.Equal(t, 2, len(des.Schema.Fields))
assert.Equal(t, "col1_f1", des.Schema.Fields[0].Name)
assert.Equal(t, "col1_f2", des.Schema.Fields[1].Name)
assert.Equal(t, schemapb.DataType_VECTOR_FLOAT, des.Schema.Fields[0].DataType)
assert.Equal(t, schemapb.DataType_VECTOR_BINARY, des.Schema.Fields[1].DataType)
assert.Equal(t, 2, len(des.Schema.Fields[0].TypeParams))
assert.Equal(t, 2, len(des.Schema.Fields[0].IndexParams))
assert.Equal(t, 2, len(des.Schema.Fields[1].TypeParams))
assert.Equal(t, 2, len(des.Schema.Fields[1].IndexParams))
assert.Equal(t, "col1_f1_tk1", des.Schema.Fields[0].TypeParams[0].Key)
assert.Equal(t, "col1_f1_tv1", des.Schema.Fields[0].TypeParams[0].Value)
assert.Equal(t, "col1_f1_ik1", des.Schema.Fields[0].IndexParams[0].Key)
assert.Equal(t, "col1_f1_iv1", des.Schema.Fields[0].IndexParams[0].Value)
assert.Equal(t, "col1_f1_tk2", des.Schema.Fields[0].TypeParams[1].Key)
assert.Equal(t, "col1_f1_tv2", des.Schema.Fields[0].TypeParams[1].Value)
assert.Equal(t, "col1_f1_ik2", des.Schema.Fields[0].IndexParams[1].Key)
assert.Equal(t, "col1_f1_iv2", des.Schema.Fields[0].IndexParams[1].Value)
assert.Equal(t, "col1_f2_tk1", des.Schema.Fields[1].TypeParams[0].Key)
assert.Equal(t, "col1_f2_tv1", des.Schema.Fields[1].TypeParams[0].Value)
assert.Equal(t, "col1_f2_ik1", des.Schema.Fields[1].IndexParams[0].Key)
assert.Equal(t, "col1_f2_iv1", des.Schema.Fields[1].IndexParams[0].Value)
assert.Equal(t, "col1_f2_tk2", des.Schema.Fields[1].TypeParams[1].Key)
assert.Equal(t, "col1_f2_tv2", des.Schema.Fields[1].TypeParams[1].Value)
assert.Equal(t, "col1_f2_ik2", des.Schema.Fields[1].IndexParams[1].Key)
assert.Equal(t, "col1_f2_iv2", des.Schema.Fields[1].IndexParams[1].Value)
reqDescribe.CollectionName.CollectionName = "colNotExist"
des, err = cli.DescribeCollection(ctx, reqDescribe)
assert.Nil(t, err)
assert.NotEqual(t, commonpb.ErrorCode_SUCCESS, des.Status.ErrorCode)
log.Printf(des.Status.Reason)
reqDescribe.CollectionName.CollectionName = "col1"
reqDescribe.Timestamp = Timestamp(10)
des, err = cli.DescribeCollection(ctx, reqDescribe)
assert.Nil(t, err)
assert.NotEqual(t, commonpb.ErrorCode_SUCCESS, des.Status.ErrorCode)
log.Printf(des.Status.Reason)
req.Timestamp = Timestamp(10)
st, err = cli.CreateCollection(ctx, &req)
assert.Nil(t, err)

84
internal/master/grpc_service.go
View File

@ -24,25 +24,24 @@ func (s *Master) CreateCollection(ctx context.Context, in *internalpb.CreateColl
},
}
response := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR,
}
var err = s.scheduler.Enqueue(t)
if err != nil {
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR,
Reason: "Enqueue failed",
}, nil
response.Reason = "Enqueue failed: " + err.Error()
return response, nil
}
err = t.WaitToFinish(ctx)
if err != nil {
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR,
Reason: "create collection failed",
}, nil
response.Reason = "Create collection failed: " + err.Error()
return response, nil
}
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_SUCCESS,
}, nil
response.ErrorCode = commonpb.ErrorCode_SUCCESS
return response, nil
}
func (s *Master) DropCollection(ctx context.Context, in *internalpb.DropCollectionRequest) (*commonpb.Status, error) {
@ -55,25 +54,24 @@ func (s *Master) DropCollection(ctx context.Context, in *internalpb.DropCollecti
},
}
response := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR,
}
var err = s.scheduler.Enqueue(t)
if err != nil {
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR,
Reason: "Enqueue failed",
}, nil
response.Reason = "Enqueue failed: " + err.Error()
return response, nil
}
err = t.WaitToFinish(ctx)
if err != nil {
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR,
Reason: "WaitToFinish failed",
}, nil
response.Reason = "Drop collection failed: " + err.Error()
return response, nil
}
return &commonpb.Status{
ErrorCode: commonpb.ErrorCode_SUCCESS,
}, nil
response.ErrorCode = commonpb.ErrorCode_SUCCESS
return response, nil
}
func (s *Master) HasCollection(ctx context.Context, in *internalpb.HasCollectionRequest) (*servicepb.BoolResponse, error) {
@ -98,13 +96,13 @@ func (s *Master) HasCollection(ctx context.Context, in *internalpb.HasCollection
var err = s.scheduler.Enqueue(t)
if err != nil {
st.Reason = "Enqueue failed"
st.Reason = "Enqueue failed: " + err.Error()
return response, nil
}
err = t.WaitToFinish(ctx)
if err != nil {
st.Reason = "WaitToFinish failed"
st.Reason = "Has collection failed: " + err.Error()
return response, nil
}
@ -124,19 +122,29 @@ func (s *Master) DescribeCollection(ctx context.Context, in *internalpb.Describe
description: nil,
}
response := &servicepb.CollectionDescription{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR,
},
Schema: nil,
}
t.(*describeCollectionTask).description = response
var err = s.scheduler.Enqueue(t)
if err != nil {
err := errors.New("Enqueue failed")
return t.(*describeCollectionTask).description, err
response.Status.Reason = "Enqueue failed: " + err.Error()
return response, nil
}
err = t.WaitToFinish(ctx)
if err != nil {
err := errors.New("WaitToFinish failed")
return t.(*describeCollectionTask).description, err
response.Status.Reason = "Describe collection failed: " + err.Error()
return response, nil
}
return t.(*describeCollectionTask).description, nil
response.Status.ErrorCode = commonpb.ErrorCode_SUCCESS
return response, nil
}
func (s *Master) ShowCollections(ctx context.Context, in *internalpb.ShowCollectionRequest) (*servicepb.StringListResponse, error) {
@ -320,14 +328,24 @@ func (s *Master) ShowPartitions(ctx context.Context, in *internalpb.ShowPartitio
var err = s.scheduler.Enqueue(t)
if err != nil {
err := errors.New("Enqueue failed")
return t.(*showPartitionTask).stringListResponse, err
return &servicepb.StringListResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR,
Reason: "Enqueue failed",
},
Values: nil,
}, nil
}
err = t.WaitToFinish(ctx)
if err != nil {
err := errors.New("WaitToFinish failed")
return t.(*showPartitionTask).stringListResponse, err
return &servicepb.StringListResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UNEXPECTED_ERROR,
Reason: "WaitToFinish failed",
},
Values: nil,
}, nil
}
return t.(*showPartitionTask).stringListResponse, nil

7
internal/master/partition_task.go
View File

@ -147,7 +147,7 @@ func (t *describePartitionTask) Ts() (Timestamp, error) {
if t.req == nil {
return 0, errors.New("null request")
}
return Timestamp(t.req.Timestamp), nil
return t.req.Timestamp, nil
}
func (t *describePartitionTask) Execute() error {
@ -161,7 +161,8 @@ func (t *describePartitionTask) Execute() error {
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_SUCCESS,
},
Name: partitionName,
Name: partitionName,
Statistics: nil,
}
t.description = &description
@ -182,7 +183,7 @@ func (t *showPartitionTask) Ts() (Timestamp, error) {
if t.req == nil {
return 0, errors.New("null request")
}
return Timestamp(t.req.Timestamp), nil
return t.req.Timestamp, nil
}
func (t *showPartitionTask) Execute() error {

73
internal/master/partition_task_test.go
View File

@ -118,7 +118,6 @@ func TestMaster_Partition(t *testing.T) {
st, _ := cli.CreateCollection(ctx, &createCollectionReq)
assert.NotNil(t, st)
assert.Equal(t, commonpb.ErrorCode_SUCCESS, st.ErrorCode)
assert.Equal(t, st.ErrorCode, commonpb.ErrorCode_SUCCESS)
createPartitionReq := internalpb.CreatePartitionRequest{
MsgType: internalpb.MsgType_kCreatePartition,
@ -130,7 +129,17 @@ func TestMaster_Partition(t *testing.T) {
st, _ = cli.CreatePartition(ctx, &createPartitionReq)
assert.NotNil(t, st)
assert.Equal(t, commonpb.ErrorCode_SUCCESS, st.ErrorCode)
assert.Equal(t, st.ErrorCode, commonpb.ErrorCode_SUCCESS)
createPartitionReq = internalpb.CreatePartitionRequest{
MsgType: internalpb.MsgType_kCreatePartition,
ReqID: 1,
Timestamp: 1,
ProxyID: 1,
PartitionName: &servicepb.PartitionName{CollectionName: "col1", Tag: "partition1"},
}
st, _ = cli.CreatePartition(ctx, &createPartitionReq)
assert.NotNil(t, st)
assert.Equal(t, commonpb.ErrorCode_UNEXPECTED_ERROR, st.ErrorCode)
createPartitionReq = internalpb.CreatePartitionRequest{
MsgType: internalpb.MsgType_kCreatePartition,
@ -142,7 +151,6 @@ func TestMaster_Partition(t *testing.T) {
st, _ = cli.CreatePartition(ctx, &createPartitionReq)
assert.NotNil(t, st)
assert.Equal(t, commonpb.ErrorCode_SUCCESS, st.ErrorCode)
assert.Equal(t, st.ErrorCode, commonpb.ErrorCode_SUCCESS)
collMeta, err := svr.mt.GetCollectionByName(sch.Name)
assert.Nil(t, err)
@ -192,6 +200,18 @@ func TestMaster_Partition(t *testing.T) {
assert.Nil(t, err)
assert.ElementsMatch(t, []string{"partition1", "partition2"}, stringList.Values)
showPartitionReq = internalpb.ShowPartitionRequest{
MsgType: internalpb.MsgType_kShowPartitions,
ReqID: 1,
Timestamp: 3,
ProxyID: 1,
CollectionName: &servicepb.CollectionName{CollectionName: "col1"},
}
stringList, _ = cli.ShowPartitions(ctx, &showPartitionReq)
assert.NotNil(t, stringList)
assert.Equal(t, commonpb.ErrorCode_UNEXPECTED_ERROR, stringList.Status.ErrorCode)
hasPartitionReq := internalpb.HasPartitionRequest{
MsgType: internalpb.MsgType_kHasPartition,
ReqID: 1,
@ -204,6 +224,18 @@ func TestMaster_Partition(t *testing.T) {
assert.Nil(t, err)
assert.True(t, hasPartition.Value)
hasPartitionReq = internalpb.HasPartitionRequest{
MsgType: internalpb.MsgType_kHasPartition,
ReqID: 1,
Timestamp: 4,
ProxyID: 1,
PartitionName: &servicepb.PartitionName{CollectionName: "col1", Tag: "partition1"},
}
hasPartition, _ = cli.HasPartition(ctx, &hasPartitionReq)
assert.NotNil(t, hasPartition)
assert.Equal(t, commonpb.ErrorCode_UNEXPECTED_ERROR, stringList.Status.ErrorCode)
hasPartitionReq = internalpb.HasPartitionRequest{
MsgType: internalpb.MsgType_kHasPartition,
ReqID: 1,
@ -228,6 +260,18 @@ func TestMaster_Partition(t *testing.T) {
assert.Nil(t, err)
assert.Equal(t, commonpb.ErrorCode_SUCCESS, st.ErrorCode)
deletePartitionReq = internalpb.DropPartitionRequest{
MsgType: internalpb.MsgType_kDropPartition,
ReqID: 1,
Timestamp: 6,
ProxyID: 1,
PartitionName: &servicepb.PartitionName{CollectionName: "col1", Tag: "partition2"},
}
st, _ = cli.DropPartition(ctx, &deletePartitionReq)
assert.NotNil(t, st)
assert.Equal(t, commonpb.ErrorCode_UNEXPECTED_ERROR, st.ErrorCode)
hasPartitionReq = internalpb.HasPartitionRequest{
MsgType: internalpb.MsgType_kHasPartition,
ReqID: 1,
@ -240,5 +284,28 @@ func TestMaster_Partition(t *testing.T) {
assert.Nil(t, err)
assert.False(t, hasPartition.Value)
describePartitionReq := internalpb.DescribePartitionRequest{
MsgType: internalpb.MsgType_kDescribePartition,
ReqID: 1,
Timestamp: 9,
ProxyID: 1,
PartitionName: &servicepb.PartitionName{CollectionName: "col1", Tag: "partition1"},
}
describePartition, err := cli.DescribePartition(ctx, &describePartitionReq)
assert.Nil(t, err)
assert.Equal(t, &servicepb.PartitionName{CollectionName: "col1", Tag: "partition1"}, describePartition.Name)
describePartitionReq = internalpb.DescribePartitionRequest{
MsgType: internalpb.MsgType_kDescribePartition,
ReqID: 1,
Timestamp: 8,
ProxyID: 1,
PartitionName: &servicepb.PartitionName{CollectionName: "col1", Tag: "partition1"},
}
describePartition, _ = cli.DescribePartition(ctx, &describePartitionReq)
assert.Equal(t, commonpb.ErrorCode_UNEXPECTED_ERROR, describePartition.Status.ErrorCode)
svr.Close()
}