Repository: matrixorigin/matrixone
Pull request #5777 (2022-10-20 04:37:40)

various memtable fixes


## What type of PR is this?

- [ ] API-change
- [ ] BUG
- [x] Improvement
- [ ] Documentation
- [ ] Feature
- [ ] Test and CI
- [ ] Code Refactoring

## Which issue(s) this PR fixes:

issue #

## What this PR does / why we need it:
memtable: add Table.FilterVersions; add error return to AbortTx
memtable: add ReverseIndexEntry; delete indexes in Table.FilterVersion
memtable: remove IndexEntry.Value
memtable: add unique index
memtable: check unique indexes in Table.Insert,Update,Upsert
memtable: check unique indexes in Table.CommitTx
memtable: add TestUniqueIndex


Patch:
diff --git a/pkg/txn/storage/memorystorage/catalog.go b/pkg/txn/storage/memorystorage/catalog.go
index 8e42829c0..6d75450bd 100644
--- a/pkg/txn/storage/memorystorage/catalog.go
+++ b/pkg/txn/storage/memorystorage/catalog.go
@@ -63,6 +63,12 @@ func (d *DatabaseRow) Indexes() []Tuple {
 	}
 }
 
+func (d *DatabaseRow) UniqueIndexes() []Tuple {
+	return []Tuple{
+		{index_AccountID_Name, Uint(d.AccountID), Text(d.Name)},
+	}
+}
+
 var _ NamedRow = new(DatabaseRow)
 
 func (d *DatabaseRow) AttrByName(handler *MemHandler, tx *Transaction, name string) (ret Nullable, err error) {
@@ -122,6 +128,12 @@ func (r *RelationRow) Indexes() []Tuple {
 	}
 }
 
+func (r *RelationRow) UniqueIndexes() []Tuple {
+	return []Tuple{
+		{index_DatabaseID_Name, r.DatabaseID, Text(r.Name)},
+	}
+}
+
 var _ NamedRow = new(RelationRow)
 
 func (r *RelationRow) AttrByName(handler *MemHandler, tx *Transaction, name string) (ret Nullable, err error) {
@@ -208,6 +220,12 @@ func (a *AttributeRow) Indexes() []Tuple {
 	}
 }
 
+func (a *AttributeRow) UniqueIndexes() []Tuple {
+	return []Tuple{
+		{index_RelationID_Name, a.RelationID, Text(a.Name)},
+	}
+}
+
 var _ NamedRow = new(AttributeRow)
 
 func (a *AttributeRow) AttrByName(handler *MemHandler, tx *Transaction, name string) (ret Nullable, err error) {
@@ -346,6 +364,12 @@ func (i *IndexRow) Indexes() []Tuple {
 	}
 }
 
+func (i *IndexRow) UniqueIndexes() []Tuple {
+	return []Tuple{
+		{index_RelationID_Name, i.RelationID, Text(i.Name)},
+	}
+}
+
 func verifyAttr(
 	names []string,
 	types []types.Type,
diff --git a/pkg/txn/storage/memorystorage/data.go b/pkg/txn/storage/memorystorage/data.go
index a1ea8ce16..95d8da2eb 100644
--- a/pkg/txn/storage/memorystorage/data.go
+++ b/pkg/txn/storage/memorystorage/data.go
@@ -44,9 +44,10 @@ func (d DataKey) Less(than DataKey) bool {
 type DataValue = []Nullable
 
 type DataRow struct {
-	key     DataKey
-	value   DataValue
-	indexes []Tuple
+	key           DataKey
+	value         DataValue
+	indexes       []Tuple
+	uniqueIndexes []Tuple
 }
 
 func (a DataRow) Key() DataKey {
@@ -61,6 +62,10 @@ func (a DataRow) Indexes() []Tuple {
 	return a.indexes
 }
 
+func (a DataRow) UniqueIndexes() []Tuple {
+	return a.uniqueIndexes
+}
+
 func (a *DataRow) String() string {
 	buf := new(strings.Builder)
 	buf.WriteString("DataRow{")
diff --git a/pkg/txn/storage/memorystorage/log_tail.go b/pkg/txn/storage/memorystorage/log_tail.go
index 61e5336c3..62e504d61 100644
--- a/pkg/txn/storage/memorystorage/log_tail.go
+++ b/pkg/txn/storage/memorystorage/log_tail.go
@@ -67,7 +67,7 @@ func (m *MemHandler) HandleGetLogTail(ctx context.Context, meta txn.TxnMeta, req
 	}
 
 	// attributes
-	attrs, err := m.attributes.Index(tx, Tuple{
+	entries, err := m.attributes.Index(tx, Tuple{
 		index_RelationID,
 		tableID,
 	})
@@ -75,14 +75,18 @@ func (m *MemHandler) HandleGetLogTail(ctx context.Context, meta txn.TxnMeta, req
 		return err
 	}
 	attrsMap := make(map[string]*AttributeRow)
-	insertNames := make([]string, 0, len(attrs))
-	deleteNames := make([]string, 0, len(attrs))
-	for _, attr := range attrs {
-		attrsMap[attr.Value.Name] = attr.Value
-		if !attr.Value.IsRowId {
-			insertNames = append(insertNames, attr.Value.Name)
-			if attr.Value.Primary {
-				deleteNames = append(deleteNames, attr.Value.Name)
+	insertNames := make([]string, 0, len(entries))
+	deleteNames := make([]string, 0, len(entries))
+	for _, entry := range entries {
+		attr, err := m.attributes.Get(tx, entry.Key)
+		if err != nil {
+			return err
+		}
+		attrsMap[attr.Name] = attr
+		if !attr.IsRowId {
+			insertNames = append(insertNames, attr.Name)
+			if attr.Primary {
+				deleteNames = append(deleteNames, attr.Name)
 			}
 		}
 	}
diff --git a/pkg/txn/storage/memorystorage/mem_handler.go b/pkg/txn/storage/memorystorage/mem_handler.go
index 1c0212d6a..6fc602a85 100644
--- a/pkg/txn/storage/memorystorage/mem_handler.go
+++ b/pkg/txn/storage/memorystorage/mem_handler.go
@@ -595,19 +595,25 @@ func (m *MemHandler) HandleDelete(ctx context.Context, meta txn.TxnMeta, req mem
 	if err != nil {
 		return err
 	}
-	if len(entries) == 1 && entries[0].Value.Name == req.ColumnName {
-		// by primary key
-		for i := 0; i < reqVecLen; i++ {
-			value := memtable.VectorAt(req.Vector, i)
-			key := DataKey{
-				tableID:    req.TableID,
-				primaryKey: Tuple{memtable.ToOrdered(value.Value)},
-			}
-			if err := m.data.Delete(tx, key); err != nil {
-				return err
+	if len(entries) == 1 {
+		attr, err := m.attributes.Get(tx, entries[0].Key)
+		if err != nil {
+			return err
+		}
+		if attr.Name == req.ColumnName {
+			// by primary key
+			for i := 0; i < reqVecLen; i++ {
+				value := memtable.VectorAt(req.Vector, i)
+				key := DataKey{
+					tableID:    req.TableID,
+					primaryKey: Tuple{memtable.ToOrdered(value.Value)},
+				}
+				if err := m.data.Delete(tx, key); err != nil {
+					return err
+				}
 			}
+			return nil
 		}
-		return nil
 	}
 
 	// by non-primary key, slow but works
@@ -625,7 +631,11 @@ func (m *MemHandler) HandleDelete(ctx context.Context, meta txn.TxnMeta, req mem
 	if len(entries) != 1 {
 		panic("impossible")
 	}
-	attrIndex := entries[0].Value.Order
+	attr, err := m.attributes.Get(tx, entries[0].Key)
+	if err != nil {
+		return err
+	}
+	attrIndex := attr.Order
 	iter := m.data.NewIter(tx)
 	defer iter.Close()
 	tableKey := DataKey{
@@ -673,13 +683,17 @@ func (m *MemHandler) HandleDeleteDatabase(ctx context.Context, meta txn.TxnMeta,
 	}
 
 	for _, entry := range entries {
+		db, err := m.databases.Get(tx, entry.Key)
+		if err != nil {
+			return err
+		}
 		if err := m.databases.Delete(tx, entry.Key); err != nil {
 			return err
 		}
-		if err := m.deleteRelationsByDBID(tx, entry.Value.ID); err != nil {
+		if err := m.deleteRelationsByDBID(tx, db.ID); err != nil {
 			return err
 		}
-		resp.ID = entry.Value.ID
+		resp.ID = db.ID
 	}
 
 	return nil
@@ -694,13 +708,17 @@ func (m *MemHandler) deleteRelationsByDBID(tx *Transaction, dbID ID) error {
 		return err
 	}
 	for _, entry := range entries {
+		rel, err := m.relations.Get(tx, entry.Key)
+		if err != nil {
+			return err
+		}
 		if err := m.relations.Delete(tx, entry.Key); err != nil {
 			return err
 		}
-		if err := m.deleteAttributesByRelationID(tx, entry.Value.ID); err != nil {
+		if err := m.deleteAttributesByRelationID(tx, rel.ID); err != nil {
 			return err
 		}
-		if err := m.deleteRelationData(tx, entry.Value.ID); err != nil {
+		if err := m.deleteRelationData(tx, rel.ID); err != nil {
 			return err
 		}
 	}
@@ -763,13 +781,17 @@ func (m *MemHandler) HandleDeleteRelation(ctx context.Context, meta txn.TxnMeta,
 		panic("impossible")
 	}
 	entry := entries[0]
+	rel, err := m.relations.Get(tx, entry.Key)
+	if err != nil {
+		return err
+	}
 	if err := m.relations.Delete(tx, entry.Key); err != nil {
 		return err
 	}
-	if err := m.deleteAttributesByRelationID(tx, entry.Value.ID); err != nil {
+	if err := m.deleteAttributesByRelationID(tx, rel.ID); err != nil {
 		return err
 	}
-	resp.ID = entry.Value.ID
+	resp.ID = rel.ID
 	return nil
 }
 
@@ -794,7 +816,11 @@ func (m *MemHandler) HandleGetDatabases(ctx context.Context, meta txn.TxnMeta, r
 	}
 
 	for _, entry := range entries {
-		resp.Names = append(resp.Names, string(entry.Value.Name))
+		db, err := m.databases.Get(tx, entry.Key)
+		if err != nil {
+			return err
+		}
+		resp.Names = append(resp.Names, string(db.Name))
 	}
 
 	return nil
@@ -811,7 +837,11 @@ func (m *MemHandler) HandleGetPrimaryKeys(ctx context.Context, meta txn.TxnMeta,
 		return err
 	}
 	for _, entry := range entries {
-		resp.Attrs = append(resp.Attrs, &entry.Value.Attribute)
+		attr, err := m.attributes.Get(tx, entry.Key)
+		if err != nil {
+			return err
+		}
+		resp.Attrs = append(resp.Attrs, &attr.Attribute)
 	}
 	return nil
 }
@@ -826,7 +856,11 @@ func (m *MemHandler) HandleGetRelations(ctx context.Context, meta txn.TxnMeta, r
 		return err
 	}
 	for _, entry := range entries {
-		resp.Names = append(resp.Names, string(entry.Value.Name))
+		rel, err := m.relations.Get(tx, entry.Key)
+		if err != nil {
+			return err
+		}
+		resp.Names = append(resp.Names, string(rel.Name))
 	}
 	return nil
 }
@@ -949,9 +983,13 @@ func (m *MemHandler) HandleGetTableDefs(ctx context.Context, meta txn.TxnMeta, r
 				TableNames: make([]string, indexLen),
 			}
 			for i, entry := range entries {
-				computeIndexDef.Names[i] = entry.Value.Name
-				computeIndexDef.Uniques[i] = entry.Value.Unique
-				computeIndexDef.TableNames[i] = entry.Value.TableName
+				index, err := m.indexes.Get(tx, entry.Key)
+				if err != nil {
+					return err
+				}
+				computeIndexDef.Names[i] = index.Name
+				computeIndexDef.Uniques[i] = index.Unique
+				computeIndexDef.TableNames[i] = index.TableName
 			}
 			resp.Defs = append(resp.Defs, computeIndexDef)
 		}
@@ -983,7 +1021,11 @@ func (m *MemHandler) HandleGetHiddenKeys(ctx context.Context, meta txn.TxnMeta,
 		return err
 	}
 	for _, entry := range entries {
-		resp.Attrs = append(resp.Attrs, &entry.Value.Attribute)
+		attr, err := m.attributes.Get(tx, entry.Key)
+		if err != nil {
+			return err
+		}
+		resp.Attrs = append(resp.Attrs, &attr.Attribute)
 	}
 	return nil
 }
@@ -1042,13 +1084,19 @@ func (m *MemHandler) HandleOpenDatabase(ctx context.Context, meta txn.TxnMeta, r
 		return err
 	}
 
-	for _, entry := range entries {
-		resp.ID = entry.Value.ID
-		resp.Name = string(entry.Value.Name)
-		return nil
+	if len(entries) == 0 {
+		return moerr.NewNoDB()
 	}
 
-	return moerr.NewNoDB()
+	entry := entries[0]
+	db, err := m.databases.Get(tx, entry.Key)
+	if err != nil {
+		return err
+	}
+	resp.ID = db.ID
+	resp.Name = string(db.Name)
+
+	return nil
 }
 
 func (m *MemHandler) HandleOpenRelation(ctx context.Context, meta txn.TxnMeta, req memoryengine.OpenRelationReq, resp *memoryengine.OpenRelationResp) error {
@@ -1065,10 +1113,14 @@ func (m *MemHandler) HandleOpenRelation(ctx context.Context, meta txn.TxnMeta, r
 		return moerr.NewNoSuchTable(req.DatabaseName, req.Name)
 	}
 	entry := entries[0]
-	resp.ID = entry.Value.ID
-	resp.Type = entry.Value.Type
-	resp.RelationName = string(entry.Value.Name)
-	db, err := m.databases.Get(tx, entry.Value.DatabaseID)
+	rel, err := m.relations.Get(tx, entry.Key)
+	if err != nil {
+		return err
+	}
+	resp.ID = rel.ID
+	resp.Type = rel.Type
+	resp.RelationName = string(rel.Name)
+	db, err := m.databases.Get(tx, rel.DatabaseID)
 	if err != nil {
 		return err
 	}
@@ -1379,7 +1431,11 @@ func (m *MemHandler) iterRelationAttributes(
 		return err
 	}
 	for _, entry := range entries {
-		if err := fn(entry.Key, entry.Value); err != nil {
+		attr, err := m.attributes.Get(tx, entry.Key)
+		if err != nil {
+			return err
+		}
+		if err := fn(entry.Key, attr); err != nil {
 			return err
 		}
 	}
diff --git a/pkg/txn/storage/memorystorage/memtable/index_iter.go b/pkg/txn/storage/memorystorage/memtable/index_iter.go
index bceeb494b..023191fbb 100644
--- a/pkg/txn/storage/memorystorage/memtable/index_iter.go
+++ b/pkg/txn/storage/memorystorage/memtable/index_iter.go
@@ -35,9 +35,25 @@ type IndexIter[
 
 func (t *Table[K, V, R]) NewIndexIter(tx *Transaction, min Tuple, max Tuple) *IndexIter[K, V] {
 	state := t.state.Load()
+	return t.newIndexIter(
+		state.indexes.Copy().Iter(),
+		state.rows,
+		tx,
+		min,
+		max,
+	)
+}
+
+func (t *Table[K, V, R]) newIndexIter(
+	iter btree.GenericIter[*IndexEntry[K, V]],
+	rows *btree.BTreeG[*PhysicalRow[K, V]],
+	tx *Transaction,
+	min Tuple,
+	max Tuple,
+) *IndexIter[K, V] {
 	return &IndexIter[K, V]{
-		iter:     state.indexes.Copy().Iter(),
-		rows:     state.rows,
+		iter:     iter,
+		rows:     rows,
 		readTime: tx.Time,
 		tx:       tx,
 		min:      min,
diff --git a/pkg/txn/storage/memorystorage/memtable/table.go b/pkg/txn/storage/memorystorage/memtable/table.go
index 6fa07e94f..c167208ee 100644
--- a/pkg/txn/storage/memorystorage/memtable/table.go
+++ b/pkg/txn/storage/memorystorage/memtable/table.go
@@ -40,16 +40,22 @@ type tableState[
 	K Ordered[K],
 	V any,
 ] struct {
-	rows    *btree.BTreeG[*PhysicalRow[K, V]]
-	indexes *btree.BTreeG[*IndexEntry[K, V]]
-	writes  *btree.BTreeG[*WriteEntry[K, V]]
+	rows                 *btree.BTreeG[*PhysicalRow[K, V]]
+	indexes              *btree.BTreeG[*IndexEntry[K, V]]
+	reverseIndexes       *btree.BTreeG[*ReverseIndexEntry[K, V]]
+	writes               *btree.BTreeG[*WriteEntry[K, V]]
+	uniqueIndexes        *btree.BTreeG[*IndexEntry[K, V]]
+	reverseUniqueIndexes *btree.BTreeG[*ReverseIndexEntry[K, V]]
 }
 
 func (t *tableState[K, V]) Copy() *tableState[K, V] {
 	return &tableState[K, V]{
-		rows:    t.rows.Copy(),
-		indexes: t.indexes.Copy(),
-		writes:  t.writes.Copy(),
+		rows:                 t.rows.Copy(),
+		indexes:              t.indexes.Copy(),
+		reverseIndexes:       t.reverseIndexes.Copy(),
+		writes:               t.writes.Copy(),
+		uniqueIndexes:        t.uniqueIndexes.Copy(),
+		reverseUniqueIndexes: t.reverseUniqueIndexes.Copy(),
 	}
 }
 
@@ -57,6 +63,7 @@ type Row[K any, V any] interface {
 	Key() K
 	Value() V
 	Indexes() []Tuple
+	UniqueIndexes() []Tuple
 }
 
 type IndexEntry[
@@ -66,8 +73,15 @@ type IndexEntry[
 	Index     Tuple
 	Key       K
 	VersionID int64
+}
 
-	Value V
+type ReverseIndexEntry[
+	K Ordered[K],
+	V any,
+] struct {
+	Key       K
+	VersionID int64
+	Index     Tuple
 }
 
 type WriteEntry[
@@ -87,9 +101,12 @@ func NewTable[
 ]() *Table[K, V, R] {
 	ret := &Table[K, V, R]{}
 	state := &tableState[K, V]{
-		rows:    btree.NewBTreeG(comparePhysicalRow[K, V]),
-		indexes: btree.NewBTreeG(compareIndexEntry[K, V]),
-		writes:  btree.NewBTreeG(compareWriteEntry[K, V]),
+		rows:                 btree.NewBTreeG(comparePhysicalRow[K, V]),
+		indexes:              btree.NewBTreeG(compareIndexEntry[K, V]),
+		reverseIndexes:       btree.NewBTreeG(compareReverseIndexEntry[K, V]),
+		writes:               btree.NewBTreeG(compareWriteEntry[K, V]),
+		uniqueIndexes:        btree.NewBTreeG(compareIndexEntry[K, V]),
+		reverseUniqueIndexes: btree.NewBTreeG(compareReverseIndexEntry[K, V]),
 	}
 	ret.state.Store(state)
 	return ret
@@ -121,6 +138,25 @@ func compareIndexEntry[
 	return a.VersionID < b.VersionID
 }
 
+func compareReverseIndexEntry[
+	K Ordered[K],
+	V any,
+](a, b *ReverseIndexEntry[K, V]) bool {
+	if a.Key.Less(b.Key) {
+		return true
+	}
+	if b.Key.Less(a.Key) {
+		return false
+	}
+	if a.VersionID < b.VersionID {
+		return true
+	}
+	if b.VersionID < a.VersionID {
+		return false
+	}
+	return a.Index.Less(b.Index)
+}
+
 func compareWriteEntry[
 	K Ordered[K],
 	V any,
@@ -171,13 +207,8 @@ func (t *Table[K, V, R]) Insert(
 		}
 
 		// index entry
-		for _, index := range row.Indexes() {
-			state.indexes.Set(&IndexEntry[K, V]{
-				Index:     index,
-				Key:       key,
-				VersionID: version.ID,
-				Value:     value,
-			})
+		if err := setIndexes(tx, t, state, key, version, row); err != nil {
+			return err
 		}
 
 		// write entry
@@ -215,13 +246,8 @@ func (t *Table[K, V, R]) Update(
 		}
 
 		// index entry
-		for _, index := range row.Indexes() {
-			state.indexes.Set(&IndexEntry[K, V]{
-				Index:     index,
-				Key:       key,
-				VersionID: version.ID,
-				Value:     value,
-			})
+		if err := setIndexes(tx, t, state, key, version, row); err != nil {
+			return err
 		}
 
 		// write entry
@@ -317,13 +343,8 @@ func (t *Table[K, V, R]) Upsert(
 		}
 
 		// index entry
-		for _, index := range row.Indexes() {
-			state.indexes.Set(&IndexEntry[K, V]{
-				Index:     index,
-				Key:       key,
-				VersionID: version.ID,
-				Value:     value,
-			})
+		if err := setIndexes(tx, t, state, key, version, row); err != nil {
+			return err
 		}
 
 		// write entry
@@ -342,6 +363,61 @@ func (t *Table[K, V, R]) Upsert(
 	})
 }
 
+func setIndexes[
+	K Ordered[K],
+	V any,
+	R Row[K, V],
+](
+	tx *Transaction,
+	table *Table[K, V, R],
+	state *tableState[K, V],
+	key K,
+	version *Version[V],
+	row R,
+) error {
+
+	// index entries
+	for _, index := range row.Indexes() {
+		state.indexes.Set(&IndexEntry[K, V]{
+			Index:     index,
+			Key:       key,
+			VersionID: version.ID,
+		})
+		state.reverseIndexes.Set(&ReverseIndexEntry[K, V]{
+			Key:       key,
+			VersionID: version.ID,
+			Index:     index,
+		})
+	}
+
+	// unique index entries
+	uniqueIndexes := row.UniqueIndexes()
+	for _, index := range uniqueIndexes {
+		iter := table.newIndexIter(
+			state.uniqueIndexes.Copy().Iter(),
+			state.rows,
+			tx,
+			index,
+			append(index, Min),
+		)
+		for ok := iter.First(); ok; ok = iter.Next() {
+			return moerr.NewDuplicate()
+		}
+		state.uniqueIndexes.Set(&IndexEntry[K, V]{
+			Index:     index,
+			Key:       key,
+			VersionID: version.ID,
+		})
+		state.reverseUniqueIndexes.Set(&ReverseIndexEntry[K, V]{
+			Key:       key,
+			VersionID: version.ID,
+			Index:     index,
+		})
+	}
+
+	return nil
+}
+
 func (t *Table[K, V, R]) Get(
 	tx *Transaction,
 	key K,
@@ -424,23 +500,59 @@ func (t *Table[K, V, R]) CommitTx(tx *Transaction) error {
 				break
 			}
 
-			physicalRow := getRowByKey(state.rows, *entry.Key)
+			key := *entry.Key
+			physicalRow := getRowByKey(state.rows, key)
 			if err := validate(physicalRow, tx); err != nil {
 				return err
 			}
 
-			// set born time and lock time to commit time
 			physicalRow = physicalRow.clone()
 			for i, version := range physicalRow.Versions {
-				if version.ID == entry.VersionID {
-					if version.LockTx == tx {
-						version.LockTime = tx.CommitTime
+				if version.ID != entry.VersionID {
+					continue
+				}
+
+				// set born time and lock time to commit time
+				if version.LockTx == tx {
+					version.LockTime = tx.CommitTime
+				}
+				if version.BornTx == tx {
+					version.BornTime = tx.CommitTime
+				}
+
+				// check unique index
+				reverseIter := state.reverseUniqueIndexes.Copy().Iter()
+				pivot := &ReverseIndexEntry[K, V]{
+					Key:       key,
+					VersionID: version.ID,
+					Index:     Tuple{Min},
+				}
+				for ok := reverseIter.Seek(pivot); ok; ok = reverseIter.Next() {
+					entry := reverseIter.Item()
+					if key.Less(entry.Key) {
+						break
 					}
-					if version.BornTx == tx {
-						version.BornTime = tx.CommitTime
+					if entry.VersionID != version.ID {
+						break
+					}
+					iter := t.newIndexIter(
+						state.uniqueIndexes.Copy().Iter(),
+						state.rows,
+						tx,
+						entry.Index,
+						append(entry.Index, Min),
+					)
+					for ok := iter.First(); ok; ok = iter.Next() {
+						index := iter.Item()
+						if index.Key.Less(entry.Key) ||
+							entry.Key.Less(index.Key) ||
+							index.VersionID != entry.VersionID {
+							return moerr.NewDuplicate()
+						}
 					}
-					physicalRow.Versions[i] = version
 				}
+
+				physicalRow.Versions[i] = version
 			}
 			state.rows.Set(physicalRow)
 
@@ -453,8 +565,79 @@ func (t *Table[K, V, R]) CommitTx(tx *Transaction) error {
 
 }
 
-func (t *Table[K, V, R]) AbortTx(tx *Transaction) {
-	t.update(func(state *tableState[K, V]) error {
+func (t *Table[K, V, R]) FilterVersions(filterFunc func(K, []Version[V]) ([]Version[V], error), keys ...K) error {
+	return t.update(func(state *tableState[K, V]) error {
+		for _, key := range keys {
+			physicalRow := getRowByKey(state.rows, key)
+			newVersions, err := filterFunc(key, physicalRow.Versions)
+			if err != nil {
+				return err
+			}
+
+			if len(newVersions) == 0 {
+				// delete
+				state.rows.Delete(physicalRow)
+			} else {
+				// update
+				physicalRow = physicalRow.clone()
+				physicalRow.Versions = newVersions
+				state.rows.Set(physicalRow)
+			}
+
+			newVersionIDSet := make(map[int64]bool)
+			for _, v := range newVersions {
+				newVersionIDSet[v.ID] = true
+			}
+
+			// remove indexes
+			iter := state.reverseIndexes.Copy().Iter()
+			for ok := iter.Seek(&ReverseIndexEntry[K, V]{
+				Key:       key,
+				VersionID: 0,
+			}); ok; ok = iter.Next() {
+				entry := iter.Item()
+				if key.Less(entry.Key) {
+					break
+				}
+				if newVersionIDSet[entry.VersionID] {
+					continue
+				}
+				state.indexes.Delete(&IndexEntry[K, V]{
+					Index:     entry.Index,
+					Key:       entry.Key,
+					VersionID: entry.VersionID,
+				})
+				state.reverseIndexes.Delete(entry)
+			}
+
+			// remove unique indexes
+			iter = state.reverseUniqueIndexes.Copy().Iter()
+			for ok := iter.Seek(&ReverseIndexEntry[K, V]{
+				Key:       key,
+				VersionID: 0,
+			}); ok; ok = iter.Next() {
+				entry := iter.Item()
+				if key.Less(entry.Key) {
+					break
+				}
+				if newVersionIDSet[entry.VersionID] {
+					continue
+				}
+				state.uniqueIndexes.Delete(&IndexEntry[K, V]{
+					Index:     entry.Index,
+					Key:       entry.Key,
+					VersionID: entry.VersionID,
+				})
+				state.reverseUniqueIndexes.Delete(entry)
+			}
+
+		}
+		return nil
+	})
+}
+
+func (t *Table[K, V, R]) AbortTx(tx *Transaction) error {
+	return t.update(func(state *tableState[K, V]) error {
 		iter := state.writes.Copy().Iter()
 		defer iter.Release()
 		pivot := &WriteEntry[K, V]{
diff --git a/pkg/txn/storage/memorystorage/memtable/transaction.go b/pkg/txn/storage/memorystorage/memtable/transaction.go
index 95a7a08d0..47794709e 100644
--- a/pkg/txn/storage/memorystorage/memtable/transaction.go
+++ b/pkg/txn/storage/memorystorage/memtable/transaction.go
@@ -26,7 +26,7 @@ type Transaction struct {
 
 type TxCommitter interface {
 	CommitTx(*Transaction) error
-	AbortTx(*Transaction)
+	AbortTx(*Transaction) error
 }
 
 func NewTransaction(
@@ -63,11 +63,15 @@ func (t *Transaction) Commit(commitTime Time) error {
 	return nil
 }
 
-func (t *Transaction) Abort() {
+func (t *Transaction) Abort() error {
+	var e error
 	for committer := range t.committers {
-		committer.AbortTx(t)
+		if err := committer.AbortTx(t); err != nil {
+			e = err
+		}
 	}
 	t.State.Store(Aborted)
+	return e
 }
 
 func (t *Transaction) Copy() *Transaction {
diff --git a/pkg/vm/engine/disttae/partition.go b/pkg/vm/engine/disttae/partition.go
index e16eca8c0..1aacce2d2 100644
--- a/pkg/vm/engine/disttae/partition.go
+++ b/pkg/vm/engine/disttae/partition.go
@@ -55,9 +55,10 @@ type DataValue struct {
 }
 
 type DataRow struct {
-	rowID   RowID
-	value   DataValue
-	indexes []memtable.Tuple
+	rowID         RowID
+	value         DataValue
+	indexes       []memtable.Tuple
+	uniqueIndexes []memtable.Tuple
 }
 
 type Op uint8
@@ -79,6 +80,10 @@ func (d *DataRow) Indexes() []memtable.Tuple {
 	return d.indexes
 }
 
+func (d *DataRow) UniqueIndexes() []memtable.Tuple {
+	return d.uniqueIndexes
+}
+
 var _ MVCC = new(Partition)
 
 func (*Partition) BlockList(ctx context.Context, ts timestamp.Timestamp, blocks []BlockMeta, entries []Entry) []BlockMeta {


Test patch:
diff --git a/pkg/txn/storage/memorystorage/memtable/table_test.go b/pkg/txn/storage/memorystorage/memtable/table_test.go
index 68fa0daf3..dfa6ca702 100644
--- a/pkg/txn/storage/memorystorage/memtable/table_test.go
+++ b/pkg/txn/storage/memorystorage/memtable/table_test.go
@@ -43,6 +43,12 @@ func (t TestRow) Indexes() []Tuple {
 	}
 }
 
+func (t TestRow) UniqueIndexes() []Tuple {
+	return []Tuple{
+		{Int(t.value)},
+	}
+}
+
 func TestTable(t *testing.T) {
 
 	table := NewTable[Int, int, TestRow]()
@@ -79,7 +85,6 @@ func TestTable(t *testing.T) {
 	assert.Nil(t, err)
 	assert.Equal(t, 1, len(entries))
 	assert.Equal(t, Int(42), entries[0].Key)
-	assert.Equal(t, 2, entries[0].Value)
 
 	// delete
 	err = table.Delete(tx, Int(42))
@@ -165,6 +170,10 @@ func (i Issue5388Row) Indexes() []Tuple {
 	return nil
 }
 
+func (i Issue5388Row) UniqueIndexes() []Tuple {
+	return nil
+}
+
 func TestIssue5388(t *testing.T) {
 	table := NewTable[memoryengine.ID, Issue5388Row, Issue5388Row]()
 
@@ -233,3 +242,23 @@ func TestIssue5388(t *testing.T) {
 	assert.Equal(t, 1, n)
 
 }
+
+func TestUniqueIndex(t *testing.T) {
+	table := NewTable[Int, int, TestRow]()
+	tx1 := NewTransaction("1", ts(1), Serializable)
+	tx2 := NewTransaction("2", ts(1), Serializable)
+
+	row := TestRow{key: 42, value: 1}
+
+	err := table.Insert(tx1, row)
+	assert.Nil(t, err)
+
+	err = table.Insert(tx2, row)
+	assert.Nil(t, err)
+
+	err = tx1.Commit(ts(2))
+	assert.Nil(t, err)
+
+	err = tx2.Commit(ts(2))
+	assert.True(t, moerr.IsMoErrCode(err, moerr.ErrDuplicate))
+}
diff --git a/pkg/txn/storage/memorystorage/storage_test.go b/pkg/txn/storage/memorystorage/storage_test.go
index a4590bc4c..98fed25b5 100644
--- a/pkg/txn/storage/memorystorage/storage_test.go
+++ b/pkg/txn/storage/memorystorage/storage_test.go
@@ -56,6 +56,10 @@ func testDatabase(
 			LogicalTime:  1,
 		},
 	}
+	defer func() {
+		err := s.Commit(ctx, txnMeta)
+		assert.Nil(t, err)
+	}()
 
 	// open database
 	{
@@ -128,7 +132,11 @@ func testDatabase(
 					memoryengine.GetDatabasesReq{},
 				)
 				assert.Nil(t, err)
-				assert.Equal(t, 0, len(resp.Names))
+				for _, name := range resp.Names {
+					if name == "foo" {
+						t.Fatal()
+					}
+				}
 			}
 		}()
 	}
@@ -652,6 +660,46 @@ func testDatabase(
 		assert.Nil(t, resp.Batch)
 	}
 
+	t.Run("duplicated db", func(t *testing.T) {
+		tx1 := txn.TxnMeta{
+			ID:     []byte("1"),
+			Status: txn.TxnStatus_Active,
+			SnapshotTS: timestamp.Timestamp{
+				PhysicalTime: 1,
+				LogicalTime:  1,
+			},
+		}
+		tx2 := txn.TxnMeta{
+			ID:     []byte("1"),
+			Status: txn.TxnStatus_Active,
+			SnapshotTS: timestamp.Timestamp{
+				PhysicalTime: 1,
+				LogicalTime:  1,
+			},
+		}
+		{
+			resp, err := testWrite[memoryengine.CreateDatabaseResp](
+				ctx, t, s, tx1,
+				memoryengine.OpCreateDatabase,
+				memoryengine.CreateDatabaseReq{
+					Name: "bar",
+				},
+			)
+			assert.Nil(t, err)
+			assert.NotEmpty(t, resp.ID)
+		}
+		{
+			_, err := testWrite[memoryengine.CreateDatabaseResp](
+				ctx, t, s, tx2,
+				memoryengine.OpCreateDatabase,
+				memoryengine.CreateDatabaseReq{
+					Name: "bar",
+				},
+			)
+			assert.NotNil(t, err)
+		}
+	})
+
 }
 
 func testRead[