tracker: fetch ref table at CREATE LIKE and more (#1105) (#1108)

Co-authored-by: lance6716 <lance6716@gmail.com>

_utils/terror_gen/errors_release.txt

@@ -122,6 +122,7 @@ ErrInvalidV1WorkerMetaPath,[code=11115:class=functional:scope=internal:level=med
 ErrFailUpdateV1DBSchema,[code=11116:class=functional:scope=internal:level=medium], "Message: fail to upgrade v1.0.x DB schema, Workaround: Please confirm that you have not violated any restrictions in the upgrade documentation."
 ErrBinlogStatusVarsParse,[code=11117:class=functional:scope=internal:level=medium], "Message: fail to parse binglog status_vars: %v, offset: %d"
 ErrVerifyHandleErrorArgs,[code=11118:class=functional:scope=internal:level=low], "Workaround: Please make sure the args are correct."
+ErrRewriteSQL,[code=11119:class=functional:scope=internal:level=high], "Message: failed to rewrite SQL for target DB, stmt: %+v, targetTableNames: %+v"
 ErrConfigCheckItemNotSupport,[code=20001:class=config:scope=internal:level=medium], "Message: checking item %s is not supported\n%s, Workaround: Please check `ignore-checking-items` config in task configuration file, which can be set including `all`/`dump_privilege`/`replication_privilege`/`version`/`binlog_enable`/`binlog_format`/`binlog_row_image`/`table_schema`/`schema_of_shard_tables`/`auto_increment_ID`."
 ErrConfigTomlTransform,[code=20002:class=config:scope=internal:level=medium], "Message: %s, Workaround: Please check the configuration file has correct TOML format."
 ErrConfigYamlTransform,[code=20003:class=config:scope=internal:level=medium], "Message: %s, Workaround: Please check the configuration file has correct YAML format."

errors.toml

@@ -742,6 +742,12 @@ description = ""
 workaround = "Please make sure the args are correct."
 tags = ["internal", "low"]
 
+[error.DM-functional-11119]
+message = "failed to rewrite SQL for target DB, stmt: %+v, targetTableNames: %+v"
+description = ""
+workaround = ""
+tags = ["internal", "high"]
+
 [error.DM-config-20001]
 message = "checking item %s is not supported\n%s"
 description = ""

pkg/parser/common.go

@@ -28,6 +28,13 @@ import (
 	"go.uber.org/zap"
 )
 
+const (
+	// SingleRenameTableNameNum stands for number of TableNames in a single table renaming. it's 4 not 2 because TiDB
+	// parser repeats first two TableNames at start
+	// ref https://github.com/pingcap/tidb/pull/3892
+	SingleRenameTableNameNum = 4
+)
+
 // Parse wraps parser.Parse(), makes `parser` suitable for dm
 func Parse(p *parser.Parser, sql, charset, collation string) (stmt []ast.StmtNode, err error) {
 	stmts, warnings, err := p.Parse(sql, charset, collation)
@@ -42,122 +49,114 @@ func Parse(p *parser.Parser, sql, charset, collation string) (stmt []ast.StmtNod
 	return stmts, terror.ErrParseSQL.Delegate(err)
 }
 
-// FetchDDLTableNames returns table names in ddl
-// the result contains [tableName] excepted create table like and rename table
-// for `create table like` DDL, result contains [sourceTableName, sourceRefTableName]
-// for rename table ddl, result contains [oldTableName, newTableName]
+// ref: https://github.com/pingcap/tidb/blob/09feccb529be2830944e11f5fed474020f50370f/server/sql_info_fetcher.go#L46
+type tableNameExtractor struct {
+	curDB string
+	names []*filter.Table
+}
+
+func (tne *tableNameExtractor) Enter(in ast.Node) (ast.Node, bool) {
+	if t, ok := in.(*ast.TableName); ok {
+		tb := &filter.Table{Schema: t.Schema.L, Name: t.Name.L}
+		if tb.Schema == "" {
+			tb.Schema = tne.curDB
+		}
+		tne.names = append(tne.names, tb)
+		return in, true
+	}
+	return in, false
+}
+
+func (tne *tableNameExtractor) Leave(in ast.Node) (ast.Node, bool) {
+	return in, true
+}
+
+// FetchDDLTableNames returns tableNames in ddl the result contains many tableName.
+// Because we use visitor pattern, first tableName is always upper-most table in ast
+// specifically, for `create table like` DDL, result contains [sourceTableName, sourceRefTableName]
+// for rename table ddl, result contains [old1, new1, old1, new1, old2, new2, old3, new3, ...] because of TiDB parser
+// for other DDL, order of tableName is the node visit order
 func FetchDDLTableNames(schema string, stmt ast.StmtNode) ([]*filter.Table, error) {
-	var res []*filter.Table
+	switch stmt.(type) {
+	case ast.DDLNode:
+	default:
+		return nil, terror.ErrUnknownTypeDDL.Generate(stmt)
+	}
+
+	// special cases: schema related SQLs doesn't have tableName
 	switch v := stmt.(type) {
 	case *ast.AlterDatabaseStmt:
-		res = append(res, genTableName(v.Name, ""))
+		return []*filter.Table{genTableName(v.Name, "")}, nil
 	case *ast.CreateDatabaseStmt:
-		res = append(res, genTableName(v.Name, ""))
+		return []*filter.Table{genTableName(v.Name, "")}, nil
 	case *ast.DropDatabaseStmt:
-		res = append(res, genTableName(v.Name, ""))
-	case *ast.CreateTableStmt:
-		res = append(res, genTableName(v.Table.Schema.O, v.Table.Name.O))
-		if v.ReferTable != nil {
-			res = append(res, genTableName(v.ReferTable.Schema.O, v.ReferTable.Name.O))
-		}
-	case *ast.DropTableStmt:
-		if len(v.Tables) != 1 {
-			return res, terror.ErrDropMultipleTables.Generate()
-		}
-		res = append(res, genTableName(v.Tables[0].Schema.O, v.Tables[0].Name.O))
-	case *ast.TruncateTableStmt:
-		res = append(res, genTableName(v.Table.Schema.O, v.Table.Name.O))
-	case *ast.AlterTableStmt:
-		res = append(res, genTableName(v.Table.Schema.O, v.Table.Name.O))
-		if v.Specs[0].NewTable != nil {
-			res = append(res, genTableName(v.Specs[0].NewTable.Schema.O, v.Specs[0].NewTable.Name.O))
-		}
-	case *ast.RenameTableStmt:
-		res = append(res, genTableName(v.OldTable.Schema.O, v.OldTable.Name.O))
-		res = append(res, genTableName(v.NewTable.Schema.O, v.NewTable.Name.O))
-	case *ast.CreateIndexStmt:
-		res = append(res, genTableName(v.Table.Schema.O, v.Table.Name.O))
-	case *ast.DropIndexStmt:
-		res = append(res, genTableName(v.Table.Schema.O, v.Table.Name.O))
-	default:
-		return res, terror.ErrUnknownTypeDDL.Generate(stmt)
+		return []*filter.Table{genTableName(v.Name, "")}, nil
+	}
+
+	e := &tableNameExtractor{
+		curDB: schema,
+		names: make([]*filter.Table, 0),
 	}
+	stmt.Accept(e)
+
+	return e.names, nil
+}
+
+type tableRenameVisitor struct {
+	targetNames []*filter.Table
+	i           int
+	hasErr      bool
+}
 
-	for i := range res {
-		if res[i].Schema == "" {
-			res[i].Schema = schema
+func (v *tableRenameVisitor) Enter(in ast.Node) (ast.Node, bool) {
+	if v.hasErr {
+		return in, true
+	}
+	if t, ok := in.(*ast.TableName); ok {
+		if v.i >= len(v.targetNames) {
+			v.hasErr = true
+			return in, true
 		}
+		t.Schema = model.NewCIStr(v.targetNames[v.i].Schema)
+		t.Name = model.NewCIStr(v.targetNames[v.i].Name)
+		v.i++
+		return in, true
 	}
+	return in, false
+}
 
-	return res, nil
+func (v *tableRenameVisitor) Leave(in ast.Node) (ast.Node, bool) {
+	if v.hasErr {
+		return in, false
+	}
+	return in, true
 }
 
 // RenameDDLTable renames table names in ddl by given `targetTableNames`
 // argument `targetTableNames` is same with return value of FetchDDLTableNames
 // returned DDL is formatted like StringSingleQuotes, KeyWordUppercase and NameBackQuotes
 func RenameDDLTable(stmt ast.StmtNode, targetTableNames []*filter.Table) (string, error) {
+	switch stmt.(type) {
+	case ast.DDLNode:
+	default:
+		return "", terror.ErrUnknownTypeDDL.Generate(stmt)
+	}
+
 	switch v := stmt.(type) {
 	case *ast.AlterDatabaseStmt:
 		v.Name = targetTableNames[0].Schema
-
 	case *ast.CreateDatabaseStmt:
 		v.Name = targetTableNames[0].Schema
-
 	case *ast.DropDatabaseStmt:
 		v.Name = targetTableNames[0].Schema
-
-	case *ast.CreateTableStmt:
-		v.Table.Schema = model.NewCIStr(targetTableNames[0].Schema)
-		v.Table.Name = model.NewCIStr(targetTableNames[0].Name)
-
-		if v.ReferTable != nil {
-			v.ReferTable.Schema = model.NewCIStr(targetTableNames[1].Schema)
-			v.ReferTable.Name = model.NewCIStr(targetTableNames[1].Name)
-		}
-
-	case *ast.DropTableStmt:
-		if len(v.Tables) > 1 {
-			return "", terror.ErrDropMultipleTables.Generate()
-		}
-
-		v.Tables[0].Schema = model.NewCIStr(targetTableNames[0].Schema)
-		v.Tables[0].Name = model.NewCIStr(targetTableNames[0].Name)
-
-	case *ast.TruncateTableStmt:
-		v.Table.Schema = model.NewCIStr(targetTableNames[0].Schema)
-		v.Table.Name = model.NewCIStr(targetTableNames[0].Name)
-
-	case *ast.DropIndexStmt:
-		v.Table.Schema = model.NewCIStr(targetTableNames[0].Schema)
-		v.Table.Name = model.NewCIStr(targetTableNames[0].Name)
-	case *ast.CreateIndexStmt:
-		v.Table.Schema = model.NewCIStr(targetTableNames[0].Schema)
-		v.Table.Name = model.NewCIStr(targetTableNames[0].Name)
-	case *ast.RenameTableStmt:
-		if len(v.TableToTables) > 1 {
-			return "", terror.ErrRenameMultipleTables.Generate()
-		}
-
-		v.TableToTables[0].OldTable.Schema = model.NewCIStr(targetTableNames[0].Schema)
-		v.TableToTables[0].OldTable.Name = model.NewCIStr(targetTableNames[0].Name)
-		v.TableToTables[0].NewTable.Schema = model.NewCIStr(targetTableNames[1].Schema)
-		v.TableToTables[0].NewTable.Name = model.NewCIStr(targetTableNames[1].Name)
-
-	case *ast.AlterTableStmt:
-		if len(v.Specs) > 1 {
-			return "", terror.ErrAlterMultipleTables.Generate()
+	default:
+		visitor := &tableRenameVisitor{
+			targetNames: targetTableNames,
 		}
-
-		v.Table.Schema = model.NewCIStr(targetTableNames[0].Schema)
-		v.Table.Name = model.NewCIStr(targetTableNames[0].Name)
-
-		if v.Specs[0].Tp == ast.AlterTableRenameTable {
-			v.Specs[0].NewTable.Schema = model.NewCIStr(targetTableNames[1].Schema)
-			v.Specs[0].NewTable.Name = model.NewCIStr(targetTableNames[1].Name)
+		stmt.Accept(visitor)
+		if visitor.hasErr {
+			return "", terror.ErrRewriteSQL.Generate(stmt, targetTableNames)
 		}
-
-	default:
-		return "", terror.ErrUnknownTypeDDL.Generate(stmt)
 	}
 
 	var b []byte

pkg/parser/common_test.go

@@ -19,6 +19,8 @@ import (
 	. "github.com/pingcap/check"
 	"github.com/pingcap/parser"
 	"github.com/pingcap/tidb-tools/pkg/filter"
+
+	"github.com/pingcap/dm/pkg/terror"
 )
 
 var _ = Suite(&testParserSuite{})
@@ -93,6 +95,28 @@ func (t *testParserSuite) TestParser(c *C) {
 	}
 }
 
+func (t *testParserSuite) TestError(c *C) {
+	p := parser.New()
+
+	// DML will report ErrUnknownTypeDDL
+	dml := "INSERT INTO `t1` VALUES (1)"
+
+	stmts, err := Parse(p, dml, "", "")
+	c.Assert(err, IsNil)
+	_, err = FetchDDLTableNames("test", stmts[0])
+	c.Assert(terror.ErrUnknownTypeDDL.Equal(err), IsTrue)
+
+	_, err = RenameDDLTable(stmts[0], nil)
+	c.Assert(terror.ErrUnknownTypeDDL.Equal(err), IsTrue)
+
+	// tableRenameVisitor with less `targetNames` won't panic
+	ddl := "create table `s1`.`t1` (id int)"
+	stmts, err = Parse(p, ddl, "", "")
+	c.Assert(err, IsNil)
+	_, err = RenameDDLTable(stmts[0], nil)
+	c.Assert(terror.ErrRewriteSQL.Equal(err), IsTrue)
+}
+
 func (t *testParserSuite) TestResolveDDL(c *C) {
 	p := parser.New()
 	expectedSQLs := [][]string{
@@ -154,8 +178,8 @@ func (t *testParserSuite) TestResolveDDL(c *C) {
 		{{genTableName("test", "t1"), genTableName("xx", "t2")}},
 		{{genTableName("test", "t1")}},
 		{{genTableName("s1", "t1")}},
-		{{genTableName("s1", "t1"), genTableName("s2", "t2")}},
-		{{genTableName("test", "t1"), genTableName("test", "t2")}, {genTableName("s1", "t1"), genTableName("test", "t2")}},
+		{{genTableName("s1", "t1"), genTableName("s2", "t2"), genTableName("s1", "t1"), genTableName("s2", "t2")}},
+		{{genTableName("test", "t1"), genTableName("test", "t2"), genTableName("test", "t1"), genTableName("test", "t2")}, {genTableName("s1", "t1"), genTableName("test", "t2"), genTableName("s1", "t1"), genTableName("test", "t2")}},
 		{{genTableName("s1", "t1")}},
 		{{genTableName("test", "t1")}},
 		{{genTableName("test", "t1")}},
@@ -165,7 +189,7 @@ func (t *testParserSuite) TestResolveDDL(c *C) {
 		{{genTableName("s1", "t1")}, {genTableName("s1", "t1"), genTableName("xx", "t2")}, {genTableName("xx", "t2")}},
 		{{genTableName("test", "t1")}},
 		{{genTableName("test", "t1")}},
-		{{genTableName("test", "t1")}},
+		{{genTableName("test", "t1"), genTableName("test", "t2")}},
 		{{genTableName("test", "t1")}},
 		{{genTableName("test", "t1")}},
 		{{genTableName("test", "t1")}},
@@ -198,8 +222,8 @@ func (t *testParserSuite) TestResolveDDL(c *C) {
 		{{genTableName("xtest", "xt1"), genTableName("xxx", "xt2")}},
 		{{genTableName("xtest", "xt1")}},
 		{{genTableName("xs1", "xt1")}},
-		{{genTableName("xs1", "xt1"), genTableName("xs2", "xt2")}},
-		{{genTableName("xtest", "xt1"), genTableName("xtest", "xt2")}, {genTableName("xs1", "xt1"), genTableName("xtest", "xt2")}},
+		{{genTableName("xs1", "xt1"), genTableName("xs2", "xt2"), genTableName("xs1", "xt1"), genTableName("xs2", "xt2")}},
+		{{genTableName("xtest", "xt1"), genTableName("xtest", "xt2"), genTableName("xtest", "xt1"), genTableName("xtest", "xt2")}, {genTableName("xs1", "xt1"), genTableName("xtest", "xt2"), genTableName("xs1", "xt1"), genTableName("xtest", "xt2")}},
 		{{genTableName("xs1", "xt1")}},
 		{{genTableName("xtest", "xt1")}},
 		{{genTableName("xtest", "xt1")}},
@@ -209,7 +233,7 @@ func (t *testParserSuite) TestResolveDDL(c *C) {
 		{{genTableName("xs1", "xt1")}, {genTableName("xs1", "xt1"), genTableName("xxx", "xt2")}, {genTableName("xxx", "xt2")}},
 		{{genTableName("xtest", "xt1")}},
 		{{genTableName("xtest", "xt1")}},
-		{{genTableName("xtest", "xt1")}},
+		{{genTableName("xtest", "xt1"), genTableName("xtest", "xt2")}},
 		{{genTableName("xtest", "xt1")}},
 		{{genTableName("xtest", "xt1")}},
 		{{genTableName("xtest", "xt1")}},
@@ -253,7 +277,7 @@ func (t *testParserSuite) TestResolveDDL(c *C) {
 		{"ALTER TABLE `xs1`.`xt1` ADD COLUMN `c1` INT", "ALTER TABLE `xs1`.`xt1` RENAME AS `xxx`.`xt2`", "ALTER TABLE `xxx`.`xt2` DROP COLUMN `c2`"},
 		{"ALTER TABLE `xtest`.`xt1` ADD COLUMN IF NOT EXISTS `c1` INT"},
 		{"ALTER TABLE `xtest`.`xt1` ADD INDEX IF NOT EXISTS(`a`) USING BTREE COMMENT 'a'"},
-		{"ALTER TABLE `xtest`.`xt1` ADD CONSTRAINT `fk_t2_id` FOREIGN KEY IF NOT EXISTS (`t2_id`) REFERENCES `t2`(`id`)"},
+		{"ALTER TABLE `xtest`.`xt1` ADD CONSTRAINT `fk_t2_id` FOREIGN KEY IF NOT EXISTS (`t2_id`) REFERENCES `xtest`.`xt2`(`id`)"},
 		{"CREATE INDEX IF NOT EXISTS `i1` ON `xtest`.`xt1` (`c1`)"},
 		{"ALTER TABLE `xtest`.`xt1` ADD PARTITION IF NOT EXISTS (PARTITION `p2` VALUES LESS THAN (MAXVALUE))"},
 		{"ALTER TABLE `xtest`.`xt1` DROP COLUMN IF EXISTS `c2`"},
@@ -295,5 +319,4 @@ func (t *testParserSuite) TestResolveDDL(c *C) {
 			c.Assert(targetSQL, Equals, targetSQLs[i][j])
 		}
 	}
-
 }

pkg/terror/error_list.go

@@ -171,6 +171,9 @@ const (
 
 	// dm/command
 	codeVerifyHandleErrorArgs
+
+	// pkg/parser
+	codeRewriteSQL
 )
 
 // Config related error code list
@@ -755,6 +758,9 @@ var (
 	// Functional error
 	ErrVerifyHandleErrorArgs = New(codeVerifyHandleErrorArgs, ClassFunctional, ScopeInternal, LevelLow, "", "Please make sure the args are correct.")
 
+	// pkg/parser
+	ErrRewriteSQL = New(codeRewriteSQL, ClassFunctional, ScopeInternal, LevelHigh, "failed to rewrite SQL for target DB, stmt: %+v, targetTableNames: %+v", "")
+
 	// Config related error
 	ErrConfigCheckItemNotSupport    = New(codeConfigCheckItemNotSupport, ClassConfig, ScopeInternal, LevelMedium, "checking item %s is not supported\n%s", "Please check `ignore-checking-items` config in task configuration file, which can be set including `all`/`dump_privilege`/`replication_privilege`/`version`/`binlog_enable`/`binlog_format`/`binlog_row_image`/`table_schema`/`schema_of_shard_tables`/`auto_increment_ID`.")
 	ErrConfigTomlTransform          = New(codeConfigTomlTransform, ClassConfig, ScopeInternal, LevelMedium, "%s", "Please check the configuration file has correct TOML format.")

pkg/utils/util.go

@@ -63,8 +63,6 @@ var (
 		"^ALTER\\s+PROCEDURE",
 
 		// view
-		"^CREATE\\s*(OR REPLACE)?\\s+(ALGORITHM\\s?=.+?)?(DEFINER\\s?=.+?)?\\s+(SQL SECURITY DEFINER)?VIEW",
-		"^DROP\\s+VIEW",
 		"^ALTER\\s+(ALGORITHM\\s?=.+?)?(DEFINER\\s?=.+?)?(SQL SECURITY DEFINER)?VIEW",
 
 		// function

syncer/filter_test.go

@@ -63,10 +63,10 @@ BEGIN
 END`, true},
 
 		// view
-		{"CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`%` SQL SECURITY DEFINER VIEW `v` AS SELECT qty, price, qty*price AS value FROM t", true},
-		{"CREATE OR REPLACE ALGORITHM=UNDEFINED DEFINER=`root`@`%` SQL SECURITY DEFINER VIEW `v` AS SELECT qty, price, qty*price AS value FROM t", true},
+		{"CREATE ALGORITHM=UNDEFINED DEFINER=`root`@`%` SQL SECURITY DEFINER VIEW `v` AS SELECT qty, price, qty*price AS value FROM t", false},
+		{"CREATE OR REPLACE ALGORITHM=UNDEFINED DEFINER=`root`@`%` SQL SECURITY DEFINER VIEW `v` AS SELECT qty, price, qty*price AS value FROM t", false},
 		{"ALTER ALGORITHM=UNDEFINED DEFINER=`root`@`%` SQL SECURITY DEFINER VIEW `v` AS SELECT qty, price, qty*price AS value FROM t", true},
-		{"DROP VIEW v", true},
+		{"DROP VIEW v", false},
 		{"CREATE TABLE `VIEW`(id int)", false},
 		{"ALTER TABLE `VIEW`(id int)", false},