Repository: pingcap/tiflow
Pull request #10708 (2024-03-05 03:40:15)

filter (ticdc): add unit test case to cover all event type for event filter
<!--
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: ref #10511 

### 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. -->

 - Unit test


#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
None
```


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

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from [asddongmen](https://github.com/asddongmen), **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":["asddongmen"]} -->
ti-chi-bot[bot]: @asddongmen: 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-engine-integration-test | 9aac2f398c27d8a0c449b568210df8c504c671c2 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_engine_integration_test/1031/display/redirect) | true | `/test engine-integration-test`
pull-cdc-integration-pulsar-test | 9aac2f398c27d8a0c449b568210df8c504c671c2 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/799/display/redirect) | true | `/test cdc-integration-pulsar-test`

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

<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/10708?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
> Merging [#10708](https://app.codecov.io/gh/pingcap/tiflow/pull/10708?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (9aac2f3) into [master](https://app.codecov.io/gh/pingcap/tiflow/commit/31dd24f902d081e50f16a19d5fb3d7b0fdfae5b5?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) (31dd24f) will **increase** coverage by `3.6791%`.
> Report is 3 commits behind head on master.
> The diff coverage is `100.0000%`.

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


| [Components](https://app.codecov.io/gh/pingcap/tiflow/pull/10708/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/10708/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `61.9587% <100.0000%> (∅)` | |
| [dm](https://app.codecov.io/gh/pingcap/tiflow/pull/10708/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `51.1990% <ø> (+0.0282%)` | :arrow_up: |
| [engine](https://app.codecov.io/gh/pingcap/tiflow/pull/10708/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `63.4289% <ø> (+0.0192%)` | :arrow_up: |

| [Flag](https://app.codecov.io/gh/pingcap/tiflow/pull/10708/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/10708/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `57.5714% <100.0000%> (+3.6791%)` | :arrow_up: |

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     #10708        +/-   ##
================================================
+ Coverage   53.8923%   57.5714%   +3.6791%     
================================================
  Files           403        850       +447     
  Lines         63651     126733     +63082     
================================================
+ Hits          34303      72962     +38659     
- Misses        26642      48333     +21691     
- Partials       2706       5438      +2732     
```

</details>
ti-chi-bot[bot]: @asddongmen: 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-engine-integration-test | 9aac2f398c27d8a0c449b568210df8c504c671c2 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_engine_integration_test/1031/display/redirect) | true | `/test engine-integration-test`
pull-cdc-integration-pulsar-test | 9aac2f398c27d8a0c449b568210df8c504c671c2 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/799/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-dm-integration-test | 9aac2f398c27d8a0c449b568210df8c504c671c2 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/1146/display/redirect) | true | `/test dm-integration-test`

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

<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]: @asddongmen: 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-engine-integration-test | 9aac2f398c27d8a0c449b568210df8c504c671c2 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_engine_integration_test/1031/display/redirect) | true | `/test engine-integration-test`
pull-cdc-integration-pulsar-test | 9aac2f398c27d8a0c449b568210df8c504c671c2 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_pulsar_test/799/display/redirect) | true | `/test cdc-integration-pulsar-test`
pull-dm-integration-test | 9aac2f398c27d8a0c449b568210df8c504c671c2 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_dm_integration_test/1146/display/redirect) | true | `/test dm-integration-test`
pull-cdc-integration-kafka-test | 9aac2f398c27d8a0c449b568210df8c504c671c2 | [link](https://do.pingcap.net/jenkins/job/pingcap/job/tiflow/job/pull_cdc_integration_kafka_test/1083/display/redirect) | true | `/test cdc-integration-kafka-test`

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

<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/OWNERS b/OWNERS
index 7b394eea9..2c803f0cb 100644
--- a/OWNERS
+++ b/OWNERS
@@ -18,6 +18,7 @@ approvers:
   - hi-rustin
   - hicqu
   - holys
+  - hongyunyan
   - IANTHEREAL
   - july2993
   - kennytm
diff --git a/cdc/entry/mounter.go b/cdc/entry/mounter.go
index 2f4a41a47..d920a55cc 100644
--- a/cdc/entry/mounter.go
+++ b/cdc/entry/mounter.go
@@ -418,13 +418,13 @@ func (m *mounter) calculateChecksum(
 	return checksum, nil
 }
 
-// return error when calculate the checksum.
+// return error when calculate the checksum failed.
 // return false if the checksum is not matched
 func (m *mounter) verifyChecksum(
 	columnInfos []*timodel.ColumnInfo, rawColumns []types.Datum, isPreRow bool,
-) (uint32, int, bool, error) {
+) (uint32, bool, error) {
 	if !m.integrity.Enabled() {
-		return 0, 0, true, nil
+		return 0, true, nil
 	}
 
 	var decoder *rowcodec.DatumMapDecoder
@@ -433,29 +433,25 @@ func (m *mounter) verifyChecksum(
 	} else {
 		decoder = m.decoder
 	}
-	if decoder == nil {
-		return 0, 0, false, errors.New("cannot found the decoder to get the checksum")
-	}
 
-	version := decoder.ChecksumVersion()
 	// if the checksum cannot be found, which means the upstream TiDB checksum is not enabled,
 	// so return matched as true to skip check the event.
 	first, ok := decoder.GetChecksum()
 	if !ok {
-		return 0, version, true, nil
+		return 0, true, nil
 	}
 
 	checksum, err := m.calculateChecksum(columnInfos, rawColumns)
 	if err != nil {
 		log.Error("failed to calculate the checksum", zap.Uint32("first", first), zap.Error(err))
-		return 0, version, false, errors.Trace(err)
+		return 0, false, err
 	}
 
 	// the first checksum matched, it hits in the most case.
 	if checksum == first {
 		log.Debug("checksum matched",
 			zap.Uint32("checksum", checksum), zap.Uint32("first", first))
-		return checksum, version, true, nil
+		return checksum, true, nil
 	}
 
 	extra, ok := decoder.GetExtraChecksum()
@@ -463,22 +459,21 @@ func (m *mounter) verifyChecksum(
 		log.Error("cannot found the extra checksum, the first checksum mismatched",
 			zap.Uint32("checksum", checksum),
 			zap.Uint32("first", first))
-		return checksum, version, false, nil
+		return checksum, false, nil
 	}
 
 	if checksum == extra {
-		log.Debug("extra checksum matched, this may happen the upstream TiDB is during the DDL"+
-			"execution phase",
+		log.Debug("extra checksum matched, this may happen the upstream TiDB is during the DDL execution phase",
 			zap.Uint32("checksum", checksum),
 			zap.Uint32("extra", extra))
-		return checksum, version, true, nil
+		return checksum, true, nil
 	}
 
 	log.Error("checksum mismatch",
 		zap.Uint32("checksum", checksum),
 		zap.Uint32("first", first),
 		zap.Uint32("extra", extra))
-	return checksum, version, false, nil
+	return checksum, false, nil
 }
 
 func (m *mounter) mountRowKVEntry(tableInfo *model.TableInfo, row *rowKVEntry, dataSize int64) (*model.RowChangedEvent, model.RowChangedDatums, error) {
@@ -494,6 +489,12 @@ func (m *mounter) mountRowKVEntry(tableInfo *model.TableInfo, row *rowKVEntry, d
 		corrupted       bool
 	)
 
+	if m.decoder != nil {
+		checksumVersion = m.decoder.ChecksumVersion()
+	} else if m.preDecoder != nil {
+		checksumVersion = m.preDecoder.ChecksumVersion()
+	}
+
 	// Decode previous columns.
 	var (
 		preCols     []*model.ColumnData
@@ -508,8 +509,11 @@ func (m *mounter) mountRowKVEntry(tableInfo *model.TableInfo, row *rowKVEntry, d
 			return nil, rawRow, errors.Trace(err)
 		}
 
-		preChecksum, checksumVersion, matched, err = m.verifyChecksum(columnInfos, preRawCols, true)
+		preChecksum, matched, err = m.verifyChecksum(columnInfos, preRawCols, true)
 		if err != nil {
+			log.Error("calculate the previous columns checksum failed",
+				zap.Any("tableInfo", tableInfo),
+				zap.Any("rawCols", preRawCols))
 			return nil, rawRow, errors.Trace(err)
 		}
 
@@ -517,10 +521,10 @@ func (m *mounter) mountRowKVEntry(tableInfo *model.TableInfo, row *rowKVEntry, d
 			log.Error("previous columns checksum mismatch",
 				zap.Uint32("checksum", preChecksum),
 				zap.Any("tableInfo", tableInfo),
-				zap.Any("row", row))
+				zap.Any("rawCols", preRawCols))
 			if m.integrity.ErrorHandle() {
 				return nil, rawRow, cerror.ErrCorruptedDataMutation.
-					GenWithStackByArgs(m.changefeedID.Namespace, m.changefeedID.ID, row)
+					GenWithStackByArgs(m.changefeedID.Namespace, m.changefeedID.ID)
 			}
 			corrupted = true
 		}
@@ -537,18 +541,21 @@ func (m *mounter) mountRowKVEntry(tableInfo *model.TableInfo, row *rowKVEntry, d
 			return nil, rawRow, errors.Trace(err)
 		}
 
-		currentChecksum, checksumVersion, matched, err = m.verifyChecksum(columnInfos, rawCols, false)
+		currentChecksum, matched, err = m.verifyChecksum(columnInfos, rawCols, false)
 		if err != nil {
+			log.Error("calculate the current columns checksum failed",
+				zap.Any("tableInfo", tableInfo),
+				zap.Any("rawCols", rawCols))
 			return nil, rawRow, errors.Trace(err)
 		}
 		if !matched {
-			log.Error("columns checksum mismatch",
+			log.Error("current columns checksum mismatch",
 				zap.Uint32("checksum", currentChecksum),
 				zap.Any("tableInfo", tableInfo),
 				zap.Any("rawCols", rawCols))
 			if m.integrity.ErrorHandle() {
 				return nil, rawRow, cerror.ErrCorruptedDataMutation.
-					GenWithStackByArgs(m.changefeedID.Namespace, m.changefeedID.ID, row)
+					GenWithStackByArgs(m.changefeedID.Namespace, m.changefeedID.ID)
 			}
 			corrupted = true
 		}
diff --git a/cdc/entry/schema/snapshot.go b/cdc/entry/schema/snapshot.go
index 94a10ae13..daf9e3078 100644
--- a/cdc/entry/schema/snapshot.go
+++ b/cdc/entry/schema/snapshot.go
@@ -123,6 +123,7 @@ func GetSchemaVersion(meta *timeta.Meta) (int64, error) {
 
 // NewSingleSnapshotFromMeta creates a new single schema snapshot from a tidb meta
 func NewSingleSnapshotFromMeta(
+	id model.ChangeFeedID,
 	meta *timeta.Meta,
 	currentTs uint64,
 	forceReplicate bool,
@@ -134,11 +135,12 @@ func NewSingleSnapshotFromMeta(
 		snap.inner.currentTs = currentTs
 		return snap, nil
 	}
-	return NewSnapshotFromMeta(meta, currentTs, forceReplicate, filter)
+	return NewSnapshotFromMeta(id, meta, currentTs, forceReplicate, filter)
 }
 
 // NewSnapshotFromMeta creates a schema snapshot from meta.
 func NewSnapshotFromMeta(
+	id model.ChangeFeedID,
 	meta *timeta.Meta,
 	currentTs uint64,
 	forceReplicate bool,
@@ -151,7 +153,8 @@ func NewSnapshotFromMeta(
 	}
 	// `tag` is used to reverse sort all versions in the generated snapshot.
 	tag := negative(currentTs)
-
+	// record all tables to be replicated for logging use
+	tables := make([]*model.TableInfo, 0, 1024)
 	for _, dbinfo := range dbinfos {
 		if filter.ShouldIgnoreSchema(dbinfo.Name.O) {
 			log.Debug("ignore database", zap.String("db", dbinfo.Name.O))
@@ -190,6 +193,8 @@ func NewSnapshotFromMeta(
 			ineligible := !tableInfo.IsEligible(forceReplicate)
 			if ineligible {
 				snap.inner.ineligibleTables.ReplaceOrInsert(versionedID{id: tableInfo.ID, tag: tag})
+			} else {
+				tables = append(tables, tableInfo)
 			}
 			if pi := tableInfo.GetPartitionInfo(); pi != nil {
 				for _, partition := range pi.Definitions {
@@ -204,6 +209,15 @@ func NewSnapshotFromMeta(
 		}
 	}
 	snap.inner.currentTs = currentTs
+	var sb strings.Builder
+	sb.WriteString(fmt.Sprintf("%d tables to be replicated: ", len(tables)))
+	for _, table := range tables {
+		sb.WriteString(fmt.Sprintf("%s.%s, ", table.TableName.Schema, table.TableName.Table))
+	}
+	log.Info("schema snapshot created",
+		zap.Stringer("changefeed", id),
+		zap.Uint64("currentTs", currentTs),
+		zap.String("tables", sb.String()))
 	return snap, nil
 }
 
diff --git a/cdc/entry/schema_storage.go b/cdc/entry/schema_storage.go
index f8e8d4072..a5c68b4dc 100644
--- a/cdc/entry/schema_storage.go
+++ b/cdc/entry/schema_storage.go
@@ -100,7 +100,7 @@ func NewSchemaStorage(
 		snap = schema.NewEmptySnapshot(forceReplicate)
 	} else {
 		meta := kv.GetSnapshotMeta(storage, startTs)
-		snap, err = schema.NewSnapshotFromMeta(meta, startTs, forceReplicate, filter)
+		snap, err = schema.NewSnapshotFromMeta(id, meta, startTs, forceReplicate, filter)
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
diff --git a/cdc/entry/validator.go b/cdc/entry/validator.go
index 3bb19bce5..e19ceb707 100644
--- a/cdc/entry/validator.go
+++ b/cdc/entry/validator.go
@@ -34,7 +34,9 @@ func VerifyTables(
 	err error,
 ) {
 	meta := kv.GetSnapshotMeta(storage, startTs)
-	snap, err := schema.NewSingleSnapshotFromMeta(meta, startTs, false /* explicitTables */, f)
+	snap, err := schema.NewSingleSnapshotFromMeta(
+		model.ChangeFeedID4Test("api", "verifyTable"),
+		meta, startTs, false /* explicitTables */, f)
 	if err != nil {
 		return nil, nil, nil, errors.Trace(err)
 	}
diff --git a/cdc/kv/client.go b/cdc/kv/client.go
index 92f0a6ef1..6a16475cd 100644
--- a/cdc/kv/client.go
+++ b/cdc/kv/client.go
@@ -478,6 +478,7 @@ func (s *eventFeedSession) eventFeed(ctx context.Context) error {
 	}()
 
 	g, ctx := errgroup.WithContext(ctx)
+	g.SetLimit(scanRegionsConcurrency)
 
 	g.Go(func() error { return s.dispatchRequest(ctx) })
 
@@ -486,8 +487,6 @@ func (s *eventFeedSession) eventFeed(ctx context.Context) error {
 	g.Go(func() error { return s.logSlowRegions(ctx) })
 
 	g.Go(func() error {
-		g, ctx := errgroup.WithContext(ctx)
-		g.SetLimit(scanRegionsConcurrency)
 		for {
 			select {
 			case <-ctx.Done():
diff --git a/cdc/metrics/metrics.go b/cdc/metrics/metrics.go
new file mode 100644
index 000000000..085999d15
--- /dev/null
+++ b/cdc/metrics/metrics.go
@@ -0,0 +1,36 @@
+// Copyright 2020 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package metrics
+
+import (
+	"github.com/prometheus/client_golang/prometheus"
+)
+
+// LagBucket returns the lag buckets for prometheus metric
+// 10 seconds is the reasonable LAG for most cases,
+// for prometheus histogram_quantile func,
+// we use small bucket distance to do accurate approximation
+func LagBucket() []float64 {
+	// 0.5s-10s in step of 0.5s
+	buckets := prometheus.LinearBuckets(0.5, 0.5, 20)
+	// 11s-20s in step of 1s
+	buckets = append(buckets, prometheus.LinearBuckets(11, 1, 10)...)
+	// 21s-591s in step of 30s
+	buckets = append(buckets, prometheus.LinearBuckets(21, 30, 20)...)
+	// 900s-3600 in step of 300s
+	buckets = append(buckets, prometheus.LinearBuckets(900, 300, 10)...)
+	// [4000, 8000, 16000, 32000]
+	buckets = append(buckets, prometheus.ExponentialBuckets(4000, 2, 4)...)
+	return buckets
+}
diff --git a/cdc/owner/metrics.go b/cdc/owner/metrics.go
index 851f813fa..e791fbba4 100644
--- a/cdc/owner/metrics.go
+++ b/cdc/owner/metrics.go
@@ -16,6 +16,7 @@ package owner
 import (
 	"time"
 
+	"github.com/pingcap/tiflow/cdc/metrics"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -56,7 +57,7 @@ var (
 			Subsystem: "owner",
 			Name:      "checkpoint_lag_histogram",
 			Help:      "checkpoint lag histogram of changefeeds",
-			Buckets:   lagBucket(),
+			Buckets:   metrics.LagBucket(),
 		}, []string{"namespace", "changefeed"})
 
 	changefeedResolvedTsGauge = prometheus.NewGaugeVec(
@@ -80,7 +81,7 @@ var (
 			Subsystem: "owner",
 			Name:      "resolved_ts_lag_histogram",
 			Help:      "resolved_ts lag histogram of changefeeds",
-			Buckets:   lagBucket(),
+			Buckets:   metrics.LagBucket(),
 		}, []string{"namespace", "changefeed"})
 
 	ownershipCounter = prometheus.NewCounter(
@@ -151,14 +152,3 @@ func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(changefeedCloseDuration)
 	registry.MustRegister(changefeedStartTimeGauge)
 }
-
-// lagBucket returns the lag buckets for prometheus metric
-// 10 seconds is the reasonable LAG for most cases,
-// for prometheus histogram_quantile func,
-// we use small bucket distance to do accurate approximation
-func lagBucket() []float64 {
-	buckets := prometheus.LinearBuckets(0.5, 0.5, 20)
-	buckets = append(buckets, prometheus.LinearBuckets(11, 1, 10)...)
-	buckets = append(buckets, prometheus.ExponentialBuckets(21, 2, 10)...)
-	return buckets
-}
diff --git a/cdc/processor/sinkmanager/manager.go b/cdc/processor/sinkmanager/manager.go
index fe5cfeefb..4792ff503 100644
--- a/cdc/processor/sinkmanager/manager.go
+++ b/cdc/processor/sinkmanager/manager.go
@@ -128,6 +128,8 @@ type SinkManager struct {
 
 	// Metric for table sink.
 	metricsTableSinkTotalRows prometheus.Counter
+
+	metricsTableSinkFlushLagDuration prometheus.Observer
 }
 
 // New creates a new sink manager.
@@ -154,6 +156,9 @@ func New(
 
 		metricsTableSinkTotalRows: tablesinkmetrics.TotalRowsCountCounter.
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
+
+		metricsTableSinkFlushLagDuration: tablesinkmetrics.TableSinkFlushLagDuration.
+			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 	}
 
 	totalQuota := changefeedInfo.Config.MemoryQuota
@@ -827,7 +832,7 @@ func (m *SinkManager) AddTable(span tablepb.Span, startTs model.Ts, targetTs mod
 			if m.sinkFactory.TryLock() {
 				defer m.sinkFactory.Unlock()
 				if m.sinkFactory.f != nil {
-					s = m.sinkFactory.f.CreateTableSink(m.changefeedID, span, startTs, m.metricsTableSinkTotalRows)
+					s = m.sinkFactory.f.CreateTableSink(m.changefeedID, span, startTs, m.up.PDClock, m.metricsTableSinkTotalRows, m.metricsTableSinkFlushLagDuration)
 					version = m.sinkFactory.version
 				}
 			}
diff --git a/cdc/sink/dmlsink/factory/factory.go b/cdc/sink/dmlsink/factory/factory.go
index 2e674cea5..6fd0dd4c2 100644
--- a/cdc/sink/dmlsink/factory/factory.go
+++ b/cdc/sink/dmlsink/factory/factory.go
@@ -132,16 +132,19 @@ func New(
 // CreateTableSink creates a TableSink by schema.
 func (s *SinkFactory) CreateTableSink(
 	changefeedID model.ChangeFeedID,
-	span tablepb.Span, startTs model.Ts,
+	span tablepb.Span,
+	startTs model.Ts,
+	PDClock pdutil.Clock,
 	totalRowsCounter prometheus.Counter,
+	flushLagDuration prometheus.Observer,
 ) tablesink.TableSink {
 	if s.txnSink != nil {
 		return tablesink.New(changefeedID, span, startTs, s.txnSink,
-			&dmlsink.TxnEventAppender{TableSinkStartTs: startTs}, totalRowsCounter)
+			&dmlsink.TxnEventAppender{TableSinkStartTs: startTs}, PDClock, totalRowsCounter, flushLagDuration)
 	}
 
 	return tablesink.New(changefeedID, span, startTs, s.rowSink,
-		&dmlsink.RowChangeEventAppender{}, totalRowsCounter)
+		&dmlsink.RowChangeEventAppender{}, PDClock, totalRowsCounter, flushLagDuration)
 }
 
 // CreateTableSinkForConsumer creates a TableSink by schema for consumer.
@@ -151,18 +154,21 @@ func (s *SinkFactory) CreateTableSink(
 func (s *SinkFactory) CreateTableSinkForConsumer(
 	changefeedID model.ChangeFeedID,
 	span tablepb.Span, startTs model.Ts,
-	totalRowsCounter prometheus.Counter,
 ) tablesink.TableSink {
 	if s.txnSink != nil {
 		return tablesink.New(changefeedID, span, startTs, s.txnSink,
 			// IgnoreStartTs is true because the consumer can
 			// **not** get the start ts of the row changed event.
 			&dmlsink.TxnEventAppender{TableSinkStartTs: startTs, IgnoreStartTs: true},
-			totalRowsCounter)
+			pdutil.NewClock4Test(),
+			prometheus.NewCounter(prometheus.CounterOpts{}),
+			prometheus.NewHistogram(prometheus.HistogramOpts{}))
 	}
 
 	return tablesink.New(changefeedID, span, startTs, s.rowSink,
-		&dmlsink.RowChangeEventAppender{}, totalRowsCounter)
+		&dmlsink.RowChangeEventAppender{}, pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 }
 
 // Close closes the sink.
diff --git a/cdc/sink/dmlsink/mq/dispatcher/event_router.go b/cdc/sink/dmlsink/mq/dispatcher/event_router.go
index e7d68014f..2d325b0c0 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/event_router.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/event_router.go
@@ -135,17 +135,19 @@ func (s *EventRouter) VerifyTables(infos []*model.TableInfo) error {
 		_, partitionDispatcher := s.matchDispatcher(table.TableName.Schema, table.TableName.Table)
 		switch v := partitionDispatcher.(type) {
 		case *partition.IndexValueDispatcher:
-			index := table.GetIndex(v.IndexName)
-			if index == nil {
-				return cerror.ErrDispatcherFailed.GenWithStack(
-					"index not found when verify the table, table: %v, index: %s", table.TableName, v.IndexName)
-			}
-			// only allow the unique index to be set.
-			// For the non-unique index, if any column belongs to the index is updated,
-			// the event is not split, it may cause incorrect data consumption.
-			if !index.Unique {
-				return cerror.ErrDispatcherFailed.GenWithStack(
-					"index is not unique when verify the table, table: %v, index: %s", table.TableName, v.IndexName)
+			if v.IndexName != "" {
+				index := table.GetIndex(v.IndexName)
+				if index == nil {
+					return cerror.ErrDispatcherFailed.GenWithStack(
+						"index not found when verify the table, table: %v, index: %s", table.TableName, v.IndexName)
+				}
+				// only allow the unique index to be set.
+				// For the non-unique index, if any column belongs to the index is updated,
+				// the event is not split, it may cause incorrect data consumption.
+				if !index.Unique {
+					return cerror.ErrDispatcherFailed.GenWithStack(
+						"index is not unique when verify the table, table: %v, index: %s", table.TableName, v.IndexName)
+				}
 			}
 		case *partition.ColumnsDispatcher:
 			_, ok := table.OffsetsByNames(v.Columns)
diff --git a/cdc/sink/metrics/tablesink/metrics.go b/cdc/sink/metrics/tablesink/metrics.go
index 4402b48ef..320dc651e 100644
--- a/cdc/sink/metrics/tablesink/metrics.go
+++ b/cdc/sink/metrics/tablesink/metrics.go
@@ -14,6 +14,7 @@
 package tablesink
 
 import (
+	"github.com/pingcap/tiflow/cdc/metrics"
 	"github.com/prometheus/client_golang/prometheus"
 )
 
@@ -26,7 +27,18 @@ var TotalRowsCountCounter = prometheus.NewCounterVec(
 		Help:      "The total count of rows that are processed by table sink",
 	}, []string{"namespace", "changefeed"})
 
+// TableSinkFlushLagDuration is the  per event flush lag calculated by ts_after_event_flushed_to_downstream - commit_ts_of_event
+var TableSinkFlushLagDuration = prometheus.NewHistogramVec(
+	prometheus.HistogramOpts{
+		Namespace: "ticdc",
+		Subsystem: "sink",
+		Name:      "flush_lag_histogram",
+		Help:      "flush lag histogram of rows that are processed by table sink",
+		Buckets:   metrics.LagBucket(),
+	}, []string{"namespace", "changefeed"})
+
 // InitMetrics registers all metrics in this file.
 func InitMetrics(registry *prometheus.Registry) {
 	registry.MustRegister(TotalRowsCountCounter)
+	registry.MustRegister(TableSinkFlushLagDuration)
 }
diff --git a/cdc/sink/tablesink/table_sink_impl.go b/cdc/sink/tablesink/table_sink_impl.go
index 76b65cc77..167f30fc9 100644
--- a/cdc/sink/tablesink/table_sink_impl.go
+++ b/cdc/sink/tablesink/table_sink_impl.go
@@ -22,7 +22,9 @@ import (
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink"
 	"github.com/pingcap/tiflow/cdc/sink/tablesink/state"
+	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/prometheus/client_golang/prometheus"
+	"github.com/tikv/client-go/v2/oracle"
 	"go.uber.org/zap"
 )
 
@@ -58,6 +60,7 @@ type EventTableSink[E dmlsink.TableEvent, P dmlsink.Appender[E]] struct {
 	backendSink     dmlsink.EventSink[E]
 	progressTracker *progressTracker
 	eventAppender   P
+	pdClock         pdutil.Clock
 	// NOTICE: It is ordered by commitTs.
 	eventBuffer []E
 	state       state.TableSinkState
@@ -66,6 +69,8 @@ type EventTableSink[E dmlsink.TableEvent, P dmlsink.Appender[E]] struct {
 
 	// For dataflow metrics.
 	metricsTableSinkTotalRows prometheus.Counter
+
+	metricsTableSinkFlushLagDuration prometheus.Observer
 }
 
 // New an eventTableSink with given backendSink and event appender.
@@ -75,20 +80,24 @@ func New[E dmlsink.TableEvent, P dmlsink.Appender[E]](
 	startTs model.Ts,
 	backendSink dmlsink.EventSink[E],
 	appender P,
+	pdClock pdutil.Clock,
 	totalRowsCounter prometheus.Counter,
+	flushLagDuration prometheus.Observer,
 ) *EventTableSink[E, P] {
 	return &EventTableSink[E, P]{
-		changefeedID:              changefeedID,
-		span:                      span,
-		startTs:                   startTs,
-		maxResolvedTs:             model.NewResolvedTs(0),
-		backendSink:               backendSink,
-		progressTracker:           newProgressTracker(span, defaultBufferSize),
-		eventAppender:             appender,
-		eventBuffer:               make([]E, 0, 1024),
-		state:                     state.TableSinkSinking,
-		lastSyncedTs:              LastSyncedTsRecord{lastSyncedTs: startTs},
-		metricsTableSinkTotalRows: totalRowsCounter,
+		changefeedID:                     changefeedID,
+		span:                             span,
+		startTs:                          startTs,
+		maxResolvedTs:                    model.NewResolvedTs(0),
+		backendSink:                      backendSink,
+		progressTracker:                  newProgressTracker(span, defaultBufferSize),
+		eventAppender:                    appender,
+		pdClock:                          pdClock,
+		eventBuffer:                      make([]E, 0, 1024),
+		state:                            state.TableSinkSinking,
+		lastSyncedTs:                     LastSyncedTsRecord{lastSyncedTs: startTs},
+		metricsTableSinkTotalRows:        totalRowsCounter,
+		metricsTableSinkFlushLagDuration: flushLagDuration,
 	}
 }
 
@@ -135,6 +144,7 @@ func (e *EventTableSink[E, P]) UpdateResolvedTs(resolvedTs model.ResolvedTs) err
 		// We have to record the event ID for the callback.
 		postEventFlushFunc := e.progressTracker.addEvent()
 		evCommitTs := ev.GetCommitTs()
+		phyCommitTs := oracle.ExtractPhysical(evCommitTs)
 		ce := &dmlsink.CallbackableEvent[E]{
 			Event: ev,
 			Callback: func() {
@@ -149,6 +159,10 @@ func (e *EventTableSink[E, P]) UpdateResolvedTs(resolvedTs model.ResolvedTs) err
 						e.lastSyncedTs.lastSyncedTs = evCommitTs
 					}
 				}
+				pdTime := e.pdClock.CurrentTime()
+				currentTs := oracle.GetPhysical(pdTime)
+				flushLag := float64(currentTs-phyCommitTs) / 1e3
+				e.metricsTableSinkFlushLagDuration.Observe(flushLag)
 				postEventFlushFunc()
 			},
 			SinkState: &e.state,
diff --git a/cmd/kafka-consumer/main.go b/cmd/kafka-consumer/main.go
index 2b4d70ea6..0c7681708 100644
--- a/cmd/kafka-consumer/main.go
+++ b/cmd/kafka-consumer/main.go
@@ -59,7 +59,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/spanz"
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/pingcap/tiflow/pkg/version"
-	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 )
 
@@ -739,7 +738,6 @@ func (c *Consumer) ConsumeClaim(session sarama.ConsumerGroupSession, claim saram
 							model.DefaultChangeFeedID("kafka-consumer"),
 							spanz.TableIDToComparableSpan(tableID),
 							events[0].CommitTs,
-							prometheus.NewCounter(prometheus.CounterOpts{}),
 						))
 					}
 					s, _ := sink.tableSinksMap.Load(tableID)
diff --git a/cmd/pulsar-consumer/main.go b/cmd/pulsar-consumer/main.go
index 0a9e06840..a21829acc 100644
--- a/cmd/pulsar-consumer/main.go
+++ b/cmd/pulsar-consumer/main.go
@@ -49,7 +49,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/spanz"
 	"github.com/pingcap/tiflow/pkg/util"
 	"github.com/pingcap/tiflow/pkg/version"
-	"github.com/prometheus/client_golang/prometheus"
 	"github.com/spf13/cobra"
 	"go.uber.org/zap"
 )
@@ -556,8 +555,7 @@ func (c *Consumer) HandleMsg(msg pulsar.Message) error {
 					tableSink := c.sinkFactory.CreateTableSinkForConsumer(
 						model.DefaultChangeFeedID("pulsar-consumer"),
 						spanz.TableIDToComparableSpan(tableID),
-						events[0].CommitTs,
-						prometheus.NewCounter(prometheus.CounterOpts{}))
+						events[0].CommitTs)
 
 					log.Info("table sink created", zap.Any("tableID", tableID),
 						zap.Any("tableSink", tableSink.GetCheckpointTs()))
diff --git a/cmd/storage-consumer/main.go b/cmd/storage-consumer/main.go
index 6e467d348..542c53a59 100644
--- a/cmd/storage-consumer/main.go
+++ b/cmd/storage-consumer/main.go
@@ -50,7 +50,6 @@ import (
 	"github.com/pingcap/tiflow/pkg/spanz"
 	putil "github.com/pingcap/tiflow/pkg/util"
 	"github.com/pingcap/tiflow/pkg/version"
-	"github.com/prometheus/client_golang/prometheus"
 	"go.uber.org/zap"
 )
 
@@ -349,8 +348,7 @@ func (c *consumer) emitDMLEvents(
 				c.tableSinkMap[tableID] = c.sinkFactory.CreateTableSinkForConsumer(
 					model.DefaultChangeFeedID(defaultChangefeedName),
 					spanz.TableIDToComparableSpan(tableID),
-					row.CommitTs,
-					prometheus.NewCounter(prometheus.CounterOpts{}))
+					row.CommitTs)
 			}
 
 			_, ok := c.tableTsMap[tableID]
diff --git a/errors.toml b/errors.toml
index 8a5cc2009..21995b0c8 100755
--- a/errors.toml
+++ b/errors.toml
@@ -203,7 +203,7 @@ failed to convert ddl '%s' to filter event type
 
 ["CDC:ErrCorruptedDataMutation"]
 error = '''
-Changefeed %s.%s stopped due to corrupted data mutation received. Corrupted mutation detail information %+v
+Changefeed %s.%s stopped due to corrupted data mutation received
 '''
 
 ["CDC:ErrCraftCodecInvalidData"]
diff --git a/go.mod b/go.mod
index 525270121..d5456c299 100644
--- a/go.mod
+++ b/go.mod
@@ -70,7 +70,7 @@ require (
 	github.com/pingcap/log v1.1.1-0.20230317032135-a0d097d16e22
 	github.com/pingcap/tidb v1.1.0-beta.0.20240228123331-27ce02afd2e3
 	github.com/pingcap/tidb-dashboard v0.0.0-20240127080020-2171c6e6b9d7
-	github.com/pingcap/tidb-tools v0.0.0-20240202030925-a6014db89eb8
+	github.com/pingcap/tidb-tools v0.0.0-20240305021104-9f9bea84490b
 	github.com/pingcap/tidb/pkg/parser v0.0.0-20240228123331-27ce02afd2e3
 	github.com/prometheus/client_golang v1.18.0
 	github.com/prometheus/client_model v0.6.0
@@ -411,3 +411,5 @@ replace gorm.io/driver/mysql v1.4.5 => gorm.io/driver/mysql v1.3.3
 replace sourcegraph.com/sourcegraph/appdash => github.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0
 
 replace sourcegraph.com/sourcegraph/appdash-data => github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67
+
+replace blainsmith.com/go/seahash => github.com/YangKeao/seahash v0.0.0-20240229041150-e7bf269c3140
diff --git a/go.sum b/go.sum
index cb4a0dbfa..338079ac1 100644
--- a/go.sum
+++ b/go.sum
@@ -1,5 +1,3 @@
-blainsmith.com/go/seahash v1.2.1 h1:6tEG97FuuhU06FdFD4tXZlyw7saa9CoXBorsc+rnuZA=
-blainsmith.com/go/seahash v1.2.1/go.mod h1:pnKU6FdQMfYOGoDBo5FPdlZG7kmOJ/nuNyfu0czty8k=
 cloud.google.com/go v0.26.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.34.0/go.mod h1:aQUYkXzVsufM+DwF1aE+0xfcU+56JwCaLick0ClmMTw=
 cloud.google.com/go v0.38.0/go.mod h1:990N+gfupTy94rShfmMCWGDn0LpTmnzTp2qbd1dvSRU=
@@ -86,6 +84,8 @@ github.com/VividCortex/ewma v1.2.0 h1:f58SaIzcDXrSy3kWaHNvuJgJ3Nmz59Zji6XoJR/q1o
 github.com/VividCortex/ewma v1.2.0/go.mod h1:nz4BbCtbLyFDeC9SUHbtcT5644juEuWfUAUnGx7j5l4=
 github.com/VividCortex/mysqlerr v1.0.0 h1:5pZ2TZA+YnzPgzBfiUWGqWmKDVNBdrkf9g+DNe1Tiq8=
 github.com/VividCortex/mysqlerr v1.0.0/go.mod h1:xERx8E4tBhLvpjzdUyQiSfUxeMcATEQrflDAfXsqcAE=
+github.com/YangKeao/seahash v0.0.0-20240229041150-e7bf269c3140 h1:t7DNyDoEPlT5YYkkTBHEH90VZ3GSZXCPD4d2zLnqhds=
+github.com/YangKeao/seahash v0.0.0-20240229041150-e7bf269c3140/go.mod h1:pnKU6FdQMfYOGoDBo5FPdlZG7kmOJ/nuNyfu0czty8k=
 github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d h1:licZJFw2RwpHMqeKTCYkitsPqHNxTmd4SNR5r94FGM8=
 github.com/acarl005/stripansi v0.0.0-20180116102854-5a71ef0e047d/go.mod h1:asat636LX7Bqt5lYEZ27JNDcqxfjdBQuJ/MM4CN/Lzo=
 github.com/ajg/form v1.5.1/go.mod h1:uL1WgH+h2mgNtvBq0339dVnzXdBETtL2LeUXaIv25UY=
@@ -877,8 +877,8 @@ github.com/pingcap/tidb v1.1.0-beta.0.20240228123331-27ce02afd2e3 h1:vLUU7S6yhHe
 github.com/pingcap/tidb v1.1.0-beta.0.20240228123331-27ce02afd2e3/go.mod h1:pwZI+N72Xb+2KsEragMzRmb4EEUFSbrOS0mMcaHMFCM=
 github.com/pingcap/tidb-dashboard v0.0.0-20240127080020-2171c6e6b9d7 h1:M+kqgNnOnzjRb5NubF7QFhyDeQKYcgtLHB12gzYPoUk=
 github.com/pingcap/tidb-dashboard v0.0.0-20240127080020-2171c6e6b9d7/go.mod h1:ucZBRz52icb23T/5Z4CsuUHmarYiin7p2MeiVBe+o8c=
-github.com/pingcap/tidb-tools v0.0.0-20240202030925-a6014db89eb8 h1:oJVvhhIpYrBxiIPUC2M9mXZw4M7EdX8DBiac35YCXzs=
-github.com/pingcap/tidb-tools v0.0.0-20240202030925-a6014db89eb8/go.mod h1:FupVn04P2/dkvzK4Tt5URj5z2Y4ILcULw7dU3GnJXgQ=
+github.com/pingcap/tidb-tools v0.0.0-20240305021104-9f9bea84490b h1:UC4lLT2OBHGImFJbiXQfIxIck7AoFLIiRJ6Eo6uFVaw=
+github.com/pingcap/tidb-tools v0.0.0-20240305021104-9f9bea84490b/go.mod h1:FupVn04P2/dkvzK4Tt5URj5z2Y4ILcULw7dU3GnJXgQ=
 github.com/pingcap/tidb/pkg/parser v0.0.0-20240228123331-27ce02afd2e3 h1:YF17teH7c/WKeIc6MO8gEKiNe25X36nKxxtPPwO+/O0=
 github.com/pingcap/tidb/pkg/parser v0.0.0-20240228123331-27ce02afd2e3/go.mod h1:MWQK6otJgZRI6zcCVPV22U4qE26qOGJnN4fq8XawgBs=
 github.com/pingcap/tipb v0.0.0-20240116032918-9bb28c43bbfc h1:sEp4lbExDfnMX8HXQyhZrhqo2/SgeFY5KOdo5akc8FM=
diff --git a/metrics/grafana/ticdc.json b/metrics/grafana/ticdc.json
index fbad520dc..037feac77 100644
--- a/metrics/grafana/ticdc.json
+++ b/metrics/grafana/ticdc.json
@@ -1768,6 +1768,122 @@
             "alignLevel": null
           }
         },
+        {
+          "aliasColors": {},
+          "bars": false,
+          "dashLength": 10,
+          "dashes": false,
+          "datasource": "${DS_TEST-CLUSTER}",
+          "description": "The lag between the commit ts of the events and the time when they are flushed to downstream. ",
+          "fieldConfig": {
+            "defaults": {},
+            "overrides": []
+          },
+          "fill": 1,
+          "fillGradient": 0,
+          "gridPos": {
+            "h": 4,
+            "w": 12,
+            "x": 0,
+            "y": 37
+          },
+          "hiddenSeries": false,
+          "id": 727,
+          "legend": {
+            "alignAsTable": true,
+            "avg": false,
+            "current": true,
+            "max": true,
+            "min": false,
+            "rightSide": true,
+            "show": true,
+            "total": false,
+            "values": true
+          },
+          "lines": true,
+          "linewidth": 1,
+          "nullPointMode": "null",
+          "options": {
+            "alertThreshold": true
+          },
+          "percentage": false,
+          "pluginVersion": "7.5.11",
+          "pointradius": 2,
+          "points": false,
+          "renderer": "flot",
+          "seriesOverrides": [],
+          "spaceLength": 10,
+          "stack": false,
+          "steppedLine": false,
+          "targets": [
+            {
+              "exemplar": true,
+              "expr": "histogram_quantile(1.00, sum(rate(ticdc_sink_flush_lag_histogram_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\"}[1m])) by (le, instance, type))",
+              "interval": "",
+              "legendFormat": "{{instance}}-{{type}}-max",
+              "queryType": "randomWalk",
+              "refId": "A"
+            },
+            {
+              "exemplar": true,
+              "expr": "histogram_quantile(0.95, sum(rate(ticdc_sink_flush_lag_histogram_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\"}[1m])) by (le, instance, type))",
+              "hide": false,
+              "interval": "",
+              "legendFormat": "{{instance}}-{{type}}-95",
+              "refId": "B"
+            },
+            {
+              "exemplar": true,
+              "expr": "histogram_quantile(0.8, sum(rate(ticdc_sink_flush_lag_histogram_bucket{k8s_cluster=\"$k8s_cluster\", tidb_cluster=\"$tidb_cluster\"}[1m])) by (le, instance, type))",
+              "hide": false,
+              "interval": "",
+              "legendFormat": "{{instance}}-{{type}}-80",
+              "refId": "C"
+            }
+          ],
+          "thresholds": [],
+          "timeFrom": null,
+          "timeRegions": [],
+          "timeShift": null,
+          "title": "Events Flush Lag",
+          "tooltip": {
+            "shared": true,
+            "sort": 0,
+            "value_type": "individual"
+          },
+          "type": "graph",
+          "xaxis": {
+            "buckets": null,
+            "mode": "time",
+            "name": null,
+            "show": true,
+            "values": []
+          },
+          "yaxes": [
+            {
+              "$$hashKey": "object:869",
+              "format": "s",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": "0",
+              "show": true
+            },
+            {
+              "$$hashKey": "object:870",
+              "format": "short",
+              "label": null,
+              "logBase": 1,
+              "max": null,
+              "min": null,
+              "show": true
+            }
+          ],
+          "yaxis": {
+            "align": false,
+            "alignLevel": null
+          }
+        },
         {
           "aliasColors": {},
           "bars": false,
diff --git a/pkg/applier/redo.go b/pkg/applier/redo.go
index 396f5b7bc..833cfc00d 100644
--- a/pkg/applier/redo.go
+++ b/pkg/applier/redo.go
@@ -29,6 +29,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/sink/tablesink"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/redo"
 	"github.com/pingcap/tiflow/pkg/sink/mysql"
 	"github.com/pingcap/tiflow/pkg/spanz"
@@ -311,7 +312,9 @@ func (ra *RedoApplier) applyRow(
 			model.DefaultChangeFeedID(applierChangefeed),
 			spanz.TableIDToComparableSpan(tableID),
 			checkpointTs,
+			pdutil.NewClock4Test(),
 			prometheus.NewCounter(prometheus.CounterOpts{}),
+			prometheus.NewHistogram(prometheus.HistogramOpts{}),
 		)
 		ra.tableSinks[tableID] = tableSink
 	}
diff --git a/pkg/errors/cdc_errors.go b/pkg/errors/cdc_errors.go
index 58d60b90e..dbc27fddb 100644
--- a/pkg/errors/cdc_errors.go
+++ b/pkg/errors/cdc_errors.go
@@ -571,8 +571,7 @@ var (
 	)
 
 	ErrCorruptedDataMutation = errors.Normalize(
-		"Changefeed %s.%s stopped due to corrupted data mutation received. "+
-			"Corrupted mutation detail information %+v",
+		"Changefeed %s.%s stopped due to corrupted data mutation received",
 		errors.RFCCodeText("CDC:ErrCorruptedDataMutation"))
 
 	// server related errors
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index d9c1bf794..0cea36228 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -132,7 +132,7 @@ func (f *sqlEventFilter) getRules(schema, table string) []*sqlEventRule {
 }
 
 // skipDDLEvent skips ddl event by its type and query.
-func (f *sqlEventFilter) shouldSkipDDL(ddl *model.DDLEvent) (bool, error) {
+func (f *sqlEventFilter) shouldSkipDDL(ddl *model.DDLEvent) (skip bool, err error) {
 	if len(f.rules) == 0 {
 		return false, nil
 	}
diff --git a/pkg/sink/codec/bootstraper.go b/pkg/sink/codec/bootstraper.go
index b2a371c77..1ec99ef05 100644
--- a/pkg/sink/codec/bootstraper.go
+++ b/pkg/sink/codec/bootstraper.go
@@ -109,7 +109,7 @@ func (b *bootstrapWorker) addEvent(
 ) error {
 	table, ok := b.activeTables.Load(row.PhysicalTableID)
 	if !ok {
-		tb := newTableStatus(key, row)
+		tb := newTableStatistic(key, row)
 		b.activeTables.Store(tb.id, tb)
 		// Send bootstrap message immediately when a new table is added
 		err := b.sendBootstrapMsg(ctx, tb)
@@ -218,7 +218,7 @@ type tableStatistic struct {
 	tableInfo atomic.Value
 }
 
-func newTableStatus(key model.TopicPartitionKey, row *model.RowChangedEvent) *tableStatistic {
+func newTableStatistic(key model.TopicPartitionKey, row *model.RowChangedEvent) *tableStatistic {
 	res := &tableStatistic{
 		id:    row.PhysicalTableID,
 		topic: key.Topic,
@@ -245,11 +245,13 @@ func (t *tableStatistic) update(row *model.RowChangedEvent, totalPartition int32
 	t.counter.Add(1)
 	t.lastMsgReceivedTime.Store(time.Now())
 
-	if t.version.Load() != row.TableInfo.UpdateTS {
+	// Note(dongmen): Rename Table DDL is a special case,
+	// the TableInfo.Name is changed but the TableInfo.UpdateTs is not changed.
+	if t.version.Load() != row.TableInfo.UpdateTS ||
+		t.tableInfo.Load().(*model.TableInfo).Name != row.TableInfo.Name {
 		t.version.Store(row.TableInfo.UpdateTS)
 		t.tableInfo.Store(row.TableInfo)
 	}
-
 	if t.totalPartition.Load() != totalPartition {
 		t.totalPartition.Store(totalPartition)
 	}
diff --git a/pkg/sink/codec/common/verify_checksum.go b/pkg/sink/codec/common/verify_checksum.go
index ebac24de2..01c913195 100644
--- a/pkg/sink/codec/common/verify_checksum.go
+++ b/pkg/sink/codec/common/verify_checksum.go
@@ -18,6 +18,7 @@ import (
 	"hash/crc32"
 	"math"
 	"strconv"
+	"strings"
 	"time"
 
 	"github.com/pingcap/errors"
@@ -154,16 +155,22 @@ func buildChecksumBytes(buf []byte, value interface{}, mysqlType byte) ([]byte,
 	case mysql.TypeTimestamp:
 		location := config.GetDefaultServerConfig().TZ
 		timestamp := value.(string)
+		// if timestamp contains microseconds,
+		// keep it in the value to match the TiDB representation.
+		format := "2006-01-02 15:04:05"
+		if strings.Contains(timestamp, ".") {
+			format = "2006-01-02 15:04:05.999999"
+		}
 
 		loc, err := util.GetTimezone(location)
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
-		t, err := time.ParseInLocation("2006-01-02 15:04:05", timestamp, loc)
+		t, err := time.ParseInLocation(format, timestamp, loc)
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
-		timestamp = t.UTC().Format("2006-01-02 15:04:05")
+		timestamp = t.UTC().Format(format)
 		buf = appendLengthValue(buf, []byte(timestamp))
 	// all encoded as string
 	case mysql.TypeDatetime, mysql.TypeDate, mysql.TypeDuration, mysql.TypeNewDate:
diff --git a/pkg/sink/codec/simple/message.go b/pkg/sink/codec/simple/message.go
index 71af215f8..b89ca2189 100644
--- a/pkg/sink/codec/simple/message.go
+++ b/pkg/sink/codec/simple/message.go
@@ -415,13 +415,19 @@ func buildRowChangedEvent(
 	result.PreColumns = model.Columns2ColumnDatas(preColumns, tableInfo)
 
 	if enableRowChecksum && msg.Checksum != nil {
+		var (
+			previousCorrupted bool
+			currentCorrupted  bool
+		)
 		err = common.VerifyChecksum(preColumns, msg.Checksum.Previous)
 		if err != nil {
-			return nil, cerror.WrapError(cerror.ErrDecodeFailed, err)
+			log.Info("checksum corrupted on the previous columns")
+			currentCorrupted = true
 		}
 		err = common.VerifyChecksum(columns, msg.Checksum.Current)
 		if err != nil {
-			return nil, cerror.WrapError(cerror.ErrDecodeFailed, err)
+			log.Info("checksum corrupted on the previous columns")
+			currentCorrupted = true
 		}
 
 		result.Checksum = &integrity.Checksum{
@@ -431,8 +437,8 @@ func buildRowChangedEvent(
 			Version:   msg.Checksum.Version,
 		}
 
-		if msg.Checksum.Corrupted {
-			log.Warn("cdc detect checksum corrupted",
+		if msg.Checksum.Corrupted || previousCorrupted {
+			log.Warn("cdc detect previous checksum corrupted",
 				zap.String("schema", msg.Schema),
 				zap.String("table", msg.Table))
 			for _, col := range preColumns {
@@ -444,6 +450,13 @@ func buildRowChangedEvent(
 					zap.Any("value", col.Value),
 					zap.Any("default", col.Default))
 			}
+			return nil, cerror.ErrDecodeFailed.GenWithStackByArgs("checksum corrupted")
+		}
+
+		if msg.Checksum.Corrupted || currentCorrupted {
+			log.Warn("cdc detect checksum corrupted",
+				zap.String("schema", msg.Schema),
+				zap.String("table", msg.Table))
 			for _, col := range columns {
 				log.Info("data corrupted, print each column for debugging",
 					zap.String("name", col.Name),
@@ -453,7 +466,9 @@ func buildRowChangedEvent(
 					zap.Any("value", col.Value),
 					zap.Any("default", col.Default))
 			}
+			return nil, cerror.ErrDecodeFailed.GenWithStackByArgs("checksum corrupted")
 		}
+
 	}
 
 	return result, nil


Test patch:
diff --git a/cdc/entry/schema_storage_test.go b/cdc/entry/schema_storage_test.go
index ba7cb9926..0b0d02d35 100644
--- a/cdc/entry/schema_storage_test.go
+++ b/cdc/entry/schema_storage_test.go
@@ -694,7 +694,7 @@ func TestCreateSnapFromMeta(t *testing.T) {
 	meta := kv.GetSnapshotMeta(store, ver.Ver)
 	f, err := filter.NewFilter(config.GetDefaultReplicaConfig(), "")
 	require.Nil(t, err)
-	snap, err := schema.NewSnapshotFromMeta(meta, ver.Ver, false, f)
+	snap, err := schema.NewSnapshotFromMeta(model.DefaultChangeFeedID("test"), meta, ver.Ver, false, f)
 	require.Nil(t, err)
 	_, ok := snap.TableByName("test", "simple_test1")
 	require.True(t, ok)
@@ -731,12 +731,12 @@ func TestExplicitTables(t *testing.T) {
 	meta1 := kv.GetSnapshotMeta(store, ver1.Ver)
 	f, err := filter.NewFilter(config.GetDefaultReplicaConfig(), "")
 	require.Nil(t, err)
-	snap1, err := schema.NewSnapshotFromMeta(meta1, ver1.Ver, true /* forceReplicate */, f)
+	snap1, err := schema.NewSnapshotFromMeta(model.DefaultChangeFeedID("test"), meta1, ver1.Ver, true /* forceReplicate */, f)
 	require.Nil(t, err)
 	meta2 := kv.GetSnapshotMeta(store, ver2.Ver)
-	snap2, err := schema.NewSnapshotFromMeta(meta2, ver2.Ver, false /* forceReplicate */, f)
+	snap2, err := schema.NewSnapshotFromMeta(model.DefaultChangeFeedID("test"), meta2, ver2.Ver, false /* forceReplicate */, f)
 	require.Nil(t, err)
-	snap3, err := schema.NewSnapshotFromMeta(meta2, ver2.Ver, true /* forceReplicate */, f)
+	snap3, err := schema.NewSnapshotFromMeta(model.DefaultChangeFeedID("test"), meta2, ver2.Ver, true /* forceReplicate */, f)
 	require.Nil(t, err)
 
 	// we don't need to count system tables since TiCDC
@@ -890,7 +890,7 @@ func TestSchemaStorage(t *testing.T) {
 		for _, job := range jobs {
 			ts := job.BinlogInfo.FinishedTS
 			meta := kv.GetSnapshotMeta(store, ts)
-			snapFromMeta, err := schema.NewSnapshotFromMeta(meta, ts, false, f)
+			snapFromMeta, err := schema.NewSnapshotFromMeta(model.DefaultChangeFeedID("test"), meta, ts, false, f)
 			require.Nil(t, err)
 			snapFromSchemaStore, err := schemaStorage.GetSnapshot(ctx, ts)
 			require.Nil(t, err)
@@ -972,7 +972,7 @@ func TestHandleKey(t *testing.T) {
 	meta := kv.GetSnapshotMeta(store, ver.Ver)
 	f, err := filter.NewFilter(config.GetDefaultReplicaConfig(), "")
 	require.Nil(t, err)
-	snap, err := schema.NewSnapshotFromMeta(meta, ver.Ver, false, f)
+	snap, err := schema.NewSnapshotFromMeta(model.DefaultChangeFeedID("test"), meta, ver.Ver, false, f)
 	require.Nil(t, err)
 	tb1, ok := snap.TableByName("test", "simple_test1")
 	require.True(t, ok)
diff --git a/cdc/owner/metrics_test.go b/cdc/metrics/metrics_test.go
similarity index 78%
rename from cdc/owner/metrics_test.go
rename to cdc/metrics/metrics_test.go
index f45c669bb..deb5822df 100644
--- a/cdc/owner/metrics_test.go
+++ b/cdc/metrics/metrics_test.go
@@ -11,7 +11,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-package owner
+package metrics
 
 import (
 	"testing"
@@ -20,10 +20,12 @@ import (
 )
 
 func TestLagBucket(t *testing.T) {
-	buckets := lagBucket()
-	require.Equal(t, 40, len(buckets))
+	buckets := LagBucket()
+	require.Equal(t, 64, len(buckets))
 	require.Equal(t, 0.5, buckets[0])
 	require.Equal(t, 1.0, buckets[1])
 	require.Equal(t, 21.0, buckets[30])
-	require.Equal(t, float64(10752), buckets[39])
+	require.Equal(t, 900.0, buckets[50])
+	require.Equal(t, 4000.0, buckets[60])
+	require.Equal(t, float64(32000), buckets[63])
 }
diff --git a/cdc/processor/sinkmanager/table_sink_wrapper_test.go b/cdc/processor/sinkmanager/table_sink_wrapper_test.go
index 9afc438cb..2a986ec13 100644
--- a/cdc/processor/sinkmanager/table_sink_wrapper_test.go
+++ b/cdc/processor/sinkmanager/table_sink_wrapper_test.go
@@ -24,6 +24,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink"
 	"github.com/pingcap/tiflow/cdc/sink/tablesink"
+	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/sink"
 	"github.com/pingcap/tiflow/pkg/spanz"
 	"github.com/prometheus/client_golang/prometheus"
@@ -105,7 +106,10 @@ func createTableSinkWrapper(
 	sink := newMockSink()
 	innerTableSink := tablesink.New[*model.RowChangedEvent](
 		changefeedID, span, model.Ts(0),
-		sink, &dmlsink.RowChangeEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.RowChangeEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 	wrapper := newTableSinkWrapper(
 		changefeedID,
 		span,
@@ -270,7 +274,9 @@ func TestTableSinkWrapperSinkVersion(t *testing.T) {
 	innerTableSink := tablesink.New[*model.RowChangedEvent](
 		model.ChangeFeedID{}, tablepb.Span{}, model.Ts(0),
 		newMockSink(), &dmlsink.RowChangeEventAppender{},
+		pdutil.NewClock4Test(),
 		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}),
 	)
 	version := new(uint64)
 
@@ -316,7 +322,9 @@ func TestTableSinkWrapperSinkInner(t *testing.T) {
 	innerTableSink := tablesink.New[*model.RowChangedEvent](
 		model.ChangeFeedID{}, tablepb.Span{}, model.Ts(0),
 		newMockSink(), &dmlsink.RowChangeEventAppender{},
+		pdutil.NewClock4Test(),
 		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}),
 	)
 	version := new(uint64)
 
diff --git a/cdc/sink/dmlsink/factory/factory_test.go b/cdc/sink/dmlsink/factory/factory_test.go
index d325938a1..0f4a729ab 100644
--- a/cdc/sink/dmlsink/factory/factory_test.go
+++ b/cdc/sink/dmlsink/factory/factory_test.go
@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink/mq/dmlproducer"
 	"github.com/pingcap/tiflow/pkg/config"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
+	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/sink/kafka"
 	"github.com/pingcap/tiflow/pkg/spanz"
 	"github.com/prometheus/client_golang/prometheus"
@@ -85,7 +86,11 @@ func TestSinkFactory(t *testing.T) {
 	require.NotNil(t, sinkFactory.txnSink)
 
 	tableSink := sinkFactory.CreateTableSink(model.DefaultChangeFeedID("1"),
-		spanz.TableIDToComparableSpan(1), 0, prometheus.NewCounter(prometheus.CounterOpts{}))
+		spanz.TableIDToComparableSpan(1),
+		0,
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 	require.NotNil(t, tableSink, "table sink can be created")
 
 	sinkFactory.Close()
diff --git a/cdc/sink/dmlsink/mq/dispatcher/event_router_test.go b/cdc/sink/dmlsink/mq/dispatcher/event_router_test.go
index e15584781..44fffc059 100644
--- a/cdc/sink/dmlsink/mq/dispatcher/event_router_test.go
+++ b/cdc/sink/dmlsink/mq/dispatcher/event_router_test.go
@@ -403,3 +403,22 @@ func TestGetTopicForDDL(t *testing.T) {
 		require.Equal(t, test.expectedTopic, d.GetTopicForDDL(test.ddl))
 	}
 }
+
+func TestVerifyTables(t *testing.T) {
+	t.Parallel()
+
+	replicaConfig := newReplicaConfig4DispatcherTest()
+	d, err := NewEventRouter(replicaConfig, config.ProtocolCanalJSON, "test", sink.KafkaScheme)
+	require.NoError(t, err)
+
+	cols := []*model.Column{
+		{
+			Name:  "id",
+			Value: 1,
+			Flag:  model.HandleKeyFlag | model.PrimaryKeyFlag,
+		},
+	}
+	tableInfo := model.BuildTableInfo("test_index_value", "table", cols, [][]int{{0}})
+	err = d.VerifyTables([]*model.TableInfo{tableInfo})
+	require.NoError(t, err)
+}
diff --git a/cdc/sink/tablesink/table_sink_impl_test.go b/cdc/sink/tablesink/table_sink_impl_test.go
index e925ea05c..87fd1af4a 100644
--- a/cdc/sink/tablesink/table_sink_impl_test.go
+++ b/cdc/sink/tablesink/table_sink_impl_test.go
@@ -22,6 +22,7 @@ import (
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/cdc/sink/dmlsink"
 	"github.com/pingcap/tiflow/cdc/sink/tablesink/state"
+	"github.com/pingcap/tiflow/pkg/pdutil"
 	"github.com/pingcap/tiflow/pkg/sink"
 	"github.com/pingcap/tiflow/pkg/spanz"
 	"github.com/prometheus/client_golang/prometheus"
@@ -167,7 +168,10 @@ func TestNewEventTableSink(t *testing.T) {
 	sink := &mockEventSink{dead: make(chan struct{})}
 	tb := New[*model.SingleTableTxn](
 		model.DefaultChangeFeedID("1"), spanz.TableIDToComparableSpan(1), model.Ts(0),
-		sink, &dmlsink.TxnEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.TxnEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 
 	require.Equal(t, model.NewResolvedTs(0), tb.maxResolvedTs, "maxResolvedTs should start from 0")
 	require.NotNil(t, sink, tb.backendSink, "backendSink should be set")
@@ -183,7 +187,10 @@ func TestAppendRowChangedEvents(t *testing.T) {
 	sink := &mockEventSink{dead: make(chan struct{})}
 	tb := New[*model.SingleTableTxn](
 		model.DefaultChangeFeedID("1"), spanz.TableIDToComparableSpan(1), model.Ts(0),
-		sink, &dmlsink.TxnEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.TxnEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 
 	tb.AppendRowChangedEvents(getTestRows()...)
 	require.Len(t, tb.eventBuffer, 7, "txn event buffer should have 7 txns")
@@ -195,7 +202,10 @@ func TestUpdateResolvedTs(t *testing.T) {
 	sink := &mockEventSink{dead: make(chan struct{})}
 	tb := New[*model.SingleTableTxn](
 		model.DefaultChangeFeedID("1"), spanz.TableIDToComparableSpan(1), model.Ts(0),
-		sink, &dmlsink.TxnEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.TxnEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 
 	tb.AppendRowChangedEvents(getTestRows()...)
 	// No event will be flushed.
@@ -245,7 +255,10 @@ func TestGetCheckpointTs(t *testing.T) {
 	sink := &mockEventSink{dead: make(chan struct{})}
 	tb := New[*model.SingleTableTxn](
 		model.DefaultChangeFeedID("1"), spanz.TableIDToComparableSpan(1), model.Ts(0),
-		sink, &dmlsink.TxnEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.TxnEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 
 	tb.AppendRowChangedEvents(getTestRows()...)
 	require.Equal(t, model.NewResolvedTs(0), tb.GetCheckpointTs(), "checkpointTs should be 0")
@@ -287,7 +300,10 @@ func TestClose(t *testing.T) {
 	sink := &mockEventSink{dead: make(chan struct{})}
 	tb := New[*model.SingleTableTxn](
 		model.DefaultChangeFeedID("1"), spanz.TableIDToComparableSpan(1), model.Ts(0),
-		sink, &dmlsink.TxnEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.TxnEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 
 	tb.AppendRowChangedEvents(getTestRows()...)
 	err := tb.UpdateResolvedTs(model.NewResolvedTs(105))
@@ -316,7 +332,10 @@ func TestOperationsAfterClose(t *testing.T) {
 	sink := &mockEventSink{dead: make(chan struct{})}
 	tb := New[*model.SingleTableTxn](
 		model.DefaultChangeFeedID("1"), spanz.TableIDToComparableSpan(1), model.Ts(0),
-		sink, &dmlsink.TxnEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.TxnEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 
 	require.True(t, tb.AsyncClose())
 
@@ -331,7 +350,10 @@ func TestCloseCancellable(t *testing.T) {
 	sink := &mockEventSink{dead: make(chan struct{})}
 	tb := New[*model.SingleTableTxn](
 		model.DefaultChangeFeedID("1"), spanz.TableIDToComparableSpan(1), model.Ts(0),
-		sink, &dmlsink.TxnEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.TxnEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 
 	tb.AppendRowChangedEvents(getTestRows()...)
 	err := tb.UpdateResolvedTs(model.NewResolvedTs(105))
@@ -361,7 +383,10 @@ func TestCloseReentrant(t *testing.T) {
 	sink := &mockEventSink{dead: make(chan struct{})}
 	tb := New[*model.SingleTableTxn](
 		model.DefaultChangeFeedID("1"), spanz.TableIDToComparableSpan(1), model.Ts(0),
-		sink, &dmlsink.TxnEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.TxnEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 
 	tb.AppendRowChangedEvents(getTestRows()...)
 	err := tb.UpdateResolvedTs(model.NewResolvedTs(105))
@@ -394,7 +419,10 @@ func TestCheckpointTsFrozenWhenStopping(t *testing.T) {
 	sink := &mockEventSink{dead: make(chan struct{})}
 	tb := New[*model.SingleTableTxn](
 		model.DefaultChangeFeedID("1"), spanz.TableIDToComparableSpan(1), model.Ts(0),
-		sink, &dmlsink.TxnEventAppender{}, prometheus.NewCounter(prometheus.CounterOpts{}))
+		sink, &dmlsink.TxnEventAppender{},
+		pdutil.NewClock4Test(),
+		prometheus.NewCounter(prometheus.CounterOpts{}),
+		prometheus.NewHistogram(prometheus.HistogramOpts{}))
 
 	tb.AppendRowChangedEvents(getTestRows()...)
 	err := tb.UpdateResolvedTs(model.NewResolvedTs(105))
diff --git a/pkg/filter/sql_event_filter_test.go b/pkg/filter/sql_event_filter_test.go
index bbf4acd5b..d23310334 100644
--- a/pkg/filter/sql_event_filter_test.go
+++ b/pkg/filter/sql_event_filter_test.go
@@ -41,167 +41,253 @@ func TestShouldSkipDDL(t *testing.T) {
 		err   error
 	}
 
-	testCases := []testCase{
-		{
-			cfg: &config.FilterConfig{
-				EventFilters: []*config.EventFilterRule{
-					{
-						Matcher:     []string{"test.t1"},
-						IgnoreEvent: []bf.EventType{bf.AllDDL},
-					},
-				},
-			},
-			cases: []innerCase{
-				{
-					schema:  "test",
-					table:   "t1",
-					query:   "alter table t1 modify column age int",
-					ddlType: timodel.ActionModifyColumn,
-					skip:    true,
-				},
-				{
-					schema:  "test",
-					table:   "t1",
-					query:   "create table t1(id int primary key)",
-					ddlType: timodel.ActionCreateTable,
-					skip:    true,
-				},
+	// filter all ddl
+	case1 := testCase{
+		cfg: &config.FilterConfig{
+			EventFilters: []*config.EventFilterRule{
 				{
-					schema:  "test",
-					table:   "t2", // table name not match
-					query:   "alter table t2 modify column age int",
-					ddlType: timodel.ActionModifyColumn,
-					skip:    false,
-				},
-				{
-					schema:  "test2", // schema name not match
-					table:   "t1",
-					query:   "alter table t1 modify column age int",
-					ddlType: timodel.ActionModifyColumn,
-					skip:    false,
+					Matcher:     []string{"test.t1"},
+					IgnoreEvent: []bf.EventType{bf.AllDDL},
 				},
 			},
 		},
-		{
-			cfg: &config.FilterConfig{
-				EventFilters: []*config.EventFilterRule{
-					{
-						Matcher:     []string{"*.t1"},
-						IgnoreEvent: []bf.EventType{bf.DropDatabase, bf.DropSchema},
-					},
-				},
+		cases: []innerCase{
+			{
+				schema:  "test",
+				table:   "t1",
+				query:   "alter table t1 modify column age int",
+				ddlType: timodel.ActionModifyColumn,
+				skip:    true,
 			},
-			cases: []innerCase{
-				{
-					schema:  "test",
-					table:   "t1",
-					query:   "alter table t1 modify column age int",
-					ddlType: timodel.ActionModifyColumn,
-					skip:    false,
-				},
-				{
-					schema:  "test",
-					table:   "t1",
-					query:   "alter table t1 drop column age",
-					ddlType: timodel.ActionDropColumn,
-					skip:    false,
-				},
-				{
-					schema:  "test2",
-					table:   "t1",
-					query:   "drop database test2",
-					ddlType: timodel.ActionDropSchema,
-					skip:    true,
+			{
+				schema:  "test",
+				table:   "t1",
+				query:   "create table t1(id int primary key)",
+				ddlType: timodel.ActionCreateTable,
+				skip:    true,
+			},
+			{
+				schema:  "test",
+				table:   "t2", // table name not match
+				query:   "alter table t2 modify column age int",
+				ddlType: timodel.ActionModifyColumn,
+				skip:    false,
+			},
+			{
+				schema:  "test2", // schema name not match
+				table:   "t1",
+				query:   "alter table t1 modify column age int",
+				ddlType: timodel.ActionModifyColumn,
+				skip:    false,
+			},
+		},
+	}
+	f, err := newSQLEventFilter(case1.cfg)
+	require.True(t, errors.ErrorEqual(err, case1.err), "case: %+s", err)
+	for _, c := range case1.cases {
+		ddl := &model.DDLEvent{
+			TableInfo: &model.TableInfo{
+				TableName: model.TableName{
+					Schema: c.schema,
+					Table:  c.table,
 				},
+			},
+			Query: c.query,
+			Type:  c.ddlType,
+		}
+		skip, err := f.shouldSkipDDL(ddl)
+		require.NoError(t, err)
+		require.Equal(t, c.skip, skip, "case: %+v", c)
+	}
+
+	// filter some ddl
+	case2 := testCase{
+		cfg: &config.FilterConfig{
+			EventFilters: []*config.EventFilterRule{
 				{
-					schema:  "test3",
-					table:   "t1",
-					query:   "drop index i3 on t1",
-					ddlType: timodel.ActionDropIndex,
-					skip:    false,
+					Matcher:     []string{"*.t1"},
+					IgnoreEvent: []bf.EventType{bf.DropDatabase, bf.DropSchema},
 				},
 			},
 		},
-		{
-			cfg: &config.FilterConfig{
-				EventFilters: []*config.EventFilterRule{
-					{
-						Matcher:   []string{"*.t1"},
-						IgnoreSQL: []string{"MODIFY COLUMN", "DROP COLUMN", "^DROP DATABASE"},
-					},
-				},
+		cases: []innerCase{
+			{
+				schema:  "test",
+				table:   "t1",
+				query:   "alter table t1 modify column age int",
+				ddlType: timodel.ActionModifyColumn,
+				skip:    false,
 			},
-			cases: []innerCase{
-				{
-					schema:  "test",
-					table:   "t1",
-					query:   "ALTER TABLE t1 MODIFY COLUMN age int(11) NOT NULL",
-					ddlType: timodel.ActionModifyColumn,
-					skip:    true,
+			{
+				schema:  "test",
+				table:   "t1",
+				query:   "alter table t1 drop column age",
+				ddlType: timodel.ActionDropColumn,
+				skip:    false,
+			},
+			{
+				schema:  "test2",
+				table:   "t1",
+				query:   "drop database test2",
+				ddlType: timodel.ActionDropSchema,
+				skip:    true,
+			},
+			{
+				schema:  "test3",
+				table:   "t1",
+				query:   "drop index i3 on t1",
+				ddlType: timodel.ActionDropIndex,
+				skip:    false,
+			},
+		},
+	}
+	f, err = newSQLEventFilter(case2.cfg)
+	require.True(t, errors.ErrorEqual(err, case2.err), "case: %+s", err)
+	for _, c := range case2.cases {
+		ddl := &model.DDLEvent{
+			TableInfo: &model.TableInfo{
+				TableName: model.TableName{
+					Schema: c.schema,
+					Table:  c.table,
 				},
+			},
+			Query: c.query,
+			Type:  c.ddlType,
+		}
+		skip, err := f.shouldSkipDDL(ddl)
+		require.NoError(t, err)
+		require.Equal(t, c.skip, skip, "case: %+v", c)
+	}
+
+	// filter ddl by IgnoreSQL
+	case3 := testCase{
+		cfg: &config.FilterConfig{
+			EventFilters: []*config.EventFilterRule{
 				{
-					schema:  "test",
-					table:   "t1",
-					query:   "ALTER TABLE t1 DROP COLUMN age",
-					ddlType: timodel.ActionDropColumn,
-					skip:    true,
+					Matcher:   []string{"*.t1"},
+					IgnoreSQL: []string{"MODIFY COLUMN", "DROP COLUMN", "^DROP DATABASE"},
 				},
-				{ // no table name
-					schema:  "test2",
-					query:   "DROP DATABASE test",
-					ddlType: timodel.ActionDropSchema,
-					skip:    true,
+			},
+		},
+		cases: []innerCase{
+			{
+				schema:  "test",
+				table:   "t1",
+				query:   "ALTER TABLE t1 MODIFY COLUMN age int(11) NOT NULL",
+				ddlType: timodel.ActionModifyColumn,
+				skip:    true,
+			},
+			{
+				schema:  "test",
+				table:   "t1",
+				query:   "ALTER TABLE t1 DROP COLUMN age",
+				ddlType: timodel.ActionDropColumn,
+				skip:    true,
+			},
+			{ // no table name
+				schema:  "test2",
+				query:   "DROP DATABASE test",
+				ddlType: timodel.ActionDropSchema,
+				skip:    true,
+			},
+			{
+				schema:  "test3",
+				table:   "t1",
+				query:   "Drop Index i1 on test3.t1",
+				ddlType: timodel.ActionDropIndex,
+				skip:    false,
+			},
+		},
+	}
+	f, err = newSQLEventFilter(case3.cfg)
+	require.True(t, errors.ErrorEqual(err, case3.err), "case: %+s", err)
+	for _, c := range case3.cases {
+		ddl := &model.DDLEvent{
+			TableInfo: &model.TableInfo{
+				TableName: model.TableName{
+					Schema: c.schema,
+					Table:  c.table,
 				},
+			},
+			Query: c.query,
+			Type:  c.ddlType,
+		}
+		skip, err := f.shouldSkipDDL(ddl)
+		require.NoError(t, err)
+		require.Equal(t, c.skip, skip, "case: %+v", c)
+	}
+
+	// config error
+	case4 := testCase{
+		cfg: &config.FilterConfig{
+			EventFilters: []*config.EventFilterRule{
 				{
-					schema:  "test3",
-					table:   "t1",
-					query:   "Drop Index i1 on test3.t1",
-					ddlType: timodel.ActionDropIndex,
-					skip:    false,
+					Matcher:     []string{"*.t1"},
+					IgnoreEvent: []bf.EventType{bf.EventType("aa")},
 				},
 			},
 		},
-		{ // config error
-			cfg: &config.FilterConfig{
-				EventFilters: []*config.EventFilterRule{
-					{
-						Matcher:     []string{"*.t1"},
-						IgnoreEvent: []bf.EventType{bf.EventType("aa")},
-					},
+		err: cerror.ErrInvalidIgnoreEventType,
+	}
+	_, err = newSQLEventFilter(case4.cfg)
+	require.True(t, errors.ErrorEqual(err, case4.err), "case: %+s", err)
+
+	// config error
+	case5 := testCase{
+		cfg: &config.FilterConfig{
+			EventFilters: []*config.EventFilterRule{
+				{
+					Matcher:   []string{"*.t1"},
+					IgnoreSQL: []string{"--6"}, // this is a valid regx
 				},
 			},
-			err: cerror.ErrInvalidIgnoreEventType,
 		},
-		{
-			cfg: &config.FilterConfig{
-				EventFilters: []*config.EventFilterRule{
-					{
-						Matcher:   []string{"*.t1"},
-						IgnoreSQL: []string{"--6"}, // this is a valid regx
-					},
+	}
+	_, err = newSQLEventFilter(case5.cfg)
+	require.True(t, errors.ErrorEqual(err, case5.err), "case: %+s", err)
+
+	// cover all ddl event types
+	allEventTypes := make([]bf.EventType, 0, len(ddlWhiteListMap))
+	for _, et := range ddlWhiteListMap {
+		allEventTypes = append(allEventTypes, et)
+	}
+	innerCases := make([]innerCase, 0, len(ddlWhiteListMap))
+	for at := range ddlWhiteListMap {
+		innerCases = append(innerCases, innerCase{
+			schema:  "test",
+			table:   "t1",
+			query:   "no matter",
+			ddlType: at,
+			skip:    true,
+		})
+	}
+	case6 := testCase{
+		cfg: &config.FilterConfig{
+			EventFilters: []*config.EventFilterRule{
+				{
+					Matcher:     []string{"*.t1"},
+					IgnoreEvent: allEventTypes,
 				},
 			},
 		},
+		cases: innerCases,
 	}
-
-	for _, tc := range testCases {
-		f, err := newSQLEventFilter(tc.cfg)
-		require.True(t, errors.ErrorEqual(err, tc.err), "case: %+s", err)
-		for _, c := range tc.cases {
-			ddl := &model.DDLEvent{
-				TableInfo: &model.TableInfo{
-					TableName: model.TableName{
-						Schema: c.schema,
-						Table:  c.table,
-					},
+	f, err = newSQLEventFilter(case6.cfg)
+	require.NoError(t, err)
+	for _, c := range case6.cases {
+		ddl := &model.DDLEvent{
+			TableInfo: &model.TableInfo{
+				TableName: model.TableName{
+					Schema: c.schema,
+					Table:  c.table,
 				},
-				Query: c.query,
-				Type:  c.ddlType,
-			}
-			skip, err := f.shouldSkipDDL(ddl)
-			require.NoError(t, err)
-			require.Equal(t, c.skip, skip, "case: %+v", c)
+			},
+			Query: c.query,
+			Type:  c.ddlType,
 		}
+		skip, err := f.shouldSkipDDL(ddl)
+		require.NoError(t, err)
+		require.Equal(t, c.skip, skip, "case: %+v", c)
 	}
 }
 
diff --git a/pkg/sink/codec/bootstraper_test.go b/pkg/sink/codec/bootstraper_test.go
index 5a1b2eb8a..1888cffb2 100644
--- a/pkg/sink/codec/bootstraper_test.go
+++ b/pkg/sink/codec/bootstraper_test.go
@@ -20,6 +20,7 @@ import (
 	"time"
 
 	timodel "github.com/pingcap/tidb/pkg/parser/model"
+	"github.com/pingcap/tiflow/cdc/entry"
 	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/stretchr/testify/require"
@@ -50,7 +51,7 @@ func getMockTableStatus(tableName string,
 		PhysicalTableID: tableID,
 		TableInfo:       tableInfo,
 	}
-	tb := newTableStatus(key, row)
+	tb := newTableStatistic(key, row)
 	return key, row, tb
 }
 
@@ -176,3 +177,47 @@ l2:
 	// The bootstrap events are sent to all partition
 	require.Equal(t, key2.TotalPartition, msgCount)
 }
+
+func TestUpdateTableStatistic(t *testing.T) {
+	helper := entry.NewSchemaTestHelper(t)
+	defer helper.Close()
+
+	sql := `create table test.t1(
+		id int primary key,
+		name varchar(64) not null,
+		age int,
+		email varchar(255) not null,
+		unique index idx_name(name),
+		index idx_age_email(age,email)
+	);`
+	tableInfo1 := helper.DDL2Event(sql).TableInfo
+	row1 := &model.RowChangedEvent{
+		PhysicalTableID: tableInfo1.ID,
+		TableInfo:       tableInfo1,
+	}
+	tableStatistic := newTableStatistic(model.TopicPartitionKey{}, row1)
+
+	// case 1: The tableStatistic should not be updated if the tableInfo is the same
+	tableStatistic.update(row1, 1)
+	require.Equal(t, tableInfo1, tableStatistic.tableInfo.Load().(*model.TableInfo))
+
+	// case 2: The tableStatistic should be updated if the tableInfo is different
+	sql = `alter table test.t1 add column address varchar(255) not null;`
+	tableInfo2 := helper.DDL2Event(sql).TableInfo
+	row2 := &model.RowChangedEvent{
+		PhysicalTableID: tableInfo2.ID,
+		TableInfo:       tableInfo2,
+	}
+	tableStatistic.update(row2, 1)
+	require.Equal(t, tableInfo2, tableStatistic.tableInfo.Load().(*model.TableInfo))
+
+	// case 3: The tableStatistic should be updated when rename table
+	sql = `alter table test.t1 rename to test.t2;`
+	tableInfo3 := helper.DDL2Event(sql).TableInfo
+	row3 := &model.RowChangedEvent{
+		PhysicalTableID: tableInfo3.ID,
+		TableInfo:       tableInfo3,
+	}
+	tableStatistic.update(row3, 1)
+	require.Equal(t, tableInfo3, tableStatistic.tableInfo.Load().(*model.TableInfo))
+}
diff --git a/pkg/sink/codec/utils/test_utils.go b/pkg/sink/codec/utils/test_utils.go
index f8e06235a..649400f48 100644
--- a/pkg/sink/codec/utils/test_utils.go
+++ b/pkg/sink/codec/utils/test_utils.go
@@ -74,6 +74,7 @@ func NewLargeEvent4Test(t *testing.T, replicaConfig *config.ReplicaConfig) (*mod
 	 	dateT date default '2023-12-27',
 	 	datetimeT datetime default '2023-12-27 12:27:23',
 	 	timestampT timestamp default now(),
+	 	timestampT2 timestamp default '2024-03-01 13:40:39.662659',
 	 	timeT time default '12:27:23',
 	 	yearT year default 2023,
 	 	enumT enum('a', 'b', 'c') default 'b',
@@ -130,6 +131,7 @@ func NewLargeEvent4Test(t *testing.T, replicaConfig *config.ReplicaConfig) (*mod
 		'2020-02-20',
 		'2020-02-20 02:20:20',
 		'2020-02-20 10:20:20',
+	    '2024-03-01 13:40:39.662659',
 		'02:20:20',
 		2020,
 		'a',
diff --git a/tests/integration_tests/mq_sink_dispatcher/conf/changefeed.toml b/tests/integration_tests/mq_sink_dispatcher/conf/changefeed.toml
index 849681cee..8ee525511 100644
--- a/tests/integration_tests/mq_sink_dispatcher/conf/changefeed.toml
+++ b/tests/integration_tests/mq_sink_dispatcher/conf/changefeed.toml
@@ -1,4 +1,5 @@
 [sink]
 dispatchers = [
+    {matcher = ['verify.t'], partition = "index-value"},
     {matcher = ['dispatcher.index'], partition = "index-value", index = "idx_a"}
 ]
diff --git a/tests/integration_tests/mq_sink_dispatcher/run.sh b/tests/integration_tests/mq_sink_dispatcher/run.sh
index ae28b63c4..6c234a6e5 100644
--- a/tests/integration_tests/mq_sink_dispatcher/run.sh
+++ b/tests/integration_tests/mq_sink_dispatcher/run.sh
@@ -26,6 +26,12 @@ function run() {
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY --loglevel "info"
 
 	changefeed_id="test"
+
+	# create a table before get pd tso to reproduce https://github.com/pingcap/tiflow/issues/10707
+	run_sql "DROP DATABASE if exists verify;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "CREATE DATABASE verify;" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "CREATE TABLE verify.t (a int primary key, b int);" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
 	# record tso before we create tables to skip the system table DDLs
 	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
 	SINK_URI="kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=canal-json&enable-tidb-extension=true"