fix: Fix the compatibility bug between stats task and segment (#36359)

issue: #33744 Signed-off-by: Cai Zhang <cai.zhang@zilliz.com>
milvus-io · Sep 20, 2024 · 4b077e1 · 4b077e1
1 parent d2c774f
commit 4b077e1
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 1 deletion.
diff --git a/internal/metastore/kv/datacoord/kv_catalog.go b/internal/metastore/kv/datacoord/kv_catalog.go
@@ -115,7 +115,7 @@ func (kc *Catalog) listSegments() ([]*datapb.SegmentInfo, error) {
 
 		// due to StatsTaskPrefix has the same prefix with SegmentPrefix, so skip it.
 		// when the WalkWithPrefix is refactored, this patch can be removed.
-		if strings.Contains(string(value), StatsTaskPrefix) {
+		if strings.Contains(string(key), StatsTaskPrefix) {
 			return nil
 		}
 

diff --git a/internal/metastore/kv/datacoord/kv_catalog_test.go b/internal/metastore/kv/datacoord/kv_catalog_test.go
@@ -22,6 +22,7 @@ import (
 	"log"
 	"math/rand"
 	"os"
+	"path"
 	"strings"
 	"testing"
 	"time"
@@ -64,6 +65,7 @@ var (
 	k6 = buildFieldBinlogPath(collectionID, partitionID, segmentID2, fieldID)
 	k7 = buildFieldDeltalogPath(collectionID, partitionID, segmentID2, fieldID)
 	k8 = buildFieldStatslogPath(collectionID, partitionID, segmentID2, fieldID)
+	k9 = buildStatsTaskKey(10000)
 
 	keys = map[string]struct{}{
 		k1: {},
@@ -74,6 +76,7 @@ var (
 		k6: {},
 		k7: {},
 		k8: {},
+		k9: {},
 	}
 
 	invalidSegment = &datapb.SegmentInfo{
@@ -202,6 +205,22 @@ func Test_ListSegments(t *testing.T) {
 		verifySegments(t, logID, ret)
 	})
 
+	t.Run("test compatibility with stats task", func(t *testing.T) {
+		metakv := mocks.NewMetaKv(t)
+		metakv.EXPECT().WalkWithPrefix(mock.Anything, mock.Anything, mock.Anything).RunAndReturn(func(s string, i int, f func([]byte, []byte) error) error {
+			if strings.HasPrefix(k9, path.Join(s)) {
+				return f([]byte(k9), nil)
+			}
+			return nil
+		})
+
+		catalog := NewCatalog(metakv, rootPath, "")
+		ret, err := catalog.ListSegments(context.TODO())
+		assert.NotNil(t, ret)
+		assert.NoError(t, err)
+		assert.Zero(t, len(ret))
+	})
+
 	t.Run("list successfully", func(t *testing.T) {
 		var savedKvs map[string]string