From 360623230fb7603abace67f57ee1925360b46976 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Wed, 10 Oct 2018 21:03:45 -0500 Subject: [PATCH] get a somewhat better groupBy working that passes new tests one test still fails due to reordering during merging --- executor.go | 261 ++++++++++++++++++++++++++++---------- executor_internal_test.go | 55 ++++++++ executor_test.go | 46 ++++++- 3 files changed, 290 insertions(+), 72 deletions(-) diff --git a/executor.go b/executor.go index 1cd95eed4..d9bdb54d2 100644 --- a/executor.go +++ b/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 +} diff --git a/executor_internal_test.go b/executor_internal_test.go index 990509128..93ba86743 100644 --- a/executor_internal_test.go +++ b/executor_internal_test.go @@ -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) + } + } + }) + } + +} diff --git a/executor_test.go b/executor_test.go index 12bf0b1e8..30a1f2890 100644 --- a/executor_test.go +++ b/executor_test.go @@ -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) {