scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 10:54:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Add GroupBy filter.

Browse source
This commit is contained in:
Ben Johnson 2018-11-21 14:42:01 -07:00
parent 73e07ae11e
commit 727659644b
No known key found for this signature in database
GPG key ID: 81741CD251883081
6 changed files with 1062 additions and 926 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

38
executor.go
View file

@ -884,6 +884,10 @@ func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call
} else if hasLimit {
limit = int(lim)
}
filter, _, err := c.CallArg("filter")
if err != nil {
return nil, err
}
// perform necessary Rows queries (any that have limit or columns args) -
// TODO, call async? would only help if multiple Rows queries had a column
@ -892,7 +896,7 @@ func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call
childRows := make([]RowIDs, len(c.Children))
for i, child := range c.Children {
if child.Name != "Rows" {
return nil, errors.Errorf("'%s' is not a valid child query for GroupBy, must be 'Rows'", c.Name)
return nil, errors.Errorf("'%s' is not a valid child query for GroupBy, must be 'Rows'", child.Name)
}
_, hasLimit, err := child.UintArg("limit")
if err != nil {
@ -915,7 +919,7 @@ func (e *executor) executeGroupBy(ctx context.Context, index string, c *pql.Call
// Execute calls in bulk on each remote node and merge.
mapFn := func(shard uint64) (interface{}, error) {
return e.executeGroupByShard(ctx, index, c, shard, childRows)
return e.executeGroupByShard(ctx, index, c, filter, shard, childRows)
}
// Merge returned results at coordinating node.
reduceFn := func(prev, v interface{}) interface{} {
@ -1028,8 +1032,15 @@ func (g GroupCount) Compare(o GroupCount) int {
return 0
}
func (e *executor) executeGroupByShard(_ context.Context, index string, c *pql.Call, shard uint64, childRows []RowIDs) ([]GroupCount, error) {
iter, err := newGroupByIterator(childRows, c.Children, index, shard, e.Holder)
func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql.Call, filter *pql.Call, shard uint64, childRows []RowIDs) (_ []GroupCount, err error) {
var filterRow *Row
if filter != nil {
if filterRow, err = e.executeBitmapCallShard(ctx, index, filter, shard); err != nil {
return nil, errors.Wrapf(err, "executing group by filter for shard %d", shard)
}
}
iter, err := newGroupByIterator(childRows, c.Children, filterRow, index, shard, e.Holder)
if err != nil {
return nil, errors.Wrapf(err, "getting group by iterator for shard %d", shard)
}
@ -2687,16 +2698,20 @@ type groupByIterator struct {
// fields and then sets the row ids.
fields []FieldRow
done bool
// Optional filter row to intersect against first level of values.
filter *Row
}
// newGroupByIterator initializes a new groupByIterator.
func newGroupByIterator(rowIDs []RowIDs, children []*pql.Call, index string, shard uint64, holder *Holder) (*groupByIterator, error) {
func newGroupByIterator(rowIDs []RowIDs, children []*pql.Call, filter *Row, index string, shard uint64, holder *Holder) (*groupByIterator, error) {
gbi := &groupByIterator{
rowIters: make([]*rowIterator, len(children)),
rows: make([]struct {
row *Row
id uint64
}, len(children)),
filter: filter,
fields: make([]FieldRow, len(children)),
}
@ -2756,6 +2771,11 @@ func newGroupByIterator(rowIDs []RowIDs, children []*pql.Call, index string, sha
}
}
// Apply filter to first level, if available.
if gbi.filter != nil && len(gbi.rows) > 0 {
gbi.rows[0].row = gbi.rows[0].row.Intersect(gbi.filter)
}
for i := 1; i < len(gbi.rows)-1; i++ {
gbi.rows[i].row = gbi.rows[i].row.Intersect(gbi.rows[i-1].row)
}
@ -2774,10 +2794,12 @@ func (gbi *groupByIterator) nextAtIdx(i int) {
if wrapped && i != 0 {
gbi.nextAtIdx(i - 1)
}
if i != 0 && i != len(gbi.rows)-1 {
gbi.rows[i].row = nr.Intersect(gbi.rows[i-1].row)
} else {
if i == 0 && gbi.filter != nil {
gbi.rows[i].row = nr.Intersect(gbi.filter)
} else if i == 0 || i == len(gbi.rows)-1 {
gbi.rows[i].row = nr
} else {
gbi.rows[i].row = nr.Intersect(gbi.rows[i-1].row)
}
gbi.rows[i].id = rowID
}

10
executor_test.go
View file

@ -2845,6 +2845,16 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
checkGroupBy(t, expected, results)
})
t.Run("Filter", func(t *testing.T) {
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 100}}, Count: 3},
{Group: []pilosa.FieldRow{{Field: "general", RowID: 10}, {Field: "sub", RowID: 110}}, Count: 1},
}
results := c.Query(t, "i", `GroupBy(Rows(field=general), Rows(field=sub), filter=Row(general=10))`).Results[0].([]pilosa.GroupCount)
checkGroupBy(t, expected, results)
})
t.Run("check field offset no limit", func(t *testing.T) {
expected := []pilosa.GroupCount{
{Group: []pilosa.FieldRow{{Field: "general", RowID: 11}}, Count: 2},

156
pql/ast.go
View file

@ -27,28 +27,34 @@ import (
type Query struct {
Calls []*Call
lastField string
lastCond Token
inList bool
callStack []*Call
callStack []*callStackElem
conditional []string
}
func (q *Query) startCall(name string) {
newCall := &Call{Name: name}
q.callStack = append(q.callStack, newCall)
q.callStack = append(q.callStack, &callStackElem{call: newCall})
if len(q.callStack) == 1 {
q.Calls = append(q.Calls, newCall)
} else {
calls := q.callStack[len(q.callStack)-2].Children
q.callStack[len(q.callStack)-2].Children = append(calls, newCall)
} else if prevElem := q.callStack[len(q.callStack)-2]; prevElem.lastField == "" {
prevElem.call.Children = append(prevElem.call.Children, newCall)
}
}
func (q *Query) endCall() {
// endCall removes the last element from the call stack and returns the call.
func (q *Query) endCall() *Call {
elem := q.callStack[len(q.callStack)-1]
q.callStack[len(q.callStack)-1] = nil
q.callStack = q.callStack[:len(q.callStack)-1]
return elem.call
}
func (q *Query) lastCallStackElem() *callStackElem {
if len(q.callStack) == 0 {
return nil
}
return q.callStack[len(q.callStack)-1]
}
func (q *Query) addPosNum(key, value string) {
@ -63,9 +69,9 @@ func (q *Query) addPosStr(key, value string) {
func (q *Query) startConditional() {
q.conditional = make([]string, 0)
call := q.callStack[len(q.callStack)-1]
if call.Args == nil {
call.Args = make(map[string]interface{})
elem := q.lastCallStackElem()
if elem.call.Args == nil {
elem.call.Args = make(map[string]interface{})
}
}
@ -89,47 +95,48 @@ func (q *Query) endConditional() {
high++
}
call := q.callStack[len(q.callStack)-1]
call.Args[field] = &Condition{Op: BETWEEN, Value: []interface{}{low, high}}
elem := q.lastCallStackElem()
elem.call.Args[field] = &Condition{Op: BETWEEN, Value: []interface{}{low, high}}
q.conditional = nil
}
func (q *Query) addField(field string) {
if q.lastField != "" {
panic(fmt.Sprintf("addField called with '%s' while field is not empty, it's: %s", field, q.lastField))
elem := q.lastCallStackElem()
if elem == nil || elem.lastField != "" {
panic(fmt.Sprintf("addField called with '%s' while field is not empty, it's: %s", field, elem.lastField))
}
q.lastField = field
call := q.callStack[len(q.callStack)-1]
if call.Args == nil {
call.Args = make(map[string]interface{})
elem.lastField = field
if elem.call.Args == nil {
elem.call.Args = make(map[string]interface{})
}
}
func (q *Query) addVal(val interface{}) {
if q.lastField == "" {
elem := q.lastCallStackElem()
if elem == nil || elem.lastField == "" {
panic(fmt.Sprintf("addVal called with '%s' when lastField is empty", val))
}
call := q.callStack[len(q.callStack)-1]
if q.inList {
list := call.Args[q.lastField].([]interface{})
call.Args[q.lastField] = append(list, val)
if elem.inList {
list := elem.call.Args[elem.lastField].([]interface{})
elem.call.Args[elem.lastField] = append(list, val)
return
}
if q.lastCond != ILLEGAL {
call.Args[q.lastField] = &Condition{
Op: q.lastCond,
if elem.lastCond != ILLEGAL {
elem.call.Args[elem.lastField] = &Condition{
Op: elem.lastCond,
Value: val,
}
} else {
call.Args[q.lastField] = val
elem.call.Args[elem.lastField] = val
}
q.lastField = ""
q.lastCond = ILLEGAL
elem.lastField = ""
elem.lastCond = ILLEGAL
}
func (q *Query) addNumVal(val string) {
if q.lastField == "" {
elem := q.lastCallStackElem()
if elem == nil || elem.lastField == "" {
panic(fmt.Sprintf("addIntVal called with '%s' when lastField is empty", val))
}
var ival interface{}
@ -142,70 +149,70 @@ func (q *Query) addNumVal(val string) {
if err != nil {
panic(err)
}
call := q.callStack[len(q.callStack)-1]
if q.inList {
if q.lastCond != ILLEGAL {
list := call.Args[q.lastField].(*Condition).Value.([]interface{})
call.Args[q.lastField] = &Condition{
Op: q.lastCond,
if elem.inList {
if elem.lastCond != ILLEGAL {
list := elem.call.Args[elem.lastField].(*Condition).Value.([]interface{})
elem.call.Args[elem.lastField] = &Condition{
Op: elem.lastCond,
Value: append(list, ival),
}
} else {
list := call.Args[q.lastField].([]interface{})
call.Args[q.lastField] = append(list, ival)
list := elem.call.Args[elem.lastField].([]interface{})
elem.call.Args[elem.lastField] = append(list, ival)
}
return
} else if q.lastCond != ILLEGAL {
call.Args[q.lastField] = &Condition{
Op: q.lastCond,
} else if elem.lastCond != ILLEGAL {
elem.call.Args[elem.lastField] = &Condition{
Op: elem.lastCond,
Value: ival,
}
} else {
call.Args[q.lastField] = ival
elem.call.Args[elem.lastField] = ival
}
q.lastField = ""
q.lastCond = ILLEGAL
elem.lastField = ""
elem.lastCond = ILLEGAL
}
func (q *Query) startList() {
call := q.callStack[len(q.callStack)-1]
if q.lastCond != ILLEGAL {
call.Args[q.lastField] = &Condition{
Op: q.lastCond,
elem := q.lastCallStackElem()
if elem.lastCond != ILLEGAL {
elem.call.Args[elem.lastField] = &Condition{
Op: elem.lastCond,
Value: make([]interface{}, 0),
}
} else {
call.Args[q.lastField] = make([]interface{}, 0)
elem.call.Args[elem.lastField] = make([]interface{}, 0)
}
q.inList = true
elem.inList = true
}
func (q *Query) endList() {
q.inList = false
q.lastField = ""
q.lastCond = ILLEGAL
elem := q.lastCallStackElem()
elem.inList = false
elem.lastField = ""
elem.lastCond = ILLEGAL
}
func (q *Query) addGT() {
q.lastCond = GT
q.lastCallStackElem().lastCond = GT
}
func (q *Query) addLT() {
q.lastCond = LT
q.lastCallStackElem().lastCond = LT
}
func (q *Query) addGTE() {
q.lastCond = GTE
q.lastCallStackElem().lastCond = GTE
}
func (q *Query) addLTE() {
q.lastCond = LTE
q.lastCallStackElem().lastCond = LTE
}
func (q *Query) addEQ() {
q.lastCond = EQ
q.lastCallStackElem().lastCond = EQ
}
func (q *Query) addNEQ() {
q.lastCond = NEQ
q.lastCallStackElem().lastCond = NEQ
}
func (q *Query) addBTWN() {
q.lastCond = BETWEEN
q.lastCallStackElem().lastCond = BETWEEN
}
// WriteCallN returns the number of mutating calls.
@ -229,6 +236,13 @@ func (q *Query) String() string {
return strings.Join(a, "\n")
}
type callStackElem struct {
call *Call
lastField string
lastCond Token
inList bool
}
// Call represents a function call in the AST.
type Call struct {
Name string
@ -329,6 +343,22 @@ func (c *Call) UintSliceArg(key string) ([]uint64, bool, error) {
}
}
// CallArg is for reading the value at key from call.Args as a Call. If the
// key is not in Call.Args, the value of the returned value will be nil, and
// the error will be nil. An error is returned if the value is not a Call.
func (c *Call) CallArg(key string) (*Call, bool, error) {
val, ok := c.Args[key]
if !ok {
return nil, false, nil
}
switch tval := val.(type) {
case *Call:
return tval, true, nil
default:
return nil, true, fmt.Errorf("could not convert %v of type %T to Call in Call.CallArg", tval, tval)
}
}
// keys returns a list of argument keys in sorted order.
func (c *Call) keys() []string {
a := make([]string, 0, len(c.Args))

1
pql/pql.peg
View file

@ -44,6 +44,7 @@ item <- ( 'null' &(comma / sp close) { p.addVal(nil) }
/ 'false' &(comma / sp close) { p.addVal(false) }
/ < '-'? [0-9]+ ('.'[0-9]*)? > { p.addNumVal(buffer[begin:end]) }
/ < '-'? '.'[0-9]+ > { p.addNumVal(buffer[begin:end]) }
/ < IDENT > { p.startCall(buffer[begin:end]) } open allargs comma? close { p.addVal(p.endCall()) }
/ < ([[A-Z]] / [0-9] / '-' / '_' / ':')+ > { p.addVal(buffer[begin:end]) }
/ < '"' doublequotedstring '"' > { s, _ := strconv.Unquote(buffer[begin:end]); p.addVal(s) }
/ '\'' < singlequotedstring > '\'' { p.addVal(buffer[begin:end]) }

1766
pql/pql.peg.go
View file

File diff suppressed because it is too large Load diff

17
pql/pqlpeg_test.go
View file

@ -612,6 +612,23 @@ func TestPQLDeepEquality(t *testing.T) {
},
},
}},
{
name: "GroupBy",
call: "GroupBy(Rows(), filter=Row(a=1))",
exp: &Call{
Name: "GroupBy",
Args: map[string]interface{}{
"filter": &Call{
Name: "Row",
Args: map[string]interface{}{
"a": int64(1),
},
},
},
Children: []*Call{
{Name: "Rows"},
},
}},
}
for i, test := range tests {