apply base in GroupBy

This commit is contained in:
Jaden Weiss 2020-06-10 16:16:53 -04:00
parent ec9474114a
commit 535257af75
No known key found for this signature in database
GPG key ID: 177F065773634B67
2 changed files with 58 additions and 2 deletions

View file

@ -1770,10 +1770,16 @@ func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call
return nil, err
}
idx := e.Holder.Index(index)
if idx == nil {
return nil, ErrIndexNotFound
}
// perform necessary Rows queries (any that have limit or columns args) -
// TODO, call async? would only help if multiple Rows queries had a column
// or limit arg.
// TODO support TopN in here would be really cool - and pretty easy I think.
bases := make(map[int]int64)
childRows := make([]RowIDs, len(c.Children))
for i, child := range c.Children {
// Check "field" first for backwards compatibility, then set _field.
@ -1793,6 +1799,18 @@ func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call
if err != nil {
return nil, errors.Wrap(err, "getting column")
}
if _, ok := child.Args["_field"].(string); !ok {
return nil, errors.Errorf("%s call must have field with valid (string) field name. Got %v of type %[2]T", child.Name, child.Args["_field"])
}
f := idx.Field(child.Args["_field"].(string))
if f == nil {
return nil, ErrFieldNotFound
}
switch f.Type() {
case FieldTypeInt:
bases[i] = f.bsiGroup(f.name).Base
}
if hasLimit || hasCol { // we need to perform this query cluster-wide ahead of executeGroupByShard
childRows[i], err = e.executeRows(ctx, index, child, shards, opt)
if err != nil {
@ -1806,7 +1824,7 @@ func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call
// Execute calls in bulk on each remote node and merge.
mapFn := func(ctx context.Context, shard uint64) (interface{}, error) {
return e.executeGroupByShard(ctx, index, c, filter, shard, childRows)
return e.executeGroupByShard(ctx, index, c, filter, shard, childRows, bases)
}
// Merge returned results at coordinating node.
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
@ -2157,7 +2175,7 @@ func applyConditionToGroupCounts(gcs []GroupCount, subj string, cond *pql.Condit
return gcs[:i]
}
func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql.Call, filter *pql.Call, shard uint64, childRows []RowIDs) (_ []GroupCount, err error) {
func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql.Call, filter *pql.Call, shard uint64, childRows []RowIDs, bases map[int]int64) (_ []GroupCount, err error) {
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeGroupByShard")
defer span.Finish()
@ -2205,6 +2223,13 @@ func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql
}
}
// Apply bases.
for i, base := range bases {
for _, r := range results {
*r.Group[i].Value += base
}
}
return results, nil
}

View file

@ -2996,6 +2996,37 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
test.CheckGroupBy(t, expected, results)
}
})
t.Run("groupbBy on ints with offset regression", func(t *testing.T) {
_, err = c[0].API.CreateField(context.Background(), "i", "hint", pilosa.OptFieldTypeInt(1, 1000))
if err != nil {
t.Fatalf("creating field: %v", err)
}
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `
Set(0, hint=1)
Set(1, hint=2)
Set(2, hint=3)
`}); err != nil {
t.Fatalf("querying remote: %v", err)
}
if res, err := c[1].API.Query(context.Background(), &pilosa.QueryRequest{
Index: "i",
Query: `GroupBy(Rows(hint))`,
}); err != nil {
t.Fatalf("GroupBy querying: %v", err)
} else {
var a, b, c int64 = 1, 2, 3
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "hint", Value: &a}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "hint", Value: &b}}, Count: 1},
{Group: []pilosa.FieldRow{{Field: "hint", Value: &c}}, Count: 1},
}
results := res.Results[0].([]pilosa.GroupCount)
test.CheckGroupBy(t, expected, results)
}
})
}
// Ensure executor returns an error if too many writes are in a single request.