test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-11-30 02:48:45 +08:00
Code Issues Actions 11 Packages Projects Releases Wiki Activity

fix releasePartition not work after loadCollection (#6025)

Browse Source 
Signed-off-by: xige-16 <xi.ge@zilliz.com>
This commit is contained in:
xige-16 2021-06-23 17:44:12 +08:00 committed by GitHub
parent 1f524bdf3e
commit 5417fee45f
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
12 changed files with 496 additions and 552 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
internal/distributed/querycoord/client/client.go
View File

@ -180,13 +180,6 @@ func (c *Client) GetStatisticsChannel(ctx context.Context) (*milvuspb.StringResp
return ret.(*milvuspb.StringResponse), err
}
func (c *Client) RegisterNode(ctx context.Context, req *querypb.RegisterNodeRequest) (*querypb.RegisterNodeResponse, error) {
ret, err := c.recall(func() (interface{}, error) {
return c.grpcClient.RegisterNode(ctx, req)
})
return ret.(*querypb.RegisterNodeResponse), err
}
func (c *Client) ShowCollections(ctx context.Context, req *querypb.ShowCollectionsRequest) (*querypb.ShowCollectionsResponse, error) {
ret, err := c.recall(func() (interface{}, error) {
return c.grpcClient.ShowCollections(ctx, req)

4
internal/distributed/querycoord/service.go
View File

@ -242,10 +242,6 @@ func (s *Server) GetStatisticsChannel(ctx context.Context, req *internalpb.GetSt
return s.queryCoord.GetStatisticsChannel(ctx)
}
func (s *Server) RegisterNode(ctx context.Context, req *querypb.RegisterNodeRequest) (*querypb.RegisterNodeResponse, error) {
return s.queryCoord.RegisterNode(ctx, req)
}
func (s *Server) ShowCollections(ctx context.Context, req *querypb.ShowCollectionsRequest) (*querypb.ShowCollectionsResponse, error) {
return s.queryCoord.ShowCollections(ctx, req)
}

14
internal/distributed/querynode/mock.go
View File

@ -22,7 +22,6 @@ import (
"github.com/milvus-io/milvus/internal/proto/indexpb"
internalPb "github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/proto/milvuspb"
"github.com/milvus-io/milvus/internal/proto/querypb"
)
const (
@ -192,16 +191,3 @@ func (index *IndexCoordMock) GetIndexFilePaths(req *indexpb.GetIndexFilePathsReq
}
return rsp, nil
}
type queryCoordMock struct{}
func (q *queryCoordMock) RegisterNode(req *querypb.RegisterNodeRequest) (*querypb.RegisterNodeResponse, error) {
return &querypb.RegisterNodeResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
},
InitParams: &internalPb.InitParams{
NodeID: int64(0),
},
}, nil
}

13
internal/proto/query_coord.proto
View File

@ -15,8 +15,6 @@ service QueryCoord {
rpc GetTimeTickChannel(internal.GetTimeTickChannelRequest) returns(milvus.StringResponse) {}
rpc GetStatisticsChannel(internal.GetStatisticsChannelRequest) returns(milvus.StringResponse){}
rpc RegisterNode(RegisterNodeRequest) returns (RegisterNodeResponse) {}
rpc ShowCollections(ShowCollectionsRequest) returns (ShowCollectionsResponse) {}
rpc ShowPartitions(ShowPartitionsRequest) returns (ShowPartitionsResponse) {}
@ -46,16 +44,6 @@ service QueryNode {
}
//--------------------query coordinator proto------------------
message RegisterNodeRequest {
common.MsgBase base = 1;
common.Address address = 2;
}
message RegisterNodeResponse {
common.Status status = 1;
internal.InitParams init_params = 2;
}
message ShowCollectionsRequest {
common.MsgBase base = 1;
int64 dbID = 2;
@ -270,6 +258,7 @@ message CollectionInfo {
repeated DmChannelInfo channel_infos = 3;
bool load_collection = 4;
schema.CollectionSchema schema = 5;
repeated int64 released_partitionIDs = 6;
}
message HandoffSegments {

432
internal/proto/querypb/query_coord.pb.go
View File

@ -136,100 +136,6 @@ func (SegmentState) EnumDescriptor() ([]byte, []int) {
}
//--------------------query coordinator proto------------------
type RegisterNodeRequest struct {
Base *commonpb.MsgBase `protobuf:"bytes,1,opt,name=base,proto3" json:"base,omitempty"`
Address *commonpb.Address `protobuf:"bytes,2,opt,name=address,proto3" json:"address,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *RegisterNodeRequest) Reset() { *m = RegisterNodeRequest{} }
func (m *RegisterNodeRequest) String() string { return proto.CompactTextString(m) }
func (*RegisterNodeRequest) ProtoMessage() {}
func (*RegisterNodeRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{0}
}
func (m *RegisterNodeRequest) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_RegisterNodeRequest.Unmarshal(m, b)
}
func (m *RegisterNodeRequest) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
return xxx_messageInfo_RegisterNodeRequest.Marshal(b, m, deterministic)
}
func (m *RegisterNodeRequest) XXX_Merge(src proto.Message) {
xxx_messageInfo_RegisterNodeRequest.Merge(m, src)
}
func (m *RegisterNodeRequest) XXX_Size() int {
return xxx_messageInfo_RegisterNodeRequest.Size(m)
}
func (m *RegisterNodeRequest) XXX_DiscardUnknown() {
xxx_messageInfo_RegisterNodeRequest.DiscardUnknown(m)
}
var xxx_messageInfo_RegisterNodeRequest proto.InternalMessageInfo
func (m *RegisterNodeRequest) GetBase() *commonpb.MsgBase {
if m != nil {
return m.Base
}
return nil
}
func (m *RegisterNodeRequest) GetAddress() *commonpb.Address {
if m != nil {
return m.Address
}
return nil
}
type RegisterNodeResponse struct {
Status *commonpb.Status `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"`
InitParams *internalpb.InitParams `protobuf:"bytes,2,opt,name=init_params,json=initParams,proto3" json:"init_params,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *RegisterNodeResponse) Reset() { *m = RegisterNodeResponse{} }
func (m *RegisterNodeResponse) String() string { return proto.CompactTextString(m) }
func (*RegisterNodeResponse) ProtoMessage() {}
func (*RegisterNodeResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{1}
}
func (m *RegisterNodeResponse) XXX_Unmarshal(b []byte) error {
return xxx_messageInfo_RegisterNodeResponse.Unmarshal(m, b)
}
func (m *RegisterNodeResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
return xxx_messageInfo_RegisterNodeResponse.Marshal(b, m, deterministic)
}
func (m *RegisterNodeResponse) XXX_Merge(src proto.Message) {
xxx_messageInfo_RegisterNodeResponse.Merge(m, src)
}
func (m *RegisterNodeResponse) XXX_Size() int {
return xxx_messageInfo_RegisterNodeResponse.Size(m)
}
func (m *RegisterNodeResponse) XXX_DiscardUnknown() {
xxx_messageInfo_RegisterNodeResponse.DiscardUnknown(m)
}
var xxx_messageInfo_RegisterNodeResponse proto.InternalMessageInfo
func (m *RegisterNodeResponse) GetStatus() *commonpb.Status {
if m != nil {
return m.Status
}
return nil
}
func (m *RegisterNodeResponse) GetInitParams() *internalpb.InitParams {
if m != nil {
return m.InitParams
}
return nil
}
type ShowCollectionsRequest struct {
Base *commonpb.MsgBase `protobuf:"bytes,1,opt,name=base,proto3" json:"base,omitempty"`
DbID int64 `protobuf:"varint,2,opt,name=dbID,proto3" json:"dbID,omitempty"`
@ -242,7 +148,7 @@ func (m *ShowCollectionsRequest) Reset() { *m = ShowCollectionsRequest{}
func (m *ShowCollectionsRequest) String() string { return proto.CompactTextString(m) }
func (*ShowCollectionsRequest) ProtoMessage() {}
func (*ShowCollectionsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{2}
return fileDescriptor_aab7cc9a69ed26e8, []int{0}
}
func (m *ShowCollectionsRequest) XXX_Unmarshal(b []byte) error {
@ -289,7 +195,7 @@ func (m *ShowCollectionsResponse) Reset() { *m = ShowCollectionsResponse
func (m *ShowCollectionsResponse) String() string { return proto.CompactTextString(m) }
func (*ShowCollectionsResponse) ProtoMessage() {}
func (*ShowCollectionsResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{3}
return fileDescriptor_aab7cc9a69ed26e8, []int{1}
}
func (m *ShowCollectionsResponse) XXX_Unmarshal(b []byte) error {
@ -337,7 +243,7 @@ func (m *ShowPartitionsRequest) Reset() { *m = ShowPartitionsRequest{} }
func (m *ShowPartitionsRequest) String() string { return proto.CompactTextString(m) }
func (*ShowPartitionsRequest) ProtoMessage() {}
func (*ShowPartitionsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{4}
return fileDescriptor_aab7cc9a69ed26e8, []int{2}
}
func (m *ShowPartitionsRequest) XXX_Unmarshal(b []byte) error {
@ -391,7 +297,7 @@ func (m *ShowPartitionsResponse) Reset() { *m = ShowPartitionsResponse{}
func (m *ShowPartitionsResponse) String() string { return proto.CompactTextString(m) }
func (*ShowPartitionsResponse) ProtoMessage() {}
func (*ShowPartitionsResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{5}
return fileDescriptor_aab7cc9a69ed26e8, []int{3}
}
func (m *ShowPartitionsResponse) XXX_Unmarshal(b []byte) error {
@ -440,7 +346,7 @@ func (m *LoadCollectionRequest) Reset() { *m = LoadCollectionRequest{} }
func (m *LoadCollectionRequest) String() string { return proto.CompactTextString(m) }
func (*LoadCollectionRequest) ProtoMessage() {}
func (*LoadCollectionRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{6}
return fileDescriptor_aab7cc9a69ed26e8, []int{4}
}
func (m *LoadCollectionRequest) XXX_Unmarshal(b []byte) error {
@ -503,7 +409,7 @@ func (m *ReleaseCollectionRequest) Reset() { *m = ReleaseCollectionReque
func (m *ReleaseCollectionRequest) String() string { return proto.CompactTextString(m) }
func (*ReleaseCollectionRequest) ProtoMessage() {}
func (*ReleaseCollectionRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{7}
return fileDescriptor_aab7cc9a69ed26e8, []int{5}
}
func (m *ReleaseCollectionRequest) XXX_Unmarshal(b []byte) error {
@ -567,7 +473,7 @@ func (m *LoadPartitionsRequest) Reset() { *m = LoadPartitionsRequest{} }
func (m *LoadPartitionsRequest) String() string { return proto.CompactTextString(m) }
func (*LoadPartitionsRequest) ProtoMessage() {}
func (*LoadPartitionsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{8}
return fileDescriptor_aab7cc9a69ed26e8, []int{6}
}
func (m *LoadPartitionsRequest) XXX_Unmarshal(b []byte) error {
@ -638,7 +544,7 @@ func (m *ReleasePartitionsRequest) Reset() { *m = ReleasePartitionsReque
func (m *ReleasePartitionsRequest) String() string { return proto.CompactTextString(m) }
func (*ReleasePartitionsRequest) ProtoMessage() {}
func (*ReleasePartitionsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{9}
return fileDescriptor_aab7cc9a69ed26e8, []int{7}
}
func (m *ReleasePartitionsRequest) XXX_Unmarshal(b []byte) error {
@ -706,7 +612,7 @@ func (m *CreateQueryChannelRequest) Reset() { *m = CreateQueryChannelReq
func (m *CreateQueryChannelRequest) String() string { return proto.CompactTextString(m) }
func (*CreateQueryChannelRequest) ProtoMessage() {}
func (*CreateQueryChannelRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{10}
return fileDescriptor_aab7cc9a69ed26e8, []int{8}
}
func (m *CreateQueryChannelRequest) XXX_Unmarshal(b []byte) error {
@ -754,7 +660,7 @@ func (m *CreateQueryChannelResponse) Reset() { *m = CreateQueryChannelRe
func (m *CreateQueryChannelResponse) String() string { return proto.CompactTextString(m) }
func (*CreateQueryChannelResponse) ProtoMessage() {}
func (*CreateQueryChannelResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{11}
return fileDescriptor_aab7cc9a69ed26e8, []int{9}
}
func (m *CreateQueryChannelResponse) XXX_Unmarshal(b []byte) error {
@ -810,7 +716,7 @@ func (m *GetPartitionStatesRequest) Reset() { *m = GetPartitionStatesReq
func (m *GetPartitionStatesRequest) String() string { return proto.CompactTextString(m) }
func (*GetPartitionStatesRequest) ProtoMessage() {}
func (*GetPartitionStatesRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{12}
return fileDescriptor_aab7cc9a69ed26e8, []int{10}
}
func (m *GetPartitionStatesRequest) XXX_Unmarshal(b []byte) error {
@ -871,7 +777,7 @@ func (m *PartitionStates) Reset() { *m = PartitionStates{} }
func (m *PartitionStates) String() string { return proto.CompactTextString(m) }
func (*PartitionStates) ProtoMessage() {}
func (*PartitionStates) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{13}
return fileDescriptor_aab7cc9a69ed26e8, []int{11}
}
func (m *PartitionStates) XXX_Unmarshal(b []byte) error {
@ -918,7 +824,7 @@ func (m *GetPartitionStatesResponse) Reset() { *m = GetPartitionStatesRe
func (m *GetPartitionStatesResponse) String() string { return proto.CompactTextString(m) }
func (*GetPartitionStatesResponse) ProtoMessage() {}
func (*GetPartitionStatesResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{14}
return fileDescriptor_aab7cc9a69ed26e8, []int{12}
}
func (m *GetPartitionStatesResponse) XXX_Unmarshal(b []byte) error {
@ -965,7 +871,7 @@ func (m *GetSegmentInfoRequest) Reset() { *m = GetSegmentInfoRequest{} }
func (m *GetSegmentInfoRequest) String() string { return proto.CompactTextString(m) }
func (*GetSegmentInfoRequest) ProtoMessage() {}
func (*GetSegmentInfoRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{15}
return fileDescriptor_aab7cc9a69ed26e8, []int{13}
}
func (m *GetSegmentInfoRequest) XXX_Unmarshal(b []byte) error {
@ -1020,7 +926,7 @@ func (m *SegmentInfo) Reset() { *m = SegmentInfo{} }
func (m *SegmentInfo) String() string { return proto.CompactTextString(m) }
func (*SegmentInfo) ProtoMessage() {}
func (*SegmentInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{16}
return fileDescriptor_aab7cc9a69ed26e8, []int{14}
}
func (m *SegmentInfo) XXX_Unmarshal(b []byte) error {
@ -1123,7 +1029,7 @@ func (m *GetSegmentInfoResponse) Reset() { *m = GetSegmentInfoResponse{}
func (m *GetSegmentInfoResponse) String() string { return proto.CompactTextString(m) }
func (*GetSegmentInfoResponse) ProtoMessage() {}
func (*GetSegmentInfoResponse) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{17}
return fileDescriptor_aab7cc9a69ed26e8, []int{15}
}
func (m *GetSegmentInfoResponse) XXX_Unmarshal(b []byte) error {
@ -1174,7 +1080,7 @@ func (m *AddQueryChannelRequest) Reset() { *m = AddQueryChannelRequest{}
func (m *AddQueryChannelRequest) String() string { return proto.CompactTextString(m) }
func (*AddQueryChannelRequest) ProtoMessage() {}
func (*AddQueryChannelRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{18}
return fileDescriptor_aab7cc9a69ed26e8, []int{16}
}
func (m *AddQueryChannelRequest) XXX_Unmarshal(b []byte) error {
@ -1245,7 +1151,7 @@ func (m *RemoveQueryChannelRequest) Reset() { *m = RemoveQueryChannelReq
func (m *RemoveQueryChannelRequest) String() string { return proto.CompactTextString(m) }
func (*RemoveQueryChannelRequest) ProtoMessage() {}
func (*RemoveQueryChannelRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{19}
return fileDescriptor_aab7cc9a69ed26e8, []int{17}
}
func (m *RemoveQueryChannelRequest) XXX_Unmarshal(b []byte) error {
@ -1318,7 +1224,7 @@ func (m *WatchDmChannelsRequest) Reset() { *m = WatchDmChannelsRequest{}
func (m *WatchDmChannelsRequest) String() string { return proto.CompactTextString(m) }
func (*WatchDmChannelsRequest) ProtoMessage() {}
func (*WatchDmChannelsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{20}
return fileDescriptor_aab7cc9a69ed26e8, []int{18}
}
func (m *WatchDmChannelsRequest) XXX_Unmarshal(b []byte) error {
@ -1405,7 +1311,7 @@ func (m *SegmentLoadInfo) Reset() { *m = SegmentLoadInfo{} }
func (m *SegmentLoadInfo) String() string { return proto.CompactTextString(m) }
func (*SegmentLoadInfo) ProtoMessage() {}
func (*SegmentLoadInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{21}
return fileDescriptor_aab7cc9a69ed26e8, []int{19}
}
func (m *SegmentLoadInfo) XXX_Unmarshal(b []byte) error {
@ -1483,7 +1389,7 @@ func (m *LoadSegmentsRequest) Reset() { *m = LoadSegmentsRequest{} }
func (m *LoadSegmentsRequest) String() string { return proto.CompactTextString(m) }
func (*LoadSegmentsRequest) ProtoMessage() {}
func (*LoadSegmentsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{22}
return fileDescriptor_aab7cc9a69ed26e8, []int{20}
}
func (m *LoadSegmentsRequest) XXX_Unmarshal(b []byte) error {
@ -1555,7 +1461,7 @@ func (m *ReleaseSegmentsRequest) Reset() { *m = ReleaseSegmentsRequest{}
func (m *ReleaseSegmentsRequest) String() string { return proto.CompactTextString(m) }
func (*ReleaseSegmentsRequest) ProtoMessage() {}
func (*ReleaseSegmentsRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{23}
return fileDescriptor_aab7cc9a69ed26e8, []int{21}
}
func (m *ReleaseSegmentsRequest) XXX_Unmarshal(b []byte) error {
@ -1630,7 +1536,7 @@ func (m *DmChannelInfo) Reset() { *m = DmChannelInfo{} }
func (m *DmChannelInfo) String() string { return proto.CompactTextString(m) }
func (*DmChannelInfo) ProtoMessage() {}
func (*DmChannelInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{24}
return fileDescriptor_aab7cc9a69ed26e8, []int{22}
}
func (m *DmChannelInfo) XXX_Unmarshal(b []byte) error {
@ -1678,7 +1584,7 @@ func (m *QueryChannelInfo) Reset() { *m = QueryChannelInfo{} }
func (m *QueryChannelInfo) String() string { return proto.CompactTextString(m) }
func (*QueryChannelInfo) ProtoMessage() {}
func (*QueryChannelInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{25}
return fileDescriptor_aab7cc9a69ed26e8, []int{23}
}
func (m *QueryChannelInfo) XXX_Unmarshal(b []byte) error {
@ -1726,6 +1632,7 @@ type CollectionInfo struct {
ChannelInfos []*DmChannelInfo `protobuf:"bytes,3,rep,name=channel_infos,json=channelInfos,proto3" json:"channel_infos,omitempty"`
LoadCollection bool `protobuf:"varint,4,opt,name=load_collection,json=loadCollection,proto3" json:"load_collection,omitempty"`
Schema *schemapb.CollectionSchema `protobuf:"bytes,5,opt,name=schema,proto3" json:"schema,omitempty"`
ReleasedPartitionIDs []int64 `protobuf:"varint,6,rep,packed,name=released_partitionIDs,json=releasedPartitionIDs,proto3" json:"released_partitionIDs,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
@ -1735,7 +1642,7 @@ func (m *CollectionInfo) Reset() { *m = CollectionInfo{} }
func (m *CollectionInfo) String() string { return proto.CompactTextString(m) }
func (*CollectionInfo) ProtoMessage() {}
func (*CollectionInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{26}
return fileDescriptor_aab7cc9a69ed26e8, []int{24}
}
func (m *CollectionInfo) XXX_Unmarshal(b []byte) error {
@ -1791,6 +1698,13 @@ func (m *CollectionInfo) GetSchema() *schemapb.CollectionSchema {
return nil
}
func (m *CollectionInfo) GetReleasedPartitionIDs() []int64 {
if m != nil {
return m.ReleasedPartitionIDs
}
return nil
}
type HandoffSegments struct {
Base *commonpb.MsgBase `protobuf:"bytes,1,opt,name=base,proto3" json:"base,omitempty"`
Infos []*SegmentLoadInfo `protobuf:"bytes,2,rep,name=infos,proto3" json:"infos,omitempty"`
@ -1803,7 +1717,7 @@ func (m *HandoffSegments) Reset() { *m = HandoffSegments{} }
func (m *HandoffSegments) String() string { return proto.CompactTextString(m) }
func (*HandoffSegments) ProtoMessage() {}
func (*HandoffSegments) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{27}
return fileDescriptor_aab7cc9a69ed26e8, []int{25}
}
func (m *HandoffSegments) XXX_Unmarshal(b []byte) error {
@ -1856,7 +1770,7 @@ func (m *LoadBalanceSegmentInfo) Reset() { *m = LoadBalanceSegmentInfo{}
func (m *LoadBalanceSegmentInfo) String() string { return proto.CompactTextString(m) }
func (*LoadBalanceSegmentInfo) ProtoMessage() {}
func (*LoadBalanceSegmentInfo) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{28}
return fileDescriptor_aab7cc9a69ed26e8, []int{26}
}
func (m *LoadBalanceSegmentInfo) XXX_Unmarshal(b []byte) error {
@ -1946,7 +1860,7 @@ func (m *LoadBalanceRequest) Reset() { *m = LoadBalanceRequest{} }
func (m *LoadBalanceRequest) String() string { return proto.CompactTextString(m) }
func (*LoadBalanceRequest) ProtoMessage() {}
func (*LoadBalanceRequest) Descriptor() ([]byte, []int) {
return fileDescriptor_aab7cc9a69ed26e8, []int{29}
return fileDescriptor_aab7cc9a69ed26e8, []int{27}
}
func (m *LoadBalanceRequest) XXX_Unmarshal(b []byte) error {
@ -1992,8 +1906,6 @@ func init() {
proto.RegisterEnum("milvus.proto.query.PartitionState", PartitionState_name, PartitionState_value)
proto.RegisterEnum("milvus.proto.query.TriggerCondition", TriggerCondition_name, TriggerCondition_value)
proto.RegisterEnum("milvus.proto.query.SegmentState", SegmentState_name, SegmentState_value)
proto.RegisterType((*RegisterNodeRequest)(nil), "milvus.proto.query.RegisterNodeRequest")
proto.RegisterType((*RegisterNodeResponse)(nil), "milvus.proto.query.RegisterNodeResponse")
proto.RegisterType((*ShowCollectionsRequest)(nil), "milvus.proto.query.ShowCollectionsRequest")
proto.RegisterType((*ShowCollectionsResponse)(nil), "milvus.proto.query.ShowCollectionsResponse")
proto.RegisterType((*ShowPartitionsRequest)(nil), "milvus.proto.query.ShowPartitionsRequest")
@ -2027,126 +1939,122 @@ func init() {
func init() { proto.RegisterFile("query_coord.proto", fileDescriptor_aab7cc9a69ed26e8) }
var fileDescriptor_aab7cc9a69ed26e8 = []byte{
// 1896 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xec, 0x59, 0x4f, 0x73, 0x1c, 0x47,
0x15, 0xd7, 0xec, 0xae, 0x56, 0xda, 0xb7, 0xff, 0xc6, 0x6d, 0x5b, 0xac, 0x97, 0x38, 0x71, 0xc6,
0x31, 0x76, 0x14, 0x22, 0xa7, 0x94, 0x40, 0x91, 0x03, 0x07, 0x4b, 0x1b, 0x8b, 0x85, 0x44, 0x11,
0x23, 0x13, 0x0a, 0x97, 0x8b, 0x61, 0x76, 0xa6, 0xb5, 0x3b, 0x95, 0x99, 0xee, 0xd5, 0xf4, 0xac,
0x65, 0xfb, 0x00, 0x45, 0x15, 0x07, 0x6e, 0x9c, 0x38, 0xc1, 0x85, 0x0b, 0x55, 0x1c, 0xf8, 0x02,
0x9c, 0xf2, 0x45, 0xa0, 0x8a, 0x22, 0x37, 0x8e, 0x7c, 0x01, 0x6a, 0xba, 0x7b, 0x66, 0xe7, 0x4f,
0xaf, 0xb4, 0x96, 0xec, 0x38, 0x45, 0x71, 0xdb, 0x7e, 0xfd, 0xfa, 0xfd, 0xef, 0xdf, 0xbc, 0x7e,
0x0b, 0x97, 0x8e, 0x67, 0x38, 0x7c, 0x6a, 0x39, 0x94, 0x86, 0xee, 0xd6, 0x34, 0xa4, 0x11, 0x45,
0x28, 0xf0, 0xfc, 0xc7, 0x33, 0x26, 0x56, 0x5b, 0x7c, 0xbf, 0xdf, 0x72, 0x68, 0x10, 0x50, 0x22,
0x68, 0xfd, 0x56, 0x96, 0xa3, 0xdf, 0xf1, 0x48, 0x84, 0x43, 0x62, 0xfb, 0xc9, 0x2e, 0x73, 0x26,
0x38, 0xb0, 0xe5, 0x4a, 0x77, 0xed, 0xc8, 0xce, 0xca, 0x37, 0x7e, 0x05, 0x97, 0x4d, 0x3c, 0xf6,
0x58, 0x84, 0xc3, 0x7d, 0xea, 0x62, 0x13, 0x1f, 0xcf, 0x30, 0x8b, 0xd0, 0x7b, 0x50, 0x1b, 0xd9,
0x0c, 0xf7, 0xb4, 0x1b, 0xda, 0x9d, 0xe6, 0xf6, 0x6b, 0x5b, 0x39, 0x2b, 0xa4, 0xfa, 0x4f, 0xd8,
0x78, 0xc7, 0x66, 0xd8, 0xe4, 0x9c, 0xe8, 0xbb, 0xb0, 0x66, 0xbb, 0x6e, 0x88, 0x19, 0xeb, 0x55,
0x4e, 0x39, 0x74, 0x4f, 0xf0, 0x98, 0x09, 0xb3, 0xf1, 0x3b, 0x0d, 0xae, 0xe4, 0x2d, 0x60, 0x53,
0x4a, 0x18, 0x46, 0xef, 0x43, 0x9d, 0x45, 0x76, 0x34, 0x63, 0xd2, 0x88, 0x6f, 0x2a, 0xe5, 0x1d,
0x72, 0x16, 0x53, 0xb2, 0xa2, 0x1d, 0x68, 0x7a, 0xc4, 0x8b, 0xac, 0xa9, 0x1d, 0xda, 0x41, 0x62,
0xc9, 0x9b, 0xf9, 0x93, 0x69, 0x84, 0x86, 0xc4, 0x8b, 0x0e, 0x38, 0xa3, 0x09, 0x5e, 0xfa, 0xdb,
0xf8, 0x39, 0x6c, 0x1c, 0x4e, 0xe8, 0xc9, 0x2e, 0xf5, 0x7d, 0xec, 0x44, 0x1e, 0x25, 0xec, 0xfc,
0x51, 0x41, 0x50, 0x73, 0x47, 0xc3, 0x01, 0x37, 0xa4, 0x6a, 0xf2, 0xdf, 0x46, 0x04, 0xdf, 0x28,
0xc9, 0xbf, 0x88, 0xcf, 0x6f, 0x41, 0xdb, 0x49, 0x65, 0x0d, 0x07, 0xb1, 0xd7, 0xd5, 0x3b, 0x55,
0x33, 0x4f, 0x34, 0x7e, 0xad, 0xc1, 0xd5, 0x58, 0xed, 0x81, 0x1d, 0x46, 0xde, 0x8b, 0xf7, 0x0a,
0x19, 0xd0, 0xca, 0x2a, 0xec, 0x55, 0xf9, 0x5e, 0x8e, 0x66, 0x1c, 0x8b, 0xc8, 0x66, 0x4d, 0xb8,
0x88, 0xe3, 0x06, 0xb4, 0xa6, 0x89, 0xa8, 0xb9, 0xdf, 0x39, 0x9a, 0xf1, 0x85, 0x06, 0x57, 0x3f,
0xa6, 0xb6, 0x3b, 0x8f, 0xf6, 0x57, 0xee, 0x36, 0xfa, 0x3e, 0xd4, 0xc5, 0x2d, 0xec, 0xd5, 0xb8,
0xae, 0x5b, 0x79, 0x5d, 0xf2, 0x86, 0xce, 0x2d, 0x3c, 0xe4, 0x04, 0x53, 0x1e, 0x32, 0xfe, 0xa8,
0x41, 0xcf, 0xc4, 0x3e, 0xb6, 0x19, 0x7e, 0x95, 0x5e, 0x6c, 0x40, 0x9d, 0x50, 0x17, 0x0f, 0x07,
0xdc, 0x8b, 0xaa, 0x29, 0x57, 0xc6, 0x97, 0x32, 0xc2, 0xaf, 0xb0, 0xb0, 0x4a, 0x95, 0x50, 0x2b,
0x57, 0x42, 0x26, 0x0b, 0xab, 0xe7, 0xc9, 0xc2, 0x17, 0xf3, 0x2c, 0x7c, 0xdd, 0x3d, 0x9d, 0x67,
0x6a, 0x35, 0x97, 0xa9, 0x9f, 0xc1, 0xb5, 0xdd, 0x10, 0xdb, 0x11, 0xfe, 0x71, 0xfc, 0x19, 0xd9,
0x9d, 0xd8, 0x84, 0x60, 0x3f, 0x71, 0xa1, 0xa8, 0x5c, 0x53, 0x28, 0xef, 0xc1, 0xda, 0x34, 0xa4,
0x4f, 0x9e, 0xa6, 0x76, 0x27, 0x4b, 0xe3, 0x4f, 0x1a, 0xf4, 0x55, 0xb2, 0x2f, 0x72, 0xbd, 0x6f,
0x43, 0x37, 0x14, 0xc6, 0x59, 0x8e, 0x90, 0xc7, 0xb5, 0x36, 0xcc, 0x8e, 0x24, 0x4b, 0x2d, 0xe8,
0x16, 0x74, 0x42, 0xcc, 0x66, 0xfe, 0x9c, 0xaf, 0xca, 0xf9, 0xda, 0x82, 0x2a, 0xd9, 0x8c, 0xbf,
0x68, 0x70, 0x6d, 0x0f, 0x47, 0x69, 0xf6, 0x62, 0x75, 0xf8, 0xeb, 0x99, 0x42, 0x23, 0x80, 0x6e,
0xc1, 0x4e, 0x74, 0x03, 0x9a, 0x19, 0x16, 0x99, 0x9f, 0x2c, 0x09, 0x7d, 0x0f, 0x56, 0xe3, 0xd0,
0x61, 0x6e, 0x51, 0x67, 0xdb, 0xd8, 0x2a, 0xf7, 0x0e, 0x5b, 0x79, 0xa9, 0xa6, 0x38, 0x60, 0xfc,
0x55, 0x83, 0xbe, 0x2a, 0x34, 0x17, 0x49, 0xdf, 0x43, 0xd8, 0x48, 0x8d, 0xb3, 0x5c, 0xcc, 0x9c,
0xd0, 0x9b, 0xf2, 0x4b, 0xc3, 0x71, 0xba, 0xb9, 0x7d, 0xf3, 0x6c, 0xf3, 0x98, 0x79, 0x35, 0x15,
0x31, 0xc8, 0x48, 0x30, 0x3c, 0xb8, 0xba, 0x87, 0xa3, 0x43, 0x3c, 0x0e, 0x30, 0x89, 0x86, 0xe4,
0x88, 0x9e, 0x3f, 0x8b, 0xaf, 0x03, 0x30, 0x29, 0x27, 0xfd, 0x84, 0x64, 0x28, 0xc6, 0xdf, 0x2b,
0xd0, 0xcc, 0x28, 0x42, 0xaf, 0x41, 0x23, 0xdd, 0x95, 0x49, 0x98, 0x13, 0x4a, 0xf9, 0xaf, 0x28,
0xf2, 0x5f, 0x48, 0x64, 0xb5, 0x9c, 0xc8, 0x05, 0x50, 0x8b, 0xae, 0xc1, 0x7a, 0x80, 0x03, 0x8b,
0x79, 0xcf, 0xb0, 0xbc, 0xda, 0x6b, 0x01, 0x0e, 0x0e, 0xbd, 0x67, 0x38, 0xde, 0x22, 0xb3, 0xc0,
0x0a, 0xe9, 0x09, 0xeb, 0xd5, 0xc5, 0x16, 0x99, 0x05, 0x26, 0x3d, 0x61, 0xe8, 0x3a, 0x80, 0x47,
0x5c, 0xfc, 0xc4, 0x22, 0x76, 0x80, 0x7b, 0x6b, 0xfc, 0x6a, 0x34, 0x38, 0x65, 0xdf, 0x0e, 0x70,
0x7c, 0xa9, 0xf9, 0x62, 0x38, 0xe8, 0xad, 0x8b, 0x83, 0x72, 0x19, 0xbb, 0x2a, 0x2f, 0xd4, 0x70,
0xd0, 0x6b, 0x88, 0x73, 0x29, 0x01, 0x7d, 0x04, 0x6d, 0xe9, 0xb7, 0x25, 0xaa, 0x0e, 0x78, 0xd5,
0xdd, 0x50, 0xa5, 0x55, 0x06, 0x50, 0xd4, 0x5c, 0x8b, 0x65, 0x56, 0xc6, 0x6f, 0x34, 0xd8, 0x28,
0xe6, 0xf2, 0x22, 0x65, 0xf7, 0x1d, 0x58, 0xf5, 0xc8, 0x11, 0x4d, 0xaa, 0xec, 0x8d, 0x53, 0xcc,
0xe1, 0xca, 0x04, 0xb7, 0xf1, 0x0f, 0x0d, 0x36, 0xee, 0xb9, 0xae, 0x0a, 0x19, 0x9f, 0xbf, 0xa6,
0xe6, 0xf9, 0xab, 0xe4, 0xf2, 0xb7, 0x0c, 0x3a, 0xbc, 0x03, 0x97, 0x0a, 0xa8, 0x27, 0xcb, 0xa0,
0x61, 0xea, 0x79, 0xdc, 0x1b, 0x0e, 0xd0, 0xdb, 0xa0, 0xe7, 0x91, 0x4f, 0x62, 0x7e, 0xc3, 0xec,
0xe6, 0xb0, 0x6f, 0x38, 0x30, 0xfe, 0xa9, 0xc1, 0x35, 0x13, 0x07, 0xf4, 0x31, 0xfe, 0xdf, 0xf5,
0xf1, 0x5f, 0x15, 0xd8, 0xf8, 0xa9, 0x1d, 0x39, 0x93, 0x41, 0x20, 0x89, 0xec, 0xd5, 0x38, 0x58,
0xb8, 0xe2, 0xb5, 0xf2, 0x15, 0x4f, 0xcb, 0x74, 0x55, 0x55, 0xa6, 0xf1, 0x33, 0x6d, 0xeb, 0xb3,
0xc4, 0xdf, 0x79, 0x99, 0x66, 0x9a, 0x98, 0xfa, 0x39, 0x9a, 0x18, 0xb4, 0x0b, 0x6d, 0xfc, 0xc4,
0xf1, 0x67, 0x2e, 0xb6, 0x84, 0xf6, 0x35, 0xae, 0xfd, 0x75, 0x85, 0xf6, 0xec, 0x1d, 0x69, 0xc9,
0x43, 0x43, 0x7e, 0x55, 0xfe, 0xad, 0x41, 0x57, 0xee, 0xc6, 0x7d, 0xdf, 0x12, 0xa8, 0x58, 0x08,
0x47, 0xa5, 0x1c, 0x8e, 0x65, 0x82, 0x9a, 0x7c, 0x6f, 0x6b, 0x99, 0xef, 0xed, 0x75, 0x80, 0x23,
0x7f, 0xc6, 0x26, 0x56, 0xe4, 0x05, 0x09, 0x26, 0x36, 0x38, 0xe5, 0x81, 0x17, 0x60, 0x74, 0x0f,
0x5a, 0x23, 0x8f, 0xf8, 0x74, 0x6c, 0x4d, 0xed, 0x68, 0x12, 0x23, 0xe3, 0x22, 0x77, 0xef, 0x7b,
0xd8, 0x77, 0x77, 0x38, 0xaf, 0xd9, 0x14, 0x67, 0x0e, 0xe2, 0x23, 0xc6, 0x9f, 0x2b, 0x70, 0x39,
0x76, 0x53, 0x7a, 0xfc, 0x12, 0x0a, 0xea, 0xc3, 0xa4, 0x14, 0xaa, 0x8b, 0xbf, 0x8b, 0x85, 0x78,
0x97, 0xcb, 0xe1, 0x3c, 0x2f, 0x0b, 0xf4, 0x23, 0xe8, 0xf8, 0xd4, 0x76, 0x2d, 0x87, 0x12, 0x97,
0x67, 0x82, 0x47, 0xb0, 0xb3, 0xfd, 0x96, 0xca, 0x84, 0x07, 0xa1, 0x37, 0x1e, 0xe3, 0x70, 0x37,
0xe1, 0x35, 0xdb, 0x3e, 0x7f, 0x57, 0xc9, 0x25, 0x47, 0x50, 0xd9, 0x20, 0xbf, 0xbc, 0x58, 0x25,
0x35, 0x50, 0x3d, 0xa5, 0xe7, 0xaa, 0x2d, 0xd1, 0x73, 0xad, 0x2a, 0xda, 0xe6, 0x7c, 0x27, 0x50,
0x2f, 0x75, 0x02, 0x0f, 0xa0, 0x9d, 0xe2, 0x0a, 0x2f, 0xfa, 0x9b, 0xd0, 0x16, 0x66, 0x59, 0x71,
0x24, 0xb0, 0x9b, 0xf4, 0xcc, 0x82, 0xf8, 0x31, 0xa7, 0xc5, 0x52, 0x53, 0xdc, 0x12, 0x1f, 0xa5,
0x86, 0x99, 0xa1, 0x18, 0xbf, 0xd7, 0x40, 0xcf, 0x22, 0x32, 0x97, 0xbc, 0x4c, 0x33, 0x7e, 0x1b,
0xba, 0x72, 0x5c, 0x94, 0xc2, 0xa2, 0x6c, 0x8f, 0x8f, 0xb3, 0xe2, 0x06, 0xe8, 0x03, 0xd8, 0x10,
0x8c, 0x25, 0x18, 0x15, 0x6d, 0xf2, 0x15, 0xbe, 0x6b, 0x16, 0xb0, 0xf4, 0xb7, 0x15, 0xe8, 0xcc,
0x0b, 0x67, 0x69, 0xab, 0x96, 0x78, 0x93, 0xa3, 0xfb, 0xd0, 0x96, 0x36, 0x58, 0xd9, 0xc2, 0x7f,
0x53, 0x55, 0x75, 0xb9, 0x88, 0x9b, 0xad, 0x0c, 0x24, 0xf2, 0x07, 0x82, 0x2c, 0xdf, 0xc4, 0x00,
0x9e, 0xfb, 0x75, 0xb3, 0xe3, 0xe7, 0x5e, 0xfc, 0x17, 0x7d, 0xfa, 0xfd, 0x12, 0xba, 0x3f, 0xb0,
0x89, 0x4b, 0x8f, 0x8e, 0x92, 0xc2, 0x3e, 0x47, 0x45, 0x7f, 0x98, 0xef, 0x4b, 0x9e, 0xe3, 0x96,
0x1b, 0x7f, 0xa8, 0xc0, 0x46, 0x4c, 0xdb, 0xb1, 0x7d, 0x9b, 0x38, 0x78, 0xf9, 0x6e, 0xf4, 0xc5,
0xe0, 0xee, 0x4d, 0x68, 0x33, 0x3a, 0x0b, 0x1d, 0x6c, 0xe5, 0x9a, 0xd2, 0x96, 0x20, 0xee, 0x8b,
0x8b, 0x79, 0x1d, 0xc0, 0x65, 0x91, 0x95, 0x7b, 0x77, 0x36, 0x5c, 0x16, 0xc9, 0xed, 0x37, 0xa0,
0x29, 0x65, 0xb8, 0x94, 0x60, 0xfe, 0xf1, 0x5a, 0x37, 0x41, 0x90, 0x06, 0x94, 0xf0, 0xfe, 0x35,
0x3e, 0xcf, 0x77, 0xd7, 0xf8, 0xee, 0x9a, 0xcb, 0x22, 0xbe, 0x75, 0x1d, 0xe0, 0xb1, 0xed, 0x7b,
0x2e, 0x2f, 0x16, 0xde, 0xa3, 0xae, 0x9b, 0x0d, 0x4e, 0x89, 0x43, 0x60, 0xfc, 0x4d, 0x03, 0x94,
0x89, 0xce, 0xf9, 0x31, 0xe7, 0x16, 0x74, 0x72, 0x7e, 0xa6, 0x83, 0xb4, 0xac, 0xa3, 0x2c, 0x06,
0xcd, 0x91, 0x50, 0x65, 0x85, 0xd8, 0x66, 0x94, 0xf0, 0xa0, 0x2d, 0x0d, 0x9a, 0xa3, 0xc4, 0xcc,
0xf8, 0xe8, 0xe6, 0x33, 0xe8, 0xe4, 0x9f, 0x3c, 0xa8, 0x05, 0xeb, 0xfb, 0x34, 0xfa, 0xe8, 0x89,
0xc7, 0x22, 0x7d, 0x05, 0x75, 0x00, 0xf6, 0x69, 0x74, 0x10, 0x62, 0x86, 0x49, 0xa4, 0x6b, 0x08,
0xa0, 0xfe, 0x29, 0x19, 0x78, 0xec, 0x73, 0xbd, 0x82, 0x2e, 0xcb, 0x37, 0xa2, 0xed, 0x0f, 0xc9,
0x27, 0x38, 0xa0, 0xe1, 0x53, 0xbd, 0x1a, 0x1f, 0x4f, 0x57, 0x35, 0xa4, 0x43, 0x2b, 0x65, 0xd9,
0x3b, 0xf8, 0x89, 0xbe, 0x8a, 0x1a, 0xb0, 0x2a, 0x7e, 0xd6, 0x37, 0x3f, 0x05, 0xbd, 0x68, 0x1e,
0x6a, 0xc2, 0xda, 0x44, 0x94, 0xba, 0xbe, 0x82, 0xba, 0xd0, 0xf4, 0xe7, 0x81, 0xd5, 0xb5, 0x98,
0x30, 0x0e, 0xa7, 0x8e, 0x0c, 0xb1, 0x5e, 0x89, 0xb5, 0xc5, 0xb1, 0x1a, 0xd0, 0x13, 0xa2, 0x57,
0x37, 0x7f, 0x08, 0xad, 0x6c, 0xa3, 0x8f, 0xd6, 0xa1, 0xb6, 0x4f, 0x09, 0xd6, 0x57, 0x62, 0xb1,
0x7b, 0x21, 0x3d, 0xf1, 0xc8, 0x58, 0xf8, 0x70, 0x3f, 0xa4, 0xcf, 0x30, 0xd1, 0x2b, 0xf1, 0x06,
0xc3, 0xb6, 0x1f, 0x6f, 0x54, 0xe3, 0x8d, 0x78, 0x81, 0x5d, 0xbd, 0xb6, 0xfd, 0x1f, 0x00, 0x10,
0xb0, 0x48, 0x69, 0xe8, 0xa2, 0x29, 0xa0, 0x3d, 0x1c, 0xed, 0xd2, 0x60, 0x4a, 0x49, 0x22, 0x9f,
0xa1, 0xf7, 0x16, 0x0c, 0x76, 0xcb, 0xac, 0xd2, 0xe4, 0xfe, 0xb7, 0x16, 0x9c, 0x28, 0xb0, 0x1b,
0x2b, 0x28, 0xe0, 0x1a, 0xe3, 0x2e, 0xe2, 0x81, 0xe7, 0x7c, 0x9e, 0x8c, 0x1a, 0x4e, 0xd1, 0x58,
0x60, 0x4d, 0x34, 0x16, 0x2e, 0xba, 0x5c, 0x1c, 0x46, 0xa1, 0x47, 0xc6, 0xc9, 0x4b, 0xc7, 0x58,
0x41, 0xc7, 0x70, 0x25, 0x7e, 0x05, 0x45, 0x76, 0xe4, 0xb1, 0xc8, 0x73, 0x58, 0xa2, 0x70, 0x7b,
0xb1, 0xc2, 0x12, 0xf3, 0x73, 0xaa, 0x74, 0xa0, 0x95, 0x1d, 0xbc, 0xa3, 0xdb, 0xaa, 0x02, 0x56,
0xfc, 0x39, 0xd0, 0xbf, 0x73, 0x36, 0x63, 0xaa, 0xc4, 0x87, 0x6e, 0x61, 0xd8, 0x8d, 0x36, 0x95,
0xd0, 0xa7, 0x9c, 0xb8, 0xf7, 0xdf, 0x59, 0x8a, 0x37, 0xd5, 0xe6, 0x41, 0x27, 0x3f, 0x60, 0x46,
0x6f, 0x2f, 0x12, 0x50, 0x1a, 0xe2, 0xf5, 0x37, 0x97, 0x61, 0x4d, 0x55, 0x3d, 0x84, 0x4e, 0x7e,
0xea, 0xa9, 0x56, 0xa5, 0x9c, 0x8c, 0xf6, 0x4f, 0x7b, 0xc9, 0x1a, 0x2b, 0xe8, 0x17, 0x70, 0xa9,
0x34, 0x6a, 0x44, 0xdf, 0x56, 0x47, 0x5d, 0x3d, 0x91, 0x3c, 0x4b, 0x83, 0xb4, 0x3e, 0xf3, 0x8d,
0x5c, 0x68, 0x7d, 0x69, 0xe6, 0xbc, 0xbc, 0xf5, 0x19, 0xf1, 0xa7, 0x59, 0xff, 0xdc, 0x1a, 0x66,
0x80, 0xca, 0xc3, 0x46, 0xf4, 0xae, 0x4a, 0xc5, 0xc2, 0x81, 0x67, 0x7f, 0x6b, 0x59, 0xf6, 0x34,
0xe5, 0x33, 0x0e, 0x09, 0xc5, 0xb9, 0x9c, 0x52, 0xed, 0xc2, 0x39, 0xa3, 0x5a, 0xed, 0xe2, 0xd9,
0x9b, 0x28, 0xea, 0xfc, 0x80, 0x44, 0x9d, 0x2b, 0xe5, 0x40, 0x4c, 0x5d, 0xd4, 0xea, 0x79, 0x8b,
0xb1, 0xb2, 0xfd, 0xe5, 0x3a, 0x34, 0xb8, 0xf3, 0x1c, 0x10, 0xfe, 0x0f, 0xba, 0x2f, 0x1e, 0x74,
0x1f, 0x41, 0xb7, 0x30, 0x66, 0x52, 0xe3, 0xa1, 0x7a, 0x16, 0x75, 0xd6, 0xc5, 0x18, 0x01, 0x2a,
0xcf, 0x78, 0xd4, 0x15, 0xba, 0x70, 0x16, 0x74, 0x96, 0x8e, 0x47, 0xd0, 0x2d, 0xcc, 0x58, 0xd4,
0x1e, 0xa8, 0x07, 0x31, 0x67, 0x49, 0xff, 0x0c, 0x5a, 0xd9, 0xd7, 0xb6, 0xfa, 0xa3, 0xa4, 0x78,
0x8f, 0xbf, 0x7a, 0x50, 0x7a, 0xf9, 0xa0, 0xfd, 0x08, 0xba, 0x85, 0x07, 0xb6, 0x3a, 0xf2, 0xea,
0x57, 0xf8, 0x59, 0xd2, 0xbf, 0x3a, 0x98, 0xd9, 0xf9, 0xe0, 0xe1, 0xf6, 0xd8, 0x8b, 0x26, 0xb3,
0x51, 0x6c, 0xc4, 0x5d, 0x71, 0xf2, 0x5d, 0x8f, 0xca, 0x5f, 0x77, 0x93, 0xfb, 0x76, 0x97, 0x0b,
0xbb, 0xcb, 0x85, 0x4d, 0x47, 0xa3, 0x3a, 0x5f, 0xbe, 0xff, 0xdf, 0x00, 0x00, 0x00, 0xff, 0xff,
0x9e, 0x3d, 0x5b, 0x45, 0x26, 0x21, 0x00, 0x00,
// 1831 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xec, 0x58, 0xcd, 0x6f, 0x24, 0x47,
0x15, 0x77, 0xcf, 0x97, 0x67, 0xde, 0x7c, 0xf5, 0x56, 0xd6, 0xc3, 0xec, 0x90, 0x4d, 0x4c, 0x6f,
0x96, 0xdd, 0x38, 0xc4, 0x1b, 0xcd, 0x06, 0x89, 0x1c, 0x38, 0x64, 0x3d, 0x59, 0x33, 0x90, 0x38,
0xa6, 0xbd, 0x04, 0xb1, 0x5a, 0xd1, 0xf4, 0x4c, 0x97, 0x67, 0x5a, 0xe9, 0xee, 0x1a, 0x77, 0xf5,
0xac, 0xbd, 0x3e, 0x20, 0x21, 0x71, 0xe6, 0xc6, 0x09, 0x2e, 0x5c, 0x90, 0x38, 0xf0, 0x0f, 0x20,
0x21, 0xe5, 0x1f, 0x01, 0x09, 0x91, 0x1b, 0xff, 0x04, 0xaa, 0x8f, 0xee, 0xe9, 0xaf, 0xb1, 0xc7,
0x76, 0x1c, 0x47, 0x88, 0x5b, 0xd7, 0xab, 0x57, 0xef, 0xbb, 0x7e, 0xfd, 0xea, 0xc1, 0xad, 0xa3,
0x39, 0xf6, 0x5f, 0x19, 0x63, 0x42, 0x7c, 0x6b, 0x7b, 0xe6, 0x93, 0x80, 0x20, 0xe4, 0xda, 0xce,
0xcb, 0x39, 0x15, 0xab, 0x6d, 0xbe, 0xdf, 0x6b, 0x8c, 0x89, 0xeb, 0x12, 0x4f, 0xd0, 0x7a, 0x8d,
0x38, 0x47, 0xaf, 0x65, 0x7b, 0x01, 0xf6, 0x3d, 0xd3, 0x09, 0x77, 0xe9, 0x78, 0x8a, 0x5d, 0x53,
0xae, 0x54, 0xcb, 0x0c, 0xcc, 0xb8, 0x7c, 0xed, 0x97, 0xd0, 0x39, 0x98, 0x92, 0xe3, 0x1d, 0xe2,
0x38, 0x78, 0x1c, 0xd8, 0xc4, 0xa3, 0x3a, 0x3e, 0x9a, 0x63, 0x1a, 0xa0, 0xf7, 0xa0, 0x34, 0x32,
0x29, 0xee, 0x2a, 0x9b, 0xca, 0xc3, 0x7a, 0xff, 0xf5, 0xed, 0x84, 0x21, 0xd2, 0x82, 0x4f, 0xe8,
0xe4, 0x89, 0x49, 0xb1, 0xce, 0x39, 0x11, 0x82, 0x92, 0x35, 0x1a, 0x0e, 0xba, 0x85, 0x4d, 0xe5,
0x61, 0x51, 0xe7, 0xdf, 0x5a, 0x00, 0xdf, 0xca, 0xc8, 0xa7, 0x33, 0xe2, 0x51, 0x8c, 0x1e, 0x43,
0x85, 0x06, 0x66, 0x30, 0xa7, 0x52, 0xc5, 0xb7, 0x73, 0x55, 0x1c, 0x70, 0x16, 0x5d, 0xb2, 0xa2,
0xb7, 0xa0, 0x39, 0x8e, 0x64, 0x0d, 0x07, 0xb4, 0x5b, 0xd8, 0x2c, 0x3e, 0x2c, 0xea, 0x49, 0xa2,
0xf6, 0x1b, 0x05, 0x36, 0x98, 0xda, 0x7d, 0xd3, 0x0f, 0xec, 0xaf, 0xde, 0x2b, 0xa4, 0x41, 0x23,
0xae, 0xb0, 0x5b, 0xe4, 0x7b, 0x09, 0x9a, 0x76, 0x24, 0x22, 0x1b, 0x37, 0xe1, 0x2a, 0x8e, 0x6b,
0xd0, 0x98, 0x85, 0xa2, 0x16, 0x7e, 0x27, 0x68, 0xda, 0x17, 0x0a, 0x6c, 0x7c, 0x4c, 0x4c, 0x6b,
0x11, 0xed, 0xaf, 0xdd, 0x6d, 0xf4, 0x43, 0xa8, 0x88, 0x92, 0xeb, 0x96, 0xb8, 0xae, 0xfb, 0x49,
0x5d, 0xb2, 0x1c, 0x17, 0x16, 0x1e, 0x70, 0x82, 0x2e, 0x0f, 0x69, 0x7f, 0x54, 0xa0, 0xab, 0x63,
0x07, 0x9b, 0x14, 0xdf, 0xa4, 0x17, 0x1d, 0xa8, 0x78, 0xc4, 0xc2, 0xc3, 0x01, 0xf7, 0xa2, 0xa8,
0xcb, 0x95, 0xf6, 0xa5, 0x8c, 0xf0, 0x0d, 0x16, 0x56, 0xa6, 0x12, 0x4a, 0xd9, 0x4a, 0x88, 0x65,
0xa1, 0x7c, 0x99, 0x2c, 0x7c, 0xb1, 0xc8, 0xc2, 0x37, 0xdd, 0xd3, 0x45, 0xa6, 0xca, 0x89, 0x4c,
0xfd, 0x02, 0xee, 0xec, 0xf8, 0xd8, 0x0c, 0xf0, 0x4f, 0x19, 0x66, 0xee, 0x4c, 0x4d, 0xcf, 0xc3,
0x4e, 0xe8, 0x42, 0x5a, 0xb9, 0x92, 0xa3, 0xbc, 0x0b, 0xeb, 0x33, 0x9f, 0x9c, 0xbc, 0x8a, 0xec,
0x0e, 0x97, 0xda, 0x9f, 0x14, 0xe8, 0xe5, 0xc9, 0xbe, 0xca, 0xf5, 0x7e, 0x00, 0x6d, 0x5f, 0x18,
0x67, 0x8c, 0x85, 0x3c, 0xae, 0xb5, 0xa6, 0xb7, 0x24, 0x59, 0x6a, 0x41, 0xf7, 0xa1, 0xe5, 0x63,
0x3a, 0x77, 0x16, 0x7c, 0x45, 0xce, 0xd7, 0x14, 0x54, 0xc9, 0xa6, 0xfd, 0x45, 0x81, 0x3b, 0xbb,
0x38, 0x88, 0xb2, 0xc7, 0xd4, 0xe1, 0x6f, 0x66, 0x0a, 0x35, 0x17, 0xda, 0x29, 0x3b, 0xd1, 0x26,
0xd4, 0x63, 0x2c, 0x32, 0x3f, 0x71, 0x12, 0xfa, 0x01, 0x94, 0x59, 0xe8, 0x30, 0xb7, 0xa8, 0xd5,
0xd7, 0xb6, 0xb3, 0x3f, 0xca, 0xed, 0xa4, 0x54, 0x5d, 0x1c, 0xd0, 0xfe, 0xaa, 0x40, 0x2f, 0x2f,
0x34, 0x57, 0x49, 0xdf, 0x73, 0xe8, 0x44, 0xc6, 0x19, 0x16, 0xa6, 0x63, 0xdf, 0x9e, 0xf1, 0x4b,
0xc3, 0x71, 0xba, 0xde, 0xbf, 0x77, 0xbe, 0x79, 0x54, 0xdf, 0x88, 0x44, 0x0c, 0x62, 0x12, 0x34,
0x1b, 0x36, 0x76, 0x71, 0x70, 0x80, 0x27, 0x2e, 0xf6, 0x82, 0xa1, 0x77, 0x48, 0x2e, 0x9f, 0xc5,
0x37, 0x00, 0xa8, 0x94, 0x13, 0xfd, 0x42, 0x62, 0x14, 0xed, 0x1f, 0x05, 0xa8, 0xc7, 0x14, 0xa1,
0xd7, 0xa1, 0x16, 0xed, 0xca, 0x24, 0x2c, 0x08, 0x99, 0xfc, 0x17, 0x72, 0xf2, 0x9f, 0x4a, 0x64,
0x31, 0x9b, 0xc8, 0x25, 0x50, 0x8b, 0xee, 0x40, 0xd5, 0xc5, 0xae, 0x41, 0xed, 0x53, 0x2c, 0xaf,
0xf6, 0xba, 0x8b, 0xdd, 0x03, 0xfb, 0x14, 0xb3, 0x2d, 0x6f, 0xee, 0x1a, 0x3e, 0x39, 0xa6, 0xdd,
0x8a, 0xd8, 0xf2, 0xe6, 0xae, 0x4e, 0x8e, 0x29, 0xba, 0x0b, 0x60, 0x7b, 0x16, 0x3e, 0x31, 0x3c,
0xd3, 0xc5, 0xdd, 0x75, 0x7e, 0x35, 0x6a, 0x9c, 0xb2, 0x67, 0xba, 0x98, 0x5d, 0x6a, 0xbe, 0x18,
0x0e, 0xba, 0x55, 0x71, 0x50, 0x2e, 0x99, 0xab, 0xf2, 0x42, 0x0d, 0x07, 0xdd, 0x9a, 0x38, 0x17,
0x11, 0xd0, 0x47, 0xd0, 0x94, 0x7e, 0x1b, 0xa2, 0xea, 0x80, 0x57, 0xdd, 0x66, 0x5e, 0x5a, 0x65,
0x00, 0x45, 0xcd, 0x35, 0x68, 0x6c, 0xa5, 0xfd, 0x56, 0x81, 0x4e, 0x3a, 0x97, 0x57, 0x29, 0xbb,
0xef, 0x43, 0xd9, 0xf6, 0x0e, 0x49, 0x58, 0x65, 0x6f, 0x9e, 0x61, 0x0e, 0x57, 0x26, 0xb8, 0xb5,
0x7f, 0x2a, 0xd0, 0xf9, 0xd0, 0xb2, 0xf2, 0x90, 0xf1, 0xe2, 0x35, 0xb5, 0xc8, 0x5f, 0x21, 0x91,
0xbf, 0x55, 0xd0, 0xe1, 0x1d, 0xb8, 0x95, 0x42, 0x3d, 0x59, 0x06, 0x35, 0x5d, 0x4d, 0xe2, 0xde,
0x70, 0x80, 0xde, 0x06, 0x35, 0x89, 0x7c, 0x12, 0xf3, 0x6b, 0x7a, 0x3b, 0x81, 0x7d, 0xc3, 0x81,
0xf6, 0x2f, 0x05, 0xee, 0xe8, 0xd8, 0x25, 0x2f, 0xf1, 0xff, 0xae, 0x8f, 0xff, 0x2e, 0x40, 0xe7,
0xe7, 0x66, 0x30, 0x9e, 0x0e, 0x5c, 0x49, 0xa4, 0x37, 0xe3, 0x60, 0xea, 0x8a, 0x97, 0xb2, 0x57,
0x3c, 0x2a, 0xd3, 0x72, 0x5e, 0x99, 0xb2, 0x37, 0xc9, 0xf6, 0x67, 0xa1, 0xbf, 0x8b, 0x32, 0x8d,
0x35, 0x31, 0x95, 0x4b, 0x34, 0x31, 0x68, 0x07, 0x9a, 0xf8, 0x64, 0xec, 0xcc, 0x2d, 0x6c, 0x08,
0xed, 0xeb, 0x5c, 0xfb, 0x1b, 0x39, 0xda, 0xe3, 0x77, 0xa4, 0x21, 0x0f, 0x0d, 0xf9, 0x55, 0xf9,
0x8f, 0x02, 0x6d, 0xb9, 0xcb, 0xfa, 0xbe, 0x15, 0x50, 0x31, 0x15, 0x8e, 0x42, 0x36, 0x1c, 0xab,
0x04, 0x35, 0xfc, 0xdf, 0x96, 0x62, 0xff, 0xdb, 0xbb, 0x00, 0x87, 0xce, 0x9c, 0x4e, 0x8d, 0xc0,
0x76, 0x43, 0x4c, 0xac, 0x71, 0xca, 0x33, 0xdb, 0xc5, 0xe8, 0x43, 0x68, 0x8c, 0x6c, 0xcf, 0x21,
0x13, 0x63, 0x66, 0x06, 0x53, 0x86, 0x8c, 0xcb, 0xdc, 0x7d, 0x6a, 0x63, 0xc7, 0x7a, 0xc2, 0x79,
0xf5, 0xba, 0x38, 0xb3, 0xcf, 0x8e, 0x68, 0x7f, 0x2e, 0xc0, 0x6b, 0xcc, 0x4d, 0xe9, 0xf1, 0x35,
0x14, 0xd4, 0x07, 0x61, 0x29, 0x14, 0x97, 0xff, 0x17, 0x53, 0xf1, 0xce, 0x96, 0xc3, 0x65, 0x5e,
0x16, 0xe8, 0x27, 0xd0, 0x72, 0x88, 0x69, 0x19, 0x63, 0xe2, 0x59, 0x3c, 0x13, 0x3c, 0x82, 0xad,
0xfe, 0x5b, 0x79, 0x26, 0x3c, 0xf3, 0xed, 0xc9, 0x04, 0xfb, 0x3b, 0x21, 0xaf, 0xde, 0x74, 0xf8,
0xbb, 0x4a, 0x2e, 0x39, 0x82, 0xca, 0x06, 0xf9, 0xfa, 0x62, 0x15, 0xd6, 0x40, 0xf1, 0x8c, 0x9e,
0xab, 0xb4, 0x42, 0xcf, 0x55, 0xce, 0x69, 0x9b, 0x93, 0x9d, 0x40, 0x25, 0xd3, 0x09, 0x3c, 0x83,
0x66, 0x84, 0x2b, 0xbc, 0xe8, 0xef, 0x41, 0x53, 0x98, 0x65, 0xb0, 0x48, 0x60, 0x2b, 0xec, 0x99,
0x05, 0xf1, 0x63, 0x4e, 0x63, 0x52, 0x23, 0xdc, 0x12, 0x3f, 0xa5, 0x9a, 0x1e, 0xa3, 0x68, 0xbf,
0x57, 0x40, 0x8d, 0x23, 0x32, 0x97, 0xbc, 0x4a, 0x33, 0xfe, 0x00, 0xda, 0x72, 0x36, 0x12, 0xc1,
0xa2, 0x6c, 0x8f, 0x8f, 0xe2, 0xe2, 0x06, 0xe8, 0x7d, 0xe8, 0x08, 0xc6, 0x0c, 0x8c, 0x8a, 0x36,
0xf9, 0x36, 0xdf, 0xd5, 0x53, 0x58, 0xfa, 0xf7, 0x02, 0xb4, 0x16, 0x85, 0xb3, 0xb2, 0x55, 0x2b,
0xbc, 0xc9, 0xd1, 0x53, 0x68, 0x4a, 0x1b, 0x8c, 0x78, 0xe1, 0x7f, 0x27, 0xaf, 0xea, 0x12, 0x11,
0xd7, 0x1b, 0x31, 0x48, 0xe4, 0x0f, 0x04, 0x59, 0xbe, 0xa1, 0x01, 0x3c, 0xf7, 0x55, 0xbd, 0xe5,
0x24, 0x5e, 0xfc, 0x57, 0x7c, 0xfa, 0xa1, 0xc7, 0xb0, 0xe1, 0x8b, 0xc2, 0xb6, 0x8c, 0x84, 0x73,
0xa2, 0x46, 0x6e, 0x87, 0x9b, 0xfb, 0xf1, 0x0e, 0xfe, 0xd7, 0xd0, 0xfe, 0x91, 0xe9, 0x59, 0xe4,
0xf0, 0x30, 0xbc, 0x0d, 0x97, 0xb8, 0x06, 0x1f, 0x24, 0x9b, 0x99, 0x0b, 0x40, 0x83, 0xf6, 0x87,
0x02, 0x74, 0x18, 0xed, 0x89, 0xe9, 0x98, 0xde, 0x18, 0xaf, 0xde, 0xc2, 0x7e, 0x35, 0x60, 0x7d,
0x0f, 0x9a, 0x94, 0xcc, 0xfd, 0x31, 0x36, 0x12, 0x9d, 0x6c, 0x43, 0x10, 0xf7, 0xc4, 0x6d, 0xbe,
0x0b, 0x60, 0xd1, 0xc0, 0x48, 0x3c, 0x56, 0x6b, 0x16, 0x0d, 0xe4, 0xf6, 0x9b, 0x50, 0x97, 0x32,
0x2c, 0xe2, 0x61, 0xfe, 0xc7, 0xab, 0xea, 0x20, 0x48, 0x03, 0xe2, 0xf1, 0xa6, 0x97, 0x9d, 0xe7,
0xbb, 0xeb, 0x7c, 0x77, 0xdd, 0xa2, 0x01, 0xdf, 0xba, 0x0b, 0xf0, 0xd2, 0x74, 0x6c, 0x8b, 0x57,
0x18, 0x6f, 0x6c, 0xab, 0x7a, 0x8d, 0x53, 0x58, 0x08, 0xb4, 0xbf, 0x29, 0x80, 0x62, 0xd1, 0xb9,
0x3c, 0x50, 0xdd, 0x87, 0x56, 0xc2, 0xcf, 0x68, 0xfa, 0x16, 0x77, 0x94, 0x32, 0xa4, 0x1d, 0x09,
0x55, 0x86, 0x8f, 0x4d, 0x4a, 0x3c, 0x1e, 0xb4, 0x95, 0x91, 0x76, 0x14, 0x9a, 0xc9, 0x8e, 0x6e,
0x9d, 0x42, 0x2b, 0xf9, 0x4e, 0x42, 0x0d, 0xa8, 0xee, 0x91, 0xe0, 0xa3, 0x13, 0x9b, 0x06, 0xea,
0x1a, 0x6a, 0x01, 0xec, 0x91, 0x60, 0xdf, 0xc7, 0x14, 0x7b, 0x81, 0xaa, 0x20, 0x80, 0xca, 0xa7,
0xde, 0xc0, 0xa6, 0x9f, 0xab, 0x05, 0xf4, 0x9a, 0x7c, 0x58, 0x9a, 0xce, 0xd0, 0xfb, 0x04, 0xbb,
0xc4, 0x7f, 0xa5, 0x16, 0xd9, 0xf1, 0x68, 0x55, 0x42, 0x2a, 0x34, 0x22, 0x96, 0xdd, 0xfd, 0x9f,
0xa9, 0x65, 0x54, 0x83, 0xb2, 0xf8, 0xac, 0x6c, 0x7d, 0x0a, 0x6a, 0xda, 0x3c, 0x54, 0x87, 0xf5,
0xa9, 0x28, 0x75, 0x75, 0x0d, 0xb5, 0xa1, 0xee, 0x2c, 0x02, 0xab, 0x2a, 0x8c, 0x30, 0xf1, 0x67,
0x63, 0x19, 0x62, 0xb5, 0xc0, 0xb4, 0xb1, 0x58, 0x0d, 0xc8, 0xb1, 0xa7, 0x16, 0xb7, 0x7e, 0x0c,
0x8d, 0xf8, 0xeb, 0x00, 0x55, 0xa1, 0xb4, 0x47, 0x3c, 0xac, 0xae, 0x31, 0xb1, 0xbb, 0x3e, 0x39,
0xb6, 0xbd, 0x89, 0xf0, 0xe1, 0xa9, 0x4f, 0x4e, 0xb1, 0xa7, 0x16, 0xd8, 0x06, 0xc5, 0xa6, 0xc3,
0x36, 0x8a, 0x6c, 0x83, 0x2d, 0xb0, 0xa5, 0x96, 0xfa, 0xbf, 0x03, 0x00, 0x81, 0xa5, 0x84, 0xf8,
0x16, 0x9a, 0x01, 0xda, 0xc5, 0xc1, 0x0e, 0x71, 0x67, 0xc4, 0x0b, 0xe5, 0x53, 0xf4, 0x5e, 0x32,
0xe4, 0xd1, 0x70, 0x38, 0xcb, 0x2a, 0x4d, 0xee, 0x7d, 0x77, 0xc9, 0x89, 0x14, 0xbb, 0xb6, 0x86,
0x5c, 0xae, 0x91, 0xb5, 0x1e, 0xcf, 0xec, 0xf1, 0xe7, 0xe1, 0x7c, 0xe2, 0x0c, 0x8d, 0x29, 0xd6,
0x50, 0x63, 0xea, 0xa2, 0xcb, 0xc5, 0x41, 0xe0, 0xdb, 0xde, 0x24, 0x7c, 0x1e, 0x69, 0x6b, 0xe8,
0x08, 0x6e, 0xb3, 0xa7, 0x53, 0x60, 0x06, 0x36, 0x0d, 0xec, 0x31, 0x0d, 0x15, 0xf6, 0x97, 0x2b,
0xcc, 0x30, 0x5f, 0x50, 0xa5, 0x03, 0xed, 0xd4, 0xf0, 0x1a, 0x6d, 0xe5, 0xa2, 0x52, 0xee, 0x04,
0xbd, 0xf7, 0xce, 0x4a, 0xbc, 0x91, 0x36, 0x1b, 0x5a, 0xc9, 0x81, 0x31, 0x7a, 0x7b, 0x99, 0x80,
0xcc, 0x50, 0xae, 0xb7, 0xb5, 0x0a, 0x6b, 0xa4, 0xea, 0x39, 0xb4, 0x92, 0x53, 0xcc, 0x7c, 0x55,
0xb9, 0x93, 0xce, 0xde, 0x59, 0x2f, 0x53, 0x6d, 0x0d, 0xfd, 0x0a, 0x6e, 0x65, 0x46, 0x87, 0xe8,
0x7b, 0x79, 0xe2, 0x97, 0x4d, 0x18, 0xcf, 0xd3, 0x20, 0xad, 0x8f, 0xfd, 0xf3, 0x96, 0x5a, 0x9f,
0x99, 0x21, 0xaf, 0x6e, 0x7d, 0x4c, 0xfc, 0x59, 0xd6, 0x5f, 0x58, 0xc3, 0x1c, 0x50, 0x76, 0x78,
0x88, 0xde, 0xcd, 0x53, 0xb1, 0x74, 0x80, 0xd9, 0xdb, 0x5e, 0x95, 0x3d, 0x4a, 0xf9, 0x9c, 0xdf,
0xd6, 0xf4, 0x9c, 0x2d, 0x57, 0xed, 0xd2, 0xb9, 0x61, 0xbe, 0xda, 0xe5, 0xb3, 0x34, 0x51, 0xd4,
0xc9, 0x81, 0x47, 0x7e, 0xae, 0x72, 0x07, 0x5c, 0xf9, 0x45, 0x9d, 0x3f, 0x3f, 0xd1, 0xd6, 0xfa,
0x5f, 0x56, 0xa1, 0xc6, 0x9d, 0x67, 0xff, 0xa1, 0xff, 0xe3, 0xe1, 0x35, 0xe0, 0xe1, 0x0b, 0x68,
0xa7, 0xc6, 0x46, 0xf9, 0x78, 0x98, 0x3f, 0x5b, 0x3a, 0xef, 0x62, 0x8c, 0x00, 0x65, 0x67, 0x36,
0xf9, 0x15, 0xba, 0x74, 0xb6, 0x73, 0x9e, 0x8e, 0x17, 0xd0, 0x4e, 0xcd, 0x4c, 0xf2, 0x3d, 0xc8,
0x1f, 0xac, 0x9c, 0x27, 0xfd, 0x33, 0x68, 0xc4, 0x5f, 0xcf, 0xe8, 0xc1, 0x32, 0x58, 0x4a, 0xbd,
0x19, 0x6f, 0x1e, 0x94, 0xae, 0x1f, 0xb4, 0x5f, 0x40, 0x3b, 0xf5, 0x60, 0xce, 0x8f, 0x7c, 0xfe,
0xab, 0xfa, 0x3c, 0xe9, 0x5f, 0x1f, 0xcc, 0x3c, 0x79, 0xff, 0x79, 0x7f, 0x62, 0x07, 0xd3, 0xf9,
0x88, 0x19, 0xf1, 0x48, 0x9c, 0x7c, 0xd7, 0x26, 0xf2, 0xeb, 0x51, 0x78, 0xdf, 0x1e, 0x71, 0x61,
0x8f, 0xb8, 0xb0, 0xd9, 0x68, 0x54, 0xe1, 0xcb, 0xc7, 0xff, 0x0d, 0x00, 0x00, 0xff, 0xff, 0x9d,
0x23, 0x87, 0xd7, 0xe3, 0x1f, 0x00, 0x00,
}
// Reference imports to suppress errors if they are not otherwise used.
@ -2164,7 +2072,6 @@ type QueryCoordClient interface {
GetComponentStates(ctx context.Context, in *internalpb.GetComponentStatesRequest, opts ...grpc.CallOption) (*internalpb.ComponentStates, error)
GetTimeTickChannel(ctx context.Context, in *internalpb.GetTimeTickChannelRequest, opts ...grpc.CallOption) (*milvuspb.StringResponse, error)
GetStatisticsChannel(ctx context.Context, in *internalpb.GetStatisticsChannelRequest, opts ...grpc.CallOption) (*milvuspb.StringResponse, error)
RegisterNode(ctx context.Context, in *RegisterNodeRequest, opts ...grpc.CallOption) (*RegisterNodeResponse, error)
ShowCollections(ctx context.Context, in *ShowCollectionsRequest, opts ...grpc.CallOption) (*ShowCollectionsResponse, error)
ShowPartitions(ctx context.Context, in *ShowPartitionsRequest, opts ...grpc.CallOption) (*ShowPartitionsResponse, error)
LoadPartitions(ctx context.Context, in *LoadPartitionsRequest, opts ...grpc.CallOption) (*commonpb.Status, error)
@ -2211,15 +2118,6 @@ func (c *queryCoordClient) GetStatisticsChannel(ctx context.Context, in *interna
return out, nil
}
func (c *queryCoordClient) RegisterNode(ctx context.Context, in *RegisterNodeRequest, opts ...grpc.CallOption) (*RegisterNodeResponse, error) {
out := new(RegisterNodeResponse)
err := c.cc.Invoke(ctx, "/milvus.proto.query.QueryCoord/RegisterNode", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *queryCoordClient) ShowCollections(ctx context.Context, in *ShowCollectionsRequest, opts ...grpc.CallOption) (*ShowCollectionsResponse, error) {
out := new(ShowCollectionsResponse)
err := c.cc.Invoke(ctx, "/milvus.proto.query.QueryCoord/ShowCollections", in, out, opts...)
@ -2306,7 +2204,6 @@ type QueryCoordServer interface {
GetComponentStates(context.Context, *internalpb.GetComponentStatesRequest) (*internalpb.ComponentStates, error)
GetTimeTickChannel(context.Context, *internalpb.GetTimeTickChannelRequest) (*milvuspb.StringResponse, error)
GetStatisticsChannel(context.Context, *internalpb.GetStatisticsChannelRequest) (*milvuspb.StringResponse, error)
RegisterNode(context.Context, *RegisterNodeRequest) (*RegisterNodeResponse, error)
ShowCollections(context.Context, *ShowCollectionsRequest) (*ShowCollectionsResponse, error)
ShowPartitions(context.Context, *ShowPartitionsRequest) (*ShowPartitionsResponse, error)
LoadPartitions(context.Context, *LoadPartitionsRequest) (*commonpb.Status, error)
@ -2331,9 +2228,6 @@ func (*UnimplementedQueryCoordServer) GetTimeTickChannel(ctx context.Context, re
func (*UnimplementedQueryCoordServer) GetStatisticsChannel(ctx context.Context, req *internalpb.GetStatisticsChannelRequest) (*milvuspb.StringResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetStatisticsChannel not implemented")
}
func (*UnimplementedQueryCoordServer) RegisterNode(ctx context.Context, req *RegisterNodeRequest) (*RegisterNodeResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method RegisterNode not implemented")
}
func (*UnimplementedQueryCoordServer) ShowCollections(ctx context.Context, req *ShowCollectionsRequest) (*ShowCollectionsResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method ShowCollections not implemented")
}
@ -2420,24 +2314,6 @@ func _QueryCoord_GetStatisticsChannel_Handler(srv interface{}, ctx context.Conte
return interceptor(ctx, in, info, handler)
}
func _QueryCoord_RegisterNode_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(RegisterNodeRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(QueryCoordServer).RegisterNode(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/milvus.proto.query.QueryCoord/RegisterNode",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(QueryCoordServer).RegisterNode(ctx, req.(*RegisterNodeRequest))
}
return interceptor(ctx, in, info, handler)
}
func _QueryCoord_ShowCollections_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(ShowCollectionsRequest)
if err := dec(in); err != nil {
@ -2616,10 +2492,6 @@ var _QueryCoord_serviceDesc = grpc.ServiceDesc{
MethodName: "GetStatisticsChannel",
Handler: _QueryCoord_GetStatisticsChannel_Handler,
},
{
MethodName: "RegisterNode",
Handler: _QueryCoord_RegisterNode_Handler,
},
{
MethodName: "ShowCollections",
Handler: _QueryCoord_ShowCollections_Handler,

194
internal/querycoord/impl.go
View File

@ -23,7 +23,6 @@ import (
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/proto/milvuspb"
"github.com/milvus-io/milvus/internal/proto/querypb"
"github.com/milvus-io/milvus/internal/util/sessionutil"
)
func (qc *QueryCoord) GetComponentStates(ctx context.Context) (*internalpb.ComponentStates, error) {
@ -70,48 +69,6 @@ func (qc *QueryCoord) GetStatisticsChannel(ctx context.Context) (*milvuspb.Strin
}, nil
}
func (qc *QueryCoord) RegisterNode(ctx context.Context, req *querypb.RegisterNodeRequest) (*querypb.RegisterNodeResponse, error) {
nodeID := req.Base.SourceID
log.Debug("register query node", zap.Any("QueryNodeID", nodeID), zap.String("address", req.Address.String()))
if _, ok := qc.cluster.nodes[nodeID]; ok {
err := errors.New("nodeID already exists")
log.Debug("register query node Failed nodeID already exist", zap.Any("QueryNodeID", nodeID), zap.String("address", req.Address.String()))
return &querypb.RegisterNodeResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
Reason: err.Error(),
},
}, err
}
session := &sessionutil.Session{
ServerID: nodeID,
Address: fmt.Sprintf("%s:%d", req.Address.Ip, req.Address.Port),
}
err := qc.cluster.RegisterNode(ctx, session, req.Base.SourceID)
if err != nil {
log.Debug("register query node new NodeClient failed", zap.Any("QueryNodeID", nodeID), zap.String("address", req.Address.String()))
return &querypb.RegisterNodeResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UnexpectedError,
},
}, err
}
log.Debug("register query node success", zap.Any("QueryNodeID", nodeID), zap.String("address", req.Address.String()))
return &querypb.RegisterNodeResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
},
InitParams: &internalpb.InitParams{
NodeID: nodeID,
//StartParams: startParams,
},
}, nil
}
func (qc *QueryCoord) ShowCollections(ctx context.Context, req *querypb.ShowCollectionsRequest) (*querypb.ShowCollectionsResponse, error) {
dbID := req.DbID
log.Debug("show collection start", zap.Int64("dbID", dbID))
@ -131,20 +88,17 @@ func (qc *QueryCoord) LoadCollection(ctx context.Context, req *querypb.LoadColle
log.Debug("LoadCollectionRequest received", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", collectionID),
zap.Stringer("schema", req.Schema))
status := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UnexpectedError,
ErrorCode: commonpb.ErrorCode_Success,
}
hasCollection := qc.meta.hasCollection(collectionID)
if hasCollection {
status.ErrorCode = commonpb.ErrorCode_Success
status.Reason = "collection has been loaded"
return status, nil
loadCollection, _ := qc.meta.getLoadCollection(collectionID)
if loadCollection {
status.Reason = "collection has been loaded"
return status, nil
}
}
//err := qs.meta.addCollection(collectionID, schema)
//if err != nil {
// log.Error(err.Error())
// return status, err
//}
loadCollectionTask := &LoadCollectionTask{
BaseTask: BaseTask{
@ -162,13 +116,12 @@ func (qc *QueryCoord) LoadCollection(ctx context.Context, req *querypb.LoadColle
err := loadCollectionTask.WaitToFinish()
if err != nil {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
return status, err
}
//qs.meta.setLoadCollection(collectionID, true)
log.Debug("LoadCollectionRequest completed", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", collectionID))
status.ErrorCode = commonpb.ErrorCode_Success
return status, nil
}
@ -181,7 +134,7 @@ func (qc *QueryCoord) ReleaseCollection(ctx context.Context, req *querypb.Releas
}
hasCollection := qc.meta.hasCollection(collectionID)
if !hasCollection {
log.Error("release collection end, query coordinator don't have the log of", zap.String("collectionID", fmt.Sprintln(collectionID)))
log.Warn("release collection end, query coordinator don't have the log of", zap.String("collectionID", fmt.Sprintln(collectionID)))
return status, nil
}
@ -237,53 +190,65 @@ func (qc *QueryCoord) LoadPartitions(ctx context.Context, req *querypb.LoadParti
partitionIDs := req.PartitionIDs
log.Debug("LoadPartitionRequest received", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", collectionID), zap.Int64s("partitionIDs", partitionIDs))
status := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_UnexpectedError,
}
hasCollection := qc.meta.hasCollection(collectionID)
if hasCollection && qc.meta.collectionInfos[collectionID].LoadCollection {
status.ErrorCode = commonpb.ErrorCode_Success
status.Reason = "collection has been loaded"
return status, nil
ErrorCode: commonpb.ErrorCode_Success,
}
if len(partitionIDs) == 0 {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
err := errors.New("partitionIDs are empty")
status.Reason = err.Error()
log.Debug("LoadPartitionRequest completed", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", req.CollectionID))
return status, err
}
partitionIDsToLoad := make([]UniqueID, 0)
for _, partitionID := range partitionIDs {
hasPartition := qc.meta.hasPartition(collectionID, partitionID)
if !hasPartition {
partitionIDsToLoad = append(partitionIDsToLoad, partitionID)
hasCollection := qc.meta.hasCollection(collectionID)
if hasCollection {
partitionIDsToLoad := make([]UniqueID, 0)
loadCollection, _ := qc.meta.getLoadCollection(collectionID)
if loadCollection {
for _, partitionID := range partitionIDs {
hasReleasePartition := qc.meta.hasReleasePartition(collectionID, partitionID)
if hasReleasePartition {
partitionIDsToLoad = append(partitionIDsToLoad, partitionID)
}
}
} else {
for _, partitionID := range partitionIDs {
hasPartition := qc.meta.hasPartition(collectionID, partitionID)
if !hasPartition {
partitionIDsToLoad = append(partitionIDsToLoad, partitionID)
}
}
}
}
req.PartitionIDs = partitionIDsToLoad
if len(req.PartitionIDs) > 0 {
loadPartitionTask := &LoadPartitionTask{
BaseTask: BaseTask{
ctx: qc.loopCtx,
Condition: NewTaskCondition(qc.loopCtx),
triggerCondition: querypb.TriggerCondition_grpcRequest,
},
LoadPartitionsRequest: req,
dataCoord: qc.dataCoordClient,
cluster: qc.cluster,
meta: qc.meta,
}
qc.scheduler.Enqueue([]task{loadPartitionTask})
err := loadPartitionTask.WaitToFinish()
if err != nil {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
return status, err
if len(partitionIDsToLoad) == 0 {
log.Debug("LoadPartitionRequest completed", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", req.CollectionID))
return status, nil
}
req.PartitionIDs = partitionIDsToLoad
}
loadPartitionTask := &LoadPartitionTask{
BaseTask: BaseTask{
ctx: qc.loopCtx,
Condition: NewTaskCondition(qc.loopCtx),
triggerCondition: querypb.TriggerCondition_grpcRequest,
},
LoadPartitionsRequest: req,
dataCoord: qc.dataCoordClient,
cluster: qc.cluster,
meta: qc.meta,
}
qc.scheduler.Enqueue([]task{loadPartitionTask})
err := loadPartitionTask.WaitToFinish()
if err != nil {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
log.Debug("LoadPartitionRequest completed", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", req.CollectionID))
return status, err
}
status.ErrorCode = commonpb.ErrorCode_Success
log.Debug("LoadPartitionRequest completed", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", req.CollectionID))
return status, nil
}
@ -296,33 +261,36 @@ func (qc *QueryCoord) ReleasePartitions(ctx context.Context, req *querypb.Releas
status := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
}
toReleasedPartitionID := make([]UniqueID, 0)
for _, partitionID := range partitionIDs {
hasPartition := qc.meta.hasPartition(collectionID, partitionID)
if hasPartition {
toReleasedPartitionID = append(toReleasedPartitionID, partitionID)
}
hasCollection := qc.meta.hasCollection(collectionID)
if !hasCollection {
log.Warn("release partitions end, query coordinator don't have the log of", zap.String("collectionID", fmt.Sprintln(collectionID)))
return status, nil
}
if len(toReleasedPartitionID) > 0 {
req.PartitionIDs = toReleasedPartitionID
releasePartitionTask := &ReleasePartitionTask{
BaseTask: BaseTask{
ctx: qc.loopCtx,
Condition: NewTaskCondition(qc.loopCtx),
triggerCondition: querypb.TriggerCondition_grpcRequest,
},
ReleasePartitionsRequest: req,
cluster: qc.cluster,
}
qc.scheduler.Enqueue([]task{releasePartitionTask})
if len(partitionIDs) == 0 {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
err := errors.New("partitionIDs are empty")
status.Reason = err.Error()
log.Debug("releasePartitionsRequest completed", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", req.CollectionID))
return status, err
}
err := releasePartitionTask.WaitToFinish()
if err != nil {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
return status, err
}
releasePartitionTask := &ReleasePartitionTask{
BaseTask: BaseTask{
ctx: qc.loopCtx,
Condition: NewTaskCondition(qc.loopCtx),
triggerCondition: querypb.TriggerCondition_grpcRequest,
},
ReleasePartitionsRequest: req,
cluster: qc.cluster,
}
qc.scheduler.Enqueue([]task{releasePartitionTask})
err := releasePartitionTask.WaitToFinish()
if err != nil {
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
return status, err
}
log.Debug("ReleasePartitionRequest completed", zap.String("role", Params.RoleName), zap.Int64("msgID", req.Base.MsgID), zap.Int64("collectionID", collectionID), zap.Int64s("partitionIDs", partitionIDs))
qc.meta.printMeta()

50
internal/querycoord/meta.go
View File

@ -170,6 +170,21 @@ func (m *meta) hasPartition(collectionID UniqueID, partitionID UniqueID) bool {
return false
}
func (m *meta) hasReleasePartition(collectionID UniqueID, partitionID UniqueID) bool {
m.RLock()
defer m.RUnlock()
if info, ok := m.collectionInfos[collectionID]; ok {
for _, id := range info.ReleasedPartitionIDs {
if partitionID == id {
return true
}
}
}
return false
}
func (m *meta) addCollection(collectionID UniqueID, schema *schemapb.CollectionSchema) error {
m.Lock()
defer m.Unlock()
@ -208,6 +223,13 @@ func (m *meta) addPartition(collectionID UniqueID, partitionID UniqueID) error {
}
}
col.PartitionIDs = append(col.PartitionIDs, partitionID)
releasedPartitionIDs := make([]UniqueID, 0)
for _, id := range col.ReleasedPartitionIDs {
if id != partitionID {
releasedPartitionIDs = append(releasedPartitionIDs, id)
}
}
col.ReleasedPartitionIDs = releasedPartitionIDs
m.partitionStates[partitionID] = querypb.PartitionState_NotPresent
log.Debug("add a partition to meta", zap.Int64s("partitionIDs", col.PartitionIDs))
err := m.saveCollectionInfo(collectionID, col)
@ -381,9 +403,26 @@ func (m *meta) releasePartition(collectionID UniqueID, partitionID UniqueID) {
}
}
info.PartitionIDs = newPartitionIDs
releasedPartitionIDs := make([]UniqueID, 0)
for _, id := range info.ReleasedPartitionIDs {
if id != partitionID {
releasedPartitionIDs = append(releasedPartitionIDs, id)
}
}
releasedPartitionIDs = append(releasedPartitionIDs, partitionID)
info.ReleasedPartitionIDs = releasedPartitionIDs
err := m.saveCollectionInfo(collectionID, info)
if err != nil {
log.Error("save collectionInfo error", zap.Any("error", err.Error()), zap.Int64("collectionID", collectionID))
}
}
for id, info := range m.segmentInfos {
if info.PartitionID == partitionID {
err := m.removeSegmentInfo(id)
if err != nil {
log.Error("delete segmentInfo error", zap.Any("error", err.Error()), zap.Int64("collectionID", collectionID), zap.Int64("segmentID", id))
}
delete(m.segmentInfos, id)
}
}
@ -579,6 +618,17 @@ func (m *meta) setLoadCollection(collectionID UniqueID, state bool) error {
return errors.New("setLoadCollection: can't find collection in collectionInfos")
}
func (m *meta) getLoadCollection(collectionID UniqueID) (bool, error) {
m.RLock()
defer m.RUnlock()
if info, ok := m.collectionInfos[collectionID]; ok {
return info.LoadCollection, nil
}
return false, errors.New("getLoadCollection: can't find collection in collectionInfos")
}
func (m *meta) printMeta() {
for id, info := range m.collectionInfos {
log.Debug("query coordinator meta: collectionInfo", zap.Int64("collectionID", id), zap.Any("info", info))

67
internal/querycoord/query_coord_test.go
View File

@ -18,7 +18,6 @@ import (
"github.com/stretchr/testify/assert"
"github.com/milvus-io/milvus/internal/msgstream"
"github.com/milvus-io/milvus/internal/proto/commonpb"
"github.com/milvus-io/milvus/internal/proto/querypb"
)
@ -53,36 +52,36 @@ func TestQueryCoord_Init(t *testing.T) {
service.Stop()
}
func TestQueryCoord_load(t *testing.T) {
ctx := context.Background()
msFactory := msgstream.NewPmsFactory()
service, err := NewQueryCoord(context.Background(), msFactory)
assert.Nil(t, err)
service.Init()
service.Start()
service.SetRootCoord(NewRootCoordMock())
service.SetDataCoord(NewDataMock())
registerNodeRequest := &querypb.RegisterNodeRequest{
Address: &commonpb.Address{},
}
service.RegisterNode(ctx, registerNodeRequest)
t.Run("Test LoadSegment", func(t *testing.T) {
loadCollectionRequest := &querypb.LoadCollectionRequest{
CollectionID: 1,
}
response, err := service.LoadCollection(ctx, loadCollectionRequest)
assert.Nil(t, err)
assert.Equal(t, response.ErrorCode, commonpb.ErrorCode_Success)
})
t.Run("Test LoadPartition", func(t *testing.T) {
loadPartitionRequest := &querypb.LoadPartitionsRequest{
CollectionID: 1,
PartitionIDs: []UniqueID{1},
}
response, err := service.LoadPartitions(ctx, loadPartitionRequest)
assert.Nil(t, err)
assert.Equal(t, response.ErrorCode, commonpb.ErrorCode_Success)
})
}
//func TestQueryCoord_load(t *testing.T) {
// ctx := context.Background()
// msFactory := msgstream.NewPmsFactory()
// service, err := NewQueryCoord(context.Background(), msFactory)
// assert.Nil(t, err)
// service.Init()
// service.Start()
// service.SetRootCoord(NewRootCoordMock())
// service.SetDataCoord(NewDataMock())
// registerNodeRequest := &querypb.RegisterNodeRequest{
// Address: &commonpb.Address{},
// }
// service.RegisterNode(ctx, registerNodeRequest)
//
// t.Run("Test LoadSegment", func(t *testing.T) {
// loadCollectionRequest := &querypb.LoadCollectionRequest{
// CollectionID: 1,
// }
// response, err := service.LoadCollection(ctx, loadCollectionRequest)
// assert.Nil(t, err)
// assert.Equal(t, response.ErrorCode, commonpb.ErrorCode_Success)
// })
//
// t.Run("Test LoadPartition", func(t *testing.T) {
// loadPartitionRequest := &querypb.LoadPartitionsRequest{
// CollectionID: 1,
// PartitionIDs: []UniqueID{1},
// }
// response, err := service.LoadPartitions(ctx, loadPartitionRequest)
// assert.Nil(t, err)
// assert.Equal(t, response.ErrorCode, commonpb.ErrorCode_Success)
// })
//}

224
internal/querycoord/task.go
View File

@ -13,6 +13,7 @@ package querycoord
import (
"context"
"errors"
"fmt"
"github.com/golang/protobuf/proto"
@ -49,9 +50,9 @@ type task interface {
SetID(id UniqueID)
Type() commonpb.MsgType
Timestamp() Timestamp
PreExecute(ctx context.Context) error
PreExecute()
Execute(ctx context.Context) error
PostExecute(ctx context.Context) error
PostExecute()
WaitToFinish() error
Notify(err error)
TaskPriority() querypb.TriggerCondition
@ -144,14 +145,17 @@ func (lct *LoadCollectionTask) Timestamp() Timestamp {
return lct.Base.Timestamp
}
func (lct *LoadCollectionTask) PreExecute(ctx context.Context) error {
func (lct *LoadCollectionTask) PreExecute() {
collectionID := lct.CollectionID
schema := lct.Schema
status := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
}
lct.result = status
log.Debug("start do LoadCollectionTask",
zap.Int64("msgID", lct.ID()),
zap.Int64("collectionID", collectionID),
zap.Stringer("schema", schema))
return nil
}
func (lct *LoadCollectionTask) Execute(ctx context.Context) error {
@ -161,6 +165,7 @@ func (lct *LoadCollectionTask) Execute(ctx context.Context) error {
}
lct.meta.addCollection(collectionID, lct.Schema)
lct.meta.setLoadCollection(collectionID, true)
showPartitionRequest := &milvuspb.ShowPartitionsRequest{
Base: &commonpb.MsgBase{
MsgType: commonpb.MsgType_ShowPartitions,
@ -173,15 +178,20 @@ func (lct *LoadCollectionTask) Execute(ctx context.Context) error {
lct.result = status
return err
}
log.Debug("loadCollectionTask: get recovery info", zap.Int64s("partitionIDs", showPartitionResponse.PartitionIDs))
toLoadPartitionIDs := make([]UniqueID, 0)
for _, id := range showPartitionResponse.PartitionIDs {
if !lct.meta.hasPartition(collectionID, id) {
toLoadPartitionIDs = append(toLoadPartitionIDs, id)
}
}
log.Debug("loadCollectionTask: toLoadPartitionIDs", zap.Int64s("partitionIDs", toLoadPartitionIDs))
segment2Binlog := make(map[UniqueID]*querypb.SegmentLoadInfo)
watchRequests := make(map[string]*querypb.WatchDmChannelsRequest)
channelsToWatch := make([]string, 0)
segmentsToLoad := make([]UniqueID, 0)
partitionIDs := showPartitionResponse.PartitionIDs
log.Debug("partitionIDs", zap.Int64s("partitionIDs", partitionIDs))
for _, partitionID := range partitionIDs {
for _, partitionID := range toLoadPartitionIDs {
getRecoveryInfoRequest := &datapb.GetRecoveryInfoRequest{
Base: lct.Base,
CollectionID: collectionID,
@ -325,14 +335,42 @@ func (lct *LoadCollectionTask) Execute(ctx context.Context) error {
return nil
}
func (lct *LoadCollectionTask) PostExecute(ctx context.Context) error {
func (lct *LoadCollectionTask) PostExecute() {
collectionID := lct.CollectionID
lct.meta.addCollection(collectionID, lct.Schema)
lct.meta.collectionInfos[collectionID].LoadCollection = true
if lct.result.ErrorCode != commonpb.ErrorCode_Success {
lct.childTasks = make([]task, 0)
for nodeID, node := range lct.cluster.nodes {
if !node.isOnService() {
continue
}
req := &querypb.ReleaseCollectionRequest{
Base: &commonpb.MsgBase{
MsgType: commonpb.MsgType_ReleaseCollection,
MsgID: lct.Base.MsgID,
Timestamp: lct.Base.Timestamp,
SourceID: lct.Base.SourceID,
},
DbID: lct.DbID,
CollectionID: lct.CollectionID,
NodeID: nodeID,
}
releaseCollectionTask := &ReleaseCollectionTask{
BaseTask: BaseTask{
ctx: lct.ctx,
Condition: NewTaskCondition(lct.ctx),
triggerCondition: querypb.TriggerCondition_grpcRequest,
},
ReleaseCollectionRequest: req,
cluster: lct.cluster,
}
lct.AddChildTask(releaseCollectionTask)
log.Debug("loadCollectionTask: add a releaseCollectionTask to loadCollectionTask's childTask", zap.Any("task", releaseCollectionTask))
}
}
log.Debug("LoadCollectionTask postExecute done",
zap.Int64("msgID", lct.ID()),
zap.Int64("collectionID", collectionID))
return nil
}
type ReleaseCollectionTask struct {
@ -353,18 +391,21 @@ func (rct *ReleaseCollectionTask) Timestamp() Timestamp {
return rct.Base.Timestamp
}
func (rct *ReleaseCollectionTask) PreExecute(ctx context.Context) error {
func (rct *ReleaseCollectionTask) PreExecute() {
collectionID := rct.CollectionID
status := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
}
rct.result = status
log.Debug("start do ReleaseCollectionTask",
zap.Int64("msgID", rct.ID()),
zap.Int64("collectionID", collectionID))
return nil
}
func (rct *ReleaseCollectionTask) Execute(ctx context.Context) error {
collectionID := rct.CollectionID
status := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
ErrorCode: commonpb.ErrorCode_UnexpectedError,
}
if rct.NodeID <= 0 {
@ -387,17 +428,22 @@ func (rct *ReleaseCollectionTask) Execute(ctx context.Context) error {
log.Debug("ReleaseCollectionTask: add a releaseCollectionTask to releaseCollectionTask's childTask", zap.Any("task", releaseCollectionTask))
}
} else {
_, err := rct.cluster.releaseCollection(ctx, rct.NodeID, rct.ReleaseCollectionRequest)
res, err := rct.cluster.releaseCollection(ctx, rct.NodeID, rct.ReleaseCollectionRequest)
if err != nil {
log.Error("ReleaseCollectionTask: release collection end, node occur error", zap.String("nodeID", fmt.Sprintln(rct.NodeID)))
status.ErrorCode = commonpb.ErrorCode_UnexpectedError
status.Reason = err.Error()
rct.result = status
return err
}
if res.ErrorCode != commonpb.ErrorCode_Success {
log.Error("ReleaseCollectionTask: release collection end, node occur error", zap.String("nodeID", fmt.Sprintln(rct.NodeID)))
err = errors.New("queryNode releaseCollection failed")
status.Reason = err.Error()
rct.result = status
return err
}
}
rct.result = status
log.Debug("ReleaseCollectionTask Execute done",
zap.Int64("msgID", rct.ID()),
zap.Int64("collectionID", collectionID),
@ -405,14 +451,13 @@ func (rct *ReleaseCollectionTask) Execute(ctx context.Context) error {
return nil
}
func (rct *ReleaseCollectionTask) PostExecute(ctx context.Context) error {
func (rct *ReleaseCollectionTask) PostExecute() {
collectionID := rct.CollectionID
log.Debug("ReleaseCollectionTask postExecute done",
zap.Int64("msgID", rct.ID()),
zap.Int64("collectionID", collectionID),
zap.Int64("nodeID", rct.NodeID))
return nil
}
type LoadPartitionTask struct {
@ -421,6 +466,7 @@ type LoadPartitionTask struct {
dataCoord types.DataCoord
cluster *queryNodeCluster
meta *meta
addCol bool
}
func (lpt *LoadPartitionTask) Marshal() string {
@ -435,19 +481,25 @@ func (lpt *LoadPartitionTask) Timestamp() Timestamp {
return lpt.Base.Timestamp
}
func (lpt *LoadPartitionTask) PreExecute(ctx context.Context) error {
func (lpt *LoadPartitionTask) PreExecute() {
collectionID := lpt.CollectionID
status := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
}
lpt.result = status
log.Debug("start do LoadPartitionTask",
zap.Int64("msgID", lpt.ID()),
zap.Int64("collectionID", collectionID))
return nil
}
func (lpt *LoadPartitionTask) Execute(ctx context.Context) error {
collectionID := lpt.CollectionID
partitionIDs := lpt.PartitionIDs
lpt.meta.addCollection(collectionID, lpt.Schema)
if !lpt.meta.hasCollection(collectionID) {
lpt.meta.addCollection(collectionID, lpt.Schema)
lpt.addCol = true
}
for _, id := range partitionIDs {
lpt.meta.addPartition(collectionID, id)
}
@ -595,20 +647,79 @@ func (lpt *LoadPartitionTask) Execute(ctx context.Context) error {
zap.Int64("msgID", lpt.ID()),
zap.Int64("collectionID", collectionID),
zap.Int64s("partitionIDs", partitionIDs))
status.ErrorCode = commonpb.ErrorCode_Success
lpt.result = status
return nil
}
func (lpt *LoadPartitionTask) PostExecute(ctx context.Context) error {
func (lpt *LoadPartitionTask) PostExecute() {
collectionID := lpt.CollectionID
partitionIDs := lpt.PartitionIDs
if lpt.result.ErrorCode != commonpb.ErrorCode_Success {
lpt.childTasks = make([]task, 0)
if lpt.addCol {
for nodeID, node := range lpt.cluster.nodes {
if !node.isOnService() {
continue
}
req := &querypb.ReleaseCollectionRequest{
Base: &commonpb.MsgBase{
MsgType: commonpb.MsgType_ReleaseCollection,
MsgID: lpt.Base.MsgID,
Timestamp: lpt.Base.Timestamp,
SourceID: lpt.Base.SourceID,
},
DbID: lpt.DbID,
CollectionID: lpt.CollectionID,
NodeID: nodeID,
}
releaseCollectionTask := &ReleaseCollectionTask{
BaseTask: BaseTask{
ctx: lpt.ctx,
Condition: NewTaskCondition(lpt.ctx),
triggerCondition: querypb.TriggerCondition_grpcRequest,
},
ReleaseCollectionRequest: req,
cluster: lpt.cluster,
}
lpt.AddChildTask(releaseCollectionTask)
log.Debug("loadPartitionTask: add a releaseCollectionTask to loadPartitionTask's childTask", zap.Any("task", releaseCollectionTask))
}
} else {
for nodeID, node := range lpt.cluster.nodes {
if !node.isOnService() {
continue
}
req := &querypb.ReleasePartitionsRequest{
Base: &commonpb.MsgBase{
MsgType: commonpb.MsgType_ReleasePartitions,
MsgID: lpt.Base.MsgID,
Timestamp: lpt.Base.Timestamp,
SourceID: lpt.Base.SourceID,
},
DbID: lpt.DbID,
CollectionID: lpt.CollectionID,
PartitionIDs: partitionIDs,
NodeID: nodeID,
}
releasePartitionTask := &ReleasePartitionTask{
BaseTask: BaseTask{
ctx: lpt.ctx,
Condition: NewTaskCondition(lpt.ctx),
triggerCondition: querypb.TriggerCondition_grpcRequest,
},
ReleasePartitionsRequest: req,
cluster: lpt.cluster,
}
lpt.AddChildTask(releasePartitionTask)
log.Debug("loadPartitionTask: add a releasePartitionTask to loadPartitionTask's childTask", zap.Any("task", releasePartitionTask))
}
}
}
log.Debug("LoadPartitionTask postExecute done",
zap.Int64("msgID", lpt.ID()),
zap.Int64("collectionID", collectionID),
zap.Int64s("partitionIDs", partitionIDs))
//lpt.cancel()
return nil
}
type ReleasePartitionTask struct {
@ -629,23 +740,29 @@ func (rpt *ReleasePartitionTask) Timestamp() Timestamp {
return rpt.Base.Timestamp
}
func (rpt *ReleasePartitionTask) PreExecute(ctx context.Context) error {
func (rpt *ReleasePartitionTask) PreExecute() {
collectionID := rpt.CollectionID
status := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
}
rpt.result = status
log.Debug("start do releasePartitionTask",
zap.Int64("msgID", rpt.ID()),
zap.Int64("collectionID", collectionID))
return nil
}
func (rpt *ReleasePartitionTask) Execute(ctx context.Context) error {
collectionID := rpt.CollectionID
partitionIDs := rpt.PartitionIDs
status := &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
ErrorCode: commonpb.ErrorCode_UnexpectedError,
}
if rpt.NodeID <= 0 {
for nodeID := range rpt.cluster.nodes {
for nodeID, node := range rpt.cluster.nodes {
if !node.isOnService() {
continue
}
req := proto.Clone(rpt.ReleasePartitionsRequest).(*querypb.ReleasePartitionsRequest)
req.NodeID = nodeID
releasePartitionTask := &ReleasePartitionTask{
@ -662,14 +779,22 @@ func (rpt *ReleasePartitionTask) Execute(ctx context.Context) error {
log.Debug("ReleasePartitionTask: add a releasePartitionTask to releasePartitionTask's childTask", zap.Any("task", releasePartitionTask))
}
} else {
status, err := rpt.cluster.releasePartitions(ctx, rpt.NodeID, rpt.ReleasePartitionsRequest)
res, err := rpt.cluster.releasePartitions(ctx, rpt.NodeID, rpt.ReleasePartitionsRequest)
if err != nil {
log.Error("ReleasePartitionsTask: release partition end, node occur error", zap.String("nodeID", fmt.Sprintln(rpt.NodeID)))
status.Reason = err.Error()
rpt.result = status
return err
}
if res.ErrorCode != commonpb.ErrorCode_Success {
log.Error("ReleasePartitionsTask: release partition end, node occur error", zap.String("nodeID", fmt.Sprintln(rpt.NodeID)))
err = errors.New("queryNode releasePartition failed")
status.Reason = err.Error()
rpt.result = status
return err
}
}
rpt.result = status
log.Debug("ReleasePartitionTask Execute done",
zap.Int64("msgID", rpt.ID()),
zap.Int64("collectionID", collectionID),
@ -678,7 +803,7 @@ func (rpt *ReleasePartitionTask) Execute(ctx context.Context) error {
return nil
}
func (rpt *ReleasePartitionTask) PostExecute(ctx context.Context) error {
func (rpt *ReleasePartitionTask) PostExecute() {
collectionID := rpt.CollectionID
partitionIDs := rpt.PartitionIDs
@ -687,7 +812,6 @@ func (rpt *ReleasePartitionTask) PostExecute(ctx context.Context) error {
zap.Int64("collectionID", collectionID),
zap.Int64s("partitionIDs", partitionIDs),
zap.Int64("nodeID", rpt.NodeID))
return nil
}
//****************************internal task*******************************//
@ -714,7 +838,7 @@ func (lst *LoadSegmentTask) Timestamp() Timestamp {
return lst.Base.Timestamp
}
func (lst *LoadSegmentTask) PreExecute(ctx context.Context) error {
func (lst *LoadSegmentTask) PreExecute() {
segmentIDs := make([]UniqueID, 0)
for _, info := range lst.Infos {
segmentIDs = append(segmentIDs, info.SegmentID)
@ -723,7 +847,6 @@ func (lst *LoadSegmentTask) PreExecute(ctx context.Context) error {
zap.Int64s("segmentIDs", segmentIDs),
zap.Int64("loaded nodeID", lst.NodeID),
zap.Int64("taskID", lst.ID()))
return nil
}
func (lst *LoadSegmentTask) Execute(ctx context.Context) error {
@ -738,10 +861,9 @@ func (lst *LoadSegmentTask) Execute(ctx context.Context) error {
zap.Int64("taskID", lst.ID()))
return nil
}
func (lst *LoadSegmentTask) PostExecute(ctx context.Context) error {
func (lst *LoadSegmentTask) PostExecute() {
log.Debug("loadSegmentTask postExecute done",
zap.Int64("taskID", lst.ID()))
return nil
}
func (lst *LoadSegmentTask) Reschedule() ([]task, error) {
@ -830,13 +952,12 @@ func (rst *ReleaseSegmentTask) Timestamp() Timestamp {
return rst.Base.Timestamp
}
func (rst *ReleaseSegmentTask) PreExecute(ctx context.Context) error {
func (rst *ReleaseSegmentTask) PreExecute() {
segmentIDs := rst.SegmentIDs
log.Debug("start do releaseSegmentTask",
zap.Int64s("segmentIDs", segmentIDs),
zap.Int64("loaded nodeID", rst.NodeID),
zap.Int64("taskID", rst.ID()))
return nil
}
func (rst *ReleaseSegmentTask) Execute(ctx context.Context) error {
@ -853,12 +974,11 @@ func (rst *ReleaseSegmentTask) Execute(ctx context.Context) error {
return nil
}
func (rst *ReleaseSegmentTask) PostExecute(ctx context.Context) error {
func (rst *ReleaseSegmentTask) PostExecute() {
segmentIDs := rst.SegmentIDs
log.Debug("releaseSegmentTask postExecute done",
zap.Int64s("segmentIDs", segmentIDs),
zap.Int64("taskID", rst.ID()))
return nil
}
type WatchDmChannelTask struct {
@ -884,7 +1004,7 @@ func (wdt *WatchDmChannelTask) Timestamp() Timestamp {
return wdt.Base.Timestamp
}
func (wdt *WatchDmChannelTask) PreExecute(ctx context.Context) error {
func (wdt *WatchDmChannelTask) PreExecute() {
channelInfos := wdt.Infos
channels := make([]string, 0)
for _, info := range channelInfos {
@ -894,7 +1014,6 @@ func (wdt *WatchDmChannelTask) PreExecute(ctx context.Context) error {
zap.Strings("dmChannels", channels),
zap.Int64("loaded nodeID", wdt.NodeID),
zap.Int64("taskID", wdt.ID()))
return nil
}
func (wdt *WatchDmChannelTask) Execute(ctx context.Context) error {
@ -910,10 +1029,9 @@ func (wdt *WatchDmChannelTask) Execute(ctx context.Context) error {
return nil
}
func (wdt *WatchDmChannelTask) PostExecute(ctx context.Context) error {
func (wdt *WatchDmChannelTask) PostExecute() {
log.Debug("watchDmChannelTask postExecute done",
zap.Int64("taskID", wdt.ID()))
return nil
}
func (wdt *WatchDmChannelTask) Reschedule() ([]task, error) {
@ -1004,14 +1122,13 @@ func (wqt *WatchQueryChannelTask) Timestamp() Timestamp {
return wqt.Base.Timestamp
}
func (wqt *WatchQueryChannelTask) PreExecute(ctx context.Context) error {
func (wqt *WatchQueryChannelTask) PreExecute() {
log.Debug("start do WatchQueryChannelTask",
zap.Int64("collectionID", wqt.CollectionID),
zap.String("queryChannel", wqt.RequestChannelID),
zap.String("queryResultChannel", wqt.ResultChannelID),
zap.Int64("loaded nodeID", wqt.NodeID),
zap.Int64("taskID", wqt.ID()))
return nil
}
func (wqt *WatchQueryChannelTask) Execute(ctx context.Context) error {
@ -1030,13 +1147,12 @@ func (wqt *WatchQueryChannelTask) Execute(ctx context.Context) error {
return nil
}
func (wqt *WatchQueryChannelTask) PostExecute(ctx context.Context) error {
func (wqt *WatchQueryChannelTask) PostExecute() {
log.Debug("WatchQueryChannelTask postExecute done",
zap.Int64("collectionID", wqt.CollectionID),
zap.String("queryChannel", wqt.RequestChannelID),
zap.String("queryResultChannel", wqt.ResultChannelID),
zap.Int64("taskID", wqt.ID()))
return nil
}
//****************************handoff task********************************//
@ -1065,12 +1181,11 @@ func (lbt *LoadBalanceTask) Timestamp() Timestamp {
return lbt.Base.Timestamp
}
func (lbt *LoadBalanceTask) PreExecute(ctx context.Context) error {
func (lbt *LoadBalanceTask) PreExecute() {
log.Debug("start do LoadBalanceTask",
zap.Int64s("sourceNodeIDs", lbt.SourceNodeIDs),
zap.Any("balanceReason", lbt.BalanceReason),
zap.Int64("taskID", lbt.ID()))
return nil
}
func (lbt *LoadBalanceTask) Execute(ctx context.Context) error {
@ -1272,7 +1387,7 @@ func (lbt *LoadBalanceTask) Execute(ctx context.Context) error {
return nil
}
func (lbt *LoadBalanceTask) PostExecute(ctx context.Context) error {
func (lbt *LoadBalanceTask) PostExecute() {
for _, id := range lbt.SourceNodeIDs {
err := lbt.cluster.removeNodeInfo(id)
if err != nil {
@ -1283,7 +1398,6 @@ func (lbt *LoadBalanceTask) PostExecute(ctx context.Context) error {
zap.Int64s("sourceNodeIDs", lbt.SourceNodeIDs),
zap.Any("balanceReason", lbt.BalanceReason),
zap.Int64("taskID", lbt.ID()))
return nil
}
func shuffleChannelsToQueryNode(dmChannels []string, cluster *queryNodeCluster) []int64 {

30
internal/querycoord/task_scheduler.go
View File

@ -438,22 +438,16 @@ func (scheduler *TaskScheduler) processTask(t task) error {
})
defer span.Finish()
span.LogFields(oplog.Int64("processTask: scheduler process PreExecute", t.ID()))
t.PreExecute()
key := fmt.Sprintf("%s/%d", taskInfoPrefix, t.ID())
err := scheduler.client.Save(key, strconv.Itoa(int(taskDoing)))
if err != nil {
log.Debug("processTask: update task state err", zap.String("reason", err.Error()))
trace.LogError(span, err)
return err
}
err = t.PreExecute(ctx)
if err != nil {
log.Debug("processTask: preExecute err", zap.String("reason", err.Error()))
trace.LogError(span, err)
return err
}
span.LogFields(oplog.Int64("processTask: scheduler process Execute", t.ID()))
err = t.Execute(ctx)
if err != nil {
@ -461,13 +455,6 @@ func (scheduler *TaskScheduler) processTask(t task) error {
trace.LogError(span, err)
return err
}
span.LogFields(oplog.Int64("processTask: scheduler process PostExecute", t.ID()))
err = t.PostExecute(ctx)
if err != nil {
log.Debug("processTask: postExecute err", zap.String("reason", err.Error()))
trace.LogError(span, err)
return err
}
for _, childTask := range t.GetChildTask() {
if childTask == nil {
@ -487,6 +474,8 @@ func (scheduler *TaskScheduler) processTask(t task) error {
kvs[stateKey] = strconv.Itoa(int(taskUndo))
err = scheduler.client.MultiSave(kvs)
if err != nil {
log.Debug("processTask: save active task info err", zap.String("reason", err.Error()))
trace.LogError(span, err)
return err
}
log.Debug("processTask: save active task to etcd", zap.Int64("parent taskID", t.ID()), zap.Int64("child taskID", childTask.ID()))
@ -498,7 +487,11 @@ func (scheduler *TaskScheduler) processTask(t task) error {
trace.LogError(span, err)
return err
}
return err
span.LogFields(oplog.Int64("processTask: scheduler process PostExecute", t.ID()))
t.PostExecute()
return nil
}
func (scheduler *TaskScheduler) scheduleLoop() {
@ -518,7 +511,7 @@ func (scheduler *TaskScheduler) scheduleLoop() {
if err != nil {
log.Error("scheduleLoop: process task error", zap.Any("error", err.Error()))
t.Notify(err)
continue
t.PostExecute()
}
if t.Type() == commonpb.MsgType_LoadCollection || t.Type() == commonpb.MsgType_LoadPartitions {
t.Notify(err)
@ -534,9 +527,6 @@ func (scheduler *TaskScheduler) scheduleLoop() {
}
}
activeTaskWg.Wait()
//if t.Type() == commonpb.MsgType_ReleaseCollection || t.Type() == commonpb.MsgType_ReleasePartitions {
// t.Notify(err)
//}
keys := make([]string, 0)
taskKey := fmt.Sprintf("%s/%d", triggerTaskPrefix, t.ID())
stateKey := fmt.Sprintf("%s/%d", taskInfoPrefix, t.ID())

12
internal/querynode/query_node_test.go
View File

@ -26,7 +26,6 @@ import (
"github.com/milvus-io/milvus/internal/msgstream"
"github.com/milvus-io/milvus/internal/proto/commonpb"
"github.com/milvus-io/milvus/internal/proto/etcdpb"
"github.com/milvus-io/milvus/internal/proto/internalpb"
"github.com/milvus-io/milvus/internal/proto/querypb"
"github.com/milvus-io/milvus/internal/proto/schemapb"
"github.com/milvus-io/milvus/internal/types"
@ -199,17 +198,6 @@ func refreshChannelNames() {
Params.StatsChannelName = Params.StatsChannelName + suffix
}
func (q *queryCoordMock) RegisterNode(ctx context.Context, req *querypb.RegisterNodeRequest) (*querypb.RegisterNodeResponse, error) {
return &querypb.RegisterNodeResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_Success,
},
InitParams: &internalpb.InitParams{
NodeID: int64(1),
},
}, nil
}
func newMessageStreamFactory() (msgstream.Factory, error) {
const receiveBufSize = 1024

1
internal/types/types.go
View File

@ -181,7 +181,6 @@ type QueryCoord interface {
Component
TimeTickProvider
RegisterNode(ctx context.Context, req *querypb.RegisterNodeRequest) (*querypb.RegisterNodeResponse, error)
ShowCollections(ctx context.Context, req *querypb.ShowCollectionsRequest) (*querypb.ShowCollectionsResponse, error)
LoadCollection(ctx context.Context, req *querypb.LoadCollectionRequest) (*commonpb.Status, error)
ReleaseCollection(ctx context.Context, req *querypb.ReleaseCollectionRequest) (*commonpb.Status, error)