mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-07 09:05:55 +00:00
Merge pull request #1647 from jaffee/new-rows-iterate
copy non roaring-import code from Todds's row-iterate PR
This commit is contained in:
Matthew Jaffee
GitHub
commit
0fc08c696a
10 changed files with 2638 additions and 166 deletions
|
|
@ -384,6 +384,15 @@ func encodeQueryResponse(m *pilosa.QueryResponse) *internal.QueryResponse {
|
|||
case bool:
|
||||
pb.Results[i].Type = queryResultTypeBool
|
||||
pb.Results[i].Changed = result
|
||||
case pilosa.RowIDs:
|
||||
pb.Results[i].Type = queryResultTypeRowIDs
|
||||
pb.Results[i].RowIDs = result
|
||||
case []pilosa.GroupCount:
|
||||
pb.Results[i].Type = queryResultTypeGroupCounts
|
||||
pb.Results[i].GroupCounts = encodeGroupCounts(result)
|
||||
case pilosa.RowIdentifiers:
|
||||
pb.Results[i].Type = queryResultTypeRowIdentifiers
|
||||
pb.Results[i].RowIdentifiers = encodeRowIdentifiers(result)
|
||||
case nil:
|
||||
pb.Results[i].Type = queryResultTypeNil
|
||||
}
|
||||
|
|
@ -932,7 +941,6 @@ func decodeQueryResponse(pb *internal.QueryResponse, m *pilosa.QueryResponse) {
|
|||
}
|
||||
m.Results = make([]interface{}, len(pb.Results))
|
||||
decodeQueryResults(pb.Results, m.Results)
|
||||
|
||||
}
|
||||
|
||||
func decodeColumnAttrSets(pb []*internal.ColumnAttrSet, m []*pilosa.ColumnAttrSet) {
|
||||
|
|
@ -962,6 +970,9 @@ const (
|
|||
queryResultTypeValCount
|
||||
queryResultTypeUint64
|
||||
queryResultTypeBool
|
||||
queryResultTypeRowIDs
|
||||
queryResultTypeGroupCounts
|
||||
queryResultTypeRowIdentifiers
|
||||
)
|
||||
|
||||
func decodeQueryResult(pb *internal.QueryResult) interface{} {
|
||||
|
|
@ -978,6 +989,12 @@ func decodeQueryResult(pb *internal.QueryResult) interface{} {
|
|||
return pb.Changed
|
||||
case queryResultTypeNil:
|
||||
return nil
|
||||
case queryResultTypeRowIDs:
|
||||
return pilosa.RowIDs(pb.RowIDs)
|
||||
case queryResultTypeRowIdentifiers:
|
||||
return decodeRowIdentifiers(pb.RowIdentifiers)
|
||||
case queryResultTypeGroupCounts:
|
||||
return decodeGroupCounts(pb.GroupCounts)
|
||||
}
|
||||
panic(fmt.Sprintf("unknown type: %d", pb.Type))
|
||||
}
|
||||
|
|
@ -1028,6 +1045,33 @@ func decodeAttr(attr *internal.Attr) (key string, value interface{}) {
|
|||
}
|
||||
}
|
||||
|
||||
func decodeRowIdentifiers(a *internal.RowIdentifiers) *pilosa.RowIdentifiers {
|
||||
return &pilosa.RowIdentifiers{
|
||||
Rows: a.Rows,
|
||||
Keys: a.Keys,
|
||||
}
|
||||
}
|
||||
|
||||
func decodeGroupCounts(a []*internal.GroupCount) []pilosa.GroupCount {
|
||||
other := make([]pilosa.GroupCount, len(a))
|
||||
for i := range a {
|
||||
other[i] = pilosa.GroupCount{
|
||||
Group: decodeFieldRows(a[i].Group),
|
||||
Count: a[i].Count,
|
||||
}
|
||||
}
|
||||
return other
|
||||
}
|
||||
|
||||
func decodeFieldRows(a []*internal.FieldRow) []pilosa.FieldRow {
|
||||
other := make([]pilosa.FieldRow, len(a))
|
||||
for i := range a {
|
||||
other[i].Field = a[i].Field
|
||||
other[i].RowID = a[i].RowID
|
||||
}
|
||||
return other
|
||||
}
|
||||
|
||||
func decodePairs(a []*internal.Pair) []pilosa.Pair {
|
||||
other := make([]pilosa.Pair, len(a))
|
||||
for i := range a {
|
||||
|
|
@ -1079,6 +1123,36 @@ func encodeRow(r *pilosa.Row) *internal.Row {
|
|||
}
|
||||
}
|
||||
|
||||
func encodeRowIdentifiers(r pilosa.RowIdentifiers) *internal.RowIdentifiers {
|
||||
return &internal.RowIdentifiers{
|
||||
Rows: r.Rows,
|
||||
Keys: r.Keys,
|
||||
//Attrs: encodeAttrs(r.Attrs),
|
||||
}
|
||||
}
|
||||
|
||||
func encodeGroupCounts(counts []pilosa.GroupCount) []*internal.GroupCount {
|
||||
result := make([]*internal.GroupCount, len(counts))
|
||||
for i := range counts {
|
||||
result[i] = &internal.GroupCount{
|
||||
Group: encodeFieldRows(counts[i].Group),
|
||||
Count: counts[i].Count,
|
||||
}
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func encodeFieldRows(a []pilosa.FieldRow) []*internal.FieldRow {
|
||||
other := make([]*internal.FieldRow, len(a))
|
||||
for i := range a {
|
||||
other[i] = &internal.FieldRow{
|
||||
Field: a[i].Field,
|
||||
RowID: a[i].RowID,
|
||||
}
|
||||
}
|
||||
return other
|
||||
}
|
||||
|
||||
func encodePairs(a pilosa.Pairs) []*internal.Pair {
|
||||
other := make([]*internal.Pair, len(a))
|
||||
for i := range a {
|
||||
|
|
|
|||
579
executor.go
579
executor.go
|
|
@ -16,6 +16,7 @@ package pilosa
|
|||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"sort"
|
||||
"time"
|
||||
|
|
@ -257,6 +258,12 @@ func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, s
|
|||
case "TopN":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
return e.executeTopN(ctx, index, c, shards, opt)
|
||||
case "Rows":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
return e.executeRows(ctx, index, c, shards, opt)
|
||||
case "GroupBy":
|
||||
e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
|
||||
return e.executeGroupBy(ctx, index, c, shards, opt)
|
||||
case "Options":
|
||||
return e.executeOptionsCall(ctx, index, c, shards, opt)
|
||||
default:
|
||||
|
|
@ -777,14 +784,323 @@ func (e *executor) executeDifferenceShard(ctx context.Context, index string, c *
|
|||
return other, nil
|
||||
}
|
||||
|
||||
// RowIdentifiers is a return type for a list of
|
||||
// row ids or row keys. The names `Rows` and `Keys`
|
||||
// are meant to follow the same convention as the
|
||||
// Row query which returns `Columns` and `Keys`.
|
||||
// TODO: Rename this to something better. Anything.
|
||||
type RowIdentifiers struct {
|
||||
Rows []uint64 `json:"rows"`
|
||||
Keys []string `json:"keys,omitempty"`
|
||||
}
|
||||
|
||||
// RowIDs is a query return type for just uint64 row ids.
|
||||
// It should only be used internally (since RowIdentifiers
|
||||
// is the external return type), but it is exported because
|
||||
// the proto package needs access to it.
|
||||
type RowIDs []uint64
|
||||
|
||||
func (r RowIDs) merge(other RowIDs, limit int) RowIDs {
|
||||
i, j := 0, 0
|
||||
result := make(RowIDs, 0)
|
||||
for i < len(r) && j < len(other) && len(result) < limit {
|
||||
av, bv := r[i], other[j]
|
||||
if av < bv {
|
||||
result = append(result, av)
|
||||
i++
|
||||
} else if av > bv {
|
||||
result = append(result, bv)
|
||||
j++
|
||||
} else {
|
||||
result = append(result, bv)
|
||||
i++
|
||||
j++
|
||||
}
|
||||
}
|
||||
for i < len(r) && len(result) < limit {
|
||||
result = append(result, r[i])
|
||||
i++
|
||||
}
|
||||
for j < len(other) && len(result) < limit {
|
||||
result = append(result, other[j])
|
||||
j++
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]GroupCount, error) {
|
||||
// validate call
|
||||
if len(c.Children) == 0 {
|
||||
return nil, errors.New("need at least one child call")
|
||||
}
|
||||
limit := int(^uint(0) >> 1)
|
||||
if lim, hasLimit, err := c.UintArg("limit"); err != nil {
|
||||
return nil, err
|
||||
} else if hasLimit {
|
||||
limit = int(lim)
|
||||
}
|
||||
|
||||
// perform necessary Rows queries (any that have limit or columns args) -
|
||||
// TODO, call async? would only help if multiple Rows queries had a column
|
||||
// or limit arg.
|
||||
// TODO support TopN in here would be really cool - and pretty easy I think.
|
||||
childRows := make([]RowIDs, len(c.Children))
|
||||
for i, child := range c.Children {
|
||||
if child.Name != "Rows" {
|
||||
return nil, errors.Errorf("'%s' is not a valid child query for GroupBy, must be 'Rows'", c.Name)
|
||||
}
|
||||
_, hasLimit, err := child.UintArg("limit")
|
||||
if err != nil {
|
||||
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
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Execute calls in bulk on each remote node and merge.
|
||||
mapFn := func(shard uint64) (interface{}, error) {
|
||||
return e.executeGroupByShard(ctx, index, c, shard, childRows)
|
||||
}
|
||||
// Merge returned results at coordinating node.
|
||||
reduceFn := func(prev, v interface{}) interface{} {
|
||||
other, _ := prev.([]GroupCount)
|
||||
return mergeGroupCounts(other, v.([]GroupCount), limit)
|
||||
}
|
||||
// Get full result set.
|
||||
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results, _ := other.([]GroupCount)
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// FieldRow is used to distinguish rows in a group by result.
|
||||
type FieldRow struct {
|
||||
Field string `json:"field"`
|
||||
RowID uint64 `json:"rowID"`
|
||||
RowKey string `json:"rowKey,omitempty"`
|
||||
}
|
||||
|
||||
func (fr FieldRow) MarshalJSON() ([]byte, error) {
|
||||
if fr.RowKey != "" {
|
||||
return json.Marshal(struct {
|
||||
Field string `json:"field"`
|
||||
RowKey string `json:"rowKey"`
|
||||
}{
|
||||
Field: fr.Field,
|
||||
RowKey: fr.RowKey,
|
||||
})
|
||||
}
|
||||
return json.Marshal(struct {
|
||||
Field string `json:"field"`
|
||||
RowID uint64 `json:"rowID"`
|
||||
}{
|
||||
Field: fr.Field,
|
||||
RowID: fr.RowID,
|
||||
})
|
||||
}
|
||||
|
||||
func (fr FieldRow) String() string {
|
||||
return fmt.Sprintf("%s.%d", fr.Field, fr.RowID)
|
||||
}
|
||||
|
||||
type GroupCount struct {
|
||||
Group []FieldRow `json:"group"`
|
||||
Count uint64 `json:"count"`
|
||||
}
|
||||
|
||||
// mergeGroupCounts merges two slices of GroupCounts throwing away any that go
|
||||
// beyond the limit. It assume that the two slices are sorted by the row ids in
|
||||
// the fields of the group counts. It may modify its arguments.
|
||||
func mergeGroupCounts(a, b []GroupCount, limit int) []GroupCount {
|
||||
if limit > len(a)+len(b) {
|
||||
limit = len(a) + len(b)
|
||||
}
|
||||
ret := make([]GroupCount, 0, limit)
|
||||
i, j := 0, 0
|
||||
for i < len(a) && j < len(b) && len(ret) < limit {
|
||||
switch a[i].Compare(b[j]) {
|
||||
case -1:
|
||||
ret = append(ret, a[i])
|
||||
i++
|
||||
case 0:
|
||||
a[i].Count += b[j].Count
|
||||
ret = append(ret, a[i])
|
||||
i++
|
||||
j++
|
||||
case 1:
|
||||
ret = append(ret, b[j])
|
||||
j++
|
||||
}
|
||||
}
|
||||
for ; i < len(a) && len(ret) < limit; i++ {
|
||||
ret = append(ret, a[i])
|
||||
}
|
||||
for ; j < len(b) && len(ret) < limit; j++ {
|
||||
ret = append(ret, b[j])
|
||||
}
|
||||
return ret
|
||||
}
|
||||
|
||||
func (g GroupCount) Compare(o GroupCount) int {
|
||||
for i := range g.Group {
|
||||
if g.Group[i].RowID < o.Group[i].RowID {
|
||||
return -1
|
||||
}
|
||||
if g.Group[i].RowID > o.Group[i].RowID {
|
||||
return 1
|
||||
}
|
||||
}
|
||||
return 0
|
||||
}
|
||||
|
||||
func (e *executor) executeGroupByShard(_ context.Context, index string, c *pql.Call, shard uint64, childRows []RowIDs) ([]GroupCount, error) {
|
||||
iter, err := newGroupByIterator(childRows, c.Children, index, shard, e.Holder)
|
||||
if err != nil {
|
||||
return nil, errors.Wrapf(err, "getting group by iterator for shard %d", shard)
|
||||
}
|
||||
if iter == nil {
|
||||
return []GroupCount{}, nil
|
||||
}
|
||||
|
||||
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)
|
||||
|
||||
num := 0
|
||||
for gc, done := iter.Next(); !done && num < limit; gc, done = iter.Next() {
|
||||
if gc.Count > 0 {
|
||||
num++
|
||||
results = append(results, gc)
|
||||
}
|
||||
}
|
||||
|
||||
return results, nil
|
||||
}
|
||||
|
||||
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 {
|
||||
return nil, errors.Wrap(err, "getting column")
|
||||
} else if ok {
|
||||
shards = []uint64{columnID / ShardWidth}
|
||||
}
|
||||
|
||||
// Execute calls in bulk on each remote node and merge.
|
||||
mapFn := func(shard uint64) (interface{}, error) {
|
||||
return e.executeRowsShard(ctx, index, c, shard)
|
||||
}
|
||||
|
||||
// Determine limit so we can use it when reducing.
|
||||
limit := int(^uint(0) >> 1)
|
||||
if lim, hasLimit, err := c.UintArg("limit"); err != nil {
|
||||
return nil, err
|
||||
} else if hasLimit {
|
||||
limit = int(lim)
|
||||
}
|
||||
|
||||
// Merge returned results at coordinating node.
|
||||
reduceFn := func(prev, v interface{}) interface{} {
|
||||
other, _ := prev.(RowIDs)
|
||||
return other.merge(v.(RowIDs), limit)
|
||||
}
|
||||
// Get full result set.
|
||||
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results, _ := other.(RowIDs)
|
||||
return results, nil
|
||||
}
|
||||
|
||||
func (e *executor) executeRowsShard(_ context.Context, index string, c *pql.Call, shard uint64) (RowIDs, error) {
|
||||
// Fetch index.
|
||||
idx := e.Holder.Index(index)
|
||||
if idx == nil {
|
||||
return nil, ErrIndexNotFound
|
||||
}
|
||||
// Fetch field name from argument.
|
||||
fieldName, ok := c.Args["field"].(string)
|
||||
if !ok {
|
||||
return nil, errors.New("Rows() argument required: field")
|
||||
}
|
||||
// Fetch field.
|
||||
f := e.Holder.Field(index, fieldName)
|
||||
if f == nil {
|
||||
return nil, ErrFieldNotFound
|
||||
}
|
||||
frag := e.Holder.fragment(index, fieldName, viewStandard, shard)
|
||||
if frag == nil {
|
||||
return make(RowIDs, 0), nil
|
||||
}
|
||||
|
||||
start := uint64(0)
|
||||
if previous, ok, err := c.UintArg("previous"); err != nil {
|
||||
return nil, errors.Wrap(err, "getting previous")
|
||||
} else if ok {
|
||||
start = previous + 1
|
||||
}
|
||||
|
||||
filters := []rowFilter{}
|
||||
if columnID, ok, err := c.UintArg("column"); err != nil {
|
||||
return nil, err
|
||||
} else if ok {
|
||||
colShard := columnID >> shardWidthExponent
|
||||
if colShard != shard {
|
||||
return RowIDs{}, nil
|
||||
}
|
||||
filters = append(filters, filterColumn(columnID))
|
||||
}
|
||||
if limit, hasLimit, err := c.UintArg("limit"); err != nil {
|
||||
return nil, errors.Wrap(err, "getting limit")
|
||||
} else if hasLimit {
|
||||
filters = append(filters, filterWithLimit(limit))
|
||||
}
|
||||
|
||||
return frag.rows(start, filters...), nil
|
||||
}
|
||||
|
||||
func (e *executor) executeBitmapShard(_ context.Context, index string, c *pql.Call, shard uint64) (*Row, error) {
|
||||
// Fetch column label from index.
|
||||
// Fetch index.
|
||||
idx := e.Holder.Index(index)
|
||||
if idx == nil {
|
||||
return nil, ErrIndexNotFound
|
||||
}
|
||||
|
||||
// Fetch field & row label based on argument.
|
||||
// Fetch field name from argument.
|
||||
fieldName, err := c.FieldArg()
|
||||
if err != nil {
|
||||
return nil, errors.New("Row() argument required: field")
|
||||
|
|
@ -1867,6 +2183,12 @@ func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
|
|||
// Positional args in new PQL syntax require special handling here.
|
||||
rowKey = "_" + rowLabel
|
||||
fieldName = callArgString(c, "_field")
|
||||
case "Rows":
|
||||
fieldName = callArgString(c, "field")
|
||||
rowKey = "previous"
|
||||
colKey = "column"
|
||||
case "GroupBy":
|
||||
return errors.Wrap(e.translateGroupByCall(index, idx, c), "translating GroupBy")
|
||||
default:
|
||||
colKey = "col"
|
||||
fieldName = callArgString(c, "field")
|
||||
|
|
@ -1942,6 +2264,61 @@ func (e *executor) translateCall(index string, idx *Index, c *pql.Call) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (e *executor) translateGroupByCall(index string, idx *Index, c *pql.Call) error {
|
||||
if c.Name != "GroupBy" {
|
||||
panic("translateGroupByCall called with '" + c.Name + "'")
|
||||
}
|
||||
|
||||
for _, child := range c.Children {
|
||||
if err := e.translateCall(index, idx, child); err != nil {
|
||||
return errors.Wrapf(err, "translating %s", child)
|
||||
}
|
||||
}
|
||||
|
||||
prev, ok := c.Args["previous"]
|
||||
if !ok {
|
||||
return nil // nothing else to be translated
|
||||
}
|
||||
previous, ok := prev.([]interface{})
|
||||
if !ok {
|
||||
return errors.Errorf("'previous' argument must be list, but got %T", prev)
|
||||
}
|
||||
if len(c.Children) != len(previous) {
|
||||
return errors.Errorf("mismatched lengths for previous: %d and children: %d in %s", len(previous), len(c.Children), c)
|
||||
}
|
||||
|
||||
fields := make([]*Field, len(c.Children))
|
||||
for i, child := range c.Children {
|
||||
fieldname := callArgString(child, "field")
|
||||
field := idx.Field(fieldname)
|
||||
if field == nil {
|
||||
return errors.Wrapf(ErrFieldNotFound, "getting field '%s' from '%s'", fieldname, child)
|
||||
}
|
||||
fields[i] = field
|
||||
}
|
||||
|
||||
for i, field := range fields {
|
||||
prev := previous[i]
|
||||
if field.keys() {
|
||||
prevStr, ok := prev.(string)
|
||||
if !ok {
|
||||
return errors.New("prev value must be a string when field 'keys' option enabled")
|
||||
}
|
||||
ids, err := e.TranslateStore.TranslateRowsToUint64(index, field.Name(), []string{prevStr})
|
||||
if err != nil {
|
||||
return errors.Wrapf(err, "translating row key '%s'", prevStr)
|
||||
}
|
||||
previous[i] = ids[0]
|
||||
} else {
|
||||
if prevStr, ok := prev.(string); ok {
|
||||
return errors.Errorf("got string row val '%s' in 'previous' for field %s which doesn't use string keys", prevStr, field.Name())
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *executor) translateResult(index string, idx *Index, call *pql.Call, result interface{}) (interface{}, error) {
|
||||
switch result := result.(type) {
|
||||
case *Row:
|
||||
|
|
@ -1977,7 +2354,62 @@ func (e *executor) translateResult(index string, idx *Index, call *pql.Call, res
|
|||
return other, nil
|
||||
}
|
||||
}
|
||||
|
||||
case []GroupCount:
|
||||
other := make([]GroupCount, 0)
|
||||
for _, gl := range result {
|
||||
|
||||
group := make([]FieldRow, len(gl.Group))
|
||||
for i, g := range gl.Group {
|
||||
group[i] = g
|
||||
|
||||
// TODO: It may be useful to cache this field lookup.
|
||||
field := idx.Field(g.Field)
|
||||
if field == nil {
|
||||
return nil, ErrFieldNotFound
|
||||
}
|
||||
if field.keys() {
|
||||
key, err := e.TranslateStore.TranslateRowToString(index, g.Field, g.RowID)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "translating row ID in Group")
|
||||
}
|
||||
group[i].RowKey = key
|
||||
}
|
||||
}
|
||||
|
||||
other = append(other, GroupCount{
|
||||
Group: group,
|
||||
Count: gl.Count,
|
||||
})
|
||||
}
|
||||
return other, nil
|
||||
|
||||
case RowIDs:
|
||||
other := RowIdentifiers{}
|
||||
|
||||
fieldName := callArgString(call, "field")
|
||||
if fieldName == "" {
|
||||
return nil, ErrFieldNotFound
|
||||
}
|
||||
|
||||
if field := idx.Field(fieldName); field == nil {
|
||||
return nil, ErrFieldNotFound
|
||||
} else if field.keys() {
|
||||
other.Keys = make([]string, len(result))
|
||||
for i, id := range result {
|
||||
key, err := e.TranslateStore.TranslateRowToString(index, fieldName, id)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "translating row ID")
|
||||
}
|
||||
other.Keys[i] = key
|
||||
}
|
||||
} else {
|
||||
other.Rows = result
|
||||
}
|
||||
|
||||
return other, nil
|
||||
}
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
|
|
@ -2026,7 +2458,7 @@ func needsShards(calls []*pql.Call) bool {
|
|||
switch call.Name {
|
||||
case "Clear", "Set", "SetRowAttrs", "SetColumnAttrs":
|
||||
continue
|
||||
case "Count", "TopN":
|
||||
case "Count", "TopN", "Rows":
|
||||
return true
|
||||
// default catches Bitmap calls
|
||||
default:
|
||||
|
|
@ -2096,3 +2528,144 @@ func isString(v interface{}) bool {
|
|||
_, ok := v.(string)
|
||||
return ok
|
||||
}
|
||||
|
||||
// groupByIterator contains several slices. Each slice contains a number of
|
||||
// elements equal to the number of fields in the group by (the number of Rows
|
||||
// calls).
|
||||
type groupByIterator struct {
|
||||
// rowIters contains a rowIterator for each of the fields in the Group By.
|
||||
rowIters []*rowIterator
|
||||
// rows contains the current row data for each of the fields in the Group
|
||||
// By. Each row is the intersection of itself and the rows of the fields
|
||||
// with an index lower than its own. This is a performance optimization so
|
||||
// that the expected common case of getting the next row in the furthest
|
||||
// field to the right require only a single intersect with the row of the
|
||||
// previous field to determine the count of the new group.
|
||||
rows []struct {
|
||||
row *Row
|
||||
id uint64
|
||||
}
|
||||
|
||||
// fields helps with the construction of GroupCount results by holding all
|
||||
// the field names that are being grouped by. Each results makes a copy of
|
||||
// fields and then sets the row ids.
|
||||
fields []FieldRow
|
||||
done bool
|
||||
}
|
||||
|
||||
// newGroupByIterator initializes a new groupByIterator.
|
||||
func newGroupByIterator(rowIDs []RowIDs, children []*pql.Call, index string, shard uint64, holder *Holder) (*groupByIterator, error) {
|
||||
gbi := &groupByIterator{
|
||||
rowIters: make([]*rowIterator, len(children)),
|
||||
rows: make([]struct {
|
||||
row *Row
|
||||
id uint64
|
||||
}, len(children)),
|
||||
fields: make([]FieldRow, len(children)),
|
||||
}
|
||||
|
||||
ignorePrev := false
|
||||
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, nil
|
||||
}
|
||||
filters := []rowFilter{}
|
||||
if len(rowIDs[i]) > 0 {
|
||||
filters = append(filters, filterWithRows(rowIDs[i]))
|
||||
}
|
||||
gbi.rowIters[i] = frag.rowIterator(i != 0, filters...)
|
||||
|
||||
prev, hasPrev, err := call.UintArg("previous")
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "getting previous")
|
||||
} else if hasPrev && !ignorePrev {
|
||||
if i == len(children)-1 {
|
||||
prev += 1
|
||||
}
|
||||
gbi.rowIters[i].Seek(prev)
|
||||
}
|
||||
nextRow, rowID, wrapped := gbi.rowIters[i].Next()
|
||||
if nextRow == nil {
|
||||
gbi.done = true
|
||||
return gbi, nil
|
||||
}
|
||||
gbi.rows[i].row = nextRow
|
||||
gbi.rows[i].id = rowID
|
||||
if hasPrev && rowID != prev {
|
||||
// ignorePrev signals that we didn't find a previous row, so all
|
||||
// Rows queries "deeper" than it need to ignore the previous
|
||||
// argument and start at the beginning.
|
||||
ignorePrev = true
|
||||
}
|
||||
if wrapped {
|
||||
// if a field has wrapped, we need to get the next row for the
|
||||
// previous field, and if that one wraps we need to keep going
|
||||
// backward.
|
||||
for j := i - 1; j >= 0; j-- {
|
||||
nextRow, rowID, wrapped := gbi.rowIters[j].Next()
|
||||
if nextRow == nil {
|
||||
gbi.done = true
|
||||
return gbi, nil
|
||||
}
|
||||
gbi.rows[j].row = nextRow
|
||||
gbi.rows[j].id = rowID
|
||||
if !wrapped {
|
||||
break
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
for i := 1; i < len(gbi.rows)-1; i++ {
|
||||
gbi.rows[i].row = gbi.rows[i].row.Intersect(gbi.rows[i-1].row)
|
||||
}
|
||||
|
||||
return gbi, nil
|
||||
}
|
||||
|
||||
// nextAtIdx is a recursive helper method for getting the next row for the field
|
||||
// at index i, and then updating the rows in the "higher" fields if it wraps.
|
||||
func (gbi *groupByIterator) nextAtIdx(i int) {
|
||||
nr, rowID, wrapped := gbi.rowIters[i].Next()
|
||||
if nr == nil {
|
||||
gbi.done = true
|
||||
return
|
||||
}
|
||||
if wrapped && i != 0 {
|
||||
gbi.nextAtIdx(i - 1)
|
||||
}
|
||||
if i != 0 && i != len(gbi.rows)-1 {
|
||||
gbi.rows[i].row = nr.Intersect(gbi.rows[i-1].row)
|
||||
} else {
|
||||
gbi.rows[i].row = nr
|
||||
}
|
||||
gbi.rows[i].id = rowID
|
||||
}
|
||||
|
||||
// Next returns a GroupCount representing the next group by record. When there
|
||||
// are no more records it will return an empty GroupCount and done==true.
|
||||
func (gbi *groupByIterator) Next() (ret GroupCount, done bool) {
|
||||
if gbi.done {
|
||||
return ret, true
|
||||
}
|
||||
if len(gbi.rows) == 1 {
|
||||
ret.Count = gbi.rows[len(gbi.rows)-1].row.Count()
|
||||
} else {
|
||||
ret.Count = gbi.rows[len(gbi.rows)-1].row.intersectionCount(gbi.rows[len(gbi.rows)-2].row)
|
||||
}
|
||||
|
||||
ret.Group = make([]FieldRow, len(gbi.rows))
|
||||
copy(ret.Group, gbi.fields)
|
||||
for i, r := range gbi.rows {
|
||||
ret.Group[i].RowID = r.id
|
||||
}
|
||||
|
||||
// set up for next call
|
||||
gbi.nextAtIdx(len(gbi.rows) - 1)
|
||||
|
||||
return ret, false
|
||||
}
|
||||
|
|
|
|||
224
executor_internal_test.go
Normal file
224
executor_internal_test.go
Normal file
|
|
@ -0,0 +1,224 @@
|
|||
package pilosa
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"io/ioutil"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/pilosa/pilosa/pql"
|
||||
)
|
||||
|
||||
func TestExecutor_TranslateGroupByCall(t *testing.T) {
|
||||
e := &executor{
|
||||
Holder: NewHolder(),
|
||||
}
|
||||
e.Holder.Path, _ = ioutil.TempDir("", "")
|
||||
err := e.Holder.Open()
|
||||
if err != nil {
|
||||
t.Fatalf("opening holder: %v", err)
|
||||
}
|
||||
|
||||
e.TranslateStore = e.Holder.translateFile
|
||||
tf, _ := ioutil.TempFile("", "")
|
||||
e.Holder.translateFile.Path = tf.Name()
|
||||
err = e.Holder.translateFile.Open()
|
||||
if err != nil {
|
||||
t.Fatalf("opening translateFile: %v", err)
|
||||
}
|
||||
|
||||
idx, err := e.Holder.CreateIndex("i", IndexOptions{})
|
||||
if err != nil {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
}
|
||||
|
||||
_, erra := idx.CreateField("ak", OptFieldKeys())
|
||||
_, errb := idx.CreateField("b")
|
||||
_, errc := idx.CreateField("ck", OptFieldKeys())
|
||||
if erra != nil || errb != nil || errc != nil {
|
||||
t.Fatalf("creating fields %v, %v, %v", erra, errb, errc)
|
||||
}
|
||||
|
||||
_, erra = e.TranslateStore.TranslateRowsToUint64("i", "ak", []string{"la"})
|
||||
_, errb = e.TranslateStore.TranslateRowsToUint64("i", "ck", []string{"ha"})
|
||||
if erra != nil || errb != nil {
|
||||
t.Fatalf("translating rows %v, %v", erra, errb)
|
||||
}
|
||||
|
||||
query, err := pql.ParseString(`GroupBy(Rows(field=ak), Rows(field=b), Rows(field=ck), previous=["la", 0, "ha"])`)
|
||||
if err != nil {
|
||||
t.Fatalf("parsing query: %v", err)
|
||||
}
|
||||
c := query.Calls[0]
|
||||
err = e.translateGroupByCall("i", idx, c)
|
||||
if err != nil {
|
||||
t.Fatalf("translating call: %v", err)
|
||||
}
|
||||
if len(c.Args["previous"].([]interface{})) != 3 {
|
||||
t.Fatalf("unexpected length for 'previous' arg %v", c.Args["previous"])
|
||||
}
|
||||
for i, v := range c.Args["previous"].([]interface{}) {
|
||||
if !isInt(v) {
|
||||
t.Fatalf("expected all items in previous to be ints, but '%v' at index %d is %[1]T", v, i)
|
||||
}
|
||||
}
|
||||
|
||||
errTests := []struct {
|
||||
pql string
|
||||
err string
|
||||
}{
|
||||
{
|
||||
pql: `GroupBy(Rows(field=notfound), previous=1)`,
|
||||
err: "'previous' argument must be list",
|
||||
},
|
||||
{
|
||||
pql: `GroupBy(Rows(field=ak), previous=["la", 0])`,
|
||||
err: "mismatched lengths",
|
||||
},
|
||||
{
|
||||
pql: `GroupBy(Rows(field=ak), previous=[1])`,
|
||||
err: "prev value must be a string",
|
||||
},
|
||||
{
|
||||
pql: `GroupBy(Rows(field=notfound), previous=[1])`,
|
||||
err: ErrFieldNotFound.Error(),
|
||||
},
|
||||
// TODO: an unknown key will actually allocate an id. this is probably bad.
|
||||
// {
|
||||
// pql: `GroupBy(Rows(field=ak), previous=["zoop"])`,
|
||||
// err: "translating row key '",
|
||||
// },
|
||||
{
|
||||
pql: `GroupBy(Rows(field=b), previous=["la"])`,
|
||||
err: "which doesn't use string keys",
|
||||
},
|
||||
}
|
||||
|
||||
for i, test := range errTests {
|
||||
t.Run(fmt.Sprintf("#%d_%s", i, test.err), func(t *testing.T) {
|
||||
query, err := pql.ParseString(test.pql)
|
||||
if err != nil {
|
||||
t.Fatalf("parsing query: %v", err)
|
||||
}
|
||||
c := query.Calls[0]
|
||||
err = e.translateGroupByCall("i", idx, c)
|
||||
if err == nil {
|
||||
t.Fatalf("expected error, but translated call is '%s", c)
|
||||
}
|
||||
if !strings.Contains(err.Error(), test.err) {
|
||||
t.Fatalf("expected '%s', got '%v'", test.err, err)
|
||||
}
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func isInt(a interface{}) bool {
|
||||
switch a.(type) {
|
||||
case int, int64, uint, uint64:
|
||||
return true
|
||||
default:
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
func TestFilterWithLimit(t *testing.T) {
|
||||
f := filterWithLimit(5)
|
||||
|
||||
for i := uint64(0); i < 5; i++ {
|
||||
include, done := f(i, i*(1<<shardVsContainerExponent), nil)
|
||||
if done {
|
||||
t.Fatalf("limit filter ended early on iteration %d", i)
|
||||
}
|
||||
if !include {
|
||||
t.Fatalf("limit filter should always include until done")
|
||||
}
|
||||
}
|
||||
inc, done := f(5, 5*(1<<shardVsContainerExponent)+1, nil)
|
||||
if !done {
|
||||
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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func TestFieldRowMarshalJSON(t *testing.T) {
|
||||
fr := FieldRow{
|
||||
Field: "blah",
|
||||
RowID: 0,
|
||||
RowKey: "ha",
|
||||
}
|
||||
b, err := json.Marshal(fr)
|
||||
if err != nil {
|
||||
t.Fatalf("marshalling fieldrow: %v", err)
|
||||
}
|
||||
if string(b) != `{"field":"blah","rowKey":"ha"}` {
|
||||
t.Fatalf("unexpected json: %s", b)
|
||||
}
|
||||
|
||||
fr = FieldRow{
|
||||
Field: "blah",
|
||||
RowID: 2,
|
||||
RowKey: "",
|
||||
}
|
||||
b, err = json.Marshal(fr)
|
||||
if err != nil {
|
||||
t.Fatalf("marshalling fieldrow: %v", err)
|
||||
}
|
||||
if string(b) != `{"field":"blah","rowID":2}` {
|
||||
t.Fatalf("unexpected json: %s", b)
|
||||
}
|
||||
}
|
||||
502
executor_test.go
502
executor_test.go
|
|
@ -1894,6 +1894,22 @@ Set(4500001, fn=4)
|
|||
t.Fatalf("wrong attrs: %v", attrst)
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("remote groupBy", func(t *testing.T) {
|
||||
if res, err := c[1].API.Query(context.Background(), &pilosa.QueryRequest{
|
||||
Index: "i",
|
||||
Query: `GroupBy(Rows(field=f))`,
|
||||
}); err != nil {
|
||||
t.Fatalf("GroupBy querying: %v", err)
|
||||
} else {
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "f", RowID: 7}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "f", RowID: 10}}, Count: 4},
|
||||
}
|
||||
results := res.Results[0].([]pilosa.GroupCount)
|
||||
checkGroupBy(t, expected, results)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
// Ensure executor returns an error if too many writes are in a single request.
|
||||
|
|
@ -2623,6 +2639,492 @@ func benchmarkExistence(nn bool, b *testing.B) {
|
|||
func BenchmarkExecutor_Existence_True(b *testing.B) { benchmarkExistence(true, b) }
|
||||
func BenchmarkExecutor_Existence_False(b *testing.B) { benchmarkExistence(false, b) }
|
||||
|
||||
func TestExecutor_Execute_Rows(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 3)
|
||||
defer c.Close()
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "general")
|
||||
c.ImportBits(t, "i", "general", [][2]uint64{
|
||||
{10, 0},
|
||||
{10, ShardWidth + 1},
|
||||
{11, 2},
|
||||
{11, ShardWidth + 2},
|
||||
{12, 2},
|
||||
{12, ShardWidth + 2},
|
||||
{13, 3},
|
||||
})
|
||||
|
||||
rows := c.Query(t, "i", `Rows(field=general)`).Results[0].(pilosa.RowIdentifiers)
|
||||
if !reflect.DeepEqual(rows, pilosa.RowIdentifiers{Rows: []uint64{10, 11, 12, 13}}) {
|
||||
t.Fatalf("unexpected rows: %+v", rows)
|
||||
}
|
||||
|
||||
rows = c.Query(t, "i", `Rows(field=general, limit=2)`).Results[0].(pilosa.RowIdentifiers)
|
||||
if !reflect.DeepEqual(rows, pilosa.RowIdentifiers{Rows: []uint64{10, 11}}) {
|
||||
t.Fatalf("unexpected rows: %+v", rows)
|
||||
}
|
||||
|
||||
rows = c.Query(t, "i", `Rows(field=general, previous=10,limit=2)`).Results[0].(pilosa.RowIdentifiers)
|
||||
if !reflect.DeepEqual(rows, pilosa.RowIdentifiers{Rows: []uint64{11, 12}}) {
|
||||
t.Fatalf("unexpected rows: %+v", rows)
|
||||
}
|
||||
|
||||
rows = c.Query(t, "i", `Rows(field=general, column=2)`).Results[0].(pilosa.RowIdentifiers)
|
||||
if !reflect.DeepEqual(rows, pilosa.RowIdentifiers{Rows: []uint64{11, 12}}) {
|
||||
t.Fatalf("unexpected rows: %+v", rows)
|
||||
}
|
||||
}
|
||||
|
||||
func TestExecutor_Execute_Rows_Keys(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
|
||||
_, err := c[0].API.CreateIndex(context.Background(), "i", pilosa.IndexOptions{Keys: true})
|
||||
if err != nil {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
}
|
||||
|
||||
_, err = c[0].API.CreateField(context.Background(), "i", "f", pilosa.OptFieldKeys())
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
|
||||
// setup some data. 10 bits in each of shards 0 through 9. starting at
|
||||
// row/col shardNum and progressing to row/col shardNum+10. Also set the
|
||||
// previous 2 for each bit if row >0.
|
||||
query := strings.Builder{}
|
||||
for shard := 0; shard < 10; shard++ {
|
||||
for i := shard; i < shard+10; i++ {
|
||||
for row := i; row >= 0 && row > i-3; row-- {
|
||||
query.WriteString(fmt.Sprintf("Set(\"%d\", f=\"%d\")", shard*pilosa.ShardWidth+i, row))
|
||||
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
_, err = c[0].API.Query(context.Background(), &pilosa.QueryRequest{
|
||||
Index: "i",
|
||||
Query: query.String(),
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("querying: %v", err)
|
||||
}
|
||||
|
||||
tests := []struct {
|
||||
q string
|
||||
exp []string
|
||||
}{
|
||||
{
|
||||
q: `Rows(field=f)`,
|
||||
exp: []string{"0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "10", "11", "12", "13", "14", "15", "16", "17", "18"},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, limit=2)`,
|
||||
exp: []string{"0", "1"},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, previous="15")`,
|
||||
exp: []string{"16", "17", "18"},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, previous="11", limit=2)`,
|
||||
exp: []string{"12", "13"},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, previous="17", limit=5)`,
|
||||
exp: []string{"18"},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, previous="18")`,
|
||||
exp: []string{},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, previous="1", limit=0)`,
|
||||
exp: []string{},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, column="1")`,
|
||||
exp: []string{"0", "1"},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, column="2")`,
|
||||
exp: []string{"0", "1", "2"},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, column="3")`,
|
||||
exp: []string{"1", "2", "3"},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, limit=2, column="3")`,
|
||||
exp: []string{"1", "2"},
|
||||
},
|
||||
{
|
||||
q: fmt.Sprintf(`Rows(field=f, previous="15", column="%d")`, ShardWidth*9+17),
|
||||
exp: []string{"16", "17"},
|
||||
},
|
||||
{
|
||||
q: fmt.Sprintf(`Rows(field=f, previous="11", limit=2, column="%d")`, ShardWidth*5+14),
|
||||
exp: []string{"12", "13"},
|
||||
},
|
||||
{
|
||||
q: fmt.Sprintf(`Rows(field=f, previous="17", limit=5, column="%d")`, ShardWidth*9+18),
|
||||
exp: []string{"18"},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, previous="18", column="19")`,
|
||||
exp: []string{},
|
||||
},
|
||||
{
|
||||
q: `Rows(field=f, previous="1", limit=0, column="0")`,
|
||||
exp: []string{},
|
||||
},
|
||||
}
|
||||
|
||||
for i, test := range tests {
|
||||
t.Run(fmt.Sprintf("#%d_%s", i, test.q), func(t *testing.T) {
|
||||
if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: test.q}); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if rows := res.Results[0].(pilosa.RowIdentifiers); !reflect.DeepEqual(
|
||||
rows, pilosa.RowIdentifiers{Keys: test.exp}) {
|
||||
t.Fatalf("\ngot: %+v\nexp: %+v", rows, pilosa.RowIdentifiers{Keys: test.exp})
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
func TestExecutor_Execute_GroupBy(t *testing.T) {
|
||||
groupByTest := func(t *testing.T, clusterSize int) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "general")
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "sub")
|
||||
c.ImportBits(t, "i", "general", [][2]uint64{
|
||||
{10, 0},
|
||||
{10, 1},
|
||||
{10, ShardWidth + 1},
|
||||
{11, 2},
|
||||
{11, ShardWidth + 2},
|
||||
{12, 2},
|
||||
{12, ShardWidth + 2},
|
||||
})
|
||||
c.ImportBits(t, "i", "sub", [][2]uint64{
|
||||
{100, 0},
|
||||
{100, 1},
|
||||
{100, 3},
|
||||
{100, ShardWidth + 1},
|
||||
|
||||
{110, 2},
|
||||
{110, 0},
|
||||
})
|
||||
|
||||
t.Run("No Field List Arguments", func(t *testing.T) {
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy()`}); err != nil {
|
||||
if !strings.Contains(err.Error(), "need at least one child call") {
|
||||
t.Fatalf("unexpected error: \"%v\"", err)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("Unknown Field ", func(t *testing.T) {
|
||||
if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(Rows(field=missing))`}); err != nil {
|
||||
if errors.Cause(err) != pilosa.ErrFieldNotFound {
|
||||
t.Fatalf("unexpected error\n\"%s\" not returned instead \n\"%s\"", pilosa.ErrFieldNotFound, err)
|
||||
}
|
||||
}
|
||||
})
|
||||
|
||||
t.Run("Basic", func(t *testing.T) {
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 3},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}, {Field: "sub", RowID: 110}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}, {Field: "sub", RowID: 110}}, Count: 1},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=general), Rows(field=sub))`).Results[0].([]pilosa.GroupCount)
|
||||
checkGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
t.Run("check field offset no limit", func(t *testing.T) {
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}}, Count: 2},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}}, Count: 2},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=general, previous=10))`).Results[0].([]pilosa.GroupCount)
|
||||
checkGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
t.Run("check field offset limit", func(t *testing.T) {
|
||||
expected := []pilosa.GroupCount{
|
||||
{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)
|
||||
checkGroupBy(t, expected, results)
|
||||
|
||||
})
|
||||
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "a")
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "b")
|
||||
c.ImportBits(t, "i", "a", [][2]uint64{
|
||||
{0, 1},
|
||||
{1, ShardWidth + 1},
|
||||
})
|
||||
c.ImportBits(t, "i", "b", [][2]uint64{
|
||||
{0, ShardWidth + 1},
|
||||
{1, 1},
|
||||
})
|
||||
|
||||
t.Run("tricky data", func(t *testing.T) {
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "a", RowID: 0}, {Field: "b", RowID: 1}}, Count: 1},
|
||||
}
|
||||
|
||||
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 previous is last result", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=wa, previous=3), Rows(field=wb, previous=3), Rows(field=wc, previous=3), limit=3)`).Results[0].([]pilosa.GroupCount)
|
||||
if len(results) > 0 {
|
||||
t.Fatalf("expected no results because previous specified last result")
|
||||
}
|
||||
})
|
||||
|
||||
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)
|
||||
})
|
||||
|
||||
// test multiple shards with distinct results (different rows) and same
|
||||
// rows to ensure ordering, limit behavior and correctness
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "ma")
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "mb")
|
||||
c.ImportBits(t, "i", "ma", [][2]uint64{
|
||||
{0, 0},
|
||||
{1, ShardWidth},
|
||||
{2, 0},
|
||||
{3, ShardWidth},
|
||||
})
|
||||
c.ImportBits(t, "i", "mb", [][2]uint64{
|
||||
{0, 0},
|
||||
{1, ShardWidth},
|
||||
{2, 0},
|
||||
{3, ShardWidth},
|
||||
})
|
||||
t.Run("distinct rows in different shards", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=ma), Rows(field=mb), limit=5)`).Results[0].([]pilosa.GroupCount)
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 0}, {Field: "mb", RowID: 0}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 0}, {Field: "mb", RowID: 2}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 1}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 3}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 2}, {Field: "mb", RowID: 0}}, Count: 1},
|
||||
}
|
||||
checkGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
t.Run("distinct rows in different shards with row limit", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=ma), Rows(field=mb, limit=2), limit=5)`).Results[0].([]pilosa.GroupCount)
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 0}, {Field: "mb", RowID: 0}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 1}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 2}, {Field: "mb", RowID: 0}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 3}, {Field: "mb", RowID: 1}}, Count: 1},
|
||||
}
|
||||
checkGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
t.Run("distinct rows in different shards with column arg", func(t *testing.T) {
|
||||
results := c.Query(t, "i", fmt.Sprintf(`GroupBy(Rows(field=ma), Rows(field=mb, column=%d), limit=5)`, ShardWidth)).Results[0].([]pilosa.GroupCount)
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 1}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 1}, {Field: "mb", RowID: 3}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 3}, {Field: "mb", RowID: 1}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "ma", RowID: 3}, {Field: "mb", RowID: 3}}, Count: 1},
|
||||
}
|
||||
checkGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "na")
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "nb")
|
||||
c.ImportBits(t, "i", "na", [][2]uint64{
|
||||
{0, 0},
|
||||
{0, ShardWidth},
|
||||
{1, 0},
|
||||
{1, ShardWidth},
|
||||
})
|
||||
c.ImportBits(t, "i", "nb", [][2]uint64{
|
||||
{0, 0},
|
||||
{0, ShardWidth},
|
||||
{1, 0},
|
||||
{1, ShardWidth},
|
||||
})
|
||||
t.Run("same rows in different shards", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=na), Rows(field=nb))`).Results[0].([]pilosa.GroupCount)
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "na", RowID: 0}, {Field: "nb", RowID: 0}}, Count: 2},
|
||||
{Group: []pilosa.FieldRow{{Field: "na", RowID: 0}, {Field: "nb", RowID: 1}}, Count: 2},
|
||||
{Group: []pilosa.FieldRow{{Field: "na", RowID: 1}, {Field: "nb", RowID: 0}}, Count: 2},
|
||||
{Group: []pilosa.FieldRow{{Field: "na", RowID: 1}, {Field: "nb", RowID: 1}}, Count: 2},
|
||||
}
|
||||
checkGroupBy(t, expected, results)
|
||||
|
||||
})
|
||||
|
||||
// test paging over results using previous. set the same bits in three
|
||||
// fields
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "ppa")
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "ppb")
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "ppc")
|
||||
c.ImportBits(t, "i", "ppa", [][2]uint64{
|
||||
{0, 0},
|
||||
{1, 0},
|
||||
{2, 0},
|
||||
{3, 0}, {3, 91000}, {3, ShardWidth}, {3, ShardWidth * 2}, {3, ShardWidth * 3},
|
||||
})
|
||||
c.ImportBits(t, "i", "ppb", [][2]uint64{
|
||||
{0, 0},
|
||||
{1, 0},
|
||||
{2, 0},
|
||||
{3, 0}, {3, 91000}, {3, ShardWidth}, {3, ShardWidth * 2}, {3, ShardWidth * 3},
|
||||
})
|
||||
c.ImportBits(t, "i", "ppc", [][2]uint64{
|
||||
{0, 0},
|
||||
{1, 0},
|
||||
{2, 0},
|
||||
{3, 0}, {3, 91000}, {3, ShardWidth}, {3, ShardWidth * 2}, {3, ShardWidth * 3},
|
||||
})
|
||||
|
||||
t.Run("test wrapping with previous", func(t *testing.T) {
|
||||
totalResults := make([]pilosa.GroupCount, 0)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=ppa), Rows(field=ppb), Rows(field=ppc), limit=3)`).Results[0].([]pilosa.GroupCount)
|
||||
totalResults = append(totalResults, results...)
|
||||
for len(totalResults) < 64 {
|
||||
lastGroup := results[len(results)-1].Group
|
||||
query := fmt.Sprintf("GroupBy(Rows(field=ppa, previous=%d), Rows(field=ppb, previous=%d), Rows(field=ppc, previous=%d), limit=3)", lastGroup[0].RowID, lastGroup[1].RowID, lastGroup[2].RowID)
|
||||
results = c.Query(t, "i", query).Results[0].([]pilosa.GroupCount)
|
||||
totalResults = append(totalResults, results...)
|
||||
}
|
||||
|
||||
expected := make([]pilosa.GroupCount, 64)
|
||||
for i := 0; i < 64; i++ {
|
||||
expected[i] = pilosa.GroupCount{Group: []pilosa.FieldRow{{Field: "ppa", RowID: uint64(i / 16)}, {Field: "ppb", RowID: uint64((i % 16) / 4)}, {Field: "ppc", RowID: uint64(i % 4)}}, Count: 1}
|
||||
}
|
||||
expected[63].Count = 5
|
||||
|
||||
checkGroupBy(t, expected, totalResults)
|
||||
})
|
||||
}
|
||||
for size := range []int{1, 3} {
|
||||
t.Run(fmt.Sprintf("%d_nodes", size), func(t *testing.T) {
|
||||
groupByTest(t, size)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func BenchmarkGroupBy(b *testing.B) {
|
||||
c := test.MustRunCluster(b, 1)
|
||||
defer c.Close()
|
||||
c.CreateField(b, "i", pilosa.IndexOptions{}, "a")
|
||||
c.CreateField(b, "i", pilosa.IndexOptions{}, "b")
|
||||
c.CreateField(b, "i", pilosa.IndexOptions{}, "c")
|
||||
// Set up identical representative data in 3 fields. In each row, we'll set
|
||||
// a certain bit pattern for 100 bits, then skip 1000 up to ShardWidth.
|
||||
bits := make([][2]uint64, 0)
|
||||
for i := uint64(0); i < ShardWidth; i++ {
|
||||
// row 0 has 100 bit runs
|
||||
bits = append(bits, [2]uint64{0, i})
|
||||
if i%2 == 1 {
|
||||
// row 1 has odd bits set
|
||||
bits = append(bits, [2]uint64{1, i})
|
||||
}
|
||||
if i%2 == 0 {
|
||||
// row 2 has even bits set
|
||||
bits = append(bits, [2]uint64{2, i})
|
||||
}
|
||||
if i%27 == 0 {
|
||||
// row 3 has every 27th bit set
|
||||
bits = append(bits, [2]uint64{3, i})
|
||||
}
|
||||
if i%100 == 99 {
|
||||
i += 1000
|
||||
}
|
||||
}
|
||||
c.ImportBits(b, "i", "a", bits)
|
||||
c.ImportBits(b, "i", "b", bits)
|
||||
c.ImportBits(b, "i", "c", bits)
|
||||
|
||||
b.Run("single shard group by", func(b *testing.B) {
|
||||
b.ResetTimer()
|
||||
b.ReportAllocs()
|
||||
for i := 0; i < b.N; i++ {
|
||||
c.Query(b, "i", `GroupBy(Rows(field=a), Rows(field=b), Rows(field=c))`)
|
||||
}
|
||||
})
|
||||
|
||||
b.Run("single shard with limit", func(b *testing.B) {
|
||||
b.ResetTimer()
|
||||
b.ReportAllocs()
|
||||
for i := 0; i < b.N; i++ {
|
||||
c.Query(b, "i", `GroupBy(Rows(field=a), Rows(field=b), Rows(field=c), limit=4)`)
|
||||
}
|
||||
})
|
||||
|
||||
// TODO benchmark over multiple shards
|
||||
|
||||
// TODO benchmark paging over large numbers of rows
|
||||
|
||||
}
|
||||
|
||||
func checkGroupBy(t *testing.T, expected, results []pilosa.GroupCount) {
|
||||
if len(results) != len(expected) {
|
||||
t.Fatalf("number of groupings mismatch:\n got:%+v\nwant:%+v\n", results, expected)
|
||||
}
|
||||
for i, result := range results {
|
||||
if !reflect.DeepEqual(expected[i], result) {
|
||||
t.Fatalf("unexpected result at %d: \n got:%+v\nwant:%+v\n", i, result, expected[i])
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func runCallTest(t *testing.T, writeQuery string, readQueries []string, indexOptions *pilosa.IndexOptions, fieldOption ...pilosa.FieldOption) []pilosa.QueryResponse {
|
||||
if indexOptions == nil {
|
||||
indexOptions = &pilosa.IndexOptions{}
|
||||
|
|
|
|||
170
fragment.go
170
fragment.go
|
|
@ -58,6 +58,9 @@ const (
|
|||
// exponent.
|
||||
shardVsContainerExponent = shardWidthExponent - 16
|
||||
|
||||
// width of roaring containers is 2^16
|
||||
containerWidth = 1 << 16
|
||||
|
||||
// snapshotExt is the file extension used for an in-process snapshot.
|
||||
snapshotExt = ".snapshotting"
|
||||
|
||||
|
|
@ -1988,15 +1991,82 @@ func (f *fragment) readCacheFromArchive(r io.Reader) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
func (f *fragment) rows() []uint64 {
|
||||
i, _ := f.storage.Containers.Iterator(0)
|
||||
rows := make([]uint64, 0)
|
||||
// rowFilter is a function signature for controlling iteration over containers
|
||||
// in a fragment. It will be invoked on each container found and returns two
|
||||
// booleans. The first is whether the row this container is in should be
|
||||
// included or skipped, and the second is whether to stop processing or
|
||||
// continue.
|
||||
type rowFilter func(rowID, key uint64, c *roaring.Container) (include, done bool)
|
||||
|
||||
// filterWithLimit returns a filter which will only allow a limited number of
|
||||
// rows to be returned. It should be applied last so that it is only called (and
|
||||
// therefore only updates its internal state) if the row is being included by
|
||||
// every other filter.
|
||||
func filterWithLimit(limit uint64) rowFilter {
|
||||
return func(rowID, key uint64, c *roaring.Container) (include, done bool) {
|
||||
if limit > 0 {
|
||||
limit--
|
||||
return true, false
|
||||
}
|
||||
return false, true
|
||||
}
|
||||
}
|
||||
|
||||
func filterColumn(col uint64) rowFilter {
|
||||
return func(rowID, key uint64, c *roaring.Container) (include, done bool) {
|
||||
colID := col % ShardWidth
|
||||
colKey := ((rowID * ShardWidth) + colID) >> 16
|
||||
colVal := uint16(colID & 0xFFFF) // columnID within the container
|
||||
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. The container iterator would need to expose a seek
|
||||
// method, and the rowFilter would need some way of communicating to
|
||||
// fragment.rows what the next rowID to seek to is.
|
||||
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
|
||||
}
|
||||
}
|
||||
|
||||
// rows returns all rows starting from 'start'. Filters will be applied in
|
||||
// order. All filters must return true to include the row. Once a row is
|
||||
// included, further containers in that row will be skipped. So, for a row to be
|
||||
// included, there must be one container in that row where all filters return
|
||||
// true. For a row to be skipped, at least one filter must return false for each
|
||||
// container in that row (it need not be the same filter for each). Any filter
|
||||
// returning done == true will cause processing to stop after all filters for
|
||||
// this container have been processed. The rows accumulated up to this point
|
||||
// (including this row if all filters passed) will be returned.
|
||||
func (f *fragment) rows(start uint64, filters ...rowFilter) []uint64 {
|
||||
startKey := rowToKey(start)
|
||||
i, _ := f.storage.Containers.Iterator(startKey)
|
||||
rows := make([]uint64, 0)
|
||||
var lastRow uint64 = math.MaxUint64
|
||||
|
||||
// Loop over the existing containers.
|
||||
for i.Next() {
|
||||
key, _ := i.Value()
|
||||
key, c := i.Value()
|
||||
|
||||
// virtual row for the current container
|
||||
vRow := key >> shardVsContainerExponent
|
||||
|
|
@ -2006,44 +2076,63 @@ func (f *fragment) rows() []uint64 {
|
|||
continue
|
||||
}
|
||||
|
||||
rows = append(rows, vRow)
|
||||
lastRow = vRow
|
||||
}
|
||||
return rows
|
||||
|
||||
}
|
||||
|
||||
func (f *fragment) rowsForColumn(columnID uint64) []uint64 {
|
||||
var colKey uint64
|
||||
|
||||
colID := columnID % ShardWidth
|
||||
i, _ := f.storage.Containers.Iterator(0)
|
||||
|
||||
colVal := uint16(colID & 0xFFFF)
|
||||
|
||||
rows := make([]uint64, 0)
|
||||
|
||||
// Loop over the existing containers.
|
||||
for i.Next() {
|
||||
key, c := i.Value()
|
||||
|
||||
// virtual row for the current container
|
||||
vRow := key >> shardVsContainerExponent
|
||||
|
||||
// column container key for virtual row
|
||||
colKey = ((vRow * ShardWidth) + colID) >> 16
|
||||
|
||||
if colKey != key {
|
||||
continue
|
||||
// apply filters
|
||||
addRow, done := true, false
|
||||
for _, filter := range filters {
|
||||
var d bool
|
||||
addRow, d = filter(vRow, key, c)
|
||||
done = done || d
|
||||
if !addRow {
|
||||
break
|
||||
}
|
||||
}
|
||||
|
||||
if c.Contains(colVal) {
|
||||
if addRow {
|
||||
lastRow = vRow
|
||||
rows = append(rows, vRow)
|
||||
}
|
||||
if done {
|
||||
return rows
|
||||
}
|
||||
}
|
||||
return rows
|
||||
}
|
||||
|
||||
type rowIterator struct {
|
||||
f *fragment
|
||||
rowIDs []uint64
|
||||
cur int
|
||||
wrap bool
|
||||
}
|
||||
|
||||
func (f *fragment) rowIterator(wrap bool, filters ...rowFilter) *rowIterator {
|
||||
return &rowIterator{
|
||||
f: f,
|
||||
rowIDs: f.rows(0, filters...), // TODO: this may be memory intensive in high cardinality cases
|
||||
wrap: wrap,
|
||||
}
|
||||
}
|
||||
|
||||
func (ri *rowIterator) Seek(rowID uint64) {
|
||||
idx := sort.Search(len(ri.rowIDs), func(i int) bool {
|
||||
return ri.rowIDs[i] >= rowID
|
||||
})
|
||||
ri.cur = idx
|
||||
}
|
||||
|
||||
func (ri *rowIterator) Next() (r *Row, rowID uint64, wrapped bool) {
|
||||
if ri.cur >= len(ri.rowIDs) {
|
||||
if !ri.wrap || len(ri.rowIDs) == 0 {
|
||||
return nil, 0, true
|
||||
}
|
||||
ri.Seek(0)
|
||||
wrapped = true
|
||||
}
|
||||
rowID = ri.rowIDs[ri.cur]
|
||||
r = ri.f.row(rowID)
|
||||
ri.cur += 1
|
||||
return r, rowID, wrapped
|
||||
}
|
||||
|
||||
// FragmentBlock represents info about a subsection of the rows in a block.
|
||||
// This is used for comparing data in remote blocks for active anti-entropy.
|
||||
type FragmentBlock struct {
|
||||
|
|
@ -2323,7 +2412,7 @@ func newRowsVector(f *fragment) *rowsVector {
|
|||
// Additionally, it returns true if a value was found,
|
||||
// otherwise it returns false.
|
||||
func (v *rowsVector) Get(colID uint64) (uint64, bool, error) {
|
||||
rows := v.f.rowsForColumn(colID)
|
||||
rows := v.f.rows(0, filterColumn(colID))
|
||||
if len(rows) > 1 {
|
||||
return 0, false, errors.New("found multiple row values for column")
|
||||
} else if len(rows) == 1 {
|
||||
|
|
@ -2332,6 +2421,13 @@ func (v *rowsVector) Get(colID uint64) (uint64, bool, error) {
|
|||
return 0, false, nil
|
||||
}
|
||||
|
||||
// rowToKey converts a Pilosa row ID to the key of the container which starts
|
||||
// that row in the bitmap which represents this entire fragment. A fragment is
|
||||
// all the rows within a shard within a field concatenated together.
|
||||
func rowToKey(rowID uint64) (key uint64) {
|
||||
return rowID * (ShardWidth / containerWidth)
|
||||
}
|
||||
|
||||
// boolVector implements the vector interface by looking
|
||||
// at data in rows 0 and 1.
|
||||
type boolVector struct {
|
||||
|
|
@ -2349,7 +2445,7 @@ func newBoolVector(f *fragment) *boolVector {
|
|||
// Additionally, it returns true if a value was found,
|
||||
// otherwise it returns false.
|
||||
func (v *boolVector) Get(colID uint64) (uint64, bool, error) {
|
||||
rows := v.f.rowsForColumn(colID)
|
||||
rows := v.f.rows(0, filterColumn(colID))
|
||||
if len(rows) > 1 {
|
||||
return 0, false, errors.New("found multiple row values for column")
|
||||
} else if len(rows) == 1 {
|
||||
|
|
|
|||
|
|
@ -1808,12 +1808,12 @@ func TestFragment_RowsIteration(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
ids := f.rows()
|
||||
ids := f.rows(0)
|
||||
if !reflect.DeepEqual(expectedAll, ids) {
|
||||
t.Fatalf("Do not match %v %v", expectedAll, ids)
|
||||
}
|
||||
|
||||
ids = f.rowsForColumn(1)
|
||||
ids = f.rows(0, filterColumn(1))
|
||||
if !reflect.DeepEqual(expectedOdd, ids) {
|
||||
t.Fatalf("Do not match %v %v", expectedOdd, ids)
|
||||
}
|
||||
|
|
@ -1832,12 +1832,12 @@ func TestFragment_RowsIteration(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
ids := f.rows()
|
||||
ids := f.rows(0)
|
||||
if !reflect.DeepEqual(expected, ids) {
|
||||
t.Fatalf("Do not match %v %v", expected, ids)
|
||||
}
|
||||
|
||||
ids = f.rowsForColumn(66000)
|
||||
ids = f.rows(0, filterColumn(66000))
|
||||
if !reflect.DeepEqual(expected, ids) {
|
||||
t.Fatalf("Do not match %v %v", expected, ids)
|
||||
}
|
||||
|
|
@ -1855,11 +1855,11 @@ func TestFragment_RowsIteration(t *testing.T) {
|
|||
t.Fatal(err)
|
||||
}
|
||||
|
||||
ids := f.rows()
|
||||
ids := f.rows(0)
|
||||
if !reflect.DeepEqual(expectedRows, ids) {
|
||||
t.Fatalf("Do not match %v %v", expectedRows, ids)
|
||||
}
|
||||
ids = f.rowsForColumn(c)
|
||||
ids = f.rows(0, filterColumn(c))
|
||||
if !reflect.DeepEqual(expectedRows, ids) {
|
||||
t.Fatalf("Do not match %v %v", expectedRows, ids)
|
||||
}
|
||||
|
|
@ -2065,3 +2065,122 @@ func calcExpected(inputs ...[]uint64) [][]uint64 {
|
|||
|
||||
return ret
|
||||
}
|
||||
|
||||
func TestFragmentRowIterator(t *testing.T) {
|
||||
t.Run("basic", func(t *testing.T) {
|
||||
f := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked)
|
||||
f.mustSetBits(0, 0)
|
||||
f.mustSetBits(1, 0)
|
||||
f.mustSetBits(2, 0)
|
||||
f.mustSetBits(3, 0)
|
||||
|
||||
iter := f.rowIterator(false)
|
||||
for i := uint64(0); i < 4; i++ {
|
||||
row, id, wrapped := iter.Next()
|
||||
if id != i {
|
||||
t.Fatalf("expected row %d but got %d", i, id)
|
||||
}
|
||||
if wrapped {
|
||||
t.Fatalf("shouldn't have wrapped")
|
||||
}
|
||||
if !reflect.DeepEqual(row.Columns(), []uint64{0}) {
|
||||
t.Fatalf("got wrong columns back on iteration %d - should just be 0 but %v", i, row.Columns())
|
||||
}
|
||||
}
|
||||
row, id, wrapped := iter.Next()
|
||||
if row != nil {
|
||||
t.Fatalf("row should be nil after iterator is exhausted, got %v", row.Columns())
|
||||
}
|
||||
if id != 0 {
|
||||
t.Fatalf("id should be 0 after iterator is exhausted, got %d", id)
|
||||
}
|
||||
if !wrapped {
|
||||
t.Fatalf("wrapped should be true after iterator is exhausted")
|
||||
}
|
||||
f.Close()
|
||||
})
|
||||
|
||||
t.Run("skipped rows", func(t *testing.T) {
|
||||
f := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked)
|
||||
f.mustSetBits(1, 0)
|
||||
f.mustSetBits(3, 0)
|
||||
f.mustSetBits(5, 0)
|
||||
f.mustSetBits(7, 0)
|
||||
|
||||
iter := f.rowIterator(false)
|
||||
for i := uint64(1); i < 8; i += 2 {
|
||||
row, id, wrapped := iter.Next()
|
||||
if id != i {
|
||||
t.Fatalf("expected row %d but got %d", i, id)
|
||||
}
|
||||
if wrapped {
|
||||
t.Fatalf("shouldn't have wrapped")
|
||||
}
|
||||
if !reflect.DeepEqual(row.Columns(), []uint64{0}) {
|
||||
t.Fatalf("got wrong columns back on iteration %d - should just be 0 but %v", i, row.Columns())
|
||||
}
|
||||
}
|
||||
row, id, wrapped := iter.Next()
|
||||
if row != nil {
|
||||
t.Fatalf("row should be nil after iterator is exhausted, got %v", row.Columns())
|
||||
}
|
||||
if id != 0 {
|
||||
t.Fatalf("id should be 0 after iterator is exhausted, got %d", id)
|
||||
}
|
||||
if !wrapped {
|
||||
t.Fatalf("wrapped should be true after iterator is exhausted")
|
||||
}
|
||||
f.Close()
|
||||
})
|
||||
|
||||
t.Run("basic wrapped", func(t *testing.T) {
|
||||
f := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked)
|
||||
f.mustSetBits(0, 0)
|
||||
f.mustSetBits(1, 0)
|
||||
f.mustSetBits(2, 0)
|
||||
f.mustSetBits(3, 0)
|
||||
|
||||
iter := f.rowIterator(true)
|
||||
for i := uint64(0); i < 5; i++ {
|
||||
row, id, wrapped := iter.Next()
|
||||
if id != i%4 {
|
||||
t.Fatalf("expected row %d but got %d", i%4, id)
|
||||
}
|
||||
if wrapped && i < 4 {
|
||||
t.Fatalf("shouldn't have wrapped")
|
||||
} else if !wrapped && i >= 4 {
|
||||
t.Fatalf("should have wrapped")
|
||||
}
|
||||
if !reflect.DeepEqual(row.Columns(), []uint64{0}) {
|
||||
t.Fatalf("got wrong columns back on iteration %d - should just be 0 but %v", i, row.Columns())
|
||||
}
|
||||
}
|
||||
f.Close()
|
||||
})
|
||||
|
||||
t.Run("skipped rows wrapped", func(t *testing.T) {
|
||||
f := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked)
|
||||
f.mustSetBits(1, 0)
|
||||
f.mustSetBits(3, 0)
|
||||
f.mustSetBits(5, 0)
|
||||
f.mustSetBits(7, 0)
|
||||
|
||||
iter := f.rowIterator(true)
|
||||
for i := uint64(1); i < 10; i += 2 {
|
||||
row, id, wrapped := iter.Next()
|
||||
if id != i%8 {
|
||||
t.Errorf("expected row %d but got %d", i%8, id)
|
||||
}
|
||||
if wrapped && i < 8 {
|
||||
t.Errorf("shouldn't have wrapped")
|
||||
} else if !wrapped && i >= 8 {
|
||||
t.Errorf("should have wrapped")
|
||||
}
|
||||
if !reflect.DeepEqual(row.Columns(), []uint64{0}) {
|
||||
t.Fatalf("got wrong columns back on iteration %d - should just be 0 but %v", i, row.Columns())
|
||||
}
|
||||
}
|
||||
f.Close()
|
||||
})
|
||||
|
||||
}
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -8,12 +8,27 @@ message Row {
|
|||
repeated Attr Attrs = 2;
|
||||
}
|
||||
|
||||
message RowIdentifiers {
|
||||
repeated uint64 Rows = 1;
|
||||
repeated string Keys = 2;
|
||||
}
|
||||
|
||||
message Pair {
|
||||
uint64 ID = 1;
|
||||
string Key = 3;
|
||||
uint64 Count = 2;
|
||||
}
|
||||
|
||||
message FieldRow{
|
||||
string Field = 1;
|
||||
uint64 RowID = 2;
|
||||
}
|
||||
|
||||
message GroupCount{
|
||||
repeated FieldRow Group = 1;
|
||||
uint64 Count = 2;
|
||||
}
|
||||
|
||||
message ValCount {
|
||||
int64 Val = 1;
|
||||
int64 Count = 2;
|
||||
|
|
@ -64,8 +79,11 @@ message QueryResult {
|
|||
Row Row = 1;
|
||||
uint64 N = 2;
|
||||
repeated Pair Pairs = 3;
|
||||
ValCount ValCount = 5;
|
||||
bool Changed = 4;
|
||||
ValCount ValCount = 5;
|
||||
repeated uint64 RowIDs = 7;
|
||||
repeated GroupCount GroupCounts = 8;
|
||||
RowIdentifiers RowIdentifiers = 9;
|
||||
}
|
||||
|
||||
message ImportRequest {
|
||||
|
|
|
|||
|
|
@ -62,7 +62,9 @@ var (
|
|||
ErrNodeNotCoordinator = errors.New("node is not the coordinator")
|
||||
ErrResizeNotRunning = errors.New("no resize job currently running")
|
||||
|
||||
ErrNotImplemented = errors.New("not implemented")
|
||||
ErrNotImplemented = errors.New("not implemented")
|
||||
ErrFieldsArgumentRequired = errors.New("fields argument required")
|
||||
ErrExpectedFieldListArgument = errors.New("expected field list argument")
|
||||
)
|
||||
|
||||
// apiMethodNotAllowedError wraps an error value indicating that a particular
|
||||
|
|
|
|||
|
|
@ -197,6 +197,79 @@ func (m *Command) RecalculateCaches() error {
|
|||
// Cluster represents a Pilosa cluster (multiple Command instances)
|
||||
type Cluster []*Command
|
||||
|
||||
// Query executes an API.Query through one of the cluster's node's API. It fails
|
||||
// the test if there is an error.
|
||||
func (c Cluster) Query(t testing.TB, index, query string) pilosa.QueryResponse {
|
||||
if len(c) == 0 {
|
||||
t.Fatal("must have at least one node in cluster to query")
|
||||
}
|
||||
|
||||
return c[0].MustQuery(t, &pilosa.QueryRequest{Index: index, Query: query})
|
||||
}
|
||||
|
||||
func (c Cluster) ImportBits(t testing.TB, index, field string, rowcols [][2]uint64) {
|
||||
byShard := make(map[uint64][][2]uint64)
|
||||
for _, rowcol := range rowcols {
|
||||
shard := rowcol[1] / pilosa.ShardWidth
|
||||
byShard[shard] = append(byShard[shard], rowcol)
|
||||
}
|
||||
|
||||
for shard, bits := range byShard {
|
||||
rowIDs := make([]uint64, len(bits))
|
||||
colIDs := make([]uint64, len(bits))
|
||||
for i, bit := range bits {
|
||||
rowIDs[i] = bit[0]
|
||||
colIDs[i] = bit[1]
|
||||
}
|
||||
nodes, err := c[0].API.ShardNodes(context.Background(), index, shard)
|
||||
if err != nil {
|
||||
t.Fatalf("getting shard nodes: %v", err)
|
||||
}
|
||||
// TODO won't be necessary to do all nodes once that works hits
|
||||
for _, node := range nodes {
|
||||
for _, com := range c {
|
||||
if com.API.Node().ID != node.ID {
|
||||
continue
|
||||
}
|
||||
err := com.API.Import(context.Background(), &pilosa.ImportRequest{
|
||||
Index: index,
|
||||
Field: field,
|
||||
Shard: shard,
|
||||
RowIDs: rowIDs,
|
||||
ColumnIDs: colIDs,
|
||||
})
|
||||
if err != nil {
|
||||
t.Fatalf("importing data: %v", err)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// CreateField creates the index (if necessary) and field specified.
|
||||
func (c Cluster) CreateField(t testing.TB, index string, iopts pilosa.IndexOptions, field string, fopts ...pilosa.FieldOption) *pilosa.Field {
|
||||
idx, err := c[0].API.CreateIndex(context.Background(), index, iopts)
|
||||
if err != nil && !strings.Contains(err.Error(), "index already exists") {
|
||||
t.Fatalf("creating index: %v", err)
|
||||
} else if err != nil { // index exists
|
||||
idx, err = c[0].API.Index(context.Background(), index)
|
||||
if err != nil {
|
||||
t.Fatalf("getting index: %v", err)
|
||||
}
|
||||
}
|
||||
if idx.Options() != iopts {
|
||||
t.Logf("existing index options:\n%v\ndon't match given opts:\n%v\n in pilosa/test.Cluster.CreateField", idx.Options(), iopts)
|
||||
}
|
||||
|
||||
f, err := c[0].API.CreateField(context.Background(), index, field, fopts...)
|
||||
// we'll assume the field doesn't exist because checking if the options
|
||||
// match seems painful.
|
||||
if err != nil {
|
||||
t.Fatalf("creating field: %v", err)
|
||||
}
|
||||
return f
|
||||
}
|
||||
|
||||
// Start runs a Cluster
|
||||
func (c Cluster) Start() error {
|
||||
var gossipSeeds = make([]string, len(c))
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue