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

add import support with aggregated op log writes

Browse source
This commit is contained in:
Matt Jaffee 2019-02-21 18:12:23 -06:00
parent dd4a8755ae
commit 537ae99fb9
No known key found for this signature in database
GPG key ID: 08A3DFFF987B11BF
3 changed files with 310 additions and 85 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

174
fragment.go
View file

@ -350,30 +350,41 @@ func (f *fragment) row(rowID uint64) *Row {
return f.unprotectedRow(rowID)
}
// unprotectedRow returns a row from the row cache if available or from storage
// (updating the cache).
func (f *fragment) unprotectedRow(rowID uint64) *Row {
r, ok := f.rowCache.Fetch(rowID)
if ok && r != nil {
return r
}
row := f.rowFromStorage(rowID)
f.rowCache.Add(rowID, row)
return row
}
// rowFromStorage clones a row data out of fragment storage and returns it as a
// Row object.
func (f *fragment) rowFromStorage(rowID uint64) *Row {
// Only use a subset of the containers.
// NOTE: The start & end ranges must be divisible by container width.
data := f.storage.OffsetRange(f.shard*ShardWidth, rowID*ShardWidth, (rowID+1)*ShardWidth)
// Reference bitmap subrange in storage.
// We Clone() data because otherwise row will contain pointers to containers in storage.
// This causes unexpected results when we cache the row and try to use it later.
// Reference bitmap subrange in storage. We Clone() data because otherwise
// row will contain pointers to containers in storage. This causes
// unexpected results when we cache the row and try to use it later.
// Basically, since we return the Row and release the fragment lock, the
// underlying fragment storage could be changed or snapshotted and thrown
// out at any point.
row := &Row{
segments: []rowSegment{{
data: *data.Clone(),
shard: f.shard,
writable: false,
writable: false, // this Row will probably be cached and shared, so it must be read only.
}},
}
row.invalidateCount()
f.rowCache.Add(rowID, row)
return row
}
@ -1454,63 +1465,41 @@ func (f *fragment) bulkImport(rowIDs, columnIDs []uint64, options *ImportOptions
return f.bulkImportStandard(rowIDs, columnIDs, options)
}
// bulkImportStandard performs a bulk import on a standard fragment.
// bulkImportStandard performs a bulk import on a standard fragment. May mutate
// its rowIDs and columnIDs arguments.
func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *ImportOptions) (err error) {
// first we'll try the "small update" path. If there aren't many bits being
// set, it isn't worth it to do a full import and snapshot. Instead we
// leverage individual set/clear bits which get set in memory and appended
// to the op log.
var localBitmap *roaring.Bitmap
smallWrite := false
f.mu.Lock()
defer f.mu.Unlock() // TODO the big import path doesn't need to acquire the lock this soon.
if len(columnIDs)+f.opN < f.MaxOpN {
for i := range rowIDs {
rowID, columnID := rowIDs[i], columnIDs[i]
if options.Clear {
_, err = f.clearBit(rowID, columnID)
} else {
_, err = f.setBit(rowID, columnID)
}
if err != nil {
return errors.Wrapf(err, "importing %dth bit clear:%v", i, options.Clear)
}
}
// forcibly recalculate the cache - setbit just calls invalidate, but in
// order to maintain parity with the "normal" bulkimport path, we want
// to recalculate it at the end of the import
f.cache.Recalculate()
return nil
} // end "small update" path - after this is the real bulk import path
smallWrite = true
// If the number of operations is small, we'll write directly to local storage
localBitmap = f.storage
} else {
// Create a temporary bitmap which will be populated by rowIDs and columnIDs
// and then merged into the existing fragment's bitmap.
localBitmap = roaring.NewBitmap()
// Create a temporary bitmap which will be populated by rowIDs and columnIDs
// and then merged into the existing fragment's bitmap.
localBitmap := roaring.NewBitmap()
// Disconnect op writer so we don't append updates.
localBitmap.OpWriter = nil
}
// Disconnect op writer so we don't append updates.
localBitmap.OpWriter = nil
// rowSet maintains the set of rowIDs present in this import.
// It allows the cache to be updated once per row, instead of once
// per bit.
// rowSet maintains the set of rowIDs present in this import. It allows the
// cache to be updated once per row, instead of once per bit. TODO: consider
// sorting by rowID/columnID first and avoiding the map allocation here. (we
// could reuse rowIDs to store the list of unique row IDs)
rowSet := make(map[uint64]struct{})
lastRowID := uint64(0)
// Process every bit by writing to a local bitmap,
// to be merged with fragment storage next.
for i := range rowIDs {
// replace columnIDs with calculated positions to avoid allocation.
for i := 0; i < len(columnIDs); i++ {
rowID, columnID := rowIDs[i], columnIDs[i]
// Determine the position of the bit in the storage.
pos, err := f.pos(rowID, columnID)
if err != nil {
return err
}
// Write to local storage.
_, err = localBitmap.Add(pos)
if err != nil {
return err
}
// Reduce the StatsD rate for high volume stats
f.stats.Count("ImportBit", 1, 0.0001)
columnIDs[i] = pos
// Add row to rowSet.
if i == 0 || rowID != lastRowID {
@ -1518,24 +1507,37 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor
rowSet[rowID] = struct{}{}
}
}
f.mu.Lock()
defer f.mu.Unlock()
// Merge localBitmap into fragment's existing data.
var results *roaring.Bitmap
positions := columnIDs
var changedN int
if options.Clear {
if f.storage.Count() > 0 {
results = f.storage.Difference(localBitmap)
changedN, err = localBitmap.RemoveN(positions...) // TODO benchmark AddN behavior with sorted/unsorted positions
} else {
changedN, err = localBitmap.AddN(positions...) // TODO benchmark RemoveN behavior with sorted/unsorted positions
}
if err != nil {
return errors.Wrap(err, "adding positions")
}
f.stats.Count("ImportBit", int64(changedN), 1)
f.opN += changedN
var results *roaring.Bitmap
if !smallWrite {
// Merge localBitmap into fragment's existing data.
if options.Clear {
if f.storage.Count() > 0 {
results = f.storage.Difference(localBitmap)
} else {
results = roaring.NewBitmap()
}
} else {
results = roaring.NewBitmap()
if f.storage.Count() > 0 {
results = f.storage.Union(localBitmap)
} else {
results = localBitmap
}
}
} else {
if f.storage.Count() > 0 {
results = f.storage.Union(localBitmap)
} else {
results = localBitmap
}
results = f.storage
}
// Update cache counts for all affected rows.
@ -1545,10 +1547,20 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor
n := results.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
f.cache.BulkAdd(rowID, n)
if smallWrite {
if _, ok := f.rowCache.Fetch(rowID); ok { // we won't update the rowCache if it wasn't already in there.
f.rowCache.Add(rowID, f.rowFromStorage(rowID))
}
}
}
f.cache.Recalculate()
return unprotectedWriteToFragment(f, results)
if !smallWrite {
return unprotectedWriteToFragment(f, results)
}
return nil
}
// bulkImportMutex performs a bulk import on a fragment while ensuring
@ -1645,23 +1657,36 @@ func (f *fragment) bulkImportMutex(rowIDs, columnIDs []uint64) error {
// importValue bulk imports a set of range-encoded values.
func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear bool) error {
f.mu.Lock()
defer f.mu.Unlock()
// Verify that there are an equal number of column ids and values.
if len(columnIDs) != len(values) {
return fmt.Errorf("mismatch of column/value len: %d != %d", len(columnIDs), len(values))
}
f.mu.RLock()
smallPath := false
if len(columnIDs)*int(bitDepth)/2+f.opN < f.MaxOpN {
smallPath = true
}
f.mu.RUnlock()
f.storage.OpWriter = nil
if !smallPath {
f.mu.Lock()
defer f.mu.Unlock()
f.storage.OpWriter = nil
}
// Process every value.
// If an error occurs then reopen the storage.
if err := func() error {
for i := range columnIDs {
columnID, value := columnIDs[i], values[i]
_, err := f.importSetValue(columnID, bitDepth, value, clear)
var err error
if smallPath {
_, err = f.setValueBase(columnID, bitDepth, value, clear)
} else {
_, err = f.importSetValue(columnID, bitDepth, value, clear)
}
if err != nil {
return errors.Wrap(err, "setting")
return errors.Wrapf(err, "setting value, smallPath:%v", smallPath)
}
}
return nil
@ -1670,8 +1695,11 @@ func (f *fragment) importValue(columnIDs, values []uint64, bitDepth uint, clear
_ = f.openStorage()
return err
}
if err := f.snapshot(); err != nil {
return errors.Wrap(err, "snapshotting")
if !smallPath {
if err := f.snapshot(); err != nil {
return errors.Wrap(err, "snapshotting")
}
}
return nil
}

109
roaring/roaring.go
View file

@ -175,6 +175,28 @@ func (b *Bitmap) Add(a ...uint64) (changed bool, err error) {
return changed, nil
}
// AddN adds values to the bitmap, appending them all to the op log in a batched write. It returns the number of changed bits.
func (b *Bitmap) AddN(a ...uint64) (changed int, err error) {
op := &op{
typ: opTypeAddBatch,
values: a,
}
if err := b.writeOp(op); err != nil {
return 0, errors.Wrap(err, "writing to op log")
}
// TODO consider applying changes in-memory first and then only writing the
// changed bits to op log?
for _, v := range a {
// Apply to the in-memory bitmap.
if b.DirectAdd(v) {
changed++
}
}
return changed, nil
}
// DirectAdd adds a value to the bitmap by bypassing the op log.
func (b *Bitmap) DirectAdd(v uint64) bool {
cont := b.Containers.GetOrCreate(highbits(v))
@ -210,6 +232,27 @@ func (b *Bitmap) Remove(a ...uint64) (changed bool, err error) {
return changed, nil
}
func (b *Bitmap) RemoveN(a ...uint64) (changed int, err error) {
op := &op{
typ: opTypeRemoveBatch,
values: a,
}
if err := b.writeOp(op); err != nil {
return 0, errors.Wrap(err, "writing to op log")
}
// TODO consider applying changes in-memory first and then only writing the
// changed bits to op log?
for _, v := range a {
// Apply to the in-memory bitmap.
if b.remove(v) {
changed++
}
}
return changed, nil
}
func (b *Bitmap) remove(v uint64) bool {
c := b.Containers.Get(highbits(v))
if c == nil {
@ -3487,26 +3530,38 @@ func shiftRun(a *Container) (*Container, bool) {
type opType uint8
const (
opTypeAdd = opType(0)
opTypeRemove = opType(1)
opTypeAdd = opType(0)
opTypeRemove = opType(1)
opTypeAddBatch = opType(2)
opTypeRemoveBatch = opType(3)
)
// op represents an operation on the bitmap.
type op struct {
typ opType
value uint64
typ opType
value uint64
values []uint64
}
// apply executes the operation against a bitmap.
func (op *op) apply(b *Bitmap) bool {
func (op *op) apply(b *Bitmap) (changed bool) {
switch op.typ {
case opTypeAdd:
return b.DirectAdd(op.value)
case opTypeRemove:
return b.remove(op.value)
case opTypeAddBatch:
for _, v := range op.values {
changed = changed || b.DirectAdd(v)
}
case opTypeRemoveBatch:
for _, v := range op.values {
changed = changed || b.remove(v)
}
default:
panic(fmt.Sprintf("invalid op type: %d", op.typ))
}
return changed
}
// WriteTo writes op to the w.
@ -3515,11 +3570,21 @@ func (op *op) WriteTo(w io.Writer) (n int64, err error) {
// Write type and value.
buf[0] = byte(op.typ)
binary.LittleEndian.PutUint64(buf[1:9], op.value)
if op.typ <= 1 {
binary.LittleEndian.PutUint64(buf[1:9], op.value)
} else {
binary.LittleEndian.PutUint64(buf[1:9], uint64(len(op.values)))
p := 13 // start of values (skip 4 for checksum)
for _, v := range op.values {
binary.LittleEndian.PutUint64(buf[p:p+8], v)
p += 8
}
}
// Add checksum at the end.
h := fnv.New32a()
h.Write(buf[0:9])
h.Write(buf[13:])
binary.LittleEndian.PutUint32(buf[9:13], h.Sum32())
// Write to writer.
@ -3527,29 +3592,49 @@ func (op *op) WriteTo(w io.Writer) (n int64, err error) {
return int64(nn), err
}
var minOpSize = 13
// UnmarshalBinary decodes data into an op.
func (op *op) UnmarshalBinary(data []byte) error {
if len(data) < op.size() {
if len(data) < minOpSize {
return fmt.Errorf("op data out of bounds: len=%d", len(data))
}
statsHit("op/UnmarshalBinary")
op.typ = opType(data[0])
// op.value will actually contain the length of values for batch ops
op.value = binary.LittleEndian.Uint64(data[1:9])
// Verify checksum.
h := fnv.New32a()
h.Write(data[0:9])
if op.typ > 1 {
if len(data) < int(13+op.value*8) {
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++ {
start := 13 + i*8
op.values[i] = binary.LittleEndian.Uint64(data[start : start+8])
}
op.value = 0
}
if chk := binary.LittleEndian.Uint32(data[9:13]); chk != h.Sum32() {
return fmt.Errorf("checksum mismatch: exp=%08x, got=%08x", h.Sum32(), chk)
}
// Read type and value.
op.typ = opType(data[0])
op.value = binary.LittleEndian.Uint64(data[1:9])
return nil
}
// size returns the encoded size of the op, in bytes.
func (*op) size() int { return 1 + 8 + 4 }
func (op *op) size() int {
if op.typ == opTypeAdd || op.typ == opTypeRemove {
return 1 + 8 + 4
}
return 1 + 8 + 4 + len(op.values)*8
}
func highbits(v uint64) uint64 { return v >> 16 }
func lowbits(v uint64) uint16 { return uint16(v & 0xFFFF) }

112
roaring/roaring_internal_test.go
View file

@ -23,6 +23,8 @@ import (
"runtime"
"strings"
"testing"
"github.com/pkg/errors"
)
// String produces a human viewable string of the contents.
@ -3437,3 +3439,113 @@ func TestShiftRun(t *testing.T) {
}
}
}
func TestOpLogWriteUnmarshal(t *testing.T) {
tests := []*op{
&op{
typ: opTypeAdd,
value: 27,
},
&op{
typ: opTypeRemove,
value: 28,
},
&op{
typ: opTypeAddBatch,
values: []uint64{1, 2, 6, 19},
},
&op{
typ: opTypeRemoveBatch,
values: []uint64{1, 2, 6, 19, 22, 44},
},
&op{
typ: opTypeAddBatch,
values: []uint64{51234567890},
},
&op{
typ: opTypeRemoveBatch,
values: []uint64{51234567890},
},
&op{
typ: opTypeAdd,
value: 0,
},
&op{
typ: opTypeRemove,
value: 0,
},
&op{
typ: opTypeAddBatch,
values: []uint64{0},
},
&op{
typ: opTypeRemoveBatch,
values: []uint64{0},
},
&op{
typ: opTypeAddBatch,
values: []uint64{},
},
&op{
typ: opTypeRemoveBatch,
values: []uint64{},
},
}
// test each one separately
for i, test := range tests {
t.Run(fmt.Sprintf("%d", i), func(t *testing.T) {
buf := &bytes.Buffer{}
if _, err := test.WriteTo(buf); err != nil {
t.Errorf("writing op %v to buffer: %v", test, err)
}
op := &op{}
if err := op.UnmarshalBinary(buf.Bytes()); err != nil {
t.Fatalf("unmarshling op: %v", err)
}
if err := compareOps(test, op); err != nil {
t.Errorf("mismatch: %v", err)
}
})
}
// now write them all to the same buffer and unmarshal one by one
t.Run("writeAllOps", func(t *testing.T) {
buf := &bytes.Buffer{}
for _, test := range tests {
_, err := test.WriteTo(buf)
if err != nil {
t.Fatalf("writing op to buffer: %v", err)
}
}
data := buf.Bytes()
offset := 0
for i, test := range tests {
op := &op{}
if err := op.UnmarshalBinary(data[offset:]); err != nil {
t.Fatalf("unmarshling op: %v", err)
}
if err := compareOps(test, op); err != nil {
t.Errorf("mismatch at %d: %v", i, err)
}
offset += op.size()
}
})
}
func compareOps(op1, op2 *op) error {
if op1.typ != op2.typ || op1.value != op2.value || len(op1.values) != len(op2.values) {
return errors.Errorf("mismatched type, value, or length: %v, %v", op1, op2)
}
for i := 0; i < len(op1.values); i++ {
if op1.values[i] != op2.values[i] {
return errors.Errorf("mismatched values at %d: %d and %d", i, op1.values[i], op2.values[i])
}
}
return nil
}