diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index b6d78799a..4ba2021bb 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -949,6 +949,10 @@ func decodeQueryResult(pb *internal.QueryResult) interface{} { return pb.Changed case queryResultTypeNil: return nil + case queryResultTypeRowIDs: + return pilosa.RowIDs(pb.RowIDs) + case queryResultTypeRowIdentifiers: + return decodeRowIdentifiers(pb.RowIdentifiers) case queryResultTypeGroupCounts: return decodeGroupCounts(pb.GroupCounts) } @@ -1001,6 +1005,13 @@ func decodeAttr(attr *internal.Attr) (key string, value interface{}) { } } +func decodeRowIdentifiers(a *internal.RowIdentifiers) *pilosa.RowIdentifiers { + return &pilosa.RowIdentifiers{ + Rows: a.Rows, + Keys: a.Keys, + } +} + func decodeGroupCounts(a []*internal.GroupCount) []pilosa.GroupCount { other := make([]pilosa.GroupCount, len(a)) for i := range a { diff --git a/executor.go b/executor.go index 3d8618b9d..3eb887b4e 100644 --- a/executor.go +++ b/executor.go @@ -18,7 +18,6 @@ import ( "context" "fmt" "sort" - "strconv" "strings" "time" @@ -768,9 +767,40 @@ func (r RowIDs) merge(other RowIDs, limit int) RowIDs { } func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]GroupCount, error) { + // validate call + if len(c.Children) == 0 { + return nil, errors.New("need at least one child call") + } + // get limit + gbLimit := int(^uint(0) >> 1) // largest signed int + if limit, hasLimit, err := c.UintArg("limit"); err != nil { + return nil, err + } else if hasLimit { + gbLimit = int(limit) + } + // perform Rows queries - TODO, call async? run per shard in + // executeGroupByShard? (note: can only do this for Rows queries which do + // not include "column" arg) + childRows := make([]RowIDs, len(c.Children)) + for i, child := range c.Children { + if child.Name != "Rows" { + return nil, errors.Errorf("'%s' is not a valid child query for GroupBy, must be 'Rows'", c.Name) + } + if limit, hasLimit, err := child.UintArg("limit"); err != nil { + return nil, err + } else if hasLimit && int(limit) > gbLimit { + child.Args["limit"] = uint64(gbLimit) + } + var err error + childRows[i], err = e.executeRows(ctx, index, child, shards, opt) + if err != nil { + return nil, errors.Wrap(err, "getting rows for ") + } + } + // Execute calls in bulk on each remote node and merge. mapFn := func(shard uint64) (interface{}, error) { - return e.executeGroupByShard(ctx, index, c, shard) + return e.executeGroupByShard(ctx, index, c, shard, childRows) } // Merge returned results at coordinating node. reduceFn := func(prev, v interface{}) interface{} { @@ -849,66 +879,39 @@ func mergeGroupCounts(gc, other []GroupCount) []GroupCount { return gc } -func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql.Call, shard uint64) ([]GroupCount, error) { +func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql.Call, shard uint64, childRows []RowIDs) ([]GroupCount, error) { // Fetch index. idx := e.Holder.Index(index) if idx == nil { return nil, ErrIndexNotFound } - // fieldDirective is a combination of field - // instructions, represented as a string with - // the form [fieldName:offset:limit] or - // [fieldName:limit]. - fieldDirectives, ok := c.Args["fields"] - if !ok { - return nil, errors.Wrap(ErrFieldsArgumentRequired, "executeGroupBy") - } - // Ensure that fieldDirectives is a list. - if _, ok := fieldDirectives.([]interface{}); !ok { - return nil, errors.Wrap(ErrExpectedFieldListArgument, "executeGroupBy") - } - // getFieldName extracts the fieldName portion of the - // fieldDirective. - getFieldName := func(s string) string { - parts := strings.Split(s, ":") - return parts[0] - } - // Ensure that all of the fields exist. - for _, fieldDirective := range fieldDirectives.([]interface{}) { - fieldName := getFieldName(fieldDirective.(string)) - f := e.Holder.Field(index, fieldName) - if f == nil { - return nil, errors.Wrap(ErrFieldNotFound, fmt.Sprintf("executeGroupBy: %s", fieldDirective.(string))) - } - } - results := make([]GroupCount, 0) var work [][]gbi - for _, fieldDirective := range fieldDirectives.([]interface{}) { - fieldName := getFieldName(fieldDirective.(string)) + for i, rowIDs := range childRows { + fieldName := c.Children[i].Args["field"].(string) // this has already been validated by this point // Fetch fragment. frag := e.Holder.fragment(index, fieldName, viewStandard, shard) if frag == nil { // this means this whole shard doesn't have all it needs to continue - return results, nil - } - // Get filter based on the field directive. - filters, err := getGroupByFilterFunction(fieldDirective.(string)) - if err != nil { - return nil, err + return []GroupCount{}, nil } set := make([]gbi, 0) - for _, rowID := range frag.rows(0, filters...) { - set = append(set, gbi{ - row: frag.row(rowID), - fieldRow: FieldRow{ - Field: fieldName, - RowID: rowID, - }, - }) + for _, rowID := range rowIDs { + rs := frag.rows(rowID, filterWithLimit(1)) + if len(rs) > 0 && rs[0] == rowID { + set = append(set, gbi{ + row: frag.row(rowID), + fieldRow: FieldRow{ + Field: fieldName, + RowID: rowID, + }, + }) + } } work = append(work, set) } + + results := make([]GroupCount, 0) for _, group := range product(work) { group.gCnt.Count = group.row.Count() if group.gCnt.Count > 0 { @@ -1017,6 +1020,10 @@ func (e *executor) executeRowsShard(ctx context.Context, index string, c *pql.Ca if columnID, ok, err := c.UintArg("column"); err != nil { return nil, err } else if ok { + colShard := columnID >> shardWidthExponent + if colShard != shard { + return RowIDs{}, nil + } filters = append(filters, filterColumn(columnID)) } if limit, hasLimit, err := c.UintArg("limit"); err != nil { @@ -1028,50 +1035,6 @@ func (e *executor) executeRowsShard(ctx context.Context, index string, c *pql.Ca return frag.rows(start, filters...), nil } -// getGroupByFilterFunction returns a rowFilter based on the -// field directive provided. -func getGroupByFilterFunction(fieldDirective string) (ret []rowFilter, err error) { - parts := strings.Split(fieldDirective, ":") - hasLimit := false - hasOffset := false - limit := uint64(0) - offset := uint64(0) - // fieldDirective can have one of the following forms: - // [fieldName] - // [fieldName:limit] - // [fieldName:offset:limit] - // - // Note that a field directive with the form - // [fieldName:offset:limit:extra] will be treated as - // [fieldName:offset:limit] (i.e. `extra` is ignored). - if len(parts) == 1 { - return ret, nil - } else if len(parts) == 2 { - hasLimit = true - if limit, err = strconv.ParseUint(parts[1], 10, 64); err != nil { - return ret, errors.Wrap(err, "getting groupby field limit only value") - } - } else { - hasOffset = true - if offset, err = strconv.ParseUint(parts[1], 10, 64); err != nil { - return ret, errors.Wrap(err, "getting groupby field offset value") - } - if parts[2] != "" { - hasLimit = true - if limit, err = strconv.ParseUint(parts[2], 10, 64); err != nil { - return ret, errors.Wrap(err, "getting groupby field limit value") - } - } - } - if hasOffset { - ret = append(ret, filterWithOffset(offset)) - } - if hasLimit { - ret = append(ret, filterWithLimit(limit)) - } - return ret, nil -} - func (e *executor) executeBitmapShard(_ context.Context, index string, c *pql.Call, shard uint64) (*Row, error) { // Fetch index. idx := e.Holder.Index(index) @@ -2319,12 +2282,6 @@ func isString(v interface{}) bool { return ok } -func filterWithOffset(offset uint64) rowFilter { - return func(rowID, key uint64, c *roaring.Container) (include, done bool) { - return rowID >= offset, false - } -} - // filterWithLimit returns a filter which will only allow a limited number of // rows to be returned. It should be applied last so that it is only called (and // therefore only updates its internal state) if the row is being included by diff --git a/executor_test.go b/executor_test.go index faa0acbf9..78e6d8790 100644 --- a/executor_test.go +++ b/executor_test.go @@ -1229,7 +1229,7 @@ Set(4500001, fn=4) t.Run("remote groupBy", func(t *testing.T) { if res, err := c[1].API.Query(context.Background(), &pilosa.QueryRequest{ Index: "i", - Query: `GroupBy(fields=[f])`, + Query: `GroupBy(Rows(field=f))`, }); err != nil { t.Fatalf("GroupBy querying: %v", err) } else { @@ -1819,25 +1819,18 @@ func TestExecutor_Execute_GroupBy(t *testing.T) { t.Run("No Field List Arguments", func(t *testing.T) { if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy()`}); err != nil { - if errors.Cause(err) != pilosa.ErrFieldsArgumentRequired { - t.Fatalf("unexpected error\n\"%s\" not returned instead \n\"%s\"", pilosa.ErrFieldsArgumentRequired, err) + if !strings.Contains(err.Error(), "need at least one child call") { + t.Fatalf("unexpected error: \"%v\"", err) } } }) t.Run("Unknown Field ", func(t *testing.T) { - if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=[missing])`}); err != nil { + if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(Rows(field=missing))`}); err != nil { if errors.Cause(err) != pilosa.ErrFieldNotFound { t.Fatalf("unexpected error\n\"%s\" not returned instead \n\"%s\"", pilosa.ErrFieldNotFound, err) } } }) - t.Run("Bad Field Format", func(t *testing.T) { - if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=missing)`}); err != nil { - if errors.Cause(err) != pilosa.ErrExpectedFieldListArgument { - t.Fatalf("unexpected error\n\"%s\" not returned instead \n\"%s\"", pilosa.ErrExpectedFieldListArgument, err) - } - } - }) t.Run("Basic", func(t *testing.T) { expected := []pilosa.GroupCount{ {Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1}, @@ -1846,7 +1839,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) { {Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 3}, } - if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=[general,sub])`}); err != nil { + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(Rows(field=general), Rows(field=sub))`}); err != nil { t.Fatal(err) } else { results := res.Results[0].([]pilosa.GroupCount) @@ -1860,7 +1853,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) { {Group: []pilosa.FieldRow{{Field: "general", RowID: 12}}, Count: 2}, } - if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=[general:11:])`}); err != nil { + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(Rows(field=general, previous=10))`}); err != nil { t.Fatal(err) } else { results := res.Results[0].([]pilosa.GroupCount) @@ -1873,7 +1866,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) { {Group: []pilosa.FieldRow{{Field: "general", RowID: 11}}, Count: 2}, } - if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=[general:11:1])`}); err != nil { + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(Rows(field=general, previous=10, limit=1))`}); err != nil { t.Fatal(err) } else { results := res.Results[0].([]pilosa.GroupCount)