simplify groupby sorting and fix bugs

Back out support for sorting on fields (only count and aggregate
supported for now).

Fix bug where default return of "true" caused sort to be unstable. (If
they are equal, Less should return false)

Fix bug where limit was being applied before sorting.

Fix bug where offset was not actually allowed to be an argument to
GroupBy (weird! guess we weren't testing that very well)

Apply "having" after calculating Count(Distinct) aggregate so that
having can apply to that.

Switch to stable sort to make testing easier.
This commit is contained in:
Matt Jaffee 2020-12-31 14:35:47 -06:00
parent 5fdae74812
commit ea539d8241
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
3 changed files with 121 additions and 61 deletions

View file

@ -2661,7 +2661,7 @@ func (g *groupCountSorter) Less(i, j int) bool {
}
}
return true
return false
}
// compareFieldRows returns -1 if a < b, +1 if a > b, and 0 if they
@ -2706,8 +2706,7 @@ func compareFieldRows(a, b FieldRow) int {
// getSorter hackily parses the sortSpec and figures out how to sort
// the GroupBy results. TODO Probably has about a billion edge case
// bugs. Also needs to take in the Call so it can figure out if fields
// are valid and what order they're in.
// bugs.
func getSorter(sortSpec string) (*groupCountSorter, error) {
gcs := &groupCountSorter{
fields: []int{},
@ -2717,19 +2716,16 @@ func getSorter(sortSpec string) (*groupCountSorter, error) {
for _, sortField := range sortOn {
sortField = strings.TrimSpace(sortField)
fieldDir := strings.Split(sortField, " ")
defaultOrder := asc
if fieldDir[0] == "count" {
gcs.fields = append(gcs.fields, -1)
defaultOrder = desc
} else if fieldDir[0] == "aggregate" {
} else if fieldDir[0] == "aggregate" || fieldDir[0] == "sum" {
gcs.fields = append(gcs.fields, -2)
defaultOrder = desc
} else {
gcs.fields = append(gcs.fields, 0) // TODO actually figure out which field. probably need the call
return nil, errors.Errorf("sorting is only supported on count, aggregate, or sum, not '%s'", fieldDir[0])
}
if len(fieldDir) == 0 {
gcs.order = append(gcs.order, defaultOrder)
if len(fieldDir) == 1 {
gcs.order = append(gcs.order, desc)
}
if fieldDir[1] == "asc" {
gcs.order = append(gcs.order, asc)
@ -2768,6 +2764,14 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
if err != nil {
return nil, errors.Wrap(err, "parsing sort spec")
}
// don't want to prematurely limit the results if we're sorting
limit = int(^uint(0) >> 1)
}
if _, hasHaving, err := c.CallArg("having"); err != nil {
return nil, errors.Wrap(err, "getting 'having' argument")
} else if hasHaving {
// don't want to prematurely limit the results if we're filtering some out
limit = int(^uint(0) >> 1)
}
idx := e.Holder.Index(index)
@ -2842,43 +2846,21 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
}
results, _ := other.([]GroupCount)
// Apply having.
if having, hasHaving, err := c.CallArg("having"); err != nil {
return nil, err
} else if hasHaving {
// parse the condition as PQL
if having.Name != "Condition" {
return nil, errors.New("the only supported having call is Condition()")
}
if len(having.Args) != 1 {
return nil, errors.New("Condition() must contain a single condition")
}
for subj, cond := range having.Args {
switch subj {
case "count", "sum":
results = applyConditionToGroupCounts(results, subj, cond.(*pql.Condition))
default:
return nil, errors.New("Condition() only supports count or sum")
}
// If there's no sorting, we want to apply limits before
// calculating the Distinct aggregate which is expensive on a
// per-result basis.
if sorter == nil {
results, err = applyLimitAndOffsetToGroupByResult(c, results)
if err != nil {
return nil, errors.Wrap(err, "applying limit/offset")
}
}
// Apply offset.
if offset, hasOffset, err := c.UintArg("offset"); err != nil {
return nil, err
} else if hasOffset {
if int(offset) < len(results) {
results = results[offset:]
}
}
// Apply limit.
if limit, hasLimit, err := c.UintArg("limit"); err != nil {
return nil, err
} else if hasLimit {
if int(limit) < len(results) {
results = results[:limit]
}
}
// TODO as an optimization, we could apply some "having"
// conditions here long as they aren't on the Count(Distinct)
// aggregate
// Calculate Count(Distinct) aggregate if requested.
aggregate, _, err := c.CallArg("aggregate")
if err == nil && aggregate != nil && aggregate.Name == "Count" && len(aggregate.Children) > 0 && aggregate.Children[0].Name == "Distinct" && !opt.Remote {
for n, gc := range results {
@ -2898,10 +2880,10 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
countDistinctIntersect := &pql.Call{
Name: "Count",
Children: []*pql.Call{
&pql.Call{
{
Name: "Distinct",
Children: []*pql.Call{
&pql.Call{
{
Name: "Intersect",
Children: intersectRows,
},
@ -2921,11 +2903,56 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
}
}
if sorter != nil {
sorter.data = results
sort.Sort(sorter)
// Apply having.
if having, hasHaving, err := c.CallArg("having"); err != nil {
return nil, err
} else if hasHaving && !opt.Remote {
// parse the condition as PQL
if having.Name != "Condition" {
return nil, errors.New("the only supported having call is Condition()")
}
if len(having.Args) != 1 {
return nil, errors.New("Condition() must contain a single condition")
}
for subj, cond := range having.Args {
switch subj {
case "count", "sum":
results = applyConditionToGroupCounts(results, subj, cond.(*pql.Condition))
default:
return nil, errors.New("Condition() only supports count or sum")
}
}
}
if sorter != nil && !opt.Remote {
sorter.data = results
sort.Stable(sorter)
results, err = applyLimitAndOffsetToGroupByResult(c, results)
if err != nil {
return nil, errors.Wrap(err, "applying limit/offset")
}
}
return results, nil
}
func applyLimitAndOffsetToGroupByResult(c *pql.Call, results []GroupCount) ([]GroupCount, error) {
// Apply offset.
if offset, hasOffset, err := c.UintArg("offset"); err != nil {
return nil, err
} else if hasOffset {
if int(offset) < len(results) {
results = results[offset:]
}
}
// Apply limit.
if limit, hasLimit, err := c.UintArg("limit"); err != nil {
return nil, err
} else if hasLimit {
if int(limit) < len(results) {
results = results[:limit]
}
}
return results, nil
}

View file

@ -6844,6 +6844,18 @@ func variousQueries(t *testing.T, clusterSize int) {
{Val: 0, Key: "userE"},
})
// Create and populate "affinity" int field with negative, positive, zero and null values.
c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "net_worth", pilosa.OptFieldTypeInt(-100000000, 100000000))
c.ImportIntKey(t, "users", "net_worth", []test.IntKey{
{Val: 1, Key: "userA"},
{Val: 10, Key: "userB"},
{Val: 100, Key: "userC"},
{Val: 1000, Key: "userD"},
{Val: 10000, Key: "userE"},
{Val: 100000, Key: "userF"},
})
c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "zip_code", pilosa.OptFieldTypeInt(0, 100000))
c.ImportIntKey(t, "users", "zip_code", []test.IntKey{
{Val: 78739, Key: "userA"},
@ -7008,6 +7020,10 @@ dog,1,0
icecream,6,0
`,
},
{
query: "GroupBy(Rows(field=likes), having=Condition(count>5))",
csvVerifier: "icecream,6,0\n",
},
{
query: "GroupBy(Rows(field=likes), filter=Row(affinity>-7))",
csvVerifier: `molecula,1,0
@ -7028,6 +7044,10 @@ dog,1,0
icecream,6,3
`,
},
{
query: "GroupBy(Rows(field=likes), aggregate=Count(Distinct(field=zip_code)), having=Condition(sum>2))",
csvVerifier: "icecream,6,3\n",
},
{
query: "GroupBy(Rows(field=likes), filter=Row(affinity>-11), aggregate=Count(Distinct(field=zip_code)))",
csvVerifier: `molecula,1,1
@ -7049,8 +7069,7 @@ icecream,5,3
`,
},
{
query: "GroupBy(Rows(field=likes), sort=\"count desc, likes asc\")",
// note, sort is in order of rowID rather than rowKey
query: "GroupBy(Rows(field=likes), sort=\"count desc\")",
csvVerifier: `icecream,6,0
molecula,1,0
pilosa,1,0
@ -7061,15 +7080,28 @@ dog,1,0
`,
},
{
query: "GroupBy(Rows(field=likes), sort=\"count desc, likes desc\")",
// note, sort is in order of rowID rather than rowKey
csvVerifier: `icecream,6,0
dog,1,0
toucan,1,0
zebra,1,0
pangolin,1,0
pilosa,1,0
molecula,1,0
query: "GroupBy(Rows(field=likes), aggregate=Sum(field=net_worth), sort=\"aggregate desc, count asc\")",
csvVerifier: `icecream,6,111111
dog,1,100000
toucan,1,10000
zebra,1,1000
pangolin,1,100
pilosa,1,10
molecula,1,1
`,
},
{
query: "GroupBy(Rows(field=likes), aggregate=Sum(field=net_worth), sort=\"aggregate desc, count asc\", limit=3)",
csvVerifier: `icecream,6,111111
dog,1,100000
toucan,1,10000
`,
},
{
query: "GroupBy(Rows(field=likes), aggregate=Sum(field=net_worth),sort=\"aggregate desc, count asc\",limit=3,offset=2)",
csvVerifier: `toucan,1,10000
zebra,1,1000
pangolin,1,100
`,
},
}
@ -7088,7 +7120,7 @@ molecula,1,0
// verify everything after header
got := csvString[strings.Index(csvString, "\n")+1:]
if got != tst.csvVerifier {
t.Errorf("expected '%s', got '%s'", tst.csvVerifier, got)
t.Errorf("expected:\n%s\ngot:\n%s", tst.csvVerifier, got)
}
// TODO: add HTTP and Postgres and ability to convert

View file

@ -441,6 +441,7 @@ var callInfoByFunc = map[string]callInfo{
prototypes: map[string]interface{}{
"filter": nil,
"limit": int64(0),
"offset": int64(0),
"previous": nil,
"aggregate": nil,
"having": nil,