ddl : Add compatibility testcases

tests/realtikvtest/addindextest/BUILD.bazel

@@ -4,6 +4,7 @@ go_library(
     name = "addindextest",
     srcs = [
         "common.go",
+        "compatibility.go",
         "workload.go",
     ],
     importpath = "github.com/pingcap/tidb/tests/realtikvtest/addindextest",
@@ -19,13 +20,18 @@ go_library(
 
 go_test(
     name = "addindextest_test",
+    timeout = "long",
     srcs = [
         "add_index_test.go",
+        "concurrent_ddl_test.go",
         "main_test.go",
+        "multi_schema_change_test.go",
+        "pitr_test.go",
     ],
     embed = [":addindextest"],
     deps = [
         "//testkit",
         "//tests/realtikvtest",
+        "@com_github_stretchr_testify//require",
     ],
 )

tests/realtikvtest/addindextest/common.go

@@ -45,6 +45,7 @@ type suiteContext struct {
 	rowNum   int
 	workload *workload
 	tkPool   *sync.Pool
+	CompCtx  *CompatibilityContext
 }
 
 func (s *suiteContext) getTestKit() *testkit.TestKit {
@@ -131,7 +132,7 @@ func createTable(tk *testkit.TestKit) {
 func insertRows(tk *testkit.TestKit) {
 	var (
 		insStr string
-		values []string = []string{
+		values = []string{
 			" (1, 1, 1, 1, 1, 1, 1, 1, 1.0, 1.0, 1111.1111, '2001-01-01', '11:11:11', '2001-01-01 11:11:11', '2001-01-01 11:11:11.123456', 1999, 'aaaa', 'aaaa', 'aaaa', 'aaaa', 'aaaa','aaaa', 'aaaa', 'aaaa', 'aaaa', 'aaaa', 'aaaa', 'aaaa', '{\"name\": \"Beijing\", \"population\": 100}')",
 			" (2, 2, 2, 2, 2, 2, 2, 2, 2.0, 2.0, 1112.1111, '2001-01-02', '11:11:12', '2001-01-02 11:11:12', '2001-01-02 11:11:12.123456', 2000, 'bbbb', 'bbbb', 'bbbb', 'bbbb', 'bbbb', 'bbbb', 'bbbb', 'bbbb', 'bbbb', 'bbbb', 'bbbb', 'bbbb', '{\"name\": \"Beijing\", \"population\": 101}')",
 			" (3, 3, 3, 3, 3, 3, 3, 3, 3.0, 3.0, 1113.1111, '2001-01-03', '11:11:13', '2001-01-03 11:11:13', '2001-01-03 11:11:11.123456', 2001, 'cccc', 'cccc', 'cccc', 'cccc', 'cccc', 'cccc', 'cccc', 'cccc', 'cccc', 'cccc', 'cccc', 'cccc', '{\"name\": \"Beijing\", \"population\": 102}')",
@@ -207,7 +208,7 @@ func insertRows(tk *testkit.TestKit) {
 	}
 }
 
-func createIndexOneCol(ctx *suiteContext, tableID int, colID int) error {
+func createIndexOneCol(ctx *suiteContext, tableID int, colID int) (err error) {
 	addIndexStr := " add index idx"
 	var ddlStr string
 	if ctx.isPK {
@@ -232,8 +233,16 @@ func createIndexOneCol(ctx *suiteContext, tableID int, colID int) error {
 			ddlStr = "alter table addindex.t" + strconv.Itoa(tableID) + addIndexStr + strconv.Itoa(colID) + "(c0, c" + strconv.Itoa(colID) + ")"
 		}
 	}
+	if ctx.CompCtx != nil && ctx.CompCtx.isMultiSchemaChange {
+		colID += 60
+		ddlStr += " , add column c" + strconv.Itoa(colID) + " int;"
+	}
 	logutil.BgLogger().Info("[add index test] createIndexOneCol", zap.String("sql", ddlStr))
-	_, err := ctx.tk.Exec(ddlStr)
+	if ctx.CompCtx != nil && ctx.CompCtx.isConcurrentDDL {
+		_, err = ctx.CompCtx.executor[tableID].tk.Exec(ddlStr)
+	} else {
+		_, err = ctx.tk.Exec(ddlStr)
+	}
 	if err != nil {
 		if ctx.isUnique || ctx.isPK {
 			require.Contains(ctx.t, err.Error(), "Duplicate entry")
@@ -244,7 +253,7 @@ func createIndexOneCol(ctx *suiteContext, tableID int, colID int) error {
 	return err
 }
 
-func createIndexTwoCols(ctx *suiteContext, tableID int, indexID int, colID1 int, colID2 int) error {
+func createIndexTwoCols(ctx *suiteContext, tableID int, indexID int, colID1 int, colID2 int) (err error) {
 	var colID1Str, colID2Str string
 	addIndexStr := " add index idx"
 	if ctx.isPK {
@@ -263,8 +272,16 @@ func createIndexTwoCols(ctx *suiteContext, tableID int, indexID int, colID1 int,
 		colID2Str = strconv.Itoa(colID2)
 	}
 	ddlStr := "alter table addindex.t" + strconv.Itoa(tableID) + addIndexStr + strconv.Itoa(indexID) + "(c" + colID1Str + ", c" + colID2Str + ")"
+	if ctx.CompCtx != nil && ctx.CompCtx.isMultiSchemaChange {
+		colID1 += 60
+		ddlStr += " , add column c" + strconv.Itoa(colID1) + " varchar(10);"
+	}
 	logutil.BgLogger().Info("[add index test] createIndexTwoCols", zap.String("sql", ddlStr))
-	_, err := ctx.tk.Exec(ddlStr)
+	if ctx.CompCtx != nil && ctx.CompCtx.isConcurrentDDL {
+		_, err = ctx.CompCtx.executor[tableID].tk.Exec(ddlStr)
+	} else {
+		_, err = ctx.tk.Exec(ddlStr)
+	}
 	if err != nil {
 		logutil.BgLogger().Error("[add index test] add index failed",
 			zap.String("sql", ddlStr), zap.Error(err))
@@ -273,32 +290,53 @@ func createIndexTwoCols(ctx *suiteContext, tableID int, indexID int, colID1 int,
 	return err
 }
 
-func checkResult(ctx *suiteContext, tableName string, indexID int) {
+func checkResult(ctx *suiteContext, tableName string, indexID int, tkID int) {
+	var err error
 	adminCheckSQL := "admin check index " + tableName + " idx" + strconv.Itoa(indexID)
-	_, err := ctx.tk.Exec(adminCheckSQL)
+	if ctx.CompCtx != nil && ctx.CompCtx.isConcurrentDDL {
+		_, err = ctx.CompCtx.executor[tkID].tk.Exec(adminCheckSQL)
+	} else {
+		_, err = ctx.tk.Exec(adminCheckSQL)
+	}
 	if err != nil {
 		logutil.BgLogger().Error("[add index test] checkResult",
 			zap.String("sql", adminCheckSQL), zap.Error(err))
 	}
 	require.NoError(ctx.t, err)
-	require.Equal(ctx.t, ctx.tk.Session().AffectedRows(), uint64(0))
-	_, err = ctx.tk.Exec("alter table " + tableName + " drop index idx" + strconv.Itoa(indexID))
+
+	if ctx.CompCtx != nil && ctx.CompCtx.isConcurrentDDL {
+		require.Equal(ctx.t, uint64(0), ctx.CompCtx.executor[tkID].tk.Session().AffectedRows())
+		_, err = ctx.CompCtx.executor[tkID].tk.Exec("alter table " + tableName + " drop index idx" + strconv.Itoa(indexID))
+	} else {
+		require.Equal(ctx.t, uint64(0), ctx.tk.Session().AffectedRows())
+		_, err = ctx.tk.Exec("alter table " + tableName + " drop index idx" + strconv.Itoa(indexID))
+	}
+
 	if err != nil {
 		logutil.BgLogger().Error("[add index test] drop index failed",
 			zap.String("sql", adminCheckSQL), zap.Error(err))
 	}
 	require.NoError(ctx.t, err)
 }
 
-func checkTableResult(ctx *suiteContext, tableName string) {
+func checkTableResult(ctx *suiteContext, tableName string, tkID int) {
+	var err error
 	adminCheckSQL := "admin check table " + tableName
-	_, err := ctx.tk.Exec(adminCheckSQL)
+	if ctx.CompCtx != nil && ctx.CompCtx.isConcurrentDDL {
+		_, err = ctx.CompCtx.executor[tkID].tk.Exec(adminCheckSQL)
+	} else {
+		_, err = ctx.tk.Exec(adminCheckSQL)
+	}
 	if err != nil {
 		logutil.BgLogger().Error("[add index test] checkTableResult",
 			zap.String("sql", adminCheckSQL), zap.Error(err))
 	}
 	require.NoError(ctx.t, err)
-	require.Equal(ctx.t, ctx.tk.Session().AffectedRows(), uint64(0))
+	if ctx.CompCtx != nil && ctx.CompCtx.isConcurrentDDL {
+		require.Equal(ctx.t, uint64(0), ctx.CompCtx.executor[tkID].tk.Session().AffectedRows())
+	} else {
+		require.Equal(ctx.t, uint64(0), ctx.tk.Session().AffectedRows())
+	}
 }
 
 func testOneColFrame(ctx *suiteContext, colIDs [][]int, f func(*suiteContext, int, string, int) error) {
@@ -318,10 +356,10 @@ func testOneColFrame(ctx *suiteContext, colIDs [][]int, f func(*suiteContext, in
 				}
 			}
 			if ctx.workload != nil {
-				_ = ctx.workload.stop(ctx)
+				_ = ctx.workload.stop(ctx, -1)
 			}
 			if err == nil {
-				checkResult(ctx, tableName, i)
+				checkResult(ctx, tableName, i, tableID)
 			}
 		}
 	}
@@ -343,10 +381,10 @@ func testTwoColsFrame(ctx *suiteContext, iIDs [][]int, jIDs [][]int, f func(*sui
 				require.NoError(ctx.t, err)
 				if ctx.workload != nil {
 					// Stop workload
-					_ = ctx.workload.stop(ctx)
+					_ = ctx.workload.stop(ctx, -1)
 				}
 				if err == nil && i != j {
-					checkResult(ctx, tableName, indexID)
+					checkResult(ctx, tableName, indexID, tableID)
 				}
 				indexID++
 			}
@@ -366,13 +404,13 @@ func testOneIndexFrame(ctx *suiteContext, colID int, f func(*suiteContext, int,
 		}
 		require.NoError(ctx.t, err)
 		if ctx.workload != nil {
-			_ = ctx.workload.stop(ctx)
+			_ = ctx.workload.stop(ctx, -1)
 		}
 		if err == nil {
 			if ctx.isPK {
-				checkTableResult(ctx, tableName)
+				checkTableResult(ctx, tableName, tableID)
 			} else {
-				checkResult(ctx, tableName, colID)
+				checkResult(ctx, tableName, colID, tableID)
 			}
 		}
 	}

tests/realtikvtest/addindextest/compatibility.go

@@ -0,0 +1,189 @@
+// Copyright 2022 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package addindextest
+
+import (
+	"strconv"
+
+	"github.com/pingcap/tidb/testkit"
+	"github.com/pingcap/tidb/util/logutil"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+var compCtx = CompatibilityContext{}
+
+type testType int8
+
+const (
+	// TestNonUnique test type of create none unique index.
+	TestNonUnique testType = iota
+	// TestUnique test type of create unique index.
+	TestUnique
+	// TestPK test type of create Primary key.
+	TestPK
+	// TestGenIndex test type of create generated col index.
+	TestGenIndex
+	// TestMultiCols test type of multi columns in one index.
+	TestMultiCols
+)
+
+// CompatibilityContext is context of compatibility test.
+type CompatibilityContext struct {
+	isMultiSchemaChange bool
+	isConcurrentDDL     bool
+	isPiTR              bool
+	executor            []*executor
+	colIIDs             [][]int
+	colJIDs             [][]int
+	tType               testType
+}
+
+type paraDDLChan struct {
+	err      error
+	finished bool
+}
+
+type executor struct {
+	id     int
+	tk     *testkit.TestKit
+	PDChan chan *paraDDLChan
+}
+
+func newExecutor(tableID int) *executor {
+	er := executor{
+		id:     tableID,
+		PDChan: make(chan *paraDDLChan, 1),
+	}
+	return &er
+}
+
+func initCompCtxParams(ctx *suiteContext) {
+	ctx.CompCtx = &compCtx
+	compCtx.isConcurrentDDL = false
+	compCtx.isMultiSchemaChange = false
+	compCtx.isPiTR = false
+}
+
+func (cCtx *CompatibilityContext) start(ctx *suiteContext) {
+	cCtx.executor = cCtx.executor[:0]
+	for i := 0; i < 3; i++ {
+		er := newExecutor(i)
+		er.tk = ctx.getTestKit()
+		cCtx.executor = append(cCtx.executor, er)
+		go cCtx.executor[i].run(ctx)
+	}
+}
+
+func (cCtx *CompatibilityContext) stop(ctx *suiteContext) error {
+	count := 3
+	for i := 0; i < 3; i++ {
+		pdChan := <-cCtx.executor[i].PDChan
+		if pdChan.err != nil {
+			require.NoError(ctx.t, pdChan.err)
+			return pdChan.err
+		}
+		if pdChan.finished {
+			count--
+			logutil.BgLogger().Info("xlc test worker", zap.Int("count", count), zap.Int("er id", i))
+			ctx.putTestKit(ctx.CompCtx.executor[i].tk)
+		}
+		if count == 0 {
+			break
+		}
+	}
+	return nil
+}
+
+func (e *executor) run(ctx *suiteContext) {
+	var (
+		err    error
+		erChan paraDDLChan
+	)
+	switch ctx.CompCtx.tType {
+	case TestNonUnique:
+		err = testOneColFramePara(ctx, e.id, ctx.CompCtx.colIIDs, addIndexNonUnique)
+	case TestUnique:
+		err = testOneColFramePara(ctx, e.id, ctx.CompCtx.colIIDs, addIndexUnique)
+	case TestPK:
+		err = testOneIndexFramePara(ctx, e.id, 0, addIndexPK)
+	case TestGenIndex:
+		err = testOneIndexFramePara(ctx, e.id, 29, addIndexGenCol)
+	case TestMultiCols:
+		err = testTwoColsFramePara(ctx, e.id, ctx.CompCtx.colIIDs, ctx.CompCtx.colJIDs, addIndexMultiCols)
+	default:
+	}
+	erChan.err = err
+	erChan.finished = true
+	e.PDChan <- &erChan
+}
+
+func testOneColFramePara(ctx *suiteContext, tableID int, colIDs [][]int, f func(*suiteContext, int, string, int) error) (err error) {
+	tableName := "addindex.t" + strconv.Itoa(tableID)
+	for _, i := range colIDs[tableID] {
+		err = f(ctx, tableID, tableName, i)
+		if err != nil {
+			if ctx.isUnique || ctx.isPK {
+				require.Contains(ctx.t, err.Error(), "Duplicate entry")
+				err = nil
+				continue
+			}
+			logutil.BgLogger().Error("[add index test] add index failed", zap.Error(err))
+			require.NoError(ctx.t, err)
+			break
+		}
+		checkResult(ctx, tableName, i, tableID)
+	}
+	return err
+}
+
+func testTwoColsFramePara(ctx *suiteContext, tableID int, iIDs [][]int, jIDs [][]int, f func(*suiteContext, int, string, int, int, int) error) (err error) {
+	tableName := "addindex.t" + strconv.Itoa(tableID)
+	indexID := 0
+	for _, i := range iIDs[tableID] {
+		for _, j := range jIDs[tableID] {
+			err = f(ctx, tableID, tableName, indexID, i, j)
+			if err != nil {
+				logutil.BgLogger().Error("[add index test] add index failed", zap.Error(err))
+			}
+			require.NoError(ctx.t, err)
+			if err == nil && i != j {
+				checkResult(ctx, tableName, indexID, tableID)
+			}
+			indexID++
+			if err != nil {
+				return err
+			}
+		}
+	}
+	return err
+}
+
+func testOneIndexFramePara(ctx *suiteContext, tableID int, colID int, f func(*suiteContext, int, string, int) error) (err error) {
+	tableName := "addindex.t" + strconv.Itoa(tableID)
+	err = f(ctx, tableID, tableName, colID)
+	if err != nil {
+		logutil.BgLogger().Error("[add index test] add index failed", zap.Error(err))
+	}
+	require.NoError(ctx.t, err)
+	if err == nil {
+		if ctx.isPK {
+			checkTableResult(ctx, tableName, tableID)
+		} else {
+			checkResult(ctx, tableName, colID, tableID)
+		}
+	}
+	return err
+}