diff --git a/client/client_it_test.go b/client/client_it_test.go index 1b5ca2884..b9137bee9 100644 --- a/client/client_it_test.go +++ b/client/client_it_test.go @@ -592,7 +592,6 @@ func TestClientAgainstCluster(t *testing.T) { target := []uint64{100} require.Equalf(t, target, resp.Result().Row().Columns, "Row Result Columns") }) - t.Run("StoreQuery", func(t *testing.T) { schema := NewSchema() testIndexStore := schema.Index("test-index-store") diff --git a/rbf/README.md b/rbf/README.md index 4f6efa072..cbc5b74a7 100644 --- a/rbf/README.md +++ b/rbf/README.md @@ -93,12 +93,12 @@ The leaf page contains a series of cells with the header of: [8] highbits [4] flag [4] child count - [*] array or RLE data + [*] array or RLE data or Handle (a pageno) to Bitmap Data -### Bitmap page +### Bitmap Data page -The data for the bitmap page takes up the entire 8KB. +The data for the bitmap data page takes up the entire 8KB. ## Proof of Concept Notes diff --git a/rbf/tx.go b/rbf/tx.go index 812d35738..52cc49366 100644 --- a/rbf/tx.go +++ b/rbf/tx.go @@ -333,7 +333,6 @@ func (tx *Tx) DeleteBitmapsWithPrefix(prefix string) error { if !strings.HasPrefix(name.(string), prefix) { continue } - // Deallocate all pages in the tree. if err := tx.deallocateTree(pgno.(uint32)); err != nil { return err @@ -1065,6 +1064,13 @@ func (tx *Tx) deallocateTree(pgno uint32) error { return tx.freePgno(pgno) case PageTypeLeaf: + for i, n := 0, readCellN(page); i < n; i++ { + if cell := readLeafCell(page, i); cell.Type == ContainerTypeBitmapPtr { + if err := tx.freePgno(toPgno(cell.Data)); err != nil { + return err + } + } + } return tx.freePgno(pgno) default: return fmt.Errorf("rbf.Tx.deallocateTree(): invalid page type: pgno=%d type=%d", pgno, typ) diff --git a/rbf/tx_test.go b/rbf/tx_test.go index e4b68f4b6..8a354a916 100644 --- a/rbf/tx_test.go +++ b/rbf/tx_test.go @@ -684,3 +684,102 @@ func TestTx_CreateBitmap(t *testing.T) { } }) } + +func TestTx_DeleteBitmapsWithPrefix(t *testing.T) { + db := MustOpenDB(t) + defer MustCloseDB(t, db) + prefix := "abc" + bitmapSize := 10000 + // var err error + // create a interleaved set up array and bitmap containers + + bits := make([]uint64, bitmapSize) + x := uint64(1) + for i := 0; i < len(bits); i++ { + bits[i] = x + x = x + 2 + } + ifError := func(err error) { + if err != nil { + t.Fatal(err) + } + } + checkInfos := func() { + tx := MustBegin(t, db, false) + defer tx.Rollback() + infos, err := tx.PageInfos() + ifError(err) + for pgno, info := range infos { + switch info := info.(type) { + case *rbf.MetaPageInfo: + fmt.Printf("%-8d ", pgno) + fmt.Printf("%-10s ", "meta") + fmt.Printf("pageN=%d,walid=%d,rootrec=%d,freelist=%d\n", info.PageN, info.WALID, info.RootRecordPageNo, info.FreelistPageNo) + + case *rbf.RootRecordPageInfo: + fmt.Printf("%-8d ", pgno) + fmt.Printf("%-10s ", "rootrec") + fmt.Printf("next=%d\n", info.Next) + + case *rbf.LeafPageInfo: + fmt.Printf("%-8d ", pgno) + fmt.Printf("%-10s ", "leaf") + fmt.Printf("flags=x%x,celln=%d\n", info.Flags, info.CellN) + + case *rbf.BranchPageInfo: + fmt.Printf("%-8d ", pgno) + fmt.Printf("%-10s ", "branch") + fmt.Printf("flags=x%x,celln=%d\n", info.Flags, info.CellN) + + case *rbf.BitmapPageInfo: + fmt.Printf("%-8d ", pgno) + fmt.Printf("%-10s ", "bitmap") + fmt.Printf("-\n") + + case *rbf.FreePageInfo: + fmt.Printf("%-8d ", pgno) + fmt.Printf("%-10s ", "free") + fmt.Printf("-\n") + + default: + t.Fatal(fmt.Sprintf("unexpected page info type %T", info)) + } + } + + } + populate := func() { + tx := MustBegin(t, db, true) + defer tx.Rollback() + for i := uint64(0); i < 16; i++ { + bm := roaring.NewBitmap() + if i%3 == 0 { + bm.Put(i, roaring.NewContainerBitmap(6144, bits)) + if _, err := tx.AddRoaring(prefix, bm); err != nil { + panic(err) + } + } else { + bm.Put(i, roaring.NewContainerArray([]uint16{2, 4, 5, 7})) + if _, err := tx.AddRoaring(prefix, bm); err != nil { + panic(err) + } + + } + } + ifError(tx.Commit()) + } + + checkInfos() + populate() + checkInfos() + ifError(db.Check()) + + tx := MustBegin(t, db, true) + tx.DeleteBitmapsWithPrefix(prefix) + ifError(tx.Commit()) + ifError(db.Check()) + checkInfos() + populate() + ifError(db.Check()) + checkInfos() + +}