From 4d1e9ed78ae4f8f71313669e70169e335cdf6788 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Mon, 10 Jun 2019 13:15:56 -0500
Subject: [PATCH 01/13] reshuffle benchmarks and include cache type in testing

It turns out there's some significant potential improvements to
be had in the case where there's no cache being used on a field, so
we add it to the benchmarks, to allow testing that.

We also make sure that `getUpdataInto` picks the requested number
of columns; if N was a point at which something weird happens,
we might only sometimes see it.
---
 fragment_internal_test.go | 140 +++++++++++++++++++++-----------------
 1 file changed, 77 insertions(+), 63 deletions(-)

diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index e45d3cce8..c91b4de18 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -2037,16 +2037,17 @@ func BenchmarkFragment_Import(b *testing.B) {
 }
 
 var (
-	rowCases         = []uint64{2, 50, 1000, 100000}
-	colCases         = []uint64{20, 1000, 50000, 500000}
-	concurrencyCases = []int{2, 16}
+	rowCases         = []uint64{2, 50, 1000, 10000, 100000}
+	colCases         = []uint64{20, 1000, 5000, 50000, 500000}
+	concurrencyCases = []int{2, 4, 16}
+	cacheCases       = []string{CacheTypeNone, CacheTypeRanked}
 )
 
 func BenchmarkImportRoaring(b *testing.B) {
 	for _, numRows := range rowCases {
 		data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
 		b.Logf("%dRows: %.2fMB\n", numRows, float64(len(data))/1024/1024)
-		for _, cacheType := range []string{CacheTypeRanked} { // CacheTypeNone didn't seem to affect the results much
+		for _, cacheType := range cacheCases {
 			b.Run(fmt.Sprintf("Rows%dCache_%s", numRows, cacheType), func(b *testing.B) {
 				b.StopTimer()
 				for i := 0; i < b.N; i++ {
@@ -2070,63 +2071,28 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) {
 		b.SkipNow()
 	}
 	for _, numRows := range rowCases {
-		data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
-		b.Logf("%dRows: %.2fMB\n", numRows, float64(len(data))/1024/1024)
+		data := make([][]byte, 0, len(concurrencyCases))
+		data = append(data, getZipfRowsSliceRoaring(numRows, 0, 0, ShardWidth))
+		b.Logf("%dRows: %.2fMB\n", numRows, float64(len(data[0]))/1024/1024)
 		for _, concurrency := range concurrencyCases {
-			b.Run(fmt.Sprintf("%dRows%dConcurrency", numRows, concurrency), func(b *testing.B) {
-				b.StopTimer()
-				frags := make([]*fragment, concurrency)
-				for i := 0; i < b.N; i++ {
-					for j := 0; j < concurrency; j++ {
-						frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), CacheTypeRanked)
-					}
-					eg := errgroup.Group{}
-					b.StartTimer()
-					for j := 0; j < concurrency; j++ {
-						j := j
-						eg.Go(func() error {
-							return frags[j].importRoaringT(data, false)
-						})
-					}
-					err := eg.Wait()
-					if err != nil {
-						b.Errorf("importing fragment: %v", err)
-					}
-					b.StopTimer()
-					for j := 0; j < concurrency; j++ {
-						frags[j].Clean(b)
-					}
-				}
-			})
-		}
-	}
-}
-func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
-	if testing.Short() {
-		b.SkipNow()
-	}
-	for _, numRows := range rowCases {
-		for _, numCols := range colCases {
-			data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
-			updata := getUpdataRoaring(numRows, numCols, 1)
-			for _, concurrency := range concurrencyCases {
-				b.Run(fmt.Sprintf("%dRows%dCols%dConcurrency", numRows, numCols, concurrency), func(b *testing.B) {
+			// add more data sets
+			for j := len(data); j < concurrency; j++ {
+				data = append(data, getZipfRowsSliceRoaring(numRows, int64(j), 0, ShardWidth))
+			}
+			for _, cacheType := range cacheCases {
+				b.Run(fmt.Sprintf("Rows%dConcurrency%dCache_%s", numRows, concurrency, cacheType), func(b *testing.B) {
 					b.StopTimer()
 					frags := make([]*fragment, concurrency)
 					for i := 0; i < b.N; i++ {
 						for j := 0; j < concurrency; j++ {
-							frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), CacheTypeRanked)
-							err := frags[j].importRoaringT(data, false)
-							if err != nil {
-								b.Fatalf("importing roaring: %v", err)
-							}
+							frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), cacheType)
 						}
 						eg := errgroup.Group{}
 						b.StartTimer()
 						for j := 0; j < concurrency; j++ {
 							j := j
 							eg.Go(func() error {
-								return frags[j].importRoaringT(updata, false)
+								return frags[j].importRoaringT(data[j], false)
 							})
 						}
 						err := eg.Wait()
@@ -2143,9 +2109,53 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
 		}
 	}
 }
+func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
+	if testing.Short() {
+		b.SkipNow()
+	}
+	for _, numRows := range rowCases {
+		for _, numCols := range colCases {
+			data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
+			updata := getUpdataRoaring(numRows, numCols, 1)
+			for _, concurrency := range concurrencyCases {
+				for _, cacheType := range cacheCases {
+					b.Run(fmt.Sprintf("Rows%dCols%dConcurrency%dCache_%s", numRows, numCols, concurrency, cacheType), func(b *testing.B) {
+						b.StopTimer()
+						frags := make([]*fragment, concurrency)
+						for i := 0; i < b.N; i++ {
+							for j := 0; j < concurrency; j++ {
+								frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), cacheType)
+								err := frags[j].importRoaringT(data, false)
+								if err != nil {
+									b.Fatalf("importing roaring: %v", err)
+								}
+							}
+							eg := errgroup.Group{}
+							b.StartTimer()
+							for j := 0; j < concurrency; j++ {
+								j := j
+								eg.Go(func() error {
+									return frags[j].importRoaringT(updata, false)
+								})
+							}
+							err := eg.Wait()
+							if err != nil {
+								b.Errorf("importing fragment: %v", err)
+							}
+							b.StopTimer()
+							for j := 0; j < concurrency; j++ {
+								frags[j].Clean(b)
+							}
+						}
+					})
+				}
+			}
+		}
+	}
+}
 
 func BenchmarkImportStandard(b *testing.B) {
-	for _, cacheType := range []string{CacheTypeRanked} {
+	for _, cacheType := range cacheCases {
 		for _, numRows := range rowCases {
 			rowIDsOrig, columnIDsOrig := getZipfRowsSliceStandard(numRows, 1, 0, ShardWidth)
 			rowIDs, columnIDs := make([]uint64, len(rowIDsOrig)), make([]uint64, len(columnIDsOrig))
@@ -2171,17 +2181,17 @@ func BenchmarkImportStandard(b *testing.B) {
 func BenchmarkImportRoaringUpdate(b *testing.B) {
 	fileSize := make(map[string]int64)
 	names := []string{}
-	for _, cacheType := range []string{CacheTypeRanked} {
-		for _, numRows := range rowCases {
-			for _, numCols := range colCases {
-				data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
-				updata := getUpdataRoaring(numRows, numCols, 1)
-				name := fmt.Sprintf("%s%dRows%dCols", cacheType, numRows, numCols)
+	for _, numRows := range rowCases {
+		data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
+		for _, numCols := range colCases {
+			updata := getUpdataRoaring(numRows, numCols, 1)
+			for _, cacheType := range cacheCases {
+				name := fmt.Sprintf("Rows%dCols%dCache_%s", numRows, numCols, cacheType)
 				names = append(names, name)
 				b.Run(name, func(b *testing.B) {
 					b.StopTimer()
 					for i := 0; i < b.N; i++ {
-						f := mustOpenFragment("i", fmt.Sprintf("r%dc%s", numRows, cacheType), viewStandard, 0, cacheType)
+						f := mustOpenFragment("i", fmt.Sprintf("r%dc%dcache_%s", numRows, numCols, cacheType), viewStandard, 0, cacheType)
 						err := f.importRoaringT(data, false)
 						if err != nil {
 							b.Errorf("import error: %v", err)
@@ -2400,19 +2410,23 @@ func getUpdataRoaring(numRows, numCols uint64, seed int64) []byte {
 	return buf.Bytes()
 }
 
-func getUpdataInto(f func(row, col uint64) bool, numRows, numCols uint64, seed int64) (changed int) {
+func getUpdataInto(f func(row, col uint64) bool, numRows, numCols uint64, seed int64) int {
 	s := rand.NewSource(seed)
 	r := rand.New(s)
 	z := rand.NewZipf(r, 1.6, 50, numRows-1)
 
-	for i := uint64(0); i < numCols; i++ {
-		col := uint64(r.Int63n(ShardWidth)) // assuming the number of repeats will be negligible
+	i := uint64(0)
+	// ensure we get exactly the number we asked for. it turns out we had
+	// a horrible pathological edge case for exactly 10,000 entries in an
+	// imported bitmap, and hit it only occasionally...
+	for i < numCols {
+		col := uint64(r.Int63n(ShardWidth))
 		row := z.Uint64()
 		if f(row, col) {
-			changed++
+			i++
 		}
 	}
-	return changed
+	return int(i)
 }
 
 // getZipfRowsSliceStandard is the same as getZipfRowsSliceRoaring, but returns

From c19b7af0d0e7556200e2167dd9980b863bdfbbcb Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Tue, 25 Jun 2019 14:26:20 -0500
Subject: [PATCH 02/13] Support direct roaring import operations

We add a new ops log type(pair), AddRoaring and RemoveRoaring,
which set and clear the bits from a provided roaring bitmap.

This also compels us to consider additional sanity checking
during tests.
---
 fragment_internal_test.go   |  19 ++
 roaring/btree.go            |   6 +-
 roaring/containers_btree.go |  11 +-
 roaring/containers_slice.go |  20 +-
 roaring/roaring.go          | 402 ++++++++++++++++++++++++++++++++++--
 5 files changed, 434 insertions(+), 24 deletions(-)

diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index c91b4de18..dfa470144 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -2476,7 +2476,25 @@ func BenchmarkFileWrite(b *testing.B) {
 
 /////////////////////////////////////////////////////////////////////
 
+func (f *fragment) sanityCheck(t testing.TB) {
+	newBM := roaring.NewFileBitmap()
+	file, err := os.Open(f.path)
+	if err != nil {
+		t.Fatalf("sanityCheck couldn't open file %s: %v", f.path, err)
+	}
+	defer file.Close()
+	data, err := ioutil.ReadAll(file)
+	err = newBM.UnmarshalBinary(data)
+	if err != nil {
+		t.Fatalf("sanityCheck couldn't read fragment %s: %v", f.path, err)
+	}
+	if equal, reason := newBM.BitwiseEqual(f.storage); !equal {
+		t.Fatalf("fragment %s: unmarshalled bitmap different: %v", f.path, reason)
+	}
+}
+
 func (f *fragment) Clean(t testing.TB) {
+	f.sanityCheck(t)
 	errc := f.Close()
 	errf := os.Remove(f.path)
 	errp := os.Remove(f.cachePath())
@@ -3021,6 +3039,7 @@ func TestUnionInPlaceMapped(t *testing.T) {
 
 	f.storage.UnionInPlace(setBM1)
 	countUnion := f.storage.Count()
+	f.snapshot()
 
 	if count0 != countF {
 		t.Fatalf("writing bitmap to storage changed count: %d => %d", count0, countF)
diff --git a/roaring/btree.go b/roaring/btree.go
index 53489af89..2a01159df 100644
--- a/roaring/btree.go
+++ b/roaring/btree.go
@@ -894,7 +894,7 @@ func (e *enumerator) Next() (k uint64, v *Container, err error) {
 }
 
 // Every iterates over a tree.
-func (e *enumerator) Every(upd func(oldV *Container, exists bool) (newV *Container, write bool)) error {
+func (e *enumerator) Every(upd func(key uint64, oldV *Container, exists bool) (newV *Container, write bool)) error {
 	if err := e.err; err != nil {
 		return err
 	}
@@ -919,10 +919,10 @@ func (e *enumerator) Every(upd func(oldV *Container, exists bool) (newV *Contain
 		}
 
 		i := e.q.d[e.i]
-		nv, write := upd(i.v, true)
+		nv, write := upd(i.k, i.v, true)
 		if write {
 			if nv == nil {
-				e.t.Delete(e.q.d[e.i].k)
+				e.t.Delete(i.k)
 			} else {
 				e.q.d[e.i].v = nv
 			}
diff --git a/roaring/containers_btree.go b/roaring/containers_btree.go
index c2dbc01f5..d00bf3247 100644
--- a/roaring/containers_btree.go
+++ b/roaring/containers_btree.go
@@ -99,6 +99,10 @@ func (btc *bTreeContainers) PutContainerValues(key uint64, typ byte, n int, mapp
 
 func (btc *bTreeContainers) Remove(key uint64) {
 	btc.tree.Delete(key)
+	if key == btc.lastKey {
+		btc.lastKey = ^uint64(0)
+		btc.lastContainer = nil
+	}
 }
 
 func (btc *bTreeContainers) GetOrCreate(key uint64) *Container {
@@ -185,6 +189,11 @@ func (btc *bTreeContainers) Reset() {
 	btc.lastContainer = nil
 }
 
+func (btc *bTreeContainers) ResetN(n int) {
+	// we ignore n because it's impractical to preallocate the tree
+	btc.Reset()
+}
+
 func (btc *bTreeContainers) Iterator(key uint64) (citer ContainerIterator, found bool) {
 	e, ok := btc.tree.Seek(key)
 	if ok {
@@ -215,7 +224,7 @@ func (btc *bTreeContainers) Update(key uint64, fn func(*Container, bool) (*Conta
 // UpdateEvery calls fn (existing-container, existed), and expects
 // (new-container, write). If write is true, the container is used to
 // replace the given container.
-func (btc *bTreeContainers) UpdateEvery(fn func(*Container, bool) (*Container, bool)) {
+func (btc *bTreeContainers) UpdateEvery(fn func(uint64, *Container, bool) (*Container, bool)) {
 	e, _ := btc.tree.Seek(0)
 	// currently not handling the error from this, but in practice it has
 	// to be io.EOF.
diff --git a/roaring/containers_slice.go b/roaring/containers_slice.go
index df44b4ff1..67e936c24 100644
--- a/roaring/containers_slice.go
+++ b/roaring/containers_slice.go
@@ -75,6 +75,10 @@ func (sc *sliceContainers) Remove(key uint64) {
 	if i < 0 {
 		return
 	}
+	if key == sc.lastKey {
+		sc.lastKey = ^uint64(0)
+		sc.lastContainer = nil
+	}
 	sc.keys = append(sc.keys[:i], sc.keys[i+1:]...)
 	sc.containers = append(sc.containers[:i], sc.containers[i+1:]...)
 
@@ -158,6 +162,18 @@ func (sc *sliceContainers) Reset() {
 	sc.lastKey = 0
 }
 
+func (sc *sliceContainers) ResetN(n int) {
+	if cap(sc.keys) < n {
+		sc.keys = make([]uint64, 0, n)
+		sc.containers = make([]*Container, 0, n)
+	} else {
+		sc.keys = sc.keys[:0]
+		sc.containers = sc.containers[:0]
+	}
+	sc.lastContainer = nil
+	sc.lastKey = 0
+}
+
 func (sc *sliceContainers) seek(key uint64) (int, bool) {
 	i := search64(sc.keys, key)
 	found := true
@@ -204,9 +220,9 @@ func (sc *sliceContainers) Update(key uint64, fn func(*Container, bool) (*Contai
 // UpdateEvery calls fn (existing-container, existed), and expects
 // (new-container, write). If write is true, the container is used to
 // replace the given container.
-func (sc *sliceContainers) UpdateEvery(fn func(*Container, bool) (*Container, bool)) {
+func (sc *sliceContainers) UpdateEvery(fn func(uint64, *Container, bool) (*Container, bool)) {
 	for i, c := range sc.containers {
-		nc, write := fn(c, true)
+		nc, write := fn(sc.keys[i], c, true)
 		if write {
 			sc.containers[i] = nc
 		}
diff --git a/roaring/roaring.go b/roaring/roaring.go
index f55cbc4a6..28892af6a 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -117,7 +117,7 @@ type Containers interface {
 	// UpdateEvery calls fn (existing-container, existed), and expects
 	// (new-container, write). If write is true, the container is used to
 	// replace the given container.
-	UpdateEvery(fn func(*Container, bool) (*Container, bool))
+	UpdateEvery(fn func(uint64, *Container, bool) (*Container, bool))
 
 	// Iterator returns a Contiterator which after a call to Next(), a call to Value() will
 	// return the first container at or after key. found will be true if a
@@ -128,6 +128,8 @@ type Containers interface {
 
 	// Reset clears the containers collection to allow for recycling during snapshot
 	Reset()
+	// ResetN clears the collection but hints at a needed size.
+	ResetN(int)
 
 	// Repair will repair the cardinality of any containers whose cardinality were corrupted
 	// due to optimized operations.
@@ -997,7 +999,7 @@ func (b *Bitmap) countEmptyContainers() int {
 
 // Optimize converts array and bitmap containers to run containers as necessary.
 func (b *Bitmap) Optimize() {
-	b.Containers.UpdateEvery(func(c *Container, existed bool) (*Container, bool) {
+	b.Containers.UpdateEvery(func(key uint64, c *Container, existed bool) (*Container, bool) {
 		return c.optimize(), true
 	})
 }
@@ -1118,6 +1120,233 @@ func (b *Bitmap) writeToUnoptimized(w io.Writer) (n int64, err error) {
 	return n, nil
 }
 
+// roaringIterator represents something which can iterate through a roaring
+// bitmap and yield information about containers, including type, size, and
+// the location of their data structures.
+type roaringIterator struct {
+	data              []byte
+	keys              int64
+	headers           []byte
+	offsets           []byte
+	currentKey        uint64
+	currentIdx        int64
+	currentType       byte
+	currentN          int
+	currentLen        int
+	currentPointer    *uint16
+	currentDataOffset uint32
+	lastErr           error
+}
+
+func newRoaringIterator(data []byte) (*roaringIterator, error) {
+	if len(data) < headerBaseSize {
+		return nil, errors.New("invalid data: not long enough to be a roaring header")
+	}
+	// Verify the first two bytes are a valid MagicNumber, and second two bytes match current storageVersion.
+	fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2]))
+	fileVersion := uint32(data[2])
+	if fileMagic != MagicNumber {
+		return nil, fmt.Errorf("invalid roaring file, magic number %v is incorrect", fileMagic)
+	}
+	if fileVersion != storageVersion {
+		return nil, fmt.Errorf("wrong roaring version, file is v%d, server requires v%d", fileVersion, storageVersion)
+	}
+	r := &roaringIterator{data: data}
+	// Read key count in bytes sizeof(cookie)+sizeof(flag):(sizeof(cookie)+sizeof(uint32)).
+	r.keys = int64(binary.LittleEndian.Uint32(data[3+1 : 8]))
+	// it could happen
+	if r.keys == 0 {
+		// not an error, exactly. it's valid and well-formed, we just have nothing to do
+		r.Done(io.EOF)
+		return r, nil
+	}
+	if int64(len(data)) < int64(headerBaseSize+(r.keys*16)) {
+		return nil, fmt.Errorf("insufficient data for header + offsets: want %d bytes, got %d",
+			headerBaseSize+(r.keys*16), len(data))
+	}
+
+	headerStart := int64(headerBaseSize)
+	headerEnd := headerStart + (r.keys * 12)
+	offsetStart := headerEnd
+	offsetEnd := offsetStart + (r.keys * 4)
+	r.headers = data[headerStart:headerEnd]
+	r.offsets = data[offsetStart:offsetEnd]
+	// set key to -1; user should call Next first.
+	r.currentIdx = -1
+	r.currentKey = ^uint64(0)
+	r.lastErr = errors.New("tried to read iterator without calling Next first")
+	return r, nil
+}
+
+// Done marks the iterator as complete, recording err as the reason why
+func (r *roaringIterator) Done(err error) {
+	r.lastErr = err
+	r.currentKey = ^uint64(0)
+	r.currentType = 0
+	r.currentN = 0
+	r.currentLen = 0
+	r.currentPointer = nil
+	r.currentDataOffset = 0
+}
+
+func (r *roaringIterator) Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
+	if r.currentIdx >= r.keys {
+		// we're already done
+		return r.Current()
+	}
+	r.currentIdx++
+	if r.currentIdx == r.keys {
+		// this is the last key. transition state to the finalized state
+		r.Done(io.EOF)
+		return r.Current()
+	}
+	header := r.headers[r.currentIdx*12:]
+	r.currentKey = binary.LittleEndian.Uint64(header[0:8])
+	r.currentType = byte(binary.LittleEndian.Uint16(header[8:10]))
+	r.currentN = int(binary.LittleEndian.Uint16(header[10:12])) + 1
+	r.currentDataOffset = binary.LittleEndian.Uint32(r.offsets[r.currentIdx*4:])
+	// a run container keeps its data after an initial 2 byte length header
+	if r.currentType == containerRun {
+		r.currentDataOffset += 2
+	}
+	if r.currentDataOffset > uint32(len(r.data)) || r.currentDataOffset < headerBaseSize {
+		r.Done(fmt.Errorf("container %d/%d, key %d, had offset %d, maximum %d",
+			r.currentIdx, r.keys, r.currentKey, r.currentDataOffset, len(r.data)))
+		return r.Current()
+	}
+	r.currentPointer = (*uint16)(unsafe.Pointer(&r.data[r.currentDataOffset]))
+	var size int
+	switch r.currentType {
+	case containerArray:
+		r.currentLen = r.currentN
+		size = r.currentLen * 2
+	case containerBitmap:
+		r.currentLen = 1024
+		size = 8192
+	case containerRun:
+		r.currentLen = int(*((*uint16)(unsafe.Pointer(&r.data[r.currentDataOffset-2]))))
+		size = r.currentLen * 4
+	}
+	if int64(r.currentDataOffset)+int64(size) > int64(len(r.data)) {
+		r.Done(fmt.Errorf("container %d/%d, key %d, had offset %d+%d size, maximum %d",
+			r.currentIdx, r.keys, r.currentKey, r.currentDataOffset, size, len(r.data)))
+		return r.Current()
+	}
+	r.lastErr = nil
+	return r.Current()
+}
+
+func (r *roaringIterator) Current() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
+	return r.currentKey, r.currentType, r.currentN, r.currentLen, r.currentPointer, r.lastErr
+}
+
+// ImportRoaringBits sets-or-clears bits based on a provided Roaring bitmap.
+// This should be equivalent to unmarshalling the bitmap, then executing
+// either `b = Union(b, newB)` or `b = Difference(b, newB)`, but with lower
+// overhead. The log parameter controls whether to write to the op log; the
+// answer should always be yes, except if you're calling using this to apply
+// the op log.
+//
+// If rowSize is non-zero, we should return a map of rows we altered,
+// where "rows" are sets of rowSize containers. Otherwise the map isn't used.
+// (This allows ImportRoaring to update caches; see fragment.go.)
+func (b *Bitmap) ImportRoaringBits(data []byte, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) {
+	if data == nil {
+		return 0, nil, errors.New("no roaring bitmap provided")
+	}
+	var itr *roaringIterator
+	var itrKey uint64
+	var itrCType byte
+	var itrN int
+	var itrLen int
+	var itrPointer *uint16
+	var itrErr error
+
+	itr, err = newRoaringIterator(data)
+	if err != nil {
+		return 0, nil, err
+	}
+	if itr == nil {
+		return 0, nil, errors.New("failed to create roaring iterator, but don't know why")
+	}
+
+	rowSet = make(map[uint64]int)
+
+	var synthC Container
+	var importUpdater func(*Container, bool) (*Container, bool)
+	var currRow uint64
+	if clear {
+		importUpdater = func(oldC *Container, existed bool) (newC *Container, write bool) {
+			existN := oldC.N()
+			if existN == 0 || !existed {
+				return nil, false
+			}
+			newC = difference(oldC, &synthC)
+			if newC.N() != existN {
+				changes := int(existN - newC.N())
+				changed += changes
+				rowSet[currRow] -= changes
+				return newC, true
+			}
+			return oldC, false
+		}
+	} else {
+		importUpdater = func(oldC *Container, existed bool) (newC *Container, write bool) {
+			existN := oldC.N()
+			if existN == maxContainerVal+1 {
+				return oldC, false
+			}
+			if existN == 0 {
+				newerC := synthC.Clone()
+				changed += int(newerC.N())
+				rowSet[currRow] += int(newerC.N())
+				return newerC, true
+			}
+			newC = oldC.unionInPlace(&synthC)
+			if newC.typeID == containerBitmap {
+				newC.Repair()
+			}
+			if newC.N() != existN {
+				changes := int(newC.N() - existN)
+				changed += changes
+				rowSet[currRow] += changes
+				return newC, true
+			}
+			return oldC, false
+		}
+	}
+	itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next()
+	for itrErr == nil {
+		synthC.typeID = itrCType
+		synthC.n = int32(itrN)
+		synthC.len = int32(itrLen)
+		synthC.cap = int32(itrLen)
+		synthC.pointer = itrPointer
+		if rowSize != 0 {
+			currRow = itrKey / rowSize
+		}
+		b.Containers.Update(itrKey, importUpdater)
+		itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next()
+	}
+	// note: if we get a non-EOF err, it's possible that we made SOME
+	// changes but didn't log them. I don't have a good solution to this.
+	if itrErr != io.EOF {
+		return changed, rowSet, itrErr
+	}
+	err = nil
+	if log {
+		op := op{opN: changed, roaring: data}
+		if clear {
+			op.typ = opTypeRemoveRoaring
+		} else {
+			op.typ = opTypeAddRoaring
+		}
+		err = b.writeOp(&op)
+	}
+	return changed, rowSet, err
+
+}
+
 // unmarshalPilosaRoaring treats data as being encoded in Pilosa's 64 bit
 // roaring format and decodes it into b.
 func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
@@ -1144,7 +1373,7 @@ func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
 	}
 
 	headerSize := headerBaseSize
-	b.Containers.Reset()
+	b.Containers.ResetN(int(keyN))
 	// Descriptive header section: Read container keys and cardinalities.
 	for i, buf := 0, data[headerSize:]; i < int(keyN); i, buf = i+1, buf[12:] {
 		b.Containers.PutContainerValues(
@@ -1233,6 +1462,18 @@ func (b *Bitmap) Iterator() *Iterator {
 	return itr
 }
 
+// OpN returns the number of write ops the bitmap is aware of in its ops
+// log.
+func (b *Bitmap) OpN() int {
+	return b.opN
+}
+
+// SetOpN lets us reset the operation count in the weird case where we know
+// we've changed an underlying file, without actually refreshing the bitmap.
+func (b *Bitmap) SetOpN(int) {
+	b.opN = 0
+}
+
 // Info returns stats for the bitmap.
 func (b *Bitmap) Info() bitmapInfo {
 	info := bitmapInfo{
@@ -3961,17 +4202,21 @@ func shiftRun(a *Container) (*Container, bool) {
 type opType uint8
 
 const (
-	opTypeAdd         = opType(0)
-	opTypeRemove      = opType(1)
-	opTypeAddBatch    = opType(2)
-	opTypeRemoveBatch = opType(3)
+	opTypeAdd           = opType(0)
+	opTypeRemove        = opType(1)
+	opTypeAddBatch      = opType(2)
+	opTypeRemoveBatch   = opType(3)
+	opTypeAddRoaring    = opType(4)
+	opTypeRemoveRoaring = opType(5)
 )
 
 // op represents an operation on the bitmap.
 type op struct {
-	typ    opType
-	value  uint64
-	values []uint64
+	typ     opType
+	opN     int
+	value   uint64
+	values  []uint64
+	roaring []byte
 }
 
 // apply executes the operation against a bitmap.
@@ -3985,6 +4230,12 @@ func (op *op) apply(b *Bitmap) (changed bool) {
 		changed = b.DirectAddN(op.values...) > 0
 	case opTypeRemoveBatch:
 		changed = b.DirectRemoveN(op.values...) > 0
+	case opTypeAddRoaring:
+		changedN, _, _ := b.ImportRoaringBits(op.roaring, false, false, 0)
+		changed = changedN != 0
+	case opTypeRemoveRoaring:
+		changedN, _, _ := b.ImportRoaringBits(op.roaring, true, false, 0)
+		changed = changedN != 0
 	default:
 		panic(fmt.Sprintf("invalid op type: %d", op.typ))
 	}
@@ -3993,29 +4244,45 @@ func (op *op) apply(b *Bitmap) (changed bool) {
 
 // WriteTo writes op to the w.
 func (op *op) WriteTo(w io.Writer) (n int64, err error) {
-	buf := make([]byte, op.size())
+	buf := make([]byte, op.encodeSize())
 
 	// Write type and value.
 	buf[0] = byte(op.typ)
-	if op.typ <= 1 {
+	switch op.typ {
+	case 0, 1:
 		binary.LittleEndian.PutUint64(buf[1:9], op.value)
-	} else {
+	case 2, 3:
 		binary.LittleEndian.PutUint64(buf[1:9], uint64(len(op.values)))
 		p := 13 // start of values (skip 4 for checksum)
 		for _, v := range op.values {
 			binary.LittleEndian.PutUint64(buf[p:p+8], v)
 			p += 8
 		}
+	case 4, 5:
+		binary.LittleEndian.PutUint64(buf[1:9], uint64(len(op.roaring)))
+		binary.LittleEndian.PutUint32(buf[13:17], uint32(op.opN))
 	}
 
 	// Add checksum at the end.
 	h := fnv.New32a()
 	_, _ = h.Write(buf[0:9])
 	_, _ = h.Write(buf[13:])
+	if op.typ == 4 || op.typ == 5 {
+		_, _ = h.Write(op.roaring)
+	}
 	binary.LittleEndian.PutUint32(buf[9:13], h.Sum32())
 
 	// Write to writer.
 	nn, err := w.Write(buf)
+	if err != nil {
+		return int64(nn), err
+	}
+	if op.typ == 4 || op.typ == 5 {
+		var nn2 int
+		// separate write so we don't have to copy the whole thing
+		nn2, err = w.Write(op.roaring)
+		nn += nn2
+	}
 	return int64(nn), err
 }
 
@@ -4030,14 +4297,16 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	statsHit("op/UnmarshalBinary")
 
 	op.typ = opType(data[0])
-	// op.value will actually contain the length of values for batch ops
+	// op.value will actually contain the length of values for batch ops, or
+	// length of the roaring bitmap for roaring bitmap ops
 	op.value = binary.LittleEndian.Uint64(data[1:9])
 
 	// Verify checksum.
 	h := fnv.New32a()
 	_, _ = h.Write(data[0:9])
 
-	if op.typ > 1 {
+	switch op.typ {
+	case 2, 3:
 		// 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 op.value > maxBatchSize {
@@ -4053,9 +4322,17 @@ func (op *op) UnmarshalBinary(data []byte) error {
 			op.values[i] = binary.LittleEndian.Uint64(data[start : start+8])
 		}
 		op.value = 0
+	case 4, 5:
+		if len(data) < int(13+4+op.value) {
+			return fmt.Errorf("op data truncated - expected %d, got %d", 13+op.value, len(data))
+		}
+		op.opN = int(binary.LittleEndian.Uint32(data[13:17]))
+		op.roaring = data[17 : 17+op.value]
+		_, _ = h.Write(data[13 : 17+op.value])
+		// op.value = 0
 	}
 	if chk := binary.LittleEndian.Uint32(data[9:13]); chk != h.Sum32() {
-		return fmt.Errorf("checksum mismatch: exp=%08x, got=%08x", h.Sum32(), chk)
+		return fmt.Errorf("checksum mismatch: type %d, exp=%08x, got=%08x", op.typ, h.Sum32(), chk)
 	}
 
 	return nil
@@ -4066,7 +4343,25 @@ func (op *op) size() int {
 	if op.typ == opTypeAdd || op.typ == opTypeRemove {
 		return 1 + 8 + 4
 	}
-	return 1 + 8 + 4 + len(op.values)*8
+	if op.typ == opTypeAddBatch || op.typ == opTypeRemoveBatch {
+		return 1 + 8 + 4 + len(op.values)*8
+	}
+	// else it's presumably roaring?
+	return 1 + 8 + 4 + 4 + len(op.roaring)
+}
+
+// size returns the size needed to encode the op, in bytes. for
+// roaring ops, this does not include the roaring data, which is
+// already encoded.
+func (op *op) encodeSize() int {
+	if op.typ == opTypeAdd || op.typ == opTypeRemove {
+		return 1 + 8 + 4
+	}
+	if op.typ == opTypeAddBatch || op.typ == opTypeRemoveBatch {
+		return 1 + 8 + 4 + len(op.values)*8
+	}
+	// else it's presumably roaring?
+	return 1 + 8 + 4 + 4
 }
 
 // count returns the number of bits the operation mutates.
@@ -4076,6 +4371,8 @@ func (op *op) count() int {
 		return 1
 	case 2, 3:
 		return len(op.values)
+	case 4, 5:
+		return op.opN
 	default:
 		panic(fmt.Sprintf("unknown operation type: %d", op.typ))
 	}
@@ -4429,6 +4726,75 @@ func xorBitmapRun(a, b *Container) *Container {
 	return output
 }
 
+// CompareEquality is used mostly in test cases to confirm that two bitmaps came
+// out the same. It does not expect corresponding opN, or OpWriter, but expects
+// identical bit contents. It does not expect identical representations; a bitmap
+// container can be identical to an array container. It returns a boolean value,
+// and also an explanation for a false value.
+func (b *Bitmap) BitwiseEqual(c *Bitmap) (bool, error) {
+	biter, _ := b.Containers.Iterator(0)
+	citer, _ := c.Containers.Iterator(0)
+	bn, cn := biter.Next(), citer.Next()
+	var bk, ck uint64
+	var bc, cc *Container
+	bct, cct := 0, 0
+	for bn && cn {
+		bk, bc = biter.Value()
+		ck, cc = citer.Value()
+		// zero containers are allowed to match no-container
+		if bk < ck {
+			if bc.N() == 0 {
+				bn = biter.Next()
+				continue
+			}
+		}
+		if ck < bk {
+			if cc.N() == 0 {
+				cn = citer.Next()
+				continue
+			}
+		}
+		bct++
+		cct++
+		if bk != ck {
+			return false, fmt.Errorf("differing keys [%d vs %d]", bk, ck)
+		}
+		diff := xor(bc, cc)
+		if diff.N() != 0 {
+			return false, fmt.Errorf("differing containers for key %d: %v vs %v", bk, bc, cc)
+		}
+		bn, cn = biter.Next(), citer.Next()
+	}
+	// only one can have containers left. they should all be empty. so we
+	// look at any remaining containers, break out of the loop if they're not
+	// empty, and otherwise keep iterating.
+	for bn {
+		bn = biter.Next()
+		bk, bc = biter.Value()
+		if bc.N() != 0 {
+			bct++
+			break
+		}
+		bn = biter.Next()
+	}
+	for cn {
+		cn = citer.Next()
+		ck, cc = biter.Value()
+		if cc.N() != 0 {
+			cct++
+			break
+		}
+		cn = biter.Next()
+	}
+	if bn {
+		return false, fmt.Errorf("container mismatch: %d vs %d containers, first bitmap has extra container %d [%d bits]", bct, cct, bk, bc)
+	}
+	if cn {
+		return false, fmt.Errorf("container mismatch: %d vs %d containers, second bitmap has extra container %d [%d bits]", bct, cct, ck, cc)
+	}
+	return true, nil
+}
+
 func bitmapsEqual(b, c *Bitmap) error { // nolint: deadcode
 	statsHit("bitmapsEqual")
 	if b.OpWriter != c.OpWriter {
@@ -4563,7 +4929,7 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error {
 	}
 	b.Flags = flags
 
-	b.Containers.Reset()
+	b.Containers.ResetN(int(keyN))
 	// Descriptive header section: Read container keys and cardinalities.
 	for i, buf := uint(0), data[header:]; i < uint(keyN); i, buf = i+1, buf[4:] {
 		card := int(binary.LittleEndian.Uint16(buf[2:4])) + 1

From 7f1763e466abd0b8cb73c9c3b7da2312ef989b9b Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Tue, 25 Jun 2019 14:29:25 -0500
Subject: [PATCH 03/13] address fuzz testing for new op types

The new op type code changed the failure mode for
one of the fuzz test issues -- and the fuzz test revealed a
bug in the code. Fixed the code, updated the test to expect
the newer, better, message.

Also fixed capitalization on the old message.
---
 roaring/fuzz_test.go | 2 +-
 roaring/roaring.go   | 6 +++++-
 2 files changed, 6 insertions(+), 2 deletions(-)

diff --git a/roaring/fuzz_test.go b/roaring/fuzz_test.go
index e56d3f0ce..da74202f0 100644
--- a/roaring/fuzz_test.go
+++ b/roaring/fuzz_test.go
@@ -30,7 +30,7 @@ func TestUnmarshalBinary(t *testing.T) {
 		{ // Checks for int overflow
 			cr: []byte("<0\x000\x00\x00\x00\x00000000000000" +
 				"0"), //"<000000000000000"
-			expected: "unmarshaling as pilosa roaring: Maximum operation size exceeded",
+			expected: "unmarshaling as pilosa roaring: unknown op type: 48",
 		},
 		{ // The next 5 check for malformed bitmaps
 			cr: []byte("<0\x0000000000000000000" +
diff --git a/roaring/roaring.go b/roaring/roaring.go
index 28892af6a..ab48cf338 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -4306,11 +4306,13 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	_, _ = h.Write(data[0:9])
 
 	switch op.typ {
+	case 0, 1:
+		// nothing to do, just being not-default
 	case 2, 3:
 		// 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 op.value > maxBatchSize {
-			return fmt.Errorf("Maximum operation size exceeded")
+			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))
@@ -4330,6 +4332,8 @@ func (op *op) UnmarshalBinary(data []byte) error {
 		op.roaring = data[17 : 17+op.value]
 		_, _ = h.Write(data[13 : 17+op.value])
 		// op.value = 0
+	default:
+		return fmt.Errorf("unknown op type: %d", op.typ)
 	}
 	if chk := binary.LittleEndian.Uint32(data[9:13]); chk != h.Sum32() {
 		return fmt.Errorf("checksum mismatch: type %d, exp=%08x, got=%08x", op.typ, h.Sum32(), chk)

From 565288f6c2442ef6f24c7a97a8253a28a6e9c2f0 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Tue, 25 Jun 2019 14:57:20 -0500
Subject: [PATCH 04/13] use ImportRoaringBits to implement importRoaring

Instead of fancy bitmap ops or ImportPositions, we use the
recently-added ImportRoaringBits operations, which can dump
themselves to op logs much more efficiently, and which are
also usually much more efficient than things like "create a
new bitmap which is a copy of the old one".
---
 fragment.go | 83 ++++++++++++++---------------------------------------
 1 file changed, 21 insertions(+), 62 deletions(-)

diff --git a/fragment.go b/fragment.go
index f81a0508a..622341574 100644
--- a/fragment.go
+++ b/fragment.go
@@ -502,7 +502,7 @@ func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err
 	delete(f.checksums, int(rowID/HashBlockSize))
 
 	// Increment number of operations until snapshot is required.
-	if err := f.incrementOpN(); err != nil {
+	if err := f.incrementOpN(1); err != nil {
 		return false, errors.Wrap(err, "incrementing")
 	}
 
@@ -566,7 +566,7 @@ func (f *fragment) unprotectedClearBit(rowID, columnID uint64) (changed bool, er
 	delete(f.checksums, int(rowID/HashBlockSize))
 
 	// Increment number of operations until snapshot is required.
-	if err := f.incrementOpN(); err != nil {
+	if err := f.incrementOpN(1); err != nil {
 		return false, errors.Wrap(err, "incrementing")
 	}
 
@@ -2046,89 +2046,48 @@ func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint
 // https://github.com/RoaringBitmap/RoaringFormatSpec or from pilosa's version
 // of the roaring format. The cache is updated to reflect the new data.
 func (f *fragment) importRoaring(ctx context.Context, data []byte, clear bool) error {
+	rowSize := uint64(1 << shardVsContainerExponent)
 	span, ctx := tracing.StartSpanFromContext(ctx, "fragment.importRoaring")
 	defer span.Finish()
 	span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.AcquireFragmentLock")
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	span.Finish()
-	bm := roaring.NewBTreeBitmap()
-	span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.UnmarshalBinary")
-	err := bm.UnmarshalBinary(data)
+	span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.ImportRoaringBits")
+	changed, rowSet, err := f.storage.ImportRoaringBits(data, clear, true, rowSize)
 	span.Finish()
 	if err != nil {
 		return err
 	}
 
-	// get a list of keys in order to update the cache
-	iter, _ := bm.Containers.Iterator(0)
-	rowSet := make(map[uint64]struct{})
-	var lastRow uint64 = math.MaxUint64
+	updateCache := f.CacheType != CacheTypeNone
+	anyChanged := false
 
-	incomingCnt := uint64(0)
-	for iter.Next() {
-		key, c := iter.Value()
-		incomingCnt += uint64(c.N())
-
-		// virtual row for the current container
-		vRow := key >> shardVsContainerExponent
-
-		// skip dups
-		if vRow == lastRow {
+	for rowID, changes := range rowSet {
+		if changes == 0 {
 			continue
 		}
-		rowSet[vRow] = struct{}{}
-		lastRow = vRow
-	}
-
-	// take smallPath? TODO - ideally instead of checking f.storage.Any(), the
-	// test here would be if the storage size (in bytes) is significantly
-	// greater than the size of the incoming bits serialized as append
-	// operations. Getting the storage size might be a bit expensive though
-	// especially if the fragment isn't mapped.
-	if incomingCnt+uint64(f.opN) <= uint64(f.MaxOpN) && f.storage.Any() {
-		toSet, toClear := bm.Slice(), []uint64{}
-		if clear {
-			toSet, toClear = toClear, toSet
+		f.rowCache.Add(rowID, nil)
+		if updateCache {
+			anyChanged = true
+			f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes))
 		}
-		span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.ImportPositions")
-		err := f.importPositions(toSet, toClear, rowSet)
-		span.Finish()
-		return err
+	}
+	// we only set this if we need to update the cache
+	if anyChanged {
+		f.cache.Recalculate()
 	}
 
-	if clear {
-		span, ctx = tracing.StartSpanFromContext(ctx, "importRoaringDifference")
-		bm = f.storage.Difference(bm)
-		span.Finish()
-	} else if f.storage.Containers.Size() >= bm.Containers.Size() {
-		span, ctx = tracing.StartSpanFromContext(ctx, "importRoaringStorageUIP")
-		f.storage.UnionInPlace(bm)
-		bm = f.storage
-		span.Finish()
-	} else {
-		span, ctx = tracing.StartSpanFromContext(ctx, "importRoaringBitmapUIP")
-		bm.UnionInPlace(f.storage)
-		span.Finish()
-	}
-
-	for rowID := range rowSet {
-		n := bm.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
-		f.cache.BulkAdd(rowID, n)
-	}
-	f.cache.Recalculate()
-
-	span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.WriteToFragment")
-	n, err := unprotectedWriteToFragment(f, bm)
-	span.LogKV("bytesWritten", n)
+	span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.incrementOpN")
+	f.incrementOpN(changed)
 	span.Finish()
 	return err
 }
 
 // incrementOpN increase the operation count by one.
 // If the count exceeds the maximum allowed then a snapshot is performed.
-func (f *fragment) incrementOpN() error {
-	f.opN++
+func (f *fragment) incrementOpN(changed int) error {
+	f.opN += changed
 	if f.opN <= f.MaxOpN {
 		return nil
 	}

From b369dace696e25e3c574f0e274f18915315e3648 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Tue, 25 Jun 2019 15:16:52 -0500
Subject: [PATCH 05/13] remap storage on reopen, instead of remarshalling it

When we do a snapshot, we may end up with containers which are
mmapped to the old file, and containers which have allocated storage
identical to the contents of the new file. It would be nicer if they
were mapped to it. But unmarshalling the entire file is expensive.

Instead, we remap it. (Or, if we couldn't mmap it, just make sure
the old stuff is no longer using the old storage space before we
munmap it.)
---
 fragment.go               | 174 ++++++++++++++++++++++++++++++--------
 fragment_internal_test.go |   6 +-
 roaring/roaring.go        |  76 +++++++++++++++++
 view.go                   |   4 +-
 4 files changed, 221 insertions(+), 39 deletions(-)

diff --git a/fragment.go b/fragment.go
index 622341574..4a42a7255 100644
--- a/fragment.go
+++ b/fragment.go
@@ -177,7 +177,7 @@ func (f *fragment) Open() error {
 	if err := func() error {
 		// Initialize storage in a function so we can close if anything goes wrong.
 		f.Logger.Debugf("open storage for index/field/view/fragment: %s/%s/%s/%d", f.index, f.field, f.view, f.shard)
-		if err := f.openStorage(); err != nil {
+		if err := f.openStorage(true); err != nil {
 			return errors.Wrap(err, "opening storage")
 		}
 
@@ -215,12 +215,43 @@ func (f *fragment) reopen() (mustClose bool, err error) {
 	return mustClose, nil
 }
 
-// openStorage opens the storage bitmap.
-func (f *fragment) openStorage() error {
+// openStorage opens the storage bitmap. Usually you also want to read in
+// the storage, but in the case where we just wrote that file, such as
+// unprotectedWriteToFragment, we could also just... not. If we didn't
+// have existing storage, we probably need to unmarshal the data. If the
+// file we're asked to open is empty, we probably don't.
+//
+// If we already had mapped storage previously, we want to unmap that, and
+// possibly remap it from the file, but we don't need a full unmarshal, just
+// an update of mapped pointers.
+//
+// unmarshalData is somewhat overloaded. it tells us whether or not we
+// need to actually create a bitmap from the data (if the data exists to
+// do this from).
+//
+// usually unmarshalData is only set to false when we're in the middle of
+// a snapshot, and unprotectedWriteToFragment just wrote the in-memory data
+// out.
+//
+// If we have existing storage data, and we successfully get new data,
+// we will unmap the existing storage data.
+//
+// This function's design is probably a problem -- it is trying to handle
+// both cases where there was existing data before, and cases where we
+// just wrote the data.
+func (f *fragment) openStorage(unmarshalData bool) error {
+	oldStorageData := f.storageData
+	// there's a few places where we might encounter an error, but need
+	// to continue past it through other error checks, before returning it.
+	var lastError error
+
 	// Create a roaring bitmap to serve as storage for the shard.
 	if f.storage == nil {
 		f.storage = roaring.NewFileBitmap()
 		f.storage.Flags = f.flags
+		// if we didn't actually have storage, we *do* need to
+		// unmarshal this data in order to have any.
+		unmarshalData = true
 	}
 	// Open the data file to be mmap'd and used as an ops log.
 	file, mustClose, err := syswrap.OpenFile(f.path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
@@ -237,13 +268,24 @@ func (f *fragment) openStorage() error {
 		return fmt.Errorf("flock: %s", err)
 	}
 
+	// data is the data we would unmarshal from, if we're unmarshalling; it might
+	// be obtained by calling ReadAll on a file.
+	//
+	// newStorageData is the data we should map things to. it is set only if
+	// mmapped; if we didn't mmap (say, we couldn't), we won't want to unmap
+	// the ioutil byte slice. (Theoretically, we shouldn't be using the mapped
+	// flag in that case...)
+	var data []byte
+	var newStorageData []byte
+
 	// If the file is empty then initialize it with an empty bitmap.
 	fi, err := f.file.Stat()
 	if err != nil {
 		return errors.Wrap(err, "statting file before")
 	} else if fi.Size() == 0 {
 		bi := bufio.NewWriter(f.file)
-		if _, err := f.storage.WriteTo(bi); err != nil {
+		var err error
+		if _, err = f.storage.WriteTo(bi); err != nil {
 			return fmt.Errorf("init storage file: %s", err)
 		}
 		bi.Flush()
@@ -251,39 +293,94 @@ func (f *fragment) openStorage() error {
 		if err != nil {
 			return errors.Wrap(err, "statting file after")
 		}
+		// there's nothing here, we're not going to try to unmarshal it.
+		unmarshalData = false
+		f.rowCache = &simpleCache{make(map[uint64]*Row)}
 	} else {
 		// Mmap the underlying file so it can be zero copied.
-		data, err := syswrap.Mmap(int(f.file.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
+		data, err = syswrap.Mmap(int(f.file.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
 		if err == syswrap.ErrMaxMapCountReached {
 			f.Logger.Debugf("maximum number of maps reached, reading file instead")
-			data, err = ioutil.ReadAll(file)
-			if err != nil {
-				return errors.Wrap(err, "failure file readall")
+			if unmarshalData {
+				data, err = ioutil.ReadAll(file)
+				if err != nil {
+					return errors.Wrap(err, "failure file readall")
+				}
 			}
 		} else if err != nil {
 			return errors.Wrap(err, "mmap failed")
 		} else {
-			f.storageData = data
-			// Advise the kernel that the mmap is accessed randomly.
-			if err := madvise(f.storageData, syscall.MADV_RANDOM); err != nil {
-				return fmt.Errorf("madvise: %s", err)
-			}
+			newStorageData = data
 		}
-
-		if err := f.storage.UnmarshalBinary(data); err != nil {
-			return fmt.Errorf("unmarshal storage: file=%s, err=%s", f.file.Name(), err)
-		}
-
 	}
 
-	f.opN = f.storage.Info().OpN
+	if unmarshalData {
+		f.storageData = newStorageData
+		// We're about to either re-read the bitmap, or fail to do so
+		// and unconditionally unmap the existing stuff. Either way, we
+		// want to unmap the old storage data after we're done here, but
+		// we can't unmap it yet because it's still live until sometime
+		// later, but we can't unmap it later, because we could return
+		// early... this is what defer is for.
+		if oldStorageData != nil {
+			defer func() {
+				unmapErr := syswrap.Munmap(oldStorageData)
+				if unmapErr != nil {
+					f.Logger.Printf("unmap of old storage failed: %s", err)
+				}
+			}()
+		}
+		// so we have a problem here: if this fails, it's unclear whether
+		// *either* or *both* of old and new storage data might be in use.
+		// So we call the thing that should unconditionally unmap both of them...
+		if err := f.storage.UnmarshalBinary(data); err != nil {
+			f.storage.RemapRoaringStorage(nil)
+			return fmt.Errorf("unmarshal storage: file=%s, err=%s", f.file.Name(), err)
+		}
+		f.rowCache = &simpleCache{make(map[uint64]*Row)}
+		f.opN = f.storage.OpN()
+		// slightly incorrect, but we can't measure it so...
+	} else {
+		// we're moving to new storage, so instead of using the OpN
+		// derived from reading that storage, we notify the bitmap that
+		// OpN is now effectively zero.
+		f.opN = 0
+		f.storage.SetOpN(0)
+		// if oldStorageData is nil, this just tries to unmap any bits that
+		// are currently mapped. otherwise, it will point them at this
+		// storage (if the containers match).
+		var mappedAny bool
+		mappedAny, lastError = f.storage.RemapRoaringStorage(newStorageData)
+		if oldStorageData != nil {
+			unmapErr := syswrap.Munmap(oldStorageData)
+			if unmapErr != nil {
+				f.Logger.Printf("unmap of old storage failed: %s", err)
+			}
+		}
+		if mappedAny {
+			// Advise the kernel that the mmap is accessed randomly.
+			if err := madvise(newStorageData, syscall.MADV_RANDOM); err != nil {
+				lastError = fmt.Errorf("madvise: %s", err)
+			}
+		} else {
+			// if we did map data, but for some reason none of it got used
+			// as backing store, we can unmap it, and set the slice to nil,
+			// so we don't keep the now-invalid slice in f.storageData.
+			if newStorageData != nil {
+				unmapErr := syswrap.Munmap(newStorageData)
+				if unmapErr != nil {
+					lastError = fmt.Errorf("unmapping unused storage data: %s", err)
+				}
+				newStorageData = nil
+			}
+		}
+		f.storageData = newStorageData
+	}
 
 	// Attach the file to the bitmap to act as a write-ahead log.
 	f.storage.OpWriter = f.file
-	f.rowCache = &simpleCache{make(map[uint64]*Row)}
-
-	return nil
 
+	return lastError
 }
 
 // openCache initializes the cache from row ids persisted to disk.
@@ -343,7 +440,7 @@ func (f *fragment) close() error {
 	}
 
 	// Close underlying storage.
-	if err := f.closeStorage(); err != nil {
+	if err := f.closeStorage(true); err != nil {
 		f.Logger.Printf("fragment: error closing storage: err=%s, path=%s", err, f.path)
 		return errors.Wrap(err, "closing storage")
 	}
@@ -374,13 +471,19 @@ func (f *fragment) safeClose() error {
 	return nil
 }
 
-func (f *fragment) closeStorage() error {
+// closeStorage attempts to close storage, including unmapping the old
+// storage if includeMap is true. This would normally make sense if you're
+// expecting to be done using the fragment, or to reload it. But it's also
+// okay to just leave stuff mmapped; you don't have to keep the file
+// descriptor open. So in some cases, we'll just leave the old mmapping
+// in place, rather than regenerating everything from the new file.
+func (f *fragment) closeStorage(includeMap bool) error {
 	// Clear the storage bitmap so it doesn't access the closed mmap.
 
 	//f.storage = roaring.NewBitmap()
 
 	// Unmap the file.
-	if f.storageData != nil {
+	if includeMap && f.storageData != nil {
 		if err := syswrap.Munmap(f.storageData); err != nil {
 			return fmt.Errorf("munmap: %s", err)
 		}
@@ -1994,8 +2097,8 @@ func (f *fragment) importValueSmallWrite(columnIDs []uint64, values []int64, bit
 		}
 		return nil
 	}(); err != nil {
-		_ = f.closeStorage()
-		_ = f.openStorage()
+		_ = f.closeStorage(true)
+		_ = f.openStorage(true)
 		return err
 	}
 	rowSet := make(map[uint64]struct{}, bitDepth+1)
@@ -2033,8 +2136,8 @@ func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint
 		}
 		return nil
 	}(); err != nil {
-		_ = f.closeStorage()
-		_ = f.openStorage()
+		_ = f.closeStorage(true)
+		_ = f.openStorage(true)
 		return err
 	}
 
@@ -2118,7 +2221,6 @@ func (f *fragment) snapshot() error {
 // unprotectedWriteToFragment writes the fragment f with bm as the data. It is unprotected, and
 // f.mu must be locked when calling it.
 func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) (n int64, err error) { // nolint: interfacer
-
 	completeMessage := fmt.Sprintf("fragment: snapshot complete %s/%s/%s/%d", f.index, f.field, f.view, f.shard)
 	start := time.Now()
 	defer track(start, completeMessage, f.stats, f.Logger)
@@ -2142,7 +2244,7 @@ func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) (n int64, err e
 	}
 
 	// Close current storage.
-	if err := f.closeStorage(); err != nil {
+	if err := f.closeStorage(false); err != nil {
 		return n, fmt.Errorf("close storage: %s", err)
 	}
 
@@ -2151,8 +2253,12 @@ func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) (n int64, err e
 		return n, fmt.Errorf("rename snapshot: %s", err)
 	}
 
+	// if we reloaded from the file, we'd end up with this bitmap
+	// as our storage. so... let's use this bitmap. as our storage.
+	f.storage = bm
+
 	// Reopen storage.
-	if err := f.openStorage(); err != nil {
+	if err := f.openStorage(false); err != nil {
 		return n, fmt.Errorf("open storage: %s", err)
 	}
 
@@ -2341,7 +2447,7 @@ func (f *fragment) readStorageFromArchive(r io.Reader) error {
 	}
 
 	// Close current storage.
-	if err := f.closeStorage(); err != nil {
+	if err := f.closeStorage(true); err != nil {
 		return errors.Wrap(err, "closing")
 	}
 
@@ -2351,7 +2457,7 @@ func (f *fragment) readStorageFromArchive(r io.Reader) error {
 	}
 
 	// Reopen storage.
-	if err := f.openStorage(); err != nil {
+	if err := f.openStorage(true); err != nil {
 		return errors.Wrap(err, "opening")
 	}
 
diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index dfa470144..703deba7d 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -117,7 +117,7 @@ func TestFragment_RowcacheMap(t *testing.T) {
 		_, _ = f.setBit(0, uint64(i*32))
 	}
 	// force snapshot so we get a mmapped row...
-	_ = f.snapshot()
+	_ = f.Snapshot()
 	row := f.row(0)
 	segment := row.Segments()[0]
 	bitmap := segment.data
@@ -3227,7 +3227,7 @@ func TestImportClearRestart(t *testing.T) {
 				f2.MaxOpN = maxOpN
 				f2.CacheType = f.CacheType
 
-				err = f.closeStorage()
+				err = f.closeStorage(true)
 				if err != nil {
 					t.Fatalf("closing storage: %v", err)
 				}
@@ -3261,7 +3261,7 @@ func TestImportClearRestart(t *testing.T) {
 				f3.MaxOpN = maxOpN
 				f3.CacheType = f.CacheType
 
-				err = f2.closeStorage()
+				err = f2.closeStorage(true)
 				if err != nil {
 					t.Fatalf("f2 closing storage: %v", err)
 				}
diff --git a/roaring/roaring.go b/roaring/roaring.go
index ab48cf338..6a360ba9b 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -1240,6 +1240,82 @@ func (r *roaringIterator) Current() (key uint64, cType byte, n int, length int,
 	return r.currentKey, r.currentType, r.currentN, r.currentLen, r.currentPointer, r.lastErr
 }
 
+// RemapRoaringStorage tries to update all containers to refer to
+// the roaring bitmap in the provided []byte. If any containers are
+// marked as mapped, but do not match the provided storage, they will
+// be unmapped. The boolean return indicates whether or not any
+// containers were mapped to the given storage.
+//
+// Regardless, after this function runs, no containers have
+// mapped storage which does not refer to data; either they got mapped
+// to the new storage, or storage was allocated for them.
+//
+// Data should be in the Pilosa roaring format.
+func (b *Bitmap) RemapRoaringStorage(data []byte) (mappedAny bool, returnErr error) {
+	if b.Containers == nil {
+		return false, nil
+	}
+	var itr *roaringIterator
+	var err error
+	var itrKey uint64
+	var itrCType byte
+	var itrN int
+	var itrPointer *uint16
+	var itrErr error
+
+	if data != nil {
+		itr, err = newRoaringIterator(data)
+	}
+	// don't return early: we still have to do the unmapping
+	if err != nil {
+		returnErr = err
+	}
+
+	if itr != nil {
+		itrKey, itrCType, itrN, _, itrPointer, itrErr = itr.Next()
+	}
+	if itrErr != nil {
+		// iterator errored out, so we won't check it in the loop below
+		itr = nil
+	}
+
+	b.Containers.UpdateEvery(func(key uint64, oldC *Container, existed bool) (newC *Container, write bool) {
+		if itr != nil {
+			for itrKey < key && itrErr == nil {
+				itrKey, itrCType, itrN, _, itrPointer, itrErr = itr.Next()
+			}
+			if itrErr != nil {
+				itr = nil
+			}
+			// container might be similar enough that we should trust it:
+			if itrKey == key && itrCType == oldC.typ() && itrN == int(oldC.N()) {
+				if oldC.frozen() {
+					// we don't use Clone, because that would copy the
+					// storage, and we don't need that.
+					var halfCopy Container
+					halfCopy = *oldC
+					halfCopy.flags &^= flagFrozen
+					newC = &halfCopy
+				} else {
+					newC = oldC
+				}
+				mappedAny = true
+				newC.pointer = itrPointer
+				newC.flags |= flagMapped
+				return newC, true
+			}
+		}
+		// if the container isn't mapped, we don't need to do anything
+		if !oldC.Mapped() {
+			return oldC, false
+		}
+		// forcibly unmap it, so the old mapping can be unmapped safely.
+		newC = oldC.unmapOrClone()
+		return newC, true
+	})
+	return mappedAny, returnErr
+}
+
 // ImportRoaringBits sets-or-clears bits based on a provided Roaring bitmap.
 // This should be equivalent to unmarshalling the bitmap, then executing
 // either `b = Union(b, newB)` or `b = Difference(b, newB)`, but with lower
diff --git a/view.go b/view.go
index 85c9ebbba..620bbcee9 100644
--- a/view.go
+++ b/view.go
@@ -441,11 +441,11 @@ func upgradeViewBSIv2(v *view, bitDepth uint) (ok bool, _ error) {
 
 		if tmpPath, err := upgradeRoaringBSIv2(frag, bitDepth); err != nil {
 			return ok, errors.Wrap(err, "upgrading bsi v2")
-		} else if err := frag.closeStorage(); err != nil {
+		} else if err := frag.closeStorage(true); err != nil {
 			return ok, errors.Wrap(err, "closing after bsi v2 upgrade")
 		} else if err := os.Rename(tmpPath, frag.path); err != nil {
 			return ok, errors.Wrap(err, "renaming after bsi v2 upgrade")
-		} else if err := frag.openStorage(); err != nil {
+		} else if err := frag.openStorage(true); err != nil {
 			return ok, errors.Wrap(err, "re-opening after bsi v2 upgrade")
 		}
 	}

From 4b657c1962406018841451b6dc6afb70b3d91880 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Tue, 25 Jun 2019 15:25:38 -0500
Subject: [PATCH 06/13] use a queue for snapshot operations

As the size of a fragment grows, the cost of snapshots
increases; with a large fragment getting a lot of large writes,
every write will trigger a snapshot, while any other writes have
to wait for that snapshot before they, too, can trigger a snapshot.

To address this, we introduce a background queue of snapshots.
In general, operations which were omitting their ops log writes
and just snapshotting no longer do; they emit an ops log. This does
mean that, in some cases, the ops log is written and then a snapshot
takes place essentially immediately, which costs us some performance.
However, that only actually happens under very light load; under
heavier load, there's generally going to be multiple writes coalesced
into each snapshot, and the ops log writes for them will be much
cheaper than a full snapshot.
---
 field.go                  |   3 +
 fragment.go               | 239 ++++++++++++++++++++++++++++----------
 fragment_internal_test.go |  39 ++++++-
 holder.go                 |  13 +++
 index.go                  |   4 +-
 roaring/roaring.go        |  23 ++--
 view.go                   |  10 +-
 7 files changed, 249 insertions(+), 82 deletions(-)

diff --git a/field.go b/field.go
index 3dd362ec0..9b49d03b6 100644
--- a/field.go
+++ b/field.go
@@ -84,6 +84,8 @@ type Field struct {
 	remoteAvailableShards *roaring.Bitmap
 
 	logger logger.Logger
+
+	snapshotQueue chan *fragment
 }
 
 // FieldOption is a functional option type for pilosa.fieldOptions.
@@ -838,6 +840,7 @@ func (f *Field) newView(path, name string) *view {
 	view.rowAttrStore = f.rowAttrStore
 	view.stats = f.Stats
 	view.broadcaster = f.broadcaster
+	view.snapshotQueue = f.snapshotQueue
 	return view
 }
 
diff --git a/fragment.go b/fragment.go
index 4a42a7255..bbf7d0afe 100644
--- a/fragment.go
+++ b/fragment.go
@@ -107,12 +107,21 @@ type fragment struct {
 	shard uint64
 
 	// File-backed storage
-	path        string
-	flags       byte // user-defined flags passed to roaring
-	file        *os.File
-	storage     *roaring.Bitmap
-	storageData []byte
-	opN         int // number of ops since snapshot
+	path               string
+	flags              byte // user-defined flags passed to roaring
+	file               *os.File
+	storage            *roaring.Bitmap
+	storageData        []byte
+	totalOpN           int64 // total opN values
+	totalOps           int64 // total ops (across all snapshots)
+	opN                int   // number of ops since snapshot (may be approximate for imports)
+	ops                int   // number of higher-level operations, as opposed to bit changes
+	snapshotsRequested int   // number of times we've requested a snapshot
+	snapshotsTaken     int   // number of actual snapshot operations
+	snapshotting       bool  // set to true when requesting a snapshot, set to false after snapshot completes
+	snapshotCond       sync.Cond
+	snapshotDelays     int
+	snapshotDelayTime  time.Duration
 
 	// Cache for row counts.
 	CacheType string // passed in by field
@@ -145,11 +154,13 @@ type fragment struct {
 	mutexVector vector
 
 	stats stats.StatsClient
+
+	snapshotQueue chan *fragment
 }
 
 // newFragment returns a new instance of Fragment.
 func newFragment(path, index, field, view string, shard uint64, flags byte) *fragment {
-	return &fragment{
+	f := &fragment{
 		path:      path,
 		index:     index,
 		field:     field,
@@ -164,11 +175,68 @@ func newFragment(path, index, field, view string, shard uint64, flags byte) *fra
 
 		stats: stats.NopStatsClient,
 	}
+	f.snapshotCond = sync.Cond{L: &f.mu}
+	return f
 }
 
 // cachePath returns the path to the fragment's cache data.
 func (f *fragment) cachePath() string { return f.path + cacheExt }
 
+// newSnapshotQueue makes a new snapshot queue, of depth N, and spawns a
+// goroutine for it.
+func newSnapshotQueue(n int, w int, l logger.Logger) chan *fragment {
+	ch := make(chan *fragment, n)
+	for i := 0; i < w; i++ {
+		go snapshotQueueWorker(ch, l)
+	}
+	return ch
+}
+
+func snapshotQueueWorker(snapshotQueue chan *fragment, l logger.Logger) {
+	for f := range snapshotQueue {
+		err := f.protectedSnapshot(true)
+		if err != nil {
+			l.Printf("snapshot error: %v", err)
+		}
+		f.snapshotCond.Broadcast()
+	}
+}
+
+// enqueueSnapshot requests that the fragment be snapshotted at some point
+// in the future, if this has not already been requested. Call this only when
+// the mutex is held.
+func (f *fragment) enqueueSnapshot() {
+	f.snapshotsRequested++
+	if f.snapshotting {
+		return
+	}
+	f.snapshotting = true
+	if f.snapshotQueue != nil {
+		select {
+		case f.snapshotQueue <- f:
+		default:
+			before := time.Now()
+			// wait forever, but notice that we're waiting
+			f.snapshotQueue <- f
+			f.snapshotDelays++
+			f.snapshotDelayTime += time.Now().Sub(before)
+			if f.snapshotDelays >= 10 {
+				f.Logger.Printf("snapshotting %s: last ten delays took %v", f.path, f.snapshotDelayTime)
+				f.snapshotDelays = 0
+				f.snapshotDelayTime = 0
+			}
+		case <-time.After(5 * time.Second):
+			f.Logger.Printf("snapshot for %s: timed out\n", f.path)
+		}
+	} else {
+		// in testing, for instance, there may be no holder, thus no one
+		// to handle these snapshots.
+		f.snapshot()
+		f.snapshotting = false
+		f.snapshotCond.Broadcast()
+	}
+}
+
 // Open opens the underlying storage.
 func (f *fragment) Open() error {
 	f.mu.Lock()
@@ -334,18 +402,21 @@ func (f *fragment) openStorage(unmarshalData bool) error {
 		// *either* or *both* of old and new storage data might be in use.
 		// So we call the thing that should unconditionally unmap both of them...
 		if err := f.storage.UnmarshalBinary(data); err != nil {
-			f.storage.RemapRoaringStorage(nil)
+			_, e2 := f.storage.RemapRoaringStorage(nil)
+			if e2 != nil {
+				return fmt.Errorf("unmarshal storage: file=%s, err=%s, clearing old mapping also failed: %v", f.file.Name(), err, e2)
+			}
 			return fmt.Errorf("unmarshal storage: file=%s, err=%s", f.file.Name(), err)
 		}
 		f.rowCache = &simpleCache{make(map[uint64]*Row)}
-		f.opN = f.storage.OpN()
-		// slightly incorrect, but we can't measure it so...
+		f.ops, f.opN = f.storage.Ops()
 	} else {
 		// we're moving to new storage, so instead of using the OpN
 		// derived from reading that storage, we notify the bitmap that
 		// OpN is now effectively zero.
 		f.opN = 0
-		f.storage.SetOpN(0)
+		f.ops = 0
+		f.storage.SetOps(0, 0)
 		// if oldStorageData is nil, this just tries to unmap any bits that
 		// are currently mapped. otherwise, it will point them at this
 		// storage (if the containers match).
@@ -429,9 +500,31 @@ func (f *fragment) openCache() error {
 func (f *fragment) Close() error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
+	for f.snapshotting {
+		f.snapshotCond.Wait()
+	}
 	return f.close()
 }
 
+// awaitSnapshot lets us delay until the snapshot gets written, preventing tests
+// from misleadingly showing amazingly fast performance because the snapshots they
+// trigger haven't happened yet.
+func (f *fragment) awaitSnapshot() {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	for f.snapshotting {
+		f.snapshotCond.Wait()
+	}
+}
+
+// protectedAwaitSnapshot assumes you already hold the lock, and waits for
+// the snapshot fairy to come along.
+func (f *fragment) protectedAwaitSnapshot() {
+	for f.snapshotting {
+		f.snapshotCond.Wait()
+	}
+}
+
 func (f *fragment) close() error {
 	// Flush cache if closing gracefully.
 	if err := f.flushCache(); err != nil {
@@ -732,14 +825,16 @@ func (f *fragment) unprotectedSetRow(row *Row, rowID uint64) (changed bool, err
 	}
 
 	// Update the row in cache.
-	n := f.storage.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
-	f.cache.BulkAdd(rowID, n)
-
-	// Snapshot storage.
-	if err := f.snapshot(); err != nil {
-		return false, errors.Wrap(err, "snapshotting")
+	if f.CacheType != CacheTypeNone {
+		n := f.storage.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
+		f.cache.BulkAdd(rowID, n)
 	}
 
+	// invalidate rowCache for this row.
+	f.rowCache.Add(rowID, nil)
+
+	// Snapshot storage.
+	f.enqueueSnapshot()
 	f.stats.Count("setRow", 1, 1.0)
 
 	return changed, nil
@@ -780,11 +875,10 @@ func (f *fragment) unprotectedClearRow(rowID uint64) (changed bool, err error) {
 
 	// Clear the row in cache.
 	f.cache.Add(rowID, 0)
+	f.rowCache.Add(rowID, nil)
 
 	// Snapshot storage.
-	if err := f.snapshot(); err != nil {
-		return false, errors.Wrap(err, "snapshotting")
-	}
+	f.enqueueSnapshot()
 
 	f.stats.Count("clearRow", 1, 1.0)
 
@@ -948,7 +1042,7 @@ func (f *fragment) setValueBase(columnID uint64, bitDepth uint, value int64, cle
 }
 
 // importSetValue is a more efficient SetValue just for imports.
-func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, clear bool) (changed bool, err error) { // nolint: unparam
+func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, clear bool) (changed int, err error) { // nolint: unparam
 	// Convert value to an unsigned representation.
 	uvalue := uint64(value)
 	if value < 0 {
@@ -965,13 +1059,13 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, c
 			if c, err := f.storage.Add(bit); err != nil {
 				return changed, errors.Wrap(err, "adding")
 			} else if c {
-				changed = true
+				changed++
 			}
 		} else {
 			if c, err := f.storage.Remove(bit); err != nil {
 				return changed, errors.Wrap(err, "removing")
 			} else if c {
-				changed = true
+				changed++
 			}
 		}
 	}
@@ -983,13 +1077,13 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, c
 		if c, err := f.storage.Remove(p); err != nil {
 			return changed, errors.Wrap(err, "removing not-null from storage")
 		} else if c {
-			changed = true
+			changed++
 		}
 	} else {
 		if c, err := f.storage.Add(p); err != nil {
 			return changed, errors.Wrap(err, "adding not-null to storage")
 		} else if c {
-			changed = true
+			changed++
 		}
 	}
 
@@ -1000,13 +1094,13 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, c
 		if c, err := f.storage.Remove(p); err != nil {
 			return changed, errors.Wrap(err, "removing sign from storage")
 		} else if c {
-			changed = true
+			changed++
 		}
 	} else {
 		if c, err := f.storage.Add(p); err != nil {
 			return changed, errors.Wrap(err, "adding sign to storage")
 		} else if c {
-			changed = true
+			changed++
 		}
 	}
 
@@ -1958,19 +2052,12 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor
 // snapshot of the fragment or just do in-memory updates while appending
 // operations to the op log.
 func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct{}) error {
-	smallWrite := false
-	if len(set)+len(clear)+f.opN < f.MaxOpN {
-		smallWrite = true
-		mustClose, err := f.reopen()
-		if err != nil {
-			return errors.Wrap(err, "reopening")
-		}
-		if mustClose {
-			defer f.safeClose()
-		}
-
-	} else {
-		f.storage.OpWriter = nil
+	mustClose, err := f.reopen()
+	if err != nil {
+		return errors.Wrap(err, "reopening")
+	}
+	if mustClose {
+		defer f.safeClose()
 	}
 
 	if len(set) > 0 {
@@ -1980,7 +2067,7 @@ func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct
 			return errors.Wrap(err, "adding positions")
 		}
 		f.stats.Count("ImportedN", int64(changedN), 1)
-		f.opN += changedN
+		f.incrementOpN(changedN)
 	}
 
 	if len(clear) > 0 {
@@ -1990,7 +2077,7 @@ func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct
 			return errors.Wrap(err, "clearing positions")
 		}
 		f.stats.Count("ClearedN", int64(changedN), 1)
-		f.opN += changedN
+		f.incrementOpN(changedN)
 	}
 
 	// Update cache counts for all affected rows.
@@ -1998,19 +2085,18 @@ func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct
 		// Invalidate block checksum.
 		delete(f.checksums, int(rowID/HashBlockSize))
 
-		n := f.storage.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
-		f.cache.BulkAdd(rowID, n)
-
-		if smallWrite {
-			f.rowCache.Add(rowID, nil)
+		if f.CacheType != CacheTypeNone {
+			n := f.storage.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
+			f.cache.BulkAdd(rowID, n)
 		}
+
+		f.rowCache.Add(rowID, nil)
 	}
 
-	f.cache.Recalculate()
-
-	if !smallWrite {
-		return f.snapshot()
+	if f.CacheType != CacheTypeNone {
+		f.cache.Recalculate()
 	}
+
 	return nil
 }
 
@@ -2123,16 +2209,18 @@ func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint
 		return errors.Wrap(f.importValueSmallWrite(columnIDs, values, bitDepth, clear), "import small write")
 	}
 
-	f.storage.OpWriter = nil
-
 	// Process every value.
 	// If an error occurs then reopen the storage.
+	f.storage.OpWriter = nil
+	totalChanges := 0
 	if err := func() (err error) {
 		for i := range columnIDs {
 			columnID, value := columnIDs[i], values[i]
-			if _, err := f.importSetValue(columnID, bitDepth, value, clear); err != nil {
+			changed, err := f.importSetValue(columnID, bitDepth, value, clear)
+			if err != nil {
 				return errors.Wrapf(err, "importSetValue")
 			}
+			totalChanges += changed
 		}
 		return nil
 	}(); err != nil {
@@ -2140,9 +2228,15 @@ func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint
 		_ = f.openStorage(true)
 		return err
 	}
+	// We don't actually care, except we want our stats to be accurate.
+	f.incrementOpN(totalChanges)
+	// in theory, this should probably have happened anyway, but if enough
+	// of the bits matched existing bits, we'll be under our opN estimate, and
+	// we want to ensure that the snapshot happens.
+	f.enqueueSnapshot()
+	f.protectedAwaitSnapshot()
 
-	err := f.snapshot()
-	return errors.Wrap(err, "snapshotting")
+	return nil
 }
 
 // importRoaring imports from the official roaring data format defined at
@@ -2190,30 +2284,51 @@ func (f *fragment) importRoaring(ctx context.Context, data []byte, clear bool) e
 // incrementOpN increase the operation count by one.
 // If the count exceeds the maximum allowed then a snapshot is performed.
 func (f *fragment) incrementOpN(changed int) error {
-	f.opN += changed
-	if f.opN <= f.MaxOpN {
+	if changed <= 0 {
 		return nil
 	}
-
-	if err := f.snapshot(); err != nil {
-		return fmt.Errorf("snapshot: %s", err)
+	f.opN += changed
+	f.ops++
+	if f.opN > f.MaxOpN {
+		f.enqueueSnapshot()
 	}
 	return nil
 }
 
-// Snapshot writes the storage bitmap to disk and reopens it.
+// Snapshot writes the storage bitmap to disk and reopens it. This may
+// coexist with existing background-queue snapshotting; it does not remove
+// things from the queue. You probably don't want to do this; use
+// enqueueSnapshot/awaitSnapshot.
 func (f *fragment) Snapshot() error {
 	f.mu.Lock()
 	defer f.mu.Unlock()
 	return f.snapshot()
 }
+
 func track(start time.Time, message string, stats stats.StatsClient, logger logger.Logger) {
 	elapsed := time.Since(start)
 	logger.Printf("%s took %s", message, elapsed)
 	stats.Histogram("snapshot", elapsed.Seconds(), 1.0)
 }
 
+// protectedSnapshot grabs the lock and unconditionally calls snapshot(). If
+// fromQueue is true, the snapshotting state is also cleared.
+func (f *fragment) protectedSnapshot(fromQueue bool) error {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	err := f.snapshot()
+	if fromQueue {
+		f.snapshotting = false
+	}
+	return err
+}
+
+// snapshot does the actual snapshot operation. it does not check or care
+// about f.snapshotting.
 func (f *fragment) snapshot() error {
+	f.totalOpN += int64(f.opN)
+	f.totalOps += int64(f.ops)
+	f.snapshotsTaken++
 	_, err := unprotectedWriteToFragment(f, f.storage)
 	return err
 }
diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index 703deba7d..1f2f5c807 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -889,7 +889,6 @@ func BenchmarkFragment_RepeatedSmallValueImports(b *testing.B) {
 								b.Fatalf("importing values: %v", err)
 							}
 						}
-
 					}
 				})
 			}
@@ -2055,6 +2054,7 @@ func BenchmarkImportRoaring(b *testing.B) {
 					b.StartTimer()
 					err := f.importRoaringT(data, false)
 					if err != nil {
+						f.awaitSnapshot()
 						f.Clean(b)
 						b.Fatalf("import error: %v", err)
 					}
@@ -2092,7 +2092,9 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) {
 						for j := 0; j < concurrency; j++ {
 							j := j
 							eg.Go(func() error {
-								return frags[j].importRoaringT(data[j], false)
+								err := frags[j].importRoaringT(data[j], false)
+								frags[j].awaitSnapshot()
+								return err
 							})
 						}
 						err := eg.Wait()
@@ -2125,7 +2127,12 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
 						for i := 0; i < b.N; i++ {
 							for j := 0; j < concurrency; j++ {
 								frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), cacheType)
-								err := frags[j].importRoaringT(data, false)
+								// the cost of actually doing the op log for the large initial data set
+								// is excessive. force storage into snapshotted state, then use import
+								// to generate an op log and/or snapshot.
+								_, _, err := frags[j].storage.ImportRoaringBits(data, false, false, 0)
+								frags[j].enqueueSnapshot()
+								frags[j].awaitSnapshot()
 								if err != nil {
 									b.Fatalf("importing roaring: %v", err)
 								}
@@ -2135,7 +2142,9 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
 							for j := 0; j < concurrency; j++ {
 								j := j
 								eg.Go(func() error {
-									return frags[j].importRoaringT(updata, false)
+									err := frags[j].importRoaringT(updata, false)
+									frags[j].awaitSnapshot()
+									return err
 								})
 							}
 							err := eg.Wait()
@@ -2192,12 +2201,18 @@ func BenchmarkImportRoaringUpdate(b *testing.B) {
 					b.StopTimer()
 					for i := 0; i < b.N; i++ {
 						f := mustOpenFragment("i", fmt.Sprintf("r%dc%dcache_%s", numRows, numCols, cacheType), viewStandard, 0, cacheType)
-						err := f.importRoaringT(data, false)
+						// the cost of actually doing the op log for the large initial data set
+						// is excessive. force storage into snapshotted state, then use import
+						// to generate an op log and/or snapshot.
+						_, _, err := f.storage.ImportRoaringBits(data, false, false, 0)
+						f.enqueueSnapshot()
+						f.awaitSnapshot()
 						if err != nil {
 							b.Errorf("import error: %v", err)
 						}
 						b.StartTimer()
 						err = f.importRoaringT(updata, false)
+						f.awaitSnapshot()
 						if err != nil {
 							f.Clean(b)
 							b.Errorf("import error: %v", err)
@@ -2494,6 +2509,7 @@ func (f *fragment) sanityCheck(t testing.TB) {
 }
 
 func (f *fragment) Clean(t testing.TB) {
+	f.awaitSnapshot()
 	f.sanityCheck(t)
 	errc := f.Close()
 	errf := os.Remove(f.path)
@@ -2501,6 +2517,10 @@ func (f *fragment) Clean(t testing.TB) {
 	if errc != nil || errf != nil {
 		t.Fatal("cleaning up fragment: ", errc, errf, errp)
 	}
+	if f.snapshotQueue != nil {
+		close(f.snapshotQueue)
+		f.snapshotQueue = nil
+	}
 	// not all fragments have cache files
 	if errp != nil && !os.IsNotExist(errp) {
 		t.Fatalf("cleaning up fragment cache: %v", errp)
@@ -2520,6 +2540,10 @@ func (f *fragment) CleanKeep(t testing.TB) {
 	if errc != nil {
 		t.Fatal("closing fragment: ", errc, errp)
 	}
+	if f.snapshotQueue != nil {
+		close(f.snapshotQueue)
+		f.snapshotQueue = nil
+	}
 	// not all fragments have cache files
 	if errp != nil && !os.IsNotExist(errp) {
 		t.Fatalf("cleaning up fragment cache: %v", errp)
@@ -2552,6 +2576,7 @@ func mustOpenFragmentFlags(index, field, view string, shard uint64, cacheType st
 	f.RowAttrStore = &memAttrStore{
 		store: make(map[uint64]map[string]interface{}),
 	}
+	f.snapshotQueue = newSnapshotQueue(1, 1, nil)
 
 	if err := f.Open(); err != nil {
 		panic(err)
@@ -3039,7 +3064,9 @@ func TestUnionInPlaceMapped(t *testing.T) {
 
 	f.storage.UnionInPlace(setBM1)
 	countUnion := f.storage.Count()
-	f.snapshot()
+	// UnionInPlace produces no ops log, we have to make it snapshot, to
+	// ensure that the on-disk representation is correct.
+	f.enqueueSnapshot()
 
 	if count0 != countF {
 		t.Fatalf("writing bitmap to storage changed count: %d => %d", count0, countF)
diff --git a/holder.go b/holder.go
index 9384d6622..692868efe 100644
--- a/holder.go
+++ b/holder.go
@@ -78,6 +78,8 @@ type Holder struct {
 	cacheFlushInterval time.Duration
 
 	Logger logger.Logger
+
+	snapshotQueue chan *fragment
 }
 
 // lockedChan looks a little ridiculous admittedly, but exists for good reason.
@@ -152,6 +154,11 @@ func (h *Holder) Open() error {
 		return errors.Wrap(err, "reading directory")
 	}
 
+	// Run snapshots asynchronously. The snapshotQueue will have a background
+	// task associated with it which flushes it and waits until this channel
+	// is closed, so we should always close this channel when done.
+	h.snapshotQueue = newSnapshotQueue(100, 2, h.Logger)
+
 	for _, fi := range fis {
 		// Skip files or hidden directories.
 		if !fi.IsDir() || strings.HasPrefix(fi.Name(), ".") {
@@ -203,6 +210,11 @@ func (h *Holder) Close() error {
 			return errors.Wrap(err, "closing index")
 		}
 	}
+	if h.snapshotQueue != nil {
+		close(h.snapshotQueue)
+		// assuming the snapshotQueueWorker has already started, this is safe.
+		h.snapshotQueue = nil
+	}
 
 	if h.translateFile != nil {
 		if err := h.translateFile.Close(); err != nil {
@@ -425,6 +437,7 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
 	index.broadcaster = h.broadcaster
 	index.newAttrStore = h.NewAttrStore
 	index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data"))
+	index.snapshotQueue = h.snapshotQueue
 	return index, nil
 }
 
diff --git a/index.go b/index.go
index bef4ff38d..bef25c9c5 100644
--- a/index.go
+++ b/index.go
@@ -53,7 +53,8 @@ type Index struct {
 	broadcaster broadcaster
 	Stats       stats.StatsClient
 
-	logger logger.Logger
+	logger        logger.Logger
+	snapshotQueue chan *fragment
 }
 
 // NewIndex returns a new instance of Index.
@@ -408,6 +409,7 @@ func (i *Index) newField(path, name string) (*Field, error) {
 	f.Stats = i.Stats
 	f.broadcaster = i.broadcaster
 	f.rowAttrStore = i.newAttrStore(filepath.Join(f.path, ".data"))
+	f.snapshotQueue = i.snapshotQueue
 	return f, nil
 }
 
diff --git a/roaring/roaring.go b/roaring/roaring.go
index 6a360ba9b..174ef39a4 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -149,8 +149,9 @@ type Bitmap struct {
 	Flags byte
 
 	// Number of bit change operations written to the writer. Some operations
-	// contain multiple values, each of those counts the number of values rather
-	// than counting as one operation.
+	// contain multiple values, so "ops" represents the number of distinct
+	// operations, while "opN" represents expected bit changes.
+	ops int
 	opN int
 
 	// Writer where operations are appended to.
@@ -1508,6 +1509,7 @@ func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
 		opr.apply(b)
 
 		// Increase the op count.
+		b.ops++
 		b.opN += opr.count()
 
 		// Move the buffer forward.
@@ -1527,6 +1529,7 @@ func (b *Bitmap) writeOp(op *op) error {
 		return err
 	}
 	b.opN += op.count()
+	b.ops++
 
 	return nil
 }
@@ -1538,22 +1541,23 @@ func (b *Bitmap) Iterator() *Iterator {
 	return itr
 }
 
-// OpN returns the number of write ops the bitmap is aware of in its ops
-// log.
-func (b *Bitmap) OpN() int {
-	return b.opN
+// Ops returns the number of write ops the bitmap is aware of in its ops
+// log, and their total bit count.
+func (b *Bitmap) Ops() (ops int, opN int) {
+	return b.ops, b.opN
 }
 
-// SetOpN lets us reset the operation count in the weird case where we know
+// SetOps lets us reset the operation count in the weird case where we know
 // we've changed an underlying file, without actually refreshing the bitmap.
-func (b *Bitmap) SetOpN(int) {
-	b.opN = 0
+func (b *Bitmap) SetOps(ops int, opN int) {
+	b.ops, b.opN = ops, opN
 }
 
 // Info returns stats for the bitmap.
 func (b *Bitmap) Info() bitmapInfo {
 	info := bitmapInfo{
 		OpN:        b.opN,
+		Ops:        b.ops,
 		Containers: make([]containerInfo, 0, b.Containers.Size()),
 	}
 
@@ -1622,6 +1626,7 @@ func (b *Bitmap) Flip(start, end uint64) *Bitmap {
 // bitmapInfo represents a point-in-time snapshot of bitmap stats.
 type bitmapInfo struct {
 	OpN        int
+	Ops        int
 	Containers []containerInfo
 }
 
diff --git a/view.go b/view.go
index 620bbcee9..4bde632f9 100644
--- a/view.go
+++ b/view.go
@@ -52,10 +52,11 @@ type view struct {
 	// Fragments by shard.
 	fragments map[uint64]*fragment
 
-	broadcaster  broadcaster
-	stats        stats.StatsClient
-	rowAttrStore AttrStore
-	logger       logger.Logger
+	broadcaster   broadcaster
+	stats         stats.StatsClient
+	rowAttrStore  AttrStore
+	logger        logger.Logger
+	snapshotQueue chan *fragment
 }
 
 // newView returns a new instance of View.
@@ -268,6 +269,7 @@ func (v *view) newFragment(path string, shard uint64) *fragment {
 	frag.CacheSize = v.cacheSize
 	frag.Logger = v.logger
 	frag.stats = v.stats
+	frag.snapshotQueue = v.snapshotQueue
 	if v.fieldType == FieldTypeMutex {
 		frag.mutexVector = newRowsVector(frag)
 	} else if v.fieldType == FieldTypeBool {

From 67830b74cf82e1e0393ae0d2242cc25ce99bf755 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Tue, 25 Jun 2019 16:57:08 -0500
Subject: [PATCH 07/13] allow importRoaring to work with official format
 roaring

I didn't think of this, because we don't use it much in the
client. This is a bit hairy because really official roaring
is two fairly different formats, one with runs and one without.
---
 roaring/roaring.go | 157 +++++++++++++++++++++++++++++++++++++++------
 1 file changed, 139 insertions(+), 18 deletions(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index 174ef39a4..0d61020e8 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -1124,7 +1124,11 @@ func (b *Bitmap) writeToUnoptimized(w io.Writer) (n int64, err error) {
 // roaringIterator represents something which can iterate through a roaring
 // bitmap and yield information about containers, including type, size, and
 // the location of their data structures.
-type roaringIterator struct {
+type roaringIterator interface {
+	Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error)
+}
+
+type baseRoaringIterator struct {
 	data              []byte
 	keys              int64
 	headers           []byte
@@ -1139,20 +1143,58 @@ type roaringIterator struct {
 	lastErr           error
 }
 
-func newRoaringIterator(data []byte) (*roaringIterator, error) {
-	if len(data) < headerBaseSize {
-		return nil, errors.New("invalid data: not long enough to be a roaring header")
+type pilosaRoaringIterator struct {
+	baseRoaringIterator
+}
+
+type officialRoaringIterator struct {
+	baseRoaringIterator
+	containerTyper func(index uint, card int) byte
+	haveRuns       bool
+}
+
+func newOfficialRoaringIterator(data []byte) (*officialRoaringIterator, error) {
+	r := &officialRoaringIterator{}
+	r.data = data
+
+	// share code with the existing unmarshal code
+	var offsetOffset, headerOffset int
+	var err error
+	var keys uint32
+
+	// we ignore the flags, since we don't have to process them for anything.
+	keys, r.containerTyper, headerOffset, offsetOffset, _, r.haveRuns, err = readOfficialHeader(data)
+	if err != nil {
+		return nil, fmt.Errorf("reading official header: %v", err)
 	}
-	// Verify the first two bytes are a valid MagicNumber, and second two bytes match current storageVersion.
-	fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2]))
+	if keys == 0 {
+		// not an error, exactly. it's valid and well-formed, we just have nothing to do
+		r.Done(io.EOF)
+		return r, nil
+	}
+	r.keys = int64(keys)
+	r.headers = data[headerOffset:offsetOffset]
+	// note: offsets are only actually used with the no-run headers.
+	if r.haveRuns {
+		// start out pointed at where the offsets would have been.
+		r.currentDataOffset = uint32(offsetOffset)
+	} else {
+		r.offsets = data[offsetOffset : offsetOffset+int(r.keys*4)]
+	}
+	// set key to -1; user should call Next first.
+	r.currentIdx = -1
+	r.currentKey = ^uint64(0)
+	r.lastErr = errors.New("tried to read iterator without calling Next first")
+	return r, nil
+}
+
+func newPilosaRoaringIterator(data []byte) (*pilosaRoaringIterator, error) {
 	fileVersion := uint32(data[2])
-	if fileMagic != MagicNumber {
-		return nil, fmt.Errorf("invalid roaring file, magic number %v is incorrect", fileMagic)
-	}
 	if fileVersion != storageVersion {
 		return nil, fmt.Errorf("wrong roaring version, file is v%d, server requires v%d", fileVersion, storageVersion)
 	}
-	r := &roaringIterator{data: data}
+	r := &pilosaRoaringIterator{}
+	r.data = data
 	// Read key count in bytes sizeof(cookie)+sizeof(flag):(sizeof(cookie)+sizeof(uint32)).
 	r.keys = int64(binary.LittleEndian.Uint32(data[3+1 : 8]))
 	// it could happen
@@ -1179,8 +1221,23 @@ func newRoaringIterator(data []byte) (*roaringIterator, error) {
 	return r, nil
 }
 
+func newRoaringIterator(data []byte) (roaringIterator, error) {
+	if len(data) < headerBaseSize {
+		return nil, errors.New("invalid data: not long enough to be a roaring header")
+	}
+	// Verify the first two bytes are a valid MagicNumber, and second two bytes match current storageVersion.
+	fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2]))
+	switch fileMagic {
+	case serialCookie, serialCookieNoRunContainer:
+		return newOfficialRoaringIterator(data)
+	case MagicNumber:
+		return newPilosaRoaringIterator(data)
+	}
+	return nil, fmt.Errorf("unknown roaring magic number %d", fileMagic)
+}
+
 // Done marks the iterator as complete, recording err as the reason why
-func (r *roaringIterator) Done(err error) {
+func (r *baseRoaringIterator) Done(err error) {
 	r.lastErr = err
 	r.currentKey = ^uint64(0)
 	r.currentType = 0
@@ -1190,7 +1247,7 @@ func (r *roaringIterator) Done(err error) {
 	r.currentDataOffset = 0
 }
 
-func (r *roaringIterator) Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
+func (r *pilosaRoaringIterator) Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
 	if r.currentIdx >= r.keys {
 		// we're already done
 		return r.Current()
@@ -1206,8 +1263,11 @@ func (r *roaringIterator) Next() (key uint64, cType byte, n int, length int, poi
 	r.currentType = byte(binary.LittleEndian.Uint16(header[8:10]))
 	r.currentN = int(binary.LittleEndian.Uint16(header[10:12])) + 1
 	r.currentDataOffset = binary.LittleEndian.Uint32(r.offsets[r.currentIdx*4:])
+
 	// a run container keeps its data after an initial 2 byte length header
+	var runCount uint16
 	if r.currentType == containerRun {
+		runCount = binary.LittleEndian.Uint16(r.data[r.currentDataOffset : r.currentDataOffset+runCountHeaderSize])
 		r.currentDataOffset += 2
 	}
 	if r.currentDataOffset > uint32(len(r.data)) || r.currentDataOffset < headerBaseSize {
@@ -1225,7 +1285,7 @@ func (r *roaringIterator) Next() (key uint64, cType byte, n int, length int, poi
 		r.currentLen = 1024
 		size = 8192
 	case containerRun:
-		r.currentLen = int(*((*uint16)(unsafe.Pointer(&r.data[r.currentDataOffset-2]))))
+		r.currentLen = int(runCount)
 		size = r.currentLen * 4
 	}
 	if int64(r.currentDataOffset)+int64(size) > int64(len(r.data)) {
@@ -1237,7 +1297,70 @@ func (r *roaringIterator) Next() (key uint64, cType byte, n int, length int, poi
 	return r.Current()
 }
 
-func (r *roaringIterator) Current() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
+func (r *officialRoaringIterator) Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
+	if r.currentIdx >= r.keys {
+		// we're already done
+		return r.Current()
+	}
+	r.currentIdx++
+	if r.currentIdx == r.keys {
+		// this is the last key. transition state to the finalized state
+		r.Done(io.EOF)
+		return r.Current()
+	}
+	header := r.headers[r.currentIdx*4:]
+	r.currentKey = uint64(binary.LittleEndian.Uint16(header[0:2]))
+	r.currentN = int(binary.LittleEndian.Uint16(header[2:4])) + 1
+	r.currentType = r.containerTyper(uint(r.currentIdx), r.currentN)
+	// with runs, we can't actually look up offsets; the format just stores
+	// things sequentially. so we have to actually track the offset in that case.
+	if !r.haveRuns {
+		r.currentDataOffset = binary.LittleEndian.Uint32(r.offsets[r.currentIdx*4:])
+	}
+	// a run container keeps its data after an initial 2 byte length header
+	var runCount uint16
+	if r.currentType == containerRun {
+		runCount = binary.LittleEndian.Uint16(r.data[r.currentDataOffset : r.currentDataOffset+runCountHeaderSize])
+		r.currentDataOffset += 2
+	}
+	if r.currentDataOffset > uint32(len(r.data)) || r.currentDataOffset < headerBaseSize {
+		r.Done(fmt.Errorf("container %d/%d, key %d, had offset %d, maximum %d",
+			r.currentIdx, r.keys, r.currentKey, r.currentDataOffset, len(r.data)))
+		return r.Current()
+	}
+	r.currentPointer = (*uint16)(unsafe.Pointer(&r.data[r.currentDataOffset]))
+	var size int
+	switch r.currentType {
+	case containerArray:
+		r.currentLen = r.currentN
+		size = r.currentLen * 2
+	case containerBitmap:
+		r.currentLen = 1024
+		size = 8192
+	case containerRun:
+		// official format stores runs as start/len, we want to convert, but since
+		// they might be mmapped, we can't write to that memory
+		newRuns := make([]interval16, runCount)
+		oldRuns := (*[65536]interval16)(unsafe.Pointer(r.currentPointer))[:runCount:runCount]
+		copy(newRuns, oldRuns)
+		for i := range newRuns {
+			newRuns[i].last += newRuns[i].start
+		}
+		r.currentPointer = (*uint16)(unsafe.Pointer(&newRuns[0]))
+		r.currentLen = int(runCount)
+		size = r.currentLen * 4
+	}
+	if int64(r.currentDataOffset)+int64(size) > int64(len(r.data)) {
+		r.Done(fmt.Errorf("container %d/%d, key %d, had offset %d+%d size, maximum %d",
+			r.currentIdx, r.keys, r.currentKey, r.currentDataOffset, size, len(r.data)))
+		return r.Current()
+	}
+	r.currentDataOffset += uint32(size)
+	r.lastErr = nil
+	return r.Current()
+}
+
+func (r *baseRoaringIterator) Current() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
 	return r.currentKey, r.currentType, r.currentN, r.currentLen, r.currentPointer, r.lastErr
 }
 
@@ -1250,13 +1373,11 @@ func (r *roaringIterator) Current() (key uint64, cType byte, n int, length int,
 // Regardless, after this function runs, no containers have
 // mapped storage which does not refer to data; either they got mapped
 // to the new storage, or storage was allocated for them.
-//
-// Data should be in the Pilosa roaring format.
 func (b *Bitmap) RemapRoaringStorage(data []byte) (mappedAny bool, returnErr error) {
 	if b.Containers == nil {
 		return false, nil
 	}
-	var itr *roaringIterator
+	var itr roaringIterator
 	var err error
 	var itrKey uint64
 	var itrCType byte
@@ -1331,7 +1452,7 @@ func (b *Bitmap) ImportRoaringBits(data []byte, clear bool, log bool, rowSize ui
 	if data == nil {
 		return 0, nil, errors.New("no roaring bitmap provided")
 	}
-	var itr *roaringIterator
+	var itr roaringIterator
 	var itrKey uint64
 	var itrCType byte
 	var itrN int

From cb50a5a48b6767a7268173f7992f103a4ec3dd89 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Tue, 25 Jun 2019 17:19:07 -0500
Subject: [PATCH 08/13] revert BSIv2 change impact on Official Roaring

The Pilosa roaring format uses two bytes of its
header, next to the magic number, for a version. The
official roaring format uses them for a container
count, if and only if it's the version of the format
that uses run-length containers.

But if it is, it really does need those bits. Also,
since we never use the official format in our internals
or snapshots, we don't have any reason to support
reading flag bits in it, since the flag bits are used
only for internals of fragments and snapshots. So
we revert the change to support flags with official
roaring bitmaps.

A couple of the fuzz tests happened to rely on this,
and we may find more issues with more fuzzing.
---
 roaring/fuzz_test.go |  6 +++---
 roaring/roaring.go   | 27 +++++++++++++++------------
 2 files changed, 18 insertions(+), 15 deletions(-)

diff --git a/roaring/fuzz_test.go b/roaring/fuzz_test.go
index da74202f0..fe8fb6cee 100644
--- a/roaring/fuzz_test.go
+++ b/roaring/fuzz_test.go
@@ -24,7 +24,7 @@ func TestUnmarshalBinary(t *testing.T) {
 		expected string
 	}{
 		{ // Checks for the zero containers situation
-			cr:       []byte(":0\x000\x01\x00\x00\x000000"), //":000000"
+			cr:       []byte(":0\x00\x00\x01\x00\x00\x000000"), //":000000"
 			expected: "reading roaring header: malformed bitmap, key-cardinality slice overruns buffer at 12",
 		},
 		{ // Checks for int overflow
@@ -58,11 +58,11 @@ func TestUnmarshalBinary(t *testing.T) {
 			expected: "unmarshaling as pilosa roaring: malformed bitmap, key-cardinality not provided for 0 containers",
 		},
 		{ // Checks for incomplete offset in readWithRuns
-			cr:       []byte(";0\x000\v00000"), //";0000000"
+			cr:       []byte(";0\x00\x00\v00000"), //";0000000"
 			expected: "reading offsets from official roaring format: offset incomplete: len=10",
 		},
 		{ // Checks for incomplete offset in readOffsets
-			cr: []byte(":0\x000\x03\x00\x00\x00000000000000" +
+			cr: []byte(":0\x00\x00\x03\x00\x00\x00000000000000" +
 				"\x00"), //:0000000000000
 			expected: "reading offsets from official roaring format: offset incomplete: len=1",
 		},
diff --git a/roaring/roaring.go b/roaring/roaring.go
index 0d61020e8..9f464d5d3 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -1128,6 +1128,8 @@ type roaringIterator interface {
 	Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error)
 }
 
+// baseRoaringIterator holds values used by both Pilosa and official Roaring
+// iterators.
 type baseRoaringIterator struct {
 	data              []byte
 	keys              int64
@@ -1163,7 +1165,7 @@ func newOfficialRoaringIterator(data []byte) (*officialRoaringIterator, error) {
 	var keys uint32
 
 	// we ignore the flags, since we don't have to process them for anything.
-	keys, r.containerTyper, headerOffset, offsetOffset, _, r.haveRuns, err = readOfficialHeader(data)
+	keys, r.containerTyper, headerOffset, offsetOffset, r.haveRuns, err = readOfficialHeader(data)
 	if err != nil {
 		return nil, fmt.Errorf("reading official header: %v", err)
 	}
@@ -5054,11 +5056,11 @@ const (
 	serialCookie               = 12347 // runs, arrays, and bitmaps
 )
 
-func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint, card int) byte, header, pos int, flags byte, haveRuns bool, err error) {
+func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint, card int) byte, header, pos int, haveRuns bool, err error) {
 	statsHit("readOfficialHeader")
 	if len(buf) < 8 {
 		err = fmt.Errorf("buffer too small, expecting at least 8 bytes, was %d", len(buf))
-		return size, containerTyper, header, pos, flags, haveRuns, err
+		return size, containerTyper, header, pos, haveRuns, err
 	}
 	cf := func(index uint, card int) (newType byte) {
 		newType = containerBitmap
@@ -5068,8 +5070,7 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
 		return newType
 	}
 	containerTyper = cf
-	cookie := binary.LittleEndian.Uint32(buf) & 0xFFFFFF
-	flags = buf[3]
+	cookie := binary.LittleEndian.Uint32(buf)
 	pos += 4
 
 	// cookie header
@@ -5084,7 +5085,7 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
 		isRunBitmapSize := (int(size) + 7) / 8
 		if pos+isRunBitmapSize > len(buf) {
 			err = fmt.Errorf("malformed bitmap, is-run bitmap overruns buffer at %d", pos+isRunBitmapSize)
-			return size, containerTyper, header, pos, flags, haveRuns, err
+			return size, containerTyper, header, pos, haveRuns, err
 		}
 
 		isRunBitmap := buf[pos : pos+isRunBitmapSize]
@@ -5097,22 +5098,22 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
 		}
 	} else {
 		err = fmt.Errorf("did not find expected serialCookie in header")
-		return size, containerTyper, header, pos, flags, haveRuns, err
+		return size, containerTyper, header, pos, haveRuns, err
 	}
 
 	header = pos
 	if size > (1 << 16) {
 		err = fmt.Errorf("it is logically impossible to have more than (1<<16) containers")
-		return size, containerTyper, header, pos, flags, haveRuns, err
+		return size, containerTyper, header, pos, haveRuns, err
 	}
 
 	// descriptive header
 	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
+		return size, containerTyper, header, pos, haveRuns, err
 	}
 	pos += 2 * 2 * int(size) // moving pos past keycount
-	return size, containerTyper, header, pos, flags, haveRuns, err
+	return size, containerTyper, header, pos, haveRuns, err
 }
 
 // UnmarshalBinary decodes b from a binary-encoded byte slice. data can be in
@@ -5129,11 +5130,13 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error {
 		return errors.Wrap(b.unmarshalPilosaRoaring(data), "unmarshaling as pilosa roaring")
 	}
 
-	keyN, containerTyper, header, pos, flags, haveRuns, err := readOfficialHeader(data)
+	keyN, containerTyper, header, pos, haveRuns, err := readOfficialHeader(data)
 	if err != nil {
 		return errors.Wrap(err, "reading roaring header")
 	}
-	b.Flags = flags
+	// Only the Pilosa roaring format has flags. The official Roaring format
+	// hasn't got space in its header for flags.
+	b.Flags = 0
 
 	b.Containers.ResetN(int(keyN))
 	// Descriptive header section: Read container keys and cardinalities.

From 5e3d01febe50e266c1dceca1a99e651460b51c32 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Wed, 26 Jun 2019 10:57:31 -0500
Subject: [PATCH 09/13] lock fragment to compute rows

If you don't hold the fragment lock when computing rows, it's
pretty reasonable for other stuff to be able to modify it -- which
could invalidate or race the enumeration.

Some calls to f.rows were being made with the lock held, others
weren't, so we introduce `f.unprotectedRows` which has the obvious
semantics. (Without which this looked great except that several
of the tests deadlocked.)
---
 fragment.go | 17 +++++++++++++----
 1 file changed, 13 insertions(+), 4 deletions(-)

diff --git a/fragment.go b/fragment.go
index bbf7d0afe..f9b7da24e 100644
--- a/fragment.go
+++ b/fragment.go
@@ -2669,6 +2669,13 @@ func filterWithRows(rows []uint64) rowFilter {
 // this container have been processed. The rows accumulated up to this point
 // (including this row if all filters passed) will be returned.
 func (f *fragment) rows(start uint64, filters ...rowFilter) []uint64 {
+	f.mu.Lock()
+	defer f.mu.Unlock()
+	return f.unprotectedRows(start, filters...)
+}
+
+// unprotectedRows calls rows without grabbing the mutex.
+func (f *fragment) unprotectedRows(start uint64, filters ...rowFilter) []uint64 {
 	startKey := rowToKey(start)
 	i, _ := f.storage.Containers.Iterator(startKey)
 	rows := make([]uint64, 0)
@@ -3096,9 +3103,10 @@ func newRowsVector(f *fragment) *rowsVector {
 
 // Get returns the rowID associated to the given colID.
 // Additionally, it returns true if a value was found,
-// otherwise it returns false.
+// otherwise it returns false. Ensure that you already
+// have the mutex before calling this.
 func (v *rowsVector) Get(colID uint64) (uint64, bool, error) {
-	rows := v.f.rows(0, filterColumn(colID))
+	rows := v.f.unprotectedRows(0, filterColumn(colID))
 	if len(rows) > 1 {
 		return 0, false, errors.New("found multiple row values for column")
 	} else if len(rows) == 1 {
@@ -3129,9 +3137,10 @@ func newBoolVector(f *fragment) *boolVector {
 
 // Get returns the rowID associated to the given colID.
 // Additionally, it returns true if a value was found,
-// otherwise it returns false.
+// otherwise it returns false. Ensure that you already
+// have the fragment mutex before calling this.
 func (v *boolVector) Get(colID uint64) (uint64, bool, error) {
-	rows := v.f.rows(0, filterColumn(colID))
+	rows := v.f.unprotectedRows(0, filterColumn(colID))
 	if len(rows) > 1 {
 		return 0, false, errors.New("found multiple row values for column")
 	} else if len(rows) == 1 {

From 17eb13702e9c48d72b1378ab317a1339d3d557e7 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Wed, 26 Jun 2019 11:10:34 -0500
Subject: [PATCH 10/13] address lint concerns

Addressing various lint.

incrementOpN no longer returns errors, because it no longer waits for
the snapshot, so checking those errors is unnecessary.

Several fields in a common embedded structure were "unused" according
to a naive checker.

Other tiny style things, and one actual unchecked error. Yay linters!
---
 fragment.go               | 29 ++++++++++++++---------------
 fragment_internal_test.go |  5 ++++-
 roaring/roaring.go        | 14 ++++++++++++--
 3 files changed, 30 insertions(+), 18 deletions(-)

diff --git a/fragment.go b/fragment.go
index f9b7da24e..e88096a8c 100644
--- a/fragment.go
+++ b/fragment.go
@@ -219,7 +219,7 @@ func (f *fragment) enqueueSnapshot() {
 			// wait forever, but notice that we're waiting
 			f.snapshotQueue <- f
 			f.snapshotDelays++
-			f.snapshotDelayTime += time.Now().Sub(before)
+			f.snapshotDelayTime += time.Since(before)
 			if f.snapshotDelays >= 10 {
 				f.Logger.Printf("snapshotting %s: last ten delays took %v", f.path, f.snapshotDelayTime)
 				f.snapshotDelays = 0
@@ -231,7 +231,10 @@ func (f *fragment) enqueueSnapshot() {
 	} else {
 		// in testing, for instance, there may be no holder, thus no one
 		// to handle these snapshots.
-		f.snapshot()
+		err := f.snapshot()
+		if err != nil {
+			f.Logger.Printf("snapshot failed: %v", err)
+		}
 		f.snapshotting = false
 		f.snapshotCond.Broadcast()
 	}
@@ -517,9 +520,9 @@ func (f *fragment) awaitSnapshot() {
 	}
 }
 
-// protectedAwaitSnapshot assumes you already hold the lock, and waits for
+// unprotectedAwaitSnapshot assumes you already hold the lock, and waits for
 // the snapshot fairy to come along.
-func (f *fragment) protectedAwaitSnapshot() {
+func (f *fragment) unprotectedAwaitSnapshot() {
 	for f.snapshotting {
 		f.snapshotCond.Wait()
 	}
@@ -698,9 +701,7 @@ func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err
 	delete(f.checksums, int(rowID/HashBlockSize))
 
 	// Increment number of operations until snapshot is required.
-	if err := f.incrementOpN(1); err != nil {
-		return false, errors.Wrap(err, "incrementing")
-	}
+	f.incrementOpN(1)
 
 	// If we're using a cache, update it. Otherwise skip the
 	// possibly-expensive count operation.
@@ -762,9 +763,7 @@ func (f *fragment) unprotectedClearBit(rowID, columnID uint64) (changed bool, er
 	delete(f.checksums, int(rowID/HashBlockSize))
 
 	// Increment number of operations until snapshot is required.
-	if err := f.incrementOpN(1); err != nil {
-		return false, errors.Wrap(err, "incrementing")
-	}
+	f.incrementOpN(1)
 
 	// If we're using a cache, update it. Otherwise skip the
 	// possibly-expensive count operation.
@@ -2230,11 +2229,12 @@ func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint
 	}
 	// We don't actually care, except we want our stats to be accurate.
 	f.incrementOpN(totalChanges)
+
 	// in theory, this should probably have happened anyway, but if enough
 	// of the bits matched existing bits, we'll be under our opN estimate, and
 	// we want to ensure that the snapshot happens.
 	f.enqueueSnapshot()
-	f.protectedAwaitSnapshot()
+	f.unprotectedAwaitSnapshot()
 
 	return nil
 }
@@ -2278,21 +2278,20 @@ func (f *fragment) importRoaring(ctx context.Context, data []byte, clear bool) e
 	span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.incrementOpN")
 	f.incrementOpN(changed)
 	span.Finish()
-	return err
+	return nil
 }
 
 // incrementOpN increase the operation count by one.
 // If the count exceeds the maximum allowed then a snapshot is performed.
-func (f *fragment) incrementOpN(changed int) error {
+func (f *fragment) incrementOpN(changed int) {
 	if changed <= 0 {
-		return nil
+		return
 	}
 	f.opN += changed
 	f.ops++
 	if f.opN > f.MaxOpN {
 		f.enqueueSnapshot()
 	}
-	return nil
 }
 
 // Snapshot writes the storage bitmap to disk and reopens it. This may
diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index 1f2f5c807..333f15b42 100644
--- a/fragment_internal_test.go
+++ b/fragment_internal_test.go
@@ -2499,10 +2499,13 @@ func (f *fragment) sanityCheck(t testing.TB) {
 	}
 	defer file.Close()
 	data, err := ioutil.ReadAll(file)
-	err = newBM.UnmarshalBinary(data)
 	if err != nil {
 		t.Fatalf("sanityCheck couldn't read fragment %s: %v", f.path, err)
 	}
+	err = newBM.UnmarshalBinary(data)
+	if err != nil {
+		t.Fatalf("sanityCheck couldn't unmarshal fragment %s: %v", f.path, err)
+	}
 	if equal, reason := newBM.BitwiseEqual(f.storage); !equal {
 		t.Fatalf("fragment %s: unmarshalled bitmap different: %v", f.path, reason)
 	}
diff --git a/roaring/roaring.go b/roaring/roaring.go
index 9f464d5d3..4948f5678 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -1145,6 +1145,17 @@ type baseRoaringIterator struct {
 	lastErr           error
 }
 
+// okay, then
+func (b *baseRoaringIterator) SilenceLint() {
+	// these are actually used in pilosaRoaringIterator or officialRoaringIterator
+	// but structcheck doesn't know that
+	_ = b.data
+	_ = b.keys
+	_ = b.offsets
+	_ = b.headers
+	_ = b.currentIdx
+}
+
 type pilosaRoaringIterator struct {
 	baseRoaringIterator
 }
@@ -1416,8 +1427,7 @@ func (b *Bitmap) RemapRoaringStorage(data []byte) (mappedAny bool, returnErr err
 				if oldC.frozen() {
 					// we don't use Clone, because that would copy the
 					// storage, and we don't need that.
-					var halfCopy Container
-					halfCopy = *oldC
+					halfCopy := *oldC
 					halfCopy.flags &^= flagFrozen
 					newC = &halfCopy
 				} else {

From b74956e48e8cf93e14a6fce2207ca77989c3c6f6 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Wed, 26 Jun 2019 14:51:25 -0500
Subject: [PATCH 11/13] drop no-longer-used timeout case

---
 fragment.go | 2 --
 1 file changed, 2 deletions(-)

diff --git a/fragment.go b/fragment.go
index e88096a8c..ff9d4fe40 100644
--- a/fragment.go
+++ b/fragment.go
@@ -225,8 +225,6 @@ func (f *fragment) enqueueSnapshot() {
 				f.snapshotDelays = 0
 				f.snapshotDelayTime = 0
 			}
-		case <-time.After(5 * time.Second):
-			f.Logger.Printf("snapshot for %s: timed out\n", f.path)
 		}
 	} else {
 		// in testing, for instance, there may be no holder, thus no one

From e1fbed51b2f2b77242246224809e0a9c4d36d596 Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Wed, 26 Jun 2019 14:51:45 -0500
Subject: [PATCH 12/13] use symbolic names for op types, add checks for invalid
 types

---
 roaring/roaring.go | 44 ++++++++++++++++++++++++++++----------------
 1 file changed, 28 insertions(+), 16 deletions(-)

diff --git a/roaring/roaring.go b/roaring/roaring.go
index 4948f5678..92ca8bb18 100644
--- a/roaring/roaring.go
+++ b/roaring/roaring.go
@@ -4463,18 +4463,20 @@ func (op *op) WriteTo(w io.Writer) (n int64, err error) {
 	// Write type and value.
 	buf[0] = byte(op.typ)
 	switch op.typ {
-	case 0, 1:
+	case opTypeAdd, opTypeRemove:
 		binary.LittleEndian.PutUint64(buf[1:9], op.value)
-	case 2, 3:
+	case opTypeAddBatch, opTypeRemoveBatch:
 		binary.LittleEndian.PutUint64(buf[1:9], uint64(len(op.values)))
 		p := 13 // start of values (skip 4 for checksum)
 		for _, v := range op.values {
 			binary.LittleEndian.PutUint64(buf[p:p+8], v)
 			p += 8
 		}
-	case 4, 5:
+	case opTypeAddRoaring, opTypeRemoveRoaring:
 		binary.LittleEndian.PutUint64(buf[1:9], uint64(len(op.roaring)))
 		binary.LittleEndian.PutUint32(buf[13:17], uint32(op.opN))
+	default:
+		return 0, fmt.Errorf("can't marshal unknown op type %d", op.typ)
 	}
 
 	// Add checksum at the end.
@@ -4520,9 +4522,9 @@ func (op *op) UnmarshalBinary(data []byte) error {
 	_, _ = h.Write(data[0:9])
 
 	switch op.typ {
-	case 0, 1:
+	case opTypeAdd, opTypeRemove:
 		// nothing to do, just being not-default
-	case 2, 3:
+	case opTypeAddBatch, opTypeRemoveBatch:
 		// 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 op.value > maxBatchSize {
@@ -4538,7 +4540,7 @@ func (op *op) UnmarshalBinary(data []byte) error {
 			op.values[i] = binary.LittleEndian.Uint64(data[start : start+8])
 		}
 		op.value = 0
-	case 4, 5:
+	case opTypeAddRoaring, opTypeRemoveRoaring:
 		if len(data) < int(13+4+op.value) {
 			return fmt.Errorf("op data truncated - expected %d, got %d", 13+op.value, len(data))
 		}
@@ -4558,28 +4560,38 @@ func (op *op) UnmarshalBinary(data []byte) error {
 
 // size returns the encoded size of the op, in bytes.
 func (op *op) size() int {
-	if op.typ == opTypeAdd || op.typ == opTypeRemove {
+	switch op.typ {
+	case opTypeAdd, opTypeRemove:
 		return 1 + 8 + 4
-	}
-	if op.typ == opTypeAddBatch || op.typ == opTypeRemoveBatch {
+	case opTypeAddBatch, opTypeRemoveBatch:
 		return 1 + 8 + 4 + len(op.values)*8
+
+	case opTypeAddRoaring, opTypeRemoveRoaring:
+		return 1 + 8 + 4 + 4 + len(op.roaring)
 	}
-	// else it's presumably roaring?
-	return 1 + 8 + 4 + 4 + len(op.roaring)
+	if roaringParanoia {
+		panic(fmt.Sprintf("op size() called on unknown op type %d", op.typ))
+	}
+	return 0
 }
 
 // size returns the size needed to encode the op, in bytes. for
 // roaring ops, this does not include the roaring data, which is
 // already encoded.
 func (op *op) encodeSize() int {
-	if op.typ == opTypeAdd || op.typ == opTypeRemove {
+	switch op.typ {
+	case opTypeAdd, opTypeRemove:
 		return 1 + 8 + 4
-	}
-	if op.typ == opTypeAddBatch || op.typ == opTypeRemoveBatch {
+	case opTypeAddBatch, opTypeRemoveBatch:
 		return 1 + 8 + 4 + len(op.values)*8
+
+	case opTypeAddRoaring, opTypeRemoveRoaring:
+		return 1 + 8 + 4 + 4
 	}
-	// else it's presumably roaring?
-	return 1 + 8 + 4 + 4
+	if roaringParanoia {
+		panic(fmt.Sprintf("op encodeSize() called on unknown op type %d", op.typ))
+	}
+	return 0
 }
 
 // count returns the number of bits the operation mutates.

From 0960d66c94adb7aac2ba08479752714c0f15b99c Mon Sep 17 00:00:00 2001
From: Seebs <seebs@seebs.net>
Date: Wed, 26 Jun 2019 15:20:10 -0500
Subject: [PATCH 13/13] update diagnostic message, use read locks for read

Annoyingly, this is actually the only place we can make
a read-only lock, because the row() call might write to
the row cache, so it needs the write lock. We might be
able to fix that later, though.
---
 fragment.go | 6 +++---
 1 file changed, 3 insertions(+), 3 deletions(-)

diff --git a/fragment.go b/fragment.go
index ff9d4fe40..94b86ac8a 100644
--- a/fragment.go
+++ b/fragment.go
@@ -221,7 +221,7 @@ func (f *fragment) enqueueSnapshot() {
 			f.snapshotDelays++
 			f.snapshotDelayTime += time.Since(before)
 			if f.snapshotDelays >= 10 {
-				f.Logger.Printf("snapshotting %s: last ten delays took %v", f.path, f.snapshotDelayTime)
+				f.Logger.Printf("snapshotting %s: last ten enqueue delays took %v", f.path, f.snapshotDelayTime)
 				f.snapshotDelays = 0
 				f.snapshotDelayTime = 0
 			}
@@ -2666,8 +2666,8 @@ func filterWithRows(rows []uint64) rowFilter {
 // this container have been processed. The rows accumulated up to this point
 // (including this row if all filters passed) will be returned.
 func (f *fragment) rows(start uint64, filters ...rowFilter) []uint64 {
-	f.mu.Lock()
-	defer f.mu.Unlock()
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	return f.unprotectedRows(start, filters...)
 }