scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-07 09:05:55 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1578 from seebs/bitmaaaapMaster

Browse source 
roaring ops log and TxBitmap fixes
This commit is contained in:
seebs 2021-04-21 12:16:47 -05:00 • committed by GitHub
commit 0adc10d2c1
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
14 changed files with 160 additions and 115 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

6
bluegreentx.go
View file

@ -539,7 +539,7 @@ func (c *blueGreenTx) isIn(index, field, view string, shard uint64, ckey uint64)
return
}
func (c *blueGreenTx) Add(index, field, view string, shard uint64, batched bool, a ...uint64) (changeCount int, err error) {
func (c *blueGreenTx) Add(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
c.checker.see(index, field, view, shard)
//vv("blueGreenTx) Add(index=%v, field=%v, view=%v, shard=%v", index, field, view, shard)
defer func() {
@ -554,10 +554,10 @@ func (c *blueGreenTx) Add(index, field, view string, shard uint64, batched bool,
a2 := make([]uint64, len(a))
copy(a2, a)
ach, errA := c.a.Add(index, field, view, shard, batched, a...)
ach, errA := c.a.Add(index, field, view, shard, a...)
_, _ = ach, errA
bch, errB := c.b.Add(index, field, view, shard, batched, a2...)
bch, errB := c.b.Add(index, field, view, shard, a2...)
if !c.o.blueGreenOff {

20
bolt.go
View file

@ -614,28 +614,16 @@ func (tx *BoltTx) RemoveContainer(index, field, view string, shard uint64, ckey
}
// Add sets all the a bits hot in the specified fragment.
func (tx *BoltTx) Add(index, field, view string, shard uint64, batched bool, a ...uint64) (changeCount int, err error) {
return tx.addOrRemove(index, field, view, shard, batched, false, a...)
func (tx *BoltTx) Add(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
return tx.addOrRemove(index, field, view, shard, false, a...)
}
// Remove clears all the specified a bits in the chosen fragment.
func (tx *BoltTx) Remove(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
const batched = false
const remove = true
return tx.addOrRemove(index, field, view, shard, batched, remove, a...)
return tx.addOrRemove(index, field, view, shard, true, a...)
}
func (tx *BoltTx) addOrRemove(index, field, view string, shard uint64, batched, remove bool, a ...uint64) (changeCount int, err error) {
// pure hack to match RoaringTx
defer func() {
if !remove && !batched {
if changeCount > 0 {
changeCount = 1
}
}
}()
func (tx *BoltTx) addOrRemove(index, field, view string, shard uint64, remove bool, a ...uint64) (changeCount int, err error) {
if len(a) == 0 {
return 0, nil
}

28
bolt_test.go
View file

@ -56,7 +56,7 @@ func BoltMustSetBitvalue(dbwrap *BoltWrapper, index, field, view string, shard u
tx, _ := dbwrap.NewTx(writable, index, Txo{})
// add a bit
changed, err := tx.Add(index, field, view, shard, doBatched, putme)
changed, err := tx.Add(index, field, view, shard, putme)
if changed != 1 {
panic("should have 1 bit changed")
}
@ -123,7 +123,7 @@ func TestBolt_DeleteFragment(t *testing.T) {
views := []string{"v1", "v2"}
for _, view := range views {
for _, v := range bits {
changed, err := tx.Add(index, field, view, shard, doBatched, v)
changed, err := tx.Add(index, field, view, shard, v)
if changed <= 0 {
panic("should have changed")
}
@ -239,7 +239,7 @@ func TestBolt_SetBitmap(t *testing.T) {
index, field, view, shard := "i", "f", "v", uint64(0)
tx, _ := dbwrap.NewTx(writable, index, Txo{})
bitvalue := uint64(0)
changed, err := tx.Add(index, field, view, shard, doBatched, bitvalue)
changed, err := tx.Add(index, field, view, shard, bitvalue)
if changed <= 0 {
panic("should have changed")
}
@ -281,14 +281,14 @@ func TestBolt_OffsetRange(t *testing.T) {
tx, _ := dbwrap.NewTx(writable, index, Txo{})
bitvalue := uint64(1 << 20)
changed, err := tx.Add(index, field, view, shard, doBatched, bitvalue)
changed, err := tx.Add(index, field, view, shard, bitvalue)
if changed <= 0 {
panic("should have changed")
}
PanicOn(err)
bitvalue2 := uint64(1<<20 + 1)
changed, err = tx.Add(index, field, view, shard, doBatched, bitvalue2)
changed, err = tx.Add(index, field, view, shard, bitvalue2)
if changed <= 0 {
panic("should have changed")
}
@ -374,7 +374,7 @@ func TestBolt_Count_dense_containers(t *testing.T) {
expected := 0
for i := uint64(0); i < (1<<16)+2; i += 2 {
changed, err := tx.Add(index, field, view, shard, doBatched, i)
changed, err := tx.Add(index, field, view, shard, i)
PanicOn(err)
if changed <= 0 {
panic("wat? should have changed")
@ -420,7 +420,7 @@ func TestBolt_ContainerIterator_on_one_bit(t *testing.T) {
bitvalue := uint64(42)
// add a bit
changed, err := tx.Add(index, field, view, shard, doBatched, bitvalue)
changed, err := tx.Add(index, field, view, shard, bitvalue)
if changed <= 0 {
panic("should have changed")
}
@ -478,7 +478,7 @@ func TestBolt_ContainerIterator_on_one_bit_fail_to_find(t *testing.T) {
searchme := putme + 1
// add a bit
changed, err := tx.Add(index, field, view, shard, doBatched, putme)
changed, err := tx.Add(index, field, view, shard, putme)
if changed <= 0 {
panic("should have changed")
}
@ -533,7 +533,7 @@ func TestBolt_ContainerIterator_empty_iteration_loop(t *testing.T) {
searchme := uint64(1 << 17) // in the next container, key:2
// add a bit
changed, err := tx.Add(index, field, view, shard, doBatched, putme)
changed, err := tx.Add(index, field, view, shard, putme)
if changed <= 0 {
panic("should have changed")
}
@ -582,7 +582,7 @@ func TestBolt_ForEach_on_one_bit(t *testing.T) {
bitvalue := uint64(42)
// add a bit
changed, err := tx.Add(index, field, view, shard, doBatched, bitvalue)
changed, err := tx.Add(index, field, view, shard, bitvalue)
if changed <= 0 {
panic("should have changed")
}
@ -1132,7 +1132,7 @@ func TestBolt_DeleteIndex(t *testing.T) {
bitvalue := uint64(777)
bits := []uint64{0, 3, 1 << 16, 1<<16 + 3, 8 << 16}
for _, v := range bits {
changed, err := tx.Add(index, field, view, shard, doBatched, v)
changed, err := tx.Add(index, field, view, shard, v)
if changed <= 0 {
panic("should have changed")
}
@ -1140,7 +1140,7 @@ func TestBolt_DeleteIndex(t *testing.T) {
}
index2 := "i2" // should not be deleted, even though it shares a prefix with 'i'
changed, err := tx.Add(index2, field, view, shard, doBatched, bitvalue)
changed, err := tx.Add(index2, field, view, shard, bitvalue)
if changed <= 0 {
panic("should have changed")
}
@ -1196,7 +1196,7 @@ func TestBolt_DeleteIndex_over100k(t *testing.T) {
//limit := uint64(101)
for v := uint64(1); v < limit; v++ {
// shift by << 16 to get into a different shard
changed, err := tx.Add(index, field, view, shard, doBatched, v<<16)
changed, err := tx.Add(index, field, view, shard, v<<16)
if changed <= 0 {
panic("should have changed")
}
@ -1208,7 +1208,7 @@ func TestBolt_DeleteIndex_over100k(t *testing.T) {
}
index2 := "i2" // should not be deleted, even though it shares a prefix with 'i'
changed, err := tx.Add(index2, field, view, shard, doBatched, bitvalue)
changed, err := tx.Add(index2, field, view, shard, bitvalue)
if changed <= 0 {
panic("should have changed")
}

4
catcher.go
View file

@ -151,7 +151,7 @@ func (c *catcherTx) IsDone() bool {
return c.b.IsDone()
}
func (c *catcherTx) Add(index, field, view string, shard uint64, batched bool, a ...uint64) (changeCount int, err error) {
func (c *catcherTx) Add(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
defer func() {
if r := recover(); r != nil {
@ -159,7 +159,7 @@ func (c *catcherTx) Add(index, field, view string, shard uint64, batched bool, a
PanicOn(r)
}
}()
return c.b.Add(index, field, view, shard, batched, a...)
return c.b.Add(index, field, view, shard, a...)
}
func (c *catcherTx) Remove(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {

11
ctl/inspect.go
View file

@ -29,6 +29,7 @@ import (
"syscall"
"text/tabwriter"
"time"
"unsafe"
"github.com/gogo/protobuf/proto"
"github.com/pilosa/pilosa/v2"
@ -360,16 +361,24 @@ func (cmd *InspectCommand) InspectFile(f *os.File, fi os.FileInfo) error {
fmt.Fprintf(cmd.Stderr, "inspect command: munmap failed: %v", err)
}
}()
mappedFrom := uintptr(unsafe.Pointer(&data[0]))
mappedTo := mappedFrom + uintptr(len(data))
// Attach the mmap file to the bitmap.
t := time.Now()
fmt.Fprintf(cmd.Stderr, "inspecting bitmap...")
var info roaring.BitmapInfo
_, _, err = roaring.InspectBinary(data, true, &info)
bitmap, _, err := roaring.InspectBinary(data, true, &info)
fmt.Fprintf(cmd.Stderr, " (%s)\n", time.Since(t))
cmd.DisplayInfo(info)
if err != nil {
return errors.Wrap(err, "inspecting")
}
mappedIn, mappedOut, unmappedIn, errs, err := bitmap.SanityCheckMapping(mappedFrom, mappedTo)
if err != nil {
fmt.Fprintf(cmd.Stderr, "sanity check: %d mapped in, %d mapped out, %d unmapped in, %d errors\n",
mappedIn, mappedOut, unmappedIn, errs)
fmt.Fprintf(cmd.Stderr, "last error: %v\n", err)
}
return nil
}

3
dbshard_internal_test.go
View file

@ -300,7 +300,6 @@ func makeRBFtestDB(path string, h *Holder, shard uint64) {
func makeTxTestDBWithViewsShards(holder *Holder, idx *Index, exp *FieldView2Shards) {
// TODO(jea): need date time quantum views!!
batched := false
for field, viewmap := range exp.m {
for view, shset := range viewmap {
@ -310,7 +309,7 @@ func makeTxTestDBWithViewsShards(holder *Holder, idx *Index, exp *FieldView2Shar
// simply write 1 bit to each shard to force its creation.
bits := []uint64{(shard << shardwidth.Exponent) + 1}
tx := idx.holder.txf.NewTx(Txo{Write: writable, Index: idx, Shard: shard})
changeCount, err := tx.Add(idx.name, field, view, shard, batched, bits...)
changeCount, err := tx.Add(idx.name, field, view, shard, bits...)
PanicOn(err)
if changeCount != len(bits) {
panic(fmt.Sprintf("writing field '%v', view '%v' shard '%v', expected changeCount to equal len bits = %v but was %v", field, view, shard, len(bits), changeCount))

5
fragment.go
View file

@ -729,7 +729,7 @@ func (f *fragment) unprotectedSetBit(tx Tx, rowID, columnID uint64) (changed boo
// Write to storage.
changeCount := 0
changeCount, err = tx.Add(f.index(), f.field(), f.view(), f.shard, doBatched, pos)
changeCount, err = tx.Add(f.index(), f.field(), f.view(), f.shard, pos)
changed = changeCount > 0
if err != nil {
return false, errors.Wrap(err, "writing")
@ -2488,8 +2488,7 @@ func (f *fragment) importPositions(tx Tx, set, clear []uint64, rowSet map[uint64
f.stats.Count(MetricImportingN, int64(len(set)), 1)
// TODO benchmark Add/RemoveN behavior with sorted/unsorted positions
// Note: AddN() avoids writing to the op-log. While Add() does.
changedN, err := tx.Add(f.index(), f.field(), f.view(), f.shard, !doBatched, set...)
changedN, err := tx.Add(f.index(), f.field(), f.view(), f.shard, set...)
if err != nil {
return errors.Wrap(err, "adding positions")
}

85
fragment_internal_test.go
View file

@ -5441,7 +5441,7 @@ func TestRemapCache(t *testing.T) {
}()
// create a container
_, err := tx.Add(index, field, view, shard, !doBatched, 65537)
_, err := tx.Add(index, field, view, shard, 65537)
if err != nil {
t.Fatalf("storage add: %v", err)
}
@ -5454,7 +5454,7 @@ func TestRemapCache(t *testing.T) {
_ = f.mustRow(tx, 0)
// add a bit that isn't in that container, so that container doesn't
// change
_, err = tx.Add(index, field, view, shard, !doBatched, 2)
_, err = tx.Add(index, field, view, shard, 2)
if err != nil {
t.Fatalf("storage add: %v", err)
}
@ -6057,3 +6057,84 @@ func TestSliceDifference(t *testing.T) {
compareSlices(t, name, tc.expected, result)
}
}
func TestBitmapGrowth(t *testing.T) {
roaringOnlyTest(t)
f, _, tx := mustOpenFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0, "")
path := f.path()
defer f.Clean(t)
const values = 500
cols := make([]uint64, values)
vals := make([]int64, values)
for i := range cols {
cols[i] = uint64(rand.Int63n(65536))
vals[i] = rand.Int63n(24)
}
err := f.importValue(tx, cols, vals, 7, false)
if err != nil {
t.Fatalf("importing values: %v", err)
}
info, err := os.Stat(path)
if err != nil {
t.Fatalf("statting %s: %v", path, err)
}
prevSize := info.Size()
prevOpN := f.opN
err = f.importValue(tx, cols, vals, 7, false)
if err != nil {
t.Fatalf("importing values: %v", err)
}
info, err = os.Stat(path)
if err != nil {
t.Fatalf("statting %s: %v", path, err)
}
deltaSize := info.Size() - prevSize
deltaOpN := f.opN - prevOpN
// This is somewhat arbitrary, but the issue tested for was that
// opN would grow by 0 or 1 with multiple KB of actual ops written.
// If deltaOpN is at least 20, we'll probably see snapshots happening
// at least occasionally, and if deltaSize is under 1024, the writes
// are probably going to be small enough that the regular backlog of
// snapshotting catches them anyway.
if deltaSize > 1024 && deltaOpN < 20 {
t.Fatalf("bitmap grew by %d bytes but OpN only grew by %d",
deltaSize, deltaOpN)
}
}
func TestTxBitmap(t *testing.T) {
f, _, tx := mustOpenFragment(t, "i", "f", viewBSIGroupPrefix+"foo", 0, "")
defer f.Clean(t)
f.MaxOpN = 8
cols := []uint64{1, 2, 3, 4, 5, 6, 65537, 131073}
vals := []int64{4, 4, 4, 4, 4, 4, 4, 4}
zeros := []int64{0, 0, 0, 0, 0, 0, 0, 0}
err := f.importValue(tx, cols, vals, 7, false)
if err != nil {
t.Fatalf("importing values: %v", err)
}
expected := []uint64{1, (4 * ShardWidth) + 1}
var got []uint64
_ = tx.ForEach("i", "f", viewBSIGroupPrefix+"foo", 0, func(i uint64) error {
got = append(got, i)
return nil
})
t.Logf("initial: %d", got)
err = f.importValue(tx, cols[1:], zeros[1:], 7, true)
if err != nil {
t.Fatalf("clearing values: %v", err)
}
got = got[:0]
_ = tx.ForEach("i", "f", viewBSIGroupPrefix+"foo", 0, func(i uint64) error {
got = append(got, i)
return nil
})
if len(got) != len(expected) {
t.Fatalf("bitmap clear unsuccessful: expected %d, got %d", expected, got)
}
for i := range expected {
if expected[i] != got[i] {
t.Fatalf("bitmap clear unsuccessful: expected %d, got %d", expected, got)
}
}
}

19
rbf.go
View file

@ -241,27 +241,16 @@ func (tx *RBFTx) RemoveContainer(index, field, view string, shard uint64, key ui
}
// Add sets all the a bits hot in the specified fragment.
func (tx *RBFTx) Add(index, field, view string, shard uint64, batched bool, a ...uint64) (changeCount int, err error) {
return tx.addOrRemove(index, field, view, shard, batched, false, a...)
func (tx *RBFTx) Add(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
return tx.addOrRemove(index, field, view, shard, false, a...)
}
// Remove clears all the specified a bits in the chosen fragment.
func (tx *RBFTx) Remove(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
const batched = false
const remove = true
return tx.addOrRemove(index, field, view, shard, batched, remove, a...)
return tx.addOrRemove(index, field, view, shard, true, a...)
}
func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, batched, remove bool, a ...uint64) (changeCount int, err error) {
// pure hack to match RoaringTx
defer func() {
if !remove && !batched {
if changeCount > 0 {
changeCount = 1
}
}
}()
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
}

11
roaring/roaring.go
View file

@ -6002,7 +6002,7 @@ func (op *op) UnmarshalBinary(data []byte) error {
switch op.typ {
case opTypeAdd, opTypeRemove:
// nothing to do, just being not-default
op.opN = 1
case opTypeAddBatch, opTypeRemoveBatch:
// This ensures that in doing 13+op.value*8, the max int won't be exceeded and a wrap around case
// (resulting in a negative value) won't occur in the slice indexing while writing
@ -6013,8 +6013,9 @@ func (op *op) UnmarshalBinary(data []byte) error {
return fmt.Errorf("op data truncated - expected %d, got %d", 13+op.value*8, len(data))
}
_, _ = h.Write(data[13 : 13+op.value*8])
op.values = make([]uint64, op.value)
for i := uint64(0); i < op.value; i++ {
op.opN = int(op.value)
op.values = make([]uint64, op.opN)
for i := range op.values {
start := 13 + i*8
op.values[i] = binary.LittleEndian.Uint64(data[start : start+8])
}
@ -6515,16 +6516,14 @@ func (b *Bitmap) BitwiseEqual(c *Bitmap) (bool, error) {
bct++
break
}
bn = biter.Next()
}
for cn {
cn = citer.Next()
ck, cc = biter.Value()
ck, cc = citer.Value()
if cc.N() != 0 {
cct++
break
}
cn = biter.Next()
}
if bn {
return false, fmt.Errorf("container mismatch: %d vs %d containers, first bitmap has extra container %d [%v bits]", bct, cct, bk, bc)

12
roaring/unmarshal_binary.go
View file

@ -58,7 +58,10 @@ func (b *Bitmap) UnmarshalBinary(data []byte) (err error) {
default:
panic("invalid container type")
}
newC.setMapped(true)
// If we're using the iterator's pointer, we're "mapped". But
// for instance, small arrays may use their own data structures,
// which is fine.
newC.setMapped(newC.pointer == itrPointer)
if !b.preferMapping {
newC = newC.unmapOrClone()
}
@ -150,10 +153,14 @@ func InspectBinary(data []byte, mapped bool, info *BitmapInfo) (b *Bitmap, mappe
default:
panic("invalid container type")
}
newC.setMapped(true)
// If our pointer isn't itrPointer, we aren't actually mapped.
newC.setMapped(newC.pointer == itrPointer)
if !mapped {
newC = newC.unmapOrClone()
}
// Pristine means this is the original object read in from
// roaring data, even if it's not mapped, which this is for
// now.
newC.flags |= flagPristine
if newC.flags&flagMapped != 0 {
mappedAny = true
@ -169,6 +176,7 @@ func InspectBinary(data []byte, mapped bool, info *BitmapInfo) (b *Bitmap, mappe
})
info.ContainerCount++
info.BitCount += uint64(newC.n)
b.Containers.Put(itrKey, newC)
itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next()
}
// note: if we get a non-EOF err, it's possible that we made SOME

28
rrtx.go
View file

@ -199,25 +199,16 @@ func (tx *RoaringTx) RemoveContainer(index, field, view string, shard uint64, ke
return nil
}
func (tx *RoaringTx) Add(index, field, view string, shard uint64, batched bool, a ...uint64) (changeCount int, err error) {
func (tx *RoaringTx) Add(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
//vv("RoaringTx.Add(index='%v', shard='%v') stack=\n%v", index, shard, stack())
b, err := tx.bitmap(index, field, view, shard)
if err != nil {
return 0, err
}
if !batched {
changed, err := b.Add(a...)
if changed {
return 1, err
}
return 0, err
}
// Note: do not replace b.AddN() with b.DirectAddN().
// DirectAddN() does not do op-log operations inside roaring
// This creates a problem because RoaringTx needs the op-log
// to know when to flush the fragment to disk.
count, err := b.AddN(a...) // AddN does oplog batches. needed to keep op-log up to date.
// DirectAddN() does not do op-log operations inside roaring, so the
// on-disk representation no longer matches the in-memory operations.
count, err := b.AddN(a...)
return count, err
}
@ -226,16 +217,7 @@ func (tx *RoaringTx) Remove(index, field, view string, shard uint64, a ...uint64
if err != nil {
return 0, err
}
changed, err := b.Remove(a...) // green TestFragment_Bug_Q2DoubleDelete
if changed {
return 1, err
} else {
return 0, err
}
// Note: don't replace b.Remove(a...) with b.RemoveN(a...) or
// with b.DirectRemoveN(a...). If you do, you'll see
// TestFragment_Bug_Q2DoubleDelete go red.
return b.RemoveN(a...)
}
func (tx *RoaringTx) Contains(index, field, view string, shard uint64, v uint64) (exists bool, err error) {

4
stattx.go
View file

@ -429,7 +429,7 @@ func (c *statTx) IsDone() (done bool) {
return c.b.IsDone()
}
func (c *statTx) Add(index, field, view string, shard uint64, batched bool, a ...uint64) (changeCount int, err error) {
func (c *statTx) Add(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {
me := kAdd
t0 := time.Now()
@ -443,7 +443,7 @@ func (c *statTx) Add(index, field, view string, shard uint64, batched bool, a ..
PanicOn(r)
}
}()
return c.b.Add(index, field, view, shard, batched, a...)
return c.b.Add(index, field, view, shard, a...)
}
func (c *statTx) Remove(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error) {

39
tx.go
View file

@ -22,11 +22,6 @@ import (
//txkey "github.com/pilosa/pilosa/v2/txkey"
)
// batch operations want Tx.Add(batched=doBatch), while bit-at-a-time want Tx.Add(batched=!doBatched)
// Used in Tx.Add() to get consistency between RoaringTx and other Tx implementations on
// the changeCount returned.
const doBatched = false // must be false, do not change this without adjusting the Add() implementations correspondingly.
// writable initializes Tx that update, use !writable for read-only.
const writable = true
@ -136,23 +131,7 @@ type Tx interface {
// in the specified fragment.
RemoveContainer(index, field, view string, shard uint64, ckey uint64) error
// Add adds the 'a' bits to the specified fragment.
//
// Using batched=true allows efficient bulk-import.
//
// Notes on the RoaringTx implementation:
// If the batched flag is true, then the roaring.Bitmap.AddN() is used, which does oplog batches.
// If the batched flag is false, then the roaring.Bitmap.Add() is used, which does simple opTypeAdd single adds.
//
// Beware: if batched is false, then changeCount will only ever be 0 or 1,
// because it calls roaring.Add().
// If batched is true, we call roaring.DirectAddN() and then changeCount
// will be accurate if the changeCount is greater than 0.
//
// Hence: only ever call Add(batched=false) if changeCount is expected to be 0 or 1.
// Or, must use Add(batched=true) if changeCount can be > 1.
//
Add(index, field, view string, shard uint64, batched bool, a ...uint64) (changeCount int, err error)
Add(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error)
// Remove removes the 'a' values from the Bitmap for the fragment.
Remove(index, field, view string, shard uint64, a ...uint64) (changeCount int, err error)
@ -278,6 +257,9 @@ type TxBitmap struct {
field string
view string
shard uint64
// Container keys we've already snagged containers for, even if
// those containers have since been deleted by remove ops.
seen map[uint64]struct{}
}
func NewTxBitmap(tx Tx, index, field, view string, shard uint64) *TxBitmap {
@ -288,6 +270,7 @@ func NewTxBitmap(tx Tx, index, field, view string, shard uint64) *TxBitmap {
field: field,
view: view,
shard: shard,
seen: make(map[uint64]struct{}),
}
}
@ -309,14 +292,14 @@ func (b *TxBitmap) Remove(a ...uint64) (changed bool, err error) {
func (b *TxBitmap) ensureContainers(a ...uint64) error {
for _, v := range a {
key := highbits(v)
if b.b.Containers.Get(key) != nil {
if _, ok := b.seen[key]; ok {
continue
}
c, err := b.tx.Container(b.index, b.field, b.view, b.shard, key)
if err != nil {
return err
}
b.seen[key] = struct{}{}
b.b.Containers.Put(key, c)
}
return nil
@ -329,6 +312,14 @@ func (b *TxBitmap) Flush() error {
if err := b.tx.PutContainer(b.index, b.field, b.view, b.shard, key, c); err != nil {
return err
}
delete(b.seen, key)
}
// remove containers we have seen but no longer have, because that means
// we deleted everything from them.
for key := range b.seen {
if err := b.tx.RemoveContainer(b.index, b.field, b.view, b.shard, key); err != nil {
return err
}
}
return nil
}