Merge pull request #163 from seebs/distinctshards

Address issues with Distinct failures in testing, or across shards, or in cases where the range of Distinct results is not the same as the range of shards available in any index.

executor.go

@@ -438,7 +438,7 @@ func (e *executor) execute(ctx context.Context, index string, q *pql.Query, shar
 		// still need to handle them. Since everything else was
 		// already precomputed by handlePreCallChildren, though,
 		// we don't need this logic in executeCall.
-		if newIndex := call.CallIndex(); newIndex != "" {
+		if newIndex := call.CallIndex(); newIndex != "" && newIndex != index {
 			v, err = e.executeCall(ctx, newIndex, call, nil, opt)
 		} else {
 			v, err = e.executeCall(ctx, index, call, shards, opt)
@@ -1006,37 +1006,14 @@ func (e *executor) executeMaxRow(ctx context.Context, index string, c *pql.Call,
 
 // executePrecomputedCall pretends to execute a call that we have a precomputed value for.
 func (e *executor) executePrecomputedCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (*Row, error) {
-	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executePrecomputedCall")
+	span, _ := tracing.StartSpanFromContext(ctx, "Executor.executePrecomputedCall")
 	defer span.Finish()
+	result := NewRow()
 
-	// Execute calls in bulk on each remote node and merge.
-	mapFn := func(shard uint64) (interface{}, error) {
-		if c.Precomputed != nil {
-			return c.Precomputed[shard], nil
-		}
-		// This might not be an error -- if there were no values, we will not have created
-		// the corresponding row.
-		return NewRow(), nil
+	for _, row := range c.Precomputed {
+		result.Merge(row.(*Row))
 	}
-
-	// Merge returned results at coordinating node.
-	reduceFn := func(prev, v interface{}) interface{} {
-		other, _ := prev.(*Row)
-		if other == nil {
-			other = NewRow()
-		}
-		other.Merge(v.(*Row))
-		return other
-	}
-
-	other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
-	if err != nil {
-		return nil, errors.Wrap(err, "map reduce")
-	}
-
-	row, _ := other.(*Row)
-
-	return row, nil
+	return result, nil
 }
 
 // executeBitmapCall executes a call that returns a bitmap.

executor_test.go

@@ -5082,3 +5082,32 @@ func TestExecutor_Execute_CountDistinct(t *testing.T) {
 		}
 	})
 }
+
+// Ensure that a top-level, bare distinct on multiple nodes
+// is handled correctly.
+func TestExecutor_BareDistinct(t *testing.T) {
+	t.Helper()
+	c := test.MustRunCluster(t, 2)
+	defer c.Close()
+
+	c.CreateField(t, "i", pilosa.IndexOptions{}, "ints",
+		pilosa.OptFieldTypeInt(0, math.MaxInt64),
+	)
+
+	// Populate integer data.
+	c.Query(t, "i", fmt.Sprintf(`
+			Set(0, ints=1)
+			Set(%d, ints=2)
+		`, ShardWidth))
+
+	for _, pql := range []string{
+		`Distinct(field="ints")`,
+		`Distinct(index="i", field="ints")`,
+	} {
+		exp := []uint64{1, 2}
+		res := c.Query(t, "i", pql).Results[0].(pilosa.SignedRow)
+		if got := res.Pos.Columns(); !reflect.DeepEqual(exp, got) {
+			t.Fatalf("expected: %v, but got: %v", exp, got)
+		}
+	}
+}