milvus-io
/
milvus
mirror of https://github.com/milvus-io/milvus.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

fix: set channel checkpoint to delta position (#32878) 

issue: https://github.com/milvus-io/milvus/issues/32853

---------

Signed-off-by: sunby <sunbingyi1992@gmail.com>
pull/32931/head
Bingyi Sun 2024-05-10 11:51:30 +08:00 committed by GitHub
parent 9a269f1489
commit b7ef8da360
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 4 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

11
internal/querycoordv2/utils/types.go
View File

@ -60,19 +60,14 @@ func MergeMetaSegmentIntoSegmentInfo(info *querypb.SegmentInfo, segments ...*met
// packSegmentLoadInfo packs SegmentLoadInfo for given segment,
// packs with index if withIndex is true, this fetch indexes from IndexCoord
func PackSegmentLoadInfo(segment *datapb.SegmentInfo, channelCheckpoint *msgpb.MsgPosition, indexes []*querypb.FieldIndexInfo) *querypb.SegmentLoadInfo {
checkpoint := segment.GetDmlPosition()
if channelCheckpoint.GetTimestamp() > checkpoint.GetTimestamp() {
checkpoint = channelCheckpoint
}
posTime := tsoutil.PhysicalTime(checkpoint.GetTimestamp())
posTime := tsoutil.PhysicalTime(channelCheckpoint.GetTimestamp())
tsLag := time.Since(posTime)
if tsLag >= 10*time.Minute {
log.Warn("delta position is quite stale",
zap.Int64("collectionID", segment.GetCollectionID()),
zap.Int64("segmentID", segment.GetID()),
zap.String("channel", segment.InsertChannel),
zap.Uint64("posTs", checkpoint.GetTimestamp()),
zap.Uint64("posTs", channelCheckpoint.GetTimestamp()),
zap.Time("posTime", posTime),
zap.Duration("tsLag", tsLag))
}
@ -87,7 +82,7 @@ func PackSegmentLoadInfo(segment *datapb.SegmentInfo, channelCheckpoint *msgpb.M
InsertChannel: segment.InsertChannel,
IndexInfos: indexes,
StartPosition: segment.GetStartPosition(),
DeltaPosition: checkpoint,
DeltaPosition: channelCheckpoint,
Level: segment.GetLevel(),
StorageVersion: segment.GetStorageVersion(),
}

2
internal/querycoordv2/utils/types_test.go
View File

@ -80,6 +80,6 @@ func Test_packLoadSegmentRequest(t *testing.T) {
req := PackSegmentLoadInfo(segmentInfo, channel.GetSeekPosition(), nil)
assert.NotNil(t, req.GetDeltaPosition())
assert.Equal(t, mockPChannel, req.GetDeltaPosition().ChannelName)
assert.Equal(t, segmentInfo.GetDmlPosition().GetTimestamp(), req.GetDeltaPosition().GetTimestamp())
assert.Equal(t, channel.SeekPosition.Timestamp, req.GetDeltaPosition().GetTimestamp())
})
}