task.go

// Copyright 2019 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package config

import (
	"encoding/json"
	"flag"
	"fmt"
	"math"
	"os"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"

	"github.com/coreos/go-semver/semver"
	"github.com/docker/go-units"
	"github.com/dustin/go-humanize"
	"github.com/pingcap/tidb/pkg/lightning/config"
	"github.com/pingcap/tidb/pkg/parser"
	"github.com/pingcap/tidb/pkg/sessionctx/variable"
	"github.com/pingcap/tidb/pkg/util/filter"
	router "github.com/pingcap/tidb/pkg/util/table-router"
	"github.com/pingcap/tiflow/dm/config/dbconfig"
	"github.com/pingcap/tiflow/dm/config/security"
	"github.com/pingcap/tiflow/dm/pkg/log"
	"github.com/pingcap/tiflow/dm/pkg/terror"
	"github.com/pingcap/tiflow/dm/pkg/utils"
	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
	"github.com/pingcap/tiflow/pkg/column-mapping"
	"go.uber.org/zap"
	"gopkg.in/yaml.v2"
)

// Online DDL Scheme.
const (
	GHOST = "gh-ost"
	PT    = "pt"
)

// shard DDL mode.
const (
	ShardPessimistic  = "pessimistic"
	ShardOptimistic   = "optimistic"
	tidbTxnMode       = "tidb_txn_mode"
	tidbTxnOptimistic = "optimistic"
)

// collation_compatible.
const (
	LooseCollationCompatible  = "loose"
	StrictCollationCompatible = "strict"
)

const (
	ValidationNone = "none"
	ValidationFast = "fast"
	ValidationFull = "full"

	DefaultValidatorWorkerCount       = 4
	DefaultValidatorValidateInterval  = 10 * time.Second
	DefaultValidatorCheckInterval     = 5 * time.Second
	DefaultValidatorRowErrorDelay     = 30 * time.Minute
	DefaultValidatorMetaFlushInterval = 5 * time.Minute
	DefaultValidatorBatchQuerySize    = 100
	DefaultValidatorMaxPendingRowSize = "500m"

	ValidatorMaxAccumulatedRow = 100000
	// PendingRow is substantial in this version (in sysbench test)
	// set to MaxInt temporaly and reset in the future.
	DefaultValidatorMaxPendingRow = math.MaxInt32
)

// default config item values.
var (
	// TaskConfig.
	defaultMetaSchema          = "dm_meta"
	defaultEnableHeartbeat     = false
	defaultIsSharding          = false
	defaultUpdateInterval      = 1
	defaultReportInterval      = 10
	defaultCollationCompatible = "loose"
	// MydumperConfig.
	defaultMydumperPath  = "./bin/mydumper"
	defaultThreads       = 4
	defaultChunkFilesize = "64"
	defaultSkipTzUTC     = true
	// LoaderConfig.
	defaultPoolSize = 16
	defaultDir      = "./dumped_data"
	// SyncerConfig.
	defaultWorkerCount             = 16
	defaultBatch                   = 100
	defaultQueueSize               = 1024 // do not give too large default value to avoid OOM
	defaultCheckpointFlushInterval = 30   // in seconds
	defaultSafeModeDuration        = strconv.Itoa(2*defaultCheckpointFlushInterval) + "s"

	// TargetDBConfig.
	defaultSessionCfg = []struct {
		key        string
		val        string
		minVersion *semver.Version
	}{
		{tidbTxnMode, tidbTxnOptimistic, semver.New("3.0.0")},
	}
)

// Meta represents binlog's meta pos
// NOTE: refine to put these config structs into pkgs
// NOTE: now, syncer does not support GTID mode and which is supported by relay.
type Meta struct {
	BinLogName string `toml:"binlog-name" yaml:"binlog-name"`
	BinLogPos  uint32 `toml:"binlog-pos" yaml:"binlog-pos"`
	BinLogGTID string `toml:"binlog-gtid" yaml:"binlog-gtid"`
}

// Verify does verification on configs
// NOTE: we can't decide to verify `binlog-name` or `binlog-gtid` until being bound to a source (with `enable-gtid` set).
func (m *Meta) Verify() error {
	if m != nil && len(m.BinLogName) == 0 && len(m.BinLogGTID) == 0 {
		return terror.ErrConfigMetaInvalid.Generate()
	}

	return nil
}

// MySQLInstance represents a sync config of a MySQL instance.
type MySQLInstance struct {
	// it represents a MySQL/MariaDB instance or a replica group
	SourceID    string   `yaml:"source-id"`
	Meta        *Meta    `yaml:"meta"`
	FilterRules []string `yaml:"filter-rules"`
	// deprecated
	ColumnMappingRules []string `yaml:"column-mapping-rules"`
	RouteRules         []string `yaml:"route-rules"`
	ExpressionFilters  []string `yaml:"expression-filters"`

	// black-white-list is deprecated, use block-allow-list instead
	BWListName string `yaml:"black-white-list"`
	BAListName string `yaml:"block-allow-list"`

	MydumperConfigName string          `yaml:"mydumper-config-name"`
	Mydumper           *MydumperConfig `yaml:"mydumper"`
	// MydumperThread is alias for Threads in MydumperConfig, and its priority is higher than Threads
	MydumperThread int `yaml:"mydumper-thread"`

	LoaderConfigName string        `yaml:"loader-config-name"`
	Loader           *LoaderConfig `yaml:"loader"`
	// LoaderThread is alias for PoolSize in LoaderConfig, and its priority is higher than PoolSize
	LoaderThread int `yaml:"loader-thread"`

	SyncerConfigName string        `yaml:"syncer-config-name"`
	Syncer           *SyncerConfig `yaml:"syncer"`
	// SyncerThread is alias for WorkerCount in SyncerConfig, and its priority is higher than WorkerCount
	SyncerThread int `yaml:"syncer-thread"`

	ContinuousValidatorConfigName string          `yaml:"validator-config-name"`
	ContinuousValidator           ValidatorConfig `yaml:"-"`
}

// VerifyAndAdjust does verification on configs, and adjust some configs.
func (m *MySQLInstance) VerifyAndAdjust() error {
	if m == nil {
		return terror.ErrConfigMySQLInstNotFound.Generate()
	}

	if m.SourceID == "" {
		return terror.ErrConfigEmptySourceID.Generate()
	}

	if err := m.Meta.Verify(); err != nil {
		return terror.Annotatef(err, "source %s", m.SourceID)
	}

	if len(m.MydumperConfigName) > 0 && m.Mydumper != nil {
		return terror.ErrConfigMydumperCfgConflict.Generate()
	}
	if len(m.LoaderConfigName) > 0 && m.Loader != nil {
		return terror.ErrConfigLoaderCfgConflict.Generate()
	}
	if len(m.SyncerConfigName) > 0 && m.Syncer != nil {
		return terror.ErrConfigSyncerCfgConflict.Generate()
	}

	if len(m.BAListName) == 0 && len(m.BWListName) != 0 {
		m.BAListName = m.BWListName
	}

	return nil
}

// MydumperConfig represents mydumper process unit's specific config.
type MydumperConfig struct {
	MydumperPath  string `yaml:"mydumper-path" toml:"mydumper-path" json:"mydumper-path"`    // mydumper binary path
	Threads       int    `yaml:"threads" toml:"threads" json:"threads"`                      // -t, --threads
	ChunkFilesize string `yaml:"chunk-filesize" toml:"chunk-filesize" json:"chunk-filesize"` // -F, --chunk-filesize
	StatementSize uint64 `yaml:"statement-size" toml:"statement-size" json:"statement-size"` // -S, --statement-size
	Rows          uint64 `yaml:"rows" toml:"rows" json:"rows"`                               // -r, --rows
	Where         string `yaml:"where" toml:"where" json:"where"`                            // --where

	SkipTzUTC bool   `yaml:"skip-tz-utc" toml:"skip-tz-utc" json:"skip-tz-utc"` // --skip-tz-utc
	ExtraArgs string `yaml:"extra-args" toml:"extra-args" json:"extra-args"`    // other extra args
	// NOTE: use LoaderConfig.Dir as --outputdir
	// TODO zxc: combine -B -T --regex with filter rules?
}

// DefaultMydumperConfig return default mydumper config for task.
func DefaultMydumperConfig() MydumperConfig {
	return MydumperConfig{
		MydumperPath:  defaultMydumperPath,
		Threads:       defaultThreads,
		ChunkFilesize: defaultChunkFilesize,
		SkipTzUTC:     defaultSkipTzUTC,
	}
}

// alias to avoid infinite recursion for UnmarshalYAML.
type rawMydumperConfig MydumperConfig

// UnmarshalYAML implements Unmarshaler.UnmarshalYAML.
func (m *MydumperConfig) UnmarshalYAML(unmarshal func(any) error) error {
	raw := rawMydumperConfig(DefaultMydumperConfig())
	if err := unmarshal(&raw); err != nil {
		return terror.ErrConfigYamlTransform.Delegate(err, "unmarshal mydumper config")
	}
	*m = MydumperConfig(raw) // raw used only internal, so no deep copy
	return nil
}

// LoadMode defines different mode used in load phase.
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 LoadModeLogical mode.
	// deprecated, use LoadModeLogical instead.
	LoadModeLoader LoadMode = "loader"
	// LoadModeLogical means use tidb backend of lightning to load data, which uses SQL to load data.
	LoadModeLogical LoadMode = "logical"
	// LoadModePhysical means use local backend of lightning to load data, which ingest SST files to load data.
	LoadModePhysical LoadMode = "physical"
	// LoadModeImportInto means use import-into backend of lightning to load data, which uses IMPORT INTO SQL statement.
	LoadModeImportInto LoadMode = "import-into"
)

// LogicalDuplicateResolveType defines the duplication resolution when meet duplicate rows for logical import.
type LogicalDuplicateResolveType string

const (
	// OnDuplicateReplace represents replace the old row with new data.
	OnDuplicateReplace LogicalDuplicateResolveType = "replace"
	// OnDuplicateError represents return an error when meet duplicate row.
	OnDuplicateError LogicalDuplicateResolveType = "error"
	// OnDuplicateIgnore represents ignore the new data when meet duplicate row.
	OnDuplicateIgnore LogicalDuplicateResolveType = "ignore"
)

// PhysicalDuplicateResolveType defines the duplication resolution when meet duplicate rows for physical import.
type PhysicalDuplicateResolveType string

const (
	// OnDuplicateNone represents do nothing when meet duplicate row and the task will continue.
	OnDuplicateNone PhysicalDuplicateResolveType = "none"
	// OnDuplicateManual represents that task should be paused when meet duplicate row to let user handle it manually.
	OnDuplicateManual PhysicalDuplicateResolveType = "manual"
)

// PhysicalPostOpLevel defines the configuration of checksum/analyze of physical import.
type PhysicalPostOpLevel string

const (
	OpLevelRequired = "required"
	OpLevelOptional = "optional"
	OpLevelOff      = "off"
)

// LoaderConfig represents loader process unit's specific config.
type LoaderConfig struct {
	PoolSize           int      `yaml:"pool-size" toml:"pool-size" json:"pool-size"`
	Dir                string   `yaml:"dir" toml:"dir" json:"dir"`
	SortingDirPhysical string   `yaml:"sorting-dir-physical" toml:"sorting-dir-physical" json:"sorting-dir-physical"`
	SQLMode            string   `yaml:"-" toml:"-" json:"-"` // wrote by dump unit (DM op) or jobmaster (DM in engine)
	ImportMode         LoadMode `yaml:"import-mode" toml:"import-mode" json:"import-mode"`
	// deprecated, use OnDuplicateLogical instead.
	OnDuplicate         LogicalDuplicateResolveType  `yaml:"on-duplicate" toml:"on-duplicate" json:"on-duplicate"`
	OnDuplicateLogical  LogicalDuplicateResolveType  `yaml:"on-duplicate-logical" toml:"on-duplicate-logical" json:"on-duplicate-logical"`
	OnDuplicatePhysical PhysicalDuplicateResolveType `yaml:"on-duplicate-physical" toml:"on-duplicate-physical" json:"on-duplicate-physical"`
	DiskQuotaPhysical   config.ByteSize              `yaml:"disk-quota-physical" toml:"disk-quota-physical" json:"disk-quota-physical"`
	ChecksumPhysical    PhysicalPostOpLevel          `yaml:"checksum-physical" toml:"checksum-physical" json:"checksum-physical"`
	Analyze             PhysicalPostOpLevel          `yaml:"analyze" toml:"analyze" json:"analyze"`
	RangeConcurrency    int                          `yaml:"range-concurrency" toml:"range-concurrency" json:"range-concurrency"`
	CompressKVPairs     string                       `yaml:"compress-kv-pairs" toml:"compress-kv-pairs" json:"compress-kv-pairs"`
	PDAddr              string                       `yaml:"pd-addr" toml:"pd-addr" json:"pd-addr"`
	// now only creating task by OpenAPI will use the `Security` field to connect PD.
	// TODO: support setting `Security` by dmctl
	Security *security.Security `yaml:"-" toml:"security" json:"security"`
}

// DefaultLoaderConfig return default loader config for task.
func DefaultLoaderConfig() LoaderConfig {
	return LoaderConfig{
		PoolSize:           defaultPoolSize,
		Dir:                defaultDir,
		ImportMode:         LoadModeLogical,
		OnDuplicateLogical: OnDuplicateReplace,
	}
}

// alias to avoid infinite recursion for UnmarshalYAML.
type rawLoaderConfig LoaderConfig

// UnmarshalYAML implements Unmarshaler.UnmarshalYAML.
func (m *LoaderConfig) UnmarshalYAML(unmarshal func(any) error) error {
	raw := rawLoaderConfig(DefaultLoaderConfig())
	if err := unmarshal(&raw); err != nil {
		return terror.ErrConfigYamlTransform.Delegate(err, "unmarshal loader config")
	}
	*m = LoaderConfig(raw) // raw used only internal, so no deep copy
	return nil
}

func (m *LoaderConfig) adjust() error {
	if m.ImportMode == "" {
		m.ImportMode = LoadModeLogical
	}
	if strings.EqualFold(string(m.ImportMode), string(LoadModeSQL)) ||
		strings.EqualFold(string(m.ImportMode), string(LoadModeLoader)) {
		m.ImportMode = LoadModeLogical
	}
	m.ImportMode = LoadMode(strings.ToLower(string(m.ImportMode)))
	switch m.ImportMode {
	case LoadModeLoader, LoadModeSQL, LoadModeLogical, LoadModePhysical, LoadModeImportInto:
	default:
		return terror.ErrConfigInvalidLoadMode.Generate(m.ImportMode)
	}

	if m.PoolSize == 0 {
		m.PoolSize = defaultPoolSize
	}

	if m.OnDuplicateLogical == "" {
		m.OnDuplicateLogical = OnDuplicateReplace
	}
	m.OnDuplicateLogical = LogicalDuplicateResolveType(strings.ToLower(string(m.OnDuplicateLogical)))
	switch m.OnDuplicateLogical {
	case OnDuplicateReplace, OnDuplicateError, OnDuplicateIgnore:
	default:
		return terror.ErrConfigInvalidDuplicateResolution.Generate(m.OnDuplicateLogical)
	}

	if m.OnDuplicatePhysical == "" {
		m.OnDuplicatePhysical = OnDuplicateNone
	}
	m.OnDuplicatePhysical = PhysicalDuplicateResolveType(strings.ToLower(string(m.OnDuplicatePhysical)))
	switch m.OnDuplicatePhysical {
	case OnDuplicateNone, OnDuplicateManual:
	default:
		return terror.ErrConfigInvalidPhysicalDuplicateResolution.Generate(m.OnDuplicatePhysical)
	}

	if m.ChecksumPhysical == "" {
		m.ChecksumPhysical = OpLevelRequired
	}
	m.ChecksumPhysical = PhysicalPostOpLevel(strings.ToLower(string(m.ChecksumPhysical)))
	switch m.ChecksumPhysical {
	case OpLevelRequired, OpLevelOptional, OpLevelOff:
	default:
		return terror.ErrConfigInvalidPhysicalChecksum.Generate(m.ChecksumPhysical)
	}

	if m.Analyze == "" {
		m.Analyze = OpLevelOptional
	}
	m.Analyze = PhysicalPostOpLevel(strings.ToLower(string(m.Analyze)))
	switch m.Analyze {
	case OpLevelRequired, OpLevelOptional, OpLevelOff:
	default:
		return terror.ErrConfigInvalidLoadAnalyze.Generate(m.Analyze)
	}

	return nil
}

// SyncerConfig represents syncer process unit's specific config.
type SyncerConfig struct {
	MetaFile    string `yaml:"meta-file" toml:"meta-file" json:"meta-file"` // meta filename, used only when load SubConfig directly
	WorkerCount int    `yaml:"worker-count" toml:"worker-count" json:"worker-count"`
	Batch       int    `yaml:"batch" toml:"batch" json:"batch"`
	QueueSize   int    `yaml:"queue-size" toml:"queue-size" json:"queue-size"`
	// checkpoint flush interval in seconds.
	CheckpointFlushInterval int `yaml:"checkpoint-flush-interval" toml:"checkpoint-flush-interval" json:"checkpoint-flush-interval"`
	// TODO: add this two new config items for openapi.
	Compact      bool `yaml:"compact" toml:"compact" json:"compact"`
	MultipleRows bool `yaml:"multiple-rows" toml:"multiple-rows" json:"multiple-rows"`

	// deprecated
	MaxRetry int `yaml:"max-retry" toml:"max-retry" json:"max-retry"`

	// deprecated
	AutoFixGTID bool `yaml:"auto-fix-gtid" toml:"auto-fix-gtid" json:"auto-fix-gtid"`
	EnableGTID  bool `yaml:"enable-gtid" toml:"enable-gtid" json:"enable-gtid"`
	// deprecated
	DisableCausality bool   `yaml:"disable-detect" toml:"disable-detect" json:"disable-detect"`
	SafeMode         bool   `yaml:"safe-mode" toml:"safe-mode" json:"safe-mode"`
	SafeModeDuration string `yaml:"safe-mode-duration" toml:"safe-mode-duration" json:"safe-mode-duration"`
	// deprecated, use `ansi-quotes` in top level config instead
	EnableANSIQuotes bool `yaml:"enable-ansi-quotes" toml:"enable-ansi-quotes" json:"enable-ansi-quotes"`
}

// IsForeignKeyChecksEnabled reports whether the downstream session keeps FK checks on.
func IsForeignKeyChecksEnabled(session map[string]string) bool {
	for key, value := range session {
		if strings.EqualFold(key, "foreign_key_checks") {
			trimmed := strings.Trim(value, " '\"")
			return variable.TiDBOptOn(trimmed)
		}
	}
	return false
}

// CheckForeignKeyChecksSyncerOptions rejects syncer options that change DML statement boundaries.
func CheckForeignKeyChecksSyncerOptions(session map[string]string, syncerCfg SyncerConfig) error {
	if !IsForeignKeyChecksEnabled(session) {
		return nil
	}
	if syncerCfg.Compact {
		return terror.ErrConfigUnsupportedForeignKeyChecksOption.Generate("compact")
	}
	if syncerCfg.MultipleRows {
		return terror.ErrConfigUnsupportedForeignKeyChecksOption.Generate("multiple-rows")
	}
	return nil
}

// DefaultSyncerConfig return default syncer config for task.
func DefaultSyncerConfig() SyncerConfig {
	return SyncerConfig{
		WorkerCount:             defaultWorkerCount,
		Batch:                   defaultBatch,
		QueueSize:               defaultQueueSize,
		CheckpointFlushInterval: defaultCheckpointFlushInterval,
		SafeModeDuration:        defaultSafeModeDuration,
	}
}

// alias to avoid infinite recursion for UnmarshalYAML.
type rawSyncerConfig SyncerConfig

// UnmarshalYAML implements Unmarshaler.UnmarshalYAML.
func (m *SyncerConfig) UnmarshalYAML(unmarshal func(any) error) error {
	raw := rawSyncerConfig(DefaultSyncerConfig())
	if err := unmarshal(&raw); err != nil {
		return terror.ErrConfigYamlTransform.Delegate(err, "unmarshal syncer config")
	}
	*m = SyncerConfig(raw) // raw used only internal, so no deep copy
	return nil
}

type ValidatorConfig struct {
	Mode               string   `yaml:"mode" toml:"mode" json:"mode"`
	WorkerCount        int      `yaml:"worker-count" toml:"worker-count" json:"worker-count"`
	ValidateInterval   Duration `yaml:"validate-interval" toml:"validate-interval" json:"validate-interval"`
	CheckInterval      Duration `yaml:"check-interval" toml:"check-interval" json:"check-interval"`
	RowErrorDelay      Duration `yaml:"row-error-delay" toml:"row-error-delay" json:"row-error-delay"`
	MetaFlushInterval  Duration `yaml:"meta-flush-interval" toml:"meta-flush-interval" json:"meta-flush-interval"`
	BatchQuerySize     int      `yaml:"batch-query-size" toml:"batch-query-size" json:"batch-query-size"`
	MaxPendingRowSize  string   `yaml:"max-pending-row-size" toml:"max-pending-row-size" json:"max-pending-row-size"`
	MaxPendingRowCount int      `yaml:"max-pending-row-count" toml:"max-pending-row-count" json:"max-pending-row-count"`
	StartTime          string   `yaml:"-" toml:"start-time" json:"-"`
}

func (v *ValidatorConfig) Adjust() error {
	if v.Mode == "" {
		v.Mode = ValidationNone
	}
	if v.Mode != ValidationNone && v.Mode != ValidationFast && v.Mode != ValidationFull {
		return terror.ErrConfigValidationMode
	}
	if v.WorkerCount <= 0 {
		v.WorkerCount = DefaultValidatorWorkerCount
	}
	if v.ValidateInterval.Duration == 0 {
		v.ValidateInterval.Duration = DefaultValidatorValidateInterval
	}
	if v.CheckInterval.Duration == 0 {
		v.CheckInterval.Duration = DefaultValidatorCheckInterval
	}
	if v.RowErrorDelay.Duration == 0 {
		v.RowErrorDelay.Duration = DefaultValidatorRowErrorDelay
	}
	if v.MetaFlushInterval.Duration == 0 {
		v.MetaFlushInterval.Duration = DefaultValidatorMetaFlushInterval
	}
	if v.BatchQuerySize == 0 {
		v.BatchQuerySize = DefaultValidatorBatchQuerySize
	}
	if v.MaxPendingRowSize == "" {
		v.MaxPendingRowSize = DefaultValidatorMaxPendingRowSize
	}

	_, err := units.RAMInBytes(v.MaxPendingRowSize)
	if err != nil {
		return err
	}
	if v.MaxPendingRowCount == 0 {
		v.MaxPendingRowCount = DefaultValidatorMaxPendingRow
	}
	return nil
}

func defaultValidatorConfig() ValidatorConfig {
	return ValidatorConfig{
		Mode: ValidationNone,
	}
}

// TaskConfig is the configuration for Task.
type TaskConfig struct {
	*flag.FlagSet `yaml:"-" toml:"-" json:"-"`

	Name                      string `yaml:"name" toml:"name" json:"name"`
	TaskMode                  string `yaml:"task-mode" toml:"task-mode" json:"task-mode"`
	IsSharding                bool   `yaml:"is-sharding" toml:"is-sharding" json:"is-sharding"`
	ShardMode                 string `yaml:"shard-mode" toml:"shard-mode" json:"shard-mode"` // when `shard-mode` set, we always enable sharding support.
	StrictOptimisticShardMode bool   `yaml:"strict-optimistic-shard-mode" toml:"strict-optimistic-shard-mode" json:"strict-optimistic-shard-mode"`
	// treat it as hidden configuration
	IgnoreCheckingItems []string `yaml:"ignore-checking-items" toml:"ignore-checking-items" json:"ignore-checking-items"`
	// we store detail status in meta
	// don't save configuration into it
	MetaSchema string `yaml:"meta-schema" toml:"meta-schema" json:"meta-schema"`
	// deprecated
	EnableHeartbeat bool `yaml:"enable-heartbeat" toml:"enable-heartbeat" json:"enable-heartbeat"`
	// deprecated
	HeartbeatUpdateInterval int `yaml:"heartbeat-update-interval" toml:"heartbeat-update-interval" json:"heartbeat-update-interval"`
	// deprecated
	HeartbeatReportInterval int    `yaml:"heartbeat-report-interval" toml:"heartbeat-report-interval" json:"heartbeat-report-interval"`
	Timezone                string `yaml:"timezone" toml:"timezone" json:"timezone"`

	// handle schema/table name mode, and only for schema/table name
	// if case insensitive, we would convert schema/table name to lower case
	CaseSensitive bool `yaml:"case-sensitive" toml:"case-sensitive" json:"case-sensitive"`

	// default "loose" handle create sql by original sql, will not add default collation as upstream
	// "strict" will add default collation as upstream, and downstream will occur error when downstream don't support
	CollationCompatible string `yaml:"collation_compatible" toml:"collation_compatible" json:"collation_compatible"`

	TargetDB *dbconfig.DBConfig `yaml:"target-database" toml:"target-database" json:"target-database"`

	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"`

	Routes  map[string]*router.TableRule   `yaml:"routes" toml:"routes" json:"routes"`
	Filters map[string]*bf.BinlogEventRule `yaml:"filters" toml:"filters" json:"filters"`
	// deprecated
	ColumnMappings map[string]*column.Rule      `yaml:"column-mappings" toml:"column-mappings" json:"column-mappings"`
	ExprFilter     map[string]*ExpressionFilter `yaml:"expression-filter" toml:"expression-filter" json:"expression-filter"`

	// black-white-list is deprecated, use block-allow-list instead
	BWList map[string]*filter.Rules `yaml:"black-white-list" toml:"black-white-list" json:"black-white-list"`
	BAList map[string]*filter.Rules `yaml:"block-allow-list" toml:"block-allow-list" json:"block-allow-list"`

	Mydumpers  map[string]*MydumperConfig  `yaml:"mydumpers" toml:"mydumpers" json:"mydumpers"`
	Loaders    map[string]*LoaderConfig    `yaml:"loaders" toml:"loaders" json:"loaders"`
	Syncers    map[string]*SyncerConfig    `yaml:"syncers" toml:"syncers" json:"syncers"`
	Validators map[string]*ValidatorConfig `yaml:"validators" toml:"validators" json:"validators"`

	CleanDumpFile bool `yaml:"clean-dump-file" toml:"clean-dump-file" json:"clean-dump-file"`
	// deprecated
	EnableANSIQuotes bool `yaml:"ansi-quotes" toml:"ansi-quotes" json:"ansi-quotes"`

	// deprecated, replaced by `start-task --remove-meta`
	RemoveMeta bool `yaml:"remove-meta"`

	// task experimental configs
	Experimental struct {
		AsyncCheckpointFlush bool `yaml:"async-checkpoint-flush" toml:"async-checkpoint-flush" json:"async-checkpoint-flush"`
	} `yaml:"experimental" toml:"experimental" json:"experimental"`
}

// NewTaskConfig creates a TaskConfig.
func NewTaskConfig() *TaskConfig {
	cfg := &TaskConfig{
		// explicitly set default value
		MetaSchema:              defaultMetaSchema,
		EnableHeartbeat:         defaultEnableHeartbeat,
		HeartbeatUpdateInterval: defaultUpdateInterval,
		HeartbeatReportInterval: defaultReportInterval,
		MySQLInstances:          make([]*MySQLInstance, 0, 5),
		IsSharding:              defaultIsSharding,
		Routes:                  make(map[string]*router.TableRule),
		Filters:                 make(map[string]*bf.BinlogEventRule),
		ColumnMappings:          make(map[string]*column.Rule),
		ExprFilter:              make(map[string]*ExpressionFilter),
		BWList:                  make(map[string]*filter.Rules),
		BAList:                  make(map[string]*filter.Rules),
		Mydumpers:               make(map[string]*MydumperConfig),
		Loaders:                 make(map[string]*LoaderConfig),
		Syncers:                 make(map[string]*SyncerConfig),
		Validators:              make(map[string]*ValidatorConfig),
		CleanDumpFile:           true,
		OnlineDDL:               true,
		CollationCompatible:     defaultCollationCompatible,
	}
	cfg.FlagSet = flag.NewFlagSet("task", flag.ContinueOnError)
	return cfg
}

// String returns the config's yaml string.
func (c *TaskConfig) String() string {
	cfg, err := yaml.Marshal(c)
	if err != nil {
		log.L().Error("marshal task config to yaml", zap.String("task", c.Name), log.ShortError(err))
	}
	return string(cfg)
}

// JSON returns the config's json string.
func (c *TaskConfig) JSON() string {
	//nolint:staticcheck
	cfg, err := json.Marshal(c)
	if err != nil {
		log.L().Error("marshal task config to json", zap.String("task", c.Name), log.ShortError(err))
	}
	return string(cfg)
}

// DecodeFile loads and decodes config from file.
func (c *TaskConfig) DecodeFile(fpath string) error {
	bs, err := os.ReadFile(fpath)
	if err != nil {
		return terror.ErrConfigReadCfgFromFile.Delegate(err, fpath)
	}

	err = yaml.UnmarshalStrict(bs, c)
	if err != nil {
		return terror.ErrConfigYamlTransform.Delegate(err)
	}

	return c.adjust()
}

// FromYaml loads config from file data.
func (c *TaskConfig) FromYaml(data string) error {
	err := yaml.UnmarshalStrict([]byte(data), c)
	if err != nil {
		return terror.ErrConfigYamlTransform.Delegate(err, "decode task config failed")
	}

	return c.adjust()
}

// RawDecode loads config from file data.
func (c *TaskConfig) RawDecode(data string) error {
	return terror.ErrConfigYamlTransform.Delegate(yaml.UnmarshalStrict([]byte(data), c), "decode task config failed")
}

// find unused items in config.
var configRefPrefixes = []string{"RouteRules", "FilterRules", "Mydumper", "Loader", "Syncer", "ExprFilter", "Validator"}

const (
	routeRulesIdx = iota
	filterRulesIdx
	mydumperIdx
	loaderIdx
	syncerIdx
	exprFilterIdx
	validatorIdx
)

// Adjust adjusts and verifies config.
func (c *TaskConfig) Adjust() error {
	if c == nil {
		return terror.ErrConfigYamlTransform.New("task config is nil")
	}
	return c.adjust()
}

func (c *TaskConfig) adjust() error {
	if len(c.Name) == 0 {
		return terror.ErrConfigNeedUniqueTaskName.Generate()
	}
	switch c.TaskMode {
	case ModeFull, ModeIncrement, ModeAll, ModeDump, ModeLoad, ModeLoadSync:
	default:
		return terror.ErrConfigInvalidTaskMode.Generate()
	}
	if c.MetaSchema == "" {
		c.MetaSchema = defaultMetaSchema
	}

	if c.ShardMode != "" && c.ShardMode != ShardPessimistic && c.ShardMode != ShardOptimistic {
		return terror.ErrConfigShardModeNotSupport.Generate(c.ShardMode)
	} else if c.ShardMode == "" && c.IsSharding {
		c.ShardMode = ShardPessimistic // use the pessimistic mode as default for back compatible.
	}
	if c.StrictOptimisticShardMode && c.ShardMode != ShardOptimistic {
		return terror.ErrConfigStrictOptimisticShardMode.Generate()
	}

	if len(c.ColumnMappings) > 0 {
		return terror.ErrConfigColumnMappingDeprecated.Generate()
	}

	if c.CollationCompatible != "" && c.CollationCompatible != LooseCollationCompatible && c.CollationCompatible != StrictCollationCompatible {
		return terror.ErrConfigCollationCompatibleNotSupport.Generate(c.CollationCompatible)
	} else if c.CollationCompatible == "" {
		c.CollationCompatible = LooseCollationCompatible
	}

	for _, item := range c.IgnoreCheckingItems {
		if err := ValidateCheckingItem(item); err != nil {
			return err
		}
	}

	if c.OnlineDDLScheme != "" && c.OnlineDDLScheme != PT && c.OnlineDDLScheme != GHOST {
		return terror.ErrConfigOnlineSchemeNotSupport.Generate(c.OnlineDDLScheme)
	} else if c.OnlineDDLScheme == PT || c.OnlineDDLScheme == GHOST {
		c.OnlineDDL = true
		log.L().Warn("'online-ddl-scheme' will be deprecated soon. Recommend that use online-ddl instead of online-ddl-scheme.")
	}

	if c.TargetDB == nil {
		return terror.ErrConfigNeedTargetDB.Generate()
	}

	if len(c.MySQLInstances) == 0 {
		return terror.ErrConfigMySQLInstsAtLeastOne.Generate()
	}

	for name, exprFilter := range c.ExprFilter {
		if exprFilter.Schema == "" {
			return terror.ErrConfigExprFilterEmptyName.Generate(name, "schema")
		}
		if exprFilter.Table == "" {
			return terror.ErrConfigExprFilterEmptyName.Generate(name, "table")
		}
		setFields := make([]string, 0, 1)
		if exprFilter.InsertValueExpr != "" {
			if err := checkValidExpr(exprFilter.InsertValueExpr); err != nil {
				return terror.ErrConfigExprFilterWrongGrammar.Generate(name, exprFilter.InsertValueExpr, err)
			}
			setFields = append(setFields, "insert: ["+exprFilter.InsertValueExpr+"]")
		}
		if exprFilter.UpdateOldValueExpr != "" || exprFilter.UpdateNewValueExpr != "" {
			if exprFilter.UpdateOldValueExpr != "" {
				if err := checkValidExpr(exprFilter.UpdateOldValueExpr); err != nil {
					return terror.ErrConfigExprFilterWrongGrammar.Generate(name, exprFilter.UpdateOldValueExpr, err)
				}
			}
			if exprFilter.UpdateNewValueExpr != "" {
				if err := checkValidExpr(exprFilter.UpdateNewValueExpr); err != nil {
					return terror.ErrConfigExprFilterWrongGrammar.Generate(name, exprFilter.UpdateNewValueExpr, err)
				}
			}
			setFields = append(setFields, "update (old value): ["+exprFilter.UpdateOldValueExpr+"] update (new value): ["+exprFilter.UpdateNewValueExpr+"]")
		}
		if exprFilter.DeleteValueExpr != "" {
			if err := checkValidExpr(exprFilter.DeleteValueExpr); err != nil {
				return terror.ErrConfigExprFilterWrongGrammar.Generate(name, exprFilter.DeleteValueExpr, err)
			}
			setFields = append(setFields, "delete: ["+exprFilter.DeleteValueExpr+"]")
		}
		if len(setFields) > 1 {
			return terror.ErrConfigExprFilterManyExpr.Generate(name, setFields)
		}
	}

	for _, validatorCfg := range c.Validators {
		if err := validatorCfg.Adjust(); err != nil {
			return err
		}
	}

	instanceIDs := make(map[string]int) // source-id -> instance-index
	globalConfigReferCount := map[string]int{}
	duplicateErrorStrings := make([]string, 0)
	for i, inst := range c.MySQLInstances {
		if err := inst.VerifyAndAdjust(); err != nil {
			return terror.Annotatef(err, "mysql-instance: %s", humanize.Ordinal(i))
		}
		if iid, ok := instanceIDs[inst.SourceID]; ok {
			return terror.ErrConfigMySQLInstSameSourceID.Generate(iid, i, inst.SourceID)
		}
		instanceIDs[inst.SourceID] = i

		switch c.TaskMode {
		case ModeFull, ModeAll, ModeDump, ModeLoad:
			if inst.Meta != nil {
				log.L().Warn("metadata will not be used. for Full/Dump/Load mode, incremental sync will never occur; for All mode, the meta dumped by MyDumper will be used", zap.Int("mysql instance", i), zap.String("task mode", c.TaskMode))
			}
		case ModeIncrement:
			if inst.Meta == nil {
				log.L().Warn("mysql-instance doesn't set meta for incremental mode, user should specify start_time to start task.", zap.String("sourceID", inst.SourceID))
			} else {
				err := inst.Meta.Verify()
				if err != nil {
					return terror.Annotatef(err, "mysql-instance: %d", i)
				}
			}
		}

		for _, name := range inst.RouteRules {
			if _, ok := c.Routes[name]; !ok {
				return terror.ErrConfigRouteRuleNotFound.Generate(i, name)
			}
			globalConfigReferCount[configRefPrefixes[routeRulesIdx]+name]++
		}
		for _, name := range inst.FilterRules {
			if _, ok := c.Filters[name]; !ok {
				return terror.ErrConfigFilterRuleNotFound.Generate(i, name)
			}
			globalConfigReferCount[configRefPrefixes[filterRulesIdx]+name]++
		}

		// only when BAList is empty use BWList
		if len(c.BAList) == 0 && len(c.BWList) != 0 {
			c.BAList = c.BWList
		}
		if _, ok := c.BAList[inst.BAListName]; len(inst.BAListName) > 0 && !ok {
			return terror.ErrConfigBAListNotFound.Generate(i, inst.BAListName)
		}

		if len(inst.MydumperConfigName) > 0 {
			rule, ok := c.Mydumpers[inst.MydumperConfigName]
			if !ok {
				return terror.ErrConfigMydumperCfgNotFound.Generate(i, inst.MydumperConfigName)
			}
			globalConfigReferCount[configRefPrefixes[mydumperIdx]+inst.MydumperConfigName]++
			if rule != nil {
				inst.Mydumper = new(MydumperConfig)
				*inst.Mydumper = *rule // ref mydumper config
			}
		}
		if inst.Mydumper == nil {
			if len(c.Mydumpers) != 0 {
				log.L().Warn("mysql instance don't refer mydumper's configuration with mydumper-config-name, the default configuration will be used", zap.String("mysql instance", inst.SourceID))
			}
			defaultCfg := DefaultMydumperConfig()
			inst.Mydumper = &defaultCfg
		} else if inst.Mydumper.ChunkFilesize == "" {
			// avoid too big dump file that can't sent concurrently
			inst.Mydumper.ChunkFilesize = defaultChunkFilesize
		}
		if inst.MydumperThread != 0 {
			inst.Mydumper.Threads = inst.MydumperThread
		}

		if HasDump(c.TaskMode) && len(inst.Mydumper.MydumperPath) == 0 {
			// only verify if set, whether is valid can only be verify when we run it
			return terror.ErrConfigMydumperPathNotValid.Generate(i)
		}

		if len(inst.LoaderConfigName) > 0 {
			rule, ok := c.Loaders[inst.LoaderConfigName]
			if !ok {
				return terror.ErrConfigLoaderCfgNotFound.Generate(i, inst.LoaderConfigName)
			}
			globalConfigReferCount[configRefPrefixes[loaderIdx]+inst.LoaderConfigName]++
			if rule != nil {
				inst.Loader = new(LoaderConfig)
				*inst.Loader = *rule // ref loader config
			}
		}
		if inst.Loader == nil {
			if len(c.Loaders) != 0 {
				log.L().Warn("mysql instance don't refer loader's configuration with loader-config-name, the default configuration will be used", zap.String("mysql instance", inst.SourceID))
			}
			defaultCfg := DefaultLoaderConfig()
			inst.Loader = &defaultCfg
		}
		if inst.LoaderThread != 0 {
			inst.Loader.PoolSize = inst.LoaderThread
		}

		// import-into does not support sharding / multi-source tasks.
		// NOTE: TaskConfig.adjust() runs before generating SubTaskConfig, so we validate it here
		// to avoid multi-source import-into tasks passing admission and failing later at runtime.
		if len(c.MySQLInstances) > 1 && strings.EqualFold(string(inst.Loader.ImportMode), string(LoadModeImportInto)) {
			return terror.ErrConfigImportIntoShardingNotSupport.Generate()
		}

		if len(inst.SyncerConfigName) > 0 {
			rule, ok := c.Syncers[inst.SyncerConfigName]
			if !ok {
				return terror.ErrConfigSyncerCfgNotFound.Generate(i, inst.SyncerConfigName)
			}
			globalConfigReferCount[configRefPrefixes[syncerIdx]+inst.SyncerConfigName]++
			if rule != nil {
				inst.Syncer = new(SyncerConfig)
				*inst.Syncer = *rule // ref syncer config
			}
		}
		if inst.Syncer == nil {
			if len(c.Syncers) != 0 {
				log.L().Warn("mysql instance don't refer syncer's configuration with syncer-config-name, the default configuration will be used", zap.String("mysql instance", inst.SourceID))
			}
			defaultCfg := DefaultSyncerConfig()
			inst.Syncer = &defaultCfg
		}
		if inst.Syncer.QueueSize == 0 {
			inst.Syncer.QueueSize = defaultQueueSize
		}
		if inst.Syncer.CheckpointFlushInterval == 0 {
			inst.Syncer.CheckpointFlushInterval = defaultCheckpointFlushInterval
		}
		if inst.Syncer.SafeModeDuration == "" {
			inst.Syncer.SafeModeDuration = strconv.Itoa(2*inst.Syncer.CheckpointFlushInterval) + "s"
		}
		if duration, err := time.ParseDuration(inst.Syncer.SafeModeDuration); err != nil {
			return terror.ErrConfigInvalidSafeModeDuration.Generate(inst.Syncer.SafeModeDuration, err)
		} else if inst.Syncer.SafeMode && duration == 0 {
			return terror.ErrConfigConfictSafeModeDurationAndSafeMode.Generate()
		}
		if inst.SyncerThread != 0 {
			inst.Syncer.WorkerCount = inst.SyncerThread
		}

		inst.ContinuousValidator = defaultValidatorConfig()
		if inst.ContinuousValidatorConfigName != "" {
			rule, ok := c.Validators[inst.ContinuousValidatorConfigName]
			if !ok {
				return terror.ErrContinuousValidatorCfgNotFound.Generate(i, inst.ContinuousValidatorConfigName)
			}
			globalConfigReferCount[configRefPrefixes[validatorIdx]+inst.ContinuousValidatorConfigName]++
			if rule != nil {
				inst.ContinuousValidator = *rule
			}
		}

		// for backward compatible, set global config `ansi-quotes: true` if any syncer is true
		if inst.Syncer.EnableANSIQuotes {
			log.L().Warn("DM could discover proper ANSI_QUOTES, `enable-ansi-quotes` is no longer take effect")
		}
		if inst.Syncer.DisableCausality {
			log.L().Warn("`disable-causality` is no longer take effect")
		}

		for _, name := range inst.ExpressionFilters {
			if _, ok := c.ExprFilter[name]; !ok {
				return terror.ErrConfigExprFilterNotFound.Generate(i, name)
			}
			globalConfigReferCount[configRefPrefixes[exprFilterIdx]+name]++
		}

		if dupeRules := checkDuplicateString(inst.RouteRules); len(dupeRules) > 0 {
			duplicateErrorStrings = append(duplicateErrorStrings, fmt.Sprintf("mysql-instance(%d)'s route-rules: %s", i, strings.Join(dupeRules, ", ")))
		}
		if dupeRules := checkDuplicateString(inst.FilterRules); len(dupeRules) > 0 {
			duplicateErrorStrings = append(duplicateErrorStrings, fmt.Sprintf("mysql-instance(%d)'s filter-rules: %s", i, strings.Join(dupeRules, ", ")))
		}
		if dupeRules := checkDuplicateString(inst.ColumnMappingRules); len(dupeRules) > 0 {
			duplicateErrorStrings = append(duplicateErrorStrings, fmt.Sprintf("mysql-instance(%d)'s column-mapping-rules: %s", i, strings.Join(dupeRules, ", ")))
		}
		if dupeRules := checkDuplicateString(inst.ExpressionFilters); len(dupeRules) > 0 {
			duplicateErrorStrings = append(duplicateErrorStrings, fmt.Sprintf("mysql-instance(%d)'s expression-filters: %s", i, strings.Join(dupeRules, ", ")))
		}
	}
	if len(duplicateErrorStrings) > 0 {
		return terror.ErrConfigDuplicateCfgItem.Generate(strings.Join(duplicateErrorStrings, "\n"))
	}

	var unusedConfigs []string
	for route := range c.Routes {
		if globalConfigReferCount[configRefPrefixes[routeRulesIdx]+route] == 0 {
			unusedConfigs = append(unusedConfigs, route)
		}
	}
	for filter := range c.Filters {
		if globalConfigReferCount[configRefPrefixes[filterRulesIdx]+filter] == 0 {
			unusedConfigs = append(unusedConfigs, filter)
		}
	}
	for mydumper := range c.Mydumpers {
		if globalConfigReferCount[configRefPrefixes[mydumperIdx]+mydumper] == 0 {
			unusedConfigs = append(unusedConfigs, mydumper)
		}
	}

	for loader, cfg := range c.Loaders {
		if cfg != nil {
			if err1 := cfg.adjust(); err1 != nil {
				return err1
			}
		}
		if globalConfigReferCount[configRefPrefixes[loaderIdx]+loader] == 0 {
			unusedConfigs = append(unusedConfigs, loader)
		}
	}
	for syncer := range c.Syncers {
		if globalConfigReferCount[configRefPrefixes[syncerIdx]+syncer] == 0 {
			unusedConfigs = append(unusedConfigs, syncer)
		}
	}
	for exprFilter := range c.ExprFilter {
		if globalConfigReferCount[configRefPrefixes[exprFilterIdx]+exprFilter] == 0 {
			unusedConfigs = append(unusedConfigs, exprFilter)
		}
	}
	for key := range c.Validators {
		if globalConfigReferCount[configRefPrefixes[validatorIdx]+key] == 0 {
			unusedConfigs = append(unusedConfigs, key)
		}
	}

	if len(unusedConfigs) != 0 {
		sort.Strings(unusedConfigs)
		return terror.ErrConfigGlobalConfigsUnused.Generate(unusedConfigs)
	}

	// we postpone default time_zone init in each unit so we won't change the config value in task/sub_task config
	if c.Timezone != "" {
		if _, err := utils.ParseTimeZone(c.Timezone); err != nil {
			return err
		}
	}
	if c.RemoveMeta {
		log.L().Warn("`remove-meta` in task config is deprecated, please use `start-task ... --remove-meta` instead")
	}

	if c.EnableHeartbeat || c.HeartbeatUpdateInterval != defaultUpdateInterval ||
		c.HeartbeatReportInterval != defaultReportInterval {
		c.EnableHeartbeat = false
		log.L().Warn("heartbeat is deprecated, needn't set it anymore.")
	}
	return nil
}

// getGenerateName generates name by rule or gets name from nameMap
// if it's a new name, increase nameIdx
// otherwise return current nameIdx.
func getGenerateName(rule any, nameIdx int, namePrefix string, nameMap map[string]string) (string, int) {
	// use json as key since no DeepEqual for rules now.
	ruleByte, err := json.Marshal(rule)
	if err != nil {
		log.L().Error(fmt.Sprintf("marshal %s rule to json", namePrefix), log.ShortError(err))
		return fmt.Sprintf("%s-%02d", namePrefix, nameIdx), nameIdx + 1
	} else if val, ok := nameMap[string(ruleByte)]; ok {
		return val, nameIdx
	}
	ruleName := fmt.Sprintf("%s-%02d", namePrefix, nameIdx+1)
	nameMap[string(ruleByte)] = ruleName
	return ruleName, nameIdx + 1
}

// checkDuplicateString checks whether the given string array has duplicate string item
// if there is duplicate, it will return **all** the duplicate strings.
func checkDuplicateString(ruleNames []string) []string {
	mp := make(map[string]bool, len(ruleNames))
	dupeArray := make([]string, 0)
	for _, name := range ruleNames {
		if added, ok := mp[name]; ok {
			if !added {
				dupeArray = append(dupeArray, name)
				mp[name] = true
			}
		} else {
			mp[name] = false
		}
	}
	return dupeArray
}

// AdjustTargetDBSessionCfg adjust session cfg of TiDB.
func AdjustTargetDBSessionCfg(dbConfig *dbconfig.DBConfig, version *semver.Version) {
	lowerMap := make(map[string]string, len(dbConfig.Session))
	for k, v := range dbConfig.Session {
		lowerMap[strings.ToLower(k)] = v
	}
	// all cfg in defaultSessionCfg should be lower case
	for _, cfg := range defaultSessionCfg {
		if _, ok := lowerMap[cfg.key]; !ok && !version.LessThan(*cfg.minVersion) {
			lowerMap[cfg.key] = cfg.val
		}
	}
	dbConfig.Session = lowerMap
}

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
}

// YamlForDowngrade returns YAML format represents of config for downgrade.
func (c *TaskConfig) YamlForDowngrade() (string, error) {
	t := NewTaskConfigForDowngrade(c)

	// try to encrypt password
	t.TargetDB.Password = utils.EncryptOrPlaintext(utils.DecryptOrPlaintext(t.TargetDB.Password))

	// omit default values, so we can ignore them for later marshal
	t.omitDefaultVals()

	return t.Yaml()
}

// MySQLInstanceForDowngrade represents a sync config of a MySQL instance for downgrade.
type MySQLInstanceForDowngrade struct {
	SourceID           string          `yaml:"source-id"`
	Meta               *Meta           `yaml:"meta"`
	FilterRules        []string        `yaml:"filter-rules"`
	ColumnMappingRules []string        `yaml:"column-mapping-rules"`
	RouteRules         []string        `yaml:"route-rules"`
	BWListName         string          `yaml:"black-white-list"`
	BAListName         string          `yaml:"block-allow-list"`
	MydumperConfigName string          `yaml:"mydumper-config-name"`
	Mydumper           *MydumperConfig `yaml:"mydumper"`
	MydumperThread     int             `yaml:"mydumper-thread"`
	LoaderConfigName   string          `yaml:"loader-config-name"`
	Loader             *LoaderConfig   `yaml:"loader"`
	LoaderThread       int             `yaml:"loader-thread"`
	SyncerConfigName   string          `yaml:"syncer-config-name"`
	Syncer             *SyncerConfig   `yaml:"syncer"`
	SyncerThread       int             `yaml:"syncer-thread"`
	// new config item
	ExpressionFilters []string `yaml:"expression-filters,omitempty"`
}

// NewMySQLInstancesForDowngrade creates []* MySQLInstanceForDowngrade.
func NewMySQLInstancesForDowngrade(mysqlInstances []*MySQLInstance) []*MySQLInstanceForDowngrade {
	mysqlInstancesForDowngrade := make([]*MySQLInstanceForDowngrade, 0, len(mysqlInstances))
	for _, m := range mysqlInstances {
		newMySQLInstance := &MySQLInstanceForDowngrade{
			SourceID:           m.SourceID,
			Meta:               m.Meta,
			FilterRules:        m.FilterRules,
			ColumnMappingRules: m.ColumnMappingRules,
			RouteRules:         m.RouteRules,
			BWListName:         m.BWListName,
			BAListName:         m.BAListName,
			MydumperConfigName: m.MydumperConfigName,
			Mydumper:           m.Mydumper,
			MydumperThread:     m.MydumperThread,
			LoaderConfigName:   m.LoaderConfigName,
			Loader:             m.Loader,
			LoaderThread:       m.LoaderThread,
			SyncerConfigName:   m.SyncerConfigName,
			Syncer:             m.Syncer,
			SyncerThread:       m.SyncerThread,
			ExpressionFilters:  m.ExpressionFilters,
		}
		mysqlInstancesForDowngrade = append(mysqlInstancesForDowngrade, newMySQLInstance)
	}
	return mysqlInstancesForDowngrade
}

// LoaderConfigForDowngrade is the base configuration for loader in v2.0.
// This config is used for downgrade(config export) from a higher dmctl version.
// When we add any new config item into LoaderConfig, we should update it also.
type LoaderConfigForDowngrade struct {
	PoolSize int    `yaml:"pool-size" toml:"pool-size" json:"pool-size"`
	Dir      string `yaml:"dir" toml:"dir" json:"dir"`
}

func NewLoaderConfigForDowngrade(loaderConfigs map[string]*LoaderConfig) map[string]*LoaderConfigForDowngrade {
	loaderConfigsForDowngrade := make(map[string]*LoaderConfigForDowngrade, len(loaderConfigs))
	for k, v := range loaderConfigs {
		loaderConfigsForDowngrade[k] = &LoaderConfigForDowngrade{
			PoolSize: v.PoolSize,
			Dir:      v.Dir,
		}
	}
	return loaderConfigsForDowngrade
}

// SyncerConfigForDowngrade is the base configuration for syncer in v2.0.
// This config is used for downgrade(config export) from a higher dmctl version.
// When we add any new config item into SyncerConfig, we should update it also.
type SyncerConfigForDowngrade struct {
	MetaFile                string `yaml:"meta-file"`
	WorkerCount             int    `yaml:"worker-count"`
	Batch                   int    `yaml:"batch"`
	QueueSize               int    `yaml:"queue-size"`
	CheckpointFlushInterval int    `yaml:"checkpoint-flush-interval"`
	MaxRetry                int    `yaml:"max-retry"`
	EnableGTID              bool   `yaml:"enable-gtid"`
	DisableCausality        bool   `yaml:"disable-detect"`
	SafeMode                bool   `yaml:"safe-mode"`
	EnableANSIQuotes        bool   `yaml:"enable-ansi-quotes"`

	SafeModeDuration string `yaml:"safe-mode-duration,omitempty"`
	Compact          bool   `yaml:"compact,omitempty"`
	MultipleRows     bool   `yaml:"multipleRows,omitempty"`
}

// NewSyncerConfigsForDowngrade converts SyncerConfig to SyncerConfigForDowngrade.
func NewSyncerConfigsForDowngrade(syncerConfigs map[string]*SyncerConfig) map[string]*SyncerConfigForDowngrade {
	syncerConfigsForDowngrade := make(map[string]*SyncerConfigForDowngrade, len(syncerConfigs))
	for configName, syncerConfig := range syncerConfigs {
		newSyncerConfig := &SyncerConfigForDowngrade{
			MetaFile:                syncerConfig.MetaFile,
			WorkerCount:             syncerConfig.WorkerCount,
			Batch:                   syncerConfig.Batch,
			QueueSize:               syncerConfig.QueueSize,
			CheckpointFlushInterval: syncerConfig.CheckpointFlushInterval,
			MaxRetry:                syncerConfig.MaxRetry,
			EnableGTID:              syncerConfig.EnableGTID,
			DisableCausality:        syncerConfig.DisableCausality,
			SafeMode:                syncerConfig.SafeMode,
			SafeModeDuration:        syncerConfig.SafeModeDuration,
			EnableANSIQuotes:        syncerConfig.EnableANSIQuotes,
			Compact:                 syncerConfig.Compact,
			MultipleRows:            syncerConfig.MultipleRows,
		}
		syncerConfigsForDowngrade[configName] = newSyncerConfig
	}
	return syncerConfigsForDowngrade
}

// omitDefaultVals change default value to empty value for new config item.
// If any default value for new config item is not empty(0 or false or nil),
// we should change it to empty.
func (c *SyncerConfigForDowngrade) omitDefaultVals() {
	if c.SafeModeDuration == strconv.Itoa(2*c.CheckpointFlushInterval)+"s" {
		c.SafeModeDuration = ""
	}
}

// TaskConfigForDowngrade is the base configuration for task in v2.0.
// This config is used for downgrade(config export) from a higher dmctl version.
// When we add any new config item into SourceConfig, we should update it also.
type TaskConfigForDowngrade struct {
	Name                    string                               `yaml:"name"`
	TaskMode                string                               `yaml:"task-mode"`
	IsSharding              bool                                 `yaml:"is-sharding"`
	ShardMode               string                               `yaml:"shard-mode"`
	IgnoreCheckingItems     []string                             `yaml:"ignore-checking-items"`
	MetaSchema              string                               `yaml:"meta-schema"`
	EnableHeartbeat         bool                                 `yaml:"enable-heartbeat"`
	HeartbeatUpdateInterval int                                  `yaml:"heartbeat-update-interval"`
	HeartbeatReportInterval int                                  `yaml:"heartbeat-report-interval"`
	Timezone                string                               `yaml:"timezone"`
	CaseSensitive           bool                                 `yaml:"case-sensitive"`
	TargetDB                *dbconfig.DBConfig                   `yaml:"target-database"`
	OnlineDDLScheme         string                               `yaml:"online-ddl-scheme"`
	Routes                  map[string]*router.TableRule         `yaml:"routes"`
	Filters                 map[string]*bf.BinlogEventRule       `yaml:"filters"`
	ColumnMappings          map[string]*column.Rule              `yaml:"column-mappings"`
	BWList                  map[string]*filter.Rules             `yaml:"black-white-list"`
	BAList                  map[string]*filter.Rules             `yaml:"block-allow-list"`
	Mydumpers               map[string]*MydumperConfig           `yaml:"mydumpers"`
	Loaders                 map[string]*LoaderConfigForDowngrade `yaml:"loaders"`
	Syncers                 map[string]*SyncerConfigForDowngrade `yaml:"syncers"`
	CleanDumpFile           bool                                 `yaml:"clean-dump-file"`
	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"`
	ShadowTableRules          []string                     `yaml:"shadow-table-rules,omitempty"`
	TrashTableRules           []string                     `yaml:"trash-table-rules,omitempty"`
	StrictOptimisticShardMode bool                         `yaml:"strict-optimistic-shard-mode,omitempty"`
}

// NewTaskConfigForDowngrade create new TaskConfigForDowngrade.
func NewTaskConfigForDowngrade(taskConfig *TaskConfig) *TaskConfigForDowngrade {
	targetDB := *taskConfig.TargetDB
	return &TaskConfigForDowngrade{
		Name:                      taskConfig.Name,
		TaskMode:                  taskConfig.TaskMode,
		IsSharding:                taskConfig.IsSharding,
		ShardMode:                 taskConfig.ShardMode,
		StrictOptimisticShardMode: taskConfig.StrictOptimisticShardMode,
		IgnoreCheckingItems:       taskConfig.IgnoreCheckingItems,
		MetaSchema:                taskConfig.MetaSchema,
		EnableHeartbeat:           taskConfig.EnableHeartbeat,
		HeartbeatUpdateInterval:   taskConfig.HeartbeatUpdateInterval,
		HeartbeatReportInterval:   taskConfig.HeartbeatReportInterval,
		Timezone:                  taskConfig.Timezone,
		CaseSensitive:             taskConfig.CaseSensitive,
		TargetDB:                  &targetDB,
		OnlineDDLScheme:           taskConfig.OnlineDDLScheme,
		Routes:                    taskConfig.Routes,
		Filters:                   taskConfig.Filters,
		ColumnMappings:            taskConfig.ColumnMappings,
		BWList:                    taskConfig.BWList,
		BAList:                    taskConfig.BAList,
		Mydumpers:                 taskConfig.Mydumpers,
		Loaders:                   NewLoaderConfigForDowngrade(taskConfig.Loaders),
		Syncers:                   NewSyncerConfigsForDowngrade(taskConfig.Syncers),
		CleanDumpFile:             taskConfig.CleanDumpFile,
		EnableANSIQuotes:          taskConfig.EnableANSIQuotes,
		RemoveMeta:                taskConfig.RemoveMeta,
		MySQLInstances:            NewMySQLInstancesForDowngrade(taskConfig.MySQLInstances),
		ExprFilter:                taskConfig.ExprFilter,
		OnlineDDL:                 taskConfig.OnlineDDL,
		ShadowTableRules:          taskConfig.ShadowTableRules,
		TrashTableRules:           taskConfig.TrashTableRules,
	}
}

// omitDefaultVals change default value to empty value for new config item.
// If any default value for new config item is not empty(0 or false or nil),
// we should change it to empty.
func (c *TaskConfigForDowngrade) omitDefaultVals() {
	if len(c.ShadowTableRules) == 1 && c.ShadowTableRules[0] == DefaultShadowTableRules {
		c.ShadowTableRules = nil
	}
	if len(c.TrashTableRules) == 1 && c.TrashTableRules[0] == DefaultTrashTableRules {
		c.TrashTableRules = nil
	}
	for _, s := range c.Syncers {
		s.omitDefaultVals()
	}
	c.OnlineDDL = false
}

// Yaml returns YAML format representation of config.
func (c *TaskConfigForDowngrade) Yaml() (string, error) {
	b, err := yaml.Marshal(c)
	return string(b), err
}