Repository: pingcap/tiflow
Pull request #10021 (2023-11-07 01:01:36)

dm: reduce gSet.String() usage by using zeroGSet for checking empty (#9944)
This is an automated cherry-pick of #9944

<!--
Thank you for contributing to TiFlow! 
Please read MD's [CONTRIBUTING](https://github.com/pingcap/tiflow/blob/master/CONTRIBUTING.md) document **BEFORE** filing this PR.
-->

### What problem does this PR solve?
<!--
Please create an issue first to describe the problem.

There MUST be one line starting with "Issue Number:  " and 
linking the relevant issues via the "close" or "ref".

For more info, check https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue.
 -->

Issue Number: close https://github.com/pingcap/tiflow/issues/9676

### What is changed and how it works?
reduce gSet.String() usage by using zeroGSet for checking empty

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test

#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
None
```


ti-chi-bot[bot]: [APPROVALNOTIFIER] This PR is **NOT APPROVED**

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please assign [5kbpers](https://github.com/5kbpers) for approval. For more information see [the Kubernetes Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

The full list of commands accepted by this bot can be found [here](https://prow.tidb.net/command-help?repo=pingcap%2Ftiflow).

<details open>
Needs approval from an approver in each of these files:

- **[OWNERS](https://github.com/pingcap/tiflow/blob/release-7.5/OWNERS)**

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["5kbpers"]} -->
ti-chi-bot[bot]: This cherry pick PR is for a release branch and has not yet been approved by triage owners. 
Adding the `do-not-merge/cherry-pick-not-approved` label.

To merge this cherry pick:
1. It must be approved by the approvers firstly.
2. **AFTER** it has been approved by approvers, please wait for the cherry-pick merging approval from triage owners.


<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository.
</details>
ti-chi-bot[bot]: @ti-chi-bot: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 580f2df6c1f5b2877bc58bd330114ba07ad46fb7 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/ghpr_verify/41/display/redirect) | true | `/test verify`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=10021). [Your PR dashboard](https://prow.tidb.net/pr?query=is%3Apr%20state%3Aopen%20author%3Ati-chi-bot).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
ti-chi-bot[bot]: @ti-chi-bot: The following tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
jenkins-ticdc/verify | 580f2df6c1f5b2877bc58bd330114ba07ad46fb7 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/ghpr_verify/41/display/redirect) | true | `/test verify`
pull-dm-integration-test | 580f2df6c1f5b2877bc58bd330114ba07ad46fb7 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_dm_integration_test/41/display/redirect) | true | `/test dm-integration-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=10021). [Your PR dashboard](https://prow.tidb.net/pr?query=is%3Apr%20state%3Aopen%20author%3Ati-chi-bot).

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/cdc/api/v2/api_helpers.go b/cdc/api/v2/api_helpers.go
index f84037747..aad7294f4 100644
--- a/cdc/api/v2/api_helpers.go
+++ b/cdc/api/v2/api_helpers.go
@@ -30,6 +30,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/owner"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/transformer/columnselector"
 	"github.com/pingcap/tiflow/cdc/sink/validator"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -518,11 +519,19 @@ func (h APIV2HelpersImpl) getVerifiedTables(
 	if err != nil {
 		return nil, nil, err
 	}
-
 	err = eventRouter.VerifyTables(tableInfos)
 	if err != nil {
 		return nil, nil, err
 	}
 
+	selectors, err := columnselector.New(replicaConfig)
+	if err != nil {
+		return nil, nil, err
+	}
+	err = selectors.VerifyTables(tableInfos, eventRouter)
+	if err != nil {
+		return nil, nil, err
+	}
+
 	return ineligibleTables, eligibleTables, nil
 }
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
index af9887b02..05a7ba86a 100644
--- a/cdc/api/v2/model.go
+++ b/cdc/api/v2/model.go
@@ -193,7 +193,8 @@ type ReplicaConfig struct {
 	Consistent                   *ConsistentConfig          `json:"consistent,omitempty"`
 	Scheduler                    *ChangefeedSchedulerConfig `json:"scheduler"`
 	Integrity                    *IntegrityConfig           `json:"integrity"`
-	ChangefeedErrorStuckDuration *JSONDuration              `json:"changefeed_error_stuck_duration,omitempty" swaggertype:"string"`
+	ChangefeedErrorStuckDuration *JSONDuration              `json:"changefeed_error_stuck_duration,omitempty"`
+	SQLMode                      string                     `json:"sql_mode,omitempty"`
 }
 
 // ToInternalReplicaConfig coverts *v2.ReplicaConfig into *config.ReplicaConfig
@@ -211,6 +212,7 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 	res.CheckGCSafePoint = c.CheckGCSafePoint
 	res.EnableSyncPoint = c.EnableSyncPoint
 	res.IgnoreIneligibleTable = c.IgnoreIneligibleTable
+	res.SQLMode = c.SQLMode
 	if c.SyncPointInterval != nil {
 		res.SyncPointInterval = &c.SyncPointInterval.duration
 	}
@@ -260,11 +262,14 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 	}
 	if c.Consistent != nil {
 		res.Consistent = &config.ConsistentConfig{
-			Level:             c.Consistent.Level,
-			MaxLogSize:        c.Consistent.MaxLogSize,
-			FlushIntervalInMs: c.Consistent.FlushIntervalInMs,
-			Storage:           c.Consistent.Storage,
-			UseFileBackend:    c.Consistent.UseFileBackend,
+			Level:                 c.Consistent.Level,
+			MaxLogSize:            c.Consistent.MaxLogSize,
+			FlushIntervalInMs:     c.Consistent.FlushIntervalInMs,
+			MetaFlushIntervalInMs: c.Consistent.MetaFlushIntervalInMs,
+			EncodingWorkerNum:     c.Consistent.EncodingWorkerNum,
+			FlushWorkerNum:        c.Consistent.FlushWorkerNum,
+			Storage:               c.Consistent.Storage,
+			UseFileBackend:        c.Consistent.UseFileBackend,
 		}
 	}
 	if c.Sink != nil {
@@ -497,6 +502,7 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 		CheckGCSafePoint:      cloned.CheckGCSafePoint,
 		EnableSyncPoint:       cloned.EnableSyncPoint,
 		BDRMode:               cloned.BDRMode,
+		SQLMode:               cloned.SQLMode,
 	}
 
 	if cloned.SyncPointInterval != nil {
@@ -741,11 +747,14 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 	}
 	if cloned.Consistent != nil {
 		res.Consistent = &ConsistentConfig{
-			Level:             cloned.Consistent.Level,
-			MaxLogSize:        cloned.Consistent.MaxLogSize,
-			FlushIntervalInMs: cloned.Consistent.FlushIntervalInMs,
-			Storage:           cloned.Consistent.Storage,
-			UseFileBackend:    cloned.Consistent.UseFileBackend,
+			Level:                 cloned.Consistent.Level,
+			MaxLogSize:            cloned.Consistent.MaxLogSize,
+			FlushIntervalInMs:     cloned.Consistent.FlushIntervalInMs,
+			MetaFlushIntervalInMs: cloned.Consistent.MetaFlushIntervalInMs,
+			EncodingWorkerNum:     c.Consistent.EncodingWorkerNum,
+			FlushWorkerNum:        c.Consistent.FlushWorkerNum,
+			Storage:               cloned.Consistent.Storage,
+			UseFileBackend:        cloned.Consistent.UseFileBackend,
 		}
 	}
 	if cloned.Mounter != nil {
@@ -933,11 +942,14 @@ type ColumnSelector struct {
 // ConsistentConfig represents replication consistency config for a changefeed
 // This is a duplicate of config.ConsistentConfig
 type ConsistentConfig struct {
-	Level             string `json:"level,omitempty"`
-	MaxLogSize        int64  `json:"max_log_size"`
-	FlushIntervalInMs int64  `json:"flush_interval"`
-	Storage           string `json:"storage,omitempty"`
-	UseFileBackend    bool   `json:"use_file_backend"`
+	Level                 string `json:"level,omitempty"`
+	MaxLogSize            int64  `json:"max_log_size"`
+	FlushIntervalInMs     int64  `json:"flush_interval"`
+	MetaFlushIntervalInMs int64  `json:"meta_flush_interval"`
+	EncodingWorkerNum     int    `json:"encoding_worker_num"`
+	FlushWorkerNum        int    `json:"flush_worker_num"`
+	Storage               string `json:"storage,omitempty"`
+	UseFileBackend        bool   `json:"use_file_backend"`
 }
 
 // ChangefeedSchedulerConfig is per changefeed scheduler settings.
diff --git a/cdc/model/changefeed.go b/cdc/model/changefeed.go
index fec7bbed7..fbf942595 100644
--- a/cdc/model/changefeed.go
+++ b/cdc/model/changefeed.go
@@ -334,6 +334,9 @@ func (info *ChangeFeedInfo) VerifyAndComplete() {
 	if info.Config.ChangefeedErrorStuckDuration == nil {
 		info.Config.ChangefeedErrorStuckDuration = defaultConfig.ChangefeedErrorStuckDuration
 	}
+	if info.Config.SQLMode == "" {
+		info.Config.SQLMode = defaultConfig.SQLMode
+	}
 	info.RmUnusedFields()
 }
 
diff --git a/cdc/model/schema_storage.go b/cdc/model/schema_storage.go
index 26d904bba..a4c50b09f 100644
--- a/cdc/model/schema_storage.go
+++ b/cdc/model/schema_storage.go
@@ -341,9 +341,9 @@ func (ti *TableInfo) IndexByName(name string) ([]string, []int, bool) {
 	return names, offset, true
 }
 
-// ColumnsByNames returns the column offsets of the corresponding columns by names
+// OffsetsByNames returns the column offsets of the corresponding columns by names
 // If any column does not exist, return false
-func (ti *TableInfo) ColumnsByNames(names []string) ([]int, bool) {
+func (ti *TableInfo) OffsetsByNames(names []string) ([]int, bool) {
 	// todo: optimize it
 	columnOffsets := make(map[string]int, len(ti.Columns))
 	for _, col := range ti.Columns {
@@ -363,3 +363,26 @@ func (ti *TableInfo) ColumnsByNames(names []string) ([]int, bool) {
 
 	return result, true
 }
+
+// GetPrimaryKeyColumnNames returns the primary key column names
+func (ti *TableInfo) GetPrimaryKeyColumnNames() map[string]struct{} {
+	result := make(map[string]struct{})
+	for _, index := range ti.Indices {
+		if index.Primary {
+			for _, col := range index.Columns {
+				result[col.Name.O] = struct{}{}
+			}
+			return result
+		}
+	}
+
+	for _, columnsOffsets := range ti.IndexColumnsOffset {
+		for _, offset := range columnsOffsets {
+			columnInfo := ti.Columns[offset]
+			if mysql.HasPriKeyFlag(columnInfo.FieldType.GetFlag()) {
+				result[columnInfo.Name.O] = struct{}{}
+			}
+		}
+	}
+	return result
+}
diff --git a/cdc/owner/ddl_sink.go b/cdc/owner/ddl_sink.go
index 5bdb8a8ca..f06d01e6f 100644
--- a/cdc/owner/ddl_sink.go
+++ b/cdc/owner/ddl_sink.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/parser"
 	"github.com/pingcap/tidb/parser/format"
+	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/ddlsink"
 	"github.com/pingcap/tiflow/cdc/sink/ddlsink/factory"
@@ -426,7 +427,18 @@ func (s *ddlSinkImpl) close(ctx context.Context) (err error) {
 
 // addSpecialComment translate tidb feature to comment
 func (s *ddlSinkImpl) addSpecialComment(ddl *model.DDLEvent) (string, error) {
-	stms, _, err := parser.New().Parse(ddl.Query, ddl.Charset, ddl.Collate)
+	p := parser.New()
+	// We need to use the correct SQL mode to parse the DDL query.
+	// Otherwise, the parser may fail to parse the DDL query.
+	// For example, it is needed to parse the following DDL query:
+	//  `alter table "t" add column "c" int default 1;`
+	// by adding `ANSI_QUOTES` to the SQL mode.
+	mode, err := mysql.GetSQLMode(s.info.Config.SQLMode)
+	if err != nil {
+		return "", errors.Trace(err)
+	}
+	p.SetSQLMode(mode)
+	stms, _, err := p.Parse(ddl.Query, ddl.Charset, ddl.Collate)
 	if err != nil {
 		return "", errors.Trace(err)
 	}
diff --git a/cdc/redo/manager.go b/cdc/redo/manager.go
index 1c6f2bd36..b06c6e3d1 100644
--- a/cdc/redo/manager.go
+++ b/cdc/redo/manager.go
@@ -274,7 +274,15 @@ func (m *logManager) Run(ctx context.Context, _ ...chan<- error) error {
 		return err
 	}
 	m.writer = w
-	return m.bgUpdateLog(ctx)
+	return m.bgUpdateLog(ctx, m.getFlushDuration())
+}
+
+func (m *logManager) getFlushDuration() time.Duration {
+	flushIntervalInMs := m.cfg.FlushIntervalInMs
+	if m.cfg.LogType == redo.RedoDDLLogFileType {
+		flushIntervalInMs = m.cfg.MetaFlushIntervalInMs
+	}
+	return time.Duration(flushIntervalInMs) * time.Millisecond
 }
 
 // WaitForReady implements pkg/util.Runnable.
@@ -427,6 +435,7 @@ func (m *logManager) flushLog(
 		log.Debug("Flush redo log",
 			zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
 			zap.String("changefeed", m.cfg.ChangeFeedID.ID),
+			zap.String("logType", m.cfg.LogType),
 			zap.Any("tableRtsMap", tableRtsMap))
 		err := m.withLock(func(m *logManager) error {
 			return m.writer.FlushLog(ctx)
@@ -483,15 +492,17 @@ func (m *logManager) onResolvedTsMsg(span tablepb.Span, resolvedTs model.Ts) {
 	}
 }
 
-func (m *logManager) bgUpdateLog(ctx context.Context) error {
+func (m *logManager) bgUpdateLog(ctx context.Context, flushDuration time.Duration) error {
 	m.releaseMemoryCbs = make([]func(), 0, 1024)
-	flushIntervalInMs := m.cfg.FlushIntervalInMs
-	ticker := time.NewTicker(time.Duration(flushIntervalInMs) * time.Millisecond)
+	ticker := time.NewTicker(flushDuration)
 	defer ticker.Stop()
 	log.Info("redo manager bgUpdateLog is running",
 		zap.String("namespace", m.cfg.ChangeFeedID.Namespace),
 		zap.String("changefeed", m.cfg.ChangeFeedID.ID),
-		zap.Int64("flushIntervalInMs", flushIntervalInMs))
+		zap.Duration("flushIntervalInMs", flushDuration),
+		zap.Int64("maxLogSize", m.cfg.MaxLogSize),
+		zap.Int("encoderWorkerNum", m.cfg.EncodingWorkerNum),
+		zap.Int("flushWorkerNum", m.cfg.FlushWorkerNum))
 
 	var err error
 	// logErrCh is used to retrieve errors from log flushing goroutines.
diff --git a/cdc/redo/meta_manager.go b/cdc/redo/meta_manager.go
index f1ad1acac..5fc437c65 100644
--- a/cdc/redo/meta_manager.go
+++ b/cdc/redo/meta_manager.go
@@ -74,6 +74,8 @@ type metaManager struct {
 	lastFlushTime          time.Time
 	cfg                    *config.ConsistentConfig
 	metricFlushLogDuration prometheus.Observer
+
+	flushIntervalInMs int64
 }
 
 // NewDisabledMetaManager creates a disabled Meta Manager.
@@ -93,12 +95,19 @@ func NewMetaManager(
 	}
 
 	m := &metaManager{
-		captureID:     config.GetGlobalServerConfig().AdvertiseAddr,
-		changeFeedID:  changefeedID,
-		uuidGenerator: uuid.NewGenerator(),
-		enabled:       true,
-		cfg:           cfg,
-		startTs:       checkpoint,
+		captureID:         config.GetGlobalServerConfig().AdvertiseAddr,
+		changeFeedID:      changefeedID,
+		uuidGenerator:     uuid.NewGenerator(),
+		enabled:           true,
+		cfg:               cfg,
+		startTs:           checkpoint,
+		flushIntervalInMs: cfg.MetaFlushIntervalInMs,
+	}
+
+	if m.flushIntervalInMs < redo.MinFlushIntervalInMs {
+		log.Warn("redo flush interval is too small, use default value",
+			zap.Int64("interval", m.flushIntervalInMs))
+		m.flushIntervalInMs = redo.DefaultMetaFlushIntervalInMs
 	}
 	return m
 }
@@ -157,7 +166,7 @@ func (m *metaManager) Run(ctx context.Context, _ ...chan<- error) error {
 	}
 	eg, egCtx := errgroup.WithContext(ctx)
 	eg.Go(func() error {
-		return m.bgFlushMeta(egCtx, m.cfg.FlushIntervalInMs)
+		return m.bgFlushMeta(egCtx)
 	})
 	eg.Go(func() error {
 		return m.bgGC(egCtx)
@@ -458,8 +467,8 @@ func (m *metaManager) Cleanup(ctx context.Context) error {
 	return m.deleteAllLogs(ctx)
 }
 
-func (m *metaManager) bgFlushMeta(egCtx context.Context, flushIntervalInMs int64) (err error) {
-	ticker := time.NewTicker(time.Duration(flushIntervalInMs) * time.Millisecond)
+func (m *metaManager) bgFlushMeta(egCtx context.Context) (err error) {
+	ticker := time.NewTicker(time.Duration(m.flushIntervalInMs) * time.Millisecond)
 	defer func() {
 		ticker.Stop()
 		log.Info("redo metaManager bgFlushMeta exits",
diff --git a/cdc/redo/writer/memory/encoding_worker.go b/cdc/redo/writer/memory/encoding_worker.go
index 57b15ce45..bd878f65b 100644
--- a/cdc/redo/writer/memory/encoding_worker.go
+++ b/cdc/redo/writer/memory/encoding_worker.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/model/codec"
 	"github.com/pingcap/tiflow/cdc/redo/writer"
 	"github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/redo"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
@@ -107,15 +108,15 @@ type encodingWorkerGroup struct {
 
 func newEncodingWorkerGroup(workerNum int) *encodingWorkerGroup {
 	if workerNum <= 0 {
-		workerNum = defaultEncodingWorkerNum
+		workerNum = redo.DefaultEncodingWorkerNum
 	}
 	inputChs := make([]chan *polymorphicRedoEvent, workerNum)
 	for i := 0; i < workerNum; i++ {
-		inputChs[i] = make(chan *polymorphicRedoEvent, defaultEncodingInputChanSize)
+		inputChs[i] = make(chan *polymorphicRedoEvent, redo.DefaultEncodingInputChanSize)
 	}
 	return &encodingWorkerGroup{
 		inputChs:  inputChs,
-		outputCh:  make(chan *polymorphicRedoEvent, defaultEncodingOutputChanSize),
+		outputCh:  make(chan *polymorphicRedoEvent, redo.DefaultEncodingOutputChanSize),
 		workerNum: workerNum,
 		closed:    make(chan struct{}),
 	}
diff --git a/cdc/redo/writer/memory/file_worker.go b/cdc/redo/writer/memory/file_worker.go
index 45b366117..3258bf169 100644
--- a/cdc/redo/writer/memory/file_worker.go
+++ b/cdc/redo/writer/memory/file_worker.go
@@ -103,7 +103,7 @@ func newFileWorkerGroup(
 	opts ...writer.Option,
 ) *fileWorkerGroup {
 	if workerNum <= 0 {
-		workerNum = defaultFlushWorkerNum
+		workerNum = redo.DefaultFlushWorkerNum
 	}
 
 	op := &writer.LogWriterOptions{}
diff --git a/cdc/redo/writer/memory/mem_log_writer.go b/cdc/redo/writer/memory/mem_log_writer.go
index 0460fdbf0..25e968c51 100644
--- a/cdc/redo/writer/memory/mem_log_writer.go
+++ b/cdc/redo/writer/memory/mem_log_writer.go
@@ -24,15 +24,6 @@ import (
 	"golang.org/x/sync/errgroup"
 )
 
-var (
-	defaultEncodingWorkerNum      = 16
-	defaultEncodingInputChanSize  = 128
-	defaultEncodingOutputChanSize = 2048
-	// Maximum allocated memory is flushWorkerNum*maxLogSize, which is
-	// `8*64MB = 512MB` by default.
-	defaultFlushWorkerNum = 8
-)
-
 var _ writer.RedoLogWriter = (*memoryLogWriter)(nil)
 
 type memoryLogWriter struct {
@@ -71,11 +62,11 @@ func NewLogWriter(
 		cancel: lwCancel,
 	}
 
-	lw.encodeWorkers = newEncodingWorkerGroup(defaultEncodingWorkerNum)
+	lw.encodeWorkers = newEncodingWorkerGroup(cfg.EncodingWorkerNum)
 	eg.Go(func() error {
 		return lw.encodeWorkers.Run(lwCtx)
 	})
-	lw.fileWorkers = newFileWorkerGroup(cfg, defaultFlushWorkerNum, extStorage, opts...)
+	lw.fileWorkers = newFileWorkerGroup(cfg, cfg.FlushWorkerNum, extStorage, opts...)
 	eg.Go(func() error {
 		return lw.fileWorkers.Run(lwCtx, lw.encodeWorkers.outputCh)
 	})
diff --git a/cdc/scheduler/internal/v3/replication/metrics.go b/cdc/scheduler/internal/v3/replication/metrics.go
index ccd546f67..a427035ef 100644
--- a/cdc/scheduler/internal/v3/replication/metrics.go
+++ b/cdc/scheduler/internal/v3/replication/metrics.go
@@ -125,6 +125,21 @@ var (
 			Name:      "slow_table_region_count",
 			Help:      "The number of regions captured by the slowest table",
 		}, []string{"namespace", "changefeed"})
+
+	slowestTablePullerResolvedTs = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "ticdc",
+			Subsystem: "scheduler",
+			Name:      "slow_table_puller_resolved_ts",
+			Help:      "Puller Slowest ResolvedTs",
+		}, []string{"namespace", "changefeed"})
+	slowestTablePullerResolvedTsLag = prometheus.NewGaugeVec(
+		prometheus.GaugeOpts{
+			Namespace: "ticdc",
+			Subsystem: "scheduler",
+			Name:      "slow_table_puller_resolved_ts_lag",
+			Help:      "Puller Slowest ResolvedTs lag",
+		}, []string{"namespace", "changefeed"})
 )
 
 // InitMetrics registers all metrics used in scheduler
@@ -144,4 +159,7 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(slowestTableStageCheckpointTsLagHistogramVec)
 	registry.MustRegister(slowestTableStageResolvedTsLagHistogramVec)
 	registry.MustRegister(slowestTableRegionGaugeVec)
+
+	registry.MustRegister(slowestTablePullerResolvedTs)
+	registry.MustRegister(slowestTablePullerResolvedTsLag)
 }
diff --git a/cdc/scheduler/internal/v3/replication/replication_manager.go b/cdc/scheduler/internal/v3/replication/replication_manager.go
index 0ec7c7a67..4228cc0ec 100644
--- a/cdc/scheduler/internal/v3/replication/replication_manager.go
+++ b/cdc/scheduler/internal/v3/replication/replication_manager.go
@@ -147,7 +147,8 @@ type Manager struct { //nolint:revive
 	maxTaskConcurrency int
 
 	changefeedID           model.ChangeFeedID
-	slowestTableID         tablepb.Span
+	slowestPuller          tablepb.Span
+	slowestSink            tablepb.Span
 	acceptAddTableTask     int
 	acceptRemoveTableTask  int
 	acceptMoveTableTask    int
@@ -589,14 +590,16 @@ func (r *Manager) AdvanceCheckpoint(
 		}
 	}()
 
+	r.slowestPuller = tablepb.Span{}
+	r.slowestSink = tablepb.Span{}
+	var slowestPullerResolvedTs uint64 = math.MaxUint64
+
 	newCheckpointTs, newResolvedTs = math.MaxUint64, math.MaxUint64
-	slowestRange := tablepb.Span{}
 	cannotProceed := false
-	lastSpan := tablepb.Span{}
 	currentTables.Iter(func(tableID model.TableID, tableStart, tableEnd tablepb.Span) bool {
 		tableSpanFound, tableHasHole := false, false
 		tableSpanStartFound, tableSpanEndFound := false, false
-		lastSpan = tablepb.Span{}
+		lastSpan := tablepb.Span{}
 		r.spans.AscendRange(tableStart, tableEnd,
 			func(span tablepb.Span, table *ReplicationSet) bool {
 				if lastSpan.TableID != 0 && !bytes.Equal(lastSpan.EndKey, span.StartKey) {
@@ -620,11 +623,19 @@ func (r *Manager) AdvanceCheckpoint(
 				// Find the minimum checkpoint ts and resolved ts.
 				if newCheckpointTs > table.Checkpoint.CheckpointTs {
 					newCheckpointTs = table.Checkpoint.CheckpointTs
-					slowestRange = span
+					r.slowestSink = span
 				}
 				if newResolvedTs > table.Checkpoint.ResolvedTs {
 					newResolvedTs = table.Checkpoint.ResolvedTs
 				}
+
+				// Find the minimum puller resolved ts.
+				if pullerCkpt, ok := table.Stats.StageCheckpoints["puller-egress"]; ok {
+					if slowestPullerResolvedTs > pullerCkpt.ResolvedTs {
+						slowestPullerResolvedTs = pullerCkpt.ResolvedTs
+						r.slowestPuller = span
+					}
+				}
 				return true
 			})
 		if !tableSpanFound || !tableSpanStartFound || !tableSpanEndFound || tableHasHole {
@@ -657,9 +668,6 @@ func (r *Manager) AdvanceCheckpoint(
 		}
 		return checkpointCannotProceed, checkpointCannotProceed
 	}
-	if slowestRange.TableID != 0 {
-		r.slowestTableID = slowestRange
-	}
 
 	// If currentTables is empty, we should advance newResolvedTs to global barrier ts and
 	// advance newCheckpointTs to min table barrier ts.
@@ -745,9 +753,9 @@ func (r *Manager) CollectMetrics() {
 	cf := r.changefeedID
 	tableGauge.
 		WithLabelValues(cf.Namespace, cf.ID).Set(float64(r.spans.Len()))
-	if table, ok := r.spans.Get(r.slowestTableID); ok {
+	if table, ok := r.spans.Get(r.slowestSink); ok {
 		slowestTableIDGauge.
-			WithLabelValues(cf.Namespace, cf.ID).Set(float64(r.slowestTableID.TableID))
+			WithLabelValues(cf.Namespace, cf.ID).Set(float64(r.slowestSink.TableID))
 		slowestTableStateGauge.
 			WithLabelValues(cf.Namespace, cf.ID).Set(float64(table.State))
 		phyCkpTs := oracle.ExtractPhysical(table.Checkpoint.CheckpointTs)
@@ -829,6 +837,17 @@ func (r *Manager) CollectMetrics() {
 			WithLabelValues(cf.Namespace, cf.ID, ReplicationSetState(s).String()).
 			Set(float64(counter))
 	}
+
+	if table, ok := r.spans.Get(r.slowestSink); ok {
+		if pullerCkpt, ok := table.Stats.StageCheckpoints["puller-egress"]; ok {
+			phyCkptTs := oracle.ExtractPhysical(pullerCkpt.ResolvedTs)
+			slowestTablePullerResolvedTs.WithLabelValues(cf.Namespace, cf.ID).Set(float64(phyCkptTs))
+
+			phyCurrentTs := oracle.ExtractPhysical(table.Stats.CurrentTs)
+			lag := float64(phyCurrentTs-phyCkptTs) / 1e3
+			slowestTablePullerResolvedTsLag.WithLabelValues(cf.Namespace, cf.ID).Set(lag)
+		}
+	}
 }
 
 // CleanMetrics cleans metrics.
diff --git a/cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink.go b/cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink.go
index 6d13ba535..140431245 100644
--- a/cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink.go
+++ b/cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink.go
@@ -80,7 +80,7 @@ type DMLSink struct {
 	alive struct {
 		sync.RWMutex
 		// msgCh is a channel to hold eventFragment.
-		msgCh  chan eventFragment
+		msgCh  *chann.DrainableChann[eventFragment]
 		isDead bool
 	}
 
@@ -142,7 +142,7 @@ func NewDMLSink(ctx context.Context,
 		cancel:          wgCancel,
 		dead:            make(chan struct{}),
 	}
-	s.alive.msgCh = make(chan eventFragment, defaultChannelSize)
+	s.alive.msgCh = chann.NewAutoDrainChann[eventFragment]()
 
 	encodedCh := make(chan eventFragment, defaultChannelSize)
 	workerChannels := make([]*chann.DrainableChann[eventFragment], cfg.WorkerCount)
@@ -150,7 +150,7 @@ func NewDMLSink(ctx context.Context,
 	// create a group of encoding workers.
 	for i := 0; i < defaultEncodingConcurrency; i++ {
 		encoder := encoderBuilder.Build()
-		s.encodingWorkers[i] = newEncodingWorker(i, s.changefeedID, encoder, s.alive.msgCh, encodedCh)
+		s.encodingWorkers[i] = newEncodingWorker(i, s.changefeedID, encoder, s.alive.msgCh.Out(), encodedCh)
 	}
 	// create defragmenter.
 	s.defragmenter = newDefragmenter(encodedCh, workerChannels)
@@ -170,7 +170,7 @@ func NewDMLSink(ctx context.Context,
 
 		s.alive.Lock()
 		s.alive.isDead = true
-		close(s.alive.msgCh)
+		s.alive.msgCh.CloseAndDrain()
 		s.alive.Unlock()
 		close(s.dead)
 
@@ -236,7 +236,7 @@ func (s *DMLSink) WriteEvents(txns ...*dmlsink.CallbackableEvent[*model.SingleTa
 
 		s.statistics.ObserveRows(txn.Event.Rows...)
 		// emit a TxnCallbackableEvent encoupled with a sequence number starting from one.
-		s.alive.msgCh <- eventFragment{
+		s.alive.msgCh.In() <- eventFragment{
 			seqNumber:      seq,
 			versionedTable: tbl,
 			event:          txn,
diff --git a/cdc/sink/dmlsink/mq/dispatcher/event_router.go b/cdc/sink/dmlsink/mq/dispatcher/event_router.go
index 53fc0c7fd..4cbaaa3c1 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/event_router.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/event_router.go
@@ -119,13 +119,14 @@ func (s *EventRouter) GetPartitionForRowChange(
 	row *model.RowChangedEvent,
 	partitionNum int32,
 ) (int32, string, error) {
-	_, partitionDispatcher := s.matchDispatcher(
-		row.Table.Schema, row.Table.Table,
-	)
+	return s.GetPartitionDispatcher(row.Table.Schema, row.Table.Table).
+		DispatchRowChangedEvent(row, partitionNum)
+}
 
-	return partitionDispatcher.DispatchRowChangedEvent(
-		row, partitionNum,
-	)
+// GetPartitionDispatcher returns the partition dispatcher for a specific table.
+func (s *EventRouter) GetPartitionDispatcher(schema, table string) partition.Dispatcher {
+	_, partitionDispatcher := s.matchDispatcher(schema, table)
+	return partitionDispatcher
 }
 
 // VerifyTables return error if any one table route rule is invalid.
@@ -147,7 +148,7 @@ func (s *EventRouter) VerifyTables(infos []*model.TableInfo) error {
 					"index is not unique when verify the table, table: %v, index: %s", table.TableName, v.IndexName)
 			}
 		case *partition.ColumnsDispatcher:
-			_, ok := table.ColumnsByNames(v.Columns)
+			_, ok := table.OffsetsByNames(v.Columns)
 			if !ok {
 				return cerror.ErrDispatcherFailed.GenWithStack(
 					"columns not found when verify the table, table: %v, columns: %v", table.TableName, v.Columns)
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go b/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
index e4a14cfcf..e63c1b17e 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/columns.go
@@ -55,7 +55,7 @@ func (r *ColumnsDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent,
 		dispatchCols = row.PreColumns
 	}
 
-	offsets, ok := row.TableInfo.ColumnsByNames(r.Columns)
+	offsets, ok := row.TableInfo.OffsetsByNames(r.Columns)
 	if !ok {
 		log.Error("columns not found when dispatch event",
 			zap.Any("tableName", row.Table),
@@ -65,7 +65,11 @@ func (r *ColumnsDispatcher) DispatchRowChangedEvent(row *model.RowChangedEvent,
 	}
 
 	for idx := 0; idx < len(r.Columns); idx++ {
-		r.hasher.Write([]byte(r.Columns[idx]), []byte(model.ColumnValueString(dispatchCols[offsets[idx]].Value)))
+		col := dispatchCols[offsets[idx]]
+		if col == nil {
+			continue
+		}
+		r.hasher.Write([]byte(r.Columns[idx]), []byte(model.ColumnValueString(col.Value)))
 	}
 
 	sum32 := r.hasher.Sum32()
diff --git a/cdc/sink/dmlsink/mq/dispatcher/partition/index_value.go b/cdc/sink/dmlsink/mq/dispatcher/partition/index_value.go
index 0a41369d5..d6e91f654 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/partition/index_value.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/partition/index_value.go
@@ -73,7 +73,11 @@ func (r *IndexValueDispatcher) DispatchRowChangedEvent(row *model.RowChangedEven
 				"index not found when dispatch event, table: %v, index: %s", row.Table, r.IndexName)
 		}
 		for idx := 0; idx < len(names); idx++ {
-			r.hasher.Write([]byte(names[idx]), []byte(model.ColumnValueString(dispatchCols[offsets[idx]].Value)))
+			col := dispatchCols[offsets[idx]]
+			if col == nil {
+				continue
+			}
+			r.hasher.Write([]byte(names[idx]), []byte(model.ColumnValueString(col.Value)))
 		}
 	}
 
diff --git a/cdc/sink/dmlsink/mq/kafka_dml_sink.go b/cdc/sink/dmlsink/mq/kafka_dml_sink.go
index c4c6bfc0f..86821ac25 100644
--- a/cdc/sink/dmlsink/mq/kafka_dml_sink.go
+++ b/cdc/sink/dmlsink/mq/kafka_dml_sink.go
@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dmlproducer"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/transformer/columnselector"
 	"github.com/pingcap/tiflow/cdc/sink/util"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -97,6 +98,11 @@ func NewKafkaDMLSink(
 		return nil, errors.Trace(err)
 	}
 
+	trans, err := columnselector.New(replicaConfig)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
 	encoderConfig, err := util.GetEncoderConfig(changefeedID, sinkURI, protocol, replicaConfig, options.MaxMessageBytes)
 	if err != nil {
 		return nil, errors.Trace(err)
@@ -118,7 +124,7 @@ func NewKafkaDMLSink(
 	concurrency := tiflowutil.GetOrZero(replicaConfig.Sink.EncoderConcurrency)
 	encoderGroup := codec.NewEncoderGroup(encoderBuilder, concurrency, changefeedID)
 	s := newDMLSink(ctx, changefeedID, dmlProducer, adminClient, topicManager,
-		eventRouter, encoderGroup, protocol, scheme, errCh)
+		eventRouter, trans, encoderGroup, protocol, scheme, errCh)
 	log.Info("DML sink producer created",
 		zap.String("namespace", changefeedID.Namespace),
 		zap.String("changefeedID", changefeedID.ID))
diff --git a/cdc/sink/dmlsink/mq/mq_dml_sink.go b/cdc/sink/dmlsink/mq/mq_dml_sink.go
index 94f363000..14ed62262 100644
--- a/cdc/sink/dmlsink/mq/mq_dml_sink.go
+++ b/cdc/sink/dmlsink/mq/mq_dml_sink.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dmlproducer"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/manager"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/transformer"
 	"github.com/pingcap/tiflow/cdc/sink/metrics"
 	"github.com/pingcap/tiflow/cdc/sink/tablesink/state"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -48,6 +49,8 @@ type dmlSink struct {
 
 	alive struct {
 		sync.RWMutex
+
+		transformer transformer.Transformer
 		// eventRouter used to route events to the right topic and partition.
 		eventRouter *dispatcher.EventRouter
 		// topicManager used to manage topics.
@@ -77,6 +80,7 @@ func newDMLSink(
 	adminClient kafka.ClusterAdminClient,
 	topicManager manager.TopicManager,
 	eventRouter *dispatcher.EventRouter,
+	transformer transformer.Transformer,
 	encoderGroup codec.EncoderGroup,
 	protocol config.Protocol,
 	scheme string,
@@ -95,6 +99,7 @@ func newDMLSink(
 		dead:        make(chan struct{}),
 		scheme:      scheme,
 	}
+	s.alive.transformer = transformer
 	s.alive.eventRouter = eventRouter
 	s.alive.topicManager = topicManager
 	s.alive.worker = worker
@@ -170,6 +175,13 @@ func (s *dmlSink) WriteEvents(txns ...*dmlsink.CallbackableEvent[*model.SingleTa
 				s.cancel(err)
 				return errors.Trace(err)
 			}
+
+			err = s.alive.transformer.Apply(row)
+			if err != nil {
+				s.cancel(err)
+				return errors.Trace(err)
+			}
+
 			index, key, err := s.alive.eventRouter.GetPartitionForRowChange(row, partitionNum)
 			if err != nil {
 				s.cancel(err)
diff --git a/cdc/sink/dmlsink/mq/pulsar_dml_sink.go b/cdc/sink/dmlsink/mq/pulsar_dml_sink.go
index 93b40f9ac..ea7716310 100644
--- a/cdc/sink/dmlsink/mq/pulsar_dml_sink.go
+++ b/cdc/sink/dmlsink/mq/pulsar_dml_sink.go
@@ -24,6 +24,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dmlproducer"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/manager"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/transformer/columnselector"
 	"github.com/pingcap/tiflow/cdc/sink/util"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -104,6 +105,11 @@ func NewPulsarDMLSink(
 		return nil, errors.Trace(err)
 	}
 
+	trans, err := columnselector.New(replicaConfig)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
 	encoderConfig, err := util.GetEncoderConfig(changefeedID, sinkURI, protocol, replicaConfig,
 		config.DefaultMaxMessageBytes)
 	if err != nil {
@@ -119,7 +125,7 @@ func NewPulsarDMLSink(
 	encoderGroup := codec.NewEncoderGroup(encoderBuilder, concurrency, changefeedID)
 
 	s := newDMLSink(ctx, changefeedID, p, nil, topicManager,
-		eventRouter, encoderGroup, protocol, scheme, errCh)
+		eventRouter, trans, encoderGroup, protocol, scheme, errCh)
 
 	return s, nil
 }
diff --git a/cdc/sink/dmlsink/mq/transformer/columnselector/column_selector.go b/cdc/sink/dmlsink/mq/transformer/columnselector/column_selector.go
new file mode 100644
index 000000000..65e2b6e15
--- /dev/null
+++ b/cdc/sink/dmlsink/mq/transformer/columnselector/column_selector.go
@@ -0,0 +1,233 @@
+// Copyright 2023 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 columnselector
+
+import (
+	filter "github.com/pingcap/tidb/util/table-filter"
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher/partition"
+	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/pingcap/tiflow/pkg/errors"
+)
+
+type selector struct {
+	tableF  filter.Filter
+	columnM filter.ColumnFilter
+}
+
+func newSelector(
+	rule *config.ColumnSelector, caseSensitive bool,
+) (*selector, error) {
+	tableM, err := filter.Parse(rule.Matcher)
+	if err != nil {
+		return nil, errors.WrapError(errors.ErrFilterRuleInvalid, err, rule.Matcher)
+	}
+	if !caseSensitive {
+		tableM = filter.CaseInsensitive(tableM)
+	}
+	columnM, err := filter.ParseColumnFilter(rule.Columns)
+	if err != nil {
+		return nil, errors.WrapError(errors.ErrFilterRuleInvalid, err, rule.Columns)
+	}
+
+	return &selector{
+		tableF:  tableM,
+		columnM: columnM,
+	}, nil
+}
+
+// Match implements Transformer interface
+func (s *selector) Match(schema, table string) bool {
+	return s.tableF.MatchTable(schema, table)
+}
+
+// Apply implements Transformer interface
+// return error if the given event cannot match the selector, or the column cannot be filtered out.
+// the caller's should make sure the given event match the selector first before apply it.
+func (s *selector) Apply(event *model.RowChangedEvent) error {
+	// defensive check, this should not happen.
+	if !s.Match(event.Table.Schema, event.Table.Table) {
+		return errors.ErrColumnSelectorFailed.GenWithStack(
+			"the given event does not match the column selector, table: %v", event.Table)
+	}
+
+	retainedColumns := make(map[string]struct{}, len(event.Columns))
+	if len(event.Columns) != 0 {
+		for idx, column := range event.Columns {
+			if s.columnM.MatchColumn(column.Name) {
+				retainedColumns[column.Name] = struct{}{}
+				continue
+			}
+			event.Columns[idx] = nil
+		}
+
+		if !verifyIndices(event.TableInfo, retainedColumns) {
+			return errors.ErrColumnSelectorFailed.GenWithStack(
+				"no primary key columns or unique key columns obtained after filter out, "+
+					"table: %+v", event.Table)
+		}
+	}
+
+	if len(event.PreColumns) != 0 {
+		clear(retainedColumns)
+		for idx, column := range event.PreColumns {
+			if s.columnM.MatchColumn(column.Name) {
+				retainedColumns[column.Name] = struct{}{}
+				continue
+			}
+			event.PreColumns[idx] = nil
+		}
+		if !verifyIndices(event.TableInfo, retainedColumns) {
+			return errors.ErrColumnSelectorFailed.GenWithStack(
+				"no primary key columns or unique key columns obtained after filter out, "+
+					"table: %+v", event.Table)
+		}
+	}
+
+	return nil
+}
+
+// ColumnSelector manages an array of selectors, the first selector match the given
+// event is used to select out columns.
+type ColumnSelector struct {
+	selectors []*selector
+}
+
+// New return a column selector
+func New(cfg *config.ReplicaConfig) (*ColumnSelector, error) {
+	selectors := make([]*selector, 0, len(cfg.Sink.ColumnSelectors))
+	for _, r := range cfg.Sink.ColumnSelectors {
+		selector, err := newSelector(r, cfg.CaseSensitive)
+		if err != nil {
+			return nil, err
+		}
+		selectors = append(selectors, selector)
+	}
+
+	return &ColumnSelector{
+		selectors: selectors,
+	}, nil
+}
+
+// Apply the column selector to the given event.
+func (c *ColumnSelector) Apply(event *model.RowChangedEvent) error {
+	for _, s := range c.selectors {
+		if s.Match(event.Table.Schema, event.Table.Table) {
+			return s.Apply(event)
+		}
+	}
+	return nil
+}
+
+// VerifyTables return the error if any given table cannot satisfy the column selector constraints.
+// 1. if the column is filter out, it must not be a part of handle key or the unique key.
+// 2. if the filtered out column is used in the column dispatcher, return error.
+func (c *ColumnSelector) VerifyTables(
+	infos []*model.TableInfo, eventRouter *dispatcher.EventRouter,
+) error {
+	if len(c.selectors) == 0 {
+		return nil
+	}
+
+	for _, table := range infos {
+		for _, s := range c.selectors {
+			if !s.Match(table.TableName.Schema, table.TableName.Table) {
+				continue
+			}
+
+			retainedColumns := make(map[string]struct{})
+			for columnID := range table.ColumnsFlag {
+				columnInfo, ok := table.GetColumnInfo(columnID)
+				if !ok {
+					return errors.ErrColumnSelectorFailed.GenWithStack(
+						"column not found when verify the table for the column selector, table: %v, column: %s",
+						table.TableName, columnInfo.Name)
+				}
+				columnName := columnInfo.Name.O
+				if s.columnM.MatchColumn(columnName) {
+					retainedColumns[columnName] = struct{}{}
+					continue
+				}
+
+				partitionDispatcher := eventRouter.GetPartitionDispatcher(table.TableName.Schema, table.TableName.Table)
+				switch v := partitionDispatcher.(type) {
+				case *partition.ColumnsDispatcher:
+					for _, col := range v.Columns {
+						if col == columnInfo.Name.O {
+							return errors.ErrColumnSelectorFailed.GenWithStack(
+								"the filtered out column is used in the column dispatcher, "+
+									"table: %v, column: %s", table.TableName, columnInfo.Name)
+						}
+					}
+				default:
+				}
+			}
+
+			if !verifyIndices(table, retainedColumns) {
+				return errors.ErrColumnSelectorFailed.GenWithStack(
+					"no primary key columns or unique key columns obtained after filter out, table: %+v", table.TableName)
+			}
+		}
+	}
+	return nil
+}
+
+// verifyIndices return true if the primary key retained,
+// else at least there are one unique key columns in the retained columns.
+func verifyIndices(table *model.TableInfo, retainedColumns map[string]struct{}) bool {
+	primaryKeyColumns := table.GetPrimaryKeyColumnNames()
+
+	retained := true
+	for name := range primaryKeyColumns {
+		if _, ok := retainedColumns[name]; !ok {
+			retained = false
+			break
+		}
+	}
+	// primary key columns are retained, return true.
+	if retained {
+		return true
+	}
+
+	// at least one unique key columns are retained, return true.
+	for _, index := range table.Indices {
+		if !index.Unique {
+			continue
+		}
+
+		retained = true
+		for _, col := range index.Columns {
+			if _, ok := retainedColumns[col.Name.O]; !ok {
+				retained = false
+				break
+			}
+		}
+		if retained {
+			return true
+		}
+	}
+	return false
+}
+
+// VerifyColumn return true if the given `schema.table` column is matched.
+func (c *ColumnSelector) VerifyColumn(schema, table, column string) bool {
+	for _, s := range c.selectors {
+		if !s.Match(schema, table) {
+			continue
+		}
+		return s.columnM.MatchColumn(column)
+	}
+	return true
+}
diff --git a/cdc/sink/dmlsink/mq/transformer/transformer.go b/cdc/sink/dmlsink/mq/transformer/transformer.go
new file mode 100644
index 000000000..2cdb8c178
--- /dev/null
+++ b/cdc/sink/dmlsink/mq/transformer/transformer.go
@@ -0,0 +1,21 @@
+// Copyright 2023 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 transformer
+
+import "github.com/pingcap/tiflow/cdc/model"
+
+// Transformer is the interface for transform the event.
+type Transformer interface {
+	Apply(event *model.RowChangedEvent) error
+}
diff --git a/dm/ctl/master/update_validation.go b/dm/ctl/master/update_validation.go
index 77c45b368..7941a28f3 100644
--- a/dm/ctl/master/update_validation.go
+++ b/dm/ctl/master/update_validation.go
@@ -18,6 +18,7 @@ import (
 
 	"github.com/pingcap/tiflow/dm/ctl/common"
 	"github.com/pingcap/tiflow/dm/pb"
+	"github.com/pingcap/tiflow/dm/pkg/binlog"
 	"github.com/spf13/cobra"
 )
 
@@ -71,15 +72,27 @@ func parseValidationUpdateArgs(cmd *cobra.Command) (validationUpdateArgs, string
 	if args.sources, err = common.GetSourceArgs(cmd); err != nil {
 		return args, err.Error(), false
 	}
-	if args.cutoverBinlogPos, err = cmd.Flags().GetString("cutover-binlog-name"); err != nil {
+	if args.cutoverBinlogPos, err = cmd.Flags().GetString("cutover-binlog-pos"); err != nil {
 		return args, err.Error(), false
 	}
+	if len(args.cutoverBinlogPos) != 0 {
+		_, err = binlog.PositionFromPosStr(args.cutoverBinlogPos)
+		if err != nil {
+			return args, err.Error(), false
+		}
+	}
+
 	if args.cutoverBinlogGTID, err = cmd.Flags().GetString("cutover-binlog-gtid"); err != nil {
 		return args, err.Error(), false
 	}
+	if len(args.cutoverBinlogGTID) != 0 && len(args.cutoverBinlogPos) != 0 {
+		return args, "you must specify either one of '--cutover-binlog-pos' or '--cutover-binlog-pos'", false
+	}
 
-	if len(cmd.Flags().Args()) != 1 {
+	if len(cmd.Flags().Args()) == 0 {
 		return args, "`task-name` should be set", false
+	} else if len(cmd.Flags().Args()) > 1 {
+		return args, "should specify only one `task-name`", false
 	}
 	args.taskName = cmd.Flags().Arg(0)
 	if len(args.taskName) == 0 {
diff --git a/dm/master/server.go b/dm/master/server.go
index 684a23e05..9dc9ea685 100644
--- a/dm/master/server.go
+++ b/dm/master/server.go
@@ -3228,15 +3228,18 @@ func (s *Server) UpdateValidation(ctx context.Context, req *pb.UpdateValidationR
 	if shouldRet {
 		return resp2, err
 	}
-	sources := req.Sources
-	if len(sources) == 0 {
-		sources = s.getTaskSourceNameList(req.TaskName)
-		if len(sources) == 0 {
-			return &pb.UpdateValidationResponse{
-				Result: false,
-				Msg:    fmt.Sprintf("task %s has no source or not exist, please check the task name and status", req.TaskName),
-			}, nil
+	resp := &pb.UpdateValidationResponse{
+		Result: false,
+	}
+	subTaskCfgs := s.scheduler.GetSubTaskCfgsByTaskAndSource(req.TaskName, req.Sources)
+	if len(subTaskCfgs) == 0 {
+		if len(req.Sources) > 0 {
+			resp.Msg = fmt.Sprintf("cannot get subtask by task name `%s` and sources `%v`",
+				req.TaskName, req.Sources)
+		} else {
+			resp.Msg = fmt.Sprintf("cannot get subtask by task name `%s`", req.TaskName)
 		}
+		return resp, nil
 	}
 
 	workerReq := workerrpc.Request{
@@ -3248,31 +3251,48 @@ func (s *Server) UpdateValidation(ctx context.Context, req *pb.UpdateValidationR
 		},
 	}
 
-	workerRespCh := make(chan *pb.CommonWorkerResponse, len(sources))
+	sourcesLen := 0
+	for _, subTaskCfg := range subTaskCfgs {
+		sourcesLen += len(subTaskCfg)
+	}
+	workerRespCh := make(chan *pb.CommonWorkerResponse, sourcesLen)
 	var wg sync.WaitGroup
-	for _, source := range sources {
-		wg.Add(1)
-		go func(source string) {
-			defer wg.Done()
-			worker := s.scheduler.GetWorkerBySource(source)
-			if worker == nil {
-				workerRespCh <- errorCommonWorkerResponse(fmt.Sprintf("source %s relevant worker-client not found", source), source, "")
-				return
-			}
-			var workerResp *pb.CommonWorkerResponse
-			resp, err := worker.SendRequest(ctx, &workerReq, s.cfg.RPCTimeout)
-			if err != nil {
-				workerResp = errorCommonWorkerResponse(err.Error(), source, worker.BaseInfo().Name)
-			} else {
-				workerResp = resp.UpdateValidation
-			}
-			workerResp.Source = source
-			workerRespCh <- workerResp
-		}(source)
+	for _, subTaskCfg := range subTaskCfgs {
+		for sourceID := range subTaskCfg {
+			wg.Add(1)
+			go func(source string) {
+				defer wg.Done()
+				sourceCfg := s.scheduler.GetSourceCfgByID(source)
+				// can't directly use subtaskCfg here, because it will be overwritten by sourceCfg
+				if sourceCfg.EnableGTID {
+					if len(req.BinlogGTID) == 0 {
+						workerRespCh <- errorCommonWorkerResponse(fmt.Sprintf("source %s didn't specify cutover-binlog-gtid when enableGTID is true", source), source, "")
+						return
+					}
+				} else if len(req.BinlogPos) == 0 {
+					workerRespCh <- errorCommonWorkerResponse(fmt.Sprintf("source %s didn't specify cutover-binlog-pos when enableGTID is false", source), source, "")
+					return
+				}
+				worker := s.scheduler.GetWorkerBySource(source)
+				if worker == nil {
+					workerRespCh <- errorCommonWorkerResponse(fmt.Sprintf("source %s relevant worker-client not found", source), source, "")
+					return
+				}
+				var workerResp *pb.CommonWorkerResponse
+				resp, err := worker.SendRequest(ctx, &workerReq, s.cfg.RPCTimeout)
+				if err != nil {
+					workerResp = errorCommonWorkerResponse(err.Error(), source, worker.BaseInfo().Name)
+				} else {
+					workerResp = resp.UpdateValidation
+				}
+				workerResp.Source = source
+				workerRespCh <- workerResp
+			}(sourceID)
+		}
 	}
 	wg.Wait()
 
-	workerResps := make([]*pb.CommonWorkerResponse, 0, len(sources))
+	workerResps := make([]*pb.CommonWorkerResponse, 0, sourcesLen)
 	for len(workerRespCh) > 0 {
 		workerResp := <-workerRespCh
 		workerResps = append(workerResps, workerResp)
diff --git a/dm/pb/dmworker.pb.go b/dm/pb/dmworker.pb.go
index f340aabbb..098648f42 100644
--- a/dm/pb/dmworker.pb.go
+++ b/dm/pb/dmworker.pb.go
@@ -2904,6 +2904,8 @@ type ValidationStatus struct {
 	ProcessedRowsStatus string         `protobuf:"bytes,8,opt,name=processedRowsStatus,proto3" json:"processedRowsStatus,omitempty"`
 	PendingRowsStatus   string         `protobuf:"bytes,9,opt,name=pendingRowsStatus,proto3" json:"pendingRowsStatus,omitempty"`
 	ErrorRowsStatus     string         `protobuf:"bytes,10,opt,name=errorRowsStatus,proto3" json:"errorRowsStatus,omitempty"`
+	CutoverBinlogPos    string         `protobuf:"bytes,11,opt,name=cutoverBinlogPos,proto3" json:"cutoverBinlogPos,omitempty"`
+	CutoverBinlogGtid   string         `protobuf:"bytes,12,opt,name=cutoverBinlogGtid,proto3" json:"cutoverBinlogGtid,omitempty"`
 }
 
 func (m *ValidationStatus) Reset()         { *m = ValidationStatus{} }
@@ -3009,6 +3011,20 @@ func (m *ValidationStatus) GetErrorRowsStatus() string {
 	return ""
 }
 
+func (m *ValidationStatus) GetCutoverBinlogPos() string {
+	if m != nil {
+		return m.CutoverBinlogPos
+	}
+	return ""
+}
+
+func (m *ValidationStatus) GetCutoverBinlogGtid() string {
+	if m != nil {
+		return m.CutoverBinlogGtid
+	}
+	return ""
+}
+
 type ValidationTableStatus struct {
 	Source   string `protobuf:"bytes,1,opt,name=source,proto3" json:"source,omitempty"`
 	SrcTable string `protobuf:"bytes,2,opt,name=srcTable,proto3" json:"srcTable,omitempty"`
@@ -3561,58 +3577,6 @@ func (m *UpdateValidationWorkerRequest) GetBinlogGTID() string {
 	return ""
 }
 
-type UpdateValidationWorkerResponse struct {
-	Result bool   `protobuf:"varint,1,opt,name=result,proto3" json:"result,omitempty"`
-	Msg    string `protobuf:"bytes,2,opt,name=msg,proto3" json:"msg,omitempty"`
-}
-
-func (m *UpdateValidationWorkerResponse) Reset()         { *m = UpdateValidationWorkerResponse{} }
-func (m *UpdateValidationWorkerResponse) String() string { return proto.CompactTextString(m) }
-func (*UpdateValidationWorkerResponse) ProtoMessage()    {}
-func (*UpdateValidationWorkerResponse) Descriptor() ([]byte, []int) {
-	return fileDescriptor_51a1b9e17fd67b10, []int{43}
-}
-func (m *UpdateValidationWorkerResponse) XXX_Unmarshal(b []byte) error {
-	return m.Unmarshal(b)
-}
-func (m *UpdateValidationWorkerResponse) XXX_Marshal(b []byte, deterministic bool) ([]byte, error) {
-	if deterministic {
-		return xxx_messageInfo_UpdateValidationWorkerResponse.Marshal(b, m, deterministic)
-	} else {
-		b = b[:cap(b)]
-		n, err := m.MarshalToSizedBuffer(b)
-		if err != nil {
-			return nil, err
-		}
-		return b[:n], nil
-	}
-}
-func (m *UpdateValidationWorkerResponse) XXX_Merge(src proto.Message) {
-	xxx_messageInfo_UpdateValidationWorkerResponse.Merge(m, src)
-}
-func (m *UpdateValidationWorkerResponse) XXX_Size() int {
-	return m.Size()
-}
-func (m *UpdateValidationWorkerResponse) XXX_DiscardUnknown() {
-	xxx_messageInfo_UpdateValidationWorkerResponse.DiscardUnknown(m)
-}
-
-var xxx_messageInfo_UpdateValidationWorkerResponse proto.InternalMessageInfo
-
-func (m *UpdateValidationWorkerResponse) GetResult() bool {
-	if m != nil {
-		return m.Result
-	}
-	return false
-}
-
-func (m *UpdateValidationWorkerResponse) GetMsg() string {
-	if m != nil {
-		return m.Msg
-	}
-	return ""
-}
-
 func init() {
 	proto.RegisterEnum("pb.TaskOp", TaskOp_name, TaskOp_value)
 	proto.RegisterEnum("pb.Stage", Stage_name, Stage_value)
@@ -3668,197 +3632,197 @@ func init() {
 	proto.RegisterType((*OperateValidationErrorRequest)(nil), "pb.OperateValidationErrorRequest")
 	proto.RegisterType((*OperateValidationErrorResponse)(nil), "pb.OperateValidationErrorResponse")
 	proto.RegisterType((*UpdateValidationWorkerRequest)(nil), "pb.UpdateValidationWorkerRequest")
-	proto.RegisterType((*UpdateValidationWorkerResponse)(nil), "pb.UpdateValidationWorkerResponse")
 }
 
 func init() { proto.RegisterFile("dmworker.proto", fileDescriptor_51a1b9e17fd67b10) }
 
 var fileDescriptor_51a1b9e17fd67b10 = []byte{
-	// 2929 bytes of a gzipped FileDescriptorProto
+	// 2947 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0xb4, 0x3a, 0xcd, 0x6f, 0xe5, 0x56,
 	0xf5, 0xcf, 0xf6, 0xfb, 0x3c, 0xef, 0x25, 0x71, 0xee, 0x64, 0xe6, 0xf7, 0x9a, 0xce, 0xbc, 0xa6,
 	0x9e, 0xaa, 0xbf, 0x34, 0x2a, 0x51, 0x1b, 0x8a, 0x8a, 0x2a, 0x41, 0xdb, 0x49, 0xa6, 0x99, 0x29,
 	0x99, 0x66, 0xc6, 0x49, 0x87, 0x15, 0x12, 0x8e, 0xdf, 0xcd, 0x8b, 0x89, 0x9f, 0xed, 0xb1, 0xfd,
-	0x12, 0x65, 0x81, 0xd8, 0x20, 0xb6, 0xb0, 0x01, 0x09, 0xc4, 0x06, 0x24, 0xb6, 0x2c, 0xf8, 0x03,
-	0x58, 0x42, 0x97, 0x15, 0x2b, 0x96, 0xa8, 0xf3, 0x37, 0xb0, 0x61, 0x81, 0xd0, 0x39, 0xf7, 0x5e,
-	0xfb, 0xfa, 0x7d, 0x64, 0x3a, 0x48, 0xec, 0x7c, 0x3e, 0x7c, 0xee, 0xb9, 0xe7, 0xfb, 0xf8, 0x3d,
+	0x12, 0x65, 0x81, 0xd8, 0xb1, 0x85, 0x0d, 0x48, 0x20, 0x36, 0x20, 0xb1, 0x65, 0xc1, 0x1f, 0xc0,
+	0x12, 0xba, 0xac, 0x58, 0xb1, 0x42, 0xa8, 0xf3, 0x37, 0xb0, 0x61, 0x81, 0xd0, 0x39, 0xf7, 0x5e,
+	0xfb, 0xfa, 0x7d, 0x64, 0x3a, 0x48, 0xec, 0x7c, 0x3e, 0xee, 0xb9, 0xc7, 0xe7, 0xfb, 0xf8, 0x3d,
 	0x58, 0x1e, 0x8e, 0x2f, 0xe3, 0xf4, 0x9c, 0xa7, 0xdb, 0x49, 0x1a, 0xe7, 0x31, 0x33, 0x93, 0x13,
 	0x67, 0x13, 0xd8, 0x93, 0x09, 0x4f, 0xaf, 0x8e, 0x72, 0x2f, 0x9f, 0x64, 0x2e, 0x7f, 0x36, 0xe1,
 	0x59, 0xce, 0x18, 0xd4, 0x23, 0x6f, 0xcc, 0xfb, 0xc6, 0x86, 0xb1, 0xd9, 0x71, 0xe9, 0xd9, 0x49,
 	0x60, 0x6d, 0x37, 0x1e, 0x8f, 0xe3, 0xe8, 0xfb, 0x24, 0xc3, 0xe5, 0x59, 0x12, 0x47, 0x19, 0x67,
 	0xb7, 0xa0, 0x99, 0xf2, 0x6c, 0x12, 0xe6, 0xc4, 0xdd, 0x76, 0x25, 0xc4, 0x6c, 0xb0, 0xc6, 0xd9,
 	0xa8, 0x6f, 0x92, 0x08, 0x7c, 0x44, 0xce, 0x2c, 0x9e, 0xa4, 0x3e, 0xef, 0x5b, 0x84, 0x94, 0x10,
-	0xe2, 0x85, 0x5e, 0xfd, 0xba, 0xc0, 0x0b, 0xc8, 0xf9, 0xa3, 0x01, 0x37, 0x2a, 0xca, 0xbd, 0xf4,
-	0x89, 0xef, 0x41, 0x4f, 0x9c, 0x21, 0x24, 0xd0, 0xb9, 0xdd, 0x1d, 0x7b, 0x3b, 0x39, 0xd9, 0x3e,
-	0xd2, 0xf0, 0x6e, 0x85, 0x8b, 0xbd, 0x0f, 0x4b, 0xd9, 0xe4, 0xe4, 0xd8, 0xcb, 0xce, 0xe5, 0x6b,
-	0xf5, 0x0d, 0x6b, 0xb3, 0xbb, 0xb3, 0x4a, 0xaf, 0xe9, 0x04, 0xb7, 0xca, 0xe7, 0xfc, 0xc1, 0x80,
-	0xee, 0xee, 0x19, 0xf7, 0x25, 0x8c, 0x8a, 0x26, 0x5e, 0x96, 0xf1, 0xa1, 0x52, 0x54, 0x40, 0x6c,
-	0x0d, 0x1a, 0x79, 0x9c, 0x7b, 0x21, 0xa9, 0xda, 0x70, 0x05, 0xc0, 0x06, 0x00, 0xd9, 0xc4, 0xf7,
-	0x79, 0x96, 0x9d, 0x4e, 0x42, 0x52, 0xb5, 0xe1, 0x6a, 0x18, 0x94, 0x76, 0xea, 0x05, 0x21, 0x1f,
-	0x92, 0x99, 0x1a, 0xae, 0x84, 0x58, 0x1f, 0x5a, 0x97, 0x5e, 0x1a, 0x05, 0xd1, 0xa8, 0xdf, 0x20,
-	0x82, 0x02, 0xf1, 0x8d, 0x21, 0xcf, 0xbd, 0x20, 0xec, 0x37, 0x37, 0x8c, 0xcd, 0x9e, 0x2b, 0x21,
-	0xe7, 0xdf, 0x06, 0xc0, 0xde, 0x64, 0x9c, 0x48, 0x35, 0x37, 0xa0, 0x4b, 0x1a, 0x1c, 0x7b, 0x27,
-	0x21, 0xcf, 0x48, 0x57, 0xcb, 0xd5, 0x51, 0x6c, 0x13, 0x56, 0xfc, 0x78, 0x9c, 0x84, 0x3c, 0xe7,
-	0x43, 0xc9, 0x85, 0xaa, 0x1b, 0xee, 0x34, 0x9a, 0xbd, 0x01, 0x4b, 0xa7, 0x41, 0x14, 0x64, 0x67,
-	0x7c, 0x78, 0xef, 0x2a, 0xe7, 0xc2, 0xe4, 0x86, 0x5b, 0x45, 0x32, 0x07, 0x7a, 0x0a, 0xe1, 0xc6,
-	0x97, 0x19, 0x5d, 0xc8, 0x70, 0x2b, 0x38, 0xf6, 0x36, 0xac, 0xf2, 0x2c, 0x0f, 0xc6, 0x5e, 0xce,
-	0x8f, 0x51, 0x15, 0x62, 0x6c, 0x10, 0xe3, 0x2c, 0x01, 0x7d, 0x7f, 0x92, 0x64, 0x74, 0x4f, 0xcb,
-	0xc5, 0x47, 0xb6, 0x0e, 0xed, 0x24, 0x8d, 0x47, 0x29, 0xcf, 0xb2, 0x7e, 0x8b, 0x42, 0xa2, 0x80,
-	0x9d, 0x2f, 0x0c, 0x80, 0x83, 0xd8, 0x1b, 0x4a, 0x03, 0xcc, 0x28, 0x2d, 0x4c, 0x30, 0xa5, 0xf4,
-	0x00, 0x80, 0x6c, 0x22, 0x58, 0x4c, 0x62, 0xd1, 0x30, 0x95, 0x03, 0xad, 0xea, 0x81, 0xf8, 0xee,
-	0x98, 0xe7, 0xde, 0xbd, 0x20, 0x0a, 0xe3, 0x91, 0x0c, 0x73, 0x0d, 0xc3, 0xde, 0x84, 0xe5, 0x12,
-	0xda, 0x3f, 0x7e, 0xb8, 0x47, 0x37, 0xed, 0xb8, 0x53, 0xd8, 0xd9, 0x6b, 0x3a, 0xbf, 0x34, 0x60,
+	0xe2, 0x85, 0x5e, 0xfd, 0xba, 0xc0, 0x0b, 0xc8, 0xf9, 0x83, 0x01, 0x37, 0x2a, 0xca, 0xbd, 0xf4,
+	0x8d, 0xef, 0x41, 0x4f, 0xdc, 0x21, 0x24, 0xd0, 0xbd, 0xdd, 0x1d, 0x7b, 0x3b, 0x39, 0xd9, 0x3e,
+	0xd2, 0xf0, 0x6e, 0x85, 0x8b, 0xbd, 0x0f, 0x4b, 0xd9, 0xe4, 0xe4, 0xd8, 0xcb, 0xce, 0xe5, 0xb1,
+	0xfa, 0x86, 0xb5, 0xd9, 0xdd, 0x59, 0xa5, 0x63, 0x3a, 0xc1, 0xad, 0xf2, 0x39, 0xbf, 0x37, 0xa0,
+	0xbb, 0x7b, 0xc6, 0x7d, 0x09, 0xa3, 0xa2, 0x89, 0x97, 0x65, 0x7c, 0xa8, 0x14, 0x15, 0x10, 0x5b,
+	0x83, 0x46, 0x1e, 0xe7, 0x5e, 0x48, 0xaa, 0x36, 0x5c, 0x01, 0xb0, 0x01, 0x40, 0x36, 0xf1, 0x7d,
+	0x9e, 0x65, 0xa7, 0x93, 0x90, 0x54, 0x6d, 0xb8, 0x1a, 0x06, 0xa5, 0x9d, 0x7a, 0x41, 0xc8, 0x87,
+	0x64, 0xa6, 0x86, 0x2b, 0x21, 0xd6, 0x87, 0xd6, 0xa5, 0x97, 0x46, 0x41, 0x34, 0xea, 0x37, 0x88,
+	0xa0, 0x40, 0x3c, 0x31, 0xe4, 0xb9, 0x17, 0x84, 0xfd, 0xe6, 0x86, 0xb1, 0xd9, 0x73, 0x25, 0xe4,
+	0xfc, 0xdb, 0x00, 0xd8, 0x9b, 0x8c, 0x13, 0xa9, 0xe6, 0x06, 0x74, 0x49, 0x83, 0x63, 0xef, 0x24,
+	0xe4, 0x19, 0xe9, 0x6a, 0xb9, 0x3a, 0x8a, 0x6d, 0xc2, 0x8a, 0x1f, 0x8f, 0x93, 0x90, 0xe7, 0x7c,
+	0x28, 0xb9, 0x50, 0x75, 0xc3, 0x9d, 0x46, 0xb3, 0x37, 0x60, 0xe9, 0x34, 0x88, 0x82, 0xec, 0x8c,
+	0x0f, 0xef, 0x5d, 0xe5, 0x5c, 0x98, 0xdc, 0x70, 0xab, 0x48, 0xe6, 0x40, 0x4f, 0x21, 0xdc, 0xf8,
+	0x32, 0xa3, 0x17, 0x32, 0xdc, 0x0a, 0x8e, 0xbd, 0x0d, 0xab, 0x3c, 0xcb, 0x83, 0xb1, 0x97, 0xf3,
+	0x63, 0x54, 0x85, 0x18, 0x1b, 0xc4, 0x38, 0x4b, 0x40, 0xdf, 0x9f, 0x24, 0x19, 0xbd, 0xa7, 0xe5,
+	0xe2, 0x23, 0x5b, 0x87, 0x76, 0x92, 0xc6, 0xa3, 0x94, 0x67, 0x59, 0xbf, 0x45, 0x21, 0x51, 0xc0,
+	0xce, 0x17, 0x06, 0xc0, 0x41, 0xec, 0x0d, 0xa5, 0x01, 0x66, 0x94, 0x16, 0x26, 0x98, 0x52, 0x7a,
+	0x00, 0x40, 0x36, 0x11, 0x2c, 0x26, 0xb1, 0x68, 0x98, 0xca, 0x85, 0x56, 0xf5, 0x42, 0x3c, 0x3b,
+	0xe6, 0xb9, 0x77, 0x2f, 0x88, 0xc2, 0x78, 0x24, 0xc3, 0x5c, 0xc3, 0xb0, 0x37, 0x61, 0xb9, 0x84,
+	0xf6, 0x8f, 0x1f, 0xee, 0xd1, 0x9b, 0x76, 0xdc, 0x29, 0xec, 0xec, 0x6b, 0x3a, 0xbf, 0x30, 0x60,
 	0xe9, 0xe8, 0xcc, 0x4b, 0x87, 0x41, 0x34, 0xda, 0x4f, 0xe3, 0x49, 0x82, 0x5e, 0xcf, 0xbd, 0x74,
 	0xc4, 0x73, 0x99, 0xbe, 0x12, 0xc2, 0xa4, 0xde, 0xdb, 0x3b, 0x40, 0xcd, 0x2d, 0x4c, 0x6a, 0x7c,
-	0x16, 0x37, 0x4f, 0xb3, 0xfc, 0x20, 0xf6, 0xbd, 0x3c, 0x88, 0x23, 0xa9, 0x78, 0x15, 0x49, 0x89,
-	0x7b, 0x15, 0xf9, 0x14, 0x79, 0x16, 0x25, 0x2e, 0x41, 0x78, 0xe3, 0x49, 0x24, 0x29, 0x0d, 0xa2,
-	0x14, 0xb0, 0xf3, 0xcf, 0x3a, 0xc0, 0xd1, 0x55, 0xe4, 0x4f, 0xc5, 0xd8, 0xfd, 0x0b, 0x1e, 0xe5,
-	0xd5, 0x18, 0x13, 0x28, 0x14, 0x26, 0x42, 0x2e, 0x51, 0xc6, 0x2d, 0x60, 0x76, 0x1b, 0x3a, 0x29,
-	0xf7, 0x79, 0x94, 0x23, 0xd1, 0x22, 0x62, 0x89, 0xc0, 0x68, 0x1a, 0x7b, 0x59, 0xce, 0xd3, 0x8a,
-	0x79, 0x2b, 0x38, 0xb6, 0x05, 0xb6, 0x0e, 0xef, 0xe7, 0xc1, 0x50, 0x9a, 0x78, 0x06, 0x8f, 0xf2,
-	0xe8, 0x12, 0x4a, 0x5e, 0x53, 0xc8, 0xd3, 0x71, 0x28, 0x4f, 0x87, 0x49, 0x9e, 0x88, 0xb2, 0x19,
-	0x3c, 0xca, 0x3b, 0x09, 0x63, 0xff, 0x3c, 0x88, 0x46, 0xe4, 0x80, 0x36, 0x99, 0xaa, 0x82, 0x63,
-	0xdf, 0x01, 0x7b, 0x12, 0xa5, 0x3c, 0x8b, 0xc3, 0x0b, 0x3e, 0x24, 0x3f, 0x66, 0xfd, 0x8e, 0x56,
-	0x76, 0x74, 0x0f, 0xbb, 0x33, 0xac, 0x9a, 0x87, 0x40, 0x54, 0x1a, 0xe9, 0xa1, 0x01, 0xc0, 0x09,
-	0x29, 0x72, 0x7c, 0x95, 0xf0, 0x7e, 0x57, 0xc4, 0x5d, 0x89, 0x61, 0xef, 0xc0, 0x8d, 0x8c, 0xfb,
-	0x71, 0x34, 0xcc, 0xee, 0xf1, 0xb3, 0x20, 0x1a, 0x3e, 0x22, 0x5b, 0xf4, 0x7b, 0x64, 0xe2, 0x79,
-	0x24, 0x8c, 0x18, 0x52, 0x7c, 0x6f, 0xef, 0xe0, 0xf0, 0x32, 0xe2, 0x69, 0x7f, 0x49, 0x44, 0x4c,
-	0x05, 0x89, 0xee, 0xf6, 0xe3, 0xe8, 0x34, 0x0c, 0xfc, 0xfc, 0x51, 0x36, 0xea, 0x2f, 0x13, 0x8f,
-	0x8e, 0x42, 0x97, 0xe6, 0x45, 0x5a, 0xaf, 0x08, 0x97, 0x16, 0x88, 0x22, 0x18, 0xdc, 0x24, 0xeb,
-	0xdb, 0x5a, 0x30, 0xb8, 0x7a, 0x30, 0x20, 0x71, 0x55, 0x0f, 0x06, 0x37, 0xc9, 0x9c, 0xdf, 0x1a,
-	0xd0, 0xd3, 0x6b, 0xbb, 0xd6, 0x75, 0x8c, 0x05, 0x5d, 0xc7, 0xd4, 0xbb, 0x0e, 0x7b, 0xab, 0xe8,
-	0x2e, 0xa2, 0x5b, 0x90, 0xfd, 0x1f, 0xa7, 0x31, 0x96, 0x61, 0x97, 0x08, 0x45, 0xc3, 0x79, 0x17,
-	0xba, 0x29, 0x0f, 0xbd, 0xab, 0xa2, 0x4d, 0x20, 0xff, 0x0a, 0xf2, 0xbb, 0x25, 0xda, 0xd5, 0x79,
-	0x9c, 0xbf, 0x9a, 0xd0, 0xd5, 0x88, 0x33, 0xb1, 0x6b, 0x7c, 0xcd, 0xd8, 0x35, 0x17, 0xc4, 0xee,
-	0x86, 0x52, 0x69, 0x72, 0xb2, 0x17, 0xa4, 0x32, 0x9d, 0x75, 0x54, 0xc1, 0x51, 0x49, 0x16, 0x1d,
-	0x85, 0xd5, 0x5e, 0x03, 0xb5, 0x54, 0x99, 0x46, 0xb3, 0x6d, 0x60, 0x84, 0xda, 0xf5, 0x72, 0xff,
-	0xec, 0xf3, 0x44, 0x46, 0x4f, 0x93, 0x42, 0x70, 0x0e, 0x85, 0xbd, 0x06, 0x8d, 0x2c, 0xf7, 0x46,
-	0x9c, 0x52, 0x65, 0x79, 0xa7, 0x43, 0xa1, 0x8d, 0x08, 0x57, 0xe0, 0x35, 0xe3, 0xb7, 0x5f, 0x60,
-	0x7c, 0xe7, 0x4f, 0x16, 0x2c, 0x55, 0xba, 0xf1, 0xbc, 0xa9, 0xa5, 0x3c, 0xd1, 0x5c, 0x70, 0xe2,
-	0x06, 0xd4, 0x27, 0x51, 0x20, 0x9c, 0xbd, 0xbc, 0xd3, 0x43, 0xfa, 0xe7, 0x51, 0x90, 0x63, 0x76,
-	0xb8, 0x44, 0xd1, 0x74, 0xaa, 0xbf, 0x28, 0x20, 0xde, 0x81, 0x1b, 0x65, 0x6a, 0xee, 0xed, 0x1d,
-	0x1c, 0xc4, 0xfe, 0x79, 0x51, 0xcb, 0xe7, 0x91, 0x18, 0x13, 0x33, 0x0b, 0x95, 0x98, 0x07, 0x35,
-	0x31, 0xb5, 0xfc, 0x3f, 0x34, 0x7c, 0x9c, 0x22, 0xc8, 0x4a, 0x32, 0xa0, 0xb4, 0xb1, 0xe2, 0x41,
-	0xcd, 0x15, 0x74, 0xf6, 0x06, 0xd4, 0x87, 0x93, 0x71, 0x22, 0x6d, 0xb5, 0x8c, 0x7c, 0x65, 0x5b,
-	0x7f, 0x50, 0x73, 0x89, 0x8a, 0x5c, 0x61, 0xec, 0x0d, 0xfb, 0x9d, 0x92, 0xab, 0xec, 0x7d, 0xc8,
-	0x85, 0x54, 0xe4, 0xc2, 0x9a, 0x41, 0xf5, 0x43, 0x72, 0x95, 0xe5, 0x1b, 0xb9, 0x90, 0xca, 0xde,
-	0x03, 0xb8, 0xf0, 0xc2, 0x60, 0x28, 0x9a, 0x45, 0x97, 0x78, 0xd7, 0x90, 0xf7, 0x69, 0x81, 0x95,
-	0x51, 0xaf, 0xf1, 0xdd, 0x6b, 0x43, 0x33, 0x13, 0xe1, 0xff, 0x5d, 0x58, 0xad, 0xf8, 0xec, 0x20,
-	0xc8, 0xc8, 0xc0, 0x82, 0xdc, 0x37, 0x16, 0x0d, 0x5a, 0xea, 0xfd, 0x01, 0x00, 0x59, 0xe2, 0x7e,
-	0x9a, 0xc6, 0xa9, 0x1a, 0xf8, 0x8c, 0x62, 0xe0, 0x73, 0xee, 0x40, 0x07, 0x2d, 0x70, 0x0d, 0x19,
-	0xaf, 0xbe, 0x88, 0x9c, 0x40, 0x8f, 0xee, 0xfc, 0xe4, 0x60, 0x01, 0x07, 0xdb, 0x81, 0x35, 0x31,
-	0x75, 0x89, 0x24, 0x78, 0x1c, 0x67, 0x01, 0x59, 0x42, 0xa4, 0xe3, 0x5c, 0x1a, 0xd6, 0x32, 0x8e,
-	0xe2, 0x8e, 0x9e, 0x1c, 0xa8, 0xb9, 0x40, 0xc1, 0xce, 0xb7, 0xa0, 0x83, 0x27, 0x8a, 0xe3, 0x36,
-	0xa1, 0x49, 0x04, 0x65, 0x07, 0xbb, 0x70, 0x82, 0x54, 0xc8, 0x95, 0x74, 0xe7, 0xe7, 0x06, 0x74,
-	0x45, 0x91, 0x13, 0x6f, 0xbe, 0x6c, 0x8d, 0xdb, 0xa8, 0xbc, 0xae, 0xaa, 0x84, 0x2e, 0x71, 0x1b,
-	0x80, 0xca, 0x94, 0x60, 0xa8, 0x97, 0x41, 0x51, 0x62, 0x5d, 0x8d, 0x03, 0x1d, 0x53, 0x42, 0x73,
-	0x4c, 0xfb, 0x6b, 0x13, 0x7a, 0xd2, 0xa5, 0x82, 0xe5, 0x7f, 0x94, 0xac, 0x32, 0x9f, 0xea, 0x7a,
-	0x3e, 0xbd, 0xa9, 0xf2, 0xa9, 0x51, 0x5e, 0xa3, 0x8c, 0xa2, 0x32, 0x9d, 0xee, 0xca, 0x74, 0x6a,
-	0x12, 0xdb, 0x92, 0x4a, 0x27, 0xc5, 0x25, 0xb2, 0xe9, 0xae, 0xcc, 0xa6, 0x56, 0xc9, 0x54, 0x84,
-	0x54, 0x91, 0x4c, 0x77, 0x65, 0x32, 0xb5, 0x4b, 0xa6, 0xc2, 0xcd, 0x2a, 0x97, 0xee, 0xb5, 0xa0,
-	0x41, 0xee, 0x74, 0x3e, 0x00, 0x5b, 0x37, 0x0d, 0xe5, 0xc4, 0x9b, 0x92, 0x58, 0x09, 0x05, 0x8d,
-	0xc9, 0x95, 0xef, 0x3e, 0x83, 0xa5, 0x4a, 0x29, 0xc2, 0x8e, 0x1f, 0x64, 0xbb, 0x5e, 0xe4, 0xf3,
-	0xb0, 0xd8, 0x3b, 0x34, 0x8c, 0x16, 0x64, 0x66, 0x29, 0x59, 0x8a, 0xa8, 0x04, 0x99, 0xb6, 0x3d,
-	0x58, 0x95, 0xed, 0xe1, 0x6f, 0x06, 0xf4, 0xf4, 0x17, 0x70, 0x01, 0xb9, 0x9f, 0xa6, 0xbb, 0xf1,
-	0x50, 0x78, 0xb3, 0xe1, 0x2a, 0x10, 0x43, 0x1f, 0x1f, 0x43, 0x2f, 0xcb, 0x64, 0x04, 0x16, 0xb0,
-	0xa4, 0x1d, 0xf9, 0x71, 0xa2, 0xf6, 0xc1, 0x02, 0x96, 0xb4, 0x03, 0x7e, 0xc1, 0x43, 0xd9, 0xa0,
-	0x0a, 0x18, 0x4f, 0x7b, 0xc4, 0xb3, 0x0c, 0xc3, 0x44, 0xd4, 0x55, 0x05, 0xe2, 0x5b, 0xae, 0x77,
+	0x16, 0x6f, 0x9e, 0x66, 0xf9, 0x41, 0xec, 0x7b, 0x79, 0x10, 0x47, 0x52, 0xf1, 0x2a, 0x92, 0x12,
+	0xf7, 0x2a, 0xf2, 0x29, 0xf2, 0x2c, 0x4a, 0x5c, 0x82, 0xf0, 0x8d, 0x27, 0x91, 0xa4, 0x34, 0x88,
+	0x52, 0xc0, 0xce, 0x3f, 0xeb, 0x00, 0x47, 0x57, 0x91, 0x3f, 0x15, 0x63, 0xf7, 0x2f, 0x78, 0x94,
+	0x57, 0x63, 0x4c, 0xa0, 0x50, 0x98, 0x08, 0xb9, 0x44, 0x19, 0xb7, 0x80, 0xd9, 0x6d, 0xe8, 0xa4,
+	0xdc, 0xe7, 0x51, 0x8e, 0x44, 0x8b, 0x88, 0x25, 0x02, 0xa3, 0x69, 0xec, 0x65, 0x39, 0x4f, 0x2b,
+	0xe6, 0xad, 0xe0, 0xd8, 0x16, 0xd8, 0x3a, 0xbc, 0x9f, 0x07, 0x43, 0x69, 0xe2, 0x19, 0x3c, 0xca,
+	0xa3, 0x97, 0x50, 0xf2, 0x9a, 0x42, 0x9e, 0x8e, 0x43, 0x79, 0x3a, 0x4c, 0xf2, 0x44, 0x94, 0xcd,
+	0xe0, 0x51, 0xde, 0x49, 0x18, 0xfb, 0xe7, 0x41, 0x34, 0x22, 0x07, 0xb4, 0xc9, 0x54, 0x15, 0x1c,
+	0xfb, 0x0e, 0xd8, 0x93, 0x28, 0xe5, 0x59, 0x1c, 0x5e, 0xf0, 0x21, 0xf9, 0x31, 0xeb, 0x77, 0xb4,
+	0xb2, 0xa3, 0x7b, 0xd8, 0x9d, 0x61, 0xd5, 0x3c, 0x04, 0xa2, 0xd2, 0x48, 0x0f, 0x0d, 0x00, 0x4e,
+	0x48, 0x91, 0xe3, 0xab, 0x84, 0xf7, 0xbb, 0x22, 0xee, 0x4a, 0x0c, 0x7b, 0x07, 0x6e, 0x64, 0xdc,
+	0x8f, 0xa3, 0x61, 0x76, 0x8f, 0x9f, 0x05, 0xd1, 0xf0, 0x11, 0xd9, 0xa2, 0xdf, 0x23, 0x13, 0xcf,
+	0x23, 0x61, 0xc4, 0x90, 0xe2, 0x7b, 0x7b, 0x07, 0x87, 0x97, 0x11, 0x4f, 0xfb, 0x4b, 0x22, 0x62,
+	0x2a, 0x48, 0x74, 0xb7, 0x1f, 0x47, 0xa7, 0x61, 0xe0, 0xe7, 0x8f, 0xb2, 0x51, 0x7f, 0x99, 0x78,
+	0x74, 0x14, 0xba, 0x34, 0x2f, 0xd2, 0x7a, 0x45, 0xb8, 0xb4, 0x40, 0x14, 0xc1, 0xe0, 0x26, 0x59,
+	0xdf, 0xd6, 0x82, 0xc1, 0xd5, 0x83, 0x01, 0x89, 0xab, 0x7a, 0x30, 0xb8, 0x49, 0xe6, 0xfc, 0xc6,
+	0x80, 0x9e, 0x5e, 0xdb, 0xb5, 0xae, 0x63, 0x2c, 0xe8, 0x3a, 0xa6, 0xde, 0x75, 0xd8, 0x5b, 0x45,
+	0x77, 0x11, 0xdd, 0x82, 0xec, 0xff, 0x38, 0x8d, 0xb1, 0x0c, 0xbb, 0x44, 0x28, 0x1a, 0xce, 0xbb,
+	0xd0, 0x4d, 0x79, 0xe8, 0x5d, 0x15, 0x6d, 0x02, 0xf9, 0x57, 0x90, 0xdf, 0x2d, 0xd1, 0xae, 0xce,
+	0xe3, 0xfc, 0xc5, 0x84, 0xae, 0x46, 0x9c, 0x89, 0x5d, 0xe3, 0x6b, 0xc6, 0xae, 0xb9, 0x20, 0x76,
+	0x37, 0x94, 0x4a, 0x93, 0x93, 0xbd, 0x20, 0x95, 0xe9, 0xac, 0xa3, 0x0a, 0x8e, 0x4a, 0xb2, 0xe8,
+	0x28, 0xac, 0xf6, 0x1a, 0xa8, 0xa5, 0xca, 0x34, 0x9a, 0x6d, 0x03, 0x23, 0xd4, 0xae, 0x97, 0xfb,
+	0x67, 0x9f, 0x27, 0x32, 0x7a, 0x9a, 0x14, 0x82, 0x73, 0x28, 0xec, 0x35, 0x68, 0x64, 0xb9, 0x37,
+	0xe2, 0x94, 0x2a, 0xcb, 0x3b, 0x1d, 0x0a, 0x6d, 0x44, 0xb8, 0x02, 0xaf, 0x19, 0xbf, 0xfd, 0x02,
+	0xe3, 0x3b, 0x7f, 0xb4, 0x60, 0xa9, 0xd2, 0x8d, 0xe7, 0x4d, 0x2d, 0xe5, 0x8d, 0xe6, 0x82, 0x1b,
+	0x37, 0xa0, 0x3e, 0x89, 0x02, 0xe1, 0xec, 0xe5, 0x9d, 0x1e, 0xd2, 0x3f, 0x8f, 0x82, 0x1c, 0xb3,
+	0xc3, 0x25, 0x8a, 0xa6, 0x53, 0xfd, 0x45, 0x01, 0xf1, 0x0e, 0xdc, 0x28, 0x53, 0x73, 0x6f, 0xef,
+	0xe0, 0x20, 0xf6, 0xcf, 0x8b, 0x5a, 0x3e, 0x8f, 0xc4, 0x98, 0x98, 0x59, 0xa8, 0xc4, 0x3c, 0xa8,
+	0x89, 0xa9, 0xe5, 0xff, 0xa1, 0xe1, 0xe3, 0x14, 0x41, 0x56, 0x92, 0x01, 0xa5, 0x8d, 0x15, 0x0f,
+	0x6a, 0xae, 0xa0, 0xb3, 0x37, 0xa0, 0x3e, 0x9c, 0x8c, 0x13, 0x69, 0xab, 0x65, 0xe4, 0x2b, 0xdb,
+	0xfa, 0x83, 0x9a, 0x4b, 0x54, 0xe4, 0x0a, 0x63, 0x6f, 0xd8, 0xef, 0x94, 0x5c, 0x65, 0xef, 0x43,
+	0x2e, 0xa4, 0x22, 0x17, 0xd6, 0x0c, 0xaa, 0x1f, 0x92, 0xab, 0x2c, 0xdf, 0xc8, 0x85, 0x54, 0xf6,
+	0x1e, 0xc0, 0x85, 0x17, 0x06, 0x43, 0xd1, 0x2c, 0xba, 0xc4, 0xbb, 0x86, 0xbc, 0x4f, 0x0b, 0xac,
+	0x8c, 0x7a, 0x8d, 0xef, 0x5e, 0x1b, 0x9a, 0x99, 0x08, 0xff, 0xef, 0xc2, 0x6a, 0xc5, 0x67, 0x07,
+	0x41, 0x46, 0x06, 0x16, 0xe4, 0xbe, 0xb1, 0x68, 0xd0, 0x52, 0xe7, 0x07, 0x00, 0x64, 0x89, 0xfb,
+	0x69, 0x1a, 0xa7, 0x6a, 0xe0, 0x33, 0x8a, 0x81, 0xcf, 0xb9, 0x03, 0x1d, 0xb4, 0xc0, 0x35, 0x64,
+	0x7c, 0xf5, 0x45, 0xe4, 0x04, 0x7a, 0xf4, 0xce, 0x4f, 0x0e, 0x16, 0x70, 0xb0, 0x1d, 0x58, 0x13,
+	0x53, 0x97, 0x48, 0x82, 0xc7, 0x71, 0x16, 0x90, 0x25, 0x44, 0x3a, 0xce, 0xa5, 0x61, 0x2d, 0xe3,
+	0x28, 0xee, 0xe8, 0xc9, 0x81, 0x9a, 0x0b, 0x14, 0xec, 0x7c, 0x0b, 0x3a, 0x78, 0xa3, 0xb8, 0x6e,
+	0x13, 0x9a, 0x44, 0x50, 0x76, 0xb0, 0x0b, 0x27, 0x48, 0x85, 0x5c, 0x49, 0x77, 0x7e, 0x66, 0x40,
+	0x57, 0x14, 0x39, 0x71, 0xf2, 0x65, 0x6b, 0xdc, 0x46, 0xe5, 0xb8, 0xaa, 0x12, 0xba, 0xc4, 0x6d,
+	0x00, 0x2a, 0x53, 0x82, 0xa1, 0x5e, 0x06, 0x45, 0x89, 0x75, 0x35, 0x0e, 0x74, 0x4c, 0x09, 0xcd,
+	0x31, 0xed, 0xaf, 0x4c, 0xe8, 0x49, 0x97, 0x0a, 0x96, 0xff, 0x51, 0xb2, 0xca, 0x7c, 0xaa, 0xeb,
+	0xf9, 0xf4, 0xa6, 0xca, 0xa7, 0x46, 0xf9, 0x1a, 0x65, 0x14, 0x95, 0xe9, 0x74, 0x57, 0xa6, 0x53,
+	0x93, 0xd8, 0x96, 0x54, 0x3a, 0x29, 0x2e, 0x91, 0x4d, 0x77, 0x65, 0x36, 0xb5, 0x4a, 0xa6, 0x22,
+	0xa4, 0x8a, 0x64, 0xba, 0x2b, 0x93, 0xa9, 0x5d, 0x32, 0x15, 0x6e, 0x56, 0xb9, 0x74, 0xaf, 0x05,
+	0x0d, 0x72, 0xa7, 0xf3, 0x01, 0xd8, 0xba, 0x69, 0x28, 0x27, 0xde, 0x94, 0xc4, 0x4a, 0x28, 0x68,
+	0x4c, 0xae, 0x3c, 0xfb, 0x0c, 0x96, 0x2a, 0xa5, 0x08, 0x3b, 0x7e, 0x90, 0xed, 0x7a, 0x91, 0xcf,
+	0xc3, 0x62, 0xef, 0xd0, 0x30, 0x5a, 0x90, 0x99, 0xa5, 0x64, 0x29, 0xa2, 0x12, 0x64, 0xda, 0xf6,
+	0x60, 0x55, 0xb6, 0x87, 0xbf, 0x1a, 0xd0, 0xd3, 0x0f, 0xe0, 0x02, 0x72, 0x3f, 0x4d, 0x77, 0xe3,
+	0xa1, 0xf0, 0x66, 0xc3, 0x55, 0x20, 0x86, 0x3e, 0x3e, 0x86, 0x5e, 0x96, 0xc9, 0x08, 0x2c, 0x60,
+	0x49, 0x3b, 0xf2, 0xe3, 0x44, 0xed, 0x83, 0x05, 0x2c, 0x69, 0x07, 0xfc, 0x82, 0x87, 0xb2, 0x41,
+	0x15, 0x30, 0xde, 0xf6, 0x88, 0x67, 0x19, 0x86, 0x89, 0xa8, 0xab, 0x0a, 0xc4, 0x53, 0xae, 0x77,
 	0xb9, 0xeb, 0x4d, 0x32, 0x2e, 0x67, 0xb6, 0x02, 0x46, 0xb3, 0xe0, 0xde, 0xea, 0xa5, 0xf1, 0x24,
 	0x52, 0x93, 0x9a, 0x86, 0x71, 0x2e, 0x61, 0xf5, 0xf1, 0x24, 0x1d, 0x71, 0x0a, 0x62, 0xb5, 0x06,
 	0xaf, 0x43, 0x3b, 0x88, 0x3c, 0x3f, 0x0f, 0x2e, 0xb8, 0xb4, 0x64, 0x01, 0x63, 0xfc, 0xe6, 0xc1,
-	0x98, 0xcb, 0x51, 0x95, 0x9e, 0x91, 0xff, 0x34, 0x08, 0x39, 0xc5, 0xb5, 0xbc, 0x92, 0x82, 0x29,
-	0x45, 0x45, 0x4f, 0x96, 0x4b, 0xae, 0x80, 0x9c, 0xdf, 0x98, 0xb0, 0x7e, 0x98, 0xf0, 0xd4, 0xcb,
-	0xb9, 0x58, 0xac, 0x8f, 0xfc, 0x33, 0x3e, 0xf6, 0x94, 0x0a, 0xb7, 0xc1, 0x8c, 0x13, 0x3a, 0x5c,
-	0xc6, 0xbb, 0x20, 0x1f, 0x26, 0xae, 0x19, 0x27, 0xa4, 0x84, 0x97, 0x9d, 0x4b, 0xdb, 0xd2, 0xf3,
-	0xc2, 0x2d, 0x7b, 0x1d, 0xda, 0x43, 0x2f, 0xf7, 0x4e, 0xbc, 0x8c, 0x2b, 0x9b, 0x2a, 0x98, 0x16,
-	0x52, 0xdc, 0xdf, 0xa4, 0x45, 0x05, 0x40, 0x92, 0xe8, 0x34, 0x69, 0x4d, 0x09, 0x21, 0xf7, 0x69,
-	0x38, 0xc9, 0xce, 0xc8, 0x8c, 0x6d, 0x57, 0x00, 0xa8, 0x4b, 0x11, 0xf3, 0x6d, 0xd9, 0x2e, 0x06,
-	0x00, 0xa7, 0x69, 0x3c, 0x16, 0x85, 0x85, 0x1a, 0x50, 0xdb, 0xd5, 0x30, 0x8a, 0x7e, 0x2c, 0xd6,
-	0x15, 0x28, 0xe9, 0x02, 0xe3, 0xe4, 0xb0, 0xf4, 0xf4, 0x5d, 0x19, 0xf6, 0x8f, 0x78, 0xee, 0xb1,
-	0x75, 0xcd, 0x1c, 0x80, 0xe6, 0x40, 0x8a, 0x34, 0xc6, 0x0b, 0xab, 0x87, 0x2a, 0x39, 0x96, 0x56,
-	0x72, 0x94, 0x05, 0xeb, 0x14, 0xe2, 0xf4, 0xec, 0xbc, 0x07, 0x6b, 0xd2, 0x23, 0x4f, 0xdf, 0xc5,
-	0x53, 0x17, 0xfa, 0x42, 0x90, 0xc5, 0xf1, 0xce, 0x5f, 0x0c, 0xb8, 0x39, 0xf5, 0xda, 0x4b, 0x7f,
-	0xaf, 0x78, 0x1f, 0xea, 0xb8, 0xf0, 0xf5, 0x2d, 0x4a, 0xcd, 0xbb, 0x78, 0xc6, 0x5c, 0x91, 0xdb,
-	0x08, 0xdc, 0x8f, 0xf2, 0xf4, 0xca, 0xa5, 0x17, 0xd6, 0x3f, 0x85, 0x4e, 0x81, 0x42, 0xb9, 0xe7,
-	0xfc, 0x4a, 0x55, 0xdf, 0x73, 0x7e, 0x85, 0x13, 0xc5, 0x85, 0x17, 0x4e, 0x84, 0x69, 0x64, 0x83,
-	0xad, 0x18, 0xd6, 0x15, 0xf4, 0x0f, 0xcc, 0x6f, 0x1b, 0xce, 0x8f, 0xa1, 0xff, 0xc0, 0x8b, 0x86,
-	0xa1, 0x8c, 0x47, 0x51, 0x14, 0xa4, 0x09, 0x5e, 0xd5, 0x4c, 0xd0, 0x45, 0x29, 0x44, 0xbd, 0x26,
-	0x1a, 0x6f, 0x43, 0xe7, 0x44, 0xb5, 0x43, 0x69, 0xf8, 0x12, 0x41, 0x31, 0xf3, 0x2c, 0xcc, 0xe4,
-	0x5a, 0x49, 0xcf, 0xce, 0x4d, 0xb8, 0xb1, 0xcf, 0x73, 0x71, 0xf6, 0xee, 0xe9, 0x48, 0x9e, 0xec,
-	0x6c, 0xc2, 0x5a, 0x15, 0x2d, 0x8d, 0x6b, 0x83, 0xe5, 0x9f, 0x16, 0xad, 0xc6, 0x3f, 0x1d, 0x39,
-	0x47, 0x70, 0x47, 0x4c, 0x4b, 0x93, 0x13, 0x54, 0x01, 0x4b, 0xdf, 0xe7, 0xc9, 0xd0, 0xcb, 0xb9,
-	0xba, 0xc4, 0x0e, 0xac, 0x65, 0x82, 0xb6, 0x7b, 0x3a, 0x3a, 0x8e, 0xc7, 0xe1, 0x51, 0x9e, 0x06,
+	0x98, 0xcb, 0x51, 0x95, 0x9e, 0x91, 0xff, 0x34, 0x08, 0x39, 0xc5, 0xb5, 0x7c, 0x25, 0x05, 0x53,
+	0x8a, 0x8a, 0x9e, 0x2c, 0x97, 0x5c, 0x01, 0x39, 0xbf, 0x36, 0x61, 0xfd, 0x30, 0xe1, 0xa9, 0x97,
+	0x73, 0xb1, 0x58, 0x1f, 0xf9, 0x67, 0x7c, 0xec, 0x29, 0x15, 0x6e, 0x83, 0x19, 0x27, 0x74, 0xb9,
+	0x8c, 0x77, 0x41, 0x3e, 0x4c, 0x5c, 0x33, 0x4e, 0x48, 0x09, 0x2f, 0x3b, 0x97, 0xb6, 0xa5, 0xe7,
+	0x85, 0x5b, 0xf6, 0x3a, 0xb4, 0x87, 0x5e, 0xee, 0x9d, 0x78, 0x19, 0x57, 0x36, 0x55, 0x30, 0x2d,
+	0xa4, 0xb8, 0xbf, 0x49, 0x8b, 0x0a, 0x80, 0x24, 0xd1, 0x6d, 0xd2, 0x9a, 0x12, 0x42, 0xee, 0xd3,
+	0x70, 0x92, 0x9d, 0x91, 0x19, 0xdb, 0xae, 0x00, 0x50, 0x97, 0x22, 0xe6, 0xdb, 0xb2, 0x5d, 0x0c,
+	0x00, 0x4e, 0xd3, 0x78, 0x2c, 0x0a, 0x0b, 0x35, 0xa0, 0xb6, 0xab, 0x61, 0x14, 0xfd, 0x58, 0xac,
+	0x2b, 0x50, 0xd2, 0x05, 0xc6, 0xc9, 0x61, 0xe9, 0xe9, 0xbb, 0x32, 0xec, 0x1f, 0xf1, 0xdc, 0x63,
+	0xeb, 0x9a, 0x39, 0x00, 0xcd, 0x81, 0x14, 0x69, 0x8c, 0x17, 0x56, 0x0f, 0x55, 0x72, 0x2c, 0xad,
+	0xe4, 0x28, 0x0b, 0xd6, 0x29, 0xc4, 0xe9, 0xd9, 0x79, 0x0f, 0xd6, 0xa4, 0x47, 0x9e, 0xbe, 0x8b,
+	0xb7, 0x2e, 0xf4, 0x85, 0x20, 0x8b, 0xeb, 0x9d, 0x3f, 0x1b, 0x70, 0x73, 0xea, 0xd8, 0x4b, 0x7f,
+	0xaf, 0x78, 0x1f, 0xea, 0xb8, 0xf0, 0xf5, 0x2d, 0x4a, 0xcd, 0xbb, 0x78, 0xc7, 0x5c, 0x91, 0xdb,
+	0x08, 0xdc, 0x8f, 0xf2, 0xf4, 0xca, 0xa5, 0x03, 0xeb, 0x9f, 0x42, 0xa7, 0x40, 0xa1, 0xdc, 0x73,
+	0x7e, 0xa5, 0xaa, 0xef, 0x39, 0xbf, 0xc2, 0x89, 0xe2, 0xc2, 0x0b, 0x27, 0xc2, 0x34, 0xb2, 0xc1,
+	0x56, 0x0c, 0xeb, 0x0a, 0xfa, 0x07, 0xe6, 0xb7, 0x0d, 0xe7, 0xc7, 0xd0, 0x7f, 0xe0, 0x45, 0xc3,
+	0x50, 0xc6, 0xa3, 0x28, 0x0a, 0xd2, 0x04, 0xaf, 0x6a, 0x26, 0xe8, 0xa2, 0x14, 0xa2, 0x5e, 0x13,
+	0x8d, 0xb7, 0xa1, 0x73, 0xa2, 0xda, 0xa1, 0x34, 0x7c, 0x89, 0xa0, 0x98, 0x79, 0x16, 0x66, 0x72,
+	0xad, 0xa4, 0x67, 0xe7, 0x26, 0xdc, 0xd8, 0xe7, 0xb9, 0xb8, 0x7b, 0xf7, 0x74, 0x24, 0x6f, 0x76,
+	0x36, 0x61, 0xad, 0x8a, 0x96, 0xc6, 0xb5, 0xc1, 0xf2, 0x4f, 0x8b, 0x56, 0xe3, 0x9f, 0x8e, 0x9c,
+	0x23, 0xb8, 0x23, 0xa6, 0xa5, 0xc9, 0x09, 0xaa, 0x80, 0xa5, 0xef, 0xf3, 0x64, 0xe8, 0xe5, 0x5c,
+	0xbd, 0xc4, 0x0e, 0xac, 0x65, 0x82, 0xb6, 0x7b, 0x3a, 0x3a, 0x8e, 0xc7, 0xe1, 0x51, 0x9e, 0x06,
 	0x91, 0x92, 0x31, 0x97, 0xe6, 0x1c, 0xc0, 0x60, 0x91, 0x50, 0xa9, 0x48, 0x1f, 0x5a, 0xf2, 0x63,
 	0x8d, 0x74, 0xb3, 0x02, 0x67, 0xfd, 0xec, 0x8c, 0x60, 0x7d, 0x9f, 0xe7, 0x33, 0x33, 0x53, 0x59,
-	0x76, 0xf0, 0x8c, 0xcf, 0xca, 0xf6, 0x58, 0xc0, 0xec, 0x1b, 0xd0, 0x3b, 0x0d, 0xc2, 0x9c, 0xa7,
-	0x72, 0xe7, 0x98, 0x89, 0xf5, 0x0a, 0xd9, 0xf9, 0xa9, 0x05, 0xf6, 0xf4, 0x31, 0x85, 0x9f, 0x8c,
+	0x76, 0xf0, 0x8e, 0xcf, 0xca, 0xf6, 0x58, 0xc0, 0xec, 0x1b, 0xd0, 0x3b, 0x0d, 0xc2, 0x9c, 0xa7,
+	0x72, 0xe7, 0x98, 0x89, 0xf5, 0x0a, 0xd9, 0xf9, 0xbb, 0x05, 0xf6, 0xf4, 0x35, 0x85, 0x9f, 0x8c,
 	0xb9, 0x55, 0xc3, 0xac, 0x54, 0x0d, 0x06, 0xf5, 0x31, 0x16, 0x76, 0x99, 0x33, 0xf8, 0x5c, 0x26,
 	0x5a, 0x7d, 0x41, 0xa2, 0x6d, 0xc2, 0x8a, 0x9c, 0xfe, 0x62, 0xb5, 0xd7, 0xc8, 0x05, 0x62, 0x0a,
 	0x8d, 0x03, 0xf3, 0x14, 0x8a, 0xd6, 0x0d, 0x51, 0x6f, 0xe6, 0x91, 0xb4, 0x69, 0xbc, 0xf5, 0x35,
 	0xa6, 0xf1, 0x44, 0x10, 0xc4, 0x27, 0x25, 0x69, 0xb2, 0xb6, 0x10, 0x3e, 0x87, 0xc4, 0xde, 0x86,
-	0xd5, 0x84, 0x47, 0xb8, 0x68, 0x6b, 0xfc, 0x1d, 0xe2, 0x9f, 0x25, 0xe0, 0x35, 0xa9, 0x55, 0x6a,
-	0xbc, 0x20, 0xae, 0x39, 0x85, 0x76, 0x7e, 0x6f, 0xc0, 0xcd, 0xd2, 0x0d, 0xf4, 0xa9, 0xec, 0x05,
-	0xdb, 0xe9, 0x3a, 0xb4, 0xb3, 0xd4, 0x27, 0x4e, 0xd5, 0x39, 0x15, 0x4c, 0x95, 0x3c, 0xcb, 0x05,
-	0x4d, 0xb6, 0x19, 0x05, 0xbf, 0xd8, 0x37, 0x7d, 0x68, 0x8d, 0xab, 0xed, 0x53, 0x82, 0xce, 0x9f,
-	0x0d, 0x78, 0x75, 0x6e, 0x54, 0xfe, 0x17, 0x9f, 0x5d, 0xa1, 0x70, 0x5d, 0x26, 0x8b, 0xd9, 0xf5,
-	0x5b, 0x02, 0xce, 0x1b, 0x1f, 0xc2, 0x52, 0x5e, 0x5a, 0x86, 0xab, 0xcf, 0xae, 0xaf, 0x54, 0x5f,
-	0xd4, 0x8c, 0xe7, 0x56, 0xf9, 0x9d, 0x73, 0x78, 0xa5, 0xa2, 0x7f, 0xa5, 0x72, 0xed, 0xd0, 0x14,
-	0x8e, 0xbc, 0x5c, 0xd6, 0xaf, 0x5b, 0x9a, 0x60, 0x31, 0xf5, 0x12, 0xd5, 0x2d, 0xf8, 0x2a, 0x89,
-	0x68, 0x56, 0x13, 0xd1, 0xf9, 0x9d, 0x09, 0x2b, 0x53, 0x47, 0xb1, 0x65, 0x30, 0x83, 0xa1, 0x74,
-	0xa4, 0x19, 0x0c, 0x17, 0x26, 0x95, 0xee, 0x5c, 0x6b, 0xca, 0xb9, 0x58, 0x46, 0x52, 0x7f, 0xcf,
-	0xcb, 0x3d, 0xd9, 0xa5, 0x15, 0x58, 0x71, 0x7b, 0x63, 0xca, 0xed, 0x7d, 0x68, 0x0d, 0xb3, 0x9c,
-	0xde, 0x12, 0xb9, 0xa3, 0x40, 0x2c, 0xc0, 0x14, 0x8d, 0xf4, 0x01, 0x48, 0xcc, 0x3d, 0x25, 0x82,
-	0x6d, 0x17, 0xab, 0x57, 0xfb, 0x5a, 0x9b, 0x48, 0xae, 0x62, 0xea, 0xe9, 0xc8, 0xd2, 0x81, 0x53,
-	0x8f, 0x16, 0x51, 0x50, 0x8d, 0xa8, 0x67, 0x53, 0x65, 0x4e, 0x3a, 0xe4, 0xa5, 0xe3, 0xe9, 0x2d,
-	0x35, 0x0c, 0x8b, 0x50, 0xba, 0x51, 0x8d, 0x88, 0xca, 0x3c, 0xfc, 0x2b, 0x03, 0xee, 0xa8, 0x96,
-	0x39, 0x3f, 0x10, 0xee, 0x6a, 0x2d, 0x6c, 0x56, 0x92, 0x6c, 0x65, 0x34, 0x45, 0x7f, 0x1c, 0x86,
-	0x62, 0xfd, 0x31, 0xd5, 0x14, 0xad, 0x30, 0x95, 0xc8, 0xb0, 0xa6, 0x4a, 0xf4, 0x1a, 0x69, 0xfb,
-	0x50, 0x7c, 0xa6, 0xaf, 0xbb, 0x02, 0x70, 0x3e, 0x85, 0xc1, 0x22, 0xbd, 0x5e, 0xd6, 0x1e, 0xce,
-	0x15, 0xdc, 0x11, 0xcd, 0xa7, 0x14, 0xa5, 0x7e, 0x94, 0x79, 0x71, 0x07, 0xa9, 0x74, 0x64, 0x73,
-	0xba, 0x23, 0x17, 0x1f, 0x0c, 0xe9, 0x23, 0xb4, 0xa5, 0x7f, 0x30, 0x44, 0x0c, 0x5e, 0x63, 0xd1,
-	0xd1, 0x2f, 0x7b, 0x8d, 0xad, 0x73, 0x68, 0x8a, 0xf1, 0x8d, 0x2d, 0x41, 0xe7, 0x61, 0x44, 0xa5,
-	0xe0, 0x30, 0xb1, 0x6b, 0xac, 0x0d, 0xf5, 0xa3, 0x3c, 0x4e, 0x6c, 0x83, 0x75, 0xa0, 0xf1, 0x18,
-	0xe7, 0x77, 0xdb, 0x64, 0x00, 0x4d, 0xac, 0xef, 0x63, 0x6e, 0x5b, 0x88, 0x3e, 0xca, 0xbd, 0x34,
-	0xb7, 0xeb, 0x88, 0x16, 0x0a, 0xd9, 0x0d, 0xb6, 0x0c, 0xf0, 0xf1, 0x24, 0x8f, 0x25, 0x5b, 0x13,
-	0x69, 0x7b, 0x3c, 0xe4, 0x39, 0xb7, 0x5b, 0x5b, 0x3f, 0xa1, 0x57, 0x46, 0x38, 0x30, 0xf4, 0xe4,
-	0x59, 0x04, 0xdb, 0x35, 0xd6, 0x02, 0xeb, 0x33, 0x7e, 0x69, 0x1b, 0xac, 0x0b, 0x2d, 0x77, 0x12,
-	0x45, 0x41, 0x34, 0x12, 0xe7, 0xd1, 0xd1, 0x43, 0xdb, 0x42, 0x02, 0x2a, 0x94, 0xf0, 0xa1, 0x5d,
-	0x67, 0x3d, 0x68, 0x7f, 0x22, 0x7f, 0x18, 0xb0, 0x1b, 0x48, 0x42, 0x36, 0x7c, 0xa7, 0x89, 0x24,
-	0x3a, 0x1c, 0xa1, 0x16, 0x42, 0xf4, 0x16, 0x42, 0xed, 0xad, 0x43, 0x68, 0xab, 0x5d, 0x95, 0xad,
-	0x40, 0x57, 0xea, 0x80, 0x28, 0xbb, 0x86, 0x17, 0xa2, 0xf1, 0xc2, 0x36, 0xf0, 0xf2, 0xb8, 0x75,
-	0xda, 0x26, 0x3e, 0xe1, 0x6a, 0x69, 0x5b, 0x64, 0x90, 0xab, 0xc8, 0xb7, 0xeb, 0xc8, 0x48, 0x2b,
-	0x8a, 0x3d, 0xdc, 0x7a, 0x04, 0x2d, 0x7a, 0x3c, 0xc4, 0xc9, 0x6b, 0x59, 0xca, 0x93, 0x18, 0xbb,
-	0x86, 0x36, 0xc5, 0xd3, 0x05, 0xb7, 0x81, 0xb6, 0xa1, 0xeb, 0x08, 0xd8, 0x44, 0x15, 0x84, 0x9d,
-	0x04, 0xc2, 0xda, 0xfa, 0x99, 0x01, 0x6d, 0xb5, 0x5c, 0xb0, 0x1b, 0xb0, 0xa2, 0x8c, 0x24, 0x51,
-	0x42, 0xe2, 0x3e, 0xcf, 0x05, 0xc2, 0x36, 0xe8, 0x80, 0x02, 0x34, 0xd1, 0xae, 0x2e, 0x1f, 0xc7,
-	0x17, 0x5c, 0x62, 0x2c, 0x3c, 0x12, 0x77, 0x59, 0x09, 0xd7, 0xf1, 0x05, 0x84, 0xa9, 0x62, 0xd9,
-	0x0d, 0x76, 0x0b, 0x18, 0x82, 0x8f, 0x82, 0x11, 0x66, 0x85, 0x98, 0xf8, 0x33, 0xbb, 0xb9, 0xf5,
-	0x11, 0xb4, 0xd5, 0x60, 0xad, 0xe9, 0xa1, 0x50, 0x85, 0x1e, 0x02, 0x61, 0x1b, 0xe5, 0xc1, 0x12,
-	0x63, 0x6e, 0x3d, 0xa5, 0x85, 0x14, 0xe7, 0x52, 0xcd, 0x32, 0x12, 0x23, 0xc3, 0xeb, 0x3c, 0x48,
-	0xa4, 0xc3, 0x79, 0x12, 0x7a, 0x7e, 0x11, 0x60, 0x17, 0x3c, 0xcd, 0x6d, 0x0b, 0x9f, 0x1f, 0x46,
-	0x3f, 0xe2, 0x3e, 0x46, 0x18, 0xba, 0x21, 0xc8, 0x72, 0xbb, 0xb1, 0x75, 0x00, 0xdd, 0xa7, 0xaa,
-	0x5f, 0x1d, 0x26, 0x78, 0x01, 0xa5, 0x5c, 0x89, 0xb5, 0x6b, 0x78, 0x26, 0x45, 0x67, 0x81, 0xb5,
-	0x0d, 0xb6, 0x0a, 0x4b, 0xe8, 0x8d, 0x12, 0x65, 0x6e, 0x3d, 0x01, 0x36, 0x5b, 0x69, 0xd1, 0x68,
-	0xa5, 0xc2, 0x76, 0x0d, 0x35, 0xf9, 0x8c, 0x5f, 0xe2, 0x33, 0xf9, 0xf0, 0xe1, 0x28, 0x8a, 0x53,
-	0x4e, 0x34, 0xe5, 0x43, 0xfa, 0xa2, 0x88, 0x08, 0x6b, 0xeb, 0xe9, 0x54, 0x4f, 0x3a, 0x4c, 0xb4,
-	0x70, 0x27, 0xd8, 0xae, 0x51, 0xf0, 0x91, 0x14, 0x81, 0x90, 0x06, 0x24, 0x31, 0x02, 0x63, 0xe2,
-	0x41, 0xbb, 0x21, 0xf7, 0x52, 0x01, 0x5b, 0x3b, 0xff, 0x6a, 0x42, 0x53, 0xa4, 0x39, 0xfb, 0x08,
-	0xba, 0xda, 0x6f, 0xb2, 0x8c, 0x1a, 0xc6, 0xec, 0x2f, 0xc8, 0xeb, 0xff, 0x37, 0x83, 0x17, 0xe5,
-	0xc1, 0xa9, 0xb1, 0x0f, 0x01, 0xca, 0x55, 0x9b, 0xdd, 0xa4, 0xf9, 0x6d, 0x7a, 0xf5, 0x5e, 0xef,
-	0xd3, 0x47, 0x9a, 0x39, 0xbf, 0x37, 0x3b, 0x35, 0xf6, 0x3d, 0x58, 0x92, 0xa5, 0x54, 0x84, 0x16,
-	0x1b, 0x68, 0x8b, 0xd2, 0x9c, 0x25, 0xfa, 0x5a, 0x61, 0x9f, 0x14, 0xc2, 0x44, 0xf8, 0xb0, 0xfe,
-	0x9c, 0xad, 0x4b, 0x88, 0x79, 0x65, 0xe1, 0x3e, 0xe6, 0xd4, 0xd8, 0x3e, 0x74, 0xc5, 0xd6, 0x24,
-	0x1a, 0xc4, 0x6d, 0xe4, 0x5d, 0xb4, 0x46, 0x5d, 0xab, 0xd0, 0x2e, 0xf4, 0xf4, 0x45, 0x87, 0x91,
-	0x25, 0xe7, 0x6c, 0x44, 0x42, 0xc8, 0xbc, 0x9d, 0xc8, 0xa9, 0x31, 0x0f, 0x6e, 0xcd, 0x5f, 0x57,
-	0xd8, 0xeb, 0xe5, 0xd7, 0xe4, 0x05, 0xfb, 0xd1, 0xba, 0x73, 0x1d, 0x4b, 0x71, 0xc4, 0x0f, 0xa0,
-	0x5f, 0x1c, 0x5e, 0x84, 0xb5, 0x8c, 0x8a, 0x81, 0x54, 0x6d, 0xc1, 0x86, 0xb3, 0xfe, 0xda, 0x42,
-	0x7a, 0x21, 0xfe, 0x18, 0x56, 0x4b, 0x86, 0x58, 0x98, 0x8f, 0xdd, 0x99, 0x79, 0xaf, 0x62, 0xd6,
-	0xc1, 0x22, 0x72, 0x21, 0xf5, 0x87, 0xe5, 0x8e, 0x5e, 0x95, 0xfc, 0xba, 0xee, 0xdb, 0xf9, 0xd2,
-	0x9d, 0xeb, 0x58, 0x8a, 0x13, 0x1e, 0xc3, 0x4a, 0xa5, 0x41, 0x2a, 0xd9, 0xd7, 0x36, 0xec, 0xeb,
-	0x02, 0xe2, 0x5e, 0xff, 0x8b, 0xaf, 0x06, 0xc6, 0x97, 0x5f, 0x0d, 0x8c, 0x7f, 0x7c, 0x35, 0x30,
-	0x7e, 0xf1, 0x7c, 0x50, 0xfb, 0xf2, 0xf9, 0xa0, 0xf6, 0xf7, 0xe7, 0x83, 0xda, 0x49, 0x93, 0xfe,
-	0xc7, 0xf1, 0xcd, 0xff, 0x04, 0x00, 0x00, 0xff, 0xff, 0x58, 0x7f, 0xb1, 0xf4, 0xd9, 0x21, 0x00,
-	0x00,
+	0xd5, 0x84, 0x47, 0xb8, 0x68, 0x6b, 0xfc, 0x1d, 0xe2, 0x9f, 0x25, 0xe0, 0x6b, 0x52, 0xab, 0xd4,
+	0x78, 0x41, 0xbc, 0xe6, 0x14, 0x1a, 0x37, 0x38, 0x7f, 0x92, 0xc7, 0x17, 0x6a, 0x55, 0xc3, 0x64,
+	0x10, 0xcb, 0xf8, 0x0c, 0x1e, 0x75, 0xa8, 0xe0, 0xc8, 0x20, 0x3d, 0xa1, 0xc3, 0x0c, 0xc1, 0xf9,
+	0x9d, 0x01, 0x37, 0x4b, 0x07, 0xd3, 0x47, 0xb8, 0x17, 0xec, 0xbd, 0xeb, 0xd0, 0xce, 0x52, 0x9f,
+	0x38, 0x55, 0x4f, 0x56, 0x30, 0xf5, 0x88, 0x2c, 0x17, 0x34, 0xd9, 0xc0, 0x14, 0xfc, 0x62, 0xaf,
+	0xf7, 0xa1, 0x35, 0xae, 0x36, 0x66, 0x09, 0x3a, 0x7f, 0x32, 0xe0, 0xd5, 0xb9, 0xf1, 0xfe, 0x5f,
+	0x7c, 0xd0, 0x85, 0x22, 0x28, 0x32, 0x59, 0x26, 0xaf, 0xdf, 0x3f, 0x70, 0x92, 0xf9, 0x10, 0x96,
+	0xf2, 0xd2, 0x32, 0x5c, 0x7d, 0xd0, 0x7d, 0xa5, 0x7a, 0x50, 0x33, 0x9e, 0x5b, 0xe5, 0x77, 0xce,
+	0xe1, 0x95, 0x8a, 0xfe, 0x95, 0x9a, 0xb8, 0x43, 0xf3, 0x3d, 0xf2, 0x72, 0x59, 0x19, 0x6f, 0x69,
+	0x82, 0xc5, 0x3c, 0x4d, 0x54, 0xb7, 0xe0, 0xab, 0xa4, 0xb8, 0x59, 0x4d, 0x71, 0xe7, 0xb7, 0x26,
+	0xac, 0x4c, 0x5d, 0xc5, 0x96, 0xc1, 0x0c, 0x86, 0xd2, 0x91, 0x66, 0x30, 0x5c, 0x98, 0xae, 0xba,
+	0x73, 0xad, 0x29, 0xe7, 0x62, 0x81, 0x4a, 0xfd, 0x3d, 0x2f, 0xf7, 0x64, 0xff, 0x57, 0x60, 0xc5,
+	0xed, 0x8d, 0x29, 0xb7, 0xf7, 0xa1, 0x35, 0xcc, 0x72, 0x3a, 0x25, 0xb2, 0x52, 0x81, 0x58, 0xda,
+	0x29, 0xce, 0xe9, 0xd3, 0x92, 0x98, 0xa8, 0x4a, 0x04, 0xdb, 0x2e, 0x96, 0xba, 0xf6, 0xb5, 0x36,
+	0x91, 0x5c, 0xc5, 0x3c, 0xd5, 0x91, 0x45, 0x09, 0xe7, 0x29, 0x2d, 0xa2, 0xa0, 0x1a, 0x51, 0xcf,
+	0xa6, 0x0a, 0xa8, 0x74, 0xc8, 0x4b, 0xc7, 0xd3, 0x5b, 0x6a, 0xcc, 0x16, 0xa1, 0x74, 0xa3, 0x1a,
+	0x11, 0x95, 0x49, 0xfb, 0x97, 0x06, 0xdc, 0x51, 0xcd, 0x78, 0x7e, 0x20, 0xdc, 0xd5, 0x9a, 0xe3,
+	0xac, 0x24, 0xd9, 0x24, 0x69, 0x3e, 0xff, 0x38, 0x0c, 0xc5, 0x62, 0x65, 0xaa, 0xf9, 0x5c, 0x61,
+	0x2a, 0x91, 0x61, 0x4d, 0x15, 0xff, 0x35, 0xd2, 0xf6, 0xa1, 0xf8, 0x01, 0xa0, 0xee, 0x0a, 0xc0,
+	0xf9, 0x14, 0x06, 0x8b, 0xf4, 0x7a, 0x59, 0x7b, 0x38, 0x57, 0x70, 0x47, 0xb4, 0xb5, 0x52, 0x94,
+	0xfa, 0xb9, 0xe7, 0xc5, 0xbd, 0xa9, 0xd2, 0xeb, 0xcd, 0xe9, 0x5e, 0x5f, 0x7c, 0x8a, 0xa4, 0xcf,
+	0xdb, 0x96, 0xfe, 0x29, 0x12, 0x31, 0x5b, 0xe7, 0xd0, 0x14, 0xc3, 0x1c, 0x5b, 0x82, 0xce, 0xc3,
+	0x88, 0xd2, 0xf7, 0x30, 0xb1, 0x6b, 0xac, 0x0d, 0xf5, 0xa3, 0x3c, 0x4e, 0x6c, 0x83, 0x75, 0xa0,
+	0xf1, 0x18, 0xa7, 0x79, 0xdb, 0x64, 0x00, 0x4d, 0xac, 0xf6, 0x63, 0x6e, 0x5b, 0x88, 0x3e, 0xca,
+	0xbd, 0x34, 0xb7, 0xeb, 0x88, 0x16, 0xfa, 0xdb, 0x0d, 0xb6, 0x0c, 0xf0, 0xf1, 0x24, 0x8f, 0x25,
+	0x5b, 0x13, 0x69, 0x7b, 0x3c, 0xe4, 0x39, 0xb7, 0x5b, 0x5b, 0x3f, 0xa1, 0x23, 0x23, 0x1c, 0x1f,
+	0x7a, 0xf2, 0x2e, 0x82, 0xed, 0x1a, 0x6b, 0x81, 0xf5, 0x19, 0xbf, 0xb4, 0x0d, 0xd6, 0x85, 0x96,
+	0x3b, 0x89, 0xa2, 0x20, 0x1a, 0x89, 0xfb, 0xe8, 0xea, 0xa1, 0x6d, 0x21, 0x01, 0x15, 0x4a, 0xf8,
+	0xd0, 0xae, 0xb3, 0x1e, 0xb4, 0x3f, 0x91, 0x3f, 0x13, 0xd8, 0x0d, 0x24, 0x21, 0x1b, 0x9e, 0x69,
+	0x22, 0x89, 0x2e, 0x47, 0xa8, 0x85, 0x10, 0x9d, 0x42, 0xa8, 0xbd, 0x75, 0x08, 0x6d, 0xb5, 0xb9,
+	0xb2, 0x15, 0xe8, 0x4a, 0x1d, 0x10, 0x65, 0xd7, 0xf0, 0x85, 0x68, 0xd8, 0xb0, 0x0d, 0x7c, 0x79,
+	0xdc, 0x41, 0x6d, 0x13, 0x9f, 0x70, 0xd1, 0xb4, 0x2d, 0x32, 0xc8, 0x55, 0xe4, 0xdb, 0x75, 0x64,
+	0xa4, 0x85, 0xc5, 0x1e, 0x6e, 0x3d, 0x82, 0x16, 0x3d, 0x1e, 0xe2, 0x1c, 0xb6, 0x2c, 0xe5, 0x49,
+	0x8c, 0x5d, 0x43, 0x9b, 0xe2, 0xed, 0x82, 0xdb, 0x40, 0xdb, 0xd0, 0xeb, 0x08, 0xd8, 0x44, 0x15,
+	0x84, 0x9d, 0x04, 0xc2, 0xda, 0xfa, 0xa9, 0x01, 0x6d, 0xb5, 0x6a, 0xb0, 0x1b, 0xb0, 0xa2, 0x8c,
+	0x24, 0x51, 0x42, 0xe2, 0x3e, 0xcf, 0x05, 0xc2, 0x36, 0xe8, 0x82, 0x02, 0x34, 0xd1, 0xae, 0x2e,
+	0x1f, 0xc7, 0x17, 0x5c, 0x62, 0x2c, 0xbc, 0x12, 0x37, 0x5b, 0x09, 0xd7, 0xf1, 0x00, 0xc2, 0x54,
+	0x65, 0xec, 0x06, 0xbb, 0x05, 0x0c, 0xc1, 0x47, 0xc1, 0x08, 0x23, 0x59, 0xcc, 0xff, 0x99, 0xdd,
+	0xdc, 0xfa, 0x08, 0xda, 0x6a, 0xcc, 0xd6, 0xf4, 0x50, 0xa8, 0x42, 0x0f, 0x81, 0xb0, 0x8d, 0xf2,
+	0x62, 0x89, 0x31, 0xb7, 0x9e, 0xd2, 0x7a, 0x8a, 0x53, 0xaa, 0x66, 0x19, 0x89, 0x91, 0xe1, 0x75,
+	0x1e, 0x24, 0xd2, 0xe1, 0x3c, 0x09, 0x3d, 0xbf, 0x08, 0xb0, 0x0b, 0x9e, 0xe6, 0xb6, 0x85, 0xcf,
+	0x0f, 0xa3, 0x1f, 0x71, 0x1f, 0x23, 0x0c, 0xdd, 0x10, 0x64, 0xb9, 0xdd, 0xd8, 0x3a, 0x80, 0xee,
+	0x53, 0xd5, 0x63, 0x0e, 0x13, 0x7c, 0x01, 0xa5, 0x5c, 0x89, 0xb5, 0x6b, 0x78, 0x27, 0x45, 0x67,
+	0x81, 0xb5, 0x0d, 0xb6, 0x0a, 0x4b, 0xe8, 0x8d, 0x12, 0x65, 0x6e, 0x3d, 0x01, 0x36, 0x5b, 0x1d,
+	0xd1, 0x68, 0xa5, 0xc2, 0x76, 0x0d, 0x35, 0xf9, 0x8c, 0x5f, 0xe2, 0x33, 0xf9, 0xf0, 0xe1, 0x28,
+	0x8a, 0x53, 0x4e, 0x34, 0xe5, 0x43, 0xfa, 0xbe, 0x88, 0x08, 0x6b, 0xeb, 0xe9, 0x54, 0x1f, 0x39,
+	0x4c, 0xb4, 0x70, 0x27, 0xd8, 0xae, 0x51, 0xf0, 0x91, 0x14, 0x81, 0x90, 0x06, 0x24, 0x31, 0x02,
+	0x63, 0xe2, 0x45, 0xbb, 0x21, 0xf7, 0x52, 0x01, 0x5b, 0x3b, 0xff, 0x6a, 0x42, 0x53, 0x54, 0x05,
+	0xf6, 0x11, 0x74, 0xb5, 0x5f, 0x68, 0x19, 0x15, 0xf9, 0xd9, 0xdf, 0x93, 0xd7, 0xff, 0x6f, 0x06,
+	0x2f, 0x2a, 0x93, 0x53, 0x63, 0x1f, 0x02, 0x94, 0x8b, 0x37, 0xbb, 0x49, 0xd3, 0xdc, 0xf4, 0x22,
+	0xbe, 0xde, 0xa7, 0x4f, 0x36, 0x73, 0x7e, 0x7d, 0x76, 0x6a, 0xec, 0x7b, 0xb0, 0x24, 0xcb, 0x9f,
+	0x08, 0x2d, 0x36, 0xd0, 0xd6, 0xa6, 0x39, 0x2b, 0xf5, 0xb5, 0xc2, 0x3e, 0x29, 0x84, 0x89, 0xf0,
+	0x61, 0xfd, 0x39, 0x3b, 0x98, 0x10, 0xf3, 0xca, 0xc2, 0xed, 0xcc, 0xa9, 0xb1, 0x7d, 0xe8, 0x8a,
+	0x1d, 0x4a, 0x14, 0xf5, 0xdb, 0xc8, 0xbb, 0x68, 0xa9, 0xba, 0x56, 0xa1, 0x5d, 0xe8, 0xe9, 0x6b,
+	0x0f, 0x23, 0x4b, 0xce, 0xd9, 0x8f, 0x84, 0x90, 0x79, 0x1b, 0x92, 0x53, 0x63, 0x1e, 0xdc, 0x9a,
+	0xbf, 0xbc, 0xb0, 0xd7, 0xcb, 0x6f, 0xcb, 0x0b, 0xb6, 0xa5, 0x75, 0xe7, 0x3a, 0x96, 0xe2, 0x8a,
+	0x1f, 0x40, 0xbf, 0xb8, 0xbc, 0x08, 0x6b, 0x19, 0x15, 0x03, 0xa9, 0xda, 0x82, 0x7d, 0x67, 0xfd,
+	0xb5, 0x85, 0xf4, 0x42, 0xfc, 0x31, 0xac, 0x96, 0x0c, 0xb1, 0x30, 0x1f, 0xbb, 0x33, 0x73, 0xae,
+	0x62, 0xd6, 0xc1, 0x22, 0x72, 0x21, 0xf5, 0x87, 0xe5, 0xc6, 0x5e, 0x95, 0xfc, 0xba, 0xee, 0xdb,
+	0xf9, 0xd2, 0x9d, 0xeb, 0x58, 0x8a, 0x1b, 0x1e, 0xc3, 0x4a, 0xa5, 0x9f, 0x2a, 0xd9, 0xd7, 0x36,
+	0xd9, 0xeb, 0x02, 0xe2, 0x5e, 0xff, 0x8b, 0xaf, 0x06, 0xc6, 0x97, 0x5f, 0x0d, 0x8c, 0x7f, 0x7c,
+	0x35, 0x30, 0x7e, 0xfe, 0x7c, 0x50, 0xfb, 0xf2, 0xf9, 0xa0, 0xf6, 0xb7, 0xe7, 0x83, 0xda, 0x49,
+	0x93, 0xfe, 0xd5, 0xf1, 0xcd, 0xff, 0x04, 0x00, 0x00, 0xff, 0xff, 0x31, 0xe0, 0x21, 0xb2, 0xe7,
+	0x21, 0x00, 0x00,
 }
 
 // Reference imports to suppress errors if they are not otherwise used.
@@ -6303,6 +6267,20 @@ func (m *ValidationStatus) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if len(m.CutoverBinlogGtid) > 0 {
+		i -= len(m.CutoverBinlogGtid)
+		copy(dAtA[i:], m.CutoverBinlogGtid)
+		i = encodeVarintDmworker(dAtA, i, uint64(len(m.CutoverBinlogGtid)))
+		i--
+		dAtA[i] = 0x62
+	}
+	if len(m.CutoverBinlogPos) > 0 {
+		i -= len(m.CutoverBinlogPos)
+		copy(dAtA[i:], m.CutoverBinlogPos)
+		i = encodeVarintDmworker(dAtA, i, uint64(len(m.CutoverBinlogPos)))
+		i--
+		dAtA[i] = 0x5a
+	}
 	if len(m.ErrorRowsStatus) > 0 {
 		i -= len(m.ErrorRowsStatus)
 		copy(dAtA[i:], m.ErrorRowsStatus)
@@ -6817,46 +6795,6 @@ func (m *UpdateValidationWorkerRequest) MarshalToSizedBuffer(dAtA []byte) (int,
 	return len(dAtA) - i, nil
 }
 
-func (m *UpdateValidationWorkerResponse) Marshal() (dAtA []byte, err error) {
-	size := m.Size()
-	dAtA = make([]byte, size)
-	n, err := m.MarshalToSizedBuffer(dAtA[:size])
-	if err != nil {
-		return nil, err
-	}
-	return dAtA[:n], nil
-}
-
-func (m *UpdateValidationWorkerResponse) MarshalTo(dAtA []byte) (int, error) {
-	size := m.Size()
-	return m.MarshalToSizedBuffer(dAtA[:size])
-}
-
-func (m *UpdateValidationWorkerResponse) MarshalToSizedBuffer(dAtA []byte) (int, error) {
-	i := len(dAtA)
-	_ = i
-	var l int
-	_ = l
-	if len(m.Msg) > 0 {
-		i -= len(m.Msg)
-		copy(dAtA[i:], m.Msg)
-		i = encodeVarintDmworker(dAtA, i, uint64(len(m.Msg)))
-		i--
-		dAtA[i] = 0x12
-	}
-	if m.Result {
-		i--
-		if m.Result {
-			dAtA[i] = 1
-		} else {
-			dAtA[i] = 0
-		}
-		i--
-		dAtA[i] = 0x8
-	}
-	return len(dAtA) - i, nil
-}
-
 func encodeVarintDmworker(dAtA []byte, offset int, v uint64) int {
 	offset -= sovDmworker(v)
 	base := offset
@@ -7834,6 +7772,14 @@ func (m *ValidationStatus) Size() (n int) {
 	if l > 0 {
 		n += 1 + l + sovDmworker(uint64(l))
 	}
+	l = len(m.CutoverBinlogPos)
+	if l > 0 {
+		n += 1 + l + sovDmworker(uint64(l))
+	}
+	l = len(m.CutoverBinlogGtid)
+	if l > 0 {
+		n += 1 + l + sovDmworker(uint64(l))
+	}
 	return n
 }
 
@@ -8038,22 +7984,6 @@ func (m *UpdateValidationWorkerRequest) Size() (n int) {
 	return n
 }
 
-func (m *UpdateValidationWorkerResponse) Size() (n int) {
-	if m == nil {
-		return 0
-	}
-	var l int
-	_ = l
-	if m.Result {
-		n += 2
-	}
-	l = len(m.Msg)
-	if l > 0 {
-		n += 1 + l + sovDmworker(uint64(l))
-	}
-	return n
-}
-
 func sovDmworker(x uint64) (n int) {
 	return (math_bits.Len64(x|1) + 6) / 7
 }
@@ -14034,6 +13964,70 @@ func (m *ValidationStatus) Unmarshal(dAtA []byte) error {
 			}
 			m.ErrorRowsStatus = string(dAtA[iNdEx:postIndex])
 			iNdEx = postIndex
+		case 11:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field CutoverBinlogPos", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowDmworker
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthDmworker
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthDmworker
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.CutoverBinlogPos = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
+		case 12:
+			if wireType != 2 {
+				return fmt.Errorf("proto: wrong wireType = %d for field CutoverBinlogGtid", wireType)
+			}
+			var stringLen uint64
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowDmworker
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				stringLen |= uint64(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
+			intStringLen := int(stringLen)
+			if intStringLen < 0 {
+				return ErrInvalidLengthDmworker
+			}
+			postIndex := iNdEx + intStringLen
+			if postIndex < 0 {
+				return ErrInvalidLengthDmworker
+			}
+			if postIndex > l {
+				return io.ErrUnexpectedEOF
+			}
+			m.CutoverBinlogGtid = string(dAtA[iNdEx:postIndex])
+			iNdEx = postIndex
 		default:
 			iNdEx = preIndex
 			skippy, err := skipDmworker(dAtA[iNdEx:])
@@ -15404,108 +15398,6 @@ func (m *UpdateValidationWorkerRequest) Unmarshal(dAtA []byte) error {
 	}
 	return nil
 }
-func (m *UpdateValidationWorkerResponse) Unmarshal(dAtA []byte) error {
-	l := len(dAtA)
-	iNdEx := 0
-	for iNdEx < l {
-		preIndex := iNdEx
-		var wire uint64
-		for shift := uint(0); ; shift += 7 {
-			if shift >= 64 {
-				return ErrIntOverflowDmworker
-			}
-			if iNdEx >= l {
-				return io.ErrUnexpectedEOF
-			}
-			b := dAtA[iNdEx]
-			iNdEx++
-			wire |= uint64(b&0x7F) << shift
-			if b < 0x80 {
-				break
-			}
-		}
-		fieldNum := int32(wire >> 3)
-		wireType := int(wire & 0x7)
-		if wireType == 4 {
-			return fmt.Errorf("proto: UpdateValidationWorkerResponse: wiretype end group for non-group")
-		}
-		if fieldNum <= 0 {
-			return fmt.Errorf("proto: UpdateValidationWorkerResponse: illegal tag %d (wire type %d)", fieldNum, wire)
-		}
-		switch fieldNum {
-		case 1:
-			if wireType != 0 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Result", wireType)
-			}
-			var v int
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowDmworker
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				v |= int(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			m.Result = bool(v != 0)
-		case 2:
-			if wireType != 2 {
-				return fmt.Errorf("proto: wrong wireType = %d for field Msg", wireType)
-			}
-			var stringLen uint64
-			for shift := uint(0); ; shift += 7 {
-				if shift >= 64 {
-					return ErrIntOverflowDmworker
-				}
-				if iNdEx >= l {
-					return io.ErrUnexpectedEOF
-				}
-				b := dAtA[iNdEx]
-				iNdEx++
-				stringLen |= uint64(b&0x7F) << shift
-				if b < 0x80 {
-					break
-				}
-			}
-			intStringLen := int(stringLen)
-			if intStringLen < 0 {
-				return ErrInvalidLengthDmworker
-			}
-			postIndex := iNdEx + intStringLen
-			if postIndex < 0 {
-				return ErrInvalidLengthDmworker
-			}
-			if postIndex > l {
-				return io.ErrUnexpectedEOF
-			}
-			m.Msg = string(dAtA[iNdEx:postIndex])
-			iNdEx = postIndex
-		default:
-			iNdEx = preIndex
-			skippy, err := skipDmworker(dAtA[iNdEx:])
-			if err != nil {
-				return err
-			}
-			if (skippy < 0) || (iNdEx+skippy) < 0 {
-				return ErrInvalidLengthDmworker
-			}
-			if (iNdEx + skippy) > l {
-				return io.ErrUnexpectedEOF
-			}
-			iNdEx += skippy
-		}
-	}
-
-	if iNdEx > l {
-		return io.ErrUnexpectedEOF
-	}
-	return nil
-}
 func skipDmworker(dAtA []byte) (n int, err error) {
 	l := len(dAtA)
 	iNdEx := 0
diff --git a/dm/pkg/binlog/event/common.go b/dm/pkg/binlog/event/common.go
index e84dc8400..5437ec4f7 100644
--- a/dm/pkg/binlog/event/common.go
+++ b/dm/pkg/binlog/event/common.go
@@ -19,6 +19,7 @@ import (
 
 	gmysql "github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
+	"github.com/pingcap/tiflow/dm/pkg/gtid"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
@@ -172,7 +173,7 @@ func GTIDIncrease(flavor string, gSet gmysql.GTIDSet) (gmysql.GTIDSet, error) {
 
 // verifySingleGTID verifies gSet whether only containing a single valid GTID.
 func verifySingleGTID(flavor string, gSet gmysql.GTIDSet) (interface{}, error) {
-	if gSet == nil || len(gSet.String()) == 0 {
+	if gtid.IsZeroGTIDSet(gSet) {
 		return nil, terror.ErrBinlogEmptyGTID.Generate()
 	}
 
diff --git a/dm/pkg/binlog/event/event.go b/dm/pkg/binlog/event/event.go
index daab72ff7..1d8ca835d 100644
--- a/dm/pkg/binlog/event/event.go
+++ b/dm/pkg/binlog/event/event.go
@@ -22,6 +22,7 @@ import (
 
 	gmysql "github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
+	"github.com/pingcap/tiflow/dm/pkg/gtid"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
@@ -717,7 +718,7 @@ func GenXIDEvent(header *replication.EventHeader, latestPos uint32, xid uint64)
 // GenMariaDBGTIDListEvent generates a MariadbGTIDListEvent.
 // ref: https://mariadb.com/kb/en/library/gtid_list_event/
 func GenMariaDBGTIDListEvent(header *replication.EventHeader, latestPos uint32, gSet gmysql.GTIDSet) (*replication.BinlogEvent, error) {
-	if gSet == nil || len(gSet.String()) == 0 {
+	if gtid.IsZeroGTIDSet(gSet) {
 		return nil, terror.ErrBinlogEmptyGTID.Generate()
 	}
 
diff --git a/dm/pkg/binlog/position.go b/dm/pkg/binlog/position.go
index 21eae14c3..a54fd70bb 100644
--- a/dm/pkg/binlog/position.go
+++ b/dm/pkg/binlog/position.go
@@ -366,8 +366,8 @@ func IsFreshPosition(location Location, flavor string, cmpGTID bool) bool {
 //
 // but if can't compare gSet1 and gSet2, will returns 0, false.
 func CompareGTID(gSet1, gSet2 gmysql.GTIDSet) (int, bool) {
-	gSetIsEmpty1 := gSet1 == nil || len(gSet1.String()) == 0
-	gSetIsEmpty2 := gSet2 == nil || len(gSet2.String()) == 0
+	gSetIsEmpty1 := gtid.IsZeroGTIDSet(gSet1)
+	gSetIsEmpty2 := gtid.IsZeroGTIDSet(gSet2)
 
 	switch {
 	case gSetIsEmpty1 && gSetIsEmpty2:
diff --git a/dm/pkg/gtid/gtid.go b/dm/pkg/gtid/gtid.go
index 98075a52d..d81734519 100644
--- a/dm/pkg/gtid/gtid.go
+++ b/dm/pkg/gtid/gtid.go
@@ -21,6 +21,11 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 )
 
+var (
+	zeroMySQLGTIDSet   = MustZeroGTIDSet(mysql.MySQLFlavor)
+	zeroMariaDBGTIDSet = MustZeroGTIDSet(mysql.MariaDBFlavor)
+)
+
 // ParserGTID parses GTID from string. If the flavor is not specified, it will
 // try mysql GTID first and then MariaDB GTID.
 func ParserGTID(flavor, gtidStr string) (mysql.GTIDSet, error) {
@@ -42,7 +47,11 @@ func ParserGTID(flavor, gtidStr string) (mysql.GTIDSet, error) {
 			gtid, err = mysql.ParseGTIDSet(fla, gtidStr)
 		}
 	case mysql.MariaDBFlavor:
-		gtid, err = mysql.ParseGTIDSet(fla, gtidStr)
+		if IsZeroMariaDBGTIDSet(gtidStr) {
+			gtid, err = mysql.ParseGTIDSet(fla, "")
+		} else {
+			gtid, err = mysql.ParseGTIDSet(fla, gtidStr)
+		}
 	case "":
 		fla = mysql.MySQLFlavor
 		gtid, err = mysql.ParseGTIDSet(fla, gtidStr)
@@ -74,9 +83,6 @@ func MustZeroGTIDSet(flavor string) mysql.GTIDSet {
 // IsZeroMySQLGTIDSet is used to meet this usage: when user wants to start binlog
 // replication from scratch, a "uuid:0" (MySQL flavor) or "0-0-0" (mariaDB) GTID
 // set must be written, in order to distinguish that user forgets to write it.
-//
-// For above two flavor, only "uuid:0" is illegal, so we use IsZeroMySQLGTIDSet
-// to handle it.
 func IsZeroMySQLGTIDSet(gStr string) bool {
 	sp := strings.Split(gStr, ",")
 	if len(sp) != 1 {
@@ -90,3 +96,28 @@ func IsZeroMySQLGTIDSet(gStr string) bool {
 	interval := strings.TrimSpace(sep[1])
 	return interval == "0"
 }
+
+// IsZeroMariaDBGTIDSet is used to meet this usage: when user wants to start binlog
+// replication from scratch, a "uuid:0" (MySQL flavor) or "0-0-0" (mariaDB) GTID
+// set must be written, in order to distinguish that user forgets to write it.
+//
+// For MariaDB, the GTID set like "0-0-0" will confuse IsZeroGTIDSet function,
+// so we also need to check the interval part.
+func IsZeroMariaDBGTIDSet(gStr string) bool {
+	sp := strings.Split(gStr, ",")
+	if len(sp) != 1 {
+		return false
+	}
+
+	sep := strings.Split(sp[0], "-")
+	if len(sep) != 3 {
+		return false
+	}
+	interval := strings.TrimSpace(sep[2])
+	return interval == "0"
+}
+
+// IsZeroGTIDSet is used to check whether a GTID set is zero.
+func IsZeroGTIDSet(gSet mysql.GTIDSet) bool {
+	return gSet == nil || zeroMySQLGTIDSet.Equal(gSet) || zeroMariaDBGTIDSet.Equal(gSet)
+}
diff --git a/dm/proto/dmworker.proto b/dm/proto/dmworker.proto
index 585e330ef..bc067104c 100644
--- a/dm/proto/dmworker.proto
+++ b/dm/proto/dmworker.proto
@@ -437,6 +437,8 @@ message ValidationStatus {
     string processedRowsStatus = 8;
     string pendingRowsStatus = 9;
     string errorRowsStatus = 10;
+    string cutoverBinlogPos = 11;
+    string cutoverBinlogGtid = 12;
 }
 
 message ValidationTableStatus {
diff --git a/dm/syncer/data_validator.go b/dm/syncer/data_validator.go
index 08083bd2b..ad150f386 100644
--- a/dm/syncer/data_validator.go
+++ b/dm/syncer/data_validator.go
@@ -1320,12 +1320,12 @@ func (v *DataValidator) OperateValidatorError(validateOp pb.ValidationErrOp, err
 
 func (v *DataValidator) UpdateValidator(req *pb.UpdateValidationWorkerRequest) error {
 	var (
-		pos *mysql.Position
+		pos = mysql.Position{}
 		gs  mysql.GTIDSet
 		err error
 	)
 	if len(req.BinlogPos) > 0 {
-		pos, err = binlog.VerifyBinlogPos(req.BinlogPos)
+		pos, err = binlog.PositionFromPosStr(req.BinlogPos)
 		if err != nil {
 			return err
 		}
@@ -1336,8 +1336,9 @@ func (v *DataValidator) UpdateValidator(req *pb.UpdateValidationWorkerRequest) e
 			return err
 		}
 	}
-	cutOverLocation := binlog.NewLocation(*pos, gs)
+	cutOverLocation := binlog.NewLocation(pos, gs)
 	v.cutOverLocation.Store(&cutOverLocation)
+	v.syncer.cutOverLocation.Store(&cutOverLocation)
 	return nil
 }
 
@@ -1404,6 +1405,14 @@ func (v *DataValidator) GetValidatorStatus() *pb.ValidationStatus {
 			validatorBinlogGtid = flushedLoc.GetGTID().String()
 		}
 	}
+	var cutoverBinlogPos, cutoverBinlogGTID string
+	if cutOverLoc := v.cutOverLocation.Load(); cutOverLoc != nil {
+		cutoverBinlogPos = cutOverLoc.Position.String()
+		if cutOverLoc.GetGTID() != nil {
+			cutoverBinlogGTID = cutOverLoc.GetGTID().String()
+		}
+	}
+
 	return &pb.ValidationStatus{
 		Task:                v.cfg.Name,
 		Source:              v.cfg.SourceID,
@@ -1415,5 +1424,7 @@ func (v *DataValidator) GetValidatorStatus() *pb.ValidationStatus {
 		ProcessedRowsStatus: processedRows,
 		PendingRowsStatus:   pendingRows,
 		ErrorRowsStatus:     errorRows,
+		CutoverBinlogPos:    cutoverBinlogPos,
+		CutoverBinlogGtid:   cutoverBinlogGTID,
 	}
 }
diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
index a160c3621..7c8232653 100644
--- a/dm/syncer/syncer.go
+++ b/dm/syncer/syncer.go
@@ -217,6 +217,8 @@ type Syncer struct {
 		isQueryEvent  bool
 	}
 
+	cutOverLocation atomic.Pointer[binlog.Location]
+
 	handleJobFunc func(*job) (bool, error)
 	flushSeq      int64
 
@@ -1112,7 +1114,12 @@ func (s *Syncer) handleJob(job *job) (added2Queue bool, err error) {
 	skipCheckFlush := false
 	defer func() {
 		if !skipCheckFlush && err == nil {
-			err = s.flushIfOutdated()
+			if cutoverLocation := s.cutOverLocation.Load(); cutoverLocation != nil && binlog.CompareLocation(*cutoverLocation, job.currentLocation, s.cfg.EnableGTID) <= 0 {
+				err = s.flushJobs()
+				s.cutOverLocation.Store(nil)
+			} else {
+				err = s.flushIfOutdated()
+			}
 		}
 	}()
 
diff --git a/errors.toml b/errors.toml
index 1563e43c0..9d5021d5d 100755
--- a/errors.toml
+++ b/errors.toml
@@ -181,6 +181,11 @@ error = '''
 Codec invalid config
 '''
 
+["CDC:ErrColumnSelectorFailed"]
+error = '''
+column selector failed
+'''
+
 ["CDC:ErrCompressionFailed"]
 error = '''
 Compression failed
diff --git a/metrics/grafana/ticdc.json b/metrics/grafana/ticdc.json
index a67871fac..407c341a6 100644
--- a/metrics/grafana/ticdc.json
+++ b/metrics/grafana/ticdc.json
@@ -3437,8 +3437,16 @@
               "hide": false,
               "interval": "",
               "intervalFactor": 1,
-              "legendFormat": "{{namespace}}-{{changefeed}}",
+              "legendFormat": "{{namespace}}-{{changefeed}}-barrier",
               "refId": "C"
+            },
+            {
+              "exemplar": true,
+              "expr": "max(ticdc_scheduler_slow_table_puller_resolved_ts{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\", changefeed=~\"$changefeed\"}) by (namespace, changefeed)",
+              "hide": false,
+              "interval": "",
+              "legendFormat": "{{namespace}}-{{changefeed}}-puller",
+              "refId": "B"
             }
           ],
           "thresholds": [],
@@ -3644,8 +3652,16 @@
               "format": "time_series",
               "interval": "",
               "intervalFactor": 1,
-              "legendFormat": "{{namespace}}-{{changefeed}}",
+              "legendFormat": "{{namespace}}-{{changefeed}}-barrier",
               "refId": "C"
+            },
+            {
+              "exemplar": true,
+              "expr": "max(ticdc_scheduler_slow_table_puller_resolved_ts_lag{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\", namespace=~\"$namespace\", changefeed=~\"$changefeed\"}) by (namespace, changefeed)",
+              "hide": false,
+              "interval": "",
+              "legendFormat": "{{namespace}}-{{changefeed}}-puller",
+              "refId": "A"
             }
           ],
           "thresholds": [],
diff --git a/pkg/config/consistent.go b/pkg/config/consistent.go
index 636edcf86..14021e207 100644
--- a/pkg/config/consistent.go
+++ b/pkg/config/consistent.go
@@ -24,11 +24,14 @@ import (
 
 // ConsistentConfig represents replication consistency config for a changefeed.
 type ConsistentConfig struct {
-	Level             string `toml:"level" json:"level"`
-	MaxLogSize        int64  `toml:"max-log-size" json:"max-log-size"`
-	FlushIntervalInMs int64  `toml:"flush-interval" json:"flush-interval"`
-	Storage           string `toml:"storage" json:"storage"`
-	UseFileBackend    bool   `toml:"use-file-backend" json:"use-file-backend"`
+	Level                 string `toml:"level" json:"level"`
+	MaxLogSize            int64  `toml:"max-log-size" json:"max-log-size"`
+	FlushIntervalInMs     int64  `toml:"flush-interval" json:"flush-interval"`
+	MetaFlushIntervalInMs int64  `toml:"meta-flush-interval" json:"meta-flush-interval"`
+	EncodingWorkerNum     int    `toml:"encoding-worker-num" json:"encoding-worker-num"`
+	FlushWorkerNum        int    `toml:"flush-worker-num" json:"flush-worker-num"`
+	Storage               string `toml:"storage" json:"storage"`
+	UseFileBackend        bool   `toml:"use-file-backend" json:"use-file-backend"`
 }
 
 // ValidateAndAdjust validates the consistency config and adjusts it if necessary.
@@ -50,6 +53,22 @@ func (c *ConsistentConfig) ValidateAndAdjust() error {
 				c.FlushIntervalInMs, redo.MinFlushIntervalInMs))
 	}
 
+	if c.MetaFlushIntervalInMs == 0 {
+		c.MetaFlushIntervalInMs = redo.DefaultMetaFlushIntervalInMs
+	}
+	if c.MetaFlushIntervalInMs < redo.MinFlushIntervalInMs {
+		return cerror.ErrInvalidReplicaConfig.FastGenByArgs(
+			fmt.Sprintf("The consistent.meta-flush-interval:%d must be equal or greater than %d",
+				c.MetaFlushIntervalInMs, redo.MinFlushIntervalInMs))
+	}
+
+	if c.EncodingWorkerNum == 0 {
+		c.EncodingWorkerNum = redo.DefaultEncodingWorkerNum
+	}
+	if c.FlushWorkerNum == 0 {
+		c.FlushWorkerNum = redo.DefaultFlushWorkerNum
+	}
+
 	uri, err := storage.ParseRawURL(c.Storage)
 	if err != nil {
 		return cerror.ErrInvalidReplicaConfig.GenWithStackByArgs(
diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go
index 2c8f37f61..ae897b729 100644
--- a/pkg/config/replica_config.go
+++ b/pkg/config/replica_config.go
@@ -23,6 +23,7 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tiflow/pkg/config/outdated"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/integrity"
@@ -38,11 +39,17 @@ const (
 	// minSyncPointRetention is the minimum of SyncPointRetention can be set.
 	minSyncPointRetention           = time.Hour * 1
 	minChangeFeedErrorStuckDuration = time.Minute * 30
+	// The default SQL Mode of TiDB: "ONLY_FULL_GROUP_BY,
+	// STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,
+	// NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+	// Note: The SQL Mode of TiDB is not the same as ORACLE.
+	// If you want to use the same SQL Mode as ORACLE, you need to add "ORACLE" to the SQL Mode.
+	defaultSQLMode = mysql.DefaultSQLMode
 )
 
 var defaultReplicaConfig = &ReplicaConfig{
 	MemoryQuota:        DefaultChangefeedMemoryQuota,
-	CaseSensitive:      true,
+	CaseSensitive:      false,
 	CheckGCSafePoint:   true,
 	EnableSyncPoint:    util.AddressOf(false),
 	SyncPointInterval:  util.AddressOf(10 * time.Minute),
@@ -72,11 +79,14 @@ var defaultReplicaConfig = &ReplicaConfig{
 		AdvanceTimeoutInSec:              util.AddressOf(DefaultAdvanceTimeoutInSec),
 	},
 	Consistent: &ConsistentConfig{
-		Level:             "none",
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		FlushIntervalInMs: redo.DefaultFlushIntervalInMs,
-		Storage:           "",
-		UseFileBackend:    false,
+		Level:                 "none",
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		FlushIntervalInMs:     redo.DefaultFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs,
+		EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+		FlushWorkerNum:        redo.DefaultFlushWorkerNum,
+		Storage:               "",
+		UseFileBackend:        false,
 	},
 	Scheduler: &ChangefeedSchedulerConfig{
 		EnableTableAcrossNodes: false,
@@ -88,6 +98,7 @@ var defaultReplicaConfig = &ReplicaConfig{
 		CorruptionHandleLevel: integrity.CorruptionHandleLevelWarn,
 	},
 	ChangefeedErrorStuckDuration: util.AddressOf(time.Minute * 30),
+	SQLMode:                      defaultSQLMode,
 }
 
 // GetDefaultReplicaConfig returns the default replica config.
@@ -139,6 +150,7 @@ type replicaConfig struct {
 	// Integrity is only available when the downstream is MQ.
 	Integrity                    *integrity.Config `toml:"integrity" json:"integrity"`
 	ChangefeedErrorStuckDuration *time.Duration    `toml:"changefeed-error-stuck-duration" json:"changefeed-error-stuck-duration,omitempty"`
+	SQLMode                      string            `toml:"sql-mode" json:"sql-mode"`
 }
 
 // Value implements the driver.Valuer interface
@@ -275,6 +287,13 @@ func (c *ReplicaConfig) ValidateAndAdjust(sinkURI *url.URL) error { // check sin
 		if err := c.Integrity.Validate(); err != nil {
 			return err
 		}
+
+		if c.Integrity.Enabled() && len(c.Sink.ColumnSelectors) != 0 {
+			log.Error("it's not allowed to enable the integrity check and column selector at the same time")
+			return cerror.ErrInvalidReplicaConfig.GenWithStack(
+				"integrity check enabled and column selector set, not allowed")
+
+		}
 	}
 
 	if c.ChangefeedErrorStuckDuration != nil &&
diff --git a/pkg/config/sink.go b/pkg/config/sink.go
index 44d59590d..da1174fe1 100644
--- a/pkg/config/sink.go
+++ b/pkg/config/sink.go
@@ -122,7 +122,7 @@ type SinkConfig struct {
 	DispatchRules []*DispatchRule `toml:"dispatchers" json:"dispatchers,omitempty"`
 	// CSVConfig is only available when the downstream is Storage.
 	CSVConfig *CSVConfig `toml:"csv" json:"csv,omitempty"`
-	// ColumnSelectors is Deprecated.
+
 	ColumnSelectors []*ColumnSelector `toml:"column-selectors" json:"column-selectors,omitempty"`
 	// SchemaRegistry is only available when the downstream is MQ using avro protocol.
 	SchemaRegistry *string `toml:"schema-registry" json:"schema-registry,omitempty"`
diff --git a/pkg/errors/cdc_errors.go b/pkg/errors/cdc_errors.go
index 22b60205b..ecbd5be84 100644
--- a/pkg/errors/cdc_errors.go
+++ b/pkg/errors/cdc_errors.go
@@ -130,6 +130,11 @@ var (
 		errors.RFCCodeText("CDC:ErrDispatcherFailed"),
 	)
 
+	ErrColumnSelectorFailed = errors.Normalize(
+		"column selector failed",
+		errors.RFCCodeText("CDC:ErrColumnSelectorFailed"),
+	)
+
 	// internal errors
 	ErrAdminStopProcessor = errors.Normalize(
 		"stop processor by admin command",
diff --git a/pkg/errors/helper.go b/pkg/errors/helper.go
index 9f5a3055b..7b36ecdee 100644
--- a/pkg/errors/helper.go
+++ b/pkg/errors/helper.go
@@ -78,6 +78,7 @@ var changefeedUnRetryableErrors = []*errors.Error{
 	ErrChangefeedUnretryable,
 	ErrCorruptedDataMutation,
 	ErrDispatcherFailed,
+	ErrColumnSelectorFailed,
 
 	ErrSinkURIInvalid,
 	ErrKafkaInvalidConfig,
diff --git a/pkg/filter/expr_filter.go b/pkg/filter/expr_filter.go
index 8bae5d2e9..6015dda85 100644
--- a/pkg/filter/expr_filter.go
+++ b/pkg/filter/expr_filter.go
@@ -14,6 +14,7 @@
 package filter
 
 import (
+	"fmt"
 	"strings"
 	"sync"
 
@@ -21,6 +22,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/parser"
+	"github.com/pingcap/tidb/parser/mysql"
 	"github.com/pingcap/tidb/planner/core"
 	"github.com/pingcap/tidb/sessionctx"
 	"github.com/pingcap/tidb/types"
@@ -77,10 +79,16 @@ func newExprFilterRule(
 
 // verifyAndInitRule will verify and init the rule.
 // It should only be called in dmlExprFilter's verify method.
-func (r *dmlExprFilterRule) verify(tableInfos []*model.TableInfo) error {
+func (r *dmlExprFilterRule) verify(tableInfos []*model.TableInfo, sqlMode string) error {
 	// verify expression filter rule syntax.
 	p := parser.New()
-	_, _, err := p.ParseSQL(completeExpression(r.config.IgnoreInsertValueExpr))
+	mode, err := mysql.GetSQLMode(sqlMode)
+	if err != nil {
+		log.Error("failed to get sql mode", zap.Error(err))
+		return cerror.ErrInvalidReplicaConfig.FastGenByArgs(fmt.Sprintf("invalid sqlMode %s", sqlMode))
+	}
+	p.SetSQLMode(mode)
+	_, _, err = p.ParseSQL(completeExpression(r.config.IgnoreInsertValueExpr))
 	if err != nil {
 		log.Error("failed to parse expression", zap.Error(err))
 		return cerror.ErrExpressionParseFailed.
@@ -347,14 +355,18 @@ func getColumnFromError(err error) string {
 
 // dmlExprFilter is a filter that filters DML events by SQL expression.
 type dmlExprFilter struct {
-	rules []*dmlExprFilterRule
+	rules   []*dmlExprFilterRule
+	sqlMODE string
 }
 
 func newExprFilter(
 	timezone string,
 	cfg *config.FilterConfig,
+	sqlMODE string,
 ) (*dmlExprFilter, error) {
-	res := &dmlExprFilter{}
+	res := &dmlExprFilter{
+		sqlMODE: sqlMODE,
+	}
 	sessCtx := utils.NewSessionCtx(map[string]string{
 		"time_zone": timezone,
 	})
@@ -382,7 +394,7 @@ func (f *dmlExprFilter) addRule(
 // verify checks if all rules in this filter is valid.
 func (f *dmlExprFilter) verify(tableInfos []*model.TableInfo) error {
 	for _, rule := range f.rules {
-		err := rule.verify(tableInfos)
+		err := rule.verify(tableInfos, f.sqlMODE)
 		if err != nil {
 			log.Error("failed to verify expression filter rule", zap.Error(err))
 			return errors.Trace(err)
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index d5674475f..a964c68e6 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -102,11 +102,11 @@ func NewFilter(cfg *config.ReplicaConfig, tz string) (Filter, error) {
 		f = tfilter.CaseInsensitive(f)
 	}
 
-	dmlExprFilter, err := newExprFilter(tz, cfg.Filter)
+	dmlExprFilter, err := newExprFilter(tz, cfg.Filter, cfg.SQLMode)
 	if err != nil {
 		return nil, err
 	}
-	sqlEventFilter, err := newSQLEventFilter(cfg.Filter)
+	sqlEventFilter, err := newSQLEventFilter(cfg.Filter, cfg.SQLMode)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index d9e80d623..fdf3ab885 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -14,6 +14,7 @@
 package filter
 
 import (
+	"fmt"
 	"sync"
 
 	"github.com/pingcap/errors"
@@ -21,6 +22,7 @@ import (
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/parser"
 	timodel "github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/parser/mysql"
 	tfilter "github.com/pingcap/tidb/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -107,9 +109,17 @@ type sqlEventFilter struct {
 	rules     []*sqlEventRule
 }
 
-func newSQLEventFilter(cfg *config.FilterConfig) (*sqlEventFilter, error) {
+func newSQLEventFilter(cfg *config.FilterConfig, sqlMode string) (*sqlEventFilter, error) {
+	p := parser.New()
+	mode, err := mysql.GetSQLMode(sqlMode)
+	if err != nil {
+		log.Error("failed to get sql mode", zap.Error(err))
+		return nil, cerror.ErrInvalidReplicaConfig.FastGenByArgs(fmt.Sprintf("invalid sqlMode %s", sqlMode))
+	}
+	p.SetSQLMode(mode)
+
 	res := &sqlEventFilter{
-		ddlParser: parser.New(),
+		ddlParser: p,
 	}
 	for _, rule := range cfg.EventFilters {
 		if err := res.addRule(rule); err != nil {
diff --git a/pkg/redo/config.go b/pkg/redo/config.go
index 82c2e6f34..eb16f754f 100644
--- a/pkg/redo/config.go
+++ b/pkg/redo/config.go
@@ -44,9 +44,22 @@ const (
 	FlushWarnDuration = time.Second * 20
 	// DefaultFlushIntervalInMs is the default flush interval for redo log.
 	DefaultFlushIntervalInMs = 2000
+	// DefaultMetaFlushIntervalInMs is the default flush interval for redo meta.
+	DefaultMetaFlushIntervalInMs = 200
 	// MinFlushIntervalInMs is the minimum flush interval for redo log.
 	MinFlushIntervalInMs = 50
 
+	// DefaultEncodingWorkerNum is the default number of encoding workers.
+	DefaultEncodingWorkerNum = 16
+	// DefaultEncodingInputChanSize is the default size of input channel for encoding worker.
+	DefaultEncodingInputChanSize = 128
+	// DefaultEncodingOutputChanSize is the default size of output channel for encoding worker.
+	DefaultEncodingOutputChanSize = 2048
+	// DefaultFlushWorkerNum is the default number of flush workers.
+	// Maximum allocated memory is flushWorkerNum*maxLogSize, which is
+	// `8*64MB = 512MB` by default.
+	DefaultFlushWorkerNum = 8
+
 	// DefaultFileMode is the default mode when operation files
 	DefaultFileMode = 0o644
 	// DefaultDirMode is the default mode when operation dir
diff --git a/pkg/sink/cloudstorage/path.go b/pkg/sink/cloudstorage/path.go
index abb947f53..97694dc6b 100644
--- a/pkg/sink/cloudstorage/path.go
+++ b/pkg/sink/cloudstorage/path.go
@@ -192,27 +192,28 @@ func (f *FilePathGenerator) CheckOrWriteSchema(
 	_, checksum := mustParseSchemaName(tblSchemaFile)
 	schemaFileCnt := 0
 	lastVersion := uint64(0)
-	prefix := fmt.Sprintf(tableSchemaPrefix+"schema_", def.Schema, def.Table)
+	subDir := fmt.Sprintf(tableSchemaPrefix, def.Schema, def.Table)
 	checksumSuffix := fmt.Sprintf("%010d.json", checksum)
-	err = f.storage.WalkDir(ctx, &storage.WalkOption{ObjPrefix: prefix},
-		func(path string, _ int64) error {
-			schemaFileCnt++
-			if !strings.HasSuffix(path, checksumSuffix) {
-				return nil
-			}
-			version, parsedChecksum := mustParseSchemaName(path)
-			if parsedChecksum != checksum {
-				// TODO: parsedChecksum should be ignored, remove this panic
-				// after the new path protocol is verified.
-				log.Panic("invalid schema file name",
-					zap.String("path", path), zap.Any("checksum", checksum))
-			}
-			if version > lastVersion {
-				lastVersion = version
-			}
+	err = f.storage.WalkDir(ctx, &storage.WalkOption{
+		SubDir:    subDir, /* use subDir to prevent walk the whole storage */
+		ObjPrefix: subDir + "schema_",
+	}, func(path string, _ int64) error {
+		schemaFileCnt++
+		if !strings.HasSuffix(path, checksumSuffix) {
 			return nil
-		},
-	)
+		}
+		version, parsedChecksum := mustParseSchemaName(path)
+		if parsedChecksum != checksum {
+			// TODO: parsedChecksum should be ignored, remove this panic
+			// after the new path protocol is verified.
+			log.Panic("invalid schema file name",
+				zap.String("path", path), zap.Any("checksum", checksum))
+		}
+		if version > lastVersion {
+			lastVersion = version
+		}
+		return nil
+	})
 	if err != nil {
 		return err
 	}
diff --git a/pkg/sink/codec/avro/avro.go b/pkg/sink/codec/avro/avro.go
index 80a863ba7..d3760c8f0 100644
--- a/pkg/sink/codec/avro/avro.go
+++ b/pkg/sink/codec/avro/avro.go
@@ -569,6 +569,9 @@ func (a *BatchEncoder) columns2AvroSchema(
 		Fields:    nil,
 	}
 	for i, col := range input.columns {
+		if col == nil {
+			continue
+		}
 		avroType, err := a.columnToAvroSchema(col, input.colInfos[i].Ft)
 		if err != nil {
 			return nil, err
diff --git a/pkg/sink/codec/avro/decoder.go b/pkg/sink/codec/avro/decoder.go
index 736bdc4a8..fb14e8af6 100644
--- a/pkg/sink/codec/avro/decoder.go
+++ b/pkg/sink/codec/avro/decoder.go
@@ -80,7 +80,7 @@ func (d *decoder) HasNext() (model.MessageType, bool, error) {
 		return model.MessageTypeRow, true, nil
 	}
 	if len(d.value) < 1 {
-		return model.MessageTypeUnknown, false, errors.ErrAvroInvalidMessage.FastGenByArgs()
+		return model.MessageTypeUnknown, false, errors.ErrAvroInvalidMessage.FastGenByArgs(d.value)
 	}
 	switch d.value[0] {
 	case magicByte:
@@ -90,7 +90,7 @@ func (d *decoder) HasNext() (model.MessageType, bool, error) {
 	case checkpointByte:
 		return model.MessageTypeResolved, true, nil
 	}
-	return model.MessageTypeUnknown, false, errors.ErrAvroInvalidMessage.FastGenByArgs()
+	return model.MessageTypeUnknown, false, errors.ErrAvroInvalidMessage.FastGenByArgs(d.value)
 }
 
 // NextRowChangedEvent returns the next row changed event if exists
diff --git a/pkg/sink/kafka/claimcheck/claim_check.go b/pkg/sink/kafka/claimcheck/claim_check.go
index 860cf0392..f88a6b10f 100644
--- a/pkg/sink/kafka/claimcheck/claim_check.go
+++ b/pkg/sink/kafka/claimcheck/claim_check.go
@@ -16,7 +16,7 @@ package claimcheck
 import (
 	"context"
 	"encoding/json"
-	"path/filepath"
+	"strings"
 	"time"
 
 	"github.com/google/uuid"
@@ -102,7 +102,7 @@ func (c *ClaimCheck) WriteMessage(ctx context.Context, key, value []byte, fileNa
 
 // FileNameWithPrefix returns the file name with prefix, the full path.
 func (c *ClaimCheck) FileNameWithPrefix(fileName string) string {
-	return filepath.Join(c.storage.URI(), fileName)
+	return strings.TrimSuffix(c.storage.URI(), "/") + "/" + fileName
 }
 
 // CleanMetrics the claim check by clean up the metrics.
diff --git a/scripts/check-diff-line-width.sh b/scripts/check-diff-line-width.sh
index af65760cc..a034d0a2b 100755
--- a/scripts/check-diff-line-width.sh
+++ b/scripts/check-diff-line-width.sh
@@ -27,6 +27,7 @@ git --no-pager diff $BASE_HASH -U0 -- cdc pkg cmd \
 	-- ':(exclude)*_gen.go' \
 	-- ':(exclude)*_gen_test.go' \
 	-- ':(exclude)*_mock.go' \
+	-- ':(exclude)*_test_data.go' \
 	-- ':(exclude)*.pb.go' |
 	grep -E '^\+' | grep -vE '^\+\+\+' | grep -vE 'json:' | grep -vE 'toml:' |
 	sed 's/\t/    /g' |


Test patch:
diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go
index 8e3b697e4..c6b999c55 100644
--- a/cdc/api/v2/model_test.go
+++ b/cdc/api/v2/model_test.go
@@ -31,7 +31,7 @@ import (
 // note: this is api published default value, not change it
 var defaultAPIConfig = &ReplicaConfig{
 	MemoryQuota:        config.DefaultChangefeedMemoryQuota,
-	CaseSensitive:      true,
+	CaseSensitive:      false,
 	CheckGCSafePoint:   true,
 	BDRMode:            util.AddressOf(false),
 	EnableSyncPoint:    util.AddressOf(false),
@@ -60,11 +60,14 @@ var defaultAPIConfig = &ReplicaConfig{
 		AdvanceTimeoutInSec:              util.AddressOf(uint(150)),
 	},
 	Consistent: &ConsistentConfig{
-		Level:             "none",
-		MaxLogSize:        64,
-		FlushIntervalInMs: redo.DefaultFlushIntervalInMs,
-		Storage:           "",
-		UseFileBackend:    false,
+		Level:                 "none",
+		MaxLogSize:            64,
+		FlushIntervalInMs:     redo.DefaultFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs,
+		EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+		FlushWorkerNum:        redo.DefaultFlushWorkerNum,
+		Storage:               "",
+		UseFileBackend:        false,
 	},
 	Scheduler: &ChangefeedSchedulerConfig{
 		EnableTableAcrossNodes: config.GetDefaultReplicaConfig().
@@ -80,6 +83,7 @@ var defaultAPIConfig = &ReplicaConfig{
 	},
 	ChangefeedErrorStuckDuration: &JSONDuration{*config.
 		GetDefaultReplicaConfig().ChangefeedErrorStuckDuration},
+	SQLMode: config.GetDefaultReplicaConfig().SQLMode,
 }
 
 func TestDefaultReplicaConfig(t *testing.T) {
diff --git a/cdc/entry/schema_storage_test.go b/cdc/entry/schema_storage_test.go
index 145ce2ee9..8cda912ba 100644
--- a/cdc/entry/schema_storage_test.go
+++ b/cdc/entry/schema_storage_test.go
@@ -549,7 +549,7 @@ func TestMultiVersionStorage(t *testing.T) {
 	jobs = append(jobs, job)
 	f, err := filter.NewFilter(config.GetDefaultReplicaConfig(), "")
 	require.Nil(t, err)
-	storage, err := NewSchemaStorage(nil, 0, false, dummyChangeFeedID, util.RoleTester, f)
+	storage, err := NewSchemaStorage(nil, 0, false, model.DefaultChangeFeedID("dummy"), util.RoleTester, f)
 	require.Nil(t, err)
 	for _, job := range jobs {
 		err := storage.HandleDDLJob(job)
@@ -883,7 +883,7 @@ func TestSchemaStorage(t *testing.T) {
 		jobs, err := getAllHistoryDDLJob(store, f)
 		require.Nil(t, err)
 
-		schemaStorage, err := NewSchemaStorage(nil, 0, false, dummyChangeFeedID, util.RoleTester, f)
+		schemaStorage, err := NewSchemaStorage(nil, 0, false, model.DefaultChangeFeedID("dummy"), util.RoleTester, f)
 		require.Nil(t, err)
 		for _, job := range jobs {
 			err := schemaStorage.HandleDDLJob(job)
@@ -1009,3 +1009,27 @@ func TestHandleKey(t *testing.T) {
 	require.True(t, ok)
 	require.Equal(t, int64(-2), tb3.HandleIndexID)
 }
+
+func TestGetPrimaryKey(t *testing.T) {
+	t.Parallel()
+
+	helper := NewSchemaTestHelper(t)
+	defer helper.Close()
+
+	sql := `create table test.t1(a int primary key, b int)`
+	job := helper.DDL2Job(sql)
+	tableInfo := model.WrapTableInfo(0, "test", 0, job.BinlogInfo.TableInfo)
+
+	names := tableInfo.GetPrimaryKeyColumnNames()
+	require.Len(t, names, 1)
+	require.Containsf(t, names, "a", "names: %v", names)
+
+	sql = `create table test.t2(a int, b int, c int, primary key(a, b))`
+	job = helper.DDL2Job(sql)
+	tableInfo = model.WrapTableInfo(0, "test", 0, job.BinlogInfo.TableInfo)
+
+	names = tableInfo.GetPrimaryKeyColumnNames()
+	require.Len(t, names, 2)
+	require.Containsf(t, names, "a", "names: %v", names)
+	require.Containsf(t, names, "b", "names: %v", names)
+}
diff --git a/cdc/model/changefeed_test.go b/cdc/model/changefeed_test.go
index 318c0242f..16dcbd5b9 100644
--- a/cdc/model/changefeed_test.go
+++ b/cdc/model/changefeed_test.go
@@ -247,7 +247,7 @@ func TestVerifyAndComplete(t *testing.T) {
 		StartTs: 417257993615179777,
 		Config: &config.ReplicaConfig{
 			MemoryQuota:        1073741824,
-			CaseSensitive:      true,
+			CaseSensitive:      false,
 			CheckGCSafePoint:   true,
 			SyncPointInterval:  util.AddressOf(time.Minute * 10),
 			SyncPointRetention: util.AddressOf(time.Hour * 24),
diff --git a/cdc/model/schema_storage_test.go b/cdc/model/schema_storage_test.go
index 0da4e7561..2cbc940bf 100644
--- a/cdc/model/schema_storage_test.go
+++ b/cdc/model/schema_storage_test.go
@@ -339,17 +339,17 @@ func TestColumnsByNames(t *testing.T) {
 	}
 
 	names := []string{"col1", "col2", "col3"}
-	offsets, ok := tableInfo.ColumnsByNames(names)
+	offsets, ok := tableInfo.OffsetsByNames(names)
 	require.True(t, ok)
 	require.Equal(t, []int{0, 1, 2}, offsets)
 
 	names = []string{"col2"}
-	offsets, ok = tableInfo.ColumnsByNames(names)
+	offsets, ok = tableInfo.OffsetsByNames(names)
 	require.True(t, ok)
 	require.Equal(t, []int{1}, offsets)
 
 	names = []string{"col1", "col-not-found"}
-	offsets, ok = tableInfo.ColumnsByNames(names)
+	offsets, ok = tableInfo.OffsetsByNames(names)
 	require.False(t, ok)
 	require.Nil(t, offsets)
 }
diff --git a/cdc/owner/changefeed_test.go b/cdc/owner/changefeed_test.go
index b184d3067..faef25010 100644
--- a/cdc/owner/changefeed_test.go
+++ b/cdc/owner/changefeed_test.go
@@ -519,9 +519,12 @@ func TestRemoveChangefeed(t *testing.T) {
 	dir := t.TempDir()
 	info.SinkURI = "mysql://"
 	info.Config.Consistent = &config.ConsistentConfig{
-		Level:             "eventual",
-		Storage:           filepath.Join("nfs://", dir),
-		FlushIntervalInMs: redo.DefaultFlushIntervalInMs,
+		Level:                 "eventual",
+		Storage:               filepath.Join("nfs://", dir),
+		FlushIntervalInMs:     redo.DefaultFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs,
+		EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+		FlushWorkerNum:        redo.DefaultFlushWorkerNum,
 	}
 	ctx = cdcContext.WithChangefeedVars(ctx, &cdcContext.ChangefeedVars{
 		ID:   ctx.ChangefeedVars().ID,
diff --git a/cdc/owner/ddl_sink_test.go b/cdc/owner/ddl_sink_test.go
index be3c3ee58..52883f2e1 100644
--- a/cdc/owner/ddl_sink_test.go
+++ b/cdc/owner/ddl_sink_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/ddlsink"
+	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/retry"
 	"github.com/stretchr/testify/require"
@@ -61,7 +62,13 @@ func (m *mockSink) GetDDL() *model.DDLEvent {
 
 func newDDLSink4Test(reportErr func(err error), reportWarn func(err error)) (DDLSink, *mockSink) {
 	mockSink := &mockSink{}
-	ddlSink := newDDLSink(model.DefaultChangeFeedID("changefeed-test"), &model.ChangeFeedInfo{}, reportErr, reportWarn)
+	ddlSink := newDDLSink(
+		model.DefaultChangeFeedID("changefeed-test"),
+		&model.ChangeFeedInfo{
+			Config: config.GetDefaultReplicaConfig(),
+		},
+		reportErr,
+		reportWarn)
 	ddlSink.(*ddlSinkImpl).sinkInitHandler = func(ctx context.Context, s *ddlSinkImpl) error {
 		s.sink = mockSink
 		return nil
@@ -545,6 +552,9 @@ func TestAddSpecialComment(t *testing.T) {
 	}
 
 	s := &ddlSinkImpl{}
+	s.info = &model.ChangeFeedInfo{
+		Config: config.GetDefaultReplicaConfig(),
+	}
 	for _, ca := range testCase {
 		re, err := s.addSpecialComment(ca.event)
 		require.Nil(t, err)
diff --git a/cdc/processor/processor_test.go b/cdc/processor/processor_test.go
index 042e243a2..b25073ba1 100644
--- a/cdc/processor/processor_test.go
+++ b/cdc/processor/processor_test.go
@@ -72,11 +72,14 @@ func newProcessor4Test(
 			tmpDir := t.TempDir()
 			redoDir := fmt.Sprintf("%s/%s", tmpDir, changefeedID)
 			dmlMgr := redo.NewDMLManager(changefeedID, &config.ConsistentConfig{
-				Level:             string(redoPkg.ConsistentLevelEventual),
-				MaxLogSize:        redoPkg.DefaultMaxLogSize,
-				FlushIntervalInMs: redoPkg.DefaultFlushIntervalInMs,
-				Storage:           "file://" + redoDir,
-				UseFileBackend:    false,
+				Level:                 string(redoPkg.ConsistentLevelEventual),
+				MaxLogSize:            redoPkg.DefaultMaxLogSize,
+				FlushIntervalInMs:     redoPkg.DefaultFlushIntervalInMs,
+				MetaFlushIntervalInMs: redoPkg.DefaultMetaFlushIntervalInMs,
+				EncodingWorkerNum:     redoPkg.DefaultEncodingWorkerNum,
+				FlushWorkerNum:        redoPkg.DefaultFlushWorkerNum,
+				Storage:               "file://" + redoDir,
+				UseFileBackend:        false,
 			})
 			p.redo.r = dmlMgr
 		}
diff --git a/cdc/redo/manager_test.go b/cdc/redo/manager_test.go
index 14db8ffc2..381f4e997 100644
--- a/cdc/redo/manager_test.go
+++ b/cdc/redo/manager_test.go
@@ -114,11 +114,14 @@ func TestLogManagerInProcessor(t *testing.T) {
 	testWriteDMLs := func(storage string, useFileBackend bool) {
 		ctx, cancel := context.WithCancel(ctx)
 		cfg := &config.ConsistentConfig{
-			Level:             string(redo.ConsistentLevelEventual),
-			MaxLogSize:        redo.DefaultMaxLogSize,
-			Storage:           storage,
-			FlushIntervalInMs: redo.MinFlushIntervalInMs,
-			UseFileBackend:    useFileBackend,
+			Level:                 string(redo.ConsistentLevelEventual),
+			MaxLogSize:            redo.DefaultMaxLogSize,
+			Storage:               storage,
+			FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+			MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
+			EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+			FlushWorkerNum:        redo.DefaultFlushWorkerNum,
+			UseFileBackend:        useFileBackend,
 		}
 		dmlMgr := NewDMLManager(model.DefaultChangeFeedID("test"), cfg)
 		var eg errgroup.Group
@@ -221,11 +224,14 @@ func TestLogManagerInOwner(t *testing.T) {
 	testWriteDDLs := func(storage string, useFileBackend bool) {
 		ctx, cancel := context.WithCancel(ctx)
 		cfg := &config.ConsistentConfig{
-			Level:             string(redo.ConsistentLevelEventual),
-			MaxLogSize:        redo.DefaultMaxLogSize,
-			Storage:           storage,
-			FlushIntervalInMs: redo.MinFlushIntervalInMs,
-			UseFileBackend:    useFileBackend,
+			Level:                 string(redo.ConsistentLevelEventual),
+			MaxLogSize:            redo.DefaultMaxLogSize,
+			Storage:               storage,
+			FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+			MetaFlushIntervalInMs: redo.DefaultMetaFlushIntervalInMs,
+			EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+			FlushWorkerNum:        redo.DefaultFlushWorkerNum,
+			UseFileBackend:        useFileBackend,
 		}
 		startTs := model.Ts(10)
 		ddlMgr := NewDDLManager(model.DefaultChangeFeedID("test"), cfg, startTs)
@@ -266,10 +272,13 @@ func TestLogManagerError(t *testing.T) {
 	defer cancel()
 
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           "blackhole-invalid://",
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               "blackhole-invalid://",
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
+		EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+		FlushWorkerNum:        redo.DefaultFlushWorkerNum,
 	}
 	logMgr := NewDMLManager(model.DefaultChangeFeedID("test"), cfg)
 	var eg errgroup.Group
@@ -317,11 +326,14 @@ func BenchmarkFileWriter(b *testing.B) {
 func runBenchTest(b *testing.B, storage string, useFileBackend bool) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           storage,
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
-		UseFileBackend:    useFileBackend,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               storage,
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
+		EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+		FlushWorkerNum:        redo.DefaultFlushWorkerNum,
+		UseFileBackend:        useFileBackend,
 	}
 	dmlMgr := NewDMLManager(model.DefaultChangeFeedID("test"), cfg)
 	var eg errgroup.Group
diff --git a/cdc/redo/meta_manager_test.go b/cdc/redo/meta_manager_test.go
index 675b96dcf..7804c4497 100644
--- a/cdc/redo/meta_manager_test.go
+++ b/cdc/redo/meta_manager_test.go
@@ -72,10 +72,13 @@ func TestInitAndWriteMeta(t *testing.T) {
 
 	startTs := uint64(10)
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           uri.String(),
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               uri.String(),
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
+		EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+		FlushWorkerNum:        redo.DefaultFlushWorkerNum,
 	}
 	m := NewMetaManager(changefeedID, cfg, startTs)
 
@@ -150,10 +153,13 @@ func TestPreCleanupAndWriteMeta(t *testing.T) {
 
 	startTs := uint64(10)
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           uri.String(),
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               uri.String(),
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
+		EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+		FlushWorkerNum:        redo.DefaultFlushWorkerNum,
 	}
 	m := NewMetaManager(changefeedID, cfg, startTs)
 
@@ -280,10 +286,13 @@ func TestGCAndCleanup(t *testing.T) {
 
 	startTs := uint64(3)
 	cfg := &config.ConsistentConfig{
-		Level:             string(redo.ConsistentLevelEventual),
-		MaxLogSize:        redo.DefaultMaxLogSize,
-		Storage:           uri.String(),
-		FlushIntervalInMs: redo.MinFlushIntervalInMs,
+		Level:                 string(redo.ConsistentLevelEventual),
+		MaxLogSize:            redo.DefaultMaxLogSize,
+		Storage:               uri.String(),
+		FlushIntervalInMs:     redo.MinFlushIntervalInMs,
+		MetaFlushIntervalInMs: redo.MinFlushIntervalInMs,
+		EncodingWorkerNum:     redo.DefaultEncodingWorkerNum,
+		FlushWorkerNum:        redo.DefaultFlushWorkerNum,
 	}
 	m := NewMetaManager(changefeedID, cfg, startTs)
 
diff --git a/cdc/redo/writer/memory/main_test.go b/cdc/redo/writer/memory/main_test.go
index 6e92d8f86..2f68850c5 100644
--- a/cdc/redo/writer/memory/main_test.go
+++ b/cdc/redo/writer/memory/main_test.go
@@ -20,10 +20,5 @@ import (
 )
 
 func TestMain(m *testing.M) {
-	bak := defaultEncodingInputChanSize
-	defer func() {
-		defaultEncodingInputChanSize = bak
-	}()
-	defaultEncodingInputChanSize = 0
 	leakutil.SetUpLeakTest(m)
 }
diff --git a/cdc/sink/dmlsink/mq/transformer/columnselector/column_selector_test.go b/cdc/sink/dmlsink/mq/transformer/columnselector/column_selector_test.go
new file mode 100644
index 000000000..30e5aa563
--- /dev/null
+++ b/cdc/sink/dmlsink/mq/transformer/columnselector/column_selector_test.go
@@ -0,0 +1,276 @@
+// Copyright 2023 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.
+
+//go:build intest
+// +build intest
+
+package columnselector
+
+import (
+	"testing"
+
+	"github.com/pingcap/tiflow/cdc/entry"
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dispatcher"
+	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/pingcap/tiflow/pkg/errors"
+	"github.com/stretchr/testify/require"
+)
+
+func TestNewColumnSelector(t *testing.T) {
+	// the column selector is not set
+	replicaConfig := config.GetDefaultReplicaConfig()
+	selectors, err := New(replicaConfig)
+	require.NoError(t, err)
+	require.NotNil(t, selectors)
+	require.Len(t, selectors.selectors, 0)
+
+	event := &model.RowChangedEvent{
+		Table: &model.TableName{
+			Schema: "test",
+			Table:  "table1",
+		},
+		Columns: []*model.Column{
+			{
+				Name:  "col1",
+				Value: []byte("val1"),
+			},
+			{
+				Name:  "col2",
+				Value: []byte("val2"),
+			},
+		},
+		PreColumns: []*model.Column{
+			{
+				Name:  "col1",
+				Value: []byte("val1"),
+			},
+			{
+				Name:  "col2",
+				Value: []byte("val2"),
+			},
+		},
+	}
+
+	for _, column := range event.Columns {
+		require.NotNil(t, column)
+	}
+	for _, column := range event.PreColumns {
+		require.NotNil(t, column)
+	}
+
+	replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{
+		{
+			Matcher: []string{"test.*"},
+			Columns: []string{"a", "b"},
+		},
+		{
+			Matcher: []string{"test1.*"},
+			Columns: []string{"*", "!a"},
+		},
+		{
+			Matcher: []string{"test2.*"},
+			Columns: []string{"co*", "!col2"},
+		},
+		{
+			Matcher: []string{"test3.*"},
+			Columns: []string{"co?1"},
+		},
+	}
+	selectors, err = New(replicaConfig)
+	require.NoError(t, err)
+	require.Len(t, selectors.selectors, 4)
+}
+
+func TestVerifyTables(t *testing.T) {
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
+
+	sql := `create table test.t1(
+    	a int primary key,
+    	b int,
+    	c int,
+    	d int,
+    	e int,
+    	unique key uk_b_c(b, c),
+    	unique key uk_d_e(d, e),
+    	key idx_c_d(c, d))`
+	job := helper.DDL2Job(sql)
+	tableInfo := model.WrapTableInfo(0, "test", 0, job.BinlogInfo.TableInfo)
+	infos := []*model.TableInfo{tableInfo}
+
+	replicaConfig := config.GetDefaultReplicaConfig()
+	replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{
+		{
+			Matcher: []string{"test.t1"},
+			Columns: []string{"a", "b"},
+		},
+	}
+	selectors, err := New(replicaConfig)
+	require.NoError(t, err)
+
+	eventRouter, err := dispatcher.NewEventRouter(replicaConfig, config.ProtocolDefault, "default", "default")
+	require.NoError(t, err)
+	// handle key column `a` is retained, so return no error
+	err = selectors.VerifyTables(infos, eventRouter)
+	require.NoError(t, err)
+
+	event4Test := func() *model.RowChangedEvent {
+		return &model.RowChangedEvent{
+			Table:     &model.TableName{Schema: "test", Table: "t1"},
+			TableInfo: tableInfo,
+			Columns: []*model.Column{
+				{
+					Name:  "a",
+					Value: []byte("1"),
+				},
+				{
+					Name:  "b",
+					Value: []byte("2"),
+				},
+				{
+					Name:  "c",
+					Value: []byte("3"),
+				},
+				{
+					Name:  "d",
+					Value: []byte("4"),
+				},
+				{
+					Name:  "e",
+					Value: []byte("5"),
+				},
+			},
+			PreColumns: []*model.Column{
+				{
+					Name:  "a",
+					Value: []byte("1"),
+				},
+				{
+					Name:  "b",
+					Value: []byte("2"),
+				},
+				{
+					Name:  "c",
+					Value: []byte("3"),
+				},
+				{
+					Name:  "d",
+					Value: []byte("4"),
+				},
+				{
+					Name:  "e",
+					Value: []byte("5"),
+				},
+			},
+		}
+	}
+
+	event := event4Test()
+	err = selectors.Apply(event)
+	require.NoError(t, err)
+	require.Nil(t, event.Columns[2])
+	require.Nil(t, event.Columns[3])
+	require.Nil(t, event.Columns[4])
+	require.Nil(t, event.PreColumns[2])
+	require.Nil(t, event.PreColumns[3])
+	require.Nil(t, event.PreColumns[4])
+
+	event = event4Test()
+	event.PreColumns = nil
+	err = selectors.Apply(event)
+	require.NoError(t, err)
+
+	event = event4Test()
+	event.Columns = nil
+	err = selectors.Apply(event)
+	require.NoError(t, err)
+
+	replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{
+		{
+			Matcher: []string{"test.t1"},
+			Columns: []string{"b", "c"},
+		},
+	}
+	selectors, err = New(replicaConfig)
+	require.NoError(t, err)
+
+	// handle key column `a` is filtered out, but unique key `uk_b_c` has all columns retained, so return no error.
+	err = selectors.VerifyTables(infos, eventRouter)
+	require.NoError(t, err)
+
+	event = event4Test()
+	err = selectors.Apply(event)
+	require.NoError(t, err)
+	require.Nil(t, event.Columns[0])
+	require.Nil(t, event.Columns[3])
+	require.Nil(t, event.Columns[4])
+	require.Nil(t, event.PreColumns[0])
+	require.Nil(t, event.PreColumns[3])
+	require.Nil(t, event.PreColumns[4])
+
+	replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{
+		{
+			Matcher: []string{"test.*"},
+			Columns: []string{"c", "d"},
+		},
+		{
+			Matcher: []string{"test.t1"},
+			Columns: []string{"a", "b"},
+		},
+	}
+	selectors, err = New(replicaConfig)
+	require.NoError(t, err)
+
+	// handle key column `a` is filtered out, no one unique key has all columns retained, so return error.
+	err = selectors.VerifyTables(infos, eventRouter)
+	require.ErrorIs(t, err, errors.ErrColumnSelectorFailed)
+
+	event = event4Test()
+	err = selectors.Apply(event)
+	require.ErrorIs(t, err, errors.ErrColumnSelectorFailed)
+}
+
+func TestVerifyTablesColumnFilteredInDispatcher(t *testing.T) {
+	replicaConfig := config.GetDefaultReplicaConfig()
+	replicaConfig.Sink.ColumnSelectors = []*config.ColumnSelector{
+		{
+			Matcher: []string{"test.*"},
+			Columns: []string{"a", "b"},
+		},
+	}
+	replicaConfig.Sink.DispatchRules = []*config.DispatchRule{
+		{
+			Matcher:       []string{"test.*"},
+			PartitionRule: "columns",
+			Columns:       []string{"c"},
+		},
+	}
+
+	selectors, err := New(replicaConfig)
+	require.NoError(t, err)
+
+	eventRouter, err := dispatcher.NewEventRouter(replicaConfig, config.ProtocolDefault, "default", "default")
+	require.NoError(t, err)
+
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
+
+	sql := `create table test.t1(a int primary key, b int, c int)`
+	job := helper.DDL2Job(sql)
+	tableInfo := model.WrapTableInfo(0, "test", 0, job.BinlogInfo.TableInfo)
+	infos := []*model.TableInfo{tableInfo}
+
+	err = selectors.VerifyTables(infos, eventRouter)
+	require.ErrorIs(t, err, errors.ErrColumnSelectorFailed)
+}
diff --git a/cdcv2/metadata/sql/tables_test.go b/cdcv2/metadata/sql/tables_test.go
index f74fa8de4..0612b0aa2 100644
--- a/cdcv2/metadata/sql/tables_test.go
+++ b/cdcv2/metadata/sql/tables_test.go
@@ -36,6 +36,7 @@ func newMockDB(t *testing.T) (*sql.DB, *gorm.DB, sqlmock.Sqlmock) {
 
 func TestAutoMigrateTables(t *testing.T) {
 	t.Parallel()
+	t.Skip("skip this test since it's unstable")
 
 	backendDB, db, mock := newMockDB(t)
 	defer backendDB.Close()
diff --git a/dm/pkg/gtid/gtid_test.go b/dm/pkg/gtid/gtid_test.go
index 275b82527..7698128c5 100644
--- a/dm/pkg/gtid/gtid_test.go
+++ b/dm/pkg/gtid/gtid_test.go
@@ -67,7 +67,7 @@ func TestParseGTIDNoFlavor(t *testing.T) {
 	require.Error(t, err)
 }
 
-func TestIsNilGTIDSet(t *testing.T) {
+func TestIsNilMySQLGTIDSet(t *testing.T) {
 	t.Parallel()
 
 	require.False(t, IsZeroMySQLGTIDSet(""))
@@ -79,6 +79,17 @@ func TestIsNilGTIDSet(t *testing.T) {
 	require.True(t, IsZeroMySQLGTIDSet(" xxxxx:0 "))
 }
 
+func TestIsNilMariaDBGTIDSet(t *testing.T) {
+	t.Parallel()
+
+	require.False(t, IsZeroMariaDBGTIDSet(""))
+	require.False(t, IsZeroMariaDBGTIDSet("xxxxx"))
+	require.False(t, IsZeroMariaDBGTIDSet("a-b-0,c-d:0"))
+	require.False(t, IsZeroMariaDBGTIDSet("xxxxx:1"))
+	require.True(t, IsZeroMariaDBGTIDSet("x-y-0"))
+	require.True(t, IsZeroMariaDBGTIDSet(" x-y-0 "))
+}
+
 func TestParseZeroAsEmptyGTIDSet(t *testing.T) {
 	t.Parallel()
 
@@ -94,3 +105,50 @@ func TestParseZeroAsEmptyGTIDSet(t *testing.T) {
 	require.NoError(t, err)
 	require.Equal(t, "", gset.String())
 }
+
+func TestIsZeroGTIDSet(t *testing.T) {
+	t.Parallel()
+
+	testCases := []struct {
+		gsetStr string
+		isZero  bool
+		flavor  string
+	}{
+		{
+			"",
+			true,
+			mysql.MySQLFlavor,
+		},
+		{
+			"",
+			true,
+			mysql.MariaDBFlavor,
+		},
+		{
+			"3ccc475b-2343-11e7-be21-6c0b84d59f30:0",
+			true,
+			mysql.MySQLFlavor,
+		},
+		{
+			"0-0-0",
+			true,
+			mysql.MariaDBFlavor,
+		},
+		{
+			"3ccc475b-2343-11e7-be21-6c0b84d59f30:1-14",
+			false,
+			mysql.MySQLFlavor,
+		},
+		{
+			"1-2-3",
+			false,
+			mysql.MariaDBFlavor,
+		},
+	}
+	for i, testCase := range testCases {
+		t.Logf("test case %d", i)
+		gset, err := ParserGTID(testCase.flavor, testCase.gsetStr)
+		require.NoError(t, err)
+		require.Equal(t, testCase.isZero, IsZeroGTIDSet(gset))
+	}
+}
diff --git a/dm/tests/dmctl_command/run.sh b/dm/tests/dmctl_command/run.sh
index aef277f41..7b22de9c1 100644
--- a/dm/tests/dmctl_command/run.sh
+++ b/dm/tests/dmctl_command/run.sh
@@ -11,6 +11,11 @@ db_name=$TEST_NAME
 # 44 normal help message + 1 "PASS" line
 help_cnt=45
 
+function get_uuid() {
+	uuid=$(echo "show variables like '%server_uuid%';" | MYSQL_PWD=123456 mysql -uroot -h$1 -P$2 | awk 'FNR == 2 {print $2}')
+	echo $uuid
+}
+
 function run() {
 	# check dmctl output with help flag
 	# it should usage for root command
@@ -427,25 +432,77 @@ function run_validation_start_stop_cmd {
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"query-status test2" \
 		"\"unit\": \"Sync\"" 1
+	uuid=($(get_uuid $MYSQL_HOST2 $MYSQL_PORT2))
+	echo "--> (fail) validation update: on non-existed validator"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update test -s $SOURCE_ID2 --cutover-binlog-gtid $uuid:1-999" \
+		"\"result\": false" 1 \
+		"validator not found for task" 1
 
 	echo "--> (success) validation start: start all tasks"
 	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"validation start --all-task --mode fast" \
 		"\"result\": true" 1
+
+	echo "--> (fail) validation update: missing task-name"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update" \
+		"\`task-name\` should be set" 1
+	echo "--> (fail) validation update: multi tasks"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update test test2" \
+		"should specify only one" 1
+	echo "--> (fail) validation update: non-existed subtask"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update xxxxx" \
+		"\"result\": false" 1 \
+		"cannot get subtask by task name \`xxxxx\`" 1
+	echo "--> (fail) validation update: non-exist source"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update -s xxxxx test" \
+		"\"result\": false" 1 \
+		"cannot get subtask by task name \`test\` and sources" 1
+	echo "--> (fail) validation update: not specify cutover-binlog-gtid"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update test -s $SOURCE_ID2" \
+		"\"result\": false" 1 \
+		"didn't specify cutover-binlog-gtid" 1
+	echo "--> (fail) validation update: not specify cutover-binlog-pos"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update test2 -s $SOURCE_ID1" \
+		"\"result\": false" 1 \
+		"didn't specify cutover-binlog-pos" 1
+	echo "--> (fail) validation update: invalid cutover-binlog-pos"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update test -s $SOURCE_ID1 --cutover-binlog-pos xxx" \
+		"invalid binlog pos" 1
+	echo "--> (fail) validation update: specify both cutover-binlog-pos and cutover-binlog-gtid"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update test -s $SOURCE_ID1 --cutover-binlog-pos '(mysql-bin.000001, 2345)' --cutover-binlog-gtid $uuid:1-999" \
+		"you must specify either one of" 1
+	echo "--> (success) validation update: on exist source"
+	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update test -s $SOURCE_ID2 --cutover-binlog-gtid $uuid:1-999" \
+		"\"result\": true" 2
+
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"validation status test" \
 		"\"result\": true" 1 \
 		"\"mode\": \"full\"" 0 \
 		"\"mode\": \"fast\"" 2 \
 		"\"stage\": \"Running\"" 2 \
-		"\"stage\": \"Stopped\"" 0
+		"\"stage\": \"Stopped\"" 0 \
+		"\"cutoverBinlogPos\"" 2 \
+		"\"cutoverBinlogGtid\": \"$uuid:1-999\"" 1
 	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
 		"validation status test2" \
 		"\"result\": true" 1 \
 		"\"mode\": \"full\"" 0 \
 		"\"mode\": \"fast\"" 1 \
 		"\"stage\": \"Running\"" 1 \
-		"\"stage\": \"Stopped\"" 0
+		"\"stage\": \"Stopped\"" 0 \
+		"\"cutoverBinlogPos\"" 1 \
+		"\"cutoverBinlogGtid\": \"$uuid:1-999\"" 0
 
 	echo "--> (success) validation stop: stop all tasks"
 	run_dm_ctl $WORK_DIR "127.0.0.1:$MASTER_PORT" \
diff --git a/dm/tests/validator_basic/conf/dm-task-standalone-long-interval.yaml b/dm/tests/validator_basic/conf/dm-task-standalone-long-interval.yaml
new file mode 100644
index 000000000..eb8d14ac3
--- /dev/null
+++ b/dm/tests/validator_basic/conf/dm-task-standalone-long-interval.yaml
@@ -0,0 +1,37 @@
+---
+name: test
+task-mode: all
+heartbeat-update-interval: 1
+heartbeat-report-interval: 1
+
+target-database:
+  host: "127.0.0.1"
+  port: 4000
+  user: "root"
+  password: ""
+
+mysql-instances:
+  - source-id: "mysql-replica-01"
+    block-allow-list:  "instance"
+    syncer-config-name: "global"
+    validator-config-name: global
+
+block-allow-list:
+  instance:
+    do-dbs: ["validator_basic"]
+
+syncers:
+  global:
+    worker-count: 16
+    batch: 100
+
+validators:
+  global:
+    mode: full
+    worker-count: 4
+    validate-interval: 1500s # increase validate-interval and meta-flush-interval to make sure the validator won't trigger validation automatically
+    check-interval: 1s
+    row-error-delay: 1s
+    meta-flush-interval: 1500s
+    max-pending-row-size: 500m
+    max-pending-row-count: 1000
diff --git a/dm/tests/validator_basic/run.sh b/dm/tests/validator_basic/run.sh
index be3f71ad6..3a664d8ea 100644
--- a/dm/tests/validator_basic/run.sh
+++ b/dm/tests/validator_basic/run.sh
@@ -8,6 +8,11 @@ WORK_DIR=$TEST_DIR/$TEST_NAME
 
 db_name=$TEST_NAME
 
+function get_gtid_executed() {
+	gtid_executed=$(echo "show master status;" | MYSQL_PWD=123456 mysql -uroot -h$1 -P$2 | awk 'FNR == 2 {print $3}')
+	echo $gtid_executed
+}
+
 function prepare_dm_and_source() {
 	cleanup_process $*
 	cleanup_data $db_name
@@ -719,6 +724,79 @@ function test_dup_autopk() {
 		"new\/ignored\/resolved: 0\/0\/0" 2
 }
 
+function test_update_validator() {
+	# backup it, will change it in the following case
+	cp $cur/conf/dm-task-standalone.yaml $cur/conf/dm-task-standalone.yaml.bak
+	cp $cur/conf/source1.yaml $cur/conf/source1.yaml.bak
+
+	trap restore_on_case_exit EXIT
+	# clear fail point
+	export GO_FAILPOINTS=""
+	cp $cur/conf/source1.yaml.bak $cur/conf/source1.yaml
+	sed -i 's/enable-gtid: false/enable-gtid: 'true'/g' $cur/conf/source1.yaml
+	sed -i 's/enable-relay: false/enable-relay: 'true'/g' $cur/conf/source1.yaml
+	run_sql_source1 "flush binary logs"
+	sleep 1
+	run_sql_source1 "purge binary logs before now()"
+	prepare_dm_and_source
+	dmctl_start_task_standalone $cur/conf/dm-task-standalone-long-interval.yaml --remove-meta
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"config source $SOURCE_ID1" \
+		"enable-gtid: true" 1 \
+		"enable-relay: true" 1
+	run_sql_source1 "insert into $db_name.t1(id, name) values(2,'b'), (3, 'c')"
+	run_sql_source1 "delete from $db_name.t1 where id=1"
+	# check validator before start, should pass
+	# validate pass
+	trigger_checkpoint_flush
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"\"processedRowsStatus\": \"insert\/update\/delete: 2\/0\/1\"" 1 \
+		"pendingRowsStatus\": \"insert\/update\/delete: 2\/0\/1" 1 \
+		"new\/ignored\/resolved: 0\/0\/0" 1 \
+		"\"cutoverBinlogGtid\": \"\"" 1
+	gtid_executed=($(get_gtid_executed $MYSQL_HOST1 $MYSQL_PORT1))
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update test -s $SOURCE_ID1 --cutover-binlog-gtid $gtid_executed" \
+		"\"result\": true" 2
+	sleep 3
+	# after 3 seconds, cutoverBinlogGtid should also not change
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"\"cutoverBinlogGtid\": \"$gtid_executed\"" 1
+	trigger_checkpoint_flush
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"\"synced\": true" 1
+	check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"\"processedRowsStatus\": \"insert\/update\/delete: 2\/0\/1\"" 1 \
+		"pendingRowsStatus\": \"insert\/update\/delete: 0\/0\/0" 1 \
+		"new\/ignored\/resolved: 0\/0\/0" 1
+	# set gtid_executed again, should be triggered by dml event
+	gtid_executed=($(get_gtid_executed $MYSQL_HOST1 $MYSQL_PORT1))
+	extracted_number=$(echo "$gtid_executed" | grep -Eo '[0-9]+$')
+	incremented_number=$((extracted_number + 1))
+	gtid_cutover=$(echo "$gtid_executed" | sed "s/${extracted_number}$/${incremented_number}/")
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"validation update test -s $SOURCE_ID1 --cutover-binlog-gtid $gtid_cutover" \
+		"\"result\": true" 2
+	run_sql_source1 "insert into $db_name.t1(id, name) values (4, 'c')"
+	# directly sleep without flush, can cutover by itself
+	sleep 1.5
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"\"synced\": true" 1
+	check_sync_diff $WORK_DIR $cur/conf/diff_config.toml
+	run_dm_ctl_with_retry $WORK_DIR "127.0.0.1:$MASTER_PORT" \
+		"query-status test" \
+		"\"processedRowsStatus\": \"insert\/update\/delete: 3\/0\/1\"" 1 \
+		"pendingRowsStatus\": \"insert\/update\/delete: 0\/0\/0" 1 \
+		"new\/ignored\/resolved: 0\/0\/0" 1 \
+		"\"cutoverBinlogGtid\": \"\"" 1
+}
+
 run_standalone $*
 validate_table_with_different_pk
 test_unsupported_table_status
@@ -726,6 +804,7 @@ stopped_validator_fail_over
 test_data_filter
 test_validation_syncer_stopped
 test_dup_autopk
+test_update_validator
 cleanup_process $*
 cleanup_data $db_name
 cleanup_data_upstream $db_name
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index 35caa03f1..b1c3aa628 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -70,6 +70,9 @@ const (
     "level": "none",
     "max-log-size": 64,
     "flush-interval": 2000,
+    "meta-flush-interval": 200,
+    "encoding-worker-num": 16,
+    "flush-worker-num": 8,
     "storage": "",
     "use-file-backend": false
   },
@@ -81,7 +84,8 @@ const (
     "integrity-check-level": "none",
     "corruption-handle-level": "warn"
  },
-  "changefeed-error-stuck-duration": 1800000000000
+  "changefeed-error-stuck-duration": 1800000000000,
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
 }`
 
 	testCfgTestServerConfigMarshal = `{
@@ -294,6 +298,9 @@ const (
     "level": "none",
     "max-log-size": 64,
     "flush-interval": 2000,
+    "meta-flush-interval": 200,
+    "encoding-worker-num": 16,
+    "flush-worker-num": 8,
     "storage": "",
     "use-file-backend": false
   },
@@ -308,7 +315,8 @@ const (
     "integrity-check-level": "none",
     "corruption-handle-level": "warn"
   },
-  "changefeed-error-stuck-duration": 1800000000000
+  "changefeed-error-stuck-duration": 1800000000000,
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
 }`
 
 	testCfgTestReplicaConfigMarshal2 = `{
@@ -441,6 +449,9 @@ const (
     "level": "none",
     "max-log-size": 64,
     "flush-interval": 2000,
+    "meta-flush-interval": 200,
+    "encoding-worker-num": 16,
+    "flush-worker-num": 8,
     "storage": "",
     "use-file-backend": false
   },
@@ -453,6 +464,7 @@ const (
     "integrity-check-level": "none",
     "corruption-handle-level": "warn"
   },
-  "changefeed-error-stuck-duration": 1800000000000
+  "changefeed-error-stuck-duration": 1800000000000,
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
 }`
 )
diff --git a/pkg/config/replica_config_test.go b/pkg/config/replica_config_test.go
index 12e1f0422..09da7e25d 100644
--- a/pkg/config/replica_config_test.go
+++ b/pkg/config/replica_config_test.go
@@ -22,6 +22,7 @@ import (
 
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/pingcap/tiflow/pkg/compression"
+	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/integrity"
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/stretchr/testify/require"
@@ -235,6 +236,22 @@ func TestReplicaConfigValidate(t *testing.T) {
 	require.Error(t, err)
 }
 
+func TestValidateIntegrity(t *testing.T) {
+	sinkURL, err := url.Parse("kafka://topic?protocol=avro")
+	require.NoError(t, err)
+
+	cfg := GetDefaultReplicaConfig()
+	cfg.Integrity.IntegrityCheckLevel = integrity.CheckLevelCorrectness
+	cfg.Sink.ColumnSelectors = []*ColumnSelector{
+		{
+			Matcher: []string{"a.b"}, Columns: []string{"c"},
+		},
+	}
+
+	err = cfg.ValidateAndAdjust(sinkURL)
+	require.ErrorIs(t, err, cerror.ErrInvalidReplicaConfig)
+}
+
 func TestValidateAndAdjust(t *testing.T) {
 	cfg := GetDefaultReplicaConfig()
 
diff --git a/pkg/filter/expr_filter_bench_test.go b/pkg/filter/expr_filter_bench_test.go
index 91c3fe903..62263e8e2 100644
--- a/pkg/filter/expr_filter_bench_test.go
+++ b/pkg/filter/expr_filter_bench_test.go
@@ -57,7 +57,7 @@ func BenchmarkSkipDML(b *testing.B) {
 	sessCtx := utils.NewSessionCtx(map[string]string{
 		"time_zone": "",
 	})
-	f, err := newExprFilter("", cfg)
+	f, err := newExprFilter("", cfg, config.GetDefaultReplicaConfig().SQLMode)
 	require.Nil(b, err)
 
 	type innerCase struct {
diff --git a/pkg/filter/expr_filter_test.go b/pkg/filter/expr_filter_test.go
index 372ace442..66fe1f02a 100644
--- a/pkg/filter/expr_filter_test.go
+++ b/pkg/filter/expr_filter_test.go
@@ -324,7 +324,7 @@ func TestShouldSkipDMLBasic(t *testing.T) {
 
 	for _, tc := range testCases {
 		tableInfo := helper.execDDL(tc.ddl)
-		f, err := newExprFilter("", tc.cfg)
+		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		for _, c := range tc.cases {
 			rowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.row, tableInfo.Columns)
@@ -441,7 +441,7 @@ func TestShouldSkipDMLError(t *testing.T) {
 
 	for _, tc := range testCases {
 		tableInfo := helper.execDDL(tc.ddl)
-		f, err := newExprFilter("", tc.cfg)
+		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		for _, c := range tc.cases {
 			rowDatums, err := utils.AdjustBinaryProtocolForDatum(sessCtx, c.row, tableInfo.Columns)
@@ -634,7 +634,7 @@ func TestShouldSkipDMLTableUpdated(t *testing.T) {
 
 	for _, tc := range testCases {
 		tableInfo := helper.execDDL(tc.ddl)
-		f, err := newExprFilter("", tc.cfg)
+		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		for _, c := range tc.cases {
 			if c.updateDDl != "" {
@@ -754,7 +754,7 @@ func TestVerify(t *testing.T) {
 			ti := helper.execDDL(ddl)
 			tableInfos = append(tableInfos, ti)
 		}
-		f, err := newExprFilter("", tc.cfg)
+		f, err := newExprFilter("", tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.Nil(t, err)
 		err = f.verify(tableInfos)
 		require.True(t, errors.ErrorEqual(tc.err, err), "case: %+v", tc, err)
diff --git a/pkg/filter/sql_event_filter_test.go b/pkg/filter/sql_event_filter_test.go
index 1fa03bedc..9e9707d07 100644
--- a/pkg/filter/sql_event_filter_test.go
+++ b/pkg/filter/sql_event_filter_test.go
@@ -171,7 +171,7 @@ func TestShouldSkipDDL(t *testing.T) {
 	}
 
 	for _, tc := range testCases {
-		f, err := newSQLEventFilter(tc.cfg)
+		f, err := newSQLEventFilter(tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 		require.True(t, errors.ErrorEqual(err, tc.err), "case: %+s", err)
 		for _, c := range tc.cases {
 			ddl := &model.DDLEvent{
@@ -298,7 +298,7 @@ func TestShouldSkipDML(t *testing.T) {
 		tc := tc
 		t.Run(tc.name, func(t *testing.T) {
 			t.Parallel()
-			f, err := newSQLEventFilter(tc.cfg)
+			f, err := newSQLEventFilter(tc.cfg, config.GetDefaultReplicaConfig().SQLMode)
 			require.NoError(t, err)
 			for _, c := range tc.cases {
 				event := &model.RowChangedEvent{
diff --git a/pkg/filter/utils_test.go b/pkg/filter/utils_test.go
index 6207ce974..c3441e108 100644
--- a/pkg/filter/utils_test.go
+++ b/pkg/filter/utils_test.go
@@ -14,12 +14,14 @@
 package filter
 
 import (
+	"fmt"
 	"testing"
 
 	"github.com/pingcap/log"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/parser"
 	timodel "github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/parser/mysql"
 	tifilter "github.com/pingcap/tidb/util/filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
@@ -174,3 +176,61 @@ func TestDDLToTypeSpecialDDL(t *testing.T) {
 		require.Equal(t, c.evenType, et, "case%v", c.ddl)
 	}
 }
+
+func TestToDDLEventWithSQLMode(t *testing.T) {
+	t.Parallel()
+	cases := []struct {
+		name    string
+		query   string
+		jobTp   timodel.ActionType
+		sqlMode string // sql mode
+		expect  bf.EventType
+		errMsg  string
+	}{
+		{
+			name:    "create table",
+			query:   "create table t1(id int primary key)",
+			jobTp:   timodel.ActionCreateTable,
+			sqlMode: config.GetDefaultReplicaConfig().SQLMode,
+			expect:  bf.CreateTable,
+		},
+		{
+			name:    "drop table",
+			query:   "drop table t1",
+			jobTp:   timodel.ActionDropTable,
+			sqlMode: config.GetDefaultReplicaConfig().SQLMode,
+			expect:  bf.DropTable,
+		},
+		{ // "" in table name or column name are not supported when sqlMode is set to ANSI_QUOTES
+			name:    "create table 2",
+			query:   `create table "t1" ("id" int primary key)`,
+			jobTp:   timodel.ActionCreateTable,
+			sqlMode: config.GetDefaultReplicaConfig().SQLMode,
+			expect:  bf.CreateTable,
+			errMsg:  "ErrConvertDDLToEventTypeFailed",
+		},
+		{ // "" in table name or column name are supported when sqlMode is set to ANSI_QUOTES
+			name:    "create table 3",
+			query:   `create table "t1" ("id" int primary key)`,
+			jobTp:   timodel.ActionCreateTable,
+			sqlMode: fmt.Sprint(config.GetDefaultReplicaConfig().SQLMode + ",ANSI_QUOTES"),
+			expect:  bf.CreateTable,
+		},
+	}
+	for _, c := range cases {
+		innerCase := c
+		t.Run(c.name, func(t *testing.T) {
+			t.Parallel()
+			p := parser.New()
+			mode, err := mysql.GetSQLMode(innerCase.sqlMode)
+			require.NoError(t, err)
+			p.SetSQLMode(mode)
+			tp, err := ddlToEventType(p, innerCase.query, innerCase.jobTp)
+			if innerCase.errMsg != "" {
+				require.Contains(t, err.Error(), innerCase.errMsg, innerCase.name)
+			} else {
+				require.Equal(t, innerCase.expect, tp)
+			}
+		})
+	}
+}
diff --git a/pkg/orchestrator/reactor_state_test.go b/pkg/orchestrator/reactor_state_test.go
index fd0b29f3a..61dc26f4b 100644
--- a/pkg/orchestrator/reactor_state_test.go
+++ b/pkg/orchestrator/reactor_state_test.go
@@ -123,6 +123,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity: config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
+						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -177,6 +178,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity: config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
+						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -236,6 +238,7 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity: config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
+						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -330,6 +333,7 @@ func TestPatchInfo(t *testing.T) {
 			Scheduler:                    defaultConfig.Scheduler,
 			Integrity:                    defaultConfig.Integrity,
 			ChangefeedErrorStuckDuration: defaultConfig.ChangefeedErrorStuckDuration,
+			SQLMode:                      defaultConfig.SQLMode,
 		},
 	}
 	cfInfo.RmUnusedFields()
@@ -352,6 +356,7 @@ func TestPatchInfo(t *testing.T) {
 			Scheduler:                    defaultConfig.Scheduler,
 			Integrity:                    defaultConfig.Integrity,
 			ChangefeedErrorStuckDuration: defaultConfig.ChangefeedErrorStuckDuration,
+			SQLMode:                      defaultConfig.SQLMode,
 		},
 	}
 	cfInfo.RmUnusedFields()
diff --git a/pkg/sink/cloudstorage/path_test.go b/pkg/sink/cloudstorage/path_test.go
index f0efcb7ac..6e97da55c 100644
--- a/pkg/sink/cloudstorage/path_test.go
+++ b/pkg/sink/cloudstorage/path_test.go
@@ -17,9 +17,14 @@ import (
 	"context"
 	"fmt"
 	"net/url"
+	"os"
+	"path/filepath"
 	"testing"
 	"time"
 
+	timodel "github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tidb/parser/mysql"
+	"github.com/pingcap/tidb/parser/types"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/engine/pkg/clock"
 	"github.com/pingcap/tiflow/pkg/config"
@@ -275,3 +280,51 @@ func TestIsSchemaFile(t *testing.T) {
 			"testCase: %s, path: %v", tt.name, tt.path)
 	}
 }
+
+func TestCheckOrWriteSchema(t *testing.T) {
+	t.Parallel()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	dir := t.TempDir()
+	f := testFilePathGenerator(ctx, t, dir)
+
+	var columns []*timodel.ColumnInfo
+	ft := types.NewFieldType(mysql.TypeLong)
+	ft.SetFlag(mysql.PriKeyFlag | mysql.NotNullFlag)
+	col := &timodel.ColumnInfo{
+		Name:         timodel.NewCIStr("Id"),
+		FieldType:    *ft,
+		DefaultValue: 10,
+	}
+	columns = append(columns, col)
+	tableInfo := &model.TableInfo{
+		TableInfo: &timodel.TableInfo{Columns: columns},
+		Version:   100,
+		TableName: model.TableName{
+			Schema:  "test",
+			Table:   "table1",
+			TableID: 20,
+		},
+	}
+
+	table := VersionedTableName{
+		TableNameWithPhysicTableID: tableInfo.TableName,
+		TableInfoVersion:           tableInfo.Version,
+	}
+
+	err := f.CheckOrWriteSchema(ctx, table, tableInfo)
+	require.NoError(t, err)
+	require.Equal(t, tableInfo.Version, f.versionMap[table])
+
+	// test only table version changed, schema file should be reused
+	table.TableInfoVersion = 101
+	err = f.CheckOrWriteSchema(ctx, table, tableInfo)
+	require.NoError(t, err)
+	require.Equal(t, tableInfo.Version, f.versionMap[table])
+
+	dir = filepath.Join(dir, "test/table1/meta")
+	cnt, err := os.ReadDir(dir)
+	require.NoError(t, err)
+	require.Equal(t, 1, len(cnt))
+}
diff --git a/pkg/sink/codec/open/open_protocol_encoder_test.go b/pkg/sink/codec/open/open_protocol_encoder_test.go
index 887831b1e..6d4bf6b02 100644
--- a/pkg/sink/codec/open/open_protocol_encoder_test.go
+++ b/pkg/sink/codec/open/open_protocol_encoder_test.go
@@ -462,7 +462,7 @@ func TestE2EClaimCheckMessage(t *testing.T) {
 	ctx := context.Background()
 	topic := ""
 
-	a := 241
+	a := 244
 	codecConfig := common.NewConfig(config.ProtocolOpen).WithMaxMessageBytes(a)
 	codecConfig.LargeMessageHandle.LargeMessageHandleOption = config.LargeMessageHandleOptionClaimCheck
 	codecConfig.LargeMessageHandle.LargeMessageHandleCompression = compression.LZ4
diff --git a/pkg/sink/kafka/claimcheck/claim_check_test.go b/pkg/sink/kafka/claimcheck/claim_check_test.go
new file mode 100644
index 000000000..802c1c441
--- /dev/null
+++ b/pkg/sink/kafka/claimcheck/claim_check_test.go
@@ -0,0 +1,34 @@
+// Copyright 2023 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 claimcheck
+
+import (
+	"context"
+	"testing"
+
+	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/stretchr/testify/require"
+)
+
+func TestClaimCheckFileName(t *testing.T) {
+	ctx := context.Background()
+	storageURI := "file:///tmp/abc/"
+	changefeedID := model.DefaultChangeFeedID("test")
+
+	claimCheck, err := New(ctx, storageURI, changefeedID)
+	require.NoError(t, err)
+
+	fileName := claimCheck.FileNameWithPrefix("file.json")
+	require.Equal(t, "file:////tmp/abc/file.json", fileName)
+}
diff --git a/tests/integration_tests/api_v2/cases.go b/tests/integration_tests/api_v2/cases.go
index f62cf45e0..9e336a016 100644
--- a/tests/integration_tests/api_v2/cases.go
+++ b/tests/integration_tests/api_v2/cases.go
@@ -79,7 +79,7 @@ var customReplicaConfig = &ReplicaConfig{
 // defaultReplicaConfig check if the default values is changed
 var defaultReplicaConfig = &ReplicaConfig{
 	MemoryQuota:      1024 * 1024 * 1024,
-	CaseSensitive:    true,
+	CaseSensitive:    false,
 	CheckGCSafePoint: true,
 	Filter: &FilterConfig{
 		Rules: []string{"*.*"},
diff --git a/tests/integration_tests/api_v2/model.go b/tests/integration_tests/api_v2/model.go
index 7ef56dffc..3c2006f5c 100644
--- a/tests/integration_tests/api_v2/model.go
+++ b/tests/integration_tests/api_v2/model.go
@@ -272,11 +272,14 @@ type ColumnSelector struct {
 // ConsistentConfig represents replication consistency config for a changefeed
 // This is a duplicate of config.ConsistentConfig
 type ConsistentConfig struct {
-	Level             string `json:"level"`
-	MaxLogSize        int64  `json:"max_log_size"`
-	FlushIntervalInMs int64  `json:"flush_interval"`
-	Storage           string `json:"storage"`
-	UseFileBackend    bool   `json:"use_file_backend"`
+	Level                 string `json:"level"`
+	MaxLogSize            int64  `json:"max_log_size"`
+	FlushIntervalInMs     int64  `json:"flush_interval"`
+	MetaFlushIntervalInMs int64  `json:"meta_flush_interval"`
+	EncoderWorkerNum      int    `json:"encoder_worker_num"`
+	FlushWorkerNum        int    `json:"flush_worker_num"`
+	Storage               string `json:"storage"`
+	UseFileBackend        bool   `json:"use_file_backend"`
 }
 
 // ChangefeedSchedulerConfig is per changefeed scheduler settings.
diff --git a/tests/integration_tests/cli/run.sh b/tests/integration_tests/cli/run.sh
index 3d27a3fbd..4e38f238e 100644
--- a/tests/integration_tests/cli/run.sh
+++ b/tests/integration_tests/cli/run.sh
@@ -73,7 +73,7 @@ function run() {
 
 	# Update changefeed failed because changefeed is running
 	cat - >"$WORK_DIR/changefeed.toml" <<EOF
-case-sensitive = false
+case-sensitive = true
 [scheduler]
 enable-table-across-nodes = true
 EOF
@@ -91,7 +91,7 @@ EOF
 	# Update changefeed
 	run_cdc_cli changefeed update --pd=$pd_addr --config="$WORK_DIR/changefeed.toml" --no-confirm --changefeed-id $uuid
 	changefeed_info=$(curl -s -X GET "http://127.0.0.1:8300/api/v2/changefeeds/$uuid/meta_info" 2>&1)
-	if [[ ! $changefeed_info == *"\"case_sensitive\":false"* ]]; then
+	if [[ ! $changefeed_info == *"\"case_sensitive\":true"* ]]; then
 		echo "[$(date)] <<<<< changefeed info is not updated as expected ${changefeed_info} >>>>>"
 		exit 1
 	fi
diff --git a/tests/integration_tests/kafka_column_selector/conf/changefeed.toml b/tests/integration_tests/kafka_column_selector/conf/changefeed.toml
new file mode 100644
index 000000000..addd3dd41
--- /dev/null
+++ b/tests/integration_tests/kafka_column_selector/conf/changefeed.toml
@@ -0,0 +1,7 @@
+[sink]
+column-selectors = [
+    {matcher = ['test.t1'], columns = ['a', 'b']},
+    {matcher = ['test.*'], columns = ["*", "!b"]},
+
+    {matcher = ['test1.t1'], columns = ['column*', '!column1']},
+]
diff --git a/tests/integration_tests/kafka_column_selector/data/data.sql b/tests/integration_tests/kafka_column_selector/data/data.sql
new file mode 100644
index 000000000..79b4dcee0
--- /dev/null
+++ b/tests/integration_tests/kafka_column_selector/data/data.sql
@@ -0,0 +1,49 @@
+drop database if exists `test`;
+create database `test`;
+use `test`;
+
+create table t1 (
+    a int primary key,
+    b int,
+    c int
+);
+
+insert into t1 values (1, 2, 3);
+insert into t1 values (2, 3, 4);
+insert into t1 values (3, 4, 5);
+
+create table t2 (
+    a int primary key,
+    b int,
+    c int
+);
+
+insert into t2 values (1, 2, 3);
+insert into t2 values (2, 3, 4);
+insert into t2 values (3, 4, 5);
+
+create table t3 (
+    a int primary key,
+    b int,
+    c int
+);
+
+insert into t3 values (1, 2, 3);
+insert into t3 values (2, 3, 4);
+insert into t3 values (3, 4, 5);
+
+drop database if exists `test1`;
+create database `test1`;
+use `test1`;
+
+create table t1 (
+    column0 int primary key,
+    column1 int,
+    column2 int
+);
+
+insert into t1 values (1, 2, 3);
+insert into t1 values (2, 3, 4);
+insert into t1 values (3, 4, 5);
+
+create table finishmark(id int primary key);
diff --git a/tests/integration_tests/kafka_column_selector/run.sh b/tests/integration_tests/kafka_column_selector/run.sh
new file mode 100644
index 000000000..e8bd3a83f
--- /dev/null
+++ b/tests/integration_tests/kafka_column_selector/run.sh
@@ -0,0 +1,52 @@
+#!/bin/bash
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+function run() {
+	# test kafka sink only in this case
+	if [ "$SINK_TYPE" != "kafka" ]; then
+		return
+	fi
+
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	# record tso before we create tables to skip the system table DDLs
+	start_ts=$(cdc cli tso query --pd=http://$UP_PD_HOST_1:$UP_PD_PORT_1)
+
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	changefeed_id="test"
+	TOPIC_NAME="column-selector-test"
+	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=canal-json&partition-num=1&enable-tidb-extension=true"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c ${changefeed_id} --config="$CUR/conf/changefeed.toml"
+
+	cdc_kafka_consumer --upstream-uri $SINK_URI --downstream-uri="mysql://root@127.0.0.1:3306/?safe-mode=true&batch-dml-enable=false" --upstream-tidb-dsn="root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/?" --config="$CUR/conf/changefeed.toml" 2>&1 &
+
+	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
+	echo "Starting build checksum checker..."
+	cd $CUR/../../utils/checksum_checker
+	if [ ! -f ./checksum_checker ]; then
+		GO111MODULE=on go build
+	fi
+
+	check_table_exists "test1.finishmark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	./checksum_checker --upstream-uri "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/" --downstream-uri "root@tcp(${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT})/" --databases "test,test1" --config="$CUR/conf/changefeed.toml"
+
+	cleanup_process $CDC_BINARY
+}
+
+trap stop_tidb_cluster EXIT
+run $*
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"
diff --git a/tests/integration_tests/kafka_column_selector_avro/conf/changefeed.toml b/tests/integration_tests/kafka_column_selector_avro/conf/changefeed.toml
new file mode 100644
index 000000000..85c3512e3
--- /dev/null
+++ b/tests/integration_tests/kafka_column_selector_avro/conf/changefeed.toml
@@ -0,0 +1,4 @@
+[sink]
+column-selectors = [
+    {matcher = ['test.*'], columns = ["*", "!b"]},
+]
diff --git a/tests/integration_tests/kafka_column_selector_avro/data/data.sql b/tests/integration_tests/kafka_column_selector_avro/data/data.sql
new file mode 100644
index 000000000..15513e5dc
--- /dev/null
+++ b/tests/integration_tests/kafka_column_selector_avro/data/data.sql
@@ -0,0 +1,15 @@
+drop database if exists `test`;
+create database `test`;
+use `test`;
+
+create table t1 (
+    a int primary key,
+    b int,
+    c int
+);
+
+insert into t1 values (1, 2, 3);
+insert into t1 values (2, 3, 4);
+insert into t1 values (3, 4, 5);
+
+create table finishmark(id int primary key);
diff --git a/tests/integration_tests/kafka_column_selector_avro/run.sh b/tests/integration_tests/kafka_column_selector_avro/run.sh
new file mode 100644
index 000000000..e98fc71fc
--- /dev/null
+++ b/tests/integration_tests/kafka_column_selector_avro/run.sh
@@ -0,0 +1,66 @@
+#!/bin/bash
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+function run() {
+	# test kafka sink only in this case
+	if [ "$SINK_TYPE" != "kafka" ]; then
+		return
+	fi
+
+	echo 'Starting schema registry...'
+	./bin/bin/schema-registry-start -daemon ./bin/etc/schema-registry/schema-registry.properties
+	i=0
+	while ! curl -o /dev/null -v -s "http://127.0.0.1:8088"; do
+		i=$(($i + 1))
+		if [ $i -gt 30 ]; then
+			echo 'Failed to start schema registry'
+			exit 1
+		fi
+		sleep 2
+	done
+
+	curl -X PUT -H "Content-Type: application/vnd.schemaregistry.v1+json" --data '{"compatibility": "NONE"}' http://127.0.0.1:8088/config
+
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	# record tso before we create tables to skip the system table DDLs
+	start_ts=$(cdc cli tso query --pd=http://$UP_PD_HOST_1:$UP_PD_PORT_1)
+
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	changefeed_id="test"
+	TOPIC_NAME="column-selector-avro-test"
+	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=avro&enable-tidb-extension=true&avro-enable-watermark=true"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" -c ${changefeed_id} --config="$CUR/conf/changefeed.toml" --schema-registry=http://127.0.0.1:8088
+
+	run_kafka_consumer $WORK_DIR $SINK_URI $CUR/conf/changefeed.toml "http://127.0.0.1:8088"
+
+	run_sql_file $CUR/data/data.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
+	echo "Starting build checksum checker..."
+	cd $CUR/../../utils/checksum_checker
+	if [ ! -f ./checksum_checker ]; then
+		GO111MODULE=on go build
+	fi
+
+	check_table_exists "test.finishmark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	./checksum_checker --upstream-uri "root@tcp(${UP_TIDB_HOST}:${UP_TIDB_PORT})/" --downstream-uri "root@tcp(${DOWN_TIDB_HOST}:${DOWN_TIDB_PORT})/" --databases "test" --config="$CUR/conf/changefeed.toml"
+
+	cleanup_process $CDC_BINARY
+}
+
+trap stop_tidb_cluster EXIT
+run $*
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"
diff --git a/tests/integration_tests/run_group.sh b/tests/integration_tests/run_group.sh
index 0364ef683..e6211a5a4 100755
--- a/tests/integration_tests/run_group.sh
+++ b/tests/integration_tests/run_group.sh
@@ -14,7 +14,7 @@ mysql_only="bdr_mode capture_suicide_while_balance_table syncpoint hang_sink_sui
 mysql_only_http="http_api http_api_tls api_v2"
 mysql_only_consistent_replicate="consistent_replicate_ddl consistent_replicate_gbk consistent_replicate_nfs consistent_replicate_storage_file consistent_replicate_storage_file_large_value consistent_replicate_storage_s3 consistent_partition_table"
 
-kafka_only="kafka_big_messages kafka_compression kafka_messages kafka_sink_error_resume mq_sink_lost_callback mq_sink_dispatcher"
+kafka_only="kafka_big_messages kafka_compression kafka_messages kafka_sink_error_resume mq_sink_lost_callback mq_sink_dispatcher kafka_column_selector kafka_column_selector_avro"
 kafka_only_protocol="canal_json_adapter_compatibility canal_json_basic multi_topics avro_basic canal_json_handle_key_only open_protocol_handle_key_only canal_json_claim_check open_protocol_claim_check"
 kafka_only_v2="kafka_big_txn_v2 kafka_big_messages_v2 multi_tables_ddl_v2 multi_topics_v2"
 
diff --git a/tests/utils/checksum_checker/main.go b/tests/utils/checksum_checker/main.go
new file mode 100644
index 000000000..5e2a86e42
--- /dev/null
+++ b/tests/utils/checksum_checker/main.go
@@ -0,0 +1,260 @@
+// Copyright 2021 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 main
+
+import (
+	"database/sql"
+	"flag"
+	"fmt"
+	"os"
+	"strings"
+	"time"
+
+	"github.com/pingcap/log"
+	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/transformer/columnselector"
+	cmdUtil "github.com/pingcap/tiflow/pkg/cmd/util"
+	"github.com/pingcap/tiflow/pkg/config"
+	"github.com/pingcap/tiflow/pkg/errors"
+	"go.uber.org/zap"
+)
+
+type options struct {
+	upstreamURI   string
+	downstreamURI string
+	dbNames       string
+	configFile    string
+}
+
+func (o *options) validate() error {
+	if o.upstreamURI == "" {
+		return errors.New("upstreamURI is required")
+	}
+	if o.downstreamURI == "" {
+		return errors.New("downstreamURI is required")
+	}
+	if len(o.dbNames) == 0 {
+		return errors.New("dbNames is required")
+	}
+	return nil
+}
+
+func main() {
+	o := &options{}
+
+	flags := flag.NewFlagSet(os.Args[0], flag.ExitOnError)
+	flags.StringVar(&o.upstreamURI, "upstream-uri", "", "upstream database uri")
+	flags.StringVar(&o.downstreamURI, "downstream-uri", "", "downstream database uri")
+	flags.StringVar(&o.dbNames, "databases", "", "database names, separate by the `,`")
+	flags.StringVar(&o.configFile, "config", "", "config file")
+	if err := flags.Parse(os.Args[1:]); err != nil {
+		log.Panic("parse args failed", zap.Error(err))
+	}
+	if err := o.validate(); err != nil {
+		log.Panic("invalid options", zap.Error(err))
+	}
+
+	upstreamDB, err := openDB(o.upstreamURI)
+	if err != nil {
+		log.Panic("cannot open db for the upstream", zap.Error(err))
+	}
+
+	downstreamDB, err := openDB(o.downstreamURI)
+	if err != nil {
+		log.Panic("cannot open db for the downstream", zap.Error(err))
+	}
+
+	replicaConfig := config.GetDefaultReplicaConfig()
+	if o.configFile != "" {
+		err = cmdUtil.StrictDecodeFile(o.configFile, "checksum checker", replicaConfig)
+		if err != nil {
+			log.Panic("cannot decode config file", zap.Error(err))
+		}
+	}
+
+	columnFilter, err := columnselector.New(replicaConfig)
+	if err != nil {
+		log.Panic("cannot create column filter", zap.Error(err))
+	}
+
+	dbNames := strings.Split(o.dbNames, ",")
+	err = compareCRC32CheckSum(upstreamDB, downstreamDB, dbNames, columnFilter)
+	if err != nil {
+		log.Panic("compare checksum failed", zap.Error(err))
+	}
+	log.Info("compare checksum passed")
+}
+
+func compareCRC32CheckSum(
+	upstream, downstream *sql.DB, dbNames []string, selector *columnselector.ColumnSelector,
+) error {
+	start := time.Now()
+	source, err := getChecksum(upstream, dbNames, selector)
+	if err != nil {
+		log.Warn("get checksum for the upstream failed", zap.Error(err))
+		return errors.Trace(err)
+	}
+	log.Info("get checksum for the upstream success",
+		zap.Duration("elapsed", time.Since(start)))
+
+	start = time.Now()
+	sink, err := getChecksum(downstream, dbNames, selector)
+	if err != nil {
+		log.Warn("get checksum for the downstream failed", zap.Error(err))
+		return errors.Trace(err)
+	}
+	log.Info("get checksum for the downstream success",
+		zap.Duration("elapsed", time.Since(start)))
+
+	if len(source) != len(sink) {
+		log.Error("source and sink have different crc32 size",
+			zap.Int("source", len(source)), zap.Int("sink", len(sink)))
+		return fmt.Errorf("source and sink have different crc32 size, source: %d, sink: %d",
+			len(source), len(sink))
+	}
+
+	for tableName, expected := range source {
+		actual, ok := sink[tableName]
+		if !ok {
+			return fmt.Errorf("table not found at sink, table: %s", tableName)
+		}
+		if expected != actual {
+			log.Error("crc32 mismatch",
+				zap.String("table", tableName), zap.Uint32("source", expected), zap.Uint32("sink", actual))
+			return fmt.Errorf("crc32 mismatch, table: %s, source: %d, sink: %d", tableName, expected, actual)
+		}
+	}
+	return nil
+}
+
+func getChecksum(
+	db *sql.DB, dbNames []string, selector *columnselector.ColumnSelector,
+) (map[string]uint32, error) {
+	result := make(map[string]uint32)
+	for _, dbName := range dbNames {
+		tables, err := getAllTables(db, dbName)
+		if err != nil {
+			return nil, err
+		}
+		for _, table := range tables {
+			tx, err := db.Begin()
+			if err != nil {
+				_ = tx.Rollback()
+				return nil, errors.Trace(err)
+			}
+			columns, err := getColumns(tx, dbName, table, selector)
+			if err != nil {
+				_ = tx.Rollback()
+				return nil, errors.Trace(err)
+			}
+			checksum, err := doChecksum(tx, dbName, table, columns)
+			if err != nil {
+				_ = tx.Rollback()
+				return nil, errors.Trace(err)
+			}
+			_ = tx.Commit()
+			result[dbName+"."+table] = checksum
+		}
+	}
+	return result, nil
+}
+
+func doChecksum(tx *sql.Tx, schema, table string, columns []string) (uint32, error) {
+	a := strings.Join(columns, "`,`")
+
+	concat := fmt.Sprintf("CONCAT_WS(',', `%s`)", a)
+	tableName := schema + "." + table
+	query := fmt.Sprintf("SELECT BIT_XOR(CRC32(%s)) AS checksum FROM %s", concat, tableName)
+	var checkSum uint32
+	rows := tx.QueryRow(query)
+	err := rows.Scan(&checkSum)
+	if err != nil {
+		log.Error("get crc32 checksum failed",
+			zap.Error(err), zap.String("table", tableName), zap.String("query", query))
+		return 0, errors.Trace(err)
+	}
+	log.Info("do checkSum success", zap.String("table", tableName), zap.Uint32("checkSum", checkSum))
+	return checkSum, nil
+}
+
+func getColumns(tx *sql.Tx, schema, table string, selector *columnselector.ColumnSelector) (result []string, err error) {
+	rows, err := tx.Query(fmt.Sprintf("SHOW COLUMNS FROM %s", schema+"."+table))
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+	defer func() {
+		if err := rows.Close(); err != nil {
+			log.Warn("close rows failed", zap.Error(err))
+		}
+	}()
+
+	for rows.Next() {
+		var t columnInfo
+		if err := rows.Scan(&t.Field, &t.Type, &t.Null, &t.Key, &t.Default, &t.Extra); err != nil {
+			return result, errors.Trace(err)
+		}
+		if selector.VerifyColumn(schema, table, t.Field) {
+			result = append(result, t.Field)
+		}
+	}
+	return result, nil
+}
+
+type columnInfo struct {
+	Field   string
+	Type    string
+	Null    string
+	Key     string
+	Default *string
+	Extra   string
+}
+
+func getAllTables(db *sql.DB, dbName string) ([]string, error) {
+	var result []string
+	dbName = strings.TrimSpace(dbName)
+	tx, err := db.Begin()
+	if err != nil {
+		_ = tx.Rollback()
+		return nil, errors.Trace(err)
+	}
+	query := fmt.Sprintf(`show full tables from %s where table_type != "VIEW"`, dbName)
+	rows, err := tx.Query(query)
+	if err != nil {
+		_ = tx.Rollback()
+		return nil, errors.Trace(err)
+	}
+	for rows.Next() {
+		var t string
+		var tt string
+		if err := rows.Scan(&t, &tt); err != nil {
+			_ = tx.Rollback()
+			return nil, errors.Trace(err)
+		}
+		result = append(result, t)
+	}
+	_ = rows.Close()
+	_ = tx.Commit()
+	return result, nil
+}
+
+func openDB(uri string) (*sql.DB, error) {
+	db, err := sql.Open("mysql", uri)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	if err := db.Ping(); err != nil {
+		return nil, errors.Trace(err)
+	}
+	return db, nil
+}