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

Merge pull request #1598 from seebs/importSpeedups

Browse source 
improve performance of importValue in most cases, switching to consistently use importPositions.
This commit is contained in:
seebs 2021-05-20 16:26:38 -05:00 • committed by GitHub
commit a46e14ef22
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
8 changed files with 318 additions and 256 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
api.go
View file

@ -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 {

87
field.go
View file

@ -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 {

50
field_internal_test.go
View file

@ -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<<bitDepth))
defer f.Close()
qcx := f.idx.holder.txf.NewQcx()
defer qcx.Abort()
name := fmt.Sprintf("Depth%d", bitDepth)
b.Run(name+"_Sparse", func(b *testing.B) {
benchmarkFieldImportValues(b, qcx, bitDepth, f, func(u uint64) uint64 { return (u + 19) & (ShardWidth - 1) })
})
b.Run(name+"_Dense", func(b *testing.B) {
benchmarkFieldImportValues(b, qcx, bitDepth, f, func(u uint64) uint64 { return (u + 1) & (ShardWidth - 1) })
})
}
}
func TestIntField_MinMaxForShard(t *testing.T) {
f := OpenField(t, OptFieldTypeInt(-100, 200))
defer f.Close()
@ -557,13 +589,6 @@ func TestIntField_MinMaxForShard(t *testing.T) {
expMax: ValCount{Val: 20, Count: 2},
expMin: ValCount{Val: 10, Count: 3},
},
{
name: "middlevals",
columnIDs: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100000000, 100000001},
values: []int64{10, 20, 10, 10, 20, 11, 12, 11, 13, 11, 44, 1},
expMax: ValCount{Val: 20, Count: 2},
expMin: ValCount{Val: 10, Count: 3},
},
} {
t.Run(test.name+strconv.Itoa(i), func(t *testing.T) {
if err := f.importValue(qcx, test.columnIDs, test.values, options); err != nil {
@ -724,13 +749,6 @@ func TestDecimalField_MinMaxForShard(t *testing.T) {
expMax: ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
expMin: ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
},
{
name: "another shard",
columnIDs: []uint64{1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 100000000, 100000001},
values: []float64{10.1, 20.2, 10.1, 10.1, 20.2, 11, 12, 11, 13, 11, 44.39, 0.23},
expMax: ValCount{DecimalVal: &pql.Decimal{Value: 20200, Scale: 3}, Count: 2},
expMin: ValCount{DecimalVal: &pql.Decimal{Value: 10100, Scale: 3}, Count: 3},
},
} {
t.Run(test.name+strconv.Itoa(i), func(t *testing.T) {
if err := f.importFloatValue(qcx, test.columnIDs, test.values, options); err != nil {

271
fragment.go
View file

@ -1136,74 +1136,6 @@ func (f *fragment) setValueBase(txOrig Tx, columnID uint64, bitDepth uint64, val
return changed, err
}
// importSetValue is a more efficient SetValue just for imports.
func (f *fragment) importSetValue(txb *TxBitmap, columnID uint64, bitDepth uint64, value int64, clear bool) (changed int, err error) { // nolint: unparam
// Convert value to an unsigned representation.
uvalue := uint64(value)
if value < 0 {
uvalue = uint64(-value)
}
for i := uint64(0); i < bitDepth; i++ {
bit, err := f.pos(uint64(bsiOffsetBit+i), columnID)
if err != nil {
return changed, errors.Wrap(err, "getting pos")
}
if uvalue&(1<<i) != 0 {
c, err := txb.Add(bit)
if err != nil {
return changed, errors.Wrap(err, "adding")
} else if c {
changed++
}
} else {
c, err := txb.Remove(bit)
if err != nil {
return changed, errors.Wrap(err, "removing")
} else if c {
changed++
}
}
}
// Mark value as set.
if p, err := f.pos(uint64(bsiExistsBit), columnID); err != nil {
return changed, errors.Wrap(err, "getting not-null pos")
} else if clear {
if c, err := txb.Remove(p); err != nil {
return changed, errors.Wrap(err, "removing not-null from storage")
} else if c {
changed++
}
} else {
if c, err := txb.Add(p); err != nil {
return changed, errors.Wrap(err, "adding not-null to storage")
} else if c {
changed++
}
}
// Mark sign bit.
if p, err := f.pos(uint64(bsiSignBit), columnID); err != nil {
return changed, errors.Wrap(err, "getting sign pos")
} else if value >= 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

54
fragment_internal_test.go
View file

@ -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<<bitDepth) - 1
prev := uint64(0)
var values []int64
var columns []uint64
for i := 0; i < n; i++ {
values = append(values, int64(i)&mask)
columns = append(columns, column)
column = cfunc(column)
if column < prev {
req := ImportValueRequest{ColumnIDs: columns, Values: values}
columns = []uint64{}
values = []int64{}
batches = append(batches, req)
}
prev = column
}
if len(columns) > 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)

20
handler.go
View file

@ -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.

5
index.go
View file

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

79
rbf.go
View file

@ -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")
}