furyhawk/milvus
1
0
Fork 0
mirror of https://github.com/milvus-io/milvus.git synced 2026-07-21 10:15:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

enhance: [2.6] simplify seek position selection in WatchDmChannels (#46568)

Browse Source 
issue: #46566
pr: #46567

Remove the complex comparison logic between seekPosition and
deleteCheckpoint. Use seekPosition directly since:

- L0 segments are loaded before consuming message stream, which contain
delete records from [deleteCheckpoint, L0.endPosition]
- DataCoord ensures seekPosition is based on channel checkpoint, updated
after data (including deletes) is flushed
- L0 segments should cover up to seekPosition, avoiding data loss
- This eliminates redundant message consumption when seekPosition >
deleteCheckpoint

Signed-off-by: Wei Liu <wei.liu@zilliz.com>
This commit is contained in:
wei liu
2025-12-30 16:59:22 +08:00
committed by Li Liu
parent 8cbb4d7d36
commit a71a1b8181
1 changed files with 19 additions and 40 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/querynodev2/services.go
+19 -40
View File
@@ -332,46 +332,25 @@ func (node *QueryNode) WatchDmChannels(ctx context.Context, req *querypb.WatchDm
return merr.Status(err), nil
}
var position *msgpb.MsgPosition
deleteCheckpoint := channel.GetDeleteCheckpoint()
channelCheckpoint := channel.GetSeekPosition()
if deleteCheckpoint == nil {
// for compatibility with old version coord, which doesn't have delete checkpoint in VchannelInfo
log.Info("no delete checkpoint found, use seek position to seek",
zap.Time("seekPosition", tsoutil.PhysicalTime(channelCheckpoint.GetTimestamp())),
)
position = &msgpb.MsgPosition{
ChannelName: channelCheckpoint.GetChannelName(),
MsgID: channelCheckpoint.GetMsgID(),
Timestamp: channelCheckpoint.GetTimestamp(),
}
} else {
if channelCheckpoint.GetTimestamp() > deleteCheckpoint.GetTimestamp() {
msg := "channel seek position is greater than delete checkpoint, use delete checkpoint to seek"
log.Info(msg,
zap.Time("seekPosition", tsoutil.PhysicalTime(channelCheckpoint.GetTimestamp())),
zap.Time("deleteCheckpoint", tsoutil.PhysicalTime(deleteCheckpoint.GetTimestamp())),
)
position = &msgpb.MsgPosition{
ChannelName: deleteCheckpoint.GetChannelName(),
MsgID: deleteCheckpoint.GetMsgID(),
Timestamp: deleteCheckpoint.GetTimestamp(),
}
} else {
msg := "channel seek position is smaller than delete checkpoint, use seek position to seek"
log.Info(msg,
zap.Time("seekPosition", tsoutil.PhysicalTime(channelCheckpoint.GetTimestamp())),
zap.Time("deleteCheckpoint", tsoutil.PhysicalTime(deleteCheckpoint.GetTimestamp())),
)
position = &msgpb.MsgPosition{
ChannelName: channelCheckpoint.GetChannelName(),
MsgID: channelCheckpoint.GetMsgID(),
Timestamp: channelCheckpoint.GetTimestamp(),
}
}
}
err = pipeline.ConsumeMsgStream(ctx, position)
// Use seekPosition directly to start consuming the message stream.
//
// Background:
// - seekPosition: channel checkpoint from DataCoord, represents the position where data has been persisted
// - deleteCheckpoint: the minimum startPosition among all L0 segments, indicates where unpersisted
// delete records begin
//
// Why we can use seekPosition directly:
// - L0 segments have already been loaded above (loadL0Segments), which contain delete records
// from [deleteCheckpoint, L0.endPosition]
// - The message stream will capture new delete records from [seekPosition, ∞)
// - DataCoord ensures that seekPosition is calculated based on channel checkpoint, which is updated
// after data (including deletes) is flushed, so L0 segments should cover up to seekPosition
// - Using seekPosition avoids redundant message consumption when seekPosition > deleteCheckpoint
log.Info("use channel seek position to seek",
zap.Time("seekPosition", tsoutil.PhysicalTime(channel.GetSeekPosition().GetTimestamp())),
zap.Time("deleteCheckpoint", tsoutil.PhysicalTime(channel.GetDeleteCheckpoint().GetTimestamp())),
)
err = pipeline.ConsumeMsgStream(ctx, channel.GetSeekPosition())
if err != nil {
err = merr.WrapErrServiceUnavailable(err.Error(), "InitPipelineFailed")
log.Warn(err.Error(),