enhance: speed up MixCoord recovery by parallelizing startup phases (#47784)

## Summary

Optimize standalone MixCoord recovery to reduce startup time by
parallelizing independent operations:

- **Parallelize DataCoord & QueryCoord startup**: both only depend on
RootCoord being ready, run them concurrently via errgroup
- **Parallelize DataCoord sub-meta loading**: load all 6 sub-metas
(indexMeta, analyzeMeta, partitionStatsMeta, compactionTaskMeta,
statsTaskMeta, snapshotMeta) concurrently alongside reloadFromKV
- **Parallelize indexMeta reload**: run ListIndexes and
ListSegmentIndexes concurrently since they update independent data
structures (m.indexes vs m.segmentIndexes/m.segmentBuildInfo)
- **Batch delete TargetManager targets**: replace per-collection
RemoveCollectionTarget calls with single RemoveWithPrefix
- **Batch etcd loading for ListCollections**: replace sequential per-key
reads with single LoadWithPrefix in RootCoord's initMetaTable

issue: #47783

## Test Plan

- [x] Unit tests pass for all changed packages (index_meta, meta,
kv_catalog, target_manager)
- [x] Race detector clean (`go test -race`)
- [x] Coverage: core functions (reloadFromKV, newIndexMeta,
RemoveCollectionTargets) at 100%; batch load functions at 81-93%

---------

Signed-off-by: sijie-ni-0214 <sijie.ni@zilliz.com>

internal/coordinator/mix_coord.go

@@ -13,6 +13,7 @@ import (
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/grpc"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
@@ -183,25 +184,34 @@ func (s *mixCoordImpl) initInternal() error {
 		return err
 	}
 
-	s.datacoordServer.SetFileResourceObserver(s.fileResourceObserver)
-	if err := s.datacoordServer.Init(); err != nil {
-		log.Error("dataCoord init failed", zap.Error(err))
-		return err
-	}
-
-	if err := s.datacoordServer.Start(); err != nil {
-		log.Error("dataCoord start failed", zap.Error(err))
-		return err
-	}
-
-	s.queryCoordServer.SetFileResourceObserver(s.fileResourceObserver)
-	if err := s.queryCoordServer.Init(); err != nil {
-		log.Error("queryCoord init failed", zap.Error(err))
-		return err
-	}
-
-	if err := s.queryCoordServer.Start(); err != nil {
-		log.Error("queryCoord start failed", zap.Error(err))
+	// DataCoord and QueryCoord are independent of each other;
+	// both only depend on RootCoord being ready. Initialize and start them in parallel.
+	g, _ := errgroup.WithContext(s.ctx)
+	g.Go(func() error {
+		s.datacoordServer.SetFileResourceObserver(s.fileResourceObserver)
+		if err := s.datacoordServer.Init(); err != nil {
+			log.Error("dataCoord init failed", zap.Error(err))
+			return err
+		}
+		if err := s.datacoordServer.Start(); err != nil {
+			log.Error("dataCoord start failed", zap.Error(err))
+			return err
+		}
+		return nil
+	})
+	g.Go(func() error {
+		s.queryCoordServer.SetFileResourceObserver(s.fileResourceObserver)
+		if err := s.queryCoordServer.Init(); err != nil {
+			log.Error("queryCoord init failed", zap.Error(err))
+			return err
+		}
+		if err := s.queryCoordServer.Start(); err != nil {
+			log.Error("queryCoord start failed", zap.Error(err))
+			return err
+		}
+		return nil
+	})
+	if err := g.Wait(); err != nil {
 		return err
 	}
 

internal/datacoord/compaction_policy_clustering_test.go

@@ -59,12 +59,12 @@ func (s *ClusteringCompactionPolicySuite) SetupTest() {
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().SaveCompactionTask(mock.Anything, mock.Anything).Return(nil).Maybe()
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil).Maybe()
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil).Maybe()
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	s.catalog = catalog
 
 	compactionTaskMeta, _ := newCompactionTaskMeta(context.TODO(), s.catalog)
 	partitionStatsMeta, _ := newPartitionStatsMeta(context.TODO(), s.catalog)
-	indexMeta, _ := newIndexMeta(context.TODO(), s.catalog)
+	indexMeta, _ := newIndexMeta(context.TODO(), s.catalog, nil)
 
 	meta := &meta{
 		segments:           NewSegmentsInfo(),

internal/datacoord/copy_segment_checker_test.go

@@ -61,7 +61,7 @@ func (s *CopySegmentCheckerSuite) SetupTest() {
 	s.catalog.EXPECT().ListCopySegmentTasks(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	s.catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	s.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	s.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)

internal/datacoord/copy_segment_inspector_test.go

@@ -57,7 +57,7 @@ func (s *CopySegmentInspectorSuite) SetupTest() {
 	s.catalog.EXPECT().ListCopySegmentTasks(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	s.catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	s.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	s.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)

internal/datacoord/copy_segment_meta_test.go

@@ -55,7 +55,7 @@ func (s *CopySegmentMetaSuite) SetupTest() {
 	s.catalog.EXPECT().ListCopySegmentTasks(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	s.catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	s.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	s.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
@@ -88,7 +88,7 @@ func (s *CopySegmentMetaSuite) TestNewCopySegmentMeta_Success() {
 	catalog.EXPECT().ListCopySegmentTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
@@ -149,7 +149,7 @@ func (s *CopySegmentMetaSuite) TestNewCopySegmentMeta_RestoreJobs() {
 	catalog.EXPECT().ListCopySegmentTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
@@ -199,7 +199,7 @@ func (s *CopySegmentMetaSuite) TestNewCopySegmentMeta_RestoreTasks() {
 	catalog.EXPECT().ListCopySegmentTasks(mock.Anything).Return(restoredTasks, nil)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
@@ -1004,7 +1004,7 @@ func (s *CopySegmentMetaSuite) TestNewCopySegmentMeta_CrashRecoveryRefFiltering(
 	catalog.EXPECT().ListCopySegmentTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)

internal/datacoord/garbage_collector_test.go

@@ -1920,7 +1920,7 @@ func TestGarbageCollector_recycleDroppedSegments_SnapshotReference(t *testing.T)
 	}).Build()
 	defer mock5.UnPatch()
 
-	mock6 := mockey.Mock((*datacoord.Catalog).ListSegmentIndexes).To(func(c *datacoord.Catalog, ctx context.Context) ([]*model.SegmentIndex, error) {
+	mock6 := mockey.Mock((*datacoord.Catalog).ListSegmentIndexes).To(func(c *datacoord.Catalog, ctx context.Context, collectionID int64) ([]*model.SegmentIndex, error) {
 		return []*model.SegmentIndex{}, nil
 	}).Build()
 	defer mock6.UnPatch()
@@ -2027,7 +2027,7 @@ func TestGarbageCollector_recycleUnusedSegIndexes_SnapshotReference(t *testing.T
 	}).Build()
 	defer mock1.UnPatch()
 
-	mock2 := mockey.Mock((*datacoord.Catalog).ListSegmentIndexes).To(func(c *datacoord.Catalog, ctx context.Context) ([]*model.SegmentIndex, error) {
+	mock2 := mockey.Mock((*datacoord.Catalog).ListSegmentIndexes).To(func(c *datacoord.Catalog, ctx context.Context, collectionID int64) ([]*model.SegmentIndex, error) {
 		return []*model.SegmentIndex{segIdx1, segIdx2}, nil
 	}).Build()
 	defer mock2.UnPatch()

internal/datacoord/import_checker_test.go

@@ -59,7 +59,7 @@ func (s *ImportCheckerSuite) SetupTest() {
 	catalog.EXPECT().ListImportTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
@@ -578,7 +578,7 @@ func TestImportCheckerCompaction(t *testing.T) {
 	catalog.EXPECT().ListImportTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)

internal/datacoord/import_inspector_test.go

@@ -58,7 +58,7 @@ func (s *ImportInspectorSuite) SetupTest() {
 	s.catalog.EXPECT().ListImportTasks(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	s.catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	s.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	s.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	s.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)

internal/datacoord/import_util_test.go

@@ -127,7 +127,7 @@ func TestImportUtil_NewImportTasks(t *testing.T) {
 	catalog := mocks.NewDataCoordCatalog(t)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().AddSegment(mock.Anything, mock.Anything).Return(nil)
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
@@ -203,7 +203,7 @@ func TestImportUtil_NewImportTasksWithDataTt(t *testing.T) {
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().AddSegment(mock.Anything, mock.Anything).Return(nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
@@ -265,7 +265,7 @@ func TestImportUtil_AssembleRequest(t *testing.T) {
 	catalog := mocks.NewDataCoordCatalog(t)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().AddSegment(mock.Anything, mock.Anything).Return(nil)
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
@@ -343,7 +343,7 @@ func TestImportUtil_AssembleRequestWithDataTt(t *testing.T) {
 	catalog := mocks.NewDataCoordCatalog(t)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().AddSegment(mock.Anything, mock.Anything).Return(nil)
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
@@ -429,7 +429,7 @@ func TestImportUtil_CheckDiskQuota(t *testing.T) {
 	catalog.EXPECT().SaveImportJob(mock.Anything, mock.Anything).Return(nil)
 	catalog.EXPECT().SavePreImportTask(mock.Anything, mock.Anything).Return(nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().AddSegment(mock.Anything, mock.Anything).Return(nil)
 	catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
@@ -616,7 +616,7 @@ func TestImportUtil_GetImportProgress(t *testing.T) {
 	catalog.EXPECT().ListImportTasks(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
 	catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, nil)
-	catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, nil)
+	catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
 	catalog.EXPECT().SaveImportJob(mock.Anything, mock.Anything).Return(nil)
 	catalog.EXPECT().SavePreImportTask(mock.Anything, mock.Anything).Return(nil)
 	catalog.EXPECT().SaveImportTask(mock.Anything, mock.Anything).Return(nil)

internal/datacoord/index_meta.go

@@ -29,6 +29,7 @@ import (
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/samber/lo"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
@@ -43,6 +44,7 @@ import (
 	"github.com/milvus-io/milvus/pkg/v2/metrics"
 	"github.com/milvus-io/milvus/pkg/v2/proto/indexpb"
 	"github.com/milvus-io/milvus/pkg/v2/proto/workerpb"
+	"github.com/milvus-io/milvus/pkg/v2/util/conc"
 	"github.com/milvus-io/milvus/pkg/v2/util/funcutil"
 	"github.com/milvus-io/milvus/pkg/v2/util/indexparams"
 	"github.com/milvus-io/milvus/pkg/v2/util/lock"
@@ -97,12 +99,14 @@ type segmentBuildInfo struct {
 	taskStats *expirable.LRU[UniqueID, *metricsinfo.IndexTaskStats]
 }
 
+const taskStatsLRUCapacity = 1024
+
 func newSegmentIndexBuildInfo() *segmentBuildInfo {
 	return &segmentBuildInfo{
 		// build ID -> segment index
 		buildID2SegmentIndex: typeutil.NewConcurrentMap[UniqueID, *model.SegmentIndex](),
 		// build ID -> task stats
-		taskStats: expirable.NewLRU[UniqueID, *metricsinfo.IndexTaskStats](1024, nil, time.Minute*30),
+		taskStats: expirable.NewLRU[UniqueID, *metricsinfo.IndexTaskStats](taskStatsLRUCapacity, nil, time.Minute*30),
 	}
 }
 
@@ -111,6 +115,18 @@ func (m *segmentBuildInfo) Add(segIdx *model.SegmentIndex) {
 	m.taskStats.Add(segIdx.BuildID, newIndexTaskStats(segIdx))
 }
 
+// AddForRecovery inserts segment index during recovery. It skips the LRU write
+// only when the LRU is already full and the index task is finished, since
+// finished tasks are unlikely to be queried and the LRU write is expensive
+// during bulk recovery.
+func (m *segmentBuildInfo) AddForRecovery(segIdx *model.SegmentIndex) {
+	m.buildID2SegmentIndex.Insert(segIdx.BuildID, segIdx)
+	if m.taskStats.Len() >= taskStatsLRUCapacity && segIdx.IndexState == commonpb.IndexState_Finished {
+		return
+	}
+	m.taskStats.Add(segIdx.BuildID, newIndexTaskStats(segIdx))
+}
+
 func (m *segmentBuildInfo) Get(key UniqueID) (*model.SegmentIndex, bool) {
 	value, exists := m.buildID2SegmentIndex.Get(key)
 	return value, exists
@@ -129,7 +145,7 @@ func (m *segmentBuildInfo) GetTaskStats() []*metricsinfo.IndexTaskStats {
 }
 
 // NewMeta creates meta from provided `kv.TxnKV`
-func newIndexMeta(ctx context.Context, catalog metastore.DataCoordCatalog) (*indexMeta, error) {
+func newIndexMeta(ctx context.Context, catalog metastore.DataCoordCatalog, collectionIDs []int64) (*indexMeta, error) {
 	mt := &indexMeta{
 		ctx:              ctx,
 		catalog:          catalog,
@@ -138,7 +154,7 @@ func newIndexMeta(ctx context.Context, catalog metastore.DataCoordCatalog) (*ind
 		segmentBuildInfo: newSegmentIndexBuildInfo(),
 		segmentIndexes:   typeutil.NewConcurrentMap[UniqueID, *typeutil.ConcurrentMap[UniqueID, *model.SegmentIndex]](),
 	}
-	err := mt.reloadFromKV()
+	err := mt.reloadFromKV(collectionIDs)
 	if err != nil {
 		return nil, err
 	}
@@ -146,30 +162,77 @@ func newIndexMeta(ctx context.Context, catalog metastore.DataCoordCatalog) (*ind
 }
 
 // reloadFromKV loads meta from KV storage
-func (m *indexMeta) reloadFromKV() error {
+func (m *indexMeta) reloadFromKV(collectionIDs []int64) error {
 	record := timerecord.NewTimeRecorder("indexMeta-reloadFromKV")
-	// load field indexes
-	fieldIndexes, err := m.catalog.ListIndexes(m.ctx)
-	if err != nil {
-		log.Error("indexMeta reloadFromKV load field indexes fail", zap.Error(err))
-		return err
-	}
-	for _, fieldIndex := range fieldIndexes {
-		m.updateCollectionIndex(fieldIndex)
-	}
-	segmentIndexes, err := m.catalog.ListSegmentIndexes(m.ctx)
-	if err != nil {
-		log.Error("indexMeta reloadFromKV load segment indexes fail", zap.Error(err))
-		return err
-	}
-	for _, segIdx := range segmentIndexes {
-		if segIdx.IndexMemSize == 0 {
-			segIdx.IndexMemSize = segIdx.IndexSerializedSize * paramtable.Get().DataCoordCfg.IndexMemSizeEstimateMultiplier.GetAsUint64()
+
+	// Parallel load and process: ListIndexes and ListSegmentIndexes have no dependency,
+	// and they update completely separate data structures so memory updates can also run in parallel.
+	g, _ := errgroup.WithContext(m.ctx)
+	g.Go(func() error {
+		fieldIndexes, err := m.catalog.ListIndexes(m.ctx)
+		if err != nil {
+			log.Error("indexMeta reloadFromKV load field indexes fail", zap.Error(err))
+			return err
 		}
-		m.updateSegmentIndex(segIdx)
-		metrics.FlushedSegmentFileNum.WithLabelValues(metrics.IndexFileLabel).Observe(float64(len(segIdx.IndexFileKeys)))
-		metrics.DataCoordStoredIndexFilesSize.WithLabelValues("", "",
-			fmt.Sprintf("%d", segIdx.CollectionID)).Add(float64(segIdx.IndexSerializedSize))
+		for _, fieldIndex := range fieldIndexes {
+			m.updateCollectionIndex(fieldIndex)
+		}
+		return nil
+	})
+	g.Go(func() error {
+		pool := conc.NewPool[any](paramtable.Get().MetaStoreCfg.ReadConcurrency.GetAsInt())
+		defer pool.Release()
+		futures := make([]*conc.Future[any], 0, len(collectionIDs))
+		collectionSegIdxes := make([][]*model.SegmentIndex, len(collectionIDs))
+		for i, collID := range collectionIDs {
+			i, collID := i, collID
+			futures = append(futures, pool.Submit(func() (any, error) {
+				segIdxes, err := m.catalog.ListSegmentIndexes(m.ctx, collID)
+				if err != nil {
+					return nil, err
+				}
+				collectionSegIdxes[i] = segIdxes
+				return nil, nil
+			}))
+		}
+		if err := conc.AwaitAll(futures...); err != nil {
+			return err
+		}
+		for _, segIdxes := range collectionSegIdxes {
+			for _, segIdx := range segIdxes {
+				if segIdx.IndexMemSize == 0 {
+					segIdx.IndexMemSize = segIdx.IndexSerializedSize * paramtable.Get().DataCoordCfg.IndexMemSizeEstimateMultiplier.GetAsUint64()
+				}
+				indexes, ok := m.segmentIndexes.Get(segIdx.SegmentID)
+				if ok {
+					indexes.Insert(segIdx.IndexID, segIdx)
+				} else {
+					indexes = typeutil.NewConcurrentMap[UniqueID, *model.SegmentIndex]()
+					indexes.Insert(segIdx.IndexID, segIdx)
+					m.segmentIndexes.Insert(segIdx.SegmentID, indexes)
+				}
+				m.segmentBuildInfo.AddForRecovery(segIdx)
+			}
+		}
+
+		// Update Prometheus metrics asynchronously to avoid blocking recovery.
+		go func() {
+			storedSizeByCollection := make(map[int64]float64)
+			for _, segIdxes := range collectionSegIdxes {
+				for _, segIdx := range segIdxes {
+					metrics.FlushedSegmentFileNum.WithLabelValues(metrics.IndexFileLabel).Observe(float64(len(segIdx.IndexFileKeys)))
+					storedSizeByCollection[segIdx.CollectionID] += float64(segIdx.IndexSerializedSize)
+				}
+			}
+			for collID, size := range storedSizeByCollection {
+				metrics.DataCoordStoredIndexFilesSize.WithLabelValues("", "",
+					fmt.Sprintf("%d", collID)).Add(size)
+			}
+		}()
+		return nil
+	})
+	if err := g.Wait(); err != nil {
+		return err
 	}
 	log.Info("indexMeta reloadFromKV done", zap.Duration("duration", record.ElapseSpan()))
 	return nil
@@ -186,7 +249,6 @@ func (m *indexMeta) updateSegmentIndex(segIdx *model.SegmentIndex) {
 	indexes, ok := m.segmentIndexes.Get(segIdx.SegmentID)
 	if ok {
 		indexes.Insert(segIdx.IndexID, segIdx)
-		m.segmentIndexes.Insert(segIdx.SegmentID, indexes)
 	} else {
 		indexes := typeutil.NewConcurrentMap[UniqueID, *model.SegmentIndex]()
 		indexes.Insert(segIdx.IndexID, segIdx)

internal/datacoord/index_meta_test.go

@@ -45,16 +45,17 @@ func TestReloadFromKV(t *testing.T) {
 	t.Run("ListIndexes_fail", func(t *testing.T) {
 		catalog := catalogmocks.NewDataCoordCatalog(t)
 		catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, errors.New("mock"))
-		_, err := newIndexMeta(context.TODO(), catalog)
+		catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, nil).Maybe()
+		_, err := newIndexMeta(context.TODO(), catalog, []int64{0})
 		assert.Error(t, err)
 	})
 
 	t.Run("ListSegmentIndexes_fails", func(t *testing.T) {
 		catalog := catalogmocks.NewDataCoordCatalog(t)
 		catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
-		catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return(nil, errors.New("mock"))
+		catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return(nil, errors.New("mock"))
 
-		_, err := newIndexMeta(context.TODO(), catalog)
+		_, err := newIndexMeta(context.TODO(), catalog, []int64{0})
 		assert.Error(t, err)
 	})
 
@@ -69,14 +70,14 @@ func TestReloadFromKV(t *testing.T) {
 			},
 		}, nil)
 
-		catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return([]*model.SegmentIndex{
+		catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{
 			{
 				SegmentID: 1,
 				IndexID:   1,
 			},
 		}, nil)
 
-		meta, err := newIndexMeta(context.TODO(), catalog)
+		meta, err := newIndexMeta(context.TODO(), catalog, []int64{0})
 		assert.NoError(t, err)
 		assert.NotNil(t, meta)
 	})
@@ -1609,6 +1610,54 @@ func TestBuildIndexTaskStatsJSON(t *testing.T) {
 	assert.Equal(t, 1, len(im.segmentBuildInfo.List()))
 }
 
+func TestSegmentBuildInfo_AddForRecovery(t *testing.T) {
+	t.Run("lru not full, all states inserted", func(t *testing.T) {
+		info := newSegmentIndexBuildInfo()
+		finished := &model.SegmentIndex{BuildID: 1, IndexState: commonpb.IndexState_Finished}
+		inProgress := &model.SegmentIndex{BuildID: 2, IndexState: commonpb.IndexState_InProgress}
+
+		info.AddForRecovery(finished)
+		info.AddForRecovery(inProgress)
+
+		assert.Equal(t, 2, len(info.List()))
+		assert.Equal(t, 2, len(info.GetTaskStats()))
+	})
+
+	t.Run("lru full, finished tasks skipped", func(t *testing.T) {
+		info := newSegmentIndexBuildInfo()
+		// Fill the LRU to capacity with in-progress tasks.
+		for i := int64(0); i < taskStatsLRUCapacity; i++ {
+			info.AddForRecovery(&model.SegmentIndex{BuildID: i, IndexState: commonpb.IndexState_InProgress})
+		}
+		assert.Equal(t, taskStatsLRUCapacity, info.taskStats.Len())
+
+		// A finished task should be skipped when LRU is full.
+		finished := &model.SegmentIndex{BuildID: taskStatsLRUCapacity + 1, IndexState: commonpb.IndexState_Finished}
+		info.AddForRecovery(finished)
+
+		_, ok := info.Get(finished.BuildID)
+		assert.True(t, ok, "buildID2SegmentIndex should still contain the entry")
+		assert.Equal(t, taskStatsLRUCapacity, info.taskStats.Len(), "LRU size should not grow")
+	})
+
+	t.Run("lru full, unfinished tasks still inserted", func(t *testing.T) {
+		info := newSegmentIndexBuildInfo()
+		for i := int64(0); i < taskStatsLRUCapacity; i++ {
+			info.AddForRecovery(&model.SegmentIndex{BuildID: i, IndexState: commonpb.IndexState_InProgress})
+		}
+		assert.Equal(t, taskStatsLRUCapacity, info.taskStats.Len())
+
+		// An unfinished task should still be inserted (evicting the oldest).
+		unissued := &model.SegmentIndex{BuildID: taskStatsLRUCapacity + 1, IndexState: commonpb.IndexState_Unissued}
+		info.AddForRecovery(unissued)
+
+		_, ok := info.Get(unissued.BuildID)
+		assert.True(t, ok)
+		// LRU size stays at capacity because the oldest entry was evicted.
+		assert.Equal(t, taskStatsLRUCapacity, info.taskStats.Len())
+	})
+}
+
 func TestMeta_GetIndexJSON(t *testing.T) {
 	m := &indexMeta{
 		indexes: map[UniqueID]map[UniqueID]*model.Index{

internal/datacoord/meta.go

@@ -30,6 +30,7 @@ import (
 	"github.com/samber/lo"
 	"go.uber.org/zap"
 	"golang.org/x/exp/maps"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/commonpb"
@@ -184,93 +185,131 @@ type dbInfo struct {
 	Properties []*commonpb.KeyValuePair
 }
 
-// NewMeta creates meta from provided `kv.TxnKV`
-func newMeta(ctx context.Context, catalog metastore.DataCoordCatalog, chunkManager storage.ChunkManager, broker broker.Broker) (*meta, error) {
-	im, err := newIndexMeta(ctx, catalog)
-	if err != nil {
-		return nil, err
-	}
-
-	am, err := newAnalyzeMeta(ctx, catalog)
-	if err != nil {
-		return nil, err
-	}
-
-	psm, err := newPartitionStatsMeta(ctx, catalog)
-	if err != nil {
-		return nil, err
-	}
-
-	ctm, err := newCompactionTaskMeta(ctx, catalog)
-	if err != nil {
-		return nil, err
-	}
-
-	stm, err := newStatsTaskMeta(ctx, catalog)
-	if err != nil {
-		return nil, err
-	}
-
-	ecrm, err := newExternalCollectionRefreshMeta(ctx, catalog)
-	if err != nil {
-		return nil, err
-	}
-
-	spm, err := newSnapshotMeta(ctx, catalog, chunkManager)
-	if err != nil {
-		return nil, err
-	}
-
-	mt := &meta{
-		ctx:                           ctx,
-		catalog:                       catalog,
-		collections:                   typeutil.NewConcurrentMap[UniqueID, *collectionInfo](),
-		segments:                      NewSegmentsInfo(),
-		channelCPs:                    newChannelCps(),
-		indexMeta:                     im,
-		analyzeMeta:                   am,
-		chunkManager:                  chunkManager,
-		partitionStatsMeta:            psm,
-		compactionTaskMeta:            ctm,
-		statsTaskMeta:                 stm,
-		externalCollectionRefreshMeta: ecrm,
-		resourceIDMap:                 make(map[int64]*internalpb.FileResourceInfo),
-		resourceVersion:               0,
-		resourceLock:                  lock.RWMutex{},
-		snapshotMeta:                  spm,
-	}
-	err = mt.reloadFromKV(ctx, broker)
-	if err != nil {
-		return nil, err
-	}
-	return mt, nil
-}
-
-// reloadFromKV loads meta from KV storage
-func (m *meta) reloadFromKV(ctx context.Context, broker broker.Broker) error {
-	record := timerecord.NewTimeRecorder("datacoord")
-
+// showCollectionIDs retrieves all collection IDs from RootCoord with retry on ErrServiceUnimplemented.
+func showCollectionIDs(ctx context.Context, broker broker.Broker) ([]int64, error) {
 	var (
 		err  error
 		resp *rootcoordpb.ShowCollectionIDsResponse
 	)
-	// retry on un implemented for compatibility
 	retryErr := retry.Handle(ctx, func() (bool, error) {
-		resp, err = broker.ShowCollectionIDs(m.ctx)
+		resp, err = broker.ShowCollectionIDs(ctx)
 		if errors.Is(err, merr.ErrServiceUnimplemented) {
 			return true, err
 		}
 		return false, err
 	})
 	if retryErr != nil {
-		return retryErr
+		return nil, retryErr
 	}
-	log.Ctx(ctx).Info("datacoord show collections done", zap.Duration("dur", record.RecordSpan()))
 
 	collectionIDs := make([]int64, 0, 4096)
 	for _, collections := range resp.GetDbCollections() {
 		collectionIDs = append(collectionIDs, collections.GetCollectionIDs()...)
 	}
+	return collectionIDs, nil
+}
+
+// NewMeta creates meta from provided `kv.TxnKV`
+func newMeta(ctx context.Context, catalog metastore.DataCoordCatalog, chunkManager storage.ChunkManager, broker broker.Broker) (*meta, error) {
+	// Fetch collection IDs first so both reloadFromKV and indexMeta can use them for per-collection loading.
+	collectionIDs, err := showCollectionIDs(ctx, broker)
+	if err != nil {
+		return nil, err
+	}
+
+	var (
+		im   *indexMeta
+		am   *analyzeMeta
+		psm  *partitionStatsMeta
+		ctm  *compactionTaskMeta
+		stm  *statsTaskMeta
+		ecrm *externalCollectionRefreshMeta
+		spm  *snapshotMeta
+	)
+
+	// Construct meta struct first so reloadFromKV can run in parallel with sub-meta loading.
+	// reloadFromKV uses m.catalog/m.segments/m.channelCPs which are independent of sub-metas.
+	mt := &meta{
+		ctx:             ctx,
+		catalog:         catalog,
+		collections:     typeutil.NewConcurrentMap[UniqueID, *collectionInfo](),
+		segments:        NewSegmentsInfo(),
+		channelCPs:      newChannelCps(),
+		chunkManager:    chunkManager,
+		resourceIDMap:   make(map[int64]*internalpb.FileResourceInfo),
+		resourceVersion: 0,
+		resourceLock:    lock.RWMutex{},
+	}
+
+	g, _ := errgroup.WithContext(ctx)
+
+	g.Go(func() error {
+		var err error
+		im, err = newIndexMeta(ctx, catalog, collectionIDs)
+		return err
+	})
+
+	g.Go(func() error {
+		var err error
+		am, err = newAnalyzeMeta(ctx, catalog)
+		return err
+	})
+
+	g.Go(func() error {
+		var err error
+		psm, err = newPartitionStatsMeta(ctx, catalog)
+		return err
+	})
+
+	g.Go(func() error {
+		var err error
+		ctm, err = newCompactionTaskMeta(ctx, catalog)
+		return err
+	})
+
+	g.Go(func() error {
+		var err error
+		stm, err = newStatsTaskMeta(ctx, catalog)
+		return err
+	})
+
+	g.Go(func() error {
+		var err error
+		ecrm, err = newExternalCollectionRefreshMeta(ctx, catalog)
+		return err
+	})
+
+	g.Go(func() error {
+		var err error
+		spm, err = newSnapshotMeta(ctx, catalog, chunkManager)
+		return err
+	})
+
+	// reloadFromKV (ListSegments, ListChannelCheckpoint) runs in parallel with sub-meta loading.
+	// It only uses mt.catalog/mt.segments/mt.channelCPs, which are independent of sub-metas.
+	g.Go(func() error {
+		return mt.reloadFromKV(ctx, collectionIDs)
+	})
+
+	if err := g.Wait(); err != nil {
+		return nil, err
+	}
+
+	// Assign sub-metas after all goroutines complete
+	mt.indexMeta = im
+	mt.analyzeMeta = am
+	mt.partitionStatsMeta = psm
+	mt.compactionTaskMeta = ctm
+	mt.statsTaskMeta = stm
+	mt.externalCollectionRefreshMeta = ecrm
+	mt.snapshotMeta = spm
+
+	return mt, nil
+}
+
+// reloadFromKV loads meta from KV storage
+func (m *meta) reloadFromKV(ctx context.Context, collectionIDs []int64) error {
+	record := timerecord.NewTimeRecorder("datacoord")
 
 	pool := conc.NewPool[any](paramtable.Get().MetaStoreCfg.ReadConcurrency.GetAsInt())
 	defer pool.Release()
@@ -288,8 +327,7 @@ func (m *meta) reloadFromKV(ctx context.Context, broker broker.Broker) error {
 			return nil, nil
 		}))
 	}
-	err = conc.AwaitAll(futures...)
-	if err != nil {
+	if err := conc.AwaitAll(futures...); err != nil {
 		return err
 	}
 

internal/datacoord/meta_test.go

@@ -93,7 +93,7 @@ func (suite *MetaReloadSuite) TestReloadFromKV() {
 		}, nil)
 		suite.catalog.EXPECT().ListSegments(mock.Anything, mock.Anything).Return(nil, errors.New("mock"))
 		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
-		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return([]*model.SegmentIndex{}, nil)
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
 		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
@@ -113,7 +113,7 @@ func (suite *MetaReloadSuite) TestReloadFromKV() {
 		suite.catalog.EXPECT().ListSegments(mock.Anything, mock.Anything).Return([]*datapb.SegmentInfo{}, nil)
 		suite.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, errors.New("mock"))
 		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
-		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return([]*model.SegmentIndex{}, nil)
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
 		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
@@ -140,7 +140,7 @@ func (suite *MetaReloadSuite) TestReloadFromKV() {
 		}, nil)
 
 		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
-		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return([]*model.SegmentIndex{}, nil)
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
 		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
@@ -170,6 +170,120 @@ func (suite *MetaReloadSuite) TestReloadFromKV() {
 		suite.MetricsEqual(metrics.DataCoordNumSegments.WithLabelValues(metrics.FlushedSegmentLabel, datapb.SegmentLevel_Legacy.String(), "unsorted", "0"), 1)
 	})
 
+	suite.Run("ListIndexes_fail", func() {
+		defer suite.resetMock()
+		brk := broker.NewMockBroker(suite.T())
+		brk.EXPECT().ShowCollectionIDs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return(nil, errors.New("mock"))
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
+		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListStatsTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListSnapshots(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshJobs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshTasks(mock.Anything).Return(nil, nil)
+
+		_, err := newMeta(ctx, suite.catalog, nil, brk)
+		suite.Error(err)
+	})
+
+	suite.Run("ListAnalyzeTasks_fail", func() {
+		defer suite.resetMock()
+		brk := broker.NewMockBroker(suite.T())
+		brk.EXPECT().ShowCollectionIDs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
+		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, errors.New("mock"))
+		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListStatsTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListSnapshots(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshJobs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshTasks(mock.Anything).Return(nil, nil)
+
+		_, err := newMeta(ctx, suite.catalog, nil, brk)
+		suite.Error(err)
+	})
+
+	suite.Run("ListPartitionStatsInfos_fail", func() {
+		defer suite.resetMock()
+		brk := broker.NewMockBroker(suite.T())
+		brk.EXPECT().ShowCollectionIDs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
+		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, errors.New("mock"))
+		suite.catalog.EXPECT().ListStatsTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListSnapshots(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshJobs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshTasks(mock.Anything).Return(nil, nil)
+
+		_, err := newMeta(ctx, suite.catalog, nil, brk)
+		suite.Error(err)
+	})
+
+	suite.Run("ListCompactionTask_fail", func() {
+		defer suite.resetMock()
+		brk := broker.NewMockBroker(suite.T())
+		brk.EXPECT().ShowCollectionIDs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
+		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, errors.New("mock"))
+		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListStatsTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListSnapshots(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshJobs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshTasks(mock.Anything).Return(nil, nil)
+
+		_, err := newMeta(ctx, suite.catalog, nil, brk)
+		suite.Error(err)
+	})
+
+	suite.Run("ListStatsTasks_fail", func() {
+		defer suite.resetMock()
+		brk := broker.NewMockBroker(suite.T())
+		brk.EXPECT().ShowCollectionIDs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
+		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListStatsTasks(mock.Anything).Return(nil, errors.New("mock"))
+		suite.catalog.EXPECT().ListSnapshots(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshJobs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshTasks(mock.Anything).Return(nil, nil)
+
+		_, err := newMeta(ctx, suite.catalog, nil, brk)
+		suite.Error(err)
+	})
+
+	suite.Run("ListSnapshots_fail", func() {
+		defer suite.resetMock()
+		brk := broker.NewMockBroker(suite.T())
+		brk.EXPECT().ShowCollectionIDs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
+		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListStatsTasks(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListSnapshots(mock.Anything).Return(nil, errors.New("mock"))
+		suite.catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshJobs(mock.Anything).Return(nil, nil)
+		suite.catalog.EXPECT().ListExternalCollectionRefreshTasks(mock.Anything).Return(nil, nil)
+
+		_, err := newMeta(ctx, suite.catalog, nil, brk)
+		suite.Error(err)
+	})
+
 	suite.Run("test list segments", func() {
 		defer suite.resetMock()
 		brk := broker.NewMockBroker(suite.T())
@@ -188,7 +302,7 @@ func (suite *MetaReloadSuite) TestReloadFromKV() {
 		}, nil)
 
 		suite.catalog.EXPECT().ListIndexes(mock.Anything).Return([]*model.Index{}, nil)
-		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything).Return([]*model.SegmentIndex{}, nil)
+		suite.catalog.EXPECT().ListSegmentIndexes(mock.Anything, mock.Anything).Return([]*model.SegmentIndex{}, nil).Maybe()
 		suite.catalog.EXPECT().ListAnalyzeTasks(mock.Anything).Return(nil, nil)
 		suite.catalog.EXPECT().ListCompactionTask(mock.Anything).Return(nil, nil)
 		suite.catalog.EXPECT().ListPartitionStatsInfos(mock.Anything).Return(nil, nil)

internal/metastore/catalog.go

@@ -201,7 +201,7 @@ type DataCoordCatalog interface {
 	DropIndex(ctx context.Context, collID, dropIdxID typeutil.UniqueID) error
 
 	CreateSegmentIndex(ctx context.Context, segIdx *model.SegmentIndex) error
-	ListSegmentIndexes(ctx context.Context) ([]*model.SegmentIndex, error)
+	ListSegmentIndexes(ctx context.Context, collectionID int64) ([]*model.SegmentIndex, error)
 	AlterSegmentIndexes(ctx context.Context, newSegIdxes []*model.SegmentIndex) error
 	DropSegmentIndex(ctx context.Context, collID, partID, segID, buildID typeutil.UniqueID) error
 
@@ -285,6 +285,7 @@ type QueryCoordCatalog interface {
 
 	SaveCollectionTargets(ctx context.Context, target ...*querypb.CollectionTarget) error
 	RemoveCollectionTarget(ctx context.Context, collectionID int64) error
+	RemoveCollectionTargets(ctx context.Context) error
 	GetCollectionTargets(ctx context.Context) (map[int64]*querypb.CollectionTarget, error)
 }
 

internal/metastore/kv/datacoord/kv_catalog.go

@@ -636,7 +636,7 @@ func (kc *Catalog) CreateSegmentIndex(ctx context.Context, segIdx *model.Segment
 	return nil
 }
 
-func (kc *Catalog) ListSegmentIndexes(ctx context.Context) ([]*model.SegmentIndex, error) {
+func (kc *Catalog) ListSegmentIndexes(ctx context.Context, collectionID int64) ([]*model.SegmentIndex, error) {
 	segIndexes := make([]*model.SegmentIndex, 0)
 	applyFn := func(key []byte, value []byte) error {
 		segmentIndexInfo := &indexpb.SegmentIndex{}
@@ -650,7 +650,8 @@ func (kc *Catalog) ListSegmentIndexes(ctx context.Context) ([]*model.SegmentInde
 		return nil
 	}
 
-	err := kc.MetaKv.WalkWithPrefix(ctx, util.SegmentIndexPrefix+"/", kc.paginationSize, applyFn)
+	prefix := buildSegmentIndexCollectionPrefix(collectionID)
+	err := kc.MetaKv.WalkWithPrefix(ctx, prefix, kc.paginationSize, applyFn)
 	if err != nil {
 		return nil, err
 	}

internal/metastore/kv/datacoord/kv_catalog_test.go

@@ -1076,7 +1076,7 @@ func TestCatalog_ListSegmentIndexes(t *testing.T) {
 			MetaKv: metakv,
 		}
 
-		segIdxes, err := catalog.ListSegmentIndexes(context.Background())
+		segIdxes, err := catalog.ListSegmentIndexes(context.Background(), 0)
 		assert.NoError(t, err)
 		assert.Equal(t, 1, len(segIdxes))
 	})
@@ -1088,7 +1088,7 @@ func TestCatalog_ListSegmentIndexes(t *testing.T) {
 			MetaKv: metakv,
 		}
 
-		_, err := catalog.ListSegmentIndexes(context.Background())
+		_, err := catalog.ListSegmentIndexes(context.Background(), 0)
 		assert.Error(t, err)
 	})
 
@@ -1101,7 +1101,7 @@ func TestCatalog_ListSegmentIndexes(t *testing.T) {
 			MetaKv: metakv,
 		}
 
-		_, err := catalog.ListSegmentIndexes(context.Background())
+		_, err := catalog.ListSegmentIndexes(context.Background(), 0)
 		assert.Error(t, err)
 	})
 }

internal/metastore/kv/datacoord/util.go

@@ -329,6 +329,10 @@ func BuildSegmentIndexKey(collectionID, partitionID, segmentID, buildID int64) s
 	return fmt.Sprintf("%s/%d/%d/%d/%d", util.SegmentIndexPrefix, collectionID, partitionID, segmentID, buildID)
 }
 
+func buildSegmentIndexCollectionPrefix(collectionID typeutil.UniqueID) string {
+	return fmt.Sprintf("%s/%d/", util.SegmentIndexPrefix, collectionID)
+}
+
 func buildCollectionPrefix(collectionID typeutil.UniqueID) string {
 	return fmt.Sprintf("%s/%d/", SegmentPrefix, collectionID)
 }

internal/metastore/kv/querycoord/kv_catalog.go

@@ -326,6 +326,10 @@ func (s Catalog) RemoveCollectionTarget(ctx context.Context, collectionID int64)
 	return s.cli.Remove(ctx, k)
 }
 
+func (s Catalog) RemoveCollectionTargets(ctx context.Context) error {
+	return s.cli.RemoveWithPrefix(ctx, CollectionTargetPrefix)
+}
+
 func (s Catalog) GetCollectionTargets(ctx context.Context) (map[int64]*querypb.CollectionTarget, error) {
 	ret := make(map[int64]*querypb.CollectionTarget)
 	applyFn := func(key []byte, value []byte) error {

internal/metastore/kv/querycoord/kv_catalog_test.go

@@ -302,6 +302,39 @@ func (suite *CatalogTestSuite) TestCollectionTarget() {
 	suite.Error(err)
 }
 
+func (suite *CatalogTestSuite) TestRemoveCollectionTargets() {
+	ctx := context.Background()
+	// save 5 targets
+	suite.catalog.SaveCollectionTargets(ctx,
+		&querypb.CollectionTarget{CollectionID: 1, Version: 1},
+		&querypb.CollectionTarget{CollectionID: 2, Version: 2},
+		&querypb.CollectionTarget{CollectionID: 3, Version: 3},
+		&querypb.CollectionTarget{CollectionID: 4, Version: 4},
+		&querypb.CollectionTarget{CollectionID: 5, Version: 5},
+	)
+
+	// remove all targets via prefix delete
+	err := suite.catalog.RemoveCollectionTargets(ctx)
+	suite.NoError(err)
+
+	targets, err := suite.catalog.GetCollectionTargets(ctx)
+	suite.NoError(err)
+	suite.Len(targets, 0)
+
+	// remove when no targets exist should be no-op
+	err = suite.catalog.RemoveCollectionTargets(ctx)
+	suite.NoError(err)
+
+	// test error from meta store
+	mockStore := mocks.NewMetaKv(suite.T())
+	mockErr := errors.New("failed to access etcd")
+	mockStore.EXPECT().RemoveWithPrefix(mock.Anything, CollectionTargetPrefix).Return(mockErr)
+
+	suite.catalog.cli = mockStore
+	err = suite.catalog.RemoveCollectionTargets(ctx)
+	suite.ErrorIs(err, mockErr)
+}
+
 func (suite *CatalogTestSuite) TestLoadRelease() {
 	// TODO(sunby): add ut
 }

internal/metastore/kv/rootcoord/kv_catalog.go

@@ -10,6 +10,7 @@ import (
 	"github.com/cockroachdb/errors"
 	"github.com/samber/lo"
 	"go.uber.org/zap"
+	"golang.org/x/sync/errgroup"
 	"google.golang.org/protobuf/proto"
 
 	"github.com/milvus-io/milvus-proto/go-api/v2/milvuspb"
@@ -521,28 +522,47 @@ func (kc *Catalog) appendPartitionAndFieldsInfo(ctx context.Context, collMeta *p
 		return collection, nil
 	}
 
-	partitions, err := kc.listPartitionsAfter210(ctx, collection.CollectionID, ts)
-	if err != nil {
+	var (
+		partitions        []*model.Partition
+		fields            []*model.Field
+		structArrayFields []*model.StructArrayField
+		functions         []*model.Function
+	)
+
+	g, gCtx := errgroup.WithContext(ctx)
+	collectionID := collection.CollectionID
+
+	g.Go(func() error {
+		var err error
+		partitions, err = kc.listPartitionsAfter210(gCtx, collectionID, ts)
+		return err
+	})
+
+	g.Go(func() error {
+		var err error
+		fields, err = kc.listFieldsAfter210(gCtx, collectionID, ts)
+		return err
+	})
+
+	g.Go(func() error {
+		var err error
+		structArrayFields, err = kc.listStructArrayFieldsAfter210(gCtx, collectionID, ts)
+		return err
+	})
+
+	g.Go(func() error {
+		var err error
+		functions, err = kc.listFunctions(gCtx, collectionID, ts)
+		return err
+	})
+
+	if err := g.Wait(); err != nil {
 		return nil, err
 	}
+
 	collection.Partitions = partitions
-
-	fields, err := kc.listFieldsAfter210(ctx, collection.CollectionID, ts)
-	if err != nil {
-		return nil, err
-	}
 	collection.Fields = fields
-
-	structArrayFields, err := kc.listStructArrayFieldsAfter210(ctx, collection.CollectionID, ts)
-	if err != nil {
-		return nil, err
-	}
 	collection.StructArrayFields = structArrayFields
-
-	functions, err := kc.listFunctions(ctx, collection.CollectionID, ts)
-	if err != nil {
-		return nil, err
-	}
 	collection.Functions = functions
 	return collection, nil
 }

internal/metastore/kv/rootcoord/kv_catalog_test.go

@@ -152,6 +152,21 @@ func TestCatalog_ListCollections(t *testing.T) {
 				return strings.HasPrefix(prefix, FieldMetaPrefix)
 			}), ts).
 			Return(nil, nil, targetErr)
+
+		// appendPartitionAndFieldsInfo runs all 4 list calls in parallel,
+		// so structArrayFields and functions calls also need mock expectations.
+		kv.On("LoadWithPrefix", mock.Anything, mock.MatchedBy(
+			func(prefix string) bool {
+				return strings.HasPrefix(prefix, StructArrayFieldMetaPrefix)
+			}), ts).
+			Return([]string{}, []string{}, nil).Maybe()
+
+		kv.On("LoadWithPrefix", mock.Anything, mock.MatchedBy(
+			func(prefix string) bool {
+				return strings.HasPrefix(prefix, FunctionMetaPrefix)
+			}), ts).
+			Return([]string{}, []string{}, nil).Maybe()
+
 		kc := NewCatalog(nil, kv)
 
 		ret, err := kc.ListCollections(ctx, util.NonDBID, ts)

internal/metastore/mocks/mock_datacoord_catalog.go

@@ -2144,9 +2144,9 @@ func (_c *DataCoordCatalog_ListPreImportTasks_Call) RunAndReturn(run func(contex
 	return _c
 }
 
-// ListSegmentIndexes provides a mock function with given fields: ctx
-func (_m *DataCoordCatalog) ListSegmentIndexes(ctx context.Context) ([]*model.SegmentIndex, error) {
-	ret := _m.Called(ctx)
+// ListSegmentIndexes provides a mock function with given fields: ctx, collectionID
+func (_m *DataCoordCatalog) ListSegmentIndexes(ctx context.Context, collectionID int64) ([]*model.SegmentIndex, error) {
+	ret := _m.Called(ctx, collectionID)
 
 	if len(ret) == 0 {
 		panic("no return value specified for ListSegmentIndexes")
@@ -2154,19 +2154,19 @@ func (_m *DataCoordCatalog) ListSegmentIndexes(ctx context.Context) ([]*model.Se
 
 	var r0 []*model.SegmentIndex
 	var r1 error
-	if rf, ok := ret.Get(0).(func(context.Context) ([]*model.SegmentIndex, error)); ok {
-		return rf(ctx)
+	if rf, ok := ret.Get(0).(func(context.Context, int64) ([]*model.SegmentIndex, error)); ok {
+		return rf(ctx, collectionID)
 	}
-	if rf, ok := ret.Get(0).(func(context.Context) []*model.SegmentIndex); ok {
-		r0 = rf(ctx)
+	if rf, ok := ret.Get(0).(func(context.Context, int64) []*model.SegmentIndex); ok {
+		r0 = rf(ctx, collectionID)
 	} else {
 		if ret.Get(0) != nil {
 			r0 = ret.Get(0).([]*model.SegmentIndex)
 		}
 	}
 
-	if rf, ok := ret.Get(1).(func(context.Context) error); ok {
-		r1 = rf(ctx)
+	if rf, ok := ret.Get(1).(func(context.Context, int64) error); ok {
+		r1 = rf(ctx, collectionID)
 	} else {
 		r1 = ret.Error(1)
 	}
@@ -2181,13 +2181,14 @@ type DataCoordCatalog_ListSegmentIndexes_Call struct {
 
 // ListSegmentIndexes is a helper method to define mock.On call
 //   - ctx context.Context
-func (_e *DataCoordCatalog_Expecter) ListSegmentIndexes(ctx interface{}) *DataCoordCatalog_ListSegmentIndexes_Call {
-	return &DataCoordCatalog_ListSegmentIndexes_Call{Call: _e.mock.On("ListSegmentIndexes", ctx)}
+//   - collectionID int64
+func (_e *DataCoordCatalog_Expecter) ListSegmentIndexes(ctx interface{}, collectionID interface{}) *DataCoordCatalog_ListSegmentIndexes_Call {
+	return &DataCoordCatalog_ListSegmentIndexes_Call{Call: _e.mock.On("ListSegmentIndexes", ctx, collectionID)}
 }
 
-func (_c *DataCoordCatalog_ListSegmentIndexes_Call) Run(run func(ctx context.Context)) *DataCoordCatalog_ListSegmentIndexes_Call {
+func (_c *DataCoordCatalog_ListSegmentIndexes_Call) Run(run func(ctx context.Context, collectionID int64)) *DataCoordCatalog_ListSegmentIndexes_Call {
 	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context))
+		run(args[0].(context.Context), args[1].(int64))
 	})
 	return _c
 }
@@ -2197,7 +2198,7 @@ func (_c *DataCoordCatalog_ListSegmentIndexes_Call) Return(_a0 []*model.SegmentI
 	return _c
 }
 
-func (_c *DataCoordCatalog_ListSegmentIndexes_Call) RunAndReturn(run func(context.Context) ([]*model.SegmentIndex, error)) *DataCoordCatalog_ListSegmentIndexes_Call {
+func (_c *DataCoordCatalog_ListSegmentIndexes_Call) RunAndReturn(run func(context.Context, int64) ([]*model.SegmentIndex, error)) *DataCoordCatalog_ListSegmentIndexes_Call {
 	_c.Call.Return(run)
 	return _c
 }

internal/metastore/mocks/mock_querycoord_catalog.go

@@ -579,6 +579,52 @@ func (_c *QueryCoordCatalog_RemoveCollectionTarget_Call) RunAndReturn(run func(c
 	return _c
 }
 
+// RemoveCollectionTargets provides a mock function with given fields: ctx
+func (_m *QueryCoordCatalog) RemoveCollectionTargets(ctx context.Context) error {
+	ret := _m.Called(ctx)
+
+	if len(ret) == 0 {
+		panic("no return value specified for RemoveCollectionTargets")
+	}
+
+	var r0 error
+	if rf, ok := ret.Get(0).(func(context.Context) error); ok {
+		r0 = rf(ctx)
+	} else {
+		r0 = ret.Error(0)
+	}
+
+	return r0
+}
+
+// QueryCoordCatalog_RemoveCollectionTargets_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'RemoveCollectionTargets'
+type QueryCoordCatalog_RemoveCollectionTargets_Call struct {
+	*mock.Call
+}
+
+// RemoveCollectionTargets is a helper method to define mock.On call
+//   - ctx context.Context
+func (_e *QueryCoordCatalog_Expecter) RemoveCollectionTargets(ctx interface{}) *QueryCoordCatalog_RemoveCollectionTargets_Call {
+	return &QueryCoordCatalog_RemoveCollectionTargets_Call{Call: _e.mock.On("RemoveCollectionTargets", ctx)}
+}
+
+func (_c *QueryCoordCatalog_RemoveCollectionTargets_Call) Run(run func(ctx context.Context)) *QueryCoordCatalog_RemoveCollectionTargets_Call {
+	_c.Call.Run(func(args mock.Arguments) {
+		run(args[0].(context.Context))
+	})
+	return _c
+}
+
+func (_c *QueryCoordCatalog_RemoveCollectionTargets_Call) Return(_a0 error) *QueryCoordCatalog_RemoveCollectionTargets_Call {
+	_c.Call.Return(_a0)
+	return _c
+}
+
+func (_c *QueryCoordCatalog_RemoveCollectionTargets_Call) RunAndReturn(run func(context.Context) error) *QueryCoordCatalog_RemoveCollectionTargets_Call {
+	_c.Call.Return(run)
+	return _c
+}
+
 // RemoveResourceGroup provides a mock function with given fields: ctx, rgName
 func (_m *QueryCoordCatalog) RemoveResourceGroup(ctx context.Context, rgName string) error {
 	ret := _m.Called(ctx, rgName)

internal/querycoordv2/meta/target_manager.go

@@ -575,11 +575,13 @@ func (mgr *TargetManager) Recover(ctx context.Context, catalog metastore.QueryCo
 			zap.Int("segmentNum", len(newTarget.GetAllSegmentIDs())),
 			zap.Int64("version", newTarget.GetTargetVersion()),
 		)
+	}
 
-		// clear target info in meta store
-		err := catalog.RemoveCollectionTarget(ctx, t.GetCollectionID())
-		if err != nil {
-			log.Warn("failed to clear collection target from etcd", zap.Error(err))
+	// Remove all target keys from etcd after in-memory recovery is done.
+	// Uses RemoveWithPrefix which is a single etcd call.
+	if len(targets) > 0 {
+		if err := catalog.RemoveCollectionTargets(ctx); err != nil {
+			log.Warn("failed to remove collection targets from etcd", zap.Error(err))
 		}
 	}
 

internal/querycoordv2/meta/target_manager_test.go

@@ -31,6 +31,7 @@ import (
 	etcdkv "github.com/milvus-io/milvus/internal/kv/etcd"
 	"github.com/milvus-io/milvus/internal/metastore"
 	"github.com/milvus-io/milvus/internal/metastore/kv/querycoord"
+	catalogmocks "github.com/milvus-io/milvus/internal/metastore/mocks"
 	. "github.com/milvus-io/milvus/internal/querycoordv2/params"
 	"github.com/milvus-io/milvus/internal/querycoordv2/session"
 	"github.com/milvus-io/milvus/pkg/v2/kv"
@@ -646,6 +647,68 @@ func (suite *TargetManagerSuite) TestRecover() {
 	suite.Len(targets, 0)
 }
 
+func TestRecoverGetTargetsFail(t *testing.T) {
+	paramtable.Init()
+	ctx := context.Background()
+	broker := NewMockBroker(t)
+	meta := NewMeta(RandomIncrementIDAllocator(), nil, session.NewNodeManager())
+	mgr := NewTargetManager(broker, meta)
+
+	mockCatalog := catalogmocks.NewQueryCoordCatalog(t)
+	mockCatalog.EXPECT().GetCollectionTargets(mock.Anything).Return(nil, errors.New("mock error"))
+
+	err := mgr.Recover(ctx, mockCatalog)
+	assert.Error(t, err)
+}
+
+func TestRecoverRemoveTargetsFail(t *testing.T) {
+	paramtable.Init()
+	ctx := context.Background()
+	broker := NewMockBroker(t)
+	meta := NewMeta(RandomIncrementIDAllocator(), nil, session.NewNodeManager())
+	mgr := NewTargetManager(broker, meta)
+
+	collectionID := int64(2001)
+	mockCatalog := catalogmocks.NewQueryCoordCatalog(t)
+	mockCatalog.EXPECT().GetCollectionTargets(mock.Anything).Return(map[int64]*querypb.CollectionTarget{
+		collectionID: {
+			CollectionID: collectionID,
+			Version:      1,
+			ChannelTargets: []*querypb.ChannelTarget{
+				{
+					ChannelName: "channel-1",
+				},
+			},
+		},
+	}, nil)
+	mockCatalog.EXPECT().RemoveCollectionTargets(mock.Anything).Return(errors.New("mock error"))
+
+	// Recover should succeed even if RemoveCollectionTargets fails (it only logs a warning)
+	err := mgr.Recover(ctx, mockCatalog)
+	assert.NoError(t, err)
+
+	// Target should still be recovered in memory
+	target := mgr.current.getCollectionTarget(collectionID)
+	assert.NotNil(t, target)
+	assert.Len(t, target.GetAllDmChannelNames(), 1)
+}
+
+func TestRecoverEmptyTargets(t *testing.T) {
+	paramtable.Init()
+	ctx := context.Background()
+	broker := NewMockBroker(t)
+	meta := NewMeta(RandomIncrementIDAllocator(), nil, session.NewNodeManager())
+	mgr := NewTargetManager(broker, meta)
+
+	mockCatalog := catalogmocks.NewQueryCoordCatalog(t)
+	mockCatalog.EXPECT().GetCollectionTargets(mock.Anything).Return(map[int64]*querypb.CollectionTarget{}, nil)
+	// RemoveCollectionTargets should NOT be called when targets are empty
+
+	err := mgr.Recover(ctx, mockCatalog)
+	assert.NoError(t, err)
+	mockCatalog.AssertNotCalled(t, "RemoveCollectionTargets", mock.Anything)
+}
+
 func (suite *TargetManagerSuite) TestGetTargetJSON() {
 	ctx := suite.ctx
 	collectionID := int64(1003)