lightning(dm): add config value to use local backend (#7448)

ref #3510

dm/_utils/terror_gen/errors_release.txt

@@ -184,7 +184,7 @@ ErrConfigOnlineDDLMistakeRegex,[code=20049:class=config:scope=internal:level=hig
 ErrOpenAPITaskConfigExist,[code=20050:class=config:scope=internal:level=low], "Message: the openapi task config for '%s' already exist, Workaround: If you want to override it, please use the overwrite flag."
 ErrOpenAPITaskConfigNotExist,[code=20051:class=config:scope=internal:level=low], "Message: the openapi task config for '%s' does not exist"
 ErrConfigCollationCompatibleNotSupport,[code=20052:class=config:scope=internal:level=medium], "Message: collation compatible %s not supported, Workaround: Please check the `collation_compatible` config in task configuration file, which can be set to `loose`/`strict`."
-ErrConfigInvalidLoadMode,[code=20053:class=config:scope=internal:level=medium], "Message: invalid load mode '%s', Workaround: Please choose a valid value in ['sql', 'loader']"
+ErrConfigInvalidLoadMode,[code=20053:class=config:scope=internal:level=medium], "Message: invalid load mode '%s', Workaround: Please choose a valid value in ['logical', 'physical']"
 ErrConfigInvalidDuplicateResolution,[code=20054:class=config:scope=internal:level=medium], "Message: invalid load on-duplicate '%s', Workaround: Please choose a valid value in ['replace', 'error', 'ignore']"
 ErrConfigValidationMode,[code=20055:class=config:scope=internal:level=high], "Message: invalid validation mode, Workaround: Please check `validation-mode` config in task configuration file."
 ErrContinuousValidatorCfgNotFound,[code=20056:class=config:scope=internal:level=medium], "Message: mysql-instance(%d)'s continuous validator config %s not exist, Workaround: Please check the `validator-config-name` config in task configuration file."

dm/config/subtask.go

@@ -591,5 +591,6 @@ func (c *SubTaskConfig) Clone() (*SubTaskConfig, error) {
 
 // NeedUseLightning returns whether need to use lightning loader.
 func (c *SubTaskConfig) NeedUseLightning() bool {
-	return (c.Mode == ModeAll || c.Mode == ModeFull) && c.ImportMode == LoadModeSQL
+	// TODO: return true after remove loader
+	return (c.Mode == ModeAll || c.Mode == ModeFull) && c.ImportMode != LoadModeLoader
 }

dm/config/task.go

@@ -22,6 +22,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 
 	"github.com/coreos/go-semver/semver"
@@ -238,9 +239,15 @@ type LoadMode string
 
 const (
 	// LoadModeSQL means write data by sql statements, uses tidb-lightning tidb backend to load data.
+	// deprecated, use LoadModeLogical instead.
 	LoadModeSQL LoadMode = "sql"
 	// LoadModeLoader is the legacy sql mode, use loader to load data. this should be replaced by sql mode in new version.
+	// deprecated, loader will be removed in future.
 	LoadModeLoader = "loader"
+	// LoadModeLogical means use tidb backend of lightning to load data, which uses SQL to load data.
+	LoadModeLogical = "logical"
+	// LoadModePhysical means use local backend of lightning to load data, which ingest SST files to load data.
+	LoadModePhysical = "physical"
 )
 
 // DuplicateResolveType defines the duplication resolution when meet duplicate rows.
@@ -289,10 +296,15 @@ func (m *LoaderConfig) UnmarshalYAML(unmarshal func(interface{}) error) error {
 
 func (m *LoaderConfig) adjust() error {
 	if m.ImportMode == "" {
-		m.ImportMode = LoadModeSQL
+		m.ImportMode = LoadModeLogical
+	}
+	if strings.EqualFold(string(m.ImportMode), string(LoadModeSQL)) {
+		m.ImportMode = LoadModeLogical
 	}
 	m.ImportMode = LoadMode(strings.ToLower(string(m.ImportMode)))
-	if m.ImportMode != LoadModeSQL && m.ImportMode != LoadModeLoader {
+	switch m.ImportMode {
+	case LoadModeLoader, LoadModeSQL, LoadModeLogical, LoadModePhysical:
+	default:
 		return terror.ErrConfigInvalidLoadMode.Generate(m.ImportMode)
 	}
 
@@ -997,11 +1009,16 @@ func AdjustDBTimeZone(config *DBConfig, timeZone string) {
 	config.Session["time_zone"] = timeZone
 }
 
-var defaultParser = parser.New()
+var (
+	defaultParser = parser.New()
+	parserMu      sync.Mutex
+)
 
 func checkValidExpr(expr string) error {
 	expr = "select " + expr
+	parserMu.Lock()
 	_, _, err := defaultParser.Parse(expr, "", "")
+	parserMu.Unlock()
 	return err
 }