diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 677f08ce0..e8de4ea1a 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -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 { diff --git a/executor.go b/executor.go index 75b4326fe..c4153ab61 100644 --- a/executor.go +++ b/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 +} diff --git a/executor_internal_test.go b/executor_internal_test.go new file mode 100644 index 000000000..629d704db --- /dev/null +++ b/executor_internal_test.go @@ -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<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{} diff --git a/fragment.go b/fragment.go index 10e5e3920..dd1d874de 100644 --- a/fragment.go +++ b/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 { diff --git a/fragment_internal_test.go b/fragment_internal_test.go index ac080b5b7..4e3007957 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -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() + }) + +} diff --git a/internal/public.pb.go b/internal/public.pb.go index cceea4806..bfa0b3be1 100644 --- a/internal/public.pb.go +++ b/internal/public.pb.go @@ -9,7 +9,10 @@ It has these top-level messages: Row + RowIdentifiers Pair + FieldRow + GroupCount ValCount Bit ColumnAttrSet @@ -74,6 +77,30 @@ func (m *Row) GetAttrs() []*Attr { return nil } +type RowIdentifiers struct { + Rows []uint64 `protobuf:"varint,1,rep,packed,name=Rows" json:"Rows,omitempty"` + Keys []string `protobuf:"bytes,2,rep,name=Keys" json:"Keys,omitempty"` +} + +func (m *RowIdentifiers) Reset() { *m = RowIdentifiers{} } +func (m *RowIdentifiers) String() string { return proto.CompactTextString(m) } +func (*RowIdentifiers) ProtoMessage() {} +func (*RowIdentifiers) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{1} } + +func (m *RowIdentifiers) GetRows() []uint64 { + if m != nil { + return m.Rows + } + return nil +} + +func (m *RowIdentifiers) GetKeys() []string { + if m != nil { + return m.Keys + } + return nil +} + type Pair struct { ID uint64 `protobuf:"varint,1,opt,name=ID,proto3" json:"ID,omitempty"` Key string `protobuf:"bytes,3,opt,name=Key,proto3" json:"Key,omitempty"` @@ -83,7 +110,7 @@ type Pair struct { func (m *Pair) Reset() { *m = Pair{} } func (m *Pair) String() string { return proto.CompactTextString(m) } func (*Pair) ProtoMessage() {} -func (*Pair) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{1} } +func (*Pair) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{2} } func (m *Pair) GetID() uint64 { if m != nil { @@ -106,6 +133,54 @@ func (m *Pair) GetCount() uint64 { return 0 } +type FieldRow struct { + Field string `protobuf:"bytes,1,opt,name=Field,proto3" json:"Field,omitempty"` + RowID uint64 `protobuf:"varint,2,opt,name=RowID,proto3" json:"RowID,omitempty"` +} + +func (m *FieldRow) Reset() { *m = FieldRow{} } +func (m *FieldRow) String() string { return proto.CompactTextString(m) } +func (*FieldRow) ProtoMessage() {} +func (*FieldRow) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{3} } + +func (m *FieldRow) GetField() string { + if m != nil { + return m.Field + } + return "" +} + +func (m *FieldRow) GetRowID() uint64 { + if m != nil { + return m.RowID + } + return 0 +} + +type GroupCount struct { + Group []*FieldRow `protobuf:"bytes,1,rep,name=Group" json:"Group,omitempty"` + Count uint64 `protobuf:"varint,2,opt,name=Count,proto3" json:"Count,omitempty"` +} + +func (m *GroupCount) Reset() { *m = GroupCount{} } +func (m *GroupCount) String() string { return proto.CompactTextString(m) } +func (*GroupCount) ProtoMessage() {} +func (*GroupCount) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{4} } + +func (m *GroupCount) GetGroup() []*FieldRow { + if m != nil { + return m.Group + } + return nil +} + +func (m *GroupCount) GetCount() uint64 { + if m != nil { + return m.Count + } + return 0 +} + type ValCount struct { Val int64 `protobuf:"varint,1,opt,name=Val,proto3" json:"Val,omitempty"` Count int64 `protobuf:"varint,2,opt,name=Count,proto3" json:"Count,omitempty"` @@ -114,7 +189,7 @@ type ValCount struct { func (m *ValCount) Reset() { *m = ValCount{} } func (m *ValCount) String() string { return proto.CompactTextString(m) } func (*ValCount) ProtoMessage() {} -func (*ValCount) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{2} } +func (*ValCount) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{5} } func (m *ValCount) GetVal() int64 { if m != nil { @@ -139,7 +214,7 @@ type Bit struct { func (m *Bit) Reset() { *m = Bit{} } func (m *Bit) String() string { return proto.CompactTextString(m) } func (*Bit) ProtoMessage() {} -func (*Bit) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{3} } +func (*Bit) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{6} } func (m *Bit) GetRowID() uint64 { if m != nil { @@ -171,7 +246,7 @@ type ColumnAttrSet struct { func (m *ColumnAttrSet) Reset() { *m = ColumnAttrSet{} } func (m *ColumnAttrSet) String() string { return proto.CompactTextString(m) } func (*ColumnAttrSet) ProtoMessage() {} -func (*ColumnAttrSet) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{4} } +func (*ColumnAttrSet) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{7} } func (m *ColumnAttrSet) GetID() uint64 { if m != nil { @@ -206,7 +281,7 @@ type Attr struct { func (m *Attr) Reset() { *m = Attr{} } func (m *Attr) String() string { return proto.CompactTextString(m) } func (*Attr) ProtoMessage() {} -func (*Attr) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{5} } +func (*Attr) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{8} } func (m *Attr) GetKey() string { if m != nil { @@ -257,7 +332,7 @@ type AttrMap struct { func (m *AttrMap) Reset() { *m = AttrMap{} } func (m *AttrMap) String() string { return proto.CompactTextString(m) } func (*AttrMap) ProtoMessage() {} -func (*AttrMap) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{6} } +func (*AttrMap) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{9} } func (m *AttrMap) GetAttrs() []*Attr { if m != nil { @@ -278,7 +353,7 @@ type QueryRequest struct { func (m *QueryRequest) Reset() { *m = QueryRequest{} } func (m *QueryRequest) String() string { return proto.CompactTextString(m) } func (*QueryRequest) ProtoMessage() {} -func (*QueryRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{7} } +func (*QueryRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{10} } func (m *QueryRequest) GetQuery() string { if m != nil { @@ -331,7 +406,7 @@ type QueryResponse struct { func (m *QueryResponse) Reset() { *m = QueryResponse{} } func (m *QueryResponse) String() string { return proto.CompactTextString(m) } func (*QueryResponse) ProtoMessage() {} -func (*QueryResponse) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{8} } +func (*QueryResponse) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{11} } func (m *QueryResponse) GetErr() string { if m != nil { @@ -355,18 +430,21 @@ func (m *QueryResponse) GetColumnAttrSets() []*ColumnAttrSet { } type QueryResult struct { - Type uint32 `protobuf:"varint,6,opt,name=Type,proto3" json:"Type,omitempty"` - Row *Row `protobuf:"bytes,1,opt,name=Row" json:"Row,omitempty"` - N uint64 `protobuf:"varint,2,opt,name=N,proto3" json:"N,omitempty"` - Pairs []*Pair `protobuf:"bytes,3,rep,name=Pairs" json:"Pairs,omitempty"` - ValCount *ValCount `protobuf:"bytes,5,opt,name=ValCount" json:"ValCount,omitempty"` - Changed bool `protobuf:"varint,4,opt,name=Changed,proto3" json:"Changed,omitempty"` + Type uint32 `protobuf:"varint,6,opt,name=Type,proto3" json:"Type,omitempty"` + Row *Row `protobuf:"bytes,1,opt,name=Row" json:"Row,omitempty"` + N uint64 `protobuf:"varint,2,opt,name=N,proto3" json:"N,omitempty"` + Pairs []*Pair `protobuf:"bytes,3,rep,name=Pairs" json:"Pairs,omitempty"` + Changed bool `protobuf:"varint,4,opt,name=Changed,proto3" json:"Changed,omitempty"` + ValCount *ValCount `protobuf:"bytes,5,opt,name=ValCount" json:"ValCount,omitempty"` + RowIDs []uint64 `protobuf:"varint,7,rep,packed,name=RowIDs" json:"RowIDs,omitempty"` + GroupCounts []*GroupCount `protobuf:"bytes,8,rep,name=GroupCounts" json:"GroupCounts,omitempty"` + RowIdentifiers *RowIdentifiers `protobuf:"bytes,9,opt,name=RowIdentifiers" json:"RowIdentifiers,omitempty"` } func (m *QueryResult) Reset() { *m = QueryResult{} } func (m *QueryResult) String() string { return proto.CompactTextString(m) } func (*QueryResult) ProtoMessage() {} -func (*QueryResult) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{9} } +func (*QueryResult) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{12} } func (m *QueryResult) GetType() uint32 { if m != nil { @@ -396,6 +474,13 @@ func (m *QueryResult) GetPairs() []*Pair { return nil } +func (m *QueryResult) GetChanged() bool { + if m != nil { + return m.Changed + } + return false +} + func (m *QueryResult) GetValCount() *ValCount { if m != nil { return m.ValCount @@ -403,11 +488,25 @@ func (m *QueryResult) GetValCount() *ValCount { return nil } -func (m *QueryResult) GetChanged() bool { +func (m *QueryResult) GetRowIDs() []uint64 { if m != nil { - return m.Changed + return m.RowIDs } - return false + return nil +} + +func (m *QueryResult) GetGroupCounts() []*GroupCount { + if m != nil { + return m.GroupCounts + } + return nil +} + +func (m *QueryResult) GetRowIdentifiers() *RowIdentifiers { + if m != nil { + return m.RowIdentifiers + } + return nil } type ImportRequest struct { @@ -424,7 +523,7 @@ type ImportRequest struct { func (m *ImportRequest) Reset() { *m = ImportRequest{} } func (m *ImportRequest) String() string { return proto.CompactTextString(m) } func (*ImportRequest) ProtoMessage() {} -func (*ImportRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{10} } +func (*ImportRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{13} } func (m *ImportRequest) GetIndex() string { if m != nil { @@ -494,7 +593,7 @@ type ImportValueRequest struct { func (m *ImportValueRequest) Reset() { *m = ImportValueRequest{} } func (m *ImportValueRequest) String() string { return proto.CompactTextString(m) } func (*ImportValueRequest) ProtoMessage() {} -func (*ImportValueRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{11} } +func (*ImportValueRequest) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{14} } func (m *ImportValueRequest) GetIndex() string { if m != nil { @@ -540,7 +639,10 @@ func (m *ImportValueRequest) GetValues() []int64 { func init() { proto.RegisterType((*Row)(nil), "internal.Row") + proto.RegisterType((*RowIdentifiers)(nil), "internal.RowIdentifiers") proto.RegisterType((*Pair)(nil), "internal.Pair") + proto.RegisterType((*FieldRow)(nil), "internal.FieldRow") + proto.RegisterType((*GroupCount)(nil), "internal.GroupCount") proto.RegisterType((*ValCount)(nil), "internal.ValCount") proto.RegisterType((*Bit)(nil), "internal.Bit") proto.RegisterType((*ColumnAttrSet)(nil), "internal.ColumnAttrSet") @@ -614,6 +716,56 @@ func (m *Row) MarshalTo(dAtA []byte) (int, error) { return i, nil } +func (m *RowIdentifiers) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *RowIdentifiers) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Rows) > 0 { + dAtA4 := make([]byte, len(m.Rows)*10) + var j3 int + for _, num := range m.Rows { + for num >= 1<<7 { + dAtA4[j3] = uint8(uint64(num)&0x7f | 0x80) + num >>= 7 + j3++ + } + dAtA4[j3] = uint8(num) + j3++ + } + dAtA[i] = 0xa + i++ + i = encodeVarintPublic(dAtA, i, uint64(j3)) + i += copy(dAtA[i:], dAtA4[:j3]) + } + if len(m.Keys) > 0 { + for _, s := range m.Keys { + dAtA[i] = 0x12 + i++ + l = len(s) + for l >= 1<<7 { + dAtA[i] = uint8(uint64(l)&0x7f | 0x80) + l >>= 7 + i++ + } + dAtA[i] = uint8(l) + i++ + i += copy(dAtA[i:], s) + } + } + return i, nil +} + func (m *Pair) Marshal() (dAtA []byte, err error) { size := m.Size() dAtA = make([]byte, size) @@ -648,6 +800,70 @@ func (m *Pair) MarshalTo(dAtA []byte) (int, error) { return i, nil } +func (m *FieldRow) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *FieldRow) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Field) > 0 { + dAtA[i] = 0xa + i++ + i = encodeVarintPublic(dAtA, i, uint64(len(m.Field))) + i += copy(dAtA[i:], m.Field) + } + if m.RowID != 0 { + dAtA[i] = 0x10 + i++ + i = encodeVarintPublic(dAtA, i, uint64(m.RowID)) + } + return i, nil +} + +func (m *GroupCount) Marshal() (dAtA []byte, err error) { + size := m.Size() + dAtA = make([]byte, size) + n, err := m.MarshalTo(dAtA) + if err != nil { + return nil, err + } + return dAtA[:n], nil +} + +func (m *GroupCount) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Group) > 0 { + for _, msg := range m.Group { + dAtA[i] = 0xa + i++ + i = encodeVarintPublic(dAtA, i, uint64(msg.Size())) + n, err := msg.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n + } + } + if m.Count != 0 { + dAtA[i] = 0x10 + i++ + i = encodeVarintPublic(dAtA, i, uint64(m.Count)) + } + return i, nil +} + func (m *ValCount) Marshal() (dAtA []byte, err error) { size := m.Size() dAtA = make([]byte, size) @@ -858,21 +1074,21 @@ func (m *QueryRequest) MarshalTo(dAtA []byte) (int, error) { i += copy(dAtA[i:], m.Query) } if len(m.Shards) > 0 { - dAtA4 := make([]byte, len(m.Shards)*10) - var j3 int + dAtA6 := make([]byte, len(m.Shards)*10) + var j5 int for _, num := range m.Shards { for num >= 1<<7 { - dAtA4[j3] = uint8(uint64(num)&0x7f | 0x80) + dAtA6[j5] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 - j3++ + j5++ } - dAtA4[j3] = uint8(num) - j3++ + dAtA6[j5] = uint8(num) + j5++ } dAtA[i] = 0x12 i++ - i = encodeVarintPublic(dAtA, i, uint64(j3)) - i += copy(dAtA[i:], dAtA4[:j3]) + i = encodeVarintPublic(dAtA, i, uint64(j5)) + i += copy(dAtA[i:], dAtA6[:j5]) } if m.ColumnAttrs { dAtA[i] = 0x18 @@ -984,11 +1200,11 @@ func (m *QueryResult) MarshalTo(dAtA []byte) (int, error) { dAtA[i] = 0xa i++ i = encodeVarintPublic(dAtA, i, uint64(m.Row.Size())) - n5, err := m.Row.MarshalTo(dAtA[i:]) + n7, err := m.Row.MarshalTo(dAtA[i:]) if err != nil { return 0, err } - i += n5 + i += n7 } if m.N != 0 { dAtA[i] = 0x10 @@ -1021,17 +1237,56 @@ func (m *QueryResult) MarshalTo(dAtA []byte) (int, error) { dAtA[i] = 0x2a i++ i = encodeVarintPublic(dAtA, i, uint64(m.ValCount.Size())) - n6, err := m.ValCount.MarshalTo(dAtA[i:]) + n8, err := m.ValCount.MarshalTo(dAtA[i:]) if err != nil { return 0, err } - i += n6 + i += n8 } if m.Type != 0 { dAtA[i] = 0x30 i++ i = encodeVarintPublic(dAtA, i, uint64(m.Type)) } + if len(m.RowIDs) > 0 { + dAtA10 := make([]byte, len(m.RowIDs)*10) + var j9 int + for _, num := range m.RowIDs { + for num >= 1<<7 { + dAtA10[j9] = uint8(uint64(num)&0x7f | 0x80) + num >>= 7 + j9++ + } + dAtA10[j9] = uint8(num) + j9++ + } + dAtA[i] = 0x3a + i++ + i = encodeVarintPublic(dAtA, i, uint64(j9)) + i += copy(dAtA[i:], dAtA10[:j9]) + } + if len(m.GroupCounts) > 0 { + for _, msg := range m.GroupCounts { + dAtA[i] = 0x42 + i++ + i = encodeVarintPublic(dAtA, i, uint64(msg.Size())) + n, err := msg.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n + } + } + if m.RowIdentifiers != nil { + dAtA[i] = 0x4a + i++ + i = encodeVarintPublic(dAtA, i, uint64(m.RowIdentifiers.Size())) + n11, err := m.RowIdentifiers.MarshalTo(dAtA[i:]) + if err != nil { + return 0, err + } + i += n11 + } return i, nil } @@ -1068,56 +1323,56 @@ func (m *ImportRequest) MarshalTo(dAtA []byte) (int, error) { i = encodeVarintPublic(dAtA, i, uint64(m.Shard)) } if len(m.RowIDs) > 0 { - dAtA8 := make([]byte, len(m.RowIDs)*10) - var j7 int + dAtA13 := make([]byte, len(m.RowIDs)*10) + var j12 int for _, num := range m.RowIDs { for num >= 1<<7 { - dAtA8[j7] = uint8(uint64(num)&0x7f | 0x80) + dAtA13[j12] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 - j7++ + j12++ } - dAtA8[j7] = uint8(num) - j7++ + dAtA13[j12] = uint8(num) + j12++ } dAtA[i] = 0x22 i++ - i = encodeVarintPublic(dAtA, i, uint64(j7)) - i += copy(dAtA[i:], dAtA8[:j7]) + i = encodeVarintPublic(dAtA, i, uint64(j12)) + i += copy(dAtA[i:], dAtA13[:j12]) } if len(m.ColumnIDs) > 0 { - dAtA10 := make([]byte, len(m.ColumnIDs)*10) - var j9 int + dAtA15 := make([]byte, len(m.ColumnIDs)*10) + var j14 int for _, num := range m.ColumnIDs { for num >= 1<<7 { - dAtA10[j9] = uint8(uint64(num)&0x7f | 0x80) + dAtA15[j14] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 - j9++ + j14++ } - dAtA10[j9] = uint8(num) - j9++ + dAtA15[j14] = uint8(num) + j14++ } dAtA[i] = 0x2a i++ - i = encodeVarintPublic(dAtA, i, uint64(j9)) - i += copy(dAtA[i:], dAtA10[:j9]) + i = encodeVarintPublic(dAtA, i, uint64(j14)) + i += copy(dAtA[i:], dAtA15[:j14]) } if len(m.Timestamps) > 0 { - dAtA12 := make([]byte, len(m.Timestamps)*10) - var j11 int + dAtA17 := make([]byte, len(m.Timestamps)*10) + var j16 int for _, num1 := range m.Timestamps { num := uint64(num1) for num >= 1<<7 { - dAtA12[j11] = uint8(uint64(num)&0x7f | 0x80) + dAtA17[j16] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 - j11++ + j16++ } - dAtA12[j11] = uint8(num) - j11++ + dAtA17[j16] = uint8(num) + j16++ } dAtA[i] = 0x32 i++ - i = encodeVarintPublic(dAtA, i, uint64(j11)) - i += copy(dAtA[i:], dAtA12[:j11]) + i = encodeVarintPublic(dAtA, i, uint64(j16)) + i += copy(dAtA[i:], dAtA17[:j16]) } if len(m.RowKeys) > 0 { for _, s := range m.RowKeys { @@ -1185,39 +1440,39 @@ func (m *ImportValueRequest) MarshalTo(dAtA []byte) (int, error) { i = encodeVarintPublic(dAtA, i, uint64(m.Shard)) } if len(m.ColumnIDs) > 0 { - dAtA14 := make([]byte, len(m.ColumnIDs)*10) - var j13 int + dAtA19 := make([]byte, len(m.ColumnIDs)*10) + var j18 int for _, num := range m.ColumnIDs { for num >= 1<<7 { - dAtA14[j13] = uint8(uint64(num)&0x7f | 0x80) + dAtA19[j18] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 - j13++ + j18++ } - dAtA14[j13] = uint8(num) - j13++ + dAtA19[j18] = uint8(num) + j18++ } dAtA[i] = 0x2a i++ - i = encodeVarintPublic(dAtA, i, uint64(j13)) - i += copy(dAtA[i:], dAtA14[:j13]) + i = encodeVarintPublic(dAtA, i, uint64(j18)) + i += copy(dAtA[i:], dAtA19[:j18]) } if len(m.Values) > 0 { - dAtA16 := make([]byte, len(m.Values)*10) - var j15 int + dAtA21 := make([]byte, len(m.Values)*10) + var j20 int for _, num1 := range m.Values { num := uint64(num1) for num >= 1<<7 { - dAtA16[j15] = uint8(uint64(num)&0x7f | 0x80) + dAtA21[j20] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 - j15++ + j20++ } - dAtA16[j15] = uint8(num) - j15++ + dAtA21[j20] = uint8(num) + j20++ } dAtA[i] = 0x32 i++ - i = encodeVarintPublic(dAtA, i, uint64(j15)) - i += copy(dAtA[i:], dAtA16[:j15]) + i = encodeVarintPublic(dAtA, i, uint64(j20)) + i += copy(dAtA[i:], dAtA21[:j20]) } if len(m.ColumnKeys) > 0 { for _, s := range m.ColumnKeys { @@ -1271,6 +1526,25 @@ func (m *Row) Size() (n int) { return n } +func (m *RowIdentifiers) Size() (n int) { + var l int + _ = l + if len(m.Rows) > 0 { + l = 0 + for _, e := range m.Rows { + l += sovPublic(uint64(e)) + } + n += 1 + sovPublic(uint64(l)) + l + } + if len(m.Keys) > 0 { + for _, s := range m.Keys { + l = len(s) + n += 1 + l + sovPublic(uint64(l)) + } + } + return n +} + func (m *Pair) Size() (n int) { var l int _ = l @@ -1287,6 +1561,34 @@ func (m *Pair) Size() (n int) { return n } +func (m *FieldRow) Size() (n int) { + var l int + _ = l + l = len(m.Field) + if l > 0 { + n += 1 + l + sovPublic(uint64(l)) + } + if m.RowID != 0 { + n += 1 + sovPublic(uint64(m.RowID)) + } + return n +} + +func (m *GroupCount) Size() (n int) { + var l int + _ = l + if len(m.Group) > 0 { + for _, e := range m.Group { + l = e.Size() + n += 1 + l + sovPublic(uint64(l)) + } + } + if m.Count != 0 { + n += 1 + sovPublic(uint64(m.Count)) + } + return n +} + func (m *ValCount) Size() (n int) { var l int _ = l @@ -1448,6 +1750,23 @@ func (m *QueryResult) Size() (n int) { if m.Type != 0 { n += 1 + sovPublic(uint64(m.Type)) } + if len(m.RowIDs) > 0 { + l = 0 + for _, e := range m.RowIDs { + l += sovPublic(uint64(e)) + } + n += 1 + sovPublic(uint64(l)) + l + } + if len(m.GroupCounts) > 0 { + for _, e := range m.GroupCounts { + l = e.Size() + n += 1 + l + sovPublic(uint64(l)) + } + } + if m.RowIdentifiers != nil { + l = m.RowIdentifiers.Size() + n += 1 + l + sovPublic(uint64(l)) + } return n } @@ -1723,6 +2042,147 @@ func (m *Row) Unmarshal(dAtA []byte) error { } return nil } +func (m *RowIdentifiers) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: RowIdentifiers: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: RowIdentifiers: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType == 0 { + var v uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + m.Rows = append(m.Rows, v) + } else if wireType == 2 { + var packedLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + packedLen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if packedLen < 0 { + return ErrInvalidLengthPublic + } + postIndex := iNdEx + packedLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + for iNdEx < postIndex { + var v uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + m.Rows = append(m.Rows, v) + } + } else { + return fmt.Errorf("proto: wrong wireType = %d for field Rows", wireType) + } + case 2: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Keys", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPublic + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Keys = append(m.Keys, string(dAtA[iNdEx:postIndex])) + iNdEx = postIndex + default: + iNdEx = preIndex + skippy, err := skipPublic(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} func (m *Pair) Unmarshal(dAtA []byte) error { l := len(dAtA) iNdEx := 0 @@ -1840,6 +2300,204 @@ func (m *Pair) Unmarshal(dAtA []byte) error { } return nil } +func (m *FieldRow) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: FieldRow: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: FieldRow: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Field", wireType) + } + var stringLen uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + stringLen |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + intStringLen := int(stringLen) + if intStringLen < 0 { + return ErrInvalidLengthPublic + } + postIndex := iNdEx + intStringLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Field = string(dAtA[iNdEx:postIndex]) + iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field RowID", wireType) + } + m.RowID = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.RowID |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + default: + iNdEx = preIndex + skippy, err := skipPublic(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} +func (m *GroupCount) Unmarshal(dAtA []byte) error { + l := len(dAtA) + iNdEx := 0 + for iNdEx < l { + preIndex := iNdEx + var wire uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + wire |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + fieldNum := int32(wire >> 3) + wireType := int(wire & 0x7) + if wireType == 4 { + return fmt.Errorf("proto: GroupCount: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: GroupCount: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Group", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthPublic + } + postIndex := iNdEx + msglen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.Group = append(m.Group, &FieldRow{}) + if err := m.Group[len(m.Group)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Count", wireType) + } + m.Count = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Count |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + default: + iNdEx = preIndex + skippy, err := skipPublic(dAtA[iNdEx:]) + if err != nil { + return err + } + if skippy < 0 { + return ErrInvalidLengthPublic + } + if (iNdEx + skippy) > l { + return io.ErrUnexpectedEOF + } + iNdEx += skippy + } + } + + if iNdEx > l { + return io.ErrUnexpectedEOF + } + return nil +} func (m *ValCount) Unmarshal(dAtA []byte) error { l := len(dAtA) iNdEx := 0 @@ -2968,6 +3626,132 @@ func (m *QueryResult) Unmarshal(dAtA []byte) error { break } } + case 7: + if wireType == 0 { + var v uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + m.RowIDs = append(m.RowIDs, v) + } else if wireType == 2 { + var packedLen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + packedLen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if packedLen < 0 { + return ErrInvalidLengthPublic + } + postIndex := iNdEx + packedLen + if postIndex > l { + return io.ErrUnexpectedEOF + } + for iNdEx < postIndex { + var v uint64 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + v |= (uint64(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + m.RowIDs = append(m.RowIDs, v) + } + } else { + return fmt.Errorf("proto: wrong wireType = %d for field RowIDs", wireType) + } + case 8: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field GroupCounts", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthPublic + } + postIndex := iNdEx + msglen + if postIndex > l { + return io.ErrUnexpectedEOF + } + m.GroupCounts = append(m.GroupCounts, &GroupCount{}) + if err := m.GroupCounts[len(m.GroupCounts)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex + case 9: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field RowIdentifiers", wireType) + } + var msglen int + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + msglen |= (int(b) & 0x7F) << shift + if b < 0x80 { + break + } + } + if msglen < 0 { + return ErrInvalidLengthPublic + } + postIndex := iNdEx + msglen + if postIndex > l { + return io.ErrUnexpectedEOF + } + if m.RowIdentifiers == nil { + m.RowIdentifiers = &RowIdentifiers{} + } + if err := m.RowIdentifiers.Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex default: iNdEx = preIndex skippy, err := skipPublic(dAtA[iNdEx:]) @@ -3748,49 +4532,56 @@ var ( func init() { proto.RegisterFile("public.proto", fileDescriptorPublic) } var fileDescriptorPublic = []byte{ - // 701 bytes of a gzipped FileDescriptorProto - 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x55, 0xcd, 0x6e, 0xd3, 0x4c, - 0x14, 0xfd, 0x26, 0x76, 0xfe, 0x6e, 0x9a, 0x7c, 0xd5, 0x08, 0x8a, 0x85, 0x50, 0xb0, 0x2c, 0x84, - 0xbc, 0x4a, 0xa5, 0xb0, 0x07, 0xd1, 0x3f, 0x29, 0xaa, 0xa8, 0xe0, 0xb6, 0x14, 0xb1, 0x74, 0x9b, - 0x51, 0x1b, 0xc9, 0xf1, 0x18, 0x7b, 0xac, 0x34, 0xcf, 0xc1, 0x86, 0x47, 0x60, 0xc1, 0x43, 0xb0, - 0xec, 0x92, 0x47, 0x80, 0xf2, 0x22, 0x68, 0xee, 0x78, 0x62, 0x37, 0x95, 0x2a, 0x16, 0xec, 0xe6, - 0x9c, 0x33, 0x73, 0x67, 0xce, 0xcc, 0xb9, 0x36, 0x6c, 0xa4, 0xc5, 0x59, 0x3c, 0x3b, 0x1f, 0xa5, - 0x99, 0x54, 0x92, 0x77, 0x66, 0x89, 0x12, 0x59, 0x12, 0xc5, 0xc1, 0x47, 0x70, 0x50, 0x2e, 0xb8, - 0x07, 0xed, 0x5d, 0x19, 0x17, 0xf3, 0x24, 0xf7, 0x98, 0xef, 0x84, 0x2e, 0x5a, 0xc8, 0x9f, 0x41, - 0xf3, 0xb5, 0x52, 0x59, 0xee, 0x35, 0x7c, 0x27, 0xec, 0x8d, 0x07, 0x23, 0xbb, 0x74, 0xa4, 0x69, - 0x34, 0x22, 0xe7, 0xe0, 0x1e, 0x8a, 0x65, 0xee, 0x39, 0xbe, 0x13, 0x76, 0x91, 0xc6, 0xc1, 0x4b, - 0x70, 0xdf, 0x46, 0xb3, 0x8c, 0x0f, 0xa0, 0x31, 0xd9, 0xf3, 0x98, 0xcf, 0x42, 0x17, 0x1b, 0x93, - 0x3d, 0xfe, 0x00, 0x9a, 0xbb, 0xb2, 0x48, 0x94, 0xd7, 0x20, 0xca, 0x00, 0xbe, 0x09, 0xce, 0xa1, - 0x58, 0x7a, 0x8e, 0xcf, 0xc2, 0x2e, 0xea, 0x61, 0x30, 0x86, 0xce, 0x69, 0x14, 0xaf, 0xd4, 0xd3, - 0x28, 0xa6, 0x22, 0x0e, 0xea, 0xe1, 0xed, 0x2a, 0x4e, 0x59, 0x25, 0x78, 0x0f, 0xce, 0xce, 0x4c, - 0x69, 0x11, 0xe5, 0x62, 0xb5, 0xab, 0x01, 0xfc, 0x31, 0x74, 0x8c, 0xab, 0xc9, 0x5e, 0xb9, 0xf7, - 0x0a, 0xf3, 0x27, 0xd0, 0x3d, 0x99, 0xcd, 0x45, 0xae, 0xa2, 0x79, 0x4a, 0x87, 0x70, 0xb0, 0x22, - 0x82, 0x0f, 0xd0, 0x37, 0x33, 0xb5, 0xdb, 0x63, 0xa1, 0xee, 0x78, 0xfa, 0xbb, 0x5b, 0xba, 0xeb, - 0xf1, 0x2b, 0x03, 0x57, 0x6b, 0x56, 0x62, 0x2b, 0x49, 0x5f, 0xe9, 0xc9, 0x32, 0x15, 0xe5, 0x49, - 0x69, 0xcc, 0x7d, 0xe8, 0x1d, 0xab, 0x6c, 0x96, 0x5c, 0x9c, 0x46, 0x71, 0x21, 0xca, 0x42, 0x75, - 0x4a, 0x7b, 0x9c, 0x24, 0xca, 0xc8, 0x2e, 0xd9, 0x58, 0x61, 0xed, 0x71, 0x47, 0xca, 0xd8, 0x88, - 0x4d, 0x9f, 0x85, 0x1d, 0xac, 0x08, 0x3e, 0x04, 0x38, 0x88, 0x65, 0x54, 0xae, 0x6d, 0xf9, 0x2c, - 0x64, 0x58, 0x63, 0x82, 0x6d, 0x68, 0xeb, 0x93, 0xbe, 0x89, 0xd2, 0xca, 0x2d, 0xbb, 0xc7, 0x6d, - 0x70, 0xcd, 0x60, 0xe3, 0x5d, 0x21, 0xb2, 0x25, 0x8a, 0x4f, 0x85, 0xc8, 0xe9, 0x55, 0x08, 0x97, - 0x2e, 0x0d, 0xe0, 0x5b, 0xd0, 0x3a, 0xbe, 0x8c, 0xb2, 0xa9, 0xb9, 0x3b, 0x17, 0x4b, 0xa4, 0xbd, - 0x56, 0x77, 0x9e, 0x93, 0xd7, 0x0e, 0xd6, 0x29, 0xbd, 0x12, 0xc5, 0x5c, 0x2a, 0x6b, 0xa6, 0x44, - 0x3c, 0x84, 0xff, 0xf7, 0xaf, 0xce, 0xe3, 0x62, 0x2a, 0x50, 0x2e, 0xcc, 0xea, 0x16, 0x4d, 0x58, - 0xa7, 0xf9, 0x73, 0x18, 0x94, 0x94, 0x4d, 0x7f, 0x9b, 0x26, 0xae, 0xb1, 0xc1, 0x67, 0x06, 0xfd, - 0xd2, 0x4a, 0x9e, 0xca, 0x24, 0x17, 0xfa, 0xbd, 0xf6, 0xb3, 0xcc, 0xbe, 0xd7, 0x7e, 0x96, 0xf1, - 0x6d, 0x68, 0xa3, 0xc8, 0x8b, 0x58, 0xd9, 0x10, 0x3c, 0xac, 0xae, 0xc5, 0xae, 0x2d, 0x62, 0x85, - 0x76, 0x16, 0x7f, 0x05, 0x83, 0x5b, 0xa1, 0x32, 0xdd, 0xd3, 0x1b, 0x3f, 0xaa, 0xd6, 0xdd, 0xd2, - 0x71, 0x6d, 0x7a, 0xf0, 0x9d, 0x41, 0xaf, 0x56, 0x99, 0x3f, 0xa5, 0x5e, 0xa6, 0x33, 0xf5, 0xc6, - 0xfd, 0xaa, 0x0a, 0xca, 0x05, 0x52, 0x97, 0x6f, 0x00, 0x3b, 0x2a, 0xf3, 0xc4, 0x8e, 0xf4, 0x2b, - 0xea, 0xfe, 0xb4, 0xdb, 0xd6, 0x5e, 0x51, 0xd3, 0x68, 0x44, 0xfa, 0x32, 0x5c, 0x46, 0xc9, 0x85, - 0x98, 0x52, 0x9e, 0x3a, 0x68, 0x21, 0x1f, 0x55, 0xfd, 0x49, 0x0f, 0xd0, 0x1b, 0xf3, 0xaa, 0x84, - 0x55, 0xb0, 0xea, 0x61, 0x1b, 0x68, 0xfd, 0x16, 0x7d, 0x13, 0xe8, 0xe0, 0x17, 0x83, 0xfe, 0x64, - 0x9e, 0xca, 0x4c, 0xd5, 0x42, 0x32, 0x49, 0xa6, 0xe2, 0xca, 0x86, 0x84, 0x80, 0x66, 0x0f, 0x66, - 0x22, 0x9e, 0xd2, 0xe9, 0xbb, 0x68, 0x80, 0x66, 0x29, 0x2c, 0x14, 0x0e, 0x17, 0x0d, 0xa0, 0x58, - 0xe8, 0x7e, 0xcf, 0x3d, 0xd7, 0x04, 0xca, 0x20, 0x1d, 0x7f, 0xdb, 0xee, 0xb9, 0xd7, 0x24, 0xa9, - 0x22, 0x74, 0xfc, 0x57, 0xfd, 0xae, 0xf3, 0xe2, 0x84, 0x0e, 0xd6, 0x18, 0x7d, 0x0f, 0x28, 0x17, - 0xf4, 0x91, 0x6b, 0xd3, 0x47, 0xce, 0x42, 0xbd, 0xd2, 0x94, 0x21, 0xb1, 0x43, 0x62, 0x8d, 0x09, - 0xbe, 0x31, 0xe0, 0xc6, 0x23, 0x35, 0xd2, 0xbf, 0x33, 0x7a, 0xbf, 0xa1, 0x2d, 0x68, 0xd1, 0x7e, - 0xd6, 0x4c, 0x89, 0xd6, 0x8e, 0xdb, 0x5e, 0x3f, 0xee, 0xce, 0xe6, 0xf5, 0xcd, 0x90, 0xfd, 0xb8, - 0x19, 0xb2, 0x9f, 0x37, 0x43, 0xf6, 0xe5, 0xf7, 0xf0, 0xbf, 0xb3, 0x16, 0xfd, 0x34, 0x5e, 0xfc, - 0x09, 0x00, 0x00, 0xff, 0xff, 0x67, 0xca, 0x55, 0x5d, 0x44, 0x06, 0x00, 0x00, + // 804 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x55, 0xcb, 0x6e, 0xdb, 0x46, + 0x14, 0xed, 0x88, 0x94, 0x44, 0x5d, 0x59, 0xaa, 0x31, 0x70, 0x5d, 0xa2, 0x30, 0x54, 0x82, 0x28, + 0x0a, 0xae, 0x64, 0x40, 0x05, 0x8c, 0xae, 0xfa, 0xf0, 0xab, 0x10, 0xdc, 0x1a, 0xcd, 0xd8, 0x71, + 0x90, 0x25, 0x6d, 0x4d, 0x6c, 0x02, 0x14, 0x87, 0xe1, 0x03, 0xb2, 0xbe, 0x23, 0x9b, 0x7c, 0x42, + 0x16, 0xf9, 0x10, 0x2f, 0x83, 0x7c, 0x41, 0xe2, 0xfc, 0x48, 0x30, 0x77, 0x38, 0x1a, 0x8a, 0x0e, + 0x8c, 0x2c, 0xb2, 0x9b, 0x73, 0x5f, 0xbc, 0xe7, 0xbe, 0x08, 0x1b, 0x69, 0x79, 0x19, 0x47, 0x57, + 0xe3, 0x34, 0x13, 0x85, 0xa0, 0x4e, 0x94, 0x14, 0x3c, 0x4b, 0xc2, 0xd8, 0x7f, 0x0e, 0x16, 0x13, + 0x0b, 0xea, 0x42, 0xf7, 0x40, 0xc4, 0xe5, 0x3c, 0xc9, 0x5d, 0xe2, 0x59, 0x81, 0xcd, 0x34, 0xa4, + 0xbf, 0x40, 0xfb, 0xef, 0xa2, 0xc8, 0x72, 0xb7, 0xe5, 0x59, 0x41, 0x7f, 0x32, 0x1c, 0x6b, 0xd7, + 0xb1, 0x14, 0x33, 0xa5, 0xa4, 0x14, 0xec, 0x13, 0xbe, 0xcc, 0x5d, 0xcb, 0xb3, 0x82, 0x1e, 0xc3, + 0xb7, 0xff, 0x3b, 0x0c, 0x99, 0x58, 0x4c, 0x67, 0x3c, 0x29, 0xa2, 0x17, 0x11, 0x57, 0x56, 0x4c, + 0x2c, 0xf4, 0x27, 0xf0, 0xbd, 0xf2, 0x6c, 0xd5, 0x3c, 0xff, 0x00, 0xfb, 0xff, 0x30, 0xca, 0xe8, + 0x10, 0x5a, 0xd3, 0x43, 0x97, 0x78, 0x24, 0xb0, 0x59, 0x6b, 0x7a, 0x48, 0xb7, 0xa0, 0x7d, 0x20, + 0xca, 0xa4, 0x70, 0x5b, 0x28, 0x52, 0x80, 0x6e, 0x82, 0x75, 0xc2, 0x97, 0xae, 0xe5, 0x91, 0xa0, + 0xc7, 0xe4, 0xd3, 0xdf, 0x03, 0xe7, 0x38, 0xe2, 0xf1, 0x4c, 0x32, 0xdb, 0x82, 0x36, 0xbe, 0x31, + 0x4c, 0x8f, 0x29, 0x20, 0xa5, 0x32, 0xb7, 0x43, 0x1d, 0x09, 0x81, 0xff, 0x2f, 0xc0, 0x3f, 0x99, + 0x28, 0x53, 0x15, 0x37, 0x80, 0x36, 0x22, 0x4c, 0xb7, 0x3f, 0xa1, 0x86, 0xb9, 0x0e, 0xce, 0x94, + 0xc1, 0x97, 0xf3, 0xf2, 0x27, 0xe0, 0x5c, 0x84, 0xf1, 0x2a, 0xc7, 0x8b, 0x30, 0xc6, 0x1c, 0x2c, + 0x26, 0x9f, 0xeb, 0x3e, 0x96, 0xf6, 0x79, 0x0a, 0xd6, 0x7e, 0x54, 0x98, 0xf4, 0x48, 0x2d, 0x3d, + 0xfa, 0x13, 0x38, 0xaa, 0x2b, 0xab, 0xbc, 0x57, 0x98, 0xee, 0x40, 0xef, 0x3c, 0x9a, 0xf3, 0xbc, + 0x08, 0xe7, 0x29, 0x96, 0xc2, 0x62, 0x46, 0xe0, 0x3f, 0x83, 0x81, 0xb2, 0x94, 0xdd, 0x3a, 0xe3, + 0xc5, 0x83, 0xca, 0x7e, 0x5d, 0x97, 0x1f, 0x56, 0xfa, 0x0d, 0x01, 0x5b, 0xea, 0xb4, 0x8a, 0xac, + 0x54, 0xb2, 0xb1, 0xe7, 0xcb, 0x94, 0x57, 0x99, 0xe2, 0x9b, 0x7a, 0xd0, 0x3f, 0x2b, 0xb2, 0x28, + 0xb9, 0xbe, 0x08, 0xe3, 0x92, 0x57, 0x81, 0xea, 0x22, 0xc9, 0x71, 0x9a, 0x14, 0x4a, 0x6d, 0x23, + 0x8d, 0x15, 0x96, 0x1c, 0xf7, 0x85, 0x88, 0x95, 0xb2, 0xed, 0x91, 0xc0, 0x61, 0x46, 0x40, 0x47, + 0x00, 0xc7, 0xb1, 0x08, 0x2b, 0xdf, 0x8e, 0x47, 0x02, 0xc2, 0x6a, 0x12, 0x7f, 0x17, 0xba, 0x32, + 0xd3, 0xff, 0xc2, 0xd4, 0xb0, 0x25, 0x8f, 0xb0, 0xf5, 0xef, 0x08, 0x6c, 0x3c, 0x29, 0x79, 0xb6, + 0x64, 0xfc, 0x65, 0xc9, 0x73, 0xec, 0x0a, 0x62, 0x3d, 0x4a, 0x08, 0xe8, 0x36, 0x74, 0xce, 0x6e, + 0xc2, 0x6c, 0xa6, 0x6a, 0x67, 0xb3, 0x0a, 0x49, 0xae, 0xa6, 0xe6, 0x39, 0x72, 0x75, 0x58, 0x5d, + 0x24, 0x3d, 0x19, 0x9f, 0x8b, 0x42, 0x93, 0xa9, 0x10, 0x0d, 0xe0, 0xfb, 0xa3, 0xdb, 0xab, 0xb8, + 0x9c, 0x71, 0x26, 0x16, 0xca, 0xbb, 0x83, 0x06, 0x4d, 0x31, 0xfd, 0x15, 0x86, 0x95, 0x48, 0x6f, + 0x6f, 0x17, 0x0d, 0x1b, 0x52, 0xff, 0x15, 0x81, 0x41, 0x45, 0x25, 0x4f, 0x45, 0x92, 0x73, 0xd9, + 0xaf, 0xa3, 0x2c, 0xd3, 0xfd, 0x3a, 0xca, 0x32, 0xba, 0x0b, 0x5d, 0xc6, 0xf3, 0x32, 0x2e, 0xf4, + 0x10, 0xfc, 0x60, 0xca, 0xa2, 0x7d, 0xcb, 0xb8, 0x60, 0xda, 0x8a, 0xfe, 0x09, 0xc3, 0xb5, 0xa1, + 0x52, 0xdb, 0xdf, 0x9f, 0xfc, 0x68, 0xfc, 0xd6, 0xf4, 0xac, 0x61, 0xee, 0xbf, 0x6f, 0x41, 0xbf, + 0x16, 0x99, 0xfe, 0x8c, 0xb7, 0x08, 0x73, 0xea, 0x4f, 0x06, 0x26, 0x8a, 0xdc, 0x34, 0xbc, 0x52, + 0x1b, 0x40, 0x4e, 0xab, 0x79, 0x22, 0xa7, 0xb2, 0x8b, 0xf2, 0x4a, 0xe8, 0xcf, 0xd6, 0xba, 0x28, + 0xc5, 0x4c, 0x29, 0xf1, 0xb2, 0xdd, 0x84, 0xc9, 0x35, 0x9f, 0xe1, 0x3c, 0x39, 0x4c, 0x43, 0x3a, + 0x36, 0xfb, 0x89, 0x0d, 0x58, 0x5b, 0x71, 0xad, 0x61, 0x66, 0x87, 0xf5, 0x40, 0xcb, 0x5e, 0x0c, + 0xaa, 0x81, 0x96, 0x2d, 0x94, 0xbb, 0x29, 0x0b, 0x8f, 0xcd, 0x57, 0x88, 0xee, 0x41, 0xdf, 0x5c, + 0x92, 0xdc, 0x75, 0x30, 0xc3, 0x2d, 0x13, 0xde, 0x28, 0x59, 0xdd, 0x90, 0xfe, 0xd5, 0xbc, 0x99, + 0x6e, 0x0f, 0x33, 0x73, 0xd7, 0xaa, 0x51, 0xd3, 0xb3, 0x86, 0xbd, 0xff, 0x91, 0xc0, 0x60, 0x3a, + 0x4f, 0x45, 0x56, 0xd4, 0xc6, 0x76, 0x9a, 0xcc, 0xf8, 0xad, 0x1e, 0x5b, 0x04, 0xe6, 0x2e, 0xb6, + 0x1a, 0x77, 0x11, 0xc7, 0x17, 0xc7, 0xd5, 0x66, 0x0a, 0xd4, 0x58, 0xda, 0x6b, 0x2c, 0x77, 0xa0, + 0xa7, 0x0f, 0x50, 0xee, 0xb6, 0x51, 0x65, 0x04, 0x72, 0x21, 0x57, 0x17, 0x48, 0x4e, 0xb0, 0x15, + 0x58, 0xac, 0x26, 0x91, 0x9d, 0x61, 0x62, 0x81, 0xc7, 0xbf, 0x8b, 0xc7, 0x5f, 0x43, 0xe9, 0xa9, + 0xc2, 0xa0, 0xd2, 0x41, 0x65, 0x4d, 0xe2, 0xbf, 0x25, 0x40, 0x15, 0x47, 0x5c, 0xed, 0x6f, 0x47, + 0xf4, 0x71, 0x42, 0xdb, 0xd0, 0xc1, 0xef, 0x69, 0x32, 0x15, 0x6a, 0xa4, 0xdb, 0x6d, 0xa6, 0xbb, + 0xbf, 0x79, 0x77, 0x3f, 0x22, 0xef, 0xee, 0x47, 0xe4, 0xc3, 0xfd, 0x88, 0xbc, 0xfe, 0x34, 0xfa, + 0xee, 0xb2, 0x83, 0xbf, 0xe1, 0xdf, 0x3e, 0x07, 0x00, 0x00, 0xff, 0xff, 0xa6, 0x62, 0xa8, 0x25, + 0x96, 0x07, 0x00, 0x00, } diff --git a/internal/public.proto b/internal/public.proto index 04c98d070..229fcfadd 100644 --- a/internal/public.proto +++ b/internal/public.proto @@ -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 { diff --git a/pilosa.go b/pilosa.go index 9798ff315..d2fdf8144 100644 --- a/pilosa.go +++ b/pilosa.go @@ -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 diff --git a/test/pilosa.go b/test/pilosa.go index 27331d53f..3bc43d593 100644 --- a/test/pilosa.go +++ b/test/pilosa.go @@ -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))