diff --git a/executor.go b/executor.go index 32a1b2f77..7482dff74 100644 --- a/executor.go +++ b/executor.go @@ -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 } diff --git a/executor_test.go b/executor_test.go index f6d711971..920977cb8 100644 --- a/executor_test.go +++ b/executor_test.go @@ -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 diff --git a/pql/ast.go b/pql/ast.go index 056a8fc85..40e2acaba 100644 --- a/pql/ast.go +++ b/pql/ast.go @@ -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,