pilosa: write lock the fragment when rowcache used

- only allocate the rowcache if it is in use (avoid allocation per fragment)
 - when the rowcache is use, fragment.go intRowIterator must write lock the
   fragment because the f.rowCache will be updated.
 - eliminate unused bitmapCache interface to keep the linter happy.
 - fixes #1035

cache.go

@@ -607,13 +607,7 @@ func (p uint64Slice) merge(other []uint64) []uint64 {
 	return ret
 }
 
-// bitmapCache provides an interface for caching full bitmaps.
-type bitmapCache interface {
-	Fetch(id uint64) (*Row, bool)
-	Add(id uint64, b *Row)
-}
-
-// simpleCache implements BitmapCache
+// simpleCache implements a bitmap Rowcache.
 // it is meant to be a short-lived cache for cases where writes are continuing to access
 // the same row within a short time frame (i.e. good for write-heavy loads)
 // A read-heavy use case would cause the cache to get bigger, potentially causing the
@@ -628,6 +622,12 @@ func (s *simpleCache) Fetch(id uint64) (*Row, bool) {
 	return m, ok
 }
 
+func newSimpleCache() *simpleCache {
+	return &simpleCache{
+		cache: make(map[uint64]*Row),
+	}
+}
+
 // Add adds the bitmap to the cache, keyed on the id. A nil row means
 // deleting the row from the cache.
 func (s *simpleCache) Add(id uint64, b *Row) {

fragment.go

@@ -144,7 +144,7 @@ type fragment struct {
 	CacheSize uint32
 
 	// Cache containing full rows (not just counts).
-	rowCache bitmapCache
+	rowCache *simpleCache
 
 	// Cached checksums for each block.
 	checksums map[int][]byte
@@ -396,9 +396,12 @@ func (f *fragment) inspectStorage(data []byte, file *os.File, newGen generation,
 // (remapping an existing bitmap to match a new backing store).
 func (f *fragment) openStorage(unmarshalData bool) error {
 
+	useRowCache := f.idx.Txf().UseRowCache()
 	if !f.idx.NeedsSnapshot() {
 		f.gen = &NopGeneration{}
-		f.rowCache = &simpleCache{make(map[uint64]*Row)}
+		if useRowCache {
+			f.rowCache = newSimpleCache()
+		}
 		f.currdata = struct{ from, to uintptr }{}
 		f.prevdata = f.currdata
 		return nil // openStorage becomes a noop under RBF, Badger, etc.
@@ -412,7 +415,9 @@ func (f *fragment) openStorage(unmarshalData bool) error {
 		// unmarshal this data in order to have any.
 		unmarshalData = true
 	}
-	f.rowCache = &simpleCache{make(map[uint64]*Row)}
+	if useRowCache {
+		f.rowCache = newSimpleCache()
+	}
 	var storageOp func([]byte, *os.File, generation, bool) (bool, error)
 	if f.holder.Opts.Inspect {
 		// note that this will unmarshal even if we already have
@@ -567,6 +572,9 @@ func (f *fragment) unprotectedRow(tx Tx, rowID uint64) (*Row, error) {
 
 	useRowCache := tx.UseRowCache()
 	if useRowCache {
+		if f.rowCache == nil {
+			f.rowCache = newSimpleCache()
+		}
 		r, ok := f.rowCache.Fetch(rowID)
 		if ok && r != nil {
 			return r, nil
@@ -693,7 +701,9 @@ func (f *fragment) unprotectedSetBit(tx Tx, rowID, columnID uint64) (changed boo
 	}
 	// Drop the rowCache entry; it's wrong, and we don't want to force
 	// a new copy if no one's reading it.
-	f.rowCache.Add(rowID, nil)
+	if tx.UseRowCache() && f.rowCache != nil {
+		f.rowCache.Add(rowID, nil)
+	}
 
 	f.stats.Count(MetricSetBit, 1, 1.0)
 
@@ -756,7 +766,9 @@ func (f *fragment) unprotectedClearBit(tx Tx, rowID, columnID uint64) (changed b
 	}
 	// Drop the rowCache entry; it's wrong, and we don't want to force
 	// a new copy if no one's reading it.
-	f.rowCache.Add(rowID, nil)
+	if tx.UseRowCache() && f.rowCache != nil {
+		f.rowCache.Add(rowID, nil)
+	}
 
 	f.stats.Count(MetricClearBit, 1, 1.0)
 
@@ -823,7 +835,9 @@ func (f *fragment) unprotectedSetRow(tx Tx, row *Row, rowID uint64) (changed boo
 	}
 
 	// invalidate rowCache for this row.
-	f.rowCache.Add(rowID, nil)
+	if tx.UseRowCache() && f.rowCache != nil {
+		f.rowCache.Add(rowID, nil)
+	}
 
 	// Snapshot storage.
 	f.holder.SnapshotQueue.Enqueue(f)
@@ -872,7 +886,9 @@ func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err e
 
 	// Clear the row in cache.
 	f.cache.Add(rowID, 0)
-	f.rowCache.Add(rowID, nil)
+	if tx.UseRowCache() && f.rowCache != nil {
+		f.rowCache.Add(rowID, nil)
+	}
 
 	// Snapshot storage.
 	f.holder.SnapshotQueue.Enqueue(f)
@@ -2293,6 +2309,8 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64
 	if f.storage != nil {
 		wp = &f.storage.OpWriter
 	}
+	useRowCache := tx.UseRowCache()
+
 	doFunc := func() error {
 		if len(set) > 0 {
 			f.stats.Count(MetricImportingN, int64(len(set)), 1)
@@ -2334,8 +2352,9 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64
 
 				f.cache.BulkAdd(rowID, n)
 			}
-
-			f.rowCache.Add(rowID, nil)
+			if useRowCache && f.rowCache != nil {
+				f.rowCache.Add(rowID, nil)
+			}
 		}
 
 		if f.CacheType != CacheTypeNone {
@@ -2465,8 +2484,10 @@ func (f *fragment) importValueSmallWrite(tx Tx, columnIDs []uint64, values []int
 		return errors.Wrap(err, "importing positions")
 	}
 
-	// Reset the rowCache.
-	f.rowCache = &simpleCache{make(map[uint64]*Row)}
+	if tx.UseRowCache() {
+		// Reset the rowCache.
+		f.rowCache = newSimpleCache()
+	}
 
 	return nil
 }
@@ -2515,8 +2536,10 @@ func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDep
 	f.opN += totalChanges
 	f.ops++
 
-	// Reset the rowCache.
-	f.rowCache = &simpleCache{make(map[uint64]*Row)}
+	if tx.UseRowCache() {
+		// Reset the rowCache.
+		f.rowCache = newSimpleCache()
+	}
 
 	// in theory, this should probably have been queued anyway, but if enough
 	// of the bits matched existing bits, we'll be under our opN estimate, and
@@ -2541,6 +2564,8 @@ func (f *fragment) importRoaring(ctx context.Context, tx Tx, data []byte, clear
 func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []byte, clear bool) error {
 	rowSize := uint64(1 << shardVsContainerExponent)
 	span, ctx := tracing.StartSpanFromContext(ctx, "importRoaring.ImportRoaringBits")
+
+	useRowCache := tx.UseRowCache()
 	var changed int
 	var rowSet map[uint64]int
 	var wp *io.Writer
@@ -2570,7 +2595,9 @@ func (f *fragment) unprotectedImportRoaring(ctx context.Context, tx Tx, data []b
 		if changes == 0 {
 			continue
 		}
-		f.rowCache.Add(rowID, nil)
+		if useRowCache && f.rowCache != nil {
+			f.rowCache.Add(rowID, nil)
+		}
 		if updateCache {
 			anyChanged = true
 			if changes < 0 {
@@ -3208,8 +3235,14 @@ func (f *fragment) intRowIterator(tx Tx, wrap bool, filters ...rowFilter) (rowIt
 	// accumulator [column ID] -> [int value]
 	acc := make(map[uint64]int64)
 
-	f.mu.RLock()
-	defer f.mu.RUnlock()
+	if tx.UseRowCache() {
+		// needs a write lock since it will update the f.rowCache
+		f.mu.Lock()
+		defer f.mu.Unlock()
+	} else {
+		f.mu.RLock()
+		defer f.mu.RUnlock()
+	}
 	if err := f.foreachRow(tx, filters, func(rid uint64) error {
 		// skip exist(0) and sign(1) rows
 		if rid == bsiExistsBit || rid == bsiSignBit {

fragment_internal_test.go

@@ -432,6 +432,7 @@ func TestFragment_SetValue(t *testing.T) {
 		f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "")
 		_ = idx
 		defer f.Clean(t)
+		defer tx.Rollback()
 
 		// Set value.
 		if changed, err := f.setValue(tx, 100, 10, 20); err != nil {

txfactory.go

@@ -27,6 +27,7 @@ import (
 	"text/tabwriter"
 
 	"github.com/pilosa/pilosa/v2/hash"
+	"github.com/pilosa/pilosa/v2/rbf"
 	"github.com/pilosa/pilosa/v2/roaring"
 	"github.com/pilosa/pilosa/v2/txkey"
 	"github.com/pkg/errors"
@@ -522,6 +523,8 @@ func NewTxFactory(txsrc string, holderDir string, holder *Holder) (f *TxFactory,
 	if len(types) == 2 {
 		f.blueGreenReg = newBlueGreenReg(types)
 		f.isBlueGreen = true
+		// blue-green can never use the rowCache.
+		rbf.SetRowcacheOn(false)
 	}
 	f.dbPerShard = f.NewDBPerShard(types, holderDir, holder)
 
@@ -538,6 +541,15 @@ func (f *TxFactory) Open() error {
 	return f.dbPerShard.LoadExistingDBs()
 }
 
+// UseRowCache can be more "global" than Tx at the moment, because
+// we are sharing the same bool flag in rbf at the moment. If
+// this changes then fragment.openStorage() will need a new way
+// to determine if it should use the rowCache. Currently it
+// doesn't have a tx Tx parameter, so we use the Txf instead.
+func (f *TxFactory) UseRowCache() bool {
+	return rbf.EnableRowCache()
+}
+
 // Txo holds the transaction options
 type Txo struct {
 	Write    bool