Merge pull request #1307 from jaffee/1292-panic-groupby-int-distinct-aggregate

Fix potential panic when grouping on int field with Distinct aggregate

executor.go

@@ -441,9 +441,11 @@ func (e *executor) handlePreCalls(ctx context.Context, qcx *Qcx, index string, c
 	// shards if the query has to go to them
 	opt.EmbeddedData = append(opt.EmbeddedData, row)
 	// and stash a copy locally, so local calls can use it
-	c.Precomputed = make(map[uint64]interface{}, len(row.segments))
-	for _, segment := range row.segments {
-		c.Precomputed[segment.shard] = &Row{segments: []rowSegment{segment}}
+	if row != nil {
+		c.Precomputed = make(map[uint64]interface{}, len(row.segments))
+		for _, segment := range row.segments {
+			c.Precomputed[segment.shard] = &Row{segments: []rowSegment{segment}}
+		}
 	}
 	return nil
 }

executor_test.go

@@ -7144,6 +7144,88 @@ pangolin,1,100
 	}
 }
 
+// TestVariousSingleShardQueries tests queries on a dataset which
+// consists of an unkeyed index, and data only in the first
+// shard. Turns out that there are some interesting failure modes
+// which only crop up with one shard. An example is that the mapReduce
+// logic does its first reduce call with a nil interface{} value, and
+// an actual result value. If this is the only reduce that gets done
+// (because there's only one shard), the result might never go through
+// normal merge/reduction logic and might e.g. be nil instead of an
+// empty struct resulting in an NPE later on.
+func TestVariousSingleShardQueries(t *testing.T) {
+	for _, clusterSize := range []int{1, 4} {
+		t.Run(fmt.Sprintf("%d-node", clusterSize), func(t *testing.T) {
+			variousSingleShardQueries(t, clusterSize)
+		})
+	}
+}
+
+func variousSingleShardQueries(t *testing.T, clusterSize int) {
+	c := test.MustRunCluster(t, clusterSize)
+	defer c.Close()
+
+	// Create and populate "likenums" similar to "likes", but without keys on the field.
+	c.CreateField(t, "events", pilosa.IndexOptions{Keys: false, TrackExistence: true}, "lostcount", pilosa.OptFieldTypeInt(0, 1000000000))
+	c.ImportIntID(t, "events", "lostcount", []test.IntID{
+		{Val: 0, ID: 1},
+		{Val: 1, ID: 2},
+		{Val: 0, ID: 3},
+		{Val: 2, ID: 4},
+		{Val: 2, ID: 5},
+		{Val: 0, ID: 6},
+		{Val: 3, ID: 7},
+		{Val: 3, ID: 8},
+		{Val: 3, ID: 9},
+		{Val: 0, ID: 10},
+	})
+
+	c.CreateField(t, "events", pilosa.IndexOptions{Keys: false, TrackExistence: true}, "jittermax", pilosa.OptFieldTypeInt(0, 1000000000))
+	c.ImportIntID(t, "events", "jittermax", []test.IntID{
+		{Val: 17, ID: 1},
+		{Val: 3, ID: 2},
+		{Val: 42, ID: 3},
+		{Val: 9, ID: 4},
+		{Val: 17, ID: 5},
+		{Val: 3, ID: 6},
+		{Val: 42, ID: 7},
+		{Val: 9, ID: 8},
+		{Val: 17, ID: 9},
+		{Val: 3, ID: 10},
+	})
+
+	tests := []struct {
+		query       string
+		csvVerifier string
+	}{
+		{
+			query: "GroupBy(Rows(lostcount), aggregate=Count(Distinct(field=jittermax)))",
+			csvVerifier: `0,4,3
+1,1,1
+2,2,2
+3,3,3
+`,
+		},
+	}
+
+	for i, tst := range tests {
+		t.Run(fmt.Sprintf("%d-%s", i, tst.query), func(t *testing.T) {
+			tr := c.QueryGRPC(t, "events", tst.query)
+			csvString, err := tableResponseToCSVString(tr)
+			if err != nil {
+				t.Fatal(err)
+			}
+			// verify everything after header
+			got := csvString[strings.Index(csvString, "\n")+1:]
+			if got != tst.csvVerifier {
+				t.Errorf("expected:\n%s\ngot:\n%s", tst.csvVerifier, got)
+			}
+
+		})
+	}
+
+}
+
 // tableResponseToCSV converts a generic TableResponse to a CSV format
 // and writes it to the writer.
 func tableResponseToCSV(m *proto.TableResponse, w io.Writer) error {

test/cluster.go

@@ -63,7 +63,7 @@ func (c *Cluster) QueryHTTP(t testing.TB, index, query string) (string, error) {
 	if len(c.Nodes) == 0 {
 		t.Fatal("must have at least one node in cluster to query")
 	}
-	
+
 	return c.Nodes[0].Query(t, index, "", query)
 }
 
@@ -189,6 +189,30 @@ func (c *Cluster) ImportIntKey(t testing.TB, index, field string, pairs []IntKey
 	}
 }
 
+type IntID struct {
+	Val int64
+	ID  uint64
+}
+
+// ImportIntID imports data into an int field in an unkeyed index.
+func (c *Cluster) ImportIntID(t testing.TB, index, field string, pairs []IntID) {
+	t.Helper()
+	importRequest := &pilosa.ImportValueRequest{
+		Index:     index,
+		Field:     field,
+		Shard:     math.MaxUint64,
+		ColumnIDs: make([]uint64, len(pairs)),
+		Values:    make([]int64, len(pairs)),
+	}
+	for i, pair := range pairs {
+		importRequest.Values[i] = pair.Val
+		importRequest.ColumnIDs[i] = pair.ID
+	}
+	if err := c.Nodes[0].API.ImportValue(context.Background(), nil, importRequest); err != nil {
+		t.Fatalf("importing IntID data: %v", err)
+	}
+}
+
 // KeyID represents a key and an ID for importing data into an index
 // and field where one uses string keys and the other does not.
 type KeyID struct {