*: refine some tests for concurrent DDL (pingcap#34215)

ref pingcap#33008

ddl/column_modify_test.go

@@ -1049,16 +1049,17 @@ func TestWriteReorgForColumnTypeChangeOnAmendTxn(t *testing.T) {
 		}()
 		hook := &ddl.TestDDLCallback{Do: dom}
 		times := 0
-		hook.OnJobUpdatedExported = func(job *model.Job) {
-			if job.Type != model.ActionModifyColumn || checkErr != nil ||
-				(job.SchemaState != startColState && job.SchemaState != commitColState) {
+		hook.OnJobRunBeforeExported = func(job *model.Job) {
+			if job.Type != model.ActionModifyColumn || checkErr != nil || job.SchemaState != startColState {
 				return
 			}
 
-			if job.SchemaState == startColState {
-				tk1.MustExec("use test")
-				tk1.MustExec("begin pessimistic;")
-				tk1.MustExec("insert into t1 values(101, 102, 103)")
+			tk1.MustExec("use test")
+			tk1.MustExec("begin pessimistic;")
+			tk1.MustExec("insert into t1 values(101, 102, 103)")
+		}
+		hook.OnJobUpdatedExported = func(job *model.Job) {
+			if job.Type != model.ActionModifyColumn || checkErr != nil || job.SchemaState != commitColState {
 				return
 			}
 			if times == 0 {

ddl/db_integration_test.go

@@ -27,6 +27,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/config"
+	"github.com/pingcap/tidb/ddl"
 	"github.com/pingcap/tidb/domain"
 	"github.com/pingcap/tidb/errno"
 	"github.com/pingcap/tidb/infoschema"
@@ -678,66 +679,31 @@ func TestTableDDLWithTimeType(t *testing.T) {
 }
 
 func TestUpdateMultipleTable(t *testing.T) {
-	store, clean := testkit.CreateMockStore(t)
+	store, dom, clean := testkit.CreateMockStoreAndDomain(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
-	tk.MustExec("create database umt_db")
-	tk.MustExec("use umt_db")
+	tk.MustExec("use test")
 	tk.MustExec("create table t1 (c1 int, c2 int)")
 	tk.MustExec("insert t1 values (1, 1), (2, 2)")
 	tk.MustExec("create table t2 (c1 int, c2 int)")
 	tk.MustExec("insert t2 values (1, 3), (2, 5)")
-	ctx := tk.Session()
-	dom := domain.GetDomain(ctx)
-	is := dom.InfoSchema()
-	db, ok := is.SchemaByName(model.NewCIStr("umt_db"))
-	require.True(t, ok)
-	t1Tbl, err := is.TableByName(model.NewCIStr("umt_db"), model.NewCIStr("t1"))
-	require.NoError(t, err)
-	t1Info := t1Tbl.Meta()
-
-	// Add a new column in write only state.
-	newColumn := &model.ColumnInfo{
-		ID:                 100,
-		Name:               model.NewCIStr("c3"),
-		Offset:             2,
-		DefaultValue:       9,
-		OriginDefaultValue: 9,
-		FieldType:          *types.NewFieldType(mysql.TypeLonglong),
-		State:              model.StateWriteOnly,
+	tk2 := testkit.NewTestKit(t, store)
+	tk2.MustExec("use test")
+
+	d := dom.DDL()
+	hook := &ddl.TestDDLCallback{Do: dom}
+	hook.OnJobUpdatedExported = func(job *model.Job) {
+		if job.SchemaState == model.StateWriteOnly {
+			tk2.MustExec("update t1, t2 set t1.c1 = 8, t2.c2 = 10 where t1.c2 = t2.c1")
+			tk2.MustQuery("select * from t1").Check(testkit.Rows("8 1", "8 2"))
+			tk2.MustQuery("select * from t2").Check(testkit.Rows("1 10", "2 10"))
+		}
 	}
-	t1Info.Columns = append(t1Info.Columns, newColumn)
-
-	err = kv.RunInNewTxn(context.Background(), store, false, func(ctx context.Context, txn kv.Transaction) error {
-		m := meta.NewMeta(txn)
-		_, err = m.GenSchemaVersion()
-		require.NoError(t, err)
-		require.Nil(t, m.UpdateTable(db.ID, t1Info))
-		return nil
-	})
-	require.NoError(t, err)
-	err = dom.Reload()
-	require.NoError(t, err)
-
-	tk.MustExec("update t1, t2 set t1.c1 = 8, t2.c2 = 10 where t1.c2 = t2.c1")
-	tk.MustQuery("select * from t1").Check(testkit.Rows("8 1", "8 2"))
-	tk.MustQuery("select * from t2").Check(testkit.Rows("1 10", "2 10"))
+	d.SetHook(hook)
 
-	newColumn.State = model.StatePublic
-
-	err = kv.RunInNewTxn(context.Background(), store, false, func(ctx context.Context, txn kv.Transaction) error {
-		m := meta.NewMeta(txn)
-		_, err = m.GenSchemaVersion()
-		require.NoError(t, err)
-		require.Nil(t, m.UpdateTable(db.ID, t1Info))
-		return nil
-	})
-	require.NoError(t, err)
-	err = dom.Reload()
-	require.NoError(t, err)
+	tk.MustExec("alter table t1 add column c3 bigint default 9")
 
 	tk.MustQuery("select * from t1").Check(testkit.Rows("8 1 9", "8 2 9"))
-	tk.MustExec("drop database umt_db")
 }
 
 func TestNullGeneratedColumn(t *testing.T) {

ddl/table_test.go

@@ -335,7 +335,8 @@ func TestRenameTables(t *testing.T) {
 	job := testRenameTables(t, ctx, d, []int64{dbInfo.ID, dbInfo.ID}, []int64{dbInfo.ID, dbInfo.ID}, []*model.CIStr{&newTblInfos[0].Name, &newTblInfos[1].Name}, []int64{tblInfos[0].ID, tblInfos[1].ID}, []*model.CIStr{&dbInfo.Name, &dbInfo.Name}, []*model.CIStr{&tblInfos[0].Name, &tblInfos[1].Name})
 
 	txn, _ := ctx.Txn(true)
-	historyJob, _ := meta.NewMeta(txn).GetHistoryDDLJob(job.ID)
+	historyJob, err := meta.NewMeta(txn).GetHistoryDDLJob(job.ID)
+	require.NoError(t, err)
 	wantTblInfos := historyJob.BinlogInfo.MultipleTableInfos
 	require.Equal(t, wantTblInfos[0].Name.L, "tt1")
 	require.Equal(t, wantTblInfos[1].Name.L, "tt2")

executor/executor_failpoint_test.go

@@ -269,12 +269,13 @@ func TestCollectCopRuntimeStats(t *testing.T) {
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test;")
 	tk.MustExec("create table t1 (a int, b int)")
+	time.Sleep(1 * time.Second)
 	tk.MustExec("set tidb_enable_collect_execution_info=1;")
 	require.NoError(t, failpoint.Enable("tikvclient/tikvStoreRespResult", `return(true)`))
 	rows := tk.MustQuery("explain analyze select * from t1").Rows()
 	require.Len(t, rows, 2)
 	explain := fmt.Sprintf("%v", rows[0])
-	require.Regexp(t, ".*rpc_num: 2, .*regionMiss:.*", explain)
+	require.Regexp(t, ".*rpc_num: .*, .*regionMiss:.*", explain)
 	require.NoError(t, failpoint.Disable("tikvclient/tikvStoreRespResult"))
 }
 

executor/executor_issue_test.go

@@ -332,19 +332,19 @@ func TestIssue28650(t *testing.T) {
 }
 
 func TestIssue30289(t *testing.T) {
+	fpName := "github.com/pingcap/tidb/executor/issue30289"
 	store, clean := testkit.CreateMockStore(t)
 	defer clean()
 	tk := testkit.NewTestKit(t, store)
 	tk.MustExec("use test")
-	fpName := "github.com/pingcap/tidb/executor/issue30289"
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t(a int)")
 	require.NoError(t, failpoint.Enable(fpName, `return(true)`))
 	defer func() {
 		require.NoError(t, failpoint.Disable(fpName))
 	}()
-	tk.MustExec("drop table if exists t")
-	tk.MustExec("create table t(a int)")
 	err := tk.QueryToErr("select /*+ hash_join(t1) */ * from t t1 join t t2 on t1.a=t2.a")
-	require.Regexp(t, "issue30289 build return error", err.Error())
+	require.EqualError(t, err, "issue30289 build return error")
 }
 
 func TestIssue29498(t *testing.T) {

executor/infoschema_cluster_table_test.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"net"
 	"net/http/httptest"
+	"strconv"
 	"strings"
 	"testing"
 	"time"
@@ -329,7 +330,8 @@ func (s *infosSchemaClusterTableSuite) TestTableStorageStats() {
 		"test 2",
 	))
 	rows := tk.MustQuery("select TABLE_NAME from information_schema.TABLE_STORAGE_STATS where TABLE_SCHEMA = 'mysql';").Rows()
-	s.Require().Len(rows, 31)
+	result := 31
+	s.Require().Len(rows, result)
 
 	// More tests about the privileges.
 	tk.MustExec("create user 'testuser'@'localhost'")
@@ -355,12 +357,12 @@ func (s *infosSchemaClusterTableSuite) TestTableStorageStats() {
 		Hostname: "localhost",
 	}, nil, nil))
 
-	tk.MustQuery("select count(1) from information_schema.TABLE_STORAGE_STATS where TABLE_SCHEMA = 'mysql'").Check(testkit.Rows("31"))
+	tk.MustQuery("select count(1) from information_schema.TABLE_STORAGE_STATS where TABLE_SCHEMA = 'mysql'").Check(testkit.Rows(strconv.Itoa(result)))
 
 	s.Require().True(tk.Session().Auth(&auth.UserIdentity{
 		Username: "testuser3",
 		Hostname: "localhost",
 	}, nil, nil))
 
-	tk.MustQuery("select count(1) from information_schema.TABLE_STORAGE_STATS where TABLE_SCHEMA = 'mysql'").Check(testkit.Rows("31"))
+	tk.MustQuery("select count(1) from information_schema.TABLE_STORAGE_STATS where TABLE_SCHEMA = 'mysql'").Check(testkit.Rows(strconv.Itoa(result)))
 }