From 70bfe86f75e05c9723975ad5ebc7f581fc3c894c Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Tue, 14 Apr 2020 14:28:17 -0500
Subject: [PATCH] in Store/SetRow, create field if it doesn't already exist

If you try to Store to a nonexistent field, we create an automatic
Set field with no cache for it, assuming it won't be used for TopN
queries. If you want TopN to work, you need to actually create it
yourself.
---
 executor.go      | 22 +++++++++++++++++-----
 executor_test.go | 14 ++++++++++++++
 2 files changed, 31 insertions(+), 5 deletions(-)

diff --git a/executor.go b/executor.go
index 702d0321c..a911eb0a7 100644
--- a/executor.go
+++ b/executor.go
@@ -2825,15 +2825,27 @@ func (e *executor) executeClearRowShard(ctx context.Context, index string, c *pq
 }
 
 // executeSetRow executes a Store() call.
-func (e *executor) executeSetRow(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (bool, error) {
+func (e *executor) executeSetRow(ctx context.Context, indexName string, c *pql.Call, shards []uint64, opt *execOptions) (bool, error) {
 	// Ensure the field type supports Store().
 	fieldName, err := c.FieldArg()
 	if err != nil {
 		return false, errors.New("field required for Store()")
 	}
-	field := e.Holder.Field(index, fieldName)
+	field := e.Holder.Field(indexName, fieldName)
 	if field == nil {
-		return false, ErrFieldNotFound
+		// Find index.
+		index := e.Holder.Index(indexName)
+		if index == nil {
+			return false, newNotFoundError(ErrIndexNotFound)
+		}
+
+		// Create field.
+		field, err = index.CreateField(fieldName, OptFieldTypeSet(CacheTypeNone, 0))
+		if err != nil {
+			// We wrap these because we want to indicate that it wasn't found,
+			// but also the problem we encountered trying to create it.
+			return false, newNotFoundError(errors.Wrap(err, "creating field"))
+		}
 	}
 	if field.Type() != FieldTypeSet {
 		return false, fmt.Errorf("can't Store() on a %s field", field.Type())
@@ -2841,7 +2853,7 @@ func (e *executor) executeSetRow(ctx context.Context, index string, c *pql.Call,
 
 	// Execute calls in bulk on each remote node and merge.
 	mapFn := func(shard uint64) (interface{}, error) {
-		return e.executeSetRowShard(ctx, index, c, shard)
+		return e.executeSetRowShard(ctx, indexName, c, shard)
 	}
 
 	// Merge returned results at coordinating node.
@@ -2853,7 +2865,7 @@ func (e *executor) executeSetRow(ctx context.Context, index string, c *pql.Call,
 		return val || prev.(bool)
 	}
 
-	result, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
+	result, err := e.mapReduce(ctx, indexName, shards, c, opt, mapFn, reduceFn)
 	return result.(bool), err
 }
 
diff --git a/executor_test.go b/executor_test.go
index 8ca45b777..ad0d1c56e 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -3713,6 +3713,20 @@ func TestExecutor_Execute_SetRow(t *testing.T) {
 		} else if bits := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(bits, []uint64{3, ShardWidth - 1, ShardWidth + 1}) {
 			t.Fatalf("unexpected columns: %+v", bits)
 		}
+
+		// Store row 10 into a table which doesn't exist.
+		if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Store(Row(f=10), nonexistent=20)`}); err != nil {
+			t.Fatal(err)
+		} else if res := res.Results[0].(bool); !res {
+			t.Fatalf("unexpected set row result: %+v", res)
+		}
+
+		// Ensure the row was populated.
+		if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Row(nonexistent=20)`}); err != nil {
+			t.Fatal(err)
+		} else if bits := res.Results[0].(*pilosa.Row).Columns(); !reflect.DeepEqual(bits, []uint64{3, ShardWidth - 1, ShardWidth + 1}) {
+			t.Fatalf("unexpected columns: %+v", bits)
+		}
 	})
 	t.Run("Set_NoSource", func(t *testing.T) {
 		c := test.MustRunCluster(t, 1)