Skip to content

Commit

Permalink
ddl: add restore deleted table (#7937)
Browse files Browse the repository at this point in the history
  • Loading branch information
crazycs520 authored Jan 16, 2019
1 parent cf456f0 commit 365264c
Show file tree
Hide file tree
Showing 23 changed files with 946 additions and 40 deletions.
1 change: 1 addition & 0 deletions ddl/ddl.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
27 changes: 27 additions & 0 deletions ddl/ddl_api.go
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
29 changes: 26 additions & 3 deletions ddl/ddl_worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -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()
Expand All @@ -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
Expand Down Expand Up @@ -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
Expand Down
40 changes: 38 additions & 2 deletions ddl/delete_range.go
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ import (
"fmt"
"math"
"sync"
"sync/atomic"

"github.com/pingcap/errors"
"github.com/pingcap/parser/model"
Expand All @@ -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()
}
Expand Down Expand Up @@ -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 {
Expand Down Expand Up @@ -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 {
Expand Down
5 changes: 5 additions & 0 deletions ddl/mock.go
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
Loading

0 comments on commit 365264c

Please sign in to comment.