From 4dc3e49a0d340d4c8aff79bc0efb5dda9495e385 Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Fri, 15 Jan 2021 22:42:37 +0300
Subject: [PATCH 01/20] Add test helper for inserting to time quantum fields

---
 test/cluster.go | 30 ++++++++++++++++++++++++++++++
 1 file changed, 30 insertions(+)

diff --git a/test/cluster.go b/test/cluster.go
index 006729ced..3af5c9b23 100644
--- a/test/cluster.go
+++ b/test/cluster.go
@@ -164,6 +164,36 @@ func (c *Cluster) ImportKeyKey(t testing.TB, index, field string, valAndRecKeys
 	}
 }
 
+// TimeQuantumKey is a string key and a string+key value
+type TimeQuantumKey struct {
+	RowKey string
+	ColKey string
+	Ts     int64
+}
+
+// ImportTimeQuantumKey imports data into an index where the index is keyd
+// and the field is a time-quantum
+func (c *Cluster) ImportTimeQuantumKey(t testing.TB, index, field string, entries []TimeQuantumKey) {
+	t.Helper()
+	importRequest := &pilosa.ImportRequest{
+		Index:      index,
+		Field:      field,
+		RowKeys:    make([]string, len(entries)),
+		ColumnKeys: make([]string, len(entries)),
+		Timestamps: make([]int64, len(entries)),
+	}
+	for i, entry := range entries {
+		importRequest.ColumnKeys[i] = entry.ColKey
+		importRequest.RowKeys[i] = entry.RowKey
+		importRequest.Timestamps[i] = entry.Ts
+
+	}
+	err := c.Nodes[0].API.Import(context.Background(), nil, importRequest)
+	if err != nil {
+		t.Fatalf("importing keykey data: %v", err)
+	}
+}
+
 // IntKey is a string key and a signed integer value.
 type IntKey struct {
 	Val int64

From 54d79d1e4d4c73b9ae7f99cd50a2d8f71b232205 Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Fri, 15 Jan 2021 22:50:54 +0300
Subject: [PATCH 02/20] Populate 'places_visited' field for 'users' index

---
 executor_test.go | 24 ++++++++++++++++++++++++
 1 file changed, 24 insertions(+)

diff --git a/executor_test.go b/executor_test.go
index 6db994817..f8e5f80a4 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -6834,6 +6834,30 @@ func variousQueries(t *testing.T, clusterSize int) {
 		{"icecream", "userF"},
 	})
 
+	// Create and populate "places_visited" time field.
+	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "places_visited", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YM")))
+	ts2019Jan01 := int64(1546300800) * 1e+9 // 2019 January 1st 0:00:00
+	ts2019Aug01 := int64(1564617600) * 1e+9 // 2019 August  1st 0:00:00
+	ts2020Jan01 := int64(1577836800) * 1e+9 // 2020 January 1st 0:00:00
+	c.ImportTimeQuantumKey(t, "users", "places_visited", []test.TimeQuantumKey{
+		// 2019 January: nairobi, paris, austin, toronto
+		{RowKey: "nairobi", ColKey: "userB", Ts: ts2019Jan01},
+		{RowKey: "paris", ColKey: "userC", Ts: ts2019Jan01},
+		{RowKey: "austin", ColKey: "userF", Ts: ts2019Jan01},
+		{RowKey: "toronto", ColKey: "userA", Ts: ts2019Jan01},
+		// 2019 August: nairobi, paris, austin, toronto
+		{RowKey: "toronto", ColKey: "userB", Ts: ts2019Aug01},
+		{RowKey: "toronto", ColKey: "userC", Ts: ts2019Aug01},
+		// 2020: toronto, mombasa, sydney, nairobi
+		{RowKey: "toronto", ColKey: "userB", Ts: ts2020Jan01},
+		{RowKey: "toronto", ColKey: "userD", Ts: ts2020Jan01},
+		{RowKey: "toronto", ColKey: "userE", Ts: ts2020Jan01},
+		{RowKey: "toronto", ColKey: "userF", Ts: ts2020Jan01},
+		{RowKey: "mombasa", ColKey: "userA", Ts: ts2020Jan01},
+		{RowKey: "sydney", ColKey: "userD", Ts: ts2020Jan01},
+		{RowKey: "nairobi", ColKey: "userE", Ts: ts2020Jan01},
+	})
+
 	// Create and populate "affinity" int field with negative, positive, zero and null values.
 	c.CreateField(t, "users", pilosa.IndexOptions{Keys: true, TrackExistence: true}, "affinity", pilosa.OptFieldTypeInt(-1000, 1000))
 	c.ImportIntKey(t, "users", "affinity", []test.IntKey{

From f91b4bc016075af824841acc24362f95ada65359 Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Fri, 15 Jan 2021 22:58:18 +0300
Subject: [PATCH 03/20] Add queries to test GroupBy Rows on time field

---
 executor_test.go | 26 ++++++++++++++++++++++++++
 1 file changed, 26 insertions(+)

diff --git a/executor_test.go b/executor_test.go
index f8e5f80a4..10db42a63 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -6895,6 +6895,32 @@ func variousQueries(t *testing.T, clusterSize int) {
 		qrVerifier  func(t *testing.T, resp pilosa.QueryResponse)
 		csvVerifier string
 	}{
+		{ // 2020 & 2019 All
+			query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2020-12-31T23:59'))`,
+			csvVerifier: `nairobi,2,0
+paris,1,0
+austin,1,0
+toronto,6,0
+mombasa,1,0
+sydney,1,0
+`,
+		},
+		{ // 2019 All
+			query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'))`,
+			csvVerifier: `nairobi,1,0
+paris,1,0
+austin,1,0
+toronto,3,0
+`,
+		},
+		{ // 2019 January only
+			query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-02-01T00:00'))`,
+			csvVerifier: `nairobi,1,0
+paris,1,0
+austin,1,0
+toronto,1,0
+`,
+		},
 		{
 			query: "Count(All())",
 			qrVerifier: func(t *testing.T, resp pilosa.QueryResponse) {

From b77f9e8a431a1fa6835a02b4544c08ed48ccdee7 Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Fri, 15 Jan 2021 23:36:54 +0300
Subject: [PATCH 04/20] Add timeFragments rowIterator

---
 executor.go | 80 +++++++++++++++++++++++++++++++++++++++++++++--------
 fragment.go | 69 +++++++++++++++++++++++++++++++++++++++++++++
 2 files changed, 137 insertions(+), 12 deletions(-)

diff --git a/executor.go b/executor.go
index 722f63d32..ff5317b04 100644
--- a/executor.go
+++ b/executor.go
@@ -7262,9 +7262,11 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children
 	idx := holder.Index(index)
 
 	var (
-		fieldName string
-		viewName  string
-		ok        bool
+		fieldName   string
+		viewName    string
+		ok          bool
+		views       []string
+		isTimeField bool
 	)
 	ignorePrev := false
 	for i, call := range children {
@@ -7278,9 +7280,42 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children
 		gbi.fields[i].Field = fieldName
 
 		switch field.Type() {
-		case FieldTypeSet, FieldTypeTime, FieldTypeMutex, FieldTypeBool:
+		case FieldTypeSet, FieldTypeMutex, FieldTypeBool:
 			viewName = viewStandard
+		case FieldTypeTime:
+			var (
+				err error
+				v   interface{}
+			)
 
+			// Parse "from" time, if set.
+			var (
+				hasFrom  bool
+				fromTime time.Time
+			)
+			if v, hasFrom = call.Args["from"]; hasFrom {
+				if fromTime, err = parseTime(v); err != nil {
+					return nil, errors.Wrap(err, "parsing from time")
+				}
+			}
+
+			// Parse "to" time, if set.
+			var (
+				hasTo  bool
+				toTime time.Time
+			)
+			if v, hasTo = call.Args["to"]; hasTo {
+				if toTime, err = parseTime(v); err != nil {
+					return nil, errors.Wrap(err, "parsing to time")
+				}
+			}
+
+			if hasTo || hasFrom {
+				views = viewsByTimeRange(viewStandard, fromTime, toTime, field.TimeQuantum())
+				isTimeField = true
+			} else {
+				viewName = viewStandard
+			}
 		case FieldTypeInt:
 			viewName = viewBSIGroupPrefix + fieldName
 
@@ -7289,11 +7324,6 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children
 				call.Name, strings.Join([]string{FieldTypeSet, FieldTypeTime, FieldTypeMutex, FieldTypeBool, FieldTypeInt}, ","))
 		}
 
-		// Fetch fragment.
-		frag := holder.fragment(index, fieldName, viewName, shard)
-		if frag == nil { // this means this whole shard doesn't have all it needs to continue
-			return nil, nil
-		}
 		filters := []roaring.BitmapFilter{}
 		if len(rowIDs[i]) > 0 {
 			filters = append(filters, roaring.NewBitmapRowsFilter(rowIDs[i]))
@@ -7305,9 +7335,35 @@ func newGroupByIterator(executor *executor, qcx *Qcx, rowIDs []RowIDs, children
 		}
 		defer finisher(&err0)
 
-		gbi.rowIters[i], err = frag.rowIterator(tx, i != 0, filters...)
-		if err != nil {
-			return nil, err
+		// Fetch fragment(s), get rowIterator
+		if isTimeField {
+			var fragments []*fragment
+			for _, viewName := range views {
+				fragment := holder.fragment(index, fieldName, viewName, shard)
+				if fragment != nil {
+					fragments = append(fragments, fragment)
+				}
+			}
+			if len(fragments) == 0 {
+				// whole shard doesn't have all it needs to continue ?
+				return nil, nil
+			}
+
+			gbi.rowIters[i], err = timeFragmentsRowIterator(fragments, tx, i != 0, filters...)
+			if err != nil {
+				return nil, err
+			}
+		} else {
+			frag := holder.fragment(index, fieldName, viewName, shard)
+			if frag == nil { // this means this whole shard doesn't have all it needs to continue
+				return nil, nil
+			}
+
+			gbi.rowIters[i], err = frag.rowIterator(tx, i != 0, filters...)
+			if err != nil {
+				return nil, err
+			}
+
 		}
 
 		prev, hasPrev, err := call.UintArg("previous")
diff --git a/fragment.go b/fragment.go
index aa93dcc9a..031043390 100644
--- a/fragment.go
+++ b/fragment.go
@@ -3329,6 +3329,75 @@ func (f *fragment) rowIterator(tx Tx, wrap bool, filters ...roaring.BitmapFilter
 	return f.setRowIterator(tx, wrap, filters...)
 }
 
+type timeRowIterator struct {
+	tx        Tx
+	fragments []*fragment
+	rows      []*Row
+	rowIDs    [][]uint64
+	cur       int
+	wrap      bool
+}
+
+func timeFragmentsRowIterator(fragments []*fragment, tx Tx, wrap bool, filters ...roaring.BitmapFilter) (rowIterator, error) {
+	if len(fragments) == 0 {
+		return nil, fmt.Errorf("there should be at least 1 fragment")
+	} else if len(fragments) == 1 {
+		return fragments[0].setRowIterator(tx, wrap, filters...)
+	}
+
+	it := &timeRowIterator{
+		tx:        tx,
+		fragments: fragments,
+		rows:      make([]*Row, len(fragments)),
+		rowIDs:    make([][]uint64, len(fragments)),
+	}
+
+	for i, f := range fragments {
+		rows, err := f.rows(context.Background(), tx, 0, filters...)
+		if err != nil {
+			return nil, err
+		}
+		it.rowIDs[i] = rows
+	}
+
+	return it, nil
+}
+
+func (it *timeRowIterator) Seek(rowID uint64) {
+	rowIDs := it.rowIDs[0]
+	idx := sort.Search(len(rowIDs), func(i int) bool {
+		return rowIDs[i] >= rowID
+	})
+	it.cur = idx
+}
+
+func (it *timeRowIterator) Next() (r *Row, rowID uint64, _ *int64, wrapped bool, err error) {
+	rowIDs := it.rowIDs[0]
+	if it.cur >= len(rowIDs) {
+		if !it.wrap || len(rowIDs) == 0 {
+			return nil, 0, nil, true, nil
+		}
+		it.Seek(0)
+		wrapped = true
+	}
+
+	id := rowIDs[it.cur]
+	// gather rows
+	for i, fragment := range it.fragments {
+		row, err := fragment.row(it.tx, id)
+		if err != nil {
+			return row, rowID, nil, wrapped, err
+		}
+		it.rows[i] = row
+	}
+
+	// union rows
+	r = it.rows[0].Union(it.rows[1:]...)
+
+	it.cur++
+	return r, rowID, nil, wrapped, nil
+}
+
 type intRowIterator struct {
 	f      *fragment
 	values int64Slice         // sorted slice of int values

From f51ff4dc85fe46b589aa1f0dffe48cbc56674f13 Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Mon, 18 Jan 2021 20:54:31 +0300
Subject: [PATCH 05/20] Add tests for Rows() on time fields

---
 executor_test.go | 12 ++++++++++++
 1 file changed, 12 insertions(+)

diff --git a/executor_test.go b/executor_test.go
index 10db42a63..613cd99c2 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -6921,6 +6921,18 @@ austin,1,0
 toronto,1,0
 `,
 		},
+		{ // 2020 & 2019 All
+			query:       `Rows(places_visited, from='2019-01-01T00:00', to='2020-12-31T23:59')`,
+			csvVerifier: "nairobi\nparis\naustin\ntoronto\nmombasa\nsydney\n",
+		},
+		{ // 2019 All
+			query:       `Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59')`,
+			csvVerifier: "nairobi\nparis\naustin\ntoronto\n",
+		},
+		{ // 2019 January only
+			query:       `Rows(places_visited, from='2019-01-01T00:00', to='2019-02-01T00:00')`,
+			csvVerifier: "nairobi\nparis\naustin\ntoronto\n",
+		},
 		{
 			query: "Count(All())",
 			qrVerifier: func(t *testing.T, resp pilosa.QueryResponse) {

From 9b23dcdd0a34e07e168dfa1d91f4891f327555be Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Tue, 19 Jan 2021 01:44:54 +0300
Subject: [PATCH 06/20] Gather rows for each fragment in a much smarter way

---
 fragment.go | 71 +++++++++++++++++++++++++++++++++++------------------
 1 file changed, 47 insertions(+), 24 deletions(-)

diff --git a/fragment.go b/fragment.go
index 031043390..5038646f3 100644
--- a/fragment.go
+++ b/fragment.go
@@ -3330,12 +3330,11 @@ func (f *fragment) rowIterator(tx Tx, wrap bool, filters ...roaring.BitmapFilter
 }
 
 type timeRowIterator struct {
-	tx        Tx
-	fragments []*fragment
-	rows      []*Row
-	rowIDs    [][]uint64
-	cur       int
-	wrap      bool
+	tx               Tx
+	cur              int
+	wrap             bool
+	allRowIDs        []uint64
+	rowIDToFragments map[uint64][]*fragment
 }
 
 func timeFragmentsRowIterator(fragments []*fragment, tx Tx, wrap bool, filters ...roaring.BitmapFilter) (rowIterator, error) {
@@ -3346,53 +3345,77 @@ func timeFragmentsRowIterator(fragments []*fragment, tx Tx, wrap bool, filters .
 	}
 
 	it := &timeRowIterator{
-		tx:        tx,
-		fragments: fragments,
-		rows:      make([]*Row, len(fragments)),
-		rowIDs:    make([][]uint64, len(fragments)),
+		tx:   tx,
+		cur:  0,
+		wrap: wrap,
 	}
 
-	for i, f := range fragments {
-		rows, err := f.rows(context.Background(), tx, 0, filters...)
+	// create a sort of inverted index that maps each
+	// rowID back to the fragments that have that rowID
+	rowIDToFragments := make(map[uint64][]*fragment)
+	for _, f := range fragments {
+		rowIDs, err := f.rows(context.Background(), tx, 0, filters...)
 		if err != nil {
 			return nil, err
 		}
-		it.rowIDs[i] = rows
+		for _, rowID := range rowIDs {
+			fs := append(rowIDToFragments[rowID], f)
+			rowIDToFragments[rowID] = fs
+		}
 	}
 
+	// if len(rowIDToFragments) == 0 what to do ??
+	// ie all fragments returned empty rowIDs, is this possible
+	// is this an error
+
+	// collect all rowIDs from inverted index to a slice
+	allRowIDs := make([]uint64, len(rowIDToFragments))
+	i := 0
+	for rowID := range rowIDToFragments {
+		allRowIDs[i] = rowID
+		i++
+	}
+
+	it.rowIDToFragments = rowIDToFragments
+	it.allRowIDs = allRowIDs
+
 	return it, nil
 }
 
 func (it *timeRowIterator) Seek(rowID uint64) {
-	rowIDs := it.rowIDs[0]
-	idx := sort.Search(len(rowIDs), func(i int) bool {
-		return rowIDs[i] >= rowID
+	idx := sort.Search(len(it.allRowIDs), func(i int) bool {
+		return it.allRowIDs[i] >= rowID
 	})
 	it.cur = idx
 }
 
 func (it *timeRowIterator) Next() (r *Row, rowID uint64, _ *int64, wrapped bool, err error) {
-	rowIDs := it.rowIDs[0]
-	if it.cur >= len(rowIDs) {
-		if !it.wrap || len(rowIDs) == 0 {
+	if it.cur >= len(it.allRowIDs) {
+		if !it.wrap || len(it.allRowIDs) == 0 {
 			return nil, 0, nil, true, nil
 		}
 		it.Seek(0)
 		wrapped = true
 	}
 
-	id := rowIDs[it.cur]
 	// gather rows
-	for i, fragment := range it.fragments {
-		row, err := fragment.row(it.tx, id)
+	rowID = it.allRowIDs[it.cur]
+	fragments := it.rowIDToFragments[rowID]
+	rows := make([]*Row, len(fragments))
+	for i, fragment := range fragments {
+		row, err := fragment.row(it.tx, rowID)
 		if err != nil {
 			return row, rowID, nil, wrapped, err
 		}
-		it.rows[i] = row
+		rows[i] = row
 	}
 
 	// union rows
-	r = it.rows[0].Union(it.rows[1:]...)
+	if len(rows) > 1 {
+		r = rows[0].Union(rows[1:]...)
+	} else {
+		r = rows[0]
+	}
 
 	it.cur++
 	return r, rowID, nil, wrapped, nil

From d26c6b048ef280eafe971ba1c97f08eb5683f2b0 Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Tue, 19 Jan 2021 02:00:50 +0300
Subject: [PATCH 07/20] Sort all rowIDs gathered before storing them

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

diff --git a/fragment.go b/fragment.go
index 5038646f3..522dc0128 100644
--- a/fragment.go
+++ b/fragment.go
@@ -3375,6 +3375,7 @@ func timeFragmentsRowIterator(fragments []*fragment, tx Tx, wrap bool, filters .
 		allRowIDs[i] = rowID
 		i++
 	}
+	sort.Slice(allRowIDs, func(i, j int) bool { return allRowIDs[i] < allRowIDs[j] })
 
 	it.rowIDToFragments = rowIDToFragments
 	it.allRowIDs = allRowIDs

From 0437f5d28a8f84ce1ef6176b92e1a551cc470c03 Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Tue, 19 Jan 2021 16:51:26 +0300
Subject: [PATCH 08/20] Handle case where row returned might be nil

---
 executor_test.go |  2 +-
 fragment.go      | 12 ++++++------
 2 files changed, 7 insertions(+), 7 deletions(-)

diff --git a/executor_test.go b/executor_test.go
index 613cd99c2..a8ba5335f 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -6845,7 +6845,7 @@ func variousQueries(t *testing.T, clusterSize int) {
 		{RowKey: "paris", ColKey: "userC", Ts: ts2019Jan01},
 		{RowKey: "austin", ColKey: "userF", Ts: ts2019Jan01},
 		{RowKey: "toronto", ColKey: "userA", Ts: ts2019Jan01},
-		// 2019 August: nairobi, paris, austin, toronto
+		// 2019 August: toronto only
 		{RowKey: "toronto", ColKey: "userB", Ts: ts2019Aug01},
 		{RowKey: "toronto", ColKey: "userC", Ts: ts2019Aug01},
 		// 2020: toronto, mombasa, sydney, nairobi
diff --git a/fragment.go b/fragment.go
index 522dc0128..72d14198a 100644
--- a/fragment.go
+++ b/fragment.go
@@ -3402,20 +3402,20 @@ func (it *timeRowIterator) Next() (r *Row, rowID uint64, _ *int64, wrapped bool,
 	// gather rows
 	rowID = it.allRowIDs[it.cur]
 	fragments := it.rowIDToFragments[rowID]
-	rows := make([]*Row, len(fragments))
-	for i, fragment := range fragments {
+	rows := make([]*Row, 0, len(fragments))
+	for _, fragment := range fragments {
 		row, err := fragment.row(it.tx, rowID)
 		if err != nil {
 			return row, rowID, nil, wrapped, err
 		}
-		rows[i] = row
+		if row != nil {
+			rows = append(rows, row)
+		}
 	}
 
 	// union rows
-	if len(rows) > 1 {
+	if len(rows) > 0 {
 		r = rows[0].Union(rows[1:]...)
-	} else {
-		r = rows[0]
 	}
 
 	it.cur++

From 113bec24748e8f5b787d433ba7e19a272f930179 Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Thu, 21 Jan 2021 15:48:52 +0300
Subject: [PATCH 09/20] Remove unnecessary nil check

---
 fragment.go | 8 ++------
 1 file changed, 2 insertions(+), 6 deletions(-)

diff --git a/fragment.go b/fragment.go
index 72d14198a..8592182f4 100644
--- a/fragment.go
+++ b/fragment.go
@@ -3408,15 +3408,11 @@ func (it *timeRowIterator) Next() (r *Row, rowID uint64, _ *int64, wrapped bool,
 		if err != nil {
 			return row, rowID, nil, wrapped, err
 		}
-		if row != nil {
-			rows = append(rows, row)
-		}
+		rows = append(rows, row)
 	}
 
 	// union rows
-	if len(rows) > 0 {
-		r = rows[0].Union(rows[1:]...)
-	}
+	r = rows[0].Union(rows[1:]...)
 
 	it.cur++
 	return r, rowID, nil, wrapped, nil

From 6cbcde3a82c0c77450dabfbca6dfa9962bfb51ec Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Thu, 21 Jan 2021 16:20:46 +0300
Subject: [PATCH 10/20] Add more tests for GroupBy on Time fields

---
 executor_test.go | 30 ++++++++++++++++++++++++++----
 1 file changed, 26 insertions(+), 4 deletions(-)

diff --git a/executor_test.go b/executor_test.go
index a8ba5335f..4f03fef9e 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -6903,6 +6903,14 @@ austin,1,0
 toronto,6,0
 mombasa,1,0
 sydney,1,0
+`,
+		},
+		{ // 2019 January only
+			query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-02-01T00:00'))`,
+			csvVerifier: `nairobi,1,0
+paris,1,0
+austin,1,0
+toronto,1,0
 `,
 		},
 		{ // 2019 All
@@ -6913,12 +6921,26 @@ austin,1,0
 toronto,3,0
 `,
 		},
-		{ // 2019 January only
-			query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-02-01T00:00'))`,
+		{ // 2019 All, this excludes userC (who likes pangolin & icecream) from the count.
+			// UserC visited Paris and Toronto in 2019
+			query: `GroupBy(
+					Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), 
+					filter=Not(Intersect(Row(likes='pangolin'), Row(likes='icecream')))
+				)`,
 			csvVerifier: `nairobi,1,0
-paris,1,0
 austin,1,0
-toronto,1,0
+toronto,2,0
+`,
+		},
+		{ // After excluding UserC, this gets the sum of the networth of everyone per cities travelled
+			query: `GroupBy(
+					Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), 
+					filter=Not(Intersect(Row(likes='pangolin'), Row(likes='icecream'))),
+					aggregate=Sum(field=net_worth)
+				)`,
+			csvVerifier: `nairobi,1,10
+austin,1,100000
+toronto,2,11
 `,
 		},
 		{ // 2020 & 2019 All

From 0d9179f10c0132e91ec5b0ebfbb1e016d9e7de16 Mon Sep 17 00:00:00 2001
From: Nia Weiss <nia.weiss@molecula.com>
Date: Thu, 21 Jan 2021 11:11:20 -0500
Subject: [PATCH 11/20] roaring: fix intersectionAnyRunBitmap when processing
 single-word runs

When a run started and ended within a single word, the entirety of the word would be checked.
This would cause small runs to be processed incorrectly, and caused Distinct-on-sets to select rows that did not match the specified filter.
---
 roaring/roaring.go                | 26 +++++++++++++-------------
 roaring/roaring_container_test.go | 12 ++++++++++++
 2 files changed, 25 insertions(+), 13 deletions(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index 2705a6ba3..9c3a97aa9 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -4144,26 +4144,26 @@ func intersectionAnyRunBitmap(a, b *Container) bool {
 	bb := b.bitmap()[:1024]
 	runs := a.runs()
 	for _, r := range runs {
-		loWord, loBit := r.Start/64, r.Start%64
-		hiWord, hiBit := r.Last/64, r.Last%64
-		if loBit != 0 {
-			w := bb[loWord]
-			mask := (uint64(1) << loBit) - 1
-			if w&^mask != 0 {
+		if r.Start/64 == r.Last/64 {
+			mask := (^uint64(0) << (r.Start % 64)) &^
+				(^uint64(0) << ((r.Last % 64) + 1))
+			if mask&bb[r.Start/64] != 0 {
 				return true
 			}
+			continue
 		}
-		for i := loWord; i < hiWord; i++ {
+
+		firstWord, lastWord := r.Start/64, r.Last/64
+		for i := firstWord + 1; i < lastWord; i++ {
 			if bb[i] != 0 {
 				return true
 			}
 		}
-		if hiBit != 0 {
-			w := bb[hiWord]
-			mask := (uint64(1) << hiBit) - 1
-			if w&mask != 0 {
-				return true
-			}
+
+		firstMask := ^uint64(0) << (r.Start % 64)
+		lastMask := ^(^uint64(0) << ((r.Last % 64) + 1))
+		if (firstMask&bb[firstWord])|(lastMask&bb[lastWord]) != 0 {
+			return true
 		}
 	}
 	return false
diff --git a/roaring/roaring_container_test.go b/roaring/roaring_container_test.go
index 3b747b5c9..9fbf4b214 100644
--- a/roaring/roaring_container_test.go
+++ b/roaring/roaring_container_test.go
@@ -97,3 +97,15 @@ func TestIntersectVariants(t *testing.T) {
 		}
 	}
 }
+
+func TestIntersectionAnyRunBitmapSingleWordRegression(t *testing.T) {
+	// In a previous version, single-word runs would match any bit within the word.
+	// Verify that this no longer happens.
+	any := intersectionAnyRunBitmap(
+		NewContainerRun([]Interval16{{1, 2}}),
+		NewContainerBitmapN([]uint64{0b1001}, 2),
+	)
+	if any {
+		t.Errorf("matched an exclusive single-word run")
+	}
+}

From b5fdc6609addcb7f942c23434fe52deaf195a474 Mon Sep 17 00:00:00 2001
From: nagamocha3000 <nagamocha@gmail.com>
Date: Fri, 22 Jan 2021 18:44:36 +0300
Subject: [PATCH 12/20] Update tests to remove sum column on GroupBy

---
 executor_test.go | 34 +++++++++++++++++-----------------
 1 file changed, 17 insertions(+), 17 deletions(-)

diff --git a/executor_test.go b/executor_test.go
index cd5dbf372..c0532b4f9 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -6926,28 +6926,28 @@ func variousQueries(t *testing.T, clusterSize int) {
 	}{
 		{ // 2020 & 2019 All
 			query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2020-12-31T23:59'))`,
-			csvVerifier: `nairobi,2,0
-paris,1,0
-austin,1,0
-toronto,6,0
-mombasa,1,0
-sydney,1,0
+			csvVerifier: `nairobi,2
+paris,1
+austin,1
+toronto,6
+mombasa,1
+sydney,1
 `,
 		},
 		{ // 2019 January only
 			query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-02-01T00:00'))`,
-			csvVerifier: `nairobi,1,0
-paris,1,0
-austin,1,0
-toronto,1,0
+			csvVerifier: `nairobi,1
+paris,1
+austin,1
+toronto,1
 `,
 		},
 		{ // 2019 All
 			query: `GroupBy(Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'))`,
-			csvVerifier: `nairobi,1,0
-paris,1,0
-austin,1,0
-toronto,3,0
+			csvVerifier: `nairobi,1
+paris,1
+austin,1
+toronto,3
 `,
 		},
 		{ // 2019 All, this excludes userC (who likes pangolin & icecream) from the count.
@@ -6956,9 +6956,9 @@ toronto,3,0
 					Rows(places_visited, from='2019-01-01T00:00', to='2019-12-31T23:59'), 
 					filter=Not(Intersect(Row(likes='pangolin'), Row(likes='icecream')))
 				)`,
-			csvVerifier: `nairobi,1,0
-austin,1,0
-toronto,2,0
+			csvVerifier: `nairobi,1
+austin,1
+toronto,2
 `,
 		},
 		{ // After excluding UserC, this gets the sum of the networth of everyone per cities travelled

From acdff02fedb8917c33d9777dcbbe6833147d9a9d Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Fri, 22 Jan 2021 11:40:10 -0700
Subject: [PATCH 13/20] Add ingest time & latency stats to query benchmarks

---
 cmd/pilosa-bench/main.go               | 45 ++++++++++++++++++++++++++
 scripts/bench_read.sh                  | 11 +++++--
 scripts/etc/gloat/query.count.yml      |  1 +
 scripts/etc/gloat/query.difference.yml |  1 +
 scripts/etc/gloat/query.groupby.yml    |  1 +
 scripts/etc/gloat/query.intersect.yml  |  1 +
 scripts/etc/gloat/query.row-bsi.yml    |  1 +
 scripts/etc/gloat/query.row-range.yml  |  1 +
 scripts/etc/gloat/query.row.yml        |  1 +
 scripts/etc/gloat/query.topk.yml       |  1 +
 scripts/etc/gloat/query.union.yml      |  1 +
 scripts/etc/gloat/query.xor.yml        |  1 +
 12 files changed, 64 insertions(+), 2 deletions(-)

diff --git a/cmd/pilosa-bench/main.go b/cmd/pilosa-bench/main.go
index b33bf0fcf..1502dae23 100644
--- a/cmd/pilosa-bench/main.go
+++ b/cmd/pilosa-bench/main.go
@@ -16,12 +16,14 @@ package main
 
 import (
 	"context"
+	"expvar"
 	"flag"
 	"fmt"
 	"io/ioutil"
 	"log"
 	"math/rand"
 	"net/http"
+	_ "net/http/pprof"
 	"os"
 	"sort"
 	"strings"
@@ -32,6 +34,14 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
+var (
+	requestCountVar          = expvar.NewInt("request_count")
+	requestCurrentLatencyVar = expvar.NewFloat("request_current_latency") // seconds
+	requestAvgLatencyVar     = expvar.NewFloat("request_avg_latency")     // seconds
+	requestTotalLatencyVar   = expvar.NewFloat("request_total_latency")   // seconds
+	requestPerSecVar         = expvar.NewFloat("request_per_sec")
+)
+
 func main() {
 	if err := run(context.Background(), os.Args[1:]); err == flag.ErrHelp {
 		os.Exit(1)
@@ -86,6 +96,13 @@ func run(ctx context.Context, args []string) (err error) {
 		return err
 	}
 
+	// Set up HTTP endpoint to provide /debug endpoints.
+	fmt.Println("Serving debug endpoint at http://localhost:7070/debug")
+	go func() { _ = http.ListenAndServe(":7070", nil) }()
+
+	// Run separate goroutine to calculate the current req/sec & latency.
+	go monitor()
+
 	// Load all id/keys for each field.
 	log.Printf("loading field identifiers")
 	fieldIDMap, err := loadFields(ctx, client)
@@ -145,10 +162,15 @@ func run(ctx context.Context, args []string) (err error) {
 		log.Printf("[query] %s", q)
 
 		g.Go(func() error {
+			t := time.Now()
 			_, err = client.Query(ctx, key.index, &pilosa.QueryRequest{Index: key.index, Query: q})
 			if err != nil {
 				return err
 			}
+			elapsed := time.Since(t).Seconds()
+			requestCountVar.Add(1)
+			requestTotalLatencyVar.Add(elapsed)
+			requestAvgLatencyVar.Set(requestTotalLatencyVar.Value() / float64(requestCountVar.Value()))
 			return nil
 		})
 	}
@@ -156,6 +178,29 @@ func run(ctx context.Context, args []string) (err error) {
 	return g.Wait()
 }
 
+// monitor runs in a separate goroutine and updates metrics.
+func monitor() {
+	ticker := time.NewTicker(1 * time.Second)
+	defer ticker.Stop()
+
+	var lastTime time.Time
+	var lastN int64
+	var lastLatency float64
+	for range ticker.C {
+		now, n := time.Now(), requestCountVar.Value()
+		latency := requestTotalLatencyVar.Value()
+
+		if !lastTime.IsZero() {
+			elapsed := lastTime.Sub(now).Seconds()
+			if n > 0 {
+				requestCurrentLatencyVar.Set((lastLatency - latency) / float64(n))
+			}
+			requestPerSecVar.Set(float64(lastN-n) / elapsed)
+		}
+		lastTime, lastN, lastLatency = now, n, latency
+	}
+}
+
 func generateQuery(typ, index, field string, info *pilosa.FieldInfo, identifiers *pilosa.RowIdentifiers, opt queryOptions) (string, error) {
 	switch typ {
 	case "row":
diff --git a/scripts/bench_read.sh b/scripts/bench_read.sh
index bfbf6baac..a070bfb94 100755
--- a/scripts/bench_read.sh
+++ b/scripts/bench_read.sh
@@ -25,14 +25,21 @@ for TYPE in row row-bsi row-range count intersect union difference xor groupby t
 do
 	WORKFLOW_PATH="${BASH_SOURCE%/*}/etc/gloat/query.${TYPE}.yml"
 	WORKFLOW_NAME="$(gloat workflow name $WORKFLOW_PATH)"
-	TITLE="$WORKFLOW_NAME, $DATE ($SHA)"
-
+	
 	# Execute RBF/Roaring benchmark.
+	STARTTIME=$(date +%s)
 	RBF_PATH=gloat/data/query/${TYPE}/rbf/${DATE}.tar.gz
 	TXSRC=rbf gloat run -v -o "$RBF_PATH" $WORKFLOW_PATH
+	RBF_ELAPSED=$(($(date +%s) - $STARTTIME))
+	RBF_LATENCY=$(gloat metric -n -name request_avg_latency "$RBF_PATH")
 
+	STARTTIME=$(date +%s)
 	ROARING_PATH=gloat/data/query/${TYPE}/roaring/${DATE}.tar.gz
 	TXSRC=roaring gloat run -v -o "$ROARING_PATH" $WORKFLOW_PATH
+	ROARING_ELAPSED=$(($(date +%s) - $STARTTIME))
+	ROARING_LATENCY=$(gloat metric -n -name request_avg_latency "$ROARING_PATH")
+
+	TITLE="$WORKFLOW_NAME, $DATE ($SHA) elapsed rbf=$RBF_ELAPSEDroaring=$ROARING_ELAPSED> latency rbf=$RBF_LATENCY roaring=$ROARING_LATENCY"
 
 	# Generate graph from results.
 	gloat graph -layout 2,5 -size 5120,820 -title "$TITLE" -name utime,stime,heap_alloc,heap_inuse,heap_objects,num_gc,rchar,wchar,syscr,syscw -series rbf,roaring -o /tmp/output.png $RBF_PATH $ROARING_PATH
diff --git a/scripts/etc/gloat/query.count.yml b/scripts/etc/gloat/query.count.yml
index 5690c13a8..b0215d62c 100644
--- a/scripts/etc/gloat/query.count.yml
+++ b/scripts/etc/gloat/query.count.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars
diff --git a/scripts/etc/gloat/query.difference.yml b/scripts/etc/gloat/query.difference.yml
index 3a73a8e46..d3b17694f 100644
--- a/scripts/etc/gloat/query.difference.yml
+++ b/scripts/etc/gloat/query.difference.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars
diff --git a/scripts/etc/gloat/query.groupby.yml b/scripts/etc/gloat/query.groupby.yml
index a5c4c5d00..25b97fce4 100644
--- a/scripts/etc/gloat/query.groupby.yml
+++ b/scripts/etc/gloat/query.groupby.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars
diff --git a/scripts/etc/gloat/query.intersect.yml b/scripts/etc/gloat/query.intersect.yml
index 00bbfbd2e..4cb0993e9 100644
--- a/scripts/etc/gloat/query.intersect.yml
+++ b/scripts/etc/gloat/query.intersect.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars
diff --git a/scripts/etc/gloat/query.row-bsi.yml b/scripts/etc/gloat/query.row-bsi.yml
index 04fe472eb..7c86b5a41 100644
--- a/scripts/etc/gloat/query.row-bsi.yml
+++ b/scripts/etc/gloat/query.row-bsi.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars
diff --git a/scripts/etc/gloat/query.row-range.yml b/scripts/etc/gloat/query.row-range.yml
index f05b5cb4c..75ab2c6d6 100644
--- a/scripts/etc/gloat/query.row-range.yml
+++ b/scripts/etc/gloat/query.row-range.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars
diff --git a/scripts/etc/gloat/query.row.yml b/scripts/etc/gloat/query.row.yml
index 142f12004..7f6997ea3 100644
--- a/scripts/etc/gloat/query.row.yml
+++ b/scripts/etc/gloat/query.row.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars
diff --git a/scripts/etc/gloat/query.topk.yml b/scripts/etc/gloat/query.topk.yml
index 10da63427..25ef7fe50 100644
--- a/scripts/etc/gloat/query.topk.yml
+++ b/scripts/etc/gloat/query.topk.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars
diff --git a/scripts/etc/gloat/query.union.yml b/scripts/etc/gloat/query.union.yml
index a2270d974..1b2777b90 100644
--- a/scripts/etc/gloat/query.union.yml
+++ b/scripts/etc/gloat/query.union.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars
diff --git a/scripts/etc/gloat/query.xor.yml b/scripts/etc/gloat/query.xor.yml
index f2eb857b8..3328fb0b1 100644
--- a/scripts/etc/gloat/query.xor.yml
+++ b/scripts/etc/gloat/query.xor.yml
@@ -8,3 +8,4 @@ health_regexp: "NORMAL"
 
 vars_urls:
   - http://localhost:10101/debug/vars
+  - http://localhost:7070/debug/vars

From 34f728521ca992ca53c1d692fd5bbc3ea24b54ad Mon Sep 17 00:00:00 2001
From: Cody Soyland <cody@soyland.com>
Date: Fri, 22 Jan 2021 14:59:28 -0600
Subject: [PATCH 14/20] Add git-submodule to manage lattice version

---
 .gitignore  |  1 -
 .gitmodules |  3 +++
 Makefile    | 10 ++++++++--
 lattice     |  1 +
 4 files changed, 12 insertions(+), 3 deletions(-)
 create mode 100644 .gitmodules
 create mode 160000 lattice

diff --git a/.gitignore b/.gitignore
index 9bf572296..82a553aed 100644
--- a/.gitignore
+++ b/.gitignore
@@ -5,7 +5,6 @@ vendor
 .DS_Store
 build
 *~
-lattice
 release-pilosa-fsck.*.*.tar.gz
 /log.*
 /tourna.log.*
diff --git a/.gitmodules b/.gitmodules
new file mode 100644
index 000000000..afb2db2d6
--- /dev/null
+++ b/.gitmodules
@@ -0,0 +1,3 @@
+[submodule "lattice"]
+	path = lattice
+	url = git@github.com:molecula/lattice.git
diff --git a/Makefile b/Makefile
index 9114fda54..a1b80b5b8 100644
--- a/Makefile
+++ b/Makefile
@@ -155,11 +155,17 @@ install:
 install-bench:
 	go install -tags='$(BUILD_TAGS)' -ldflags $(LDFLAGS) $(FLAGS) ./cmd/pilosa-bench
 
+# Ensure lattice is cloned and the pinned version is checked out
 lattice:
-	git clone git@github.com:molecula/lattice.git
+	git submodule update --init
 
+# Build the lattice assets
 build-lattice: lattice require-yarn
-	cd lattice && git pull && yarn install && yarn build
+	cd lattice && yarn install && yarn build
+
+# Upgrade lattice to the latest version
+upgrade-lattice: lattice
+	git submodule update --remote
 
 # `go generate` protocol buffers
 generate-protoc: require-protoc require-protoc-gen-gofast
diff --git a/lattice b/lattice
new file mode 160000
index 000000000..36f453c1e
--- /dev/null
+++ b/lattice
@@ -0,0 +1 @@
+Subproject commit 36f453c1ea3bf86c546a8ad4a88f2a926724d683

From 671c7262c9f756785f8adc1e67c07c3d715bd23b Mon Sep 17 00:00:00 2001
From: Cody Soyland <cody@soyland.com>
Date: Fri, 22 Jan 2021 16:26:09 -0600
Subject: [PATCH 15/20] Add lattice to PHONY

---
 Makefile | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/Makefile b/Makefile
index a1b80b5b8..2e6cc7331 100644
--- a/Makefile
+++ b/Makefile
@@ -1,4 +1,4 @@
-.PHONY: build check-clean clean build-lattice cover cover-viz default docker docker-build docker-test docker-tag-push generate generate-protoc generate-pql generate-statik gometalinter install install-build-deps install-golangci-lint install-gometalinter install-protoc install-protoc-gen-gofast install-peg install-statik prerelease prerelease-upload release release-build test testv testv-race testvsub testvsub-race  test-txstore-rbf
+.PHONY: build check-clean clean build-lattice cover cover-viz default docker docker-build docker-test docker-tag-push generate generate-protoc generate-pql generate-statik gometalinter install install-build-deps install-golangci-lint install-gometalinter install-protoc install-protoc-gen-gofast install-peg install-statik prerelease prerelease-upload release release-build test testv testv-race testvsub testvsub-race  test-txstore-rbf lattice
 
 CLONE_URL=github.com/pilosa/pilosa
 MOD_VERSION=v2

From 852057533be7109e1fd51c85777fecfb11fbdc99 Mon Sep 17 00:00:00 2001
From: Alan Bernstein <alan.aaron.bernstein@gmail.com>
Date: Fri, 22 Jan 2021 16:51:17 -0600
Subject: [PATCH 16/20] Add lattice submodule upgrade instructions

---
 README.md | 9 +++++++++
 1 file changed, 9 insertions(+)

diff --git a/README.md b/README.md
index 4543a3524..f8ff52c19 100644
--- a/README.md
+++ b/README.md
@@ -58,6 +58,15 @@ Check out how the Pilosa [Data Model](https://www.pilosa.com/docs/data-model/) w
 You can interact with Pilosa directly in the console using the [Pilosa Query Language](https://www.pilosa.com/docs/query-language/) (PQL).
 
 
+## Upgrading UI
+
+Lattice is now a submodule of Pilosa, for the purpose of keeping the frontend and backend components of the UI system synchronized. When making a change to the UI that requires changing both the backend and frontend, follow steps in this order:
+
+1. merge frontend (Lattice) PR
+2. run `make upgrade-lattice`
+3. run `git add lattice`, `git commit -m"Upgrade Lattice"`, `git push`
+4. merge backend (Pilosa) PR
+
 ## Client Libraries
 
 There are supported libraries for the following languages:

From 649202b0816bcebd05d17f43ec435bd69e7a7ac9 Mon Sep 17 00:00:00 2001
From: Cody Soyland <cody@soyland.com>
Date: Mon, 21 Dec 2020 13:25:15 -0600
Subject: [PATCH 17/20] Fix cluster size setter: expose failures

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

diff --git a/executor_test.go b/executor_test.go
index c0532b4f9..6a202ca3a 100644
--- a/executor_test.go
+++ b/executor_test.go
@@ -5517,7 +5517,7 @@ func TestExecutor_Execute_DistinctFailure(t *testing.T) {
 
 func TestExecutor_Execute_GroupBy(t *testing.T) {
 	groupByTest := func(t *testing.T, clusterSize int) {
-		c := test.MustRunCluster(t, 1)
+		c := test.MustRunCluster(t, clusterSize)
 		defer c.Close()
 		c.CreateField(t, "i", pilosa.IndexOptions{}, "general")
 		c.CreateField(t, "i", pilosa.IndexOptions{}, "sub")
@@ -5924,7 +5924,7 @@ func TestExecutor_Execute_GroupBy(t *testing.T) {
 		})
 
 	}
-	for size := range []int{1, 3} {
+	for _, size := range []int{1, 3} {
 		t.Run(fmt.Sprintf("%d_nodes", size), func(t *testing.T) {
 			groupByTest(t, size)
 		})

From dde318ac8cb929a2c2fb7b12c431d99e8fd648b6 Mon Sep 17 00:00:00 2001
From: Nia Weiss <nia.weiss@molecula.com>
Date: Mon, 25 Jan 2021 10:15:45 -0500
Subject: [PATCH 18/20] Move globally computed GroupBy rows calls into
 EmbeddedData

This fixes a bug where a globally computed Rows call would be computed with a subset of the shards.
---
 executor.go | 17 +++++++++++++++++
 pql/ast.go  |  1 +
 row.go      |  4 ++++
 3 files changed, 22 insertions(+)

diff --git a/executor.go b/executor.go
index 0c02fd6c8..b1a0ae1eb 100644
--- a/executor.go
+++ b/executor.go
@@ -2797,6 +2797,12 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
 		}
 
 		if hasLimit || hasCol { // we need to perform this query cluster-wide ahead of executeGroupByShard
+			if idx, ok := child.Args["valueidx"].(int64); ok {
+				// The rows query was already completed on the initiating node.
+				childRows[i] = opt.EmbeddedData[idx].Columns()
+				continue
+			}
+
 			childRows[i], err = e.executeRows(ctx, qcx, index, child, shards, opt)
 			if err != nil {
 				return nil, errors.Wrap(err, "getting rows for ")
@@ -2804,6 +2810,13 @@ func (e *executor) executeGroupBy(ctx context.Context, qcx *Qcx, index string, c
 			if len(childRows[i]) == 0 { // there are no results because this field has no values.
 				return &GroupCounts{}, nil
 			}
+
+			// Stuff the result into opt.EmbeddedData so that it gets sent to other nodes in the map-reduce.
+			// This is flagged as "NoSplit" to ensure that the entire row gets sent out.
+			rowsRow := NewRow(childRows[i]...)
+			rowsRow.NoSplit = true
+			child.Args["valueidx"] = int64(len(opt.EmbeddedData))
+			opt.EmbeddedData = append(opt.EmbeddedData, rowsRow)
 		}
 	}
 
@@ -5527,6 +5540,10 @@ func makeEmbeddedDataForShards(allRows []*Row, shards []uint64) []*Row {
 		if row == nil || len(row.segments) == 0 {
 			continue
 		}
+		if row.NoSplit {
+			newRows[i] = row
+			continue
+		}
 		segments := row.segments
 		segmentIndex := 0
 		newRows[i] = &Row{
diff --git a/pql/ast.go b/pql/ast.go
index 40e2acaba..54ae2dee7 100644
--- a/pql/ast.go
+++ b/pql/ast.go
@@ -388,6 +388,7 @@ var callInfoByFunc = map[string]callInfo{
 			"from":     nil,
 			"to":       nil,
 			"like":     "",
+			"valueidx": int64(0),
 		},
 	},
 	"Shift": {allowUnknown: false,
diff --git a/row.go b/row.go
index 46a0ec478..3ff47cba5 100644
--- a/row.go
+++ b/row.go
@@ -42,6 +42,10 @@ type Row struct {
 	// query. Knowing the index and field, we can figure out how to
 	// interpret the row data.
 	Field string
+
+	// NoSplit indicates that this row may not be split.
+	// This is used for `Rows` calls in a GroupBy.
+	NoSplit bool
 }
 
 // NewRow returns a new instance of Row.

From d1a9c91a96df1a208e8a7f1450743e73ebc75a81 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Mon, 25 Jan 2021 14:41:05 -0600
Subject: [PATCH 19/20] Test CountRange on non-container ranges.

This test was supposed to check against all the container types,
but especially bitmaps, but turns out not to work because the
containers turn into RLE containers. Oops. Now, we start with
every-other-bit for the first 8k, then start filling in the holes,
so we get some bitmap containers and then start generating
RLE containers.
---
 tx_internal_test.go | 27 ++++++++++++++++++++-------
 1 file changed, 20 insertions(+), 7 deletions(-)

diff --git a/tx_internal_test.go b/tx_internal_test.go
index a5da9cccd..3ff4ddd63 100644
--- a/tx_internal_test.go
+++ b/tx_internal_test.go
@@ -37,20 +37,29 @@ func requireCountRangeSampleData(tb testing.TB) (*fragment, Tx) {
 		// request that each container get its own copy of the bitmap.
 		var bitmapSample [1025]uint64
 		for i := range arraySample {
-			arraySample[i] = uint16(i)
+			arraySample[i] = uint16(i * 2)
 		}
-		for i := 0; i < 4096/64; i++ {
-			bitmapSample[i] = ^uint64(0)
+		// Put corresponding bits in the bitmap...
+		for i := 0; i < 4096/32; i++ {
+			// bit 0 is 0x1, bit 2 is 0x4, so even-numbered bits
+			// are 0x5555....
+			bitmapSample[i] = 0x5555555555555555
 		}
 		bm := roaring.NewSliceBitmap()
 		for n := 0; n < 4096 && n < countRangeMaxN; n++ {
 			c := roaring.NewContainerArray(arraySample[:n])
 			bm.Put(uint64(n), c)
 		}
-		for n := 4096; n < countRangeMaxN; n++ {
+		// Start filling in the missing bits. This starts us out with
+		// bitmap containers, but then eventually converts to things
+		// that are more likely to be run containers. At the end of this,
+		// we should have exactly the first 8,192 bits set, for a single
+		// run of 8k.
+		for n := 4096; n < 8192; n++ {
 			c := roaring.NewContainerBitmapN(bitmapSample[:], int32(n))
 			bm.Put(uint64(n), c)
-			bitmapSample[n/64] |= 1 << (n % 64)
+			w := n - 4096
+			bitmapSample[w/32] |= 1 << (((n % 32) * 2) + 1)
 		}
 		var asBytes bytes.Buffer
 		n, err := bm.WriteTo(&asBytes)
@@ -90,11 +99,14 @@ func TestTx_CountRange(t *testing.T) {
 	expected := uint64(0)
 	j := uint64(0)
 	for i := uint64(0); i < countRangeMaxN; i += 7 {
+		expected += i
 		if i%4 == 3 {
 			expected -= (j * 7) + 21
 			j += 7
 		}
-		got, err := tx.CountRange("i", "f", viewStandard, 0, uint64(j)<<16, uint64(i)<<16)
+		// Every other bit gets set, for a total of i bits in container
+		// i, so they're all in the first (i*2) bits of the container.
+		got, err := tx.CountRange("i", "f", viewStandard, 0, uint64(j)<<16, (uint64(i)<<16)+(i*2))
 		if err != nil {
 			t.Fatalf("counting range: %v", err)
 		}
@@ -102,7 +114,8 @@ func TestTx_CountRange(t *testing.T) {
 			t.Fatalf("counting from container %d to %d, expected %d, got %d",
 				j, i, expected, got)
 		}
-		expected += (i * 7) + 21
+		// The -i here undoes the +i at the top of this loop.
+		expected += (i * 7) + 21 - i
 	}
 }
 

From a238afb21a75a0ed584d8c82c3164cba034d0792 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Mon, 25 Jan 2021 14:50:37 -0600
Subject: [PATCH 20/20] Handle BitmapPtr cells in countRange

We need to be able to count bits in BitmapPtr containers. This only
comes up if you have a non-container-aligned range count, which we
never do in real production yet, but the API allows it so it should
work. In order to do this, we need to provide the tx to countRange
so it can grab pages as needed. Arguably, we should be able to avoid
actually creating/copying that page since we're only using it
internally, never returning it, but this is a pretty rare case
and probably not performance-critical.
---
 rbf/rbf.go | 6 +++++-
 rbf/tx.go  | 7 +++----
 2 files changed, 8 insertions(+), 5 deletions(-)

diff --git a/rbf/rbf.go b/rbf/rbf.go
index f13a5a7fc..365907121 100644
--- a/rbf/rbf.go
+++ b/rbf/rbf.go
@@ -462,7 +462,7 @@ func (c *leafCell) lastValue(tx *Tx) uint16 {
 // We have to take int32 rather than uint16 because the interval is [start, end),
 // and otherwise we have no way to ask to count the entire container (the
 // high bit will be missed).
-func (c *leafCell) countRange(start, end int32) (n int) {
+func (c *leafCell) countRange(tx *Tx, start, end int32) (n int) {
 	// If the full range is being queried, simply use the precalculated count.
 	if start == 0 && end > math.MaxUint16 {
 		return c.BitN
@@ -475,6 +475,10 @@ func (c *leafCell) countRange(start, end int32) (n int) {
 		return int(roaring.RunCountRange(toInterval16(c.Data), start, end))
 	case ContainerTypeBitmap:
 		return int(roaring.BitmapCountRange(toArray64(c.Data), start, end))
+	case ContainerTypeBitmapPtr:
+		_, a, err := tx.leafCellBitmap(toPgno(c.Data))
+		panicOn(err)
+		return int(roaring.BitmapCountRange(a, start, end))
 	default:
 		panic(fmt.Sprintf("invalid container type: %d", c.Type))
 	}
diff --git a/rbf/tx.go b/rbf/tx.go
index 705edfc5b..a952104ec 100644
--- a/rbf/tx.go
+++ b/rbf/tx.go
@@ -1318,7 +1318,6 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) {
 	} else if err != nil {
 		return 0, err
 	}
-
 	var n uint64
 	for {
 		if err := csr.Next(); err == io.EOF {
@@ -1341,7 +1340,7 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) {
 
 		// If range is entirely in one container then just count that range.
 		if skey == ekey {
-			return uint64(c.countRange(int32(lowbits(start)), ebits)), nil
+			return uint64(c.countRange(tx, int32(lowbits(start)), ebits)), nil
 		}
 		// INVAR: skey < ekey
 
@@ -1351,7 +1350,7 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) {
 			break
 		}
 		if k == skey {
-			n += uint64(c.countRange(int32(lowbits(start)), roaring.MaxContainerVal+1))
+			n += uint64(c.countRange(tx, int32(lowbits(start)), roaring.MaxContainerVal+1))
 			continue
 		}
 		if k < ekey {
@@ -1359,7 +1358,7 @@ func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) {
 			continue
 		}
 		if k == ekey && ebits > 0 {
-			n += uint64(c.countRange(0, ebits))
+			n += uint64(c.countRange(tx, 0, ebits))
 			break
 		}
 	}