scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-10-06 19:07:50 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge branch 'master' into master

Browse source
This commit is contained in:
asvetlik 2019-07-03 15:06:58 -05:00 • committed by GitHub
commit 42a1d851cb
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
20 changed files with 2386 additions and 332 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

32
cmd/server.go
View file

@ -46,22 +46,24 @@ on the configured port.`,
return errors.Wrap(err, "running server")
}
// Initialize tracing in the command since it is global.
var cfg jaegercfg.Configuration
cfg.ServiceName = "pilosa"
cfg.Sampler = &jaegercfg.SamplerConfig{
Type: Server.Config.Tracing.SamplerType,
Param: Server.Config.Tracing.SamplerParam,
if Server.Config.Tracing.SamplerType != "off" {
// Initialize tracing in the command since it is global.
var cfg jaegercfg.Configuration
cfg.ServiceName = "pilosa"
cfg.Sampler = &jaegercfg.SamplerConfig{
Type: Server.Config.Tracing.SamplerType,
Param: Server.Config.Tracing.SamplerParam,
}
cfg.Reporter = &jaegercfg.ReporterConfig{
LocalAgentHostPort: Server.Config.Tracing.AgentHostPort,
}
tracer, closer, err := cfg.NewTracer()
if err != nil {
return errors.Wrap(err, "initializing jaeger tracer")
}
defer closer.Close()
tracing.GlobalTracer = opentracing.NewTracer(tracer)
}
cfg.Reporter = &jaegercfg.ReporterConfig{
LocalAgentHostPort: Server.Config.Tracing.AgentHostPort,
}
tracer, closer, err := cfg.NewTracer()
if err != nil {
return errors.Wrap(err, "initializing jaeger tracer")
}
defer closer.Close()
tracing.GlobalTracer = opentracing.NewTracer(tracer)
return errors.Wrap(Server.Wait(), "waiting on Server")
},

2
ctl/server.go
View file

@ -77,7 +77,7 @@ func BuildServerFlags(cmd *cobra.Command, srv *server.Command) {
// Tracing
flags.StringVarP(&srv.Config.Tracing.AgentHostPort, "tracing.agent-host-port", "", srv.Config.Tracing.AgentHostPort, "Jaeger agent host:port.")
flags.StringVarP(&srv.Config.Tracing.SamplerType, "tracing.sampler-type", "", srv.Config.Tracing.SamplerType, "Jaeger sampler type.")
flags.StringVarP(&srv.Config.Tracing.SamplerType, "tracing.sampler-type", "", srv.Config.Tracing.SamplerType, "Jaeger sampler type or 'off' to disable tracing completely.")
flags.Float64VarP(&srv.Config.Tracing.SamplerParam, "tracing.sampler-param", "", srv.Config.Tracing.SamplerParam, "Jaeger sampler parameter.")
// Profiling

2
docs/configuration.md
View file

@ -377,7 +377,7 @@ The config file is in the [toml format](https://github.com/toml-lang/toml) and h
#### Tracing Sampler Type
* Description: Jaeger sampler type (const, probabilistic, ratelimiting, or remote)
* Description: Jaeger sampler type (const, probabilistic, ratelimiting, or remote). Set to 'off' to disable tracing completely.
* Flag: `tracing.sampler-type`
* Env: `PILOSA_TRACING_SAMPLER_TYPE`
* Config:

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",
},

297
roaring/fuzzer.go
View file

@ -16,6 +16,15 @@
package roaring
import (
"encoding/binary"
"fmt"
"io/ioutil"
"reflect"
)
// FuzzBitmapUnmarshalBinary fuzz tests the unmarshaling of binary
// to both Pilosa and official roaring formats.
func FuzzBitmapUnmarshalBinary(data []byte) int {
b := NewBitmap()
err := b.UnmarshalBinary(data)
@ -24,3 +33,291 @@ func FuzzBitmapUnmarshalBinary(data []byte) int {
}
return 1
}
// FuzzRoaringOps fuzz tests different operations on roaring bitmaps,
// comparing the results to a naive implementation of the operations
// on uint64 slices.
func FuzzRoaringOps(data []byte) int {
// number of uint64s not to include
const reserved = 4
// flipping is too inefficient for large values of end - start
// and will cause go-fuzz to hang if not controlled.
const maxFlips = 1000000
arr := bytesToUint64s(data)
if len(arr) <= reserved {
return 0
}
// start > end is possible. This will test correctness in unexpected conditions.
start, end, split, rand := arr[0], arr[1], int(arr[2]), arr[3]
// don't include start, end, split, rand in the slices
if split < reserved {
split = reserved
}
// ensure slice in bounds
if split > len(arr) {
split = len(arr)
}
// using removeSliceDuplicates guarantees that the slice inputs of the
// following functions do not have duplicates and are sorted, just as
// the Roaring Bitmap implementations are.
s1 := removeSliceDuplicates(arr[reserved:split])
s2 := removeSliceDuplicates(arr[split:])
if len(s1) == 0 {
s1 = nil
}
if len(s2) == 0 {
s2 = nil
}
bm1 := NewBitmap(arr[reserved:split]...)
bm2 := NewBitmap(arr[split:]...)
expected := s1
actual := bm1.Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("first slice:\n expected: %v\n got: %v", expected, actual))
}
expected = s2
actual = bm2.Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("second slice:\n expected: %v\n got: %v", expected, actual))
}
// Pure functions
expected = []uint64{maxInSlice(s1), maxInSlice(s2)}
actual = []uint64{bm1.Max(), bm2.Max()}
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("max values:\n expected: %v\n got: %v", expected, actual))
}
expected = intersectSlice(s1, s2)
actual = bm1.Intersect(bm2).Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("intersection:\n expected: %v\n got: %v", expected, actual))
}
expected = unionSlice(s1, s2)
actual = bm1.Union(bm2).Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("union:\n expected: %v\n got: %v", expected, actual))
}
expected = differenceSlice(s1, s2)
actual = bm1.Difference(bm2).Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("difference:\n expected: %v\n got: %v", expected, actual))
}
expected = xorSlice(s1, s2)
actual = bm1.Xor(bm2).Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("XOR:\n expected: %v\n got: %v", expected, actual))
}
if (len(s1) > 0) != bm1.Any() {
panic(fmt.Sprintf("any:\n %v has %v values but got %v which has %v values", s1, len(s1), bm1.Slice(), bm1.Any()))
}
if (len(s2) > 0) != bm2.Any() {
panic(fmt.Sprintf("any:\n %v has %v values but got %v which has %v values", s2, len(s2), bm2.Slice(), bm2.Any()))
}
expected = []uint64{uint64(len(s1)), uint64(len(s2))}
actual = []uint64{bm1.Count(), bm2.Count()}
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("count:\n expected: %v\n got: %v", expected, actual))
}
expect := countRangeSlice(s1, start, end)
got := bm1.CountRange(start, end)
if expect != got {
panic(fmt.Sprintf("count range:\n count from %v to %v in slice %v and bitmap %v:\n expected %v got %v",
start, end, s1, bm1.Slice(), expect, got))
}
expect = countRangeSlice(s2, start, end)
got = bm2.CountRange(start, end)
if expect != got {
panic(fmt.Sprintf("count range:\n count from %v to %v in slice %v and bitmap %v:\n expected %v got %v",
start, end, s2, bm2.Slice(), expect, got))
}
expected = rangeSlice(s1, start, end)
actual = bm1.SliceRange(start, end)
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("slice range:\n from %v to %v in slice %v and bitmap %v:\n expected %v\n got %v",
start, end, s1, bm1.Slice(), expected, actual))
}
expected = rangeSlice(s2, start, end)
actual = bm2.SliceRange(start, end)
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("slice range:\n from %v to %v in slice %v and bitmap %v:\n expected %v\n got %v",
start, end, s2, bm2.Slice(), expected, actual))
}
expect = uint64(len(intersectSlice(s1, s2)))
got = bm1.IntersectionCount(bm2)
if expect != got {
panic(fmt.Sprintf("intersection count:\n expected %v got %v", expect, got))
}
_, found := containedInSlice(s1, rand)
if found != bm1.Contains(rand) {
panic(fmt.Sprintf("contains:\n %v contains %v: %v\n %v contains %v: %v", s1, rand, found,
bm1.Slice(), rand, bm1.Contains(rand)))
}
_, found = containedInSlice(s2, rand)
if found != bm2.Contains(rand) {
panic(fmt.Sprintf("contains:\n %v contains %v: %v\n %v contains %v: %v", s2, rand, found,
bm2.Slice(), rand, bm2.Contains(rand)))
}
if end-start < maxFlips {
expected = flipSlice(s1, start, end)
actual = bm1.Flip(start, end).Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("flip:\n from %v to %v in slice %v and bitmap %v\n expected %v\n got %v",
start, end, s1, bm1.Slice(), expected, actual))
}
expected = flipSlice(s2, start, end)
actual = bm2.Flip(start, end).Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("flip:\n from %v to %v in slice %v and bitmap %v\n expected %v\n got %v",
start, end, s2, bm2.Slice(), expected, actual))
}
}
expected = make([]uint64, 0)
actual = make([]uint64, 0)
forEachInSlice(s1, func(v uint64) { expected = append(expected, v) })
bm1.ForEach(func(v uint64) { actual = append(actual, v) })
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("for each:\n expected %v\n got %v", expected, actual))
}
expected = make([]uint64, 0)
actual = make([]uint64, 0)
forEachInSlice(s2, func(v uint64) { expected = append(expected, v) })
bm2.ForEach(func(v uint64) { actual = append(actual, v) })
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("for each:\n expected %v\n got %v", expected, actual))
}
expected = make([]uint64, 0)
actual = make([]uint64, 0)
forEachInRangeSlice(s1, start, end, func(v uint64) { expected = append(expected, v) })
bm1.ForEachRange(start, end, func(v uint64) { actual = append(actual, v) })
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("for each in range:\n expected %v\n got %v", expected, actual))
}
expected = make([]uint64, 0)
actual = make([]uint64, 0)
forEachInRangeSlice(s2, start, end, func(v uint64) { expected = append(expected, v) })
bm2.ForEachRange(start, end, func(v uint64) { actual = append(actual, v) })
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("for each in range:\n expected %v\n got %v", expected, actual))
}
// Impure functions
// The following tests operations that mutate bitmaps.
nbm1, nbm2 := bm1.Clone(), bm2.Clone()
expected = shiftSlice(s1, 1)
tempBM, _ := nbm1.Shift(1)
actual = tempBM.Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("shift:\n in slice %v and bitmap %v \n expected %v\n got %v",
s1, bm1.Slice(), expected, actual))
}
expected = shiftSlice(s2, 1)
tempBM, _ = nbm2.Shift(1)
actual = tempBM.Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("shift:\n in slice %v and bitmap %v \n expected %v\n got %v",
s2, bm2.Slice(), expected, actual))
}
// reuse start and end as random values
rand2 := start
rand3 := end
nbm1 = bm1.Clone()
expected, echanged := addNToSlice(s1, rand)
achanged := nbm1.DirectAddN(rand)
actual = nbm1.Slice()
if echanged != achanged || !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("directAddN:\n adding %v in slice %v and bitmap %v \n expected %v and %v changed \n got %v and %v changed",
rand, s1, bm1.Slice(), expected, echanged, actual, achanged))
}
nbm2 = bm2.Clone()
expected, echanged = addNToSlice(s2, rand2, rand3)
achanged = nbm2.DirectAddN(rand2, rand3)
actual = nbm2.Slice()
if echanged != achanged || !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("directAddN:\n adding %v and %v in slice %v and bitmap %v \n expected %v and %v changed \n got %v and %v changed",
rand2, rand3, s2, bm2.Slice(), expected, echanged, actual, achanged))
}
nbm1 = bm1.Clone()
expected, echanged = removeNFromSlice(s1, rand2, rand3)
achanged = nbm1.DirectRemoveN(rand2, rand3)
actual = nbm1.Slice()
if echanged != achanged || !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("directRemoveN\n removing %v and %v in slice %v and bitmap %v \n expected %v and %v changed \n got %v and %v changed",
rand2, rand3, s1, bm1.Slice(), expected, echanged, actual, achanged))
}
nbm2 = bm2.Clone()
expected, echanged = removeNFromSlice(s2, rand)
achanged = nbm2.DirectRemoveN(rand)
actual = nbm2.Slice()
if echanged != achanged || !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("directRemoveN:\n removing %v in slice %v and bitmap %v \n expected %v and %v changed \n got %v and %v changed",
rand, s2, bm2.Slice(), expected, echanged, actual, achanged))
}
nbm1, nbm2 = bm1.Clone(), bm2.Clone()
expected = unionSlice(s1, s2)
nbm1.UnionInPlace(nbm2)
actual = nbm1.Slice()
if !reflect.DeepEqual(expected, actual) {
panic(fmt.Sprintf("union in place:\n expected %v\n got %v", expected, actual))
}
return 1
}
func bytesToUint64s(data []byte) []uint64 {
const uint64Size = 8
size := len(data) / uint64Size
slice := make([]uint64, 0)
for i := 0; i < size; i++ {
offset := i * uint64Size
num := binary.LittleEndian.Uint64(data[offset : offset+uint64Size])
slice = append(slice, num)
}
return slice
}
// copy and paste the following to a main file to run.
// path should be the absolute path to the corpus.
// make sure filename is not already in the corpus.
func addSliceToCorpus(slice []uint64, filename, path string) {
data := uint64sToBytes(slice)
err := ioutil.WriteFile(path+"/"+filename, data, 0777)
if err != nil {
fmt.Printf("could not write to file: %v\n", err)
}
}
func uint64sToBytes(slice []uint64) []byte {
const uint64Size = 8
size := len(slice) * uint64Size
data := make([]byte, size)
for i := 0; i < len(slice); i++ {
offset := i * uint64Size
binary.LittleEndian.PutUint64(data[offset:offset+uint64Size], slice[i])
}
return data
}

321
roaring/naive.go Normal file
View file

@ -0,0 +1,321 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package roaring
import (
"math"
"sort"
)
// The following functions reimplement Roaring Bitmap methods, but done naively on
// uint64 slices. Most of these functions are inefficient, which is acceptable because
// this is purely for testing consistency with Roaring internal operations. Thus, the
// functions should be easily guaranteed to produce the correct results.
func sortSlice(slice []uint64) {
sort.Slice(slice, func(i, j int) bool { return slice[i] < slice[j] })
}
// removeSliceDuplicates removes duplicate values
// in the slice and sorts the output.
func removeSliceDuplicates(slice []uint64) []uint64 {
// just throw slice into a map and
// get the values out again
hash := make(map[uint64]bool)
for _, val := range slice {
hash[val] = true
}
unique := make([]uint64, 0)
for key := range hash {
unique = append(unique, key)
}
if len(unique) == 0 {
return nil
}
sortSlice(unique)
return unique
}
// intersect intersects two []uint64s, removing any duplicates
// and sorting the final output.
func intersectSlice(s1, s2 []uint64) []uint64 {
// throw both slices in maps
hash1 := make(map[uint64]bool)
for _, val := range s1 {
hash1[val] = true
}
hash2 := make(map[uint64]bool)
for _, val := range s2 {
hash2[val] = true
}
intersection := make([]uint64, 0)
// look for keys from hash1 also in hash2
for key := range hash1 {
if _, found := hash2[key]; found {
intersection = append(intersection, key)
}
}
if len(intersection) == 0 {
return nil
}
sortSlice(intersection)
return intersection
}
// union unions two []uint64s and sorts the output.
func unionSlice(s1, s2 []uint64) []uint64 {
// just dump both slices in a map
// and get the values out again
hash := make(map[uint64]bool)
for _, val := range s1 {
hash[val] = true
}
for _, val := range s2 {
hash[val] = true
}
union := make([]uint64, 0)
for key := range hash {
union = append(union, key)
}
if len(union) == 0 {
return nil
}
sortSlice(union)
return union
}
// maxSlice returns the max in the slice.
func maxInSlice(slice []uint64) uint64 {
if len(slice) == 0 {
return 0
}
max := uint64(0)
for _, val := range slice {
if val > max {
max = val
}
}
return max
}
// differenceSlice returns a slice containing the values
// present in the first slice but not in the second.
func differenceSlice(s1, s2 []uint64) []uint64 {
// throw s2 in a map, check if each value
// in s1 is also in that map
hash := make(map[uint64]bool)
for _, val := range s2 {
hash[val] = true
}
diff := make([]uint64, 0)
for _, val := range s1 {
if _, found := hash[val]; !found {
diff = append(diff, val)
}
}
// make sure duplicates in s1 are not added
diff = removeSliceDuplicates(diff)
return diff
}
// xorSlice returns an array containing the values
// present in exactly one of the two slices.
func xorSlice(s1, s2 []uint64) []uint64 {
// throw both slices in maps
hash1 := make(map[uint64]bool)
for _, val := range s1 {
hash1[val] = true
}
hash2 := make(map[uint64]bool)
for _, val := range s2 {
hash2[val] = true
}
xor := make([]uint64, 0)
// add all values in hash1 not in hash2
for key := range hash1 {
if _, found := hash2[key]; !found {
xor = append(xor, key)
}
}
// add all values in hash2 not in hash1
for key := range hash2 {
if _, found := hash1[key]; !found {
xor = append(xor, key)
}
}
if len(xor) == 0 {
return nil
}
sortSlice(xor)
return xor
}
// shiftSlice adds n to each element and sorts the slice, but ignores any values that
// will cause an overflow. This does not modify the original slice, unlike the Roaring implementation.
func shiftSlice(slice []uint64, n int) []uint64 {
shifted := make([]uint64, 0)
for _, val := range slice {
if uint64(n) <= math.MaxUint64-val {
shifted = append(shifted, val+uint64(n))
}
}
if len(shifted) == 0 {
return nil
}
sortSlice(shifted)
return shifted
}
// forEachSlice executes fn for each element in the slice.
func forEachInSlice(slice []uint64, fn func(uint64)) {
for _, val := range slice {
fn(val)
}
}
// forEachRangeSlice executes fn for each element in slice that is in [start, end).
func forEachInRangeSlice(slice []uint64, start, end uint64, fn func(uint64)) {
for _, val := range slice {
if start <= val && val < end {
fn(val)
}
}
}
// containedInSlice returns the index of the first instance of v and true
// if v is in slice and returns -1 and false otherwise.
func containedInSlice(slice []uint64, v uint64) (int, bool) {
for idx := range slice {
if v == slice[idx] {
return idx, true
}
}
return -1, false
}
// addNToSlice adds the contents of a to slice and returns the new slice and
// number of values successfully added. This somewhat mimics *Bitmap.DirectAddN
// and but does not modify slice in place, so it returns that new slice instead.
func addNToSlice(slice []uint64, a ...uint64) ([]uint64, int) {
newSlice := make([]uint64, len(slice))
copy(newSlice, slice)
changed := 0
for _, val := range a {
if _, found := containedInSlice(newSlice, val); !found {
newSlice = append(newSlice, val)
changed++
}
}
if len(newSlice) == 0 {
return nil, changed
}
sortSlice(newSlice)
return newSlice, changed
}
// removeNFromSlice removes the contents of a from slice and returns the new slice and
// number of values successfully removed. This somewhat mimics *Bitmap.DirectRemoveN
// and but does not modify slice in place, so it returns that new slice instead.
func removeNFromSlice(slice []uint64, a ...uint64) ([]uint64, int) {
newSlice := make([]uint64, len(slice))
copy(newSlice, slice)
changed := 0
for _, val := range a {
if i, found := containedInSlice(newSlice, val); found {
newSlice = append(newSlice[:i], newSlice[i+1:]...)
changed++
}
}
if len(newSlice) == 0 {
return nil, changed
}
sortSlice(newSlice)
return newSlice, changed
}
// countRangeSlice returns the number of values in slice that are in [start, end).
func countRangeSlice(slice []uint64, start, end uint64) uint64 {
count := uint64(0)
for _, val := range slice {
if start <= val && val < end {
count++
}
}
return count
}
// rangeSlice returns a sorted slice of integers between [start, end).
func rangeSlice(slice []uint64, start, end uint64) []uint64 {
newSlice := make([]uint64, 0)
for _, val := range slice {
if start <= val && val < end {
newSlice = append(newSlice, val)
}
}
if len(newSlice) == 0 {
return nil
}
sortSlice(newSlice)
return newSlice
}
// flipSplice returns a slice containing all numbers in [start, end]
// that are not in the original slice, as well as the numbers in the
// original slice not in [start, end].
func flipSlice(slice []uint64, start, end uint64) []uint64 {
if start > end {
sortSlice(slice)
return slice
}
flipped := make([]uint64, 0)
// add values in slice outside [start, end]
hash := make(map[uint64]bool)
for _, val := range slice {
hash[val] = true
}
for val := range hash {
if val < start || val > end {
flipped = append(flipped, val)
}
}
for i := start; i <= end; i++ {
if _, found := containedInSlice(slice, i); !found {
flipped = append(flipped, i)
}
}
if len(flipped) == 0 {
return nil
}
sortSlice(flipped)
return flipped
}

459
roaring/naive_test.go Normal file
View file

@ -0,0 +1,459 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
package roaring
import (
"reflect"
"testing"
)
func TestSortSlice(t *testing.T) {
tests := []struct{ a, expected []uint64 }{
{
a: []uint64{1, 3, 2, 8, 5, 21, 13},
expected: []uint64{1, 2, 3, 5, 8, 13, 21},
},
}
for _, test := range tests {
sortSlice(test.a)
if !reflect.DeepEqual(test.a, test.expected) {
t.Fatalf("unexpected sorting: %v", test.a)
}
}
}
func TestRemoveSliceDuplicates(t *testing.T) {
tests := []struct{ a, expected []uint64 }{
{
a: []uint64{2, 3, 2, 1, 2, 5, 8, 5, 13, 3, 2, 5, 144},
expected: []uint64{1, 2, 3, 5, 8, 13, 144},
},
{
a: []uint64{2, 3, 2, 1, 2, 5, 8, 5, 13, 3, 2, 5, 144, 21, 8, 3, 3, 5, 5, 1, 34, 21, 21},
expected: []uint64{1, 2, 3, 5, 8, 13, 21, 34, 144},
},
}
for _, test := range tests {
got := removeSliceDuplicates(test.a)
if !reflect.DeepEqual(got, test.expected) {
t.Fatalf("expected %v, got %v", test.expected, got)
}
}
}
func TestIntersectSlice(t *testing.T) {
tests := []struct{ a, b, expected []uint64 }{
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{2, 1, 9, 5, 12},
expected: []uint64{1, 5, 9},
},
{
a: []uint64{2, 1, 9, 5, 12},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64{1, 5, 9},
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{1, 5, 9},
expected: []uint64{1, 5, 9},
},
}
for _, test := range tests {
got := intersectSlice(test.a, test.b)
if !reflect.DeepEqual(test.expected, got) {
t.Fatalf("expected %v, got %v", test.expected, got)
}
}
}
func TestUnionSlice(t *testing.T) {
tests := []struct{ a, b, expected []uint64 }{
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{2, 1, 9, 5, 12},
expected: []uint64{1, 2, 4, 5, 9, 12, 13, 24},
},
{
a: []uint64{2, 1, 9, 5, 12},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64{1, 2, 4, 5, 9, 12, 13, 24},
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{1, 5, 9},
expected: []uint64{1, 4, 5, 9, 13, 24},
},
}
for _, test := range tests {
got := unionSlice(test.a, test.b)
if !reflect.DeepEqual(test.expected, got) {
t.Fatalf("expected %v, got %v", test.expected, got)
}
}
}
func TestMaxInSlice(t *testing.T) {
a := []uint64{1, 4, 9, 5, 24, 13}
v := maxInSlice(a)
if uint64(24) != v {
t.Fatalf("expected %v, but got %v", uint64(24), v)
}
for i := uint64(1000); i <= uint64(100000); i++ {
a = append(a, i)
if v = maxInSlice(a); v != i {
t.Fatalf("expected %v, but got %v", i, v)
}
}
}
func TestDifferenceSlice(t *testing.T) {
tests := []struct{ a, b, expected []uint64 }{
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{2, 1, 9, 5, 12},
expected: []uint64{4, 13, 24},
},
{
a: []uint64{2, 1, 9, 5, 12},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64{2, 12},
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64(nil),
},
}
for _, test := range tests {
got := differenceSlice(test.a, test.b)
if !reflect.DeepEqual(test.expected, got) {
t.Fatalf("expected %v, got %v", test.expected, got)
}
}
}
func TestXorSlice(t *testing.T) {
tests := []struct{ a, b, expected []uint64 }{
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{2, 1, 9, 5, 12},
expected: []uint64{2, 4, 12, 13, 24},
},
{
a: []uint64{2, 1, 9, 5, 12},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64{2, 4, 12, 13, 24},
},
{
a: []uint64{2, 4, 12, 13, 24},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64{1, 2, 5, 9, 12},
},
{
a: []uint64{2, 4, 12, 13, 24},
b: []uint64{1, 2, 5, 9, 12},
expected: []uint64{1, 4, 5, 9, 13, 24},
},
}
for _, test := range tests {
got := xorSlice(test.a, test.b)
if !reflect.DeepEqual(test.expected, got) {
t.Fatalf("expected %v, got %v", test.expected, got)
}
}
}
func TestShiftSlice(t *testing.T) {
tests := []struct {
a []uint64
shift int
expected []uint64
}{
{
a: []uint64{1, 4, 9, 5, 24, 13},
shift: 12,
expected: []uint64{13, 16, 17, 21, 25, 36},
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
shift: 0,
expected: []uint64{1, 4, 5, 9, 13, 24},
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
shift: 1,
expected: []uint64{2, 5, 6, 10, 14, 25},
},
}
for _, test := range tests {
got := shiftSlice(test.a, test.shift)
if !reflect.DeepEqual(got, test.expected) {
t.Fatalf("expected %v, got %v", test.expected, got)
}
}
}
func TestForEachInSlice(t *testing.T) {
a := []uint64{1, 4, 9, 5, 24, 13}
c := make([]uint64, 0)
forEachInSlice(a, func(v uint64) {
c = append(c, v+1)
})
if !reflect.DeepEqual(c, []uint64{2, 5, 10, 6, 25, 14}) {
t.Fatalf("unexpected values: %v", c)
}
}
func TestForEachInRangeSlice(t *testing.T) {
a := []uint64{1, 4, 9, 5, 24, 13}
c := make([]uint64, 0)
forEachInRangeSlice(a, uint64(3), uint64(12), func(v uint64) {
c = append(c, v+1)
})
if !reflect.DeepEqual(c, []uint64{5, 10, 6}) {
t.Fatalf("unexpected values: %v", c)
}
}
func TestContainedInSlice(t *testing.T) {
tests := []struct {
a []uint64
c uint64
index int
found bool
}{
{
a: []uint64{1, 4, 9, 5, 24, 13},
c: uint64(4),
index: 1,
found: true,
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
c: uint64(12),
index: -1,
found: false,
},
}
for _, test := range tests {
idx, found := containedInSlice(test.a, test.c)
if found != test.found {
t.Fatalf("expected value of found: %v, got %v", test.found, found)
}
if idx != test.index {
t.Fatalf("expected index %v, got %v", test.index, idx)
}
}
}
func TestAddNToSlice(t *testing.T) {
tests := []struct {
a []uint64
b []uint64
expected []uint64
changed int
}{
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{2, 1, 9, 5, 12},
expected: []uint64{1, 2, 4, 5, 9, 12, 13, 24},
changed: 2,
},
{
a: []uint64{2, 1, 9, 5, 12},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64{1, 2, 4, 5, 9, 12, 13, 24},
changed: 3,
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64{1, 4, 5, 9, 13, 24},
changed: 0,
},
}
for _, test := range tests {
got, changed := addNToSlice(test.a, test.b...)
if !reflect.DeepEqual(test.expected, got) {
t.Fatalf("expected slices %v, got %v", test.expected, got)
}
if changed != test.changed {
t.Fatalf("expected changed %v, got %v", test.changed, changed)
}
}
}
func TestRemoveNFromSlice(t *testing.T) {
tests := []struct {
a []uint64
b []uint64
expected []uint64
changed int
}{
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{2, 1, 9, 5, 12},
expected: []uint64{4, 13, 24},
changed: 3,
},
{
a: []uint64{2, 1, 9, 5, 12},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64{2, 12},
changed: 3,
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
b: []uint64{1, 4, 9, 5, 24, 13},
expected: []uint64(nil),
changed: 6,
},
}
for _, test := range tests {
got, changed := removeNFromSlice(test.a, test.b...)
if !reflect.DeepEqual(test.expected, got) {
t.Fatalf("expected slices %v, got %v", test.expected, got)
}
if changed != test.changed {
t.Fatalf("expected changed %v, got %v", test.changed, changed)
}
}
}
func TestCountRangeSlice(t *testing.T) {
tests := []struct {
a []uint64
start uint64
end uint64
expected uint64
}{
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(3),
end: uint64(12),
expected: uint64(3),
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(0),
end: uint64(25),
expected: uint64(6),
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(12),
end: uint64(4),
expected: uint64(0),
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(4),
end: uint64(4),
expected: uint64(0),
},
}
for _, test := range tests {
got := countRangeSlice(test.a, test.start, test.end)
if got != test.expected {
t.Fatalf("expected %v, got %v", test.expected, got)
}
}
}
func TestRangeSlice(t *testing.T) {
tests := []struct {
a []uint64
start uint64
end uint64
expected []uint64
}{
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(3),
end: uint64(12),
expected: []uint64{4, 5, 9},
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(0),
end: uint64(25),
expected: []uint64{1, 4, 5, 9, 13, 24},
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(5),
end: uint64(5),
expected: []uint64(nil),
},
}
for _, test := range tests {
got := rangeSlice(test.a, test.start, test.end)
if !reflect.DeepEqual(got, test.expected) {
t.Fatalf("expected %v, got %v", test.expected, got)
}
}
}
func TestFlipSlice(t *testing.T) {
tests := []struct {
a []uint64
start uint64
end uint64
expected []uint64
}{
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(3),
end: uint64(12),
expected: []uint64{1, 3, 6, 7, 8, 10, 11, 12, 13, 24},
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(13),
end: uint64(12),
expected: []uint64{1, 4, 5, 9, 13, 24},
},
{
a: []uint64{1, 4, 9, 5, 24, 13},
start: uint64(9),
end: uint64(13),
expected: []uint64{1, 4, 5, 10, 11, 12, 24},
},
}
for _, test := range tests {
got := flipSlice(test.a, test.start, test.end)
if !reflect.DeepEqual(got, test.expected) {
t.Fatalf("expected %v, got %v", test.expected, got)
}
}
}

669
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.
@ -431,7 +434,7 @@ func (b *Bitmap) Size() int {
// CountRange returns the number of bits set between [start, end).
func (b *Bitmap) CountRange(start, end uint64) (n uint64) {
if roaringParanoia {
if roaringSentinel {
if start > end {
panic(fmt.Sprintf("counting in range but %v > %v", start, end))
}
@ -494,7 +497,7 @@ func (b *Bitmap) Slice() []uint64 {
// SliceRange returns a slice of integers between [start, end).
func (b *Bitmap) SliceRange(start, end uint64) []uint64 {
if roaringParanoia {
if roaringSentinel {
if start > end {
panic(fmt.Sprintf("getting slice in range but %v > %v", start, end))
}
@ -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()),
}
@ -1271,7 +1725,7 @@ func (b *Bitmap) Check() error {
// Flip performs a logical negate of the bits in the range [start,end].
func (b *Bitmap) Flip(start, end uint64) *Bitmap {
if roaringParanoia {
if roaringSentinel {
if start > end {
panic(fmt.Sprintf("flipping in range but %v > %v", start, end))
}
@ -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

102
roaring/roaring_internal_test.go
View file

@ -3356,38 +3356,82 @@ func runContainerFunc(f interface{}, c ...*Container) *Container {
return nil
}
func TestUnmarshalOfficialRoaring(t *testing.T) {
//generated serialize image from java(clojure) with arrays
rbContainerWithTwoArrays, _ := hex.DecodeString("3A300000020000000000020001000000180000001E0000000100020003000100")
bm := NewBitmap()
er := bm.UnmarshalBinary(rbContainerWithTwoArrays)
if er != nil {
t.Fatalf("UnmarshalOfficialRoaring %s", er)
func TestUnmarshalRoaringWithNoErrors(t *testing.T) {
testCases := []struct {
roaringData string
roaringFileName string
count uint64
expectedBits string
}{
{ // generated serialize image from java(clojure) with arrays
roaringData: "3A300000020000000000020001000000180000001E0000000100020003000100",
count: 4,
expectedBits: "[1 2 3 65537]",
},
{ // generated serialize image from java(clojure) with a run and array
roaringData: "3B3001000100000900010000000100010009000100",
count: 11,
expectedBits: "[1 2 3 4 5 6 7 8 9 10 65537]",
},
{ // had to use an external file because emacs was barfing on the long line :()
roaringFileName: "testdata/bitmapcontainer.roaringbitmap",
count: 10000,
expectedBits: "X",
},
}
if bm.Count() != 4 {
t.Fatalf("unexpected bitmap %v expected bits [1 2 3 65537]", bm.Slice())
}
//generated serialize image from java(clojure) with a run and array
rbContainerWithRLEandArray, _ := hex.DecodeString("3B3001000100000900010000000100010009000100")
bm = NewBitmap()
er = bm.UnmarshalBinary(rbContainerWithRLEandArray)
if er != nil {
t.Fatalf("UnmarshalOfficialRoaring %s", er)
}
if bm.Count() != 11 {
t.Fatalf("unexpected bitmap %v expected bits [1 2 3 4 5 6 7 8 9 10 65537]", bm.Slice())
}
//had to use an external file because emacs was barfing on the long line :()
_bitmap_array_container, _ := ioutil.ReadFile("testdata/bitmapcontainer.roaringbitmap")
bm = NewBitmap()
er = bm.UnmarshalBinary(_bitmap_array_container)
if er != nil {
t.Fatalf("UnmarshalOfficialRoaring %s", er)
}
if bm.Count() != 10000 {
t.Fatalf("expecting X got %d", bm.Count())
var testContainer []byte
var err error
for _, testCase := range testCases {
if testCase.roaringFileName == "" {
testContainer, err = hex.DecodeString(testCase.roaringData)
if err != nil {
t.Fatalf("hex decode %s", err)
}
} else {
testContainer, _ = ioutil.ReadFile(testCase.roaringFileName)
}
bm := NewBitmap()
err = bm.UnmarshalBinary(testContainer)
if err != nil {
t.Fatalf("UnmarshalOfficialRoaring %s", err)
}
if bm.Count() != testCase.count {
t.Fatalf("expecting %s got %d", testCase.expectedBits, bm.Count())
}
}
}
func TestUnmarshalRoaringWithErrors(t *testing.T) {
//testing bitmaps with no containers
noContainers := []struct {
hexString string
expectedError string
}{
{ // Runs a bitmap without runs and no containers through the official roaring
hexString: "3A30000000000000",
expectedError: "reading roaring header: malformed bitmap, key-cardinality slice overruns buffer at 8",
},
{ // Runs a bitmap with runs and no containers through the official roaring
hexString: "3B30000000000000",
expectedError: "reading roaring header: malformed bitmap, key-cardinality slice overruns buffer at 9",
},
{ // Runs a bitmap in the Pilosa format through the Pilosa roaring
hexString: "3C30000000000000",
},
}
for _, loopContainers := range noContainers {
zeroContainers, err := hex.DecodeString(loopContainers.hexString)
if err != nil {
t.Fatalf("hex decode %s", err)
}
bm := NewBitmap()
err = bm.UnmarshalBinary(zeroContainers)
if err != nil {
if err.Error() != loopContainers.expectedError {
t.Fatalf("Expected: %s, Got: %s", loopContainers.expectedError, err)
}
}
}
}
func BenchmarkUnionBitmapBitmapInPlace(b *testing.B) {

19
roaring/roaring_nop_sentinel.go Normal file
View file

@ -0,0 +1,19 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// +build !roaringsentinel
package roaring
const roaringSentinel = false

19
roaring/roaring_sentinel.go Normal file
View file

@ -0,0 +1,19 @@
// Copyright 2017 Pilosa Corp.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// +build roaringsentinel
package roaring
const roaringSentinel = true

18
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 {
@ -278,7 +280,9 @@ func (v *view) newFragment(path string, shard uint64) *fragment {
// deleteFragment removes the fragment from the view.
func (v *view) deleteFragment(shard uint64) error {
fragment := v.Fragment(shard)
v.mu.Lock()
defer v.mu.Unlock()
fragment := v.fragments[shard]
if fragment == nil {
return ErrFragmentNotFound
}
@ -439,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")
}
}