Repository: pingcap/tiflow
Pull request #11253 (2024-06-05 08:18:36)

ticdc: change the calculation of first sync point ts
<!--
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/11212

### What is changed and how it works?
In this PR, we change the calculation of the first sync point ts.
As before, we use startTs as the first sync point ts.
Afer the PR, the first-sync-point-ts = k * sync-point-ts-interval,
and we will choose the minimal k that satisfy the first-sync-point-ts >= startTs.

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Unit test
 - Integration test
 - Manual test (add detailed scripts or steps below)


#### 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
Change the calculation of first sync point ts.
```


ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: To remove the `do-not-merge/needs-linked-issue` label, please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

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 ask for approval from [hongyunyan](https://github.com/hongyunyan) and additionally assign [benmeadowcroft](https://github.com/benmeadowcroft) for approval(**Please ensuring that each of them provides their approval before proceeding**). 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/master/OWNERS)**
- **[cdc/api/v2/OWNERS](https://github.com/pingcap/tiflow/blob/master/cdc/api/v2/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":["benmeadowcroft","hongyunyan"]} -->
ti-chi-bot[bot]: 
[FORMAT CHECKER NOTIFICATION]

**Notice**: To remove the `do-not-merge/needs-linked-issue` label, please provide the linked issue number on one line in the PR body, for example: `Issue Number: close #123` or `Issue Number: ref #456`.

<sub>:open_book: For more info, you can check the ["Contribute Code"](https://pingcap.github.io/tidb-dev-guide/contribute-to-tidb/contribute-code.html#referring-to-an-issue) section in the development guide.</sub>


<!--Checker Notification Identifier-->

ti-chi-bot[bot]: @hongyunyan: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

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

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

<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]: @hongyunyan: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

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

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

<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]: @hongyunyan: 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 | dca7a3ea115a75c254af231765588465fc695eaf | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/2109/display/redirect) | true | `/test verify`
pull-dm-integration-test | dca7a3ea115a75c254af231765588465fc695eaf | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/2317/display/redirect) | true | `/test dm-integration-test`

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

<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 -->
hongyunyan: /retest
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pingcap/tiflow/pull/11253?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
Attention: Patch coverage is `51.28205%` with `19 lines` in your changes missing coverage. Please review.
> Project coverage is 57.6403%. Comparing base [(`667cea0`)](https://app.codecov.io/gh/pingcap/tiflow/commit/667cea0175676e32b5f5dcc9bc3bcc1ea267a61a?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) to head [(`dca7a3e`)](https://app.codecov.io/gh/pingcap/tiflow/commit/dca7a3ea115a75c254af231765588465fc695eaf?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap).
> Report is 2 commits behind head on master.

<details><summary>Additional details and impacted files</summary>


| [Components](https://app.codecov.io/gh/pingcap/tiflow/pull/11253/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | Coverage Δ | |
|---|---|---|
| [cdc](https://app.codecov.io/gh/pingcap/tiflow/pull/11253/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `61.4625% <51.2820%> (-0.0486%)` | :arrow_down: |
| [dm](https://app.codecov.io/gh/pingcap/tiflow/pull/11253/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `51.2031% <ø> (+0.0020%)` | :arrow_up: |
| [engine](https://app.codecov.io/gh/pingcap/tiflow/pull/11253/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `63.3808% <ø> (-0.0212%)` | :arrow_down: |

| [Flag](https://app.codecov.io/gh/pingcap/tiflow/pull/11253/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | Coverage Δ | |
|---|---|---|
| [unit](https://app.codecov.io/gh/pingcap/tiflow/pull/11253/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `57.6403% <51.2820%> (-0.0247%)` | :arrow_down: |

Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap#carryforward-flags-in-the-pull-request-comment) to find out more.

```diff
@@               Coverage Diff                @@
##             master     #11253        +/-   ##
================================================
- Coverage   57.6650%   57.6403%   -0.0247%     
================================================
  Files           850        850                
  Lines        126059     126087        +28     
================================================
- Hits          72692      72677        -15     
- Misses        47971      48005        +34     
- Partials       5396       5405         +9     
```

</details>

Patch:
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
index 5bb6e3568..6a224ca2e 100644
--- a/cdc/api/v2/model.go
+++ b/cdc/api/v2/model.go
@@ -317,6 +317,7 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 				BasicPassword:           c.Sink.PulsarConfig.BasicPassword,
 				AuthTLSCertificatePath:  c.Sink.PulsarConfig.AuthTLSCertificatePath,
 				AuthTLSPrivateKeyPath:   c.Sink.PulsarConfig.AuthTLSPrivateKeyPath,
+				OutputRawChangeEvent:    c.Sink.PulsarConfig.OutputRawChangeEvent,
 			}
 			if c.Sink.PulsarConfig.OAuth2 != nil {
 				pulsarConfig.OAuth2 = &config.OAuth2{
@@ -402,6 +403,7 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 				CodecConfig:                  codeConfig,
 				LargeMessageHandle:           largeMessageHandle,
 				GlueSchemaRegistryConfig:     glueSchemaRegistryConfig,
+				OutputRawChangeEvent:         c.Sink.KafkaConfig.OutputRawChangeEvent,
 			}
 		}
 		var mysqlConfig *config.MySQLConfig
@@ -427,13 +429,14 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 		var cloudStorageConfig *config.CloudStorageConfig
 		if c.Sink.CloudStorageConfig != nil {
 			cloudStorageConfig = &config.CloudStorageConfig{
-				WorkerCount:         c.Sink.CloudStorageConfig.WorkerCount,
-				FlushInterval:       c.Sink.CloudStorageConfig.FlushInterval,
-				FileSize:            c.Sink.CloudStorageConfig.FileSize,
-				OutputColumnID:      c.Sink.CloudStorageConfig.OutputColumnID,
-				FileExpirationDays:  c.Sink.CloudStorageConfig.FileExpirationDays,
-				FileCleanupCronSpec: c.Sink.CloudStorageConfig.FileCleanupCronSpec,
-				FlushConcurrency:    c.Sink.CloudStorageConfig.FlushConcurrency,
+				WorkerCount:          c.Sink.CloudStorageConfig.WorkerCount,
+				FlushInterval:        c.Sink.CloudStorageConfig.FlushInterval,
+				FileSize:             c.Sink.CloudStorageConfig.FileSize,
+				OutputColumnID:       c.Sink.CloudStorageConfig.OutputColumnID,
+				FileExpirationDays:   c.Sink.CloudStorageConfig.FileExpirationDays,
+				FileCleanupCronSpec:  c.Sink.CloudStorageConfig.FileCleanupCronSpec,
+				FlushConcurrency:     c.Sink.CloudStorageConfig.FlushConcurrency,
+				OutputRawChangeEvent: c.Sink.CloudStorageConfig.OutputRawChangeEvent,
 			}
 		}
 		var debeziumConfig *config.DebeziumConfig
@@ -666,6 +669,7 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 				CodecConfig:                  codeConfig,
 				LargeMessageHandle:           largeMessageHandle,
 				GlueSchemaRegistryConfig:     glueSchemaRegistryConfig,
+				OutputRawChangeEvent:         cloned.Sink.KafkaConfig.OutputRawChangeEvent,
 			}
 		}
 		var mysqlConfig *MySQLConfig
@@ -708,6 +712,7 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 				BasicPassword:           cloned.Sink.PulsarConfig.BasicPassword,
 				AuthTLSCertificatePath:  cloned.Sink.PulsarConfig.AuthTLSCertificatePath,
 				AuthTLSPrivateKeyPath:   cloned.Sink.PulsarConfig.AuthTLSPrivateKeyPath,
+				OutputRawChangeEvent:    cloned.Sink.PulsarConfig.OutputRawChangeEvent,
 			}
 			if cloned.Sink.PulsarConfig.OAuth2 != nil {
 				pulsarConfig.OAuth2 = &PulsarOAuth2{
@@ -722,13 +727,14 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 		var cloudStorageConfig *CloudStorageConfig
 		if cloned.Sink.CloudStorageConfig != nil {
 			cloudStorageConfig = &CloudStorageConfig{
-				WorkerCount:         cloned.Sink.CloudStorageConfig.WorkerCount,
-				FlushInterval:       cloned.Sink.CloudStorageConfig.FlushInterval,
-				FileSize:            cloned.Sink.CloudStorageConfig.FileSize,
-				OutputColumnID:      cloned.Sink.CloudStorageConfig.OutputColumnID,
-				FileExpirationDays:  cloned.Sink.CloudStorageConfig.FileExpirationDays,
-				FileCleanupCronSpec: cloned.Sink.CloudStorageConfig.FileCleanupCronSpec,
-				FlushConcurrency:    cloned.Sink.CloudStorageConfig.FlushConcurrency,
+				WorkerCount:          cloned.Sink.CloudStorageConfig.WorkerCount,
+				FlushInterval:        cloned.Sink.CloudStorageConfig.FlushInterval,
+				FileSize:             cloned.Sink.CloudStorageConfig.FileSize,
+				OutputColumnID:       cloned.Sink.CloudStorageConfig.OutputColumnID,
+				FileExpirationDays:   cloned.Sink.CloudStorageConfig.FileExpirationDays,
+				FileCleanupCronSpec:  cloned.Sink.CloudStorageConfig.FileCleanupCronSpec,
+				FlushConcurrency:     cloned.Sink.CloudStorageConfig.FlushConcurrency,
+				OutputRawChangeEvent: cloned.Sink.CloudStorageConfig.OutputRawChangeEvent,
 			}
 		}
 		var debeziumConfig *DebeziumConfig
@@ -1194,6 +1200,7 @@ type PulsarConfig struct {
 	AuthTLSCertificatePath  *string       `json:"auth-tls-certificate-path,omitempty"`
 	AuthTLSPrivateKeyPath   *string       `json:"auth-tls-private-key-path,omitempty"`
 	OAuth2                  *PulsarOAuth2 `json:"oauth2,omitempty"`
+	OutputRawChangeEvent    *bool         `json:"output-raw-change-event,omitempty"`
 }
 
 // PulsarOAuth2 is the configuration for OAuth2
@@ -1243,6 +1250,7 @@ type KafkaConfig struct {
 	CodecConfig                  *CodecConfig              `json:"codec_config,omitempty"`
 	LargeMessageHandle           *LargeMessageHandleConfig `json:"large_message_handle,omitempty"`
 	GlueSchemaRegistryConfig     *GlueSchemaRegistryConfig `json:"glue_schema_registry_config,omitempty"`
+	OutputRawChangeEvent         *bool                     `json:"output_raw_change_event,omitempty"`
 }
 
 // MySQLConfig represents a MySQL sink configuration
@@ -1266,13 +1274,14 @@ type MySQLConfig struct {
 
 // CloudStorageConfig represents a cloud storage sink configuration
 type CloudStorageConfig struct {
-	WorkerCount         *int    `json:"worker_count,omitempty"`
-	FlushInterval       *string `json:"flush_interval,omitempty"`
-	FileSize            *int    `json:"file_size,omitempty"`
-	OutputColumnID      *bool   `json:"output_column_id,omitempty"`
-	FileExpirationDays  *int    `json:"file_expiration_days,omitempty"`
-	FileCleanupCronSpec *string `json:"file_cleanup_cron_spec,omitempty"`
-	FlushConcurrency    *int    `json:"flush_concurrency,omitempty"`
+	WorkerCount          *int    `json:"worker_count,omitempty"`
+	FlushInterval        *string `json:"flush_interval,omitempty"`
+	FileSize             *int    `json:"file_size,omitempty"`
+	OutputColumnID       *bool   `json:"output_column_id,omitempty"`
+	FileExpirationDays   *int    `json:"file_expiration_days,omitempty"`
+	FileCleanupCronSpec  *string `json:"file_cleanup_cron_spec,omitempty"`
+	FlushConcurrency     *int    `json:"flush_concurrency,omitempty"`
+	OutputRawChangeEvent *bool   `json:"output_raw_change_event,omitempty"`
 }
 
 // ChangefeedStatus holds common information of a changefeed in cdc
diff --git a/cdc/entry/mounter.go b/cdc/entry/mounter.go
index 82be98ea6..c9513f683 100644
--- a/cdc/entry/mounter.go
+++ b/cdc/entry/mounter.go
@@ -380,11 +380,11 @@ func parseJob(v []byte, startTs, CRTs uint64, fromHistoryTable bool) (*timodel.J
 	}
 
 	if fromHistoryTable {
-		// we only want to get `create table` ddl from tidb_ddl_history, so we just throw out others ddls.
+		// we only want to get `create table` and `create tables` ddl from tidb_ddl_history, so we just throw out others ddls.
 		// We only want the job with `JobStateSynced`, which is means the ddl job is done successfully.
 		// Besides, to satisfy the subsequent processing,
 		// We need to set the job to be Done to make it will replay in schemaStorage
-		if job.Type != timodel.ActionCreateTable || job.State != timodel.JobStateSynced {
+		if (job.Type != timodel.ActionCreateTable && job.Type != timodel.ActionCreateTables) || job.State != timodel.JobStateSynced {
 			return nil, nil
 		}
 		job.State = timodel.JobStateDone
diff --git a/cdc/entry/schema/snapshot.go b/cdc/entry/schema/snapshot.go
index e01821ef5..0af4f370b 100644
--- a/cdc/entry/schema/snapshot.go
+++ b/cdc/entry/schema/snapshot.go
@@ -43,7 +43,7 @@ func (s *Snapshot) PreTableInfo(job *timodel.Job) (*model.TableInfo, error) {
 	switch job.Type {
 	case timodel.ActionCreateSchema, timodel.ActionModifySchemaCharsetAndCollate, timodel.ActionDropSchema:
 		return nil, nil
-	case timodel.ActionCreateTable, timodel.ActionCreateView, timodel.ActionRecoverTable:
+	case timodel.ActionCreateTable, timodel.ActionCreateTables, timodel.ActionCreateView, timodel.ActionRecoverTable:
 		// no pre table info
 		return nil, nil
 	case timodel.ActionRenameTable, timodel.ActionDropTable, timodel.ActionDropView, timodel.ActionTruncateTable, timodel.ActionAlterTablePartitioning, timodel.ActionRemovePartitioning:
@@ -472,6 +472,15 @@ func (s *Snapshot) DoHandleDDL(job *timodel.Job) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
+	case timodel.ActionCreateTables:
+		multiTableInfos := job.BinlogInfo.MultipleTableInfos
+		for _, tableInfo := range multiTableInfos {
+			err := s.inner.createTable(model.WrapTableInfo(job.SchemaID, job.SchemaName,
+				job.BinlogInfo.FinishedTS, tableInfo), job.BinlogInfo.FinishedTS)
+			if err != nil {
+				return errors.Trace(err)
+			}
+		}
 	case timodel.ActionCreateTable, timodel.ActionCreateView, timodel.ActionRecoverTable:
 		err := s.inner.createTable(getWrapTableInfo(job), job.BinlogInfo.FinishedTS)
 		if err != nil {
diff --git a/cdc/entry/schema_storage.go b/cdc/entry/schema_storage.go
index c4ba9c38d..faaa6d4fe 100644
--- a/cdc/entry/schema_storage.go
+++ b/cdc/entry/schema_storage.go
@@ -16,6 +16,7 @@ package entry
 import (
 	"context"
 	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -407,6 +408,20 @@ func (s *schemaStorage) BuildDDLEvents(
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
+	case timodel.ActionCreateTables:
+		if job.BinlogInfo != nil && job.BinlogInfo.MultipleTableInfos != nil {
+			querys := strings.Split(job.Query, ";")
+			multiTableInfos := job.BinlogInfo.MultipleTableInfos
+			for index, tableInfo := range multiTableInfos {
+				newTableInfo := model.WrapTableInfo(job.SchemaID, job.SchemaName, job.BinlogInfo.FinishedTS, tableInfo)
+				job.Query = querys[index]
+				event := new(model.DDLEvent)
+				event.FromJob(job, nil, newTableInfo)
+				ddlEvents = append(ddlEvents, event)
+			}
+		} else {
+			return nil, errors.Errorf("there is no multiple table infos in the create tables job: %s", job)
+		}
 	default:
 		// parse preTableInfo
 		preSnap, err := s.GetSnapshot(ctx, job.BinlogInfo.FinishedTS-1)
diff --git a/cdc/model/sink.go b/cdc/model/sink.go
index 217df3ae8..7d12b07d9 100644
--- a/cdc/model/sink.go
+++ b/cdc/model/sink.go
@@ -272,7 +272,7 @@ func (r *RedoLog) GetCommitTs() Ts {
 }
 
 // TrySplitAndSortUpdateEvent redo log do nothing
-func (r *RedoLog) TrySplitAndSortUpdateEvent(_ string) error {
+func (r *RedoLog) TrySplitAndSortUpdateEvent(_ string, _ bool) error {
 	return nil
 }
 
@@ -444,7 +444,7 @@ func (r *RowChangedEvent) GetCommitTs() uint64 {
 }
 
 // TrySplitAndSortUpdateEvent do nothing
-func (r *RowChangedEvent) TrySplitAndSortUpdateEvent(_ string) error {
+func (r *RowChangedEvent) TrySplitAndSortUpdateEvent(_ string, _ bool) error {
 	return nil
 }
 
@@ -1140,10 +1140,19 @@ func (t *SingleTableTxn) GetPhysicalTableID() int64 {
 }
 
 // TrySplitAndSortUpdateEvent split update events if unique key is updated
-func (t *SingleTableTxn) TrySplitAndSortUpdateEvent(scheme string) error {
-	if !t.shouldSplitUpdateEvent(scheme) {
+func (t *SingleTableTxn) TrySplitAndSortUpdateEvent(scheme string, outputRawChangeEvent bool) error {
+	if sink.IsMySQLCompatibleScheme(scheme) || outputRawChangeEvent {
+		// For MySQL Sink, all update events will be split into insert and delete at the puller side
+		// according to whether the changefeed is in safemode. We don't split update event here(in sink)
+		// since there may be OOM issues. For more information, ref https://github.com/tikv/tikv/issues/17062.
+		//
+		// For the Kafka and Storage sink, the outputRawChangeEvent parameter is introduced to control
+		// split behavior. TiCDC only output original change event if outputRawChangeEvent is true.
 		return nil
 	}
+
+	// Try to split update events for the Kafka and Storage sink if outputRawChangeEvent is false.
+	// Note it is only for backward compatibility, and we should remove this logic in the future.
 	newRows, err := trySplitAndSortUpdateEvent(t.Rows)
 	if err != nil {
 		return errors.Trace(err)
@@ -1152,21 +1161,6 @@ func (t *SingleTableTxn) TrySplitAndSortUpdateEvent(scheme string) error {
 	return nil
 }
 
-// Whether split a single update event into delete and insert events？
-//
-// For the MySQL Sink, we don't split any update event.
-// This may cause error like "duplicate entry" when sink to the downstream.
-// This kind of error will cause the changefeed to restart,
-// and then the related update rows will be splitted to insert and delete at puller side.
-//
-// For the Kafka and Storage sink, always split a single unique key changed update event, since:
-// 1. Avro and CSV does not output the previous column values for the update event, so it would
-// cause consumer missing data if the unique key changed event is not split.
-// 2. Index-Value Dispatcher cannot work correctly if the unique key changed event isn't split.
-func (t *SingleTableTxn) shouldSplitUpdateEvent(sinkScheme string) bool {
-	return !sink.IsMySQLCompatibleScheme(sinkScheme)
-}
-
 // trySplitAndSortUpdateEvent try to split update events if unique key is updated
 // returns true if some updated events is split
 func trySplitAndSortUpdateEvent(
@@ -1176,8 +1170,7 @@ func trySplitAndSortUpdateEvent(
 	split := false
 	for _, e := range events {
 		if e == nil {
-			log.Warn("skip emit nil event",
-				zap.Any("event", e))
+			log.Warn("skip emit nil event", zap.Any("event", e))
 			continue
 		}
 
@@ -1187,8 +1180,7 @@ func trySplitAndSortUpdateEvent(
 		// begin; insert into t (id) values (1); delete from t where id=1; commit;
 		// Just ignore these row changed events.
 		if colLen == 0 && preColLen == 0 {
-			log.Warn("skip emit empty row event",
-				zap.Any("event", e))
+			log.Warn("skip emit empty row event", zap.Any("event", e))
 			continue
 		}
 
@@ -1222,7 +1214,7 @@ func isNonEmptyUniqueOrHandleCol(col *ColumnData, tableInfo *TableInfo) bool {
 
 // ShouldSplitUpdateEvent determines if the split event is needed to align the old format based on
 // whether the handle key column or unique key has been modified.
-// If  is modified, we need to use splitUpdateEvent to split the update event into a delete and an insert event.
+// If is modified, we need to use splitUpdateEvent to split the update event into a delete and an insert event.
 func ShouldSplitUpdateEvent(updateEvent *RowChangedEvent) bool {
 	// nil event will never be split.
 	if updateEvent == nil {
@@ -1265,6 +1257,13 @@ func SplitUpdateEvent(
 	// NOTICE: clean up pre cols for insert event.
 	insertEvent.PreColumns = nil
 
+	log.Debug("split update event", zap.Uint64("startTs", updateEvent.StartTs),
+		zap.Uint64("commitTs", updateEvent.CommitTs),
+		zap.String("schema", updateEvent.TableInfo.TableName.Schema),
+		zap.String("table", updateEvent.TableInfo.GetTableName()),
+		zap.Any("preCols", updateEvent.PreColumns),
+		zap.Any("cols", updateEvent.Columns))
+
 	return &deleteEvent, &insertEvent, nil
 }
 
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
index fbeec7289..18fc88bdb 100755
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -631,7 +631,15 @@ LOOP2:
 
 	c.barriers = newBarriers()
 	if util.GetOrZero(cfInfo.Config.EnableSyncPoint) {
-		c.barriers.Update(syncPointBarrier, c.resolvedTs)
+		// firstSyncPointStartTs = k * syncPointInterval，
+		// which >= startTs, and choose the minimal k
+		syncPointInterval := util.GetOrZero(cfInfo.Config.SyncPointInterval)
+		k := oracle.GetTimeFromTS(c.resolvedTs).Sub(time.Unix(0, 0)) / syncPointInterval
+		if oracle.GetTimeFromTS(c.resolvedTs).Sub(time.Unix(0, 0))%syncPointInterval != 0 || oracle.ExtractLogical(c.resolvedTs) != 0 {
+			k += 1
+		}
+		firstSyncPointTs := oracle.GoTimeToTS(time.Unix(0, 0).Add(k * syncPointInterval))
+		c.barriers.Update(syncPointBarrier, firstSyncPointTs)
 	}
 	c.barriers.Update(finishBarrier, cfInfo.GetTargetTs())
 
@@ -811,6 +819,20 @@ func (c *changefeed) releaseResources(ctx context.Context) {
 	c.initialized.Store(false)
 	c.isReleased = true
 
+	// when closing a changefeed, we must clean the warningCh.
+	// otherwise, the old warning errors will be handled when the reused changefeed instance is ticked again
+OUT:
+	for {
+		select {
+		case err := <-c.warningCh:
+			log.Warn("drain owner warnings",
+				zap.String("namespace", c.id.Namespace),
+				zap.String("changefeed", c.id.ID),
+				zap.Error(err))
+		default:
+			break OUT
+		}
+	}
 	log.Info("changefeed closed",
 		zap.String("namespace", c.id.Namespace),
 		zap.String("changefeed", c.id.ID),
diff --git a/cdc/owner/feed_state_manager.go b/cdc/owner/feed_state_manager.go
index fc83a2eb4..1c2c87cd8 100644
--- a/cdc/owner/feed_state_manager.go
+++ b/cdc/owner/feed_state_manager.go
@@ -468,6 +468,7 @@ func (m *feedStateManager) HandleWarning(errs ...*model.RunningError) {
 				"it will be failed",
 				zap.String("namespace", m.state.GetID().Namespace),
 				zap.String("changefeed", m.state.GetID().ID),
+				zap.Time("checkpointTsAdvanced", m.checkpointTsAdvanced),
 				zap.Uint64("checkpointTs", m.state.GetChangefeedStatus().CheckpointTs),
 				zap.Duration("checkpointTime", currTime.Sub(ckptTime)),
 			)
diff --git a/cdc/processor/sourcemanager/sorter/pebble/event_sorter.go b/cdc/processor/sourcemanager/sorter/pebble/event_sorter.go
index 2122359e9..443f1d345 100644
--- a/cdc/processor/sourcemanager/sorter/pebble/event_sorter.go
+++ b/cdc/processor/sourcemanager/sorter/pebble/event_sorter.go
@@ -495,7 +495,11 @@ func (s *EventSorter) handleEvents(
 		if !ok {
 			return
 		}
-		batchCh <- batchEvent
+		select {
+		case <-s.closed:
+			return
+		case batchCh <- batchEvent:
+		}
 	}
 }
 
diff --git a/cdc/puller/ddl_puller.go b/cdc/puller/ddl_puller.go
index e50ad06bf..4ced26544 100644
--- a/cdc/puller/ddl_puller.go
+++ b/cdc/puller/ddl_puller.go
@@ -17,6 +17,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -390,6 +391,28 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 			return false, cerror.WrapError(cerror.ErrHandleDDLFailed,
 				errors.Trace(err), job.Query, job.StartTS, job.StartTS)
 		}
+	case timodel.ActionCreateTables:
+		// we only use multiTableInfos and Querys when we generate job event
+		// So if some table should be discard, we just need to delete the info from multiTableInfos and Querys
+		var newMultiTableInfos []*timodel.TableInfo
+		var newQuerys []string
+
+		multiTableInfos := job.BinlogInfo.MultipleTableInfos
+		querys := strings.Split(job.Query, ";")
+
+		for index, tableInfo := range multiTableInfos {
+			// judge each table whether need to be skip
+			if p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, tableInfo.Name.O) {
+				continue
+			}
+			newMultiTableInfos = append(newMultiTableInfos, multiTableInfos[index])
+			newQuerys = append(newQuerys, querys[index])
+		}
+
+		skip = len(newMultiTableInfos) == 0
+
+		job.BinlogInfo.MultipleTableInfos = newMultiTableInfos
+		job.Query = strings.Join(newQuerys, ";")
 	case timodel.ActionRenameTable:
 		oldTable, ok := snap.PhysicalTableByID(job.TableID)
 		if !ok {
diff --git a/cdc/sink/ddlsink/mq/ddlproducer/kafka_ddl_producer.go b/cdc/sink/ddlsink/mq/ddlproducer/kafka_ddl_producer.go
index f3abfd54b..7a2d2b384 100644
--- a/cdc/sink/ddlsink/mq/ddlproducer/kafka_ddl_producer.go
+++ b/cdc/sink/ddlsink/mq/ddlproducer/kafka_ddl_producer.go
@@ -69,8 +69,7 @@ func (k *kafkaDDLProducer) SyncBroadcastMessage(ctx context.Context, topic strin
 	case <-ctx.Done():
 		return ctx.Err()
 	default:
-		err := k.syncProducer.SendMessages(ctx, topic,
-			totalPartitionsNum, message.Key, message.Value)
+		err := k.syncProducer.SendMessages(ctx, topic, totalPartitionsNum, message)
 		return cerror.WrapError(cerror.ErrKafkaSendMessage, err)
 	}
 }
@@ -89,8 +88,7 @@ func (k *kafkaDDLProducer) SyncSendMessage(ctx context.Context, topic string,
 	case <-ctx.Done():
 		return errors.Trace(ctx.Err())
 	default:
-		err := k.syncProducer.SendMessage(ctx, topic,
-			partitionNum, message.Key, message.Value)
+		err := k.syncProducer.SendMessage(ctx, topic, partitionNum, message)
 		return cerror.WrapError(cerror.ErrKafkaSendMessage, err)
 	}
 }
diff --git a/cdc/sink/dmlsink/blackhole/black_hole_dml_sink.go b/cdc/sink/dmlsink/blackhole/black_hole_dml_sink.go
index be470b6d3..83f321dcd 100644
--- a/cdc/sink/dmlsink/blackhole/black_hole_dml_sink.go
+++ b/cdc/sink/dmlsink/blackhole/black_hole_dml_sink.go
@@ -47,9 +47,9 @@ func (s *DMLSink) WriteEvents(rows ...*dmlsink.CallbackableEvent[*model.RowChang
 	return
 }
 
-// Scheme return the scheme of the sink.
-func (s *DMLSink) Scheme() string {
-	return sink.BlackHoleScheme
+// SchemeOption returns the scheme and the option.
+func (s *DMLSink) SchemeOption() (string, bool) {
+	return sink.BlackHoleScheme, true
 }
 
 // Close do nothing.
diff --git a/cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink.go b/cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink.go
index e1dca4941..f93315305 100644
--- a/cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink.go
+++ b/cdc/sink/dmlsink/cloudstorage/cloud_storage_dml_sink.go
@@ -72,8 +72,9 @@ type eventFragment struct {
 // messages to individual dmlWorkers.
 // The dmlWorkers will write the encoded messages to external storage in parallel between different tables.
 type DMLSink struct {
-	changefeedID model.ChangeFeedID
-	scheme       string
+	changefeedID         model.ChangeFeedID
+	scheme               string
+	outputRawChangeEvent bool
 	// last sequence number
 	lastSeqNum uint64
 	// encodingWorkers defines a group of workers for encoding events.
@@ -144,13 +145,14 @@ func NewDMLSink(ctx context.Context,
 
 	wgCtx, wgCancel := context.WithCancel(ctx)
 	s := &DMLSink{
-		changefeedID:    changefeedID,
-		scheme:          strings.ToLower(sinkURI.Scheme),
-		encodingWorkers: make([]*encodingWorker, defaultEncodingConcurrency),
-		workers:         make([]*dmlWorker, cfg.WorkerCount),
-		statistics:      metrics.NewStatistics(wgCtx, changefeedID, sink.TxnSink),
-		cancel:          wgCancel,
-		dead:            make(chan struct{}),
+		changefeedID:         changefeedID,
+		scheme:               strings.ToLower(sinkURI.Scheme),
+		outputRawChangeEvent: replicaConfig.Sink.CloudStorageConfig.GetOutputRawChangeEvent(),
+		encodingWorkers:      make([]*encodingWorker, defaultEncodingConcurrency),
+		workers:              make([]*dmlWorker, cfg.WorkerCount),
+		statistics:           metrics.NewStatistics(wgCtx, changefeedID, sink.TxnSink),
+		cancel:               wgCancel,
+		dead:                 make(chan struct{}),
 	}
 	s.alive.msgCh = chann.NewAutoDrainChann[eventFragment]()
 
@@ -295,7 +297,7 @@ func (s *DMLSink) Dead() <-chan struct{} {
 	return s.dead
 }
 
-// Scheme returns the sink scheme.
-func (s *DMLSink) Scheme() string {
-	return s.scheme
+// SchemeOption returns the scheme and the option.
+func (s *DMLSink) SchemeOption() (string, bool) {
+	return s.scheme, s.outputRawChangeEvent
 }
diff --git a/cdc/sink/dmlsink/event.go b/cdc/sink/dmlsink/event.go
index 8df85cab2..7126be06b 100644
--- a/cdc/sink/dmlsink/event.go
+++ b/cdc/sink/dmlsink/event.go
@@ -23,7 +23,7 @@ type TableEvent interface {
 	// GetCommitTs returns the commit timestamp of the event.
 	GetCommitTs() uint64
 	// TrySplitAndSortUpdateEvent split the update to delete and insert if the unique key is updated
-	TrySplitAndSortUpdateEvent(scheme string) error
+	TrySplitAndSortUpdateEvent(scheme string, outputRawChangeEvent bool) error
 }
 
 // CallbackFunc is the callback function for callbackable event.
diff --git a/cdc/sink/dmlsink/event_sink.go b/cdc/sink/dmlsink/event_sink.go
index 76a179ecc..660ed1900 100644
--- a/cdc/sink/dmlsink/event_sink.go
+++ b/cdc/sink/dmlsink/event_sink.go
@@ -18,10 +18,8 @@ type EventSink[E TableEvent] interface {
 	// WriteEvents writes events to the sink.
 	// This is an asynchronously and thread-safe method.
 	WriteEvents(events ...*CallbackableEvent[E]) error
-
-	// Scheme returns the sink scheme.
-	Scheme() string
-
+	// SchemeOption returns the sink scheme and whether the sink should output raw change event.
+	SchemeOption() (scheme string, outputRawChangeEvent bool)
 	// Close closes the sink. Can be called with `WriteEvents` concurrently.
 	Close()
 	// The EventSink meets internal errors and has been dead already.
diff --git a/cdc/sink/dmlsink/mq/dmlproducer/kafka_dml_producer.go b/cdc/sink/dmlsink/mq/dmlproducer/kafka_dml_producer.go
index fec416491..0e5765ca9 100644
--- a/cdc/sink/dmlsink/mq/dmlproducer/kafka_dml_producer.go
+++ b/cdc/sink/dmlsink/mq/dmlproducer/kafka_dml_producer.go
@@ -119,8 +119,7 @@ func (k *kafkaDMLProducer) AsyncSendMessage(
 		k.failpointCh <- errors.New("kafka sink injected error")
 		failpoint.Return(nil)
 	})
-	return k.asyncProducer.AsyncSend(ctx, topic, partition,
-		message.Key, message.Value, message.Callback)
+	return k.asyncProducer.AsyncSend(ctx, topic, partition, message)
 }
 
 func (k *kafkaDMLProducer) Close() {
diff --git a/cdc/sink/dmlsink/mq/kafka_dml_sink.go b/cdc/sink/dmlsink/mq/kafka_dml_sink.go
index 3841c2a72..a0b0c4193 100644
--- a/cdc/sink/dmlsink/mq/kafka_dml_sink.go
+++ b/cdc/sink/dmlsink/mq/kafka_dml_sink.go
@@ -122,8 +122,8 @@ func NewKafkaDMLSink(
 	metricsCollector := factory.MetricsCollector(tiflowutil.RoleProcessor, adminClient)
 	dmlProducer := producerCreator(ctx, changefeedID, asyncProducer, metricsCollector, errCh, failpointCh)
 	encoderGroup := codec.NewEncoderGroup(replicaConfig.Sink, encoderBuilder, changefeedID)
-	s := newDMLSink(ctx, changefeedID, dmlProducer, adminClient, topicManager,
-		eventRouter, trans, encoderGroup, protocol, scheme, errCh)
+	s := newDMLSink(ctx, changefeedID, dmlProducer, adminClient, topicManager, eventRouter, trans, encoderGroup,
+		protocol, scheme, replicaConfig.Sink.KafkaConfig.GetOutputRawChangeEvent(), errCh)
 	log.Info("DML sink producer created",
 		zap.String("namespace", changefeedID.Namespace),
 		zap.String("changefeedID", changefeedID.ID))
diff --git a/cdc/sink/dmlsink/mq/mq_dml_sink.go b/cdc/sink/dmlsink/mq/mq_dml_sink.go
index e6d2105b6..5bed18936 100644
--- a/cdc/sink/dmlsink/mq/mq_dml_sink.go
+++ b/cdc/sink/dmlsink/mq/mq_dml_sink.go
@@ -70,7 +70,8 @@ type dmlSink struct {
 	wg   sync.WaitGroup
 	dead chan struct{}
 
-	scheme string
+	scheme               string
+	outputRawChangeEvent bool
 }
 
 func newDMLSink(
@@ -84,6 +85,7 @@ func newDMLSink(
 	encoderGroup codec.EncoderGroup,
 	protocol config.Protocol,
 	scheme string,
+	outputRawChangeEvent bool,
 	errCh chan error,
 ) *dmlSink {
 	ctx, cancel := context.WithCancelCause(ctx)
@@ -91,13 +93,14 @@ func newDMLSink(
 	worker := newWorker(changefeedID, protocol, producer, encoderGroup, statistics)
 
 	s := &dmlSink{
-		id:          changefeedID,
-		protocol:    protocol,
-		adminClient: adminClient,
-		ctx:         ctx,
-		cancel:      cancel,
-		dead:        make(chan struct{}),
-		scheme:      scheme,
+		id:                   changefeedID,
+		protocol:             protocol,
+		adminClient:          adminClient,
+		ctx:                  ctx,
+		cancel:               cancel,
+		dead:                 make(chan struct{}),
+		scheme:               scheme,
+		outputRawChangeEvent: outputRawChangeEvent,
 	}
 	s.alive.transformer = transformer
 	s.alive.eventRouter = eventRouter
@@ -236,6 +239,6 @@ func (s *dmlSink) Dead() <-chan struct{} {
 }
 
 // Scheme returns the scheme of this sink.
-func (s *dmlSink) Scheme() string {
-	return s.scheme
+func (s *dmlSink) SchemeOption() (string, bool) {
+	return s.scheme, s.outputRawChangeEvent
 }
diff --git a/cdc/sink/dmlsink/mq/pulsar_dml_sink.go b/cdc/sink/dmlsink/mq/pulsar_dml_sink.go
index 089f521fd..f9934b591 100644
--- a/cdc/sink/dmlsink/mq/pulsar_dml_sink.go
+++ b/cdc/sink/dmlsink/mq/pulsar_dml_sink.go
@@ -123,8 +123,8 @@ func NewPulsarDMLSink(
 
 	encoderGroup := codec.NewEncoderGroup(replicaConfig.Sink, encoderBuilder, changefeedID)
 
-	s := newDMLSink(ctx, changefeedID, p, nil, topicManager,
-		eventRouter, trans, encoderGroup, protocol, scheme, errCh)
+	s := newDMLSink(ctx, changefeedID, p, nil, topicManager, eventRouter, trans, encoderGroup,
+		protocol, scheme, pConfig.GetOutputRawChangeEvent(), errCh)
 
 	return s, nil
 }
diff --git a/cdc/sink/dmlsink/txn/txn_dml_sink.go b/cdc/sink/dmlsink/txn/txn_dml_sink.go
index 964d11b6e..f5b2cb78e 100644
--- a/cdc/sink/dmlsink/txn/txn_dml_sink.go
+++ b/cdc/sink/dmlsink/txn/txn_dml_sink.go
@@ -180,6 +180,6 @@ func (s *dmlSink) Dead() <-chan struct{} {
 	return s.dead
 }
 
-func (s *dmlSink) Scheme() string {
-	return s.scheme
+func (s *dmlSink) SchemeOption() (string, bool) {
+	return s.scheme, false
 }
diff --git a/cdc/sink/dmlsink/txn/worker.go b/cdc/sink/dmlsink/txn/worker.go
index d71be238e..d9727618b 100644
--- a/cdc/sink/dmlsink/txn/worker.go
+++ b/cdc/sink/dmlsink/txn/worker.go
@@ -46,6 +46,8 @@ type worker struct {
 	flushInterval            time.Duration
 	hasPending               bool
 	postTxnExecutedCallbacks []func()
+
+	lastSlowConflictDetectLog map[model.TableID]time.Time
 }
 
 func newWorker(ctx context.Context, changefeedID model.ChangeFeedID,
@@ -70,6 +72,8 @@ func newWorker(ctx context.Context, changefeedID model.ChangeFeedID,
 		flushInterval:            backend.MaxFlushInterval(),
 		hasPending:               false,
 		postTxnExecutedCallbacks: make([]func(), 0, 1024),
+
+		lastSlowConflictDetectLog: make(map[model.TableID]time.Time),
 	}
 }
 
@@ -87,6 +91,9 @@ func (w *worker) runLoop(txnCh <-chan causality.TxnWithNotifier[*txnEvent]) erro
 		zap.String("changefeedID", w.changefeed),
 		zap.Int("workerID", w.ID))
 
+	cleanSlowLogHistory := time.NewTicker(time.Hour)
+	defer cleanSlowLogHistory.Stop()
+
 	start := time.Now()
 	for {
 		select {
@@ -95,6 +102,15 @@ func (w *worker) runLoop(txnCh <-chan causality.TxnWithNotifier[*txnEvent]) erro
 				zap.String("changefeedID", w.changefeed),
 				zap.Int("workerID", w.ID))
 			return nil
+		case <-cleanSlowLogHistory.C:
+			lastSlowConflictDetectLog := w.lastSlowConflictDetectLog
+			w.lastSlowConflictDetectLog = make(map[model.TableID]time.Time)
+			now := time.Now()
+			for tableID, lastLog := range lastSlowConflictDetectLog {
+				if now.Sub(lastLog) <= time.Minute {
+					w.lastSlowConflictDetectLog[tableID] = lastLog
+				}
+			}
 		case txn := <-txnCh:
 			// we get the data from txnCh until no more data here or reach the state that can be flushed.
 			// If no more data in txnCh, and also not reach the state that can be flushed,
@@ -151,8 +167,24 @@ func (w *worker) onEvent(txn *txnEvent, postTxnExecuted func()) bool {
 		return false
 	}
 
-	w.metricConflictDetectDuration.Observe(txn.conflictResolved.Sub(txn.start).Seconds())
+	conflictDetectTime := txn.conflictResolved.Sub(txn.start).Seconds()
+	w.metricConflictDetectDuration.Observe(conflictDetectTime)
 	w.metricQueueDuration.Observe(time.Since(txn.start).Seconds())
+
+	// Log tables which conflict detect time larger than 1 minute.
+	if conflictDetectTime > float64(60) {
+		now := time.Now()
+		// Log slow conflict detect tables every minute.
+		if lastLog, ok := w.lastSlowConflictDetectLog[txn.Event.PhysicalTableID]; !ok || now.Sub(lastLog) > time.Minute {
+			log.Warn("Transaction dmlSink finds a slow transaction in conflict detector",
+				zap.String("changefeedID", w.changefeed),
+				zap.Int("workerID", w.ID),
+				zap.Int64("TableID", txn.Event.PhysicalTableID),
+				zap.Float64("seconds", conflictDetectTime))
+			w.lastSlowConflictDetectLog[txn.Event.PhysicalTableID] = now
+		}
+	}
+
 	w.metricTxnWorkerHandledRows.Add(float64(len(txn.Event.Rows)))
 	w.postTxnExecutedCallbacks = append(w.postTxnExecutedCallbacks, postTxnExecuted)
 	return w.backend.OnTxnEvent(txn.TxnCallbackableEvent)
diff --git a/cdc/sink/tablesink/table_sink_impl.go b/cdc/sink/tablesink/table_sink_impl.go
index 167f30fc9..7e6b7d715 100644
--- a/cdc/sink/tablesink/table_sink_impl.go
+++ b/cdc/sink/tablesink/table_sink_impl.go
@@ -138,7 +138,7 @@ func (e *EventTableSink[E, P]) UpdateResolvedTs(resolvedTs model.ResolvedTs) err
 
 	resolvedCallbackableEvents := make([]*dmlsink.CallbackableEvent[E], 0, len(resolvedEvents))
 	for _, ev := range resolvedEvents {
-		if err := ev.TrySplitAndSortUpdateEvent(e.backendSink.Scheme()); err != nil {
+		if err := ev.TrySplitAndSortUpdateEvent(e.backendSink.SchemeOption()); err != nil {
 			return SinkInternalError{err}
 		}
 		// We have to record the event ID for the callback.
diff --git a/docs/swagger/docs.go b/docs/swagger/docs.go
index 4f605257d..ad9763034 100644
--- a/docs/swagger/docs.go
+++ b/docs/swagger/docs.go
@@ -1447,6 +1447,10 @@ var doc = `{
                 "output-column-id": {
                     "type": "boolean"
                 },
+                "output-raw-change-event": {
+                    "description": "OutputRawChangeEvent controls whether to split the update pk/uk events.",
+                    "type": "boolean"
+                },
                 "worker-count": {
                     "type": "integer"
                 }
@@ -1602,6 +1606,10 @@ var doc = `{
                 "max-message-bytes": {
                     "type": "integer"
                 },
+                "output-raw-change-event": {
+                    "description": "OutputRawChangeEvent controls whether to split the update pk/uk events.",
+                    "type": "boolean"
+                },
                 "partition-num": {
                     "type": "integer"
                 },
@@ -1817,6 +1825,10 @@ var doc = `{
                     "description": "Set the operation timeout (default: 30 seconds)\nProducer-create, subscribe and unsubscribe operations will be retried until this interval, after which the\noperation will be marked as failed",
                     "type": "integer"
                 },
+                "output-raw-change-event": {
+                    "description": "OutputRawChangeEvent controls whether to split the update pk/uk events.",
+                    "type": "boolean"
+                },
                 "pulsar-producer-cache-size": {
                     "description": "PulsarProducerCacheSize is the size of the cache of pulsar producers",
                     "type": "integer"
@@ -2452,6 +2464,9 @@ var doc = `{
                 "output_column_id": {
                     "type": "boolean"
                 },
+                "output_raw_change_event": {
+                    "type": "boolean"
+                },
                 "worker_count": {
                     "type": "integer"
                 }
@@ -2723,6 +2738,9 @@ var doc = `{
                 "max_message_bytes": {
                     "type": "integer"
                 },
+                "output_raw_change_event": {
+                    "type": "boolean"
+                },
                 "partition_num": {
                     "type": "integer"
                 },
@@ -2944,6 +2962,9 @@ var doc = `{
                 "operation-timeout": {
                     "type": "integer"
                 },
+                "output-raw-change-event": {
+                    "type": "boolean"
+                },
                 "pulsar-producer-cache-size": {
                     "type": "integer"
                 },
diff --git a/docs/swagger/swagger.json b/docs/swagger/swagger.json
index 38f0c2008..8d4f5d190 100644
--- a/docs/swagger/swagger.json
+++ b/docs/swagger/swagger.json
@@ -1428,6 +1428,10 @@
                 "output-column-id": {
                     "type": "boolean"
                 },
+                "output-raw-change-event": {
+                    "description": "OutputRawChangeEvent controls whether to split the update pk/uk events.",
+                    "type": "boolean"
+                },
                 "worker-count": {
                     "type": "integer"
                 }
@@ -1583,6 +1587,10 @@
                 "max-message-bytes": {
                     "type": "integer"
                 },
+                "output-raw-change-event": {
+                    "description": "OutputRawChangeEvent controls whether to split the update pk/uk events.",
+                    "type": "boolean"
+                },
                 "partition-num": {
                     "type": "integer"
                 },
@@ -1798,6 +1806,10 @@
                     "description": "Set the operation timeout (default: 30 seconds)\nProducer-create, subscribe and unsubscribe operations will be retried until this interval, after which the\noperation will be marked as failed",
                     "type": "integer"
                 },
+                "output-raw-change-event": {
+                    "description": "OutputRawChangeEvent controls whether to split the update pk/uk events.",
+                    "type": "boolean"
+                },
                 "pulsar-producer-cache-size": {
                     "description": "PulsarProducerCacheSize is the size of the cache of pulsar producers",
                     "type": "integer"
@@ -2433,6 +2445,9 @@
                 "output_column_id": {
                     "type": "boolean"
                 },
+                "output_raw_change_event": {
+                    "type": "boolean"
+                },
                 "worker_count": {
                     "type": "integer"
                 }
@@ -2704,6 +2719,9 @@
                 "max_message_bytes": {
                     "type": "integer"
                 },
+                "output_raw_change_event": {
+                    "type": "boolean"
+                },
                 "partition_num": {
                     "type": "integer"
                 },
@@ -2925,6 +2943,9 @@
                 "operation-timeout": {
                     "type": "integer"
                 },
+                "output-raw-change-event": {
+                    "type": "boolean"
+                },
                 "pulsar-producer-cache-size": {
                     "type": "integer"
                 },
diff --git a/docs/swagger/swagger.yaml b/docs/swagger/swagger.yaml
index 11113b710..9cb8b63d7 100644
--- a/docs/swagger/swagger.yaml
+++ b/docs/swagger/swagger.yaml
@@ -40,6 +40,10 @@ definitions:
         type: string
       output-column-id:
         type: boolean
+      output-raw-change-event:
+        description: OutputRawChangeEvent controls whether to split the update pk/uk
+          events.
+        type: boolean
       worker-count:
         type: integer
     type: object
@@ -147,6 +151,10 @@ definitions:
         $ref: '#/definitions/config.LargeMessageHandleConfig'
       max-message-bytes:
         type: integer
+      output-raw-change-event:
+        description: OutputRawChangeEvent controls whether to split the update pk/uk
+          events.
+        type: boolean
       partition-num:
         type: integer
       read-timeout:
@@ -306,6 +314,10 @@ definitions:
           Producer-create, subscribe and unsubscribe operations will be retried until this interval, after which the
           operation will be marked as failed
         type: integer
+      output-raw-change-event:
+        description: OutputRawChangeEvent controls whether to split the update pk/uk
+          events.
+        type: boolean
       pulsar-producer-cache-size:
         description: PulsarProducerCacheSize is the size of the cache of pulsar producers
         type: integer
@@ -764,6 +776,8 @@ definitions:
         type: string
       output_column_id:
         type: boolean
+      output_raw_change_event:
+        type: boolean
       worker_count:
         type: integer
     type: object
@@ -943,6 +957,8 @@ definitions:
         $ref: '#/definitions/v2.LargeMessageHandleConfig'
       max_message_bytes:
         type: integer
+      output_raw_change_event:
+        type: boolean
       partition_num:
         type: integer
       read_timeout:
@@ -1088,6 +1104,8 @@ definitions:
         $ref: '#/definitions/v2.PulsarOAuth2'
       operation-timeout:
         type: integer
+      output-raw-change-event:
+        type: boolean
       pulsar-producer-cache-size:
         type: integer
       pulsar-version:
diff --git a/errors.toml b/errors.toml
index ac4413f46..8ae2a3abb 100755
--- a/errors.toml
+++ b/errors.toml
@@ -916,6 +916,11 @@ error = '''
 cdc server is not ready
 '''
 
+["CDC:ErrSinkIncompatibleConfig"]
+error = '''
+incompatible configuration %s
+'''
+
 ["CDC:ErrSinkInvalidConfig"]
 error = '''
 sink config invalid
diff --git a/pkg/config/sink.go b/pkg/config/sink.go
index 44e1a3dbc..9954fa4f5 100644
--- a/pkg/config/sink.go
+++ b/pkg/config/sink.go
@@ -134,8 +134,6 @@ type SinkConfig struct {
 
 	// DispatchRules is only available when the downstream is MQ.
 	DispatchRules []*DispatchRule `toml:"dispatchers" json:"dispatchers,omitempty"`
-	// CSVConfig is only available when the downstream is Storage.
-	CSVConfig *CSVConfig `toml:"csv" json:"csv,omitempty"`
 
 	ColumnSelectors []*ColumnSelector `toml:"column-selectors" json:"column-selectors,omitempty"`
 	// SchemaRegistry is only available when the downstream is MQ using avro protocol.
@@ -194,9 +192,10 @@ type SinkConfig struct {
 	// Debezium only. Whether schema should be excluded in the output.
 	DebeziumDisableSchema *bool `toml:"debezium-disable-schema" json:"debezium-disable-schema,omitempty"`
 
+	// CSVConfig is only available when the downstream is Storage.
+	CSVConfig *CSVConfig `toml:"csv" json:"csv,omitempty"`
 	// OpenProtocol related configurations
 	OpenProtocol *OpenProtocolConfig `toml:"open" json:"open,omitempty"`
-
 	// DebeziumConfig related configurations
 	Debezium *DebeziumConfig `toml:"debezium" json:"debezium,omitempty"`
 }
@@ -434,6 +433,17 @@ type KafkaConfig struct {
 	CodecConfig                  *CodecConfig              `toml:"codec-config" json:"codec-config,omitempty"`
 	LargeMessageHandle           *LargeMessageHandleConfig `toml:"large-message-handle" json:"large-message-handle,omitempty"`
 	GlueSchemaRegistryConfig     *GlueSchemaRegistryConfig `toml:"glue-schema-registry-config" json:"glue-schema-registry-config"`
+
+	// OutputRawChangeEvent controls whether to split the update pk/uk events.
+	OutputRawChangeEvent *bool `toml:"output-raw-change-event" json:"output-raw-change-event,omitempty"`
+}
+
+// GetOutputRawChangeEvent returns the value of OutputRawChangeEvent
+func (k *KafkaConfig) GetOutputRawChangeEvent() bool {
+	if k == nil || k.OutputRawChangeEvent == nil {
+		return false
+	}
+	return *k.OutputRawChangeEvent
 }
 
 // MaskSensitiveData masks sensitive data in KafkaConfig
@@ -588,6 +598,9 @@ type PulsarConfig struct {
 	// and 'type' always use 'client_credentials'
 	OAuth2 *OAuth2 `toml:"oauth2" json:"oauth2,omitempty"`
 
+	// OutputRawChangeEvent controls whether to split the update pk/uk events.
+	OutputRawChangeEvent *bool `toml:"output-raw-change-event" json:"output-raw-change-event,omitempty"`
+
 	// BrokerURL is used to configure service brokerUrl for the Pulsar service.
 	// This parameter is a part of the `sink-uri`. Internal use only.
 	BrokerURL string `toml:"-" json:"-"`
@@ -595,6 +608,14 @@ type PulsarConfig struct {
 	SinkURI *url.URL `toml:"-" json:"-"`
 }
 
+// GetOutputRawChangeEvent returns the value of OutputRawChangeEvent
+func (c *PulsarConfig) GetOutputRawChangeEvent() bool {
+	if c == nil || c.OutputRawChangeEvent == nil {
+		return false
+	}
+	return *c.OutputRawChangeEvent
+}
+
 // MaskSensitiveData masks sensitive data in PulsarConfig
 func (c *PulsarConfig) MaskSensitiveData() {
 	if c.AuthenticationToken != nil {
@@ -653,6 +674,17 @@ type CloudStorageConfig struct {
 	FileExpirationDays  *int    `toml:"file-expiration-days" json:"file-expiration-days,omitempty"`
 	FileCleanupCronSpec *string `toml:"file-cleanup-cron-spec" json:"file-cleanup-cron-spec,omitempty"`
 	FlushConcurrency    *int    `toml:"flush-concurrency" json:"flush-concurrency,omitempty"`
+
+	// OutputRawChangeEvent controls whether to split the update pk/uk events.
+	OutputRawChangeEvent *bool `toml:"output-raw-change-event" json:"output-raw-change-event,omitempty"`
+}
+
+// GetOutputRawChangeEvent returns the value of OutputRawChangeEvent
+func (c *CloudStorageConfig) GetOutputRawChangeEvent() bool {
+	if c == nil || c.OutputRawChangeEvent == nil {
+		return false
+	}
+	return *c.OutputRawChangeEvent
 }
 
 func (s *SinkConfig) validateAndAdjust(sinkURI *url.URL) error {
@@ -796,21 +828,64 @@ func (s *SinkConfig) validateAndAdjustSinkURI(sinkURI *url.URL) error {
 		return err
 	}
 
-	// Adjust that protocol is compatible with the scheme. For testing purposes,
-	// any protocol should be legal for blackhole.
-	if sink.IsMQScheme(sinkURI.Scheme) || sink.IsStorageScheme(sinkURI.Scheme) {
-		_, err := ParseSinkProtocolFromString(util.GetOrZero(s.Protocol))
-		if err != nil {
-			return err
-		}
-	} else if sink.IsMySQLCompatibleScheme(sinkURI.Scheme) && s.Protocol != nil {
+	log.Info("succeed to parse parameter from sink uri",
+		zap.String("protocol", util.GetOrZero(s.Protocol)),
+		zap.String("txnAtomicity", string(util.GetOrZero(s.TxnAtomicity))))
+
+	// Check that protocol config is compatible with the scheme.
+	if sink.IsMySQLCompatibleScheme(sinkURI.Scheme) && s.Protocol != nil {
 		return cerror.ErrSinkURIInvalid.GenWithStackByArgs(fmt.Sprintf("protocol %s "+
 			"is incompatible with %s scheme", util.GetOrZero(s.Protocol), sinkURI.Scheme))
 	}
+	// For testing purposes, any protocol should be legal for blackhole.
+	if sink.IsMQScheme(sinkURI.Scheme) || sink.IsStorageScheme(sinkURI.Scheme) {
+		return s.ValidateProtocol(sinkURI.Scheme)
+	}
+	return nil
+}
 
-	log.Info("succeed to parse parameter from sink uri",
-		zap.String("protocol", util.GetOrZero(s.Protocol)),
-		zap.String("txnAtomicity", string(util.GetOrZero(s.TxnAtomicity))))
+// ValidateProtocol validates the protocol configuration.
+func (s *SinkConfig) ValidateProtocol(scheme string) error {
+	protocol, err := ParseSinkProtocolFromString(util.GetOrZero(s.Protocol))
+	if err != nil {
+		return err
+	}
+	outputOldValue := false
+	switch protocol {
+	case ProtocolOpen:
+		if s.OpenProtocol != nil {
+			outputOldValue = s.OpenProtocol.OutputOldValue
+		}
+	case ProtocolDebezium:
+		if s.Debezium != nil {
+			outputOldValue = s.Debezium.OutputOldValue
+		}
+	case ProtocolCsv:
+		if s.CSVConfig != nil {
+			outputOldValue = s.CSVConfig.OutputOldValue
+		}
+	case ProtocolAvro:
+		outputOldValue = false
+	default:
+		return nil
+	}
+
+	outputRawChangeEvent := false
+	switch scheme {
+	case sink.KafkaScheme, sink.KafkaSSLScheme:
+		outputRawChangeEvent = s.KafkaConfig.GetOutputRawChangeEvent()
+	case sink.PulsarScheme, sink.PulsarSSLScheme:
+		outputRawChangeEvent = s.PulsarConfig.GetOutputRawChangeEvent()
+	default:
+		outputRawChangeEvent = s.CloudStorageConfig.GetOutputRawChangeEvent()
+	}
+
+	if outputRawChangeEvent && !outputOldValue {
+		// TODO: return error if we do not need to keep backward compatibility.
+		log.Warn(fmt.Sprintf("TiCDC will not split the update pk/uk events if output-raw-change-event is true(scheme: %s).", scheme) +
+			fmt.Sprintf("It is not recommended to set output-old-value to false(protocol: %s) in this case.", protocol) +
+			"Otherwise, there may be data consistency issues in update pk/uk scenarios due to lack of old value information.")
+	}
 	return nil
 }
 
diff --git a/pkg/errors/cdc_errors.go b/pkg/errors/cdc_errors.go
index e3a2fada8..cba2637a1 100644
--- a/pkg/errors/cdc_errors.go
+++ b/pkg/errors/cdc_errors.go
@@ -400,6 +400,10 @@ var (
 		"sink config invalid",
 		errors.RFCCodeText("CDC:ErrSinkInvalidConfig"),
 	)
+	ErrSinkIncompatibleConfig = errors.Normalize(
+		"incompatible configuration %s",
+		errors.RFCCodeText("CDC:ErrSinkIncompatibleConfig"),
+	)
 	ErrCraftCodecInvalidData = errors.Normalize(
 		"craft codec invalid data",
 		errors.RFCCodeText("CDC:ErrCraftCodecInvalidData"),
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index d7c73a729..3441322ae 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -37,6 +37,7 @@ var ddlWhiteListMap = map[timodel.ActionType]bf.EventType{
 
 	// table related DDLs
 	timodel.ActionCreateTable:                  bf.CreateTable,
+	timodel.ActionCreateTables:                 bf.CreateTable,
 	timodel.ActionDropTable:                    bf.DropTable,
 	timodel.ActionTruncateTable:                bf.TruncateTable,
 	timodel.ActionRenameTable:                  bf.RenameTable,
diff --git a/pkg/sink/codec/csv/csv_message.go b/pkg/sink/codec/csv/csv_message.go
index ccea08f28..32486b2fe 100644
--- a/pkg/sink/codec/csv/csv_message.go
+++ b/pkg/sink/codec/csv/csv_message.go
@@ -174,6 +174,11 @@ func (c *csvMessage) decode(datums []types.Datum) error {
 	} else {
 		c.commitTs = 0
 	}
+	if c.config.OutputOldValue {
+		// When c.config.OutputOldValue, we need an extra column "is-updated".
+		// TODO: use this flag to guarantee data consistency in update uk/pk scenario.
+		dataColIdx++
+	}
 	c.columns = c.columns[:0]
 
 	for i := dataColIdx; i < len(datums); i++ {
diff --git a/pkg/sink/codec/utils/field_types.go b/pkg/sink/codec/utils/field_types.go
index b6f8e1b51..729737d19 100644
--- a/pkg/sink/codec/utils/field_types.go
+++ b/pkg/sink/codec/utils/field_types.go
@@ -15,33 +15,8 @@ package utils
 
 import (
 	"strings"
-
-	"github.com/pingcap/tidb/pkg/parser/charset"
-	"github.com/pingcap/tidb/pkg/parser/mysql"
-	"github.com/pingcap/tidb/pkg/parser/types"
-	"github.com/pingcap/tiflow/cdc/model"
 )
 
-// SetBinChsClnFlag set the binary charset flag.
-func SetBinChsClnFlag(ft *types.FieldType) *types.FieldType {
-	ft.SetCharset(charset.CharsetBin)
-	ft.SetCollate(charset.CollationBin)
-	ft.AddFlag(mysql.BinaryFlag)
-	return ft
-}
-
-// SetUnsigned set the unsigned flag.
-func SetUnsigned(ft *types.FieldType) *types.FieldType {
-	ft.SetFlag(uint(model.UnsignedFlag))
-	return ft
-}
-
-// SetElems set the elems to the ft
-func SetElems(ft *types.FieldType, elems []string) *types.FieldType {
-	ft.SetElems(elems)
-	return ft
-}
-
 // when encoding the canal format, for unsigned mysql type, add `unsigned` keyword.
 // it should have the form `t unsigned`, such as `int unsigned`
 func withUnsigned4MySQLType(mysqlType string, unsigned bool) string {
diff --git a/pkg/sink/kafka/factory.go b/pkg/sink/kafka/factory.go
index d0b721d66..a8a8f05e0 100644
--- a/pkg/sink/kafka/factory.go
+++ b/pkg/sink/kafka/factory.go
@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	"github.com/pingcap/tiflow/pkg/util"
 	"go.uber.org/zap"
 )
@@ -48,15 +49,13 @@ type SyncProducer interface {
 	// of the produced message, or an error if the message failed to produce.
 	SendMessage(ctx context.Context,
 		topic string, partitionNum int32,
-		key []byte, value []byte) error
+		message *common.Message) error
 
 	// SendMessages produces a given set of messages, and returns only when all
 	// messages in the set have either succeeded or failed. Note that messages
 	// can succeed and fail individually; if some succeed and some fail,
 	// SendMessages will return an error.
-	SendMessages(ctx context.Context,
-		topic string, partitionNum int32,
-		key []byte, value []byte) error
+	SendMessages(ctx context.Context, topic string, partitionNum int32, message *common.Message) error
 
 	// Close shuts down the producer; you must call this function before a producer
 	// object passes out of scope, as it may otherwise leak memory.
@@ -75,9 +74,7 @@ type AsyncProducer interface {
 
 	// AsyncSend is the input channel for the user to write messages to that they
 	// wish to send.
-	AsyncSend(ctx context.Context, topic string,
-		partition int32, key []byte, value []byte,
-		callback func()) error
+	AsyncSend(ctx context.Context, topic string, partition int32, message *common.Message) error
 
 	// AsyncRunCallback process the messages that has sent to kafka,
 	// and run tha attached callback. the caller should call this
@@ -92,29 +89,26 @@ type saramaSyncProducer struct {
 }
 
 func (p *saramaSyncProducer) SendMessage(
-	ctx context.Context,
+	_ context.Context,
 	topic string, partitionNum int32,
-	key []byte, value []byte,
+	message *common.Message,
 ) error {
 	_, _, err := p.producer.SendMessage(&sarama.ProducerMessage{
 		Topic:     topic,
-		Key:       sarama.ByteEncoder(key),
-		Value:     sarama.ByteEncoder(value),
+		Key:       sarama.ByteEncoder(message.Key),
+		Value:     sarama.ByteEncoder(message.Value),
 		Partition: partitionNum,
 	})
 	return err
 }
 
-func (p *saramaSyncProducer) SendMessages(ctx context.Context,
-	topic string, partitionNum int32,
-	key []byte, value []byte,
-) error {
+func (p *saramaSyncProducer) SendMessages(ctx context.Context, topic string, partitionNum int32, message *common.Message) error {
 	msgs := make([]*sarama.ProducerMessage, partitionNum)
 	for i := 0; i < int(partitionNum); i++ {
 		msgs[i] = &sarama.ProducerMessage{
 			Topic:     topic,
-			Key:       sarama.ByteEncoder(key),
-			Value:     sarama.ByteEncoder(value),
+			Key:       sarama.ByteEncoder(message.Key),
+			Value:     sarama.ByteEncoder(message.Value),
 			Partition: int32(i),
 		}
 	}
@@ -259,19 +253,13 @@ func (p *saramaAsyncProducer) AsyncRunCallback(
 
 // AsyncSend is the input channel for the user to write messages to that they
 // wish to send.
-func (p *saramaAsyncProducer) AsyncSend(ctx context.Context,
-	topic string,
-	partition int32,
-	key []byte,
-	value []byte,
-	callback func(),
-) error {
+func (p *saramaAsyncProducer) AsyncSend(ctx context.Context, topic string, partition int32, message *common.Message) error {
 	msg := &sarama.ProducerMessage{
 		Topic:     topic,
 		Partition: partition,
-		Key:       sarama.StringEncoder(key),
-		Value:     sarama.ByteEncoder(value),
-		Metadata:  callback,
+		Key:       sarama.StringEncoder(message.Key),
+		Value:     sarama.ByteEncoder(message.Value),
+		Metadata:  message.Callback,
 	}
 	select {
 	case <-ctx.Done():
diff --git a/pkg/sink/kafka/mock_factory.go b/pkg/sink/kafka/mock_factory.go
index 676f2e74c..2d5b479b4 100644
--- a/pkg/sink/kafka/mock_factory.go
+++ b/pkg/sink/kafka/mock_factory.go
@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tiflow/cdc/model"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	"github.com/pingcap/tiflow/pkg/util"
 )
 
@@ -91,30 +92,27 @@ type MockSaramaSyncProducer struct {
 
 // SendMessage implement the SyncProducer interface.
 func (m *MockSaramaSyncProducer) SendMessage(
-	ctx context.Context,
+	_ context.Context,
 	topic string, partitionNum int32,
-	key []byte, value []byte,
+	message *common.Message,
 ) error {
 	_, _, err := m.Producer.SendMessage(&sarama.ProducerMessage{
 		Topic:     topic,
-		Key:       sarama.ByteEncoder(key),
-		Value:     sarama.ByteEncoder(value),
+		Key:       sarama.ByteEncoder(message.Key),
+		Value:     sarama.ByteEncoder(message.Value),
 		Partition: partitionNum,
 	})
 	return err
 }
 
 // SendMessages implement the SyncProducer interface.
-func (m *MockSaramaSyncProducer) SendMessages(ctx context.Context,
-	topic string, partitionNum int32,
-	key []byte, value []byte,
-) error {
+func (m *MockSaramaSyncProducer) SendMessages(ctx context.Context, topic string, partitionNum int32, message *common.Message) error {
 	msgs := make([]*sarama.ProducerMessage, partitionNum)
 	for i := 0; i < int(partitionNum); i++ {
 		msgs[i] = &sarama.ProducerMessage{
 			Topic:     topic,
-			Key:       sarama.ByteEncoder(key),
-			Value:     sarama.ByteEncoder(value),
+			Key:       sarama.ByteEncoder(message.Key),
+			Value:     sarama.ByteEncoder(message.Value),
 			Partition: int32(i),
 		}
 	}
@@ -166,16 +164,13 @@ func (p *MockSaramaAsyncProducer) AsyncRunCallback(
 }
 
 // AsyncSend implement the AsyncProducer interface.
-func (p *MockSaramaAsyncProducer) AsyncSend(ctx context.Context, topic string,
-	partition int32, key []byte, value []byte,
-	callback func(),
-) error {
+func (p *MockSaramaAsyncProducer) AsyncSend(ctx context.Context, topic string, partition int32, message *common.Message) error {
 	msg := &sarama.ProducerMessage{
 		Topic:     topic,
 		Partition: partition,
-		Key:       sarama.StringEncoder(key),
-		Value:     sarama.ByteEncoder(value),
-		Metadata:  callback,
+		Key:       sarama.StringEncoder(message.Key),
+		Value:     sarama.ByteEncoder(message.Value),
+		Metadata:  message.Callback,
 	}
 	select {
 	case <-ctx.Done():
diff --git a/pkg/sink/kafka/v2/factory.go b/pkg/sink/kafka/v2/factory.go
index 870ec48cf..7d0346f74 100644
--- a/pkg/sink/kafka/v2/factory.go
+++ b/pkg/sink/kafka/v2/factory.go
@@ -26,6 +26,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/security"
+	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	pkafka "github.com/pingcap/tiflow/pkg/sink/kafka"
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/segmentio/kafka-go"
@@ -275,13 +276,13 @@ type syncWriter struct {
 func (s *syncWriter) SendMessage(
 	ctx context.Context,
 	topic string, partitionNum int32,
-	key []byte, value []byte,
+	message *common.Message,
 ) error {
 	return s.w.WriteMessages(ctx, kafka.Message{
 		Topic:     topic,
 		Partition: int(partitionNum),
-		Key:       key,
-		Value:     value,
+		Key:       message.Key,
+		Value:     message.Value,
 	})
 }
 
@@ -289,17 +290,13 @@ func (s *syncWriter) SendMessage(
 // messages in the set have either succeeded or failed. Note that messages
 // can succeed and fail individually; if some succeed and some fail,
 // SendMessages will return an error.
-func (s *syncWriter) SendMessages(
-	ctx context.Context,
-	topic string, partitionNum int32,
-	key []byte, value []byte,
-) error {
+func (s *syncWriter) SendMessages(ctx context.Context, topic string, partitionNum int32, message *common.Message) error {
 	msgs := make([]kafka.Message, int(partitionNum))
 	for i := 0; i < int(partitionNum); i++ {
 		msgs[i] = kafka.Message{
 			Topic:     topic,
-			Key:       key,
-			Value:     value,
+			Key:       message.Key,
+			Value:     message.Value,
 			Partition: i,
 		}
 	}
@@ -363,10 +360,7 @@ func (a *asyncWriter) Close() {
 
 // AsyncSend is the input channel for the user to write messages to that they
 // wish to send.
-func (a *asyncWriter) AsyncSend(ctx context.Context, topic string,
-	partition int32, key []byte, value []byte,
-	callback func(),
-) error {
+func (a *asyncWriter) AsyncSend(ctx context.Context, topic string, partition int32, message *common.Message) error {
 	select {
 	case <-ctx.Done():
 		return errors.Trace(ctx.Err())
@@ -375,9 +369,9 @@ func (a *asyncWriter) AsyncSend(ctx context.Context, topic string,
 	return a.w.WriteMessages(ctx, kafka.Message{
 		Topic:      topic,
 		Partition:  int(partition),
-		Key:        key,
-		Value:      value,
-		WriterData: callback,
+		Key:        message.Key,
+		Value:      message.Value,
+		WriterData: message.Callback,
 	})
 }
 


Test patch:
diff --git a/cdc/model/sink_test.go b/cdc/model/sink_test.go
index cd285b3ff..f77e0f2df 100644
--- a/cdc/model/sink_test.go
+++ b/cdc/model/sink_test.go
@@ -604,21 +604,32 @@ func TestTxnTrySplitAndSortUpdateEvent(t *testing.T) {
 		Rows: []*RowChangedEvent{ukUpdatedEvent},
 	}
 
-	err := txn.TrySplitAndSortUpdateEvent(sink.KafkaScheme)
+	outputRawChangeEvent := true
+	notOutputRawChangeEvent := false
+	err := txn.TrySplitAndSortUpdateEvent(sink.KafkaScheme, outputRawChangeEvent)
+	require.NoError(t, err)
+	require.Len(t, txn.Rows, 1)
+	err = txn.TrySplitAndSortUpdateEvent(sink.KafkaScheme, notOutputRawChangeEvent)
 	require.NoError(t, err)
 	require.Len(t, txn.Rows, 2)
 
 	txn = &SingleTableTxn{
 		Rows: []*RowChangedEvent{ukUpdatedEvent},
 	}
-	err = txn.TrySplitAndSortUpdateEvent(sink.MySQLScheme)
+	err = txn.TrySplitAndSortUpdateEvent(sink.MySQLScheme, outputRawChangeEvent)
+	require.NoError(t, err)
+	require.Len(t, txn.Rows, 1)
+	err = txn.TrySplitAndSortUpdateEvent(sink.MySQLScheme, notOutputRawChangeEvent)
 	require.NoError(t, err)
 	require.Len(t, txn.Rows, 1)
 
 	txn2 := &SingleTableTxn{
 		Rows: []*RowChangedEvent{ukUpdatedEvent, ukUpdatedEvent},
 	}
-	err = txn.TrySplitAndSortUpdateEvent(sink.MySQLScheme)
+	err = txn.TrySplitAndSortUpdateEvent(sink.MySQLScheme, outputRawChangeEvent)
+	require.NoError(t, err)
+	require.Len(t, txn2.Rows, 2)
+	err = txn.TrySplitAndSortUpdateEvent(sink.MySQLScheme, notOutputRawChangeEvent)
 	require.NoError(t, err)
 	require.Len(t, txn2.Rows, 2)
 }
diff --git a/cdc/owner/changefeed_test.go b/cdc/owner/changefeed_test.go
index 0866638b0..c384ecd59 100644
--- a/cdc/owner/changefeed_test.go
+++ b/cdc/owner/changefeed_test.go
@@ -684,13 +684,7 @@ func TestBarrierAdvance(t *testing.T) {
 			require.Nil(t, err)
 			err = cf.handleBarrier(ctx, state.Info, state.Status, barrier)
 
-			nextSyncPointTs := oracle.GoTimeToTS(
-				oracle.GetTimeFromTS(state.Status.CheckpointTs + 10).
-					Add(util.GetOrZero(changefeedInfo.Config.SyncPointInterval)),
-			)
-
 			require.Nil(t, err)
-			require.Equal(t, nextSyncPointTs, barrier.GlobalBarrierTs)
 			require.Less(t, state.Status.CheckpointTs+10, barrier.GlobalBarrierTs)
 			require.Less(t, barrier.GlobalBarrierTs, cf.ddlManager.ddlResolvedTs)
 		}
diff --git a/cdc/processor/sinkmanager/table_sink_wrapper_test.go b/cdc/processor/sinkmanager/table_sink_wrapper_test.go
index 2a986ec13..f94f889f8 100644
--- a/cdc/processor/sinkmanager/table_sink_wrapper_test.go
+++ b/cdc/processor/sinkmanager/table_sink_wrapper_test.go
@@ -52,8 +52,8 @@ func (m *mockSink) WriteEvents(events ...*dmlsink.CallbackableEvent[*model.RowCh
 	return nil
 }
 
-func (m *mockSink) Scheme() string {
-	return sink.BlackHoleScheme
+func (m *mockSink) SchemeOption() (string, bool) {
+	return sink.BlackHoleScheme, false
 }
 
 func (m *mockSink) GetEvents() []*dmlsink.CallbackableEvent[*model.RowChangedEvent] {
diff --git a/cdc/sink/tablesink/table_sink_impl_test.go b/cdc/sink/tablesink/table_sink_impl_test.go
index 87fd1af4a..b7c55d38f 100644
--- a/cdc/sink/tablesink/table_sink_impl_test.go
+++ b/cdc/sink/tablesink/table_sink_impl_test.go
@@ -50,8 +50,8 @@ func (m *mockEventSink) Dead() <-chan struct{} {
 	return m.dead
 }
 
-func (m *mockEventSink) Scheme() string {
-	return sink.BlackHoleScheme
+func (m *mockEventSink) SchemeOption() (string, bool) {
+	return sink.BlackHoleScheme, false
 }
 
 // acknowledge the txn events by call the callback function.
diff --git a/pkg/filter/filter_test.go b/pkg/filter/filter_test.go
index 9ed19e04c..1d296eb0d 100644
--- a/pkg/filter/filter_test.go
+++ b/pkg/filter/filter_test.go
@@ -371,7 +371,7 @@ func TestShouldDiscardDDL(t *testing.T) {
 }
 
 func TestIsAllowedDDL(t *testing.T) {
-	require.Len(t, ddlWhiteListMap, 36)
+	require.Len(t, ddlWhiteListMap, 37)
 	type testCase struct {
 		timodel.ActionType
 		allowed bool
diff --git a/pkg/sink/codec/csv/csv_decoder_test.go b/pkg/sink/codec/csv/csv_decoder_test.go
index b5ba921dc..81e77ae5e 100644
--- a/pkg/sink/codec/csv/csv_decoder_test.go
+++ b/pkg/sink/codec/csv/csv_decoder_test.go
@@ -16,9 +16,7 @@ import (
 	"context"
 	"testing"
 
-	timodel "github.com/pingcap/tidb/pkg/parser/model"
-	"github.com/pingcap/tidb/pkg/parser/mysql"
-	"github.com/pingcap/tidb/pkg/parser/types"
+	"github.com/pingcap/tiflow/cdc/entry"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	"github.com/stretchr/testify/require"
@@ -32,45 +30,21 @@ func TestCSVBatchDecoder(t *testing.T) {
 "U","employee","hr",433305438660591630,102,"Alex","Alice","2018-06-15","Beijing"
 `
 	ctx := context.Background()
-	tableInfo := &model.TableInfo{
-		TableName: model.TableName{
-			Schema: "hr",
-			Table:  "employee",
-		},
-		TableInfo: &timodel.TableInfo{
-			Name: timodel.NewCIStr("employee"),
-			Columns: []*timodel.ColumnInfo{
-				{
-					Name:      timodel.NewCIStr("Id"),
-					FieldType: *types.NewFieldType(mysql.TypeInt24),
-				},
-				{
-					Name:      timodel.NewCIStr("LastName"),
-					FieldType: *types.NewFieldType(mysql.TypeVarchar),
-				},
-				{
-					Name:      timodel.NewCIStr("FirstName"),
-					FieldType: *types.NewFieldType(mysql.TypeVarchar),
-				},
-				{
-					Name:      timodel.NewCIStr("HireDate"),
-					FieldType: *types.NewFieldType(mysql.TypeDate),
-				},
-				{
-					Name:      timodel.NewCIStr("OfficeLocation"),
-					FieldType: *types.NewFieldType(mysql.TypeVarchar),
-				},
-			},
-		},
-	}
-	decoder, err := NewBatchDecoder(ctx, &common.Config{
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
+
+	_ = helper.DDL2Event("create database hr")
+	createTableDDL := helper.DDL2Event("create table hr.employee(Id int, LastName varchar(255), FirstName varchar(255), HireDate date, OfficeLocation varchar(255))")
+
+	codecConfig := &common.Config{
 		Delimiter:       ",",
 		Quote:           "\"",
 		Terminator:      "\n",
 		NullString:      "\\N",
 		IncludeCommitTs: true,
-	}, tableInfo, []byte(csvData))
-	require.Nil(t, err)
+	}
+	decoder, err := NewBatchDecoder(ctx, codecConfig, createTableDDL.TableInfo, []byte(csvData))
+	require.NoError(t, err)
 
 	for i := 0; i < 5; i++ {
 		tp, hasNext, err := decoder.HasNext()
diff --git a/pkg/sink/codec/csv/csv_encoder_test.go b/pkg/sink/codec/csv/csv_encoder_test.go
index 4f3661ebf..588c91b7b 100644
--- a/pkg/sink/codec/csv/csv_encoder_test.go
+++ b/pkg/sink/codec/csv/csv_encoder_test.go
@@ -16,40 +16,29 @@ package csv
 import (
 	"testing"
 
-	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tiflow/cdc/entry"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	"github.com/stretchr/testify/require"
 )
 
 func TestCSVBatchCodec(t *testing.T) {
-	tableInfo := model.BuildTableInfo("test", "table1", []*model.Column{{
-		Name: "tiny",
-		Type: mysql.TypeTiny,
-	}}, nil)
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
+
+	ddl := helper.DDL2Event("create table test.table1(col1 int primary key)")
+	event1 := helper.DML2Event("insert into test.table1 values (1)", "test", "table1")
+	event2 := helper.DML2Event("insert into test.table1 values (2)", "test", "table1")
+
 	testCases := []*model.SingleTableTxn{
 		{
 			Rows: []*model.RowChangedEvent{
-				{
-					CommitTs:  1,
-					TableInfo: tableInfo,
-					Columns: model.Columns2ColumnDatas([]*model.Column{{
-						Name:  "tiny",
-						Value: int64(1),
-					}}, tableInfo),
-				},
-				{
-					CommitTs:  2,
-					TableInfo: tableInfo,
-					Columns: model.Columns2ColumnDatas([]*model.Column{{
-						Name:  "tiny",
-						Value: int64(2),
-					}}, tableInfo),
-				},
+				event1,
+				event2,
 			},
 		},
 		{
-			TableInfo: tableInfo,
+			TableInfo: ddl.TableInfo,
 			Rows:      nil,
 		},
 	}
@@ -85,19 +74,14 @@ func TestCSVAppendRowChangedEventWithCallback(t *testing.T) {
 	require.NotNil(t, encoder)
 
 	count := 0
-	tableInfo := model.BuildTableInfo("test", "table1", []*model.Column{{
-		Name:  "tiny",
-		Value: int64(1), Type: mysql.TypeTiny,
-	}}, nil)
-	row := &model.RowChangedEvent{
-		CommitTs:  1,
-		TableInfo: tableInfo,
 
-		Columns: model.Columns2ColumnDatas([]*model.Column{{Name: "tiny", Value: int64(1)}}, tableInfo),
-	}
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
 
+	_ = helper.DDL2Event("create table test.table1(col1 int primary key)")
+	row := helper.DML2Event("insert into test.table1 values (1)", "test", "table1")
 	txn := &model.SingleTableTxn{
-		TableInfo: tableInfo,
+		TableInfo: row.TableInfo,
 		Rows:      []*model.RowChangedEvent{row},
 	}
 	callback := func() {
diff --git a/pkg/sink/codec/maxwell/maxwell_encoder_test.go b/pkg/sink/codec/maxwell/maxwell_encoder_test.go
index 84f190d37..870998f6e 100644
--- a/pkg/sink/codec/maxwell/maxwell_encoder_test.go
+++ b/pkg/sink/codec/maxwell/maxwell_encoder_test.go
@@ -17,30 +17,22 @@ import (
 	"context"
 	"testing"
 
-	timodel "github.com/pingcap/tidb/pkg/parser/model"
-	"github.com/pingcap/tidb/pkg/parser/mysql"
+	"github.com/pingcap/tiflow/cdc/entry"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	"github.com/stretchr/testify/require"
 )
 
 func TestMaxwellBatchCodec(t *testing.T) {
-	t.Parallel()
-	newEncoder := newBatchEncoder
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
 
-	tableInfo := model.BuildTableInfo("a", "b", []*model.Column{{Name: "col1", Type: mysql.TypeLong}}, nil)
-	rowCases := [][]*model.RowChangedEvent{{{
-		CommitTs:  1,
-		TableInfo: tableInfo,
-		Columns: model.Columns2ColumnDatas([]*model.Column{
-			{
-				Name:  "col1",
-				Value: 10,
-			},
-		}, tableInfo),
-	}}, {}}
+	ddlEvent := helper.DDL2Event("create table test.t(col1 int primary key)")
+	dmlEvent := helper.DML2Event("insert into test.t values (10)", "test", "t")
+
+	rowCases := [][]*model.RowChangedEvent{{dmlEvent}, {}}
 	for _, cs := range rowCases {
-		encoder := newEncoder(&common.Config{})
+		encoder := newBatchEncoder(&common.Config{})
 		for _, row := range cs {
 			err := encoder.AppendRowChangedEvent(context.Background(), "", row, nil)
 			require.Nil(t, err)
@@ -54,19 +46,9 @@ func TestMaxwellBatchCodec(t *testing.T) {
 		require.Equal(t, len(cs), messages[0].GetRowsCount())
 	}
 
-	ddlCases := [][]*model.DDLEvent{{{
-		CommitTs: 1,
-		TableInfo: &model.TableInfo{
-			TableName: model.TableName{
-				Schema: "a", Table: "b",
-			},
-			TableInfo: &timodel.TableInfo{},
-		},
-		Query: "create table a",
-		Type:  1,
-	}}}
+	ddlCases := [][]*model.DDLEvent{{ddlEvent}}
 	for _, cs := range ddlCases {
-		encoder := newEncoder(&common.Config{})
+		encoder := newBatchEncoder(&common.Config{})
 		for _, ddl := range cs {
 			msg, err := encoder.EncodeDDLEvent(ddl)
 			require.Nil(t, err)
@@ -79,17 +61,12 @@ func TestMaxwellAppendRowChangedEventWithCallback(t *testing.T) {
 	encoder := newBatchEncoder(&common.Config{})
 	require.NotNil(t, encoder)
 
-	count := 0
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
 
-	tableInfo := model.BuildTableInfo("a", "b", []*model.Column{{Name: "col1", Type: mysql.TypeVarchar}}, nil)
-	row := &model.RowChangedEvent{
-		CommitTs:  1,
-		TableInfo: tableInfo,
-		Columns: model.Columns2ColumnDatas([]*model.Column{{
-			Name:  "col1",
-			Value: []byte("aa"),
-		}}, tableInfo),
-	}
+	_ = helper.DDL2Event("create table test.t(col1 varchar(255) primary key)")
+	row := helper.DML2Event("insert into test.t values ('aa')", "test", "t")
+	count := 0
 
 	tests := []struct {
 		row      *model.RowChangedEvent
diff --git a/pkg/sink/codec/maxwell/maxwell_message_test.go b/pkg/sink/codec/maxwell/maxwell_message_test.go
index d4905e83c..f2631c36f 100644
--- a/pkg/sink/codec/maxwell/maxwell_message_test.go
+++ b/pkg/sink/codec/maxwell/maxwell_message_test.go
@@ -16,8 +16,7 @@ package maxwell
 import (
 	"testing"
 
-	"github.com/pingcap/tidb/pkg/parser/mysql"
-	"github.com/pingcap/tiflow/cdc/model"
+	"github.com/pingcap/tiflow/cdc/entry"
 	"github.com/stretchr/testify/require"
 )
 
@@ -42,20 +41,11 @@ func TestMaxwellFormatCol(t *testing.T) {
 }
 
 func TestEncodeBinaryToMaxwell(t *testing.T) {
-	t.Parallel()
-
-	cols := []*model.Column{{
-		Name:  "varbinary",
-		Type:  mysql.TypeVarchar,
-		Value: []uint8("测试varbinary"),
-		Flag:  model.BinaryFlag,
-	}}
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
 
-	tableInfo := model.BuildTableInfo("a", "b", cols, nil)
-	e := &model.RowChangedEvent{
-		TableInfo: tableInfo,
-		Columns:   model.Columns2ColumnDatas(cols, tableInfo),
-	}
+	_ = helper.DDL2Event("create table test.t(col1 varbinary(255) primary key)")
+	e := helper.DML2Event("insert into test.t values ('测试varbinary')", "test", "t")
 
 	key, msg := rowChangeToMaxwellMsg(e, false)
 	require.NotNil(t, key)
diff --git a/pkg/sink/kafka/v2/factory_test.go b/pkg/sink/kafka/v2/factory_test.go
index 1f65850be..c86bc5554 100644
--- a/pkg/sink/kafka/v2/factory_test.go
+++ b/pkg/sink/kafka/v2/factory_test.go
@@ -23,6 +23,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/security"
+	"github.com/pingcap/tiflow/pkg/sink/codec/common"
 	pkafka "github.com/pingcap/tiflow/pkg/sink/kafka"
 	v2mock "github.com/pingcap/tiflow/pkg/sink/kafka/v2/mock"
 	"github.com/pingcap/tiflow/pkg/util"
@@ -213,7 +214,9 @@ func TestSyncWriterSendMessage(t *testing.T) {
 			require.Equal(t, 3, msgs[0].Partition)
 			return errors.New("fake")
 		})
-	require.NotNil(t, w.SendMessage(context.Background(), "topic", 3, []byte{'1'}, []byte{}))
+
+	message := &common.Message{Key: []byte{'1'}, Value: []byte{}}
+	require.NotNil(t, w.SendMessage(context.Background(), "topic", 3, message))
 }
 
 func TestSyncWriterSendMessages(t *testing.T) {
@@ -224,7 +227,8 @@ func TestSyncWriterSendMessages(t *testing.T) {
 			require.Equal(t, 3, len(msgs))
 			return errors.New("fake")
 		})
-	require.NotNil(t, w.SendMessages(context.Background(), "topic", 3, []byte{'1'}, []byte{}))
+	message := &common.Message{Key: []byte{'1'}, Value: []byte{}}
+	require.NotNil(t, w.SendMessages(context.Background(), "topic", 3, message))
 }
 
 func TestSyncWriterClose(t *testing.T) {
@@ -244,14 +248,14 @@ func TestAsyncWriterAsyncSend(t *testing.T) {
 
 	ctx, cancel := context.WithCancel(context.Background())
 
-	callback := func() {}
+	message := &common.Message{Key: []byte{'1'}, Value: []byte{}, Callback: func() {}}
 	mw.EXPECT().WriteMessages(gomock.Any(), gomock.Any()).Return(nil)
-	err := w.AsyncSend(ctx, "topic", 1, []byte{'1'}, []byte{}, callback)
+	err := w.AsyncSend(ctx, "topic", 1, message)
 	require.NoError(t, err)
 
 	cancel()
 
-	err = w.AsyncSend(ctx, "topic", 1, []byte{'1'}, []byte{}, callback)
+	err = w.AsyncSend(ctx, "topic", 1, nil)
 	require.ErrorIs(t, err, context.Canceled)
 }
 
diff --git a/tests/integration_tests/batch_add_table/conf/changefeed.toml b/tests/integration_tests/batch_add_table/conf/changefeed.toml
new file mode 100644
index 000000000..c125b104e
--- /dev/null
+++ b/tests/integration_tests/batch_add_table/conf/changefeed.toml
@@ -0,0 +1,2 @@
+[filter]
+rules = ['*.*','!test.t_1']
\ No newline at end of file
diff --git a/tests/integration_tests/batch_add_table/data/prepare.sql b/tests/integration_tests/batch_add_table/data/prepare.sql
index 83674029f..1f37cf311 100644
--- a/tests/integration_tests/batch_add_table/data/prepare.sql
+++ b/tests/integration_tests/batch_add_table/data/prepare.sql
@@ -7,9 +7,31 @@ create table a2 (id int primary key);
 create table a3 (id int primary key);
 create table a4 (id int primary key);
 create table a5 (id int primary key);
+create table a6 (id int primary key);
+create table a7 (id int primary key);
+create table a8 (id int primary key);
+create table a9 (id int primary key);
+create table a10 (id int primary key);
+create table a11 (id int primary key);
+create table a12 (id int primary key);
+create table a13 (id int primary key);
+create table a14 (id int primary key);
+create table a15 (id int primary key);
+create table a16 (id int primary key);
 
 insert into a1 values (1);
 insert into a2 values (1);
 insert into a3 values (1);
 insert into a4 values (1);
-insert into a5 values (1);
\ No newline at end of file
+insert into a5 values (1);
+insert into a6 values (1);
+insert into a7 values (1);
+insert into a8 values (1);
+insert into a9 values (1);
+insert into a10 values (1);
+insert into a11 values (1);
+insert into a12 values (1);
+insert into a13 values (1);
+insert into a14 values (1);
+insert into a15 values (1);
+insert into a16 values (1);
\ No newline at end of file
diff --git a/tests/integration_tests/batch_add_table/data/test.sql b/tests/integration_tests/batch_add_table/data/test.sql
index 77afc7191..f51b65b34 100644
--- a/tests/integration_tests/batch_add_table/data/test.sql
+++ b/tests/integration_tests/batch_add_table/data/test.sql
@@ -7,6 +7,16 @@ insert into a2 values (2);
 insert into a3 values (2);
 insert into a4 values (2);
 insert into a5 values (2);
-
+insert into a6 values (2);
+insert into a7 values (2);
+insert into a8 values (2);
+insert into a9 values (2);
+insert into a10 values (2);
+insert into a11 values (2);
+insert into a12 values (2);
+insert into a13 values (2);
+insert into a14 values (2);
+insert into a15 values (2);
+insert into a16 values (2);
 
 create table finish_mark (id int primary key);
\ No newline at end of file
diff --git a/tests/integration_tests/batch_add_table/run.sh b/tests/integration_tests/batch_add_table/run.sh
index 9cce075c5..427a496e2 100644
--- a/tests/integration_tests/batch_add_table/run.sh
+++ b/tests/integration_tests/batch_add_table/run.sh
@@ -17,9 +17,6 @@ function run_with_fast_create_table() {
 
 	run_sql "set global tidb_enable_fast_create_table=on" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
-	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	run_sql_file $CUR/data/prepare.sql ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
 
 	TOPIC_NAME="ticdc-batch-add-table-test-$RANDOM"
@@ -32,16 +29,31 @@ function run_with_fast_create_table() {
 		;;
 	*) SINK_URI="mysql://normal:123456@127.0.0.1:3306/" ;;
 	esac
-	run_cdc_cli changefeed create --sink-uri="$SINK_URI"
+	run_cdc_cli changefeed create --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml"
 	case $SINK_TYPE in
 	kafka) run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=4&version=${KAFKA_VERSION}&max-message-bytes=10485760" ;;
 	storage) run_storage_consumer $WORK_DIR $SINK_URI "" "" ;;
 	pulsar) run_pulsar_consumer --upstream-uri $SINK_URI ;;
 	esac
 
+	## to generate batch create ddl. In changefeed.toml, we filter test.t_1
+	for ((i = 1; i <= 100; i++)); do
+		mysql -h ${UP_TIDB_HOST} -P ${UP_TIDB_PORT} -u root -D "test" -e "create table t_$i (a int primary key , b int)" &
+	done
+
+	check_table_exists test.t_100 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	for ((i = 1; i <= 100; i++)); do
+		mysql -h ${UP_TIDB_HOST} -P ${UP_TIDB_PORT} -u root -D "test" -e "insert into t_$i values(1,2)"
+	done
+
+	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql_file $CUR/data/test.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
 	check_table_exists batch_add_table.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	# check the ddl of this table is skipped
+	check_table_not_exists test.t_1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	cleanup_process $CDC_BINARY
@@ -56,9 +68,6 @@ function run_without_fast_create_table() {
 
 	run_sql "set global tidb_enable_fast_create_table=off" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
-	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	run_sql_file $CUR/data/prepare.sql ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
 
 	TOPIC_NAME="ticdc-batch-add-table-test-$RANDOM"
@@ -78,7 +87,9 @@ function run_without_fast_create_table() {
 	pulsar) run_pulsar_consumer --upstream-uri $SINK_URI ;;
 	esac
 
+	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql_file $CUR/data/test.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
 	check_table_exists batch_add_table.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed1.toml b/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed1.toml
new file mode 100644
index 000000000..ff4fe5d67
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed1.toml
@@ -0,0 +1,27 @@
+# Case 1: default configuration where `csv.output-old-value=false` and `sink.cloud-storage-config.output-raw-change-event=false`
+# Split and sort update pk/uk events in table sink.
+
+[filter]
+rules = ['test.*']
+
+[sink]
+protocol = "csv"
+# Line terminator. Empty value means "\r\n" (CRLF) is line terminators. The default value is empty.
+terminator = "\n"
+# Directory date separator, Optional values are `none`, `year`, `month`, `date`. The default value is none.
+date-separator = 'day'
+
+[sink.cloud-storage-config]
+output-raw-change-event = false
+
+[sink.csv]
+# Delimiter between fields. Must be ASCII characters. The default value is ','.
+delimiter = ','
+# Quoting character. Empty value means no quoting. The default value is '"'.
+quote = '"'
+# Representation of null values in CSV files, the default value is '\N'
+null = '\N'
+# Include commit-ts in the row data. The default value is false.
+include-commit-ts = true
+# Whether to output the value before the row data changes. The default value is false. 
+output-old-value = false
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed2.toml b/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed2.toml
new file mode 100644
index 000000000..96057f4dd
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed2.toml
@@ -0,0 +1,26 @@
+# Case 2: Split all update events in csv encoder.
+
+[filter]
+rules = ['test.*']
+
+[sink]
+protocol = "csv"
+# Line terminator. Empty value means "\r\n" (CRLF) is line terminators. The default value is empty.
+terminator = "\n"
+# Directory date separator, Optional values are `none`, `year`, `month`, `date`. The default value is none.
+date-separator = 'day'
+
+[sink.cloud-storage-config]
+output-raw-change-event = true
+
+[sink.csv]
+# Delimiter between fields. Must be ASCII characters. The default value is ','.
+delimiter = ','
+# Quoting character. Empty value means no quoting. The default value is '"'.
+quote = '"'
+# Representation of null values in CSV files, the default value is '\N'
+null = '\N'
+# Include commit-ts in the row data. The default value is false.
+include-commit-ts = true
+# Whether to output the value before the row data changes. The default value is false. 
+output-old-value = true
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed3.toml b/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed3.toml
new file mode 100644
index 000000000..033e2bd1a
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed3.toml
@@ -0,0 +1,26 @@
+# Case 3: Don't split any update event
+
+[filter]
+rules = ['test.*']
+
+[sink]
+protocol = "csv"
+# Line terminator. Empty value means "\r\n" (CRLF) is line terminators. The default value is empty.
+terminator = "\n"
+# Directory date separator, Optional values are `none`, `year`, `month`, `date`. The default value is none.
+date-separator = 'day'
+
+[sink.cloud-storage-config]
+output-raw-change-event = true
+
+[sink.csv]
+# Delimiter between fields. Must be ASCII characters. The default value is ','.
+delimiter = ','
+# Quoting character. Empty value means no quoting. The default value is '"'.
+quote = '"'
+# Representation of null values in CSV files, the default value is '\N'
+null = '\N'
+# Include commit-ts in the row data. The default value is false.
+include-commit-ts = true
+# Whether to output the value before the row data changes. The default value is false. 
+output-old-value = false
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed4.toml b/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed4.toml
new file mode 100644
index 000000000..112cfe4a0
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/conf/changefeed4.toml
@@ -0,0 +1,26 @@
+# Case 4: Split and sort update pk/uk events in table sink. Split other update events in csv encoder.
+
+[filter]
+rules = ['test.*']
+
+[sink]
+protocol = "csv"
+# Line terminator. Empty value means "\r\n" (CRLF) is line terminators. The default value is empty.
+terminator = "\n"
+# Directory date separator, Optional values are `none`, `year`, `month`, `date`. The default value is none.
+date-separator = 'day'
+
+[sink.cloud-storage-config]
+output-raw-change-event = false
+
+[sink.csv]
+# Delimiter between fields. Must be ASCII characters. The default value is ','.
+delimiter = ','
+# Quoting character. Empty value means no quoting. The default value is '"'.
+quote = '"'
+# Representation of null values in CSV files, the default value is '\N'
+null = '\N'
+# Include commit-ts in the row data. The default value is false.
+include-commit-ts = true
+# Whether to output the value before the row data changes. The default value is false. 
+output-old-value = true
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/conf/diff_config.toml b/tests/integration_tests/csv_storage_update_pk_clustered/conf/diff_config.toml
new file mode 100644
index 000000000..8edf2368f
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/conf/diff_config.toml
@@ -0,0 +1,29 @@
+# diff Configuration.
+
+check-thread-count = 4
+
+export-fix-sql = true
+
+check-struct-only = false
+
+[task]
+    output-dir = "/tmp/tidb_cdc_test/csv_storage_update_pk_clustered/sync_diff/output"
+
+    source-instances = ["mysql1"]
+
+    target-instance = "tidb0"
+
+    target-check-tables = ["test.?*"]
+
+[data-sources]
+[data-sources.mysql1]
+    host = "127.0.0.1"
+    port = 4000
+    user = "root"
+    password = ""
+
+[data-sources.tidb0]
+    host = "127.0.0.1"
+    port = 3306
+    user = "root"
+    password = ""
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/data/prepare.sql b/tests/integration_tests/csv_storage_update_pk_clustered/data/prepare.sql
new file mode 100644
index 000000000..506a6e757
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/data/prepare.sql
@@ -0,0 +1,27 @@
+drop database if exists `test`;
+create database `test`;
+use `test`;
+
+CREATE TABLE `update_pk` (
+  `id` int PRIMARY KEY CLUSTERED,
+  `pad` varchar(100) NOT NULL
+);
+INSERT INTO `update_pk` (`id`, `pad`) VALUES (1, 'example1'), (2, 'example2');
+INSERT INTO `update_pk` (`id`, `pad`) VALUES (10, 'example10'), (20, 'example20');
+INSERT INTO `update_pk` (`id`, `pad`) VALUES (100, 'example100');
+INSERT INTO `update_pk` (`id`, `pad`) VALUES (1000, 'example1000');
+
+SHOW INDEX FROM update_pk;
+
+CREATE TABLE `update_uk` (
+  `id` int PRIMARY KEY CLUSTERED,
+  `uk` int NOT NULL,
+  `pad` varchar(100) NOT NULL,
+  UNIQUE KEY `uk` (`uk`)
+);
+INSERT INTO `update_uk` (`id`, `uk`, `pad`) VALUES (1, 1, 'example1'), (2, 2, 'example2');
+INSERT INTO `update_uk` (`id`, `uk`, `pad`) VALUES (10, 10, 'example10'), (20, 20, 'example20');
+INSERT INTO `update_uk` (`id`, `uk`, `pad`) VALUES (100, 100, 'example100');
+INSERT INTO `update_uk` (`id`, `uk`, `pad`) VALUES (1000, 1000, 'example1000');
+
+SHOW INDEX FROM update_uk;
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/data/run.sql b/tests/integration_tests/csv_storage_update_pk_clustered/data/run.sql
new file mode 100644
index 000000000..86e7d7e7d
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/data/run.sql
@@ -0,0 +1,40 @@
+USE `test`;
+
+-- update_pk --
+
+BEGIN; -- Note: multi-row exchange
+UPDATE update_pk SET id = 3 WHERE id = 1;
+UPDATE update_pk SET id = 1 WHERE id = 2;
+UPDATE update_pk SET id = 2 WHERE id = 3;
+COMMIT;
+
+BEGIN; -- Note: multi-row update with no order dependency
+UPDATE update_pk SET id = 30 WHERE id = 10;
+UPDATE update_pk SET id = 40 WHERE id = 20;
+COMMIT;
+
+BEGIN; -- Single row update
+UPDATE update_pk SET id = 200 WHERE id = 100;
+COMMIT;
+
+-- Normal update
+UPDATE update_pk SET pad='example1001' WHERE id = 1000;
+
+-- update_uk --
+BEGIN; -- Note: multi-row exchange
+UPDATE update_uk SET uk = 3 WHERE uk = 1;
+UPDATE update_uk SET uk = 1 WHERE uk = 2;
+UPDATE update_uk SET uk = 2 WHERE uk = 3;
+COMMIT;
+
+BEGIN; -- Note: multi-row update with no order dependency
+UPDATE update_uk SET uk = 30 WHERE uk = 10;
+UPDATE update_uk SET uk = 40 WHERE uk = 20;
+COMMIT;
+
+BEGIN; -- Single row update
+UPDATE update_uk SET uk = 200 WHERE uk = 100;
+COMMIT;
+
+-- Normal update
+UPDATE update_uk SET pad='example1001' WHERE uk = 1000;
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed1_pk.res b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed1_pk.res
new file mode 100644
index 000000000..ad6016c80
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed1_pk.res
@@ -0,0 +1,22 @@
+"I","update_pk","test",450253245302439944,1,"example1"
+"I","update_pk","test",450253245302439944,2,"example2"
+"I","update_pk","test",450253245302439946,10,"example10"
+"I","update_pk","test",450253245302439946,20,"example20"
+"I","update_pk","test",450253245302439947,100,"example100"
+"I","update_pk","test",450253245302439948,1000,"example1000"
+
+# translate to normal update in upstream
+"U","update_pk","test",450253245485940746,1,"example2"
+"U","update_pk","test",450253245485940746,2,"example1"
+
+# split and sort in upstream
+"D","update_pk","test",450253245485940749,10,"example10"
+"D","update_pk","test",450253245485940749,20,"example20"
+"I","update_pk","test",450253245485940749,30,"example10"
+"I","update_pk","test",450253245485940749,40,"example20"
+
+# split and sort in upstream
+"D","update_pk","test",450253245485940752,100,"example100"
+"I","update_pk","test",450253245485940752,200,"example100"
+
+"U","update_pk","test",450253245485940753,1000,"example1001"
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed1_uk.res b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed1_uk.res
new file mode 100644
index 000000000..ebe3a6352
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed1_uk.res
@@ -0,0 +1,24 @@
+"I","update_uk","test",450253245446619144,1,1,"example1"
+"I","update_uk","test",450253245446619144,2,2,"example2"
+"I","update_uk","test",450253245446619146,10,10,"example10"
+"I","update_uk","test",450253245446619146,20,20,"example20"
+"I","update_uk","test",450253245446619147,100,100,"example100"
+"I","update_uk","test",450253245446619148,1000,1000,"example1000"
+
+# split and sort in table sink
+"D","update_uk","test",450253245499047940,1,1,"example1"
+"D","update_uk","test",450253245499047940,2,2,"example2"
+"I","update_uk","test",450253245499047940,1,2,"example1"
+"I","update_uk","test",450253245499047940,2,1,"example2"
+
+# split and sort in table sink
+"D","update_uk","test",450253245499047943,10,10,"example10"
+"D","update_uk","test",450253245499047943,20,20,"example20"
+"I","update_uk","test",450253245499047943,10,30,"example10"
+"I","update_uk","test",450253245499047943,20,40,"example20"
+
+# split and sort in table sink
+"D","update_uk","test",450253245499047946,100,100,"example100"
+"I","update_uk","test",450253245499047946,100,200,"example100"
+
+"U","update_uk","test",450253245512155140,1000,1000,"example1001"
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed2_pk.res b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed2_pk.res
new file mode 100644
index 000000000..fc3ea45b6
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed2_pk.res
@@ -0,0 +1,26 @@
+"I","update_pk","test",450253245302439944,false,1,"example1"
+"I","update_pk","test",450253245302439944,false,2,"example2"
+"I","update_pk","test",450253245302439946,false,10,"example10"
+"I","update_pk","test",450253245302439946,false,20,"example20"
+"I","update_pk","test",450253245302439947,false,100,"example100"
+"I","update_pk","test",450253245302439948,false,1000,"example1000"
+
+# translate to normal update in upstream, split in csv encoder
+"D","update_pk","test",450253245485940746,true,1,"example1"
+"I","update_pk","test",450253245485940746,true,1,"example2"
+"D","update_pk","test",450253245485940746,true,2,"example2"
+"I","update_pk","test",450253245485940746,true,2,"example1"
+
+# split and sort in upstream
+"D","update_pk","test",450253245485940749,false,10,"example10"
+"D","update_pk","test",450253245485940749,false,20,"example20"
+"I","update_pk","test",450253245485940749,false,30,"example10"
+"I","update_pk","test",450253245485940749,false,40,"example20"
+
+# split and sort in upstream
+"D","update_pk","test",450253245485940752,false,100,"example100"
+"I","update_pk","test",450253245485940752,false,200,"example100"
+
+# normal update event, split in csv encoder
+"D","update_pk","test",450253245485940753,true,1000,"example1000"
+"I","update_pk","test",450253245485940753,true,1000,"example1001"
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed2_uk.res b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed2_uk.res
new file mode 100644
index 000000000..5e7f2ce0e
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed2_uk.res
@@ -0,0 +1,26 @@
+"I","update_uk","test",450253245446619144,false,1,1,"example1"
+"I","update_uk","test",450253245446619144,false,2,2,"example2"
+"I","update_uk","test",450253245446619146,false,10,10,"example10"
+"I","update_uk","test",450253245446619146,false,20,20,"example20"
+"I","update_uk","test",450253245446619147,false,100,100,"example100"
+"I","update_uk","test",450253245446619148,false,1000,1000,"example1000"
+
+# split in csv encoder, data is consistent since delete by pk
+"D","update_uk","test",450253245499047940,true,1,1,"example1"
+"I","update_uk","test",450253245499047940,true,1,2,"example1"
+"D","update_uk","test",450253245499047940,true,2,2,"example2"
+"I","update_uk","test",450253245499047940,true,2,1,"example2"
+
+# split in csv encoder
+"D","update_uk","test",450253245499047943,true,10,10,"example10"
+"I","update_uk","test",450253245499047943,true,10,30,"example10"
+"D","update_uk","test",450253245499047943,true,20,20,"example20"
+"I","update_uk","test",450253245499047943,true,20,40,"example20"
+
+# split in csv encoder
+"D","update_uk","test",450253245499047946,true,100,100,"example100"
+"I","update_uk","test",450253245499047946,true,100,200,"example100"
+
+# normal update event, also split in csv encoder
+"D","update_uk","test",450253245512155140,true,1000,1000,"example1000"
+"I","update_uk","test",450253245512155140,true,1000,1000,"example1001"
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed3_pk.res b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed3_pk.res
new file mode 100644
index 000000000..9f4448e5d
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed3_pk.res
@@ -0,0 +1,24 @@
+"I","update_pk","test",450253245302439944,1,"example1"
+"I","update_pk","test",450253245302439944,2,"example2"
+"I","update_pk","test",450253245302439946,10,"example10"
+"I","update_pk","test",450253245302439946,20,"example20"
+"I","update_pk","test",450253245302439947,100,"example100"
+"I","update_pk","test",450253245302439948,1000,"example1000"
+
+# output raw change event
+# translate to normal update in upstream
+"U","update_pk","test",450253245485940746,1,"example2"
+"U","update_pk","test",450253245485940746,2,"example1"
+
+# split and sort in upstream
+"D","update_pk","test",450253245485940749,10,"example10"
+"D","update_pk","test",450253245485940749,20,"example20"
+"I","update_pk","test",450253245485940749,30,"example10"
+"I","update_pk","test",450253245485940749,40,"example20"
+
+# split and sort in upstream
+"D","update_pk","test",450253245485940752,100,"example100"
+"I","update_pk","test",450253245485940752,200,"example100"
+
+# normal update event
+"U","update_pk","test",450253245485940753,1000,"example1001"
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed3_uk.res b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed3_uk.res
new file mode 100644
index 000000000..66348746d
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed3_uk.res
@@ -0,0 +1,14 @@
+"I","update_uk","test",450253245446619144,1,1,"example1"
+"I","update_uk","test",450253245446619144,2,2,"example2"
+"I","update_uk","test",450253245446619146,10,10,"example10"
+"I","update_uk","test",450253245446619146,20,20,"example20"
+"I","update_uk","test",450253245446619147,100,100,"example100"
+"I","update_uk","test",450253245446619148,1000,1000,"example1000"
+
+# output raw change event, data is consistent since replace by pk/uk
+"U","update_uk","test",450253245499047940,1,2,"example1"
+"U","update_uk","test",450253245499047940,2,1,"example2"
+"U","update_uk","test",450253245499047943,10,30,"example10"
+"U","update_uk","test",450253245499047943,20,40,"example20"
+"U","update_uk","test",450253245499047946,100,200,"example100"
+"U","update_uk","test",450253245512155140,1000,1000,"example1001"
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed4_pk.res b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed4_pk.res
new file mode 100644
index 000000000..fc3ea45b6
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed4_pk.res
@@ -0,0 +1,26 @@
+"I","update_pk","test",450253245302439944,false,1,"example1"
+"I","update_pk","test",450253245302439944,false,2,"example2"
+"I","update_pk","test",450253245302439946,false,10,"example10"
+"I","update_pk","test",450253245302439946,false,20,"example20"
+"I","update_pk","test",450253245302439947,false,100,"example100"
+"I","update_pk","test",450253245302439948,false,1000,"example1000"
+
+# translate to normal update in upstream, split in csv encoder
+"D","update_pk","test",450253245485940746,true,1,"example1"
+"I","update_pk","test",450253245485940746,true,1,"example2"
+"D","update_pk","test",450253245485940746,true,2,"example2"
+"I","update_pk","test",450253245485940746,true,2,"example1"
+
+# split and sort in upstream
+"D","update_pk","test",450253245485940749,false,10,"example10"
+"D","update_pk","test",450253245485940749,false,20,"example20"
+"I","update_pk","test",450253245485940749,false,30,"example10"
+"I","update_pk","test",450253245485940749,false,40,"example20"
+
+# split and sort in upstream
+"D","update_pk","test",450253245485940752,false,100,"example100"
+"I","update_pk","test",450253245485940752,false,200,"example100"
+
+# normal update event, split in csv encoder
+"D","update_pk","test",450253245485940753,true,1000,"example1000"
+"I","update_pk","test",450253245485940753,true,1000,"example1001"
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed4_uk.res b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed4_uk.res
new file mode 100644
index 000000000..ea6448686
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/result/changefeed4_uk.res
@@ -0,0 +1,26 @@
+"I","update_uk","test",450253245446619144,false,1,1,"example1"
+"I","update_uk","test",450253245446619144,false,2,2,"example2"
+"I","update_uk","test",450253245446619146,false,10,10,"example10"
+"I","update_uk","test",450253245446619146,false,20,20,"example20"
+"I","update_uk","test",450253245446619147,false,100,100,"example100"
+"I","update_uk","test",450253245446619148,false,1000,1000,"example1000"
+
+# split and sort in table sink
+"D","update_uk","test",450253245499047940,false,1,1,"example1"
+"D","update_uk","test",450253245499047940,false,2,2,"example2"
+"I","update_uk","test",450253245499047940,false,1,2,"example1"
+"I","update_uk","test",450253245499047940,false,2,1,"example2"
+
+# split and sort in table sink
+"D","update_uk","test",450253245499047943,false,10,10,"example10"
+"D","update_uk","test",450253245499047943,false,20,20,"example20"
+"I","update_uk","test",450253245499047943,false,10,30,"example10"
+"I","update_uk","test",450253245499047943,false,20,40,"example20"
+
+# split and sort in table sink
+"D","update_uk","test",450253245499047946,false,100,100,"example100"
+"I","update_uk","test",450253245499047946,false,100,200,"example100"
+
+# normal update event, split in csv encoder
+"D","update_uk","test",450253245512155140,true,1000,1000,"example1000"
+"I","update_uk","test",450253245512155140,true,1000,1000,"example1001"
diff --git a/tests/integration_tests/csv_storage_update_pk_clustered/run.sh b/tests/integration_tests/csv_storage_update_pk_clustered/run.sh
new file mode 100644
index 000000000..8606518da
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_clustered/run.sh
@@ -0,0 +1,54 @@
+#!/bin/bash
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+function run_changefeed() {
+	local changefeed_id=$1
+	local start_ts=$2
+	local expected_split_count=$3
+	SINK_URI="file://$WORK_DIR/storage_test/$changefeed_id?flush-interval=5s"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --config=$CUR/conf/$changefeed_id.toml -c "$changefeed_id"
+
+	run_storage_consumer $WORK_DIR $SINK_URI $CUR/conf/$changefeed_id.toml $changefeed_id
+	sleep 8
+	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml 100
+
+	real_split_count=$(grep "split update event" $WORK_DIR/cdc.log | wc -l)
+	if [[ $real_split_count -ne $expected_split_count ]]; then
+		echo "expected split count $expected_split_count, real split count $real_split_count"
+		exit 1
+	fi
+	run_sql "drop database if exists test" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+}
+
+function run() {
+	if [ "$SINK_TYPE" != "storage" ]; then
+		return
+	fi
+
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+	start_tidb_cluster --workdir $WORK_DIR
+	cd $WORK_DIR
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+
+	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql_file $CUR/data/run.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
+	run_changefeed "changefeed1" $start_ts 5
+	run_changefeed "changefeed2" $start_ts 5
+	run_changefeed "changefeed3" $start_ts 5
+	run_changefeed "changefeed4" $start_ts 10
+}
+
+trap stop_tidb_cluster EXIT
+run $*
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed1.toml b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed1.toml
new file mode 100644
index 000000000..ff4fe5d67
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed1.toml
@@ -0,0 +1,27 @@
+# Case 1: default configuration where `csv.output-old-value=false` and `sink.cloud-storage-config.output-raw-change-event=false`
+# Split and sort update pk/uk events in table sink.
+
+[filter]
+rules = ['test.*']
+
+[sink]
+protocol = "csv"
+# Line terminator. Empty value means "\r\n" (CRLF) is line terminators. The default value is empty.
+terminator = "\n"
+# Directory date separator, Optional values are `none`, `year`, `month`, `date`. The default value is none.
+date-separator = 'day'
+
+[sink.cloud-storage-config]
+output-raw-change-event = false
+
+[sink.csv]
+# Delimiter between fields. Must be ASCII characters. The default value is ','.
+delimiter = ','
+# Quoting character. Empty value means no quoting. The default value is '"'.
+quote = '"'
+# Representation of null values in CSV files, the default value is '\N'
+null = '\N'
+# Include commit-ts in the row data. The default value is false.
+include-commit-ts = true
+# Whether to output the value before the row data changes. The default value is false. 
+output-old-value = false
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed2.toml b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed2.toml
new file mode 100644
index 000000000..96057f4dd
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed2.toml
@@ -0,0 +1,26 @@
+# Case 2: Split all update events in csv encoder.
+
+[filter]
+rules = ['test.*']
+
+[sink]
+protocol = "csv"
+# Line terminator. Empty value means "\r\n" (CRLF) is line terminators. The default value is empty.
+terminator = "\n"
+# Directory date separator, Optional values are `none`, `year`, `month`, `date`. The default value is none.
+date-separator = 'day'
+
+[sink.cloud-storage-config]
+output-raw-change-event = true
+
+[sink.csv]
+# Delimiter between fields. Must be ASCII characters. The default value is ','.
+delimiter = ','
+# Quoting character. Empty value means no quoting. The default value is '"'.
+quote = '"'
+# Representation of null values in CSV files, the default value is '\N'
+null = '\N'
+# Include commit-ts in the row data. The default value is false.
+include-commit-ts = true
+# Whether to output the value before the row data changes. The default value is false. 
+output-old-value = true
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed3.toml b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed3.toml
new file mode 100644
index 000000000..033e2bd1a
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed3.toml
@@ -0,0 +1,26 @@
+# Case 3: Don't split any update event
+
+[filter]
+rules = ['test.*']
+
+[sink]
+protocol = "csv"
+# Line terminator. Empty value means "\r\n" (CRLF) is line terminators. The default value is empty.
+terminator = "\n"
+# Directory date separator, Optional values are `none`, `year`, `month`, `date`. The default value is none.
+date-separator = 'day'
+
+[sink.cloud-storage-config]
+output-raw-change-event = true
+
+[sink.csv]
+# Delimiter between fields. Must be ASCII characters. The default value is ','.
+delimiter = ','
+# Quoting character. Empty value means no quoting. The default value is '"'.
+quote = '"'
+# Representation of null values in CSV files, the default value is '\N'
+null = '\N'
+# Include commit-ts in the row data. The default value is false.
+include-commit-ts = true
+# Whether to output the value before the row data changes. The default value is false. 
+output-old-value = false
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed4.toml b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed4.toml
new file mode 100644
index 000000000..112cfe4a0
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/changefeed4.toml
@@ -0,0 +1,26 @@
+# Case 4: Split and sort update pk/uk events in table sink. Split other update events in csv encoder.
+
+[filter]
+rules = ['test.*']
+
+[sink]
+protocol = "csv"
+# Line terminator. Empty value means "\r\n" (CRLF) is line terminators. The default value is empty.
+terminator = "\n"
+# Directory date separator, Optional values are `none`, `year`, `month`, `date`. The default value is none.
+date-separator = 'day'
+
+[sink.cloud-storage-config]
+output-raw-change-event = false
+
+[sink.csv]
+# Delimiter between fields. Must be ASCII characters. The default value is ','.
+delimiter = ','
+# Quoting character. Empty value means no quoting. The default value is '"'.
+quote = '"'
+# Representation of null values in CSV files, the default value is '\N'
+null = '\N'
+# Include commit-ts in the row data. The default value is false.
+include-commit-ts = true
+# Whether to output the value before the row data changes. The default value is false. 
+output-old-value = true
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/diff_config.toml b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/diff_config.toml
new file mode 100644
index 000000000..0714c0c18
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/conf/diff_config.toml
@@ -0,0 +1,29 @@
+# diff Configuration.
+
+check-thread-count = 4
+
+export-fix-sql = true
+
+check-struct-only = false
+
+[task]
+    output-dir = "/tmp/tidb_cdc_test/csv_storage_update_pk_nonclustered/sync_diff-<suffix>/output"
+
+    source-instances = ["mysql1"]
+
+    target-instance = "tidb0"
+
+    target-check-tables = ["test.?*"]
+
+[data-sources]
+[data-sources.mysql1]
+    host = "127.0.0.1"
+    port = 4000
+    user = "root"
+    password = ""
+
+[data-sources.tidb0]
+    host = "127.0.0.1"
+    port = 3306
+    user = "root"
+    password = ""
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/data/prepare.sql b/tests/integration_tests/csv_storage_update_pk_nonclustered/data/prepare.sql
new file mode 100644
index 000000000..f3cd4ca4d
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/data/prepare.sql
@@ -0,0 +1,28 @@
+drop database if exists `test`;
+create database `test`;
+use `test`;
+
+CREATE TABLE `update_pk` (
+  `id` int PRIMARY KEY NONCLUSTERED,
+  `pad` varchar(100) NOT NULL
+);
+INSERT INTO `update_pk` (`id`, `pad`) VALUES (1, 'example1'), (2, 'example2');
+INSERT INTO `update_pk` (`id`, `pad`) VALUES (10, 'example10'), (20, 'example20');
+INSERT INTO `update_pk` (`id`, `pad`) VALUES (100, 'example100');
+INSERT INTO `update_pk` (`id`, `pad`) VALUES (1000, 'example1000');
+
+
+SHOW INDEX FROM update_pk;
+
+CREATE TABLE `update_uk` (
+  `id` int PRIMARY KEY NONCLUSTERED,
+  `uk` int NOT NULL,
+  `pad` varchar(100) NOT NULL,
+  UNIQUE KEY `uk` (`uk`)
+);
+INSERT INTO `update_uk` (`id`, `uk`, `pad`) VALUES (1, 1, 'example1'), (2, 2, 'example2');
+INSERT INTO `update_uk` (`id`, `uk`, `pad`) VALUES (10, 10, 'example10'), (20, 20, 'example20');
+INSERT INTO `update_uk` (`id`, `uk`, `pad`) VALUES (100, 100, 'example100');
+INSERT INTO `update_uk` (`id`, `uk`, `pad`) VALUES (1000, 1000, 'example1000');
+
+SHOW INDEX FROM update_uk;
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/data/run.sql b/tests/integration_tests/csv_storage_update_pk_nonclustered/data/run.sql
new file mode 100644
index 000000000..86e7d7e7d
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/data/run.sql
@@ -0,0 +1,40 @@
+USE `test`;
+
+-- update_pk --
+
+BEGIN; -- Note: multi-row exchange
+UPDATE update_pk SET id = 3 WHERE id = 1;
+UPDATE update_pk SET id = 1 WHERE id = 2;
+UPDATE update_pk SET id = 2 WHERE id = 3;
+COMMIT;
+
+BEGIN; -- Note: multi-row update with no order dependency
+UPDATE update_pk SET id = 30 WHERE id = 10;
+UPDATE update_pk SET id = 40 WHERE id = 20;
+COMMIT;
+
+BEGIN; -- Single row update
+UPDATE update_pk SET id = 200 WHERE id = 100;
+COMMIT;
+
+-- Normal update
+UPDATE update_pk SET pad='example1001' WHERE id = 1000;
+
+-- update_uk --
+BEGIN; -- Note: multi-row exchange
+UPDATE update_uk SET uk = 3 WHERE uk = 1;
+UPDATE update_uk SET uk = 1 WHERE uk = 2;
+UPDATE update_uk SET uk = 2 WHERE uk = 3;
+COMMIT;
+
+BEGIN; -- Note: multi-row update with no order dependency
+UPDATE update_uk SET uk = 30 WHERE uk = 10;
+UPDATE update_uk SET uk = 40 WHERE uk = 20;
+COMMIT;
+
+BEGIN; -- Single row update
+UPDATE update_uk SET uk = 200 WHERE uk = 100;
+COMMIT;
+
+-- Normal update
+UPDATE update_uk SET pad='example1001' WHERE uk = 1000;
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed1_pk.res b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed1_pk.res
new file mode 100644
index 000000000..08f6eedb8
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed1_pk.res
@@ -0,0 +1,24 @@
+"I","update_pk","test",450250823741472787,1,"example1"
+"I","update_pk","test",450250823741472787,2,"example2"
+"I","update_pk","test",450250823741472790,10,"example10"
+"I","update_pk","test",450250823741472790,20,"example20"
+"I","update_pk","test",450250823741472791,100,"example100"
+"I","update_pk","test",450250823741472792,1000,"example1000"
+
+# split and sort in table sink
+"D","update_pk","test",450250823807270922,1,"example1"
+"D","update_pk","test",450250823807270922,2,"example2"
+"I","update_pk","test",450250823807270922,2,"example1"
+"I","update_pk","test",450250823807270922,1,"example2"
+
+# split and sort in table sink
+"D","update_pk","test",450250823807270925,10,"example10"
+"D","update_pk","test",450250823807270925,20,"example20"
+"I","update_pk","test",450250823807270925,30,"example10"
+"I","update_pk","test",450250823807270925,40,"example20"
+
+# split and sort in table sink
+"D","update_pk","test",450250823807270927,100,"example100"
+"I","update_pk","test",450250823807270927,200,"example100"
+
+"U","update_pk","test",450250823807270928,1000,"example1001"
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed1_uk.res b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed1_uk.res
new file mode 100644
index 000000000..b26f2219a
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed1_uk.res
@@ -0,0 +1,24 @@
+"I","update_uk","test",450250823780794385,1,1,"example1"
+"I","update_uk","test",450250823780794385,2,2,"example2"
+"I","update_uk","test",450250823780794387,10,10,"example10"
+"I","update_uk","test",450250823780794387,20,20,"example20"
+"I","update_uk","test",450250823780794389,100,100,"example100"
+"I","update_uk","test",450250823780794390,1000,1000,"example1000"
+
+# split and sort in table sink
+"D","update_uk","test",450250823807270931,1,1,"example1"
+"D","update_uk","test",450250823807270931,2,2,"example2"
+"I","update_uk","test",450250823807270931,1,2,"example1"
+"I","update_uk","test",450250823807270931,2,1,"example2"
+
+# split and sort in table sink
+"D","update_uk","test",450250823820115970,10,10,"example10"
+"D","update_uk","test",450250823820115970,20,20,"example20"
+"I","update_uk","test",450250823820115970,10,30,"example10"
+"I","update_uk","test",450250823820115970,20,40,"example20"
+
+# split and sort in table sink
+"D","update_uk","test",450250823820115973,100,100,"example100"
+"I","update_uk","test",450250823820115973,100,200,"example100"
+
+"U","update_uk","test",450250823820115977,1000,1000,"example1001"
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed2_pk.res b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed2_pk.res
new file mode 100644
index 000000000..e2713a94f
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed2_pk.res
@@ -0,0 +1,28 @@
+"I","update_pk","test",450250823741472787,false,1,"example1"
+"I","update_pk","test",450250823741472787,false,2,"example2"
+"I","update_pk","test",450250823741472790,false,10,"example10"
+"I","update_pk","test",450250823741472790,false,20,"example20"
+"I","update_pk","test",450250823741472791,false,100,"example100"
+"I","update_pk","test",450250823741472792,false,1000,"example1000"
+
+# split in csv encoder
+# DIFF_RES: REPLACE INTO `test`.`update_pk`(`id`,`pad`) VALUES (2,'example1');
+# lost id=2 since delete are not sorted before insert within single txn
+"D","update_pk","test",450250823807270922,true,1,"example1"
+"I","update_pk","test",450250823807270922,true,2,"example1"
+"D","update_pk","test",450250823807270922,true,2,"example2"
+"I","update_pk","test",450250823807270922,true,1,"example2"
+
+# split in csv encoder
+"D","update_pk","test",450250823807270925,true,10,"example10"
+"I","update_pk","test",450250823807270925,true,30,"example10"
+"D","update_pk","test",450250823807270925,true,20,"example20"
+"I","update_pk","test",450250823807270925,true,40,"example20"
+
+# split in csv encoder
+"D","update_pk","test",450250823807270927,true,100,"example100"
+"I","update_pk","test",450250823807270927,true,200,"example100"
+
+# normal update event, also split in csv encoder
+"D","update_pk","test",450250823807270928,true,1000,"example1000"
+"I","update_pk","test",450250823807270928,true,1000,"example1001"
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed2_uk.res b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed2_uk.res
new file mode 100644
index 000000000..1783ee5a0
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed2_uk.res
@@ -0,0 +1,26 @@
+"I","update_uk","test",450250823780794385,false,1,1,"example1"
+"I","update_uk","test",450250823780794385,false,2,2,"example2"
+"I","update_uk","test",450250823780794387,false,10,10,"example10"
+"I","update_uk","test",450250823780794387,false,20,20,"example20"
+"I","update_uk","test",450250823780794389,false,100,100,"example100"
+"I","update_uk","test",450250823780794390,false,1000,1000,"example1000"
+
+# split in csv encoder, data is consistent since delete by pk
+"D","update_uk","test",450250823807270931,true,1,1,"example1"
+"I","update_uk","test",450250823807270931,true,1,2,"example1"
+"D","update_uk","test",450250823807270931,true,2,2,"example2"
+"I","update_uk","test",450250823807270931,true,2,1,"example2"
+
+# split in csv encoder
+"D","update_uk","test",450250823820115970,true,10,10,"example10"
+"I","update_uk","test",450250823820115970,true,10,30,"example10"
+"D","update_uk","test",450250823820115970,true,20,20,"example20"
+"I","update_uk","test",450250823820115970,true,20,40,"example20"
+
+# split in csv encoder
+"D","update_uk","test",450250823820115973,true,100,100,"example100"
+"I","update_uk","test",450250823820115973,true,100,200,"example100"
+
+# normal update event, also split in csv encoder
+"D","update_uk","test",450250823820115977,true,1000,1000,"example1000"
+"I","update_uk","test",450250823820115977,true,1000,1000,"example1001"
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed3_pk.res b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed3_pk.res
new file mode 100644
index 000000000..54595b6f4
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed3_pk.res
@@ -0,0 +1,22 @@
+"I","update_pk","test",450250823741472787,1,"example1"
+"I","update_pk","test",450250823741472787,2,"example2"
+"I","update_pk","test",450250823741472790,10,"example10"
+"I","update_pk","test",450250823741472790,20,"example20"
+"I","update_pk","test",450250823741472791,100,"example100"
+"I","update_pk","test",450250823741472792,1000,"example1000"
+
+# output raw change event
+"U","update_pk","test",450250823807270922,2,"example1"
+"U","update_pk","test",450250823807270922,1,"example2"
+
+
+# DIFF_RES:
+# DELETE FROM `test`.`update_pk` WHERE `id` = 10 AND `pad` = 'example10' LIMIT 1;
+# DELETE FROM `test`.`update_pk` WHERE `id` = 20 AND `pad` = 'example20' LIMIT 1;
+# DELETE FROM `test`.`update_pk` WHERE `id` = 100 AND `pad` = 'example100' LIMIT 1;
+# old row is not deleted since lack of old value
+"U","update_pk","test",450250823807270925,30,"example10"
+"U","update_pk","test",450250823807270925,40,"example20"
+"U","update_pk","test",450250823807270927,200,"example100"
+
+"U","update_pk","test",450250823807270928,1000,"example1001"
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed3_uk.res b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed3_uk.res
new file mode 100644
index 000000000..ed01246a1
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed3_uk.res
@@ -0,0 +1,14 @@
+"I","update_uk","test",450250823780794385,1,1,"example1"
+"I","update_uk","test",450250823780794385,2,2,"example2"
+"I","update_uk","test",450250823780794387,10,10,"example10"
+"I","update_uk","test",450250823780794387,20,20,"example20"
+"I","update_uk","test",450250823780794389,100,100,"example100"
+"I","update_uk","test",450250823780794390,1000,1000,"example1000"
+
+# output raw change event, data is consistent since replace by pk/uk
+"U","update_uk","test",450250823807270931,1,2,"example1"
+"U","update_uk","test",450250823807270931,2,1,"example2"
+"U","update_uk","test",450250823820115970,10,30,"example10"
+"U","update_uk","test",450250823820115970,20,40,"example20"
+"U","update_uk","test",450250823820115973,100,200,"example100"
+"U","update_uk","test",450250823820115977,1000,1000,"example1001"
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed4_pk.res b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed4_pk.res
new file mode 100644
index 000000000..79dc50dbf
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed4_pk.res
@@ -0,0 +1,26 @@
+"I","update_pk","test",450250823741472787,false,1,"example1"
+"I","update_pk","test",450250823741472787,false,2,"example2"
+"I","update_pk","test",450250823741472790,false,10,"example10"
+"I","update_pk","test",450250823741472790,false,20,"example20"
+"I","update_pk","test",450250823741472791,false,100,"example100"
+"I","update_pk","test",450250823741472792,false,1000,"example1000"
+
+# split and sort in table sink
+"D","update_pk","test",450250823807270922,false,1,"example1"
+"D","update_pk","test",450250823807270922,false,2,"example2"
+"I","update_pk","test",450250823807270922,false,2,"example1"
+"I","update_pk","test",450250823807270922,false,1,"example2"
+
+# split and sort in table sink
+"D","update_pk","test",450250823807270925,false,10,"example10"
+"D","update_pk","test",450250823807270925,false,20,"example20"
+"I","update_pk","test",450250823807270925,false,30,"example10"
+"I","update_pk","test",450250823807270925,false,40,"example20"
+
+# split and sort in table sink
+"D","update_pk","test",450250823807270927,false,100,"example100"
+"I","update_pk","test",450250823807270927,false,200,"example100"
+
+# normal update event, split in csv encoder
+"D","update_pk","test",450250823807270928,true,1000,"example1000"
+"I","update_pk","test",450250823807270928,true,1000,"example1001"
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed4_uk.res b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed4_uk.res
new file mode 100644
index 000000000..2225f55ff
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/result/changefeed4_uk.res
@@ -0,0 +1,26 @@
+"I","update_uk","test",450250823780794385,false,1,1,"example1"
+"I","update_uk","test",450250823780794385,false,2,2,"example2"
+"I","update_uk","test",450250823780794387,false,10,10,"example10"
+"I","update_uk","test",450250823780794387,false,20,20,"example20"
+"I","update_uk","test",450250823780794389,false,100,100,"example100"
+"I","update_uk","test",450250823780794390,false,1000,1000,"example1000"
+
+# split and sort in table sink
+"D","update_uk","test",450250823807270931,false,1,1,"example1"
+"D","update_uk","test",450250823807270931,false,2,2,"example2"
+"I","update_uk","test",450250823807270931,false,1,2,"example1"
+"I","update_uk","test",450250823807270931,false,2,1,"example2"
+
+# split and sort in table sink
+"D","update_uk","test",450250823820115970,false,10,10,"example10"
+"D","update_uk","test",450250823820115970,false,20,20,"example20"
+"I","update_uk","test",450250823820115970,false,10,30,"example10"
+"I","update_uk","test",450250823820115970,false,20,40,"example20"
+
+# split and sort in table sink
+"D","update_uk","test",450250823820115973,false,100,100,"example100"
+"I","update_uk","test",450250823820115973,false,100,200,"example100"
+
+# normal update event, split in csv encoder
+"D","update_uk","test",450250823820115977,true,1000,1000,"example1000"
+"I","update_uk","test",450250823820115977,true,1000,1000,"example1001"
\ No newline at end of file
diff --git a/tests/integration_tests/csv_storage_update_pk_nonclustered/run.sh b/tests/integration_tests/csv_storage_update_pk_nonclustered/run.sh
new file mode 100644
index 000000000..adbf3392f
--- /dev/null
+++ b/tests/integration_tests/csv_storage_update_pk_nonclustered/run.sh
@@ -0,0 +1,64 @@
+#!/bin/bash
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+function run_changefeed() {
+	local changefeed_id=$1
+	local start_ts=$2
+	local expected_split_count=$3
+	local should_pass_check=$4
+	SINK_URI="file://$WORK_DIR/storage_test/$changefeed_id?flush-interval=5s"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --config=$CUR/conf/$changefeed_id.toml -c "$changefeed_id"
+
+	run_storage_consumer $WORK_DIR $SINK_URI $CUR/conf/$changefeed_id.toml $changefeed_id
+	sleep 8
+
+	cp $CUR/conf/diff_config.toml $WORK_DIR/diff_config.toml
+	sed -i "s/<suffix>/$changefeed_id/" $WORK_DIR/diff_config.toml
+	if [[ $should_pass_check == true ]]; then
+		check_sync_diff $WORK_DIR $WORK_DIR/diff_config.toml 100
+	else
+		check_sync_diff $WORK_DIR $WORK_DIR/diff_config.toml 30 && exit 1 || echo "check_sync_diff failed as expected for $changefeed_id"
+	fi
+
+	real_split_count=$(grep "split update event" $WORK_DIR/cdc.log | wc -l)
+	if [[ $real_split_count -ne $expected_split_count ]]; then
+		echo "expected split count $expected_split_count, real split count $real_split_count"
+		exit 1
+	fi
+	run_sql "drop database if exists test" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+}
+
+function run() {
+	if [ "$SINK_TYPE" != "storage" ]; then
+		return
+	fi
+
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+	start_tidb_cluster --workdir $WORK_DIR
+	cd $WORK_DIR
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+
+	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql_file $CUR/data/run.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
+	run_changefeed "changefeed1" $start_ts 10 true
+	# changefeed2 fail since delete events are not sorted
+	run_changefeed "changefeed2" $start_ts 10 false
+	# changefeed3 fail since update pk/uk events are not split
+	run_changefeed "changefeed3" $start_ts 10 false
+	run_changefeed "changefeed4" $start_ts 20 true
+}
+
+trap stop_tidb_cluster EXIT
+run $*
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"
diff --git a/tests/integration_tests/run_group.sh b/tests/integration_tests/run_group.sh
index bbed31655..f8f15673e 100755
--- a/tests/integration_tests/run_group.sh
+++ b/tests/integration_tests/run_group.sh
@@ -11,7 +11,7 @@ group_num=${group#G}
 # Other tests that only support mysql: batch_update_to_no_batch ddl_reentrant
 # changefeed_fast_fail changefeed_resume_with_checkpoint_ts sequence
 # multi_cdc_cluster capture_suicide_while_balance_table
-mysql_only="bdr_mode capture_suicide_while_balance_table syncpoint hang_sink_suicide server_config_compatibility changefeed_dup_error_restart"
+mysql_only="bdr_mode capture_suicide_while_balance_table syncpoint syncpoint_check_ts hang_sink_suicide server_config_compatibility changefeed_dup_error_restart"
 mysql_only_http="http_api http_api_tls api_v2 http_api_tls_with_user_auth cli_tls_with_auth"
 mysql_only_consistent_replicate="consistent_replicate_ddl consistent_replicate_gbk consistent_replicate_nfs consistent_replicate_storage_file consistent_replicate_storage_file_large_value consistent_replicate_storage_s3 consistent_partition_table"
 
@@ -19,7 +19,7 @@ kafka_only="kafka_big_messages kafka_compression kafka_messages kafka_sink_error
 kafka_only_protocol="kafka_simple_basic kafka_simple_basic_avro kafka_simple_handle_key_only kafka_simple_handle_key_only_avro kafka_simple_claim_check kafka_simple_claim_check_avro canal_json_adapter_compatibility canal_json_basic canal_json_content_compatible multi_topics avro_basic canal_json_handle_key_only open_protocol_handle_key_only canal_json_claim_check open_protocol_claim_check"
 kafka_only_v2="kafka_big_messages_v2 multi_tables_ddl_v2 multi_topics_v2"
 
-storage_only="lossy_ddl storage_csv_update"
+storage_only="lossy_ddl storage_csv_update csv_storage_update_pk_clustered csv_storage_update_pk_nonclustered"
 storage_only_csv="storage_cleanup csv_storage_basic csv_storage_multi_tables_ddl csv_storage_partition_table"
 storage_only_canal_json="canal_json_storage_basic canal_json_storage_partition_table"
 
diff --git a/tests/integration_tests/syncpoint_check_ts/conf/changefeed.toml b/tests/integration_tests/syncpoint_check_ts/conf/changefeed.toml
new file mode 100644
index 000000000..d01100dc3
--- /dev/null
+++ b/tests/integration_tests/syncpoint_check_ts/conf/changefeed.toml
@@ -0,0 +1,2 @@
+enable-sync-point = true
+sync-point-interval = "30s"
diff --git a/tests/integration_tests/syncpoint_check_ts/run.sh b/tests/integration_tests/syncpoint_check_ts/run.sh
new file mode 100644
index 000000000..f6fb07cf9
--- /dev/null
+++ b/tests/integration_tests/syncpoint_check_ts/run.sh
@@ -0,0 +1,72 @@
+#!/bin/bash
+
+# [DISCRIPTION]:
+#   The test is used to check the first sync point ts value when we set different SyncPointInterval
+# [STEP]:
+#   1. Create changefeed with --sync-point --sync-interval=30s
+#   2. After test, get fisrt syncpoints from tidb_cdc.syncpoint_v1
+#   3. check the first syncpoint ts value
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+CDC_COUNT=3
+DB_COUNT=4
+
+function run() {
+	if [ "$SINK_TYPE" != "mysql" ]; then
+		echo "kafka downstream isn't support syncpoint record"
+		return
+	fi
+
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+	run_sql "SET GLOBAL tidb_enable_external_ts_read = on;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	# this test contains `set global tidb_external_ts = ?` , which requires super privilege, so we
+	# can't use the normal user
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml"
+
+	run_sql "SET GLOBAL tidb_enable_external_ts_read = off;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	sleep 60
+
+	run_sql "SELECT primary_ts, secondary_ts FROM tidb_cdc.syncpoint_v1 order by primary_ts limit 1;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	echo "____________________________________"
+	cat "$OUT_DIR/sql_res.$TEST_NAME.txt"
+	primary_ts=($(grep primary_ts $OUT_DIR/sql_res.$TEST_NAME.txt | awk -F ": " '{print $2}'))
+	echo "primary_ts is " $primary_ts "start_ts is " $start_ts
+
+	shifted_primary_ts=$(($primary_ts >> 18))
+	if [ $(($shifted_primary_ts % 30000)) -eq 0 ]; then
+		pre_primary_ts=$(($primary_ts >> 18 - 30000))
+		pre_primary_ts=$(($pre_primary_ts << 18))
+		if [ $pre_primary_ts -lt $start_ts ]; then
+			echo "check successfully"
+		else
+			echo "primary ts is not correct, there should be a smaller primary ts as the first sync point ts"
+			exit 1
+		fi
+	else
+		echo "primary ts is not correct, the difference between it and the time.Unix(0,0) should be the integer multiples of the syncPointInterval "
+		exit 1
+	fi
+
+	cleanup_process $CDC_BINARY
+}
+
+trap stop_tidb_cluster EXIT
+run $*
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"