scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-08-28 02:44:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #2024 from seebs/unmarshal3

Browse source 
Unmarshal3
This commit is contained in:
Matthew Jaffee 2019-07-01 14:39:40 -05:00 • committed by GitHub
commit 9df46353e7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
11 changed files with 1173 additions and 282 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

3
field.go
View file

@ -84,6 +84,8 @@ type Field struct {
remoteAvailableShards *roaring.Bitmap
logger logger.Logger
snapshotQueue chan *fragment
}
// FieldOption is a functional option type for pilosa.fieldOptions.
@ -838,6 +840,7 @@ func (f *Field) newView(path, name string) *view {
view.rowAttrStore = f.rowAttrStore
view.stats = f.Stats
view.broadcaster = f.broadcaster
view.snapshotQueue = f.snapshotQueue
return view
}

514
fragment.go
View file

@ -107,12 +107,21 @@ type fragment struct {
shard uint64
// File-backed storage
path string
flags byte // user-defined flags passed to roaring
file *os.File
storage *roaring.Bitmap
storageData []byte
opN int // number of ops since snapshot
path string
flags byte // user-defined flags passed to roaring
file *os.File
storage *roaring.Bitmap
storageData []byte
totalOpN int64 // total opN values
totalOps int64 // total ops (across all snapshots)
opN int // number of ops since snapshot (may be approximate for imports)
ops int // number of higher-level operations, as opposed to bit changes
snapshotsRequested int // number of times we've requested a snapshot
snapshotsTaken int // number of actual snapshot operations
snapshotting bool // set to true when requesting a snapshot, set to false after snapshot completes
snapshotCond sync.Cond
snapshotDelays int
snapshotDelayTime time.Duration
// Cache for row counts.
CacheType string // passed in by field
@ -145,11 +154,13 @@ type fragment struct {
mutexVector vector
stats stats.StatsClient
snapshotQueue chan *fragment
}
// newFragment returns a new instance of Fragment.
func newFragment(path, index, field, view string, shard uint64, flags byte) *fragment {
return &fragment{
f := &fragment{
path: path,
index: index,
field: field,
@ -164,11 +175,69 @@ func newFragment(path, index, field, view string, shard uint64, flags byte) *fra
stats: stats.NopStatsClient,
}
f.snapshotCond = sync.Cond{L: &f.mu}
return f
}
// cachePath returns the path to the fragment's cache data.
func (f *fragment) cachePath() string { return f.path + cacheExt }
// newSnapshotQueue makes a new snapshot queue, of depth N, and spawns a
// goroutine for it.
func newSnapshotQueue(n int, w int, l logger.Logger) chan *fragment {
ch := make(chan *fragment, n)
for i := 0; i < w; i++ {
go snapshotQueueWorker(ch, l)
}
return ch
}
func snapshotQueueWorker(snapshotQueue chan *fragment, l logger.Logger) {
for f := range snapshotQueue {
err := f.protectedSnapshot(true)
if err != nil {
l.Printf("snapshot error: %v", err)
}
f.snapshotCond.Broadcast()
}
}
// enqueueSnapshot requests that the fragment be snapshotted at some point
// in the future, if this has not already been requested. Call this only when
// the mutex is held.
func (f *fragment) enqueueSnapshot() {
f.snapshotsRequested++
if f.snapshotting {
return
}
f.snapshotting = true
if f.snapshotQueue != nil {
select {
case f.snapshotQueue <- f:
default:
before := time.Now()
// wait forever, but notice that we're waiting
f.snapshotQueue <- f
f.snapshotDelays++
f.snapshotDelayTime += time.Since(before)
if f.snapshotDelays >= 10 {
f.Logger.Printf("snapshotting %s: last ten enqueue delays took %v", f.path, f.snapshotDelayTime)
f.snapshotDelays = 0
f.snapshotDelayTime = 0
}
}
} else {
// in testing, for instance, there may be no holder, thus no one
// to handle these snapshots.
err := f.snapshot()
if err != nil {
f.Logger.Printf("snapshot failed: %v", err)
}
f.snapshotting = false
f.snapshotCond.Broadcast()
}
}
// Open opens the underlying storage.
func (f *fragment) Open() error {
f.mu.Lock()
@ -177,7 +246,7 @@ func (f *fragment) Open() error {
if err := func() error {
// Initialize storage in a function so we can close if anything goes wrong.
f.Logger.Debugf("open storage for index/field/view/fragment: %s/%s/%s/%d", f.index, f.field, f.view, f.shard)
if err := f.openStorage(); err != nil {
if err := f.openStorage(true); err != nil {
return errors.Wrap(err, "opening storage")
}
@ -215,12 +284,43 @@ func (f *fragment) reopen() (mustClose bool, err error) {
return mustClose, nil
}
// openStorage opens the storage bitmap.
func (f *fragment) openStorage() error {
// openStorage opens the storage bitmap. Usually you also want to read in
// the storage, but in the case where we just wrote that file, such as
// unprotectedWriteToFragment, we could also just... not. If we didn't
// have existing storage, we probably need to unmarshal the data. If the
// file we're asked to open is empty, we probably don't.
//
// If we already had mapped storage previously, we want to unmap that, and
// possibly remap it from the file, but we don't need a full unmarshal, just
// an update of mapped pointers.
//
// unmarshalData is somewhat overloaded. it tells us whether or not we
// need to actually create a bitmap from the data (if the data exists to
// do this from).
//
// usually unmarshalData is only set to false when we're in the middle of
// a snapshot, and unprotectedWriteToFragment just wrote the in-memory data
// out.
//
// If we have existing storage data, and we successfully get new data,
// we will unmap the existing storage data.
//
// This function's design is probably a problem -- it is trying to handle
// both cases where there was existing data before, and cases where we
// just wrote the data.
func (f *fragment) openStorage(unmarshalData bool) error {
oldStorageData := f.storageData
// there's a few places where we might encounter an error, but need
// to continue past it through other error checks, before returning it.
var lastError error
// Create a roaring bitmap to serve as storage for the shard.
if f.storage == nil {
f.storage = roaring.NewFileBitmap()
f.storage.Flags = f.flags
// if we didn't actually have storage, we *do* need to
// unmarshal this data in order to have any.
unmarshalData = true
}
// Open the data file to be mmap'd and used as an ops log.
file, mustClose, err := syswrap.OpenFile(f.path, os.O_RDWR|os.O_CREATE|os.O_APPEND, 0666)
@ -237,13 +337,24 @@ func (f *fragment) openStorage() error {
return fmt.Errorf("flock: %s", err)
}
// data is the data we would unmarshal from, if we're unmarshalling; it might
// be obtained by calling ReadAll on a file.
//
// newStorageData is the data we should map things to. it is set only if
// mmapped; if we didn't mmap (say, we couldn't), we won't want to unmap
// the ioutil byte slice. (Theoretically, we shouldn't be using the mapped
// flag in that case...)
var data []byte
var newStorageData []byte
// If the file is empty then initialize it with an empty bitmap.
fi, err := f.file.Stat()
if err != nil {
return errors.Wrap(err, "statting file before")
} else if fi.Size() == 0 {
bi := bufio.NewWriter(f.file)
if _, err := f.storage.WriteTo(bi); err != nil {
var err error
if _, err = f.storage.WriteTo(bi); err != nil {
return fmt.Errorf("init storage file: %s", err)
}
bi.Flush()
@ -251,39 +362,97 @@ func (f *fragment) openStorage() error {
if err != nil {
return errors.Wrap(err, "statting file after")
}
// there's nothing here, we're not going to try to unmarshal it.
unmarshalData = false
f.rowCache = &simpleCache{make(map[uint64]*Row)}
} else {
// Mmap the underlying file so it can be zero copied.
data, err := syswrap.Mmap(int(f.file.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
data, err = syswrap.Mmap(int(f.file.Fd()), 0, int(fi.Size()), syscall.PROT_READ, syscall.MAP_SHARED)
if err == syswrap.ErrMaxMapCountReached {
f.Logger.Debugf("maximum number of maps reached, reading file instead")
data, err = ioutil.ReadAll(file)
if err != nil {
return errors.Wrap(err, "failure file readall")
if unmarshalData {
data, err = ioutil.ReadAll(file)
if err != nil {
return errors.Wrap(err, "failure file readall")
}
}
} else if err != nil {
return errors.Wrap(err, "mmap failed")
} else {
f.storageData = data
// Advise the kernel that the mmap is accessed randomly.
if err := madvise(f.storageData, syscall.MADV_RANDOM); err != nil {
return fmt.Errorf("madvise: %s", err)
}
newStorageData = data
}
if err := f.storage.UnmarshalBinary(data); err != nil {
return fmt.Errorf("unmarshal storage: file=%s, err=%s", f.file.Name(), err)
}
}
f.opN = f.storage.Info().OpN
if unmarshalData {
f.storageData = newStorageData
// We're about to either re-read the bitmap, or fail to do so
// and unconditionally unmap the existing stuff. Either way, we
// want to unmap the old storage data after we're done here, but
// we can't unmap it yet because it's still live until sometime
// later, but we can't unmap it later, because we could return
// early... this is what defer is for.
if oldStorageData != nil {
defer func() {
unmapErr := syswrap.Munmap(oldStorageData)
if unmapErr != nil {
f.Logger.Printf("unmap of old storage failed: %s", err)
}
}()
}
// so we have a problem here: if this fails, it's unclear whether
// *either* or *both* of old and new storage data might be in use.
// So we call the thing that should unconditionally unmap both of them...
if err := f.storage.UnmarshalBinary(data); err != nil {
_, e2 := f.storage.RemapRoaringStorage(nil)
if e2 != nil {
return fmt.Errorf("unmarshal storage: file=%s, err=%s, clearing old mapping also failed: %v", f.file.Name(), err, e2)
}
return fmt.Errorf("unmarshal storage: file=%s, err=%s", f.file.Name(), err)
}
f.rowCache = &simpleCache{make(map[uint64]*Row)}
f.ops, f.opN = f.storage.Ops()
} else {
// we're moving to new storage, so instead of using the OpN
// derived from reading that storage, we notify the bitmap that
// OpN is now effectively zero.
f.opN = 0
f.ops = 0
f.storage.SetOps(0, 0)
// if oldStorageData is nil, this just tries to unmap any bits that
// are currently mapped. otherwise, it will point them at this
// storage (if the containers match).
var mappedAny bool
mappedAny, lastError = f.storage.RemapRoaringStorage(newStorageData)
if oldStorageData != nil {
unmapErr := syswrap.Munmap(oldStorageData)
if unmapErr != nil {
f.Logger.Printf("unmap of old storage failed: %s", err)
}
}
if mappedAny {
// Advise the kernel that the mmap is accessed randomly.
if err := madvise(newStorageData, syscall.MADV_RANDOM); err != nil {
lastError = fmt.Errorf("madvise: %s", err)
}
} else {
// if we did map data, but for some reason none of it got used
// as backing store, we can unmap it, and set the slice to nil,
// so we don't keep the now-invalid slice in f.storageData.
if newStorageData != nil {
unmapErr := syswrap.Munmap(newStorageData)
if unmapErr != nil {
lastError = fmt.Errorf("unmapping unused storage data: %s", err)
}
newStorageData = nil
}
}
f.storageData = newStorageData
}
// Attach the file to the bitmap to act as a write-ahead log.
f.storage.OpWriter = f.file
f.rowCache = &simpleCache{make(map[uint64]*Row)}
return nil
return lastError
}
// openCache initializes the cache from row ids persisted to disk.
@ -332,9 +501,31 @@ func (f *fragment) openCache() error {
func (f *fragment) Close() error {
f.mu.Lock()
defer f.mu.Unlock()
for f.snapshotting {
f.snapshotCond.Wait()
}
return f.close()
}
// awaitSnapshot lets us delay until the snapshot gets written, preventing tests
// from misleadingly showing amazingly fast performance because the snapshots they
// trigger haven't happened yet.
func (f *fragment) awaitSnapshot() {
f.mu.Lock()
defer f.mu.Unlock()
for f.snapshotting {
f.snapshotCond.Wait()
}
}
// unprotectedAwaitSnapshot assumes you already hold the lock, and waits for
// the snapshot fairy to come along.
func (f *fragment) unprotectedAwaitSnapshot() {
for f.snapshotting {
f.snapshotCond.Wait()
}
}
func (f *fragment) close() error {
// Flush cache if closing gracefully.
if err := f.flushCache(); err != nil {
@ -343,7 +534,7 @@ func (f *fragment) close() error {
}
// Close underlying storage.
if err := f.closeStorage(); err != nil {
if err := f.closeStorage(true); err != nil {
f.Logger.Printf("fragment: error closing storage: err=%s, path=%s", err, f.path)
return errors.Wrap(err, "closing storage")
}
@ -374,13 +565,19 @@ func (f *fragment) safeClose() error {
return nil
}
func (f *fragment) closeStorage() error {
// closeStorage attempts to close storage, including unmapping the old
// storage if includeMap is true. This would normally make sense if you're
// expecting to be done using the fragment, or to reload it. But it's also
// okay to just leave stuff mmapped; you don't have to keep the file
// descriptor open. So in some cases, we'll just leave the old mmapping
// in place, rather than regenerating everything from the new file.
func (f *fragment) closeStorage(includeMap bool) error {
// Clear the storage bitmap so it doesn't access the closed mmap.
//f.storage = roaring.NewBitmap()
// Unmap the file.
if f.storageData != nil {
if includeMap && f.storageData != nil {
if err := syswrap.Munmap(f.storageData); err != nil {
return fmt.Errorf("munmap: %s", err)
}
@ -502,9 +699,7 @@ func (f *fragment) unprotectedSetBit(rowID, columnID uint64) (changed bool, err
delete(f.checksums, int(rowID/HashBlockSize))
// Increment number of operations until snapshot is required.
if err := f.incrementOpN(); err != nil {
return false, errors.Wrap(err, "incrementing")
}
f.incrementOpN(1)
// If we're using a cache, update it. Otherwise skip the
// possibly-expensive count operation.
@ -566,9 +761,7 @@ func (f *fragment) unprotectedClearBit(rowID, columnID uint64) (changed bool, er
delete(f.checksums, int(rowID/HashBlockSize))
// Increment number of operations until snapshot is required.
if err := f.incrementOpN(); err != nil {
return false, errors.Wrap(err, "incrementing")
}
f.incrementOpN(1)
// If we're using a cache, update it. Otherwise skip the
// possibly-expensive count operation.
@ -629,14 +822,16 @@ func (f *fragment) unprotectedSetRow(row *Row, rowID uint64) (changed bool, err
}
// Update the row in cache.
n := f.storage.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
f.cache.BulkAdd(rowID, n)
// Snapshot storage.
if err := f.snapshot(); err != nil {
return false, errors.Wrap(err, "snapshotting")
if f.CacheType != CacheTypeNone {
n := f.storage.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
f.cache.BulkAdd(rowID, n)
}
// invalidate rowCache for this row.
f.rowCache.Add(rowID, nil)
// Snapshot storage.
f.enqueueSnapshot()
f.stats.Count("setRow", 1, 1.0)
return changed, nil
@ -677,11 +872,10 @@ func (f *fragment) unprotectedClearRow(rowID uint64) (changed bool, err error) {
// Clear the row in cache.
f.cache.Add(rowID, 0)
f.rowCache.Add(rowID, nil)
// Snapshot storage.
if err := f.snapshot(); err != nil {
return false, errors.Wrap(err, "snapshotting")
}
f.enqueueSnapshot()
f.stats.Count("clearRow", 1, 1.0)
@ -845,7 +1039,7 @@ func (f *fragment) setValueBase(columnID uint64, bitDepth uint, value int64, cle
}
// importSetValue is a more efficient SetValue just for imports.
func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, clear bool) (changed bool, err error) { // nolint: unparam
func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, clear bool) (changed int, err error) { // nolint: unparam
// Convert value to an unsigned representation.
uvalue := uint64(value)
if value < 0 {
@ -862,13 +1056,13 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, c
if c, err := f.storage.Add(bit); err != nil {
return changed, errors.Wrap(err, "adding")
} else if c {
changed = true
changed++
}
} else {
if c, err := f.storage.Remove(bit); err != nil {
return changed, errors.Wrap(err, "removing")
} else if c {
changed = true
changed++
}
}
}
@ -880,13 +1074,13 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, c
if c, err := f.storage.Remove(p); err != nil {
return changed, errors.Wrap(err, "removing not-null from storage")
} else if c {
changed = true
changed++
}
} else {
if c, err := f.storage.Add(p); err != nil {
return changed, errors.Wrap(err, "adding not-null to storage")
} else if c {
changed = true
changed++
}
}
@ -897,13 +1091,13 @@ func (f *fragment) importSetValue(columnID uint64, bitDepth uint, value int64, c
if c, err := f.storage.Remove(p); err != nil {
return changed, errors.Wrap(err, "removing sign from storage")
} else if c {
changed = true
changed++
}
} else {
if c, err := f.storage.Add(p); err != nil {
return changed, errors.Wrap(err, "adding sign to storage")
} else if c {
changed = true
changed++
}
}
@ -1855,19 +2049,12 @@ func (f *fragment) bulkImportStandard(rowIDs, columnIDs []uint64, options *Impor
// snapshot of the fragment or just do in-memory updates while appending
// operations to the op log.
func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct{}) error {
smallWrite := false
if len(set)+len(clear)+f.opN < f.MaxOpN {
smallWrite = true
mustClose, err := f.reopen()
if err != nil {
return errors.Wrap(err, "reopening")
}
if mustClose {
defer f.safeClose()
}
} else {
f.storage.OpWriter = nil
mustClose, err := f.reopen()
if err != nil {
return errors.Wrap(err, "reopening")
}
if mustClose {
defer f.safeClose()
}
if len(set) > 0 {
@ -1877,7 +2064,7 @@ func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct
return errors.Wrap(err, "adding positions")
}
f.stats.Count("ImportedN", int64(changedN), 1)
f.opN += changedN
f.incrementOpN(changedN)
}
if len(clear) > 0 {
@ -1887,7 +2074,7 @@ func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct
return errors.Wrap(err, "clearing positions")
}
f.stats.Count("ClearedN", int64(changedN), 1)
f.opN += changedN
f.incrementOpN(changedN)
}
// Update cache counts for all affected rows.
@ -1895,19 +2082,18 @@ func (f *fragment) importPositions(set, clear []uint64, rowSet map[uint64]struct
// Invalidate block checksum.
delete(f.checksums, int(rowID/HashBlockSize))
n := f.storage.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
f.cache.BulkAdd(rowID, n)
if smallWrite {
f.rowCache.Add(rowID, nil)
if f.CacheType != CacheTypeNone {
n := f.storage.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
f.cache.BulkAdd(rowID, n)
}
f.rowCache.Add(rowID, nil)
}
f.cache.Recalculate()
if !smallWrite {
return f.snapshot()
if f.CacheType != CacheTypeNone {
f.cache.Recalculate()
}
return nil
}
@ -1994,8 +2180,8 @@ func (f *fragment) importValueSmallWrite(columnIDs []uint64, values []int64, bit
}
return nil
}(); err != nil {
_ = f.closeStorage()
_ = f.openStorage()
_ = f.closeStorage(true)
_ = f.openStorage(true)
return err
}
rowSet := make(map[uint64]struct{}, bitDepth+1)
@ -2020,138 +2206,126 @@ func (f *fragment) importValue(columnIDs []uint64, values []int64, bitDepth uint
return errors.Wrap(f.importValueSmallWrite(columnIDs, values, bitDepth, clear), "import small write")
}
f.storage.OpWriter = nil
// Process every value.
// If an error occurs then reopen the storage.
f.storage.OpWriter = nil
totalChanges := 0
if err := func() (err error) {
for i := range columnIDs {
columnID, value := columnIDs[i], values[i]
if _, err := f.importSetValue(columnID, bitDepth, value, clear); err != nil {
changed, err := f.importSetValue(columnID, bitDepth, value, clear)
if err != nil {
return errors.Wrapf(err, "importSetValue")
}
totalChanges += changed
}
return nil
}(); err != nil {
_ = f.closeStorage()
_ = f.openStorage()
_ = f.closeStorage(true)
_ = f.openStorage(true)
return err
}
// We don't actually care, except we want our stats to be accurate.
f.incrementOpN(totalChanges)
err := f.snapshot()
return errors.Wrap(err, "snapshotting")
// in theory, this should probably have happened 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.
f.enqueueSnapshot()
f.unprotectedAwaitSnapshot()
return nil
}
// importRoaring imports from the official roaring data format defined at
// https://github.com/RoaringBitmap/RoaringFormatSpec or from pilosa's version
// of the roaring format. The cache is updated to reflect the new data.
func (f *fragment) importRoaring(ctx context.Context, data []byte, clear bool) error {
rowSize := uint64(1 << shardVsContainerExponent)
span, ctx := tracing.StartSpanFromContext(ctx, "fragment.importRoaring")
defer span.Finish()
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.AcquireFragmentLock")
f.mu.Lock()
defer f.mu.Unlock()
span.Finish()
bm := roaring.NewBTreeBitmap()
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.UnmarshalBinary")
err := bm.UnmarshalBinary(data)
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaring.ImportRoaringBits")
changed, rowSet, err := f.storage.ImportRoaringBits(data, clear, true, rowSize)
span.Finish()
if err != nil {
return err
}
// get a list of keys in order to update the cache
iter, _ := bm.Containers.Iterator(0)
rowSet := make(map[uint64]struct{})
var lastRow uint64 = math.MaxUint64
updateCache := f.CacheType != CacheTypeNone
anyChanged := false
incomingCnt := uint64(0)
for iter.Next() {
key, c := iter.Value()
incomingCnt += uint64(c.N())
// virtual row for the current container
vRow := key >> shardVsContainerExponent
// skip dups
if vRow == lastRow {
for rowID, changes := range rowSet {
if changes == 0 {
continue
}
rowSet[vRow] = struct{}{}
lastRow = vRow
}
// take smallPath? TODO - ideally instead of checking f.storage.Any(), the
// test here would be if the storage size (in bytes) is significantly
// greater than the size of the incoming bits serialized as append
// operations. Getting the storage size might be a bit expensive though
// especially if the fragment isn't mapped.
if incomingCnt+uint64(f.opN) <= uint64(f.MaxOpN) && f.storage.Any() {
toSet, toClear := bm.Slice(), []uint64{}
if clear {
toSet, toClear = toClear, toSet
f.rowCache.Add(rowID, nil)
if updateCache {
anyChanged = true
f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes))
}
span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.ImportPositions")
err := f.importPositions(toSet, toClear, rowSet)
span.Finish()
return err
}
// we only set this if we need to update the cache
if anyChanged {
f.cache.Recalculate()
}
if clear {
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaringDifference")
bm = f.storage.Difference(bm)
span.Finish()
} else if f.storage.Containers.Size() >= bm.Containers.Size() {
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaringStorageUIP")
f.storage.UnionInPlace(bm)
bm = f.storage
span.Finish()
} else {
span, ctx = tracing.StartSpanFromContext(ctx, "importRoaringBitmapUIP")
bm.UnionInPlace(f.storage)
span.Finish()
}
for rowID := range rowSet {
n := bm.CountRange(rowID*ShardWidth, (rowID+1)*ShardWidth)
f.cache.BulkAdd(rowID, n)
}
f.cache.Recalculate()
span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.WriteToFragment")
n, err := unprotectedWriteToFragment(f, bm)
span.LogKV("bytesWritten", n)
span, _ = tracing.StartSpanFromContext(ctx, "importRoaring.incrementOpN")
f.incrementOpN(changed)
span.Finish()
return err
return nil
}
// incrementOpN increase the operation count by one.
// If the count exceeds the maximum allowed then a snapshot is performed.
func (f *fragment) incrementOpN() error {
f.opN++
if f.opN <= f.MaxOpN {
return nil
func (f *fragment) incrementOpN(changed int) {
if changed <= 0 {
return
}
if err := f.snapshot(); err != nil {
return fmt.Errorf("snapshot: %s", err)
f.opN += changed
f.ops++
if f.opN > f.MaxOpN {
f.enqueueSnapshot()
}
return nil
}
// Snapshot writes the storage bitmap to disk and reopens it.
// Snapshot writes the storage bitmap to disk and reopens it. This may
// coexist with existing background-queue snapshotting; it does not remove
// things from the queue. You probably don't want to do this; use
// enqueueSnapshot/awaitSnapshot.
func (f *fragment) Snapshot() error {
f.mu.Lock()
defer f.mu.Unlock()
return f.snapshot()
}
func track(start time.Time, message string, stats stats.StatsClient, logger logger.Logger) {
elapsed := time.Since(start)
logger.Printf("%s took %s", message, elapsed)
stats.Histogram("snapshot", elapsed.Seconds(), 1.0)
}
// protectedSnapshot grabs the lock and unconditionally calls snapshot(). If
// fromQueue is true, the snapshotting state is also cleared.
func (f *fragment) protectedSnapshot(fromQueue bool) error {
f.mu.Lock()
defer f.mu.Unlock()
err := f.snapshot()
if fromQueue {
f.snapshotting = false
}
return err
}
// snapshot does the actual snapshot operation. it does not check or care
// about f.snapshotting.
func (f *fragment) snapshot() error {
f.totalOpN += int64(f.opN)
f.totalOps += int64(f.ops)
f.snapshotsTaken++
_, err := unprotectedWriteToFragment(f, f.storage)
return err
}
@ -2159,7 +2333,6 @@ func (f *fragment) snapshot() error {
// unprotectedWriteToFragment writes the fragment f with bm as the data. It is unprotected, and
// f.mu must be locked when calling it.
func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) (n int64, err error) { // nolint: interfacer
completeMessage := fmt.Sprintf("fragment: snapshot complete %s/%s/%s/%d", f.index, f.field, f.view, f.shard)
start := time.Now()
defer track(start, completeMessage, f.stats, f.Logger)
@ -2183,7 +2356,7 @@ func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) (n int64, err e
}
// Close current storage.
if err := f.closeStorage(); err != nil {
if err := f.closeStorage(false); err != nil {
return n, fmt.Errorf("close storage: %s", err)
}
@ -2192,8 +2365,12 @@ func unprotectedWriteToFragment(f *fragment, bm *roaring.Bitmap) (n int64, err e
return n, fmt.Errorf("rename snapshot: %s", err)
}
// if we reloaded from the file, we'd end up with this bitmap
// as our storage. so... let's use this bitmap. as our storage.
f.storage = bm
// Reopen storage.
if err := f.openStorage(); err != nil {
if err := f.openStorage(false); err != nil {
return n, fmt.Errorf("open storage: %s", err)
}
@ -2382,7 +2559,7 @@ func (f *fragment) readStorageFromArchive(r io.Reader) error {
}
// Close current storage.
if err := f.closeStorage(); err != nil {
if err := f.closeStorage(true); err != nil {
return errors.Wrap(err, "closing")
}
@ -2392,7 +2569,7 @@ func (f *fragment) readStorageFromArchive(r io.Reader) error {
}
// Reopen storage.
if err := f.openStorage(); err != nil {
if err := f.openStorage(true); err != nil {
return errors.Wrap(err, "opening")
}
@ -2489,6 +2666,13 @@ func filterWithRows(rows []uint64) rowFilter {
// this container have been processed. The rows accumulated up to this point
// (including this row if all filters passed) will be returned.
func (f *fragment) rows(start uint64, filters ...rowFilter) []uint64 {
f.mu.RLock()
defer f.mu.RUnlock()
return f.unprotectedRows(start, filters...)
}
// unprotectedRows calls rows without grabbing the mutex.
func (f *fragment) unprotectedRows(start uint64, filters ...rowFilter) []uint64 {
startKey := rowToKey(start)
i, _ := f.storage.Containers.Iterator(startKey)
rows := make([]uint64, 0)
@ -2916,9 +3100,10 @@ func newRowsVector(f *fragment) *rowsVector {
// Get returns the rowID associated to the given colID.
// Additionally, it returns true if a value was found,
// otherwise it returns false.
// otherwise it returns false. Ensure that you already
// have the mutex before calling this.
func (v *rowsVector) Get(colID uint64) (uint64, bool, error) {
rows := v.f.rows(0, filterColumn(colID))
rows := v.f.unprotectedRows(0, filterColumn(colID))
if len(rows) > 1 {
return 0, false, errors.New("found multiple row values for column")
} else if len(rows) == 1 {
@ -2949,9 +3134,10 @@ func newBoolVector(f *fragment) *boolVector {
// Get returns the rowID associated to the given colID.
// Additionally, it returns true if a value was found,
// otherwise it returns false.
// otherwise it returns false. Ensure that you already
// have the fragment mutex before calling this.
func (v *boolVector) Get(colID uint64) (uint64, bool, error) {
rows := v.f.rows(0, filterColumn(colID))
rows := v.f.unprotectedRows(0, filterColumn(colID))
if len(rows) > 1 {
return 0, false, errors.New("found multiple row values for column")
} else if len(rows) == 1 {

199
fragment_internal_test.go
View file

@ -117,7 +117,7 @@ func TestFragment_RowcacheMap(t *testing.T) {
_, _ = f.setBit(0, uint64(i*32))
}
// force snapshot so we get a mmapped row...
_ = f.snapshot()
_ = f.Snapshot()
row := f.row(0)
segment := row.Segments()[0]
bitmap := segment.data
@ -889,7 +889,6 @@ func BenchmarkFragment_RepeatedSmallValueImports(b *testing.B) {
b.Fatalf("importing values: %v", err)
}
}
}
})
}
@ -2037,16 +2036,17 @@ func BenchmarkFragment_Import(b *testing.B) {
}
var (
rowCases = []uint64{2, 50, 1000, 100000}
colCases = []uint64{20, 1000, 50000, 500000}
concurrencyCases = []int{2, 16}
rowCases = []uint64{2, 50, 1000, 10000, 100000}
colCases = []uint64{20, 1000, 5000, 50000, 500000}
concurrencyCases = []int{2, 4, 16}
cacheCases = []string{CacheTypeNone, CacheTypeRanked}
)
func BenchmarkImportRoaring(b *testing.B) {
for _, numRows := range rowCases {
data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
b.Logf("%dRows: %.2fMB\n", numRows, float64(len(data))/1024/1024)
for _, cacheType := range []string{CacheTypeRanked} { // CacheTypeNone didn't seem to affect the results much
for _, cacheType := range cacheCases {
b.Run(fmt.Sprintf("Rows%dCache_%s", numRows, cacheType), func(b *testing.B) {
b.StopTimer()
for i := 0; i < b.N; i++ {
@ -2054,6 +2054,7 @@ func BenchmarkImportRoaring(b *testing.B) {
b.StartTimer()
err := f.importRoaringT(data, false)
if err != nil {
f.awaitSnapshot()
f.Clean(b)
b.Fatalf("import error: %v", err)
}
@ -2070,63 +2071,30 @@ func BenchmarkImportRoaringConcurrent(b *testing.B) {
b.SkipNow()
}
for _, numRows := range rowCases {
data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
b.Logf("%dRows: %.2fMB\n", numRows, float64(len(data))/1024/1024)
data := make([][]byte, 0, len(concurrencyCases))
data = append(data, getZipfRowsSliceRoaring(numRows, 0, 0, ShardWidth))
b.Logf("%dRows: %.2fMB\n", numRows, float64(len(data[0]))/1024/1024)
for _, concurrency := range concurrencyCases {
b.Run(fmt.Sprintf("%dRows%dConcurrency", numRows, concurrency), func(b *testing.B) {
b.StopTimer()
frags := make([]*fragment, concurrency)
for i := 0; i < b.N; i++ {
for j := 0; j < concurrency; j++ {
frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), CacheTypeRanked)
}
eg := errgroup.Group{}
b.StartTimer()
for j := 0; j < concurrency; j++ {
j := j
eg.Go(func() error {
return frags[j].importRoaringT(data, false)
})
}
err := eg.Wait()
if err != nil {
b.Errorf("importing fragment: %v", err)
}
b.StopTimer()
for j := 0; j < concurrency; j++ {
frags[j].Clean(b)
}
}
})
}
}
}
func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
if testing.Short() {
b.SkipNow()
}
for _, numRows := range rowCases {
for _, numCols := range colCases {
data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
updata := getUpdataRoaring(numRows, numCols, 1)
for _, concurrency := range concurrencyCases {
b.Run(fmt.Sprintf("%dRows%dCols%dConcurrency", numRows, numCols, concurrency), func(b *testing.B) {
// add more data sets
for j := len(data); j < concurrency; j++ {
data = append(data, getZipfRowsSliceRoaring(numRows, int64(j), 0, ShardWidth))
}
for _, cacheType := range cacheCases {
b.Run(fmt.Sprintf("Rows%dConcurrency%dCache_%s", numRows, concurrency, cacheType), func(b *testing.B) {
b.StopTimer()
frags := make([]*fragment, concurrency)
for i := 0; i < b.N; i++ {
for j := 0; j < concurrency; j++ {
frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), CacheTypeRanked)
err := frags[j].importRoaringT(data, false)
if err != nil {
b.Fatalf("importing roaring: %v", err)
}
frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), cacheType)
}
eg := errgroup.Group{}
b.StartTimer()
for j := 0; j < concurrency; j++ {
j := j
eg.Go(func() error {
return frags[j].importRoaringT(updata, false)
err := frags[j].importRoaringT(data[j], false)
frags[j].awaitSnapshot()
return err
})
}
err := eg.Wait()
@ -2143,9 +2111,60 @@ func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
}
}
}
func BenchmarkImportRoaringUpdateConcurrent(b *testing.B) {
if testing.Short() {
b.SkipNow()
}
for _, numRows := range rowCases {
for _, numCols := range colCases {
data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
updata := getUpdataRoaring(numRows, numCols, 1)
for _, concurrency := range concurrencyCases {
for _, cacheType := range cacheCases {
b.Run(fmt.Sprintf("Rows%dCols%dConcurrency%dCache_%s", numRows, numCols, concurrency, cacheType), func(b *testing.B) {
b.StopTimer()
frags := make([]*fragment, concurrency)
for i := 0; i < b.N; i++ {
for j := 0; j < concurrency; j++ {
frags[j] = mustOpenFragment("i", "f", viewStandard, uint64(j), cacheType)
// the cost of actually doing the op log for the large initial data set
// is excessive. force storage into snapshotted state, then use import
// to generate an op log and/or snapshot.
_, _, err := frags[j].storage.ImportRoaringBits(data, false, false, 0)
frags[j].enqueueSnapshot()
frags[j].awaitSnapshot()
if err != nil {
b.Fatalf("importing roaring: %v", err)
}
}
eg := errgroup.Group{}
b.StartTimer()
for j := 0; j < concurrency; j++ {
j := j
eg.Go(func() error {
err := frags[j].importRoaringT(updata, false)
frags[j].awaitSnapshot()
return err
})
}
err := eg.Wait()
if err != nil {
b.Errorf("importing fragment: %v", err)
}
b.StopTimer()
for j := 0; j < concurrency; j++ {
frags[j].Clean(b)
}
}
})
}
}
}
}
}
func BenchmarkImportStandard(b *testing.B) {
for _, cacheType := range []string{CacheTypeRanked} {
for _, cacheType := range cacheCases {
for _, numRows := range rowCases {
rowIDsOrig, columnIDsOrig := getZipfRowsSliceStandard(numRows, 1, 0, ShardWidth)
rowIDs, columnIDs := make([]uint64, len(rowIDsOrig)), make([]uint64, len(columnIDsOrig))
@ -2171,23 +2190,29 @@ func BenchmarkImportStandard(b *testing.B) {
func BenchmarkImportRoaringUpdate(b *testing.B) {
fileSize := make(map[string]int64)
names := []string{}
for _, cacheType := range []string{CacheTypeRanked} {
for _, numRows := range rowCases {
for _, numCols := range colCases {
data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
updata := getUpdataRoaring(numRows, numCols, 1)
name := fmt.Sprintf("%s%dRows%dCols", cacheType, numRows, numCols)
for _, numRows := range rowCases {
data := getZipfRowsSliceRoaring(numRows, 1, 0, ShardWidth)
for _, numCols := range colCases {
updata := getUpdataRoaring(numRows, numCols, 1)
for _, cacheType := range cacheCases {
name := fmt.Sprintf("Rows%dCols%dCache_%s", numRows, numCols, cacheType)
names = append(names, name)
b.Run(name, func(b *testing.B) {
b.StopTimer()
for i := 0; i < b.N; i++ {
f := mustOpenFragment("i", fmt.Sprintf("r%dc%s", numRows, cacheType), viewStandard, 0, cacheType)
err := f.importRoaringT(data, false)
f := mustOpenFragment("i", fmt.Sprintf("r%dc%dcache_%s", numRows, numCols, cacheType), viewStandard, 0, cacheType)
// the cost of actually doing the op log for the large initial data set
// is excessive. force storage into snapshotted state, then use import
// to generate an op log and/or snapshot.
_, _, err := f.storage.ImportRoaringBits(data, false, false, 0)
f.enqueueSnapshot()
f.awaitSnapshot()
if err != nil {
b.Errorf("import error: %v", err)
}
b.StartTimer()
err = f.importRoaringT(updata, false)
f.awaitSnapshot()
if err != nil {
f.Clean(b)
b.Errorf("import error: %v", err)
@ -2400,19 +2425,23 @@ func getUpdataRoaring(numRows, numCols uint64, seed int64) []byte {
return buf.Bytes()
}
func getUpdataInto(f func(row, col uint64) bool, numRows, numCols uint64, seed int64) (changed int) {
func getUpdataInto(f func(row, col uint64) bool, numRows, numCols uint64, seed int64) int {
s := rand.NewSource(seed)
r := rand.New(s)
z := rand.NewZipf(r, 1.6, 50, numRows-1)
for i := uint64(0); i < numCols; i++ {
col := uint64(r.Int63n(ShardWidth)) // assuming the number of repeats will be negligible
i := uint64(0)
// ensure we get exactly the number we asked for. it turns out we had
// a horrible pathological edge case for exactly 10,000 entries in an
// imported bitmap, and hit it only occasionally...
for i < numCols {
col := uint64(r.Int63n(ShardWidth))
row := z.Uint64()
if f(row, col) {
changed++
i++
}
}
return changed
return int(i)
}
// getZipfRowsSliceStandard is the same as getZipfRowsSliceRoaring, but returns
@ -2462,13 +2491,39 @@ func BenchmarkFileWrite(b *testing.B) {
/////////////////////////////////////////////////////////////////////
func (f *fragment) sanityCheck(t testing.TB) {
newBM := roaring.NewFileBitmap()
file, err := os.Open(f.path)
if err != nil {
t.Fatalf("sanityCheck couldn't open file %s: %v", f.path, err)
}
defer file.Close()
data, err := ioutil.ReadAll(file)
if err != nil {
t.Fatalf("sanityCheck couldn't read fragment %s: %v", f.path, err)
}
err = newBM.UnmarshalBinary(data)
if err != nil {
t.Fatalf("sanityCheck couldn't unmarshal fragment %s: %v", f.path, err)
}
if equal, reason := newBM.BitwiseEqual(f.storage); !equal {
t.Fatalf("fragment %s: unmarshalled bitmap different: %v", f.path, reason)
}
}
func (f *fragment) Clean(t testing.TB) {
f.awaitSnapshot()
f.sanityCheck(t)
errc := f.Close()
errf := os.Remove(f.path)
errp := os.Remove(f.cachePath())
if errc != nil || errf != nil {
t.Fatal("cleaning up fragment: ", errc, errf, errp)
}
if f.snapshotQueue != nil {
close(f.snapshotQueue)
f.snapshotQueue = nil
}
// not all fragments have cache files
if errp != nil && !os.IsNotExist(errp) {
t.Fatalf("cleaning up fragment cache: %v", errp)
@ -2488,6 +2543,10 @@ func (f *fragment) CleanKeep(t testing.TB) {
if errc != nil {
t.Fatal("closing fragment: ", errc, errp)
}
if f.snapshotQueue != nil {
close(f.snapshotQueue)
f.snapshotQueue = nil
}
// not all fragments have cache files
if errp != nil && !os.IsNotExist(errp) {
t.Fatalf("cleaning up fragment cache: %v", errp)
@ -2520,6 +2579,7 @@ func mustOpenFragmentFlags(index, field, view string, shard uint64, cacheType st
f.RowAttrStore = &memAttrStore{
store: make(map[uint64]map[string]interface{}),
}
f.snapshotQueue = newSnapshotQueue(1, 1, nil)
if err := f.Open(); err != nil {
panic(err)
@ -3007,6 +3067,9 @@ func TestUnionInPlaceMapped(t *testing.T) {
f.storage.UnionInPlace(setBM1)
countUnion := f.storage.Count()
// UnionInPlace produces no ops log, we have to make it snapshot, to
// ensure that the on-disk representation is correct.
f.enqueueSnapshot()
if count0 != countF {
t.Fatalf("writing bitmap to storage changed count: %d => %d", count0, countF)
@ -3194,7 +3257,7 @@ func TestImportClearRestart(t *testing.T) {
f2.MaxOpN = maxOpN
f2.CacheType = f.CacheType
err = f.closeStorage()
err = f.closeStorage(true)
if err != nil {
t.Fatalf("closing storage: %v", err)
}
@ -3228,7 +3291,7 @@ func TestImportClearRestart(t *testing.T) {
f3.MaxOpN = maxOpN
f3.CacheType = f.CacheType
err = f2.closeStorage()
err = f2.closeStorage(true)
if err != nil {
t.Fatalf("f2 closing storage: %v", err)
}

13
holder.go
View file

@ -78,6 +78,8 @@ type Holder struct {
cacheFlushInterval time.Duration
Logger logger.Logger
snapshotQueue chan *fragment
}
// lockedChan looks a little ridiculous admittedly, but exists for good reason.
@ -152,6 +154,11 @@ func (h *Holder) Open() error {
return errors.Wrap(err, "reading directory")
}
// Run snapshots asynchronously. The snapshotQueue will have a background
// task associated with it which flushes it and waits until this channel
// is closed, so we should always close this channel when done.
h.snapshotQueue = newSnapshotQueue(100, 2, h.Logger)
for _, fi := range fis {
// Skip files or hidden directories.
if !fi.IsDir() || strings.HasPrefix(fi.Name(), ".") {
@ -203,6 +210,11 @@ func (h *Holder) Close() error {
return errors.Wrap(err, "closing index")
}
}
if h.snapshotQueue != nil {
close(h.snapshotQueue)
// assuming the snapshotQueueWorker has already started, this is safe.
h.snapshotQueue = nil
}
if h.translateFile != nil {
if err := h.translateFile.Close(); err != nil {
@ -425,6 +437,7 @@ func (h *Holder) newIndex(path, name string) (*Index, error) {
index.broadcaster = h.broadcaster
index.newAttrStore = h.NewAttrStore
index.columnAttrs = h.NewAttrStore(filepath.Join(index.path, ".data"))
index.snapshotQueue = h.snapshotQueue
return index, nil
}

4
index.go
View file

@ -53,7 +53,8 @@ type Index struct {
broadcaster broadcaster
Stats stats.StatsClient
logger logger.Logger
logger logger.Logger
snapshotQueue chan *fragment
}
// NewIndex returns a new instance of Index.
@ -408,6 +409,7 @@ func (i *Index) newField(path, name string) (*Field, error) {
f.Stats = i.Stats
f.broadcaster = i.broadcaster
f.rowAttrStore = i.newAttrStore(filepath.Join(f.path, ".data"))
f.snapshotQueue = i.snapshotQueue
return f, nil
}

6
roaring/btree.go
View file

@ -894,7 +894,7 @@ func (e *enumerator) Next() (k uint64, v *Container, err error) {
}
// Every iterates over a tree.
func (e *enumerator) Every(upd func(oldV *Container, exists bool) (newV *Container, write bool)) error {
func (e *enumerator) Every(upd func(key uint64, oldV *Container, exists bool) (newV *Container, write bool)) error {
if err := e.err; err != nil {
return err
}
@ -919,10 +919,10 @@ func (e *enumerator) Every(upd func(oldV *Container, exists bool) (newV *Contain
}
i := e.q.d[e.i]
nv, write := upd(i.v, true)
nv, write := upd(i.k, i.v, true)
if write {
if nv == nil {
e.t.Delete(e.q.d[e.i].k)
e.t.Delete(i.k)
} else {
e.q.d[e.i].v = nv
}

11
roaring/containers_btree.go
View file

@ -99,6 +99,10 @@ func (btc *bTreeContainers) PutContainerValues(key uint64, typ byte, n int, mapp
func (btc *bTreeContainers) Remove(key uint64) {
btc.tree.Delete(key)
if key == btc.lastKey {
btc.lastKey = ^uint64(0)
btc.lastContainer = nil
}
}
func (btc *bTreeContainers) GetOrCreate(key uint64) *Container {
@ -185,6 +189,11 @@ func (btc *bTreeContainers) Reset() {
btc.lastContainer = nil
}
func (btc *bTreeContainers) ResetN(n int) {
// we ignore n because it's impractical to preallocate the tree
btc.Reset()
}
func (btc *bTreeContainers) Iterator(key uint64) (citer ContainerIterator, found bool) {
e, ok := btc.tree.Seek(key)
if ok {
@ -215,7 +224,7 @@ func (btc *bTreeContainers) Update(key uint64, fn func(*Container, bool) (*Conta
// UpdateEvery calls fn (existing-container, existed), and expects
// (new-container, write). If write is true, the container is used to
// replace the given container.
func (btc *bTreeContainers) UpdateEvery(fn func(*Container, bool) (*Container, bool)) {
func (btc *bTreeContainers) UpdateEvery(fn func(uint64, *Container, bool) (*Container, bool)) {
e, _ := btc.tree.Seek(0)
// currently not handling the error from this, but in practice it has
// to be io.EOF.

20
roaring/containers_slice.go
View file

@ -75,6 +75,10 @@ func (sc *sliceContainers) Remove(key uint64) {
if i < 0 {
return
}
if key == sc.lastKey {
sc.lastKey = ^uint64(0)
sc.lastContainer = nil
}
sc.keys = append(sc.keys[:i], sc.keys[i+1:]...)
sc.containers = append(sc.containers[:i], sc.containers[i+1:]...)
@ -158,6 +162,18 @@ func (sc *sliceContainers) Reset() {
sc.lastKey = 0
}
func (sc *sliceContainers) ResetN(n int) {
if cap(sc.keys) < n {
sc.keys = make([]uint64, 0, n)
sc.containers = make([]*Container, 0, n)
} else {
sc.keys = sc.keys[:0]
sc.containers = sc.containers[:0]
}
sc.lastContainer = nil
sc.lastKey = 0
}
func (sc *sliceContainers) seek(key uint64) (int, bool) {
i := search64(sc.keys, key)
found := true
@ -204,9 +220,9 @@ func (sc *sliceContainers) Update(key uint64, fn func(*Container, bool) (*Contai
// UpdateEvery calls fn (existing-container, existed), and expects
// (new-container, write). If write is true, the container is used to
// replace the given container.
func (sc *sliceContainers) UpdateEvery(fn func(*Container, bool) (*Container, bool)) {
func (sc *sliceContainers) UpdateEvery(fn func(uint64, *Container, bool) (*Container, bool)) {
for i, c := range sc.containers {
nc, write := fn(c, true)
nc, write := fn(sc.keys[i], c, true)
if write {
sc.containers[i] = nc
}

8
roaring/fuzz_test.go
View file

@ -24,13 +24,13 @@ func TestUnmarshalBinary(t *testing.T) {
expected string
}{
{ // Checks for the zero containers situation
cr: []byte(":0\x000\x01\x00\x00\x000000"), //":000000"
cr: []byte(":0\x00\x00\x01\x00\x00\x000000"), //":000000"
expected: "reading roaring header: malformed bitmap, key-cardinality slice overruns buffer at 12",
},
{ // Checks for int overflow
cr: []byte("<0\x000\x00\x00\x00\x00000000000000" +
"0"), //"<000000000000000"
expected: "unmarshaling as pilosa roaring: Maximum operation size exceeded",
expected: "unmarshaling as pilosa roaring: unknown op type: 48",
},
{ // The next 5 check for malformed bitmaps
cr: []byte("<0\x0000000000000000000" +
@ -58,11 +58,11 @@ func TestUnmarshalBinary(t *testing.T) {
expected: "unmarshaling as pilosa roaring: malformed bitmap, key-cardinality not provided for 0 containers",
},
{ // Checks for incomplete offset in readWithRuns
cr: []byte(";0\x000\v00000"), //";00 00000"
cr: []byte(";0\x00\x00\v00000"), //";00 00000"
expected: "reading offsets from official roaring format: offset incomplete: len=10",
},
{ // Checks for incomplete offset in readOffsets
cr: []byte(":0\x000\x03\x00\x00\x00000000000000" +
cr: []byte(":0\x00\x00\x03\x00\x00\x00000000000000" +
"\x00"), //:0000000000000
expected: "reading offsets from official roaring format: offset incomplete: len=1",
},

663
roaring/roaring.go
View file

@ -117,7 +117,7 @@ type Containers interface {
// UpdateEvery calls fn (existing-container, existed), and expects
// (new-container, write). If write is true, the container is used to
// replace the given container.
UpdateEvery(fn func(*Container, bool) (*Container, bool))
UpdateEvery(fn func(uint64, *Container, bool) (*Container, bool))
// Iterator returns a Contiterator which after a call to Next(), a call to Value() will
// return the first container at or after key. found will be true if a
@ -128,6 +128,8 @@ type Containers interface {
// Reset clears the containers collection to allow for recycling during snapshot
Reset()
// ResetN clears the collection but hints at a needed size.
ResetN(int)
// Repair will repair the cardinality of any containers whose cardinality were corrupted
// due to optimized operations.
@ -147,8 +149,9 @@ type Bitmap struct {
Flags byte
// Number of bit change operations written to the writer. Some operations
// contain multiple values, each of those counts the number of values rather
// than counting as one operation.
// contain multiple values, so "ops" represents the number of distinct
// operations, while "opN" represents expected bit changes.
ops int
opN int
// Writer where operations are appended to.
@ -997,7 +1000,7 @@ func (b *Bitmap) countEmptyContainers() int {
// Optimize converts array and bitmap containers to run containers as necessary.
func (b *Bitmap) Optimize() {
b.Containers.UpdateEvery(func(c *Container, existed bool) (*Container, bool) {
b.Containers.UpdateEvery(func(key uint64, c *Container, existed bool) (*Container, bool) {
return c.optimize(), true
})
}
@ -1118,6 +1121,442 @@ func (b *Bitmap) writeToUnoptimized(w io.Writer) (n int64, err error) {
return n, nil
}
// roaringIterator represents something which can iterate through a roaring
// bitmap and yield information about containers, including type, size, and
// the location of their data structures.
type roaringIterator interface {
Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error)
}
// baseRoaringIterator holds values used by both Pilosa and official Roaring
// iterators.
type baseRoaringIterator struct {
data []byte
keys int64
headers []byte
offsets []byte
currentKey uint64
currentIdx int64
currentType byte
currentN int
currentLen int
currentPointer *uint16
currentDataOffset uint32
lastErr error
}
// okay, then
func (b *baseRoaringIterator) SilenceLint() {
// these are actually used in pilosaRoaringIterator or officialRoaringIterator
// but structcheck doesn't know that
_ = b.data
_ = b.keys
_ = b.offsets
_ = b.headers
_ = b.currentIdx
}
type pilosaRoaringIterator struct {
baseRoaringIterator
}
type officialRoaringIterator struct {
baseRoaringIterator
containerTyper func(index uint, card int) byte
haveRuns bool
}
func newOfficialRoaringIterator(data []byte) (*officialRoaringIterator, error) {
r := &officialRoaringIterator{}
r.data = data
// share code with the existing unmarshal code
var offsetOffset, headerOffset int
var err error
var keys uint32
// we ignore the flags, since we don't have to process them for anything.
keys, r.containerTyper, headerOffset, offsetOffset, r.haveRuns, err = readOfficialHeader(data)
if err != nil {
return nil, fmt.Errorf("reading official header: %v", err)
}
if keys == 0 {
// not an error, exactly. it's valid and well-formed, we just have nothing to do
r.Done(io.EOF)
return r, nil
}
r.keys = int64(keys)
r.headers = data[headerOffset:offsetOffset]
// note: offsets are only actually used with the no-run headers.
if r.haveRuns {
// start out pointed at where the offsets would have been.
r.currentDataOffset = uint32(offsetOffset)
} else {
r.offsets = data[offsetOffset : offsetOffset+int(r.keys*4)]
}
// set key to -1; user should call Next first.
r.currentIdx = -1
r.currentKey = ^uint64(0)
r.lastErr = errors.New("tried to read iterator without calling Next first")
return r, nil
}
func newPilosaRoaringIterator(data []byte) (*pilosaRoaringIterator, error) {
fileVersion := uint32(data[2])
if fileVersion != storageVersion {
return nil, fmt.Errorf("wrong roaring version, file is v%d, server requires v%d", fileVersion, storageVersion)
}
r := &pilosaRoaringIterator{}
r.data = data
// Read key count in bytes sizeof(cookie)+sizeof(flag):(sizeof(cookie)+sizeof(uint32)).
r.keys = int64(binary.LittleEndian.Uint32(data[3+1 : 8]))
// it could happen
if r.keys == 0 {
// not an error, exactly. it's valid and well-formed, we just have nothing to do
r.Done(io.EOF)
return r, nil
}
if int64(len(data)) < int64(headerBaseSize+(r.keys*16)) {
return nil, fmt.Errorf("insufficient data for header + offsets: want %d bytes, got %d",
headerBaseSize+(r.keys*16), len(data))
}
headerStart := int64(headerBaseSize)
headerEnd := headerStart + (r.keys * 12)
offsetStart := headerEnd
offsetEnd := offsetStart + (r.keys * 4)
r.headers = data[headerStart:headerEnd]
r.offsets = data[offsetStart:offsetEnd]
// set key to -1; user should call Next first.
r.currentIdx = -1
r.currentKey = ^uint64(0)
r.lastErr = errors.New("tried to read iterator without calling Next first")
return r, nil
}
func newRoaringIterator(data []byte) (roaringIterator, error) {
if len(data) < headerBaseSize {
return nil, errors.New("invalid data: not long enough to be a roaring header")
}
// Verify the first two bytes are a valid MagicNumber, and second two bytes match current storageVersion.
fileMagic := uint32(binary.LittleEndian.Uint16(data[0:2]))
switch fileMagic {
case serialCookie, serialCookieNoRunContainer:
return newOfficialRoaringIterator(data)
case MagicNumber:
return newPilosaRoaringIterator(data)
}
return nil, fmt.Errorf("unknown roaring magic number %d", fileMagic)
}
// Done marks the iterator as complete, recording err as the reason why
func (r *baseRoaringIterator) Done(err error) {
r.lastErr = err
r.currentKey = ^uint64(0)
r.currentType = 0
r.currentN = 0
r.currentLen = 0
r.currentPointer = nil
r.currentDataOffset = 0
}
func (r *pilosaRoaringIterator) Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
if r.currentIdx >= r.keys {
// we're already done
return r.Current()
}
r.currentIdx++
if r.currentIdx == r.keys {
// this is the last key. transition state to the finalized state
r.Done(io.EOF)
return r.Current()
}
header := r.headers[r.currentIdx*12:]
r.currentKey = binary.LittleEndian.Uint64(header[0:8])
r.currentType = byte(binary.LittleEndian.Uint16(header[8:10]))
r.currentN = int(binary.LittleEndian.Uint16(header[10:12])) + 1
r.currentDataOffset = binary.LittleEndian.Uint32(r.offsets[r.currentIdx*4:])
// a run container keeps its data after an initial 2 byte length header
var runCount uint16
if r.currentType == containerRun {
runCount = binary.LittleEndian.Uint16(r.data[r.currentDataOffset : r.currentDataOffset+runCountHeaderSize])
r.currentDataOffset += 2
}
if r.currentDataOffset > uint32(len(r.data)) || r.currentDataOffset < headerBaseSize {
r.Done(fmt.Errorf("container %d/%d, key %d, had offset %d, maximum %d",
r.currentIdx, r.keys, r.currentKey, r.currentDataOffset, len(r.data)))
return r.Current()
}
r.currentPointer = (*uint16)(unsafe.Pointer(&r.data[r.currentDataOffset]))
var size int
switch r.currentType {
case containerArray:
r.currentLen = r.currentN
size = r.currentLen * 2
case containerBitmap:
r.currentLen = 1024
size = 8192
case containerRun:
r.currentLen = int(runCount)
size = r.currentLen * 4
}
if int64(r.currentDataOffset)+int64(size) > int64(len(r.data)) {
r.Done(fmt.Errorf("container %d/%d, key %d, had offset %d+%d size, maximum %d",
r.currentIdx, r.keys, r.currentKey, r.currentDataOffset, size, len(r.data)))
return r.Current()
}
r.lastErr = nil
return r.Current()
}
func (r *officialRoaringIterator) Next() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
if r.currentIdx >= r.keys {
// we're already done
return r.Current()
}
r.currentIdx++
if r.currentIdx == r.keys {
// this is the last key. transition state to the finalized state
r.Done(io.EOF)
return r.Current()
}
header := r.headers[r.currentIdx*4:]
r.currentKey = uint64(binary.LittleEndian.Uint16(header[0:2]))
r.currentN = int(binary.LittleEndian.Uint16(header[2:4])) + 1
r.currentType = r.containerTyper(uint(r.currentIdx), r.currentN)
// with runs, we can't actually look up offsets; the format just stores
// things sequentially. so we have to actually track the offset in that case.
if !r.haveRuns {
r.currentDataOffset = binary.LittleEndian.Uint32(r.offsets[r.currentIdx*4:])
}
// a run container keeps its data after an initial 2 byte length header
var runCount uint16
if r.currentType == containerRun {
runCount = binary.LittleEndian.Uint16(r.data[r.currentDataOffset : r.currentDataOffset+runCountHeaderSize])
r.currentDataOffset += 2
}
if r.currentDataOffset > uint32(len(r.data)) || r.currentDataOffset < headerBaseSize {
r.Done(fmt.Errorf("container %d/%d, key %d, had offset %d, maximum %d",
r.currentIdx, r.keys, r.currentKey, r.currentDataOffset, len(r.data)))
return r.Current()
}
r.currentPointer = (*uint16)(unsafe.Pointer(&r.data[r.currentDataOffset]))
var size int
switch r.currentType {
case containerArray:
r.currentLen = r.currentN
size = r.currentLen * 2
case containerBitmap:
r.currentLen = 1024
size = 8192
case containerRun:
// official format stores runs as start/len, we want to convert, but since
// they might be mmapped, we can't write to that memory
newRuns := make([]interval16, runCount)
oldRuns := (*[65536]interval16)(unsafe.Pointer(r.currentPointer))[:runCount:runCount]
copy(newRuns, oldRuns)
for i := range newRuns {
newRuns[i].last += newRuns[i].start
}
r.currentPointer = (*uint16)(unsafe.Pointer(&newRuns[0]))
r.currentLen = int(runCount)
size = r.currentLen * 4
}
if int64(r.currentDataOffset)+int64(size) > int64(len(r.data)) {
r.Done(fmt.Errorf("container %d/%d, key %d, had offset %d+%d size, maximum %d",
r.currentIdx, r.keys, r.currentKey, r.currentDataOffset, size, len(r.data)))
return r.Current()
}
r.currentDataOffset += uint32(size)
r.lastErr = nil
return r.Current()
}
func (r *baseRoaringIterator) Current() (key uint64, cType byte, n int, length int, pointer *uint16, err error) {
return r.currentKey, r.currentType, r.currentN, r.currentLen, r.currentPointer, r.lastErr
}
// RemapRoaringStorage tries to update all containers to refer to
// the roaring bitmap in the provided []byte. If any containers are
// marked as mapped, but do not match the provided storage, they will
// be unmapped. The boolean return indicates whether or not any
// containers were mapped to the given storage.
//
// Regardless, after this function runs, no containers have
// mapped storage which does not refer to data; either they got mapped
// to the new storage, or storage was allocated for them.
func (b *Bitmap) RemapRoaringStorage(data []byte) (mappedAny bool, returnErr error) {
if b.Containers == nil {
return false, nil
}
var itr roaringIterator
var err error
var itrKey uint64
var itrCType byte
var itrN int
var itrPointer *uint16
var itrErr error
if data != nil {
itr, err = newRoaringIterator(data)
}
// don't return early: we still have to do the unmapping
if err != nil {
returnErr = err
}
if itr != nil {
itrKey, itrCType, itrN, _, itrPointer, itrErr = itr.Next()
}
if itrErr != nil {
// iterator errored out, so we won't check it in the loop below
itr = nil
}
b.Containers.UpdateEvery(func(key uint64, oldC *Container, existed bool) (newC *Container, write bool) {
if itr != nil {
for itrKey < key && itrErr == nil {
itrKey, itrCType, itrN, _, itrPointer, itrErr = itr.Next()
}
if itrErr != nil {
itr = nil
}
// container might be similar enough that we should trust it:
if itrKey == key && itrCType == oldC.typ() && itrN == int(oldC.N()) {
if oldC.frozen() {
// we don't use Clone, because that would copy the
// storage, and we don't need that.
halfCopy := *oldC
halfCopy.flags &^= flagFrozen
newC = &halfCopy
} else {
newC = oldC
}
mappedAny = true
newC.pointer = itrPointer
newC.flags |= flagMapped
return newC, true
}
}
// if the container isn't mapped, we don't need to do anything
if !oldC.Mapped() {
return oldC, false
}
// forcibly unmap it, so the old mapping can be unmapped safely.
newC = oldC.unmapOrClone()
return newC, true
})
return mappedAny, returnErr
}
// ImportRoaringBits sets-or-clears bits based on a provided Roaring bitmap.
// This should be equivalent to unmarshalling the bitmap, then executing
// either `b = Union(b, newB)` or `b = Difference(b, newB)`, but with lower
// overhead. The log parameter controls whether to write to the op log; the
// answer should always be yes, except if you're calling using this to apply
// the op log.
//
// If rowSize is non-zero, we should return a map of rows we altered,
// where "rows" are sets of rowSize containers. Otherwise the map isn't used.
// (This allows ImportRoaring to update caches; see fragment.go.)
func (b *Bitmap) ImportRoaringBits(data []byte, clear bool, log bool, rowSize uint64) (changed int, rowSet map[uint64]int, err error) {
if data == nil {
return 0, nil, errors.New("no roaring bitmap provided")
}
var itr roaringIterator
var itrKey uint64
var itrCType byte
var itrN int
var itrLen int
var itrPointer *uint16
var itrErr error
itr, err = newRoaringIterator(data)
if err != nil {
return 0, nil, err
}
if itr == nil {
return 0, nil, errors.New("failed to create roaring iterator, but don't know why")
}
rowSet = make(map[uint64]int)
var synthC Container
var importUpdater func(*Container, bool) (*Container, bool)
var currRow uint64
if clear {
importUpdater = func(oldC *Container, existed bool) (newC *Container, write bool) {
existN := oldC.N()
if existN == 0 || !existed {
return nil, false
}
newC = difference(oldC, &synthC)
if newC.N() != existN {
changes := int(existN - newC.N())
changed += changes
rowSet[currRow] -= changes
return newC, true
}
return oldC, false
}
} else {
importUpdater = func(oldC *Container, existed bool) (newC *Container, write bool) {
existN := oldC.N()
if existN == maxContainerVal+1 {
return oldC, false
}
if existN == 0 {
newerC := synthC.Clone()
changed += int(newerC.N())
rowSet[currRow] += int(newerC.N())
return newerC, true
}
newC = oldC.unionInPlace(&synthC)
if newC.typeID == containerBitmap {
newC.Repair()
}
if newC.N() != existN {
changes := int(newC.N() - existN)
changed += changes
rowSet[currRow] += changes
return newC, true
}
return oldC, false
}
}
itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next()
for itrErr == nil {
synthC.typeID = itrCType
synthC.n = int32(itrN)
synthC.len = int32(itrLen)
synthC.cap = int32(itrLen)
synthC.pointer = itrPointer
if rowSize != 0 {
currRow = itrKey / rowSize
}
b.Containers.Update(itrKey, importUpdater)
itrKey, itrCType, itrN, itrLen, itrPointer, itrErr = itr.Next()
}
// note: if we get a non-EOF err, it's possible that we made SOME
// changes but didn't log them. I don't have a good solution to this.
if itrErr != io.EOF {
return changed, rowSet, itrErr
}
err = nil
if log {
op := op{opN: changed, roaring: data}
if clear {
op.typ = opTypeRemoveRoaring
} else {
op.typ = opTypeAddRoaring
}
err = b.writeOp(&op)
}
return changed, rowSet, err
}
// unmarshalPilosaRoaring treats data as being encoded in Pilosa's 64 bit
// roaring format and decodes it into b.
func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
@ -1144,7 +1583,7 @@ func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
}
headerSize := headerBaseSize
b.Containers.Reset()
b.Containers.ResetN(int(keyN))
// Descriptive header section: Read container keys and cardinalities.
for i, buf := 0, data[headerSize:]; i < int(keyN); i, buf = i+1, buf[12:] {
b.Containers.PutContainerValues(
@ -1203,6 +1642,7 @@ func (b *Bitmap) unmarshalPilosaRoaring(data []byte) error {
opr.apply(b)
// Increase the op count.
b.ops++
b.opN += opr.count()
// Move the buffer forward.
@ -1222,6 +1662,7 @@ func (b *Bitmap) writeOp(op *op) error {
return err
}
b.opN += op.count()
b.ops++
return nil
}
@ -1233,10 +1674,23 @@ func (b *Bitmap) Iterator() *Iterator {
return itr
}
// Ops returns the number of write ops the bitmap is aware of in its ops
// log, and their total bit count.
func (b *Bitmap) Ops() (ops int, opN int) {
return b.ops, b.opN
}
// SetOps lets us reset the operation count in the weird case where we know
// we've changed an underlying file, without actually refreshing the bitmap.
func (b *Bitmap) SetOps(ops int, opN int) {
b.ops, b.opN = ops, opN
}
// Info returns stats for the bitmap.
func (b *Bitmap) Info() bitmapInfo {
info := bitmapInfo{
OpN: b.opN,
Ops: b.ops,
Containers: make([]containerInfo, 0, b.Containers.Size()),
}
@ -1305,6 +1759,7 @@ func (b *Bitmap) Flip(start, end uint64) *Bitmap {
// bitmapInfo represents a point-in-time snapshot of bitmap stats.
type bitmapInfo struct {
OpN int
Ops int
Containers []containerInfo
}
@ -3961,17 +4416,21 @@ func shiftRun(a *Container) (*Container, bool) {
type opType uint8
const (
opTypeAdd = opType(0)
opTypeRemove = opType(1)
opTypeAddBatch = opType(2)
opTypeRemoveBatch = opType(3)
opTypeAdd = opType(0)
opTypeRemove = opType(1)
opTypeAddBatch = opType(2)
opTypeRemoveBatch = opType(3)
opTypeAddRoaring = opType(4)
opTypeRemoveRoaring = opType(5)
)
// op represents an operation on the bitmap.
type op struct {
typ opType
value uint64
values []uint64
typ opType
opN int
value uint64
values []uint64
roaring []byte
}
// apply executes the operation against a bitmap.
@ -3985,6 +4444,12 @@ func (op *op) apply(b *Bitmap) (changed bool) {
changed = b.DirectAddN(op.values...) > 0
case opTypeRemoveBatch:
changed = b.DirectRemoveN(op.values...) > 0
case opTypeAddRoaring:
changedN, _, _ := b.ImportRoaringBits(op.roaring, false, false, 0)
changed = changedN != 0
case opTypeRemoveRoaring:
changedN, _, _ := b.ImportRoaringBits(op.roaring, true, false, 0)
changed = changedN != 0
default:
panic(fmt.Sprintf("invalid op type: %d", op.typ))
}
@ -3993,29 +4458,47 @@ func (op *op) apply(b *Bitmap) (changed bool) {
// WriteTo writes op to the w.
func (op *op) WriteTo(w io.Writer) (n int64, err error) {
buf := make([]byte, op.size())
buf := make([]byte, op.encodeSize())
// Write type and value.
buf[0] = byte(op.typ)
if op.typ <= 1 {
switch op.typ {
case opTypeAdd, opTypeRemove:
binary.LittleEndian.PutUint64(buf[1:9], op.value)
} else {
case opTypeAddBatch, opTypeRemoveBatch:
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
}
case opTypeAddRoaring, opTypeRemoveRoaring:
binary.LittleEndian.PutUint64(buf[1:9], uint64(len(op.roaring)))
binary.LittleEndian.PutUint32(buf[13:17], uint32(op.opN))
default:
return 0, fmt.Errorf("can't marshal unknown op type %d", op.typ)
}
// Add checksum at the end.
h := fnv.New32a()
_, _ = h.Write(buf[0:9])
_, _ = h.Write(buf[13:])
if op.typ == 4 || op.typ == 5 {
_, _ = h.Write(op.roaring)
}
binary.LittleEndian.PutUint32(buf[9:13], h.Sum32())
// Write to writer.
nn, err := w.Write(buf)
if err != nil {
return int64(nn), err
}
if op.typ == 4 || op.typ == 5 {
var nn2 int
// separate write so we don't have to copy the whole thing
nn2, err = w.Write(op.roaring)
nn += nn2
}
return int64(nn), err
}
@ -4030,18 +4513,22 @@ func (op *op) UnmarshalBinary(data []byte) error {
statsHit("op/UnmarshalBinary")
op.typ = opType(data[0])
// op.value will actually contain the length of values for batch ops
// op.value will actually contain the length of values for batch ops, or
// length of the roaring bitmap for roaring bitmap ops
op.value = binary.LittleEndian.Uint64(data[1:9])
// Verify checksum.
h := fnv.New32a()
_, _ = h.Write(data[0:9])
if op.typ > 1 {
switch op.typ {
case opTypeAdd, opTypeRemove:
// nothing to do, just being not-default
case opTypeAddBatch, opTypeRemoveBatch:
// This ensures that in doing 13+op.value*8, the max int won't be exceeded and a wrap around case
// (resulting in a negative value) won't occur in the slice indexing while writing
if op.value > maxBatchSize {
return fmt.Errorf("Maximum operation size exceeded")
return fmt.Errorf("maximum operation size exceeded")
}
if len(data) < int(13+op.value*8) {
return fmt.Errorf("op data truncated - expected %d, got %d", 13+op.value*8, len(data))
@ -4053,9 +4540,19 @@ func (op *op) UnmarshalBinary(data []byte) error {
op.values[i] = binary.LittleEndian.Uint64(data[start : start+8])
}
op.value = 0
case opTypeAddRoaring, opTypeRemoveRoaring:
if len(data) < int(13+4+op.value) {
return fmt.Errorf("op data truncated - expected %d, got %d", 13+op.value, len(data))
}
op.opN = int(binary.LittleEndian.Uint32(data[13:17]))
op.roaring = data[17 : 17+op.value]
_, _ = h.Write(data[13 : 17+op.value])
// op.value = 0
default:
return fmt.Errorf("unknown op type: %d", op.typ)
}
if chk := binary.LittleEndian.Uint32(data[9:13]); chk != h.Sum32() {
return fmt.Errorf("checksum mismatch: exp=%08x, got=%08x", h.Sum32(), chk)
return fmt.Errorf("checksum mismatch: type %d, exp=%08x, got=%08x", op.typ, h.Sum32(), chk)
}
return nil
@ -4063,10 +4560,38 @@ func (op *op) UnmarshalBinary(data []byte) error {
// size returns the encoded size of the op, in bytes.
func (op *op) size() int {
if op.typ == opTypeAdd || op.typ == opTypeRemove {
switch op.typ {
case opTypeAdd, opTypeRemove:
return 1 + 8 + 4
case opTypeAddBatch, opTypeRemoveBatch:
return 1 + 8 + 4 + len(op.values)*8
case opTypeAddRoaring, opTypeRemoveRoaring:
return 1 + 8 + 4 + 4 + len(op.roaring)
}
return 1 + 8 + 4 + len(op.values)*8
if roaringParanoia {
panic(fmt.Sprintf("op size() called on unknown op type %d", op.typ))
}
return 0
}
// size returns the size needed to encode the op, in bytes. for
// roaring ops, this does not include the roaring data, which is
// already encoded.
func (op *op) encodeSize() int {
switch op.typ {
case opTypeAdd, opTypeRemove:
return 1 + 8 + 4
case opTypeAddBatch, opTypeRemoveBatch:
return 1 + 8 + 4 + len(op.values)*8
case opTypeAddRoaring, opTypeRemoveRoaring:
return 1 + 8 + 4 + 4
}
if roaringParanoia {
panic(fmt.Sprintf("op encodeSize() called on unknown op type %d", op.typ))
}
return 0
}
// count returns the number of bits the operation mutates.
@ -4076,6 +4601,8 @@ func (op *op) count() int {
return 1
case 2, 3:
return len(op.values)
case 4, 5:
return op.opN
default:
panic(fmt.Sprintf("unknown operation type: %d", op.typ))
}
@ -4429,6 +4956,75 @@ func xorBitmapRun(a, b *Container) *Container {
return output
}
// CompareEquality is used mostly in test cases to confirm that two bitmaps came
// out the same. It does not expect corresponding opN, or OpWriter, but expects
// identical bit contents. It does not expect identical representations; a bitmap
// container can be identical to an array container. It returns a boolean value,
// and also an explanation for a false value.
func (b *Bitmap) BitwiseEqual(c *Bitmap) (bool, error) {
biter, _ := b.Containers.Iterator(0)
citer, _ := c.Containers.Iterator(0)
bn, cn := biter.Next(), citer.Next()
var bk, ck uint64
var bc, cc *Container
bct, cct := 0, 0
for bn && cn {
bk, bc = biter.Value()
ck, cc = citer.Value()
// zero containers are allowed to match no-container
if bk < ck {
if bc.N() == 0 {
bn = biter.Next()
continue
}
}
if ck < bk {
if cc.N() == 0 {
cn = citer.Next()
continue
}
}
bct++
cct++
if bk != ck {
return false, fmt.Errorf("differing keys [%d vs %d]", bk, ck)
}
diff := xor(bc, cc)
if diff.N() != 0 {
return false, fmt.Errorf("differing containers for key %d: %v vs %v", bk, bc, cc)
}
bn, cn = biter.Next(), citer.Next()
}
// only one can have containers left. they should all be empty. so we
// look at any remaining containers, break out of the loop if they're not
// empty, and otherwise keep iterating.
for bn {
bn = biter.Next()
bk, bc = biter.Value()
if bc.N() != 0 {
bct++
break
}
bn = biter.Next()
}
for cn {
cn = citer.Next()
ck, cc = biter.Value()
if cc.N() != 0 {
cct++
break
}
cn = biter.Next()
}
if bn {
return false, fmt.Errorf("container mismatch: %d vs %d containers, first bitmap has extra container %d [%d bits]", bct, cct, bk, bc)
}
if cn {
return false, fmt.Errorf("container mismatch: %d vs %d containers, second bitmap has extra container %d [%d bits]", bct, cct, ck, cc)
}
return true, nil
}
func bitmapsEqual(b, c *Bitmap) error { // nolint: deadcode
statsHit("bitmapsEqual")
if b.OpWriter != c.OpWriter {
@ -4482,11 +5078,11 @@ const (
serialCookie = 12347 // runs, arrays, and bitmaps
)
func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint, card int) byte, header, pos int, flags byte, haveRuns bool, err error) {
func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint, card int) byte, header, pos int, haveRuns bool, err error) {
statsHit("readOfficialHeader")
if len(buf) < 8 {
err = fmt.Errorf("buffer too small, expecting at least 8 bytes, was %d", len(buf))
return size, containerTyper, header, pos, flags, haveRuns, err
return size, containerTyper, header, pos, haveRuns, err
}
cf := func(index uint, card int) (newType byte) {
newType = containerBitmap
@ -4496,8 +5092,7 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
return newType
}
containerTyper = cf
cookie := binary.LittleEndian.Uint32(buf) & 0xFFFFFF
flags = buf[3]
cookie := binary.LittleEndian.Uint32(buf)
pos += 4
// cookie header
@ -4512,7 +5107,7 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
isRunBitmapSize := (int(size) + 7) / 8
if pos+isRunBitmapSize > len(buf) {
err = fmt.Errorf("malformed bitmap, is-run bitmap overruns buffer at %d", pos+isRunBitmapSize)
return size, containerTyper, header, pos, flags, haveRuns, err
return size, containerTyper, header, pos, haveRuns, err
}
isRunBitmap := buf[pos : pos+isRunBitmapSize]
@ -4525,22 +5120,22 @@ func readOfficialHeader(buf []byte) (size uint32, containerTyper func(index uint
}
} else {
err = fmt.Errorf("did not find expected serialCookie in header")
return size, containerTyper, header, pos, flags, haveRuns, err
return size, containerTyper, header, pos, haveRuns, err
}
header = pos
if size > (1 << 16) {
err = fmt.Errorf("it is logically impossible to have more than (1<<16) containers")
return size, containerTyper, header, pos, flags, haveRuns, err
return size, containerTyper, header, pos, haveRuns, err
}
// descriptive header
if pos+2*2*int(size) >= len(buf) {
err = fmt.Errorf("malformed bitmap, key-cardinality slice overruns buffer at %d", pos+2*2*int(size))
return size, containerTyper, header, pos, flags, haveRuns, err
return size, containerTyper, header, pos, haveRuns, err
}
pos += 2 * 2 * int(size) // moving pos past keycount
return size, containerTyper, header, pos, flags, haveRuns, err
return size, containerTyper, header, pos, haveRuns, err
}
// UnmarshalBinary decodes b from a binary-encoded byte slice. data can be in
@ -4557,13 +5152,15 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error {
return errors.Wrap(b.unmarshalPilosaRoaring(data), "unmarshaling as pilosa roaring")
}
keyN, containerTyper, header, pos, flags, haveRuns, err := readOfficialHeader(data)
keyN, containerTyper, header, pos, haveRuns, err := readOfficialHeader(data)
if err != nil {
return errors.Wrap(err, "reading roaring header")
}
b.Flags = flags
// Only the Pilosa roaring format has flags. The official Roaring format
// hasn't got space in its header for flags.
b.Flags = 0
b.Containers.Reset()
b.Containers.ResetN(int(keyN))
// Descriptive header section: Read container keys and cardinalities.
for i, buf := uint(0), data[header:]; i < uint(keyN); i, buf = i+1, buf[4:] {
card := int(binary.LittleEndian.Uint16(buf[2:4])) + 1

14
view.go
View file

@ -52,10 +52,11 @@ type view struct {
// Fragments by shard.
fragments map[uint64]*fragment
broadcaster broadcaster
stats stats.StatsClient
rowAttrStore AttrStore
logger logger.Logger
broadcaster broadcaster
stats stats.StatsClient
rowAttrStore AttrStore
logger logger.Logger
snapshotQueue chan *fragment
}
// newView returns a new instance of View.
@ -268,6 +269,7 @@ func (v *view) newFragment(path string, shard uint64) *fragment {
frag.CacheSize = v.cacheSize
frag.Logger = v.logger
frag.stats = v.stats
frag.snapshotQueue = v.snapshotQueue
if v.fieldType == FieldTypeMutex {
frag.mutexVector = newRowsVector(frag)
} else if v.fieldType == FieldTypeBool {
@ -441,11 +443,11 @@ func upgradeViewBSIv2(v *view, bitDepth uint) (ok bool, _ error) {
if tmpPath, err := upgradeRoaringBSIv2(frag, bitDepth); err != nil {
return ok, errors.Wrap(err, "upgrading bsi v2")
} else if err := frag.closeStorage(); err != nil {
} else if err := frag.closeStorage(true); err != nil {
return ok, errors.Wrap(err, "closing after bsi v2 upgrade")
} else if err := os.Rename(tmpPath, frag.path); err != nil {
return ok, errors.Wrap(err, "renaming after bsi v2 upgrade")
} else if err := frag.openStorage(); err != nil {
} else if err := frag.openStorage(true); err != nil {
return ok, errors.Wrap(err, "re-opening after bsi v2 upgrade")
}
}