ddl: add check table compatibility for temporary table (#24501)

pingcap · May 12, 2021 · 89c0e6c · 89c0e6c
1 parent 6a69ee7
commit 89c0e6c
Show file tree

Hide file tree

Showing 6 changed files with 58 additions and 2 deletions.
diff --git a/errors.toml b/errors.toml
@@ -1281,6 +1281,11 @@ error = '''
 Unknown SEQUENCE: '%-.300s'
 '''
 
+["schema:8003"]
+error = '''
+TiDB admin check table failed.
+'''
+
 ["schema:8020"]
 error = '''
 Table '%s' was locked in %s by %v

diff --git a/executor/admin_test.go b/executor/admin_test.go
@@ -76,6 +76,22 @@ func (s *testSuite5) TestAdminCheckIndex(c *C) {
 	check()
 }
 
+func (s *testSuite5) TestAdminCheckIndexInTemporaryMode(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists temporary_admin_test;")
+	tk.MustExec("create global temporary table temporary_admin_test (c1 int, c2 int, c3 int default 1, primary key (c1), index (c1), unique key(c2)) ON COMMIT DELETE ROWS;")
+	tk.MustExec("insert temporary_admin_test (c1, c2) values (1, 1), (2, 2), (3, 3);")
+	tk.MustGetErrCode("admin check table temporary_admin_test;", mysql.ErrAdminCheckTable)
+	tk.MustExec("drop table if exists temporary_admin_test;")
+
+	tk.MustExec("drop table if exists non_temporary_admin_test;")
+	tk.MustExec("create table non_temporary_admin_test (c1 int, c2 int, c3 int default 1, primary key (c1), index (c1), unique key(c2));")
+	tk.MustExec("insert non_temporary_admin_test (c1, c2) values (1, 1), (2, 2), (3, 3);")
+	tk.MustExec("admin check table non_temporary_admin_test;")
+	tk.MustExec("drop table if exists non_temporary_admin_test;")
+}
+
 func (s *testSuite5) TestAdminRecoverIndex(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")

diff --git a/executor/ddl.go b/executor/ddl.go
@@ -311,8 +311,12 @@ func (e *DDLExec) dropTableObject(objects []*ast.TableName, obt objectType, ifEx
 		if isSystemTable(tn.Schema.L, tn.Name.L) {
 			return errors.Errorf("Drop tidb system table '%s.%s' is forbidden", tn.Schema.L, tn.Name.L)
 		}
-
-		if obt == tableObject && config.CheckTableBeforeDrop {
+		tableInfo, err := e.is.TableByName(tn.Schema, tn.Name)
+		if err != nil {
+			return err
+		}
+		tempTableType := tableInfo.Meta().TempTableType
+		if obt == tableObject && config.CheckTableBeforeDrop && tempTableType == model.TempTableNone {
 			logutil.BgLogger().Warn("admin check table before drop",
 				zap.String("database", fullti.Schema.O),
 				zap.String("table", fullti.Name.O),

diff --git a/expression/integration_test.go b/expression/integration_test.go
@@ -9161,8 +9161,10 @@ func (s *testIntegrationSuite) TestIssue24429(c *C) {
 
 	tk.MustExec("set @@sql_mode = ANSI_QUOTES;")
 	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t;")
 	tk.MustExec("create table t (a int);")
 	tk.MustQuery(`select t."a"=10 from t;`).Check(testkit.Rows())
+	tk.MustExec("drop table if exists t;")
 }
 
 func (s *testIntegrationSuite) TestVitessHash(c *C) {

diff --git a/infoschema/error.go b/infoschema/error.go
@@ -69,4 +69,6 @@ var (
 	ErrTableLocked = dbterror.ClassSchema.NewStd(mysql.ErrTableLocked)
 	// ErrWrongObject returns when the table/view/sequence is not the expected object.
 	ErrWrongObject = dbterror.ClassSchema.NewStd(mysql.ErrWrongObject)
+	// ErrAdminCheckTable returns when the check table in temporary mode.
+	ErrAdminCheckTable = dbterror.ClassSchema.NewStd(mysql.ErrAdminCheckTable)
 )
diff --git a/planner/core/preprocess.go b/planner/core/preprocess.go
@@ -125,6 +125,8 @@ type preprocessor struct {
 
 func (p *preprocessor) Enter(in ast.Node) (out ast.Node, skipChildren bool) {
 	switch node := in.(type) {
+	case *ast.AdminStmt:
+		p.checkAdminCheckTableGrammar(node)
 	case *ast.DeleteStmt:
 		p.stmtTp = TypeDelete
 	case *ast.SelectStmt:
@@ -557,6 +559,31 @@ func (p *preprocessor) checkDropDatabaseGrammar(stmt *ast.DropDatabaseStmt) {
 	}
 }
 
+func (p *preprocessor) checkAdminCheckTableGrammar(stmt *ast.AdminStmt) {
+	for _, table := range stmt.Tables {
+		currentDB := p.ctx.GetSessionVars().CurrentDB
+		if table.Schema.String() != "" {
+			currentDB = table.Schema.L
+		}
+		if currentDB == "" {
+			p.err = errors.Trace(ErrNoDB)
+			return
+		}
+		sName := model.NewCIStr(currentDB)
+		tName := table.Name
+		tableInfo, err := p.is.TableByName(sName, tName)
+		if err != nil {
+			p.err = err
+			return
+		}
+		tempTableType := tableInfo.Meta().TempTableType
+		if stmt.Tp == ast.AdminCheckTable && tempTableType != model.TempTableNone {
+			p.err = infoschema.ErrAdminCheckTable
+			return
+		}
+	}
+}
+
 func (p *preprocessor) checkCreateTableGrammar(stmt *ast.CreateTableStmt) {
 	tName := stmt.Table.Name.String()
 	if isIncorrectName(tName) {