remove noFilter and filterWithOffsetLimit

can use an empty list of filters and a list of offsetFilter followed by limit
filter respectively

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
 }

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)
 			}
 		}