diff --git a/internal/coordinator/mix_coord.go b/internal/coordinator/mix_coord.go index 2b32af73683..49db314d979 100644 --- a/internal/coordinator/mix_coord.go +++ b/internal/coordinator/mix_coord.go @@ -549,6 +549,11 @@ func (s *mixCoordImpl) HasCollection(ctx context.Context, req *milvuspb.HasColle return s.rootcoordServer.HasCollection(ctx, req) } +// IsCollectionAvailable is an in-process, positive-only GC fast path. +func (s *mixCoordImpl) IsCollectionAvailable(collectionID int64) bool { + return s.rootcoordServer.IsCollectionAvailable(collectionID) +} + func (s *mixCoordImpl) DescribeCollection(ctx context.Context, req *milvuspb.DescribeCollectionRequest) (*milvuspb.DescribeCollectionResponse, error) { return s.rootcoordServer.DescribeCollection(ctx, req) } diff --git a/internal/coordinator/mix_coord_test.go b/internal/coordinator/mix_coord_test.go index 2a802faae7d..e485d1b4765 100644 --- a/internal/coordinator/mix_coord_test.go +++ b/internal/coordinator/mix_coord_test.go @@ -32,6 +32,7 @@ import ( "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" "github.com/milvus-io/milvus-proto/go-api/v3/milvuspb" "github.com/milvus-io/milvus/internal/datacoord" + "github.com/milvus-io/milvus/internal/datacoord/broker" "github.com/milvus-io/milvus/internal/querycoordv2" "github.com/milvus-io/milvus/internal/rootcoord" "github.com/milvus-io/milvus/internal/util/dependency" @@ -47,6 +48,19 @@ import ( "github.com/milvus-io/milvus/pkg/v3/util/tikv" ) +func TestMixCoordCollectionAvailability(t *testing.T) { + mockey.PatchConvey("broker uses the in-process rootcoord availability check", t, func() { + core := &rootcoord.Core{} + mockey.Mock((*rootcoord.Core).IsCollectionAvailable).To(func(receiver *rootcoord.Core, id int64) bool { + assert.Same(t, core, receiver) + return id == 123 + }).Build() + b := broker.NewCoordinatorBroker(&mixCoordImpl{rootcoordServer: core}) + assert.True(t, b.IsCollectionAvailable(123)) + assert.False(t, b.IsCollectionAvailable(124)) + }) +} + func TestMixcoord_EnableActiveStandby(t *testing.T) { randVal := rand.Int() paramtable.Init() diff --git a/internal/datacoord/broker/coordinator_broker.go b/internal/datacoord/broker/coordinator_broker.go index 456c4ba05e1..18ca5ec92df 100644 --- a/internal/datacoord/broker/coordinator_broker.go +++ b/internal/datacoord/broker/coordinator_broker.go @@ -66,6 +66,17 @@ type coordinatorBroker struct { mixCoord types.MixCoord } +// CollectionAvailability is an optional in-process fast path. Only true is +// conclusive; false must fall back to HasCollection before deciding to GC. +type CollectionAvailability interface { + IsCollectionAvailable(collectionID int64) bool +} + +func (b *coordinatorBroker) IsCollectionAvailable(collectionID int64) bool { + checker, ok := b.mixCoord.(CollectionAvailability) + return ok && checker.IsCollectionAvailable(collectionID) +} + func NewCoordinatorBroker(mixCoord types.MixCoord) *coordinatorBroker { return &coordinatorBroker{ mixCoord: mixCoord, diff --git a/internal/datacoord/checkpoint_availability_test.go b/internal/datacoord/checkpoint_availability_test.go new file mode 100644 index 00000000000..ea1450270b9 --- /dev/null +++ b/internal/datacoord/checkpoint_availability_test.go @@ -0,0 +1,172 @@ +// Licensed to the LF AI & Data foundation under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package datacoord + +import ( + "context" + "testing" + "time" + + "github.com/stretchr/testify/mock" + "github.com/stretchr/testify/require" + + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus-proto/go-api/v3/milvuspb" + "github.com/milvus-io/milvus-proto/go-api/v3/msgpb" + "github.com/milvus-io/milvus/internal/datacoord/broker" + catalogmocks "github.com/milvus-io/milvus/internal/metastore/mocks" + "github.com/milvus-io/milvus/internal/types" + "github.com/milvus-io/milvus/pkg/v3/util/merr" +) + +type checkpointAvailabilityCoord struct { + types.MixCoord + available bool + checks int +} + +func (c *checkpointAvailabilityCoord) IsCollectionAvailable(int64) bool { + c.checks++ + return c.available +} + +func TestCheckpointAvailabilityFastPath(t *testing.T) { + catalog := catalogmocks.NewDataCoordCatalog(t) + coord := &checkpointAvailabilityCoord{available: true} + checkpoints := map[string]*msgpb.MsgPosition{ + "cluster-rootcoord-dm_0_123v0": nil, + "cluster-rootcoord-dm_0_124v0": nil, + } + catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(checkpoints, nil).Times(3) + m := &meta{catalog: catalog, channelCPs: newChannelCps()} + m.channelCPs.checkpoints = checkpoints + gc := newGarbageCollector(m, newMockHandlerWithMeta(m), GcOption{broker: broker.NewCoordinatorBroker(coord)}) + gc.recycleChannelCPMeta(context.Background(), nil) + require.Equal(t, 2, coord.checks) + require.Len(t, m.channelCPs.checkpoints, 2) + // The embedded nil RPC interface and strict catalog mock reject any + // DescribeCollection/GcConfirm/Drop call on a positive cache hit. + ctx, cancel := context.WithCancel(context.Background()) + cancel() + gc.recycleChannelCPMeta(ctx, nil) + require.Equal(t, 2, coord.checks, "a canceled sweep must stop before checking live IDs") + pauseRecords := NewGCPauseRecords() + _, err := pauseRecords.Insert("test", time.Now().Add(time.Minute)) + require.NoError(t, err) + gc.pausedCollection.Insert(123, pauseRecords) + gc.recycleChannelCPMeta(context.Background(), nil) + require.Equal(t, 3, coord.checks, "paused collections must not reach the fast path") +} + +type checkpointLookupCoord struct { + types.MixCoord + describe func(context.Context, *milvuspb.DescribeCollectionRequest) (*milvuspb.DescribeCollectionResponse, error) +} + +func (c *checkpointLookupCoord) DescribeCollection(ctx context.Context, req *milvuspb.DescribeCollectionRequest) (*milvuspb.DescribeCollectionResponse, error) { + return c.describe(ctx, req) +} + +type checkpointCacheMissCoord struct { + *checkpointLookupCoord +} + +func (c *checkpointCacheMissCoord) IsCollectionAvailable(int64) bool { return false } + +func TestCheckpointAvailabilityFallback(t *testing.T) { + for _, withCache := range []bool{false, true} { + for _, tc := range []struct { + name string + status *commonpb.Status + rpcErr error + confirm bool + dropped bool + }{ + {"available", merr.Success(), nil, false, false}, + {"missing-unconfirmed", merr.Status(merr.WrapErrCollectionNotFound(123)), nil, false, false}, + {"missing-confirmed", merr.Status(merr.WrapErrCollectionNotFound(123)), nil, true, true}, + {"not-ready", merr.Status(merr.ErrServiceNotReady), nil, false, false}, + {"deadline", nil, context.DeadlineExceeded, false, false}, + } { + name := tc.name + "/no-cache" + if withCache { + name = tc.name + "/cache-miss" + } + t.Run(name, func(t *testing.T) { + catalog := catalogmocks.NewDataCoordCatalog(t) + checkpoints := map[string]*msgpb.MsgPosition{ + "cluster-rootcoord-dm_0_123v0": nil, + "cluster-rootcoord-dm_1_123v0": nil, + "cluster-rootcoord-dm_0_124v0": nil, + } + catalog.EXPECT().ListChannelCheckpoint(mock.Anything).Return(checkpoints, nil).Once() + calls := 0 + var previousOuter context.Context + coord := &checkpointLookupCoord{describe: func(ctx context.Context, req *milvuspb.DescribeCollectionRequest) (*milvuspb.DescribeCollectionResponse, error) { + calls++ + require.Empty(t, req.CollectionName) + require.Contains(t, []int64{123, 124}, req.CollectionID) + return &milvuspb.DescribeCollectionResponse{Status: tc.status}, tc.rpcErr + }} + var mixCoord types.MixCoord = coord + if withCache { + mixCoord = &checkpointCacheMissCoord{coord} + } + // Capture the GC's outer timeout as well as exercising the real + // broker's status/error classification and ID-based request. + actualBroker := broker.NewCoordinatorBroker(mixCoord) + wrapped := &checkpointContextBroker{Broker: actualBroker, check: func(ctx context.Context) { + if previousOuter != nil { + require.ErrorIs(t, previousOuter.Err(), context.Canceled) + } + previousOuter = ctx + }} + if tc.rpcErr == nil && merr.Code(merr.Error(tc.status)) == merr.Code(merr.ErrCollectionNotFound) { + catalog.EXPECT().GcConfirm(mock.Anything, mock.Anything, int64(-1)).Return(tc.confirm).Twice() + } + if tc.dropped { + catalog.EXPECT().DropChannelCheckpoint(mock.Anything, mock.Anything).Return(nil).Times(3) + } + m := &meta{catalog: catalog, channelCPs: newChannelCps()} + m.channelCPs.checkpoints = checkpoints + gc := newGarbageCollector(m, newMockHandlerWithMeta(m), GcOption{broker: wrapped}) + gc.recycleChannelCPMeta(context.Background(), nil) + require.Equal(t, 2, calls, "fallback results must be reused across channels of one collection") + require.ErrorIs(t, previousOuter.Err(), context.Canceled) + if tc.dropped { + require.Empty(t, m.channelCPs.checkpoints) + } else { + require.Len(t, m.channelCPs.checkpoints, 3) + } + }) + } + } +} + +type checkpointContextBroker struct { + broker.Broker + check func(context.Context) +} + +func (b *checkpointContextBroker) HasCollection(ctx context.Context, id int64) (bool, error) { + b.check(ctx) + return b.Broker.HasCollection(ctx, id) +} + +func (b *checkpointContextBroker) IsCollectionAvailable(id int64) bool { + return b.Broker.(broker.CollectionAvailability).IsCollectionAvailable(id) +} diff --git a/internal/datacoord/garbage_collector.go b/internal/datacoord/garbage_collector.go index bc6e3f1ee26..e25e0425164 100644 --- a/internal/datacoord/garbage_collector.go +++ b/internal/datacoord/garbage_collector.go @@ -1681,6 +1681,7 @@ func (gc *garbageCollector) recycleChannelCPMeta(ctx context.Context, signal <-c } collectionID2GcStatus := make(map[int64]bool) + availability, _ := gc.option.broker.(broker.CollectionAvailability) skippedCnt := 0 mlog.Info(ctx, "start to GC channel cp", mlog.Int("vchannelCPCnt", len(channelCPs))) @@ -1699,15 +1700,21 @@ func (gc *garbageCollector) recycleChannelCPMeta(ctx context.Context, signal <-c continue } + if ctx.Err() != nil { + return + } + // A positive resident lookup avoids schema conversion, task scheduling, + // per-collection timeouts and retaining live IDs for the entire sweep. + if availability != nil && availability.IsCollectionAvailable(collectionID) { + skippedCnt++ + continue + } + _, ok := collectionID2GcStatus[collectionID] if !ok { - if ctx.Err() != nil { - // process canceled, stop. - return - } timeoutCtx, cancel := context.WithTimeout(ctx, 3*time.Second) - defer cancel() has, err := gc.option.broker.HasCollection(timeoutCtx, collectionID) + cancel() if err == nil && !has { collectionID2GcStatus[collectionID] = gc.meta.catalog.GcConfirm(ctx, collectionID, -1) } else { diff --git a/internal/datacoord/index_meta.go b/internal/datacoord/index_meta.go index fa665d0006b..7ea3fada62e 100644 --- a/internal/datacoord/index_meta.go +++ b/internal/datacoord/index_meta.go @@ -83,6 +83,7 @@ type indexMeta struct { fieldIndexLock sync.RWMutex indexes map[UniqueID]map[UniqueID]*model.Index storedIndexSize storedIndexSizeTracker + taskCounts indexTaskCounts collectionOnce sync.Once collectionLock *lock.KeyLock[UniqueID] @@ -213,6 +214,7 @@ func (m *indexMeta) reloadFromKV() error { var segmentIndexes []*model.SegmentIndex recoveredIndexSizes := make(map[UniqueID]map[UniqueID]uint64) + var recoveredTaskCounts indexTaskCounts g, _ := errgroup.WithContext(m.ctx) g.Go(func() error { fieldIndexScanStart := time.Now() @@ -269,6 +271,8 @@ func (m *indexMeta) reloadFromKV() error { indexes.Insert(segIdx.IndexID, segIdx) m.segmentIndexes.Insert(segIdx.SegmentID, indexes) } + old, _ := m.segmentBuildInfo.Get(segIdx.BuildID) + recoveredTaskCounts.replaceTask(old, segIdx, nil) m.segmentBuildInfo.AddForRecovery(segIdx) completed := recoveredSegmentIndexes + 1 if completed%segmentIndexCacheRecoveryProgressLogInterval == 0 && completed < len(segmentIndexes) { @@ -291,6 +295,14 @@ func (m *indexMeta) reloadFromKV() error { // have completed. DropIndex then updates the gauge without scanning all // segment indexes. m.storedIndexSize.recover(m.indexes, recoveredIndexSizes) + // Both walkers have finished: activate counts only for surviving indexes. + // Recovery folds counts into the existing build pass, never a second List. + for _, indexes := range m.indexes { + for _, index := range indexes { + recoveredTaskCounts.replaceIndex(nil, index) + } + } + m.taskCounts = recoveredTaskCounts // Update the index file-count histogram asynchronously. The stored-size // gauge is initialized synchronously above so DDL callbacks can safely use @@ -306,6 +318,7 @@ func (m *indexMeta) reloadFromKV() error { } func (m *indexMeta) updateCollectionIndex(index *model.Index) { + m.taskCounts.replaceIndex(m.indexes[index.CollectionID][index.IndexID], index) if _, ok := m.indexes[index.CollectionID]; !ok { m.indexes[index.CollectionID] = make(map[UniqueID]*model.Index) } @@ -358,6 +371,8 @@ func (m *indexMeta) lockCollections(collectionIDs ...UniqueID) func() { } func (m *indexMeta) updateSegmentIndex(segIdx *model.SegmentIndex) { + old, _ := m.segmentBuildInfo.Get(segIdx.BuildID) + m.taskCounts.replaceTask(old, segIdx, m.indexes) indexes, ok := m.segmentIndexes.Get(segIdx.SegmentID) if ok { indexes.Insert(segIdx.IndexID, segIdx) @@ -393,39 +408,11 @@ func (m *indexMeta) updateSegIndexMeta(segIdx *model.SegmentIndex, updateFunc fu } func (m *indexMeta) updateIndexTasksMetrics() { - taskMetrics := make(map[indexpb.JobState]int) - taskMetrics[indexpb.JobState_JobStateNone] = 0 - taskMetrics[indexpb.JobState_JobStateInit] = 0 - taskMetrics[indexpb.JobState_JobStateInProgress] = 0 - taskMetrics[indexpb.JobState_JobStateFinished] = 0 - taskMetrics[indexpb.JobState_JobStateFailed] = 0 - taskMetrics[indexpb.JobState_JobStateRetry] = 0 - for _, segIdx := range m.segmentBuildInfo.List() { - if segIdx.IsDeleted || !m.IsIndexExist(segIdx.CollectionID, segIdx.IndexID) { - continue - } - - switch segIdx.IndexState { - case commonpb.IndexState_IndexStateNone: - taskMetrics[indexpb.JobState_JobStateNone]++ - case commonpb.IndexState_Unissued: - taskMetrics[indexpb.JobState_JobStateInit]++ - case commonpb.IndexState_InProgress: - taskMetrics[indexpb.JobState_JobStateInProgress]++ - case commonpb.IndexState_Finished: - taskMetrics[indexpb.JobState_JobStateFinished]++ - case commonpb.IndexState_Failed: - taskMetrics[indexpb.JobState_JobStateFailed]++ - case commonpb.IndexState_Retry: - taskMetrics[indexpb.JobState_JobStateRetry]++ - } - } - + taskMetrics := m.indexTaskCountsSnapshot() jobType := indexpb.JobType_JobTypeIndexJob.String() - for k, v := range taskMetrics { - metrics.IndexStatsTaskNum.WithLabelValues(jobType, k.String()).Set(float64(v)) + for state, count := range taskMetrics { + metrics.IndexStatsTaskNum.WithLabelValues(jobType, indexTaskMetricStates[state].String()).Set(float64(count)) } - mlog.Info(m.ctx, "update index metric", mlog.Int("collectionNum", len(taskMetrics))) } func checkIdenticalJSON(index *model.Index, req *indexpb.CreateIndexRequest) bool { @@ -911,7 +898,7 @@ func (m *indexMeta) MarkIndexAsDeleted(ctx context.Context, collID UniqueID, ind deletedIndexIDs := make([]UniqueID, 0, len(indexes)) m.fieldIndexLock.Lock() for _, index := range indexes { - m.indexes[index.CollectionID][index.IndexID] = index + m.updateCollectionIndex(index) deletedIndexIDs = append(deletedIndexIDs, index.IndexID) } m.fieldIndexLock.Unlock() @@ -1398,6 +1385,7 @@ func (m *indexMeta) RemoveSegmentIndexes(ctx context.Context, candidates []*mode } } } + m.taskCounts.replaceTask(segIdx, nil, m.indexes) m.segmentBuildInfo.Remove(segIdx.BuildID) } m.fieldIndexLock.Unlock() @@ -1437,6 +1425,7 @@ func (m *indexMeta) RemoveSegmentIndex(ctx context.Context, buildID UniqueID) er } } + m.taskCounts.replaceTask(segIdx, nil, m.indexes) m.segmentBuildInfo.Remove(buildID) m.fieldIndexLock.Unlock() @@ -1471,6 +1460,7 @@ func (m *indexMeta) RemoveIndex(ctx context.Context, collID, indexID UniqueID) e } m.fieldIndexLock.Lock() + m.taskCounts.replaceIndex(m.indexes[collID][indexID], nil) delete(m.indexes[collID], indexID) collectionRemoved := len(m.indexes[collID]) == 0 if collectionRemoved { @@ -1541,6 +1531,7 @@ func (m *indexMeta) RemoveIndexes(ctx context.Context, candidates []*model.Index removedCollections := make(map[UniqueID]struct{}) m.fieldIndexLock.Lock() for _, index := range currentIndexes { + m.taskCounts.replaceIndex(index, nil) delete(m.indexes[index.CollectionID], index.IndexID) if len(m.indexes[index.CollectionID]) == 0 { delete(m.indexes, index.CollectionID) diff --git a/internal/datacoord/index_task_counts.go b/internal/datacoord/index_task_counts.go new file mode 100644 index 00000000000..ace9b3e6416 --- /dev/null +++ b/internal/datacoord/index_task_counts.go @@ -0,0 +1,116 @@ +// Licensed to the LF AI & Data foundation under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package datacoord + +import ( + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus/internal/metastore/model" + "github.com/milvus-io/milvus/pkg/v3/proto/indexpb" +) + +var indexTaskMetricStates = [...]indexpb.JobState{ + indexpb.JobState_JobStateNone, + indexpb.JobState_JobStateInit, + indexpb.JobState_JobStateInProgress, + indexpb.JobState_JobStateFinished, + indexpb.JobState_JobStateFailed, + indexpb.JobState_JobStateRetry, +} + +type indexTaskStateCounts [len(indexTaskMetricStates)]int64 + +// indexTaskCounts is protected by indexMeta.fieldIndexLock, together with +// resident metadata publication. It adds no per-build objects or pointers. +// Raw counts include orphaned/dropped field indexes, but exclude deleted +// tasks. Keeping them allows index DDL to adjust totals without visiting builds. +type indexTaskCounts struct { + byIndex map[[2]int64]indexTaskStateCounts + total indexTaskStateCounts +} + +func indexTaskStateSlot(state commonpb.IndexState) int { + switch state { + case commonpb.IndexState_IndexStateNone: + return 0 + case commonpb.IndexState_Unissued: + return 1 + case commonpb.IndexState_InProgress: + return 2 + case commonpb.IndexState_Finished: + return 3 + case commonpb.IndexState_Failed: + return 4 + case commonpb.IndexState_Retry: + return 5 + default: + return -1 + } +} + +func (c *indexTaskCounts) adjustTask(task *model.SegmentIndex, delta int64, indexes map[UniqueID]map[UniqueID]*model.Index) { + if task == nil || task.IsDeleted { + return + } + slot := indexTaskStateSlot(task.IndexState) + if slot < 0 { + return + } + if c.byIndex == nil { + c.byIndex = make(map[[2]int64]indexTaskStateCounts) + } + key := [2]int64{task.CollectionID, task.IndexID} + counts := c.byIndex[key] + counts[slot] += delta + if counts == (indexTaskStateCounts{}) { + delete(c.byIndex, key) + } else { + c.byIndex[key] = counts + } + if index := indexes[task.CollectionID][task.IndexID]; index != nil && !index.IsDeleted { + c.total[slot] += delta + } +} + +func (c *indexTaskCounts) replaceTask(old, current *model.SegmentIndex, indexes map[UniqueID]map[UniqueID]*model.Index) { + if old != nil && current != nil && old.CollectionID == current.CollectionID && + old.IndexID == current.IndexID && old.IndexState == current.IndexState && old.IsDeleted == current.IsDeleted { + return + } + c.adjustTask(old, -1, indexes) + c.adjustTask(current, 1, indexes) +} + +func (c *indexTaskCounts) replaceIndex(old, current *model.Index) { + for _, change := range [...]struct { + index *model.Index + delta int64 + }{{old, -1}, {current, 1}} { + if change.index == nil || change.index.IsDeleted { + continue + } + counts := c.byIndex[[2]int64{change.index.CollectionID, change.index.IndexID}] + for state, count := range counts { + c.total[state] += count * change.delta + } + } +} + +func (m *indexMeta) indexTaskCountsSnapshot() indexTaskStateCounts { + m.fieldIndexLock.RLock() + defer m.fieldIndexLock.RUnlock() + return m.taskCounts.total +} diff --git a/internal/datacoord/index_task_counts_test.go b/internal/datacoord/index_task_counts_test.go new file mode 100644 index 00000000000..d43e744acf6 --- /dev/null +++ b/internal/datacoord/index_task_counts_test.go @@ -0,0 +1,384 @@ +// Licensed to the LF AI & Data foundation under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package datacoord + +import ( + "context" + "fmt" + "math/rand" + "sync" + "testing" + + "github.com/prometheus/client_golang/prometheus/testutil" + "github.com/stretchr/testify/require" + + "github.com/milvus-io/milvus-proto/go-api/v3/commonpb" + "github.com/milvus-io/milvus/internal/metastore" + "github.com/milvus-io/milvus/internal/metastore/model" + "github.com/milvus-io/milvus/pkg/v3/metrics" + "github.com/milvus-io/milvus/pkg/v3/proto/indexpb" + "github.com/milvus-io/milvus/pkg/v3/proto/workerpb" + "github.com/milvus-io/milvus/pkg/v3/util/merr" + "github.com/milvus-io/milvus/pkg/v3/util/typeutil" +) + +type taskCountsCatalog struct { + metastore.DataCoordCatalog + indexes []*model.Index + tasks []*model.SegmentIndex + err error +} + +func (c *taskCountsCatalog) ListIndexes(context.Context) ([]*model.Index, error) { + return c.indexes, c.err +} + +func (c *taskCountsCatalog) ListSegmentIndexes(context.Context, int64) ([]*model.SegmentIndex, error) { + return c.tasks, c.err +} +func (c *taskCountsCatalog) CreateIndex(context.Context, *model.Index) error { return c.err } +func (c *taskCountsCatalog) AlterIndexes(context.Context, []*model.Index) error { return c.err } +func (c *taskCountsCatalog) CreateSegmentIndex(context.Context, *model.SegmentIndex) error { + return c.err +} + +func (c *taskCountsCatalog) AlterSegmentIndexes(context.Context, []*model.SegmentIndex) error { + return c.err +} + +func (c *taskCountsCatalog) DropSegmentIndex(context.Context, int64, int64, int64, int64) error { + return c.err +} + +func (c *taskCountsCatalog) DropSegmentIndexes(context.Context, []*model.SegmentIndex) error { + return c.err +} +func (c *taskCountsCatalog) DropIndex(context.Context, int64, int64) error { return c.err } +func (c *taskCountsCatalog) DropIndexes(context.Context, []*model.Index) error { return c.err } + +// The reference retains the old full-scan predicate. Tests compare after every +// publication, including failed persistence, not just the final aggregate. +func scanIndexTaskCounts(m *indexMeta) indexTaskStateCounts { + var counts indexTaskStateCounts + for _, task := range m.segmentBuildInfo.List() { + if task.IsDeleted || !m.IsIndexExist(task.CollectionID, task.IndexID) { + continue + } + // These are the existing wire enum values, independently of the new mapper. + if state := int(task.IndexState); state >= 0 && state < len(counts) { + counts[state]++ + } + } + return counts +} + +func assertIndexTaskCounts(t *testing.T, m *indexMeta) { + t.Helper() + require.Equal(t, scanIndexTaskCounts(m), m.indexTaskCountsSnapshot()) + for _, counts := range m.taskCounts.byIndex { + for _, count := range counts { + require.GreaterOrEqual(t, count, int64(0)) + } + require.NotEqual(t, indexTaskStateCounts{}, counts, "empty buckets must be reclaimed") + } +} + +func TestIndexTaskCountsRecovery(t *testing.T) { + catalog := &taskCountsCatalog{indexes: []*model.Index{ + {CollectionID: 1, IndexID: 10}, + {CollectionID: 2, IndexID: 10, IsDeleted: true}, + }} + for state := 0; state < 6; state++ { + for _, coll := range []int64{1, 2, 3} { + catalog.tasks = append(catalog.tasks, &model.SegmentIndex{ + CollectionID: coll, IndexID: 10, SegmentID: int64(state)*10 + coll, + BuildID: int64(state)*10 + coll, IndexState: commonpb.IndexState(state), + }) + } + } + // Deleted and unknown-state tasks must not contribute. + catalog.tasks = append(catalog.tasks, + &model.SegmentIndex{CollectionID: 1, IndexID: 10, BuildID: 100, IndexState: commonpb.IndexState_Finished, IsDeleted: true}, + &model.SegmentIndex{CollectionID: 1, IndexID: 10, BuildID: 101, IndexState: commonpb.IndexState(100)}, + ) + m, err := newIndexMeta(context.Background(), catalog) + require.NoError(t, err) + assertIndexTaskCounts(t, m) + require.Equal(t, indexTaskStateCounts{1, 1, 1, 1, 1, 1}, m.indexTaskCountsSnapshot()) + m.updateIndexTasksMetrics() + for _, label := range []string{"JobStateNone", "JobStateInit", "JobStateInProgress", "JobStateFinished", "JobStateFailed", "JobStateRetry"} { + require.Equal(t, float64(1), testutil.ToFloat64(metrics.IndexStatsTaskNum.WithLabelValues("JobTypeIndexJob", label))) + } + // Creating the missing field index activates its existing raw counts. + require.NoError(t, m.CreateIndex(context.Background(), &model.Index{CollectionID: 3, IndexID: 10})) + assertIndexTaskCounts(t, m) + require.Equal(t, indexTaskStateCounts{2, 2, 2, 2, 2, 2}, m.indexTaskCountsSnapshot()) + require.NoError(t, m.MarkIndexAsDeleted(context.Background(), 1, nil)) + require.NoError(t, m.MarkIndexAsDeleted(context.Background(), 3, nil)) + m.updateIndexTasksMetrics() + for _, state := range indexTaskMetricStates { + require.Zero(t, testutil.ToFloat64(metrics.IndexStatsTaskNum.WithLabelValues(indexpb.JobType_JobTypeIndexJob.String(), state.String()))) + } + assertIndexTaskCounts(t, m) +} + +func TestIndexTaskCountsLifecycle(t *testing.T) { + ctx := context.Background() + catalog := &taskCountsCatalog{} + m, err := newIndexMeta(ctx, catalog) + require.NoError(t, err) + require.NoError(t, m.CreateIndex(ctx, &model.Index{CollectionID: 1, IndexID: 10})) + task := &model.SegmentIndex{CollectionID: 1, IndexID: 10, SegmentID: 20, BuildID: 30} + require.NoError(t, m.AddSegmentIndex(ctx, task)) + assertIndexTaskCounts(t, m) + // An identical ID retry replaces rather than increments. + require.NoError(t, m.AddSegmentIndex(ctx, model.CloneSegmentIndex(task))) + require.Equal(t, indexTaskStateCounts{0, 1}, m.indexTaskCountsSnapshot()) + require.NoError(t, m.BuildIndex(30)) + require.NoError(t, m.UpdateVersion(30, 100)) + assertIndexTaskCounts(t, m) + for _, state := range []commonpb.IndexState{commonpb.IndexState_Failed, commonpb.IndexState_Retry, commonpb.IndexState_InProgress} { + require.NoError(t, m.UpdateIndexState(30, state, "")) + assertIndexTaskCounts(t, m) + } + finish := &workerpb.IndexTaskInfo{BuildID: 30, State: commonpb.IndexState_Finished} + require.NoError(t, m.FinishTask(finish)) + require.NoError(t, m.FinishTask(finish)) + require.Equal(t, indexTaskStateCounts{0, 0, 0, 1}, m.indexTaskCountsSnapshot()) + require.NoError(t, m.MarkIndexAsDeleted(ctx, 1, []int64{10})) + require.NoError(t, m.MarkIndexAsDeleted(ctx, 1, []int64{10})) + require.NoError(t, m.FinishTask(finish)) + assertIndexTaskCounts(t, m) + require.Equal(t, indexTaskStateCounts{}, m.indexTaskCountsSnapshot(), "late completion cannot reactivate a dropped index") + // Replacing field metadata must re-activate retained counts. + require.NoError(t, m.AlterIndex(ctx, &model.Index{CollectionID: 1, IndexID: 10})) + assertIndexTaskCounts(t, m) + require.NoError(t, m.DeleteTask(30)) + require.NoError(t, m.DeleteTask(30)) + require.NoError(t, m.RemoveSegmentIndex(ctx, 30)) + require.NoError(t, m.RemoveSegmentIndex(ctx, 30)) + assertIndexTaskCounts(t, m) + require.Empty(t, m.taskCounts.byIndex) + + // Copy paths can create already-finished tasks, including two build IDs + // for the same (segment,index): the metric has always counted build IDs. + for _, id := range []int64{40, 41} { + require.NoError(t, m.AddSegmentIndex(ctx, &model.SegmentIndex{CollectionID: 1, IndexID: 10, SegmentID: 20, BuildID: id, IndexState: commonpb.IndexState_Finished})) + } + require.Equal(t, indexTaskStateCounts{0, 0, 0, 2}, m.indexTaskCountsSnapshot()) + stale, _ := m.GetIndexJob(40) + require.NoError(t, m.UpdateVersion(40, 200)) + removed, err := m.RemoveSegmentIndexes(ctx, []*model.SegmentIndex{stale}) + require.NoError(t, err) + require.Zero(t, removed) + assertIndexTaskCounts(t, m) + current, _ := m.GetIndexJob(40) + removed, err = m.RemoveSegmentIndexes(ctx, []*model.SegmentIndex{current, current}) + require.NoError(t, err) + require.Equal(t, 1, removed) + assertIndexTaskCounts(t, m) + require.NoError(t, m.RemoveIndex(ctx, 1, 10)) + assertIndexTaskCounts(t, m) + require.NoError(t, m.FinishTask(&workerpb.IndexTaskInfo{BuildID: 41, State: commonpb.IndexState_Finished})) + require.Equal(t, indexTaskStateCounts{}, m.indexTaskCountsSnapshot()) + require.NoError(t, m.RemoveSegmentIndex(ctx, 41)) + require.Empty(t, m.taskCounts.byIndex) +} + +func TestIndexTaskCountsRecoveryReplacement(t *testing.T) { + catalog := &taskCountsCatalog{ + indexes: []*model.Index{{CollectionID: 1, IndexID: 10}}, + tasks: []*model.SegmentIndex{ + {CollectionID: 1, IndexID: 10, SegmentID: 20, BuildID: 30, IndexState: commonpb.IndexState_Unissued}, + {CollectionID: 1, IndexID: 10, SegmentID: 20, BuildID: 30, IndexState: commonpb.IndexState_Finished}, + {CollectionID: 1, IndexID: 10, SegmentID: 20, BuildID: 30, IndexState: commonpb.IndexState_Finished}, + }, + } + m, err := newIndexMeta(context.Background(), catalog) + require.NoError(t, err) + assertIndexTaskCounts(t, m) + require.Equal(t, indexTaskStateCounts{0, 0, 0, 1}, m.indexTaskCountsSnapshot()) +} + +func TestIndexTaskCountsPersistenceFailures(t *testing.T) { + ctx := context.Background() + operations := map[string]func(*indexMeta) error{ + "create": func(m *indexMeta) error { return m.CreateIndex(ctx, &model.Index{CollectionID: 2, IndexID: 11}) }, + "alter-index": func(m *indexMeta) error { + return m.AlterIndex(ctx, &model.Index{CollectionID: 1, IndexID: 10, IsDeleted: true}) + }, + "add": func(m *indexMeta) error { + return m.AddSegmentIndex(ctx, &model.SegmentIndex{CollectionID: 1, IndexID: 10, SegmentID: 21, BuildID: 31}) + }, + "state": func(m *indexMeta) error { return m.UpdateIndexState(30, commonpb.IndexState_Retry, "") }, + "version": func(m *indexMeta) error { return m.UpdateVersion(30, 100) }, + "build": func(m *indexMeta) error { return m.BuildIndex(30) }, + "finish": func(m *indexMeta) error { + return m.FinishTask(&workerpb.IndexTaskInfo{BuildID: 30, State: commonpb.IndexState_Finished}) + }, + "delete-task": func(m *indexMeta) error { return m.DeleteTask(30) }, + "mark-deleted": func(m *indexMeta) error { return m.MarkIndexAsDeleted(ctx, 1, nil) }, + "remove-task": func(m *indexMeta) error { return m.RemoveSegmentIndex(ctx, 30) }, + "remove-tasks": func(m *indexMeta) error { _, err := m.RemoveSegmentIndexes(ctx, m.segmentBuildInfo.List()); return err }, + "remove-index": func(m *indexMeta) error { return m.RemoveIndex(ctx, 1, 10) }, + "remove-indexes": func(m *indexMeta) error { _, err := m.RemoveIndexes(ctx, m.GetDeletedIndexes()); return err }, + } + for name, operation := range operations { + t.Run(name, func(t *testing.T) { + catalog := &taskCountsCatalog{ + indexes: []*model.Index{{CollectionID: 1, IndexID: 10, IsDeleted: name == "remove-indexes"}}, + tasks: []*model.SegmentIndex{{CollectionID: 1, IndexID: 10, SegmentID: 20, BuildID: 30, IndexState: commonpb.IndexState_Unissued}}, + } + m, err := newIndexMeta(ctx, catalog) + require.NoError(t, err) + before := m.indexTaskCountsSnapshot() + catalog.err = merr.ErrServiceUnavailable + require.ErrorIs(t, operation(m), catalog.err) + require.Equal(t, before, m.indexTaskCountsSnapshot()) + assertIndexTaskCounts(t, m) + catalog.err = nil + require.NoError(t, operation(m)) + assertIndexTaskCounts(t, m) + }) + } +} + +func TestIndexTaskCountsRandomized(t *testing.T) { + ctx := context.Background() + m, err := newIndexMeta(ctx, &taskCountsCatalog{}) + require.NoError(t, err) + rng := rand.New(rand.NewSource(42)) + for i := 0; i < 500; i++ { + buildID := int64(rng.Intn(24)) + collID := buildID % 3 + indexID := buildID % 2 + switch rng.Intn(7) { + case 0: + require.NoError(t, m.CreateIndex(ctx, &model.Index{CollectionID: collID, IndexID: indexID})) + case 1: + require.NoError(t, m.AddSegmentIndex(ctx, &model.SegmentIndex{CollectionID: collID, IndexID: indexID, SegmentID: buildID, BuildID: buildID, IndexState: commonpb.IndexState(rng.Intn(7))})) + case 2: + if _, ok := m.GetIndexJob(buildID); ok { + require.NoError(t, m.UpdateIndexState(buildID, commonpb.IndexState(rng.Intn(7)), "")) + } + case 3: + require.NoError(t, m.DeleteTask(buildID)) + case 4: + require.NoError(t, m.RemoveSegmentIndex(ctx, buildID)) + case 5: + require.NoError(t, m.MarkIndexAsDeleted(ctx, collID, []int64{indexID})) + case 6: + require.NoError(t, m.RemoveIndex(ctx, collID, indexID)) + } + assertIndexTaskCounts(t, m) + } +} + +func TestIndexTaskCountsConcurrent(t *testing.T) { + ctx := context.Background() + m, err := newIndexMeta(ctx, &taskCountsCatalog{}) + require.NoError(t, err) + const builds = 16 + require.NoError(t, m.CreateIndex(ctx, &model.Index{CollectionID: 1, IndexID: 10})) + for id := int64(0); id < builds; id++ { + require.NoError(t, m.AddSegmentIndex(ctx, &model.SegmentIndex{CollectionID: 1, IndexID: 10, SegmentID: id, BuildID: id})) + } + var wg sync.WaitGroup + for id := int64(0); id < builds; id++ { + wg.Go(func() { + for i := 0; i < 20; i++ { + if err := m.UpdateIndexState(id, commonpb.IndexState(i%6), ""); err != nil { + t.Error(err) + } + snapshot := m.indexTaskCountsSnapshot() + var total int64 + for _, n := range snapshot { + if n < 0 { + t.Error("negative task count") + } + total += n + } + if total != 0 && total != builds { + t.Errorf("torn index publication: %d", total) + } + } + }) + } + wg.Go(func() { + for i := 0; i < 20; i++ { + if err := m.MarkIndexAsDeleted(ctx, 1, nil); err != nil { + t.Error(err) + } + if err := m.AlterIndex(ctx, &model.Index{CollectionID: 1, IndexID: 10}); err != nil { + t.Error(err) + } + } + }) + wg.Wait() + assertIndexTaskCounts(t, m) + require.Zero(t, testing.AllocsPerRun(100, func() { m.indexTaskCountsSnapshot() })) +} + +func BenchmarkIndexTaskCounts(b *testing.B) { + for _, size := range []int{1000, 100000} { + b.Run(fmt.Sprint(size), func(b *testing.B) { + m := &indexMeta{ + ctx: context.Background(), + indexes: map[int64]map[int64]*model.Index{1: {10: {CollectionID: 1, IndexID: 10}}}, + segmentBuildInfo: newSegmentIndexBuildInfo(), + segmentIndexes: typeutil.NewConcurrentMap[int64, *typeutil.ConcurrentMap[int64, *model.SegmentIndex]](), + } + for i := 0; i < size; i++ { + m.updateSegmentIndex(&model.SegmentIndex{CollectionID: 1, IndexID: 10, SegmentID: int64(i), BuildID: int64(i), IndexState: commonpb.IndexState_Finished}) + } + b.Run("full-scan", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + _ = scanIndexTaskCounts(m) + } + }) + b.Run("snapshot", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + _ = m.indexTaskCountsSnapshot() + } + }) + b.Run("publish", func(b *testing.B) { + m.updateIndexTasksMetrics() + b.ReportAllocs() + for b.Loop() { + m.updateIndexTasksMetrics() + } + }) + }) + } +} + +func BenchmarkIndexTaskCountsStorage(b *testing.B) { + const indexes = 100000 + b.ReportAllocs() + b.ReportMetric(indexes, "indexes/op") + for b.Loop() { + var counts indexTaskCounts + for id := int64(0); id < indexes; id++ { + counts.adjustTask(&model.SegmentIndex{CollectionID: 1, IndexID: id, IndexState: commonpb.IndexState_Finished}, 1, nil) + } + if len(counts.byIndex) != indexes { + b.Fatal("unexpected index count") + } + } +} diff --git a/internal/rootcoord/meta_table.go b/internal/rootcoord/meta_table.go index d3a9cf56d54..af3c57049c1 100644 --- a/internal/rootcoord/meta_table.go +++ b/internal/rootcoord/meta_table.go @@ -1150,6 +1150,15 @@ func (mt *MetaTable) getCollectionByNameInternal(ctx context.Context, dbName str return filterUnavailablePartition(coll), nil } +// IsCollectionAvailable is a positive-only check of current resident metadata. +// A miss does not prove absence: callers must use the normal lookup before GC. +func (mt *MetaTable) IsCollectionAvailable(collectionID int64) bool { + mt.ddLock.RLock() + defer mt.ddLock.RUnlock() + collection := mt.collID2Meta[collectionID] + return collection != nil && collection.Available() +} + func (mt *MetaTable) GetCollectionByID(ctx context.Context, dbName string, collectionID UniqueID, ts Timestamp, allowUnavailable bool) (*model.Collection, error) { mt.ddLock.RLock() defer mt.ddLock.RUnlock() diff --git a/internal/rootcoord/meta_table_test.go b/internal/rootcoord/meta_table_test.go index 303f8a816c7..9fa505e1557 100644 --- a/internal/rootcoord/meta_table_test.go +++ b/internal/rootcoord/meta_table_test.go @@ -19,6 +19,7 @@ package rootcoord import ( "context" "math/rand" + "sync" "testing" "time" @@ -52,6 +53,67 @@ import ( "github.com/milvus-io/milvus/pkg/v3/util/typeutil" ) +func TestCollectionAvailability(t *testing.T) { + mt := &MetaTable{collID2Meta: map[int64]*model.Collection{ + 1: {CollectionID: 1, State: pb.CollectionState_CollectionCreated}, + 2: {CollectionID: 2, State: pb.CollectionState_CollectionCreating}, + 3: {CollectionID: 3, State: pb.CollectionState_CollectionDropping}, + 4: nil, + }} + core := &Core{ctx: context.Background(), meta: mt} + core.UpdateStateCode(commonpb.StateCode_Healthy) + require.True(t, core.IsCollectionAvailable(1)) + for _, id := range []int64{2, 3, 4, 5} { + require.False(t, core.IsCollectionAvailable(id)) + } + // No scheduler, catalog or schema is needed on either cache outcome. + core.UpdateStateCode(commonpb.StateCode_Abnormal) + require.False(t, core.IsCollectionAvailable(1)) + core.UpdateStateCode(commonpb.StateCode_Healthy) + mt.ddLock.Lock() + mt.collID2Meta[1] = &model.Collection{CollectionID: 1, State: pb.CollectionState_CollectionDropping} + mt.ddLock.Unlock() + require.False(t, core.IsCollectionAvailable(1)) + core.meta = nil + require.False(t, core.IsCollectionAvailable(1)) + + require.Zero(t, testing.AllocsPerRun(100, func() { mt.IsCollectionAvailable(1) })) +} + +func BenchmarkCollectionAvailability(b *testing.B) { + mt := &MetaTable{collID2Meta: map[int64]*model.Collection{ + 1: {CollectionID: 1, State: pb.CollectionState_CollectionCreated}, + }} + b.ReportAllocs() + for b.Loop() { + if !mt.IsCollectionAvailable(1) { + b.Fatal("available collection not found") + } + } +} + +func TestCollectionAvailabilityConcurrent(t *testing.T) { + mt := &MetaTable{collID2Meta: make(map[int64]*model.Collection)} + var wg sync.WaitGroup + wg.Go(func() { + for i := 0; i < 100; i++ { + mt.ddLock.Lock() + mt.collID2Meta[1] = &model.Collection{CollectionID: 1, State: pb.CollectionState_CollectionCreated} + mt.ddLock.Unlock() + mt.ddLock.Lock() + delete(mt.collID2Meta, 1) + mt.ddLock.Unlock() + } + }) + wg.Go(func() { + for i := 0; i < 100; i++ { + mt.IsCollectionAvailable(1) + } + }) + wg.Wait() + require.False(t, mt.IsCollectionAvailable(1)) +} + func TestMetaTable_DescribeAliasAllowsConcurrentReaders(t *testing.T) { const ( collectionID = int64(100) diff --git a/internal/rootcoord/root_coord.go b/internal/rootcoord/root_coord.go index b5c031d368f..caebd3f094f 100644 --- a/internal/rootcoord/root_coord.go +++ b/internal/rootcoord/root_coord.go @@ -1155,6 +1155,16 @@ func (c *Core) TruncateCollection(ctx context.Context, in *milvuspb.TruncateColl }, nil } +// IsCollectionAvailable lets in-process GC skip live collections without +// materializing their schemas. False means unknown, not safe to delete. +func (c *Core) IsCollectionAvailable(collectionID int64) bool { + if c.GetStateCode() != commonpb.StateCode_Healthy { + return false + } + checker, ok := c.meta.(interface{ IsCollectionAvailable(int64) bool }) + return ok && checker.IsCollectionAvailable(collectionID) +} + // HasCollection check collection existence func (c *Core) HasCollection(ctx context.Context, in *milvuspb.HasCollectionRequest) (*milvuspb.BoolResponse, error) { if err := merr.CheckHealthy(c.GetStateCode()); err != nil { diff --git a/pkg/util/funcutil/func.go b/pkg/util/funcutil/func.go index 0bc60328da2..d3772e75d36 100644 --- a/pkg/util/funcutil/func.go +++ b/pkg/util/funcutil/func.go @@ -449,9 +449,10 @@ func ConvertChannelName(chanName string, tokenFrom string, tokenTo string) (stri return strings.Replace(chanName, tokenFrom, tokenTo, 1), nil } +var collectionIDFromVChannelPattern = regexp.MustCompile(`.*_(\d+)v\d+`) + func GetCollectionIDFromVChannel(vChannelName string) int64 { - re := regexp.MustCompile(`.*_(\d+)v\d+`) - matches := re.FindStringSubmatch(vChannelName) + matches := collectionIDFromVChannelPattern.FindStringSubmatch(vChannelName) if len(matches) > 1 { number, err := strconv.ParseInt(matches[1], 0, 64) if err == nil { diff --git a/pkg/util/funcutil/func_test.go b/pkg/util/funcutil/func_test.go index 262a898e467..6955aa8043b 100644 --- a/pkg/util/funcutil/func_test.go +++ b/pkg/util/funcutil/func_test.go @@ -23,6 +23,7 @@ import ( "fmt" "net" "reflect" + "regexp" "strconv" "testing" "time" @@ -250,6 +251,28 @@ func TestGetCollectionIDFromVChannel(t *testing.T) { assert.Equal(t, int64(-1), collectionID) } +func BenchmarkGetCollectionIDFromVChannel(b *testing.B) { + const channel = "cluster-rootcoord-dm_3_449684528748778322v0" + b.Run("compile-per-call", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + matches := regexp.MustCompile(`.*_(\d+)v\d+`).FindStringSubmatch(channel) + id, err := strconv.ParseInt(matches[1], 0, 64) + if err != nil || id != 449684528748778322 { + b.Fatal("unexpected collection ID") + } + } + }) + b.Run("shared-regexp", func(b *testing.B) { + b.ReportAllocs() + for b.Loop() { + if GetCollectionIDFromVChannel(channel) != 449684528748778322 { + b.Fatal("unexpected collection ID") + } + } + }) +} + func TestParseVChannel(t *testing.T) { pchannel, collectionID, index, err := ParseVChannel("by-dev-rootcoord-dml_0_12345v2") require.NoError(t, err)