enhance: [2.4] Batch forward delete when using DirectForward (#37076)

internal/distributed/querynode/client/client.go

@@ -332,3 +332,16 @@ func (c *Client) Delete(ctx context.Context, req *querypb.DeleteRequest, _ ...gr
 		return client.Delete(ctx, req)
 	})
 }
+
+// DeleteBatch is the API to apply same delete data into multiple segments.
+// it's basically same as `Delete` but cost less memory pressure.
+func (c *Client) DeleteBatch(ctx context.Context, req *querypb.DeleteBatchRequest, _ ...grpc.CallOption) (*querypb.DeleteBatchResponse, error) {
+	req = typeutil.Clone(req)
+	commonpbutil.UpdateMsgBase(
+		req.GetBase(),
+		commonpbutil.FillMsgBaseFromClient(c.nodeID),
+	)
+	return wrapGrpcCall(ctx, c, func(client querypb.QueryNodeClient) (*querypb.DeleteBatchResponse, error) {
+		return client.DeleteBatch(ctx, req)
+	})
+}

internal/distributed/querynode/client/client_test.go

@@ -108,6 +108,9 @@ func Test_NewClient(t *testing.T) {
 		r20, err := client.SearchSegments(ctx, nil)
 		retCheck(retNotNil, r20, err)
 
+		r21, err := client.DeleteBatch(ctx, nil)
+		retCheck(retNotNil, r21, err)
+
 		// stream rpc
 		client, err := client.QueryStream(ctx, nil)
 		retCheck(retNotNil, client, err)

internal/distributed/querynode/service.go

@@ -383,3 +383,9 @@ func (s *Server) SyncDistribution(ctx context.Context, req *querypb.SyncDistribu
 func (s *Server) Delete(ctx context.Context, req *querypb.DeleteRequest) (*commonpb.Status, error) {
 	return s.querynode.Delete(ctx, req)
 }
+
+// DeleteBatch is the API to apply same delete data into multiple segments.
+// it's basically same as `Delete` but cost less memory pressure.
+func (s *Server) DeleteBatch(ctx context.Context, req *querypb.DeleteBatchRequest) (*querypb.DeleteBatchResponse, error) {
+	return s.querynode.DeleteBatch(ctx, req)
+}

internal/distributed/querynode/service_test.go

@@ -32,6 +32,7 @@ import (
 	"github.com/milvus-io/milvus/internal/proto/internalpb"
 	"github.com/milvus-io/milvus/internal/proto/querypb"
 	"github.com/milvus-io/milvus/internal/types"
+	"github.com/milvus-io/milvus/pkg/util/merr"
 	"github.com/milvus-io/milvus/pkg/util/paramtable"
 )
 
@@ -270,6 +271,15 @@ func Test_NewServer(t *testing.T) {
 		assert.Equal(t, commonpb.ErrorCode_Success, resp.GetStatus().GetErrorCode())
 	})
 
+	t.Run("DeleteBatch", func(t *testing.T) {
+		mockQN.EXPECT().DeleteBatch(mock.Anything, mock.Anything).Return(&querypb.DeleteBatchResponse{
+			Status: merr.Success(),
+		}, nil)
+
+		resp, err := server.DeleteBatch(ctx, &querypb.DeleteBatchRequest{})
+		assert.NoError(t, merr.CheckRPCCall(resp, err))
+	})
+
 	err = server.Stop()
 	assert.NoError(t, err)
 }

internal/proto/query_coord.proto

@@ -172,6 +172,10 @@ service QueryNode {
     }
     rpc Delete(DeleteRequest) returns (common.Status) {
     }
+    // DeleteBatch is the API to apply same delete data into multiple segments.
+    // it's basically same as `Delete` but cost less memory pressure.
+    rpc DeleteBatch(DeleteBatchRequest) returns (DeleteBatchResponse) {
+    } 
 }
 
 // --------------------QueryCoord grpc request and response proto------------------
@@ -772,6 +776,25 @@ message DeleteRequest {
     DataScope scope = 8;
 }
 
+message DeleteBatchRequest {
+    common.MsgBase base = 1;
+    int64 collection_id = 2;
+    int64 partition_id = 3;
+    string vchannel_name = 4;
+    repeated int64 segment_ids = 5;
+    schema.IDs primary_keys = 6;
+    repeated uint64 timestamps = 7;
+    DataScope scope = 8;
+}
+
+// DeleteBatchResponse returns failed/missing segment ids
+// cannot just using common.Status to handle partial failure logic
+message DeleteBatchResponse {
+    common.Status status = 1;
+    repeated int64 failed_ids = 2;
+    repeated int64 missing_ids = 3;
+}
+
 message ActivateCheckerRequest {
     common.MsgBase base = 1;
     int32 checkerID = 2;