remove string keys on delete to allow for reuse

boltdb/translate.go

@@ -12,7 +12,8 @@ import (
 	"sync"
 	"time"
 
-	"github.com/molecula/featurebase/v3"
+	pilosa "github.com/molecula/featurebase/v3"
+	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/pkg/errors"
 	bolt "go.etcd.io/bbolt"
 
@@ -32,6 +33,8 @@ var (
 
 	bucketKeys = []byte("keys")
 	bucketIDs  = []byte("ids")
+	bucketFree = []byte("free")
+	FreeKey    = []byte("free")
 )
 
 const (
@@ -119,6 +122,8 @@ func (s *TranslateStore) Open() (err error) {
 			return err
 		} else if _, err := tx.CreateBucketIfNotExists(bucketIDs); err != nil {
 			return err
+		} else if _, err := tx.CreateBucketIfNotExists(bucketFree); err != nil {
+			return err
 		}
 		return nil
 	}); err != nil {
@@ -445,7 +450,7 @@ func (r *TranslateEntryReader) Close() error {
 	return nil
 }
 
-// ReadEntry reads the next entry from the underlying translate store.
+// ReadEntry reads th next entry from the underlying translate store.
 func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
 	// Ensure reader has not been closed before read.
 	select {
@@ -498,6 +503,88 @@ func (r *TranslateEntryReader) ReadEntry(entry *pilosa.TranslateEntry) error {
 	}
 }
 
+type boltWrapper struct {
+	tx *bolt.Tx
+	db *bolt.DB
+}
+
+func (w *boltWrapper) Commit() error {
+	if w.tx != nil {
+		return w.tx.Commit()
+	}
+	return nil
+}
+
+func (w *boltWrapper) Rollback() {
+	if w.tx != nil {
+		w.tx.Rollback()
+	}
+}
+func (s *TranslateStore) FreeIDs() (*roaring.Bitmap, error) {
+	result := roaring.NewBitmap()
+	err := s.db.View(func(tx *bolt.Tx) error {
+		bkt := tx.Bucket(bucketFree)
+		if bkt == nil {
+			return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys)
+		}
+		b := bkt.Get(FreeKey)
+		err := result.UnmarshalBinary(b)
+		if err != nil {
+			return err
+		}
+		return nil
+	})
+	return result, err
+}
+func (s *TranslateStore) MergeFree(tx *bolt.Tx, newIDs *roaring.Bitmap) error {
+	bkt := tx.Bucket(bucketFree)
+	b := bkt.Get(FreeKey)
+	buf := new(bytes.Buffer)
+	if b != nil { //if existing combine with newIDs
+		before := roaring.NewBitmap()
+		err := before.UnmarshalBinary(b)
+		if err != nil {
+			return err
+		}
+		final := newIDs.Union(before)
+		_, err = final.WriteTo(buf)
+		if err != nil {
+			return err
+		}
+	} else {
+		newIDs.WriteTo(buf)
+	}
+	return bkt.Put(FreeKey, buf.Bytes())
+}
+
+// Delete removes the lookeup pairs in order to make avialble for reuse but doesn't commit the
+// transaction for that is tied to the associated rbf transaction being successful
+func (s *TranslateStore) Delete(records *roaring.Bitmap) (pilosa.Commitor, error) {
+	tx, err := s.db.Begin(true)
+	if err != nil {
+		return nil, err
+	}
+	keyBucket := tx.Bucket(bucketKeys)
+	idBucket := tx.Bucket(bucketIDs)
+	ids := records.Slice()
+	for i := range ids {
+		id := u64tob(ids[i])
+		boltKey := idBucket.Get(id)
+		err = keyBucket.Delete(boltKey)
+		if err != nil {
+			tx.Rollback()
+			return &boltWrapper{}, err
+		}
+		err = idBucket.Delete(id)
+		if err != nil {
+			tx.Rollback()
+			return &boltWrapper{}, err
+		}
+
+	}
+	return &boltWrapper{tx: tx}, s.MergeFree(tx, records)
+}
+
 // emptyKey is a sentinel byte slice which stands for "" as a key.
 var emptyKey = []byte{
 	0x00, 0x00, 0x00,

boltdb/translate_test.go

@@ -10,8 +10,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/molecula/featurebase/v3"
+	pilosa "github.com/molecula/featurebase/v3"
 	"github.com/molecula/featurebase/v3/boltdb"
+	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/molecula/featurebase/v3/testhook"
 	"github.com/molecula/featurebase/v3/topology"
 )
@@ -385,7 +386,53 @@ func MustNewTranslateStore(tb testing.TB) *boltdb.TranslateStore {
 	s.Path = f.Name()
 	return s
 }
+func TestTranslateStore_Delete(t *testing.T) {
+	s := MustOpenNewTranslateStore(t)
+	defer MustCloseTranslateStore(s)
 
+	// Setup initial keys.
+	ids, err := s.CreateKeys("foo", "bar", "deleteme")
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	records := roaring.NewBitmap(ids["deleteme"])
+	c, err := s.Delete(records)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = c.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	r, e := s.FreeIDs()
+	if e != nil {
+		t.Fatal(err)
+	}
+	freeids := r.Slice()
+	if len(freeids) == 0 {
+		t.Fatalf("expected to have free id")
+	}
+	if freeids[0] != ids["deleteme"] {
+		t.Fatalf("expected [%v] and got %v", ids["deleteme"], freeids[0])
+	}
+
+	records2 := roaring.NewBitmap(ids["foo"])
+	c, err = s.Delete(records2)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = c.Commit(); err != nil {
+		t.Fatal(err)
+	}
+	r, e = s.FreeIDs()
+	if e != nil {
+		t.Fatal(err)
+	}
+	freeids = r.Slice()
+	if len(freeids) != 2 {
+		t.Fatalf("expected to have 2 free ids")
+	}
+}
 func TestTranslateStore_ReadWrite(t *testing.T) {
 	t.Run("WriteTo_ReadFrom", func(t *testing.T) {
 		s := MustOpenNewTranslateStore(t)

catcher.go

@@ -26,6 +26,11 @@ func init() {
 
 var _ Tx = (*catcherTx)(nil)
 
+func (c *catcherTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) {
+	c.b.RemoveChannel(index, field, view, shard, a, resChan)
+	return
+}
+
 func (c *catcherTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator {
 	return c.b.NewTxIterator(index, field, view, shard)
 }

dbshard_internal_test.go

@@ -3,6 +3,7 @@ package pilosa
 
 import (
 	"fmt"
+	"math/rand"
 	"os"
 	"path/filepath"
 	"strings"
@@ -296,3 +297,67 @@ func Test_DBPerShard_GetFieldView2Shards_map_from_RBF(t *testing.T) {
 		panic(fmt.Sprintf("expected '%v' but got view2shard '%v'", exp, view2shard))
 	}
 }
+func TestTXBigDelete(t *testing.T) {
+	if _, ok := os.LookupEnv("GAUNTLET"); !ok {
+		t.Skip("only running this test if GAUNTLET is set")
+	}
+
+	f, idx, tx := mustOpenFragment(t, "i", "f", viewStandard, 0, "")
+	_ = idx
+	defer f.Clean(t)
+	result := make(map[uint64]struct{})
+	var set, none []uint64
+	accum := uint64(0)
+	N := 1000000
+	fmt.Println("build big set")
+	rand.Seed(0)
+	for i := 0; i < N; i++ {
+		bd := rand.Intn(15)
+		accum += uint64(bd)
+		for row := uint64(0); row < uint64(rand.Intn(10)); row++ {
+			pos, _ := f.pos(row, accum)
+			set = append(set, pos)
+		}
+	}
+	fmt.Println("set")
+	err := f.importPositions(tx, set, none, result)
+	PanicOn(err)
+	PanicOn(tx.Commit())
+
+	// Close and reopen the fragment & verify the data.
+	fmt.Println("repopen")
+	err = f.Reopen() // roaring data not being flushed? red on roaring
+	if err != nil {
+		t.Fatal(err)
+	}
+	fmt.Println("clear")
+	tx = idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Fragment: f, Shard: f.shard})
+	row, er := f.row(tx, 5)
+	PanicOn(er)
+	fmt.Println(row.Count())
+	cols := row.Columns()
+	subset := make([]uint64, len(cols))
+	for i := range cols {
+		pos, e := f.pos(5, cols[i])
+		PanicOn(e)
+		subset[i] = pos
+	}
+	PanicOn(f.importPositions(tx, none, subset, result))
+	row, er = f.row(tx, 5)
+	PanicOn(er)
+	fmt.Println(row.Count())
+	PanicOn(tx.Commit())
+
+	fmt.Println("reopen")
+	err = f.Reopen() // roaring data not being flushed? red on roaring
+	if err != nil {
+		t.Fatal(err)
+	}
+	tx = idx.holder.txf.NewTx(Txo{Write: !writable, Index: idx, Fragment: f, Shard: f.shard})
+	row, er = f.row(tx, 5)
+	PanicOn(er)
+	fmt.Println("delete count should be 0", row.Count())
+	row, er = f.row(tx, 2)
+	PanicOn(er)
+	fmt.Println(row.Count())
+}

delete_test.go

@@ -8,7 +8,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/molecula/featurebase/v3"
+	pilosa "github.com/molecula/featurebase/v3"
+	"github.com/molecula/featurebase/v3/disco"
 	"github.com/molecula/featurebase/v3/test"
 	"github.com/stretchr/testify/require"
 )
@@ -49,6 +50,21 @@ func TestExecutor_DeleteRecords(t *testing.T) {
 		})
 
 	}
+	setupBig := func(t *testing.T, r *require.Assertions, c *test.Cluster, Rows uint64) {
+		t.Helper()
+		fieldName := "setfield"
+		c.CreateField(t, indexName, pilosa.IndexOptions{TrackExistence: true}, fieldName)
+		rows := make([][2]uint64, ShardWidth*Rows)
+		for columnID := uint64(0); columnID < ShardWidth; columnID++ {
+			for rowID := uint64(0); rowID < Rows; rowID++ {
+				if rowID == 0 || (columnID%rowID+1) != 0 {
+					rows[rowID] = [2]uint64{rowID, columnID}
+				}
+			}
+		}
+		c.ImportBits(t, indexName, "setfield", rows)
+	}
+
 	setupKeys := func(t *testing.T, r *require.Assertions, c *test.Cluster) {
 		t.Helper()
 		c.CreateField(t, indexName, pilosa.IndexOptions{Keys: true, TrackExistence: true}, "timefield", pilosa.OptFieldKeys(), pilosa.OptFieldTypeTime(pilosa.TimeQuantum("YMDH")))
@@ -131,6 +147,12 @@ func TestExecutor_DeleteRecords(t *testing.T) {
 			m = resp.Results[0].(pilosa.ExtractedTable)
 			after := convertKey(m.Columns)
 			require.Equal([]string{"B", "C", "D", "two"}, after, "these keyed records after delete")
+			//validate that column keys got deleted
+			node := c.GetNode(0)
+			keys := []string{"A", "one"}
+			res, err := node.API.FindIndexKeys(context.Background(), indexName, keys...)
+			require.Nil(err)
+			require.Empty(res)
 		})
 		t.Run("Delete Row", func(t *testing.T) {
 			setup(t, require, c)
@@ -200,8 +222,33 @@ func TestExecutor_DeleteRecords(t *testing.T) {
 			require.Equal([]uint64{0, 1}, after, "these records should be remaining")
 		})
 	})
+	t.Run("DeleteRecordsBigWithRestart", func(t *testing.T) {
+		c := test.MustNewCluster(t, 1)
+		for _, n := range c.Nodes {
+			n.Config.Cluster.ReplicaN = 1
+		}
+		err := c.Start()
+		defer c.Close()
+		require.NoError(err, "Start cluster DeleteRecordsBig")
+		setupBig(t, require, c, 16)
+		defer tearDown(t, require, c)
+		node := c.GetNode(0)
+		resp := c.Query(t, indexName, `Delete(Row(setfield=12))`)
+		require.NotNil(resp, "Response should not be nil")
+		require.NotEmpty(resp.Results)
+		require.Equal(true, resp.Results[0], "Change should have happened")
+		resp = c.Query(t, indexName, `Count(Row(setfield=12))`)
+		require.NotNil(resp, "Response should not be nil")
+		require.NotEmpty(resp.Results)
+		require.Equal(uint64(0), resp.Results[0], "Should have removed")
+		err = node.Reopen()
+		require.NoError(err, "restart cluster DeleteRecordsBig")
+		err = c.AwaitState(disco.ClusterStateNormal, 10*time.Second)
+		require.NoError(err, "backToNormal")
 
+	})
 }
+
 func convert(before []pilosa.ExtractedTableColumn) []uint64 {
 	result := make([]uint64, 0)
 	for _, i := range before {

executor.go

@@ -8252,72 +8252,117 @@ func (e *executor) executeDeleteRecords(ctx context.Context, qcx *Qcx, index str
 	return n, nil
 }
 
-func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (bool, error) {
+func transactExistRow(ctx context.Context, idx *Index, shard uint64, frag *fragment, src *Row) (uint64, error) {
+	tx := idx.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
+	rows, err := frag.rows(ctx, tx, 1)
+	if err != nil {
+		tx.Rollback()
+		return 0, err
 
+	}
+	rowID := uint64(len(rows) + 1)
+	_, err = frag.setRow(tx, src, rowID)
+	if err != nil {
+		tx.Rollback()
+		return 0, err
+	}
+	return rowID, tx.Commit()
+}
+func (e *executor) executeDeleteRecordFromShard(ctx context.Context, qcx *Qcx, index string, c *pql.Call, shard uint64) (changed bool, err error) {
 	span, _ := tracing.StartSpanFromContext(ctx, "Executor.executeDeleteRecordFromShard")
 	defer span.Finish()
 	//need to build the bitmap in the call
 	child := c.Children[0]
-	row, err := e.executeBitmapCallShard(ctx, qcx, index, child, shard)
-	if err != nil {
-		return false, err
+	src, er := e.executeBitmapCallShard(ctx, qcx, index, child, shard)
+	if er != nil {
+		err = er
+		return
 	}
-	if len(row.segments) == 0 { //nothing to remove
-		return false, nil
+	if len(src.segments) == 0 { //nothing to remove
+		return
+	}
+	columns := src.segments[0].data //should only be one segment
+	if columns.Count() == 0 {
+		return
 	}
-
 	// Fetch index.
 	idx := e.Holder.Index(index)
 	if idx == nil {
-		return false, newNotFoundError(ErrIndexNotFound, index)
+		err = newNotFoundError(ErrIndexNotFound, index)
+		return
 	}
 
-	return DeleteRows(row, idx, shard)
+	return DeleteRows(ctx, src, idx, shard)
 }
 
-func DeleteRows(row *Row, idx *Index, shard uint64) (bool, error) {
-	tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
-	defer tx.Rollback()
-
+func DeleteRows(ctx context.Context, row *Row, idx *Index, shard uint64) (bool, error) {
+	var existenceFragment *fragment
+	var deletedRowID uint64
+	var commitor Commitor = &NopCommitor{}
+	var err error
 	columns := row.segments[0].data //should only be one segment
-	if columns.Count() == 0 {
-		return false, nil
-	}
-	columnIDs := make([]uint64, 0)
-	none := make([]uint64, 0) // no bits will be set
-
-	changed := false
-	colCounts := make([]int, 0)
-	toClear := columnIDs[:0]
-	rowSet := make(map[uint64]struct{})
-	callback := func(pos uint64) error {
-		toClear = append(toClear, pos)
-		rowID := pos / ShardWidth
-		rowSet[rowID] = struct{}{}
-		return nil
-	}
-	findExisting := roaring.NewBitmapBitmapFilter(columns, callback)
-
-	clearFragment := func(frag *fragment) (bool, error) {
-		// re-zero these
-		toClear = columnIDs[:0]
-		rowSet = make(map[uint64]struct{})
-
-		err := tx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting)
+	if idx.Keys() {
+		columns := row.segments[0].data
+		//store columns in exits field ToBeDelete row commited
+		existenceFragment = idx.Holder().fragment(idx.Name(), existenceFieldName, viewStandard, shard)
+		if existenceFragment == nil {
+			//no exists field
+			return false, errors.New("can't bulk delete without existence field")
+		}
+		deletedRowID, err = transactExistRow(ctx, idx, shard, existenceFragment, row)
+		commitor, err = deleteKeyTranslation(ctx, idx, shard, columns)
 		if err != nil {
 			return false, err
 		}
-		colCounts = append(colCounts, len(toClear))
-		// this will be the remove part
-		if len(toClear) > 0 {
-			err = frag.importPositions(tx, none, toClear, rowSet)
-			if err != nil {
-				return false, err
-			}
-			return true, nil
-		}
-		return false, nil
 	}
+	writeTx := idx.Txf().NewTx(Txo{Write: writable, Index: idx, Shard: shard})
+	defer writeTx.Rollback()
+	if err != nil {
+		return false, err
+	}
+	changed := false
+	defer func() {
+		//if there is an error on the bit clearing rollback the keys
+		if err != nil {
+			changed = false
+			commitor.Rollback()
+			return
+		}
+		// if there is an error in the key commit, then rollback the delete
+		// write records before keys to remove possiblity of unmatch keys=records
+		err = writeTx.Commit()
+		if err != nil {
+			changed = false
+			commitor.Rollback()
+			return
+		}
+		if er := commitor.Commit(); er != nil {
+			err = er
+		}
+		if err != nil {
+			idx.Holder().Logger.Errorf("problems committing delete in rbf %v shard %v", err, shard)
+		}
+
+	}()
+	findExisting := roaring.NewBitmapBitmapFilter(columns, func(p uint64) error { return nil })
+	resChan := make(chan countResults)
+	clearFragment := func(frag *fragment) (bool, error) {
+		posChan := make(chan uint64, 8192)
+		findExisting.SetCallback(func(pos uint64) error {
+			posChan <- pos
+			return nil
+		})
+		go writeTx.RemoveChannel(frag.index(), frag.field(), frag.view(), frag.shard, posChan, resChan)
+
+		err = writeTx.ApplyFilter(frag.index(), frag.field(), frag.view(), frag.shard, 0, findExisting)
+		close(posChan)
+		if err != nil {
+			return false, err
+		}
+		r := <-resChan
+		return r.changeCount > 0, r.err
+	}
+
 	for _, field := range idx.Fields() {
 		for _, view := range field.views() {
 
@@ -8332,7 +8377,33 @@ func DeleteRows(row *Row, idx *Index, shard uint64) (bool, error) {
 			if c {
 				changed = true
 			}
+
 		}
 	}
-	return changed, tx.Commit()
+	close(resChan)
+	if existenceFragment != nil { //a string keys have been deleted and the deleteRow was created
+		existenceFragment.clearRow(writeTx, deletedRowID)
+	}
+	return changed, nil
+}
+
+type Commitor interface {
+	Rollback()
+	Commit() error
+}
+type NopCommitor struct {
+}
+
+func (c *NopCommitor) Rollback() {
+
+}
+func (c *NopCommitor) Commit() error {
+	return nil
+}
+
+func deleteKeyTranslation(ctx context.Context, idx *Index, shard uint64, records *roaring.Bitmap) (Commitor, error) {
+	// ShardToShardParition ...
+	paritionID := topology.ShardToShardPartition(idx.name, shard, idx.holder.partitionN)
+
+	return idx.TranslateStore(paritionID).Delete(records)
 }

go.mod

@@ -12,6 +12,7 @@ require (
 	github.com/benbjohnson/immutable v0.3.0
 	github.com/buger/jsonparser v1.1.1
 	github.com/cespare/xxhash v1.1.0
+	github.com/claygod/PiHex v0.0.0-20200916193129-5277802bfd7b // indirect
 	github.com/davecgh/go-spew v1.1.1
 	github.com/desertbit/timer v0.0.0-20180107155436-c41aec40b27f // indirect
 	github.com/dustin/go-humanize v1.0.0 // indirect

go.sum

@@ -54,6 +54,8 @@ github.com/cespare/xxhash v1.1.0/go.mod h1:XrSqR1VqqWfGrhpAt58auRo0WTKS1nRRg3ghf
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
 github.com/chzyer/test v0.0.0-20180213035817-a1ea475d72b1/go.mod h1:Q3SI9o4m/ZMnBNeIyt5eFwwo7qiLfzFZmjNmxjkiQlU=
+github.com/claygod/PiHex v0.0.0-20200916193129-5277802bfd7b h1:LmxuKRxYbpulBnhu2ZYLfN92Zs2uitai6s6hpmCIZ1Q=
+github.com/claygod/PiHex v0.0.0-20200916193129-5277802bfd7b/go.mod h1:iQyqZlmS/QK9N12+07jX1OO2xlzguGIE7vDmHh3TX+E=
 github.com/client9/misspell v0.3.4/go.mod h1:qj6jICC3Q7zFZvVWo7KLAzC3yx5G7kyvSDkc90ppPyw=
 github.com/cncf/udpa/go v0.0.0-20191209042840-269d4d468f6f/go.mod h1:M8M6+tZqaGXZJjfX53e64911xZQV5JYwmTeXPW+k8Sc=
 github.com/cockroachdb/datadriven v0.0.0-20190809214429-80d97fb3cbaa h1:OaNxuTZr7kxeODyLWsRMC+OD03aFUH+mW6r2d+MWa5Y=

holder.go

@@ -324,7 +324,7 @@ func (h *Holder) processDeleteInflight() error {
 					}
 					inprocessRowIDs = inprocessRowIDs.Union(row)
 				}
-				DeleteRows(inprocessRowIDs, index, shard)
+				DeleteRows(context.Background(), inprocessRowIDs, index, shard)
 			}
 		}
 	}

rbf.go

@@ -223,6 +223,74 @@ func (tx *RBFTx) Remove(index, field, view string, shard uint64, a ...uint64) (c
 // which is expensive in practice and only really useful occasionally.
 const sortedParanoia = false
 
+type countResults struct {
+	changeCount int
+	err         error
+}
+
+// RemoveChannel provides a method of streaming in bits or positions and not requiring a large buffer like add and remove
+// the bits are input via the posChanel and the results are returned via the retChannel
+func (tx *RBFTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) {
+	name := rbfName(index, field, view, shard)
+	var lastHi uint64 = math.MaxUint64 // highbits is always less than this starter.
+	var rc *roaring.Container
+	var hi uint64
+	var lo uint16
+	var err error
+	changeCount := 0
+	i := 0
+	for v := range a {
+		hi, lo = highbits(v), lowbits(v)
+		if hi != lastHi {
+			// either first time through, or changed to a different container.
+			// do we need put the last updated container now?
+			if i > 0 {
+				// not first time through, write what we got.
+				if rc == nil || (rc.N() == 0) {
+					err = tx.tx.RemoveContainer(name, lastHi)
+					if err != nil {
+						resChan <- countResults{0, errors.Wrap(err, "failed to remove container")}
+						return
+					}
+				} else {
+					err = tx.tx.PutContainer(name, lastHi, rc)
+					if err != nil {
+						resChan <- countResults{0, errors.Wrap(err, "failed to put container")}
+						return
+					}
+				}
+			}
+			// get the next container
+			rc, err = tx.tx.Container(name, hi)
+			if err != nil {
+				resChan <- countResults{0, errors.Wrap(err, "failed to retrieve container")}
+				return
+			}
+		} // else same container, keep adding bits to rct.
+		chng := false
+		rc, chng = rc.Remove(lo)
+		if chng {
+			changeCount++
+		}
+		lastHi = hi
+		i++
+	}
+	// write the last updates.
+	if rc == nil || rc.N() == 0 {
+		err = tx.tx.RemoveContainer(name, hi)
+		if err != nil {
+			resChan <- countResults{0, errors.Wrap(err, "failed to remove container")}
+			return
+		}
+	} else {
+		err = tx.tx.PutContainer(name, hi, rc)
+		if err != nil {
+			resChan <- countResults{0, errors.Wrap(err, "put to remove container")}
+			return
+		}
+	}
+	resChan <- countResults{changeCount, nil}
+}
 func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove bool, a ...uint64) (changeCount int, err error) {
 	if len(a) == 0 {
 		return 0, nil

rbf/cursor.go

@@ -474,9 +474,11 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
 		writeCellN(buf[:], len(group))
 
 		offset := dataOffset(len(group))
+		X := 0
 		for j, cell := range group {
 			writeLeafCell(buf[:], j, offset, cell)
 			offset += align8(cell.Size())
+			X++
 		}
 
 		if err := c.tx.writePage(buf[:]); err != nil {
@@ -614,7 +616,6 @@ func (c *Cursor) deleteLeafCell(key uint64) (err error) {
 	copy(cells[elem.index:], cells[elem.index+1:])
 	cells[len(cells)-1] = leafCell{}
 	cells = cells[:len(cells)-1]
-
 	// Write cells to page.
 	buf := allocPage()
 	writePageNo(buf[:], elem.pgno)
@@ -626,12 +627,14 @@ func (c *Cursor) deleteLeafCell(key uint64) (err error) {
 		writeLeafCell(buf[:], j, offset, cell)
 		offset += align8(cell.Size())
 	}
+
 	if err := c.tx.writePage(buf[:]); err != nil {
 		return err
 	}
 
 	// Update the parent's reference key if it's changed.
 	if c.stack.top > 0 && oldPageKey != cells[0].Key {
+
 		return c.updateBranchCell(c.stack.top-1, cells[0].Key)
 	}
 	return nil

rbf/db.go

@@ -412,7 +412,6 @@ func (db *DB) checkpoint() (err error) {
 // Close closes the database.
 func (db *DB) Close() (err error) {
 	// TODO(bbj): Add wait group to hang until last Tx is complete.
-
 	// Wait for writer lock.
 	db.rwmu.Lock()
 	defer db.rwmu.Unlock()

rbf/tx.go

@@ -669,6 +669,7 @@ func (tx *Tx) container(name string, key uint64) (*roaring.Container, error) {
 func (tx *Tx) PutContainer(name string, key uint64, ct *roaring.Container) error {
 	tx.mu.Lock()
 	defer tx.mu.Unlock()
+
 	return tx.putContainer(name, key, ct)
 }
 
@@ -725,7 +726,6 @@ func (tx *Tx) removeContainer(name string, key uint64) error {
 	if exact, err := c.Seek(key); err != nil || !exact {
 		return err
 	}
-
 	return c.deleteLeafCell(key)
 }
 
@@ -1125,7 +1125,7 @@ func (tx *Tx) readPage(pgno uint32) (_ []byte, isHeap bool, err error) {
 
 	// Verify page number requested is within current size of database.
 	pageN := readMetaPageN(tx.meta[:])
-	if pgno > pageN {
+	if pgno >= pageN {
 		return nil, false, fmt.Errorf("rbf: page read out of bounds: pgno=%d max=%d", pgno, pageN-1)
 	}
 
@@ -1932,6 +1932,7 @@ func (tx *Tx) Pages(pgnos []uint32) ([]Page, error) {
 // PageInfos returns meta data about all pages in the database.
 func (tx *Tx) PageInfos() ([]PageInfo, error) {
 	var errorList ErrorList
+
 	infos := make([]PageInfo, tx.PageN())
 
 	// Read meta page info.

roaring/filter.go

@@ -584,6 +584,10 @@ type BitmapBitmapFilter struct {
 	callback    func(uint64) error
 }
 
+func (b *BitmapBitmapFilter) SetCallback(cb func(uint64) error) {
+	b.callback = cb
+}
+
 func (b *BitmapBitmapFilter) ConsiderKey(key FilterKey, n int32) FilterResult {
 	pos := key & keyMask
 	if b.containers[pos] == nil || n == 0 {

roaring/roaring.go

@@ -675,6 +675,47 @@ func (b *Bitmap) Intersect(other *Bitmap) *Bitmap {
 	return output
 }
 
+func (b *Bitmap) Hash(hash uint64) uint64 {
+	const (
+		offset = 14695981039346656037
+		prime  = 1099511628211
+	)
+	if hash == 0 {
+		hash = uint64(offset)
+	}
+
+	it, _ := b.Containers.Iterator(0)
+	for it.Next() {
+		ki, _ := it.Value()
+		hash ^= uint64(ki)
+		hash *= prime
+	}
+
+	it, _ = b.Containers.Iterator(0)
+	for it.Next() {
+		_, ci := it.Value()
+		hash ^= 0
+		hash *= prime
+		if ci.N() > 0 {
+			var bytes []byte
+			switch ci.typ() {
+
+			case ContainerArray:
+				bytes = fromArray16(ci.array())
+			case ContainerBitmap:
+				bytes = fromArray64(ci.bitmap())
+			case ContainerRun:
+				bytes = fromInterval16(ci.runs())
+			}
+			for _, b := range bytes {
+				hash ^= uint64(b)
+				hash *= prime
+			}
+		}
+	}
+	return hash
+}
+
 type mutableContainersIterator struct {
 	c Containers
 
@@ -7488,3 +7529,13 @@ func (c *Container) Slice() (r []uint16) {
 	}
 	return r
 }
+
+func fromArray16(a []uint16) []byte {
+	return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*2 : len(a)*2]
+}
+func fromArray64(a []uint64) []byte {
+	return (*[8192]byte)(unsafe.Pointer(&a[0]))[:8192:8192]
+}
+func fromInterval16(a []Interval16) []byte {
+	return (*[8192]byte)(unsafe.Pointer(&a[0]))[: len(a)*4 : len(a)*4]
+}

roaring/roaring_internal_test.go

@@ -4825,3 +4825,26 @@ func TestVariousBitmap(t *testing.T) {
 		t.Fatal("nil AddN should be 0")
 	}
 }
+func TestBitmapHash(t *testing.T) {
+	a, b := NewContainerBitmapN(getFullBitmap(), MaxContainerVal+1), NewContainerBitmapN(getFullBitmap(), MaxContainerVal+1)
+	arr := NewContainerArray([]uint16{1, 2, 3, 5, 8})
+	run := NewContainerRun([]Interval16{{Start: 0, Last: 32}})
+	ba := NewBitmap()
+	bb := NewBitmap()
+	ba.Containers.Put(1, arr)
+	ba.Containers.Put(2, run)
+	ba.Containers.Put(101, a)
+	ba.Containers.Put(102, a)
+
+	bb.Containers.Put(1, arr)
+	bb.Containers.Put(2, run)
+	bb.Containers.Put(101, b)
+	bb.Containers.Put(102, b)
+	if ba.Hash(0) != bb.Hash(0) {
+		t.Fatal("hash should be equal")
+	}
+	bb.Containers.Put(103, b)
+	if ba.Hash(0) == bb.Hash(0) {
+		t.Fatal("hash should be different")
+	}
+}

row.go

@@ -122,6 +122,15 @@ func (r *Row) ToTable() (*pb.TableResponse, error) {
 	return pb.RowsToTable(r, n)
 }
 
+// Hash calculate checksum code be useful in block hash join
+func (r *Row) Hash() uint64 {
+	hash := uint64(0)
+	for i := range r.segments {
+		hash = r.segments[i].data.Hash(hash)
+	}
+	return hash
+}
+
 // ToRows implements the ToRowser interface.
 func (r *Row) ToRows(callback func(*pb.RowResponse) error) error {
 	if len(r.Keys) > 0 {

stattx.go

@@ -159,6 +159,7 @@ const (
 	kOffsetRange
 	kLast // mark the end, always keep this last. The following aren't tracked atm:
 	kType
+	kRemoveChannel
 )
 
 func (k kall) String() string {
@@ -205,6 +206,8 @@ func (k kall) String() string {
 		return "kLast"
 	case kType:
 		return "kType"
+	case kRemoveChannel:
+		return "kRemoveChannel"
 	}
 	vprint.PanicOn(fmt.Sprintf("unknown kall '%v'", int(k)))
 	return ""
@@ -221,6 +224,15 @@ func (c *statTx) NewTxIterator(index, field, view string, shard uint64) *roaring
 	}()
 	return c.b.NewTxIterator(index, field, view, shard)
 }
+func (c *statTx) RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults) {
+	me := kRemoveChannel
+	t0 := time.Now()
+	defer func() {
+		c.stats.add(me, time.Since(t0))
+	}()
+	c.b.RemoveChannel(index, field, view, shard, a, resChan)
+	return
+}
 
 func (c *statTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) {
 	me := kImportRoaringBits

translate.go

@@ -11,6 +11,7 @@ import (
 	"sync"
 
 	"github.com/molecula/featurebase/v3/ingest"
+	"github.com/molecula/featurebase/v3/roaring"
 	"github.com/molecula/featurebase/v3/topology"
 	"github.com/pkg/errors"
 )
@@ -84,6 +85,8 @@ type TranslateStore interface { // TODO: refactor this interface; readonly shoul
 	// It should read from the reader and replace the data store with
 	// the read payload.
 	ReadFrom(io.Reader) (int64, error)
+
+	Delete(records *roaring.Bitmap) (Commitor, error)
 }
 
 // This implements ingest's key translator interface, which differs
@@ -420,6 +423,16 @@ func (s *InMemTranslateStore) SetReadOnly(v bool) {
 	defer s.mu.Unlock()
 	s.readOnly = v
 }
+func (s *InMemTranslateStore) Delete(records *roaring.Bitmap) (Commitor, error) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	for _, id := range records.Slice() {
+		key := s.keysByID[id]
+		delete(s.keysByID, id)
+		delete(s.idsByKey, key)
+	}
+	return &NopCommitor{}, nil
+}
 
 // FindKeys looks up the ID for each key.
 // Keys are not created if they do not exist.

tx.go

@@ -133,6 +133,7 @@ type Tx interface {
 	GetSortedFieldViewList(idx *Index, shard uint64) (fvs []txkey.FieldView, err error)
 
 	GetFieldSizeBytes(index, field string) (uint64, error)
+	RemoveChannel(index, field, view string, shard uint64, a chan uint64, resChan chan countResults)
 }
 
 // GenericApplyFilter implements ApplyFilter in terms of tx.ContainerIterator,

tx_test.go

@@ -242,5 +242,4 @@ func TestAPI_ImportAtomicRecord(t *testing.T) {
 	if iraBit {
 		PanicOn("IRA bit should have been cleared")
 	}
-
 }