loader, syncer: support create/drop view

_utils/terror_gen/errors_release.txt

@@ -247,6 +247,7 @@ ErrLoadUnitNoTableFile,[code=34013:class=load-unit:scope=internal:level=high], "
 ErrLoadUnitDumpDirNotFound,[code=34014:class=load-unit:scope=internal:level=high], "Message: %s does not exist or it's not a dir"
 ErrLoadUnitDuplicateTableFile,[code=34015:class=load-unit:scope=internal:level=high], "Message: invalid table schema file, duplicated item - %s"
 ErrLoadUnitGenBAList,[code=34016:class=load-unit:scope=internal:level=high], "Message: generate block allow list, Workaround: Please check the `block-allow-list` config in task configuration file."
+ErrLoadUnitNoTableFileForView,[code=34017:class=load-unit:scope=internal:level=high], "Message: invalid view sql file, cannot find table - %s"
 ErrSyncerUnitPanic,[code=36001:class=sync-unit:scope=internal:level=high], "Message: panic error: %v"
 ErrSyncUnitInvalidTableName,[code=36002:class=sync-unit:scope=internal:level=high], "Message: extract table name for DML error: %s"
 ErrSyncUnitTableNameQuery,[code=36003:class=sync-unit:scope=internal:level=high], "Message: table name parse error: %s"

dumpling/dumpling.go

@@ -218,6 +218,7 @@ func (m *Dumpling) constructArgs() (*export.Config, error) {
 	dumpConfig.TableFilter = tableFilter
 	dumpConfig.CompleteInsert = true // always keep column name in `INSERT INTO` statements.
 	dumpConfig.Logger = m.logger.Logger
+	dumpConfig.NoViews = false
 
 	if cfg.Threads > 0 {
 		dumpConfig.Threads = cfg.Threads

errors.toml

@@ -1492,6 +1492,12 @@ description = ""
 workaround = "Please check the `block-allow-list` config in task configuration file."
 tags = ["internal", "high"]
 
+[error.DM-load-unit-34017]
+message = "invalid view sql file, cannot find table - %s"
+description = ""
+workaround = ""
+tags = ["internal", "high"]
+
 [error.DM-sync-unit-36001]
 message = "panic error: %v"
 description = ""

go.mod

@@ -18,7 +18,7 @@ require (
 	github.com/kami-zh/go-capturer v0.0.0-20171211120116-e492ea43421d
 	github.com/pingcap/br v5.0.0-rc.0.20201223100334-c344d1edf20c+incompatible // indirect
 	github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712
-	github.com/pingcap/dumpling v0.0.0-20201224084933-34903c7475cf
+	github.com/pingcap/dumpling v0.0.0-20201230072552-5f1acb878f3e
 	github.com/pingcap/errors v0.11.5-0.20201126102027-b0a155152ca3
 	github.com/pingcap/failpoint v0.0.0-20200702092429-9f69995143ce
 	github.com/pingcap/log v0.0.0-20201112100606-8f1e84a3abc8

go.sum

@@ -811,8 +811,8 @@ github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712 h1:R8gStypOBmpnHEx1q
 github.com/pingcap/check v0.0.0-20200212061837-5e12011dc712/go.mod h1:PYMCGwN0JHjoqGr3HrZoD+b8Tgx8bKnArhSq8YVzUMc=
 github.com/pingcap/dm v1.1.0-alpha.0.20200521025928-83063141c5fd/go.mod h1:I5AAhwb0JPfLZINukML5VU9rB6mCcVA/Jq5OFFtMuEk=
 github.com/pingcap/dumpling v0.0.0-20200423082233-887d037b5b5c/go.mod h1:VJTcnA0MLL9tzDceTDoRh3k5UnOq9Hk6wh/ATo+B8I8=
-github.com/pingcap/dumpling v0.0.0-20201224084933-34903c7475cf h1:D1ujZCR0h3BS4ppd+nR870k26NGDpVCYeKrNLOHPlNU=
-github.com/pingcap/dumpling v0.0.0-20201224084933-34903c7475cf/go.mod h1:qHuvF07zoRcpovYVKqbGortzOyct/e9SdWa3wGop9sE=
+github.com/pingcap/dumpling v0.0.0-20201230072552-5f1acb878f3e h1:2s0tKThi9KzkQhPt002wr4GxebjYveO376LD2G+ALtE=
+github.com/pingcap/dumpling v0.0.0-20201230072552-5f1acb878f3e/go.mod h1:qHuvF07zoRcpovYVKqbGortzOyct/e9SdWa3wGop9sE=
 github.com/pingcap/errcode v0.0.0-20180921232412-a1a7271709d9/go.mod h1:4b2X8xSqxIroj/IZ9MX/VGZhAwc11wB9wRIzHvz6SeM=
 github.com/pingcap/errcode v0.3.0/go.mod h1:4b2X8xSqxIroj/IZ9MX/VGZhAwc11wB9wRIzHvz6SeM=
 github.com/pingcap/errors v0.11.0 h1:DCJQB8jrHbQ1VVlMFIrbj2ApScNNotVmkSNplu2yUt4=

loader/loader.go

@@ -28,6 +28,9 @@ import (
 	"sync/atomic"
 	"time"
 
+	"github.com/pingcap/parser"
+	tmysql "github.com/pingcap/parser/mysql"
+
 	"github.com/pingcap/dm/dm/config"
 	"github.com/pingcap/dm/dm/pb"
 	"github.com/pingcap/dm/dm/unit"
@@ -36,6 +39,7 @@ import (
 	"github.com/pingcap/dm/pkg/dumpling"
 	fr "github.com/pingcap/dm/pkg/func-rollback"
 	"github.com/pingcap/dm/pkg/log"
+	parserpkg "github.com/pingcap/dm/pkg/parser"
 	"github.com/pingcap/dm/pkg/terror"
 	"github.com/pingcap/dm/pkg/utils"
 
@@ -384,7 +388,9 @@ type Loader struct {
 
 	// db -> tables
 	// table -> data files
-	db2Tables  map[string]Tables2DataFiles
+	db2Tables map[string]Tables2DataFiles
+	// db -> views
+	db2Views   map[string]map[string]struct{}
 	tableInfos map[string]*tableInfo
 
 	// for every worker goroutine, not for every data file
@@ -421,6 +427,7 @@ func NewLoader(cfg *config.SubTaskConfig) *Loader {
 	loader := &Loader{
 		cfg:        cfg,
 		db2Tables:  make(map[string]Tables2DataFiles),
+		db2Views:   make(map[string]map[string]struct{}),
 		tableInfos: make(map[string]*tableInfo),
 		workerWg:   new(sync.WaitGroup),
 		logger:     log.With(zap.String("task", cfg.Name), zap.String("unit", "load")),
@@ -946,6 +953,44 @@ func (l *Loader) prepareTableFiles(files map[string]struct{}) error {
 	return nil
 }
 
+func (l *Loader) prepareViewFiles(files map[string]struct{}) error {
+	for file := range files {
+		if !strings.HasSuffix(file, "-schema-view.sql") {
+			continue
+		}
+
+		idx := strings.LastIndex(file, "-schema-view.sql")
+		name := file[:idx]
+		fields := strings.Split(name, ".")
+		if len(fields) != 2 {
+			l.logger.Warn("invalid view file", zap.String("file", file))
+			continue
+		}
+
+		db, table := fields[0], fields[1]
+		if l.skipSchemaAndTable(&filter.Table{Schema: db, Name: table}) {
+			l.logger.Warn("ignore view file", zap.String("view file", file))
+			continue
+		}
+		// because there's a table file for view file, we skip this check
+		tables := l.db2Tables[db]
+		if _, ok := tables[table]; !ok {
+			return terror.ErrLoadUnitNoTableFileForView.Generate(file)
+		}
+
+		views, ok := l.db2Views[db]
+		if !ok {
+			l.db2Views[db] = map[string]struct{}{}
+			views = l.db2Views[db]
+		}
+		views[table] = struct{}{}
+
+		l.totalFileCount.Add(1) // for view
+	}
+
+	return nil
+}
+
 func (l *Loader) prepareDataFiles(files map[string]struct{}) error {
 	var dataFilesNumber float64
 
@@ -1046,6 +1091,11 @@ func (l *Loader) prepare() error {
 		return err
 	}
 
+	// Sql file for create view
+	if err := l.prepareViewFiles(files); err != nil {
+		return err
+	}
+
 	// Sql file for restore data
 	return l.prepareDataFiles(files)
 }
@@ -1084,6 +1134,104 @@ func (l *Loader) restoreTable(ctx context.Context, conn *DBConn, sqlFile, schema
 	return nil
 }
 
+// restoreView drops dummy table and create view
+func (l *Loader) restoreView(ctx context.Context, conn *DBConn, sqlFile, schema, view string) error {
+	// dumpling will generate such a viewFile
+	// /*!40101 SET NAMES binary*/;
+	// DROP TABLE IF EXISTS `v2`;
+	// DROP VIEW IF EXISTS `v2`;
+	// SET @PREV_CHARACTER_SET_CLIENT=@@CHARACTER_SET_CLIENT;
+	// SET @PREV_CHARACTER_SET_RESULTS=@@CHARACTER_SET_RESULTS;
+	// SET @PREV_COLLATION_CONNECTION=@@COLLATION_CONNECTION;
+	// SET character_set_client = utf8;
+	// SET character_set_results = utf8;
+	// SET collation_connection = utf8_general_ci;
+	// CREATE ALGORITHM=UNDEFINED DEFINER="root"@"localhost" SQL SECURITY DEFINER VIEW "all_mode"."v2" AS select "all_mode"."t2"."id" AS "id" from "all_mode"."t2";
+	// SET character_set_client = @PREV_CHARACTER_SET_CLIENT;
+	// SET character_set_results = @PREV_CHARACTER_SET_RESULTS;
+	// SET collation_connection = @PREV_COLLATION_CONNECTION;
+	f, err := os.Open(sqlFile)
+	if err != nil {
+		return terror.ErrLoadUnitReadSchemaFile.Delegate(err)
+	}
+	defer f.Close()
+
+	tctx := tcontext.NewContext(ctx, l.logger)
+
+	var sqls []string
+	dstSchema, dstView := fetchMatchedLiteral(tctx, l.tableRouter, schema, view)
+	sqls = append(sqls, fmt.Sprintf("USE `%s`;", unescapePercent(dstSchema, l.logger)))
+	sqlMode, err := tmysql.GetSQLMode(l.cfg.SQLMode)
+	if err != nil {
+		// should not happened
+		return terror.ErrGetSQLModeFromStr.Generate(l.cfg.SQLMode)
+	}
+
+	data := make([]byte, 0, 1024*1024)
+	br := bufio.NewReader(f)
+	for {
+		line, err := br.ReadString('\n')
+		if err == io.EOF {
+			break
+		}
+
+		realLine := strings.TrimSpace(line[:len(line)-1])
+		if len(realLine) == 0 {
+			continue
+		}
+
+		data = append(data, []byte(realLine)...)
+		if data[len(data)-1] == ';' {
+			query := string(data)
+			data = data[0:0]
+			if strings.HasPrefix(query, "/*") && strings.HasSuffix(query, "*/;") {
+				continue
+			}
+
+			// handle route-rules below. we could skip SET and only check DROP/CREATE
+			if strings.HasPrefix(query, "DROP") {
+				query = renameShardingTable(query, view, dstView, false)
+			} else if strings.HasPrefix(query, "CREATE") {
+				// create view statement could be complicated because it has a select
+				p := parser.New()
+				p.SetSQLMode(sqlMode)
+				stmt, err := p.ParseOneStmt(query, "", "")
+				if err != nil {
+					return terror.ErrLoadUnitParseStatement.Generate(query)
+				}
+
+				tableNames, err := parserpkg.FetchDDLTableNames(schema, stmt)
+				if err != nil {
+					return terror.WithScope(err, terror.ScopeInternal)
+				}
+				targetTableNames := make([]*filter.Table, 0, len(tableNames))
+				for i := range tableNames {
+					dstSchema, dstTable := fetchMatchedLiteral(tctx, l.tableRouter, tableNames[i].Schema, tableNames[i].Name)
+					tableName := &filter.Table{
+						Schema: dstSchema,
+						Name:   dstTable,
+					}
+					targetTableNames = append(targetTableNames, tableName)
+				}
+				query, err = parserpkg.RenameDDLTable(stmt, targetTableNames)
+				if err != nil {
+					return terror.WithScope(err, terror.ScopeInternal)
+				}
+			}
+
+			l.logger.Debug("view create statement", zap.String("sql", query))
+
+			sqls = append(sqls, query)
+		}
+	}
+	err = conn.executeSQL(tctx, sqls)
+	if err != nil {
+		return terror.WithScope(err, terror.ScopeDownstream)
+	}
+
+	return nil
+}
+
 // restoreStruture creates schema or table
 func (l *Loader) restoreStructure(ctx context.Context, conn *DBConn, sqlFile string, schema string, table string) error {
 	f, err := os.Open(sqlFile)
@@ -1194,18 +1342,9 @@ func (l *Loader) restoreData(ctx context.Context) error {
 	}
 
 	dispatchMap := make(map[string]*fileJob)
-
-	// restore db in sort
-	dbs := make([]string, 0, len(l.db2Tables))
-	for db := range l.db2Tables {
-		dbs = append(dbs, db)
-	}
-
 	tctx := tcontext.NewContext(ctx, l.logger)
 
-	for _, db := range dbs {
-		tables := l.db2Tables[db]
-
+	for db, tables := range l.db2Tables {
 		// create db
 		dbFile := fmt.Sprintf("%s/%s-schema-create.sql", l.cfg.Dir, db)
 		l.logger.Info("start to create schema", zap.String("schema file", dbFile))
@@ -1274,7 +1413,19 @@ func (l *Loader) restoreData(ctx context.Context) error {
 			}
 		}
 	}
-	l.logger.Info("finish to create tables", zap.Duration("cost time", time.Since(begin)))
+
+	for db, views := range l.db2Views {
+		for view := range views {
+			viewFile := fmt.Sprintf("%s/%s.%s-schema-view.sql", l.cfg.Dir, db, view)
+			l.logger.Info("start to create view", zap.String("view file", viewFile))
+			err := l.restoreView(ctx, dbConn, viewFile, db, view)
+			if err != nil {
+				return err
+			}
+			l.logger.Info("finish to create view", zap.String("view file", viewFile))
+		}
+	}
+	l.logger.Info("finish to create tables and views", zap.Duration("cost time", time.Since(begin)))
 
 	// a simple and naive approach to dispatch files randomly based on the feature of golang map(range by random)
 	for _, j := range dispatchMap {

pkg/parser/common.go

@@ -51,14 +51,27 @@ type tableNameExtractor struct {
 }
 
 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}
+	switch n := in.(type) {
+	case *ast.TableName:
+		tb := &filter.Table{Schema: n.Schema.L, Name: n.Name.L}
 		if tb.Schema == "" {
 			tb.Schema = tne.curDB
 		}
 		tne.names = append(tne.names, tb)
 		return in, true
+	case *ast.ColumnName:
+		tb := &filter.Table{Schema: n.Schema.L, Name: n.Table.L}
+		// this column has specified a table, such as
+		// CREATE VIEW `v1` AS SELECT `t1`.`c1` AS `c1` FROM `t1`
+		if tb.Name != "" {
+			if tb.Schema == "" {
+				tb.Schema = tne.curDB
+			}
+			tne.names = append(tne.names, tb)
+		}
+		return in, true
 	}
+
 	return in, false
 }
 
@@ -107,16 +120,29 @@ func (v *tableRenameVisitor) Enter(in ast.Node) (ast.Node, bool) {
 	if v.hasErr {
 		return in, true
 	}
-	if t, ok := in.(*ast.TableName); ok {
+	switch n := in.(type) {
+	case *ast.TableName:
 		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)
+		n.Schema = model.NewCIStr(v.targetNames[v.i].Schema)
+		n.Name = model.NewCIStr(v.targetNames[v.i].Name)
 		v.i++
 		return in, true
+	case *ast.ColumnName:
+		if n.Table.L != "" {
+			if v.i >= len(v.targetNames) {
+				v.hasErr = true
+				return in, true
+			}
+			n.Schema = model.NewCIStr(v.targetNames[v.i].Schema)
+			n.Table = model.NewCIStr(v.targetNames[v.i].Name)
+			v.i++
+		}
+		return in, true
 	}
+
 	return in, false
 }
 
@@ -167,6 +193,31 @@ func RenameDDLTable(stmt ast.StmtNode, targetTableNames []*filter.Table) (string
 	return bf.String(), nil
 }
 
+type dbNameAppender struct {
+	curDB model.CIStr
+}
+
+func (v *dbNameAppender) Enter(in ast.Node) (ast.Node, bool) {
+	switch n := in.(type) {
+	case *ast.TableName:
+		if n.Schema.O == "" {
+			n.Schema = v.curDB
+		}
+		return in, true
+	case *ast.ColumnName:
+		if n.Table.O != "" && n.Schema.O == "" {
+			n.Schema = v.curDB
+		}
+		return in, true
+	}
+
+	return in, false
+}
+
+func (v *dbNameAppender) Leave(in ast.Node) (ast.Node, bool) {
+	return in, true
+}
+
 // SplitDDL splits multiple operations in one DDL statement into multiple DDL statements
 // returned DDL is formatted like StringSingleQuotes, KeyWordUppercase and NameBackQuotes
 // if fail to restore, it would not restore the value of `stmt` (it changes it's values if `stmt` is one of  DropTableStmt, RenameTableStmt, AlterTableStmt)
@@ -309,6 +360,9 @@ func SplitDDL(stmt ast.StmtNode, schema string) (sqls []string, err error) {
 		v.Table = table
 
 		return sqls, nil
+	case *ast.CreateViewStmt:
+		visitor := &dbNameAppender{curDB: schemaName}
+		v.Accept(visitor)
 	default:
 		return nil, terror.ErrUnknownTypeDDL.Generate(stmt)
 	}