barak/cayley
1
0
Fork 0
Code Issues Pull requests Releases Wiki Activity

Merge branch 'master' of https://github.com/google/cayley into operators

Browse source 
Conflicts:
	query/gremlin/build_iterator.go
This commit is contained in:
Matei Chiperi 2014-09-08 17:04:48 -07:00
commit a718130f4a
83 changed files with 2370 additions and 2659 deletions
Download patch file Download diff file Expand all files Collapse all files

2
graph/bolt/all_iterator.go
View file

@ -103,7 +103,7 @@ func (it *AllIterator) Next() bool {
} else {
k, _ := cur.Seek(last)
if !bytes.Equal(k, last) {
return fmt.Errorf("Couldn't pick up after", k)
return fmt.Errorf("could not pick up after %v", k)
}
}
for i < bufferSize {

28
graph/bolt/iterator.go
View file

@ -32,7 +32,7 @@ import (
var (
boltType graph.Type
bufferSize = 50
errNotExist = errors.New("Quad does not exist")
errNotExist = errors.New("quad does not exist")
)
func init() {
@ -43,7 +43,7 @@ type Iterator struct {
uid uint64
tags graph.Tagger
bucket []byte
checkId []byte
checkID []byte
dir quad.Direction
qs *QuadStore
result *Token
@ -56,7 +56,7 @@ type Iterator struct {
func NewIterator(bucket []byte, d quad.Direction, value graph.Value, qs *QuadStore) *Iterator {
tok := value.(*Token)
if !bytes.Equal(tok.bucket, nodeBucket) {
glog.Error("Creating an iterator from a non-node value.")
glog.Error("creating an iterator from a non-node value")
return &Iterator{done: true}
}
@ -68,8 +68,8 @@ func NewIterator(bucket []byte, d quad.Direction, value graph.Value, qs *QuadSto
size: qs.SizeOf(value),
}
it.checkId = make([]byte, len(tok.key))
copy(it.checkId, tok.key)
it.checkID = make([]byte, len(tok.key))
copy(it.checkID, tok.key)
return &it
}
@ -101,7 +101,7 @@ func (it *Iterator) TagResults(dst map[string]graph.Value) {
}
func (it *Iterator) Clone() graph.Iterator {
out := NewIterator(it.bucket, it.dir, &Token{nodeBucket, it.checkId}, it.qs)
out := NewIterator(it.bucket, it.dir, &Token{nodeBucket, it.checkID}, it.qs)
out.Tagger().CopyFrom(it)
return out
}
@ -134,8 +134,8 @@ func (it *Iterator) Next() bool {
b := tx.Bucket(it.bucket)
cur := b.Cursor()
if last == nil {
k, _ := cur.Seek(it.checkId)
if bytes.HasPrefix(k, it.checkId) {
k, _ := cur.Seek(it.checkID)
if bytes.HasPrefix(k, it.checkID) {
var out []byte
out = make([]byte, len(k))
copy(out, k)
@ -148,12 +148,12 @@ func (it *Iterator) Next() bool {
} else {
k, _ := cur.Seek(last)
if !bytes.Equal(k, last) {
return fmt.Errorf("Couldn't pick up after", k)
return fmt.Errorf("could not pick up after %v", k)
}
}
for i < bufferSize {
k, v := cur.Next()
if k == nil || !bytes.HasPrefix(k, it.checkId) {
if k == nil || !bytes.HasPrefix(k, it.checkID) {
it.buffer = append(it.buffer, nil)
break
}
@ -170,7 +170,7 @@ func (it *Iterator) Next() bool {
})
if err != nil {
if err != errNotExist {
glog.Error("Error nexting in database: ", err)
glog.Errorf("Error nexting in database: %v", err)
}
it.done = true
return false
@ -272,8 +272,8 @@ func (it *Iterator) Contains(v graph.Value) bool {
return false
}
offset := PositionOf(val, it.dir, it.qs)
if bytes.HasPrefix(val.key[offset:], it.checkId) {
// You may ask, why don't we check to see if it's a valid (not deleted) triple
if bytes.HasPrefix(val.key[offset:], it.checkID) {
// You may ask, why don't we check to see if it's a valid (not deleted) quad
// again?
//
// We've already done that -- in order to get the graph.Value token in the
@ -299,7 +299,7 @@ func (it *Iterator) DebugString(indent int) string {
it.tags.Tags(),
it.dir,
it.size,
it.qs.NameOf(&Token{it.bucket, it.checkId}),
it.qs.NameOf(&Token{it.bucket, it.checkID}),
)
}

72
graph/bolt/quadstore.go
View file

@ -32,7 +32,7 @@ import (
)
func init() {
graph.RegisterTripleStore("bolt", true, newQuadStore, createNewBolt)
graph.RegisterQuadStore("bolt", true, newQuadStore, createNewBolt)
}
var (
@ -77,7 +77,7 @@ func createNewBolt(path string, _ graph.Options) error {
return nil
}
func newQuadStore(path string, options graph.Options) (graph.TripleStore, error) {
func newQuadStore(path string, options graph.Options) (graph.QuadStore, error) {
var qs QuadStore
var err error
db, err := bolt.Open(path, 0600, nil)
@ -101,20 +101,20 @@ func (qs *QuadStore) createBuckets() error {
for _, index := range [][4]quad.Direction{spo, osp, pos, cps} {
_, err = tx.CreateBucket(bucketFor(index))
if err != nil {
return fmt.Errorf("Couldn't create bucket: %s", err)
return fmt.Errorf("could not create bucket: %s", err)
}
}
_, err = tx.CreateBucket(logBucket)
if err != nil {
return fmt.Errorf("Couldn't create bucket: %s", err)
return fmt.Errorf("could not create bucket: %s", err)
}
_, err = tx.CreateBucket(nodeBucket)
if err != nil {
return fmt.Errorf("Couldn't create bucket: %s", err)
return fmt.Errorf("could not create bucket: %s", err)
}
_, err = tx.CreateBucket(metaBucket)
if err != nil {
return fmt.Errorf("Couldn't create bucket: %s", err)
return fmt.Errorf("could not create bucket: %s", err)
}
return nil
})
@ -136,18 +136,28 @@ func bucketFor(d [4]quad.Direction) []byte {
return []byte{d[0].Prefix(), d[1].Prefix(), d[2].Prefix(), d[3].Prefix()}
}
func (qs *QuadStore) createKeyFor(d [4]quad.Direction, triple quad.Quad) []byte {
func hashOf(s string) []byte {
h := hashPool.Get().(hash.Hash)
h.Reset()
defer hashPool.Put(h)
key := make([]byte, 0, hashSize)
h.Write([]byte(s))
key = h.Sum(key)
return key
}
func (qs *QuadStore) createKeyFor(d [4]quad.Direction, q quad.Quad) []byte {
key := make([]byte, 0, (hashSize * 4))
key = append(key, qs.convertStringToByteHash(triple.Get(d[0]))...)
key = append(key, qs.convertStringToByteHash(triple.Get(d[1]))...)
key = append(key, qs.convertStringToByteHash(triple.Get(d[2]))...)
key = append(key, qs.convertStringToByteHash(triple.Get(d[3]))...)
key = append(key, hashOf(q.Get(d[0]))...)
key = append(key, hashOf(q.Get(d[1]))...)
key = append(key, hashOf(q.Get(d[2]))...)
key = append(key, hashOf(q.Get(d[3]))...)
return key
}
func (qs *QuadStore) createValueKeyFor(s string) []byte {
key := make([]byte, 0, hashSize)
key = append(key, qs.convertStringToByteHash(s)...)
key = append(key, hashOf(s)...)
return key
}
@ -173,13 +183,13 @@ var (
)
func (qs *QuadStore) ApplyDeltas(deltas []graph.Delta) error {
old_size := qs.size
old_horizon := qs.horizon
oldSize := qs.size
oldHorizon := qs.horizon
err := qs.db.Update(func(tx *bolt.Tx) error {
b := tx.Bucket(logBucket)
b.FillPercent = localFillPercent
resizeMap := make(map[string]int64)
size_change := int64(0)
sizeChange := int64(0)
for _, d := range deltas {
bytes, err := json.Marshal(d)
if err != nil {
@ -205,7 +215,7 @@ func (qs *QuadStore) ApplyDeltas(deltas []graph.Delta) error {
if d.Quad.Label != "" {
resizeMap[d.Quad.Label] += delta
}
size_change += delta
sizeChange += delta
qs.horizon = d.ID
}
for k, v := range resizeMap {
@ -216,14 +226,14 @@ func (qs *QuadStore) ApplyDeltas(deltas []graph.Delta) error {
}
}
}
qs.size += size_change
qs.size += sizeChange
return qs.WriteHorizonAndSize(tx)
})
if err != nil {
glog.Error("Couldn't write to DB for Delta set. Error: ", err)
qs.horizon = old_horizon
qs.size = old_size
qs.horizon = oldHorizon
qs.size = oldSize
return err
}
return nil
@ -243,11 +253,11 @@ func (qs *QuadStore) buildQuadWrite(tx *bolt.Tx, q quad.Quad, id int64, isAdd bo
}
if isAdd && len(entry.History)%2 == 1 {
glog.Error("Adding a valid triple ", entry)
glog.Error("Adding a valid quad ", entry)
return graph.ErrQuadExists
}
if !isAdd && len(entry.History)%2 == 0 {
glog.Error("Deleting an invalid triple ", entry)
glog.Error("Deleting an invalid quad ", entry)
return graph.ErrQuadNotExist
}
@ -373,22 +383,12 @@ func (qs *QuadStore) Quad(k graph.Value) quad.Quad {
return json.Unmarshal(data, &q)
})
if err != nil {
glog.Error("Error getting triple: ", err)
glog.Error("Error getting quad: ", err)
return quad.Quad{}
}
return q
}
func (qs *QuadStore) convertStringToByteHash(s string) []byte {
h := hashPool.Get().(hash.Hash)
h.Reset()
defer hashPool.Put(h)
key := make([]byte, 0, hashSize)
h.Write([]byte(s))
key = h.Sum(key)
return key
}
func (qs *QuadStore) ValueOf(s string) graph.Value {
return &Token{
bucket: nodeBucket,
@ -459,7 +459,7 @@ func (qs *QuadStore) getMetadata() error {
return err
}
func (qs *QuadStore) TripleIterator(d quad.Direction, val graph.Value) graph.Iterator {
func (qs *QuadStore) QuadIterator(d quad.Direction, val graph.Value) graph.Iterator {
var bucket []byte
switch d {
case quad.Subject:
@ -480,11 +480,11 @@ func (qs *QuadStore) NodesAllIterator() graph.Iterator {
return NewAllIterator(nodeBucket, quad.Any, qs)
}
func (qs *QuadStore) TriplesAllIterator() graph.Iterator {
func (qs *QuadStore) QuadsAllIterator() graph.Iterator {
return NewAllIterator(posBucket, quad.Predicate, qs)
}
func (qs *QuadStore) TripleDirection(val graph.Value, d quad.Direction) graph.Value {
func (qs *QuadStore) QuadDirection(val graph.Value, d quad.Direction) graph.Value {
v := val.(*Token)
offset := PositionOf(v, d, qs)
if offset != -1 {
@ -503,5 +503,5 @@ func compareTokens(a, b graph.Value) bool {
}
func (qs *QuadStore) FixedIterator() graph.FixedIterator {
return iterator.NewFixedIteratorWithCompare(compareTokens)
return iterator.NewFixed(compareTokens)
}

8
graph/bolt/quadstore_iterator_optimize.go
View file

@ -19,16 +19,16 @@ import (
"github.com/google/cayley/graph/iterator"
)
func (ts *QuadStore) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) {
func (qs *QuadStore) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) {
switch it.Type() {
case graph.LinksTo:
return ts.optimizeLinksTo(it.(*iterator.LinksTo))
return qs.optimizeLinksTo(it.(*iterator.LinksTo))
}
return it, false
}
func (ts *QuadStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool) {
func (qs *QuadStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool) {
subs := it.SubIterators()
if len(subs) != 1 {
return it, false
@ -41,7 +41,7 @@ func (ts *QuadStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool
panic("unexpected size during optimize")
}
val := primary.Result()
newIt := ts.TripleIterator(it.Direction(), val)
newIt := qs.QuadIterator(it.Direction(), val)
nt := newIt.Tagger()
nt.CopyFrom(it)
for _, tag := range primary.Tagger().Tags() {

4
graph/iterator.go
View file

@ -260,7 +260,7 @@ func (t Type) String() string {
type StatsContainer struct {
IteratorStats
Kind string
Uid uint64
UID uint64
SubIts []StatsContainer
}
@ -268,7 +268,7 @@ func DumpStats(it Iterator) StatsContainer {
var out StatsContainer
out.IteratorStats = it.Stats()
out.Kind = it.Type().String()
out.Uid = it.UID()
out.UID = it.UID()
for _, sub := range it.SubIterators() {
out.SubIts = append(out.SubIts, DumpStats(sub))
}

4
graph/iterator/all_iterator.go
View file

@ -17,9 +17,9 @@ package iterator
// Defines one of the base iterators, the All iterator. Which, logically
// enough, represents all nodes or all links in the graph.
//
// This particular file is actually vestigial. It's up to the TripleStore to give
// This particular file is actually vestigial. It's up to the QuadStore to give
// us an All iterator that represents all things in the graph. So this is
// really the All iterator for the MemTripleStore. That said, it *is* one of
// really the All iterator for the memstore.QuadStore. That said, it *is* one of
// the base iterators, and it helps just to see it here.
import (

2
graph/iterator/and_iterator.go
View file

@ -1,5 +1,5 @@
// Defines the And iterator, one of the base iterators. And requires no
// knowledge of the constituent TripleStore; its sole purpose is to act as an
// knowledge of the constituent QuadStore; its sole purpose is to act as an
// intersection operator across the subiterators it is given. If one iterator
// contains [1,3,5] and another [2,3,4] -- then And is an iterator that
// 'contains' [3]

2
graph/iterator/and_iterator_optimize.go
View file

@ -110,7 +110,7 @@ func closeIteratorList(its []graph.Iterator, except graph.Iterator) {
// Find if there is a single subiterator which is a valid replacement for this
// And.
func (_ *And) optimizeReplacement(its []graph.Iterator) graph.Iterator {
func (*And) optimizeReplacement(its []graph.Iterator) graph.Iterator {
// If we were created with no SubIterators, we're as good as Null.
if len(its) == 0 {
return &Null{}

2
graph/iterator/and_iterator_optimize_test.go
View file

@ -27,7 +27,7 @@ import (
func TestIteratorPromotion(t *testing.T) {
all := NewInt64(1, 3)
fixed := newFixed()
fixed := NewFixed(Identity)
fixed.Add(3)
a := NewAnd()
a.AddSubIterator(all)

10
graph/iterator/and_iterator_test.go
View file

@ -22,7 +22,7 @@ import (
// Make sure that tags work on the And.
func TestTag(t *testing.T) {
fix1 := newFixed()
fix1 := NewFixed(Identity)
fix1.Add(234)
fix1.Tagger().Add("foo")
and := NewAnd()
@ -55,12 +55,12 @@ func TestTag(t *testing.T) {
// Do a simple itersection of fixed values.
func TestAndAndFixedIterators(t *testing.T) {
fix1 := newFixed()
fix1 := NewFixed(Identity)
fix1.Add(1)
fix1.Add(2)
fix1.Add(3)
fix1.Add(4)
fix2 := newFixed()
fix2 := NewFixed(Identity)
fix2.Add(3)
fix2.Add(4)
fix2.Add(5)
@ -93,12 +93,12 @@ func TestAndAndFixedIterators(t *testing.T) {
// If there's no intersection, the size should still report the same,
// but there should be nothing to Next()
func TestNonOverlappingFixedIterators(t *testing.T) {
fix1 := newFixed()
fix1 := NewFixed(Identity)
fix1.Add(1)
fix1.Add(2)
fix1.Add(3)
fix1.Add(4)
fix2 := newFixed()
fix2 := NewFixed(Identity)
fix2.Add(5)
fix2.Add(6)
fix2.Add(7)

24
graph/iterator/fixed_iterator.go
View file

@ -18,7 +18,7 @@ package iterator
// contains an explicit fixed array of values.
//
// A fixed iterator requires an Equality function to be passed to it, by reason that graph.Value, the
// opaque Triple store value, may not answer to ==.
// opaque Quad store value, may not answer to ==.
import (
"fmt"
@ -42,24 +42,16 @@ type Fixed struct {
type Equality func(a, b graph.Value) bool
// Define an equality function of purely ==, which works for native types.
func BasicEquality(a, b graph.Value) bool {
if a == b {
return true
}
return false
func Identity(a, b graph.Value) bool {
return a == b
}
// Creates a new Fixed iterator based around == equality.
func newFixed() *Fixed {
return NewFixedIteratorWithCompare(BasicEquality)
}
// Creates a new Fixed iterator with a custom comparitor.
func NewFixedIteratorWithCompare(compareFn Equality) *Fixed {
// Creates a new Fixed iterator with a custom comparator.
func NewFixed(cmp Equality) *Fixed {
return &Fixed{
uid: NextUID(),
values: make([]graph.Value, 0, 20),
cmp: compareFn,
cmp: cmp,
}
}
@ -88,7 +80,7 @@ func (it *Fixed) TagResults(dst map[string]graph.Value) {
}
func (it *Fixed) Clone() graph.Iterator {
out := NewFixedIteratorWithCompare(it.cmp)
out := NewFixed(it.cmp)
for _, val := range it.values {
out.Add(val)
}
@ -108,7 +100,7 @@ func (it *Fixed) DebugString(indent int) string {
if len(it.values) > 0 {
value = fmt.Sprint(it.values[0])
}
return fmt.Sprintf("%s(%s %d tags: %s Size: %d id0: %d)",
return fmt.Sprintf("%s(%s %d tags: %s Size: %d id0: %s)",
strings.Repeat(" ", indent),
it.Type(),
it.UID(),

42
graph/iterator/hasa_iterator.go
View file

@ -21,7 +21,7 @@ package iterator
//
// HasA is weird in that it may return the same value twice if on the Next()
// path. That's okay -- in reality, it can be viewed as returning the value for
// a new triple, but to make logic much simpler, here we have the HasA.
// a new quad, but to make logic much simpler, here we have the HasA.
//
// Likewise, it's important to think about Contains()ing a HasA. When given a
// value to check, it means "Check all predicates that have this value for your
@ -43,13 +43,13 @@ import (
"github.com/google/cayley/quad"
)
// A HasA consists of a reference back to the graph.TripleStore that it references,
// a primary subiterator, a direction in which the triples for that subiterator point,
// A HasA consists of a reference back to the graph.QuadStore that it references,
// a primary subiterator, a direction in which the quads for that subiterator point,
// and a temporary holder for the iterator generated on Contains().
type HasA struct {
uid uint64
tags graph.Tagger
ts graph.TripleStore
qs graph.QuadStore
primaryIt graph.Iterator
dir quad.Direction
resultIt graph.Iterator
@ -57,12 +57,12 @@ type HasA struct {
runstats graph.IteratorStats
}
// Construct a new HasA iterator, given the triple subiterator, and the triple
// Construct a new HasA iterator, given the quad subiterator, and the quad
// direction for which it stands.
func NewHasA(ts graph.TripleStore, subIt graph.Iterator, d quad.Direction) *HasA {
func NewHasA(qs graph.QuadStore, subIt graph.Iterator, d quad.Direction) *HasA {
return &HasA{
uid: NextUID(),
ts: ts,
qs: qs,
primaryIt: subIt,
dir: d,
}
@ -89,7 +89,7 @@ func (it *HasA) Tagger() *graph.Tagger {
}
func (it *HasA) Clone() graph.Iterator {
out := NewHasA(it.ts, it.primaryIt.Clone(), it.dir)
out := NewHasA(it.qs, it.primaryIt.Clone(), it.dir)
out.tags.CopyFrom(it)
return out
}
@ -98,7 +98,7 @@ func (it *HasA) Clone() graph.Iterator {
func (it *HasA) Direction() quad.Direction { return it.dir }
// Pass the Optimize() call along to the subiterator. If it becomes Null,
// then the HasA becomes Null (there are no triples that have any directions).
// then the HasA becomes Null (there are no quads that have any directions).
func (it *HasA) Optimize() (graph.Iterator, bool) {
newPrimary, changed := it.primaryIt.Optimize()
if changed {
@ -140,34 +140,34 @@ func (it *HasA) DebugString(indent int) string {
}
// Check a value against our internal iterator. In order to do this, we must first open a new
// iterator of "triples that have `val` in our direction", given to us by the triple store,
// iterator of "quads that have `val` in our direction", given to us by the quad store,
// and then Next() values out of that iterator and Contains() them against our subiterator.
func (it *HasA) Contains(val graph.Value) bool {
graph.ContainsLogIn(it, val)
it.runstats.Contains += 1
if glog.V(4) {
glog.V(4).Infoln("Id is", it.ts.NameOf(val))
glog.V(4).Infoln("Id is", it.qs.NameOf(val))
}
// TODO(barakmich): Optimize this
if it.resultIt != nil {
it.resultIt.Close()
}
it.resultIt = it.ts.TripleIterator(it.dir, val)
it.resultIt = it.qs.QuadIterator(it.dir, val)
return graph.ContainsLogOut(it, val, it.NextContains())
}
// NextContains() is shared code between Contains() and GetNextResult() -- calls next on the
// result iterator (a triple iterator based on the last checked value) and returns true if
// result iterator (a quad iterator based on the last checked value) and returns true if
// another match is made.
func (it *HasA) NextContains() bool {
for graph.Next(it.resultIt) {
it.runstats.ContainsNext += 1
link := it.resultIt.Result()
if glog.V(4) {
glog.V(4).Infoln("Quad is", it.ts.Quad(link))
glog.V(4).Infoln("Quad is", it.qs.Quad(link))
}
if it.primaryIt.Contains(link) {
it.result = it.ts.TripleDirection(link, it.dir)
it.result = it.qs.QuadDirection(link, it.dir)
return true
}
}
@ -192,7 +192,7 @@ func (it *HasA) NextPath() bool {
}
// Next advances the iterator. This is simpler than Contains. We have a
// subiterator we can get a value from, and we can take that resultant triple,
// subiterator we can get a value from, and we can take that resultant quad,
// pull our direction out of it, and return that.
func (it *HasA) Next() bool {
graph.NextLogIn(it)
@ -206,7 +206,7 @@ func (it *HasA) Next() bool {
return graph.NextLogOut(it, 0, false)
}
tID := it.primaryIt.Result()
val := it.ts.TripleDirection(tID, it.dir)
val := it.qs.QuadDirection(tID, it.dir)
it.result = val
return graph.NextLogOut(it, val, true)
}
@ -217,20 +217,20 @@ func (it *HasA) Result() graph.Value {
// GetStats() returns the statistics on the HasA iterator. This is curious. Next
// cost is easy, it's an extra call or so on top of the subiterator Next cost.
// ContainsCost involves going to the graph.TripleStore, iterating out values, and hoping
// ContainsCost involves going to the graph.QuadStore, iterating out values, and hoping
// one sticks -- potentially expensive, depending on fanout. Size, however, is
// potentially smaller. we know at worst it's the size of the subiterator, but
// if there are many repeated values, it could be much smaller in totality.
func (it *HasA) Stats() graph.IteratorStats {
subitStats := it.primaryIt.Stats()
// TODO(barakmich): These should really come from the triplestore itself
// TODO(barakmich): These should really come from the quadstore itself
// and be optimized.
faninFactor := int64(1)
fanoutFactor := int64(30)
nextConstant := int64(2)
tripleConstant := int64(1)
quadConstant := int64(1)
return graph.IteratorStats{
NextCost: tripleConstant + subitStats.NextCost,
NextCost: quadConstant + subitStats.NextCost,
ContainsCost: (fanoutFactor * nextConstant) * subitStats.ContainsCost,
Size: faninFactor * subitStats.Size,
Next: it.runstats.Next,

20
graph/iterator/linksto_iterator.go
View file

@ -37,13 +37,13 @@ import (
"github.com/google/cayley/quad"
)
// A LinksTo has a reference back to the graph.TripleStore (to create the iterators
// A LinksTo has a reference back to the graph.QuadStore (to create the iterators
// for each node) the subiterator, and the direction the iterator comes from.
// `next_it` is the tempoarary iterator held per result in `primary_it`.
type LinksTo struct {
uid uint64
tags graph.Tagger
ts graph.TripleStore
qs graph.QuadStore
primaryIt graph.Iterator
dir quad.Direction
nextIt graph.Iterator
@ -53,10 +53,10 @@ type LinksTo struct {
// Construct a new LinksTo iterator around a direction and a subiterator of
// nodes.
func NewLinksTo(ts graph.TripleStore, it graph.Iterator, d quad.Direction) *LinksTo {
func NewLinksTo(qs graph.QuadStore, it graph.Iterator, d quad.Direction) *LinksTo {
return &LinksTo{
uid: NextUID(),
ts: ts,
qs: qs,
primaryIt: it,
dir: d,
nextIt: &Null{},
@ -80,7 +80,7 @@ func (it *LinksTo) Tagger() *graph.Tagger {
}
func (it *LinksTo) Clone() graph.Iterator {
out := NewLinksTo(it.ts, it.primaryIt.Clone(), it.dir)
out := NewLinksTo(it.qs, it.primaryIt.Clone(), it.dir)
out.tags.CopyFrom(it)
return out
}
@ -120,7 +120,7 @@ func (it *LinksTo) DebugString(indent int) string {
func (it *LinksTo) Contains(val graph.Value) bool {
graph.ContainsLogIn(it, val)
it.runstats.Contains += 1
node := it.ts.TripleDirection(val, it.dir)
node := it.qs.QuadDirection(val, it.dir)
if it.primaryIt.Contains(node) {
it.result = val
return graph.ContainsLogOut(it, val, true)
@ -143,10 +143,10 @@ func (it *LinksTo) Optimize() (graph.Iterator, bool) {
return it.primaryIt, true
}
}
// Ask the graph.TripleStore if we can be replaced. Often times, this is a great
// Ask the graph.QuadStore if we can be replaced. Often times, this is a great
// optimization opportunity (there's a fixed iterator underneath us, for
// example).
newReplacement, hasOne := it.ts.OptimizeIterator(it)
newReplacement, hasOne := it.qs.OptimizeIterator(it)
if hasOne {
it.Close()
return newReplacement, true
@ -170,7 +170,7 @@ func (it *LinksTo) Next() bool {
return graph.NextLogOut(it, 0, false)
}
it.nextIt.Close()
it.nextIt = it.ts.TripleIterator(it.dir, it.primaryIt.Result())
it.nextIt = it.qs.QuadIterator(it.dir, it.primaryIt.Result())
// Recurse -- return the first in the next set.
return it.Next()
@ -197,7 +197,7 @@ func (it *LinksTo) Type() graph.Type { return graph.LinksTo }
// Return a guess as to how big or costly it is to next the iterator.
func (it *LinksTo) Stats() graph.IteratorStats {
subitStats := it.primaryIt.Stats()
// TODO(barakmich): These should really come from the triplestore itself
// TODO(barakmich): These should really come from the quadstore itself
fanoutFactor := int64(20)
checkConstant := int64(1)
nextConstant := int64(2)

16
graph/iterator/linksto_iterator_test.go
View file

@ -21,23 +21,23 @@ import (
)
func TestLinksTo(t *testing.T) {
ts := &store{
qs := &store{
data: []string{1: "cool"},
iter: newFixed(),
iter: NewFixed(Identity),
}
ts.iter.(*Fixed).Add(2)
fixed := newFixed()
val := ts.ValueOf("cool")
qs.iter.(*Fixed).Add(2)
fixed := NewFixed(Identity)
val := qs.ValueOf("cool")
if val != 1 {
t.Fatalf("Failed to return correct value, got:%v expect:1", val)
}
fixed.Add(val)
lto := NewLinksTo(ts, fixed, quad.Object)
lto := NewLinksTo(qs, fixed, quad.Object)
if !lto.Next() {
t.Error("At least one triple matches the fixed object")
t.Error("At least one quad matches the fixed object")
}
val = lto.Result()
if val != 2 {
t.Errorf("Quad index 2, such as %s, should match %s", ts.Quad(2), ts.Quad(val))
t.Errorf("Quad index 2, such as %s, should match %s", qs.Quad(2), qs.Quad(val))
}
}

14
graph/iterator/mock_ts_test.go
View file

@ -14,14 +14,12 @@
package iterator
// A quickly mocked version of the TripleStore interface, for use in tests.
// Can better used Mock.Called but will fill in as needed.
import (
"github.com/google/cayley/graph"
"github.com/google/cayley/quad"
)
// store is a mocked version of the QuadStore interface, for use in tests.
type store struct {
data []string
iter graph.Iterator
@ -40,13 +38,13 @@ func (qs *store) ApplyDeltas([]graph.Delta) error { return nil }
func (qs *store) Quad(graph.Value) quad.Quad { return quad.Quad{} }
func (qs *store) TripleIterator(d quad.Direction, i graph.Value) graph.Iterator {
func (qs *store) QuadIterator(d quad.Direction, i graph.Value) graph.Iterator {
return qs.iter
}
func (qs *store) NodesAllIterator() graph.Iterator { return &Null{} }
func (qs *store) TriplesAllIterator() graph.Iterator { return &Null{} }
func (qs *store) QuadsAllIterator() graph.Iterator { return &Null{} }
func (qs *store) NameOf(v graph.Value) string {
i := v.(int)
@ -67,11 +65,11 @@ func (qs *store) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) {
}
func (qs *store) FixedIterator() graph.FixedIterator {
return NewFixedIteratorWithCompare(BasicEquality)
return NewFixed(Identity)
}
func (qs *store) Close() {}
func (qs *store) TripleDirection(graph.Value, quad.Direction) graph.Value { return 0 }
func (qs *store) QuadDirection(graph.Value, quad.Direction) graph.Value { return 0 }
func (qs *store) RemoveTriple(t quad.Quad) {}
func (qs *store) RemoveQuad(t quad.Quad) {}

10
graph/iterator/or_iterator_test.go
View file

@ -31,11 +31,11 @@ func iterated(it graph.Iterator) []int {
func TestOrIteratorBasics(t *testing.T) {
or := NewOr()
f1 := newFixed()
f1 := NewFixed(Identity)
f1.Add(1)
f1.Add(2)
f1.Add(3)
f2 := newFixed()
f2 := NewFixed(Identity)
f2.Add(3)
f2.Add(9)
f2.Add(20)
@ -77,11 +77,11 @@ func TestOrIteratorBasics(t *testing.T) {
func TestShortCircuitingOrBasics(t *testing.T) {
var or *Or
f1 := newFixed()
f1 := NewFixed(Identity)
f1.Add(1)
f1.Add(2)
f1.Add(3)
f2 := newFixed()
f2 := NewFixed(Identity)
f2.Add(3)
f2.Add(9)
f2.Add(20)
@ -133,7 +133,7 @@ func TestShortCircuitingOrBasics(t *testing.T) {
// Check that it pulls the second iterator's numbers if the first is empty.
or = NewShortCircuitOr()
or.AddSubIterator(newFixed())
or.AddSubIterator(NewFixed(Identity))
or.AddSubIterator(f2)
expect = []int{3, 9, 20, 21}
for i := 0; i < 2; i++ {

120
graph/iterator/query_shape.go
View file

@ -20,7 +20,7 @@ import (
)
type Node struct {
Id int `json:"id"`
ID int `json:"id"`
Tags []string `json:"tags,omitempty"`
Values []string `json:"values,omitempty"`
IsLinkNode bool `json:"is_link_node"`
@ -37,47 +37,47 @@ type Link struct {
type queryShape struct {
nodes []Node
links []Link
ts graph.TripleStore
nodeId int
hasaIds []int
qs graph.QuadStore
nodeID int
hasaIDs []int
hasaDirs []quad.Direction
}
func OutputQueryShapeForIterator(it graph.Iterator, ts graph.TripleStore, outputMap map[string]interface{}) {
qs := &queryShape{
ts: ts,
nodeId: 1,
func OutputQueryShapeForIterator(it graph.Iterator, qs graph.QuadStore, outputMap map[string]interface{}) {
s := &queryShape{
qs: qs,
nodeID: 1,
}
node := qs.MakeNode(it.Clone())
qs.AddNode(node)
outputMap["nodes"] = qs.nodes
outputMap["links"] = qs.links
node := s.MakeNode(it.Clone())
s.AddNode(node)
outputMap["nodes"] = s.nodes
outputMap["links"] = s.links
}
func (qs *queryShape) AddNode(n *Node) {
qs.nodes = append(qs.nodes, *n)
func (s *queryShape) AddNode(n *Node) {
s.nodes = append(s.nodes, *n)
}
func (qs *queryShape) AddLink(l *Link) {
qs.links = append(qs.links, *l)
func (s *queryShape) AddLink(l *Link) {
s.links = append(s.links, *l)
}
func (qs *queryShape) LastHasa() (int, quad.Direction) {
return qs.hasaIds[len(qs.hasaIds)-1], qs.hasaDirs[len(qs.hasaDirs)-1]
func (s *queryShape) LastHasa() (int, quad.Direction) {
return s.hasaIDs[len(s.hasaIDs)-1], s.hasaDirs[len(s.hasaDirs)-1]
}
func (qs *queryShape) PushHasa(i int, d quad.Direction) {
qs.hasaIds = append(qs.hasaIds, i)
qs.hasaDirs = append(qs.hasaDirs, d)
func (s *queryShape) PushHasa(i int, d quad.Direction) {
s.hasaIDs = append(s.hasaIDs, i)
s.hasaDirs = append(s.hasaDirs, d)
}
func (qs *queryShape) RemoveHasa() {
qs.hasaIds = qs.hasaIds[:len(qs.hasaIds)-1]
qs.hasaDirs = qs.hasaDirs[:len(qs.hasaDirs)-1]
func (s *queryShape) RemoveHasa() {
s.hasaIDs = s.hasaIDs[:len(s.hasaIDs)-1]
s.hasaDirs = s.hasaDirs[:len(s.hasaDirs)-1]
}
func (qs *queryShape) StealNode(left *Node, right *Node) {
func (s *queryShape) StealNode(left *Node, right *Node) {
for _, v := range right.Values {
left.Values = append(left.Values, v)
}
@ -86,88 +86,88 @@ func (qs *queryShape) StealNode(left *Node, right *Node) {
}
left.IsLinkNode = left.IsLinkNode || right.IsLinkNode
left.IsFixed = left.IsFixed || right.IsFixed
for i, link := range qs.links {
for i, link := range s.links {
rewrite := false
if link.LinkNode == right.Id {
link.LinkNode = left.Id
if link.LinkNode == right.ID {
link.LinkNode = left.ID
rewrite = true
}
if link.Source == right.Id {
link.Source = left.Id
if link.Source == right.ID {
link.Source = left.ID
rewrite = true
}
if link.Target == right.Id {
link.Target = left.Id
if link.Target == right.ID {
link.Target = left.ID
rewrite = true
}
if rewrite {
qs.links = append(append(qs.links[:i], qs.links[i+1:]...), link)
s.links = append(append(s.links[:i], s.links[i+1:]...), link)
}
}
}
func (qs *queryShape) MakeNode(it graph.Iterator) *Node {
n := Node{Id: qs.nodeId}
func (s *queryShape) MakeNode(it graph.Iterator) *Node {
n := Node{ID: s.nodeID}
for _, tag := range it.Tagger().Tags() {
n.Tags = append(n.Tags, tag)
}
for k, _ := range it.Tagger().Fixed() {
for k := range it.Tagger().Fixed() {
n.Tags = append(n.Tags, k)
}
switch it.Type() {
case graph.And:
for _, sub := range it.SubIterators() {
qs.nodeId++
newNode := qs.MakeNode(sub)
s.nodeID++
newNode := s.MakeNode(sub)
if sub.Type() != graph.Or {
qs.StealNode(&n, newNode)
s.StealNode(&n, newNode)
} else {
qs.AddNode(newNode)
qs.AddLink(&Link{n.Id, newNode.Id, 0, 0})
s.AddNode(newNode)
s.AddLink(&Link{n.ID, newNode.ID, 0, 0})
}
}
case graph.Fixed:
n.IsFixed = true
for graph.Next(it) {
n.Values = append(n.Values, qs.ts.NameOf(it.Result()))
n.Values = append(n.Values, s.qs.NameOf(it.Result()))
}
case graph.HasA:
hasa := it.(*HasA)
qs.PushHasa(n.Id, hasa.dir)
qs.nodeId++
newNode := qs.MakeNode(hasa.primaryIt)
qs.AddNode(newNode)
qs.RemoveHasa()
s.PushHasa(n.ID, hasa.dir)
s.nodeID++
newNode := s.MakeNode(hasa.primaryIt)
s.AddNode(newNode)
s.RemoveHasa()
case graph.Or:
for _, sub := range it.SubIterators() {
qs.nodeId++
newNode := qs.MakeNode(sub)
s.nodeID++
newNode := s.MakeNode(sub)
if sub.Type() == graph.Or {
qs.StealNode(&n, newNode)
s.StealNode(&n, newNode)
} else {
qs.AddNode(newNode)
qs.AddLink(&Link{n.Id, newNode.Id, 0, 0})
s.AddNode(newNode)
s.AddLink(&Link{n.ID, newNode.ID, 0, 0})
}
}
case graph.LinksTo:
n.IsLinkNode = true
lto := it.(*LinksTo)
qs.nodeId++
newNode := qs.MakeNode(lto.primaryIt)
hasaID, hasaDir := qs.LastHasa()
s.nodeID++
newNode := s.MakeNode(lto.primaryIt)
hasaID, hasaDir := s.LastHasa()
if (hasaDir == quad.Subject && lto.dir == quad.Object) ||
(hasaDir == quad.Object && lto.dir == quad.Subject) {
qs.AddNode(newNode)
s.AddNode(newNode)
if hasaDir == quad.Subject {
qs.AddLink(&Link{hasaID, newNode.Id, 0, n.Id})
s.AddLink(&Link{hasaID, newNode.ID, 0, n.ID})
} else {
qs.AddLink(&Link{newNode.Id, hasaID, 0, n.Id})
s.AddLink(&Link{newNode.ID, hasaID, 0, n.ID})
}
} else if lto.primaryIt.Type() == graph.Fixed {
qs.StealNode(&n, newNode)
s.StealNode(&n, newNode)
} else {
qs.AddNode(newNode)
s.AddNode(newNode)
}
case graph.Optional:
// Unsupported, for the moment

48
graph/iterator/query_shape_test.go
View file

@ -22,23 +22,23 @@ import (
"github.com/google/cayley/quad"
)
func hasaWithTag(ts graph.TripleStore, tag string, target string) *HasA {
func hasaWithTag(qs graph.QuadStore, tag string, target string) *HasA {
and := NewAnd()
obj := ts.FixedIterator()
obj.Add(ts.ValueOf(target))
obj := qs.FixedIterator()
obj.Add(qs.ValueOf(target))
obj.Tagger().Add(tag)
and.AddSubIterator(NewLinksTo(ts, obj, quad.Object))
and.AddSubIterator(NewLinksTo(qs, obj, quad.Object))
pred := ts.FixedIterator()
pred.Add(ts.ValueOf("status"))
and.AddSubIterator(NewLinksTo(ts, pred, quad.Predicate))
pred := qs.FixedIterator()
pred.Add(qs.ValueOf("status"))
and.AddSubIterator(NewLinksTo(qs, pred, quad.Predicate))
return NewHasA(ts, and, quad.Subject)
return NewHasA(qs, and, quad.Subject)
}
func TestQueryShape(t *testing.T) {
ts := &store{
qs := &store{
data: []string{
1: "cool",
2: "status",
@ -48,11 +48,11 @@ func TestQueryShape(t *testing.T) {
}
// Given a single linkage iterator's shape.
hasa := hasaWithTag(ts, "tag", "cool")
hasa := hasaWithTag(qs, "tag", "cool")
hasa.Tagger().Add("top")
shape := make(map[string]interface{})
OutputQueryShapeForIterator(hasa, ts, shape)
OutputQueryShapeForIterator(hasa, qs, shape)
nodes := shape["nodes"].([]Node)
if len(nodes) != 3 {
@ -77,14 +77,14 @@ func TestQueryShape(t *testing.T) {
// Link should be correctly typed.
nodes = shape["nodes"].([]Node)
link := shape["links"].([]Link)[0]
if link.Source != nodes[2].Id {
t.Errorf("Failed to get correct link source, got:%v expect:%v", link.Source, nodes[2].Id)
if link.Source != nodes[2].ID {
t.Errorf("Failed to get correct link source, got:%v expect:%v", link.Source, nodes[2].ID)
}
if link.Target != nodes[0].Id {
t.Errorf("Failed to get correct link target, got:%v expect:%v", link.Target, nodes[0].Id)
if link.Target != nodes[0].ID {
t.Errorf("Failed to get correct link target, got:%v expect:%v", link.Target, nodes[0].ID)
}
if link.LinkNode != nodes[1].Id {
t.Errorf("Failed to get correct link node, got:%v expect:%v", link.LinkNode, nodes[1].Id)
if link.LinkNode != nodes[1].ID {
t.Errorf("Failed to get correct link node, got:%v expect:%v", link.LinkNode, nodes[1].ID)
}
if link.Pred != 0 {
t.Errorf("Failed to get correct number of predecessors:%v expect:0", link.Pred)
@ -93,23 +93,23 @@ func TestQueryShape(t *testing.T) {
// Given a name-of-an-and-iterator's shape.
andInternal := NewAnd()
hasa1 := hasaWithTag(ts, "tag1", "cool")
hasa1 := hasaWithTag(qs, "tag1", "cool")
hasa1.Tagger().Add("hasa1")
andInternal.AddSubIterator(hasa1)
hasa2 := hasaWithTag(ts, "tag2", "fun")
hasa2 := hasaWithTag(qs, "tag2", "fun")
hasa2.Tagger().Add("hasa2")
andInternal.AddSubIterator(hasa2)
pred := ts.FixedIterator()
pred.Add(ts.ValueOf("name"))
pred := qs.FixedIterator()
pred.Add(qs.ValueOf("name"))
and := NewAnd()
and.AddSubIterator(NewLinksTo(ts, andInternal, quad.Subject))
and.AddSubIterator(NewLinksTo(ts, pred, quad.Predicate))
and.AddSubIterator(NewLinksTo(qs, andInternal, quad.Subject))
and.AddSubIterator(NewLinksTo(qs, pred, quad.Predicate))
shape = make(map[string]interface{})
OutputQueryShapeForIterator(NewHasA(ts, and, quad.Object), ts, shape)
OutputQueryShapeForIterator(NewHasA(qs, and, quad.Object), qs, shape)
links = shape["links"].([]Link)
if len(links) != 3 {

28
graph/iterator/value_comparison_iterator.go
View file

@ -38,10 +38,10 @@ import (
type Operator int
const (
kCompareLT Operator = iota
kCompareLTE
kCompareGT
kCompareGTE
compareLT Operator = iota
compareLTE
compareGT
compareGTE
// Why no Equals? Because that's usually an AndIterator.
)
@ -51,17 +51,17 @@ type Comparison struct {
subIt graph.Iterator
op Operator
val interface{}
ts graph.TripleStore
qs graph.QuadStore
result graph.Value
}
func NewComparison(sub graph.Iterator, op Operator, val interface{}, ts graph.TripleStore) *Comparison {
func NewComparison(sub graph.Iterator, op Operator, val interface{}, qs graph.QuadStore) *Comparison {
return &Comparison{
uid: NextUID(),
subIt: sub,
op: op,
val: val,
ts: ts,
qs: qs,
}
}
@ -73,7 +73,7 @@ func (it *Comparison) UID() uint64 {
// and our operator, determine whether or not we meet the requirement.
func (it *Comparison) doComparison(val graph.Value) bool {
//TODO(barakmich): Implement string comparison.
nodeStr := it.ts.NameOf(val)
nodeStr := it.qs.NameOf(val)
switch cVal := it.val.(type) {
case int:
cInt := int64(cVal)
@ -99,13 +99,13 @@ func (it *Comparison) Close() {
func RunIntOp(a int64, op Operator, b int64) bool {
switch op {
case kCompareLT:
case compareLT:
return a < b
case kCompareLTE:
case compareLTE:
return a <= b
case kCompareGT:
case compareGT:
return a > b
case kCompareGTE:
case compareGTE:
return a >= b
default:
log.Fatal("Unknown operator type")
@ -122,7 +122,7 @@ func (it *Comparison) Tagger() *graph.Tagger {
}
func (it *Comparison) Clone() graph.Iterator {
out := NewComparison(it.subIt.Clone(), it.op, it.val, it.ts)
out := NewComparison(it.subIt.Clone(), it.op, it.val, it.qs)
out.tags.CopyFrom(it)
return out
}
@ -154,7 +154,7 @@ func (it *Comparison) NextPath() bool {
return false
}
if it.doComparison(it.subIt.Result()) {
return true
break
}
}
it.result = it.subIt.Result()

24
graph/iterator/value_comparison_iterator_test.go
View file

@ -24,7 +24,7 @@ import (
var simpleStore = &store{data: []string{"0", "1", "2", "3", "4", "5"}}
func simpleFixedIterator() *Fixed {
f := newFixed()
f := NewFixed(Identity)
for i := 0; i < 5; i++ {
f.Add(i)
}
@ -40,37 +40,37 @@ var comparisonTests = []struct {
{
message: "successful int64 less than comparison",
operand: int64(3),
operator: kCompareLT,
operator: compareLT,
expect: []string{"0", "1", "2"},
},
{
message: "empty int64 less than comparison",
operand: int64(0),
operator: kCompareLT,
operator: compareLT,
expect: nil,
},
{
message: "successful int64 greater than comparison",
operand: int64(2),
operator: kCompareGT,
operator: compareGT,
expect: []string{"3", "4"},
},
{
message: "successful int64 greater than or equal comparison",
operand: int64(2),
operator: kCompareGTE,
operator: compareGTE,
expect: []string{"2", "3", "4"},
},
}
func TestValueComparison(t *testing.T) {
for _, test := range comparisonTests {
ts := simpleStore
vc := NewComparison(simpleFixedIterator(), test.operator, test.operand, ts)
qs := simpleStore
vc := NewComparison(simpleFixedIterator(), test.operator, test.operand, qs)
var got []string
for vc.Next() {
got = append(got, ts.NameOf(vc.Result()))
got = append(got, qs.NameOf(vc.Result()))
}
if !reflect.DeepEqual(got, test.expect) {
t.Errorf("Failed to show %s, got:%q expect:%q", test.message, got, test.expect)
@ -86,25 +86,25 @@ var vciContainsTests = []struct {
}{
{
message: "1 is less than 2",
operator: kCompareGTE,
operator: compareGTE,
check: 1,
expect: false,
},
{
message: "2 is greater than or equal to 2",
operator: kCompareGTE,
operator: compareGTE,
check: 2,
expect: true,
},
{
message: "3 is greater than or equal to 2",
operator: kCompareGTE,
operator: compareGTE,
check: 3,
expect: true,
},
{
message: "5 is absent from iterator",
operator: kCompareGTE,
operator: compareGTE,
check: 5,
expect: false,
},

14
graph/leveldb/all_iterator.go
View file

@ -34,12 +34,12 @@ type AllIterator struct {
dir quad.Direction
open bool
iter ldbit.Iterator
ts *TripleStore
qs *QuadStore
ro *opt.ReadOptions
result graph.Value
}
func NewAllIterator(prefix string, d quad.Direction, ts *TripleStore) *AllIterator {
func NewAllIterator(prefix string, d quad.Direction, qs *QuadStore) *AllIterator {
opts := &opt.ReadOptions{
DontFillCache: true,
}
@ -47,11 +47,11 @@ func NewAllIterator(prefix string, d quad.Direction, ts *TripleStore) *AllIterat
it := AllIterator{
uid: iterator.NextUID(),
ro: opts,
iter: ts.db.NewIterator(nil, opts),
iter: qs.db.NewIterator(nil, opts),
prefix: []byte(prefix),
dir: d,
open: true,
ts: ts,
qs: qs,
}
it.iter.Seek(it.prefix)
@ -71,7 +71,7 @@ func (it *AllIterator) UID() uint64 {
func (it *AllIterator) Reset() {
if !it.open {
it.iter = it.ts.db.NewIterator(nil, it.ro)
it.iter = it.qs.db.NewIterator(nil, it.ro)
it.open = true
}
it.iter.Seek(it.prefix)
@ -96,7 +96,7 @@ func (it *AllIterator) TagResults(dst map[string]graph.Value) {
}
func (it *AllIterator) Clone() graph.Iterator {
out := NewAllIterator(string(it.prefix), it.dir, it.ts)
out := NewAllIterator(string(it.prefix), it.dir, it.qs)
out.tags.CopyFrom(it)
return out
}
@ -151,7 +151,7 @@ func (it *AllIterator) Close() {
}
func (it *AllIterator) Size() (int64, bool) {
size, err := it.ts.SizeOfPrefix(it.prefix)
size, err := it.qs.SizeOfPrefix(it.prefix)
if err == nil {
return size, false
}

20
graph/leveldb/iterator.go
View file

@ -33,17 +33,17 @@ type Iterator struct {
uid uint64
tags graph.Tagger
nextPrefix []byte
checkId []byte
checkID []byte
dir quad.Direction
open bool
iter ldbit.Iterator
qs *TripleStore
qs *QuadStore
ro *opt.ReadOptions
originalPrefix string
result graph.Value
}
func NewIterator(prefix string, d quad.Direction, value graph.Value, qs *TripleStore) *Iterator {
func NewIterator(prefix string, d quad.Direction, value graph.Value, qs *QuadStore) *Iterator {
vb := value.(Token)
p := make([]byte, 0, 2+hashSize)
p = append(p, []byte(prefix)...)
@ -56,7 +56,7 @@ func NewIterator(prefix string, d quad.Direction, value graph.Value, qs *TripleS
it := Iterator{
uid: iterator.NextUID(),
nextPrefix: p,
checkId: vb,
checkID: vb,
dir: d,
originalPrefix: prefix,
ro: opts,
@ -106,7 +106,7 @@ func (it *Iterator) TagResults(dst map[string]graph.Value) {
}
func (it *Iterator) Clone() graph.Iterator {
out := NewIterator(it.originalPrefix, it.dir, Token(it.checkId), it.qs)
out := NewIterator(it.originalPrefix, it.dir, Token(it.checkID), it.qs)
out.tags.CopyFrom(it)
return out
}
@ -173,7 +173,7 @@ func (it *Iterator) SubIterators() []graph.Iterator {
return nil
}
func PositionOf(prefix []byte, d quad.Direction, qs *TripleStore) int {
func PositionOf(prefix []byte, d quad.Direction, qs *QuadStore) int {
if bytes.Equal(prefix, []byte("sp")) {
switch d {
case quad.Subject:
@ -231,8 +231,8 @@ func (it *Iterator) Contains(v graph.Value) bool {
return false
}
offset := PositionOf(val[0:2], it.dir, it.qs)
if bytes.HasPrefix(val[offset:], it.checkId[1:]) {
// You may ask, why don't we check to see if it's a valid (not deleted) triple
if bytes.HasPrefix(val[offset:], it.checkID[1:]) {
// You may ask, why don't we check to see if it's a valid (not deleted) quad
// again?
//
// We've already done that -- in order to get the graph.Value token in the
@ -247,7 +247,7 @@ func (it *Iterator) Contains(v graph.Value) bool {
}
func (it *Iterator) Size() (int64, bool) {
return it.qs.SizeOf(Token(it.checkId)), true
return it.qs.SizeOf(Token(it.checkID)), true
}
func (it *Iterator) DebugString(indent int) string {
@ -259,7 +259,7 @@ func (it *Iterator) DebugString(indent int) string {
it.tags.Tags(),
it.dir,
size,
it.qs.NameOf(Token(it.checkId)),
it.qs.NameOf(Token(it.checkID)),
)
}

118
graph/leveldb/leveldb_test.go
View file

@ -27,8 +27,8 @@ import (
"github.com/google/cayley/writer"
)
func makeTripleSet() []quad.Quad {
tripleSet := []quad.Quad{
func makeQuadSet() []quad.Quad {
quadSet := []quad.Quad{
{"A", "follows", "B", ""},
{"C", "follows", "B", ""},
{"C", "follows", "D", ""},
@ -41,10 +41,10 @@ func makeTripleSet() []quad.Quad {
{"D", "status", "cool", "status_graph"},
{"G", "status", "cool", "status_graph"},
}
return tripleSet
return quadSet
}
func iteratedTriples(qs graph.TripleStore, it graph.Iterator) []quad.Quad {
func iteratedQuads(qs graph.QuadStore, it graph.Iterator) []quad.Quad {
var res ordered
for graph.Next(it) {
res = append(res, qs.Quad(it.Result()))
@ -80,7 +80,7 @@ func (o ordered) Less(i, j int) bool {
}
func (o ordered) Swap(i, j int) { o[i], o[j] = o[j], o[i] }
func iteratedNames(qs graph.TripleStore, it graph.Iterator) []string {
func iteratedNames(qs graph.QuadStore, it graph.Iterator) []string {
var res []string
for graph.Next(it) {
res = append(res, qs.NameOf(it.Result()))
@ -101,9 +101,9 @@ func TestCreateDatabase(t *testing.T) {
t.Fatal("Failed to create LevelDB database.")
}
qs, err := newTripleStore(tmpDir, nil)
qs, err := newQuadStore(tmpDir, nil)
if qs == nil || err != nil {
t.Error("Failed to create leveldb TripleStore.")
t.Error("Failed to create leveldb QuadStore.")
}
if s := qs.Size(); s != 0 {
t.Errorf("Unexpected size, got:%d expected:0", s)
@ -131,20 +131,25 @@ func TestLoadDatabase(t *testing.T) {
t.Fatal("Failed to create LevelDB database.")
}
qs, err := newTripleStore(tmpDir, nil)
qs, err := newQuadStore(tmpDir, nil)
if qs == nil || err != nil {
t.Error("Failed to create leveldb TripleStore.")
t.Error("Failed to create leveldb QuadStore.")
}
w, _ := writer.NewSingleReplication(qs, nil)
w.AddQuad(quad.Quad{"Something", "points_to", "Something Else", "context"})
w.AddQuad(quad.Quad{
Subject: "Something",
Predicate: "points_to",
Object: "Something Else",
Label: "context",
})
for _, pq := range []string{"Something", "points_to", "Something Else", "context"} {
if got := qs.NameOf(qs.ValueOf(pq)); got != pq {
t.Errorf("Failed to roundtrip %q, got:%q expect:%q", pq, got, pq)
}
}
if s := qs.Size(); s != 1 {
t.Errorf("Unexpected triplestore size, got:%d expect:1", s)
t.Errorf("Unexpected quadstore size, got:%d expect:1", s)
}
qs.Close()
@ -152,31 +157,36 @@ func TestLoadDatabase(t *testing.T) {
if err != nil {
t.Fatal("Failed to create LevelDB database.")
}
qs, err = newTripleStore(tmpDir, nil)
qs, err = newQuadStore(tmpDir, nil)
if qs == nil || err != nil {
t.Error("Failed to create leveldb TripleStore.")
t.Error("Failed to create leveldb QuadStore.")
}
w, _ = writer.NewSingleReplication(qs, nil)
ts2, didConvert := qs.(*TripleStore)
ts2, didConvert := qs.(*QuadStore)
if !didConvert {
t.Errorf("Could not convert from generic to LevelDB TripleStore")
t.Errorf("Could not convert from generic to LevelDB QuadStore")
}
w.AddQuadSet(makeTripleSet())
w.AddQuadSet(makeQuadSet())
if s := qs.Size(); s != 11 {
t.Errorf("Unexpected triplestore size, got:%d expect:11", s)
t.Errorf("Unexpected quadstore size, got:%d expect:11", s)
}
if s := ts2.SizeOf(qs.ValueOf("B")); s != 5 {
t.Errorf("Unexpected triplestore size, got:%d expect:5", s)
t.Errorf("Unexpected quadstore size, got:%d expect:5", s)
}
w.RemoveQuad(quad.Quad{"A", "follows", "B", ""})
w.RemoveQuad(quad.Quad{
Subject: "A",
Predicate: "follows",
Object: "B",
Label: "",
})
if s := qs.Size(); s != 10 {
t.Errorf("Unexpected triplestore size after RemoveTriple, got:%d expect:10", s)
t.Errorf("Unexpected quadstore size after RemoveQuad, got:%d expect:10", s)
}
if s := ts2.SizeOf(qs.ValueOf("B")); s != 4 {
t.Errorf("Unexpected triplestore size, got:%d expect:4", s)
t.Errorf("Unexpected quadstore size, got:%d expect:4", s)
}
qs.Close()
@ -195,13 +205,13 @@ func TestIterator(t *testing.T) {
t.Fatal("Failed to create LevelDB database.")
}
qs, err := newTripleStore(tmpDir, nil)
qs, err := newQuadStore(tmpDir, nil)
if qs == nil || err != nil {
t.Error("Failed to create leveldb TripleStore.")
t.Error("Failed to create leveldb QuadStore.")
}
w, _ := writer.NewSingleReplication(qs, nil)
w.AddQuadSet(makeTripleSet())
w.AddQuadSet(makeQuadSet())
var it graph.Iterator
it = qs.NodesAllIterator()
@ -262,19 +272,19 @@ func TestIterator(t *testing.T) {
*/
it.Reset()
it = qs.TriplesAllIterator()
it = qs.QuadsAllIterator()
graph.Next(it)
triple := qs.Quad(it.Result())
set := makeTripleSet()
q := qs.Quad(it.Result())
set := makeQuadSet()
var ok bool
for _, t := range set {
if t.String() == triple.String() {
if t.String() == q.String() {
ok = true
break
}
}
if !ok {
t.Errorf("Failed to find %q during iteration, got:%q", triple, set)
t.Errorf("Failed to find %q during iteration, got:%q", q, set)
}
qs.Close()
@ -290,14 +300,14 @@ func TestSetIterator(t *testing.T) {
t.Fatalf("Failed to create working directory")
}
qs, err := newTripleStore(tmpDir, nil)
qs, err := newQuadStore(tmpDir, nil)
if qs == nil || err != nil {
t.Error("Failed to create leveldb TripleStore.")
t.Error("Failed to create leveldb QuadStore.")
}
defer qs.Close()
w, _ := writer.NewSingleReplication(qs, nil)
w.AddQuadSet(makeTripleSet())
w.AddQuadSet(makeQuadSet())
expect := []quad.Quad{
{"C", "follows", "B", ""},
@ -306,46 +316,46 @@ func TestSetIterator(t *testing.T) {
sort.Sort(ordered(expect))
// Subject iterator.
it := qs.TripleIterator(quad.Subject, qs.ValueOf("C"))
it := qs.QuadIterator(quad.Subject, qs.ValueOf("C"))
if got := iteratedTriples(qs, it); !reflect.DeepEqual(got, expect) {
if got := iteratedQuads(qs, it); !reflect.DeepEqual(got, expect) {
t.Errorf("Failed to get expected results, got:%v expect:%v", got, expect)
}
it.Reset()
and := iterator.NewAnd()
and.AddSubIterator(qs.TriplesAllIterator())
and.AddSubIterator(qs.QuadsAllIterator())
and.AddSubIterator(it)
if got := iteratedTriples(qs, and); !reflect.DeepEqual(got, expect) {
if got := iteratedQuads(qs, and); !reflect.DeepEqual(got, expect) {
t.Errorf("Failed to get confirm expected results, got:%v expect:%v", got, expect)
}
// Object iterator.
it = qs.TripleIterator(quad.Object, qs.ValueOf("F"))
it = qs.QuadIterator(quad.Object, qs.ValueOf("F"))
expect = []quad.Quad{
{"B", "follows", "F", ""},
{"E", "follows", "F", ""},
}
sort.Sort(ordered(expect))
if got := iteratedTriples(qs, it); !reflect.DeepEqual(got, expect) {
if got := iteratedQuads(qs, it); !reflect.DeepEqual(got, expect) {
t.Errorf("Failed to get expected results, got:%v expect:%v", got, expect)
}
and = iterator.NewAnd()
and.AddSubIterator(qs.TripleIterator(quad.Subject, qs.ValueOf("B")))
and.AddSubIterator(qs.QuadIterator(quad.Subject, qs.ValueOf("B")))
and.AddSubIterator(it)
expect = []quad.Quad{
{"B", "follows", "F", ""},
}
if got := iteratedTriples(qs, and); !reflect.DeepEqual(got, expect) {
if got := iteratedQuads(qs, and); !reflect.DeepEqual(got, expect) {
t.Errorf("Failed to get confirm expected results, got:%v expect:%v", got, expect)
}
// Predicate iterator.
it = qs.TripleIterator(quad.Predicate, qs.ValueOf("status"))
it = qs.QuadIterator(quad.Predicate, qs.ValueOf("status"))
expect = []quad.Quad{
{"B", "status", "cool", "status_graph"},
@ -353,12 +363,12 @@ func TestSetIterator(t *testing.T) {
{"G", "status", "cool", "status_graph"},
}
sort.Sort(ordered(expect))
if got := iteratedTriples(qs, it); !reflect.DeepEqual(got, expect) {
if got := iteratedQuads(qs, it); !reflect.DeepEqual(got, expect) {
t.Errorf("Failed to get expected results from predicate iterator, got:%v expect:%v", got, expect)
}
// Label iterator.
it = qs.TripleIterator(quad.Label, qs.ValueOf("status_graph"))
it = qs.QuadIterator(quad.Label, qs.ValueOf("status_graph"))
expect = []quad.Quad{
{"B", "status", "cool", "status_graph"},
@ -366,20 +376,20 @@ func TestSetIterator(t *testing.T) {
{"G", "status", "cool", "status_graph"},
}
sort.Sort(ordered(expect))
if got := iteratedTriples(qs, it); !reflect.DeepEqual(got, expect) {
if got := iteratedQuads(qs, it); !reflect.DeepEqual(got, expect) {
t.Errorf("Failed to get expected results from predicate iterator, got:%v expect:%v", got, expect)
}
it.Reset()
// Order is important
and = iterator.NewAnd()
and.AddSubIterator(qs.TripleIterator(quad.Subject, qs.ValueOf("B")))
and.AddSubIterator(qs.QuadIterator(quad.Subject, qs.ValueOf("B")))
and.AddSubIterator(it)
expect = []quad.Quad{
{"B", "status", "cool", "status_graph"},
}
if got := iteratedTriples(qs, and); !reflect.DeepEqual(got, expect) {
if got := iteratedQuads(qs, and); !reflect.DeepEqual(got, expect) {
t.Errorf("Failed to get confirm expected results, got:%v expect:%v", got, expect)
}
it.Reset()
@ -387,12 +397,12 @@ func TestSetIterator(t *testing.T) {
// Order is important
and = iterator.NewAnd()
and.AddSubIterator(it)
and.AddSubIterator(qs.TripleIterator(quad.Subject, qs.ValueOf("B")))
and.AddSubIterator(qs.QuadIterator(quad.Subject, qs.ValueOf("B")))
expect = []quad.Quad{
{"B", "status", "cool", "status_graph"},
}
if got := iteratedTriples(qs, and); !reflect.DeepEqual(got, expect) {
if got := iteratedQuads(qs, and); !reflect.DeepEqual(got, expect) {
t.Errorf("Failed to get confirm expected results, got:%v expect:%v", got, expect)
}
}
@ -405,13 +415,13 @@ func TestOptimize(t *testing.T) {
if err != nil {
t.Fatalf("Failed to create working directory")
}
qs, err := newTripleStore(tmpDir, nil)
qs, err := newQuadStore(tmpDir, nil)
if qs == nil || err != nil {
t.Error("Failed to create leveldb TripleStore.")
t.Error("Failed to create leveldb QuadStore.")
}
w, _ := writer.NewSingleReplication(qs, nil)
w.AddQuadSet(makeTripleSet())
w.AddQuadSet(makeQuadSet())
// With an linksto-fixed pair
fixed := qs.FixedIterator()
@ -428,9 +438,9 @@ func TestOptimize(t *testing.T) {
t.Errorf("Optimized iterator type does not match original, got:%v expect:%v", newIt.Type(), Type())
}
newTriples := iteratedTriples(qs, newIt)
oldTriples := iteratedTriples(qs, oldIt)
if !reflect.DeepEqual(newTriples, oldTriples) {
newQuads := iteratedQuads(qs, newIt)
oldQuads := iteratedQuads(qs, oldIt)
if !reflect.DeepEqual(newQuads, oldQuads) {
t.Errorf("Optimized iteration does not match original")
}

165
graph/leveldb/triplestore.go → graph/leveldb/quadstore.go
View file

@ -36,7 +36,7 @@ import (
)
func init() {
graph.RegisterTripleStore("leveldb", true, newTripleStore, createNewLevelDB)
graph.RegisterQuadStore("leveldb", true, newQuadStore, createNewLevelDB)
}
const (
@ -57,7 +57,7 @@ func (t Token) Key() interface{} {
return string(t)
}
type TripleStore struct {
type QuadStore struct {
dbOpts *opt.Options
db *leveldb.DB
path string
@ -72,11 +72,11 @@ func createNewLevelDB(path string, _ graph.Options) error {
opts := &opt.Options{}
db, err := leveldb.OpenFile(path, opts)
if err != nil {
glog.Errorf("Error: couldn't create database: %v", err)
glog.Errorf("Error: could not create database: %v", err)
return err
}
defer db.Close()
qs := &TripleStore{}
qs := &QuadStore{}
qs.db = db
qs.writeopts = &opt.WriteOptions{
Sync: true,
@ -85,31 +85,31 @@ func createNewLevelDB(path string, _ graph.Options) error {
return nil
}
func newTripleStore(path string, options graph.Options) (graph.TripleStore, error) {
var qs TripleStore
func newQuadStore(path string, options graph.Options) (graph.QuadStore, error) {
var qs QuadStore
var err error
qs.path = path
cache_size := DefaultCacheSize
cacheSize := DefaultCacheSize
if val, ok := options.IntKey("cache_size_mb"); ok {
cache_size = val
cacheSize = val
}
qs.dbOpts = &opt.Options{
BlockCache: cache.NewLRUCache(cache_size * opt.MiB),
BlockCache: cache.NewLRUCache(cacheSize * opt.MiB),
}
qs.dbOpts.ErrorIfMissing = true
write_buffer_mb := DefaultWriteBufferSize
if val, ok := options.IntKey("write_buffer_mb"); ok {
write_buffer_mb = val
writeBufferSize := DefaultWriteBufferSize
if val, ok := options.IntKey("writeBufferSize"); ok {
writeBufferSize = val
}
qs.dbOpts.WriteBuffer = write_buffer_mb * opt.MiB
qs.dbOpts.WriteBuffer = writeBufferSize * opt.MiB
qs.writeopts = &opt.WriteOptions{
Sync: false,
}
qs.readopts = &opt.ReadOptions{}
db, err := leveldb.OpenFile(qs.path, qs.dbOpts)
if err != nil {
glog.Errorln("Error, couldn't open! ", err)
glog.Errorln("Error, could not open! ", err)
return nil, err
}
qs.db = db
@ -121,7 +121,7 @@ func newTripleStore(path string, options graph.Options) (graph.TripleStore, erro
return &qs, nil
}
func (qs *TripleStore) GetStats() string {
func (qs *QuadStore) GetStats() string {
out := ""
stats, err := qs.db.GetProperty("leveldb.stats")
if err == nil {
@ -131,36 +131,39 @@ func (qs *TripleStore) GetStats() string {
return out
}
func (qs *TripleStore) Size() int64 {
func (qs *QuadStore) Size() int64 {
return qs.size
}
func (qs *TripleStore) Horizon() int64 {
func (qs *QuadStore) Horizon() int64 {
return qs.horizon
}
func (qa *TripleStore) createDeltaKeyFor(d graph.Delta) []byte {
key := make([]byte, 0, 19)
key = append(key, 'd')
key = append(key, []byte(fmt.Sprintf("%018x", d.ID))...)
func hashOf(s string) []byte {
h := hashPool.Get().(hash.Hash)
h.Reset()
defer hashPool.Put(h)
key := make([]byte, 0, hashSize)
h.Write([]byte(s))
key = h.Sum(key)
return key
}
func (qs *TripleStore) createKeyFor(d [4]quad.Direction, triple quad.Quad) []byte {
func (qs *QuadStore) createKeyFor(d [4]quad.Direction, q quad.Quad) []byte {
key := make([]byte, 0, 2+(hashSize*3))
// TODO(kortschak) Remove dependence on String() method.
key = append(key, []byte{d[0].Prefix(), d[1].Prefix()}...)
key = append(key, qs.convertStringToByteHash(triple.Get(d[0]))...)
key = append(key, qs.convertStringToByteHash(triple.Get(d[1]))...)
key = append(key, qs.convertStringToByteHash(triple.Get(d[2]))...)
key = append(key, qs.convertStringToByteHash(triple.Get(d[3]))...)
key = append(key, hashOf(q.Get(d[0]))...)
key = append(key, hashOf(q.Get(d[1]))...)
key = append(key, hashOf(q.Get(d[2]))...)
key = append(key, hashOf(q.Get(d[3]))...)
return key
}
func (qs *TripleStore) createValueKeyFor(s string) []byte {
func (qs *QuadStore) createValueKeyFor(s string) []byte {
key := make([]byte, 0, 1+hashSize)
key = append(key, []byte("z")...)
key = append(key, qs.convertStringToByteHash(s)...)
key = append(key, hashOf(s)...)
return key
}
@ -177,16 +180,16 @@ var (
cps = [4]quad.Direction{quad.Label, quad.Predicate, quad.Subject, quad.Object}
)
func (qs *TripleStore) ApplyDeltas(deltas []graph.Delta) error {
func (qs *QuadStore) ApplyDeltas(deltas []graph.Delta) error {
batch := &leveldb.Batch{}
resizeMap := make(map[string]int64)
size_change := int64(0)
sizeChange := int64(0)
for _, d := range deltas {
bytes, err := json.Marshal(d)
if err != nil {
return err
}
batch.Put(qs.createDeltaKeyFor(d), bytes)
batch.Put(keyFor(d), bytes)
err = qs.buildQuadWrite(batch, d.Quad, d.ID, d.Action == graph.Add)
if err != nil {
return err
@ -201,7 +204,7 @@ func (qs *TripleStore) ApplyDeltas(deltas []graph.Delta) error {
if d.Quad.Label != "" {
resizeMap[d.Quad.Label] += delta
}
size_change += delta
sizeChange += delta
qs.horizon = d.ID
}
for k, v := range resizeMap {
@ -214,18 +217,25 @@ func (qs *TripleStore) ApplyDeltas(deltas []graph.Delta) error {
}
err := qs.db.Write(batch, qs.writeopts)
if err != nil {
glog.Error("Couldn't write to DB for tripleset.")
glog.Error("could not write to DB for quadset.")
return err
}
qs.size += size_change
qs.size += sizeChange
return nil
}
func (qs *TripleStore) buildQuadWrite(batch *leveldb.Batch, q quad.Quad, id int64, isAdd bool) error {
func keyFor(d graph.Delta) []byte {
key := make([]byte, 0, 19)
key = append(key, 'd')
key = append(key, []byte(fmt.Sprintf("%018x", d.ID))...)
return key
}
func (qs *QuadStore) buildQuadWrite(batch *leveldb.Batch, q quad.Quad, id int64, isAdd bool) error {
var entry IndexEntry
data, err := qs.db.Get(qs.createKeyFor(spo, q), qs.readopts)
if err != nil && err != leveldb.ErrNotFound {
glog.Error("Couldn't access DB to prepare index: ", err)
glog.Error("could not access DB to prepare index: ", err)
return err
}
if err == nil {
@ -241,12 +251,12 @@ func (qs *TripleStore) buildQuadWrite(batch *leveldb.Batch, q quad.Quad, id int6
if isAdd && len(entry.History)%2 == 0 {
glog.Error("Entry History is out of sync for", entry)
return errors.New("Odd index history")
return errors.New("odd index history")
}
bytes, err := json.Marshal(entry)
if err != nil {
glog.Errorf("Couldn't write to buffer for entry %#v: %s", entry, err)
glog.Errorf("could not write to buffer for entry %#v: %s", entry, err)
return err
}
batch.Put(qs.createKeyFor(spo, q), bytes)
@ -263,7 +273,7 @@ type ValueData struct {
Size int64
}
func (qs *TripleStore) UpdateValueKeyBy(name string, amount int64, batch *leveldb.Batch) error {
func (qs *QuadStore) UpdateValueKeyBy(name string, amount int64, batch *leveldb.Batch) error {
value := &ValueData{name, amount}
key := qs.createValueKeyFor(name)
b, err := qs.db.Get(key, qs.readopts)
@ -278,7 +288,7 @@ func (qs *TripleStore) UpdateValueKeyBy(name string, amount int64, batch *leveld
if b != nil && err != leveldb.ErrNotFound {
err = json.Unmarshal(b, value)
if err != nil {
glog.Errorf("Error: couldn't reconstruct value: %v", err)
glog.Errorf("Error: could not reconstruct value: %v", err)
return err
}
value.Size += amount
@ -292,7 +302,7 @@ func (qs *TripleStore) UpdateValueKeyBy(name string, amount int64, batch *leveld
// Repackage and rewrite.
bytes, err := json.Marshal(&value)
if err != nil {
glog.Errorf("Couldn't write to buffer for value %s: %s", name, err)
glog.Errorf("could not write to buffer for value %s: %s", name, err)
return err
}
if batch == nil {
@ -303,85 +313,75 @@ func (qs *TripleStore) UpdateValueKeyBy(name string, amount int64, batch *leveld
return nil
}
func (qs *TripleStore) Close() {
func (qs *QuadStore) Close() {
buf := new(bytes.Buffer)
err := binary.Write(buf, binary.LittleEndian, qs.size)
if err == nil {
werr := qs.db.Put([]byte("__size"), buf.Bytes(), qs.writeopts)
if werr != nil {
glog.Error("Couldn't write size before closing!")
glog.Error("could not write size before closing!")
}
} else {
glog.Errorf("Couldn't convert size before closing!")
glog.Errorf("could not convert size before closing!")
}
buf.Reset()
err = binary.Write(buf, binary.LittleEndian, qs.horizon)
if err == nil {
werr := qs.db.Put([]byte("__horizon"), buf.Bytes(), qs.writeopts)
if werr != nil {
glog.Error("Couldn't write horizon before closing!")
glog.Error("could not write horizon before closing!")
}
} else {
glog.Errorf("Couldn't convert horizon before closing!")
glog.Errorf("could not convert horizon before closing!")
}
qs.db.Close()
qs.open = false
}
func (qs *TripleStore) Quad(k graph.Value) quad.Quad {
var triple quad.Quad
func (qs *QuadStore) Quad(k graph.Value) quad.Quad {
var q quad.Quad
b, err := qs.db.Get(k.(Token), qs.readopts)
if err != nil && err != leveldb.ErrNotFound {
glog.Error("Error: couldn't get triple from DB.")
glog.Error("Error: could not get quad from DB.")
return quad.Quad{}
}
if err == leveldb.ErrNotFound {
// No harm, no foul.
return quad.Quad{}
}
err = json.Unmarshal(b, &triple)
err = json.Unmarshal(b, &q)
if err != nil {
glog.Error("Error: couldn't reconstruct triple.")
glog.Error("Error: could not reconstruct quad.")
return quad.Quad{}
}
return triple
return q
}
func (qs *TripleStore) convertStringToByteHash(s string) []byte {
h := hashPool.Get().(hash.Hash)
h.Reset()
defer hashPool.Put(h)
key := make([]byte, 0, hashSize)
h.Write([]byte(s))
key = h.Sum(key)
return key
}
func (qs *TripleStore) ValueOf(s string) graph.Value {
func (qs *QuadStore) ValueOf(s string) graph.Value {
return Token(qs.createValueKeyFor(s))
}
func (qs *TripleStore) valueData(value_key []byte) ValueData {
func (qs *QuadStore) valueData(key []byte) ValueData {
var out ValueData
if glog.V(3) {
glog.V(3).Infof("%s %v", string(value_key[0]), value_key)
glog.V(3).Infof("%c %v", key[0], key)
}
b, err := qs.db.Get(value_key, qs.readopts)
b, err := qs.db.Get(key, qs.readopts)
if err != nil && err != leveldb.ErrNotFound {
glog.Errorln("Error: couldn't get value from DB")
glog.Errorln("Error: could not get value from DB")
return out
}
if b != nil && err != leveldb.ErrNotFound {
err = json.Unmarshal(b, &out)
if err != nil {
glog.Errorln("Error: couldn't reconstruct value")
glog.Errorln("Error: could not reconstruct value")
return ValueData{}
}
}
return out
}
func (qs *TripleStore) NameOf(k graph.Value) string {
func (qs *QuadStore) NameOf(k graph.Value) string {
if k == nil {
glog.V(2).Info("k was nil")
return ""
@ -389,18 +389,18 @@ func (qs *TripleStore) NameOf(k graph.Value) string {
return qs.valueData(k.(Token)).Name
}
func (qs *TripleStore) SizeOf(k graph.Value) int64 {
func (qs *QuadStore) SizeOf(k graph.Value) int64 {
if k == nil {
return 0
}
return int64(qs.valueData(k.(Token)).Size)
}
func (qs *TripleStore) getInt64ForKey(key string, empty int64) (int64, error) {
func (qs *QuadStore) getInt64ForKey(key string, empty int64) (int64, error) {
var out int64
b, err := qs.db.Get([]byte(key), qs.readopts)
if err != nil && err != leveldb.ErrNotFound {
glog.Errorln("Couldn't read " + key + ": " + err.Error())
glog.Errorln("could not read " + key + ": " + err.Error())
return 0, err
}
if err == leveldb.ErrNotFound {
@ -410,13 +410,13 @@ func (qs *TripleStore) getInt64ForKey(key string, empty int64) (int64, error) {
buf := bytes.NewBuffer(b)
err = binary.Read(buf, binary.LittleEndian, &out)
if err != nil {
glog.Errorln("Error: couldn't parse", key)
glog.Errorln("Error: could not parse", key)
return 0, err
}
return out, nil
}
func (qs *TripleStore) getMetadata() error {
func (qs *QuadStore) getMetadata() error {
var err error
qs.size, err = qs.getInt64ForKey("__size", 0)
if err != nil {
@ -426,7 +426,7 @@ func (qs *TripleStore) getMetadata() error {
return err
}
func (qs *TripleStore) SizeOfPrefix(pre []byte) (int64, error) {
func (qs *QuadStore) SizeOfPrefix(pre []byte) (int64, error) {
limit := make([]byte, len(pre))
copy(limit, pre)
end := len(limit) - 1
@ -441,7 +441,7 @@ func (qs *TripleStore) SizeOfPrefix(pre []byte) (int64, error) {
return 0, nil
}
func (qs *TripleStore) TripleIterator(d quad.Direction, val graph.Value) graph.Iterator {
func (qs *QuadStore) QuadIterator(d quad.Direction, val graph.Value) graph.Iterator {
var prefix string
switch d {
case quad.Subject:
@ -458,28 +458,27 @@ func (qs *TripleStore) TripleIterator(d quad.Direction, val graph.Value) graph.I
return NewIterator(prefix, d, val, qs)
}
func (qs *TripleStore) NodesAllIterator() graph.Iterator {
func (qs *QuadStore) NodesAllIterator() graph.Iterator {
return NewAllIterator("z", quad.Any, qs)
}
func (qs *TripleStore) TriplesAllIterator() graph.Iterator {
func (qs *QuadStore) QuadsAllIterator() graph.Iterator {
return NewAllIterator("po", quad.Predicate, qs)
}
func (qs *TripleStore) TripleDirection(val graph.Value, d quad.Direction) graph.Value {
func (qs *QuadStore) QuadDirection(val graph.Value, d quad.Direction) graph.Value {
v := val.(Token)
offset := PositionOf(v[0:2], d, qs)
if offset != -1 {
return Token(append([]byte("z"), v[offset:offset+hashSize]...))
} else {
return Token(qs.Quad(val).Get(d))
}
return Token(qs.Quad(val).Get(d))
}
func compareBytes(a, b graph.Value) bool {
return bytes.Equal(a.(Token), b.(Token))
}
func (qs *TripleStore) FixedIterator() graph.FixedIterator {
return iterator.NewFixedIteratorWithCompare(compareBytes)
func (qs *QuadStore) FixedIterator() graph.FixedIterator {
return iterator.NewFixed(compareBytes)
}

8
graph/leveldb/triplestore_iterator_optimize.go → graph/leveldb/quadstore_iterator_optimize.go
View file

@ -19,16 +19,16 @@ import (
"github.com/google/cayley/graph/iterator"
)
func (ts *TripleStore) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) {
func (qs *QuadStore) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) {
switch it.Type() {
case graph.LinksTo:
return ts.optimizeLinksTo(it.(*iterator.LinksTo))
return qs.optimizeLinksTo(it.(*iterator.LinksTo))
}
return it, false
}
func (ts *TripleStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool) {
func (qs *QuadStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool) {
subs := it.SubIterators()
if len(subs) != 1 {
return it, false
@ -41,7 +41,7 @@ func (ts *TripleStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bo
panic("unexpected size during optimize")
}
val := primary.Result()
newIt := ts.TripleIterator(it.Direction(), val)
newIt := qs.QuadIterator(it.Direction(), val)
nt := newIt.Tagger()
nt.CopyFrom(it)
for _, tag := range primary.Tagger().Tags() {

40
graph/memstore/all_iterator.go
View file

@ -21,48 +21,50 @@ import (
type AllIterator struct {
iterator.Int64
ts *TripleStore
qs *QuadStore
}
type NodesAllIterator AllIterator
type QuadsAllIterator AllIterator
type (
nodesAllIterator AllIterator
quadsAllIterator AllIterator
)
func NewMemstoreNodesAllIterator(ts *TripleStore) *NodesAllIterator {
var out NodesAllIterator
out.Int64 = *iterator.NewInt64(1, ts.idCounter-1)
out.ts = ts
func newNodesAllIterator(qs *QuadStore) *nodesAllIterator {
var out nodesAllIterator
out.Int64 = *iterator.NewInt64(1, qs.nextID-1)
out.qs = qs
return &out
}
// No subiterators.
func (it *NodesAllIterator) SubIterators() []graph.Iterator {
func (it *nodesAllIterator) SubIterators() []graph.Iterator {
return nil
}
func (it *NodesAllIterator) Next() bool {
func (it *nodesAllIterator) Next() bool {
if !it.Int64.Next() {
return false
}
_, ok := it.ts.revIdMap[it.Int64.Result().(int64)]
_, ok := it.qs.revIDMap[it.Int64.Result().(int64)]
if !ok {
return it.Next()
}
return true
}
func NewMemstoreQuadsAllIterator(ts *TripleStore) *QuadsAllIterator {
var out QuadsAllIterator
out.Int64 = *iterator.NewInt64(1, ts.quadIdCounter-1)
out.ts = ts
func newQuadsAllIterator(qs *QuadStore) *quadsAllIterator {
var out quadsAllIterator
out.Int64 = *iterator.NewInt64(1, qs.nextQuadID-1)
out.qs = qs
return &out
}
func (qit *QuadsAllIterator) Next() bool {
out := qit.Int64.Next()
func (it *quadsAllIterator) Next() bool {
out := it.Int64.Next()
if out {
i64 := qit.Int64.Result().(int64)
if qit.ts.log[i64].DeletedBy != 0 || qit.ts.log[i64].Action == graph.Delete {
return qit.Next()
i64 := it.Int64.Result().(int64)
if it.qs.log[i64].DeletedBy != 0 || it.qs.log[i64].Action == graph.Delete {
return it.Next()
}
}
return out

10
graph/memstore/iterator.go
View file

@ -26,7 +26,7 @@ import (
type Iterator struct {
uid uint64
ts *TripleStore
qs *QuadStore
tags graph.Tagger
tree *b.Tree
iter *b.Enumerator
@ -38,14 +38,14 @@ func cmp(a, b int64) int {
return int(a - b)
}
func NewIterator(tree *b.Tree, data string, ts *TripleStore) *Iterator {
func NewIterator(tree *b.Tree, data string, qs *QuadStore) *Iterator {
iter, err := tree.SeekFirst()
if err != nil {
iter = nil
}
return &Iterator{
uid: iterator.NextUID(),
ts: ts,
qs: qs,
tree: tree,
iter: iter,
data: data,
@ -96,7 +96,7 @@ func (it *Iterator) Clone() graph.Iterator {
m := &Iterator{
uid: iterator.NextUID(),
ts: it.ts,
qs: it.qs,
tree: it.tree,
iter: iter,
data: it.data,
@ -109,7 +109,7 @@ func (it *Iterator) Clone() graph.Iterator {
func (it *Iterator) Close() {}
func (it *Iterator) checkValid(index int64) bool {
return it.ts.log[index].DeletedBy == 0
return it.qs.log[index].DeletedBy == 0
}
func (it *Iterator) Next() bool {

136
graph/memstore/triplestore.go → graph/memstore/quadstore.go
View file

@ -26,8 +26,8 @@ import (
)
func init() {
graph.RegisterTripleStore("memstore", false, func(string, graph.Options) (graph.TripleStore, error) {
return newTripleStore(), nil
graph.RegisterQuadStore("memstore", false, func(string, graph.Options) (graph.QuadStore, error) {
return newQuadStore(), nil
}, nil)
}
@ -69,38 +69,38 @@ type LogEntry struct {
DeletedBy int64
}
type TripleStore struct {
idCounter int64
quadIdCounter int64
idMap map[string]int64
revIdMap map[int64]string
log []LogEntry
size int64
index QuadDirectionIndex
type QuadStore struct {
nextID int64
nextQuadID int64
idMap map[string]int64
revIDMap map[int64]string
log []LogEntry
size int64
index QuadDirectionIndex
// vip_index map[string]map[int64]map[string]map[int64]*b.Tree
}
func newTripleStore() *TripleStore {
return &TripleStore{
func newQuadStore() *QuadStore {
return &QuadStore{
idMap: make(map[string]int64),
revIdMap: make(map[int64]string),
revIDMap: make(map[int64]string),
// Sentinel null entry so indices start at 1
log: make([]LogEntry, 1, 200),
index: NewQuadDirectionIndex(),
idCounter: 1,
quadIdCounter: 1,
index: NewQuadDirectionIndex(),
nextID: 1,
nextQuadID: 1,
}
}
func (ts *TripleStore) ApplyDeltas(deltas []graph.Delta) error {
func (qs *QuadStore) ApplyDeltas(deltas []graph.Delta) error {
for _, d := range deltas {
var err error
if d.Action == graph.Add {
err = ts.AddDelta(d)
err = qs.AddDelta(d)
} else {
err = ts.RemoveDelta(d)
err = qs.RemoveDelta(d)
}
if err != nil {
return err
@ -111,7 +111,7 @@ func (ts *TripleStore) ApplyDeltas(deltas []graph.Delta) error {
const maxInt = int(^uint(0) >> 1)
func (ts *TripleStore) indexOf(t quad.Quad) (int64, bool) {
func (qs *QuadStore) indexOf(t quad.Quad) (int64, bool) {
min := maxInt
var tree *b.Tree
for d := quad.Subject; d <= quad.Label; d++ {
@ -119,12 +119,12 @@ func (ts *TripleStore) indexOf(t quad.Quad) (int64, bool) {
if d == quad.Label && sid == "" {
continue
}
id, ok := ts.idMap[sid]
id, ok := qs.idMap[sid]
// If we've never heard about a node, it must not exist
if !ok {
return 0, false
}
index, ok := ts.index.Get(d, id)
index, ok := qs.index.Get(d, id)
if !ok {
// If it's never been indexed in this direction, it can't exist.
return 0, false
@ -133,35 +133,35 @@ func (ts *TripleStore) indexOf(t quad.Quad) (int64, bool) {
min, tree = l, index
}
}
it := NewIterator(tree, "", ts)
it := NewIterator(tree, "", qs)
for it.Next() {
val := it.Result()
if t == ts.log[val.(int64)].Quad {
if t == qs.log[val.(int64)].Quad {
return val.(int64), true
}
}
return 0, false
}
func (ts *TripleStore) AddDelta(d graph.Delta) error {
if _, exists := ts.indexOf(d.Quad); exists {
func (qs *QuadStore) AddDelta(d graph.Delta) error {
if _, exists := qs.indexOf(d.Quad); exists {
return graph.ErrQuadExists
}
qid := ts.quadIdCounter
ts.log = append(ts.log, LogEntry{Delta: d})
ts.size++
ts.quadIdCounter++
qid := qs.nextQuadID
qs.log = append(qs.log, LogEntry{Delta: d})
qs.size++
qs.nextQuadID++
for dir := quad.Subject; dir <= quad.Label; dir++ {
sid := d.Quad.Get(dir)
if dir == quad.Label && sid == "" {
continue
}
if _, ok := ts.idMap[sid]; !ok {
ts.idMap[sid] = ts.idCounter
ts.revIdMap[ts.idCounter] = sid
ts.idCounter++
if _, ok := qs.idMap[sid]; !ok {
qs.idMap[sid] = qs.nextID
qs.revIDMap[qs.nextID] = sid
qs.nextID++
}
}
@ -169,8 +169,8 @@ func (ts *TripleStore) AddDelta(d graph.Delta) error {
if dir == quad.Label && d.Quad.Get(dir) == "" {
continue
}
id := ts.idMap[d.Quad.Get(dir)]
tree := ts.index.Tree(dir, id)
id := qs.idMap[d.Quad.Get(dir)]
tree := qs.index.Tree(dir, id)
tree.Set(qid, struct{}{})
}
@ -178,43 +178,43 @@ func (ts *TripleStore) AddDelta(d graph.Delta) error {
return nil
}
func (ts *TripleStore) RemoveDelta(d graph.Delta) error {
prevQuadID, exists := ts.indexOf(d.Quad)
func (qs *QuadStore) RemoveDelta(d graph.Delta) error {
prevQuadID, exists := qs.indexOf(d.Quad)
if !exists {
return graph.ErrQuadNotExist
}
quadID := ts.quadIdCounter
ts.log = append(ts.log, LogEntry{Delta: d})
ts.log[prevQuadID].DeletedBy = quadID
ts.size--
ts.quadIdCounter++
quadID := qs.nextQuadID
qs.log = append(qs.log, LogEntry{Delta: d})
qs.log[prevQuadID].DeletedBy = quadID
qs.size--
qs.nextQuadID++
return nil
}
func (ts *TripleStore) Quad(index graph.Value) quad.Quad {
return ts.log[index.(int64)].Quad
func (qs *QuadStore) Quad(index graph.Value) quad.Quad {
return qs.log[index.(int64)].Quad
}
func (ts *TripleStore) TripleIterator(d quad.Direction, value graph.Value) graph.Iterator {
index, ok := ts.index.Get(d, value.(int64))
func (qs *QuadStore) QuadIterator(d quad.Direction, value graph.Value) graph.Iterator {
index, ok := qs.index.Get(d, value.(int64))
data := fmt.Sprintf("dir:%s val:%d", d, value.(int64))
if ok {
return NewIterator(index, data, ts)
return NewIterator(index, data, qs)
}
return &iterator.Null{}
}
func (ts *TripleStore) Horizon() int64 {
return ts.log[len(ts.log)-1].ID
func (qs *QuadStore) Horizon() int64 {
return qs.log[len(qs.log)-1].ID
}
func (ts *TripleStore) Size() int64 {
return ts.size
func (qs *QuadStore) Size() int64 {
return qs.size
}
func (ts *TripleStore) DebugPrint() {
for i, l := range ts.log {
func (qs *QuadStore) DebugPrint() {
for i, l := range qs.log {
if i == 0 {
continue
}
@ -222,29 +222,29 @@ func (ts *TripleStore) DebugPrint() {
}
}
func (ts *TripleStore) ValueOf(name string) graph.Value {
return ts.idMap[name]
func (qs *QuadStore) ValueOf(name string) graph.Value {
return qs.idMap[name]
}
func (ts *TripleStore) NameOf(id graph.Value) string {
return ts.revIdMap[id.(int64)]
func (qs *QuadStore) NameOf(id graph.Value) string {
return qs.revIDMap[id.(int64)]
}
func (ts *TripleStore) TriplesAllIterator() graph.Iterator {
return NewMemstoreQuadsAllIterator(ts)
func (qs *QuadStore) QuadsAllIterator() graph.Iterator {
return newQuadsAllIterator(qs)
}
func (ts *TripleStore) FixedIterator() graph.FixedIterator {
return iterator.NewFixedIteratorWithCompare(iterator.BasicEquality)
func (qs *QuadStore) FixedIterator() graph.FixedIterator {
return iterator.NewFixed(iterator.Identity)
}
func (ts *TripleStore) TripleDirection(val graph.Value, d quad.Direction) graph.Value {
name := ts.Quad(val).Get(d)
return ts.ValueOf(name)
func (qs *QuadStore) QuadDirection(val graph.Value, d quad.Direction) graph.Value {
name := qs.Quad(val).Get(d)
return qs.ValueOf(name)
}
func (ts *TripleStore) NodesAllIterator() graph.Iterator {
return NewMemstoreNodesAllIterator(ts)
func (qs *QuadStore) NodesAllIterator() graph.Iterator {
return newNodesAllIterator(qs)
}
func (ts *TripleStore) Close() {}
func (qs *QuadStore) Close() {}

8
graph/memstore/triplestore_iterator_optimize.go → graph/memstore/quadstore_iterator_optimize.go
View file

@ -19,16 +19,16 @@ import (
"github.com/google/cayley/graph/iterator"
)
func (ts *TripleStore) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) {
func (qs *QuadStore) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) {
switch it.Type() {
case graph.LinksTo:
return ts.optimizeLinksTo(it.(*iterator.LinksTo))
return qs.optimizeLinksTo(it.(*iterator.LinksTo))
}
return it, false
}
func (ts *TripleStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool) {
func (qs *QuadStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool) {
subs := it.SubIterators()
if len(subs) != 1 {
return it, false
@ -41,7 +41,7 @@ func (ts *TripleStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bo
panic("unexpected size during optimize")
}
val := primary.Result()
newIt := ts.TripleIterator(it.Direction(), val)
newIt := qs.QuadIterator(it.Direction(), val)
nt := newIt.Tagger()
nt.CopyFrom(it)
for _, tag := range primary.Tagger().Tags() {

81
graph/memstore/triplestore_test.go → graph/memstore/quadstore_test.go
View file

@ -52,14 +52,14 @@ var simpleGraph = []quad.Quad{
{"G", "status", "cool", "status_graph"},
}
func makeTestStore(data []quad.Quad) (*TripleStore, graph.QuadWriter, []pair) {
func makeTestStore(data []quad.Quad) (*QuadStore, graph.QuadWriter, []pair) {
seen := make(map[string]struct{})
ts := newTripleStore()
qs := newQuadStore()
var (
val int64
ind []pair
)
writer, _ := writer.NewSingleReplication(ts, nil)
writer, _ := writer.NewSingleReplication(qs, nil)
for _, t := range data {
for _, qp := range []string{t.Subject, t.Predicate, t.Object, t.Label} {
if _, ok := seen[qp]; !ok && qp != "" {
@ -71,7 +71,7 @@ func makeTestStore(data []quad.Quad) (*TripleStore, graph.QuadWriter, []pair) {
writer.AddQuad(t)
}
return ts, writer, ind
return qs, writer, ind
}
type pair struct {
@ -80,12 +80,12 @@ type pair struct {
}
func TestMemstore(t *testing.T) {
ts, _, index := makeTestStore(simpleGraph)
if size := ts.Size(); size != int64(len(simpleGraph)) {
t.Errorf("Triple store has unexpected size, got:%d expected %d", size, len(simpleGraph))
qs, _, index := makeTestStore(simpleGraph)
if size := qs.Size(); size != int64(len(simpleGraph)) {
t.Errorf("Quad store has unexpected size, got:%d expected %d", size, len(simpleGraph))
}
for _, test := range index {
v := ts.ValueOf(test.query)
v := qs.ValueOf(test.query)
switch v := v.(type) {
default:
t.Errorf("ValueOf(%q) returned unexpected type, got:%T expected int64", test.query, v)
@ -98,21 +98,21 @@ func TestMemstore(t *testing.T) {
}
func TestIteratorsAndNextResultOrderA(t *testing.T) {
ts, _, _ := makeTestStore(simpleGraph)
qs, _, _ := makeTestStore(simpleGraph)
fixed := ts.FixedIterator()
fixed.Add(ts.ValueOf("C"))
fixed := qs.FixedIterator()
fixed.Add(qs.ValueOf("C"))
fixed2 := ts.FixedIterator()
fixed2.Add(ts.ValueOf("follows"))
fixed2 := qs.FixedIterator()
fixed2.Add(qs.ValueOf("follows"))
all := ts.NodesAllIterator()
all := qs.NodesAllIterator()
innerAnd := iterator.NewAnd()
innerAnd.AddSubIterator(iterator.NewLinksTo(ts, fixed2, quad.Predicate))
innerAnd.AddSubIterator(iterator.NewLinksTo(ts, all, quad.Object))
innerAnd.AddSubIterator(iterator.NewLinksTo(qs, fixed2, quad.Predicate))
innerAnd.AddSubIterator(iterator.NewLinksTo(qs, all, quad.Object))
hasa := iterator.NewHasA(ts, innerAnd, quad.Subject)
hasa := iterator.NewHasA(qs, innerAnd, quad.Subject)
outerAnd := iterator.NewAnd()
outerAnd.AddSubIterator(fixed)
outerAnd.AddSubIterator(hasa)
@ -121,8 +121,8 @@ func TestIteratorsAndNextResultOrderA(t *testing.T) {
t.Error("Expected one matching subtree")
}
val := outerAnd.Result()
if ts.NameOf(val) != "C" {
t.Errorf("Matching subtree should be %s, got %s", "barak", ts.NameOf(val))
if qs.NameOf(val) != "C" {
t.Errorf("Matching subtree should be %s, got %s", "barak", qs.NameOf(val))
}
var (
@ -130,7 +130,7 @@ func TestIteratorsAndNextResultOrderA(t *testing.T) {
expect = []string{"B", "D"}
)
for {
got = append(got, ts.NameOf(all.Result()))
got = append(got, qs.NameOf(all.Result()))
if !outerAnd.NextPath() {
break
}
@ -147,12 +147,12 @@ func TestIteratorsAndNextResultOrderA(t *testing.T) {
}
func TestLinksToOptimization(t *testing.T) {
ts, _, _ := makeTestStore(simpleGraph)
qs, _, _ := makeTestStore(simpleGraph)
fixed := ts.FixedIterator()
fixed.Add(ts.ValueOf("cool"))
fixed := qs.FixedIterator()
fixed.Add(qs.ValueOf("cool"))
lto := iterator.NewLinksTo(ts, fixed, quad.Object)
lto := iterator.NewLinksTo(qs, fixed, quad.Object)
lto.Tagger().Add("foo")
newIt, changed := lto.Optimize()
@ -164,32 +164,37 @@ func TestLinksToOptimization(t *testing.T) {
}
v := newIt.(*Iterator)
v_clone := v.Clone()
if v_clone.DebugString(0) != v.DebugString(0) {
t.Fatal("Wrong iterator. Got ", v_clone.DebugString(0))
vClone := v.Clone()
if vClone.DebugString(0) != v.DebugString(0) {
t.Fatal("Wrong iterator. Got ", vClone.DebugString(0))
}
vt := v_clone.Tagger()
vt := vClone.Tagger()
if len(vt.Tags()) < 1 || vt.Tags()[0] != "foo" {
t.Fatal("Tag on LinksTo did not persist")
}
}
func TestRemoveTriple(t *testing.T) {
ts, w, _ := makeTestStore(simpleGraph)
func TestRemoveQuad(t *testing.T) {
qs, w, _ := makeTestStore(simpleGraph)
w.RemoveQuad(quad.Quad{"E", "follows", "F", ""})
w.RemoveQuad(quad.Quad{
Subject: "E",
Predicate: "follows",
Object: "F",
Label: "",
})
fixed := ts.FixedIterator()
fixed.Add(ts.ValueOf("E"))
fixed := qs.FixedIterator()
fixed.Add(qs.ValueOf("E"))
fixed2 := ts.FixedIterator()
fixed2.Add(ts.ValueOf("follows"))
fixed2 := qs.FixedIterator()
fixed2.Add(qs.ValueOf("follows"))
innerAnd := iterator.NewAnd()
innerAnd.AddSubIterator(iterator.NewLinksTo(ts, fixed, quad.Subject))
innerAnd.AddSubIterator(iterator.NewLinksTo(ts, fixed2, quad.Predicate))
innerAnd.AddSubIterator(iterator.NewLinksTo(qs, fixed, quad.Subject))
innerAnd.AddSubIterator(iterator.NewLinksTo(qs, fixed2, quad.Predicate))
hasa := iterator.NewHasA(ts, innerAnd, quad.Object)
hasa := iterator.NewHasA(qs, innerAnd, quad.Object)
newIt, _ := hasa.Optimize()
if graph.Next(newIt) {

10
graph/mongo/iterator.go
View file

@ -30,7 +30,7 @@ import (
type Iterator struct {
uid uint64
tags graph.Tagger
qs *TripleStore
qs *QuadStore
dir quad.Direction
iter *mgo.Iter
hash string
@ -42,7 +42,7 @@ type Iterator struct {
result graph.Value
}
func NewIterator(qs *TripleStore, collection string, d quad.Direction, val graph.Value) *Iterator {
func NewIterator(qs *QuadStore, collection string, d quad.Direction, val graph.Value) *Iterator {
name := qs.NameOf(val)
constraint := bson.M{d.String(): name}
@ -68,7 +68,7 @@ func NewIterator(qs *TripleStore, collection string, d quad.Direction, val graph
}
}
func NewAllIterator(qs *TripleStore, collection string) *Iterator {
func NewAllIterator(qs *QuadStore, collection string) *Iterator {
size, err := qs.db.C(collection).Count()
if err != nil {
// FIXME(kortschak) This should be passed back rather than just logging.
@ -130,7 +130,7 @@ func (it *Iterator) Clone() graph.Iterator {
func (it *Iterator) Next() bool {
var result struct {
Id string `bson:"_id"`
ID string `bson:"_id"`
Added []int64 `bson:"Added"`
Deleted []int64 `bson:"Deleted"`
}
@ -145,7 +145,7 @@ func (it *Iterator) Next() bool {
if it.collection == "quads" && len(result.Added) <= len(result.Deleted) {
return it.Next()
}
it.result = result.Id
it.result = result.ID
return true
}

23
graph/mongo/lru.go
View file

@ -18,45 +18,48 @@ import (
"container/list"
)
type IDLru struct {
// TODO(kortschak) Reimplement without container/list.
// cache implements an LRU cache.
type cache struct {
cache map[string]*list.Element
priority *list.List
maxSize int
}
type KV struct {
type kv struct {
key string
value string
}
func NewIDLru(size int) *IDLru {
var lru IDLru
func newCache(size int) *cache {
var lru cache
lru.maxSize = size
lru.priority = list.New()
lru.cache = make(map[string]*list.Element)
return &lru
}
func (lru *IDLru) Put(key string, value string) {
func (lru *cache) Put(key string, value string) {
if _, ok := lru.Get(key); ok {
return
}
if len(lru.cache) == lru.maxSize {
lru.removeOldest()
}
lru.priority.PushFront(KV{key: key, value: value})
lru.priority.PushFront(kv{key: key, value: value})
lru.cache[key] = lru.priority.Front()
}
func (lru *IDLru) Get(key string) (string, bool) {
func (lru *cache) Get(key string) (string, bool) {
if element, ok := lru.cache[key]; ok {
lru.priority.MoveToFront(element)
return element.Value.(KV).value, true
return element.Value.(kv).value, true
}
return "", false
}
func (lru *IDLru) removeOldest() {
func (lru *cache) removeOldest() {
last := lru.priority.Remove(lru.priority.Back())
delete(lru.cache, last.(KV).key)
delete(lru.cache, last.(kv).key)
}

80
graph/mongo/triplestore.go → graph/mongo/quadstore.go
View file

@ -30,7 +30,7 @@ import (
)
func init() {
graph.RegisterTripleStore("mongo", true, newTripleStore, createNewMongoGraph)
graph.RegisterQuadStore("mongo", true, newQuadStore, createNewMongoGraph)
}
const DefaultDBName = "cayley"
@ -42,10 +42,10 @@ var (
hashSize = sha1.Size
)
type TripleStore struct {
type QuadStore struct {
session *mgo.Session
db *mgo.Database
idCache *IDLru
ids *cache
}
func createNewMongoGraph(addr string, options graph.Options) error {
@ -84,8 +84,8 @@ func createNewMongoGraph(addr string, options graph.Options) error {
return nil
}
func newTripleStore(addr string, options graph.Options) (graph.TripleStore, error) {
var qs TripleStore
func newQuadStore(addr string, options graph.Options) (graph.QuadStore, error) {
var qs QuadStore
conn, err := mgo.Dial(addr)
if err != nil {
return nil, err
@ -97,19 +97,19 @@ func newTripleStore(addr string, options graph.Options) (graph.TripleStore, erro
}
qs.db = conn.DB(dbName)
qs.session = conn
qs.idCache = NewIDLru(1 << 16)
qs.ids = newCache(1 << 16)
return &qs, nil
}
func (qs *TripleStore) getIdForQuad(t quad.Quad) string {
id := qs.convertStringToByteHash(t.Subject)
id += qs.convertStringToByteHash(t.Predicate)
id += qs.convertStringToByteHash(t.Object)
id += qs.convertStringToByteHash(t.Label)
func (qs *QuadStore) getIDForQuad(t quad.Quad) string {
id := hashOf(t.Subject)
id += hashOf(t.Predicate)
id += hashOf(t.Object)
id += hashOf(t.Label)
return id
}
func (qs *TripleStore) convertStringToByteHash(s string) string {
func hashOf(s string) string {
h := hashPool.Get().(hash.Hash)
h.Reset()
defer hashPool.Put(h)
@ -121,7 +121,7 @@ func (qs *TripleStore) convertStringToByteHash(s string) string {
}
type MongoNode struct {
Id string `bson:"_id"`
ID string `bson:"_id"`
Name string `bson:"Name"`
Size int `bson:"Size"`
}
@ -133,11 +133,11 @@ type MongoLogEntry struct {
Timestamp int64
}
func (qs *TripleStore) updateNodeBy(node_name string, inc int) error {
node := qs.ValueOf(node_name)
func (qs *QuadStore) updateNodeBy(name string, inc int) error {
node := qs.ValueOf(name)
doc := bson.M{
"_id": node.(string),
"Name": node_name,
"Name": name,
}
upsert := bson.M{
"$setOnInsert": doc,
@ -153,7 +153,7 @@ func (qs *TripleStore) updateNodeBy(node_name string, inc int) error {
return err
}
func (qs *TripleStore) updateQuad(q quad.Quad, id int64, proc graph.Procedure) error {
func (qs *QuadStore) updateQuad(q quad.Quad, id int64, proc graph.Procedure) error {
var setname string
if proc == graph.Add {
setname = "Added"
@ -166,14 +166,14 @@ func (qs *TripleStore) updateQuad(q quad.Quad, id int64, proc graph.Procedure) e
setname: id,
},
}
_, err := qs.db.C("quads").UpsertId(qs.getIdForQuad(q), upsert)
_, err := qs.db.C("quads").UpsertId(qs.getIDForQuad(q), upsert)
if err != nil {
glog.Errorf("Error: %v", err)
}
return err
}
func (qs *TripleStore) checkValid(key string) bool {
func (qs *QuadStore) checkValid(key string) bool {
var indexEntry struct {
Added []int64 `bson:"Added"`
Deleted []int64 `bson:"Deleted"`
@ -192,7 +192,7 @@ func (qs *TripleStore) checkValid(key string) bool {
return true
}
func (qs *TripleStore) updateLog(d graph.Delta) error {
func (qs *QuadStore) updateLog(d graph.Delta) error {
var action string
if d.Action == graph.Add {
action = "Add"
@ -202,7 +202,7 @@ func (qs *TripleStore) updateLog(d graph.Delta) error {
entry := MongoLogEntry{
LogID: d.ID,
Action: action,
Key: qs.getIdForQuad(d.Quad),
Key: qs.getIDForQuad(d.Quad),
Timestamp: d.Timestamp.UnixNano(),
}
err := qs.db.C("log").Insert(entry)
@ -212,12 +212,12 @@ func (qs *TripleStore) updateLog(d graph.Delta) error {
return err
}
func (qs *TripleStore) ApplyDeltas(in []graph.Delta) error {
func (qs *QuadStore) ApplyDeltas(in []graph.Delta) error {
qs.session.SetSafe(nil)
ids := make(map[string]int)
// Pre-check the existence condition.
for _, d := range in {
key := qs.getIdForQuad(d.Quad)
key := qs.getIDForQuad(d.Quad)
switch d.Action {
case graph.Add:
if qs.checkValid(key) {
@ -266,7 +266,7 @@ func (qs *TripleStore) ApplyDeltas(in []graph.Delta) error {
return nil
}
func (qs *TripleStore) Quad(val graph.Value) quad.Quad {
func (qs *QuadStore) Quad(val graph.Value) quad.Quad {
var q quad.Quad
err := qs.db.C("quads").FindId(val.(string)).One(&q)
if err != nil {
@ -275,24 +275,24 @@ func (qs *TripleStore) Quad(val graph.Value) quad.Quad {
return q
}
func (qs *TripleStore) TripleIterator(d quad.Direction, val graph.Value) graph.Iterator {
func (qs *QuadStore) QuadIterator(d quad.Direction, val graph.Value) graph.Iterator {
return NewIterator(qs, "quads", d, val)
}
func (qs *TripleStore) NodesAllIterator() graph.Iterator {
func (qs *QuadStore) NodesAllIterator() graph.Iterator {
return NewAllIterator(qs, "nodes")
}
func (qs *TripleStore) TriplesAllIterator() graph.Iterator {
func (qs *QuadStore) QuadsAllIterator() graph.Iterator {
return NewAllIterator(qs, "quads")
}
func (qs *TripleStore) ValueOf(s string) graph.Value {
return qs.convertStringToByteHash(s)
func (qs *QuadStore) ValueOf(s string) graph.Value {
return hashOf(s)
}
func (qs *TripleStore) NameOf(v graph.Value) string {
val, ok := qs.idCache.Get(v.(string))
func (qs *QuadStore) NameOf(v graph.Value) string {
val, ok := qs.ids.Get(v.(string))
if ok {
return val
}
@ -301,11 +301,11 @@ func (qs *TripleStore) NameOf(v graph.Value) string {
if err != nil {
glog.Errorf("Error: Couldn't retrieve node %s %v", v, err)
}
qs.idCache.Put(v.(string), node.Name)
qs.ids.Put(v.(string), node.Name)
return node.Name
}
func (qs *TripleStore) Size() int64 {
func (qs *QuadStore) Size() int64 {
// TODO(barakmich): Make size real; store it in the log, and retrieve it.
count, err := qs.db.C("quads").Count()
if err != nil {
@ -315,7 +315,7 @@ func (qs *TripleStore) Size() int64 {
return int64(count)
}
func (qs *TripleStore) Horizon() int64 {
func (qs *QuadStore) Horizon() int64 {
var log MongoLogEntry
err := qs.db.C("log").Find(nil).Sort("-LogID").One(&log)
if err != nil {
@ -327,19 +327,15 @@ func (qs *TripleStore) Horizon() int64 {
return log.LogID
}
func compareStrings(a, b graph.Value) bool {
return a.(string) == b.(string)
func (qs *QuadStore) FixedIterator() graph.FixedIterator {
return iterator.NewFixed(iterator.Identity)
}
func (qs *TripleStore) FixedIterator() graph.FixedIterator {
return iterator.NewFixedIteratorWithCompare(compareStrings)
}
func (qs *TripleStore) Close() {
func (qs *QuadStore) Close() {
qs.db.Session.Close()
}
func (qs *TripleStore) TripleDirection(in graph.Value, d quad.Direction) graph.Value {
func (qs *QuadStore) QuadDirection(in graph.Value, d quad.Direction) graph.Value {
// Maybe do the trick here
var offset int
switch d {

8
graph/mongo/triplestore_iterator_optimize.go → graph/mongo/quadstore_iterator_optimize.go
View file

@ -19,16 +19,16 @@ import (
"github.com/google/cayley/graph/iterator"
)
func (ts *TripleStore) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) {
func (qs *QuadStore) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) {
switch it.Type() {
case graph.LinksTo:
return ts.optimizeLinksTo(it.(*iterator.LinksTo))
return qs.optimizeLinksTo(it.(*iterator.LinksTo))
}
return it, false
}
func (ts *TripleStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool) {
func (qs *QuadStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool) {
subs := it.SubIterators()
if len(subs) != 1 {
return it, false
@ -41,7 +41,7 @@ func (ts *TripleStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bo
panic("unexpected size during optimize")
}
val := primary.Result()
newIt := ts.TripleIterator(it.Direction(), val)
newIt := qs.QuadIterator(it.Direction(), val)
nt := newIt.Tagger()
nt.CopyFrom(it)
for _, tag := range primary.Tagger().Tags() {

62
graph/triplestore.go → graph/quadstore.go
View file

@ -14,12 +14,12 @@
package graph
// Defines the TripleStore interface. Every backing store must implement at
// Defines the QuadStore interface. Every backing store must implement at
// least this interface.
//
// Most of these are pretty straightforward. As long as we can surface this
// interface, the rest of the stack will "just work" and we can connect to any
// triple backing store we prefer.
// quad backing store we prefer.
import (
"errors"
@ -28,71 +28,73 @@ import (
"github.com/google/cayley/quad"
)
// Value defines an opaque "triple store value" type. However the backend wishes
// to implement it, a Value is merely a token to a triple or a node that the
// Value defines an opaque "quad store value" type. However the backend wishes
// to implement it, a Value is merely a token to a quad or a node that the
// backing store itself understands, and the base iterators pass around.
//
// For example, in a very traditional, graphd-style graph, these are int64s
// (guids of the primitives). In a very direct sort of graph, these could be
// pointers to structs, or merely triples, or whatever works best for the
// pointers to structs, or merely quads, or whatever works best for the
// backing store.
//
// These must be comparable, or implement a `Key() interface{}` function
// so that they may be stored in maps.
type Value interface{}
type TripleStore interface {
type QuadStore interface {
// The only way in is through building a transaction, which
// is done by a replication strategy.
ApplyDeltas([]Delta) error
// Given an opaque token, returns the triple for that token from the store.
// Given an opaque token, returns the quad for that token from the store.
Quad(Value) quad.Quad
// Given a direction and a token, creates an iterator of links which have
// that node token in that directional field.
TripleIterator(quad.Direction, Value) Iterator
QuadIterator(quad.Direction, Value) Iterator
// Returns an iterator enumerating all nodes in the graph.
NodesAllIterator() Iterator
// Returns an iterator enumerating all links in the graph.
TriplesAllIterator() Iterator
QuadsAllIterator() Iterator
// Given a node ID, return the opaque token used by the TripleStore
// Given a node ID, return the opaque token used by the QuadStore
// to represent that id.
ValueOf(string) Value
// Given an opaque token, return the node that it represents.
NameOf(Value) string
// Returns the number of triples currently stored.
// Returns the number of quads currently stored.
Size() int64
// The last replicated transaction ID that this triplestore has verified.
// The last replicated transaction ID that this quadstore has verified.
Horizon() int64
// Creates a fixed iterator which can compare Values
FixedIterator() FixedIterator
// Optimize an iterator in the context of the triple store.
// Optimize an iterator in the context of the quad store.
// Suppose we have a better index for the passed tree; this
// gives the TripleStore the opportunity to replace it
// gives the QuadStore the opportunity to replace it
// with a more efficient iterator.
OptimizeIterator(it Iterator) (Iterator, bool)
// Close the triple store and clean up. (Flush to disk, cleanly
// Close the quad store and clean up. (Flush to disk, cleanly
// sever connections, etc)
Close()
// Convenience function for speed. Given a triple token and a direction
// return the node token for that direction. Sometimes, a TripleStore
// Convenience function for speed. Given a quad token and a direction
// return the node token for that direction. Sometimes, a QuadStore
// can do this without going all the way to the backing store, and
// gives the TripleStore the opportunity to make this optimization.
// gives the QuadStore the opportunity to make this optimization.
//
// Iterators will call this. At worst, a valid implementation is
// ts.IdFor(ts.quad.Quad(id).Get(dir))
TripleDirection(id Value, d quad.Direction) Value
//
// qs.ValueOf(qs.Quad(id).Get(dir))
//
QuadDirection(id Value, d quad.Direction) Value
}
type Options map[string]interface{}
@ -133,16 +135,16 @@ func (d Options) BoolKey(key string) (bool, bool) {
return false, false
}
var ErrCannotBulkLoad = errors.New("triplestore: cannot bulk load")
var ErrCannotBulkLoad = errors.New("quadstore: cannot bulk load")
type BulkLoader interface {
// BulkLoad loads Quads from a quad.Unmarshaler in bulk to the TripleStore.
// BulkLoad loads Quads from a quad.Unmarshaler in bulk to the QuadStore.
// It returns ErrCannotBulkLoad if bulk loading is not possible. For example if
// you cannot load in bulk to a non-empty database, and the db is non-empty.
BulkLoad(quad.Unmarshaler) error
}
type NewStoreFunc func(string, Options) (TripleStore, error)
type NewStoreFunc func(string, Options) (QuadStore, error)
type InitStoreFunc func(string, Options) error
type register struct {
@ -153,9 +155,9 @@ type register struct {
var storeRegistry = make(map[string]register)
func RegisterTripleStore(name string, persists bool, newFunc NewStoreFunc, initFunc InitStoreFunc) {
func RegisterQuadStore(name string, persists bool, newFunc NewStoreFunc, initFunc InitStoreFunc) {
if _, found := storeRegistry[name]; found {
panic("already registered TripleStore " + name)
panic("already registered QuadStore " + name)
}
storeRegistry[name] = register{
newFunc: newFunc,
@ -164,27 +166,27 @@ func RegisterTripleStore(name string, persists bool, newFunc NewStoreFunc, initF
}
}
func NewTripleStore(name, dbpath string, opts Options) (TripleStore, error) {
func NewQuadStore(name, dbpath string, opts Options) (QuadStore, error) {
r, registered := storeRegistry[name]
if !registered {
return nil, errors.New("triplestore: name '" + name + "' is not registered")
return nil, errors.New("quadstore: name '" + name + "' is not registered")
}
return r.newFunc(dbpath, opts)
}
func InitTripleStore(name, dbpath string, opts Options) error {
func InitQuadStore(name, dbpath string, opts Options) error {
r, registered := storeRegistry[name]
if registered {
return r.initFunc(dbpath, opts)
}
return errors.New("triplestore: name '" + name + "' is not registered")
return errors.New("quadstore: name '" + name + "' is not registered")
}
func IsPersistent(name string) bool {
return storeRegistry[name].isPersistent
}
func TripleStores() []string {
func QuadStores() []string {
t := make([]string, 0, len(storeRegistry))
for n := range storeRegistry {
t = append(t, n)

18
graph/quadwriter.go
View file

@ -16,9 +16,9 @@ package graph
// Defines the interface for consistent replication of a graph instance.
//
// Separate from the backend, this dictates how individual triples get
// Separate from the backend, this dictates how individual quads get
// identified and replicated consistently across (potentially) multiple
// instances. The simplest case is to keep an append-only log of triple
// instances. The simplest case is to keep an append-only log of quad
// changes.
import (
@ -44,7 +44,7 @@ type Delta struct {
}
type Handle struct {
QuadStore TripleStore
QuadStore QuadStore
QuadWriter QuadWriter
}
@ -54,8 +54,8 @@ func (h *Handle) Close() {
}
var (
ErrQuadExists = errors.New("Quad exists")
ErrQuadNotExist = errors.New("Quad doesn't exist")
ErrQuadExists = errors.New("quad exists")
ErrQuadNotExist = errors.New("quad does not exist")
)
type QuadWriter interface {
@ -73,23 +73,23 @@ type QuadWriter interface {
Close() error
}
type NewQuadWriterFunc func(TripleStore, Options) (QuadWriter, error)
type NewQuadWriterFunc func(QuadStore, Options) (QuadWriter, error)
var writerRegistry = make(map[string]NewQuadWriterFunc)
func RegisterWriter(name string, newFunc NewQuadWriterFunc) {
if _, found := writerRegistry[name]; found {
panic("already registered TripleWriter " + name)
panic("already registered QuadWriter " + name)
}
writerRegistry[name] = newFunc
}
func NewQuadWriter(name string, ts TripleStore, opts Options) (QuadWriter, error) {
func NewQuadWriter(name string, qs QuadStore, opts Options) (QuadWriter, error) {
newFunc, hasNew := writerRegistry[name]
if !hasNew {
return nil, errors.New("replication: name '" + name + "' is not registered")
}
return newFunc(ts, opts)
return newFunc(qs, opts)
}
func WriterMethods() []string {