diff --git a/executor.go b/executor.go index 7c23ac6e4..d6b9ed46e 100644 --- a/executor.go +++ b/executor.go @@ -891,13 +891,13 @@ func (e *executor) executeGroupByShard(ctx context.Context, index string, c *pql return results, nil } // Get filter based on the field directive. - filter, err := getGroupByFilterFunction(fieldDirective.(string)) + filters, err := getGroupByFilterFunction(fieldDirective.(string)) if err != nil { return nil, err } set := make([]gbi, 0) - for _, rowID := range frag.rowsWithFilter(0, filter) { + for _, rowID := range frag.rows(0, filters...) { set = append(set, gbi{ row: frag.row(rowID), fieldRow: FieldRow{ @@ -1012,31 +1012,30 @@ func (e *executor) executeRowsShard(ctx context.Context, index string, c *pql.Ca start = previous + 1 } - filter := noFilter + filters := []rowFilter{} if limit, hasLimit, err := c.UintArg("limit"); err != nil { return nil, errors.Wrap(err, "getting limit") } else if hasLimit { - filter = (&filterWithLimit{limit: limit}).filter + filters = append(filters, (&filterWithLimit{limit: limit}).filter) } if columnID, ok, err := c.UintArg("column"); err != nil { return nil, err } else if ok { - return frag.rowsForColumnWithFilter(start, columnID, filter), nil + return frag.rowsForColumnWithFilter(start, columnID, filters...), nil } else { - return frag.rowsWithFilter(start, filter), nil + return frag.rows(start, filters...), nil } } // getGroupByFilterFunction returns a rowFilter based on the // field directive provided. -func getGroupByFilterFunction(fieldDirective string) (rowFilter, error) { +func getGroupByFilterFunction(fieldDirective string) (ret []rowFilter, err error) { parts := strings.Split(fieldDirective, ":") hasLimit := false hasOffset := false limit := uint64(0) offset := uint64(0) - var err error // fieldDirective can have one of the following forms: // [fieldName] // [fieldName:limit] @@ -1046,33 +1045,31 @@ func getGroupByFilterFunction(fieldDirective string) (rowFilter, error) { // [fieldName:offset:limit:extra] will be treated as // [fieldName:offset:limit] (i.e. `extra` is ignored). if len(parts) == 1 { - return noFilter, nil + return ret, nil } else if len(parts) == 2 { hasLimit = true if limit, err = strconv.ParseUint(parts[1], 10, 64); err != nil { - return nil, errors.Wrap(err, "getting groupby field limit only value") + return ret, errors.Wrap(err, "getting groupby field limit only value") } } else { hasOffset = true if offset, err = strconv.ParseUint(parts[1], 10, 64); err != nil { - return nil, errors.Wrap(err, "getting groupby field offset value") + return ret, errors.Wrap(err, "getting groupby field offset value") } if parts[2] != "" { hasLimit = true if limit, err = strconv.ParseUint(parts[2], 10, 64); err != nil { - return nil, errors.Wrap(err, "getting groupby field limit value") + return ret, errors.Wrap(err, "getting groupby field limit value") } } } - if hasOffset && hasLimit { - f := filterWithOffsetLimit{offset: offset, limit: limit} - return f.filter, nil - } else if hasLimit { - f := filterWithLimit{limit: limit} - return f.filter, nil + if hasOffset { + ret = append(ret, (&filterWithOffset{offset: offset}).filter) } - f := filterWithOffset{offset: offset} - return f.filter, nil + if hasLimit { + ret = append(ret, (&filterWithLimit{limit: limit}).filter) + } + return ret, nil } func (e *executor) executeBitmapShard(_ context.Context, index string, c *pql.Call, shard uint64) (*Row, error) { @@ -2322,22 +2319,6 @@ func isString(v interface{}) bool { return ok } -// Filters to be used with RowsWithFilter queries. -type filterWithOffsetLimit struct { - offset, limit uint64 -} - -func (fol *filterWithOffsetLimit) filter(rowID uint64) (bool, bool) { - if rowID >= fol.offset { - if fol.limit > 0 { - fol.limit-- - return true, false - } - return false, true - } - return false, false -} - type filterWithOffset struct { offset uint64 } diff --git a/fragment.go b/fragment.go index a29f68109..49850f800 100644 --- a/fragment.go +++ b/fragment.go @@ -1769,17 +1769,14 @@ func (f *fragment) readCacheFromArchive(r io.Reader) error { // processing. type rowFilter func(rowID uint64) (bool, bool) -// noFilter is a filter function which has no restrictions. -var noFilter = func(rowID uint64) (bool, bool) { return true, false } - // rows returns all rows by calling rowsWithFilter() // with a completely unrestrictive filter. -func (f *fragment) rows(start uint64) []uint64 { - return f.rowsWithFilter(start, noFilter) +func (f *fragment) rows(start uint64, filters ...rowFilter) []uint64 { + return f.rowsWithFilter(start, filters...) } -func (f *fragment) rowsWithFilter(start uint64, filter rowFilter) []uint64 { +func (f *fragment) rowsWithFilter(start uint64, filters ...rowFilter) []uint64 { startKey := rowToKey(start) i, _ := f.storage.Containers.Iterator(startKey) rows := make([]uint64, 0) @@ -1797,24 +1794,32 @@ func (f *fragment) rowsWithFilter(start uint64, filter rowFilter) []uint64 { continue } - // apply filter - if addRow, breakOut := filter(vRow); breakOut { - break - } else if addRow { + // apply filters + addRow := true + for _, filter := range filters { + add, done := filter(vRow) + if done { + return rows + } + addRow = add && addRow + if !addRow { + break + } + } + if addRow { rows = append(rows, vRow) } lastRow = vRow } return rows - } func (f *fragment) rowsForColumn(columnID uint64) []uint64 { - return f.rowsForColumnWithFilter(0, columnID, noFilter) + return f.rowsForColumnWithFilter(0, columnID) } -func (f *fragment) rowsForColumnWithFilter(start, columnID uint64, filter rowFilter) []uint64 { +func (f *fragment) rowsForColumnWithFilter(start, columnID uint64, filters ...rowFilter) []uint64 { if columnID/ShardWidth != f.shard { panic(fmt.Sprintln("fragment.rowsForColumn should never be called with a columnID which is not in the fragment's shard", columnID, columnID/ShardWidth, f.shard)) @@ -1845,9 +1850,18 @@ func (f *fragment) rowsForColumnWithFilter(start, columnID uint64, filter rowFil // apply filter if c.Contains(colVal) { - if addRow, breakOut := filter(vRow); breakOut { - break - } else if addRow { + addRow := true + for _, filter := range filters { + add, done := filter(vRow) + if done { + return rows + } + addRow = add && addRow + if !addRow { + break + } + } + if addRow { rows = append(rows, vRow) } }