onlineddl: support regular expressions rules for PT/GH-OST (#2139) (#…

…2156)

_utils/terror_gen/errors_release.txt

@@ -177,6 +177,7 @@ ErrConfigGenBAList,[code=20044:class=config:scope=internal:level=high], "Message
 ErrConfigGenTableRouter,[code=20045:class=config:scope=internal:level=high], "Message: generate table router error, Workaround: Please check the `routes` config in task configuration file."
 ErrConfigGenColumnMapping,[code=20046:class=config:scope=internal:level=high], "Message: generate column mapping error, Workaround: Please check the `column-mappings` config in task configuration file."
 ErrConfigInvalidChunkFileSize,[code=20047:class=config:scope=internal:level=high], "Message: invalid `chunk-filesize` %v, Workaround: Please check the `chunk-filesize` config in task configuration file."
+ErrConfigOnlineDDLInvalidRegex,[code=20048:class=config:scope=internal:level=high], "Message: config '%s' regex pattern '%s' invalid, reason: %s, Workaround: Please check if params is correctly in the configuration file."
 ErrBinlogExtractPosition,[code=22001:class=binlog-op:scope=internal:level=high]
 ErrBinlogInvalidFilename,[code=22002:class=binlog-op:scope=internal:level=high], "Message: invalid binlog filename"
 ErrBinlogParsePosFromStr,[code=22003:class=binlog-op:scope=internal:level=high]

dm/config/subtask.go

@@ -19,6 +19,7 @@ import (
 	"flag"
 	"fmt"
 	"os"
+	"regexp"
 	"strconv"
 	"strings"
 
@@ -40,6 +41,9 @@ const (
 	ModeAll       = "all"
 	ModeFull      = "full"
 	ModeIncrement = "incremental"
+
+	DefaultShadowTableRules = "^_(.+)_(?:new|gho)$"
+	DefaultTrashTableRules  = "^_(.+)_(?:ghc|del|old)$"
 )
 
 var defaultMaxIdleConns = 2
@@ -192,6 +196,11 @@ type SubTaskConfig struct {
 	IsSharding bool   `toml:"is-sharding" json:"is-sharding"`
 	ShardMode  string `toml:"shard-mode" json:"shard-mode"`
 	OnlineDDL  bool   `toml:"online-ddl" json:"online-ddl"`
+
+	// pt/gh-ost name rule, support regex
+	ShadowTableRules []string `yaml:"shadow-table-rules" toml:"shadow-table-rules" json:"shadow-table-rules"`
+	TrashTableRules  []string `yaml:"trash-table-rules" toml:"trash-table-rules" json:"trash-table-rules"`
+
 	// deprecated
 	OnlineDDLScheme string `toml:"online-ddl-scheme" json:"online-ddl-scheme"`
 
@@ -317,6 +326,29 @@ func (c *SubTaskConfig) Decode(data string, verifyDecryptPassword bool) error {
 	return c.Adjust(verifyDecryptPassword)
 }
 
+func adjustOnlineTableRules(ruleType string, rules []string) ([]string, error) {
+	adjustedRules := make([]string, 0, len(rules))
+	for _, r := range rules {
+		if !strings.HasPrefix(r, "^") {
+			r = "^" + r
+		}
+
+		if !strings.HasSuffix(r, "$") {
+			r += "$"
+		}
+
+		p, err := regexp.Compile(r)
+		if err != nil {
+			return rules, terror.ErrConfigOnlineDDLInvalidRegex.Generate(ruleType, r, "fail to compile: "+err.Error())
+		}
+		if p.NumSubexp() != 1 {
+			return rules, terror.ErrConfigOnlineDDLInvalidRegex.Generate(ruleType, r, "rule isn't contains exactly one submatch")
+		}
+		adjustedRules = append(adjustedRules, r)
+	}
+	return adjustedRules, nil
+}
+
 // Adjust adjusts and verifies configs.
 func (c *SubTaskConfig) Adjust(verifyDecryptPassword bool) error {
 	if c.Name == "" {
@@ -343,6 +375,26 @@ func (c *SubTaskConfig) Adjust(verifyDecryptPassword bool) error {
 		log.L().Warn("'online-ddl-scheme' will be deprecated soon. Recommend that use online-ddl instead of online-ddl-scheme.")
 	}
 
+	if len(c.ShadowTableRules) == 0 {
+		c.ShadowTableRules = []string{DefaultShadowTableRules}
+	} else {
+		shadowTableRule, err := adjustOnlineTableRules("shadow-table-rules", c.ShadowTableRules)
+		if err != nil {
+			return err
+		}
+		c.ShadowTableRules = shadowTableRule
+	}
+
+	if len(c.TrashTableRules) == 0 {
+		c.TrashTableRules = []string{DefaultTrashTableRules}
+	} else {
+		trashTableRule, err := adjustOnlineTableRules("trash-table-rules", c.TrashTableRules)
+		if err != nil {
+			return err
+		}
+		c.TrashTableRules = trashTableRule
+	}
+
 	if c.MetaSchema == "" {
 		c.MetaSchema = defaultMetaSchema
 	}

dm/config/task.go

@@ -302,6 +302,10 @@ type TaskConfig struct {
 	MySQLInstances []*MySQLInstance `yaml:"mysql-instances" toml:"mysql-instances" json:"mysql-instances"`
 
 	OnlineDDL bool `yaml:"online-ddl" toml:"online-ddl" json:"online-ddl"`
+	// pt/gh-ost name rule,support regex
+	ShadowTableRules []string `yaml:"shadow-table-rules" toml:"shadow-table-rules" json:"shadow-table-rules"`
+	TrashTableRules  []string `yaml:"trash-table-rules" toml:"trash-table-rules" json:"trash-table-rules"`
+
 	// deprecated
 	OnlineDDLScheme string `yaml:"online-ddl-scheme" toml:"online-ddl-scheme" json:"online-ddl-scheme"`
 
@@ -708,6 +712,8 @@ func (c *TaskConfig) SubTaskConfigs(sources map[string]DBConfig) ([]*SubTaskConf
 		cfg.IsSharding = c.IsSharding
 		cfg.ShardMode = c.ShardMode
 		cfg.OnlineDDL = c.OnlineDDL
+		cfg.TrashTableRules = c.TrashTableRules
+		cfg.ShadowTableRules = c.ShadowTableRules
 		cfg.IgnoreCheckingItems = c.IgnoreCheckingItems
 		cfg.Name = c.Name
 		cfg.Mode = c.TaskMode
@@ -1053,9 +1059,11 @@ type TaskConfigForDowngrade struct {
 	EnableANSIQuotes        bool                           `yaml:"ansi-quotes"`
 	RemoveMeta              bool                           `yaml:"remove-meta"`
 	// new config item
-	MySQLInstances []*MySQLInstanceForDowngrade `yaml:"mysql-instances"`
-	ExprFilter     map[string]*ExpressionFilter `yaml:"expression-filter,omitempty"`
-	OnlineDDL      bool                         `yaml:"online-ddl,omitempty"`
+	MySQLInstances   []*MySQLInstanceForDowngrade `yaml:"mysql-instances"`
+	ExprFilter       map[string]*ExpressionFilter `yaml:"expression-filter,omitempty"`
+	OnlineDDL        bool                         `yaml:"online-ddl,omitempty"`
+	ShadowTableRules []string                     `yaml:"shadow-table-rules,omitempty"`
+	TrashTableRules  []string                     `yaml:"trash-table-rules,omitempty"`
 }
 
 // NewTaskConfigForDowngrade create new TaskConfigForDowngrade.
@@ -1088,6 +1096,8 @@ func NewTaskConfigForDowngrade(taskConfig *TaskConfig) *TaskConfigForDowngrade {
 		MySQLInstances:          NewMySQLInstancesForDowngrade(taskConfig.MySQLInstances),
 		ExprFilter:              taskConfig.ExprFilter,
 		OnlineDDL:               taskConfig.OnlineDDL,
+		ShadowTableRules:        taskConfig.ShadowTableRules,
+		TrashTableRules:         taskConfig.TrashTableRules,
 	}
 }
 
@@ -1100,6 +1110,12 @@ func (c *TaskConfigForDowngrade) omitDefaultVals() {
 			delete(c.TargetDB.Session, "time_zone")
 		}
 	}
+	if len(c.ShadowTableRules) == 1 && c.ShadowTableRules[0] == DefaultShadowTableRules {
+		c.ShadowTableRules = nil
+	}
+	if len(c.TrashTableRules) == 1 && c.TrashTableRules[0] == DefaultTrashTableRules {
+		c.TrashTableRules = nil
+	}
 }
 
 // Yaml returns YAML format representation of config.

dm/config/task_test.go

@@ -648,6 +648,8 @@ func (t *testConfig) TestGenAndFromSubTaskConfigs(c *C) {
 			IsSharding:              true,
 			ShardMode:               shardMode,
 			OnlineDDL:               onlineDDL,
+			ShadowTableRules:        []string{DefaultShadowTableRules},
+			TrashTableRules:         []string{DefaultTrashTableRules},
 			CaseSensitive:           true,
 			Name:                    name,
 			Mode:                    taskMode,

errors.toml

@@ -1072,6 +1072,12 @@ description = ""
 workaround = "Please check the `chunk-filesize` config in task configuration file."
 tags = ["internal", "high"]
 
+[error.DM-config-20048]
+message = "config '%s' regex pattern '%s' invalid, reason: %s"
+description = ""
+workaround = "Please check if params is correctly in the configuration file."
+tags = ["internal", "high"]
+
 [error.DM-binlog-op-22001]
 message = ""
 description = ""

pkg/terror/error_list.go

@@ -242,6 +242,7 @@ const (
 	codeConfigGenTableRouter
 	codeConfigGenColumnMapping
 	codeConfigInvalidChunkFileSize
+	codeConfigOnlineDDLInvalidRegex
 )
 
 // Binlog operation error code list.
@@ -877,6 +878,8 @@ var (
 	ErrConfigGenTableRouter         = New(codeConfigGenTableRouter, ClassConfig, ScopeInternal, LevelHigh, "generate table router error", "Please check the `routes` config in task configuration file.")
 	ErrConfigGenColumnMapping       = New(codeConfigGenColumnMapping, ClassConfig, ScopeInternal, LevelHigh, "generate column mapping error", "Please check the `column-mappings` config in task configuration file.")
 	ErrConfigInvalidChunkFileSize   = New(codeConfigInvalidChunkFileSize, ClassConfig, ScopeInternal, LevelHigh, "invalid `chunk-filesize` %v", "Please check the `chunk-filesize` config in task configuration file.")
+	ErrConfigOnlineDDLInvalidRegex  = New(codeConfigOnlineDDLInvalidRegex, ClassConfig, ScopeInternal, LevelHigh,
+		"config '%s' regex pattern '%s' invalid, reason: %s", "Please check if params is correctly in the configuration file.")
 
 	// Binlog operation error.
 	ErrBinlogExtractPosition = New(codeBinlogExtractPosition, ClassBinlogOp, ScopeInternal, LevelHigh, "", "")

syncer/job.go

@@ -18,9 +18,9 @@ import (
 	"time"
 
 	"github.com/go-mysql-org/go-mysql/replication"
+	"github.com/pingcap/tidb-tools/pkg/filter"
 
 	"github.com/pingcap/dm/pkg/binlog"
-	"github.com/pingcap/tidb-tools/pkg/filter"
 )
 
 type opType byte

syncer/online-ddl-tools/online_ddl.go

@@ -16,7 +16,7 @@ package onlineddl
 import (
 	"encoding/json"
 	"fmt"
-	"strings"
+	"regexp"
 	"sync"
 
 	"github.com/pingcap/failpoint"
@@ -379,13 +379,33 @@ func (s *Storage) CheckAndUpdate(
 // (_*).*_old ghost trash table
 // we don't support `--new-table-name` flag.
 type RealOnlinePlugin struct {
-	storage *Storage
+	storage    *Storage
+	shadowRegs []*regexp.Regexp
+	trashRegs  []*regexp.Regexp
 }
 
 // NewRealOnlinePlugin returns real online plugin.
 func NewRealOnlinePlugin(tctx *tcontext.Context, cfg *config.SubTaskConfig) (OnlinePlugin, error) {
+	shadowRegs := make([]*regexp.Regexp, 0, len(cfg.ShadowTableRules))
+	trashRegs := make([]*regexp.Regexp, 0, len(cfg.TrashTableRules))
+	for _, sg := range cfg.ShadowTableRules {
+		shadowReg, err := regexp.Compile(sg)
+		if err != nil {
+			return nil, terror.ErrConfigOnlineDDLInvalidRegex.Generate("shadow-table-rules", sg, "fail to compile: "+err.Error())
+		}
+		shadowRegs = append(shadowRegs, shadowReg)
+	}
+	for _, tg := range cfg.TrashTableRules {
+		trashReg, err := regexp.Compile(tg)
+		if err != nil {
+			return nil, terror.ErrConfigOnlineDDLInvalidRegex.Generate("trash-table-rules", tg, "fail to compile: "+err.Error())
+		}
+		trashRegs = append(trashRegs, trashReg)
+	}
 	r := &RealOnlinePlugin{
-		storage: NewOnlineDDLStorage(tcontext.Background().WithLogger(tctx.L().WithFields(zap.String("online ddl", ""))), cfg), // create a context for logger
+		storage:    NewOnlineDDLStorage(tcontext.Background().WithLogger(tctx.L().WithFields(zap.String("online ddl", ""))), cfg), // create a context for logger
+		shadowRegs: shadowRegs,
+		trashRegs:  trashRegs,
 	}
 
 	return r, r.storage.Init(tctx)
@@ -486,12 +506,14 @@ func (r *RealOnlinePlugin) Finish(tctx *tcontext.Context, table *filter.Table) e
 // TableType implements interface.
 func (r *RealOnlinePlugin) TableType(table string) TableType {
 	// 5 is _ _gho/ghc/del or _ _old/new
-	if len(table) > 5 && strings.HasPrefix(table, "_") {
-		if strings.HasSuffix(table, "_gho") || strings.HasSuffix(table, "_new") {
+	for _, shadowReg := range r.shadowRegs {
+		if shadowReg.MatchString(table) {
 			return GhostTable
 		}
+	}
 
-		if strings.HasSuffix(table, "_ghc") || strings.HasSuffix(table, "_del") || strings.HasSuffix(table, "_old") {
+	for _, trashReg := range r.trashRegs {
+		if trashReg.MatchString(table) {
 			return TrashTable
 		}
 	}
@@ -500,9 +522,18 @@ func (r *RealOnlinePlugin) TableType(table string) TableType {
 
 // RealName implements interface.
 func (r *RealOnlinePlugin) RealName(table string) string {
-	tp := r.TableType(table)
-	if tp == GhostTable || tp == TrashTable {
-		table = table[1 : len(table)-4]
+	for _, shadowReg := range r.shadowRegs {
+		shadowRes := shadowReg.FindStringSubmatch(table)
+		if len(shadowRes) > 1 {
+			return shadowRes[1]
+		}
+	}
+
+	for _, trashReg := range r.trashRegs {
+		trashRes := trashReg.FindStringSubmatch(table)
+		if len(trashRes) > 1 {
+			return trashRes[1]
+		}
 	}
 	return table
 }

syncer/safe-mode/mode.go

@@ -16,12 +16,12 @@ package mode
 import (
 	"sync"
 
+	"github.com/pingcap/tidb-tools/pkg/filter"
 	"go.uber.org/zap"
 
 	tcontext "github.com/pingcap/dm/pkg/context"
 	"github.com/pingcap/dm/pkg/terror"
 	"github.com/pingcap/dm/pkg/utils"
-	"github.com/pingcap/tidb-tools/pkg/filter"
 )
 
 // SafeMode controls whether enable safe-mode through a mechanism similar to reference-count

syncer/syncer_test.go

@@ -141,14 +141,16 @@ func (s *testSyncerSuite) SetUpSuite(c *C) {
 		Dir: loaderDir,
 	}
 	s.cfg = &config.SubTaskConfig{
-		From:         config.GetDBConfigFromEnv(),
-		To:           config.GetDBConfigFromEnv(),
-		ServerID:     101,
-		MetaSchema:   "test",
-		Name:         "syncer_ut",
-		Mode:         config.ModeIncrement,
-		Flavor:       "mysql",
-		LoaderConfig: loaderCfg,
+		From:             config.GetDBConfigFromEnv(),
+		To:               config.GetDBConfigFromEnv(),
+		ServerID:         101,
+		MetaSchema:       "test",
+		Name:             "syncer_ut",
+		ShadowTableRules: []string{config.DefaultShadowTableRules},
+		TrashTableRules:  []string{config.DefaultTrashTableRules},
+		Mode:             config.ModeIncrement,
+		Flavor:           "mysql",
+		LoaderConfig:     loaderCfg,
 	}
 	s.cfg.From.Adjust()
 	s.cfg.To.Adjust()

tests/dmctl_basic/conf/get_task.yaml

@@ -63,6 +63,8 @@ mysql-instances:
   syncer: null
   syncer-thread: 0
 online-ddl: false
+shadow-table-rules: []
+trash-table-rules: []
 online-ddl-scheme: ""
 routes:
   route-01:

tests/import_v10x/conf/task.yaml

@@ -57,6 +57,8 @@ mysql-instances:
   syncer: null
   syncer-thread: 0
 online-ddl: false
+shadow-table-rules: []
+trash-table-rules: []
 online-ddl-scheme: ""
 routes: {}
 filters: {}