Clear intersecting regions in the cache when inserting a region

Signed-off-by: Yilin Chen <sticnarf@gmail.com>

internal/locate/region_cache.go

@@ -1316,6 +1316,11 @@ func (c *RegionCache) insertRegionToCache(cachedRegion *Region) {
 	if !ok || latest.GetVer() < newVer.GetVer() || latest.GetConfVer() < newVer.GetConfVer() {
 		c.mu.latestVersions[cachedRegion.VerID().id] = newVer
 	}
+	// The intersecting regions in the cache are probably stale, clear them.
+	deleted := c.mu.sorted.removeIntersecting(cachedRegion)
+	for _, r := range deleted {
+		c.removeVersionFromCache(r.cachedRegion.VerID(), r.cachedRegion.GetID())
+	}
 }
 
 // searchCachedRegion finds a region from cache by key. Like `getCachedRegion`,

internal/locate/sorted_btree.go

@@ -94,6 +94,28 @@ func (s *SortedRegions) AscendGreaterOrEqual(startKey, endKey []byte, limit int)
 	return regions
 }
 
+// removeIntersecting removes all items that have intersection with the key range of given region.
+// If the region itself is in the cache, it's not removed.
+func (s *SortedRegions) removeIntersecting(r *Region) []*btreeItem {
+	var deleted []*btreeItem
+	s.b.AscendGreaterOrEqual(newBtreeSearchItem(r.StartKey()), func(item btree.Item) bool {
+		region := item.(*btreeItem).cachedRegion
+		// Skip the item that is equal to the given region.
+		if region.VerID() == r.VerID() {
+			return true
+		}
+		if len(r.EndKey()) > 0 && bytes.Compare(region.StartKey(), r.EndKey()) >= 0 {
+			return false
+		}
+		deleted = append(deleted, item.(*btreeItem))
+		return true
+	})
+	for _, item := range deleted {
+		s.b.Delete(item)
+	}
+	return deleted
+}
+
 // Clear removes all items from the btree.
 func (s *SortedRegions) Clear() {
 	s.b.Clear(false)

internal/locate/sorted_btree_generic.go

@@ -94,6 +94,27 @@ func (s *SortedRegions) AscendGreaterOrEqual(startKey, endKey []byte, limit int)
 	return regions
 }
 
+// removeIntersecting removes all items that have intersection with the key range of given region.
+// If the region itself is in the cache, it's not removed.
+func (s *SortedRegions) removeIntersecting(r *Region) []*btreeItem {
+	var deleted []*btreeItem
+	s.b.AscendGreaterOrEqual(newBtreeSearchItem(r.StartKey()), func(item *btreeItem) bool {
+		// Skip the item that is equal to the given region.
+		if item.cachedRegion.VerID() == r.VerID() {
+			return true
+		}
+		if len(r.EndKey()) > 0 && bytes.Compare(item.cachedRegion.StartKey(), r.EndKey()) >= 0 {
+			return false
+		}
+		deleted = append(deleted, item)
+		return true
+	})
+	for _, item := range deleted {
+		s.b.Delete(item)
+	}
+	return deleted
+}
+
 // Clear removes all items from the btree.
 func (s *SortedRegions) Clear() {
 	s.b.Clear(false)