diff --git a/api.go b/api.go index 75becc899..37d385791 100644 --- a/api.go +++ b/api.go @@ -1568,7 +1568,15 @@ func (api *API) ImportValueWithTx(ctx context.Context, qcx *Qcx, req *ImportValu } if !options.Presorted { + // horrible hackery: we implement a secondary key so we can + // get a stable sort without using sort.Stable + req.scratch = make([]int, len(req.ColumnIDs)) + for i := range req.scratch { + req.scratch[i] = i + } sort.Sort(req) + // don't keep that list around since we don't need it anymore + req.scratch = nil } isLocalQcx := false if qcx == nil { diff --git a/field.go b/field.go index 2eafcd4fa..3f2c4c534 100644 --- a/field.go +++ b/field.go @@ -1525,6 +1525,9 @@ func (f *Field) Import(qcx *Qcx, rowIDs, columnIDs []uint64, timestamps []*time. return nil } +// importFloatValue imports floating point values. In current usage, this +// should only ever be called with data for a single shard; the API calls +// around this are splitting it up per shard. func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, options *ImportOptions) error { // convert values to int64 values based on scale ivalues := make([]int64, len(values)) @@ -1540,6 +1543,9 @@ func (f *Field) importFloatValue(qcx *Qcx, columnIDs []uint64, values []float64, return f.importValue(qcx, columnIDs, ivalues, options) } +// importFloatValue imports timestamp values. In current usage, this +// should only ever be called with data for a single shard; the API calls +// around this are splitting it up per shard. func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time.Time, options *ImportOptions) error { ivalues := make([]int64, len(values)) bsig := f.bsiGroup(f.name) @@ -1553,8 +1559,17 @@ func (f *Field) importTimestampValue(qcx *Qcx, columnIDs []uint64, values []time return f.importValue(qcx, columnIDs, ivalues, options) } -// importValue bulk imports range-encoded value data. +// importValue bulk imports range-encoded value data. This function should +// only be called with data for a single shard; the API calls that wrap +// this handle splitting the data up per-shard. func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, options *ImportOptions) (err0 error) { + // no data to import + if len(columnIDs) == 0 { + return nil + } + if len(values) != len(columnIDs) { + return fmt.Errorf("importValue: mismatch between column IDs and values: %d != %d", len(columnIDs), len(values)) + } viewName := viewBSIGroupPrefix + f.name // Get the bsiGroup so we know bitDepth. bsig := f.bsiGroup(f.name) @@ -1565,13 +1580,10 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option // We want to determine the required bit depth, in case the field doesn't // have as many bits currently as would be needed to represent these values, // but only if the values are in-range for the field. - var min, max int64 - if len(values) > 0 { - min, max = values[0], values[0] - } + min, max := values[0], values[0] - // Split import data by fragment. - dataByFragment := make(map[importKey]importValueData) + // Check for minimum/maximum in case we need to expand the field's + // stated bit depth. for i := range columnIDs { columnID, value := columnIDs[i], values[i] if value > bsig.Max { @@ -1585,15 +1597,6 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option if value < min { min = value } - - // Attach value to each bsiGroup view. - for _, name := range []string{viewName} { - key := importKey{View: name, Shard: columnID / ShardWidth} - data := dataByFragment[key] - data.ColumnIDs = append(data.ColumnIDs, columnID) - data.Values = append(data.Values, value) - dataByFragment[key] = data - } } // Determine the highest bit depth required by the min & max. @@ -1612,40 +1615,36 @@ func (f *Field) importValue(qcx *Qcx, columnIDs []uint64, values []int64, option } f.mu.Unlock() - // Import into each fragment. - for key, data := range dataByFragment { - // The view must already exist (i.e. we can't create it) - // because we need to know bitDepth (based on min/max value). - view, err := f.createViewIfNotExists(key.View) - if err != nil { - return errors.Wrap(err, "creating view") - } + // Since all data should be for the same shard, we can just compute + // this from the first value. + shard := columnIDs[0] / ShardWidth - frag, err := view.CreateFragmentIfNotExists(key.Shard) - if err != nil { - return errors.Wrap(err, "creating fragment") - } + view, err := f.createViewIfNotExists(viewName) + if err != nil { + return errors.Wrap(err, "creating view") + } - baseValues := make([]int64, len(data.Values)) - for i, value := range data.Values { - baseValues[i] = value - bsig.Base - } + frag, err := view.CreateFragmentIfNotExists(shard) + if err != nil { + return errors.Wrap(err, "creating fragment") + } - // now we know which shard we discovered. - tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: f.idx, Shard: frag.shard}) - if err != nil { - return err - } - // by deferring, even though we are in loop, we get en-mass commit at once if they all succeed, - // or en-mass rollback if any fail. - defer finisher(&err0) - - if err = frag.importValue(tx, data.ColumnIDs, baseValues, requiredDepth, options.Clear); err != nil { - return err + if bsig.Base != 0 { + for i, v := range values { + values[i] = v - bsig.Base } } - return nil + // now we know which shard we discovered. + tx, finisher, err := qcx.GetTx(Txo{Write: writable, Index: f.idx, Shard: frag.shard}) + if err != nil { + return err + } + // defer the finisher, so it will check the error returned and + // possibly rollback. + defer finisher(&err0) + + return frag.importValue(tx, columnIDs, values, requiredDepth, options.Clear) } func (f *Field) importRoaring(ctx context.Context, tx Tx, data []byte, shard uint64, viewName string, clear bool) error { diff --git a/field_internal_test.go b/field_internal_test.go index 8a9ba7ebe..337909494 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -203,7 +203,7 @@ type TestField struct { } // NewTestField returns a new instance of TestField d/0. -func NewTestField(t *testing.T, opts FieldOption) *TestField { +func NewTestField(t testing.TB, opts FieldOption) *TestField { path, err := testhook.TempDirInDir(t, *TempDir, "pilosa-field-") if err != nil { t.Fatal(err) @@ -230,7 +230,7 @@ func NewTestField(t *testing.T, opts FieldOption) *TestField { } // OpenField returns a new, opened field at a temporary path. -func OpenField(t *testing.T, opts FieldOption) *TestField { +func OpenField(t testing.TB, opts FieldOption) *TestField { f := NewTestField(t, opts) return f } @@ -509,6 +509,38 @@ func TestBSIGroup_importValue(t *testing.T) { } // loop } +// benchmarkImportValues is a helper function to explore, very roughly, the cost +// of setting values using the special setter used for imports. +func benchmarkFieldImportValues(b *testing.B, qcx *Qcx, bitDepth uint64, f *TestField, cfunc func(uint64) uint64) { + batches := makeBenchmarkImportValueData(b, bitDepth, cfunc) + for _, req := range batches { + err := f.importValue(qcx, req.ColumnIDs, req.Values, &ImportOptions{}) + if err != nil { + b.Fatalf("error importing values: %s", err) + } + } +} + +// Benchmark performance of setValue for BSI ranges. +func BenchmarkField_ImportValue(b *testing.B) { + depths := []uint64{4, 8, 16, 32} + + for _, bitDepth := range depths { + f := OpenField(b, OptFieldTypeInt(0, 1<= 0 || clear { - if c, err := txb.Remove(p); err != nil { - return changed, errors.Wrap(err, "removing sign from storage") - } else if c { - changed++ - } - } else { - if c, err := txb.Add(p); err != nil { - return changed, errors.Wrap(err, "adding sign to storage") - } else if c { - changed++ - } - } - - return changed, nil -} - // sum returns the sum of a given bsiGroup as well as the number of columns involved. // A bitmap can be passed in to optionally filter the computed columns. func (f *fragment) sum(tx Tx, filter *Row, bitDepth uint64) (sum int64, count uint64, err error) { @@ -2283,6 +2215,34 @@ func (f *fragment) bulkImport(tx Tx, rowIDs, columnIDs []uint64, options *Import return f.bulkImportStandard(tx, rowIDs, columnIDs, options) } +// rowColumnSet is a sortable set of row and column IDs which +// correspond, allowing us to ensure that we produce values in +// a predictable order +type rowColumnSet struct { + r []uint64 + c []uint64 +} + +func (r rowColumnSet) Len() int { + return len(r.r) +} + +func (r rowColumnSet) Swap(i, j int) { + r.r[i], r.r[j] = r.r[j], r.r[i] + r.c[i], r.c[j] = r.c[j], r.c[i] +} + +// Sort by row ID first, column second, to sort by fragment position +func (r rowColumnSet) Less(i, j int) bool { + if r.r[i] < r.r[j] { + return true + } + if r.r[i] > r.r[j] { + return false + } + return r.c[i] < r.c[j] +} + // bulkImportStandard performs a bulk import on a standard fragment. May mutate // its rowIDs and columnIDs arguments. func (f *fragment) bulkImportStandard(tx Tx, rowIDs, columnIDs []uint64, options *ImportOptions) (err error) { @@ -2294,13 +2254,21 @@ func (f *fragment) bulkImportStandard(tx Tx, rowIDs, columnIDs []uint64, options lastRowID := uint64(1 << 63) // replace columnIDs with calculated positions to avoid allocation. + sort.Sort(rowColumnSet{r: rowIDs, c: columnIDs}) + prevRow, prevCol := ^uint64(0), ^uint64(0) + next := 0 for i := 0; i < len(columnIDs); i++ { rowID, columnID := rowIDs[i], columnIDs[i] + if rowID == prevRow && columnID == prevCol { + continue + } + prevRow, prevCol = rowID, columnID pos, err := f.pos(rowID, columnID) if err != nil { return err } - columnIDs[i] = pos + columnIDs[next] = pos + next++ // Add row to rowSet. if rowID != lastRowID { @@ -2308,7 +2276,7 @@ func (f *fragment) bulkImportStandard(tx Tx, rowIDs, columnIDs []uint64, options rowSet[rowID] = struct{}{} } } - positions := columnIDs + positions := columnIDs[:next] f.mu.Lock() defer f.mu.Unlock() if options.Clear { @@ -2613,57 +2581,6 @@ func (f *fragment) bulkImportMutex(tx Tx, rowIDs, columnIDs []uint64) error { return errors.Wrap(f.importPositions(tx, toSet, toClear, rowSet), "importing positions") } -func (f *fragment) importValueSmallWrite(tx Tx, columnIDs []uint64, values []int64, bitDepth uint64, clear bool) error { - // TODO figure out how to avoid re-allocating these each time. Probably - // possible to store them on the fragment with a capacity based on - // MaxOpN. For now, we know that the total number of bits to be - // set+cleared is len(values)*(bitDepth+1), so we make each slice - // slightly more than half of that to try to avoid reallocation. - toSet := make([]uint64, 0, len(columnIDs)*int(bitDepth+1)*(5/8)) - toClear := make([]uint64, 0, len(columnIDs)*int(bitDepth+1)*(5/8)) - colSet := make(map[uint64]struct{}, len(columnIDs)) - - if err := func() (err error) { - for i := len(columnIDs) - 1; i >= 0; i-- { - columnID, value := columnIDs[i], values[i] - if _, ok := colSet[columnID]; ok { - continue - } - - colSet[columnID] = struct{}{} - toSet, toClear, err = f.positionsForValue(columnID, bitDepth, value, clear, toSet, toClear) - if err != nil { - return errors.Wrap(err, "getting positions for value") - } - } - return nil - }(); err != nil { - errOpenStorage := f.openStorage(true) - if errOpenStorage != nil { - f.Logger.Errorf("failed to import data into fragment: %v", err) - f.Logger.Errorf("recovery with openStorage failed for fragment: %v", errOpenStorage) - f.Logger.Debugf("%s", debug.Stack()) - os.Exit(1) - } - return err - } - rowSet := make(map[uint64]struct{}, bitDepth+1) - for i := uint64(0); i < bitDepth+1; i++ { - rowSet[uint64(i)] = struct{}{} - } - err := f.importPositions(tx, toSet, toClear, rowSet) - if err != nil { - return errors.Wrap(err, "importing positions") - } - - if tx.UseRowCache() { - // Reset the rowCache. - f.rowCache = newSimpleCache() - } - - return nil -} - // importValue bulk imports a set of range-encoded values. func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDepth uint64, clear bool) error { f.mu.Lock() @@ -2673,56 +2590,84 @@ func (f *fragment) importValue(tx Tx, columnIDs []uint64, values []int64, bitDep if len(columnIDs) != len(values) { return fmt.Errorf("mismatch of column/value len: %d != %d", len(columnIDs), len(values)) } - - if len(columnIDs)*int(bitDepth+1)+f.opN < f.MaxOpN { - return errors.Wrap(f.importValueSmallWrite(tx, columnIDs, values, bitDepth, clear), "import small write") + positionsByDepth := make([][]uint64, bitDepth+2) + toSetByDepth := make([]int, bitDepth+2) + toClearByDepth := make([]int, bitDepth+2) + batchSize := len(columnIDs) + if batchSize > 65536 { + batchSize = 65536 + } + for i := 0; i < int(bitDepth)+2; i++ { + positionsByDepth[i] = make([]uint64, batchSize) + toClearByDepth[i] = batchSize } - // Process every value. - // If an error occurs then reopen the storage. - if f.storage != nil { - f.storage.OpWriter = nil - } - - var totalChanges int - if err := func() (err error) { - // Build changes into temporary bitmap. - txb := NewTxBitmap(tx, f.index(), f.field(), f.view(), f.shard) - for i := range columnIDs { - columnID, value := columnIDs[i], values[i] - if _, err := f.importSetValue(txb, columnID, bitDepth, value, clear); err != nil { - return errors.Wrapf(err, "importSetValue") + row := 0 + columnID := uint64(0) + value := int64(0) + // arbitrarily set prev to be not equal to the first column ID + // we will encounter. + prev := columnIDs[len(columnIDs)-1] + 1 + for len(columnIDs) > 0 { + downTo := len(columnIDs) - batchSize + if downTo < 0 { + downTo = 0 + } + for i := range positionsByDepth { + toSetByDepth[i] = 0 + toClearByDepth[i] = batchSize + } + for i := len(columnIDs) - 1; i >= downTo; i-- { + columnID, value = columnIDs[i], values[i] + columnID = columnID % ShardWidth + if columnID == prev { + continue + } + prev = columnID + row = 0 + if clear { + toClearByDepth[row]-- + positionsByDepth[row][toClearByDepth[row]] = columnID + } else { + positionsByDepth[row][toSetByDepth[row]] = columnID + toSetByDepth[row]++ + } + row++ + columnID += ShardWidth + if value < 0 { + positionsByDepth[row][toSetByDepth[row]] = columnID + toSetByDepth[row]++ + value *= -1 + } else { + toClearByDepth[row]-- + positionsByDepth[row][toClearByDepth[row]] = columnID + } + row++ + columnID += ShardWidth + for j := 0; j < int(bitDepth); j++ { + if value&1 != 0 { + positionsByDepth[row][toSetByDepth[row]] = columnID + toSetByDepth[row]++ + } else { + toClearByDepth[row]-- + positionsByDepth[row][toClearByDepth[row]] = columnID + } + row++ + columnID += ShardWidth + value >>= 1 } } - // Flush changes in bulk back to the transaction. - return txb.Flush() - }(); err != nil { - errOpenStorage := f.openStorage(true) - if errOpenStorage != nil { - f.Logger.Errorf("failed to import data into fragment: %v", err) - f.Logger.Errorf("recovery with openStorage failed for fragment: %v", errOpenStorage) - f.Logger.Debugf("%s", debug.Stack()) - os.Exit(1) + for i := range positionsByDepth { + err := f.importPositions(tx, positionsByDepth[i][:toSetByDepth[i]], positionsByDepth[i][toClearByDepth[i]:], nil) + if err != nil { + return errors.Wrap(err, "importing positions") + } } - return err - } - // Keep stats accurate. We don't call incrementOpN here because it may - // or may not enqueue a request, which would then be in the queue - // taking up space and otherwise being a possible nuisance, when we're - // about to force a snapshot anyway. - f.opN += totalChanges - f.ops++ - - if tx.UseRowCache() { - // Reset the rowCache. - f.rowCache = newSimpleCache() + columnIDs = columnIDs[:downTo] } - // in theory, this should probably have been queued anyway, but if enough - // of the bits matched existing bits, we'll be under our opN estimate, and - // we want to ensure that the snapshot happens. - return f.holder.SnapshotQueue.Immediate(f) + return nil } // importRoaring imports from the official roaring data format defined at diff --git a/fragment_internal_test.go b/fragment_internal_test.go index a3336487e..e95351b5a 100644 --- a/fragment_internal_test.go +++ b/fragment_internal_test.go @@ -1020,34 +1020,58 @@ func BenchmarkFragment_SetValue(b *testing.B) { } } +func makeBenchmarkImportValueData(b *testing.B, bitDepth uint64, cfunc func(uint64) uint64) []ImportValueRequest { + b.StopTimer() + column := uint64(0) + // we don't average an alloc-per-bit, so we use a much larger N to get + // meaningful data from -benchmem + n := b.N * 10000 + batches := make([]ImportValueRequest, 0, (n/ShardWidth)+1) + mask := int64(1< 0 { + req := ImportValueRequest{ColumnIDs: columns, Values: values} + batches = append(batches, req) + } + b.StartTimer() + return batches +} + // benchmarkImportValues is a helper function to explore, very roughly, the cost // of setting values using the special setter used for imports. func benchmarkImportValues(b *testing.B, tx Tx, bitDepth uint64, f *fragment, cfunc func(uint64) uint64) { - column := uint64(0) - b.StopTimer() - columns := make([]uint64, b.N) - values := make([]int64, b.N) - for i := 0; i < b.N; i++ { - values[i] = int64(i) - columns[i] = column - column = cfunc(column) - } - b.StartTimer() - err := f.importValue(tx, columns, values, bitDepth, false) - if err != nil { - b.Fatalf("error importing values: %s", err) + batches := makeBenchmarkImportValueData(b, bitDepth, cfunc) + for _, req := range batches { + err := f.importValue(tx, req.ColumnIDs, req.Values, bitDepth, false) + if err != nil { + b.Fatalf("error importing values: %s", err) + } } } // Benchmark performance of setValue for BSI ranges. func BenchmarkFragment_ImportValue(b *testing.B) { - depths := []uint64{4, 8, 16} + depths := []uint64{4, 8, 16, 32} for _, bitDepth := range depths { name := fmt.Sprintf("Depth%d", bitDepth) f, idx, tx := mustOpenBSIFragment(b, "i", "f", viewBSIGroupPrefix+"foo", 0) _ = idx b.Run(name+"_Sparse", func(b *testing.B) { - benchmarkImportValues(b, tx, bitDepth, f, func(u uint64) uint64 { return (u + 70000) & (ShardWidth - 1) }) + benchmarkImportValues(b, tx, bitDepth, f, func(u uint64) uint64 { return (u + 19) & (ShardWidth - 1) }) }) f.Clean(b) f, idx, tx = mustOpenBSIFragment(b, "i", "f", viewBSIGroupPrefix+"foo", 0) diff --git a/handler.go b/handler.go index 6276ac535..ee9a7d8b5 100644 --- a/handler.go +++ b/handler.go @@ -123,6 +123,7 @@ type ImportValueRequest struct { TimestampValues []time.Time StringValues []string Clear bool + scratch []int // scratch space to allow us to get a stable sort in reasonable time } // AtomicRecord applies all its Ivr and Ivr atomically, in a Tx. @@ -138,8 +139,20 @@ type AtomicRecord struct { Ir []*ImportRequest // other field types, e.g. single bit } -func (ivr *ImportValueRequest) Len() int { return len(ivr.ColumnIDs) } -func (ivr *ImportValueRequest) Less(i, j int) bool { return ivr.ColumnIDs[i] < ivr.ColumnIDs[j] } +func (ivr *ImportValueRequest) Len() int { return len(ivr.ColumnIDs) } +func (ivr *ImportValueRequest) Less(i, j int) bool { + if ivr.ColumnIDs[i] < ivr.ColumnIDs[j] { + return true + } + if ivr.ColumnIDs[i] > ivr.ColumnIDs[j] { + return false + } + if len(ivr.scratch) > 0 { + return ivr.scratch[i] < ivr.scratch[j] + } + return false +} + func (ivr *ImportValueRequest) Swap(i, j int) { ivr.ColumnIDs[i], ivr.ColumnIDs[j] = ivr.ColumnIDs[j], ivr.ColumnIDs[i] if len(ivr.Values) > 0 { @@ -151,6 +164,9 @@ func (ivr *ImportValueRequest) Swap(i, j int) { } else if len(ivr.StringValues) > 0 { ivr.StringValues[i], ivr.StringValues[j] = ivr.StringValues[j], ivr.StringValues[i] } + if len(ivr.scratch) > 0 { + ivr.scratch[i], ivr.scratch[j] = ivr.scratch[j], ivr.scratch[i] + } } // Validate ensures that the payload of the request is valid. diff --git a/index.go b/index.go index 460ef5434..18fa1d49e 100644 --- a/index.go +++ b/index.go @@ -861,11 +861,6 @@ type importData struct { ColumnIDs []uint64 } -type importValueData struct { - ColumnIDs []uint64 - Values []int64 -} - // FormatQualifiedIndexName generates a qualified name for the index to be used with Tx operations. func FormatQualifiedIndexName(index string) string { return fmt.Sprintf("%s\x00", index) diff --git a/rbf.go b/rbf.go index ff2eea25c..f1b4a98c6 100644 --- a/rbf.go +++ b/rbf.go @@ -21,7 +21,6 @@ import ( "io/ioutil" "math" "os" - "sort" "strings" "sync" "sync/atomic" @@ -250,21 +249,79 @@ func (tx *RBFTx) Remove(index, field, view string, shard uint64, a ...uint64) (c return tx.addOrRemove(index, field, view, shard, true, a...) } +// sortedParanoia is a flag to enable a check for unsorted inputs to addOrRemove, +// which is expensive in practice and only really useful occasionally. +const sortedParanoia = false + 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 } - - // have to sort, b/c input is not always sorted. - sort.Slice(a, func(i, j int) bool { return a[i] < a[j] }) + name := rbfName(index, field, view, shard) + // this special case can/should possibly go away, except that it + // turns out to be by far the most common case, and we need to know + // there's at least two items to simplify the check-sorted thing. + if len(a) == 1 { + hi, lo := highbits(a[0]), lowbits(a[0]) + rc, err := tx.tx.Container(name, hi) + if err != nil { + return 0, errors.Wrap(err, "failed to retrieve container") + } + if remove { + if rc.N() == 0 { + return 0, nil + } + rc, chng := rc.Remove(lo) + if !chng { + return 0, nil + } + if rc.N() == 0 { + err = tx.tx.RemoveContainer(name, hi) + } else { + err = tx.tx.PutContainer(name, hi, rc) + } + if err != nil { + return 0, err + } + return 1, nil + } else { + rc, chng := rc.Add(lo) + if !chng { + return 0, nil + } + err = tx.tx.PutContainer(name, hi, rc) + if err != nil { + return 0, err + } + return 1, nil + } + } var lastHi uint64 = math.MaxUint64 // highbits is always less than this starter. var rc *roaring.Container var hi uint64 var lo uint16 + // we can accept sorted either ascending or descending. + sign := a[1] - a[0] + prev := a[0] - sign + sign >>= 63 for i, v := range a { - + // This check is noticably expensive (a few percent in some + // use cases) and as long as it passes occasionally it's probably + // not important to run it all the time, and anyway panic is + // not a good choice outside of testing. + if sortedParanoia { + if (v-prev)>>63 != sign { + explain := fmt.Sprintf("addOrRemove: %d < %d != %d < %d", v, prev, a[1], a[0]) + panic(explain) + } + if v == prev { + explain := fmt.Sprintf("addOrRemove: %d twice", v) + panic(explain) + } + } + prev = v hi, lo = highbits(v), lowbits(v) if hi != lastHi { // either first time through, or changed to a different container. @@ -272,19 +329,19 @@ func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove boo if i > 0 { // not first time through, write what we got. if remove && (rc == nil || rc.N() == 0) { - err = tx.RemoveContainer(index, field, view, shard, lastHi) + err = tx.tx.RemoveContainer(name, lastHi) if err != nil { return 0, errors.Wrap(err, "failed to remove container") } } else { - err = tx.PutContainer(index, field, view, shard, lastHi, rc) + err = tx.tx.PutContainer(name, lastHi, rc) if err != nil { return 0, errors.Wrap(err, "failed to put container") } } } // get the next container - rc, err = tx.Container(index, field, view, shard, hi) + rc, err = tx.tx.Container(name, hi) if err != nil { return 0, errors.Wrap(err, "failed to retrieve container") } @@ -305,12 +362,12 @@ func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove boo // write the last updates. if remove { if rc == nil || rc.N() == 0 { - err = tx.RemoveContainer(index, field, view, shard, hi) + err = tx.tx.RemoveContainer(name, hi) if err != nil { return 0, errors.Wrap(err, "failed to remove container") } } else { - err = tx.PutContainer(index, field, view, shard, hi, rc) + err = tx.tx.PutContainer(name, hi, rc) if err != nil { return 0, errors.Wrap(err, "failed to put container") } @@ -319,7 +376,7 @@ func (tx *RBFTx) addOrRemove(index, field, view string, shard uint64, remove boo if rc == nil || rc.N() == 0 { panic("there should be no way to have an empty bitmap AFTER an Add() operation") } - err = tx.PutContainer(index, field, view, shard, hi, rc) + err = tx.tx.PutContainer(name, hi, rc) if err != nil { return 0, errors.Wrap(err, "failed to put container") }