ddl: avoid DDL retry taking too long (#19328)

ddl/ddl_worker.go

@@ -481,7 +481,14 @@ func (w *worker) handleDDLJobQueue(d *ddlCtx) error {
 				// then shouldn't discard the KV modification.
 				// And the job state is rollback done, it means the job was already finished, also shouldn't discard too.
 				// Otherwise, we should discard the KV modification when running job.
+<<<<<<< HEAD
 				txn.Discard()
+=======
+				txn.Reset()
+				// If error happens after updateSchemaVersion(), then the schemaVer is updated.
+				// Result in the retry duration is up to 2 * lease.
+				schemaVer = 0
+>>>>>>> 33f3843... ddl: avoid DDL retry taking too long (#19328)
 			}
 			err = w.updateDDLJob(t, job, runJobErr != nil)
 			if err = w.handleUpdateJobError(t, job, err); err != nil {

ddl/mock.go

@@ -19,6 +19,7 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/parser/ast"
 	"github.com/pingcap/parser/charset"
 	"github.com/pingcap/parser/model"
@@ -105,6 +106,11 @@ func (s *MockSchemaSyncer) OwnerCheckAllVersions(ctx context.Context, latestVer
 	for {
 		select {
 		case <-ctx.Done():
+			failpoint.Inject("checkOwnerCheckAllVersionsWaitTime", func(v failpoint.Value) {
+				if v.(bool) {
+					panic("shouldn't happen")
+				}
+			})
 			return errors.Trace(ctx.Err())
 		case <-ticker.C:
 			ver := atomic.LoadInt64(&s.selfSchemaVersion)

ddl/serial_test.go

@@ -1146,3 +1146,159 @@ func (s *testSerialSuite) TestForbidUnsupportedCollations(c *C) {
 	// mustGetUnsupportedCollation("create database ucd collate utf8mb4_unicode_ci", errMsgUnsupportedUnicodeCI)
 	// mustGetUnsupportedCollation("alter table t convert to collate utf8mb4_unicode_ci", "utf8mb4_unicode_ci")
 }
+<<<<<<< HEAD
+=======
+
+func (s *testSerialSuite) TestInvisibleIndex(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t,t1,t2,t3,t4,t5,t6")
+
+	// The DDL statement related to invisible index.
+	showIndexes := "select index_name, is_visible from information_schema.statistics where table_schema = 'test' and table_name = 't'"
+	// 1. Create table with invisible index
+	tk.MustExec("create table t (a int, b int, unique (a) invisible)")
+	tk.MustQuery(showIndexes).Check(testkit.Rows("a NO"))
+	tk.MustExec("insert into t values (1, 2)")
+	tk.MustQuery("select * from t").Check(testkit.Rows("1 2"))
+	// 2. Drop invisible index
+	tk.MustExec("alter table t drop index a")
+	tk.MustQuery(showIndexes).Check(testkit.Rows())
+	tk.MustExec("insert into t values (3, 4)")
+	tk.MustQuery("select * from t").Check(testkit.Rows("1 2", "3 4"))
+	// 3. Add an invisible index
+	tk.MustExec("alter table t add index (b) invisible")
+	tk.MustQuery(showIndexes).Check(testkit.Rows("b NO"))
+	tk.MustExec("insert into t values (5, 6)")
+	tk.MustQuery("select * from t").Check(testkit.Rows("1 2", "3 4", "5 6"))
+	// 4. Drop it
+	tk.MustExec("alter table t drop index b")
+	tk.MustQuery(showIndexes).Check(testkit.Rows())
+	tk.MustExec("insert into t values (7, 8)")
+	tk.MustQuery("select * from t").Check(testkit.Rows("1 2", "3 4", "5 6", "7 8"))
+	// 5. Create a multiple-column invisible index
+	tk.MustExec("alter table t add index a_b(a, b) invisible")
+	tk.MustQuery(showIndexes).Check(testkit.Rows("a_b NO", "a_b NO"))
+	tk.MustExec("insert into t values (9, 10)")
+	tk.MustQuery("select * from t").Check(testkit.Rows("1 2", "3 4", "5 6", "7 8", "9 10"))
+	// 6. Drop it
+	tk.MustExec("alter table t drop index a_b")
+	tk.MustQuery(showIndexes).Check(testkit.Rows())
+	tk.MustExec("insert into t values (11, 12)")
+	tk.MustQuery("select * from t").Check(testkit.Rows("1 2", "3 4", "5 6", "7 8", "9 10", "11 12"))
+
+	defer config.RestoreFunc()()
+	config.UpdateGlobal(func(conf *config.Config) {
+		conf.AlterPrimaryKey = true
+	})
+
+	// Limitation: Primary key cannot be invisible index
+	tk.MustGetErrCode("create table t1 (a int, primary key (a) invisible)", errno.ErrPKIndexCantBeInvisible)
+	tk.MustGetErrCode("create table t1 (a int, b int, primary key (a, b) invisible)", errno.ErrPKIndexCantBeInvisible)
+	tk.MustExec("create table t1 (a int, b int)")
+	tk.MustGetErrCode("alter table t1 add primary key(a) invisible", errno.ErrPKIndexCantBeInvisible)
+	tk.MustGetErrCode("alter table t1 add primary key(a, b) invisible", errno.ErrPKIndexCantBeInvisible)
+
+	// Implicit primary key cannot be invisible index
+	// Create a implicit primary key
+	tk.MustGetErrCode("create table t2(a int not null, unique (a) invisible)", errno.ErrPKIndexCantBeInvisible)
+	// Column `a` become implicit primary key after DDL statement on itself
+	tk.MustExec("create table t2(a int not null)")
+	tk.MustGetErrCode("alter table t2 add unique (a) invisible", errno.ErrPKIndexCantBeInvisible)
+	tk.MustExec("create table t3(a int, unique index (a) invisible)")
+	tk.MustGetErrCode("alter table t3 modify column a int not null", errno.ErrPKIndexCantBeInvisible)
+	// Only first unique column can be implicit primary
+	tk.MustExec("create table t4(a int not null, b int not null, unique (a), unique (b) invisible)")
+	showIndexes = "select index_name, is_visible from information_schema.statistics where table_schema = 'test' and table_name = 't4'"
+	tk.MustQuery(showIndexes).Check(testkit.Rows("a YES", "b NO"))
+	tk.MustExec("insert into t4 values (1, 2)")
+	tk.MustQuery("select * from t4").Check(testkit.Rows("1 2"))
+	tk.MustGetErrCode("create table t5(a int not null, b int not null, unique (b) invisible, unique (a))", errno.ErrPKIndexCantBeInvisible)
+	// Column `b` become implicit primary key after DDL statement on other columns
+	tk.MustExec("create table t5(a int not null, b int not null, unique (a), unique (b) invisible)")
+	tk.MustGetErrCode("alter table t5 drop index a", errno.ErrPKIndexCantBeInvisible)
+	tk.MustGetErrCode("alter table t5 modify column a int null", errno.ErrPKIndexCantBeInvisible)
+	// If these is a explicit primary key, no key will become implicit primary key
+	tk.MustExec("create table t6 (a int not null, b int, unique (a) invisible, primary key(b))")
+	showIndexes = "select index_name, is_visible from information_schema.statistics where table_schema = 'test' and table_name = 't6'"
+	tk.MustQuery(showIndexes).Check(testkit.Rows("a NO", "PRIMARY YES"))
+	tk.MustExec("insert into t6 values (1, 2)")
+	tk.MustQuery("select * from t6").Check(testkit.Rows("1 2"))
+	tk.MustGetErrCode("alter table t6 drop primary key", errno.ErrPKIndexCantBeInvisible)
+}
+
+func (s *testSerialSuite) TestCreateClusteredIndex(c *C) {
+	tk := testkit.NewTestKitWithInit(c, s.store)
+	tk.Se.GetSessionVars().EnableClusteredIndex = true
+	tk.MustExec("CREATE TABLE t1 (a int primary key, b int)")
+	tk.MustExec("CREATE TABLE t2 (a varchar(255) primary key, b int)")
+	tk.MustExec("CREATE TABLE t3 (a int, b int, c int, primary key (a, b))")
+	tk.MustExec("CREATE TABLE t4 (a int, b int, c int)")
+	ctx := tk.Se.(sessionctx.Context)
+	is := domain.GetDomain(ctx).InfoSchema()
+	tbl, err := is.TableByName(model.NewCIStr("test"), model.NewCIStr("t1"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().PKIsHandle, IsTrue)
+	c.Assert(tbl.Meta().IsCommonHandle, IsFalse)
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t2"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().IsCommonHandle, IsTrue)
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t3"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().IsCommonHandle, IsTrue)
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t4"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().IsCommonHandle, IsFalse)
+
+	config.UpdateGlobal(func(conf *config.Config) {
+		conf.AlterPrimaryKey = true
+	})
+	tk.MustExec("CREATE TABLE t5 (a varchar(255) primary key, b int)")
+	tk.MustExec("CREATE TABLE t6 (a int, b int, c int, primary key (a, b))")
+	is = domain.GetDomain(ctx).InfoSchema()
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t5"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().IsCommonHandle, IsFalse)
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t6"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().IsCommonHandle, IsFalse)
+	config.UpdateGlobal(func(conf *config.Config) {
+		conf.AlterPrimaryKey = false
+	})
+
+	tk.MustExec("CREATE TABLE t21 like t2")
+	tk.MustExec("CREATE TABLE t31 like t3")
+	is = domain.GetDomain(ctx).InfoSchema()
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t21"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().IsCommonHandle, IsTrue)
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t31"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().IsCommonHandle, IsTrue)
+
+	tk.Se.GetSessionVars().EnableClusteredIndex = false
+	tk.MustExec("CREATE TABLE t7 (a varchar(255) primary key, b int)")
+	is = domain.GetDomain(ctx).InfoSchema()
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t7"))
+	c.Assert(err, IsNil)
+	c.Assert(tbl.Meta().IsCommonHandle, IsFalse)
+}
+
+func (s *testSerialSuite) TestCreateTableNoBlock(c *C) {
+	tk := testkit.NewTestKitWithInit(c, s.store)
+	c.Assert(failpoint.Enable("github.com/pingcap/tidb/ddl/checkOwnerCheckAllVersionsWaitTime", `return(true)`), IsNil)
+	defer func() {
+		c.Assert(failpoint.Disable("github.com/pingcap/tidb/ddl/checkOwnerCheckAllVersionsWaitTime"), IsNil)
+	}()
+	save := variable.GetDDLErrorCountLimit()
+	variable.SetDDLErrorCountLimit(1)
+	defer func() {
+		variable.SetDDLErrorCountLimit(save)
+	}()
+
+	tk.MustExec("drop table if exists t")
+	_, err := tk.Exec("create table t(a int)")
+	c.Assert(err, NotNil)
+}
+>>>>>>> 33f3843... ddl: avoid DDL retry taking too long (#19328)

ddl/table.go

@@ -75,6 +75,13 @@ func onCreateTable(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error)
 		if err != nil {
 			return ver, errors.Trace(err)
 		}
+
+		failpoint.Inject("checkOwnerCheckAllVersionsWaitTime", func(val failpoint.Value) {
+			if val.(bool) {
+				failpoint.Return(ver, errors.New("mock create table error"))
+			}
+		})
+
 		// Finish this job.
 		job.FinishTableJob(model.JobStateDone, model.StatePublic, ver, tbInfo)
 		asyncNotifyEvent(d, &util.Event{Tp: model.ActionCreateTable, TableInfo: tbInfo})