Repository: pingcap/tiflow
Pull request #11262 (2024-06-05 15:01:01)

ddl(ticdc): support batch create table ddl
<!--
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/11009

### What is changed and how it works?


### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Integration test
 - Manual test (add detailed scripts or steps below)
1. create changefeed.
2. set global tidb_enable_fast_create_table=on; for the upstream tidb'
3. concurrently create 1000 tables in upstream, and insert data
4. check the tables and data in the downstream tidb
![image](https://github.com/pingcap/tiflow/assets/26538495/7e5e671f-a0ad-434a-b1ff-9d07599eb285)


#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
`None`.
```


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

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from [hongyunyan](https://github.com/hongyunyan), **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)**

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":["hongyunyan"]} -->
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 | 991c19c8de678727962cebedcf156a1e52164283 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/2117/display/redirect) | true | `/test verify`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11262). [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 | 991c19c8de678727962cebedcf156a1e52164283 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/2117/display/redirect) | true | `/test verify`
pull-dm-integration-test | 991c19c8de678727962cebedcf156a1e52164283 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/2325/display/redirect) | true | `/test dm-integration-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11262). [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]: 
[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 tests **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-dm-integration-test | 991c19c8de678727962cebedcf156a1e52164283 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/2325/display/redirect) | true | `/test dm-integration-test`
jenkins-ticdc/verify | 779e5a37995f8cc3e562ea7a39a9bf7b756d0d5a | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/2118/display/redirect) | true | `/test verify`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11262). [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 | 779e5a37995f8cc3e562ea7a39a9bf7b756d0d5a | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/2118/display/redirect) | true | `/test verify`
pull-dm-integration-test | 779e5a37995f8cc3e562ea7a39a9bf7b756d0d5a | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/2326/display/redirect) | true | `/test dm-integration-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11262). [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]: 
[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]: 
[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]: 
[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 | e130e571a03c6d4f9e2ab161a60509b656c9d865 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/2121/display/redirect) | true | `/test verify`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11262). [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 | c621bc5449668ff67efcb39dfc9752e3dabf5e30 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/2122/display/redirect) | true | `/test verify`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11262). [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 | d36b2143c3d94e9e39ae9107c99927f4ee33ef5c | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/ghpr_verify/2123/display/redirect) | true | `/test verify`
pull-dm-integration-test | d36b2143c3d94e9e39ae9107c99927f4ee33ef5c | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/2331/display/redirect) | true | `/test dm-integration-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11262). [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 -->
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pingcap/tiflow/pull/11262?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 `0%` with `33 lines` in your changes missing coverage. Please review.
> Project coverage is 57.6325%. Comparing base [(`971bfcd`)](https://app.codecov.io/gh/pingcap/tiflow/commit/971bfcdb28fcb71b2fa5c0673afcd5996f084ad9?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) to head [(`9dbf262`)](https://app.codecov.io/gh/pingcap/tiflow/commit/9dbf2625e6613e5547a944644293126dab8d27fc?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap).

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


| [Components](https://app.codecov.io/gh/pingcap/tiflow/pull/11262/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/11262/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `61.4404% <0.0000%> (-0.0361%)` | :arrow_down: |
| [dm](https://app.codecov.io/gh/pingcap/tiflow/pull/11262/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `51.2091% <ø> (+0.0322%)` | :arrow_up: |
| [engine](https://app.codecov.io/gh/pingcap/tiflow/pull/11262/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `63.3879% <ø> (-0.0071%)` | :arrow_down: |

| [Flag](https://app.codecov.io/gh/pingcap/tiflow/pull/11262/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/11262/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `57.6325% <0.0000%> (-0.0050%)` | :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     #11262        +/-   ##
================================================
- Coverage   57.6375%   57.6325%   -0.0050%     
================================================
  Files           850        850                
  Lines        126055     126085        +30     
================================================
+ Hits          72655      72666        +11     
- Misses        47998      48021        +23     
+ Partials       5402       5398         -4     
```

</details>
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
--- | --- | --- | --- | ---
pull-dm-integration-test | d36b2143c3d94e9e39ae9107c99927f4ee33ef5c | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/2331/display/redirect) | true | `/test dm-integration-test`
pull-cdc-integration-mysql-test | 9dbf2625e6613e5547a944644293126dab8d27fc | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_test/2160/display/redirect) | true | `/test cdc-integration-mysql-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11262). [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
--- | --- | --- | --- | ---
pull-cdc-integration-mysql-test | 9dbf2625e6613e5547a944644293126dab8d27fc | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_test/2160/display/redirect) | true | `/test cdc-integration-mysql-test`
pull-cdc-integration-kafka-test | 9dbf2625e6613e5547a944644293126dab8d27fc | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/2233/display/redirect) | true | `/test cdc-integration-kafka-test`

[Full PR test history](https://prow.tidb.net/pr-history?org=pingcap&repo=tiflow&pr=11262). [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 -->

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..f19716f69 100755
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -811,6 +811,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/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..ebcb41681 100755
--- a/tests/integration_tests/run_group.sh
+++ b/tests/integration_tests/run_group.sh
@@ -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"