From 54679c12c4364de125b51f0daaab21dd9a5dee31 Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Tue, 14 Jan 2020 22:10:12 -0600
Subject: [PATCH] post merge, needs review of TODOs

---
 cluster.go       |  2 +-
 executor.go      | 24 ++++++++++++++++++++++--
 executor_test.go |  1 +
 field.go         |  2 +-
 4 files changed, 25 insertions(+), 4 deletions(-)

diff --git a/cluster.go b/cluster.go
index 5763d740c..140babc11 100644
--- a/cluster.go
+++ b/cluster.go
@@ -869,7 +869,7 @@ func shardPartition(index string, shard uint64, partitionN int) int {
 	return int(h.Sum64() % uint64(partitionN))
 }
 
-// keyPartition returns the partition that a shard belongs to.
+// keyPartition returns the partition that a key belongs to.
 func (c *cluster) keyPartition(index, key string) int {
 	return keyPartition(index, key, c.partitionN)
 }
diff --git a/executor.go b/executor.go
index a36ea9d0e..a832dab5e 100644
--- a/executor.go
+++ b/executor.go
@@ -3689,7 +3689,27 @@ func (e *executor) translateCall(indexName string, idx *Index, isDefaultIndex bo
 					c.Args[rowKey] = rowID
 				}
 			} else if field.Keys() {
-				if c.Args[rowKey] != nil && !isString(c.Args[rowKey]) {
+				if c.Args[rowKey] != nil && isCondition(c.Args[rowKey]) {
+					// In the case where a field has a foreign index with keys,
+					// allow `== "key"` or `!= "key"` to be used against the BSI
+					// field.
+					cond := c.Args[rowKey].(*pql.Condition)
+					if isString(cond.Value) {
+						switch cond.Op {
+						case pql.EQ, pql.NEQ:
+							id, err := field.TranslateStore().TranslateKey(cond.Value.(string))
+							if err != nil {
+								return errors.Wrap(err, "translating key")
+							}
+							c.Args[rowKey] = &pql.Condition{
+								Op:    cond.Op,
+								Value: id,
+							}
+						default:
+							return errors.Errorf("conditional is not supported with string predicates: %s", cond.Op)
+						}
+					}
+				} else if c.Args[rowKey] != nil && !isString(c.Args[rowKey]) {
 					// allow passing row id directly (this can come in handy, but make sure it is a valid row id)
 					if !isValidID(c.Args[rowKey]) {
 						return errors.Errorf("row value must be a string or non-negative integer, but got: %v of %[1]T", c.Args[rowKey])
@@ -3854,7 +3874,7 @@ func (e *executor) translateResult(index string, idx *Index, call *pql.Call, res
 		// In the case where a field/foreignIndex doesn't exist,
 		// fall back to using the index translateStore.
 		if store == nil && idx.Keys() {
-			store = idx.translateStore
+			store = nil // TODO: this may need to be idx.TranslateStore(?)
 		}
 
 		if store != nil {
diff --git a/executor_test.go b/executor_test.go
index e3002a70c..b7d3e3d14 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -3912,6 +3912,7 @@ func TestExecutor_ForeignIndex(t *testing.T) {
 		t.Fatalf("unexpected columns: %v", neq.Columns())
 	}
 
+	// TODO: this test is failing because field `color` is being associated to index `parent`
 	join := c.Query(t, "parent", `Intersect(Row(general=3), Distinct(Row(color="blue"), index="child", field="parent_id"))`).Results[0].(*pilosa.Row)
 	if !reflect.DeepEqual(join.Keys, []string{"one"}) {
 		t.Fatalf("unexpected keys: %v", join.Keys)
diff --git a/field.go b/field.go
index 0c41534d0..f47f79342 100644
--- a/field.go
+++ b/field.go
@@ -565,7 +565,7 @@ func (f *Field) applyForeignIndex() error {
 		return errors.Wrapf(ErrForeignIndexNotFound, "%s", f.options.ForeignIndex)
 	} else if foreignIndex.Keys() {
 		f.usesKeys = true
-		f.translateStore = foreignIndex.translateStore
+		f.translateStore = foreignIndex.TranslateStore(0) // TODO: this is wrong
 		return nil
 	}
 	return f.applyTranslateStore()