address comment

ddl/db_change_test.go

@@ -1012,7 +1012,7 @@ func (s *testStateChangeSuite) TestParallelDDLBeforeRunDDLJob(c *C) {
 		se1.SetConnectionID(2)
 		_, err2 := se1.Execute(context.Background(), "alter table test_table add column c2 int")
 		c.Assert(err2, NotNil)
-		c.Assert(err2.Error(), Equals, "[schema:1060]Duplicate column name 'c2'")
+		c.Assert(strings.Contains(err2.Error(), "Information schema is changed"), IsTrue)
 	}()
 
 	wg.Wait()
@@ -1034,3 +1034,35 @@ func (s *testStateChangeSuite) TestParallelAlterSchemaCharsetAndCollate(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustQuery(sql).Check(testkit.Rows("utf8mb4 utf8mb4_general_ci"))
 }
+
+// TestParallelTruncateTableAndAddColumn tests add column when truncate table.
+func (s *testStateChangeSuite) TestParallelTruncateTableAndAddColumn(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk2 := testkit.NewTestKit(c, s.store)
+	tk2.MustExec("use test")
+	tk.MustExec("use test")
+	tk.MustExec("create database if not exists test_truncate_table")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(c1 int)")
+	defer tk.MustExec("drop table t;")
+	var addColumnErr error
+	var wg sync.WaitGroup
+	hook := &ddl.TestDDLCallback{}
+	hook.OnJobRunBeforeExported = func(job *model.Job) {
+		if job.Type == model.ActionTruncateTable && job.State == model.JobStateNone {
+			wg.Add(1)
+			go func() {
+				_, addColumnErr = tk2.Exec("alter table t add column c3 int")
+				wg.Done()
+			}()
+		}
+	}
+	originalHook := s.dom.DDL().GetHook()
+	s.dom.DDL().(ddl.DDLForTest).SetHook(hook)
+	_, err := tk.Exec("truncate table t")
+	wg.Wait()
+	c.Assert(addColumnErr, NotNil)
+	c.Assert(addColumnErr.Error(), Equals, "[domain:2]Information schema is changed. [try again later]")
+	c.Assert(err, IsNil)
+	s.dom.DDL().(ddl.DDLForTest).SetHook(originalHook)
+}

ddl/db_test.go

@@ -3387,38 +3387,6 @@ func (s *testDBSuite2) TestDDLWithInvalidTableInfo(c *C) {
 	c.Assert(err.Error(), Equals, "[parser:1064]You have an error in your SQL syntax; check the manual that corresponds to your TiDB version for the right syntax to use line 1 column 94 near \"then (b / a) end));\" ")
 }
 
-// TestParallelTruncateTableAndAddColumn tests add column when truncate table.
-func (s *testDBSuite5) TestParallelTruncateTableAndAddColumn(c *C) {
-	s.tk = testkit.NewTestKit(c, s.store)
-	tk2 := testkit.NewTestKit(c, s.store)
-	tk2.MustExec("use test_db")
-	s.mustExec(c, "use test_db")
-	s.mustExec(c, "create database if not exists test_truncate_table")
-	s.mustExec(c, "drop table if exists t")
-	s.mustExec(c, "create table t(c1 int)")
-	defer s.mustExec(c, "drop table t;")
-	var addColumnErr error
-	var wg sync.WaitGroup
-	hook := &ddl.TestDDLCallback{}
-	hook.OnJobRunBeforeExported = func(job *model.Job) {
-		if job.Type == model.ActionTruncateTable && job.State == model.JobStateNone {
-			wg.Add(1)
-			go func() {
-				_, addColumnErr = tk2.Exec("alter table t add column c3 int")
-				wg.Done()
-			}()
-		}
-	}
-	originalHook := s.dom.DDL().GetHook()
-	s.dom.DDL().(ddl.DDLForTest).SetHook(hook)
-	_, err := s.tk.Exec("truncate table t")
-	wg.Wait()
-	c.Assert(addColumnErr, NotNil)
-	c.Assert(addColumnErr.Error(), Equals, "[domain:2]Information schema is changed. [try again later]")
-	c.Assert(err, IsNil)
-	s.dom.DDL().(ddl.DDLForTest).SetHook(originalHook)
-}
-
 func init() {
 	// Make sure it will only be executed once.
 	domain.SchemaOutOfDateRetryInterval = int64(50 * time.Millisecond)

ddl/ddl.go

@@ -582,7 +582,6 @@ func (d *ddl) doDDLJob(ctx sessionctx.Context, job *model.Job) error {
 		return errors.Trace(err)
 	}
 	ctx.GetSessionVars().StmtCtx.IsDDLJobInQueue = true
-	defer func() { ctx.GetSessionVars().StmtCtx.IsDDLJobInQueue = false }()
 
 	// Notice worker that we push a new job and wait the job done.
 	d.asyncNotifyWorker(job.Type)

executor/ddl.go

@@ -49,14 +49,6 @@ type DDLExec struct {
 
 // toErr converts the error to the ErrInfoSchemaChanged when the schema is outdated.
 func (e *DDLExec) toErr(err error) error {
-	if e.ctx.GetSessionVars().StmtCtx.IsDDLJobInQueue {
-		return err
-	}
-
-	if !infoschema.ErrTableNotExists.Equal(err) {
-		return err
-	}
-
 	// Before the DDL job is ready, it encouters an error that may be due to the outdated schema information.
 	// After the DDL job is ready, the ErrInfoSchemaChanged error won't happen because we are getting the schema directly from storage.
 	// So we needn't to consider this condition.
@@ -87,6 +79,8 @@ func (e *DDLExec) Next(ctx context.Context, req *chunk.Chunk) (err error) {
 		return err
 	}
 
+	defer func() { e.ctx.GetSessionVars().StmtCtx.IsDDLJobInQueue = false }()
+
 	switch x := e.stmt.(type) {
 	case *ast.AlterDatabaseStmt:
 		err = e.executeAlterDatabase(x)
@@ -121,7 +115,12 @@ func (e *DDLExec) Next(ctx context.Context, req *chunk.Chunk) (err error) {
 
 	}
 	if err != nil {
-		return e.toErr(err)
+		if (e.ctx.GetSessionVars().StmtCtx.IsDDLJobInQueue && infoschema.ErrTableNotExists.Equal(err)) ||
+			!e.ctx.GetSessionVars().StmtCtx.IsDDLJobInQueue {
+			return e.toErr(err)
+		}
+		return err
+
 	}
 
 	dom := domain.GetDomain(e.ctx)