fix store deltaRowcount

Signed-off-by: yangxuan <[email protected]>
milvus-io · Nov 1, 2024 · b88de14 · b88de14
1 parent ad04c2f
commit b88de14
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 5 deletions.
diff --git a/internal/datacoord/meta.go b/internal/datacoord/meta.go
@@ -853,6 +853,10 @@ func AddBinlogsOperator(segmentID int64, binlogs, statslogs, deltalogs []*datapb
 		segment.Binlogs = mergeFieldBinlogs(segment.GetBinlogs(), binlogs)
 		segment.Statslogs = mergeFieldBinlogs(segment.GetStatslogs(), statslogs)
 		segment.Deltalogs = mergeFieldBinlogs(segment.GetDeltalogs(), deltalogs)
+		if len(deltalogs) > 0 {
+			segment.deltaRowcount.Store(-1)
+		}
+
 		modPack.increments[segmentID] = metastore.BinlogsIncrement{
 			Segment: segment.SegmentInfo,
 		}

diff --git a/internal/datacoord/meta_test.go b/internal/datacoord/meta_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/samber/lo"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
 	"github.com/stretchr/testify/suite"
 	"google.golang.org/protobuf/proto"
 
@@ -727,13 +728,15 @@ func TestUpdateSegmentsInfo(t *testing.T) {
 		meta, err := newMemoryMeta()
 		assert.NoError(t, err)
 
-		segment1 := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
+		segment1 := NewSegmentInfo(&datapb.SegmentInfo{
 			ID: 1, State: commonpb.SegmentState_Growing,
 			Binlogs:   []*datapb.FieldBinlog{getFieldBinlogIDs(1, 2)},
 			Statslogs: []*datapb.FieldBinlog{getFieldBinlogIDs(1, 2)},
-		}}
+		})
 		err = meta.AddSegment(context.TODO(), segment1)
 		assert.NoError(t, err)
+		require.EqualValues(t, -1, segment1.deltaRowcount.Load())
+		assert.EqualValues(t, 0, segment1.getDeltaCount())
 
 		err = meta.UpdateSegmentsInfo(
 			UpdateStatusOperator(1, commonpb.SegmentState_Flushing),
@@ -748,6 +751,9 @@ func TestUpdateSegmentsInfo(t *testing.T) {
 		assert.NoError(t, err)
 
 		updated := meta.GetHealthySegment(1)
+		assert.EqualValues(t, -1, updated.deltaRowcount.Load())
+		assert.EqualValues(t, 1, updated.getDeltaCount())
+
 		expected := &SegmentInfo{SegmentInfo: &datapb.SegmentInfo{
 			ID: 1, State: commonpb.SegmentState_Flushing, NumOfRows: 10,
 			StartPosition: &msgpb.MsgPosition{MsgID: []byte{1, 2, 3}},

diff --git a/internal/datacoord/segment_info.go b/internal/datacoord/segment_info.go
@@ -508,10 +508,9 @@ func (s *SegmentInfo) getSegmentSize() int64 {
 	return s.size.Load()
 }
 
-// L1 segment deltaCount changes in any state
-// L0 segment deltaCount won't change
+// Any edits on deltalogs of flushed segments will reset deltaRowcount to -1
 func (s *SegmentInfo) getDeltaCount() int64 {
-	if s.deltaRowcount.Load() < 0 || s.GetLevel() != datapb.SegmentLevel_L0 {
+	if s.deltaRowcount.Load() < 0 || s.GetState() != commonpb.SegmentState_Flushed {
 		var rc int64
 		for _, deltaLogs := range s.GetDeltalogs() {
 			for _, l := range deltaLogs.GetBinlogs() {