From d261e5d870d16d7db164767f990e34917ee6e7dd Mon Sep 17 00:00:00 2001 From: Barak Michener Date: Mon, 11 Aug 2014 02:54:14 -0400 Subject: [PATCH] wip --- graph/bolt/all_iterator.go | 176 +++++++++++ graph/bolt/iterator.go | 289 ++++++++++++++++++ graph/bolt/quadstore.go | 478 ++++++++++++++++++++++++++++++ graph/bolt/quadstore_iterator_optimize.go | 55 ++++ 4 files changed, 998 insertions(+) create mode 100644 graph/bolt/all_iterator.go create mode 100644 graph/bolt/iterator.go create mode 100644 graph/bolt/quadstore.go create mode 100644 graph/bolt/quadstore_iterator_optimize.go diff --git a/graph/bolt/all_iterator.go b/graph/bolt/all_iterator.go new file mode 100644 index 0000000..95fe04b --- /dev/null +++ b/graph/bolt/all_iterator.go @@ -0,0 +1,176 @@ +// Copyright 2014 The Cayley Authors. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package bolt + +import ( + "bytes" + "fmt" + "strings" + + "github.com/google/cayley/graph" + "github.com/google/cayley/graph/iterator" + "github.com/google/cayley/quad" +) + +type AllIterator struct { + uid uint64 + tags graph.Tagger + prefix []byte + dir quad.Direction + open bool + qs *QuadStore + result graph.Value +} + +func NewAllIterator(prefix string, d quad.Direction, ts *QuadStore) *AllIterator { + opts := &opt.ReadOptions{ + DontFillCache: true, + } + + it := AllIterator{ + uid: iterator.NextUID(), + ro: opts, + iter: ts.db.NewIterator(nil, opts), + prefix: []byte(prefix), + dir: d, + open: true, + ts: ts, + } + + it.iter.Seek(it.prefix) + if !it.iter.Valid() { + // FIXME(kortschak) What are the semantics here? Is this iterator usable? + // If not, we should return nil *Iterator and an error. + it.open = false + it.iter.Release() + } + + return &it +} + +func (it *AllIterator) UID() uint64 { + return it.uid +} + +func (it *AllIterator) Reset() { + if !it.open { + it.iter = it.ts.db.NewIterator(nil, it.ro) + it.open = true + } + it.iter.Seek(it.prefix) + if !it.iter.Valid() { + it.open = false + it.iter.Release() + } +} + +func (it *AllIterator) Tagger() *graph.Tagger { + return &it.tags +} + +func (it *AllIterator) TagResults(dst map[string]graph.Value) { + for _, tag := range it.tags.Tags() { + dst[tag] = it.Result() + } + + for tag, value := range it.tags.Fixed() { + dst[tag] = value + } +} + +func (it *AllIterator) Clone() graph.Iterator { + out := NewAllIterator(string(it.prefix), it.dir, it.ts) + out.tags.CopyFrom(it) + return out +} + +func (it *AllIterator) Next() bool { + if !it.open { + it.result = nil + return false + } + var out []byte + out = make([]byte, len(it.iter.Key())) + copy(out, it.iter.Key()) + it.iter.Next() + if !it.iter.Valid() { + it.Close() + } + if !bytes.HasPrefix(out, it.prefix) { + it.Close() + return false + } + it.result = Token(out) + return true +} + +func (it *AllIterator) ResultTree() *graph.ResultTree { + return graph.NewResultTree(it.Result()) +} + +func (it *AllIterator) Result() graph.Value { + return it.result +} + +func (it *AllIterator) NextPath() bool { + return false +} + +// No subiterators. +func (it *AllIterator) SubIterators() []graph.Iterator { + return nil +} + +func (it *AllIterator) Contains(v graph.Value) bool { + it.result = v + return true +} + +func (it *AllIterator) Close() { + if it.open { + it.iter.Release() + it.open = false + } +} + +func (it *AllIterator) Size() (int64, bool) { + size, err := it.ts.SizeOfPrefix(it.prefix) + if err == nil { + return size, false + } + // INT64_MAX + return int64(^uint64(0) >> 1), false +} + +func (it *AllIterator) DebugString(indent int) string { + size, _ := it.Size() + return fmt.Sprintf("%s(%s tags: %v leveldb size:%d %s %p)", strings.Repeat(" ", indent), it.Type(), it.tags.Tags(), size, it.dir, it) +} + +func (it *AllIterator) Type() graph.Type { return graph.All } +func (it *AllIterator) Sorted() bool { return false } + +func (it *AllIterator) Optimize() (graph.Iterator, bool) { + return it, false +} + +func (it *AllIterator) Stats() graph.IteratorStats { + s, _ := it.Size() + return graph.IteratorStats{ + ContainsCost: 1, + NextCost: 2, + Size: s, + } +} diff --git a/graph/bolt/iterator.go b/graph/bolt/iterator.go new file mode 100644 index 0000000..a1a1da6 --- /dev/null +++ b/graph/bolt/iterator.go @@ -0,0 +1,289 @@ +// Copyright 2014 The Cayley Authors. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package bolt + +import ( + "bytes" + "encoding/json" + "fmt" + "strings" + + "github.com/barakmich/glog" + ldbit "github.com/syndtr/goleveldb/leveldb/iterator" + "github.com/syndtr/goleveldb/leveldb/opt" + + "github.com/google/cayley/graph" + "github.com/google/cayley/graph/iterator" + "github.com/google/cayley/quad" +) + +type Iterator struct { + uid uint64 + tags graph.Tagger + nextPrefix []byte + checkId []byte + dir quad.Direction + open bool + iter ldbit.Iterator + qs *QuadStore + ro *opt.ReadOptions + originalPrefix string + result graph.Value +} + +func NewIterator(prefix string, d quad.Direction, value graph.Value, qs *QuadStore) graph.Iterator { + vb := value.(Token) + p := make([]byte, 0, 2+qs.hasher.Size()) + p = append(p, []byte(prefix)...) + p = append(p, []byte(vb[1:])...) + + opts := &opt.ReadOptions{ + DontFillCache: true, + } + + it := Iterator{ + uid: iterator.NextUID(), + nextPrefix: p, + checkId: vb, + dir: d, + originalPrefix: prefix, + ro: opts, + iter: qs.db.NewIterator(nil, opts), + open: true, + qs: qs, + } + + ok := it.iter.Seek(it.nextPrefix) + if !ok { + it.open = false + it.iter.Release() + glog.Error("Opening LevelDB iterator couldn't seek to location ", it.nextPrefix) + return &iterator.Null{} + } + + return &it +} + +func (it *Iterator) UID() uint64 { + return it.uid +} + +func (it *Iterator) Reset() { + if !it.open { + it.iter = it.qs.db.NewIterator(nil, it.ro) + it.open = true + } + ok := it.iter.Seek(it.nextPrefix) + if !ok { + it.open = false + it.iter.Release() + } +} + +func (it *Iterator) Tagger() *graph.Tagger { + return &it.tags +} + +func (it *Iterator) TagResults(dst map[string]graph.Value) { + for _, tag := range it.tags.Tags() { + dst[tag] = it.Result() + } + + for tag, value := range it.tags.Fixed() { + dst[tag] = value + } +} + +func (it *Iterator) Clone() graph.Iterator { + out := NewIterator(it.originalPrefix, it.dir, Token(it.checkId), it.qs) + out.Tagger().CopyFrom(it) + return out +} + +func (it *Iterator) Close() { + if it.open { + it.iter.Release() + it.open = false + } +} + +func (it *Iterator) isLiveValue(val []byte) bool { + var entry IndexEntry + json.Unmarshal(val, &entry) + return len(entry.History)%2 != 0 +} + +func (it *Iterator) Next() bool { + if it.iter == nil { + it.result = nil + return false + } + if !it.open { + it.result = nil + return false + } + if !it.iter.Valid() { + it.result = nil + it.Close() + return false + } + if bytes.HasPrefix(it.iter.Key(), it.nextPrefix) { + if !it.isLiveValue(it.iter.Value()) { + return it.Next() + } + out := make([]byte, len(it.iter.Key())) + copy(out, it.iter.Key()) + it.result = Token(out) + ok := it.iter.Next() + if !ok { + it.Close() + } + return true + } + it.Close() + it.result = nil + return false +} + +func (it *Iterator) ResultTree() *graph.ResultTree { + return graph.NewResultTree(it.Result()) +} + +func (it *Iterator) Result() graph.Value { + return it.result +} + +func (it *Iterator) NextPath() bool { + return false +} + +// No subiterators. +func (it *Iterator) SubIterators() []graph.Iterator { + return nil +} + +func PositionOf(prefix []byte, d quad.Direction, qs *QuadStore) int { + if bytes.Equal(prefix, []byte("sp")) { + switch d { + case quad.Subject: + return 2 + case quad.Predicate: + return qs.hasher.Size() + 2 + case quad.Object: + return 2*qs.hasher.Size() + 2 + case quad.Label: + return 3*qs.hasher.Size() + 2 + } + } + if bytes.Equal(prefix, []byte("po")) { + switch d { + case quad.Subject: + return 2*qs.hasher.Size() + 2 + case quad.Predicate: + return 2 + case quad.Object: + return qs.hasher.Size() + 2 + case quad.Label: + return 3*qs.hasher.Size() + 2 + } + } + if bytes.Equal(prefix, []byte("os")) { + switch d { + case quad.Subject: + return qs.hasher.Size() + 2 + case quad.Predicate: + return 2*qs.hasher.Size() + 2 + case quad.Object: + return 2 + case quad.Label: + return 3*qs.hasher.Size() + 2 + } + } + if bytes.Equal(prefix, []byte("cp")) { + switch d { + case quad.Subject: + return 2*qs.hasher.Size() + 2 + case quad.Predicate: + return qs.hasher.Size() + 2 + case quad.Object: + return 3*qs.hasher.Size() + 2 + case quad.Label: + return 2 + } + } + panic("unreachable") +} + +func (it *Iterator) Contains(v graph.Value) bool { + val := v.(Token) + if val[0] == 'z' { + 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 + // again? + // + // We've already done that -- in order to get the graph.Value token in the + // first place, we had to have done the check already; it came from a Next(). + // + // However, if it ever starts coming from somewhere else, it'll be more + // efficient to change the interface of the graph.Value for LevelDB to a + // struct with a flag for isValid, to save another random read. + return true + } + return false +} + +func (it *Iterator) Size() (int64, bool) { + return it.qs.SizeOf(Token(it.checkId)), true +} + +func (it *Iterator) DebugString(indent int) string { + size, _ := it.Size() + return fmt.Sprintf("%s(%s %d tags: %v dir: %s size:%d %s)", + strings.Repeat(" ", indent), + it.Type(), + it.UID(), + it.tags.Tags(), + it.dir, + size, + it.qs.NameOf(Token(it.checkId)), + ) +} + +var levelDBType graph.Type + +func init() { + levelDBType = graph.RegisterIterator("leveldb") +} + +func Type() graph.Type { return levelDBType } + +func (it *Iterator) Type() graph.Type { return levelDBType } +func (it *Iterator) Sorted() bool { return false } + +func (it *Iterator) Optimize() (graph.Iterator, bool) { + return it, false +} + +func (it *Iterator) Stats() graph.IteratorStats { + s, _ := it.Size() + return graph.IteratorStats{ + ContainsCost: 1, + NextCost: 2, + Size: s, + } +} diff --git a/graph/bolt/quadstore.go b/graph/bolt/quadstore.go new file mode 100644 index 0000000..974da37 --- /dev/null +++ b/graph/bolt/quadstore.go @@ -0,0 +1,478 @@ +// Copyright 2014 The Cayley Authors. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package bolt + +import ( + "bytes" + "crypto/sha1" + "encoding/binary" + "encoding/json" + "fmt" + "hash" + + "github.com/barakmich/glog" + "github.com/boltdb/bolt" + + "github.com/google/cayley/graph" + "github.com/google/cayley/graph/iterator" + "github.com/google/cayley/quad" +) + +func init() { + graph.RegisterTripleStore("bolt", true, newQuadStore, createNewLevelDB) +} + +type Token struct { + bucket []byte + key []byte +} + +func (t *Token) Key() interface{} { + return fmt.Sprint(t.bucket, t.data) +} + +type QuadStore struct { + db *bolt.DB + path string + open bool + size int64 + horizon int64 + hasher hash.Hash +} + +func createNewBolt(path string, _ graph.Options) error { + opts := &opt.Options{} + db, err := bolt.Open(path, 0600, nil) + if err != nil { + glog.Errorf("Error: couldn't create Bolt database: %v", err) + return err + } + defer db.Close() + qs := &QuadStore{} + qs.db = db + qs.writeopts = &opt.WriteOptions{ + Sync: true, + } + err = qs.createBuckets() + if err != nil { + return err + } + qs.Close() + return nil +} + +func newQuadStore(path string, options graph.Options) (graph.TripleStore, error) { + var qs QuadStore + var err error + qs.hasher = sha1.New() + db, err := bolt.Open(path, 0600, nil) + if err != nil { + glog.Errorln("Error, couldn't open! ", err) + return nil, err + } + qs.db = db + err = qs.getMetadata() + if err != nil { + return nil, err + } + return &qs, nil +} + +func (qs *QuadStore) createBuckets() error { + return db.Update(func(tx *bolt.Tx) error { + var err error + for _, bucket := range [][]byte{spo, osp, pos, cps} { + _, err = tx.CreateBucket(bucketFor(bucket)) + if err != nil { + return fmt.Errorf("Couldn't create bucket: %s", err) + } + } + _, err = tx.CreateBucket(logBucket) + if err != nil { + return fmt.Errorf("Couldn't create bucket: %s", err) + } + _, err = tx.CreateBucket(nodeBucket) + if err != nil { + return fmt.Errorf("Couldn't create bucket: %s", err) + } + _, err = tx.CreateBucket(metaBucket) + if err != nil { + return fmt.Errorf("Couldn't create bucket: %s", err) + } + }) +} + +func (qs *QuadStore) Size() int64 { + return qs.size +} + +func (qs *QuadStore) Horizon() int64 { + return qs.horizon +} + +func (qa *QuadStore) createDeltaKeyFor(d *graph.Delta) []byte { + return []byte(fmt.Sprintf("%018x", d.ID)) +} + +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 { + key := make([]byte, 0, (qs.hasher.Size() * 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]))...) + return key +} + +func (qs *QuadStore) createValueKeyFor(s string) []byte { + key := make([]byte, 0, qs.hasher.Size()) + key = append(key, qs.convertStringToByteHash(s)...) + return key +} + +type IndexEntry struct { + quad.Quad + History []int64 +} + +// Short hand for direction permutations. +var ( + spo = bucketFor([4]quad.Direction{quad.Subject, quad.Predicate, quad.Object, quad.Label}) + osp = bucketFor([4]quad.Direction{quad.Object, quad.Subject, quad.Predicate, quad.Label}) + pos = bucketFor([4]quad.Direction{quad.Predicate, quad.Object, quad.Subject, quad.Label}) + cps = bucketFor([4]quad.Direction{quad.Label, quad.Predicate, quad.Subject, quad.Object}) +) + +var logBucket = []byte("log") +var nodeBucket = []byte("node") +var metaBucket = []byte("meta") + +func (qs *QuadStore) ApplyDeltas(deltas []*graph.Delta) error { + batch := &leveldb.Batch{} + var size_change int64 + var new_horizon int64 + err := qs.db.Update(func(tx *bolt.Tx) error { + var b *bolt.Bucket + var err error + resizeMap := make(map[string]int64) + size_change = int64(0) + for _, d := range deltas { + bytes, err := json.Marshal(d) + if err != nil { + return err + } + b = tx.Bucket(logBucket) + err = b.Put(qs.createDeltaKeyFor(d), bytes) + if err != nil { + return err + } + err = qs.buildQuadWrite(tx, d.Quad, d.ID, d.Action == graph.Add) + if err != nil { + return err + } + delta := int64(1) + if d.Action == graph.Delete { + delta = int64(-1) + } + resizeMap[d.Quad.Subject] += delta + resizeMap[d.Quad.Predicate] += delta + resizeMap[d.Quad.Object] += delta + if d.Quad.Label != "" { + resizeMap[d.Quad.Label] += delta + } + size_change += delta + new_horizon = d.ID + } + for k, v := range resizeMap { + if v != 0 { + err := qs.UpdateValueKeyBy(k, v, tx) + if err != nil { + return err + } + } + } + return nil + }) + + if err != nil { + glog.Error("Couldn't write to DB for Delta set. Error: ", err) + return err + } + qs.size += size_change + qs.horizon = new_horizon + return nil +} + +func (qs *QuadStore) buildQuadWrite(tx *bolt.Tx, q quad.Quad, id int64, isAdd bool) error { + var entry IndexEntry + b := tx.Bucket(bucketFor(spo)) + + data := b.Get(qs.createKeyFor(spo, q)) + if data != nil { + // We got something. + err = json.Unmarshal(data, &entry) + if err != nil { + return err + } + } else { + entry.Quad = q + } + + if isAdd && len(entry.History)%2 == 1 { + glog.Error("Adding a valid triple ", entry) + return graph.ErrQuadExists + } + if !isAdd && len(entry.History)%2 == 0 { + glog.Error("Deleting an invalid triple ", entry) + return graph.ErrQuadNotExist + } + + entry.History = append(entry.History, id) + + bytes, err := json.Marshal(entry) + if err != nil { + glog.Errorf("Couldn't write to buffer for entry %#v: %s", entry, err) + return err + } + for _, bucket := range [][4]quad.Direction{spo, osp, pos, cps} { + if bucket == cps && q.Get(quad.Label) == "" { + continue + } + b := tx.Bucket(bucketFor(bucket)) + err = b.Put(qs.createKeyFor(bucket, q), bytes) + if err != nil { + return err + } + } + return nil +} + +type ValueData struct { + Name string + Size int64 +} + +func (qs *QuadStore) UpdateValueKeyBy(name string, amount int64, tx *bolt.Tx) error { + value := &ValueData{name, amount} + b := tx.Bucket(nodeBucket) + key := qs.createValueKeyFor(name) + data := b.Get(key) + + if data != nil { + // Node exists in the database -- unmarshal and update. + err = json.Unmarshal(b, value) + if err != nil { + glog.Errorf("Error: couldn't reconstruct value: %v", err) + return err + } + value.Size += amount + } + + // Are we deleting something? + if value.Size <= 0 { + value.Size = 0 + } + + // Repackage and rewrite. + bytes, err := json.Marshal(&value) + if err != nil { + glog.Errorf("Couldn't write to buffer for value %s: %s", name, err) + return err + } + err = b.Put(key, bytes) + return err +} + +func (qs *QuadStore) Close() { + qs.db.Update(func(tx *bolt.Tx) error { + buf := new(bytes.Buffer) + err := binary.Write(buf, binary.LittleEndian, qs.size) + if err == nil { + b := tx.Bucket(metaBucket) + werr := b.Put([]byte("size"), buf.Bytes()) + if werr != nil { + glog.Error("Couldn't write size before closing!") + return werr + } + } else { + glog.Errorf("Couldn't convert size before closing!") + return err + } + buf.Reset() + err = binary.Write(buf, binary.LittleEndian, qs.horizon) + if err == nil { + b := tx.Bucket(metaBucket) + werr := b.Put([]byte("horizon"), buf.Bytes()) + if werr != nil { + glog.Error("Couldn't write horizon before closing!") + return werr + } + } else { + glog.Errorf("Couldn't convert horizon before closing!") + } + return err + }) + qs.db.Close() + qs.open = false +} + +func (qs *QuadStore) Quad(k graph.Value) quad.Quad { + var q quad.Quad + tok := k.(*Token) + err := qs.db.View(func(tx *bolt.Tx) error { + b := tx.Bucket(tok.bucket) + data := qs.db.Get(tok.key, qs.readopts) + if data == nil { + // No harm, no foul. + return nil + } + err = json.Unmarshal(data, &q) + return err + }) + if err != nil { + glog.Error("Error getting triple: ", err) + return quad.Quad{} + } + return q +} + +func (qs *QuadStore) convertStringToByteHash(s string) []byte { + qs.hasher.Reset() + key := make([]byte, 0, qs.hasher.Size()) + qs.hasher.Write([]byte(s)) + key = qs.hasher.Sum(key) + return key +} + +func (qs *QuadStore) ValueOf(s string) graph.Value { + return &Token{ + bucket: nodeBucket, + key: qs.createValueKeyFor(s), + } +} + +func (qs *QuadStore) valueData(t *Token) ValueData { + var out ValueData + if glog.V(3) { + glog.V(3).Infof("%s %v", string(t.bucket), t.key) + } + err := qs.db.View(func(tx *bolt.Tx) error { + b := tx.Bucket(t.bucket) + data := b.Get(t.key) + if data != nil { + return json.Unmarshal(data, &out) + } + return nil + }) + if err != nil { + glog.Errorln("Error: couldn't get value") + return ValueData{} + } + return out +} + +func (qs *QuadStore) NameOf(k graph.Value) string { + if k == nil { + glog.V(2).Info("k was nil") + return "" + } + return qs.valueData(k.(*Token)).Name +} + +func (qs *QuadStore) SizeOf(k graph.Value) int64 { + if k == nil { + return 0 + } + return int64(qs.valueData(k.(*Token)).Size) +} + +func (qs *QuadStore) getInt64ForKey(tx *bolt.Tx, key string, empty int64) (int64, error) { + var out int64 + b := tx.Bucket(metaBucket) + data := b.Get([]byte(key)) + if data == nil { + return empty, nil + } + buf := bytes.NewBuffer(data) + err = binary.Read(buf, binary.LittleEndian, &out) + if err != nil { + return 0, err + } + return out, nil +} + +func (qs *QuadStore) getMetadata() error { + err := qs.db.View(func(tx *bolt.Tx) error { + var err error + qs.size, err = qs.getInt64ForKey("size", 0) + if err != nil { + return err + } + qs.horizon, err = qs.getInt64ForKey("horizon", 0) + return err + }) + return err +} + +func (qs *QuadStore) TripleIterator(d quad.Direction, val graph.Value) graph.Iterator { + var prefix []byte + switch d { + case quad.Subject: + prefix = spo + case quad.Predicate: + prefix = pos + case quad.Object: + prefix = osp + case quad.Label: + prefix = cps + default: + panic("unreachable " + d.String()) + } + return NewIterator(prefix, d, val, qs) +} + +func (qs *QuadStore) NodesAllIterator() graph.Iterator { + return NewAllIterator(nodeBucket, quad.Any, qs) +} + +func (qs *QuadStore) TriplesAllIterator() graph.Iterator { + return NewAllIterator(pos, quad.Predicate, qs) +} + +func (qs *QuadStore) TripleDirection(val graph.Value, d quad.Direction) graph.Value { + v := val.(*Token) + offset := PositionOf(v, d, qs) + if offset != -1 { + return &Token{ + bucket: nodeBucket, + key: v[offset : offset+qs.hasher.Size()], + } + } else { + return qs.ValueOf(qs.Quad(v).Get(d)) + } +} + +func compareTokens(a, b graph.Value) bool { + atok := a.(*Token) + btok := b.(*Token) + return bytes.Equal(atok.key, btok.key) && atok.bucket == btok.bucket +} + +func (qs *QuadStore) FixedIterator() graph.FixedIterator { + return iterator.NewFixedIteratorWithCompare(compareTokens) +} diff --git a/graph/bolt/quadstore_iterator_optimize.go b/graph/bolt/quadstore_iterator_optimize.go new file mode 100644 index 0000000..e893020 --- /dev/null +++ b/graph/bolt/quadstore_iterator_optimize.go @@ -0,0 +1,55 @@ +// Copyright 2014 The Cayley Authors. All rights reserved. +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +package bolt + +import ( + "github.com/google/cayley/graph" + "github.com/google/cayley/graph/iterator" +) + +func (ts *QuadStore) OptimizeIterator(it graph.Iterator) (graph.Iterator, bool) { + switch it.Type() { + case graph.LinksTo: + return ts.optimizeLinksTo(it.(*iterator.LinksTo)) + + } + return it, false +} + +func (ts *QuadStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool) { + subs := it.SubIterators() + if len(subs) != 1 { + return it, false + } + primary := subs[0] + if primary.Type() == graph.Fixed { + size, _ := primary.Size() + if size == 1 { + if !graph.Next(primary) { + panic("unexpected size during optimize") + } + val := primary.Result() + newIt := ts.TripleIterator(it.Direction(), val) + nt := newIt.Tagger() + nt.CopyFrom(it) + for _, tag := range primary.Tagger().Tags() { + nt.AddFixed(tag, val) + } + it.Close() + return newIt, true + } + } + return it, false +}