mirror of
https://github.com/featurebasedb/featurebase.git
synced 2026-09-10 07:01:01 +00:00
Implement pilosa.Tx for RBF
This commit is contained in:
Ben Johnson
parent
4cdf62ab89
commit
64de208170
22 changed files with 882 additions and 170 deletions
|
|
@ -165,6 +165,8 @@ func TestAPI_ImportColumnAttrs(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestAPI_Import(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
c := test.MustRunCluster(t, 2,
|
||||
[]server.CommandOption{
|
||||
server.OptCommandServerOptions(
|
||||
|
|
@ -276,6 +278,8 @@ func TestAPI_Import(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestAPI_ImportValue(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
c := test.MustRunCluster(t, 2,
|
||||
[]server.CommandOption{
|
||||
server.OptCommandServerOptions(
|
||||
|
|
|
|||
|
|
@ -276,7 +276,6 @@ func badgerPath(path string) string {
|
|||
// the existing instance. This insures only one badgerDB
|
||||
// per bpath in this pilosa node.
|
||||
func (r *badgerRegistrar) openBadgerDBWrapper(bpath string) (*BadgerDBWrapper, error) {
|
||||
|
||||
// now that newTxFactory can call us directly, we might not
|
||||
// have the -badgerdb suffix.
|
||||
if !strings.HasSuffix(bpath, "-badgerdb") {
|
||||
|
|
|
|||
|
|
@ -3624,6 +3624,8 @@ func TestExecutor_Execute_FieldValue(t *testing.T) {
|
|||
|
||||
// Ensure an all query can be executed.
|
||||
func TestExecutor_Execute_All(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
t.Run("ColumnID", func(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
|
|
@ -4026,7 +4028,6 @@ func TestExecutor_Execute_ClearRow(t *testing.T) {
|
|||
|
||||
// Ensure a row can be set.
|
||||
func TestExecutor_Execute_SetRow(t *testing.T) {
|
||||
|
||||
t.Run("Set_NewRow", func(t *testing.T) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
|
|
@ -4385,6 +4386,8 @@ func TestExecutor_Execute_Query_Error(t *testing.T) {
|
|||
}
|
||||
|
||||
func TestExecutor_GroupByStrings(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
c.CreateField(t, "istring", pilosa.IndexOptions{Keys: true}, "generals", pilosa.OptFieldKeys())
|
||||
|
|
@ -4841,6 +4844,8 @@ func sameStringSlice(x, y []string) bool {
|
|||
}
|
||||
|
||||
func TestExecutor_Execute_GroupBy(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
groupByTest := func(t *testing.T, clusterSize int) {
|
||||
c := test.MustRunCluster(t, 1)
|
||||
defer c.Close()
|
||||
|
|
|
|||
|
|
@ -3025,13 +3025,15 @@ func (f *fragment) rows(ctx context.Context, tx Tx, start uint64, filters ...row
|
|||
|
||||
// unprotectedRows calls rows without grabbing the mutex.
|
||||
func (f *fragment) unprotectedRows(ctx context.Context, tx Tx, start uint64, filters ...rowFilter) ([]uint64, error) {
|
||||
rows := make([]uint64, 0)
|
||||
startKey := rowToKey(start)
|
||||
i, _, err := tx.ContainerIterator(f.index, f.field, f.view, f.shard, startKey)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if i == nil {
|
||||
return rows, nil
|
||||
}
|
||||
defer i.Close() // must close iterators allocated on a Tx
|
||||
rows := make([]uint64, 0)
|
||||
var lastRow uint64 = math.MaxUint64
|
||||
|
||||
// Loop over the existing containers.
|
||||
|
|
|
|||
|
|
@ -1786,6 +1786,8 @@ func TestFragment_LRUCache_Persistence(t *testing.T) {
|
|||
|
||||
// Ensure a fragment's cache can be persisted between restarts.
|
||||
func TestFragment_RankCache_Persistence(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
index := mustOpenIndex(IndexOptions{})
|
||||
defer index.Close()
|
||||
|
||||
|
|
@ -1847,6 +1849,8 @@ func TestFragment_RankCache_Persistence(t *testing.T) {
|
|||
|
||||
// Ensure a fragment can be copied to another fragment.
|
||||
func TestFragment_WriteTo_ReadFrom(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
f0, idx := mustOpenFragment("i", "f", viewStandard, 0, "")
|
||||
_ = idx
|
||||
defer f0.Clean(t)
|
||||
|
|
@ -4196,6 +4200,8 @@ func TestFragmentRowIterator(t *testing.T) {
|
|||
})
|
||||
|
||||
t.Run("skipped rows wrapped", func(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
f, idx := mustOpenFragment("i", "f", "v", 0, CacheTypeRanked)
|
||||
_ = idx
|
||||
defer f.Clean(t)
|
||||
|
|
@ -4396,6 +4402,8 @@ func TestFragmentRowIterator_WithTxCommit(t *testing.T) {
|
|||
row, id, _, wrapped, err := iter.Next()
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
} else if row == nil {
|
||||
t.Fatal("expected row")
|
||||
}
|
||||
if id != i%8 {
|
||||
t.Errorf("expected row %d but got %d", i%8, id)
|
||||
|
|
@ -5306,6 +5314,7 @@ func check(t *testing.T, tx Tx, f *fragment, exp map[uint64]map[uint64]struct{})
|
|||
}
|
||||
|
||||
func TestImportValueConcurrent(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
f, idx := mustOpenBSIFragment("i", "f", viewBSIGroupPrefix+"foo", 0)
|
||||
switch idx.Txf.TxType() {
|
||||
|
|
@ -5631,3 +5640,9 @@ func TestFragment_Bug_Q2DoubleDelete(t *testing.T) {
|
|||
t.Fatalf("expected nothing got %v", res)
|
||||
}
|
||||
}
|
||||
|
||||
func skipForRBF(tb testing.TB) {
|
||||
if os.Getenv("PILOSA_TXSRC") == "rbf" {
|
||||
tb.Skip("skip for RBF")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -33,6 +33,7 @@ import (
|
|||
|
||||
func TestHolder_Open(t *testing.T) {
|
||||
skipForBadger := os.Getenv("PILOSA_TXSRC") == "badger"
|
||||
skipForRBF := os.Getenv("PILOSA_TXSRC") == "rbf"
|
||||
|
||||
t.Run("ErrIndexName", func(t *testing.T) {
|
||||
h := test.MustOpenHolder()
|
||||
|
|
@ -169,6 +170,8 @@ func TestHolder_Open(t *testing.T) {
|
|||
t.Run("ErrFragmentStoragePermission", func(t *testing.T) {
|
||||
if skipForBadger {
|
||||
t.Skip("skipping for badger")
|
||||
} else if skipForRBF {
|
||||
t.Skip("skipping for rbf")
|
||||
}
|
||||
if os.Geteuid() == 0 {
|
||||
t.Skip("Skipping permissions test since user is root.")
|
||||
|
|
@ -208,6 +211,8 @@ func TestHolder_Open(t *testing.T) {
|
|||
t.Run("ErrFragmentStorageCorrupt", func(t *testing.T) {
|
||||
if skipForBadger {
|
||||
t.Skip("skipping for badger")
|
||||
} else if skipForRBF {
|
||||
t.Skip("skipping for rbf")
|
||||
}
|
||||
|
||||
h := test.MustOpenHolder()
|
||||
|
|
@ -244,6 +249,8 @@ func TestHolder_Open(t *testing.T) {
|
|||
t.Run("ErrFragmentStorageRecoverable", func(t *testing.T) {
|
||||
if skipForBadger {
|
||||
t.Skip("skipping for badger")
|
||||
} else if skipForRBF {
|
||||
t.Skip("skipping for rbf")
|
||||
}
|
||||
|
||||
h := test.MustOpenHolder()
|
||||
|
|
@ -410,6 +417,8 @@ func TestHolder_HasData(t *testing.T) {
|
|||
|
||||
// Ensure holder can delete an index and its underlying files.
|
||||
func TestHolder_DeleteIndex(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
hldr := test.MustOpenHolder()
|
||||
defer hldr.Close()
|
||||
|
||||
|
|
@ -705,6 +714,8 @@ func TestHolderSyncer_TimeQuantum(t *testing.T) {
|
|||
|
||||
// Ensure holder can sync integer views with a remote holder.
|
||||
func TestHolderSyncer_IntField(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
t.Run("BasicSync", func(t *testing.T) {
|
||||
c := test.MustNewCluster(t, 2)
|
||||
c[0].Config.Cluster.ReplicaN = 2
|
||||
|
|
|
|||
|
|
@ -86,6 +86,8 @@ func forceSnapshotsCheckMapping(t *testing.T) {
|
|||
// in newGeneration in generation.go. So this is probably useless but it's
|
||||
// a failure mode we've been bitten by once...
|
||||
func TestMmapBehavior(t *testing.T) {
|
||||
skipForRBF(t)
|
||||
|
||||
var changed bool
|
||||
var original uint64
|
||||
defer func() {
|
||||
|
|
|
|||
|
|
@ -15,6 +15,7 @@
|
|||
package pilosa_test
|
||||
|
||||
import (
|
||||
"os"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
|
|
@ -55,3 +56,9 @@ func TestAddressWithDefaults(t *testing.T) {
|
|||
}
|
||||
}
|
||||
}
|
||||
|
||||
func skipForRBF(tb testing.TB) {
|
||||
if os.Getenv("PILOSA_TXSRC") == "rbf" {
|
||||
tb.Skip("skip for RBF")
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -84,7 +84,7 @@ func runAdd(runs []roaring.Interval16, v uint16) ([]roaring.Interval16, bool) {
|
|||
}
|
||||
return runs, true
|
||||
}
|
||||
func checkRun(runs []roaring.Interval16, key uint64) leafCell {
|
||||
func checkRun(runs []roaring.Interval16, bitN int, key uint64) leafCell {
|
||||
if len(runs) >= RLEMaxSize {
|
||||
//convertToBitmap
|
||||
bitmap := make([]uint64, BitmapN)
|
||||
|
|
@ -123,9 +123,9 @@ func checkRun(runs []roaring.Interval16, key uint64) leafCell {
|
|||
n += popcount(v)
|
||||
}
|
||||
|
||||
return leafCell{Key: key, N: int(n), Type: ContainerTypeBitmap, Data: fromArray64(bitmap)}
|
||||
return leafCell{Key: key, N: int(n), BitN: int(n), Type: ContainerTypeBitmap, Data: fromArray64(bitmap)}
|
||||
}
|
||||
return leafCell{Key: key, N: len(runs), Type: ContainerTypeRLE, Data: fromInterval16(runs)}
|
||||
return leafCell{Key: key, N: len(runs), BitN: int(bitN + 1), Type: ContainerTypeRLE, Data: fromInterval16(runs)}
|
||||
}
|
||||
|
||||
// Add sets a bit on the underlying bitmap.
|
||||
|
|
@ -136,7 +136,7 @@ func (c *Cursor) Add(v uint64) (changed bool, err error) {
|
|||
if exact, err := c.Seek(hi); err != nil {
|
||||
return false, err
|
||||
} else if !exact {
|
||||
return true, c.putLeafCell(leafCell{Key: hi, Type: ContainerTypeArray, N: 1, Data: fromArray16([]uint16{lo})})
|
||||
return true, c.putLeafCell(leafCell{Key: hi, Type: ContainerTypeArray, N: 1, BitN: 1, Data: fromArray16([]uint16{lo})})
|
||||
}
|
||||
|
||||
// If the container exists and bit is not set then update the page.
|
||||
|
|
@ -155,7 +155,7 @@ func (c *Cursor) Add(v uint64) (changed bool, err error) {
|
|||
copy(other, a[:i])
|
||||
other[i] = lo
|
||||
copy(other[i+1:], a[i:])
|
||||
return true, c.putLeafCell(leafCell{Key: cell.Key, Type: ContainerTypeArray, N: len(other), Data: fromArray16(other)})
|
||||
return true, c.putLeafCell(leafCell{Key: cell.Key, Type: ContainerTypeArray, N: len(other), BitN: cell.BitN + 1, Data: fromArray16(other)})
|
||||
|
||||
case ContainerTypeRLE:
|
||||
runs := toInterval16(cell.Data)
|
||||
|
|
@ -163,7 +163,7 @@ func (c *Cursor) Add(v uint64) (changed bool, err error) {
|
|||
copy(c.rle[:], runs)
|
||||
run, added := runAdd(c.rle[:len(runs)], lo)
|
||||
if added {
|
||||
leaf := checkRun(run, cell.Key)
|
||||
leaf := checkRun(run, cell.BitN, cell.Key)
|
||||
return true, c.putLeafCell(leaf)
|
||||
}
|
||||
return false, nil
|
||||
|
|
@ -185,6 +185,8 @@ func (c *Cursor) Add(v uint64) (changed bool, err error) {
|
|||
if err := c.tx.writeBitmapPage(pgno, fromArray64(a)); err != nil {
|
||||
return false, err
|
||||
}
|
||||
// TODO(bbj): Update parent cell with new BitN.
|
||||
|
||||
return true, nil
|
||||
default:
|
||||
return false, fmt.Errorf("rbf.Cursor.Add(): invalid container type: %d", cell.Type)
|
||||
|
|
@ -220,7 +222,7 @@ func (c *Cursor) Remove(v uint64) (changed bool, err error) {
|
|||
other := make([]uint16, len(a)-1)
|
||||
copy(other[:i], a[:i])
|
||||
copy(other[i:], a[i+1:])
|
||||
return true, c.putLeafCell(leafCell{Key: cell.Key, Type: ContainerTypeArray, N: len(other), Data: fromArray16(other)})
|
||||
return true, c.putLeafCell(leafCell{Key: cell.Key, Type: ContainerTypeArray, N: len(other), BitN: cell.BitN - 1, Data: fromArray16(other)})
|
||||
|
||||
case ContainerTypeRLE:
|
||||
r := toInterval16(cell.Data)
|
||||
|
|
@ -265,6 +267,8 @@ func (c *Cursor) Remove(v uint64) (changed bool, err error) {
|
|||
if err := c.tx.writeBitmapPage(pgno, fromArray64(a)); err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
// TODO(bbj): Update parent cell to decrement BitN.
|
||||
return true, nil
|
||||
default:
|
||||
return false, fmt.Errorf("rbf.Cursor.Add(): invalid container type: %d", cell.Type)
|
||||
|
|
@ -356,7 +360,10 @@ func (c *Cursor) putLeafCell(in leafCell) (err error) {
|
|||
}
|
||||
in.Data = fromArray64(a)
|
||||
cell.Type = ContainerTypeBitmapPtr
|
||||
bitmapPgno, _ := c.tx.allocate()
|
||||
bitmapPgno, err := c.tx.allocate()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cell.Data = fromPgno(bitmapPgno)
|
||||
}
|
||||
|
||||
|
|
@ -812,6 +819,7 @@ func (c *Cursor) Seek(key uint64) (exact bool, err error) {
|
|||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
switch typ := readFlags(buf); typ {
|
||||
case PageTypeBranch:
|
||||
n := readCellN(buf)
|
||||
|
|
@ -1089,6 +1097,7 @@ func (c *Cursor) goNextPage() error {
|
|||
func ConvertToLeafArgs(key uint64, c *roaring.Container) (result leafCell) {
|
||||
result.Key = key
|
||||
result.N = int(c.N())
|
||||
result.BitN = int(c.N())
|
||||
result.Type = ContainerTypeNone
|
||||
if c.N() == 0 {
|
||||
return
|
||||
|
|
@ -1136,7 +1145,7 @@ func (c *Cursor) merge(key uint64, data *roaring.Container) (bool, error) {
|
|||
if err != nil {
|
||||
return false, errors.Wrap(err, "cursor.merge")
|
||||
}
|
||||
container = roaring.NewContainerBitmap(cell.N, d)
|
||||
container = roaring.NewContainerBitmap(cell.BitN, d)
|
||||
case ContainerTypeRLE:
|
||||
d := toInterval16(cell.Data)
|
||||
container = roaring.NewContainerRun(d)
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ func TestCursor_FirstNext(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -95,7 +95,7 @@ func TestCursor_FirstNext_Quick(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
// Insert values in random order.
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
|
|
@ -153,7 +153,7 @@ func TestCursor_LastPrev(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -217,7 +217,7 @@ func TestCursor_LastPrev_Quick(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
// Insert values in random order.
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
|
|
@ -276,7 +276,7 @@ func TestCursor_Union(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -322,7 +322,7 @@ func TestCursor_Union(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
values := GenerateValues(rand, 10000)
|
||||
rows := ToRows(values)
|
||||
|
||||
|
|
@ -356,7 +356,7 @@ func TestCursor_Intersect(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
row := make([]uint64, rbf.ShardWidth/64)
|
||||
|
||||
|
|
@ -403,7 +403,7 @@ func TestCursor_Intersect(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
values := GenerateValues(rand, rand.Intn(10000))
|
||||
rows := ToRows(values)
|
||||
|
||||
|
|
@ -447,7 +447,7 @@ func TestCursor_AddRoaring(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -466,7 +466,7 @@ func TestCursor_AddRoaring(t *testing.T) {
|
|||
return bm
|
||||
}(),
|
||||
wantChanged: false,
|
||||
wantErr: true},
|
||||
wantErr: false},
|
||||
{
|
||||
name: "initial Array",
|
||||
fieldview: "x",
|
||||
|
|
@ -614,7 +614,7 @@ func TestCursor_RLETesting(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
//setup RLE
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -696,10 +696,10 @@ func TestCursor_RLETesting(t *testing.T) {
|
|||
}
|
||||
for _, tt := range tests {
|
||||
t.Run(tt.name, func(t *testing.T) {
|
||||
changed, err := tx.Add("x", tt.args...)
|
||||
changeCount, err := tx.Add("x", tt.args...)
|
||||
if tt.wantErr && err == nil {
|
||||
t.Errorf("No Error %v", err)
|
||||
} else if tt.wantChanged && !changed {
|
||||
} else if tt.wantChanged && changeCount == 0 {
|
||||
t.Errorf("No Change %v", err)
|
||||
} else if err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -734,7 +734,7 @@ func TestCursor_RLEConversion(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
//setup RLE with full container
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -840,7 +840,7 @@ func TestCursor_UpdateBranchCells(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -914,7 +914,7 @@ func TestCursor_SplitBranchCells(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -964,7 +964,7 @@ func TestCursor_RemoveCells(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -1006,7 +1006,7 @@ func TestCursor_PlayContainer(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -1041,7 +1041,7 @@ func TestCursor_OneBitmap(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -1075,7 +1075,7 @@ func TestCursor_GenerateAll(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -93,10 +93,11 @@ func (c *Cursor) Dump(name string) {
|
|||
fmt.Fprintf(bufStdout, "\n}")
|
||||
bufStdout.Flush()
|
||||
}
|
||||
func (c *Cursor) Row(rowID uint64) (*roaring.Bitmap, error) {
|
||||
|
||||
func (c *Cursor) Row(shard, rowID uint64) (*roaring.Bitmap, error) {
|
||||
base := rowID * ShardWidth
|
||||
|
||||
offset := uint64(c.tx.db.Shard * ShardWidth)
|
||||
offset := uint64(shard * ShardWidth)
|
||||
off := highbits(offset)
|
||||
hi0, hi1 := highbits(base), highbits((rowID+1)*ShardWidth)
|
||||
c.stack.index = 0
|
||||
|
|
|
|||
15
rbf/db.go
15
rbf/db.go
|
|
@ -54,27 +54,18 @@ type DB struct {
|
|||
|
||||
// The maximum allowed database size. Required by mmap.
|
||||
MaxSize int64
|
||||
Shard int
|
||||
}
|
||||
|
||||
// NewDB returns a new instance of DB.
|
||||
func NewDB(path string) *DB {
|
||||
return NewDBWithShard(path, 0)
|
||||
}
|
||||
func NewDBWithShard(path string, shard int) *DB {
|
||||
return &DB{
|
||||
txs: make(map[*Tx]struct{}),
|
||||
pageMap: immutable.NewMap(&uint32Hasher{}),
|
||||
Path: path,
|
||||
MaxSize: DefaultMaxSize,
|
||||
Shard: shard,
|
||||
}
|
||||
}
|
||||
|
||||
func (db *DB) DeleteFragment(index, field, view string, shard uint64, frag interface{}) error {
|
||||
panic("TODO: implement rbf.DB.DeleteFragment")
|
||||
}
|
||||
|
||||
// DataPath returns the path to the data file for the DB.
|
||||
func (db *DB) DataPath() string {
|
||||
return filepath.Join(db.Path, "data")
|
||||
|
|
@ -101,7 +92,7 @@ func (db *DB) Open() (err error) {
|
|||
db.mu.Lock()
|
||||
defer db.mu.Unlock()
|
||||
|
||||
if err := os.MkdirAll(filepath.Dir(db.Path), 0755); err != nil {
|
||||
if err := os.MkdirAll(db.Path, 0755); err != nil {
|
||||
return err
|
||||
} else if db.file, err = os.OpenFile(db.DataPath(), os.O_WRONLY|os.O_CREATE, 0666); err != nil {
|
||||
return fmt.Errorf("open file: %w", err)
|
||||
|
|
@ -573,7 +564,7 @@ func (db *DB) Begin(writable bool) (_ *Tx, err error) {
|
|||
// This page is only written at the end of a dirty transaction.
|
||||
page, err := db.readPage(db.pageMap, 0)
|
||||
if err != nil {
|
||||
_ = tx.Rollback()
|
||||
tx.Rollback()
|
||||
return nil, err
|
||||
}
|
||||
copy(tx.meta[:], page)
|
||||
|
|
@ -617,7 +608,7 @@ func (db *DB) Check() error {
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
defer tx.Rollback()
|
||||
return tx.Check()
|
||||
}
|
||||
|
||||
|
|
|
|||
|
|
@ -119,14 +119,13 @@ func TestDB_Recovery(t *testing.T) {
|
|||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
if exists, err := tx.Contains("x", uint64(len(a))); exists || err != nil {
|
||||
t.Fatalf("Contains()=<%v,%#v>", exists, err)
|
||||
} else if exists, err := tx.Contains("x", uint64(len(a)-1)); !exists || err != nil {
|
||||
t.Fatalf("Contains()=<%v,%#v>", exists, err)
|
||||
} else if err := tx.Rollback(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tx.Rollback()
|
||||
})
|
||||
}
|
||||
|
|
|
|||
69
rbf/rbf.go
69
rbf/rbf.go
|
|
@ -21,8 +21,11 @@ import (
|
|||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"os"
|
||||
"unsafe"
|
||||
|
||||
"github.com/pilosa/pilosa/v2/roaring"
|
||||
"github.com/pilosa/pilosa/v2/shardwidth"
|
||||
)
|
||||
|
||||
|
|
@ -81,6 +84,8 @@ var (
|
|||
ErrTxClosed = errors.New("transaction closed")
|
||||
ErrTxNotWritable = errors.New("transaction not writable")
|
||||
ErrBitmapNameRequired = errors.New("bitmap name required")
|
||||
ErrBitmapNotFound = errors.New("bitmap not found")
|
||||
ErrBitmapExists = errors.New("bitmap already exists")
|
||||
)
|
||||
|
||||
// Debug is just a temporary flag used for debugging.
|
||||
|
|
@ -259,6 +264,7 @@ type leafCell struct {
|
|||
Key uint64
|
||||
Type int
|
||||
N int
|
||||
BitN int
|
||||
Data []byte
|
||||
}
|
||||
|
||||
|
|
@ -361,6 +367,49 @@ func (c *leafCell) firstValue() uint16 {
|
|||
}
|
||||
}
|
||||
|
||||
// lastValue the last value from the container.
|
||||
func (c *leafCell) lastValue() uint16 {
|
||||
switch c.Type {
|
||||
case ContainerTypeArray:
|
||||
a := toArray16(c.Data)
|
||||
return a[len(a)-1]
|
||||
case ContainerTypeRLE:
|
||||
r := toInterval16(c.Data)
|
||||
return r[len(r)-1].Last
|
||||
case ContainerTypeBitmap:
|
||||
a := toArray64(c.Data)
|
||||
for i := len(a) - 1; i >= 0; i-- {
|
||||
for j := 63; j >= 0; j-- {
|
||||
if a[i]&(1<<j) != 0 {
|
||||
return (uint16(i) * 64) + uint16(j)
|
||||
}
|
||||
}
|
||||
}
|
||||
panic(fmt.Sprintf("rbf.leafCell.firstValue(): no values set in bitmap container: key=%d", c.Key))
|
||||
default:
|
||||
panic(fmt.Sprintf("invalid container type: %d", c.Type))
|
||||
}
|
||||
}
|
||||
|
||||
// countRange returns the bit count within the given range.
|
||||
func (c *leafCell) countRange(start, end uint16) (n int) {
|
||||
// If the full range is being queried, simply use the precalculated count.
|
||||
if start == 0 && end == math.MaxUint16 {
|
||||
return c.BitN
|
||||
}
|
||||
|
||||
switch c.Type {
|
||||
case ContainerTypeArray:
|
||||
return int(roaring.ArrayCountRange(toArray16(c.Data), int32(start), int32(end)))
|
||||
case ContainerTypeRLE:
|
||||
return int(roaring.RunCountRange(toInterval16(c.Data), int32(start), int32(end)))
|
||||
case ContainerTypeBitmap:
|
||||
return int(roaring.BitmapCountRange(toArray64(c.Data), int32(start), int32(end)))
|
||||
default:
|
||||
panic(fmt.Sprintf("invalid container type: %d", c.Type))
|
||||
}
|
||||
}
|
||||
|
||||
func readLeafCellKey(page []byte, i int) uint64 {
|
||||
offset := readCellOffset(page, i)
|
||||
return *(*uint64)(unsafe.Pointer(&page[offset]))
|
||||
|
|
@ -373,7 +422,8 @@ func readLeafCell(page []byte, i int) leafCell {
|
|||
var cell leafCell
|
||||
cell.Key = *(*uint64)(unsafe.Pointer(&buf[0]))
|
||||
cell.Type = int(*(*uint32)(unsafe.Pointer(&buf[8])))
|
||||
cell.N = int(*(*uint32)(unsafe.Pointer(&buf[12])))
|
||||
cell.N = int(*(*uint16)(unsafe.Pointer(&buf[12])))
|
||||
cell.BitN = int(*(*uint16)(unsafe.Pointer(&buf[14])))
|
||||
|
||||
switch cell.Type {
|
||||
case ContainerTypeArray:
|
||||
|
|
@ -410,7 +460,8 @@ func writeLeafCell(page []byte, i, offset int, cell leafCell) {
|
|||
writeCellOffset(page, i, offset)
|
||||
*(*uint64)(unsafe.Pointer(&page[offset])) = cell.Key
|
||||
*(*uint32)(unsafe.Pointer(&page[offset+8])) = uint32(cell.Type)
|
||||
*(*uint32)(unsafe.Pointer(&page[offset+12])) = uint32(cell.N)
|
||||
*(*uint16)(unsafe.Pointer(&page[offset+12])) = uint16(cell.N)
|
||||
*(*uint16)(unsafe.Pointer(&page[offset+14])) = uint16(cell.BitN)
|
||||
assert(offset+16+len(cell.Data) <= PageSize)
|
||||
copy(page[offset+16:], cell.Data)
|
||||
}
|
||||
|
|
@ -486,8 +537,11 @@ func search(n int, f func(int) int) (index int, exact bool) {
|
|||
return i, false
|
||||
}
|
||||
|
||||
/*
|
||||
func pagedumpi(b []byte, indent string, writer io.Writer) {
|
||||
func Pagedump(b []byte, indent string, writer io.Writer) {
|
||||
if writer == nil {
|
||||
writer = os.Stderr
|
||||
}
|
||||
|
||||
pgno := readPageNo(b)
|
||||
if pgno == Magic32() {
|
||||
fmt.Fprintf(writer, "==META\n")
|
||||
|
|
@ -501,6 +555,7 @@ func pagedumpi(b []byte, indent string, writer io.Writer) {
|
|||
// the page alone so this will output !PAGE for bitmap pages & invalid pages.
|
||||
switch {
|
||||
case flags&PageTypeLeaf != 0:
|
||||
fmt.Fprintf(writer, "==LEAF pgno=%d flags=%d n=%d\n", pgno, flags, cellN)
|
||||
for i := 0; i < cellN; i++ {
|
||||
cell := readLeafCell(b, i)
|
||||
switch cell.Type {
|
||||
|
|
@ -525,7 +580,6 @@ func pagedumpi(b []byte, indent string, writer io.Writer) {
|
|||
fmt.Fprintf(writer, "==!PAGE %d flags=%d\n", pgno, flags)
|
||||
}
|
||||
}
|
||||
*/
|
||||
|
||||
func Walk(tx *Tx, pgno uint32, v func(uint32, []*RootRecord)) {
|
||||
for pgno := readMetaRootRecordPageNo(tx.meta[:]); pgno != 0; {
|
||||
|
|
@ -563,3 +617,8 @@ func RowValues(b []uint64) []uint64 {
|
|||
}
|
||||
return a
|
||||
}
|
||||
|
||||
// func caller(skip int) string {
|
||||
// _, file, line, _ := runtime.Caller(skip + 1)
|
||||
// return fmt.Sprintf("%s:%d", file, line)
|
||||
// }
|
||||
|
|
|
|||
|
|
@ -119,14 +119,6 @@ func MustBegin(tb testing.TB, db *rbf.DB, writable bool) *rbf.Tx {
|
|||
return tx
|
||||
}
|
||||
|
||||
// MustRollback rolls back a transaction or fails.
|
||||
func MustRollback(tb testing.TB, tx *rbf.Tx) {
|
||||
tb.Helper()
|
||||
if err := tx.Rollback(); err != nil && err != rbf.ErrTxClosed {
|
||||
tb.Logf("rollback error: %q", err)
|
||||
}
|
||||
}
|
||||
|
||||
// MustAddRandom adds values to a bitmap in a random order.
|
||||
func MustAddRandom(tb testing.TB, rand *rand.Rand, tx *rbf.Tx, name string, values ...uint64) {
|
||||
tb.Helper()
|
||||
|
|
|
|||
548
rbf/tx.go
548
rbf/tx.go
|
|
@ -16,7 +16,10 @@ package rbf
|
|||
import (
|
||||
"fmt"
|
||||
"io"
|
||||
"math"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/benbjohnson/immutable"
|
||||
"github.com/pilosa/pilosa/v2/roaring"
|
||||
|
|
@ -24,6 +27,7 @@ import (
|
|||
|
||||
// Tx represents a transaction.
|
||||
type Tx struct {
|
||||
mu sync.RWMutex
|
||||
db *DB // parent db
|
||||
meta [PageSize]byte // copy of current meta page
|
||||
walID int64 // max WAL ID at start of tx
|
||||
|
|
@ -32,8 +36,16 @@ type Tx struct {
|
|||
dirty bool // if true, changes have been made
|
||||
}
|
||||
|
||||
// Writable returns true if the transaction can mutate data.
|
||||
func (tx *Tx) Writable() bool {
|
||||
return tx.writable
|
||||
}
|
||||
|
||||
// Commit completes the transaction and persists data changes.
|
||||
func (tx *Tx) Commit() error {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return ErrTxClosed
|
||||
}
|
||||
|
|
@ -57,9 +69,14 @@ func (tx *Tx) Commit() error {
|
|||
return tx.db.removeTx(tx)
|
||||
}
|
||||
|
||||
func (tx *Tx) Rollback() error {
|
||||
func (tx *Tx) Rollback() {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
|
||||
// TODO(bbj): Invalidate DB if rollback fails. Possibly attempt reopen?
|
||||
|
||||
if tx.db == nil {
|
||||
return ErrTxClosed
|
||||
return
|
||||
}
|
||||
|
||||
// If any pages have been written, ensure we write a new meta page with
|
||||
|
|
@ -67,29 +84,26 @@ func (tx *Tx) Rollback() error {
|
|||
// discard pages in the transaction during playback of the WAL on open.
|
||||
if tx.dirty {
|
||||
if err := tx.writeMetaPage(MetaPageFlagRollback); err != nil {
|
||||
return err
|
||||
panic(err)
|
||||
} else if err := tx.db.SyncWAL(); err != nil {
|
||||
return err
|
||||
panic(err)
|
||||
}
|
||||
}
|
||||
|
||||
if err := tx.db.checkpoint(); err != nil {
|
||||
return err
|
||||
}
|
||||
_ = tx.db.checkpoint() // TODO: Check error
|
||||
|
||||
// Disconnect transaction from DB.
|
||||
err := tx.db.removeTx(tx)
|
||||
_ = err
|
||||
/*
|
||||
if err != nil {
|
||||
//TODO need to fix this error
|
||||
}
|
||||
*/
|
||||
return nil
|
||||
_ = tx.db.removeTx(tx) // TODO: Check error
|
||||
}
|
||||
|
||||
// Root returns the root page number for a bitmap. Returns 0 if the bitmap does not exist.
|
||||
func (tx *Tx) Root(name string) (uint32, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
return tx.root(name)
|
||||
}
|
||||
|
||||
func (tx *Tx) root(name string) (uint32, error) {
|
||||
records, err := tx.rootRecords()
|
||||
if err != nil {
|
||||
return 0, err
|
||||
|
|
@ -97,14 +111,43 @@ func (tx *Tx) Root(name string) (uint32, error) {
|
|||
|
||||
i := sort.Search(len(records), func(i int) bool { return records[i].Name >= name })
|
||||
if i >= len(records) || records[i].Name != name {
|
||||
return 0, fmt.Errorf("bitmap not found: %q", name)
|
||||
return 0, ErrBitmapNotFound
|
||||
}
|
||||
return records[i].Pgno, nil
|
||||
}
|
||||
|
||||
// BitmapNames returns a list of all bitmap names.
|
||||
func (tx *Tx) BitmapNames() ([]string, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return nil, ErrTxClosed
|
||||
}
|
||||
|
||||
// Read list of root records.
|
||||
records, err := tx.rootRecords()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Convert to a list of strings.
|
||||
names := make([]string, len(records))
|
||||
for i := range records {
|
||||
names[i] = records[i].Name
|
||||
}
|
||||
return names, nil
|
||||
}
|
||||
|
||||
// CreateBitmap creates a new empty bitmap with the given name.
|
||||
// Returns an error if the bitmap already exists.
|
||||
func (tx *Tx) CreateBitmap(name string) error {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
return tx.createBitmap(name)
|
||||
}
|
||||
|
||||
func (tx *Tx) createBitmap(name string) error {
|
||||
if tx.db == nil {
|
||||
return ErrTxClosed
|
||||
} else if !tx.writable {
|
||||
|
|
@ -122,7 +165,7 @@ func (tx *Tx) CreateBitmap(name string) error {
|
|||
// Find btree by name. Exit if already exists.
|
||||
index := sort.Search(len(records), func(i int) bool { return records[i].Name >= name })
|
||||
if index < len(records) && records[index].Name == name {
|
||||
return fmt.Errorf("bitmap already exists: %q", name)
|
||||
return ErrBitmapExists
|
||||
}
|
||||
//fmt.Println("CREATE BITMAP", name, index)
|
||||
|
||||
|
|
@ -153,6 +196,22 @@ func (tx *Tx) CreateBitmap(name string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// CreateBitmapIfNotExists creates a new empty bitmap with the given name.
|
||||
// This is a no-op if the bitmap already exists.
|
||||
func (tx *Tx) CreateBitmapIfNotExists(name string) error {
|
||||
if err := tx.CreateBitmap(name); err != nil && err != ErrBitmapExists {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (tx *Tx) createBitmapIfNotExists(name string) error {
|
||||
if err := tx.createBitmap(name); err != nil && err != ErrBitmapExists {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
/*
|
||||
func dump(r []*RootRecord) {
|
||||
for _, i := range r {
|
||||
|
|
@ -165,6 +224,9 @@ func dump(r []*RootRecord) {
|
|||
// DeleteBitmap removes a bitmap with the given name.
|
||||
// Returns an error if the bitmap does not exist.
|
||||
func (tx *Tx) DeleteBitmap(name string) error {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return ErrTxClosed
|
||||
} else if !tx.writable {
|
||||
|
|
@ -200,9 +262,55 @@ func (tx *Tx) DeleteBitmap(name string) error {
|
|||
return nil
|
||||
}
|
||||
|
||||
// DeleteBitmapsWithPrefix removes all bitmaps with a given prefix.
|
||||
func (tx *Tx) DeleteBitmapsWithPrefix(prefix string) error {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return ErrTxClosed
|
||||
} else if !tx.writable {
|
||||
return ErrTxNotWritable
|
||||
}
|
||||
|
||||
// Read list of root records.
|
||||
records, err := tx.rootRecords()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for i := 0; i < len(records); i++ {
|
||||
record := records[i]
|
||||
|
||||
// Skip bitmaps without matching prefix.
|
||||
if !strings.HasPrefix(record.Name, prefix) {
|
||||
continue
|
||||
}
|
||||
|
||||
// Deallocate all pages in the tree.
|
||||
if err := tx.deallocateTree(record.Pgno); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Delete from record list.
|
||||
records = append(records[:i], records[i+1:]...)
|
||||
i--
|
||||
}
|
||||
|
||||
// Rewrite record pages.
|
||||
if err := tx.writeRootRecordPages(records); err != nil {
|
||||
return fmt.Errorf("write bitmaps: %w", err)
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// RenameBitmap updates the name of an existing bitmap.
|
||||
// Returns an error if the bitmap does not exist.
|
||||
func (tx *Tx) RenameBitmap(oldname, newname string) error {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return ErrTxClosed
|
||||
} else if !tx.writable {
|
||||
|
|
@ -314,78 +422,103 @@ func (tx *Tx) writeRootRecordPages(records []*RootRecord) (err error) {
|
|||
}
|
||||
|
||||
// Add sets a given bit on the bitmap.
|
||||
func (tx *Tx) Add(name string, a ...uint64) (changed bool, err error) {
|
||||
func (tx *Tx) Add(name string, a ...uint64) (changeCount int, err error) {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return false, ErrTxClosed
|
||||
return 0, ErrTxClosed
|
||||
} else if !tx.writable {
|
||||
return false, ErrTxNotWritable
|
||||
return 0, ErrTxNotWritable
|
||||
} else if name == "" {
|
||||
return false, ErrBitmapNameRequired
|
||||
return 0, ErrBitmapNameRequired
|
||||
}
|
||||
|
||||
c, err := tx.Cursor(name)
|
||||
if err := tx.createBitmapIfNotExists(name); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return false, err
|
||||
return 0, err
|
||||
}
|
||||
for _, v := range a {
|
||||
if vchanged, err := c.Add(v); err != nil {
|
||||
return changed, err
|
||||
return changeCount, err
|
||||
} else if vchanged {
|
||||
changed = true
|
||||
changeCount++
|
||||
}
|
||||
}
|
||||
return changed, nil
|
||||
return changeCount, nil
|
||||
}
|
||||
|
||||
// Remove unsets a given bit on the bitmap.
|
||||
func (tx *Tx) Remove(name string, a ...uint64) (changed bool, err error) {
|
||||
func (tx *Tx) Remove(name string, a ...uint64) (changeCount int, err error) {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return false, ErrTxClosed
|
||||
return 0, ErrTxClosed
|
||||
} else if !tx.writable {
|
||||
return false, ErrTxNotWritable
|
||||
return 0, ErrTxNotWritable
|
||||
} else if name == "" {
|
||||
return false, ErrBitmapNameRequired
|
||||
return 0, ErrBitmapNameRequired
|
||||
}
|
||||
|
||||
c, err := tx.Cursor(name)
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return false, err
|
||||
return 0, err
|
||||
} else if c == nil {
|
||||
return 0, nil
|
||||
}
|
||||
for _, v := range a {
|
||||
if vchanged, err := c.Remove(v); err != nil {
|
||||
return changed, err
|
||||
return changeCount, err
|
||||
} else if vchanged {
|
||||
changed = true
|
||||
changeCount++
|
||||
}
|
||||
}
|
||||
return changed, nil
|
||||
return changeCount, nil
|
||||
}
|
||||
|
||||
// Contains returns true if the given bit is set on the bitmap.
|
||||
func (tx *Tx) Contains(name string, v uint64) (bool, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return false, ErrTxClosed
|
||||
} else if name == "" {
|
||||
return false, ErrBitmapNameRequired
|
||||
}
|
||||
|
||||
c, err := tx.Cursor(name)
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return false, err
|
||||
} else if c == nil {
|
||||
return false, nil
|
||||
}
|
||||
return c.Contains(v)
|
||||
}
|
||||
|
||||
// Cursor returns an instance of a cursor this bitmap.
|
||||
func (tx *Tx) Cursor(name string) (*Cursor, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
return tx.cursor(name)
|
||||
}
|
||||
|
||||
func (tx *Tx) cursor(name string) (*Cursor, error) {
|
||||
if tx.db == nil {
|
||||
return nil, ErrTxClosed
|
||||
} else if name == "" {
|
||||
return nil, ErrBitmapNameRequired
|
||||
}
|
||||
|
||||
root, err := tx.Root(name)
|
||||
if err != nil {
|
||||
root, err := tx.root(name)
|
||||
if err == ErrBitmapNotFound {
|
||||
return nil, nil
|
||||
} else if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
|
|
@ -394,8 +527,109 @@ func (tx *Tx) Cursor(name string) (*Cursor, error) {
|
|||
return &c, nil
|
||||
}
|
||||
|
||||
// RoaringBitmap returns a bitmap as a Roaring bitmap.
|
||||
func (tx *Tx) RoaringBitmap(name string) (*roaring.Bitmap, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return nil, ErrTxClosed
|
||||
} else if name == "" {
|
||||
return nil, ErrBitmapNameRequired
|
||||
}
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if c == nil {
|
||||
return roaring.NewSliceBitmap(), nil
|
||||
}
|
||||
|
||||
other := roaring.NewSliceBitmap()
|
||||
if err := c.First(); err == io.EOF {
|
||||
return other, nil
|
||||
} else if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
for {
|
||||
if err := c.Next(); err == io.EOF {
|
||||
return other, nil
|
||||
} else if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
cell := c.cell()
|
||||
other.Containers.Put(cell.Key, toContainer(cell, tx))
|
||||
}
|
||||
}
|
||||
|
||||
// Container returns a Roaring container by key.
|
||||
func (tx *Tx) Container(name string, key uint64) (*roaring.Container, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return nil, ErrTxClosed
|
||||
} else if name == "" {
|
||||
return nil, ErrBitmapNameRequired
|
||||
}
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
} else if c == nil {
|
||||
return nil, err
|
||||
} else if exact, err := c.Seek(key); err != nil || !exact {
|
||||
return nil, err
|
||||
}
|
||||
return toContainer(c.cell(), tx), nil
|
||||
}
|
||||
|
||||
// PutContainer inserts a container into a bitmap. Overwrites if key already exists.
|
||||
func (tx *Tx) PutContainer(name string, key uint64, cont *roaring.Container) error {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
|
||||
cell := ConvertToLeafArgs(key, cont)
|
||||
if cell.BitN == 0 {
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := tx.createBitmapIfNotExists(name); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return err
|
||||
} else if _, err := c.Seek(cell.Key); err != nil {
|
||||
return err
|
||||
}
|
||||
return c.putLeafCell(cell)
|
||||
}
|
||||
|
||||
// RemoveContainer removes a container from the bitmap by key.
|
||||
func (tx *Tx) RemoveContainer(name string, key uint64) error {
|
||||
tx.mu.Lock()
|
||||
defer tx.mu.Unlock()
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return err
|
||||
} else if c == nil {
|
||||
return nil
|
||||
} else if exact, err := c.Seek(key); err != nil || !exact {
|
||||
return err
|
||||
}
|
||||
return c.deleteLeafCell(key)
|
||||
}
|
||||
|
||||
// Check verifies the integrity of the database.
|
||||
func (tx *Tx) Check() error {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
if tx.db == nil {
|
||||
return ErrTxClosed
|
||||
}
|
||||
|
|
@ -677,12 +911,20 @@ func (tx *Tx) writeMetaPage(flag uint32) error {
|
|||
}
|
||||
|
||||
func (tx *Tx) AddRoaring(name string, bm *roaring.Bitmap) (changed bool, err error) {
|
||||
c, err := tx.Cursor(name)
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
if err := tx.createBitmapIfNotExists(name); err != nil {
|
||||
return false, err
|
||||
}
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return false, err
|
||||
}
|
||||
return c.AddRoaring(bm)
|
||||
}
|
||||
|
||||
func (tx *Tx) leafCellBitmap(pgno uint32) (uint32, []uint64, error) {
|
||||
page, err := tx.readPage(pgno)
|
||||
if err != nil {
|
||||
|
|
@ -690,3 +932,231 @@ func (tx *Tx) leafCellBitmap(pgno uint32) (uint32, []uint64, error) {
|
|||
}
|
||||
return pgno, toArray64(page), err
|
||||
}
|
||||
|
||||
func (tx *Tx) ContainerIterator(name string, key uint64) (citer roaring.ContainerIterator, found bool, err error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
// TODO(bbj): Don't return error if bitmap is simply not found?
|
||||
return nil, false, err
|
||||
} else if c == nil {
|
||||
return nil, false, nil
|
||||
} else if err := c.First(); err != nil {
|
||||
return nil, false, err
|
||||
}
|
||||
return &containerIterator{cursor: c}, true, nil
|
||||
}
|
||||
|
||||
func (tx *Tx) ForEach(name string, fn func(i uint64) error) error {
|
||||
return tx.ForEachRange(name, 0, math.MaxUint64, fn)
|
||||
}
|
||||
|
||||
func (tx *Tx) ForEachRange(name string, start, end uint64, fn func(uint64) error) error {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return err
|
||||
} else if c == nil {
|
||||
return nil
|
||||
} else if _, err := c.Seek(highbits(start)); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
for {
|
||||
if err := c.Next(); err == io.EOF {
|
||||
return nil
|
||||
} else if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
switch cell := c.cell(); cell.Type {
|
||||
case ContainerTypeArray:
|
||||
for _, lo := range toArray16(cell.Data) {
|
||||
v := cell.Key<<16 | uint64(lo)
|
||||
if v < start {
|
||||
continue
|
||||
} else if v > end {
|
||||
return nil
|
||||
} else if err := fn(v); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
case ContainerTypeRLE:
|
||||
for _, r := range toInterval16(cell.Data) {
|
||||
for lo := int(r.Start); lo <= int(r.Last); lo++ {
|
||||
v := cell.Key<<16 | uint64(lo)
|
||||
if v < start {
|
||||
continue
|
||||
} else if v > end {
|
||||
return nil
|
||||
} else if err := fn(v); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
case ContainerTypeBitmap:
|
||||
for i, bits := range toArray64(cell.Data) {
|
||||
for j := uint(0); j < 64; j++ {
|
||||
if bits&(1<<j) != 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
v := cell.Key<<16 | (uint64(i) * 64) | uint64(j)
|
||||
if v < start {
|
||||
continue
|
||||
} else if v > end {
|
||||
return nil
|
||||
} else if err := fn(v); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
}
|
||||
default:
|
||||
panic(fmt.Sprintf("invalid container type: %d", cell.Type))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (tx *Tx) Count(name string) (uint64, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
} else if c == nil {
|
||||
return 0, nil
|
||||
} else if err := c.First(); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
var n uint64
|
||||
for {
|
||||
if err := c.Next(); err == io.EOF {
|
||||
break
|
||||
} else if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
n += uint64(c.cell().BitN)
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func (tx *Tx) Max(name string) (uint64, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
} else if c == nil {
|
||||
return 0, nil
|
||||
} else if err := c.Last(); err == io.EOF {
|
||||
return 0, nil
|
||||
} else if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
cell := c.cell()
|
||||
return uint64((cell.Key << 16) | uint64(cell.lastValue())), nil
|
||||
}
|
||||
|
||||
func (tx *Tx) Min(name string) (uint64, bool, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return 0, false, err
|
||||
} else if c == nil {
|
||||
return 0, false, nil
|
||||
} else if err := c.First(); err == io.EOF {
|
||||
return 0, false, nil
|
||||
} else if err != nil {
|
||||
return 0, false, err
|
||||
}
|
||||
|
||||
cell := c.cell()
|
||||
return uint64((cell.Key << 16) | uint64(cell.firstValue())), true, nil
|
||||
}
|
||||
|
||||
func (tx *Tx) UnionInPlace(name string, others ...*roaring.Bitmap) error {
|
||||
panic("TODO")
|
||||
}
|
||||
|
||||
func (tx *Tx) CountRange(name string, start, end uint64) (uint64, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
c, err := tx.cursor(name)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
} else if c == nil {
|
||||
return 0, nil
|
||||
}
|
||||
|
||||
if err := c.First(); err == io.EOF {
|
||||
return 0, nil
|
||||
} else if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
var n uint64
|
||||
for {
|
||||
if err := c.Next(); err == io.EOF {
|
||||
break
|
||||
} else if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
|
||||
cell := c.cell()
|
||||
if cell.Key > highbits(end) {
|
||||
break
|
||||
}
|
||||
|
||||
if cell.Key == highbits(start) {
|
||||
n += uint64(cell.countRange(lowbits(start), math.MaxUint16))
|
||||
} else if cell.Key == highbits(end) {
|
||||
n += uint64(cell.countRange(0, lowbits(end)))
|
||||
} else {
|
||||
n += uint64(cell.BitN)
|
||||
}
|
||||
}
|
||||
return n, nil
|
||||
}
|
||||
|
||||
func (tx *Tx) OffsetRange(name string, offset, start, end uint64) (*roaring.Bitmap, error) {
|
||||
tx.mu.RLock()
|
||||
defer tx.mu.RUnlock()
|
||||
|
||||
b, err := tx.RoaringBitmap(name)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return b.OffsetRange(offset, start, end), nil
|
||||
}
|
||||
|
||||
// containerIterator wraps Cursor to implement roaring.ContainerIterator.
|
||||
type containerIterator struct {
|
||||
cursor *Cursor
|
||||
}
|
||||
|
||||
// Close is a no-op. It exists to implement the roaring.ContainerIterator interface.
|
||||
func (itr *containerIterator) Close() {}
|
||||
|
||||
// Next moves the iterator to the next container.
|
||||
func (itr *containerIterator) Next() bool {
|
||||
err := itr.cursor.Next()
|
||||
return err != nil
|
||||
}
|
||||
|
||||
// Value returns the current key & container.
|
||||
func (itr *containerIterator) Value() (uint64, *roaring.Container) {
|
||||
cell := itr.cursor.cell()
|
||||
return cell.Key, toContainer(cell, itr.cursor.tx)
|
||||
}
|
||||
|
|
|
|||
|
|
@ -29,13 +29,13 @@ func TestTx_CommitRollback(t *testing.T) {
|
|||
defer MustCloseDB(t, db)
|
||||
|
||||
// Create bitmap in transaction but rollback.
|
||||
if tx, err := db.Begin(true); err != nil {
|
||||
tx, err := db.Begin(true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
} else if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if err := tx.Rollback(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tx.Rollback()
|
||||
|
||||
// Create bitmap in transaction again but commit.
|
||||
if tx, err := db.Begin(true); err != nil {
|
||||
|
|
@ -49,8 +49,8 @@ func TestTx_CommitRollback(t *testing.T) {
|
|||
// Create bitmap again but it should fail as it already exists.
|
||||
if tx, err := db.Begin(true); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if err := tx.CreateBitmap("x"); err == nil || err.Error() != `bitmap already exists: "x"` {
|
||||
_ = tx.Rollback()
|
||||
} else if err := tx.CreateBitmap("x"); err == nil || err != rbf.ErrBitmapExists {
|
||||
tx.Rollback()
|
||||
t.Fatal(err)
|
||||
} else if err := tx.Commit(); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -62,13 +62,13 @@ func TestTx_CommitRollback(t *testing.T) {
|
|||
defer func() { MustCloseDB(t, db) }()
|
||||
|
||||
// Create bitmap in transaction but rollback.
|
||||
if tx, err := db.Begin(true); err != nil {
|
||||
tx, err := db.Begin(true)
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
} else if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if err := tx.Rollback(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
tx.Rollback()
|
||||
db = MustReopenDB(t, db)
|
||||
|
||||
// Create bitmap in transaction again but commit.
|
||||
|
|
@ -84,8 +84,8 @@ func TestTx_CommitRollback(t *testing.T) {
|
|||
// Create bitmap again but it should fail as it already exists.
|
||||
if tx, err := db.Begin(true); err != nil {
|
||||
t.Fatal(err)
|
||||
} else if err := tx.CreateBitmap("x"); err == nil || err.Error() != `bitmap already exists: "x"` {
|
||||
_ = tx.Rollback()
|
||||
} else if err := tx.CreateBitmap("x"); err == nil || err != rbf.ErrBitmapExists {
|
||||
tx.Rollback()
|
||||
t.Fatal(err)
|
||||
} else if err := tx.Commit(); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -102,7 +102,7 @@ func TestTx_CommitRollback(t *testing.T) {
|
|||
tx0 := MustBegin(t, db, true)
|
||||
go func() {
|
||||
<-ch0
|
||||
_ = tx0.Rollback()
|
||||
tx0.Rollback()
|
||||
}()
|
||||
|
||||
// Start separate write transaction in different goroutine.
|
||||
|
|
@ -134,7 +134,7 @@ func TestTx_Add(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -167,7 +167,7 @@ func TestTx_DeleteBitmap(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
// Create bitmap & add value.
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
|
|
@ -192,7 +192,7 @@ func TestTx_RenameBitmap(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
// Create bitmap & add value.
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
|
|
@ -226,7 +226,7 @@ func TestTx_Add_Quick(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
values := GenerateValues(rand, 10000)
|
||||
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
|
|
@ -265,7 +265,7 @@ func TestTx_AddRemove_Quick(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
values := GenerateValues(rand, 10000)
|
||||
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
|
|
@ -314,7 +314,7 @@ func TestTx_Multiple_CreateBitmap(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
values := GenerateValues(rand, 2)
|
||||
|
||||
if err := tx.CreateBitmap("x/1"); err != nil {
|
||||
|
|
@ -332,7 +332,7 @@ func TestTx_Multiple_CreateBitmap(t *testing.T) {
|
|||
}
|
||||
|
||||
tx1 := MustBegin(t, db, true)
|
||||
defer func() { _ = tx1.Rollback() }()
|
||||
defer tx1.Rollback()
|
||||
|
||||
if err := tx1.CreateBitmap("x/2"); err != nil {
|
||||
t.Fatal(err)
|
||||
|
|
@ -353,7 +353,7 @@ func TestTx_CursorCrashArray(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -379,7 +379,7 @@ func TestTx_CursorCrashBitmap(t *testing.T) {
|
|||
db := MustOpenDB(t)
|
||||
defer MustCloseDB(t, db)
|
||||
tx := MustBegin(t, db, true)
|
||||
defer MustRollback(t, tx)
|
||||
defer tx.Rollback()
|
||||
if err := tx.CreateBitmap("x"); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
|
@ -418,7 +418,7 @@ func BenchmarkTx_Add(b *testing.B) {
|
|||
db := MustOpenDB(b)
|
||||
defer MustCloseDB(b, db)
|
||||
tx := MustBegin(b, db, true)
|
||||
defer MustRollback(b, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
for _, v := range values {
|
||||
if _, err := tx.Add("x", v); err != nil {
|
||||
|
|
@ -446,7 +446,7 @@ func BenchmarkTx_Contains(b *testing.B) {
|
|||
db := MustOpenDB(b)
|
||||
defer MustCloseDB(b, db)
|
||||
tx := MustBegin(b, db, true)
|
||||
defer MustRollback(b, tx)
|
||||
defer tx.Rollback()
|
||||
|
||||
b.ResetTimer()
|
||||
t := time.Now()
|
||||
|
|
|
|||
|
|
@ -2860,20 +2860,19 @@ func (c *Container) countRange(start, end int32) (n int32) {
|
|||
return 0
|
||||
}
|
||||
if c.isArray() {
|
||||
return c.arrayCountRange(start, end)
|
||||
return ArrayCountRange(c.array(), start, end)
|
||||
} else if c.isRun() {
|
||||
return c.runCountRange(start, end)
|
||||
return RunCountRange(c.runs(), start, end)
|
||||
}
|
||||
return c.bitmapCountRange(start, end)
|
||||
return BitmapCountRange(c.bitmap(), start, end)
|
||||
}
|
||||
|
||||
func (c *Container) arrayCountRange(start, end int32) (n int32) {
|
||||
func ArrayCountRange(array []uint16, start, end int32) (n int32) {
|
||||
if roaringParanoia {
|
||||
if start > end {
|
||||
panic(fmt.Sprintf("counting in range but %v > %v", start, end))
|
||||
}
|
||||
}
|
||||
array := c.array()
|
||||
i := int32(sort.Search(len(array), func(i int) bool { return int32(array[i]) >= start }))
|
||||
for ; i < int32(len(array)); i++ {
|
||||
v := int32(array[i])
|
||||
|
|
@ -2885,7 +2884,7 @@ func (c *Container) arrayCountRange(start, end int32) (n int32) {
|
|||
return n
|
||||
}
|
||||
|
||||
func (c *Container) bitmapCountRange(start, end int32) int32 {
|
||||
func BitmapCountRange(bitmap []uint64, start, end int32) int32 {
|
||||
if roaringParanoia {
|
||||
if start > end {
|
||||
panic(fmt.Sprintf("counting in range but %v > %v", start, end))
|
||||
|
|
@ -2894,7 +2893,6 @@ func (c *Container) bitmapCountRange(start, end int32) int32 {
|
|||
var n uint64
|
||||
i, j := start/64, end/64
|
||||
// Special case when start and end fall in the same word.
|
||||
bitmap := c.bitmap()
|
||||
if i == j {
|
||||
offi, offj := uint(start%64), uint(64-end%64)
|
||||
n += popcount((bitmap[i] >> offi) << (offj + offi))
|
||||
|
|
@ -2921,13 +2919,13 @@ func (c *Container) bitmapCountRange(start, end int32) int32 {
|
|||
return int32(n)
|
||||
}
|
||||
|
||||
func (c *Container) runCountRange(start, end int32) (n int32) {
|
||||
// RunCountRange returns the ranged bit count for RLE pairs.
|
||||
func RunCountRange(runs []Interval16, start, end int32) (n int32) {
|
||||
if roaringParanoia {
|
||||
if start > end {
|
||||
panic(fmt.Sprintf("counting in range but %v > %v", start, end))
|
||||
}
|
||||
}
|
||||
runs := c.runs()
|
||||
for _, iv := range runs {
|
||||
// iv is before range
|
||||
if int32(iv.Last) < start {
|
||||
|
|
@ -3837,12 +3835,12 @@ func (c *Container) check() error {
|
|||
a.Append(fmt.Errorf("array count mismatch: count=%d, n=%d", len(array), c.N()))
|
||||
}
|
||||
} else if c.isRun() {
|
||||
n := c.runCountRange(0, MaxContainerVal+1)
|
||||
n := RunCountRange(c.runs(), 0, MaxContainerVal+1)
|
||||
if n != c.N() {
|
||||
a.Append(fmt.Errorf("run count mismatch: count=%d, n=%d", n, c.N()))
|
||||
}
|
||||
} else if c.isBitmap() {
|
||||
if n := c.bitmapCountRange(0, MaxContainerVal+1); n != c.N() {
|
||||
if n := BitmapCountRange(c.bitmap(), 0, MaxContainerVal+1); n != c.N() {
|
||||
a.Append(fmt.Errorf("bitmap count mismatch: count=%d, n=%d", n, c.N()))
|
||||
}
|
||||
} else {
|
||||
|
|
@ -4052,7 +4050,7 @@ func intersectionCountRunRun(a, b *Container) (n int32) {
|
|||
func intersectionCountBitmapRun(a, b *Container) (n int32) {
|
||||
statsHit("intersectionCount/BitmapRun")
|
||||
for _, iv := range b.runs() {
|
||||
n += a.bitmapCountRange(int32(iv.Start), int32(iv.Last)+1)
|
||||
n += BitmapCountRange(a.bitmap(), int32(iv.Start), int32(iv.Last)+1)
|
||||
}
|
||||
return n
|
||||
}
|
||||
|
|
|
|||
|
|
@ -131,7 +131,7 @@ func TestContainerRunAdd2(t *testing.T) {
|
|||
|
||||
func TestRunCountRange(t *testing.T) {
|
||||
c := NewContainerRun(nil)
|
||||
cnt := c.runCountRange(2, 9)
|
||||
cnt := RunCountRange(c.runs(), 2, 9)
|
||||
if cnt != 0 {
|
||||
t.Fatalf("should get 0 from empty container, but got: %v", cnt)
|
||||
}
|
||||
|
|
@ -139,7 +139,7 @@ func TestRunCountRange(t *testing.T) {
|
|||
c.add(6)
|
||||
c.add(7)
|
||||
|
||||
cnt = c.runCountRange(2, 9)
|
||||
cnt = RunCountRange(c.runs(), 2, 9)
|
||||
if cnt != 3 {
|
||||
t.Fatalf("should get 3 from interval within range, but got: %v", cnt)
|
||||
}
|
||||
|
|
@ -149,52 +149,52 @@ func TestRunCountRange(t *testing.T) {
|
|||
c.add(10)
|
||||
c.add(11)
|
||||
|
||||
cnt = c.runCountRange(4, 8)
|
||||
cnt = RunCountRange(c.runs(), 4, 8)
|
||||
if cnt != 3 {
|
||||
t.Fatalf("should get 3 from range overlaps front of interval, but got: %v", cnt)
|
||||
}
|
||||
|
||||
cnt = c.runCountRange(5, 8)
|
||||
cnt = RunCountRange(c.runs(), 5, 8)
|
||||
if cnt != 3 {
|
||||
t.Fatalf("should get 3 from range within interval, but got: %v", cnt)
|
||||
}
|
||||
|
||||
cnt = c.runCountRange(6, 8)
|
||||
cnt = RunCountRange(c.runs(), 6, 8)
|
||||
if cnt != 2 {
|
||||
t.Fatalf("should get 2 from range within interval, but got: %v", cnt)
|
||||
}
|
||||
|
||||
cnt = c.runCountRange(3, 9)
|
||||
cnt = RunCountRange(c.runs(), 3, 9)
|
||||
if cnt != 4 {
|
||||
t.Fatalf("should get 4 from range overlaps front of interval, but got: %v", cnt)
|
||||
}
|
||||
|
||||
cnt = c.runCountRange(9, 14)
|
||||
cnt = RunCountRange(c.runs(), 9, 14)
|
||||
if cnt != 3 {
|
||||
t.Fatalf("should get 3 from range overlaps back of interval, but got: %v", cnt)
|
||||
}
|
||||
|
||||
cnt = c.runCountRange(8, 10)
|
||||
cnt = RunCountRange(c.runs(), 8, 10)
|
||||
if cnt != 2 {
|
||||
t.Fatalf("should get 2 from range within interval, but got: %v", cnt)
|
||||
}
|
||||
|
||||
cnt = c.runCountRange(8, 11)
|
||||
cnt = RunCountRange(c.runs(), 8, 11)
|
||||
if cnt != 3 {
|
||||
t.Fatalf("should get 3 from range within interval, but got: %v", cnt)
|
||||
}
|
||||
|
||||
cnt = c.runCountRange(8, 12)
|
||||
cnt = RunCountRange(c.runs(), 8, 12)
|
||||
if cnt != 4 {
|
||||
t.Fatalf("should get 4 from range overlaps back of interval, but got: %v", cnt)
|
||||
}
|
||||
|
||||
cnt = c.runCountRange(5, 12)
|
||||
cnt = RunCountRange(c.runs(), 5, 12)
|
||||
if cnt != 7 {
|
||||
t.Fatalf("should get 7 from interval within range, but got: %v", cnt)
|
||||
}
|
||||
|
||||
cnt = c.runCountRange(5, 11)
|
||||
cnt = RunCountRange(c.runs(), 5, 11)
|
||||
if cnt != 6 {
|
||||
t.Fatalf("should get 6 from interval equal to range, but got: %v", cnt)
|
||||
}
|
||||
|
|
@ -203,7 +203,7 @@ func TestRunCountRange(t *testing.T) {
|
|||
c.add(19)
|
||||
c.add(18)
|
||||
|
||||
cnt = c.runCountRange(1, 22)
|
||||
cnt = RunCountRange(c.runs(), 1, 22)
|
||||
if cnt != 10 {
|
||||
t.Fatalf("should get 10 from multiple ranges in interval, but got: %v", cnt)
|
||||
}
|
||||
|
|
@ -211,7 +211,7 @@ func TestRunCountRange(t *testing.T) {
|
|||
c.add(13)
|
||||
c.add(14)
|
||||
|
||||
cnt = c.runCountRange(6, 18)
|
||||
cnt = RunCountRange(c.runs(), 6, 18)
|
||||
if cnt != 9 {
|
||||
t.Fatalf("should get 9 from multiple ranges overlapping both sides, but got: %v", cnt)
|
||||
}
|
||||
|
|
@ -263,7 +263,7 @@ func TestBitmapCountRange(t *testing.T) {
|
|||
|
||||
for i, test := range tests {
|
||||
c.setBitmap(test.bitmap[:])
|
||||
if ret := c.bitmapCountRange(test.start, test.end); ret != test.exp {
|
||||
if ret := BitmapCountRange(c.bitmap(), test.start, test.end); ret != test.exp {
|
||||
t.Fatalf("test #%v count of %v from %v to %v should be %v but got %v", i, test.bitmap, test.start, test.end, test.exp, ret)
|
||||
}
|
||||
}
|
||||
|
|
|
|||
146
tx.go
146
tx.go
|
|
@ -21,8 +21,10 @@ import (
|
|||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
|
||||
"github.com/pilosa/pilosa/v2/rbf"
|
||||
"github.com/pilosa/pilosa/v2/roaring"
|
||||
"github.com/pkg/errors"
|
||||
)
|
||||
|
|
@ -872,3 +874,147 @@ func (tx *RoaringTx) RoaringBitmapReader(index, field, view string, shard uint64
|
|||
r = file
|
||||
return
|
||||
}
|
||||
|
||||
type RBFTx struct {
|
||||
tx *rbf.Tx
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Type() string {
|
||||
return RBFTxn
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Rollback() {
|
||||
tx.tx.Rollback()
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Commit() error {
|
||||
return tx.tx.Commit()
|
||||
}
|
||||
|
||||
func (tx *RBFTx) RoaringBitmap(index, field, view string, shard uint64) (*roaring.Bitmap, error) {
|
||||
return tx.tx.RoaringBitmap(rbfName(field, view, shard))
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Container(index, field, view string, shard uint64, key uint64) (*roaring.Container, error) {
|
||||
return tx.tx.Container(rbfName(field, view, shard), key)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) PutContainer(index, field, view string, shard uint64, key uint64, c *roaring.Container) error {
|
||||
return tx.tx.PutContainer(rbfName(field, view, shard), key, c)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) RemoveContainer(index, field, view string, shard uint64, key uint64) error {
|
||||
return tx.tx.RemoveContainer(rbfName(field, view, shard), key)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Add(index, field, view string, shard uint64, batched bool, a ...uint64) (changeCount int, err error) {
|
||||
return tx.tx.Add(rbfName(field, view, shard), a...)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Remove(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
|
||||
return tx.tx.Remove(rbfName(field, view, shard), a...)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Contains(index, field, view string, shard uint64, v uint64) (exists bool, err error) {
|
||||
return tx.tx.Contains(rbfName(field, view, shard), v)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) ContainerIterator(index, field, view string, shard uint64, key uint64) (citer roaring.ContainerIterator, found bool, err error) {
|
||||
return tx.tx.ContainerIterator(rbfName(field, view, shard), key)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) ForEach(index, field, view string, shard uint64, fn func(i uint64) error) error {
|
||||
return tx.tx.ForEach(rbfName(field, view, shard), fn)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) ForEachRange(index, field, view string, shard uint64, start, end uint64, fn func(uint64) error) error {
|
||||
return tx.tx.ForEachRange(rbfName(field, view, shard), start, end, fn)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Count(index, field, view string, shard uint64) (uint64, error) {
|
||||
return tx.tx.Count(rbfName(field, view, shard))
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Max(index, field, view string, shard uint64) (uint64, error) {
|
||||
return tx.tx.Max(rbfName(field, view, shard))
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Min(index, field, view string, shard uint64) (uint64, bool, error) {
|
||||
return tx.tx.Min(rbfName(field, view, shard))
|
||||
}
|
||||
|
||||
func (tx *RBFTx) UnionInPlace(index, field, view string, shard uint64, others ...*roaring.Bitmap) error {
|
||||
return tx.tx.UnionInPlace(rbfName(field, view, shard), others...)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) CountRange(index, field, view string, shard uint64, start, end uint64) (uint64, error) {
|
||||
return tx.tx.CountRange(rbfName(field, view, shard), start, end)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) OffsetRange(index, field, view string, shard uint64, offset, start, end uint64) (*roaring.Bitmap, error) {
|
||||
return tx.tx.OffsetRange(rbfName(field, view, shard), offset, start, end)
|
||||
}
|
||||
|
||||
func (tx *RBFTx) IncrementOpN(index, field, view string, shard uint64, changedN int) {}
|
||||
|
||||
func (tx *RBFTx) ImportRoaringBits(index, field, view string, shard uint64, rit roaring.RoaringIterator, clear bool, log bool, rowSize uint64, data []byte) (changed int, rowSet map[uint64]int, err error) {
|
||||
// TODO: Implement RBFTX.ImportRoaringBits"
|
||||
return 0, make(map[uint64]int), nil
|
||||
}
|
||||
|
||||
func (tx *RBFTx) RoaringBitmapReader(index, field, view string, shard uint64, fragmentPathForRoaring string) (r io.ReadCloser, sz int64, err error) {
|
||||
panic("TODO: Implement RBFTx.RoaringBitmapReader()")
|
||||
}
|
||||
|
||||
func (tx *RBFTx) SliceOfShards(index, field, view, optionalViewPath string) (sliceOfShards []uint64, err error) {
|
||||
prefix := rbfFieldViewPrefix(field, view)
|
||||
|
||||
names, err := tx.tx.BitmapNames()
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Iterate over shard names and collect shards from matching field/view prefix.
|
||||
for _, name := range names {
|
||||
if !strings.HasPrefix(name, prefix) {
|
||||
continue
|
||||
}
|
||||
|
||||
s := strings.TrimPrefix(name, prefix)
|
||||
shard, err := strconv.ParseUint(s, 10, 64)
|
||||
if err != nil {
|
||||
return nil, errors.Wrap(err, "parse shard id from rbf key")
|
||||
}
|
||||
sliceOfShards = append(sliceOfShards, shard)
|
||||
}
|
||||
return sliceOfShards, nil
|
||||
}
|
||||
|
||||
func (tx *RBFTx) NewTxIterator(index, field, view string, shard uint64) *roaring.Iterator {
|
||||
b, err := tx.RoaringBitmap(index, field, view, shard)
|
||||
panicOn(err)
|
||||
return b.Iterator()
|
||||
}
|
||||
|
||||
func (tx *RBFTx) Pointer() string {
|
||||
return fmt.Sprintf("%p", tx)
|
||||
}
|
||||
|
||||
// Readonly is true if the transaction is not read-and-write, but only doing reads.
|
||||
func (tx *RBFTx) Readonly() bool {
|
||||
return !tx.tx.Writable()
|
||||
}
|
||||
|
||||
func (tx *RBFTx) UseRowCache() bool {
|
||||
return false
|
||||
}
|
||||
|
||||
// rbfName returns a NULL-separated key used for identifying bitmap maps in RBF.
|
||||
func rbfName(field, view string, shard uint64) string {
|
||||
return fmt.Sprintf("%s\x00%s\x00%d", field, view, shard)
|
||||
}
|
||||
|
||||
// rbfFieldViewPrefix returns a NULL-separated prefix for keys in RBF.
|
||||
func rbfFieldViewPrefix(field, view string) string {
|
||||
return fmt.Sprintf("%s\x00%s\x00", field, view)
|
||||
}
|
||||
|
|
|
|||
40
txfactory.go
40
txfactory.go
|
|
@ -81,6 +81,8 @@ func (f *TxFactory) Store() TxStore {
|
|||
// case blueGreenRoaringBadger:
|
||||
}
|
||||
panic(fmt.Sprintf("unknown f.typeOfTx type: '%v'", f.typeOfTx))
|
||||
=======
|
||||
>>>>>>> Implement pilosa.Tx for RBF
|
||||
}
|
||||
*/
|
||||
|
||||
|
|
@ -168,6 +170,7 @@ func NewTxFactory(txsrc string, dir, name string) (f *TxFactory, err error) {
|
|||
typeOfTx: ty,
|
||||
roaringDB: NewRoaringStore(),
|
||||
}
|
||||
|
||||
switch ty {
|
||||
case badgerTxn, blueGreenBadgerRoaring, blueGreenRoaringBadger, blueGreenBadgerRBF, blueGreenRBFBadger:
|
||||
|
||||
|
|
@ -193,10 +196,9 @@ func NewTxFactory(txsrc string, dir, name string) (f *TxFactory, err error) {
|
|||
|
||||
switch ty {
|
||||
case rbfTxn, blueGreenRBFRoaring, blueGreenRoaringRBF, blueGreenBadgerRBF, blueGreenRBFBadger:
|
||||
path := dir + sep + name + ".rbf"
|
||||
f.rbfDB = rbf.NewDB(path)
|
||||
f.rbfDB = rbf.NewDB(filepath.Join(dir, "db.rbf"))
|
||||
if err := f.rbfDB.Open(); err != nil {
|
||||
return nil, errors.Wrap(err, fmt.Sprintf("cannot open rbf db. path='%v'", path))
|
||||
return nil, errors.Wrap(err, "cannot open rbf db")
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -240,8 +242,16 @@ func (f *TxFactory) DeleteFragmentFromStore(index, field, view string, shard uin
|
|||
case badgerTxn:
|
||||
return f.badgerDB.DeleteFragment(index, field, view, shard, frag)
|
||||
case rbfTxn:
|
||||
//return f.rbfDB.DeleteFragment(index, field, view, shard, frag)
|
||||
return nil
|
||||
tx, err := f.rbfDB.Begin(true)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer tx.Rollback()
|
||||
|
||||
if err := tx.DeleteBitmapsWithPrefix(rbfFieldViewPrefix(field, view)); err != nil {
|
||||
return err
|
||||
}
|
||||
return tx.Commit()
|
||||
case blueGreenBadgerRoaring:
|
||||
_ = f.badgerDB.DeleteFragment(index, field, view, shard, frag)
|
||||
return f.roaringDB.DeleteFragment(index, field, view, shard, frag)
|
||||
|
|
@ -263,9 +273,7 @@ func (f *TxFactory) CloseIndex(idx *Index) error {
|
|||
//return f.badgerDB.Close()
|
||||
return nil
|
||||
case rbfTxn:
|
||||
// for same reason as above may not be able to close here.
|
||||
//return f.rbfDB.Close()
|
||||
return nil
|
||||
return f.rbfDB.Close()
|
||||
case blueGreenBadgerRoaring:
|
||||
return nil
|
||||
case blueGreenRoaringBadger:
|
||||
|
|
@ -275,7 +283,6 @@ func (f *TxFactory) CloseIndex(idx *Index) error {
|
|||
}
|
||||
|
||||
func (f *TxFactory) NewTx(o Txo) Tx {
|
||||
|
||||
indexName := ""
|
||||
if o.Index != nil {
|
||||
indexName = o.Index.name
|
||||
|
|
@ -288,14 +295,11 @@ func (f *TxFactory) NewTx(o Txo) Tx {
|
|||
btx := f.badgerDB.NewBadgerTx(o.Write, indexName)
|
||||
return btx
|
||||
case rbfTxn:
|
||||
panic("todo rbfTxn creation")
|
||||
/*
|
||||
rbftx, err := f.rbfDB.Begin(o.Write)
|
||||
if err != nil {
|
||||
errors.Wrap(err, "rbfDB.Begin transaction errored")
|
||||
}
|
||||
return rbftx
|
||||
*/
|
||||
tx, err := f.rbfDB.Begin(o.Write)
|
||||
if err != nil {
|
||||
panic(err) // TODO: Add error return on NewTx()
|
||||
}
|
||||
return &RBFTx{tx: tx}
|
||||
case blueGreenBadgerRoaring:
|
||||
btx := f.badgerDB.NewBadgerTx(o.Write, indexName)
|
||||
rtx := &RoaringTx{write: o.Write, Field: o.Field, Index: o.Index, fragment: o.Fragment}
|
||||
|
|
@ -379,7 +383,6 @@ func (idx *Index) StringifiedBadgerKeys(optionalUseThisTx Tx) string {
|
|||
// index directory, not including the name of the index itself.
|
||||
// The path should not start with the path separator sep ('/' or '\\') rune.
|
||||
func fragmentSpecFromRoaringPath(path string) (field, view string, shard uint64, err error) {
|
||||
|
||||
if len(path) == 0 {
|
||||
err = fmt.Errorf("fragmentSpecFromRoaringPath error: path '%v' too short", path)
|
||||
return
|
||||
|
|
@ -408,7 +411,6 @@ func fragmentSpecFromRoaringPath(path string) (field, view string, shard uint64,
|
|||
}
|
||||
|
||||
func (idx *Index) StringifiedRoaringKeys() (r string) {
|
||||
|
||||
paths, err := listFilesUnderDir(idx.path, false, "", true)
|
||||
panicOn(err)
|
||||
index := idx.name
|
||||
|
|
|
|||
|
|
@ -63,7 +63,7 @@ func (rbc *RBFConverter) Convert(index, field, view string, shard uint64, rb *ro
|
|||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer func() { _ = tx.Rollback() }()
|
||||
defer tx.Rollback()
|
||||
|
||||
name := fmt.Sprintf("%s/%s", field, view)
|
||||
err = tx.CreateBitmap(name)
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue