bootstrap: add cluster_id to the mysql.tidb table (#59511) (#65564)

close #59476

Author: ti-chi-bot

br/pkg/restore/snap_client/systable_restore_test.go

@@ -393,7 +393,7 @@ func TestCheckPrivilegeTableRowsCollateCompatibility(t *testing.T) {
 //
 // The above variables are in the file br/pkg/restore/systable_restore.go
 func TestMonitorTheSystemTableIncremental(t *testing.T) {
-	require.Equal(t, int64(225), session.CurrentBootstrapVersion)
+	require.Equal(t, int64(226), session.CurrentBootstrapVersion)
 }
 
 func TestIsStatsTemporaryTable(t *testing.T) {

pkg/ddl/schematracker/checker.go

@@ -575,6 +575,7 @@ func (d *Checker) DoDDLJobWrapper(ctx sessionctx.Context, jobW *ddl.JobWrapper)
 
 type storageAndMore interface {
 	kv.Storage
+	kv.StorageWithPD
 	kv.EtcdBackend
 	helper.Storage
 }

pkg/session/bootstrap.go

@@ -860,6 +860,8 @@ const (
 	tidbDefOOMAction = "default_oom_action"
 	// The variable name in mysql.tidb table and it records the current DDLTableVersion
 	tidbDDLTableVersion = "ddl_table_version"
+	// The variable name in mysql.tidb table and it records the cluster id of this cluster
+	tidbClusterID = "cluster_id"
 	// Const for TiDB server version 2.
 	version2  = 2
 	version3  = 3
@@ -1254,16 +1256,19 @@ const (
 	// Add index on user field for some mysql tables.
 	version225 = 225
 
+	//   insert `cluster_id` into the `mysql.tidb` table.
+	version226 = 226
+
 	// ...
-	// [version226, version238] is the version range reserved for patches of 8.5.x
+	// [version227, version238] is the version range reserved for patches of 8.5.x
 	// ...
 	// next version should start with 239
 
 )
 
 // currentBootstrapVersion is defined as a variable, so we can modify its value for testing.
 // please make sure this is the largest version
-var currentBootstrapVersion int64 = version225
+var currentBootstrapVersion int64 = version226
 
 // DDL owner key's expired time is ManagerSessionTTL seconds, we should wait the time and give more time to have a chance to finish it.
 var internalSQLTimeout = owner.ManagerSessionTTL + 15
@@ -1444,6 +1449,7 @@ var (
 		upgradeToVer223,
 		upgradeToVer224,
 		upgradeToVer225,
+		upgradeToVer226,
 	}
 )
 
@@ -3332,6 +3338,7 @@ func upgradeToVer225(s sessiontypes.Session, ver int64) {
 	if ver >= version225 {
 		return
 	}
+
 	doReentrantDDL(s, "ALTER TABLE mysql.user ADD INDEX i_user (user)", dbterror.ErrDupKeyName)
 	doReentrantDDL(s, "ALTER TABLE mysql.global_priv ADD INDEX i_user (user)", dbterror.ErrDupKeyName)
 	doReentrantDDL(s, "ALTER TABLE mysql.db ADD INDEX i_user (user)", dbterror.ErrDupKeyName)
@@ -3341,6 +3348,30 @@ func upgradeToVer225(s sessiontypes.Session, ver int64) {
 	doReentrantDDL(s, "ALTER TABLE mysql.default_roles ADD INDEX i_user (user)", dbterror.ErrDupKeyName)
 }
 
+// writeClusterID writes cluster id into mysql.tidb
+func writeClusterID(s sessiontypes.Session) {
+	ctx, cancel := context.WithTimeout(context.Background(), time.Duration(internalSQLTimeout)*time.Second)
+	defer cancel()
+
+	clusterID := s.GetDomain().(*domain.Domain).GetPDClient().GetClusterID(ctx)
+
+	mustExecute(s, `INSERT HIGH_PRIORITY INTO %n.%n VALUES (%?, %?, "TiDB Cluster ID.") ON DUPLICATE KEY UPDATE VARIABLE_VALUE= %?`,
+		mysql.SystemDB,
+		mysql.TiDBTable,
+		tidbClusterID,
+		clusterID,
+		clusterID,
+	)
+}
+
+func upgradeToVer226(s sessiontypes.Session, ver int64) {
+	if ver >= version226 {
+		return
+	}
+
+	writeClusterID(s)
+}
+
 // initGlobalVariableIfNotExists initialize a global variable with specific val if it does not exist.
 func initGlobalVariableIfNotExists(s sessiontypes.Session, name string, val any) {
 	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnBootstrap)
@@ -3592,6 +3623,8 @@ func doDMLWorks(s sessiontypes.Session) {
 
 	writeDDLTableVersion(s)
 
+	writeClusterID(s)
+
 	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnBootstrap)
 	_, err := s.ExecuteInternal(ctx, "COMMIT")
 	if err != nil {

pkg/session/bootstrap_test.go

@@ -2591,3 +2591,61 @@ func TestTiDBUpgradeToVer219(t *testing.T) {
 	require.Contains(t, string(chk.GetRow(0).GetBytes(1)), "idx_schema_table_state")
 	require.Contains(t, string(chk.GetRow(0).GetBytes(1)), "idx_schema_table_partition_state")
 }
+
+func TestWriteClusterIDToMySQLTiDBWhenUpgradingTo225(t *testing.T) {
+	ctx := context.Background()
+	store, dom := CreateStoreAndBootstrap(t)
+	defer func() { require.NoError(t, store.Close()) }()
+
+	// `cluster_id` is inserted for a new TiDB cluster.
+	se := CreateSessionAndSetID(t, store)
+	r := MustExecToRecodeSet(t, se, `select VARIABLE_VALUE from mysql.tidb where VARIABLE_NAME='cluster_id'`)
+	req := r.NewChunk(nil)
+	err := r.Next(ctx, req)
+	require.NoError(t, err)
+	require.Equal(t, 1, req.NumRows())
+	require.NotEmpty(t, req.GetRow(0).GetBytes(0))
+	require.NoError(t, r.Close())
+	se.Close()
+
+	// bootstrap as version224
+	ver224 := version224
+	seV224 := CreateSessionAndSetID(t, store)
+	txn, err := store.Begin()
+	require.NoError(t, err)
+	m := meta.NewMutator(txn)
+	err = m.FinishBootstrap(int64(ver224))
+	require.NoError(t, err)
+	revertVersionAndVariables(t, seV224, ver224)
+	// remove the cluster_id entry from mysql.tidb table
+	MustExec(t, seV224, "delete from mysql.tidb where variable_name='cluster_id'")
+	err = txn.Commit(ctx)
+	require.NoError(t, err)
+	ver, err := getBootstrapVersion(seV224)
+	require.NoError(t, err)
+	require.Equal(t, int64(ver224), ver)
+	seV224.Close()
+
+	// upgrade to current version
+	dom.Close()
+	storeBootstrappedLock.Lock()
+	delete(storeBootstrapped, store.UUID())
+	storeBootstrappedLock.Unlock()
+	domCurVer, err := BootstrapSession(store)
+	require.NoError(t, err)
+	defer domCurVer.Close()
+	seCurVer := CreateSessionAndSetID(t, store)
+	ver, err = getBootstrapVersion(seCurVer)
+	require.NoError(t, err)
+	require.Equal(t, currentBootstrapVersion, ver)
+
+	// check if the cluster_id has been set in the `mysql.tidb` table during upgrade
+	r = MustExecToRecodeSet(t, seCurVer, `select VARIABLE_VALUE from mysql.tidb where VARIABLE_NAME='cluster_id'`)
+	req = r.NewChunk(nil)
+	err = r.Next(ctx, req)
+	require.NoError(t, err)
+	require.Equal(t, 1, req.NumRows())
+	require.NotEmpty(t, req.GetRow(0).GetBytes(0))
+	require.NoError(t, r.Close())
+	seCurVer.Close()
+}