sql/index/pilosa: return all index ids in lookups

sql/index/pilosa/index.go

@@ -108,6 +108,9 @@ func (idx *pilosaIndex) Get(keys ...interface{}) (sql.IndexLookup, error) {
 		mapping:     idx.mapping,
 		keys:        keys,
 		expressions: idx.expressions,
+		indexes: map[string]struct{}{
+			idx.ID(): struct{}{},
+		},
 	}, nil
 }
 
@@ -165,6 +168,9 @@ func (idx *pilosaIndex) AscendGreaterOrEqual(keys ...interface{}) (sql.IndexLook
 		mapping:     idx.mapping,
 		keys:        keys,
 		expressions: idx.expressions,
+		indexes: map[string]struct{}{
+			idx.ID(): struct{}{},
+		},
 	}, keys, nil), nil
 }
 
@@ -179,6 +185,9 @@ func (idx *pilosaIndex) AscendLessThan(keys ...interface{}) (sql.IndexLookup, er
 		mapping:     idx.mapping,
 		keys:        keys,
 		expressions: idx.expressions,
+		indexes: map[string]struct{}{
+			idx.ID(): struct{}{},
+		},
 	}, nil, keys), nil
 }
 
@@ -196,6 +205,9 @@ func (idx *pilosaIndex) AscendRange(greaterOrEqual, lessThan []interface{}) (sql
 		index:       idx.index,
 		mapping:     idx.mapping,
 		expressions: idx.expressions,
+		indexes: map[string]struct{}{
+			idx.ID(): struct{}{},
+		},
 	}, greaterOrEqual, lessThan), nil
 }
 
@@ -211,6 +223,9 @@ func (idx *pilosaIndex) DescendGreater(keys ...interface{}) (sql.IndexLookup, er
 		keys:        keys,
 		expressions: idx.expressions,
 		reverse:     true,
+		indexes: map[string]struct{}{
+			idx.ID(): struct{}{},
+		},
 	}, keys, nil), nil
 }
 
@@ -226,6 +241,9 @@ func (idx *pilosaIndex) DescendLessOrEqual(keys ...interface{}) (sql.IndexLookup
 		keys:        keys,
 		expressions: idx.expressions,
 		reverse:     true,
+		indexes: map[string]struct{}{
+			idx.ID(): struct{}{},
+		},
 	}, nil, keys), nil
 }
 
@@ -244,6 +262,9 @@ func (idx *pilosaIndex) DescendRange(lessOrEqual, greaterThan []interface{}) (sq
 		mapping:     idx.mapping,
 		expressions: idx.expressions,
 		reverse:     true,
+		indexes: map[string]struct{}{
+			idx.ID(): struct{}{},
+		},
 	}, greaterThan, lessOrEqual), nil
 }
 
@@ -258,6 +279,9 @@ func (idx *pilosaIndex) Not(keys ...interface{}) (sql.IndexLookup, error) {
 		mapping:     idx.mapping,
 		keys:        keys,
 		expressions: idx.expressions,
+		indexes: map[string]struct{}{
+			idx.ID(): struct{}{},
+		},
 	}, nil
 }
 

sql/index/pilosa/lookup.go

@@ -4,6 +4,7 @@ import (
 	"bytes"
 	"encoding/gob"
 	"io"
+	"sort"
 	"strings"
 	"time"
 
@@ -63,6 +64,7 @@ type (
 		keys        []interface{}
 		expressions []string
 		operations  []*lookupOperation
+		indexes     map[string]struct{}
 	}
 
 	lookupOperation struct {
@@ -165,7 +167,7 @@ func (l *indexLookup) Values(p sql.Partition) (sql.IndexValueIter, error) {
 }
 
 func (l *indexLookup) Indexes() []string {
-	return []string{l.id}
+	return sortedIndexes(l.indexes)
 }
 
 // IsMergeable implements sql.Mergeable interface.
@@ -181,6 +183,9 @@ func (l *indexLookup) IsMergeable(lookup sql.IndexLookup) bool {
 func (l *indexLookup) Intersection(lookups ...sql.IndexLookup) sql.IndexLookup {
 	lookup := *l
 	for _, li := range lookups {
+		for _, idx := range li.Indexes() {
+			lookup.indexes[idx] = struct{}{}
+		}
 		lookup.operations = append(lookup.operations, &lookupOperation{li, intersect})
 	}
 
@@ -191,6 +196,9 @@ func (l *indexLookup) Intersection(lookups ...sql.IndexLookup) sql.IndexLookup {
 func (l *indexLookup) Union(lookups ...sql.IndexLookup) sql.IndexLookup {
 	lookup := *l
 	for _, li := range lookups {
+		for _, idx := range li.Indexes() {
+			lookup.indexes[idx] = struct{}{}
+		}
 		lookup.operations = append(lookup.operations, &lookupOperation{li, union})
 	}
 
@@ -201,6 +209,9 @@ func (l *indexLookup) Union(lookups ...sql.IndexLookup) sql.IndexLookup {
 func (l *indexLookup) Difference(lookups ...sql.IndexLookup) sql.IndexLookup {
 	lookup := *l
 	for _, li := range lookups {
+		for _, idx := range li.Indexes() {
+			lookup.indexes[idx] = struct{}{}
+		}
 		lookup.operations = append(lookup.operations, &lookupOperation{li, difference})
 	}
 
@@ -214,6 +225,7 @@ type filteredLookup struct {
 	keys        []interface{}
 	expressions []string
 	operations  []*lookupOperation
+	indexes     map[string]struct{}
 
 	reverse bool
 	filter  func(int, []byte) (bool, error)
@@ -320,7 +332,7 @@ func (l *filteredLookup) Values(p sql.Partition) (sql.IndexValueIter, error) {
 }
 
 func (l *filteredLookup) Indexes() []string {
-	return []string{l.id}
+	return sortedIndexes(l.indexes)
 }
 
 // IsMergeable implements sql.Mergeable interface.
@@ -335,6 +347,9 @@ func (l *filteredLookup) IsMergeable(lookup sql.IndexLookup) bool {
 func (l *filteredLookup) Intersection(lookups ...sql.IndexLookup) sql.IndexLookup {
 	lookup := *l
 	for _, li := range lookups {
+		for _, idx := range li.Indexes() {
+			lookup.indexes[idx] = struct{}{}
+		}
 		lookup.operations = append(lookup.operations, &lookupOperation{li, intersect})
 	}
 
@@ -345,6 +360,9 @@ func (l *filteredLookup) Intersection(lookups ...sql.IndexLookup) sql.IndexLooku
 func (l *filteredLookup) Union(lookups ...sql.IndexLookup) sql.IndexLookup {
 	lookup := *l
 	for _, li := range lookups {
+		for _, idx := range li.Indexes() {
+			lookup.indexes[idx] = struct{}{}
+		}
 		lookup.operations = append(lookup.operations, &lookupOperation{li, union})
 	}
 
@@ -355,6 +373,9 @@ func (l *filteredLookup) Union(lookups ...sql.IndexLookup) sql.IndexLookup {
 func (l *filteredLookup) Difference(lookups ...sql.IndexLookup) sql.IndexLookup {
 	lookup := *l
 	for _, li := range lookups {
+		for _, idx := range li.Indexes() {
+			lookup.indexes[idx] = struct{}{}
+		}
 		lookup.operations = append(lookup.operations, &lookupOperation{li, difference})
 	}
 
@@ -379,6 +400,7 @@ type negateLookup struct {
 	mapping     *mapping
 	keys        []interface{}
 	expressions []string
+	indexes     map[string]struct{}
 	operations  []*lookupOperation
 }
 
@@ -491,7 +513,7 @@ func (l *negateLookup) Values(p sql.Partition) (sql.IndexValueIter, error) {
 }
 
 func (l *negateLookup) Indexes() []string {
-	return []string{l.id}
+	return sortedIndexes(l.indexes)
 }
 
 // IsMergeable implements sql.Mergeable interface.
@@ -507,6 +529,9 @@ func (l *negateLookup) IsMergeable(lookup sql.IndexLookup) bool {
 func (l *negateLookup) Intersection(lookups ...sql.IndexLookup) sql.IndexLookup {
 	lookup := *l
 	for _, li := range lookups {
+		for _, idx := range li.Indexes() {
+			lookup.indexes[idx] = struct{}{}
+		}
 		lookup.operations = append(lookup.operations, &lookupOperation{li, intersect})
 	}
 
@@ -517,6 +542,9 @@ func (l *negateLookup) Intersection(lookups ...sql.IndexLookup) sql.IndexLookup
 func (l *negateLookup) Union(lookups ...sql.IndexLookup) sql.IndexLookup {
 	lookup := *l
 	for _, li := range lookups {
+		for _, idx := range li.Indexes() {
+			lookup.indexes[idx] = struct{}{}
+		}
 		lookup.operations = append(lookup.operations, &lookupOperation{li, union})
 	}
 
@@ -527,6 +555,9 @@ func (l *negateLookup) Union(lookups ...sql.IndexLookup) sql.IndexLookup {
 func (l *negateLookup) Difference(lookups ...sql.IndexLookup) sql.IndexLookup {
 	lookup := *l
 	for _, li := range lookups {
+		for _, idx := range li.Indexes() {
+			lookup.indexes[idx] = struct{}{}
+		}
 		lookup.operations = append(lookup.operations, &lookupOperation{li, difference})
 	}
 
@@ -596,7 +627,7 @@ func compare(a, b interface{}) (int, error) {
 			return 0, nil
 		}
 
-		if !a {
+		if a == false {
 			return -1, nil
 		}
 
@@ -734,3 +765,13 @@ func compare(a, b interface{}) (int, error) {
 		return 0, errUnknownType.New(a)
 	}
 }
+
+func sortedIndexes(indexes map[string]struct{}) []string {
+	var result = make([]string, 0, len(indexes))
+	for idx := range indexes {
+		result = append(result, idx)
+	}
+
+	sort.Strings(result)
+	return result
+}

sql/index/pilosa/lookup_test.go

@@ -204,10 +204,20 @@ func TestMergeable(t *testing.T) {
 
 func TestIndexes(t *testing.T) {
 	testCases := []sql.IndexLookup{
-		&indexLookup{id: "foo"},
-		&negateLookup{id: "foo"},
-		&ascendLookup{filteredLookup: &filteredLookup{id: "foo"}},
-		&descendLookup{filteredLookup: &filteredLookup{id: "foo"}},
+		&indexLookup{id: "foo", indexes: map[string]struct{}{"foo": struct{}{}}},
+		&negateLookup{id: "foo", indexes: map[string]struct{}{"foo": struct{}{}}},
+		&ascendLookup{
+			filteredLookup: &filteredLookup{
+				id:      "foo",
+				indexes: map[string]struct{}{"foo": struct{}{}},
+			},
+		},
+		&descendLookup{
+			filteredLookup: &filteredLookup{
+				id:      "foo",
+				indexes: map[string]struct{}{"foo": struct{}{}},
+			},
+		},
 	}
 
 	for _, tt := range testCases {
@@ -216,3 +226,50 @@ func TestIndexes(t *testing.T) {
 		})
 	}
 }
+
+func TestLookupIndexes(t *testing.T) {
+	require := require.New(t)
+
+	lookups := []sql.IndexLookup{
+		&indexLookup{
+			id:      "1",
+			indexes: map[string]struct{}{"1": struct{}{}},
+		},
+		&negateLookup{
+			id:      "2",
+			indexes: map[string]struct{}{"2": struct{}{}},
+		},
+		&ascendLookup{filteredLookup: &filteredLookup{
+			id:      "3",
+			indexes: map[string]struct{}{"3": struct{}{}},
+		}},
+		&descendLookup{filteredLookup: &filteredLookup{
+			id:      "4",
+			indexes: map[string]struct{}{"4": struct{}{}},
+		}},
+		&filteredLookup{
+			id:      "5",
+			indexes: map[string]struct{}{"5": struct{}{}},
+		},
+	}
+
+	expected := []string{"1", "2", "3", "4", "5"}
+
+	// All possible permutations of operations between all the different kinds
+	// of lookups are tested.
+	for i := 0; i < len(lookups); i++ {
+		var op sql.SetOperations
+		var others []sql.IndexLookup
+		for j := 0; j < len(lookups); j++ {
+			if i == j {
+				op = lookups[i].(sql.SetOperations)
+			} else {
+				others = append(others, lookups[j])
+			}
+		}
+
+		require.Equal(expected, op.Union(others...).Indexes())
+		require.Equal(expected, op.Difference(others...).Indexes())
+		require.Equal(expected, op.Intersection(others...).Indexes())
+	}
+}