Merge branch 'master' into cicd-smoketest

cache.go

@@ -567,37 +567,6 @@ func (p uint64Slice) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
 func (p uint64Slice) Len() int           { return len(p) }
 func (p uint64Slice) Less(i, j int) bool { return p[i] < p[j] }
 
-// 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
-// node to run out of memory.
-type simpleCache struct {
-	cache map[uint64]*Row
-}
-
-// Fetch retrieves the bitmap at the id in the cache.
-func (s *simpleCache) Fetch(id uint64) (*Row, bool) {
-	m, ok := s.cache[id]
-	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) {
-	if b != nil {
-		s.cache[id] = b
-	} else {
-		delete(s.cache, id)
-	}
-}
-
 // nopCache represents a no-op Cache implementation.
 type nopCache struct {
 	stats stats.StatsClient

ctl/server.go

@@ -84,9 +84,6 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
 	flags.StringVar(&srv.Config.Storage.Backend, "storage.backend", storage.DefaultBackend, fmt.Sprintf("transaction/storage to use: one of roaring or rbf. The default is: %v. The env var PILOSA_STORAGE_BACKEND is over-ridden by --storage.backend option on the command line.", storage.DefaultBackend))
 	flags.BoolVar(&srv.Config.Storage.FsyncEnabled, "storage.fsync", true, "enable fsync fully safe flush-to-disk")
 
-	// RowcacheOn
-	flags.BoolVar((&srv.Config.RowcacheOn), "rowcache-on", srv.Config.RowcacheOn, "Do not use, permanently disabled. Flag exists for backwards compatibility and will be removed.")
-
 	// RBF specific flags. See pilosa/rbf/cfg/cfg.go for definitions.
 	srv.Config.RBFConfig.DefineFlags(flags)
 

fragment.go

@@ -34,7 +34,6 @@ import (
 	"github.com/molecula/featurebase/v2/roaring"
 	"github.com/molecula/featurebase/v2/shardwidth"
 	"github.com/molecula/featurebase/v2/stats"
-	"github.com/molecula/featurebase/v2/storage"
 	"github.com/molecula/featurebase/v2/testhook"
 	"github.com/molecula/featurebase/v2/topology"
 	"github.com/molecula/featurebase/v2/tracing"
@@ -163,9 +162,6 @@ type fragment struct {
 
 	CacheSize uint32
 
-	// Cache containing full rows (not just counts).
-	rowCache *simpleCache
-
 	// Cached checksums for each block.
 	checksums map[int][]byte
 
@@ -427,13 +423,8 @@ func (f *fragment) inspectStorage(data []byte, file *os.File, newGen generation,
 // logic is now mostly in importStorage (reading in a bitmap) and applyStorage
 // (remapping an existing bitmap to match a new backing store).
 func (f *fragment) openStorage(unmarshalData bool) error {
-
-	useRowCache := storage.RowCacheEnabled()
 	if !f.idx.NeedsSnapshot() {
 		f.gen = &NopGeneration{}
-		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.
@@ -447,9 +438,7 @@ func (f *fragment) openStorage(unmarshalData bool) error {
 		// unmarshal this data in order to have any.
 		unmarshalData = true
 	}
-	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
@@ -612,24 +601,10 @@ func (f *fragment) mustRow(tx Tx, rowID uint64) *Row {
 // unprotectedRow returns a row from the row cache if available or from storage
 // (updating the cache).
 func (f *fragment) unprotectedRow(tx Tx, rowID uint64) (*Row, error) {
-
-	useRowCache := storage.RowCacheEnabled()
-	if useRowCache {
-		if f.rowCache == nil {
-			f.rowCache = newSimpleCache()
-		}
-		r, ok := f.rowCache.Fetch(rowID)
-		if ok && r != nil {
-			return r, nil
-		}
-	}
 	row, err := f.rowFromStorage(tx, rowID)
 	if err != nil {
 		return nil, err
 	}
-	if useRowCache {
-		f.rowCache.Add(rowID, row)
-	}
 	return row, nil
 }
 
@@ -742,11 +717,6 @@ func (f *fragment) unprotectedSetBit(tx Tx, rowID, columnID uint64) (changed boo
 		}
 		f.cache.Add(rowID, n)
 	}
-	// Drop the rowCache entry; it's wrong, and we don't want to force
-	// a new copy if no one's reading it.
-	if storage.RowCacheEnabled() && f.rowCache != nil {
-		f.rowCache.Add(rowID, nil)
-	}
 
 	f.stats.Count(MetricSetBit, 1, 1.0)
 
@@ -807,11 +777,6 @@ func (f *fragment) unprotectedClearBit(tx Tx, rowID, columnID uint64) (changed b
 		}
 		f.cache.Add(rowID, n)
 	}
-	// Drop the rowCache entry; it's wrong, and we don't want to force
-	// a new copy if no one's reading it.
-	if storage.RowCacheEnabled() && f.rowCache != nil {
-		f.rowCache.Add(rowID, nil)
-	}
 
 	f.stats.Count(MetricClearBit, 1, 1.0)
 
@@ -877,11 +842,6 @@ func (f *fragment) unprotectedSetRow(tx Tx, row *Row, rowID uint64) (changed boo
 		}
 	}
 
-	// invalidate rowCache for this row.
-	if storage.RowCacheEnabled() && f.rowCache != nil {
-		f.rowCache.Add(rowID, nil)
-	}
-
 	// Snapshot storage.
 	f.holder.SnapshotQueue.Enqueue(f)
 	f.stats.Count("setRow", 1, 1.0)
@@ -929,9 +889,6 @@ func (f *fragment) unprotectedClearRow(tx Tx, rowID uint64) (changed bool, err e
 
 	// Clear the row in cache.
 	f.cache.Add(rowID, 0)
-	if storage.RowCacheEnabled() && f.rowCache != nil {
-		f.rowCache.Add(rowID, nil)
-	}
 
 	// Snapshot storage.
 	f.holder.SnapshotQueue.Enqueue(f)
@@ -2416,7 +2373,6 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64
 	if f.storage != nil {
 		wp = &f.storage.OpWriter
 	}
-	useRowCache := storage.RowCacheEnabled()
 
 	doFunc := func() error {
 		if len(set) > 0 {
@@ -2457,9 +2413,6 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64
 
 				f.cache.BulkAdd(rowID, n)
 			}
-			if useRowCache && f.rowCache != nil {
-				f.rowCache.Add(rowID, nil)
-			}
 		}
 
 		if f.CacheType != CacheTypeNone {
@@ -3328,14 +3281,8 @@ func (f *fragment) intRowIterator(tx Tx, wrap bool, filters ...roaring.BitmapFil
 	// accumulator [column ID] -> [int value]
 	acc := make(map[uint64]int64)
 
-	if storage.RowCacheEnabled() {
-		// 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()
-	}
+	f.mu.RLock()
+	defer f.mu.RUnlock()
 	callback := func(rid uint64) error {
 		// skip exist(0) and sign(1) rows
 		if rid == bsiExistsBit || rid == bsiSignBit {

fragment_internal_test.go

@@ -115,59 +115,6 @@ func TestFragment_ClearBit(t *testing.T) {
 	}
 }
 
-/* We suspect this test is no longer valid under the new Tx
-   framework in which we always copy mmap-ed rows before
-   returning them. So we will comment it out for now.
-   If someone knows any reason for this to stick around,
-   let us know; we couldn't figure out how to adapt
-   to do a meaningful test under Tx. - jaten / tgruben
-
-// What about rowcache timing.
-func TestFragment_RowcacheMap(t *testing.T) {
-	var done int64
-	f, _, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "")
-
-	// Under -race, this test turns out to take a fairly long time
-	// to run with larger OpN, because we write 50,000 bits to
-	// the bitmap, and everything is being race-detected, and we don't
-	// actually need that many to get the result we care about.
-	f.MaxOpN = 2000
-	defer f.Clean(t) // failing here with     TestFragment_RowcacheMap: fragment_internal_test.go:2859: fragment /var/folders/2x/hm9gp5ys3k9gmm5f_vzm_6wc0000gn/T/pilosa-fragment-001943331: unmarshalled bitmap different: differing containers for key 0: <run container, N=10131, len 2000x interval> vs <run container, N=12000, len 2000x interval>
-
-	ch := make(chan struct{})
-
-	for i := 0; i < f.MaxOpN; i++ {
-		_, _ = f.setBit(tx, 0, uint64(i*32))
-	}
-	// force snapshot so we get a mmapped row...
-	_ = f.Snapshot()
-	row := f.mustRow(tx, 0)
-	tx.Commit(0)
-	segment := row.Segments()[0]
-	bitmap := segment.data
-
-	// request information from the frozen bitmap we got back
-	go func() {
-		for atomic.LoadInt64(&done) == 0 {
-			for i := 0; i < f.MaxOpN; i++ {
-				_ = bitmap.Contains(uint64(i * 32))
-			}
-		}
-		close(ch)
-	}()
-
-	// modify the original bitmap, until it causes a snapshot, which
-	// then invalidates the other map...
-	for j := 0; j < 5; j++ {
-		for i := 0; i < f.MaxOpN; i++ {
-			_, _ = f.setBit(tx, 0, uint64(i*32+j+1))
-		}
-	}
-	atomic.StoreInt64(&done, 1)
-	<-ch
-}
-*/
-
 // Ensure a fragment can clear a row.
 func TestFragment_ClearRow(t *testing.T) {
 	f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "")

holder.go

@@ -149,9 +149,6 @@ type HolderOpts struct {
 	// StorageBackend controls the tx/storage engine we instatiate. Set by
 	// server.go OptServerStorageConfig
 	StorageBackend string
-
-	// RowcacheOn, if true, turns on the row cache for all storage backends.
-	RowcacheOn bool
 }
 
 func (h *Holder) StartTransaction(ctx context.Context, id string, timeout time.Duration, exclusive bool) (*Transaction, error) {
@@ -213,7 +210,6 @@ type HolderConfig struct {
 	CacheFlushInterval   time.Duration
 	StatsClient          stats.StatsClient
 	Logger               logger.Logger
-	RowcacheOn           bool
 
 	StorageConfig       *storage.Config
 	RBFConfig           *rbfcfg.Config
@@ -273,7 +269,7 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
 		sharder:              cfg.Sharder,
 		schemator:            cfg.Schemator,
 		Logger:               cfg.Logger,
-		Opts:                 HolderOpts{StorageBackend: cfg.StorageConfig.Backend, RowcacheOn: cfg.RowcacheOn},
+		Opts:                 HolderOpts{StorageBackend: cfg.StorageConfig.Backend},
 
 		SnapshotQueue: defaultSnapshotQueue,
 
@@ -284,8 +280,6 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
 		indexes: make(map[string]*Index),
 	}
 
-	storage.SetRowCacheOn(cfg.RowcacheOn)
-
 	txf, err := NewTxFactory(cfg.StorageConfig.Backend, h.IndexesPath(), h)
 	vprint.PanicOn(err)
 	h.txf = txf

qa/tf/.modules/featurebase-cluster/main.tf

@@ -83,6 +83,12 @@ resource "aws_instance" "fb_ingest" {
 resource "aws_key_pair" "gitlab-featurebase-ci" {
   key_name   = "${var.cluster_prefix}-gitlab-ci"
   public_key = "ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQC91hhpVHNonAG7ku2ugpxEskf9KHeyHJPQJT26OHrMUw7R+T5A8TjqSzTau07sXQ/E9SO3ebV8SJ5PqeaQOnQB8VEvVNK0DjQH7ppvNg1Rfs42FZT9ttzTMvOjsSbK3vZTHXdoKQEdC9NxBwSkFIRGQojK1HUOq9xGrw31fA1OjSwlpLcbx7yyg18lcqW6UOptnVR8U9Yy9qQ5jZF1HtkQ6L9J+gv4o1UyNAUK2bopeGiXpBc3PQ/CFaFT2h/aqLBP66qAHsHVyAFD3PIRtplC5EHa8jXDgLacEls0uF7Q3kRPxvzcuo4g4VkOn1rDy9qH3vd2hT3aKVnM73FIDUiL"
+  
+  tags = {
+    Prefix = "${var.cluster_prefix}"
+    Name   = "${var.cluster_prefix}-gitlab-featurebase-ci"
+    Role   = "ssh_keypair"
+  }
 }
 
 resource "aws_security_group" "featurebase" {
@@ -156,7 +162,9 @@ resource "aws_security_group" "featurebase" {
   }
 
   tags = {
-    Name = "allow_featurebase"
+    Prefix = "${var.cluster_prefix}"
+    Name   = "${var.cluster_prefix}-allow_featurebase"
+    Role   = "allow_featurebase"
   }
 }
 
@@ -199,13 +207,21 @@ resource "aws_security_group" "ingest" {
   }
 
   tags = {
-    Name = "allow_ingest"
+    Prefix = "${var.cluster_prefix}"
+    Name   = "${var.cluster_prefix}-allow_ingest"
+    Role   = "allow_ingest"
   }
 }
 
 resource "aws_iam_instance_profile" "fb_cluster_node_profile" {
   name = "${var.cluster_prefix}-fb_cluster_node_profile"
   role = aws_iam_role.fb_cluster_node_role.name
+
+  tags = {
+    Prefix = "${var.cluster_prefix}"
+    Name   = "${var.cluster_prefix}-fb_cluster_node_profile"
+    Role   = "fb_cluster_node_profile"
+  }
 }
 
 resource "aws_iam_role" "fb_cluster_node_role" {
@@ -239,4 +255,9 @@ resource "aws_iam_role" "fb_cluster_node_role" {
     })
   }
 
+  tags = {
+    Prefix = "${var.cluster_prefix}"
+    Name   = "${var.cluster_prefix}-fb_cluster_node_role"
+    Role   = "fb_cluster_node_role"
+  }
 }

rbf/cursorx.go

@@ -7,10 +7,8 @@ import (
 	"io"
 	"math"
 	"os"
-	"unsafe"
 
 	"github.com/molecula/featurebase/v2/roaring"
-	"github.com/molecula/featurebase/v2/storage"
 	"github.com/pkg/errors"
 )
 
@@ -162,8 +160,8 @@ func intoContainer(l leafCell, tx *Tx, replacing *roaring.Container, target []by
 	orig := l.Data
 	var cpMaybe []byte
 	var mapped bool
-	if storage.RowCacheEnabled() || tx.db.cfg.DoAllocZero {
-		// make a copy, otherwise the rowCache will see corrupted data
+	if tx.db.cfg.DoAllocZero {
+		// make a copy so no one will see corrupted data
 		// or mmapped data that may disappear.
 		cpMaybe = target[:len(orig)]
 		copy(cpMaybe, orig)
@@ -179,10 +177,6 @@ func intoContainer(l leafCell, tx *Tx, replacing *roaring.Container, target []by
 	case ContainerTypeBitmapPtr:
 		_, bm, _ := tx.leafCellBitmap(toPgno(cpMaybe))
 		cloneMaybe := bm
-		if storage.RowCacheEnabled() {
-			cloneMaybe = (*[1024]uint64)(unsafe.Pointer(&target[0]))[:1024]
-			copy(cloneMaybe, bm)
-		}
 		c = roaring.RemakeContainerBitmap(replacing, cloneMaybe)
 	case ContainerTypeBitmap:
 		c = roaring.RemakeContainerBitmapN(replacing, toArray64(cpMaybe), int32(l.BitN))
@@ -205,8 +199,8 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
 	orig := l.Data
 	var cpMaybe []byte
 	var mapped bool
-	if storage.RowCacheEnabled() || tx.db.cfg.DoAllocZero {
-		// make a copy, otherwise the rowCache will see corrupted data
+	if tx.db.cfg.DoAllocZero {
+		// make a copy, otherwise someone could see corrupted data
 		// or mmapped data that may disappear.
 		cpMaybe = make([]byte, len(orig))
 		copy(cpMaybe, orig)
@@ -222,10 +216,6 @@ func toContainer(l leafCell, tx *Tx) (c *roaring.Container) {
 	case ContainerTypeBitmapPtr:
 		_, bm, _ := tx.leafCellBitmap(toPgno(cpMaybe))
 		cloneMaybe := bm
-		if storage.RowCacheEnabled() {
-			cloneMaybe = make([]uint64, len(bm))
-			copy(cloneMaybe, bm)
-		}
 		c = roaring.NewContainerBitmap(-1, cloneMaybe)
 	case ContainerTypeBitmap:
 		c = roaring.NewContainerBitmap(-1, toArray64(cpMaybe))

rbf/db_test.go

@@ -3,6 +3,7 @@ package rbf_test
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"math/rand"
 	"net"
@@ -46,6 +47,35 @@ func TestDB_WAL(t *testing.T) {
 		}
 	})
 
+	t.Run("ErrTxTooLargeWithBitmap", func(t *testing.T) {
+		config := rbfcfg.NewDefaultConfig()
+		config.MaxWALSize = 5 * rbf.PageSize
+
+		db := MustOpenDB(t, config)
+		defer MustCloseDB(t, db)
+
+		tx := MustBegin(t, db, true)
+		defer tx.Rollback()
+
+		if err := tx.CreateBitmap("x"); err != nil {
+			t.Fatal(err)
+		}
+
+		// Fill array until it has the maximum number of elements.
+		for i := uint64(0); i < rbf.ArrayMaxSize; i++ {
+			if _, err := tx.Add("x", i); err != nil {
+				t.Fatal(err)
+			}
+		}
+
+		// Issuing one more item to a full array should convert it to a bitmap
+		// page and cause the write to return "tx too large". Previous to the
+		// FB-828 fix, this would write past the mmap size so it was inaccessible.
+		if _, err := tx.Add("x", rbf.ArrayMaxSize); err == nil || !errors.Is(err, rbf.ErrTxTooLarge) {
+			t.Fatalf("unexpected error: %#v", err)
+		}
+	})
+
 	t.Run("Halt", func(t *testing.T) {
 		if testing.Short() {
 			t.Skip("-short enabled, skipping")

rbf/tx.go

@@ -1163,7 +1163,8 @@ func (tx *Tx) writeBitmapPage(pgno uint32, page []byte) error {
 }
 
 func (tx *Tx) checkTxSize() error {
-	if (tx.walPageN+tx.dirtyN())*PageSize >= len(tx.db.wal) {
+	pageN := tx.walPageN + len(tx.dirtyPages) + (len(tx.dirtyBitmapPages) * 2)
+	if pageN*PageSize >= len(tx.db.wal) {
 		return ErrTxTooLarge
 	}
 	return nil

server.go

@@ -341,15 +341,6 @@ func OptServerStorageConfig(cfg *storage.Config) ServerOption {
 	}
 }
 
-// OptServerRowcacheOn is a functional option on Server
-// used to turn on the row cache.
-func OptServerRowcacheOn(rowcacheOn bool) ServerOption {
-	return func(s *Server) error {
-		s.holderConfig.RowcacheOn = rowcacheOn
-		return nil
-	}
-}
-
 // OptServerRBFConfig conveys the RBF flags to the Holder.
 func OptServerRBFConfig(cfg *rbfcfg.Config) ServerOption {
 	return func(s *Server) error {

server/config.go

@@ -203,11 +203,6 @@ type Config struct {
 	// "rbf".
 	Storage *storage.Config `toml:"storage"`
 
-	// RowcacheOn permanently disabled. No longer useful w/ RBF. Left
-	// for backward compatibility but will be removed in a future
-	// version.
-	RowcacheOn bool `toml:"rowcache-on"`
-
 	// RBFConfig defines all externally configurable RBF flags.
 	RBFConfig *rbfcfg.Config `toml:"rbf"`
 

server/server.go

@@ -485,7 +485,6 @@ func (m *Command) SetupServer() error {
 		pilosa.OptServerClusterName(m.Config.Cluster.Name),
 		pilosa.OptServerSerializer(proto.Serializer{}),
 		pilosa.OptServerStorageConfig(m.Config.Storage),
-		pilosa.OptServerRowcacheOn(false),
 		pilosa.OptServerRBFConfig(m.Config.RBFConfig),
 		pilosa.OptServerMaxQueryMemory(m.Config.MaxQueryMemory),
 		pilosa.OptServerQueryHistoryLength(m.Config.QueryHistoryLength),

storage/cache.go

@@ -1,24 +0,0 @@
-// Copyright 2021 Molecula Corp. All rights reserved.
-package storage
-
-import (
-	"sync/atomic"
-)
-
-// if enableRowCache, then we must not return mmap-ed memory
-// directly, but only a copy.
-var enableRowcache int64 = 1
-
-// SetRowCacheOn should only be called in NewHolder before
-// all other reads.
-func SetRowCacheOn(on bool) {
-	if on {
-		atomic.StoreInt64(&enableRowcache, 1)
-	} else {
-		atomic.StoreInt64(&enableRowcache, 0)
-	}
-}
-
-func RowCacheEnabled() bool {
-	return atomic.LoadInt64(&enableRowcache) == 1
-}