ddl: add restore deleted table

ddl/ddl.go

@@ -214,6 +214,7 @@ type DDL interface {
 	CreateView(ctx sessionctx.Context, stmt *ast.CreateViewStmt) error
 	CreateTableWithLike(ctx sessionctx.Context, ident, referIdent ast.Ident, ifNotExists bool) error
 	DropTable(ctx sessionctx.Context, tableIdent ast.Ident) (err error)
+	RestoreTable(ctx sessionctx.Context, tbInfo *model.TableInfo, schemaID, autoID, dropJobID int64, snapshotTS uint64) (err error)
 	DropView(ctx sessionctx.Context, tableIdent ast.Ident) (err error)
 	CreateIndex(ctx sessionctx.Context, tableIdent ast.Ident, unique bool, indexName model.CIStr,
 		columnNames []*ast.IndexColName, indexOption *ast.IndexOption) error

ddl/ddl_api.go

@@ -1064,6 +1064,33 @@ func (d *ddl) CreateTable(ctx sessionctx.Context, s *ast.CreateTableStmt) (err e
 	return errors.Trace(err)
 }
 
+func (d *ddl) RestoreTable(ctx sessionctx.Context, tbInfo *model.TableInfo, schemaID, autoID, dropJobID int64, snapshotTS uint64) (err error) {
+	is := d.GetInformationSchema(ctx)
+	// Check schema exist.
+	schema, ok := is.SchemaByID(schemaID)
+	if !ok {
+		return errors.Trace(infoschema.ErrDatabaseNotExists.GenWithStackByArgs(
+			fmt.Sprintf("(Schema ID %d)", schemaID),
+		))
+	}
+	// Check not exist table with same name.
+	if ok := is.TableExists(schema.Name, tbInfo.Name); ok {
+		return infoschema.ErrTableExists.GenWithStackByArgs(tbInfo.Name)
+	}
+
+	tbInfo.State = model.StateNone
+	job := &model.Job{
+		SchemaID:   schemaID,
+		TableID:    tbInfo.ID,
+		Type:       model.ActionRestoreTable,
+		BinlogInfo: &model.HistoryInfo{},
+		Args:       []interface{}{tbInfo, autoID, dropJobID, snapshotTS, restoreTableCheckFlagNone},
+	}
+	err = d.doDDLJob(ctx, job)
+	err = d.callHookOnChanged(err)
+	return errors.Trace(err)
+}
+
 func (d *ddl) CreateView(ctx sessionctx.Context, s *ast.CreateViewStmt) (err error) {
 	ident := ast.Ident{Name: s.ViewName.Name, Schema: s.ViewName.Schema}
 	is := d.GetInformationSchema(ctx)

ddl/ddl_worker.go

@@ -277,9 +277,13 @@ func (w *worker) finishDDLJob(t *meta.Meta, job *model.Job) (err error) {
 		case model.ActionDropSchema, model.ActionDropTable, model.ActionTruncateTable, model.ActionDropIndex, model.ActionDropTablePartition, model.ActionTruncateTablePartition:
 			err = w.deleteRange(job)
 		}
-		if err != nil {
-			return errors.Trace(err)
-		}
+	}
+	switch job.Type {
+	case model.ActionRestoreTable:
+		err = finishRestoreTable(w, t, job)
+	}
+	if err != nil {
+		return errors.Trace(err)
 	}
 
 	_, err = t.DeQueueDDLJob()
@@ -293,6 +297,23 @@ func (w *worker) finishDDLJob(t *meta.Meta, job *model.Job) (err error) {
 	return errors.Trace(err)
 }
 
+func finishRestoreTable(w *worker, t *meta.Meta, job *model.Job) error {
+	tbInfo := &model.TableInfo{}
+	var autoID, dropJobID, restoreTableCheckFlag int64
+	var snapshotTS uint64
+	err := job.DecodeArgs(tbInfo, &autoID, &dropJobID, &snapshotTS, &restoreTableCheckFlag)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if restoreTableCheckFlag == restoreTableCheckFlagEnableGC {
+		err = enableGC(w)
+		if err != nil {
+			return errors.Trace(err)
+		}
+	}
+	return nil
+}
+
 func isDependencyJobDone(t *meta.Meta, job *model.Job) (bool, error) {
 	if job.DependencyID == noneDependencyJob {
 		return true, nil
@@ -497,6 +518,8 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 		ver, err = onAddTablePartition(t, job)
 	case model.ActionModifyTableCharsetAndCollate:
 		ver, err = onModifyTableCharsetAndCollate(t, job)
+	case model.ActionRestoreTable:
+		ver, err = w.onRestoreTable(d, t, job)
 	default:
 		// Invalid job, cancel it.
 		job.State = model.JobStateCancelled

ddl/delete_range.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"math"
 	"sync"
+	"sync/atomic"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/parser/model"
@@ -38,9 +39,16 @@ const (
 	delBackLog   = 128
 )
 
+// enableEmulatorGC means whether to enable emulator GC. The default is enable.
+// In some unit tests, we want to stop emulator GC, then wen can set enableEmulatorGC to 0.
+var emulatorGCEnable = int32(1)
+
 type delRangeManager interface {
 	// addDelRangeJob add a DDL job into gc_delete_range table.
 	addDelRangeJob(job *model.Job) error
+	// removeFromGCDeleteRange removes the deleting table job from gc_delete_range table by jobID and tableID.
+	// It's use for recover the table that was mistakenly deleted.
+	removeFromGCDeleteRange(jobID, tableID int64) error
 	start()
 	clear()
 }
@@ -90,6 +98,17 @@ func (dr *delRange) addDelRangeJob(job *model.Job) error {
 	return nil
 }
 
+// removeFromGCDeleteRange implements delRangeManager interface.
+func (dr *delRange) removeFromGCDeleteRange(jobID, tableID int64) error {
+	ctx, err := dr.sessPool.get()
+	if err != nil {
+		return errors.Trace(err)
+	}
+	defer dr.sessPool.put(ctx)
+	err = util.RemoveFromGCDeleteRange(ctx, jobID, tableID)
+	return errors.Trace(err)
+}
+
 // start implements delRangeManager interface.
 func (dr *delRange) start() {
 	if !dr.storeSupport {
@@ -117,11 +136,28 @@ func (dr *delRange) startEmulator() {
 		case <-dr.quitCh:
 			return
 		}
-		err := dr.doDelRangeWork()
-		terror.Log(errors.Trace(err))
+		if IsEmulatorGCEnable() {
+			err := dr.doDelRangeWork()
+			terror.Log(errors.Trace(err))
+		}
 	}
 }
 
+// EmulatorGCEnable enables emulator gc. It exports for testing.
+func EmulatorGCEnable() {
+	atomic.StoreInt32(&emulatorGCEnable, 1)
+}
+
+// EmulatorGCDisable disables emulator gc. It exports for testing.
+func EmulatorGCDisable() {
+	atomic.StoreInt32(&emulatorGCEnable, 0)
+}
+
+// IsEmulatorGCEnable indicates whether emulator GC enabled. It exports for testing.
+func IsEmulatorGCEnable() bool {
+	return atomic.LoadInt32(&emulatorGCEnable) == 1
+}
+
 func (dr *delRange) doDelRangeWork() error {
 	ctx, err := dr.sessPool.get()
 	if err != nil {

ddl/mock.go

@@ -126,6 +126,11 @@ func (dr *mockDelRange) addDelRangeJob(job *model.Job) error {
 	return nil
 }
 
+// removeFromGCDeleteRange implements delRangeManager interface.
+func (dr *mockDelRange) removeFromGCDeleteRange(jobID, tableID int64) error {
+	return nil
+}
+
 // start implements delRangeManager interface.
 func (dr *mockDelRange) start() {
 	return