From 537ae99fb9c8445ff97f2cd9094681472a81a2c9 Mon Sep 17 00:00:00 2001 From: Matt Jaffee Date: Thu, 21 Feb 2019 18:12:23 -0600 Subject: [PATCH] add import support with aggregated op log writes --- fragment.go | 174 ++++++++++++++++++------------- roaring/roaring.go | 109 ++++++++++++++++--- roaring/roaring_internal_test.go | 112 ++++++++++++++++++++ 3 files changed, 310 insertions(+), 85 deletions(-) diff --git a/fragment.go b/fragment.go index fe7de7b49..0486904e7 100644 --- a/fragment.go +++ b/fragment.go @@ -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 } diff --git a/roaring/roaring.go b/roaring/roaring.go index 818184148..e97803812 100644 --- a/roaring/roaring.go +++ b/roaring/roaring.go @@ -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) } diff --git a/roaring/roaring_internal_test.go b/roaring/roaring_internal_test.go index 5ca0edb2b..337213afd 100644 --- a/roaring/roaring_internal_test.go +++ b/roaring/roaring_internal_test.go @@ -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 +}