Repository: ovn-org/libovsdb
Pull request #222 (2021-08-26 16:47:03)

Feature: Support Conditional Monitoring
Fixes: #187 
Fixes: #228 


Patch:
diff --git a/cache/cache.go b/cache/cache.go
index 9b9c896..01a5aab 100644
--- a/cache/cache.go
+++ b/cache/cache.go
@@ -78,7 +78,7 @@ func (r *RowCache) Row(uuid string) model.Model {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
 	if row, ok := r.cache[uuid]; ok {
-		return row.(model.Model)
+		return model.Clone(row)
 	}
 	return nil
 }
@@ -144,7 +144,7 @@ func (r *RowCache) Create(uuid string, m model.Model, checkIndexes bool) error {
 			r.indexes[k1][k2] = v2
 		}
 	}
-	r.cache[uuid] = m
+	r.cache[uuid] = model.Clone(m)
 	return nil
 }
 
@@ -155,7 +155,7 @@ func (r *RowCache) Update(uuid string, m model.Model, checkIndexes bool) error {
 	if _, ok := r.cache[uuid]; !ok {
 		return fmt.Errorf("row %s does not exist", uuid)
 	}
-	oldRow := r.cache[uuid]
+	oldRow := model.Clone(r.cache[uuid])
 	oldInfo, err := mapper.NewInfo(&r.schema, oldRow)
 	if err != nil {
 		return err
@@ -213,7 +213,7 @@ func (r *RowCache) Update(uuid string, m model.Model, checkIndexes bool) error {
 			delete(r.indexes[k1], k2)
 		}
 	}
-	r.cache[uuid] = m
+	r.cache[uuid] = model.Clone(m)
 	return nil
 }
 
@@ -487,6 +487,15 @@ func (t *TableCache) Update(context interface{}, tableUpdates ovsdb.TableUpdates
 	t.Populate(tableUpdates)
 }
 
+// Update2 implements the update method of the NotificationHandler interface
+// this populates the cache with new updates
+func (t *TableCache) Update2(context interface{}, tableUpdates ovsdb.TableUpdates2) {
+	if len(tableUpdates) == 0 {
+		return
+	}
+	t.Populate2(tableUpdates)
+}
+
 // Locked implements the locked method of the NotificationHandler interface
 func (t *TableCache) Locked([]interface{}) {
 }
@@ -550,6 +559,66 @@ func (t *TableCache) Populate(tableUpdates ovsdb.TableUpdates) {
 	}
 }
 
+// Populate2 adds data to the cache and places an event on the channel
+func (t *TableCache) Populate2(tableUpdates ovsdb.TableUpdates2) {
+	t.mutex.Lock()
+	defer t.mutex.Unlock()
+	for table := range t.dbModel.Types() {
+		updates, ok := tableUpdates[table]
+		if !ok {
+			continue
+		}
+		tCache := t.cache[table]
+		for uuid, row := range updates {
+			switch {
+			case row.Initial != nil:
+				m, err := t.CreateModel(table, row.Initial, uuid)
+				if err != nil {
+					panic(err)
+				}
+				if err := tCache.Create(uuid, m, false); err != nil {
+					panic(err)
+				}
+				t.eventProcessor.AddEvent(addEvent, table, nil, m)
+			case row.Insert != nil:
+				m, err := t.CreateModel(table, row.Insert, uuid)
+				if err != nil {
+					panic(err)
+				}
+				if err := tCache.Create(uuid, m, false); err != nil {
+					panic(err)
+				}
+				t.eventProcessor.AddEvent(addEvent, table, nil, m)
+			case row.Modify != nil:
+				modified := tCache.Row(uuid)
+				if modified == nil {
+					panic(fmt.Errorf("row with uuid %s does not exist", uuid))
+				}
+				err := t.ApplyModifications(table, modified, *row.Modify)
+				if err != nil {
+					panic(err)
+				}
+				existing := tCache.Row(uuid)
+				if !reflect.DeepEqual(modified, existing) {
+					if err := tCache.Update(uuid, modified, false); err != nil {
+						panic(err)
+					}
+					t.eventProcessor.AddEvent(updateEvent, table, existing, modified)
+				}
+			case row.Delete != nil:
+				m, err := t.CreateModel(table, row.Delete, uuid)
+				if err != nil {
+					panic(err)
+				}
+				if err := tCache.Delete(uuid); err != nil {
+					panic(err)
+				}
+				t.eventProcessor.AddEvent(deleteEvent, table, m, nil)
+			}
+		}
+	}
+}
+
 // Purge drops all data in the cache and reinitializes it using the
 // provided schema
 func (t *TableCache) Purge(schema *ovsdb.DatabaseSchema) {
@@ -710,6 +779,114 @@ func (t *TableCache) CreateModel(tableName string, row *ovsdb.Row, uuid string)
 	return model, nil
 }
 
+// ApplyModifications applies the contents of a RowUpdate2.Modify to a model
+func (t *TableCache) ApplyModifications(tableName string, base model.Model, update ovsdb.Row) error {
+	table := t.mapper.Schema.Table(tableName)
+	if table == nil {
+		return fmt.Errorf("table %s not found", tableName)
+	}
+	schema := t.schema.Table(tableName)
+	if schema == nil {
+		return fmt.Errorf("no schema for table %s", tableName)
+	}
+	info, err := mapper.NewInfo(schema, base)
+	if err != nil {
+		return err
+	}
+	for k, v := range update {
+		if k == "_uuid" {
+			continue
+		}
+		value, err := ovsdb.OvsToNative(schema.Column(k), v)
+		if err != nil {
+			return err
+		}
+		nv := reflect.ValueOf(value)
+
+		switch nv.Kind() {
+		case reflect.Slice, reflect.Array:
+			// The difference between two sets are all elements that only belong to one of the sets.
+
+			// Iterate new values
+			for i := 0; i < nv.Len(); i++ {
+
+				// search for match in base values
+				baseValue, err := info.FieldByColumn(k)
+				if err != nil {
+					return err
+				}
+				bv := reflect.ValueOf(baseValue)
+				var found bool
+				for j := 0; j < bv.Len(); j++ {
+					if bv.Index(j).Interface() == nv.Index(i).Interface() {
+						// found a match, delete from slice
+						found = true
+						newValue := reflect.AppendSlice(bv.Slice(0, j), bv.Slice(j+1, bv.Len()))
+						err = info.SetField(k, newValue.Interface())
+						if err != nil {
+							return err
+						}
+						break
+					}
+				}
+				if !found {
+					newValue := reflect.Append(bv, nv.Index(i))
+					err = info.SetField(k, newValue.Interface())
+					if err != nil {
+						return err
+					}
+				}
+			}
+
+		case reflect.Map:
+			// The difference between two maps are all key-value pairs whose keys appears in only one of the maps,
+			// plus the key-value pairs whose keys appear in both maps but with different values.
+			// For the latter elements, <row> includes the value from the new column.
+			iter := nv.MapRange()
+
+			baseValue, err := info.FieldByColumn(k)
+			if err != nil {
+				return err
+			}
+
+			bv := reflect.ValueOf(baseValue)
+			if bv.IsNil() {
+				bv = reflect.MakeMap(nv.Type())
+			}
+
+			for iter.Next() {
+				mk := iter.Key()
+				mv := iter.Value()
+
+				existingValue := bv.MapIndex(mk)
+
+				// key does not exist, add it
+				if !existingValue.IsValid() {
+					bv.SetMapIndex(mk, mv)
+				} else if reflect.DeepEqual(mv.Interface(), existingValue.Interface()) {
+					// delete it
+					bv.SetMapIndex(mk, reflect.Value{})
+				} else {
+					// set new value
+					bv.SetMapIndex(mk, mv)
+				}
+			}
+			err = info.SetField(k, bv.Interface())
+			if err != nil {
+				return err
+			}
+
+		default:
+			// For columns with single value, the difference is the value of the new column.
+			err = info.SetField(k, value)
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return nil
+}
+
 func valueFromIndex(info *mapper.Info, index index) (interface{}, error) {
 	columns := index.columns()
 	if len(columns) > 1 {
diff --git a/client/api.go b/client/api.go
index 5a5837a..04fd5c4 100644
--- a/client/api.go
+++ b/client/api.go
@@ -1,8 +1,10 @@
 package client
 
 import (
+	"encoding/json"
 	"errors"
 	"fmt"
+	"log"
 	"reflect"
 
 	"github.com/ovn-org/libovsdb/cache"
@@ -220,7 +222,10 @@ func (a api) Get(m model.Model) error {
 	if found == nil {
 		return ErrNotFound
 	}
-	reflect.ValueOf(m).Elem().Set(reflect.Indirect(reflect.ValueOf(found)))
+
+	foundBytes, _ := json.Marshal(found)
+	_ = json.Unmarshal(foundBytes, m)
+
 	return nil
 }
 
@@ -359,6 +364,7 @@ func (a api) Update(model model.Model, fields ...interface{}) ([]ovsdb.Operation
 
 	for colName, column := range tableSchema.Columns {
 		if !column.Mutable() {
+			log.Printf("libovsdb: removing immutable field %s", colName)
 			delete(row, colName)
 		}
 	}
diff --git a/client/client.go b/client/client.go
index 80cd4ea..0e17523 100644
--- a/client/client.go
+++ b/client/client.go
@@ -16,7 +16,6 @@ import (
 	"github.com/cenkalti/backoff/v4"
 	"github.com/cenkalti/rpc2"
 	"github.com/cenkalti/rpc2/jsonrpc"
-	"github.com/google/uuid"
 	"github.com/ovn-org/libovsdb/cache"
 	"github.com/ovn-org/libovsdb/mapper"
 	"github.com/ovn-org/libovsdb/model"
@@ -36,6 +35,12 @@ const serverDB = "_Server"
 // ErrNotConnected is an error returned when the client is not connected
 var ErrNotConnected = errors.New("not connected")
 
+// ErrAlreadyConnected is an error returned when the client is already connected
+var ErrAlreadyConnected = errors.New("already connected")
+
+// ErrUnsupportedRPC is an error returned when an unsupported RPC method is called
+var ErrUnsupportedRPC = errors.New("unsupported rpc")
+
 // Client represents an OVSDB Client Connection
 // It provides all the necessary functionality to Connect to a server,
 // perform transactions, and build your own replica of the database with
@@ -52,21 +57,14 @@ type Client interface {
 	DisconnectNotify() chan struct{}
 	Echo(context.Context) error
 	Transact(context.Context, ...ovsdb.Operation) ([]ovsdb.OperationResult, error)
-	Monitor(context.Context, ...TableMonitor) (MonitorCookie, error)
+	Monitor(context.Context, *Monitor) (MonitorCookie, error)
 	MonitorAll(context.Context) (MonitorCookie, error)
 	MonitorCancel(ctx context.Context, cookie MonitorCookie) error
-	NewTableMonitor(m model.Model, fields ...interface{}) TableMonitor
+	NewMonitor(...MonitorOption) *Monitor
 	CurrentEndpoint() string
 	API
 }
 
-// MonitorCookie is the struct we pass to correlate from updates back to their
-// originating Monitor request.
-type MonitorCookie struct {
-	DatabaseName string `json:"databaseName"`
-	ID           string `json:"id"`
-}
-
 // ovsdbClient is an OVSDB client
 type ovsdbClient struct {
 	options        *options
@@ -96,7 +94,7 @@ type database struct {
 	api API
 
 	// any ongoing monitors, so we can re-create them if we disconnect
-	monitors      map[string][]TableMonitor
+	monitors      map[string]*Monitor
 	monitorsMutex sync.Mutex
 }
 
@@ -115,7 +113,7 @@ func newOVSDBClient(databaseModel *model.DBModel, opts ...Option) (*ovsdbClient,
 		databases: map[string]*database{
 			databaseModel.Name(): {
 				model:    databaseModel,
-				monitors: make(map[string][]TableMonitor),
+				monitors: make(map[string]*Monitor),
 			},
 		},
 		disconnect: make(chan struct{}),
@@ -134,7 +132,7 @@ func newOVSDBClient(databaseModel *model.DBModel, opts ...Option) (*ovsdbClient,
 		}
 		ovs.databases[serverDB] = &database{
 			model:    sm,
-			monitors: make(map[string][]TableMonitor),
+			monitors: make(map[string]*Monitor),
 		}
 	}
 	return ovs, nil
@@ -146,9 +144,11 @@ func newOVSDBClient(databaseModel *model.DBModel, opts ...Option) (*ovsdbClient,
 // If no WithEndpoint option is supplied, the default of unix:/var/run/openvswitch/ovsdb.sock is used
 func (o *ovsdbClient) Connect(ctx context.Context) error {
 	if err := o.connect(ctx, false); err != nil {
+		if err == ErrAlreadyConnected {
+			return nil
+		}
 		return err
 	}
-
 	if o.options.leaderOnly {
 		if err := o.watchForLeaderChange(); err != nil {
 			return err
@@ -161,7 +161,7 @@ func (o *ovsdbClient) connect(ctx context.Context, reconnect bool) error {
 	o.rpcMutex.Lock()
 	defer o.rpcMutex.Unlock()
 	if o.rpcClient != nil {
-		return nil
+		return ErrAlreadyConnected
 	}
 
 	connected := false
@@ -202,8 +202,9 @@ func (o *ovsdbClient) connect(ctx context.Context, reconnect bool) error {
 			db.monitorsMutex.Lock()
 			defer db.monitorsMutex.Unlock()
 			for id, request := range db.monitors {
-				err := o.monitor(ctx, MonitorCookie{DatabaseName: dbName, ID: id}, true, request...)
+				err := o.monitor(ctx, MonitorCookie{DatabaseName: dbName, ID: id}, true, request)
 				if err != nil {
+					o.rpcClient = nil
 					return err
 				}
 			}
@@ -242,9 +243,7 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) error {
 		return fmt.Errorf("failed to open connection: %w", err)
 	}
 
-	if err = o.createRPC2Client(c); err != nil {
-		return fmt.Errorf("failed to open RPC connection: %w", err)
-	}
+	o.createRPC2Client(c)
 
 	// from now on, if err is nil, always tear down the RPC session
 	defer func() {
@@ -328,7 +327,7 @@ func (o *ovsdbClient) tryEndpoint(ctx context.Context, u *url.URL) error {
 // createRPC2Client creates an rpcClient using the provided connection
 // It is also responsible for setting up go routines for client-side event handling
 // Should only be called when the mutex is held
-func (o *ovsdbClient) createRPC2Client(conn net.Conn) error {
+func (o *ovsdbClient) createRPC2Client(conn net.Conn) {
 	o.stopCh = make(chan struct{})
 	o.rpcClient = rpc2.NewClientWithCodec(jsonrpc.NewJSONCodec(conn))
 	o.rpcClient.SetBlocking(true)
@@ -338,8 +337,13 @@ func (o *ovsdbClient) createRPC2Client(conn net.Conn) error {
 	o.rpcClient.Handle("update", func(_ *rpc2.Client, args []json.RawMessage, reply *[]interface{}) error {
 		return o.update(args, reply)
 	})
+	o.rpcClient.Handle("update2", func(_ *rpc2.Client, args []json.RawMessage, reply *[]interface{}) error {
+		return o.update2(args, reply)
+	})
+	o.rpcClient.Handle("update3", func(_ *rpc2.Client, args []json.RawMessage, reply *[]interface{}) error {
+		return o.update3(args, reply)
+	})
 	go o.rpcClient.Run()
-	return nil
 }
 
 // isEndpointLeader returns true if the currently connected endpoint is leader.
@@ -488,6 +492,71 @@ func (o *ovsdbClient) update(params []json.RawMessage, reply *[]interface{}) err
 	return nil
 }
 
+// update2 handling from ovsdb-server.7
+func (o *ovsdbClient) update2(params []json.RawMessage, reply *[]interface{}) error {
+	cookie := MonitorCookie{}
+	if len(params) > 2 {
+		return fmt.Errorf("update2 requires exactly 2 args")
+	}
+	err := json.Unmarshal(params[0], &cookie)
+	if err != nil {
+		return err
+	}
+	var updates ovsdb.TableUpdates2
+	err = json.Unmarshal(params[1], &updates)
+	if err != nil {
+		return err
+	}
+	db := o.databases[cookie.DatabaseName]
+	if db == nil {
+		return fmt.Errorf("update: invalid database name: %s unknown", cookie.DatabaseName)
+	}
+	// Update the local DB cache with the tableUpdates
+	db.cacheMutex.RLock()
+	db.cache.Update2(cookie, updates)
+	db.cacheMutex.RUnlock()
+	*reply = []interface{}{}
+	return nil
+}
+
+// update3 handling from ovsdb-server.7
+func (o *ovsdbClient) update3(params []json.RawMessage, reply *[]interface{}) error {
+	cookie := MonitorCookie{}
+	if len(params) > 3 {
+		return fmt.Errorf("update requires exactly 3 args")
+	}
+	err := json.Unmarshal(params[0], &cookie)
+	if err != nil {
+		return err
+	}
+	var lastTransactionID string
+	err = json.Unmarshal(params[1], &lastTransactionID)
+	if err != nil {
+		return err
+	}
+	var updates ovsdb.TableUpdates2
+	err = json.Unmarshal(params[2], &updates)
+	if err != nil {
+		return err
+	}
+
+	db := o.databases[cookie.DatabaseName]
+	if db == nil {
+		return fmt.Errorf("update: invalid database name: %s unknown", cookie.DatabaseName)
+	}
+	db.monitorsMutex.Lock()
+	mon := db.monitors[cookie.ID]
+	mon.LastTransactionID = lastTransactionID
+	db.monitorsMutex.Unlock()
+
+	// Update the local DB cache with the tableUpdates
+	db.cacheMutex.RLock()
+	db.cache.Update2(cookie, updates)
+	db.cacheMutex.RUnlock()
+	*reply = []interface{}{}
+	return nil
+}
+
 // getSchema returns the schema in use for the provided database name
 // RFC 7047 : get_schema
 // Should only be called when mutex is held
@@ -557,11 +626,11 @@ func (o *ovsdbClient) transact(ctx context.Context, dbName string, operation ...
 
 // MonitorAll is a convenience method to monitor every table/column
 func (o *ovsdbClient) MonitorAll(ctx context.Context) (MonitorCookie, error) {
-	var options []TableMonitor
+	m := newMonitor()
 	for name := range o.primaryDB().model.Types() {
-		options = append(options, TableMonitor{Table: name})
+		m.Tables = append(m.Tables, TableMonitor{Table: name})
 	}
-	return o.Monitor(ctx, options...)
+	return o.Monitor(ctx, m)
 }
 
 // MonitorCancel will request cancel a previously issued monitor request
@@ -590,51 +659,30 @@ func (o *ovsdbClient) MonitorCancel(ctx context.Context, cookie MonitorCookie) e
 	return nil
 }
 
-// TableMonitor is a table to be monitored
-type TableMonitor struct {
-	// Table is the table to be monitored
-	Table string
-	// Fields are the fields in the model to monitor
-	// If none are supplied, all fields will be used
-	Fields []interface{}
-	// Error will contain any errors caught in the creation of a TableMonitor
-	Error error
-}
-
-func (o *ovsdbClient) NewTableMonitor(m model.Model, fields ...interface{}) TableMonitor {
-	tableName := o.primaryDB().model.FindTable(reflect.TypeOf(m))
-	if tableName == "" {
-		return TableMonitor{
-			Error: fmt.Errorf("object of type %s is not part of the DBModel", reflect.TypeOf(m)),
-		}
-	}
-	return TableMonitor{
-		Table:  tableName,
-		Fields: fields,
-	}
-}
-
-func newMonitorCookie(dbName string) MonitorCookie {
-	return MonitorCookie{
-		DatabaseName: dbName,
-		ID:           uuid.NewString(),
-	}
-}
-
 // Monitor will provide updates for a given table/column
 // and populate the cache with them. Subsequent updates will be processed
 // by the Update Notifications
 // RFC 7047 : monitor
-func (o *ovsdbClient) Monitor(ctx context.Context, options ...TableMonitor) (MonitorCookie, error) {
+func (o *ovsdbClient) Monitor(ctx context.Context, monitor *Monitor) (MonitorCookie, error) {
 	cookie := newMonitorCookie(o.primaryDBName)
-	return cookie, o.monitor(ctx, cookie, false, options...)
+	err := o.monitor(ctx, cookie, false, monitor)
+	if err != nil && err == ErrUnsupportedRPC {
+		return cookie, o.monitor(ctx, cookie, false, monitor)
+	}
+	return cookie, err
 }
 
-func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconnecting bool, options ...TableMonitor) error {
-	if len(options) == 0 {
-		return fmt.Errorf("no monitor options provided")
+func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconnecting bool, monitor *Monitor) error {
+	if len(monitor.Tables) == 0 {
+		return fmt.Errorf("at least one table should be monitored")
+	}
+	if len(monitor.Errors) != 0 {
+		var errString []string
+		for _, err := range monitor.Errors {
+			errString = append(errString, err.Error())
+		}
+		return fmt.Errorf(strings.Join(errString, ". "))
 	}
-	var reply ovsdb.TableUpdates
 	dbName := cookie.DatabaseName
 	db := o.databases[dbName]
 	db.schemaMutex.RLock()
@@ -642,10 +690,7 @@ func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconne
 	db.schemaMutex.RUnlock()
 	typeMap := o.databases[dbName].model.Types()
 	requests := make(map[string]ovsdb.MonitorRequest)
-	for _, o := range options {
-		if o.Error != nil {
-			return o.Error
-		}
+	for _, o := range monitor.Tables {
 		m, ok := typeMap[o.Table]
 		if !ok {
 			return fmt.Errorf("type for table %s does not exist in model", o.Table)
@@ -656,7 +701,13 @@ func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconne
 		}
 		requests[o.Table] = *request
 	}
-	args := ovsdb.NewMonitorArgs(dbName, cookie, requests)
+
+	var args []interface{}
+	if monitor.Method == ovsdb.ConditionalMonitorSinceRPC {
+		args = ovsdb.NewMonitorCondSinceArgs(dbName, cookie, requests, monitor.LastTransactionID)
+	} else {
+		args = ovsdb.NewMonitorArgs(dbName, cookie, requests)
+	}
 
 	// if we're reconnecting, we already hold the rpcMutex
 	if !reconnecting {
@@ -667,20 +718,63 @@ func (o *ovsdbClient) monitor(ctx context.Context, cookie MonitorCookie, reconne
 		return ErrNotConnected
 	}
 
-	err := o.rpcClient.CallWithContext(ctx, "monitor", args, &reply)
+	var err error
+	var tableUpdates interface{}
+
+	switch monitor.Method {
+	case ovsdb.MonitorRPC:
+		var reply ovsdb.TableUpdates
+		err = o.rpcClient.CallWithContext(ctx, monitor.Method, args, &reply)
+		tableUpdates = reply
+	case ovsdb.ConditionalMonitorRPC:
+		var reply ovsdb.TableUpdates2
+		err = o.rpcClient.CallWithContext(ctx, monitor.Method, args, &reply)
+		tableUpdates = reply
+	case ovsdb.ConditionalMonitorSinceRPC:
+		var reply ovsdb.MonitorCondSinceReply
+		err = o.rpcClient.CallWithContext(ctx, monitor.Method, args, &reply)
+		if err == nil && reply.Found {
+			monitor.LastTransactionID = reply.LastTransactionID
+		}
+		tableUpdates = reply.Updates
+	default:
+		return fmt.Errorf("unsupported monitor method: %v", monitor.Method)
+	}
+
 	if err != nil {
 		if err == rpc2.ErrShutdown {
 			return ErrNotConnected
 		}
+		// TODO: Match unsupported RPC method error
+		if monitor.Method == ovsdb.ConditionalMonitorSinceRPC {
+			log.Printf("libovsdb: method monitor_cond_since not supported, falling back to monitor_cond: %v", err.Error())
+			monitor.Method = ovsdb.ConditionalMonitorRPC
+			// return to ensure that we release any held mutexes, retry must happen from a calling function
+			return ErrUnsupportedRPC
+		} else if monitor.Method == ovsdb.ConditionalMonitorRPC {
+			log.Printf("libovsdb: method monitor_cond not supported, falling back to monitor: %v", err.Error())
+			monitor.Method = ovsdb.MonitorRPC
+			// return to ensure that we release any held mutexes, retry must happen from a calling function
+			return ErrUnsupportedRPC
+		}
 		return err
 	}
+
 	if !reconnecting {
 		db := o.databases[dbName]
 		db.monitorsMutex.Lock()
-		db.monitors[cookie.ID] = options
+		db.monitors[cookie.ID] = monitor
 		db.monitorsMutex.Unlock()
 	}
-	o.databases[dbName].cache.Populate(reply)
+
+	if monitor.Method == ovsdb.MonitorRPC {
+		u := tableUpdates.(ovsdb.TableUpdates)
+		o.databases[dbName].cache.Populate(u)
+	} else {
+		u := tableUpdates.(ovsdb.TableUpdates2)
+		o.databases[dbName].cache.Populate2(u)
+	}
+
 	return nil
 }
 
@@ -717,11 +811,11 @@ func (o *ovsdbClient) watchForLeaderChange() error {
 		},
 	})
 
-	err := o.monitor(context.Background(), newMonitorCookie(serverDB), false,
-		TableMonitor{
-			Table: "Database",
-		},
-	)
+	m := newMonitor()
+	// NOTE: _Server does not support monitor_cond_since
+	m.Method = ovsdb.ConditionalMonitorRPC
+	m.Tables = []TableMonitor{{Table: "Database"}}
+	err := o.monitor(context.Background(), newMonitorCookie(serverDB), false, m)
 	if err != nil {
 		return err
 	}
@@ -760,6 +854,9 @@ func (o *ovsdbClient) handleDisconnectNotification() {
 			defer cancel()
 			err := o.connect(ctx, true)
 			if err != nil {
+				if err == ErrAlreadyConnected {
+					panic(err)
+				}
 				log.Printf("libovsdb: failed to reconnect: %s", err)
 			}
 			return err
@@ -790,7 +887,7 @@ func (o *ovsdbClient) handleDisconnectNotification() {
 
 		db.monitorsMutex.Lock()
 		defer db.monitorsMutex.Unlock()
-		db.monitors = make(map[string][]TableMonitor)
+		db.monitors = make(map[string]*Monitor)
 	}
 
 	o.shutdownMutex.Lock()
diff --git a/client/monitor.go b/client/monitor.go
new file mode 100644
index 0000000..d2fd731
--- /dev/null
+++ b/client/monitor.go
@@ -0,0 +1,100 @@
+package client
+
+import (
+	"fmt"
+	"reflect"
+
+	"github.com/google/uuid"
+	"github.com/ovn-org/libovsdb/model"
+	"github.com/ovn-org/libovsdb/ovsdb"
+)
+
+const emptyUUID = "00000000-0000-0000-0000-000000000000"
+
+// Monitor represents a monitor
+type Monitor struct {
+	Method            string
+	Tables            []TableMonitor
+	Errors            []error
+	LastTransactionID string
+}
+
+// newMonitor creates a new *Monitor with default values
+func newMonitor() *Monitor {
+	return &Monitor{
+		Method:            ovsdb.ConditionalMonitorSinceRPC,
+		Errors:            make([]error, 0),
+		LastTransactionID: emptyUUID,
+	}
+}
+
+// NewMonitor creates a new Monitor with the provided options
+func (o *ovsdbClient) NewMonitor(opts ...MonitorOption) *Monitor {
+	m := newMonitor()
+	for _, opt := range opts {
+		err := opt(o, m)
+		if err != nil {
+			m.Errors = append(m.Errors, err)
+		}
+	}
+	return m
+}
+
+// MonitorOption adds Tables to a Monitor
+type MonitorOption func(o *ovsdbClient, m *Monitor) error
+
+// MonitorCookie is the struct we pass to correlate from updates back to their
+// originating Monitor request.
+type MonitorCookie struct {
+	DatabaseName string `json:"databaseName"`
+	ID           string `json:"id"`
+}
+
+func newMonitorCookie(dbName string) MonitorCookie {
+	return MonitorCookie{
+		DatabaseName: dbName,
+		ID:           uuid.NewString(),
+	}
+}
+
+// TableMonitor is a table to be monitored
+type TableMonitor struct {
+	// Table is the table to be monitored
+	Table string
+	// Condition is the condition under which the table should be monitored
+	Condition model.Condition
+	// Fields are the fields in the model to monitor
+	// If none are supplied, all fields will be used
+	Fields []interface{}
+}
+
+func WithTable(m model.Model, fields ...interface{}) MonitorOption {
+	return func(o *ovsdbClient, monitor *Monitor) error {
+		tableName := o.primaryDB().model.FindTable(reflect.TypeOf(m))
+		if tableName == "" {
+			return fmt.Errorf("object of type %s is not part of the DBModel", reflect.TypeOf(m))
+		}
+		tableMonitor := TableMonitor{
+			Table:  tableName,
+			Fields: fields,
+		}
+		monitor.Tables = append(monitor.Tables, tableMonitor)
+		return nil
+	}
+}
+
+func WithConditionalTable(m model.Model, condition model.Condition, fields ...interface{}) MonitorOption {
+	return func(o *ovsdbClient, monitor *Monitor) error {
+		tableName := o.primaryDB().model.FindTable(reflect.TypeOf(m))
+		if tableName == "" {
+			return fmt.Errorf("object of type %s is not part of the DBModel", reflect.TypeOf(m))
+		}
+		tableMonitor := TableMonitor{
+			Table:     tableName,
+			Condition: condition,
+			Fields:    fields,
+		}
+		monitor.Tables = append(monitor.Tables, tableMonitor)
+		return nil
+	}
+}
diff --git a/example/play_with_ovs/main.go b/example/play_with_ovs/main.go
index 7694918..b4e1bd3 100644
--- a/example/play_with_ovs/main.go
+++ b/example/play_with_ovs/main.go
@@ -122,8 +122,10 @@ func main() {
 	})
 	_, err = ovs.Monitor(
 		context.TODO(),
-		ovs.NewTableMonitor(&vswitchd.OpenvSwitch{}),
-		ovs.NewTableMonitor(&vswitchd.Bridge{}),
+		ovs.NewMonitor(
+			client.WithTable(&vswitchd.OpenvSwitch{}),
+			client.WithTable(&vswitchd.Bridge{}),
+		),
 	)
 	if err != nil {
 		log.Fatal(err)
diff --git a/mapper/info.go b/mapper/info.go
index fd1ceb1..1ad981b 100644
--- a/mapper/info.go
+++ b/mapper/info.go
@@ -20,7 +20,7 @@ type Info struct {
 func (i *Info) FieldByColumn(column string) (interface{}, error) {
 	fieldName, ok := i.fields[column]
 	if !ok {
-		return nil, fmt.Errorf("column %s not found in orm info", column)
+		return nil, fmt.Errorf("FieldByColumn: column %s not found in orm info", column)
 	}
 	return reflect.ValueOf(i.obj).Elem().FieldByName(fieldName).Interface(), nil
 }
@@ -35,7 +35,7 @@ func (i *Info) hasColumn(column string) bool {
 func (i *Info) SetField(column string, value interface{}) error {
 	fieldName, ok := i.fields[column]
 	if !ok {
-		return fmt.Errorf("column %s not found in orm info", column)
+		return fmt.Errorf("SetField: column %s not found in orm info", column)
 	}
 	fieldValue := reflect.ValueOf(i.obj).Elem().FieldByName(fieldName)
 
diff --git a/model/model.go b/model/model.go
index 3c201e3..ef77fc9 100644
--- a/model/model.go
+++ b/model/model.go
@@ -1,6 +1,7 @@
 package model
 
 import (
+	"encoding/json"
 	"fmt"
 	"reflect"
 
@@ -24,6 +25,15 @@ import (
 //}
 type Model interface{}
 
+// Clone creates a deep copy of a model
+func Clone(a Model) Model {
+	val := reflect.Indirect(reflect.ValueOf(a))
+	b := reflect.New(val.Type()).Interface()
+	aBytes, _ := json.Marshal(a)
+	_ = json.Unmarshal(aBytes, b)
+	return b
+}
+
 // DBModel is a Database model
 type DBModel struct {
 	name  string
diff --git a/ovsdb/notation.go b/ovsdb/notation.go
index 1b5bf0d..ec3aa0e 100644
--- a/ovsdb/notation.go
+++ b/ovsdb/notation.go
@@ -82,6 +82,7 @@ type MonitorRequests struct {
 // MonitorRequest represents a monitor request according to RFC7047
 type MonitorRequest struct {
 	Columns []string       `json:"columns,omitempty"`
+	Where   []Condition    `json:"where,omitempty"`
 	Select  *MonitorSelect `json:"select,omitempty"`
 }
 
diff --git a/ovsdb/rpc.go b/ovsdb/rpc.go
index 0604aca..f1e5980 100644
--- a/ovsdb/rpc.go
+++ b/ovsdb/rpc.go
@@ -1,5 +1,14 @@
 package ovsdb
 
+const (
+	// MonitorRPC is the monitor RPC method
+	MonitorRPC = "monitor"
+	// ConditionalMonitorRPC is the monitor_cond
+	ConditionalMonitorRPC = "monitor_cond"
+	// ConditionalMonitorSinceRPC is the monitor_cond_since RPC method
+	ConditionalMonitorSinceRPC = "monitor_cond_since"
+)
+
 // NewEchoArgs creates a new set of arguments for an echo RPC
 func NewEchoArgs() []interface{} {
 	return []interface{}{"libovsdb echo"}
@@ -34,6 +43,11 @@ func NewMonitorArgs(database string, value interface{}, requests map[string]Moni
 	return []interface{}{database, value, requests}
 }
 
+// NewMonitorCondSinceArgs creates a new set of arguments for a monitor_cond_since RPC
+func NewMonitorCondSinceArgs(database string, value interface{}, requests map[string]MonitorRequest, lastTransactionID string) []interface{} {
+	return []interface{}{database, value, requests, lastTransactionID}
+}
+
 // NewMonitorCancelArgs creates a new set of arguments for a monitor_cancel RPC
 func NewMonitorCancelArgs(value interface{}) []interface{} {
 	return []interface{}{value}
@@ -49,6 +63,9 @@ type NotificationHandler interface {
 	// RFC 7047 section 4.1.6 Update Notification
 	Update(context interface{}, tableUpdates TableUpdates)
 
+	// ovsdb-server.7 update2 notifications
+	Update2(context interface{}, tableUpdates TableUpdates2)
+
 	// RFC 7047 section 4.1.9 Locked Notification
 	Locked([]interface{})
 
diff --git a/ovsdb/update3.go b/ovsdb/update3.go
new file mode 100644
index 0000000..a24ce64
--- /dev/null
+++ b/ovsdb/update3.go
@@ -0,0 +1,51 @@
+package ovsdb
+
+import (
+	"encoding/json"
+	"fmt"
+)
+
+type MonitorCondSinceReply struct {
+	Found             bool
+	LastTransactionID string
+	Updates           TableUpdates2
+}
+
+func (m MonitorCondSinceReply) MarshalJSON() ([]byte, error) {
+	v := []interface{}{m.Found, m.LastTransactionID, m.Updates}
+	return json.Marshal(v)
+}
+
+func (m *MonitorCondSinceReply) UnmarshalJSON(b []byte) error {
+	var v []json.RawMessage
+	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))
+	}
+
+	var found bool
+	err = json.Unmarshal(v[0], &found)
+	if err != nil {
+		return err
+	}
+
+	var lastTransactionID string
+	err = json.Unmarshal(v[1], &lastTransactionID)
+	if err != nil {
+		return err
+	}
+
+	var updates TableUpdates2
+	err = json.Unmarshal(v[2], &updates)
+	if err != nil {
+		return err
+	}
+
+	m.Found = found
+	m.LastTransactionID = lastTransactionID
+	m.Updates = updates
+	return nil
+}
diff --git a/ovsdb/updates.go b/ovsdb/updates.go
index 0681503..9702b64 100644
--- a/ovsdb/updates.go
+++ b/ovsdb/updates.go
@@ -15,6 +15,14 @@ func (t TableUpdates) AddTableUpdate(table string, update TableUpdate) {
 	}
 }
 
+func (t TableUpdates) FromTableUpdates2(tu TableUpdates2) {
+	for k, v := range tu {
+		var u TableUpdate
+		u.FromTableUpdate2(v)
+		t[k] = u
+	}
+}
+
 func (t TableUpdates) Merge(update TableUpdates) {
 	for k, v := range update {
 		t.AddTableUpdate(k, v)
@@ -33,6 +41,12 @@ func (t TableUpdate) AddRowUpdate(uuid string, update *RowUpdate) {
 	}
 }
 
+func (t TableUpdate) FromTableUpdate2(tu TableUpdate2) {
+	for k, v := range tu {
+		t[k] = &RowUpdate{Old: v.Old, New: v.New}
+	}
+}
+
 // RowUpdate represents a row update according to RFC7047
 type RowUpdate struct {
 	New *Row `json:"new,omitempty"`
diff --git a/ovsdb/updates2.go b/ovsdb/updates2.go
new file mode 100644
index 0000000..7b269ec
--- /dev/null
+++ b/ovsdb/updates2.go
@@ -0,0 +1,73 @@
+package ovsdb
+
+// TableUpdates2 is an object that maps from a table name to a
+// TableUpdate2
+type TableUpdates2 map[string]TableUpdate2
+
+// TableUpdate2 is an object that maps from the row's UUID to a
+// RowUpdate2
+type TableUpdate2 map[string]*RowUpdate2
+
+// RowUpdate2 represents a row update according to ovsdb-server.7
+type RowUpdate2 struct {
+	Initial *Row `json:"initial,omitempty"`
+	Insert  *Row `json:"insert,omitempty"`
+	Modify  *Row `json:"modify,omitempty"`
+	Delete  *Row `json:"delete,omitempty"`
+	Old     *Row `json:"-"`
+	New     *Row `json:"-"`
+}
+
+// AddTableUpdate adds a new TableUpdate to a TableUpdates
+func (t TableUpdates2) AddTableUpdate(table string, update TableUpdate2) {
+	if _, ok := t[table]; !ok {
+		t[table] = update
+	} else {
+		for uuid, row := range update {
+			t[table].AddRowUpdate(uuid, row)
+		}
+	}
+}
+
+func (t TableUpdates2) Merge(update TableUpdates2) {
+	for k, v := range update {
+		t.AddTableUpdate(k, v)
+	}
+}
+
+func (t TableUpdate2) AddRowUpdate(uuid string, update *RowUpdate2) {
+	if _, ok := t[uuid]; !ok {
+		t[uuid] = update
+	} else {
+		t[uuid].Merge(update)
+	}
+}
+
+func (r *RowUpdate2) Merge(new *RowUpdate2) {
+	// old.Delete() cannot be true, as we can't modify a row after it was deleted
+	if r.Delete != nil {
+		return
+	}
+	// we should never get two insert events
+	if r.Insert != nil && new.Insert != nil {
+		return
+	}
+	if r.Insert != nil && new.Modify != nil {
+		r.Insert = new.Modify
+		return
+	}
+	if r.Insert != nil && new.Delete != nil {
+		r.Insert = nil
+		r.Delete = new.Delete
+		return
+	}
+	if r.Modify != nil && new.Modify != nil {
+		r.Modify = new.Modify
+		return
+	}
+	if r.Modify != nil && new.Delete != nil {
+		r.Modify = nil
+		r.Delete = new.Delete
+		return
+	}
+}
diff --git a/server/database.go b/server/database.go
index ceec0c4..ffb7c0e 100644
--- a/server/database.go
+++ b/server/database.go
@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"sync"
 
+	"github.com/google/uuid"
 	"github.com/ovn-org/libovsdb/cache"
 	"github.com/ovn-org/libovsdb/model"
 	"github.com/ovn-org/libovsdb/ovsdb"
@@ -13,7 +14,7 @@ import (
 type Database interface {
 	CreateDatabase(database string, model *ovsdb.DatabaseSchema) error
 	Exists(database string) bool
-	Commit(database string, updates ovsdb.TableUpdates) error
+	Commit(database string, id uuid.UUID, updates ovsdb.TableUpdates2) error
 	CheckIndexes(database string, table string, m model.Model) error
 	List(database, table string, conditions ...ovsdb.Condition) ([]model.Model, error)
 	Get(database, table string, uuid string) (model.Model, error)
@@ -56,14 +57,14 @@ func (db *inMemoryDatabase) Exists(name string) bool {
 	return ok
 }
 
-func (db *inMemoryDatabase) Commit(database string, updates ovsdb.TableUpdates) error {
+func (db *inMemoryDatabase) Commit(database string, id uuid.UUID, updates ovsdb.TableUpdates2) error {
 	if !db.Exists(database) {
 		return fmt.Errorf("db does not exist")
 	}
 	db.mutex.RLock()
 	targetDb := db.databases[database]
 	db.mutex.RLock()
-	targetDb.Populate(updates)
+	targetDb.Populate2(updates)
 	return nil
 }
 
diff --git a/server/monitor.go b/server/monitor.go
index 8472b87..19ba1bd 100644
--- a/server/monitor.go
+++ b/server/monitor.go
@@ -2,11 +2,11 @@ package server
 
 import (
 	"encoding/json"
-	"fmt"
 	"log"
 	"sync"
 
 	"github.com/cenkalti/rpc2"
+	"github.com/google/uuid"
 	"github.com/ovn-org/libovsdb/ovsdb"
 )
 
@@ -27,13 +27,43 @@ func newConnectionMonitors() *connectionMonitors {
 // will be reflected
 type monitor struct {
 	id      string
+	kind    monitorKind
 	request map[string]*ovsdb.MonitorRequest
 	client  *rpc2.Client
 }
 
+type monitorKind int
+
+const (
+	monitorKindOriginal monitorKind = iota
+	monitorKindConditional
+	monitorKindConditionalSince
+)
+
 func newMonitor(id string, request map[string]*ovsdb.MonitorRequest, client *rpc2.Client) *monitor {
 	m := &monitor{
 		id:      id,
+		kind:    monitorKindOriginal,
+		request: request,
+		client:  client,
+	}
+	return m
+}
+
+func newConditionalMonitor(id string, request map[string]*ovsdb.MonitorRequest, client *rpc2.Client) *monitor {
+	m := &monitor{
+		id:      id,
+		kind:    monitorKindConditional,
+		request: request,
+		client:  client,
+	}
+	return m
+}
+
+func newConditionalSinceMonitor(id string, request map[string]*ovsdb.MonitorRequest, client *rpc2.Client) *monitor {
+	m := &monitor{
+		id:      id,
+		kind:    monitorKindConditional,
 		request: request,
 		client:  client,
 	}
@@ -53,7 +83,45 @@ func (m *monitor) Send(update ovsdb.TableUpdates) {
 	}
 	args := []interface{}{json.RawMessage([]byte(m.id)), update}
 	var reply interface{}
-	err := m.client.Call("update", args, &reply)
+	err := m.client.Call("update2", args, &reply)
+	if err != nil {
+		log.Printf("client error handling update rpc: %v", err)
+	}
+}
+
+// Send2 will send an update if it matches the tables and monitor select arguments
+// we take the update by value (not reference) so we can mutate it in place before
+// queuing it for dispatch
+func (m *monitor) Send2(update ovsdb.TableUpdates2) {
+	// remove updates for tables that we aren't watching
+	if len(m.request) != 0 {
+		m.filter2(update)
+	}
+	if len(update) == 0 {
+		return
+	}
+	args := []interface{}{json.RawMessage([]byte(m.id)), update}
+	var reply interface{}
+	err := m.client.Call("update2", args, &reply)
+	if err != nil {
+		log.Printf("client error handling update rpc: %v", err)
+	}
+}
+
+// Send3 will send an update if it matches the tables and monitor select arguments
+// we take the update by value (not reference) so we can mutate it in place before
+// queuing it for dispatch
+func (m *monitor) Send3(id uuid.UUID, update ovsdb.TableUpdates2) {
+	// remove updates for tables that we aren't watching
+	if len(m.request) != 0 {
+		m.filter2(update)
+	}
+	if len(update) == 0 {
+		return
+	}
+	args := []interface{}{json.RawMessage([]byte(m.id)), id.String(), update}
+	var reply interface{}
+	err := m.client.Call("update2", args, &reply)
 	if err != nil {
 		log.Printf("client error handling update rpc: %v", err)
 	}
@@ -64,7 +132,6 @@ func (m *monitor) filter(update ovsdb.TableUpdates) {
 	if len(m.request) != 0 {
 		for table, u := range update {
 			if _, ok := m.request[table]; !ok {
-				fmt.Println("dropping table update")
 				delete(update, table)
 				continue
 			}
@@ -108,3 +175,62 @@ func (m *monitor) filter(update ovsdb.TableUpdates) {
 		}
 	}
 }
+
+func (m *monitor) filter2(update ovsdb.TableUpdates2) {
+	// remove updates for tables that we aren't watching
+	if len(m.request) != 0 {
+		for table, u := range update {
+			if _, ok := m.request[table]; !ok {
+				delete(update, table)
+				continue
+			}
+			for uuid, row := range u {
+				switch {
+				case row.Insert != nil && m.request[table].Select.Insert():
+					fallthrough
+				case row.Modify != nil && m.request[table].Select.Modify():
+					fallthrough
+				case row.Delete != nil && m.request[table].Select.Delete():
+					if len(m.request[table].Columns) > 0 {
+						cols := make(map[string]bool)
+						for _, c := range m.request[table].Columns {
+							cols[c] = true
+						}
+						if row.Insert != nil {
+							new := *row.Insert
+							for k := range new {
+								if _, ok := cols[k]; !ok {
+									delete(new, k)
+								}
+							}
+							row.Insert = &new
+							update[table][uuid] = row
+						}
+						if row.Modify != nil {
+							new := *row.Modify
+							for k := range new {
+								if _, ok := cols[k]; !ok {
+									delete(new, k)
+								}
+							}
+							row.Modify = &new
+							update[table][uuid] = row
+						}
+						if row.Delete != nil {
+							old := *row.Delete
+							for k := range old {
+								if _, ok := cols[k]; !ok {
+									delete(old, k)
+								}
+							}
+							row.Delete = &old
+							update[table][uuid] = row
+						}
+					}
+				default:
+					delete(u, uuid)
+				}
+			}
+		}
+	}
+}
diff --git a/server/mutate.go b/server/mutate.go
index 5f66d6b..41afb53 100644
--- a/server/mutate.go
+++ b/server/mutate.go
@@ -25,28 +25,37 @@ func insertToSlice(a, b reflect.Value) reflect.Value {
 	return reflect.Append(a, b)
 }
 
-func mutate(current interface{}, mutator ovsdb.Mutator, value interface{}) interface{} {
+func mutate(current interface{}, mutator ovsdb.Mutator, value interface{}) (interface{}, interface{}) {
 	switch current.(type) {
 	case bool, string:
-		return current
+		return current, value
 	}
 	switch mutator {
 	case ovsdb.MutateOperationInsert:
-		return mutateInsert(current, value)
+		// for insert, the delta will be the new value added
+		return mutateInsert(current, value), value
 	case ovsdb.MutateOperationDelete:
-		return mutateDelete(current, value)
+		// for delete, the delta will be the value removed
+		return mutateDelete(current, value), value
 	case ovsdb.MutateOperationAdd:
-		return mutateAdd(current, value)
+		// for add, the delta is the new value
+		new := mutateAdd(current, value)
+		return new, new
 	case ovsdb.MutateOperationSubtract:
-		return mutateSubtract(current, value)
+		// for subtract, the delta is the new value
+		new := mutateSubtract(current, value)
+		return new, new
 	case ovsdb.MutateOperationMultiply:
-		return mutateMultiply(current, value)
+		new := mutateMultiply(current, value)
+		return new, new
 	case ovsdb.MutateOperationDivide:
-		return mutateDivide(current, value)
+		new := mutateDivide(current, value)
+		return new, new
 	case ovsdb.MutateOperationModulo:
-		return mutateModulo(current, value)
+		new := mutateModulo(current, value)
+		return new, new
 	}
-	return current
+	return current, value
 }
 
 func mutateInsert(current, value interface{}) interface{} {
diff --git a/server/server.go b/server/server.go
index ed4de01..64576b9 100644
--- a/server/server.go
+++ b/server/server.go
@@ -58,6 +58,8 @@ func NewOvsdbServer(db Database, models ...DatabaseModel) (*OvsdbServer, error)
 	o.srv.Handle("transact", o.Transact)
 	o.srv.Handle("cancel", o.Cancel)
 	o.srv.Handle("monitor", o.Monitor)
+	o.srv.Handle("monitor_cond", o.MonitorCond)
+	o.srv.Handle("monitor_cond_since", o.MonitorCondSince)
 	o.srv.Handle("monitor_cancel", o.MonitorCancel)
 	o.srv.Handle("steal", o.Steal)
 	o.srv.Handle("unlock", o.Unlock)
@@ -177,8 +179,9 @@ func (o *OvsdbServer) Transact(client *rpc2.Client, args []json.RawMessage, repl
 	}
 	response, updates := o.transact(db, ops)
 	*reply = response
-	o.processMonitors(updates)
-	return o.db.Commit(db, updates)
+	transactionID := uuid.New()
+	o.processMonitors(transactionID, updates)
+	return o.db.Commit(db, transactionID, updates)
 }
 
 func deepCopy(a ovsdb.TableUpdates) (ovsdb.TableUpdates, error) {
@@ -191,6 +194,16 @@ func deepCopy(a ovsdb.TableUpdates) (ovsdb.TableUpdates, error) {
 	return b, err
 }
 
+func deepCopy2(a ovsdb.TableUpdates2) (ovsdb.TableUpdates2, error) {
+	var b ovsdb.TableUpdates2
+	raw, err := json.Marshal(a)
+	if err != nil {
+		return b, err
+	}
+	err = json.Unmarshal(raw, &b)
+	return b, err
+}
+
 // Cancel cancels the last transaction
 func (o *OvsdbServer) Cancel(client *rpc2.Client, args []interface{}, reply *[]interface{}) error {
 	return fmt.Errorf("not implemented")
@@ -237,6 +250,84 @@ func (o *OvsdbServer) Monitor(client *rpc2.Client, args []json.RawMessage, reply
 	return nil
 }
 
+// MonitorCond monitors a given database table and provides updates to the client via an RPC callback
+func (o *OvsdbServer) MonitorCond(client *rpc2.Client, args []json.RawMessage, reply *ovsdb.TableUpdates2) error {
+	var db string
+	if err := json.Unmarshal(args[0], &db); 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")
+	}
+	value := string(args[1])
+	var request map[string]*ovsdb.MonitorRequest
+	if err := json.Unmarshal(args[2], &request); err != nil {
+		return err
+	}
+	o.monitorMutex.Lock()
+	defer o.monitorMutex.Unlock()
+	clientMonitors, ok := o.monitors[client]
+	if !ok {
+		o.monitors[client] = newConnectionMonitors()
+	} else {
+		if _, ok := clientMonitors.monitors[value]; ok {
+			return fmt.Errorf("monitor with that value already exists")
+		}
+	}
+	tableUpdates := make(ovsdb.TableUpdates2)
+	for t, request := range request {
+		rows := o.Select(db, t, nil, request.Columns)
+		for i := range rows.Rows {
+			tu := make(ovsdb.TableUpdate2)
+			uuid := rows.Rows[i]["_uuid"].(ovsdb.UUID).GoUUID
+			tu[uuid] = &ovsdb.RowUpdate2{Initial: &rows.Rows[i]}
+			tableUpdates.AddTableUpdate(t, tu)
+		}
+	}
+	*reply = tableUpdates
+	o.monitors[client].monitors[value] = newConditionalMonitor(value, request, client)
+	return nil
+}
+
+// MonitorCondSince monitors a given database table and provides updates to the client via an RPC callback
+func (o *OvsdbServer) MonitorCondSince(client *rpc2.Client, args []json.RawMessage, reply *ovsdb.MonitorCondSinceReply) error {
+	var db string
+	if err := json.Unmarshal(args[0], &db); 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")
+	}
+	value := string(args[1])
+	var request map[string]*ovsdb.MonitorRequest
+	if err := json.Unmarshal(args[2], &request); err != nil {
+		return err
+	}
+	o.monitorMutex.Lock()
+	defer o.monitorMutex.Unlock()
+	clientMonitors, ok := o.monitors[client]
+	if !ok {
+		o.monitors[client] = newConnectionMonitors()
+	} else {
+		if _, ok := clientMonitors.monitors[value]; ok {
+			return fmt.Errorf("monitor with that value already exists")
+		}
+	}
+	tableUpdates := make(ovsdb.TableUpdates2)
+	for t, request := range request {
+		rows := o.Select(db, t, nil, request.Columns)
+		for i := range rows.Rows {
+			tu := make(ovsdb.TableUpdate2)
+			uuid := rows.Rows[i]["_uuid"].(ovsdb.UUID).GoUUID
+			tu[uuid] = &ovsdb.RowUpdate2{Initial: &rows.Rows[i]}
+			tableUpdates.AddTableUpdate(t, tu)
+		}
+	}
+	*reply = ovsdb.MonitorCondSinceReply{Found: false, LastTransactionID: "00000000-0000-0000-000000000000", Updates: tableUpdates}
+	o.monitors[client].monitors[value] = newConditionalSinceMonitor(value, request, client)
+	return nil
+}
+
 // MonitorCancel cancels a monitor on a given table
 func (o *OvsdbServer) MonitorCancel(client *rpc2.Client, args []interface{}, reply *[]interface{}) error {
 	return fmt.Errorf("not implemented")
@@ -265,15 +356,26 @@ func (o *OvsdbServer) Echo(client *rpc2.Client, args []interface{}, reply *[]int
 	return nil
 }
 
-func (o *OvsdbServer) processMonitors(update ovsdb.TableUpdates) {
+func (o *OvsdbServer) processMonitors(id uuid.UUID, update ovsdb.TableUpdates2) {
 	o.monitorMutex.RLock()
 	for _, c := range o.monitors {
 		for _, m := range c.monitors {
-			// Deep copy for every monitor since each one filters
-			// the update for relevant tables and removes items
-			// from the update array
-			dbUpdates, _ := deepCopy(update)
-			m.Send(dbUpdates)
+			switch m.kind {
+			case monitorKindOriginal:
+				var updates ovsdb.TableUpdates
+				updates.FromTableUpdates2(update)
+				// Deep copy for every monitor since each one filters
+				// the update for relevant tables and removes items
+				// from the update array
+				dbUpdates, _ := deepCopy(updates)
+				m.Send(dbUpdates)
+			case monitorKindConditional:
+				dbUpdates, _ := deepCopy2(update)
+				m.Send2(dbUpdates)
+			case monitorKindConditionalSince:
+				dbUpdates, _ := deepCopy2(update)
+				m.Send3(id, dbUpdates)
+			}
 		}
 	}
 	o.monitorMutex.RUnlock()
diff --git a/server/transact.go b/server/transact.go
index 6585b9f..21d3d1b 100644
--- a/server/transact.go
+++ b/server/transact.go
@@ -10,9 +10,9 @@ import (
 	"github.com/ovn-org/libovsdb/ovsdb"
 )
 
-func (o *OvsdbServer) transact(name string, operations []ovsdb.Operation) ([]ovsdb.OperationResult, ovsdb.TableUpdates) {
+func (o *OvsdbServer) transact(name string, operations []ovsdb.Operation) ([]ovsdb.OperationResult, ovsdb.TableUpdates2) {
 	results := []ovsdb.OperationResult{}
-	updates := make(ovsdb.TableUpdates)
+	updates := make(ovsdb.TableUpdates2)
 	for _, op := range operations {
 		switch op.Op {
 		case ovsdb.OperationInsert:
@@ -65,7 +65,7 @@ func (o *OvsdbServer) transact(name string, operations []ovsdb.Operation) ([]ovs
 	return results, updates
 }
 
-func (o *OvsdbServer) Insert(database string, table string, rowUUID string, row ovsdb.Row) (ovsdb.OperationResult, ovsdb.TableUpdates) {
+func (o *OvsdbServer) Insert(database string, table string, rowUUID string, row ovsdb.Row) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
 	if !o.db.Exists(database) {
 		return ovsdb.OperationResult{
 			Error: "database does not exist",
@@ -134,11 +134,12 @@ func (o *OvsdbServer) Insert(database string, table string, rowUUID string, row
 	result := ovsdb.OperationResult{
 		UUID: ovsdb.UUID{GoUUID: rowUUID},
 	}
-	return result, ovsdb.TableUpdates{
+	return result, ovsdb.TableUpdates2{
 		table: {
 			rowUUID: {
-				New: &resultRow,
-				Old: nil,
+				Insert: &resultRow,
+				New:    &resultRow,
+				Old:    nil,
 			},
 		},
 	}
@@ -173,7 +174,7 @@ func (o *OvsdbServer) Select(database string, table string, where []ovsdb.Condit
 	}
 }
 
-func (o *OvsdbServer) Update(database, table string, where []ovsdb.Condition, row ovsdb.Row) (ovsdb.OperationResult, ovsdb.TableUpdates) {
+func (o *OvsdbServer) Update(database, table string, where []ovsdb.Condition, row ovsdb.Row) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
 	if !o.db.Exists(database) {
 		return ovsdb.OperationResult{
 			Error: "database does not exist",
@@ -185,7 +186,7 @@ func (o *OvsdbServer) Update(database, table string, where []ovsdb.Condition, ro
 
 	m := mapper.NewMapper(dbModel.Schema)
 	schema := dbModel.Schema.Table(table)
-	tableUpdate := make(ovsdb.TableUpdate)
+	tableUpdate := make(ovsdb.TableUpdate2)
 	rows, err := o.db.List(database, table, where...)
 	if err != nil {
 		return ovsdb.OperationResult{
@@ -217,6 +218,7 @@ func (o *OvsdbServer) Update(database, table string, where []ovsdb.Condition, ro
 			panic(err)
 		}
 
+		rowDelta := ovsdb.NewRow()
 		for column, value := range row {
 			colSchema := schema.Column(column)
 			if colSchema == nil {
@@ -233,6 +235,14 @@ func (o *OvsdbServer) Update(database, table string, where []ovsdb.Condition, ro
 					Details: fmt.Sprintf("column %s is of table %s not mutable", column, table),
 				}, nil
 			}
+			old, err := info.FieldByColumn(column)
+			if err != nil {
+				panic(err)
+			}
+			oldValue, err := ovsdb.NativeToOvs(colSchema, old)
+			if err != nil {
+				panic(err)
+			}
 			native, err := ovsdb.OvsToNative(colSchema, value)
 			if err != nil {
 				panic(err)
@@ -241,6 +251,7 @@ func (o *OvsdbServer) Update(database, table string, where []ovsdb.Condition, ro
 			if err != nil {
 				panic(err)
 			}
+			rowDelta[column] = diff(oldValue, value)
 		}
 
 		newRow, err := m.NewRow(table, new)
@@ -262,20 +273,21 @@ func (o *OvsdbServer) Update(database, table string, where []ovsdb.Condition, ro
 			}, nil
 		}
 
-		tableUpdate.AddRowUpdate(uuid.(string), &ovsdb.RowUpdate{
-			Old: &oldRow,
-			New: &newRow,
+		tableUpdate.AddRowUpdate(uuid.(string), &ovsdb.RowUpdate2{
+			Modify: &rowDelta,
+			Old:    &oldRow,
+			New:    &newRow,
 		})
 	}
 	// FIXME: We need to filter the returned columns
 	return ovsdb.OperationResult{
 			Count: len(rows),
-		}, ovsdb.TableUpdates{
+		}, ovsdb.TableUpdates2{
 			table: tableUpdate,
 		}
 }
 
-func (o *OvsdbServer) Mutate(database, table string, where []ovsdb.Condition, mutations []ovsdb.Mutation) (ovsdb.OperationResult, ovsdb.TableUpdates) {
+func (o *OvsdbServer) Mutate(database, table string, where []ovsdb.Condition, mutations []ovsdb.Mutation) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
 	if !o.db.Exists(database) {
 		return ovsdb.OperationResult{
 			Error: "database does not exist",
@@ -288,7 +300,7 @@ func (o *OvsdbServer) Mutate(database, table string, where []ovsdb.Condition, mu
 	m := mapper.NewMapper(dbModel.Schema)
 	schema := dbModel.Schema.Table(table)
 
-	tableUpdate := make(ovsdb.TableUpdate)
+	tableUpdate := make(ovsdb.TableUpdate2)
 
 	rows, err := o.db.List(database, table, where...)
 	if err != nil {
@@ -321,6 +333,8 @@ func (o *OvsdbServer) Mutate(database, table string, where []ovsdb.Condition, mu
 		if err != nil {
 			panic(err)
 		}
+
+		rowDelta := ovsdb.NewRow()
 		for _, mutation := range mutations {
 			column := schema.Column(mutation.Column)
 			var nativeValue interface{}
@@ -345,10 +359,12 @@ func (o *OvsdbServer) Mutate(database, table string, where []ovsdb.Condition, mu
 			if err != nil {
 				panic(err)
 			}
-			newValue := mutate(current, mutation.Mutator, nativeValue)
+			newValue, delta := mutate(current, mutation.Mutator, nativeValue)
 			if err := info.SetField(mutation.Column, newValue); err != nil {
 				panic(err)
 			}
+			rowDelta[mutation.Column] = delta
+
 			newRow, err := m.NewRow(table, new)
 			if err != nil {
 				panic(err)
@@ -366,20 +382,19 @@ func (o *OvsdbServer) Mutate(database, table string, where []ovsdb.Condition, mu
 					Error: err.Error(),
 				}, nil
 			}
-			tableUpdate.AddRowUpdate(uuid.(string), &ovsdb.RowUpdate{
-				Old: &oldRow,
-				New: &newRow,
+			tableUpdate.AddRowUpdate(uuid.(string), &ovsdb.RowUpdate2{
+				Modify: &newRow,
 			})
 		}
 	}
 	return ovsdb.OperationResult{
 			Count: len(rows),
-		}, ovsdb.TableUpdates{
+		}, ovsdb.TableUpdates2{
 			table: tableUpdate,
 		}
 }
 
-func (o *OvsdbServer) Delete(database, table string, where []ovsdb.Condition) (ovsdb.OperationResult, ovsdb.TableUpdates) {
+func (o *OvsdbServer) Delete(database, table string, where []ovsdb.Condition) (ovsdb.OperationResult, ovsdb.TableUpdates2) {
 	if !o.db.Exists(database) {
 		return ovsdb.OperationResult{
 			Error: "database does not exist",
@@ -390,7 +405,7 @@ func (o *OvsdbServer) Delete(database, table string, where []ovsdb.Condition) (o
 	o.modelsMutex.Unlock()
 	m := mapper.NewMapper(dbModel.Schema)
 	schema := dbModel.Schema.Table(table)
-	tableUpdate := make(ovsdb.TableUpdate)
+	tableUpdate := make(ovsdb.TableUpdate2)
 	rows, err := o.db.List(database, table, where...)
 	if err != nil {
 		panic(err)
@@ -402,14 +417,13 @@ func (o *OvsdbServer) Delete(database, table string, where []ovsdb.Condition) (o
 		if err != nil {
 			panic(err)
 		}
-		tableUpdate.AddRowUpdate(uuid.(string), &ovsdb.RowUpdate{
-			Old: &oldRow,
-			New: nil,
+		tableUpdate.AddRowUpdate(uuid.(string), &ovsdb.RowUpdate2{
+			Delete: &oldRow,
 		})
 	}
 	return ovsdb.OperationResult{
 			Count: len(rows),
-		}, ovsdb.TableUpdates{
+		}, ovsdb.TableUpdates2{
 			table: tableUpdate,
 		}
 }
@@ -438,3 +452,71 @@ func (o *OvsdbServer) Assert(database, table, lock string) ovsdb.OperationResult
 	e := ovsdb.NotSupported{}
 	return ovsdb.OperationResult{Error: e.Error()}
 }
+
+func diff(a interface{}, b interface{}) interface{} {
+	switch a.(type) {
+	case ovsdb.OvsSet:
+		// original value
+		original := a.(ovsdb.OvsSet)
+		// replacement value
+		replacement := b.(ovsdb.OvsSet)
+		var c []interface{}
+		for _, originalElem := range original.GoSet {
+			found := false
+			for _, replacementElem := range replacement.GoSet {
+				if originalElem == replacementElem {
+					found = true
+					break
+				}
+			}
+			if !found {
+				// remove from client
+				c = append(c, originalElem)
+			}
+		}
+		for _, replacementElem := range replacement.GoSet {
+			found := false
+			for _, originalElem := range original.GoSet {
+				if replacementElem == originalElem {
+					found = true
+					break
+				}
+			}
+			if !found {
+				// add to client
+				c = append(c, replacementElem)
+			}
+		}
+		cSet, _ := ovsdb.NewOvsSet(c)
+		return cSet
+	case ovsdb.OvsMap:
+		originalMap := a.(ovsdb.OvsMap)
+		replacementMap := b.(ovsdb.OvsMap)
+		c := make(map[interface{}]interface{})
+		for k, v := range originalMap.GoMap {
+			// if key exists in replacement map
+			if _, ok := replacementMap.GoMap[k]; ok {
+				// and values are not equal
+				if originalMap.GoMap[k] != replacementMap.GoMap[k] {
+					// add to diff
+					c[k] = replacementMap.GoMap[k]
+				}
+			} else {
+				// if key does not exist in replacement map
+				// add old value so it's deleted by client
+				c[k] = v
+			}
+		}
+		for k, v := range replacementMap.GoMap {
+			// if key does not exist in original map
+			if _, ok := originalMap.GoMap[k]; !ok {
+				// add old value so it's added by client
+				c[k] = v
+			}
+		}
+		cMap, _ := ovsdb.NewOvsMap(c)
+		return cMap
+	default:
+		return b
+	}
+}


Test patch:
diff --git a/cache/cache_test.go b/cache/cache_test.go
index a9450b5..44aba58 100644
--- a/cache/cache_test.go
+++ b/cache/cache_test.go
@@ -1,9 +1,9 @@
 package cache
 
 import (
-	"testing"
-
 	"encoding/json"
+	"reflect"
+	"testing"
 
 	"github.com/ovn-org/libovsdb/mapper"
 	"github.com/ovn-org/libovsdb/model"
@@ -750,6 +750,153 @@ func TestTableCache_populate(t *testing.T) {
 	assert.False(t, ok)
 }
 
+func TestTableCachePopulate(t *testing.T) {
+	t.Log("Create")
+	db, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &testModel{}})
+	assert.Nil(t, err)
+	var schema ovsdb.DatabaseSchema
+	err = json.Unmarshal([]byte(`
+		 {"name": "TestDB",
+		  "tables": {
+		    "Open_vSwitch": {
+			  "indexes": [["foo"]],
+		      "columns": {
+		        "foo": {
+			  "type": "string"
+			},
+			"bar": {
+				"type": "string"
+			  }
+		      }
+		    }
+		 }
+	     }
+	`), &schema)
+	assert.Nil(t, err)
+	tc, err := NewTableCache(&schema, db, nil)
+	assert.Nil(t, err)
+
+	testRow := ovsdb.Row(map[string]interface{}{"_uuid": "test", "foo": "bar"})
+	testRowModel := &testModel{UUID: "test", Foo: "bar"}
+	updates := ovsdb.TableUpdates{
+		"Open_vSwitch": {
+			"test": &ovsdb.RowUpdate{
+				Old: nil,
+				New: &testRow,
+			},
+		},
+	}
+	tc.Populate(updates)
+
+	got := tc.Table("Open_vSwitch").Row("test")
+	assert.Equal(t, testRowModel, got)
+
+	t.Log("Update")
+	updatedRow := ovsdb.Row(map[string]interface{}{"_uuid": "test", "foo": "quux"})
+	updatedRowModel := &testModel{UUID: "test", Foo: "quux"}
+	updates["Open_vSwitch"]["test"] = &ovsdb.RowUpdate{
+		Old: &testRow,
+		New: &updatedRow,
+	}
+	tc.Populate(updates)
+
+	got = tc.cache["Open_vSwitch"].cache["test"]
+	assert.Equal(t, updatedRowModel, got)
+
+	t.Log("Delete")
+	updates["Open_vSwitch"]["test"] = &ovsdb.RowUpdate{
+		Old: &updatedRow,
+		New: nil,
+	}
+
+	tc.Populate(updates)
+
+	_, ok := tc.cache["Open_vSwitch"].cache["test"]
+	assert.False(t, ok)
+}
+
+func TestTableCachePopulate2(t *testing.T) {
+	db, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &testModel{}})
+	assert.Nil(t, err)
+	var schema ovsdb.DatabaseSchema
+	err = json.Unmarshal([]byte(`
+		 {"name": "TestDB",
+		  "tables": {
+		    "Open_vSwitch": {
+			  "indexes": [["foo"]],
+		      "columns": {
+		        "foo": {
+			  "type": "string"
+			},
+			"bar": {
+				"type": "string"
+			  }
+		      }
+		    }
+		 }
+	     }
+	`), &schema)
+	assert.Nil(t, err)
+	tc, err := NewTableCache(&schema, db, nil)
+	assert.Nil(t, err)
+
+	testRow := ovsdb.Row(map[string]interface{}{"_uuid": "test", "foo": "bar"})
+	testRowModel := &testModel{UUID: "test", Foo: "bar"}
+	updates := ovsdb.TableUpdates2{
+		"Open_vSwitch": {
+			"test": &ovsdb.RowUpdate2{
+				Initial: &testRow,
+			},
+		},
+	}
+
+	t.Log("Initial")
+	tc.Populate2(updates)
+	got := tc.Table("Open_vSwitch").Row("test")
+	assert.Equal(t, testRowModel, got)
+
+	t.Log("Insert")
+	testRow2 := ovsdb.Row(map[string]interface{}{"_uuid": "test2", "foo": "bar2"})
+	testRowModel2 := &testModel{UUID: "test2", Foo: "bar2"}
+	updates = ovsdb.TableUpdates2{
+		"Open_vSwitch": {
+			"test2": &ovsdb.RowUpdate2{
+				Insert: &testRow2,
+			},
+		},
+	}
+	tc.Populate2(updates)
+	got = tc.Table("Open_vSwitch").Row("test2")
+	assert.Equal(t, testRowModel2, got)
+
+	t.Log("Update")
+	updatedRow := ovsdb.Row(map[string]interface{}{"foo": "quux"})
+	updatedRowModel := &testModel{UUID: "test", Foo: "quux"}
+	updates = ovsdb.TableUpdates2{
+		"Open_vSwitch": {
+			"test": &ovsdb.RowUpdate2{
+				Modify: &updatedRow,
+			},
+		},
+	}
+	tc.Populate2(updates)
+	got = tc.cache["Open_vSwitch"].cache["test"]
+	assert.Equal(t, updatedRowModel, got)
+
+	t.Log("Delete")
+	deletedRow := ovsdb.Row(map[string]interface{}{"_uuid": "test", "foo": "quux"})
+	updates = ovsdb.TableUpdates2{
+		"Open_vSwitch": {
+			"test": &ovsdb.RowUpdate2{
+				Delete: &deletedRow,
+			},
+		},
+	}
+	tc.Populate2(updates)
+	_, ok := tc.cache["Open_vSwitch"].cache["test"]
+	assert.False(t, ok)
+}
+
 func TestEventProcessor_AddEvent(t *testing.T) {
 	ep := newEventProcessor(16)
 	var events []event
@@ -1025,3 +1172,113 @@ func TestTableCacheRowByModelMultiIndex(t *testing.T) {
 		assert.Nil(t, baz)
 	})
 }
+
+func TestTableCacheApplyModifications(t *testing.T) {
+	type testDBModel struct {
+		Value string            `ovsdb:"value"`
+		Set   []string          `ovsdb:"set"`
+		Map   map[string]string `ovsdb:"map"`
+		Map2  map[string]string `ovsdb:"map2"`
+	}
+	aFooSet, _ := ovsdb.NewOvsSet([]string{"foo"})
+	aFooBarSet, _ := ovsdb.NewOvsSet([]string{"foo", "bar"})
+	aFooMap, _ := ovsdb.NewOvsMap(map[string]string{"foo": "bar"})
+	aBarMap, _ := ovsdb.NewOvsMap(map[string]string{"bar": "baz"})
+	aBarBazMap, _ := ovsdb.NewOvsMap(map[string]string{
+		"bar": "baz",
+		"baz": "quux",
+	})
+	tests := []struct {
+		name     string
+		update   ovsdb.Row
+		base     *testDBModel
+		expected *testDBModel
+	}{
+		{
+			"replace value",
+			ovsdb.Row{"value": "bar"},
+			&testDBModel{Value: "foo"},
+			&testDBModel{Value: "bar"},
+		},
+
+		{
+			"add to set",
+			ovsdb.Row{"set": aFooSet},
+			&testDBModel{Set: []string{}},
+			&testDBModel{Set: []string{"foo"}},
+		},
+		{
+			"remove from set",
+			ovsdb.Row{"set": aFooSet},
+			&testDBModel{Set: []string{"foo"}},
+			&testDBModel{Set: []string{}},
+		},
+		{
+			"add and remove from set",
+			ovsdb.Row{"set": aFooBarSet},
+			&testDBModel{Set: []string{"foo"}},
+			&testDBModel{Set: []string{"bar"}},
+		},
+
+		{
+			"replace map value",
+			ovsdb.Row{"map": aFooMap},
+			&testDBModel{Map: map[string]string{"foo": "baz"}},
+			&testDBModel{Map: map[string]string{"foo": "bar"}},
+		},
+		{
+			"add map key",
+			ovsdb.Row{"map": aBarMap},
+			&testDBModel{Map: map[string]string{"foo": "bar"}},
+			&testDBModel{Map: map[string]string{"foo": "bar", "bar": "baz"}},
+		},
+		{
+			"delete map key",
+			ovsdb.Row{"map": aFooMap},
+			&testDBModel{Map: map[string]string{"foo": "bar"}},
+			&testDBModel{Map: map[string]string{}},
+		},
+		{
+			"multiple map operations",
+			ovsdb.Row{"map": aBarBazMap, "map2": aFooMap},
+			&testDBModel{Map: map[string]string{"foo": "bar"}},
+			&testDBModel{
+				Map:  map[string]string{"foo": "bar", "bar": "baz", "baz": "quux"},
+				Map2: map[string]string{"foo": "bar"},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			db, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &testModel{}})
+			assert.Nil(t, err)
+			var schema ovsdb.DatabaseSchema
+			err = json.Unmarshal([]byte(`
+			  {
+				"name": "TestDB",
+				"tables": {
+				  "Open_vSwitch": {
+				    "indexes": [["foo"]],
+					"columns": {
+					  "value": { "type": "string" },
+					  "set": { "type": { "key": { "type": "string" }, "min": 0,	"max": "unlimited" } },
+					  "map": { "type": { "key": "string", "max": "unlimited", "min": 0, "value": "string" } },
+					  "map2": { "type": { "key": "string", "max": "unlimited", "min": 0, "value": "string" } }
+					}
+				  }
+				}
+			  }
+			`), &schema)
+			require.NoError(t, err)
+			tc, err := NewTableCache(&schema, db, nil)
+			assert.Nil(t, err)
+			original := model.Clone(tt.base).(*testDBModel)
+			err = tc.ApplyModifications("Open_vSwitch", original, tt.update)
+			require.NoError(t, err)
+			require.Equal(t, tt.expected, original)
+			if reflect.DeepEqual(original, tt.base) {
+				t.Error("original and base are equal")
+			}
+		})
+	}
+}
diff --git a/client/monitor_test.go b/client/monitor_test.go
new file mode 100644
index 0000000..b1c28d3
--- /dev/null
+++ b/client/monitor_test.go
@@ -0,0 +1,19 @@
+package client
+
+import (
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestWithTable(t *testing.T) {
+	client, err := newOVSDBClient(defDB)
+	assert.NoError(t, err)
+	m := newMonitor()
+	opt := WithTable(&OpenvSwitch{})
+
+	err = opt(client, m)
+	assert.NoError(t, err)
+
+	assert.Equal(t, 1, len(m.Tables))
+}
diff --git a/model/model_test.go b/model/model_test.go
index 1579fad..6eaaee4 100644
--- a/model/model_test.go
+++ b/model/model_test.go
@@ -334,3 +334,13 @@ func TestValidate(t *testing.T) {
 	}
 
 }
+
+func TestClone(t *testing.T) {
+	a := &modelB{UID: "foo", Foo: "bar", Bar: "baz"}
+	b := Clone(a).(*modelB)
+	assert.Equal(t, a, b)
+	a.UID = "baz"
+	assert.NotEqual(t, a, b)
+	b.UID = "quux"
+	assert.NotEqual(t, a, b)
+}
diff --git a/server/monitor_test.go b/server/monitor_test.go
index ffac715..c0071d8 100644
--- a/server/monitor_test.go
+++ b/server/monitor_test.go
@@ -27,50 +27,50 @@ func TestMonitorFilter(t *testing.T) {
 	}
 	tests := []struct {
 		name     string
-		update   ovsdb.TableUpdates
-		expected ovsdb.TableUpdates
+		update   ovsdb.TableUpdates2
+		expected ovsdb.TableUpdates2
 	}{
 		{
 			"not filtered",
-			ovsdb.TableUpdates{
-				"Bridge": ovsdb.TableUpdate{
-					"foo": &ovsdb.RowUpdate{
-						Old: nil, New: &bridgeRow,
+			ovsdb.TableUpdates2{
+				"Bridge": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRow,
 					},
 				},
 			},
-			ovsdb.TableUpdates{
-				"Bridge": ovsdb.TableUpdate{
-					"foo": &ovsdb.RowUpdate{
-						Old: nil, New: &bridgeRow,
+			ovsdb.TableUpdates2{
+				"Bridge": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRow,
 					},
 				},
 			},
 		},
 		{
 			"removed table",
-			ovsdb.TableUpdates{
-				"Open_vSwitch": ovsdb.TableUpdate{
-					"foo": &ovsdb.RowUpdate{
-						Old: nil, New: &bridgeRow,
+			ovsdb.TableUpdates2{
+				"Open_vSwitch": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRow,
 					},
 				},
 			},
-			ovsdb.TableUpdates{},
+			ovsdb.TableUpdates2{},
 		},
 		{
 			"removed column",
-			ovsdb.TableUpdates{
-				"Bridge": ovsdb.TableUpdate{
-					"foo": &ovsdb.RowUpdate{
-						Old: nil, New: &bridgeRowWithIDs,
+			ovsdb.TableUpdates2{
+				"Bridge": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRowWithIDs,
 					},
 				},
 			},
-			ovsdb.TableUpdates{
-				"Bridge": ovsdb.TableUpdate{
-					"foo": &ovsdb.RowUpdate{
-						Old: nil, New: &bridgeRow,
+			ovsdb.TableUpdates2{
+				"Bridge": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRow,
 					},
 				},
 			},
@@ -78,7 +78,82 @@ func TestMonitorFilter(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			monitor.filter(tt.update)
+			monitor.filter2(tt.update)
+			assert.Equal(t, tt.expected, tt.update)
+		})
+	}
+}
+
+func TestMonitorFilter2(t *testing.T) {
+	monitor := monitor{
+		request: map[string]*ovsdb.MonitorRequest{
+			"Bridge": {
+				Columns: []string{"name"},
+				Select:  ovsdb.NewDefaultMonitorSelect(),
+			},
+		},
+	}
+	bridgeRow := ovsdb.Row{
+		"name": "bar",
+	}
+	bridgeRowWithIDs := ovsdb.Row{
+		"name":         "bar",
+		"external_ids": map[string]string{"foo": "bar"},
+	}
+	tests := []struct {
+		name     string
+		update   ovsdb.TableUpdates2
+		expected ovsdb.TableUpdates2
+	}{
+		{
+			"not filtered",
+			ovsdb.TableUpdates2{
+				"Bridge": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRow,
+					},
+				},
+			},
+			ovsdb.TableUpdates2{
+				"Bridge": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRow,
+					},
+				},
+			},
+		},
+		{
+			"removed table",
+			ovsdb.TableUpdates2{
+				"Open_vSwitch": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRow,
+					},
+				},
+			},
+			ovsdb.TableUpdates2{},
+		},
+		{
+			"removed column",
+			ovsdb.TableUpdates2{
+				"Bridge": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRowWithIDs,
+					},
+				},
+			},
+			ovsdb.TableUpdates2{
+				"Bridge": ovsdb.TableUpdate2{
+					"foo": &ovsdb.RowUpdate2{
+						Insert: &bridgeRow,
+					},
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			monitor.filter2(tt.update)
 			assert.Equal(t, tt.expected, tt.update)
 		})
 	}
diff --git a/server/mutate_test.go b/server/mutate_test.go
index da2ded3..0234a4e 100644
--- a/server/mutate_test.go
+++ b/server/mutate_test.go
@@ -46,7 +46,7 @@ func TestMutateAdd(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := mutate(tt.current, tt.mutator, tt.value)
+			got, _ := mutate(tt.current, tt.mutator, tt.value)
 			if !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("mutate() = %v, want %v", got, tt.want)
 			}
@@ -94,7 +94,7 @@ func TestMutateSubtract(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := mutate(tt.current, tt.mutator, tt.value)
+			got, _ := mutate(tt.current, tt.mutator, tt.value)
 			if !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("mutate() = %v, want %v", got, tt.want)
 			}
@@ -142,7 +142,7 @@ func TestMutateMultiply(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := mutate(tt.current, tt.mutator, tt.value)
+			got, _ := mutate(tt.current, tt.mutator, tt.value)
 			if !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("mutate() = %v, want %v", got, tt.want)
 			}
@@ -189,7 +189,7 @@ func TestMutateDivide(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := mutate(tt.current, tt.mutator, tt.value)
+			got, _ := mutate(tt.current, tt.mutator, tt.value)
 			if !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("mutate() = %v, want %v", got, tt.want)
 			}
@@ -222,7 +222,7 @@ func TestMutateModulo(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := mutate(tt.current, tt.mutator, tt.value)
+			got, _ := mutate(tt.current, tt.mutator, tt.value)
 			if !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("mutate() = %v, want %v", got, tt.want)
 			}
@@ -277,7 +277,7 @@ func TestMutateInsert(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := mutate(tt.current, tt.mutator, tt.value)
+			got, _ := mutate(tt.current, tt.mutator, tt.value)
 			if !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("mutate() = %v, want %v", got, tt.want)
 			}
@@ -337,7 +337,7 @@ func TestMutateDelete(t *testing.T) {
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
-			got := mutate(tt.current, tt.mutator, tt.value)
+			got, _ := mutate(tt.current, tt.mutator, tt.value)
 			if !reflect.DeepEqual(got, tt.want) {
 				t.Errorf("mutate() = %v, want %v", got, tt.want)
 			}
diff --git a/server/server_integration_test.go b/server/server_integration_test.go
index 9417c2b..a794e78 100644
--- a/server/server_integration_test.go
+++ b/server/server_integration_test.go
@@ -215,7 +215,6 @@ func TestClientServerMonitor(t *testing.T) {
 			}
 		},
 		UpdateFunc: func(table string, old, new model.Model) {
-			fmt.Println("got an update")
 			if table == "Open_vSwitch" {
 				ov := new.(*ovsType)
 				assert.Equal(t, 1, len(ov.Bridges))
@@ -344,7 +343,8 @@ func TestClientServerInsertAndDelete(t *testing.T) {
 func TestClientServerInsertDuplicate(t *testing.T) {
 	defDB, err := model.NewDBModel("Open_vSwitch", map[string]model.Model{
 		"Open_vSwitch": &ovsType{},
-		"Bridge":       &bridgeType{}})
+		"Bridge":       &bridgeType{},
+	})
 	require.Nil(t, err)
 
 	schema, err := getSchema()
@@ -496,6 +496,6 @@ func TestClientServerInsertAndUpdate(t *testing.T) {
 		if err != nil {
 			return false
 		}
-		return reflect.DeepEqual(br, bridgeRow)
+		return reflect.DeepEqual(br.ExternalIds, bridgeRow.ExternalIds)
 	}, 2*time.Second, 500*time.Millisecond)
 }
diff --git a/server/server_test.go b/server/server_test.go
index 1a6da6e..604300d 100644
--- a/server/server_test.go
+++ b/server/server_test.go
@@ -104,7 +104,7 @@ func TestOvsdbServerMonitor(t *testing.T) {
 	updates.Merge(update3)
 	_, update4 := o.Insert("Open_vSwitch", "Bridge", quuxUUID, ovsdb.Row{"name": "quux"})
 	updates.Merge(update4)
-	err = o.db.Commit("Open_vSwitch", updates)
+	err = o.db.Commit("Open_vSwitch", uuid.New(), updates)
 	require.NoError(t, err)
 
 	db, err := json.Marshal("Open_vSwitch")
diff --git a/server/transact_test.go b/server/transact_test.go
index 497e913..50be96c 100644
--- a/server/transact_test.go
+++ b/server/transact_test.go
@@ -56,7 +56,7 @@ func TestMutateOp(t *testing.T) {
 	require.Nil(t, err)
 
 	updates.Merge(update2)
-	err = o.db.Commit("Open_vSwitch", updates)
+	err = o.db.Commit("Open_vSwitch", uuid.New(), updates)
 	require.NoError(t, err)
 
 	gotResult, gotUpdate := o.Mutate(
@@ -73,13 +73,11 @@ func TestMutateOp(t *testing.T) {
 
 	bridgeSet, err := ovsdb.NewOvsSet([]ovsdb.UUID{{GoUUID: bridgeUUID}})
 	assert.Nil(t, err)
-	assert.Equal(t, ovsdb.TableUpdates{
-		"Open_vSwitch": ovsdb.TableUpdate{
-			ovsUUID: &ovsdb.RowUpdate{
-				Old: &ovsdb.Row{
-					"_uuid": ovsdb.UUID{GoUUID: ovsUUID},
-				},
-				New: &ovsdb.Row{
+	assert.Equal(t, ovsdb.TableUpdates2{
+		"Open_vSwitch": ovsdb.TableUpdate2{
+			ovsUUID: &ovsdb.RowUpdate2{
+				Modify: &ovsdb.Row{
+					// TODO: _uuid should be filtered
 					"_uuid":   ovsdb.UUID{GoUUID: ovsUUID},
 					"bridges": bridgeSet,
 				},
@@ -104,19 +102,14 @@ func TestMutateOp(t *testing.T) {
 	)
 	assert.Equal(t, ovsdb.OperationResult{Count: 1}, gotResult)
 
-	oldExternalIds, err := ovsdb.NewOvsMap(bridge.ExternalIds)
+	// oldExternalIds, err := ovsdb.NewOvsMap(bridge.ExternalIds)
 	assert.Nil(t, err)
 	newExternalIds, err := ovsdb.NewOvsMap(map[string]string{"waldo": "fred"})
 	assert.Nil(t, err)
-	assert.Equal(t, ovsdb.TableUpdates{
-		"Bridge": ovsdb.TableUpdate{
-			bridgeUUID: &ovsdb.RowUpdate{
-				Old: &ovsdb.Row{
-					"_uuid":        ovsdb.UUID{GoUUID: bridgeUUID},
-					"name":         "foo",
-					"external_ids": oldExternalIds,
-				},
-				New: &ovsdb.Row{
+	assert.Equal(t, ovsdb.TableUpdates2{
+		"Bridge": ovsdb.TableUpdate2{
+			bridgeUUID: &ovsdb.RowUpdate2{
+				Modify: &ovsdb.Row{
 					"_uuid":        ovsdb.UUID{GoUUID: bridgeUUID},
 					"name":         "foo",
 					"external_ids": newExternalIds,
@@ -125,3 +118,68 @@ func TestMutateOp(t *testing.T) {
 		},
 	}, gotUpdate)
 }
+
+func TestDiff(t *testing.T) {
+	originSet, _ := ovsdb.NewOvsSet([]interface{}{"foo"})
+
+	originSetAdd, _ := ovsdb.NewOvsSet([]interface{}{"bar"})
+	setAddDiff, _ := ovsdb.NewOvsSet([]interface{}{"foo", "bar"})
+
+	originSetDel, _ := ovsdb.NewOvsSet([]interface{}{})
+	setDelDiff, _ := ovsdb.NewOvsSet([]interface{}{"foo"})
+
+	originMap, _ := ovsdb.NewOvsMap(map[interface{}]interface{}{"foo": "bar"})
+
+	originMapAdd, _ := ovsdb.NewOvsMap(map[interface{}]interface{}{"foo": "bar", "baz": "quux"})
+	originMapAddDiff, _ := ovsdb.NewOvsMap(map[interface{}]interface{}{"baz": "quux"})
+
+	originMapDel, _ := ovsdb.NewOvsMap(map[interface{}]interface{}{})
+	originMapDelDiff, _ := ovsdb.NewOvsMap(map[interface{}]interface{}{"foo": "bar"})
+
+	originMapReplace, _ := ovsdb.NewOvsMap(map[interface{}]interface{}{"foo": "baz"})
+	originMapReplaceDiff, _ := ovsdb.NewOvsMap(map[interface{}]interface{}{"foo": "baz"})
+
+	tests := []struct {
+		name     string
+		a        interface{}
+		b        interface{}
+		expected interface{}
+	}{
+		{
+			"add to set",
+			originSet,
+			originSetAdd,
+			setAddDiff,
+		},
+		{
+			"delete from set",
+			originSet,
+			originSetDel,
+			setDelDiff,
+		},
+		{
+			"add to map",
+			originMap,
+			originMapAdd,
+			originMapAddDiff,
+		},
+		{
+			"delete from map",
+			originMap,
+			originMapDel,
+			originMapDelDiff,
+		},
+		{
+			"replace in map",
+			originMap,
+			originMapReplace,
+			originMapReplaceDiff,
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			res := diff(tt.a, tt.b)
+			assert.Equal(t, tt.expected, res)
+		})
+	}
+}
diff --git a/test/ovs/ovs_integration_test.go b/test/ovs/ovs_integration_test.go
index fd91c80..1056a96 100644
--- a/test/ovs/ovs_integration_test.go
+++ b/test/ovs/ovs_integration_test.go
@@ -89,7 +89,12 @@ func (suite *OVSIntegrationSuite) SetupSuite() {
 
 	// give ovsdb-server some time to start up
 
-	_, err = suite.client.MonitorAll(context.TODO())
+	_, err = suite.client.Monitor(context.TODO(),
+		suite.client.NewMonitor(
+			client.WithTable(&ovsType{}),
+			client.WithTable(&bridgeType{}),
+		),
+	)
 	require.NoError(suite.T(), err)
 }
 
@@ -140,10 +145,18 @@ type ipfixType struct {
 	Targets []string `ovsdb:"targets"`
 }
 
+// queueType is the simplified ORM model of the Queue table
+type queueType struct {
+	UUID string `ovsdb:"_uuid"`
+	DSCP *int   `ovsdb:"dscp"`
+}
+
 var defDB, _ = model.NewDBModel("Open_vSwitch", map[string]model.Model{
 	"Open_vSwitch": &ovsType{},
 	"Bridge":       &bridgeType{},
-	"IPFIX":        &ipfixType{}})
+	"IPFIX":        &ipfixType{},
+	"Queue":        &queueType{},
+})
 
 func (suite *OVSIntegrationSuite) TestConnectReconnect() {
 	assert.True(suite.T(), suite.client.Connected())
@@ -216,7 +229,12 @@ func (suite *OVSIntegrationSuite) TestConnectReconnect() {
 	err = suite.client.Echo(context.TODO())
 	assert.NoError(suite.T(), err)
 
-	_, err = suite.client.MonitorAll(context.TODO())
+	_, err = suite.client.Monitor(context.TODO(),
+		suite.client.NewMonitor(
+			client.WithTable(&ovsType{}),
+			client.WithTable(&bridgeType{}),
+		),
+	)
 	require.NoError(suite.T(), err)
 
 	// assert cache has been re-populated
@@ -454,28 +472,30 @@ func (suite *OVSIntegrationSuite) TestColumnSchemaValidationIntegration() {
 }
 
 func (suite *OVSIntegrationSuite) TestMonitorCancelIntegration() {
-	requests := make(map[string]ovsdb.MonitorRequest)
-	requests["Bridge"] = ovsdb.MonitorRequest{
-		Columns: []string{"name"},
-		Select:  ovsdb.NewDefaultMonitorSelect(),
-	}
-
 	monitorID, err := suite.client.Monitor(
 		context.TODO(),
-		suite.client.NewTableMonitor(&ovsType{}),
-		suite.client.NewTableMonitor(&bridgeType{}),
+		suite.client.NewMonitor(
+			client.WithTable(&queueType{}),
+		),
 	)
 	require.NoError(suite.T(), err)
 
+	uuid, err := suite.createQueue("test1", 0)
+	require.NoError(suite.T(), err)
+	require.Eventually(suite.T(), func() bool {
+		q := &queueType{UUID: uuid}
+		err = suite.client.Get(q)
+		return err == nil
+	}, 2*time.Second, 500*time.Millisecond)
+
 	err = suite.client.MonitorCancel(context.TODO(), monitorID)
 	assert.NoError(suite.T(), err)
 
-	uuid, err := suite.createBridge("br-monitor")
+	uuid, err = suite.createQueue("test2", 1)
 	require.NoError(suite.T(), err)
-
-	assert.Eventually(suite.T(), func() bool {
-		br := &bridgeType{UUID: uuid}
-		err = suite.client.Get(br)
+	assert.Never(suite.T(), func() bool {
+		q := &queueType{UUID: uuid}
+		err = suite.client.Get(q)
 		return err == nil
 	}, 2*time.Second, 500*time.Millisecond)
 }
@@ -509,13 +529,17 @@ func (suite *OVSIntegrationSuite) TestUpdate() {
 	err = suite.client.Get(bridgeRow)
 	require.NoError(suite.T(), err)
 
+	// try to modify immutable field
+	bridgeRow.Name = "br-update2"
+	_, err = suite.client.Where(bridgeRow).Update(bridgeRow, &bridgeRow.Name)
+	require.Error(suite.T(), err)
+	bridgeRow.Name = "br-update"
 	// update many fields
-	bridgeRow.UUID = uuid
 	bridgeRow.ExternalIds["baz"] = "foobar"
 	bridgeRow.OtherConfig = map[string]string{"foo": "bar"}
 	ops, err := suite.client.Where(bridgeRow).Update(bridgeRow)
 	require.NoError(suite.T(), err)
-	reply, err := suite.client.Transact(context.TODO(), ops...)
+	reply, err := suite.client.Transact(context.Background(), ops...)
 	require.NoError(suite.T(), err)
 	opErrs, err := ovsdb.CheckOperationResults(reply, ops)
 	require.NoErrorf(suite.T(), err, "%+v", opErrs)
@@ -526,24 +550,17 @@ func (suite *OVSIntegrationSuite) TestUpdate() {
 		if err != nil {
 			return false
 		}
-		return reflect.DeepEqual(bridgeRow, br)
-	}, 2*time.Second, 500*time.Millisecond)
-
-	// try to modify immutable field
-	bridgeRow.Name = "br-update2"
-	_, err = suite.client.Where(bridgeRow).Update(bridgeRow, &bridgeRow.Name)
-	require.Error(suite.T(), err)
-	// set name back again
-	bridgeRow.Name = "br-update"
+		return reflect.DeepEqual(br, bridgeRow)
+	}, 2*time.Second, 50*time.Millisecond)
 
 	newExternalIds := map[string]string{"foo": "bar"}
 	bridgeRow.ExternalIds = newExternalIds
 	ops, err = suite.client.Where(bridgeRow).Update(bridgeRow, &bridgeRow.ExternalIds)
 	require.NoError(suite.T(), err)
-	reply, err = suite.client.Transact(context.TODO(), ops...)
-	require.NoError(suite.T(), err)
-	_, err = ovsdb.CheckOperationResults(reply, ops)
+	reply, err = suite.client.Transact(context.Background(), ops...)
 	require.NoError(suite.T(), err)
+	opErr, err := ovsdb.CheckOperationResults(reply, ops)
+	require.NoErrorf(suite.T(), err, "%Populate2+v", opErr)
 
 	assert.Eventually(suite.T(), func() bool {
 		br := &bridgeType{UUID: uuid}
@@ -551,7 +568,7 @@ func (suite *OVSIntegrationSuite) TestUpdate() {
 		if err != nil {
 			return false
 		}
-		return reflect.DeepEqual(bridgeRow, br)
+		return reflect.DeepEqual(br, bridgeRow)
 	}, 2*time.Second, 500*time.Millisecond)
 }
 
@@ -638,3 +655,17 @@ func (suite *OVSIntegrationSuite) TestCreateIPFIX() {
 	require.Empty(suite.T(), ipfixes)
 
 }
+
+func (suite *OVSIntegrationSuite) createQueue(queueName string, dscp int) (string, error) {
+	q := queueType{
+		DSCP: &dscp,
+	}
+
+	insertOp, err := suite.client.Create(&q)
+	require.NoError(suite.T(), err)
+	reply, err := suite.client.Transact(context.TODO(), insertOp...)
+	require.NoError(suite.T(), err)
+
+	_, err = ovsdb.CheckOperationResults(reply, insertOp)
+	return reply[0].UUID.GoUUID, err
+}