statistics: skip stats cache update when incoming table is older

A stale async histogram load could overwrite a newer analyzed cache entry
because the cache accepted any table object regardless of version. Guard
`Update` and `CopyAndUpdate` with a version check so an older table cannot
replace a newer one for the same physical table.

Issue Number: ref #67355

Author: 0xPoe

pkg/statistics/handle/cache/BUILD.bazel

@@ -45,7 +45,7 @@ go_test(
     ],
     embed = [":cache"],
     flaky = True,
-    shard_count = 3,
+    shard_count = 4,
     deps = [
         "//pkg/config",
         "//pkg/statistics",

pkg/statistics/handle/cache/statscache_test.go

@@ -118,6 +118,30 @@ func TestUpdateStatsHealthyMetrics(t *testing.T) {
 	}
 }
 
+func TestUpdateSkipsOlderTableVersion(t *testing.T) {
+	current := newMockTable(t, 1, false, 1280, 0, 1)
+	current.Version = 2
+	stale := newMockTable(t, 1, false, 640, 0, 1)
+	stale.Version = 1
+
+	cache := &StatsCache{
+		c: newMockStatsCacheInner(current),
+	}
+
+	cache.Update([]*statistics.Table{stale}, nil, false)
+
+	got, ok := cache.c.Get(current.PhysicalID)
+	require.True(t, ok)
+	require.Equal(t, uint64(2), got.Version)
+	require.Equal(t, int64(1280), got.RealtimeCount)
+
+	copied := cache.CopyAndUpdate([]*statistics.Table{stale}, nil)
+	got, ok = copied.c.Get(current.PhysicalID)
+	require.True(t, ok)
+	require.Equal(t, uint64(2), got.Version)
+	require.Equal(t, int64(1280), got.RealtimeCount)
+}
+
 func newMockTable(t *testing.T, physicalID int64, pseudo bool, realtimeCount, modifyCount int64, analyzeVersion uint64) *statistics.Table {
 	hist := statistics.NewHistColl(physicalID, realtimeCount, modifyCount, 0, 0)
 	table := &statistics.Table{
@@ -162,27 +186,33 @@ func (m *mockStatsCacheInner) Values() []*statistics.Table {
 }
 
 func (m *mockStatsCacheInner) Get(id int64) (*statistics.Table, bool) {
-	panic("not implemented")
+	tbl, ok := m.items[id]
+	return tbl, ok
 }
 
 func (m *mockStatsCacheInner) Put(id int64, tbl *statistics.Table) bool {
-	panic("not implemented")
+	m.items[id] = tbl
+	return true
 }
 
 func (m *mockStatsCacheInner) Del(id int64) {
-	panic("not implemented")
+	delete(m.items, id)
 }
 
 func (m *mockStatsCacheInner) Cost() int64 {
 	panic("not implemented")
 }
 
 func (m *mockStatsCacheInner) Len() int {
-	panic("not implemented")
+	return len(m.items)
 }
 
 func (m *mockStatsCacheInner) Copy() internal.StatsCacheInner {
-	panic("not implemented")
+	copied := make(map[int64]*statistics.Table, len(m.items))
+	for id, tbl := range m.items {
+		copied[id] = tbl
+	}
+	return &mockStatsCacheInner{items: copied}
 }
 
 func (*mockStatsCacheInner) SetCapacity(int64) {

pkg/statistics/handle/cache/statscacheinner.go

@@ -143,11 +143,22 @@ func (sc *StatsCache) Version() uint64 {
 	return sc.maxTblStatsVer.Load()
 }
 
+// shouldSkipTableUpdate reports whether the incoming table stats are older than
+// what is already cached for the same physical table. Skipping the update
+// prevents a stale async histogram load from overwriting a newer analyze result.
+func (sc *StatsCache) shouldSkipTableUpdate(tbl *statistics.Table) bool {
+	current, ok := sc.c.Get(tbl.PhysicalID)
+	return ok && current != nil && current.Version > tbl.Version
+}
+
 // CopyAndUpdate copies a new cache and updates the new statistics table cache. It is only used in the COW mode.
 func (sc *StatsCache) CopyAndUpdate(tables []*statistics.Table, deletedIDs []int64) *StatsCache {
 	newCache := &StatsCache{c: sc.c.Copy()}
 	newCache.maxTblStatsVer.Store(sc.maxTblStatsVer.Load())
 	for _, tbl := range tables {
+		if newCache.shouldSkipTableUpdate(tbl) {
+			continue
+		}
 		id := tbl.PhysicalID
 		newCache.c.Put(id, tbl)
 	}
@@ -167,6 +178,9 @@ func (sc *StatsCache) CopyAndUpdate(tables []*statistics.Table, deletedIDs []int
 // Update updates the new statistics table cache.
 func (sc *StatsCache) Update(tables []*statistics.Table, deletedIDs []int64, skipMoveForwardStatsCache bool) {
 	for _, tbl := range tables {
+		if sc.shouldSkipTableUpdate(tbl) {
+			continue
+		}
 		id := tbl.PhysicalID
 		metrics.UpdateCounter.Inc()
 		sc.c.Put(id, tbl)