enhance: reduce coordinator metadata lock contention (#51255)

issue: #51333

## Summary

- serialize DataCoord checkpoint mutations per channel so unrelated
channel catalog writes can proceed concurrently
- keep overlapping single, batch, mark, and drop operations serialized
with deterministic multi-channel lock ordering
- use a read lock for the read-only RootCoord `DescribeAlias` path
- exclude all newly added metrics and profiling instrumentation

---------

Signed-off-by: sunby <sunbingyi1992@gmail.com>
Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>

internal/datacoord/meta.go

@@ -23,6 +23,7 @@ import (
 	"fmt"
 	"math"
 	"path"
+	"sort"
 	"strconv"
 	"strings"
 	"time"
@@ -138,13 +139,15 @@ func (m *meta) GetSnapshotMeta() *snapshotMeta {
 
 type channelCPs struct {
 	lock.RWMutex
-	checkpoints map[string]*msgpb.MsgPosition
-	cond        *syncutil.ContextCond
+	checkpoints  map[string]*msgpb.MsgPosition
+	channelLocks *lock.KeyLock[string]
+	cond         *syncutil.ContextCond
 }
 
 func newChannelCps() *channelCPs {
 	cp := &channelCPs{
-		checkpoints: make(map[string]*msgpb.MsgPosition),
+		checkpoints:  make(map[string]*msgpb.MsgPosition),
+		channelLocks: lock.NewKeyLock[string](),
 	}
 	// use the same lock as channelCPs
 	cp.cond = syncutil.NewContextCond(&cp.RWMutex)
@@ -2722,16 +2725,20 @@ func (m *meta) UpdateChannelCheckpoint(ctx context.Context, vChannel string, pos
 		pos = minGrowingCP
 	}
 
-	m.channelCPs.Lock()
-	defer m.channelCPs.Unlock()
+	m.channelCPs.channelLocks.Lock(vChannel)
+	defer m.channelCPs.channelLocks.Unlock(vChannel)
 
+	m.channelCPs.RLock()
 	oldPosition, ok := m.channelCPs.checkpoints[vChannel]
+	m.channelCPs.RUnlock()
 	if !ok || oldPosition.Timestamp < pos.Timestamp || (oldPosition.Timestamp == pos.Timestamp && !bytes.Equal(oldPosition.MsgID, pos.MsgID)) {
 		err := m.catalog.SaveChannelCheckpoint(ctx, vChannel, pos)
 		if err != nil {
 			return err
 		}
+		m.channelCPs.Lock()
 		m.channelCPs.checkpoints[vChannel] = pos
+		m.channelCPs.Unlock()
 		ts, _ := tsoutil.ParseTS(pos.Timestamp)
 		mlog.Info(context.TODO(), "UpdateChannelCheckpoint done",
 			mlog.String("vChannel", vChannel),
@@ -2750,8 +2757,8 @@ func (m *meta) UpdateChannelCheckpoint(ctx context.Context, vChannel string, pos
 // UpdateChannelCheckpoint can overwrite it, and removes the channel-checkpoint
 // lag metric.
 func (m *meta) MarkChannelCheckpointDropped(ctx context.Context, channel string) error {
-	m.channelCPs.Lock()
-	defer m.channelCPs.Unlock()
+	m.channelCPs.channelLocks.Lock(channel)
+	defer m.channelCPs.channelLocks.Unlock(channel)
 
 	cp := &msgpb.MsgPosition{
 		ChannelName: channel,
@@ -2763,7 +2770,9 @@ func (m *meta) MarkChannelCheckpointDropped(ctx context.Context, channel string)
 		return err
 	}
 
+	m.channelCPs.Lock()
 	m.channelCPs.checkpoints[channel] = cp
+	m.channelCPs.Unlock()
 
 	metrics.DataCoordCheckpointUnixSeconds.DeleteLabelValues(paramtable.GetStringNodeID(), channel)
 	return nil
@@ -2786,24 +2795,52 @@ func (m *meta) UpdateChannelCheckpoints(ctx context.Context, positions []*msgpb.
 		}
 	}
 
-	m.channelCPs.Lock()
-	defer m.channelCPs.Unlock()
-	toUpdates := lo.Filter(positions, func(pos *msgpb.MsgPosition, _ int) bool {
+	validPositions := lo.Filter(positions, func(pos *msgpb.MsgPosition, _ int) bool {
 		if pos == nil || (pos.GetMsgID() == nil && pos.GetWALName() != commonpb.WALName_WoodPecker) || pos.GetChannelName() == "" {
 			mlog.Warn(context.TODO(), "illegal channel cp", mlog.Any("pos", pos))
 			return false
 		}
+		return true
+	})
+	channelSet := make(map[string]struct{}, len(validPositions))
+	for _, pos := range validPositions {
+		channelSet[pos.GetChannelName()] = struct{}{}
+	}
+	channels := make([]string, 0, len(channelSet))
+	for channel := range channelSet {
+		channels = append(channels, channel)
+	}
+	sort.Strings(channels)
+	for _, channel := range channels {
+		m.channelCPs.channelLocks.Lock(channel)
+	}
+	defer func() {
+		for i := len(channels) - 1; i >= 0; i-- {
+			m.channelCPs.channelLocks.Unlock(channels[i])
+		}
+	}()
+
+	m.channelCPs.RLock()
+	toUpdates := lo.Filter(validPositions, func(pos *msgpb.MsgPosition, _ int) bool {
 		vChannel := pos.GetChannelName()
 		oldPosition, ok := m.channelCPs.checkpoints[vChannel]
 		return !ok || oldPosition.Timestamp < pos.Timestamp || (oldPosition.Timestamp == pos.Timestamp && !bytes.Equal(oldPosition.MsgID, pos.MsgID))
 	})
+	m.channelCPs.RUnlock()
 	err := m.catalog.SaveChannelCheckpoints(ctx, toUpdates)
 	if err != nil {
 		return err
 	}
+	m.channelCPs.Lock()
 	for _, pos := range toUpdates {
 		channel := pos.GetChannelName()
 		m.channelCPs.checkpoints[channel] = pos
+	}
+	// broadcast the change of channel checkpoint for TruncateCollection op to drop segments
+	m.channelCPs.cond.UnsafeBroadcast()
+	m.channelCPs.Unlock()
+	for _, pos := range toUpdates {
+		channel := pos.GetChannelName()
 		mlog.Info(context.TODO(), "UpdateChannelCheckpoint done", mlog.String("channel", channel),
 			mlog.Stringer("walName", pos.WALName),
 			mlog.Uint64("ts", pos.GetTimestamp()),
@@ -2811,8 +2848,6 @@ func (m *meta) UpdateChannelCheckpoints(ctx context.Context, positions []*msgpb.
 		ts, _ := tsoutil.ParseTS(pos.Timestamp)
 		metrics.DataCoordCheckpointUnixSeconds.WithLabelValues(paramtable.GetStringNodeID(), channel).Set(float64(ts.Unix()))
 	}
-	// broadcast the change of channel checkpoint for TruncateCollection op to drop segments
-	m.channelCPs.cond.UnsafeBroadcast()
 	return nil
 }
 
@@ -2827,13 +2862,15 @@ func (m *meta) GetChannelCheckpoint(vChannel string) *msgpb.MsgPosition {
 }
 
 func (m *meta) DropChannelCheckpoint(vChannel string) error {
-	m.channelCPs.Lock()
-	defer m.channelCPs.Unlock()
+	m.channelCPs.channelLocks.Lock(vChannel)
+	defer m.channelCPs.channelLocks.Unlock(vChannel)
 	err := m.catalog.DropChannelCheckpoint(m.ctx, vChannel)
 	if err != nil {
 		return err
 	}
+	m.channelCPs.Lock()
 	delete(m.channelCPs.checkpoints, vChannel)
+	m.channelCPs.Unlock()
 	metrics.DataCoordCheckpointUnixSeconds.DeleteLabelValues(paramtable.GetStringNodeID(), vChannel)
 	mlog.Info(context.TODO(), "DropChannelCheckpoint done", mlog.String("vChannel", vChannel))
 	return nil

internal/datacoord/meta_test.go

@@ -4746,6 +4746,175 @@ func equalCollectionInfo(t *testing.T, a *collectionInfo, b *collectionInfo) {
 	assert.Equal(t, a.StartPositions, b.StartPositions)
 }
 
+func TestUpdateChannelCheckpoint_DifferentChannelsPersistConcurrently(t *testing.T) {
+	const (
+		channel1 = "channel-1"
+		channel2 = "channel-2"
+	)
+
+	catalog := mocks2.NewDataCoordCatalog(t)
+	channel1Entered := make(chan struct{})
+	releaseChannel1 := make(chan struct{})
+	channel2Entered := make(chan struct{})
+	catalog.EXPECT().SaveChannelCheckpoint(mock.Anything, channel1, mock.Anything).
+		RunAndReturn(func(context.Context, string, *msgpb.MsgPosition) error {
+			close(channel1Entered)
+			<-releaseChannel1
+			return nil
+		}).Once()
+	catalog.EXPECT().SaveChannelCheckpoint(mock.Anything, channel2, mock.Anything).
+		RunAndReturn(func(context.Context, string, *msgpb.MsgPosition) error {
+			close(channel2Entered)
+			return nil
+		}).Once()
+
+	meta := &meta{
+		ctx:         context.Background(),
+		catalog:     catalog,
+		collections: typeutil.NewConcurrentMap[UniqueID, *collectionInfo](),
+		segments:    NewSegmentsInfo(),
+		channelCPs:  newChannelCps(),
+	}
+	channel1Done := make(chan error, 1)
+	go func() {
+		channel1Done <- meta.UpdateChannelCheckpoint(context.Background(), channel1, &msgpb.MsgPosition{
+			ChannelName: channel1,
+			MsgID:       []byte{1},
+			Timestamp:   1,
+		})
+	}()
+
+	select {
+	case <-channel1Entered:
+	case <-time.After(3 * time.Second):
+		close(releaseChannel1)
+		select {
+		case <-channel1Done:
+		case <-time.After(3 * time.Second):
+		}
+		t.Fatal("channel-1 did not enter SaveChannelCheckpoint")
+	}
+
+	channel2Done := make(chan error, 1)
+	go func() {
+		channel2Done <- meta.UpdateChannelCheckpoint(context.Background(), channel2, &msgpb.MsgPosition{
+			ChannelName: channel2,
+			MsgID:       []byte{2},
+			Timestamp:   2,
+		})
+	}()
+
+	channel2PersistedConcurrently := false
+	select {
+	case <-channel2Entered:
+		channel2PersistedConcurrently = true
+	case <-time.After(3 * time.Second):
+	}
+
+	close(releaseChannel1)
+	errs := make(map[string]error, 2)
+	for channel, done := range map[string]<-chan error{
+		channel1: channel1Done,
+		channel2: channel2Done,
+	} {
+		select {
+		case err := <-done:
+			errs[channel] = err
+		case <-time.After(3 * time.Second):
+			assert.Failf(t, "checkpoint update did not finish", "channel: %s", channel)
+		}
+	}
+	for _, err := range errs {
+		assert.NoError(t, err)
+	}
+	require.True(t, channel2PersistedConcurrently, "channel-2 did not enter catalog while channel-1 was blocked")
+}
+
+func TestUpdateChannelCheckpoints_SerializesWithSingleUpdateOnSameChannel(t *testing.T) {
+	const channel = "channel-1"
+
+	catalog := mocks2.NewDataCoordCatalog(t)
+	batchEntered := make(chan struct{})
+	releaseBatch := make(chan struct{})
+	singleEntered := make(chan struct{})
+	catalog.EXPECT().SaveChannelCheckpoints(mock.Anything, mock.Anything).
+		RunAndReturn(func(context.Context, []*msgpb.MsgPosition) error {
+			close(batchEntered)
+			<-releaseBatch
+			return nil
+		}).Once()
+	catalog.EXPECT().SaveChannelCheckpoint(mock.Anything, channel, mock.Anything).
+		RunAndReturn(func(context.Context, string, *msgpb.MsgPosition) error {
+			close(singleEntered)
+			return nil
+		}).Once()
+
+	meta := &meta{
+		ctx:         context.Background(),
+		catalog:     catalog,
+		collections: typeutil.NewConcurrentMap[UniqueID, *collectionInfo](),
+		segments:    NewSegmentsInfo(),
+		channelCPs:  newChannelCps(),
+	}
+	batchDone := make(chan error, 1)
+	go func() {
+		batchDone <- meta.UpdateChannelCheckpoints(context.Background(), []*msgpb.MsgPosition{{
+			ChannelName: channel,
+			MsgID:       []byte{1},
+			Timestamp:   1,
+		}})
+	}()
+
+	select {
+	case <-batchEntered:
+	case <-time.After(3 * time.Second):
+		close(releaseBatch)
+		select {
+		case <-batchDone:
+		case <-time.After(3 * time.Second):
+		}
+		t.Fatal("batch update did not enter SaveChannelCheckpoints")
+	}
+
+	singleStarted := make(chan struct{})
+	singleDone := make(chan error, 1)
+	go func() {
+		close(singleStarted)
+		singleDone <- meta.UpdateChannelCheckpoint(context.Background(), channel, &msgpb.MsgPosition{
+			ChannelName: channel,
+			MsgID:       []byte{2},
+			Timestamp:   2,
+		})
+	}()
+	<-singleStarted
+
+	singleEnteredBeforeBatchRelease := false
+	select {
+	case <-singleEntered:
+		singleEnteredBeforeBatchRelease = true
+	case <-time.After(2 * time.Second):
+	}
+
+	close(releaseBatch)
+	errs := make([]error, 0, 2)
+	for operation, done := range map[string]<-chan error{
+		"batch":  batchDone,
+		"single": singleDone,
+	} {
+		select {
+		case err := <-done:
+			errs = append(errs, err)
+		case <-time.After(3 * time.Second):
+			assert.Failf(t, "checkpoint update did not finish", "operation: %s", operation)
+		}
+	}
+	for _, err := range errs {
+		assert.NoError(t, err)
+	}
+	assert.False(t, singleEnteredBeforeBatchRelease, "single update entered catalog before the overlapping batch completed")
+	require.Equal(t, uint64(2), meta.GetChannelCheckpoint(channel).GetTimestamp())
+}
+
 func TestChannelCP(t *testing.T) {
 	mockVChannel := "fake-by-dev-rootcoord-dml-1-testchannelcp-v0"
 	mockPChannel := "fake-by-dev-rootcoord-dml-1"

internal/rootcoord/meta_table.go

@@ -1592,8 +1592,8 @@ func (mt *MetaTable) CheckIfAliasAlterable(ctx context.Context, dbName string, a
 }
 
 func (mt *MetaTable) DescribeAlias(ctx context.Context, dbName string, alias string, ts Timestamp) (string, error) {
-	mt.ddLock.Lock()
-	defer mt.ddLock.Unlock()
+	mt.ddLock.RLock()
+	defer mt.ddLock.RUnlock()
 
 	if dbName == "" {
 		mlog.Warn(ctx, "db name is empty", mlog.String("alias", alias))

internal/rootcoord/meta_table_test.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"math/rand"
 	"testing"
+	"time"
 
 	"github.com/cockroachdb/errors"
 	"github.com/stretchr/testify/assert"
@@ -51,6 +52,46 @@ import (
 	"github.com/milvus-io/milvus/pkg/v3/util/typeutil"
 )
 
+func TestMetaTable_DescribeAliasAllowsConcurrentReaders(t *testing.T) {
+	const (
+		collectionID   = int64(100)
+		collectionName = "test_metatable_describe_alias"
+		aliasName      = "a_alias"
+	)
+	meta := &MetaTable{
+		collID2Meta: map[typeutil.UniqueID]*model.Collection{
+			collectionID: {
+				CollectionID: collectionID,
+				Name:         collectionName,
+			},
+		},
+		aliases: newNameDb(),
+	}
+	meta.aliases.insert(util.DefaultDBName, aliasName, collectionID)
+
+	type result struct {
+		collectionName string
+		err            error
+	}
+	resultCh := make(chan result, 1)
+	meta.ddLock.RLock()
+	go func() {
+		collectionName, err := meta.DescribeAlias(context.Background(), util.DefaultDBName, aliasName, 0)
+		resultCh <- result{collectionName: collectionName, err: err}
+	}()
+
+	select {
+	case result := <-resultCh:
+		meta.ddLock.RUnlock()
+		require.NoError(t, result.err)
+		assert.Equal(t, collectionName, result.collectionName)
+	case <-time.After(3 * time.Second):
+		meta.ddLock.RUnlock()
+		<-resultCh
+		t.Fatal("DescribeAlias blocked behind another reader")
+	}
+}
+
 func generateMetaTable(_ *testing.T) *MetaTable {
 	kv, _ := kvfactory.GetEtcdAndPath()
 	path := funcutil.RandomString(10)