From ed82a535e542aed22016d4d3a749ec092a3ed7aa Mon Sep 17 00:00:00 2001
From: Travis <travis@pilosa.com>
Date: Mon, 18 Nov 2019 11:39:50 -0600
Subject: [PATCH] Fix ranked cache logic to support reducing cached values
 below the threshold.

Prior to this commit, if a cache value was reduced to a value
that fell below the threshold, the operation would be ignored
and the cached value would remain at the old, higher value.

This commit also fixes logic which reduces a cached value within
the framework of uint64 values by subracting the absolute value
of the negative value (since adding a negitive doesn't work with
unsigned integers).
---
 cache.go      |  2 ++
 cache_test.go | 27 ++++++++++++++++++++++++---
 fragment.go   | 11 ++++++++++-
 3 files changed, 36 insertions(+), 4 deletions(-)

diff --git a/cache.go b/cache.go
index de836b8c8..05572fa11 100644
--- a/cache.go
+++ b/cache.go
@@ -172,6 +172,7 @@ func (c *rankCache) Add(id uint64, n uint64) {
 	// unless the count is 0, which is effectively used
 	// to clear the cache value.
 	if n < c.thresholdValue && n > 0 {
+		delete(c.entries, id)
 		return
 	}
 
@@ -185,6 +186,7 @@ func (c *rankCache) BulkAdd(id uint64, n uint64) {
 	c.mu.Lock()
 	defer c.mu.Unlock()
 	if n < c.thresholdValue {
+		delete(c.entries, id)
 		return
 	}
 
diff --git a/cache_test.go b/cache_test.go
index de1bd3a0b..d4c0982e9 100644
--- a/cache_test.go
+++ b/cache_test.go
@@ -20,8 +20,8 @@ import (
 	"github.com/pilosa/pilosa/v2"
 )
 
-// Ensure a bitmap query can be executed.
-func TestCache_Rank(t *testing.T) {
+// Ensure cache stays constrained to its configured size.
+func TestCache_Rank_Size(t *testing.T) {
 	cacheSize := uint32(3)
 	cache := pilosa.NewRankCache(cacheSize)
 	for i := 1; i < int(2*cacheSize); i++ {
@@ -31,5 +31,26 @@ func TestCache_Rank(t *testing.T) {
 	if cache.Len() != int(cacheSize) {
 		t.Fatalf("unexpected cache Size: %d!=%d expected\n", cache.Len(), cacheSize)
 	}
-
+}
+
+// Ensure cache entries set below threshold are handled appropriately.
+func TestCache_Rank_Threshold(t *testing.T) {
+	cacheSize := uint32(5)
+	cache := pilosa.NewRankCache(cacheSize)
+	for i := 1; i < int(2*cacheSize); i++ {
+		cache.Add(uint64(i), 3)
+	}
+
+	// Set the cache value for rows 4 and 5 to a number below the threshold
+	// value (which is 3), and ensure that they gets zeroed out.
+	cache.Add(4, 1)
+	cache.BulkAdd(5, 1)
+	cache.Recalculate()
+
+	if cache.Get(4) != 0 {
+		t.Fatalf("unexpected cache value after Add: %d!=%d expected\n", cache.Get(4), 0)
+	}
+	if cache.Get(5) != 0 {
+		t.Fatalf("unexpected cache value after BulkAdd: %d!=%d expected\n", cache.Get(5), 0)
+	}
 }
diff --git a/fragment.go b/fragment.go
index aa65574c6..e6c714937 100644
--- a/fragment.go
+++ b/fragment.go
@@ -2099,7 +2099,16 @@ func (f *fragment) importRoaring(ctx context.Context, data []byte, clear bool) e
 		f.rowCache.Add(rowID, nil)
 		if updateCache {
 			anyChanged = true
-			f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes))
+			if changes < 0 {
+				absChanges := uint64(-1 * changes)
+				if absChanges <= f.cache.Get(rowID) {
+					f.cache.BulkAdd(rowID, f.cache.Get(rowID)-absChanges)
+				} else {
+					f.cache.BulkAdd(rowID, 0)
+				}
+			} else {
+				f.cache.BulkAdd(rowID, f.cache.Get(rowID)+uint64(changes))
+			}
 		}
 	}
 	// we only set this if we need to update the cache