From f15cb9e05ca4a5b1098f2be67851d74d01811d0c Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Thu, 30 May 2019 15:15:08 +0300
Subject: [PATCH 01/27] added roaring min

---
 roaring/containers.go       |  7 ++++
 roaring/containers_btree.go |  7 ++++
 roaring/containers_test.go  | 44 ++++++++++++++++++++++++-
 roaring/roaring.go          | 65 ++++++++++++++++++++++++++++++++++---
 roaring/roaring_test.go     | 26 +++++++++++++++
 5 files changed, 144 insertions(+), 5 deletions(-)

diff --git a/roaring/containers.go b/roaring/containers.go
index dc5d4118d..0cd515b9f 100644
--- a/roaring/containers.go
+++ b/roaring/containers.go
@@ -114,6 +114,13 @@ func (sc *sliceContainers) Clone() Containers {
 	return other
 }
 
+func (sc *sliceContainers) First() (key uint64, c *Container) {
+	if len(sc.keys) == 0 {
+		return 0, nil
+	}
+	return sc.keys[0], sc.containers[0]
+}
+
 func (sc *sliceContainers) Last() (key uint64, c *Container) {
 	if len(sc.keys) == 0 {
 		return 0, nil
diff --git a/roaring/containers_btree.go b/roaring/containers_btree.go
index 5934a1244..11f282a83 100644
--- a/roaring/containers_btree.go
+++ b/roaring/containers_btree.go
@@ -142,6 +142,13 @@ func (btc *bTreeContainers) Clone() Containers {
 	return nbtc
 }
 
+func (btc *bTreeContainers) First() (key uint64, c *Container) {
+	if btc.tree.Len() == 0 {
+		return 0, nil
+	}
+	return btc.tree.First()
+}
+
 func (btc *bTreeContainers) Last() (key uint64, c *Container) {
 	if btc.tree.Len() == 0 {
 		return 0, nil
diff --git a/roaring/containers_test.go b/roaring/containers_test.go
index ad95f2f79..6d159b73d 100644
--- a/roaring/containers_test.go
+++ b/roaring/containers_test.go
@@ -15,6 +15,7 @@
 package roaring
 
 import (
+	"reflect"
 	"testing"
 )
 
@@ -98,5 +99,46 @@ func testContainersIterator(cs Containers, t *testing.T) {
 	if itr.Next() {
 		t.Fatalf("itr should be done, but got true")
 	}
-
+}
+
+func TestContainers(t *testing.T) {
+	cs := NewFileBitmap().Containers
+	first := NewContainerArray([]uint16{1, 2, 3})
+	last := NewContainerArray([]uint16{1, 2, 3, 4, 5, 6})
+	cs.Put(3, first)
+	cs.Put(6, last)
+
+	key, container := cs.First()
+	if key != 3 {
+		t.Fatalf("cs.First key 3 != %d", key)
+	}
+	if !reflect.DeepEqual(first, container) {
+		t.Fatalf("cs.First container %v != %v", first, container)
+	}
+
+	key, container = cs.Last()
+	if key != 6 {
+		t.Fatalf("cs.First key 6 != %d", key)
+	}
+	if !reflect.DeepEqual(last, container) {
+		t.Fatalf("cs.First container %v != %v", last, container)
+	}
+
+	cs = NewFileBitmap().Containers
+
+	key, container = cs.First()
+	if key != 0 {
+		t.Fatalf("cs.First key 0 != %d", key)
+	}
+	if nil != container {
+		t.Fatalf("cs.First container nil != %v", container)
+	}
+
+	key, container = cs.Last()
+	if key != 0 {
+		t.Fatalf("cs.First key 0 != %d", key)
+	}
+	if nil != container {
+		t.Fatalf("cs.First container nil != %v", container)
+	}
 }
diff --git a/roaring/roaring.go b/roaring/roaring.go
index 2d1a2d08a..73b7e3dc6 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -95,6 +95,9 @@ type Containers interface {
 	// Clone does a deep copy of Containers, including cloning all containers contained.
 	Clone() Containers
 
+	// First returns the lowest key and associated container.
+	First() (key uint64, c *Container)
+
 	// Last returns the highest key and associated container.
 	Last() (key uint64, c *Container)
 
@@ -320,6 +323,18 @@ func (b *Bitmap) remove(v uint64) bool {
 	return c.remove(lowbits(v))
 }
 
+// Min returns the lowest value in the bitmap.
+// Second return value is true if containers exist in the bitmap.
+func (b *Bitmap) Min() (uint64, bool) {
+	if b.Containers.Size() == 0 {
+		return 0, false
+	}
+
+	hb, c := b.Containers.First()
+	lb, ok := c.min()
+	return hb<<16 | uint64(lb), ok
+}
+
 // Max returns the highest value in the bitmap.
 // Returns zero if the bitmap is empty.
 func (b *Bitmap) Max() uint64 {
@@ -1860,6 +1875,17 @@ func (c *Container) runRemove(v uint16) bool {
 	return true
 }
 
+// min returns the minimum value in the container.
+func (c *Container) min() (uint16, bool) {
+	if c.isArray() {
+		return c.arrayMin()
+	} else if c.isRun() {
+		return c.runMin()
+	} else {
+		return c.bitmapMin()
+	}
+}
+
 // max returns the maximum value in the container.
 func (c *Container) max() uint16 {
 	if c.isArray() {
@@ -1871,6 +1897,15 @@ func (c *Container) max() uint16 {
 	}
 }
 
+// Second result value is true if array is non-empty.
+func (c *Container) arrayMin() (uint16, bool) {
+	array := c.array()
+	if len(array) == 0 {
+		return 0, false
+	}
+	return array[0], true
+}
+
 func (c *Container) arrayMax() uint16 {
 	array := c.array()
 	if len(array) == 0 {
@@ -1879,21 +1914,43 @@ func (c *Container) arrayMax() uint16 {
 	return array[len(array)-1]
 }
 
+// Second result value is true if array is non-empty.
+func (c *Container) bitmapMin() (uint16, bool) {
+	bitmap := c.bitmap()
+	for i := 0; i < len(bitmap); i++ {
+		// If value is zero then skip.
+		v := bitmap[i]
+		if v != 0 {
+			r := bits.TrailingZeros64(v)
+			return uint16(r + i*64), true
+		}
+	}
+	return 0, false
+}
+
 func (c *Container) bitmapMax() uint16 {
 	// Search bitmap in reverse order.
 	bitmap := c.bitmap()
-	for i := len(bitmap); i > 0; i-- {
+	for i := len(bitmap) - 1; i > 0; i-- {
 		// If value is zero then skip.
-		v := bitmap[i-1]
+		v := bitmap[i]
 		if v != 0 {
 			r := bits.LeadingZeros64(v)
-			return uint16((i-1)*64 + 63 - r)
+			return uint16(i*64 + 63 - r)
 		}
-
 	}
 	return 0
 }
 
+// Second result value is true if array is non-empty.
+func (c *Container) runMin() (uint16, bool) {
+	runs := c.runs()
+	if len(runs) == 0 {
+		return 0, false
+	}
+	return runs[0].start, true
+}
+
 func (c *Container) runMax() uint16 {
 	runs := c.runs()
 	if len(runs) == 0 {
diff --git a/roaring/roaring_test.go b/roaring/roaring_test.go
index b4f792629..210ffa94d 100644
--- a/roaring/roaring_test.go
+++ b/roaring/roaring_test.go
@@ -309,6 +309,32 @@ func TestBitmap_Max(t *testing.T) {
 	}
 }
 
+// Ensure bitmap can return the lowest value.
+func TestBitmap_Min(t *testing.T) {
+	bm := roaring.NewFileBitmap()
+	for i := uint64(100000); i > 0; i-- {
+		if _, err := bm.Add(i); err != nil {
+			t.Fatalf("adding bits: %v", err)
+		}
+
+		v, ok := bm.Min()
+		if !ok {
+			t.Fatalf("ok should be true")
+		}
+
+		if v != i {
+			t.Fatalf("min: got=%d; want=%d", v, i)
+		}
+	}
+
+	// empty bitmap
+	bm = roaring.NewFileBitmap()
+	_, ok := bm.Min()
+	if ok {
+		t.Fatalf("ok should be false")
+	}
+}
+
 // Ensure CountRange is correct even if rangekey is prior to initial container.
 func TestBitmap_BitmapCountRangeEdgeCase(t *testing.T) {
 	s := uint64(2009 * pilosa.ShardWidth)

From 778ae1e8e25f8c1fb6b80dd555efed156a2cfd4b Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Thu, 30 May 2019 15:20:04 +0300
Subject: [PATCH 02/27] added enterprise btree first

---
 enterprise/b/containers_btree.go | 7 +++++++
 1 file changed, 7 insertions(+)

diff --git a/enterprise/b/containers_btree.go b/enterprise/b/containers_btree.go
index a051d7cde..eb7358a5d 100644
--- a/enterprise/b/containers_btree.go
+++ b/enterprise/b/containers_btree.go
@@ -149,6 +149,13 @@ func (btc *bTreeContainers) Clone() roaring.Containers {
 	return nbtc
 }
 
+func (btc *bTreeContainers) First() (key uint64, c *roaring.Container) {
+	if btc.tree.Len() == 0 {
+		return 0, nil
+	}
+	return btc.tree.First()
+}
+
 func (btc *bTreeContainers) Last() (key uint64, c *roaring.Container) {
 	if btc.tree.Len() == 0 {
 		return 0, nil

From d2aca3bbfc643d60b2d6accd02a3b575cf4aa1c4 Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Fri, 31 May 2019 15:32:15 +0300
Subject: [PATCH 03/27] Added MinRow and MaxRow calls

---
 executor.go | 141 ++++++++++++++++++++++++++++++++++++++++++++++++++++
 fragment.go |  20 +++++++-
 go.sum      |   1 +
 3 files changed, 160 insertions(+), 2 deletions(-)

diff --git a/executor.go b/executor.go
index 607840a42..56785ad48 100644
--- a/executor.go
+++ b/executor.go
@@ -264,6 +264,12 @@ func (e *executor) executeCall(ctx context.Context, index string, c *pql.Call, s
 	case "Max":
 		e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
 		return e.executeMax(ctx, index, c, shards, opt)
+	case "MinRow":
+		e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
+		return e.executeMinRow(ctx, index, c, shards, opt)
+	case "MaxRow":
+		e.Holder.Stats.CountWithCustomTags(c.Name, 1, 1.0, []string{indexTag})
+		return e.executeMaxRow(ctx, index, c, shards, opt)
 	case "Clear":
 		return e.executeClearBit(ctx, index, c, opt)
 	case "ClearRow":
@@ -468,6 +474,74 @@ func (e *executor) executeMax(ctx context.Context, index string, c *pql.Call, sh
 	return other, nil
 }
 
+// executeMinRow executes a MinRow() call.
+func (e *executor) executeMinRow(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMinRow")
+	defer span.Finish()
+
+	if field := c.Args["field"]; field == "" {
+		return ValCount{}, errors.New("MinRow(): field required")
+	}
+
+	// Execute calls in bulk on each remote node and merge.
+	mapFn := func(shard uint64) (interface{}, error) {
+		return e.executeMinRowShard(ctx, index, c, shard)
+	}
+
+	// Merge returned results at coordinating node.
+	reduceFn := func(prev, v interface{}) interface{} {
+		// if minRowID exists, and if it is smaller than the other one return it.
+		// otherwise return the minRowID of the one which exists.
+		prevp, _ := prev.(Pair)
+		vp, _ := v.(Pair)
+		if prevp.Count > 0 && vp.Count > 0 {
+			if prevp.ID < vp.ID {
+				return prevp
+			}
+			return vp
+		} else if prevp.Count > 0 {
+			return prevp
+		}
+		return vp
+	}
+
+	return e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
+}
+
+// executeMinRow executes a MaxRow() call.
+func (e *executor) executeMaxRow(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (interface{}, error) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxRow")
+	defer span.Finish()
+
+	if field := c.Args["field"]; field == "" {
+		return ValCount{}, errors.New("MaxRow(): field required")
+	}
+
+	// Execute calls in bulk on each remote node and merge.
+	mapFn := func(shard uint64) (interface{}, error) {
+		return e.executeMaxRowShard(ctx, index, c, shard)
+	}
+
+	// Merge returned results at coordinating node.
+	reduceFn := func(prev, v interface{}) interface{} {
+		// if minRowID exists, and if it is smaller than the other one return it.
+		// otherwise return the minRowID of the one which exists.
+		prevp, _ := prev.(Pair)
+		vp, _ := v.(Pair)
+		if prevp.Count > 0 && vp.Count > 0 {
+			if prevp.ID > vp.ID {
+				return prevp
+			}
+			return vp
+		} else if prevp.Count > 0 {
+			return prevp
+		}
+		return vp
+	}
+
+	return e.mapReduce(ctx, index, shards, c, opt, mapFn, reduceFn)
+}
+
 // executeBitmapCall executes a call that returns a bitmap.
 func (e *executor) executeBitmapCall(ctx context.Context, index string, c *pql.Call, shards []uint64, opt *execOptions) (*Row, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeBitmapCall")
@@ -648,6 +722,32 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal
 	}, nil
 }
 
+// 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) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMinRowShard")
+	defer span.Finish()
+
+	fieldName, _ := c.Args["field"].(string)
+	field := e.Holder.Field(index, fieldName)
+	if field == nil {
+		return Pair{}, nil
+	}
+
+	fragment := e.Holder.fragment(index, fieldName, viewStandard, shard)
+	if fragment == nil {
+		return Pair{}, nil
+	}
+
+	count := uint64(1)
+	if !fragment.hasRowID {
+		count = 0
+	}
+	return Pair{
+		ID:    fragment.minRowID,
+		Count: count,
+	}, nil
+}
+
 // executeMaxShard calculates the max for bsiGroups on a shard.
 func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
 	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxShard")
@@ -689,6 +789,32 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal
 	}, 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) {
+	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxRowShard")
+	defer span.Finish()
+
+	fieldName, _ := c.Args["field"].(string)
+	field := e.Holder.Field(index, fieldName)
+	if field == nil {
+		return Pair{}, nil
+	}
+
+	fragment := e.Holder.fragment(index, fieldName, viewStandard, shard)
+	if fragment == nil {
+		return Pair{}, nil
+	}
+
+	count := uint64(1)
+	if !fragment.hasRowID {
+		count = 0
+	}
+	return Pair{
+		ID:    fragment.maxRowID,
+		Count: count,
+	}, nil
+}
+
 // executeTopN executes a TopN() call.
 // This first performs the TopN() to determine the top results and then
 // requeries to retrieve the full counts for each of the top results.
@@ -2624,6 +2750,21 @@ func (e *executor) translateResult(index string, idx *Index, call *pql.Call, res
 			return other, nil
 		}
 
+	case Pair:
+		if fieldName := callArgString(call, "field"); fieldName != "" {
+			field := idx.Field(fieldName)
+			if field == nil {
+				return nil, fmt.Errorf("field %q not found", fieldName)
+			}
+			if field.keys() {
+				key, err := e.TranslateStore.TranslateRowToString(index, fieldName, result.ID)
+				if err != nil {
+					return nil, err
+				}
+				return Pair{Key: key, Count: result.Count}, nil
+			}
+		}
+
 	case []Pair:
 		if fieldName := callArgString(call, "_field"); fieldName != "" {
 			field := idx.Field(fieldName)
diff --git a/fragment.go b/fragment.go
index d9b9e8128..b162b965e 100644
--- a/fragment.go
+++ b/fragment.go
@@ -118,6 +118,8 @@ type fragment struct {
 
 	// Stats reporting.
 	maxRowID uint64
+	minRowID uint64
+	hasRowID bool
 
 	// Cache containing full rows (not just counts).
 	rowCache bitmapCache
@@ -188,8 +190,10 @@ func (f *fragment) Open() error {
 		f.checksums = make(map[int][]byte)
 
 		// Read last bit to determine max row.
-		pos := f.storage.Max()
-		f.maxRowID = pos / ShardWidth
+		f.maxRowID = f.storage.Max() / ShardWidth
+		min, ok := f.storage.Min()
+		f.minRowID = min / ShardWidth
+		f.hasRowID = ok
 		f.stats.Gauge("rows", float64(f.maxRowID), 1.0)
 		return nil
 	}(); err != nil {
@@ -519,6 +523,10 @@ func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err
 		f.maxRowID = rowID
 		f.stats.Gauge("rows", float64(f.maxRowID), 1.0)
 	}
+	if !f.hasRowID || rowID < f.minRowID {
+		f.minRowID = rowID
+		f.hasRowID = true
+	}
 
 	return changed, nil
 }
@@ -1023,6 +1031,14 @@ func (f *fragment) maxUnsigned(filter *Row, bitDepth uint) (max int64, count uin
 	return max, count
 }
 
+func (f *fragment) rowIDMin() (uint64, bool) {
+	return f.minRowID, f.hasRowID
+}
+
+func (f *fragment) rowIDMax() uint64 {
+	return f.maxRowID
+}
+
 // 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 {
diff --git a/go.sum b/go.sum
index 674e2adea..a66c2be51 100644
--- a/go.sum
+++ b/go.sum
@@ -86,6 +86,7 @@ github.com/shirou/gopsutil v2.18.12+incompatible h1:1eaJvGomDnH74/5cF4CTmTbLHAri
 github.com/shirou/gopsutil v2.18.12+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
 github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4 h1:udFKJ0aHUL60LboW/A+DfgoHVedieIzIXE8uylPue0U=
 github.com/shirou/w32 v0.0.0-20160930032740-bb4de0191aa4/go.mod h1:qsXQc7+bwAM3Q1u/4XEfrquwF8Lw7D7y5cD8CuHnfIc=
+github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72 h1:qLC7fQah7D6K1B0ujays3HV9gkFtllcxhzImRR7ArPQ=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spf13/afero v1.1.2 h1:m8/z1t7/fwjysjQRYbP0RD+bUIF/8tJwPdEZsI83ACI=
 github.com/spf13/afero v1.1.2/go.mod h1:j4pytiNVoe2o6bmDsKpLACNPDBIoEAkihy7loJ1B0CQ=

From dd728f28ede2a4372903a27d5b2d0c5acf4f2d3f Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Fri, 31 May 2019 17:13:25 +0300
Subject: [PATCH 04/27] remove unused code

---
 executor.go | 6 ------
 fragment.go | 8 --------
 2 files changed, 14 deletions(-)

diff --git a/executor.go b/executor.go
index 56785ad48..12789c3dd 100644
--- a/executor.go
+++ b/executor.go
@@ -724,9 +724,6 @@ func (e *executor) executeMinShard(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) {
-	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMinRowShard")
-	defer span.Finish()
-
 	fieldName, _ := c.Args["field"].(string)
 	field := e.Holder.Field(index, fieldName)
 	if field == nil {
@@ -750,9 +747,6 @@ func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql.
 
 // executeMaxShard calculates the max for bsiGroups on a shard.
 func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
-	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxShard")
-	defer span.Finish()
-
 	var filter *Row
 	if len(c.Children) == 1 {
 		row, err := e.executeBitmapCallShard(ctx, index, c.Children[0], shard)
diff --git a/fragment.go b/fragment.go
index b162b965e..87e7915b1 100644
--- a/fragment.go
+++ b/fragment.go
@@ -1031,14 +1031,6 @@ func (f *fragment) maxUnsigned(filter *Row, bitDepth uint) (max int64, count uin
 	return max, count
 }
 
-func (f *fragment) rowIDMin() (uint64, bool) {
-	return f.minRowID, f.hasRowID
-}
-
-func (f *fragment) rowIDMax() uint64 {
-	return f.maxRowID
-}
-
 // 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 {

From 592a191aee6b46dc353f5d9c49615cbcb76ee9f2 Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Fri, 31 May 2019 17:21:44 +0300
Subject: [PATCH 05/27] translate key into Pair only for MinRow, MaxRow

---
 executor.go | 4 ++++
 1 file changed, 4 insertions(+)

diff --git a/executor.go b/executor.go
index 12789c3dd..9f0ee8dac 100644
--- a/executor.go
+++ b/executor.go
@@ -2755,6 +2755,10 @@ func (e *executor) translateResult(index string, idx *Index, call *pql.Call, res
 				if err != nil {
 					return nil, err
 				}
+				if call.Name == "MinRow" || call.Name == "MaxRow" {
+					result.Key = key
+					return result, nil
+				}
 				return Pair{Key: key, Count: result.Count}, nil
 			}
 		}

From 1379cbbcd6f6fc56cd682c152a564c2d766075fa Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Fri, 31 May 2019 17:35:50 +0300
Subject: [PATCH 06/27] remove unused code

---
 executor.go | 3 ---
 1 file changed, 3 deletions(-)

diff --git a/executor.go b/executor.go
index 9f0ee8dac..8b07f9c9c 100644
--- a/executor.go
+++ b/executor.go
@@ -785,9 +785,6 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal
 
 // 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) {
-	span, ctx := tracing.StartSpanFromContext(ctx, "Executor.executeMaxRowShard")
-	defer span.Finish()
-
 	fieldName, _ := c.Args["field"].(string)
 	field := e.Holder.Field(index, fieldName)
 	if field == nil {

From 8be7bd69569d6b0e4a7a3d7086084f559d2e7574 Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Mon, 3 Jun 2019 13:56:50 +0300
Subject: [PATCH 07/27] add tests for MinRow and MaxRow

---
 executor.go      | 46 +++++++++++------------
 executor_test.go | 97 ++++++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 120 insertions(+), 23 deletions(-)

diff --git a/executor.go b/executor.go
index 8b07f9c9c..3529995f5 100644
--- a/executor.go
+++ b/executor.go
@@ -722,29 +722,6 @@ func (e *executor) executeMinShard(ctx context.Context, index string, c *pql.Cal
 	}, nil
 }
 
-// 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) {
-	fieldName, _ := c.Args["field"].(string)
-	field := e.Holder.Field(index, fieldName)
-	if field == nil {
-		return Pair{}, nil
-	}
-
-	fragment := e.Holder.fragment(index, fieldName, viewStandard, shard)
-	if fragment == nil {
-		return Pair{}, nil
-	}
-
-	count := uint64(1)
-	if !fragment.hasRowID {
-		count = 0
-	}
-	return Pair{
-		ID:    fragment.minRowID,
-		Count: count,
-	}, nil
-}
-
 // executeMaxShard calculates the max for bsiGroups on a shard.
 func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Call, shard uint64) (ValCount, error) {
 	var filter *Row
@@ -783,6 +760,29 @@ func (e *executor) executeMaxShard(ctx context.Context, index string, c *pql.Cal
 	}, nil
 }
 
+// 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) {
+	fieldName, _ := c.Args["field"].(string)
+	field := e.Holder.Field(index, fieldName)
+	if field == nil {
+		return Pair{}, nil
+	}
+
+	fragment := e.Holder.fragment(index, fieldName, viewStandard, shard)
+	if fragment == nil {
+		return Pair{}, nil
+	}
+
+	count := uint64(1)
+	if !fragment.hasRowID {
+		count = 0
+	}
+	return Pair{
+		ID:    fragment.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) {
 	fieldName, _ := c.Args["field"].(string)
diff --git a/executor_test.go b/executor_test.go
index 3565c40fd..be10b80bb 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -1415,6 +1415,103 @@ func TestExecutor_Execute_MinMax(t *testing.T) {
 	})
 }
 
+// Ensure MinRow() and MaxRow() queries can be executed.
+func TestExecutor_Execute_MinMaxRow(t *testing.T) {
+	t.Run("RowID", func(t *testing.T) {
+		c := test.MustRunCluster(t, 1)
+		defer c.Close()
+		hldr := test.Holder{Holder: c[0].Server.Holder()}
+
+		idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := idx.CreateField("f", pilosa.OptFieldTypeDefault()); err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `
+			Set(0, f=7000)
+			Set(3, f=50)
+			Set(` + strconv.Itoa(ShardWidth+1) + `, f=10000)
+			Set(1000, f=1)
+			Set(` + strconv.Itoa(ShardWidth+2) + `, f=5000)
+		`}); err != nil {
+			t.Fatal(err)
+		}
+
+		t.Run("MinRow", func(t *testing.T) {
+			result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "MinRow(field=f)"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			target := pilosa.Pair{ID: 1, Count: 1}
+			if !reflect.DeepEqual(target, result.Results[0]) {
+				t.Fatalf("unexpected result %v != %v", target, result.Results[0])
+			}
+		})
+
+		t.Run("MaxRow", func(t *testing.T) {
+			result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "MaxRow(field=f)"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			target := pilosa.Pair{ID: 10000, Count: 1}
+			if !reflect.DeepEqual(target, result.Results[0]) {
+				t.Fatalf("unexpected result %v != %v", target, result.Results[0])
+			}
+		})
+	})
+
+	t.Run("RowKey", func(t *testing.T) {
+		c := test.MustRunCluster(t, 1)
+		defer c.Close()
+		hldr := test.Holder{Holder: c[0].Server.Holder()}
+
+		idx, err := hldr.CreateIndex("i", pilosa.IndexOptions{})
+		if err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := idx.CreateField("f", pilosa.OptFieldKeys()); err != nil {
+			t.Fatal(err)
+		}
+
+		if _, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: `
+			Set(0, f="seven-thousand")
+			Set(3, f="fifty")
+			Set(` + strconv.Itoa(ShardWidth+1) + `, f="ten-thousand")
+			Set(1000, f="one")
+			Set(` + strconv.Itoa(ShardWidth+2) + `, f="five-thousand")
+		`}); err != nil {
+			t.Fatal(err)
+		}
+
+		t.Run("MinRow", func(t *testing.T) {
+			result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "MinRow(field=f)"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			target := pilosa.Pair{Key: "seven-thousand", ID: 1, Count: 1}
+			if !reflect.DeepEqual(target, result.Results[0]) {
+				t.Fatalf("unexpected result %v != %v", target, result.Results[0])
+			}
+		})
+
+		t.Run("MaxRow", func(t *testing.T) {
+			result, err := c[0].API.Query(context.Background(), &pilosa.QueryRequest{Index: "i", Query: "MaxRow(field=f)"})
+			if err != nil {
+				t.Fatal(err)
+			}
+			target := pilosa.Pair{Key: "five-thousand", ID: 5, Count: 1}
+			if !reflect.DeepEqual(target, result.Results[0]) {
+				t.Fatalf("unexpected result %v != %v", target, result.Results[0])
+			}
+		})
+	})
+}
+
 // Ensure a Sum() query can be executed.
 func TestExecutor_Execute_Sum(t *testing.T) {
 	t.Run("ColumnID", func(t *testing.T) {

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 08/27] 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 {

From 13a42d9c0783fdf25e79b832611124d938c8cb23 Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Tue, 11 Jun 2019 16:58:32 +0300
Subject: [PATCH 09/27] replaced min code with bmp.iterator

---
 executor.go                 |  2 +-
 roaring/containers_btree.go |  7 -----
 roaring/containers_slice.go |  7 -----
 roaring/containers_test.go  | 43 -----------------------------
 roaring/roaring.go          | 55 ++-----------------------------------
 5 files changed, 3 insertions(+), 111 deletions(-)

diff --git a/executor.go b/executor.go
index 7d474b6a4..7eb13dd79 100644
--- a/executor.go
+++ b/executor.go
@@ -789,7 +789,7 @@ func (e *executor) executeMinRowShard(ctx context.Context, index string, c *pql.
 	}, nil
 }
 
-// executeMaxRowShard returns the minimum row ID for a shard.
+// executeMaxRowShard returns the maximum 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 {
diff --git a/roaring/containers_btree.go b/roaring/containers_btree.go
index 2b1dd55a7..c2dbc01f5 100644
--- a/roaring/containers_btree.go
+++ b/roaring/containers_btree.go
@@ -164,13 +164,6 @@ func (btc *bTreeContainers) Freeze() Containers {
 	return nbtc
 }
 
-func (btc *bTreeContainers) First() (key uint64, c *Container) {
-	if btc.tree.Len() == 0 {
-		return 0, nil
-	}
-	return btc.tree.First()
-}
-
 func (btc *bTreeContainers) Last() (key uint64, c *Container) {
 	if btc.tree.Len() == 0 {
 		return 0, nil
diff --git a/roaring/containers_slice.go b/roaring/containers_slice.go
index 7a5722cf4..cbff4f179 100644
--- a/roaring/containers_slice.go
+++ b/roaring/containers_slice.go
@@ -131,13 +131,6 @@ func (sc *sliceContainers) Freeze() Containers {
 	return other
 }
 
-func (sc *sliceContainers) First() (key uint64, c *Container) {
-	if len(sc.keys) == 0 {
-		return 0, nil
-	}
-	return sc.keys[0], sc.containers[0]
-}
-
 func (sc *sliceContainers) Last() (key uint64, c *Container) {
 	if len(sc.keys) == 0 {
 		return 0, nil
diff --git a/roaring/containers_test.go b/roaring/containers_test.go
index 9d5e1610d..1869d31af 100644
--- a/roaring/containers_test.go
+++ b/roaring/containers_test.go
@@ -15,7 +15,6 @@
 package roaring
 
 import (
-	"reflect"
 	"testing"
 )
 
@@ -100,45 +99,3 @@ func testContainersIterator(cs Containers, t *testing.T) {
 		t.Fatalf("itr should be done, but got true")
 	}
 }
-
-func TestContainers(t *testing.T) {
-	cs := NewFileBitmap().Containers
-	first := NewContainerArray([]uint16{1, 2, 3})
-	last := NewContainerArray([]uint16{1, 2, 3, 4, 5, 6})
-	cs.Put(3, first)
-	cs.Put(6, last)
-
-	key, container := cs.First()
-	if key != 3 {
-		t.Fatalf("cs.First key 3 != %d", key)
-	}
-	if !reflect.DeepEqual(first, container) {
-		t.Fatalf("cs.First container %v != %v", first, container)
-	}
-
-	key, container = cs.Last()
-	if key != 6 {
-		t.Fatalf("cs.First key 6 != %d", key)
-	}
-	if !reflect.DeepEqual(last, container) {
-		t.Fatalf("cs.First container %v != %v", last, container)
-	}
-
-	cs = NewFileBitmap().Containers
-
-	key, container = cs.First()
-	if key != 0 {
-		t.Fatalf("cs.First key 0 != %d", key)
-	}
-	if nil != container {
-		t.Fatalf("cs.First container nil != %v", container)
-	}
-
-	key, container = cs.Last()
-	if key != 0 {
-		t.Fatalf("cs.First key 0 != %d", key)
-	}
-	if nil != container {
-		t.Fatalf("cs.First container nil != %v", container)
-	}
-}
diff --git a/roaring/roaring.go b/roaring/roaring.go
index 89dca0f17..7b7b5d144 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -103,9 +103,6 @@ type Containers interface {
 	// are shared (but marked as frozen).
 	Freeze() Containers
 
-	// First returns the lowest key and associated container.
-	First() (key uint64, c *Container)
-
 	// Last returns the highest key and associated container.
 	Last() (key uint64, c *Container)
 
@@ -385,13 +382,8 @@ func (b *Bitmap) remove(v uint64) bool {
 // Min returns the lowest value in the bitmap.
 // Second return value is true if containers exist in the bitmap.
 func (b *Bitmap) Min() (uint64, bool) {
-	if b.Containers.Size() == 0 {
-		return 0, false
-	}
-
-	hb, c := b.Containers.First()
-	lb, ok := c.min()
-	return hb<<16 | uint64(lb), ok
+	v, eof := b.Iterator().Next()
+	return v, !eof
 }
 
 // Max returns the highest value in the bitmap.
@@ -1995,17 +1987,6 @@ func (c *Container) runRemove(v uint16) (*Container, bool) {
 	return c, true
 }
 
-// min returns the minimum value in the container.
-func (c *Container) min() (uint16, bool) {
-	if c.isArray() {
-		return c.arrayMin()
-	} else if c.isRun() {
-		return c.runMin()
-	} else {
-		return c.bitmapMin()
-	}
-}
-
 // max returns the maximum value in the container.
 func (c *Container) max() uint16 {
 	if c == nil || c.N() == 0 {
@@ -2021,34 +2002,11 @@ func (c *Container) max() uint16 {
 	}
 }
 
-// Second result value is true if array is non-empty.
-func (c *Container) arrayMin() (uint16, bool) {
-	array := c.array()
-	if len(array) == 0 {
-		return 0, false
-	}
-	return array[0], true
-}
-
 func (c *Container) arrayMax() uint16 {
 	array := c.array()
 	return array[len(array)-1]
 }
 
-// Second result value is true if array is non-empty.
-func (c *Container) bitmapMin() (uint16, bool) {
-	bitmap := c.bitmap()
-	for i := 0; i < len(bitmap); i++ {
-		// If value is zero then skip.
-		v := bitmap[i]
-		if v != 0 {
-			r := bits.TrailingZeros64(v)
-			return uint16(r + i*64), true
-		}
-	}
-	return 0, false
-}
-
 func (c *Container) bitmapMax() uint16 {
 	// Search bitmap in reverse order.
 	bitmap := c.bitmap()
@@ -2063,15 +2021,6 @@ func (c *Container) bitmapMax() uint16 {
 	return 0
 }
 
-// Second result value is true if array is non-empty.
-func (c *Container) runMin() (uint16, bool) {
-	runs := c.runs()
-	if len(runs) == 0 {
-		return 0, false
-	}
-	return runs[0].start, true
-}
-
 func (c *Container) runMax() uint16 {
 	runs := c.runs()
 	if len(runs) == 0 {

From a1f6321b1dfbdc0d95e6eb9bfc1adae87b9ebac2 Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Thu, 13 Jun 2019 11:54:52 -0500
Subject: [PATCH 10/27] Added a test for slice bounds out of range

---
 roaring/fuzz_test.go | 23 +++++++++++++++++++++++
 1 file changed, 23 insertions(+)
 create mode 100644 roaring/fuzz_test.go

diff --git a/roaring/fuzz_test.go b/roaring/fuzz_test.go
new file mode 100644
index 000000000..c951b412e
--- /dev/null
+++ b/roaring/fuzz_test.go
@@ -0,0 +1,23 @@
+package roaring
+
+import ( 
+	"testing"
+)
+
+func TestUnmarshalBinary(t *testing.T) {
+	b := NewBitmap()
+	confirmedCrashers := []struct {
+		cr []byte
+	} {
+		{cr : []byte(":000000")},
+		{cr : []byte("<000000000000000")},
+	}
+
+	for _, crash := range confirmedCrashers {
+		err := b.UnmarshalBinary(crash.cr)
+		if err != nil {
+			t.Error("Known crasher failed.")
+		}
+	}
+	
+}
\ No newline at end of file

From 0a87d8108f82a865ebcd91e4278661d05a853b8a Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Fri, 14 Jun 2019 10:05:06 -0500
Subject: [PATCH 11/27] Added the actual bytes and their respective errors

---
 roaring/fuzz_test.go | 18 +++++++++++++-----
 1 file changed, 13 insertions(+), 5 deletions(-)

diff --git a/roaring/fuzz_test.go b/roaring/fuzz_test.go
index c951b412e..50a1492a2 100644
--- a/roaring/fuzz_test.go
+++ b/roaring/fuzz_test.go
@@ -8,16 +8,24 @@ func TestUnmarshalBinary(t *testing.T) {
 	b := NewBitmap()
 	confirmedCrashers := []struct {
 		cr []byte
+		expected string
 	} {
-		{cr : []byte(":000000")},
-		{cr : []byte("<000000000000000")},
+		{
+			cr : []byte(":0\x000\x01\x00\x00\x000000"),					//":000000"
+			expected : "reading roaring header: malformed bitmap, key-cardinality slice overruns buffer at 12",
+		},				
+		{
+			cr : []byte("<0\x000\x00\x00\x00\x00000000000000" +
+			"0"),														//"<000000000000000"
+			expected : "unmarshaling as pilosa roaring: too big",
+		},		
 	}
 
 	for _, crash := range confirmedCrashers {
 		err := b.UnmarshalBinary(crash.cr)
-		if err != nil {
-			t.Error("Known crasher failed.")
-		}
+		if err.Error() != crash.expected {
+			t.Errorf("Expected: %s, Got: %s", crash.expected, err)
+		} 
 	}
 	
 }
\ No newline at end of file

From 622fba4f2752503f62f2b4d90ca8837c425c19bf Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Fri, 14 Jun 2019 10:06:37 -0500
Subject: [PATCH 12/27] Fixed the <000000000 bug by adding if statement

---
 roaring/roaring.go | 3 +++
 1 file changed, 3 insertions(+)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index 8930981e0..5437bba83 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -3963,6 +3963,9 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	_, _ = h.Write(data[0:9])
 
 	if op.typ > 1 {
+		if 1152921504606847000 < int(op.value) {
+			return fmt.Errorf("too big")
+		}
 		if len(data) < int(13+op.value*8) {
 			return fmt.Errorf("op data truncated - expected %d, got %d", 13+op.value*8, len(data))
 		}

From 1e7638677b117269292155384911886f23959f78 Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Fri, 14 Jun 2019 10:08:23 -0500
Subject: [PATCH 13/27] Fixed the :000000 bug by adding an = in
 readOfficalHeader

---
 roaring/roaring.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index 5437bba83..023c75bc4 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -4458,7 +4458,7 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
 	}
 
 	// descriptive header
-	if pos+2*2*int(size) > len(buf) {
+	if pos+2*2*int(size) >= len(buf) {
 		err = fmt.Errorf("malformed bitmap, key-cardinality slice overruns buffer at %d", pos+2*2*int(size))
 		return size, containerTyper, header, pos, flags, haveRuns, err
 	}

From 56659f9d7b36f6d89b27dd87786e10f73b501165 Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Fri, 14 Jun 2019 11:47:33 -0500
Subject: [PATCH 14/27] Added Licensing

---
 roaring/fuzz_test.go | 13 +++++++++++++
 1 file changed, 13 insertions(+)

diff --git a/roaring/fuzz_test.go b/roaring/fuzz_test.go
index 50a1492a2..b11d00337 100644
--- a/roaring/fuzz_test.go
+++ b/roaring/fuzz_test.go
@@ -1,3 +1,16 @@
+// Copyright 2017 Pilosa Corp.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
 package roaring
 
 import ( 

From 77cb1ea6d8a3e1f8d54dbcf2690034bf3aa83a83 Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Fri, 14 Jun 2019 13:44:04 -0500
Subject: [PATCH 15/27] Claified the arithmetic behind the max op.value

---
 roaring/roaring.go | 5 ++++-
 1 file changed, 4 insertions(+), 1 deletion(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index 31916445f..f1593df9c 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -3961,7 +3961,10 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	_, _ = h.Write(data[0:9])
 
 	if op.typ > 1 {
-		if 1152921504606847000 < int(op.value) {
+		// The maximum integer value for a int64 is 9223372036854775807
+		maxInt := 9223372036854775807
+		maxOpValue := maxInt/8-13
+		if maxOpValue < int(op.value) {
 			return fmt.Errorf("too big")
 		}
 		if len(data) < int(13+op.value*8) {

From 734daf79ee99d9486b43602c9d398c3851572aaa Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Fri, 14 Jun 2019 14:30:21 -0500
Subject: [PATCH 16/27] Simplified the if statement and made the calculation
 more precise

---
 roaring/roaring.go | 11 +++++------
 1 file changed, 5 insertions(+), 6 deletions(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index f1593df9c..118a2f0ac 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -23,6 +23,7 @@ import (
 	"math/bits"
 	"sort"
 	"unsafe"
+	"math"
 
 	"github.com/pkg/errors"
 )
@@ -3956,17 +3957,15 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	// op.value will actually contain the length of values for batch ops
 	op.value = binary.LittleEndian.Uint64(data[1:9])
 
+	if math.MaxInt64/8-13 < int(op.value){
+		return fmt.Errorf("too big")
+	}
+
 	// Verify checksum.
 	h := fnv.New32a()
 	_, _ = h.Write(data[0:9])
 
 	if op.typ > 1 {
-		// The maximum integer value for a int64 is 9223372036854775807
-		maxInt := 9223372036854775807
-		maxOpValue := maxInt/8-13
-		if maxOpValue < int(op.value) {
-			return fmt.Errorf("too big")
-		}
 		if len(data) < int(13+op.value*8) {
 			return fmt.Errorf("op data truncated - expected %d, got %d", 13+op.value*8, len(data))
 		}

From 06aa2cf98e455b76222e2324479e53ff2cabe861 Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Sat, 15 Jun 2019 15:15:07 +0300
Subject: [PATCH 17/27] updated for feedback from PR 1983

---
 fragment.go        | 27 +++++++++++++--------------
 roaring/roaring.go |  6 +++---
 2 files changed, 16 insertions(+), 17 deletions(-)

diff --git a/fragment.go b/fragment.go
index c6cc03291..728fe9dee 100644
--- a/fragment.go
+++ b/fragment.go
@@ -118,8 +118,6 @@ type fragment struct {
 
 	// Stats reporting.
 	maxRowID uint64
-	minRowID uint64
-	hasRowID bool
 
 	// Cache containing full rows (not just counts).
 	rowCache bitmapCache
@@ -191,9 +189,6 @@ func (f *fragment) Open() error {
 
 		// Read last bit to determine max row.
 		f.maxRowID = f.storage.Max() / ShardWidth
-		min, ok := f.storage.Min()
-		f.minRowID = min / ShardWidth
-		f.hasRowID = ok
 		f.stats.Gauge("rows", float64(f.maxRowID), 1.0)
 		return nil
 	}(); err != nil {
@@ -525,10 +520,6 @@ func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err
 		f.maxRowID = rowID
 		f.stats.Gauge("rows", float64(f.maxRowID), 1.0)
 	}
-	if !f.hasRowID || rowID < f.minRowID {
-		f.minRowID = rowID
-		f.hasRowID = true
-	}
 
 	return changed, nil
 }
@@ -1040,12 +1031,13 @@ func (f *fragment) maxUnsigned(filter *Row, bitDepth uint) (max int64, count uin
 // 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 {
+	minRowID, hasRowID := f.minRowID()
+	if hasRowID {
 		if filter == nil {
-			return f.minRowID, 1
+			return minRowID, 1
 		}
 		// iterate from min row ID and return the first that intersects with filter.
-		for i := f.minRowID; i <= f.maxRowID; i++ {
+		for i := minRowID; i <= f.maxRowID; i++ {
 			row := f.row(i).Intersect(filter)
 			count := row.Count()
 			if count > 0 {
@@ -1060,12 +1052,14 @@ func (f *fragment) minRow(filter *Row) (uint64, uint64) {
 // 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 {
+	minRowID, hasRowID := f.minRowID()
+	if 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-- {
+		// TODO: implement reverse container iteration to improve performance here for sparse data. --Jaffee
+		for i := f.maxRowID; i >= minRowID; i-- {
 			row := f.row(i).Intersect(filter)
 			count := row.Count()
 			if count > 0 {
@@ -2419,6 +2413,11 @@ func (f *fragment) readCacheFromArchive(r io.Reader) error {
 	return nil
 }
 
+func (f *fragment) minRowID() (uint64, bool) {
+	min, ok := f.storage.Min()
+	return min / ShardWidth, ok
+}
+
 // rowFilter is a function signature for controlling iteration over containers
 // in a fragment. It will be invoked on each container found and returns two
 // booleans. The first is whether the row this container is in should be
diff --git a/roaring/roaring.go b/roaring/roaring.go
index 7addfc2f4..f7c3feb37 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -2008,12 +2008,12 @@ func (c *Container) arrayMax() uint16 {
 func (c *Container) bitmapMax() uint16 {
 	// Search bitmap in reverse order.
 	bitmap := c.bitmap()
-	for i := len(bitmap) - 1; i > 0; i-- {
+	for i := len(bitmap); i > 0; i-- {
 		// If value is zero then skip.
-		v := bitmap[i]
+		v := bitmap[i-1]
 		if v != 0 {
 			r := bits.LeadingZeros64(v)
-			return uint16(i*64 + 63 - r)
+			return uint16((i-1)*64 + 63 - r)
 		}
 	}
 	return 0

From 5c59449ed2326ac8dc7d915260bbd1055079bd0d Mon Sep 17 00:00:00 2001
From: Yuce Tekol <yucetekol@gmail.com>
Date: Mon, 17 Jun 2019 16:12:43 +0300
Subject: [PATCH 18/27] reset roaring.go and added bitmap.Min

---
 roaring/roaring.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index f7c3feb37..7ac6671b8 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -2015,6 +2015,7 @@ func (c *Container) bitmapMax() uint16 {
 			r := bits.LeadingZeros64(v)
 			return uint16((i-1)*64 + 63 - r)
 		}
+
 	}
 	return 0
 }

From 413492552c0e13333d56b17186de9f50be19b69f Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Mon, 17 Jun 2019 08:36:56 -0500
Subject: [PATCH 19/27] Rearranged if statement and declared maxOpSize value

---
 roaring/roaring.go | 8 ++++----
 1 file changed, 4 insertions(+), 4 deletions(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index 118a2f0ac..218227a61 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -3945,6 +3945,7 @@ func (op *op) WriteTo(w io.Writer) (n int64, err error) {
 }
 
 var minOpSize = 13
+var maxOpSize = math.MaxInt64/8 - 13
 
 // UnmarshalBinary decodes data into an op.
 func (op *op) UnmarshalBinary(data []byte) error {
@@ -3957,15 +3958,14 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	// op.value will actually contain the length of values for batch ops
 	op.value = binary.LittleEndian.Uint64(data[1:9])
 
-	if math.MaxInt64/8-13 < int(op.value){
-		return fmt.Errorf("too big")
-	}
-
 	// Verify checksum.
 	h := fnv.New32a()
 	_, _ = h.Write(data[0:9])
 
 	if op.typ > 1 {
+		if maxOpSize < int(op.value){
+			return fmt.Errorf("too big")
+		}
 		if len(data) < int(13+op.value*8) {
 			return fmt.Errorf("op data truncated - expected %d, got %d", 13+op.value*8, len(data))
 		}

From d9f2792d1fab73face48fbc7fcd10d0eac78c348 Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Mon, 17 Jun 2019 14:11:37 -0500
Subject: [PATCH 20/27] Reworded max int error and reset max int value

---
 roaring/roaring.go | 7 +++----
 1 file changed, 3 insertions(+), 4 deletions(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index 218227a61..db2910a8e 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -23,7 +23,6 @@ import (
 	"math/bits"
 	"sort"
 	"unsafe"
-	"math"
 
 	"github.com/pkg/errors"
 )
@@ -3945,7 +3944,7 @@ func (op *op) WriteTo(w io.Writer) (n int64, err error) {
 }
 
 var minOpSize = 13
-var maxOpSize = math.MaxInt64/8 - 13
+var maxOpN = 1000000
 
 // UnmarshalBinary decodes data into an op.
 func (op *op) UnmarshalBinary(data []byte) error {
@@ -3963,8 +3962,8 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	_, _ = h.Write(data[0:9])
 
 	if op.typ > 1 {
-		if maxOpSize < int(op.value){
-			return fmt.Errorf("too big")
+		if maxOpN < int(op.value){
+			return fmt.Errorf("Maximum operation size exceeded")
 		}
 		if len(data) < int(13+op.value*8) {
 			return fmt.Errorf("op data truncated - expected %d, got %d", 13+op.value*8, len(data))

From d24a15794775413e164e86e0ef0307a55fb6f78d Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Mon, 17 Jun 2019 15:58:20 -0500
Subject: [PATCH 21/27] Addressed review feedback

---
 roaring/fuzz_test.go | 2 +-
 roaring/roaring.go   | 6 ++++--
 2 files changed, 5 insertions(+), 3 deletions(-)

diff --git a/roaring/fuzz_test.go b/roaring/fuzz_test.go
index b11d00337..69362d947 100644
--- a/roaring/fuzz_test.go
+++ b/roaring/fuzz_test.go
@@ -30,7 +30,7 @@ func TestUnmarshalBinary(t *testing.T) {
 		{
 			cr : []byte("<0\x000\x00\x00\x00\x00000000000000" +
 			"0"),														//"<000000000000000"
-			expected : "unmarshaling as pilosa roaring: too big",
+			expected : "unmarshaling as pilosa roaring: Maximum operation size exceeded",
 		},		
 	}
 
diff --git a/roaring/roaring.go b/roaring/roaring.go
index db2910a8e..e10793915 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -3944,7 +3944,7 @@ func (op *op) WriteTo(w io.Writer) (n int64, err error) {
 }
 
 var minOpSize = 13
-var maxOpN = 1000000
+var maxBatchSize = 1<<59
 
 // UnmarshalBinary decodes data into an op.
 func (op *op) UnmarshalBinary(data []byte) error {
@@ -3962,7 +3962,9 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	_, _ = h.Write(data[0:9])
 
 	if op.typ > 1 {
-		if maxOpN < int(op.value){
+		// This ensures that in doing 13+op.value*8, the max int won't be exceeded and a wrap around case
+		// (resulting in a negative value) won't occur in the slice indexing while writing
+		if int(op.value) > maxBatchSize {
 			return fmt.Errorf("Maximum operation size exceeded")
 		}
 		if len(data) < int(13+op.value*8) {

From cae1a76629fbe20a2d394bc2e4a457ff05ae23e6 Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Mon, 17 Jun 2019 16:47:28 -0500
Subject: [PATCH 22/27] Fixed typo

---
 roaring/README.md | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/roaring/README.md b/roaring/README.md
index 044c98b83..53269cf88 100644
--- a/roaring/README.md
+++ b/roaring/README.md
@@ -18,7 +18,7 @@ The fuzzer needs some input to start the fuzzing with. Copy some sample Pilosa f
 
 Once you have copied your sample inputs, you are ready to run the fuzzer:
 
-`go-fuzz -bin=roaring-fuzz.zip -workdir=workdir -func=FuzzBitmapUnmarshalBianry`
+`go-fuzz -bin=roaring-fuzz.zip -workdir=workdir -func=FuzzBitmapUnmarshalBinary`
 
 ## Understanding the Fuzzer Output
 

From 97f525ff06b48a748abbc5dcfae51ec23dfb62db Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Tue, 18 Jun 2019 16:40:10 -0500
Subject: [PATCH 23/27] Removed fuzz_test.go

---
 roaring/fuzz_test.go | 44 --------------------------------------------
 1 file changed, 44 deletions(-)
 delete mode 100644 roaring/fuzz_test.go

diff --git a/roaring/fuzz_test.go b/roaring/fuzz_test.go
deleted file mode 100644
index b11d00337..000000000
--- a/roaring/fuzz_test.go
+++ /dev/null
@@ -1,44 +0,0 @@
-// Copyright 2017 Pilosa Corp.
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-package roaring
-
-import ( 
-	"testing"
-)
-
-func TestUnmarshalBinary(t *testing.T) {
-	b := NewBitmap()
-	confirmedCrashers := []struct {
-		cr []byte
-		expected string
-	} {
-		{
-			cr : []byte(":0\x000\x01\x00\x00\x000000"),					//":000000"
-			expected : "reading roaring header: malformed bitmap, key-cardinality slice overruns buffer at 12",
-		},				
-		{
-			cr : []byte("<0\x000\x00\x00\x00\x00000000000000" +
-			"0"),														//"<000000000000000"
-			expected : "unmarshaling as pilosa roaring: too big",
-		},		
-	}
-
-	for _, crash := range confirmedCrashers {
-		err := b.UnmarshalBinary(crash.cr)
-		if err.Error() != crash.expected {
-			t.Errorf("Expected: %s, Got: %s", crash.expected, err)
-		} 
-	}
-	
-}
\ No newline at end of file

From 2d151cd41a434ed7780165c54c2945001f0ead81 Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Tue, 18 Jun 2019 16:48:03 -0500
Subject: [PATCH 24/27] Making CI happy

---
 roaring/roaring.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index e10793915..cb1a7cb9b 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -3944,7 +3944,7 @@ func (op *op) WriteTo(w io.Writer) (n int64, err error) {
 }
 
 var minOpSize = 13
-var maxBatchSize = 1<<59
+var maxBatchSize = uint64(1<<59)
 
 // UnmarshalBinary decodes data into an op.
 func (op *op) UnmarshalBinary(data []byte) error {
@@ -3964,7 +3964,7 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	if op.typ > 1 {
 		// This ensures that in doing 13+op.value*8, the max int won't be exceeded and a wrap around case
 		// (resulting in a negative value) won't occur in the slice indexing while writing
-		if int(op.value) > maxBatchSize {
+		if int(op.value) > int(maxBatchSize) {
 			return fmt.Errorf("Maximum operation size exceeded")
 		}
 		if len(data) < int(13+op.value*8) {

From 3eed3b472fc6a3fae895374cadbc2ef52d169697 Mon Sep 17 00:00:00 2001
From: Ashley Svetlik <ash.svetlik@gmail.com>
Date: Wed, 19 Jun 2019 10:36:04 -0500
Subject: [PATCH 25/27] Corrected If statement logic error

---
 roaring/roaring.go | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index cb1a7cb9b..d7334be05 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -3964,7 +3964,7 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	if op.typ > 1 {
 		// This ensures that in doing 13+op.value*8, the max int won't be exceeded and a wrap around case
 		// (resulting in a negative value) won't occur in the slice indexing while writing
-		if int(op.value) > int(maxBatchSize) {
+		if op.value > maxBatchSize {
 			return fmt.Errorf("Maximum operation size exceeded")
 		}
 		if len(data) < int(13+op.value*8) {

From 481c85acae6dc3d3bdae92410a47a1d86d57c410 Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Thu, 20 Jun 2019 17:32:57 -0500
Subject: [PATCH 26/27] more info if nodeleave confirmation queries fail

---
 cluster.go | 6 ++++--
 1 file changed, 4 insertions(+), 2 deletions(-)

diff --git a/cluster.go b/cluster.go
index cf0090c0f..4c32f6b42 100644
--- a/cluster.go
+++ b/cluster.go
@@ -1709,15 +1709,17 @@ func confirmNodeDown(uri URI, log logger.Logger) bool {
 		log.Printf("bad request:%s %s", u.String(), err)
 		return false
 	}
-
 	for i := 0; i < confirmDownRetries; i++ {
 		resp, err := http.DefaultClient.Do(req.WithContext(ctx))
+		var bod []byte
 		if err == nil {
+			bod, err = ioutil.ReadAll(resp.Body)
 			if resp.StatusCode == 200 {
 				return false
 			}
 		}
-		log.Printf("NodeLeave Timeout with %s %d", uri.HostPort(), i)
+
+		log.Printf("NodeLeave confirm with %s %d. err: '%v' bod: '%s'", uri.HostPort(), i, err, bod)
 		time.Sleep(confirmDownSleep * time.Second)
 	}
 	return true

From c0d067b7ee4ad937c13bf5c3c3a8b2228af22a8a Mon Sep 17 00:00:00 2001
From: Matt Jaffee <matthew.jaffee@gmail.com>
Date: Fri, 21 Jun 2019 12:15:58 -0500
Subject: [PATCH 27/27] move context timeout inside loop, so context gets a
 fresh deadline

---
 cluster.go | 4 ++--
 1 file changed, 2 insertions(+), 2 deletions(-)

diff --git a/cluster.go b/cluster.go
index 4c32f6b42..c972daadd 100644
--- a/cluster.go
+++ b/cluster.go
@@ -1702,14 +1702,14 @@ func confirmNodeDown(uri URI, log logger.Logger) bool {
 		Host:   uri.HostPort(),
 		Path:   "version",
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), confirmDownTimeout*time.Second)
-	defer cancel()
 	req, err := http.NewRequest("GET", u.String(), nil)
 	if err != nil {
 		log.Printf("bad request:%s %s", u.String(), err)
 		return false
 	}
 	for i := 0; i < confirmDownRetries; i++ {
+		ctx, cancel := context.WithTimeout(context.Background(), confirmDownTimeout*time.Second)
+		defer cancel()
 		resp, err := http.DefaultClient.Do(req.WithContext(ctx))
 		var bod []byte
 		if err == nil {