fix intersected stale check (#1090)

Signed-off-by: Cabinfever_B <[email protected]>
tikv · Dec 25, 2023 · db2e85c · db2e85c
1 parent d0629e1
commit db2e85c
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 9 deletions.
diff --git a/internal/locate/region_cache_test.go b/internal/locate/region_cache_test.go
@@ -1883,21 +1883,30 @@ func (s *testRegionCacheWithDelaySuite) TestInsertStaleRegion() {
 		invalidReason: r.invalidReason,
 	}
 	fakeRegion.setStore(r.getStore().clone())
-
+	keya := mocktikv.NewMvccKey([]byte("a"))
+	keyb := mocktikv.NewMvccKey([]byte("b"))
+	keyc := mocktikv.NewMvccKey([]byte("c"))
+	newRegionID := s.cluster.AllocID()
 	newPeersIDs := s.cluster.AllocIDs(1)
-	s.cluster.Split(r.GetID(), s.cluster.AllocID(), []byte("c"), newPeersIDs, newPeersIDs[0])
+	s.cluster.Split(r.GetID(), newRegionID, []byte("b"), newPeersIDs, newPeersIDs[0])
 	newPeersIDs = s.cluster.AllocIDs(1)
-	s.cluster.Split(r.GetID(), s.cluster.AllocID(), []byte("b"), newPeersIDs, newPeersIDs[0])
+	s.cluster.Split(newRegionID, s.cluster.AllocID(), []byte("c"), newPeersIDs, newPeersIDs[0])
 
 	r.invalidate(Other)
-	r2, err := s.cache.findRegionByKey(s.bo, []byte("c"), false)
+	r2, err := s.cache.findRegionByKey(s.bo, keyc, false)
 	s.NoError(err)
 	s.Equal([]byte("c"), r2.StartKey())
-	r2, err = s.cache.findRegionByKey(s.bo, []byte("b"), false)
+	r2, err = s.cache.findRegionByKey(s.bo, keyb, false)
 	s.NoError(err)
 	s.Equal([]byte("b"), r2.StartKey())
+	ra, err := s.cache.loadRegion(s.bo, keya, false)
+	s.NoError(err)
+	s.cache.mu.Lock()
+	stale := s.cache.insertRegionToCache(ra, true, true)
+	s.cache.mu.Unlock()
+	s.True(stale)
 
-	stale := !s.cache.insertRegionToCache(fakeRegion, true, true)
+	stale = !s.cache.insertRegionToCache(fakeRegion, true, true)
 	s.True(stale)
 
 	rs, err := s.cache.scanRegionsFromCache(s.bo, []byte(""), []byte(""), 100)

diff --git a/internal/locate/sorted_btree.go b/internal/locate/sorted_btree.go
@@ -99,16 +99,16 @@ func (s *SortedRegions) removeIntersecting(r *Region, verID RegionVerID) ([]*btr
 	var deleted []*btreeItem
 	var stale bool
 	s.b.AscendGreaterOrEqual(newBtreeSearchItem(r.StartKey()), func(item *btreeItem) bool {
+		if len(r.EndKey()) > 0 && bytes.Compare(item.cachedRegion.StartKey(), r.EndKey()) >= 0 {
+			return false
+		}
 		if item.cachedRegion.meta.GetRegionEpoch().GetVersion() > verID.ver {
 			logutil.BgLogger().Debug("get stale region",
 				zap.Uint64("region", verID.GetID()), zap.Uint64("ver", verID.GetVer()), zap.Uint64("conf", verID.GetConfVer()),
 				zap.Uint64("intersecting-ver", item.cachedRegion.meta.GetRegionEpoch().GetVersion()))
 			stale = true
 			return false
 		}
-		if len(r.EndKey()) > 0 && bytes.Compare(item.cachedRegion.StartKey(), r.EndKey()) >= 0 {
-			return false
-		}
 		deleted = append(deleted, item)
 		return true
 	})