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

Add query service module 

Signed-off-by: xige-16 <xi.ge@zilliz.com>
pull/4973/head^2
xige-16 2021-01-16 15:31:10 +08:00 committed by yefu.chen
parent 37688559f1
commit 0164e78bb5
10 changed files with 385 additions and 234 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

66
internal/distributed/queryservice/client.go
View File

@ -1,12 +1,70 @@
package queryservice
import (
"context"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
"github.com/zilliztech/milvus-distributed/internal/proto/querypb"
)
type Client struct {
ctx context.Context
querypb.QueryServiceClient
grpcClient querypb.QueryServiceClient
}
func (c *Client) Init() {
panic("implement me")
}
func (c *Client) Start() {
panic("implement me")
}
func (c *Client) Stop() {
panic("implement me")
}
func (c *Client) GetServiceStates() (internalpb2.ServiceStates, error) {
panic("implement me")
}
func (c *Client) GetTimeTickChannel() (string, error) {
panic("implement me")
}
func (c *Client) GetStatisticsChannel() (string, error) {
panic("implement me")
}
func (c *Client) RegisterNode(req querypb.RegisterNodeRequest) (querypb.RegisterNodeResponse, error) {
panic("implement me")
}
func (c *Client) ShowCollections(req querypb.ShowCollectionRequest) (querypb.ShowCollectionResponse, error) {
panic("implement me")
}
func (c *Client) LoadCollection(req querypb.LoadCollectionRequest) error {
panic("implement me")
}
func (c *Client) ReleaseCollection(req querypb.ReleaseCollectionRequest) error {
panic("implement me")
}
func (c *Client) ShowPartitions(req querypb.ShowPartitionRequest) (querypb.ShowPartitionResponse, error) {
panic("implement me")
}
func (c *Client) LoadPartitions(req querypb.LoadPartitionRequest) error {
panic("implement me")
}
func (c *Client) ReleasePartitions(req querypb.ReleasePartitionRequest) error {
panic("implement me")
}
func (c *Client) CreateQueryChannel() (querypb.CreateQueryChannelResponse, error) {
panic("implement me")
}
func (c *Client) GetPartitionStates(req querypb.PartitionStatesRequest) (querypb.PartitionStatesResponse, error) {
panic("implement me")
}

45
internal/distributed/queryservice/service.go
View File

@ -3,47 +3,72 @@ package queryservice
import (
"google.golang.org/grpc"
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
"github.com/zilliztech/milvus-distributed/internal/proto/querypb"
queryServiceImpl "github.com/zilliztech/milvus-distributed/internal/queryservice"
)
type Server struct {
grpcServer *grpc.Server
queryService queryServiceImpl.QueryService
queryService queryServiceImpl.Interface
}
func (s *Server) Init() {
panic("implement me")
}
func (s *Server) Start() {
panic("implement me")
}
func (s *Server) Stop() {
panic("implement me")
}
func (s *Server) GetServiceStates() (internalpb2.ServiceStates, error) {
panic("implement me")
}
func (s *Server) GetTimeTickChannel() (string, error) {
panic("implement me")
}
func (s *Server) GetStatisticsChannel() (string, error) {
panic("implement me")
}
func (s *Server) RegisterNode(req querypb.RegisterNodeRequest) (querypb.RegisterNodeResponse, error) {
return s.queryService.RegisterNode(req)
panic("implement me")
}
func (s *Server) ShowCollections(req querypb.ShowCollectionRequest) (querypb.ShowCollectionResponse, error) {
return s.ShowCollections(req)
panic("implement me")
}
func (s *Server) LoadCollection(req querypb.LoadCollectionRequest) error {
return s.LoadCollection(req)
panic("implement me")
}
func (s *Server) ReleaseCollection(req querypb.ReleaseCollectionRequest) error {
return s.ReleaseCollection(req)
panic("implement me")
}
func (s *Server) ShowPartitions(req querypb.ShowPartitionRequest) (querypb.ShowPartitionResponse, error) {
return s.ShowPartitions(req)
panic("implement me")
}
func (s *Server) GetPartitionStates(req querypb.PartitionStatesRequest) (querypb.PartitionStatesResponse, error) {
return s.GetPartitionStates(req)
panic("implement me")
}
func (s *Server) LoadPartitions(req querypb.LoadPartitionRequest) error {
return s.LoadPartitions(req)
panic("implement me")
}
func (s *Server) ReleasePartitions(req querypb.ReleasePartitionRequest) error {
return s.ReleasePartitions(req)
panic("implement me")
}
func (s *Server) CreateQueryChannel() (querypb.CreateQueryChannelResponse, error) {
return s.CreateQueryChannel()
panic("implement me")
}

55
internal/proto/query_service.proto
View File

@ -7,50 +7,49 @@ option go_package = "github.com/zilliztech/milvus-distributed/internal/proto/que
import "common.proto";
import "internal.proto";
message RegisterNodeRequest {
common.MsgBase base = 1;
common.Address address = 2;
}
message RegisterNodeResponse {
internal.InitParams init_params = 1;
common.Status status = 1;
internal.InitParams init_params = 2;
}
message ShowCollectionRequest {
common.MsgBase base = 1;
int64 dbID = 2;
int64 dbID = 2;
}
message ShowCollectionResponse {
repeated int64 collectionIDs = 1;
message ShowCollectionResponse {
common.Status status = 1;
repeated int64 collectionIDs = 2;
}
message LoadCollectionRequest {
common.MsgBase base = 1;
int64 dbID = 2;
int64 dbID = 2;
int64 collectionID = 3;
}
message ReleaseCollectionRequest {
common.MsgBase base = 1;
int64 dbID = 2;
int64 dbID = 2;
int64 collectionID = 3;
}
message ShowPartitionRequest {
common.MsgBase base = 1;
int64 dbID = 2;
int64 collectionID = 3;
int64 dbID = 2;
int64 collectionID = 3;
}
message ShowPartitionResponse {
repeated int64 partitionIDs = 1;
common.Status status = 1;
repeated int64 partitionIDs = 2;
}
enum PartitionState {
NotExist = 0;
NotPresent = 1;
@ -66,18 +65,18 @@ message PartitionStates {
PartitionState state = 2;
}
message PartitionStatesRequest {
message PartitionStatesRequest {
common.MsgBase base = 1;
int64 dbID = 2;
int64 dbID = 2;
int64 collectionID = 3;
repeated int64 partitionIDs = 4;
}
message PartitionStatesResponse {
repeated PartitionStates partition_descriptions = 1;
common.Status status = 1;
repeated PartitionStates partition_descriptions = 2;
}
message LoadPartitionRequest {
common.MsgBase base = 1;
int64 dbID = 2;
@ -93,22 +92,22 @@ message ReleasePartitionRequest {
}
message CreateQueryChannelResponse {
string request_channel = 1;
string result_channel = 2;
common.Status status = 1;
string request_channel = 2;
string result_channel = 3;
}
message AddQueryChannelsRequest {
common.MsgBase base = 1;
string request_channelID = 2;
string result_channelID = 3;
}
message RemoveQueryChannelsRequest {
common.MsgBase base = 1;
string request_channelID = 2;
string result_channelID = 3;
common.Status status = 1;
common.MsgBase base = 2;
string request_channelID = 3;
string result_channelID = 4;
}
message WatchDmChannelsRequest {
@ -116,7 +115,6 @@ message WatchDmChannelsRequest {
repeated string channelIDs = 2;
}
message LoadSegmentRequest {
common.MsgBase base = 1;
int64 dbID = 2;
@ -151,13 +149,6 @@ message ServiceStatesResponse {
}
service QueryService {
/**
* @brief This method is used to create collection
*
* @param CollectionSchema, use to provide collection information to be created.
*
* @return Status
*/
rpc RegisterNode(RegisterNodeRequest) returns (RegisterNodeResponse) {}
rpc ShowCollections(ShowCollectionRequest) returns (ShowCollectionResponse) {}

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

@ -114,7 +114,8 @@ func (m *RegisterNodeRequest) GetAddress() *commonpb.Address {
}
type RegisterNodeResponse struct {
InitParams *internalpb2.InitParams `protobuf:"bytes,1,opt,name=init_params,json=initParams,proto3" json:"init_params,omitempty"`
Status *commonpb.Status `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"`
InitParams *internalpb2.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:"-"`
@ -145,6 +146,13 @@ func (m *RegisterNodeResponse) XXX_DiscardUnknown() {
var xxx_messageInfo_RegisterNodeResponse proto.InternalMessageInfo
func (m *RegisterNodeResponse) GetStatus() *commonpb.Status {
if m != nil {
return m.Status
}
return nil
}
func (m *RegisterNodeResponse) GetInitParams() *internalpb2.InitParams {
if m != nil {
return m.InitParams
@ -200,10 +208,11 @@ func (m *ShowCollectionRequest) GetDbID() int64 {
}
type ShowCollectionResponse struct {
CollectionIDs []int64 `protobuf:"varint,1,rep,packed,name=collectionIDs,proto3" json:"collectionIDs,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
Status *commonpb.Status `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"`
CollectionIDs []int64 `protobuf:"varint,2,rep,packed,name=collectionIDs,proto3" json:"collectionIDs,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *ShowCollectionResponse) Reset() { *m = ShowCollectionResponse{} }
@ -231,6 +240,13 @@ func (m *ShowCollectionResponse) XXX_DiscardUnknown() {
var xxx_messageInfo_ShowCollectionResponse proto.InternalMessageInfo
func (m *ShowCollectionResponse) GetStatus() *commonpb.Status {
if m != nil {
return m.Status
}
return nil
}
func (m *ShowCollectionResponse) GetCollectionIDs() []int64 {
if m != nil {
return m.CollectionIDs
@ -404,10 +420,11 @@ func (m *ShowPartitionRequest) GetCollectionID() int64 {
}
type ShowPartitionResponse struct {
PartitionIDs []int64 `protobuf:"varint,1,rep,packed,name=partitionIDs,proto3" json:"partitionIDs,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
Status *commonpb.Status `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"`
PartitionIDs []int64 `protobuf:"varint,2,rep,packed,name=partitionIDs,proto3" json:"partitionIDs,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *ShowPartitionResponse) Reset() { *m = ShowPartitionResponse{} }
@ -435,6 +452,13 @@ func (m *ShowPartitionResponse) XXX_DiscardUnknown() {
var xxx_messageInfo_ShowPartitionResponse proto.InternalMessageInfo
func (m *ShowPartitionResponse) GetStatus() *commonpb.Status {
if m != nil {
return m.Status
}
return nil
}
func (m *ShowPartitionResponse) GetPartitionIDs() []int64 {
if m != nil {
return m.PartitionIDs
@ -553,7 +577,8 @@ func (m *PartitionStatesRequest) GetPartitionIDs() []int64 {
}
type PartitionStatesResponse struct {
PartitionDescriptions []*PartitionStates `protobuf:"bytes,1,rep,name=partition_descriptions,json=partitionDescriptions,proto3" json:"partition_descriptions,omitempty"`
Status *commonpb.Status `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"`
PartitionDescriptions []*PartitionStates `protobuf:"bytes,2,rep,name=partition_descriptions,json=partitionDescriptions,proto3" json:"partition_descriptions,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
@ -584,6 +609,13 @@ func (m *PartitionStatesResponse) XXX_DiscardUnknown() {
var xxx_messageInfo_PartitionStatesResponse proto.InternalMessageInfo
func (m *PartitionStatesResponse) GetStatus() *commonpb.Status {
if m != nil {
return m.Status
}
return nil
}
func (m *PartitionStatesResponse) GetPartitionDescriptions() []*PartitionStates {
if m != nil {
return m.PartitionDescriptions
@ -718,11 +750,12 @@ func (m *ReleasePartitionRequest) GetPartitionIDs() []int64 {
}
type CreateQueryChannelResponse struct {
RequestChannel string `protobuf:"bytes,1,opt,name=request_channel,json=requestChannel,proto3" json:"request_channel,omitempty"`
ResultChannel string `protobuf:"bytes,2,opt,name=result_channel,json=resultChannel,proto3" json:"result_channel,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
Status *commonpb.Status `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"`
RequestChannel string `protobuf:"bytes,2,opt,name=request_channel,json=requestChannel,proto3" json:"request_channel,omitempty"`
ResultChannel string `protobuf:"bytes,3,opt,name=result_channel,json=resultChannel,proto3" json:"result_channel,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
}
func (m *CreateQueryChannelResponse) Reset() { *m = CreateQueryChannelResponse{} }
@ -750,6 +783,13 @@ func (m *CreateQueryChannelResponse) XXX_DiscardUnknown() {
var xxx_messageInfo_CreateQueryChannelResponse proto.InternalMessageInfo
func (m *CreateQueryChannelResponse) GetStatus() *commonpb.Status {
if m != nil {
return m.Status
}
return nil
}
func (m *CreateQueryChannelResponse) GetRequestChannel() string {
if m != nil {
return m.RequestChannel
@ -820,9 +860,10 @@ func (m *AddQueryChannelsRequest) GetResultChannelID() string {
}
type RemoveQueryChannelsRequest struct {
Base *commonpb.MsgBase `protobuf:"bytes,1,opt,name=base,proto3" json:"base,omitempty"`
RequestChannelID string `protobuf:"bytes,2,opt,name=request_channelID,json=requestChannelID,proto3" json:"request_channelID,omitempty"`
ResultChannelID string `protobuf:"bytes,3,opt,name=result_channelID,json=resultChannelID,proto3" json:"result_channelID,omitempty"`
Status *commonpb.Status `protobuf:"bytes,1,opt,name=status,proto3" json:"status,omitempty"`
Base *commonpb.MsgBase `protobuf:"bytes,2,opt,name=base,proto3" json:"base,omitempty"`
RequestChannelID string `protobuf:"bytes,3,opt,name=request_channelID,json=requestChannelID,proto3" json:"request_channelID,omitempty"`
ResultChannelID string `protobuf:"bytes,4,opt,name=result_channelID,json=resultChannelID,proto3" json:"result_channelID,omitempty"`
XXX_NoUnkeyedLiteral struct{} `json:"-"`
XXX_unrecognized []byte `json:"-"`
XXX_sizecache int32 `json:"-"`
@ -853,6 +894,13 @@ func (m *RemoveQueryChannelsRequest) XXX_DiscardUnknown() {
var xxx_messageInfo_RemoveQueryChannelsRequest proto.InternalMessageInfo
func (m *RemoveQueryChannelsRequest) GetStatus() *commonpb.Status {
if m != nil {
return m.Status
}
return nil
}
func (m *RemoveQueryChannelsRequest) GetBase() *commonpb.MsgBase {
if m != nil {
return m.Base
@ -1249,79 +1297,80 @@ func init() {
func init() { proto.RegisterFile("query_service.proto", fileDescriptor_5fcb6756dc1afb8d) }
var fileDescriptor_5fcb6756dc1afb8d = []byte{
// 1137 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xdc, 0x57, 0xeb, 0x6e, 0x1b, 0x45,
0x14, 0xf6, 0xc6, 0x49, 0xa8, 0x4f, 0x1c, 0xdb, 0x99, 0x5c, 0xb5, 0x20, 0x54, 0x86, 0x42, 0x6e,
0xe0, 0xa0, 0x54, 0x42, 0x48, 0x48, 0xa0, 0x24, 0xae, 0x22, 0x4b, 0x34, 0x84, 0x4d, 0x2b, 0x44,
0xda, 0xca, 0xac, 0x77, 0x07, 0x7b, 0xda, 0xbd, 0xb8, 0x3b, 0xe3, 0x94, 0xe4, 0x0f, 0x54, 0xe2,
0x1d, 0x78, 0x00, 0x10, 0xbc, 0x01, 0xef, 0xc1, 0x0b, 0x20, 0xf1, 0x26, 0x68, 0x66, 0xd7, 0x9b,
0xdd, 0xf5, 0xac, 0xed, 0xb6, 0x94, 0x44, 0xfc, 0xdb, 0x39, 0xf3, 0xcd, 0xf9, 0xbe, 0x39, 0x73,
0xfb, 0x16, 0x16, 0x9f, 0xf6, 0x49, 0x70, 0xde, 0x62, 0x24, 0x38, 0xa3, 0x16, 0xa9, 0xf7, 0x02,
0x9f, 0xfb, 0x08, 0xb9, 0xd4, 0x39, 0xeb, 0xb3, 0xb0, 0x55, 0x97, 0x08, 0xbd, 0x6c, 0xf9, 0xae,
0xeb, 0x7b, 0x61, 0x4c, 0xaf, 0x50, 0x8f, 0x93, 0xc0, 0x33, 0x9d, 0xb0, 0x8d, 0x7f, 0x80, 0x45,
0x83, 0x74, 0x28, 0xe3, 0x24, 0x38, 0xf2, 0x6d, 0x62, 0x90, 0xa7, 0x7d, 0xc2, 0x38, 0xfa, 0x08,
0xa6, 0xdb, 0x26, 0x23, 0x6b, 0xda, 0x4d, 0x6d, 0x63, 0x6e, 0xf7, 0xad, 0x7a, 0x2a, 0x6f, 0x94,
0xf0, 0x2e, 0xeb, 0xec, 0x9b, 0x8c, 0x18, 0x12, 0x89, 0x3e, 0x86, 0x37, 0x4c, 0xdb, 0x0e, 0x08,
0x63, 0x6b, 0x53, 0x23, 0x06, 0xed, 0x85, 0x18, 0x63, 0x00, 0xc6, 0xa7, 0xb0, 0x94, 0x16, 0xc0,
0x7a, 0xbe, 0xc7, 0x08, 0xda, 0x87, 0x39, 0xea, 0x51, 0xde, 0xea, 0x99, 0x81, 0xe9, 0xb2, 0x48,
0xc8, 0x3b, 0xe9, 0x9c, 0xf1, 0x5c, 0x9a, 0x1e, 0xe5, 0xc7, 0x12, 0x68, 0x00, 0x8d, 0xbf, 0xf1,
0x23, 0x58, 0x3e, 0xe9, 0xfa, 0xcf, 0x0e, 0x7c, 0xc7, 0x21, 0x16, 0xa7, 0xbe, 0xf7, 0xf2, 0xd3,
0x43, 0x30, 0x6d, 0xb7, 0x9b, 0x0d, 0x39, 0xb7, 0xa2, 0x21, 0xbf, 0xf1, 0x67, 0xb0, 0x92, 0x4d,
0x1f, 0x89, 0xbf, 0x05, 0xf3, 0x56, 0x1c, 0x6d, 0x36, 0x84, 0xfc, 0xe2, 0x46, 0xd1, 0x48, 0x07,
0xf1, 0x73, 0x0d, 0x96, 0xbf, 0xf0, 0x4d, 0xfb, 0x35, 0xe9, 0x43, 0x18, 0xca, 0x49, 0xc2, 0xb5,
0xa2, 0xec, 0x4b, 0xc5, 0xf0, 0x4f, 0x1a, 0xac, 0x19, 0xc4, 0x21, 0x26, 0x23, 0x57, 0x29, 0xe3,
0x47, 0x0d, 0x96, 0x44, 0x2d, 0x8f, 0xcd, 0x80, 0xd3, 0xab, 0x91, 0xf0, 0x69, 0xb8, 0x59, 0x12,
0x0a, 0xa2, 0xc5, 0xc4, 0x50, 0xee, 0x0d, 0x82, 0x97, 0x6b, 0x99, 0x8a, 0x61, 0x17, 0xaa, 0xf1,
0xc0, 0x13, 0x6e, 0x72, 0xc2, 0xd0, 0x4d, 0x98, 0x4b, 0x40, 0xe4, 0x04, 0x8a, 0x46, 0x32, 0x84,
0x3e, 0x81, 0x19, 0x26, 0xb0, 0x52, 0x6a, 0x65, 0x17, 0xd7, 0x87, 0x4f, 0x6f, 0x3d, 0x9d, 0xd5,
0x08, 0x07, 0xe0, 0xdf, 0x34, 0x58, 0xc9, 0xf0, 0xfd, 0xe7, 0x05, 0x1b, 0xaa, 0xcb, 0xb4, 0xa2,
0x2e, 0x7d, 0x58, 0x1d, 0xd2, 0x19, 0x95, 0xf5, 0x14, 0x56, 0x62, 0x68, 0xcb, 0x26, 0xcc, 0x0a,
0x68, 0x4f, 0x7c, 0x87, 0x05, 0x9e, 0xdb, 0x7d, 0x77, 0x7c, 0x39, 0x98, 0xb1, 0x1c, 0xa7, 0x68,
0x24, 0x32, 0xe0, 0x5f, 0x35, 0x58, 0x12, 0x27, 0xeb, 0xea, 0xb6, 0xd3, 0x44, 0xd5, 0xf9, 0x5d,
0x83, 0xd5, 0xe8, 0xf0, 0x5d, 0x73, 0xa5, 0x0e, 0xe8, 0x07, 0x01, 0x31, 0x39, 0xf9, 0x4a, 0x2c,
0xc3, 0x41, 0xd7, 0xf4, 0x3c, 0xe2, 0xc4, 0x4b, 0xb9, 0x0e, 0xd5, 0x20, 0x94, 0xdd, 0xb2, 0xc2,
0x2e, 0x29, 0xbb, 0x64, 0x54, 0xa2, 0x70, 0x34, 0x00, 0xbd, 0x07, 0x95, 0x80, 0xb0, 0xbe, 0x73,
0x89, 0x9b, 0x92, 0xb8, 0xf9, 0x30, 0x1a, 0xc1, 0xf0, 0x2f, 0x1a, 0xac, 0xee, 0xd9, 0x76, 0x92,
0xeb, 0x15, 0xf6, 0xf7, 0x36, 0x2c, 0x64, 0xd4, 0x45, 0x45, 0x2a, 0x19, 0xb5, 0xb4, 0xbe, 0x66,
0x03, 0x6d, 0x42, 0x2d, 0xad, 0x30, 0x2a, 0x5a, 0xc9, 0xa8, 0xa6, 0x34, 0x36, 0x1b, 0xe2, 0x10,
0xea, 0x06, 0x71, 0xfd, 0x33, 0x72, 0xcd, 0x85, 0x3e, 0x86, 0x95, 0xaf, 0x4d, 0x6e, 0x75, 0x1b,
0xee, 0xab, 0x6b, 0x7c, 0x1b, 0x20, 0xe6, 0x13, 0x2f, 0x7d, 0x71, 0xa3, 0x64, 0x24, 0x22, 0xf8,
0x2f, 0x0d, 0x90, 0x38, 0x79, 0x27, 0xa4, 0xe3, 0x12, 0x8f, 0x5f, 0xcb, 0xdd, 0x2c, 0x26, 0xc1,
0x42, 0x7d, 0x02, 0x31, 0x23, 0x11, 0x89, 0x08, 0xd2, 0xe1, 0xc6, 0x77, 0x94, 0x38, 0xb6, 0xe8,
0x9d, 0x95, 0xbd, 0x71, 0x1b, 0xff, 0xad, 0xc1, 0x72, 0x74, 0x66, 0xff, 0xb7, 0x73, 0x7c, 0xae,
0x81, 0x7e, 0x48, 0xf8, 0x3d, 0xea, 0x92, 0x7b, 0xd4, 0x7a, 0x92, 0x3d, 0xee, 0xb7, 0x61, 0x56,
0x3c, 0x43, 0xfd, 0x81, 0x2b, 0x7b, 0x53, 0x39, 0xd5, 0x13, 0x09, 0x31, 0x22, 0x28, 0xaa, 0xc3,
0x22, 0xa7, 0x2e, 0x69, 0x71, 0x6a, 0x3d, 0x19, 0xda, 0xde, 0x0b, 0x3c, 0x4d, 0xd5, 0x6c, 0xe0,
0x67, 0xb0, 0x7a, 0x48, 0xb8, 0x48, 0xc2, 0xfe, 0x15, 0xfe, 0x75, 0xa8, 0x8a, 0x2f, 0x36, 0xc4,
0x5d, 0x61, 0x09, 0x8e, 0x66, 0x03, 0xff, 0xac, 0xc1, 0xf2, 0x49, 0xe8, 0xaa, 0x33, 0x2f, 0xd6,
0x4b, 0xf1, 0x36, 0x61, 0x5e, 0x78, 0x74, 0x12, 0xb4, 0xe4, 0xd3, 0x3d, 0x70, 0xc7, 0xb7, 0x72,
0x9c, 0x6c, 0x9a, 0xb9, 0x1c, 0x0e, 0x0d, 0x5b, 0x5b, 0x17, 0x50, 0x49, 0xbf, 0x7f, 0xa8, 0x0c,
0x37, 0x8e, 0x7c, 0x7e, 0xe7, 0x7b, 0xca, 0x78, 0xad, 0x80, 0x2a, 0x00, 0x47, 0x3e, 0x3f, 0x0e,
0x08, 0x23, 0x1e, 0xaf, 0x69, 0x08, 0x60, 0xf6, 0x4b, 0xaf, 0x21, 0xfa, 0xa6, 0xd0, 0x62, 0x64,
0x50, 0x4c, 0xa7, 0xe9, 0xdd, 0x25, 0xae, 0x1f, 0x9c, 0xd7, 0x8a, 0x62, 0x78, 0xdc, 0x9a, 0x46,
0x35, 0x28, 0xc7, 0x90, 0xc3, 0xe3, 0xfb, 0xb5, 0x19, 0x54, 0x82, 0x99, 0xf0, 0x73, 0x76, 0xf7,
0x8f, 0x12, 0x94, 0xe5, 0x35, 0x17, 0x09, 0x44, 0x16, 0x94, 0x93, 0xbe, 0x1d, 0xad, 0xab, 0x9e,
0x6b, 0xc5, 0xaf, 0x85, 0xbe, 0x31, 0x1e, 0x18, 0xd6, 0x1b, 0x17, 0xd0, 0x63, 0xa8, 0xa6, 0x1d,
0x36, 0x43, 0x9b, 0xaa, 0xe1, 0x4a, 0x97, 0xaf, 0x6f, 0x4d, 0x02, 0x8d, 0xb9, 0x3a, 0x50, 0x49,
0xf9, 0x3f, 0x86, 0x36, 0xf2, 0xc6, 0x67, 0x1f, 0x6b, 0x7d, 0x73, 0x02, 0x64, 0x4c, 0xf4, 0x0d,
0x54, 0x52, 0xde, 0x24, 0x87, 0x48, 0xe5, 0x5f, 0xf4, 0x51, 0x5b, 0x0e, 0x17, 0x50, 0x0b, 0x16,
0xb2, 0x7e, 0x82, 0xa1, 0x6d, 0x75, 0xc1, 0x95, 0xb6, 0x63, 0x1c, 0xc1, 0x69, 0xa8, 0xfd, 0xb2,
0x80, 0xea, 0xf5, 0x50, 0xfe, 0xd5, 0x8c, 0xcb, 0xfd, 0x6d, 0x2c, 0x3e, 0x91, 0xfe, 0x83, 0x11,
0xe2, 0x5f, 0x98, 0xa1, 0x0d, 0x68, 0xd8, 0xc5, 0x20, 0x5d, 0x39, 0xe8, 0x8e, 0xdb, 0xe3, 0xe7,
0x7a, 0x5d, 0x45, 0x9f, 0xef, 0x84, 0x42, 0x8e, 0xe1, 0xab, 0xf3, 0xc5, 0x39, 0xf2, 0xaf, 0x5f,
0x5c, 0x40, 0x0f, 0xa1, 0x9a, 0xb9, 0x1b, 0x47, 0x12, 0x6c, 0xe7, 0x10, 0xa8, 0x2e, 0x57, 0x5c,
0x40, 0xbe, 0x9c, 0x41, 0xf6, 0x77, 0x66, 0x6b, 0x12, 0x3b, 0x1e, 0x2d, 0xc3, 0xf6, 0x44, 0xd8,
0x98, 0xf0, 0x01, 0xd4, 0x84, 0x9a, 0xe4, 0xcd, 0x37, 0x72, 0x3e, 0xea, 0xd3, 0xa6, 0xba, 0xb2,
0x71, 0x61, 0xf7, 0xcf, 0x69, 0x28, 0xc9, 0xa5, 0x92, 0xb7, 0xd4, 0x23, 0xa8, 0x66, 0x8c, 0xa5,
0xfa, 0x78, 0xe4, 0xb8, 0xcf, 0x71, 0x1b, 0xcc, 0x02, 0x34, 0xec, 0x08, 0x51, 0x5d, 0xbd, 0x87,
0xf3, 0x9c, 0xe3, 0x38, 0x92, 0x87, 0x50, 0xcd, 0xd8, 0x39, 0xf5, 0xe2, 0xa8, 0x3d, 0xdf, 0xb8,
0xec, 0xf7, 0xa1, 0x9c, 0xf0, 0x6f, 0x0c, 0xbd, 0x9f, 0x77, 0xbe, 0xd3, 0xee, 0x67, 0x5c, 0xda,
0x07, 0x50, 0x4d, 0xbb, 0xa6, 0x9c, 0x9b, 0x5c, 0x69, 0xad, 0xc6, 0x25, 0xf7, 0x60, 0x61, 0x68,
0xc7, 0xbe, 0xc6, 0x0d, 0xbb, 0xbf, 0x77, 0xfa, 0x79, 0x87, 0xf2, 0x6e, 0xbf, 0x2d, 0x94, 0xec,
0x5c, 0x50, 0xc7, 0xa1, 0x17, 0x9c, 0x58, 0xdd, 0x9d, 0x30, 0xcb, 0x87, 0x36, 0x65, 0x3c, 0xa0,
0xed, 0x3e, 0x27, 0xf6, 0xce, 0xe0, 0x65, 0xdf, 0x91, 0xa9, 0x77, 0x64, 0xea, 0x5e, 0xbb, 0x3d,
0x2b, 0x9b, 0xb7, 0xff, 0x09, 0x00, 0x00, 0xff, 0xff, 0x84, 0x5b, 0x26, 0x0a, 0xc7, 0x13, 0x00,
0x00,
// 1158 bytes of a gzipped FileDescriptorProto
0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xdc, 0x57, 0x5d, 0x6f, 0x1b, 0x45,
0x17, 0xf6, 0xc6, 0x4e, 0xde, 0xfa, 0xc4, 0xb1, 0x9d, 0xc9, 0xa7, 0xf6, 0x45, 0xa8, 0x0c, 0x85,
0xe6, 0x03, 0x1c, 0x94, 0x4a, 0x88, 0x3b, 0x94, 0xc4, 0x55, 0x64, 0x89, 0x86, 0xb0, 0x69, 0x85,
0x48, 0x5b, 0x99, 0xf5, 0xee, 0x10, 0x4f, 0xbb, 0x1f, 0xee, 0xce, 0x38, 0x25, 0xb9, 0x81, 0x4a,
0xdc, 0x73, 0xc9, 0x35, 0x02, 0x81, 0xc4, 0x0f, 0xe0, 0x7f, 0x70, 0x8d, 0x84, 0xc4, 0x3f, 0x41,
0x3b, 0xb3, 0xde, 0xec, 0xda, 0xb3, 0xb6, 0x53, 0xa7, 0xa4, 0xe2, 0x6e, 0x66, 0xf6, 0xcc, 0x79,
0x9e, 0x73, 0xe6, 0xcc, 0xd9, 0x67, 0x60, 0xe1, 0x59, 0x97, 0x04, 0x67, 0x4d, 0x46, 0x82, 0x53,
0x6a, 0x91, 0x5a, 0x27, 0xf0, 0xb9, 0x8f, 0x90, 0x4b, 0x9d, 0xd3, 0x2e, 0x93, 0xb3, 0x9a, 0xb0,
0xd0, 0x4b, 0x96, 0xef, 0xba, 0xbe, 0x27, 0xd7, 0xf4, 0x32, 0xf5, 0x38, 0x09, 0x3c, 0xd3, 0x91,
0x73, 0xfc, 0x0d, 0x2c, 0x18, 0xe4, 0x84, 0x32, 0x4e, 0x82, 0x03, 0xdf, 0x26, 0x06, 0x79, 0xd6,
0x25, 0x8c, 0xa3, 0x0f, 0xa0, 0xd0, 0x32, 0x19, 0x59, 0xd5, 0x6e, 0x6a, 0x6b, 0xb3, 0xdb, 0x6f,
0xd4, 0x52, 0x7e, 0x23, 0x87, 0xf7, 0xd8, 0xc9, 0xae, 0xc9, 0x88, 0x21, 0x2c, 0xd1, 0x87, 0xf0,
0x3f, 0xd3, 0xb6, 0x03, 0xc2, 0xd8, 0xea, 0xd4, 0x90, 0x4d, 0x3b, 0xd2, 0xc6, 0xe8, 0x19, 0xe3,
0xef, 0x35, 0x58, 0x4c, 0x33, 0x60, 0x1d, 0xdf, 0x63, 0x04, 0xdd, 0x81, 0x19, 0xc6, 0x4d, 0xde,
0x65, 0x11, 0x89, 0xff, 0x2b, 0xfd, 0x1d, 0x09, 0x13, 0x23, 0x32, 0x45, 0xbb, 0x30, 0x4b, 0x3d,
0xca, 0x9b, 0x1d, 0x33, 0x30, 0xdd, 0x1e, 0x93, 0xb7, 0xd2, 0x3b, 0xe3, 0x0c, 0x34, 0x3c, 0xca,
0x0f, 0x85, 0xa1, 0x01, 0x34, 0x1e, 0xe3, 0xc7, 0xb0, 0x74, 0xd4, 0xf6, 0x9f, 0xef, 0xf9, 0x8e,
0x43, 0x2c, 0x4e, 0x7d, 0xef, 0xe5, 0x93, 0x82, 0xa0, 0x60, 0xb7, 0x1a, 0x75, 0xc1, 0x23, 0x6f,
0x88, 0x31, 0x66, 0xb0, 0xdc, 0xef, 0x7e, 0x92, 0x88, 0x6f, 0xc1, 0x9c, 0x15, 0xbb, 0x6a, 0xd4,
0xc3, 0x98, 0xf3, 0x6b, 0x79, 0x23, 0xbd, 0x88, 0x5f, 0x68, 0xb0, 0xf4, 0x89, 0x6f, 0xda, 0xaf,
0x28, 0x28, 0x84, 0xa1, 0x94, 0x04, 0x5c, 0xcd, 0x8b, 0x6f, 0xa9, 0x35, 0xfc, 0x9d, 0x06, 0xab,
0x06, 0x71, 0x88, 0xc9, 0xc8, 0x75, 0xd2, 0xf8, 0x56, 0x83, 0xc5, 0xf0, 0x00, 0x0e, 0xcd, 0x80,
0xd3, 0xeb, 0xa1, 0xd0, 0x91, 0x15, 0x96, 0x60, 0x30, 0x49, 0x05, 0x60, 0x28, 0x75, 0x7a, 0x9e,
0x2e, 0x0a, 0x20, 0xb5, 0x86, 0x5d, 0xa8, 0xc4, 0x68, 0xe1, 0x76, 0xc2, 0xd0, 0x4d, 0x98, 0x4d,
0x98, 0x08, 0xc0, 0xbc, 0x91, 0x5c, 0x42, 0x1f, 0xc1, 0x74, 0x08, 0x41, 0x44, 0x7c, 0xe5, 0x6d,
0x5c, 0x1b, 0xec, 0x2e, 0xb5, 0xb4, 0x57, 0x43, 0x6e, 0xc0, 0xbf, 0x68, 0xb0, 0xdc, 0x87, 0xf7,
0xaf, 0x67, 0x79, 0x20, 0x2f, 0x05, 0x45, 0x5e, 0x7e, 0xd3, 0x60, 0x65, 0x80, 0xe8, 0x24, 0x87,
0x71, 0x0c, 0xcb, 0x31, 0x40, 0xd3, 0x26, 0xcc, 0x0a, 0x68, 0x27, 0x1c, 0xcb, 0x63, 0x99, 0xdd,
0x7e, 0x7b, 0x74, 0x12, 0x99, 0xb1, 0x14, 0xbb, 0xa8, 0x27, 0x3c, 0xe0, 0x9f, 0x35, 0x58, 0x0c,
0x2f, 0xf1, 0xf5, 0x55, 0xee, 0x58, 0x39, 0xfd, 0x55, 0x83, 0x95, 0xe8, 0x9e, 0xbf, 0xe6, 0x4c,
0x7f, 0xd4, 0x40, 0xdf, 0x0b, 0x88, 0xc9, 0xc9, 0x67, 0xe1, 0x39, 0xec, 0xb5, 0x4d, 0xcf, 0x23,
0xce, 0x64, 0x05, 0x70, 0x1b, 0x2a, 0x81, 0x0c, 0xb6, 0x69, 0x49, 0x7f, 0x82, 0x7a, 0xd1, 0x28,
0x47, 0xcb, 0x11, 0x0a, 0x7a, 0x07, 0xca, 0x01, 0x61, 0x5d, 0xe7, 0xc2, 0x2e, 0x2f, 0xec, 0xe6,
0xe4, 0x6a, 0x64, 0x86, 0x7f, 0xd2, 0x60, 0x65, 0xc7, 0xb6, 0x93, 0x04, 0x27, 0xb8, 0x4b, 0x9b,
0x30, 0xdf, 0xc7, 0x2e, 0x4a, 0x6d, 0xd1, 0xa8, 0xa6, 0xf9, 0x35, 0xea, 0x68, 0x1d, 0xaa, 0x69,
0x86, 0x51, 0xaa, 0x8b, 0x46, 0x25, 0xc5, 0xb1, 0x51, 0xc7, 0x7f, 0x6a, 0xa0, 0x1b, 0xc4, 0xf5,
0x4f, 0x89, 0x92, 0xe8, 0x4b, 0x65, 0xb2, 0x17, 0xdd, 0xd4, 0x64, 0xd1, 0xe5, 0x2f, 0x11, 0x5d,
0x41, 0x1d, 0xdd, 0x13, 0x58, 0xfe, 0xdc, 0xe4, 0x56, 0xbb, 0xee, 0x4e, 0x7e, 0x02, 0x6f, 0x02,
0xc4, 0x78, 0xb2, 0x29, 0x14, 0x8d, 0xc4, 0x0a, 0xfe, 0x4b, 0x03, 0x14, 0x5e, 0xf2, 0x23, 0x72,
0xe2, 0x12, 0x8f, 0xbf, 0x96, 0x17, 0x27, 0x0c, 0x82, 0x49, 0x7e, 0xa1, 0xc5, 0xb4, 0xb0, 0x48,
0xac, 0x20, 0x1d, 0x6e, 0x7c, 0x45, 0x89, 0x63, 0x87, 0x5f, 0x67, 0xc4, 0xd7, 0x78, 0x8e, 0xff,
0xd6, 0x60, 0x29, 0x6a, 0x0f, 0xff, 0xd9, 0x18, 0x5f, 0x68, 0xa0, 0xef, 0x13, 0x7e, 0x9f, 0xba,
0xe4, 0x3e, 0xb5, 0x9e, 0x5e, 0x49, 0x63, 0xa9, 0xc1, 0x02, 0xa7, 0x2e, 0x69, 0x72, 0x6a, 0x3d,
0x1d, 0xb8, 0xbc, 0xf3, 0x3c, 0x0d, 0xd5, 0xa8, 0xe3, 0xe7, 0xb0, 0xb2, 0x4f, 0x78, 0xe8, 0x84,
0x5d, 0x55, 0x63, 0x0b, 0x47, 0x6c, 0x00, 0xbb, 0xcc, 0x12, 0x18, 0x8d, 0x3a, 0xfe, 0x41, 0x83,
0xa5, 0x23, 0xf9, 0x2c, 0xb9, 0x8a, 0x3f, 0x6a, 0x03, 0xe6, 0xc2, 0x47, 0x0e, 0x09, 0x9a, 0x42,
0x5b, 0xf4, 0x44, 0xfd, 0xad, 0x0c, 0x51, 0x9f, 0x46, 0x2e, 0xc9, 0xad, 0x72, 0xb6, 0x71, 0x0e,
0xe5, 0xf4, 0xaf, 0x16, 0x95, 0xe0, 0xc6, 0x81, 0xcf, 0xef, 0x7e, 0x4d, 0x19, 0xaf, 0xe6, 0x50,
0x19, 0xe0, 0xc0, 0xe7, 0x87, 0x01, 0x61, 0xc4, 0xe3, 0x55, 0x0d, 0x01, 0xcc, 0x7c, 0xea, 0xd5,
0xc3, 0x6f, 0x53, 0x68, 0x21, 0x52, 0x50, 0xa6, 0xd3, 0xf0, 0xee, 0x11, 0xd7, 0x0f, 0xce, 0xaa,
0xf9, 0x70, 0x7b, 0x3c, 0x2b, 0xa0, 0x2a, 0x94, 0x62, 0x93, 0xfd, 0xc3, 0x07, 0xd5, 0x69, 0x54,
0x84, 0x69, 0x39, 0x9c, 0xd9, 0xfe, 0xbd, 0x08, 0x25, 0xd1, 0x1b, 0x23, 0x82, 0xc8, 0x82, 0x52,
0xf2, 0xdd, 0x83, 0x6e, 0xab, 0x94, 0x81, 0xe2, 0x6d, 0xa6, 0xaf, 0x8d, 0x36, 0x94, 0xf9, 0xc6,
0x39, 0xf4, 0x04, 0x2a, 0xe9, 0xc7, 0x06, 0x43, 0xeb, 0xaa, 0xed, 0xca, 0x07, 0x8f, 0xbe, 0x31,
0x8e, 0x69, 0x8c, 0x75, 0x02, 0xe5, 0x94, 0xaa, 0x65, 0x68, 0x2d, 0x6b, 0x7f, 0xbf, 0x2e, 0xd0,
0xd7, 0xc7, 0xb0, 0x8c, 0x81, 0xbe, 0x80, 0x72, 0x4a, 0x06, 0x65, 0x00, 0xa9, 0xa4, 0x92, 0x3e,
0xac, 0xe4, 0x70, 0x0e, 0x35, 0x61, 0xbe, 0x5f, 0xba, 0x30, 0xb4, 0xa9, 0x4e, 0xb8, 0x52, 0xe1,
0x8c, 0x02, 0x38, 0x96, 0xdc, 0x2f, 0x12, 0xa8, 0x3e, 0x0f, 0xe5, 0x5b, 0x6d, 0x94, 0xef, 0x2f,
0x63, 0xf2, 0x09, 0xf7, 0xef, 0x0d, 0x21, 0x7f, 0x69, 0x84, 0x16, 0xa0, 0x41, 0xbd, 0x84, 0x74,
0xe5, 0xa6, 0xbb, 0x6e, 0x87, 0x9f, 0xe9, 0x35, 0x15, 0x7c, 0xb6, 0xe6, 0x92, 0x18, 0x83, 0xad,
0xf3, 0xf2, 0x18, 0xd9, 0xed, 0x17, 0xe7, 0xd0, 0x23, 0xa8, 0xf4, 0xf5, 0xc6, 0xa1, 0x00, 0x9b,
0x19, 0x00, 0xaa, 0xe6, 0x8a, 0x73, 0xc8, 0x17, 0x11, 0xf4, 0xbf, 0xb7, 0x36, 0xc6, 0x51, 0xfe,
0xd1, 0x31, 0x6c, 0x8e, 0x65, 0x1b, 0x03, 0x3e, 0x84, 0x6a, 0xc8, 0x26, 0xd9, 0xf9, 0x86, 0xc6,
0xa3, 0xbe, 0x6d, 0xaa, 0x96, 0x8d, 0x73, 0xdb, 0x7f, 0x14, 0xa0, 0x28, 0x8e, 0x4a, 0x74, 0xa9,
0xc7, 0x50, 0xe9, 0x53, 0xa3, 0xea, 0xeb, 0x91, 0x21, 0x59, 0x47, 0x15, 0x98, 0x05, 0x68, 0x50,
0x46, 0xa2, 0x9a, 0xba, 0x86, 0xb3, 0xe4, 0xe6, 0x28, 0x90, 0x47, 0x50, 0xe9, 0x93, 0x73, 0xea,
0xc3, 0x51, 0x6b, 0xbe, 0x51, 0xde, 0x1f, 0x40, 0x29, 0xa1, 0xdf, 0x18, 0x7a, 0x37, 0xeb, 0x7e,
0xa7, 0xd5, 0xcf, 0x28, 0xb7, 0x0f, 0xa1, 0x92, 0x56, 0x4d, 0x19, 0x9d, 0x5c, 0x29, 0xad, 0x46,
0x39, 0xf7, 0x60, 0x7e, 0xa0, 0x62, 0x5f, 0x61, 0xc1, 0xee, 0xee, 0x1c, 0x7f, 0x7c, 0x42, 0x79,
0xbb, 0xdb, 0x0a, 0x99, 0x6c, 0x9d, 0x53, 0xc7, 0xa1, 0xe7, 0x9c, 0x58, 0xed, 0x2d, 0xe9, 0xe5,
0x7d, 0x9b, 0x32, 0x1e, 0xd0, 0x56, 0x97, 0x13, 0x7b, 0xab, 0xf7, 0x67, 0xdf, 0x12, 0xae, 0xb7,
0x84, 0xeb, 0x4e, 0xab, 0x35, 0x23, 0xa6, 0x77, 0xfe, 0x09, 0x00, 0x00, 0xff, 0xff, 0x32, 0x8c,
0x07, 0xbf, 0x08, 0x15, 0x00, 0x00,
}
// Reference imports to suppress errors if they are not otherwise used.
@ -1336,12 +1385,6 @@ const _ = grpc.SupportPackageIsVersion4
//
// For semantics around ctx use and closing/ending streaming RPCs, please refer to https://godoc.org/google.golang.org/grpc#ClientConn.NewStream.
type QueryServiceClient interface {
//*
// @brief This method is used to create collection
//
// @param CollectionSchema, use to provide collection information to be created.
//
// @return Status
RegisterNode(ctx context.Context, in *RegisterNodeRequest, opts ...grpc.CallOption) (*RegisterNodeResponse, error)
ShowCollections(ctx context.Context, in *ShowCollectionRequest, opts ...grpc.CallOption) (*ShowCollectionResponse, error)
ShowPartitions(ctx context.Context, in *ShowPartitionRequest, opts ...grpc.CallOption) (*ShowPartitionResponse, error)
@ -1474,12 +1517,6 @@ func (c *queryServiceClient) GetServiceStates(ctx context.Context, in *commonpb.
// QueryServiceServer is the server API for QueryService service.
type QueryServiceServer interface {
//*
// @brief This method is used to create collection
//
// @param CollectionSchema, use to provide collection information to be created.
//
// @return Status
RegisterNode(context.Context, *RegisterNodeRequest) (*RegisterNodeResponse, error)
ShowCollections(context.Context, *ShowCollectionRequest) (*ShowCollectionResponse, error)
ShowPartitions(context.Context, *ShowPartitionRequest) (*ShowPartitionResponse, error)

14
internal/queryservice/interface.go
View File

@ -1,18 +1,22 @@
package queryserviceimpl
import "github.com/zilliztech/milvus-distributed/internal/proto/querypb"
import (
"github.com/zilliztech/milvus-distributed/internal/proto/querypb"
"github.com/zilliztech/milvus-distributed/internal/util/typeutil"
)
type ServiceBase = typeutil.Service
type Interface interface {
RegisterNode(req querypb.RegisterNodeRequest) (querypb.RegisterNodeResponse, error)
ServiceBase
RegisterNode(req querypb.RegisterNodeRequest) (querypb.RegisterNodeResponse, error)
ShowCollections(req querypb.ShowCollectionRequest) (querypb.ShowCollectionResponse, error)
LoadCollection(req querypb.LoadCollectionRequest) error
ReleaseCollection(req querypb.ReleaseCollectionRequest) error
ShowPartitions(req querypb.ShowPartitionRequest) (querypb.ShowPartitionResponse, error)
GetPartitionStates(req querypb.PartitionStatesRequest) (querypb.PartitionStatesResponse, error)
LoadPartitions(req querypb.LoadPartitionRequest) error
ReleasePartitions(req querypb.ReleasePartitionRequest) error
CreateQueryChannel() (querypb.CreateQueryChannelResponse, error)
GetPartitionStates(req querypb.PartitionStatesRequest) (querypb.PartitionStatesResponse, error)
}

55
internal/queryservice/query_service.go → internal/queryservice/queryservice.go
View File

@ -1,42 +1,71 @@
package queryserviceimpl
import "github.com/zilliztech/milvus-distributed/internal/proto/querypb"
import (
"github.com/zilliztech/milvus-distributed/internal/proto/internalpb2"
"github.com/zilliztech/milvus-distributed/internal/proto/querypb"
)
type QueryService struct {
}
//serverBase interface
func (qs *QueryService) Init() {
panic("implement me")
}
func (qs *QueryService) Start() {
panic("implement me")
}
func (qs *QueryService) Stop() {
panic("implement me")
}
func (qs *QueryService) GetServiceStates() (internalpb2.ServiceStates, error) {
panic("implement me")
}
func (qs *QueryService) GetTimeTickChannel() (string, error) {
panic("implement me")
}
func (qs *QueryService) GetStatisticsChannel() (string, error) {
panic("implement me")
}
//queryService interface
func (qs *QueryService) RegisterNode(req querypb.RegisterNodeRequest) (querypb.RegisterNodeResponse, error) {
return querypb.RegisterNodeResponse{}, nil
panic("implement me")
}
func (qs *QueryService) ShowCollections(req querypb.ShowCollectionRequest) (querypb.ShowCollectionResponse, error) {
return querypb.ShowCollectionResponse{}, nil
panic("implement me")
}
func (qs *QueryService) LoadCollection(req querypb.LoadCollectionRequest) error {
return nil
panic("implement me")
}
func (qs *QueryService) ReleaseCollection(req querypb.ReleaseCollectionRequest) error {
return nil
panic("implement me")
}
func (qs *QueryService) ShowPartitions(req querypb.ShowPartitionRequest) (querypb.ShowPartitionResponse, error) {
return querypb.ShowPartitionResponse{}, nil
}
func (qs *QueryService) GetPartitionStates(req querypb.PartitionStatesRequest) (querypb.PartitionStatesResponse, error) {
return querypb.PartitionStatesResponse{}, nil
panic("implement me")
}
func (qs *QueryService) LoadPartitions(req querypb.LoadPartitionRequest) error {
return nil
panic("implement me")
}
func (qs *QueryService) ReleasePartitions(req querypb.ReleasePartitionRequest) error {
return nil
panic("implement me")
}
func (qs *QueryService) CreateQueryChannel() (querypb.CreateQueryChannelResponse, error) {
return querypb.CreateQueryChannelResponse{}, nil
panic("implement me")
}
func (qs *QueryService) GetPartitionStates(req querypb.PartitionStatesRequest) (querypb.PartitionStatesResponse, error) {
panic("implement me")
}

6
internal/writenode/data_sync_service_test.go
View File

@ -130,7 +130,7 @@ func TestDataSyncService_Start(t *testing.T) {
InsertRequest: internalPb.InsertRequest{
MsgType: commonpb.MsgType_kInsert,
ReqID: UniqueID(0),
CollectionName: "col1",
CollectionName: "coll1",
PartitionTag: "default",
SegmentID: UniqueID(1),
ChannelID: UniqueID(0),
@ -206,7 +206,7 @@ func TestDataSyncService_Start(t *testing.T) {
<-ctx.Done()
}
func newMeta() *etcdpb.CollectionMeta {
func newMeta() {
ETCDAddr := Params.EtcdAddress
MetaRootPath := Params.MetaRootPath
@ -375,6 +375,4 @@ func newMeta() *etcdpb.CollectionMeta {
segBytes := proto.MarshalTextString(&segSch)
kvClient.Save("/segment/"+strconv.FormatInt(segSch.SegmentID, 10), segBytes)
return &collection
}

72
internal/writenode/flow_graph_insert_buffer_node.go
View File

@ -127,8 +127,24 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
}
}
// Timestamps
_, ok = idata.Data[1].(*storage.Int64FieldData)
if !ok {
idata.Data[1] = &storage.Int64FieldData{
Data: []int64{},
NumRows: 0,
}
}
tsData := idata.Data[1].(*storage.Int64FieldData)
for _, ts := range msg.Timestamps {
tsData.Data = append(tsData.Data, int64(ts))
}
tsData.NumRows += len(msg.Timestamps)
span.LogFields(oplog.Int("tsData numRows", tsData.NumRows))
// 1.1 Get CollectionMeta from etcd
collection, err := ibNode.replica.getCollectionByName(collectionName)
//collSchema, err := ibNode.getCollectionSchemaByName(collectionName)
if err != nil {
// GOOSE TODO add error handler
log.Println("bbb, Get meta wrong:", err)
@ -167,20 +183,18 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
fieldData := idata.Data[field.FieldID].(*storage.FloatVectorFieldData)
var offset int
for _, blob := range msg.RowData {
offset = 0
for j := 0; j < dim; j++ {
var v float32
buf := bytes.NewBuffer(blob.GetValue()[pos+offset:])
buf := bytes.NewBuffer(blob.GetValue()[pos:])
if err := binary.Read(buf, binary.LittleEndian, &v); err != nil {
log.Println("binary.read float32 err:", err)
}
fieldData.Data = append(fieldData.Data, v)
offset += int(unsafe.Sizeof(*(&v)))
pos += int(unsafe.Sizeof(*(&v)))
}
}
pos += offset
fieldData.NumRows += len(msg.RowIDs)
case schemapb.DataType_VECTOR_BINARY:
@ -208,15 +222,13 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
}
fieldData := idata.Data[field.FieldID].(*storage.BinaryVectorFieldData)
var offset int
for _, blob := range msg.RowData {
bv := blob.GetValue()[pos+offset : pos+(dim/8)]
bv := blob.GetValue()[pos : pos+(dim/8)]
fieldData.Data = append(fieldData.Data, bv...)
offset = len(bv)
pos += len(bv)
}
pos += offset
fieldData.NumRows += len(msg.RowData)
fieldData.NumRows += len(msg.RowData)
case schemapb.DataType_BOOL:
if _, ok := idata.Data[field.FieldID]; !ok {
idata.Data[field.FieldID] = &storage.BoolFieldData{
@ -226,18 +238,17 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
}
fieldData := idata.Data[field.FieldID].(*storage.BoolFieldData)
var v bool
for _, blob := range msg.RowData {
var v bool
buf := bytes.NewReader(blob.GetValue()[pos:])
if err := binary.Read(buf, binary.LittleEndian, &v); err != nil {
log.Println("binary.Read bool failed:", err)
}
fieldData.Data = append(fieldData.Data, v)
pos += int(unsafe.Sizeof(*(&v)))
}
pos += int(unsafe.Sizeof(*(&v)))
fieldData.NumRows += len(msg.RowIDs)
fieldData.NumRows += len(msg.RowIDs)
case schemapb.DataType_INT8:
if _, ok := idata.Data[field.FieldID]; !ok {
idata.Data[field.FieldID] = &storage.Int8FieldData{
@ -247,15 +258,15 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
}
fieldData := idata.Data[field.FieldID].(*storage.Int8FieldData)
var v int8
for _, blob := range msg.RowData {
var v int8
buf := bytes.NewReader(blob.GetValue()[pos:])
if err := binary.Read(buf, binary.LittleEndian, &v); err != nil {
log.Println("binary.Read int8 failed:", err)
}
fieldData.Data = append(fieldData.Data, v)
pos += int(unsafe.Sizeof(*(&v)))
}
pos += int(unsafe.Sizeof(*(&v)))
fieldData.NumRows += len(msg.RowIDs)
case schemapb.DataType_INT16:
@ -267,15 +278,16 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
}
fieldData := idata.Data[field.FieldID].(*storage.Int16FieldData)
var v int16
for _, blob := range msg.RowData {
var v int16
buf := bytes.NewReader(blob.GetValue()[pos:])
if err := binary.Read(buf, binary.LittleEndian, &v); err != nil {
log.Println("binary.Read int16 failed:", err)
}
fieldData.Data = append(fieldData.Data, v)
pos += int(unsafe.Sizeof(*(&v)))
}
pos += int(unsafe.Sizeof(*(&v)))
fieldData.NumRows += len(msg.RowIDs)
case schemapb.DataType_INT32:
@ -287,15 +299,15 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
}
fieldData := idata.Data[field.FieldID].(*storage.Int32FieldData)
var v int32
for _, blob := range msg.RowData {
var v int32
buf := bytes.NewReader(blob.GetValue()[pos:])
if err := binary.Read(buf, binary.LittleEndian, &v); err != nil {
log.Println("binary.Read int32 failed:", err)
}
fieldData.Data = append(fieldData.Data, v)
pos += int(unsafe.Sizeof(*(&v)))
}
pos += int(unsafe.Sizeof(*(&v)))
fieldData.NumRows += len(msg.RowIDs)
case schemapb.DataType_INT64:
@ -308,24 +320,27 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
fieldData := idata.Data[field.FieldID].(*storage.Int64FieldData)
switch field.FieldID {
case 0: // rowIDs
case 0:
fieldData.Data = append(fieldData.Data, msg.RowIDs...)
fieldData.NumRows += len(msg.RowIDs)
case 1: // Timestamps
case 1:
// Timestamps
for _, ts := range msg.Timestamps {
fieldData.Data = append(fieldData.Data, int64(ts))
}
fieldData.NumRows += len(msg.Timestamps)
default:
var v int64
for _, blob := range msg.RowData {
var v int64
buf := bytes.NewBuffer(blob.GetValue()[pos:])
if err := binary.Read(buf, binary.LittleEndian, &v); err != nil {
log.Println("binary.Read int64 failed:", err)
}
fieldData.Data = append(fieldData.Data, v)
pos += int(unsafe.Sizeof(*(&v)))
}
pos += int(unsafe.Sizeof(*(&v)))
fieldData.NumRows += len(msg.RowIDs)
}
@ -338,15 +353,16 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
}
fieldData := idata.Data[field.FieldID].(*storage.FloatFieldData)
var v float32
for _, blob := range msg.RowData {
var v float32
buf := bytes.NewBuffer(blob.GetValue()[pos:])
if err := binary.Read(buf, binary.LittleEndian, &v); err != nil {
log.Println("binary.Read float32 failed:", err)
}
fieldData.Data = append(fieldData.Data, v)
pos += int(unsafe.Sizeof(*(&v)))
}
pos += int(unsafe.Sizeof(*(&v)))
fieldData.NumRows += len(msg.RowIDs)
case schemapb.DataType_DOUBLE:
@ -358,16 +374,16 @@ func (ibNode *insertBufferNode) Operate(in []*Msg) []*Msg {
}
fieldData := idata.Data[field.FieldID].(*storage.DoubleFieldData)
var v float64
for _, blob := range msg.RowData {
var v float64
buf := bytes.NewBuffer(blob.GetValue()[pos:])
if err := binary.Read(buf, binary.LittleEndian, &v); err != nil {
log.Println("binary.Read float64 failed:", err)
}
fieldData.Data = append(fieldData.Data, v)
pos += int(unsafe.Sizeof(*(&v)))
}
pos += int(unsafe.Sizeof(*(&v)))
fieldData.NumRows += len(msg.RowIDs)
}
}

54
internal/writenode/flow_graph_insert_buffer_node_test.go
View File

@ -9,7 +9,6 @@ import (
"testing"
"time"
"github.com/golang/protobuf/proto"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
etcdkv "github.com/zilliztech/milvus-distributed/internal/kv/etcd"
@ -47,16 +46,11 @@ func TestFlowGraphInputBufferNode_Operate(t *testing.T) {
assert.Equal(t, testPath, fService.metaTable.client.(*etcdkv.EtcdKV).GetPath("."))
go fService.start()
collMeta := newMeta()
schemaBlob := proto.MarshalTextString(collMeta.Schema)
require.NotEqual(t, "", schemaBlob)
replica := newReplica()
err = replica.addCollection(collMeta.ID, schemaBlob)
require.NoError(t, err)
// Params.FlushInsertBufSize = 2
replica := newReplica()
iBNode := newInsertBufferNode(ctx, insertChan, replica)
newMeta()
inMsg := genInsertMsg()
var iMsg flowgraph.Msg = &inMsg
iBNode.Operate([]*flowgraph.Msg{&iMsg})
@ -76,12 +70,14 @@ func genInsertMsg() insertMsg {
binary.LittleEndian.PutUint32(buf, math.Float32bits(ele))
rawData = append(rawData, buf...)
}
log.Println(len(rawData))
// Binary vector
// Dimension of binary vector is 32
// size := 4, = 32 / 8
var bvector = []byte{255, 255, 255, 0}
rawData = append(rawData, bvector...)
log.Println(len(rawData))
// Bool
var fieldBool = true
@ -91,6 +87,7 @@ func genInsertMsg() insertMsg {
}
rawData = append(rawData, buf.Bytes()...)
log.Println(len(rawData))
// int8
var dataInt8 int8 = 100
@ -99,6 +96,7 @@ func genInsertMsg() insertMsg {
panic(err)
}
rawData = append(rawData, bint8.Bytes()...)
log.Println(len(rawData))
// int16
var dataInt16 int16 = 200
@ -107,6 +105,7 @@ func genInsertMsg() insertMsg {
panic(err)
}
rawData = append(rawData, bint16.Bytes()...)
log.Println(len(rawData))
// int32
var dataInt32 int32 = 300
@ -115,14 +114,16 @@ func genInsertMsg() insertMsg {
panic(err)
}
rawData = append(rawData, bint32.Bytes()...)
log.Println(len(rawData))
// int64
var dataInt64 int64 = 400
var dataInt64 int64 = 300
bint64 := new(bytes.Buffer)
if err := binary.Write(bint64, binary.LittleEndian, dataInt64); err != nil {
panic(err)
}
rawData = append(rawData, bint64.Bytes()...)
log.Println(len(rawData))
// float32
var datafloat float32 = 1.1
@ -131,6 +132,7 @@ func genInsertMsg() insertMsg {
panic(err)
}
rawData = append(rawData, bfloat32.Bytes()...)
log.Println(len(rawData))
// float64
var datafloat64 float64 = 2.2
@ -139,7 +141,7 @@ func genInsertMsg() insertMsg {
panic(err)
}
rawData = append(rawData, bfloat64.Bytes()...)
log.Println("Test rawdata length:", len(rawData))
log.Println(len(rawData))
timeRange := TimeRange{
timestampMin: 0,
@ -168,32 +170,16 @@ func genInsertMsg() insertMsg {
InsertRequest: internalpb.InsertRequest{
MsgType: commonpb.MsgType_kInsert,
ReqID: UniqueID(0),
CollectionName: "col1",
CollectionName: "coll1",
PartitionTag: "default",
SegmentID: UniqueID(1),
ChannelID: UniqueID(0),
ProxyID: UniqueID(0),
Timestamps: []Timestamp{
Timestamp(i + 1000),
Timestamp(i + 1000),
Timestamp(i + 1000),
Timestamp(i + 1000),
Timestamp(i + 1000),
},
RowIDs: []UniqueID{
UniqueID(i),
UniqueID(i),
UniqueID(i),
UniqueID(i),
UniqueID(i),
},
Timestamps: []Timestamp{Timestamp(i + 1000)},
RowIDs: []UniqueID{UniqueID(i)},
RowData: []*commonpb.Blob{
{Value: rawData},
{Value: rawData},
{Value: rawData},
{Value: rawData},
{Value: rawData},
},
},
}
@ -207,11 +193,9 @@ func genInsertMsg() insertMsg {
},
},
FlushMsg: internalpb.FlushMsg{
MsgType: commonpb.MsgType_kFlush,
SegmentID: UniqueID(1),
Timestamp: Timestamp(2000),
CollectionID: UniqueID(1),
PartitionTag: "default",
MsgType: commonpb.MsgType_kFlush,
SegmentID: UniqueID(1),
Timestamp: Timestamp(2000),
},
}
iMsg.flushMessages = append(iMsg.flushMessages, &fmsg)

9
scripts/proto_gen_go.sh
View File

@ -45,6 +45,15 @@ ${protoc} --go_out=plugins=grpc,paths=source_relative:./etcdpb etcd_meta.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./indexcgopb index_cgo_msg.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./internalpb internal_msg.proto
PROTOBUF_GOOGLE_SRC=${SCRIPTS_DIR}/../cmake_build/thirdparty/protobuf/protobuf-src/src/google
if [ -d ${PROTOBUF_GOOGLE_SRC} ]; then
echo ${PROTOBUF_GOOGLE_SRC}
ln -snf ${PROTOBUF_GOOGLE_SRC} google
${protoc} --go_out=plugins=grpc,paths=source_relative:./internalpb msg_header.proto
unlink google
fi
${protoc} --go_out=plugins=grpc,paths=source_relative:./servicepb service_msg.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./servicepb service.proto
${protoc} --go_out=plugins=grpc,paths=source_relative:./masterpb master.proto