Modifications to support byte array indexing and transactions for KV stores

Author: Luis Fung

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/containerd/continuity v0.0.0-20181203112020-004b46473808 // indirect
 	github.com/cznic/mathutil v0.0.0-20170313102836-1447ad269d64
 	github.com/d4l3k/messagediff v1.2.1 // indirect
+	github.com/deckarep/golang-set v1.7.1
 	github.com/dennwc/graphql v0.0.0-20180603144102-12cfed44bc5d
 	github.com/dgraph-io/badger v1.5.4
 	github.com/dgryski/go-farm v0.0.0-20190104051053-3adb47b1fb0f // indirect
@@ -20,6 +21,7 @@ require (
 	github.com/docker/go-connections v0.4.0 // indirect
 	github.com/docker/go-units v0.3.3 // indirect
 	github.com/dop251/goja v0.0.0-20190105122144-6d5bf35058fa
+	github.com/emirpasic/gods v1.12.0
 	github.com/flimzy/diff v0.1.4 // indirect
 	github.com/flimzy/kivik v1.8.1 // indirect
 	github.com/flimzy/testy v0.0.13 // indirect

graph/kv/indexing.go

@@ -25,11 +25,11 @@ import (
 
 	"github.com/cayleygraph/cayley/clog"
 	"github.com/cayleygraph/cayley/graph"
-	"github.com/cayleygraph/cayley/graph/log"
+	graphlog "github.com/cayleygraph/cayley/graph/log"
 	"github.com/cayleygraph/cayley/graph/proto"
 	"github.com/cayleygraph/cayley/quad"
 	"github.com/cayleygraph/cayley/quad/pquads"
-	"github.com/tylertreat/BoomFilters"
+	boom "github.com/tylertreat/BoomFilters"
 )
 
 var (
@@ -829,6 +829,9 @@ func (qs *QuadStore) resolveQuadValues(ctx context.Context, tx BucketTx, vals []
 		} else if v == nil {
 			continue
 		}
+		if bytes, ok := v.(quad.Bytes); ok {
+			v = bytes.TypedString()
+		}
 		inds = append(inds, i)
 		keys = append(keys, bucketKeyForVal(v))
 	}

graph/transaction.go

@@ -14,14 +14,17 @@
 
 package graph
 
-import "github.com/cayleygraph/cayley/quad"
+import (
+	"github.com/cayleygraph/cayley/internal/mapset"
+	"github.com/cayleygraph/cayley/quad"
+)
 
 // Transaction stores a bunch of Deltas to apply together in an atomic step on the database.
 type Transaction struct {
 	// Deltas stores the deltas in the right order
 	Deltas []Delta
 	// deltas stores the deltas in a map to avoid duplications
-	deltas map[Delta]struct{}
+	deltas mapset.Map
 }
 
 // NewTransaction initialize a new transaction.
@@ -31,7 +34,7 @@ func NewTransaction() *Transaction {
 
 // NewTransactionN initialize a new transaction with a predefined capacity.
 func NewTransactionN(n int) *Transaction {
-	return &Transaction{Deltas: make([]Delta, 0, n), deltas: make(map[Delta]struct{}, n)}
+	return &Transaction{Deltas: make([]Delta, 0, n), deltas: mapset.NewMapWithComparator(mapset.GenericComparator)}
 }
 
 // AddQuad adds a new quad to the transaction if it is not already present in it.
@@ -40,8 +43,8 @@ func NewTransactionN(n int) *Transaction {
 func (t *Transaction) AddQuad(q quad.Quad) {
 	ad, rd := createDeltas(q)
 
-	if _, adExists := t.deltas[ad]; !adExists {
-		if _, rdExists := t.deltas[rd]; rdExists {
+	if !t.deltas.Contains(ad) {
+		if t.deltas.Contains(rd) {
 			t.deleteDelta(rd)
 		} else {
 			t.addDelta(ad)
@@ -55,10 +58,10 @@ func (t *Transaction) AddQuad(q quad.Quad) {
 func (t *Transaction) RemoveQuad(q quad.Quad) {
 	ad, rd := createDeltas(q)
 
-	if _, adExists := t.deltas[ad]; adExists {
+	if t.deltas.Contains(ad) {
 		t.deleteDelta(ad)
 	} else {
-		if _, rdExists := t.deltas[rd]; !rdExists {
+		if !t.deltas.Contains(rd) {
 			t.addDelta(rd)
 		}
 	}
@@ -78,11 +81,11 @@ func createDeltas(q quad.Quad) (ad, rd Delta) {
 
 func (t *Transaction) addDelta(d Delta) {
 	t.Deltas = append(t.Deltas, d)
-	t.deltas[d] = struct{}{}
+	t.deltas.Put(d, struct{}{})
 }
 
 func (t *Transaction) deleteDelta(d Delta) {
-	delete(t.deltas, d)
+	t.deltas.Remove(d)
 
 	for i, id := range t.Deltas {
 		if id == d {

internal/mapset/comparator.go

@@ -0,0 +1,57 @@
+package mapset
+
+import (
+	"time"
+
+	"github.com/emirpasic/gods/utils"
+)
+
+var (
+	ByteComparator   = utils.ByteComparator
+	StringComparator = utils.StringComparator
+	RuneComparator   = utils.RuneComparator
+	TimeComparator   = utils.TimeComparator
+
+	IntComparator   = utils.IntComparator
+	Int8Comparator  = utils.Int8Comparator
+	Int16Comparator = utils.Int16Comparator
+	Int32Comparator = utils.Int32Comparator
+	Int64Comparator = utils.Int64Comparator
+
+	UIntComparator   = utils.UIntComparator
+	UInt8Comparator  = utils.UInt8Comparator
+	UInt16Comparator = utils.UInt16Comparator
+	UInt32Comparator = utils.UInt32Comparator
+	UInt64Comparator = utils.UInt64Comparator
+
+	Float32Comparator = utils.Float32Comparator
+	Float64Comparator = utils.Float64Comparator
+
+	GenericComparator = genericComparator
+)
+
+func genericComparator(a, b interface{}) int {
+	switch a.(type) {
+	case string:
+		return StringComparator(a, b)
+	case rune:
+		return RuneComparator(a, b)
+	case []byte:
+		return ByteComparator(a, b)
+	case int:
+		return IntComparator(a, b)
+	case uint:
+		return UIntComparator(a, b)
+	case float32:
+		return Float32Comparator(a, b)
+	case float64:
+		return Float64Comparator(a, b)
+	case time.Time:
+		return TimeComparator(a, b)
+	}
+	if a == b {
+		return 0
+	}
+	return -2
+	// panic(fmt.Sprintf("unknonw comparative type: %#v, %#v", a, b))
+}

internal/mapset/map.go

@@ -0,0 +1,93 @@
+package mapset
+
+import (
+	"github.com/emirpasic/gods/trees/btree"
+	"github.com/emirpasic/gods/utils"
+)
+
+type Map interface {
+	Put(k, v interface{}) bool
+
+	Get(k interface{}) (interface{}, bool)
+
+	Remove(k interface{})
+
+	Contains(k ...interface{}) bool
+
+	Each(func(k, v interface{}) int)
+
+	Keys() []interface{}
+
+	Values() []interface{}
+
+	Size() int
+
+	Clear()
+
+	String() string
+}
+
+func NewBytesMap() Map {
+	return NewMapWithComparator(utils.ByteComparator)
+}
+
+func NewMapWithComparator(cmp func(a, b interface{}) int) Map {
+	m := &btreeMap{
+		inner: btree.NewWith(10, cmp),
+	}
+	return m
+}
+
+type btreeMap struct {
+	inner *btree.Tree
+	cmp   func(a, b interface{}) int
+}
+
+func (m btreeMap) Each(f func(k, v interface{}) int) {
+	it := m.inner.Iterator()
+	for it.Next() {
+		f(it.Key(), it.Value())
+	}
+}
+
+func (m *btreeMap) Put(k, v interface{}) bool {
+	m.inner.Put(k, v)
+	return true
+}
+
+func (m btreeMap) Get(k interface{}) (interface{}, bool) {
+	return m.inner.Get(k)
+}
+
+func (m *btreeMap) Remove(k interface{}) {
+	m.inner.Remove(k)
+}
+
+func (b *btreeMap) Contains(k ...interface{}) bool {
+	for _, n := range k {
+		if _, exists := b.inner.Get(n); exists {
+			return true
+		}
+	}
+	return false
+}
+
+func (m *btreeMap) Clear() {
+	m.inner.Clear()
+}
+
+func (m btreeMap) Size() int {
+	return m.inner.Size()
+}
+
+func (m btreeMap) Keys() []interface{} {
+	return m.inner.Keys()
+}
+
+func (m btreeMap) Values() []interface{} {
+	return m.inner.Values()
+}
+
+func (m btreeMap) String() string {
+	return m.inner.String()
+}

internal/mapset/map_test.go

@@ -0,0 +1,63 @@
+package mapset
+
+import (
+	"testing"
+)
+
+func TestMap(t *testing.T) {
+	tree := NewMapWithComparator(func(a, b interface{}) int {
+		aa := a.(int)
+		bb := b.(int)
+		switch {
+		case aa > bb:
+			return 1
+		case aa < bb:
+			return -1
+		default:
+			return 0
+		}
+	})
+	tree.Put(1, "a")
+	tree.Put(2, "b")
+	tree.Put(3, "c")
+	tree.Put(4, "d")
+	tree.Put(5, "e")
+	tree.Put(6, "f")
+	tree.Put(7, "g")
+
+	tests := [][]interface{}{
+		{0, nil, false},
+		{1, "a", true},
+		{2, "b", true},
+		{3, "c", true},
+		{4, "d", true},
+		{5, "e", true},
+		{6, "f", true},
+		{7, "g", true},
+		{8, nil, false},
+	}
+
+	// Test values
+	for _, test := range tests {
+		if value, found := tree.Get(test[0]); value != test[1] || found != test[2] {
+			t.Errorf("Got %v,%v expected %v,%v", value, found, test[1], test[2])
+		}
+	}
+
+	// Test updates
+	sz := tree.Size()
+	tree.Put(7, "g")
+	tree.Put(7, "this doesn't matter either...")
+	if sz != tree.Size() {
+		t.Errorf("Got %v expected %v", tree.Size(), sz)
+	}
+
+	// Test contains
+	if !tree.Contains(7) {
+		t.Errorf("Got false, expected true")
+	}
+
+	if tree.Contains(10) {
+		t.Errorf("Got true, expected false")
+	}
+}