mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-08-28 10:54:59 +00:00
get a somewhat better groupBy working that passes new tests
one test still fails due to reordering during merging
This commit is contained in:
Matt Jaffee
parent
603b0e5369
commit
360623230f
3 changed files with 290 additions and 72 deletions
261
executor.go
261
executor.go
|
|
@ -834,13 +834,6 @@ func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call
|
|||
if len(c.Children) == 0 {
|
||||
return nil, errors.New("need at least one child call")
|
||||
}
|
||||
// get limit
|
||||
gbLimit := int(^uint(0) >> 1) // largest signed int
|
||||
if limit, hasLimit, err := c.UintArg("limit"); err != nil {
|
||||
return nil, err
|
||||
} else if hasLimit {
|
||||
gbLimit = int(limit)
|
||||
}
|
||||
// perform Rows queries - TODO, call async? run per shard in
|
||||
// executeGroupByShard? (note: can only do this for Rows queries which do
|
||||
// not include "column" arg)
|
||||
|
|
@ -849,15 +842,22 @@ func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call
|
|||
if child.Name != "Rows" {
|
||||
return nil, errors.Errorf("'%s' is not a valid child query for GroupBy, must be 'Rows'", c.Name)
|
||||
}
|
||||
if limit, hasLimit, err := child.UintArg("limit"); err != nil {
|
||||
return nil, err
|
||||
} else if !hasLimit || int(limit) > gbLimit {
|
||||
child.Args["limit"] = uint64(gbLimit)
|
||||
}
|
||||
var err error
|
||||
childRows[i], err = e.executeRows(ctx, index, child, shards, opt)
|
||||
_, hasLimit, err := child.UintArg("limit")
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "getting rows for ")
|
||||
return nil, errors.Wrap(err, "getting limit")
|
||||
}
|
||||
_, hasCol, err := child.UintArg("column")
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "getting column")
|
||||
}
|
||||
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 {
|
||||
return nil, errors.Wrap(err, "getting rows for ")
|
||||
}
|
||||
if len(childRows[i]) == 0 { // there are no results because this field has no values.
|
||||
return []GroupCount{}, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -943,44 +943,58 @@ func mergeGroupCounts(gc, other []GroupCount) []GroupCount {
|
|||
}
|
||||
|
||||
func (e *executor) executeGroupByShard(_ context.Context, index string, c *pql.Call, shard uint64, childRows []RowIDs) ([]GroupCount, error) {
|
||||
// Fetch index.
|
||||
idx := e.Holder.Index(index)
|
||||
if idx == nil {
|
||||
return nil, ErrIndexNotFound
|
||||
iter := newGroupByIterator(childRows, c.Children, index, shard, e.Holder)
|
||||
if iter == nil {
|
||||
return []GroupCount{}, nil
|
||||
}
|
||||
|
||||
var work [][]gbi
|
||||
for i, rowIDs := range childRows {
|
||||
fieldName := c.Children[i].Args["field"].(string) // this has already been validated by this point
|
||||
// Fetch fragment.
|
||||
frag := e.Holder.fragment(index, fieldName, viewStandard, shard)
|
||||
if frag == nil { // this means this whole shard doesn't have all it needs to continue
|
||||
return []GroupCount{}, nil
|
||||
}
|
||||
// var work [][]gbi
|
||||
// for i, rowIDs := range childRows {
|
||||
// fieldName := c.Children[i].Args["field"].(string) // this has already been validated by this point
|
||||
// // Fetch fragment.
|
||||
// frag := e.Holder.fragment(index, fieldName, viewStandard, shard)
|
||||
// if frag == nil { // this means this whole shard doesn't have all it needs to continue
|
||||
// return []GroupCount{}, nil
|
||||
// }
|
||||
|
||||
set := make([]gbi, 0)
|
||||
for _, rowID := range rowIDs {
|
||||
rs := frag.rows(rowID, filterWithLimit(1))
|
||||
if len(rs) > 0 && rs[0] == rowID {
|
||||
set = append(set, gbi{
|
||||
row: frag.row(rowID),
|
||||
fieldRow: FieldRow{
|
||||
Field: fieldName,
|
||||
RowID: rowID,
|
||||
},
|
||||
})
|
||||
}
|
||||
}
|
||||
work = append(work, set)
|
||||
// set := make([]gbi, 0)
|
||||
// for _, rowID := range rowIDs {
|
||||
// rs := frag.rows(rowID, filterWithLimit(1))
|
||||
// if len(rs) > 0 && rs[0] == rowID {
|
||||
// set = append(set, gbi{
|
||||
// row: frag.row(rowID),
|
||||
// fieldRow: FieldRow{
|
||||
// Field: fieldName,
|
||||
// RowID: rowID,
|
||||
// },
|
||||
// })
|
||||
// }
|
||||
// }
|
||||
// work = append(work, set)
|
||||
// }
|
||||
limit := int(^uint(0) >> 1)
|
||||
if lim, hasLimit, err := c.UintArg("limit"); err != nil {
|
||||
return nil, err
|
||||
} else if hasLimit {
|
||||
limit = int(lim)
|
||||
}
|
||||
|
||||
results := make([]GroupCount, 0)
|
||||
for _, group := range product(work) {
|
||||
group.gCnt.Count = group.row.Count()
|
||||
if group.gCnt.Count > 0 {
|
||||
results = append(results, group.gCnt)
|
||||
|
||||
num := 0
|
||||
for pp, done := iter.Next(); !done && num < limit; pp, done = iter.Next() {
|
||||
if pp.gCnt.Count > 0 {
|
||||
num++
|
||||
results = append(results, pp.gCnt)
|
||||
}
|
||||
}
|
||||
|
||||
// for _, group := range product(work) {
|
||||
// group.gCnt.Count = group.row.Count()
|
||||
// if group.gCnt.Count > 0 {
|
||||
// results = append(results, group.gCnt)
|
||||
// }
|
||||
// }
|
||||
return results, nil
|
||||
}
|
||||
|
||||
|
|
@ -990,32 +1004,32 @@ type ppi struct {
|
|||
gCnt GroupCount
|
||||
}
|
||||
|
||||
// product generates the cartesian product of the input
|
||||
// using tail recursion.
|
||||
func product(input [][]gbi) []ppi {
|
||||
if len(input) == 0 { // base return empty list
|
||||
return []ppi{
|
||||
{gCnt: GroupCount{Group: make([]FieldRow, 0)}},
|
||||
}
|
||||
}
|
||||
// // product generates the cartesian product of the input
|
||||
// // using tail recursion.
|
||||
// func product(input [][]gbi) []ppi {
|
||||
// if len(input) == 0 { // base return empty list
|
||||
// return []ppi{
|
||||
// {gCnt: GroupCount{Group: make([]FieldRow, 0)}},
|
||||
// }
|
||||
// }
|
||||
|
||||
res := make([]ppi, 0)
|
||||
head := input[0] // take first element of the list
|
||||
tail := product(input[1:]) // invoke product on remaining element
|
||||
for h := range head { // for each head
|
||||
for t := range tail { // iterate over the tail
|
||||
s := ppi{gCnt: GroupCount{Group: make([]FieldRow, 0)}}
|
||||
s.gCnt.Group = append([]FieldRow{head[h].fieldRow}, tail[t].gCnt.Group...) // had to insert at the front to match input order
|
||||
if tail[t].row != nil { // first time around nothing to intersect
|
||||
s.row = head[h].row.Intersect(tail[t].row)
|
||||
} else {
|
||||
s.row = head[h].row
|
||||
}
|
||||
res = append(res, s)
|
||||
}
|
||||
}
|
||||
return res
|
||||
}
|
||||
// res := make([]ppi, 0)
|
||||
// head := input[0] // take first element of the list
|
||||
// tail := product(input[1:]) // invoke product on remaining element
|
||||
// for h := range head { // for each head
|
||||
// for t := range tail { // iterate over the tail
|
||||
// s := ppi{gCnt: GroupCount{Group: make([]FieldRow, 0)}}
|
||||
// s.gCnt.Group = append([]FieldRow{head[h].fieldRow}, tail[t].gCnt.Group...) // had to insert at the front to match input order
|
||||
// if tail[t].row != nil { // first time around nothing to intersect
|
||||
// s.row = head[h].row.Intersect(tail[t].row)
|
||||
// } else {
|
||||
// s.row = head[h].row
|
||||
// }
|
||||
// res = append(res, s)
|
||||
// }
|
||||
// }
|
||||
// return res
|
||||
// }
|
||||
|
||||
func (e *executor) executeRows(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (RowIDs, error) {
|
||||
if columnID, ok, err := c.UintArg("column"); err != nil {
|
||||
|
|
@ -2555,3 +2569,110 @@ func filterColumn(col uint64) rowFilter {
|
|||
return colKey == key && c.Contains(colVal), false
|
||||
}
|
||||
}
|
||||
|
||||
// TODO: this works, but it would be more performant if the fragment could seek to the
|
||||
// next row in the rows list rather than asking the filter for each container
|
||||
// serially.
|
||||
func filterWithRows(rows []uint64) rowFilter {
|
||||
loc := 0
|
||||
return func(rowID, key uint64, c *roaring.Container) (include, done bool) {
|
||||
if loc >= len(rows) {
|
||||
return false, true
|
||||
}
|
||||
i := sort.Search(len(rows[loc:]), func(i int) bool {
|
||||
return rows[loc+i] >= rowID
|
||||
})
|
||||
loc += i
|
||||
if loc >= len(rows) {
|
||||
return false, true
|
||||
}
|
||||
if rows[loc] == rowID {
|
||||
if loc == len(rows)-1 {
|
||||
done = true
|
||||
}
|
||||
return true, done
|
||||
}
|
||||
return false, false
|
||||
}
|
||||
}
|
||||
|
||||
type groupByIterator struct {
|
||||
fragments []*fragment
|
||||
current []uint64
|
||||
rows []RowIDs
|
||||
fields []FieldRow
|
||||
}
|
||||
|
||||
func newGroupByIterator(rowIDs []RowIDs, children []*pql.Call, index string, shard uint64, holder *Holder) *groupByIterator {
|
||||
gbi := &groupByIterator{
|
||||
fragments: make([]*fragment, len(rowIDs)),
|
||||
current: make([]uint64, len(rowIDs)),
|
||||
rows: rowIDs,
|
||||
fields: make([]FieldRow, len(rowIDs)),
|
||||
}
|
||||
for i, call := range children {
|
||||
fieldName := call.Args["field"].(string) // this has already been validated by this point
|
||||
gbi.fields[i].Field = fieldName
|
||||
// Fetch fragment.
|
||||
frag := holder.fragment(index, fieldName, viewStandard, shard)
|
||||
if frag == nil { // this means this whole shard doesn't have all it needs to continue
|
||||
return nil
|
||||
}
|
||||
gbi.fragments[i] = frag
|
||||
if prev, hasPrev, err := call.UintArg("previous"); err != nil {
|
||||
panic("getting prev")
|
||||
} else if hasPrev {
|
||||
gbi.current[i] = prev
|
||||
if i == len(children)-1 {
|
||||
gbi.current[i] += 1
|
||||
}
|
||||
}
|
||||
}
|
||||
return gbi
|
||||
}
|
||||
|
||||
func (gbi *groupByIterator) Next() (ret ppi, done bool) {
|
||||
rows := make([]*Row, len(gbi.current))
|
||||
ret.gCnt.Group = make([]FieldRow, len(gbi.current))
|
||||
copy(ret.gCnt.Group, gbi.fields)
|
||||
wrap := false
|
||||
|
||||
// build rows slice
|
||||
for i := len(gbi.current) - 1; i >= 0; i-- {
|
||||
if wrap {
|
||||
gbi.current[i] += 1
|
||||
wrap = false
|
||||
}
|
||||
frag := gbi.fragments[i]
|
||||
filters := []rowFilter{}
|
||||
if len(gbi.rows[i]) > 0 {
|
||||
filters = append(filters, filterWithRows(gbi.rows[i]))
|
||||
}
|
||||
filters = append(filters, filterWithLimit(1))
|
||||
rowIDs := frag.rows(gbi.current[i], filters...)
|
||||
if len(rowIDs) == 0 && i != 0 { // wrap around
|
||||
wrap = true
|
||||
gbi.current[i] = 0
|
||||
if len(gbi.rows[i]) > 0 {
|
||||
gbi.current[i] = gbi.rows[i][0]
|
||||
}
|
||||
rowIDs = frag.rows(gbi.current[i], filters...)
|
||||
}
|
||||
if len(rowIDs) != 0 {
|
||||
rows[i] = frag.row(rowIDs[0])
|
||||
gbi.current[i] = rowIDs[0]
|
||||
ret.gCnt.Group[i].RowID = rowIDs[0]
|
||||
} else {
|
||||
return ppi{}, true
|
||||
}
|
||||
}
|
||||
gbi.current[len(gbi.current)-1] += 1
|
||||
|
||||
// build ppi from rows
|
||||
ret.row = rows[0]
|
||||
for _, row := range rows[1:] {
|
||||
ret.row = ret.row.Intersect(row)
|
||||
}
|
||||
ret.gCnt.Count = ret.row.Count()
|
||||
return ret, false
|
||||
}
|
||||
|
|
|
|||
|
|
@ -138,3 +138,58 @@ func TestFilterWithLimit(t *testing.T) {
|
|||
t.Fatalf("limit filter should have been done, but got inc: %v done: %v", inc, done)
|
||||
}
|
||||
}
|
||||
|
||||
func TestFilterWithRows(t *testing.T) {
|
||||
tests := []struct {
|
||||
rows []uint64
|
||||
callWith []uint64
|
||||
expect [][2]bool
|
||||
}{
|
||||
{
|
||||
rows: []uint64{},
|
||||
callWith: []uint64{0},
|
||||
expect: [][2]bool{{false, true}},
|
||||
},
|
||||
{
|
||||
rows: []uint64{0},
|
||||
callWith: []uint64{0},
|
||||
expect: [][2]bool{{true, true}},
|
||||
},
|
||||
{
|
||||
rows: []uint64{1},
|
||||
callWith: []uint64{0, 2},
|
||||
expect: [][2]bool{{false, false}, {false, true}},
|
||||
},
|
||||
{
|
||||
rows: []uint64{0},
|
||||
callWith: []uint64{1, 2},
|
||||
expect: [][2]bool{{false, true}, {false, true}},
|
||||
},
|
||||
{
|
||||
rows: []uint64{3, 9},
|
||||
callWith: []uint64{1, 2, 3, 10},
|
||||
expect: [][2]bool{{false, false}, {false, false}, {true, false}, {false, true}},
|
||||
},
|
||||
{
|
||||
rows: []uint64{0, 1, 2},
|
||||
callWith: []uint64{0, 1, 2},
|
||||
expect: [][2]bool{{true, false}, {true, false}, {true, true}},
|
||||
},
|
||||
}
|
||||
|
||||
for num, test := range tests {
|
||||
t.Run(fmt.Sprintf("%d_%v_with_%v", num, test.rows, test.callWith), func(t *testing.T) {
|
||||
if len(test.callWith) != len(test.expect) {
|
||||
t.Fatalf("Badly specified test - must expect the same number of values as calls.")
|
||||
}
|
||||
f := filterWithRows(test.rows)
|
||||
for i, id := range test.callWith {
|
||||
inc, done := f(id, 0, nil)
|
||||
if inc != test.expect[i][0] || done != test.expect[i][1] {
|
||||
t.Fatalf("Calling with %d\nexp: %v,%v\ngot: %v,%v", id, test.expect[i][0], test.expect[i][1], inc, done)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -2412,7 +2412,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}}, Count: 2},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=general, previous=10, limit=1))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=general, previous=10), limit=1)`).Results[0].([]pilosa.GroupCount)
|
||||
checkGroupBy(t, expected, results)
|
||||
|
||||
})
|
||||
|
|
@ -2436,6 +2436,48 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
results := c.Query(t, "i", `GroupBy(Rows(field=a), Rows(field=b), limit=1)`).Results[0].([]pilosa.GroupCount)
|
||||
checkGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
// set the same bits in a single shard in three fields
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "wa")
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "wb")
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "wc")
|
||||
c.ImportBits(t, "i", "wa", [][2]uint64{
|
||||
{0, 0}, {0, 1}, {0, 2}, // all
|
||||
{1, 1}, // odds
|
||||
{2, 0}, {2, 2}, // evens
|
||||
{3, 3}, // no overlap
|
||||
})
|
||||
c.ImportBits(t, "i", "wb", [][2]uint64{
|
||||
{0, 0}, {0, 1}, {0, 2},
|
||||
{1, 1},
|
||||
{2, 0}, {2, 2},
|
||||
{3, 3},
|
||||
})
|
||||
c.ImportBits(t, "i", "wc", [][2]uint64{
|
||||
{0, 0}, {0, 1}, {0, 2},
|
||||
{1, 1},
|
||||
{2, 0}, {2, 2},
|
||||
{3, 3},
|
||||
})
|
||||
|
||||
t.Run("test wrapping with previous", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=wa), Rows(field=wb), Rows(field=wc, previous=1), limit=3)`).Results[0].([]pilosa.GroupCount)
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "wa", RowID: 0}, {Field: "wb", RowID: 0}, {Field: "wc", RowID: 2}}, Count: 2},
|
||||
{Group: []pilosa.FieldRow{{Field: "wa", RowID: 0}, {Field: "wb", RowID: 1}, {Field: "wc", RowID: 0}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "wa", RowID: 0}, {Field: "wb", RowID: 1}, {Field: "wc", RowID: 1}}, Count: 1},
|
||||
}
|
||||
checkGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
t.Run("test wrapping multiple", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=wa), Rows(field=wb, previous=2), Rows(field=wc, previous=2), limit=1)`).Results[0].([]pilosa.GroupCount)
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "wa", RowID: 1}, {Field: "wb", RowID: 0}, {Field: "wc", RowID: 0}}, Count: 1},
|
||||
}
|
||||
checkGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
}
|
||||
|
||||
func checkGroupBy(t *testing.T, expected, results []pilosa.GroupCount) {
|
||||
|
|
@ -2450,7 +2492,7 @@ func checkGroupBy(t *testing.T, expected, results []pilosa.GroupCount) {
|
|||
return true
|
||||
}
|
||||
if len(results) != len(expected) {
|
||||
t.Fatalf("number of groupings mismatch: \n%+v\n%+v\n", results, expected)
|
||||
t.Fatalf("number of groupings mismatch:\n got:%+v\nwant:%+v\n", results, expected)
|
||||
}
|
||||
for _, result := range results {
|
||||
if notIn(result, expected) {
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue