test/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2024-12-02 11:59:00 +08:00
Code Issues Actions 12 Packages Projects Releases Wiki Activity

Use tt msg stream for consume delete msg (#21478) (#21500)

Browse Source 
Signed-off-by: cai.zhang <cai.zhang@zilliz.com>
This commit is contained in:
cai.zhang 2023-01-05 11:03:35 +08:00 committed by GitHub
parent 58b79eb74c
commit 9dd2e1dcd8
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 19 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
internal/querynode/mock_test.go
View File

@ -1903,7 +1903,7 @@ func (mm *mockMsgStreamFactory) NewMsgStream(ctx context.Context) (msgstream.Msg
}
func (mm *mockMsgStreamFactory) NewTtMsgStream(ctx context.Context) (msgstream.MsgStream, error) {
return nil, nil
return mm.mockMqStream, nil
}
func (mm *mockMsgStreamFactory) NewQueryMsgStream(ctx context.Context) (msgstream.MsgStream, error) {

31
internal/querynode/segment_loader.go
View File

@ -710,7 +710,7 @@ func (loader *segmentLoader) loadDeltaLogs(ctx context.Context, segment *Segment
func (loader *segmentLoader) FromDmlCPLoadDelete(ctx context.Context, collectionID int64, position *internalpb.MsgPosition,
segmentIDs []int64) error {
startTs := time.Now()
stream, err := loader.factory.NewMsgStream(ctx)
stream, err := loader.factory.NewTtMsgStream(ctx)
if err != nil {
return err
}
@ -764,6 +764,7 @@ func (loader *segmentLoader) FromDmlCPLoadDelete(ctx context.Context, collection
zap.String("seekPos", position.String()),
zap.Any("lastMsg", lastMsgID), // use any in case of nil
)
hasMore := true
for hasMore {
select {
@ -803,21 +804,19 @@ func (loader *segmentLoader) FromDmlCPLoadDelete(ctx context.Context, collection
return err
}
}
ret, err := lastMsgID.LessOrEqualThan(tsMsg.Position().MsgID)
if err != nil {
log.Warn("check whether current MsgID less than last MsgID failed",
zap.Int64("collectionID", collectionID),
zap.String("channel", pChannelName),
zap.Error(err),
)
return err
}
if ret {
hasMore = false
break
}
}
ret, err := lastMsgID.LessOrEqualThan(msgPack.EndPositions[0].MsgID)
if err != nil {
log.Warn("check whether current MsgID less than last MsgID failed",
zap.Int64("collectionID", collectionID),
zap.String("channel", pChannelName),
zap.Error(err),
)
return err
}
if ret {
hasMore = false
break
}
}
}

4
internal/querynode/segment_loader_test.go
View File

@ -849,7 +849,9 @@ func testConsumingDeltaMsg(ctx context.Context, t *testing.T, position *msgstrea
msgChan <- nil
deleteMsg1 := genDeleteMsg(defaultCollectionID+1, schemapb.DataType_Int64, defaultDelLength)
deleteMsg2 := genDeleteMsg(defaultCollectionID, schemapb.DataType_Int64, defaultDelLength)
msgChan <- &msgstream.MsgPack{Msgs: []msgstream.TsMsg{deleteMsg1, deleteMsg2}}
msgChan <- &msgstream.MsgPack{Msgs: []msgstream.TsMsg{deleteMsg1, deleteMsg2},
StartPositions: []*internalpb.MsgPosition{genMsgStreamBaseMsg().MsgPosition},
EndPositions: []*internalpb.MsgPosition{genMsgStreamBaseMsg().MsgPosition}}
}
if closedStream {
close(msgChan)