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

Implement distributed api, delete partition states api 

Signed-off-by: bigsheeper <yihao.dai@zilliz.com>
pull/4973/head^2
bigsheeper 2021-01-22 11:17:18 +08:00 committed by yefu.chen
parent ef79b8d8c7
commit 46a0a6133a
7 changed files with 262 additions and 159 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
internal/distributed/querynode/client.go
View File

@ -13,28 +13,28 @@ type Client struct {
grpcClient querypb.QueryNodeClient
}
func (c *Client) Init() error {
panic("implement me")
}
func (c *Client) Start() error {
panic("implement me")
}
func (c *Client) Stop() error {
panic("implement me")
}
func (c *Client) GetComponentStates() (*internalpb2.ComponentStates, error) {
panic("implement me")
states, err := c.grpcClient.GetComponentStates(context.TODO(), nil)
if err != nil {
return nil, err
}
return states.ServerStates, nil
}
func (c *Client) GetTimeTickChannel() (string, error) {
panic("implement me")
response, err := c.grpcClient.GetTimeTickChannel(context.TODO(), nil)
if err != nil {
return "", err
}
return response.TimeTickChannelID, nil
}
func (c *Client) GetStatisticsChannel() (string, error) {
panic("implement me")
response, err := c.grpcClient.GetStatsChannel(context.TODO(), nil)
if err != nil {
return "", err
}
return response.StatsChannelID, nil
}
func (c *Client) AddQueryChannel(in *querypb.AddQueryChannelsRequest) (*commonpb.Status, error) {
@ -56,7 +56,3 @@ func (c *Client) LoadSegments(in *querypb.LoadSegmentRequest) (*commonpb.Status,
func (c *Client) ReleaseSegments(in *querypb.ReleaseSegmentRequest) (*commonpb.Status, error) {
return c.grpcClient.ReleaseSegments(context.TODO(), in)
}
func (c *Client) GetPartitionState(in *querypb.PartitionStatesRequest) (*querypb.PartitionStatesResponse, error) {
return c.grpcClient.GetPartitionState(context.TODO(), in)
}

61
internal/distributed/querynode/service.go
View File

@ -13,7 +13,7 @@ import (
type Server struct {
grpcServer *grpc.Server
node querynode.Node
node *querynode.QueryNode
}
func NewServer(ctx context.Context, queryNodeID uint64) *Server {
@ -36,9 +36,59 @@ func (s *Server) StartGrpcServer() {
}
}
func (s *Server) Start() {
func (s *Server) Init() error {
return s.Init()
}
func (s *Server) Start() error {
go s.StartGrpcServer()
s.node.Start()
return s.node.Start()
}
func (s *Server) Stop() error {
return s.Stop()
}
func (s *Server) GetTimeTickChannel(ctx context.Context, in *commonpb.Empty) (*querypb.GetTimeTickChannelResponse, error) {
// ignore ctx and in
channel, err := s.node.GetTimeTickChannel()
if err != nil {
return nil, err
}
return &querypb.GetTimeTickChannelResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_SUCCESS,
},
TimeTickChannelID: channel,
}, nil
}
func (s *Server) GetStatsChannel(ctx context.Context, in *commonpb.Empty) (*querypb.GetStatsChannelResponse, error) {
// ignore ctx and in
channel, err := s.node.GetStatisticsChannel()
if err != nil {
return nil, err
}
return &querypb.GetStatsChannelResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_SUCCESS,
},
StatsChannelID: channel,
}, nil
}
func (s *Server) GetComponentStates(ctx context.Context, in *commonpb.Empty) (*querypb.ServiceStatesResponse, error) {
// ignore ctx and in
componentStates, err := s.node.GetComponentStates()
if err != nil {
return nil, err
}
return &querypb.ServiceStatesResponse{
Status: &commonpb.Status{
ErrorCode: commonpb.ErrorCode_SUCCESS,
},
ServerStates: componentStates,
}, nil
}
func (s *Server) AddQueryChannel(ctx context.Context, in *querypb.AddQueryChannelsRequest) (*commonpb.Status, error) {
@ -65,8 +115,3 @@ func (s *Server) ReleaseSegments(ctx context.Context, in *querypb.ReleaseSegment
// ignore ctx
return s.node.ReleaseSegments(in)
}
func (s *Server) GetPartitionState(ctx context.Context, in *querypb.PartitionStatesRequest) (*querypb.PartitionStatesResponse, error) {
// ignore ctx
return s.node.GetPartitionState(in)
}

5
internal/proto/query_service.proto
View File

@ -167,10 +167,13 @@ service QueryService {
}
service QueryNode {
rpc GetTimeTickChannel(common.Empty) returns (GetTimeTickChannelResponse) {}
rpc GetStatsChannel(common.Empty) returns (GetStatsChannelResponse) {}
rpc GetComponentStates(common.Empty) returns (ServiceStatesResponse) {}
rpc AddQueryChannel(AddQueryChannelsRequest) returns (common.Status) {}
rpc RemoveQueryChannel(RemoveQueryChannelsRequest) returns (common.Status) {}
rpc WatchDmChannels(WatchDmChannelsRequest) returns (common.Status) {}
rpc LoadSegments(LoadSegmentRequest) returns (common.Status) {}
rpc ReleaseSegments(ReleaseSegmentRequest) returns (common.Status) {}
rpc GetPartitionState(PartitionStatesRequest) returns (PartitionStatesResponse) {}
}

294
internal/proto/querypb/query_service.pb.go
View File

@ -1297,81 +1297,81 @@ func init() {
func init() { proto.RegisterFile("query_service.proto", fileDescriptor_5fcb6756dc1afb8d) }
var fileDescriptor_5fcb6756dc1afb8d = []byte{
// 1174 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xcc, 0x57, 0x5b, 0x6f, 0x1b, 0x45,
0x14, 0xf6, 0xc6, 0x4e, 0xa8, 0x4f, 0x1c, 0xdb, 0x99, 0x5c, 0xb5, 0xa0, 0xaa, 0x0c, 0xd0, 0xe6,
0x02, 0x0e, 0x4a, 0x25, 0xc4, 0x1b, 0x4a, 0xe2, 0x2a, 0xb2, 0xa0, 0x21, 0x6c, 0x5a, 0x21, 0x42,
0x23, 0xb3, 0xde, 0x1d, 0x92, 0x69, 0xf7, 0xe2, 0xee, 0x8c, 0x53, 0x92, 0x17, 0xa8, 0xc4, 0x3b,
0xcf, 0x3c, 0x22, 0x10, 0x48, 0xfc, 0x02, 0xfe, 0x06, 0xcf, 0xf0, 0xc2, 0x3f, 0x41, 0x3b, 0xb3,
0xde, 0xec, 0xae, 0xc7, 0x97, 0xd4, 0x69, 0xd3, 0xb7, 0x9d, 0xd9, 0x33, 0xe7, 0xfb, 0xce, 0x99,
0x39, 0x67, 0xbe, 0x81, 0xb9, 0xa7, 0x1d, 0x12, 0x9c, 0x35, 0x19, 0x09, 0x4e, 0xa9, 0x45, 0x6a,
0xed, 0xc0, 0xe7, 0x3e, 0x42, 0x2e, 0x75, 0x4e, 0x3b, 0x4c, 0x8e, 0x6a, 0xc2, 0x42, 0x2f, 0x59,
0xbe, 0xeb, 0xfa, 0x9e, 0x9c, 0xd3, 0xcb, 0xd4, 0xe3, 0x24, 0xf0, 0x4c, 0x47, 0x8e, 0xf1, 0xf7,
0x30, 0x67, 0x90, 0x63, 0xca, 0x38, 0x09, 0xf6, 0x7c, 0x9b, 0x18, 0xe4, 0x69, 0x87, 0x30, 0x8e,
0x3e, 0x84, 0x42, 0xcb, 0x64, 0x64, 0x59, 0xbb, 0xa5, 0xad, 0x4c, 0x6f, 0xbe, 0x55, 0x4b, 0xf9,
0x8d, 0x1c, 0xde, 0x67, 0xc7, 0xdb, 0x26, 0x23, 0x86, 0xb0, 0x44, 0x1f, 0xc1, 0x1b, 0xa6, 0x6d,
0x07, 0x84, 0xb1, 0xe5, 0x89, 0x01, 0x8b, 0xb6, 0xa4, 0x8d, 0xd1, 0x35, 0xc6, 0x3f, 0x69, 0x30,
0x9f, 0x66, 0xc0, 0xda, 0xbe, 0xc7, 0x08, 0xba, 0x0b, 0x53, 0x8c, 0x9b, 0xbc, 0xc3, 0x22, 0x12,
0x6f, 0x2a, 0xfd, 0x1d, 0x08, 0x13, 0x23, 0x32, 0x45, 0xdb, 0x30, 0x4d, 0x3d, 0xca, 0x9b, 0x6d,
0x33, 0x30, 0xdd, 0x2e, 0x93, 0xb7, 0xd3, 0x2b, 0xe3, 0x0c, 0x34, 0x3c, 0xca, 0xf7, 0x85, 0xa1,
0x01, 0x34, 0xfe, 0xc6, 0x47, 0xb0, 0x70, 0x70, 0xe2, 0x3f, 0xdb, 0xf1, 0x1d, 0x87, 0x58, 0x9c,
0xfa, 0xde, 0x8b, 0x27, 0x05, 0x41, 0xc1, 0x6e, 0x35, 0xea, 0x82, 0x47, 0xde, 0x10, 0xdf, 0x98,
0xc1, 0x62, 0xd6, 0xfd, 0x38, 0x11, 0xbf, 0x0b, 0x33, 0x56, 0xec, 0xaa, 0x51, 0x0f, 0x63, 0xce,
0xaf, 0xe4, 0x8d, 0xf4, 0x24, 0x7e, 0xae, 0xc1, 0xc2, 0x67, 0xbe, 0x69, 0xbf, 0xa4, 0xa0, 0x10,
0x86, 0x52, 0x12, 0x70, 0x39, 0x2f, 0xfe, 0xa5, 0xe6, 0xf0, 0x8f, 0x1a, 0x2c, 0x1b, 0xc4, 0x21,
0x26, 0x23, 0xd7, 0x49, 0xe3, 0x07, 0x0d, 0xe6, 0xc3, 0x0d, 0xd8, 0x37, 0x03, 0x4e, 0xaf, 0x87,
0x42, 0x5b, 0x9e, 0xb0, 0x04, 0x83, 0x71, 0x4e, 0x00, 0x86, 0x52, 0xbb, 0xeb, 0xe9, 0xe2, 0x00,
0xa4, 0xe6, 0xb0, 0x0b, 0x95, 0x18, 0x2d, 0x5c, 0x4e, 0x18, 0xba, 0x05, 0xd3, 0x09, 0x13, 0x01,
0x98, 0x37, 0x92, 0x53, 0xe8, 0x63, 0x98, 0x0c, 0x21, 0x88, 0x88, 0xaf, 0xbc, 0x89, 0x6b, 0xbd,
0xdd, 0xa5, 0x96, 0xf6, 0x6a, 0xc8, 0x05, 0xf8, 0x77, 0x0d, 0x16, 0x33, 0x78, 0xaf, 0x3c, 0xcb,
0x3d, 0x79, 0x29, 0x28, 0xf2, 0xf2, 0xa7, 0x06, 0x4b, 0x3d, 0x44, 0xc7, 0xd9, 0x8c, 0x43, 0x58,
0x8c, 0x01, 0x9a, 0x36, 0x61, 0x56, 0x40, 0xdb, 0xe1, 0xb7, 0xdc, 0x96, 0xe9, 0xcd, 0x77, 0x86,
0x27, 0x91, 0x19, 0x0b, 0xb1, 0x8b, 0x7a, 0xc2, 0x03, 0xfe, 0x4d, 0x83, 0xf9, 0xb0, 0x88, 0xaf,
0xef, 0xe4, 0x8e, 0x94, 0xd3, 0x3f, 0x34, 0x58, 0x8a, 0xea, 0xfc, 0x35, 0x67, 0xfa, 0x8b, 0x06,
0xfa, 0x4e, 0x40, 0x4c, 0x4e, 0xbe, 0x08, 0xf7, 0x61, 0xe7, 0xc4, 0xf4, 0x3c, 0xe2, 0x8c, 0x77,
0x00, 0xee, 0x40, 0x25, 0x90, 0xc1, 0x36, 0x2d, 0xe9, 0x4f, 0x50, 0x2f, 0x1a, 0xe5, 0x68, 0x3a,
0x42, 0x41, 0xef, 0x41, 0x39, 0x20, 0xac, 0xe3, 0x5c, 0xd8, 0xe5, 0x85, 0xdd, 0x8c, 0x9c, 0x8d,
0xcc, 0xf0, 0xaf, 0x1a, 0x2c, 0x6d, 0xd9, 0x76, 0x92, 0xe0, 0x18, 0xb5, 0xb4, 0x0e, 0xb3, 0x19,
0x76, 0x51, 0x6a, 0x8b, 0x46, 0x35, 0xcd, 0xaf, 0x51, 0x47, 0xab, 0x50, 0x4d, 0x33, 0x8c, 0x52,
0x5d, 0x34, 0x2a, 0x29, 0x8e, 0x8d, 0x3a, 0xfe, 0x47, 0x03, 0xdd, 0x20, 0xae, 0x7f, 0x4a, 0x94,
0x44, 0x5f, 0x28, 0x93, 0xdd, 0xe8, 0x26, 0xc6, 0x8b, 0x2e, 0x7f, 0x89, 0xe8, 0x0a, 0xea, 0xe8,
0x1e, 0xc3, 0xe2, 0x97, 0x26, 0xb7, 0x4e, 0xea, 0xee, 0xf8, 0x3b, 0x70, 0x13, 0x20, 0xc6, 0x93,
0x4d, 0xa1, 0x68, 0x24, 0x66, 0xf0, 0xbf, 0x1a, 0xa0, 0xb0, 0xc8, 0x0f, 0xc8, 0xb1, 0x4b, 0x3c,
0xfe, 0xea, 0x0b, 0x27, 0x73, 0x2f, 0x14, 0x7a, 0xef, 0x85, 0x9b, 0x00, 0x4c, 0xb2, 0x0b, 0x43,
0x98, 0x14, 0x85, 0x95, 0x98, 0x41, 0x3a, 0xdc, 0xf8, 0x96, 0x12, 0xc7, 0x0e, 0xff, 0x4e, 0x89,
0xbf, 0xf1, 0x18, 0xff, 0xa7, 0xc1, 0x42, 0xd4, 0x1c, 0xae, 0x2d, 0xc2, 0x11, 0x5a, 0xc3, 0x58,
0x31, 0x3e, 0xd7, 0x40, 0xdf, 0x25, 0xfc, 0x01, 0x75, 0xc9, 0x03, 0x6a, 0x3d, 0xb9, 0x92, 0xb6,
0x52, 0x83, 0x39, 0x4e, 0x5d, 0xd2, 0xe4, 0xd4, 0x7a, 0xd2, 0x53, 0xba, 0xb3, 0x3c, 0x0d, 0xd5,
0xa8, 0xe3, 0x67, 0xb0, 0xb4, 0x4b, 0x78, 0xe8, 0x84, 0x5d, 0x55, 0x5b, 0x0b, 0xbf, 0x58, 0x0f,
0x76, 0x99, 0x25, 0x30, 0x1a, 0x75, 0xfc, 0xb3, 0x06, 0x0b, 0x07, 0xf2, 0x51, 0x72, 0x15, 0xf7,
0xe9, 0xa7, 0x30, 0x13, 0x3e, 0x71, 0x48, 0xd0, 0x14, 0xca, 0xa2, 0x2b, 0xe9, 0x6f, 0xf7, 0x91,
0xf4, 0x3b, 0xbe, 0xdb, 0xf6, 0x3d, 0xe2, 0xf1, 0x08, 0xbb, 0x24, 0x17, 0xcb, 0xd1, 0xda, 0x39,
0x94, 0xd3, 0x57, 0x2d, 0x2a, 0xc1, 0x8d, 0x3d, 0x9f, 0xdf, 0xfb, 0x8e, 0x32, 0x5e, 0xcd, 0xa1,
0x32, 0xc0, 0x9e, 0xcf, 0xf7, 0x03, 0xc2, 0x88, 0xc7, 0xab, 0x1a, 0x02, 0x98, 0xfa, 0xdc, 0xab,
0x87, 0xff, 0x26, 0xd0, 0x5c, 0xa4, 0xa0, 0x4c, 0xa7, 0xe1, 0xdd, 0x27, 0xae, 0x1f, 0x9c, 0x55,
0xf3, 0xe1, 0xf2, 0x78, 0x54, 0x40, 0x55, 0x28, 0xc5, 0x26, 0xbb, 0xfb, 0x0f, 0xab, 0x93, 0xa8,
0x08, 0x93, 0xf2, 0x73, 0x6a, 0xf3, 0xaf, 0x22, 0x94, 0x44, 0x6f, 0x8c, 0x92, 0x83, 0x2c, 0x28,
0x25, 0xdf, 0x3d, 0xe8, 0x8e, 0x4a, 0x19, 0x28, 0xde, 0x66, 0xfa, 0xca, 0x70, 0x43, 0x99, 0x71,
0x9c, 0x43, 0x8f, 0xa1, 0x92, 0x7e, 0x6c, 0x30, 0xb4, 0xaa, 0x5a, 0xae, 0x7c, 0xf0, 0xe8, 0x6b,
0xa3, 0x98, 0xc6, 0x58, 0xc7, 0x50, 0x4e, 0xa9, 0x5a, 0x86, 0x56, 0xfa, 0xad, 0xcf, 0xea, 0x02,
0x7d, 0x75, 0x04, 0xcb, 0x18, 0xe8, 0x2b, 0x28, 0xa7, 0x64, 0x50, 0x1f, 0x20, 0x95, 0x54, 0xd2,
0x07, 0x1d, 0x3a, 0x9c, 0x43, 0x4d, 0x98, 0xcd, 0x4a, 0x17, 0x86, 0xd6, 0xd5, 0x09, 0x57, 0x2a,
0x9c, 0x61, 0x00, 0x87, 0x92, 0xfb, 0x45, 0x02, 0xd5, 0xfb, 0xa1, 0x7c, 0xab, 0x0d, 0xf3, 0xfd,
0x4d, 0x4c, 0x3e, 0xe1, 0xfe, 0xfd, 0x01, 0xe4, 0x2f, 0x8d, 0xd0, 0x02, 0xd4, 0xab, 0x97, 0x90,
0xae, 0x5c, 0x74, 0xcf, 0x6d, 0xf3, 0x33, 0xbd, 0xa6, 0x82, 0xef, 0xaf, 0xb9, 0x24, 0x46, 0x6f,
0xf3, 0xbc, 0x3c, 0x46, 0xff, 0x06, 0x8c, 0x73, 0xe8, 0x11, 0x54, 0x32, 0xdd, 0x71, 0x20, 0xc0,
0x7a, 0x1f, 0x00, 0x55, 0x7b, 0xc5, 0x39, 0xe4, 0x8b, 0x08, 0xb2, 0xef, 0xad, 0xb5, 0x51, 0x94,
0x7f, 0xb4, 0x0d, 0xeb, 0x23, 0xd9, 0xc6, 0x80, 0x47, 0x02, 0x30, 0xd3, 0xfb, 0x06, 0x46, 0xa4,
0xae, 0x37, 0x55, 0xdb, 0xc6, 0xb9, 0xcd, 0xbf, 0x0b, 0x50, 0x14, 0x9b, 0x25, 0xfa, 0xd4, 0x11,
0x54, 0x32, 0x7a, 0x54, 0x5d, 0x20, 0x7d, 0x44, 0xeb, 0xb0, 0x23, 0x66, 0x01, 0xea, 0x15, 0x92,
0xa8, 0xa6, 0x3e, 0xc5, 0xfd, 0x04, 0xe7, 0x30, 0x90, 0x47, 0x50, 0xc9, 0x08, 0x3a, 0xf5, 0xf6,
0xa8, 0x55, 0xdf, 0x30, 0xef, 0x0f, 0xa1, 0x94, 0x50, 0x70, 0x0c, 0xdd, 0xee, 0x57, 0xe1, 0x69,
0x05, 0x34, 0xcc, 0xed, 0xd7, 0x50, 0x49, 0x2b, 0xa7, 0x3e, 0xbd, 0x5c, 0x29, 0xaf, 0x86, 0x39,
0xf7, 0x60, 0xb6, 0xe7, 0xcc, 0xbe, 0xc4, 0x23, 0xbb, 0xbd, 0x75, 0xf8, 0xc9, 0x31, 0xe5, 0x27,
0x9d, 0x56, 0xc8, 0x64, 0xe3, 0x9c, 0x3a, 0x0e, 0x3d, 0xe7, 0xc4, 0x3a, 0xd9, 0x90, 0x5e, 0x3e,
0xb0, 0x29, 0xe3, 0x01, 0x6d, 0x75, 0x38, 0xb1, 0x37, 0xba, 0xb7, 0xfb, 0x86, 0x70, 0xbd, 0x21,
0x5c, 0xb7, 0x5b, 0xad, 0x29, 0x31, 0xbc, 0xfb, 0x7f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x90, 0x66,
0x45, 0xf8, 0x0a, 0x15, 0x00, 0x00,
// 1172 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xd4, 0x57, 0xdd, 0x6e, 0x1b, 0x45,
0x14, 0xf6, 0xc6, 0x8e, 0xa9, 0x4f, 0x1c, 0xdb, 0x99, 0xfc, 0x6a, 0x41, 0x55, 0x19, 0xa0, 0xcd,
0x0f, 0x38, 0x28, 0x95, 0x10, 0x77, 0x28, 0x89, 0xab, 0xc8, 0x82, 0x86, 0xb0, 0x69, 0x85, 0x08,
0x8d, 0xcc, 0x7a, 0x77, 0x48, 0xa6, 0xdd, 0x1f, 0x77, 0x67, 0x9c, 0x92, 0xdc, 0x40, 0x11, 0xf7,
0x5c, 0x73, 0x89, 0x40, 0x20, 0xf1, 0x04, 0xbc, 0x0b, 0xdc, 0xf0, 0x26, 0x68, 0x67, 0xd6, 0x9b,
0xdd, 0xf5, 0x38, 0x76, 0xea, 0xb4, 0x81, 0xbb, 0x9d, 0xd9, 0x33, 0xe7, 0xfb, 0xce, 0x99, 0x39,
0x67, 0xbe, 0x81, 0xd9, 0xa7, 0x5d, 0x12, 0x9c, 0xb6, 0x18, 0x09, 0x4e, 0xa8, 0x45, 0xea, 0x9d,
0xc0, 0xe7, 0x3e, 0x42, 0x2e, 0x75, 0x4e, 0xba, 0x4c, 0x8e, 0xea, 0xc2, 0x42, 0x2f, 0x5b, 0xbe,
0xeb, 0xfa, 0x9e, 0x9c, 0xd3, 0x2b, 0xd4, 0xe3, 0x24, 0xf0, 0x4c, 0x47, 0x8e, 0xf1, 0xb7, 0x30,
0x6b, 0x90, 0x23, 0xca, 0x38, 0x09, 0x76, 0x7d, 0x9b, 0x18, 0xe4, 0x69, 0x97, 0x30, 0x8e, 0xde,
0x87, 0x42, 0xdb, 0x64, 0x64, 0x49, 0xbb, 0xa5, 0x2d, 0x4f, 0x6d, 0xbc, 0x51, 0x4f, 0xf9, 0x8d,
0x1c, 0xde, 0x67, 0x47, 0x5b, 0x26, 0x23, 0x86, 0xb0, 0x44, 0x1f, 0xc0, 0x6b, 0xa6, 0x6d, 0x07,
0x84, 0xb1, 0xa5, 0x89, 0x0b, 0x16, 0x6d, 0x4a, 0x1b, 0xa3, 0x67, 0x8c, 0x7f, 0xd4, 0x60, 0x2e,
0xcd, 0x80, 0x75, 0x7c, 0x8f, 0x11, 0x74, 0x17, 0x8a, 0x8c, 0x9b, 0xbc, 0xcb, 0x22, 0x12, 0xaf,
0x2b, 0xfd, 0xed, 0x0b, 0x13, 0x23, 0x32, 0x45, 0x5b, 0x30, 0x45, 0x3d, 0xca, 0x5b, 0x1d, 0x33,
0x30, 0xdd, 0x1e, 0x93, 0x37, 0xd3, 0x2b, 0xe3, 0x0c, 0x34, 0x3d, 0xca, 0xf7, 0x84, 0xa1, 0x01,
0x34, 0xfe, 0xc6, 0x87, 0x30, 0xbf, 0x7f, 0xec, 0x3f, 0xdb, 0xf6, 0x1d, 0x87, 0x58, 0x9c, 0xfa,
0xde, 0x8b, 0x27, 0x05, 0x41, 0xc1, 0x6e, 0x37, 0x1b, 0x82, 0x47, 0xde, 0x10, 0xdf, 0x98, 0xc1,
0x42, 0xd6, 0xfd, 0x38, 0x11, 0xbf, 0x0d, 0xd3, 0x56, 0xec, 0xaa, 0xd9, 0x08, 0x63, 0xce, 0x2f,
0xe7, 0x8d, 0xf4, 0x24, 0x7e, 0xae, 0xc1, 0xfc, 0x27, 0xbe, 0x69, 0xbf, 0xa4, 0xa0, 0x10, 0x86,
0x72, 0x12, 0x70, 0x29, 0x2f, 0xfe, 0xa5, 0xe6, 0xf0, 0x0f, 0x1a, 0x2c, 0x19, 0xc4, 0x21, 0x26,
0x23, 0xd7, 0x49, 0xe3, 0x3b, 0x0d, 0xe6, 0xc2, 0x0d, 0xd8, 0x33, 0x03, 0x4e, 0xaf, 0x87, 0x42,
0x47, 0x9e, 0xb0, 0x04, 0x83, 0x71, 0x4e, 0x00, 0x86, 0x72, 0xa7, 0xe7, 0xe9, 0xfc, 0x00, 0xa4,
0xe6, 0xb0, 0x0b, 0xd5, 0x18, 0x2d, 0x5c, 0x4e, 0x18, 0xba, 0x05, 0x53, 0x09, 0x13, 0x01, 0x98,
0x37, 0x92, 0x53, 0xe8, 0x43, 0x98, 0x0c, 0x21, 0x88, 0x88, 0xaf, 0xb2, 0x81, 0xeb, 0xfd, 0xdd,
0xa5, 0x9e, 0xf6, 0x6a, 0xc8, 0x05, 0xf8, 0x37, 0x0d, 0x16, 0x32, 0x78, 0xaf, 0x3c, 0xcb, 0x7d,
0x79, 0x29, 0x28, 0xf2, 0xf2, 0x87, 0x06, 0x8b, 0x7d, 0x44, 0xc7, 0xd9, 0x8c, 0x03, 0x58, 0x88,
0x01, 0x5a, 0x36, 0x61, 0x56, 0x40, 0x3b, 0xe1, 0xb7, 0xdc, 0x96, 0xa9, 0x8d, 0xb7, 0x86, 0x27,
0x91, 0x19, 0xf3, 0xb1, 0x8b, 0x46, 0xc2, 0x03, 0xfe, 0x55, 0x83, 0xb9, 0xb0, 0x88, 0xaf, 0xef,
0xe4, 0x8e, 0x94, 0xd3, 0xdf, 0x35, 0x58, 0x8c, 0xea, 0xfc, 0x3f, 0xce, 0xf4, 0x67, 0x0d, 0xf4,
0xed, 0x80, 0x98, 0x9c, 0x7c, 0x16, 0xee, 0xc3, 0xf6, 0xb1, 0xe9, 0x79, 0xc4, 0x19, 0xef, 0x00,
0xdc, 0x81, 0x6a, 0x20, 0x83, 0x6d, 0x59, 0xd2, 0x9f, 0xa0, 0x5e, 0x32, 0x2a, 0xd1, 0x74, 0x84,
0x82, 0xde, 0x81, 0x4a, 0x40, 0x58, 0xd7, 0x39, 0xb7, 0xcb, 0x0b, 0xbb, 0x69, 0x39, 0x1b, 0x99,
0xe1, 0x5f, 0x34, 0x58, 0xdc, 0xb4, 0xed, 0x24, 0xc1, 0x31, 0x6a, 0x69, 0x0d, 0x66, 0x32, 0xec,
0xa2, 0xd4, 0x96, 0x8c, 0x5a, 0x9a, 0x5f, 0xb3, 0x81, 0x56, 0xa0, 0x96, 0x66, 0x18, 0xa5, 0xba,
0x64, 0x54, 0x53, 0x1c, 0x9b, 0x0d, 0xfc, 0x97, 0x06, 0xba, 0x41, 0x5c, 0xff, 0x84, 0x28, 0x89,
0xbe, 0x50, 0x26, 0x7b, 0xd1, 0x4d, 0x8c, 0x17, 0x5d, 0xfe, 0x12, 0xd1, 0x15, 0xd4, 0xd1, 0x3d,
0x86, 0x85, 0xcf, 0x4d, 0x6e, 0x1d, 0x37, 0xdc, 0xf1, 0x77, 0xe0, 0x26, 0x40, 0x8c, 0x27, 0x9b,
0x42, 0xc9, 0x48, 0xcc, 0xe0, 0xbf, 0x35, 0x40, 0x61, 0x91, 0xef, 0x93, 0x23, 0x97, 0x78, 0xfc,
0xd5, 0x17, 0x4e, 0xe6, 0x5e, 0x28, 0xf4, 0xdf, 0x0b, 0x37, 0x01, 0x98, 0x64, 0x17, 0x86, 0x30,
0x29, 0x0a, 0x2b, 0x31, 0x83, 0x74, 0xb8, 0xf1, 0x35, 0x25, 0x8e, 0x1d, 0xfe, 0x2d, 0x8a, 0xbf,
0xf1, 0x18, 0xff, 0xa3, 0xc1, 0x7c, 0xd4, 0x1c, 0xae, 0x2d, 0xc2, 0x11, 0x5a, 0xc3, 0x58, 0x31,
0x3e, 0xd7, 0x40, 0xdf, 0x21, 0xfc, 0x01, 0x75, 0xc9, 0x03, 0x6a, 0x3d, 0xb9, 0x92, 0xb6, 0x52,
0x87, 0x59, 0x4e, 0x5d, 0xd2, 0xe2, 0xd4, 0x7a, 0xd2, 0x57, 0xba, 0x33, 0x3c, 0x0d, 0xd5, 0x6c,
0xe0, 0x67, 0xb0, 0xb8, 0x43, 0x78, 0xe8, 0x84, 0x5d, 0x55, 0x5b, 0x0b, 0xbf, 0x58, 0x1f, 0x76,
0x85, 0x25, 0x30, 0x9a, 0x0d, 0xfc, 0x93, 0x06, 0xf3, 0xfb, 0xf2, 0x51, 0x72, 0x15, 0xf7, 0xe9,
0xc7, 0x30, 0x1d, 0x3e, 0x71, 0x48, 0xd0, 0x12, 0xca, 0xa2, 0x27, 0xe9, 0x6f, 0x0f, 0x90, 0xf4,
0xdb, 0xbe, 0xdb, 0xf1, 0x3d, 0xe2, 0xf1, 0x08, 0xbb, 0x2c, 0x17, 0xcb, 0xd1, 0xea, 0x19, 0x54,
0xd2, 0x57, 0x2d, 0x2a, 0xc3, 0x8d, 0x5d, 0x9f, 0xdf, 0xfb, 0x86, 0x32, 0x5e, 0xcb, 0xa1, 0x0a,
0xc0, 0xae, 0xcf, 0xf7, 0x02, 0xc2, 0x88, 0xc7, 0x6b, 0x1a, 0x02, 0x28, 0x7e, 0xea, 0x35, 0xc2,
0x7f, 0x13, 0x68, 0x36, 0x52, 0x50, 0xa6, 0xd3, 0xf4, 0xee, 0x13, 0xd7, 0x0f, 0x4e, 0x6b, 0xf9,
0x70, 0x79, 0x3c, 0x2a, 0xa0, 0x1a, 0x94, 0x63, 0x93, 0x9d, 0xbd, 0x87, 0xb5, 0x49, 0x54, 0x82,
0x49, 0xf9, 0x59, 0xdc, 0xf8, 0xb3, 0x04, 0x65, 0xd1, 0x1b, 0xa3, 0xe4, 0x20, 0x0b, 0xca, 0xc9,
0x77, 0x0f, 0xba, 0xa3, 0x52, 0x06, 0x8a, 0xb7, 0x99, 0xbe, 0x3c, 0xdc, 0x50, 0x66, 0x1c, 0xe7,
0xd0, 0x63, 0xa8, 0xa6, 0x1f, 0x1b, 0x0c, 0xad, 0xa8, 0x96, 0x2b, 0x1f, 0x3c, 0xfa, 0xea, 0x28,
0xa6, 0x31, 0xd6, 0x11, 0x54, 0x52, 0xaa, 0x96, 0xa1, 0xe5, 0x41, 0xeb, 0xb3, 0xba, 0x40, 0x5f,
0x19, 0xc1, 0x32, 0x06, 0xfa, 0x02, 0x2a, 0x29, 0x19, 0x34, 0x00, 0x48, 0x25, 0x95, 0xf4, 0x8b,
0x0e, 0x1d, 0xce, 0xa1, 0x16, 0xcc, 0x64, 0xa5, 0x0b, 0x43, 0x6b, 0xea, 0x84, 0x2b, 0x15, 0xce,
0x30, 0x80, 0x03, 0xc9, 0xfd, 0x3c, 0x81, 0xea, 0xfd, 0x50, 0xbe, 0xd5, 0x86, 0xf9, 0xfe, 0x2a,
0x26, 0x9f, 0x70, 0xff, 0xee, 0x05, 0xe4, 0x2f, 0x8d, 0xd0, 0x06, 0xd4, 0xaf, 0x97, 0x90, 0xae,
0x5c, 0x74, 0xcf, 0xed, 0xf0, 0x53, 0xbd, 0xae, 0x82, 0x1f, 0xac, 0xb9, 0x24, 0x46, 0x7f, 0xf3,
0xbc, 0x3c, 0xc6, 0xe0, 0x06, 0x8c, 0x73, 0xe8, 0x11, 0x54, 0x33, 0xdd, 0xf1, 0x42, 0x80, 0xb5,
0x01, 0x00, 0xaa, 0xf6, 0x8a, 0x73, 0xc8, 0x17, 0x11, 0x64, 0xdf, 0x5b, 0xab, 0xa3, 0x28, 0xff,
0x68, 0x1b, 0xd6, 0x46, 0xb2, 0x8d, 0x01, 0x0f, 0x05, 0x60, 0xa6, 0xf7, 0x5d, 0x18, 0x91, 0xba,
0xde, 0x54, 0x6d, 0x1b, 0xe7, 0x36, 0xbe, 0x2f, 0x42, 0x49, 0x6c, 0x96, 0xe8, 0x53, 0xff, 0xff,
0xfd, 0x79, 0xb9, 0xe9, 0x42, 0x87, 0x50, 0xcd, 0x08, 0x76, 0x75, 0x07, 0x19, 0xa0, 0xea, 0x87,
0xd5, 0xa0, 0x05, 0xa8, 0x5f, 0x69, 0xa3, 0xba, 0xba, 0xcc, 0x07, 0x29, 0xf2, 0x61, 0x20, 0x8f,
0xa0, 0x9a, 0x51, 0xbc, 0xea, 0xf3, 0xab, 0x96, 0xc5, 0xc3, 0xbc, 0x3f, 0x84, 0x72, 0x42, 0xe2,
0x32, 0x74, 0x7b, 0x50, 0x0b, 0x4c, 0x4b, 0xc4, 0x61, 0x6e, 0xbf, 0x84, 0x6a, 0x5a, 0x5a, 0x0e,
0xb8, 0xec, 0x94, 0xfa, 0x73, 0x88, 0xf3, 0xad, 0xcd, 0x83, 0x8f, 0x8e, 0x28, 0x3f, 0xee, 0xb6,
0xc3, 0x3f, 0xeb, 0x67, 0xd4, 0x71, 0xe8, 0x19, 0x27, 0xd6, 0xf1, 0xba, 0x5c, 0xf5, 0x9e, 0x4d,
0x19, 0x0f, 0x68, 0xbb, 0xcb, 0x89, 0xbd, 0xde, 0x93, 0x23, 0xeb, 0xc2, 0xd5, 0xba, 0x40, 0xed,
0xb4, 0xdb, 0x45, 0x31, 0xbc, 0xfb, 0x6f, 0x00, 0x00, 0x00, 0xff, 0xff, 0x88, 0x5d, 0xc7, 0x98,
0xbb, 0x15, 0x00, 0x00,
}
// Reference imports to suppress errors if they are not otherwise used.
@ -1854,12 +1854,14 @@ var _QueryService_serviceDesc = grpc.ServiceDesc{
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
type QueryNodeClient interface {
GetTimeTickChannel(ctx context.Context, in *commonpb.Empty, opts ...grpc.CallOption) (*GetTimeTickChannelResponse, error)
GetStatsChannel(ctx context.Context, in *commonpb.Empty, opts ...grpc.CallOption) (*GetStatsChannelResponse, error)
GetComponentStates(ctx context.Context, in *commonpb.Empty, opts ...grpc.CallOption) (*ServiceStatesResponse, error)
AddQueryChannel(ctx context.Context, in *AddQueryChannelsRequest, opts ...grpc.CallOption) (*commonpb.Status, error)
RemoveQueryChannel(ctx context.Context, in *RemoveQueryChannelsRequest, opts ...grpc.CallOption) (*commonpb.Status, error)
WatchDmChannels(ctx context.Context, in *WatchDmChannelsRequest, opts ...grpc.CallOption) (*commonpb.Status, error)
LoadSegments(ctx context.Context, in *LoadSegmentRequest, opts ...grpc.CallOption) (*commonpb.Status, error)
ReleaseSegments(ctx context.Context, in *ReleaseSegmentRequest, opts ...grpc.CallOption) (*commonpb.Status, error)
GetPartitionState(ctx context.Context, in *PartitionStatesRequest, opts ...grpc.CallOption) (*PartitionStatesResponse, error)
}
type queryNodeClient struct {
@ -1870,6 +1872,33 @@ func NewQueryNodeClient(cc *grpc.ClientConn) QueryNodeClient {
return &queryNodeClient{cc}
}
func (c *queryNodeClient) GetTimeTickChannel(ctx context.Context, in *commonpb.Empty, opts ...grpc.CallOption) (*GetTimeTickChannelResponse, error) {
out := new(GetTimeTickChannelResponse)
err := c.cc.Invoke(ctx, "/milvus.proto.query.QueryNode/GetTimeTickChannel", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *queryNodeClient) GetStatsChannel(ctx context.Context, in *commonpb.Empty, opts ...grpc.CallOption) (*GetStatsChannelResponse, error) {
out := new(GetStatsChannelResponse)
err := c.cc.Invoke(ctx, "/milvus.proto.query.QueryNode/GetStatsChannel", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *queryNodeClient) GetComponentStates(ctx context.Context, in *commonpb.Empty, opts ...grpc.CallOption) (*ServiceStatesResponse, error) {
out := new(ServiceStatesResponse)
err := c.cc.Invoke(ctx, "/milvus.proto.query.QueryNode/GetComponentStates", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
func (c *queryNodeClient) AddQueryChannel(ctx context.Context, in *AddQueryChannelsRequest, opts ...grpc.CallOption) (*commonpb.Status, error) {
out := new(commonpb.Status)
err := c.cc.Invoke(ctx, "/milvus.proto.query.QueryNode/AddQueryChannel", in, out, opts...)
@ -1915,29 +1944,31 @@ func (c *queryNodeClient) ReleaseSegments(ctx context.Context, in *ReleaseSegmen
return out, nil
}
func (c *queryNodeClient) GetPartitionState(ctx context.Context, in *PartitionStatesRequest, opts ...grpc.CallOption) (*PartitionStatesResponse, error) {
out := new(PartitionStatesResponse)
err := c.cc.Invoke(ctx, "/milvus.proto.query.QueryNode/GetPartitionState", in, out, opts...)
if err != nil {
return nil, err
}
return out, nil
}
// QueryNodeServer is the server API for QueryNode service.
type QueryNodeServer interface {
GetTimeTickChannel(context.Context, *commonpb.Empty) (*GetTimeTickChannelResponse, error)
GetStatsChannel(context.Context, *commonpb.Empty) (*GetStatsChannelResponse, error)
GetComponentStates(context.Context, *commonpb.Empty) (*ServiceStatesResponse, error)
AddQueryChannel(context.Context, *AddQueryChannelsRequest) (*commonpb.Status, error)
RemoveQueryChannel(context.Context, *RemoveQueryChannelsRequest) (*commonpb.Status, error)
WatchDmChannels(context.Context, *WatchDmChannelsRequest) (*commonpb.Status, error)
LoadSegments(context.Context, *LoadSegmentRequest) (*commonpb.Status, error)
ReleaseSegments(context.Context, *ReleaseSegmentRequest) (*commonpb.Status, error)
GetPartitionState(context.Context, *PartitionStatesRequest) (*PartitionStatesResponse, error)
}
// UnimplementedQueryNodeServer can be embedded to have forward compatible implementations.
type UnimplementedQueryNodeServer struct {
}
func (*UnimplementedQueryNodeServer) GetTimeTickChannel(ctx context.Context, req *commonpb.Empty) (*GetTimeTickChannelResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetTimeTickChannel not implemented")
}
func (*UnimplementedQueryNodeServer) GetStatsChannel(ctx context.Context, req *commonpb.Empty) (*GetStatsChannelResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetStatsChannel not implemented")
}
func (*UnimplementedQueryNodeServer) GetComponentStates(ctx context.Context, req *commonpb.Empty) (*ServiceStatesResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetComponentStates not implemented")
}
func (*UnimplementedQueryNodeServer) AddQueryChannel(ctx context.Context, req *AddQueryChannelsRequest) (*commonpb.Status, error) {
return nil, status.Errorf(codes.Unimplemented, "method AddQueryChannel not implemented")
}
@ -1953,14 +1984,65 @@ func (*UnimplementedQueryNodeServer) LoadSegments(ctx context.Context, req *Load
func (*UnimplementedQueryNodeServer) ReleaseSegments(ctx context.Context, req *ReleaseSegmentRequest) (*commonpb.Status, error) {
return nil, status.Errorf(codes.Unimplemented, "method ReleaseSegments not implemented")
}
func (*UnimplementedQueryNodeServer) GetPartitionState(ctx context.Context, req *PartitionStatesRequest) (*PartitionStatesResponse, error) {
return nil, status.Errorf(codes.Unimplemented, "method GetPartitionState not implemented")
}
func RegisterQueryNodeServer(s *grpc.Server, srv QueryNodeServer) {
s.RegisterService(&_QueryNode_serviceDesc, srv)
}
func _QueryNode_GetTimeTickChannel_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(commonpb.Empty)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(QueryNodeServer).GetTimeTickChannel(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/milvus.proto.query.QueryNode/GetTimeTickChannel",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(QueryNodeServer).GetTimeTickChannel(ctx, req.(*commonpb.Empty))
}
return interceptor(ctx, in, info, handler)
}
func _QueryNode_GetStatsChannel_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(commonpb.Empty)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(QueryNodeServer).GetStatsChannel(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/milvus.proto.query.QueryNode/GetStatsChannel",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(QueryNodeServer).GetStatsChannel(ctx, req.(*commonpb.Empty))
}
return interceptor(ctx, in, info, handler)
}
func _QueryNode_GetComponentStates_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(commonpb.Empty)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(QueryNodeServer).GetComponentStates(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/milvus.proto.query.QueryNode/GetComponentStates",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(QueryNodeServer).GetComponentStates(ctx, req.(*commonpb.Empty))
}
return interceptor(ctx, in, info, handler)
}
func _QueryNode_AddQueryChannel_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(AddQueryChannelsRequest)
if err := dec(in); err != nil {
@ -2051,28 +2133,22 @@ func _QueryNode_ReleaseSegments_Handler(srv interface{}, ctx context.Context, de
return interceptor(ctx, in, info, handler)
}
func _QueryNode_GetPartitionState_Handler(srv interface{}, ctx context.Context, dec func(interface{}) error, interceptor grpc.UnaryServerInterceptor) (interface{}, error) {
in := new(PartitionStatesRequest)
if err := dec(in); err != nil {
return nil, err
}
if interceptor == nil {
return srv.(QueryNodeServer).GetPartitionState(ctx, in)
}
info := &grpc.UnaryServerInfo{
Server: srv,
FullMethod: "/milvus.proto.query.QueryNode/GetPartitionState",
}
handler := func(ctx context.Context, req interface{}) (interface{}, error) {
return srv.(QueryNodeServer).GetPartitionState(ctx, req.(*PartitionStatesRequest))
}
return interceptor(ctx, in, info, handler)
}
var _QueryNode_serviceDesc = grpc.ServiceDesc{
ServiceName: "milvus.proto.query.QueryNode",
HandlerType: (*QueryNodeServer)(nil),
Methods: []grpc.MethodDesc{
{
MethodName: "GetTimeTickChannel",
Handler: _QueryNode_GetTimeTickChannel_Handler,
},
{
MethodName: "GetStatsChannel",
Handler: _QueryNode_GetStatsChannel_Handler,
},
{
MethodName: "GetComponentStates",
Handler: _QueryNode_GetComponentStates_Handler,
},
{
MethodName: "AddQueryChannel",
Handler: _QueryNode_AddQueryChannel_Handler,
@ -2093,10 +2169,6 @@ var _QueryNode_serviceDesc = grpc.ServiceDesc{
MethodName: "ReleaseSegments",
Handler: _QueryNode_ReleaseSegments_Handler,
},
{
MethodName: "GetPartitionState",
Handler: _QueryNode_GetPartitionState_Handler,
},
},
Streams: []grpc.StreamDesc{},
Metadata: "query_service.proto",

23
internal/querynode/query_node.go
View File

@ -19,10 +19,6 @@ import (
"io"
"sync/atomic"
"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
queryserviceimpl "github.com/zilliztech/milvus-distributed/internal/queryservice"
"github.com/opentracing/opentracing-go"
"github.com/uber/jaeger-client-go/config"
@ -31,11 +27,11 @@ import (
"github.com/zilliztech/milvus-distributed/internal/proto/commonpb"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
queryPb "github.com/zilliztech/milvus-distributed/internal/proto/querypb"
queryserviceimpl "github.com/zilliztech/milvus-distributed/internal/queryservice"
"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
)
type Node interface {
typeutil.Service
GetComponentStates() (*internalpb2.ComponentStates, error)
GetTimeTickChannel() (string, error)
GetStatisticsChannel() (string, error)
@ -45,10 +41,11 @@ type Node interface {
WatchDmChannels(in *queryPb.WatchDmChannelsRequest) (*commonpb.Status, error)
LoadSegments(in *queryPb.LoadSegmentRequest) (*commonpb.Status, error)
ReleaseSegments(in *queryPb.ReleaseSegmentRequest) (*commonpb.Status, error)
GetPartitionState(in *queryPb.PartitionStatesRequest) (*queryPb.PartitionStatesResponse, error)
}
type QueryNode struct {
typeutil.Service
queryNodeLoopCtx context.Context
queryNodeLoopCancel context.CancelFunc
@ -71,12 +68,7 @@ type QueryNode struct {
closer io.Closer
}
func NewQueryNode(ctx context.Context, queryNodeID uint64) Node {
var node Node = newQueryNode(ctx, queryNodeID)
return node
}
func newQueryNode(ctx context.Context, queryNodeID uint64) *QueryNode {
func NewQueryNode(ctx context.Context, queryNodeID uint64) *QueryNode {
ctx1, cancel := context.WithCancel(ctx)
node := &QueryNode{
queryNodeLoopCtx: ctx1,
@ -385,8 +377,3 @@ func (node *QueryNode) ReleaseSegments(in *queryPb.ReleaseSegmentRequest) (*comm
}
return nil, nil
}
func (node *QueryNode) GetPartitionState(in *queryPb.PartitionStatesRequest) (*queryPb.PartitionStatesResponse, error) {
// TODO: implement
return nil, nil
}

2
internal/querynode/query_node_test.go
View File

@ -130,7 +130,7 @@ func newQueryNodeMock() *QueryNode {
ctx = context.Background()
}
svr := newQueryNode(ctx, 0)
svr := NewQueryNode(ctx, 0)
return svr
}

2
internal/querynode/search_service_test.go
View File

@ -213,7 +213,7 @@ func TestSearch_Search(t *testing.T) {
}
func TestSearch_SearchMultiSegments(t *testing.T) {
node := newQueryNode(context.Background(), 0)
node := NewQueryNode(context.Background(), 0)
initTestMeta(t, node, "collection0", 0, 0)
pulsarURL := Params.PulsarAddress