Repository: cayleygraph/cayley
Pull request #832 (2019-09-23 01:09:10)

Define and implement a new query interface
Current query interface (in Go) is weird. The user must know that `Execute` should be called in a goroutine, and then returned values should be `Collated`. This is a very non-intuitive way to expose query results.

This PR defines a new interface that returns a regular iterator. The type of query results is controlled by an option. This way the query language will collate value internally, providing only useful values to the user.

This is a breaking change, since it replaces the old interface completely. However, internal query interface was not documented, and was not intended for external use, so I think it's worth changing anyway.

This change also adds an extendable `Options` parameter to `Execute`, unlocking future progress on #821.

<!-- Reviewable:start -->
---
This change is [<img src="https://reviewable.io/review_button.svg" height="34" align="absmiddle" alt="Reviewable"/>](https://reviewable.io/reviews/cayleygraph/cayley/832)
<!-- Reviewable:end -->



Patch:
diff --git a/cmd/cayley/command/repl.go b/cmd/cayley/command/repl.go
index fc31af9..2fdd766 100644
--- a/cmd/cayley/command/repl.go
+++ b/cmd/cayley/command/repl.go
@@ -14,10 +14,8 @@ import (
 	"github.com/spf13/viper"
 
 	"github.com/cayleygraph/cayley/clog"
-	"github.com/cayleygraph/cayley/graph"
 	"github.com/cayleygraph/cayley/internal/repl"
 	"github.com/cayleygraph/cayley/query"
-	"github.com/cayleygraph/quad"
 )
 
 const (
@@ -117,38 +115,21 @@ func NewQueryCmd() *cobra.Command {
 			if err != nil {
 				return err
 			}
-
-			l := query.GetLanguage(lang)
-			if l == nil {
-				return fmt.Errorf("unknown query language: %q", lang)
-			}
 			enc := json.NewEncoder(os.Stdout)
-			sess := l.Session(h)
-			ch := make(chan query.Result, 100)
-			go sess.Execute(ctx, querystr, ch, limit)
-			for i := 0; limit <= 0 || i < limit; i++ {
-				select {
-				case <-ctx.Done():
-					return ctx.Err()
-				case r, ok := <-ch:
-					if !ok {
-						return nil
-					} else if err = r.Err(); err != nil {
-						return err
-					}
-					obj := r.Result()
-					switch p := obj.(type) {
-					case map[string]graph.Ref:
-						m := make(map[string]quad.Value, len(p))
-						for k, v := range p {
-							m[k] = h.NameOf(v)
-						}
-						obj = m
-					}
-					enc.Encode(obj)
+			it, err := query.Execute(ctx, h, lang, querystr, query.Options{
+				Collation: query.JSON,
+				Limit:     limit,
+			})
+			if err != nil {
+				return err
+			}
+			defer it.Close()
+			for i := 0; it.Next(ctx) && (limit <= 0 || i < limit); i++ {
+				if err = enc.Encode(it.Result()); err != nil {
+					return err
 				}
 			}
-			return nil
+			return it.Err()
 		},
 	}
 	registerQueryFlags(cmd)
diff --git a/internal/http/query.go b/internal/http/query.go
index c56d333..d67f433 100644
--- a/internal/http/query.go
+++ b/internal/http/query.go
@@ -120,27 +120,25 @@ func (api *API) ServeV1Query(w http.ResponseWriter, r *http.Request, params http
 		errFunc(w, err)
 		return
 	}
-	code := string(bodyBytes)
-
-	c := make(chan query.Result, 5)
-	go ses.Execute(ctx, code, c, limit)
+	it, err := ses.Execute(ctx, string(bodyBytes), query.Options{
+		Collation: query.JSON,
+		Limit:     limit,
+	})
+	if err != nil {
+		errFunc(w, err)
+		return
+	}
+	defer it.Close()
 
-	for res := range c {
-		if err := res.Err(); err != nil {
-			if err == nil {
-				continue // wait for results channel to close
-			}
-			errFunc(w, err)
-			return
-		}
-		ses.Collate(res)
+	var out []interface{}
+	for it.Next(ctx) {
+		out = append(out, it.Result())
 	}
-	output, err := ses.Results()
-	if err != nil {
+	if err = it.Err(); err != nil {
 		errFunc(w, err)
 		return
 	}
-	_ = WriteResult(w, output)
+	_ = WriteResult(w, out)
 }
 
 func (api *API) ServeV1Shape(w http.ResponseWriter, r *http.Request, params httprouter.Params) {
diff --git a/internal/repl/repl.go b/internal/repl/repl.go
index 46b2b9e..015eb26 100644
--- a/internal/repl/repl.go
+++ b/internal/repl/repl.go
@@ -53,15 +53,21 @@ func Run(ctx context.Context, qu string, ses query.REPLSession) error {
 		}
 	}()
 	fmt.Printf("\n")
-	c := make(chan query.Result, 5)
-	go ses.Execute(ctx, qu, c, 100)
-	for res := range c {
-		if err := res.Err(); err != nil {
-			return err
-		}
-		fmt.Print(ses.FormatREPL(res))
+	it, err := ses.Execute(ctx, qu, query.Options{
+		Collation: query.REPL,
+		Limit:     100,
+	})
+	if err != nil {
+		return err
+	}
+	defer it.Close()
+	for it.Next(ctx) {
+		fmt.Print(it.Result())
 		nResults++
 	}
+	if err := it.Err(); err != nil {
+		return err
+	}
 	if nResults > 0 {
 		results := "Result"
 		if nResults > 1 {
diff --git a/query/gizmo/gizmo.go b/query/gizmo/gizmo.go
index 19ef03a..2e2e90d 100644
--- a/query/gizmo/gizmo.go
+++ b/query/gizmo/gizmo.go
@@ -16,6 +16,7 @@ package gizmo
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"reflect"
 	"sort"
@@ -25,7 +26,6 @@ import (
 
 	"github.com/dop251/goja"
 
-	"github.com/cayleygraph/cayley/clog"
 	"github.com/cayleygraph/cayley/graph"
 	"github.com/cayleygraph/cayley/graph/iterator"
 	"github.com/cayleygraph/cayley/query"
@@ -96,15 +96,13 @@ type Session struct {
 	last string
 	p    *goja.Program
 
-	out   chan query.Result
+	out   chan *Result
 	ctx   context.Context
 	limit int
 	count int
 
-	// used only to collate web results
-	dataOutput []interface{}
-	err        error
-	shape      map[string]interface{}
+	err   error
+	shape map[string]interface{}
 }
 
 func (s *Session) context() context.Context {
@@ -197,7 +195,7 @@ func (s *Session) runIteratorWithCallback(it graph.Iterator, callback goja.Value
 }
 
 func (s *Session) send(ctx context.Context, r *Result) bool {
-	if s.limit >= 0 && s.count >= s.limit {
+	if s.limit > 0 && s.count >= s.limit {
 		return false
 	}
 	if s.out == nil {
@@ -212,7 +210,7 @@ func (s *Session) send(ctx context.Context, r *Result) bool {
 		return false
 	}
 	s.count++
-	return s.limit < 0 || s.count < s.limit
+	return s.limit <= 0 || s.count < s.limit
 }
 
 func (s *Session) runIterator(it graph.Iterator) error {
@@ -256,20 +254,24 @@ func (r *Result) Result() interface{} {
 	}
 	return r.Val
 }
-func (r *Result) Err() error { return nil }
 
-func (s *Session) run(qu string) (v goja.Value, err error) {
+func (s *Session) compile(qu string) error {
 	var p *goja.Program
 	if s.last == qu && s.last != "" {
 		p = s.p
 	} else {
+		var err error
 		p, err = goja.Compile("", qu, false)
 		if err != nil {
-			return
+			return err
 		}
 		s.last, s.p = qu, p
 	}
-	v, err = s.vm.RunProgram(p)
+	return nil
+}
+
+func (s *Session) run() (goja.Value, error) {
+	v, err := s.vm.RunProgram(s.p)
 	if e, ok := err.(*goja.Exception); ok && e.Value() != nil {
 		if er, ok := e.Value().Export().(error); ok {
 			err = er
@@ -277,42 +279,123 @@ func (s *Session) run(qu string) (v goja.Value, err error) {
 	}
 	return v, err
 }
-func (s *Session) Execute(ctx context.Context, qu string, out chan query.Result, limit int) {
-	defer close(out)
-	s.out = out
-	s.limit = limit
+func (s *Session) Execute(ctx context.Context, qu string, opt query.Options) (query.Iterator, error) {
+	switch opt.Collation {
+	case query.Raw, query.JSON, query.REPL:
+	default:
+		return nil, &query.ErrUnsupportedCollation{Collation: opt.Collation}
+	}
+	if err := s.compile(qu); err != nil {
+		return nil, err
+	}
+	s.limit = opt.Limit
 	s.count = 0
+	ctx, cancel := context.WithCancel(context.Background())
 	s.ctx = ctx
-	done := make(chan struct{})
-	defer close(done)
-	go func() {
-		select {
-		case <-ctx.Done():
-			s.vm.Interrupt(ctx.Err())
-		case <-done:
-		}
-	}()
-	v, err := s.run(qu)
-	if err != nil {
-		select {
-		case <-ctx.Done():
-		case out <- query.ErrorResult(err):
-		}
+	return &results{
+		col: opt.Collation,
+		s:   s,
+		ctx: ctx, cancel: cancel,
+	}, nil
+}
+
+type results struct {
+	s      *Session
+	col    query.Collation
+	ctx    context.Context
+	cancel func()
+
+	running bool
+	errc    chan error
+
+	err error
+	cur *Result
+}
+
+func (it *results) stop(err error) {
+	it.cancel()
+	if !it.running {
 		return
 	}
-	if !goja.IsNull(v) && !goja.IsUndefined(v) {
-		s.send(ctx, &Result{Meta: true, Val: v.Export()})
+	it.s.vm.Interrupt(err)
+	it.running = false
+}
+
+func (it *results) Next(ctx context.Context) bool {
+	if it.errc == nil {
+		it.s.out = make(chan *Result)
+		it.errc = make(chan error, 1)
+		it.running = true
+		go func() {
+			defer close(it.errc)
+			v, err := it.s.run()
+			if err != nil {
+				it.errc <- err
+				return
+			}
+			if !goja.IsNull(v) && !goja.IsUndefined(v) {
+				it.s.send(it.ctx, &Result{Meta: true, Val: v.Export()})
+			}
+		}()
+	}
+	select {
+	case r := <-it.s.out:
+		it.cur = r
+		return true
+	case err := <-it.errc:
+		if err != nil {
+			it.err = err
+		}
+		return false
+	case <-ctx.Done():
+		it.err = ctx.Err()
+		it.stop(it.err)
+		return false
 	}
 }
 
-func (s *Session) FormatREPL(result query.Result) string {
-	if err := result.Err(); err != nil {
-		return fmt.Sprintf("error: %v", err)
+func (it *results) Result() interface{} {
+	if it.cur == nil {
+		return nil
 	}
-	data, ok := result.(*Result)
-	if !ok {
-		return fmt.Sprintf("Error: unexpected result type: %T\n", result)
+	switch it.col {
+	case query.Raw:
+		return it.cur
+	case query.JSON:
+		return it.jsonResult()
+	case query.REPL:
+		return it.replResult()
+	}
+	return nil
+}
+
+func (it *results) jsonResult() interface{} {
+	data := it.cur
+	if data.Meta {
+		return nil
+	}
+	if data.Val != nil {
+		return data.Val
+	}
+	obj := make(map[string]interface{})
+	tags := data.Tags
+	var tagKeys []string
+	for k := range tags {
+		tagKeys = append(tagKeys, k)
+	}
+	sort.Strings(tagKeys)
+	for _, k := range tagKeys {
+		if name := it.s.qs.NameOf(tags[k]); name != nil {
+			obj[k] = quadValueToNative(name)
+		} else {
+			delete(obj, k)
+		}
 	}
+	return obj
+}
+
+func (it *results) replResult() interface{} {
+	data := it.cur
 	if data.Meta {
 		if data.Val != nil {
 			s := data.Val
@@ -339,7 +422,7 @@ func (s *Session) FormatREPL(result query.Result) string {
 			if k == "$_" {
 				continue
 			}
-			out += fmt.Sprintf("%s : %s\n", k, quadValueToString(s.qs.NameOf(tags[k])))
+			out += fmt.Sprintf("%s : %s\n", k, quadValueToString(it.s.qs.NameOf(tags[k])))
 		}
 	} else {
 		switch export := data.Val.(type) {
@@ -358,59 +441,25 @@ func (s *Session) FormatREPL(result query.Result) string {
 	return out
 }
 
+func (it *results) Err() error {
+	return it.err
+}
+
+func (it *results) Close() error {
+	it.stop(errors.New("iterator closed"))
+	return nil
+}
+
 // Web stuff
 
 func (s *Session) ShapeOf(qu string) (interface{}, error) {
 	s.shape = make(map[string]interface{})
-	_, err := s.run(qu)
+	err := s.compile(qu)
+	if err != nil {
+		return nil, err
+	}
+	_, err = s.run()
 	out := s.shape
 	s.shape = nil
 	return out, err
 }
-
-func (s *Session) Collate(result query.Result) {
-	if err := result.Err(); err != nil {
-		s.err = err
-		return
-	}
-	data, ok := result.(*Result)
-	if !ok {
-		clog.Errorf("unexpected result type: %T", result)
-		return
-	} else if data.Meta {
-		return
-	}
-	if data.Val != nil {
-		s.dataOutput = append(s.dataOutput, data.Val)
-		return
-	}
-	obj := make(map[string]interface{})
-	tags := data.Tags
-	var tagKeys []string
-	for k := range tags {
-		tagKeys = append(tagKeys, k)
-	}
-	sort.Strings(tagKeys)
-	for _, k := range tagKeys {
-		if name := s.qs.NameOf(tags[k]); name != nil {
-			obj[k] = quadValueToNative(name)
-		} else {
-			delete(obj, k)
-		}
-	}
-	if len(obj) != 0 {
-		s.dataOutput = append(s.dataOutput, obj)
-	}
-}
-
-func (s *Session) Results() (interface{}, error) {
-	defer s.Clear()
-	if s.err != nil {
-		return nil, s.err
-	}
-	return s.dataOutput, nil
-}
-
-func (s *Session) Clear() {
-	s.dataOutput = nil
-}
diff --git a/query/graphql/graphql.go b/query/graphql/graphql.go
index d64ab75..5288872 100644
--- a/query/graphql/graphql.go
+++ b/query/graphql/graphql.go
@@ -53,42 +53,64 @@ func NewSession(qs graph.QuadStore) *Session {
 	return &Session{qs: qs}
 }
 
-type resultMap map[string]interface{}
-
-func (resultMap) Err() error            { return nil }
-func (m resultMap) Result() interface{} { return map[string]interface{}(m) }
-
 type Session struct {
 	qs graph.QuadStore
 }
 
-func (s *Session) Execute(ctx context.Context, qu string, out chan query.Result, limit int) {
-	defer close(out)
-	q, err := Parse(strings.NewReader(qu))
-	if err != nil {
-		select {
-		case out <- query.ErrorResult(err):
-		case <-ctx.Done():
-		}
-		return
+func (s *Session) Execute(ctx context.Context, qu string, opt query.Options) (query.Iterator, error) {
+	switch opt.Collation {
+	case query.Raw, query.JSON, query.REPL:
+	default:
+		return nil, &query.ErrUnsupportedCollation{Collation: opt.Collation}
 	}
-	m, err := q.Execute(ctx, s.qs)
-	var r query.Result
+	q, err := Parse(strings.NewReader(qu))
 	if err != nil {
-		r = query.ErrorResult(err)
-	} else {
-		r = resultMap(m)
+		return nil, err
 	}
-	select {
-	case out <- r:
-	case <-ctx.Done():
+	return &results{
+		s:   s,
+		q:   q,
+		col: opt.Collation,
+	}, nil
+}
+
+type results struct {
+	s   *Session
+	q   *Query
+	col query.Collation
+	res map[string]interface{}
+	err error
+}
+
+func (it *results) Next(ctx context.Context) bool {
+	if it.q == nil {
+		return false
 	}
+	it.res, it.err = it.q.Execute(ctx, it.s.qs)
+	it.q = nil
+	return it.err == nil && len(it.res) != 0
 }
-func (s *Session) FormatREPL(result query.Result) string {
-	data, _ := json.MarshalIndent(result, "", "   ")
+
+func (it *results) Result() interface{} {
+	if len(it.res) == 0 {
+		return nil
+	}
+	if it.col != query.REPL {
+		return it.res
+	}
+	data, _ := json.MarshalIndent(it.res, "", "   ")
 	return string(data)
 }
 
+func (it *results) Err() error {
+	return it.err
+}
+
+func (it *results) Close() error {
+	it.q = nil
+	return nil
+}
+
 // Configurable keywords and special field names.
 var (
 	ValueKey = "id"
diff --git a/query/mql/session.go b/query/mql/session.go
index 883b740..0c74c2d 100644
--- a/query/mql/session.go
+++ b/query/mql/session.go
@@ -17,8 +17,6 @@ package mql
 import (
 	"context"
 	"encoding/json"
-	"fmt"
-	"sort"
 
 	"github.com/cayleygraph/cayley/graph"
 	"github.com/cayleygraph/cayley/graph/iterator"
@@ -43,8 +41,7 @@ func init() {
 }
 
 type Session struct {
-	qs    graph.QuadStore
-	query *Query
+	qs graph.QuadStore
 }
 
 func NewSession(qs graph.QuadStore) *Session {
@@ -57,10 +54,10 @@ func (s *Session) ShapeOf(query string) (interface{}, error) {
 	if err != nil {
 		return nil, err
 	}
-	s.query = NewQuery(s)
-	s.query.BuildIteratorTree(mqlQuery)
+	q := NewQuery(s)
+	q.BuildIteratorTree(mqlQuery)
 	output := make(map[string]interface{})
-	iterator.OutputQueryShapeForIterator(s.query.it, s.qs, output)
+	iterator.OutputQueryShapeForIterator(q.it, s.qs, output)
 	nodes := make([]iterator.Node, 0)
 	for _, n := range output["nodes"].([]iterator.Node) {
 		n.Tags = nil
@@ -70,81 +67,80 @@ func (s *Session) ShapeOf(query string) (interface{}, error) {
 	return output, nil
 }
 
-func (s *Session) Execute(ctx context.Context, input string, c chan query.Result, limit int) {
-	defer close(c)
-	var mqlQuery interface{}
-	if err := json.Unmarshal([]byte(input), &mqlQuery); err != nil {
-		select {
-		case c <- query.ErrorResult(err):
-		case <-ctx.Done():
+type mqlIterator struct {
+	q   *Query
+	col query.Collation
+	it  graph.Iterator
+	res []interface{}
+}
+
+func (it *mqlIterator) Next(ctx context.Context) bool {
+	// TODO: stream results
+	if it.res != nil {
+		if len(it.res) == 0 {
+			return false
 		}
-		return
+		it.res = it.res[1:]
+		return len(it.res) != 0
 	}
-	s.query = NewQuery(s)
-	s.query.BuildIteratorTree(mqlQuery)
-	if s.query.isError() {
-		select {
-		case c <- query.ErrorResult(s.query.err):
-		case <-ctx.Done():
+	for it.it.Next(ctx) {
+		m := make(map[string]graph.Ref)
+		it.it.TagResults(m)
+		it.q.treeifyResult(m)
+		for it.it.NextPath(ctx) {
+			m = make(map[string]graph.Ref, len(m))
+			it.it.TagResults(m)
+			it.q.treeifyResult(m)
 		}
-		return
 	}
-
-	it := s.query.it
-	err := graph.Iterate(ctx, it).Limit(limit).TagEach(func(tags map[string]graph.Ref) {
-		select {
-		case c <- query.TagMapResult(tags):
-		case <-ctx.Done():
-		}
-	})
-	if err != nil {
-		select {
-		case c <- query.ErrorResult(err):
-		case <-ctx.Done():
-		}
+	if err := it.it.Err(); err != nil {
+		return false
 	}
+	it.q.buildResults()
+	it.res = it.q.results
+	return len(it.res) != 0
 }
 
-func (s *Session) FormatREPL(result query.Result) string {
-	tags, ok := result.Result().(map[string]graph.Ref)
-	if !ok {
-		return ""
-	}
-	out := fmt.Sprintln("****")
-	tagKeys := make([]string, len(tags))
-	s.query.treeifyResult(tags)
-	s.query.buildResults()
-	r, _ := json.MarshalIndent(s.query.results, "", " ")
-	fmt.Println(string(r))
-	i := 0
-	for k := range tags {
-		tagKeys[i] = string(k)
-		i++
-	}
-	sort.Strings(tagKeys)
-	for _, k := range tagKeys {
-		if k == "$_" {
-			continue
-		}
-		out += fmt.Sprintf("%s : %s\n", k, s.qs.NameOf(tags[k]))
+func (it *mqlIterator) Result() interface{} {
+	if len(it.res) == 0 {
+		return nil
 	}
-	return out
+	return it.res[0]
 }
 
-func (s *Session) Collate(result query.Result) {
-	res, ok := result.Result().(map[string]graph.Ref)
-	if !ok {
-		return
-	}
-	s.query.treeifyResult(res)
+func (it *mqlIterator) Err() error {
+	return it.it.Err()
+}
+
+func (it *mqlIterator) Close() error {
+	return it.it.Close()
 }
 
-func (s *Session) Results() (interface{}, error) {
-	s.query.buildResults()
-	if s.query.isError() {
-		return nil, s.query.err
+func (s *Session) Execute(ctx context.Context, input string, opt query.Options) (query.Iterator, error) {
+	switch opt.Collation {
+	case query.REPL, query.JSON:
+	default:
+		return nil, &query.ErrUnsupportedCollation{Collation: opt.Collation}
+	}
+	var mqlQuery interface{}
+	if err := json.Unmarshal([]byte(input), &mqlQuery); err != nil {
+		return nil, err
+	}
+	q := NewQuery(s)
+	q.BuildIteratorTree(mqlQuery)
+	if q.isError() {
+		return nil, q.err
+	}
+
+	it := q.it
+	if opt.Limit > 0 {
+		it = iterator.NewLimit(it, int64(opt.Limit))
 	}
-	return s.query.results, nil
+	return &mqlIterator{
+		q:   q,
+		col: opt.Collation,
+		it:  it,
+	}, nil
 }
 
 func (s *Session) Clear() {
diff --git a/query/session.go b/query/session.go
index c200f1f..2fbe2bc 100644
--- a/query/session.go
+++ b/query/session.go
@@ -18,6 +18,7 @@ package query
 import (
 	"context"
 	"errors"
+	"fmt"
 	"io"
 
 	"github.com/cayleygraph/cayley/graph"
@@ -25,53 +26,62 @@ import (
 
 var ErrParseMore = errors.New("query: more input required")
 
-type Result interface {
-	Result() interface{}
-	Err() error
+type ErrUnsupportedCollation struct {
+	Collation Collation
 }
 
-func ErrorResult(err error) Result {
-	return errResult{err: err}
+func (e *ErrUnsupportedCollation) Error() string {
+	return fmt.Sprintf("unsupported collation: %v", e.Collation)
 }
 
-type errResult struct {
-	err error
+// Iterator for query results.
+type Iterator interface {
+	// Next advances the iterator to the next value, which will then be available through
+	// the Result method. It returns false if no further advancement is possible, or if an
+	// error was encountered during iteration.  Err should be consulted to distinguish
+	// between the two cases.
+	Next(ctx context.Context) bool
+	// Results returns the current result. The type depends on the collation mode of the query.
+	Result() interface{}
+	// Err returns any error that was encountered by the Iterator.
+	Err() error
+	// Close the iterator and do internal cleanup.
+	Close() error
 }
 
-func (errResult) Result() interface{} { return nil }
-func (e errResult) Err() error        { return e.err }
+// Collation of results.
+type Collation int
 
-func TagMapResult(m map[string]graph.Ref) Result {
-	return tagMap(m)
-}
-
-type tagMap map[string]graph.Ref
+const (
+	// Raw collates results as maps or arrays of graph.Refs or any other query-native or graph-native data type.
+	Raw = Collation(iota)
+	// REPL collates results as strings which will be used in CLI.
+	REPL = Collation(iota)
+	// JSON collates results as maps, arrays and values, that can be encoded to JSON.
+	JSON
+)
 
-func (m tagMap) Result() interface{} { return map[string]graph.Ref(m) }
-func (tagMap) Err() error            { return nil }
+// Options for the query execution.
+type Options struct {
+	Limit     int
+	Collation Collation
+}
 
 type Session interface {
-	// Runs the query and returns individual results on the channel.
-	//
-	// Channel will be closed when function returns.
-	Execute(ctx context.Context, query string, out chan Result, limit int)
+	// Execute runs the query and returns an iterator over the results.
+	// Type of results depends on Collation. See Options for details.
+	Execute(ctx context.Context, query string, opt Options) (Iterator, error)
 }
 
-// TODO(dennwc): review HTTP interface (Collate is weird)
 // TODO(dennwc): specify exact type to return from ShapeOf
 // TODO(dennwc): add context to ShapeOf?
 
 type HTTP interface {
 	Session
 	ShapeOf(string) (interface{}, error)
-	Collate(Result)
-	Results() (interface{}, error)
 }
 
-type REPLSession interface {
-	Session
-	FormatREPL(Result) string
-}
+type REPLSession = Session
 
 // ResponseWriter is a subset of http.ResponseWriter
 type ResponseWriter interface {
@@ -83,7 +93,7 @@ type ResponseWriter interface {
 type Language struct {
 	Name    string
 	Session func(graph.QuadStore) Session
-	REPL    func(graph.QuadStore) REPLSession
+	REPL    func(graph.QuadStore) REPLSession // deprecated
 	HTTP    func(graph.QuadStore) HTTP
 
 	// Custom HTTP handlers
@@ -126,3 +136,14 @@ func Languages() []string {
 	}
 	return out
 }
+
+// Execute runs the query in a given language and returns an iterator over the results.
+// Type of results depends on Collation. See Options for details.
+func Execute(ctx context.Context, qs graph.QuadStore, lang, query string, opt Options) (Iterator, error) {
+	l := GetLanguage(lang)
+	if l == nil {
+		return nil, fmt.Errorf("unsupported language: %q", lang)
+	}
+	sess := l.Session(qs)
+	return sess.Execute(ctx, query, opt)
+}
diff --git a/query/sexp/session.go b/query/sexp/session.go
index 49aa2f5..71f88d0 100644
--- a/query/sexp/session.go
+++ b/query/sexp/session.go
@@ -23,6 +23,7 @@ import (
 	"sort"
 
 	"github.com/cayleygraph/cayley/graph"
+	"github.com/cayleygraph/cayley/graph/iterator"
 	"github.com/cayleygraph/cayley/query"
 )
 
@@ -74,32 +75,55 @@ func (s *Session) Parse(input string) error {
 	return errors.New("invalid syntax")
 }
 
-func (s *Session) Execute(ctx context.Context, input string, out chan query.Result, limit int) {
-	defer close(out)
+func (s *Session) Execute(ctx context.Context, input string, opt query.Options) (query.Iterator, error) {
+	switch opt.Collation {
+	case query.Raw, query.REPL:
+	default:
+		return nil, &query.ErrUnsupportedCollation{Collation: opt.Collation}
+	}
 	it := BuildIteratorTreeForQuery(s.qs, input)
-	err := graph.Iterate(ctx, it).Paths(true).Limit(limit).TagEach(func(tags map[string]graph.Ref) {
-		select {
-		case out <- query.TagMapResult(tags):
-		case <-ctx.Done():
-		}
-	})
-	if err != nil {
-		select {
-		case out <- query.ErrorResult(err):
-		case <-ctx.Done():
-		}
+	if err := it.Err(); err != nil {
+		return nil, err
 	}
+	if opt.Limit > 0 {
+		it = iterator.NewLimit(it, int64(opt.Limit))
+	}
+	return &results{
+		s:   s,
+		col: opt.Collation,
+		it:  it,
+	}, nil
+}
+
+type results struct {
+	s        *Session
+	col      query.Collation
+	it       graph.Iterator
+	nextPath bool
 }
 
-func (s *Session) FormatREPL(result query.Result) string {
-	out := fmt.Sprintln("****")
-	tags, ok := result.Result().(map[string]graph.Ref)
-	if !ok {
-		return ""
+func (it *results) Next(ctx context.Context) bool {
+	if it.nextPath && it.it.NextPath(ctx) {
+		return true
+	}
+	it.nextPath = false
+	if it.it.Next(ctx) {
+		it.nextPath = true
+		return true
 	}
-	tagKeys := make([]string, len(tags))
+	return false
+}
+
+func (it *results) Result() interface{} {
+	m := make(map[string]graph.Ref)
+	it.it.TagResults(m)
+	if it.col == query.Raw {
+		return m
+	}
+	out := "****\n"
+	tagKeys := make([]string, len(m))
 	i := 0
-	for k := range tags {
+	for k := range m {
 		tagKeys[i] = k
 		i++
 	}
@@ -108,7 +132,15 @@ func (s *Session) FormatREPL(result query.Result) string {
 		if k == "$_" {
 			continue
 		}
-		out += fmt.Sprintf("%s : %s\n", k, s.qs.NameOf(tags[k]))
+		out += fmt.Sprintf("%s : %s\n", k, it.s.qs.NameOf(m[k]))
 	}
 	return out
 }
+
+func (it *results) Err() error {
+	return it.it.Err()
+}
+
+func (it *results) Close() error {
+	return it.it.Close()
+}
diff --git a/server/http/api_v2.go b/server/http/api_v2.go
index da3f65b..ac4903d 100644
--- a/server/http/api_v2.go
+++ b/server/http/api_v2.go
@@ -488,23 +488,23 @@ func (api *APIv2) ServeQuery(w http.ResponseWriter, r *http.Request) {
 		clog.Infof("query: %s: %q", lang, qu)
 	}
 
-	c := make(chan query.Result, 5)
-	go ses.Execute(ctx, qu, c, api.limit)
+	it, err := ses.Execute(ctx, qu, query.Options{
+		Collation: query.JSON,
+		Limit:     api.limit,
+	})
+	if err != nil {
+		errFunc(w, err)
+		return
+	}
+	defer it.Close()
 
-	for res := range c {
-		if err := res.Err(); err != nil {
-			if err == nil {
-				continue // wait for results channel to close
-			}
-			errFunc(w, err)
-			return
-		}
-		ses.Collate(res)
+	var out []interface{}
+	for it.Next(ctx) {
+		out = append(out, it.Result())
 	}
-	output, err := ses.Results()
-	if err != nil {
+	if err = it.Err(); err != nil {
 		errFunc(w, err)
 		return
 	}
-	writeResults(w, output)
+	writeResults(w, out)
 }


Test patch:
diff --git a/graph/graphtest/integration.go b/graph/graphtest/integration.go
index a3d1edd..400062c 100644
--- a/graph/graphtest/integration.go
+++ b/graph/graphtest/integration.go
@@ -507,26 +507,22 @@ func checkQueries(t *testing.T, qs graph.QuadStore, timeout time.Duration) {
 			}
 			start := time.Now()
 			ses := gizmo.NewSession(qs)
-			c := make(chan query.Result, 5)
 			ctx := context.Background()
 			if timeout > 0 {
 				var cancel func()
 				ctx, cancel = context.WithTimeout(ctx, timeout)
 				defer cancel()
 			}
-			go ses.Execute(ctx, test.query, c, -1)
+			it, err := ses.Execute(ctx, test.query, query.Options{
+				Collation: query.JSON,
+			})
+			if err != nil {
+				t.Fatal(err)
+			}
+			defer it.Close()
 			var got []interface{}
-			for r := range c {
-				if err := r.Err(); err != nil {
-					t.Error("Error:", err)
-					continue
-				}
-				ses.Collate(r)
-				j, err := ses.Results()
-				if j == nil && err == nil {
-					continue
-				}
-				got = append(got, j.([]interface{})...)
+			for it.Next(ctx) {
+				got = append(got, it.Result())
 			}
 			t.Logf("%12v %v", time.Since(start), test.message)
 
@@ -580,26 +576,34 @@ func benchmarkQueries(b *testing.B, gen testutil.DatabaseFunc) {
 			b.StopTimer()
 			b.ResetTimer()
 			for i := 0; i < b.N; i++ {
-				c := make(chan query.Result, 5)
-				ctx := context.Background()
-				var cancel func()
-				if timeout > 0 {
-					ctx, cancel = context.WithTimeout(ctx, timeout)
-				}
-				ses := gizmo.NewSession(qs)
-				b.StartTimer()
-				go ses.Execute(ctx, bench.query, c, -1)
-				n := 0
-				for range c {
-					n++
-				}
-				b.StopTimer()
-				if n != len(bench.expect) {
-					b.Fatalf("unexpected number of results: %d vs %d", n, len(bench.expect))
-				}
-				if cancel != nil {
-					cancel()
-				}
+				func() {
+					ctx := context.Background()
+					if timeout > 0 {
+						var cancel func()
+						ctx, cancel = context.WithTimeout(ctx, timeout)
+						defer cancel()
+					}
+					ses := gizmo.NewSession(qs)
+					b.StartTimer()
+					it, err := ses.Execute(ctx, bench.query, query.Options{
+						Collation: query.Raw,
+					})
+					if err != nil {
+						b.Fatal(err)
+					}
+					defer it.Close()
+					n := 0
+					for it.Next(ctx) {
+						n++
+					}
+					if err = it.Err(); err != nil {
+						b.Fatal(err)
+					}
+					b.StopTimer()
+					if n != len(bench.expect) {
+						b.Fatalf("unexpected number of results: %d vs %d", n, len(bench.expect))
+					}
+				}()
 			}
 		})
 	}
diff --git a/query/gizmo/gizmo_test.go b/query/gizmo/gizmo_test.go
index 9ab72d8..fc46ded 100644
--- a/query/gizmo/gizmo_test.go
+++ b/query/gizmo/gizmo_test.go
@@ -622,18 +622,20 @@ var testQueries = []struct {
 
 func runQueryGetTag(rec func(), g []quad.Quad, qu string, tag string, limit int) ([]string, error) {
 	js := makeTestSession(g)
-	c := make(chan query.Result, 1)
-	go func() {
-		defer rec()
-		js.Execute(context.TODO(), qu, c, limit)
-	}()
+	ctx := context.TODO()
+	it, err := js.Execute(ctx, qu, query.Options{
+		Collation: query.Raw,
+		Limit:     limit,
+	})
+	if err != nil {
+		return nil, err
+	}
+	defer it.Close()
+	defer rec()
 
 	var results []string
-	for res := range c {
-		if err := res.Err(); err != nil {
-			return results, err
-		}
-		data := res.(*Result)
+	for it.Next(ctx) {
+		data := it.Result().(*Result)
 		if data.Val == nil {
 			if val := data.Tags[tag]; val != nil {
 				results = append(results, quadValueToString(js.qs.NameOf(val)))
@@ -647,6 +649,9 @@ func runQueryGetTag(rec func(), g []quad.Quad, qu string, tag string, limit int)
 			}
 		}
 	}
+	if err := it.Err(); err != nil {
+		return results, err
+	}
 	return results, nil
 }
 
@@ -718,17 +723,24 @@ func TestIssue160(t *testing.T) {
 	}
 
 	ses := makeTestSession(issue160TestGraph)
-	c := make(chan query.Result, 5)
-	go ses.Execute(context.TODO(), qu, c, 100)
+	ctx := context.TODO()
+	it, err := ses.Execute(ctx, qu, query.Options{
+		Collation: query.REPL,
+		Limit:     100,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer it.Close()
 	var got []string
-	for res := range c {
+	for it.Next(ctx) {
 		func() {
 			defer func() {
 				if r := recover(); r != nil {
 					t.Errorf("Unexpected panic: %v", r)
 				}
 			}()
-			got = append(got, ses.FormatREPL(res))
+			got = append(got, it.Result().(string))
 		}()
 	}
 	sort.Strings(got)
diff --git a/query/mql/mql_test.go b/query/mql/mql_test.go
index 8c62e1a..5e41653 100644
--- a/query/mql/mql_test.go
+++ b/query/mql/mql_test.go
@@ -17,9 +17,10 @@ package mql
 import (
 	"context"
 	"encoding/json"
-	"reflect"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
 	"github.com/cayleygraph/cayley/graph"
 	"github.com/cayleygraph/cayley/graph/graphtest/testutil"
 	_ "github.com/cayleygraph/cayley/graph/memstore"
@@ -160,31 +161,29 @@ var testQueries = []struct {
 	},
 }
 
-func runQuery(g []quad.Quad, qu string) interface{} {
+func runQuery(t testing.TB, g []quad.Quad, qu string) interface{} {
 	s := makeTestSession(g)
-	c := make(chan query.Result, 5)
-	go s.Execute(context.TODO(), qu, c, -1)
-	for result := range c {
-		s.Collate(result)
+	ctx := context.TODO()
+	it, err := s.Execute(ctx, qu, query.Options{Collation: query.JSON})
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer it.Close()
+	var out []interface{}
+	for it.Next(ctx) {
+		out = append(out, it.Result())
 	}
-	result, _ := s.Results()
-	return result
+	return out
 }
 
 func TestMQL(t *testing.T) {
 	simpleGraph := testutil.LoadGraph(t, "../../data/testdata.nq")
 	for _, test := range testQueries {
 		t.Run(test.message, func(t *testing.T) {
-			got := runQuery(simpleGraph, test.query)
+			got := runQuery(t, simpleGraph, test.query)
 			var expect interface{}
 			json.Unmarshal([]byte(test.expect), &expect)
-			if !reflect.DeepEqual(got, expect) {
-				b, err := json.MarshalIndent(got, "", " ")
-				if err != nil {
-					t.Fatalf("unexpected JSON marshal error: %v", err)
-				}
-				t.Errorf("got: %s expected: %s", b, test.expect)
-			}
+			require.Equal(t, expect, got)
 		})
 	}
 }