From b64a3e0c68f666067e5c3a67475f296bdc2758dc Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Mon, 3 Jun 2019 16:29:04 +0300
Subject: [PATCH] adds filter support to MinRow and MaxRow

---
 executor.go | 32 ++++++++++++++++++++++----------
 fragment.go | 40 ++++++++++++++++++++++++++++++++++++++++
 2 files changed, 62 insertions(+), 10 deletions(-)

diff --git a/executor.go b/executor.go
index 3529995f5..7d474b6a4 100644
--- a/executor.go
+++ b/executor.go
@@ -762,6 +762,15 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal
 
 // executeMinRowShard returns the minimum row ID for a shard.
 func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) {
+	var filter *Row
+	if len(c.Children) == 1 {
+		row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
+		if err != nil {
+			return Pair{}, err
+		}
+		filter = row
+	}
+
 	fieldName, _ := c.Args["field"].(string)
 	field := e.Holder.Field(index, fieldName)
 	if field == nil {
@@ -773,18 +782,24 @@ func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql.
 		return Pair{}, nil
 	}
 
-	count := uint64(1)
-	if !fragment.hasRowID {
-		count = 0
-	}
+	minRowID, count := fragment.minRow(filter)
 	return Pair{
-		ID:    fragment.minRowID,
+		ID:    minRowID,
 		Count: count,
 	}, nil
 }
 
 // executeMaxRowShard returns the minimum row ID for a shard.
 func (e *executor) executeMaxRowShard(ctx context.Context, index string, c *pql.Call, shard uint64) (Pair, error) {
+	var filter *Row
+	if len(c.Children) == 1 {
+		row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
+		if err != nil {
+			return Pair{}, err
+		}
+		filter = row
+	}
+
 	fieldName, _ := c.Args["field"].(string)
 	field := e.Holder.Field(index, fieldName)
 	if field == nil {
@@ -796,12 +811,9 @@ func (e *executor) executeMaxRowShard(ctx context.Context, index string, c *pql.
 		return Pair{}, nil
 	}
 
-	count := uint64(1)
-	if !fragment.hasRowID {
-		count = 0
-	}
+	maxRowID, count := fragment.maxRow(filter)
 	return Pair{
-		ID:    fragment.maxRowID,
+		ID:    maxRowID,
 		Count: count,
 	}, nil
 }
diff --git a/fragment.go b/fragment.go
index 642f378bb..c6cc03291 100644
--- a/fragment.go
+++ b/fragment.go
@@ -1036,6 +1036,46 @@ func (f *fragment) maxUnsigned(filter *Row, bitDepth uint) (max int64, count uin
 	return max, count
 }
 
+// minRow returns minRowID of the rows in the filter and its count.
+// if filter is nil, it returns fragment.minRowID, 1
+// if fragment has no rows, it returns 0, 0
+func (f *fragment) minRow(filter *Row) (uint64, uint64) {
+	if f.hasRowID {
+		if filter == nil {
+			return f.minRowID, 1
+		}
+		// iterate from min row ID and return the first that intersects with filter.
+		for i := f.minRowID; i <= f.maxRowID; i++ {
+			row := f.row(i).Intersect(filter)
+			count := row.Count()
+			if count > 0 {
+				return i, count
+			}
+		}
+	}
+	return 0, 0
+}
+
+// maxRow returns maxRowID of the rows in the filter and its count.
+// if filter is nil, it returns fragment.maxRowID, 1
+// if fragment has no rows, it returns 0, 0
+func (f *fragment) maxRow(filter *Row) (uint64, uint64) {
+	if f.hasRowID {
+		if filter == nil {
+			return f.maxRowID, 1
+		}
+		// iterate back from max row ID and return the first that intersects with filter.
+		for i := f.maxRowID; i >= f.minRowID; i-- {
+			row := f.row(i).Intersect(filter)
+			count := row.Count()
+			if count > 0 {
+				return i, count
+			}
+		}
+	}
+	return 0, 0
+}
+
 // rangeOp returns bitmaps with a bsiGroup value encoding matching the predicate.
 func (f *fragment) rangeOp(op pql.Token, bitDepth uint, predicate int64) (*Row, error) {
 	switch op {