Fix filters by using NaiveFilter

mount/mount.go

@@ -141,14 +141,6 @@ func (h *querySet) next() (query.Result, bool) {
 	next := head.next
 
 	for head.advance() {
-		if head.next.Error == nil {
-			for _, f := range h.query.Filters {
-				if !f.Filter(head.next.Entry) {
-					continue
-				}
-			}
-			// can limit and offset be implemented here?
-		}
 		heap.Fix(h, 0)
 		return next, true
 	}
@@ -235,6 +227,8 @@ func (d *Datastore) Query(q query.Query) (query.Results, error) {
 	// offset needs to be applied after the results are aggregated
 	offset := q.Offset
 	q.Offset = 0
+	filters := q.Filters
+	q.Filters = []query.Filter{}
 
 	queries := &querySet{
 		query: q,
@@ -262,6 +256,12 @@ func (d *Datastore) Query(q query.Query) (query.Results, error) {
 		Close: queries.close,
 	})
 
+	if len(filters) > 0 {
+		for _, f := range filters {
+			qr = query.NaiveFilter(qr, f)
+		}
+	}
+
 	if offset > 0 {
 		qr = query.NaiveOffset(qr, offset)
 	}

mount/mount_test.go

@@ -452,6 +452,97 @@ func TestQueryLimitAndOffsetCrossWithSort(t *testing.T) {
 	}
 }
 
+func TestQueryFilterAcrossMountsWithSort(t *testing.T) {
+	mapds1 := sync.MutexWrap(datastore.NewMapDatastore())
+	mapds2 := sync.MutexWrap(datastore.NewMapDatastore())
+	mapds3 := sync.MutexWrap(datastore.NewMapDatastore())
+	m := mount.New([]mount.Mount{
+		{Prefix: datastore.NewKey("/rok"), Datastore: mapds1},
+		{Prefix: datastore.NewKey("/zoo"), Datastore: mapds2},
+		{Prefix: datastore.NewKey("/noop"), Datastore: mapds3},
+	})
+
+	m.Put(datastore.NewKey("/rok/0"), []byte("ghi"))
+	m.Put(datastore.NewKey("/zoo/0"), []byte("123"))
+	m.Put(datastore.NewKey("/rok/1"), []byte("def"))
+	m.Put(datastore.NewKey("/zoo/1"), []byte("167"))
+	m.Put(datastore.NewKey("/zoo/2"), []byte("345"))
+	m.Put(datastore.NewKey("/rok/3"), []byte("abc"))
+	m.Put(datastore.NewKey("/zoo/3"), []byte("456"))
+
+	f := &query.FilterKeyCompare{Op: query.Equal, Key: "/rok/3"}
+	q := query.Query{Filters: []query.Filter{f}}
+	res, err := m.Query(q)
+	if err != nil {
+		t.Fatalf("Query fail: %v\n", err)
+	}
+
+	entries, err := res.Rest()
+	if err != nil {
+		t.Fatalf("Query Results.Rest fail: %v\n", err)
+	}
+
+	expect := []string{
+		"/rok/3",
+	}
+
+	if len(entries) != len(expect) {
+		t.Fatalf("expected %d entries, but got %d", len(expect), len(entries))
+	}
+
+	for i, e := range expect {
+		if e != entries[i].Key {
+			t.Errorf("expected key %s, but got %s", e, entries[i].Key)
+		}
+	}
+
+	err = res.Close()
+	if err != nil {
+		t.Errorf("result.Close failed %d", err)
+	}
+}
+
+func TestQueryLimit(t *testing.T) {
+	mapds1 := sync.MutexWrap(datastore.NewMapDatastore())
+	m := mount.New([]mount.Mount{
+		{Prefix: datastore.NewKey("/rok"), Datastore: mapds1},
+	})
+
+	m.Put(datastore.NewKey("/rok/0"), []byte("ghi"))
+	m.Put(datastore.NewKey("/rok/1"), []byte("def"))
+	m.Put(datastore.NewKey("/rok/3"), []byte("abc"))
+
+	q := query.Query{Limit: 1, Orders: []query.Order{query.OrderByKeyDescending{}}}
+	res, err := m.Query(q)
+	if err != nil {
+		t.Fatalf("Query fail: %v\n", err)
+	}
+
+	entries, err := res.Rest()
+	if err != nil {
+		t.Fatalf("Query Results.Rest fail: %v\n", err)
+	}
+
+	expect := []string{
+		"/rok/3",
+	}
+
+	if len(entries) != len(expect) {
+		t.Fatalf("expected %d entries, but got %d", len(expect), len(entries))
+	}
+
+	for i, e := range expect {
+		if e != entries[i].Key {
+			t.Errorf("expected key %s, but got %s", e, entries[i].Key)
+		}
+	}
+
+	err = res.Close()
+	if err != nil {
+		t.Errorf("result.Close failed %d", err)
+	}
+}
+
 func TestLookupPrio(t *testing.T) {
 	mapds0 := datastore.NewMapDatastore()
 	mapds1 := datastore.NewMapDatastore()