infoschema: refine info cache logic to reduce the impact of DDL on in…

…formation schema cache (#48284)

close #48285

pkg/infoschema/cache.go

@@ -99,15 +99,24 @@ 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) {
+			// is.timestamp == 0 means the schema ts is unknown, so we can't use it, then just skip it.
+			// ts < is.timestamp means the schema is newer than ts, so we can't use it too, just skip it to find the older one.
+			continue
+		}
+		// ts >= is.timestamp must be true after the above condition.
+		if i == 0 {
+			// the first element is the latest schema, so we can return it directly.
+			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 {
+			// This first condition is to make sure the schema version is continuous. If last(cache[i-1]) schema-version is 10,
+			// but current(cache[i]) schema-version is not 9, then current schema is not suitable for ts.
+			// The second condition is to make sure the cache[i-1].timestamp > ts >= cache[i].timestamp, then the current schema is suitable for ts.
 			return is.infoschema, true
 		}
+		// current schema is not suitable for ts, then break the loop to avoid the unnecessary search.
+		break
 	}
 
 	logutil.BgLogger().Debug("SCHEMA CACHE no schema found")

pkg/infoschema/test/cachetest/BUILD.bazel

@@ -8,7 +8,7 @@ go_test(
         "main_test.go",
     ],
     flaky = True,
-    shard_count = 6,
+    shard_count = 7,
     deps = [
         "//pkg/infoschema",
         "//pkg/testkit/testsetup",

pkg/infoschema/test/cachetest/cache_test.go

@@ -15,6 +15,7 @@
 package cachetest
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/pingcap/tidb/pkg/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())
+}