mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
Renamed Sum call to SumReduce call and it returns SumCount; removed Average call
This commit is contained in:
parent
f6ce36c02c
commit
ebee6e4deb
2 changed files with 16 additions and 112 deletions
55
executor.go
55
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
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
}
|
||||
})
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue