Repository: milvus-io/milvus
File: internal/querycoordv2/meta/collection_manager.go (Go)

Code before:
}

func NewCollectionManager(store Store) *CollectionManager {
	return &CollectionManager{
		collections: make(map[int64]*Collection),
		partitions:  make(map[int64]*Partition),
		store:       store,
	}
}

// Recover recovers collections from kv store,
// panics if failed
func (m *CollectionManager) Recover(broker Broker) error {
	collections, err := m.store.GetCollections()
	if err != nil {
		return err
	}
	partitions, err := m.store.GetPartitions()
	if err != nil {
		return err
	}

	for _, collection := range collections {
		// Collections not loaded done should be deprecated
		if collection.GetStatus() != querypb.LoadStatus_Loaded || collection.GetReplicaNumber() <= 0 {
			log.Info("skip recovery and release collection",
				zap.Int64("collectionID", collection.GetCollectionID()),
				zap.String("status", collection.GetStatus().String()),
				zap.Int32("replicaNumber", collection.GetReplicaNumber()),
			)
			m.store.ReleaseCollection(collection.GetCollectionID())
			continue
		}
		m.collections[collection.CollectionID] = &Collection{
			CollectionLoadInfo: collection,
		}
	}

	for collection, partitions := range partitions {
		sawLoaded := false
		for _, partition := range partitions {
			// Partitions not loaded done should be deprecated
			if partition.GetStatus() != querypb.LoadStatus_Loaded || partition.GetReplicaNumber() <= 0 {
				log.Info("skip recovery and release partition",
					zap.Int64("collectionID", collection),
					zap.Int64("partitionID", partition.GetPartitionID()),
					zap.String("status", partition.GetStatus().String()),
					zap.Int32("replicaNumber", partition.GetReplicaNumber()),
				)
				m.store.ReleasePartition(collection, partition.GetPartitionID())
				continue

Review comment (performance):
this is a performance bottleneck if many collections loaded, may optimize it by concurrently recover collections

Code after:
	store       Store
}

func NewCollectionManager(store Store) *CollectionManager {
	return &CollectionManager{
		collections: make(map[int64]*Collection),
		partitions:  make(map[int64]*Partition),
		store:       store,
	}
}

// Recover recovers collections from kv store,
// panics if failed
func (m *CollectionManager) Recover(broker Broker) error {
	collections, err := m.store.GetCollections()
	if err != nil {
		return err
	}
	partitions, err := m.store.GetPartitions()
	if err != nil {
		return err
	}

	for _, collection := range collections {
		// Dropped collection should be deprecated
		_, err = broker.GetCollectionSchema(context.Background(), collection.GetCollectionID())
		if errors.Is(err, merr.ErrCollectionNotFound) {
			log.Info("skip dropped collection during recovery", zap.Int64("collection", collection.GetCollectionID()))
			m.store.ReleaseCollection(collection.GetCollectionID())
			continue
		}
		if err != nil {
			return err
		}
		// Collections not loaded done should be deprecated
		if collection.GetStatus() != querypb.LoadStatus_Loaded || collection.GetReplicaNumber() <= 0 {
			log.Info("skip recovery and release collection",
				zap.Int64("collectionID", collection.GetCollectionID()),
				zap.String("status", collection.GetStatus().String()),
				zap.Int32("replicaNumber", collection.GetReplicaNumber()),
			)
			m.store.ReleaseCollection(collection.GetCollectionID())
			continue
		}
		m.collections[collection.CollectionID] = &Collection{
			CollectionLoadInfo: collection,
		}
	}

	for collection, partitions := range partitions {
		existPartitions, err := broker.GetPartitions(context.Background(), collection)