Use channelsMgr to manage search message stream (#5819)

Signed-off-by: dragondriver <jiquan.long@zilliz.com>

go.sum

@@ -319,6 +319,7 @@ github.com/prometheus/procfs v0.1.3/go.mod h1:lV6e/gmhEcM9IjHGsFOCxxuZ+z1YqCvr4O
 github.com/prometheus/tsdb v0.7.1/go.mod h1:qhTCs0VvXwvX/y3TZrWD7rabWM+ijKTux40TwIPHuXU=
 github.com/protocolbuffers/protobuf v3.17.0+incompatible h1:MYhKKlaNOl8FB3F4u6oM2AlpcyLtT+p8Ec1w/9YeHss=
 github.com/protocolbuffers/protobuf v3.17.2+incompatible h1:ZM85np8wovAzyCCjAF9oZydM0VZh13Khe1633aDtJhA=
+github.com/protocolbuffers/protobuf v3.17.3+incompatible h1:weIpdqbAakIy/7PnlmdSBnPdODTtUySpnf3LyypYPwA=
 github.com/quasilyte/go-ruleguard v0.2.1 h1:56eRm0daAyny9UhJnmtJW/UyLZQusukBAB8oT8AHKHo=
 github.com/quasilyte/go-ruleguard v0.2.1/go.mod h1:hN2rVc/uS4bQhQKTio2XaSJSafJwqBUWWwtssT3cQmc=
 github.com/rivo/tview v0.0.0-20200219210816-cd38d7432498/go.mod h1:6lkG1x+13OShEf0EaOCaTQYyB7d5nSbb181KtjlS+84=

internal/distributed/proxynode/client/client.go

@@ -135,6 +135,10 @@ func (c *Client) InvalidateCollectionMetaCache(ctx context.Context, req *proxypb
 	})
 	return ret.(*commonpb.Status), err
 }
-func (c *Client) ReleaseDQLMessageStream(ctx context.Context, in *proxypb.ReleaseDQLMessageStreamRequest) (*commonpb.Status, error) {
-	panic("not implemented") // TODO: Implement
+
+func (c *Client) ReleaseDQLMessageStream(ctx context.Context, req *proxypb.ReleaseDQLMessageStreamRequest) (*commonpb.Status, error) {
+	ret, err := c.recall(func() (interface{}, error) {
+		return c.grpcClient.ReleaseDQLMessageStream(ctx, req)
+	})
+	return ret.(*commonpb.Status), err
 }

internal/distributed/proxynode/service.go

@@ -274,8 +274,8 @@ func (s *Server) InvalidateCollectionMetaCache(ctx context.Context, request *pro
 	return s.proxynode.InvalidateCollectionMetaCache(ctx, request)
 }
 
-func (s *Server) ReleaseDQLMessageStream(ctx context.Context, in *proxypb.ReleaseDQLMessageStreamRequest) (*commonpb.Status, error) {
-	panic("not implement")
+func (s *Server) ReleaseDQLMessageStream(ctx context.Context, request *proxypb.ReleaseDQLMessageStreamRequest) (*commonpb.Status, error) {
+	return s.proxynode.ReleaseDQLMessageStream(ctx, request)
 }
 
 func (s *Server) CreateCollection(ctx context.Context, request *milvuspb.CreateCollectionRequest) (*commonpb.Status, error) {

internal/proto/query_service.proto

@@ -107,6 +107,7 @@ message ReleasePartitionsRequest {
 
 message CreateQueryChannelRequest {
   int64 collectionID = 1;
+  int64 proxyID = 2;
 }
 
 message CreateQueryChannelResponse {

internal/proto/querypb/query_service.pb.go

@@ -680,6 +680,7 @@ func (m *ReleasePartitionsRequest) GetPartitionIDs() []int64 {
 
 type CreateQueryChannelRequest struct {
 	CollectionID         int64    `protobuf:"varint,1,opt,name=collectionID,proto3" json:"collectionID,omitempty"`
+	ProxyID              int64    `protobuf:"varint,2,opt,name=proxyID,proto3" json:"proxyID,omitempty"`
 	XXX_NoUnkeyedLiteral struct{} `json:"-"`
 	XXX_unrecognized     []byte   `json:"-"`
 	XXX_sizecache        int32    `json:"-"`
@@ -717,6 +718,13 @@ func (m *CreateQueryChannelRequest) GetCollectionID() int64 {
 	return 0
 }
 
+func (m *CreateQueryChannelRequest) GetProxyID() int64 {
+	if m != nil {
+		return m.ProxyID
+	}
+	return 0
+}
+
 type CreateQueryChannelResponse struct {
 	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"`
@@ -1987,123 +1995,123 @@ func init() {
 func init() { proto.RegisterFile("query_service.proto", fileDescriptor_5fcb6756dc1afb8d) }
 
 var fileDescriptor_5fcb6756dc1afb8d = []byte{
-	// 1845 bytes of a gzipped FileDescriptorProto
+	// 1856 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xec, 0x58, 0xcd, 0x73, 0x1c, 0x47,
 	0x15, 0xd7, 0xec, 0xae, 0x56, 0xda, 0xb7, 0x5f, 0x93, 0xb6, 0x2d, 0xd6, 0x4b, 0x9c, 0x38, 0xe3,
-	0x84, 0x38, 0x0a, 0x91, 0x53, 0x4a, 0xa0, 0xc8, 0x81, 0x02, 0x4b, 0x1b, 0x8b, 0x85, 0x44, 0x11,
-	0x23, 0x13, 0xaa, 0x5c, 0x2e, 0x96, 0xd1, 0x4c, 0x6b, 0x77, 0x2a, 0x33, 0xdd, 0xab, 0xe9, 0x5e,
-	0xcb, 0xf1, 0x01, 0x8a, 0x2a, 0x2e, 0x9c, 0x38, 0x71, 0xe2, 0xc4, 0x85, 0x82, 0x03, 0xff, 0x03,
-	0xff, 0x08, 0x50, 0x14, 0xb9, 0x71, 0xe6, 0x4e, 0x4d, 0x77, 0xcf, 0xec, 0x7c, 0xf4, 0x4a, 0x6b,
-	0xc9, 0x8e, 0x53, 0x14, 0xb7, 0xe9, 0xd7, 0xaf, 0xdf, 0x77, 0xff, 0xe6, 0xf5, 0x83, 0x2b, 0x27,
-	0x33, 0x1c, 0x7d, 0x3e, 0x62, 0x38, 0x7a, 0xe4, 0xbb, 0x78, 0x6b, 0x1a, 0x51, 0x4e, 0x11, 0x0a,
-	0xfd, 0xe0, 0xd1, 0x8c, 0xc9, 0xd5, 0x96, 0xe0, 0xe8, 0xb7, 0x5c, 0x1a, 0x86, 0x94, 0x48, 0x5a,
-	0xbf, 0x95, 0xe5, 0xe8, 0x77, 0x7c, 0xc2, 0x71, 0x44, 0x9c, 0x20, 0xd9, 0x65, 0xee, 0x04, 0x87,
-	0x8e, 0x5a, 0x21, 0xcf, 0xe1, 0x4e, 0x5e, 0x83, 0xf5, 0x4b, 0xb8, 0x62, 0xe3, 0xb1, 0xcf, 0x38,
-	0x8e, 0xf6, 0xa9, 0x87, 0x6d, 0x7c, 0x32, 0xc3, 0x8c, 0xa3, 0x77, 0xa1, 0x76, 0xe4, 0x30, 0xdc,
-	0x33, 0x6e, 0x1a, 0xb7, 0x9b, 0xdb, 0x2f, 0x6f, 0xe5, 0xec, 0x50, 0x06, 0x7c, 0xcc, 0xc6, 0x3b,
-	0x0e, 0xc3, 0xb6, 0xe0, 0x44, 0xdf, 0x86, 0x35, 0xc7, 0xf3, 0x22, 0xcc, 0x58, 0xaf, 0x72, 0xc6,
-	0xa1, 0xbb, 0x92, 0xc7, 0x4e, 0x98, 0xad, 0xdf, 0x1a, 0x70, 0x35, 0x6f, 0x01, 0x9b, 0x52, 0xc2,
-	0x30, 0x7a, 0x0f, 0xea, 0x8c, 0x3b, 0x7c, 0xc6, 0x94, 0x11, 0x5f, 0xd7, 0xca, 0x3b, 0x14, 0x2c,
-	0xb6, 0x62, 0x45, 0x3b, 0xd0, 0xf4, 0x89, 0xcf, 0x47, 0x53, 0x27, 0x72, 0xc2, 0xc4, 0x92, 0xd7,
-	0xf2, 0x27, 0xd3, 0x18, 0x0d, 0x89, 0xcf, 0x0f, 0x04, 0xa3, 0x0d, 0x7e, 0xfa, 0x6d, 0xfd, 0x0c,
-	0x36, 0x0e, 0x27, 0xf4, 0x74, 0x97, 0x06, 0x01, 0x76, 0xb9, 0x4f, 0x09, 0xbb, 0x78, 0x54, 0x10,
-	0xd4, 0xbc, 0xa3, 0xe1, 0x40, 0x18, 0x52, 0xb5, 0xc5, 0xb7, 0xc5, 0xe1, 0x6b, 0x25, 0xf9, 0x97,
-	0xf1, 0xf9, 0x75, 0x68, 0xbb, 0xa9, 0xac, 0xe1, 0x20, 0xf6, 0xba, 0x7a, 0xbb, 0x6a, 0xe7, 0x89,
-	0xd6, 0xaf, 0x0c, 0xb8, 0x16, 0xab, 0x3d, 0x70, 0x22, 0xee, 0x3f, 0x7b, 0xaf, 0x90, 0x05, 0xad,
-	0xac, 0xc2, 0x5e, 0x55, 0xec, 0xe5, 0x68, 0xd6, 0x89, 0x8c, 0x6c, 0xd6, 0x84, 0xcb, 0x38, 0x6e,
-	0x41, 0x6b, 0x9a, 0x88, 0x9a, 0xfb, 0x9d, 0xa3, 0x59, 0x7f, 0x35, 0xe0, 0xda, 0x47, 0xd4, 0xf1,
-	0xe6, 0xd1, 0xfe, 0xd2, 0xdd, 0x46, 0xdf, 0x85, 0xba, 0xbc, 0x87, 0xbd, 0x9a, 0xd0, 0xf5, 0x46,
-	0x5e, 0x97, 0xba, 0xa3, 0x73, 0x0b, 0x0f, 0x05, 0xc1, 0x56, 0x87, 0xac, 0x5f, 0x1b, 0xd0, 0xb3,
-	0x71, 0x80, 0x1d, 0x86, 0x5f, 0xa0, 0x17, 0xd6, 0x17, 0x2a, 0x92, 0x2f, 0xb0, 0x80, 0x4a, 0x19,
-	0xaf, 0x95, 0x33, 0x9e, 0x89, 0xf6, 0xea, 0x45, 0xa2, 0xfd, 0xa7, 0x79, 0xb4, 0xbf, 0xe2, 0x9e,
-	0x5a, 0xdf, 0x83, 0xeb, 0xbb, 0x11, 0x76, 0x38, 0xfe, 0x71, 0xfc, 0x63, 0xd8, 0x9d, 0x38, 0x84,
-	0xe0, 0x20, 0x31, 0xb5, 0xa8, 0xc4, 0xd0, 0xa4, 0xf4, 0x0f, 0x06, 0xf4, 0x75, 0x12, 0x2e, 0x73,
-	0x29, 0xdf, 0x84, 0x6e, 0x24, 0x4d, 0x18, 0xb9, 0x52, 0x9e, 0xf0, 0xbd, 0x61, 0x77, 0x14, 0x59,
-	0x69, 0x41, 0x6f, 0x40, 0x27, 0xc2, 0x6c, 0x16, 0xcc, 0xf9, 0xaa, 0x82, 0xaf, 0x2d, 0xa9, 0x8a,
-	0xcd, 0xfa, 0xb3, 0x01, 0xd7, 0xf7, 0x30, 0x4f, 0x73, 0x11, 0xab, 0xc3, 0x5f, 0xd1, 0x84, 0x84,
-	0xd0, 0x2d, 0xd8, 0x89, 0x6e, 0x42, 0x33, 0xc3, 0xa2, 0xb2, 0x90, 0x25, 0xa1, 0xef, 0xc0, 0x6a,
-	0x1c, 0x3a, 0x2c, 0x2c, 0xea, 0x6c, 0x5b, 0x5b, 0xe5, 0x7f, 0xfe, 0x56, 0x5e, 0xaa, 0x2d, 0x0f,
-	0x58, 0x7f, 0x31, 0xa0, 0xaf, 0x0b, 0xcd, 0x65, 0xd2, 0xf7, 0x00, 0x36, 0x52, 0xe3, 0x46, 0x1e,
-	0x66, 0x6e, 0xe4, 0x4f, 0xc5, 0x15, 0x10, 0xe8, 0xda, 0xdc, 0xbe, 0x75, 0xbe, 0x79, 0xcc, 0xbe,
-	0x96, 0x8a, 0x18, 0x64, 0x24, 0x58, 0x3e, 0x5c, 0xdb, 0xc3, 0xfc, 0x10, 0x8f, 0x43, 0x4c, 0xf8,
-	0x90, 0x1c, 0xd3, 0x8b, 0x67, 0xf1, 0x15, 0x00, 0xa6, 0xe4, 0xa4, 0xc0, 0x9f, 0xa1, 0x58, 0x7f,
-	0xab, 0x40, 0x33, 0xa3, 0x08, 0xbd, 0x0c, 0x8d, 0x74, 0x57, 0x25, 0x61, 0x4e, 0x28, 0xe5, 0xbf,
-	0xa2, 0xc9, 0x7f, 0x21, 0x91, 0xd5, 0x72, 0x22, 0x37, 0xa0, 0x4e, 0xa8, 0x87, 0x87, 0x03, 0x01,
-	0xf3, 0x55, 0x5b, 0xad, 0xd0, 0x75, 0x58, 0x0f, 0x71, 0x38, 0x62, 0xfe, 0x13, 0x2c, 0x20, 0xa9,
-	0x6a, 0xaf, 0x85, 0x38, 0x3c, 0xf4, 0x9f, 0xe0, 0x78, 0x8b, 0xcc, 0xc2, 0x51, 0x44, 0x4f, 0x59,
-	0xaf, 0x2e, 0xb7, 0xc8, 0x2c, 0xb4, 0xe9, 0x29, 0x43, 0x37, 0x00, 0x7c, 0xe2, 0xe1, 0xc7, 0x23,
-	0xe2, 0x84, 0xb8, 0xb7, 0x26, 0xae, 0x46, 0x43, 0x50, 0xf6, 0x9d, 0x10, 0xa3, 0x1e, 0xac, 0x89,
-	0xc5, 0x70, 0xd0, 0x5b, 0x97, 0x07, 0xd5, 0x32, 0x76, 0x55, 0x5d, 0xa8, 0xe1, 0xa0, 0xd7, 0x90,
-	0xe7, 0x52, 0x02, 0xfa, 0x10, 0xda, 0xca, 0xef, 0x91, 0xac, 0x3a, 0x10, 0x55, 0x77, 0x53, 0x97,
-	0x56, 0x15, 0x40, 0x59, 0x73, 0x2d, 0x96, 0x59, 0xc5, 0xff, 0xa4, 0x8d, 0x62, 0x2e, 0x2f, 0x53,
-	0x76, 0xdf, 0x82, 0x55, 0x9f, 0x1c, 0xd3, 0xa4, 0xca, 0x5e, 0x3d, 0xc3, 0x1c, 0xa1, 0x4c, 0x72,
-	0x5b, 0x7f, 0x37, 0x60, 0xe3, 0xae, 0xe7, 0xe9, 0xf0, 0xef, 0xe9, 0x6b, 0x6a, 0x9e, 0xbf, 0x4a,
-	0x2e, 0x7f, 0xcb, 0xa0, 0xc3, 0xdb, 0xf0, 0x52, 0x01, 0xf5, 0x54, 0x19, 0x34, 0x6c, 0x33, 0x8f,
-	0x7b, 0xc3, 0x01, 0x7a, 0x0b, 0xcc, 0x3c, 0xf2, 0x0d, 0x07, 0xa2, 0x30, 0x1a, 0x76, 0x37, 0x87,
-	0x7d, 0xc3, 0x81, 0xf5, 0x4f, 0x03, 0xae, 0xdb, 0x38, 0xa4, 0x8f, 0xf0, 0xff, 0xae, 0x8f, 0xff,
-	0xaa, 0xc0, 0xc6, 0x4f, 0x1d, 0xee, 0x4e, 0x06, 0xa1, 0x22, 0xb2, 0x17, 0xe3, 0x60, 0xe1, 0x8a,
-	0xd7, 0xca, 0x57, 0x3c, 0x2d, 0xd3, 0x55, 0x5d, 0x99, 0xc6, 0xcf, 0xab, 0xad, 0x4f, 0x13, 0x7f,
-	0xe7, 0x65, 0x9a, 0x69, 0x49, 0xea, 0x17, 0x68, 0x49, 0xd0, 0x2e, 0xb4, 0xf1, 0x63, 0x37, 0x98,
-	0x79, 0x78, 0x24, 0xb5, 0xaf, 0x09, 0xed, 0xaf, 0x68, 0xb4, 0x67, 0xef, 0x48, 0x4b, 0x1d, 0x1a,
-	0x8a, 0xab, 0xf2, 0x6f, 0x03, 0xba, 0x6a, 0x37, 0xee, 0xe2, 0x96, 0x40, 0xc5, 0x42, 0x38, 0x2a,
-	0xe5, 0x70, 0x2c, 0x13, 0xd4, 0xe4, 0x7f, 0x5b, 0xcb, 0xfc, 0x6f, 0x6f, 0x00, 0x1c, 0x07, 0x33,
-	0x36, 0x19, 0x71, 0x3f, 0x4c, 0x30, 0xb1, 0x21, 0x28, 0xf7, 0xfd, 0x10, 0xa3, 0xbb, 0xd0, 0x3a,
-	0xf2, 0x49, 0x40, 0xc7, 0xa3, 0xa9, 0xc3, 0x27, 0x31, 0x32, 0x2e, 0x72, 0xf7, 0x9e, 0x8f, 0x03,
-	0x6f, 0x47, 0xf0, 0xda, 0x4d, 0x79, 0xe6, 0x20, 0x3e, 0x62, 0xfd, 0xb1, 0x02, 0x57, 0x62, 0x37,
-	0x95, 0xc7, 0xcf, 0xa1, 0xa0, 0x3e, 0x48, 0x4a, 0xa1, 0xba, 0xf8, 0xbf, 0x58, 0x88, 0x77, 0xb9,
-	0x1c, 0x2e, 0xf2, 0x1e, 0x40, 0x3f, 0x82, 0x4e, 0x40, 0x1d, 0x6f, 0xe4, 0x52, 0xe2, 0x89, 0x4c,
-	0x88, 0x08, 0x76, 0xb6, 0x5f, 0xd7, 0x99, 0x70, 0x3f, 0xf2, 0xc7, 0x63, 0x1c, 0xed, 0x26, 0xbc,
-	0x76, 0x3b, 0x10, 0xaf, 0x21, 0xb5, 0x14, 0x08, 0xaa, 0xda, 0xdd, 0xe7, 0x17, 0xab, 0xa4, 0x06,
-	0xaa, 0x67, 0xf4, 0x5c, 0xb5, 0x25, 0x7a, 0xae, 0x55, 0x4d, 0xbb, 0x9f, 0xef, 0x04, 0xea, 0xa5,
-	0x4e, 0xe0, 0x3e, 0xb4, 0x53, 0x5c, 0x11, 0x45, 0x7f, 0x0b, 0xda, 0xd2, 0xac, 0x51, 0x1c, 0x09,
-	0xec, 0x25, 0x9d, 0xb1, 0x24, 0x7e, 0x24, 0x68, 0xb1, 0xd4, 0x14, 0xb7, 0xe4, 0x4f, 0xa9, 0x61,
-	0x67, 0x28, 0xd6, 0xef, 0x0c, 0x30, 0xb3, 0x88, 0x2c, 0x24, 0x2f, 0xd1, 0x72, 0xc7, 0xed, 0xb1,
-	0x1c, 0xf4, 0xcc, 0x61, 0x51, 0xb5, 0xc7, 0x27, 0x59, 0x71, 0x03, 0xf4, 0x3e, 0x6c, 0x48, 0xc6,
-	0x12, 0x8c, 0xca, 0x36, 0xf9, 0xaa, 0xd8, 0xb5, 0x0b, 0x58, 0xfa, 0x0f, 0x03, 0x3a, 0xf3, 0xc2,
-	0x59, 0xda, 0xaa, 0x25, 0x5e, 0xd2, 0xe8, 0x1e, 0xb4, 0x95, 0x0d, 0xa3, 0x6c, 0xe1, 0xbf, 0xa6,
-	0xab, 0xba, 0x5c, 0xc4, 0xed, 0x56, 0x06, 0x12, 0x2f, 0x0b, 0x86, 0xd6, 0x2f, 0xa0, 0xfb, 0x03,
-	0x87, 0x78, 0xf4, 0xf8, 0x38, 0xa9, 0xd7, 0x0b, 0x14, 0xea, 0x07, 0xf9, 0x76, 0xe3, 0x29, 0x2e,
-	0xaf, 0xf5, 0xfb, 0x0a, 0x6c, 0xc4, 0xb4, 0x1d, 0x27, 0x70, 0x88, 0x8b, 0x97, 0x6f, 0x32, 0x9f,
-	0x0d, 0x9c, 0xde, 0x82, 0x36, 0xa3, 0xb3, 0xc8, 0xc5, 0xa3, 0x5c, 0xaf, 0xd9, 0x92, 0xc4, 0x7d,
-	0x79, 0xdf, 0x6e, 0x00, 0x78, 0x8c, 0x27, 0x1c, 0x0a, 0x5f, 0x3d, 0xc6, 0xd5, 0xf6, 0xab, 0xd0,
-	0x54, 0x32, 0x3c, 0x4a, 0xb0, 0x48, 0xc3, 0xba, 0x0d, 0x92, 0x34, 0xa0, 0x44, 0xb4, 0xa5, 0xf1,
-	0x79, 0xb1, 0xbb, 0x26, 0x76, 0xd7, 0x3c, 0xc6, 0xc5, 0xd6, 0x0d, 0x80, 0x47, 0x4e, 0xe0, 0x7b,
-	0xa2, 0x06, 0x44, 0xeb, 0xb9, 0x6e, 0x37, 0x04, 0x25, 0x0e, 0x81, 0xf5, 0x1b, 0x43, 0xe2, 0x6e,
-	0x3e, 0x3a, 0x17, 0x49, 0xd1, 0xf7, 0xf3, 0x29, 0xda, 0xd4, 0xa5, 0x48, 0x9f, 0x07, 0x95, 0xa9,
-	0xcd, 0x27, 0xd0, 0xc9, 0x3f, 0x4c, 0x50, 0x0b, 0xd6, 0xf7, 0x29, 0xff, 0xf0, 0xb1, 0xcf, 0xb8,
-	0xb9, 0x82, 0x3a, 0x00, 0xfb, 0x94, 0x1f, 0x44, 0x98, 0x61, 0xc2, 0x4d, 0x03, 0x01, 0xd4, 0x3f,
-	0x21, 0x03, 0x9f, 0x7d, 0x66, 0x56, 0xd0, 0x15, 0xf5, 0x92, 0x73, 0x82, 0x21, 0xf9, 0x18, 0x87,
-	0x34, 0xfa, 0xdc, 0xac, 0xc6, 0xc7, 0xd3, 0x55, 0x0d, 0x99, 0xd0, 0x4a, 0x59, 0xf6, 0x0e, 0x7e,
-	0x62, 0xae, 0xa2, 0x06, 0xac, 0xca, 0xcf, 0xfa, 0xe6, 0x27, 0x60, 0x16, 0x91, 0x17, 0x35, 0x61,
-	0x6d, 0x22, 0x2b, 0xd7, 0x5c, 0x41, 0x5d, 0x68, 0x06, 0x73, 0xeb, 0x4d, 0x23, 0x26, 0x8c, 0xa3,
-	0xa9, 0xab, 0xc0, 0xd7, 0xac, 0xc4, 0xda, 0xe2, 0x04, 0x0e, 0xe8, 0x29, 0x31, 0xab, 0x9b, 0x3f,
-	0x84, 0x56, 0xb6, 0x1d, 0x47, 0xeb, 0x50, 0xdb, 0xa7, 0x04, 0x9b, 0x2b, 0xb1, 0xd8, 0xbd, 0x88,
-	0x9e, 0xfa, 0x64, 0x2c, 0x7d, 0xb8, 0x17, 0xd1, 0x27, 0x98, 0x98, 0x95, 0x78, 0x83, 0x61, 0x27,
-	0x88, 0x37, 0xaa, 0xf1, 0x46, 0xbc, 0xc0, 0x9e, 0x59, 0xdb, 0xfe, 0x0f, 0x40, 0x4b, 0x80, 0xd7,
-	0xa1, 0x1c, 0x05, 0xa3, 0x29, 0xa0, 0x3d, 0xcc, 0x77, 0x69, 0x38, 0xa5, 0x24, 0xd1, 0xc0, 0xd0,
-	0xbb, 0x0b, 0xc6, 0xa6, 0x65, 0x56, 0x65, 0x74, 0xff, 0x1b, 0x0b, 0x4e, 0x14, 0xd8, 0xad, 0x15,
-	0x14, 0x0a, 0x8d, 0xf1, 0xdf, 0xfe, 0xbe, 0xef, 0x7e, 0x96, 0x8c, 0x04, 0xce, 0xd0, 0x58, 0x60,
-	0x4d, 0x34, 0x16, 0x6e, 0xae, 0x5a, 0x1c, 0xf2, 0xc8, 0x27, 0xe3, 0xe4, 0x45, 0x62, 0xad, 0xa0,
-	0x13, 0xb8, 0x1a, 0xbf, 0x56, 0xb8, 0xc3, 0x7d, 0xc6, 0x7d, 0x97, 0x25, 0x0a, 0xb7, 0x17, 0x2b,
-	0x2c, 0x31, 0x3f, 0xa5, 0x4a, 0x17, 0x5a, 0xd9, 0xb1, 0x36, 0x7a, 0x53, 0x57, 0xc0, 0x9a, 0xd1,
-	0x7b, 0xff, 0xf6, 0xf9, 0x8c, 0xa9, 0x92, 0x00, 0xba, 0x85, 0x51, 0x32, 0xd2, 0x5e, 0x14, 0xfd,
-	0x3c, 0xbb, 0xff, 0xf6, 0x52, 0xbc, 0xa9, 0x36, 0x1f, 0x3a, 0xf9, 0xf1, 0x2d, 0x7a, 0x6b, 0x91,
-	0x80, 0xd2, 0xe8, 0xac, 0xbf, 0xb9, 0x0c, 0x6b, 0xaa, 0xea, 0x01, 0x74, 0xf2, 0xb3, 0x46, 0xbd,
-	0x2a, 0xed, 0x3c, 0xb2, 0x7f, 0xd6, 0x8b, 0xd3, 0x5a, 0x41, 0x3f, 0x87, 0x97, 0x4a, 0x03, 0x3e,
-	0xf4, 0x4d, 0x7d, 0xd4, 0xf5, 0x73, 0xc0, 0xf3, 0x34, 0x28, 0xeb, 0xe7, 0x51, 0x5c, 0x6c, 0x7d,
-	0x69, 0xa2, 0xbb, 0xbc, 0xf5, 0x19, 0xf1, 0x67, 0x59, 0xff, 0xd4, 0x1a, 0x66, 0x80, 0xca, 0x43,
-	0x41, 0xf4, 0x8e, 0x4e, 0xc5, 0xc2, 0xf1, 0x63, 0x7f, 0x6b, 0x59, 0xf6, 0x34, 0xe5, 0x33, 0x01,
-	0x09, 0xc5, 0xf9, 0x99, 0x56, 0xed, 0xc2, 0x79, 0xa0, 0x5e, 0xed, 0xe2, 0x19, 0x99, 0x2c, 0xea,
-	0xfc, 0x20, 0x43, 0x9f, 0x2b, 0xed, 0xe0, 0x4a, 0x5f, 0xd4, 0xfa, 0xb9, 0x88, 0xb5, 0xb2, 0xfd,
-	0xc5, 0x3a, 0x34, 0x84, 0xf3, 0x02, 0x10, 0xfe, 0x0f, 0xba, 0xcf, 0x1e, 0x74, 0x1f, 0x42, 0xb7,
-	0x30, 0x0e, 0xd2, 0xe3, 0xa1, 0x7e, 0x66, 0x74, 0xde, 0xc5, 0x38, 0x02, 0x54, 0x9e, 0xc5, 0xe8,
-	0x2b, 0x74, 0xe1, 0xcc, 0xe6, 0x3c, 0x1d, 0x0f, 0xa1, 0x5b, 0x98, 0x85, 0xe8, 0x3d, 0xd0, 0x0f,
-	0x4c, 0xce, 0x93, 0xfe, 0x29, 0xb4, 0xb2, 0xaf, 0x62, 0xfd, 0x4f, 0x49, 0xf3, 0x6e, 0x7e, 0xf1,
-	0xa0, 0xf4, 0xfc, 0x41, 0xfb, 0x21, 0x74, 0x0b, 0x0f, 0x61, 0x7d, 0xe4, 0xf5, 0xaf, 0xe5, 0xf3,
-	0xa4, 0x7f, 0x79, 0x30, 0xb3, 0xf3, 0xfe, 0x83, 0xed, 0xb1, 0xcf, 0x27, 0xb3, 0xa3, 0xd8, 0x88,
-	0x3b, 0xf2, 0xe4, 0x3b, 0x3e, 0x55, 0x5f, 0x77, 0x92, 0xfb, 0x76, 0x47, 0x08, 0xbb, 0x23, 0x84,
-	0x4d, 0x8f, 0x8e, 0xea, 0x62, 0xf9, 0xde, 0x7f, 0x03, 0x00, 0x00, 0xff, 0xff, 0xde, 0x2e, 0x8d,
-	0xea, 0x88, 0x20, 0x00, 0x00,
+	0x04, 0x3b, 0x0a, 0x91, 0x53, 0x4a, 0xa0, 0xc8, 0x81, 0x2a, 0x2c, 0x6d, 0x2c, 0x16, 0x12, 0x45,
+	0x8c, 0x4c, 0x28, 0x5c, 0x2e, 0x96, 0xd1, 0x4c, 0x6b, 0x35, 0x95, 0x99, 0xee, 0xd5, 0x74, 0xaf,
+	0x65, 0xfb, 0x00, 0x45, 0x15, 0x17, 0x4e, 0x9c, 0x38, 0x71, 0xe2, 0x42, 0xc1, 0x81, 0xff, 0x81,
+	0x7f, 0x04, 0x28, 0x8a, 0xdc, 0x38, 0x73, 0xa7, 0xa6, 0xbb, 0x67, 0x76, 0x3e, 0x7a, 0xa5, 0xb5,
+	0x64, 0x47, 0x29, 0x8a, 0xdb, 0xf4, 0xeb, 0xd7, 0xef, 0xbb, 0x7f, 0xf3, 0xfa, 0xc1, 0x95, 0xe3,
+	0x29, 0x8e, 0x9e, 0x8e, 0x18, 0x8e, 0x1e, 0xfb, 0x2e, 0xde, 0x98, 0x44, 0x94, 0x53, 0x84, 0x42,
+	0x3f, 0x78, 0x3c, 0x65, 0x72, 0xb5, 0x21, 0x38, 0xfa, 0x2d, 0x97, 0x86, 0x21, 0x25, 0x92, 0xd6,
+	0x6f, 0x65, 0x39, 0xfa, 0x1d, 0x9f, 0x70, 0x1c, 0x11, 0x27, 0x48, 0x76, 0x99, 0x7b, 0x84, 0x43,
+	0x47, 0xad, 0x90, 0xe7, 0x70, 0x27, 0xaf, 0xc1, 0xfa, 0x25, 0x5c, 0xb1, 0xf1, 0xd8, 0x67, 0x1c,
+	0x47, 0xbb, 0xd4, 0xc3, 0x36, 0x3e, 0x9e, 0x62, 0xc6, 0xd1, 0x7b, 0x50, 0x3b, 0x70, 0x18, 0xee,
+	0x19, 0x37, 0x8d, 0x3b, 0xcd, 0xcd, 0x57, 0x37, 0x72, 0x76, 0x28, 0x03, 0x3e, 0x61, 0xe3, 0x2d,
+	0x87, 0x61, 0x5b, 0x70, 0xa2, 0x6f, 0xc3, 0x8a, 0xe3, 0x79, 0x11, 0x66, 0xac, 0x57, 0x39, 0xe5,
+	0xd0, 0x3d, 0xc9, 0x63, 0x27, 0xcc, 0xd6, 0x6f, 0x0d, 0xb8, 0x9a, 0xb7, 0x80, 0x4d, 0x28, 0x61,
+	0x18, 0xbd, 0x0f, 0x75, 0xc6, 0x1d, 0x3e, 0x65, 0xca, 0x88, 0xaf, 0x6b, 0xe5, 0xed, 0x0b, 0x16,
+	0x5b, 0xb1, 0xa2, 0x2d, 0x68, 0xfa, 0xc4, 0xe7, 0xa3, 0x89, 0x13, 0x39, 0x61, 0x62, 0xc9, 0x1b,
+	0xf9, 0x93, 0x69, 0x8c, 0x86, 0xc4, 0xe7, 0x7b, 0x82, 0xd1, 0x06, 0x3f, 0xfd, 0xb6, 0x7e, 0x06,
+	0x6b, 0xfb, 0x47, 0xf4, 0x64, 0x9b, 0x06, 0x01, 0x76, 0xb9, 0x4f, 0x09, 0x3b, 0x7f, 0x54, 0x10,
+	0xd4, 0xbc, 0x83, 0xe1, 0x40, 0x18, 0x52, 0xb5, 0xc5, 0xb7, 0xc5, 0xe1, 0x6b, 0x25, 0xf9, 0x17,
+	0xf1, 0xf9, 0x4d, 0x68, 0xbb, 0xa9, 0xac, 0xe1, 0x20, 0xf6, 0xba, 0x7a, 0xa7, 0x6a, 0xe7, 0x89,
+	0xd6, 0xaf, 0x0c, 0xb8, 0x16, 0xab, 0xdd, 0x73, 0x22, 0xee, 0xbf, 0x78, 0xaf, 0x90, 0x05, 0xad,
+	0xac, 0xc2, 0x5e, 0x55, 0xec, 0xe5, 0x68, 0xd6, 0xb1, 0x8c, 0x6c, 0xd6, 0x84, 0x8b, 0x38, 0x6e,
+	0x41, 0x6b, 0x92, 0x88, 0x9a, 0xf9, 0x9d, 0xa3, 0x59, 0x7f, 0x35, 0xe0, 0xda, 0xc7, 0xd4, 0xf1,
+	0x66, 0xd1, 0xfe, 0xd2, 0xdd, 0x46, 0xdf, 0x85, 0xba, 0xbc, 0x87, 0xbd, 0x9a, 0xd0, 0xf5, 0x56,
+	0x5e, 0x97, 0xba, 0xa3, 0x33, 0x0b, 0xf7, 0x05, 0xc1, 0x56, 0x87, 0xac, 0x5f, 0x1b, 0xd0, 0xb3,
+	0x71, 0x80, 0x1d, 0x86, 0x2f, 0xd1, 0x0b, 0xeb, 0x0b, 0x15, 0xc9, 0x4b, 0x2c, 0xa0, 0x52, 0xc6,
+	0x6b, 0xe5, 0x8c, 0x67, 0xa2, 0xbd, 0x7c, 0x9e, 0x68, 0xff, 0x69, 0x16, 0xed, 0xaf, 0xb8, 0xa7,
+	0xd6, 0x4f, 0xe1, 0xfa, 0x76, 0x84, 0x1d, 0x8e, 0x7f, 0x14, 0xff, 0x18, 0xb6, 0x8f, 0x1c, 0x42,
+	0x70, 0x90, 0x98, 0x5a, 0x54, 0x62, 0x68, 0x94, 0xf4, 0x60, 0x65, 0x12, 0xd1, 0x27, 0x4f, 0x53,
+	0xfb, 0x92, 0xa5, 0xf5, 0x07, 0x03, 0xfa, 0x3a, 0xd9, 0x17, 0xb9, 0xae, 0xb7, 0xa1, 0x1b, 0x49,
+	0xe3, 0x46, 0xae, 0x94, 0x27, 0xb4, 0x36, 0xec, 0x8e, 0x22, 0x2b, 0x2d, 0xe8, 0x2d, 0xe8, 0x44,
+	0x98, 0x4d, 0x83, 0x19, 0x5f, 0x55, 0xf0, 0xb5, 0x25, 0x55, 0xb1, 0x59, 0x7f, 0x36, 0xe0, 0xfa,
+	0x0e, 0xe6, 0x69, 0x96, 0x62, 0x75, 0xf8, 0x2b, 0x9a, 0xaa, 0x10, 0xba, 0x05, 0x3b, 0xd1, 0x4d,
+	0x68, 0x66, 0x58, 0x54, 0x7e, 0xb2, 0x24, 0xf4, 0x1d, 0x58, 0x8e, 0x43, 0x87, 0x85, 0x45, 0x9d,
+	0x4d, 0x6b, 0xa3, 0xdc, 0x0d, 0x6c, 0xe4, 0xa5, 0xda, 0xf2, 0x80, 0xf5, 0x17, 0x03, 0xfa, 0xba,
+	0xd0, 0x5c, 0x24, 0x7d, 0x0f, 0x61, 0x2d, 0x35, 0x6e, 0xe4, 0x61, 0xe6, 0x46, 0xfe, 0x44, 0x5c,
+	0x0e, 0x81, 0xbb, 0xcd, 0xcd, 0x5b, 0x67, 0x9b, 0xc7, 0xec, 0x6b, 0xa9, 0x88, 0x41, 0x46, 0x82,
+	0xe5, 0xc3, 0xb5, 0x1d, 0xcc, 0xf7, 0xf1, 0x38, 0xc4, 0x84, 0x0f, 0xc9, 0x21, 0x3d, 0x7f, 0x16,
+	0x5f, 0x03, 0x60, 0x4a, 0x4e, 0xfa, 0x4b, 0xc8, 0x50, 0xac, 0xbf, 0x55, 0xa0, 0x99, 0x51, 0x84,
+	0x5e, 0x85, 0x46, 0xba, 0xab, 0x92, 0x30, 0x23, 0x94, 0xf2, 0x5f, 0xd1, 0xe4, 0xbf, 0x90, 0xc8,
+	0x6a, 0x39, 0x91, 0x6b, 0x50, 0x27, 0xd4, 0xc3, 0xc3, 0x81, 0xf8, 0x01, 0x54, 0x6d, 0xb5, 0x42,
+	0xd7, 0x61, 0x35, 0xc4, 0xe1, 0x88, 0xf9, 0xcf, 0xb0, 0x00, 0xab, 0xaa, 0xbd, 0x12, 0xe2, 0x70,
+	0xdf, 0x7f, 0x86, 0xe3, 0x2d, 0x32, 0x0d, 0x47, 0x11, 0x3d, 0x61, 0xbd, 0xba, 0xdc, 0x22, 0xd3,
+	0xd0, 0xa6, 0x27, 0x0c, 0xdd, 0x00, 0xf0, 0x89, 0x87, 0x9f, 0x8c, 0x88, 0x13, 0xe2, 0xde, 0x8a,
+	0xb8, 0x1a, 0x0d, 0x41, 0xd9, 0x75, 0x42, 0x1c, 0x5f, 0x6a, 0xb1, 0x18, 0x0e, 0x7a, 0xab, 0xf2,
+	0xa0, 0x5a, 0xc6, 0xae, 0xaa, 0x0b, 0x35, 0x1c, 0xf4, 0x1a, 0xf2, 0x5c, 0x4a, 0x40, 0x1f, 0x41,
+	0x5b, 0xf9, 0x3d, 0x92, 0x55, 0x07, 0xa2, 0xea, 0x6e, 0xea, 0xd2, 0xaa, 0x02, 0x28, 0x6b, 0xae,
+	0xc5, 0x32, 0xab, 0xf8, 0x6f, 0xb5, 0x56, 0xcc, 0xe5, 0x45, 0xca, 0xee, 0x5b, 0xb0, 0xec, 0x93,
+	0x43, 0x9a, 0x54, 0xd9, 0xeb, 0xa7, 0x98, 0x23, 0x94, 0x49, 0x6e, 0xeb, 0xef, 0x06, 0xac, 0xdd,
+	0xf3, 0x3c, 0x1d, 0x32, 0x3e, 0x7f, 0x4d, 0xcd, 0xf2, 0x57, 0xc9, 0xe5, 0x6f, 0x11, 0x74, 0x78,
+	0x07, 0x5e, 0x29, 0xa0, 0x9e, 0x2a, 0x83, 0x86, 0x6d, 0xe6, 0x71, 0x6f, 0x38, 0x40, 0x6f, 0x83,
+	0x99, 0x47, 0xbe, 0xe1, 0x40, 0x14, 0x46, 0xc3, 0xee, 0xe6, 0xb0, 0x6f, 0x38, 0xb0, 0xfe, 0x69,
+	0xc0, 0x75, 0x1b, 0x87, 0xf4, 0x31, 0xfe, 0xdf, 0xf5, 0xf1, 0x5f, 0x15, 0x58, 0xfb, 0x89, 0xc3,
+	0xdd, 0xa3, 0x41, 0xa8, 0x88, 0xec, 0x72, 0x1c, 0x2c, 0x5c, 0xf1, 0x5a, 0xf9, 0x8a, 0xa7, 0x65,
+	0xba, 0xac, 0x2b, 0xd3, 0xf8, 0xe1, 0xb5, 0xf1, 0x59, 0xe2, 0xef, 0xac, 0x4c, 0x33, 0xcd, 0x4a,
+	0xfd, 0x1c, 0xcd, 0x0a, 0xda, 0x86, 0x36, 0x7e, 0xe2, 0x06, 0x53, 0x0f, 0x8f, 0xa4, 0xf6, 0x15,
+	0xa1, 0xfd, 0x35, 0x8d, 0xf6, 0xec, 0x1d, 0x69, 0xa9, 0x43, 0x43, 0x71, 0x55, 0xfe, 0x6d, 0x40,
+	0x57, 0xed, 0xc6, 0xfd, 0xdd, 0x02, 0xa8, 0x58, 0x08, 0x47, 0xa5, 0x1c, 0x8e, 0x45, 0x82, 0x9a,
+	0xfc, 0x6f, 0x6b, 0x99, 0xff, 0xed, 0x0d, 0x80, 0xc3, 0x60, 0xca, 0x8e, 0x46, 0xdc, 0x0f, 0x13,
+	0x4c, 0x6c, 0x08, 0xca, 0x03, 0x3f, 0xc4, 0xe8, 0x1e, 0xb4, 0x0e, 0x7c, 0x12, 0xd0, 0xf1, 0x68,
+	0xe2, 0xf0, 0xa3, 0x18, 0x19, 0xe7, 0xb9, 0x7b, 0xdf, 0xc7, 0x81, 0xb7, 0x25, 0x78, 0xed, 0xa6,
+	0x3c, 0xb3, 0x17, 0x1f, 0xb1, 0xfe, 0x58, 0x81, 0x2b, 0xb1, 0x9b, 0xca, 0xe3, 0x97, 0x50, 0x50,
+	0x1f, 0x26, 0xa5, 0x50, 0x9d, 0xff, 0x5f, 0x2c, 0xc4, 0xbb, 0x5c, 0x0e, 0xe7, 0x79, 0x29, 0xa0,
+	0x1f, 0x42, 0x27, 0xa0, 0x8e, 0x37, 0x72, 0x29, 0xf1, 0x44, 0x26, 0x44, 0x04, 0x3b, 0x9b, 0x6f,
+	0xea, 0x4c, 0x78, 0x10, 0xf9, 0xe3, 0x31, 0x8e, 0xb6, 0x13, 0x5e, 0xbb, 0x1d, 0x88, 0x77, 0x92,
+	0x5a, 0x0a, 0x04, 0x55, 0x8d, 0xf0, 0xcb, 0x8b, 0x55, 0x52, 0x03, 0xd5, 0x53, 0x7a, 0xae, 0xda,
+	0x02, 0x3d, 0xd7, 0xb2, 0xe6, 0x21, 0x90, 0xef, 0x04, 0xea, 0xa5, 0x4e, 0xe0, 0x01, 0xb4, 0x53,
+	0x5c, 0x11, 0x45, 0x7f, 0x0b, 0xda, 0xd2, 0xac, 0x51, 0x1c, 0x09, 0xec, 0x25, 0x3d, 0xb3, 0x24,
+	0x7e, 0x2c, 0x68, 0xb1, 0xd4, 0x14, 0xb7, 0xe4, 0x4f, 0xa9, 0x61, 0x67, 0x28, 0xd6, 0xef, 0x0c,
+	0x30, 0xb3, 0x88, 0x2c, 0x24, 0x2f, 0xd2, 0x8c, 0xdf, 0x86, 0xae, 0x1c, 0x01, 0xcd, 0x60, 0x51,
+	0xb5, 0xc7, 0xc7, 0x59, 0x71, 0x03, 0xf4, 0x01, 0xac, 0x49, 0xc6, 0x12, 0x8c, 0xca, 0x36, 0xf9,
+	0xaa, 0xd8, 0xb5, 0x0b, 0x58, 0xfa, 0x0f, 0x03, 0x3a, 0xb3, 0xc2, 0x59, 0xd8, 0xaa, 0x05, 0xde,
+	0xd8, 0xe8, 0x3e, 0xb4, 0x95, 0x0d, 0xa3, 0x6c, 0xe1, 0xbf, 0xa1, 0xab, 0xba, 0x5c, 0xc4, 0xed,
+	0x56, 0x06, 0x12, 0x2f, 0x0a, 0x86, 0xd6, 0x2f, 0xa0, 0xfb, 0x7d, 0x87, 0x78, 0xf4, 0xf0, 0x30,
+	0xa9, 0xd7, 0x73, 0x14, 0xea, 0x87, 0xf9, 0x76, 0xe3, 0x39, 0x2e, 0xaf, 0xf5, 0xfb, 0x0a, 0xac,
+	0xc5, 0xb4, 0x2d, 0x27, 0x70, 0x88, 0x8b, 0x17, 0x6f, 0x32, 0x5f, 0x0c, 0x9c, 0xde, 0x82, 0x36,
+	0xa3, 0xd3, 0xc8, 0xc5, 0xa3, 0x5c, 0xaf, 0xd9, 0x92, 0xc4, 0x5d, 0x79, 0xdf, 0x6e, 0x00, 0x78,
+	0x8c, 0x27, 0x1c, 0x0a, 0x5f, 0x3d, 0xc6, 0xd5, 0xf6, 0xeb, 0xd0, 0x54, 0x32, 0x3c, 0x4a, 0xb0,
+	0x48, 0xc3, 0xaa, 0x0d, 0x92, 0x34, 0xa0, 0x44, 0xb4, 0xa5, 0xf1, 0x79, 0xb1, 0xbb, 0x22, 0x76,
+	0x57, 0x3c, 0xc6, 0xc5, 0xd6, 0x0d, 0x80, 0xc7, 0x4e, 0xe0, 0x7b, 0xa2, 0x06, 0x44, 0xeb, 0xb9,
+	0x6a, 0x37, 0x04, 0x25, 0x0e, 0x81, 0xf5, 0x1b, 0x43, 0xe2, 0x6e, 0x3e, 0x3a, 0xe7, 0x49, 0xd1,
+	0xf7, 0xf2, 0x29, 0x5a, 0xd7, 0xa5, 0x48, 0x9f, 0x07, 0x95, 0xa9, 0xf5, 0x67, 0xd0, 0xc9, 0x3f,
+	0x4c, 0x50, 0x0b, 0x56, 0x77, 0x29, 0xff, 0xe8, 0x89, 0xcf, 0xb8, 0xb9, 0x84, 0x3a, 0x00, 0xbb,
+	0x94, 0xef, 0x45, 0x98, 0x61, 0xc2, 0x4d, 0x03, 0x01, 0xd4, 0x3f, 0x25, 0x03, 0x9f, 0x7d, 0x6e,
+	0x56, 0xd0, 0x15, 0xf5, 0x92, 0x73, 0x82, 0x21, 0xf9, 0x04, 0x87, 0x34, 0x7a, 0x6a, 0x56, 0xe3,
+	0xe3, 0xe9, 0xaa, 0x86, 0x4c, 0x68, 0xa5, 0x2c, 0x3b, 0x7b, 0x3f, 0x36, 0x97, 0x51, 0x03, 0x96,
+	0xe5, 0x67, 0x7d, 0xfd, 0x53, 0x30, 0x8b, 0xc8, 0x8b, 0x9a, 0xb0, 0x72, 0x24, 0x2b, 0xd7, 0x5c,
+	0x42, 0x5d, 0x68, 0x06, 0x33, 0xeb, 0x4d, 0x23, 0x26, 0x8c, 0xa3, 0x89, 0xab, 0xc0, 0xd7, 0xac,
+	0xc4, 0xda, 0xe2, 0x04, 0x0e, 0xe8, 0x09, 0x31, 0xab, 0xeb, 0x3f, 0x80, 0x56, 0xb6, 0x1d, 0x47,
+	0xab, 0x50, 0xdb, 0xa5, 0x04, 0x9b, 0x4b, 0xb1, 0xd8, 0x9d, 0x88, 0x9e, 0xf8, 0x64, 0x2c, 0x7d,
+	0xb8, 0x1f, 0xd1, 0x67, 0x98, 0x98, 0x95, 0x78, 0x83, 0x61, 0x27, 0x88, 0x37, 0xaa, 0xf1, 0x46,
+	0xbc, 0xc0, 0x9e, 0x59, 0xdb, 0xfc, 0x0f, 0x40, 0x4b, 0x80, 0xd7, 0xbe, 0x1c, 0x12, 0xa3, 0x09,
+	0xa0, 0x1d, 0xcc, 0xb7, 0x69, 0x38, 0xa1, 0x24, 0xd1, 0xc0, 0xd0, 0x7b, 0x73, 0x06, 0xaa, 0x65,
+	0x56, 0x65, 0x74, 0xff, 0x1b, 0x73, 0x4e, 0x14, 0xd8, 0xad, 0x25, 0x14, 0x0a, 0x8d, 0xf1, 0xdf,
+	0xfe, 0x81, 0xef, 0x7e, 0x9e, 0x8c, 0x04, 0x4e, 0xd1, 0x58, 0x60, 0x4d, 0x34, 0x16, 0x6e, 0xae,
+	0x5a, 0xec, 0xf3, 0xc8, 0x27, 0xe3, 0xe4, 0x45, 0x62, 0x2d, 0xa1, 0x63, 0xb8, 0x1a, 0xbf, 0x56,
+	0xb8, 0xc3, 0x7d, 0xc6, 0x7d, 0x97, 0x25, 0x0a, 0x37, 0xe7, 0x2b, 0x2c, 0x31, 0x3f, 0xa7, 0x4a,
+	0x17, 0x5a, 0xd9, 0x81, 0x37, 0xba, 0xad, 0x2b, 0x60, 0xcd, 0x50, 0xbe, 0x7f, 0xe7, 0x6c, 0xc6,
+	0x54, 0x49, 0x00, 0xdd, 0xc2, 0x90, 0x19, 0x69, 0x2f, 0x8a, 0x7e, 0xd2, 0xdd, 0x7f, 0x67, 0x21,
+	0xde, 0x54, 0x9b, 0x0f, 0x9d, 0xfc, 0x60, 0x17, 0xbd, 0x3d, 0x4f, 0x40, 0x69, 0xa8, 0xd6, 0x5f,
+	0x5f, 0x84, 0x35, 0x55, 0xf5, 0x10, 0x3a, 0xf9, 0x29, 0xa4, 0x5e, 0x95, 0x76, 0x52, 0xd9, 0x3f,
+	0xed, 0xc5, 0x69, 0x2d, 0xa1, 0x9f, 0xc3, 0x2b, 0xa5, 0xd1, 0x1f, 0xfa, 0xa6, 0x3e, 0xea, 0xfa,
+	0x09, 0xe1, 0x59, 0x1a, 0x94, 0xf5, 0xb3, 0x28, 0xce, 0xb7, 0xbe, 0x34, 0xeb, 0x5d, 0xdc, 0xfa,
+	0x8c, 0xf8, 0xd3, 0xac, 0x7f, 0x6e, 0x0d, 0x53, 0x40, 0xe5, 0xa1, 0x20, 0x7a, 0x57, 0xa7, 0x62,
+	0xee, 0x60, 0xb2, 0xbf, 0xb1, 0x28, 0x7b, 0x9a, 0xf2, 0xa9, 0x80, 0x84, 0xe2, 0xfc, 0x4c, 0xab,
+	0x76, 0xee, 0x3c, 0x50, 0xaf, 0x76, 0xfe, 0x8c, 0x4c, 0x16, 0x75, 0x7e, 0x90, 0xa1, 0xcf, 0x95,
+	0x76, 0x70, 0xa5, 0x2f, 0x6a, 0xfd, 0x5c, 0xc4, 0x5a, 0xda, 0xfc, 0x62, 0x15, 0x1a, 0xc2, 0x79,
+	0x01, 0x08, 0xff, 0x07, 0xdd, 0x17, 0x0f, 0xba, 0x8f, 0xa0, 0x5b, 0x18, 0x07, 0xe9, 0xf1, 0x50,
+	0x3f, 0x33, 0x3a, 0xeb, 0x62, 0x1c, 0x00, 0x2a, 0xcf, 0x62, 0xf4, 0x15, 0x3a, 0x77, 0x66, 0x73,
+	0x96, 0x8e, 0x47, 0xd0, 0x2d, 0xcc, 0x42, 0xf4, 0x1e, 0xe8, 0x07, 0x26, 0x67, 0x49, 0xff, 0x0c,
+	0x5a, 0xd9, 0x57, 0xb1, 0xfe, 0xa7, 0xa4, 0x79, 0x37, 0x5f, 0x3e, 0x28, 0xbd, 0x7c, 0xd0, 0x7e,
+	0x04, 0xdd, 0xc2, 0x43, 0x58, 0x1f, 0x79, 0xfd, 0x6b, 0xf9, 0x2c, 0xe9, 0x5f, 0x1e, 0xcc, 0x6c,
+	0x7d, 0xf0, 0x70, 0x73, 0xec, 0xf3, 0xa3, 0xe9, 0x41, 0x6c, 0xc4, 0x5d, 0x79, 0xf2, 0x5d, 0x9f,
+	0xaa, 0xaf, 0xbb, 0xc9, 0x7d, 0xbb, 0x2b, 0x84, 0xdd, 0x15, 0xc2, 0x26, 0x07, 0x07, 0x75, 0xb1,
+	0x7c, 0xff, 0xbf, 0x01, 0x00, 0x00, 0xff, 0xff, 0x28, 0x13, 0x87, 0x0e, 0xa2, 0x20, 0x00, 0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.

internal/proxynode/channels_mgr.go

@@ -65,6 +65,7 @@ func getUniqueIntGeneratorIns() uniqueIntGenerator {
 }
 
 type getChannelsFuncType = func(collectionID UniqueID) (map[vChan]pChan, error)
+type repackFuncType = func(tsMsgs []msgstream.TsMsg, hashKeys [][]int32) (map[int32]*msgstream.MsgPack, error)
 
 type getChannelsService interface {
 	GetChannels(collectionID UniqueID) (map[vChan]pChan, error)
@@ -105,6 +106,13 @@ func (m *mockGetChannelsService) GetChannels(collectionID UniqueID) (map[vChan]p
 	return channels, nil
 }
 
+type streamType int
+
+const (
+	dmlStreamType streamType = iota
+	dqlStreamType
+)
+
 type singleTypeChannelsMgr struct {
 	collectionID2VIDs map[UniqueID][]int // id are sorted
 	collMtx           sync.RWMutex
@@ -121,6 +129,10 @@ type singleTypeChannelsMgr struct {
 
 	getChannelsFunc getChannelsFuncType
 
+	repackFunc repackFuncType
+
+	singleStreamType streamType
+
 	msgStreamFactory msgstream.Factory
 }
 
@@ -327,29 +339,20 @@ func (mgr *singleTypeChannelsMgr) createMsgStream(collectionID UniqueID) error {
 	vchans := getAllKeys(channels)
 	mgr.updateVChans(id, vchans)
 
-	stream, err := mgr.msgStreamFactory.NewMsgStream(context.Background())
+	var stream msgstream.MsgStream
+	if mgr.singleStreamType == dqlStreamType {
+		stream, err = mgr.msgStreamFactory.NewQueryMsgStream(context.Background())
+	} else {
+		stream, err = mgr.msgStreamFactory.NewMsgStream(context.Background())
+	}
 	if err != nil {
 		return err
 	}
 	pchans := getAllValues(channels)
 	stream.AsProducer(pchans)
-	repack := func(tsMsgs []msgstream.TsMsg, hashKeys [][]int32) (map[int32]*msgstream.MsgPack, error) {
-		// after assigning segment id to msg, tsMsgs was already re-bucketed
-		pack := make(map[int32]*msgstream.MsgPack)
-		for idx, msg := range tsMsgs {
-			if len(hashKeys[idx]) <= 0 {
-				continue
-			}
-			key := hashKeys[idx][0]
-			_, ok := pack[key]
-			if !ok {
-				pack[key] = &msgstream.MsgPack{}
-			}
-			pack[key].Msgs = append(pack[key].Msgs, msg)
-		}
-		return pack, nil
+	if mgr.repackFunc != nil {
+		stream.SetRepackFunc(mgr.repackFunc)
 	}
-	stream.SetRepackFunc(repack)
 	runtime.SetFinalizer(stream, func(stream msgstream.MsgStream) {
 		stream.Close()
 	})
@@ -398,7 +401,12 @@ func (mgr *singleTypeChannelsMgr) removeAllStream() error {
 	return nil
 }
 
-func newSingleTypeChannelsMgr(getChannelsFunc getChannelsFuncType, msgStreamFactory msgstream.Factory) *singleTypeChannelsMgr {
+func newSingleTypeChannelsMgr(
+	getChannelsFunc getChannelsFuncType,
+	msgStreamFactory msgstream.Factory,
+	repackFunc repackFuncType,
+	singleStreamType streamType,
+) *singleTypeChannelsMgr {
 	return &singleTypeChannelsMgr{
 		collectionID2VIDs:         make(map[UniqueID][]int),
 		id2vchans:                 make(map[int][]vChan),
@@ -406,6 +414,8 @@ func newSingleTypeChannelsMgr(getChannelsFunc getChannelsFuncType, msgStreamFact
 		id2UsageHistogramOfStream: make(map[int]int),
 		vchans2pchans:             make(map[vChan]pChan),
 		getChannelsFunc:           getChannelsFunc,
+		repackFunc:                repackFunc,
+		singleStreamType:          singleStreamType,
 		msgStreamFactory:          msgStreamFactory,
 	}
 }
@@ -455,9 +465,15 @@ func (mgr *channelsMgrImpl) removeAllDMLStream() error {
 	return mgr.dmlChannelsMgr.removeAllStream()
 }
 
-func newChannelsMgr(getDmlChannelsFunc getChannelsFuncType, getDqlChannelsFunc getChannelsFuncType, msgStreamFactory msgstream.Factory) channelsMgr {
+func newChannelsMgr(
+	getDmlChannelsFunc getChannelsFuncType,
+	dmlRepackFunc repackFuncType,
+	getDqlChannelsFunc getChannelsFuncType,
+	dqlRepackFunc repackFuncType,
+	msgStreamFactory msgstream.Factory,
+) channelsMgr {
 	return &channelsMgrImpl{
-		dmlChannelsMgr: newSingleTypeChannelsMgr(getDmlChannelsFunc, msgStreamFactory),
-		dqlChannelsMgr: newSingleTypeChannelsMgr(getDqlChannelsFunc, msgStreamFactory),
+		dmlChannelsMgr: newSingleTypeChannelsMgr(getDmlChannelsFunc, msgStreamFactory, dmlRepackFunc, dmlStreamType),
+		dqlChannelsMgr: newSingleTypeChannelsMgr(getDqlChannelsFunc, msgStreamFactory, dqlRepackFunc, dqlStreamType),
 	}
 }

internal/proxynode/channels_mgr_test.go

@@ -26,7 +26,7 @@ func TestChannelsMgrImpl_getChannels(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	collID := UniqueID(getUniqueIntGeneratorIns().get())
@@ -44,7 +44,7 @@ func TestChannelsMgrImpl_getVChannels(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	collID := UniqueID(getUniqueIntGeneratorIns().get())
@@ -62,7 +62,7 @@ func TestChannelsMgrImpl_createDMLMsgStream(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	collID := UniqueID(getUniqueIntGeneratorIns().get())
@@ -84,7 +84,7 @@ func TestChannelsMgrImpl_getDMLMsgStream(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	collID := UniqueID(getUniqueIntGeneratorIns().get())
@@ -102,7 +102,7 @@ func TestChannelsMgrImpl_removeDMLMsgStream(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	collID := UniqueID(getUniqueIntGeneratorIns().get())
@@ -129,7 +129,7 @@ func TestChannelsMgrImpl_removeAllDMLMsgStream(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	num := 10
@@ -144,7 +144,7 @@ func TestChannelsMgrImpl_createDQLMsgStream(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	collID := UniqueID(getUniqueIntGeneratorIns().get())
@@ -166,7 +166,7 @@ func TestChannelsMgrImpl_getDQLMsgStream(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	collID := UniqueID(getUniqueIntGeneratorIns().get())
@@ -184,7 +184,7 @@ func TestChannelsMgrImpl_removeDQLMsgStream(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	collID := UniqueID(getUniqueIntGeneratorIns().get())
@@ -211,7 +211,7 @@ func TestChannelsMgrImpl_removeAllDQLMsgStream(t *testing.T) {
 	master := newMockGetChannelsService()
 	query := newMockGetChannelsService()
 	factory := msgstream.NewSimpleMsgStreamFactory()
-	mgr := newChannelsMgr(master.GetChannels, query.GetChannels, factory)
+	mgr := newChannelsMgr(master.GetChannels, nil, query.GetChannels, nil, factory)
 	defer mgr.removeAllDMLStream()
 
 	num := 10

internal/proxynode/impl.go

@@ -91,6 +91,25 @@ func (node *ProxyNode) InvalidateCollectionMetaCache(ctx context.Context, reques
 	}, nil
 }
 
+func (node *ProxyNode) ReleaseDQLMessageStream(ctx context.Context, request *proxypb.ReleaseDQLMessageStreamRequest) (*commonpb.Status, error) {
+	log.Debug("ReleaseDQLMessageStream",
+		zap.Any("role", Params.RoleName),
+		zap.Any("db", request.DbID),
+		zap.Any("collection", request.CollectionID))
+
+	_ = node.chMgr.removeDQLStream(request.CollectionID)
+
+	log.Debug("ReleaseDQLMessageStream Done",
+		zap.Any("role", Params.RoleName),
+		zap.Any("db", request.DbID),
+		zap.Any("collection", request.CollectionID))
+
+	return &commonpb.Status{
+		ErrorCode: commonpb.ErrorCode_Success,
+		Reason:    "",
+	}, nil
+}
+
 func (node *ProxyNode) CreateCollection(ctx context.Context, request *milvuspb.CreateCollectionRequest) (*commonpb.Status, error) {
 	cct := &CreateCollectionTask{
 		ctx:                     ctx,
@@ -279,6 +298,7 @@ func (node *ProxyNode) ReleaseCollection(ctx context.Context, request *milvuspb.
 		Condition:                NewTaskCondition(ctx),
 		ReleaseCollectionRequest: request,
 		queryService:             node.queryService,
+		chMgr:                    node.chMgr,
 	}
 
 	err := node.sched.DdQueue.Enqueue(rct)
@@ -1128,10 +1148,9 @@ func (node *ProxyNode) Search(ctx context.Context, request *milvuspb.SearchReque
 			},
 			ResultChannelID: strconv.FormatInt(Params.ProxyID, 10),
 		},
-		queryMsgStream: node.queryMsgStream,
-		resultBuf:      make(chan []*internalpb.SearchResults),
-		query:          request,
-		chMgr:          node.chMgr,
+		resultBuf: make(chan []*internalpb.SearchResults),
+		query:     request,
+		chMgr:     node.chMgr,
 	}
 
 	err := node.sched.DqQueue.Enqueue(qt)
@@ -1201,9 +1220,8 @@ func (node *ProxyNode) Retrieve(ctx context.Context, request *milvuspb.RetrieveR
 			},
 			ResultChannelID: strconv.FormatInt(Params.ProxyID, 10),
 		},
-		queryMsgStream: node.queryMsgStream,
-		resultBuf:      make(chan []*internalpb.RetrieveResults),
-		retrieve:       request,
+		resultBuf: make(chan []*internalpb.RetrieveResults),
+		retrieve:  request,
 	}
 
 	err := node.sched.DqQueue.Enqueue(rt)
@@ -1352,10 +1370,9 @@ func (node *ProxyNode) Query(ctx context.Context, request *milvuspb.QueryRequest
 				},
 				ResultChannelID: strconv.FormatInt(Params.ProxyID, 10),
 			},
-			queryMsgStream: node.queryMsgStream,
-			resultBuf:      make(chan []*internalpb.RetrieveResults),
-			retrieve:       retrieveRequest,
-			chMgr:          node.chMgr,
+			resultBuf: make(chan []*internalpb.RetrieveResults),
+			retrieve:  retrieveRequest,
+			chMgr:     node.chMgr,
 		}
 
 		err := node.sched.DqQueue.Enqueue(rt)

internal/proxynode/paramtable.go

@@ -44,9 +44,7 @@ type ParamTable struct {
 	ProxyID                    UniqueID
 	TimeTickInterval           time.Duration
 	K2SChannelNames            []string
-	SearchChannelNames         []string
 	SearchResultChannelNames   []string
-	RetrieveChannelNames       []string
 	RetrieveResultChannelNames []string
 	ProxySubName               string
 	ProxyTimeTickChannelNames  []string

internal/proxynode/proxy_node.go

@@ -70,8 +70,7 @@ type ProxyNode struct {
 
 	session *sessionutil.Session
 
-	queryMsgStream msgstream.MsgStream
-	msFactory      msgstream.Factory
+	msFactory msgstream.Factory
 
 	// Add callback functions at different stages
 	startCallbacks []func()
@@ -147,13 +146,9 @@ func (node *ProxyNode) Init() error {
 		}
 		log.Debug("ProxyNode CreateQueryChannel success")
 
-		Params.SearchChannelNames = []string{resp.RequestChannel}
 		Params.SearchResultChannelNames = []string{resp.ResultChannel}
-		Params.RetrieveChannelNames = []string{resp.RequestChannel}
 		Params.RetrieveResultChannelNames = []string{resp.ResultChannel}
-		log.Debug("ProxyNode CreateQueryChannel success", zap.Any("SearchChannelNames", Params.SearchChannelNames))
 		log.Debug("ProxyNode CreateQueryChannel success", zap.Any("SearchResultChannelNames", Params.SearchResultChannelNames))
-		log.Debug("ProxyNode CreateQueryChannel success", zap.Any("RetrieveChannelNames", Params.RetrieveChannelNames))
 		log.Debug("ProxyNode CreateQueryChannel success", zap.Any("RetrieveResultChannelNames", Params.RetrieveResultChannelNames))
 	}
 
@@ -171,12 +166,6 @@ func (node *ProxyNode) Init() error {
 		return err
 	}
 
-	node.queryMsgStream, _ = node.msFactory.NewQueryMsgStream(node.ctx)
-	node.queryMsgStream.AsProducer(Params.SearchChannelNames)
-	// FIXME(wxyu): use log.Debug instead
-	log.Debug("proxynode", zap.Strings("proxynode AsProducer:", Params.SearchChannelNames))
-	log.Debug("create query message stream ...")
-
 	idAllocator, err := allocator.NewIDAllocator(node.ctx, Params.MetaRootPath, Params.EtcdEndpoints)
 
 	if err != nil {
@@ -245,9 +234,26 @@ func (node *ProxyNode) Init() error {
 
 		return ret, nil
 	}
-	mockQueryService := newMockGetChannelsService()
+	getDqlChannelsFunc := func(collectionID UniqueID) (map[vChan]pChan, error) {
+		req := &querypb.CreateQueryChannelRequest{
+			CollectionID: collectionID,
+			ProxyID:      node.session.ServerID,
+		}
+		resp, err := node.queryService.CreateQueryChannel(node.ctx, req)
+		if err != nil {
+			return nil, err
+		}
+		if resp.Status.ErrorCode != commonpb.ErrorCode_Success {
+			return nil, errors.New(resp.Status.Reason)
+		}
 
-	chMgr := newChannelsMgr(getDmlChannelsFunc, mockQueryService.GetChannels, node.msFactory)
+		m := make(map[vChan]pChan)
+		m[resp.RequestChannel] = resp.RequestChannel
+
+		return m, nil
+	}
+
+	chMgr := newChannelsMgr(getDmlChannelsFunc, defaultInsertRepackFunc, getDqlChannelsFunc, nil, node.msFactory)
 	node.chMgr = chMgr
 
 	node.sched, err = NewTaskScheduler(node.ctx, node.idAllocator, node.tsoAllocator, node.msFactory)
@@ -324,12 +330,6 @@ func (node *ProxyNode) Start() error {
 	}
 	log.Debug("init global meta cache ...")
 
-	initGlobalInsertChannelsMap(node)
-	log.Debug("init global insert channels map ...")
-
-	node.queryMsgStream.Start()
-	log.Debug("start query message stream ...")
-
 	node.sched.Start()
 	log.Debug("start scheduler ...")
 
@@ -364,11 +364,9 @@ func (node *ProxyNode) Start() error {
 func (node *ProxyNode) Stop() error {
 	node.cancel()
 
-	globalInsertChannelsMap.CloseAllMsgStream()
 	node.idAllocator.Close()
 	node.segAssigner.Close()
 	node.sched.Close()
-	node.queryMsgStream.Close()
 	node.tick.Close()
 	err := node.chTicker.close()
 	if err != nil {

internal/proxynode/repack_func.go

@@ -32,3 +32,20 @@ func insertRepackFunc(tsMsgs []msgstream.TsMsg,
 	}
 	return result, nil
 }
+
+func defaultInsertRepackFunc(tsMsgs []msgstream.TsMsg, hashKeys [][]int32) (map[int32]*msgstream.MsgPack, error) {
+	// after assigning segment id to msg, tsMsgs was already re-bucketed
+	pack := make(map[int32]*msgstream.MsgPack)
+	for idx, msg := range tsMsgs {
+		if len(hashKeys[idx]) <= 0 {
+			continue
+		}
+		key := hashKeys[idx][0]
+		_, ok := pack[key]
+		if !ok {
+			pack[key] = &msgstream.MsgPack{}
+		}
+		pack[key].Msgs = append(pack[key].Msgs, msg)
+	}
+	return pack, nil
+}

internal/proxynode/task.go

@@ -972,6 +972,7 @@ func (dct *DropCollectionTask) Execute(ctx context.Context) error {
 	}
 
 	_ = dct.chMgr.removeDMLStream(collID)
+	_ = dct.chMgr.removeDQLStream(collID)
 
 	return nil
 }
@@ -984,12 +985,11 @@ func (dct *DropCollectionTask) PostExecute(ctx context.Context) error {
 type SearchTask struct {
 	Condition
 	*internalpb.SearchRequest
-	ctx            context.Context
-	queryMsgStream msgstream.MsgStream
-	resultBuf      chan []*internalpb.SearchResults
-	result         *milvuspb.SearchResults
-	query          *milvuspb.SearchRequest
-	chMgr          channelsMgr
+	ctx       context.Context
+	resultBuf chan []*internalpb.SearchResults
+	result    *milvuspb.SearchResults
+	query     *milvuspb.SearchRequest
+	chMgr     channelsMgr
 }
 
 func (st *SearchTask) TraceCtx() context.Context {
@@ -1183,7 +1183,29 @@ func (st *SearchTask) Execute(ctx context.Context) error {
 		Msgs:    make([]msgstream.TsMsg, 1),
 	}
 	msgPack.Msgs[0] = tsMsg
-	err := st.queryMsgStream.Produce(&msgPack)
+
+	collectionName := st.query.CollectionName
+	collID, err := globalMetaCache.GetCollectionID(ctx, collectionName)
+	if err != nil { // err is not nil if collection not exists
+		return err
+	}
+
+	stream, err := st.chMgr.getDQLStream(collID)
+	if err != nil {
+		err = st.chMgr.createDQLStream(collID)
+		if err != nil {
+			st.result.Status.ErrorCode = commonpb.ErrorCode_UnexpectedError
+			st.result.Status.Reason = err.Error()
+			return err
+		}
+		stream, err = st.chMgr.getDQLStream(collID)
+		if err != nil {
+			st.result.Status.ErrorCode = commonpb.ErrorCode_UnexpectedError
+			st.result.Status.Reason = err.Error()
+			return err
+		}
+	}
+	err = stream.Produce(&msgPack)
 	log.Debug("proxynode", zap.Int("length of searchMsg", len(msgPack.Msgs)))
 	log.Debug("proxy node sent one searchMsg",
 		zap.Any("collectionID", st.CollectionID),
@@ -1461,12 +1483,11 @@ func (st *SearchTask) PostExecute(ctx context.Context) error {
 type RetrieveTask struct {
 	Condition
 	*internalpb.RetrieveRequest
-	ctx            context.Context
-	queryMsgStream msgstream.MsgStream
-	resultBuf      chan []*internalpb.RetrieveResults
-	result         *milvuspb.RetrieveResults
-	retrieve       *milvuspb.RetrieveRequest
-	chMgr          channelsMgr
+	ctx       context.Context
+	resultBuf chan []*internalpb.RetrieveResults
+	result    *milvuspb.RetrieveResults
+	retrieve  *milvuspb.RetrieveRequest
+	chMgr     channelsMgr
 }
 
 func (rt *RetrieveTask) TraceCtx() context.Context {
@@ -1584,7 +1605,7 @@ func (rt *RetrieveTask) PreExecute(ctx context.Context) error {
 		rt.OutputFields = rt.retrieve.OutputFields
 	}
 
-	rt.ResultChannelID = Params.RetrieveChannelNames[0]
+	rt.ResultChannelID = Params.RetrieveResultChannelNames[0]
 	rt.DbID = 0 // todo(yukun)
 
 	rt.CollectionID = collectionID
@@ -1646,7 +1667,29 @@ func (rt *RetrieveTask) Execute(ctx context.Context) error {
 		Msgs:    make([]msgstream.TsMsg, 1),
 	}
 	msgPack.Msgs[0] = tsMsg
-	err := rt.queryMsgStream.Produce(&msgPack)
+
+	collectionName := rt.retrieve.CollectionName
+	collID, err := globalMetaCache.GetCollectionID(ctx, collectionName)
+	if err != nil {
+		return err
+	}
+
+	stream, err := rt.chMgr.getDQLStream(collID)
+	if err != nil {
+		err = rt.chMgr.createDQLStream(collID)
+		if err != nil {
+			rt.result.Status.ErrorCode = commonpb.ErrorCode_UnexpectedError
+			rt.result.Status.Reason = err.Error()
+			return err
+		}
+		stream, err = rt.chMgr.getDQLStream(collID)
+		if err != nil {
+			rt.result.Status.ErrorCode = commonpb.ErrorCode_UnexpectedError
+			rt.result.Status.Reason = err.Error()
+			return err
+		}
+	}
+	err = stream.Produce(&msgPack)
 	log.Debug("proxynode", zap.Int("length of retrieveMsg", len(msgPack.Msgs)))
 	if err != nil {
 		log.Debug("Failed to send retrieve request.",
@@ -3415,6 +3458,7 @@ type ReleaseCollectionTask struct {
 	ctx          context.Context
 	queryService types.QueryService
 	result       *commonpb.Status
+	chMgr        channelsMgr
 }
 
 func (rct *ReleaseCollectionTask) TraceCtx() context.Context {
@@ -3482,7 +3526,11 @@ func (rct *ReleaseCollectionTask) Execute(ctx context.Context) (err error) {
 		DbID:         0,
 		CollectionID: collID,
 	}
+
 	rct.result, err = rct.queryService.ReleaseCollection(ctx, request)
+
+	_ = rct.chMgr.removeDQLStream(collID)
+
 	return err
 }
 

internal/proxynode/task_scheduler.go

@@ -524,7 +524,7 @@ func (sched *TaskScheduler) queryLoop() {
 
 type resultBufHeader struct {
 	usedVChans                  map[interface{}]struct{} // set of vChan
-	usedChans                   map[interface{}]struct{} // set of Chan todo
+	usedChans                   map[interface{}]struct{} // set of pChan todo
 	receivedVChansSet           map[interface{}]struct{} // set of vChan
 	receivedSealedSegmentIDsSet map[interface{}]struct{} // set of UniqueID
 	receivedGlobalSegmentIDsSet map[interface{}]struct{} // set of UniqueID
@@ -699,6 +699,7 @@ func (sched *TaskScheduler) collectResultLoop() {
 			if msgPack == nil {
 				continue
 			}
+
 			for _, tsMsg := range msgPack.Msgs {
 				sp, ctx := trace.StartSpanFromContext(tsMsg.TraceCtx())
 				tsMsg.SetTraceCtx(ctx)