Repository: cayleygraph/cayley (go)

Issue:
NextPath is broken on Postgres
**Description**

https://discourse.cayley.io/t/path-save-lost-some-data-when-using-sql-backends/1192

**Steps to reproduce the issue:**
1. Initialize a clean PG instance
2. Load test data
3. Execute the query:
```
g.V("<dani>", "<bob>").Save("<follows>", "target").All()
```

**Received results:**

```json
{"id":"bob","target":"fred"}
{"id":"dani","target":"greg"}
```

**Expected results:**

```json
{"id":"bob","target":"fred"}
{"id":"dani","target":"bob"}
{"id":"dani","target":"greg"}
```

**Environment details:**

Backend database: `postgres`


Pull request:
Fix NextPath for optimized queries in SQL
When optimizing queries for SQL, we missed the fact that `NodesFrom` and derived queries make use of `NextPath` which is not implemented on SQL iterators.

This PR fixes it by emulating the same behavior for both `Next` and `Contains`.

Fixes #724

<!-- 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/848)
<!-- Reviewable:end -->


Patch:
diff --git a/graph/sql/iterator.go b/graph/sql/iterator.go
index 3f2c7f7..58b43b0 100644
--- a/graph/sql/iterator.go
+++ b/graph/sql/iterator.go
@@ -255,89 +255,56 @@ func newIteratorNext(qs *QuadStore, s Select) *iteratorNext {
 type iteratorNext struct {
 	iteratorBase
 	cursor *sql.Rows
+	// TODO(dennwc): nextPath workaround; remove when we get rid of NextPath in general
+	nextPathRes  graph.Ref
+	nextPathTags map[string]graph.Ref
 }
 
-func (it *iteratorNext) TagResults(m map[string]graph.Ref) {
-	for tag, val := range it.tags {
-		m[tag] = val
+func (it *iteratorNext) Next(ctx context.Context) bool {
+	if it.err != nil {
+		return false
 	}
-}
-
-func (it *iteratorNext) Result() graph.Ref {
-	return it.res
-}
-
-func (it *iteratorNext) ensureColumns() {
-	if it.cols != nil {
-		return
+	if it.cursor == nil {
+		it.cursor, it.err = it.qs.Query(ctx, it.query)
 	}
-	it.cols = it.query.Columns()
-	it.cind = make(map[quad.Direction]int, len(quad.Directions)+1)
-	for i, name := range it.cols {
-		if !strings.HasPrefix(name, tagPref) {
-			continue
+	// TODO(dennwc): this loop exists only because of nextPath workaround
+	for {
+		if it.err != nil {
+			return false
 		}
-		if name == tagNode {
-			it.cind[quad.Any] = i
-			continue
+		if it.nextPathRes != nil {
+			it.res = it.nextPathRes
+			it.tags = it.nextPathTags
+			it.nextPathRes = nil
+			it.nextPathTags = nil
+			return true
 		}
-		name = name[len(tagPref):]
-		for _, d := range quad.Directions {
-			if name == d.String() {
-				it.cind[d] = i
-				break
-			}
+		if !it.cursor.Next() {
+			it.err = it.cursor.Err()
+			it.cursor.Close()
+			return false
 		}
-	}
-}
 
-func (it *iteratorNext) scanValue(r *sql.Rows) bool {
-	it.ensureColumns()
-	nodes := make([]NodeHash, len(it.cols))
-	pointers := make([]interface{}, len(nodes))
-	for i := range pointers {
-		pointers[i] = &nodes[i]
-	}
-	if err := r.Scan(pointers...); err != nil {
-		it.err = err
-		return false
-	}
-	it.tags = make(map[string]graph.Ref)
-	for i, name := range it.cols {
-		if !strings.Contains(name, tagPref) {
-			it.tags[name] = nodes[i].ValueHash
+		prev := it.res
+		if !it.scanValue(it.cursor) {
+			return false
 		}
-	}
-	if len(it.cind) > 1 {
-		var q QuadHashes
-		for _, d := range quad.Directions {
-			i, ok := it.cind[d]
-			if !ok {
-				it.err = fmt.Errorf("cannot find quad %v in query output (columns: %v)", d, it.cols)
-				return false
-			}
-			q.Set(d, nodes[i].ValueHash)
+		if !it.query.nextPath {
+			return true
 		}
-		it.res = q
-		return true
-	}
-	i, ok := it.cind[quad.Any]
-	if !ok {
-		it.err = fmt.Errorf("cannot find node hash in query output (columns: %v, cind: %v)", it.cols, it.cind)
-		return false
+		if prev == nil || prev.Key() != it.res.Key() {
+			return true
+		}
+		// skip the same main key if in nextPath mode
+		// the user should receive accept those results via NextPath of the iterator
 	}
-	it.res = nodes[i]
-	return true
 }
 
-func (it *iteratorNext) Next(ctx context.Context) bool {
+func (it *iteratorNext) NextPath(ctx context.Context) bool {
 	if it.err != nil {
 		return false
 	}
-	if it.cursor == nil {
-		it.cursor, it.err = it.qs.Query(ctx, it.query)
-	}
-	if it.err != nil {
+	if !it.query.nextPath {
 		return false
 	}
 	if !it.cursor.Next() {
@@ -345,21 +312,21 @@ func (it *iteratorNext) Next(ctx context.Context) bool {
 		it.cursor.Close()
 		return false
 	}
-	return it.scanValue(it.cursor)
-}
-
-func (it *iteratorNext) NextPath(ctx context.Context) bool {
+	prev := it.res
+	if !it.scanValue(it.cursor) {
+		return false
+	}
+	if prev.Key() == it.res.Key() {
+		return true
+	}
+	// different main keys - return false, but keep this results for the Next
+	it.nextPathRes = it.res
+	it.nextPathTags = it.tags
+	it.res = nil
+	it.tags = nil
 	return false
 }
 
-func (it *iteratorNext) Err() error {
-	return it.err
-}
-
-func (it *iteratorNext) String() string {
-	return it.query.SQL(NewBuilder(it.qs.flavor.QueryDialect))
-}
-
 func (it *iteratorNext) Close() error {
 	if it.cursor != nil {
 		it.cursor.Close()
@@ -376,83 +343,8 @@ func newIteratorContains(qs *QuadStore, s Select) *iteratorContains {
 
 type iteratorContains struct {
 	iteratorBase
-}
-
-func (it *iteratorContains) TagResults(m map[string]graph.Ref) {
-	for tag, val := range it.tags {
-		m[tag] = val
-	}
-}
-
-func (it *iteratorContains) Result() graph.Ref {
-	return it.res
-}
-
-func (it *iteratorContains) ensureColumns() {
-	if it.cols != nil {
-		return
-	}
-	it.cols = it.query.Columns()
-	it.cind = make(map[quad.Direction]int, len(quad.Directions)+1)
-	for i, name := range it.cols {
-		if !strings.HasPrefix(name, tagPref) {
-			continue
-		}
-		if name == tagNode {
-			it.cind[quad.Any] = i
-			continue
-		}
-		name = name[len(tagPref):]
-		for _, d := range quad.Directions {
-			if name == d.String() {
-				it.cind[d] = i
-				break
-			}
-		}
-	}
-}
-
-func (it *iteratorContains) scanValue(r *sql.Rows) bool {
-	it.ensureColumns()
-	nodes := make([]NodeHash, len(it.cols))
-	pointers := make([]interface{}, len(nodes))
-	for i := range pointers {
-		pointers[i] = &nodes[i]
-	}
-	if err := r.Scan(pointers...); err != nil {
-		it.err = err
-		return false
-	}
-	it.tags = make(map[string]graph.Ref)
-	for i, name := range it.cols {
-		if !strings.Contains(name, tagPref) {
-			it.tags[name] = nodes[i].ValueHash
-		}
-	}
-	if len(it.cind) > 1 {
-		var q QuadHashes
-		for _, d := range quad.Directions {
-			i, ok := it.cind[d]
-			if !ok {
-				it.err = fmt.Errorf("cannot find quad %v in query output (columns: %v)", d, it.cols)
-				return false
-			}
-			q.Set(d, nodes[i].ValueHash)
-		}
-		it.res = q
-		return true
-	}
-	i, ok := it.cind[quad.Any]
-	if !ok {
-		it.err = fmt.Errorf("cannot find node hash in query output (columns: %v, cind: %v)", it.cols, it.cind)
-		return false
-	}
-	it.res = nodes[i]
-	return true
-}
-
-func (it *iteratorContains) NextPath(ctx context.Context) bool {
-	return false
+	// TODO(dennwc): nextPath workaround; remove when we get rid of NextPath in general
+	nextPathRows *sql.Rows
 }
 
 func (it *iteratorContains) Contains(ctx context.Context, v graph.Ref) bool {
@@ -489,7 +381,14 @@ func (it *iteratorContains) Contains(ctx context.Context, v graph.Ref) bool {
 		it.err = err
 		return false
 	}
-	defer rows.Close()
+	if it.query.nextPath {
+		if it.nextPathRows != nil {
+			_ = it.nextPathRows.Close()
+		}
+		it.nextPathRows = rows
+	} else {
+		defer rows.Close()
+	}
 	if !rows.Next() {
 		it.err = rows.Err()
 		return false
@@ -497,14 +396,23 @@ func (it *iteratorContains) Contains(ctx context.Context, v graph.Ref) bool {
 	return it.scanValue(rows)
 }
 
-func (it *iteratorContains) Err() error {
-	return it.err
-}
-
-func (it *iteratorContains) String() string {
-	return it.query.SQL(NewBuilder(it.qs.flavor.QueryDialect))
+func (it *iteratorContains) NextPath(ctx context.Context) bool {
+	if it.err != nil {
+		return false
+	}
+	if !it.query.nextPath {
+		return false
+	}
+	if !it.nextPathRows.Next() {
+		it.err = it.nextPathRows.Err()
+		return false
+	}
+	return it.scanValue(it.nextPathRows)
 }
 
 func (it *iteratorContains) Close() error {
+	if it.nextPathRows != nil {
+		return it.nextPathRows.Close()
+	}
 	return nil
 }
diff --git a/graph/sql/optimizer.go b/graph/sql/optimizer.go
index 6d84750..30a436e 100644
--- a/graph/sql/optimizer.go
+++ b/graph/sql/optimizer.go
@@ -399,6 +399,8 @@ func (opt *Optimizer) optimizeNodesFrom(s shape.NodesFrom) (shape.Shape, bool) {
 	if !found {
 		return s, false
 	}
+	// NodesFrom implies that the iterator will use NextPath
+	sel.nextPath = true
 	return sel, true
 }
 
@@ -410,6 +412,8 @@ func (opt *Optimizer) optimizeQuadsAction(s shape.QuadsAction) (shape.Shape, boo
 		From: []Source{
 			Table{Name: "quads"},
 		},
+		// NodesFrom (that is a part of QuadsAction) implies that the iterator will use NextPath
+		nextPath: true,
 	}
 	var dirs []quad.Direction
 	for d := range s.Save {
@@ -522,11 +526,13 @@ func (opt *Optimizer) optimizeIntersect(s shape.Intersect) (shape.Shape, bool) {
 	}
 	sec := sels[1:]
 
+	nextPath := false
 	for _, s2 := range sec {
 		// merge From, Where and Params
 		pri.From = append(pri.From, s2.From...)
 		pri.Where = append(pri.Where, s2.Where...)
 		pri.Params = append(pri.Params, s2.Params...)
+		nextPath = nextPath || s2.nextPath
 		// also find and remove primary tag, but add the same field to WHERE
 		ok := false
 		for _, f := range s2.Fields {
@@ -550,6 +556,7 @@ func (opt *Optimizer) optimizeIntersect(s shape.Intersect) (shape.Shape, bool) {
 		}
 	}
 	if len(other) == 1 {
+		pri.nextPath = pri.nextPath || nextPath
 		return pri, true
 	}
 	other[0] = pri
diff --git a/graph/sql/shape.go b/graph/sql/shape.go
index 1a0af89..4b98ee1 100644
--- a/graph/sql/shape.go
+++ b/graph/sql/shape.go
@@ -266,6 +266,10 @@ type Select struct {
 	Params []Value
 	Limit  int64
 	Offset int64
+
+	// TODO(dennwc): this field in unexported because we don't want it to a be a part of the API
+	//               however, it's necessary to make NodesFrom optimizations to work with SQL
+	nextPath bool
 }
 
 func (s Select) Clone() Select {


Test patch:
diff --git a/graph/path/pathtest/pathtest.go b/graph/path/pathtest/pathtest.go
index aa7f683..14ac270 100644
--- a/graph/path/pathtest/pathtest.go
+++ b/graph/path/pathtest/pathtest.go
@@ -236,6 +236,18 @@ func testSet(qs graph.QuadStore) []test {
 			tag:     "who",
 			expect:  []quad.Value{vGreg, vDani, vBob},
 		},
+		{
+			message: "save with a next path",
+			path:    StartPath(qs, vDani, vBob).Save(vFollows, "target"),
+			tag:     "target",
+			expect:  []quad.Value{vBob, vFred, vGreg},
+		},
+		{
+			message: "save all with a next path",
+			path:    StartPath(qs).Save(vFollows, "target"),
+			tag:     "target",
+			expect:  []quad.Value{vBob, vBob, vBob, vDani, vFred, vFred, vGreg, vGreg},
+		},
 		{
 			message: "simple Has",
 			path:    StartPath(qs).Has(vStatus, vCool),