Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ddl, domain: make schema correct after canceling jobs (#7997) #8312

Merged
merged 3 commits into from
Nov 16, 2018
Merged
Show file tree
Hide file tree
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 14 additions & 11 deletions ddl/ddl_worker.go
Original file line number Diff line number Diff line change
Expand Up @@ -194,18 +194,20 @@ func (d *ddl) finishDDLJob(t *meta.Meta, job *model.Job) (err error) {
metrics.DDLWorkerHistogram.WithLabelValues(metrics.WorkerFinishDDLJob, metrics.RetLabel(err)).Observe(time.Since(startTime).Seconds())
}()

switch job.Type {
case model.ActionAddIndex:
if job.State != model.JobStateRollbackDone {
break
if !job.IsCancelled() {
switch job.Type {
case model.ActionAddIndex:
if job.State != model.JobStateRollbackDone {
break
}
// After rolling back an AddIndex operation, we need to use delete-range to delete the half-done index data.
err = d.deleteRange(job)
case model.ActionDropSchema, model.ActionDropTable, model.ActionTruncateTable, model.ActionDropIndex:
err = d.deleteRange(job)
}
if err != nil {
return errors.Trace(err)
}
// After rolling back an AddIndex operation, we need to use delete-range to delete the half-done index data.
err = d.deleteRange(job)
case model.ActionDropSchema, model.ActionDropTable, model.ActionTruncateTable, model.ActionDropIndex:
err = d.deleteRange(job)
}
if err != nil {
return errors.Trace(err)
}

_, err = t.DeQueueDDLJob()
Expand Down Expand Up @@ -292,6 +294,7 @@ func (d *ddl) handleDDLJobQueue(shouldCleanJobs bool) error {
// and retry later if the job is not cancelled.
schemaVer, runJobErr = d.runDDLJob(t, job)
if job.IsCancelled() {
txn.Reset()
err = d.finishDDLJob(t, job)
return errors.Trace(err)
}
Expand Down
116 changes: 116 additions & 0 deletions ddl/fail_db_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -14,11 +14,51 @@
package ddl_test

import (
"time"

gofail "github.com/etcd-io/gofail/runtime"
. "github.com/pingcap/check"
"github.com/pingcap/tidb/domain"
"github.com/pingcap/tidb/kv"
"github.com/pingcap/tidb/parser"
"github.com/pingcap/tidb/session"
"github.com/pingcap/tidb/store/mockstore"
"github.com/pingcap/tidb/util/testleak"
"golang.org/x/net/context"
)

var _ = Suite(&testFailDBSuite{})

type testFailDBSuite struct {
lease time.Duration
store kv.Storage
dom *domain.Domain
se session.Session
p *parser.Parser
}

func (s *testFailDBSuite) SetUpSuite(c *C) {
testleak.BeforeTest()
s.lease = 200 * time.Millisecond
var err error
s.store, err = mockstore.NewMockTikvStore()
c.Assert(err, IsNil)
session.SetSchemaLease(s.lease)
s.dom, err = session.BootstrapSession(s.store)
c.Assert(err, IsNil)
s.se, err = session.CreateSession4Test(s.store)
c.Assert(err, IsNil)
s.p = parser.New()
}

func (s *testFailDBSuite) TearDownSuite(c *C) {
s.se.Execute(context.Background(), "drop database if exists test_db_state")
s.se.Close()
s.dom.Close()
s.store.Close()
testleak.AfterTest(c)()
}

// TestInitializeOffsetAndState tests the case that the column's offset and state don't be initialized in the file of ddl_api.go when
// doing the operation of 'modify column'.
func (s *testStateChangeSuite) TestInitializeOffsetAndState(c *C) {
Expand All @@ -33,3 +73,79 @@ func (s *testStateChangeSuite) TestInitializeOffsetAndState(c *C) {
c.Assert(err, IsNil)
gofail.Disable("github.com/pingcap/tidb/ddl/uninitializedOffsetAndState")
}

// TestHalfwayCancelOperations tests the case that the schema is correct after the execution of operations are cancelled halfway.
func (s *testFailDBSuite) TestHalfwayCancelOperations(c *C) {
gofail.Enable("github.com/pingcap/tidb/ddl/truncateTableErr", `return(true)`)
defer gofail.Disable("github.com/pingcap/tidb/ddl/truncateTableErr")

// test for truncating table
_, err := s.se.Execute(context.Background(), "create database cancel_job_db")
c.Assert(err, IsNil)
_, err = s.se.Execute(context.Background(), "use cancel_job_db")
c.Assert(err, IsNil)
_, err = s.se.Execute(context.Background(), "create table t(a int)")
c.Assert(err, IsNil)
_, err = s.se.Execute(context.Background(), "insert into t values(1)")
c.Assert(err, IsNil)
_, err = s.se.Execute(context.Background(), "truncate table t")
c.Assert(err, NotNil)
// Make sure that the table's data has not been deleted.
rs, err := s.se.Execute(context.Background(), "select count(*) from t")
c.Assert(err, IsNil)
chk := rs[0].NewChunk()
err = rs[0].Next(context.Background(), chk)
c.Assert(err, IsNil)
c.Assert(chk.NumRows() == 0, IsFalse)
row := chk.GetRow(0)
c.Assert(row.Len(), Equals, 1)
c.Assert(row.GetInt64(0), DeepEquals, int64(1))
c.Assert(rs[0].Close(), IsNil)
// Reload schema.
s.dom.ResetHandle(s.store)
err = s.dom.DDL().GetHook().OnChanged(nil)
c.Assert(err, IsNil)
s.se, err = session.CreateSession4Test(s.store)
c.Assert(err, IsNil)
_, err = s.se.Execute(context.Background(), "use cancel_job_db")
c.Assert(err, IsNil)
// Test schema is correct.
_, err = s.se.Execute(context.Background(), "select * from t")
c.Assert(err, IsNil)

// test for renaming table
gofail.Enable("github.com/pingcap/tidb/ddl/errRenameTable", `return(true)`)
defer gofail.Disable("github.com/pingcap/tidb/ddl/errRenameTable")
_, err = s.se.Execute(context.Background(), "create table tx(a int)")
c.Assert(err, IsNil)
_, err = s.se.Execute(context.Background(), "insert into tx values(1)")
c.Assert(err, IsNil)
_, err = s.se.Execute(context.Background(), "rename table tx to ty")
c.Assert(err, NotNil)
// Make sure that the table's data has not been deleted.
rs, err = s.se.Execute(context.Background(), "select count(*) from tx")
c.Assert(err, IsNil)
chk = rs[0].NewChunk()
err = rs[0].Next(context.Background(), chk)
c.Assert(err, IsNil)
c.Assert(chk.NumRows() == 0, IsFalse)
row = chk.GetRow(0)
c.Assert(row.Len(), Equals, 1)
c.Assert(row.GetInt64(0), DeepEquals, int64(1))
c.Assert(rs[0].Close(), IsNil)
// Reload schema.
s.dom.ResetHandle(s.store)
err = s.dom.DDL().GetHook().OnChanged(nil)
c.Assert(err, IsNil)
s.se, err = session.CreateSession4Test(s.store)
c.Assert(err, IsNil)
_, err = s.se.Execute(context.Background(), "use cancel_job_db")
c.Assert(err, IsNil)
// Test schema is correct.
_, err = s.se.Execute(context.Background(), "select * from tx")
c.Assert(err, IsNil)

// clean up
_, err = s.se.Execute(context.Background(), "drop database cancel_job_db")
c.Assert(err, IsNil)
}
12 changes: 12 additions & 0 deletions ddl/table.go
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,13 @@ func (d *ddl) onTruncateTable(t *meta.Meta, job *model.Job) (ver int64, _ error)
job.State = model.JobStateCancelled
return ver, errors.Trace(err)
}

// gofail: var truncateTableErr bool
// if truncateTableErr {
// job.State = model.JobStateCancelled
// return ver, errors.New("occur an error after dropping table.")
// }

tblInfo.ID = newTableID
err = t.CreateTable(schemaID, tblInfo)
if err != nil {
Expand Down Expand Up @@ -313,6 +320,11 @@ func (d *ddl) onRenameTable(t *meta.Meta, job *model.Job) (ver int64, _ error) {
job.State = model.JobStateCancelled
return ver, errors.Trace(err)
}
// gofail: var renameTableErr bool
// if renameTableErr {
// job.State = model.JobStateCancelled
// return ver, errors.New("occur an error after renaming table.")
// }
tblInfo.Name = tableName
err = t.CreateTable(newSchemaID, tblInfo)
if err != nil {
Expand Down
5 changes: 5 additions & 0 deletions domain/domain.go
Original file line number Diff line number Diff line change
Expand Up @@ -463,6 +463,11 @@ func NewDomain(store kv.Storage, ddlLease time.Duration, statsLease time.Duratio
}
}

// ResetHandle resets the domain's infoschema handle. It is used for testing.
func (do *Domain) ResetHandle(store kv.Storage) {
do.infoHandle = infoschema.NewHandle(store)
}

// Init initializes a domain.
func (do *Domain) Init(ddlLease time.Duration, sysFactory func(*Domain) (pools.Resource, error)) error {
if ebd, ok := do.store.(EtcdBackend); ok {
Expand Down