enhance: remove SuffixSnapshot mechanism from RootCoord metadata storage (#47960)

## Summary

Remove the entire SuffixSnapshot time-travel layer from RootCoord
metadata storage, replacing it with plain KV operations.

Replace SnapShotKV with TxnKV in Catalog (~30 call sites)
Delete suffix_snapshot.go (~750 lines) and tests (~917 lines)
Remove SnapShotKV interface from pkg/kv/kv.go
Remove snapshot GC config params (ttl, reserveTime)
Keep IsTombstone/ComposeSnapshotKey constants for migration tool
compatibility
Why: Investigation confirmed snapshot keys are write-only artifacts that
nobody reads:

No client/proxy sends non-zero timestamp for time-travel metadata
queries
Tombstone sweeper calls RemoveCollection(ts=0), bypassing snapshot
tombstone writes entirely
Snapshot keys accumulate as orphans — the old GC couldn't fully clean
them because ts=0 deletion path never wrote tombstone markers
Net reduction: -2180 lines

related: #48697

## Test plan
- [x] Unit tests for GC batch logic (expiry, cursor progress,
latest-per-group protection, tombstone cleanup)
- [x] Unit tests for Save/Load with time-travel verification
- [x] `go vet` passes on all changed packages
- [ ] CI passes

Signed-off-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: MrPresent-Han <chun.han@gmail.com>
Co-authored-by: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Makefile

@@ -566,7 +566,6 @@ generate-mockery-kv: getdeps
 	$(INSTALL_PATH)/mockery --name=TxnKV --dir=$(PWD)/pkg/kv --output=$(PWD)/internal/kv/mocks --filename=txn_kv.go --with-expecter
 	$(INSTALL_PATH)/mockery --name=MetaKv --dir=$(PWD)/pkg/kv --output=$(PWD)/internal/kv/mocks --filename=meta_kv.go --with-expecter
 	$(INSTALL_PATH)/mockery --name=WatchKV --dir=$(PWD)/pkg/kv --output=$(PWD)/internal/kv/mocks --filename=watch_kv.go --with-expecter
-	$(INSTALL_PATH)/mockery --name=SnapShotKV --dir=$(PWD)/pkg/kv --output=$(PWD)/internal/kv/mocks --filename=snapshot_kv.go --with-expecter
 	$(INSTALL_PATH)/mockery --name=Predicate --dir=$(PWD)/pkg/kv/predicates --output=$(PWD)/internal/kv/predicates --filename=mock_predicate.go --with-expecter --inpackage
 
 generate-mockery-chunk-manager: getdeps

cmd/tools/migration/mmap/tool/main.go

@@ -126,31 +126,21 @@ func prepareRootCoordMeta(ctx context.Context, allocator tso.Allocator) (rootcoo
 	switch paramtable.Get().MetaStoreCfg.MetaStoreType.GetValue() {
 	case util.MetaStoreTypeEtcd:
 		var metaKV kv.MetaKv
-		var ss *kvmetestore.SuffixSnapshot
 		var err error
 
 		if metaKV, err = metaKVCreator(); err != nil {
 			panic(err)
 		}
-
-		if ss, err = kvmetestore.NewSuffixSnapshot(metaKV, kvmetestore.SnapshotsSep, paramtable.Get().EtcdCfg.MetaRootPath.GetValue(), kvmetestore.SnapshotPrefix); err != nil {
-			panic(err)
-		}
-		catalog = kvmetestore.NewCatalog(metaKV, ss)
+		catalog = kvmetestore.NewCatalog(metaKV)
 	case util.MetaStoreTypeTiKV:
 		log.Ctx(ctx).Info("Using tikv as meta storage.")
 		var metaKV kv.MetaKv
-		var ss *kvmetestore.SuffixSnapshot
 		var err error
 
 		if metaKV, err = metaKVCreator(); err != nil {
 			panic(err)
 		}
-
-		if ss, err = kvmetestore.NewSuffixSnapshot(metaKV, kvmetestore.SnapshotsSep, paramtable.Get().TiKVCfg.MetaRootPath.GetValue(), kvmetestore.SnapshotPrefix); err != nil {
-			panic(err)
-		}
-		catalog = kvmetestore.NewCatalog(metaKV, ss)
+		catalog = kvmetestore.NewCatalog(metaKV)
 	default:
 		panic(fmt.Sprintf("MetaStoreType %s not supported", paramtable.Get().MetaStoreCfg.MetaStoreType.GetValue()))
 	}

configs/milvus.yaml

@@ -66,9 +66,6 @@ etcd:
 
 metastore:
   type: etcd # Default value: etcd, Valid values: [etcd, tikv]
-  snapshot:
-    ttl: 86400 # snapshot ttl in seconds
-    reserveTime: 3600 # snapshot reserve time in seconds
   maxEtcdTxnNum: 64 # maximum number of operations in a single etcd transaction
 
 # Related configuration of tikv, used to store Milvus metadata.

internal/kv/mock_snapshot_kv.go

@@ -1,66 +0,0 @@
-package kv
-
-import (
-	"context"
-
-	"github.com/milvus-io/milvus/pkg/v2/kv"
-	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
-)
-
-var _ kv.SnapShotKV = &mockSnapshotKV{}
-
-type mockSnapshotKV struct {
-	kv.SnapShotKV
-	SaveFunc                         func(ctx context.Context, key string, value string, ts typeutil.Timestamp) error
-	LoadFunc                         func(ctx context.Context, key string, ts typeutil.Timestamp) (string, error)
-	MultiSaveFunc                    func(ctx context.Context, kvs map[string]string, ts typeutil.Timestamp) error
-	LoadWithPrefixFunc               func(ctx context.Context, key string, ts typeutil.Timestamp) ([]string, []string, error)
-	MultiSaveAndRemoveWithPrefixFunc func(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error
-	MultiSaveAndRemoveFunc           func(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error
-}
-
-func NewMockSnapshotKV() *mockSnapshotKV {
-	return &mockSnapshotKV{}
-}
-
-func (m mockSnapshotKV) Save(ctx context.Context, key string, value string, ts typeutil.Timestamp) error {
-	if m.SaveFunc != nil {
-		return m.SaveFunc(ctx, key, value, ts)
-	}
-	return nil
-}
-
-func (m mockSnapshotKV) Load(ctx context.Context, key string, ts typeutil.Timestamp) (string, error) {
-	if m.LoadFunc != nil {
-		return m.LoadFunc(ctx, key, ts)
-	}
-	return "", nil
-}
-
-func (m mockSnapshotKV) MultiSave(ctx context.Context, kvs map[string]string, ts typeutil.Timestamp) error {
-	if m.MultiSaveFunc != nil {
-		return m.MultiSaveFunc(ctx, kvs, ts)
-	}
-	return nil
-}
-
-func (m mockSnapshotKV) LoadWithPrefix(ctx context.Context, key string, ts typeutil.Timestamp) ([]string, []string, error) {
-	if m.LoadWithPrefixFunc != nil {
-		return m.LoadWithPrefixFunc(ctx, key, ts)
-	}
-	return nil, nil, nil
-}
-
-func (m mockSnapshotKV) MultiSaveAndRemoveWithPrefix(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error {
-	if m.MultiSaveAndRemoveWithPrefixFunc != nil {
-		return m.MultiSaveAndRemoveWithPrefixFunc(ctx, saves, removals, ts)
-	}
-	return nil
-}
-
-func (m mockSnapshotKV) MultiSaveAndRemove(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error {
-	if m.MultiSaveAndRemoveFunc != nil {
-		return m.MultiSaveAndRemoveFunc(ctx, saves, removals, ts)
-	}
-	return nil
-}

internal/kv/mock_snapshot_kv_test.go

@@ -1,106 +0,0 @@
-package kv
-
-import (
-	"context"
-	"testing"
-
-	"github.com/stretchr/testify/assert"
-
-	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
-)
-
-func Test_mockSnapshotKV_Save(t *testing.T) {
-	t.Run("func not set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		err := snapshot.Save(context.TODO(), "k", "v", 0)
-		assert.NoError(t, err)
-	})
-	t.Run("func set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		snapshot.SaveFunc = func(ctx context.Context, key string, value string, ts typeutil.Timestamp) error {
-			return nil
-		}
-		err := snapshot.Save(context.TODO(), "k", "v", 0)
-		assert.NoError(t, err)
-	})
-}
-
-func Test_mockSnapshotKV_Load(t *testing.T) {
-	t.Run("func not set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		_, err := snapshot.Load(context.TODO(), "k", 0)
-		assert.NoError(t, err)
-	})
-	t.Run("func set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		snapshot.LoadFunc = func(ctx context.Context, key string, ts typeutil.Timestamp) (string, error) {
-			return "", nil
-		}
-		_, err := snapshot.Load(context.TODO(), "k", 0)
-		assert.NoError(t, err)
-	})
-}
-
-func Test_mockSnapshotKV_MultiSave(t *testing.T) {
-	t.Run("func not set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		err := snapshot.MultiSave(context.TODO(), nil, 0)
-		assert.NoError(t, err)
-	})
-	t.Run("func set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		snapshot.MultiSaveFunc = func(ctx context.Context, kvs map[string]string, ts typeutil.Timestamp) error {
-			return nil
-		}
-		err := snapshot.MultiSave(context.TODO(), nil, 0)
-		assert.NoError(t, err)
-	})
-}
-
-func Test_mockSnapshotKV_LoadWithPrefix(t *testing.T) {
-	t.Run("func not set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		_, _, err := snapshot.LoadWithPrefix(context.TODO(), "prefix", 0)
-		assert.NoError(t, err)
-	})
-	t.Run("func set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		snapshot.LoadWithPrefixFunc = func(ctx context.Context, key string, ts typeutil.Timestamp) ([]string, []string, error) {
-			return nil, nil, nil
-		}
-		_, _, err := snapshot.LoadWithPrefix(context.TODO(), "prefix", 0)
-		assert.NoError(t, err)
-	})
-}
-
-func Test_mockSnapshotKV_MultiSaveAndRemoveWithPrefix(t *testing.T) {
-	t.Run("func not set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		err := snapshot.MultiSaveAndRemoveWithPrefix(context.TODO(), nil, nil, 0)
-		assert.NoError(t, err)
-	})
-	t.Run("func set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		snapshot.MultiSaveAndRemoveWithPrefixFunc = func(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error {
-			return nil
-		}
-		err := snapshot.MultiSaveAndRemoveWithPrefix(context.TODO(), nil, nil, 0)
-		assert.NoError(t, err)
-	})
-}
-
-func Test_mockSnapshotKV_MultiSaveAndRemove(t *testing.T) {
-	t.Run("func not set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		err := snapshot.MultiSaveAndRemove(context.TODO(), nil, nil, 0)
-		assert.NoError(t, err)
-	})
-	t.Run("func set", func(t *testing.T) {
-		snapshot := NewMockSnapshotKV()
-		snapshot.MultiSaveAndRemoveWithPrefixFunc = func(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error {
-			return nil
-		}
-		err := snapshot.MultiSaveAndRemove(context.TODO(), nil, nil, 0)
-		assert.NoError(t, err)
-	})
-}

internal/kv/mocks/snapshot_kv.go

@@ -1,358 +0,0 @@
-// Code generated by mockery v2.53.3. DO NOT EDIT.
-
-package mocks
-
-import (
-	context "context"
-
-	mock "github.com/stretchr/testify/mock"
-)
-
-// SnapShotKV is an autogenerated mock type for the SnapShotKV type
-type SnapShotKV struct {
-	mock.Mock
-}
-
-type SnapShotKV_Expecter struct {
-	mock *mock.Mock
-}
-
-func (_m *SnapShotKV) EXPECT() *SnapShotKV_Expecter {
-	return &SnapShotKV_Expecter{mock: &_m.Mock}
-}
-
-// Load provides a mock function with given fields: ctx, key, ts
-func (_m *SnapShotKV) Load(ctx context.Context, key string, ts uint64) (string, error) {
-	ret := _m.Called(ctx, key, ts)
-
-	if len(ret) == 0 {
-		panic("no return value specified for Load")
-	}
-
-	var r0 string
-	var r1 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, uint64) (string, error)); ok {
-		return rf(ctx, key, ts)
-	}
-	if rf, ok := ret.Get(0).(func(context.Context, string, uint64) string); ok {
-		r0 = rf(ctx, key, ts)
-	} else {
-		r0 = ret.Get(0).(string)
-	}
-
-	if rf, ok := ret.Get(1).(func(context.Context, string, uint64) error); ok {
-		r1 = rf(ctx, key, ts)
-	} else {
-		r1 = ret.Error(1)
-	}
-
-	return r0, r1
-}
-
-// SnapShotKV_Load_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Load'
-type SnapShotKV_Load_Call struct {
-	*mock.Call
-}
-
-// Load is a helper method to define mock.On call
-//   - ctx context.Context
-//   - key string
-//   - ts uint64
-func (_e *SnapShotKV_Expecter) Load(ctx interface{}, key interface{}, ts interface{}) *SnapShotKV_Load_Call {
-	return &SnapShotKV_Load_Call{Call: _e.mock.On("Load", ctx, key, ts)}
-}
-
-func (_c *SnapShotKV_Load_Call) Run(run func(ctx context.Context, key string, ts uint64)) *SnapShotKV_Load_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(string), args[2].(uint64))
-	})
-	return _c
-}
-
-func (_c *SnapShotKV_Load_Call) Return(_a0 string, _a1 error) *SnapShotKV_Load_Call {
-	_c.Call.Return(_a0, _a1)
-	return _c
-}
-
-func (_c *SnapShotKV_Load_Call) RunAndReturn(run func(context.Context, string, uint64) (string, error)) *SnapShotKV_Load_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
-// LoadWithPrefix provides a mock function with given fields: ctx, key, ts
-func (_m *SnapShotKV) LoadWithPrefix(ctx context.Context, key string, ts uint64) ([]string, []string, error) {
-	ret := _m.Called(ctx, key, ts)
-
-	if len(ret) == 0 {
-		panic("no return value specified for LoadWithPrefix")
-	}
-
-	var r0 []string
-	var r1 []string
-	var r2 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, uint64) ([]string, []string, error)); ok {
-		return rf(ctx, key, ts)
-	}
-	if rf, ok := ret.Get(0).(func(context.Context, string, uint64) []string); ok {
-		r0 = rf(ctx, key, ts)
-	} else {
-		if ret.Get(0) != nil {
-			r0 = ret.Get(0).([]string)
-		}
-	}
-
-	if rf, ok := ret.Get(1).(func(context.Context, string, uint64) []string); ok {
-		r1 = rf(ctx, key, ts)
-	} else {
-		if ret.Get(1) != nil {
-			r1 = ret.Get(1).([]string)
-		}
-	}
-
-	if rf, ok := ret.Get(2).(func(context.Context, string, uint64) error); ok {
-		r2 = rf(ctx, key, ts)
-	} else {
-		r2 = ret.Error(2)
-	}
-
-	return r0, r1, r2
-}
-
-// SnapShotKV_LoadWithPrefix_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'LoadWithPrefix'
-type SnapShotKV_LoadWithPrefix_Call struct {
-	*mock.Call
-}
-
-// LoadWithPrefix is a helper method to define mock.On call
-//   - ctx context.Context
-//   - key string
-//   - ts uint64
-func (_e *SnapShotKV_Expecter) LoadWithPrefix(ctx interface{}, key interface{}, ts interface{}) *SnapShotKV_LoadWithPrefix_Call {
-	return &SnapShotKV_LoadWithPrefix_Call{Call: _e.mock.On("LoadWithPrefix", ctx, key, ts)}
-}
-
-func (_c *SnapShotKV_LoadWithPrefix_Call) Run(run func(ctx context.Context, key string, ts uint64)) *SnapShotKV_LoadWithPrefix_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(string), args[2].(uint64))
-	})
-	return _c
-}
-
-func (_c *SnapShotKV_LoadWithPrefix_Call) Return(_a0 []string, _a1 []string, _a2 error) *SnapShotKV_LoadWithPrefix_Call {
-	_c.Call.Return(_a0, _a1, _a2)
-	return _c
-}
-
-func (_c *SnapShotKV_LoadWithPrefix_Call) RunAndReturn(run func(context.Context, string, uint64) ([]string, []string, error)) *SnapShotKV_LoadWithPrefix_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
-// MultiSave provides a mock function with given fields: ctx, kvs, ts
-func (_m *SnapShotKV) MultiSave(ctx context.Context, kvs map[string]string, ts uint64) error {
-	ret := _m.Called(ctx, kvs, ts)
-
-	if len(ret) == 0 {
-		panic("no return value specified for MultiSave")
-	}
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, map[string]string, uint64) error); ok {
-		r0 = rf(ctx, kvs, ts)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// SnapShotKV_MultiSave_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'MultiSave'
-type SnapShotKV_MultiSave_Call struct {
-	*mock.Call
-}
-
-// MultiSave is a helper method to define mock.On call
-//   - ctx context.Context
-//   - kvs map[string]string
-//   - ts uint64
-func (_e *SnapShotKV_Expecter) MultiSave(ctx interface{}, kvs interface{}, ts interface{}) *SnapShotKV_MultiSave_Call {
-	return &SnapShotKV_MultiSave_Call{Call: _e.mock.On("MultiSave", ctx, kvs, ts)}
-}
-
-func (_c *SnapShotKV_MultiSave_Call) Run(run func(ctx context.Context, kvs map[string]string, ts uint64)) *SnapShotKV_MultiSave_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(map[string]string), args[2].(uint64))
-	})
-	return _c
-}
-
-func (_c *SnapShotKV_MultiSave_Call) Return(_a0 error) *SnapShotKV_MultiSave_Call {
-	_c.Call.Return(_a0)
-	return _c
-}
-
-func (_c *SnapShotKV_MultiSave_Call) RunAndReturn(run func(context.Context, map[string]string, uint64) error) *SnapShotKV_MultiSave_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
-// MultiSaveAndRemove provides a mock function with given fields: ctx, saves, removals, ts
-func (_m *SnapShotKV) MultiSaveAndRemove(ctx context.Context, saves map[string]string, removals []string, ts uint64) error {
-	ret := _m.Called(ctx, saves, removals, ts)
-
-	if len(ret) == 0 {
-		panic("no return value specified for MultiSaveAndRemove")
-	}
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, map[string]string, []string, uint64) error); ok {
-		r0 = rf(ctx, saves, removals, ts)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// SnapShotKV_MultiSaveAndRemove_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'MultiSaveAndRemove'
-type SnapShotKV_MultiSaveAndRemove_Call struct {
-	*mock.Call
-}
-
-// MultiSaveAndRemove is a helper method to define mock.On call
-//   - ctx context.Context
-//   - saves map[string]string
-//   - removals []string
-//   - ts uint64
-func (_e *SnapShotKV_Expecter) MultiSaveAndRemove(ctx interface{}, saves interface{}, removals interface{}, ts interface{}) *SnapShotKV_MultiSaveAndRemove_Call {
-	return &SnapShotKV_MultiSaveAndRemove_Call{Call: _e.mock.On("MultiSaveAndRemove", ctx, saves, removals, ts)}
-}
-
-func (_c *SnapShotKV_MultiSaveAndRemove_Call) Run(run func(ctx context.Context, saves map[string]string, removals []string, ts uint64)) *SnapShotKV_MultiSaveAndRemove_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(map[string]string), args[2].([]string), args[3].(uint64))
-	})
-	return _c
-}
-
-func (_c *SnapShotKV_MultiSaveAndRemove_Call) Return(_a0 error) *SnapShotKV_MultiSaveAndRemove_Call {
-	_c.Call.Return(_a0)
-	return _c
-}
-
-func (_c *SnapShotKV_MultiSaveAndRemove_Call) RunAndReturn(run func(context.Context, map[string]string, []string, uint64) error) *SnapShotKV_MultiSaveAndRemove_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
-// MultiSaveAndRemoveWithPrefix provides a mock function with given fields: ctx, saves, removals, ts
-func (_m *SnapShotKV) MultiSaveAndRemoveWithPrefix(ctx context.Context, saves map[string]string, removals []string, ts uint64) error {
-	ret := _m.Called(ctx, saves, removals, ts)
-
-	if len(ret) == 0 {
-		panic("no return value specified for MultiSaveAndRemoveWithPrefix")
-	}
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, map[string]string, []string, uint64) error); ok {
-		r0 = rf(ctx, saves, removals, ts)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// SnapShotKV_MultiSaveAndRemoveWithPrefix_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'MultiSaveAndRemoveWithPrefix'
-type SnapShotKV_MultiSaveAndRemoveWithPrefix_Call struct {
-	*mock.Call
-}
-
-// MultiSaveAndRemoveWithPrefix is a helper method to define mock.On call
-//   - ctx context.Context
-//   - saves map[string]string
-//   - removals []string
-//   - ts uint64
-func (_e *SnapShotKV_Expecter) MultiSaveAndRemoveWithPrefix(ctx interface{}, saves interface{}, removals interface{}, ts interface{}) *SnapShotKV_MultiSaveAndRemoveWithPrefix_Call {
-	return &SnapShotKV_MultiSaveAndRemoveWithPrefix_Call{Call: _e.mock.On("MultiSaveAndRemoveWithPrefix", ctx, saves, removals, ts)}
-}
-
-func (_c *SnapShotKV_MultiSaveAndRemoveWithPrefix_Call) Run(run func(ctx context.Context, saves map[string]string, removals []string, ts uint64)) *SnapShotKV_MultiSaveAndRemoveWithPrefix_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(map[string]string), args[2].([]string), args[3].(uint64))
-	})
-	return _c
-}
-
-func (_c *SnapShotKV_MultiSaveAndRemoveWithPrefix_Call) Return(_a0 error) *SnapShotKV_MultiSaveAndRemoveWithPrefix_Call {
-	_c.Call.Return(_a0)
-	return _c
-}
-
-func (_c *SnapShotKV_MultiSaveAndRemoveWithPrefix_Call) RunAndReturn(run func(context.Context, map[string]string, []string, uint64) error) *SnapShotKV_MultiSaveAndRemoveWithPrefix_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
-// Save provides a mock function with given fields: ctx, key, value, ts
-func (_m *SnapShotKV) Save(ctx context.Context, key string, value string, ts uint64) error {
-	ret := _m.Called(ctx, key, value, ts)
-
-	if len(ret) == 0 {
-		panic("no return value specified for Save")
-	}
-
-	var r0 error
-	if rf, ok := ret.Get(0).(func(context.Context, string, string, uint64) error); ok {
-		r0 = rf(ctx, key, value, ts)
-	} else {
-		r0 = ret.Error(0)
-	}
-
-	return r0
-}
-
-// SnapShotKV_Save_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'Save'
-type SnapShotKV_Save_Call struct {
-	*mock.Call
-}
-
-// Save is a helper method to define mock.On call
-//   - ctx context.Context
-//   - key string
-//   - value string
-//   - ts uint64
-func (_e *SnapShotKV_Expecter) Save(ctx interface{}, key interface{}, value interface{}, ts interface{}) *SnapShotKV_Save_Call {
-	return &SnapShotKV_Save_Call{Call: _e.mock.On("Save", ctx, key, value, ts)}
-}
-
-func (_c *SnapShotKV_Save_Call) Run(run func(ctx context.Context, key string, value string, ts uint64)) *SnapShotKV_Save_Call {
-	_c.Call.Run(func(args mock.Arguments) {
-		run(args[0].(context.Context), args[1].(string), args[2].(string), args[3].(uint64))
-	})
-	return _c
-}
-
-func (_c *SnapShotKV_Save_Call) Return(_a0 error) *SnapShotKV_Save_Call {
-	_c.Call.Return(_a0)
-	return _c
-}
-
-func (_c *SnapShotKV_Save_Call) RunAndReturn(run func(context.Context, string, string, uint64) error) *SnapShotKV_Save_Call {
-	_c.Call.Return(run)
-	return _c
-}
-
-// NewSnapShotKV creates a new instance of SnapShotKV. It also registers a testing interface on the mock and a cleanup function to assert the mocks expectations.
-// The first argument is typically a *testing.T value.
-func NewSnapShotKV(t interface {
-	mock.TestingT
-	Cleanup(func())
-}) *SnapShotKV {
-	mock := &SnapShotKV{}
-	mock.Mock.Test(t)
-
-	t.Cleanup(func() { mock.AssertExpectations(t) })
-
-	return mock
-}

internal/metastore/kv/rootcoord/kv_catalog.go

@@ -40,15 +40,14 @@ import (
 // prefix/file_resource/resource_id             -> Resource
 
 type Catalog struct {
-	Txn      kv.TxnKV
-	Snapshot kv.SnapShotKV
+	Txn kv.TxnKV
 
 	pool *conc.Pool[any]
 }
 
-func NewCatalog(metaKV kv.TxnKV, ss kv.SnapShotKV) metastore.RootCoordCatalog {
+func NewCatalog(metaKV kv.TxnKV) metastore.RootCoordCatalog {
 	ioPool := conc.NewPool[any](paramtable.Get().MetaStoreCfg.ReadConcurrency.GetAsInt())
-	return &Catalog{Txn: metaKV, Snapshot: ss, pool: ioPool}
+	return &Catalog{Txn: metaKV, pool: ioPool}
 }
 
 func BuildCollectionKey(dbID typeutil.UniqueID, collectionID typeutil.UniqueID) string {
@@ -113,17 +112,16 @@ func BuildAliasPrefixWithDB(dbID int64) string {
 	return fmt.Sprintf("%s/%s/%d/", DatabaseMetaPrefix, Aliases, dbID)
 }
 
-// since SnapshotKV may save both snapshot key and the original key if the original key is newest
-func batchMultiSaveAndRemove(ctx context.Context, snapshot kv.SnapShotKV, limit int, saves map[string]string, removals []string, ts typeutil.Timestamp) error {
+func batchMultiSaveAndRemove(ctx context.Context, txn kv.TxnKV, limit int, saves map[string]string, removals []string) error {
 	saveFn := func(partialKvs map[string]string) error {
-		return snapshot.MultiSave(ctx, partialKvs, ts)
+		return txn.MultiSave(ctx, partialKvs)
 	}
 	if err := etcd.SaveByBatchWithLimit(saves, limit, saveFn); err != nil {
 		return err
 	}
 
 	removeFn := func(partialKeys []string) error {
-		return snapshot.MultiSaveAndRemove(ctx, nil, partialKeys, ts)
+		return txn.MultiSaveAndRemove(ctx, nil, partialKeys)
 	}
 	return etcd.RemoveByBatchWithLimit(removals, limit, removeFn)
 }
@@ -135,7 +133,7 @@ func (kc *Catalog) CreateDatabase(ctx context.Context, db *model.Database, ts ty
 	if err != nil {
 		return err
 	}
-	return kc.Snapshot.Save(ctx, key, string(v), ts)
+	return kc.Txn.Save(ctx, key, string(v))
 }
 
 func (kc *Catalog) AlterDatabase(ctx context.Context, newColl *model.Database, ts typeutil.Timestamp) error {
@@ -145,22 +143,27 @@ func (kc *Catalog) AlterDatabase(ctx context.Context, newColl *model.Database, t
 	if err != nil {
 		return err
 	}
-	return kc.Snapshot.Save(ctx, key, string(v), ts)
+	return kc.Txn.Save(ctx, key, string(v))
 }
 
 func (kc *Catalog) DropDatabase(ctx context.Context, dbID int64, ts typeutil.Timestamp) error {
 	key := BuildDatabaseKey(dbID)
-	return kc.Snapshot.MultiSaveAndRemove(ctx, nil, []string{key}, ts)
+	return kc.Txn.MultiSaveAndRemove(ctx, nil, []string{key})
 }
 
 func (kc *Catalog) ListDatabases(ctx context.Context, ts typeutil.Timestamp) ([]*model.Database, error) {
-	_, vals, err := kc.Snapshot.LoadWithPrefix(ctx, DBInfoMetaPrefix+"/", ts)
+	_, vals, err := kc.Txn.LoadWithPrefix(ctx, DBInfoMetaPrefix+"/")
 	if err != nil {
 		return nil, err
 	}
 
 	dbs := make([]*model.Database, 0, len(vals))
 	for _, val := range vals {
+		// Skip tombstone values left by old SuffixSnapshot (DropDatabase with ts!=0
+		// wrote a 3-byte tombstone instead of deleting the key).
+		if IsTombstone(val) {
+			continue
+		}
 		dbMeta := &pb.DatabaseInfo{}
 		err := proto.Unmarshal([]byte(val), dbMeta)
 		if err != nil {
@@ -240,18 +243,18 @@ func (kc *Catalog) CreateCollection(ctx context.Context, coll *model.Collection,
 	// cause the idempotency check in AddCollection to short-circuit and skip saving fields.
 	maxTxnNum := paramtable.Get().MetaStoreCfg.MaxEtcdTxnNum.GetAsInt()
 	if err := etcd.SaveByBatchWithLimit(kvs, maxTxnNum, func(partialKvs map[string]string) error {
-		return kc.Snapshot.MultiSave(ctx, partialKvs, ts)
+		return kc.Txn.MultiSave(ctx, partialKvs)
 	}); err != nil {
 		return err
 	}
 
 	// Save the collection key last — this is the "commit point" that makes the collection visible.
-	return kc.Snapshot.Save(ctx, k1, string(v1), ts)
+	return kc.Txn.Save(ctx, k1, string(v1))
 }
 
 func (kc *Catalog) loadCollectionFromDb(ctx context.Context, dbID int64, collectionID typeutil.UniqueID, ts typeutil.Timestamp) (*pb.CollectionInfo, error) {
 	collKey := BuildCollectionKey(dbID, collectionID)
-	collVal, err := kc.Snapshot.Load(ctx, collKey, ts)
+	collVal, err := kc.Txn.Load(ctx, collKey)
 	if err != nil {
 		return nil, merr.WrapErrCollectionNotFound(collectionID, err.Error())
 	}
@@ -309,7 +312,7 @@ func (kc *Catalog) CreatePartition(ctx context.Context, dbID int64, partition *m
 		if err != nil {
 			return err
 		}
-		return kc.Snapshot.Save(ctx, k, string(v), ts)
+		return kc.Txn.Save(ctx, k, string(v))
 	}
 
 	if partitionExistByID(collMeta, partition.PartitionID) {
@@ -331,7 +334,7 @@ func (kc *Catalog) CreatePartition(ctx context.Context, dbID int64, partition *m
 	if err != nil {
 		return err
 	}
-	return kc.Snapshot.Save(ctx, k, string(v), ts)
+	return kc.Txn.Save(ctx, k, string(v))
 }
 
 func (kc *Catalog) CreateAlias(ctx context.Context, alias *model.Alias, ts typeutil.Timestamp) error {
@@ -344,7 +347,7 @@ func (kc *Catalog) CreateAlias(ctx context.Context, alias *model.Alias, ts typeu
 		return err
 	}
 	kvs := map[string]string{k: string(v)}
-	return kc.Snapshot.MultiSaveAndRemove(ctx, kvs, []string{oldKBefore210, oldKeyWithoutDb}, ts)
+	return kc.Txn.MultiSaveAndRemove(ctx, kvs, []string{oldKBefore210, oldKeyWithoutDb})
 }
 
 func (kc *Catalog) AlterCredential(ctx context.Context, credential *model.Credential) error {
@@ -367,7 +370,7 @@ func (kc *Catalog) AlterCredential(ctx context.Context, credential *model.Creden
 
 func (kc *Catalog) listPartitionsAfter210(ctx context.Context, collectionID typeutil.UniqueID, ts typeutil.Timestamp) ([]*model.Partition, error) {
 	prefix := BuildPartitionPrefix(collectionID)
-	_, values, err := kc.Snapshot.LoadWithPrefix(ctx, prefix, ts)
+	_, values, err := kc.Txn.LoadWithPrefix(ctx, prefix)
 	if err != nil {
 		return nil, err
 	}
@@ -384,7 +387,7 @@ func (kc *Catalog) listPartitionsAfter210(ctx context.Context, collectionID type
 }
 
 func (kc *Catalog) batchListPartitionsAfter210(ctx context.Context, ts typeutil.Timestamp) (map[int64][]*model.Partition, error) {
-	_, values, err := kc.Snapshot.LoadWithPrefix(ctx, PartitionMetaPrefix+"/", ts)
+	_, values, err := kc.Txn.LoadWithPrefix(ctx, PartitionMetaPrefix+"/")
 	if err != nil {
 		return nil, err
 	}
@@ -411,7 +414,7 @@ func fieldVersionAfter210(collMeta *pb.CollectionInfo) bool {
 
 func (kc *Catalog) listFieldsAfter210(ctx context.Context, collectionID typeutil.UniqueID, ts typeutil.Timestamp) ([]*model.Field, error) {
 	prefix := BuildFieldPrefix(collectionID)
-	_, values, err := kc.Snapshot.LoadWithPrefix(ctx, prefix, ts)
+	_, values, err := kc.Txn.LoadWithPrefix(ctx, prefix)
 	if err != nil {
 		return nil, err
 	}
@@ -428,7 +431,7 @@ func (kc *Catalog) listFieldsAfter210(ctx context.Context, collectionID typeutil
 }
 
 func (kc *Catalog) batchListFieldsAfter210(ctx context.Context, ts typeutil.Timestamp) (map[int64][]*model.Field, error) {
-	keys, values, err := kc.Snapshot.LoadWithPrefix(ctx, FieldMetaPrefix+"/", ts)
+	keys, values, err := kc.Txn.LoadWithPrefix(ctx, FieldMetaPrefix+"/")
 	if err != nil {
 		return nil, err
 	}
@@ -455,7 +458,7 @@ func (kc *Catalog) batchListFieldsAfter210(ctx context.Context, ts typeutil.Time
 
 func (kc *Catalog) listStructArrayFieldsAfter210(ctx context.Context, collectionID typeutil.UniqueID, ts typeutil.Timestamp) ([]*model.StructArrayField, error) {
 	prefix := BuildStructArrayFieldPrefix(collectionID)
-	_, values, err := kc.Snapshot.LoadWithPrefix(ctx, prefix, ts)
+	_, values, err := kc.Txn.LoadWithPrefix(ctx, prefix)
 	if err != nil {
 		return nil, err
 	}
@@ -473,7 +476,7 @@ func (kc *Catalog) listStructArrayFieldsAfter210(ctx context.Context, collection
 
 func (kc *Catalog) listFunctions(ctx context.Context, collectionID typeutil.UniqueID, ts typeutil.Timestamp) ([]*model.Function, error) {
 	prefix := BuildFunctionPrefix(collectionID)
-	_, values, err := kc.Snapshot.LoadWithPrefix(ctx, prefix, ts)
+	_, values, err := kc.Txn.LoadWithPrefix(ctx, prefix)
 	if err != nil {
 		return nil, err
 	}
@@ -490,7 +493,7 @@ func (kc *Catalog) listFunctions(ctx context.Context, collectionID typeutil.Uniq
 }
 
 func (kc *Catalog) batchListFunctions(ctx context.Context, ts typeutil.Timestamp) (map[int64][]*model.Function, error) {
-	keys, values, err := kc.Snapshot.LoadWithPrefix(ctx, FunctionMetaPrefix+"/", ts)
+	keys, values, err := kc.Txn.LoadWithPrefix(ctx, FunctionMetaPrefix+"/")
 	if err != nil {
 		return nil, err
 	}
@@ -680,17 +683,16 @@ func (kc *Catalog) DropCollection(ctx context.Context, collectionInfo *model.Col
 	// delMetakeysSnap = append(delMetakeysSnap, buildPartitionPrefix(collectionInfo.CollectionID))
 	// delMetakeysSnap = append(delMetakeysSnap, buildFieldPrefix(collectionInfo.CollectionID))
 
-	// Though batchMultiSaveAndRemoveWithPrefix is not atomic enough, we can promise atomicity outside.
-	// If we found collection under dropping state, we'll know that gc is not completely on this collection.
-	// However, if we remove collection first, we cannot remove other metas.
-	// since SnapshotKV may save both snapshot key and the original key if the original key is newest
+	// Remove related metadata first, then the collection key itself.
+	// If RootCoord crashes in between, the collection stays in Dropping state
+	// and the tombstone sweeper will retry on next startup.
 	maxTxnNum := paramtable.Get().MetaStoreCfg.MaxEtcdTxnNum.GetAsInt()
-	if err := batchMultiSaveAndRemove(ctx, kc.Snapshot, maxTxnNum, nil, delMetakeysSnap, ts); err != nil {
+	if err := batchMultiSaveAndRemove(ctx, kc.Txn, maxTxnNum, nil, delMetakeysSnap); err != nil {
 		return err
 	}
 
 	// if we found collection dropping, we should try removing related resources.
-	return kc.Snapshot.MultiSaveAndRemove(ctx, nil, collectionKeys, ts)
+	return kc.Txn.MultiSaveAndRemove(ctx, nil, collectionKeys)
 }
 
 func (kc *Catalog) alterModifyCollection(ctx context.Context, oldColl *model.Collection, newColl *model.Collection, ts typeutil.Timestamp, fieldModify bool) error {
@@ -761,7 +763,7 @@ func (kc *Catalog) alterModifyCollection(ctx context.Context, oldColl *model.Col
 
 	maxTxnNum := paramtable.Get().MetaStoreCfg.MaxEtcdTxnNum.GetAsInt()
 	return etcd.SaveByBatchWithLimit(saves, maxTxnNum, func(partialKvs map[string]string) error {
-		return kc.Snapshot.MultiSave(ctx, partialKvs, ts)
+		return kc.Txn.MultiSave(ctx, partialKvs)
 	})
 }
 
@@ -787,7 +789,7 @@ func (kc *Catalog) AlterCollectionDB(ctx context.Context, oldColl *model.Collect
 	}
 	saves := map[string]string{newKey: string(value)}
 
-	return kc.Snapshot.MultiSaveAndRemove(ctx, saves, []string{oldKey}, ts)
+	return kc.Txn.MultiSaveAndRemove(ctx, saves, []string{oldKey})
 }
 
 func (kc *Catalog) alterModifyPartition(ctx context.Context, oldPart *model.Partition, newPart *model.Partition, ts typeutil.Timestamp) error {
@@ -804,7 +806,7 @@ func (kc *Catalog) alterModifyPartition(ctx context.Context, oldPart *model.Part
 	if err != nil {
 		return err
 	}
-	return kc.Snapshot.Save(ctx, key, string(value), ts)
+	return kc.Txn.Save(ctx, key, string(value))
 }
 
 func (kc *Catalog) AlterPartition(ctx context.Context, dbID int64, oldPart *model.Partition, newPart *model.Partition, alterType metastore.AlterType, ts typeutil.Timestamp) error {
@@ -848,7 +850,7 @@ func (kc *Catalog) DropPartition(ctx context.Context, dbID int64, collectionID t
 
 	if partitionVersionAfter210(collMeta) {
 		k := BuildPartitionKey(collectionID, partitionID)
-		return kc.Snapshot.MultiSaveAndRemove(ctx, nil, []string{k}, ts)
+		return kc.Txn.MultiSaveAndRemove(ctx, nil, []string{k})
 	}
 
 	k := BuildCollectionKey(util.NonDBID, collectionID)
@@ -857,7 +859,7 @@ func (kc *Catalog) DropPartition(ctx context.Context, dbID int64, collectionID t
 	if err != nil {
 		return err
 	}
-	return kc.Snapshot.Save(ctx, k, string(v), ts)
+	return kc.Txn.Save(ctx, k, string(v))
 }
 
 func (kc *Catalog) DropCredential(ctx context.Context, username string) error {
@@ -890,12 +892,12 @@ func (kc *Catalog) DropAlias(ctx context.Context, dbID int64, alias string, ts t
 	oldKBefore210 := BuildAliasKey210(alias)
 	oldKeyWithoutDb := BuildAliasKey(alias)
 	k := BuildAliasKeyWithDB(dbID, alias)
-	return kc.Snapshot.MultiSaveAndRemove(ctx, nil, []string{k, oldKeyWithoutDb, oldKBefore210}, ts)
+	return kc.Txn.MultiSaveAndRemove(ctx, nil, []string{k, oldKeyWithoutDb, oldKBefore210})
 }
 
 func (kc *Catalog) GetCollectionByName(ctx context.Context, dbID int64, dbName string, collectionName string, ts typeutil.Timestamp) (*model.Collection, error) {
 	prefix := getDatabasePrefix(dbID)
-	_, vals, err := kc.Snapshot.LoadWithPrefix(ctx, prefix, ts)
+	_, vals, err := kc.Txn.LoadWithPrefix(ctx, prefix)
 	if err != nil {
 		log.Ctx(ctx).Warn("get collection meta fail", zap.String("collectionName", collectionName), zap.Error(err))
 		return nil, err
@@ -919,7 +921,7 @@ func (kc *Catalog) GetCollectionByName(ctx context.Context, dbID int64, dbName s
 
 func (kc *Catalog) ListCollections(ctx context.Context, dbID int64, ts typeutil.Timestamp) ([]*model.Collection, error) {
 	prefix := getDatabasePrefix(dbID)
-	_, vals, err := kc.Snapshot.LoadWithPrefix(ctx, prefix, ts)
+	_, vals, err := kc.Txn.LoadWithPrefix(ctx, prefix)
 	if err != nil {
 		log.Ctx(ctx).Error("get collections meta fail",
 			zap.String("prefix", prefix),
@@ -975,13 +977,16 @@ func (kc *Catalog) fixDefaultDBIDConsistency(ctx context.Context, collMeta *pb.C
 }
 
 func (kc *Catalog) listAliasesBefore210(ctx context.Context, ts typeutil.Timestamp) ([]*model.Alias, error) {
-	_, values, err := kc.Snapshot.LoadWithPrefix(ctx, CollectionAliasMetaPrefix210+"/", ts)
+	_, values, err := kc.Txn.LoadWithPrefix(ctx, CollectionAliasMetaPrefix210+"/")
 	if err != nil {
 		return nil, err
 	}
 	// aliases before 210 stored by CollectionInfo.
 	aliases := make([]*model.Alias, 0, len(values))
 	for _, value := range values {
+		if IsTombstone(value) {
+			continue
+		}
 		coll := &pb.CollectionInfo{}
 		err := proto.Unmarshal([]byte(value), coll)
 		if err != nil {
@@ -999,13 +1004,16 @@ func (kc *Catalog) listAliasesBefore210(ctx context.Context, ts typeutil.Timesta
 
 func (kc *Catalog) listAliasesAfter210WithDb(ctx context.Context, dbID int64, ts typeutil.Timestamp) ([]*model.Alias, error) {
 	prefix := BuildAliasPrefixWithDB(dbID)
-	_, values, err := kc.Snapshot.LoadWithPrefix(ctx, prefix, ts)
+	_, values, err := kc.Txn.LoadWithPrefix(ctx, prefix)
 	if err != nil {
 		return nil, err
 	}
 	// aliases after 210 stored by AliasInfo.
 	aliases := make([]*model.Alias, 0, len(values))
 	for _, value := range values {
+		if IsTombstone(value) {
+			continue
+		}
 		info := &pb.AliasInfo{}
 		err := proto.Unmarshal([]byte(value), info)
 		if err != nil {

internal/metastore/kv/rootcoord/legacy_snapshot_gc.go

@@ -0,0 +1,143 @@
+package rootcoord
+
+import (
+	"context"
+	"fmt"
+	"strings"
+	"sync"
+	"time"
+
+	"github.com/cockroachdb/errors"
+	"go.uber.org/zap"
+
+	"github.com/milvus-io/milvus/pkg/v2/kv"
+	"github.com/milvus-io/milvus/pkg/v2/log"
+	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
+	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
+)
+
+const (
+	legacyGCBatchSize = 2000
+	legacyGCInterval  = 5 * time.Second
+)
+
+// errBatchFull is a sentinel error to stop WalkWithPrefix after collecting enough keys.
+var errBatchFull = fmt.Errorf("legacy GC batch full")
+
+var legacyGCOnce sync.Once
+
+// StartLegacySnapshotGC starts a background goroutine to clean up legacy
+// snapshot keys (prefix "snapshots/") left from before the SuffixSnapshot removal.
+// The goroutine self-terminates once all keys are cleaned.
+// Safe to call multiple times — only the first call starts the goroutine.
+//
+// Safety: it ONLY operates on the "snapshots/" prefix, which is disjoint from all
+// plain key paths (root-coord/, datacoord-meta/, etc.). WalkWithPrefix physically
+// cannot match a plain key, so this cannot corrupt any active metadata.
+func StartLegacySnapshotGC(ctx context.Context, metaKV kv.MetaKv) {
+	legacyGCOnce.Do(func() {
+		go runLegacySnapshotGC(ctx, metaKV)
+	})
+}
+
+func runLegacySnapshotGC(ctx context.Context, metaKV kv.MetaKv) {
+	logger := log.Ctx(ctx)
+	logger.Info("legacy snapshot GC started",
+		zap.String("prefix", SnapshotPrefix+"/"),
+		zap.Int("batchSize", legacyGCBatchSize),
+		zap.Duration("interval", legacyGCInterval))
+
+	ticker := time.NewTicker(legacyGCInterval)
+	defer ticker.Stop()
+
+	totalDeleted := 0
+	startTime := time.Now()
+
+	for {
+		select {
+		case <-ctx.Done():
+			logger.Info("legacy snapshot GC stopped (context canceled)",
+				zap.Int("totalDeleted", totalDeleted),
+				zap.Duration("totalElapsed", time.Since(startTime)))
+			return
+		case <-ticker.C:
+			deleted, err := gcLegacySnapshotBatch(ctx, metaKV)
+			if err != nil {
+				logger.Warn("legacy snapshot GC batch error",
+					zap.Error(err),
+					zap.Int("totalDeleted", totalDeleted))
+				continue
+			}
+			if deleted == 0 {
+				logger.Info("legacy snapshot GC complete",
+					zap.Int("totalDeleted", totalDeleted),
+					zap.Duration("totalElapsed", time.Since(startTime)))
+				return
+			}
+			totalDeleted += deleted
+			logger.Info("legacy snapshot GC batch done",
+				zap.Int("deleted", deleted),
+				zap.Int("totalDeleted", totalDeleted),
+				zap.Duration("elapsed", time.Since(startTime)))
+		}
+	}
+}
+
+// gcLegacySnapshotBatch collects up to legacyGCBatchSize snapshot keys and
+// deletes them in batched etcd transactions.
+// Returns (0, nil) when no more snapshot keys exist.
+func gcLegacySnapshotBatch(ctx context.Context, metaKV kv.MetaKv) (int, error) {
+	// Phase 1: Collect keys under "snapshots/" prefix.
+	// MetaKv internally prepends rootPath (e.g. "by-dev/meta/"), so this scans
+	// "by-dev/meta/snapshots/...". Plain keys live under "by-dev/meta/root-coord/",
+	// "by-dev/meta/datacoord-meta/", etc. — no overlap is possible.
+	keys := make([]string, 0, legacyGCBatchSize)
+	err := metaKV.WalkWithPrefix(ctx, SnapshotPrefix+"/", legacyGCBatchSize,
+		func(k []byte, v []byte) error {
+			keys = append(keys, string(k))
+			if len(keys) >= legacyGCBatchSize {
+				return errBatchFull
+			}
+			return nil
+		})
+	if err != nil && !errors.Is(err, errBatchFull) {
+		return 0, err
+	}
+
+	if len(keys) == 0 {
+		return 0, nil
+	}
+
+	// Phase 2: Strip rootPath prefix to get relative keys.
+	// WalkWithPrefix returns full etcd paths (with rootPath), but MultiRemove
+	// expects relative paths (it adds rootPath internally).
+	// GetPath("x") returns rootPath + "/" + "x". So GetPath(SnapshotPrefix) gives
+	// us the full etcd prefix for snapshot keys, which we can use to strip.
+	snapshotFullPrefix := metaKV.GetPath(SnapshotPrefix)
+	relativeKeys := make([]string, 0, len(keys))
+	for _, key := range keys {
+		if !strings.HasPrefix(key, snapshotFullPrefix) {
+			log.Warn("legacy snapshot GC: key not under expected prefix, skipping",
+				zap.String("key", key), zap.String("expectedPrefix", snapshotFullPrefix))
+			continue
+		}
+		// Reconstruct relative key: "snapshots/" + suffix after full prefix
+		rel := SnapshotPrefix + key[len(snapshotFullPrefix):]
+		relativeKeys = append(relativeKeys, rel)
+	}
+
+	if len(relativeKeys) == 0 {
+		return 0, nil
+	}
+
+	// Phase 3: Delete in batched transactions.
+	maxTxnNum := paramtable.Get().MetaStoreCfg.MaxEtcdTxnNum.GetAsInt()
+	removeFn := func(partialKeys []string) error {
+		return metaKV.MultiRemove(ctx, partialKeys)
+	}
+	if err := etcd.RemoveByBatchWithLimit(relativeKeys, maxTxnNum, removeFn); err != nil {
+		return 0, err
+	}
+
+	return len(relativeKeys), nil
+}

internal/metastore/kv/rootcoord/legacy_snapshot_gc_test.go

@@ -0,0 +1,246 @@
+package rootcoord
+
+import (
+	"context"
+	"fmt"
+	"math/rand"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/mock"
+	"github.com/stretchr/testify/require"
+
+	etcdkv "github.com/milvus-io/milvus/internal/kv/etcd"
+	"github.com/milvus-io/milvus/internal/kv/mocks"
+	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
+)
+
+func Test_gcLegacySnapshotBatch(t *testing.T) {
+	etcdCli, err := etcd.GetEtcdClient(
+		Params.EtcdCfg.UseEmbedEtcd.GetAsBool(),
+		Params.EtcdCfg.EtcdUseSSL.GetAsBool(),
+		Params.EtcdCfg.Endpoints.GetAsStrings(),
+		Params.EtcdCfg.EtcdTLSCert.GetValue(),
+		Params.EtcdCfg.EtcdTLSKey.GetValue(),
+		Params.EtcdCfg.EtcdTLSCACert.GetValue(),
+		Params.EtcdCfg.EtcdTLSMinVersion.GetValue())
+	require.NoError(t, err)
+	defer etcdCli.Close()
+
+	rootPath := fmt.Sprintf("/test/meta/legacy-gc-%d", rand.Int())
+	metaKV := etcdkv.NewEtcdKV(etcdCli, rootPath)
+	defer metaKV.Close()
+	defer metaKV.RemoveWithPrefix(context.TODO(), "")
+
+	ctx := context.TODO()
+
+	// Seed 50 snapshot keys (to be deleted)
+	for i := 0; i < 50; i++ {
+		key := fmt.Sprintf("%s/root-coord/fields/%d/0_ts%d", SnapshotPrefix, 100+i, 438497159122780160+i)
+		err := metaKV.Save(ctx, key, fmt.Sprintf("snapshot-value-%d", i))
+		require.NoError(t, err)
+	}
+
+	// Seed 5 plain keys (must NOT be deleted)
+	for i := 0; i < 5; i++ {
+		key := fmt.Sprintf("root-coord/fields/%d/0", 200+i)
+		err := metaKV.Save(ctx, key, fmt.Sprintf("plain-value-%d", i))
+		require.NoError(t, err)
+	}
+
+	t.Run("batch deletes snapshot keys", func(t *testing.T) {
+		deleted, err := gcLegacySnapshotBatch(ctx, metaKV)
+		assert.NoError(t, err)
+		assert.Equal(t, 50, deleted)
+	})
+
+	t.Run("returns zero when no more snapshot keys", func(t *testing.T) {
+		deleted, err := gcLegacySnapshotBatch(ctx, metaKV)
+		assert.NoError(t, err)
+		assert.Equal(t, 0, deleted, "should find no more snapshot keys")
+	})
+
+	t.Run("plain keys are untouched", func(t *testing.T) {
+		for i := 0; i < 5; i++ {
+			key := fmt.Sprintf("root-coord/fields/%d/0", 200+i)
+			val, err := metaKV.Load(ctx, key)
+			assert.NoError(t, err, "plain key %s should still exist", key)
+			assert.Equal(t, fmt.Sprintf("plain-value-%d", i), val)
+		}
+	})
+}
+
+func Test_gcLegacySnapshotBatch_LargeBatch(t *testing.T) {
+	etcdCli, err := etcd.GetEtcdClient(
+		Params.EtcdCfg.UseEmbedEtcd.GetAsBool(),
+		Params.EtcdCfg.EtcdUseSSL.GetAsBool(),
+		Params.EtcdCfg.Endpoints.GetAsStrings(),
+		Params.EtcdCfg.EtcdTLSCert.GetValue(),
+		Params.EtcdCfg.EtcdTLSKey.GetValue(),
+		Params.EtcdCfg.EtcdTLSCACert.GetValue(),
+		Params.EtcdCfg.EtcdTLSMinVersion.GetValue())
+	require.NoError(t, err)
+	defer etcdCli.Close()
+
+	rootPath := fmt.Sprintf("/test/meta/legacy-gc-large-%d", rand.Int())
+	metaKV := etcdkv.NewEtcdKV(etcdCli, rootPath)
+	defer metaKV.Close()
+	defer metaKV.RemoveWithPrefix(context.TODO(), "")
+
+	ctx := context.TODO()
+
+	// Seed 5000 snapshot keys — more than one batch (legacyGCBatchSize=2000)
+	for i := 0; i < 5000; i++ {
+		key := fmt.Sprintf("%s/root-coord/fields/%d/0_ts%d", SnapshotPrefix, i, 438497159122780160+i)
+		err := metaKV.Save(ctx, key, fmt.Sprintf("v%d", i))
+		require.NoError(t, err)
+	}
+
+	// Batch 1: should delete legacyGCBatchSize keys
+	deleted1, err := gcLegacySnapshotBatch(ctx, metaKV)
+	assert.NoError(t, err)
+	assert.Equal(t, legacyGCBatchSize, deleted1)
+
+	// Batch 2: should delete another legacyGCBatchSize keys
+	deleted2, err := gcLegacySnapshotBatch(ctx, metaKV)
+	assert.NoError(t, err)
+	assert.Equal(t, legacyGCBatchSize, deleted2)
+
+	// Batch 3: remaining keys
+	deleted3, err := gcLegacySnapshotBatch(ctx, metaKV)
+	assert.NoError(t, err)
+	assert.Equal(t, 5000-2*legacyGCBatchSize, deleted3)
+
+	// Batch 4: no more keys
+	deleted4, err := gcLegacySnapshotBatch(ctx, metaKV)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, deleted4)
+}
+
+func Test_gcLegacySnapshotBatch_WalkError(t *testing.T) {
+	kvMock := mocks.NewMetaKv(t)
+	kvMock.EXPECT().WalkWithPrefix(mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		Return(fmt.Errorf("etcd unavailable"))
+
+	deleted, err := gcLegacySnapshotBatch(context.TODO(), kvMock)
+	assert.Error(t, err)
+	assert.Equal(t, 0, deleted)
+}
+
+func Test_gcLegacySnapshotBatch_PrefixMismatch(t *testing.T) {
+	// Simulate: WalkWithPrefix returns keys that don't match expected prefix.
+	// This should never happen in practice but tests the safety guard.
+	kvMock := mocks.NewMetaKv(t)
+	kvMock.EXPECT().WalkWithPrefix(mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		RunAndReturn(func(ctx context.Context, prefix string, paginationSize int, fn func([]byte, []byte) error) error {
+			// Return a key with wrong prefix
+			return fn([]byte("wrong-prefix/some-key"), []byte("value"))
+		})
+	kvMock.EXPECT().GetPath(SnapshotPrefix).Return("root/snapshots")
+
+	deleted, err := gcLegacySnapshotBatch(context.TODO(), kvMock)
+	assert.NoError(t, err)
+	assert.Equal(t, 0, deleted, "mismatched keys should be skipped, nothing deleted")
+}
+
+func Test_gcLegacySnapshotBatch_DeleteError(t *testing.T) {
+	etcdCli, err := etcd.GetEtcdClient(
+		Params.EtcdCfg.UseEmbedEtcd.GetAsBool(),
+		Params.EtcdCfg.EtcdUseSSL.GetAsBool(),
+		Params.EtcdCfg.Endpoints.GetAsStrings(),
+		Params.EtcdCfg.EtcdTLSCert.GetValue(),
+		Params.EtcdCfg.EtcdTLSKey.GetValue(),
+		Params.EtcdCfg.EtcdTLSCACert.GetValue(),
+		Params.EtcdCfg.EtcdTLSMinVersion.GetValue())
+	require.NoError(t, err)
+	defer etcdCli.Close()
+
+	rootPath := fmt.Sprintf("/test/meta/legacy-gc-delerr-%d", rand.Int())
+	metaKV := etcdkv.NewEtcdKV(etcdCli, rootPath)
+	defer metaKV.Close()
+	defer metaKV.RemoveWithPrefix(context.TODO(), "")
+
+	ctx := context.TODO()
+
+	// Seed some snapshot keys
+	for i := 0; i < 3; i++ {
+		key := fmt.Sprintf("%s/root-coord/fields/%d/0_ts%d", SnapshotPrefix, i, 438497159122780160+i)
+		err := metaKV.Save(ctx, key, fmt.Sprintf("v%d", i))
+		require.NoError(t, err)
+	}
+
+	// Use canceled context to trigger delete error
+	canceledCtx, cancel := context.WithCancel(ctx)
+	cancel()
+
+	deleted, err := gcLegacySnapshotBatch(canceledCtx, metaKV)
+	// Either returns error (context canceled) or succeeds — both are acceptable.
+	// The key point is it doesn't panic.
+	_ = deleted
+	_ = err
+}
+
+func Test_runLegacySnapshotGC_ContextCancel(t *testing.T) {
+	// Test that runLegacySnapshotGC exits when context is canceled.
+	kvMock := mocks.NewMetaKv(t)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel() // cancel immediately
+
+	// runLegacySnapshotGC should exit quickly via ctx.Done()
+	done := make(chan struct{})
+	go func() {
+		runLegacySnapshotGC(ctx, kvMock)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// success — goroutine exited
+	case <-time.After(3 * time.Second):
+		t.Fatal("runLegacySnapshotGC did not exit after context cancel")
+	}
+}
+
+func Test_runLegacySnapshotGC_SelfTerminate(t *testing.T) {
+	// Test that runLegacySnapshotGC self-terminates when no keys found.
+	kvMock := mocks.NewMetaKv(t)
+	// WalkWithPrefix returns 0 keys → deleted=0 → goroutine exits
+	kvMock.EXPECT().WalkWithPrefix(mock.Anything, mock.Anything, mock.Anything, mock.Anything).
+		Return(nil)
+
+	done := make(chan struct{})
+	go func() {
+		runLegacySnapshotGC(context.Background(), kvMock)
+		close(done)
+	}()
+
+	select {
+	case <-done:
+		// success — goroutine self-terminated
+	case <-time.After(10 * time.Second):
+		t.Fatal("runLegacySnapshotGC did not self-terminate")
+	}
+}
+
+func Test_LegacySnapshotUtils(t *testing.T) {
+	t.Run("IsTombstone", func(t *testing.T) {
+		assert.True(t, IsTombstone(string(SuffixSnapshotTombstone)))
+		assert.False(t, IsTombstone("normal-value"))
+		assert.False(t, IsTombstone(""))
+	})
+
+	t.Run("ConstructTombstone", func(t *testing.T) {
+		tomb := ConstructTombstone()
+		assert.Equal(t, SuffixSnapshotTombstone, tomb)
+		tomb[0] = 0x00
+		assert.NotEqual(t, SuffixSnapshotTombstone, tomb)
+	})
+
+	t.Run("ComposeSnapshotKey", func(t *testing.T) {
+		key := ComposeSnapshotKey("snapshots/", "root-coord/fields/100/0", "_ts", 12345)
+		assert.Contains(t, key, "snapshots/")
+		assert.Contains(t, key, "root-coord/fields/100/0_ts12345")
+	})
+}

internal/metastore/kv/rootcoord/rootcoord_constant.go

@@ -1,9 +1,11 @@
 package rootcoord
 
 import (
+	"bytes"
 	"fmt"
 
 	"github.com/milvus-io/milvus/pkg/v2/util"
+	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )
 
 const (
@@ -82,3 +84,23 @@ func getDatabasePrefix(dbID int64) string {
 func BuildPrivilegeGroupkey(groupName string) string {
 	return fmt.Sprintf("%s/%s", PrivilegeGroupPrefix, groupName)
 }
+
+// Legacy snapshot utilities — kept for migration tool compatibility only.
+
+// SuffixSnapshotTombstone is the tombstone marker used in legacy snapshot keys.
+var SuffixSnapshotTombstone = []byte{0xE2, 0x9B, 0xBC}
+
+// IsTombstone checks whether the value is a legacy tombstone marker.
+func IsTombstone(value string) bool {
+	return bytes.Equal([]byte(value), SuffixSnapshotTombstone)
+}
+
+// ConstructTombstone returns a copy of the tombstone marker.
+func ConstructTombstone() []byte {
+	return append([]byte{}, SuffixSnapshotTombstone...)
+}
+
+// ComposeSnapshotKey builds a legacy snapshot key from prefix, key, separator, and timestamp.
+func ComposeSnapshotKey(snapshotPrefix string, key string, separator string, ts typeutil.Timestamp) string {
+	return util.GetPath(snapshotPrefix, fmt.Sprintf("%s%s%d", key, separator, ts))
+}

internal/metastore/kv/rootcoord/suffix_snapshot.go

@@ -1,750 +0,0 @@
-// 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 rootcoord
-
-import (
-	"bytes"
-	"context"
-	"fmt"
-	"sort"
-	"strconv"
-	"strings"
-	"sync"
-	"time"
-
-	"github.com/cockroachdb/errors"
-	"go.uber.org/zap"
-
-	"github.com/milvus-io/milvus/pkg/v2/common"
-	"github.com/milvus-io/milvus/pkg/v2/kv"
-	"github.com/milvus-io/milvus/pkg/v2/log"
-	"github.com/milvus-io/milvus/pkg/v2/util"
-	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
-	"github.com/milvus-io/milvus/pkg/v2/util/merr"
-	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
-	"github.com/milvus-io/milvus/pkg/v2/util/retry"
-	"github.com/milvus-io/milvus/pkg/v2/util/tsoutil"
-	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
-)
-
-// SuffixSnapshotTombstone special value for tombstone mark
-var SuffixSnapshotTombstone = []byte{0xE2, 0x9B, 0xBC}
-
-// IsTombstone used in migration tool also.
-func IsTombstone(value string) bool {
-	return bytes.Equal([]byte(value), SuffixSnapshotTombstone)
-}
-
-// ConstructTombstone used in migration tool also.
-func ConstructTombstone() []byte {
-	return common.CloneByteSlice(SuffixSnapshotTombstone)
-}
-
-// SuffixSnapshot implements SnapshotKV
-// this is a simple replacement for MetaSnapshot, which is not available due to etcd compaction
-// SuffixSnapshot record timestamp as prefix of a key under the Snapshot prefix path
-type SuffixSnapshot struct {
-	// internal kv which SuffixSnapshot based on
-	kv.MetaKv
-	// rw mutex provided range lock
-	sync.RWMutex
-	// lastestTS latest timestamp for each key
-	// note that this map is lazy loaded
-	// which means if a key is never used in current session, no ts related to the key is stored
-	// so is ts is not recorded in map, a `load ts` process will be triggered
-	// a `0` for a key means that the key is not a snapshot key
-	lastestTS map[string]typeutil.Timestamp
-	// separator is the conjunction string between actual key and trailing ts
-	// used in composing ts-key and check whether a key is a ts-key
-	separator string
-	// rootPrefix is used to hide detail rootPrefix in kv
-	rootPrefix string
-	// rootLen pre calculated offset when hiding root prefix
-	rootLen int
-	// snapshotPrefix serves a prefix stores snapshot with ts
-	snapshotPrefix string
-	// snapshotLen pre calculated offset when parsing snapshot key
-	snapshotLen int
-
-	paginationSize int
-
-	closeGC chan struct{}
-}
-
-// tsv struct stores kv with timestamp
-type tsv struct {
-	value string
-	ts    typeutil.Timestamp
-}
-
-// type conversion make sure implementation
-var _ kv.SnapShotKV = (*SuffixSnapshot)(nil)
-
-// NewSuffixSnapshot creates a NewSuffixSnapshot with provided kv
-func NewSuffixSnapshot(metaKV kv.MetaKv, sep, root, snapshot string) (*SuffixSnapshot, error) {
-	if metaKV == nil {
-		return nil, retry.Unrecoverable(errors.New("MetaKv is nil"))
-	}
-
-	// ensure snapshot has trailing '/'
-	if !strings.HasSuffix(snapshot, "/") {
-		snapshot += "/"
-	}
-	snapshotLen := len(snapshot)
-	// keep empty root unchanged to preserve rootLen=0 behavior.
-	if root != "" && !strings.HasSuffix(root, "/") {
-		root += "/"
-	}
-	rootLen := len(root)
-
-	ss := &SuffixSnapshot{
-		MetaKv:         metaKV,
-		lastestTS:      make(map[string]typeutil.Timestamp),
-		separator:      sep,
-		snapshotPrefix: snapshot,
-		snapshotLen:    snapshotLen,
-		rootPrefix:     root,
-		rootLen:        rootLen,
-		paginationSize: paramtable.Get().MetaStoreCfg.PaginationSize.GetAsInt(),
-		closeGC:        make(chan struct{}, 1),
-	}
-	go ss.startBackgroundGC(context.TODO())
-	return ss, nil
-}
-
-// isTombstone helper function to check whether is tombstone mark
-func (ss *SuffixSnapshot) isTombstone(value string) bool {
-	return IsTombstone(value)
-}
-
-// hideRootPrefix helper function to hide root prefix from key
-func (ss *SuffixSnapshot) hideRootPrefix(value string) string {
-	return value[ss.rootLen:]
-}
-
-// composeSnapshotPrefix build a prefix for load snapshots
-// formated like [snapshotPrefix]key[sep]
-func (ss *SuffixSnapshot) composeSnapshotPrefix(key string) string {
-	return ss.snapshotPrefix + key + ss.separator
-}
-
-// ComposeSnapshotKey used in migration tool also, in case of any rules change.
-func ComposeSnapshotKey(snapshotPrefix string, key string, separator string, ts typeutil.Timestamp) string {
-	// [key][sep][ts]
-	return util.GetPath(snapshotPrefix, fmt.Sprintf("%s%s%d", key, separator, ts))
-}
-
-// composeTSKey unified tsKey composing method
-// uses key, ts and separator to form a key
-func (ss *SuffixSnapshot) composeTSKey(key string, ts typeutil.Timestamp) string {
-	// [key][sep][ts]
-	return ComposeSnapshotKey(ss.snapshotPrefix, key, ss.separator, ts)
-}
-
-// isTSKey checks whether a key is in ts-key format
-// if true, also returns parsed ts value
-func (ss *SuffixSnapshot) isTSKey(key string) (typeutil.Timestamp, bool) {
-	// not in snapshot path
-	if !strings.HasPrefix(key, ss.snapshotPrefix) {
-		return 0, false
-	}
-	key = key[ss.snapshotLen:]
-	idx := strings.LastIndex(key, ss.separator)
-	if idx == -1 {
-		return 0, false
-	}
-	ts, err := strconv.ParseUint(key[idx+len(ss.separator):], 10, 64)
-	if err != nil {
-		return 0, false
-	}
-	return ts, true
-}
-
-// isTSOfKey check whether a key is in ts-key format of provided group key
-// if true, laso returns parsed ts value
-func (ss *SuffixSnapshot) isTSOfKey(key string, groupKey string) (typeutil.Timestamp, bool) {
-	// not in snapshot path
-	if !strings.HasPrefix(key, ss.snapshotPrefix) {
-		return 0, false
-	}
-	key = key[ss.snapshotLen:]
-
-	idx := strings.LastIndex(key, ss.separator)
-	if idx == -1 {
-		return 0, false
-	}
-	if key[:idx] != groupKey {
-		return 0, false
-	}
-	ts, err := strconv.ParseUint(key[idx+len(ss.separator):], 10, 64)
-	if err != nil {
-		return 0, false
-	}
-	return ts, true
-}
-
-// checkKeyTS checks provided key's latest ts is before provided ts
-// lock is needed
-func (ss *SuffixSnapshot) checkKeyTS(ctx context.Context, key string, ts typeutil.Timestamp) (bool, error) {
-	latest, has := ss.lastestTS[key]
-	if !has {
-		err := ss.loadLatestTS(ctx, key)
-		if err != nil {
-			return false, err
-		}
-		latest = ss.lastestTS[key]
-	}
-	return latest <= ts, nil
-}
-
-// loadLatestTS load the loatest ts for specified key
-func (ss *SuffixSnapshot) loadLatestTS(ctx context.Context, key string) error {
-	prefix := ss.composeSnapshotPrefix(key)
-	keys, _, err := ss.MetaKv.LoadWithPrefix(ctx, prefix)
-	if err != nil {
-		log.Warn("SuffixSnapshot MetaKv LoadWithPrefix failed", zap.String("key", key),
-			zap.Error(err))
-		return err
-	}
-	tss := make([]typeutil.Timestamp, 0, len(keys))
-	for _, key := range keys {
-		ts, ok := ss.isTSKey(ss.hideRootPrefix(key))
-		if ok {
-			tss = append(tss, ts)
-		}
-	}
-	if len(tss) == 0 {
-		ss.lastestTS[key] = 0
-		return nil
-	}
-	sort.Slice(tss, func(i, j int) bool {
-		return tss[i] > tss[j]
-	})
-	ss.lastestTS[key] = tss[0]
-	return nil
-}
-
-// binarySearchRecords returns value in sorted order with index i
-// which satisfies records[i].ts <= ts && records[i+1].ts > ts
-func binarySearchRecords(records []tsv, ts typeutil.Timestamp) (string, bool) {
-	if len(records) == 0 {
-		return "", false
-	}
-	// sort records by ts
-	sort.Slice(records, func(i, j int) bool {
-		return records[i].ts < records[j].ts
-	})
-
-	if ts < records[0].ts {
-		return "", false
-	}
-	if ts > records[len(records)-1].ts {
-		return records[len(records)-1].value, true
-	}
-	i, j := 0, len(records)
-	for i+1 < j {
-		k := (i + j) / 2
-		// log.Warn("", zap.Int("i", i), zap.Int("j", j), zap.Int("k", k))
-		if records[k].ts == ts {
-			return records[k].value, true
-		}
-
-		if records[k].ts < ts {
-			i = k
-		} else {
-			j = k
-		}
-	}
-	return records[i].value, true
-}
-
-// Save stores key-value pairs with timestamp
-// if ts == 0, SuffixSnapshot works as a MetaKv
-// otherwise, SuffixSnapshot will store a ts-key as "key[sep]ts"-value pair in snapshot path
-// and for acceleration store original key-value if ts is the latest
-func (ss *SuffixSnapshot) Save(ctx context.Context, key string, value string, ts typeutil.Timestamp) error {
-	// if ts == 0, act like MetaKv
-	// will not update lastestTs since ts not not valid
-	if ts == 0 {
-		return ss.MetaKv.Save(ctx, key, value)
-	}
-
-	ss.Lock()
-	defer ss.Unlock()
-
-	tsKey := ss.composeTSKey(key, ts)
-
-	// provided key value is latest
-	// stores both tsKey and original key
-	after, err := ss.checkKeyTS(ctx, key, ts)
-	if err != nil {
-		return err
-	}
-	if after {
-		err := ss.MetaKv.MultiSave(ctx, map[string]string{
-			key:   value,
-			tsKey: value,
-		})
-		// update latestTS only when MultiSave succeeds
-		if err == nil {
-			ss.lastestTS[key] = ts
-		}
-		return err
-	}
-
-	// modifying history key, just save tskey-value
-	return ss.MetaKv.Save(ctx, tsKey, value)
-}
-
-func (ss *SuffixSnapshot) Load(ctx context.Context, key string, ts typeutil.Timestamp) (string, error) {
-	// if ts == 0 or typeutil.MaxTimestamp, load latest by definition
-	// and with acceleration logic, just do load key will do
-	if ts == 0 || ts == typeutil.MaxTimestamp {
-		value, err := ss.MetaKv.Load(ctx, key)
-		if ss.isTombstone(value) {
-			return "", errors.New("no value found")
-		}
-		return value, err
-	}
-
-	ss.Lock()
-	after, err := ss.checkKeyTS(ctx, key, ts)
-	ss.Unlock()
-
-	ss.RLock()
-	defer ss.RUnlock()
-	// ts after latest ts, load key as acceleration
-	if err != nil {
-		return "", err
-	}
-	if after {
-		value, err := ss.MetaKv.Load(ctx, key)
-		if ss.isTombstone(value) {
-			return "", errors.New("no value found")
-		}
-		return value, err
-	}
-
-	// before ts, do time travel
-	// 1. load all tsKey with key/ prefix
-	keys, values, err := ss.MetaKv.LoadWithPrefix(ctx, ss.composeSnapshotPrefix(key))
-	if err != nil {
-		log.Warn("prefixSnapshot MetaKv LoadWithPrefix failed", zap.String("key", key), zap.Error(err))
-		return "", err
-	}
-
-	records := make([]tsv, 0, len(keys))
-	// 2. validate key and parse ts
-	for i := 0; i < len(keys); i++ {
-		ts, ok := ss.isTSKey(ss.hideRootPrefix(keys[i]))
-		if !ok {
-			continue
-		}
-		records = append(records, tsv{
-			value: values[i],
-			ts:    ts,
-		})
-	}
-
-	if len(records) == 0 {
-		return "", errors.New("not value found")
-	}
-
-	// 3. find i which ts[i] <= ts && ts[i+1] > ts
-	// corner cases like len(records)==0, ts < records[0].ts is covered in binarySearch
-	// binary search
-	value, found := binarySearchRecords(records, ts)
-	if !found {
-		log.Warn("not found")
-		return "", errors.New("no value found")
-	}
-	// check whether value is tombstone
-	if ss.isTombstone(value) {
-		log.Warn("tombstone", zap.String("value", value))
-		return "", errors.New("not value found")
-	}
-	return value, nil
-}
-
-// MultiSave save multiple kvs
-// if ts == 0, act like MetaKv
-// each key-value will be treated using same logic like Save
-func (ss *SuffixSnapshot) MultiSave(ctx context.Context, kvs map[string]string, ts typeutil.Timestamp) error {
-	// if ts == 0, act like MetaKv
-	if ts == 0 {
-		return ss.MetaKv.MultiSave(ctx, kvs)
-	}
-	ss.Lock()
-	defer ss.Unlock()
-	var err error
-
-	// process each key, checks whether is the latest
-	execute, updateList, err := ss.generateSaveExecute(ctx, kvs, ts)
-	if err != nil {
-		return err
-	}
-
-	// multi save execute map; if succeeds, update ts in the update list
-	err = ss.MetaKv.MultiSave(ctx, execute)
-	if err == nil {
-		for _, key := range updateList {
-			ss.lastestTS[key] = ts
-		}
-	}
-	return err
-}
-
-// generateSaveExecute examine each key is the after the corresponding latest
-// returns calculated execute map and update ts list
-func (ss *SuffixSnapshot) generateSaveExecute(ctx context.Context, kvs map[string]string, ts typeutil.Timestamp) (map[string]string, []string, error) {
-	var after bool
-	var err error
-	execute := make(map[string]string)
-	updateList := make([]string, 0, len(kvs))
-	for key, value := range kvs {
-		tsKey := ss.composeTSKey(key, ts)
-		// provided key value is latest
-		// stores both tsKey and original key
-		after, err = ss.checkKeyTS(ctx, key, ts)
-		if err != nil {
-			return nil, nil, err
-		}
-		if after {
-			execute[key] = value
-			execute[tsKey] = value
-			updateList = append(updateList, key)
-		} else {
-			execute[tsKey] = value
-		}
-	}
-	return execute, updateList, nil
-}
-
-// LoadWithPrefix load keys with provided prefix and returns value in the ts
-func (ss *SuffixSnapshot) LoadWithPrefix(ctx context.Context, key string, ts typeutil.Timestamp) ([]string, []string, error) {
-	// ts 0 case shall be treated as fetch latest/current value
-	if ts == 0 || ts == typeutil.MaxTimestamp {
-		fks := make([]string, 0)
-		fvs := make([]string, 0)
-		// hide rootPrefix from return value
-		applyFn := func(key []byte, value []byte) error {
-			// filters tombstone
-			if ss.isTombstone(string(value)) {
-				return nil
-			}
-			fks = append(fks, ss.hideRootPrefix(string(key)))
-			fvs = append(fvs, string(value))
-			return nil
-		}
-
-		err := ss.MetaKv.WalkWithPrefix(ctx, key, ss.paginationSize, applyFn)
-		return fks, fvs, err
-	}
-	ss.Lock()
-	defer ss.Unlock()
-
-	resultKeys := make([]string, 0)
-	resultValues := make([]string, 0)
-
-	latestOriginalKey := ""
-	tValueGroups := make([]tsv, 0)
-
-	prefix := ss.snapshotPrefix + key
-	appendResultFn := func(ts typeutil.Timestamp) {
-		value, ok := binarySearchRecords(tValueGroups, ts)
-		if !ok || ss.isTombstone(value) {
-			return
-		}
-
-		resultKeys = append(resultKeys, latestOriginalKey)
-		resultValues = append(resultValues, value)
-	}
-
-	err := ss.MetaKv.WalkWithPrefix(ctx, prefix, ss.paginationSize, func(k []byte, v []byte) error {
-		sKey := string(k)
-		sValue := string(v)
-
-		snapshotKey := ss.hideRootPrefix(sKey)
-		curOriginalKey, err := ss.getOriginalKey(snapshotKey)
-		if err != nil {
-			return err
-		}
-
-		// reset if starting look up a new key group
-		if latestOriginalKey != "" && latestOriginalKey != curOriginalKey {
-			appendResultFn(ts)
-			tValueGroups = make([]tsv, 0)
-		}
-
-		targetTs, ok := ss.isTSKey(snapshotKey)
-		if !ok {
-			log.Warn("skip key because it doesn't contain ts", zap.String("key", key))
-			return nil
-		}
-
-		tValueGroups = append(tValueGroups, tsv{value: sValue, ts: targetTs})
-		latestOriginalKey = curOriginalKey
-		return nil
-	})
-	if err != nil {
-		return nil, nil, err
-	}
-
-	appendResultFn(ts)
-	return resultKeys, resultValues, nil
-}
-
-// MultiSaveAndRemove save muiltple kvs and remove as well
-// if ts == 0, act like MetaKv
-// each key-value will be treated in same logic like Save
-func (ss *SuffixSnapshot) MultiSaveAndRemove(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error {
-	// if ts == 0, act like MetaKv
-	if ts == 0 {
-		return ss.MetaKv.MultiSaveAndRemove(ctx, saves, removals)
-	}
-	ss.Lock()
-	defer ss.Unlock()
-	var err error
-
-	// process each key, checks whether is the latest
-	execute, updateList, err := ss.generateSaveExecute(ctx, saves, ts)
-	if err != nil {
-		return err
-	}
-
-	// load each removal, change execution to adding tombstones
-	for _, removal := range removals {
-		// if save batch contains removal, skip remove op
-		if _, ok := execute[removal]; ok {
-			continue
-		}
-		value, err := ss.MetaKv.Load(ctx, removal)
-		if err != nil {
-			log.Warn("SuffixSnapshot MetaKv Load failed", zap.String("key", removal), zap.Error(err))
-			if errors.Is(err, merr.ErrIoKeyNotFound) {
-				continue
-			}
-			return err
-		}
-		// add tombstone to original key and add ts entry
-		if IsTombstone(value) {
-			continue
-		}
-		execute[removal] = string(SuffixSnapshotTombstone)
-		execute[ss.composeTSKey(removal, ts)] = string(SuffixSnapshotTombstone)
-		updateList = append(updateList, removal)
-	}
-
-	maxTxnNum := paramtable.Get().MetaStoreCfg.MaxEtcdTxnNum.GetAsInt()
-	err = etcd.SaveByBatchWithLimit(execute, maxTxnNum, func(partialKvs map[string]string) error {
-		return ss.MetaKv.MultiSave(ctx, partialKvs)
-	})
-	if err == nil {
-		for _, key := range updateList {
-			ss.lastestTS[key] = ts
-		}
-	}
-	return err
-}
-
-// MultiSaveAndRemoveWithPrefix save muiltple kvs and remove as well
-// if ts == 0, act like MetaKv
-// each key-value will be treated in same logic like Save
-func (ss *SuffixSnapshot) MultiSaveAndRemoveWithPrefix(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error {
-	// if ts == 0, act like MetaKv
-	if ts == 0 {
-		return ss.MetaKv.MultiSaveAndRemoveWithPrefix(ctx, saves, removals)
-	}
-	ss.Lock()
-	defer ss.Unlock()
-	var err error
-
-	// process each key, checks whether is the latest
-	execute, updateList, err := ss.generateSaveExecute(ctx, saves, ts)
-	if err != nil {
-		return err
-	}
-
-	// load each removal, change execution to adding tombstones
-	for _, removal := range removals {
-		keys, values, err := ss.MetaKv.LoadWithPrefix(ctx, removal)
-		if err != nil {
-			log.Warn("SuffixSnapshot MetaKv LoadwithPrefix failed", zap.String("key", removal), zap.Error(err))
-			return err
-		}
-
-		// add tombstone to original key and add ts entry
-		for idx, key := range keys {
-			if IsTombstone(values[idx]) {
-				continue
-			}
-			key = ss.hideRootPrefix(key)
-			execute[key] = string(SuffixSnapshotTombstone)
-			execute[ss.composeTSKey(key, ts)] = string(SuffixSnapshotTombstone)
-			updateList = append(updateList, key)
-		}
-	}
-
-	// multi save execute map; if succeeds, update ts in the update list
-	err = ss.MetaKv.MultiSave(ctx, execute)
-	if err == nil {
-		for _, key := range updateList {
-			ss.lastestTS[key] = ts
-		}
-	}
-	return err
-}
-
-func (ss *SuffixSnapshot) Close() {
-	close(ss.closeGC)
-}
-
-// startBackgroundGC the data will clean up if key ts!=0 and expired
-func (ss *SuffixSnapshot) startBackgroundGC(ctx context.Context) {
-	log := log.Ctx(ctx)
-	log.Debug("suffix snapshot GC goroutine start!")
-	ticker := time.NewTicker(60 * time.Minute)
-	defer ticker.Stop()
-
-	for {
-		select {
-		case <-ss.closeGC:
-			log.Warn("quit suffix snapshot GC goroutine!")
-			return
-		case now := <-ticker.C:
-			err := ss.removeExpiredKvs(ctx, now)
-			if err != nil {
-				log.Warn("remove expired data fail during GC", zap.Error(err))
-			}
-		}
-	}
-}
-
-func (ss *SuffixSnapshot) getOriginalKey(snapshotKey string) (string, error) {
-	if !strings.HasPrefix(snapshotKey, ss.snapshotPrefix) {
-		return "", fmt.Errorf("get original key failed, invaild snapshot key:%s", snapshotKey)
-	}
-	// collect keys that parent node is snapshot node if the corresponding the latest ts is expired.
-	idx := strings.LastIndex(snapshotKey, ss.separator)
-	if idx == -1 {
-		return "", fmt.Errorf("get original key failed, snapshot key:%s", snapshotKey)
-	}
-	prefix := snapshotKey[:idx]
-	return prefix[ss.snapshotLen:], nil
-}
-
-func (ss *SuffixSnapshot) batchRemoveExpiredKvs(ctx context.Context, keyGroup []string, originalKey string, includeOriginalKey bool) error {
-	if includeOriginalKey {
-		keyGroup = append(keyGroup, originalKey)
-	}
-
-	// to protect txn finished with ascend order, reverse the latest kv with tombstone to tail of array
-	sort.Strings(keyGroup)
-	if !includeOriginalKey && len(keyGroup) > 0 {
-		// keep the latest snapshot key for historical version compatibility
-		keyGroup = keyGroup[0 : len(keyGroup)-1]
-	}
-	removeFn := func(partialKeys []string) error {
-		return ss.MetaKv.MultiRemove(ctx, partialKeys)
-	}
-	maxTxnNum := paramtable.Get().MetaStoreCfg.MaxEtcdTxnNum.GetAsInt()
-	return etcd.RemoveByBatchWithLimit(keyGroup, maxTxnNum, removeFn)
-}
-
-// removeExpiredKvs removes expired key-value pairs from the snapshot
-// It walks through all keys with the snapshot prefix, groups them by original key,
-// and removes expired versions or all versions if the original key has been deleted
-func (ss *SuffixSnapshot) removeExpiredKvs(ctx context.Context, now time.Time) error {
-	log := log.Ctx(ctx)
-	ttlTime := paramtable.Get().ServiceParam.MetaStoreCfg.SnapshotTTLSeconds.GetAsDuration(time.Second)
-	reserveTime := paramtable.Get().ServiceParam.MetaStoreCfg.SnapshotReserveTimeSeconds.GetAsDuration(time.Second)
-
-	candidateExpiredKeys := make([]string, 0)
-	latestOriginalKey := ""
-	latestOriginValue := ""
-	totalVersions := 0
-
-	// cleanFn processes a group of keys for a single original key
-	cleanFn := func(curOriginalKey string) error {
-		if curOriginalKey == "" {
-			return nil
-		}
-		if ss.isTombstone(latestOriginValue) {
-			// If deleted, remove all versions including the original key
-			return ss.batchRemoveExpiredKvs(ctx, candidateExpiredKeys, curOriginalKey, totalVersions == len(candidateExpiredKeys))
-		}
-
-		// If not deleted, check for expired versions
-		expiredKeys := make([]string, 0)
-		for _, key := range candidateExpiredKeys {
-			ts, _ := ss.isTSKey(key)
-			expireTime, _ := tsoutil.ParseTS(ts)
-			if expireTime.Add(ttlTime).Before(now) {
-				expiredKeys = append(expiredKeys, key)
-			}
-		}
-		if len(expiredKeys) > 0 {
-			return ss.batchRemoveExpiredKvs(ctx, expiredKeys, curOriginalKey, false)
-		}
-		return nil
-	}
-
-	// Walk through all keys with the snapshot prefix
-	err := ss.MetaKv.WalkWithPrefix(ctx, ss.snapshotPrefix, ss.paginationSize, func(k []byte, v []byte) error {
-		key := ss.hideRootPrefix(string(k))
-		ts, ok := ss.isTSKey(key)
-		if !ok {
-			log.Warn("Skip key because it doesn't contain ts", zap.String("key", key))
-			return nil
-		}
-
-		curOriginalKey, err := ss.getOriginalKey(key)
-		if err != nil {
-			log.Error("Failed to parse the original key for GC", zap.String("key", key), zap.Error(err))
-			return err
-		}
-
-		// If we've moved to a new original key, process the previous group
-		if latestOriginalKey != "" && latestOriginalKey != curOriginalKey {
-			if err := cleanFn(latestOriginalKey); err != nil {
-				return err
-			}
-
-			candidateExpiredKeys = make([]string, 0)
-			totalVersions = 0
-		}
-
-		latestOriginalKey = curOriginalKey
-		latestOriginValue = string(v)
-		totalVersions++
-
-		// Record versions that are already expired but not removed
-		time, _ := tsoutil.ParseTS(ts)
-		if time.Add(reserveTime).Before(now) {
-			candidateExpiredKeys = append(candidateExpiredKeys, key)
-		}
-
-		return nil
-	})
-	if err != nil {
-		log.Error("Error occurred during WalkWithPrefix", zap.Error(err))
-		return err
-	}
-
-	// Process the last group of keys
-	return cleanFn(latestOriginalKey)
-}

internal/metastore/kv/rootcoord/suffix_snapshot_test.go

@@ -1,917 +0,0 @@
-// 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 rootcoord
-
-import (
-	"context"
-	"fmt"
-	"math/rand"
-	"os"
-	"path"
-	"sort"
-	"testing"
-	"time"
-
-	"github.com/cockroachdb/errors"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/mock"
-	"github.com/stretchr/testify/require"
-
-	etcdkv "github.com/milvus-io/milvus/internal/kv/etcd"
-	"github.com/milvus-io/milvus/internal/kv/mocks"
-	"github.com/milvus-io/milvus/pkg/v2/util/etcd"
-	"github.com/milvus-io/milvus/pkg/v2/util/paramtable"
-	"github.com/milvus-io/milvus/pkg/v2/util/tsoutil"
-	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
-)
-
-var snapshotPrefix = "snapshots"
-
-var Params = paramtable.Get()
-
-func TestMain(m *testing.M) {
-	paramtable.Init()
-	code := m.Run()
-	os.Exit(code)
-}
-
-func Test_binarySearchRecords(t *testing.T) {
-	type testcase struct {
-		records     []tsv
-		ts          typeutil.Timestamp
-		expected    string
-		shouldFound bool
-	}
-
-	cases := []testcase{
-		{
-			records:     []tsv{},
-			ts:          0,
-			expected:    "",
-			shouldFound: false,
-		},
-		{
-			records: []tsv{
-				{
-					ts:    101,
-					value: "abc",
-				},
-			},
-			ts:          100,
-			expected:    "",
-			shouldFound: false,
-		},
-		{
-			records: []tsv{
-				{
-					ts:    100,
-					value: "a",
-				},
-			},
-			ts:          100,
-			expected:    "a",
-			shouldFound: true,
-		},
-		{
-			records: []tsv{
-				{
-					ts:    100,
-					value: "a",
-				},
-				{
-					ts:    200,
-					value: "b",
-				},
-			},
-			ts:          100,
-			expected:    "a",
-			shouldFound: true,
-		},
-		{
-			records: []tsv{
-				{
-					ts:    100,
-					value: "a",
-				},
-				{
-					ts:    200,
-					value: "b",
-				},
-				{
-					ts:    300,
-					value: "c",
-				},
-			},
-			ts:          150,
-			expected:    "a",
-			shouldFound: true,
-		},
-		{
-			records: []tsv{
-				{
-					ts:    100,
-					value: "a",
-				},
-				{
-					ts:    200,
-					value: "b",
-				},
-				{
-					ts:    300,
-					value: "c",
-				},
-			},
-			ts:          300,
-			expected:    "c",
-			shouldFound: true,
-		},
-		{
-			records: []tsv{
-				{
-					ts:    100,
-					value: "a",
-				},
-				{
-					ts:    200,
-					value: "b",
-				},
-				{
-					ts:    300,
-					value: "c",
-				},
-			},
-			ts:          201,
-			expected:    "b",
-			shouldFound: true,
-		},
-		{
-			records: []tsv{
-				{
-					ts:    100,
-					value: "a",
-				},
-				{
-					ts:    200,
-					value: "b",
-				},
-				{
-					ts:    300,
-					value: "c",
-				},
-			},
-			ts:          301,
-			expected:    "c",
-			shouldFound: true,
-		},
-	}
-	for _, c := range cases {
-		result, found := binarySearchRecords(c.records, c.ts)
-		assert.Equal(t, c.expected, result)
-		assert.Equal(t, c.shouldFound, found)
-	}
-}
-
-func Test_ComposeIsTsKey(t *testing.T) {
-	sep := "_ts"
-	ss, err := NewSuffixSnapshot(etcdkv.NewEtcdKV(nil, ""), sep, "", snapshotPrefix)
-	require.Nil(t, err)
-	defer ss.Close()
-
-	type testcase struct {
-		key         string
-		expected    uint64
-		shouldFound bool
-	}
-	testCases := []testcase{
-		{
-			key:         ss.composeTSKey("key", 100),
-			expected:    100,
-			shouldFound: true,
-		},
-		{
-			key:         ss.composeTSKey("other-key", 65536),
-			expected:    65536,
-			shouldFound: true,
-		},
-		{
-			key:         "snapshots/test/1000",
-			expected:    0,
-			shouldFound: false,
-		},
-		{
-			key:         "snapshots",
-			expected:    0,
-			shouldFound: false,
-		},
-	}
-	for _, c := range testCases {
-		ts, found := ss.isTSKey(c.key)
-		assert.EqualValues(t, c.expected, ts)
-		assert.Equal(t, c.shouldFound, found)
-	}
-}
-
-func TestComposeSnapshotKey(t *testing.T) {
-	key := "root-coord/collection/1"
-	sep := "_ts"
-	ts := typeutil.Timestamp(100)
-
-	// ComposeSnapshotKey should normalize snapshotPrefix with a trailing slash.
-	assert.Equal(t, "snapshots/"+key+sep+"100", ComposeSnapshotKey("snapshots", key, sep, ts))
-	assert.Equal(t, "snapshots/"+key+sep+"100", ComposeSnapshotKey("snapshots/", key, sep, ts))
-}
-
-func Test_SuffixSnapshotEmptyRootPrefix(t *testing.T) {
-	sep := "_ts"
-	ss, err := NewSuffixSnapshot(etcdkv.NewEtcdKV(nil, ""), sep, "", snapshotPrefix)
-	require.NoError(t, err)
-	defer ss.Close()
-
-	// Empty root should not trim the first byte when hiding root prefix.
-	key := ss.composeTSKey("k", 100)
-	assert.Equal(t, key, ss.hideRootPrefix(key))
-}
-
-func Test_SuffixSnaphotIsTSOfKey(t *testing.T) {
-	sep := "_ts"
-	ss, err := NewSuffixSnapshot(etcdkv.NewEtcdKV(nil, ""), sep, "", snapshotPrefix)
-	require.Nil(t, err)
-	defer ss.Close()
-
-	type testcase struct {
-		key         string
-		target      string
-		expected    uint64
-		shouldFound bool
-	}
-	testCases := []testcase{
-		{
-			key:         ss.composeTSKey("key", 100),
-			target:      "key",
-			expected:    100,
-			shouldFound: true,
-		},
-		{
-			key:         ss.composeTSKey("other-key", 65536),
-			target:      "other-key",
-			expected:    65536,
-			shouldFound: true,
-		},
-		{
-			key:         ss.composeTSKey("other-key", 65536),
-			target:      "key",
-			expected:    0,
-			shouldFound: false,
-		},
-		{
-			key:         "snapshots/test/1000",
-			expected:    0,
-			shouldFound: false,
-		},
-		{
-			key:         "snapshots",
-			expected:    0,
-			shouldFound: false,
-		},
-	}
-	for _, c := range testCases {
-		ts, found := ss.isTSOfKey(c.key, c.target)
-		assert.EqualValues(t, c.expected, ts)
-		assert.Equal(t, c.shouldFound, found)
-	}
-}
-
-func Test_SuffixSnapshotLoad(t *testing.T) {
-	rand.Seed(time.Now().UnixNano())
-	randVal := rand.Int()
-
-	rootPath := fmt.Sprintf("/test/meta/%d", randVal)
-	sep := "_ts"
-
-	etcdCli, err := etcd.GetEtcdClient(
-		Params.EtcdCfg.UseEmbedEtcd.GetAsBool(),
-		Params.EtcdCfg.EtcdUseSSL.GetAsBool(),
-		Params.EtcdCfg.Endpoints.GetAsStrings(),
-		Params.EtcdCfg.EtcdTLSCert.GetValue(),
-		Params.EtcdCfg.EtcdTLSKey.GetValue(),
-		Params.EtcdCfg.EtcdTLSCACert.GetValue(),
-		Params.EtcdCfg.EtcdTLSMinVersion.GetValue())
-	require.Nil(t, err)
-	defer etcdCli.Close()
-	etcdkv := etcdkv.NewEtcdKV(etcdCli, rootPath)
-	defer etcdkv.Close()
-
-	var vtso typeutil.Timestamp
-	ftso := func() typeutil.Timestamp {
-		return vtso
-	}
-
-	ss, err := NewSuffixSnapshot(etcdkv, sep, rootPath, snapshotPrefix)
-	assert.NoError(t, err)
-	assert.NotNil(t, ss)
-	defer ss.Close()
-
-	for i := 0; i < 20; i++ {
-		vtso = typeutil.Timestamp(100 + i*5)
-		ts := ftso()
-		err = ss.Save(context.TODO(), "key", fmt.Sprintf("value-%d", i), ts)
-		assert.NoError(t, err)
-		assert.Equal(t, vtso, ts)
-	}
-	for i := 0; i < 20; i++ {
-		val, err := ss.Load(context.TODO(), "key", typeutil.Timestamp(100+i*5+2))
-		t.Log("ts:", typeutil.Timestamp(100+i*5+2), i, val)
-		assert.NoError(t, err)
-		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
-	}
-	val, err := ss.Load(context.TODO(), "key", 0)
-	assert.NoError(t, err)
-	assert.Equal(t, "value-19", val)
-
-	for i := 0; i < 20; i++ {
-		val, err := ss.Load(context.TODO(), "key", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, val, fmt.Sprintf("value-%d", i))
-	}
-
-	ss.RemoveWithPrefix(context.TODO(), "")
-}
-
-func Test_SuffixSnapshotMultiSave(t *testing.T) {
-	rand.Seed(time.Now().UnixNano())
-	randVal := rand.Int()
-
-	rootPath := fmt.Sprintf("/test/meta/%d", randVal)
-	sep := "_ts"
-	etcdCli, err := etcd.GetEtcdClient(
-		Params.EtcdCfg.UseEmbedEtcd.GetAsBool(),
-		Params.EtcdCfg.EtcdUseSSL.GetAsBool(),
-		Params.EtcdCfg.Endpoints.GetAsStrings(),
-		Params.EtcdCfg.EtcdTLSCert.GetValue(),
-		Params.EtcdCfg.EtcdTLSKey.GetValue(),
-		Params.EtcdCfg.EtcdTLSCACert.GetValue(),
-		Params.EtcdCfg.EtcdTLSMinVersion.GetValue())
-	require.Nil(t, err)
-	defer etcdCli.Close()
-	etcdkv := etcdkv.NewEtcdKV(etcdCli, rootPath)
-	defer etcdkv.Close()
-
-	var vtso typeutil.Timestamp
-	ftso := func() typeutil.Timestamp {
-		return vtso
-	}
-
-	ss, err := NewSuffixSnapshot(etcdkv, sep, rootPath, snapshotPrefix)
-	assert.NoError(t, err)
-	assert.NotNil(t, ss)
-	defer ss.Close()
-
-	for i := 0; i < 20; i++ {
-		saves := map[string]string{"k1": fmt.Sprintf("v1-%d", i), "k2": fmt.Sprintf("v2-%d", i)}
-		vtso = typeutil.Timestamp(100 + i*5)
-		ts := ftso()
-		err = ss.MultiSave(context.TODO(), saves, ts)
-		assert.NoError(t, err)
-		assert.Equal(t, vtso, ts)
-	}
-	for i := 0; i < 20; i++ {
-		keys, vals, err := ss.LoadWithPrefix(context.TODO(), "k", typeutil.Timestamp(100+i*5+2))
-		t.Log(i, keys, vals)
-		assert.NoError(t, err)
-		assert.Equal(t, len(keys), len(vals))
-		assert.Equal(t, len(keys), 2)
-		assert.Equal(t, keys[0], "k1")
-		assert.Equal(t, keys[1], "k2")
-		assert.Equal(t, vals[0], fmt.Sprintf("v1-%d", i))
-		assert.Equal(t, vals[1], fmt.Sprintf("v2-%d", i))
-	}
-	keys, vals, err := ss.LoadWithPrefix(context.TODO(), "k", 0)
-	assert.NoError(t, err)
-	assert.Equal(t, len(keys), len(vals))
-	assert.Equal(t, len(keys), 2)
-	assert.Equal(t, keys[0], "k1")
-	assert.Equal(t, keys[1], "k2")
-	assert.Equal(t, vals[0], "v1-19")
-	assert.Equal(t, vals[1], "v2-19")
-
-	for i := 0; i < 20; i++ {
-		keys, vals, err := ss.LoadWithPrefix(context.TODO(), "k", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, len(keys), len(vals))
-		assert.Equal(t, len(keys), 2)
-		assert.ElementsMatch(t, keys, []string{"k1", "k2"})
-		assert.ElementsMatch(t, vals, []string{fmt.Sprintf("v1-%d", i), fmt.Sprintf("v2-%d", i)})
-	}
-	// mix non ts k-v
-	err = ss.Save(context.TODO(), "kextra", "extra-value", 0)
-	assert.NoError(t, err)
-	keys, vals, err = ss.LoadWithPrefix(context.TODO(), "k", typeutil.Timestamp(300))
-	assert.NoError(t, err)
-	assert.Equal(t, len(keys), len(vals))
-	assert.Equal(t, len(keys), 2)
-	assert.ElementsMatch(t, keys, []string{"k1", "k2"})
-	assert.ElementsMatch(t, vals, []string{"v1-19", "v2-19"})
-
-	// clean up
-	ss.RemoveWithPrefix(context.TODO(), "")
-}
-
-func Test_SuffixSnapshotRemoveExpiredKvs(t *testing.T) {
-	rand.Seed(time.Now().UnixNano())
-	randVal := rand.Int()
-
-	rootPath := fmt.Sprintf("/test/meta/remove-expired-test-%d", randVal)
-	sep := "_ts"
-
-	etcdCli, err := etcd.GetEtcdClient(
-		Params.EtcdCfg.UseEmbedEtcd.GetAsBool(),
-		Params.EtcdCfg.EtcdUseSSL.GetAsBool(),
-		Params.EtcdCfg.Endpoints.GetAsStrings(),
-		Params.EtcdCfg.EtcdTLSCert.GetValue(),
-		Params.EtcdCfg.EtcdTLSKey.GetValue(),
-		Params.EtcdCfg.EtcdTLSCACert.GetValue(),
-		Params.EtcdCfg.EtcdTLSMinVersion.GetValue())
-	assert.NoError(t, err)
-	defer etcdCli.Close()
-	etcdkv := etcdkv.NewEtcdKV(etcdCli, rootPath)
-	assert.NoError(t, err)
-	defer etcdkv.Close()
-
-	ss, err := NewSuffixSnapshot(etcdkv, sep, rootPath, snapshotPrefix)
-	assert.NoError(t, err)
-	assert.NotNil(t, ss)
-	defer ss.Close()
-
-	saveFn := func(key, value string, ts typeutil.Timestamp) {
-		err = ss.Save(context.TODO(), key, value, ts)
-		assert.NoError(t, err)
-	}
-
-	multiSaveFn := func(kvs map[string]string, ts typeutil.Timestamp) {
-		err = ss.MultiSave(context.TODO(), kvs, ts)
-		assert.NoError(t, err)
-	}
-
-	now := time.Now()
-	ftso := func(ts int) typeutil.Timestamp {
-		return tsoutil.ComposeTS(now.Add(-1*time.Duration(ts)*time.Hour).UnixMilli(), 0)
-	}
-
-	getKey := func(prefix string, id int) string {
-		return fmt.Sprintf("%s-%d", prefix, id)
-	}
-
-	generateTestData := func(prefix string, kCnt int, kVersion int, expiredKeyCnt int) {
-		var value string
-		cnt := 0
-		for i := 0; i < kVersion; i++ {
-			kvs := make(map[string]string)
-			ts := ftso((i + 1) * 2)
-			for v := 0; v < kCnt; v++ {
-				if i == 0 && v%2 == 0 && cnt < expiredKeyCnt {
-					value = string(SuffixSnapshotTombstone)
-					cnt++
-				} else {
-					value = "v"
-				}
-
-				kvs[getKey(prefix, v)] = value
-				if v%25 == 0 {
-					multiSaveFn(kvs, ts)
-					kvs = make(map[string]string)
-				}
-			}
-			multiSaveFn(kvs, ts)
-		}
-	}
-
-	countPrefix := func(prefix string) int {
-		cnt := 0
-		err := etcdkv.WalkWithPrefix(context.TODO(), "", 10, func(key []byte, value []byte) error {
-			cnt++
-			return nil
-		})
-		assert.NoError(t, err)
-		return cnt
-	}
-
-	getPrefix := func(prefix string) []string {
-		var res []string
-		_ = etcdkv.WalkWithPrefix(context.TODO(), "", 10, func(key []byte, value []byte) error {
-			res = append(res, string(key))
-			return nil
-		})
-		return res
-	}
-
-	t.Run("Mixed test ", func(t *testing.T) {
-		prefix := fmt.Sprintf("prefix%d", rand.Int())
-		keyCnt := 500
-		keyVersion := 3
-		expiredKCnt := 100
-		generateTestData(prefix, keyCnt, keyVersion, expiredKCnt)
-
-		cnt := countPrefix(prefix)
-		assert.Equal(t, keyCnt*keyVersion+keyCnt, cnt)
-
-		err = ss.removeExpiredKvs(context.TODO(), now)
-		assert.NoError(t, err)
-
-		cnt = countPrefix(prefix)
-		assert.Equal(t, keyCnt*keyVersion+keyCnt-(expiredKCnt*keyVersion+expiredKCnt), cnt)
-
-		// clean all data
-		err := etcdkv.RemoveWithPrefix(context.TODO(), "")
-		assert.NoError(t, err)
-	})
-
-	t.Run("partial expired and all expired", func(t *testing.T) {
-		prefix := fmt.Sprintf("prefix%d", rand.Int())
-		value := "v"
-		ts := ftso(1)
-		saveFn(getKey(prefix, 0), value, ts)
-		ts = ftso(2)
-		saveFn(getKey(prefix, 0), value, ts)
-		ts = ftso(3)
-		saveFn(getKey(prefix, 0), value, ts)
-
-		// insert partial expired kv
-		ts = ftso(2)
-		saveFn(getKey(prefix, 1), string(SuffixSnapshotTombstone), ts)
-		ts = ftso(4)
-		saveFn(getKey(prefix, 1), value, ts)
-		ts = ftso(6)
-		saveFn(getKey(prefix, 1), value, ts)
-
-		// insert all expired kv
-		ts = ftso(1)
-		saveFn(getKey(prefix, 2), string(SuffixSnapshotTombstone), ts)
-		ts = ftso(2)
-		saveFn(getKey(prefix, 2), value, ts)
-		ts = ftso(3)
-		saveFn(getKey(prefix, 2), value, ts)
-
-		cnt := countPrefix(prefix)
-		assert.Equal(t, 12, cnt)
-
-		// err = ss.removeExpiredKvs(now, time.Duration(50)*time.Millisecond)
-		err = ss.removeExpiredKvs(context.TODO(), now)
-		assert.NoError(t, err)
-
-		cnt = countPrefix(prefix)
-		assert.Equal(t, 4, cnt)
-
-		// clean all data
-		err := etcdkv.RemoveWithPrefix(context.TODO(), "")
-		assert.NoError(t, err)
-	})
-
-	t.Run("partial 24 expired and all expired", func(t *testing.T) {
-		prefix := fmt.Sprintf("prefix%d", rand.Int())
-		value := "v"
-		ts := ftso(100)
-		saveFn(getKey(prefix, 0), value, ts)
-		ts = ftso(200)
-		saveFn(getKey(prefix, 0), value, ts)
-		ts = ftso(300)
-		saveFn(getKey(prefix, 0), value, ts)
-
-		// insert partial expired kv
-		ts = ftso(2)
-		saveFn(getKey(prefix, 1), string(SuffixSnapshotTombstone), ts)
-		ts = ftso(4)
-		saveFn(getKey(prefix, 1), value, ts)
-		ts = ftso(6)
-		saveFn(getKey(prefix, 1), value, ts)
-
-		// insert all expired kv
-		ts = ftso(1)
-		saveFn(getKey(prefix, 2), string(SuffixSnapshotTombstone), ts)
-		ts = ftso(2)
-		saveFn(getKey(prefix, 2), value, ts)
-		ts = ftso(3)
-		saveFn(getKey(prefix, 2), value, ts)
-
-		cnt := countPrefix(prefix)
-		assert.Equal(t, 12, cnt)
-
-		// err = ss.removeExpiredKvs(now, time.Duration(50)*time.Millisecond)
-		err = ss.removeExpiredKvs(context.TODO(), now)
-		assert.NoError(t, err)
-
-		cnt = countPrefix(prefix)
-		assert.Equal(t, 2, cnt)
-		res := getPrefix(prefix)
-		sort.Strings(res)
-		keepKey := getKey(prefix, 0)
-		keepTs := ftso(100)
-		assert.Equal(t, []string{path.Join(rootPath, keepKey), path.Join(rootPath, ss.composeTSKey(keepKey, keepTs))}, res)
-
-		// clean all data
-		err := etcdkv.RemoveWithPrefix(context.TODO(), "")
-		assert.NoError(t, err)
-	})
-
-	t.Run("parse ts fail", func(t *testing.T) {
-		prefix := fmt.Sprintf("prefix%d", rand.Int())
-		key := fmt.Sprintf("%s-%s", prefix, "ts_error-ts")
-		err = etcdkv.Save(context.TODO(), ss.composeSnapshotPrefix(key), "")
-		assert.NoError(t, err)
-
-		err = ss.removeExpiredKvs(context.TODO(), now)
-		assert.NoError(t, err)
-
-		cnt := countPrefix(prefix)
-		assert.Equal(t, 1, cnt)
-
-		// clean all data
-		err := etcdkv.RemoveWithPrefix(context.TODO(), "")
-		assert.NoError(t, err)
-	})
-
-	t.Run("test walk kv data fail", func(t *testing.T) {
-		sep := "_ts"
-		rootPath := "root/"
-		kv := mocks.NewMetaKv(t)
-		kv.EXPECT().
-			WalkWithPrefix(mock.Anything, mock.Anything, mock.Anything, mock.Anything).
-			Return(errors.New("error"))
-
-		ss, err := NewSuffixSnapshot(kv, sep, rootPath, snapshotPrefix)
-		assert.NotNil(t, ss)
-		assert.NoError(t, err)
-
-		err = ss.removeExpiredKvs(context.TODO(), time.Now())
-		assert.Error(t, err)
-	})
-}
-
-func Test_SuffixSnapshotMultiSaveAndRemoveWithPrefix(t *testing.T) {
-	rand.Seed(time.Now().UnixNano())
-	randVal := rand.Int()
-
-	rootPath := fmt.Sprintf("/test/meta/%d", randVal)
-	sep := "_ts"
-
-	etcdCli, err := etcd.GetEtcdClient(
-		Params.EtcdCfg.UseEmbedEtcd.GetAsBool(),
-		Params.EtcdCfg.EtcdUseSSL.GetAsBool(),
-		Params.EtcdCfg.Endpoints.GetAsStrings(),
-		Params.EtcdCfg.EtcdTLSCert.GetValue(),
-		Params.EtcdCfg.EtcdTLSKey.GetValue(),
-		Params.EtcdCfg.EtcdTLSCACert.GetValue(),
-		Params.EtcdCfg.EtcdTLSMinVersion.GetValue())
-	require.Nil(t, err)
-	defer etcdCli.Close()
-	etcdkv := etcdkv.NewEtcdKV(etcdCli, rootPath)
-	require.Nil(t, err)
-	defer etcdkv.Close()
-
-	var vtso typeutil.Timestamp
-	ftso := func() typeutil.Timestamp {
-		return vtso
-	}
-
-	ss, err := NewSuffixSnapshot(etcdkv, sep, rootPath, snapshotPrefix)
-	assert.NoError(t, err)
-	assert.NotNil(t, ss)
-	defer ss.Close()
-
-	for i := 0; i < 20; i++ {
-		vtso = typeutil.Timestamp(100 + i*5)
-		ts := ftso()
-		err = ss.Save(context.TODO(), fmt.Sprintf("kd-%04d", i), fmt.Sprintf("value-%d", i), ts)
-		assert.NoError(t, err)
-		assert.Equal(t, vtso, ts)
-	}
-	for i := 20; i < 40; i++ {
-		sm := map[string]string{"ks": fmt.Sprintf("value-%d", i)}
-		dm := []string{fmt.Sprintf("kd-%04d", i-20)}
-		vtso = typeutil.Timestamp(100 + i*5)
-		ts := ftso()
-		err = ss.MultiSaveAndRemoveWithPrefix(context.TODO(), sm, dm, ts)
-		assert.NoError(t, err)
-		assert.Equal(t, vtso, ts)
-	}
-	for i := 0; i < 20; i++ {
-		val, err := ss.Load(context.TODO(), fmt.Sprintf("kd-%04d", i), typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
-		_, vals, err := ss.LoadWithPrefix(context.TODO(), "kd-", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, i+1, len(vals))
-	}
-	for i := 20; i < 40; i++ {
-		val, err := ss.Load(context.TODO(), "ks", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
-		_, vals, err := ss.LoadWithPrefix(context.TODO(), "kd-", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, 39-i, len(vals))
-	}
-
-	for i := 0; i < 20; i++ {
-		val, err := ss.Load(context.TODO(), fmt.Sprintf("kd-%04d", i), typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
-		_, vals, err := ss.LoadWithPrefix(context.TODO(), "kd-", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, i+1, len(vals))
-	}
-	for i := 20; i < 40; i++ {
-		val, err := ss.Load(context.TODO(), "ks", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
-		_, vals, err := ss.LoadWithPrefix(context.TODO(), "kd-", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, 39-i, len(vals))
-	}
-	// try to load
-	_, err = ss.Load(context.TODO(), "kd-0000", 500)
-	assert.Error(t, err)
-	_, err = ss.Load(context.TODO(), "kd-0000", 0)
-	assert.Error(t, err)
-	_, err = ss.Load(context.TODO(), "kd-0000", 1)
-	assert.Error(t, err)
-
-	// cleanup
-	ss.MultiSaveAndRemoveWithPrefix(context.TODO(), map[string]string{}, []string{""}, 0)
-}
-
-func Test_SuffixSnapshotMultiSaveAndRemove(t *testing.T) {
-	rand.Seed(time.Now().UnixNano())
-	randVal := rand.Int()
-
-	rootPath := fmt.Sprintf("/test/meta/%d", randVal)
-	sep := "_ts"
-
-	etcdCli, err := etcd.GetEtcdClient(
-		Params.EtcdCfg.UseEmbedEtcd.GetAsBool(),
-		Params.EtcdCfg.EtcdUseSSL.GetAsBool(),
-		Params.EtcdCfg.Endpoints.GetAsStrings(),
-		Params.EtcdCfg.EtcdTLSCert.GetValue(),
-		Params.EtcdCfg.EtcdTLSKey.GetValue(),
-		Params.EtcdCfg.EtcdTLSCACert.GetValue(),
-		Params.EtcdCfg.EtcdTLSMinVersion.GetValue())
-	require.Nil(t, err)
-	defer etcdCli.Close()
-	etcdkv := etcdkv.NewEtcdKV(etcdCli, rootPath)
-	require.Nil(t, err)
-	defer etcdkv.Close()
-
-	var vtso typeutil.Timestamp
-	ftso := func() typeutil.Timestamp {
-		return vtso
-	}
-
-	ss, err := NewSuffixSnapshot(etcdkv, sep, rootPath, snapshotPrefix)
-	assert.NoError(t, err)
-	assert.NotNil(t, ss)
-	defer ss.Close()
-
-	for i := 0; i < 20; i++ {
-		vtso = typeutil.Timestamp(100 + i*5)
-		ts := ftso()
-		err = ss.Save(context.TODO(), fmt.Sprintf("kd-%04d", i), fmt.Sprintf("value-%d", i), ts)
-		assert.NoError(t, err)
-		assert.Equal(t, vtso, ts)
-	}
-	for i := 20; i < 40; i++ {
-		sm := map[string]string{"ks": fmt.Sprintf("value-%d", i)}
-		dm := []string{fmt.Sprintf("kd-%04d", i-20)}
-		vtso = typeutil.Timestamp(100 + i*5)
-		ts := ftso()
-		err = ss.MultiSaveAndRemove(context.TODO(), sm, dm, ts)
-		assert.NoError(t, err)
-		assert.Equal(t, vtso, ts)
-	}
-	for i := 0; i < 20; i++ {
-		val, err := ss.Load(context.TODO(), fmt.Sprintf("kd-%04d", i), typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
-		_, vals, err := ss.LoadWithPrefix(context.TODO(), "kd-", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, i+1, len(vals))
-	}
-	for i := 20; i < 40; i++ {
-		val, err := ss.Load(context.TODO(), "ks", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, fmt.Sprintf("value-%d", i), val)
-		_, vals, err := ss.LoadWithPrefix(context.TODO(), "kd-", typeutil.Timestamp(100+i*5+2))
-		assert.NoError(t, err)
-		assert.Equal(t, 39-i, len(vals))
-	}
-
-	// try to load
-	_, err = ss.Load(context.TODO(), "kd-0000", 500)
-	assert.Error(t, err)
-	_, err = ss.Load(context.TODO(), "kd-0000", 0)
-	assert.Error(t, err)
-	_, err = ss.Load(context.TODO(), "kd-0000", 1)
-	assert.Error(t, err)
-
-	// cleanup
-	ss.MultiSaveAndRemoveWithPrefix(context.TODO(), map[string]string{}, []string{""}, 0)
-}
-
-func TestSuffixSnapshot_LoadWithPrefix(t *testing.T) {
-	rand.Seed(time.Now().UnixNano())
-	randVal := rand.Int()
-
-	rootPath := fmt.Sprintf("/test/meta/loadWithPrefix-test-%d", randVal)
-	sep := "_ts"
-
-	etcdCli, err := etcd.GetEtcdClient(
-		Params.EtcdCfg.UseEmbedEtcd.GetAsBool(),
-		Params.EtcdCfg.EtcdUseSSL.GetAsBool(),
-		Params.EtcdCfg.Endpoints.GetAsStrings(),
-		Params.EtcdCfg.EtcdTLSCert.GetValue(),
-		Params.EtcdCfg.EtcdTLSKey.GetValue(),
-		Params.EtcdCfg.EtcdTLSCACert.GetValue(),
-		Params.EtcdCfg.EtcdTLSMinVersion.GetValue())
-	assert.NoError(t, err)
-	defer etcdCli.Close()
-	etcdkv := etcdkv.NewEtcdKV(etcdCli, rootPath)
-	assert.NoError(t, err)
-	defer etcdkv.Close()
-
-	ss, err := NewSuffixSnapshot(etcdkv, sep, rootPath, snapshotPrefix)
-	assert.NoError(t, err)
-	assert.NotNil(t, ss)
-	defer ss.Close()
-
-	t.Run("parse ts fail", func(t *testing.T) {
-		prefix := fmt.Sprintf("prefix%d", rand.Int())
-		key := fmt.Sprintf("%s-%s", prefix, "ts_error-ts")
-		err = etcdkv.Save(context.TODO(), ss.composeSnapshotPrefix(key), "")
-		assert.NoError(t, err)
-
-		keys, values, err := ss.LoadWithPrefix(context.TODO(), prefix, 100)
-		assert.NoError(t, err)
-		assert.Equal(t, 0, len(keys))
-		assert.Equal(t, 0, len(values))
-
-		// clean all data
-		err = etcdkv.RemoveWithPrefix(context.TODO(), "")
-		assert.NoError(t, err)
-	})
-
-	t.Run("test walk kv data fail", func(t *testing.T) {
-		sep := "_ts"
-		rootPath := "root/"
-		kv := mocks.NewMetaKv(t)
-		kv.EXPECT().
-			WalkWithPrefix(mock.Anything, mock.Anything, mock.Anything, mock.Anything).
-			Return(errors.New("error"))
-
-		ss, err := NewSuffixSnapshot(kv, sep, rootPath, snapshotPrefix)
-		assert.NotNil(t, ss)
-		assert.NoError(t, err)
-
-		keys, values, err := ss.LoadWithPrefix(context.TODO(), "t", 100)
-		assert.Error(t, err)
-		assert.Nil(t, keys)
-		assert.Nil(t, values)
-	})
-}
-
-func Test_getOriginalKey(t *testing.T) {
-	sep := "_ts"
-	rootPath := "root/"
-	kv := mocks.NewMetaKv(t)
-	ss, err := NewSuffixSnapshot(kv, sep, rootPath, snapshotPrefix)
-	assert.NotNil(t, ss)
-	assert.NoError(t, err)
-
-	t.Run("match prefix fail", func(t *testing.T) {
-		ret, err := ss.getOriginalKey("non-snapshots/k1")
-		assert.Equal(t, "", ret)
-		assert.Error(t, err)
-	})
-
-	t.Run("find separator fail", func(t *testing.T) {
-		ret, err := ss.getOriginalKey("snapshots/k1")
-		assert.Equal(t, "", ret)
-		assert.Error(t, err)
-	})
-
-	t.Run("ok", func(t *testing.T) {
-		ret, err := ss.getOriginalKey("snapshots/prefix-1_ts438497159122780160")
-		assert.Equal(t, "prefix-1", ret)
-		assert.NoError(t, err)
-	})
-}

internal/rootcoord/meta_table_test.go

@@ -51,7 +51,7 @@ func generateMetaTable(_ *testing.T) *MetaTable {
 	kv, _ := kvfactory.GetEtcdAndPath()
 	path := funcutil.RandomString(10)
 	catalogKV := etcdkv.NewEtcdKV(kv, path)
-	return &MetaTable{catalog: rootcoord.NewCatalog(catalogKV, nil)}
+	return &MetaTable{catalog: rootcoord.NewCatalog(catalogKV)}
 }
 
 func buildAlterUserMessage(credInfo *internalpb.CredentialInfo, timetick uint64) message.BroadcastResultAlterUserMessageV2 {
@@ -2610,9 +2610,7 @@ func TestMetaTable_TruncateCollection(t *testing.T) {
 	kv, _ := kvfactory.GetEtcdAndPath()
 	path := funcutil.RandomString(10) + "/meta"
 	catalogKV := etcdkv.NewEtcdKV(kv, path)
-	ss, err := rootcoord.NewSuffixSnapshot(catalogKV, rootcoord.SnapshotsSep, path, rootcoord.SnapshotPrefix)
-	require.NoError(t, err)
-	catalog := rootcoord.NewCatalog(catalogKV, ss)
+	catalog := rootcoord.NewCatalog(catalogKV)
 
 	allocator := mocktso.NewAllocator(t)
 	allocator.EXPECT().GenerateTSO(mock.Anything).Return(1000, nil)

internal/rootcoord/root_coord.go

@@ -392,24 +392,14 @@ func (c *Core) initMetaTable(initCtx context.Context) error {
 		switch Params.MetaStoreCfg.MetaStoreType.GetValue() {
 		case util.MetaStoreTypeEtcd:
 			log.Ctx(initCtx).Info("Using etcd as meta storage.")
-			var ss *kvmetastore.SuffixSnapshot
-			var err error
-
 			metaKV := c.metaKVCreator()
-			if ss, err = kvmetastore.NewSuffixSnapshot(metaKV, kvmetastore.SnapshotsSep, Params.EtcdCfg.MetaRootPath.GetValue(), kvmetastore.SnapshotPrefix); err != nil {
-				return err
-			}
-			catalog = kvmetastore.NewCatalog(metaKV, ss)
+			kvmetastore.StartLegacySnapshotGC(c.ctx, metaKV)
+			catalog = kvmetastore.NewCatalog(metaKV)
 		case util.MetaStoreTypeTiKV:
 			log.Ctx(initCtx).Info("Using tikv as meta storage.")
-			var ss *kvmetastore.SuffixSnapshot
-			var err error
-
 			metaKV := c.metaKVCreator()
-			if ss, err = kvmetastore.NewSuffixSnapshot(metaKV, kvmetastore.SnapshotsSep, Params.TiKVCfg.MetaRootPath.GetValue(), kvmetastore.SnapshotPrefix); err != nil {
-				return err
-			}
-			catalog = kvmetastore.NewCatalog(metaKV, ss)
+			kvmetastore.StartLegacySnapshotGC(c.ctx, metaKV)
+			catalog = kvmetastore.NewCatalog(metaKV)
 		default:
 			return retry.Unrecoverable(fmt.Errorf("not supported meta store: %s", Params.MetaStoreCfg.MetaStoreType.GetValue()))
 		}

internal/rootcoord/root_coord_test.go

@@ -84,15 +84,13 @@ func initStreamingSystemAndCore(t *testing.T) *Core {
 	path := funcutil.RandomString(10) + "/meta"
 	catalogKV := etcdkv.NewEtcdKV(kv, path)
 
-	ss, err := rootcoord.NewSuffixSnapshot(catalogKV, rootcoord.SnapshotsSep, path, rootcoord.SnapshotPrefix)
-	require.NoError(t, err)
 	testDB := newNameDb()
 	collID2Meta := make(map[typeutil.UniqueID]*model.Collection)
 	tso := mocktso.NewAllocator(t)
 	tso.EXPECT().GenerateTSO(mock.Anything).Return(uint64(1), nil).Maybe()
 	core := newTestCore(withHealthyCode(),
 		withMeta(&MetaTable{
-			catalog:     rootcoord.NewCatalog(catalogKV, ss),
+			catalog:     rootcoord.NewCatalog(catalogKV),
 			names:       testDB,
 			aliases:     newNameDb(),
 			dbName2Meta: make(map[string]*model.Database),

pkg/kv/kv.go

@@ -22,7 +22,6 @@ import (
 	clientv3 "go.etcd.io/etcd/client/v3"
 
 	"github.com/milvus-io/milvus/pkg/v2/kv/predicates"
-	"github.com/milvus-io/milvus/pkg/v2/util/typeutil"
 )
 
 // CompareFailedError is a helper type for checking MetaKv CompareAndSwap series func error type
@@ -84,15 +83,3 @@ type WatchKV interface {
 	WatchWithPrefix(ctx context.Context, key string) clientv3.WatchChan
 	WatchWithRevision(ctx context.Context, key string, revision int64) clientv3.WatchChan
 }
-
-// SnapShotKV is TxnKV for snapshot data. It must save timestamp.
-//
-//go:generate mockery --name=SnapShotKV --with-expecter
-type SnapShotKV interface {
-	Save(ctx context.Context, key string, value string, ts typeutil.Timestamp) error
-	Load(ctx context.Context, key string, ts typeutil.Timestamp) (string, error)
-	MultiSave(ctx context.Context, kvs map[string]string, ts typeutil.Timestamp) error
-	LoadWithPrefix(ctx context.Context, key string, ts typeutil.Timestamp) ([]string, []string, error)
-	MultiSaveAndRemove(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error
-	MultiSaveAndRemoveWithPrefix(ctx context.Context, saves map[string]string, removals []string, ts typeutil.Timestamp) error
-}

pkg/util/paramtable/service_param.go

@@ -510,12 +510,10 @@ It is recommended to change this parameter before starting Milvus for the first
 }
 
 type MetaStoreConfig struct {
-	MetaStoreType              ParamItem `refreshable:"false"`
-	SnapshotTTLSeconds         ParamItem `refreshable:"true"`
-	SnapshotReserveTimeSeconds ParamItem `refreshable:"true"`
-	PaginationSize             ParamItem `refreshable:"true"`
-	ReadConcurrency            ParamItem `refreshable:"true"`
-	MaxEtcdTxnNum              ParamItem `refreshable:"true"`
+	MetaStoreType   ParamItem `refreshable:"false"`
+	PaginationSize  ParamItem `refreshable:"true"`
+	ReadConcurrency ParamItem `refreshable:"true"`
+	MaxEtcdTxnNum   ParamItem `refreshable:"true"`
 }
 
 func (p *MetaStoreConfig) Init(base *BaseTable) {
@@ -528,24 +526,6 @@ func (p *MetaStoreConfig) Init(base *BaseTable) {
 	}
 	p.MetaStoreType.Init(base.mgr)
 
-	p.SnapshotTTLSeconds = ParamItem{
-		Key:          "metastore.snapshot.ttl",
-		Version:      "2.4.14",
-		DefaultValue: "86400",
-		Doc:          `snapshot ttl in seconds`,
-		Export:       true,
-	}
-	p.SnapshotTTLSeconds.Init(base.mgr)
-
-	p.SnapshotReserveTimeSeconds = ParamItem{
-		Key:          "metastore.snapshot.reserveTime",
-		Version:      "2.4.14",
-		DefaultValue: "3600",
-		Doc:          `snapshot reserve time in seconds`,
-		Export:       true,
-	}
-	p.SnapshotReserveTimeSeconds.Init(base.mgr)
-
 	p.PaginationSize = ParamItem{
 		Key:          "metastore.paginationSize",
 		Version:      "2.5.1",

pkg/util/paramtable/service_param_test.go

@@ -362,8 +362,6 @@ func TestServiceParam(t *testing.T) {
 		Params := &SParams.MetaStoreCfg
 
 		assert.Equal(t, util.MetaStoreTypeEtcd, Params.MetaStoreType.GetValue())
-		assert.Equal(t, 86400*time.Second, Params.SnapshotTTLSeconds.GetAsDuration(time.Second))
-		assert.Equal(t, 3600*time.Second, Params.SnapshotReserveTimeSeconds.GetAsDuration(time.Second))
 		assert.Equal(t, 100000, Params.PaginationSize.GetAsInt())
 		assert.Equal(t, 32, Params.ReadConcurrency.GetAsInt())
 	})