Change SaveBinLogPath proto & func (#5551)

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>

internal/dataservice/grpc_services.go

@@ -316,8 +316,8 @@ func (s *Server) SaveBinlogPaths(ctx context.Context, req *datapb.SaveBinlogPath
 	}
 
 	// set segment to SegmentState_Flushing
-	err = s.meta.FlushSegmentWithBinlogAndPos(req.SegmentID, req.DmlPosition,
-		req.DdlPosition, meta)
+	err = s.meta.FlushSegmentWithBinlogAndPos(req.SegmentID, nil,
+		nil, meta)
 	if err != nil {
 		resp.Reason = err.Error()
 		return resp, nil

internal/dataservice/server_test.go

@@ -499,34 +499,6 @@ func TestSaveBinlogPaths(t *testing.T) {
 					TsBinlogPath:  "/by-dev/test/0/ts/Allo8",
 				},
 			},
-			DmlPosition: &datapb.PositionPair{
-				StartPosition: &internalpb.MsgPosition{
-					ChannelName: "ch1",
-					MsgID:       []byte{1, 2, 3},
-					MsgGroup:    "",
-					Timestamp:   0,
-				},
-				EndPosition: &internalpb.MsgPosition{
-					ChannelName: "ch1",
-					MsgID:       []byte{3, 4, 5},
-					MsgGroup:    "",
-					Timestamp:   0,
-				},
-			},
-			DdlPosition: &datapb.PositionPair{
-				StartPosition: &internalpb.MsgPosition{
-					ChannelName: "ch2",
-					MsgID:       []byte{1, 2, 3},
-					MsgGroup:    "",
-					Timestamp:   0,
-				},
-				EndPosition: &internalpb.MsgPosition{
-					ChannelName: "ch2",
-					MsgID:       []byte{3, 4, 5},
-					MsgGroup:    "",
-					Timestamp:   0,
-				},
-			},
 		})
 		assert.Nil(t, err)
 		assert.EqualValues(t, resp.ErrorCode, commonpb.ErrorCode_Success)
@@ -814,34 +786,6 @@ func TestGetVChannelPos(t *testing.T) {
 		CollectionID:      0,
 		Field2BinlogPaths: []*datapb.ID2PathList{},
 		DdlBinlogPaths:    []*datapb.DDLBinlogMeta{},
-		DmlPosition: &datapb.PositionPair{
-			StartPosition: &internalpb.MsgPosition{
-				ChannelName: "ch1",
-				MsgID:       []byte{1, 2, 3},
-				MsgGroup:    "",
-				Timestamp:   0,
-			},
-			EndPosition: &internalpb.MsgPosition{
-				ChannelName: "ch1",
-				MsgID:       []byte{3, 4, 5},
-				MsgGroup:    "",
-				Timestamp:   0,
-			},
-		},
-		DdlPosition: &datapb.PositionPair{
-			StartPosition: &internalpb.MsgPosition{
-				ChannelName: "ch2",
-				MsgID:       []byte{1, 2, 3},
-				MsgGroup:    "",
-				Timestamp:   0,
-			},
-			EndPosition: &internalpb.MsgPosition{
-				ChannelName: "ch2",
-				MsgID:       []byte{3, 4, 5},
-				MsgGroup:    "",
-				Timestamp:   0,
-			},
-		},
 	}
 	status, err := svr.SaveBinlogPaths(context.TODO(), req)
 	assert.Nil(t, err)
@@ -874,10 +818,10 @@ func TestGetVChannelPos(t *testing.T) {
 		assert.Nil(t, err)
 		assert.EqualValues(t, 1, len(pair))
 		assert.EqualValues(t, 0, pair[0].CollectionID)
-		assert.EqualValues(t, []byte{1, 2, 3}, pair[0].DmlPosition.StartPosition.MsgID)
-		assert.EqualValues(t, []byte{3, 4, 5}, pair[0].DmlPosition.EndPosition.MsgID)
-		assert.EqualValues(t, []byte{1, 2, 3}, pair[0].DdlPosition.StartPosition.MsgID)
-		assert.EqualValues(t, []byte{3, 4, 5}, pair[0].DdlPosition.EndPosition.MsgID)
+		// assert.EqualValues(t, []byte{1, 2, 3}, pair[0].DmlPosition.StartPosition.MsgID)
+		// assert.EqualValues(t, []byte{3, 4, 5}, pair[0].DmlPosition.EndPosition.MsgID)
+		// assert.EqualValues(t, []byte{1, 2, 3}, pair[0].DdlPosition.StartPosition.MsgID)
+		// assert.EqualValues(t, []byte{3, 4, 5}, pair[0].DdlPosition.EndPosition.MsgID)
 	})
 }
 

internal/proto/data_service.proto

@@ -256,8 +256,15 @@ message SaveBinlogPathsRequest {
   int64 collectionID = 3;
   repeated ID2PathList field2BinlogPaths = 4;
   repeated DDLBinlogMeta ddlBinlogPaths = 5;
-  PositionPair dml_position = 6;
-  PositionPair ddl_position =7;
+  repeated CheckPoint checkPoints = 6;
+  bool flushed = 7;
+}
+
+message CheckPoint {
+  int64 segmentID = 1;
+  int64 collectionID = 2;
+  internal.MsgPosition position = 3;
+  int64 rows = 4;
 }
 
 message DataNodeTtMsg {
@@ -266,11 +273,6 @@ message DataNodeTtMsg {
     uint64 timestamp = 3;
 }
 
-message CheckPoint {
-  int64 segmentID = 1;
-  internal.MsgPosition position = 2;
-  int64 num_of_rows = 3;
-}
 
 message VchannelInfo {
   int64 collectionID = 1;