From 831b14c23e4126f2764c121ff7451efd26be27fe Mon Sep 17 00:00:00 2001
From: Barak Michener <me@barakmich.com>
Date: Thu, 12 Nov 2015 14:46:31 -0500
Subject: [PATCH] graph/sql: Use an IN clause when optimizing a fixed iterator

---
 graph/sql/optimizers.go        | 30 ++++++++++++++++++++++++++++++
 graph/sql/sql_link_iterator.go | 14 ++++++++++++--
 2 files changed, 42 insertions(+), 2 deletions(-)

diff --git a/graph/sql/optimizers.go b/graph/sql/optimizers.go
index 538ddd7..c54c973 100644
--- a/graph/sql/optimizers.go
+++ b/graph/sql/optimizers.go
@@ -169,6 +169,36 @@ func (qs *QuadStore) optimizeLinksTo(it *iterator.LinksTo) (graph.Iterator, bool
 			}
 			it.Close()
 			return newIt, true
+		} else if size > 1 {
+			var vals []string
+			for graph.Next(primary) {
+				vals = append(vals, qs.NameOf(primary.Result()))
+			}
+			lsql := &SQLLinkIterator{
+				constraints: []constraint{
+					constraint{
+						dir:  it.Direction(),
+						vals: vals,
+					},
+				},
+				tableName: newTableName(),
+				size:      0,
+			}
+			l := &SQLIterator{
+				uid: iterator.NextUID(),
+				qs:  qs,
+				sql: lsql,
+			}
+			nt := l.Tagger()
+			nt.CopyFrom(it)
+			for _, t := range primary.Tagger().Tags() {
+				lsql.tagdirs = append(lsql.tagdirs, tagDir{
+					dir: it.Direction(),
+					tag: t,
+				})
+			}
+			it.Close()
+			return l, true
 		}
 	case sqlType:
 		p := primary.(*SQLIterator)
diff --git a/graph/sql/sql_link_iterator.go b/graph/sql/sql_link_iterator.go
index f2750c3..79b3569 100644
--- a/graph/sql/sql_link_iterator.go
+++ b/graph/sql/sql_link_iterator.go
@@ -220,8 +220,18 @@ func (l *SQLLinkIterator) buildWhere() (string, []string) {
 	var q []string
 	var vals []string
 	for _, c := range l.constraints {
-		q = append(q, fmt.Sprintf("%s.%s_hash = ?", l.tableName, c.dir))
-		vals = append(vals, hashOf(c.vals[0]))
+		if len(c.vals) == 1 {
+			q = append(q, fmt.Sprintf("%s.%s_hash = ?", l.tableName, c.dir))
+			vals = append(vals, hashOf(c.vals[0]))
+		} else if len(c.vals) > 1 {
+			subq := fmt.Sprintf("%s.%s_hash IN ", l.tableName, c.dir)
+			valslots := strings.Join(strings.Split(strings.Repeat("?", len(c.vals)), ""), ", ")
+			subq += fmt.Sprintf("(%s)", valslots)
+			q = append(q, subq)
+			for _, v := range c.vals {
+				vals = append(vals, hashOf(v))
+			}
+		}
 	}
 	for _, i := range l.nodeIts {
 		t := i.it.tableID()