From 54ce53732713eee358b231d1689df9f1900585a1 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Mon, 17 Sep 2018 16:01:51 -0500 Subject: [PATCH] copy non roaring-import code from Todds's row-iterate PR tests passing --- encoding/proto/proto.go | 26 ++ executor.go | 391 +++++++++++++++++++++++++++- executor_test.go | 143 +++++++++++ fragment.go | 49 +++- internal/public.pb.go | 552 ++++++++++++++++++++++++++++++++-------- internal/public.proto | 9 +- pilosa.go | 4 +- 7 files changed, 1051 insertions(+), 123 deletions(-) diff --git a/encoding/proto/proto.go b/encoding/proto/proto.go index 0fcd9df3a..e133de7fc 100644 --- a/encoding/proto/proto.go +++ b/encoding/proto/proto.go @@ -360,6 +360,12 @@ 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.GroupByCounts: + pb.Results[i].Type = queryResultTypeGroupByCounts + pb.Results[i].GroupByCounts = encodeGroupByCount(result) case nil: pb.Results[i].Type = queryResultTypeNil } @@ -922,6 +928,8 @@ const ( queryResultTypeValCount queryResultTypeUint64 queryResultTypeBool + queryResultTypeRowIDs + queryResultTypeGroupByCounts ) func decodeQueryResult(pb *internal.QueryResult) interface{} { @@ -938,6 +946,8 @@ func decodeQueryResult(pb *internal.QueryResult) interface{} { return pb.Changed case queryResultTypeNil: return nil + case queryResultTypeGroupByCounts: + return decodeGroupByCounts(pb.GroupByCounts) } panic(fmt.Sprintf("unknown type: %d", pb.Type)) } @@ -988,6 +998,14 @@ func decodeAttr(attr *internal.Attr) (key string, value interface{}) { } } +func decodeGroupByCounts(a []*internal.GroupLine) pilosa.GroupByCounts { + gbc := make(pilosa.GroupByCounts, 0) + for i := range a { + gbc = append(gbc, pilosa.GroupLine{a[i].Groups, a[i].Total}) + } + return gbc +} + func decodePairs(a []*internal.Pair) []pilosa.Pair { other := make([]pilosa.Pair, len(a)) for i := range a { @@ -1039,6 +1057,14 @@ func encodeRow(r *pilosa.Row) *internal.Row { } } +func encodeGroupByCount(counts pilosa.GroupByCounts) []*internal.GroupLine { + result := make([]*internal.GroupLine, len(counts)) + for i := range counts { + result[i] = &internal.GroupLine{Groups: counts[i].Groups, Total: counts[i].Total} + } + return result +} + 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 b0083a514..56ab05269 100644 --- a/executor.go +++ b/executor.go @@ -18,6 +18,8 @@ import ( "context" "fmt" "sort" + "strconv" + "strings" "time" "github.com/pilosa/pilosa/pql" @@ -194,6 +196,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: @@ -714,14 +722,355 @@ func (e *executor) executeDifferenceShard(ctx context.Context, index string, c * return other, nil } +type RowIDs []uint64 + +func (r RowIDs) Merge(other RowIDs) RowIDs { + i, j := 0, 0 + result := make(RowIDs, 0) + for i < len(r) && j < len(other) { + 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) { + result = append(result, r[i]) + i++ + } + for j < len(other) { + result = append(result, other[j]) + j++ + } + return result +} +func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (GroupByCounts, error) { + // Execute calls in bulk on each remote node and merge. + mapFn := func(shard uint64) (interface{}, error) { + return e.executeGroupByShard(ctx, index, c, shard) + } + // Merge returned results at coordinating node. + reduceFn := func(prev, v interface{}) interface{} { + other, _ := prev.(GroupByCounts) + return other.Merge(v.(GroupByCounts)) + } + // Get full result set. + other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn) + if err != nil { + return nil, err + } + results, _ := other.(GroupByCounts) + // 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 +} + +// gbi is a groupBy item. +type gbi struct { + row *Row + fieldKey string + rowID uint64 +} +type GroupLine struct { + Groups []string + Total uint64 +} +type GroupByCounts []GroupLine + +func (gbc GroupByCounts) Merge(other GroupByCounts) GroupByCounts { + m := make(map[string]struct { + i int + total uint64 + }) + for i := range gbc { + m[strings.Join(gbc[i].Groups, "-")] = struct { + i int + total uint64 + }{total: gbc[i].Total, i: i} + } + for i := range other { + key := strings.Join(other[i].Groups, "-") + o, found := m[key] + if found { + gbc[o.i].Total += other[i].Total + } else { + gbc = append(gbc, other[i]) + } + } + return gbc +} +func makeGroup(parts []gbi) GroupLine { + var other *Row + line := GroupLine{} + for i, o := range parts { + if i == 0 { + other = o.row + } else { + other = other.Intersect(o.row) + } + line.Groups = append(line.Groups, o.fieldKey) + } + line.Total = other.Count() + return line +} +func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql.Call, shard uint64) (GroupByCounts, error) { + // Fetch index. + idx := e.Holder.Index(index) + if idx == nil { + return nil, ErrIndexNotFound + } + // fieldDirective is a combination of field + // instructions, represented as a string with + // the form [fieldName:offset:limit] or + // [fieldName:limit]. + fieldDirectives, ok := c.Args["fields"] + if !ok { + return nil, errors.Wrap(ErrFieldsArgumentRequired, "executeGroupBy") + } + // Ensure that fieldDirectives is a list. + if _, ok := fieldDirectives.([]interface{}); !ok { + return nil, errors.Wrap(ErrExpectedFieldListArgument, "executeGroupBy") + } + // getFieldName extracts the fieldName portion of the + // fieldDirective. + getFieldName := func(s string) string { + parts := strings.Split(s, ":") + return parts[0] + } + // Ensure that all of the fields exist. + for _, fieldDirective := range fieldDirectives.([]interface{}) { + fieldName := getFieldName(fieldDirective.(string)) + f := e.Holder.Field(index, fieldName) + if f == nil { + return nil, errors.Wrap(ErrFieldNotFound, fmt.Sprintf("executeGroupBy: %s", fieldDirective.(string))) + } + } + results := make(GroupByCounts, 0) + var work listOfGBILists + for _, fieldDirective := range fieldDirectives.([]interface{}) { + fieldName := getFieldName(fieldDirective.(string)) + // Fetch fragment. + frag := e.Holder.fragment(index, fieldName, viewStandard, shard) + if frag == nil { // this means this whole shard doesn't have all it needs to continue + return results, nil + } + // Get filter based on the field directive. + filter, err := getGroupByFilterFunction(fieldDirective.(string)) + if err != nil { + return nil, err + } + set := make(gbiList, 0) + for _, rowID := range frag.rowsWithFilter(filter) { + set = append(set, gbi{ + row: frag.row(rowID), + rowID: rowID, + fieldKey: fmt.Sprintf("%s.%d", fieldName, rowID), + }) + } + work = append(work, set) + } + for _, group := range product(work) { + group.gl.Total = group.row.Count() + if group.gl.Total > 0 { + results = append(results, group.gl) + } + } + return results, nil +} + +type gbiList []gbi +type listOfGBILists []gbiList + +// pi is a product process item. +type pi struct { + row *Row + gl GroupLine +} +type piList []pi + +// product generates the cartiesian product of the input +// using tail recursion +func product(input listOfGBILists) piList { + res := make(piList, 0) + if len(input) == 0 { //base return empty list + res = append(res, pi{gl: GroupLine{Groups: make([]string, 0)}}) + } else { + res = productHelper(input, res) + } + return res +} +func productHelper(lists listOfGBILists, res piList) piList { + head := lists[0] //take first element of the list + tail := product(lists[1:]) //invoke product on remaining element + for h := range head { // for each head + for t := range tail { //iterate over the tail + s := pi{gl: GroupLine{Groups: make([]string, 0)}} + s.gl.Groups = append([]string{head[h].fieldKey}, tail[t].gl.Groups...) //had to insert at the front to match input order + if tail[t].row != nil { //first time around nothing to intersect + s.row = head[h].row.Intersect(tail[t].row) + } else { + s.row = head[h].row + } + res = append(res, s) + } + } + return res +} +func (e *executor) executeRows(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (RowIDs, error) { + // Execute calls in bulk on each remote node and merge. + mapFn := func(shard uint64) (interface{}, error) { + return e.executeRowsShard(ctx, index, c, shard) + } + // Merge returned results at coordinating node. + reduceFn := func(prev, v interface{}) interface{} { + other, _ := prev.(RowIDs) + return other.Merge(v.(RowIDs)) + } + // Get full result set. + other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn) + if err != nil { + return nil, err + } + results, _ := other.(RowIDs) + // 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 +} +func (e *executor) executeRowsShard(ctx 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 + } + if columnID, ok, err := c.UintArg("column"); err != nil { + return nil, err + } else if ok { + // TODO: it's possible that filters could be applied here, so this returns too early. + return frag.rowsForColumn(columnID), nil + } + filter := getFilterFunction(c) + return frag.rowsWithFilter(filter), nil +} + +// getGroupByFilterFunction returns a rowFilter based on the +// field directive provided. +func getGroupByFilterFunction(fieldDirective string) (rowFilter, error) { + parts := strings.Split(fieldDirective, ":") + hasLimit := false + hasOffset := false + limit := uint64(0) + offset := uint64(0) + var err error + // fieldDirective can have one of the following forms: + // [fieldName] + // [fieldName:limit] + // [fieldName:offset:limit] + // + // Note that a field directive with the form + // [fieldName:offset:limit:extra] will be treated as + // [fieldName:offset:limit] (i.e. `extra` is ignored). + if len(parts) == 1 { + return noFilter, nil + } else if len(parts) == 2 { + hasLimit = true + if limit, err = strconv.ParseUint(parts[1], 10, 64); err != nil { + return nil, errors.Wrap(err, "getting groupby field limit only value") + } + } else { + hasOffset = true + if offset, err = strconv.ParseUint(parts[1], 10, 64); err != nil { + return nil, errors.Wrap(err, "getting groupby field offset value") + } + if parts[2] != "" { + hasLimit = true + if limit, err = strconv.ParseUint(parts[2], 10, 64); err != nil { + return nil, errors.Wrap(err, "getting groupby field limit value") + } + } + } + if hasOffset && hasLimit { + f := filterWithOffsetLimit{offset: offset, limit: limit} + return f.filter, nil + } else if hasLimit { + f := filterWithLimit{limit: limit} + return f.filter, nil + } + f := filterWithOffset{offset: offset} + return f.filter, nil +} +func getFilterFunction(c *pql.Call) rowFilter { + offset, hasOffset, _ := c.UintArg("shardoffset") + limit, hasLimit, _ := c.UintArg("shardlimit") + if hasOffset && hasLimit { + f := filterWithOffsetLimit{offset: offset, limit: limit} + return f.filter + } else if hasOffset { + f := filterWithOffset{offset: offset} + return f.filter + } else if hasLimit { + f := filterWithLimit{limit: limit} + return f.filter + } + return noFilter +} + 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") @@ -1787,7 +2136,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: @@ -1845,3 +2194,39 @@ func isString(v interface{}) bool { _, ok := v.(string) return ok } + +// Filters to be used with RowsWithFilter queries. +type filterWithOffsetLimit struct { + offset, limit uint64 +} + +func (fol *filterWithOffsetLimit) filter(rowID uint64) (bool, bool) { + if rowID >= fol.offset { + if fol.limit > 0 { + fol.limit-- + return true, false + } + return false, true + } + return false, false +} + +type filterWithOffset struct { + offset uint64 +} + +func (fo *filterWithOffset) filter(rowID uint64) (bool, bool) { + return rowID >= fo.offset, false +} + +type filterWithLimit struct { + limit uint64 +} + +func (fl *filterWithLimit) filter(rowID uint64) (bool, bool) { // nolint: unparam + if fl.limit > 0 { + fl.limit-- + return true, false + } + return false, true +} diff --git a/executor_test.go b/executor_test.go index ee919f001..84e00936d 100644 --- a/executor_test.go +++ b/executor_test.go @@ -1225,6 +1225,22 @@ Set(4500001, fn=4) t.Fatalf("wrong attrs: %v", attrst) } }) + + t.Run("remote groupBy", func(t *testing.T) { + if res, err := c[1].API.Query(context.Background(), &pilosa.QueryRequest{ + Index: "i", + Query: `GroupBy(fields=[f])`, + }); err != nil { + t.Fatalf("GroupBy querying: %v", err) + } else { + expected := pilosa.GroupByCounts{ + {Groups: []string{"f.10"}, Total: 4}, + {Groups: []string{"f.7"}, Total: 1}, + } + results := res.Results[0].(pilosa.GroupByCounts) + checkGroupBy(expected, results, t) + } + }) } // Ensure executor returns an error if too many writes are in a single request. @@ -1622,3 +1638,130 @@ func benchmarkExistence(nn bool, b *testing.B) { func BenchmarkExecutor_Existence_True(b *testing.B) { benchmarkExistence(true, b) } func BenchmarkExecutor_Existence_False(b *testing.B) { benchmarkExistence(false, b) } + +func TestExecutor_Execute_Rows(t *testing.T) { + c := test.MustRunCluster(t, 1) + defer c.Close() + hldr := test.Holder{Holder: c[0].Server.Holder()} + hldr.SetBit("i", "general", 10, 0) + hldr.SetBit("i", "general", 10, ShardWidth+1) + hldr.SetBit("i", "general", 11, 2) + hldr.SetBit("i", "general", 11, ShardWidth+2) + hldr.SetBit("i", "general", 12, 2) + hldr.SetBit("i", "general", 12, ShardWidth+2) + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Rows(field=general)`}); err != nil { + t.Fatal(err) + } else if columns := res.Results[0].(pilosa.RowIDs); !reflect.DeepEqual(columns, pilosa.RowIDs{10, 11, 12}) { + t.Fatalf("unexpected columns: %+v", columns) + } + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Rows(field=general, limit=2)`}); err != nil { + t.Fatal(err) + } else if columns := res.Results[0].(pilosa.RowIDs); !reflect.DeepEqual(columns, pilosa.RowIDs{10, 11}) { + t.Fatalf("unexpected columns: %+v", columns) + } + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Rows(field=general, offset=1,limit=2)`}); err != nil { + t.Fatal(err) + } else if columns := res.Results[0].(pilosa.RowIDs); !reflect.DeepEqual(columns, pilosa.RowIDs{11, 12}) { + t.Fatalf("unexpected columns: %+v", columns) + } + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Rows(field=general, column=2)`}); err != nil { + t.Fatal(err) + } else if columns := res.Results[0].(pilosa.RowIDs); !reflect.DeepEqual(columns, pilosa.RowIDs{11, 12}) { + t.Fatalf("unexpected columns: %+v", columns) + } +} +func TestExecutor_Execute_GroupBy(t *testing.T) { + c := test.MustRunCluster(t, 1) + defer c.Close() + hldr := test.Holder{Holder: c[0].Server.Holder()} + hldr.SetBit("i", "general", 10, 0) + hldr.SetBit("i", "general", 10, 1) + hldr.SetBit("i", "general", 10, ShardWidth+1) + hldr.SetBit("i", "general", 11, 2) + hldr.SetBit("i", "general", 11, ShardWidth+2) + hldr.SetBit("i", "general", 12, 2) + hldr.SetBit("i", "general", 12, ShardWidth+2) + hldr.SetBit("i", "sub", 10, 0) + hldr.SetBit("i", "sub", 10, 1) + hldr.SetBit("i", "sub", 10, 3) + hldr.SetBit("i", "sub", 11, 2) + hldr.SetBit("i", "sub", 11, 0) + expected := pilosa.GroupByCounts{ + {Groups: []string{"general.10", "sub.11"}, Total: 1}, + {Groups: []string{"general.11", "sub.11"}, Total: 1}, + {Groups: []string{"general.12", "sub.11"}, Total: 1}, + {Groups: []string{"general.10", "sub.10"}, Total: 2}, + } + 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 errors.Cause(err) != pilosa.ErrFieldsArgumentRequired { + t.Fatalf("unexpected error\n\"%s\" not returned instead \n\"%s\"", pilosa.ErrFieldsArgumentRequired, err) + } + } + }) + t.Run("Unknown Field ", func(t *testing.T) { + if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=[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("Bad Field Format", func(t *testing.T) { + if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=missing)`}); err != nil { + if errors.Cause(err) != pilosa.ErrExpectedFieldListArgument { + t.Fatalf("unexpected error\n\"%s\" not returned instead \n\"%s\"", pilosa.ErrExpectedFieldListArgument, err) + } + } + }) + t.Run("Basic", func(t *testing.T) { + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=[general,sub])`}); err != nil { + t.Fatal(err) + } else { + results := res.Results[0].(pilosa.GroupByCounts) + checkGroupBy(expected, results, t) + } + }) + expected = pilosa.GroupByCounts{ + {Groups: []string{"general.11"}, Total: 2}, + {Groups: []string{"general.12"}, Total: 2}, + } + t.Run("check field offset no limit", func(t *testing.T) { + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=[general:11:])`}); err != nil { + t.Fatal(err) + } else { + results := res.Results[0].(pilosa.GroupByCounts) + checkGroupBy(expected, results, t) + } + }) + expected = pilosa.GroupByCounts{ + {Groups: []string{"general.11"}, Total: 2}, + } + t.Run("check field offset limit", func(t *testing.T) { + if res, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `GroupBy(fields=[general:11:1])`}); err != nil { + t.Fatal(err) + } else { + results := res.Results[0].(pilosa.GroupByCounts) + checkGroupBy(expected, results, t) + } + }) +} +func checkGroupBy(expected, results pilosa.GroupByCounts, t *testing.T) { + notIn := func(item pilosa.GroupLine, expected pilosa.GroupByCounts) bool { + for i := range expected { + if item.Total == expected[i].Total { + if reflect.DeepEqual(item.Groups, expected[i].Groups) { + return false + } + } + } + return true + } + if len(results) != len(expected) { + t.Fatalf("number of groupings mismatch: \n%+v\n%+v\n", results, expected) + } + for _, result := range results { + if notIn(result, expected) { + t.Fatalf("unexpected grouping: \n%+v\n\n\n%+v\n", result, expected) + } + } +} diff --git a/fragment.go b/fragment.go index 7f651b135..bcd8e9b13 100644 --- a/fragment.go +++ b/fragment.go @@ -1758,10 +1758,27 @@ func (f *fragment) readCacheFromArchive(r io.Reader) error { return nil } +// rowFilter is a filter function which takes a rowID +// and determines if that row should be included in +// the result set. Additionally, it signals whether +// to halt processing any more rows. The two bool +// returned are (1) include row, (2) break further +// processing. +type rowFilter func(rowID uint64) (bool, bool) + +// noFilter is a filter function which has no restrictions. +var noFilter = func(rowID uint64) (bool, bool) { return true, false } + +// rows returns all rows by calling rowsWithFilter() +// with a completely unrestrictive filter. + func (f *fragment) rows() []uint64 { + return f.rowsWithFilter(noFilter) +} + +func (f *fragment) rowsWithFilter(filter rowFilter) []uint64 { i, _ := f.storage.Containers.Iterator(0) rows := make([]uint64, 0) - var lastRow uint64 = math.MaxUint64 // Loop over the existing containers. @@ -1776,22 +1793,33 @@ func (f *fragment) rows() []uint64 { continue } - rows = append(rows, vRow) + // apply filter + if addRow, breakOut := filter(vRow); breakOut { + break + } else if addRow { + rows = append(rows, vRow) + } + lastRow = vRow } return rows } +// rowsForColumn is similar to the rows method, but isolated +// to a single column. func (f *fragment) rowsForColumn(columnID uint64) []uint64 { - var colKey uint64 + return f.rowsForColumnWithFilter(columnID, noFilter) +} + +func (f *fragment) rowsForColumnWithFilter(columnID uint64, filter rowFilter) []uint64 { + i, _ := f.storage.Containers.Iterator(0) + rows := make([]uint64, 0) colID := columnID % ShardWidth - i, _ := f.storage.Containers.Iterator(0) + colVal := uint16(colID & 0xFFFF) // columnID within the container - colVal := uint16(colID & 0xFFFF) - - rows := make([]uint64, 0) + var colKey uint64 // Loop over the existing containers. for i.Next() { @@ -1807,8 +1835,13 @@ func (f *fragment) rowsForColumn(columnID uint64) []uint64 { continue } + // apply filter if c.Contains(colVal) { - rows = append(rows, vRow) + if addRow, breakOut := filter(vRow); breakOut { + break + } else if addRow { + rows = append(rows, vRow) + } } } return rows diff --git a/internal/public.pb.go b/internal/public.pb.go index cceea4806..8139d3a28 100644 --- a/internal/public.pb.go +++ b/internal/public.pb.go @@ -10,6 +10,7 @@ It has these top-level messages: Row Pair + GroupLine ValCount Bit ColumnAttrSet @@ -106,6 +107,30 @@ func (m *Pair) GetCount() uint64 { return 0 } +type GroupLine struct { + Groups []string `protobuf:"bytes,1,rep,name=Groups" json:"Groups,omitempty"` + Total uint64 `protobuf:"varint,2,opt,name=Total,proto3" json:"Total,omitempty"` +} + +func (m *GroupLine) Reset() { *m = GroupLine{} } +func (m *GroupLine) String() string { return proto.CompactTextString(m) } +func (*GroupLine) ProtoMessage() {} +func (*GroupLine) Descriptor() ([]byte, []int) { return fileDescriptorPublic, []int{2} } + +func (m *GroupLine) GetGroups() []string { + if m != nil { + return m.Groups + } + return nil +} + +func (m *GroupLine) GetTotal() uint64 { + if m != nil { + return m.Total + } + 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 +139,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{3} } func (m *ValCount) GetVal() int64 { if m != nil { @@ -139,7 +164,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{4} } func (m *Bit) GetRowID() uint64 { if m != nil { @@ -171,7 +196,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{5} } func (m *ColumnAttrSet) GetID() uint64 { if m != nil { @@ -206,7 +231,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{6} } func (m *Attr) GetKey() string { if m != nil { @@ -257,7 +282,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{7} } func (m *AttrMap) GetAttrs() []*Attr { if m != nil { @@ -278,7 +303,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{8} } func (m *QueryRequest) GetQuery() string { if m != nil { @@ -331,7 +356,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{9} } func (m *QueryResponse) GetErr() string { if m != nil { @@ -355,18 +380,20 @@ 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"` + GroupByCounts []*GroupLine `protobuf:"bytes,8,rep,name=GroupByCounts" json:"GroupByCounts,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{10} } func (m *QueryResult) GetType() uint32 { if m != nil { @@ -396,6 +423,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 +437,18 @@ 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) GetGroupByCounts() []*GroupLine { + if m != nil { + return m.GroupByCounts + } + return nil } type ImportRequest struct { @@ -424,7 +465,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{11} } func (m *ImportRequest) GetIndex() string { if m != nil { @@ -494,7 +535,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{12} } func (m *ImportValueRequest) GetIndex() string { if m != nil { @@ -541,6 +582,7 @@ func (m *ImportValueRequest) GetValues() []int64 { func init() { proto.RegisterType((*Row)(nil), "internal.Row") proto.RegisterType((*Pair)(nil), "internal.Pair") + proto.RegisterType((*GroupLine)(nil), "internal.GroupLine") proto.RegisterType((*ValCount)(nil), "internal.ValCount") proto.RegisterType((*Bit)(nil), "internal.Bit") proto.RegisterType((*ColumnAttrSet)(nil), "internal.ColumnAttrSet") @@ -648,6 +690,44 @@ func (m *Pair) MarshalTo(dAtA []byte) (int, error) { return i, nil } +func (m *GroupLine) 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 *GroupLine) MarshalTo(dAtA []byte) (int, error) { + var i int + _ = i + var l int + _ = l + if len(m.Groups) > 0 { + for _, s := range m.Groups { + dAtA[i] = 0xa + 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) + } + } + if m.Total != 0 { + dAtA[i] = 0x10 + i++ + i = encodeVarintPublic(dAtA, i, uint64(m.Total)) + } + return i, nil +} + func (m *ValCount) Marshal() (dAtA []byte, err error) { size := m.Size() dAtA = make([]byte, size) @@ -1032,6 +1112,35 @@ func (m *QueryResult) MarshalTo(dAtA []byte) (int, error) { i++ i = encodeVarintPublic(dAtA, i, uint64(m.Type)) } + if len(m.RowIDs) > 0 { + dAtA8 := make([]byte, len(m.RowIDs)*10) + var j7 int + for _, num := range m.RowIDs { + for num >= 1<<7 { + dAtA8[j7] = uint8(uint64(num)&0x7f | 0x80) + num >>= 7 + j7++ + } + dAtA8[j7] = uint8(num) + j7++ + } + dAtA[i] = 0x3a + i++ + i = encodeVarintPublic(dAtA, i, uint64(j7)) + i += copy(dAtA[i:], dAtA8[:j7]) + } + if len(m.GroupByCounts) > 0 { + for _, msg := range m.GroupByCounts { + 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 + } + } return i, nil } @@ -1068,26 +1177,9 @@ 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 - for _, num := range m.RowIDs { - for num >= 1<<7 { - dAtA8[j7] = uint8(uint64(num)&0x7f | 0x80) - num >>= 7 - j7++ - } - dAtA8[j7] = uint8(num) - j7++ - } - dAtA[i] = 0x22 - i++ - i = encodeVarintPublic(dAtA, i, uint64(j7)) - i += copy(dAtA[i:], dAtA8[:j7]) - } - if len(m.ColumnIDs) > 0 { - dAtA10 := make([]byte, len(m.ColumnIDs)*10) + dAtA10 := make([]byte, len(m.RowIDs)*10) var j9 int - for _, num := range m.ColumnIDs { + for _, num := range m.RowIDs { for num >= 1<<7 { dAtA10[j9] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 @@ -1096,16 +1188,15 @@ func (m *ImportRequest) MarshalTo(dAtA []byte) (int, error) { dAtA10[j9] = uint8(num) j9++ } - dAtA[i] = 0x2a + dAtA[i] = 0x22 i++ i = encodeVarintPublic(dAtA, i, uint64(j9)) i += copy(dAtA[i:], dAtA10[:j9]) } - if len(m.Timestamps) > 0 { - dAtA12 := make([]byte, len(m.Timestamps)*10) + if len(m.ColumnIDs) > 0 { + dAtA12 := make([]byte, len(m.ColumnIDs)*10) var j11 int - for _, num1 := range m.Timestamps { - num := uint64(num1) + for _, num := range m.ColumnIDs { for num >= 1<<7 { dAtA12[j11] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 @@ -1114,11 +1205,29 @@ func (m *ImportRequest) MarshalTo(dAtA []byte) (int, error) { dAtA12[j11] = uint8(num) j11++ } - dAtA[i] = 0x32 + dAtA[i] = 0x2a i++ i = encodeVarintPublic(dAtA, i, uint64(j11)) i += copy(dAtA[i:], dAtA12[:j11]) } + if len(m.Timestamps) > 0 { + dAtA14 := make([]byte, len(m.Timestamps)*10) + var j13 int + for _, num1 := range m.Timestamps { + num := uint64(num1) + for num >= 1<<7 { + dAtA14[j13] = uint8(uint64(num)&0x7f | 0x80) + num >>= 7 + j13++ + } + dAtA14[j13] = uint8(num) + j13++ + } + dAtA[i] = 0x32 + i++ + i = encodeVarintPublic(dAtA, i, uint64(j13)) + i += copy(dAtA[i:], dAtA14[:j13]) + } if len(m.RowKeys) > 0 { for _, s := range m.RowKeys { dAtA[i] = 0x3a @@ -1185,27 +1294,9 @@ 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 - for _, num := range m.ColumnIDs { - for num >= 1<<7 { - dAtA14[j13] = uint8(uint64(num)&0x7f | 0x80) - num >>= 7 - j13++ - } - dAtA14[j13] = uint8(num) - j13++ - } - dAtA[i] = 0x2a - i++ - i = encodeVarintPublic(dAtA, i, uint64(j13)) - i += copy(dAtA[i:], dAtA14[:j13]) - } - if len(m.Values) > 0 { - dAtA16 := make([]byte, len(m.Values)*10) + dAtA16 := make([]byte, len(m.ColumnIDs)*10) var j15 int - for _, num1 := range m.Values { - num := uint64(num1) + for _, num := range m.ColumnIDs { for num >= 1<<7 { dAtA16[j15] = uint8(uint64(num)&0x7f | 0x80) num >>= 7 @@ -1214,11 +1305,29 @@ func (m *ImportValueRequest) MarshalTo(dAtA []byte) (int, error) { dAtA16[j15] = uint8(num) j15++ } - dAtA[i] = 0x32 + dAtA[i] = 0x2a i++ i = encodeVarintPublic(dAtA, i, uint64(j15)) i += copy(dAtA[i:], dAtA16[:j15]) } + if len(m.Values) > 0 { + dAtA18 := make([]byte, len(m.Values)*10) + var j17 int + for _, num1 := range m.Values { + num := uint64(num1) + for num >= 1<<7 { + dAtA18[j17] = uint8(uint64(num)&0x7f | 0x80) + num >>= 7 + j17++ + } + dAtA18[j17] = uint8(num) + j17++ + } + dAtA[i] = 0x32 + i++ + i = encodeVarintPublic(dAtA, i, uint64(j17)) + i += copy(dAtA[i:], dAtA18[:j17]) + } if len(m.ColumnKeys) > 0 { for _, s := range m.ColumnKeys { dAtA[i] = 0x3a @@ -1287,6 +1396,21 @@ func (m *Pair) Size() (n int) { return n } +func (m *GroupLine) Size() (n int) { + var l int + _ = l + if len(m.Groups) > 0 { + for _, s := range m.Groups { + l = len(s) + n += 1 + l + sovPublic(uint64(l)) + } + } + if m.Total != 0 { + n += 1 + sovPublic(uint64(m.Total)) + } + return n +} + func (m *ValCount) Size() (n int) { var l int _ = l @@ -1448,6 +1572,19 @@ 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.GroupByCounts) > 0 { + for _, e := range m.GroupByCounts { + l = e.Size() + n += 1 + l + sovPublic(uint64(l)) + } + } return n } @@ -1840,6 +1977,104 @@ func (m *Pair) Unmarshal(dAtA []byte) error { } return nil } +func (m *GroupLine) 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: GroupLine: wiretype end group for non-group") + } + if fieldNum <= 0 { + return fmt.Errorf("proto: GroupLine: illegal tag %d (wire type %d)", fieldNum, wire) + } + switch fieldNum { + case 1: + if wireType != 2 { + return fmt.Errorf("proto: wrong wireType = %d for field Groups", 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.Groups = append(m.Groups, string(dAtA[iNdEx:postIndex])) + iNdEx = postIndex + case 2: + if wireType != 0 { + return fmt.Errorf("proto: wrong wireType = %d for field Total", wireType) + } + m.Total = 0 + for shift := uint(0); ; shift += 7 { + if shift >= 64 { + return ErrIntOverflowPublic + } + if iNdEx >= l { + return io.ErrUnexpectedEOF + } + b := dAtA[iNdEx] + iNdEx++ + m.Total |= (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 +3203,99 @@ 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 GroupByCounts", 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.GroupByCounts = append(m.GroupByCounts, &GroupLine{}) + if err := m.GroupByCounts[len(m.GroupByCounts)-1].Unmarshal(dAtA[iNdEx:postIndex]); err != nil { + return err + } + iNdEx = postIndex default: iNdEx = preIndex skippy, err := skipPublic(dAtA[iNdEx:]) @@ -3748,49 +4076,53 @@ 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, + // 760 bytes of a gzipped FileDescriptorProto + 0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xac, 0x55, 0xcd, 0x6e, 0xd3, 0x4a, + 0x14, 0xbe, 0x13, 0x3b, 0x89, 0x73, 0xd2, 0xe4, 0x56, 0x73, 0xef, 0xed, 0xb5, 0x50, 0x15, 0x2c, + 0x0b, 0x21, 0xaf, 0x52, 0x29, 0xac, 0xba, 0x01, 0x91, 0xfe, 0xa0, 0xa8, 0x50, 0xc1, 0xb4, 0x14, + 0xb1, 0x74, 0x9b, 0x51, 0x6b, 0xc9, 0xf1, 0x18, 0x7b, 0xac, 0x34, 0xcf, 0xd1, 0x0d, 0x8f, 0xc0, + 0x82, 0x07, 0xe9, 0x92, 0x47, 0x80, 0xf2, 0x22, 0x68, 0xce, 0x78, 0x62, 0x27, 0x95, 0x2a, 0x16, + 0xec, 0xfc, 0x7d, 0x67, 0xe6, 0xcc, 0xf9, 0xce, 0x9f, 0x61, 0x23, 0x2d, 0xce, 0xe3, 0xe8, 0x62, + 0x98, 0x66, 0x42, 0x0a, 0xea, 0x44, 0x89, 0xe4, 0x59, 0x12, 0xc6, 0xfe, 0x47, 0xb0, 0x98, 0x98, + 0x53, 0x17, 0xda, 0x7b, 0x22, 0x2e, 0x66, 0x49, 0xee, 0x12, 0xcf, 0x0a, 0x6c, 0x66, 0x20, 0x7d, + 0x02, 0xcd, 0x97, 0x52, 0x66, 0xb9, 0xdb, 0xf0, 0xac, 0xa0, 0x3b, 0xea, 0x0f, 0xcd, 0xd5, 0xa1, + 0xa2, 0x99, 0x36, 0x52, 0x0a, 0xf6, 0x11, 0x5f, 0xe4, 0xae, 0xe5, 0x59, 0x41, 0x87, 0xe1, 0xb7, + 0xff, 0x1c, 0xec, 0xb7, 0x61, 0x94, 0xd1, 0x3e, 0x34, 0x26, 0xfb, 0x2e, 0xf1, 0x48, 0x60, 0xb3, + 0xc6, 0x64, 0x9f, 0xfe, 0x0b, 0xcd, 0x3d, 0x51, 0x24, 0xd2, 0x6d, 0x20, 0xa5, 0x01, 0xdd, 0x04, + 0xeb, 0x88, 0x2f, 0x5c, 0xcb, 0x23, 0x41, 0x87, 0xa9, 0x4f, 0x7f, 0x17, 0x3a, 0xaf, 0x32, 0x51, + 0xa4, 0xaf, 0xa3, 0x84, 0xd3, 0x2d, 0x68, 0x21, 0xd0, 0xf1, 0x75, 0x58, 0x89, 0x94, 0xb3, 0x53, + 0x21, 0xc3, 0xd8, 0x38, 0x43, 0xe0, 0x8f, 0xc0, 0x39, 0x0b, 0xe3, 0xa5, 0xe3, 0xb3, 0x30, 0xc6, + 0xf7, 0x2d, 0xa6, 0x3e, 0x57, 0x03, 0xb0, 0xca, 0x00, 0xfc, 0xf7, 0x60, 0x8d, 0x23, 0xa9, 0x8c, + 0x4c, 0xcc, 0x97, 0x01, 0x6b, 0x40, 0x1f, 0x81, 0xa3, 0x13, 0x32, 0xd9, 0x2f, 0x5f, 0x5a, 0x62, + 0xba, 0x0d, 0x9d, 0xd3, 0x68, 0xc6, 0x73, 0x19, 0xce, 0x52, 0x8c, 0xdf, 0x62, 0x15, 0xe1, 0x7f, + 0x80, 0x9e, 0x3e, 0xa9, 0x12, 0x75, 0xc2, 0xe5, 0xbd, 0x74, 0xfc, 0x5e, 0x82, 0xef, 0xa7, 0xe7, + 0x0b, 0x01, 0x5b, 0xd9, 0x8c, 0x89, 0x2c, 0x4d, 0xaa, 0x1a, 0xa7, 0x8b, 0x94, 0x97, 0x91, 0xe2, + 0x37, 0xf5, 0xa0, 0x7b, 0x22, 0xb3, 0x28, 0xb9, 0x3c, 0x0b, 0xe3, 0x82, 0x97, 0x8e, 0xea, 0x94, + 0xd2, 0x38, 0x49, 0xa4, 0x36, 0xdb, 0x28, 0x63, 0x89, 0x95, 0xc6, 0xb1, 0x10, 0xb1, 0x36, 0x36, + 0x3d, 0x12, 0x38, 0xac, 0x22, 0xe8, 0x00, 0xe0, 0x30, 0x16, 0x61, 0x79, 0xb7, 0xe5, 0x91, 0x80, + 0xb0, 0x1a, 0xe3, 0xef, 0x40, 0x5b, 0x45, 0xfa, 0x26, 0x4c, 0x2b, 0xb5, 0xe4, 0x01, 0xb5, 0xfe, + 0x2d, 0x81, 0x8d, 0x77, 0x05, 0xcf, 0x16, 0x8c, 0x7f, 0x2a, 0x78, 0x8e, 0x55, 0x41, 0x5c, 0xaa, + 0xd4, 0x40, 0x35, 0xc5, 0xc9, 0x55, 0x98, 0x4d, 0x75, 0xee, 0x6c, 0x56, 0x22, 0xa5, 0xb5, 0xca, + 0x79, 0x8e, 0x5a, 0x1d, 0x56, 0xa7, 0xd4, 0x4d, 0xc6, 0x67, 0x42, 0x1a, 0x31, 0x25, 0xa2, 0x01, + 0xfc, 0x7d, 0x70, 0x7d, 0x11, 0x17, 0x53, 0xce, 0xc4, 0x5c, 0xdf, 0x6e, 0xe1, 0x81, 0x75, 0x9a, + 0x3e, 0x85, 0x7e, 0x49, 0x99, 0xc1, 0x69, 0xe3, 0xc1, 0x35, 0xd6, 0xbf, 0x21, 0xd0, 0x2b, 0xa5, + 0xe4, 0xa9, 0x48, 0x72, 0xae, 0xea, 0x75, 0x90, 0x65, 0xa6, 0x5e, 0x07, 0x59, 0x46, 0x77, 0xa0, + 0xcd, 0x78, 0x5e, 0xc4, 0xd2, 0x34, 0xc1, 0x7f, 0x55, 0x5a, 0xcc, 0xdd, 0x22, 0x96, 0xcc, 0x9c, + 0xa2, 0x2f, 0xa0, 0xbf, 0xd2, 0x54, 0x7a, 0xf0, 0xba, 0xa3, 0xff, 0xab, 0x7b, 0x2b, 0x76, 0xb6, + 0x76, 0xdc, 0xbf, 0x69, 0x40, 0xb7, 0xe6, 0x99, 0x3e, 0xc6, 0x35, 0x80, 0x31, 0x75, 0x47, 0xbd, + 0xca, 0x0b, 0x13, 0x73, 0x86, 0x0b, 0x62, 0x03, 0xc8, 0x71, 0xd9, 0x4f, 0xe4, 0x58, 0x55, 0x51, + 0x8d, 0xb6, 0x79, 0xb6, 0x56, 0x45, 0x45, 0x33, 0x6d, 0xc4, 0xa5, 0x72, 0x15, 0x26, 0x97, 0x7c, + 0x8a, 0xfd, 0xe4, 0x30, 0x03, 0xe9, 0xb0, 0x9a, 0x4f, 0x2c, 0x40, 0x77, 0x44, 0x2b, 0x17, 0xc6, + 0xc2, 0xaa, 0x19, 0x36, 0x0d, 0xad, 0x6a, 0xd1, 0x2b, 0x1b, 0x5a, 0x95, 0x50, 0xcd, 0xa6, 0x4a, + 0x3c, 0x16, 0x5f, 0x23, 0xba, 0x0b, 0x3d, 0xdc, 0x0d, 0xe3, 0x05, 0xde, 0xcd, 0x5d, 0x07, 0x63, + 0xfc, 0xa7, 0x7a, 0x60, 0xb9, 0x55, 0xd8, 0xea, 0x49, 0xff, 0x07, 0x81, 0xde, 0x64, 0x96, 0x8a, + 0x4c, 0xd6, 0xfa, 0x6e, 0x92, 0x4c, 0xf9, 0xb5, 0xe9, 0x3b, 0x04, 0x8a, 0x3d, 0x8c, 0x78, 0x3c, + 0xc5, 0x84, 0x74, 0x98, 0x06, 0x8a, 0xc5, 0xfe, 0xc3, 0x7e, 0xb3, 0x99, 0x06, 0xb5, 0x30, 0xed, + 0x95, 0x30, 0xb7, 0xa1, 0x63, 0x36, 0x48, 0xee, 0x36, 0xd1, 0x54, 0x11, 0x6a, 0xa2, 0x96, 0x2b, + 0x44, 0xb5, 0xa0, 0x15, 0x58, 0xac, 0xc6, 0xa8, 0xd4, 0x32, 0x31, 0xc7, 0x95, 0xdb, 0xc6, 0x7d, + 0x68, 0xa0, 0xba, 0xa9, 0xdd, 0xa0, 0xd1, 0x41, 0x63, 0x8d, 0xf1, 0xbf, 0x12, 0xa0, 0x5a, 0x23, + 0xce, 0xe6, 0x9f, 0x13, 0xfa, 0xb0, 0xa0, 0x2d, 0x68, 0xe1, 0x7b, 0x46, 0x4c, 0x89, 0xd6, 0xc2, + 0x6d, 0xaf, 0x87, 0x3b, 0xde, 0xbc, 0xbd, 0x1b, 0x90, 0x6f, 0x77, 0x03, 0xf2, 0xfd, 0x6e, 0x40, + 0x3e, 0xff, 0x1c, 0xfc, 0x75, 0xde, 0xc2, 0x5f, 0xd8, 0xb3, 0x5f, 0x01, 0x00, 0x00, 0xff, 0xff, + 0x22, 0x44, 0x8c, 0x98, 0xd2, 0x06, 0x00, 0x00, } diff --git a/internal/public.proto b/internal/public.proto index 04c98d070..f3f362fd1 100644 --- a/internal/public.proto +++ b/internal/public.proto @@ -14,6 +14,11 @@ message Pair { uint64 Count = 2; } +message GroupLine{ + repeated string Groups = 1; + uint64 Total=2; +} + message ValCount { int64 Val = 1; int64 Count = 2; @@ -64,8 +69,10 @@ 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 GroupLine GroupByCounts = 8; } message ImportRequest { diff --git a/pilosa.go b/pilosa.go index 7378eb628..96b06d9e6 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