Add seek position to WatchDmChannels response (#5601)

Signed-off-by: sunby <bingyi.sun@zilliz.com>

internal/dataservice/binlog_helper.go

@@ -184,9 +184,14 @@ func (s *Server) GetVChanPositions(vchans []vchannel) ([]*datapb.VchannelInfo, e
 		segments := s.meta.GetSegmentsByChannel(vchan.DmlChannel)
 		flushedSegmentIDs := make([]UniqueID, 0)
 		unflushedCheckpoints := make([]*datapb.CheckPoint, 0)
+		var seekPosition *internalpb.MsgPosition
+		var useUnflushedPosition bool
 		for _, s := range segments {
 			if s.State == commonpb.SegmentState_Flushing || s.State == commonpb.SegmentState_Flushed {
 				flushedSegmentIDs = append(flushedSegmentIDs, s.ID)
+				if seekPosition == nil || (!useUnflushedPosition && s.DmlPosition.Timestamp > seekPosition.Timestamp) {
+					seekPosition = s.DmlPosition
+				}
 				continue
 			}
 
@@ -200,13 +205,19 @@ func (s *Server) GetVChanPositions(vchans []vchannel) ([]*datapb.VchannelInfo, e
 				NumOfRows: s.NumOfRows,
 			}
 			unflushedCheckpoints = append(unflushedCheckpoints, cp)
+
+			if seekPosition == nil || (useUnflushedPosition && s.DmlPosition.Timestamp < seekPosition.Timestamp) {
+				useUnflushedPosition = true
+				seekPosition = s.DmlPosition
+			}
 		}
 
 		pairs = append(pairs, &datapb.VchannelInfo{
 			CollectionID:    vchan.CollectionID,
 			ChannelName:     vchan.DmlChannel,
-			FlushedSegments: flushedSegmentIDs,
+			SeekPosition:    seekPosition,
 			CheckPoints:     unflushedCheckpoints,
+			FlushedSegments: flushedSegmentIDs,
 		})
 	}
 	return pairs, nil

internal/dataservice/cluster.go

@@ -117,7 +117,7 @@ func (c *cluster) watch(nodes []*datapb.DataNodeInfo) []*datapb.DataNodeInfo {
 				})
 			}
 		}
-		_, err := c.posProvider.GetVChanPositions(uncompletes)
+		vchanInfos, err := c.posProvider.GetVChanPositions(uncompletes)
 		if err != nil {
 			log.Warn("get vchannel position failed", zap.Error(err))
 			continue
@@ -131,8 +131,7 @@ func (c *cluster) watch(nodes []*datapb.DataNodeInfo) []*datapb.DataNodeInfo {
 			Base: &commonpb.MsgBase{
 				SourceID: Params.NodeID,
 			},
-			// TODO change to VChannel info
-			// Vchannels: pairs,
+			Vchannels: vchanInfos,
 		}
 		resp, err := cli.WatchDmChannels(c.ctx, req)
 		if err != nil {

internal/dataservice/meta.go

@@ -255,6 +255,10 @@ func (m *meta) SaveBinlogAndCheckPoints(segID UniqueID, flushed bool,
 			log.Warn("Failed to find segment", zap.Int64("id", cp.SegmentID))
 			continue
 		}
+		if segment.DmlPosition != nil && segment.DmlPosition.Timestamp >= cp.Position.Timestamp {
+			// segment position in etcd is larger than checkpoint, then dont change it
+			continue
+		}
 		segment.DmlPosition = cp.Position
 		segment.NumOfRows = cp.NumOfRows
 		segBytes := proto.MarshalTextString(segInfo)

internal/proto/data_service.proto

@@ -173,8 +173,9 @@ message GetSegmentInfoChannelRequest {
 message VchannelInfo {
   int64 collectionID = 1;
   string channelName = 2;
-  repeated CheckPoint checkPoints = 3;
-  repeated int64 flushedSegments = 4;
+  internal.MsgPosition seek_position = 3;
+  repeated CheckPoint checkPoints = 4;
+  repeated int64 flushedSegments = 5;
 }
 
 message WatchDmChannelsRequest {