From e651153f1cf72786e6cadcf2cd9dae5e78926f20 Mon Sep 17 00:00:00 2001
From: Ben Johnson <benbjohnson@yahoo.com>
Date: Fri, 17 Aug 2018 11:11:47 -0600
Subject: [PATCH] Add TagValueSeriesIDCache.Delete().

---
 tsdb/index/tsi1/cache.go | 55 ++++++++++++++++++++++++++++------------
 tsdb/index/tsi1/index.go | 14 ++++++++--
 2 files changed, 51 insertions(+), 18 deletions(-)

diff --git a/tsdb/index/tsi1/cache.go b/tsdb/index/tsi1/cache.go
index 1f8c69b0a0..6bcf84c5a7 100644
--- a/tsdb/index/tsi1/cache.go
+++ b/tsdb/index/tsi1/cache.go
@@ -141,28 +141,51 @@ EVICT:
 	c.checkEviction()
 }
 
+// Delete removes x from the tuple {name, key, value} if it exists.
+// This method takes a lock on the underlying SeriesIDSet.
+func (c *TagValueSeriesIDCache) Delete(name, key, value []byte, x uint64) {
+	c.Lock()
+	c.delete(name, key, value, x)
+	c.Unlock()
+}
+
+// delete removes x from the tuple {name, key, value} if it exists.
+func (c *TagValueSeriesIDCache) delete(name, key, value []byte, x uint64) {
+	if mmap, ok := c.cache[string(name)]; ok {
+		if tkmap, ok := mmap[string(key)]; ok {
+			if ele, ok := tkmap[string(value)]; ok {
+				if ss := ele.Value.(*seriesIDCacheElement).SeriesIDSet; ss != nil {
+					ele.Value.(*seriesIDCacheElement).SeriesIDSet.Remove(x)
+				}
+			}
+		}
+	}
+}
+
 // checkEviction checks if the cache is too big, and evicts the least recently used
 // item if it is.
 func (c *TagValueSeriesIDCache) checkEviction() {
-	if c.evictor.Len() > c.capacity {
-		e := c.evictor.Back() // Least recently used item.
-		listElement := e.Value.(*seriesIDCacheElement)
-		name := listElement.name
-		key := listElement.key
-		value := listElement.value
+	if c.evictor.Len() <= c.capacity {
+		return
+	}
 
-		c.evictor.Remove(e)                                       // Remove from evictor
-		delete(c.cache[string(name)][string(key)], string(value)) // Remove from hashmap of items.
+	e := c.evictor.Back() // Least recently used item.
+	listElement := e.Value.(*seriesIDCacheElement)
+	name := listElement.name
+	key := listElement.key
+	value := listElement.value
 
-		// Check if there are no more tag values for the tag key.
-		if len(c.cache[string(name)][string(key)]) == 0 {
-			delete(c.cache[string(name)], string(key))
-		}
+	c.evictor.Remove(e)                                       // Remove from evictor
+	delete(c.cache[string(name)][string(key)], string(value)) // Remove from hashmap of items.
 
-		// Check there are no more tag keys for the measurement.
-		if len(c.cache[string(name)]) == 0 {
-			delete(c.cache, string(name))
-		}
+	// Check if there are no more tag values for the tag key.
+	if len(c.cache[string(name)][string(key)]) == 0 {
+		delete(c.cache[string(name)], string(key))
+	}
+
+	// Check there are no more tag keys for the measurement.
+	if len(c.cache[string(name)]) == 0 {
+		delete(c.cache, string(name))
 	}
 }
 
diff --git a/tsdb/index/tsi1/index.go b/tsdb/index/tsi1/index.go
index fcece16de0..24e7df97cf 100644
--- a/tsdb/index/tsi1/index.go
+++ b/tsdb/index/tsi1/index.go
@@ -789,8 +789,18 @@ func (i *Index) DropSeries(seriesID uint64, key []byte, cascade bool) error {
 		return nil
 	}
 
-	// Extract measurement name.
-	name, _ := models.ParseKeyBytes(key)
+	// Extract measurement name & tags.
+	name, tags := models.ParseKeyBytes(key)
+
+	// If there are cached sets for any of the tag pairs, they will need to be
+	// updated with the series id.
+	i.tagValueCache.RLock()
+	if i.tagValueCache.measurementContainsSets(name) {
+		for _, pair := range tags {
+			i.tagValueCache.delete(name, pair.Key, pair.Value, seriesID) // Takes a lock on the series id set
+		}
+	}
+	i.tagValueCache.RUnlock()
 
 	// Check if that was the last series for the measurement in the entire index.
 	if ok, err := i.MeasurementHasSeries(name); err != nil {