session: use mDDLTableVersion key to control backfill tables

meta/meta.go

@@ -630,6 +630,12 @@ func (m *Meta) SetMDLTables() error {
 	return errors.Trace(err)
 }
 
+// SetBackfillTables write a key into storage.
+func (m *Meta) SetBackfillTables() error {
+	err := m.txn.Set(mDDLTableVersion, []byte("3"))
+	return errors.Trace(err)
+}
+
 // CreateMySQLDatabaseIfNotExists creates mysql schema and return its DB ID.
 func (m *Meta) CreateMySQLDatabaseIfNotExists() (int64, error) {
 	id, err := m.GetSystemDBID()
@@ -681,7 +687,24 @@ func (m *Meta) CheckMDLTableExists() (bool, error) {
 	if err != nil {
 		return false, errors.Trace(err)
 	}
-	return bytes.Equal(v, []byte("2")), nil
+	ver, err := strconv.ParseUint(string(v), 10, 64)
+	if err != nil {
+		return false, errors.Trace(err)
+	}
+	return ver >= 2, nil
+}
+
+// CheckBackfillTableExists check if the tables related to concurrent DDL exists.
+func (m *Meta) CheckBackfillTableExists() (bool, error) {
+	v, err := m.txn.Get(mDDLTableVersion)
+	if err != nil {
+		return false, errors.Trace(err)
+	}
+	ver, err := strconv.ParseUint(string(v), 10, 64)
+	if err != nil {
+		return false, errors.Trace(err)
+	}
+	return ver >= 3, nil
 }
 
 // SetMetadataLock sets the metadata lock.

session/session.go

@@ -3108,36 +3108,20 @@ func InitDDLJobTables(store kv.Storage) error {
 	return kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), store, true, func(ctx context.Context, txn kv.Transaction) error {
 		t := meta.NewMeta(txn)
 		exists, err := t.CheckDDLTableExists()
-		if err != nil {
+		if err != nil || exists {
 			return errors.Trace(err)
 		}
 		dbID, err := t.CreateMySQLDatabaseIfNotExists()
 		if err != nil {
 			return err
 		}
-		if exists {
-			return initBackfillJobTables(store, t, dbID)
-		}
-
 		if err = createAndSplitTables(store, t, dbID, DDLJobTables); err != nil {
 			return err
 		}
-		if err = initBackfillJobTables(store, t, dbID); err != nil {
-			return err
-		}
 		return t.SetDDLTables()
 	})
 }
 
-// initBackfillJobTables is to create tidb_ddl_backfill and tidb_ddl_backfill_history.
-func initBackfillJobTables(store kv.Storage, t *meta.Meta, dbID int64) error {
-	tblExist, err := t.CheckTableExists(dbID, BackfillTables[0].id)
-	if err != nil || tblExist {
-		return errors.Trace(err)
-	}
-	return createAndSplitTables(store, t, dbID, BackfillTables)
-}
-
 func createAndSplitTables(store kv.Storage, t *meta.Meta, dbID int64, tables []tableBasicInfo) error {
 	tableIDs := make([]int64, 0, len(tables))
 	for _, tbl := range tables {
@@ -3199,6 +3183,25 @@ func InitMDLTable(store kv.Storage) error {
 	})
 }
 
+// InitBackfillTable is to create tidb_ddl_backfill and tidb_ddl_backfill_history, which is used for distreorg.
+func InitBackfillTable(store kv.Storage) error {
+	return kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), store, true, func(ctx context.Context, txn kv.Transaction) error {
+		t := meta.NewMeta(txn)
+		exists, err := t.CheckBackfillTableExists()
+		if err != nil || exists {
+			return errors.Trace(err)
+		}
+		dbID, err := t.CreateMySQLDatabaseIfNotExists()
+		if err != nil {
+			return err
+		}
+		if err = createAndSplitTables(store, t, dbID, BackfillTables); err != nil {
+			return err
+		}
+		return t.SetBackfillTables()
+	})
+}
+
 // InitMDLVariableForBootstrap initializes the metadata lock variable.
 func InitMDLVariableForBootstrap(store kv.Storage) error {
 	err := kv.RunInNewTxn(kv.WithInternalSourceType(context.Background(), kv.InternalTxnDDL), store, true, func(ctx context.Context, txn kv.Transaction) error {
@@ -3280,6 +3283,10 @@ func BootstrapSession(store kv.Storage) (*domain.Domain, error) {
 	if err != nil {
 		return nil, err
 	}
+	err = InitBackfillTable(store)
+	if err != nil {
+		return nil, err
+	}
 	ver := getStoreBootstrapVersion(store)
 	if ver == notBootstrapped {
 		runInBootstrapSession(store, bootstrap)

session/session_test.go

@@ -37,10 +37,16 @@ func TestInitMetaTable(t *testing.T) {
 		tk.MustExec(sql.SQL)
 	}
 
+	for _, sql := range session.BackfillTables {
+		tk.MustExec(sql.SQL)
+	}
+
 	tbls := map[string]struct{}{
-		"tidb_ddl_job":     {},
-		"tidb_ddl_reorg":   {},
-		"tidb_ddl_history": {},
+		"tidb_ddl_job":              {},
+		"tidb_ddl_reorg":            {},
+		"tidb_ddl_history":          {},
+		"tidb_ddl_backfill":         {},
+		"tidb_ddl_backfill_history": {},
 	}
 
 	for tbl := range tbls {