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 f81a0508a..94b86ac8a 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,69 @@ 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.Since(before)
+			if f.snapshotDelays >= 10 {
+				f.Logger.Printf("snapshotting %s: last ten enqueue delays took %v", f.path, f.snapshotDelayTime)
+				f.snapshotDelays = 0
+				f.snapshotDelayTime = 0
+			}
+		}
+	} else {
+		// in testing, for instance, there may be no holder, thus no one
+		// to handle these snapshots.
+		err := f.snapshot()
+		if err != nil {
+			f.Logger.Printf("snapshot failed: %v", err)
+		}
+		f.snapshotting = false
+		f.snapshotCond.Broadcast()
+	}
+}
+
 // Open opens the underlying storage.
 func (f *fragment) Open() error {
 	f.mu.Lock()
@@ -177,7 +246,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 +284,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 +337,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 +362,97 @@ 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 {
+			_, 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.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.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).
+		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.
@@ -332,9 +501,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()
+	}
+}
+
+// unprotectedAwaitSnapshot assumes you already hold the lock, and waits for
+// the snapshot fairy to come along.
+func (f *fragment) unprotectedAwaitSnapshot() {
+	for f.snapshotting {
+		f.snapshotCond.Wait()
+	}
+}
+
 func (f *fragment) close() error {
 	// Flush cache if closing gracefully.
 	if err := f.flushCache(); err != nil {
@@ -343,7 +534,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 +565,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)
 		}
@@ -502,9 +699,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 {
-		return false, errors.Wrap(err, "incrementing")
-	}
+	f.incrementOpN(1)
 
 	// If we're using a cache, update it. Otherwise skip the
 	// possibly-expensive count operation.
@@ -566,9 +761,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 {
-		return false, errors.Wrap(err, "incrementing")
-	}
+	f.incrementOpN(1)
 
 	// If we're using a cache, update it. Otherwise skip the
 	// possibly-expensive count operation.
@@ -629,14 +822,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
@@ -677,11 +872,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)
 
@@ -845,7 +1039,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 {
@@ -862,13 +1056,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++
 			}
 		}
 	}
@@ -880,13 +1074,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++
 		}
 	}
 
@@ -897,13 +1091,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++
 		}
 	}
 
@@ -1855,19 +2049,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 {
@@ -1877,7 +2064,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 {
@@ -1887,7 +2074,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.
@@ -1895,19 +2082,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
 }
 
@@ -1994,8 +2180,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)
@@ -2020,138 +2206,126 @@ 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 {
-		_ = f.closeStorage()
-		_ = f.openStorage()
+		_ = f.closeStorage(true)
+		_ = f.openStorage(true)
 		return err
 	}
+	// We don't actually care, except we want our stats to be accurate.
+	f.incrementOpN(totalChanges)
 
-	err := f.snapshot()
-	return errors.Wrap(err, "snapshotting")
+	// 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.unprotectedAwaitSnapshot()
+
+	return nil
 }
 
 // importRoaring imports from the official roaring data format defined at
 // 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
+	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() error {
-	f.opN++
-	if f.opN <= f.MaxOpN {
-		return nil
+func (f *fragment) incrementOpN(changed int) {
+	if changed <= 0 {
+		return
 	}
-
-	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
 }
@@ -2159,7 +2333,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)
@@ -2183,7 +2356,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)
 	}
 
@@ -2192,8 +2365,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)
 	}
 
@@ -2382,7 +2559,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")
 	}
 
@@ -2392,7 +2569,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")
 	}
 
@@ -2489,6 +2666,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.RLock()
+	defer f.mu.RUnlock()
+	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)
@@ -2916,9 +3100,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 {
@@ -2949,9 +3134,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 {
diff --git a/fragment_internal_test.go b/fragment_internal_test.go
index e45d3cce8..333f15b42 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
@@ -889,7 +889,6 @@ func BenchmarkFragment_RepeatedSmallValueImports(b *testing.B) {
 								b.Fatalf("importing values: %v", err)
 							}
 						}
-
 					}
 				})
 			}
@@ -2037,16 +2036,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++ {
@@ -2054,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)
 					}
@@ -2070,63 +2071,30 @@ 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)
+								err := frags[j].importRoaringT(data[j], false)
+								frags[j].awaitSnapshot()
+								return err
 							})
 						}
 						err := eg.Wait()
@@ -2143,9 +2111,60 @@ 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)
+								// 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)
+								}
+							}
+							eg := errgroup.Group{}
+							b.StartTimer()
+							for j := 0; j < concurrency; j++ {
+								j := j
+								eg.Go(func() error {
+									err := frags[j].importRoaringT(updata, false)
+									frags[j].awaitSnapshot()
+									return err
+								})
+							}
+							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,23 +2190,29 @@ 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)
-						err := f.importRoaringT(data, false)
+						f := mustOpenFragment("i", fmt.Sprintf("r%dc%dcache_%s", numRows, numCols, cacheType), viewStandard, 0, cacheType)
+						// 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)
@@ -2400,19 +2425,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
@@ -2462,13 +2491,39 @@ 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)
+	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)
+	}
+}
+
 func (f *fragment) Clean(t testing.TB) {
+	f.awaitSnapshot()
+	f.sanityCheck(t)
 	errc := f.Close()
 	errf := os.Remove(f.path)
 	errp := os.Remove(f.cachePath())
 	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)
@@ -2488,6 +2543,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)
@@ -2520,6 +2579,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)
@@ -3007,6 +3067,9 @@ func TestUnionInPlaceMapped(t *testing.T) {
 
 	f.storage.UnionInPlace(setBM1)
 	countUnion := f.storage.Count()
+	// 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)
@@ -3194,7 +3257,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)
 				}
@@ -3228,7 +3291,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/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/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/fuzz_test.go b/roaring/fuzz_test.go
index e56d3f0ce..fe8fb6cee 100644
--- a/roaring/fuzz_test.go
+++ b/roaring/fuzz_test.go
@@ -24,13 +24,13 @@ 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
 			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" +
@@ -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 f55cbc4a6..92ca8bb18 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.
@@ -147,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.
@@ -997,7 +1000,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 +1121,442 @@ 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 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
+	headers           []byte
+	offsets           []byte
+	currentKey        uint64
+	currentIdx        int64
+	currentType       byte
+	currentN          int
+	currentLen        int
+	currentPointer    *uint16
+	currentDataOffset uint32
+	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
+}
+
+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)
+	}
+	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 fileVersion != storageVersion {
+		return nil, fmt.Errorf("wrong roaring version, file is v%d, server requires v%d", fileVersion, storageVersion)
+	}
+	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
+	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
+}
+
+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 *baseRoaringIterator) 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 *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()
+	}
+	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
+	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:
+		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.lastErr = nil
+	return r.Current()
+}
+
+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
+}
+
+// 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.
+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.
+					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
+// 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 +1583,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(
@@ -1203,6 +1642,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.
@@ -1222,6 +1662,7 @@ func (b *Bitmap) writeOp(op *op) error {
 		return err
 	}
 	b.opN += op.count()
+	b.ops++
 
 	return nil
 }
@@ -1233,10 +1674,23 @@ func (b *Bitmap) Iterator() *Iterator {
 	return itr
 }
 
+// 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
+}
+
+// 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) 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()),
 	}
 
@@ -1305,6 +1759,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
 }
 
@@ -3961,17 +4416,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 +4444,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 +4458,47 @@ 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 opTypeAdd, opTypeRemove:
 		binary.LittleEndian.PutUint64(buf[1:9], op.value)
-	} else {
+	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 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.
 	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,18 +4513,22 @@ 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 opTypeAdd, opTypeRemove:
+		// nothing to do, just being not-default
+	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 {
-			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))
@@ -4053,9 +4540,19 @@ func (op *op) UnmarshalBinary(data []byte) error {
 			op.values[i] = binary.LittleEndian.Uint64(data[start : start+8])
 		}
 		op.value = 0
+	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))
+		}
+		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
+	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: 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
@@ -4063,10 +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
+	case opTypeAddBatch, opTypeRemoveBatch:
+		return 1 + 8 + 4 + len(op.values)*8
+
+	case opTypeAddRoaring, opTypeRemoveRoaring:
+		return 1 + 8 + 4 + 4 + len(op.roaring)
 	}
-	return 1 + 8 + 4 + len(op.values)*8
+	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 {
+	switch op.typ {
+	case opTypeAdd, opTypeRemove:
+		return 1 + 8 + 4
+	case opTypeAddBatch, opTypeRemoveBatch:
+		return 1 + 8 + 4 + len(op.values)*8
+
+	case opTypeAddRoaring, opTypeRemoveRoaring:
+		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.
@@ -4076,6 +4601,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 +4956,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 {
@@ -4482,11 +5078,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
@@ -4496,8 +5092,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
@@ -4512,7 +5107,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]
@@ -4525,22 +5120,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
@@ -4557,13 +5152,15 @@ 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.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
diff --git a/view.go b/view.go
index 85c9ebbba..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 {
@@ -441,11 +443,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")
 		}
 	}