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

pingcap · Nov 16, 2018 · 8714977 · 8714977
1 parent 34d9050
commit 8714977
Show file tree

Hide file tree

Showing 4 changed files with 147 additions and 11 deletions.
diff --git a/ddl/ddl_worker.go b/ddl/ddl_worker.go
@@ -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()
@@ -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)
 			}

diff --git a/ddl/fail_db_test.go b/ddl/fail_db_test.go
@@ -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) {
@@ -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)
+}
diff --git a/ddl/table.go b/ddl/table.go
@@ -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 {
@@ -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 {

diff --git a/domain/domain.go b/domain/domain.go
@@ -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 {