From 79ff11884a53eaf708d2a5b80276910d74eaab17 Mon Sep 17 00:00:00 2001
From: Hangjie Mo <mohangjie1995@gmail.com>
Date: Thu, 2 Feb 2023 14:57:01 +0800
Subject: [PATCH] update

---
 meta/meta.go       | 25 ++++++++++++++++++++++++-
 session/session.go | 41 ++++++++++++++++++++++++-----------------
 2 files changed, 48 insertions(+), 18 deletions(-)

diff --git a/meta/meta.go b/meta/meta.go
index 36a4b0730d00e..7fb3fc7fbfdbf 100644
--- a/meta/meta.go
+++ b/meta/meta.go
@@ -647,6 +647,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()
@@ -698,7 +704,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.
diff --git a/session/session.go b/session/session.go
index 11226483660bb..f9a4fa2e2fe5f 100644
--- a/session/session.go
+++ b/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)