mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-05 16:15:56 +00:00
Merge branch 'master' into bnm-fix-1080
This commit is contained in:
nagamocha3000
GitHub
commit
16b5fbe40b
33 changed files with 1067 additions and 735 deletions
2
api.go
2
api.go
|
|
@ -161,7 +161,7 @@ func (api *API) Query(ctx context.Context, req *QueryRequest) (QueryResponse, er
|
|||
}
|
||||
|
||||
if !req.Remote {
|
||||
defer api.tracker.Finish(api.tracker.Start(req.Query, api.server.nodeID, req.Index, start))
|
||||
defer api.tracker.Finish(api.tracker.Start(req.Query, req.SQLQuery, api.server.nodeID, req.Index, start))
|
||||
}
|
||||
// TODO can we get rid of exec options and pass the QueryRequest directly to executor?
|
||||
execOpts := &execOptions{
|
||||
|
|
|
|||
|
|
@ -536,7 +536,7 @@ func (s Serializer) encodeQueryResponse(m *pilosa.QueryResponse) *internal.Query
|
|||
case pilosa.ExtractedIDMatrix:
|
||||
pb.Results[i].Type = queryResultTypeExtractedIDMatrix
|
||||
pb.Results[i].ExtractedIDMatrix = s.endcodeExtractedIDMatrix(result)
|
||||
case []pilosa.GroupCount:
|
||||
case *pilosa.GroupCounts:
|
||||
pb.Results[i].Type = queryResultTypeGroupCounts
|
||||
pb.Results[i].GroupCounts = s.encodeGroupCounts(result)
|
||||
case pilosa.RowIdentifiers:
|
||||
|
|
@ -1423,7 +1423,7 @@ func (s Serializer) decodeQueryResult(pb *internal.QueryResult) interface{} {
|
|||
case queryResultTypeRowIdentifiers:
|
||||
return s.decodeRowIdentifiers(pb.RowIdentifiers)
|
||||
case queryResultTypeGroupCounts:
|
||||
return s.decodeGroupCounts(pb.GroupCounts)
|
||||
return s.decodeGroupCounts(pb.GroupCounts, pb.OldGroupCounts)
|
||||
case queryResultTypePair:
|
||||
return s.decodePair(pb.Pairs[0])
|
||||
case queryResultTypePairField:
|
||||
|
|
@ -1587,16 +1587,22 @@ func (s Serializer) decodeRowIdentifiers(a *internal.RowIdentifiers) *pilosa.Row
|
|||
}
|
||||
}
|
||||
|
||||
func (s Serializer) decodeGroupCounts(a []*internal.GroupCount) []pilosa.GroupCount {
|
||||
other := make([]pilosa.GroupCount, len(a))
|
||||
for i := range a {
|
||||
func (s Serializer) decodeGroupCounts(a *internal.GroupCounts, b []*internal.GroupCount) *pilosa.GroupCounts {
|
||||
// Workaround: If we get an old-style "[]*GroupCount", we translate it.
|
||||
if a == nil {
|
||||
a = &internal.GroupCounts{Aggregate: "", Groups: b}
|
||||
}
|
||||
other := make([]pilosa.GroupCount, len(a.Groups))
|
||||
for i, gc := range a.Groups {
|
||||
other[i] = pilosa.GroupCount{
|
||||
Group: s.decodeFieldRows(a[i].Group),
|
||||
Count: a[i].Count,
|
||||
Sum: a[i].Sum,
|
||||
Group: s.decodeFieldRows(gc.Group),
|
||||
Count: gc.Count,
|
||||
// note: not renaming the `internal` structure members now
|
||||
// to avoid breaking protobuf interactions.
|
||||
Agg: gc.Agg,
|
||||
}
|
||||
}
|
||||
return other
|
||||
return pilosa.NewGroupCounts(a.Aggregate, other...)
|
||||
}
|
||||
|
||||
func (s Serializer) decodeFieldRows(a []*internal.FieldRow) []pilosa.FieldRow {
|
||||
|
|
@ -1723,13 +1729,17 @@ func (s Serializer) encodeRowIdentifiers(r pilosa.RowIdentifiers) *internal.RowI
|
|||
}
|
||||
}
|
||||
|
||||
func (s Serializer) encodeGroupCounts(counts []pilosa.GroupCount) []*internal.GroupCount {
|
||||
result := make([]*internal.GroupCount, len(counts))
|
||||
for i := range counts {
|
||||
result[i] = &internal.GroupCount{
|
||||
Group: s.encodeFieldRows(counts[i].Group),
|
||||
Count: counts[i].Count,
|
||||
Sum: counts[i].Sum,
|
||||
func (s Serializer) encodeGroupCounts(counts *pilosa.GroupCounts) *internal.GroupCounts {
|
||||
groups := counts.Groups()
|
||||
result := &internal.GroupCounts{
|
||||
Groups: make([]*internal.GroupCount, len(groups)),
|
||||
Aggregate: counts.AggregateColumn(),
|
||||
}
|
||||
for i, gc := range groups {
|
||||
result.Groups[i] = &internal.GroupCount{
|
||||
Group: s.encodeFieldRows(gc.Group),
|
||||
Count: gc.Count,
|
||||
Agg: gc.Agg,
|
||||
}
|
||||
}
|
||||
return result
|
||||
|
|
|
|||
209
executor.go
209
executor.go
|
|
@ -24,6 +24,7 @@ import (
|
|||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
"unsafe"
|
||||
|
||||
"github.com/pilosa/pilosa/v2/pql"
|
||||
pb "github.com/pilosa/pilosa/v2/proto"
|
||||
|
|
@ -312,6 +313,8 @@ func (e *executor) safeCopy(resp QueryResponse) (out QueryResponse) {
|
|||
out.Results = append(out.Results, x)
|
||||
case []GroupCount:
|
||||
out.Results = append(out.Results, x)
|
||||
case *GroupCounts:
|
||||
out.Results = append(out.Results, x)
|
||||
case ExtractedTable:
|
||||
out.Results = append(out.Results, x)
|
||||
case ExtractedIDMatrix:
|
||||
|
|
@ -2653,10 +2656,10 @@ func (g *groupCountSorter) Less(i, j int) bool {
|
|||
} else if gci.Count > gcj.Count {
|
||||
return fieldOrder == desc
|
||||
}
|
||||
case -2: // aggregate/Sum
|
||||
if gci.Sum < gcj.Sum {
|
||||
case -2: // Aggregate
|
||||
if gci.Agg < gcj.Agg {
|
||||
return fieldOrder == asc
|
||||
} else if gci.Sum > gcj.Sum {
|
||||
} else if gci.Agg > gcj.Agg {
|
||||
return fieldOrder == desc
|
||||
}
|
||||
default:
|
||||
|
|
@ -2702,7 +2705,19 @@ func getSorter(sortSpec string) (*groupCountSorter, error) {
|
|||
return gcs, nil
|
||||
}
|
||||
|
||||
func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) ([]GroupCount, error) {
|
||||
// findGroupCounts gets a safe-to-use but possibly empty []GroupCount from
|
||||
// an interface which might be a *GroupCounts or a []GroupCount.
|
||||
func findGroupCounts(v interface{}) []GroupCount {
|
||||
switch gc := v.(type) {
|
||||
case []GroupCount:
|
||||
return gc
|
||||
case *GroupCounts:
|
||||
return gc.Groups()
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shards []uint64, opt *execOptions) (*GroupCounts, error) {
|
||||
span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeGroupBy")
|
||||
defer span.Finish()
|
||||
// validate call
|
||||
|
|
@ -2787,7 +2802,7 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
|
|||
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
|
||||
return &GroupCounts{}, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -2798,11 +2813,11 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
|
|||
}
|
||||
// Merge returned results at coordinating node.
|
||||
reduceFn := func(ctx context.Context, prev, v interface{}) interface{} {
|
||||
other, _ := prev.([]GroupCount)
|
||||
other := findGroupCounts(prev)
|
||||
if err := ctx.Err(); err != nil {
|
||||
return err
|
||||
}
|
||||
return mergeGroupCounts(other, v.([]GroupCount), limit)
|
||||
return mergeGroupCounts(other, findGroupCounts(v), limit)
|
||||
}
|
||||
// Get full result set.
|
||||
other, err := e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
|
||||
|
|
@ -2869,7 +2884,7 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
|
|||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
results[n].Sum = int64(aggregateCount[0].(uint64))
|
||||
results[n].Agg = int64(aggregateCount[0].(uint64))
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -2907,7 +2922,16 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
|
|||
|
||||
}
|
||||
|
||||
return results, nil
|
||||
aggType := ""
|
||||
if aggregate != nil {
|
||||
switch aggregate.Name {
|
||||
case "Sum":
|
||||
aggType = "sum"
|
||||
case "Count":
|
||||
aggType = "aggregate"
|
||||
}
|
||||
}
|
||||
return NewGroupCounts(aggType, results...), nil
|
||||
}
|
||||
|
||||
func applyLimitAndOffsetToGroupByResult(c *pql.Call, results []GroupCount) ([]GroupCount, error) {
|
||||
|
|
@ -2992,17 +3016,70 @@ func (fr FieldRow) String() string {
|
|||
return fmt.Sprintf("%s.%d.%s", fr.Field, fr.RowID, fr.RowKey)
|
||||
}
|
||||
|
||||
type aggregateType int
|
||||
|
||||
const (
|
||||
nilAggregate aggregateType = 0
|
||||
sumAggregate aggregateType = 1
|
||||
distinctAggregate aggregateType = 2
|
||||
)
|
||||
|
||||
// GroupCounts is a list of GroupCount.
|
||||
type GroupCounts []GroupCount
|
||||
type GroupCounts struct {
|
||||
groups []GroupCount
|
||||
aggregateType aggregateType
|
||||
}
|
||||
|
||||
// AggregateColumn gives the likely column name to use for aggregates, because
|
||||
// for historical reasons we used "sum" when it was a sum, but don't want to
|
||||
// use that when it's something else. This will likely get revisited.
|
||||
func (g *GroupCounts) AggregateColumn() string {
|
||||
switch g.aggregateType {
|
||||
case sumAggregate:
|
||||
return "sum"
|
||||
case distinctAggregate:
|
||||
return "aggregate"
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
// Groups is a convenience method to let us not worry as much about the
|
||||
// potentially-nil nature of a *GroupCounts.
|
||||
func (g *GroupCounts) Groups() []GroupCount {
|
||||
if g == nil {
|
||||
return nil
|
||||
}
|
||||
return g.groups
|
||||
}
|
||||
|
||||
// NewGroupCounts creates a GroupCounts with the given type and slice
|
||||
// of GroupCount objects. There's intentionally no externally-accessible way
|
||||
// to change the []GroupCount after creation.
|
||||
func NewGroupCounts(agg string, groups ...GroupCount) *GroupCounts {
|
||||
var aggType aggregateType
|
||||
switch agg {
|
||||
case "sum":
|
||||
aggType = sumAggregate
|
||||
case "aggregate":
|
||||
aggType = distinctAggregate
|
||||
case "":
|
||||
aggType = nilAggregate
|
||||
default:
|
||||
panic(fmt.Sprintf("invalid aggregate type %q", agg))
|
||||
}
|
||||
return &GroupCounts{aggregateType: aggType, groups: groups}
|
||||
}
|
||||
|
||||
// ToTable implements the ToTabler interface.
|
||||
func (g GroupCounts) ToTable() (*pb.TableResponse, error) {
|
||||
return pb.RowsToTable(&g, len(g))
|
||||
func (g *GroupCounts) ToTable() (*pb.TableResponse, error) {
|
||||
return pb.RowsToTable(g, len(g.Groups()))
|
||||
}
|
||||
|
||||
// ToRows implements the ToRowser interface.
|
||||
func (g GroupCounts) ToRows(callback func(*pb.RowResponse) error) error {
|
||||
for i, gc := range g {
|
||||
func (g *GroupCounts) ToRows(callback func(*pb.RowResponse) error) error {
|
||||
agg := g.AggregateColumn()
|
||||
for i, gc := range g.Groups() {
|
||||
var ci []*pb.ColumnInfo
|
||||
if i == 0 {
|
||||
for _, fieldRow := range gc.Group {
|
||||
|
|
@ -3015,7 +3092,10 @@ func (g GroupCounts) ToRows(callback func(*pb.RowResponse) error) error {
|
|||
}
|
||||
}
|
||||
ci = append(ci, &pb.ColumnInfo{Name: "count", Datatype: "uint64"})
|
||||
ci = append(ci, &pb.ColumnInfo{Name: "sum", Datatype: "int64"})
|
||||
if agg != "" {
|
||||
ci = append(ci, &pb.ColumnInfo{Name: agg, Datatype: "int64"})
|
||||
}
|
||||
|
||||
}
|
||||
rowResp := &pb.RowResponse{
|
||||
Headers: ci,
|
||||
|
|
@ -3032,9 +3112,11 @@ func (g GroupCounts) ToRows(callback func(*pb.RowResponse) error) error {
|
|||
}
|
||||
}
|
||||
rowResp.Columns = append(rowResp.Columns,
|
||||
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: gc.Count}},
|
||||
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: gc.Sum}},
|
||||
)
|
||||
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Uint64Val{Uint64Val: gc.Count}})
|
||||
if agg != "" {
|
||||
rowResp.Columns = append(rowResp.Columns,
|
||||
&pb.ColumnResponse{ColumnVal: &pb.ColumnResponse_Int64Val{Int64Val: gc.Agg}})
|
||||
}
|
||||
if err := callback(rowResp); err != nil {
|
||||
return errors.Wrap(err, "calling callback")
|
||||
}
|
||||
|
|
@ -3042,18 +3124,51 @@ func (g GroupCounts) ToRows(callback func(*pb.RowResponse) error) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// MarshalJSON makes GroupCounts satisfy interface json.Marshaler and
|
||||
// customizes the JSON output of the aggregate field label.
|
||||
func (g *GroupCounts) MarshalJSON() ([]byte, error) {
|
||||
groups := g.Groups()
|
||||
var counts interface{} = groups
|
||||
|
||||
if len(groups) == 0 {
|
||||
return []byte("[]"), nil
|
||||
}
|
||||
switch g.aggregateType {
|
||||
case sumAggregate:
|
||||
counts = *(*[]groupCountSum)(unsafe.Pointer(&groups))
|
||||
case distinctAggregate:
|
||||
counts = *(*[]groupCountAggregate)(unsafe.Pointer(&groups))
|
||||
}
|
||||
return json.Marshal(counts)
|
||||
}
|
||||
|
||||
// GroupCount represents a result item for a group by query.
|
||||
type GroupCount struct {
|
||||
Group []FieldRow `json:"group"`
|
||||
Count uint64 `json:"count"`
|
||||
Sum int64 `json:"sum"`
|
||||
Agg int64 `json:"-"`
|
||||
}
|
||||
|
||||
type groupCountSum struct {
|
||||
Group []FieldRow `json:"group"`
|
||||
Count uint64 `json:"count"`
|
||||
Agg int64 `json:"sum"`
|
||||
}
|
||||
|
||||
type groupCountAggregate struct {
|
||||
Group []FieldRow `json:"group"`
|
||||
Count uint64 `json:"count"`
|
||||
Agg int64 `json:"aggregate"`
|
||||
}
|
||||
|
||||
var _ GroupCount = GroupCount(groupCountSum{})
|
||||
var _ GroupCount = GroupCount(groupCountAggregate{})
|
||||
|
||||
func (g *GroupCount) Clone() (r *GroupCount) {
|
||||
r = &GroupCount{
|
||||
Group: make([]FieldRow, len(g.Group)),
|
||||
Count: g.Count,
|
||||
Sum: g.Sum,
|
||||
Agg: g.Agg,
|
||||
}
|
||||
for i := range g.Group {
|
||||
r.Group[i] = *(g.Group[i].Clone())
|
||||
|
|
@ -3077,7 +3192,7 @@ func mergeGroupCounts(a, b []GroupCount, limit int) []GroupCount {
|
|||
i++
|
||||
case 0:
|
||||
a[i].Count += b[j].Count
|
||||
a[i].Sum += b[j].Sum
|
||||
a[i].Agg += b[j].Agg
|
||||
ret = append(ret, a[i])
|
||||
i++
|
||||
j++
|
||||
|
|
@ -3184,27 +3299,27 @@ func (g GroupCount) satisfiesCondition(subj string, cond *pql.Condition) bool {
|
|||
return false
|
||||
}
|
||||
if cond.Op == pql.EQ {
|
||||
if g.Sum == val {
|
||||
if g.Agg == val {
|
||||
return true
|
||||
}
|
||||
} else if cond.Op == pql.NEQ {
|
||||
if g.Sum != val {
|
||||
if g.Agg != val {
|
||||
return true
|
||||
}
|
||||
} else if cond.Op == pql.LT {
|
||||
if g.Sum < val {
|
||||
if g.Agg < val {
|
||||
return true
|
||||
}
|
||||
} else if cond.Op == pql.LTE {
|
||||
if g.Sum <= val {
|
||||
if g.Agg <= val {
|
||||
return true
|
||||
}
|
||||
} else if cond.Op == pql.GT {
|
||||
if g.Sum > val {
|
||||
if g.Agg > val {
|
||||
return true
|
||||
}
|
||||
} else if cond.Op == pql.GTE {
|
||||
if g.Sum >= val {
|
||||
if g.Agg >= val {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
|
@ -3214,19 +3329,19 @@ func (g GroupCount) satisfiesCondition(subj string, cond *pql.Condition) bool {
|
|||
return false
|
||||
}
|
||||
if cond.Op == pql.BETWEEN {
|
||||
if val[0] <= g.Sum && g.Sum <= val[1] {
|
||||
if val[0] <= g.Agg && g.Agg <= val[1] {
|
||||
return true
|
||||
}
|
||||
} else if cond.Op == pql.BTWN_LT_LTE {
|
||||
if val[0] < g.Sum && g.Sum <= val[1] {
|
||||
if val[0] < g.Agg && g.Agg <= val[1] {
|
||||
return true
|
||||
}
|
||||
} else if cond.Op == pql.BTWN_LTE_LT {
|
||||
if val[0] <= g.Sum && g.Sum < val[1] {
|
||||
if val[0] <= g.Agg && g.Agg < val[1] {
|
||||
return true
|
||||
}
|
||||
} else if cond.Op == pql.BTWN_LT_LT {
|
||||
if val[0] < g.Sum && g.Sum < val[1] {
|
||||
if val[0] < g.Agg && g.Agg < val[1] {
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
|
@ -6497,10 +6612,11 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
|
|||
}
|
||||
}
|
||||
|
||||
case []GroupCount:
|
||||
case *GroupCounts:
|
||||
fieldIDs := make(map[*Field]map[uint64]struct{})
|
||||
foreignIDs := make(map[*Field]map[uint64]struct{})
|
||||
for _, gl := range result {
|
||||
groups := result.Groups()
|
||||
for _, gl := range groups {
|
||||
for _, g := range gl.Group {
|
||||
field := idx.Field(g.Field)
|
||||
if field == nil {
|
||||
|
|
@ -6511,7 +6627,7 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
|
|||
if fi := field.ForeignIndex(); fi != "" {
|
||||
m, ok := foreignIDs[field]
|
||||
if !ok {
|
||||
m = make(map[uint64]struct{}, len(result))
|
||||
m = make(map[uint64]struct{}, len(groups))
|
||||
foreignIDs[field] = m
|
||||
}
|
||||
|
||||
|
|
@ -6522,7 +6638,7 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
|
|||
|
||||
m, ok := fieldIDs[field]
|
||||
if !ok {
|
||||
m = make(map[uint64]struct{}, len(result))
|
||||
m = make(map[uint64]struct{}, len(groups))
|
||||
fieldIDs[field] = m
|
||||
}
|
||||
|
||||
|
|
@ -6549,8 +6665,11 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
|
|||
foreignTranslations[field.Name()] = trans
|
||||
}
|
||||
|
||||
other := make([]GroupCount, 0)
|
||||
for _, gl := range result {
|
||||
// We are reluctant to smash result, and I'm not sure we need
|
||||
// to be but I'm not sure we don't need to be.
|
||||
newGroups := make([]GroupCount, len(groups))
|
||||
copy(newGroups, groups)
|
||||
for gi, gl := range groups {
|
||||
|
||||
group := make([]FieldRow, len(gl.Group))
|
||||
for i, g := range gl.Group {
|
||||
|
|
@ -6563,15 +6682,15 @@ func (e *executor) translateResult(ctx context.Context, index string, idx *Index
|
|||
|
||||
group[i] = g
|
||||
}
|
||||
|
||||
other = append(other, GroupCount{
|
||||
Group: group,
|
||||
Count: gl.Count,
|
||||
Sum: gl.Sum,
|
||||
})
|
||||
// Replace with translated group.
|
||||
newGroups[gi].Group = group
|
||||
}
|
||||
other := &GroupCounts{}
|
||||
if result != nil {
|
||||
other.aggregateType = result.aggregateType
|
||||
}
|
||||
other.groups = newGroups
|
||||
return other, nil
|
||||
|
||||
case RowIDs:
|
||||
fieldName := callArgString(call, "_field")
|
||||
if fieldName == "" {
|
||||
|
|
@ -7494,7 +7613,7 @@ func (gbi *groupByIterator) Next(ctx context.Context) (ret GroupCount, done bool
|
|||
return ret, false, err
|
||||
}
|
||||
ret.Count = uint64(result.Count)
|
||||
ret.Sum = result.Val
|
||||
ret.Agg = result.Val
|
||||
}
|
||||
}
|
||||
if ret.Count == 0 {
|
||||
|
|
|
|||
|
|
@ -164,7 +164,7 @@ func TestExecutor_GroupCountCondition(t *testing.T) {
|
|||
},
|
||||
},
|
||||
{
|
||||
groupCount: GroupCount{Sum: 100},
|
||||
groupCount: GroupCount{Agg: 100},
|
||||
checks: []condCheck{
|
||||
{cond: "sum == 99", exp: false},
|
||||
{cond: "sum != 99", exp: true},
|
||||
|
|
@ -196,7 +196,7 @@ func TestExecutor_GroupCountCondition(t *testing.T) {
|
|||
},
|
||||
},
|
||||
{
|
||||
groupCount: GroupCount{Sum: -100},
|
||||
groupCount: GroupCount{Agg: -100},
|
||||
checks: []condCheck{
|
||||
{cond: "sum == -99", exp: false},
|
||||
{cond: "sum != -99", exp: true},
|
||||
|
|
|
|||
187
executor_test.go
187
executor_test.go
|
|
@ -3066,7 +3066,7 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "f", RowID: 7}}, Count: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "f", RowID: 10}}, Count: 4},
|
||||
}
|
||||
results := res.Results[0].([]pilosa.GroupCount)
|
||||
results := res.Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
}
|
||||
})
|
||||
|
|
@ -3108,7 +3108,7 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "fint", Value: &d}}, Count: 1},
|
||||
}
|
||||
|
||||
results := res.Results[0].([]pilosa.GroupCount)
|
||||
results := res.Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
}
|
||||
})
|
||||
|
|
@ -3139,7 +3139,7 @@ func TestExecutor_Execute_Remote_Row(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "hint", Value: &c}}, Count: 1},
|
||||
}
|
||||
|
||||
results := res.Results[0].([]pilosa.GroupCount)
|
||||
results := res.Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
}
|
||||
})
|
||||
|
|
@ -5074,20 +5074,20 @@ func TestExecutor_GroupByStrings(t *testing.T) {
|
|||
{
|
||||
query: "GroupBy(Rows(generals), aggregate=Sum(field=v))",
|
||||
expected: []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "generals", RowID: 1, RowKey: "r1"}}, Count: 5, Sum: 25},
|
||||
{Group: []pilosa.FieldRow{{Field: "generals", RowID: 2, RowKey: "r2"}}, Count: 5, Sum: 30},
|
||||
{Group: []pilosa.FieldRow{{Field: "generals", RowID: 1, RowKey: "r1"}}, Count: 5, Agg: 25},
|
||||
{Group: []pilosa.FieldRow{{Field: "generals", RowID: 2, RowKey: "r2"}}, Count: 5, Agg: 30},
|
||||
},
|
||||
},
|
||||
{
|
||||
query: "GroupBy(Rows(generals), aggregate=Sum(field=v), having=Condition(sum>25))",
|
||||
expected: []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "generals", RowID: 2, RowKey: "r2"}}, Count: 5, Sum: 30},
|
||||
{Group: []pilosa.FieldRow{{Field: "generals", RowID: 2, RowKey: "r2"}}, Count: 5, Agg: 30},
|
||||
},
|
||||
},
|
||||
{
|
||||
query: "GroupBy(Rows(generals), aggregate=Sum(field=v), having=Condition(-5<sum<27))",
|
||||
expected: []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "generals", RowID: 1, RowKey: "r1"}}, Count: 5, Sum: 25},
|
||||
{Group: []pilosa.FieldRow{{Field: "generals", RowID: 1, RowKey: "r1"}}, Count: 5, Agg: 25},
|
||||
},
|
||||
},
|
||||
{
|
||||
|
|
@ -5100,52 +5100,52 @@ func TestExecutor_GroupByStrings(t *testing.T) {
|
|||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v1}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v2}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v3}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v4}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v5}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v6}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v7}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v8}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v9}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "v", Value: &v10}},
|
||||
Count: 1,
|
||||
Sum: 0,
|
||||
Agg: 0,
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
@ -5155,12 +5155,12 @@ func TestExecutor_GroupByStrings(t *testing.T) {
|
|||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "vv", Value: &v3}},
|
||||
Count: 3,
|
||||
Sum: 9,
|
||||
Agg: 9,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "vv", Value: &v4}},
|
||||
Count: 4,
|
||||
Sum: 16,
|
||||
Agg: 16,
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
@ -5170,12 +5170,12 @@ func TestExecutor_GroupByStrings(t *testing.T) {
|
|||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "nv", Value: &nv4}},
|
||||
Count: 4,
|
||||
Sum: -16,
|
||||
Agg: -16,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "nv", Value: &nv3}},
|
||||
Count: 3,
|
||||
Sum: -9,
|
||||
Agg: -9,
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
@ -5185,12 +5185,12 @@ func TestExecutor_GroupByStrings(t *testing.T) {
|
|||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "nv", Value: &nv4}},
|
||||
Count: 4,
|
||||
Sum: -16,
|
||||
Agg: -16,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{pilosa.FieldRow{Field: "nv", Value: &nv3}},
|
||||
Count: 3,
|
||||
Sum: -9,
|
||||
Agg: -9,
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
@ -5203,7 +5203,7 @@ func TestExecutor_GroupByStrings(t *testing.T) {
|
|||
pilosa.FieldRow{Field: "nv", Value: &nv3},
|
||||
},
|
||||
Count: 3,
|
||||
Sum: 9,
|
||||
Agg: 9,
|
||||
},
|
||||
{
|
||||
Group: []pilosa.FieldRow{
|
||||
|
|
@ -5211,7 +5211,7 @@ func TestExecutor_GroupByStrings(t *testing.T) {
|
|||
pilosa.FieldRow{Field: "nv", Value: &nv4},
|
||||
},
|
||||
Count: 4,
|
||||
Sum: 16,
|
||||
Agg: 16,
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
@ -5226,7 +5226,7 @@ func TestExecutor_GroupByStrings(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
results := r.Results[0].([]pilosa.GroupCount)
|
||||
results := r.Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, tst.expected, results)
|
||||
})
|
||||
}
|
||||
|
|
@ -5487,6 +5487,34 @@ func sameStringSlice(x, y []string) bool {
|
|||
return len(diff) == 0
|
||||
}
|
||||
|
||||
func TestExecutor_Execute_DistinctFailure(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "general")
|
||||
c.CreateField(t, "i", pilosa.IndexOptions{}, "v", pilosa.OptFieldTypeInt(0, 1000))
|
||||
c.ImportBits(t, "i", "general", [][2]uint64{
|
||||
{10, 0},
|
||||
{10, 1},
|
||||
{10, ShardWidth + 1},
|
||||
{11, 2},
|
||||
{11, ShardWidth + 2},
|
||||
{12, 2},
|
||||
{12, ShardWidth + 2},
|
||||
})
|
||||
|
||||
if _, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(0, v=10)`}); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if _, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Set(1, v=100)`}); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
t.Run("BasicDistinct", func(t *testing.T) {
|
||||
if _, err := c.GetNode(0).API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `Distinct(field="v")`}); err != nil {
|
||||
t.Fatalf("unexpected error: \"%v\"", err)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
func TestExecutor_Execute_GroupBy(t *testing.T) {
|
||||
groupByTest := func(t *testing.T, clusterSize int) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
|
|
@ -5547,7 +5575,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}, {Field: "sub", RowID: 110}}, Count: 1},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=general), Rows(sub))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(field=general), Rows(sub))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
|
|
@ -5559,7 +5587,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}, {Field: "sub", RowID: 110}}, Count: 1},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
|
|
@ -5569,50 +5597,50 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), filter=Row(general=10))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), filter=Row(general=10))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
t.Run("Aggregate", func(t *testing.T) {
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 2, Sum: 110},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1, Sum: 10},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 2, Agg: 110},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1, Agg: 10},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), aggregate=Sum(field=v))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), aggregate=Sum(field=v))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
t.Run("AggregateCountDistinct", func(t *testing.T) {
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 3, Sum: 2},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1, Sum: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}, {Field: "sub", RowID: 110}}, Count: 1, Sum: 0},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}, {Field: "sub", RowID: 110}}, Count: 1, Sum: 0},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 3, Agg: 2},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1, Agg: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}, {Field: "sub", RowID: 110}}, Count: 1, Agg: 0},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}, {Field: "sub", RowID: 110}}, Count: 1, Agg: 0},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), aggregate=Count(Distinct(field=v)))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), aggregate=Count(Distinct(field=v)))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
t.Run("AggregateCountDistinctFilter", func(t *testing.T) {
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 1, Sum: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 1, Agg: 1},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), filter=Row(v > 10), aggregate=Count(Distinct(field=v)))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), filter=Row(v > 10), aggregate=Count(Distinct(field=v)))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
t.Run("AggregateCountDistinctFilterDistinct", func(t *testing.T) {
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 3, Sum: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1, Sum: 0},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}, {Field: "sub", RowID: 110}}, Count: 1, Sum: 0},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}, {Field: "sub", RowID: 110}}, Count: 1, Sum: 0},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 3, Agg: 1},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1, Agg: 0},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}, {Field: "sub", RowID: 110}}, Count: 1, Agg: 0},
|
||||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}, {Field: "sub", RowID: 110}}, Count: 1, Agg: 0},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), aggregate=Count(Distinct(Row(v > 10), field=v)))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general), Rows(sub), aggregate=Count(Distinct(Row(v > 10), field=v)))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
|
|
@ -5622,7 +5650,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 12}}, Count: 2},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general, previous=10))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general, previous=10))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
|
|
@ -5631,7 +5659,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}}, Count: 2},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general, previous=10), limit=1)`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(general, previous=10), limit=1)`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
|
||||
})
|
||||
|
|
@ -5652,7 +5680,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "a", RowID: 0}, {Field: "b", RowID: 1}}, Count: 1},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(a), Rows(b), limit=1)`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(a), Rows(b), limit=1)`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
|
|
@ -5680,7 +5708,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("test wrapping with previous", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(wa), Rows(wb), Rows(wc, previous=1), limit=3)`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(wa), Rows(wb), Rows(wc, previous=1), limit=3)`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
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},
|
||||
|
|
@ -5690,14 +5718,14 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("test previous is last result", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(wa, previous=3), Rows(wb, previous=3), Rows(wc, previous=3), limit=3)`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(wa, previous=3), Rows(wb, previous=3), Rows(wc, previous=3), limit=3)`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
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(wa), Rows(wb, previous=2), Rows(wc, previous=2), limit=1)`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(wa), Rows(wb, previous=2), Rows(wc, previous=2), limit=1)`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
expected := []pilosa.GroupCount{
|
||||
{Group: []pilosa.FieldRow{{Field: "wa", RowID: 1}, {Field: "wb", RowID: 0}, {Field: "wc", RowID: 0}}, Count: 1},
|
||||
}
|
||||
|
|
@ -5721,7 +5749,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{3, ShardWidth},
|
||||
})
|
||||
t.Run("distinct rows in different shards", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(ma), Rows(mb), limit=5)`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(ma), Rows(mb), limit=5)`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
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},
|
||||
|
|
@ -5733,7 +5761,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("distinct rows in different shards with row limit", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(ma), Rows(mb, limit=2), limit=5)`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(ma), Rows(mb, limit=2), limit=5)`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
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},
|
||||
|
|
@ -5744,7 +5772,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("distinct rows in different shards with column arg", func(t *testing.T) {
|
||||
results := c.Query(t, "i", fmt.Sprintf(`GroupBy(Rows(ma), Rows(mb, column=%d), limit=5)`, ShardWidth)).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", fmt.Sprintf(`GroupBy(Rows(ma), Rows(mb, column=%d), limit=5)`, ShardWidth)).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
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},
|
||||
|
|
@ -5769,7 +5797,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{1, ShardWidth},
|
||||
})
|
||||
t.Run("same rows in different shards", func(t *testing.T) {
|
||||
results := c.Query(t, "i", `GroupBy(Rows(na), Rows(nb))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(na), Rows(nb))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
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},
|
||||
|
|
@ -5806,12 +5834,12 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
|
||||
t.Run("test wrapping with previous", func(t *testing.T) {
|
||||
totalResults := make([]pilosa.GroupCount, 0)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(ppa), Rows(ppb), Rows(ppc), limit=3)`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(ppa), Rows(ppb), Rows(ppc), limit=3)`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
totalResults = append(totalResults, results...)
|
||||
for len(totalResults) < 64 {
|
||||
lastGroup := results[len(results)-1].Group
|
||||
query := fmt.Sprintf("GroupBy(Rows(ppa, previous=%d), Rows(ppb, previous=%d), Rows(ppc, previous=%d), limit=3)", lastGroup[0].RowID, lastGroup[1].RowID, lastGroup[2].RowID)
|
||||
results = c.Query(t, "i", query).Results[0].([]pilosa.GroupCount)
|
||||
results = c.Query(t, "i", query).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
totalResults = append(totalResults, results...)
|
||||
}
|
||||
|
||||
|
|
@ -5853,7 +5881,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "generalk", RowID: 3, RowKey: "twelve"}, {Field: "subk", RowID: 2, RowKey: "one-hundred-ten"}}, Count: 1},
|
||||
}
|
||||
|
||||
results := c.Query(t, "i", `GroupBy(Rows(generalk), Rows(subk))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "i", `GroupBy(Rows(generalk), Rows(subk))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
|
||||
})
|
||||
|
|
@ -5891,7 +5919,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
|
|||
{Group: []pilosa.FieldRow{{Field: "child", RowID: 2, RowKey: "three"}}, Count: 2},
|
||||
}
|
||||
|
||||
results := c.Query(t, "fic", `GroupBy(Rows(child))`).Results[0].([]pilosa.GroupCount)
|
||||
results := c.Query(t, "fic", `GroupBy(Rows(child))`).Results[0].(*pilosa.GroupCounts).Groups()
|
||||
test.CheckGroupBy(t, expected, results)
|
||||
})
|
||||
|
||||
|
|
@ -6477,10 +6505,11 @@ func TestExecutor_Execute_CountDistinct(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
gc, ok := resp.Results[0].([]pilosa.GroupCount)
|
||||
gcc, ok := resp.Results[0].(*pilosa.GroupCounts)
|
||||
if !ok {
|
||||
t.Fatalf("invalid response type, expected: []pilosa.GroupCount, got: %T", resp.Results[0])
|
||||
}
|
||||
gc := gcc.Groups()
|
||||
if len(gc) != 2 {
|
||||
t.Fatalf("invalid group count length, expected: 2, got: %v", len(gc))
|
||||
}
|
||||
|
|
@ -7095,13 +7124,13 @@ toronto,2,11
|
|||
},
|
||||
{
|
||||
query: "GroupBy(Rows(field=likes))",
|
||||
csvVerifier: `molecula,1,0
|
||||
pilosa,1,0
|
||||
pangolin,1,0
|
||||
zebra,1,0
|
||||
toucan,1,0
|
||||
dog,1,0
|
||||
icecream,6,0
|
||||
csvVerifier: `molecula,1
|
||||
pilosa,1
|
||||
pangolin,1
|
||||
zebra,1
|
||||
toucan,1
|
||||
dog,1
|
||||
icecream,6
|
||||
`,
|
||||
},
|
||||
{
|
||||
|
|
@ -7112,15 +7141,15 @@ zebra,1,1000
|
|||
},
|
||||
{
|
||||
query: "GroupBy(Rows(field=likes), having=Condition(count>5))",
|
||||
csvVerifier: "icecream,6,0\n",
|
||||
csvVerifier: "icecream,6\n",
|
||||
},
|
||||
{
|
||||
query: "GroupBy(Rows(field=likes), filter=Row(affinity>-7))",
|
||||
csvVerifier: `molecula,1,0
|
||||
pangolin,1,0
|
||||
zebra,1,0
|
||||
toucan,1,0
|
||||
icecream,4,0
|
||||
csvVerifier: `molecula,1
|
||||
pangolin,1
|
||||
zebra,1
|
||||
toucan,1
|
||||
icecream,4
|
||||
`,
|
||||
},
|
||||
{
|
||||
|
|
@ -7160,13 +7189,13 @@ icecream,5,3
|
|||
},
|
||||
{
|
||||
query: "GroupBy(Rows(field=likes), sort=\"count desc\")",
|
||||
csvVerifier: `icecream,6,0
|
||||
molecula,1,0
|
||||
pilosa,1,0
|
||||
pangolin,1,0
|
||||
zebra,1,0
|
||||
toucan,1,0
|
||||
dog,1,0
|
||||
csvVerifier: `icecream,6
|
||||
molecula,1
|
||||
pilosa,1
|
||||
pangolin,1
|
||||
zebra,1
|
||||
toucan,1
|
||||
dog,1
|
||||
`,
|
||||
},
|
||||
{
|
||||
|
|
|
|||
|
|
@ -29,6 +29,9 @@ type QueryRequest struct {
|
|||
// The query string to parse and execute.
|
||||
Query string
|
||||
|
||||
// The SQL source query, if applicable.
|
||||
SQLQuery string
|
||||
|
||||
// The shards to include in the query execution.
|
||||
// If empty, all shards are included.
|
||||
Shards []uint64
|
||||
|
|
|
|||
|
|
@ -1182,7 +1182,7 @@ func (h *Handler) handleGetActiveQueries(w http.ResponseWriter, r *http.Request)
|
|||
}
|
||||
}
|
||||
for i, q := range queries {
|
||||
_, err := fmt.Fprintf(w, "%*s%q\n", -(maxlen + 2), durations[i], q.Query)
|
||||
_, err := fmt.Fprintf(w, "%*s%q\n", -(maxlen + 2), durations[i], q.PQL)
|
||||
if err != nil {
|
||||
h.logger.Printf("sending GetActiveQueries response: %s", err)
|
||||
return
|
||||
|
|
|
|||
|
|
@ -5620,10 +5620,7 @@ func (m *IndexMeta) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -6028,10 +6025,7 @@ func (m *FieldOptions) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -6114,10 +6108,7 @@ func (m *ImportResponse) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -6302,10 +6293,7 @@ func (m *BlockDataRequest) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -6508,10 +6496,7 @@ func (m *BlockDataResponse) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -6638,10 +6623,7 @@ func (m *Cache) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -6788,7 +6770,7 @@ func (m *MaxShards) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > postIndex {
|
||||
|
|
@ -6805,10 +6787,7 @@ func (m *MaxShards) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -6942,10 +6921,7 @@ func (m *CreateShardMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -7028,10 +7004,7 @@ func (m *DeleteIndexMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -7169,10 +7142,7 @@ func (m *CreateIndexMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -7342,10 +7312,7 @@ func (m *CreateFieldMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -7460,10 +7427,7 @@ func (m *DeleteFieldMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -7597,10 +7561,7 @@ func (m *DeleteAvailableShardMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -7770,10 +7731,7 @@ func (m *Field) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -7858,10 +7816,7 @@ func (m *Schema) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -8033,10 +7988,7 @@ func (m *Index) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -8170,10 +8122,7 @@ func (m *URI) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -8380,10 +8329,7 @@ func (m *Node) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -8498,10 +8444,7 @@ func (m *NodeStateMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -8607,10 +8550,7 @@ func (m *NodeEventMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -8767,10 +8707,7 @@ func (m *NodeStatus) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -8906,10 +8843,7 @@ func (m *IndexStatus) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -9087,10 +9021,7 @@ func (m *FieldStatus) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -9275,10 +9206,7 @@ func (m *ClusterStatus) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -9431,10 +9359,7 @@ func (m *BSIGroup) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -9581,10 +9506,7 @@ func (m *CreateViewMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -9731,10 +9653,7 @@ func (m *DeleteViewMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -10016,10 +9935,7 @@ func (m *ResizeInstruction) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -10221,10 +10137,7 @@ func (m *ResizeSource) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -10362,10 +10275,7 @@ func (m *TranslationResizeSource) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -10503,10 +10413,7 @@ func (m *ResizeInstructionComplete) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -10593,10 +10500,7 @@ func (m *SetCoordinatorMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -10683,10 +10587,7 @@ func (m *UpdateCoordinatorMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -10801,10 +10702,7 @@ func (m *Topology) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -10855,10 +10753,7 @@ func (m *RecalculateCaches) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -10977,10 +10872,7 @@ func (m *TransactionMessage) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -11177,10 +11069,7 @@ func (m *Transaction) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
@ -11231,10 +11120,7 @@ func (m *TransactionStats) Unmarshal(dAtA []byte) error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if skippy < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) < 0 {
|
||||
if (skippy < 0) || (iNdEx+skippy) < 0 {
|
||||
return ErrInvalidLengthPrivate
|
||||
}
|
||||
if (iNdEx + skippy) > l {
|
||||
|
|
|
|||
File diff suppressed because it is too large
Load diff
|
|
@ -102,7 +102,7 @@ message FieldRow {
|
|||
message GroupCount{
|
||||
repeated FieldRow Group = 1;
|
||||
uint64 Count = 2;
|
||||
int64 Sum = 3;
|
||||
int64 Agg = 3;
|
||||
}
|
||||
|
||||
message ValCount {
|
||||
|
|
@ -159,9 +159,15 @@ message QueryResult {
|
|||
uint64 N = 2;
|
||||
repeated Pair Pairs = 3;
|
||||
bool Changed = 4;
|
||||
ValCount ValCount = 5;
|
||||
ValCount ValCount = 5;
|
||||
repeated uint64 RowIDs = 7;
|
||||
repeated GroupCount GroupCounts = 8;
|
||||
// In the past, GroupCounts was a []GroupCount which did not indicate
|
||||
// whether it had an aggregate, or which aggregate it had. We've
|
||||
// updated this, but we keep this here so that messages using the old
|
||||
// format can get a best-effort treatment rather than causing panics.
|
||||
// Later this can almost certainly go away, but leave a comment warning
|
||||
// people that 8 is Spoken For if you do that, please.
|
||||
repeated GroupCount OldGroupCounts = 8;
|
||||
RowIdentifiers RowIdentifiers = 9;
|
||||
SignedRow SignedRow = 10;
|
||||
PairsField PairsField = 11;
|
||||
|
|
@ -169,6 +175,7 @@ message QueryResult {
|
|||
ExtractedIDMatrix ExtractedIDMatrix = 13;
|
||||
ExtractedTable ExtractedTable = 14;
|
||||
RowMatrix RowMatrix = 15;
|
||||
GroupCounts GroupCounts = 16;
|
||||
}
|
||||
|
||||
message ImportRequest {
|
||||
|
|
@ -254,3 +261,8 @@ message ImportColumnAttrsRequest {
|
|||
repeated uint64 ColumnIDs = 5;
|
||||
int64 IndexCreatedAt = 6;
|
||||
}
|
||||
|
||||
message GroupCounts{
|
||||
string Aggregate = 1;
|
||||
repeated GroupCount Groups = 2;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,9 @@
|
|||
name: "GitHub Import Load Testing (1 day)"
|
||||
|
||||
main: "pilosa server --data-dir ${TMPDIR} --txsrc ${TXSRC}"
|
||||
load: "molecula-consumer-github -i events -d id --record-type event --batch-size=100000 --start-time 2020-01-01T00:00:00Z --end-time 2020-01-01T23:00:00Z --cache-dir .githubarchive"
|
||||
load: "molecula-consumer-github -i events -d id --record-type event --batch-size=100000 --start-time 2020-01-01T00:00:00Z --end-time 2020-01-01T23:00:00Z --cache-dir ~/.githubarchive"
|
||||
|
||||
health_url: "http://localhost:10101/status"
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
- http://localhost:7070/debug/vars
|
||||
|
|
|
|||
|
|
@ -1,7 +1,9 @@
|
|||
name: "GitHub Import Load Testing (1 month)"
|
||||
|
||||
main: "pilosa server --data-dir ${TMPDIR} --txsrc ${TXSRC}"
|
||||
load: "molecula-consumer-github -i events -d id --record-type event --batch-size=100000 --start-time 2020-01-01T00:00:00Z --end-time 2020-01-31T23:00:00Z --cache-dir .githubarchive"
|
||||
load: "molecula-consumer-github -i events -d id --record-type event --batch-size=100000 --start-time 2020-01-01T00:00:00Z --end-time 2020-01-31T23:00:00Z --cache-dir ~/.githubarchive"
|
||||
|
||||
health_url: "http://localhost:10101/status"
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
- http://localhost:7070/debug/vars
|
||||
|
|
|
|||
|
|
@ -1,7 +1,9 @@
|
|||
name: "GitHub Import Load Testing (1 week)"
|
||||
|
||||
main: "pilosa server --data-dir ${TMPDIR} --txsrc ${TXSRC}"
|
||||
load: "molecula-consumer-github -i events -d id --record-type event --batch-size=100000 --start-time 2020-01-01T00:00:00Z --end-time 2020-01-06T23:00:00Z --cache-dir .githubarchive"
|
||||
load: "molecula-consumer-github -i events -d id --record-type event --batch-size=100000 --start-time 2020-01-01T00:00:00Z --end-time 2020-01-06T23:00:00Z --cache-dir ~/.githubarchive"
|
||||
|
||||
health_url: "http://localhost:10101/status"
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
- http://localhost:7070/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type count -rate 100 -n 3000"
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type difference -rate 10 -n 300"
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type groupby -rate 100 -n 3000"
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type intersect -rate 100 -n 3000"
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type row -rate 100 -n 3000"
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type row-range -rate 10 -n 300 -from 2020-01-01T00:00:00Z -
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type row -rate 100 -n 3000"
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type row-range -rate 10 -n 300 -from 2020-01-01T00:00:00Z -
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type union -rate 10 -n 300"
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -6,4 +6,5 @@ load: "pilosa-bench -type xor -rate 10 -n 300"
|
|||
health_url: "http://localhost:10101/status"
|
||||
health_regexp: "NORMAL"
|
||||
|
||||
debug_url: "http://localhost:10101/debug"
|
||||
vars_urls:
|
||||
- http://localhost:10101/debug/vars
|
||||
|
|
|
|||
|
|
@ -456,7 +456,8 @@ func ToTablerWrapper(result interface{}) (pb.ToTabler, error) {
|
|||
if !ok {
|
||||
switch v := result.(type) {
|
||||
case []pilosa.GroupCount:
|
||||
toTabler = pilosa.GroupCounts(v)
|
||||
gc := pilosa.NewGroupCounts("", v...)
|
||||
toTabler = gc
|
||||
case uint64:
|
||||
toTabler = ResultUint64(v)
|
||||
case bool:
|
||||
|
|
@ -477,7 +478,8 @@ func ToRowserWrapper(result interface{}) (pb.ToRowser, error) {
|
|||
if !ok {
|
||||
switch v := result.(type) {
|
||||
case []pilosa.GroupCount:
|
||||
toRowser = pilosa.GroupCounts(v)
|
||||
gc := pilosa.NewGroupCounts("", v...)
|
||||
toRowser = gc
|
||||
case uint64:
|
||||
toRowser = ResultUint64(v)
|
||||
case bool:
|
||||
|
|
|
|||
|
|
@ -138,7 +138,7 @@ func TestGRPC(t *testing.T) {
|
|||
},
|
||||
// []GroupCount (uint64)
|
||||
{
|
||||
[]pilosa.GroupCount{
|
||||
pilosa.NewGroupCounts("", []pilosa.GroupCount{
|
||||
pilosa.GroupCount{
|
||||
Group: []pilosa.FieldRow{
|
||||
{Field: "a", RowID: 10},
|
||||
|
|
@ -160,22 +160,21 @@ func TestGRPC(t *testing.T) {
|
|||
},
|
||||
Count: 789,
|
||||
},
|
||||
},
|
||||
}...),
|
||||
[]expHeader{
|
||||
{"a", "uint64"},
|
||||
{"b", "uint64"},
|
||||
{"count", "uint64"},
|
||||
{"sum", "int64"},
|
||||
},
|
||||
[][]expColumn{
|
||||
{uint64(10), uint64(11), uint64(123), int64(0)},
|
||||
{uint64(10), uint64(12), uint64(456), int64(0)},
|
||||
{int64(va), int64(vb), uint64(789), int64(0)},
|
||||
{uint64(10), uint64(11), uint64(123)},
|
||||
{uint64(10), uint64(12), uint64(456)},
|
||||
{int64(va), int64(vb), uint64(789)},
|
||||
},
|
||||
},
|
||||
// []GroupCount (string)
|
||||
// []GroupCount (string) + sum
|
||||
{
|
||||
[]pilosa.GroupCount{
|
||||
pilosa.NewGroupCounts("sum", []pilosa.GroupCount{
|
||||
pilosa.GroupCount{
|
||||
Group: []pilosa.FieldRow{
|
||||
{Field: "a", RowKey: "ten"},
|
||||
|
|
@ -190,7 +189,7 @@ func TestGRPC(t *testing.T) {
|
|||
},
|
||||
Count: 456,
|
||||
},
|
||||
},
|
||||
}...),
|
||||
[]expHeader{
|
||||
{"a", "string"},
|
||||
{"b", "string"},
|
||||
|
|
@ -292,7 +291,12 @@ func TestGRPC(t *testing.T) {
|
|||
}
|
||||
|
||||
// Ensure headers match.
|
||||
for i, header := range table.GetHeaders() {
|
||||
headers := table.GetHeaders()
|
||||
if len(headers) < len(test.expHeaders) {
|
||||
t.Fatalf("test %d expected %d headers, got %d, first missing header %q",
|
||||
ti, len(test.expHeaders), len(headers), test.expHeaders[len(headers)].name)
|
||||
}
|
||||
for i, header := range headers {
|
||||
if header.Name != test.expHeaders[i].name {
|
||||
t.Fatalf("test %d expected header name: %s, but got: %s", ti, test.expHeaders[i].name, header.Name)
|
||||
}
|
||||
|
|
@ -303,7 +307,12 @@ func TestGRPC(t *testing.T) {
|
|||
|
||||
// Ensure column data matches.
|
||||
for i, row := range table.GetRows() {
|
||||
for j, column := range row.GetColumns() {
|
||||
columns := row.GetColumns()
|
||||
if len(columns) != len(test.expColumns[i]) {
|
||||
t.Fatalf("test %d expected %d columns, got %d in row %d",
|
||||
ti, len(test.expColumns[i]), len(columns), i)
|
||||
}
|
||||
for j, column := range columns {
|
||||
switch v := test.expColumns[i][j].(type) {
|
||||
case string:
|
||||
val := column.GetStringVal()
|
||||
|
|
|
|||
|
|
@ -37,6 +37,7 @@ import (
|
|||
"github.com/pilosa/pilosa/v2/encoding/proto"
|
||||
"github.com/pilosa/pilosa/v2/http"
|
||||
"github.com/pilosa/pilosa/v2/pql"
|
||||
pb "github.com/pilosa/pilosa/v2/proto"
|
||||
"github.com/pilosa/pilosa/v2/server"
|
||||
"github.com/pilosa/pilosa/v2/test"
|
||||
)
|
||||
|
|
@ -1502,6 +1503,16 @@ func TestQueryHistory(t *testing.T) {
|
|||
|
||||
test.Do(t, "POST", cmd.URL()+"/index/i0", "")
|
||||
test.Do(t, "POST", cmd.URL()+"/index/i0/field/f0", "")
|
||||
|
||||
gh := server.NewGRPCHandler(cmd.API)
|
||||
_, err = gh.QuerySQLUnary(context.Background(), &pb.QuerySQLRequest{
|
||||
Sql: `select * from i0`,
|
||||
})
|
||||
|
||||
if err != nil {
|
||||
t.Fatalf("QuerySQLUnary failed: %v", err)
|
||||
}
|
||||
|
||||
test.Do(t, "POST", cmd.URL()+"/index/i0/query", "Set(0, f0=0)")
|
||||
test.Do(t, "POST", cmd.URL()+"/index/i0/query", "Set(3000000, f0=0)")
|
||||
test.Do(t, "POST", cmd.URL()+"/index/i0/query", "TopN(f0)")
|
||||
|
|
@ -1511,7 +1522,7 @@ func TestQueryHistory(t *testing.T) {
|
|||
t.Fatalf("unexpected status code: %d", w.Code)
|
||||
}
|
||||
|
||||
ret := make([]pilosa.PastQueryStatus, 3)
|
||||
ret := make([]pilosa.PastQueryStatus, 4)
|
||||
b, err := ioutil.ReadAll(w.Body)
|
||||
if err != nil {
|
||||
t.Fatalf("reading: %v", err)
|
||||
|
|
@ -1522,10 +1533,10 @@ func TestQueryHistory(t *testing.T) {
|
|||
}
|
||||
|
||||
// verify result length
|
||||
if len(ret) != 3 {
|
||||
if len(ret) != 4 {
|
||||
// each set query executes on both nodes once
|
||||
// topn query gets added to history on node0 once, node1 twice
|
||||
t.Fatalf("expected list of length 3, got %d", len(ret))
|
||||
t.Fatalf("expected list of length 4, got %d\n%+v", len(ret), ret)
|
||||
}
|
||||
|
||||
// verify sort order
|
||||
|
|
@ -1543,8 +1554,14 @@ func TestQueryHistory(t *testing.T) {
|
|||
if ret[0].Node != cluster.GetNode(0).Server.NodeID() {
|
||||
t.Fatalf("response value for 'Node' was '%s', expected '%s'", ret[0].Node, cluster.GetNode(0).Server.NodeID())
|
||||
}
|
||||
if ret[0].Query != "TopN(f0)" {
|
||||
t.Fatalf("response value for 'Query' was '%s', expected 'TopN(f0)'", ret[0].Query)
|
||||
if ret[3].PQL != "Extract(All(),Rows(f0))" {
|
||||
t.Fatalf("response value for 'PQL' was '%s', expected 'Extract(All(),Rows(f0))'", ret[0].PQL)
|
||||
}
|
||||
if ret[3].SQL != "select * from i0" {
|
||||
t.Fatalf("response value for 'SQL' was '%s', expected 'select * from i0'", ret[0].SQL)
|
||||
}
|
||||
if ret[0].PQL != "TopN(f0)" {
|
||||
t.Fatalf("response value for 'PQL' was '%s', expected 'TopN(f0)'", ret[0].PQL)
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
44
server/pg.go
44
server/pg.go
|
|
@ -252,27 +252,38 @@ func pgWriteExtractedTable(w pg.QueryResultWriter, tbl pilosa.ExtractedTable) er
|
|||
return nil
|
||||
}
|
||||
|
||||
func pgWriteGroupCount(w pg.QueryResultWriter, counts []pilosa.GroupCount) error {
|
||||
if len(counts) == 0 {
|
||||
func pgWriteGroupCount(w pg.QueryResultWriter, counts *pilosa.GroupCounts) error {
|
||||
groups := counts.Groups()
|
||||
if len(groups) == 0 {
|
||||
// Not enough information is available to construct the header.
|
||||
// This is a significant flaw in the data type.
|
||||
return nil
|
||||
}
|
||||
expectedLen := len(groups[0].Group) + 1
|
||||
|
||||
headers := make([]pg.ColumnInfo, len(counts[0].Group)+2)
|
||||
for i, g := range counts[0].Group {
|
||||
agg := counts.AggregateColumn()
|
||||
if agg != "" {
|
||||
expectedLen++
|
||||
}
|
||||
|
||||
headers := make([]pg.ColumnInfo, expectedLen)
|
||||
for i, g := range groups[0].Group {
|
||||
headers[i] = pg.ColumnInfo{
|
||||
Name: g.Field,
|
||||
Type: pg.TypeCharoid,
|
||||
}
|
||||
}
|
||||
headers[len(headers)-2] = pg.ColumnInfo{
|
||||
next := len(groups[0].Group)
|
||||
headers[next] = pg.ColumnInfo{
|
||||
Name: "count",
|
||||
Type: pg.TypeCharoid,
|
||||
}
|
||||
headers[len(headers)-1] = pg.ColumnInfo{
|
||||
Name: "sum",
|
||||
Type: pg.TypeCharoid,
|
||||
if agg != "" {
|
||||
next++
|
||||
headers[next] = pg.ColumnInfo{
|
||||
Name: agg,
|
||||
Type: pg.TypeCharoid,
|
||||
}
|
||||
}
|
||||
err := w.WriteHeader(headers...)
|
||||
if err != nil {
|
||||
|
|
@ -280,8 +291,10 @@ func pgWriteGroupCount(w pg.QueryResultWriter, counts []pilosa.GroupCount) error
|
|||
}
|
||||
|
||||
vals := make([]string, len(headers))
|
||||
for _, gc := range counts {
|
||||
for j, g := range gc.Group {
|
||||
for _, gc := range groups {
|
||||
var j int
|
||||
var g pilosa.FieldRow
|
||||
for j, g = range gc.Group {
|
||||
var v string
|
||||
switch {
|
||||
case g.Value != nil:
|
||||
|
|
@ -293,8 +306,12 @@ func pgWriteGroupCount(w pg.QueryResultWriter, counts []pilosa.GroupCount) error
|
|||
}
|
||||
vals[j] = v
|
||||
}
|
||||
vals[len(vals)-2] = strconv.FormatUint(gc.Count, 10)
|
||||
vals[len(vals)-1] = strconv.FormatInt(gc.Sum, 10)
|
||||
j++
|
||||
vals[j] = strconv.FormatUint(gc.Count, 10)
|
||||
if agg != "" {
|
||||
j++
|
||||
vals[j] = strconv.FormatInt(gc.Agg, 10)
|
||||
}
|
||||
|
||||
err := w.WriteRowText(vals...)
|
||||
if err != nil {
|
||||
|
|
@ -370,6 +387,9 @@ func pgWriteResult(w pg.QueryResultWriter, result interface{}) error {
|
|||
case pilosa.ExtractedTable:
|
||||
return pgWriteExtractedTable(w, result)
|
||||
case []pilosa.GroupCount:
|
||||
gc := pilosa.NewGroupCounts("", result...)
|
||||
return pgWriteGroupCount(w, gc)
|
||||
case *pilosa.GroupCounts:
|
||||
return pgWriteGroupCount(w, result)
|
||||
case pb.ToRowser: // we should avoid protobuf where we can...
|
||||
return pgWriteRowser(w, result)
|
||||
|
|
|
|||
|
|
@ -169,11 +169,10 @@ func TestPostgresHandler(t *testing.T) {
|
|||
Columns: []pg.ColumnInfo{
|
||||
{Name: "set", Type: pg.TypeCharoid},
|
||||
{Name: "count", Type: pg.TypeCharoid},
|
||||
{Name: "sum", Type: pg.TypeCharoid},
|
||||
},
|
||||
Data: [][]string{
|
||||
{"4", "3", "0"},
|
||||
{"5", "3", "0"},
|
||||
{"4", "3"},
|
||||
{"5", "3"},
|
||||
},
|
||||
},
|
||||
},
|
||||
|
|
|
|||
|
|
@ -298,7 +298,7 @@ func TestMain_GroupBy(t *testing.T) {
|
|||
if res, err := m.QueryProtobuf("i", `GroupBy(Rows(generalk), Rows(subk))`); err != nil {
|
||||
t.Fatal(err)
|
||||
} else {
|
||||
test.CheckGroupBy(t, expected, res.Results[0].([]pilosa.GroupCount))
|
||||
test.CheckGroupBy(t, expected, res.Results[0].(*pilosa.GroupCounts).Groups())
|
||||
}
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -40,6 +40,7 @@ type MappedSQL struct {
|
|||
Statement sqlparser.Statement
|
||||
Mask QueryMask
|
||||
Tables []string
|
||||
SQL string
|
||||
}
|
||||
|
||||
// Mapper is responsible for mapping a SQL query to structure representation
|
||||
|
|
@ -109,5 +110,6 @@ func (m *Mapper) MapSQL(sql string) (*MappedSQL, error) {
|
|||
Statement: stmt,
|
||||
Mask: qm,
|
||||
Tables: tableNames,
|
||||
SQL: sql,
|
||||
}, nil
|
||||
}
|
||||
|
|
|
|||
|
|
@ -43,6 +43,7 @@ func NewSelectHandler(api *pilosa.API) *SelectHandler {
|
|||
// Handle executes mapped SQL
|
||||
func (s *SelectHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.ToRowser, error) {
|
||||
stmt, ok := mapped.Statement.(*sqlparser.Select)
|
||||
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("statement is not type select: %T", mapped.Statement)
|
||||
}
|
||||
|
|
@ -50,7 +51,7 @@ func (s *SelectHandler) Handle(ctx context.Context, mapped *MappedSQL) (pproto.T
|
|||
if err != nil {
|
||||
return nil, errors.Wrap(err, "mapping select")
|
||||
}
|
||||
return s.execMappingResult(ctx, mr)
|
||||
return s.execMappingResult(ctx, mr, mapped.SQL)
|
||||
}
|
||||
|
||||
func (s *SelectHandler) mapSelect(ctx context.Context, selectStmt *sqlparser.Select, qm QueryMask) (*MappingResult, error) {
|
||||
|
|
@ -74,12 +75,12 @@ func (s *SelectHandler) mapSelect(ctx context.Context, selectStmt *sqlparser.Sel
|
|||
return mr, nil
|
||||
}
|
||||
|
||||
func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult) (pproto.ToRowser, error) {
|
||||
func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult, sql string) (pproto.ToRowser, error) {
|
||||
if mr.Query == "" {
|
||||
return nil, errors.New("no pql query created")
|
||||
}
|
||||
|
||||
resp, err := s.api.Query(ctx, &pilosa.QueryRequest{Index: mr.IndexName, Query: mr.Query})
|
||||
resp, err := s.api.Query(ctx, &pilosa.QueryRequest{Index: mr.IndexName, Query: mr.Query, SQLQuery: sql})
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "doing pql query")
|
||||
}
|
||||
|
|
@ -90,7 +91,7 @@ func (s *SelectHandler) execMappingResult(ctx context.Context, mr *MappingResult
|
|||
case pproto.ToRowser:
|
||||
result = res
|
||||
case []pilosa.GroupCount:
|
||||
result = pilosa.GroupCounts(res)
|
||||
result = pilosa.NewGroupCounts("", res...)
|
||||
case uint64:
|
||||
result = pproto.ConstRowser{
|
||||
{
|
||||
|
|
|
|||
26
tracker.go
26
tracker.go
|
|
@ -21,14 +21,16 @@ import (
|
|||
)
|
||||
|
||||
type ActiveQueryStatus struct {
|
||||
Query string `json:"query"`
|
||||
PQL string `json:"PQL"`
|
||||
SQL string `json:"SQL,omitempty"`
|
||||
Node string `json:"node"`
|
||||
Index string `json:"index"`
|
||||
Age time.Duration `json:"age"`
|
||||
}
|
||||
|
||||
type PastQueryStatus struct {
|
||||
Query string `json:"query"`
|
||||
PQL string `json:"PQL"`
|
||||
SQL string `json:"SQL,omitempty"`
|
||||
Node string `json:"nodeID"`
|
||||
Index string `json:"index"`
|
||||
Start time.Time `json:"start"`
|
||||
|
|
@ -36,14 +38,16 @@ type PastQueryStatus struct {
|
|||
}
|
||||
|
||||
type activeQuery struct {
|
||||
query string
|
||||
PQL string
|
||||
SQL string
|
||||
node string
|
||||
index string
|
||||
started time.Time
|
||||
}
|
||||
|
||||
type pastQuery struct {
|
||||
query string
|
||||
PQL string
|
||||
SQL string
|
||||
node string
|
||||
index string
|
||||
started time.Time
|
||||
|
|
@ -123,7 +127,7 @@ func newQueryTracker(historyLength int) *queryTracker {
|
|||
select {
|
||||
case update := <-updates:
|
||||
if update.end {
|
||||
pq := pastQuery{update.q.query, update.q.node, update.q.index, update.q.started, update.endTime.Sub(update.q.started)}
|
||||
pq := pastQuery{update.q.PQL, update.q.SQL, update.q.node, update.q.index, update.q.started, update.endTime.Sub(update.q.started)}
|
||||
tracker.history.add(pq)
|
||||
delete(activeQueries, update.q)
|
||||
} else {
|
||||
|
|
@ -146,8 +150,8 @@ func newQueryTracker(historyLength int) *queryTracker {
|
|||
return tracker
|
||||
}
|
||||
|
||||
func (t *queryTracker) Start(query, nodeID, index string, start time.Time) *activeQuery {
|
||||
q := &activeQuery{query, nodeID, index, start}
|
||||
func (t *queryTracker) Start(pql, sql, nodeID, index string, start time.Time) *activeQuery {
|
||||
q := &activeQuery{pql, sql, nodeID, index, start}
|
||||
t.updates <- queryStatusUpdate{q, false, time.Time{}}
|
||||
return q
|
||||
}
|
||||
|
|
@ -166,9 +170,9 @@ func (t *queryTracker) ActiveQueries() []ActiveQueryStatus {
|
|||
return true
|
||||
case queries[i].started.After(queries[j].started):
|
||||
return false
|
||||
case queries[i].query < queries[j].query:
|
||||
case queries[i].PQL < queries[j].PQL:
|
||||
return true
|
||||
case queries[i].query > queries[j].query:
|
||||
case queries[i].PQL > queries[j].PQL:
|
||||
return false
|
||||
default:
|
||||
return false
|
||||
|
|
@ -177,7 +181,7 @@ func (t *queryTracker) ActiveQueries() []ActiveQueryStatus {
|
|||
now := time.Now()
|
||||
out := make([]ActiveQueryStatus, len(queries))
|
||||
for i, v := range queries {
|
||||
out[i] = ActiveQueryStatus{v.query, v.node, v.index, now.Sub(v.started)}
|
||||
out[i] = ActiveQueryStatus{v.PQL, v.SQL, v.node, v.index, now.Sub(v.started)}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
|
@ -186,7 +190,7 @@ func (t *queryTracker) PastQueries() []PastQueryStatus {
|
|||
queries := t.history.slice()
|
||||
out := make([]PastQueryStatus, len(queries))
|
||||
for i, v := range queries {
|
||||
out[i] = PastQueryStatus{v.query, v.node, v.index, v.started, v.runtime}
|
||||
out[i] = PastQueryStatus{v.PQL, v.SQL, v.node, v.index, v.started, v.runtime}
|
||||
}
|
||||
return out
|
||||
|
||||
|
|
|
|||
|
|
@ -47,11 +47,11 @@ func TestRingBuffer(t *testing.T) {
|
|||
}
|
||||
|
||||
for n, q := range buffer.slice() {
|
||||
if q.query != tests[k].queries[n] {
|
||||
t.Fatalf("test[%d], buffer[%d] expected querystring '%s', found '%s'", k, n, tests[k].queries[n], q.query)
|
||||
if q.PQL != tests[k].queries[n] {
|
||||
t.Fatalf("test[%d], buffer[%d] expected querystring '%s', found '%s'", k, n, tests[k].queries[n], q.PQL)
|
||||
}
|
||||
}
|
||||
buffer.add(pastQuery{query: fmt.Sprintf("%d", k)})
|
||||
buffer.add(pastQuery{PQL: fmt.Sprintf("%d", k)})
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -63,13 +63,13 @@ func TestQueryTracker(t *testing.T) {
|
|||
t.Fatalf("expected no active queries; found %v", queries)
|
||||
}
|
||||
|
||||
qs := tracker.Start("test query", "node0", "i", time.Now())
|
||||
qs := tracker.Start("test query", "test SQL", "node0", "i", time.Now())
|
||||
|
||||
var queries []ActiveQueryStatus
|
||||
for len(queries) < 1 {
|
||||
queries = tracker.ActiveQueries()
|
||||
}
|
||||
if len(queries) > 1 || queries[0].Query != "test query" {
|
||||
if len(queries) > 1 || queries[0].PQL != "test query" {
|
||||
t.Fatalf("unexpected queries: %v", queries)
|
||||
}
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue