Merge branch 'master' into mmap

.gitignore

@@ -5,7 +5,6 @@ vendor
 .DS_Store
 build
 *~
-lattice
 release-pilosa-fsck.*.*.tar.gz
 /log.*
 /tourna.log.*

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "lattice"]
+	path = lattice
+	url = git@github.com:molecula/lattice.git

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
@@ -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

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:

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":

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{
@@ -7381,9 +7398,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 {
@@ -7397,9 +7416,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
 
@@ -7408,11 +7460,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]))
@@ -7424,9 +7471,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")

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)
 		})
@@ -6863,6 +6863,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: toronto only
+		{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{
@@ -6900,6 +6924,66 @@ 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
+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
+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
+paris,1
+austin,1
+toronto,3
+`,
+		},
+		{ // 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
+austin,1
+toronto,2
+`,
+		},
+		{ // 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
+			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) {

fragment.go

@@ -3329,6 +3329,95 @@ func (f *fragment) rowIterator(tx Tx, wrap bool, filters ...roaring.BitmapFilter
 	return f.setRowIterator(tx, wrap, filters...)
 }
 
+type timeRowIterator struct {
+	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) {
+	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,
+		cur:  0,
+		wrap: wrap,
+	}
+
+	// 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
+		}
+		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++
+	}
+	sort.Slice(allRowIDs, func(i, j int) bool { return allRowIDs[i] < allRowIDs[j] })
+
+	it.rowIDToFragments = rowIDToFragments
+	it.allRowIDs = allRowIDs
+
+	return it, nil
+}
+
+func (it *timeRowIterator) Seek(rowID uint64) {
+	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) {
+	if it.cur >= len(it.allRowIDs) {
+		if !it.wrap || len(it.allRowIDs) == 0 {
+			return nil, 0, nil, true, nil
+		}
+		it.Seek(0)
+		wrapped = true
+	}
+
+	// gather rows
+	rowID = it.allRowIDs[it.cur]
+	fragments := it.rowIDToFragments[rowID]
+	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 = append(rows, row)
+	}
+
+	// union rows
+	r = rows[0].Union(rows[1:]...)
+
+	it.cur++
+	return r, rowID, nil, wrapped, nil
+}
+
 type intRowIterator struct {
 	f      *fragment
 	values int64Slice         // sorted slice of int values

lattice

@@ -0,0 +1 @@
+Subproject commit 36f453c1ea3bf86c546a8ad4a88f2a926724d683

pql/ast.go

@@ -388,6 +388,7 @@ var callInfoByFunc = map[string]callInfo{
 			"from":     nil,
 			"to":       nil,
 			"like":     "",
+			"valueidx": int64(0),
 		},
 	},
 	"Shift": {allowUnknown: false,

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

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

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

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

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.

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

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

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

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

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

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

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

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

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

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

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

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

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