admin/milvus
1
0
Fork 0
mirror of https://gitee.com/milvus-io/milvus.git synced 2025-12-08 10:08:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: compacted segment still buffers delta data (#28816)

Browse Source 
Related to #28628
Compacted segment syncing counter is not set correctly in sync task and
the bf write buffer shall not use compacted segment as candidate when
buffering delta data

---------

Signed-off-by: Congqi Xia <congqi.xia@zilliz.com>
This commit is contained in:
congqixia 2023-11-30 10:20:28 +08:00 committed by GitHub
parent f9bb8e9648
commit 2cd8daaf0b
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
3 changed files with 10 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

5
internal/datanode/syncmgr/task.go
View File

@ -87,8 +87,9 @@ func (t *SyncTask) Run() error {
t.segment, has = t.metacache.GetSegmentByID(t.segmentID) t.segment, has = t.metacache.GetSegmentByID(t.segmentID)
if !has { if !has {
log.Warn("failed to sync data, segment not found in metacache") log.Warn("failed to sync data, segment not found in metacache")
err := merr.WrapErrSegmentNotFound(t.segmentID)
t.handleError(err) t.handleError(err)
return merr.WrapErrSegmentNotFound(t.segmentID) return err
} }
if t.segment.CompactTo() == metacache.NullSegment { if t.segment.CompactTo() == metacache.NullSegment {
@ -141,7 +142,7 @@ func (t *SyncTask) Run() error {
actions = append(actions, metacache.UpdateState(commonpb.SegmentState_Flushed)) actions = append(actions, metacache.UpdateState(commonpb.SegmentState_Flushed))
} }
t.metacache.UpdateSegments(metacache.MergeSegmentAction(actions...), metacache.WithSegmentIDs(t.segmentID)) t.metacache.UpdateSegments(metacache.MergeSegmentAction(actions...), metacache.WithSegmentIDs(t.segment.SegmentID()))
log.Info("task done") log.Info("task done")
return nil return nil

3
internal/datanode/writebuffer/bf_write_buffer.go
View File

@ -53,6 +53,9 @@ func (wb *bfWriteBuffer) BufferData(insertMsgs []*msgstream.InsertMsg, deleteMsg
segments := wb.metaCache.GetSegmentsBy(metacache.WithPartitionID(delMsg.PartitionID), segments := wb.metaCache.GetSegmentsBy(metacache.WithPartitionID(delMsg.PartitionID),
metacache.WithSegmentState(commonpb.SegmentState_Growing, commonpb.SegmentState_Flushing, commonpb.SegmentState_Flushed)) metacache.WithSegmentState(commonpb.SegmentState_Growing, commonpb.SegmentState_Flushing, commonpb.SegmentState_Flushed))
for _, segment := range segments { for _, segment := range segments {
if segment.CompactTo() != 0 {
continue
}
var deletePks []storage.PrimaryKey var deletePks []storage.PrimaryKey
var deleteTss []typeutil.Timestamp var deleteTss []typeutil.Timestamp
for idx, pk := range pks { for idx, pk := range pks {

5
internal/datanode/writebuffer/bf_write_buffer_test.go
View File

@ -256,7 +256,10 @@ func (s *BFWriteBufferSuite) TestAutoSyncWithStorageV2() {
s.NoError(err) s.NoError(err)
seg := metacache.NewSegmentInfo(&datapb.SegmentInfo{ID: 1000}, metacache.NewBloomFilterSet()) seg := metacache.NewSegmentInfo(&datapb.SegmentInfo{ID: 1000}, metacache.NewBloomFilterSet())
s.metacache.EXPECT().GetSegmentsBy(mock.Anything, mock.Anything).Return([]*metacache.SegmentInfo{seg}) segCompacted := metacache.NewSegmentInfo(&datapb.SegmentInfo{ID: 1000}, metacache.NewBloomFilterSet())
metacache.CompactTo(2001)(segCompacted)
s.metacache.EXPECT().GetSegmentsBy(mock.Anything, mock.Anything).Return([]*metacache.SegmentInfo{seg, segCompacted})
s.metacache.EXPECT().GetSegmentByID(int64(1000)).Return(nil, false) s.metacache.EXPECT().GetSegmentByID(int64(1000)).Return(nil, false)
s.metacache.EXPECT().GetSegmentByID(int64(1002)).Return(seg, true) s.metacache.EXPECT().GetSegmentByID(int64(1002)).Return(seg, true)
s.metacache.EXPECT().GetSegmentIDsBy(mock.Anything).Return([]int64{1002}) s.metacache.EXPECT().GetSegmentIDsBy(mock.Anything).Return([]int64{1002})