infoschema: refine info cache logic to reduce the impact of DDL on information schema cache

infoschema/cache.go

@@ -109,15 +109,16 @@ func (h *InfoCache) getSchemaByTimestampNoLock(ts uint64) (InfoSchema, bool) {
 	// moreover, the most likely hit element in the array is the first one in steady mode
 	// thus it may have better performance than binary search
 	for i, is := range h.cache {
-		if is.timestamp == 0 || (i > 0 && h.cache[i-1].infoschema.SchemaMetaVersion() != is.infoschema.SchemaMetaVersion()+1) {
-			// the schema version doesn't have a timestamp or there is a gap in the schema cache
-			// ignore all the schema cache equals or less than this version in search by timestamp
-			break
+		if is.timestamp == 0 || ts < uint64(is.timestamp) {
+			continue
+		}
+		if i == 0 {
+			return is.infoschema, true
 		}
-		if ts >= uint64(is.timestamp) {
-			// found the largest version before the given ts
+		if h.cache[i-1].infoschema.SchemaMetaVersion() == is.infoschema.SchemaMetaVersion()+1 && uint64(h.cache[i-1].timestamp) > ts {
 			return is.infoschema, true
 		}
+		break
 	}
 
 	logutil.BgLogger().Debug("SCHEMA CACHE no schema found")

infoschema/cache_test.go

@@ -15,6 +15,7 @@
 package infoschema_test
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/pingcap/tidb/infoschema"
@@ -211,3 +212,82 @@ func TestReSize(t *testing.T) {
 	require.Nil(t, ic.GetByVersion(3))
 	require.Equal(t, is4, ic.GetByVersion(4))
 }
+
+func TestCacheWithSchemaTsZero(t *testing.T) {
+	ic := infoschema.NewCache(16)
+	require.NotNil(t, ic)
+
+	for i := 1; i <= 8; i++ {
+		ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, int64(i)), uint64(i))
+	}
+
+	checkFn := func(start, end int64, exist bool) {
+		require.True(t, start <= end)
+		latestSchemaVersion := ic.GetLatest().SchemaMetaVersion()
+		for ts := start; ts <= end; ts++ {
+			is := ic.GetBySnapshotTS(uint64(ts))
+			if exist {
+				require.NotNil(t, is, fmt.Sprintf("ts %d", ts))
+				if ts > latestSchemaVersion {
+					require.Equal(t, latestSchemaVersion, is.SchemaMetaVersion(), fmt.Sprintf("ts %d", ts))
+				} else {
+					require.Equal(t, ts, is.SchemaMetaVersion(), fmt.Sprintf("ts %d", ts))
+				}
+			} else {
+				require.Nil(t, is, fmt.Sprintf("ts %d", ts))
+			}
+		}
+	}
+	checkFn(1, 8, true)
+	checkFn(8, 10, true)
+
+	// mock for meet error There is no Write MVCC info for the schema version
+	ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 9), 0)
+	checkFn(1, 7, true)
+	checkFn(8, 9, false)
+	checkFn(9, 10, false)
+
+	for i := 10; i <= 16; i++ {
+		ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, int64(i)), uint64(i))
+		checkFn(1, 7, true)
+		checkFn(8, 9, false)
+		checkFn(10, 16, true)
+	}
+	require.Equal(t, 16, ic.Size())
+
+	// refill the cache
+	ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 9), 9)
+	checkFn(1, 16, true)
+	require.Equal(t, 16, ic.Size())
+
+	// Test more than capacity
+	ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 17), 17)
+	checkFn(1, 1, false)
+	checkFn(2, 17, true)
+	checkFn(2, 20, true)
+	require.Equal(t, 16, ic.Size())
+
+	// Test for there is a hole in the middle.
+	ic = infoschema.NewCache(16)
+
+	// mock for restart with full load the latest version schema.
+	ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, 100), 100)
+	checkFn(1, 99, false)
+	checkFn(100, 100, true)
+
+	for i := 1; i <= 16; i++ {
+		ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, int64(i)), uint64(i))
+	}
+	checkFn(1, 1, false)
+	checkFn(2, 15, true)
+	checkFn(16, 16, false)
+	checkFn(100, 100, true)
+	require.Equal(t, 16, ic.Size())
+
+	for i := 85; i < 100; i++ {
+		ic.Insert(infoschema.MockInfoSchemaWithSchemaVer(nil, int64(i)), uint64(i))
+	}
+	checkFn(1, 84, false)
+	checkFn(85, 100, true)
+	require.Equal(t, 16, ic.Size())
+}