diff --git a/executor.go b/executor.go index dfd63be90..8aaac05dc 100644 --- a/executor.go +++ b/executor.go @@ -161,9 +161,9 @@ func (e *Executor) executeCall(ctx context.Context, index string, c *pql.Call, s indexTag := fmt.Sprintf("index:%s", index) // Special handling for mutation and top-n calls. switch c.Name { - case "Average": + case "SumReduce": e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag}) - return e.executeAverage(ctx, index, c, slices, opt) + return e.executeSumReduce(ctx, index, c, slices, opt) case "ClearBit": return e.executeClearBit(ctx, index, c, opt) case "Count": @@ -177,9 +177,6 @@ func (e *Executor) executeCall(ctx context.Context, index string, c *pql.Call, s return nil, e.executeSetRowAttrs(ctx, index, c, opt) case "SetColumnAttrs": return nil, e.executeSetColumnAttrs(ctx, index, c, opt) - case "Sum": - e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag}) - return e.executeSum(ctx, index, c, slices, opt) case "TopN": e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag}) return e.executeTopN(ctx, index, c, slices, opt) @@ -208,16 +205,16 @@ func (e *Executor) validateCallArgs(c *pql.Call) error { return nil } -// executeAverage executes an average() call. -func (e *Executor) executeAverage(ctx context.Context, index string, c *pql.Call, slices []uint64, opt *ExecOptions) (int64, error) { +// executeSumReduce executes a SumCount() call. +func (e *Executor) executeSumReduce(ctx context.Context, index string, c *pql.Call, slices []uint64, opt *ExecOptions) (SumCount, error) { if frame, _ := c.Args["frame"]; frame == "" { - return 0, errors.New("Average(): frame required") + return SumCount{}, errors.New("SumReduce(): frame required") } else if field, _ := c.Args["field"]; field == "" { - return 0, errors.New("Average(): field required") + return SumCount{}, errors.New("SumReduce(): field required") } if len(c.Children) > 1 { - return 0, errors.New("Average() only accepts a single bitmap input") + return SumCount{}, errors.New("SumReduce() only accepts a single bitmap input") } // Execute calls in bulk on each remote node and merge. @@ -233,14 +230,14 @@ func (e *Executor) executeAverage(ctx context.Context, index string, c *pql.Call result, err := e.mapReduce(ctx, index, slices, c, opt, mapFn, reduceFn) if err != nil { - return 0, err + return SumCount{}, err } other, _ := result.(SumCount) if other.Count == 0 { - return 0, nil + return SumCount{}, nil } - return other.Sum / other.Count, nil + return other, nil } // executeBitmapCall executes a call that returns a bitmap. @@ -330,38 +327,6 @@ func (e *Executor) executeBitmapCallSlice(ctx context.Context, index string, c * } } -// executeSum executes a sum() call. -func (e *Executor) executeSum(ctx context.Context, index string, c *pql.Call, slices []uint64, opt *ExecOptions) (int64, error) { - if frame, _ := c.Args["frame"]; frame == "" { - return 0, errors.New("Sum(): frame required") - } else if field, _ := c.Args["field"]; field == "" { - return 0, errors.New("Sum(): field required") - } - - if len(c.Children) > 1 { - return 0, errors.New("Sum() only accepts a single bitmap input") - } - - // Execute calls in bulk on each remote node and merge. - mapFn := func(slice uint64) (interface{}, error) { - return e.executeSumCountSlice(ctx, index, c, slice) - } - - // Merge returned results at coordinating node. - reduceFn := func(prev, v interface{}) interface{} { - other, _ := prev.(SumCount) - return other.Add(v.(SumCount)) - } - - result, err := e.mapReduce(ctx, index, slices, c, opt, mapFn, reduceFn) - if err != nil { - return 0, err - } - other, _ := result.(SumCount) - - return other.Sum, nil -} - // executeSumCountSlice executes calculates the sum & count for fields on a slice. func (e *Executor) executeSumCountSlice(ctx context.Context, index string, c *pql.Call, slice uint64) (SumCount, error) { var filter *Bitmap diff --git a/executor_test.go b/executor_test.go index 2b99d60c7..896c7f7b8 100644 --- a/executor_test.go +++ b/executor_test.go @@ -600,8 +600,8 @@ func TestExecutor_Execute_TopN_Attr_Src(t *testing.T) { } } -// Ensure a Sum() query can be executed. -func TestExecutor_Execute_Sum(t *testing.T) { +// Ensure a SumReduce() query can be executed. +func TestExecutor_Execute_SumReduce(t *testing.T) { hldr := test.MustOpenHolder() defer hldr.Close() e := test.NewExecutor(hldr.Holder, test.NewCluster(1)) @@ -645,78 +645,17 @@ func TestExecutor_Execute_Sum(t *testing.T) { } t.Run("NoFilter", func(t *testing.T) { - if result, err := e.Execute(context.Background(), "i", test.MustParse(`Sum(frame=f, field=foo)`), nil, nil); err != nil { + if result, err := e.Execute(context.Background(), "i", test.MustParse(`SumReduce(frame=f, field=foo)`), nil, nil); err != nil { t.Fatal(err) - } else if result[0] != int64(200) { + } else if !reflect.DeepEqual(result[0], pilosa.SumCount{Sum: 200, Count: 5}) { t.Fatalf("unexpected result: %s", spew.Sdump(result)) } }) t.Run("WithFilter", func(t *testing.T) { - if result, err := e.Execute(context.Background(), "i", test.MustParse(`Sum(Bitmap(frame=f, rowID=0), frame=f, field=foo)`), nil, nil); err != nil { + if result, err := e.Execute(context.Background(), "i", test.MustParse(`SumReduce(Bitmap(frame=f, rowID=0), frame=f, field=foo)`), nil, nil); err != nil { t.Fatal(err) - } else if result[0] != int64(80) { - t.Fatalf("unexpected result: %s", spew.Sdump(result)) - } - }) -} - -// Ensure a Average() query can be executed. -func TestExecutor_Execute_Average(t *testing.T) { - hldr := test.MustOpenHolder() - defer hldr.Close() - e := test.NewExecutor(hldr.Holder, test.NewCluster(1)) - - idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{}) - if err != nil { - t.Fatal(err) - } - - if _, err := idx.CreateFrame("f", pilosa.FrameOptions{ - RangeEnabled: true, - Fields: []*pilosa.Field{ - {Name: "foo", Type: pilosa.FieldTypeInt, Min: 10, Max: 100}, - {Name: "bar", Type: pilosa.FieldTypeInt, Min: 0, Max: 100000}, - }, - }); err != nil { - t.Fatal(err) - } - - if _, err := idx.CreateFrame("other", pilosa.FrameOptions{ - RangeEnabled: true, - Fields: []*pilosa.Field{ - {Name: "foo", Type: pilosa.FieldTypeInt, Min: 0, Max: 1000}, - }, - }); err != nil { - t.Fatal(err) - } - - if _, err := e.Execute(context.Background(), "i", test.MustParse(` - SetBit(frame=f, rowID=0, columnID=0) - SetBit(frame=f, rowID=0, columnID=`+strconv.Itoa(SliceWidth+2)+`) - - SetFieldValue(frame=f, foo=20, bar=2000, columnID=0) - SetFieldValue(frame=f, foo=30, columnID=`+strconv.Itoa(SliceWidth)+`) - SetFieldValue(frame=f, foo=40, columnID=`+strconv.Itoa(SliceWidth+2)+`) - SetFieldValue(frame=f, foo=50, columnID=`+strconv.Itoa((5*SliceWidth)+100)+`) - SetFieldValue(frame=f, foo=60, columnID=`+strconv.Itoa(SliceWidth+1)+`) - SetFieldValue(frame=other, foo=1000, columnID=0) - `), nil, nil); err != nil { - t.Fatal(err) - } - - t.Run("NoFilter", func(t *testing.T) { - if result, err := e.Execute(context.Background(), "i", test.MustParse(`Average(frame=f, field=foo)`), nil, nil); err != nil { - t.Fatal(err) - } else if result[0] != int64(40) { - t.Fatalf("unexpected result: %s", spew.Sdump(result)) - } - }) - - t.Run("WithFilter", func(t *testing.T) { - if result, err := e.Execute(context.Background(), "i", test.MustParse(`Average(Bitmap(frame=f, rowID=0), frame=f, field=foo)`), nil, nil); err != nil { - t.Fatal(err) - } else if result[0] != int64(30) { + } else if !reflect.DeepEqual(result[0], pilosa.SumCount{Sum: 80, Count: 2}) { t.Fatalf("unexpected result: %s", spew.Sdump(result)) } })