Repository: ovn-org/libovsdb
Pull request #123 (2021-05-21 18:38:44)

Increase Update Performance
This PR builds on #122 

1. Simplifies the marshal/unmarshal of the structs in the `update` code path (TableUpdates, TableUpdate)
1. Adds some missing fields in the `Operation` struct
1. Adds an `Echo` method on the `client` for calling the server
1. Ensures that our `update` method response contains either an error or a result to avoid upsetting some RPC clients.
1. Uses partial unmarshalling via `json.RawMessage` in the `update` arguments so we can avoid having to marshal/unmarshal to get the arguments in to the right type
1. Re-writes the benchmarks to use `[]byte`


Comparison of benchmarks:
```
name       old time/op  new time/op  delta
Update1-8  43.0µs ± 1%  14.6µs ± 1%  -65.96%  (p=0.000 n=3+3)
Update2-8  59.2µs ± 3%  21.6µs ± 2%  -63.59%  (p=0.000 n=3+3)
Update3-8  77.3µs ± 1%  28.6µs ± 1%  -63.02%  (p=0.000 n=3+3)
Update5-8   108µs ± 2%    43µs ± 2%  -60.71%  (p=0.000 n=3+3)
Update8-8   157µs ± 2%    64µs ± 1%  -59.29%  (p=0.000 n=3+3)
```

In terms of real world performance, with `stress` we're ~6 seconds better on 3000 inserts

**Before**
```
$ go run ./cmd/stress -verbose -ninserts 3000 -ovsdb tcp:localhost:49154
rootUUID is 2333c4e6-5de6-4c8f-a7ba-3f62557d3957

Summary:
        Insertions: 3000
        Deletions: 0
go run ./cmd/stress -verbose -ninserts 3000 -ovsdb tcp:localhost:49154  87.58s user 2.94s system 67% cpu 2:14.57 total
```

*After*
```
$ time go run ./cmd/stress -verbose -ninserts 3000 -ovsdb tcp:localhost:49153 
rootUUID is 28ddc871-f67a-405c-ac47-7beb6b94526a
Summary:
        Insertions: 3000
        Deletions: 0
go run ./cmd/stress -verbose -ninserts 3000 -ovsdb tcp:localhost:49153  78.19s user 2.54s system 63% cpu 2:08.10 total
```

Fixes: #109 

coveralls: ## Pull Request Test Coverage Report for [Build 864803620](https://coveralls.io/builds/39901514)

* **78** of **127**   **(61.42%)**  changed or added relevant lines in **8** files are covered.
* **2** unchanged lines in **1** file lost coverage.
* Overall coverage decreased (**-1.08%**) to **71.367%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [ovsdb/row.go](https://coveralls.io/builds/39901514/source?filename=ovsdb%2Frow.go#L25) | 0 | 3 | 0.0%
| [ovsdb/rpc.go](https://coveralls.io/builds/39901514/source?filename=ovsdb%2Frpc.go#L4) | 0 | 3 | 0.0%
| [ovsdb/updates.go](https://coveralls.io/builds/39901514/source?filename=ovsdb%2Fupdates.go#L18) | 45 | 53 | 84.91%
| [client/client.go](https://coveralls.io/builds/39901514/source?filename=client%2Fclient.go#L193) | 11 | 22 | 50.0%
| [ovsdb/mutation.go](https://coveralls.io/builds/39901514/source?filename=ovsdb%2Fmutation.go#L43) | 11 | 35 | 31.43%
<!-- | **Total:** | **78** | **127** | **61.42%** | -->

|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [client/client.go](https://coveralls.io/builds/39901514/source?filename=client%2Fclient.go#L194) | 2 | 69.51% |
<!-- | **Total:** | **2** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/39901514/badge)](https://coveralls.io/builds/39901514) |
| :-- | --: |
| Change from base [Build 859723497](https://coveralls.io/builds/39847414): |  -1.08% |
| Covered Lines: | 1812 |
| Relevant Lines: | 2539 |

---
##### 💛  - [Coveralls](https://coveralls.io)


Patch:
diff --git a/client/api.go b/client/api.go
index 9202d3f..646971b 100644
--- a/client/api.go
+++ b/client/api.go
@@ -295,7 +295,7 @@ func (a api) Create(models ...model.Model) ([]ovsdb.Operation, error) {
 
 // Mutate returns the operations needed to transform the one Model into another one
 func (a api) Mutate(model model.Model, mutationObjs ...model.Mutation) ([]ovsdb.Operation, error) {
-	var mutations []interface{}
+	var mutations []ovsdb.Mutation
 	var operations []ovsdb.Operation
 
 	if len(mutationObjs) < 1 {
@@ -328,7 +328,7 @@ func (a api) Mutate(model model.Model, mutationObjs ...model.Mutation) ([]ovsdb.
 		if err != nil {
 			return nil, err
 		}
-		mutations = append(mutations, mutation)
+		mutations = append(mutations, *mutation)
 	}
 	for _, condition := range conditions {
 		operations = append(operations,
diff --git a/client/cache.go b/client/cache.go
index 24d656c..847da59 100644
--- a/client/cache.go
+++ b/client/cache.go
@@ -143,7 +143,7 @@ func (t *TableCache) Tables() []string {
 // Update implements the update method of the NotificationHandler interface
 // this populates the cache with new updates
 func (t *TableCache) Update(context interface{}, tableUpdates ovsdb.TableUpdates) {
-	if len(tableUpdates.Updates) == 0 {
+	if len(tableUpdates) == 0 {
 		return
 	}
 	t.populate(tableUpdates)
@@ -170,7 +170,7 @@ func (t *TableCache) populate(tableUpdates ovsdb.TableUpdates) {
 	t.cacheMutex.Lock()
 	defer t.cacheMutex.Unlock()
 	for table := range t.dbModel.Types() {
-		updates, ok := tableUpdates.Updates[table]
+		updates, ok := tableUpdates[table]
 		if !ok {
 			continue
 		}
@@ -180,16 +180,16 @@ func (t *TableCache) populate(tableUpdates ovsdb.TableUpdates) {
 			tCache = t.cache[table]
 		}
 		tCache.mutex.Lock()
-		for uuid, row := range updates.Rows {
-			if !reflect.DeepEqual(row.New, ovsdb.Row{}) {
-				newModel, err := t.createModel(table, &row.New, uuid)
+		for uuid, row := range updates {
+			if row.New != nil {
+				newModel, err := t.createModel(table, row.New, uuid)
 				if err != nil {
 					panic(err)
 				}
 				if existing, ok := tCache.cache[uuid]; ok {
 					if !reflect.DeepEqual(newModel, existing) {
 						tCache.cache[uuid] = newModel
-						oldModel, err := t.createModel(table, &row.Old, uuid)
+						oldModel, err := t.createModel(table, row.Old, uuid)
 						if err != nil {
 							panic(err)
 						}
@@ -202,7 +202,7 @@ func (t *TableCache) populate(tableUpdates ovsdb.TableUpdates) {
 				t.eventProcessor.AddEvent(addEvent, table, nil, newModel)
 				continue
 			} else {
-				oldModel, err := t.createModel(table, &row.Old, uuid)
+				oldModel, err := t.createModel(table, row.Old, uuid)
 				if err != nil {
 					panic(err)
 				}
diff --git a/client/client.go b/client/client.go
index 6205939..c833ce5 100644
--- a/client/client.go
+++ b/client/client.go
@@ -92,8 +92,8 @@ func newRPC2Client(conn net.Conn, database *model.DBModel) (*OvsdbClient, error)
 	ovs.rpcClient.Handle("echo", func(_ *rpc2.Client, args []interface{}, reply *[]interface{}) error {
 		return ovs.echo(args, reply)
 	})
-	ovs.rpcClient.Handle("update", func(_ *rpc2.Client, args []interface{}, _ *[]interface{}) error {
-		return ovs.update(args)
+	ovs.rpcClient.Handle("update", func(_ *rpc2.Client, args []json.RawMessage, reply *[]interface{}) error {
+		return ovs.update(args, reply)
 	})
 	go ovs.rpcClient.Run()
 	go ovs.handleDisconnectNotification()
@@ -188,36 +188,27 @@ func (ovs *OvsdbClient) echo(args []interface{}, reply *[]interface{}) error {
 }
 
 // RFC 7047 : Update Notification Section 4.1.6
-// Processing "params": [<json-value>, <table-updates>]
-func (ovs *OvsdbClient) update(params []interface{}) error {
-	if len(params) < 2 {
-		return fmt.Errorf("invalid update message")
+func (ovs *OvsdbClient) update(args []json.RawMessage, reply *[]interface{}) error {
+	var value string
+	if len(args) > 2 {
+		return fmt.Errorf("update requires exactly 2 args")
 	}
-	// Ignore params[0] as we dont use the <json-value> currently for comparison
-
-	raw, ok := params[1].(map[string]interface{})
-	if !ok {
-		return fmt.Errorf("invalid update message")
-	}
-	var rowUpdates map[string]map[string]ovsdb.RowUpdate
-
-	b, err := json.Marshal(raw)
+	err := json.Unmarshal(args[0], &value)
 	if err != nil {
 		return err
 	}
-	err = json.Unmarshal(b, &rowUpdates)
+	var updates ovsdb.TableUpdates
+	err = json.Unmarshal(args[1], &updates)
 	if err != nil {
 		return err
 	}
-
 	// Update the local DB cache with the tableUpdates
-	tableUpdates := getTableUpdatesFromRawUnmarshal(rowUpdates)
 	ovs.handlersMutex.Lock()
 	defer ovs.handlersMutex.Unlock()
 	for _, handler := range ovs.handlers {
-		handler.Update(params[0], tableUpdates)
+		handler.Update(value, updates)
 	}
-
+	*reply = []interface{}{}
 	return nil
 }
 
@@ -303,11 +294,7 @@ func (ovs OvsdbClient) Monitor(jsonContext interface{}, requests map[string]ovsd
 	var reply ovsdb.TableUpdates
 
 	args := ovsdb.NewMonitorArgs(ovs.Schema.Name, jsonContext, requests)
-
-	// This totally sucks. Refer to golang JSON issue #6213
-	var response map[string]map[string]ovsdb.RowUpdate
-	err := ovs.rpcClient.Call("monitor", args, &response)
-	reply = getTableUpdatesFromRawUnmarshal(response)
+	err := ovs.rpcClient.Call("monitor", args, &reply)
 	if err != nil {
 		return err
 	}
@@ -315,14 +302,18 @@ func (ovs OvsdbClient) Monitor(jsonContext interface{}, requests map[string]ovsd
 	return nil
 }
 
-func getTableUpdatesFromRawUnmarshal(raw map[string]map[string]ovsdb.RowUpdate) ovsdb.TableUpdates {
-	var tableUpdates ovsdb.TableUpdates
-	tableUpdates.Updates = make(map[string]ovsdb.TableUpdate)
-	for table, update := range raw {
-		tableUpdate := ovsdb.TableUpdate{Rows: update}
-		tableUpdates.Updates[table] = tableUpdate
+// Echo tests the liveness of the OVSDB connetion
+func (ovs *OvsdbClient) Echo() error {
+	args := ovsdb.NewEchoArgs()
+	var reply []interface{}
+	err := ovs.rpcClient.Call("echo", args, &reply)
+	if err != nil {
+		return err
 	}
-	return tableUpdates
+	if !reflect.DeepEqual(args, reply) {
+		return fmt.Errorf("incorrect server response: %v, %v", args, reply)
+	}
+	return nil
 }
 
 func (ovs *OvsdbClient) clearConnection() {
diff --git a/mapper/mapper.go b/mapper/mapper.go
index d490090..dd958d9 100644
--- a/mapper/mapper.go
+++ b/mapper/mapper.go
@@ -286,7 +286,7 @@ func (m Mapper) NewCondition(tableName string, data interface{}, field interface
 
 // newMutation creates a RFC7047 mutation object based on an ORM object and the mutation fields (in native format)
 // It takes care of field validation against the column type
-func (m Mapper) NewMutation(tableName string, data interface{}, column string, mutator ovsdb.Mutator, value interface{}) ([]interface{}, error) {
+func (m Mapper) NewMutation(tableName string, data interface{}, column string, mutator ovsdb.Mutator, value interface{}) (*ovsdb.Mutation, error) {
 	table := m.Schema.Table(tableName)
 	if table == nil {
 		return nil, newErrNoTable(tableName)
@@ -325,7 +325,7 @@ func (m Mapper) NewMutation(tableName string, data interface{}, column string, m
 		}
 	}
 
-	return []interface{}{column, mutator, ovsValue}, nil
+	return &ovsdb.Mutation{Column: column, Mutator: mutator, Value: ovsValue}, nil
 }
 
 // equalIndexes returns whether both models are equal from the DB point of view
diff --git a/ovsdb/mutation.go b/ovsdb/mutation.go
new file mode 100644
index 0000000..64441b4
--- /dev/null
+++ b/ovsdb/mutation.go
@@ -0,0 +1,76 @@
+package ovsdb
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+type Mutator string
+
+const (
+	MutateOperationDelete    Mutator = "delete"
+	MutateOperationInsert    Mutator = "insert"
+	MutateOperationAdd       Mutator = "+="
+	MutateOperationSubstract Mutator = "-="
+	MutateOperationMultiply  Mutator = "*="
+	MutateOperationDivide    Mutator = "/="
+	MutateOperationModulo    Mutator = "%="
+)
+
+// Mutation is described in RFC 7047: 5.1
+type Mutation struct {
+	Column  string
+	Mutator Mutator
+	Value   interface{}
+}
+
+// NewMutation returns a new mutation
+func NewMutation(column string, mutator Mutator, value interface{}) *Mutation {
+	return &Mutation{
+		Column:  column,
+		Mutator: mutator,
+		Value:   value,
+	}
+}
+
+// MarshalJSON marshals a mutation to a 3 element JSON array
+func (m Mutation) MarshalJSON() ([]byte, error) {
+	v := []interface{}{m.Column, m.Mutator, m.Value}
+	return json.Marshal(v)
+}
+
+// UnmarshalJSON converts a 3 element JSON array to a Mutation
+func (m Mutation) UnmarshalJSON(b []byte) error {
+	var v []interface{}
+	err := json.Unmarshal(b, &v)
+	if err != nil {
+		return err
+	}
+	if len(v) != 3 {
+		return fmt.Errorf("expected a 3 element json array. there are %d elements", len(v))
+	}
+	ok := false
+	m.Column, ok = v[0].(string)
+	if !ok {
+		return fmt.Errorf("expected column name %v to be a valid string", v[0])
+	}
+	mutatorString, ok := v[1].(string)
+	if !ok {
+		return fmt.Errorf("expected mutator %v to be a valid string", v[1])
+	}
+	mutator := Mutator(mutatorString)
+	switch mutator {
+	case MutateOperationDelete:
+	case MutateOperationInsert:
+	case MutateOperationAdd:
+	case MutateOperationSubstract:
+	case MutateOperationMultiply:
+	case MutateOperationDivide:
+	case MutateOperationModulo:
+		m.Mutator = mutator
+	default:
+		return fmt.Errorf("%s is not a valid mutator", mutator)
+	}
+	m.Value = v[2]
+	return nil
+}
diff --git a/ovsdb/notation.go b/ovsdb/notation.go
index 338339a..ddb2ffd 100644
--- a/ovsdb/notation.go
+++ b/ovsdb/notation.go
@@ -1,6 +1,21 @@
 package ovsdb
 
-import "encoding/json"
+import (
+	"encoding/json"
+)
+
+const (
+	OperationInsert  = "insert"
+	OperationSelect  = "select"
+	OperationUpdate  = "update"
+	OperationMutate  = "mutate"
+	OperationDelete  = "delete"
+	OperationWait    = "wait"
+	OperationCommit  = "commit"
+	OperationAbort   = "abort"
+	OperationComment = "comment"
+	OperationAssert  = "assert"
+)
 
 // Operation represents an operation according to RFC7047 section 5.2
 type Operation struct {
@@ -9,10 +24,13 @@ type Operation struct {
 	Row       map[string]interface{}   `json:"row,omitempty"`
 	Rows      []map[string]interface{} `json:"rows,omitempty"`
 	Columns   []string                 `json:"columns,omitempty"`
-	Mutations []interface{}            `json:"mutations,omitempty"`
+	Mutations []Mutation               `json:"mutations,omitempty"`
 	Timeout   int                      `json:"timeout,omitempty"`
 	Where     []Condition              `json:"where,omitempty"`
 	Until     string                   `json:"until,omitempty"`
+	Durable   *bool                    `json:"durable,omitempty"`
+	Comment   *string                  `json:"comment,omitempty"`
+	Lock      *string                  `json:"lock,omitempty"`
 	UUIDName  string                   `json:"uuid-name,omitempty"`
 }
 
@@ -57,37 +75,12 @@ type MonitorRequest struct {
 	Select  *MonitorSelect `json:"select,omitempty"`
 }
 
-// TableUpdates is a collection of TableUpdate entries
-// We cannot use TableUpdates directly by json encoding by inlining the TableUpdate Map
-// structure till GoLang issue #6213 makes it.
-// The only option is to go with raw map[string]map[string]interface{} option :-( that sucks !
-// Refer to client.go : MonitorAll() function for more details
-type TableUpdates struct {
-	Updates map[string]TableUpdate `json:"updates"`
-}
-
-// TableUpdate represents a table update according to RFC7047
-type TableUpdate struct {
-	Rows map[string]RowUpdate `json:"rows"`
-}
-
-// RowUpdate represents a row update according to RFC7047
-type RowUpdate struct {
-	New Row `json:"new,omitempty"`
-	Old Row `json:"old,omitempty"`
-}
-
 // OvsdbError is an OVS Error Condition
 type OvsdbError struct {
 	Error   string `json:"error"`
 	Details string `json:"details,omitempty"`
 }
 
-// NewMutation creates a new mutation as specified in RFC7047
-func NewMutation(column string, mutator string, value interface{}) []interface{} {
-	return []interface{}{column, mutator, value}
-}
-
 // TransactResponse represents the response to a Transact Operation
 type TransactResponse struct {
 	Result []OperationResult `json:"result"`
@@ -110,7 +103,6 @@ func ovsSliceToGoNotation(val interface{}) (interface{}, error) {
 		if err != nil {
 			return nil, err
 		}
-
 		switch sl[0] {
 		case "uuid", "named-uuid":
 			var uuid UUID
@@ -129,15 +121,3 @@ func ovsSliceToGoNotation(val interface{}) (interface{}, error) {
 	}
 	return val, nil
 }
-
-type Mutator string
-
-const (
-	MutateOperationDelete    Mutator = "delete"
-	MutateOperationInsert    Mutator = "insert"
-	MutateOperationAdd       Mutator = "+="
-	MutateOperationSubstract Mutator = "-="
-	MutateOperationMultiply  Mutator = "*="
-	MutateOperationDivide    Mutator = "/="
-	MutateOperationModulo    Mutator = "%="
-)
diff --git a/ovsdb/row.go b/ovsdb/row.go
index e770ea9..57f722b 100644
--- a/ovsdb/row.go
+++ b/ovsdb/row.go
@@ -22,6 +22,10 @@ func (r *Row) UnmarshalJSON(b []byte) (err error) {
 	return err
 }
 
+func (r Row) MarshalJSON() ([]byte, error) {
+	return json.Marshal(r.Fields)
+}
+
 // ResultRow is an properly unmarshalled row returned by Transact
 type ResultRow map[string]interface{}
 
diff --git a/ovsdb/rpc.go b/ovsdb/rpc.go
index 4b66210..da31366 100644
--- a/ovsdb/rpc.go
+++ b/ovsdb/rpc.go
@@ -1,5 +1,10 @@
 package ovsdb
 
+// NewEchoArgs creates a new set of arguments for an echo RPC
+func NewEchoArgs() []interface{} {
+	return []interface{}{"libovsdb echo"}
+}
+
 // NewGetSchemaArgs creates a new set of arguments for a get_schemas RPC
 func NewGetSchemaArgs(schema string) []interface{} {
 	return []interface{}{schema}
diff --git a/ovsdb/updates.go b/ovsdb/updates.go
new file mode 100644
index 0000000..8e3b62f
--- /dev/null
+++ b/ovsdb/updates.go
@@ -0,0 +1,85 @@
+package ovsdb
+
+// TableUpdates is an object that maps from a table name to a
+// TableUpdate
+type TableUpdates map[string]TableUpdate
+
+// AddTableUpdate adds a new TableUpdate to a TableUpdates
+func (t TableUpdates) AddTableUpdate(table string, update TableUpdate) {
+	if _, ok := t[table]; !ok {
+		t[table] = update
+	} else {
+		for uuid, row := range update {
+			t[table].AddRowUpdate(uuid, row)
+		}
+	}
+}
+
+func (t TableUpdates) Merge(update TableUpdates) {
+	for k, v := range update {
+		t.AddTableUpdate(k, v)
+	}
+}
+
+// TableUpdate is an object that maps from the row's UUID to a
+// RowUpdate
+type TableUpdate map[string]*RowUpdate
+
+func (t TableUpdate) AddRowUpdate(uuid string, update *RowUpdate) {
+	if _, ok := t[uuid]; !ok {
+		t[uuid] = update
+	} else {
+		t[uuid].Merge(update)
+	}
+}
+
+// RowUpdate represents a row update according to RFC7047
+type RowUpdate struct {
+	New *Row `json:"new,omitempty"`
+	Old *Row `json:"old,omitempty"`
+}
+
+// Insert returns true if this is an update for an insert operation
+func (r RowUpdate) Insert() bool {
+	return r.New != nil && r.Old == nil
+}
+
+// Insert returns true if this is an update for a modify operation
+func (r RowUpdate) Modify() bool {
+	return r.New != nil && r.Old != nil
+}
+
+// Insert returns true if this is an update for a delete operation
+func (r RowUpdate) Delete() bool {
+	return r.New == nil && r.Old != nil
+}
+
+func (r *RowUpdate) Merge(new *RowUpdate) {
+	// old.Delete() cannot be true, as we can't modify a row after it was deleted
+	if r.Delete() {
+		return
+	}
+	// we should never get two insert events
+	if r.Insert() && new.Insert() {
+		return
+	}
+	if r.Insert() && new.Modify() {
+		r.Old = nil
+		r.New = new.New
+		return
+	}
+	if r.Insert() && new.Delete() {
+		r.Old = new.Old
+		r.New = nil
+		return
+	}
+	if r.Modify() && new.Modify() {
+		r.New = new.New
+		return
+	}
+	if r.Modify() && new.Delete() {
+		r.Old = new.Old
+		r.New = nil
+		return
+	}
+}


Test patch:
diff --git a/client/api_test.go b/client/api_test.go
index 6452c04..e1f0526 100644
--- a/client/api_test.go
+++ b/client/api_test.go
@@ -662,7 +662,7 @@ func TestAPIMutate(t *testing.T) {
 				{
 					Op:        opMutate,
 					Table:     "Logical_Switch_Port",
-					Mutations: []interface{}{[]interface{}{"tag", ovsdb.MutateOperationInsert, testOvsSet(t, []int{5})}},
+					Mutations: []ovsdb.Mutation{{Column: "tag", Mutator: ovsdb.MutateOperationInsert, Value: testOvsSet(t, []int{5})}},
 					Where:     []ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: aUUID0}}},
 				},
 			},
@@ -686,7 +686,7 @@ func TestAPIMutate(t *testing.T) {
 				{
 					Op:        opMutate,
 					Table:     "Logical_Switch_Port",
-					Mutations: []interface{}{[]interface{}{"external_ids", ovsdb.MutateOperationDelete, testOvsSet(t, []string{"foo"})}},
+					Mutations: []ovsdb.Mutation{{Column: "external_ids", Mutator: ovsdb.MutateOperationDelete, Value: testOvsSet(t, []string{"foo"})}},
 					Where:     []ovsdb.Condition{{Column: "name", Function: ovsdb.ConditionEqual, Value: "lsp2"}},
 				},
 			},
@@ -710,7 +710,7 @@ func TestAPIMutate(t *testing.T) {
 				{
 					Op:        opMutate,
 					Table:     "Logical_Switch_Port",
-					Mutations: []interface{}{[]interface{}{"external_ids", ovsdb.MutateOperationInsert, testOvsMap(t, map[string]string{"bar": "baz"})}},
+					Mutations: []ovsdb.Mutation{{Column: "external_ids", Mutator: ovsdb.MutateOperationInsert, Value: testOvsMap(t, map[string]string{"bar": "baz"})}},
 					Where:     []ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: aUUID2}}},
 				},
 			},
@@ -734,13 +734,13 @@ func TestAPIMutate(t *testing.T) {
 				{
 					Op:        opMutate,
 					Table:     "Logical_Switch_Port",
-					Mutations: []interface{}{[]interface{}{"external_ids", ovsdb.MutateOperationInsert, testOvsMap(t, map[string]string{"bar": "baz"})}},
+					Mutations: []ovsdb.Mutation{{Column: "external_ids", Mutator: ovsdb.MutateOperationInsert, Value: testOvsMap(t, map[string]string{"bar": "baz"})}},
 					Where:     []ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: aUUID0}}},
 				},
 				{
 					Op:        opMutate,
 					Table:     "Logical_Switch_Port",
-					Mutations: []interface{}{[]interface{}{"external_ids", ovsdb.MutateOperationInsert, testOvsMap(t, map[string]string{"bar": "baz"})}},
+					Mutations: []ovsdb.Mutation{{Column: "external_ids", Mutator: ovsdb.MutateOperationInsert, Value: testOvsMap(t, map[string]string{"bar": "baz"})}},
 					Where:     []ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: aUUID1}}},
 				},
 			},
diff --git a/client/cache_test.go b/client/cache_test.go
index e65cb31..a210fad 100644
--- a/client/cache_test.go
+++ b/client/cache_test.go
@@ -310,17 +310,13 @@ func TestTableCache_populate(t *testing.T) {
 	tc, err := newTableCache(&schema, db)
 	assert.Nil(t, err)
 
-	testRow := ovsdb.Row{Fields: map[string]interface{}{"_uuid": "test", "foo": "bar"}}
+	testRow := &ovsdb.Row{Fields: map[string]interface{}{"_uuid": "test", "foo": "bar"}}
 	testRowModel := &testModel{UUID: "test", Foo: "bar"}
 	updates := ovsdb.TableUpdates{
-		Updates: map[string]ovsdb.TableUpdate{
-			"Open_vSwitch": {
-				Rows: map[string]ovsdb.RowUpdate{
-					"test": {
-						Old: ovsdb.Row{},
-						New: testRow,
-					},
-				},
+		"Open_vSwitch": {
+			"test": &ovsdb.RowUpdate{
+				Old: nil,
+				New: testRow,
 			},
 		},
 	}
@@ -330,19 +326,11 @@ func TestTableCache_populate(t *testing.T) {
 	assert.Equal(t, testRowModel, got)
 
 	t.Log("Update")
-	updatedRow := ovsdb.Row{Fields: map[string]interface{}{"_uuid": "test", "foo": "quux"}}
+	updatedRow := &ovsdb.Row{Fields: map[string]interface{}{"_uuid": "test", "foo": "quux"}}
 	updatedRowModel := &testModel{UUID: "test", Foo: "quux"}
-	updates = ovsdb.TableUpdates{
-		Updates: map[string]ovsdb.TableUpdate{
-			"Open_vSwitch": {
-				Rows: map[string]ovsdb.RowUpdate{
-					"test": {
-						Old: testRow,
-						New: updatedRow,
-					},
-				},
-			},
-		},
+	updates["Open_vSwitch"]["test"] = &ovsdb.RowUpdate{
+		Old: testRow,
+		New: updatedRow,
 	}
 	tc.populate(updates)
 
@@ -350,17 +338,9 @@ func TestTableCache_populate(t *testing.T) {
 	assert.Equal(t, updatedRowModel, got)
 
 	t.Log("Delete")
-	updates = ovsdb.TableUpdates{
-		Updates: map[string]ovsdb.TableUpdate{
-			"Open_vSwitch": {
-				Rows: map[string]ovsdb.RowUpdate{
-					"test": {
-						Old: updatedRow,
-						New: ovsdb.Row{},
-					},
-				},
-			},
-		},
+	updates["Open_vSwitch"]["test"] = &ovsdb.RowUpdate{
+		Old: updatedRow,
+		New: nil,
 	}
 
 	tc.populate(updates)
diff --git a/client/client_test.go b/client/client_test.go
index 9c9a564..cfd1f06 100644
--- a/client/client_test.go
+++ b/client/client_test.go
@@ -1,7 +1,9 @@
 package client
 
 import (
+	"encoding/json"
 	"reflect"
+	"strings"
 	"sync"
 	"testing"
 
@@ -28,110 +30,142 @@ func testOvsMap(t *testing.T, set interface{}) *ovsdb.OvsMap {
 	return oMap
 }
 
-func updateBenchmark(bridges []string, b *testing.B) {
-	bridgeInsert := ovsdb.TableUpdate{
-		Rows: make(map[string]ovsdb.RowUpdate),
-	}
-	for _, br := range bridges {
-		r := newBridgeRow(br)
-		bridgeInsert.Rows[br] = ovsdb.RowUpdate{New: r}
-	}
-	ovsUpdate := ovsdb.TableUpdate{
-		Rows: map[string]ovsdb.RowUpdate{
-			"829f8534-94a8-468e-9176-132738cf260a": {Old: newOvsRow([]string{}), New: newOvsRow(bridges)},
-		},
-	}
-	tu := map[string]interface{}{
-		"Open_vSwitch": ovsUpdate,
-		"Bridge":       bridgeInsert,
-	}
+func updateBenchmark(updates []byte, b *testing.B) {
 	ovs := OvsdbClient{
 		handlers:      []ovsdb.NotificationHandler{},
 		handlersMutex: &sync.Mutex{},
 	}
 	for n := 0; n < b.N; n++ {
-		params := []interface{}{"v1", tu}
-		if len(params) != 2 {
-			b.Fatalf("Params not 2")
-		}
-		err := ovs.update(params)
+		params := []json.RawMessage{[]byte(`"v1"`), updates}
+		var reply []interface{}
+		err := ovs.update(params, &reply)
 		if err != nil {
 			b.Fatal(err)
 		}
 	}
 }
 
-func newBridgeRow(name string) ovsdb.Row {
-	return ovsdb.Row{
-		Fields: map[string]interface{}{
-			"connection_mode":       []string{},
-			"controller":            []string{},
-			"datapath_id":           "blablabla",
-			"datapath_type":         "",
-			"datapath_version":      "",
-			"external_ids":          map[string]string{"foo": "bar"},
-			"fail_mode":             []string{},
-			"flood_vlans":           []string{},
-			"flow_tables":           map[string]string{},
-			"ipfix":                 []string{},
-			"mcast_snooping_enable": false,
-			"mirrors":               []string{},
-			"name":                  name,
-			"netflow":               []string{},
-			"other_config":          map[string]string{"baz": "quux"},
-			"ports":                 []string{},
-			"protocols":             []string{},
-			"rstp_enable":           false,
-			"rstp_status":           map[string]string{},
-			"sflow":                 []string{},
-			"status":                map[string]string{},
-			"stp_enable":            false,
-		},
-	}
+func newBridgeRow(name string) string {
+	return `{
+		"connection_mode": [ "set", [] ],
+		"controller": [ "set", [] ],
+		"datapath_id": "blablabla",
+		"datapath_type": "",
+		"datapath_version": "",
+		"external_ids": [ "map", [["foo","bar"]]],
+		"fail_mode": [ "set", [] ],
+		"flood_vlans": [ "set", [] ],
+		"flow_tables": [ "map", [] ],
+		"ipfix": [ "set", [] ],
+		"mcast_snooping_enable": false,
+		"mirrors": [ "set", [] ],
+		"name": "` + name + `",
+		"netflow": [ "set", [] ],
+		"other_config": [ "map", [["bar","quux"]]],
+		"ports": [ "set", [] ],
+		"protocols": [ "set", [] ],
+		"rstp_enable": false,
+		"rstp_status": [ "map", [] ],
+		"sflow": [ "set", [] ],
+		"status": [ "map", [] ],
+		"stp_enable": false
+	}`
 }
 
-func newOvsRow(bridges []string) ovsdb.Row {
-	return ovsdb.Row{
-		Fields: map[string]interface{}{
-			"bridges":          bridges,
-			"cur_cfg":          0,
-			"datapath_types":   []string{},
-			"datapaths":        map[string]string{},
-			"db_version":       "8.2.0",
-			"dpdk_initialized": false,
-			"dpdk_version":     []string{},
-			"external_ids":     map[string]string{"system-id": "829f8534-94a8-468e-9176-132738cf260a"},
-			"iface_types":      []string{},
-			"manager_options":  "6e4cd5fc-f51a-462a-b3d6-a696af6d7a84",
-			"next_cfg":         0,
-			"other_config":     map[string]string{},
-			"ovs_version":      "2.15.90",
-			"ssl":              []string{},
-			"statistics":       map[string]string{},
-			"system_type":      "docker-ovs",
-			"system_version":   "0.1",
-		},
-	}
+func newOvsRow(bridges ...string) string {
+	return `{
+		"bridges": [ "set", ["` + strings.Join(bridges, `","`) + `"]],
+		"cur_cfg": 0,
+		"datapath_types": [ "set", [] ],
+		"datapaths": [ "map", [] ],
+		"db_version":       "8.2.0",
+		"dpdk_initialized": false,
+		"dpdk_version":     [ "set", [] ],
+		"external_ids":     [ "map", [["system-id","829f8534-94a8-468e-9176-132738cf260a"]]],
+		"iface_types":      [ "set", [] ],
+		"manager_options":  "6e4cd5fc-f51a-462a-b3d6-a696af6d7a84",
+		"next_cfg":         0,
+		"other_config":     [ "map", [] ],
+		"ovs_version":      "2.15.90",
+		"ssl":              [ "set", [] ],
+		"statistics":       [ "map", [] ],
+		"system_type":      "docker-ovs",
+		"system_version":   "0.1"
+	}`
 }
 
 func BenchmarkUpdate1(b *testing.B) {
-	updateBenchmark([]string{"foo"}, b)
+	update := []byte(`{
+		"Open_vSwitch": {
+			"ovs": ` + newOvsRow("foo") + `
+		},
+		"Bridge": {
+			"foo": ` + newBridgeRow("foo") + `
+		}
+	}`)
+	updateBenchmark(update, b)
 }
 
 func BenchmarkUpdate2(b *testing.B) {
-	updateBenchmark([]string{"foo", "bar"}, b)
+	update := []byte(`{
+		"Open_vSwitch": {
+			"ovs": ` + newOvsRow("foo", "bar") + `
+		},
+		"Bridge": {
+			"foo": ` + newBridgeRow("foo") + `,
+			"bar": ` + newBridgeRow("bar") + `
+		}
+	}`)
+	updateBenchmark(update, b)
 }
 
 func BenchmarkUpdate3(b *testing.B) {
-	updateBenchmark([]string{"foo", "bar", "baz"}, b)
+	update := []byte(`{
+		"Open_vSwitch": {
+			"ovs": ` + newOvsRow("foo", "bar", "baz") + `
+		},
+		"Bridge": {
+			"foo": ` + newBridgeRow("foo") + `,
+			"bar": ` + newBridgeRow("bar") + `,
+			"baz": ` + newBridgeRow("baz") + `
+		}
+	}`)
+	updateBenchmark(update, b)
 }
 
 func BenchmarkUpdate5(b *testing.B) {
-	updateBenchmark([]string{"foo", "bar", "baz", "quux", "foofoo"}, b)
+	update := []byte(`{
+		"Open_vSwitch": {
+			"ovs": ` + newOvsRow("foo", "bar", "baz", "quux", "foofoo") + `
+		},
+		"Bridge": {
+			"foo": ` + newBridgeRow("foo") + `,
+			"bar": ` + newBridgeRow("bar") + `,
+			"baz": ` + newBridgeRow("baz") + `,
+			"quux": ` + newBridgeRow("quux") + `,
+			"foofoo": ` + newBridgeRow("foofoo") + `
+		}
+	}`)
+	updateBenchmark(update, b)
 }
 
 func BenchmarkUpdate8(b *testing.B) {
-	updateBenchmark([]string{"foo", "bar", "baz", "quux", "foofoo", "foobar", "foobaz", "fooquux"}, b)
+	update := []byte(`{
+		"Open_vSwitch": {
+			"ovs": ` + newOvsRow("foo", "bar", "baz", "quux", "foofoo", "foobar", "foobaz", "fooquux") + `
+		},
+		"Bridge": {
+			"foo": ` + newBridgeRow("foo") + `,
+			"bar": ` + newBridgeRow("bar") + `,
+			"baz": ` + newBridgeRow("baz") + `,
+			"quux": ` + newBridgeRow("quux") + `,
+			"foofoo": ` + newBridgeRow("foofoo") + `,
+			"foobar": ` + newBridgeRow("foobar") + `,
+			"foobaz": ` + newBridgeRow("foobaz") + `,
+			"fooquux": ` + newBridgeRow("fooquux") + `
+		}
+	}`)
+	updateBenchmark(update, b)
 }
 
 func TestEcho(t *testing.T) {
@@ -155,25 +189,17 @@ func TestUpdate(t *testing.T) {
 		handlers:      []ovsdb.NotificationHandler{},
 		handlersMutex: &sync.Mutex{},
 	}
-	// Update notification should fail for arrays of size < 2
-	err := ovs.update([]interface{}{"hello"})
-	if err == nil {
-		t.Error("Expected: error for a dummy request")
+	var reply []interface{}
+	validUpdate := ovsdb.TableUpdates{
+		"table": {
+			"uuid": &ovsdb.RowUpdate{},
+		},
 	}
-
-	// Update notification should fail if arg[1] is not map[string]map[string]RowUpdate type
-	err = ovs.update([]interface{}{"hello", "gophers"})
-	if err == nil {
-		t.Error("Expected: error for a dummy request")
+	b, err := json.Marshal(validUpdate)
+	if err != nil {
+		t.Fatal(err)
 	}
-
-	// Valid dummy update should pass
-	validUpdate := make(map[string]interface{})
-	validRowUpdate := make(map[string]ovsdb.RowUpdate)
-	validRowUpdate["uuid"] = ovsdb.RowUpdate{}
-	validUpdate["table"] = validRowUpdate
-
-	err = ovs.update([]interface{}{"hello", validUpdate})
+	err = ovs.update([]json.RawMessage{[]byte(`"hello"`), b}, &reply)
 	if err != nil {
 		t.Error(err)
 	}
diff --git a/mapper/mapper_test.go b/mapper/mapper_test.go
index 8e62278..967d72d 100644
--- a/mapper/mapper_test.go
+++ b/mapper/mapper_test.go
@@ -922,7 +922,7 @@ func TestMapperMutation(t *testing.T) {
 		name     string
 		column   string
 		obj      testType
-		expected []interface{}
+		expected *ovsdb.Mutation
 		mutator  ovsdb.Mutator
 		value    interface{}
 		err      bool
@@ -941,7 +941,7 @@ func TestMapperMutation(t *testing.T) {
 			obj:      testType{},
 			mutator:  ovsdb.MutateOperationAdd,
 			value:    1,
-			expected: []interface{}{"int", ovsdb.MutateOperationAdd, 1},
+			expected: ovsdb.NewMutation("int", ovsdb.MutateOperationAdd, 1),
 			err:      false,
 		},
 		{
@@ -950,7 +950,7 @@ func TestMapperMutation(t *testing.T) {
 			obj:      testType{},
 			mutator:  ovsdb.MutateOperationModulo,
 			value:    2,
-			expected: []interface{}{"int", ovsdb.MutateOperationModulo, 2},
+			expected: ovsdb.NewMutation("int", ovsdb.MutateOperationModulo, 2),
 			err:      false,
 		},
 		{
@@ -967,7 +967,7 @@ func TestMapperMutation(t *testing.T) {
 			obj:      testType{},
 			mutator:  ovsdb.MutateOperationInsert,
 			value:    []string{"foo"},
-			expected: []interface{}{"set", ovsdb.MutateOperationInsert, testOvsSet(t, []string{"foo"})},
+			expected: ovsdb.NewMutation("set", ovsdb.MutateOperationInsert, testOvsSet(t, []string{"foo"})),
 			err:      false,
 		},
 		{
@@ -976,7 +976,7 @@ func TestMapperMutation(t *testing.T) {
 			obj:      testType{},
 			mutator:  ovsdb.MutateOperationDelete,
 			value:    []string{"foo"},
-			expected: []interface{}{"set", ovsdb.MutateOperationDelete, testOvsSet(t, []string{"foo"})},
+			expected: ovsdb.NewMutation("set", ovsdb.MutateOperationDelete, testOvsSet(t, []string{"foo"})),
 			err:      false,
 		},
 		{
@@ -985,7 +985,7 @@ func TestMapperMutation(t *testing.T) {
 			obj:      testType{},
 			mutator:  ovsdb.MutateOperationDelete,
 			value:    []string{"foo", "bar"},
-			expected: []interface{}{"map", ovsdb.MutateOperationDelete, testOvsSet(t, []string{"foo", "bar"})},
+			expected: ovsdb.NewMutation("map", ovsdb.MutateOperationDelete, testOvsSet(t, []string{"foo", "bar"})),
 			err:      false,
 		},
 		{
@@ -994,7 +994,7 @@ func TestMapperMutation(t *testing.T) {
 			obj:      testType{},
 			mutator:  ovsdb.MutateOperationInsert,
 			value:    map[string]string{"foo": "bar"},
-			expected: []interface{}{"map", ovsdb.MutateOperationInsert, testOvsMap(t, map[string]string{"foo": "bar"})},
+			expected: ovsdb.NewMutation("map", ovsdb.MutateOperationInsert, testOvsMap(t, map[string]string{"foo": "bar"})),
 			err:      false,
 		},
 	}
diff --git a/ovsdb/updates_test.go b/ovsdb/updates_test.go
new file mode 100644
index 0000000..9d4098c
--- /dev/null
+++ b/ovsdb/updates_test.go
@@ -0,0 +1,192 @@
+package ovsdb
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestAddTableUpdate(t *testing.T) {
+	tests := []struct {
+		name     string
+		initial  TableUpdates
+		table    string
+		update   TableUpdate
+		expected TableUpdates
+	}{
+		{
+			"new table",
+			TableUpdates{},
+			"foo",
+			TableUpdate{},
+			TableUpdates{"foo": TableUpdate{}},
+		},
+		{
+			"existing table",
+			TableUpdates{
+				"foo": {"bar": {Old: nil, New: nil}},
+			},
+			"foo",
+			TableUpdate{"baz": {Old: nil, New: nil}},
+			TableUpdates{
+				"foo": {
+					"bar": {Old: nil, New: nil},
+					"baz": {Old: nil, New: nil},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.initial.AddTableUpdate(tt.table, tt.update)
+			assert.Equal(t, tt.expected, tt.initial)
+		})
+	}
+}
+
+func TestAddRowUpdate(t *testing.T) {
+	tests := []struct {
+		name     string
+		initial  TableUpdate
+		uuid     string
+		update   *RowUpdate
+		expected TableUpdate
+	}{
+		{
+			"new row",
+			TableUpdate{},
+			"foo",
+			&RowUpdate{},
+			TableUpdate{"foo": {}},
+		},
+		{
+			"update existing row",
+			TableUpdate{
+				"foo": {
+					Old: nil,
+					New: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+				},
+			},
+			"foo",
+			&RowUpdate{
+				Old: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+				New: &Row{Fields: map[string]interface{}{"foo": "baz"}},
+			},
+			TableUpdate{
+				"foo": {
+					Old: nil,
+					New: &Row{Fields: map[string]interface{}{"foo": "baz"}},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.initial.AddRowUpdate(tt.uuid, tt.update)
+			assert.Equal(t, tt.expected, tt.initial)
+		})
+	}
+}
+
+func TestAddRowUpdateMerge(t *testing.T) {
+	tests := []struct {
+		name     string
+		initial  *RowUpdate
+		new      *RowUpdate
+		expected *RowUpdate
+	}{
+		{
+			"insert then modify",
+			&RowUpdate{
+				Old: nil,
+				New: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+			},
+			&RowUpdate{
+				Old: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+				New: &Row{Fields: map[string]interface{}{"foo": "baz"}},
+			},
+			&RowUpdate{
+				Old: nil,
+				New: &Row{Fields: map[string]interface{}{"foo": "baz"}},
+			},
+		},
+		{
+			"insert then delete",
+			&RowUpdate{
+				New: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+			},
+			&RowUpdate{
+				Old: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+			},
+			&RowUpdate{
+				Old: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+				New: nil,
+			},
+		},
+		{
+			"modify then delete",
+			&RowUpdate{
+				New: &Row{Fields: map[string]interface{}{"foo": "baz"}},
+				Old: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+			},
+			&RowUpdate{
+				Old: &Row{Fields: map[string]interface{}{"foo": "baz"}},
+			},
+			&RowUpdate{
+				Old: &Row{Fields: map[string]interface{}{"foo": "baz"}},
+				New: nil,
+			},
+		},
+		{
+			"modify then modify",
+			&RowUpdate{
+				New: &Row{Fields: map[string]interface{}{"foo": "baz"}},
+				Old: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+			},
+			&RowUpdate{
+				New: &Row{Fields: map[string]interface{}{"foo": "quux"}},
+				Old: &Row{Fields: map[string]interface{}{"foo": "baz"}},
+			},
+			&RowUpdate{
+				New: &Row{Fields: map[string]interface{}{"foo": "quux"}},
+				Old: &Row{Fields: map[string]interface{}{"foo": "bar"}},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tt.initial.Merge(tt.new)
+			assert.Equal(t, tt.expected, tt.initial)
+		})
+	}
+}
+
+func TestRowUpdateInsert(t *testing.T) {
+	u1 := RowUpdate{Old: nil, New: &Row{}}
+	u2 := RowUpdate{Old: &Row{}, New: &Row{}}
+	u3 := RowUpdate{Old: &Row{}, New: nil}
+
+	assert.True(t, u1.Insert())
+	assert.False(t, u2.Insert())
+	assert.False(t, u3.Insert())
+}
+
+func TestRowUpdateModify(t *testing.T) {
+	u1 := RowUpdate{Old: nil, New: &Row{}}
+	u2 := RowUpdate{Old: &Row{}, New: &Row{}}
+	u3 := RowUpdate{Old: &Row{}, New: nil}
+
+	assert.False(t, u1.Modify())
+	assert.True(t, u2.Modify())
+	assert.False(t, u3.Modify())
+}
+
+func TestRowUpdateDelete(t *testing.T) {
+	u1 := RowUpdate{Old: nil, New: &Row{}}
+	u2 := RowUpdate{Old: &Row{}, New: &Row{}}
+	u3 := RowUpdate{Old: &Row{}, New: nil}
+
+	assert.False(t, u1.Delete())
+	assert.False(t, u2.Delete())
+	assert.True(t, u3.Delete())
+}