From cf1de78efd27244e16617e8281488961e29a07bd Mon Sep 17 00:00:00 2001 From: Todd Gruben Date: Fri, 25 Feb 2022 10:38:20 -0600 Subject: [PATCH] remove string keys on delete to allow for reuse --- boltdb/translate.go | 91 ++++++++++++++++- boltdb/translate_test.go | 49 ++++++++- catcher.go | 5 + dbshard_internal_test.go | 65 ++++++++++++ delete_test.go | 49 ++++++++- executor.go | 167 ++++++++++++++++++++++--------- go.mod | 1 + go.sum | 2 + holder.go | 2 +- rbf.go | 68 +++++++++++++ rbf/cursor.go | 5 +- rbf/db.go | 1 - rbf/tx.go | 5 +- roaring/filter.go | 4 + roaring/roaring.go | 51 ++++++++++ roaring/roaring_internal_test.go | 23 +++++ row.go | 9 ++ stattx.go | 12 +++ translate.go | 13 +++ tx.go | 1 + tx_test.go | 1 - 21 files changed, 566 insertions(+), 58 deletions(-) diff --git a/boltdb/translate.go b/boltdb/translate.go index 8ff56f7e4..a5a17e962 100644 --- a/boltdb/translate.go +++ b/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, diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index 201971644..ed367c8b7 100644 --- a/boltdb/translate_test.go +++ b/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) diff --git a/catcher.go b/catcher.go index a1f128d65..0a75ac9f7 100644 --- a/catcher.go +++ b/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) } diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index 38f752425..1c173eb9c 100644 --- a/dbshard_internal_test.go +++ b/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()) +} diff --git a/delete_test.go b/delete_test.go index df977513b..63c77be34 100644 --- a/delete_test.go +++ b/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 { diff --git a/executor.go b/executor.go index 7dfc994f2..d9aba15c3 100644 --- a/executor.go +++ b/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) } diff --git a/go.mod b/go.mod index 529f2a29d..f089da554 100644 --- a/go.mod +++ b/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 diff --git a/go.sum b/go.sum index 79fc9f1f6..2e0ea4375 100644 --- a/go.sum +++ b/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= diff --git a/holder.go b/holder.go index 7e7b4c056..4e9030f40 100644 --- a/holder.go +++ b/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) } } } diff --git a/rbf.go b/rbf.go index b79b2eb03..5360958cb 100644 --- a/rbf.go +++ b/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 diff --git a/rbf/cursor.go b/rbf/cursor.go index 41e9e4d4f..c359d723a 100644 --- a/rbf/cursor.go +++ b/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 diff --git a/rbf/db.go b/rbf/db.go index 4076fecca..ec75a6064 100644 --- a/rbf/db.go +++ b/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() diff --git a/rbf/tx.go b/rbf/tx.go index 17647caa3..f02a83783 100644 --- a/rbf/tx.go +++ b/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. diff --git a/roaring/filter.go b/roaring/filter.go index 513f8e8f0..fd1856af4 100644 --- a/roaring/filter.go +++ b/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 { diff --git a/roaring/roaring.go b/roaring/roaring.go index f632aaca4..9f2ef5039 100644 --- a/roaring/roaring.go +++ b/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] +} diff --git a/roaring/roaring_internal_test.go b/roaring/roaring_internal_test.go index 7fc0e5bb1..00534cb69 100644 --- a/roaring/roaring_internal_test.go +++ b/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") + } +} diff --git a/row.go b/row.go index 82c8c1029..1639f84ed 100644 --- a/row.go +++ b/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 { diff --git a/stattx.go b/stattx.go index 4780f70bc..5ce265e90 100644 --- a/stattx.go +++ b/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 diff --git a/translate.go b/translate.go index cc36ebbe2..9d5909f28 100644 --- a/translate.go +++ b/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. diff --git a/tx.go b/tx.go index 65a3e3c3f..70d7a724a 100644 --- a/tx.go +++ b/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, diff --git a/tx_test.go b/tx_test.go index 80d72b51c..117198c94 100644 --- a/tx_test.go +++ b/tx_test.go @@ -242,5 +242,4 @@ func TestAPI_ImportAtomicRecord(t *testing.T) { if iraBit { PanicOn("IRA bit should have been cleared") } - }