Add DeleteAll method to store type (#402)

## Describe your changes and provide context
Iterating with mergeiterator to get all keys and then deleting is
extremely slow when there are >10 layers of `cachekv`. This PR adds a
more efficient function to delete all keys within a range, without
having to make recursive calls like mergeiterator.

## Testing performed to validate your change
unit test on cachekv

---------

Co-authored-by: Philip Su <[email protected]>

server/mock/store.go

@@ -230,6 +230,10 @@ func (kv kvStore) VersionExists(version int64) bool {
 	panic("not implemented")
 }
 
+func (kv kvStore) DeleteAll(start, end []byte) error {
+	panic("not implemented")
+}
+
 func NewCommitMultiStore() sdk.CommitMultiStore {
 	return multiStore{kv: make(map[sdk.StoreKey]kvStore)}
 }

store/cachekv/store.go

@@ -372,3 +372,18 @@ func (store *Store) isDeleted(key string) bool {
 func (store *Store) GetParent() types.KVStore {
 	return store.parent
 }
+
+func (store *Store) DeleteAll(start, end []byte) error {
+	store.dirtyItems(start, end)
+	// memdb iterator
+	cachedIter, err := store.sortedCache.Iterator(start, end)
+	if err != nil {
+		return err
+	}
+	defer cachedIter.Close()
+	for ; cachedIter.Valid(); cachedIter.Next() {
+		// `Delete` would not touch sortedCache so it's okay to perform inside iterator
+		store.Delete(cachedIter.Key())
+	}
+	return nil
+}

store/cachekv/store_test.go

@@ -65,6 +65,16 @@ func TestCacheKVStore(t *testing.T) {
 
 	// GetParent returns parent store
 	require.NotNil(t, st.GetParent())
+
+	// DeleteAll deletes all entries in cache but not affect mem
+	st = cachekv.NewStore(mem, types.NewKVStoreKey("CacheKvTest"), types.DefaultCacheSizeLimit)
+	mem.Set(keyFmt(1), valFmt(1))
+	st.Set(keyFmt(1), valFmt(2))
+	st.Set(keyFmt(2), valFmt(3))
+	require.Nil(t, st.DeleteAll(nil, nil))
+	require.Nil(t, st.Get(keyFmt(1)))
+	require.Nil(t, st.Get(keyFmt(2)))
+	require.Equal(t, valFmt(1), mem.Get(keyFmt(1)))
 }
 
 func TestCacheKVStoreNoNilSet(t *testing.T) {

store/dbadapter/store.go

@@ -99,5 +99,18 @@ func (dsa Store) VersionExists(version int64) bool {
 	panic("no versioning for dbadater")
 }
 
+func (dsa Store) DeleteAll(start, end []byte) error {
+	iter := dsa.Iterator(start, end)
+	keys := [][]byte{}
+	for ; iter.Valid(); iter.Next() {
+		keys = append(keys, iter.Key())
+	}
+	iter.Close()
+	for _, key := range keys {
+		dsa.Delete(key)
+	}
+	return nil
+}
+
 // dbm.DB implements KVStore so we can CacheKVStore it.
 var _ types.KVStore = Store{}

store/dbadapter/store_test.go

@@ -13,6 +13,7 @@ import (
 	"github.com/cosmos/cosmos-sdk/store/dbadapter"
 	"github.com/cosmos/cosmos-sdk/store/types"
 	"github.com/cosmos/cosmos-sdk/tests/mocks"
+	dbm "github.com/tendermint/tm-db"
 )
 
 var errFoo = errors.New("dummy")
@@ -74,6 +75,17 @@ func TestAccessors(t *testing.T) {
 	require.Panics(t, func() { store.ReverseIterator(start, end) })
 }
 
+func TestDeleteAll(t *testing.T) {
+	mem := dbadapter.Store{DB: dbm.NewMemDB()}
+	mem.Set([]byte("1"), []byte("2"))
+	mem.Set([]byte("3"), []byte("4"))
+	require.NotNil(t, mem.Get([]byte("1")))
+	require.NotNil(t, mem.Get([]byte("3")))
+	require.Nil(t, mem.DeleteAll(nil, nil))
+	require.Nil(t, mem.Get([]byte("1")))
+	require.Nil(t, mem.Get([]byte("3")))
+}
+
 func TestCacheWraps(t *testing.T) {
 	mockCtrl := gomock.NewController(t)
 	mockDB := mocks.NewMockDB(mockCtrl)

store/gaskv/store.go

@@ -130,6 +130,10 @@ func (gs *Store) VersionExists(version int64) bool {
 	return gs.parent.VersionExists(version)
 }
 
+func (gs *Store) DeleteAll(start, end []byte) error {
+	return gs.parent.DeleteAll(start, end)
+}
+
 type gasIterator struct {
 	gasMeter  types.GasMeter
 	gasConfig types.GasConfig

store/iavl/store.go

@@ -423,6 +423,19 @@ func (st *Store) Query(req abci.RequestQuery) (res abci.ResponseQuery) {
 	return res
 }
 
+func (st *Store) DeleteAll(start, end []byte) error {
+	iter := st.Iterator(start, end)
+	keys := [][]byte{}
+	for ; iter.Valid(); iter.Next() {
+		keys = append(keys, iter.Key())
+	}
+	iter.Close()
+	for _, key := range keys {
+		st.Delete(key)
+	}
+	return nil
+}
+
 // Takes a MutableTree, a key, and a flag for creating existence or absence proof and returns the
 // appropriate merkle.Proof. Since this must be called after querying for the value, this function should never error
 // Thus, it will panic on error rather than returning it

store/iavl/store_test.go

@@ -463,6 +463,21 @@ func TestIAVLNoPrune(t *testing.T) {
 	}
 }
 
+func TestIAVLStoreDeleteAll(t *testing.T) {
+	db := dbm.NewMemDB()
+	tree, err := iavl.NewMutableTree(db, cacheSize, false)
+	require.NoError(t, err)
+
+	iavlStore := UnsafeNewStore(tree)
+	iavlStore.Set([]byte("1"), []byte("2"))
+	iavlStore.Set([]byte("3"), []byte("4"))
+	require.NotNil(t, iavlStore.Get([]byte("1")))
+	require.NotNil(t, iavlStore.Get([]byte("3")))
+	require.Nil(t, iavlStore.DeleteAll(nil, nil))
+	require.Nil(t, iavlStore.Get([]byte("1")))
+	require.Nil(t, iavlStore.Get([]byte("3")))
+}
+
 func TestIAVLStoreQuery(t *testing.T) {
 	db := dbm.NewMemDB()
 	tree, err := iavl.NewMutableTree(db, cacheSize, false)

store/listenkv/store.go

@@ -161,3 +161,7 @@ func (s *Store) onWrite(delete bool, key, value []byte) {
 		l.OnWrite(s.parentStoreKey, key, value, delete)
 	}
 }
+
+func (s *Store) DeleteAll(start, end []byte) error {
+	return s.parent.DeleteAll(start, end)
+}

store/prefix/store.go

@@ -90,6 +90,18 @@ func (s Store) Delete(key []byte) {
 	s.parent.Delete(s.key(key))
 }
 
+func (s Store) DeleteAll(start, end []byte) error {
+	newstart := cloneAppend(s.prefix, start)
+
+	var newend []byte
+	if end == nil {
+		newend = cpIncr(s.prefix)
+	} else {
+		newend = cloneAppend(s.prefix, end)
+	}
+	return s.parent.DeleteAll(newstart, newend)
+}
+
 // Implements KVStore
 // Check https://github.com/tendermint/tendermint/blob/master/libs/db/prefix_db.go#L106
 func (s Store) Iterator(start, end []byte) types.Iterator {

store/tracekv/store.go

@@ -186,6 +186,10 @@ func (tkv *Store) VersionExists(version int64) bool {
 	return tkv.parent.VersionExists(version)
 }
 
+func (tkv *Store) DeleteAll(start, end []byte) error {
+	return tkv.parent.DeleteAll(start, end)
+}
+
 // writeOperation writes a KVStore operation to the underlying io.Writer as
 // JSON-encoded data where the key/value pair is base64 encoded.
 func writeOperation(w io.Writer, op operation, tc types.TraceContext, key, value []byte) {

store/types/store.go

@@ -255,6 +255,8 @@ type KVStore interface {
 	GetWorkingHash() ([]byte, error)
 
 	VersionExists(version int64) bool
+
+	DeleteAll(start, end []byte) error
 }
 
 // Iterator is an alias db's Iterator for convenience.

storev2/commitment/store.go

@@ -182,3 +182,16 @@ func (st *Store) VersionExists(version int64) bool {
 	// one version per SC tree
 	return version == st.tree.Version()
 }
+
+func (st *Store) DeleteAll(start, end []byte) error {
+	iter := st.Iterator(start, end)
+	keys := [][]byte{}
+	for ; iter.Valid(); iter.Next() {
+		keys = append(keys, iter.Key())
+	}
+	iter.Close()
+	for _, key := range keys {
+		st.Delete(key)
+	}
+	return nil
+}

storev2/state/store.go

@@ -134,3 +134,16 @@ func (st *Store) VersionExists(version int64) bool {
 	}
 	return version >= earliest
 }
+
+func (st *Store) DeleteAll(start, end []byte) error {
+	iter := st.Iterator(start, end)
+	keys := [][]byte{}
+	for ; iter.Valid(); iter.Next() {
+		keys = append(keys, iter.Key())
+	}
+	iter.Close()
+	for _, key := range keys {
+		st.Delete(key)
+	}
+	return nil
+}