syncer: improve coverage for syncer/ddl.go to 85.1% (#1109) (#1120)

Co-authored-by: lance6716 <[email protected]>

syncer/ddl.go

@@ -127,7 +127,7 @@ func (s *Syncer) parseDDLSQL(sql string, p *parser.Parser, schema string) (resul
 /// resolveDDLSQL do two things
 // * it splits multiple operations in one DDL statement into multiple DDL statements
 // * try to apply online ddl by given online
-// return @spilted sqls, @online ddl table names, @error
+// return @spilt sqls, @online ddl table names, @error
 func (s *Syncer) resolveDDLSQL(tctx *tcontext.Context, p *parser.Parser, stmt ast.StmtNode, schema string) (sqls []string, tables map[string]*filter.Table, err error) {
 	sqls, err = parserpkg.SplitDDL(stmt, schema)
 	if err != nil {

syncer/ddl_test.go

@@ -14,14 +14,18 @@
 package syncer
 
 import (
+	"errors"
+	"fmt"
+
 	"github.com/pingcap/dm/dm/config"
 	tcontext "github.com/pingcap/dm/pkg/context"
 	parserpkg "github.com/pingcap/dm/pkg/parser"
 	"github.com/pingcap/dm/pkg/utils"
 
-	sqlmock "github.com/DATA-DOG/go-sqlmock"
+	"github.com/DATA-DOG/go-sqlmock"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/parser"
+	"github.com/pingcap/parser/ast"
 	"github.com/pingcap/tidb-tools/pkg/filter"
 	router "github.com/pingcap/tidb-tools/pkg/table-router"
 )
@@ -87,7 +91,7 @@ func (s *testSyncerSuite) TestCommentQuote(c *C) {
 	c.Assert(sqls[0], Equals, expectedSQL)
 }
 
-func (s *testSyncerSuite) TestresolveDDLSQL(c *C) {
+func (s *testSyncerSuite) TestResolveDDLSQL(c *C) {
 	// duplicate with pkg/parser
 	sqls := []string{
 		"create schema `s1`",
@@ -296,6 +300,13 @@ func (s *testSyncerSuite) TestParseDDLSQL(c *C) {
 			isDDL:    false,
 			hasError: false,
 		},
+		{
+			sql:      "NOT A SQL",
+			schema:   "",
+			ignore:   false,
+			isDDL:    false,
+			hasError: true,
+		},
 	}
 
 	cfg := &config.SubTaskConfig{
@@ -358,3 +369,155 @@ func (s *testSyncerSuite) TestResolveGeneratedColumnSQL(c *C) {
 		c.Assert(getSQL, Equals, tc.expected)
 	}
 }
+
+func (s *testSyncerSuite) TestResolveOnlineDDL(c *C) {
+	cases := []struct {
+		onlineType string
+		trashName  string
+		ghostname  string
+	}{
+		{
+			config.GHOST,
+			"_t1_del",
+			"_t1_gho",
+		},
+		{
+			config.PT,
+			"_t1_old",
+			"_t1_new",
+		},
+	}
+	tctx := tcontext.Background()
+	p := parser.New()
+
+	for _, ca := range cases {
+		fn := OnlineDDLSchemes[ca.onlineType]
+		plugin, err := fn(tctx, s.cfg)
+		c.Assert(err, IsNil)
+		syncer := NewSyncer(s.cfg, nil)
+		syncer.onlineDDL = plugin
+		c.Assert(plugin.Clear(tctx), IsNil)
+
+		// real table
+		sql := "ALTER TABLE `test`.`t1` ADD COLUMN `n` INT"
+		stmt, err := p.ParseOneStmt(sql, "", "")
+		c.Assert(err, IsNil)
+		sqls, tables, err := syncer.resolveDDLSQL(tctx, p, stmt, "test")
+		c.Assert(err, IsNil)
+		c.Assert(sqls, HasLen, 1)
+		c.Assert(sqls[0], Equals, sql)
+		c.Assert(tables, HasLen, 0)
+
+		// trash table
+		sql = fmt.Sprintf("CREATE TABLE IF NOT EXISTS `test`.`%s` (`n` INT)", ca.trashName)
+		stmt, err = p.ParseOneStmt(sql, "", "")
+		c.Assert(err, IsNil)
+		sqls, tables, err = syncer.resolveDDLSQL(tctx, p, stmt, "test")
+		c.Assert(err, IsNil)
+		c.Assert(sqls, HasLen, 0)
+		c.Assert(tables, HasLen, 0)
+
+		// ghost table
+		sql = fmt.Sprintf("ALTER TABLE `test`.`%s` ADD COLUMN `n` INT", ca.ghostname)
+		newSQL := "ALTER TABLE `test`.`t1` ADD COLUMN `n` INT"
+		stmt, err = p.ParseOneStmt(sql, "", "")
+		c.Assert(err, IsNil)
+		sqls, tables, err = syncer.resolveDDLSQL(tctx, p, stmt, "test")
+		c.Assert(err, IsNil)
+		c.Assert(sqls, HasLen, 0)
+		c.Assert(tables, HasLen, 0)
+		sql = fmt.Sprintf("RENAME TABLE `test`.`%s` TO `test`.`t1`", ca.ghostname)
+		stmt, err = p.ParseOneStmt(sql, "", "")
+		c.Assert(err, IsNil)
+		sqls, tables, err = syncer.resolveDDLSQL(tctx, p, stmt, "test")
+		c.Assert(err, IsNil)
+		c.Assert(sqls, HasLen, 1)
+		c.Assert(sqls[0], Equals, newSQL)
+		tableName := &filter.Table{Schema: "test", Name: ca.ghostname}
+		c.Assert(tables, DeepEquals, map[string]*filter.Table{tableName.String(): tableName})
+	}
+}
+
+func (s *testSyncerSuite) TestDropSchemaInSharding(c *C) {
+	var (
+		targetDB  = "target_db"
+		targetTbl = "tbl"
+		sourceDB  = "db1"
+		source1   = "`db1`.`tbl1`"
+		source2   = "`db1`.`tbl2`"
+		tctx      = tcontext.Background()
+	)
+	clone, _ := s.cfg.Clone()
+	clone.ShardMode = config.ShardPessimistic
+	syncer := NewSyncer(clone, nil)
+	_, _, _, _, err := syncer.sgk.AddGroup(targetDB, targetTbl, []string{source1}, nil, true)
+	c.Assert(err, IsNil)
+	_, _, _, _, err = syncer.sgk.AddGroup(targetDB, targetTbl, []string{source2}, nil, true)
+	c.Assert(err, IsNil)
+	c.Assert(syncer.sgk.Groups(), HasLen, 2)
+	c.Assert(syncer.dropSchemaInSharding(tctx, sourceDB), IsNil)
+	c.Assert(syncer.sgk.Groups(), HasLen, 0)
+}
+
+type mockOnlinePlugin struct {
+	toFinish map[string]struct{}
+}
+
+func (m mockOnlinePlugin) Apply(tctx *tcontext.Context, tables []*filter.Table, statement string, stmt ast.StmtNode) ([]string, string, string, error) {
+	return nil, "", "", nil
+}
+
+func (m mockOnlinePlugin) Finish(tctx *tcontext.Context, schema, table string) error {
+	k := schema + table
+	if _, ok := m.toFinish[k]; !ok {
+		return errors.New("finish table not found")
+	}
+	delete(m.toFinish, schema+table)
+	return nil
+}
+
+func (m mockOnlinePlugin) TableType(table string) TableType {
+	return ""
+}
+
+func (m mockOnlinePlugin) RealName(schema, table string) (string, string) {
+	return "", ""
+}
+
+func (m mockOnlinePlugin) ResetConn(tctx *tcontext.Context) error {
+	return nil
+}
+
+func (m mockOnlinePlugin) Clear(tctx *tcontext.Context) error {
+	return nil
+}
+
+func (m mockOnlinePlugin) Close() {
+}
+
+func (s *testSyncerSuite) TestClearOnlineDDL(c *C) {
+	var (
+		targetDB  = "target_db"
+		targetTbl = "tbl"
+		source1   = "`db1`.`tbl1`"
+		key1      = "db1tbl1"
+		source2   = "`db1`.`tbl2`"
+		key2      = "db1tbl2"
+		tctx      = tcontext.Background()
+	)
+	clone, _ := s.cfg.Clone()
+	clone.ShardMode = config.ShardPessimistic
+	syncer := NewSyncer(clone, nil)
+	mock := mockOnlinePlugin{
+		map[string]struct{}{key1: {}, key2: {}},
+	}
+	syncer.onlineDDL = mock
+
+	_, _, _, _, err := syncer.sgk.AddGroup(targetDB, targetTbl, []string{source1}, nil, true)
+	c.Assert(err, IsNil)
+	_, _, _, _, err = syncer.sgk.AddGroup(targetDB, targetTbl, []string{source2}, nil, true)
+	c.Assert(err, IsNil)
+
+	c.Assert(syncer.clearOnlineDDL(tctx, targetDB, targetTbl), IsNil)
+	c.Assert(mock.toFinish, HasLen, 0)
+}

syncer/pt_osc.go

@@ -28,7 +28,7 @@ import (
 
 // PT handles pt online schema changes
 // (_*).*_new ghost table
-// (_*).*_old ghost transh table
+// (_*).*_old ghost trash table
 // we don't support `--new-table-name` flag
 type PT struct {
 	storge *OnlineDDLStorage
@@ -140,7 +140,7 @@ func (p *PT) Finish(tcxt *tcontext.Context, schema, table string) error {
 
 // TableType implements interface
 func (p *PT) TableType(table string) TableType {
-	// 5 is _ _gho/ghc/del
+	// 5 is _ _old/new
 	if len(table) > 5 {
 		if strings.HasPrefix(table, "_") && strings.HasSuffix(table, "_new") {
 			return ghostTable