scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

get GroupBy working with "Rows" child calls, remove fieldDirectives

Browse source 
had to implement decoders for RowIDs and RowIdentifiers - a sign that we need
better testing of remote Rows calls
This commit is contained in:
Matt Jaffee 2018-10-08 19:04:29 -05:00
parent c172ca0680
commit cbd7e945b2
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
3 changed files with 71 additions and 110 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

11
encoding/proto/proto.go
View file

@ -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 {

149
executor.go
View file

@ -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

21
executor_test.go
View file

@ -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)