Update triple API to support lookup options.

Initially thought as not needed, this was an ommision that should have not been done. This commits break all the drivers implementing tha graph interface.

bql/planner/data_access.go

@@ -358,7 +358,7 @@ func simpleFetch(ctx context.Context, gs []storage.Graph, cls *semantic.GraphCla
 			wg.Add(1)
 			go func() {
 				defer wg.Done()
-				tErr = g.Triples(ctx, ts)
+				tErr = g.Triples(ctx, lo, ts)
 			}()
 			aErr = addTriples(ts, cls, tbl)
 			wg.Wait()

bql/planner/data_access_test.go

@@ -222,7 +222,7 @@ func TestDataAccessAddTriples(t *testing.T) {
 	ts := make(chan *triple.Triple)
 	go func() {
 		defer wg.Done()
-		if err := g.Triples(ctx, ts); err != nil {
+		if err := g.Triples(ctx, storage.DefaultLookup, ts); err != nil {
 			t.Fatal(err)
 		}
 	}()

bql/planner/planner_test.go

@@ -57,7 +57,7 @@ func insertTest(t *testing.T) {
 	i := 0
 	ts := make(chan *triple.Triple)
 	go func() {
-		if err := g.Triples(ctx, ts); err != nil {
+		if err := g.Triples(ctx, storage.DefaultLookup, ts); err != nil {
 			t.Error(err)
 		}
 	}()
@@ -95,7 +95,7 @@ func deleteTest(t *testing.T) {
 	}
 	i := 0
 	ts := make(chan *triple.Triple)
-	if err := g.Triples(ctx, ts); err != nil {
+	if err := g.Triples(ctx, storage.DefaultLookup, ts); err != nil {
 		t.Error(err)
 	}
 	for _ = range ts {
@@ -246,7 +246,7 @@ func populateTestStore(t *testing.T) storage.Store {
 	}
 	trpls := make(chan *triple.Triple)
 	go func() {
-		if err := g.Triples(ctx, trpls); err != nil {
+		if err := g.Triples(ctx, storage.DefaultLookup, trpls); err != nil {
 			t.Fatal(err)
 		}
 	}()
@@ -272,7 +272,7 @@ func populateBenchmarkStore(b *testing.B) storage.Store {
 	}
 	trpls := make(chan *triple.Triple)
 	go func() {
-		if err := g.Triples(ctx, trpls); err != nil {
+		if err := g.Triples(ctx, storage.DefaultLookup, trpls); err != nil {
 			b.Fatal(err)
 		}
 	}()

io/io.go

@@ -66,7 +66,7 @@ func WriteGraph(ctx context.Context, w io.Writer, g storage.Graph) (int, error)
 	wg.Add(1)
 	go func() {
 		defer wg.Done()
-		tErr = g.Triples(ctx, ts)
+		tErr = g.Triples(ctx, storage.DefaultLookup, ts)
 	}()
 	for t := range ts {
 		if wErr != nil {

io/io_test.go

@@ -21,6 +21,7 @@ import (
 
 	"golang.org/x/net/context"
 
+	"github.com/google/badwolf/storage"
 	"github.com/google/badwolf/storage/memory"
 	"github.com/google/badwolf/triple"
 	"github.com/google/badwolf/triple/literal"
@@ -110,7 +111,7 @@ func TestSerializationContents(t *testing.T) {
 	gs := 0
 	gtrpls := make(chan *triple.Triple)
 	go func() {
-		if err := g.Triples(ctx, gtrpls); err != nil {
+		if err := g.Triples(ctx, storage.DefaultLookup, gtrpls); err != nil {
 			t.Errorf("g.Triples failed to retrieve triples with error %v", err)
 		}
 	}()
@@ -121,7 +122,7 @@ func TestSerializationContents(t *testing.T) {
 	gos := 0
 	g2trpls := make(chan *triple.Triple)
 	go func() {
-		if err := g2.Triples(ctx, g2trpls); err != nil {
+		if err := g2.Triples(ctx, storage.DefaultLookup, g2trpls); err != nil {
 			t.Errorf("g2.Triples failed to retrieve triples with error %v", err)
 		}
 	}()

storage/memory/memory.go

@@ -476,16 +476,19 @@ func (m *memory) Exist(ctx context.Context, t *triple.Triple) (bool, error) {
 
 // Triples allows to iterate over all available triples by pushing them to the
 // provided channel.
-func (m *memory) Triples(ctx context.Context, trpls chan<- *triple.Triple) error {
+func (m *memory) Triples(ctx context.Context, lo *storage.LookupOptions, trpls chan<- *triple.Triple) error {
 	if trpls == nil {
 		return fmt.Errorf("cannot provide an empty channel")
 	}
 	m.rwmu.RLock()
 	defer m.rwmu.RUnlock()
 	defer close(trpls)
 
+	ckr := newChecker(lo)
 	for _, t := range m.idx {
-		trpls <- t
+		if ckr.CheckAndUpdate(t.Predicate()) {
+			trpls <- t
+		}
 	}
 	return nil
 }

storage/memory/memory_test.go

@@ -513,7 +513,7 @@ func TestTriples(t *testing.T) {
 	// to call the graph operation on a separated goroutine using a sync.WaitGroup
 	// to collect the error code eventually.
 	trpls := make(chan *triple.Triple, 100)
-	if err := g.Triples(ctx, trpls); err != nil {
+	if err := g.Triples(ctx, storage.DefaultLookup, trpls); err != nil {
 		t.Fatal(err)
 	}
 	cnt := 0

storage/storage.go

@@ -241,5 +241,5 @@ type Graph interface {
 	// Triples pushes to the provided channel all available triples in the graph.
 	// The function does not return immediately but spawns a goroutine to satisfy
 	// elements in the channel.
-	Triples(ctx context.Context, trpls chan<- *triple.Triple) error
+	Triples(ctx context.Context, lo *LookupOptions, trpls chan<- *triple.Triple) error
 }

tools/vcli/bw/export/export.go

@@ -72,7 +72,7 @@ func Eval(ctx context.Context, usage string, args []string, store storage.Store,
 	chn := make(chan *triple.Triple, bulkSize)
 	for _, vg := range sgs {
 		go func(g storage.Graph) {
-			err := g.Triples(ctx, chn)
+			err := g.Triples(ctx, storage.DefaultLookup, chn)
 			mu.Lock()
 			errs = append(errs, err)
 			mu.Unlock()