feat(orm): range/prefix deletion support (#11103)

## Description

This adds `DeleteBy...` methods to the ORM codegen for the primary key + every unique index.



---

### Author Checklist

*All items are required. Please add a note to the item if the item is not applicable and
please add links to any relevant follow up issues.*

I have...

- [ ] included the correct [type prefix](https://github.com/commitizen/conventional-commit-types/blob/v3.0.0/index.json) in the PR title
- [ ] added `!` to the type prefix if API or client breaking change
- [ ] targeted the correct branch (see [PR Targeting](https://github.com/cosmos/cosmos-sdk/blob/master/CONTRIBUTING.md#pr-targeting))
- [ ] provided a link to the relevant issue or specification
- [ ] followed the guidelines for [building modules](https://github.com/cosmos/cosmos-sdk/blob/master/docs/building-modules)
- [ ] included the necessary unit and integration [tests](https://github.com/cosmos/cosmos-sdk/blob/master/CONTRIBUTING.md#testing)
- [ ] added a changelog entry to `CHANGELOG.md`
- [ ] included comments for [documenting Go code](https://blog.golang.org/godoc)
- [ ] updated the relevant documentation or specification
- [ ] reviewed "Files changed" and left comments if necessary
- [ ] confirmed all CI checks have passed

### Reviewers Checklist

*All items are required. Please add a note if the item is not applicable and please add
your handle next to the items reviewed if you only reviewed selected items.*

I have...

- [ ] confirmed the correct [type prefix](https://github.com/commitizen/conventional-commit-types/blob/v3.0.0/index.json) in the PR title
- [ ] confirmed `!` in the type prefix if API or client breaking change
- [ ] confirmed all author checklist items have been addressed 
- [ ] reviewed state machine logic
- [ ] reviewed API design and naming
- [ ] reviewed documentation is accurate
- [ ] reviewed tests and test coverage
- [ ] manually tested (if applicable)

orm/internal/codegen/table.go

@@ -65,21 +65,24 @@ func (t tableGen) genStoreInterface() {
 	t.P("Delete(ctx ", contextPkg.Ident("Context"), ", ", t.param(t.msg.GoIdent.GoName), " *", t.QualifiedGoIdent(t.msg.GoIdent), ") error")
 	t.P("Has(ctx ", contextPkg.Ident("Context"), ", ", t.fieldsArgs(t.primaryKeyFields.Names()), ") (found bool, err error)")
 	t.P("Get(ctx ", contextPkg.Ident("Context"), ", ", t.fieldsArgs(t.primaryKeyFields.Names()), ") (*", t.QualifiedGoIdent(t.msg.GoIdent), ", error)")
+
 	for _, idx := range t.uniqueIndexes {
 		t.genUniqueIndexSig(idx)
 	}
 	t.P("List(ctx ", contextPkg.Ident("Context"), ", prefixKey ", t.indexKeyInterfaceName(), ", opts ...", ormListPkg.Ident("Option"), ") ", "(", t.iteratorName(), ", error)")
 	t.P("ListRange(ctx ", contextPkg.Ident("Context"), ", from, to ", t.indexKeyInterfaceName(), ", opts ...", ormListPkg.Ident("Option"), ") ", "(", t.iteratorName(), ", error)")
+	t.P("DeleteBy(ctx ", contextPkg.Ident("Context"), ", prefixKey ", t.indexKeyInterfaceName(), ") error")
+	t.P("DeleteRange(ctx ", contextPkg.Ident("Context"), ", from, to ", t.indexKeyInterfaceName(), ") error")
 	t.P()
 	t.P("doNotImplement()")
 	t.P("}")
 	t.P()
 }
 
 // returns the has and get (in that order) function signature for unique indexes.
-func (t tableGen) uniqueIndexSig(idx *ormv1alpha1.SecondaryIndexDescriptor) (string, string) {
-	fieldsSlc := strings.Split(idx.Fields, ",")
-	camelFields := t.fieldsToCamelCase(idx.Fields)
+func (t tableGen) uniqueIndexSig(idxFields string) (string, string) {
+	fieldsSlc := strings.Split(idxFields, ",")
+	camelFields := t.fieldsToCamelCase(idxFields)
 
 	hasFuncName := "HasBy" + camelFields
 	getFuncName := "GetBy" + camelFields
@@ -91,7 +94,7 @@ func (t tableGen) uniqueIndexSig(idx *ormv1alpha1.SecondaryIndexDescriptor) (str
 }
 
 func (t tableGen) genUniqueIndexSig(idx *ormv1alpha1.SecondaryIndexDescriptor) {
-	hasSig, getSig := t.uniqueIndexSig(idx)
+	hasSig, getSig := t.uniqueIndexSig(idx.Fields)
 	t.P(hasSig)
 	t.P(getSig)
 }
@@ -197,7 +200,7 @@ func (t tableGen) genStoreImpl() {
 
 	for _, idx := range t.uniqueIndexes {
 		fields := strings.Split(idx.Fields, ",")
-		hasName, getName := t.uniqueIndexSig(idx)
+		hasName, getName := t.uniqueIndexSig(idx.Fields)
 
 		// has
 		t.P("func (", receiverVar, " ", t.messageStoreReceiverName(t.msg), ") ", hasName, "{")
@@ -243,6 +246,20 @@ func (t tableGen) genStoreImpl() {
 	t.P("}")
 	t.P()
 
+	// DeleteBy
+	t.P(receiver, "DeleteBy(ctx ", contextPkg.Ident("Context"), ", prefixKey ", t.indexKeyInterfaceName(), ") error {")
+	t.P("return ", receiverVar, ".table.GetIndexByID(prefixKey.id()).DeleteBy(ctx, prefixKey.values()...)")
+	t.P("}")
+	t.P()
+	t.P()
+
+	// DeleteRange
+	t.P(receiver, "DeleteRange(ctx ", contextPkg.Ident("Context"), ", from, to ", t.indexKeyInterfaceName(), ") error {")
+	t.P("return ", receiverVar, ".table.GetIndexByID(from.id()).DeleteRange(ctx, from.values(), to.values())")
+	t.P("}")
+	t.P()
+	t.P()
+
 	t.P(receiver, "doNotImplement() {}")
 	t.P()
 }

orm/model/ormtable/batch.go

@@ -11,14 +11,13 @@ type batchIndexCommitmentWriter struct {
 func newBatchIndexCommitmentWriter(store Backend) *batchIndexCommitmentWriter {
 	return &batchIndexCommitmentWriter{
 		Backend: store,
-		// optimal array capacities are estimated here:
 		commitmentWriter: &batchStoreWriter{
 			ReadonlyStore: store.CommitmentStoreReader(),
-			writes:        make([]batchWriterEntry, 0, 2),
+			curBuf:        make([]*batchWriterEntry, 0, capacity),
 		},
 		indexWriter: &batchStoreWriter{
 			ReadonlyStore: store.IndexStoreReader(),
-			writes:        make([]batchWriterEntry, 0, 16),
+			curBuf:        make([]*batchWriterEntry, 0, capacity),
 		},
 	}
 }
@@ -33,12 +32,12 @@ func (w *batchIndexCommitmentWriter) IndexStore() kv.Store {
 
 // Write flushes any pending writes.
 func (w *batchIndexCommitmentWriter) Write() error {
-	err := flushWrites(w.Backend.CommitmentStore(), w.commitmentWriter.writes)
+	err := flushWrites(w.Backend.CommitmentStore(), w.commitmentWriter)
 	if err != nil {
 		return err
 	}
 
-	err = flushWrites(w.Backend.IndexStore(), w.indexWriter.writes)
+	err = flushWrites(w.Backend.IndexStore(), w.indexWriter)
 	if err != nil {
 		return err
 	}
@@ -49,15 +48,25 @@ func (w *batchIndexCommitmentWriter) Write() error {
 	return err
 }
 
-func flushWrites(writer kv.Store, writes []batchWriterEntry) error {
+func flushWrites(store kv.Store, writer *batchStoreWriter) error {
+	for _, buf := range writer.prevBufs {
+		err := flushBuf(store, buf)
+		if err != nil {
+			return err
+		}
+	}
+	return flushBuf(store, writer.curBuf)
+}
+
+func flushBuf(store kv.Store, writes []*batchWriterEntry) error {
 	for _, write := range writes {
 		if !write.delete {
-			err := writer.Set(write.key, write.value)
+			err := store.Set(write.key, write.value)
 			if err != nil {
 				return err
 			}
 		} else {
-			err := writer.Delete(write.key)
+			err := store.Delete(write.key)
 			if err != nil {
 				return err
 			}
@@ -69,8 +78,10 @@ func flushWrites(writer kv.Store, writes []batchWriterEntry) error {
 // Close discards any pending writes and should generally be called using
 // a defer statement.
 func (w *batchIndexCommitmentWriter) Close() {
-	w.commitmentWriter.writes = nil
-	w.indexWriter.writes = nil
+	w.commitmentWriter.prevBufs = nil
+	w.commitmentWriter.curBuf = nil
+	w.indexWriter.prevBufs = nil
+	w.indexWriter.curBuf = nil
 }
 
 type batchWriterEntry struct {
@@ -80,17 +91,29 @@ type batchWriterEntry struct {
 
 type batchStoreWriter struct {
 	kv.ReadonlyStore
-	writes []batchWriterEntry
+	prevBufs [][]*batchWriterEntry
+	curBuf   []*batchWriterEntry
 }
 
+const capacity = 16
+
 func (b *batchStoreWriter) Set(key, value []byte) error {
-	b.writes = append(b.writes, batchWriterEntry{key: key, value: value})
+	b.append(&batchWriterEntry{key: key, value: value})
 	return nil
 }
 
 func (b *batchStoreWriter) Delete(key []byte) error {
-	b.writes = append(b.writes, batchWriterEntry{key: key, delete: true})
+	b.append(&batchWriterEntry{key: key, delete: true})
 	return nil
 }
 
+func (b *batchStoreWriter) append(entry *batchWriterEntry) {
+	if len(b.curBuf) == capacity {
+		b.prevBufs = append(b.prevBufs, b.curBuf)
+		b.curBuf = make([]*batchWriterEntry, 0, capacity)
+	}
+
+	b.curBuf = append(b.curBuf, entry)
+}
+
 var _ Backend = &batchIndexCommitmentWriter{}

orm/model/ormtable/index.go

@@ -38,6 +38,12 @@ type Index interface {
 	// Range iteration is inclusive at both ends.
 	ListRange(ctx context.Context, from, to []interface{}, options ...ormlist.Option) (Iterator, error)
 
+	// DeleteBy deletes any entries which match the provided prefix key.
+	DeleteBy(context context.Context, prefixKey ...interface{}) error
+
+	// DeleteRange deletes any entries between the provided range keys.
+	DeleteRange(context context.Context, from, to []interface{}) error
+
 	// MessageType returns the protobuf message type of the index.
 	MessageType() protoreflect.MessageType
 
@@ -64,9 +70,6 @@ type UniqueIndex interface {
 
 	// Get retrieves the message if one exists for the provided key values.
 	Get(context context.Context, message proto.Message, keyValues ...interface{}) (found bool, err error)
-
-	// DeleteByKey deletes the message if one exists in for the provided key values.
-	DeleteByKey(context context.Context, keyValues ...interface{}) error
 }
 
 type indexer interface {

orm/model/ormtable/index_impl.go

@@ -24,6 +24,24 @@ type indexKeyIndex struct {
 	getReadBackend func(context.Context) (ReadBackend, error)
 }
 
+func (i indexKeyIndex) DeleteBy(ctx context.Context, keyValues ...interface{}) error {
+	it, err := i.List(ctx, keyValues)
+	if err != nil {
+		return err
+	}
+
+	return i.primaryKey.deleteByIterator(ctx, it)
+}
+
+func (i indexKeyIndex) DeleteRange(ctx context.Context, from, to []interface{}) error {
+	it, err := i.ListRange(ctx, from, to)
+	if err != nil {
+		return err
+	}
+
+	return i.primaryKey.deleteByIterator(ctx, it)
+}
+
 func (i indexKeyIndex) List(ctx context.Context, prefixKey []interface{}, options ...ormlist.Option) (Iterator, error) {
 	backend, err := i.getReadBackend(ctx)
 	if err != nil {

orm/model/ormtable/iterator.go

@@ -15,6 +15,10 @@ import (
 )
 
 // Iterator defines the interface for iterating over indexes.
+//
+// WARNING: it is generally unsafe to mutate a table while iterating over it.
+// Instead you should do reads and writes separately, or use a helper
+// function like DeleteBy which does this efficiently.
 type Iterator interface {
 
 	// Next advances the iterator and returns true if a valid entry is found.
@@ -213,6 +217,7 @@ func (i *indexIterator) Next() bool {
 		i.started = true
 	} else {
 		i.iterator.Next()
+		i.indexValues = nil
 	}
 
 	return i.iterator.Valid()