Repository: pingcap/tiflow
Pull request #11155 (2024-05-21 10:01:19)

(redo)ticdc: fix the event orderliness in redo log (#11117)
This is an automated cherry-pick of #11117

<!--
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/11096

### What is changed and how it works?
1. Fix the less function of logHeap to make dml event of redo log in the same txn sorted as delete/update/insert.
2. use startTs instead of commitTs in redo apply to distinguish different txns.

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test
 - Manual test (add detailed scripts or steps below)
 
    create mysql sink changefeed with redo log on
    run gotpc workload
    After 30m, pause changefeed and run redo apply

#### 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]: 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]: [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 [liuzix](https://github.com/liuzix) for approval. For more information see [the 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":["liuzix"]} -->
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 | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/ghpr_verify/463/display/redirect) | true | `/test verify`
pull-cdc-integration-pulsar-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_pulsar_test/391/display/redirect) | true | `/test cdc-integration-pulsar-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11155). [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 | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/ghpr_verify/463/display/redirect) | true | `/test verify`
pull-cdc-integration-pulsar-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_pulsar_test/391/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-mysql-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_mysql_test/497/display/redirect) | true | `/test cdc-integration-mysql-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11155). [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
--- | --- | --- | --- | ---
pull-cdc-integration-pulsar-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_pulsar_test/391/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-mysql-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_mysql_test/497/display/redirect) | true | `/test cdc-integration-mysql-test`
pull-cdc-integration-kafka-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_kafka_test/656/display/redirect) | true | `/test cdc-integration-kafka-test`
jenkins-ticdc/verify | a50f69fd893a9628aefb8b003192c47d72d5a44b | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/ghpr_verify/466/display/redirect) | true | `/test verify`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11155). [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
--- | --- | --- | --- | ---
pull-cdc-integration-pulsar-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_pulsar_test/391/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-kafka-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_kafka_test/656/display/redirect) | true | `/test cdc-integration-kafka-test`
jenkins-ticdc/verify | a50f69fd893a9628aefb8b003192c47d72d5a44b | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/ghpr_verify/466/display/redirect) | true | `/test verify`
pull-cdc-integration-mysql-test | a50f69fd893a9628aefb8b003192c47d72d5a44b | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_mysql_test/499/display/redirect) | true | `/test cdc-integration-mysql-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11155). [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
--- | --- | --- | --- | ---
pull-cdc-integration-pulsar-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_pulsar_test/391/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-cdc-integration-kafka-test | ad0018682712efe1bf311290bd615fbdb0f3e852 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_kafka_test/656/display/redirect) | true | `/test cdc-integration-kafka-test`
jenkins-ticdc/verify | a50f69fd893a9628aefb8b003192c47d72d5a44b | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/ghpr_verify/466/display/redirect) | true | `/test verify`
pull-cdc-integration-mysql-test | a50f69fd893a9628aefb8b003192c47d72d5a44b | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_cdc_integration_mysql_test/499/display/redirect) | true | `/test cdc-integration-mysql-test`
pull-dm-integration-test | a50f69fd893a9628aefb8b003192c47d72d5a44b | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/release-7.5/job/pull_dm_integration_test/447/display/redirect) | true | `/test dm-integration-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11155). [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/middleware/middleware.go b/cdc/api/middleware/middleware.go
index 0681e1846..51214fa28 100644
--- a/cdc/api/middleware/middleware.go
+++ b/cdc/api/middleware/middleware.go
@@ -46,7 +46,7 @@ func LogMiddleware() gin.HandlerFunc {
 			stdErr = err.Err
 		}
 		version := c.Request.Header.Get(ClientVersionHeader)
-		log.Info(path,
+		log.Info("cdc open api request",
 			zap.Int("status", c.Writer.Status()),
 			zap.String("method", c.Request.Method),
 			zap.String("path", path),
diff --git a/cdc/api/v1/api.go b/cdc/api/v1/api.go
index 531f4855f..08d4c6caa 100644
--- a/cdc/api/v1/api.go
+++ b/cdc/api/v1/api.go
@@ -738,7 +738,8 @@ func (h *OpenAPI) ListProcessor(c *gin.Context) {
 	for i, info := range infos {
 		resp := &model.ProcessorCommonInfo{
 			Namespace: info.CfID.Namespace,
-			CfID:      info.CfID.ID, CaptureID: info.CaptureID,
+			CfID:      info.CfID.ID,
+			CaptureID: info.CaptureID,
 		}
 		resps[i] = resp
 	}
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
index cfbe3a2a0..d02392273 100644
--- a/cdc/api/v2/model.go
+++ b/cdc/api/v2/model.go
@@ -18,9 +18,9 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	filter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/integrity"
diff --git a/cdc/async/async_initializer.go b/cdc/async/async_initializer.go
new file mode 100644
index 000000000..5bd43405c
--- /dev/null
+++ b/cdc/async/async_initializer.go
@@ -0,0 +1,100 @@
+// Copyright 2024 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 async
+
+import (
+	"context"
+	"sync"
+
+	"github.com/pingcap/log"
+	cdcContext "github.com/pingcap/tiflow/pkg/context"
+	"github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/workerpool"
+	"go.uber.org/atomic"
+	"go.uber.org/zap"
+)
+
+// Initializer is a helper struct to initialize a changefeed asynchronously.
+type Initializer struct {
+	// state related fields
+	initialized  *atomic.Bool
+	initializing *atomic.Bool
+	initError    *atomic.Error
+
+	// func to cancel the changefeed initialization
+	cancelInitialize context.CancelFunc
+	initialWaitGroup sync.WaitGroup
+}
+
+// NewInitializer creates a new initializer.
+func NewInitializer() *Initializer {
+	return &Initializer{
+		initialized:  atomic.NewBool(false),
+		initializing: atomic.NewBool(false),
+		initError:    atomic.NewError(nil),
+	}
+}
+
+// TryInitialize tries to initialize the module asynchronously.
+// returns true if the module is already initialized or initialized successfully.
+// returns false if the module is initializing or failed to initialize.
+// returns error if the module failed to initialize.
+// It will only initialize the module once.
+// It's not thread-safe.
+func (initializer *Initializer) TryInitialize(ctx cdcContext.Context,
+	initFunc func(ctx cdcContext.Context) error,
+	pool workerpool.AsyncPool,
+) (bool, error) {
+	if initializer.initialized.Load() {
+		return true, nil
+	}
+	if initializer.initializing.CompareAndSwap(false, true) {
+		initialCtx, cancelInitialize := cdcContext.WithCancel(ctx)
+		initializer.initialWaitGroup.Add(1)
+		initializer.cancelInitialize = cancelInitialize
+		log.Info("submit async initializer task to the worker pool")
+		err := pool.Go(initialCtx, func() {
+			defer initializer.initialWaitGroup.Done()
+			if err := initFunc(initialCtx); err != nil {
+				initializer.initError.Store(errors.Trace(err))
+			} else {
+				initializer.initialized.Store(true)
+			}
+		})
+		if err != nil {
+			log.Error("failed to submit async initializer task to the worker pool", zap.Error(err))
+			initializer.initialWaitGroup.Done()
+			return false, errors.Trace(err)
+		}
+	}
+	if initializer.initError.Load() != nil {
+		return false, errors.Trace(initializer.initError.Load())
+	}
+	return initializer.initialized.Load(), nil
+}
+
+// Terminate terminates the initializer.
+// It will cancel the initialization if it is initializing. and wait for the initialization to finish.
+// It's not thread-safe.
+func (initializer *Initializer) Terminate() {
+	if initializer.initializing.Load() {
+		if initializer.cancelInitialize != nil {
+			initializer.cancelInitialize()
+		}
+		initializer.initialWaitGroup.Wait()
+	}
+	initializer.initializing.Store(false)
+	initializer.initialized.Store(false)
+	initializer.initError.Store(nil)
+}
diff --git a/cdc/capture/capture.go b/cdc/capture/capture.go
index 44bc2f8f8..e22bb1ee4 100644
--- a/cdc/capture/capture.go
+++ b/cdc/capture/capture.go
@@ -40,6 +40,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/upstream"
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/pingcap/tiflow/pkg/version"
+	"github.com/pingcap/tiflow/pkg/workerpool"
 	pd "github.com/tikv/pd/client"
 	"go.etcd.io/etcd/client/v3/concurrency"
 	"go.etcd.io/etcd/server/v3/mvcc"
@@ -48,7 +49,11 @@ import (
 	"golang.org/x/time/rate"
 )
 
-const cleanMetaDuration = 10 * time.Second
+const (
+	cleanMetaDuration = 10 * time.Second
+	// changefeedAsyncInitWorkerCount is the size of the worker pool for changefeed initialization processing.
+	changefeedAsyncInitWorkerCount = 8
+)
 
 // Capture represents a Capture server, it monitors the changefeed
 // information in etcd and schedules Task on it.
@@ -97,6 +102,9 @@ type captureImpl struct {
 
 	sortEngineFactory *factory.SortEngineFactory
 
+	// ChangefeedThreadPool is the thread pool for changefeed initialization
+	ChangefeedThreadPool workerpool.AsyncPool
+
 	// MessageServer is the receiver of the messages from the other nodes.
 	// It should be recreated each time the capture is restarted.
 	MessageServer *p2p.MessageServer
@@ -263,7 +271,7 @@ func (c *captureImpl) reset(ctx context.Context) error {
 	messageClientConfig.AdvertisedAddr = advertiseAddr
 
 	c.MessageRouter = p2p.NewMessageRouterWithLocalClient(c.info.ID, c.config.Security, messageClientConfig)
-
+	c.ChangefeedThreadPool = workerpool.NewDefaultAsyncPool(changefeedAsyncInitWorkerCount)
 	log.Info("capture initialized", zap.Any("capture", c.info))
 	return nil
 }
@@ -342,11 +350,12 @@ func (c *captureImpl) run(stdCtx context.Context) error {
 	stdCtx, cancel := context.WithCancel(stdCtx)
 
 	ctx := cdcContext.NewContext(stdCtx, &cdcContext.GlobalVars{
-		CaptureInfo:       c.info,
-		EtcdClient:        c.EtcdClient,
-		MessageServer:     c.MessageServer,
-		MessageRouter:     c.MessageRouter,
-		SortEngineFactory: c.sortEngineFactory,
+		CaptureInfo:          c.info,
+		EtcdClient:           c.EtcdClient,
+		MessageServer:        c.MessageServer,
+		MessageRouter:        c.MessageRouter,
+		SortEngineFactory:    c.sortEngineFactory,
+		ChangefeedThreadPool: c.ChangefeedThreadPool,
 	})
 	g.Go(func() error {
 		// when the campaignOwner returns an error, it means that the owner throws
@@ -398,6 +407,14 @@ func (c *captureImpl) run(stdCtx context.Context) error {
 		return c.MessageServer.Run(ctx, c.MessageRouter.GetLocalChannel())
 	})
 
+	poolCtx, cancelPool := context.WithCancel(stdCtx)
+	defer func() {
+		cancelPool()
+		log.Info("workerpool exited", zap.Error(err))
+	}()
+	g.Go(func() error {
+		return c.ChangefeedThreadPool.Run(poolCtx)
+	})
 	return errors.Trace(g.Wait())
 }
 
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
index 8950a62ce..73ac938ca 100755
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/pkg/errno"
+	"github.com/pingcap/tiflow/cdc/async"
 	"github.com/pingcap/tiflow/cdc/entry"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/puller"
@@ -97,7 +98,9 @@ type changefeed struct {
 	wg sync.WaitGroup
 
 	// state related fields
-	initialized bool
+	initialized *atomic.Bool
+	initializer *async.Initializer
+
 	// isRemoved is true if the changefeed is removed,
 	// which means it will be removed from memory forever
 	isRemoved bool
@@ -178,9 +181,11 @@ func newChangefeed(
 		newDDLPuller:          puller.NewDDLPuller,
 		newSink:               newDDLSink,
 		newDownstreamObserver: observer.NewObserver,
+		initialized:           atomic.NewBool(false),
 	}
 	c.newScheduler = newScheduler
 	c.cfg = cfg
+	c.initializer = async.NewInitializer()
 	return c
 }
 
@@ -249,7 +254,7 @@ func (c *changefeed) Tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 	err := c.tick(ctx, captures)
 
 	// The tick duration is recorded only if changefeed has completed initialization
-	if c.initialized {
+	if c.initialized.Load() {
 		costTime := time.Since(startTime)
 		if costTime > changefeedLogsWarnDuration {
 			log.Warn("changefeed tick took too long",
@@ -339,9 +344,32 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 	if !c.preflightCheck(captures) {
 		return nil
 	}
+	if c.state.Status == nil {
+		// If `c.state.Status` is nil it means the changefeed struct is just created, it needs to
+		//  1. use startTs as checkpointTs and resolvedTs, if it's a new created changefeed; or
+		//  2. load checkpointTs and resolvedTs from etcd, if it's an existing changefeed.
+		// And then it can continue to initialize.
+		return nil
+	}
 
-	if err := c.initialize(ctx); err != nil {
-		return errors.Trace(err)
+	if !c.initialized.Load() {
+		info, err := c.state.Info.Clone()
+		if err != nil {
+			return errors.Trace(err)
+		}
+		checkpointTs := c.state.Status.CheckpointTs
+		minTableBarrierTs := c.state.Status.MinTableBarrierTs
+		initialized, err := c.initializer.TryInitialize(ctx,
+			func(ctx cdcContext.Context) error {
+				return c.initialize(ctx, info, checkpointTs, minTableBarrierTs)
+			},
+			ctx.GlobalVars().ChangefeedThreadPool)
+		if err != nil {
+			return errors.Trace(err)
+		}
+		if !initialized {
+			return nil
+		}
 	}
 
 	select {
@@ -462,12 +490,12 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 	return nil
 }
 
-func (c *changefeed) initialize(ctx cdcContext.Context) (err error) {
-	if c.initialized || c.state.Status == nil {
-		// If `c.state.Status` is nil it means the changefeed struct is just created, it needs to
-		//  1. use startTs as checkpointTs and resolvedTs, if it's a new created changefeed; or
-		//  2. load checkpointTs and resolvedTs from etcd, if it's an existing changefeed.
-		// And then it can continue to initialize.
+func (c *changefeed) initialize(ctx cdcContext.Context,
+	info *model.ChangeFeedInfo,
+	checkpointTs uint64,
+	minTableBarrierTs uint64,
+) (err error) {
+	if c.initialized.Load() {
 		return nil
 	}
 	c.isReleased = false
@@ -492,13 +520,10 @@ LOOP2:
 		}
 	}
 
-	checkpointTs := c.state.Status.CheckpointTs
 	if c.resolvedTs == 0 {
 		c.resolvedTs = checkpointTs
 	}
 
-	minTableBarrierTs := c.state.Status.MinTableBarrierTs
-
 	failpoint.Inject("NewChangefeedNoRetryError", func() {
 		failpoint.Return(cerror.ErrStartTsBeforeGC.GenWithStackByArgs(checkpointTs-300, checkpointTs))
 	})
@@ -506,7 +531,7 @@ LOOP2:
 		failpoint.Return(errors.New("failpoint injected retriable error"))
 	})
 
-	if c.state.Info.Config.CheckGCSafePoint {
+	if info.Config.CheckGCSafePoint {
 		// Check TiDB GC safepoint does not exceed the checkpoint.
 		//
 		// We update TTL to 10 minutes,
@@ -560,19 +585,19 @@ LOOP2:
 	}
 
 	c.barriers = newBarriers()
-	if util.GetOrZero(c.state.Info.Config.EnableSyncPoint) {
+	if util.GetOrZero(info.Config.EnableSyncPoint) {
 		c.barriers.Update(syncPointBarrier, c.resolvedTs)
 	}
-	c.barriers.Update(finishBarrier, c.state.Info.GetTargetTs())
+	c.barriers.Update(finishBarrier, info.GetTargetTs())
 
-	filter, err := pfilter.NewFilter(c.state.Info.Config, "")
+	filter, err := pfilter.NewFilter(info.Config, "")
 	if err != nil {
 		return errors.Trace(err)
 	}
 	c.schema, err = entry.NewSchemaStorage(
 		c.upstream.KVStorage,
 		ddlStartTs,
-		c.state.Info.Config.ForceReplicate,
+		info.Config.ForceReplicate,
 		c.id,
 		util.RoleOwner,
 		filter)
@@ -587,14 +612,14 @@ LOOP2:
 	if err != nil {
 		return errors.Trace(err)
 	}
-	c.state.Info.Config.Sink.TiDBSourceID = sourceID
+	info.Config.Sink.TiDBSourceID = sourceID
 	log.Info("set source id",
 		zap.Uint64("sourceID", sourceID),
 		zap.String("namespace", c.id.Namespace),
 		zap.String("changefeed", c.id.ID),
 	)
 
-	c.ddlSink = c.newSink(c.id, c.state.Info, ctx.Throw, func(err error) {
+	c.ddlSink = c.newSink(c.id, info, ctx.Throw, func(err error) {
 		select {
 		case <-ctx.Done():
 		case c.warningCh <- err:
@@ -603,7 +628,7 @@ LOOP2:
 	c.ddlSink.run(cancelCtx)
 
 	c.ddlPuller, err = c.newDDLPuller(cancelCtx,
-		c.state.Info.Config,
+		info.Config,
 		c.upstream, ddlStartTs,
 		c.id,
 		c.schema,
@@ -618,13 +643,13 @@ LOOP2:
 		ctx.Throw(c.ddlPuller.Run(cancelCtx))
 	}()
 
-	c.downstreamObserver, err = c.newDownstreamObserver(ctx, c.id, c.state.Info.SinkURI, c.state.Info.Config)
+	c.downstreamObserver, err = c.newDownstreamObserver(ctx, c.id, info.SinkURI, info.Config)
 	if err != nil {
 		return err
 	}
 	c.observerLastTick = atomic.NewTime(time.Time{})
 
-	c.redoDDLMgr = redo.NewDDLManager(c.id, c.state.Info.Config.Consistent, ddlStartTs)
+	c.redoDDLMgr = redo.NewDDLManager(c.id, info.Config.Consistent, ddlStartTs)
 	if c.redoDDLMgr.Enabled() {
 		c.wg.Add(1)
 		go func() {
@@ -633,7 +658,7 @@ LOOP2:
 		}()
 	}
 
-	c.redoMetaMgr = redo.NewMetaManager(c.id, c.state.Info.Config.Consistent, checkpointTs)
+	c.redoMetaMgr = redo.NewMetaManager(c.id, info.Config.Consistent, checkpointTs)
 	if c.redoMetaMgr.Enabled() {
 		c.wg.Add(1)
 		go func() {
@@ -645,7 +670,7 @@ LOOP2:
 		zap.String("namespace", c.id.Namespace),
 		zap.String("changefeed", c.id.ID))
 
-	downstreamType, err := c.state.Info.DownstreamType()
+	downstreamType, err := info.DownstreamType()
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -653,7 +678,7 @@ LOOP2:
 	c.ddlManager = newDDLManager(
 		c.id,
 		ddlStartTs,
-		c.state.Status.CheckpointTs,
+		checkpointTs,
 		c.ddlSink,
 		filter,
 		c.ddlPuller,
@@ -661,13 +686,13 @@ LOOP2:
 		c.redoDDLMgr,
 		c.redoMetaMgr,
 		downstreamType,
-		util.GetOrZero(c.state.Info.Config.BDRMode),
+		util.GetOrZero(info.Config.BDRMode),
 	)
 
 	// create scheduler
 	cfg := *c.cfg
-	cfg.ChangefeedSettings = c.state.Info.Config.Scheduler
-	epoch := c.state.Info.Epoch
+	cfg.ChangefeedSettings = info.Config.Scheduler
+	epoch := info.Epoch
 	c.scheduler, err = c.newScheduler(ctx, c.upstream, epoch, &cfg, c.redoMetaMgr)
 	if err != nil {
 		return errors.Trace(err)
@@ -675,8 +700,8 @@ LOOP2:
 
 	c.initMetrics()
 
-	c.initialized = true
-	c.metricsChangefeedCreateTimeGuage.Set(float64(oracle.GetPhysical(c.state.Info.CreateTime)))
+	c.initialized.Store(true)
+	c.metricsChangefeedCreateTimeGuage.Set(float64(oracle.GetPhysical(info.CreateTime)))
 	c.metricsChangefeedRestartTimeGauge.Set(float64(oracle.GetPhysical(time.Now())))
 	log.Info("changefeed initialized",
 		zap.String("namespace", c.state.ID.Namespace),
@@ -684,7 +709,7 @@ LOOP2:
 		zap.Uint64("changefeedEpoch", epoch),
 		zap.Uint64("checkpointTs", checkpointTs),
 		zap.Uint64("resolvedTs", c.resolvedTs),
-		zap.String("info", c.state.Info.String()))
+		zap.String("info", info.String()))
 
 	return nil
 }
@@ -718,6 +743,7 @@ func (c *changefeed) initMetrics() {
 
 // releaseResources is idempotent.
 func (c *changefeed) releaseResources(ctx cdcContext.Context) {
+	c.initializer.Terminate()
 	c.cleanupMetrics()
 	if c.isReleased {
 		return
@@ -727,7 +753,9 @@ func (c *changefeed) releaseResources(ctx cdcContext.Context) {
 	c.cleanupRedoManager(ctx)
 	c.cleanupChangefeedServiceGCSafePoints(ctx)
 
-	c.cancel()
+	if c.cancel != nil {
+		c.cancel()
+	}
 	c.cancel = func() {}
 
 	if c.ddlPuller != nil {
@@ -758,7 +786,7 @@ func (c *changefeed) releaseResources(ctx cdcContext.Context) {
 
 	c.schema = nil
 	c.barriers = nil
-	c.initialized = false
+	c.initialized.Store(false)
 	c.isReleased = true
 
 	log.Info("changefeed closed",
diff --git a/cdc/processor/processor.go b/cdc/processor/processor.go
index 7b44d3c07..8098bb324 100644
--- a/cdc/processor/processor.go
+++ b/cdc/processor/processor.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
 	"github.com/pingcap/log"
+	"github.com/pingcap/tiflow/cdc/async"
 	"github.com/pingcap/tiflow/cdc/entry"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/processor/sinkmanager"
@@ -46,6 +47,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/tikv/client-go/v2/oracle"
+	"go.uber.org/atomic"
 	"go.uber.org/zap"
 	"golang.org/x/sync/errgroup"
 )
@@ -89,7 +91,8 @@ type processor struct {
 
 	sinkManager component[*sinkmanager.SinkManager]
 
-	initialized bool
+	initialized *atomic.Bool
+	initializer *async.Initializer
 
 	lazyInit func(ctx cdcContext.Context) error
 	newAgent func(
@@ -431,6 +434,8 @@ func NewProcessor(
 		latestInfo:      info,
 		latestStatus:    status,
 
+		initialized: atomic.NewBool(false),
+
 		ownerCaptureInfoClient: ownerCaptureInfoClient,
 
 		metricSyncTableNumGauge: syncTableNumGauge.
@@ -447,6 +452,7 @@ func NewProcessor(
 	p.lazyInit = p.lazyInitImpl
 	p.newAgent = p.newAgentImpl
 	p.cfg = cfg
+	p.initializer = async.NewInitializer()
 	return p
 }
 
@@ -495,6 +501,16 @@ func (p *processor) Tick(
 	ctx cdcContext.Context,
 	info *model.ChangeFeedInfo, status *model.ChangeFeedStatus,
 ) (error, error) {
+	if !p.initialized.Load() {
+		initialized, err := p.initializer.TryInitialize(ctx, p.lazyInit, ctx.GlobalVars().ChangefeedThreadPool)
+		if err != nil {
+			return errors.Trace(err), nil
+		}
+		if !initialized {
+			return nil, nil
+		}
+	}
+
 	p.latestInfo = info
 	p.latestStatus = status
 
@@ -519,8 +535,7 @@ func (p *processor) Tick(
 		return nil, nil
 	}
 	startTime := time.Now()
-	warning := p.handleWarnings()
-	err := p.tick(ctx)
+	err, warning := p.tick(ctx)
 	costTime := time.Since(startTime)
 	if costTime > processorLogsWarnDuration {
 		log.Warn("processor tick took too long",
@@ -555,17 +570,15 @@ func (p *processor) handleWarnings() error {
 	return err
 }
 
-func (p *processor) tick(ctx cdcContext.Context) error {
+func (p *processor) tick(ctx cdcContext.Context) (error, error) {
+	warning := p.handleWarnings()
 	if err := p.handleErrorCh(); err != nil {
-		return errors.Trace(err)
-	}
-	if err := p.lazyInit(ctx); err != nil {
-		return errors.Trace(err)
+		return errors.Trace(err), warning
 	}
 
 	barrier, err := p.agent.Tick(ctx)
 	if err != nil {
-		return errors.Trace(err)
+		return errors.Trace(err), warning
 	}
 
 	if barrier != nil && barrier.GlobalBarrierTs != 0 {
@@ -573,12 +586,12 @@ func (p *processor) tick(ctx cdcContext.Context) error {
 	}
 	p.doGCSchemaStorage()
 
-	return nil
+	return nil, warning
 }
 
 // lazyInitImpl create Filter, SchemaStorage, Mounter instances at the first tick.
 func (p *processor) lazyInitImpl(etcdCtx cdcContext.Context) (err error) {
-	if p.initialized {
+	if p.initialized.Load() {
 		return nil
 	}
 
@@ -594,7 +607,11 @@ func (p *processor) lazyInitImpl(etcdCtx cdcContext.Context) (err error) {
 	if err != nil {
 		return errors.Trace(err)
 	}
-	p.filter, err = filter.NewFilter(p.latestInfo.Config, util.GetTimeZoneName(tz))
+
+	// Clone the config to avoid data race
+	cfConfig := p.latestInfo.Config.Clone()
+
+	p.filter, err = filter.NewFilter(cfConfig, util.GetTimeZoneName(tz))
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -607,8 +624,8 @@ func (p *processor) lazyInitImpl(etcdCtx cdcContext.Context) (err error) {
 	p.ddlHandler.spawn(prcCtx)
 
 	p.mg.r = entry.NewMounterGroup(p.ddlHandler.r.schemaStorage,
-		p.latestInfo.Config.Mounter.WorkerNum,
-		p.filter, tz, p.changefeedID, p.latestInfo.Config.Integrity)
+		cfConfig.Mounter.WorkerNum,
+		p.filter, tz, p.changefeedID, cfConfig.Integrity)
 	p.mg.name = "MounterGroup"
 	p.mg.changefeedID = p.changefeedID
 	p.mg.spawn(prcCtx)
@@ -617,9 +634,11 @@ func (p *processor) lazyInitImpl(etcdCtx cdcContext.Context) (err error) {
 	if err != nil {
 		return errors.Trace(err)
 	}
-	p.latestInfo.Config.Sink.TiDBSourceID = sourceID
+	cfConfig.Sink.TiDBSourceID = sourceID
+	log.Info("get sourceID from PD", zap.Uint64("sourceID", sourceID),
+		zap.Stringer("changefeedID", p.changefeedID))
 
-	p.redo.r = redo.NewDMLManager(p.changefeedID, p.latestInfo.Config.Consistent)
+	p.redo.r = redo.NewDMLManager(p.changefeedID, cfConfig.Consistent)
 	p.redo.name = "RedoManager"
 	p.redo.changefeedID = p.changefeedID
 	p.redo.spawn(prcCtx)
@@ -639,14 +658,14 @@ func (p *processor) lazyInitImpl(etcdCtx cdcContext.Context) (err error) {
 	}
 	p.sourceManager.r = sourcemanager.New(
 		p.changefeedID, p.upstream, p.mg.r,
-		sortEngine, util.GetOrZero(p.latestInfo.Config.BDRMode),
+		sortEngine, util.GetOrZero(cfConfig.BDRMode),
 		pullerSafeModeAtStart)
 	p.sourceManager.name = "SourceManager"
 	p.sourceManager.changefeedID = p.changefeedID
 	p.sourceManager.spawn(prcCtx)
 
 	p.sinkManager.r = sinkmanager.New(
-		p.changefeedID, p.latestInfo, p.upstream,
+		p.changefeedID, p.latestInfo.SinkURI, cfConfig, p.upstream,
 		p.ddlHandler.r.schemaStorage, p.redo.r, p.sourceManager.r)
 	p.sinkManager.name = "SinkManager"
 	p.sinkManager.changefeedID = p.changefeedID
@@ -659,7 +678,7 @@ func (p *processor) lazyInitImpl(etcdCtx cdcContext.Context) (err error) {
 		return err
 	}
 
-	p.initialized = true
+	p.initialized.Store(true)
 	log.Info("processor initialized",
 		zap.String("capture", p.captureInfo.ID),
 		zap.String("namespace", p.changefeedID.Namespace),
@@ -836,7 +855,7 @@ func (p *processor) doGCSchemaStorage() {
 func (p *processor) refreshMetrics() {
 	// Before the processor is initialized, we should not refresh metrics.
 	// Otherwise, it will cause panic.
-	if !p.initialized {
+	if !p.initialized.Load() {
 		return
 	}
 	p.metricSyncTableNumGauge.Set(float64(p.sinkManager.r.GetAllCurrentTableSpansCount()))
@@ -847,7 +866,7 @@ func (p *processor) Close() error {
 	log.Info("processor closing ...",
 		zap.String("namespace", p.changefeedID.Namespace),
 		zap.String("changefeed", p.changefeedID.ID))
-
+	p.initializer.Terminate()
 	// clean up metrics first to avoid some metrics are not cleaned up
 	// when error occurs during closing the processor
 	p.cleanupMetrics()
@@ -924,6 +943,10 @@ func (p *processor) cleanupMetrics() {
 
 // WriteDebugInfo write the debug info to Writer
 func (p *processor) WriteDebugInfo(w io.Writer) error {
+	if !p.initialized.Load() {
+		fmt.Fprintln(w, "processor is not initialized")
+		return nil
+	}
 	fmt.Fprintf(w, "%+v\n%+v\n", *p.latestInfo, *p.latestStatus)
 	spans := p.sinkManager.r.GetAllCurrentTableSpans()
 	for _, span := range spans {
diff --git a/cdc/processor/sinkmanager/manager.go b/cdc/processor/sinkmanager/manager.go
index 4e3a14042..2ce276a47 100644
--- a/cdc/processor/sinkmanager/manager.go
+++ b/cdc/processor/sinkmanager/manager.go
@@ -34,6 +34,7 @@ import (
 	tablesinkmetrics "github.com/pingcap/tiflow/cdc/sink/metrics/tablesink"
 	"github.com/pingcap/tiflow/cdc/sink/tablesink"
 	"github.com/pingcap/tiflow/pkg/config"
+	pconfig "github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/retry"
 	"github.com/pingcap/tiflow/pkg/spanz"
@@ -66,7 +67,8 @@ type TableStats struct {
 type SinkManager struct {
 	changefeedID model.ChangeFeedID
 
-	changefeedInfo *model.ChangeFeedInfo
+	sinkURI string
+	config  *pconfig.ReplicaConfig
 
 	// up is the upstream and used to get the current pd time.
 	up *upstream.Upstream
@@ -137,19 +139,20 @@ type SinkManager struct {
 // New creates a new sink manager.
 func New(
 	changefeedID model.ChangeFeedID,
-	changefeedInfo *model.ChangeFeedInfo,
+	sinkURI string,
+	config *pconfig.ReplicaConfig,
 	up *upstream.Upstream,
 	schemaStorage entry.SchemaStorage,
 	redoDMLMgr redo.DMLManager,
 	sourceManager *sourcemanager.SourceManager,
 ) *SinkManager {
 	m := &SinkManager{
-		changefeedID:   changefeedID,
-		changefeedInfo: changefeedInfo,
-		up:             up,
-		schemaStorage:  schemaStorage,
-		sourceManager:  sourceManager,
-
+		changefeedID:        changefeedID,
+		up:                  up,
+		schemaStorage:       schemaStorage,
+		sourceManager:       sourceManager,
+		sinkURI:             sinkURI,
+		config:              config,
 		sinkProgressHeap:    newTableProgresses(),
 		sinkWorkers:         make([]*sinkWorker, 0, sinkWorkerNum),
 		sinkTaskChan:        make(chan *sinkTask),
@@ -163,7 +166,7 @@ func New(
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 	}
 
-	totalQuota := changefeedInfo.Config.MemoryQuota
+	totalQuota := config.MemoryQuota
 	if redoDMLMgr != nil && redoDMLMgr.Enabled() {
 		m.redoDMLMgr = redoDMLMgr
 		m.redoProgressHeap = newTableProgresses()
@@ -171,9 +174,9 @@ func New(
 		m.redoTaskChan = make(chan *redoTask)
 		m.redoWorkerAvailable = make(chan struct{}, 1)
 
-		consistentMemoryUsage := changefeedInfo.Config.Consistent.MemoryUsage
+		consistentMemoryUsage := m.config.Consistent.MemoryUsage
 		if consistentMemoryUsage == nil {
-			consistentMemoryUsage = config.GetDefaultReplicaConfig().Consistent.MemoryUsage
+			consistentMemoryUsage = pconfig.GetDefaultReplicaConfig().Consistent.MemoryUsage
 		}
 
 		redoQuota := totalQuota * consistentMemoryUsage.MemoryQuotaPercentage / 100
@@ -208,7 +211,7 @@ func (m *SinkManager) Run(ctx context.Context, warnings ...chan<- error) (err er
 			zap.Error(err))
 	}()
 
-	splitTxn := util.GetOrZero(m.changefeedInfo.Config.Sink.TxnAtomicity).ShouldSplitTxn()
+	splitTxn := util.GetOrZero(m.config.Sink.TxnAtomicity).ShouldSplitTxn()
 
 	gcErrors := make(chan error, 16)
 	sinkErrors := make(chan error, 16)
@@ -334,8 +337,8 @@ func (m *SinkManager) needsStuckCheck() bool {
 func (m *SinkManager) initSinkFactory() (chan error, uint64) {
 	m.sinkFactory.Lock()
 	defer m.sinkFactory.Unlock()
-	uri := m.changefeedInfo.SinkURI
-	cfg := m.changefeedInfo.Config
+	uri := m.sinkURI
+	cfg := m.config
 
 	if m.sinkFactory.f != nil {
 		return m.sinkFactory.errors, m.sinkFactory.version
@@ -1022,7 +1025,7 @@ func (m *SinkManager) GetTableStats(span tablepb.Span) TableStats {
 	m.sinkMemQuota.Release(span, checkpointTs)
 	m.redoMemQuota.Release(span, checkpointTs)
 
-	advanceTimeoutInSec := util.GetOrZero(m.changefeedInfo.Config.Sink.AdvanceTimeoutInSec)
+	advanceTimeoutInSec := util.GetOrZero(m.config.Sink.AdvanceTimeoutInSec)
 	if advanceTimeoutInSec <= 0 {
 		advanceTimeoutInSec = config.DefaultAdvanceTimeoutInSec
 	}
diff --git a/cdc/redo/reader/reader.go b/cdc/redo/reader/reader.go
index 25c5ea8cb..8f7a7a61b 100644
--- a/cdc/redo/reader/reader.go
+++ b/cdc/redo/reader/reader.go
@@ -364,6 +364,7 @@ func (h logHeap) Len() int {
 }
 
 func (h logHeap) Less(i, j int) bool {
+	// we separate ddl and dml, so we only need to compare dml with dml, and ddl with ddl.
 	if h[i].data.Type == model.RedoLogTypeDDL {
 		if h[i].data.RedoDDL.DDL == nil {
 			return true
@@ -381,10 +382,19 @@ func (h logHeap) Less(i, j int) bool {
 		return false
 	}
 
-	if h[i].data.RedoRow.Row.CommitTs == h[j].data.RedoRow.Row.CommitTs &&
-		h[i].data.RedoRow.Row.StartTs < h[j].data.RedoRow.Row.StartTs {
-		return true
+	if h[i].data.RedoRow.Row.CommitTs == h[j].data.RedoRow.Row.CommitTs {
+		if h[i].data.RedoRow.Row.StartTs != h[j].data.RedoRow.Row.StartTs {
+			return h[i].data.RedoRow.Row.StartTs < h[j].data.RedoRow.Row.StartTs
+		}
+		// in the same txn, we need to sort by delete/update/insert order
+		if h[i].data.RedoRow.Row.IsDelete() {
+			return true
+		} else if h[i].data.RedoRow.Row.IsUpdate() {
+			return !h[j].data.RedoRow.Row.IsDelete()
+		}
+		return false
 	}
+
 	return h[i].data.RedoRow.Row.CommitTs < h[j].data.RedoRow.Row.CommitTs
 }
 
diff --git a/cdc/redo/writer/memory/encoding_worker.go b/cdc/redo/writer/memory/encoding_worker.go
index 07e9824d1..9b0cc6a86 100644
--- a/cdc/redo/writer/memory/encoding_worker.go
+++ b/cdc/redo/writer/memory/encoding_worker.go
@@ -104,7 +104,7 @@ type encodingWorkerGroup struct {
 	workerNum  int
 	nextWorker atomic.Uint64
 
-	closed chan struct{}
+	closed chan error
 }
 
 func newEncodingWorkerGroup(cfg *writer.LogWriterConfig) *encodingWorkerGroup {
@@ -121,19 +121,20 @@ func newEncodingWorkerGroup(cfg *writer.LogWriterConfig) *encodingWorkerGroup {
 		inputChs:   inputChs,
 		outputCh:   make(chan *polymorphicRedoEvent, redo.DefaultEncodingOutputChanSize),
 		workerNum:  workerNum,
-		closed:     make(chan struct{}),
+		closed:     make(chan error, 1),
 	}
 }
 
 func (e *encodingWorkerGroup) Run(ctx context.Context) (err error) {
 	defer func() {
-		close(e.closed)
+		log.Warn("redo encoding workers closed",
+			zap.String("namespace", e.changefeed.Namespace),
+			zap.String("changefeed", e.changefeed.ID),
+			zap.Error(err))
 		if err != nil && errors.Cause(err) != context.Canceled {
-			log.Warn("redo fileWorkerGroup closed with error",
-				zap.String("namespace", e.changefeed.Namespace),
-				zap.String("changefeed", e.changefeed.ID),
-				zap.Error(err))
+			e.closed <- err
 		}
+		close(e.closed)
 	}()
 	eg, egCtx := errgroup.WithContext(ctx)
 	for i := 0; i < e.workerNum; i++ {
@@ -183,8 +184,8 @@ func (e *encodingWorkerGroup) input(
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case <-e.closed:
-		return errors.ErrRedoWriterStopped.GenWithStackByArgs("encoding worker is closed")
+	case err := <-e.closed:
+		return errors.WrapError(errors.ErrRedoWriterStopped, err, "encoding worker is closed")
 	case e.inputChs[idx] <- event:
 		return nil
 	}
@@ -196,8 +197,8 @@ func (e *encodingWorkerGroup) output(
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case <-e.closed:
-		return errors.ErrRedoWriterStopped.GenWithStackByArgs("encoding worker is closed")
+	case err := <-e.closed:
+		return errors.WrapError(errors.ErrRedoWriterStopped, err, "encoding worker is closed")
 	case e.outputCh <- event:
 		return nil
 	}
@@ -221,8 +222,8 @@ func (e *encodingWorkerGroup) FlushAll(ctx context.Context) error {
 	select {
 	case <-ctx.Done():
 		return ctx.Err()
-	case <-e.closed:
-		return errors.ErrRedoWriterStopped.GenWithStackByArgs("encoding worker is closed")
+	case err := <-e.closed:
+		return errors.WrapError(errors.ErrRedoWriterStopped, err, "encoding worker is closed")
 	case <-flushCh:
 	}
 	return nil
@@ -245,8 +246,8 @@ func (e *encodingWorkerGroup) broadcastAndWaitEncoding(ctx context.Context) erro
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
-		case <-e.closed:
-			return errors.ErrRedoWriterStopped.GenWithStackByArgs("encoding worker is closed")
+		case err := <-e.closed:
+			return errors.WrapError(errors.ErrRedoWriterStopped, err, "encoding worker is closed")
 		case <-ch:
 		}
 	}
diff --git a/cdc/redo/writer/memory/file_worker.go b/cdc/redo/writer/memory/file_worker.go
index d566ddf66..2e8fe622d 100644
--- a/cdc/redo/writer/memory/file_worker.go
+++ b/cdc/redo/writer/memory/file_worker.go
@@ -155,12 +155,10 @@ func (f *fileWorkerGroup) Run(
 ) (err error) {
 	defer func() {
 		f.close()
-		if err != nil && errors.Cause(err) != context.Canceled {
-			log.Warn("redo file workers closed with error",
-				zap.String("namespace", f.cfg.ChangeFeedID.Namespace),
-				zap.String("changefeed", f.cfg.ChangeFeedID.ID),
-				zap.Error(err))
-		}
+		log.Warn("redo file workers closed",
+			zap.String("namespace", f.cfg.ChangeFeedID.Namespace),
+			zap.String("changefeed", f.cfg.ChangeFeedID.ID),
+			zap.Error(err))
 	}()
 
 	eg, egCtx := errgroup.WithContext(ctx)
diff --git a/cdc/scheduler/internal/v3/coordinator.go b/cdc/scheduler/internal/v3/coordinator.go
index aaaa77300..853759850 100644
--- a/cdc/scheduler/internal/v3/coordinator.go
+++ b/cdc/scheduler/internal/v3/coordinator.go
@@ -32,7 +32,6 @@ import (
 	"github.com/pingcap/tiflow/cdc/scheduler/internal/v3/transport"
 	"github.com/pingcap/tiflow/cdc/scheduler/schedulepb"
 	"github.com/pingcap/tiflow/pkg/config"
-	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/p2p"
 	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/spanz"
@@ -191,9 +190,8 @@ func (c *coordinator) DrainCapture(target model.CaptureID) (int, error) {
 			zap.String("target", target),
 			zap.String("namespace", c.changefeedID.Namespace),
 			zap.String("changefeed", c.changefeedID.ID))
-		// return false to let client retry.
-		return 0, cerror.ErrSchedulerRequestFailed.
-			GenWithStack("not all captures initialized")
+		// return count 1 to let client retry.
+		return 1, nil
 	}
 
 	var count int
diff --git a/cmd/dm-syncer/config.go b/cmd/dm-syncer/config.go
index 35bd99771..020ee1ab9 100644
--- a/cmd/dm-syncer/config.go
+++ b/cmd/dm-syncer/config.go
@@ -21,12 +21,12 @@ import (
 	"github.com/BurntSushi/toml"
 	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/pingcap/errors"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
 	"github.com/pingcap/tiflow/dm/config"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/pkg/log"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/version"
 )
 
diff --git a/dm/config/source_config.go b/dm/config/source_config.go
index 584eb1b5d..a6cf46df8 100644
--- a/dm/config/source_config.go
+++ b/dm/config/source_config.go
@@ -27,7 +27,6 @@ import (
 
 	"github.com/BurntSushi/toml"
 	"github.com/go-mysql-org/go-mysql/mysql"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
 	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
@@ -35,6 +34,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v2"
 )
diff --git a/dm/config/subtask.go b/dm/config/subtask.go
index 5aff14117..a98429d7d 100644
--- a/dm/config/subtask.go
+++ b/dm/config/subtask.go
@@ -28,8 +28,6 @@ import (
 	"time"
 
 	"github.com/BurntSushi/toml"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	extstorage "github.com/pingcap/tidb/br/pkg/storage"
 	"github.com/pingcap/tidb/pkg/util/dbutil"
 	"github.com/pingcap/tidb/pkg/util/filter"
@@ -41,6 +39,8 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	"github.com/pingcap/tiflow/engine/pkg/promutil"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
+	"github.com/pingcap/tiflow/pkg/column-mapping"
 	"github.com/pingcap/tiflow/pkg/version"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
diff --git a/dm/config/task.go b/dm/config/task.go
index 67860d91a..031fece84 100644
--- a/dm/config/task.go
+++ b/dm/config/task.go
@@ -28,8 +28,6 @@ import (
 	"github.com/coreos/go-semver/semver"
 	"github.com/docker/go-units"
 	"github.com/dustin/go-humanize"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	"github.com/pingcap/tidb/pkg/lightning/config"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/util/filter"
@@ -38,6 +36,8 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
+	"github.com/pingcap/tiflow/pkg/column-mapping"
 	"go.uber.org/zap"
 	"gopkg.in/yaml.v2"
 )
diff --git a/dm/config/task_converters.go b/dm/config/task_converters.go
index 960623fe1..306037a28 100644
--- a/dm/config/task_converters.go
+++ b/dm/config/task_converters.go
@@ -17,8 +17,6 @@ import (
 	"fmt"
 	"strings"
 
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
@@ -27,6 +25,8 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/storage"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
+	"github.com/pingcap/tiflow/pkg/column-mapping"
 	"go.uber.org/zap"
 )
 
diff --git a/dm/syncer/ddl.go b/dm/syncer/ddl.go
index cb5f53532..b89a0fcb3 100644
--- a/dm/syncer/ddl.go
+++ b/dm/syncer/ddl.go
@@ -20,7 +20,6 @@ import (
 
 	"github.com/go-mysql-org/go-mysql/replication"
 	"github.com/pingcap/failpoint"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	tidbddl "github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -48,6 +47,7 @@ import (
 	onlineddl "github.com/pingcap/tiflow/dm/syncer/online-ddl-tools"
 	sm "github.com/pingcap/tiflow/dm/syncer/safe-mode"
 	"github.com/pingcap/tiflow/dm/syncer/shardddl"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
 )
diff --git a/dm/syncer/filter.go b/dm/syncer/filter.go
index 799398ca6..11863921d 100644
--- a/dm/syncer/filter.go
+++ b/dm/syncer/filter.go
@@ -15,11 +15,11 @@ package syncer
 
 import (
 	"github.com/go-mysql-org/go-mysql/replication"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	onlineddl "github.com/pingcap/tiflow/dm/syncer/online-ddl-tools"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"go.uber.org/zap"
 )
 
diff --git a/dm/syncer/syncer.go b/dm/syncer/syncer.go
index 8cbadbb62..b55f07389 100644
--- a/dm/syncer/syncer.go
+++ b/dm/syncer/syncer.go
@@ -31,7 +31,6 @@ import (
 	mysql2 "github.com/go-sql-driver/mysql"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	tidbddl "github.com/pingcap/tidb/pkg/ddl"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -68,6 +67,7 @@ import (
 	sm "github.com/pingcap/tiflow/dm/syncer/safe-mode"
 	"github.com/pingcap/tiflow/dm/syncer/shardddl"
 	"github.com/pingcap/tiflow/dm/unit"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/errorutil"
 	"github.com/pingcap/tiflow/pkg/sqlmodel"
 	clientv3 "go.etcd.io/etcd/client/v3"
diff --git a/dm/worker/source_worker.go b/dm/worker/source_worker.go
index 35faf0f54..48886a103 100644
--- a/dm/worker/source_worker.go
+++ b/dm/worker/source_worker.go
@@ -23,7 +23,6 @@ import (
 	"github.com/golang/protobuf/proto"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/dm/config"
 	"github.com/pingcap/tiflow/dm/pb"
 	"github.com/pingcap/tiflow/dm/pkg/binlog"
@@ -35,6 +34,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/streamer"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
 	"github.com/pingcap/tiflow/dm/relay"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.uber.org/atomic"
 	"go.uber.org/zap"
diff --git a/engine/jobmaster/dm/config/config.go b/engine/jobmaster/dm/config/config.go
index 396db5945..2cd4f3105 100644
--- a/engine/jobmaster/dm/config/config.go
+++ b/engine/jobmaster/dm/config/config.go
@@ -20,14 +20,14 @@ import (
 
 	"github.com/dustin/go-humanize"
 	"github.com/google/uuid"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	"github.com/pingcap/tidb-tools/pkg/column-mapping"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
 	"github.com/pingcap/tiflow/dm/config"
 	dmconfig "github.com/pingcap/tiflow/dm/config"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/master"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
+	"github.com/pingcap/tiflow/pkg/column-mapping"
 	"github.com/pingcap/tiflow/pkg/errors"
 	"go.uber.org/atomic"
 	"gopkg.in/yaml.v2"
diff --git a/pkg/applier/redo.go b/pkg/applier/redo.go
index 51859757f..ce8514d64 100644
--- a/pkg/applier/redo.go
+++ b/pkg/applier/redo.go
@@ -577,10 +577,10 @@ func (t *tempTxnInsertEventStorage) readNextEvent() (*model.RowChangedEvent, err
 // when the update event is an update to the handle key or the non empty unique key.
 // deferred insert event means all delete events and update events in the same transaction are emitted before this insert event
 type updateEventSplitter struct {
-	rd              reader.RedoLogReader
-	rdFinished      bool
-	tempStorage     *tempTxnInsertEventStorage
-	prevTxnCommitTs model.Ts
+	rd             reader.RedoLogReader
+	rdFinished     bool
+	tempStorage    *tempTxnInsertEventStorage
+	prevTxnStartTs model.Ts
 	// pendingEvent is the event that trigger the process to emit events from tempStorage, it can be
 	// 1) an insert event in the same transaction(because there will be no more update and delete events in the same transaction)
 	// 2) a new event in the next transaction
@@ -592,17 +592,17 @@ type updateEventSplitter struct {
 
 func newUpdateEventSplitter(rd reader.RedoLogReader, dir string) *updateEventSplitter {
 	return &updateEventSplitter{
-		rd:              rd,
-		rdFinished:      false,
-		tempStorage:     newTempTxnInsertEventStorage(defaultFlushThreshold, dir),
-		prevTxnCommitTs: 0,
+		rd:             rd,
+		rdFinished:     false,
+		tempStorage:    newTempTxnInsertEventStorage(defaultFlushThreshold, dir),
+		prevTxnStartTs: 0,
 	}
 }
 
 // processEvent return (event to emit, pending event)
 func processEvent(
 	event *model.RowChangedEvent,
-	prevTxnCommitTs model.Ts,
+	prevTxnStartTs model.Ts,
 	tempStorage *tempTxnInsertEventStorage,
 ) (*model.RowChangedEvent, *model.RowChangedEvent, error) {
 	if event == nil {
@@ -610,7 +610,7 @@ func processEvent(
 	}
 
 	// meet a new transaction
-	if prevTxnCommitTs != 0 && prevTxnCommitTs != event.CommitTs {
+	if prevTxnStartTs != 0 && prevTxnStartTs != event.StartTs {
 		if tempStorage.hasEvent() {
 			// emit the insert events in the previous transaction
 			return nil, event, nil
@@ -643,7 +643,8 @@ func (u *updateEventSplitter) checkEventOrder(event *model.RowChangedEvent) {
 	if event == nil {
 		return
 	}
-	if event.CommitTs > u.prevTxnCommitTs {
+	// meeet a new transaction
+	if event.StartTs != u.prevTxnStartTs {
 		u.meetInsertInCurTxn = false
 		return
 	}
@@ -666,7 +667,7 @@ func (u *updateEventSplitter) readNextRow(ctx context.Context) (*model.RowChange
 			}
 			var event *model.RowChangedEvent
 			var err error
-			event, u.pendingEvent, err = processEvent(u.pendingEvent, u.prevTxnCommitTs, u.tempStorage)
+			event, u.pendingEvent, err = processEvent(u.pendingEvent, u.prevTxnStartTs, u.tempStorage)
 			if err != nil {
 				return nil, err
 			}
@@ -693,10 +694,10 @@ func (u *updateEventSplitter) readNextRow(ctx context.Context) (*model.RowChange
 			u.rdFinished = true
 		} else {
 			u.checkEventOrder(event)
-			prevTxnCommitTS := u.prevTxnCommitTs
-			u.prevTxnCommitTs = event.CommitTs
+			prevTxnStartTs := u.prevTxnStartTs
+			u.prevTxnStartTs = event.StartTs
 			var err error
-			event, u.pendingEvent, err = processEvent(event, prevTxnCommitTS, u.tempStorage)
+			event, u.pendingEvent, err = processEvent(event, prevTxnStartTs, u.tempStorage)
 			if err != nil {
 				return nil, err
 			}
diff --git a/pkg/binlog-filter/filter.go b/pkg/binlog-filter/filter.go
new file mode 100644
index 000000000..5fa84cd4a
--- /dev/null
+++ b/pkg/binlog-filter/filter.go
@@ -0,0 +1,431 @@
+// Copyright 2018 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 filter
+
+import (
+	"regexp"
+	"strings"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/log"
+	selector "github.com/pingcap/tidb/pkg/util/table-rule-selector"
+	"go.uber.org/zap"
+)
+
+// ActionType indicates how to handle matched items
+type ActionType string
+
+// show that how to handle rules
+const (
+	Ignore ActionType = "Ignore"
+	Do     ActionType = "Do"
+	Error  ActionType = "Error"
+)
+
+// EventType is DML/DDL Event type
+type EventType string
+
+// show DML/DDL Events
+const (
+	ddl             EventType = "ddl"
+	dml             EventType = "dml"
+	incompatibleDDL EventType = "incompatible DDL"
+
+	// it indicates all dml/ddl events in rule
+	AllEvent EventType = "all"
+	AllDDL   EventType = "all ddl"
+	AllDML   EventType = "all dml"
+
+	// it indicates no any dml/ddl events in rule,
+	// and equals empty rule.DDLEvent/DMLEvent
+	NoneEvent EventType = "none"
+	NoneDDL   EventType = "none ddl"
+	NoneDML   EventType = "none dml"
+
+	InsertEvent EventType = "insert"
+	UpdateEvent EventType = "update"
+	DeleteEvent EventType = "delete"
+
+	CreateDatabase EventType = "create database"
+	DropDatabase   EventType = "drop database"
+	AlterDatabase  EventType = "alter database"
+	CreateTable    EventType = "create table"
+	DropTable      EventType = "drop table"
+	TruncateTable  EventType = "truncate table"
+	RenameTable    EventType = "rename table"
+	CreateIndex    EventType = "create index"
+	DropIndex      EventType = "drop index"
+	CreateView     EventType = "create view"
+	DropView       EventType = "drop view"
+	AlterTable     EventType = "alter table"
+
+	CreateSchema EventType = "create schema" // alias of CreateDatabase
+	DropSchema   EventType = "drop schema"   // alias of DropDatabase
+	AlterSchema  EventType = "alter schema"  // alias of AlterDatabase
+
+	AddTablePartition      EventType = "add table partition"
+	DropTablePartition     EventType = "drop table partition"
+	TruncateTablePartition EventType = "truncate table partition"
+	// if need, add more	AlertTableOption     = "alert table option"
+
+	IncompatibleDDLChanges EventType = "incompatible ddl changes"
+	ValueRangeDecrease     EventType = "value range decrease"
+	PrecisionDecrease      EventType = "precision decrease"
+	ModifyColumn           EventType = "modify column"
+	RenameColumn           EventType = "rename column"
+	RenameIndex            EventType = "rename index"
+	DropColumn             EventType = "drop column"
+	DropPrimaryKey         EventType = "drop primary key"
+	DropUniqueKey          EventType = "drop unique key"
+	ModifyDefaultValue     EventType = "modify default value"
+	ModifyConstraint       EventType = "modify constraint"
+	ModifyColumnsOrder     EventType = "modify columns order"
+	ModifyCharset          EventType = "modify charset"
+	ModifyCollation        EventType = "modify collation"
+	RemoveAutoIncrement    EventType = "remove auto increment"
+	ModifyStorageEngine    EventType = "modify storage engine"
+	ReorganizePartition    EventType = "reorganize table partition"
+	RebuildPartition       EventType = "rebuild table partition"
+	CoalescePartition      EventType = "coalesce table partition"
+	SplitPartition         EventType = "split table partition"
+	ExchangePartition      EventType = "exchange table partition"
+
+	ModifySchemaCharsetAndCollate EventType = "modify schema charset and collate"
+	ModifyTableCharsetAndCollate  EventType = "modify table charset and collate"
+	ModifyTableComment            EventType = "modify table comment"
+	RecoverTable                  EventType = "recover table"
+	AlterTablePartitioning        EventType = "alter table partitioning"
+	RemovePartitioning            EventType = "remove table partitioning"
+	AddColumn                     EventType = "add column"
+	SetDefaultValue               EventType = "set default value"
+	RebaseAutoID                  EventType = "rebase auto id"
+	AddPrimaryKey                 EventType = "add primary key"
+	AlterIndexVisibility          EventType = "alter index visibility"
+	AlterTTLInfo                  EventType = "alter ttl info"
+	AlterTTLRemove                EventType = "alter ttl remove"
+	MultiSchemaChange             EventType = "multi schema change"
+
+	// NullEvent is used to represents unsupported ddl event type when we
+	// convert a ast.StmtNode or a string to EventType.
+	NullEvent EventType = ""
+)
+
+// ClassifyEvent classify event into dml/ddl
+func ClassifyEvent(event EventType) (EventType, error) {
+	switch event {
+	case InsertEvent,
+		UpdateEvent,
+		DeleteEvent:
+		return dml, nil
+	case CreateDatabase,
+		AlterDatabase,
+		AlterSchema,
+		CreateTable,
+		CreateIndex,
+		CreateView,
+		DropView,
+		AlterTable,
+		CreateSchema,
+		AddTablePartition:
+		return ddl, nil
+	case NullEvent:
+		return NullEvent, nil
+	case ValueRangeDecrease,
+		PrecisionDecrease,
+		ModifyColumn,
+		RenameColumn,
+		RenameIndex,
+		DropColumn,
+		DropPrimaryKey,
+		DropUniqueKey,
+		ModifyDefaultValue,
+		ModifyConstraint,
+		ModifyColumnsOrder,
+		ModifyCharset,
+		ModifyCollation,
+		RemoveAutoIncrement,
+		ModifyStorageEngine,
+		ReorganizePartition,
+		RebuildPartition,
+		CoalescePartition,
+		SplitPartition,
+		ExchangePartition,
+
+		DropDatabase,
+		DropTable,
+		DropIndex,
+		RenameTable,
+		TruncateTable,
+		DropSchema,
+		DropTablePartition,
+		TruncateTablePartition,
+
+		ModifySchemaCharsetAndCollate,
+		ModifyTableCharsetAndCollate,
+		ModifyTableComment,
+		RecoverTable,
+		AlterTablePartitioning,
+		RemovePartitioning,
+		AddColumn,
+		SetDefaultValue,
+		RebaseAutoID,
+		AddPrimaryKey,
+		AlterIndexVisibility,
+		AlterTTLInfo,
+		AlterTTLRemove,
+		MultiSchemaChange:
+		return incompatibleDDL, nil
+	default:
+		return NullEvent, errors.NotValidf("event type %s", event)
+	}
+}
+
+// BinlogEventRule is a rule to filter binlog events
+type BinlogEventRule struct {
+	SchemaPattern string      `json:"schema-pattern" toml:"schema-pattern" yaml:"schema-pattern"`
+	TablePattern  string      `json:"table-pattern" toml:"table-pattern" yaml:"table-pattern"`
+	Events        []EventType `json:"events" toml:"events" yaml:"events"`
+	SQLPattern    []string    `json:"sql-pattern" toml:"sql-pattern" yaml:"sql-pattern"` // regular expression
+	sqlRegularExp *regexp.Regexp
+
+	Action ActionType `json:"action" toml:"action" yaml:"action"`
+}
+
+// ToLower covert schema/table pattern to lower case
+func (b *BinlogEventRule) ToLower() {
+	b.SchemaPattern = strings.ToLower(b.SchemaPattern)
+	b.TablePattern = strings.ToLower(b.TablePattern)
+}
+
+// Valid checks validity of rule.
+func (b *BinlogEventRule) Valid() error {
+	if len(b.SQLPattern) > 0 {
+		reg, err := regexp.Compile("(?i)" + strings.Join(b.SQLPattern, "|"))
+		if err != nil {
+			return errors.Annotatef(err, "compile regular expression %+v", b.SQLPattern)
+		}
+		b.sqlRegularExp = reg
+	}
+
+	if b.Action != Do && b.Action != Ignore && b.Action != Error {
+		return errors.Errorf("action of binlog event rule %+v should not be empty", b)
+	}
+
+	for i := range b.Events {
+		et, err := toEventType(string(b.Events[i]))
+		if err != nil {
+			return errors.NotValidf("event type %s", b.Events[i])
+		}
+		b.Events[i] = et
+	}
+	return nil
+}
+
+// BinlogEvent filters binlog events by given rules
+type BinlogEvent struct {
+	selector.Selector
+
+	caseSensitive bool
+}
+
+// NewBinlogEvent returns a binlog event filter
+func NewBinlogEvent(caseSensitive bool, rules []*BinlogEventRule) (*BinlogEvent, error) {
+	b := &BinlogEvent{
+		Selector:      selector.NewTrieSelector(),
+		caseSensitive: caseSensitive,
+	}
+
+	for _, rule := range rules {
+		if err := b.AddRule(rule); err != nil {
+			log.Error("invalid binlog event rule", zap.Any("rule", rule), zap.Error(err))
+		}
+	}
+
+	return b, nil
+}
+
+// AddRule adds a rule into binlog event filter
+func (b *BinlogEvent) AddRule(rule *BinlogEventRule) error {
+	if b == nil || rule == nil {
+		return nil
+	}
+	err := rule.Valid()
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if !b.caseSensitive {
+		rule.ToLower()
+	}
+
+	err = b.Insert(rule.SchemaPattern, rule.TablePattern, rule, selector.Insert)
+	if err != nil {
+		return errors.Annotatef(err, "add rule %+v into binlog event filter", rule)
+	}
+
+	return nil
+}
+
+// UpdateRule updates binlog event filter rule
+func (b *BinlogEvent) UpdateRule(rule *BinlogEventRule) error {
+	if b == nil || rule == nil {
+		return nil
+	}
+	err := rule.Valid()
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if !b.caseSensitive {
+		rule.ToLower()
+	}
+
+	err = b.Insert(rule.SchemaPattern, rule.TablePattern, rule, selector.Replace)
+	if err != nil {
+		return errors.Annotatef(err, "update rule %+v into binlog event filter", rule)
+	}
+
+	return nil
+}
+
+// RemoveRule removes a rule from binlog event filter
+func (b *BinlogEvent) RemoveRule(rule *BinlogEventRule) error {
+	if b == nil || rule == nil {
+		return nil
+	}
+	if !b.caseSensitive {
+		rule.ToLower()
+	}
+
+	err := b.Remove(rule.SchemaPattern, rule.TablePattern)
+	if err != nil {
+		return errors.Annotatef(err, "remove rule %+v", rule)
+	}
+
+	return nil
+}
+
+// Filter filters events or queries by given rules
+// returns action and error
+func (b *BinlogEvent) Filter(schema, table string, event EventType, rawQuery string) (ActionType, error) {
+	if b == nil {
+		return Do, nil
+	}
+
+	tp, err := ClassifyEvent(event)
+	if err != nil {
+		return Ignore, errors.Trace(err)
+	}
+
+	schemaL, tableL := schema, table
+	if !b.caseSensitive {
+		schemaL, tableL = strings.ToLower(schema), strings.ToLower(table)
+	}
+
+	rules := b.Match(schemaL, tableL)
+	if len(rules) == 0 {
+		return Do, nil
+	}
+
+	for _, rule := range rules {
+		binlogEventRule, ok := rule.(*BinlogEventRule)
+		if !ok {
+			return "", errors.NotValidf("rule %+v", rule)
+		}
+
+		if tp != NullEvent {
+			matched := b.matchEvent(tp, event, binlogEventRule.Events)
+
+			if matched {
+				// ignore has highest priority
+				if binlogEventRule.Action == Ignore {
+					return Ignore, nil
+				}
+				if binlogEventRule.Action == Error {
+					return Error, nil
+				}
+			} else {
+				if binlogEventRule.Action == Do {
+					return Ignore, nil
+				}
+			}
+		}
+
+		if len(rawQuery) > 0 {
+			if len(binlogEventRule.SQLPattern) == 0 {
+				// sql pattern is disabled , just continue
+				continue
+			}
+
+			matched := binlogEventRule.sqlRegularExp.FindStringIndex(rawQuery) != nil
+			if matched {
+				// Ignore has highest priority
+				if binlogEventRule.Action == Ignore {
+					return Ignore, nil
+				}
+				if binlogEventRule.Action == Error {
+					return Error, nil
+				}
+			} else {
+				if binlogEventRule.Action == Do {
+					return Ignore, nil
+				}
+			}
+		}
+	}
+
+	return Do, nil
+}
+
+func (b *BinlogEvent) matchEvent(tp, event EventType, rules []EventType) bool {
+	for _, rule := range rules {
+		if rule == AllEvent {
+			return true
+		}
+
+		if rule == NoneEvent {
+			return false
+		}
+
+		if tp == ddl || tp == incompatibleDDL {
+			if rule == AllDDL {
+				return true
+			}
+
+			if rule == NoneDDL {
+				return false
+			}
+		}
+
+		if tp == dml {
+			if rule == AllDML {
+				return true
+			}
+
+			if rule == NoneDML {
+				return false
+			}
+		}
+
+		if tp == incompatibleDDL {
+			if rule == IncompatibleDDLChanges {
+				return true
+			}
+		}
+
+		if rule == event {
+			return true
+		}
+	}
+
+	return false
+}
diff --git a/pkg/binlog-filter/util.go b/pkg/binlog-filter/util.go
new file mode 100644
index 000000000..e076ba62e
--- /dev/null
+++ b/pkg/binlog-filter/util.go
@@ -0,0 +1,132 @@
+// Copyright 2018 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 filter
+
+import (
+	"strings"
+
+	"github.com/pingcap/errors"
+	"github.com/pingcap/tidb/pkg/parser/ast"
+)
+
+// AstToDDLEvent returns filter.DDLEvent
+func AstToDDLEvent(node ast.StmtNode) EventType {
+	switch n := node.(type) {
+	case *ast.CreateDatabaseStmt:
+		return CreateDatabase
+	case *ast.DropDatabaseStmt:
+		return DropDatabase
+	case *ast.CreateTableStmt:
+		return CreateTable
+	case *ast.DropTableStmt:
+		if n.IsView {
+			return DropView
+		}
+		return DropTable
+	case *ast.TruncateTableStmt:
+		return TruncateTable
+	case *ast.RenameTableStmt:
+		return RenameTable
+	case *ast.CreateIndexStmt:
+		return CreateIndex
+	case *ast.DropIndexStmt:
+		return DropIndex
+	case *ast.AlterTableStmt:
+		return AlterTable
+	case *ast.CreateViewStmt:
+		return CreateView
+	case *ast.AlterDatabaseStmt:
+		return AlterDatabase
+	}
+
+	return NullEvent
+}
+
+// toEventType converts event type string to EventType and check if it is valid.
+func toEventType(es string) (EventType, error) {
+	event := EventType(strings.ToLower(es))
+	switch event {
+	case AllEvent,
+		AllDDL,
+		AllDML,
+		NullEvent,
+		NoneEvent,
+		NoneDDL,
+		NoneDML,
+		InsertEvent,
+		UpdateEvent,
+		DeleteEvent,
+		CreateDatabase,
+		DropDatabase,
+		AlterDatabase,
+		CreateTable,
+		DropTable,
+		TruncateTable,
+		RenameTable,
+		CreateIndex,
+		DropIndex,
+		CreateView,
+		DropView,
+		AlterTable,
+		AddTablePartition,
+		DropTablePartition,
+		TruncateTablePartition,
+
+		IncompatibleDDLChanges,
+		ValueRangeDecrease,
+		PrecisionDecrease,
+		ModifyColumn,
+		RenameColumn,
+		RenameIndex,
+		DropColumn,
+		DropPrimaryKey,
+		DropUniqueKey,
+		ModifyDefaultValue,
+		ModifyConstraint,
+		ModifyColumnsOrder,
+		ModifyCharset,
+		ModifyCollation,
+		RemoveAutoIncrement,
+		ModifyStorageEngine,
+		ReorganizePartition,
+		RebuildPartition,
+		CoalescePartition,
+		SplitPartition,
+		ExchangePartition,
+
+		ModifySchemaCharsetAndCollate,
+		ModifyTableCharsetAndCollate,
+		ModifyTableComment,
+		RecoverTable,
+		AlterTablePartitioning,
+		RemovePartitioning,
+		AddColumn,
+		SetDefaultValue,
+		RebaseAutoID,
+		AddPrimaryKey,
+		AlterIndexVisibility,
+		AlterTTLInfo,
+		AlterTTLRemove,
+		MultiSchemaChange:
+		return event, nil
+	case CreateSchema: // alias of CreateDatabase
+		return CreateDatabase, nil
+	case DropSchema: // alias of DropDatabase
+		return DropDatabase, nil
+	case AlterSchema:
+		return AlterDatabase, nil
+	default:
+		return NullEvent, errors.NotValidf("event type %s", es)
+	}
+}
diff --git a/pkg/column-mapping/column.go b/pkg/column-mapping/column.go
new file mode 100644
index 000000000..77e1a4fd2
--- /dev/null
+++ b/pkg/column-mapping/column.go
@@ -0,0 +1,538 @@
+// Copyright 2018 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 column
+
+import (
+	"fmt"
+	"strconv"
+	"strings"
+	"sync"
+
+	"github.com/pingcap/errors"
+	selector "github.com/pingcap/tidb/pkg/util/table-rule-selector"
+)
+
+var (
+	// for partition ID, ref definition of partitionID
+	instanceIDBitSize       = 4
+	schemaIDBitSize         = 7
+	tableIDBitSize          = 8
+	maxOriginID       int64 = 17592186044416
+)
+
+// SetPartitionRule sets bit size of schema ID and table ID
+func SetPartitionRule(instanceIDSize, schemaIDSize, tableIDSize int) {
+	instanceIDBitSize = instanceIDSize
+	schemaIDBitSize = schemaIDSize
+	tableIDBitSize = tableIDSize
+	maxOriginID = 1 << uint(64-instanceIDSize-schemaIDSize-tableIDSize-1)
+}
+
+// Expr indicates how to handle column mapping
+type Expr string
+
+// poor Expr
+const (
+	AddPrefix   Expr = "add prefix"
+	AddSuffix   Expr = "add suffix"
+	PartitionID Expr = "partition id"
+)
+
+// Exprs is some built-in expression for column mapping
+// only supports some poor expressions now,
+// we would unify tableInfo later and support more
+var Exprs = map[Expr]func(*mappingInfo, []interface{}) ([]interface{}, error){
+	AddPrefix: addPrefix, // arguments contains prefix
+	AddSuffix: addSuffix, // arguments contains suffix
+	// arguments contains [instance_id, prefix of schema, prefix of table]
+	// we would compute a ID like
+	// [1:1 bit][2:9 bits][3:10 bits][4:44 bits] int64  (using default bits length)
+	// # 1 useless, no reason
+	// # 2 schema ID (schema suffix)
+	// # 3 table ID (table suffix)
+	// # 4 origin ID (>= 0, <= 17592186044415)
+	//
+	// others: schema = arguments[1]  or  arguments[1] + arguments[3] + schema suffix
+	//         table  = arguments[2]  or  arguments[2] + arguments[3] + table suffix
+	//  example: schema = schema_1 table = t_1
+	//        => arguments[1] = "schema", arguments[2] = "t", arguments[3] = "_"
+	//  if arguments[1]/arguments[2] == "", it means we don't use schemaID/tableID to compute partition ID
+	//  if length of arguments is < 4, arguments[3] is set to "" (empty string)
+	PartitionID: partitionID,
+}
+
+// Rule is a rule to map column
+// TODO: we will do it later, if we need to implement a real column mapping, we need table structure of source and target system
+type Rule struct {
+	PatternSchema    string   `yaml:"schema-pattern" json:"schema-pattern" toml:"schema-pattern"`
+	PatternTable     string   `yaml:"table-pattern" json:"table-pattern" toml:"table-pattern"`
+	SourceColumn     string   `yaml:"source-column" json:"source-column" toml:"source-column"` // modify, add refer column, ignore
+	TargetColumn     string   `yaml:"target-column" json:"target-column" toml:"target-column"` // add column, modify
+	Expression       Expr     `yaml:"expression" json:"expression" toml:"expression"`
+	Arguments        []string `yaml:"arguments" json:"arguments" toml:"arguments"`
+	CreateTableQuery string   `yaml:"create-table-query" json:"create-table-query" toml:"create-table-query"`
+}
+
+// ToLower covert schema/table parttern to lower case
+func (r *Rule) ToLower() {
+	r.PatternSchema = strings.ToLower(r.PatternSchema)
+	r.PatternTable = strings.ToLower(r.PatternTable)
+}
+
+// Valid checks validity of rule.
+// add prefix/suffix: it should have target column and one argument
+// partition id: it should have 3 to 4 arguments
+func (r *Rule) Valid() error {
+	if _, ok := Exprs[r.Expression]; !ok {
+		return errors.NotFoundf("expression %s", r.Expression)
+	}
+
+	if r.TargetColumn == "" {
+		return errors.NotValidf("rule need to be applied a target column")
+	}
+
+	if r.Expression == AddPrefix || r.Expression == AddSuffix {
+		if len(r.Arguments) != 1 {
+			return errors.NotValidf("arguments %v for add prefix/suffix", r.Arguments)
+		}
+	}
+
+	if r.Expression == PartitionID {
+		switch len(r.Arguments) {
+		case 3, 4:
+			break
+		default:
+			return errors.NotValidf("arguments %v for patition id", r.Arguments)
+		}
+	}
+
+	return nil
+}
+
+// Adjust normalizes the rule into an easier-to-process form, e.g. filling in
+// optional arguments with the default values.
+func (r *Rule) Adjust() {
+	if r.Expression == PartitionID && len(r.Arguments) == 3 {
+		r.Arguments = append(r.Arguments, "")
+	}
+}
+
+// check source and target position
+func (r *Rule) adjustColumnPosition(source, target int) (int, int, error) {
+	// if not found target, ignore it
+	if target == -1 {
+		return source, target, errors.NotFoundf("target column %s", r.TargetColumn)
+	}
+
+	return source, target, nil
+}
+
+type mappingInfo struct {
+	ignore         bool
+	sourcePosition int
+	targetPosition int
+	rule           *Rule
+
+	instanceID int64
+	schemaID   int64
+	tableID    int64
+}
+
+// Mapping maps column to something by rules
+type Mapping struct {
+	selector.Selector
+
+	caseSensitive bool
+
+	cache struct {
+		sync.RWMutex
+		infos map[string]*mappingInfo
+	}
+}
+
+// NewMapping returns a column mapping
+func NewMapping(caseSensitive bool, rules []*Rule) (*Mapping, error) {
+	m := &Mapping{
+		Selector:      selector.NewTrieSelector(),
+		caseSensitive: caseSensitive,
+	}
+	m.resetCache()
+
+	for _, rule := range rules {
+		if err := m.AddRule(rule); err != nil {
+			return nil, errors.Annotatef(err, "initial rule %+v in mapping", rule)
+		}
+	}
+
+	return m, nil
+}
+
+func (m *Mapping) addOrUpdateRule(rule *Rule, isUpdate bool) error {
+	if m == nil || rule == nil {
+		return nil
+	}
+
+	err := rule.Valid()
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if !m.caseSensitive {
+		rule.ToLower()
+	}
+	rule.Adjust()
+
+	m.resetCache()
+	if isUpdate {
+		err = m.Insert(rule.PatternSchema, rule.PatternTable, rule, selector.Replace)
+	} else {
+		err = m.Insert(rule.PatternSchema, rule.PatternTable, rule, selector.Insert)
+	}
+	if err != nil {
+		var method string
+		if isUpdate {
+			method = "update"
+		} else {
+			method = "add"
+		}
+		return errors.Annotatef(err, "%s rule %+v into mapping", method, rule)
+	}
+
+	return nil
+}
+
+// AddRule adds a rule into mapping
+func (m *Mapping) AddRule(rule *Rule) error {
+	return m.addOrUpdateRule(rule, false)
+}
+
+// UpdateRule updates mapping rule
+func (m *Mapping) UpdateRule(rule *Rule) error {
+	return m.addOrUpdateRule(rule, true)
+}
+
+// RemoveRule removes a rule from mapping
+func (m *Mapping) RemoveRule(rule *Rule) error {
+	if m == nil || rule == nil {
+		return nil
+	}
+	if !m.caseSensitive {
+		rule.ToLower()
+	}
+
+	m.resetCache()
+	err := m.Remove(rule.PatternSchema, rule.PatternTable)
+	if err != nil {
+		return errors.Annotatef(err, "remove rule %+v from mapping", rule)
+	}
+
+	return nil
+}
+
+// HandleRowValue handles row value
+func (m *Mapping) HandleRowValue(schema, table string, columns []string, vals []interface{}) ([]interface{}, []int, error) {
+	if m == nil {
+		return vals, nil, nil
+	}
+
+	schemaL, tableL := schema, table
+	if !m.caseSensitive {
+		schemaL, tableL = strings.ToLower(schema), strings.ToLower(table)
+	}
+
+	info, err := m.queryColumnInfo(schemaL, tableL, columns)
+	if err != nil {
+		return nil, nil, errors.Trace(err)
+	}
+	if info.ignore {
+		return vals, nil, nil
+	}
+
+	exp, ok := Exprs[info.rule.Expression]
+	if !ok {
+		return nil, nil, errors.NotFoundf("column mapping expression %s", info.rule.Expression)
+	}
+
+	vals, err = exp(info, vals)
+	if err != nil {
+		return nil, nil, errors.Trace(err)
+	}
+
+	return vals, []int{info.sourcePosition, info.targetPosition}, nil
+}
+
+// HandleDDL handles ddl
+func (m *Mapping) HandleDDL(schema, table string, columns []string, statement string) (string, []int, error) {
+	if m == nil {
+		return statement, nil, nil
+	}
+
+	schemaL, tableL := schema, table
+	if !m.caseSensitive {
+		schemaL, tableL = strings.ToLower(schema), strings.ToLower(table)
+	}
+
+	info, err := m.queryColumnInfo(schemaL, tableL, columns)
+	if err != nil {
+		return statement, nil, errors.Trace(err)
+	}
+
+	if info.ignore {
+		return statement, nil, nil
+	}
+
+	m.resetCache()
+	// only output erro now, wait fix it manually
+	return statement, nil, errors.Errorf("ddl %s @ column mapping rule %s/%s:%+v not implemented", statement, schema, table, info.rule)
+}
+
+func (m *Mapping) queryColumnInfo(schema, table string, columns []string) (*mappingInfo, error) {
+	m.cache.RLock()
+	ci, ok := m.cache.infos[tableName(schema, table)]
+	m.cache.RUnlock()
+	if ok {
+		return ci, nil
+	}
+
+	info := &mappingInfo{
+		ignore: true,
+	}
+	rules := m.Match(schema, table)
+	if len(rules) == 0 {
+		m.cache.Lock()
+		m.cache.infos[tableName(schema, table)] = info
+		m.cache.Unlock()
+
+		return info, nil
+	}
+
+	var (
+		schemaRules []*Rule
+		tableRules  = make([]*Rule, 0, 1)
+	)
+	// classify rules into schema level rules and table level
+	// table level rules have highest priority
+	for i := range rules {
+		rule, ok := rules[i].(*Rule)
+		if !ok {
+			return nil, errors.NotValidf("column mapping rule %+v", rules[i])
+		}
+
+		if len(rule.PatternTable) == 0 {
+			schemaRules = append(schemaRules, rule)
+		} else {
+			tableRules = append(tableRules, rule)
+		}
+	}
+
+	// only support one expression for one table now, refine it later
+	var rule *Rule
+	if len(table) == 0 || len(tableRules) == 0 {
+		if len(schemaRules) != 1 {
+			return nil, errors.NotSupportedf("`%s`.`%s` matches %d schema column mapping rules which should be one. It's", schema, table, len(schemaRules))
+		}
+
+		rule = schemaRules[0]
+	} else {
+		if len(tableRules) != 1 {
+			return nil, errors.NotSupportedf("`%s`.`%s` matches %d table column mapping rules which should be one. It's", schema, table, len(tableRules))
+		}
+
+		rule = tableRules[0]
+	}
+	if rule == nil {
+		m.cache.Lock()
+		m.cache.infos[tableName(schema, table)] = info
+		m.cache.Unlock()
+
+		return info, nil
+	}
+
+	// compute source and target column position
+	sourcePosition := findColumnPosition(columns, rule.SourceColumn)
+	targetPosition := findColumnPosition(columns, rule.TargetColumn)
+
+	sourcePosition, targetPosition, err := rule.adjustColumnPosition(sourcePosition, targetPosition)
+	if err != nil {
+		return nil, errors.Trace(err)
+	}
+
+	info = &mappingInfo{
+		sourcePosition: sourcePosition,
+		targetPosition: targetPosition,
+		rule:           rule,
+	}
+
+	// if expr is partition ID, compute schema and table ID
+	if rule.Expression == PartitionID {
+		info.instanceID, info.schemaID, info.tableID, err = computePartitionID(schema, table, rule)
+		if err != nil {
+			return nil, errors.Trace(err)
+		}
+	}
+
+	m.cache.Lock()
+	m.cache.infos[tableName(schema, table)] = info
+	m.cache.Unlock()
+
+	return info, nil
+}
+
+func (m *Mapping) resetCache() {
+	m.cache.Lock()
+	m.cache.infos = make(map[string]*mappingInfo)
+	m.cache.Unlock()
+}
+
+func findColumnPosition(cols []string, col string) int {
+	for i := range cols {
+		if cols[i] == col {
+			return i
+		}
+	}
+
+	return -1
+}
+
+func tableName(schema, table string) string {
+	return fmt.Sprintf("`%s`.`%s`", schema, table)
+}
+
+func addPrefix(info *mappingInfo, vals []interface{}) ([]interface{}, error) {
+	prefix := info.rule.Arguments[0]
+	originStr, ok := vals[info.targetPosition].(string)
+	if !ok {
+		return nil, errors.NotValidf("column %d value is not string, but %v, which is", info.targetPosition, vals[info.targetPosition])
+	}
+
+	// fast to concatenated string
+	rawByte := make([]byte, 0, len(prefix)+len(originStr))
+	rawByte = append(rawByte, prefix...)
+	rawByte = append(rawByte, originStr...)
+
+	vals[info.targetPosition] = string(rawByte)
+	return vals, nil
+}
+
+func addSuffix(info *mappingInfo, vals []interface{}) ([]interface{}, error) {
+	suffix := info.rule.Arguments[0]
+	originStr, ok := vals[info.targetPosition].(string)
+	if !ok {
+		return nil, errors.NotValidf("column %d value is not string, but %v, which is", info.targetPosition, vals[info.targetPosition])
+	}
+
+	rawByte := make([]byte, 0, len(suffix)+len(originStr))
+	rawByte = append(rawByte, originStr...)
+	rawByte = append(rawByte, suffix...)
+
+	vals[info.targetPosition] = string(rawByte)
+	return vals, nil
+}
+
+func partitionID(info *mappingInfo, vals []interface{}) ([]interface{}, error) {
+	// only int64 now
+	var (
+		originID int64
+		err      error
+		isChars  bool
+	)
+
+	switch rawID := vals[info.targetPosition].(type) {
+	case int:
+		originID = int64(rawID)
+	case int8:
+		originID = int64(rawID)
+	case int32:
+		originID = int64(rawID)
+	case int64:
+		originID = rawID
+	case uint:
+		originID = int64(rawID)
+	case uint16:
+		originID = int64(rawID)
+	case uint32:
+		originID = int64(rawID)
+	case uint64:
+		originID = int64(rawID)
+	case string:
+		originID, err = strconv.ParseInt(rawID, 10, 64)
+		if err != nil {
+			return nil, errors.NotValidf("column %d value is not int, but %v, which is", info.targetPosition, vals[info.targetPosition])
+		}
+		isChars = true
+	default:
+		return nil, errors.NotValidf("type %T(%v)", vals[info.targetPosition], vals[info.targetPosition])
+	}
+
+	if originID >= maxOriginID || originID < 0 {
+		return nil, errors.NotValidf("id must less than %d, greater than or equal to 0, but got %d, which is", maxOriginID, originID)
+	}
+
+	originID = info.instanceID | info.schemaID | info.tableID | originID
+	if isChars {
+		vals[info.targetPosition] = strconv.FormatInt(originID, 10)
+	} else {
+		vals[info.targetPosition] = originID
+	}
+
+	return vals, nil
+}
+
+func computePartitionID(schema, table string, rule *Rule) (instanceID int64, schemaID int64, tableID int64, err error) {
+	shiftCnt := uint(63)
+	if instanceIDBitSize > 0 && len(rule.Arguments[0]) > 0 {
+		var instanceIDUnsign uint64
+		shiftCnt = shiftCnt - uint(instanceIDBitSize)
+		instanceIDUnsign, err = strconv.ParseUint(rule.Arguments[0], 10, instanceIDBitSize)
+		if err != nil {
+			return
+		}
+		instanceID = int64(instanceIDUnsign << shiftCnt)
+	}
+
+	sep := rule.Arguments[3]
+
+	if schemaIDBitSize > 0 && len(rule.Arguments[1]) > 0 {
+		shiftCnt = shiftCnt - uint(schemaIDBitSize)
+		schemaID, err = computeID(schema, rule.Arguments[1], sep, schemaIDBitSize, shiftCnt)
+		if err != nil {
+			return
+		}
+	}
+
+	if tableIDBitSize > 0 && len(rule.Arguments[2]) > 0 {
+		shiftCnt = shiftCnt - uint(tableIDBitSize)
+		tableID, err = computeID(table, rule.Arguments[2], sep, tableIDBitSize, shiftCnt)
+	}
+
+	return
+}
+
+func computeID(name string, prefix, sep string, bitSize int, shiftCount uint) (int64, error) {
+	if name == prefix {
+		return 0, nil
+	}
+
+	prefix += sep
+	if len(prefix) >= len(name) || prefix != name[:len(prefix)] {
+		return 0, errors.NotValidf("%s is not the prefix of %s", prefix, name)
+	}
+
+	idStr := name[len(prefix):]
+	id, err := strconv.ParseUint(idStr, 10, bitSize)
+	if err != nil {
+		return 0, errors.NotValidf("the suffix of %s can't be converted to int64", idStr)
+	}
+
+	return int64(id << shiftCount), nil
+}
diff --git a/pkg/config/filter.go b/pkg/config/filter.go
index fb77779b7..22a5bca1f 100644
--- a/pkg/config/filter.go
+++ b/pkg/config/filter.go
@@ -14,8 +14,8 @@
 package config
 
 import (
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	filter "github.com/pingcap/tidb/pkg/util/table-filter"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 )
 
 // FilterConfig represents filter config for a changefeed
diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go
index 704aa58e5..b66970515 100644
--- a/pkg/config/replica_config.go
+++ b/pkg/config/replica_config.go
@@ -45,6 +45,9 @@ const (
 	// 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
+
+	// DefaultTiDBSourceID is the default source ID of TiDB cluster.
+	DefaultTiDBSourceID = 1
 )
 
 var defaultReplicaConfig = &ReplicaConfig{
@@ -75,7 +78,7 @@ var defaultReplicaConfig = &ReplicaConfig{
 		EnableKafkaSinkV2:                util.AddressOf(false),
 		OnlyOutputUpdatedColumns:         util.AddressOf(false),
 		DeleteOnlyOutputHandleKeyColumns: util.AddressOf(false),
-		TiDBSourceID:                     1,
+		TiDBSourceID:                     DefaultTiDBSourceID,
 		AdvanceTimeoutInSec:              util.AddressOf(DefaultAdvanceTimeoutInSec),
 		SendBootstrapIntervalInSec:       util.AddressOf(DefaultSendBootstrapIntervalInSec),
 		SendBootstrapInMsgCount:          util.AddressOf(DefaultSendBootstrapInMsgCount),
diff --git a/pkg/context/context.go b/pkg/context/context.go
index ce2acdac8..e00f2c71b 100644
--- a/pkg/context/context.go
+++ b/pkg/context/context.go
@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/etcd"
 	"github.com/pingcap/tiflow/pkg/p2p"
+	"github.com/pingcap/tiflow/pkg/workerpool"
 	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
 )
@@ -43,6 +44,9 @@ type GlobalVars struct {
 	// MessageServer and MessageRouter are for peer-messaging
 	MessageServer *p2p.MessageServer
 	MessageRouter p2p.MessageRouter
+
+	// ChangefeedThreadPool is the thread pool for changefeed initialization
+	ChangefeedThreadPool workerpool.AsyncPool
 }
 
 // ChangefeedVars contains some vars which can be used anywhere in a pipeline
@@ -194,6 +198,7 @@ func NewContext4Test(baseCtx context.Context, withChangefeedVars bool) Context {
 		EtcdClient: &etcd.CDCEtcdClientImpl{
 			ClusterID: etcd.DefaultCDCClusterID,
 		},
+		ChangefeedThreadPool: &NonAsyncPool{},
 	})
 	if withChangefeedVars {
 		ctx = WithChangefeedVars(ctx, &ChangefeedVars{
@@ -212,3 +217,18 @@ func NewContext4Test(baseCtx context.Context, withChangefeedVars bool) Context {
 func NewBackendContext4Test(withChangefeedVars bool) Context {
 	return NewContext4Test(context.Background(), withChangefeedVars)
 }
+
+// NonAsyncPool is a dummy implementation of workerpool.AsyncPool, which runs tasks synchronously.
+// It is used in tests to avoid the overhead of asynchronous task scheduling.
+type NonAsyncPool struct{}
+
+// Go runs the task synchronously.
+func (f *NonAsyncPool) Go(_ context.Context, fn func()) error {
+	fn()
+	return nil
+}
+
+// Run does nothing.
+func (f *NonAsyncPool) Run(_ context.Context) error {
+	return nil
+}
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index 01147c7c0..39410d908 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -19,11 +19,11 @@ import (
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"go.uber.org/zap"
diff --git a/pkg/filter/utils.go b/pkg/filter/utils.go
index ef5ddef46..906226d0f 100644
--- a/pkg/filter/utils.go
+++ b/pkg/filter/utils.go
@@ -16,11 +16,11 @@ package filter
 import (
 	"fmt"
 
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/parser"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	tifilter "github.com/pingcap/tidb/pkg/util/filter"
 	tfilter "github.com/pingcap/tidb/pkg/util/table-filter"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 )
diff --git a/pkg/pdutil/utils.go b/pkg/pdutil/utils.go
index 9949c2c00..afb210a56 100644
--- a/pkg/pdutil/utils.go
+++ b/pkg/pdutil/utils.go
@@ -18,6 +18,7 @@ import (
 	"strconv"
 
 	"github.com/pingcap/log"
+	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
@@ -29,11 +30,12 @@ const sourceIDName = "source_id"
 func GetSourceID(ctx context.Context, pdClient pd.Client) (uint64, error) {
 	// only nil in test case
 	if pdClient == nil {
-		return 1, nil
+		return config.DefaultTiDBSourceID, nil
 	}
+
 	// The default value of sourceID is 1,
 	// which means the sourceID is not changed by user.
-	sourceID := uint64(1)
+	sourceID := uint64(config.DefaultTiDBSourceID)
 	sourceIDConfig, _, err := pdClient.LoadGlobalConfig(ctx, []string{sourceIDName}, "")
 	if err != nil {
 		return 0, cerror.WrapError(cerror.ErrPDEtcdAPIError, err)
diff --git a/pkg/sink/codec/simple/message.go b/pkg/sink/codec/simple/message.go
index be112b1d3..c7e80f330 100644
--- a/pkg/sink/codec/simple/message.go
+++ b/pkg/sink/codec/simple/message.go
@@ -415,8 +415,9 @@ func buildRowChangedEvent(
 			Version:   msg.Checksum.Version,
 		}
 
-		if msg.Checksum.Corrupted || previousCorrupted {
-			log.Warn("consumer detect previous checksum corrupted",
+		corrupted := msg.Checksum.Corrupted || previousCorrupted || currentCorrupted
+		if corrupted {
+			log.Warn("consumer detect checksum corrupted",
 				zap.String("schema", msg.Schema),
 				zap.String("table", msg.Table))
 			for idx, col := range result.PreColumns {
@@ -429,13 +430,6 @@ func buildRowChangedEvent(
 					zap.Any("value", col.Value),
 					zap.Any("default", colInfo.GetDefaultValue()))
 			}
-			return nil, cerror.ErrDecodeFailed.GenWithStackByArgs("checksum corrupted")
-		}
-
-		if msg.Checksum.Corrupted || currentCorrupted {
-			log.Warn("consumer detect checksum corrupted",
-				zap.String("schema", msg.Schema),
-				zap.String("table", msg.Table))
 			for idx, col := range result.Columns {
 				colInfo := tableInfo.Columns[idx]
 				log.Info("data corrupted, print each column for debugging",
diff --git a/pkg/sink/codec/simple/mock/marshaller.go b/pkg/sink/codec/simple/mock/marshaller.go
new file mode 100644
index 000000000..b1b7e95f2
--- /dev/null
+++ b/pkg/sink/codec/simple/mock/marshaller.go
@@ -0,0 +1,94 @@
+// Code generated by MockGen. DO NOT EDIT.
+// Source: pkg/sink/codec/simple/marshaller.go
+
+// Package mock_simple is a generated GoMock package.
+package mock_simple
+
+import (
+	reflect "reflect"
+
+	gomock "github.com/golang/mock/gomock"
+	model "github.com/pingcap/tiflow/cdc/model"
+)
+
+// Mockmarshaller is a mock of marshaller interface.
+type Mockmarshaller struct {
+	ctrl     *gomock.Controller
+	recorder *MockmarshallerMockRecorder
+}
+
+// MockmarshallerMockRecorder is the mock recorder for Mockmarshaller.
+type MockmarshallerMockRecorder struct {
+	mock *Mockmarshaller
+}
+
+// NewMockmarshaller creates a new mock instance.
+func NewMockmarshaller(ctrl *gomock.Controller) *Mockmarshaller {
+	mock := &Mockmarshaller{ctrl: ctrl}
+	mock.recorder = &MockmarshallerMockRecorder{mock}
+	return mock
+}
+
+// EXPECT returns an object that allows the caller to indicate expected use.
+func (m *Mockmarshaller) EXPECT() *MockmarshallerMockRecorder {
+	return m.recorder
+}
+
+// MarshalCheckpoint mocks base method.
+func (m *Mockmarshaller) MarshalCheckpoint(ts uint64) ([]byte, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "MarshalCheckpoint", ts)
+	ret0, _ := ret[0].([]byte)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// MarshalCheckpoint indicates an expected call of MarshalCheckpoint.
+func (mr *MockmarshallerMockRecorder) MarshalCheckpoint(ts interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarshalCheckpoint", reflect.TypeOf((*Mockmarshaller)(nil).MarshalCheckpoint), ts)
+}
+
+// MarshalDDLEvent mocks base method.
+func (m *Mockmarshaller) MarshalDDLEvent(event *model.DDLEvent) ([]byte, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "MarshalDDLEvent", event)
+	ret0, _ := ret[0].([]byte)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// MarshalDDLEvent indicates an expected call of MarshalDDLEvent.
+func (mr *MockmarshallerMockRecorder) MarshalDDLEvent(event interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarshalDDLEvent", reflect.TypeOf((*Mockmarshaller)(nil).MarshalDDLEvent), event)
+}
+
+// MarshalRowChangedEvent mocks base method.
+func (m *Mockmarshaller) MarshalRowChangedEvent(event *model.RowChangedEvent, handleKeyOnly bool, claimCheckFileName string) ([]byte, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "MarshalRowChangedEvent", event, handleKeyOnly, claimCheckFileName)
+	ret0, _ := ret[0].([]byte)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// MarshalRowChangedEvent indicates an expected call of MarshalRowChangedEvent.
+func (mr *MockmarshallerMockRecorder) MarshalRowChangedEvent(event, handleKeyOnly, claimCheckFileName interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "MarshalRowChangedEvent", reflect.TypeOf((*Mockmarshaller)(nil).MarshalRowChangedEvent), event, handleKeyOnly, claimCheckFileName)
+}
+
+// Unmarshal mocks base method.
+func (m *Mockmarshaller) Unmarshal(data []byte, v any) error {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "Unmarshal", data, v)
+	ret0, _ := ret[0].(error)
+	return ret0
+}
+
+// Unmarshal indicates an expected call of Unmarshal.
+func (mr *MockmarshallerMockRecorder) Unmarshal(data, v interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "Unmarshal", reflect.TypeOf((*Mockmarshaller)(nil).Unmarshal), data, v)
+}
diff --git a/pkg/sink/mysql/config.go b/pkg/sink/mysql/config.go
index fadd9f7e1..344731ece 100644
--- a/pkg/sink/mysql/config.go
+++ b/pkg/sink/mysql/config.go
@@ -139,6 +139,7 @@ func NewConfig() *Config {
 		BatchDMLEnable:         defaultBatchDMLEnable,
 		MultiStmtEnable:        defaultMultiStmtEnable,
 		CachePrepStmts:         defaultCachePrepStmts,
+		SourceID:               config.DefaultTiDBSourceID,
 	}
 }
 
@@ -198,7 +199,20 @@ func (c *Config) Apply(
 	getMultiStmtEnable(urlParameter, &c.MultiStmtEnable)
 	getCachePrepStmts(urlParameter, &c.CachePrepStmts)
 	c.ForceReplicate = replicaConfig.ForceReplicate
-	c.SourceID = replicaConfig.Sink.TiDBSourceID
+
+	// Note(dongmen): The TiDBSourceID should never be 0 here, but we have found that
+	// in some problematic cases, the TiDBSourceID is 0 since something went wrong in the
+	// configuration process. So we need to check it here again.
+	// We do this is because it can cause the data to be inconsistent if the TiDBSourceID is 0
+	// in BDR Mode cluster.
+	if replicaConfig.Sink.TiDBSourceID == 0 {
+		log.Error("The TiDB source ID should never be set to 0. Please report it as a bug. The default value will be used: 1.",
+			zap.Uint64("tidbSourceID", replicaConfig.Sink.TiDBSourceID))
+		c.SourceID = config.DefaultTiDBSourceID
+	} else {
+		c.SourceID = replicaConfig.Sink.TiDBSourceID
+		log.Info("TiDB source ID is set", zap.Uint64("sourceID", c.SourceID))
+	}
 
 	return nil
 }
diff --git a/scripts/generate-mock.sh b/scripts/generate-mock.sh
index ffd9aff64..6cde7cc77 100755
--- a/scripts/generate-mock.sh
+++ b/scripts/generate-mock.sh
@@ -42,6 +42,7 @@ fi
 "$MOCKGEN" -source pkg/sink/kafka/v2/client.go -destination pkg/sink/kafka/v2/mock/client_mock.go
 "$MOCKGEN" -source pkg/sink/kafka/v2/gssapi.go -destination pkg/sink/kafka/v2/mock/gssapi_mock.go
 "$MOCKGEN" -source pkg/sink/kafka/v2/writer.go -destination pkg/sink/kafka/v2/mock/writer_mock.go
+"$MOCKGEN" -source pkg/sink/codec/simple/marshaller.go -destination pkg/sink/codec/simple/mock/marshaller.go
 
 # DM mock
 "$MOCKGEN" -package pbmock -destination dm/pbmock/dmmaster.go github.com/pingcap/tiflow/dm/pb MasterClient,MasterServer


Test patch:
diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go
index 0c864e508..3b6c68b36 100644
--- a/cdc/api/v2/model_test.go
+++ b/cdc/api/v2/model_test.go
@@ -19,9 +19,9 @@ import (
 	"testing"
 	"time"
 
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	filter "github.com/pingcap/tidb/pkg/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/redo"
 	"github.com/pingcap/tiflow/pkg/util"
diff --git a/cdc/async/async_initializer_test.go b/cdc/async/async_initializer_test.go
new file mode 100644
index 000000000..c6e42514b
--- /dev/null
+++ b/cdc/async/async_initializer_test.go
@@ -0,0 +1,139 @@
+// Copyright 2024 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 async
+
+import (
+	"context"
+	"sync"
+	"testing"
+	"time"
+
+	"github.com/pingcap/errors"
+	cdcContext "github.com/pingcap/tiflow/pkg/context"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/atomic"
+)
+
+type fakePool struct {
+	f           func()
+	submitErr   error
+	submitTimes int
+}
+
+func (f *fakePool) Go(_ context.Context, fn func()) error {
+	f.f = fn
+	f.submitTimes++
+	return f.submitErr
+}
+
+func (f *fakePool) Run(_ context.Context) error {
+	return nil
+}
+
+func TestTryInitialize(t *testing.T) {
+	initializer := NewInitializer()
+	pool := &cdcContext.NonAsyncPool{}
+	initialized, err := initializer.TryInitialize(cdcContext.NewContext4Test(context.Background(), false),
+		func(ctx cdcContext.Context) error {
+			return nil
+		}, pool)
+	require.Nil(t, err)
+	require.True(t, initialized)
+	// Try to initialize again
+	initialized, err = initializer.TryInitialize(cdcContext.NewContext4Test(context.Background(),
+		false), func(ctx cdcContext.Context) error {
+		return nil
+	}, pool)
+	require.Nil(t, err)
+	require.True(t, initialized)
+	// init failed
+	initializer = NewInitializer()
+	initialized, err = initializer.TryInitialize(cdcContext.NewContext4Test(context.Background(), false),
+		func(ctx cdcContext.Context) error {
+			return errors.New("failed to init")
+		}, pool)
+	require.NotNil(t, err)
+	require.False(t, initializer.initialized.Load())
+	require.True(t, initializer.initializing.Load())
+	require.False(t, initialized)
+	initialized, err = initializer.TryInitialize(cdcContext.NewContext4Test(context.Background(), false),
+		func(ctx cdcContext.Context) error {
+			return errors.New("failed to init")
+		}, pool)
+	require.NotNil(t, err)
+	require.False(t, initializer.initialized.Load())
+	require.True(t, initializer.initializing.Load())
+	require.False(t, initialized)
+
+	// test submit error
+	initializer = NewInitializer()
+	initialized, err = initializer.TryInitialize(cdcContext.NewContext4Test(context.Background(), false),
+		func(ctx cdcContext.Context) error {
+			return nil
+		}, &fakePool{submitErr: errors.New("submit error")})
+	require.NotNil(t, err)
+	require.False(t, initialized)
+	require.False(t, initializer.initialized.Load())
+	require.True(t, initializer.initializing.Load())
+}
+
+func TestTerminate(t *testing.T) {
+	initializer := NewInitializer()
+	pool := &cdcContext.NonAsyncPool{}
+	initialized, err := initializer.TryInitialize(cdcContext.NewContext4Test(context.Background(), false),
+
+		func(ctx cdcContext.Context) error {
+			return nil
+		}, pool)
+	require.Nil(t, err)
+	require.True(t, initialized)
+	initializer.Terminate()
+	require.False(t, initializer.initialized.Load())
+	require.False(t, initializer.initializing.Load())
+
+	// test submit error
+	initializer = NewInitializer()
+	fpool := &fakePool{}
+	initialized, err = initializer.TryInitialize(cdcContext.NewContext4Test(context.Background(), false),
+		func(ctx cdcContext.Context) error {
+			return nil
+		}, fpool)
+	require.Nil(t, err)
+	require.False(t, initialized)
+	require.True(t, initializer.initializing.Load())
+	require.Equal(t, 1, fpool.submitTimes)
+
+	initialized, err = initializer.TryInitialize(cdcContext.NewContext4Test(context.Background(), false),
+		func(ctx cdcContext.Context) error {
+			return nil
+		}, fpool)
+	require.Nil(t, err)
+	require.False(t, initialized)
+	require.True(t, initializer.initializing.Load())
+	require.Equal(t, 1, fpool.submitTimes)
+
+	wg := sync.WaitGroup{}
+	wg.Add(1)
+	terminated := atomic.NewInt32(1)
+	go func() {
+		defer wg.Done()
+		initializer.Terminate()
+		require.Equal(t, int32(2), terminated.Swap(3))
+	}()
+	require.Equal(t, int32(1), terminated.Swap(2))
+	time.Sleep(1 * time.Second)
+	fpool.f()
+	wg.Wait()
+	require.Equal(t, int32(3), terminated.Load())
+}
diff --git a/cdc/owner/changefeed_test.go b/cdc/owner/changefeed_test.go
index 642b10002..e59ccabcf 100644
--- a/cdc/owner/changefeed_test.go
+++ b/cdc/owner/changefeed_test.go
@@ -575,7 +575,7 @@ func testChangefeedReleaseResource(
 	// initialize
 	cf.Tick(ctx, captures)
 	tester.MustApplyPatches()
-	require.Equal(t, cf.initialized, expectedInitialized)
+	require.Equal(t, cf.initialized.Load(), expectedInitialized)
 
 	// remove changefeed from state manager by admin job
 	cf.feedStateManager.PushAdminJob(&model.AdminJob{
diff --git a/cdc/processor/processor_test.go b/cdc/processor/processor_test.go
index b25073ba1..b1575309b 100644
--- a/cdc/processor/processor_test.go
+++ b/cdc/processor/processor_test.go
@@ -18,11 +18,13 @@ import (
 	"encoding/json"
 	"fmt"
 	"math"
+	"os"
 	"sync/atomic"
 	"testing"
 
 	"github.com/pingcap/errors"
 	"github.com/pingcap/failpoint"
+	"github.com/pingcap/tiflow/cdc/async"
 	"github.com/pingcap/tiflow/cdc/entry"
 	"github.com/pingcap/tiflow/cdc/entry/schema"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -62,7 +64,7 @@ func newProcessor4Test(
 	// Some cases want to send errors to the processor without initializing it.
 	p.sinkManager.errors = make(chan error, 16)
 	p.lazyInit = func(ctx cdcContext.Context) error {
-		if p.initialized {
+		if p.initialized.Load() {
 			return nil
 		}
 
@@ -101,9 +103,10 @@ func newProcessor4Test(
 		// otherwise the sinkManager will not receive the resolvedTs.
 		p.sourceManager.r.OnResolve(p.sinkManager.r.UpdateReceivedSorterResolvedTs)
 
-		p.initialized = true
+		p.initialized.Store(true)
 		return nil
 	}
+	p.initializer = async.NewInitializer()
 
 	p.ddlHandler.r = &ddlHandler{
 		schemaStorage: &mockSchemaStorage{t: t, resolvedTs: math.MaxUint64},
@@ -221,6 +224,7 @@ func TestTableExecutorAddingTableIndirectly(t *testing.T) {
 
 	// init tick
 	checkChangefeedNormal(changefeed)
+	require.Nil(t, p.lazyInit(ctx))
 	createTaskPosition(changefeed, p.captureInfo)
 	tester.MustApplyPatches()
 	changefeed.PatchStatus(func(status *model.ChangeFeedStatus) (*model.ChangeFeedStatus, bool, error) {
@@ -309,6 +313,7 @@ func TestTableExecutorAddingTableIndirectlyWithRedoEnabled(t *testing.T) {
 
 	// init tick
 	checkChangefeedNormal(changefeed)
+	require.Nil(t, p.lazyInit(ctx))
 	createTaskPosition(changefeed, p.captureInfo)
 	tester.MustApplyPatches()
 	changefeed.PatchStatus(func(status *model.ChangeFeedStatus) (*model.ChangeFeedStatus, bool, error) {
@@ -406,6 +411,7 @@ func TestProcessorError(t *testing.T) {
 	p, tester, changefeed := initProcessor4Test(ctx, t, &liveness, false)
 
 	// init tick
+	require.Nil(t, p.lazyInit(ctx))
 	err, _ := p.Tick(ctx, changefeed.Info, changefeed.Status)
 	require.Nil(t, err)
 	createTaskPosition(changefeed, p.captureInfo)
@@ -430,6 +436,7 @@ func TestProcessorError(t *testing.T) {
 
 	p, tester, changefeed = initProcessor4Test(ctx, t, &liveness, false)
 	// init tick
+	require.Nil(t, p.lazyInit(ctx))
 	err, _ = p.Tick(ctx, changefeed.Info, changefeed.Status)
 	require.Nil(t, err)
 	createTaskPosition(changefeed, p.captureInfo)
@@ -452,9 +459,9 @@ func TestProcessorExit(t *testing.T) {
 	ctx := cdcContext.NewBackendContext4Test(true)
 	liveness := model.LivenessCaptureAlive
 	p, tester, changefeed := initProcessor4Test(ctx, t, &liveness, false)
-	// var err error
 	// init tick
 	checkChangefeedNormal(changefeed)
+	require.Nil(t, p.lazyInit(ctx))
 	createTaskPosition(changefeed, p.captureInfo)
 	tester.MustApplyPatches()
 
@@ -479,6 +486,7 @@ func TestProcessorClose(t *testing.T) {
 	p, tester, changefeed := initProcessor4Test(ctx, t, &liveness, false)
 	// init tick
 	checkChangefeedNormal(changefeed)
+	require.Nil(t, p.lazyInit(ctx))
 	createTaskPosition(changefeed, p.captureInfo)
 	tester.MustApplyPatches()
 
@@ -518,6 +526,7 @@ func TestProcessorClose(t *testing.T) {
 	p, tester, changefeed = initProcessor4Test(ctx, t, &liveness, false)
 	// init tick
 	checkChangefeedNormal(changefeed)
+	require.Nil(t, p.lazyInit(ctx))
 	createTaskPosition(changefeed, p.captureInfo)
 	tester.MustApplyPatches()
 
@@ -563,6 +572,7 @@ func TestPositionDeleted(t *testing.T) {
 	p, tester, changefeed := initProcessor4Test(ctx, t, &liveness, false)
 	// init tick
 	checkChangefeedNormal(changefeed)
+	require.Nil(t, p.lazyInit(ctx))
 	createTaskPosition(changefeed, p.captureInfo)
 	tester.MustApplyPatches()
 	require.Contains(t, changefeed.TaskPositions, p.captureInfo.ID)
@@ -608,6 +618,7 @@ func TestSchemaGC(t *testing.T) {
 	var err error
 	// init tick
 	checkChangefeedNormal(changefeed)
+	require.Nil(t, p.lazyInit(ctx))
 	createTaskPosition(changefeed, p.captureInfo)
 	tester.MustApplyPatches()
 
@@ -674,6 +685,7 @@ func TestUpdateBarrierTs(t *testing.T) {
 
 	// init tick
 	checkChangefeedNormal(changefeed)
+	require.Nil(t, p.lazyInit(ctx))
 	createTaskPosition(changefeed, p.captureInfo)
 	tester.MustApplyPatches()
 	require.Contains(t, changefeed.TaskPositions, p.captureInfo.ID)
@@ -721,6 +733,7 @@ func TestProcessorLiveness(t *testing.T) {
 	p, tester, changefeed := initProcessor4Test(ctx, t, &liveness, false)
 
 	// First tick for creating position.
+	require.Nil(t, p.lazyInit(ctx))
 	err, _ := p.Tick(ctx, changefeed.Info, changefeed.Status)
 	require.Nil(t, err)
 	tester.MustApplyPatches()
@@ -754,6 +767,7 @@ func TestProcessorDostNotStuckInInit(t *testing.T) {
 	ctx := cdcContext.NewBackendContext4Test(true)
 	liveness := model.LivenessCaptureAlive
 	p, tester, changefeed := initProcessor4Test(ctx, t, &liveness, false)
+	require.Nil(t, p.lazyInit(ctx))
 
 	// First tick for creating position.
 	err, _ := p.Tick(ctx, changefeed.Info, changefeed.Status)
@@ -771,3 +785,9 @@ func TestProcessorDostNotStuckInInit(t *testing.T) {
 	require.Nil(t, p.Close())
 	tester.MustApplyPatches()
 }
+
+func TestProcessorNotInitialized(t *testing.T) {
+	liveness := model.LivenessCaptureAlive
+	p, _, _ := initProcessor4Test(cdcContext.NewContext4Test(context.Background(), true), t, &liveness, false)
+	require.Nil(t, p.WriteDebugInfo(os.Stdout))
+}
diff --git a/cdc/processor/sinkmanager/manager_test_helper.go b/cdc/processor/sinkmanager/manager_test_helper.go
index 7726dbd5b..73163921d 100644
--- a/cdc/processor/sinkmanager/manager_test_helper.go
+++ b/cdc/processor/sinkmanager/manager_test_helper.go
@@ -70,7 +70,8 @@ func CreateManagerWithMemEngine(
 	go func() { handleError(sourceManager.Run(ctx)) }()
 	sourceManager.WaitForReady(ctx)
 
-	sinkManager := New(changefeedID, changefeedInfo, up, schemaStorage, nil, sourceManager)
+	sinkManager := New(changefeedID, changefeedInfo.SinkURI,
+		changefeedInfo.Config, up, schemaStorage, nil, sourceManager)
 	go func() { handleError(sinkManager.Run(ctx)) }()
 	sinkManager.WaitForReady(ctx)
 
@@ -90,6 +91,7 @@ func NewManagerWithMemEngine(
 	mg := &entry.MockMountGroup{}
 	schemaStorage := &entry.MockSchemaStorage{Resolved: math.MaxUint64}
 	sourceManager := sourcemanager.NewForTest(changefeedID, up, mg, sortEngine, false)
-	sinkManager := New(changefeedID, changefeedInfo, up, schemaStorage, redoMgr, sourceManager)
+	sinkManager := New(changefeedID, changefeedInfo.SinkURI,
+		changefeedInfo.Config, up, schemaStorage, redoMgr, sourceManager)
 	return sinkManager, sourceManager, sortEngine
 }
diff --git a/cdc/redo/reader/reader_test.go b/cdc/redo/reader/reader_test.go
index 1a40f055b..8cf9627e7 100644
--- a/cdc/redo/reader/reader_test.go
+++ b/cdc/redo/reader/reader_test.go
@@ -244,3 +244,515 @@ func genMetaFile(t *testing.T, dir string, meta *common.LogMeta) {
 	_, err = f.Write(data)
 	require.Nil(t, err)
 }
+
+func TestLogHeapLess(t *testing.T) {
+	tests := []struct {
+		name   string
+		h      logHeap
+		i      int
+		j      int
+		expect bool
+	}{
+		{
+			name: "Delete before Update",
+			h: logHeap{
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 2,
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 2,
+									},
+								},
+								Columns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 3,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			i:      0,
+			j:      1,
+			expect: true,
+		},
+		{
+			name: "Update before Insert",
+			h: logHeap{
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 2,
+									},
+								},
+								Columns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 3,
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								Columns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 1,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			i:      0,
+			j:      1,
+			expect: true,
+		},
+		{
+			name: "Update before Delete",
+			h: logHeap{
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 2,
+									},
+								},
+								Columns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 3,
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 1,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			i:      0,
+			j:      1,
+			expect: false,
+		},
+		{
+			name: "Update before Update",
+			h: logHeap{
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 2,
+									},
+								},
+								Columns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 3,
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 1,
+									},
+								},
+								Columns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 4,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			i:      0,
+			j:      1,
+			expect: true,
+		},
+		{
+			name: "Delete before Insert",
+			h: logHeap{
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 1,
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								Columns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 3,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			i:      0,
+			j:      1,
+			expect: true,
+		},
+		{
+			name: "Same type of operations, different commit ts",
+			h: logHeap{
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+							},
+						},
+					},
+				},
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 200,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+							},
+						},
+					},
+				},
+			},
+			i:      0,
+			j:      1,
+			expect: true,
+		},
+		{
+			name: "Same type of operations, same commit ts, different startTs",
+			h: logHeap{
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								StartTs:  80,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+							},
+						},
+					},
+				},
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								StartTs:  90,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+							},
+						},
+					},
+				},
+			},
+			i:      0,
+			j:      1,
+			expect: true,
+		},
+		{
+			name: "Same type of operations, same commit ts",
+			h: logHeap{
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 2,
+									},
+								},
+								Columns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 3,
+									},
+								},
+							},
+						},
+					},
+				},
+				{
+					data: &model.RedoLog{
+						Type: model.RedoLogTypeRow,
+						RedoRow: model.RedoRowChangedEvent{
+							Row: &model.RowChangedEvent{
+								CommitTs: 100,
+								Table: &model.TableName{
+									Schema:      "test",
+									Table:       "table",
+									TableID:     1,
+									IsPartition: false,
+								},
+								PreColumns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 1,
+									},
+								},
+								Columns: []*model.Column{
+									{
+										Name:  "col-1",
+										Value: 1,
+									}, {
+										Name:  "col-2",
+										Value: 3,
+									},
+								},
+							},
+						},
+					},
+				},
+			},
+			i:      0,
+			j:      1,
+			expect: true,
+		},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			if got := tt.h.Less(tt.i, tt.j); got != tt.expect {
+				t.Errorf("logHeap.Less() = %v, want %v", got, tt.expect)
+			}
+		})
+	}
+}
diff --git a/cdc/redo/writer/memory/mem_log_writer_test.go b/cdc/redo/writer/memory/mem_log_writer_test.go
index bba18c806..64fbe9b2d 100644
--- a/cdc/redo/writer/memory/mem_log_writer_test.go
+++ b/cdc/redo/writer/memory/mem_log_writer_test.go
@@ -88,11 +88,9 @@ func testWriteEvents(t *testing.T, events []writer.RedoEvent) {
 	require.NoError(t, err)
 
 	require.ErrorIs(t, lw.Close(), context.Canceled)
-	require.Eventually(t, func() bool {
-		err = lw.WriteEvents(ctx, events...)
-		return err != nil
-	}, 2*time.Second, 10*time.Millisecond)
-	require.ErrorContains(t, err, "redo log writer stopped")
+
+	err = lw.WriteEvents(ctx, events...)
+	require.NoError(t, err)
 	err = lw.FlushLog(ctx)
-	require.ErrorContains(t, err, "redo log writer stopped")
+	require.NoError(t, err)
 }
diff --git a/cdc/scheduler/internal/v3/coordinator_test.go b/cdc/scheduler/internal/v3/coordinator_test.go
index 8caacc30a..3e3f2460d 100644
--- a/cdc/scheduler/internal/v3/coordinator_test.go
+++ b/cdc/scheduler/internal/v3/coordinator_test.go
@@ -29,7 +29,6 @@ import (
 	"github.com/pingcap/tiflow/cdc/scheduler/internal/v3/transport"
 	"github.com/pingcap/tiflow/cdc/scheduler/schedulepb"
 	"github.com/pingcap/tiflow/pkg/config"
-	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/leakutil"
 	"github.com/pingcap/tiflow/pkg/spanz"
 	"github.com/pingcap/tiflow/pkg/version"
@@ -403,8 +402,8 @@ func TestCoordinatorDrainCapture(t *testing.T) {
 	coord.captureM.SetInitializedForTests(true)
 	coord.captureM.Captures["a"] = &member.CaptureStatus{State: member.CaptureStateUninitialized}
 	count, err := coord.DrainCapture("a")
-	require.ErrorIs(t, err, cerror.ErrSchedulerRequestFailed)
-	require.Equal(t, 0, count)
+	require.Nil(t, err)
+	require.Equal(t, 1, count)
 
 	coord.captureM.Captures["a"] = &member.CaptureStatus{State: member.CaptureStateInitialized}
 	coord.replicationM = replication.NewReplicationManager(10, model.ChangeFeedID{})
diff --git a/dm/config/source_config_test.go b/dm/config/source_config_test.go
index fc4234a89..0eeb5fc9a 100644
--- a/dm/config/source_config_test.go
+++ b/dm/config/source_config_test.go
@@ -26,9 +26,9 @@ import (
 
 	"github.com/DATA-DOG/go-sqlmock"
 	"github.com/go-mysql-org/go-mysql/mysql"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/dm/pkg/conn"
 	tcontext "github.com/pingcap/tiflow/dm/pkg/context"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/stretchr/testify/require"
 )
 
diff --git a/dm/config/task_converters_test.go b/dm/config/task_converters_test.go
index 223558c75..597abe3cb 100644
--- a/dm/config/task_converters_test.go
+++ b/dm/config/task_converters_test.go
@@ -17,12 +17,12 @@ import (
 	"testing"
 
 	"github.com/pingcap/check"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/openapi"
 	"github.com/pingcap/tiflow/dm/openapi/fixtures"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/stretchr/testify/require"
 )
 
diff --git a/dm/config/task_test.go b/dm/config/task_test.go
index 4afc47751..0d68c3054 100644
--- a/dm/config/task_test.go
+++ b/dm/config/task_test.go
@@ -22,12 +22,12 @@ import (
 	"testing"
 
 	"github.com/coreos/go-semver/semver"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	router "github.com/pingcap/tidb/pkg/util/table-router"
 	"github.com/pingcap/tiflow/dm/config/dbconfig"
 	"github.com/pingcap/tiflow/dm/config/security"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/stretchr/testify/require"
 )
 
diff --git a/dm/master/bootstrap_test.go b/dm/master/bootstrap_test.go
index b09b9fbfc..25e2c26f7 100644
--- a/dm/master/bootstrap_test.go
+++ b/dm/master/bootstrap_test.go
@@ -24,7 +24,6 @@ import (
 
 	"github.com/golang/mock/gomock"
 	. "github.com/pingcap/check"
-	filter "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/dm/config"
 	"github.com/pingcap/tiflow/dm/master/workerrpc"
 	"github.com/pingcap/tiflow/dm/pb"
@@ -33,6 +32,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/ha"
 	"github.com/pingcap/tiflow/dm/pkg/log"
 	"github.com/pingcap/tiflow/dm/pkg/terror"
+	filter "github.com/pingcap/tiflow/pkg/binlog-filter"
 	clientv3 "go.etcd.io/etcd/client/v3"
 	"go.etcd.io/etcd/tests/v3/integration"
 )
diff --git a/dm/syncer/filter_test.go b/dm/syncer/filter_test.go
index d166a4fbc..579b14ed4 100644
--- a/dm/syncer/filter_test.go
+++ b/dm/syncer/filter_test.go
@@ -21,7 +21,6 @@ import (
 	"github.com/go-mysql-org/go-mysql/mysql"
 	"github.com/go-mysql-org/go-mysql/replication"
 	. "github.com/pingcap/check"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/util/filter"
 	"github.com/pingcap/tiflow/dm/config"
@@ -32,6 +31,7 @@ import (
 	"github.com/pingcap/tiflow/dm/pkg/schema"
 	"github.com/pingcap/tiflow/dm/pkg/utils"
 	"github.com/pingcap/tiflow/dm/syncer/dbconn"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 )
 
 type testFilterSuite struct {
diff --git a/dm/syncer/syncer_test.go b/dm/syncer/syncer_test.go
index 624ae290e..a592ad115 100644
--- a/dm/syncer/syncer_test.go
+++ b/dm/syncer/syncer_test.go
@@ -31,8 +31,6 @@ import (
 	_ "github.com/go-sql-driver/mysql"
 	. "github.com/pingcap/check"
 	"github.com/pingcap/failpoint"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
-	cm "github.com/pingcap/tidb-tools/pkg/column-mapping"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	"github.com/pingcap/tidb/pkg/parser"
 	"github.com/pingcap/tidb/pkg/parser/ast"
@@ -58,6 +56,8 @@ import (
 	"github.com/pingcap/tiflow/dm/syncer/binlogstream"
 	"github.com/pingcap/tiflow/dm/syncer/dbconn"
 	"github.com/pingcap/tiflow/dm/syncer/metrics"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
+	cm "github.com/pingcap/tiflow/pkg/column-mapping"
 	"github.com/pingcap/tiflow/pkg/errorutil"
 	"github.com/pingcap/tiflow/pkg/sqlmodel"
 	"github.com/stretchr/testify/require"
diff --git a/pkg/binlog-filter/filter_test.go b/pkg/binlog-filter/filter_test.go
new file mode 100644
index 000000000..6d4e97782
--- /dev/null
+++ b/pkg/binlog-filter/filter_test.go
@@ -0,0 +1,332 @@
+// Copyright 2018 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 filter
+
+import (
+	"testing"
+
+	"github.com/pingcap/errors"
+	selector "github.com/pingcap/tidb/pkg/util/table-rule-selector"
+	"github.com/stretchr/testify/require"
+)
+
+func TestFilter(t *testing.T) {
+	rules := []*BinlogEventRule{
+		{"Test_1_*", "abc*", []EventType{DeleteEvent, InsertEvent, CreateIndex, DropIndex, DropView}, []string{"^DROP\\s+PROCEDURE", "^CREATE\\s+PROCEDURE"}, nil, Ignore},
+		{"xxx_*", "abc_*", []EventType{AllDML, NoneDDL}, nil, nil, Ignore},
+		{"yyy_*", "abc_*", []EventType{EventType("ALL DML")}, nil, nil, Do},
+		{"Test_1_*", "abc*", []EventType{"wrong event"}, []string{"^DROP\\s+PROCEDURE", "^CREATE\\s+PROCEDURE"}, nil, Ignore},
+		{"cdc", "t1", []EventType{RebaseAutoID}, nil, nil, Ignore},
+	}
+
+	cases := []struct {
+		schema, table string
+		event         EventType
+		sql           string
+		action        ActionType
+	}{
+		{"test_1_a", "abc1", DeleteEvent, "", Ignore},
+		{"test_1_a", "abc1", InsertEvent, "", Ignore},
+		{"test_1_a", "abc1", UpdateEvent, "", Do},
+		{"test_1_a", "abc1", CreateIndex, "", Ignore},
+		{"test_1_a", "abc1", RenameTable, "", Do},
+		{"test_1_a", "abc1", NullEvent, "drop procedure abc", Ignore},
+		{"test_1_a", "abc1", NullEvent, "create procedure abc", Ignore},
+		{"test_1_a", "abc1", NullEvent, "create function abc", Do},
+		{"xxx_1", "abc_1", NullEvent, "create function abc", Do},
+		{"xxx_1", "abc_1", InsertEvent, "", Ignore},
+		{"xxx_1", "abc_1", CreateIndex, "", Do},
+		{"yyy_1", "abc_1", InsertEvent, "", Do},
+		{"yyy_1", "abc_1", CreateIndex, "", Ignore},
+		{"test_1_a", "abc1", DropView, "", Ignore},
+		{"cdc", "t1", RebaseAutoID, "", Ignore},
+	}
+
+	// initial binlog event filter
+	filter, err := NewBinlogEvent(false, rules)
+	require.NoError(t, err)
+
+	// insert duplicate rules
+	for _, rule := range rules {
+		err = filter.AddRule(rule)
+		require.Error(t, err)
+	}
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// update rules
+	rules[0].Events = []EventType{}
+	rules[1].Action = Do
+	rules[2].Events = []EventType{"ALL DDL"}
+	rules = rules[:3]
+	for _, rule := range rules {
+		err = filter.UpdateRule(rule)
+		require.NoError(t, err)
+	}
+
+	cases[0].action = Do      // delete
+	cases[1].action = Do      // insert
+	cases[3].action = Do      // create index
+	cases[9].action = Do      // match all event and insert
+	cases[10].action = Ignore // match none event and create index
+	cases[11].action = Ignore // no match
+	cases[12].action = Do     // match all ddl
+	cases[13].action = Do     // match all ddl
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// test multiple rules
+	rule := &BinlogEventRule{"test_*", "ab*", []EventType{InsertEvent, AllDDL}, []string{"^DROP\\s+PROCEDURE"}, nil, Do}
+	err = filter.AddRule(rule)
+	require.NoError(t, err)
+	cases[0].action = Ignore // delete
+	cases[2].action = Ignore // update
+	cases[4].action = Do     // rename table
+	cases[7].action = Ignore // create function
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// remove rule
+	err = filter.RemoveRule(rules[0])
+	require.NoError(t, err)
+	// remove not existing rule
+	err = filter.RemoveRule(rules[0])
+	require.Error(t, err)
+	cases[3].action = Do // create index
+	cases[5].action = Do // drop procedure
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// mismatched
+	action, err := filter.Filter("xxx_a", "", InsertEvent, "")
+	require.NoError(t, err)
+	require.Equal(t, Do, action)
+
+	// invalid rule
+	err = filter.Selector.Insert("test_1_*", "abc*", "error", selector.Insert)
+	require.NoError(t, err)
+	_, err = filter.Filter("test_1_a", "abc", InsertEvent, "")
+	require.Error(t, err)
+}
+
+func TestCaseSensitive(t *testing.T) {
+	// we test case insensitive in TestFilter
+	rules := []*BinlogEventRule{
+		{"Test_1_*", "abc*", []EventType{DeleteEvent, InsertEvent, CreateIndex, DropIndex}, []string{"^DROP\\s+PROCEDURE", "^CREATE\\s+PROCEDURE"}, nil, Ignore},
+		{"xxx_*", "abc_*", []EventType{AllDML, NoneDDL}, nil, nil, Ignore},
+	}
+
+	cases := []struct {
+		schema, table string
+		event         EventType
+		sql           string
+		action        ActionType
+	}{
+		{"test_1_a", "abc1", DeleteEvent, "", Do},
+		{"test_1_a", "abc1", InsertEvent, "", Do},
+		{"test_1_a", "abc1", UpdateEvent, "", Do},
+		{"test_1_a", "abc1", CreateIndex, "", Do},
+		{"test_1_a", "abc1", RenameTable, "", Do},
+		{"test_1_a", "abc1", NullEvent, "drop procedure abc", Do},
+		{"test_1_a", "abc1", NullEvent, "create procedure abc", Do},
+		{"test_1_a", "abc1", NullEvent, "create function abc", Do},
+		{"xxx_1", "abc_1", NullEvent, "create function abc", Do},
+		{"xxx_1", "abc_1", InsertEvent, "", Ignore},
+		{"xxx_1", "abc_1", CreateIndex, "", Do},
+	}
+
+	// initial binlog event filter
+	filter, err := NewBinlogEvent(true, rules)
+	require.NoError(t, err)
+
+	// insert duplicate rules
+	for _, rule := range rules {
+		err = filter.AddRule(rule)
+		require.Error(t, err)
+	}
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, cs.event, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+}
+
+func TestGlobalFilter(t *testing.T) {
+	schemaRule := &BinlogEventRule{
+		SchemaPattern: "*",
+		SQLPattern:    []string{"^FLUSH"},
+		Action:        Ignore,
+	}
+	tableRule := &BinlogEventRule{
+		SchemaPattern: "*",
+		TablePattern:  "*",
+		SQLPattern:    []string{"^FLUSH"},
+		Action:        Ignore,
+	}
+
+	cases := []struct {
+		schema string
+		table  string
+		sql    string
+		action ActionType
+	}{
+		{
+			schema: "db",
+			table:  "tbl",
+			sql:    "FLUSH ENGINE LOGS",
+			action: Ignore,
+		},
+		{
+			schema: "db",
+			table:  "",
+			sql:    "FLUSH ENGINE LOGS",
+			action: Ignore,
+		},
+		{
+			schema: "",
+			table:  "tbl",
+			sql:    "FLUSH ENGINE LOGS",
+			action: Ignore,
+		},
+		{
+			schema: "",
+			table:  "",
+			sql:    "FLUSH ENGINE LOGS",
+			action: Ignore,
+		},
+	}
+
+	// initial binlog event filter with schema rule
+	filter, err := NewBinlogEvent(false, []*BinlogEventRule{schemaRule})
+	require.NoError(t, err)
+
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, NullEvent, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+
+	// remove schema rule
+	err = filter.RemoveRule(schemaRule)
+	require.NoError(t, err)
+
+	// add table rule
+	err = filter.AddRule(tableRule)
+	require.NoError(t, err)
+
+	for _, cs := range cases {
+		action, err := filter.Filter(cs.schema, cs.table, NullEvent, cs.sql)
+		require.NoError(t, err)
+		require.Equal(t, cs.action, action)
+	}
+}
+
+func TestToEventType(t *testing.T) {
+	cases := []struct {
+		eventStr string
+		event    EventType
+		err      error
+	}{
+		{"", NullEvent, nil},
+		{"insert", InsertEvent, nil},
+		{"Insert", InsertEvent, nil},
+		{"update", UpdateEvent, nil},
+		{"UPDATE", UpdateEvent, nil},
+		{"delete", DeleteEvent, nil},
+		{"create", NullEvent, errors.NotValidf("event type %s", "create")},
+		{"create schema", CreateDatabase, nil},
+		{"create SCHEMA", CreateDatabase, nil},
+		{"create database", CreateDatabase, nil},
+		{"drop schema", DropDatabase, nil},
+		{"drop Schema", DropDatabase, nil},
+		{"drop database", DropDatabase, nil},
+		{"alter database", AlterDatabase, nil},
+		{"alter schema", AlterDatabase, nil},
+		{"create index", CreateIndex, nil},
+		{"add table partition", AddTablePartition, nil},
+		{"drop taBle partition", DropTablePartition, nil},
+		{"truncate tablE parTition", TruncateTablePartition, nil},
+		{"rebase auto id", RebaseAutoID, nil},
+		{"xxx", NullEvent, errors.NotValidf("event type %s", "xxx")},
+		{"I don't know", NullEvent, errors.NotValidf("event type %s", "I don't know")},
+	}
+
+	for _, cs := range cases {
+		event, err := toEventType(cs.eventStr)
+		require.Equal(t, cs.event, event)
+		if err != nil {
+			require.ErrorContains(t, err, cs.err.Error())
+		} else {
+			require.NoError(t, err)
+		}
+	}
+}
+
+func TestClassifyEvent(t *testing.T) {
+	cases := []struct {
+		event    EventType
+		evenType EventType
+		err      error
+	}{
+		{NullEvent, NullEvent, nil},
+		// dml
+		{InsertEvent, dml, nil},
+		{UpdateEvent, dml, nil},
+		{DeleteEvent, dml, nil},
+		// ddl
+		{CreateDatabase, ddl, nil},
+		{CreateSchema, ddl, nil},
+		{DropDatabase, incompatibleDDL, nil},
+		{DropSchema, incompatibleDDL, nil},
+		{AlterSchema, ddl, nil},
+		{CreateTable, ddl, nil},
+		{DropTable, incompatibleDDL, nil},
+		{TruncateTable, incompatibleDDL, nil},
+		{RenameTable, incompatibleDDL, nil},
+		{CreateIndex, ddl, nil},
+		{DropIndex, incompatibleDDL, nil},
+		{CreateView, ddl, nil},
+		{DropView, ddl, nil},
+		{AlterTable, ddl, nil},
+		{AddTablePartition, ddl, nil},
+		{DropTablePartition, incompatibleDDL, nil},
+		{RebaseAutoID, incompatibleDDL, nil},
+		{TruncateTablePartition, incompatibleDDL, nil},
+		{"create", NullEvent, errors.NotValidf("event type %s", "create")},
+		{EventType("xxx"), NullEvent, errors.NotValidf("event type %s", "xxx")},
+		{EventType("I don't know"), NullEvent, errors.NotValidf("event type %s", "I don't know")},
+	}
+
+	for _, cs := range cases {
+		et, err := ClassifyEvent(cs.event)
+		require.Equal(t, cs.evenType, et)
+		if err != nil {
+			require.ErrorContains(t, err, cs.err.Error())
+		} else {
+			require.NoError(t, err)
+		}
+	}
+}
diff --git a/pkg/column-mapping/column_test.go b/pkg/column-mapping/column_test.go
new file mode 100644
index 000000000..6a8ddb98e
--- /dev/null
+++ b/pkg/column-mapping/column_test.go
@@ -0,0 +1,278 @@
+// Copyright 2018 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 column
+
+import (
+	"fmt"
+	"testing"
+
+	"github.com/stretchr/testify/require"
+)
+
+func TestRule(t *testing.T) {
+	// test invalid rules
+	inValidRule := &Rule{"test*", "abc*", "id", "id", "Error", nil, "xxx"}
+	require.Error(t, inValidRule.Valid())
+
+	inValidRule.TargetColumn = ""
+	require.Error(t, inValidRule.Valid())
+
+	inValidRule.Expression = AddPrefix
+	inValidRule.TargetColumn = "id"
+	require.Error(t, inValidRule.Valid())
+
+	inValidRule.Arguments = []string{"1"}
+	require.NoError(t, inValidRule.Valid())
+
+	inValidRule.Expression = PartitionID
+	require.Error(t, inValidRule.Valid())
+
+	inValidRule.Arguments = []string{"1", "test_", "t_"}
+	require.NoError(t, inValidRule.Valid())
+}
+
+func TestHandle(t *testing.T) {
+	rules := []*Rule{
+		{"Test*", "xxx*", "", "id", AddPrefix, []string{"instance_id:"}, "xx"},
+	}
+
+	// initial column mapping
+	m, err := NewMapping(false, rules)
+	require.NoError(t, err)
+	require.Len(t, m.cache.infos, 0)
+
+	// test add prefix, add suffix is similar
+	vals, poss, err := m.HandleRowValue("test", "xxx", []string{"age", "id"}, []interface{}{1, "1"})
+	require.NoError(t, err)
+	require.Equal(t, []interface{}{1, "instance_id:1"}, vals)
+	require.Equal(t, []int{-1, 1}, poss)
+
+	// test cache
+	vals, poss, err = m.HandleRowValue("test", "xxx", []string{"name"}, []interface{}{1, "1"})
+	require.NoError(t, err)
+	require.Equal(t, []interface{}{1, "instance_id:1"}, vals)
+	require.Equal(t, []int{-1, 1}, poss)
+
+	// test resetCache
+	m.resetCache()
+	_, _, err = m.HandleRowValue("test", "xxx", []string{"name"}, []interface{}{"1"})
+
+	require.Error(t, err)
+
+	// test DDL
+	_, _, err = m.HandleDDL("test", "xxx", []string{"id", "age"}, "create table xxx")
+	require.Error(t, err)
+
+	statement, poss, err := m.HandleDDL("abc", "xxx", []string{"id", "age"}, "create table xxx")
+	require.NoError(t, err)
+	require.Equal(t, "create table xxx", statement)
+	require.Nil(t, poss)
+}
+
+func TestQueryColumnInfo(t *testing.T) {
+	SetPartitionRule(4, 7, 8)
+	rules := []*Rule{
+		{"test*", "xxx*", "", "id", PartitionID, []string{"8", "test_", "xxx_"}, "xx"},
+	}
+
+	// initial column mapping
+	m, err := NewMapping(false, rules)
+	require.NoError(t, err)
+
+	// test mismatch
+	info, err := m.queryColumnInfo("test_2", "t_1", []string{"id", "name"})
+	require.NoError(t, err)
+	require.True(t, info.ignore)
+
+	// test matched
+	info, err = m.queryColumnInfo("test_2", "xxx_1", []string{"id", "name"})
+	require.NoError(t, err)
+	require.Equal(t, &mappingInfo{
+		sourcePosition: -1,
+		targetPosition: 0,
+		rule:           rules[0],
+		instanceID:     int64(8 << 59),
+		schemaID:       int64(2 << 52),
+		tableID:        int64(1 << 44),
+	}, info)
+
+	m.resetCache()
+	SetPartitionRule(0, 0, 3)
+	info, err = m.queryColumnInfo("test_2", "xxx_1", []string{"id", "name"})
+	require.NoError(t, err)
+	require.Equal(t, &mappingInfo{
+		sourcePosition: -1,
+		targetPosition: 0,
+		rule:           rules[0],
+		instanceID:     int64(0),
+		schemaID:       int64(0),
+		tableID:        int64(1 << 60),
+	}, info)
+}
+
+func TestSetPartitionRule(t *testing.T) {
+	SetPartitionRule(4, 7, 8)
+	require.Equal(t, 4, instanceIDBitSize)
+	require.Equal(t, 7, schemaIDBitSize)
+	require.Equal(t, 8, tableIDBitSize)
+	require.Equal(t, int64(1<<44), maxOriginID)
+
+	SetPartitionRule(0, 3, 4)
+	require.Equal(t, 0, instanceIDBitSize)
+	require.Equal(t, 3, schemaIDBitSize)
+	require.Equal(t, 4, tableIDBitSize)
+	require.Equal(t, int64(1<<56), maxOriginID)
+}
+
+func TestComputePartitionID(t *testing.T) {
+	SetPartitionRule(4, 7, 8)
+
+	rule := &Rule{
+		Arguments: []string{"test", "t"},
+	}
+	_, _, _, err := computePartitionID("test_1", "t_1", rule)
+	require.Error(t, err)
+	_, _, _, err = computePartitionID("test", "t", rule)
+	require.Error(t, err)
+
+	rule = &Rule{
+		Arguments: []string{"2", "test", "t", "_"},
+	}
+	instanceID, schemaID, tableID, err := computePartitionID("test_1", "t_1", rule)
+	require.NoError(t, err)
+	require.Equal(t, int64(2<<59), instanceID)
+	require.Equal(t, int64(1<<52), schemaID)
+	require.Equal(t, int64(1<<44), tableID)
+
+	// test default partition ID to zero
+	instanceID, schemaID, tableID, err = computePartitionID("test", "t_3", rule)
+	require.NoError(t, err)
+	require.Equal(t, int64(2<<59), instanceID)
+	require.Equal(t, int64(0), schemaID)
+	require.Equal(t, int64(3<<44), tableID)
+
+	instanceID, schemaID, tableID, err = computePartitionID("test_5", "t", rule)
+	require.NoError(t, err)
+	require.Equal(t, int64(2<<59), instanceID)
+	require.Equal(t, int64(5<<52), schemaID)
+	require.Equal(t, int64(0), tableID)
+
+	_, _, _, err = computePartitionID("unrelated", "t_6", rule)
+	require.ErrorContains(t, err, "test_ is not the prefix of unrelated")
+
+	_, _, _, err = computePartitionID("test", "x", rule)
+	require.ErrorContains(t, err, "t_ is not the prefix of x")
+
+	_, _, _, err = computePartitionID("test_0", "t_0xa", rule)
+	require.ErrorContains(t, err, "the suffix of 0xa can't be converted to int64")
+
+	_, _, _, err = computePartitionID("test_0", "t_", rule)
+	require.ErrorContains(t, err, "t_ is not the prefix of t_") // needs a better error message
+
+	_, _, _, err = computePartitionID("testx", "t_3", rule)
+	require.ErrorContains(t, err, "test_ is not the prefix of testx")
+
+	SetPartitionRule(4, 0, 8)
+	rule = &Rule{
+		Arguments: []string{"2", "test_", "t_", ""},
+	}
+	instanceID, schemaID, tableID, err = computePartitionID("test_1", "t_1", rule)
+	require.NoError(t, err)
+	require.Equal(t, int64(2<<59), instanceID)
+	require.Equal(t, int64(0), schemaID)
+	require.Equal(t, int64(1<<51), tableID)
+
+	instanceID, schemaID, tableID, err = computePartitionID("test_", "t_", rule)
+	require.NoError(t, err)
+	require.Equal(t, int64(2<<59), instanceID)
+	require.Equal(t, int64(0), schemaID)
+	require.Equal(t, int64(0), tableID)
+
+	// test ignore instance ID
+	SetPartitionRule(4, 7, 8)
+	rule = &Rule{
+		Arguments: []string{"", "test_", "t_", ""},
+	}
+	instanceID, schemaID, tableID, err = computePartitionID("test_1", "t_1", rule)
+	require.NoError(t, err)
+	require.Equal(t, int64(0), instanceID)
+	require.Equal(t, int64(1<<56), schemaID)
+	require.Equal(t, int64(1<<48), tableID)
+
+	// test ignore schema ID
+	rule = &Rule{
+		Arguments: []string{"2", "", "t_", ""},
+	}
+	instanceID, schemaID, tableID, err = computePartitionID("test_1", "t_1", rule)
+	require.NoError(t, err)
+	require.Equal(t, int64(2<<59), instanceID)
+	require.Equal(t, int64(0), schemaID)
+	require.Equal(t, int64(1<<51), tableID)
+
+	// test ignore schema ID
+	rule = &Rule{
+		Arguments: []string{"2", "test_", "", ""},
+	}
+	instanceID, schemaID, tableID, err = computePartitionID("test_1", "t_1", rule)
+	require.NoError(t, err)
+	require.Equal(t, int64(2<<59), instanceID)
+	require.Equal(t, int64(1<<52), schemaID)
+	require.Equal(t, int64(0), tableID)
+}
+
+func TestPartitionID(t *testing.T) {
+	SetPartitionRule(4, 7, 8)
+	info := &mappingInfo{
+		instanceID:     int64(2 << 59),
+		schemaID:       int64(1 << 52),
+		tableID:        int64(1 << 44),
+		targetPosition: 1,
+	}
+
+	// test wrong type
+	_, err := partitionID(info, []interface{}{1, "ha"})
+	require.Error(t, err)
+
+	// test exceed maxOriginID
+	_, err = partitionID(info, []interface{}{"ha", 1 << 44})
+	require.Error(t, err)
+
+	vals, err := partitionID(info, []interface{}{"ha", 1})
+	require.NoError(t, err)
+	require.Equal(t, []interface{}{"ha", int64(2<<59 | 1<<52 | 1<<44 | 1)}, vals)
+
+	info.instanceID = 0
+	vals, err = partitionID(info, []interface{}{"ha", "123"})
+	require.NoError(t, err)
+	require.Equal(t, []interface{}{"ha", fmt.Sprintf("%d", int64(1<<52|1<<44|123))}, vals)
+}
+
+func TestCaseSensitive(t *testing.T) {
+	// we test case insensitive in TestHandle
+	rules := []*Rule{
+		{"Test*", "xxx*", "", "id", AddPrefix, []string{"instance_id:"}, "xx"},
+	}
+
+	// case sensitive
+	// initial column mapping
+	m, err := NewMapping(true, rules)
+	require.NoError(t, err)
+	require.Len(t, m.cache.infos, 0)
+
+	// test add prefix, add suffix is similar
+	vals, poss, err := m.HandleRowValue("test", "xxx", []string{"age", "id"}, []interface{}{1, "1"})
+	require.NoError(t, err)
+	require.Equal(t, []interface{}{1, "1"}, vals)
+	require.Nil(t, poss)
+}
diff --git a/pkg/filter/filter_test.go b/pkg/filter/filter_test.go
index c88e99afe..35356e77e 100644
--- a/pkg/filter/filter_test.go
+++ b/pkg/filter/filter_test.go
@@ -16,9 +16,9 @@ package filter
 import (
 	"testing"
 
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/stretchr/testify/require"
 )
diff --git a/pkg/filter/sql_event_filter_test.go b/pkg/filter/sql_event_filter_test.go
index 9b27fa2b9..c0bc1f5bf 100644
--- a/pkg/filter/sql_event_filter_test.go
+++ b/pkg/filter/sql_event_filter_test.go
@@ -17,8 +17,8 @@ import (
 	"testing"
 
 	"github.com/pingcap/errors"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/cdc/model"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/stretchr/testify/require"
diff --git a/pkg/filter/utils_test.go b/pkg/filter/utils_test.go
index 6f7297fa5..ef7540f7c 100644
--- a/pkg/filter/utils_test.go
+++ b/pkg/filter/utils_test.go
@@ -18,11 +18,11 @@ import (
 	"testing"
 
 	"github.com/pingcap/log"
-	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/pkg/parser"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	tifilter "github.com/pingcap/tidb/pkg/util/filter"
+	bf "github.com/pingcap/tiflow/pkg/binlog-filter"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/stretchr/testify/require"
diff --git a/pkg/sink/codec/simple/encoder_test.go b/pkg/sink/codec/simple/encoder_test.go
index 5cc83a9ad..ff32ff18f 100644
--- a/pkg/sink/codec/simple/encoder_test.go
+++ b/pkg/sink/codec/simple/encoder_test.go
@@ -24,6 +24,7 @@ import (
 	"time"
 
 	"github.com/DATA-DOG/go-sqlmock"
+	"github.com/golang/mock/gomock"
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
 	"github.com/pingcap/tidb/pkg/parser/mysql"
 	"github.com/pingcap/tiflow/cdc/entry"
@@ -33,6 +34,7 @@ import (
 	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/integrity"
 	"github.com/pingcap/tiflow/pkg/sink/codec/common"
+	mock_simple "github.com/pingcap/tiflow/pkg/sink/codec/simple/mock"
 	"github.com/pingcap/tiflow/pkg/sink/codec/utils"
 	"github.com/stretchr/testify/require"
 )
@@ -84,7 +86,7 @@ func TestEncodeCheckpoint(t *testing.T) {
 func TestEncodeDMLEnableChecksum(t *testing.T) {
 	replicaConfig := config.GetDefaultReplicaConfig()
 	replicaConfig.Integrity.IntegrityCheckLevel = integrity.CheckLevelCorrectness
-	createTableDDL, insertEvent, _, _ := utils.NewLargeEvent4Test(t, replicaConfig)
+	createTableDDL, _, updateEvent, _ := utils.NewLargeEvent4Test(t, replicaConfig)
 	rand.New(rand.NewSource(time.Now().Unix())).Shuffle(len(createTableDDL.TableInfo.Columns), func(i, j int) {
 		createTableDDL.TableInfo.Columns[i],
 			createTableDDL.TableInfo.Columns[j] = createTableDDL.TableInfo.Columns[j],
@@ -127,7 +129,7 @@ func TestEncodeDMLEnableChecksum(t *testing.T) {
 			_, err = dec.NextDDLEvent()
 			require.NoError(t, err)
 
-			err = enc.AppendRowChangedEvent(ctx, "", insertEvent, func() {})
+			err = enc.AppendRowChangedEvent(ctx, "", updateEvent, func() {})
 			require.NoError(t, err)
 
 			messages := enc.Build()
@@ -143,11 +145,53 @@ func TestEncodeDMLEnableChecksum(t *testing.T) {
 
 			decodedRow, err := dec.NextRowChangedEvent()
 			require.NoError(t, err)
-			require.Equal(t, insertEvent.Checksum.Current, decodedRow.Checksum.Current)
-			require.Equal(t, insertEvent.Checksum.Previous, decodedRow.Checksum.Previous)
+			require.Equal(t, updateEvent.Checksum.Current, decodedRow.Checksum.Current)
+			require.Equal(t, updateEvent.Checksum.Previous, decodedRow.Checksum.Previous)
 			require.False(t, decodedRow.Checksum.Corrupted)
 		}
 	}
+
+	// tamper the checksum, to test error case
+	updateEvent.Checksum.Current = 1
+	updateEvent.Checksum.Previous = 2
+
+	b, err := NewBuilder(ctx, codecConfig)
+	require.NoError(t, err)
+	enc := b.Build()
+
+	dec, err := NewDecoder(ctx, codecConfig, nil)
+	require.NoError(t, err)
+	m, err := enc.EncodeDDLEvent(createTableDDL)
+	require.NoError(t, err)
+
+	err = dec.AddKeyValue(m.Key, m.Value)
+	require.NoError(t, err)
+
+	messageType, hasNext, err := dec.HasNext()
+	require.NoError(t, err)
+	require.True(t, hasNext)
+	require.Equal(t, model.MessageTypeDDL, messageType)
+
+	_, err = dec.NextDDLEvent()
+	require.NoError(t, err)
+
+	err = enc.AppendRowChangedEvent(ctx, "", updateEvent, func() {})
+	require.NoError(t, err)
+
+	messages := enc.Build()
+	require.Len(t, messages, 1)
+
+	err = dec.AddKeyValue(messages[0].Key, messages[0].Value)
+	require.NoError(t, err)
+
+	messageType, hasNext, err = dec.HasNext()
+	require.NoError(t, err)
+	require.True(t, hasNext)
+	require.Equal(t, model.MessageTypeRow, messageType)
+
+	decodedRow, err := dec.NextRowChangedEvent()
+	require.Error(t, err)
+	require.Nil(t, decodedRow)
 }
 
 func TestEncodeDDLSequence(t *testing.T) {
@@ -1652,3 +1696,40 @@ func TestDecoder(t *testing.T) {
 	err = decoder.AddKeyValue(nil, nil)
 	require.ErrorIs(t, err, errors.ErrCodecDecode)
 }
+
+func TestMarshallerError(t *testing.T) {
+	ctx := context.Background()
+	codecConfig := common.NewConfig(config.ProtocolSimple)
+
+	b, err := NewBuilder(ctx, codecConfig)
+	require.NoError(t, err)
+	enc := b.Build()
+
+	mockMarshaller := mock_simple.NewMockmarshaller(gomock.NewController(t))
+	enc.(*encoder).marshaller = mockMarshaller
+
+	mockMarshaller.EXPECT().MarshalCheckpoint(gomock.Any()).Return(nil, errors.ErrEncodeFailed)
+	_, err = enc.EncodeCheckpointEvent(123)
+	require.ErrorIs(t, err, errors.ErrEncodeFailed)
+
+	mockMarshaller.EXPECT().MarshalDDLEvent(gomock.Any()).Return(nil, errors.ErrEncodeFailed)
+	_, err = enc.EncodeDDLEvent(&model.DDLEvent{})
+	require.ErrorIs(t, err, errors.ErrEncodeFailed)
+
+	mockMarshaller.EXPECT().MarshalRowChangedEvent(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, errors.ErrEncodeFailed)
+	err = enc.AppendRowChangedEvent(ctx, "", &model.RowChangedEvent{}, func() {})
+	require.ErrorIs(t, err, errors.ErrEncodeFailed)
+
+	dec, err := NewDecoder(ctx, codecConfig, nil)
+	require.NoError(t, err)
+	dec.marshaller = mockMarshaller
+
+	mockMarshaller.EXPECT().Unmarshal(gomock.Any(), gomock.Any()).Return(errors.ErrDecodeFailed)
+	err = dec.AddKeyValue([]byte("key"), []byte("value"))
+	require.NoError(t, err)
+
+	messageType, hasNext, err := dec.HasNext()
+	require.ErrorIs(t, err, errors.ErrDecodeFailed)
+	require.False(t, hasNext)
+	require.Equal(t, model.MessageTypeUnknown, messageType)
+}
diff --git a/scripts/download-integration-test-binaries.sh b/scripts/download-integration-test-binaries.sh
index e383e1fb6..ec12db471 100755
--- a/scripts/download-integration-test-binaries.sh
+++ b/scripts/download-integration-test-binaries.sh
@@ -31,7 +31,7 @@ branch=${1:-master}
 # binaries.
 community=${2:-false}
 # Specify which version of the community binaries that will be utilized.
-ver=${3:-v6.5.2}
+ver=${3:-v7.5.2}
 # Specify which os that will be used to pack the binaries.
 os=${4:-linux}
 # Specify which architecture that will be used to pack the binaries.
@@ -130,7 +130,7 @@ function download_binaries() {
 	minio_download_url="${file_server_url}/download/minio.tar.gz"
 	go_ycsb_download_url="${file_server_url}/download/builds/pingcap/go-ycsb/test-br/go-ycsb"
 	etcd_download_url="${file_server_url}/download/builds/pingcap/cdc/etcd-v3.4.7-linux-amd64.tar.gz"
-	sync_diff_inspector_url="${file_server_url}/download/builds/pingcap/cdc/sync_diff_inspector_hash-d671b084_linux-amd64.tar.gz"
+	sync_diff_inspector_url="${file_server_url}/download/builds/pingcap/cdc/sync_diff_inspector_hash-79f1fd1e_linux-amd64.tar.gz"
 	jq_download_url="${file_server_url}/download/builds/pingcap/test/jq-1.6/jq-linux64"
 	schema_registry_url="${file_server_url}/download/builds/pingcap/cdc/schema-registry.tar.gz"
 
diff --git a/tests/integration_tests/http_api/run.sh b/tests/integration_tests/http_api/run.sh
index c95795971..db5604066 100644
--- a/tests/integration_tests/http_api/run.sh
+++ b/tests/integration_tests/http_api/run.sh
@@ -76,8 +76,8 @@ function run() {
 	sequential_cases2=(
 		"list_changefeed"
 		"get_changefeed"
-		"resume_changefeed"
 		"pause_changefeed"
+		"resume_changefeed"
 		"rebalance_table"
 		"move_table"
 		"get_processor"
diff --git a/tests/integration_tests/http_api/util/test_case.py b/tests/integration_tests/http_api/util/test_case.py
index 756758ea8..9c2a14888 100644
--- a/tests/integration_tests/http_api/util/test_case.py
+++ b/tests/integration_tests/http_api/util/test_case.py
@@ -287,7 +287,7 @@ def get_processor():
     assert resp.status_code == rq.codes.ok
     data = resp.json()[0]
     time.sleep(2)
-    url = base_url + "/" + data["changefeed_id"] + "/" + data["capture_id"]
+    url = base_url + "/changefeed-test1/" + data["capture_id"]
     resp = requests_get_with_retry(url)
     # print error message for debug 
     if (resp.status_code != rq.codes.ok):
diff --git a/tests/integration_tests/http_api_tls/util/test_case.py b/tests/integration_tests/http_api_tls/util/test_case.py
index 16a7be18e..6a7d436f9 100644
--- a/tests/integration_tests/http_api_tls/util/test_case.py
+++ b/tests/integration_tests/http_api_tls/util/test_case.py
@@ -255,7 +255,7 @@ def get_processor():
     assert resp.status_code == rq.codes.ok
     data = resp.json()[0]
     time.sleep(2)
-    url = base_url + "/" + data["changefeed_id"] + "/" + data["capture_id"]
+    url = base_url + "/changefeed-test1/" + data["capture_id"]
     resp = rq.get(url, cert=CERT, verify=VERIFY)
     # print error message for debug 
     if (resp.status_code != rq.codes.ok):
@@ -269,7 +269,7 @@ def get_processor():
     resp = rq.get(url, cert=CERT, verify=VERIFY)
     assert resp.status_code == rq.codes.bad_request
 
-    print("pass test: get processors")
+    print("pass test: get processor")
 
 
 def check_health():