Merge pull request #368 from travisturner/performance-enhancements

Performance enhancements

attr.go

@@ -265,6 +265,8 @@ func txUpdateAttrs(tx *bolt.Tx, id uint64, m map[string]interface{}) (map[string
 			attr[k] = uint64(v)
 		case uint:
 			attr[k] = uint64(v)
+		case float64:
+			attr[k] = uint64(v)
 		case int64:
 			attr[k] = uint64(v)
 		case string, uint64, bool:

bitmap.go

@@ -174,6 +174,22 @@ func (b *Bitmap) InvalidateCount() {
 	}
 }
 
+//increment the bitmap cached counter, note this is an optimization that assumes that the caller is aware the size increased
+func (b *Bitmap) IncrementCount(i uint64) {
+	seg := b.segment(i / SliceWidth)
+	if seg != nil {
+		seg.n++
+	}
+}
+func (b *Bitmap) DecrementCount(i uint64) {
+	seg := b.segment(i / SliceWidth)
+	if seg != nil {
+		if seg.n > 0 {
+			seg.n--
+		}
+	}
+}
+
 // Count returns the number of set bits in the bitmap.
 func (b *Bitmap) Count() uint64 {
 	var n uint64
@@ -312,7 +328,6 @@ func (s *BitmapSegment) Difference(other *BitmapSegment) *BitmapSegment {
 // SetBit sets the i-th bit of the bitmap.
 func (s *BitmapSegment) SetBit(i uint64) (changed bool) {
 	s.ensureWritable()
-
 	changed, _ = s.data.Add(i)
 	if changed {
 		s.n++

cache.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"io"
 	"sort"
+	"sync"
 	"time"
 
 	"github.com/golang/groupcache/lru"
@@ -14,6 +15,7 @@ import (
 // Cache represents a cache for bitmap counts.
 type Cache interface {
 	Add(bitmapID uint64, n uint64)
+	BulkAdd(bitmapID uint64, n uint64)
 	Get(bitmapID uint64) uint64
 	Len() int
 
@@ -43,6 +45,10 @@ func NewLRUCache(maxEntries int) *LRUCache {
 	return c
 }
 
+func (c *LRUCache) BulkAdd(bitmapID, n uint64) {
+	c.Add(bitmapID, n)
+}
+
 // Add adds a bitmap to the cache.
 func (c *LRUCache) Add(bitmapID, n uint64) {
 	c.cache.Add(bitmapID, n)
@@ -92,6 +98,7 @@ var _ Cache = &LRUCache{}
 
 // RankCache represents a cache with sorted entries.
 type RankCache struct {
+	mu       sync.Mutex
 	entries  map[uint64]uint64
 	rankings []BitmapPair // cached, ordered list
 
@@ -112,33 +119,47 @@ func NewRankCache() *RankCache {
 
 // Add adds a bitmap to the cache.
 func (c *RankCache) Add(bitmapID uint64, n uint64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
 	// Ignore if the bit count on the bitmap is below the threshold.
 	if n < c.ThresholdValue {
 		return
 	}
 
-	// Add to cache.
 	c.entries[bitmapID] = n
 
-	// If size is larger than the threshold then trim it.
-	if len(c.entries) > c.ThresholdLength {
-		c.update()
-		for id, n := range c.entries {
-			if n <= c.ThresholdValue {
-				delete(c.entries, id)
-			}
-		}
+	c.invalidate()
+}
+
+// BulkAdd adds a bitmap to the cache unsorted. You should Invalidate after completion.
+func (c *RankCache) BulkAdd(bitmapID uint64, n uint64) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if n < c.ThresholdValue {
+		return
 	}
+
+	c.entries[bitmapID] = n
 }
 
 // Get returns a bitmap with a given id.
-func (c *RankCache) Get(bitmapID uint64) uint64 { return c.entries[bitmapID] }
+func (c *RankCache) Get(bitmapID uint64) uint64 {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.entries[bitmapID]
+}
 
 // Len returns the number of items in the cache.
-func (c *RankCache) Len() int { return len(c.entries) }
+func (c *RankCache) Len() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return len(c.entries)
+}
 
 // BitmapIDs returns a list of all bitmap IDs in the cache.
 func (c *RankCache) BitmapIDs() []uint64 {
+	c.mu.Lock()
+	defer c.mu.Unlock()
 	a := make([]uint64, 0, len(c.entries))
 	for id := range c.entries {
 		a = append(a, id)
@@ -147,22 +168,25 @@ func (c *RankCache) BitmapIDs() []uint64 {
 	return a
 }
 
-// Invalidate reorders the entries, if necessary.
-func (c *RankCache) Invalidate() {
-	// Update if there aren't many items or it hasn't been updated recently.
-	if len(c.rankings) < 50 || (c.updateN > 0 && time.Since(c.updateTime) > 5*time.Minute) {
-		c.update()
-	}
-}
-
 // update reorders the entries by rank.
-func (c *RankCache) update() {
+func (c *RankCache) Invalidate() {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	c.invalidate()
+
+}
+func (c *RankCache) invalidate() {
+	// Don't invalidate more than once every X seconds.
+	// TODO: consider making this configurable.
+	if time.Now().Sub(c.updateTime).Seconds() < 10 {
+		return
+	}
 	// Convert cache to a sorted list.
 	rankings := make([]BitmapPair, 0, len(c.entries))
-	for id, n := range c.entries {
+	for id, cnt := range c.entries {
 		rankings = append(rankings, BitmapPair{
 			ID:    id,
-			Count: n,
+			Count: cnt,
 		})
 	}
 	sort.Sort(BitmapPairs(rankings))
@@ -177,6 +201,15 @@ func (c *RankCache) update() {
 
 	// Reset counters.
 	c.updateTime, c.updateN = time.Now(), 0
+
+	// If size is larger than the threshold then trim it.
+	if len(c.entries) > c.ThresholdLength {
+		for id, cnt := range c.entries {
+			if cnt <= c.ThresholdValue {
+				delete(c.entries, id)
+			}
+		}
+	}
 }
 
 // Top returns an ordered list of bitmaps.
@@ -233,6 +266,26 @@ func (p Pairs) Swap(i, j int)      { p[i], p[j] = p[j], p[i] }
 func (p Pairs) Len() int           { return len(p) }
 func (p Pairs) Less(i, j int) bool { return p[i].Count > p[j].Count }
 
+type PairHeap struct {
+	Pairs
+}
+
+func (p PairHeap) Less(i, j int) bool { return p.Pairs[i].Count < p.Pairs[j].Count }
+
+func (h *Pairs) Push(x interface{}) {
+	// Push and Pop use pointer receivers because they modify the slice's length,
+	// not just its contents.
+	*h = append(*h, x.(Pair))
+}
+
+func (h *Pairs) Pop() interface{} {
+	old := *h
+	n := len(old)
+	x := old[n-1]
+	*h = old[0 : n-1]
+	return x
+}
+
 // Add merges other into p and returns a new slice.
 func (p Pairs) Add(other []Pair) []Pair {
 	// Create lookup of key/counts.
@@ -327,3 +380,27 @@ func (p uint64Slice) merge(other []uint64) []uint64 {
 
 	return ret
 }
+
+// BitmapCache provides an interface for caching full bitmaps.
+type BitmapCache interface {
+	Fetch(id uint64) (*Bitmap, bool)
+	Add(id uint64, b *Bitmap)
+}
+
+// SimpleCache implements BitmapCache
+// it is meant to be a short-lived cache for cases where writes are continuing to access
+// the same bit within a short time frame (i.e. good for write-heavy loads)
+// A read-heavy use case would cause the cache to get bigger, potentially causing the
+// node to run out of memory.
+type SimpleCache struct {
+	cache map[uint64]*Bitmap
+}
+
+func (s *SimpleCache) Fetch(id uint64) (*Bitmap, bool) {
+	m, ok := s.cache[id]
+	return m, ok
+}
+
+func (s *SimpleCache) Add(id uint64, b *Bitmap) {
+	s.cache[id] = b
+}

executor.go

@@ -52,13 +52,15 @@ func (e *Executor) Execute(ctx context.Context, db string, q *pql.Query, slices
 
 	// If slices aren't specified, then include all of them.
 	if len(slices) == 0 {
-		// Round up the number of slices.
-		maxSlice := e.Index.DB(db).MaxSlice()
+		if needsSlices(q.Calls) {
+			// Round up the number of slices.
+			maxSlice := e.Index.DB(db).MaxSlice()
 
-		// Generate a slices of all slices.
-		slices = make([]uint64, maxSlice+1)
-		for i := range slices {
-			slices[i] = uint64(i)
+			// Generate a slices of all slices.
+			slices = make([]uint64, maxSlice+1)
+			for i := range slices {
+				slices[i] = uint64(i)
+			}
 		}
 	}
 
@@ -168,6 +170,10 @@ func (e *Executor) executeBitmapCallSlice(ctx context.Context, db string, c *pql
 // requeries to retrieve the full counts for each of the top results.
 func (e *Executor) executeTopN(ctx context.Context, db string, c *pql.Call, slices []uint64, opt *ExecOptions) ([]Pair, error) {
 	bitmapIDs, _ := c.Args["ids"].([]uint64)
+	var n uint64
+	if nval, ok := c.Args["n"]; ok {
+		n = nval.(uint64)
+	}
 
 	// Execute original query.
 	pairs, err := e.executeTopNSlices(ctx, db, c, slices, opt)
@@ -180,21 +186,29 @@ func (e *Executor) executeTopN(ctx context.Context, db string, c *pql.Call, slic
 	if len(pairs) == 0 || len(bitmapIDs) > 0 || opt.Remote {
 		return pairs, nil
 	}
-
 	// Only the original caller should refetch the full counts.
 	other := c.Clone()
-	other.Args["n"] = 0
+
+	// Double the size of n for other calls in order to...
+	// TODO: travis review
+	other.Args["n"] = len(bitmapIDs) * 2
 
 	ids := Pairs(pairs).Keys()
 	sort.Sort(uint64Slice(ids))
 	other.Args["ids"] = ids
 
-	return e.executeTopNSlices(ctx, db, other, slices, opt)
+	trimmedList, err := e.executeTopNSlices(ctx, db, other, slices, opt)
+	if err != nil {
+		return nil, err
+	}
+
+	if n != 0 && int(n) < len(trimmedList) {
+		trimmedList = trimmedList[0:n]
+	}
+	return trimmedList, nil
 }
 
 func (e *Executor) executeTopNSlices(ctx context.Context, db string, c *pql.Call, slices []uint64, opt *ExecOptions) ([]Pair, error) {
-	n, _ := c.Args["n"].(uint64)
-
 	// Execute calls in bulk on each remote node and merge.
 	mapFn := func(slice uint64) (interface{}, error) {
 		return e.executeTopNSlice(ctx, db, c, slice)
@@ -215,11 +229,6 @@ func (e *Executor) executeTopNSlices(ctx context.Context, db string, c *pql.Call
 	// Sort final merged results.
 	sort.Sort(Pairs(results))
 
-	// Only keep the top n after sorting.
-	if n > 0 && len(results) > int(n) {
-		results = results[0:n]
-	}
-
 	return results, nil
 }
 
@@ -954,6 +963,7 @@ func (e *Executor) mapper(ctx context.Context, ch chan mapResponse, nodes []*Nod
 			if n.Host == e.Host {
 				resp.result, resp.err = e.mapperLocal(ctx, nodeSlices, mapFn, reduceFn)
 			} else if !opt.Remote {
+
 				results, err := e.exec(ctx, n, db, &pql.Query{Calls: []*pql.Call{c}}, nodeSlices, opt)
 				if len(results) > 0 {
 					resp.result = results[0]
@@ -1052,3 +1062,21 @@ func hasOnlySetBitmapAttrs(calls []*pql.Call) bool {
 	}
 	return true
 }
+
+func needsSlices(calls []*pql.Call) bool {
+	if len(calls) == 0 {
+		return false
+	}
+	for _, call := range calls {
+		switch call.Name {
+		case "ClearBit", "Profile", "SetBit", "SetBitmapAttrs", "SetProfileAttrs":
+			continue
+		case "Count", "TopN":
+			return true
+		// default catches Bitmap calls
+		default:
+			return true
+		}
+	}
+	return false
+}

executor_test.go

@@ -260,6 +260,40 @@ func TestExecutor_Execute_TopN_fill(t *testing.T) {
 	}
 }
 
+// Ensure
+func TestExecutor_Execute_TopN_fill_small(t *testing.T) {
+	idx := MustOpenIndex()
+	defer idx.Close()
+
+	idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0)
+	idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(0, SliceWidth)
+	idx.MustCreateFragmentIfNotExists("d", "f", 2).SetBit(0, 2*SliceWidth)
+	idx.MustCreateFragmentIfNotExists("d", "f", 3).SetBit(0, 3*SliceWidth)
+	idx.MustCreateFragmentIfNotExists("d", "f", 4).SetBit(0, 4*SliceWidth)
+
+	idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(1, 0)
+	idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(1, 1)
+
+	idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(2, SliceWidth)
+	idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(2, SliceWidth+1)
+
+	idx.MustCreateFragmentIfNotExists("d", "f", 2).SetBit(3, 2*SliceWidth)
+	idx.MustCreateFragmentIfNotExists("d", "f", 2).SetBit(3, 2*SliceWidth+1)
+
+	idx.MustCreateFragmentIfNotExists("d", "f", 3).SetBit(4, 3*SliceWidth)
+	idx.MustCreateFragmentIfNotExists("d", "f", 3).SetBit(4, 3*SliceWidth+1)
+
+	// Execute query.
+	e := NewExecutor(idx.Index, NewCluster(1))
+	if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame=f, n=1)`), nil, nil); err != nil {
+		t.Fatal(err)
+	} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
+		{Key: 0, Count: 5},
+	}}) {
+		t.Fatalf("unexpected result: %s", spew.Sdump(result))
+	}
+}
+
 // Ensure a TopN() query with a source bitmap can be executed.
 func TestExecutor_Execute_TopN_Src(t *testing.T) {
 	idx := MustOpenIndex()
@@ -293,6 +327,52 @@ func TestExecutor_Execute_TopN_Src(t *testing.T) {
 	}
 }
 
+//Ensure TopN handles Attribute filters
+func TestExecutor_Execute_TopN_Attr(t *testing.T) {
+	//
+	idx := MustOpenIndex()
+	defer idx.Close()
+	idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0)
+	idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1)
+	idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth)
+
+	if err := idx.Frame("d", "f").BitmapAttrStore().SetAttrs(10, map[string]interface{}{"category": uint64(123)}); err != nil {
+		t.Fatal(err)
+	}
+	e := NewExecutor(idx.Index, NewCluster(1))
+	if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(frame="f", n=1, field="category", filters=[123])`), nil, nil); err != nil {
+		t.Fatal(err)
+	} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
+		{Key: 10, Count: 1},
+	}}) {
+		t.Fatalf("unexpected result: %s", spew.Sdump(result))
+	}
+
+}
+
+//Ensure TopN handles Attribute filters with source bitmap
+func TestExecutor_Execute_TopN_Attr_Src(t *testing.T) {
+	//
+	idx := MustOpenIndex()
+	defer idx.Close()
+	idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 0)
+	idx.MustCreateFragmentIfNotExists("d", "f", 0).SetBit(0, 1)
+	idx.MustCreateFragmentIfNotExists("d", "f", 1).SetBit(10, SliceWidth)
+
+	if err := idx.Frame("d", "f").BitmapAttrStore().SetAttrs(10, map[string]interface{}{"category": uint64(123)}); err != nil {
+		t.Fatal(err)
+	}
+	e := NewExecutor(idx.Index, NewCluster(1))
+	if result, err := e.Execute(context.Background(), "d", MustParse(`TopN(Bitmap(id=10,frame=f),frame="f", n=1, field="category", filters=[123])`), nil, nil); err != nil {
+		t.Fatal(err)
+	} else if !reflect.DeepEqual(result, []interface{}{[]pilosa.Pair{
+		{Key: 10, Count: 1},
+	}}) {
+		t.Fatalf("unexpected result: %s", spew.Sdump(result))
+	}
+
+}
+
 // Ensure a range query can be executed.
 func TestExecutor_Execute_Range(t *testing.T) {
 	idx := MustOpenIndex()

fragment.go

@@ -4,6 +4,7 @@ import (
 	"archive/tar"
 	"bufio"
 	"bytes"
+	"container/heap"
 	"context"
 	"crypto/sha1"
 	"encoding/binary"
@@ -49,7 +50,7 @@ const (
 
 const (
 	// DefaultFragmentMaxOpN is the default value for Fragment.MaxOpN.
-	DefaultFragmentMaxOpN = 1000
+	DefaultFragmentMaxOpN = 2000
 )
 
 // Fragment represents the intersection of a frame and slice in a database.
@@ -68,9 +69,12 @@ type Fragment struct {
 	storageData []byte
 	opN         int // number of ops since snapshot
 
-	// Bitmap cache.
+	// Cache for bitmap counts.
 	cache Cache
 
+	// Cache containing full bitmaps (not just counts).
+	bitmapCache BitmapCache
+
 	// Cached checksums for each block.
 	checksums map[int][]byte
 
@@ -203,6 +207,7 @@ func (f *Fragment) openStorage() error {
 
 	// Attach the file to the bitmap to act as a write-ahead log.
 	f.storage.OpWriter = f.file
+	f.bitmapCache = &SimpleCache{make(map[uint64]*Bitmap)}
 
 	return nil
 
@@ -239,9 +244,11 @@ func (f *Fragment) openCache() error {
 	// Read in all bitmaps by ID.
 	// This will cause them to be added to the cache.
 	for _, bitmapID := range pb.BitmapIDs {
-		n := f.storage.CountRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth)
-		f.cache.Add(bitmapID, n)
+		//n := f.storage.CountRange(bitmapID*SliceWidth, (bitmapID+1)*SliceWidth)
+		n := f.bitmap(bitmapID, true, true).Count()
+		f.cache.BulkAdd(bitmapID, n)
 	}
+	f.cache.Invalidate()
 
 	return nil
 }
@@ -305,26 +312,37 @@ func (f *Fragment) logger() *log.Logger { return log.New(f.LogOutput, "", log.Ls
 func (f *Fragment) Bitmap(bitmapID uint64) *Bitmap {
 	f.mu.Lock()
 	defer f.mu.Unlock()
-	return f.bitmap(bitmapID)
+	return f.bitmap(bitmapID, true, true)
 }
 
-func (f *Fragment) bitmap(bitmapID uint64) *Bitmap {
+func (f *Fragment) bitmap(bitmapID uint64, checkBitmapCache bool, updateBitmapCache bool) *Bitmap {
+
+	if checkBitmapCache {
+		r, ok := f.bitmapCache.Fetch(bitmapID)
+		if ok && r != nil {
+			return r
+		}
+	}
+
 	// Only use a subset of the containers.
 	// NOTE: The start & end ranges must be divisible by
 	data := f.storage.OffsetRange(f.slice*SliceWidth, bitmapID*SliceWidth, (bitmapID+1)*SliceWidth)
 
 	// Reference bitmap subrange in storage.
+	// We Clone() data because otherwise bm will contains pointers to containers in storage.
+	// This causes unexpected results when we cache the bitmap and try to use it later.
 	bm := &Bitmap{
 		segments: []BitmapSegment{{
-			data:     *data,
+			data:     *data.Clone(),
 			slice:    f.slice,
 			writable: false,
 		}},
 	}
 	bm.InvalidateCount()
 
-	// Update cache.
-	f.cache.Add(bitmapID, bm.Count())
+	if updateBitmapCache {
+		f.bitmapCache.Add(bitmapID, bm)
+	}
 
 	return bm
 }
@@ -337,31 +355,39 @@ func (f *Fragment) SetBit(bitmapID, profileID uint64) (changed bool, err error)
 	return f.setBit(bitmapID, profileID)
 }
 
-func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, bool error) {
-	// Determine the position of the bit in the storage.
+func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, err error) {
 	changed = false
+	// Determine the position of the bit in the storage.
 	pos, err := f.pos(bitmapID, profileID)
 	if err != nil {
 		return false, err
 	}
 
 	// Write to storage.
+
 	if changed, err = f.storage.Add(pos); err != nil {
 		return false, err
 	}
 
+	// Don't update the cache if nothing changed.
+	if !changed {
+		return changed, nil
+	}
+
 	// Invalidate block checksum.
 	delete(f.checksums, int(bitmapID/HashBlockSize))
 
-	// If the number of operations exceeds the limit then snapshot.
+	// Increment number of operations until snapshot is required.
 	if err := f.incrementOpN(); err != nil {
 		return false, err
 	}
 
+	// Get the bitmap from bitmapCache or fragment.storage.
+	bm := f.bitmap(bitmapID, true, true)
+	bm.SetBit(profileID)
+
 	// Update the cache.
-	if f.bitmap(bitmapID).SetBit(profileID) {
-		changed = true
-	}
+	f.cache.Add(bitmapID, bm.Count())
 
 	f.stats.Count("setN", 1)
 
@@ -376,7 +402,8 @@ func (f *Fragment) ClearBit(bitmapID, profileID uint64) (bool, error) {
 	return f.clearBit(bitmapID, profileID)
 }
 
-func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) {
+func (f *Fragment) clearBit(bitmapID, profileID uint64) (changed bool, err error) {
+	changed = false
 	// Determine the position of the bit in the storage.
 	pos, err := f.pos(bitmapID, profileID)
 	if err != nil {
@@ -384,11 +411,15 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) {
 	}
 
 	// Write to storage.
-	changed, err := f.storage.Remove(pos)
-	if err != nil {
+	if changed, err = f.storage.Remove(pos); err != nil {
 		return false, err
 	}
 
+	// Don't update the cache if nothing changed.
+	if !changed {
+		return changed, nil
+	}
+
 	// Invalidate block checksum.
 	delete(f.checksums, int(bitmapID/HashBlockSize))
 
@@ -397,10 +428,12 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) {
 		return false, err
 	}
 
+	// Get the bitmap from bitmapCache or fragment.storage.
+	bm := f.bitmap(bitmapID, true, true)
+	bm.ClearBit(profileID)
+
 	// Update the cache.
-	if f.bitmap(bitmapID).ClearBit(profileID) {
-		return true, nil
-	}
+	f.cache.Add(bitmapID, bm.Count())
 
 	f.stats.Count("clearN", 1)
 
@@ -453,7 +486,8 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
 	}
 
 	// Iterate over rankings and add to results until we have enough.
-	results := make([]Pair, 0, opt.N)
+	//results := make(PairHeap, 0, opt.N)
+	results := &PairHeap{}
 	for _, pair := range pairs {
 		bitmapID, n := pair.ID, pair.Count
 
@@ -477,7 +511,7 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
 		}
 
 		// The initial n pairs should simply be added to the results.
-		if opt.N == 0 || len(results) < opt.N {
+		if opt.N == 0 || results.Len() < opt.N {
 			// Calculate count and append.
 			count := n
 			if opt.Src != nil {
@@ -486,26 +520,22 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
 			if count == 0 {
 				continue
 			}
-			results = append(results, Pair{Key: bitmapID, Count: count})
+			heap.Push(results, Pair{Key: bitmapID, Count: count})
 
 			// If we reach the requested number of pairs and we are not computing
 			// intersections then simply exit. If we are intersecting then sort
 			// and then only keep pairs that are higher than the lowest count.
-			if opt.N > 0 && len(results) == opt.N {
+			if opt.N > 0 && results.Len() == opt.N {
 				if opt.Src == nil {
 					break
 				}
-				sort.Sort(Pairs(results))
 			}
 			continue
 		}
 
 		// Retrieve the lowest count we have.
 		// If it's too low then don't try finding anymore pairs.
-		threshold := results[len(results)-1].Count
-		if threshold < MinThreshold {
-			break
-		}
+		threshold := results.Pairs[0].Count
 
 		// If the bitmap doesn't have enough bits set before the intersection
 		// then we can assume that any remaing bitmaps also have a count too low.
@@ -515,22 +545,22 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
 
 		// Calculate the intersecting bit count and skip if it's below our
 		// last bitmap in our current result set.
+
 		count := opt.Src.IntersectionCount(f.Bitmap(bitmapID))
 		if count < threshold {
 			continue
 		}
 
-		// Swap out the last pair for this new count.
-		results[len(results)-1] = Pair{Key: bitmapID, Count: count}
-
-		// If it's count is also higher than the second to last item then resort.
-		if len(results) >= 2 && count > results[len(results)-2].Count {
-			sort.Sort(Pairs(results))
-		}
+		heap.Push(results, Pair{Key: bitmapID, Count: count})
 	}
-
-	sort.Sort(Pairs(results))
-	return results, nil
+	r := make(Pairs, results.Len(), results.Len())
+	x := results.Len()
+	i := 1
+	for results.Len() > 0 {
+		r[x-i] = heap.Pop(results).(Pair)
+		i++
+	}
+	return r, nil
 }
 
 func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair {
@@ -543,23 +573,27 @@ func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair {
 	}
 
 	// Otherwise retrieve specific bitmaps.
-	pairs := make([]BitmapPair, len(bitmapIDs))
-	for i, bitmapID := range bitmapIDs {
+	pairs := make([]BitmapPair, 0, len(bitmapIDs))
+	for _, bitmapID := range bitmapIDs {
 		// Look up cache first, if available.
 		if n := f.cache.Get(bitmapID); n > 0 {
-			pairs[i] = BitmapPair{
+			pairs = append(pairs, BitmapPair{
 				ID:    bitmapID,
 				Count: n,
-			}
+			})
 			continue
 		}
 
-		// Otherwise load from storage.
-		pairs[i] = BitmapPair{
-			ID:    bitmapID,
-			Count: f.Bitmap(bitmapID).Count(),
+		bm := f.Bitmap(bitmapID)
+		if bm.Count() > 0 {
+			// Otherwise load from storage.
+			pairs = append(pairs, BitmapPair{
+				ID:    bitmapID,
+				Count: bm.Count(),
+			})
 		}
 	}
+	sort.Sort(BitmapPairs(pairs))
 	return pairs
 }
 
@@ -838,7 +872,6 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 	// Process every bit.
 	// If an error occurs then reopen the storage.
 	lastID := uint64(0)
-	bmCounter := 0
 	if err := func() error {
 		set := make(map[uint64]struct{})
 		for i := range bitmapIDs {
@@ -851,7 +884,7 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 			}
 
 			// Write to storage.
-			changed, err := f.storage.Add(pos)
+			_, err = f.storage.Add(pos)
 			if err != nil {
 				return err
 			}
@@ -863,9 +896,6 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 				lastID = bitmapID
 				set[bitmapID] = struct{}{}
 			}
-			if changed {
-				bmCounter += 1
-			}
 
 			// Invalidate block checksum.
 			delete(f.checksums, int(bitmapID/HashBlockSize))
@@ -873,7 +903,10 @@ func (f *Fragment) Import(bitmapIDs, profileIDs []uint64) error {
 
 		// Update cache counts for all bitmaps.
 		for bitmapID := range set {
-			f.cache.Add(bitmapID, f.bitmap(bitmapID).Count())
+			// Import should ALWAYS have bitmap() load a new bm from fragment.storage
+			// because the bitmap that's in bitmapCache hasn't been updated with
+			// this import's data.
+			f.cache.BulkAdd(bitmapID, f.bitmap(bitmapID, false, false).Count())
 		}
 
 		f.cache.Invalidate()
@@ -912,10 +945,15 @@ func (f *Fragment) Snapshot() error {
 	defer f.mu.Unlock()
 	return f.snapshot()
 }
+func track(start time.Time, name string, logger *log.Logger) {
+	elapsed := time.Since(start)
+	logger.Printf("%s took %s", name, elapsed)
+}
 
 func (f *Fragment) snapshot() error {
 	logger := f.logger()
 	logger.Printf("fragment: snapshotting %s/%s/%d", f.db, f.frame, f.slice)
+	defer track(time.Now(), fmt.Sprintf("fragment: snapshot complete %s/%s/%d", f.db, f.frame, f.slice), logger)
 
 	// Create a temporary file to snapshot to.
 	snapshotPath := f.path + SnapshotExt
@@ -1216,7 +1254,6 @@ func (s *FragmentSyncer) SyncFragment() error {
 	// Determine replica set.
 	nodes := s.Cluster.FragmentNodes(s.Fragment.DB(), s.Fragment.Slice())
 	if len(nodes) == 1 {
-		//fmt.Println("no place to replicate", s.Fragment.DB(), s.Fragment.Frame(), s.Fragment.Slice())
 		return nil
 	}
 

handler.go

@@ -205,7 +205,10 @@ func (h *Handler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		http.NotFound(w, r)
 	}
 
-	h.logger().Printf("%s %s %.03fs", r.Method, r.URL.String(), time.Since(t).Seconds())
+	dif := time.Since(t).Seconds()
+	if dif > 90 {
+		h.logger().Printf("%s %s %.03fs", r.Method, r.URL.String(), dif)
+	}
 }
 
 // handleGetSchema handles GET /schema requests.

roaring/roaring.go

@@ -444,33 +444,56 @@ func (b *Bitmap) removeEmptyContainers() {
 		i++
 	}
 }
+func (b *Bitmap) countEmptyContainers() int {
+	result := 0
+	for i := 0; i < len(b.containers); {
+		c := b.containers[i]
+
+		if c.n == 0 {
+			result++
+		}
+		i++
+	}
+	return result
+}
 
 // WriteTo writes b to w.
 func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) {
 	// Remove empty containers before persisting.
-	b.removeEmptyContainers()
+	//b.removeEmptyContainers()
+	containerCount := len(b.keys) - b.countEmptyContainers()
 
 	// Build header before writing individual container blocks.
-	buf := make([]byte, headerSize+(len(b.keys)*(4+8+4)))
+	buf := make([]byte, headerSize+(containerCount*(4+8+4)))
 	binary.LittleEndian.PutUint32(buf[0:], cookie)
-	binary.LittleEndian.PutUint32(buf[4:], uint32(len(b.keys)))
-
+	binary.LittleEndian.PutUint32(buf[4:], uint32(containerCount))
+	empty := 0
 	// Encode keys and cardinality.
 	for i, key := range b.keys {
 		c := b.containers[i]
 
 		// Verify container count before writing.
-		count := c.count()
-		assert(c.count() == c.n, "cannot write container count, mismatch: count=%d, n=%d", count, c.n)
-
-		binary.LittleEndian.PutUint64(buf[headerSize+i*12:], uint64(key))
-		binary.LittleEndian.PutUint32(buf[headerSize+i*12+8:], uint32(c.n-1))
+		// TODO: instead of commenting this out, we need to make it a configuration option
+		//count := c.count()
+		//assert(c.count() == c.n, "cannot write container count, mismatch: count=%d, n=%d", count, c.n)
+		if c.n > 0 {
+			binary.LittleEndian.PutUint64(buf[headerSize+(i-empty)*12:], uint64(key))
+			binary.LittleEndian.PutUint32(buf[headerSize+(i-empty)*12+8:], uint32(c.n-1))
+		} else {
+			empty++
+		}
 	}
 
 	// Write the offset for each container block.
 	offset := uint32(len(buf))
+	empty = 0
 	for i, c := range b.containers {
-		binary.LittleEndian.PutUint32(buf[headerSize+(len(b.keys)*12)+(i*4):], uint32(offset))
+
+		if c.n > 0 {
+			binary.LittleEndian.PutUint32(buf[headerSize+(containerCount*12)+((i-empty)*4):], uint32(offset))
+		} else {
+			empty++
+		}
 		offset += uint32(c.size())
 	}
 
@@ -483,10 +506,12 @@ func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) {
 
 	// Write each container block.
 	for _, c := range b.containers {
-		nn, err := c.WriteTo(w)
-		n += nn
-		if err != nil {
-			return n, err
+		if c.n > 0 {
+			nn, err := c.WriteTo(w)
+			n += nn
+			if err != nil {
+				return n, err
+			}
 		}
 	}
 
@@ -532,9 +557,10 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error {
 		c := b.containers[i]
 		if c.n <= ArrayMaxSize {
 			c.array = (*[0xFFFFFFF]uint32)(unsafe.Pointer(&data[offset]))[:c.n]
-			for _, v := range c.array {
-				assert(lowbits(uint64(v)) == v, "array value out of range: %d", v)
-			}
+			// TODO: instead of commenting this out, we need to make it a configuration option
+			//for _, v := range c.array {
+			//    assert(lowbits(uint64(v)) == v, "array value out of range: %d", v)
+			//}
 			opsOffset = int(offset) + len(c.array)*4
 		} else {
 			c.bitmap = (*[0xFFFFFFF]uint64)(unsafe.Pointer(&data[offset]))[:bitmapN]
@@ -542,8 +568,9 @@ func (b *Bitmap) UnmarshalBinary(data []byte) error {
 		}
 
 		// Verify container count on load.
-		count := c.count()
-		assert(c.count() == c.n, "container count mismatch: count=%d, n=%d", count, c.n)
+		// TODO: instead of commenting this out, we need to make it a configuration option
+		//count := c.count()
+		//assert(c.count() == c.n, "container count mismatch: count=%d, n=%d", count, c.n)
 	}
 
 	// Read ops log until the end of the file.
@@ -1074,9 +1101,10 @@ func (c *container) arrayWriteTo(w io.Writer) (n int64, err error) {
 	}
 
 	// Verify all elements are valid.
-	for _, v := range c.array {
-		assert(lowbits(uint64(v)) == v, "cannot write array value out of range: %d", v)
-	}
+	// TODO: instead of commenting this out, we need to make it a configuration option
+	//for _, v := range c.array {
+	//  	assert(lowbits(uint64(v)) == v, "cannot write array value out of range: %d", v)
+	//}
 
 	nn, err := w.Write((*[0xFFFFFFF]byte)(unsafe.Pointer(&c.array[0]))[:4*c.n])
 	return int64(nn), err