ddl: use real start ts for recover table snapshot

pkg/ddl/ddl.go

@@ -1331,19 +1331,30 @@ var (
 	RunInGoTest bool
 )
 
+// GetRecoverSnapshotTS returns the snapshot timestamp for reconstructing
+// metadata from a finished drop/truncate table or drop schema job.
+func GetRecoverSnapshotTS(job *model.Job) uint64 {
+	if job.RealStartTS != 0 {
+		return job.RealStartTS
+	}
+	return job.StartTS
+}
+
 // GetDropOrTruncateTableInfoFromJobsByStore implements GetDropOrTruncateTableInfoFromJobs
 func GetDropOrTruncateTableInfoFromJobsByStore(jobs []*model.Job, gcSafePoint uint64, getTable func(uint64, int64, int64) (*model.TableInfo, error), fn func(*model.Job, *model.TableInfo) (bool, error)) (bool, error) {
 	for _, job := range jobs {
+		if job.Type != model.ActionDropTable && job.Type != model.ActionTruncateTable {
+			continue
+		}
+
+		snapshotTS := GetRecoverSnapshotTS(job)
 		// Check GC safe point for getting snapshot infoSchema.
-		err := gcutil.ValidateSnapshotWithGCSafePoint(job.StartTS, gcSafePoint)
+		err := gcutil.ValidateSnapshotWithGCSafePoint(snapshotTS, gcSafePoint)
 		if err != nil {
 			return false, err
 		}
-		if job.Type != model.ActionDropTable && job.Type != model.ActionTruncateTable {
-			continue
-		}
 
-		tbl, err := getTable(job.StartTS, job.SchemaID, job.TableID)
+		tbl, err := getTable(snapshotTS, job.SchemaID, job.TableID)
 		if err != nil {
 			if meta.ErrDBNotExists.Equal(err) {
 				// The dropped/truncated DDL maybe execute failed that caused by the parallel DDL execution,

pkg/ddl/tests/serial/BUILD.bazel

@@ -8,7 +8,7 @@ go_test(
         "serial_test.go",
     ],
     flaky = True,
-    shard_count = 20,
+    shard_count = 22,
     deps = [
         "//pkg/config",
         "//pkg/config/kerneltype",

pkg/ddl/tests/serial/serial_test.go

@@ -18,6 +18,7 @@ import (
 	"context"
 	"fmt"
 	"math"
+	"strconv"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -643,6 +644,213 @@ func TestRecoverTableByJobID(t *testing.T) {
 	require.Equal(t, false, gcEnable)
 }
 
+func TestRecoverTableUsesRealStartTSForQueuedDropTable(t *testing.T) {
+	store := createMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("create database if not exists test_recover")
+	tk.MustExec("use test_recover")
+	tk.MustExec("drop table if exists t_recover_snapshot")
+	tk.MustExec("create table t_recover_snapshot (id int primary key, col_a int, col_b int)")
+	tk.MustExec("insert into t_recover_snapshot values (1, 11, 21)")
+
+	defer func(originGC bool) {
+		if originGC {
+			util.EmulatorGCEnable()
+		} else {
+			util.EmulatorGCDisable()
+		}
+	}(util.IsEmulatorGCEnable())
+	util.EmulatorGCDisable()
+
+	var pauseSchedule atomic.Bool
+	waitSchCh := make(chan struct{})
+	var closeSchedule sync.Once
+	releaseSchedule := func() {
+		pauseSchedule.Store(false)
+		closeSchedule.Do(func() { close(waitSchCh) })
+	}
+	t.Cleanup(releaseSchedule)
+	testfailpoint.EnableCall(t, "github.com/pingcap/tidb/pkg/ddl/beforeLoadAndDeliverJobs", func() {
+		if pauseSchedule.Load() {
+			<-waitSchCh
+		}
+	})
+	pauseSchedule.Store(true)
+
+	submittedCh := make(chan struct{}, 2)
+	submitGate := make(chan struct{})
+	testfailpoint.EnableCall(t, "github.com/pingcap/tidb/pkg/ddl/waitJobSubmitted", func() {
+		submittedCh <- struct{}{}
+		<-submitGate
+	})
+	waitSubmitted := func() {
+		select {
+		case <-submittedCh:
+			submitGate <- struct{}{}
+		case <-time.After(5 * time.Second):
+			require.FailNow(t, "DDL job was not submitted")
+		}
+	}
+
+	// Two independent sessions are needed so the drop-table job can be queued
+	// after drop-column is submitted but before drop-column has changed metadata.
+	tkAlter := testkit.NewTestKit(t, store)
+	tkAlter.MustExec("use test_recover")
+	alterDoneCh := make(chan error, 1)
+	go func() {
+		_, err := tkAlter.Exec("alter table t_recover_snapshot drop column col_a")
+		alterDoneCh <- err
+	}()
+	waitSubmitted()
+
+	tkDrop := testkit.NewTestKit(t, store)
+	tkDrop.MustExec("use test_recover")
+	dropDoneCh := make(chan error, 1)
+	go func() {
+		_, err := tkDrop.Exec("drop table t_recover_snapshot")
+		dropDoneCh <- err
+	}()
+	waitSubmitted()
+
+	testfailpoint.Disable(t, "github.com/pingcap/tidb/pkg/ddl/waitJobSubmitted")
+	releaseSchedule()
+	require.NoError(t, <-alterDoneCh)
+	require.NoError(t, <-dropDoneCh)
+
+	getHistoryJobID := func(jobType string) int64 {
+		rows := tk.MustQuery(fmt.Sprintf(
+			"admin show ddl jobs where db_name = 'test_recover' and table_name = 't_recover_snapshot' and job_type = '%s'",
+			jobType,
+		)).Rows()
+		require.NotEmpty(t, rows)
+		jobID, err := strconv.ParseInt(rows[0][0].(string), 10, 64)
+		require.NoError(t, err)
+		return jobID
+	}
+
+	dropJobID := getHistoryJobID("drop table")
+	dropJob, err := ddl.GetHistoryJobByID(tk.Session(), dropJobID)
+	require.NoError(t, err)
+	require.NotNil(t, dropJob)
+	require.Greater(t, dropJob.RealStartTS, dropJob.StartTS)
+
+	gcTimeFormat := "20060102-15:04:05 -0700 MST"
+	timeBeforeDrop := time.Now().Add(-48 * time.Hour).Format(gcTimeFormat)
+	safePointSQL := `INSERT HIGH_PRIORITY INTO mysql.tidb VALUES ('tikv_gc_safe_point', '%[1]s', '')
+			       ON DUPLICATE KEY
+			       UPDATE variable_value = '%[1]s'`
+	tk.MustExec("delete from mysql.tidb where variable_name in ('tikv_gc_safe_point','tikv_gc_enable')")
+	tk.MustExec(fmt.Sprintf(safePointSQL, timeBeforeDrop))
+	require.NoError(t, gcutil.EnableGC(tk.Session()))
+
+	tk.MustExec(fmt.Sprintf("recover table by job %d", dropJobID))
+	tk.MustQuery("select column_name from information_schema.columns where table_schema = 'test_recover' and table_name = 't_recover_snapshot' order by ordinal_position").Check(testkit.Rows("id", "col_b"))
+	tk.MustQuery("select id, col_b from t_recover_snapshot").Check(testkit.Rows("1 21"))
+
+	recoverJobID := getHistoryJobID("recover table")
+	recoverJob, err := ddl.GetHistoryJobByID(tk.Session(), recoverJobID)
+	require.NoError(t, err)
+	require.NotNil(t, recoverJob)
+	require.NotNil(t, recoverJob.BinlogInfo.TableInfo)
+	colNames := make([]string, 0, len(recoverJob.BinlogInfo.TableInfo.Columns))
+	for _, col := range recoverJob.BinlogInfo.TableInfo.Columns {
+		colNames = append(colNames, col.Name.L)
+	}
+	require.Equal(t, []string{"id", "col_b"}, colNames)
+}
+
+func TestFlashbackDatabaseUsesRealStartTSForQueuedDropSchema(t *testing.T) {
+	store := createMockStore(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("drop database if exists test_recover_schema_snapshot")
+	tk.MustExec("create database test_recover_schema_snapshot")
+	tk.MustExec("create table test_recover_schema_snapshot.t (id int primary key, col_a int, col_b int)")
+	tk.MustExec("insert into test_recover_schema_snapshot.t values (1, 11, 21)")
+
+	defer func(originGC bool) {
+		if originGC {
+			util.EmulatorGCEnable()
+		} else {
+			util.EmulatorGCDisable()
+		}
+	}(util.IsEmulatorGCEnable())
+	util.EmulatorGCDisable()
+
+	var pauseSchedule atomic.Bool
+	waitSchCh := make(chan struct{})
+	var closeSchedule sync.Once
+	releaseSchedule := func() {
+		pauseSchedule.Store(false)
+		closeSchedule.Do(func() { close(waitSchCh) })
+	}
+	t.Cleanup(releaseSchedule)
+	testfailpoint.EnableCall(t, "github.com/pingcap/tidb/pkg/ddl/beforeLoadAndDeliverJobs", func() {
+		if pauseSchedule.Load() {
+			<-waitSchCh
+		}
+	})
+	pauseSchedule.Store(true)
+
+	submittedCh := make(chan struct{}, 2)
+	submitGate := make(chan struct{})
+	testfailpoint.EnableCall(t, "github.com/pingcap/tidb/pkg/ddl/waitJobSubmitted", func() {
+		submittedCh <- struct{}{}
+		<-submitGate
+	})
+	waitSubmitted := func() {
+		select {
+		case <-submittedCh:
+			submitGate <- struct{}{}
+		case <-time.After(5 * time.Second):
+			require.FailNow(t, "DDL job was not submitted")
+		}
+	}
+
+	tkAlter := testkit.NewTestKit(t, store)
+	tkAlter.MustExec("use test_recover_schema_snapshot")
+	alterDoneCh := make(chan error, 1)
+	go func() {
+		_, err := tkAlter.Exec("alter table t drop column col_a")
+		alterDoneCh <- err
+	}()
+	waitSubmitted()
+
+	tkDrop := testkit.NewTestKit(t, store)
+	dropDoneCh := make(chan error, 1)
+	go func() {
+		_, err := tkDrop.Exec("drop database test_recover_schema_snapshot")
+		dropDoneCh <- err
+	}()
+	waitSubmitted()
+
+	testfailpoint.Disable(t, "github.com/pingcap/tidb/pkg/ddl/waitJobSubmitted")
+	releaseSchedule()
+	require.NoError(t, <-alterDoneCh)
+	require.NoError(t, <-dropDoneCh)
+
+	rows := tk.MustQuery("admin show ddl jobs where db_name = 'test_recover_schema_snapshot' and job_type = 'drop schema'").Rows()
+	require.NotEmpty(t, rows)
+	dropJobID, err := strconv.ParseInt(rows[0][0].(string), 10, 64)
+	require.NoError(t, err)
+	dropJob, err := ddl.GetHistoryJobByID(tk.Session(), dropJobID)
+	require.NoError(t, err)
+	require.NotNil(t, dropJob)
+	require.Greater(t, dropJob.RealStartTS, dropJob.StartTS)
+
+	gcTimeFormat := "20060102-15:04:05 -0700 MST"
+	timeBeforeDrop := time.Now().Add(-48 * time.Hour).Format(gcTimeFormat)
+	safePointSQL := `INSERT HIGH_PRIORITY INTO mysql.tidb VALUES ('tikv_gc_safe_point', '%[1]s', '')
+			       ON DUPLICATE KEY
+			       UPDATE variable_value = '%[1]s'`
+	tk.MustExec("delete from mysql.tidb where variable_name in ('tikv_gc_safe_point','tikv_gc_enable')")
+	tk.MustExec(fmt.Sprintf(safePointSQL, timeBeforeDrop))
+	require.NoError(t, gcutil.EnableGC(tk.Session()))
+
+	tk.MustExec("flashback database test_recover_schema_snapshot")
+	tk.MustQuery("select column_name from information_schema.columns where table_schema = 'test_recover_schema_snapshot' and table_name = 't' order by ordinal_position").Check(testkit.Rows("id", "col_b"))
+	tk.MustQuery("select id, col_b from test_recover_schema_snapshot.t").Check(testkit.Rows("1 21"))
+}
+
 func TestRecoverTableByJobIDFail(t *testing.T) {
 	store := createMockStore(t)
 	tk := testkit.NewTestKit(t, store)

pkg/executor/ddl.go

@@ -449,7 +449,8 @@ func (e *DDLExec) executeRecoverTable(s *ast.RecoverTableStmt) error {
 		return infoschema.ErrTableExists.GenWithStack("Table '%-.192s' already been recover to '%-.192s', can't be recover repeatedly", tblInfo.Name.O, tbl.Meta().Name.O)
 	}
 
-	m := domain.GetDomain(e.Ctx()).GetSnapshotMeta(job.StartTS)
+	snapshotTS := ddl.GetRecoverSnapshotTS(job)
+	m := domain.GetDomain(e.Ctx()).GetSnapshotMeta(snapshotTS)
 	autoIDs, err := m.GetAutoIDAccessors(job.SchemaID, job.TableID).Get()
 	if err != nil {
 		return err
@@ -459,7 +460,7 @@ func (e *DDLExec) executeRecoverTable(s *ast.RecoverTableStmt) error {
 		SchemaID:      job.SchemaID,
 		TableInfo:     tblInfo,
 		DropJobID:     job.ID,
-		SnapshotTS:    job.StartTS,
+		SnapshotTS:    snapshotTS,
 		AutoIDs:       autoIDs,
 		OldSchemaName: job.SchemaName,
 		OldTableName:  tblInfo.Name.L,
@@ -487,14 +488,15 @@ func (e *DDLExec) getRecoverTableByJobID(s *ast.RecoverTableStmt, dom *domain.Do
 		return nil, nil, errors.Errorf("Job %v type is %v, not dropped/truncated table", job.ID, job.Type)
 	}
 
+	snapshotTS := ddl.GetRecoverSnapshotTS(job)
 	// Check GC safe point for getting snapshot infoSchema.
-	err = gcutil.ValidateSnapshot(e.Ctx(), job.StartTS)
+	err = gcutil.ValidateSnapshot(e.Ctx(), snapshotTS)
 	if err != nil {
 		return nil, nil, err
 	}
 
 	// Get the snapshot infoSchema before drop table.
-	snapInfo, err := dom.GetSnapshotInfoSchema(job.StartTS)
+	snapInfo, err := dom.GetSnapshotInfoSchema(snapshotTS)
 	if err != nil {
 		return nil, nil, err
 	}
@@ -514,7 +516,7 @@ func (e *DDLExec) getRecoverTableByJobID(s *ast.RecoverTableStmt, dom *domain.Do
 }
 
 // GetDropOrTruncateTableInfoFromJobs gets the dropped/truncated table information from DDL jobs,
-// it will use the `start_ts` of DDL job as snapshot to get the dropped/truncated table information.
+// it will use the recover snapshot of DDL job to get the dropped/truncated table information.
 func GetDropOrTruncateTableInfoFromJobs(jobs []*model.Job, gcSafePoint uint64, dom *domain.Domain, fn func(*model.Job, *model.TableInfo) (bool, error)) (bool, error) {
 	getTable := func(startTS uint64, schemaID int64, tableID int64) (*model.TableInfo, error) {
 		snapMeta := dom.GetSnapshotMeta(startTS)
@@ -613,7 +615,8 @@ func (e *DDLExec) executeFlashbackTable(s *ast.FlashBackTableStmt) error {
 		return infoschema.ErrTableExists.GenWithStack("Table '%-.192s' already been flashback to '%-.192s', can't be flashback repeatedly", s.Table.Name.O, tbl.Meta().Name.O)
 	}
 
-	m := domain.GetDomain(e.Ctx()).GetSnapshotMeta(job.StartTS)
+	snapshotTS := ddl.GetRecoverSnapshotTS(job)
+	m := domain.GetDomain(e.Ctx()).GetSnapshotMeta(snapshotTS)
 	autoIDs, err := m.GetAutoIDAccessors(job.SchemaID, job.TableID).Get()
 	if err != nil {
 		return err
@@ -623,7 +626,7 @@ func (e *DDLExec) executeFlashbackTable(s *ast.FlashBackTableStmt) error {
 		SchemaID:      job.SchemaID,
 		TableInfo:     tblInfo,
 		DropJobID:     job.ID,
-		SnapshotTS:    job.StartTS,
+		SnapshotTS:    snapshotTS,
 		AutoIDs:       autoIDs,
 		OldSchemaName: job.SchemaName,
 		OldTableName:  s.Table.Name.L,
@@ -672,15 +675,17 @@ func (e *DDLExec) getRecoverDBByName(schemaName ast.CIStr) (recoverSchemaInfo *m
 	dom := domain.GetDomain(e.Ctx())
 	fn := func(jobs []*model.Job) (bool, error) {
 		for _, job := range jobs {
+			if job.Type != model.ActionDropSchema {
+				continue
+			}
+
+			snapshotTS := ddl.GetRecoverSnapshotTS(job)
 			// Check GC safe point for getting snapshot infoSchema.
-			err = gcutil.ValidateSnapshotWithGCSafePoint(job.StartTS, gcSafePoint)
+			err = gcutil.ValidateSnapshotWithGCSafePoint(snapshotTS, gcSafePoint)
 			if err != nil {
 				return false, err
 			}
-			if job.Type != model.ActionDropSchema {
-				continue
-			}
-			snapMeta := dom.GetSnapshotMeta(job.StartTS)
+			snapMeta := dom.GetSnapshotMeta(snapshotTS)
 			schemaInfo, err := snapMeta.GetDatabase(job.SchemaID)
 			if err != nil {
 				return false, err
@@ -698,7 +703,7 @@ func (e *DDLExec) getRecoverDBByName(schemaName ast.CIStr) (recoverSchemaInfo *m
 				DBInfo:              schemaInfo,
 				LoadTablesOnExecute: true,
 				DropJobID:           job.ID,
-				SnapshotTS:          job.StartTS,
+				SnapshotTS:          snapshotTS,
 				OldSchemaName:       schemaName,
 			}
 			return true, nil