Repository: ovn-org/libovsdb
Pull request #306 (2022-05-10 09:44:33)

Add support for client indexes
A user may specify additional indexes at runtime through the client DB
model at client initialization. Indexes are specified in sets per table.
Each set consists of columns and optionally a key within the column that
can be addressed if the column type is a map.

The main use case is performance. Running a predicate on all the rows of
a table can be costly. Introducing new indexes in the schema is not backwards
compatible or might not align with the general concept of the schema but those
additional indexes might be useful for specific uses of that schema that 
would otherwise rely on running less performant predicates frequently.

For example, with OVN-K, checking whether ~40 LBs are present in the cache
by a predicate that matches by name on a table that contains ~200K rows took 
around 3 seconds. While surely there is different ways to optimize a predicate, 
for OVN-K which gives & frequently addresses LBs by unique names would 
make sense to have them indexed, something that is not possible to do with 
the schema without breaking backward compatibility.

This can be enhanced in the future with more use cases. For example, supporting
primary and secondary client indexes where primary client indexes could be 
checked for duplicates against a transaction before sending it to the server.

This is implemented in the cache tracking client indexes as if they were
schema indexes. Indexes map to a set of uuids instead of a single uuid
where the set will consist of a single uuid for schema index but might
consist of multiple uuids for client indexes. A new `RowsByModel` method
queries the cache considering schema and client indexes. `RowsByCondition`
also makes use of the new client indexes naturally. The legacy
`RowByModel` method considers only schema indexes. Schema indexes have
priority over client indexes in general.

API wise, client indexes are leveraged through the conditional
APIs from `Where(...)` and `WhereAll(...)`.

So where you would have before something like
```
// slow predicate run on all the LB table rows...
WhereCache(func (lb *LoadBalancer) bool {
    return lb.ExternalIds["myIdKey"] == "myIdValue"
}).List(ctx, &results)
```

Now you can have
```
dbModel, err := nbdb.FullDatabaseModel()
db.SetIndexes(map[string][]model.ClientIndex{
  "Load_Balancer": {{Columns: []model.ColumnKey{Column: "external_ids", Key: "myIdKey"}}}
})

// connect and everything....

lb := &LoadBalancer{
    ExternalIds: map[string]string{"myIdKey": "myIdValue"},
}
// quick indexed result
Where(lb).List(ctx, &results)
```

Other changes in this context:

* Cherry-picked @dave-tucker's [api: Allow use of Where for client-side ops](https://github.com/ovn-org/libovsdb/commit/b54c97c4b95409f779814be465a21f252082554a) 
  so that client indexes can be leveraged through the `Where.List` and `WhereAll.List` 
  conditional APIs. 
* As improvements to the above, run the predicates on shallow copies of model, 
  remove the unnecessary clone and return cache matches directly instead of 
  in-directing through RowsByCondition.
* Fixed an issue with RowsByCondition that was implementing an `Or` between the 
  conditions instead of an `And`. also improved performance.
* Instead of translating model conditions to ovsdb conditions and
  sending those to the server, perform a query on the cache through
  `RowsByModel` or `RowsByCondition` and generate uuid based conditions
  off those results. Fallback to the original conditionals if no cache hits are obtained.

  This is similar to what was already being done for the predicate conditional, this PR
  does the same for the equality and explicit conditionals.

  This has two benefits:

  * OVSDB server does not index schema indexes, it only ensures uniqueness
    but operations based on those indexes are not necessarily fast. By
    querying the cache for the uuids we improve the performance of index
    based operations. Users will no longer be required to do a `Get` before any 
    server side op to ensure this is optimized through the uuid.
  * The conditionals APIs can fully benefit of client indexes for both cache and 
    server side ops.
  
**RFCs**

* The `ClientDBModel`API could be better but I did not want to complicate it 
  unnecessarily. Feedback welcome.
* With this change we rely more on the cache. But the cache doesn't seem to
  be aware at all of monitored vs non-monitored columns and will happily index
  zero values from non-monitored columns. I am not worried in excess about it
  in the context of this PR as this is an issue already present, but perhaps we
  ought to do something about it.


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

* **352** of **444**   **(79.28%)**  changed or added relevant lines in **5** files are covered.
* **2** unchanged lines in **2** files lost coverage.
* Overall coverage increased (+**0.7%**) to **74.527%**

---

|  Changes Missing Coverage | Covered Lines | Changed/Added Lines | % |
| :-----|--------------|--------|---: |
| [client/api.go](https://coveralls.io/builds/48981510/source?filename=client%2Fapi.go#L154) | 17 | 21 | 80.95%
| [cache/uuidset.go](https://coveralls.io/builds/48981510/source?filename=cache%2Fuuidset.go#L31) | 53 | 62 | 85.48%
| [client/condition.go](https://coveralls.io/builds/48981510/source?filename=client%2Fcondition.go#L30) | 71 | 85 | 83.53%
| [cache/cache.go](https://coveralls.io/builds/48981510/source?filename=cache%2Fcache.go#L351) | 201 | 224 | 89.73%
| [model/client.go](https://coveralls.io/builds/48981510/source?filename=model%2Fclient.go#L44) | 10 | 52 | 19.23%
<!-- | **Total:** | **352** | **444** | **79.28%** | -->

|  Files with Coverage Reduction | New Missed Lines | % |
| :-----|--------------|--: |
| [client/api.go](https://coveralls.io/builds/48981510/source?filename=client%2Fapi.go#L155) | 1 | 81.01% |
| [client/condition.go](https://coveralls.io/builds/48981510/source?filename=client%2Fcondition.go#L213) | 1 | 84.85% |
<!-- | **Total:** | **2** |  | -->

|  Totals | [![Coverage Status](https://coveralls.io/builds/48981510/badge)](https://coveralls.io/builds/48981510) |
| :-- | --: |
| Change from base [Build 2233077523](https://coveralls.io/builds/48618761): |  0.7% |
| Covered Lines: | 4801 |
| Relevant Lines: | 6442 |

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

jcaamano: Added a couple of commits to benchmark & optimize `RowsByCondition` so that it makes use of indexes as much as possible.

Results before this PR:
```
goos: linux
goarch: amd64
pkg: github.com/ovn-org/libovsdb/cache
cpu: Intel Core Processor (Skylake, IBRS)
BenchmarkRowsByCondition/by_uuid-4         	      38	  28737606 ns/op	 6850409 B/op	  140222 allocs/op
BenchmarkRowsByCondition/by_single_column_squema_index-4         	    1686	    742480 ns/op	     925 B/op	      16 allocs/op
BenchmarkRowsByCondition/by_multi_column_squema_index-4          	      19	  60704196 ns/op	13540612 B/op	  280450 allocs/op
BenchmarkRowsByCondition/by_two_squema_indexes-4                 	     817	   1537409 ns/op	    1516 B/op	      29 allocs/op
BenchmarkRowsByCondition/by_squema_index_and_non-index-4         	      37	  31416413 ns/op	 6772972 B/op	  140246 allocs/op
BenchmarkRowsByCondition/by_single_non_index-4                   	      37	  30643479 ns/op	 6770317 B/op	  140226 allocs/op
BenchmarkRowsByCondition/by_multiple_non_indexes-4               	      19	  60734832 ns/op	13542517 B/op	  280456 allocs/op
PASS
ok  	github.com/ovn-org/libovsdb/cache	8.821s
```

Results before optimization commit (includes using single-column indexes, match all logic and avoids unnecessary clones):
```
goos: linux
goarch: amd64
pkg: github.com/ovn-org/libovsdb/cache
cpu: Intel Core Processor (Skylake, IBRS)
BenchmarkRowsByCondition/by_uuid-4         	  272794	      4057 ns/op	    1181 B/op	      20 allocs/op
BenchmarkRowsByCondition/by_single_column_squema_index-4         	  285651	      4229 ns/op	    1181 B/op	      20 allocs/op
BenchmarkRowsByCondition/by_single_column_client_index-4         	  288646	      4283 ns/op	    1181 B/op	      20 allocs/op
BenchmarkRowsByCondition/by_multi_column_squema_index-4          	     400	   2986806 ns/op	  241852 B/op	   20036 allocs/op
BenchmarkRowsByCondition/by_two_squema_indexes-4                 	  224726	      5065 ns/op	    1645 B/op	      26 allocs/op
BenchmarkRowsByCondition/by_squema_index_and_non-index-4         	  201648	      5581 ns/op	    1749 B/op	      32 allocs/op
BenchmarkRowsByCondition/by_single_non_index-4                   	     387	   3047246 ns/op	  241289 B/op	   20024 allocs/op
BenchmarkRowsByCondition/by_multiple_non_indexes-4               	     392	   2976273 ns/op	  241866 B/op	   20036 allocs/op
PASS
ok  	github.com/ovn-org/libovsdb/cache	11.000s
```

Results after optimization (includes using multi-column indexes):
```
goos: linux
goarch: amd64
pkg: github.com/ovn-org/libovsdb/cache
cpu: Intel Core Processor (Skylake, IBRS)
BenchmarkRowsByCondition/by_uuid-4         	  269812	      4106 ns/op	    1221 B/op	      22 allocs/op
BenchmarkRowsByCondition/by_single_column_squema_index-4         	  198980	      5833 ns/op	    1797 B/op	      36 allocs/op
BenchmarkRowsByCondition/by_single_column_client_index-4         	  178773	      6490 ns/op	    2013 B/op	      45 allocs/op
BenchmarkRowsByCondition/by_multi_column_squema_index-4          	   92376	     12630 ns/op	    5090 B/op	     112 allocs/op
BenchmarkRowsByCondition/by_two_squema_indexes-4                 	  109209	     10330 ns/op	    3957 B/op	      88 allocs/op
BenchmarkRowsByCondition/by_squema_index_and_non-index-4         	  112953	     10330 ns/op	    3957 B/op	      93 allocs/op
BenchmarkRowsByCondition/by_single_non_index-4                   	     397	   3073535 ns/op	  241862 B/op	   20042 allocs/op
BenchmarkRowsByCondition/by_multiple_non_indexes-4               	     391	   2958540 ns/op	  244022 B/op	   20099 allocs/op
PASS
ok  	github.com/ovn-org/libovsdb/cache	10.820s
```

Being able to use multi-column indexes does have a cost for the single-column index performance but I think it makes sense to have it. It's an overall good improvement for a start and I am sure this can be optimized further in various ways. Ideas welcome though if you see easy picks.
jcaamano: Pushed an update that shortcuts RowsByCondition when we have reduced the possible index matches to 1. That will help in most situations where I would think we would expect a single match. As said, this can be tuned and optimized further but don't want to go overboard with no experience.

Other than that just added some additional tests and improvement to godoc  that come to mind as times goes on.

Updated results, added a benchmark that uses many conditions.

Results before this PR:
```
goos: linux
goarch: amd64
pkg: github.com/ovn-org/libovsdb/cache
cpu: Intel Core Processor (Skylake, IBRS)
BenchmarkRowsByCondition/by_uuid-4         	      39	  31181645 ns/op	 7156346 B/op	  140223 allocs/op
BenchmarkRowsByCondition/by_single_column_squema_index-4         	    1608	    719762 ns/op	     956 B/op	      16 allocs/op
BenchmarkRowsByCondition/by_multi_column_squema_index-4          	      16	  65336847 ns/op	14151228 B/op	  280448 allocs/op
BenchmarkRowsByCondition/by_two_squema_indexes-4                 	     800	   1582913 ns/op	    1577 B/op	      29 allocs/op
BenchmarkRowsByCondition/by_squema_index_and_non-index-4         	      33	  33603397 ns/op	 7077108 B/op	  140241 allocs/op
BenchmarkRowsByCondition/by_single_non_index-4                   	      32	  32456669 ns/op	 7076734 B/op	  140228 allocs/op
BenchmarkRowsByCondition/by_multiple_non_indexes-4               	      18	  64228400 ns/op	14151768 B/op	  280449 allocs/op
BenchmarkRowsByCondition/by_many_conditions-4                    	       8	 141383613 ns/op	33114615 B/op	  600961 allocs/op
PASS
ok  	github.com/ovn-org/libovsdb/cache	9.916s
```

Results before optimization commit (includes using single-column indexes, match all logic and avoids unnecessary clones):
```
goos: linux
goarch: amd64
pkg: github.com/ovn-org/libovsdb/cache
cpu: Intel Core Processor (Skylake, IBRS)
BenchmarkRowsByCondition/by_uuid-4         	  207000	      5366 ns/op	    1854 B/op	      32 allocs/op
BenchmarkRowsByCondition/by_single_column_squema_index-4         	  215368	      5775 ns/op	    1854 B/op	      32 allocs/op
BenchmarkRowsByCondition/by_single_column_client_index-4         	     166	   6692064 ns/op	 1842801 B/op	   50041 allocs/op
BenchmarkRowsByCondition/by_multi_column_squema_index-4          	     370	   2896103 ns/op	  242527 B/op	   20048 allocs/op
BenchmarkRowsByCondition/by_two_squema_indexes-4                 	  161007	      6484 ns/op	    2318 B/op	      38 allocs/op
BenchmarkRowsByCondition/by_squema_index_and_non-index-4         	  154776	      6872 ns/op	    2428 B/op	      44 allocs/op
BenchmarkRowsByCondition/by_single_non_index-4                   	     387	   3063313 ns/op	  241962 B/op	   20036 allocs/op
BenchmarkRowsByCondition/by_multiple_non_indexes-4               	     396	   2947208 ns/op	  242533 B/op	   20048 allocs/op
BenchmarkRowsByCondition/by_many_conditions-4                    	   89371	     12340 ns/op	    5422 B/op	      93 allocs/op
PASS
ok  	github.com/ovn-org/libovsdb/cache	13.949s
```

Results after optimization (includes using multi-column indexes):
```
goos: linux
goarch: amd64
pkg: github.com/ovn-org/libovsdb/cache
cpu: Intel Core Processor (Skylake, IBRS)
BenchmarkRowsByCondition/by_uuid-4         	  208352	      5577 ns/op	    1894 B/op	      34 allocs/op
BenchmarkRowsByCondition/by_single_column_squema_index-4         	  160498	      7098 ns/op	    2428 B/op	      47 allocs/op
BenchmarkRowsByCondition/by_single_column_client_index-4         	  111760	      9768 ns/op	    3677 B/op	      75 allocs/op
BenchmarkRowsByCondition/by_multi_column_squema_index-4          	   79140	     14264 ns/op	    5667 B/op	     123 allocs/op
BenchmarkRowsByCondition/by_two_squema_indexes-4                 	  149360	      7876 ns/op	    2910 B/op	      54 allocs/op
BenchmarkRowsByCondition/by_squema_index_and_non-index-4         	  128370	      7967 ns/op	    2910 B/op	      54 allocs/op
BenchmarkRowsByCondition/by_single_non_index-4                   	     390	   3027351 ns/op	  242536 B/op	   20054 allocs/op
BenchmarkRowsByCondition/by_multiple_non_indexes-4               	     403	   2975605 ns/op	  244693 B/op	   20111 allocs/op
BenchmarkRowsByCondition/by_many_conditions-4                    	   95650	     12171 ns/op	    5694 B/op	      89 allocs/op
PASS
ok  	github.com/ovn-org/libovsdb/cache	12.210s
```

Was curious to have a benchmark of index vs predicate. It turns out there was one for predicates but I don't it was working fine (too good to be true) so I updated it. Results: 
```
BenchmarkAPIListPredicate/ApiListPredicate:_predicate_returns_none-4         	     363	   3117351 ns/op	  930387 B/op	   20010 allocs/op
BenchmarkAPIListPredicate/ApiListPredicate:_predicate_returns_all-4          	     225	   5167648 ns/op	 2445557 B/op	   30223 allocs/op
BenchmarkAPIListPredicate/ApiListPredicate:_predicate_on_an_arbitrary_condition-4         	     332	   3627668 ns/op	 1123775 B/op	   21152 allocs/op
BenchmarkAPIListPredicate/ApiListPredicate:_predicate_matches_name-4                      	     363	   3292371 ns/op	  930691 B/op	   20012 allocs/op
BenchmarkAPIListPredicate/ApiListPredicate:_by_index,_no_predicate-4                      	   45225	     25121 ns/op	   84076 B/op	      32 allocs/op
```
jcaamano: Would `ColumnKey` be better served typed simpler as `map[string][]interface{}` as a map of columns to optional keys. 
Or at least 
```
struct {
  Column string
  Keys []interface{}
}
````
Really nothing prevents using multiple keys within the same map column and thus might be cleaner that way.
flavio-fernandes: SUPER NIT: fix typo in commit description of commit
https://github.com/ovn-org/libovsdb/pull/306/commits/0d2fa37e8d4fb3d291768aaac80e32c4c66a85da
```
Do use RowsByCondition to add support to match against the cache for the
 --------------------------------------------------------^
explicit conditional.
```
flavio-fernandes: @amorenoz @dave-tucker PTAL as well!
dcbw: @jcaamano DCO check failure?
jcaamano: > @jcaamano DCO check failure?
> 
> EDIT: ah your fixup. I assume you'll rebase + squash before merge

rebased & squashed

I am going to run this by OVN-K to make sure everything looks fine. 

I would also like feedback from @dave-tucker and @amorenoz at least on the API changes and the RFCs.

Patch:
diff --git a/README.md b/README.md
index ce7e463..e551cd4 100644
--- a/README.md
+++ b/README.md
@@ -62,8 +62,8 @@ to a Model's field, a `ovsdb.ConditionFunction` and a value. The type of the val
 The resulting `ConditionalAPI` will create one operation per condition, so all the rows that match *any* of the specified conditions will be affected.
 
 If no conditions are provided, `Where()` will create a `ConditionalAPI` based on the index information that the provided Model contains.
-It will check the field corresponding to the `_uuid` column as well as all the other schema-defined indexes. The first field with
-non-default value will be used for the condition.
+It will check the field corresponding to the `_uuid` column as well as all the other schema-defined or client-defined indexes in that order of priority.
+The first available index will be used to generate a condition.
 
 **WhereAll()**: `WhereAll()` behaves like `Where()` but with *AND* semantics. The resulting `ConditionalAPI` will put all the
 conditions into a single operation. Therefore the operation will affect the rows that satisfy *all* the conditions.
@@ -81,6 +81,41 @@ create an equality condition (using `ovsdb.ConditionEqual`) on the `_uuid` field
 
 The table is inferred from the type that the function accepts as only argument.
 
+### Client indexes
+
+The client will track schema indexes and use them when appropriate in `Get`, `Where`, and `WhereAll` as explained above.
+
+Additional indexes can be specified for a client instance to track. Just as schema indexes, client indexes are specified in sets per table.
+where each set consists of the columns that compose the index. Unlike schema indexes, a key within a column can be addressed if the column
+type is a map.
+
+Client indexes are leveraged through `Where`, and `WhereAll`. Since client indexes value uniqueness is not enforced as it happens with schema indexes,
+conditions based on them can match multiple rows.
+
+Indexed based operations generally provide better performance than operations based on explicit conditions.
+
+As an example, where you would have:
+
+    // slow predicate run on all the LB table rows...
+    ovn.WhereCache(func (lb *LoadBalancer) bool {
+        return lb.ExternalIds["myIdKey"] == "myIdValue"
+    }).List(ctx, &results)
+
+can now be improved with:
+
+    dbModel, err := nbdb.FullDatabaseModel()
+    db.SetIndexes(map[string][]model.ClientIndex{
+        "Load_Balancer": {{Columns: []model.ColumnKey{Column: "external_ids", Key: "myIdKey"}}}
+    })
+
+    // connect ....
+
+    lb := &LoadBalancer{
+        ExternalIds: map[string]string{"myIdKey": "myIdValue"},
+    }
+    // quick indexed result
+    ovn.Where(lb).List(ctx, &results)
+
 ## Documentation
 
 This package is divided into several sub-packages. Documentation for each sub-package is available at [pkg.go.dev][doc]:
diff --git a/cache/cache.go b/cache/cache.go
index 762eefc..5653145 100644
--- a/cache/cache.go
+++ b/cache/cache.go
@@ -9,6 +9,7 @@ import (
 	"log"
 	"os"
 	"reflect"
+	"sort"
 	"strings"
 	"sync"
 
@@ -25,6 +26,7 @@ const (
 	deleteEvent     = "delete"
 	bufferSize      = 65536
 	columnDelimiter = ","
+	keyDelimiter    = "|"
 )
 
 // ErrCacheInconsistent is an error that can occur when an operation
@@ -68,32 +70,84 @@ func NewIndexExistsError(table string, value interface{}, index string, new, exi
 }
 
 // map of unique values to uuids
-type valueToUUID map[interface{}]string
+type valueToUUIDs map[interface{}]uuidset
 
-// map of column name(s) to a unique values, to UUIDs
-type columnToValue map[index]valueToUUID
+// map of column name(s) to unique values, to UUIDs
+type columnToValue map[index]valueToUUIDs
 
 // index is the type used to implement multiple cache indexes
 type index string
 
-// columns returns the columns that conform the index
-func (i index) columns() []string {
-	return strings.Split(string(i), columnDelimiter)
+// indexType is the type of index
+type indexType uint
+
+const (
+	schemaIndexType indexType = iota
+	clientIndexType
+)
+
+// indexSpec contains details about an index
+type indexSpec struct {
+	index     index
+	columns   []model.ColumnKey
+	indexType indexType
+}
+
+func (s indexSpec) isClientIndex() bool {
+	return s.indexType == clientIndexType
+}
+
+func (s indexSpec) isSchemaIndex() bool {
+	return s.indexType == schemaIndexType
 }
 
 // newIndex builds a index from a list of columns
-func newIndex(columns ...string) index {
+func newIndexFromColumns(columns ...string) index {
+	sort.Strings(columns)
 	return index(strings.Join(columns, columnDelimiter))
 }
 
+// newIndexFromColumnKeys builds a index from a list of column keys
+func newIndexFromColumnKeys(columnsKeys ...model.ColumnKey) index {
+	// RFC 7047 says that Indexes is a [<column-set>] and "Each <column-set> is a set of
+	// columns whose values, taken together within any given row, must be
+	// unique within the table". We'll store the column names, separated by comma
+	// as we'll assume (RFC is not clear), that comma isn't valid in a <id>
+	columns := make([]string, 0, len(columnsKeys))
+	columnsMap := map[string]struct{}{}
+	for _, columnKey := range columnsKeys {
+		var column string
+		if columnKey.Key != nil {
+			column = fmt.Sprintf("%s%s%v", columnKey.Column, keyDelimiter, columnKey.Key)
+		} else {
+			column = columnKey.Column
+		}
+		if _, found := columnsMap[column]; !found {
+			columns = append(columns, column)
+			columnsMap[column] = struct{}{}
+		}
+	}
+	return newIndexFromColumns(columns...)
+}
+
+// newColumnKeysFromColumns builds a list of column keys from a list of columns
+func newColumnKeysFromColumns(columns ...string) []model.ColumnKey {
+	columnKeys := make([]model.ColumnKey, len(columns))
+	for i, column := range columns {
+		columnKeys[i] = model.ColumnKey{Column: column}
+	}
+	return columnKeys
+}
+
 // RowCache is a collections of Models hashed by UUID
 type RowCache struct {
-	name     string
-	dbModel  model.DatabaseModel
-	dataType reflect.Type
-	cache    map[string]model.Model
-	indexes  columnToValue
-	mutex    sync.RWMutex
+	name       string
+	dbModel    model.DatabaseModel
+	dataType   reflect.Type
+	cache      map[string]model.Model
+	indexSpecs []indexSpec
+	indexes    columnToValue
+	mutex      sync.RWMutex
 }
 
 // rowByUUID returns one model from the cache by UUID. Caller must hold the row
@@ -112,33 +166,73 @@ func (r *RowCache) Row(uuid string) model.Model {
 	return r.rowByUUID(uuid)
 }
 
-// RowByModel searches the cache using a the indexes for a provided model
-func (r *RowCache) RowByModel(m model.Model) model.Model {
+func (r *RowCache) rowsByModel(m model.Model, useClientIndexes bool) map[string]model.Model {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
 	if reflect.TypeOf(m) != r.dataType {
 		return nil
 	}
 	info, _ := r.dbModel.NewModelInfo(m)
-	uuid, err := info.FieldByColumn("_uuid")
+	field, err := info.FieldByColumn("_uuid")
 	if err != nil {
 		return nil
 	}
-	if uuid.(string) != "" {
-		return r.rowByUUID(uuid.(string))
+	uuid := field.(string)
+	if uuid != "" {
+		row := r.rowByUUID(uuid)
+		if row != nil {
+			return map[string]model.Model{uuid: row}
+		}
 	}
-	for index, vals := range r.indexes {
-		val, err := valueFromIndex(info, index)
+	// indexSpecs are ordered, schema indexes go first, then client indexes
+	for _, indexSpec := range r.indexSpecs {
+		if indexSpec.isClientIndex() && !useClientIndexes {
+			// Given the ordered indexSpecs, we can break here if we reach the
+			// first client index
+			break
+		}
+		val, err := valueFromIndex(info, indexSpec.columns)
 		if err != nil {
 			continue
 		}
-		if uuid, ok := vals[val]; ok {
-			return r.rowByUUID(uuid)
+		vals := r.indexes[indexSpec.index]
+		uuids, ok := vals[val]
+		if !ok || uuids.empty() {
+			continue
+		}
+		results := make(map[string]model.Model, len(uuids))
+		for uuid := range uuids {
+			results[uuid] = r.rowByUUID(uuid)
 		}
+		return results
 	}
 	return nil
 }
 
+// RowByModel searches the cache by UUID and schema indexes. UUID search is
+// performed first. Then schema indexes are evaluated in turn by the same order
+// with which they are defined in the schema. The model for the first matching
+// index is returned along with its UUID. An empty string and nil is returned if
+// no Model is found.
+func (r *RowCache) RowByModel(m model.Model) (string, model.Model) {
+	models := r.rowsByModel(m, false)
+	for uuid, model := range models {
+		return uuid, model
+	}
+	return "", nil
+}
+
+// RowsByModel searches the cache by UUID, schema indexes and client indexes.
+// UUID search is performed first. Schema indexes are evaluated next in turn by
+// the same order with which they are defined in the schema. Finally, client
+// indexes are evaluated in turn by the same order with which they are defined
+// in the client DB model. The models for the first matching index are returned,
+// which might be more than 1 if they were found through a client index since in
+// that case uniqueness is not enforced. Nil is returned if no Model is found.
+func (r *RowCache) RowsByModel(m model.Model) map[string]model.Model {
+	return r.rowsByModel(m, true)
+}
+
 // Create writes the provided content to the cache
 func (r *RowCache) Create(uuid string, m model.Model, checkIndexes bool) error {
 	r.mutex.Lock()
@@ -153,25 +247,31 @@ func (r *RowCache) Create(uuid string, m model.Model, checkIndexes bool) error {
 	if err != nil {
 		return err
 	}
-	newIndexes := newColumnToValue(r.dbModel.Schema.Table(r.name).Indexes)
-	for index, vals := range r.indexes {
-		val, err := valueFromIndex(info, index)
+	newIndexes := r.newIndexes()
+	for _, indexSpec := range r.indexSpecs {
+		index := indexSpec.index
+		val, err := valueFromIndex(info, indexSpec.columns)
 		if err != nil {
 			return err
 		}
 
-		if existing, ok := vals[val]; ok && checkIndexes {
-			return NewIndexExistsError(r.name, val, string(index), uuid, existing)
+		vals := r.indexes[index]
+		if existing, ok := vals[val]; ok && !existing.empty() && checkIndexes && indexSpec.isSchemaIndex() {
+			return NewIndexExistsError(r.name, val, string(index), uuid, existing.getAny())
 		}
 
-		newIndexes[index][val] = uuid
+		uuidset := newUUIDSet(uuid)
+		if indexSpec.isSchemaIndex() {
+			newIndexes[index][val] = uuidset
+		} else {
+			newIndexes[index][val] = addUUIDSet(r.indexes[index][val], uuidset)
+		}
 	}
 
 	// write indexes
 	for k1, v1 := range newIndexes {
-		vals := r.indexes[k1]
 		for k2, v2 := range v1 {
-			vals[k2] = v2
+			r.indexes[k1][k2] = v2
 		}
 	}
 	r.cache[uuid] = model.Clone(m)
@@ -194,17 +294,16 @@ func (r *RowCache) Update(uuid string, m model.Model, checkIndexes bool) error {
 	if err != nil {
 		return err
 	}
-	indexes := r.dbModel.Schema.Table(r.name).Indexes
-	newIndexes := newColumnToValue(indexes)
-	oldIndexes := newColumnToValue(indexes)
+	newIndexes := r.newIndexes()
 	var errs []error
-	for index, vals := range r.indexes {
+	for _, indexSpec := range r.indexSpecs {
+		index := indexSpec.index
 		var err error
-		oldVal, err := valueFromIndex(oldInfo, index)
+		oldVal, err := valueFromIndex(oldInfo, indexSpec.columns)
 		if err != nil {
 			return err
 		}
-		newVal, err := valueFromIndex(newInfo, index)
+		newVal, err := valueFromIndex(newInfo, indexSpec.columns)
 		if err != nil {
 			return err
 		}
@@ -216,34 +315,37 @@ func (r *RowCache) Update(uuid string, m model.Model, checkIndexes bool) error {
 		// old and new values are NOT the same
 
 		// check that there are no conflicts
-		if conflict, ok := vals[newVal]; ok && checkIndexes && conflict != uuid {
+		vals := r.indexes[index]
+		if existing, ok := vals[newVal]; ok && indexSpec.isSchemaIndex() && checkIndexes && !existing.empty() && !existing.has(uuid) {
 			errs = append(errs, NewIndexExistsError(
 				r.name,
 				newVal,
 				string(index),
 				uuid,
-				conflict,
+				existing.getAny(),
 			))
 		}
 
-		newIndexes[index][newVal] = uuid
-		oldIndexes[index][oldVal] = ""
+		uuidset := newUUIDSet(uuid)
+		if indexSpec.isSchemaIndex() {
+			newIndexes[index][newVal] = uuidset
+			newIndexes[index][oldVal] = nil
+		} else {
+			newIndexes[index][newVal] = addUUIDSet(r.indexes[index][newVal], uuidset)
+			newIndexes[index][oldVal] = substractUUIDSet(r.indexes[index][oldVal], uuidset)
+		}
 	}
 	if len(errs) > 0 {
 		return fmt.Errorf("%+v", errs)
 	}
 	// write indexes
 	for k1, v1 := range newIndexes {
-		vals := r.indexes[k1]
 		for k2, v2 := range v1 {
-			vals[k2] = v2
-		}
-	}
-	// delete old indexes
-	for k1, v1 := range oldIndexes {
-		vals := r.indexes[k1]
-		for k2 := range v1 {
-			delete(vals, k2)
+			if len(v2) == 0 {
+				delete(r.indexes[k1], k2)
+			} else {
+				r.indexes[k1][k2] = v2
+			}
 		}
 	}
 	r.cache[uuid] = model.Clone(m)
@@ -255,22 +357,30 @@ func (r *RowCache) IndexExists(row model.Model) error {
 	if err != nil {
 		return err
 	}
-	uuid, err := info.FieldByColumn("_uuid")
+	field, err := info.FieldByColumn("_uuid")
 	if err != nil {
 		return nil
 	}
-	for index, vals := range r.indexes {
-		val, err := valueFromIndex(info, index)
+	uuid := field.(string)
+	for _, indexSpec := range r.indexSpecs {
+		if !indexSpec.isSchemaIndex() {
+			// Given the ordered indexSpecs, we can break here if we reach the
+			// first non schema index
+			break
+		}
+		index := indexSpec.index
+		val, err := valueFromIndex(info, indexSpec.columns)
 		if err != nil {
 			continue
 		}
-		if existing, ok := vals[val]; ok && existing != uuid.(string) {
+		vals := r.indexes[index]
+		if existing := vals[val]; !existing.empty() && !existing.has(uuid) {
 			return NewIndexExistsError(
 				r.name,
 				val,
 				string(index),
-				uuid.(string),
-				existing,
+				uuid,
+				existing.getAny(),
 			)
 		}
 	}
@@ -289,16 +399,22 @@ func (r *RowCache) Delete(uuid string) error {
 	if err != nil {
 		return err
 	}
-	for index, vals := range r.indexes {
-		oldVal, err := valueFromIndex(oldInfo, index)
+	for _, indexSpec := range r.indexSpecs {
+		index := indexSpec.index
+		oldVal, err := valueFromIndex(oldInfo, indexSpec.columns)
 		if err != nil {
 			return err
 		}
 		// only remove the index if it is pointing to this uuid
 		// otherwise we can cause a consistency issue if we've processed
 		// updates out of order
-		if vals[oldVal] == uuid {
-			delete(vals, oldVal)
+		vals := r.indexes[index]
+		existing, ok := vals[oldVal]
+		if ok {
+			existing.remove(uuid)
+			if len(existing) == 0 {
+				delete(vals, oldVal)
+			}
 		}
 	}
 	delete(r.cache, uuid)
@@ -331,73 +447,277 @@ func (r *RowCache) RowsShallow() map[string]model.Model {
 	return result
 }
 
+// uuidsByConditionsAsIndexes checks possible indexes that can be built with a
+// subset of the provided conditions and returns the uuids for the models that
+// match that subset of conditions. If no conditions could be used as indexes,
+// returns nil. Note that this method does not necessarily match all the
+// provided conditions. Thus the caller is required to evaluate all the
+// conditions against the returned candidates. This is only useful to obtain, as
+// quick as possible, via indexes, a reduced list of candidate models that might
+// match all conditions, which should be better than just evaluating all
+// conditions against all rows of a table.
+//nolint:gocyclo // warns overall function is complex but ignores inner functions
+func (r *RowCache) uuidsByConditionsAsIndexes(conditions []ovsdb.Condition, nativeValues []interface{}) (uuidset, error) {
+	type indexableCondition struct {
+		column      string
+		keys        []interface{}
+		nativeValue interface{}
+	}
+
+	// build an indexable condition, more appropriate for our processing, from
+	// an ovsdb condition. Only equality based conditions can be used as indexes
+	// (or `includes` conditions on map values).
+	toIndexableCondition := func(condition ovsdb.Condition, nativeValue interface{}) *indexableCondition {
+		if condition.Column == "_uuid" {
+			return nil
+		}
+		if condition.Function != ovsdb.ConditionEqual && condition.Function != ovsdb.ConditionIncludes {
+			return nil
+		}
+		v := reflect.ValueOf(nativeValue)
+		if !v.IsValid() {
+			return nil
+		}
+		isSet := v.Kind() == reflect.Slice || v.Kind() == reflect.Array
+		if condition.Function == ovsdb.ConditionIncludes && isSet {
+			return nil
+		}
+		keys := []interface{}{}
+		if v.Kind() == reflect.Map && condition.Function == ovsdb.ConditionIncludes {
+			for _, key := range v.MapKeys() {
+				keys = append(keys, key.Interface())
+			}
+		}
+		return &indexableCondition{
+			column:      condition.Column,
+			keys:        keys,
+			nativeValue: nativeValue,
+		}
+	}
+
+	// for any given set of conditions, we need to check if an index uses the
+	// same fields as the conditions
+	indexMatchesConditions := func(spec indexSpec, conditions []*indexableCondition) bool {
+		columnKeys := []model.ColumnKey{}
+		for _, condition := range conditions {
+			if len(condition.keys) == 0 {
+				columnKeys = append(columnKeys, model.ColumnKey{Column: condition.column})
+				continue
+			}
+			for _, key := range condition.keys {
+				columnKeys = append(columnKeys, model.ColumnKey{Column: condition.column, Key: key})
+			}
+		}
+		index := newIndexFromColumnKeys(columnKeys...)
+		return index == spec.index
+	}
+
+	// for a specific set of conditions, check if an index can be built from
+	// them and return the associated UUIDs
+	evaluateConditionSetAsIndex := func(conditions []*indexableCondition) (uuidset, error) {
+		// build a model with the values from the conditions
+		m, err := r.dbModel.NewModel(r.name)
+		if err != nil {
+			return nil, err
+		}
+		info, err := r.dbModel.NewModelInfo(m)
+		if err != nil {
+			return nil, err
+		}
+		for _, conditions := range conditions {
+			err := info.SetField(conditions.column, conditions.nativeValue)
+			if err != nil {
+				return nil, err
+			}
+		}
+		for _, spec := range r.indexSpecs {
+			if !indexMatchesConditions(spec, conditions) {
+				continue
+			}
+			// if we have an index for those conditions, calculate the index
+			// value. The models mapped to that value match the conditions.
+			v, err := valueFromIndex(info, spec.columns)
+			if err != nil {
+				return nil, err
+			}
+			if v != nil {
+				uuids := r.indexes[spec.index][v]
+				if uuids == nil {
+					// this set of conditions was represented by an index but
+					// had no matches, return an empty set
+					uuids = uuidset{}
+				}
+				return uuids, nil
+			}
+		}
+		return nil, nil
+	}
+
+	// set of uuids that match the conditions as we evaluate them
+	var matching uuidset
+
+	// attempt to evaluate a set of conditions via indexes and intersect the
+	// results against matches of previous sets
+	intersectUUIDsFromConditionSet := func(indexableConditions []*indexableCondition) (bool, error) {
+		uuids, err := evaluateConditionSetAsIndex(indexableConditions)
+		if err != nil {
+			return true, err
+		}
+		if matching == nil {
+			matching = uuids
+		} else if uuids != nil {
+			matching = intersectUUIDSets(matching, uuids)
+		}
+		if matching != nil && len(matching) <= 1 {
+			// if we had no matches or a single match, no point in continuing
+			// searching for additional indexes. If we had a single match, it's
+			// cheaper to just evaluate all conditions on it.
+			return true, nil
+		}
+		return false, nil
+	}
+
+	// First, filter out conditions that cannot be matched against indexes. With
+	// the remaining conditions build all possible subsets (the power set of all
+	// conditions) and for any subset that is an index, intersect the obtained
+	// uuids with the ones obtained from previous subsets
+	matchUUIDsFromConditionsPowerSet := func() error {
+		ps := [][]*indexableCondition{}
+		// prime the power set with a first empty subset
+		ps = append(ps, []*indexableCondition{})
+		for i, condition := range conditions {
+			nativeValue := nativeValues[i]
+			iCondition := toIndexableCondition(condition, nativeValue)
+			// this is not a condition we can use as an index, skip it
+			if iCondition == nil {
+				continue
+			}
+			// the power set is built appending the subsets that result from
+			// adding each item to each of the previous subsets
+			ss := make([][]*indexableCondition, len(ps))
+			for j := range ss {
+				ss[j] = make([]*indexableCondition, len(ps[j]), len(ps[j])+1)
+				copy(ss[j], ps[j])
+				ss[j] = append(ss[j], iCondition)
+				// as we add them to the power set, attempt to evaluate this
+				// subset of conditions as indexes
+				stop, err := intersectUUIDsFromConditionSet(ss[j])
+				if stop || err != nil {
+					return err
+				}
+			}
+			ps = append(ps, ss...)
+		}
+		return nil
+	}
+
+	// finally
+	err := matchUUIDsFromConditionsPowerSet()
+	return matching, err
+}
+
+// RowsByCondition searches models in the cache that match all conditions
 func (r *RowCache) RowsByCondition(conditions []ovsdb.Condition) (map[string]model.Model, error) {
+	r.mutex.RLock()
+	defer r.mutex.RUnlock()
 	results := make(map[string]model.Model)
 	schema := r.dbModel.Schema.Table(r.name)
+
+	// no conditions matches all rows
 	if len(conditions) == 0 {
-		for uuid, row := range r.Rows() {
-			results[uuid] = row
+		for uuid := range r.cache {
+			results[uuid] = r.rowByUUID(uuid)
 		}
 		return results, nil
 	}
 
+	// one pass to obtain the native values
+	nativeValues := make([]interface{}, 0, len(conditions))
 	for _, condition := range conditions {
-		if condition.Column == "_uuid" {
-			ovsdbUUID, ok := condition.Value.(ovsdb.UUID)
+		tSchema := schema.Column(condition.Column)
+		nativeValue, err := ovsdb.OvsToNative(tSchema, condition.Value)
+		if err != nil {
+			return nil, err
+		}
+		nativeValues = append(nativeValues, nativeValue)
+	}
+
+	// obtain all possible matches using conditions as indexes
+	matching, err := r.uuidsByConditionsAsIndexes(conditions, nativeValues)
+	if err != nil {
+		return nil, err
+	}
+
+	// From the matches obtained with indexes, which might have not used all
+	// conditions, continue trimming down the list explicitly evaluating the
+	// conditions.
+	for i, condition := range conditions {
+		matchingCondition := uuidset{}
+
+		if condition.Column == "_uuid" && (condition.Function == ovsdb.ConditionEqual || condition.Function == ovsdb.ConditionIncludes) {
+			uuid, ok := nativeValues[i].(string)
 			if !ok {
-				panic(fmt.Sprintf("%+v is not an ovsdb uuid", ovsdbUUID))
-			}
-			uuid := ovsdbUUID.GoUUID
-			for rowUUID, row := range r.Rows() {
-				ok, err := condition.Function.Evaluate(rowUUID, uuid)
-				if err != nil {
-					return nil, err
-				}
-				if ok {
-					results[rowUUID] = row
-				}
+				panic(fmt.Sprintf("%+v is not a uuid", nativeValues[i]))
 			}
-		} else if index, err := r.Index(condition.Column); err == nil {
-			for k, rowUUID := range index {
-				tSchema := schema.Columns[condition.Column]
-				nativeValue, err := ovsdb.OvsToNative(tSchema, condition.Value)
-				if err != nil {
-					return nil, err
-				}
-				ok, err := condition.Function.Evaluate(k, nativeValue)
-				if err != nil {
-					return nil, err
-				}
-				if ok {
-					row := r.Row(rowUUID)
-					results[rowUUID] = row
-				}
+			if _, found := r.cache[uuid]; found {
+				matchingCondition.add(uuid)
 			}
 		} else {
-			for uuid, row := range r.Rows() {
+			matchCondition := func(uuid string) error {
+				row := r.cache[uuid]
 				info, err := r.dbModel.NewModelInfo(row)
 				if err != nil {
-					return nil, err
+					return err
 				}
 				value, err := info.FieldByColumn(condition.Column)
 				if err != nil {
-					return nil, err
-				}
-				tSchema := schema.Columns[condition.Column]
-				nativeValue, err := ovsdb.OvsToNative(tSchema, condition.Value)
-				if err != nil {
-					return nil, err
+					return err
 				}
-				ok, err := condition.Function.Evaluate(value, nativeValue)
+				ok, err := condition.Function.Evaluate(value, nativeValues[i])
 				if err != nil {
-					return nil, err
+					return err
 				}
 				if ok {
-					results[uuid] = row
+					matchingCondition.add(uuid)
+				}
+				return nil
+			}
+			if matching != nil {
+				// we just need to consider rows that matched previous
+				// conditions
+				for uuid := range matching {
+					err = matchCondition(uuid)
+					if err != nil {
+						return nil, err
+					}
+				}
+			} else {
+				// If this is the first condition we are able to check, just run
+				// it by whole table
+				for uuid := range r.cache {
+					err = matchCondition(uuid)
+					if err != nil {
+						return nil, err
+					}
 				}
 			}
 		}
+		if matching == nil {
+			matching = matchingCondition
+		} else {
+			matching = intersectUUIDSets(matching, matchingCondition)
+		}
+		if matching.empty() {
+			// no models match the conditions checked up to now, no need to
+			// check remaining conditions
+			break
+		}
+	}
+
+	for uuid := range matching {
+		results[uuid] = r.rowByUUID(uuid)
 	}
+
 	return results, nil
 }
 
@@ -408,14 +728,19 @@ func (r *RowCache) Len() int {
 	return len(r.cache)
 }
 
-func (r *RowCache) Index(columns ...string) (map[interface{}]string, error) {
+func (r *RowCache) Index(columns ...string) (map[interface{}][]string, error) {
 	r.mutex.RLock()
 	defer r.mutex.RUnlock()
-	index, ok := r.indexes[newIndex(columns...)]
+	spec := newIndexFromColumns(columns...)
+	index, ok := r.indexes[spec]
 	if !ok {
-		return nil, fmt.Errorf("%s is not an index", index)
+		return nil, fmt.Errorf("%v is not an index", columns)
+	}
+	dbIndex := make(map[interface{}][]string, len(index))
+	for k, v := range index {
+		dbIndex[k] = v.list()
 	}
-	return index, nil
+	return dbIndex, nil
 }
 
 // EventHandler can handle events when the contents of the cache changes
@@ -740,29 +1065,49 @@ func (t *TableCache) Run(stopCh <-chan struct{}) {
 // newRowCache creates a new row cache with the provided data
 // if the data is nil, and empty RowCache will be created
 func newRowCache(name string, dbModel model.DatabaseModel, dataType reflect.Type) *RowCache {
+	schemaIndexes := dbModel.Schema.Table(name).Indexes
+	clientIndexes := dbModel.Client().Indexes(name)
+
 	r := &RowCache{
-		name:     name,
-		dbModel:  dbModel,
-		indexes:  newColumnToValue(dbModel.Schema.Table(name).Indexes),
-		dataType: dataType,
-		cache:    make(map[string]model.Model),
-		mutex:    sync.RWMutex{},
+		name:       name,
+		dbModel:    dbModel,
+		indexSpecs: make([]indexSpec, 0, len(schemaIndexes)+len(clientIndexes)),
+		dataType:   dataType,
+		cache:      make(map[string]model.Model),
+		mutex:      sync.RWMutex{},
+	}
+
+	// respect the order of indexes, add first schema indexes, then client
+	// indexes
+	indexes := map[index]indexSpec{}
+	for _, columns := range schemaIndexes {
+		columnKeys := newColumnKeysFromColumns(columns...)
+		index := newIndexFromColumnKeys(columnKeys...)
+		spec := indexSpec{index: index, columns: columnKeys, indexType: schemaIndexType}
+		r.indexSpecs = append(r.indexSpecs, spec)
+		indexes[index] = spec
+	}
+	for _, clientIndex := range clientIndexes {
+		columnKeys := clientIndex.Columns
+		index := newIndexFromColumnKeys(columnKeys...)
+		// if this is already a DB index, ignore
+		if _, ok := indexes[index]; ok {
+			continue
+		}
+		spec := indexSpec{index: index, columns: columnKeys, indexType: clientIndexType}
+		r.indexSpecs = append(r.indexSpecs, spec)
+		indexes[index] = spec
 	}
+
+	r.indexes = r.newIndexes()
 	return r
 }
 
-func newColumnToValue(schemaIndexes [][]string) columnToValue {
-	// RFC 7047 says that Indexes is a [<column-set>] and "Each <column-set> is a set of
-	// columns whose values, taken together within any given row, must be
-	// unique within the table". We'll store the column names, separated by comma
-	// as we'll assume (RFC is not clear), that comma isn't valid in a <id>
-	var indexes []index
-	for i := range schemaIndexes {
-		indexes = append(indexes, newIndex(schemaIndexes[i]...))
-	}
+func (r *RowCache) newIndexes() columnToValue {
 	c := make(columnToValue)
-	for _, index := range indexes {
-		c[index] = make(valueToUUID)
+	for _, indexSpec := range r.indexSpecs {
+		index := indexSpec.index
+		c[index] = make(valueToUUIDs)
 	}
 	return c
 }
@@ -1032,13 +1377,12 @@ func (t *TableCache) ApplyModifications(tableName string, base model.Model, upda
 	return nil
 }
 
-func valueFromIndex(info *mapper.Info, index index) (interface{}, error) {
-	columns := index.columns()
-	if len(columns) > 1 {
+func valueFromIndex(info *mapper.Info, columnKeys []model.ColumnKey) (interface{}, error) {
+	if len(columnKeys) > 1 {
 		var buf bytes.Buffer
 		enc := gob.NewEncoder(&buf)
-		for _, column := range columns {
-			val, err := info.FieldByColumn(column)
+		for _, columnKey := range columnKeys {
+			val, err := valueFromColumnKey(info, columnKey)
 			if err != nil {
 				return "", err
 			}
@@ -1051,9 +1395,37 @@ func valueFromIndex(info *mapper.Info, index index) (interface{}, error) {
 		val := hex.EncodeToString(h.Sum(buf.Bytes()))
 		return val, nil
 	}
-	val, err := info.FieldByColumn(columns[0])
+	val, err := valueFromColumnKey(info, columnKeys[0])
+	if err != nil {
+		return "", err
+	}
+	return val, err
+}
+
+func valueFromColumnKey(info *mapper.Info, columnKey model.ColumnKey) (interface{}, error) {
+	val, err := info.FieldByColumn(columnKey.Column)
 	if err != nil {
 		return nil, err
 	}
+	if columnKey.Key != nil {
+		val, err = valueFromMap(val, columnKey.Key)
+		if err != nil {
+			return "", fmt.Errorf("can't get key value from map: %v", err)
+		}
+	}
 	return val, err
 }
+
+func valueFromMap(aMap interface{}, key interface{}) (interface{}, error) {
+	m := reflect.ValueOf(aMap)
+	if m.Kind() != reflect.Map {
+		return nil, fmt.Errorf("expected map but got %s", m.Kind())
+	}
+	v := m.MapIndex(reflect.ValueOf(key))
+	if !v.IsValid() {
+		// return the zero value for the map value type
+		return reflect.Indirect(reflect.New(m.Type().Elem())).Interface(), nil
+	}
+
+	return v.Interface(), nil
+}
diff --git a/cache/uuidset.go b/cache/uuidset.go
new file mode 100644
index 0000000..fe3c5d8
--- /dev/null
+++ b/cache/uuidset.go
@@ -0,0 +1,89 @@
+package cache
+
+type void struct{}
+type uuidset map[string]void
+
+func newUUIDSet(uuids ...string) uuidset {
+	s := uuidset{}
+	for _, uuid := range uuids {
+		s[uuid] = void{}
+	}
+	return s
+}
+
+func (s uuidset) add(uuid string) {
+	s[uuid] = void{}
+}
+
+func (s uuidset) remove(uuid string) {
+	delete(s, uuid)
+}
+
+func (s uuidset) has(uuid string) bool {
+	_, ok := s[uuid]
+	return ok
+}
+
+func (s uuidset) getAny() string {
+	for k := range s {
+		return k
+	}
+	return ""
+}
+
+func (s uuidset) list() []string {
+	uuids := make([]string, 0, len(s))
+	for uuid := range s {
+		uuids = append(uuids, uuid)
+	}
+	return uuids
+}
+
+func (s uuidset) empty() bool {
+	return len(s) == 0
+}
+
+func addUUIDSet(s1, s2 uuidset) uuidset {
+	if len(s2) == 0 {
+		return s1
+	}
+	if s1 == nil {
+		s1 = uuidset{}
+	}
+	for uuid := range s2 {
+		s1.add(uuid)
+	}
+	return s1
+}
+
+func substractUUIDSet(s1, s2 uuidset) uuidset {
+	if len(s1) == 0 || len(s2) == 0 {
+		return s1
+	}
+	for uuid := range s2 {
+		s1.remove(uuid)
+	}
+	return s1
+}
+
+func intersectUUIDSets(s1, s2 uuidset) uuidset {
+	if len(s1) == 0 || len(s2) == 0 {
+		return nil
+	}
+	var big uuidset
+	var small uuidset
+	if len(s1) > len(s2) {
+		big = s1
+		small = s2
+	} else {
+		big = s2
+		small = s1
+	}
+	f := uuidset{}
+	for uuid := range small {
+		if big.has(uuid) {
+			f.add(uuid)
+		}
+	}
+	return f
+}
diff --git a/client/api.go b/client/api.go
index 0ca3f9b..4d93eff 100644
--- a/client/api.go
+++ b/client/api.go
@@ -29,7 +29,7 @@ type API interface {
 	// Create a ConditionalAPI from a Model's index data or a list of Conditions
 	// where operations apply to elements that match any of the conditions
 	// If no condition is given, it will match the values provided in model.Model according
-	// to the database index.
+	// to the indexes.
 	Where(model.Model, ...model.Condition) ConditionalAPI
 
 	// Create a ConditionalAPI from a Model's index data or a list of Conditions
@@ -147,35 +147,35 @@ func (a api) List(ctx context.Context, result interface{}) error {
 	}
 	i := resultVal.Len()
 
-	for _, row := range tableCache.RowsShallow() {
-		if i >= resultVal.Cap() {
-			break
+	var rows map[string]model.Model
+	if a.cond != nil {
+		rows, err = a.cond.Matches()
+		if err != nil {
+			return err
 		}
+	} else {
+		rows = tableCache.Rows()
+	}
 
-		if a.cond != nil {
-			if matches, err := a.cond.Matches(row); err != nil {
-				return err
-			} else if !matches {
-				continue
-			}
+	for _, row := range rows {
+		if i >= resultVal.Cap() {
+			break
 		}
-		// clone only the models that match the predicate
-		m := model.Clone(row)
-
-		appendValue(reflect.ValueOf(m))
+		appendValue(reflect.ValueOf(row))
 		i++
 	}
+
 	return nil
 }
 
 // Where returns a conditionalAPI based on a Condition list
 func (a api) Where(model model.Model, cond ...model.Condition) ConditionalAPI {
-	return newConditionalAPI(a.cache, a.conditionFromModel(false, model, cond...), a.logger)
+	return newConditionalAPI(a.cache, a.conditionFromModel(false, false, model, cond...), a.logger)
 }
 
 // Where returns a conditionalAPI based on a Condition list
 func (a api) WhereAll(model model.Model, cond ...model.Condition) ConditionalAPI {
-	return newConditionalAPI(a.cache, a.conditionFromModel(true, model, cond...), a.logger)
+	return newConditionalAPI(a.cache, a.conditionFromModel(true, false, model, cond...), a.logger)
 }
 
 // Where returns a conditionalAPI based a Predicate
@@ -199,7 +199,7 @@ func (a api) conditionFromFunc(predicate interface{}) Conditional {
 }
 
 // FromModel returns a Condition from a model and a list of fields
-func (a api) conditionFromModel(any bool, model model.Model, cond ...model.Condition) Conditional {
+func (a api) conditionFromModel(any bool, cache bool, model model.Model, cond ...model.Condition) Conditional {
 	var conditional Conditional
 	var err error
 
@@ -209,13 +209,12 @@ func (a api) conditionFromModel(any bool, model model.Model, cond ...model.Condi
 	}
 
 	if len(cond) == 0 {
-		conditional, err = newEqualityConditional(a.cache.DatabaseModel(), tableName, any, model)
+		conditional, err = newEqualityConditional(tableName, a.cache, model)
 		if err != nil {
 			conditional = newErrorConditional(err)
 		}
-
 	} else {
-		conditional, err = newExplicitConditional(a.cache.DatabaseModel(), tableName, any, model, cond...)
+		conditional, err = newExplicitConditional(tableName, a.cache, any, model, cond...)
 		if err != nil {
 			conditional = newErrorConditional(err)
 		}
@@ -240,7 +239,7 @@ func (a api) Get(ctx context.Context, m model.Model) error {
 		return ErrNotFound
 	}
 
-	found := tableCache.RowByModel(m)
+	_, found := tableCache.RowByModel(m)
 	if found == nil {
 		return ErrNotFound
 	}
diff --git a/client/condition.go b/client/condition.go
index f67e6e2..759469a 100644
--- a/client/condition.go
+++ b/client/condition.go
@@ -16,118 +16,147 @@ type Conditional interface {
 	// Generate returns a list of lists of conditions to be used in Operations
 	// Each element in the (outer) list corresponds to an operation
 	Generate() ([][]ovsdb.Condition, error)
-	// matches returns true if a model matches the condition
-	Matches(m model.Model) (bool, error)
+	// Returns the models that match the conditions
+	Matches() (map[string]model.Model, error)
 	// returns the table that this condition is associated with
 	Table() string
 }
 
-// equalityConditional uses the information available in a model to generate conditions
-// The conditions are based on the equality of the first available index.
-// The priority of indexes is: uuid, {schema index}
-type equalityConditional struct {
-	dbModel   model.DatabaseModel
-	tableName string
-	info      *mapper.Info
-	singleOp  bool
+func generateConditionsFromModels(dbModel model.DatabaseModel, models map[string]model.Model) ([][]ovsdb.Condition, error) {
+	anyConditions := make([][]ovsdb.Condition, 0, len(models))
+	for _, model := range models {
+		info, err := dbModel.NewModelInfo(model)
+		if err != nil {
+			return nil, err
+		}
+		allConditions, err := dbModel.Mapper.NewEqualityCondition(info)
+		if err != nil {
+			return nil, err
+		}
+		anyConditions = append(anyConditions, allConditions)
+	}
+	return anyConditions, nil
 }
 
-func (c *equalityConditional) Matches(m model.Model) (bool, error) {
-	info, err := c.dbModel.NewModelInfo(m)
-	if err != nil {
-		return false, err
+func generateOvsdbConditionsFromModelConditions(dbModel model.DatabaseModel, info *mapper.Info, conditions []model.Condition, singleOp bool) ([][]ovsdb.Condition, error) {
+	anyConditions := [][]ovsdb.Condition{}
+	if singleOp {
+		anyConditions = append(anyConditions, []ovsdb.Condition{})
 	}
-	return c.dbModel.Mapper.EqualFields(c.info, info)
+	for _, condition := range conditions {
+		ovsdbCond, err := dbModel.Mapper.NewCondition(info, condition.Field, condition.Function, condition.Value)
+		if err != nil {
+			return nil, err
+		}
+		allConditions := []ovsdb.Condition{*ovsdbCond}
+		if singleOp {
+			anyConditions[0] = append(anyConditions[0], allConditions...)
+		} else {
+			anyConditions = append(anyConditions, allConditions)
+		}
+	}
+	return anyConditions, nil
+}
+
+// equalityConditional uses the indexes available in a provided model to find a
+// matching model in the database.
+type equalityConditional struct {
+	tableName string
+	model     model.Model
+	cache     *cache.TableCache
 }
 
 func (c *equalityConditional) Table() string {
 	return c.tableName
 }
 
-// Generate returns a condition based on the model and the field pointers
-func (c *equalityConditional) Generate() ([][]ovsdb.Condition, error) {
-	var result [][]ovsdb.Condition
-
-	conds, err := c.dbModel.Mapper.NewEqualityCondition(c.info)
-	if err != nil {
-		return nil, err
+// Returns the models that match the indexes available through the provided
+// model.
+func (c *equalityConditional) Matches() (map[string]model.Model, error) {
+	tableCache := c.cache.Table(c.tableName)
+	if tableCache == nil {
+		return nil, ErrNotFound
 	}
-	if c.singleOp {
-		result = append(result, conds)
-	} else {
-		for _, c := range conds {
-			result = append(result, []ovsdb.Condition{c})
-		}
+	return tableCache.RowsByModel(c.model), nil
+}
+
+// Generate conditions based on the equality of the first available index. If
+// the index can be matched against a model in the cache, the condition will be
+// based on the UUID of the found model. Otherwise, the conditions will be based
+// on the index.
+func (c *equalityConditional) Generate() ([][]ovsdb.Condition, error) {
+	models, _ := c.Matches()
+	if len(models) == 0 {
+		// no cache hits, generate condition from model we were given
+		return generateConditionsFromModels(c.cache.DatabaseModel(), map[string]model.Model{"": c.model})
 	}
-	return result, nil
+	return generateConditionsFromModels(c.cache.DatabaseModel(), models)
 }
 
 // NewEqualityCondition creates a new equalityConditional
-func newEqualityConditional(dbModel model.DatabaseModel, table string, all bool, model model.Model, fields ...interface{}) (Conditional, error) {
-	info, err := dbModel.NewModelInfo(model)
-	if err != nil {
-		return nil, err
-	}
+func newEqualityConditional(table string, cache *cache.TableCache, model model.Model) (Conditional, error) {
 	return &equalityConditional{
-		dbModel:   dbModel,
 		tableName: table,
-		info:      info,
-		singleOp:  all,
+		model:     model,
+		cache:     cache,
 	}, nil
 }
 
 // explicitConditional generates conditions based on the provided Condition list
 type explicitConditional struct {
-	dbModel    model.DatabaseModel
-	tableName  string
-	info       *mapper.Info
-	conditions []model.Condition
-	singleOp   bool
-}
-
-func (c *explicitConditional) Matches(m model.Model) (bool, error) {
-	return false, fmt.Errorf("cannot perform cache comparisons using explicit conditions")
+	tableName     string
+	anyConditions [][]ovsdb.Condition
+	cache         *cache.TableCache
 }
 
 func (c *explicitConditional) Table() string {
 	return c.tableName
 }
 
-// Generate returns a condition based on the model and the field pointers
-func (c *explicitConditional) Generate() ([][]ovsdb.Condition, error) {
-	var result [][]ovsdb.Condition
-	var conds []ovsdb.Condition
-
-	for _, cond := range c.conditions {
-		ovsdbCond, err := c.dbModel.Mapper.NewCondition(c.info, cond.Field, cond.Function, cond.Value)
+// Returns the models that match the conditions
+func (c *explicitConditional) Matches() (map[string]model.Model, error) {
+	tableCache := c.cache.Table(c.tableName)
+	if tableCache == nil {
+		return nil, ErrNotFound
+	}
+	found := map[string]model.Model{}
+	for _, allConditions := range c.anyConditions {
+		models, err := tableCache.RowsByCondition(allConditions)
 		if err != nil {
 			return nil, err
 		}
-		if c.singleOp {
-			conds = append(conds, *ovsdbCond)
-		} else {
-			result = append(result, []ovsdb.Condition{*ovsdbCond})
+		for uuid, model := range models {
+			found[uuid] = model
 		}
-
 	}
-	if c.singleOp {
-		result = append(result, conds)
+	return found, nil
+}
+
+// Generate returns conditions based on the provided Condition list
+func (c *explicitConditional) Generate() ([][]ovsdb.Condition, error) {
+	models, _ := c.Matches()
+	if len(models) == 0 {
+		// no cache hits, return conditions we were given
+		return c.anyConditions, nil
 	}
-	return result, nil
+	return generateConditionsFromModels(c.cache.DatabaseModel(), models)
 }
 
-// newIndexCondition creates a new equalityConditional
-func newExplicitConditional(dbModel model.DatabaseModel, table string, all bool, model model.Model, cond ...model.Condition) (Conditional, error) {
+// newExplicitConditional creates a new explicitConditional
+func newExplicitConditional(table string, cache *cache.TableCache, all bool, model model.Model, cond ...model.Condition) (Conditional, error) {
+	dbModel := cache.DatabaseModel()
 	info, err := dbModel.NewModelInfo(model)
 	if err != nil {
 		return nil, err
 	}
+	anyConditions, err := generateOvsdbConditionsFromModelConditions(dbModel, info, cond, all)
+	if err != nil {
+		return nil, err
+	}
 	return &explicitConditional{
-		dbModel:    dbModel,
-		tableName:  table,
-		info:       info,
-		conditions: cond,
-		singleOp:   all,
+		tableName:     table,
+		anyConditions: anyConditions,
+		cache:         cache,
 	}, nil
 }
 
@@ -141,9 +170,22 @@ type predicateConditional struct {
 
 // matches returns the result of the execution of the predicate
 // Type verifications are not performed
-func (c *predicateConditional) Matches(model model.Model) (bool, error) {
-	ret := reflect.ValueOf(c.predicate).Call([]reflect.Value{reflect.ValueOf(model)})
-	return ret[0].Bool(), nil
+// Returns the models that match the conditions
+func (c *predicateConditional) Matches() (map[string]model.Model, error) {
+	tableCache := c.cache.Table(c.tableName)
+	if tableCache == nil {
+		return nil, ErrNotFound
+	}
+	found := map[string]model.Model{}
+	// run the predicate on a shallow copy of the models for speed and only
+	// clone the matches
+	for u, m := range tableCache.RowsShallow() {
+		ret := reflect.ValueOf(c.predicate).Call([]reflect.Value{reflect.ValueOf(m)})
+		if ret[0].Bool() {
+			found[u] = model.Clone(m)
+		}
+	}
+	return found, nil
 }
 
 func (c *predicateConditional) Table() string {
@@ -153,29 +195,8 @@ func (c *predicateConditional) Table() string {
 // generate returns a list of conditions that match, by _uuid equality, all the objects that
 // match the predicate
 func (c *predicateConditional) Generate() ([][]ovsdb.Condition, error) {
-	allConditions := make([][]ovsdb.Condition, 0)
-	tableCache := c.cache.Table(c.tableName)
-	if tableCache == nil {
-		return nil, ErrNotFound
-	}
-	for _, row := range tableCache.Rows() {
-		match, err := c.Matches(row)
-		if err != nil {
-			return nil, err
-		}
-		if match {
-			info, err := c.cache.DatabaseModel().NewModelInfo(row)
-			if err != nil {
-				return nil, err
-			}
-			elemCond, err := c.cache.Mapper().NewEqualityCondition(info)
-			if err != nil {
-				return nil, err
-			}
-			allConditions = append(allConditions, elemCond)
-		}
-	}
-	return allConditions, nil
+	models, _ := c.Matches()
+	return generateConditionsFromModels(c.cache.DatabaseModel(), models)
 }
 
 // newPredicateConditional creates a new predicateConditional
@@ -193,8 +214,8 @@ type errorConditional struct {
 	err error
 }
 
-func (e *errorConditional) Matches(model.Model) (bool, error) {
-	return false, e.err
+func (e *errorConditional) Matches() (map[string]model.Model, error) {
+	return nil, e.err
 }
 
 func (e *errorConditional) Table() string {
diff --git a/client/doc.go b/client/doc.go
index 7f7dae5..90e409e 100644
--- a/client/doc.go
+++ b/client/doc.go
@@ -61,9 +61,8 @@ Conditions must refer to fields of the provided Model (via pointer to fields). E
 		Value: []string{"portUUID"},
 	    })
 
-If no client.Condition is provided, the client will create a default Condition based on the Model's data.
-The first non-null field that corresponds to a database index will be used. Therefore the following
-two statements are equivalent:
+If no client.Condition is provided, the client will use any of fields that correspond to indexes to
+generate an appropriate condition. Therefore the following two statements are equivalent:
 
 	ls = &MyLogicalSwitch {UUID:"myUUID"}
 
@@ -92,8 +91,11 @@ For example, the following operation will delete all the Logical Switches named
 
 To create a Condition that matches all of the conditions simultaneously (i.e: AND semantics), use WhereAll().
 
-Where() and WhereAll() inject conditions into operations that will be evaluated by the server.
-However, to perform searches on the local cache, a more flexible mechanism is available: WhereCache()
+Where() or WhereAll() evaluate the provided index values or explicit conditions against the cache and generate
+conditions based on the UUIDs of matching models. If no matches are found in the cache, the generated conditions
+will be based on the index or condition fields themselves.
+
+A more flexible mechanism to search the cache is available: WhereCache()
 
 WhereCache() accepts a function that takes any Model as argument and returns a boolean.
 It is used to search the cache so commonly used with List() function. For example:
@@ -111,7 +113,7 @@ same condition using Where() or WhereAll() which will be more efficient.
 
 Get
 
-Get() operation is a simple operation capable of retrieving one Model based on some of its indexes. E.g:
+Get() operation is a simple operation capable of retrieving one Model based on some of its schema indexes. E.g:
 
 	ls := &LogicalSwitch{UUID:"myUUID"}
 	err := ovs.Get(ls)
diff --git a/model/client.go b/model/client.go
index fe1746d..5eb6862 100644
--- a/model/client.go
+++ b/model/client.go
@@ -8,10 +8,22 @@ import (
 	"github.com/ovn-org/libovsdb/ovsdb"
 )
 
+// ColumnKey addresses a column and optionally a key within a column
+type ColumnKey struct {
+	Column string
+	Key    interface{}
+}
+
+// ClientIndex defines a client index by a set of columns
+type ClientIndex struct {
+	Columns []ColumnKey
+}
+
 // ClientDBModel contains the client information needed to build a DatabaseModel
 type ClientDBModel struct {
-	name  string
-	types map[string]reflect.Type
+	name    string
+	types   map[string]reflect.Type
+	indexes map[string][]ClientIndex
 }
 
 // NewModel returns a new instance of a model from a specific string
@@ -29,6 +41,28 @@ func (db ClientDBModel) Name() string {
 	return db.name
 }
 
+// Indexes returns the client indexes for a model
+func (db ClientDBModel) Indexes(table string) []ClientIndex {
+	if len(db.indexes) == 0 {
+		return nil
+	}
+	if _, ok := db.indexes[table]; ok {
+		return copyIndexes(db.indexes)[table]
+	}
+	return nil
+}
+
+// SetIndexes sets the client indexes. Client indexes are optional, similar to
+// schema indexes and are only tracked in the specific client instances that are
+// provided with this client model. A client index may point to multiple models
+// as uniqueness is not enforced. They are defined per table and multiple
+// indexes can be defined for a table. Each index consists of a set of columns.
+// If the column is a map, specific keys of that map can be addressed for the
+// index.
+func (db *ClientDBModel) SetIndexes(indexes map[string][]ClientIndex) {
+	db.indexes = copyIndexes(indexes)
+}
+
 // Validate validates the DatabaseModel against the input schema
 // Returns all the errors detected
 func (db ClientDBModel) validate(schema ovsdb.DatabaseSchema) []error {
@@ -38,6 +72,7 @@ func (db ClientDBModel) validate(schema ovsdb.DatabaseSchema) []error {
 			db.name, schema.Name))
 	}
 
+	infos := make(map[string]*mapper.Info, len(db.types))
 	for tableName := range db.types {
 		tableSchema := schema.Table(tableName)
 		if tableSchema == nil {
@@ -49,8 +84,41 @@ func (db ClientDBModel) validate(schema ovsdb.DatabaseSchema) []error {
 			errors = append(errors, err)
 			continue
 		}
-		if _, err := mapper.NewInfo(tableName, tableSchema, model); err != nil {
+		info, err := mapper.NewInfo(tableName, tableSchema, model)
+		if err != nil {
 			errors = append(errors, err)
+			continue
+		}
+		infos[tableName] = info
+	}
+
+	for tableName, indexSets := range db.indexes {
+		info, ok := infos[tableName]
+		if !ok {
+			errors = append(errors, fmt.Errorf("database model contains a client index for table %s that does not exist in schema", tableName))
+			continue
+		}
+		for _, indexSet := range indexSets {
+			for _, indexColumn := range indexSet.Columns {
+				f, err := info.FieldByColumn(indexColumn.Column)
+				if err != nil {
+					errors = append(
+						errors,
+						fmt.Errorf("database model contains a client index for column %s that does not exist in table %s",
+							indexColumn.Column,
+							tableName))
+					continue
+				}
+				if indexColumn.Key != nil && reflect.ValueOf(f).Kind() != reflect.Map {
+					errors = append(
+						errors,
+						fmt.Errorf("database model contains a client index for key %s in column %s of table %s that is not a map",
+							indexColumn.Key,
+							indexColumn.Column,
+							tableName))
+					continue
+				}
+			}
 		}
 	}
 	return errors
@@ -83,3 +151,21 @@ func NewClientDBModel(name string, models map[string]Model) (ClientDBModel, erro
 		name:  name,
 	}, nil
 }
+
+func copyIndexes(src map[string][]ClientIndex) map[string][]ClientIndex {
+	if len(src) == 0 {
+		return nil
+	}
+	dst := make(map[string][]ClientIndex, len(src))
+	for table, indexSets := range src {
+		dst[table] = make([]ClientIndex, 0, len(indexSets))
+		for _, indexSet := range indexSets {
+			indexSetCopy := ClientIndex{
+				Columns: make([]ColumnKey, len(indexSet.Columns)),
+			}
+			copy(indexSetCopy.Columns, indexSet.Columns)
+			dst[table] = append(dst[table], indexSetCopy)
+		}
+	}
+	return dst
+}


Test patch:
diff --git a/cache/cache_test.go b/cache/cache_test.go
index a59b6c8..47be02c 100644
--- a/cache/cache_test.go
+++ b/cache/cache_test.go
@@ -3,6 +3,7 @@ package cache
 import (
 	"encoding/json"
 	"fmt"
+	"math/rand"
 	"reflect"
 	"testing"
 
@@ -163,7 +164,96 @@ func TestRowCacheCreate(t *testing.T) {
 				assert.Error(t, err)
 			} else {
 				assert.Nil(t, err)
-				assert.Equal(t, tt.uuid, rc.indexes["foo"][tt.model.Foo])
+				assert.Len(t, rc.indexes["foo"][tt.model.Foo], 1)
+				assert.Equal(t, tt.uuid, rc.indexes["foo"][tt.model.Foo].getAny())
+			}
+		})
+	}
+}
+
+func TestRowCacheCreateClientIndex(t *testing.T) {
+	var schema ovsdb.DatabaseSchema
+	db, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &testModel{}})
+	db.SetIndexes(map[string][]model.ClientIndex{
+		"Open_vSwitch": {
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "foo",
+					},
+				},
+			},
+		},
+	})
+	require.Nil(t, err)
+	err = json.Unmarshal([]byte(`{
+		"name": "Open_vSwitch",
+		"tables": {
+		  "Open_vSwitch": {
+		    "columns": {
+		      "foo": {
+			    "type": "string"
+			  },
+			  "bar": {
+			    "type": "string"
+			  }
+		    }
+		  }
+		}
+	}`), &schema)
+	require.Nil(t, err)
+	testData := Data{
+		"Open_vSwitch": map[string]model.Model{"bar": &testModel{Foo: "bar"}},
+	}
+
+	dbModel, errs := model.NewDatabaseModel(schema, db)
+	require.Empty(t, errs)
+
+	tests := []struct {
+		name     string
+		uuid     string
+		model    *testModel
+		wantErr  bool
+		expected valueToUUIDs
+	}{
+		{
+			name:    "inserts a new row",
+			uuid:    "foo",
+			model:   &testModel{Foo: "foo"},
+			wantErr: false,
+			expected: valueToUUIDs{
+				"foo": newUUIDSet("foo"),
+				"bar": newUUIDSet("bar"),
+			},
+		},
+		{
+			name:    "error duplicate uuid",
+			uuid:    "bar",
+			model:   &testModel{Foo: "foo"},
+			wantErr: true,
+		},
+		{
+			name:    "inserts duplicate index",
+			uuid:    "baz",
+			model:   &testModel{Foo: "bar"},
+			wantErr: false,
+			expected: valueToUUIDs{
+				"bar": newUUIDSet("bar", "baz"),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc, err := NewTableCache(dbModel, testData, nil)
+			require.Nil(t, err)
+			rc := tc.Table("Open_vSwitch")
+			require.NotNil(t, rc)
+			err = rc.Create(tt.uuid, tt.model, true)
+			if tt.wantErr {
+				require.Error(t, err)
+			} else {
+				require.Nil(t, err)
+				require.Equal(t, tt.expected, rc.indexes["foo"])
 			}
 		})
 	}
@@ -191,6 +281,7 @@ func TestRowCacheCreateMultiIndex(t *testing.T) {
 	     }
 	`), &schema)
 	require.Nil(t, err)
+	index := newIndexFromColumns("foo", "bar")
 	testData := Data{
 		"Open_vSwitch": map[string]model.Model{"bar": &testModel{Foo: "bar", Bar: "bar"}},
 	}
@@ -255,9 +346,184 @@ func TestRowCacheCreateMultiIndex(t *testing.T) {
 				assert.Nil(t, err)
 				mapperInfo, err := dbModel.NewModelInfo(tt.model)
 				require.Nil(t, err)
-				h, err := valueFromIndex(mapperInfo, newIndex("foo", "bar"))
+				h, err := valueFromIndex(mapperInfo, newColumnKeysFromColumns("foo", "bar"))
 				require.Nil(t, err)
-				assert.Equal(t, tt.uuid, rc.indexes["foo,bar"][h])
+				assert.Len(t, rc.indexes[index][h], 1)
+				assert.Equal(t, tt.uuid, rc.indexes[index][h].getAny())
+			}
+		})
+	}
+}
+
+func TestRowCacheCreateMultiClientIndex(t *testing.T) {
+	type testModel struct {
+		UUID string            `ovsdb:"_uuid"`
+		Foo  string            `ovsdb:"foo"`
+		Bar  map[string]string `ovsdb:"bar"`
+	}
+	var schema ovsdb.DatabaseSchema
+	db, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &testModel{}})
+	require.Nil(t, err)
+
+	db.SetIndexes(map[string][]model.ClientIndex{
+		"Open_vSwitch": {
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "foo",
+					},
+					{
+						Column: "bar",
+						Key:    "bar",
+					},
+				},
+			},
+		},
+	})
+	index := newIndexFromColumnKeys(db.Indexes("Open_vSwitch")[0].Columns...)
+
+	err = json.Unmarshal([]byte(`{
+		"name": "Open_vSwitch",
+		"tables": {
+		  "Open_vSwitch": {
+		    "columns": {
+		      "foo": {
+			    "type": "string"
+			  },
+			  "bar": {
+				"type": {
+					"key": "string",
+					"value": "string",
+					"min": 0, 
+					"max": "unlimited"
+				}
+			  }
+		    }
+		  }
+		}
+	}`), &schema)
+	require.Nil(t, err)
+
+	testData := Data{
+		"Open_vSwitch": map[string]model.Model{"bar": &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}}},
+	}
+	dbModel, errs := model.NewDatabaseModel(schema, db)
+	require.Empty(t, errs)
+
+	type expected struct {
+		index model.Model
+		uuids uuidset
+	}
+
+	tests := []struct {
+		name     string
+		uuid     string
+		model    *testModel
+		wantErr  bool
+		expected []expected
+	}{
+		{
+			name:    "inserts a new row",
+			uuid:    "foo",
+			model:   &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar"),
+				},
+				{
+					index: &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}},
+					uuids: newUUIDSet("foo"),
+				},
+			},
+		},
+		{
+			name:    "error duplicate uuid",
+			uuid:    "bar",
+			model:   &testModel{Foo: "foo", Bar: map[string]string{"bar": "bar"}},
+			wantErr: true,
+		},
+		{
+			name:    "inserts duplicate index",
+			uuid:    "baz",
+			model:   &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar", "baz"),
+				},
+			},
+		},
+		{
+			name:    "new row with one duplicate value",
+			uuid:    "baz",
+			model:   &testModel{Foo: "foo", Bar: map[string]string{"bar": "bar"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar"),
+				},
+				{
+					index: &testModel{Foo: "foo", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("baz"),
+				},
+			},
+		},
+		{
+			name:    "new row with other duplicate value",
+			uuid:    "baz",
+			model:   &testModel{Foo: "bar", Bar: map[string]string{"bar": "foo"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar"),
+				},
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "foo"}},
+					uuids: newUUIDSet("baz"),
+				},
+			},
+		},
+		{
+			name:    "new row with nil map index",
+			uuid:    "baz",
+			model:   &testModel{Foo: "bar"},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar"),
+				},
+				{
+					index: &testModel{Foo: "bar"},
+					uuids: newUUIDSet("baz"),
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc, err := NewTableCache(dbModel, testData, nil)
+			require.Nil(t, err)
+			rc := tc.Table("Open_vSwitch")
+			require.NotNil(t, rc)
+			err = rc.Create(tt.uuid, tt.model, true)
+			if tt.wantErr {
+				require.Error(t, err)
+			} else {
+				require.Nil(t, err)
+				require.Len(t, rc.indexes[index], len(tt.expected))
+				for _, expected := range tt.expected {
+					mapperInfo, err := dbModel.NewModelInfo(expected.index)
+					require.Nil(t, err)
+					h, err := valueFromIndex(mapperInfo, db.Indexes("Open_vSwitch")[0].Columns)
+					require.Nil(t, err)
+					require.Equal(t, expected.uuids, rc.indexes[index][h], expected.index)
+				}
 			}
 		})
 	}
@@ -341,7 +607,119 @@ func TestRowCacheUpdate(t *testing.T) {
 				assert.Error(t, err)
 			} else {
 				assert.Nil(t, err)
-				assert.Equal(t, tt.uuid, rc.indexes["foo"][tt.model.Foo])
+				assert.Len(t, rc.indexes["foo"][tt.model.Foo], 1)
+				assert.Equal(t, tt.uuid, rc.indexes["foo"][tt.model.Foo].getAny())
+			}
+		})
+	}
+}
+
+func TestRowCacheUpdateClientIndex(t *testing.T) {
+	var schema ovsdb.DatabaseSchema
+	db, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &testModel{}})
+	require.Nil(t, err)
+	db.SetIndexes(map[string][]model.ClientIndex{
+		"Open_vSwitch": {
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "foo",
+					},
+				},
+			},
+		},
+	})
+	err = json.Unmarshal([]byte(`{
+		"name": "Open_vSwitch",
+		"tables": {
+		  "Open_vSwitch": {
+		    "columns": {
+		      "foo": {
+			    "type": "string"
+			  },
+			  "bar": {
+			    "type": "string"
+			  }
+		    }
+		  }
+		}
+	}`), &schema)
+	require.Nil(t, err)
+	testData := Data{
+		"Open_vSwitch": map[string]model.Model{
+			"foo":    &testModel{Foo: "foo", Bar: "foo"},
+			"bar":    &testModel{Foo: "bar", Bar: "bar"},
+			"foobar": &testModel{Foo: "bar", Bar: "foobar"},
+		},
+	}
+	dbModel, errs := model.NewDatabaseModel(schema, db)
+	require.Empty(t, errs)
+
+	tests := []struct {
+		name     string
+		uuid     string
+		model    *testModel
+		wantErr  bool
+		expected valueToUUIDs
+	}{
+		{
+			name:    "error if row does not exist",
+			uuid:    "baz",
+			model:   &testModel{Foo: "baz"},
+			wantErr: true,
+		},
+		{
+			name:    "update non-index",
+			uuid:    "foo",
+			model:   &testModel{Foo: "foo", Bar: "bar"},
+			wantErr: false,
+			expected: valueToUUIDs{
+				"foo": newUUIDSet("foo"),
+				"bar": newUUIDSet("bar", "foobar"),
+			},
+		},
+		{
+			name:    "update unique index to new index",
+			uuid:    "foo",
+			model:   &testModel{Foo: "baz"},
+			wantErr: false,
+			expected: valueToUUIDs{
+				"baz": newUUIDSet("foo"),
+				"bar": newUUIDSet("bar", "foobar"),
+			},
+		},
+		{
+			name:    "update unique index to existing index",
+			uuid:    "foo",
+			model:   &testModel{Foo: "bar"},
+			wantErr: false,
+			expected: valueToUUIDs{
+				"bar": newUUIDSet("foo", "bar", "foobar"),
+			},
+		},
+		{
+			name:    "update multi index to different index",
+			uuid:    "foobar",
+			model:   &testModel{Foo: "foo"},
+			wantErr: false,
+			expected: valueToUUIDs{
+				"foo": newUUIDSet("foo", "foobar"),
+				"bar": newUUIDSet("bar"),
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc, err := NewTableCache(dbModel, testData, nil)
+			require.Nil(t, err)
+			rc := tc.Table("Open_vSwitch")
+			require.NotNil(t, rc)
+			err = rc.Update(tt.uuid, tt.model, true)
+			if tt.wantErr {
+				require.Error(t, err)
+			} else {
+				require.Nil(t, err)
+				require.Equal(t, tt.expected, rc.indexes["foo"])
 			}
 		})
 	}
@@ -369,6 +747,7 @@ func TestRowCacheUpdateMultiIndex(t *testing.T) {
 	     }
 	`), &schema)
 	require.Nil(t, err)
+	index := newIndexFromColumns("foo", "bar")
 	testData := Data{
 		"Open_vSwitch": map[string]model.Model{
 			"bar":    &testModel{Foo: "bar", Bar: "bar"},
@@ -422,9 +801,207 @@ func TestRowCacheUpdateMultiIndex(t *testing.T) {
 				assert.Nil(t, err)
 				mapperInfo, err := dbModel.NewModelInfo(tt.model)
 				require.Nil(t, err)
-				h, err := valueFromIndex(mapperInfo, newIndex("foo", "bar"))
+				h, err := valueFromIndex(mapperInfo, newColumnKeysFromColumns("foo", "bar"))
 				require.Nil(t, err)
-				assert.Equal(t, tt.uuid, rc.indexes["foo,bar"][h])
+				assert.Len(t, rc.indexes[index][h], 1)
+				assert.Equal(t, tt.uuid, rc.indexes[index][h].getAny())
+			}
+		})
+	}
+}
+
+func TestRowCacheUpdateMultiClientIndex(t *testing.T) {
+	type testModel struct {
+		UUID string            `ovsdb:"_uuid"`
+		Foo  string            `ovsdb:"foo"`
+		Bar  map[string]string `ovsdb:"bar"`
+		Baz  string            `ovsdb:"baz"`
+	}
+	var schema ovsdb.DatabaseSchema
+	db, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &testModel{}})
+	require.Nil(t, err)
+
+	db.SetIndexes(map[string][]model.ClientIndex{
+		"Open_vSwitch": {
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "foo",
+					},
+					{
+						Column: "bar",
+						Key:    "bar",
+					},
+				},
+			},
+		},
+	})
+	index := newIndexFromColumnKeys(db.Indexes("Open_vSwitch")[0].Columns...)
+
+	err = json.Unmarshal([]byte(`{
+		"name": "Open_vSwitch",
+		"tables": {
+		  "Open_vSwitch": {
+		    "columns": {
+		      "foo": {
+			    "type": "string"
+			  },
+			  "bar": {
+				"type": {
+					"key": "string",
+					"value": "string",
+					"min": 0, 
+					"max": "unlimited"
+				}
+			  },
+			  "baz": {
+			    "type": "string"
+			  }
+		    }
+		  }
+		}
+	}`), &schema)
+	require.Nil(t, err)
+
+	testData := Data{
+		"Open_vSwitch": map[string]model.Model{
+			"foo":    &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}},
+			"bar":    &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+			"foobar": &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+		},
+	}
+	dbModel, errs := model.NewDatabaseModel(schema, db)
+	require.Empty(t, errs)
+
+	type expected struct {
+		index model.Model
+		uuids uuidset
+	}
+
+	tests := []struct {
+		name     string
+		uuid     string
+		model    *testModel
+		wantErr  bool
+		expected []expected
+	}{
+		{
+			name:    "error if row does not exist",
+			uuid:    "baz",
+			model:   &testModel{Foo: "baz", Bar: map[string]string{"bar": "baz"}},
+			wantErr: true,
+		},
+		{
+			name:  "update non-index",
+			uuid:  "foo",
+			model: &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}, Baz: "bar"},
+			expected: []expected{
+				{
+					index: &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}},
+					uuids: newUUIDSet("foo"),
+				},
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar", "foobar"),
+				},
+			},
+		},
+		{
+			name:    "update one index column",
+			uuid:    "foo",
+			model:   &testModel{Foo: "foo", Bar: map[string]string{"bar": "baz"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "foo", Bar: map[string]string{"bar": "baz"}},
+					uuids: newUUIDSet("foo"),
+				},
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar", "foobar"),
+				},
+			},
+		},
+		{
+			name:    "update other index column",
+			uuid:    "foo",
+			model:   &testModel{Foo: "baz", Bar: map[string]string{"bar": "foo"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "baz", Bar: map[string]string{"bar": "foo"}},
+					uuids: newUUIDSet("foo"),
+				},
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar", "foobar"),
+				},
+			},
+		},
+		{
+			name:    "update both index columns",
+			uuid:    "foo",
+			model:   &testModel{Foo: "baz", Bar: map[string]string{"bar": "baz"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "baz", Bar: map[string]string{"bar": "baz"}},
+					uuids: newUUIDSet("foo"),
+				},
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar", "foobar"),
+				},
+			},
+		},
+		{
+			name:    "update unique index to existing index",
+			uuid:    "foo",
+			model:   &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("foo", "bar", "foobar"),
+				},
+			},
+		},
+		{
+			name:    "update multi index to different index",
+			uuid:    "foobar",
+			model:   &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}},
+					uuids: newUUIDSet("foo", "foobar"),
+				},
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar"),
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc, err := NewTableCache(dbModel, testData, nil)
+			require.Nil(t, err)
+			rc := tc.Table("Open_vSwitch")
+			require.NotNil(t, rc)
+			err = rc.Update(tt.uuid, tt.model, true)
+			if tt.wantErr {
+				require.Error(t, err)
+			} else {
+				require.Nil(t, err)
+				require.Len(t, rc.indexes[index], len(tt.expected))
+				for _, expectedUUID := range tt.expected {
+					mapperInfo, err := dbModel.NewModelInfo(expectedUUID.index)
+					require.Nil(t, err)
+					h, err := valueFromIndex(mapperInfo, db.Indexes("Open_vSwitch")[0].Columns)
+					require.Nil(t, err)
+					require.Equal(t, expectedUUID.uuids, rc.indexes[index][h], expectedUUID.index)
+				}
 			}
 		})
 	}
@@ -490,7 +1067,137 @@ func TestRowCacheDelete(t *testing.T) {
 				assert.Error(t, err)
 			} else {
 				require.Nil(t, err)
-				assert.Equal(t, "", rc.indexes["foo"][tt.model.Foo])
+				assert.Nil(t, rc.indexes["foo"][tt.model.Foo])
+			}
+		})
+	}
+}
+
+func TestRowCacheDeleteClientIndex(t *testing.T) {
+	type testModel struct {
+		UUID string            `ovsdb:"_uuid"`
+		Foo  string            `ovsdb:"foo"`
+		Bar  map[string]string `ovsdb:"bar"`
+	}
+	var schema ovsdb.DatabaseSchema
+	db, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &testModel{}})
+	require.Nil(t, err)
+
+	db.SetIndexes(map[string][]model.ClientIndex{
+		"Open_vSwitch": {
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "foo",
+					},
+					{
+						Column: "bar",
+						Key:    "bar",
+					},
+				},
+			},
+		},
+	})
+	index := newIndexFromColumnKeys(db.Indexes("Open_vSwitch")[0].Columns...)
+
+	err = json.Unmarshal([]byte(`{
+		"name": "Open_vSwitch",
+		"tables": {
+		  "Open_vSwitch": {
+		    "columns": {
+		      "foo": {
+			    "type": "string"
+			  },
+			  "bar": {
+				"type": {
+					"key": "string",
+					"value": "string",
+					"min": 0, 
+					"max": "unlimited"
+				}
+			  }
+		    }
+		  }
+		}
+	}`), &schema)
+	require.Nil(t, err)
+
+	testData := Data{
+		"Open_vSwitch": map[string]model.Model{
+			"foo":    &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}},
+			"bar":    &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+			"foobar": &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+		},
+	}
+	dbModel, errs := model.NewDatabaseModel(schema, db)
+	require.Empty(t, errs)
+
+	type expected struct {
+		index model.Model
+		uuids uuidset
+	}
+
+	tests := []struct {
+		name     string
+		uuid     string
+		model    *testModel
+		wantErr  bool
+		expected []expected
+	}{
+		{
+			name:    "error if row does not exist",
+			uuid:    "baz",
+			model:   &testModel{Foo: "baz", Bar: map[string]string{"bar": "baz"}},
+			wantErr: true,
+		},
+		{
+			name:    "delete a row with unique index",
+			uuid:    "foo",
+			model:   &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar", "foobar"),
+				},
+			},
+		},
+		{
+			name:    "delete a row with duplicated index",
+			uuid:    "foobar",
+			model:   &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+			wantErr: false,
+			expected: []expected{
+				{
+					index: &testModel{Foo: "foo", Bar: map[string]string{"bar": "foo"}},
+					uuids: newUUIDSet("foo"),
+				},
+				{
+					index: &testModel{Foo: "bar", Bar: map[string]string{"bar": "bar"}},
+					uuids: newUUIDSet("bar"),
+				},
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc, err := NewTableCache(dbModel, testData, nil)
+			require.Nil(t, err)
+			rc := tc.Table("Open_vSwitch")
+			require.NotNil(t, rc)
+			err = rc.Delete(tt.uuid)
+			if tt.wantErr {
+				require.Error(t, err)
+			} else {
+				require.Nil(t, err)
+				require.Len(t, rc.indexes[index], len(tt.expected))
+				for _, expected := range tt.expected {
+					mapperInfo, err := dbModel.NewModelInfo(expected.index)
+					require.Nil(t, err)
+					h, err := valueFromIndex(mapperInfo, db.Indexes("Open_vSwitch")[0].Columns)
+					require.Nil(t, err)
+					require.Equal(t, expected.uuids, rc.indexes[index][h], expected.index)
+				}
 			}
 		})
 	}
@@ -1064,7 +1771,7 @@ func TestTableCachePopulate2BrokenIndexes(t *testing.T) {
 	assert.False(t, ok)
 
 	t.Log("Lookup Original Insert By Index")
-	result := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "bar"})
+	_, result := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "bar"})
 	require.NotNil(t, result)
 }
 
@@ -1108,6 +1815,27 @@ func TestIndex(t *testing.T) {
 	}
 	db, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &indexTestModel{}})
 	assert.Nil(t, err)
+	db.SetIndexes(map[string][]model.ClientIndex{
+		"Open_vSwitch": {
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "bar",
+					},
+				},
+			},
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "foo",
+					},
+					{
+						Column: "baz",
+					},
+				},
+			},
+		},
+	})
 	var schema ovsdb.DatabaseSchema
 	err = json.Unmarshal([]byte(`
 		 {"name": "Open_vSwitch",
@@ -1134,34 +1862,44 @@ func TestIndex(t *testing.T) {
 	assert.Empty(t, errs)
 	tc, err := NewTableCache(dbModel, nil, nil)
 	assert.Nil(t, err)
+	table := tc.Table("Open_vSwitch")
+
 	obj := &indexTestModel{
 		UUID: "test1",
 		Foo:  "foo",
 		Bar:  "bar",
 		Baz:  42,
 	}
-	table := tc.Table("Open_vSwitch")
-
 	err = table.Create(obj.UUID, obj, true)
 	assert.Nil(t, err)
+
+	obj2 := &indexTestModel{
+		UUID: "test2",
+		Foo:  "foo2",
+		Bar:  "bar",
+		Baz:  78,
+	}
+	err = table.Create(obj2.UUID, obj2, true)
+	assert.Nil(t, err)
+
 	t.Run("Index by single column", func(t *testing.T) {
 		idx, err := table.Index("foo")
 		assert.Nil(t, err)
 		info, err := dbModel.NewModelInfo(obj)
 		assert.Nil(t, err)
-		v, err := valueFromIndex(info, newIndex("foo"))
+		v, err := valueFromIndex(info, newColumnKeysFromColumns("foo"))
 		assert.Nil(t, err)
-		assert.Equal(t, idx[v], obj.UUID)
+		assert.ElementsMatch(t, idx[v], []string{obj.UUID})
 	})
 	t.Run("Index by single column miss", func(t *testing.T) {
 		idx, err := table.Index("foo")
 		assert.Nil(t, err)
-		obj2 := obj
-		obj2.Foo = "wrong"
+		obj3 := *obj
+		obj3.Foo = "wrong"
 		assert.Nil(t, err)
-		info, err := dbModel.NewModelInfo(obj2)
+		info, err := dbModel.NewModelInfo(&obj3)
 		assert.Nil(t, err)
-		v, err := valueFromIndex(info, newIndex("foo"))
+		v, err := valueFromIndex(info, newColumnKeysFromColumns("foo"))
 		assert.Nil(t, err)
 		_, ok := idx[v]
 		assert.False(t, ok)
@@ -1179,25 +1917,39 @@ func TestIndex(t *testing.T) {
 		assert.Nil(t, err)
 		info, err := dbModel.NewModelInfo(obj)
 		assert.Nil(t, err)
-		v, err := valueFromIndex(info, newIndex("bar", "baz"))
+		v, err := valueFromIndex(info, newColumnKeysFromColumns("bar", "baz"))
 		assert.Nil(t, err)
-		assert.Equal(t, idx[v], obj.UUID)
+		assert.ElementsMatch(t, idx[v], []string{obj.UUID})
 	})
 	t.Run("Index by multi-column miss", func(t *testing.T) {
 		idx, err := table.Index("bar", "baz")
 		assert.Nil(t, err)
-		obj2 := obj
-		obj2.Baz++
-		info, err := dbModel.NewModelInfo(obj)
+		obj3 := *obj
+		obj3.Baz++
+		info, err := dbModel.NewModelInfo(&obj3)
 		assert.Nil(t, err)
-		v, err := valueFromIndex(info, newIndex("bar", "baz"))
+		v, err := valueFromIndex(info, newColumnKeysFromColumns("bar", "baz"))
 		assert.Nil(t, err)
 		_, ok := idx[v]
 		assert.False(t, ok)
 	})
-	t.Run("Index type", func(t *testing.T) {
-		idx := newIndex("foo", "bar")
-		assert.Equal(t, idx.columns(), []string{"foo", "bar"})
+	t.Run("Client index by single column", func(t *testing.T) {
+		idx, err := table.Index("bar")
+		assert.Nil(t, err)
+		info, err := dbModel.NewModelInfo(obj)
+		assert.Nil(t, err)
+		v, err := valueFromIndex(info, newColumnKeysFromColumns("bar"))
+		assert.Nil(t, err)
+		assert.ElementsMatch(t, idx[v], []string{obj.UUID, obj2.UUID})
+	})
+	t.Run("Client index by multiple column", func(t *testing.T) {
+		idx, err := table.Index("foo", "baz")
+		assert.Nil(t, err)
+		info, err := dbModel.NewModelInfo(obj)
+		assert.Nil(t, err)
+		v, err := valueFromIndex(info, newColumnKeysFromColumns("foo", "baz"))
+		assert.Nil(t, err)
+		assert.ElementsMatch(t, idx[v], []string{obj.UUID})
 	})
 }
 
@@ -1242,18 +1994,18 @@ func TestTableCacheRowByModelSingleIndex(t *testing.T) {
 	myFoo, tc := setupRowByModelSingleIndex(t)
 
 	t.Run("get foo by index", func(t *testing.T) {
-		foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "foo"})
+		_, foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "foo"})
 		assert.NotNil(t, foo)
 		assert.Equal(t, myFoo, foo)
 	})
 
 	t.Run("get non-existent item by index", func(t *testing.T) {
-		baz := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "baz"})
+		_, baz := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "baz"})
 		assert.Nil(t, baz)
 	})
 
 	t.Run("no index data", func(t *testing.T) {
-		foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Bar: "foo"})
+		_, foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Bar: "foo"})
 		assert.Nil(t, foo)
 	})
 }
@@ -1354,19 +2106,19 @@ func TestTableCacheRowByModelTwoIndexes(t *testing.T) {
 	require.NoError(t, err)
 
 	t.Run("get foo by Foo index", func(t *testing.T) {
-		foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "foo"})
+		_, foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "foo"})
 		assert.NotNil(t, foo)
 		assert.Equal(t, myFoo, foo)
 	})
 
 	t.Run("get foo by Bar index", func(t *testing.T) {
-		foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Bar: "foo"})
+		_, foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Bar: "foo"})
 		assert.NotNil(t, foo)
 		assert.Equal(t, myFoo, foo)
 	})
 
 	t.Run("get non-existent item by index", func(t *testing.T) {
-		baz := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "baz"})
+		_, baz := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "baz"})
 		assert.Nil(t, baz)
 	})
 
@@ -1404,18 +2156,18 @@ func TestTableCacheRowByModelMultiIndex(t *testing.T) {
 	require.NoError(t, err)
 
 	t.Run("incomplete index", func(t *testing.T) {
-		foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "foo"})
+		_, foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "foo"})
 		assert.Nil(t, foo)
 	})
 
 	t.Run("get foo by index", func(t *testing.T) {
-		foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "foo", Bar: "foo"})
+		_, foo := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "foo", Bar: "foo"})
 		assert.NotNil(t, foo)
 		assert.Equal(t, myFoo, foo)
 	})
 
 	t.Run("get non-existent item by index", func(t *testing.T) {
-		baz := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "baz", Bar: "baz"})
+		_, baz := tc.Table("Open_vSwitch").RowByModel(&testModel{Foo: "baz", Bar: "baz"})
 		assert.Nil(t, baz)
 	})
 }
@@ -1563,3 +2315,550 @@ func TestTableCacheApplyModifications(t *testing.T) {
 		})
 	}
 }
+
+func TestTableCacheRowsByModel(t *testing.T) {
+	type testModel struct {
+		UUID string            `ovsdb:"_uuid"`
+		Foo  string            `ovsdb:"foo"`
+		Bar  string            `ovsdb:"bar"`
+		Baz  map[string]string `ovsdb:"baz"`
+	}
+	var schema ovsdb.DatabaseSchema
+	db, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &testModel{}})
+	require.NoError(t, err)
+	db.SetIndexes(map[string][]model.ClientIndex{
+		"Open_vSwitch": {
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "bar",
+					},
+				},
+			},
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "bar",
+					},
+					{
+						Column: "baz",
+						Key:    "baz",
+					},
+				},
+			},
+		},
+	})
+	err = json.Unmarshal([]byte(`{
+		"name": "Open_vSwitch",
+		"tables": {
+		  "Open_vSwitch": {
+			"indexes": [["foo"]],
+			"columns": {	
+			  "foo": {
+				"type": "string"
+			  },
+			  "bar": {
+				"type": "string"
+			  },
+			  "baz": {
+				"type": {
+					"key": "string",
+					"value": "string",
+					"min": 0, 
+					"max": "unlimited"
+				}
+			  }
+			}
+		  }
+		}
+	}`), &schema)
+	require.NoError(t, err)
+
+	testData := Data{
+		"Open_vSwitch": map[string]model.Model{
+			"foo":    &testModel{Foo: "foo", Bar: "foo", Baz: map[string]string{"baz": "foo", "other": "other"}},
+			"bar":    &testModel{Foo: "bar", Bar: "bar", Baz: map[string]string{"baz": "bar", "other": "other"}},
+			"foobar": &testModel{Foo: "foobar", Bar: "bar", Baz: map[string]string{"baz": "foobar", "other": "other"}},
+			"baz":    &testModel{Foo: "baz", Bar: "baz", Baz: map[string]string{"baz": "baz", "other": "other"}},
+			"quux":   &testModel{Foo: "quux", Bar: "quux", Baz: map[string]string{"baz": "quux", "other": "other"}},
+			"quuz":   &testModel{Foo: "quuz", Bar: "quux", Baz: map[string]string{"baz": "quux", "other": "other"}},
+		},
+	}
+	dbModel, errs := model.NewDatabaseModel(schema, db)
+	require.Empty(t, errs)
+
+	tests := []struct {
+		name  string
+		model model.Model
+		rows  map[string]model.Model
+	}{
+		{
+			"by non index, no result",
+			&testModel{Foo: "no", Bar: "no", Baz: map[string]string{"baz": "no"}},
+			nil,
+		},
+		{
+			"by single column client index, single result",
+			&testModel{Bar: "foo"},
+			map[string]model.Model{
+				"foo": testData["Open_vSwitch"]["foo"],
+			},
+		},
+		{
+			"by single column client index, multiple results",
+			&testModel{Bar: "bar"},
+			map[string]model.Model{
+				"bar":    testData["Open_vSwitch"]["bar"],
+				"foobar": testData["Open_vSwitch"]["foobar"],
+			},
+		},
+		{
+			"by multi column client index, single result",
+			&testModel{Bar: "baz", Baz: map[string]string{"baz": "baz"}},
+			map[string]model.Model{
+				"baz": testData["Open_vSwitch"]["baz"],
+			},
+		},
+		{
+			"by client index, multiple results",
+			&testModel{Bar: "quux", Baz: map[string]string{"baz": "quux"}},
+			map[string]model.Model{
+				"quux": testData["Open_vSwitch"]["quux"],
+				"quuz": testData["Open_vSwitch"]["quuz"],
+			},
+		},
+		{
+			"by schema index prioritized over client index",
+			&testModel{Foo: "foo", Bar: "bar", Baz: map[string]string{"baz": "bar"}},
+			map[string]model.Model{
+				"foo": testData["Open_vSwitch"]["foo"],
+			},
+		},
+	}
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc, err := NewTableCache(dbModel, testData, nil)
+			require.NoError(t, err)
+			rows := tc.Table("Open_vSwitch").RowsByModel(tt.model)
+			require.Equal(t, tt.rows, rows)
+		})
+	}
+}
+
+type rowsByConditionTestModel struct {
+	UUID   string            `ovsdb:"_uuid"`
+	Foo    string            `ovsdb:"foo"`
+	Bar    string            `ovsdb:"bar"`
+	Baz    string            `ovsdb:"baz"`
+	Quux   string            `ovsdb:"quux"`
+	Quuz   string            `ovsdb:"quuz"`
+	FooBar map[string]string `ovsdb:"foobar"`
+	Empty  string            `ovsdb:"empty"`
+}
+
+func setupRowsByConditionCache(t require.TestingT) *TableCache {
+	var schema ovsdb.DatabaseSchema
+	db, err := model.NewClientDBModel("Open_vSwitch", map[string]model.Model{"Open_vSwitch": &rowsByConditionTestModel{}})
+	require.NoError(t, err)
+	db.SetIndexes(map[string][]model.ClientIndex{
+		"Open_vSwitch": {
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "foobar",
+						Key:    "foobar",
+					},
+				},
+			},
+			{
+				Columns: []model.ColumnKey{
+					{
+						Column: "empty",
+					},
+				},
+			},
+		},
+	})
+	err = json.Unmarshal([]byte(`{
+		"name": "Open_vSwitch",
+		"tables": {
+		  "Open_vSwitch": {
+			"indexes": [["foo"], ["bar"], ["quux", "quuz"]],
+			"columns": {	
+			  "foo": {
+				"type": "string"
+			  },
+			  "bar": {
+				"type": "string"
+			  },
+			  "baz": {
+				"type": "string"
+			  },
+			  "quux": {
+			    "type": "string"
+			  },
+			  "quuz": {
+			    "type": "string"
+			  },
+			  "foobar": {
+				"type": {
+					"key": "string",
+					"value": "string",
+					"min": 0, 
+					"max": "unlimited"
+				}
+			  },
+			  "empty": {
+			    "type": "string"
+			  }
+			}
+		  }
+		}
+	}`), &schema)
+	require.NoError(t, err)
+
+	dbModel, errs := model.NewDatabaseModel(schema, db)
+	require.Empty(t, errs)
+	tc, err := NewTableCache(dbModel, nil, nil)
+	require.NoError(t, err)
+	return tc
+}
+
+func TestTableCacheRowsByCondition(t *testing.T) {
+	testData := map[string]*rowsByConditionTestModel{
+		"foo":  {UUID: "foo", Foo: "foo", Bar: "foo", Baz: "foo", Quux: "foo", Quuz: "quuz", FooBar: map[string]string{"foobar": "foo"}},
+		"bar":  {UUID: "bar", Foo: "bar", Bar: "bar", Baz: "bar", Quux: "bar", Quuz: "quuz", FooBar: map[string]string{"foobar": "bar"}},
+		"baz":  {UUID: "baz", Foo: "baz", Bar: "baz", Baz: "baz", Quux: "baz", Quuz: "quuz", FooBar: map[string]string{"foobar": "baz"}},
+		"quux": {UUID: "quux", Foo: "quux", Bar: "quux", Baz: "quux", Quux: "quux", Quuz: "quuz", FooBar: map[string]string{"foobar": "baz"}},
+		"quuz": {UUID: "quuz", Foo: "quuz", Bar: "quuz", Baz: "quuz", Quux: "quuz", Quuz: "quuz", FooBar: map[string]string{"foobar": "baz"}},
+	}
+
+	tests := []struct {
+		name       string
+		conditions []ovsdb.Condition
+		// uuids that could be found evaluating conditions as indexes
+		uuidsByConditionsAsIndexes uuidset
+		// rows that could be found evaluating all conditions
+		rowsByCondition map[string]model.Model
+	}{
+		{
+			"by equal uuid",
+			[]ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: "foo"}}},
+			nil,
+			map[string]model.Model{"foo": testData["foo"]},
+		},
+		{
+			"by includes uuid",
+			[]ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionIncludes, Value: ovsdb.UUID{GoUUID: "foo"}}},
+			nil,
+			map[string]model.Model{"foo": testData["foo"]},
+		},
+		{
+			"by non equal uuid, multiple results",
+			[]ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionNotEqual, Value: ovsdb.UUID{GoUUID: "foo"}}},
+			nil,
+			map[string]model.Model{
+				"bar":  testData["bar"],
+				"baz":  testData["baz"],
+				"quux": testData["quux"],
+				"quuz": testData["quuz"],
+			},
+		},
+		{
+			"by excludes uuid, multiple results",
+			[]ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionExcludes, Value: ovsdb.UUID{GoUUID: "foo"}}},
+			nil,
+			map[string]model.Model{
+				"bar":  testData["bar"],
+				"baz":  testData["baz"],
+				"quux": testData["quux"],
+				"quuz": testData["quuz"],
+			},
+		},
+		{
+			"by schema index",
+			[]ovsdb.Condition{{Column: "foo", Function: ovsdb.ConditionEqual, Value: "foo"}},
+			newUUIDSet("foo"),
+			map[string]model.Model{"foo": testData["foo"]},
+		},
+		{
+			"by schema index, no results",
+			[]ovsdb.Condition{{Column: "foo", Function: ovsdb.ConditionEqual, Value: "foobar"}},
+			newUUIDSet(),
+			map[string]model.Model{},
+		},
+		{
+			"by multi column schema index",
+			[]ovsdb.Condition{
+				{Column: "quux", Function: ovsdb.ConditionEqual, Value: "foo"},
+				{Column: "quuz", Function: ovsdb.ConditionEqual, Value: "quuz"},
+			},
+			newUUIDSet("foo"),
+			map[string]model.Model{"foo": testData["foo"]},
+		},
+		{
+			"by multi column schema index, no results",
+			[]ovsdb.Condition{
+				{Column: "quux", Function: ovsdb.ConditionEqual, Value: "foobar"},
+				{Column: "quuz", Function: ovsdb.ConditionEqual, Value: "quuz"},
+			},
+			newUUIDSet(),
+			map[string]model.Model{},
+		},
+		{
+			"by client index",
+			[]ovsdb.Condition{{Column: "foobar", Function: ovsdb.ConditionIncludes, Value: ovsdb.OvsMap{GoMap: map[interface{}]interface{}{"foobar": "bar"}}}},
+			newUUIDSet("bar"),
+			map[string]model.Model{"bar": testData["bar"]},
+		},
+		{
+			"by client index, no results",
+			[]ovsdb.Condition{{Column: "foobar", Function: ovsdb.ConditionIncludes, Value: ovsdb.OvsMap{GoMap: map[interface{}]interface{}{"foobar": "foobar"}}}},
+			newUUIDSet(),
+			map[string]model.Model{},
+		},
+		{
+			"by client index, multiple results",
+			[]ovsdb.Condition{{Column: "foobar", Function: ovsdb.ConditionIncludes, Value: ovsdb.OvsMap{GoMap: map[interface{}]interface{}{"foobar": "baz"}}}},
+			newUUIDSet("baz", "quux", "quuz"),
+			map[string]model.Model{
+				"baz":  testData["baz"],
+				"quux": testData["quux"],
+				"quuz": testData["quuz"],
+			},
+		},
+		{
+			"by zero client index, multiple results",
+			[]ovsdb.Condition{{Column: "empty", Function: ovsdb.ConditionEqual, Value: ""}},
+			newUUIDSet("foo", "bar", "baz", "quux", "quuz"),
+			map[string]model.Model{
+				"foo":  testData["foo"],
+				"bar":  testData["bar"],
+				"baz":  testData["baz"],
+				"quux": testData["quux"],
+				"quuz": testData["quuz"],
+			},
+		},
+		{
+			"by non index",
+			[]ovsdb.Condition{{Column: "baz", Function: ovsdb.ConditionEqual, Value: "baz"}},
+			nil,
+			map[string]model.Model{"baz": testData["baz"]},
+		},
+		{
+			"by two uuids, no results",
+			[]ovsdb.Condition{
+				{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: "foo"}},
+				{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: "bar"}},
+			},
+			nil,
+			map[string]model.Model{},
+		},
+		{
+			"by uuid and schema index",
+			[]ovsdb.Condition{
+				{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: "foo"}},
+				{Column: "foo", Function: ovsdb.ConditionEqual, Value: "foo"},
+			},
+			newUUIDSet("foo"),
+			map[string]model.Model{"foo": testData["foo"]},
+		},
+		{
+			"by uuid and schema index, no results",
+			[]ovsdb.Condition{
+				{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: "foo"}},
+				{Column: "foo", Function: ovsdb.ConditionEqual, Value: "bar"},
+			},
+			newUUIDSet("bar"),
+			map[string]model.Model{},
+		},
+		{
+			"by schema index and non-index",
+			[]ovsdb.Condition{
+				{Column: "foo", Function: ovsdb.ConditionEqual, Value: "foo"},
+				{Column: "baz", Function: ovsdb.ConditionEqual, Value: "foo"},
+			},
+			newUUIDSet("foo"),
+			map[string]model.Model{"foo": testData["foo"]},
+		},
+		{
+			"by schema index and non-index, no results",
+			[]ovsdb.Condition{
+				{Column: "foo", Function: ovsdb.ConditionEqual, Value: "foo"},
+				{Column: "baz", Function: ovsdb.ConditionEqual, Value: "baz"},
+			},
+			newUUIDSet("foo"),
+			map[string]model.Model{},
+		},
+		{
+			"by uuid, schema index, and non-index",
+			[]ovsdb.Condition{
+				{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: "foo"}},
+				{Column: "foo", Function: ovsdb.ConditionEqual, Value: "foo"},
+				{Column: "bar", Function: ovsdb.ConditionEqual, Value: "foo"},
+				{Column: "baz", Function: ovsdb.ConditionEqual, Value: "foo"},
+			},
+			newUUIDSet("foo"),
+			map[string]model.Model{"foo": testData["foo"]},
+		},
+		{
+			"by client index, and non-index, multiple results",
+			[]ovsdb.Condition{
+				{Column: "foobar", Function: ovsdb.ConditionIncludes, Value: ovsdb.OvsMap{GoMap: map[interface{}]interface{}{"foobar": "baz"}}},
+				{Column: "quuz", Function: ovsdb.ConditionEqual, Value: "quuz"},
+			},
+			newUUIDSet("baz", "quux", "quuz"),
+			map[string]model.Model{
+				"baz":  testData["baz"],
+				"quux": testData["quux"],
+				"quuz": testData["quuz"],
+			},
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			tc := setupRowsByConditionCache(t)
+			rc := tc.Table("Open_vSwitch")
+			for _, m := range testData {
+				err := rc.Create(m.UUID, m, true)
+				require.NoError(t, err)
+			}
+
+			nativeValues := make([]interface{}, 0, len(tt.conditions))
+			for _, condition := range tt.conditions {
+				cSchema := rc.dbModel.Schema.Tables["Open_vSwitch"].Column(condition.Column)
+				nativeValue, err := ovsdb.OvsToNative(cSchema, condition.Value)
+				require.NoError(t, err)
+				nativeValues = append(nativeValues, nativeValue)
+			}
+
+			uuids, err := tc.Table("Open_vSwitch").uuidsByConditionsAsIndexes(tt.conditions, nativeValues)
+			require.NoError(t, err)
+			require.Equal(t, tt.uuidsByConditionsAsIndexes, uuids)
+
+			rows, err := tc.Table("Open_vSwitch").RowsByCondition(tt.conditions)
+			require.NoError(t, err)
+			require.Equal(t, tt.rowsByCondition, rows)
+		})
+	}
+}
+
+func BenchmarkRowsByCondition(b *testing.B) {
+	tc := setupRowsByConditionCache(b)
+	rc := tc.Table("Open_vSwitch")
+
+	models := []*rowsByConditionTestModel{}
+	for i := 0; i < numRows; i++ {
+		model := &rowsByConditionTestModel{
+			UUID:   fmt.Sprintf("UUID-%d", i),
+			Foo:    fmt.Sprintf("Foo-%d", i),
+			Bar:    fmt.Sprintf("Bar-%d", i),
+			Baz:    fmt.Sprintf("Baz-%d", i),
+			Quux:   fmt.Sprintf("Quux-%d", i),
+			Quuz:   fmt.Sprintf("Quuz-%d", i),
+			FooBar: map[string]string{"foobar": fmt.Sprintf("FooBar-%d", i)},
+		}
+		err := rc.Create(model.UUID, model, true)
+		require.NoError(b, err)
+		models = append(models, model)
+	}
+
+	rand.Seed(int64(b.N))
+
+	benchmarks := []struct {
+		name    string
+		prepare func(int) []ovsdb.Condition
+	}{
+		{
+			name: "by uuid",
+			prepare: func(i int) []ovsdb.Condition {
+				return []ovsdb.Condition{
+					{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: models[i].UUID}},
+				}
+			},
+		},
+		{
+			name: "by single column squema index",
+			prepare: func(i int) []ovsdb.Condition {
+				return []ovsdb.Condition{
+					{Column: "foo", Function: ovsdb.ConditionEqual, Value: models[i].Foo},
+				}
+			},
+		},
+		{
+			name: "by single column client index",
+			prepare: func(i int) []ovsdb.Condition {
+				return []ovsdb.Condition{
+					{Column: "foobar", Function: ovsdb.ConditionIncludes, Value: ovsdb.OvsMap{GoMap: map[interface{}]interface{}{"foobar": models[i].FooBar["foobar"]}}},
+				}
+			},
+		},
+		{
+			name: "by multi column squema index",
+			prepare: func(i int) []ovsdb.Condition {
+				return []ovsdb.Condition{
+					{Column: "quux", Function: ovsdb.ConditionEqual, Value: models[i].Quux},
+					{Column: "quuz", Function: ovsdb.ConditionEqual, Value: models[i].Quuz},
+				}
+			},
+		},
+		{
+			name: "by two squema indexes",
+			prepare: func(i int) []ovsdb.Condition {
+				return []ovsdb.Condition{
+					{Column: "foo", Function: ovsdb.ConditionEqual, Value: models[i].Foo},
+					{Column: "bar", Function: ovsdb.ConditionEqual, Value: models[i].Bar},
+				}
+			},
+		},
+		{
+			name: "by squema index and non-index",
+			prepare: func(i int) []ovsdb.Condition {
+				return []ovsdb.Condition{
+					{Column: "foo", Function: ovsdb.ConditionEqual, Value: models[i].Foo},
+					{Column: "quuz", Function: ovsdb.ConditionEqual, Value: models[i].Quuz},
+				}
+			},
+		},
+		{
+			name: "by single non index",
+			prepare: func(i int) []ovsdb.Condition {
+				return []ovsdb.Condition{
+					{Column: "quuz", Function: ovsdb.ConditionEqual, Value: models[i].Quuz},
+				}
+			},
+		},
+		{
+			name: "by multiple non indexes",
+			prepare: func(i int) []ovsdb.Condition {
+				return []ovsdb.Condition{
+					{Column: "baz", Function: ovsdb.ConditionEqual, Value: models[i].Baz},
+					{Column: "quuz", Function: ovsdb.ConditionEqual, Value: models[i].Quuz},
+				}
+			},
+		},
+		{
+			name: "by many conditions",
+			prepare: func(i int) []ovsdb.Condition {
+				return []ovsdb.Condition{
+					{Column: "foo", Function: ovsdb.ConditionEqual, Value: models[i].Foo},
+					{Column: "bar", Function: ovsdb.ConditionEqual, Value: models[i].Bar},
+					{Column: "baz", Function: ovsdb.ConditionEqual, Value: models[i].Baz},
+					{Column: "quux", Function: ovsdb.ConditionEqual, Value: models[i].Quux},
+					{Column: "quuz", Function: ovsdb.ConditionEqual, Value: models[i].Quuz},
+					{Column: "foobar", Function: ovsdb.ConditionIncludes, Value: ovsdb.OvsMap{GoMap: map[interface{}]interface{}{"foobar": models[i].FooBar["foobar"]}}},
+				}
+			},
+		},
+	}
+	for _, bm := range benchmarks {
+		b.Run(bm.name, func(b *testing.B) {
+			for i := 0; i < b.N; i++ {
+				results, err := rc.RowsByCondition(bm.prepare(rand.Intn(numRows)))
+				require.NoError(b, err)
+				require.Len(b, results, 1)
+			}
+		})
+	}
+}
diff --git a/client/api_test.go b/client/api_test.go
index 51fdf91..c878191 100644
--- a/client/api_test.go
+++ b/client/api_test.go
@@ -3,6 +3,7 @@ package client
 import (
 	"context"
 	"fmt"
+	"math/rand"
 	"strings"
 	"testing"
 
@@ -252,7 +253,7 @@ func TestAPIListPredicate(t *testing.T) {
 
 	for _, tt := range test {
 		t.Run(fmt.Sprintf("ApiListPredicate: %s", tt.name), func(t *testing.T) {
-			var result []testLogicalSwitch
+			var result []*testLogicalSwitch
 			api := newAPI(tcache, &discardLogger)
 			cond := api.WhereCache(tt.predicate)
 			err := cond.List(context.Background(), &result)
@@ -262,13 +263,95 @@ func TestAPIListPredicate(t *testing.T) {
 				if !assert.Nil(t, err) {
 					t.Log(err)
 				}
-				assert.ElementsMatchf(t, tt.content, tt.content, "Content should match")
+				assert.ElementsMatchf(t, tt.content, result, "Content should match")
 			}
 
 		})
 	}
 }
 
+func TestAPIListWhereConditions(t *testing.T) {
+	lscacheList := []model.Model{
+		&testLogicalSwitchPort{
+			UUID: aUUID0,
+			Name: "lsp0",
+			Type: "",
+		},
+		&testLogicalSwitchPort{
+			UUID: aUUID1,
+			Name: "lsp1",
+			Type: "router",
+		},
+		&testLogicalSwitchPort{
+			UUID: aUUID2,
+			Name: "lsp2",
+			Type: "router",
+		},
+		&testLogicalSwitchPort{
+			UUID: aUUID3,
+			Name: "lsp3",
+			Type: "localnet",
+		},
+	}
+	lscache := map[string]model.Model{}
+	for i := range lscacheList {
+		lscache[lscacheList[i].(*testLogicalSwitchPort).UUID] = lscacheList[i]
+	}
+	testData := cache.Data{
+		"Logical_Switch_Port": lscache,
+	}
+	tcache := apiTestCache(t, testData)
+
+	test := []struct {
+		desc       string
+		matchNames []string
+		matchTypes []string
+		matchAll   bool
+		result     []model.Model
+	}{
+		{
+			desc:       "any conditions",
+			matchNames: []string{"lsp0"},
+			matchTypes: []string{"router"},
+			matchAll:   false,
+			result:     lscacheList[0:3],
+		},
+		{
+			desc:       "all conditions",
+			matchNames: []string{"lsp1"},
+			matchTypes: []string{"router"},
+			matchAll:   true,
+			result:     lscacheList[1:2],
+		},
+	}
+
+	for _, tt := range test {
+		t.Run(fmt.Sprintf("TestAPIListWhereConditions: %s", tt.desc), func(t *testing.T) {
+			var result []*testLogicalSwitchPort
+			api := newAPI(tcache, &discardLogger)
+			testObj := &testLogicalSwitchPort{}
+			conds := []model.Condition{}
+			for _, name := range tt.matchNames {
+				cond := model.Condition{Field: &testObj.Name, Function: ovsdb.ConditionEqual, Value: name}
+				conds = append(conds, cond)
+			}
+			for _, atype := range tt.matchTypes {
+				cond := model.Condition{Field: &testObj.Type, Function: ovsdb.ConditionEqual, Value: atype}
+				conds = append(conds, cond)
+			}
+			var capi ConditionalAPI
+			if tt.matchAll {
+				capi = api.WhereAll(testObj, conds...)
+			} else {
+				capi = api.Where(testObj, conds...)
+			}
+			err := capi.List(context.Background(), &result)
+			assert.NoError(t, err)
+			assert.ElementsMatchf(t, tt.result, result, "Content should match")
+		})
+	}
+}
+
 func TestAPIListFields(t *testing.T) {
 	lspcacheList := []model.Model{
 		&testLogicalSwitchPort{
@@ -305,19 +388,16 @@ func TestAPIListFields(t *testing.T) {
 	}
 	tcache := apiTestCache(t, testData)
 
-	testObj := testLogicalSwitchPort{}
-
 	test := []struct {
 		name    string
 		fields  []interface{}
 		prepare func(*testLogicalSwitchPort)
 		content []model.Model
-		err     bool
 	}{
 		{
-			name:    "empty object must match everything",
-			content: lspcacheList,
-			err:     false,
+			name:    "No match",
+			prepare: func(t *testLogicalSwitchPort) {},
+			content: []model.Model{},
 		},
 		{
 			name: "List unique by UUID",
@@ -325,7 +405,6 @@ func TestAPIListFields(t *testing.T) {
 				t.UUID = aUUID0
 			},
 			content: []model.Model{lspcache[aUUID0]},
-			err:     false,
 		},
 		{
 			name: "List unique by Index",
@@ -333,24 +412,19 @@ func TestAPIListFields(t *testing.T) {
 				t.Name = "lsp2"
 			},
 			content: []model.Model{lspcache[aUUID2]},
-			err:     false,
 		},
 	}
 
 	for _, tt := range test {
 		t.Run(fmt.Sprintf("ApiListFields: %s", tt.name), func(t *testing.T) {
-			var result []testLogicalSwitchPort
+			var result []*testLogicalSwitchPort
 			// Clean object
-			testObj = testLogicalSwitchPort{}
+			testObj := testLogicalSwitchPort{}
+			tt.prepare(&testObj)
 			api := newAPI(tcache, &discardLogger)
 			err := api.Where(&testObj).List(context.Background(), &result)
-			if tt.err {
-				assert.NotNil(t, err)
-			} else {
-				assert.Nil(t, err)
-				assert.ElementsMatchf(t, tt.content, tt.content, "Content should match")
-			}
-
+			assert.Nil(t, err)
+			assert.ElementsMatchf(t, tt.content, result, "Content should match")
 		})
 	}
 
@@ -458,7 +532,7 @@ func TestConditionFromModel(t *testing.T) {
 		t.Run(fmt.Sprintf("conditionFromModel: %s", tt.name), func(t *testing.T) {
 			cache := apiTestCache(t, nil)
 			apiIface := newAPI(cache, &discardLogger)
-			condition := apiIface.(api).conditionFromModel(false, tt.model, tt.conds...)
+			condition := apiIface.(api).conditionFromModel(false, false, tt.model, tt.conds...)
 			if tt.err {
 				assert.IsType(t, &errorConditional{}, condition)
 			} else {
@@ -713,7 +787,7 @@ func TestAPIMutate(t *testing.T) {
 		{
 			name: "select by UUID addElement to set",
 			condition: func(a API) ConditionalAPI {
-				return a.Where(&testLogicalSwitch{
+				return a.Where(&testLogicalSwitchPort{
 					UUID: aUUID0,
 				})
 			},
@@ -735,7 +809,7 @@ func TestAPIMutate(t *testing.T) {
 			err: false,
 		},
 		{
-			name: "select by name delete element from map",
+			name: "select by name delete element from map with cache",
 			condition: func(a API) ConditionalAPI {
 				return a.Where(&testLogicalSwitchPort{
 					Name: "lsp2",
@@ -753,7 +827,31 @@ func TestAPIMutate(t *testing.T) {
 					Op:        ovsdb.OperationMutate,
 					Table:     "Logical_Switch_Port",
 					Mutations: []ovsdb.Mutation{{Column: "external_ids", Mutator: ovsdb.MutateOperationDelete, Value: testOvsSet(t, []string{"foo"})}},
-					Where:     []ovsdb.Condition{{Column: "name", Function: ovsdb.ConditionEqual, Value: "lsp2"}},
+					Where:     []ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: aUUID2}}},
+				},
+			},
+			err: false,
+		},
+		{
+			name: "select by name delete element from map with no cache",
+			condition: func(a API) ConditionalAPI {
+				return a.Where(&testLogicalSwitchPort{
+					Name: "foo",
+				})
+			},
+			mutations: []model.Mutation{
+				{
+					Field:   &testObj.ExternalIds,
+					Mutator: ovsdb.MutateOperationDelete,
+					Value:   []string{"foo"},
+				},
+			},
+			result: []ovsdb.Operation{
+				{
+					Op:        ovsdb.OperationMutate,
+					Table:     "Logical_Switch_Port",
+					Mutations: []ovsdb.Mutation{{Column: "external_ids", Mutator: ovsdb.MutateOperationDelete, Value: testOvsSet(t, []string{"foo"})}},
+					Where:     []ovsdb.Condition{{Column: "name", Function: ovsdb.ConditionEqual, Value: "foo"}},
 				},
 			},
 			err: false,
@@ -901,7 +999,28 @@ func TestAPIUpdate(t *testing.T) {
 			err: false,
 		},
 		{
-			name: "select by index change multiple field",
+			name: "select by index change multiple field with no cache",
+			condition: func(a API) ConditionalAPI {
+				return a.Where(&testLogicalSwitchPort{
+					Name: "foo",
+				})
+			},
+			prepare: func(t *testLogicalSwitchPort) {
+				t.Type = "somethingElse"
+				t.Tag = &six
+			},
+			result: []ovsdb.Operation{
+				{
+					Op:    ovsdb.OperationUpdate,
+					Table: "Logical_Switch_Port",
+					Row:   testRow,
+					Where: []ovsdb.Condition{{Column: "name", Function: ovsdb.ConditionEqual, Value: "foo"}},
+				},
+			},
+			err: false,
+		},
+		{
+			name: "select by index change multiple field with cache",
 			condition: func(a API) ConditionalAPI {
 				return a.Where(&testLogicalSwitchPort{
 					Name: "lsp1",
@@ -916,7 +1035,7 @@ func TestAPIUpdate(t *testing.T) {
 					Op:    ovsdb.OperationUpdate,
 					Table: "Logical_Switch_Port",
 					Row:   testRow,
-					Where: []ovsdb.Condition{{Column: "name", Function: ovsdb.ConditionEqual, Value: "lsp1"}},
+					Where: []ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: aUUID1}}},
 				},
 			},
 			err: false,
@@ -1015,16 +1134,42 @@ func TestAPIUpdate(t *testing.T) {
 			err: false,
 		},
 		{
-			name: "select by field inequality change multiple field",
+			name: "select by field inequality change multiple field with cache",
 			condition: func(a API) ConditionalAPI {
 				t := testLogicalSwitchPort{
-					Type:    "sometype",
+					Type:    "someType",
 					Enabled: &trueVal,
 				}
 				return a.Where(&t, model.Condition{
 					Field:    &t.Type,
 					Function: ovsdb.ConditionNotEqual,
-					Value:    "sometype",
+					Value:    "someType",
+				})
+			},
+			prepare: func(t *testLogicalSwitchPort) {
+				t.Tag = &six
+			},
+			result: []ovsdb.Operation{
+				{
+					Op:    ovsdb.OperationUpdate,
+					Table: "Logical_Switch_Port",
+					Row:   tagRow,
+					Where: []ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: aUUID2}}},
+				},
+			},
+			err: false,
+		},
+		{
+			name: "select by field inequality change multiple field with no cache",
+			condition: func(a API) ConditionalAPI {
+				t := testLogicalSwitchPort{
+					Type:    "sometype",
+					Enabled: &trueVal,
+				}
+				return a.Where(&t, model.Condition{
+					Field:    &t.Tag,
+					Function: ovsdb.ConditionNotEqual,
+					Value:    &one,
 				})
 			},
 			prepare: func(t *testLogicalSwitchPort) {
@@ -1035,7 +1180,7 @@ func TestAPIUpdate(t *testing.T) {
 					Op:    ovsdb.OperationUpdate,
 					Table: "Logical_Switch_Port",
 					Row:   tagRow,
-					Where: []ovsdb.Condition{{Column: "type", Function: ovsdb.ConditionNotEqual, Value: "sometype"}},
+					Where: []ovsdb.Condition{{Column: "tag", Function: ovsdb.ConditionNotEqual, Value: testOvsSet(t, &one)}},
 				},
 			},
 			err: false,
@@ -1140,7 +1285,7 @@ func TestAPIDelete(t *testing.T) {
 			err: false,
 		},
 		{
-			name: "select by index",
+			name: "select by index with cache",
 			condition: func(a API) ConditionalAPI {
 				return a.Where(&testLogicalSwitchPort{
 					Name: "lsp1",
@@ -1150,7 +1295,23 @@ func TestAPIDelete(t *testing.T) {
 				{
 					Op:    ovsdb.OperationDelete,
 					Table: "Logical_Switch_Port",
-					Where: []ovsdb.Condition{{Column: "name", Function: ovsdb.ConditionEqual, Value: "lsp1"}},
+					Where: []ovsdb.Condition{{Column: "_uuid", Function: ovsdb.ConditionEqual, Value: ovsdb.UUID{GoUUID: aUUID1}}},
+				},
+			},
+			err: false,
+		},
+		{
+			name: "select by index with no cache",
+			condition: func(a API) ConditionalAPI {
+				return a.Where(&testLogicalSwitchPort{
+					Name: "foo",
+				})
+			},
+			result: []ovsdb.Operation{
+				{
+					Op:    ovsdb.OperationDelete,
+					Table: "Logical_Switch_Port",
+					Where: []ovsdb.Condition{{Column: "name", Function: ovsdb.ConditionEqual, Value: "foo"}},
 				},
 			},
 			err: false,
@@ -1273,14 +1434,14 @@ func TestAPIDelete(t *testing.T) {
 	}
 }
 
-func BenchmarkAPIListPredicate(b *testing.B) {
+func BenchmarkAPIList(b *testing.B) {
 	const numRows = 10000
 
-	lscacheList := make([]model.Model, 0, numRows)
+	lscacheList := make([]*testLogicalSwitchPort, 0, numRows)
 
 	for i := 0; i < numRows; i++ {
 		lscacheList = append(lscacheList,
-			&testLogicalSwitch{
+			&testLogicalSwitchPort{
 				UUID:        uuid.New().String(),
 				Name:        fmt.Sprintf("ls%d", i),
 				ExternalIds: map[string]string{"foo": "bar"},
@@ -1288,47 +1449,64 @@ func BenchmarkAPIListPredicate(b *testing.B) {
 	}
 	lscache := map[string]model.Model{}
 	for i := range lscacheList {
-		lscache[lscacheList[i].(*testLogicalSwitch).UUID] = lscacheList[i]
+		lscache[lscacheList[i].UUID] = lscacheList[i]
 	}
 	testData := cache.Data{
-		"Logical_Switch": lscache,
+		"Logical_Switch_Port": lscache,
 	}
 	tcache := apiTestCache(b, testData)
 
+	rand.Seed(int64(b.N))
+	var index int
+
 	test := []struct {
 		name      string
 		predicate interface{}
 	}{
 		{
-			name: "none",
-			predicate: func(t *testLogicalSwitch) bool {
+			name: "predicate returns none",
+			predicate: func(t *testLogicalSwitchPort) bool {
 				return false
 			},
 		},
 		{
-			name: "all",
-			predicate: func(t *testLogicalSwitch) bool {
+			name: "predicate returns all",
+			predicate: func(t *testLogicalSwitchPort) bool {
 				return true
 			},
 		},
 		{
-			name: "arbitrary condition",
-			predicate: func(t *testLogicalSwitch) bool {
+			name: "predicate on an arbitrary condition",
+			predicate: func(t *testLogicalSwitchPort) bool {
 				return strings.HasPrefix(t.Name, "ls1")
 			},
 		},
+		{
+			name: "predicate matches name",
+			predicate: func(t *testLogicalSwitchPort) bool {
+				return t.Name == lscacheList[index].Name
+			},
+		},
+		{
+			name: "by index, no predicate",
+		},
 	}
 
 	for _, tt := range test {
-		b.Run(fmt.Sprintf("ApiListPredicate: %s", tt.name), func(b *testing.B) {
-			var result []testLogicalSwitch
-			api := newAPI(tcache, &discardLogger)
-			cond := api.WhereCache(tt.predicate)
-			err := cond.List(context.Background(), &result)
-			if err != nil {
-				b.Fatal(err)
+		b.Run(tt.name, func(b *testing.B) {
+			for i := 0; i < b.N; i++ {
+				index = rand.Intn(numRows)
+				var result []*testLogicalSwitchPort
+				api := newAPI(tcache, &discardLogger)
+				var cond ConditionalAPI
+				if tt.predicate != nil {
+					cond = api.WhereCache(tt.predicate)
+				} else {
+					cond = api.Where(lscacheList[index])
+				}
+				err := cond.List(context.Background(), &result)
+				assert.NoError(b, err)
 			}
-			b.Logf("got %d rows", len(result))
 		})
 	}
 }
diff --git a/client/condition_test.go b/client/condition_test.go
index 1c980a9..6e0be86 100644
--- a/client/condition_test.go
+++ b/client/condition_test.go
@@ -50,8 +50,7 @@ func TestEqualityConditional(t *testing.T) {
 		name      string
 		model     model.Model
 		condition [][]ovsdb.Condition
-		matches   map[model.Model]bool
-		all       bool
+		matches   map[string]model.Model
 		err       bool
 	}{
 		{
@@ -64,61 +63,30 @@ func TestEqualityConditional(t *testing.T) {
 						Function: ovsdb.ConditionEqual,
 						Value:    ovsdb.UUID{GoUUID: aUUID0},
 					}}},
-			matches: map[model.Model]bool{
-				&testLogicalSwitchPort{UUID: aUUID0}:              true,
-				&testLogicalSwitchPort{UUID: aUUID1}:              false,
-				&testLogicalSwitchPort{UUID: aUUID0, Name: "foo"}: true,
-			},
-		},
-		{
-			name:  "by uuid all",
-			model: &testLogicalSwitchPort{UUID: aUUID0, Name: "different"},
-			condition: [][]ovsdb.Condition{
-				{
-					{
-						Column:   "_uuid",
-						Function: ovsdb.ConditionEqual,
-						Value:    ovsdb.UUID{GoUUID: aUUID0},
-					}}},
-			matches: map[model.Model]bool{
-				&testLogicalSwitchPort{UUID: aUUID0}:              true,
-				&testLogicalSwitchPort{UUID: aUUID1}:              false,
-				&testLogicalSwitchPort{UUID: aUUID0, Name: "foo"}: true,
-			},
-			all: true,
+			matches: map[string]model.Model{aUUID0: lspcacheList[0]},
 		},
 		{
-			name:  "by index",
+			name:  "by index with cache",
 			model: &testLogicalSwitchPort{Name: "lsp1"},
 			condition: [][]ovsdb.Condition{
 				{
 					{
-						Column:   "name",
+						Column:   "_uuid",
 						Function: ovsdb.ConditionEqual,
-						Value:    "lsp1",
+						Value:    ovsdb.UUID{GoUUID: aUUID1},
 					}}},
-			matches: map[model.Model]bool{
-				&testLogicalSwitchPort{UUID: aUUID1}:               false,
-				&testLogicalSwitchPort{UUID: aUUID1, Name: "lsp1"}: true,
-				&testLogicalSwitchPort{UUID: aUUID0, Name: "lsp1"}: true,
-			},
+			matches: map[string]model.Model{aUUID1: lspcacheList[1]},
 		},
 		{
-			name:  "by index all",
-			model: &testLogicalSwitchPort{Name: "lsp1"},
+			name:  "by index with no cache",
+			model: &testLogicalSwitchPort{Name: "foo"},
 			condition: [][]ovsdb.Condition{
 				{
 					{
 						Column:   "name",
 						Function: ovsdb.ConditionEqual,
-						Value:    "lsp1",
+						Value:    "foo",
 					}}},
-			matches: map[model.Model]bool{
-				&testLogicalSwitchPort{UUID: aUUID1}:               false,
-				&testLogicalSwitchPort{UUID: aUUID1, Name: "lsp1"}: true,
-				&testLogicalSwitchPort{UUID: aUUID0, Name: "lsp1"}: true,
-			},
-			all: true,
 		},
 		{
 			name:  "by non index",
@@ -128,17 +96,11 @@ func TestEqualityConditional(t *testing.T) {
 	}
 	for _, tt := range test {
 		t.Run(fmt.Sprintf("Equality Conditional: %s", tt.name), func(t *testing.T) {
-			cond, err := newEqualityConditional(tcache.DatabaseModel(), "Logical_Switch_Port", tt.all, tt.model)
+			cond, err := newEqualityConditional("Logical_Switch_Port", tcache, tt.model)
 			assert.Nil(t, err)
-			for model, shouldMatch := range tt.matches {
-				matches, err := cond.Matches(model)
-				if tt.err {
-					assert.NotNil(t, err)
-				} else {
-					assert.Nil(t, err)
-					assert.Equalf(t, shouldMatch, matches, fmt.Sprintf("Match on model %#+v should be %v", model, shouldMatch))
-				}
-			}
+			matches, err := cond.Matches()
+			assert.Nil(t, err)
+			assert.Equal(t, tt.matches, matches)
 			generated, err := cond.Generate()
 			if tt.err {
 				assert.NotNil(t, err)
@@ -190,7 +152,7 @@ func TestPredicateConditional(t *testing.T) {
 		name      string
 		predicate interface{}
 		condition [][]ovsdb.Condition
-		matches   map[model.Model]bool
+		matches   map[string]model.Model
 		err       bool
 	}{
 		{
@@ -205,11 +167,7 @@ func TestPredicateConditional(t *testing.T) {
 						Function: ovsdb.ConditionEqual,
 						Value:    ovsdb.UUID{GoUUID: aUUID0},
 					}}},
-			matches: map[model.Model]bool{
-				&testLogicalSwitchPort{UUID: aUUID0}:              true,
-				&testLogicalSwitchPort{UUID: aUUID1}:              false,
-				&testLogicalSwitchPort{UUID: aUUID0, Name: "foo"}: true,
-			},
+			matches: map[string]model.Model{aUUID0: lspcacheList[0]},
 		},
 		{
 			name: "by random field",
@@ -229,9 +187,9 @@ func TestPredicateConditional(t *testing.T) {
 						Function: ovsdb.ConditionEqual,
 						Value:    ovsdb.UUID{GoUUID: aUUID2},
 					}}},
-			matches: map[model.Model]bool{
-				&testLogicalSwitchPort{UUID: aUUID1, Enabled: &trueVal}:  false,
-				&testLogicalSwitchPort{UUID: aUUID1, Enabled: &falseVal}: true,
+			matches: map[string]model.Model{
+				aUUID1: lspcacheList[1],
+				aUUID2: lspcacheList[2],
 			},
 		},
 	}
@@ -239,15 +197,9 @@ func TestPredicateConditional(t *testing.T) {
 		t.Run(fmt.Sprintf("Predicate Conditional: %s", tt.name), func(t *testing.T) {
 			cond, err := newPredicateConditional("Logical_Switch_Port", tcache, tt.predicate)
 			assert.Nil(t, err)
-			for model, shouldMatch := range tt.matches {
-				matches, err := cond.Matches(model)
-				if tt.err {
-					assert.NotNil(t, err)
-				} else {
-					assert.Nil(t, err)
-					assert.Equalf(t, shouldMatch, matches, fmt.Sprintf("Match on model %#+v should be %v", model, shouldMatch))
-				}
-			}
+			matches, err := cond.Matches()
+			assert.Nil(t, err)
+			assert.Equal(t, tt.matches, matches)
 			generated, err := cond.Generate()
 			if tt.err {
 				assert.NotNil(t, err)
@@ -259,37 +211,8 @@ func TestPredicateConditional(t *testing.T) {
 	}
 }
 
-func TestExplicitConditional(t *testing.T) {
-	lspcacheList := []model.Model{
-		&testLogicalSwitchPort{
-			UUID:        aUUID0,
-			Name:        "lsp0",
-			ExternalIds: map[string]string{"foo": "bar"},
-			Enabled:     &trueVal,
-		},
-		&testLogicalSwitchPort{
-			UUID:        aUUID1,
-			Name:        "lsp1",
-			ExternalIds: map[string]string{"foo": "baz"},
-			Enabled:     &falseVal,
-		},
-		&testLogicalSwitchPort{
-			UUID:        aUUID2,
-			Name:        "lsp2",
-			ExternalIds: map[string]string{"unique": "id"},
-			Enabled:     &falseVal,
-		},
-		&testLogicalSwitchPort{
-			UUID:        aUUID3,
-			Name:        "lsp3",
-			ExternalIds: map[string]string{"foo": "baz"},
-			Enabled:     &trueVal,
-		},
-	}
+func TestExplicitConditionalWithNoCache(t *testing.T) {
 	lspcache := map[string]model.Model{}
-	for i := range lspcacheList {
-		lspcache[lspcacheList[i].(*testLogicalSwitchPort).UUID] = lspcacheList[i]
-	}
 	testData := cache.Data{
 		"Logical_Switch_Port": lspcache,
 	}
@@ -430,11 +353,266 @@ func TestExplicitConditional(t *testing.T) {
 		},
 	}
 	for _, tt := range test {
-		t.Run(fmt.Sprintf("Explicit Conditional: %s", tt.name), func(t *testing.T) {
-			cond, err := newExplicitConditional(tcache.DatabaseModel(), "Logical_Switch_Port", tt.all, testObj, tt.args...)
+		t.Run(fmt.Sprintf("Explicit Conditional with no cache: %s", tt.name), func(t *testing.T) {
+			cond, err := newExplicitConditional("Logical_Switch_Port", tcache, tt.all, testObj, tt.args...)
+			assert.Nil(t, err)
+			generated, err := cond.Generate()
+			if tt.err {
+				assert.NotNil(t, err)
+			} else {
+				assert.Nil(t, err)
+				assert.ElementsMatch(t, tt.result, generated)
+			}
+		})
+	}
+}
+
+func TestExplicitConditionalWithCache(t *testing.T) {
+	lspcacheList := []model.Model{
+		&testLogicalSwitchPort{
+			UUID:        aUUID0,
+			Name:        "lsp0",
+			ExternalIds: map[string]string{"foo": "bar"},
+			Enabled:     &trueVal,
+		},
+		&testLogicalSwitchPort{
+			UUID:        aUUID1,
+			Name:        "lsp1",
+			ExternalIds: map[string]string{"foo": "baz"},
+			Enabled:     &falseVal,
+		},
+		&testLogicalSwitchPort{
+			UUID:        aUUID2,
+			Name:        "lsp2",
+			ExternalIds: map[string]string{"unique": "id"},
+			Enabled:     &falseVal,
+		},
+		&testLogicalSwitchPort{
+			UUID:        aUUID3,
+			Name:        "lsp3",
+			ExternalIds: map[string]string{"foo": "baz"},
+			Enabled:     &trueVal,
+		},
+	}
+	lspcache := map[string]model.Model{}
+	for i := range lspcacheList {
+		lspcache[lspcacheList[i].(*testLogicalSwitchPort).UUID] = lspcacheList[i]
+	}
+	testData := cache.Data{
+		"Logical_Switch_Port": lspcache,
+	}
+	tcache := apiTestCache(t, testData)
+
+	testObj := &testLogicalSwitchPort{}
+
+	test := []struct {
+		name   string
+		args   []model.Condition
+		result [][]ovsdb.Condition
+		all    bool
+		err    bool
+	}{
+		{
+			name: "inequality comparison",
+			args: []model.Condition{
+				{
+					Field:    &testObj.Name,
+					Function: ovsdb.ConditionNotEqual,
+					Value:    "lsp0",
+				},
+			},
+			result: [][]ovsdb.Condition{
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID1},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID2},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID3},
+					},
+				},
+			},
+		},
+		{
+			name: "inequality comparison all",
+			args: []model.Condition{
+				{
+					Field:    &testObj.Name,
+					Function: ovsdb.ConditionNotEqual,
+					Value:    "lsp0",
+				},
+			},
+			result: [][]ovsdb.Condition{
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID1},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID2},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID3},
+					},
+				},
+			},
+			all: true,
+		},
+		{
+			name: "map comparison",
+			args: []model.Condition{
+				{
+					Field:    &testObj.ExternalIds,
+					Function: ovsdb.ConditionIncludes,
+					Value:    map[string]string{"foo": "baz"},
+				},
+			},
+			result: [][]ovsdb.Condition{
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID1},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID3},
+					},
+				},
+			},
+		},
+		{
+			name: "set comparison",
+			args: []model.Condition{
+				{
+					Field:    &testObj.Enabled,
+					Function: ovsdb.ConditionEqual,
+					Value:    &trueVal,
+				},
+			},
+			result: [][]ovsdb.Condition{
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID0},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID3},
+					},
+				},
+			},
+		},
+		{
+			name: "multiple conditions",
+			args: []model.Condition{
+				{
+					Field:    &testObj.Enabled,
+					Function: ovsdb.ConditionEqual,
+					Value:    &trueVal,
+				},
+				{
+					Field:    &testObj.Name,
+					Function: ovsdb.ConditionNotEqual,
+					Value:    "foo",
+				},
+			},
+			result: [][]ovsdb.Condition{
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID0},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID1},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID2},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID3},
+					},
+				},
+			},
+		},
+		{
+			name: "multiple conditions all",
+			args: []model.Condition{
+				{
+					Field:    &testObj.Enabled,
+					Function: ovsdb.ConditionEqual,
+					Value:    &trueVal,
+				},
+				{
+					Field:    &testObj.Name,
+					Function: ovsdb.ConditionNotEqual,
+					Value:    "foo",
+				},
+			},
+			result: [][]ovsdb.Condition{
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID0},
+					},
+				},
+				{
+					{
+						Column:   "_uuid",
+						Function: ovsdb.ConditionEqual,
+						Value:    ovsdb.UUID{GoUUID: aUUID3},
+					},
+				},
+			},
+			all: true,
+		},
+	}
+	for _, tt := range test {
+		t.Run(fmt.Sprintf("Explicit Conditional with cache: %s", tt.name), func(t *testing.T) {
+			cond, err := newExplicitConditional("Logical_Switch_Port", tcache, tt.all, testObj, tt.args...)
 			assert.Nil(t, err)
-			_, err = cond.Matches(testObj)
-			assert.NotNilf(t, err, "Explicit conditions should fail to match on cache")
 			generated, err := cond.Generate()
 			if tt.err {
 				assert.NotNil(t, err)