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

drop "batched" flag from Add operation

Browse source 
The "batched" flag creates a complexity which is that the return value of Add
might or might not be meaningful, but it doesn't really buy us very much.

If we are concerned about the ops log size of writing single ops as 21-byte
arrays of 1 op rather than as 13-byte ops, we can make the AddN code smarter
about how it writes ops. And probably should.

Along with this, change Remove to use the batched operation form, which
writes a more meaningful ops log, and return a meaningful value for changes
made. Otherwise, it ends up writing potentially thousands of ops to the
ops log without reporting any OpN, because the number of ops written isn't
the same as the number of changes those ops made. This could result in
files growing by megabytes without OpN changing.

There was a comment here about a test failing with RemoveN. I can't prove
it, but I strongly suspect that this was actually a result of that test
case hitting a particular bug that we eventually fixed, and which we might
have fixed sooner if we'd realized why using RemoveN made that test
fail.
This commit is contained in:
Seebs 2021-03-29 10:01:21 -05:00
parent 3ef2c84d3c
commit a94c745a7f
11 changed files with 41 additions and 104 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

@ -625,28 +625,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

@ -55,7 +55,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")
}
@ -122,7 +122,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")
}
@ -238,7 +238,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")
}
@ -280,14 +280,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")
}
@ -373,7 +373,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")
@ -419,7 +419,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")
}
@ -477,7 +477,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")
}
@ -532,7 +532,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")
}
@ -581,7 +581,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")
}
@ -1131,7 +1131,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")
}
@ -1139,7 +1139,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")
}
@ -1195,7 +1195,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")
}
@ -1207,7 +1207,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
panic(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) {

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

@ -708,7 +708,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")
@ -2465,8 +2465,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")
}

4
fragment_internal_test.go
View file

@ -5426,7 +5426,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)
}
@ -5439,7 +5439,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)
}

20
rbf.go
View file

@ -30,6 +30,7 @@ import (
rbfcfg "github.com/pilosa/pilosa/v2/rbf/cfg"
"github.com/pilosa/pilosa/v2/roaring"
txkey "github.com/pilosa/pilosa/v2/short_txkey"
//txkey "github.com/pilosa/pilosa/v2/txkey"
"github.com/pkg/errors"
)
@ -241,27 +242,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
}

28
rrtx.go
View file

@ -200,25 +200,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
}
@ -227,16 +218,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

@ -445,7 +445,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()
@ -459,7 +459,7 @@ func (c *statTx) Add(index, field, view string, shard uint64, batched bool, a ..
panic(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) {

23
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)