WIP increase caching and topn adjustments

cache.go

@@ -14,6 +14,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
 
@@ -22,7 +23,6 @@ type Cache interface {
 
 	// Updates the cache, if necessary.
 	Invalidate()
-	Refresh()
 
 	// Returns an ordered list of the top ranked bitmaps.
 	Top() []BitmapPair
@@ -44,6 +44,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)
@@ -87,7 +91,7 @@ func (c *LRUCache) Top() []BitmapPair {
 }
 
 func (c *LRUCache) onEvicted(key lru.Key, _ interface{}) { delete(c.counts, key.(uint64)) }
-func (c *LRUCache) Refresh() { 
+func (c *LRUCache) Refresh() {
 }
 
 // Ensure LRUCache implements Cache.
@@ -122,18 +126,26 @@ func (c *RankCache) Add(bitmapID uint64, n uint64) {
 
 	c.entries[bitmapID] = n
 
-
+	c.Invalidate()
 	// 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)
-				}
+		for id, n := range c.entries {
+			if n <= c.ThresholdValue {
+				delete(c.entries, id)
 			}
+		}
 	}
 }
 
+// Add adds a bitmap to the cache unsorted you should Invalidate after completion
+func (c *RankCache) BulkAdd(bitmapID uint64, n uint64) {
+	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] }
 
@@ -150,19 +162,9 @@ 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()
-	}
-}
-func (c *RankCache) Refresh() { 
-		c.update()
-}
-
 // update reorders the entries by rank.
-func (c *RankCache) update() {
+func (c *RankCache) Invalidate() {
+	//fmt.Println("RankCache Update")
 	// Convert cache to a sorted list.
 	rankings := make([]BitmapPair, 0, len(c.entries))
 	for id, n := range c.entries {

fragment.go

@@ -53,21 +53,22 @@ const (
 	//TODO CHANGING FOR TEST TO 10x
 	DefaultFragmentMaxOpN = 2000
 )
+
 type BitmapCacher interface {
-	Fetch(id uint64)(*Bitmap,bool)
+	Fetch(id uint64) (*Bitmap, bool)
 	Add(id uint64, b *Bitmap)
 }
 
+type Simple struct {
+	cache map[uint64]*Bitmap
+}
 
-type Simple struct{
-	cache	map[uint64]*Bitmap
+func (s *Simple) Fetch(id uint64) (*Bitmap, bool) {
+	m, ok := s.cache[id]
+	return m, ok
 }
-func (s *Simple)Fetch(id uint64)(*Bitmap,bool){
-	m,ok:=s.cache[id]
-	return m,ok
-}
-func (s *Simple)Add(id uint64,p*Bitmap){
-	s.cache[id]=p
+func (s *Simple) Add(id uint64, p *Bitmap) {
+	s.cache[id] = p
 }
 
 // Fragment represents the intersection of a frame and slice in a database.
@@ -105,8 +106,7 @@ type Fragment struct {
 	BitmapAttrStore *AttrStore
 
 	stats StatsClient
-	turbo  BitmapCacher
-	
+	turbo BitmapCacher
 }
 
 // NewFragment returns a new instance of Fragment.
@@ -173,6 +173,7 @@ func (f *Fragment) Open() error {
 
 // openStorage opens the storage bitmap.
 func (f *Fragment) openStorage() error {
+	//f.logger().Printf("Open Storage %s/%s/%d", f.db, f.frame, f.slice)
 	// Create a roaring bitmap to serve as storage for the slice.
 	f.storage = roaring.NewBitmap()
 
@@ -260,9 +261,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, false).Count()
+		f.cache.BulkAdd(bitmapID, n)
 	}
+	f.cache.Invalidate()
 
 	return nil
 }
@@ -326,15 +329,14 @@ 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, false)
 }
 
-
-func (f *Fragment) bitmap(bitmapID uint64) *Bitmap {
-	r,ok:=f.turbo.Fetch(bitmapID)
-		if ok && r != nil{
-			return r
-		}
+func (f *Fragment) bitmap(bitmapID uint64, updateCache bool) *Bitmap {
+	r, ok := f.turbo.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)
@@ -349,9 +351,11 @@ func (f *Fragment) bitmap(bitmapID uint64) *Bitmap {
 	}
 	bm.InvalidateCount()
 
-	// Update cache.
-	f.cache.Add(bitmapID, bm.Count())
-	f.turbo.Add(bitmapID, bm)
+	if updateCache {
+		// Update cache.
+		f.cache.Add(bitmapID, bm.Count())
+		f.turbo.Add(bitmapID, bm)
+	}
 
 	return bm
 }
@@ -391,7 +395,7 @@ func (f *Fragment) setBit(bitmapID, profileID uint64) (changed bool, bool error)
 	}
 
 	// Update the cache.
-	if f.bitmap(bitmapID).SetBit(profileID) {
+	if f.bitmap(bitmapID, true).SetBit(profileID) {
 		changed = true
 	}
 
@@ -435,7 +439,7 @@ func (f *Fragment) clearBit(bitmapID, profileID uint64) (bool, error) {
 	}
 
 	// Update the cache.
-	if f.bitmap(bitmapID).ClearBit(profileID) {
+	if f.bitmap(bitmapID, true).ClearBit(profileID) {
 		return true, nil
 	}
 
@@ -570,7 +574,16 @@ func (f *Fragment) Top(opt TopOptions) ([]Pair, error) {
 	return results, nil
 }
 
+func debugDumpPairs(pairs []BitmapPair) {
+	fmt.Println("=====Start")
+	for i, pair := range pairs {
+		fmt.Println(i, pair.ID, pair.Count)
+	}
+	fmt.Println("=====Stop")
+}
+
 func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair {
+	//fmt.Println("DEBUG topBitmapPairs")
 	// If no specific bitmaps are requested, retrieve top bitmaps.
 	if len(bitmapIDs) == 0 {
 		f.mu.Lock()
@@ -597,6 +610,8 @@ func (f *Fragment) topBitmapPairs(bitmapIDs []uint64) []BitmapPair {
 			Count: f.Bitmap(bitmapID).Count(),
 		}
 	}
+	sort.Sort(BitmapPairs(pairs))
+	//debugDumpPairs(pairs)
 	return pairs
 }
 
@@ -906,10 +921,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())
+			f.cache.BulkAdd(bitmapID, f.bitmap(bitmapID, false).Count())
 		}
 
-		f.cache.Refresh()
+		f.cache.Invalidate()
 		return nil
 	}(); err != nil {
 		_ = f.closeStorage()

roaring/roaring.go

@@ -467,7 +467,7 @@ func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) {
 	buf := make([]byte, headerSize+(containerCount*(4+8+4)))
 	binary.LittleEndian.PutUint32(buf[0:], cookie)
 	binary.LittleEndian.PutUint32(buf[4:], uint32(containerCount))
-	empty:=0
+	empty := 0
 	// Encode keys and cardinality.
 	for i, key := range b.keys {
 		c := b.containers[i]
@@ -479,20 +479,20 @@ func (b *Bitmap) WriteTo(w io.Writer) (n int64, err error) {
 		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{
+		} else {
 			empty++
 		}
 	}
 
 	// Write the offset for each container block.
 	offset := uint32(len(buf))
-	empty=0
+	empty = 0
 	for i, c := range b.containers {
 
 		if c.n > 0 {
-		binary.LittleEndian.PutUint32(buf[headerSize+(containerCount*12)+((i-empty)*4):], uint32(offset))
-		}else{
-		  empty++
+			binary.LittleEndian.PutUint32(buf[headerSize+(containerCount*12)+((i-empty)*4):], uint32(offset))
+		} else {
+			empty++
 		}
 		offset += uint32(c.size())
 	}
@@ -1392,6 +1392,7 @@ func unionArrayBitmap(a, b *container) *container {
 			break
 		} else if i >= len(a.array) {
 			output.add(vb)
+			continue
 		} else if eof {
 			output.add(a.array[i])
 			i++