Repository: ovn-org/libovsdb
Pull request #340 (2022-10-13 11:23:52)

server: index uniqueness is a deferred constraint
Per RFC7047, index duplication is a commit constraint and errors should
be reported as an additional operation result. Verification should be deferred 
until all operations have been processed and can be taken into account.

This commit also includes:

* moving the database & transaction to its own package
for better clarity & organization.
*  Make sure that the server transact has conformance with 
RFC7047, that is operations are processed until first error,
a result is provided for each operation, being null for non executed
operations, and include an additional result in case of a deferred
constraint validation error.
* Other small fixes and improvements made along the way.

Excerpts from the RFC7047.

  * About transact:

       ```
       The database server executes each of the specified operations in the
       specified order, except if an operation fails, then the remaining
       operations are not executed.
       ```

  * About results

    ```
     In general, "result" contains some number of successful results,
     possibly followed by an error, in turn followed by enough JSON null
     values to match the number of elements in "params".  There is one
     exception: if all of the operations succeed, but the results cannot
     be committed, then "result" will have one more element than "params",
     with the additional element being an <error>.
     ```

* About indexes

    ```
    If "indexes" is specified, it must be an array of zero or more
    <column-set>s.  A <column-set> is an array of one or more strings,
    each of which names a column.  Each <column-set> is a set of
    columns whose values, taken together within any given row, must be
    unique within the table.  This is a "deferred" constraint,
    enforced only at transaction commit time, after unreferenced rows
    are deleted and dangling weak references are removed.  Ephemeral
    columns may not be part of indexes.
    ```

Signed-off-by: Jaime Caamaño Ruiz <jcaamano@redhat.com>


Patch:
diff --git a/cache/cache.go b/cache/cache.go
index c871161..8d17037 100644
--- a/cache/cache.go
+++ b/cache/cache.go
@@ -57,14 +57,14 @@ type ErrIndexExists struct {
 	Value    interface{}
 	Index    string
 	New      string
-	Existing string
+	Existing []string
 }
 
 func (e *ErrIndexExists) Error() string {
 	return fmt.Sprintf("cannot insert %s in the %s table. item %s has identical indexes. index: %s, value: %v", e.New, e.Table, e.Existing, e.Index, e.Value)
 }
 
-func NewIndexExistsError(table string, value interface{}, index string, new, existing string) *ErrIndexExists {
+func NewIndexExistsError(table string, value interface{}, index string, new string, existing []string) *ErrIndexExists {
 	return &ErrIndexExists{
 		table, value, index, new, existing,
 	}
@@ -167,6 +167,13 @@ func (r *RowCache) Row(uuid string) model.Model {
 	return r.rowByUUID(uuid)
 }
 
+func (r *RowCache) HasRow(uuid string) bool {
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
+	_, found := r.cache[uuid]
+	return found
+}
+
 // rowsByModels searches the cache to find all rows matching any of the provided
 // models, either by UUID or indexes. An error is returned if the model schema
 // has no UUID field, or if the provided models are not all the same type.
@@ -264,7 +271,7 @@ func (r *RowCache) Create(uuid string, m model.Model, checkIndexes bool) error {
 	if err != nil {
 		return err
 	}
-	newIndexes := r.newIndexes()
+	addIndexes := r.newIndexes()
 	for _, indexSpec := range r.indexSpecs {
 		index := indexSpec.index
 		val, err := valueFromIndex(info, indexSpec.columns)
@@ -272,25 +279,29 @@ func (r *RowCache) Create(uuid string, m model.Model, checkIndexes bool) error {
 			return err
 		}
 
+		uuidset := newUUIDSet(uuid)
+
 		vals := r.indexes[index]
-		if existing, ok := vals[val]; ok && !existing.empty() && checkIndexes && indexSpec.isSchemaIndex() {
-			return NewIndexExistsError(r.name, val, string(index), uuid, existing.getAny())
+		existing := vals[val]
+		if checkIndexes && indexSpec.isSchemaIndex() && !existing.empty() && !existing.equals(uuidset) {
+			return NewIndexExistsError(r.name, val, string(index), uuid, existing.list())
 		}
 
-		uuidset := newUUIDSet(uuid)
-		if indexSpec.isSchemaIndex() {
-			newIndexes[index][val] = uuidset
-		} else {
-			newIndexes[index][val] = addUUIDSet(r.indexes[index][val], uuidset)
-		}
+		addIndexes[index][val] = uuidset
 	}
 
 	// write indexes
-	for k1, v1 := range newIndexes {
-		for k2, v2 := range v1 {
-			r.indexes[k1][k2] = v2
+	for _, indexSpec := range r.indexSpecs {
+		index := indexSpec.index
+		for k, v := range addIndexes[index] {
+			if indexSpec.isSchemaIndex() {
+				r.indexes[index][k] = v
+			} else {
+				r.indexes[index][k] = addUUIDSet(r.indexes[index][k], v)
+			}
 		}
 	}
+
 	r.cache[uuid] = model.Clone(m)
 	return nil
 }
@@ -311,7 +322,9 @@ func (r *RowCache) Update(uuid string, m model.Model, checkIndexes bool) (model.
 	if err != nil {
 		return nil, err
 	}
-	newIndexes := r.newIndexes()
+
+	addIndexes := r.newIndexes()
+	removeIndexes := r.newIndexes()
 	var errs []error
 	for _, indexSpec := range r.indexSpecs {
 		index := indexSpec.index
@@ -331,44 +344,51 @@ func (r *RowCache) Update(uuid string, m model.Model, checkIndexes bool) (model.
 		}
 		// old and new values are NOT the same
 
+		uuidset := newUUIDSet(uuid)
+
 		// check that there are no conflicts
 		vals := r.indexes[index]
-		if existing, ok := vals[newVal]; ok && indexSpec.isSchemaIndex() && checkIndexes && !existing.empty() && !existing.has(uuid) {
+		existing := vals[newVal]
+		if checkIndexes && indexSpec.isSchemaIndex() && !existing.empty() && !existing.equals(uuidset) {
 			errs = append(errs, NewIndexExistsError(
 				r.name,
 				newVal,
 				string(index),
 				uuid,
-				existing.getAny(),
+				existing.list(),
 			))
 		}
 
-		uuidset := newUUIDSet(uuid)
-		if indexSpec.isSchemaIndex() {
-			newIndexes[index][newVal] = uuidset
-			newIndexes[index][oldVal] = nil
-		} else {
-			newIndexes[index][newVal] = addUUIDSet(r.indexes[index][newVal], uuidset)
-			newIndexes[index][oldVal] = substractUUIDSet(r.indexes[index][oldVal], uuidset)
-		}
+		addIndexes[index][newVal] = uuidset
+		removeIndexes[index][oldVal] = uuidset
 	}
 	if len(errs) > 0 {
 		return nil, fmt.Errorf("%+v", errs)
 	}
+
 	// write indexes
-	for k1, v1 := range newIndexes {
-		for k2, v2 := range v1 {
-			if len(v2) == 0 {
-				delete(r.indexes[k1], k2)
+	for _, indexSpec := range r.indexSpecs {
+		index := indexSpec.index
+		for k, v := range addIndexes[index] {
+			if indexSpec.isSchemaIndex() {
+				r.indexes[index][k] = v
 			} else {
-				r.indexes[k1][k2] = v2
+				r.indexes[index][k] = addUUIDSet(r.indexes[index][k], v)
+			}
+		}
+		for k, v := range removeIndexes[index] {
+			if indexSpec.isSchemaIndex() || substractUUIDSet(r.indexes[index][k], v).empty() {
+				delete(r.indexes[index], k)
 			}
 		}
 	}
+
 	r.cache[uuid] = model.Clone(m)
 	return oldRow, nil
 }
 
+// IndexExists checks if any of the schema indexes of the provided model is
+// already in the cache under a different UUID.
 func (r *RowCache) IndexExists(row model.Model) error {
 	info, err := r.dbModel.NewModelInfo(row)
 	if err != nil {
@@ -391,13 +411,14 @@ func (r *RowCache) IndexExists(row model.Model) error {
 			continue
 		}
 		vals := r.indexes[index]
-		if existing := vals[val]; !existing.empty() && !existing.has(uuid) {
+		existing := vals[val]
+		if !existing.empty() && !existing.equals(newUUIDSet(uuid)) {
 			return NewIndexExistsError(
 				r.name,
 				val,
 				string(index),
 				uuid,
-				existing.getAny(),
+				existing.list(),
 			)
 		}
 	}
@@ -416,24 +437,31 @@ func (r *RowCache) Delete(uuid string) error {
 	if err != nil {
 		return err
 	}
+
+	removeIndexes := r.newIndexes()
 	for _, indexSpec := range r.indexSpecs {
 		index := indexSpec.index
 		oldVal, err := valueFromIndex(oldInfo, indexSpec.columns)
 		if err != nil {
 			return err
 		}
-		// only remove the index if it is pointing to this uuid
-		// otherwise we can cause a consistency issue if we've processed
-		// updates out of order
-		vals := r.indexes[index]
-		existing, ok := vals[oldVal]
-		if ok {
-			existing.remove(uuid)
-			if len(existing) == 0 {
-				delete(vals, oldVal)
+
+		removeIndexes[index][oldVal] = newUUIDSet(uuid)
+	}
+
+	// write indexes
+	for _, indexSpec := range r.indexSpecs {
+		index := indexSpec.index
+		for k, v := range removeIndexes[index] {
+			// only remove the index if it is pointing to this uuid
+			// otherwise we can cause a consistency issue if we've processed
+			// updates out of order
+			if substractUUIDSet(r.indexes[index][k], v).empty() {
+				delete(r.indexes[index], k)
 			}
 		}
 	}
+
 	delete(r.cache, uuid)
 	return nil
 }
@@ -473,6 +501,7 @@ func (r *RowCache) RowsShallow() map[string]model.Model {
 // quick as possible, via indexes, a reduced list of candidate models that might
 // match all conditions, which should be better than just evaluating all
 // conditions against all rows of a table.
+//
 //nolint:gocyclo // warns overall function is complex but ignores inner functions
 func (r *RowCache) uuidsByConditionsAsIndexes(conditions []ovsdb.Condition, nativeValues []interface{}) (uuidset, error) {
 	type indexableCondition struct {
diff --git a/cache/uuidset.go b/cache/uuidset.go
index fe3c5d8..f7c1397 100644
--- a/cache/uuidset.go
+++ b/cache/uuidset.go
@@ -24,6 +24,18 @@ func (s uuidset) has(uuid string) bool {
 	return ok
 }
 
+func (s uuidset) equals(o uuidset) bool {
+	if len(s) != len(o) {
+		return false
+	}
+	for uuid := range s {
+		if !o.has(uuid) {
+			return false
+		}
+	}
+	return true
+}
+
 func (s uuidset) getAny() string {
 	for k := range s {
 		return k
diff --git a/server/database.go b/database/database.go
similarity index 99%
rename from server/database.go
rename to database/database.go
index 0c0e55d..727b576 100644
--- a/server/database.go
+++ b/database/database.go
@@ -1,4 +1,4 @@
-package server
+package database
 
 import (
 	"fmt"
diff --git a/server/errors.go b/database/errors.go
similarity index 96%
rename from server/errors.go
rename to database/errors.go
index a0845d9..979752c 100644
--- a/server/errors.go
+++ b/database/errors.go
@@ -1,4 +1,4 @@
-package server
+package database
 
 import (
 	"fmt"
diff --git a/server/mutate.go b/database/mutate.go
similarity index 99%
rename from server/mutate.go
rename to database/mutate.go
index a761345..e2c335a 100644
--- a/server/mutate.go
+++ b/database/mutate.go
@@ -1,4 +1,4 @@
-package server
+package database
 
 import (
 	"reflect"
diff --git a/server/transact.go b/database/transaction.go
similarity index 75%
rename from server/transact.go
rename to database/transaction.go
index 7bfacf6..9de266b 100644
--- a/server/transact.go
+++ b/database/transaction.go
@@ -1,97 +1,148 @@
-package server
+package database
 
 import (
 	"fmt"
 	"reflect"
 	"time"
 
+	"github.com/go-logr/logr"
 	"github.com/google/uuid"
 	"github.com/ovn-org/libovsdb/cache"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
 )
 
-func (o *OvsdbServer) transact(name string, operations []ovsdb.Operation) ([]ovsdb.OperationResult, ovsdb.TableUpdates2) {
-	o.modelsMutex.Lock()
-	dbModel := o.models[name]
-	o.modelsMutex.Unlock()
-	transaction := o.NewTransaction(dbModel, name, o.db)
+type Transaction struct {
+	ID          uuid.UUID
+	Cache       *cache.TableCache
+	DeletedRows map[string]struct{}
+	Model       model.DatabaseModel
+	DbName      string
+	Database    Database
+}
 
-	results := []ovsdb.OperationResult{}
+func NewTransaction(model model.DatabaseModel, dbName string, database Database, logger *logr.Logger) Transaction {
+	if logger != nil {
+		l := logger.WithName("transaction")
+		logger = &l
+	}
+	cache, err := cache.NewTableCache(model, nil, logger)
+	if err != nil {
+		panic(err)
+	}
+	return Transaction{
+		ID:          uuid.New(),
+		Cache:       cache,
+		DeletedRows: make(map[string]struct{}),
+		Model:       model,
+		DbName:      dbName,
+		Database:    database,
+	}
+}
+
+func (t *Transaction) Transact(operations []ovsdb.Operation) ([]*ovsdb.OperationResult, ovsdb.TableUpdates2) {
+	results := []*ovsdb.OperationResult{}
 	updates := make(ovsdb.TableUpdates2)
 
-	// simple case: database name does not exist
-	if !o.db.Exists(name) {
-		r := ovsdb.OperationResult{
-			Error: "database does not exist",
+	var r ovsdb.OperationResult
+	for _, op := range operations {
+		// if we had a previous error, just append a nil result for every op
+		// after that
+		if r.Error != "" {
+			results = append(results, nil)
+			continue
 		}
-		for range operations {
-			results = append(results, r)
+
+		// simple case: database name does not exist
+		if !t.Database.Exists(t.DbName) {
+			r = ovsdb.OperationResult{
+				Error: "database does not exist",
+			}
+			results = append(results, &r)
+			continue
 		}
-		return results, updates
-	}
 
-	for _, op := range operations {
 		switch op.Op {
 		case ovsdb.OperationInsert:
-			r, tu := transaction.Insert(op.Table, op.UUIDName, op.Row)
-			results = append(results, r)
+			var tu ovsdb.TableUpdates2
+			r, tu = t.Insert(op.Table, op.UUIDName, op.Row)
 			if tu != nil {
 				updates.Merge(tu)
-				if err := transaction.Cache.Populate2(tu); err != nil {
+				if err := t.Cache.Populate2(tu); err != nil {
 					panic(err)
 				}
 			}
 		case ovsdb.OperationSelect:
-			r := transaction.Select(op.Table, op.Where, op.Columns)
-			results = append(results, r)
+			r = t.Select(op.Table, op.Where, op.Columns)
 		case ovsdb.OperationUpdate:
-			r, tu := transaction.Update(name, op.Table, op.Where, op.Row)
-			results = append(results, r)
+			var tu ovsdb.TableUpdates2
+			r, tu = t.Update(op.Table, op.Where, op.Row)
 			if tu != nil {
 				updates.Merge(tu)
-				if err := transaction.Cache.Populate2(tu); err != nil {
+				if err := t.Cache.Populate2(tu); err != nil {
 					panic(err)
 				}
 			}
 		case ovsdb.OperationMutate:
-			r, tu := transaction.Mutate(name, op.Table, op.Where, op.Mutations)
-			results = append(results, r)
+			var tu ovsdb.TableUpdates2
+			r, tu = t.Mutate(op.Table, op.Where, op.Mutations)
 			if tu != nil {
 				updates.Merge(tu)
-				if err := transaction.Cache.Populate2(tu); err != nil {
+				if err := t.Cache.Populate2(tu); err != nil {
 					panic(err)
 				}
 			}
 		case ovsdb.OperationDelete:
-			r, tu := transaction.Delete(name, op.Table, op.Where)
-			results = append(results, r)
+			var tu ovsdb.TableUpdates2
+			r, tu = t.Delete(op.Table, op.Where)
 			if tu != nil {
 				updates.Merge(tu)
-				if err := transaction.Cache.Populate2(tu); err != nil {
+				if err := t.Cache.Populate2(tu); err != nil {
 					panic(err)
 				}
 			}
 		case ovsdb.OperationWait:
-			r := transaction.Wait(name, op.Table, op.Timeout, op.Where, op.Columns, op.Until, op.Rows)
-			results = append(results, r)
+			r = t.Wait(op.Table, op.Timeout, op.Where, op.Columns, op.Until, op.Rows)
 		case ovsdb.OperationCommit:
 			durable := op.Durable
-			r := transaction.Commit(name, op.Table, *durable)
-			results = append(results, r)
+			r = t.Commit(op.Table, *durable)
 		case ovsdb.OperationAbort:
-			r := transaction.Abort(name, op.Table)
-			results = append(results, r)
+			r = t.Abort(op.Table)
 		case ovsdb.OperationComment:
-			r := transaction.Comment(name, op.Table, *op.Comment)
-			results = append(results, r)
+			r = t.Comment(op.Table, *op.Comment)
 		case ovsdb.OperationAssert:
-			r := transaction.Assert(name, op.Table, *op.Lock)
-			results = append(results, r)
+			r = t.Assert(op.Table, *op.Lock)
 		default:
-			return nil, updates
+			e := ovsdb.NotSupported{}
+			r = ovsdb.OperationResult{
+				Error: e.Error(),
+			}
+		}
+
+		result := r
+		results = append(results, &result)
+	}
+
+	// if an operation failed, no need to do any further validation
+	if r.Error != "" {
+		return results, updates
+	}
+
+	// check index constraints
+	if err := t.checkIndexes(); err != nil {
+		if indexExists, ok := err.(*cache.ErrIndexExists); ok {
+			e := ovsdb.ConstraintViolation{}
+			results = append(results, &ovsdb.OperationResult{
+				Error:   e.Error(),
+				Details: newIndexExistsDetails(*indexExists),
+			})
+		} else {
+			results = append(results, &ovsdb.OperationResult{
+				Error: err.Error(),
+			})
 		}
 	}
+
 	return results, updates
 }
 
@@ -110,14 +161,19 @@ func (t *Transaction) rowsFromTransactionCacheAndDatabase(table string, where []
 	for rowUUID, row := range rows {
 		if txnRow, found := txnRows[rowUUID]; found {
 			rows[rowUUID] = txnRow
+			// delete txnRows so that only inserted rows remain in txnRows
+			delete(txnRows, rowUUID)
 		} else {
 			// warm the transaction cache with the current contents of the row
 			if err := t.Cache.Table(table).Create(rowUUID, row, false); err != nil {
 				return nil, fmt.Errorf("failed warming transaction cache row %s %v for table %s: %v", rowUUID, row, table, err)
 			}
-			txnRows[rowUUID] = row
 		}
 	}
+	// add rows that have been inserted in this transaction
+	for rowUUID, row := range txnRows {
+		rows[rowUUID] = row
+	}
 	// exclude deleted rows
 	for rowUUID := range t.DeletedRows {
 		delete(rows, rowUUID)
@@ -125,15 +181,43 @@ func (t *Transaction) rowsFromTransactionCacheAndDatabase(table string, where []
 	return rows, nil
 }
 
-func (t *Transaction) checkIndexes(table string, model model.Model) error {
-	// check for index conflicts. First check on transaction cache, followed by
-	// the database's
-	targetTable := t.Cache.Table(table)
-	err := targetTable.IndexExists(model)
-	if err == nil {
-		err = t.Database.CheckIndexes(t.DbName, table, model)
+// checkIndexes checks that there are no index conflicts:
+// - no duplicate indexes among any two rows operated with in the transaction
+// - no duplicate indexes of any transaction row with any database row
+func (t *Transaction) checkIndexes() error {
+	// check for index conflicts.
+	tables := t.Cache.Tables()
+	for _, table := range tables {
+		tc := t.Cache.Table(table)
+		for _, row := range tc.RowsShallow() {
+			err := tc.IndexExists(row)
+			if err != nil {
+				return err
+			}
+			err = t.Database.CheckIndexes(t.DbName, table, row)
+			errIndexExists, isErrIndexExists := err.(*cache.ErrIndexExists)
+			if err == nil {
+				continue
+			}
+			if !isErrIndexExists {
+				return err
+			}
+			for _, existing := range errIndexExists.Existing {
+				if _, isDeleted := t.DeletedRows[existing]; isDeleted {
+					// this model is deleted in the transaction, ignore it
+					continue
+				}
+				if tc.HasRow(existing) {
+					// this model is updated in the transaction and was not
+					// detected as a duplicate, so an index must have been
+					// updated, ignore it
+					continue
+				}
+				return err
+			}
+		}
 	}
-	return err
+	return nil
 }
 
 func (t *Transaction) Insert(table string, rowUUID string, row ovsdb.Row) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
@@ -179,20 +263,6 @@ func (t *Transaction) Insert(table string, rowUUID string, row ovsdb.Row) (ovsdb
 		}, nil
 	}
 
-	// check for index conflicts
-	if err := t.checkIndexes(table, model); err != nil {
-		if indexExists, ok := err.(*cache.ErrIndexExists); ok {
-			e := ovsdb.ConstraintViolation{}
-			return ovsdb.OperationResult{
-				Error:   e.Error(),
-				Details: newIndexExistsDetails(*indexExists),
-			}, nil
-		}
-		return ovsdb.OperationResult{
-			Error: err.Error(),
-		}, nil
-	}
-
 	result := ovsdb.OperationResult{
 		UUID: ovsdb.UUID{GoUUID: rowUUID},
 	}
@@ -233,7 +303,7 @@ func (t *Transaction) Select(table string, where []ovsdb.Condition, columns []st
 	}
 }
 
-func (t *Transaction) Update(database, table string, where []ovsdb.Condition, row ovsdb.Row) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
+func (t *Transaction) Update(table string, where []ovsdb.Condition, row ovsdb.Row) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
 	dbModel := t.Model
 	m := dbModel.Mapper
 	schema := dbModel.Schema.Table(table)
@@ -327,20 +397,6 @@ func (t *Transaction) Update(database, table string, where []ovsdb.Condition, ro
 			panic(err)
 		}
 
-		// check for index conflicts
-		if err := t.checkIndexes(table, new); err != nil {
-			if indexExists, ok := err.(*cache.ErrIndexExists); ok {
-				e := ovsdb.ConstraintViolation{}
-				return ovsdb.OperationResult{
-					Error:   e.Error(),
-					Details: newIndexExistsDetails(*indexExists),
-				}, nil
-			}
-			return ovsdb.OperationResult{
-				Error: err.Error(),
-			}, nil
-		}
-
 		tableUpdate.AddRowUpdate(uuid, &ovsdb.RowUpdate2{
 			Modify: &rowDelta,
 			Old:    &oldRow,
@@ -355,7 +411,7 @@ func (t *Transaction) Update(database, table string, where []ovsdb.Condition, ro
 		}
 }
 
-func (t *Transaction) Mutate(database, table string, where []ovsdb.Condition, mutations []ovsdb.Mutation) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
+func (t *Transaction) Mutate(table string, where []ovsdb.Condition, mutations []ovsdb.Mutation) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
 	dbModel := t.Model
 	m := dbModel.Mapper
 	schema := dbModel.Schema.Table(table)
@@ -452,20 +508,6 @@ func (t *Transaction) Mutate(database, table string, where []ovsdb.Condition, mu
 			}
 		}
 
-		// check indexes
-		if err := t.checkIndexes(table, new); err != nil {
-			if indexExists, ok := err.(*cache.ErrIndexExists); ok {
-				e := ovsdb.ConstraintViolation{}
-				return ovsdb.OperationResult{
-					Error:   e.Error(),
-					Details: newIndexExistsDetails(*indexExists),
-				}, nil
-			}
-			return ovsdb.OperationResult{
-				Error: err.Error(),
-			}, nil
-		}
-
 		newRow, err := m.NewRow(newInfo)
 		if err != nil {
 			panic(err)
@@ -485,7 +527,7 @@ func (t *Transaction) Mutate(database, table string, where []ovsdb.Condition, mu
 		}
 }
 
-func (t *Transaction) Delete(database, table string, where []ovsdb.Condition) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
+func (t *Transaction) Delete(table string, where []ovsdb.Condition) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
 	dbModel := t.Model
 	m := dbModel.Mapper
 	tableUpdate := make(ovsdb.TableUpdate2)
@@ -515,7 +557,7 @@ func (t *Transaction) Delete(database, table string, where []ovsdb.Condition) (o
 		}
 }
 
-func (t *Transaction) Wait(database, table string, timeout *int, where []ovsdb.Condition, columns []string, until string, rows []ovsdb.Row) ovsdb.OperationResult {
+func (t *Transaction) Wait(table string, timeout *int, where []ovsdb.Condition, columns []string, until string, rows []ovsdb.Row) ovsdb.OperationResult {
 	start := time.Now()
 
 	if until != "!=" && until != "==" {
@@ -614,22 +656,22 @@ Loop:
 	return ovsdb.OperationResult{Error: e.Error()}
 }
 
-func (t *Transaction) Commit(database, table string, durable bool) ovsdb.OperationResult {
+func (t *Transaction) Commit(table string, durable bool) ovsdb.OperationResult {
 	e := ovsdb.NotSupported{}
 	return ovsdb.OperationResult{Error: e.Error()}
 }
 
-func (t *Transaction) Abort(database, table string) ovsdb.OperationResult {
+func (t *Transaction) Abort(table string) ovsdb.OperationResult {
 	e := ovsdb.NotSupported{}
 	return ovsdb.OperationResult{Error: e.Error()}
 }
 
-func (t *Transaction) Comment(database, table string, comment string) ovsdb.OperationResult {
+func (t *Transaction) Comment(table string, comment string) ovsdb.OperationResult {
 	e := ovsdb.NotSupported{}
 	return ovsdb.OperationResult{Error: e.Error()}
 }
 
-func (t *Transaction) Assert(database, table, lock string) ovsdb.OperationResult {
+func (t *Transaction) Assert(table, lock string) ovsdb.OperationResult {
 	e := ovsdb.NotSupported{}
 	return ovsdb.OperationResult{Error: e.Error()}
 }
diff --git a/example/ovsdb-server/main.go b/example/ovsdb-server/main.go
index f65f0c2..e958280 100644
--- a/example/ovsdb-server/main.go
+++ b/example/ovsdb-server/main.go
@@ -13,6 +13,7 @@ import (
 	"time"
 
 	"github.com/ovn-org/libovsdb/client"
+	"github.com/ovn-org/libovsdb/database"
 	"github.com/ovn-org/libovsdb/example/vswitchd"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
@@ -57,7 +58,7 @@ func main() {
 		log.Fatal(err)
 	}
 
-	ovsDB := server.NewInMemoryDatabase(map[string]model.ClientDBModel{
+	ovsDB := database.NewInMemoryDatabase(map[string]model.ClientDBModel{
 		schema.Name: clientDBModel,
 	})
 
diff --git a/server/server.go b/server/server.go
index 9be6886..94543de 100644
--- a/server/server.go
+++ b/server/server.go
@@ -14,7 +14,7 @@ import (
 	"github.com/go-logr/logr"
 	"github.com/go-logr/stdr"
 	"github.com/google/uuid"
-	"github.com/ovn-org/libovsdb/cache"
+	"github.com/ovn-org/libovsdb/database"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
 )
@@ -24,7 +24,7 @@ type OvsdbServer struct {
 	srv          *rpc2.Server
 	listener     net.Listener
 	done         chan struct{}
-	db           Database
+	db           database.Database
 	ready        bool
 	readyMutex   sync.RWMutex
 	models       map[string]model.DatabaseModel
@@ -36,7 +36,7 @@ type OvsdbServer struct {
 }
 
 // NewOvsdbServer returns a new OvsdbServer
-func NewOvsdbServer(db Database, models ...model.DatabaseModel) (*OvsdbServer, error) {
+func NewOvsdbServer(db database.Database, models ...model.DatabaseModel) (*OvsdbServer, error) {
 	l := stdr.NewWithOptions(log.New(os.Stderr, "", log.LstdFlags), stdr.Options{LogCaller: stdr.All}).WithName("server")
 	stdr.SetVerbosity(5)
 	o := &OvsdbServer{
@@ -163,32 +163,8 @@ func (o *OvsdbServer) GetSchema(client *rpc2.Client, args []interface{}, reply *
 	return nil
 }
 
-type Transaction struct {
-	ID          uuid.UUID
-	Cache       *cache.TableCache
-	DeletedRows map[string]struct{}
-	Model       model.DatabaseModel
-	DbName      string
-	Database    Database
-}
-
-func (o *OvsdbServer) NewTransaction(model model.DatabaseModel, dbName string, database Database) Transaction {
-	cache, err := cache.NewTableCache(model, nil, &o.logger)
-	if err != nil {
-		panic(err)
-	}
-	return Transaction{
-		ID:          uuid.New(),
-		Cache:       cache,
-		DeletedRows: make(map[string]struct{}),
-		Model:       model,
-		DbName:      dbName,
-		Database:    database,
-	}
-}
-
 // Transact issues a new database transaction and returns the results
-func (o *OvsdbServer) Transact(client *rpc2.Client, args []json.RawMessage, reply *[]ovsdb.OperationResult) error {
+func (o *OvsdbServer) Transact(client *rpc2.Client, args []json.RawMessage, reply *[]*ovsdb.OperationResult) error {
 	// While allowing other rpc handlers to run in parallel, this ovsdb server expects transactions
 	// to be serialized. The following mutex ensures that.
 	// Ref: https://github.com/cenkalti/rpc2/blob/c1acbc6ec984b7ae6830b6a36b62f008d5aefc4c/client.go#L187
@@ -203,9 +179,6 @@ func (o *OvsdbServer) Transact(client *rpc2.Client, args []json.RawMessage, repl
 	if err != nil {
 		return fmt.Errorf("database %v is not a string", args[0])
 	}
-	if !o.db.Exists(db) {
-		return fmt.Errorf("db does not exist")
-	}
 	var ops []ovsdb.Operation
 	namedUUID := make(map[string]ovsdb.UUID)
 	for i := 1; i < len(args); i++ {
@@ -237,9 +210,9 @@ func (o *OvsdbServer) Transact(client *rpc2.Client, args []json.RawMessage, repl
 	}
 	response, updates := o.transact(db, ops)
 	*reply = response
-	for i, operResult := range response {
+	for _, operResult := range response {
 		if operResult.Error != "" {
-			o.logger.Error(errors.New("failed to process operation"), "Skipping transaction DB commit due to error", "operations", ops, "failed operation", ops[i], "operation error", operResult.Error)
+			o.logger.Error(errors.New("failed to process operation"), "Skipping transaction DB commit due to error", "operations", ops, "results", response, "operation error", operResult.Error)
 			return nil
 		}
 	}
@@ -248,6 +221,14 @@ func (o *OvsdbServer) Transact(client *rpc2.Client, args []json.RawMessage, repl
 	return o.db.Commit(db, transactionID, updates)
 }
 
+func (o *OvsdbServer) transact(name string, operations []ovsdb.Operation) ([]*ovsdb.OperationResult, ovsdb.TableUpdates2) {
+	o.modelsMutex.Lock()
+	dbModel := o.models[name]
+	o.modelsMutex.Unlock()
+	transaction := database.NewTransaction(dbModel, name, o.db, &o.logger)
+	return transaction.Transact(operations)
+}
+
 func deepCopy(a ovsdb.TableUpdates) (ovsdb.TableUpdates, error) {
 	var b ovsdb.TableUpdates
 	raw, err := json.Marshal(a)
@@ -301,7 +282,7 @@ func (o *OvsdbServer) Monitor(client *rpc2.Client, args []json.RawMessage, reply
 	o.modelsMutex.Lock()
 	dbModel := o.models[db]
 	o.modelsMutex.Unlock()
-	transaction := o.NewTransaction(dbModel, db, o.db)
+	transaction := database.NewTransaction(dbModel, db, o.db, &o.logger)
 
 	tableUpdates := make(ovsdb.TableUpdates)
 	for t, request := range request {
@@ -348,7 +329,7 @@ func (o *OvsdbServer) MonitorCond(client *rpc2.Client, args []json.RawMessage, r
 	o.modelsMutex.Lock()
 	dbModel := o.models[db]
 	o.modelsMutex.Unlock()
-	transaction := o.NewTransaction(dbModel, db, o.db)
+	transaction := database.NewTransaction(dbModel, db, o.db, &o.logger)
 
 	tableUpdates := make(ovsdb.TableUpdates2)
 	for t, request := range request {
@@ -393,7 +374,7 @@ func (o *OvsdbServer) MonitorCondSince(client *rpc2.Client, args []json.RawMessa
 	o.modelsMutex.Lock()
 	dbModel := o.models[db]
 	o.modelsMutex.Unlock()
-	transaction := o.NewTransaction(dbModel, db, o.db)
+	transaction := database.NewTransaction(dbModel, db, o.db, &o.logger)
 
 	tableUpdates := make(ovsdb.TableUpdates2)
 	for t, request := range request {


Test patch:
diff --git a/client/client_test.go b/client/client_test.go
index 2850268..b4a5e0a 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -16,6 +16,7 @@ import (
 	"github.com/cenkalti/rpc2"
 	"github.com/google/uuid"
 	"github.com/ovn-org/libovsdb/cache"
+	db "github.com/ovn-org/libovsdb/database"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
 	"github.com/ovn-org/libovsdb/ovsdb/serverdb"
@@ -870,7 +871,7 @@ func newOVSDBServer(t *testing.T, dbModel model.ClientDBModel, schema ovsdb.Data
 	require.NoError(t, err)
 	serverSchema := serverdb.Schema()
 
-	db := server.NewInMemoryDatabase(map[string]model.ClientDBModel{
+	db := db.NewInMemoryDatabase(map[string]model.ClientDBModel{
 		schema.Name:       dbModel,
 		serverSchema.Name: serverDBModel,
 	})
diff --git a/server/mutate_test.go b/database/mutate_test.go
similarity index 99%
rename from server/mutate_test.go
rename to database/mutate_test.go
index b7559bf..73b1a0f 100644
--- a/server/mutate_test.go
+++ b/database/mutate_test.go
@@ -1,4 +1,4 @@
-package server
+package database
 
 import (
 	"testing"
diff --git a/server/transact_test.go b/database/transaction_test.go
similarity index 66%
rename from server/transact_test.go
rename to database/transaction_test.go
index a962c05..130d504 100644
--- a/server/transact_test.go
+++ b/database/transaction_test.go
@@ -1,4 +1,4 @@
-package server
+package database
 
 import (
 	"encoding/json"
@@ -6,42 +6,45 @@ import (
 	"time"
 
 	"github.com/google/uuid"
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+
 	"github.com/ovn-org/libovsdb/mapper"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
-	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
+
+	. "github.com/ovn-org/libovsdb/test"
 )
 
 func TestWaitOpEquals(t *testing.T) {
 	defDB, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
-		"Open_vSwitch": &ovsType{},
-		"Bridge":       &bridgeType{}})
+		"Open_vSwitch": &OvsType{},
+		"Bridge":       &BridgeType{}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	schema, err := getSchema()
+	schema, err := GetSchema()
 	if err != nil {
 		t.Fatal(err)
 	}
-	ovsDB := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	db := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	err = db.CreateDatabase("Open_vSwitch", schema)
+	require.NoError(t, err)
 	dbModel, errs := model.NewDatabaseModel(schema, defDB)
 	require.Empty(t, errs)
-	o, err := NewOvsdbServer(ovsDB, dbModel)
-	require.Nil(t, err)
 
 	ovsUUID := uuid.NewString()
 	bridgeUUID := uuid.NewString()
 
 	m := mapper.NewMapper(schema)
 
-	ovs := ovsType{}
+	ovs := OvsType{}
 	info, err := dbModel.NewModelInfo(&ovs)
 	require.NoError(t, err)
 	ovsRow, err := m.NewRow(info)
 	require.Nil(t, err)
 
-	bridge := bridgeType{
+	bridge := BridgeType{
 		Name: "foo",
 		ExternalIds: map[string]string{
 			"foo":   "bar",
@@ -54,7 +57,7 @@ func TestWaitOpEquals(t *testing.T) {
 	bridgeRow, err := m.NewRow(bridgeInfo)
 	require.Nil(t, err)
 
-	transaction := o.NewTransaction(dbModel, "Open_vSwitch", o.db)
+	transaction := NewTransaction(dbModel, "Open_vSwitch", db, nil)
 
 	res, updates := transaction.Insert("Open_vSwitch", ovsUUID, ovsRow)
 	_, err = ovsdb.CheckOperationResults([]ovsdb.OperationResult{res}, []ovsdb.Operation{{Op: "insert"}})
@@ -65,13 +68,12 @@ func TestWaitOpEquals(t *testing.T) {
 	require.Nil(t, err)
 
 	updates.Merge(update2)
-	err = o.db.Commit("Open_vSwitch", uuid.New(), updates)
+	err = db.Commit("Open_vSwitch", uuid.New(), updates)
 	require.NoError(t, err)
 
 	timeout := 0
 	// Attempt to wait for row with name foo to appear
 	gotResult := transaction.Wait(
-		"Open_vSwitch",
 		"Bridge",
 		&timeout,
 		[]ovsdb.Condition{ovsdb.NewCondition("name", ovsdb.ConditionEqual, "foo")},
@@ -84,7 +86,6 @@ func TestWaitOpEquals(t *testing.T) {
 
 	// Attempt to wait for 2 rows, where one does not exist
 	gotResult = transaction.Wait(
-		"Open_vSwitch",
 		"Bridge",
 		&timeout,
 		[]ovsdb.Condition{ovsdb.NewCondition("name", ovsdb.ConditionEqual, "foo")},
@@ -103,7 +104,6 @@ func TestWaitOpEquals(t *testing.T) {
 	require.Nil(t, err)
 	// Attempt to wait for a row, with multiple columns specified
 	gotResult = transaction.Wait(
-		"Open_vSwitch",
 		"Bridge",
 		&timeout,
 		[]ovsdb.Condition{ovsdb.NewCondition("name", ovsdb.ConditionEqual, "foo")},
@@ -116,7 +116,6 @@ func TestWaitOpEquals(t *testing.T) {
 
 	// Attempt to wait for a row, with multiple columns, but not specified in row filtering
 	gotResult = transaction.Wait(
-		"Open_vSwitch",
 		"Bridge",
 		&timeout,
 		[]ovsdb.Condition{ovsdb.NewCondition("name", ovsdb.ConditionEqual, "foo")},
@@ -130,7 +129,6 @@ func TestWaitOpEquals(t *testing.T) {
 	// Attempt to get something with a non-zero timeout that will fail
 	timeout = 400
 	gotResult = transaction.Wait(
-		"Open_vSwitch",
 		"Bridge",
 		&timeout,
 		[]ovsdb.Condition{ovsdb.NewCondition("name", ovsdb.ConditionEqual, "foo")},
@@ -145,33 +143,33 @@ func TestWaitOpEquals(t *testing.T) {
 
 func TestWaitOpNotEquals(t *testing.T) {
 	defDB, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
-		"Open_vSwitch": &ovsType{},
-		"Bridge":       &bridgeType{}})
+		"Open_vSwitch": &OvsType{},
+		"Bridge":       &BridgeType{}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	schema, err := getSchema()
+	schema, err := GetSchema()
 	if err != nil {
 		t.Fatal(err)
 	}
-	ovsDB := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	db := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	err = db.CreateDatabase("Open_vSwitch", schema)
+	require.NoError(t, err)
 	dbModel, errs := model.NewDatabaseModel(schema, defDB)
 	require.Empty(t, errs)
-	o, err := NewOvsdbServer(ovsDB, dbModel)
-	require.Nil(t, err)
 
 	ovsUUID := uuid.NewString()
 	bridgeUUID := uuid.NewString()
 
 	m := mapper.NewMapper(schema)
 
-	ovs := ovsType{}
+	ovs := OvsType{}
 	info, err := dbModel.NewModelInfo(&ovs)
 	require.NoError(t, err)
 	ovsRow, err := m.NewRow(info)
 	require.Nil(t, err)
 
-	bridge := bridgeType{
+	bridge := BridgeType{
 		Name: "foo",
 		ExternalIds: map[string]string{
 			"foo":   "bar",
@@ -184,7 +182,7 @@ func TestWaitOpNotEquals(t *testing.T) {
 	bridgeRow, err := m.NewRow(bridgeInfo)
 	require.Nil(t, err)
 
-	transaction := o.NewTransaction(dbModel, "Open_vSwitch", o.db)
+	transaction := NewTransaction(dbModel, "Open_vSwitch", db, nil)
 
 	res, updates := transaction.Insert("Open_vSwitch", ovsUUID, ovsRow)
 	_, err = ovsdb.CheckOperationResults([]ovsdb.OperationResult{res}, []ovsdb.Operation{{Op: "insert"}})
@@ -195,13 +193,12 @@ func TestWaitOpNotEquals(t *testing.T) {
 	require.Nil(t, err)
 
 	updates.Merge(update2)
-	err = o.db.Commit("Open_vSwitch", uuid.New(), updates)
+	err = db.Commit("Open_vSwitch", uuid.New(), updates)
 	require.NoError(t, err)
 
 	timeout := 0
 	// Attempt a wait where no entry with name blah should exist
 	gotResult := transaction.Wait(
-		"Open_vSwitch",
 		"Bridge",
 		&timeout,
 		[]ovsdb.Condition{ovsdb.NewCondition("name", ovsdb.ConditionEqual, "foo")},
@@ -214,7 +211,6 @@ func TestWaitOpNotEquals(t *testing.T) {
 
 	// Attempt another wait with multiple rows specified, one that would match, and one that doesn't
 	gotResult = transaction.Wait(
-		"Open_vSwitch",
 		"Bridge",
 		&timeout,
 		[]ovsdb.Condition{ovsdb.NewCondition("name", ovsdb.ConditionEqual, "foo")},
@@ -234,7 +230,6 @@ func TestWaitOpNotEquals(t *testing.T) {
 	require.Nil(t, err)
 	// Attempt to wait for a row, with multiple columns specified and one is not a match
 	gotResult = transaction.Wait(
-		"Open_vSwitch",
 		"Bridge",
 		&timeout,
 		[]ovsdb.Condition{ovsdb.NewCondition("name", ovsdb.ConditionEqual, "foo")},
@@ -249,7 +244,6 @@ func TestWaitOpNotEquals(t *testing.T) {
 	start := time.Now()
 	timeout = 200
 	gotResult = transaction.Wait(
-		"Open_vSwitch",
 		"Bridge",
 		&timeout,
 		[]ovsdb.Condition{ovsdb.NewCondition("name", ovsdb.ConditionEqual, "foo")},
@@ -267,33 +261,33 @@ func TestWaitOpNotEquals(t *testing.T) {
 
 func TestMutateOp(t *testing.T) {
 	defDB, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
-		"Open_vSwitch": &ovsType{},
-		"Bridge":       &bridgeType{}})
+		"Open_vSwitch": &OvsType{},
+		"Bridge":       &BridgeType{}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	schema, err := getSchema()
+	schema, err := GetSchema()
 	if err != nil {
 		t.Fatal(err)
 	}
-	ovsDB := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	db := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	err = db.CreateDatabase("Open_vSwitch", schema)
+	require.NoError(t, err)
 	dbModel, errs := model.NewDatabaseModel(schema, defDB)
 	require.Empty(t, errs)
-	o, err := NewOvsdbServer(ovsDB, dbModel)
-	require.Nil(t, err)
 
 	ovsUUID := uuid.NewString()
 	bridgeUUID := uuid.NewString()
 
 	m := mapper.NewMapper(schema)
 
-	ovs := ovsType{}
+	ovs := OvsType{}
 	info, err := dbModel.NewModelInfo(&ovs)
 	require.NoError(t, err)
 	ovsRow, err := m.NewRow(info)
 	require.Nil(t, err)
 
-	bridge := bridgeType{
+	bridge := BridgeType{
 		Name: "foo",
 		ExternalIds: map[string]string{
 			"foo":   "bar",
@@ -306,7 +300,7 @@ func TestMutateOp(t *testing.T) {
 	bridgeRow, err := m.NewRow(bridgeInfo)
 	require.Nil(t, err)
 
-	transaction := o.NewTransaction(dbModel, "Open_vSwitch", o.db)
+	transaction := NewTransaction(dbModel, "Open_vSwitch", db, nil)
 
 	res, updates := transaction.Insert("Open_vSwitch", ovsUUID, ovsRow)
 	_, err = ovsdb.CheckOperationResults([]ovsdb.OperationResult{res}, []ovsdb.Operation{{Op: "insert"}})
@@ -317,11 +311,10 @@ func TestMutateOp(t *testing.T) {
 	require.Nil(t, err)
 
 	updates.Merge(update2)
-	err = o.db.Commit("Open_vSwitch", uuid.New(), updates)
+	err = db.Commit("Open_vSwitch", uuid.New(), updates)
 	require.NoError(t, err)
 
 	gotResult, gotUpdate := transaction.Mutate(
-		"Open_vSwitch",
 		"Open_vSwitch",
 		[]ovsdb.Condition{
 			ovsdb.NewCondition("_uuid", ovsdb.ConditionEqual, ovsdb.UUID{GoUUID: ovsUUID}),
@@ -331,7 +324,7 @@ func TestMutateOp(t *testing.T) {
 		},
 	)
 	assert.Equal(t, ovsdb.OperationResult{Count: 1}, gotResult)
-	err = o.db.Commit("Open_vSwitch", uuid.New(), gotUpdate)
+	err = db.Commit("Open_vSwitch", uuid.New(), gotUpdate)
 	require.NoError(t, err)
 
 	bridgeSet, err := ovsdb.NewOvsSet([]ovsdb.UUID{{GoUUID: bridgeUUID}})
@@ -360,7 +353,6 @@ func TestMutateOp(t *testing.T) {
 	keyValueDelete, err := ovsdb.NewOvsMap(map[string]string{"baz": "quux"})
 	assert.Nil(t, err)
 	gotResult, gotUpdate = transaction.Mutate(
-		"Open_vSwitch",
 		"Bridge",
 		[]ovsdb.Condition{
 			ovsdb.NewCondition("_uuid", ovsdb.ConditionEqual, ovsdb.UUID{GoUUID: bridgeUUID}),
@@ -523,24 +515,24 @@ func TestDiff(t *testing.T) {
 func TestOvsdbServerInsert(t *testing.T) {
 	t.Skip("need a helper for comparing rows as map elements aren't in same order")
 	defDB, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
-		"Open_vSwitch": &ovsType{},
-		"Bridge":       &bridgeType{}})
+		"Open_vSwitch": &OvsType{},
+		"Bridge":       &BridgeType{}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	schema, err := getSchema()
+	schema, err := GetSchema()
 	if err != nil {
 		t.Fatal(err)
 	}
-	ovsDB := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	db := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	err = db.CreateDatabase("Open_vSwitch", schema)
+	require.NoError(t, err)
 	dbModel, errs := model.NewDatabaseModel(schema, defDB)
 	require.Empty(t, errs)
-	o, err := NewOvsdbServer(ovsDB, dbModel)
-	require.Nil(t, err)
 	m := mapper.NewMapper(schema)
 
 	gromit := "gromit"
-	bridge := bridgeType{
+	bridge := BridgeType{
 		Name:         "foo",
 		DatapathType: "bar",
 		DatapathID:   &gromit,
@@ -556,17 +548,17 @@ func TestOvsdbServerInsert(t *testing.T) {
 	bridgeRow, err := m.NewRow(bridgeInfo)
 	require.Nil(t, err)
 
-	transaction := o.NewTransaction(dbModel, "Open_vSwitch", o.db)
+	transaction := NewTransaction(dbModel, "Open_vSwitch", db, nil)
 
 	res, updates := transaction.Insert("Bridge", bridgeUUID, bridgeRow)
 	_, err = ovsdb.CheckOperationResults([]ovsdb.OperationResult{res}, []ovsdb.Operation{{Op: "insert"}})
 	require.NoError(t, err)
 
-	err = ovsDB.Commit("Open_vSwitch", uuid.New(), updates)
+	err = db.Commit("Open_vSwitch", uuid.New(), updates)
 	assert.NoError(t, err)
 
 	bridge.UUID = bridgeUUID
-	br, err := o.db.Get("Open_vSwitch", "Bridge", bridgeUUID)
+	br, err := db.Get("Open_vSwitch", "Bridge", bridgeUUID)
 	assert.NoError(t, err)
 	assert.Equal(t, &bridge, br)
 	assert.Equal(t, ovsdb.TableUpdates2{
@@ -581,24 +573,24 @@ func TestOvsdbServerInsert(t *testing.T) {
 
 func TestOvsdbServerUpdate(t *testing.T) {
 	defDB, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
-		"Open_vSwitch": &ovsType{},
-		"Bridge":       &bridgeType{}})
+		"Open_vSwitch": &OvsType{},
+		"Bridge":       &BridgeType{}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	schema, err := getSchema()
+	schema, err := GetSchema()
 	if err != nil {
 		t.Fatal(err)
 	}
-	ovsDB := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	db := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	err = db.CreateDatabase("Open_vSwitch", schema)
+	require.NoError(t, err)
 	dbModel, errs := model.NewDatabaseModel(schema, defDB)
 	require.Empty(t, errs)
-	o, err := NewOvsdbServer(ovsDB, dbModel)
-	require.Nil(t, err)
 	m := mapper.NewMapper(schema)
 
 	christmas := "christmas"
-	bridge := bridgeType{
+	bridge := BridgeType{
 		Name:       "foo",
 		DatapathID: &christmas,
 		ExternalIds: map[string]string{
@@ -613,13 +605,13 @@ func TestOvsdbServerUpdate(t *testing.T) {
 	bridgeRow, err := m.NewRow(bridgeInfo)
 	require.Nil(t, err)
 
-	transaction := o.NewTransaction(dbModel, "Open_vSwitch", o.db)
+	transaction := NewTransaction(dbModel, "Open_vSwitch", db, nil)
 
 	res, updates := transaction.Insert("Bridge", bridgeUUID, bridgeRow)
 	_, err = ovsdb.CheckOperationResults([]ovsdb.OperationResult{res}, []ovsdb.Operation{{Op: "insert"}})
 	require.NoError(t, err)
 
-	err = ovsDB.Commit("Open_vSwitch", uuid.New(), updates)
+	err = db.Commit("Open_vSwitch", uuid.New(), updates)
 	assert.NoError(t, err)
 
 	halloween, _ := ovsdb.NewOvsSet([]string{"halloween"})
@@ -669,7 +661,7 @@ func TestOvsdbServerUpdate(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			res, updates := transaction.Update(
-				"Open_vSwitch", "Bridge",
+				"Bridge",
 				[]ovsdb.Condition{{
 					Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: bridgeUUID},
 				}}, tt.row)
@@ -677,9 +669,9 @@ func TestOvsdbServerUpdate(t *testing.T) {
 			require.NoErrorf(t, err, "%+v", errs)
 
 			bridge.UUID = bridgeUUID
-			row, err := o.db.Get("Open_vSwitch", "Bridge", bridgeUUID)
+			row, err := db.Get("Open_vSwitch", "Bridge", bridgeUUID)
 			assert.NoError(t, err)
-			br := row.(*bridgeType)
+			br := row.(*BridgeType)
 			assert.NotEqual(t, br, bridgeRow)
 			assert.Equal(t, tt.expected.Modify, updates["Bridge"][bridgeUUID].Modify)
 		})
@@ -688,47 +680,59 @@ func TestOvsdbServerUpdate(t *testing.T) {
 
 func TestMultipleOps(t *testing.T) {
 	defDB, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
-		"Open_vSwitch": &ovsType{},
-		"Bridge":       &bridgeType{}})
+		"Open_vSwitch": &OvsType{},
+		"Bridge":       &BridgeType{}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	schema, err := getSchema()
+	schema, err := GetSchema()
 	if err != nil {
 		t.Fatal(err)
 	}
-	ovsDB := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	db := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	err = db.CreateDatabase("Open_vSwitch", schema)
+	require.NoError(t, err)
 	dbModel, errs := model.NewDatabaseModel(schema, defDB)
 	require.Empty(t, errs)
-	o, err := NewOvsdbServer(ovsDB, dbModel)
-	require.Nil(t, err)
-	m := mapper.NewMapper(schema)
+
+	var ops []ovsdb.Operation
+	var op ovsdb.Operation
 
 	bridgeUUID := uuid.NewString()
-	bridge := bridgeType{
-		Name: "a_bridge_to_nowhere",
-		Ports: []string{
-			"port1",
-			"port10",
+	op = ovsdb.Operation{
+		Op:       ovsdb.OperationInsert,
+		Table:    "Bridge",
+		UUIDName: bridgeUUID,
+		Row: ovsdb.Row{
+			"name": "a_bridge_to_nowhere",
 		},
 	}
-	bridgeInfo, err := dbModel.NewModelInfo(&bridge)
-	require.NoError(t, err)
-	bridgeRow, err := m.NewRow(bridgeInfo)
-	require.Nil(t, err)
-
-	transaction := o.NewTransaction(dbModel, "Open_vSwitch", o.db)
-
-	res, updates := transaction.Insert("Bridge", bridgeUUID, bridgeRow)
-	_, err = ovsdb.CheckOperationResults([]ovsdb.OperationResult{res}, []ovsdb.Operation{{Op: "insert"}})
-	require.Nil(t, err)
-
-	err = o.db.Commit("Open_vSwitch", uuid.New(), updates)
-	require.NoError(t, err)
+	ops = append(ops, op)
 
-	var ops []ovsdb.Operation
-	var op ovsdb.Operation
+	op = ovsdb.Operation{
+		Op:    ovsdb.OperationUpdate,
+		Table: "Bridge",
+		Where: []ovsdb.Condition{
+			ovsdb.NewCondition("_uuid", ovsdb.ConditionEqual, ovsdb.UUID{GoUUID: bridgeUUID}),
+		},
+		Row: ovsdb.Row{
+			"ports": ovsdb.OvsSet{GoSet: []interface{}{ovsdb.UUID{GoUUID: "port1"}, ovsdb.UUID{GoUUID: "port10"}}},
+		},
+	}
+	ops = append(ops, op)
 
+	transaction := NewTransaction(dbModel, "Open_vSwitch", db, nil)
+	results, _ := transaction.Transact(ops)
+	assert.Len(t, results, len(ops))
+	assert.NotNil(t, results[0])
+	assert.Empty(t, results[0].Error)
+	assert.Equal(t, 0, results[0].Count)
+	assert.Equal(t, bridgeUUID, results[0].UUID.GoUUID)
+	assert.NotNil(t, results[1])
+	assert.Equal(t, 1, results[1].Count)
+	assert.Empty(t, results[1].Error)
+
+	ops = ops[:0]
 	portA, err := ovsdb.NewOvsSet([]ovsdb.UUID{{GoUUID: "portA"}})
 	require.NoError(t, err)
 	op = ovsdb.Operation{
@@ -757,15 +761,13 @@ func TestMultipleOps(t *testing.T) {
 	}
 	ops = append(ops, op2)
 
-	results, updates := o.transact("Open_vSwitch", ops)
+	results, updates := transaction.Transact(ops)
 	require.Len(t, results, len(ops))
 	for _, result := range results {
-		assert.Equal(t, "", result.Error)
+		assert.Empty(t, result.Error)
+		assert.Equal(t, 1, result.Count)
 	}
 
-	err = o.db.Commit("Open_vSwitch", uuid.New(), updates)
-	require.NoError(t, err)
-
 	modifiedPorts, err := ovsdb.NewOvsSet([]ovsdb.UUID{{GoUUID: "portA"}, {GoUUID: "portB"}, {GoUUID: "portC"}})
 	assert.Nil(t, err)
 
@@ -796,3 +798,251 @@ func TestMultipleOps(t *testing.T) {
 	}, updates)
 
 }
+
+func TestOvsdbServerDbDoesNotExist(t *testing.T) {
+	defDB, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
+		"Open_vSwitch": &OvsType{},
+		"Bridge":       &BridgeType{}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	schema, err := GetSchema()
+	if err != nil {
+		t.Fatal(err)
+	}
+	db := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	err = db.CreateDatabase("Open_vSwitch", schema)
+	require.NoError(t, err)
+	dbModel, errs := model.NewDatabaseModel(schema, defDB)
+	require.Empty(t, errs)
+
+	ops := []ovsdb.Operation{
+		{
+			Op:       ovsdb.OperationInsert,
+			Table:    "Bridge",
+			UUIDName: uuid.NewString(),
+			Row: ovsdb.Row{
+				"name": "bridge",
+			},
+		},
+		{
+			Op:    ovsdb.OperationUpdate,
+			Table: "Bridge",
+			Where: []ovsdb.Condition{
+				{
+					Column:   "name",
+					Function: ovsdb.ConditionEqual,
+					Value:    "bridge",
+				},
+			},
+			Row: ovsdb.Row{
+				"datapath_type": "type",
+			},
+		},
+	}
+
+	transaction := NewTransaction(dbModel, "nonexsitent_db", db, nil)
+	res, _ := transaction.Transact(ops)
+	assert.Len(t, res, len(ops))
+	assert.Equal(t, "database does not exist", res[0].Error)
+	assert.Nil(t, res[1])
+}
+
+func TestCheckIndexes(t *testing.T) {
+	clientDbModel, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
+		"Open_vSwitch":              &OvsType{},
+		"Bridge":                    &BridgeType{},
+		"Flow_Sample_Collector_Set": &FlowSampleCollectorSetType{},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	schema, err := GetSchema()
+	if err != nil {
+		t.Fatal(err)
+	}
+	db := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": clientDbModel})
+	err = db.CreateDatabase("Open_vSwitch", schema)
+	require.NoError(t, err)
+	dbModel, errs := model.NewDatabaseModel(schema, clientDbModel)
+	require.Empty(t, errs)
+
+	bridgeUUID := uuid.NewString()
+	fscsUUID := uuid.NewString()
+	ops := []ovsdb.Operation{
+		{
+			Table:    "Bridge",
+			Op:       ovsdb.OperationInsert,
+			UUIDName: bridgeUUID,
+			Row: ovsdb.Row{
+				"name": "a_bridge_to_nowhere",
+			},
+		},
+		{
+			Table:    "Flow_Sample_Collector_Set",
+			Op:       ovsdb.OperationInsert,
+			UUIDName: fscsUUID,
+			Row: ovsdb.Row{
+				"id":     1,
+				"bridge": ovsdb.UUID{GoUUID: bridgeUUID},
+			},
+		},
+		{
+			Table: "Flow_Sample_Collector_Set",
+			Op:    ovsdb.OperationInsert,
+			Row: ovsdb.Row{
+				"id":     2,
+				"bridge": ovsdb.UUID{GoUUID: bridgeUUID},
+			},
+		},
+	}
+
+	transaction := NewTransaction(dbModel, "Open_vSwitch", db, nil)
+	results, updates := transaction.Transact(ops)
+	require.Len(t, results, len(ops))
+	for _, result := range results {
+		assert.Equal(t, "", result.Error)
+	}
+	err = db.Commit("Open_vSwitch", uuid.New(), updates)
+	require.NoError(t, err)
+
+	tests := []struct {
+		desc        string
+		ops         func() []ovsdb.Operation
+		expectedErr string
+	}{
+		{
+			"Inserting an existing database index should fail",
+			func() []ovsdb.Operation {
+				return []ovsdb.Operation{
+					{
+						Table: "Flow_Sample_Collector_Set",
+						Op:    ovsdb.OperationInsert,
+						Row: ovsdb.Row{
+							"id":     1,
+							"bridge": ovsdb.UUID{GoUUID: bridgeUUID},
+						},
+					},
+				}
+			},
+			"constraint violation",
+		},
+		{
+			"Updating an index to an existing database index should fail",
+			func() []ovsdb.Operation {
+				return []ovsdb.Operation{
+					{
+						Table: "Flow_Sample_Collector_Set",
+						Op:    ovsdb.OperationUpdate,
+						Row: ovsdb.Row{
+							"id":     2,
+							"bridge": ovsdb.UUID{GoUUID: bridgeUUID},
+						},
+						Where: []ovsdb.Condition{
+							ovsdb.NewCondition("_uuid", ovsdb.ConditionEqual, ovsdb.UUID{GoUUID: fscsUUID}),
+						},
+					},
+				}
+			},
+			"constraint violation",
+		},
+		{
+			"Updating an index to an existing transaction index should fail",
+			func() []ovsdb.Operation {
+				return []ovsdb.Operation{
+					{
+						Table: "Flow_Sample_Collector_Set",
+						Op:    ovsdb.OperationInsert,
+						Row: ovsdb.Row{
+							"id":     3,
+							"bridge": ovsdb.UUID{GoUUID: bridgeUUID},
+						},
+					},
+					{
+						Table: "Flow_Sample_Collector_Set",
+						Op:    ovsdb.OperationUpdate,
+						Row: ovsdb.Row{
+							"id":     3,
+							"bridge": ovsdb.UUID{GoUUID: bridgeUUID},
+						},
+						Where: []ovsdb.Condition{
+							ovsdb.NewCondition("_uuid", ovsdb.ConditionEqual, ovsdb.UUID{GoUUID: fscsUUID}),
+						},
+					},
+				}
+			},
+			"constraint violation",
+		},
+		{
+			"Updating an index to an old index that is updated in the same transaction should succeed",
+			func() []ovsdb.Operation {
+				return []ovsdb.Operation{
+					{
+						Table: "Flow_Sample_Collector_Set",
+						Op:    ovsdb.OperationInsert,
+						Row: ovsdb.Row{
+							"id":     1,
+							"bridge": ovsdb.UUID{GoUUID: bridgeUUID},
+						},
+					},
+					{
+						Table: "Flow_Sample_Collector_Set",
+						Op:    ovsdb.OperationUpdate,
+						Row: ovsdb.Row{
+							"id":     3,
+							"bridge": ovsdb.UUID{GoUUID: bridgeUUID},
+						},
+						Where: []ovsdb.Condition{
+							ovsdb.NewCondition("_uuid", ovsdb.ConditionEqual, ovsdb.UUID{GoUUID: fscsUUID}),
+						},
+					},
+				}
+			},
+			"",
+		},
+		{
+			"Updating an index to a old index that is deleted in the same transaction should succeed",
+			func() []ovsdb.Operation {
+				return []ovsdb.Operation{
+					{
+						Table: "Flow_Sample_Collector_Set",
+						Op:    ovsdb.OperationInsert,
+						Row: ovsdb.Row{
+							"id":     1,
+							"bridge": ovsdb.UUID{GoUUID: bridgeUUID},
+						},
+					},
+					{
+						Table: "Flow_Sample_Collector_Set",
+						Op:    ovsdb.OperationDelete,
+						Where: []ovsdb.Condition{
+							ovsdb.NewCondition("_uuid", ovsdb.ConditionEqual, ovsdb.UUID{GoUUID: fscsUUID}),
+						},
+					},
+				}
+			},
+			"",
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.desc, func(t *testing.T) {
+			transaction := NewTransaction(dbModel, "Open_vSwitch", db, nil)
+			ops := tt.ops()
+			results, _ := transaction.Transact(ops)
+			var err string
+			for _, result := range results {
+				if result.Error != "" {
+					err = result.Error
+					break
+				}
+			}
+			require.Equal(t, tt.expectedErr, err, "got a different error than expected")
+			if tt.expectedErr != "" {
+				require.Len(t, results, len(ops)+1)
+			} else {
+				require.Len(t, results, len(ops))
+			}
+		})
+	}
+}
diff --git a/server/server_integration_test.go b/server/server_integration_test.go
index 536fd0f..2b9b4cb 100644
--- a/server/server_integration_test.go
+++ b/server/server_integration_test.go
@@ -5,7 +5,6 @@ import (
 	"fmt"
 	"math/rand"
 	"os"
-	"path/filepath"
 	"reflect"
 	"sync"
 	"testing"
@@ -13,58 +12,26 @@ import (
 
 	"github.com/ovn-org/libovsdb/cache"
 	"github.com/ovn-org/libovsdb/client"
+	"github.com/ovn-org/libovsdb/database"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
-)
-
-// bridgeType is the simplified ORM model of the Bridge table
-type bridgeType struct {
-	UUID         string            `ovsdb:"_uuid"`
-	Name         string            `ovsdb:"name"`
-	DatapathType string            `ovsdb:"datapath_type"`
-	DatapathID   *string           `ovsdb:"datapath_id"`
-	OtherConfig  map[string]string `ovsdb:"other_config"`
-	ExternalIds  map[string]string `ovsdb:"external_ids"`
-	Ports        []string          `ovsdb:"ports"`
-	Status       map[string]string `ovsdb:"status"`
-}
 
-// ovsType is the simplified ORM model of the Bridge table
-type ovsType struct {
-	UUID    string   `ovsdb:"_uuid"`
-	Bridges []string `ovsdb:"bridges"`
-}
-
-func getSchema() (ovsdb.DatabaseSchema, error) {
-	wd, err := os.Getwd()
-	if err != nil {
-		return ovsdb.DatabaseSchema{}, err
-	}
-	path := filepath.Join(wd, "testdata", "ovslite.json")
-	f, err := os.Open(path)
-	if err != nil {
-		return ovsdb.DatabaseSchema{}, err
-	}
-	defer f.Close()
-	schema, err := ovsdb.SchemaFromFile(f)
-	if err != nil {
-		return ovsdb.DatabaseSchema{}, err
-	}
-	return schema, nil
-}
+	. "github.com/ovn-org/libovsdb/test"
+)
 
 func buildTestServerAndClient(t *testing.T) (client.Client, func()) {
 	defDB, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
-		"Open_vSwitch": &ovsType{},
-		"Bridge":       &bridgeType{}})
+		"Open_vSwitch": &OvsType{},
+		"Bridge":       &BridgeType{}})
 	require.Nil(t, err)
 
-	schema, err := getSchema()
+	schema, err := GetSchema()
 	require.Nil(t, err)
 
-	ovsDB := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	ovsDB := database.NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+
 	rand.Seed(time.Now().UnixNano())
 	tmpfile := fmt.Sprintf("/tmp/ovsdb-%d.sock", rand.Intn(10000))
 	defer os.Remove(tmpfile)
@@ -110,7 +77,7 @@ func TestClientServerInsert(t *testing.T) {
 	require.NoError(t, err)
 
 	wallace := "wallace"
-	bridgeRow := &bridgeType{
+	bridgeRow := &BridgeType{
 		Name:         "foo",
 		DatapathType: "bar",
 		DatapathID:   &wallace,
@@ -126,12 +93,12 @@ func TestClientServerInsert(t *testing.T) {
 
 	uuid := reply[0].UUID.GoUUID
 	require.Eventually(t, func() bool {
-		br := &bridgeType{UUID: uuid}
+		br := &BridgeType{UUID: uuid}
 		err := ovs.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 
-	br := &bridgeType{UUID: uuid}
+	br := &BridgeType{UUID: uuid}
 	err = ovs.Get(context.Background(), br)
 	require.NoError(t, err)
 
@@ -145,10 +112,10 @@ func TestClientServerMonitor(t *testing.T) {
 	ovs, close := buildTestServerAndClient(t)
 	defer close()
 
-	ovsRow := &ovsType{
+	ovsRow := &OvsType{
 		UUID: "ovs",
 	}
-	bridgeRow := &bridgeType{
+	bridgeRow := &BridgeType{
 		UUID:        "foo",
 		Name:        "foo",
 		ExternalIds: map[string]string{"go": "awesome", "docker": "made-for-each-other"},
@@ -161,7 +128,7 @@ func TestClientServerMonitor(t *testing.T) {
 	ovs.Cache().AddEventHandler(&cache.EventHandlerFuncs{
 		AddFunc: func(table string, model model.Model) {
 			if table == "Bridge" {
-				br := model.(*bridgeType)
+				br := model.(*BridgeType)
 				assert.Equal(t, bridgeRow.Name, br.Name)
 				assert.Equal(t, bridgeRow.ExternalIds, br.ExternalIds)
 				seenMutex.Lock()
@@ -176,7 +143,7 @@ func TestClientServerMonitor(t *testing.T) {
 		},
 		UpdateFunc: func(table string, old, new model.Model) {
 			if table == "Open_vSwitch" {
-				ov := new.(*ovsType)
+				ov := new.(*OvsType)
 				assert.Equal(t, 1, len(ov.Bridges))
 				seenMutex.Lock()
 				seenMutation = true
@@ -241,7 +208,7 @@ func TestClientServerInsertAndDelete(t *testing.T) {
 	_, err := ovs.MonitorAll(context.Background())
 	require.NoError(t, err)
 
-	bridgeRow := &bridgeType{
+	bridgeRow := &BridgeType{
 		Name:        "foo",
 		ExternalIds: map[string]string{"go": "awesome", "docker": "made-for-each-other"},
 	}
@@ -255,7 +222,7 @@ func TestClientServerInsertAndDelete(t *testing.T) {
 
 	uuid := reply[0].UUID.GoUUID
 	assert.Eventually(t, func() bool {
-		br := &bridgeType{UUID: uuid}
+		br := &BridgeType{UUID: uuid}
 		err := ovs.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
@@ -275,7 +242,7 @@ func TestClientServerInsertDuplicate(t *testing.T) {
 	ovs, close := buildTestServerAndClient(t)
 	defer close()
 
-	bridgeRow := &bridgeType{
+	bridgeRow := &BridgeType{
 		Name:        "foo",
 		ExternalIds: map[string]string{"go": "awesome", "docker": "made-for-each-other"},
 	}
@@ -291,9 +258,9 @@ func TestClientServerInsertDuplicate(t *testing.T) {
 	reply, err = ovs.Transact(context.Background(), ops...)
 	require.Nil(t, err)
 	opErrs, err := ovsdb.CheckOperationResults(reply, ops)
+	require.Nil(t, opErrs)
 	require.Error(t, err)
-	require.Error(t, opErrs[0])
-	require.IsTypef(t, &ovsdb.ConstraintViolation{}, opErrs[0], opErrs[0].Error())
+	require.IsTypef(t, &ovsdb.ConstraintViolation{}, err, err.Error())
 }
 
 func TestClientServerInsertAndUpdate(t *testing.T) {
@@ -303,7 +270,7 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 	_, err := ovs.MonitorAll(context.Background())
 	require.NoError(t, err)
 
-	bridgeRow := &bridgeType{
+	bridgeRow := &BridgeType{
 		Name:        "br-update",
 		ExternalIds: map[string]string{"go": "awesome", "docker": "made-for-each-other"},
 	}
@@ -317,7 +284,7 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 
 	uuid := reply[0].UUID.GoUUID
 	assert.Eventually(t, func() bool {
-		br := &bridgeType{UUID: uuid}
+		br := &BridgeType{UUID: uuid}
 		err := ovs.Get(context.Background(), br)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
@@ -342,7 +309,7 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 	require.NoErrorf(t, err, "%+v", opErrs)
 
 	require.Eventually(t, func() bool {
-		br := &bridgeType{UUID: uuid}
+		br := &BridgeType{UUID: uuid}
 		err = ovs.Get(context.Background(), br)
 		if err != nil {
 			return false
@@ -360,7 +327,7 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 	require.NoErrorf(t, err, "%+v", opErr)
 
 	assert.Eventually(t, func() bool {
-		br := &bridgeType{UUID: uuid}
+		br := &BridgeType{UUID: uuid}
 		err = ovs.Get(context.Background(), br)
 		if err != nil {
 			return false
@@ -368,7 +335,7 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 		return reflect.DeepEqual(br.ExternalIds, bridgeRow.ExternalIds)
 	}, 2*time.Second, 500*time.Millisecond)
 
-	br := &bridgeType{UUID: uuid}
+	br := &BridgeType{UUID: uuid}
 	err = ovs.Get(context.Background(), br)
 	assert.NoError(t, err)
 
@@ -383,7 +350,7 @@ func TestUnsetOptional(t *testing.T) {
 
 	// Create the default bridge which has an optional DatapathID set
 	optional := "optional"
-	br := bridgeType{
+	br := BridgeType{
 		Name:       "br-with-optional",
 		DatapathID: &optional,
 	}
diff --git a/server/server_test.go b/server/server_test.go
index fde1823..900f57c 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -5,10 +5,13 @@ import (
 	"testing"
 
 	"github.com/google/uuid"
+	"github.com/ovn-org/libovsdb/database"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
+
+	. "github.com/ovn-org/libovsdb/test"
 )
 
 func TestExpandNamedUUID(t *testing.T) {
@@ -67,16 +70,16 @@ func TestExpandNamedUUID(t *testing.T) {
 
 func TestOvsdbServerMonitor(t *testing.T) {
 	defDB, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{
-		"Open_vSwitch": &ovsType{},
-		"Bridge":       &bridgeType{}})
+		"Open_vSwitch": &OvsType{},
+		"Bridge":       &BridgeType{}})
 	if err != nil {
 		t.Fatal(err)
 	}
-	schema, err := getSchema()
+	schema, err := GetSchema()
 	if err != nil {
 		t.Fatal(err)
 	}
-	ovsDB := NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
+	ovsDB := database.NewInMemoryDatabase(map[string]model.ClientDBModel{"Open_vSwitch": defDB})
 	dbModel, errs := model.NewDatabaseModel(schema, defDB)
 	require.Empty(t, errs)
 	o, err := NewOvsdbServer(ovsDB, dbModel)
@@ -98,7 +101,7 @@ func TestOvsdbServerMonitor(t *testing.T) {
 	bazUUID := uuid.NewString()
 	quuxUUID := uuid.NewString()
 
-	transaction := o.NewTransaction(dbModel, "Open_vSwitch", o.db)
+	transaction := database.NewTransaction(dbModel, "Open_vSwitch", o.db, &o.logger)
 
 	_, updates := transaction.Insert("Bridge", fooUUID, ovsdb.Row{"name": "foo"})
 	_, update2 := transaction.Insert("Bridge", barUUID, ovsdb.Row{"name": "bar"})
diff --git a/server/testdata/ovslite.json b/server/testdata/ovslite.json
deleted file mode 100644
index d8c402f..0000000
--- a/server/testdata/ovslite.json
+++ /dev/null
@@ -1,81 +0,0 @@
-{
-    "name": "Open_vSwitch",
-    "version": "0.0.1",
-    "tables": {
-        "Open_vSwitch": {
-            "columns": {
-                "bridges": {
-                    "type": {
-                        "key": {
-                            "type": "uuid",
-                            "refTable": "Bridge"
-                        },
-                        "min": 0,
-                        "max": "unlimited"
-                    }
-                }
-            },
-            "isRoot": true,
-            "maxRows": 1
-        },
-        "Bridge": {
-            "columns": {
-                "name": {
-                    "type": "string",
-                    "mutable": false
-                },
-                "datapath_type": {
-                    "type": "string"
-                },
-                "datapath_id": {
-                    "type": {
-                        "key": "string",
-                        "min": 0,
-                        "max": 1
-                    },
-                    "ephemeral": true
-                },
-                "ports": {
-                    "type": {
-                        "key": {
-                            "type": "uuid",
-                            "refTable": "Port"
-                        },
-                        "min": 0,
-                        "max": "unlimited"
-                    }
-                },
-                "status": {
-                    "type": {
-                        "key": "string",
-                        "value": "string",
-                        "min": 0,
-                        "max": "unlimited"
-                    },
-                    "ephemeral": true
-                },
-                "other_config": {
-                    "type": {
-                        "key": "string",
-                        "value": "string",
-                        "min": 0,
-                        "max": "unlimited"
-                    }
-                },
-                "external_ids": {
-                    "type": {
-                        "key": "string",
-                        "value": "string",
-                        "min": 0,
-                        "max": "unlimited"
-                    }
-                }
-            },
-            "indexes": [
-                [
-                    "name"
-                ]
-            ]
-        }
-    }
-}
\ No newline at end of file
diff --git a/test/test_data.go b/test/test_data.go
new file mode 100644
index 0000000..4b1b73a
--- /dev/null
+++ b/test/test_data.go
@@ -0,0 +1,163 @@
+package test
+
+import (
+	"encoding/json"
+
+	"github.com/ovn-org/libovsdb/ovsdb"
+)
+
+const schema = `
+{
+    "name": "Open_vSwitch",
+    "version": "0.0.1",
+    "tables": {
+        "Open_vSwitch": {
+            "columns": {
+                "bridges": {
+                    "type": {
+                        "key": {
+                            "type": "uuid",
+                            "refTable": "Bridge"
+                        },
+                        "min": 0,
+                        "max": "unlimited"
+                    }
+                }
+            },
+            "isRoot": true,
+            "maxRows": 1
+        },
+        "Bridge": {
+            "columns": {
+                "name": {
+                    "type": "string",
+                    "mutable": false
+                },
+                "datapath_type": {
+                    "type": "string"
+                },
+                "datapath_id": {
+                    "type": {
+                        "key": "string",
+                        "min": 0,
+                        "max": 1
+                    },
+                    "ephemeral": true
+                },
+                "ports": {
+                    "type": {
+                        "key": {
+                            "type": "uuid",
+                            "refTable": "Port"
+                        },
+                        "min": 0,
+                        "max": "unlimited"
+                    }
+                },
+                "status": {
+                    "type": {
+                        "key": "string",
+                        "value": "string",
+                        "min": 0,
+                        "max": "unlimited"
+                    },
+                    "ephemeral": true
+                },
+                "other_config": {
+                    "type": {
+                        "key": "string",
+                        "value": "string",
+                        "min": 0,
+                        "max": "unlimited"
+                    }
+                },
+                "external_ids": {
+                    "type": {
+                        "key": "string",
+                        "value": "string",
+                        "min": 0,
+                        "max": "unlimited"
+                    }
+                }
+            },
+            "indexes": [
+                [
+                    "name"
+                ]
+            ]
+        },
+        "Flow_Sample_Collector_Set": {
+            "columns": {
+                "id": {
+                    "type": {
+                        "key": {
+                            "type": "integer",
+                            "minInteger": 0,
+                            "maxInteger": 4294967295
+                        },
+                        "min": 1,
+                        "max": 1
+                    }
+                },
+                "bridge": {
+                    "type": {
+                        "key": {
+                            "type": "uuid",
+                            "refTable": "Bridge"
+                        },
+                        "min": 1,
+                        "max": 1
+                    }
+                },
+                "external_ids": {
+                    "type": {
+                        "key": "string",
+                        "value": "string",
+                        "min": 0,
+                        "max": "unlimited"
+                    }
+                }
+            },
+            "isRoot": true,
+            "indexes": [
+                [
+                    "id",
+                    "bridge"
+                ]
+            ]
+        }
+    }
+}
+`
+
+// BridgeType is the simplified ORM model of the Bridge table
+type BridgeType struct {
+	UUID         string            `ovsdb:"_uuid"`
+	Name         string            `ovsdb:"name"`
+	DatapathType string            `ovsdb:"datapath_type"`
+	DatapathID   *string           `ovsdb:"datapath_id"`
+	OtherConfig  map[string]string `ovsdb:"other_config"`
+	ExternalIds  map[string]string `ovsdb:"external_ids"`
+	Ports        []string          `ovsdb:"ports"`
+	Status       map[string]string `ovsdb:"status"`
+}
+
+// OvsType is the simplified ORM model of the Bridge table
+type OvsType struct {
+	UUID    string   `ovsdb:"_uuid"`
+	Bridges []string `ovsdb:"bridges"`
+}
+
+type FlowSampleCollectorSetType struct {
+	UUID        string            `ovsdb:"_uuid"`
+	Bridge      string            `ovsdb:"bridge"`
+	ExternalIDs map[string]string `ovsdb:"external_ids"`
+	ID          int               `ovsdb:"id"`
+	IPFIX       *string           // `ovsdb:"ipfix"`
+}
+
+func GetSchema() (ovsdb.DatabaseSchema, error) {
+	var dbSchema ovsdb.DatabaseSchema
+	err := json.Unmarshal([]byte(schema), &dbSchema)
+	return dbSchema, err
+}