Repository: pingcap/tiflow
Pull request #10369 (2023-12-27 05:24:59)

conflict detector(cdc): fix potential dead lock when a transaction generated multiple same hash (#10335)
This is an automated cherry-pick of #10335

<!--
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/10334

### What is changed and how it works?

- Deduplicate the generated hash in a single transaction to prevent self cyclic dependency in the conflict detector.
- Refine the conflict detector, make it more readable


### 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
Fix sink-to-mysql checkpoint-ts stuck issue in some rare cases.
```


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

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

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

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

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

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

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


<details>

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

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

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

<details>

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

Patch:
diff --git a/cdc/api/v2/api.go b/cdc/api/v2/api.go
index 09a2849cd..a6b8b48c4 100644
--- a/cdc/api/v2/api.go
+++ b/cdc/api/v2/api.go
@@ -63,6 +63,7 @@ func RegisterOpenAPIV2Routes(router *gin.Engine, api OpenAPIV2) {
 	changefeedGroup.POST("/:changefeed_id/resume", changefeedOwnerMiddleware, api.resumeChangefeed)
 	changefeedGroup.POST("/:changefeed_id/pause", changefeedOwnerMiddleware, api.pauseChangefeed)
 	changefeedGroup.GET("/:changefeed_id/status", changefeedOwnerMiddleware, api.status)
+	changefeedGroup.GET("/:changefeed_id/synced", changefeedOwnerMiddleware, api.synced)
 
 	// capture apis
 	captureGroup := v2.Group("/captures")
diff --git a/cdc/api/v2/changefeed.go b/cdc/api/v2/changefeed.go
index 1cc070d76..120675726 100644
--- a/cdc/api/v2/changefeed.go
+++ b/cdc/api/v2/changefeed.go
@@ -48,6 +48,8 @@ const (
 	apiOpVarChangefeedID = "changefeed_id"
 	// apiOpVarNamespace is the key of changefeed namespace in HTTP API
 	apiOpVarNamespace = "namespace"
+	// timeout for pd client
+	timeout = 30 * time.Second
 )
 
 // createChangefeed handles create changefeed request,
@@ -889,6 +891,142 @@ func (h *OpenAPIV2) status(c *gin.Context) {
 	})
 }
 
+// synced get the synced status of a changefeed
+// @Summary Get synced status
+// @Description get the synced status of a changefeed
+// @Tags changefeed,v2
+// @Accept json
+// @Produce json
+// @Param changefeed_id  path  string  true  "changefeed_id"
+// @Param namespace query string false "default"
+// @Success 200 {object} SyncedStatus
+// @Failure 500,400 {object} model.HTTPError
+// @Router /api/v2/changefeeds/{changefeed_id}/synced [get]
+func (h *OpenAPIV2) synced(c *gin.Context) {
+	ctx := c.Request.Context()
+
+	namespace := getNamespaceValueWithDefault(c)
+	changefeedID := model.ChangeFeedID{Namespace: namespace, ID: c.Param(apiOpVarChangefeedID)}
+	if err := model.ValidateChangefeedID(changefeedID.ID); err != nil {
+		_ = c.Error(cerror.ErrAPIInvalidParam.GenWithStack("invalid changefeed_id: %s",
+			changefeedID.ID))
+		return
+	}
+
+	status, err := h.capture.StatusProvider().GetChangeFeedSyncedStatus(ctx, changefeedID)
+	if err != nil {
+		_ = c.Error(err)
+		return
+	}
+
+	log.Info("Get changefeed synced status:", zap.Any("status", status), zap.Any("changefeedID", changefeedID))
+
+	cfg := &ChangefeedConfig{ReplicaConfig: GetDefaultReplicaConfig()}
+	if (status.SyncedCheckInterval != 0) && (status.CheckpointInterval != 0) {
+		cfg.ReplicaConfig.SyncedStatus.CheckpointInterval = status.CheckpointInterval
+		cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval = status.SyncedCheckInterval
+	}
+
+	// try to get pd client to get pd time, and determine synced status based on the pd time
+	if len(cfg.PDAddrs) == 0 {
+		up, err := getCaptureDefaultUpstream(h.capture)
+		if err != nil {
+			_ = c.Error(err)
+			return
+		}
+		cfg.PDConfig = getUpstreamPDConfig(up)
+	}
+	credential := cfg.PDConfig.toCredential()
+
+	timeoutCtx, cancel := context.WithTimeout(ctx, timeout)
+	defer cancel()
+
+	pdClient, err := h.helpers.getPDClient(timeoutCtx, cfg.PDAddrs, credential)
+	if err != nil {
+		// case 1. we can't get pd client, pd may be unavailable.
+		//         if pullerResolvedTs - checkpointTs > checkpointInterval, data is not synced
+		//         otherwise, if pd is unavailable, we decide data whether is synced based on
+		//         the time difference between current time and lastSyncedTs.
+		var message string
+		if (oracle.ExtractPhysical(status.PullerResolvedTs) - oracle.ExtractPhysical(status.CheckpointTs)) >
+			cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000 {
+			message = fmt.Sprintf("%s. Besides the data is not finish syncing", err.Error())
+		} else {
+			message = fmt.Sprintf("%s. You should check the pd status first. If pd status is normal, means we don't finish sync data. "+
+				"If pd is offline, please check whether we satisfy the condition that "+
+				"the time difference from lastSyncedTs to the current time from the time zone of pd is greater than %v secs. "+
+				"If it's satisfied, means the data syncing is totally finished", err, cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval)
+		}
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           false,
+			SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
+			PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
+			LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
+			NowTs:            model.JSONTime(time.Unix(0, 0)),
+			Info:             message,
+		})
+		return
+	}
+	defer pdClient.Close()
+	// get time from pd
+	physicalNow, _, _ := pdClient.GetTS(ctx)
+
+	// We can normally get pd time. Thus we determine synced status based on physicalNow, lastSyncedTs, checkpointTs and pullerResolvedTs
+	if (physicalNow-oracle.ExtractPhysical(status.LastSyncedTs) > cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval*1000) &&
+		(physicalNow-oracle.ExtractPhysical(status.CheckpointTs) < cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000) {
+		// case 2: If physcialNow - lastSyncedTs > SyncedCheckInterval && physcialNow - CheckpointTs < CheckpointInterval
+		//         --> reach strict synced status
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           true,
+			SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
+			PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
+			LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
+			NowTs:            model.JSONTime(time.Unix(physicalNow/1e3, 0)),
+			Info:             "Data syncing is finished",
+		})
+		return
+	}
+
+	if physicalNow-oracle.ExtractPhysical(status.LastSyncedTs) > cfg.ReplicaConfig.SyncedStatus.SyncedCheckInterval*1000 {
+		// case 3: If physcialNow - lastSyncedTs > SyncedCheckInterval && physcialNow - CheckpointTs > CheckpointInterval
+		//         we should consider the situation that pd or tikv region is not healthy to block the advancing resolveTs.
+		//         if pullerResolvedTs - checkpointTs > CheckpointInterval-->  data is not synced
+		//         otherwise, if pd & tikv is healthy --> data is not synced
+		//                    if not healthy --> data is synced
+		var message string
+		if (oracle.ExtractPhysical(status.PullerResolvedTs) - oracle.ExtractPhysical(status.CheckpointTs)) <
+			cfg.ReplicaConfig.SyncedStatus.CheckpointInterval*1000 {
+			message = fmt.Sprintf("Please check whether PD is online and TiKV Regions are all available. " +
+				"If PD is offline or some TiKV regions are not available, it means that the data syncing process is complete. " +
+				"To check whether TiKV regions are all available, " +
+				"you can view 'TiKV-Details' > 'Resolved-Ts' > 'Max Leader Resolved TS gap' on Grafana. " +
+				"If the gap is large, such as a few minutes, it means that some regions in TiKV are unavailable. " +
+				"Otherwise, if the gap is small and PD is online, it means the data syncing is incomplete, so please wait")
+		} else {
+			message = "The data syncing is not finished, please wait"
+		}
+		c.JSON(http.StatusOK, SyncedStatus{
+			Synced:           false,
+			SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
+			PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
+			LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
+			NowTs:            model.JSONTime(time.Unix(physicalNow/1e3, 0)),
+			Info:             message,
+		})
+		return
+	}
+
+	// case	4: If physcialNow - lastSyncedTs < SyncedCheckInterval --> data is not synced
+	c.JSON(http.StatusOK, SyncedStatus{
+		Synced:           false,
+		SinkCheckpointTs: model.JSONTime(oracle.GetTimeFromTS(status.CheckpointTs)),
+		PullerResolvedTs: model.JSONTime(oracle.GetTimeFromTS(status.PullerResolvedTs)),
+		LastSyncedTs:     model.JSONTime(oracle.GetTimeFromTS(status.LastSyncedTs)),
+		NowTs:            model.JSONTime(time.Unix(physicalNow/1e3, 0)),
+		Info:             "The data syncing is not finished, please wait",
+	})
+}
+
 func toAPIModel(
 	info *model.ChangeFeedInfo,
 	resolvedTs uint64,
diff --git a/cdc/api/v2/model.go b/cdc/api/v2/model.go
index 0564fb0d1..efd17488b 100644
--- a/cdc/api/v2/model.go
+++ b/cdc/api/v2/model.go
@@ -103,6 +103,15 @@ type ChangefeedCommonInfo struct {
 	RunningError   *model.RunningError `json:"error"`
 }
 
+// SyncedStatusConfig represents synced check interval config for a changefeed
+type SyncedStatusConfig struct {
+	// The minimum interval between the latest synced ts and now required to reach synced state
+	SyncedCheckInterval int64 `json:"synced_check_interval"`
+	// The maximum interval between latest checkpoint ts and now or
+	// between latest sink's checkpoint ts and puller's checkpoint ts required to reach synced state
+	CheckpointInterval int64 `json:"checkpoint_interval"`
+}
+
 // MarshalJSON marshal changefeed common info to json
 // we need to set feed state to normal if it is uninitialized and pending to warning
 // to hide the detail of uninitialized and pending state from user
@@ -195,6 +204,7 @@ type ReplicaConfig struct {
 	Integrity                    *IntegrityConfig           `json:"integrity"`
 	ChangefeedErrorStuckDuration *JSONDuration              `json:"changefeed_error_stuck_duration,omitempty"`
 	SQLMode                      string                     `json:"sql_mode,omitempty"`
+	SyncedStatus                 *SyncedStatusConfig        `json:"synced_status,omitempty"`
 }
 
 // ToInternalReplicaConfig coverts *v2.ReplicaConfig into *config.ReplicaConfig
@@ -273,6 +283,12 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 			Compression:           c.Consistent.Compression,
 			FlushConcurrency:      c.Consistent.FlushConcurrency,
 		}
+		if c.Consistent.MemoryUsage != nil {
+			res.Consistent.MemoryUsage = &config.ConsistentMemoryUsage{
+				MemoryQuotaPercentage: c.Consistent.MemoryUsage.MemoryQuotaPercentage,
+				EventCachePercentage:  c.Consistent.MemoryUsage.EventCachePercentage,
+			}
+		}
 	}
 	if c.Sink != nil {
 		var dispatchRules []*config.DispatchRule
@@ -489,6 +505,12 @@ func (c *ReplicaConfig) toInternalReplicaConfigWithOriginConfig(
 	if c.ChangefeedErrorStuckDuration != nil {
 		res.ChangefeedErrorStuckDuration = &c.ChangefeedErrorStuckDuration.duration
 	}
+	if c.SyncedStatus != nil {
+		res.SyncedStatus = &config.SyncedStatusConfig{
+			SyncedCheckInterval: c.SyncedStatus.SyncedCheckInterval,
+			CheckpointInterval:  c.SyncedStatus.CheckpointInterval,
+		}
+	}
 	return res
 }
 
@@ -760,7 +782,14 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 			Compression:           cloned.Consistent.Compression,
 			FlushConcurrency:      cloned.Consistent.FlushConcurrency,
 		}
+		if cloned.Consistent.MemoryUsage != nil {
+			res.Consistent.MemoryUsage = &ConsistentMemoryUsage{
+				MemoryQuotaPercentage: cloned.Consistent.MemoryUsage.MemoryQuotaPercentage,
+				EventCachePercentage:  cloned.Consistent.MemoryUsage.EventCachePercentage,
+			}
+		}
 	}
+
 	if cloned.Mounter != nil {
 		res.Mounter = &MounterConfig{
 			WorkerNum: cloned.Mounter.WorkerNum,
@@ -783,6 +812,12 @@ func ToAPIReplicaConfig(c *config.ReplicaConfig) *ReplicaConfig {
 	if cloned.ChangefeedErrorStuckDuration != nil {
 		res.ChangefeedErrorStuckDuration = &JSONDuration{*cloned.ChangefeedErrorStuckDuration}
 	}
+	if cloned.SyncedStatus != nil {
+		res.SyncedStatus = &SyncedStatusConfig{
+			SyncedCheckInterval: cloned.SyncedStatus.SyncedCheckInterval,
+			CheckpointInterval:  cloned.SyncedStatus.CheckpointInterval,
+		}
+	}
 	return res
 }
 
@@ -956,6 +991,14 @@ type ConsistentConfig struct {
 	UseFileBackend        bool   `json:"use_file_backend"`
 	Compression           string `json:"compression,omitempty"`
 	FlushConcurrency      int    `json:"flush_concurrency,omitempty"`
+
+	MemoryUsage *ConsistentMemoryUsage `json:"memory_usage"`
+}
+
+// ConsistentMemoryUsage represents memory usage of Consistent module.
+type ConsistentMemoryUsage struct {
+	MemoryQuotaPercentage uint64 `json:"memory_quota_percentage"`
+	EventCachePercentage  uint64 `json:"event_cache_percentage"`
 }
 
 // ChangefeedSchedulerConfig is per changefeed scheduler settings.
@@ -1014,6 +1057,16 @@ type ChangeFeedInfo struct {
 	TaskStatus     []model.CaptureTaskStatus `json:"task_status,omitempty"`
 }
 
+// SyncedStatus describes the detail of a changefeed's synced status
+type SyncedStatus struct {
+	Synced           bool           `json:"synced"`
+	SinkCheckpointTs model.JSONTime `json:"sink_checkpoint_ts"`
+	PullerResolvedTs model.JSONTime `json:"puller_resolved_ts"`
+	LastSyncedTs     model.JSONTime `json:"last_synced_ts"`
+	NowTs            model.JSONTime `json:"now_ts"`
+	Info             string         `json:"info"`
+}
+
 // RunningError represents some running error from cdc components,
 // such as processor.
 type RunningError struct {
diff --git a/cdc/entry/mounter.go b/cdc/entry/mounter.go
index 57316aae6..8304bbdeb 100644
--- a/cdc/entry/mounter.go
+++ b/cdc/entry/mounter.go
@@ -16,6 +16,7 @@ package entry
 import (
 	"bytes"
 	"context"
+	"encoding/hex"
 	"encoding/json"
 	"fmt"
 	"math"
@@ -172,10 +173,18 @@ func (m *mounter) unmarshalAndMountRowChanged(ctx context.Context, raw *model.Ra
 		}
 		tableInfo, exist := snap.PhysicalTableByID(physicalTableID)
 		if !exist {
+			// for truncate table and truncate table partition DDL, the table ID is changed, but DML can be inserted to TiKV with old table ID.
+			// normally, cdc will close the old table pipeline and create a new one, and these invalid DMLs keys will not be pulled by CDC,
+			// but if redo is enabled or push based table pipeline is enabled, puller and mounter are not blocked by barrier ts.
+			// So some invalid DML keys will be decoded before processor removing the table pipeline
 			if snap.IsTruncateTableID(physicalTableID) {
 				log.Debug("skip the DML of truncated table", zap.Uint64("ts", raw.CRTs), zap.Int64("tableID", physicalTableID))
 				return nil, nil
 			}
+			log.Error("can not found table schema",
+				zap.Uint64("ts", raw.CRTs),
+				zap.String("key", hex.EncodeToString(raw.Key)),
+				zap.Int64("tableID", physicalTableID))
 			return nil, cerror.ErrSnapshotTableNotFound.GenWithStackByArgs(physicalTableID)
 		}
 		if bytes.HasPrefix(key, recordPrefix) {
diff --git a/cdc/entry/schema/snapshot.go b/cdc/entry/schema/snapshot.go
index 5367a038e..46416437d 100644
--- a/cdc/entry/schema/snapshot.go
+++ b/cdc/entry/schema/snapshot.go
@@ -468,11 +468,16 @@ func (s *Snapshot) DoHandleDDL(job *timodel.Job) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
-	case timodel.ActionTruncateTablePartition,
+	case timodel.ActionTruncateTablePartition:
+		err := s.inner.updatePartition(getWrapTableInfo(job), true, job.BinlogInfo.FinishedTS)
+		if err != nil {
+			return errors.Trace(err)
+		}
+	case
 		timodel.ActionAddTablePartition,
 		timodel.ActionDropTablePartition,
 		timodel.ActionReorganizePartition:
-		err := s.inner.updatePartition(getWrapTableInfo(job), job.BinlogInfo.FinishedTS)
+		err := s.inner.updatePartition(getWrapTableInfo(job), false, job.BinlogInfo.FinishedTS)
 		if err != nil {
 			return errors.Trace(err)
 		}
@@ -863,7 +868,7 @@ func (s *snapshot) doCreateTable(tbInfo *model.TableInfo, currentTs uint64) {
 }
 
 // updatePartition updates partition info for `tbInfo`.
-func (s *snapshot) updatePartition(tbInfo *model.TableInfo, currentTs uint64) error {
+func (s *snapshot) updatePartition(tbInfo *model.TableInfo, isTruncate bool, currentTs uint64) error {
 	oldTbInfo, ok := s.physicalTableByID(tbInfo.ID)
 	if !ok {
 		return cerror.ErrSnapshotTableNotFound.GenWithStackByArgs(tbInfo.ID)
@@ -888,6 +893,7 @@ func (s *snapshot) updatePartition(tbInfo *model.TableInfo, currentTs uint64) er
 	for _, partition := range oldPi.Definitions {
 		s.partitions.ReplaceOrInsert(newVersionedID(partition.ID, tag))
 	}
+	newPartitionIDMap := make(map[int64]struct{}, len(newPi.NewPartitionIDs))
 	for _, partition := range newPi.Definitions {
 		vid := newVersionedID(partition.ID, tag)
 		vid.target = tbInfo
@@ -895,6 +901,14 @@ func (s *snapshot) updatePartition(tbInfo *model.TableInfo, currentTs uint64) er
 		if ineligible {
 			s.ineligibleTables.ReplaceOrInsert(newVersionedID(partition.ID, tag))
 		}
+		newPartitionIDMap[partition.ID] = struct{}{}
+	}
+	if isTruncate {
+		for _, partition := range oldPi.Definitions {
+			if _, ok := newPartitionIDMap[partition.ID]; !ok {
+				s.truncatedTables.ReplaceOrInsert(newVersionedID(partition.ID, tag))
+			}
+		}
 	}
 	s.currentTs = currentTs
 
@@ -984,7 +998,7 @@ func (s *snapshot) exchangePartition(targetTable *model.TableInfo, currentTS uin
 	// ref: https://github.com/pingcap/tidb/issues/43819
 	targetTable.SchemaID = oldTable.SchemaID
 	targetTable.TableName = oldTable.TableName
-	err = s.updatePartition(targetTable, currentTS)
+	err = s.updatePartition(targetTable, false, currentTS)
 	if err != nil {
 		return errors.Trace(err)
 	}
diff --git a/cdc/model/changefeed.go b/cdc/model/changefeed.go
index c9d62df19..8df08f7e0 100644
--- a/cdc/model/changefeed.go
+++ b/cdc/model/changefeed.go
@@ -337,6 +337,9 @@ func (info *ChangeFeedInfo) VerifyAndComplete() {
 	if info.Config.SQLMode == "" {
 		info.Config.SQLMode = defaultConfig.SQLMode
 	}
+	if info.Config.SyncedStatus == nil {
+		info.Config.SyncedStatus = defaultConfig.SyncedStatus
+	}
 	info.RmUnusedFields()
 }
 
@@ -628,3 +631,12 @@ type ChangeFeedStatusForAPI struct {
 	ResolvedTs   uint64 `json:"resolved-ts"`
 	CheckpointTs uint64 `json:"checkpoint-ts"`
 }
+
+// ChangeFeedSyncedStatusForAPI uses to transfer the synced status of changefeed for API.
+type ChangeFeedSyncedStatusForAPI struct {
+	CheckpointTs        uint64 `json:"checkpoint-ts"`
+	LastSyncedTs        uint64 `json:"last-sync-time"`
+	PullerResolvedTs    uint64 `json:"puller-resolved-ts"`
+	SyncedCheckInterval int64  `json:"synced-check-interval"`
+	CheckpointInterval  int64  `json:"checkpoint-interval"`
+}
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
old mode 100644
new mode 100755
index 35cfd6ba6..4fee2e802
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -16,6 +16,7 @@ package owner
 import (
 	"context"
 	"fmt"
+	"math"
 	"strings"
 	"sync"
 	"time"
@@ -34,6 +35,7 @@ import (
 	cdcContext "github.com/pingcap/tiflow/pkg/context"
 	cerror "github.com/pingcap/tiflow/pkg/errors"
 	"github.com/pingcap/tiflow/pkg/filter"
+	pfilter "github.com/pingcap/tiflow/pkg/filter"
 	"github.com/pingcap/tiflow/pkg/orchestrator"
 	"github.com/pingcap/tiflow/pkg/pdutil"
 	redoCfg "github.com/pingcap/tiflow/pkg/redo"
@@ -76,6 +78,13 @@ type changefeed struct {
 	feedStateManager *feedStateManager
 	resolvedTs       model.Ts
 
+	// lastSyncedTs is the lastest resolvedTs that has been synced to downstream.
+	// pullerResolvedTs is the minimum resolvedTs of all pullers.
+	// we don't need to initialize lastSyncedTs and pullerResolvedTs specially
+	// because it will be updated in tick.
+	lastSyncedTs     model.Ts
+	pullerResolvedTs model.Ts
+
 	// ddl related fields
 	ddlManager  *ddlManager
 	redoDDLMgr  redo.DDLManager
@@ -123,7 +132,7 @@ type changefeed struct {
 		startTs uint64,
 		changefeed model.ChangeFeedID,
 		schemaStorage entry.SchemaStorage,
-		filter filter.Filter,
+		filter pfilter.Filter,
 	) (puller.DDLPuller, error)
 
 	newSink func(
@@ -181,7 +190,7 @@ func newChangefeed4Test(
 		startTs uint64,
 		changefeed model.ChangeFeedID,
 		schemaStorage entry.SchemaStorage,
-		filter filter.Filter,
+		filter pfilter.Filter,
 	) (puller.DDLPuller, error),
 	newSink func(
 		changefeedID model.ChangeFeedID, info *model.ChangeFeedInfo,
@@ -372,19 +381,36 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 		return nil
 	}
 
-	newCheckpointTs, newResolvedTs, err := c.scheduler.Tick(
+	watermark, err := c.scheduler.Tick(
 		ctx, preCheckpointTs, allPhysicalTables, captures,
 		barrier)
 	if err != nil {
 		return errors.Trace(err)
 	}
+	if c.lastSyncedTs < watermark.LastSyncedTs {
+		c.lastSyncedTs = watermark.LastSyncedTs
+	} else if c.lastSyncedTs > watermark.LastSyncedTs {
+		log.Warn("LastSyncedTs should not be greater than newLastSyncedTs",
+			zap.Uint64("c.LastSyncedTs", c.lastSyncedTs),
+			zap.Uint64("newLastSyncedTs", watermark.LastSyncedTs))
+	}
+
+	if watermark.PullerResolvedTs != scheduler.CheckpointCannotProceed && watermark.PullerResolvedTs != math.MaxUint64 {
+		if watermark.PullerResolvedTs > c.pullerResolvedTs {
+			c.pullerResolvedTs = watermark.PullerResolvedTs
+		} else if watermark.PullerResolvedTs < c.pullerResolvedTs {
+			log.Warn("the newPullerResolvedTs should not be smaller than c.pullerResolvedTs",
+				zap.Uint64("c.pullerResolvedTs", c.pullerResolvedTs),
+				zap.Uint64("newPullerResolvedTs", watermark.PullerResolvedTs))
+		}
+	}
 
 	pdTime := c.upstream.PDClock.CurrentTime()
 	currentTs := oracle.GetPhysical(pdTime)
 
 	// CheckpointCannotProceed implies that not all tables are being replicated normally,
 	// so in that case there is no need to advance the global watermarks.
-	if newCheckpointTs == scheduler.CheckpointCannotProceed {
+	if watermark.CheckpointTs == scheduler.CheckpointCannotProceed {
 		if c.state.Status != nil {
 			// We should keep the metrics updated even if the scheduler cannot
 			// advance the watermarks for now.
@@ -395,13 +421,13 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 
 	log.Debug("owner prepares to update status",
 		zap.Uint64("prevResolvedTs", c.resolvedTs),
-		zap.Uint64("newResolvedTs", newResolvedTs),
-		zap.Uint64("newCheckpointTs", newCheckpointTs),
+		zap.Uint64("newResolvedTs", watermark.ResolvedTs),
+		zap.Uint64("newCheckpointTs", watermark.CheckpointTs),
 		zap.String("namespace", c.id.Namespace),
 		zap.String("changefeed", c.id.ID))
 	// resolvedTs should never regress.
-	if newResolvedTs > c.resolvedTs {
-		c.resolvedTs = newResolvedTs
+	if watermark.ResolvedTs > c.resolvedTs {
+		c.resolvedTs = watermark.ResolvedTs
 	}
 
 	// MinTableBarrierTs should never regress
@@ -415,13 +441,17 @@ func (c *changefeed) tick(ctx cdcContext.Context, captures map[model.CaptureID]*
 				zap.String("namespace", c.id.Namespace),
 				zap.String("changefeed", c.id.ID),
 				zap.Uint64("keepCheckpoint", c.state.Status.CheckpointTs),
-				zap.Uint64("skipCheckpoint", newCheckpointTs))
-			newCheckpointTs = c.state.Status.CheckpointTs
+				zap.Uint64("skipCheckpoint", watermark.CheckpointTs))
+			watermark.CheckpointTs = c.state.Status.CheckpointTs
 		}
 	})
 
-	c.updateStatus(newCheckpointTs, barrier.MinTableBarrierTs)
-	c.updateMetrics(currentTs, newCheckpointTs, c.resolvedTs)
+	failpoint.Inject("ChangefeedOwnerNotUpdateCheckpoint", func() {
+		watermark.CheckpointTs = c.state.Status.CheckpointTs
+	})
+
+	c.updateStatus(watermark.CheckpointTs, barrier.MinTableBarrierTs)
+	c.updateMetrics(currentTs, watermark.CheckpointTs, c.resolvedTs)
 	c.tickDownstreamObserver(ctx)
 
 	return nil
@@ -619,6 +649,7 @@ LOOP2:
 		ddlStartTs,
 		c.state.Status.CheckpointTs,
 		c.ddlSink,
+		filter,
 		c.ddlPuller,
 		c.schema,
 		c.redoDDLMgr,
diff --git a/cdc/owner/ddl_manager.go b/cdc/owner/ddl_manager.go
index 0d65fb61f..5423793ac 100644
--- a/cdc/owner/ddl_manager.go
+++ b/cdc/owner/ddl_manager.go
@@ -112,6 +112,8 @@ type ddlManager struct {
 	// tableCheckpoint store the tableCheckpoint of each table. We need to wait
 	// for the tableCheckpoint to reach the next ddl commitTs before executing the ddl
 	tableCheckpoint map[model.TableName]model.Ts
+	filter          filter.Filter
+
 	// pendingDDLs store the pending DDL events of all tables
 	// the DDL events in the same table are ordered by commitTs.
 	pendingDDLs map[model.TableName][]*model.DDLEvent
@@ -137,6 +139,7 @@ func newDDLManager(
 	startTs model.Ts,
 	checkpointTs model.Ts,
 	ddlSink DDLSink,
+	filter filter.Filter,
 	ddlPuller puller.DDLPuller,
 	schema *schemaWrap4Owner,
 	redoManager redo.DDLManager,
@@ -155,6 +158,7 @@ func newDDLManager(
 	return &ddlManager{
 		changfeedID:     changefeedID,
 		ddlSink:         ddlSink,
+		filter:          filter,
 		ddlPuller:       ddlPuller,
 		schema:          schema,
 		redoDDLManager:  redoManager,
@@ -248,7 +252,14 @@ func (m *ddlManager) tick(
 			// Send DDL events to redo log.
 			if m.redoDDLManager.Enabled() {
 				for _, event := range events {
-					err := m.redoDDLManager.EmitDDLEvent(ctx, event)
+					skip, _, err := m.shouldSkipDDL(event)
+					if err != nil {
+						return nil, nil, errors.Trace(err)
+					}
+					if skip {
+						continue
+					}
+					err = m.redoDDLManager.EmitDDLEvent(ctx, event)
 					if err != nil {
 						return nil, nil, err
 					}
@@ -338,26 +349,34 @@ func (m *ddlManager) shouldExecDDL(nextDDL *model.DDLEvent) bool {
 	return checkpointReachBarrier && redoCheckpointReachBarrier && redoDDLResolvedTsExceedBarrier
 }
 
+func (m *ddlManager) shouldSkipDDL(ddl *model.DDLEvent) (bool, string, error) {
+	ignored, err := m.filter.ShouldIgnoreDDLEvent(ddl)
+	if err != nil {
+		return false, "", errors.Trace(err)
+	}
+	if ignored {
+		return true, "ddl is ignored by event filter rule, skip it", nil
+	}
+
+	// In a BDR mode cluster, TiCDC can receive DDLs from all roles of TiDB.
+	// However, CDC only executes the DDLs from the TiDB that has BDRRolePrimary role.
+	if m.BDRMode {
+		return true, "changefeed is in BDRMode, skip all ddl in release 6.5", nil
+	}
+	return false, "", nil
+}
+
 // executeDDL executes ddlManager.executingDDL.
 func (m *ddlManager) executeDDL(ctx context.Context) error {
 	if m.executingDDL == nil {
 		return nil
 	}
-
-	// If changefeed is in BDRMode, skip ddl.
-	if m.BDRMode {
-		log.Info("changefeed is in BDRMode, skip a ddl event",
-			zap.String("namespace", m.changfeedID.Namespace),
-			zap.String("ID", m.changfeedID.ID),
-			zap.Any("ddlEvent", m.executingDDL))
-		tableName := m.executingDDL.TableInfo.TableName
-		// Set it to nil first to accelerate GC.
-		m.pendingDDLs[tableName][0] = nil
-		m.pendingDDLs[tableName] = m.pendingDDLs[tableName][1:]
-		m.schema.DoGC(m.executingDDL.CommitTs - 1)
-		m.justSentDDL = m.executingDDL
-		m.executingDDL = nil
-		m.cleanCache()
+	skip, cleanMsg, err := m.shouldSkipDDL(m.executingDDL)
+	if err != nil {
+		return errors.Trace(err)
+	}
+	if skip {
+		m.cleanCache(cleanMsg)
 		return nil
 	}
 
@@ -379,22 +398,10 @@ func (m *ddlManager) executeDDL(ctx context.Context) error {
 
 	done, err := m.ddlSink.emitDDLEvent(ctx, m.executingDDL)
 	if err != nil {
-		return err
+		return errors.Trace(err)
 	}
 	if done {
-		tableName := m.executingDDL.TableInfo.TableName
-		log.Info("execute a ddl event successfully",
-			zap.String("ddl", m.executingDDL.Query),
-			zap.Uint64("commitTs", m.executingDDL.CommitTs),
-			zap.Stringer("table", tableName),
-		)
-		// Set it to nil first to accelerate GC.
-		m.pendingDDLs[tableName][0] = nil
-		m.pendingDDLs[tableName] = m.pendingDDLs[tableName][1:]
-		m.schema.DoGC(m.executingDDL.CommitTs - 1)
-		m.justSentDDL = m.executingDDL
-		m.executingDDL = nil
-		m.cleanCache()
+		m.cleanCache("execute a ddl event successfully")
 	}
 	return nil
 }
@@ -591,9 +598,21 @@ func (m *ddlManager) getSnapshotTs() (ts uint64) {
 }
 
 // cleanCache cleans the tableInfoCache and physicalTablesCache.
-// It should be called after a DDL is applied to schema or a DDL
-// is sent to downstream successfully.
-func (m *ddlManager) cleanCache() {
+// It should be called after a DDL is skipped or sent to downstream successfully.
+func (m *ddlManager) cleanCache(msg string) {
+	tableName := m.executingDDL.TableInfo.TableName
+	log.Info(msg, zap.String("ddl", m.executingDDL.Query),
+		zap.String("namespace", m.changfeedID.Namespace),
+		zap.String("changefeed", m.changfeedID.ID),
+		zap.Any("ddlEvent", m.executingDDL))
+
+	// Set it to nil first to accelerate GC.
+	m.pendingDDLs[tableName][0] = nil
+	m.pendingDDLs[tableName] = m.pendingDDLs[tableName][1:]
+	m.schema.DoGC(m.executingDDL.CommitTs - 1)
+	m.justSentDDL = m.executingDDL
+	m.executingDDL = nil
+
 	m.tableInfoCache = nil
 	m.physicalTablesCache = nil
 }
diff --git a/cdc/owner/mock/status_provider_mock.go b/cdc/owner/mock/status_provider_mock.go
index 224428776..8b570b0bc 100644
--- a/cdc/owner/mock/status_provider_mock.go
+++ b/cdc/owner/mock/status_provider_mock.go
@@ -125,6 +125,21 @@ func (mr *MockStatusProviderMockRecorder) GetChangeFeedStatus(ctx, changefeedID
 	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChangeFeedStatus", reflect.TypeOf((*MockStatusProvider)(nil).GetChangeFeedStatus), ctx, changefeedID)
 }
 
+// GetChangeFeedSyncedStatus mocks base method.
+func (m *MockStatusProvider) GetChangeFeedSyncedStatus(ctx context.Context, changefeedID model.ChangeFeedID) (*model.ChangeFeedSyncedStatusForAPI, error) {
+	m.ctrl.T.Helper()
+	ret := m.ctrl.Call(m, "GetChangeFeedSyncedStatus", ctx, changefeedID)
+	ret0, _ := ret[0].(*model.ChangeFeedSyncedStatusForAPI)
+	ret1, _ := ret[1].(error)
+	return ret0, ret1
+}
+
+// GetChangeFeedSyncedStatus indicates an expected call of GetChangeFeedSyncedStatus.
+func (mr *MockStatusProviderMockRecorder) GetChangeFeedSyncedStatus(ctx, changefeedID interface{}) *gomock.Call {
+	mr.mock.ctrl.T.Helper()
+	return mr.mock.ctrl.RecordCallWithMethodType(mr.mock, "GetChangeFeedSyncedStatus", reflect.TypeOf((*MockStatusProvider)(nil).GetChangeFeedSyncedStatus), ctx, changefeedID)
+}
+
 // GetProcessors mocks base method.
 func (m *MockStatusProvider) GetProcessors(ctx context.Context) ([]*model.ProcInfoSnap, error) {
 	m.ctrl.T.Helper()
diff --git a/cdc/owner/owner.go b/cdc/owner/owner.go
index 8e16ef203..6519d3e14 100644
--- a/cdc/owner/owner.go
+++ b/cdc/owner/owner.go
@@ -492,6 +492,25 @@ func (o *ownerImpl) handleQueries(query *Query) error {
 			}
 		}
 		query.Data = ret
+	case QueryChangeFeedSyncedStatus:
+		cfReactor, ok := o.changefeeds[query.ChangeFeedID]
+		if !ok {
+			query.Data = nil
+			return nil
+		}
+		ret := &model.ChangeFeedSyncedStatusForAPI{}
+		ret.LastSyncedTs = cfReactor.lastSyncedTs
+		ret.CheckpointTs = cfReactor.state.Status.CheckpointTs
+		ret.PullerResolvedTs = cfReactor.pullerResolvedTs
+
+		if cfReactor.state == nil || cfReactor.state.Info == nil || cfReactor.state.Info.Config == nil {
+			ret.CheckpointInterval = 0
+			ret.SyncedCheckInterval = 0
+		} else {
+			ret.CheckpointInterval = cfReactor.state.Info.Config.SyncedStatus.CheckpointInterval
+			ret.SyncedCheckInterval = cfReactor.state.Info.Config.SyncedStatus.SyncedCheckInterval
+		}
+		query.Data = ret
 	case QueryAllTaskStatuses:
 		cfReactor, ok := o.changefeeds[query.ChangeFeedID]
 		if !ok {
diff --git a/cdc/owner/schema.go b/cdc/owner/schema.go
index b2f0ea39a..da8b208da 100644
--- a/cdc/owner/schema.go
+++ b/cdc/owner/schema.go
@@ -265,22 +265,15 @@ func (s *schemaWrap4Owner) filterDDLEvents(ddlEvents []*model.DDLEvent) ([]*mode
 			err     error
 		)
 		if event.Type == timodel.ActionRenameTable {
-			ignored, err = s.filter.ShouldDiscardDDL(
-				event.StartTs,
+			ignored = s.filter.ShouldDiscardDDL(
 				event.Type,
 				event.PreTableInfo.TableName.Schema,
-				event.PreTableInfo.TableName.Table,
-				event.Query)
-			if err != nil {
-				return nil, errors.Trace(err)
-			}
+				event.PreTableInfo.TableName.Table)
 		} else {
-			ignored, err = s.filter.ShouldDiscardDDL(
-				event.StartTs,
+			ignored = s.filter.ShouldDiscardDDL(
 				event.Type,
 				event.TableInfo.TableName.Schema,
-				event.TableInfo.TableName.Table,
-				event.Query)
+				event.TableInfo.TableName.Table)
 			if err != nil {
 				return nil, errors.Trace(err)
 			}
diff --git a/cdc/owner/status_provider.go b/cdc/owner/status_provider.go
index 9cfdff04f..93c16171c 100644
--- a/cdc/owner/status_provider.go
+++ b/cdc/owner/status_provider.go
@@ -33,6 +33,9 @@ type StatusProvider interface {
 	// GetAllChangeFeedInfo returns all changefeeds' info.
 	GetAllChangeFeedInfo(ctx context.Context) (map[model.ChangeFeedID]*model.ChangeFeedInfo, error)
 
+	// GetChangeFeedSyncedStatus returns a changefeeds' synced status.
+	GetChangeFeedSyncedStatus(ctx context.Context, changefeedID model.ChangeFeedID) (*model.ChangeFeedSyncedStatusForAPI, error)
+
 	// GetChangeFeedInfo returns a changefeeds' info.
 	GetChangeFeedInfo(ctx context.Context, changefeedID model.ChangeFeedID) (*model.ChangeFeedInfo, error)
 
@@ -68,7 +71,9 @@ const (
 	// QueryHealth is the type of query cluster health info.
 	QueryHealth
 	// QueryOwner is the type of query changefeed owner
-	QueryOwner = 6
+	QueryOwner
+	// QueryChangeFeedSyncedStatus is the type of query changefeed synced status
+	QueryChangeFeedSyncedStatus
 )
 
 // Query wraps query command and return results.
@@ -126,6 +131,22 @@ func (p *ownerStatusProvider) GetAllChangeFeedInfo(ctx context.Context) (
 	return query.Data.(map[model.ChangeFeedID]*model.ChangeFeedInfo), nil
 }
 
+func (p *ownerStatusProvider) GetChangeFeedSyncedStatus(ctx context.Context,
+	changefeedID model.ChangeFeedID,
+) (*model.ChangeFeedSyncedStatusForAPI, error) {
+	query := &Query{
+		Tp:           QueryChangeFeedSyncedStatus,
+		ChangeFeedID: changefeedID,
+	}
+	if err := p.sendQueryToOwner(ctx, query); err != nil {
+		return nil, errors.Trace(err)
+	}
+	if query.Data == nil {
+		return nil, cerror.ErrChangeFeedNotExists.GenWithStackByArgs(changefeedID)
+	}
+	return query.Data.(*model.ChangeFeedSyncedStatusForAPI), nil
+}
+
 func (p *ownerStatusProvider) GetChangeFeedInfo(ctx context.Context,
 	changefeedID model.ChangeFeedID,
 ) (*model.ChangeFeedInfo, error) {
diff --git a/cdc/processor/processor.go b/cdc/processor/processor.go
index 58205b110..bc983b780 100644
--- a/cdc/processor/processor.go
+++ b/cdc/processor/processor.go
@@ -363,6 +363,7 @@ func (p *processor) GetTableSpanStatus(span tablepb.Span, collectStat bool) tabl
 		Checkpoint: tablepb.Checkpoint{
 			CheckpointTs: sinkStats.CheckpointTs,
 			ResolvedTs:   sinkStats.ResolvedTs,
+			LastSyncedTs: sinkStats.LastSyncedTs,
 		},
 		State: state,
 		Stats: stats,
diff --git a/cdc/processor/sinkmanager/manager.go b/cdc/processor/sinkmanager/manager.go
index a2ae9c22e..dd277b0fc 100644
--- a/cdc/processor/sinkmanager/manager.go
+++ b/cdc/processor/sinkmanager/manager.go
@@ -58,6 +58,7 @@ const (
 type TableStats struct {
 	CheckpointTs model.Ts
 	ResolvedTs   model.Ts
+	LastSyncedTs model.Ts
 	BarrierTs    model.Ts
 }
 
@@ -157,6 +158,7 @@ func New(
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 	}
 
+	totalQuota := changefeedInfo.Config.MemoryQuota
 	if redoDMLMgr != nil && redoDMLMgr.Enabled() {
 		m.redoDMLMgr = redoDMLMgr
 		m.redoProgressHeap = newTableProgresses()
@@ -164,18 +166,26 @@ func New(
 		m.redoTaskChan = make(chan *redoTask)
 		m.redoWorkerAvailable = make(chan struct{}, 1)
 
-		// Use 3/4 memory quota as redo quota, and 1/2 again for redo cache.
-		m.sinkMemQuota = memquota.NewMemQuota(changefeedID, changefeedInfo.Config.MemoryQuota/4*1, "sink")
-		redoQuota := changefeedInfo.Config.MemoryQuota / 4 * 3
+		consistentMemoryUsage := changefeedInfo.Config.Consistent.MemoryUsage
+		if consistentMemoryUsage == nil {
+			consistentMemoryUsage = config.GetDefaultReplicaConfig().Consistent.MemoryUsage
+		}
+
+		redoQuota := totalQuota * consistentMemoryUsage.MemoryQuotaPercentage / 100
+		sinkQuota := totalQuota - redoQuota
+		m.sinkMemQuota = memquota.NewMemQuota(changefeedID, sinkQuota, "sink")
 		m.redoMemQuota = memquota.NewMemQuota(changefeedID, redoQuota, "redo")
-		m.eventCache = newRedoEventCache(changefeedID, redoQuota/2*1)
+
+		eventCache := redoQuota * consistentMemoryUsage.EventCachePercentage / 100
+		if eventCache > 0 {
+			m.eventCache = newRedoEventCache(changefeedID, eventCache)
+		}
 	} else {
-		m.sinkMemQuota = memquota.NewMemQuota(changefeedID, changefeedInfo.Config.MemoryQuota, "sink")
+		m.sinkMemQuota = memquota.NewMemQuota(changefeedID, totalQuota, "sink")
 		m.redoMemQuota = memquota.NewMemQuota(changefeedID, 0, "redo")
 	}
 
 	m.ready = make(chan struct{})
-
 	return m
 }
 
@@ -998,6 +1008,7 @@ func (m *SinkManager) GetTableStats(span tablepb.Span) TableStats {
 	tableSink := value.(*tableSinkWrapper)
 
 	checkpointTs := tableSink.getCheckpointTs()
+	lastSyncedTs := tableSink.getLastSyncedTs()
 	m.sinkMemQuota.Release(span, checkpointTs)
 	m.redoMemQuota.Release(span, checkpointTs)
 
@@ -1040,6 +1051,7 @@ func (m *SinkManager) GetTableStats(span tablepb.Span) TableStats {
 	return TableStats{
 		CheckpointTs: checkpointTs.ResolvedMark(),
 		ResolvedTs:   resolvedTs,
+		LastSyncedTs: lastSyncedTs,
 		BarrierTs:    tableSink.barrierTs.Load(),
 	}
 }
diff --git a/cdc/processor/sinkmanager/table_sink_wrapper.go b/cdc/processor/sinkmanager/table_sink_wrapper.go
index 45ce8c55e..9d4938fc4 100644
--- a/cdc/processor/sinkmanager/table_sink_wrapper.go
+++ b/cdc/processor/sinkmanager/table_sink_wrapper.go
@@ -58,6 +58,7 @@ type tableSinkWrapper struct {
 		advanced     time.Time
 		resolvedTs   model.ResolvedTs
 		checkpointTs model.ResolvedTs
+		lastSyncedTs model.Ts
 	}
 
 	// state used to control the lifecycle of the table.
@@ -223,6 +224,15 @@ func (t *tableSinkWrapper) updateResolvedTs(ts model.ResolvedTs) error {
 	return t.tableSink.s.UpdateResolvedTs(ts)
 }
 
+func (t *tableSinkWrapper) getLastSyncedTs() uint64 {
+	t.tableSink.RLock()
+	defer t.tableSink.RUnlock()
+	if t.tableSink.s != nil {
+		return t.tableSink.s.GetLastSyncedTs()
+	}
+	return t.tableSink.lastSyncedTs
+}
+
 func (t *tableSinkWrapper) getCheckpointTs() model.ResolvedTs {
 	t.tableSink.RLock()
 	defer t.tableSink.RUnlock()
@@ -362,6 +372,7 @@ func (t *tableSinkWrapper) doTableSinkClear() {
 		t.tableSink.checkpointTs = checkpointTs
 	}
 	t.tableSink.resolvedTs = checkpointTs
+	t.tableSink.lastSyncedTs = t.tableSink.s.GetLastSyncedTs()
 	t.tableSink.advanced = time.Now()
 	t.tableSink.innerMu.Unlock()
 	t.tableSink.s = nil
diff --git a/cdc/processor/tablepb/table.pb.go b/cdc/processor/tablepb/table.pb.go
index da01e6d59..c4b86e834 100644
--- a/cdc/processor/tablepb/table.pb.go
+++ b/cdc/processor/tablepb/table.pb.go
@@ -117,6 +117,7 @@ var xxx_messageInfo_Span proto.InternalMessageInfo
 type Checkpoint struct {
 	CheckpointTs Ts `protobuf:"varint,1,opt,name=checkpoint_ts,json=checkpointTs,proto3,casttype=Ts" json:"checkpoint_ts,omitempty"`
 	ResolvedTs   Ts `protobuf:"varint,2,opt,name=resolved_ts,json=resolvedTs,proto3,casttype=Ts" json:"resolved_ts,omitempty"`
+	LastSyncedTs Ts `protobuf:"varint,3,opt,name=last_synced_ts,json=lastSyncedTs,proto3,casttype=Ts" json:"last_synced_ts,omitempty"`
 }
 
 func (m *Checkpoint) Reset()         { *m = Checkpoint{} }
@@ -166,6 +167,13 @@ func (m *Checkpoint) GetResolvedTs() Ts {
 	return 0
 }
 
+func (m *Checkpoint) GetLastSyncedTs() Ts {
+	if m != nil {
+		return m.LastSyncedTs
+	}
+	return 0
+}
+
 // Stats holds a statistic for a table.
 type Stats struct {
 	// Number of captured regions.
@@ -329,50 +337,52 @@ func init() {
 func init() { proto.RegisterFile("processor/tablepb/table.proto", fileDescriptor_ae83c9c6cf5ef75c) }
 
 var fileDescriptor_ae83c9c6cf5ef75c = []byte{
-	// 688 bytes of a gzipped FileDescriptorProto
+	// 706 bytes of a gzipped FileDescriptorProto
 	0x1f, 0x8b, 0x08, 0x00, 0x00, 0x00, 0x00, 0x00, 0x02, 0xff, 0x9c, 0x54, 0xbf, 0x6f, 0xd3, 0x4e,
-	0x1c, 0xb5, 0x9d, 0x5f, 0xcd, 0x27, 0xf9, 0x7e, 0xe5, 0xde, 0xb7, 0xed, 0x37, 0x44, 0x22, 0x31,
-	0x51, 0x81, 0xaa, 0x95, 0x1c, 0x28, 0x0b, 0xea, 0xd6, 0xb4, 0x80, 0xaa, 0x0a, 0x09, 0xb9, 0x81,
-	0x81, 0x25, 0xf2, 0x8f, 0xc3, 0xb5, 0x1a, 0xce, 0x96, 0xef, 0xd2, 0x2a, 0x1b, 0x23, 0xca, 0x02,
-	0x13, 0x62, 0x89, 0xd4, 0x3f, 0xa7, 0x63, 0x47, 0x06, 0x14, 0x41, 0x2a, 0x66, 0xf6, 0x4e, 0xe8,
-	0xee, 0xdc, 0xb8, 0x09, 0x0c, 0xa1, 0x4b, 0x72, 0xbe, 0xf7, 0x3e, 0xcf, 0xef, 0xbd, 0x3b, 0x19,
-	0x6e, 0x47, 0x71, 0xe8, 0x62, 0x4a, 0xc3, 0xb8, 0xc9, 0x6c, 0xa7, 0x8b, 0x23, 0x47, 0xfe, 0x9b,
-	0x51, 0x1c, 0xb2, 0x10, 0xad, 0x46, 0x01, 0xf1, 0x5d, 0x3b, 0x32, 0x59, 0xf0, 0xa6, 0x1b, 0x9e,
-	0x98, 0xae, 0xe7, 0x9a, 0x93, 0x09, 0x33, 0x99, 0xa8, 0x2e, 0xf9, 0xa1, 0x1f, 0x8a, 0x81, 0x26,
-	0x5f, 0xc9, 0xd9, 0xc6, 0x07, 0x15, 0xb2, 0x07, 0x91, 0x4d, 0xd0, 0x43, 0x58, 0x10, 0xcc, 0x4e,
-	0xe0, 0x55, 0x54, 0x43, 0x5d, 0xcb, 0xb4, 0x56, 0xc6, 0xa3, 0x7a, 0xa1, 0xcd, 0xf7, 0xf6, 0x76,
-	0x2f, 0xd3, 0xa5, 0x55, 0x10, 0xbc, 0x3d, 0x0f, 0xad, 0x42, 0x91, 0x32, 0x3b, 0x66, 0x9d, 0x23,
-	0xdc, 0xaf, 0x68, 0x86, 0xba, 0x56, 0x6e, 0x15, 0x2e, 0x47, 0xf5, 0xcc, 0x3e, 0xee, 0x5b, 0x0b,
-	0x02, 0xd9, 0xc7, 0x7d, 0x64, 0x40, 0x01, 0x13, 0x4f, 0x70, 0x32, 0xd3, 0x9c, 0x3c, 0x26, 0xde,
-	0x3e, 0xee, 0x6f, 0x95, 0xdf, 0x9f, 0xd6, 0x95, 0xcf, 0xa7, 0x75, 0xe5, 0xdd, 0x57, 0x43, 0x69,
-	0x38, 0x00, 0x3b, 0x87, 0xd8, 0x3d, 0x8a, 0xc2, 0x80, 0x30, 0xb4, 0x01, 0xff, 0xb8, 0x93, 0xa7,
-	0x0e, 0xa3, 0xc2, 0x5b, 0xb6, 0x95, 0xbf, 0x1c, 0xd5, 0xb5, 0x36, 0xb5, 0xca, 0x29, 0xd8, 0xa6,
-	0xe8, 0x3e, 0x94, 0x62, 0x4c, 0xc3, 0xee, 0x31, 0xf6, 0x38, 0x55, 0x9b, 0xa2, 0xc2, 0x15, 0xd4,
-	0xa6, 0x8d, 0x1f, 0x1a, 0xe4, 0x0e, 0x98, 0xcd, 0x28, 0xba, 0x03, 0xe5, 0x18, 0xfb, 0x41, 0x48,
-	0x3a, 0x6e, 0xd8, 0x23, 0x4c, 0xca, 0x5b, 0x25, 0xb9, 0xb7, 0xc3, 0xb7, 0xd0, 0x5d, 0x00, 0xb7,
-	0x17, 0xc7, 0x58, 0xbe, 0x7f, 0x5a, 0xb4, 0x98, 0x20, 0x6d, 0x8a, 0x18, 0x2c, 0x52, 0x66, 0xfb,
-	0xb8, 0x93, 0x5a, 0xa2, 0x95, 0x8c, 0x91, 0x59, 0x2b, 0x6d, 0x6e, 0x9b, 0xf3, 0x9c, 0x90, 0x29,
-	0x1c, 0xf1, 0x5f, 0x1f, 0xa7, 0x0d, 0xd0, 0x27, 0x84, 0xc5, 0xfd, 0x56, 0xf6, 0x6c, 0x54, 0x57,
-	0x2c, 0x9d, 0xce, 0x80, 0xdc, 0x9c, 0x63, 0xc7, 0x71, 0x80, 0x63, 0x6e, 0x2e, 0x3b, 0x6d, 0x2e,
-	0x41, 0xda, 0xb4, 0xda, 0x83, 0xe5, 0x3f, 0xea, 0x22, 0x1d, 0x32, 0xfc, 0x64, 0x78, 0xec, 0xa2,
-	0xc5, 0x97, 0xe8, 0x29, 0xe4, 0x8e, 0xed, 0x6e, 0x0f, 0x8b, 0xa4, 0xa5, 0xcd, 0x07, 0xf3, 0x79,
-	0x4f, 0x85, 0x2d, 0x39, 0xbe, 0xa5, 0x3d, 0x56, 0x1b, 0x3f, 0x35, 0x28, 0x89, 0x6b, 0xc3, 0xa3,
-	0xf5, 0xe8, 0x4d, 0x2e, 0xd9, 0x2e, 0x64, 0x69, 0x64, 0x93, 0x4a, 0x4e, 0xb8, 0x59, 0x9f, 0xb3,
-	0xc9, 0xc8, 0x26, 0x49, 0x65, 0x62, 0x9a, 0x87, 0xa2, 0xcc, 0x66, 0x32, 0xd4, 0xbf, 0xf3, 0x86,
-	0x9a, 0x58, 0xc7, 0x96, 0x1c, 0x47, 0xaf, 0x00, 0xd2, 0xe3, 0x15, 0xf7, 0xf9, 0x06, 0x0d, 0x25,
-	0xce, 0xae, 0x29, 0xa1, 0x67, 0xd2, 0x9f, 0x3c, 0xc1, 0xd2, 0xe6, 0xc6, 0x5f, 0x5c, 0x98, 0x44,
-	0x4d, 0xce, 0xaf, 0x7f, 0xd2, 0x00, 0x52, 0xdb, 0xa8, 0x01, 0x85, 0x97, 0xe4, 0x88, 0x84, 0x27,
-	0x44, 0x57, 0xaa, 0xcb, 0x83, 0xa1, 0xb1, 0x98, 0x82, 0x09, 0x80, 0x0c, 0xc8, 0x6f, 0x3b, 0x14,
-	0x13, 0xa6, 0xab, 0xd5, 0xa5, 0xc1, 0xd0, 0xd0, 0x53, 0x8a, 0xdc, 0x47, 0xf7, 0xa0, 0xf8, 0x22,
-	0xc6, 0x91, 0x1d, 0x07, 0xc4, 0xd7, 0xb5, 0xea, 0xff, 0x83, 0xa1, 0xf1, 0x5f, 0x4a, 0x9a, 0x40,
-	0x68, 0x15, 0x16, 0xe4, 0x03, 0xf6, 0xf4, 0x4c, 0x75, 0x65, 0x30, 0x34, 0xd0, 0x2c, 0x0d, 0x7b,
-	0x68, 0x1d, 0x4a, 0x16, 0x8e, 0xba, 0x81, 0x6b, 0x33, 0xae, 0x97, 0xad, 0xde, 0x1a, 0x0c, 0x8d,
-	0xe5, 0x6b, 0x5d, 0xa7, 0x20, 0x57, 0x3c, 0x60, 0x61, 0xc4, 0xdb, 0xd0, 0x73, 0xb3, 0x8a, 0x57,
-	0x08, 0x4f, 0x29, 0xd6, 0xd8, 0xd3, 0xf3, 0xb3, 0x29, 0x13, 0xa0, 0xf5, 0xfc, 0xfc, 0x7b, 0x4d,
-	0x39, 0x1b, 0xd7, 0xd4, 0xf3, 0x71, 0x4d, 0xfd, 0x36, 0xae, 0xa9, 0x1f, 0x2f, 0x6a, 0xca, 0xf9,
-	0x45, 0x4d, 0xf9, 0x72, 0x51, 0x53, 0x5e, 0x37, 0xfd, 0x80, 0x1d, 0xf6, 0x1c, 0xd3, 0x0d, 0xdf,
-	0x36, 0x93, 0xea, 0x9b, 0xb2, 0xfa, 0xa6, 0xeb, 0xb9, 0xcd, 0xdf, 0xbe, 0xbf, 0x4e, 0x5e, 0x7c,
-	0x3e, 0x1f, 0xfd, 0x0a, 0x00, 0x00, 0xff, 0xff, 0x80, 0x0e, 0x45, 0x99, 0x9b, 0x05, 0x00, 0x00,
+	0x1c, 0xb5, 0x9d, 0x5f, 0xcd, 0x27, 0xf9, 0x56, 0xee, 0x7d, 0xdb, 0x52, 0x22, 0x91, 0x98, 0xa8,
+	0x40, 0xd5, 0x22, 0x07, 0xca, 0x82, 0xba, 0x35, 0x2d, 0xa0, 0xaa, 0x42, 0x42, 0x4e, 0x60, 0x60,
+	0x89, 0x1c, 0xfb, 0x70, 0xad, 0x86, 0xb3, 0xe5, 0xbb, 0xb4, 0xca, 0xc6, 0x88, 0xb2, 0xd0, 0x09,
+	0xb1, 0x44, 0xea, 0x9f, 0xd3, 0xb1, 0x23, 0x03, 0x8a, 0x20, 0x15, 0x33, 0x7b, 0x27, 0x74, 0x77,
+	0x6e, 0xdc, 0x04, 0x86, 0xd0, 0x25, 0x39, 0xdf, 0x7b, 0x9f, 0xa7, 0xf7, 0xde, 0x9d, 0x0e, 0xee,
+	0x84, 0x51, 0xe0, 0x60, 0x4a, 0x83, 0xa8, 0xc6, 0xec, 0x76, 0x07, 0x87, 0x6d, 0xf9, 0x6f, 0x86,
+	0x51, 0xc0, 0x02, 0xb4, 0x1a, 0xfa, 0xc4, 0x73, 0xec, 0xd0, 0x64, 0xfe, 0xbb, 0x4e, 0x70, 0x6c,
+	0x3a, 0xae, 0x63, 0x8e, 0x27, 0xcc, 0x78, 0xa2, 0xb4, 0xe8, 0x05, 0x5e, 0x20, 0x06, 0x6a, 0x7c,
+	0x25, 0x67, 0xab, 0x9f, 0x54, 0x48, 0x37, 0x42, 0x9b, 0xa0, 0xc7, 0x30, 0x27, 0x98, 0x2d, 0xdf,
+	0x5d, 0x51, 0x0d, 0x75, 0x2d, 0x55, 0x5f, 0x1e, 0x0d, 0x2b, 0xb9, 0x26, 0xdf, 0xdb, 0xdb, 0xbd,
+	0x4c, 0x96, 0x56, 0x4e, 0xf0, 0xf6, 0x5c, 0xb4, 0x0a, 0x79, 0xca, 0xec, 0x88, 0xb5, 0x0e, 0x71,
+	0x6f, 0x45, 0x33, 0xd4, 0xb5, 0x62, 0x3d, 0x77, 0x39, 0xac, 0xa4, 0xf6, 0x71, 0xcf, 0x9a, 0x13,
+	0xc8, 0x3e, 0xee, 0x21, 0x03, 0x72, 0x98, 0xb8, 0x82, 0x93, 0x9a, 0xe4, 0x64, 0x31, 0x71, 0xf7,
+	0x71, 0x6f, 0xab, 0xf8, 0xf1, 0xb4, 0xa2, 0x7c, 0x39, 0xad, 0x28, 0x1f, 0xbe, 0x19, 0x4a, 0xf5,
+	0x44, 0x05, 0xd8, 0x39, 0xc0, 0xce, 0x61, 0x18, 0xf8, 0x84, 0xa1, 0x0d, 0xf8, 0xcf, 0x19, 0x7f,
+	0xb5, 0x18, 0x15, 0xe6, 0xd2, 0xf5, 0xec, 0xe5, 0xb0, 0xa2, 0x35, 0xa9, 0x55, 0x4c, 0xc0, 0x26,
+	0x45, 0x0f, 0xa0, 0x10, 0x61, 0x1a, 0x74, 0x8e, 0xb0, 0xcb, 0xa9, 0xda, 0x04, 0x15, 0xae, 0xa0,
+	0x26, 0x45, 0x0f, 0x61, 0xbe, 0x63, 0x53, 0xd6, 0xa2, 0x3d, 0xe2, 0x48, 0x6e, 0x6a, 0x52, 0x96,
+	0xa3, 0x0d, 0x01, 0x36, 0x69, 0xf5, 0xa7, 0x06, 0x99, 0x06, 0xb3, 0x19, 0x45, 0x77, 0xa1, 0x18,
+	0x61, 0xcf, 0x0f, 0x48, 0xcb, 0x09, 0xba, 0x84, 0x49, 0x33, 0x56, 0x41, 0xee, 0xed, 0xf0, 0x2d,
+	0x74, 0x0f, 0xc0, 0xe9, 0x46, 0x11, 0x96, 0x6e, 0x27, 0x2d, 0xe4, 0x63, 0xa4, 0x49, 0x11, 0x83,
+	0x05, 0xca, 0x6c, 0x0f, 0xb7, 0x92, 0x00, 0xdc, 0x44, 0x6a, 0xad, 0xb0, 0xb9, 0x6d, 0xce, 0x72,
+	0xa0, 0xa6, 0x70, 0xc4, 0x7f, 0x3d, 0x9c, 0xf4, 0x45, 0x9f, 0x11, 0x16, 0xf5, 0xea, 0xe9, 0xb3,
+	0x61, 0x45, 0xb1, 0x74, 0x3a, 0x05, 0x72, 0x73, 0x6d, 0x3b, 0x8a, 0x7c, 0x1c, 0x71, 0x73, 0xe9,
+	0x49, 0x73, 0x31, 0xd2, 0xa4, 0xa5, 0x2e, 0x2c, 0xfd, 0x55, 0x17, 0xe9, 0x90, 0xe2, 0x07, 0xc9,
+	0x63, 0xe7, 0x2d, 0xbe, 0x44, 0xcf, 0x21, 0x73, 0x64, 0x77, 0xba, 0x58, 0x24, 0x2d, 0x6c, 0x3e,
+	0x9a, 0xcd, 0x7b, 0x22, 0x6c, 0xc9, 0xf1, 0x2d, 0xed, 0xa9, 0x5a, 0xfd, 0xa5, 0x41, 0x41, 0xdc,
+	0x32, 0x1e, 0xad, 0x4b, 0x6f, 0x72, 0x27, 0x77, 0x21, 0x4d, 0x43, 0x9b, 0xac, 0x64, 0x84, 0x9b,
+	0xf5, 0x19, 0x9b, 0x0c, 0x6d, 0x12, 0x57, 0x26, 0xa6, 0x79, 0x28, 0xca, 0x6c, 0x26, 0x43, 0xcd,
+	0xcf, 0x1a, 0x6a, 0x6c, 0x1d, 0x5b, 0x72, 0x1c, 0xbd, 0x01, 0x48, 0x8e, 0x57, 0x5c, 0xb1, 0x1b,
+	0x34, 0x14, 0x3b, 0xbb, 0xa6, 0x84, 0x5e, 0x48, 0x7f, 0xf2, 0x04, 0x0b, 0x9b, 0x1b, 0xff, 0x70,
+	0x61, 0x62, 0x35, 0x39, 0xbf, 0xfe, 0x59, 0x03, 0x48, 0x6c, 0xa3, 0x2a, 0xe4, 0x5e, 0x93, 0x43,
+	0x12, 0x1c, 0x13, 0x5d, 0x29, 0x2d, 0xf5, 0x07, 0xc6, 0x42, 0x02, 0xc6, 0x00, 0x32, 0x20, 0xbb,
+	0xdd, 0xa6, 0x98, 0x30, 0x5d, 0x2d, 0x2d, 0xf6, 0x07, 0x86, 0x9e, 0x50, 0xe4, 0x3e, 0xba, 0x0f,
+	0xf9, 0x57, 0x11, 0x0e, 0xed, 0xc8, 0x27, 0x9e, 0xae, 0x95, 0x6e, 0xf5, 0x07, 0xc6, 0xff, 0x09,
+	0x69, 0x0c, 0xa1, 0x55, 0x98, 0x93, 0x1f, 0xd8, 0xd5, 0x53, 0xa5, 0xe5, 0xfe, 0xc0, 0x40, 0xd3,
+	0x34, 0xec, 0xa2, 0x75, 0x28, 0x58, 0x38, 0xec, 0xf8, 0x8e, 0xcd, 0xb8, 0x5e, 0xba, 0x74, 0xbb,
+	0x3f, 0x30, 0x96, 0xae, 0x75, 0x9d, 0x80, 0x5c, 0xb1, 0xc1, 0x82, 0x90, 0xb7, 0xa1, 0x67, 0xa6,
+	0x15, 0xaf, 0x10, 0x9e, 0x52, 0xac, 0xb1, 0xab, 0x67, 0xa7, 0x53, 0xc6, 0x40, 0xfd, 0xe5, 0xf9,
+	0x8f, 0xb2, 0x72, 0x36, 0x2a, 0xab, 0xe7, 0xa3, 0xb2, 0xfa, 0x7d, 0x54, 0x56, 0x4f, 0x2e, 0xca,
+	0xca, 0xf9, 0x45, 0x59, 0xf9, 0x7a, 0x51, 0x56, 0xde, 0xd6, 0x3c, 0x9f, 0x1d, 0x74, 0xdb, 0xa6,
+	0x13, 0xbc, 0xaf, 0xc5, 0xd5, 0xd7, 0x64, 0xf5, 0x35, 0xc7, 0x75, 0x6a, 0x7f, 0x3c, 0xd7, 0xed,
+	0xac, 0x78, 0x6d, 0x9f, 0xfc, 0x0e, 0x00, 0x00, 0xff, 0xff, 0xa3, 0xeb, 0x06, 0x9b, 0xca, 0x05,
+	0x00, 0x00,
 }
 
 func (m *Span) Marshal() (dAtA []byte, err error) {
@@ -437,6 +447,11 @@ func (m *Checkpoint) MarshalToSizedBuffer(dAtA []byte) (int, error) {
 	_ = i
 	var l int
 	_ = l
+	if m.LastSyncedTs != 0 {
+		i = encodeVarintTable(dAtA, i, uint64(m.LastSyncedTs))
+		i--
+		dAtA[i] = 0x18
+	}
 	if m.ResolvedTs != 0 {
 		i = encodeVarintTable(dAtA, i, uint64(m.ResolvedTs))
 		i--
@@ -618,6 +633,9 @@ func (m *Checkpoint) Size() (n int) {
 	if m.ResolvedTs != 0 {
 		n += 1 + sovTable(uint64(m.ResolvedTs))
 	}
+	if m.LastSyncedTs != 0 {
+		n += 1 + sovTable(uint64(m.LastSyncedTs))
+	}
 	return n
 }
 
@@ -879,6 +897,25 @@ func (m *Checkpoint) Unmarshal(dAtA []byte) error {
 					break
 				}
 			}
+		case 3:
+			if wireType != 0 {
+				return fmt.Errorf("proto: wrong wireType = %d for field LastSyncedTs", wireType)
+			}
+			m.LastSyncedTs = 0
+			for shift := uint(0); ; shift += 7 {
+				if shift >= 64 {
+					return ErrIntOverflowTable
+				}
+				if iNdEx >= l {
+					return io.ErrUnexpectedEOF
+				}
+				b := dAtA[iNdEx]
+				iNdEx++
+				m.LastSyncedTs |= Ts(b&0x7F) << shift
+				if b < 0x80 {
+					break
+				}
+			}
 		default:
 			iNdEx = preIndex
 			skippy, err := skipTable(dAtA[iNdEx:])
diff --git a/cdc/processor/tablepb/table.proto b/cdc/processor/tablepb/table.proto
index 97de3a23e..3727edc0c 100644
--- a/cdc/processor/tablepb/table.proto
+++ b/cdc/processor/tablepb/table.proto
@@ -57,6 +57,7 @@ enum TableState {
 message Checkpoint {
     uint64 checkpoint_ts = 1 [(gogoproto.casttype) = "Ts"];
     uint64 resolved_ts = 2 [(gogoproto.casttype) = "Ts"];
+    uint64 last_synced_ts = 3 [(gogoproto.casttype) = "Ts"];
 }
 
 // Stats holds a statistic for a table.
diff --git a/cdc/puller/ddl_puller.go b/cdc/puller/ddl_puller.go
index 041ebc5dc..8afe10f1b 100644
--- a/cdc/puller/ddl_puller.go
+++ b/cdc/puller/ddl_puller.go
@@ -297,11 +297,7 @@ func (p *ddlJobPullerImpl) handleRenameTables(job *timodel.Job) (skip bool, err
 		if !ok {
 			shouldDiscardOldTable = true
 		} else {
-			shouldDiscardOldTable, err = p.filter.ShouldDiscardDDL(job.StartTS,
-				job.Type, oldSchemaNames[i].O, oldTable.Name.O, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
+			shouldDiscardOldTable = p.filter.ShouldDiscardDDL(job.Type, oldSchemaNames[i].O, oldTable.Name.O)
 		}
 
 		newSchemaName, ok := snap.SchemaByID(newSchemaIDs[i])
@@ -309,11 +305,7 @@ func (p *ddlJobPullerImpl) handleRenameTables(job *timodel.Job) (skip bool, err
 			// the new table name does not hit the filter rule, so we should discard the table.
 			shouldDiscardNewTable = true
 		} else {
-			shouldDiscardNewTable, err = p.filter.ShouldDiscardDDL(job.StartTS,
-				job.Type, newSchemaName.Name.O, newTableNames[i].O, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
+			shouldDiscardNewTable = p.filter.ShouldDiscardDDL(job.Type, newSchemaName.Name.O, newTableNames[i].O)
 		}
 
 		if shouldDiscardOldTable && shouldDiscardNewTable {
@@ -425,12 +417,7 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 	snap := p.schemaStorage.GetLastSnapshot()
 	if err := snap.FillSchemaName(job); err != nil {
 		log.Info("failed to fill schema name for ddl job", zap.Error(err))
-		discard, fErr := p.filter.
-			ShouldDiscardDDL(job.StartTS, job.Type, job.SchemaName, job.TableName, job.Query)
-		if fErr != nil {
-			return false, errors.Trace(fErr)
-		}
-		if discard {
+		if p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.TableName) {
 			return true, nil
 		}
 		return true, errors.Trace(err)
@@ -453,11 +440,7 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 		oldTable, ok := snap.PhysicalTableByID(job.TableID)
 		if !ok {
 			// 1. If we can not find the old table, and the new table name is in filter rule, return error.
-			discard, err := p.filter.
-				ShouldDiscardDDL(job.StartTS, job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
+			discard := p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O)
 			if !discard {
 				return true, cerror.ErrSyncRenameTableFailed.GenWithStackByArgs(job.TableID, job.Query)
 			}
@@ -468,16 +451,8 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 				zap.String("oldSchemaName", oldTable.TableName.Schema))
 			// since we can find the old table, we must can find the old schema.
 			// 2. If we can find the preTableInfo, we filter it by the old table name.
-			skipByOldTableName, err := p.filter.ShouldDiscardDDL(job.StartTS,
-				job.Type, oldTable.TableName.Schema, oldTable.TableName.Table, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
-			skipByNewTableName, err := p.filter.ShouldDiscardDDL(job.StartTS,
-				job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O, job.Query)
-			if err != nil {
-				return true, errors.Trace(err)
-			}
+			skipByOldTableName := p.filter.ShouldDiscardDDL(job.Type, oldTable.TableName.Schema, oldTable.TableName.Table)
+			skipByNewTableName := p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.BinlogInfo.TableInfo.Name.O)
 			// 3. If its old table name is not in filter rule, and its new table name in filter rule, return error.
 			if skipByOldTableName && !skipByNewTableName {
 				return true, cerror.ErrSyncRenameTableFailed.GenWithStackByArgs(job.TableID, job.Query)
@@ -492,11 +467,7 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 		if job.BinlogInfo.TableInfo != nil {
 			job.TableName = job.BinlogInfo.TableInfo.Name.O
 		}
-		skip, err = p.filter.
-			ShouldDiscardDDL(job.StartTS, job.Type, job.SchemaName, job.TableName, job.Query)
-		if err != nil {
-			return false, errors.Trace(err)
-		}
+		skip = p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, job.TableName)
 	}
 
 	if skip {
diff --git a/cdc/puller/frontier/frontier.go b/cdc/puller/frontier/frontier.go
index f993dba2b..d01a3d01b 100644
--- a/cdc/puller/frontier/frontier.go
+++ b/cdc/puller/frontier/frontier.go
@@ -78,7 +78,7 @@ func (s *spanFrontier) Frontier() uint64 {
 func (s *spanFrontier) Forward(regionID uint64, span tablepb.Span, ts uint64) {
 	// it's the fast part to detect if the region is split or merged,
 	// if not we can update the minTsHeap with use new ts directly
-	if n, ok := s.cachedRegions[regionID]; ok && n.regionID != fakeRegionID && n.end != nil {
+	if n, ok := s.cachedRegions[regionID]; ok && n.regionID == regionID && n.end != nil {
 		if bytes.Equal(n.Key(), span.StartKey) && bytes.Equal(n.End(), span.EndKey) {
 			s.minTsHeap.UpdateKey(n.Value(), ts)
 			return
@@ -100,6 +100,7 @@ func (s *spanFrontier) insert(regionID uint64, span tablepb.Span, ts uint64) {
 		if bytes.Equal(seekRes.Node().Key(), span.StartKey) &&
 			bytes.Equal(next.Key(), span.EndKey) {
 			s.minTsHeap.UpdateKey(seekRes.Node().Value(), ts)
+			delete(s.cachedRegions, seekRes.Node().regionID)
 			if regionID != fakeRegionID {
 				s.cachedRegions[regionID] = seekRes.Node()
 				s.cachedRegions[regionID].regionID = regionID
diff --git a/cdc/scheduler/internal/scheduler.go b/cdc/scheduler/internal/scheduler.go
index 111911b56..28e241b9d 100644
--- a/cdc/scheduler/internal/scheduler.go
+++ b/cdc/scheduler/internal/scheduler.go
@@ -47,7 +47,7 @@ type Scheduler interface {
 		// ddl jobs that need to be replicated. The Scheduler will
 		// broadcast the barrierTs to all captures through the Heartbeat.
 		barrier *schedulepb.BarrierWithMinTs,
-	) (newCheckpointTs, newResolvedTs model.Ts, err error)
+	) (watermark schedulepb.Watermark, err error)
 
 	// MoveTable requests that a table be moved to target.
 	// It is thread-safe.
diff --git a/cdc/scheduler/internal/v3/coordinator.go b/cdc/scheduler/internal/v3/coordinator.go
index e27bdf49c..aaaa77300 100644
--- a/cdc/scheduler/internal/v3/coordinator.go
+++ b/cdc/scheduler/internal/v3/coordinator.go
@@ -126,7 +126,7 @@ func (c *coordinator) Tick(
 	// All captures that are alive according to the latest Etcd states.
 	aliveCaptures map[model.CaptureID]*model.CaptureInfo,
 	barrier *schedulepb.BarrierWithMinTs,
-) (newCheckpointTs, newResolvedTs model.Ts, err error) {
+) (watermark schedulepb.Watermark, err error) {
 	startTime := time.Now()
 	defer func() {
 		costTime := time.Since(startTime)
@@ -272,7 +272,7 @@ func (c *coordinator) poll(
 	currentTables []model.TableID,
 	aliveCaptures map[model.CaptureID]*model.CaptureInfo,
 	barrier *schedulepb.BarrierWithMinTs,
-) (newCheckpointTs, newResolvedTs model.Ts, err error) {
+) (watermark schedulepb.Watermark, err error) {
 	c.maybeCollectMetrics()
 	if c.compat.UpdateCaptureInfo(aliveCaptures) {
 		spanReplicationEnabled := c.compat.CheckSpanReplicationEnabled()
@@ -283,7 +283,12 @@ func (c *coordinator) poll(
 
 	recvMsgs, err := c.recvMsgs(ctx)
 	if err != nil {
-		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
+		}, errors.Trace(err)
 	}
 
 	var msgBuf []*schedulepb.Message
@@ -295,7 +300,12 @@ func (c *coordinator) poll(
 	// Handle received messages to advance replication set.
 	msgs, err = c.replicationM.HandleMessage(recvMsgs)
 	if err != nil {
-		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
+		}, errors.Trace(err)
 	}
 	msgBuf = append(msgBuf, msgs...)
 
@@ -309,13 +319,13 @@ func (c *coordinator) poll(
 	if !c.captureM.CheckAllCaptureInitialized() {
 		// Skip generating schedule tasks for replication manager,
 		// as not all capture are initialized.
-		newCheckpointTs, newResolvedTs = c.replicationM.AdvanceCheckpoint(&c.tableRanges, pdTime, barrier, c.redoMetaManager)
+		watermark = c.replicationM.AdvanceCheckpoint(&c.tableRanges, pdTime, barrier, c.redoMetaManager)
 		// tick capture manager after checkpoint calculation to take account resolvedTs in barrier
 		// when redo is enabled
 		msgs = c.captureM.Tick(c.replicationM.ReplicationSets(),
 			c.schedulerM.DrainingTarget(), barrier.Barrier)
 		msgBuf = append(msgBuf, msgs...)
-		return newCheckpointTs, newResolvedTs, c.sendMsgs(ctx, msgBuf)
+		return watermark, c.sendMsgs(ctx, msgBuf)
 	}
 
 	// Handle capture membership changes.
@@ -323,7 +333,12 @@ func (c *coordinator) poll(
 		msgs, err = c.replicationM.HandleCaptureChanges(
 			changes.Init, changes.Removed, checkpointTs)
 		if err != nil {
-			return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+			return schedulepb.Watermark{
+				CheckpointTs:     checkpointCannotProceed,
+				ResolvedTs:       checkpointCannotProceed,
+				LastSyncedTs:     checkpointCannotProceed,
+				PullerResolvedTs: checkpointCannotProceed,
+			}, errors.Trace(err)
 		}
 		msgBuf = append(msgBuf, msgs...)
 	}
@@ -339,12 +354,17 @@ func (c *coordinator) poll(
 	// Handle generated schedule tasks.
 	msgs, err = c.replicationM.HandleTasks(allTasks)
 	if err != nil {
-		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
+		}, errors.Trace(err)
 	}
 	msgBuf = append(msgBuf, msgs...)
 
 	// Checkpoint calculation
-	newCheckpointTs, newResolvedTs = c.replicationM.AdvanceCheckpoint(&c.tableRanges, pdTime, barrier, c.redoMetaManager)
+	watermark = c.replicationM.AdvanceCheckpoint(&c.tableRanges, pdTime, barrier, c.redoMetaManager)
 
 	// tick capture manager after checkpoint calculation to take account resolvedTs in barrier
 	// when redo is enabled
@@ -355,10 +375,15 @@ func (c *coordinator) poll(
 	// Send new messages.
 	err = c.sendMsgs(ctx, msgBuf)
 	if err != nil {
-		return checkpointCannotProceed, checkpointCannotProceed, errors.Trace(err)
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
+		}, errors.Trace(err)
 	}
 
-	return newCheckpointTs, newResolvedTs, nil
+	return watermark, nil
 }
 
 func (c *coordinator) recvMsgs(ctx context.Context) ([]*schedulepb.Message, error) {
diff --git a/cdc/scheduler/internal/v3/replication/replication_manager.go b/cdc/scheduler/internal/v3/replication/replication_manager.go
index 4228cc0ec..8196a06a0 100644
--- a/cdc/scheduler/internal/v3/replication/replication_manager.go
+++ b/cdc/scheduler/internal/v3/replication/replication_manager.go
@@ -557,9 +557,9 @@ func (r *Manager) AdvanceCheckpoint(
 	currentPDTime time.Time,
 	barrier *schedulepb.BarrierWithMinTs,
 	redoMetaManager redo.MetaManager,
-) (newCheckpointTs, newResolvedTs model.Ts) {
+) (watermark schedulepb.Watermark) {
 	var redoFlushedResolvedTs model.Ts
-	limitBarrierWithRedo := func(newCheckpointTs, newResolvedTs uint64) (uint64, uint64) {
+	limitBarrierWithRedo := func(watermark *schedulepb.Watermark) {
 		flushedMeta := redoMetaManager.GetFlushedMeta()
 		redoFlushedResolvedTs = flushedMeta.ResolvedTs
 		log.Debug("owner gets flushed redo meta",
@@ -567,18 +567,17 @@ func (r *Manager) AdvanceCheckpoint(
 			zap.String("changefeed", r.changefeedID.ID),
 			zap.Uint64("flushedCheckpointTs", flushedMeta.CheckpointTs),
 			zap.Uint64("flushedResolvedTs", flushedMeta.ResolvedTs))
-		if flushedMeta.ResolvedTs < newResolvedTs {
-			newResolvedTs = flushedMeta.ResolvedTs
+		if flushedMeta.ResolvedTs < watermark.ResolvedTs {
+			watermark.ResolvedTs = flushedMeta.ResolvedTs
 		}
 
-		if newCheckpointTs > newResolvedTs {
-			newCheckpointTs = newResolvedTs
+		if watermark.CheckpointTs > watermark.ResolvedTs {
+			watermark.CheckpointTs = watermark.ResolvedTs
 		}
 
-		if barrier.GlobalBarrierTs > newResolvedTs {
-			barrier.GlobalBarrierTs = newResolvedTs
+		if barrier.GlobalBarrierTs > watermark.ResolvedTs {
+			barrier.GlobalBarrierTs = watermark.ResolvedTs
 		}
-		return newCheckpointTs, newResolvedTs
 	}
 	defer func() {
 		if redoFlushedResolvedTs != 0 && barrier.GlobalBarrierTs > redoFlushedResolvedTs {
@@ -592,9 +591,14 @@ func (r *Manager) AdvanceCheckpoint(
 
 	r.slowestPuller = tablepb.Span{}
 	r.slowestSink = tablepb.Span{}
-	var slowestPullerResolvedTs uint64 = math.MaxUint64
 
-	newCheckpointTs, newResolvedTs = math.MaxUint64, math.MaxUint64
+	watermark = schedulepb.Watermark{
+		CheckpointTs:     math.MaxUint64,
+		ResolvedTs:       math.MaxUint64,
+		LastSyncedTs:     0,
+		PullerResolvedTs: math.MaxUint64,
+	}
+
 	cannotProceed := false
 	currentTables.Iter(func(tableID model.TableID, tableStart, tableEnd tablepb.Span) bool {
 		tableSpanFound, tableHasHole := false, false
@@ -621,21 +625,26 @@ func (r *Manager) AdvanceCheckpoint(
 				}
 
 				// Find the minimum checkpoint ts and resolved ts.
-				if newCheckpointTs > table.Checkpoint.CheckpointTs {
-					newCheckpointTs = table.Checkpoint.CheckpointTs
+				if watermark.CheckpointTs > table.Checkpoint.CheckpointTs {
+					watermark.CheckpointTs = table.Checkpoint.CheckpointTs
 					r.slowestSink = span
 				}
-				if newResolvedTs > table.Checkpoint.ResolvedTs {
-					newResolvedTs = table.Checkpoint.ResolvedTs
+				if watermark.ResolvedTs > table.Checkpoint.ResolvedTs {
+					watermark.ResolvedTs = table.Checkpoint.ResolvedTs
 				}
 
+				// Find the max lastSyncedTs of all tables.
+				if watermark.LastSyncedTs < table.Checkpoint.LastSyncedTs {
+					watermark.LastSyncedTs = table.Checkpoint.LastSyncedTs
+				}
 				// Find the minimum puller resolved ts.
 				if pullerCkpt, ok := table.Stats.StageCheckpoints["puller-egress"]; ok {
-					if slowestPullerResolvedTs > pullerCkpt.ResolvedTs {
-						slowestPullerResolvedTs = pullerCkpt.ResolvedTs
+					if watermark.PullerResolvedTs > pullerCkpt.ResolvedTs {
+						watermark.PullerResolvedTs = pullerCkpt.ResolvedTs
 						r.slowestPuller = span
 					}
 				}
+
 				return true
 			})
 		if !tableSpanFound || !tableSpanStartFound || !tableSpanEndFound || tableHasHole {
@@ -663,32 +672,39 @@ func (r *Manager) AdvanceCheckpoint(
 	if cannotProceed {
 		if redoMetaManager.Enabled() {
 			// If redo is enabled, GlobalBarrierTs should be limited by redo flushed meta.
-			newResolvedTs = barrier.RedoBarrierTs
-			limitBarrierWithRedo(newCheckpointTs, newResolvedTs)
+			watermark.ResolvedTs = barrier.RedoBarrierTs
+			watermark.LastSyncedTs = checkpointCannotProceed
+			watermark.PullerResolvedTs = checkpointCannotProceed
+			limitBarrierWithRedo(&watermark)
+		}
+		return schedulepb.Watermark{
+			CheckpointTs:     checkpointCannotProceed,
+			ResolvedTs:       checkpointCannotProceed,
+			LastSyncedTs:     checkpointCannotProceed,
+			PullerResolvedTs: checkpointCannotProceed,
 		}
-		return checkpointCannotProceed, checkpointCannotProceed
 	}
 
 	// If currentTables is empty, we should advance newResolvedTs to global barrier ts and
 	// advance newCheckpointTs to min table barrier ts.
-	if newResolvedTs == math.MaxUint64 || newCheckpointTs == math.MaxUint64 {
-		if newCheckpointTs != newResolvedTs || currentTables.Len() != 0 {
+	if watermark.ResolvedTs == math.MaxUint64 || watermark.CheckpointTs == math.MaxUint64 {
+		if watermark.CheckpointTs != watermark.ResolvedTs || currentTables.Len() != 0 {
 			log.Panic("schedulerv3: newCheckpointTs and newResolvedTs should be both maxUint64 "+
 				"if currentTables is empty",
-				zap.Uint64("newCheckpointTs", newCheckpointTs),
-				zap.Uint64("newResolvedTs", newResolvedTs),
+				zap.Uint64("newCheckpointTs", watermark.CheckpointTs),
+				zap.Uint64("newResolvedTs", watermark.ResolvedTs),
 				zap.Any("currentTables", currentTables))
 		}
-		newResolvedTs = barrier.GlobalBarrierTs
-		newCheckpointTs = barrier.MinTableBarrierTs
+		watermark.ResolvedTs = barrier.GlobalBarrierTs
+		watermark.CheckpointTs = barrier.MinTableBarrierTs
 	}
 
-	if newCheckpointTs > barrier.MinTableBarrierTs {
-		newCheckpointTs = barrier.MinTableBarrierTs
+	if watermark.CheckpointTs > barrier.MinTableBarrierTs {
+		watermark.CheckpointTs = barrier.MinTableBarrierTs
 		// TODO: add panic after we fix the bug that newCheckpointTs > minTableBarrierTs.
 		// log.Panic("schedulerv3: newCheckpointTs should not be larger than minTableBarrierTs",
-		// 	zap.Uint64("newCheckpointTs", newCheckpointTs),
-		// 	zap.Uint64("newResolvedTs", newResolvedTs),
+		// 	zap.Uint64("newCheckpointTs", watermark.CheckpointTs),
+		// 	zap.Uint64("newResolvedTs", watermark.ResolvedTs),
 		// 	zap.Any("currentTables", currentTables.currentTables),
 		// 	zap.Any("barrier", barrier.Barrier),
 		// 	zap.Any("minTableBarrierTs", barrier.MinTableBarrierTs))
@@ -697,7 +713,7 @@ func (r *Manager) AdvanceCheckpoint(
 	// If changefeed's checkpoint lag is larger than 30s,
 	// log the 4 slowlest table infos every minute, which can
 	// help us find the problematic tables.
-	checkpointLag := currentPDTime.Sub(oracle.GetTimeFromTS(newCheckpointTs))
+	checkpointLag := currentPDTime.Sub(oracle.GetTimeFromTS(watermark.CheckpointTs))
 	if checkpointLag > logSlowTablesLagThreshold &&
 		time.Since(r.lastLogSlowTablesTime) > logSlowTablesInterval {
 		r.logSlowTableInfo(currentPDTime)
@@ -705,19 +721,19 @@ func (r *Manager) AdvanceCheckpoint(
 	}
 
 	if redoMetaManager.Enabled() {
-		if newResolvedTs > barrier.RedoBarrierTs {
-			newResolvedTs = barrier.RedoBarrierTs
+		if watermark.ResolvedTs > barrier.RedoBarrierTs {
+			watermark.ResolvedTs = barrier.RedoBarrierTs
 		}
-		redoMetaManager.UpdateMeta(newCheckpointTs, newResolvedTs)
+		redoMetaManager.UpdateMeta(watermark.CheckpointTs, watermark.ResolvedTs)
 		log.Debug("owner updates redo meta",
 			zap.String("namespace", r.changefeedID.Namespace),
 			zap.String("changefeed", r.changefeedID.ID),
-			zap.Uint64("newCheckpointTs", newCheckpointTs),
-			zap.Uint64("newResolvedTs", newResolvedTs))
-		return limitBarrierWithRedo(newCheckpointTs, newResolvedTs)
+			zap.Uint64("newCheckpointTs", watermark.CheckpointTs),
+			zap.Uint64("newResolvedTs", watermark.ResolvedTs))
+		limitBarrierWithRedo(&watermark)
 	}
 
-	return newCheckpointTs, newResolvedTs
+	return watermark
 }
 
 func (r *Manager) logSlowTableInfo(currentPDTime time.Time) {
diff --git a/cdc/scheduler/internal/v3/replication/replication_set.go b/cdc/scheduler/internal/v3/replication/replication_set.go
index fec432499..ea1a93f41 100644
--- a/cdc/scheduler/internal/v3/replication/replication_set.go
+++ b/cdc/scheduler/internal/v3/replication/replication_set.go
@@ -1020,6 +1020,11 @@ func (r *ReplicationSet) updateCheckpointAndStats(
 			zap.Any("checkpointTs", r.Checkpoint.CheckpointTs),
 			zap.Any("resolvedTs", r.Checkpoint.ResolvedTs))
 	}
+
+	if r.Checkpoint.LastSyncedTs < checkpoint.LastSyncedTs {
+		r.Checkpoint.LastSyncedTs = checkpoint.LastSyncedTs
+	}
+
 	r.Stats = stats
 }
 
diff --git a/cdc/scheduler/schedulepb/watermark.go b/cdc/scheduler/schedulepb/watermark.go
new file mode 100644
index 000000000..81123734e
--- /dev/null
+++ b/cdc/scheduler/schedulepb/watermark.go
@@ -0,0 +1,24 @@
+// Copyright 2023 PingCAP, Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package schedulepb
+
+import "github.com/pingcap/tiflow/cdc/model"
+
+// Watermark contains various ts variables to make code easier
+type Watermark struct {
+	CheckpointTs     model.Ts
+	ResolvedTs       model.Ts
+	LastSyncedTs     model.Ts
+	PullerResolvedTs model.Ts
+}
diff --git a/cdc/sink/dmlsink/txn/event.go b/cdc/sink/dmlsink/txn/event.go
index 18ea2090f..52dcf90e3 100644
--- a/cdc/sink/dmlsink/txn/event.go
+++ b/cdc/sink/dmlsink/txn/event.go
@@ -41,11 +41,44 @@ func (e *txnEvent) OnConflictResolved() {
 	e.conflictResolved = time.Now()
 }
 
-// ConflictKeys implements causality.txnEvent interface.
-func (e *txnEvent) ConflictKeys(numSlots uint64) []uint64 {
-	keys := genTxnKeys(e.TxnCallbackableEvent.Event)
-	sort.Slice(keys, func(i, j int) bool { return keys[i]%numSlots < keys[j]%numSlots })
-	return keys
+// GenSortedDedupKeysHash implements causality.txnEvent interface.
+func (e *txnEvent) GenSortedDedupKeysHash(numSlots uint64) []uint64 {
+	hashes := genTxnKeys(e.TxnCallbackableEvent.Event)
+
+	// Sort and dedup hashes.
+	// Sort hashes by `hash % numSlots` to avoid deadlock, and then dedup
+	// hashes, so the same txn will not check confict with the same hash twice to
+	// prevent potential cyclic self dependency in the causality dependency
+	// graph.
+	return sortAndDedupHashes(hashes, numSlots)
+}
+
+func sortAndDedupHashes(hashes []uint64, numSlots uint64) []uint64 {
+	if len(hashes) == 0 {
+		return nil
+	}
+
+	// Sort hashes by `hash % numSlots` to avoid deadlock.
+	sort.Slice(hashes, func(i, j int) bool { return hashes[i]%numSlots < hashes[j]%numSlots })
+
+	// Dedup hashes
+	last := hashes[0]
+	j := 1
+	for i, hash := range hashes {
+		if i == 0 {
+			// skip first one, start checking duplication from 2nd one
+			continue
+		}
+		if hash == last {
+			continue
+		}
+		last = hash
+		hashes[j] = hash
+		j++
+	}
+	hashes = hashes[:j]
+
+	return hashes
 }
 
 // genTxnKeys returns hash keys for `txn`.
diff --git a/cdc/sink/dmlsink/txn/worker.go b/cdc/sink/dmlsink/txn/worker.go
index 673a53b5a..6fae294dd 100644
--- a/cdc/sink/dmlsink/txn/worker.go
+++ b/cdc/sink/dmlsink/txn/worker.go
@@ -29,7 +29,7 @@ import (
 
 type txnWithNotifier struct {
 	*txnEvent
-	wantMore func()
+	postTxnExecuted func()
 }
 
 type worker struct {
@@ -49,9 +49,9 @@ type worker struct {
 	metricTxnWorkerHandledRows   prometheus.Counter
 
 	// Fields only used in the background loop.
-	flushInterval     time.Duration
-	hasPending        bool
-	wantMoreCallbacks []func()
+	flushInterval            time.Duration
+	hasPending               bool
+	postTxnExecutedCallbacks []func()
 }
 
 func newWorker(ctx context.Context, changefeedID model.ChangeFeedID,
@@ -73,17 +73,19 @@ func newWorker(ctx context.Context, changefeedID model.ChangeFeedID,
 		metricTxnWorkerBusyRatio:     txn.WorkerBusyRatio.WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 		metricTxnWorkerHandledRows:   txn.WorkerHandledRows.WithLabelValues(changefeedID.Namespace, changefeedID.ID, wid),
 
-		flushInterval:     backend.MaxFlushInterval(),
-		hasPending:        false,
-		wantMoreCallbacks: make([]func(), 0, 1024),
+		flushInterval:            backend.MaxFlushInterval(),
+		hasPending:               false,
+		postTxnExecutedCallbacks: make([]func(), 0, 1024),
 	}
 }
 
 // Add adds a txnEvent to the worker.
-// The worker will call unlock() when it's ready to receive more events.
-// In other words, it maybe advances the conflict detector.
-func (w *worker) Add(txn *txnEvent, unlock func()) {
-	w.txnCh.In() <- txnWithNotifier{txn, unlock}
+// The worker will call postTxnExecuted() after the txn executed.
+// The postTxnExecuted will remove the txn related Node in the conflict detector's
+// dependency graph and resolve related dependencies for these transacitons
+// which depend on this executed txn.
+func (w *worker) Add(txn *txnEvent, postTxnExecuted func()) {
+	w.txnCh.In() <- txnWithNotifier{txn, postTxnExecuted}
 }
 
 func (w *worker) close() {
@@ -154,15 +156,15 @@ func (w *worker) onEvent(txn txnWithNotifier) bool {
 		// The table where the event comes from is in stopping, so it's safe
 		// to drop the event directly.
 		txn.txnEvent.Callback()
-		// Still necessary to append the wantMore callback into the pending list.
-		w.wantMoreCallbacks = append(w.wantMoreCallbacks, txn.wantMore)
+		// Still necessary to append the callbacks into the pending list.
+		w.postTxnExecutedCallbacks = append(w.postTxnExecutedCallbacks, txn.postTxnExecuted)
 		return false
 	}
 
 	w.metricConflictDetectDuration.Observe(txn.conflictResolved.Sub(txn.start).Seconds())
 	w.metricQueueDuration.Observe(time.Since(txn.start).Seconds())
 	w.metricTxnWorkerHandledRows.Add(float64(len(txn.Event.Rows)))
-	w.wantMoreCallbacks = append(w.wantMoreCallbacks, txn.wantMore)
+	w.postTxnExecutedCallbacks = append(w.postTxnExecutedCallbacks, txn.postTxnExecuted)
 	return w.backend.OnTxnEvent(txn.txnEvent.TxnCallbackableEvent)
 }
 
@@ -185,13 +187,13 @@ func (w *worker) doFlush(flushTimeSlice *time.Duration) error {
 			return err
 		}
 		// Flush successfully, call callbacks to notify conflict detector.
-		for _, wantMore := range w.wantMoreCallbacks {
-			wantMore()
+		for _, postTxnExecuted := range w.postTxnExecutedCallbacks {
+			postTxnExecuted()
 		}
-		w.wantMoreCallbacks = w.wantMoreCallbacks[:0]
-		if cap(w.wantMoreCallbacks) > 1024 {
+		w.postTxnExecutedCallbacks = w.postTxnExecutedCallbacks[:0]
+		if cap(w.postTxnExecutedCallbacks) > 1024 {
 			// Resize the buffer if it's too big.
-			w.wantMoreCallbacks = make([]func(), 0, 1024)
+			w.postTxnExecutedCallbacks = make([]func(), 0, 1024)
 		}
 	}
 
diff --git a/cdc/sink/tablesink/table_sink.go b/cdc/sink/tablesink/table_sink.go
index 69f6f61db..e5e8cf19b 100644
--- a/cdc/sink/tablesink/table_sink.go
+++ b/cdc/sink/tablesink/table_sink.go
@@ -32,6 +32,11 @@ type TableSink interface {
 	// For example, calculating the current progress from the statistics of the table sink.
 	// This is a thread-safe method.
 	GetCheckpointTs() model.ResolvedTs
+	// GetLastSyncedTs returns the last synced ts of table sink.
+	// the last synced ts means the biggest commits of the events
+	// that have been flushed to the downstream.
+	// This is a thread-safe method.
+	GetLastSyncedTs() model.Ts
 	// Close closes the table sink.
 	// After it returns, no more events will be sent out from this capture.
 	Close()
diff --git a/cdc/sink/tablesink/table_sink_impl.go b/cdc/sink/tablesink/table_sink_impl.go
index 5ee1b869d..76b65cc77 100644
--- a/cdc/sink/tablesink/table_sink_impl.go
+++ b/cdc/sink/tablesink/table_sink_impl.go
@@ -15,6 +15,7 @@ package tablesink
 
 import (
 	"sort"
+	"sync"
 
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -31,6 +32,21 @@ var (
 	_ TableSink = (*EventTableSink[*model.SingleTableTxn, *dmlsink.TxnEventAppender])(nil)
 )
 
+// LastSyncedTsRecord is used to record the last synced ts of table sink with lock
+// lastSyncedTs means the biggest commits of the events
+// that have been flushed to the downstream.
+type LastSyncedTsRecord struct {
+	sync.Mutex
+	lastSyncedTs model.Ts
+}
+
+// getLastSyncedTs get value from LastSyncedTsRecord
+func (r *LastSyncedTsRecord) getLastSyncedTs() model.Ts {
+	r.Lock()
+	defer r.Unlock()
+	return r.lastSyncedTs
+}
+
 // EventTableSink is a table sink that can write events.
 type EventTableSink[E dmlsink.TableEvent, P dmlsink.Appender[E]] struct {
 	changefeedID model.ChangeFeedID
@@ -46,6 +62,8 @@ type EventTableSink[E dmlsink.TableEvent, P dmlsink.Appender[E]] struct {
 	eventBuffer []E
 	state       state.TableSinkState
 
+	lastSyncedTs LastSyncedTsRecord
+
 	// For dataflow metrics.
 	metricsTableSinkTotalRows prometheus.Counter
 }
@@ -69,6 +87,7 @@ func New[E dmlsink.TableEvent, P dmlsink.Appender[E]](
 		eventAppender:             appender,
 		eventBuffer:               make([]E, 0, 1024),
 		state:                     state.TableSinkSinking,
+		lastSyncedTs:              LastSyncedTsRecord{lastSyncedTs: startTs},
 		metricsTableSinkTotalRows: totalRowsCounter,
 	}
 }
@@ -114,9 +133,24 @@ func (e *EventTableSink[E, P]) UpdateResolvedTs(resolvedTs model.ResolvedTs) err
 			return SinkInternalError{err}
 		}
 		// We have to record the event ID for the callback.
+		postEventFlushFunc := e.progressTracker.addEvent()
+		evCommitTs := ev.GetCommitTs()
 		ce := &dmlsink.CallbackableEvent[E]{
-			Event:     ev,
-			Callback:  e.progressTracker.addEvent(),
+			Event: ev,
+			Callback: func() {
+				// Due to multi workers will call this callback concurrently,
+				// we need to add lock to protect lastSyncedTs
+				// we need make a performance test for it
+				{
+					e.lastSyncedTs.Lock()
+					defer e.lastSyncedTs.Unlock()
+
+					if e.lastSyncedTs.lastSyncedTs < evCommitTs {
+						e.lastSyncedTs.lastSyncedTs = evCommitTs
+					}
+				}
+				postEventFlushFunc()
+			},
 			SinkState: &e.state,
 		}
 		resolvedCallbackableEvents = append(resolvedCallbackableEvents, ce)
@@ -140,6 +174,13 @@ func (e *EventTableSink[E, P]) GetCheckpointTs() model.ResolvedTs {
 	return e.progressTracker.advance()
 }
 
+// GetLastSyncedTs returns the last synced ts of table sink.
+// lastSyncedTs means the biggest commits of all the events
+// that have been flushed to the downstream.
+func (e *EventTableSink[E, P]) GetLastSyncedTs() model.Ts {
+	return e.lastSyncedTs.getLastSyncedTs()
+}
+
 // Close closes the table sink.
 // After it returns, no more events will be sent out from this capture.
 func (e *EventTableSink[E, P]) Close() {
diff --git a/cmd/filter-helper/main.go b/cmd/filter-helper/main.go
index c3286bc5d..d6c11a623 100644
--- a/cmd/filter-helper/main.go
+++ b/cmd/filter-helper/main.go
@@ -18,6 +18,7 @@ import (
 	"strings"
 
 	timodel "github.com/pingcap/tidb/parser/model"
+	"github.com/pingcap/tiflow/cdc/model"
 	"github.com/pingcap/tiflow/pkg/cmd/util"
 	"github.com/pingcap/tiflow/pkg/config"
 	"github.com/pingcap/tiflow/pkg/filter"
@@ -79,13 +80,8 @@ func runFilter(cmd *cobra.Command, args []string) {
 		}
 		fmt.Printf("Table: %s, Not matched filter rule\n", table)
 	case "ddl":
-		startTs := uint64(0)
 		ddlType := timodel.ActionCreateTable
-		discard, err := ft.ShouldDiscardDDL(startTs,
-			ddlType,
-			tableAndSchema[0],
-			tableAndSchema[1],
-			ddl)
+		discard := ft.ShouldDiscardDDL(ddlType, tableAndSchema[0], tableAndSchema[1])
 		if err != nil {
 			fmt.Printf("filter ddl error: %s, error: %v\n", ddl, err)
 			return
@@ -94,6 +90,25 @@ func runFilter(cmd *cobra.Command, args []string) {
 			fmt.Printf("DDL: %s, should be discard by event filter rule\n", ddl)
 			return
 		}
+		ignored, err := ft.ShouldIgnoreDDLEvent(&model.DDLEvent{
+			StartTs: uint64(0),
+			Query:   ddl,
+			Type:    ddlType,
+			TableInfo: &model.TableInfo{
+				TableName: model.TableName{
+					Schema: tableAndSchema[0],
+					Table:  tableAndSchema[1],
+				},
+			},
+		})
+		if err != nil {
+			fmt.Printf("filter ddl error: %s, error: %v\n", ddl, err)
+			return
+		}
+		if ignored {
+			fmt.Printf("DDL: %s, should be ignored by event filter rule\n", ddl)
+			return
+		}
 		fmt.Printf("DDL: %s, should not be discard by event filter rule\n", ddl)
 	default:
 		fmt.Printf("unknown target: %s", target)
diff --git a/go.mod b/go.mod
index ee2f2b3ac..d029cfc69 100644
--- a/go.mod
+++ b/go.mod
@@ -382,3 +382,9 @@ replace go.opencensus.io => go.opencensus.io v0.23.1-0.20220331163232-052120675f
 replace github.com/go-ldap/ldap/v3 v3.4.4 => github.com/yangkeao/ldap/v3 v3.4.5-0.20230421065457-369a3bab1117
 
 replace github.com/tildeleb/hashland => leb.io/hashland v0.1.5
+
+// TODO: `sourcegraph.com/sourcegraph/appdash` has been archived, and the original host has been removed.
+// Please remove these dependencies.
+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
diff --git a/go.sum b/go.sum
index 6c06a6def..b6ea8cbc9 100644
--- a/go.sum
+++ b/go.sum
@@ -1184,6 +1184,10 @@ github.com/smartystreets/goconvey v1.6.4/go.mod h1:syvi0/a8iFYH4r/RixwvyeAJjdLS9
 github.com/soheilhy/cmux v0.1.4/go.mod h1:IM3LyeVVIOuxMH7sFAkER9+bJ4dT7Ms6E4xg4kGIyLM=
 github.com/soheilhy/cmux v0.1.5 h1:jjzc5WVemNEDTLwv9tlmemhC73tI08BNOIGwBOo10Js=
 github.com/soheilhy/cmux v0.1.5/go.mod h1:T7TcVDs9LWfQgPlPsdngu6I6QIoyIFZDDC6sNE1GqG0=
+github.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0 h1:IJ3DuWHPTJrsqtIqjfdmPTELdTFGefvrOa2eTeRBleQ=
+github.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0/go.mod h1:V952P4GGl1v/MMynLwxVdWEbSZJx+n0oOO3ljnez+WU=
+github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67 h1:8ZnTA26bBOoPkAbbitKPgNlpw0Bwt7ZlpYgZWHWJR/w=
+github.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67/go.mod h1:tNZjgbYncKL5HxvDULAr/mWDmFz4B7H8yrXEDlnoIiw=
 github.com/spaolacci/murmur3 v0.0.0-20180118202830-f09979ecbc72/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
 github.com/spaolacci/murmur3 v1.1.0 h1:7c1g84S4BPRrfL5Xrdp6fOJ206sU9y293DDHaoy0bLI=
 github.com/spaolacci/murmur3 v1.1.0/go.mod h1:JwIasOWyU6f++ZhiEuf87xNszmSA2myDM2Kzu9HwQUA=
@@ -2133,9 +2137,5 @@ sigs.k8s.io/structured-merge-diff/v4 v4.2.3/go.mod h1:qjx8mGObPmV2aSZepjQjbmb2ih
 sigs.k8s.io/yaml v1.2.0/go.mod h1:yfXDCHCao9+ENCvLSE62v9VSji2MKu5jeNfTrofGhJc=
 sigs.k8s.io/yaml v1.3.0 h1:a2VclLzOGrwOHDiV8EfBGhvjHvP46CtW5j6POvhYGGo=
 sigs.k8s.io/yaml v1.3.0/go.mod h1:GeOyir5tyXNByN85N/dRIT9es5UQNerPYEKK56eTBm8=
-sourcegraph.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0 h1:ucqkfpjg9WzSUubAO62csmucvxl4/JeW3F4I4909XkM=
-sourcegraph.com/sourcegraph/appdash v0.0.0-20190731080439-ebfcffb1b5c0/go.mod h1:hI742Nqp5OhwiqlzhgfbWU4mW4yO10fP+LoT9WOswdU=
-sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67 h1:e1sMhtVq9AfcEy8AXNb8eSg6gbzfdpYhoNqnPJa+GzI=
-sourcegraph.com/sourcegraph/appdash-data v0.0.0-20151005221446-73f23eafcf67/go.mod h1:L5q+DGLGOQFpo1snNEkLOJT2d1YTW66rWNzatr3He1k=
 upper.io/db.v3 v3.7.1+incompatible h1:GiK/NmDUClH3LrZd54qj5OQsz8brGFv652QXyRXtg2U=
 upper.io/db.v3 v3.7.1+incompatible/go.mod h1:FgTdD24eBjJAbPKsQSiHUNgXjOR4Lub3u1UMHSIh82Y=
diff --git a/pkg/causality/conflict_detector.go b/pkg/causality/conflict_detector.go
index 641dde8b8..f1f7495e7 100644
--- a/pkg/causality/conflict_detector.go
+++ b/pkg/causality/conflict_detector.go
@@ -44,8 +44,8 @@ type ConflictDetector[Worker worker[Txn], Txn txnEvent] struct {
 }
 
 type txnFinishedEvent struct {
-	node         *internal.Node
-	conflictKeys []uint64
+	node           *internal.Node
+	sortedKeysHash []uint64
 }
 
 // NewConflictDetector creates a new ConflictDetector.
@@ -73,20 +73,29 @@ func NewConflictDetector[Worker worker[Txn], Txn txnEvent](
 
 // Add pushes a transaction to the ConflictDetector.
 //
-// NOTE: if multiple threads access this concurrently, Txn.ConflictKeys must be sorted.
+// NOTE: if multiple threads access this concurrently,
+// Txn.GenSortedDedupKeysHash must be sorted by the slot index.
 func (d *ConflictDetector[Worker, Txn]) Add(txn Txn) {
-	conflictKeys := txn.ConflictKeys(d.numSlots)
+	sortedKeysHash := txn.GenSortedDedupKeysHash(d.numSlots)
 	node := internal.NewNode()
 	node.OnResolved = func(workerID int64) {
-		unlock := func() {
+		// This callback is called after the transaction is executed.
+		postTxnExecuted := func() {
+			// After this transaction is executed, we can remove the node from the graph,
+			// and resolve related dependencies for these transacitons which depend on this
+			// executed transaction.
 			node.Remove()
-			d.garbageNodes.In() <- txnFinishedEvent{node, conflictKeys}
+
+			// Send this node to garbageNodes to GC it from the slots if this node is still
+			// occupied related slots.
+			d.garbageNodes.In() <- txnFinishedEvent{node, sortedKeysHash}
 		}
-		d.sendToWorker(txn, unlock, workerID)
+		// Send this txn to related worker as soon as all dependencies are resolved.
+		d.sendToWorker(txn, postTxnExecuted, workerID)
 	}
 	node.RandWorkerID = func() int64 { return d.nextWorkerID.Add(1) % int64(len(d.workers)) }
 	node.OnNotified = func(callback func()) { d.notifiedNodes.In() <- callback }
-	d.slots.Add(node, conflictKeys)
+	d.slots.Add(node, sortedKeysHash)
 }
 
 // Close closes the ConflictDetector.
@@ -110,18 +119,18 @@ func (d *ConflictDetector[Worker, Txn]) runBackgroundTasks() {
 			}
 		case event := <-d.garbageNodes.Out():
 			if event.node != nil {
-				d.slots.Free(event.node, event.conflictKeys)
+				d.slots.Free(event.node, event.sortedKeysHash)
 			}
 		}
 	}
 }
 
 // sendToWorker should not call txn.Callback if it returns an error.
-func (d *ConflictDetector[Worker, Txn]) sendToWorker(txn Txn, unlock func(), workerID int64) {
+func (d *ConflictDetector[Worker, Txn]) sendToWorker(txn Txn, postTxnExecuted func(), workerID int64) {
 	if workerID < 0 {
 		panic("must assign with a valid workerID")
 	}
 	txn.OnConflictResolved()
 	worker := d.workers[workerID]
-	worker.Add(txn, unlock)
+	worker.Add(txn, postTxnExecuted)
 }
diff --git a/pkg/causality/internal/node.go b/pkg/causality/internal/node.go
index c9c7a3a29..cfcd5fc2f 100644
--- a/pkg/causality/internal/node.go
+++ b/pkg/causality/internal/node.go
@@ -48,7 +48,7 @@ type Node struct {
 	// Immutable fields.
 	id int64
 
-	// Set the callback that the node is resolved.
+	// Called when all dependencies are resolved.
 	OnResolved func(id workerID)
 	// Set the id generator to get a random ID.
 	RandWorkerID func() workerID
@@ -56,10 +56,10 @@ type Node struct {
 	OnNotified func(callback func())
 
 	// Following fields are used for notifying a node's dependers lock-free.
-	totalDependees    int32
-	resolvedDependees int32
-	removedDependees  int32
-	resolvedList      []int64
+	totalDependencies    int32
+	resolvedDependencies int32
+	removedDependencies  int32
+	resolvedList         []int64
 
 	// Following fields are protected by `mu`.
 	mu sync.Mutex
@@ -87,9 +87,9 @@ func NewNode() (ret *Node) {
 		ret.id = genNextNodeID()
 		ret.OnResolved = nil
 		ret.RandWorkerID = nil
-		ret.totalDependees = 0
-		ret.resolvedDependees = 0
-		ret.removedDependees = 0
+		ret.totalDependencies = 0
+		ret.resolvedDependencies = 0
+		ret.removedDependencies = 0
 		ret.resolvedList = nil
 		ret.assignedTo = unassigned
 		ret.removed = false
@@ -105,36 +105,43 @@ func (n *Node) NodeID() int64 {
 }
 
 // DependOn implements interface internal.SlotNode.
-func (n *Node) DependOn(unresolvedDeps map[int64]*Node, resolvedDeps int) {
-	resolvedDependees, removedDependees := int32(0), int32(0)
+func (n *Node) DependOn(dependencyNodes map[int64]*Node, noDependencyKeyCnt int) {
+	resolvedDependencies, removedDependencies := int32(0), int32(0)
 
 	depend := func(target *Node) {
 		if target == nil {
 			// For a given Node, every dependency corresponds to a target.
 			// If target is nil it means the dependency doesn't conflict
 			// with any other nodes. However it's still necessary to track
-			// it because Node.tryResolve needs to know it.
-			resolvedDependees = stdatomic.AddInt32(&n.resolvedDependees, 1)
-			stdatomic.StoreInt64(&n.resolvedList[resolvedDependees-1], assignedToAny)
-			removedDependees = stdatomic.AddInt32(&n.removedDependees, 1)
+			// it because Node.tryResolve needs to counting the number of
+			// resolved dependencies.
+			resolvedDependencies = stdatomic.AddInt32(&n.resolvedDependencies, 1)
+			stdatomic.StoreInt64(&n.resolvedList[resolvedDependencies-1], assignedToAny)
+			removedDependencies = stdatomic.AddInt32(&n.removedDependencies, 1)
 			return
 		}
 
 		if target.id == n.id {
 			panic("you cannot depend on yourself")
 		}
-		// Lock target and insert `n` into target.dependers.
+
+		// The target node might be removed or modified in other places, for example
+		// after its corresponding transaction has been executed.
 		target.mu.Lock()
 		defer target.mu.Unlock()
 
 		if target.assignedTo != unassigned {
 			// The target has already been assigned to a worker.
-			resolvedDependees = stdatomic.AddInt32(&n.resolvedDependees, 1)
-			stdatomic.StoreInt64(&n.resolvedList[resolvedDependees-1], target.assignedTo)
+			// In this case, record the worker ID in `resolvedList`, and this node
+			// probably can be sent to the same worker and executed sequentially.
+			resolvedDependencies = stdatomic.AddInt32(&n.resolvedDependencies, 1)
+			stdatomic.StoreInt64(&n.resolvedList[resolvedDependencies-1], target.assignedTo)
 		}
+
+		// Add the node to the target's dependers if the target has not been removed.
 		if target.removed {
 			// The target has already been removed.
-			removedDependees = stdatomic.AddInt32(&n.removedDependees, 1)
+			removedDependencies = stdatomic.AddInt32(&n.removedDependencies, 1)
 		} else if _, exist := target.getOrCreateDependers().ReplaceOrInsert(n); exist {
 			// Should never depend on a target redundantly.
 			panic("should never exist")
@@ -143,23 +150,24 @@ func (n *Node) DependOn(unresolvedDeps map[int64]*Node, resolvedDeps int) {
 
 	// Re-allocate ID in `DependOn` instead of creating the node, because the node can be
 	// pending in slots after it's created.
+	// ?: why gen new ID here?
 	n.id = genNextNodeID()
 
-	// `totalDependees` and `resolvedList` must be initialized before depending on any targets.
-	n.totalDependees = int32(len(unresolvedDeps) + resolvedDeps)
-	n.resolvedList = make([]int64, 0, n.totalDependees)
-	for i := 0; i < int(n.totalDependees); i++ {
+	// `totalDependcies` and `resolvedList` must be initialized before depending on any targets.
+	n.totalDependencies = int32(len(dependencyNodes) + noDependencyKeyCnt)
+	n.resolvedList = make([]int64, 0, n.totalDependencies)
+	for i := 0; i < int(n.totalDependencies); i++ {
 		n.resolvedList = append(n.resolvedList, unassigned)
 	}
 
-	for _, target := range unresolvedDeps {
-		depend(target)
+	for _, node := range dependencyNodes {
+		depend(node)
 	}
-	for i := 0; i < resolvedDeps; i++ {
+	for i := 0; i < noDependencyKeyCnt; i++ {
 		depend(nil)
 	}
 
-	n.maybeResolve(resolvedDependees, removedDependees)
+	n.maybeResolve(resolvedDependencies, removedDependencies)
 }
 
 // Remove implements interface internal.SlotNode.
@@ -171,8 +179,8 @@ func (n *Node) Remove() {
 	if n.dependers != nil {
 		// `mu` must be holded during accessing dependers.
 		n.dependers.Ascend(func(node *Node) bool {
-			removedDependees := stdatomic.AddInt32(&node.removedDependees, 1)
-			node.maybeResolve(0, removedDependees)
+			removedDependencies := stdatomic.AddInt32(&node.removedDependencies, 1)
+			node.maybeResolve(0, removedDependencies)
 			return true
 		})
 		n.dependers.Clear(true)
@@ -217,9 +225,9 @@ func (n *Node) assignTo(workerID int64) bool {
 	if n.dependers != nil {
 		// `mu` must be holded during accessing dependers.
 		n.dependers.Ascend(func(node *Node) bool {
-			resolvedDependees := stdatomic.AddInt32(&node.resolvedDependees, 1)
-			stdatomic.StoreInt64(&node.resolvedList[resolvedDependees-1], n.assignedTo)
-			node.maybeResolve(resolvedDependees, 0)
+			resolvedDependencies := stdatomic.AddInt32(&node.resolvedDependencies, 1)
+			stdatomic.StoreInt64(&node.resolvedList[resolvedDependencies-1], n.assignedTo)
+			node.maybeResolve(resolvedDependencies, 0)
 			return true
 		})
 	}
@@ -227,14 +235,16 @@ func (n *Node) assignTo(workerID int64) bool {
 	return true
 }
 
-func (n *Node) maybeResolve(resolvedDependees, removedDependees int32) {
-	if workerNum, ok := n.tryResolve(resolvedDependees, removedDependees); ok {
+func (n *Node) maybeResolve(resolvedDependencies, removedDependencies int32) {
+	if workerNum, ok := n.tryResolve(resolvedDependencies, removedDependencies); ok {
 		if workerNum < 0 {
 			panic("Node.tryResolve must return a valid worker ID")
 		}
 		if n.OnNotified != nil {
+			// Notify the conflict detector background worker to assign the node to the worker asynchronously.
 			n.OnNotified(func() { n.assignTo(workerNum) })
 		} else {
+			// Assign the node to the worker directly.
 			n.assignTo(workerNum)
 		}
 	}
@@ -244,39 +254,50 @@ func (n *Node) maybeResolve(resolvedDependees, removedDependees int32) {
 // Returns (_, false) if there is a conflict,
 // returns (rand, true) if there is no conflict,
 // returns (N, true) if only worker N can be used.
-func (n *Node) tryResolve(resolvedDependees, removedDependees int32) (int64, bool) {
-	assignedTo, resolved := n.doResolve(resolvedDependees, removedDependees)
+func (n *Node) tryResolve(resolvedDependencies, removedDependencies int32) (int64, bool) {
+	assignedTo, resolved := n.doResolve(resolvedDependencies, removedDependencies)
 	if resolved && assignedTo == assignedToAny {
 		assignedTo = n.RandWorkerID()
 	}
 	return assignedTo, resolved
 }
 
-func (n *Node) doResolve(resolvedDependees, removedDependees int32) (int64, bool) {
-	if n.totalDependees == 0 {
+func (n *Node) doResolve(resolvedDependencies, removedDependencies int32) (int64, bool) {
+	if n.totalDependencies == 0 {
 		// No conflicts, can select any workers.
 		return assignedToAny, true
 	}
 
-	if resolvedDependees == n.totalDependees {
+	if resolvedDependencies == n.totalDependencies {
 		firstDep := stdatomic.LoadInt64(&n.resolvedList[0])
 		hasDiffDep := false
-		for i := 1; i < int(n.totalDependees); i++ {
+		for i := 1; i < int(n.totalDependencies); i++ {
 			curr := stdatomic.LoadInt64(&n.resolvedList[i])
+			// // Todo: simplify assign to logic, only resolve dependencies nodes after
+			// // corresponding transactions are executed.
+			// //
+			// // In DependOn, depend(nil) set resolvedList[i] to assignedToAny
+			// // for these no dependecy keys.
+			// if curr == assignedToAny {
+			// 	continue
+			// }
 			if firstDep != curr {
 				hasDiffDep = true
 				break
 			}
 		}
 		if !hasDiffDep {
-			// If all dependees are assigned to one same worker, we can assign
-			// this node to the same worker directly.
+			// If all dependency nodes are assigned to the same worker, we can assign
+			// this node to the same worker directly, and they will execute sequentially.
+			// On the other hand, if dependency nodes are assigned to different workers,
+			// This node has to wait all dependency txn executed and all depencecy nodes
+			// are removed.
 			return firstDep, true
 		}
 	}
 
-	// All dependees are removed, so assign the node to any worker is fine.
-	if removedDependees == n.totalDependees {
+	// All dependcies are removed, so assign the node to any worker is fine.
+	if removedDependencies == n.totalDependencies {
 		return assignedToAny, true
 	}
 
diff --git a/pkg/causality/internal/slots.go b/pkg/causality/internal/slots.go
index cc70a0c41..d6d131e02 100644
--- a/pkg/causality/internal/slots.go
+++ b/pkg/causality/internal/slots.go
@@ -28,7 +28,7 @@ type SlotNode[T any] interface {
 	// NodeID tells the node's ID.
 	NodeID() int64
 	// Construct a dependency on `others`.
-	DependOn(unresolvedDeps map[int64]T, resolvedDeps int)
+	DependOn(dependencyNodes map[int64]T, noDependencyKeyCnt int)
 	// Remove the node itself and notify all dependers.
 	Remove()
 	// Free the node itself and remove it from the graph.
@@ -56,32 +56,43 @@ func NewSlots[E SlotNode[E]](numSlots uint64) *Slots[E] {
 }
 
 // Add adds an elem to the slots and calls DependOn for elem.
-func (s *Slots[E]) Add(elem E, keys []uint64) {
-	unresolvedDeps := make(map[int64]E, len(keys))
-	resolvedDeps := 0
+func (s *Slots[E]) Add(elem E, hashes []uint64) {
+	dependencyNodes := make(map[int64]E, len(hashes))
+	noDependecyCnt := 0
 
 	var lastSlot uint64 = math.MaxUint64
-	for _, key := range keys {
-		slotIdx := getSlot(key, s.numSlots)
+	for _, hash := range hashes {
+		// lock the slot that the node belongs to.
+		slotIdx := getSlot(hash, s.numSlots)
 		if lastSlot != slotIdx {
 			s.slots[slotIdx].mu.Lock()
 			lastSlot = slotIdx
 		}
-		if tail, ok := s.slots[slotIdx].nodes[key]; ok {
-			prevID := tail.NodeID()
-			unresolvedDeps[prevID] = tail
+
+		// If there is a node occpuied the same hash slot, we may have conflict with it.
+		// Add the conflict node to the dependencyNodes.
+		if prevNode, ok := s.slots[slotIdx].nodes[hash]; ok {
+			prevID := prevNode.NodeID()
+			// If there are multiple hashes conflicts with the same node, we only need to
+			// depend on the node once.
+			dependencyNodes[prevID] = prevNode
 		} else {
-			resolvedDeps += 1
+			noDependecyCnt += 1
 		}
-		s.slots[slotIdx].nodes[key] = elem
+		// Add this node to the slot, make sure new coming nodes with the same hash should
+		// depend on this node.
+		s.slots[slotIdx].nodes[hash] = elem
 	}
-	elem.DependOn(unresolvedDeps, resolvedDeps)
+
+	// Construct the dependency graph based on collected `dependencyNodes` and with corresponding
+	// slots locked.
+	elem.DependOn(dependencyNodes, noDependecyCnt)
 
 	// Lock those slots one by one and then unlock them one by one, so that
 	// we can avoid 2 transactions get executed interleaved.
 	lastSlot = math.MaxUint64
-	for _, key := range keys {
-		slotIdx := getSlot(key, s.numSlots)
+	for _, hash := range hashes {
+		slotIdx := getSlot(hash, s.numSlots)
 		if lastSlot != slotIdx {
 			s.slots[slotIdx].mu.Unlock()
 			lastSlot = slotIdx
@@ -90,18 +101,21 @@ func (s *Slots[E]) Add(elem E, keys []uint64) {
 }
 
 // Free removes an element from the Slots.
-func (s *Slots[E]) Free(elem E, keys []uint64) {
-	for _, key := range keys {
-		slotIdx := getSlot(key, s.numSlots)
+func (s *Slots[E]) Free(elem E, hashes []uint64) {
+	for _, hash := range hashes {
+		slotIdx := getSlot(hash, s.numSlots)
 		s.slots[slotIdx].mu.Lock()
-		if tail, ok := s.slots[slotIdx].nodes[key]; ok && tail.NodeID() == elem.NodeID() {
-			delete(s.slots[slotIdx].nodes, key)
+		// Remove the node from the slot.
+		// If the node is not in the slot, it means the node has been replaced by new node with the same hash,
+		// in this case we don't need to remove it from the slot.
+		if tail, ok := s.slots[slotIdx].nodes[hash]; ok && tail.NodeID() == elem.NodeID() {
+			delete(s.slots[slotIdx].nodes, hash)
 		}
 		s.slots[slotIdx].mu.Unlock()
 	}
 	elem.Free()
 }
 
-func getSlot(key, numSlots uint64) uint64 {
-	return key % numSlots
+func getSlot(hash, numSlots uint64) uint64 {
+	return hash % numSlots
 }
diff --git a/pkg/causality/worker.go b/pkg/causality/worker.go
index fb2234b26..738ba8891 100644
--- a/pkg/causality/worker.go
+++ b/pkg/causality/worker.go
@@ -13,19 +13,15 @@
 
 package causality
 
-type (
-	conflictKey = uint64
-)
-
 type txnEvent interface {
 	// OnConflictResolved is called when the event leaves ConflictDetector.
 	OnConflictResolved()
 
-	// Keys are in range [0, math.MaxUint64) and must be deduped.
+	// Hashes are in range [0, math.MaxUint64) and must be deduped.
 	//
 	// NOTE: if the conflict detector is accessed by multiple threads concurrently,
-	// ConflictKeys must also be sorted based on `key % numSlots`.
-	ConflictKeys(numSlots uint64) []conflictKey
+	// GenSortedDedupKeysHash must also be sorted based on `key % numSlots`.
+	GenSortedDedupKeysHash(numSlots uint64) []uint64
 }
 
 type worker[Txn txnEvent] interface {
diff --git a/pkg/config/consistent.go b/pkg/config/consistent.go
index c8f93c6ea..3409cf832 100644
--- a/pkg/config/consistent.go
+++ b/pkg/config/consistent.go
@@ -35,6 +35,17 @@ type ConsistentConfig struct {
 	UseFileBackend        bool   `toml:"use-file-backend" json:"use-file-backend"`
 	Compression           string `toml:"compression" json:"compression"`
 	FlushConcurrency      int    `toml:"flush-concurrency" json:"flush-concurrency,omitempty"`
+
+	MemoryUsage *ConsistentMemoryUsage `toml:"memory-usage" json:"memory-usage"`
+}
+
+// ConsistentMemoryUsage represents memory usage of Consistent module.
+type ConsistentMemoryUsage struct {
+	// ReplicaConfig.MemoryQuota * MemoryQuotaPercentage / 100 will be used for redo events.
+	MemoryQuotaPercentage uint64 `toml:"memory-quota-percentage" json:"memory-quota-percentage"`
+	// ReplicaConfig.MemoryQuota * MemoryQuotaPercentage / 100 * EventCachePercentage / 100
+	// will be used for redo cache.
+	EventCachePercentage uint64 `toml:"event-cache-percentage" json:"event-cache-percentage"`
 }
 
 // ValidateAndAdjust validates the consistency config and adjusts it if necessary.
diff --git a/pkg/config/replica_config.go b/pkg/config/replica_config.go
index 333a1e9f9..577937bdb 100644
--- a/pkg/config/replica_config.go
+++ b/pkg/config/replica_config.go
@@ -88,6 +88,10 @@ var defaultReplicaConfig = &ReplicaConfig{
 		Storage:               "",
 		UseFileBackend:        false,
 		Compression:           "",
+		MemoryUsage: &ConsistentMemoryUsage{
+			MemoryQuotaPercentage: 50,
+			EventCachePercentage:  0,
+		},
 	},
 	Scheduler: &ChangefeedSchedulerConfig{
 		EnableTableAcrossNodes: false,
@@ -100,6 +104,7 @@ var defaultReplicaConfig = &ReplicaConfig{
 	},
 	ChangefeedErrorStuckDuration: util.AddressOf(time.Minute * 30),
 	SQLMode:                      defaultSQLMode,
+	SyncedStatus:                 &SyncedStatusConfig{SyncedCheckInterval: 5 * 60, CheckpointInterval: 15},
 }
 
 // GetDefaultReplicaConfig returns the default replica config.
@@ -149,9 +154,10 @@ type replicaConfig struct {
 	// Scheduler is the configuration for scheduler.
 	Scheduler *ChangefeedSchedulerConfig `toml:"scheduler" json:"scheduler"`
 	// Integrity is only available when the downstream is MQ.
-	Integrity                    *integrity.Config `toml:"integrity" json:"integrity"`
-	ChangefeedErrorStuckDuration *time.Duration    `toml:"changefeed-error-stuck-duration" json:"changefeed-error-stuck-duration,omitempty"`
-	SQLMode                      string            `toml:"sql-mode" json:"sql-mode"`
+	Integrity                    *integrity.Config   `toml:"integrity" json:"integrity"`
+	ChangefeedErrorStuckDuration *time.Duration      `toml:"changefeed-error-stuck-duration" json:"changefeed-error-stuck-duration,omitempty"`
+	SQLMode                      string              `toml:"sql-mode" json:"sql-mode"`
+	SyncedStatus                 *SyncedStatusConfig `toml:"synced-status" json:"synced-status,omitempty"`
 }
 
 // Value implements the driver.Valuer interface
diff --git a/pkg/config/synced_status_config.go b/pkg/config/synced_status_config.go
new file mode 100644
index 000000000..144851f83
--- /dev/null
+++ b/pkg/config/synced_status_config.go
@@ -0,0 +1,23 @@
+// 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 config
+
+// SyncedStatusConfig represents synced check interval config for a changefeed
+type SyncedStatusConfig struct {
+	// The minimum interval between the latest synced ts and now required to reach synced state
+	SyncedCheckInterval int64 `toml:"synced-check-interval" json:"synced-check-interval"`
+	// The maximum interval between latest checkpoint ts and now or
+	// between latest sink's checkpoint ts and puller's checkpoint ts required to reach synced state
+	CheckpointInterval int64 `toml:"checkpoint-interval" json:"checkpoint-interval"`
+}
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index a964c68e6..d61a89792 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -64,12 +64,16 @@ var allowDDLList = []timodel.ActionType{
 // Filter are safe for concurrent use.
 // TODO: find a better way to abstract this interface.
 type Filter interface {
-	// ShouldIgnoreDMLEvent returns true and nil if the DML event should be ignored.
+	// ShouldIgnoreDMLEvent returns true if the DML event should be ignored.
 	ShouldIgnoreDMLEvent(dml *model.RowChangedEvent, rawRow model.RowChangedDatums, tableInfo *model.TableInfo) (bool, error)
+	// ShouldIgnoreDDLEvent returns true if the DDL event should be ignored.
+	// If a ddl is ignored, it will be applied to cdc's schema storage,
+	// but will not be sent to downstream.
+	ShouldIgnoreDDLEvent(ddl *model.DDLEvent) (bool, error)
 	// ShouldDiscardDDL returns true if this DDL should be discarded.
 	// If a ddl is discarded, it will neither be applied to cdc's schema storage
 	// nor sent to downstream.
-	ShouldDiscardDDL(startTs uint64, ddlType timodel.ActionType, schema, table, query string) (bool, error)
+	ShouldDiscardDDL(ddlType timodel.ActionType, schema, table string) bool
 	// ShouldIgnoreTable returns true if the table should be ignored.
 	ShouldIgnoreTable(schema, table string) bool
 	// ShouldIgnoreSchema returns true if the schema should be ignored.
@@ -146,31 +150,41 @@ func (f *filter) ShouldIgnoreDMLEvent(
 	return f.dmlExprFilter.shouldSkipDML(dml, rawRow, ti)
 }
 
-// ShouldDiscardDDL returns true if this DDL should be discarded.
-// If a ddl is discarded, it will not be applied to cdc's schema storage
-// and sent to downstream.
-func (f *filter) ShouldDiscardDDL(startTs uint64, ddlType timodel.ActionType, schema, table, query string) (discard bool, err error) {
-	discard = !isAllowedDDL(ddlType)
-	if discard {
-		return
-	}
-
-	discard = f.shouldIgnoreStartTs(startTs)
-	if discard {
-		return
+// ShouldDiscardDDL checks if a DDL should be discarded by conditions below:
+// 0. By allow list.
+// 1. By schema name.
+// 2. By table name.
+func (f *filter) ShouldDiscardDDL(ddlType timodel.ActionType, schema, table string) bool {
+	if !isAllowedDDL(ddlType) {
+		return true
 	}
 
 	if IsSchemaDDL(ddlType) {
-		discard = !f.tableFilter.MatchSchema(schema)
-	} else {
-		discard = f.ShouldIgnoreTable(schema, table)
+		return f.ShouldIgnoreSchema(schema)
 	}
+	return f.ShouldIgnoreTable(schema, table)
+}
 
-	if discard {
-		return
+// ShouldIgnoreDDLEvent checks if a DDL event should be ignore by conditions below:
+// 0. By startTs.
+// 1. By ddl type.
+// 2. By ddl query.
+//
+// If a ddl is ignored, it will be applied to cdc's schema storage,
+// but will not be sent to downstream.
+// Note that a ignored ddl is different from a discarded ddl. For example, suppose
+// we have a changefeed-test with the following config:
+//   - table filter: rules = ['test.*']
+//   - event-filters: matcher = ["test.worker"] ignore-event = ["create table"]
+//
+// Then, for the following DDLs:
+//  1. `CREATE TABLE test.worker` will be ignored, but the table will be replicated by changefeed-test.
+//  2. `CREATE TABLE other.worker` will be discarded, and the table will not be replicated by changefeed-test.
+func (f *filter) ShouldIgnoreDDLEvent(ddl *model.DDLEvent) (bool, error) {
+	if f.shouldIgnoreStartTs(ddl.StartTs) {
+		return true, nil
 	}
-
-	return f.sqlEventFilter.shouldSkipDDL(ddlType, schema, table, query)
+	return f.sqlEventFilter.shouldSkipDDL(ddl)
 }
 
 // ShouldIgnoreTable returns true if the specified table should be ignored by this changefeed.
diff --git a/pkg/filter/sql_event_filter.go b/pkg/filter/sql_event_filter.go
index fdf3ab885..39235eb05 100644
--- a/pkg/filter/sql_event_filter.go
+++ b/pkg/filter/sql_event_filter.go
@@ -21,7 +21,6 @@ import (
 	"github.com/pingcap/log"
 	bf "github.com/pingcap/tidb-tools/pkg/binlog-filter"
 	"github.com/pingcap/tidb/parser"
-	timodel "github.com/pingcap/tidb/parser/model"
 	"github.com/pingcap/tidb/parser/mysql"
 	tfilter "github.com/pingcap/tidb/util/table-filter"
 	"github.com/pingcap/tiflow/cdc/model"
@@ -155,22 +154,22 @@ func (f *sqlEventFilter) getRules(schema, table string) []*sqlEventRule {
 }
 
 // skipDDLEvent skips ddl event by its type and query.
-func (f *sqlEventFilter) shouldSkipDDL(
-	ddlType timodel.ActionType, schema, table, query string,
-) (bool, error) {
+func (f *sqlEventFilter) shouldSkipDDL(ddl *model.DDLEvent) (bool, error) {
+	schema := ddl.TableInfo.TableName.Schema
+	table := ddl.TableInfo.TableName.Table
 	log.Info("sql event filter handle ddl event",
-		zap.Any("ddlType", ddlType), zap.String("schema", schema),
-		zap.String("table", table), zap.String("query", query))
+		zap.Any("ddlType", ddl.Type), zap.String("schema", schema),
+		zap.String("table", table), zap.String("query", ddl.Query))
 	f.pLock.Lock()
-	evenType, err := ddlToEventType(f.ddlParser, query, ddlType)
+	evenType, err := ddlToEventType(f.ddlParser, ddl.Query, ddl.Type)
 	f.pLock.Unlock()
 	if err != nil {
 		return false, err
 	}
 	if evenType == bf.NullEvent {
 		log.Warn("sql event filter unsupported ddl type, do nothing",
-			zap.String("type", ddlType.String()),
-			zap.String("query", query))
+			zap.String("type", ddl.Type.String()),
+			zap.String("query", ddl.Query))
 		return false, nil
 	}
 
@@ -179,7 +178,7 @@ func (f *sqlEventFilter) shouldSkipDDL(
 		action, err := rule.bf.Filter(
 			binlogFilterSchemaPlaceholder,
 			binlogFilterTablePlaceholder,
-			evenType, query)
+			evenType, ddl.Query)
 		if err != nil {
 			return false, errors.Trace(err)
 		}


Test patch:
diff --git a/cdc/api/v1/api_test.go b/cdc/api/v1/api_test.go
index 24b08e2d4..eb3f702bf 100644
--- a/cdc/api/v1/api_test.go
+++ b/cdc/api/v1/api_test.go
@@ -100,6 +100,13 @@ func (p *mockStatusProvider) GetCaptures(ctx context.Context) ([]*model.CaptureI
 	return args.Get(0).([]*model.CaptureInfo), args.Error(1)
 }
 
+func (p *mockStatusProvider) GetChangeFeedSyncedStatus(ctx context.Context,
+	changefeedID model.ChangeFeedID,
+) (*model.ChangeFeedSyncedStatusForAPI, error) {
+	args := p.Called(ctx)
+	return args.Get(0).(*model.ChangeFeedSyncedStatusForAPI), args.Error(1)
+}
+
 func (p *mockStatusProvider) IsHealthy(ctx context.Context) (bool, error) {
 	args := p.Called(ctx)
 	return args.Get(0).(bool), args.Error(1)
diff --git a/cdc/api/v2/api_test.go b/cdc/api/v2/api_test.go
index 96929887a..98072ecc6 100644
--- a/cdc/api/v2/api_test.go
+++ b/cdc/api/v2/api_test.go
@@ -49,7 +49,7 @@ func (c *mockPDClient) UpdateServiceGCSafePoint(ctx context.Context,
 
 // GetTS of mockPDClient returns a mock tso
 func (c *mockPDClient) GetTS(ctx context.Context) (int64, int64, error) {
-	return c.logicTime, c.timestamp, nil
+	return c.timestamp, c.logicTime, nil
 }
 
 // GetClusterID of mockPDClient returns a mock ClusterID
@@ -62,13 +62,14 @@ func (c *mockPDClient) Close() {}
 
 type mockStatusProvider struct {
 	owner.StatusProvider
-	changefeedStatus   *model.ChangeFeedStatusForAPI
-	changefeedInfo     *model.ChangeFeedInfo
-	processors         []*model.ProcInfoSnap
-	taskStatus         map[model.CaptureID]*model.TaskStatus
-	changefeedInfos    map[model.ChangeFeedID]*model.ChangeFeedInfo
-	changefeedStatuses map[model.ChangeFeedID]*model.ChangeFeedStatusForAPI
-	err                error
+	changefeedStatus       *model.ChangeFeedStatusForAPI
+	changefeedInfo         *model.ChangeFeedInfo
+	processors             []*model.ProcInfoSnap
+	taskStatus             map[model.CaptureID]*model.TaskStatus
+	changefeedInfos        map[model.ChangeFeedID]*model.ChangeFeedInfo
+	changefeedStatuses     map[model.ChangeFeedID]*model.ChangeFeedStatusForAPI
+	changeFeedSyncedStatus *model.ChangeFeedSyncedStatusForAPI
+	err                    error
 }
 
 // GetChangeFeedStatus returns a changefeeds' runtime status.
@@ -120,6 +121,14 @@ func (m *mockStatusProvider) GetAllChangeFeedStatuses(_ context.Context) (
 	return m.changefeedStatuses, m.err
 }
 
+// GetChangeFeedSyncedStatus returns a mock changefeed status.
+func (m *mockStatusProvider) GetChangeFeedSyncedStatus(_ context.Context, changefeedID model.ChangeFeedID) (
+	*model.ChangeFeedSyncedStatusForAPI,
+	error,
+) {
+	return m.changeFeedSyncedStatus, m.err
+}
+
 func (m *mockStatusProvider) IsChangefeedOwner(_ context.Context, id model.ChangeFeedID) (bool, error) {
 	return true, nil
 }
diff --git a/cdc/api/v2/changefeed_test.go b/cdc/api/v2/changefeed_test.go
index e37cb4b4c..b150a3199 100644
--- a/cdc/api/v2/changefeed_test.go
+++ b/cdc/api/v2/changefeed_test.go
@@ -964,6 +964,213 @@ func TestPauseChangefeed(t *testing.T) {
 	require.Equal(t, "{}", w.Body.String())
 }
 
+func TestChangefeedSynced(t *testing.T) {
+	syncedInfo := testCase{url: "/api/v2/changefeeds/%s/synced?namespace=abc", method: "GET"}
+	helpers := NewMockAPIV2Helpers(gomock.NewController(t))
+	cp := mock_capture.NewMockCapture(gomock.NewController(t))
+	owner := mock_owner.NewMockOwner(gomock.NewController(t))
+	apiV2 := NewOpenAPIV2ForTest(cp, helpers)
+	router := newRouter(apiV2)
+
+	statusProvider := &mockStatusProvider{}
+
+	cp.EXPECT().StatusProvider().Return(statusProvider).AnyTimes()
+	cp.EXPECT().IsReady().Return(true).AnyTimes()
+	cp.EXPECT().IsController().Return(true).AnyTimes()
+	cp.EXPECT().GetOwner().Return(owner, nil).AnyTimes()
+
+	pdClient := &mockPDClient{}
+	mockUpManager := upstream.NewManager4Test(pdClient)
+	cp.EXPECT().GetUpstreamManager().Return(mockUpManager, nil).AnyTimes()
+
+	{
+		// case 1: invalid changefeed id
+		w := httptest.NewRecorder()
+		invalidID := "@^Invalid"
+		req, _ := http.NewRequestWithContext(context.Background(),
+			syncedInfo.method, fmt.Sprintf(syncedInfo.url, invalidID), nil)
+		router.ServeHTTP(w, req)
+		respErr := model.HTTPError{}
+		err := json.NewDecoder(w.Body).Decode(&respErr)
+		require.Nil(t, err)
+		require.Contains(t, respErr.Code, "ErrAPIInvalidParam")
+	}
+
+	{
+		// case 2: not existed changefeed id
+		validID := changeFeedID.ID
+		statusProvider.err = cerrors.ErrChangeFeedNotExists.GenWithStackByArgs(validID)
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(context.Background(), syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID), nil)
+		router.ServeHTTP(w, req)
+		respErr := model.HTTPError{}
+		err := json.NewDecoder(w.Body).Decode(&respErr)
+		require.Nil(t, err)
+		require.Contains(t, respErr.Code, "ErrChangeFeedNotExists")
+		require.Equal(t, http.StatusBadRequest, w.Code)
+	}
+
+	validID := changeFeedID.ID
+	cfInfo := &model.ChangeFeedInfo{
+		ID: validID,
+	}
+	statusProvider.err = nil
+	statusProvider.changefeedInfo = cfInfo
+	{
+		helpers.EXPECT().getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, cerrors.ErrAPIGetPDClientFailed).Times(1)
+		// case3: pd is offline，resolvedTs - checkpointTs > 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153217279 << 18,
+			LastSyncedTs:     1701153217279 << 18,
+			PullerResolvedTs: 1701153247279 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "[CDC:ErrAPIGetPDClientFailed]failed to get PDClient to connect PD, "+
+			"please recheck. Besides the data is not finish syncing", resp.Info)
+	}
+
+	{
+		helpers.EXPECT().getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).Return(nil, cerrors.ErrAPIGetPDClientFailed).Times(1)
+		// case4: pd is offline，resolvedTs - checkpointTs < 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153217279 << 18,
+			LastSyncedTs:     1701153217279 << 18,
+			PullerResolvedTs: 1701153217479 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "[CDC:ErrAPIGetPDClientFailed]failed to get PDClient to connect PD, please recheck. "+
+			"You should check the pd status first. If pd status is normal, means we don't finish sync data. "+
+			"If pd is offline, please check whether we satisfy the condition that "+
+			"the time difference from lastSyncedTs to the current time from the time zone of pd is greater than 300 secs. "+
+			"If it's satisfied, means the data syncing is totally finished", resp.Info)
+	}
+
+	helpers.EXPECT().getPDClient(gomock.Any(), gomock.Any(), gomock.Any()).Return(pdClient, nil).AnyTimes()
+	pdClient.logicTime = 1000
+	pdClient.timestamp = 1701153217279
+	{
+		// case5: pdTs - lastSyncedTs > 5min, pdTs - checkpointTs < 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153217209 << 18,
+			LastSyncedTs:     1701152217279 << 18,
+			PullerResolvedTs: 1701153217229 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, true, resp.Synced)
+		require.Equal(t, "Data syncing is finished", resp.Info)
+	}
+
+	{
+		// case6: pdTs - lastSyncedTs > 5min, pdTs - checkpointTs > 15s, resolvedTs - checkpointTs < 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153201279 << 18,
+			LastSyncedTs:     1701152217279 << 18,
+			PullerResolvedTs: 1701153201379 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "Please check whether PD is online and TiKV Regions are all available. "+
+			"If PD is offline or some TiKV regions are not available, it means that the data syncing process is complete. "+
+			"To check whether TiKV regions are all available, "+
+			"you can view 'TiKV-Details' > 'Resolved-Ts' > 'Max Leader Resolved TS gap' on Grafana. "+
+			"If the gap is large, such as a few minutes, it means that some regions in TiKV are unavailable. "+
+			"Otherwise, if the gap is small and PD is online, it means the data syncing is incomplete, so please wait", resp.Info)
+	}
+
+	{
+		// case7: pdTs - lastSyncedTs > 5min, pdTs - checkpointTs > 15s, resolvedTs - checkpointTs > 15s
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153201279 << 18,
+			LastSyncedTs:     1701152207279 << 18,
+			PullerResolvedTs: 1701153218279 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "The data syncing is not finished, please wait", resp.Info)
+	}
+
+	{
+		// case8: pdTs - lastSyncedTs < 5min
+		statusProvider.changeFeedSyncedStatus = &model.ChangeFeedSyncedStatusForAPI{
+			CheckpointTs:     1701153217279 << 18,
+			LastSyncedTs:     1701153213279 << 18,
+			PullerResolvedTs: 1701153217279 << 18,
+		}
+		w := httptest.NewRecorder()
+		req, _ := http.NewRequestWithContext(
+			context.Background(),
+			syncedInfo.method,
+			fmt.Sprintf(syncedInfo.url, validID),
+			nil,
+		)
+		router.ServeHTTP(w, req)
+		require.Equal(t, http.StatusOK, w.Code)
+		resp := SyncedStatus{}
+		err := json.NewDecoder(w.Body).Decode(&resp)
+		require.Nil(t, err)
+		require.Equal(t, false, resp.Synced)
+		require.Equal(t, "The data syncing is not finished, please wait", resp.Info)
+	}
+}
+
 func TestHasRunningImport(t *testing.T) {
 	integration.BeforeTestExternal(t)
 	testEtcdCluster := integration.NewClusterV3(
diff --git a/cdc/api/v2/model_test.go b/cdc/api/v2/model_test.go
index c6b999c55..2b7311489 100644
--- a/cdc/api/v2/model_test.go
+++ b/cdc/api/v2/model_test.go
@@ -68,6 +68,10 @@ var defaultAPIConfig = &ReplicaConfig{
 		FlushWorkerNum:        redo.DefaultFlushWorkerNum,
 		Storage:               "",
 		UseFileBackend:        false,
+		MemoryUsage: &ConsistentMemoryUsage{
+			MemoryQuotaPercentage: 50,
+			EventCachePercentage:  0,
+		},
 	},
 	Scheduler: &ChangefeedSchedulerConfig{
 		EnableTableAcrossNodes: config.GetDefaultReplicaConfig().
@@ -83,7 +87,8 @@ var defaultAPIConfig = &ReplicaConfig{
 	},
 	ChangefeedErrorStuckDuration: &JSONDuration{*config.
 		GetDefaultReplicaConfig().ChangefeedErrorStuckDuration},
-	SQLMode: config.GetDefaultReplicaConfig().SQLMode,
+	SQLMode:      config.GetDefaultReplicaConfig().SQLMode,
+	SyncedStatus: (*SyncedStatusConfig)(config.GetDefaultReplicaConfig().SyncedStatus),
 }
 
 func TestDefaultReplicaConfig(t *testing.T) {
diff --git a/cdc/entry/schema/snapshot_test.go b/cdc/entry/schema/snapshot_test.go
index d8248784f..68106599e 100644
--- a/cdc/entry/schema/snapshot_test.go
+++ b/cdc/entry/schema/snapshot_test.go
@@ -219,19 +219,19 @@ func TestUpdatePartition(t *testing.T) {
 	oldTb = newTbInfo(1, "DB_1", 11)
 	oldTb.Partition = nil
 	require.Nil(t, snap.inner.createTable(oldTb, 110))
-	require.Error(t, snap.inner.updatePartition(newTbInfo(1, "DB_1", 11), 120))
+	require.Error(t, snap.inner.updatePartition(newTbInfo(1, "DB_1", 11), false, 120))
 
 	// updatePartition fails if the new table is not partitioned.
 	require.Nil(t, snap.inner.dropTable(11, 130))
 	require.Nil(t, snap.inner.createTable(newTbInfo(1, "DB_1", 11), 140))
 	newTb = newTbInfo(1, "DB_1", 11)
 	newTb.Partition = nil
-	require.Error(t, snap.inner.updatePartition(newTb, 150))
+	require.Error(t, snap.inner.updatePartition(newTb, false, 150))
 	snap1 = snap.Copy()
 
 	newTb = newTbInfo(1, "DB_1", 11)
 	newTb.Partition.Definitions[0] = timodel.PartitionDefinition{ID: 11 + 65536*2}
-	require.Nil(t, snap.inner.updatePartition(newTb, 160))
+	require.Nil(t, snap.inner.updatePartition(newTb, false, 160))
 	snap2 = snap.Copy()
 
 	info, _ = snap1.PhysicalTableByID(11)
@@ -253,6 +253,31 @@ func TestUpdatePartition(t *testing.T) {
 	require.True(t, snap2.IsIneligibleTableID(11+65536*2))
 }
 
+func TestTruncateTablePartition(t *testing.T) {
+	var oldTb, newTb *model.TableInfo
+
+	snap := NewEmptySnapshot(false)
+	require.Nil(t, snap.inner.createSchema(newDBInfo(1), 100))
+
+	// updatePartition fails if the old table is not partitioned.
+	oldTb = newTbInfo(1, "DB_1", 11)
+	oldTb.Partition = nil
+	require.Nil(t, snap.inner.createTable(oldTb, 110))
+	require.Error(t, snap.inner.updatePartition(newTbInfo(1, "DB_1", 11), false, 120))
+
+	// updatePartition fails if the new table is not partitioned.
+	require.Nil(t, snap.inner.dropTable(11, 130))
+	require.Nil(t, snap.inner.createTable(newTbInfo(1, "DB_1", 11), 140))
+	newTb = newTbInfo(1, "DB_1", 11)
+	newTb.Partition = nil
+	require.Error(t, snap.inner.updatePartition(newTb, false, 150))
+
+	newTb = newTbInfo(1, "DB_1", 11)
+	newTb.Partition.Definitions[0] = timodel.PartitionDefinition{ID: 11 + 65536*2}
+	require.Nil(t, snap.inner.updatePartition(newTb, true, 160))
+	require.True(t, snap.IsTruncateTableID(11+65536))
+}
+
 func TestExchangePartition(t *testing.T) {
 	var targetTb, sourceTb *model.TableInfo
 
diff --git a/cdc/owner/changefeed_test.go b/cdc/owner/changefeed_test.go
index faef25010..d7c6a3075 100644
--- a/cdc/owner/changefeed_test.go
+++ b/cdc/owner/changefeed_test.go
@@ -165,9 +165,14 @@ func (m *mockScheduler) Tick(
 	currentTables []model.TableID,
 	captures map[model.CaptureID]*model.CaptureInfo,
 	barrier *schedulepb.BarrierWithMinTs,
-) (newCheckpointTs, newResolvedTs model.Ts, err error) {
+) (watermark schedulepb.Watermark, err error) {
 	m.currentTables = currentTables
-	return barrier.MinTableBarrierTs, barrier.GlobalBarrierTs, nil
+	return schedulepb.Watermark{
+		CheckpointTs:     barrier.MinTableBarrierTs,
+		ResolvedTs:       barrier.GlobalBarrierTs,
+		LastSyncedTs:     scheduler.CheckpointCannotProceed,
+		PullerResolvedTs: scheduler.CheckpointCannotProceed,
+	}, nil
 }
 
 // MoveTable is used to trigger manual table moves.
diff --git a/cdc/owner/ddl_manager_test.go b/cdc/owner/ddl_manager_test.go
index 3255608be..520b6e0f0 100644
--- a/cdc/owner/ddl_manager_test.go
+++ b/cdc/owner/ddl_manager_test.go
@@ -44,6 +44,7 @@ func createDDLManagerForTest(t *testing.T) *ddlManager {
 		startTs,
 		checkpointTs,
 		ddlSink,
+		f,
 		ddlPuller,
 		schema,
 		redo.NewDisabledDDLManager(),
diff --git a/cdc/processor/sinkmanager/manager_test.go b/cdc/processor/sinkmanager/manager_test.go
index b1447fe19..4b245ea26 100644
--- a/cdc/processor/sinkmanager/manager_test.go
+++ b/cdc/processor/sinkmanager/manager_test.go
@@ -30,10 +30,13 @@ import (
 )
 
 func getChangefeedInfo() *model.ChangeFeedInfo {
+	replicaConfig := config.GetDefaultReplicaConfig()
+	replicaConfig.Consistent.MemoryUsage.MemoryQuotaPercentage = 75
+	replicaConfig.Consistent.MemoryUsage.EventCachePercentage = 50
 	return &model.ChangeFeedInfo{
 		Error:   nil,
 		SinkURI: "blackhole://",
-		Config:  config.GetDefaultReplicaConfig(),
+		Config:  replicaConfig,
 	}
 }
 
@@ -93,6 +96,13 @@ func addTableAndAddEventsToSortEngine(
 				CRTs:   4,
 			},
 		},
+		{
+			CRTs: 6,
+			RawKV: &model.RawKVEntry{
+				OpType: model.OpTypeResolved,
+				CRTs:   6,
+			},
+		},
 	}
 	for _, event := range events {
 		engine.Add(span, event)
@@ -174,6 +184,8 @@ func TestRemoveTable(t *testing.T) {
 }
 
 func TestGenerateTableSinkTaskWithBarrierTs(t *testing.T) {
+	t.Skip("Temporary disable this case since it is not stable")
+
 	t.Parallel()
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -195,10 +207,16 @@ func TestGenerateTableSinkTaskWithBarrierTs(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
-		tableSink, ok := manager.tableSinks.Load(span)
-		require.True(t, ok)
-		checkpointTS := tableSink.(*tableSinkWrapper).getCheckpointTs()
-		return checkpointTS.ResolvedMark() == 4
+		s := manager.GetTableStats(span)
+		return s.CheckpointTs == 4 && s.LastSyncedTs == 4
+	}, 5*time.Second, 10*time.Millisecond)
+
+	manager.UpdateBarrierTs(6, nil)
+	manager.UpdateReceivedSorterResolvedTs(span, 6)
+	manager.schemaStorage.AdvanceResolvedTs(6)
+	require.Eventually(t, func() bool {
+		s := manager.GetTableStats(span)
+		return s.CheckpointTs == 6 && s.LastSyncedTs == 4
 	}, 5*time.Second, 10*time.Millisecond)
 }
 
@@ -226,10 +244,8 @@ func TestGenerateTableSinkTaskWithResolvedTs(t *testing.T) {
 	require.NoError(t, err)
 
 	require.Eventually(t, func() bool {
-		tableSink, ok := manager.tableSinks.Load(span)
-		require.True(t, ok)
-		checkpointTS := tableSink.(*tableSinkWrapper).getCheckpointTs()
-		return checkpointTS.ResolvedMark() == 3
+		s := manager.GetTableStats(span)
+		return s.CheckpointTs == 3 && s.LastSyncedTs == 3
 	}, 5*time.Second, 10*time.Millisecond)
 }
 
@@ -257,7 +273,7 @@ func TestGetTableStatsToReleaseMemQuota(t *testing.T) {
 
 	require.Eventually(t, func() bool {
 		s := manager.GetTableStats(span)
-		return manager.sinkMemQuota.GetUsedBytes() == 0 && s.CheckpointTs == 4
+		return manager.sinkMemQuota.GetUsedBytes() == 0 && s.CheckpointTs == 4 && s.LastSyncedTs == 4
 	}, 5*time.Second, 10*time.Millisecond)
 }
 
diff --git a/cdc/puller/ddl_puller_test.go b/cdc/puller/ddl_puller_test.go
index 8b6a7dcd3..b36114f15 100644
--- a/cdc/puller/ddl_puller_test.go
+++ b/cdc/puller/ddl_puller_test.go
@@ -439,7 +439,7 @@ func TestHandleJob(t *testing.T) {
 		job = helper.DDL2Job("alter table test1.t1 add column c1 int")
 		skip, err = ddlJobPullerImpl.handleJob(job)
 		require.NoError(t, err)
-		require.True(t, skip)
+		require.False(t, skip)
 
 		job = helper.DDL2Job("create table test1.testStartTs(id int)")
 		skip, err = ddlJobPullerImpl.handleJob(job)
@@ -450,7 +450,7 @@ func TestHandleJob(t *testing.T) {
 		job.StartTS = 1
 		skip, err = ddlJobPullerImpl.handleJob(job)
 		require.NoError(t, err)
-		require.True(t, skip)
+		require.False(t, skip)
 
 		job = helper.DDL2Job("create table test1.t2(id int)")
 		skip, err = ddlJobPullerImpl.handleJob(job)
diff --git a/cdc/puller/frontier/frontier_test.go b/cdc/puller/frontier/frontier_test.go
index 2ca327160..cb2e1cd11 100644
--- a/cdc/puller/frontier/frontier_test.go
+++ b/cdc/puller/frontier/frontier_test.go
@@ -15,6 +15,7 @@ package frontier
 
 import (
 	"bytes"
+	"fmt"
 	"math"
 	"math/rand"
 	"sort"
@@ -435,3 +436,21 @@ func TestFrontierEntries(t *testing.T) {
 	require.Equal(t, []byte("a"), []byte(slowestRange.StartKey))
 	require.Equal(t, []byte("b"), []byte(slowestRange.EndKey))
 }
+
+func TestMergeSpitWithDifferentRegionID(t *testing.T) {
+	frontier := NewFrontier(100, tablepb.Span{StartKey: []byte("a"), EndKey: []byte("c")})
+	frontier.Forward(1, tablepb.Span{StartKey: []byte("a"), EndKey: []byte("b")}, 1222)
+	frontier.Forward(2, tablepb.Span{StartKey: []byte("b"), EndKey: []byte("c")}, 102)
+	frontier.Forward(4, tablepb.Span{StartKey: []byte("b"), EndKey: []byte("c")}, 103)
+	frontier.Forward(1, tablepb.Span{StartKey: []byte("a"), EndKey: []byte("c")}, 104)
+	frontier.Forward(1, tablepb.Span{StartKey: []byte("a"), EndKey: []byte("b")}, 1223)
+	frontier.Forward(3, tablepb.Span{StartKey: []byte("b"), EndKey: []byte("c")}, 105)
+	frontier.Forward(2, tablepb.Span{StartKey: []byte("b"), EndKey: []byte("c")}, 107)
+	frontier.(*spanFrontier).spanList.Entries(func(node *skipListNode) bool {
+		fmt.Printf("%d:[%s: %s) %d\n", node.regionID,
+			string(node.Key()),
+			string(node.End()), node.value.key)
+		return true
+	})
+	require.Equal(t, uint64(107), frontier.Frontier())
+}
diff --git a/cdc/scheduler/internal/v3/coordinator_test.go b/cdc/scheduler/internal/v3/coordinator_test.go
index 8d691b4e0..8caacc30a 100644
--- a/cdc/scheduler/internal/v3/coordinator_test.go
+++ b/cdc/scheduler/internal/v3/coordinator_test.go
@@ -251,7 +251,7 @@ func TestCoordinatorHeartbeat(t *testing.T) {
 	ctx := context.Background()
 	currentTables := []model.TableID{1, 2, 3}
 	aliveCaptures := map[model.CaptureID]*model.CaptureInfo{"a": {}, "b": {}}
-	_, _, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	msgs := trans.SendBuffer
 	require.Len(t, msgs, 2)
@@ -283,7 +283,7 @@ func TestCoordinatorHeartbeat(t *testing.T) {
 		},
 	})
 	trans.SendBuffer = []*schedulepb.Message{}
-	_, _, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	require.True(t, coord.captureM.CheckAllCaptureInitialized())
 	msgs = trans.SendBuffer
@@ -324,7 +324,7 @@ func TestCoordinatorAddCapture(t *testing.T) {
 	ctx := context.Background()
 	currentTables := []model.TableID{1, 2, 3}
 	aliveCaptures := map[model.CaptureID]*model.CaptureInfo{"a": {}, "b": {}}
-	_, _, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	msgs = trans.SendBuffer
 	require.Len(t, msgs, 1)
@@ -340,7 +340,7 @@ func TestCoordinatorAddCapture(t *testing.T) {
 		HeartbeatResponse: &schedulepb.HeartbeatResponse{},
 	})
 	trans.SendBuffer = []*schedulepb.Message{}
-	_, _, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	msgs = trans.SendBuffer
 	require.Len(t, msgs, 1)
@@ -381,7 +381,7 @@ func TestCoordinatorRemoveCapture(t *testing.T) {
 	ctx := context.Background()
 	currentTables := []model.TableID{1, 2, 3}
 	aliveCaptures := map[model.CaptureID]*model.CaptureInfo{"a": {}, "b": {}}
-	_, _, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 	msgs = trans.SendBuffer
 	require.Len(t, msgs, 1)
@@ -456,7 +456,7 @@ func TestCoordinatorAdvanceCheckpoint(t *testing.T) {
 	ctx := context.Background()
 	currentTables := []model.TableID{1, 2}
 	aliveCaptures := map[model.CaptureID]*model.CaptureInfo{"a": {}, "b": {}}
-	_, _, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
+	_, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(0))
 	require.Nil(t, err)
 
 	// Initialize captures.
@@ -482,24 +482,40 @@ func TestCoordinatorAdvanceCheckpoint(t *testing.T) {
 					Span:  spanz.TableIDToComparableSpan(1),
 					State: tablepb.TableStateReplicating,
 					Checkpoint: tablepb.Checkpoint{
-						CheckpointTs: 2, ResolvedTs: 4,
+						CheckpointTs: 2, ResolvedTs: 4, LastSyncedTs: 3,
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(5),
+							},
+						},
 					},
 				},
 				{
 					Span:  spanz.TableIDToComparableSpan(2),
 					State: tablepb.TableStateReplicating,
 					Checkpoint: tablepb.Checkpoint{
-						CheckpointTs: 2, ResolvedTs: 4,
+						CheckpointTs: 2, ResolvedTs: 4, LastSyncedTs: 4,
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(6),
+							},
+						},
 					},
 				},
 			},
 		},
 	})
-	cts, rts, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(5))
+	watermark, err := coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(5))
 	require.Nil(t, err)
 	require.True(t, coord.captureM.CheckAllCaptureInitialized())
-	require.EqualValues(t, 2, cts)
-	require.EqualValues(t, 4, rts)
+	require.EqualValues(t, 2, watermark.CheckpointTs)
+	require.EqualValues(t, 4, watermark.ResolvedTs)
+	require.EqualValues(t, 4, watermark.LastSyncedTs)
+	require.EqualValues(t, 5, watermark.PullerResolvedTs)
 
 	// Checkpoint should be advanced even if there is an uninitialized capture.
 	aliveCaptures["c"] = &model.CaptureInfo{}
@@ -517,24 +533,40 @@ func TestCoordinatorAdvanceCheckpoint(t *testing.T) {
 					Span:  spanz.TableIDToComparableSpan(1),
 					State: tablepb.TableStateReplicating,
 					Checkpoint: tablepb.Checkpoint{
-						CheckpointTs: 3, ResolvedTs: 5,
+						CheckpointTs: 3, ResolvedTs: 5, LastSyncedTs: 4,
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(7),
+							},
+						},
 					},
 				},
 				{
 					Span:  spanz.TableIDToComparableSpan(2),
 					State: tablepb.TableStateReplicating,
 					Checkpoint: tablepb.Checkpoint{
-						CheckpointTs: 4, ResolvedTs: 5,
+						CheckpointTs: 4, ResolvedTs: 5, LastSyncedTs: 6,
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(7),
+							},
+						},
 					},
 				},
 			},
 		},
 	})
-	cts, rts, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(5))
+	watermark, err = coord.poll(ctx, 0, currentTables, aliveCaptures, schedulepb.NewBarrierWithMinTs(5))
 	require.Nil(t, err)
 	require.False(t, coord.captureM.CheckAllCaptureInitialized())
-	require.EqualValues(t, 3, cts)
-	require.EqualValues(t, 5, rts)
+	require.EqualValues(t, 3, watermark.CheckpointTs)
+	require.EqualValues(t, 5, watermark.ResolvedTs)
+	require.EqualValues(t, 6, watermark.LastSyncedTs)
+	require.EqualValues(t, 7, watermark.PullerResolvedTs)
 }
 
 func TestCoordinatorDropMsgIfChangefeedEpochMismatch(t *testing.T) {
diff --git a/cdc/scheduler/internal/v3/replication/replication_manager_test.go b/cdc/scheduler/internal/v3/replication/replication_manager_test.go
index 632a0835a..a899bc4a9 100644
--- a/cdc/scheduler/internal/v3/replication/replication_manager_test.go
+++ b/cdc/scheduler/internal/v3/replication/replication_manager_test.go
@@ -15,6 +15,7 @@ package replication
 
 import (
 	"context"
+	"math"
 	"testing"
 	"time"
 
@@ -620,6 +621,14 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(10),
 					ResolvedTs:   model.Ts(20),
+					LastSyncedTs: model.Ts(15),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(30),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -635,6 +644,14 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(15),
 					ResolvedTs:   model.Ts(30),
+					LastSyncedTs: model.Ts(20),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(40),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -646,21 +663,30 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	// no tables are replicating, resolvedTs should be advanced to globalBarrierTs and checkpoint
 	// should be advanced to minTableBarrierTs.
 	currentTables := &TableRanges{}
-	checkpoint, resolved := r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(5), redoMetaManager)
-	require.Equal(t, model.Ts(5), checkpoint)
-	require.Equal(t, model.Ts(5), resolved)
+	watermark := r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(5), redoMetaManager)
+	require.Equal(t, model.Ts(5), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(5), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(0), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(math.MaxUint64), watermark.PullerResolvedTs)
 
 	// all tables are replicating
 	currentTables.UpdateTables([]model.TableID{1, 2})
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, model.Ts(10), checkpoint)
-	require.Equal(t, model.Ts(20), resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, model.Ts(10), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(20), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(20), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(30), watermark.PullerResolvedTs)
 
 	// some table not exist yet.
 	currentTables.UpdateTables([]model.TableID{1, 2, 3})
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, checkpointCannotProceed, checkpoint)
-	require.Equal(t, checkpointCannotProceed, resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, checkpointCannotProceed, watermark.CheckpointTs)
+	require.Equal(t, checkpointCannotProceed, watermark.ResolvedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.LastSyncedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.PullerResolvedTs)
 
 	span3 := spanz.TableIDToComparableSpan(3)
 	rs, err = NewReplicationSet(span3, model.Ts(5),
@@ -671,6 +697,14 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(5),
 					ResolvedTs:   model.Ts(40),
+					LastSyncedTs: model.Ts(30),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(50),
+						},
+					},
 				},
 			},
 			"2": {
@@ -679,14 +713,25 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(5),
 					ResolvedTs:   model.Ts(40),
+					LastSyncedTs: model.Ts(32),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(50),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
 	require.NoError(t, err)
 	r.spans.ReplaceOrInsert(span3, rs)
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, model.Ts(5), checkpoint)
-	require.Equal(t, model.Ts(20), resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, model.Ts(5), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(20), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(32), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(30), watermark.PullerResolvedTs)
 
 	currentTables.UpdateTables([]model.TableID{1, 2, 3, 4})
 	span4 := spanz.TableIDToComparableSpan(4)
@@ -698,14 +743,25 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(3),
 					ResolvedTs:   model.Ts(10),
+					LastSyncedTs: model.Ts(5),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(12),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
 	require.NoError(t, err)
 	r.spans.ReplaceOrInsert(span4, rs)
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, model.Ts(3), checkpoint)
-	require.Equal(t, model.Ts(10), resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, model.Ts(3), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(10), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(32), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(12), watermark.PullerResolvedTs)
 
 	// Split table 5 into 2 spans.
 	currentTables.UpdateTables([]model.TableID{1, 2, 3, 4, 5})
@@ -722,28 +778,45 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 					Checkpoint: tablepb.Checkpoint{
 						CheckpointTs: model.Ts(3),
 						ResolvedTs:   model.Ts(10),
+						LastSyncedTs: model.Ts(8),
+					},
+					Stats: tablepb.Stats{
+						StageCheckpoints: map[string]tablepb.Checkpoint{
+							"puller-egress": {
+								ResolvedTs: model.Ts(11),
+							},
+						},
 					},
 				},
 			}, model.ChangeFeedID{})
 		require.NoError(t, err)
 		r.spans.ReplaceOrInsert(span, rs)
 	}
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, model.Ts(3), checkpoint)
-	require.Equal(t, model.Ts(10), resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, model.Ts(3), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(10), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(32), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(11), watermark.PullerResolvedTs)
 
 	// The start span is missing
 	rs5_1, _ := r.spans.Delete(span5_1)
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, checkpointCannotProceed, checkpoint)
-	require.Equal(t, checkpointCannotProceed, resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, checkpointCannotProceed, watermark.CheckpointTs)
+	require.Equal(t, checkpointCannotProceed, watermark.ResolvedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.LastSyncedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.PullerResolvedTs)
 
 	// The end span is missing
 	r.spans.ReplaceOrInsert(span5_1, rs5_1)
 	r.spans.Delete(span5_2)
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
-	require.Equal(t, checkpointCannotProceed, checkpoint)
-	require.Equal(t, checkpointCannotProceed, resolved)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), schedulepb.NewBarrierWithMinTs(30), redoMetaManager)
+	require.Equal(t, checkpointCannotProceed, watermark.CheckpointTs)
+	require.Equal(t, checkpointCannotProceed, watermark.ResolvedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.LastSyncedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.PullerResolvedTs)
 
 	// redo is enabled
 	currentTables.UpdateTables([]model.TableID{4})
@@ -756,6 +829,14 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(10),
 					ResolvedTs:   model.Ts(15),
+					LastSyncedTs: model.Ts(12),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(16),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -764,9 +845,12 @@ func TestReplicationManagerAdvanceCheckpoint(t *testing.T) {
 	barrier := schedulepb.NewBarrierWithMinTs(30)
 	redoMetaManager.enable = true
 	redoMetaManager.resolvedTs = 9
-	checkpoint, resolved = r.AdvanceCheckpoint(currentTables, time.Now(), barrier, redoMetaManager)
-	require.Equal(t, model.Ts(9), resolved)
-	require.Equal(t, model.Ts(9), checkpoint)
+	watermark = r.AdvanceCheckpoint(
+		currentTables, time.Now(), barrier, redoMetaManager)
+	require.Equal(t, model.Ts(9), watermark.ResolvedTs)
+	require.Equal(t, model.Ts(9), watermark.CheckpointTs)
+	require.Equal(t, model.Ts(12), watermark.LastSyncedTs)
+	require.Equal(t, model.Ts(16), watermark.PullerResolvedTs)
 	require.Equal(t, model.Ts(9), barrier.GetGlobalBarrierTs())
 }
 
@@ -782,6 +866,14 @@ func TestReplicationManagerAdvanceCheckpointWithRedoEnabled(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(10),
 					ResolvedTs:   model.Ts(20),
+					LastSyncedTs: model.Ts(12),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(26),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -797,6 +889,14 @@ func TestReplicationManagerAdvanceCheckpointWithRedoEnabled(t *testing.T) {
 				Checkpoint: tablepb.Checkpoint{
 					CheckpointTs: model.Ts(15),
 					ResolvedTs:   model.Ts(30),
+					LastSyncedTs: model.Ts(18),
+				},
+				Stats: tablepb.Stats{
+					StageCheckpoints: map[string]tablepb.Checkpoint{
+						"puller-egress": {
+							ResolvedTs: model.Ts(39),
+						},
+					},
 				},
 			},
 		}, model.ChangeFeedID{})
@@ -809,9 +909,13 @@ func TestReplicationManagerAdvanceCheckpointWithRedoEnabled(t *testing.T) {
 	currentTables := &TableRanges{}
 	currentTables.UpdateTables([]model.TableID{1, 2, 3})
 	barrier := schedulepb.NewBarrierWithMinTs(30)
-	checkpoint, resolved := r.AdvanceCheckpoint(currentTables, time.Now(), barrier, redoMetaManager)
-	require.Equal(t, checkpointCannotProceed, checkpoint)
-	require.Equal(t, checkpointCannotProceed, resolved)
+	watermark := r.AdvanceCheckpoint(
+		currentTables,
+		time.Now(), barrier, redoMetaManager)
+	require.Equal(t, checkpointCannotProceed, watermark.CheckpointTs)
+	require.Equal(t, checkpointCannotProceed, watermark.ResolvedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.LastSyncedTs)
+	require.Equal(t, checkpointCannotProceed, watermark.PullerResolvedTs)
 	require.Equal(t, uint64(25), barrier.Barrier.GetGlobalBarrierTs())
 }
 
diff --git a/cdc/sink/dmlsink/txn/event_test.go b/cdc/sink/dmlsink/txn/event_test.go
index c0dfb0049..cf8c2181e 100644
--- a/cdc/sink/dmlsink/txn/event_test.go
+++ b/cdc/sink/dmlsink/txn/event_test.go
@@ -203,3 +203,30 @@ func TestGenKeys(t *testing.T) {
 		require.Equal(t, tc.expected, keys)
 	}
 }
+
+func TestSortAndDedupHash(t *testing.T) {
+	// If a transaction contains multiple rows, those rows may generate the same hash
+	// in some rare cases. We should dedup these hashes to avoid unnecessary self cyclic
+	// dependency in the causality dependency graph.
+	t.Parallel()
+	testCases := []struct {
+		hashes   []uint64
+		expected []uint64
+	}{{
+		// No duplicate hashes
+		hashes:   []uint64{1, 2, 3, 4, 5},
+		expected: []uint64{1, 2, 3, 4, 5},
+	}, {
+		// Duplicate hashes
+		hashes:   []uint64{1, 2, 3, 4, 5, 1, 2, 3, 4, 5},
+		expected: []uint64{1, 2, 3, 4, 5},
+	}, {
+		// Has hash value larger than slots count, should sort by `hash % numSlots` first.
+		hashes:   []uint64{4, 9, 9, 3},
+		expected: []uint64{9, 3, 4},
+	}}
+
+	for _, tc := range testCases {
+		require.Equal(t, tc.expected, sortAndDedupHashes(tc.hashes, 8))
+	}
+}
diff --git a/cdc/sink/tablesink/table_sink_impl_test.go b/cdc/sink/tablesink/table_sink_impl_test.go
index 95a26711a..34e334f33 100644
--- a/cdc/sink/tablesink/table_sink_impl_test.go
+++ b/cdc/sink/tablesink/table_sink_impl_test.go
@@ -247,6 +247,7 @@ func TestGetCheckpointTs(t *testing.T) {
 
 	tb.AppendRowChangedEvents(getTestRows()...)
 	require.Equal(t, model.NewResolvedTs(0), tb.GetCheckpointTs(), "checkpointTs should be 0")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(0), "lastSyncedTs should be not updated")
 
 	// One event will be flushed.
 	err := tb.UpdateResolvedTs(model.NewResolvedTs(101))
@@ -254,11 +255,13 @@ func TestGetCheckpointTs(t *testing.T) {
 	require.Equal(t, model.NewResolvedTs(0), tb.GetCheckpointTs(), "checkpointTs should be 0")
 	sink.acknowledge(101)
 	require.Equal(t, model.NewResolvedTs(101), tb.GetCheckpointTs(), "checkpointTs should be 101")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(101), "lastSyncedTs should be the same as the flushed event")
 
 	// Flush all events.
 	err = tb.UpdateResolvedTs(model.NewResolvedTs(105))
 	require.Nil(t, err)
 	require.Equal(t, model.NewResolvedTs(101), tb.GetCheckpointTs(), "checkpointTs should be 101")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(101), "lastSyncedTs should be not updated")
 
 	// Only acknowledge some events.
 	sink.acknowledge(102)
@@ -268,10 +271,12 @@ func TestGetCheckpointTs(t *testing.T) {
 		tb.GetCheckpointTs(),
 		"checkpointTs should still be 101",
 	)
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(102), "lastSyncedTs should be updated")
 
 	// Ack all events.
 	sink.acknowledge(105)
 	require.Equal(t, model.NewResolvedTs(105), tb.GetCheckpointTs(), "checkpointTs should be 105")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(105), "lastSyncedTs should be updated")
 }
 
 func TestClose(t *testing.T) {
@@ -404,4 +409,5 @@ func TestCheckpointTsFrozenWhenStopping(t *testing.T) {
 	currentTs := tb.GetCheckpointTs()
 	sink.acknowledge(105)
 	require.Equal(t, currentTs, tb.GetCheckpointTs(), "checkpointTs should not be updated")
+	require.Equal(t, tb.lastSyncedTs.getLastSyncedTs(), uint64(105), "lastSyncedTs should not change")
 }
diff --git a/pkg/causality/tests/integration_test.go b/pkg/causality/tests/integration_test.go
index 9ed509531..35de5533d 100644
--- a/pkg/causality/tests/integration_test.go
+++ b/pkg/causality/tests/integration_test.go
@@ -40,7 +40,7 @@ func TestConflictBasics(t *testing.T) {
 		numWorkers, numSlots, newUniformGenerator(workingSetSize, batchSize, numSlots),
 	).WithExecFunc(
 		func(txn *txnForTest) error {
-			for _, key := range txn.ConflictKeys(numSlots) {
+			for _, key := range txn.GenSortedDedupKeysHash(numSlots) {
 				// Access a position in the array without synchronization,
 				// so that if causality check is buggy, the Go race detection would fail.
 				conflictArray[key]++
diff --git a/pkg/causality/tests/worker.go b/pkg/causality/tests/worker.go
index 901cc24f5..f78f406ba 100644
--- a/pkg/causality/tests/worker.go
+++ b/pkg/causality/tests/worker.go
@@ -27,7 +27,7 @@ type txnForTest struct {
 
 func (t *txnForTest) OnConflictResolved() {}
 
-func (t *txnForTest) ConflictKeys(numSlots uint64) []uint64 {
+func (t *txnForTest) GenSortedDedupKeysHash(numSlots uint64) []uint64 {
 	return t.keys
 }
 
diff --git a/pkg/config/config_test_data.go b/pkg/config/config_test_data.go
index d68f7fe32..063c0e334 100644
--- a/pkg/config/config_test_data.go
+++ b/pkg/config/config_test_data.go
@@ -75,7 +75,11 @@ const (
     "encoding-worker-num": 16,
     "flush-worker-num": 8,
     "storage": "",
-    "use-file-backend": false
+    "use-file-backend": false,
+    "memory-usage": {
+        "memory-quota-percentage": 50,
+        "event-cache-percentage": 0
+    }
   },
   "scheduler": {
     "enable-table-across-nodes": false,
@@ -86,7 +90,11 @@ const (
     "corruption-handle-level": "warn"
  },
   "changefeed-error-stuck-duration": 1800000000000,
-  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION",
+  "synced-status": {
+    "synced-check-interval": 300,
+    "checkpoint-interval": 15
+  }
 }`
 
 	testCfgTestServerConfigMarshal = `{
@@ -308,7 +316,11 @@ const (
     "encoding-worker-num": 16,
     "flush-worker-num": 8,
     "storage": "",
-    "use-file-backend": false
+    "use-file-backend": false,
+    "memory-usage": {
+        "memory-quota-percentage": 50,
+        "event-cache-percentage": 0
+    }
   },
   "scheduler": {
     "enable-table-across-nodes": true,
@@ -322,7 +334,11 @@ const (
     "corruption-handle-level": "warn"
   },
   "changefeed-error-stuck-duration": 1800000000000,
-  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION",
+  "synced-status": {
+    "synced-check-interval": 300,
+    "checkpoint-interval": 15
+  }
 }`
 
 	testCfgTestReplicaConfigMarshal2 = `{
@@ -460,7 +476,11 @@ const (
     "encoding-worker-num": 16,
     "flush-worker-num": 8,
     "storage": "",
-    "use-file-backend": false
+    "use-file-backend": false,
+    "memory-usage": {
+        "memory-quota-percentage": 50,
+        "event-cache-percentage": 0
+    }
   },
   "scheduler": {
     "enable-table-across-nodes": true,
@@ -472,6 +492,10 @@ const (
     "corruption-handle-level": "warn"
   },
   "changefeed-error-stuck-duration": 1800000000000,
-  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION"
+  "sql-mode":"ONLY_FULL_GROUP_BY,STRICT_TRANS_TABLES,NO_ZERO_IN_DATE,NO_ZERO_DATE,ERROR_FOR_DIVISION_BY_ZERO,NO_AUTO_CREATE_USER,NO_ENGINE_SUBSTITUTION",
+  "synced-status": {
+    "synced-check-interval": 300,
+    "checkpoint-interval": 15
+  }
 }`
 )
diff --git a/pkg/filter/filter_test.go b/pkg/filter/filter_test.go
index 33cd51d69..9d59ec5dd 100644
--- a/pkg/filter/filter_test.go
+++ b/pkg/filter/filter_test.go
@@ -152,8 +152,9 @@ func TestShouldIgnoreDMLEvent(t *testing.T) {
 	}
 }
 
-func TestShouldDiscardDDL(t *testing.T) {
+func TestShouldIgnoreDDL(t *testing.T) {
 	t.Parallel()
+
 	testCases := []struct {
 		cases []struct {
 			startTs uint64
@@ -166,137 +167,199 @@ func TestShouldDiscardDDL(t *testing.T) {
 		rules        []string
 		ignoredTs    []uint64
 		eventFilters []*config.EventFilterRule
-	}{{
-		// Ignore by table name cases.
-		cases: []struct {
-			startTs uint64
-			schema  string
-			table   string
-			query   string
-			ddlType timodel.ActionType
-			ignore  bool
-		}{
-			{1, "sns", "", "create database test", timodel.ActionCreateSchema, false},
-			{1, "sns", "", "drop database test", timodel.ActionDropSchema, false},
-			{
-				1, "sns", "", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
-				timodel.ActionModifySchemaCharsetAndCollate, false,
+	}{
+		{ // cases ignore by startTs
+			cases: []struct {
+				startTs uint64
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{1, "ts", "", "create database test", timodel.ActionCreateSchema, true},
+				{2, "ts", "student", "drop database test2", timodel.ActionDropSchema, true},
+				{
+					3, "ts", "teacher", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
+					timodel.ActionModifySchemaCharsetAndCollate, true,
+				},
+				{4, "ts", "man", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+				{5, "ts", "fruit", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+				{6, "ts", "insect", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
 			},
-			{1, "ecom", "", "create database test", timodel.ActionCreateSchema, false},
-			{1, "ecom", "aa", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
-			{1, "ecom", "", "create database test", timodel.ActionCreateSchema, false},
-			{1, "test", "", "create database test", timodel.ActionCreateSchema, true},
+			rules:     []string{"*.*"},
+			ignoredTs: []uint64{1, 2, 3},
 		},
-		rules: []string{"sns.*", "ecom.*", "!sns.log", "!ecom.test"},
-		// Ignore by schema name cases.
-	}, {
-		cases: []struct {
-			startTs uint64
-			schema  string
-			table   string
-			query   string
-			ddlType timodel.ActionType
-			ignore  bool
-		}{
-			{1, "schema", "C1", "create database test", timodel.ActionCreateSchema, false},
-			{1, "test", "", "drop database test1", timodel.ActionDropSchema, true},
-			{
-				1, "dbname", "", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
-				timodel.ActionModifySchemaCharsetAndCollate, true,
+		{ // cases ignore by ddl type.
+			cases: []struct {
+				startTs uint64
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{1, "event", "", "drop table t1", timodel.ActionDropTable, true},
+				{1, "event", "January", "drop index i on t1", timodel.ActionDropIndex, true},
+				{1, "event", "February", "drop index x2 on t2", timodel.ActionDropIndex, true},
+				{1, "event", "March", "create table t2(age int)", timodel.ActionCreateTable, false},
+				{1, "event", "April", "create table t2(age int)", timodel.ActionCreateTable, false},
+				{1, "event", "May", "create table t2(age int)", timodel.ActionCreateTable, false},
 			},
-			{1, "test", "aa", "create table test.t1(a int primary key)", timodel.ActionCreateTable, true},
-			{1, "schema", "C1", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
-			{1, "schema", "", "create table test.t1(a int primary key)", timodel.ActionCreateTable, true},
-		},
-		rules: []string{"schema.C1"},
-	}, { // cases ignore by startTs
-		cases: []struct {
-			startTs uint64
-			schema  string
-			table   string
-			query   string
-			ddlType timodel.ActionType
-			ignore  bool
-		}{
-			{1, "ts", "", "create database test", timodel.ActionCreateSchema, true},
-			{2, "ts", "student", "drop database test2", timodel.ActionDropSchema, true},
-			{
-				3, "ts", "teacher", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
-				timodel.ActionModifySchemaCharsetAndCollate, true,
+			rules: []string{"*.*"},
+			eventFilters: []*config.EventFilterRule{
+				{
+					Matcher: []string{"event.*"},
+					IgnoreEvent: []bf.EventType{
+						bf.AlterTable, bf.DropTable,
+					},
+				},
 			},
-			{4, "ts", "man", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
-			{5, "ts", "fruit", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
-			{6, "ts", "insect", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+			ignoredTs: []uint64{},
 		},
-		rules:     []string{"*.*"},
-		ignoredTs: []uint64{1, 2, 3},
-	}, { // cases ignore by ddl type.
-		cases: []struct {
-			startTs uint64
-			schema  string
-			table   string
-			query   string
-			ddlType timodel.ActionType
-			ignore  bool
-		}{
-			{1, "event", "", "drop table t1", timodel.ActionDropTable, true},
-			{1, "event", "January", "drop index i on t1", timodel.ActionDropIndex, true},
-			{1, "event", "February", "drop index x2 on t2", timodel.ActionDropIndex, true},
-			{1, "event", "March", "create table t2(age int)", timodel.ActionCreateTable, false},
-			{1, "event", "April", "create table t2(age int)", timodel.ActionCreateTable, false},
-			{1, "event", "May", "create table t2(age int)", timodel.ActionCreateTable, false},
-		},
-		rules: []string{"*.*"},
-		eventFilters: []*config.EventFilterRule{
-			{
-				Matcher: []string{"event.*"},
-				IgnoreEvent: []bf.EventType{
-					bf.AlterTable, bf.DropTable,
+		{ // cases ignore by ddl query
+			cases: []struct {
+				startTs uint64
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{1, "sql_pattern", "t1", "CREATE DATABASE sql_pattern", timodel.ActionCreateSchema, false},
+				{1, "sql_pattern", "t1", "DROP DATABASE sql_pattern", timodel.ActionDropSchema, true},
+				{
+					1, "sql_pattern", "t1",
+					"ALTER DATABASE `test_db` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci'",
+					timodel.ActionModifySchemaCharsetAndCollate,
+					false,
 				},
+				{1, "sql_pattern", "t1", "CREATE TABLE t1(id int primary key)", timodel.ActionCreateTable, false},
+				{1, "sql_pattern", "t1", "DROP TABLE t1", timodel.ActionDropTable, true},
+				{1, "sql_pattern", "t1", "CREATE VIEW test.v AS SELECT * FROM t", timodel.ActionCreateView, true},
 			},
+			rules: []string{"*.*"},
+			eventFilters: []*config.EventFilterRule{
+				{
+					Matcher:   []string{"sql_pattern.*"},
+					IgnoreSQL: []string{"^DROP TABLE", "^CREATE VIEW", "^DROP DATABASE"},
+				},
+			},
+			ignoredTs: []uint64{},
 		},
-	}, { // cases ignore by ddl query
-		cases: []struct {
-			startTs uint64
+	}
+
+	for _, ftc := range testCases {
+		filter, err := NewFilter(&config.ReplicaConfig{
+			Filter: &config.FilterConfig{
+				Rules:            ftc.rules,
+				EventFilters:     ftc.eventFilters,
+				IgnoreTxnStartTs: ftc.ignoredTs,
+			},
+		}, "")
+		require.Nil(t, err)
+		for _, tc := range ftc.cases {
+			ddl := &model.DDLEvent{
+				StartTs: tc.startTs,
+				TableInfo: &model.TableInfo{
+					TableName: model.TableName{
+						Schema: tc.schema,
+						Table:  tc.table,
+					},
+				},
+				Query: tc.query,
+				Type:  tc.ddlType,
+			}
+			ignore, err := filter.ShouldIgnoreDDLEvent(ddl)
+			require.NoError(t, err)
+			require.Equal(t, tc.ignore, ignore, "%#v", tc)
+		}
+	}
+}
+
+func TestShouldDiscardDDL(t *testing.T) {
+	t.Parallel()
+	testCases := []struct {
+		cases []struct {
 			schema  string
 			table   string
 			query   string
 			ddlType timodel.ActionType
 			ignore  bool
-		}{
-			{1, "sql_pattern", "t1", "CREATE DATABASE sql_pattern", timodel.ActionCreateSchema, false},
-			{1, "sql_pattern", "t1", "DROP DATABASE sql_pattern", timodel.ActionDropSchema, true},
-			{
-				1, "sql_pattern", "t1",
-				"ALTER DATABASE `test_db` CHARACTER SET 'utf8' COLLATE 'utf8_general_ci'",
-				timodel.ActionModifySchemaCharsetAndCollate,
-				false,
+		}
+		rules        []string
+		eventFilters []*config.EventFilterRule
+	}{
+		{
+			// Discard by not allowed DDL type cases.
+			cases: []struct {
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{"sns", "", "create database test", timodel.ActionCreateSchema, false},
+				{"sns", "", "drop database test", timodel.ActionDropSchema, false},
+				{"test", "", "create database test", timodel.ActionCreateSequence, true},
 			},
-			{1, "sql_pattern", "t1", "CREATE TABLE t1(id int primary key)", timodel.ActionCreateTable, false},
-			{1, "sql_pattern", "t1", "DROP TABLE t1", timodel.ActionDropTable, true},
-			{1, "sql_pattern", "t1", "CREATE VIEW test.v AS SELECT * FROM t", timodel.ActionCreateView, true},
+			rules: []string{"*.*"},
 		},
-		rules: []string{"*.*"},
-		eventFilters: []*config.EventFilterRule{
-			{
-				Matcher:   []string{"sql_pattern.*"},
-				IgnoreSQL: []string{"^DROP TABLE", "^CREATE VIEW", "^DROP DATABASE"},
+		{
+			// Discard by table name cases.
+			cases: []struct {
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{"sns", "", "create database test", timodel.ActionCreateSchema, false},
+				{"sns", "", "drop database test", timodel.ActionDropSchema, false},
+				{
+					"sns", "", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
+					timodel.ActionModifySchemaCharsetAndCollate, false,
+				},
+				{"ecom", "", "create database test", timodel.ActionCreateSchema, false},
+				{"ecom", "aa", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+				{"ecom", "", "create database test", timodel.ActionCreateSchema, false},
+				{"test", "", "create database test", timodel.ActionCreateSchema, true},
 			},
+			rules: []string{"sns.*", "ecom.*", "!sns.log", "!ecom.test"},
 		},
-	}}
+		{
+			// Discard by schema name cases.
+			cases: []struct {
+				schema  string
+				table   string
+				query   string
+				ddlType timodel.ActionType
+				ignore  bool
+			}{
+				{"schema", "C1", "create database test", timodel.ActionCreateSchema, false},
+				{"test", "", "drop database test1", timodel.ActionDropSchema, true},
+				{
+					"dbname", "", "ALTER DATABASE dbname CHARACTER SET utf8 COLLATE utf8_general_ci",
+					timodel.ActionModifySchemaCharsetAndCollate, true,
+				},
+				{"test", "aa", "create table test.t1(a int primary key)", timodel.ActionCreateTable, true},
+				{"schema", "C1", "create table test.t1(a int primary key)", timodel.ActionCreateTable, false},
+				{"schema", "", "create table test.t1(a int primary key)", timodel.ActionCreateTable, true},
+			},
+			rules: []string{"schema.C1"},
+		},
+	}
 
 	for _, ftc := range testCases {
 		filter, err := NewFilter(&config.ReplicaConfig{
 			Filter: &config.FilterConfig{
-				Rules:            ftc.rules,
-				IgnoreTxnStartTs: ftc.ignoredTs,
-				EventFilters:     ftc.eventFilters,
+				Rules:        ftc.rules,
+				EventFilters: ftc.eventFilters,
 			},
 		}, "")
 		require.Nil(t, err)
 		for _, tc := range ftc.cases {
-			ignore, err := filter.ShouldDiscardDDL(tc.startTs, tc.ddlType, tc.schema, tc.table, tc.query)
-			require.Nil(t, err, "%#v", tc)
+			ignore := filter.ShouldDiscardDDL(tc.ddlType, tc.schema, tc.table)
 			require.Equal(t, tc.ignore, ignore, "%#v", tc)
 		}
 	}
diff --git a/pkg/filter/sql_event_filter_test.go b/pkg/filter/sql_event_filter_test.go
index 9e9707d07..9b27fa2b9 100644
--- a/pkg/filter/sql_event_filter_test.go
+++ b/pkg/filter/sql_event_filter_test.go
@@ -183,8 +183,7 @@ func TestShouldSkipDDL(t *testing.T) {
 				},
 				Query: c.query,
 			}
-			skip, err := f.shouldSkipDDL(ddl.Type,
-				ddl.TableInfo.TableName.Schema, ddl.TableInfo.TableName.Table, ddl.Query)
+			skip, err := f.shouldSkipDDL(ddl)
 			require.NoError(t, err)
 			require.Equal(t, c.skip, skip, "case: %+v", c)
 		}
diff --git a/pkg/orchestrator/reactor_state_test.go b/pkg/orchestrator/reactor_state_test.go
index 7296da25f..3b2d8c474 100644
--- a/pkg/orchestrator/reactor_state_test.go
+++ b/pkg/orchestrator/reactor_state_test.go
@@ -131,7 +131,8 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity:  config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
-						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
+						SQLMode:      config.GetDefaultReplicaConfig().SQLMode,
+						SyncedStatus: config.GetDefaultReplicaConfig().SyncedStatus,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -194,7 +195,8 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Consistent: config.GetDefaultReplicaConfig().Consistent,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
-						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
+						SQLMode:      config.GetDefaultReplicaConfig().SQLMode,
+						SyncedStatus: config.GetDefaultReplicaConfig().SyncedStatus,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -262,7 +264,8 @@ func TestChangefeedStateUpdate(t *testing.T) {
 						Integrity:  config.GetDefaultReplicaConfig().Integrity,
 						ChangefeedErrorStuckDuration: config.
 							GetDefaultReplicaConfig().ChangefeedErrorStuckDuration,
-						SQLMode: config.GetDefaultReplicaConfig().SQLMode,
+						SQLMode:      config.GetDefaultReplicaConfig().SQLMode,
+						SyncedStatus: config.GetDefaultReplicaConfig().SyncedStatus,
 					},
 				},
 				Status: &model.ChangeFeedStatus{CheckpointTs: 421980719742451713},
@@ -358,6 +361,7 @@ func TestPatchInfo(t *testing.T) {
 			Integrity:                    defaultConfig.Integrity,
 			ChangefeedErrorStuckDuration: defaultConfig.ChangefeedErrorStuckDuration,
 			SQLMode:                      defaultConfig.SQLMode,
+			SyncedStatus:                 defaultConfig.SyncedStatus,
 		},
 	}
 	cfInfo.RmUnusedFields()
@@ -381,6 +385,7 @@ func TestPatchInfo(t *testing.T) {
 			Integrity:                    defaultConfig.Integrity,
 			ChangefeedErrorStuckDuration: defaultConfig.ChangefeedErrorStuckDuration,
 			SQLMode:                      defaultConfig.SQLMode,
+			SyncedStatus:                 defaultConfig.SyncedStatus,
 		},
 	}
 	cfInfo.RmUnusedFields()
diff --git a/tests/integration_tests/event_filter/conf/cf.toml b/tests/integration_tests/event_filter/conf/cf.toml
index 2ea6083c5..d3a36188b 100644
--- a/tests/integration_tests/event_filter/conf/cf.toml
+++ b/tests/integration_tests/event_filter/conf/cf.toml
@@ -6,3 +6,10 @@ matcher = ["event_filter.t1"]
 ignore-event = ["drop table", "delete"]
 ignore-insert-value-expr = "id = 2 or city = 'tokyo'"
 
+[[filter.event-filters]]
+matcher = ["event_filter.t_truncate"]
+ignore-event = ["truncate table"]
+
+[[filter.event-filters]]
+matcher = ["event_filter.t_alter"]
+ignore-event = ["alter table"]
diff --git a/tests/integration_tests/event_filter/conf/diff_config.toml b/tests/integration_tests/event_filter/conf/diff_config.toml
index 75593ed89..fff11e2ee 100644
--- a/tests/integration_tests/event_filter/conf/diff_config.toml
+++ b/tests/integration_tests/event_filter/conf/diff_config.toml
@@ -13,7 +13,7 @@ source-instances = ["tidb0"]
 
 target-instance = "mysql1"
 
-target-check-tables = ["event_filter.t2"]
+target-check-tables = ["event_filter.t_normal", "event_filter.t_truncate", "event_filter.t_alter"]
 
 [data-sources]
 [data-sources.tidb0]
diff --git a/tests/integration_tests/event_filter/data/test.sql b/tests/integration_tests/event_filter/data/test.sql
index a3ff2e479..6af0efed2 100644
--- a/tests/integration_tests/event_filter/data/test.sql
+++ b/tests/integration_tests/event_filter/data/test.sql
@@ -38,8 +38,8 @@ WHERE id = 4;
 /* ignore by event type*/
 DROP TABLE t1;
 
-/* all event of t2 will be replicated to downstream */
-CREATE TABLE t2 (
+/* all event of t_normal will be replicated to downstream */
+CREATE TABLE t_normal (
                     id INT,
                     name varchar(128),
                     country char(32),
@@ -48,16 +48,45 @@ CREATE TABLE t2 (
                     gender char(32),
                     PRIMARY KEY (id)
 );
-INSERT INTO t2
+INSERT INTO t_normal
 VALUES (1, 'guagua', "china", "chengdu", 1, "female");
 
-INSERT INTO t2
+INSERT INTO t_normal
 VALUES (2, 'huahua', "china", "chengdu", 2, "female");
 
-INSERT INTO t2
+INSERT INTO t_normal
 VALUES (3, 'xigua', "japan", "tokyo", 2, "male");
 
-INSERT INTO t2
+INSERT INTO t_normal
 VALUES (4, 'yuko', "japan", "nagoya", 33, "female");
 
-create table finish_mark(id int primary key);
+CREATE TABLE t_truncate (
+                    id INT,
+                    name varchar(128),
+                    PRIMARY KEY (id)
+);
+CREATE TABLE t_alter
+(
+    id          INT AUTO_INCREMENT,
+    t_boolean   BOOLEAN,
+    t_bigint    DECIMAL(38, 19),
+    t_double    DOUBLE,
+    t_decimal   DECIMAL(38, 19),
+    t_bit       BIT(64),
+    t_date      DATE,
+    t_datetime  DATETIME,
+    t_timestamp TIMESTAMP NULL,
+    t_time      TIME,
+    t_year      YEAR,
+    t_char      CHAR,
+    t_varchar   VARCHAR(10),
+    t_blob      BLOB,
+    t_text      TEXT,
+    t_enum      ENUM ('enum1', 'enum2', 'enum3'),
+    t_set       SET ('a', 'b', 'c'),
+    t_json      JSON,
+    PRIMARY KEY (id)
+) ENGINE = InnoDB
+  DEFAULT CHARSET = utf8
+  COLLATE = utf8_bin;
+  
\ No newline at end of file
diff --git a/tests/integration_tests/event_filter/data/test_alter.sql b/tests/integration_tests/event_filter/data/test_alter.sql
new file mode 100644
index 000000000..0e6fbabd2
--- /dev/null
+++ b/tests/integration_tests/event_filter/data/test_alter.sql
@@ -0,0 +1,49 @@
+USE `event_filter`;
+
+/* test  */
+ALTER TABLE t_alter MODIFY t_bigint BIGINT;
+
+-- make sure `nullable` can be handled by the mounter and mq encoding protocol
+INSERT INTO t_alter() VALUES ();
+
+INSERT INTO t_alter( t_boolean, t_bigint, t_double, t_decimal, t_bit
+                           , t_date, t_datetime, t_timestamp, t_time, t_year
+                           , t_char, t_varchar, t_blob, t_text, t_enum
+                           , t_set, t_json)
+VALUES ( true, 9223372036854775807, 123.123, 123456789012.123456789012, b'1000001'
+       , '1000-01-01', '9999-12-31 23:59:59', '19731230153000', '23:59:59', 1970
+       , '测', '测试', 'blob', '测试text', 'enum2'
+       , 'a,b', NULL);
+
+INSERT INTO t_alter( t_boolean, t_bigint, t_double, t_decimal, t_bit
+                           , t_date, t_datetime, t_timestamp, t_time, t_year
+                           , t_char, t_varchar, t_blob, t_text, t_enum
+                           , t_set, t_json)
+VALUES ( false, 666, 123.777, 123456789012.123456789012, b'1000001'
+       , '1000-01-01', '9999-12-31 23:59:59', '19731230153000', '23:59:59', 1970
+       , '测', '测试', 'blob', '测试text11', 'enum3'
+       , 'a,b', NULL);
+
+UPDATE t_alter
+SET t_bigint = 555
+WHERE id = 1;
+
+INSERT INTO t_alter( t_boolean, t_bigint, t_double, t_decimal, t_bit
+                           , t_date, t_datetime, t_timestamp, t_time, t_year
+                           , t_char, t_varchar, t_blob, t_text, t_enum
+                           , t_set, t_json)
+VALUES ( true, 9223372036875807, 153.123, 123456669012.123456789012, b'1010001'
+       , '2000-01-01', '9999-12-31 23:59:59', '19731230153000', '23:59:59', 1970
+       , '测', '测试', 'blob', '测试text', 'enum1'
+       , 'a,b', '{
+    "key1": "value1",
+    "key2": "value2"
+  }');
+
+UPDATE t_alter
+SET t_bigint = 888,
+    t_json   = '{
+      "key0": "value0",
+      "key2": "value2"
+    }'
+WHERE id = 2;
diff --git a/tests/integration_tests/event_filter/data/test_truncate.sql b/tests/integration_tests/event_filter/data/test_truncate.sql
new file mode 100644
index 000000000..170d94c9c
--- /dev/null
+++ b/tests/integration_tests/event_filter/data/test_truncate.sql
@@ -0,0 +1,15 @@
+USE `event_filter`;
+
+/* test new physical table is replicated */
+TRUNCATE TABLE t_truncate;
+INSERT INTO t_truncate
+VALUES (1, 'guagua');
+
+INSERT INTO t_truncate
+VALUES (2, 'huahua');
+
+INSERT INTO t_truncate
+VALUES (3, 'xigua');
+
+INSERT INTO t_truncate
+VALUES (4, 'yuko');
diff --git a/tests/integration_tests/event_filter/run.sh b/tests/integration_tests/event_filter/run.sh
index 62983b3d2..ce68a3c27 100644
--- a/tests/integration_tests/event_filter/run.sh
+++ b/tests/integration_tests/event_filter/run.sh
@@ -38,8 +38,9 @@ function run() {
 	# make suer table t1 is deleted in upstream and exists in downstream
 	check_table_not_exists "event_filter.t1" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	check_table_exists "event_filter.t1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-	check_table_exists "event_filter.t2" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-	check_table_exists "event_filter.finish_mark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_normal" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_truncate" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.t_alter" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 
 	# check those rows that are not filtered are synced to downstream
 	run_sql "select count(1) from event_filter.t1;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
@@ -53,7 +54,14 @@ function run() {
 	run_sql "select count(5) from event_filter.t1 where id=4;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_contains "count(5): 1"
 
-	# check table t2 is replicated
+	run_sql "TRUNCATE TABLE event_filter.t_truncate;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	run_sql_file $CUR/data/test_truncate.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "ALTER TABLE event_filter.t_alter MODIFY t_bigint BIGINT;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	run_sql_file $CUR/data/test_alter.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+	run_sql "create table event_filter.finish_mark(id int primary key);" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	check_table_exists "event_filter.finish_mark" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	# check table t_normal is replicated
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	cleanup_process $CDC_BINARY
diff --git a/tests/integration_tests/run_group.sh b/tests/integration_tests/run_group.sh
index e6211a5a4..bf4f9c2d9 100755
--- a/tests/integration_tests/run_group.sh
+++ b/tests/integration_tests/run_group.sh
@@ -40,7 +40,7 @@ groups=(
 	["G06"]='sink_retry changefeed_error ddl_sequence resourcecontrol'
 	["G07"]='kv_client_stream_reconnect cdc split_region'
 	["G08"]='processor_err_chan changefeed_reconstruct multi_capture'
-	["G09"]='gc_safepoint changefeed_pause_resume cli savepoint'
+	["G09"]='gc_safepoint changefeed_pause_resume cli savepoint synced_status'
 	["G10"]='default_value simple cdc_server_tips event_filter'
 	["G11"]='resolve_lock move_table autorandom generate_column'
 	["G12"]='many_pk_or_uk capture_session_done_during_task ddl_attributes'
diff --git a/tests/integration_tests/synced_status/conf/changefeed-redo.toml b/tests/integration_tests/synced_status/conf/changefeed-redo.toml
new file mode 100644
index 000000000..8f67ecb9d
--- /dev/null
+++ b/tests/integration_tests/synced_status/conf/changefeed-redo.toml
@@ -0,0 +1,7 @@
+[synced-status]
+synced-check-interval = 120 
+checkpoint-interval = 20
+
+[consistent]
+level="eventual"
+storage = "file:///tmp/tidb_cdc_test/synced_status/redo"
\ No newline at end of file
diff --git a/tests/integration_tests/synced_status/conf/changefeed.toml b/tests/integration_tests/synced_status/conf/changefeed.toml
new file mode 100644
index 000000000..0d8aa19fb
--- /dev/null
+++ b/tests/integration_tests/synced_status/conf/changefeed.toml
@@ -0,0 +1,3 @@
+[synced-status]
+synced-check-interval = 120 
+checkpoint-interval = 20
\ No newline at end of file
diff --git a/tests/integration_tests/synced_status/run.sh b/tests/integration_tests/synced_status/run.sh
new file mode 100644
index 000000000..ffb7d6892
--- /dev/null
+++ b/tests/integration_tests/synced_status/run.sh
@@ -0,0 +1,308 @@
+#!/bin/bash
+
+# [DISCRIPTION]:
+#   This test is related to
+#   It will test the sync status request of cdc server in the following scenarios:
+#   1. The sync status request of cdc server when the upstream cluster is available
+#      1.1 pdNow - lastSyncedTs > threshold, pdNow - checkpointTs < threshold
+#      1.2 pdNow - lastSyncedTs < threshold
+#      1.3 pdNow - lastSyncedTs > threshold, pdNow - checkpointTs < threshold, resolvedTs - checkpointTs > threshold
+#   2. The sync status request of cdc server when the upstream pd is unavailable
+#      2.1 resolvedTs - checkpointTs < threshold
+#   3. The sync status request of cdc server when the upstream tikv is unavailable
+#      3.1 pdNow - lastSyncedTs > threshold, pdNow - checkpointTs > threshold, resolvedTs - checkpointTs < threshold
+#      3.2 pdNow - lastSyncedTs < threshold
+#   4. The sync status request of cdc server when the downstream tidb is available
+#      4.1 pdNow - lastSyncedTs > threshold, pdNow - checkpointTs < threshold
+#      4.2 pdNow - lastSyncedTs < threshold
+# [STEP]:
+#   1. Create changefeed with synced-time-config = xx
+#   2. insert data to upstream cluster, and do the related actions for each scenarios
+#   3. do the query of synced status of cdc server
+#   4. check the info and status of query
+
+set -xeu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+CDC_COUNT=3
+DB_COUNT=4
+
+function kill_pd() {
+	info=$(ps aux | grep pd-server | grep $WORK_DIR) || true
+	$(ps aux | grep pd-server | grep $WORK_DIR | awk '{print $2}' | xargs kill -9 &>/dev/null) || true
+}
+
+function kill_tikv() {
+	info=$(ps aux | grep tikv-server | grep $WORK_DIR) || true
+	$(ps aux | grep tikv-server | grep $WORK_DIR | awk '{print $2}' | xargs kill -9 &>/dev/null) || true
+}
+
+function kill_tidb() {
+	info=$(ps aux | grep tidb-server | grep $WORK_DIR) || true
+	$(ps aux | grep tidb-server | grep $WORK_DIR | awk '{print $2}' | xargs kill -9 &>/dev/null) || true
+}
+
+function run_normal_case_and_unavailable_pd() {
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	config_path=$1
+
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --changefeed-id="test-1" --config="$CUR/$config_path"
+
+	# case 1: test in available cluster
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+
+	status=$(echo $synced_status | jq '.synced')
+	sink_checkpoint_ts=$(echo $synced_status | jq -r '.sink_checkpoint_ts')
+	puller_resolved_ts=$(echo $synced_status | jq -r '.puller_resolved_ts')
+	last_synced_ts=$(echo $synced_status | jq -r '.last_synced_ts')
+	if [ $status != true ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	# the timestamp for puller_resolved_ts is 0 when do data insert
+	if [ "$puller_resolved_ts" != "1970-01-01 08:00:00.000" ]; then
+		echo "puller_resolved_ts is not 1970-01-01 08:00:00.000"
+		exit 1
+	fi
+	# the timestamp for last_synced_ts is 0 when do data insert
+	if [ "$last_synced_ts" != "1970-01-01 08:00:00.000" ]; then
+		echo "last_synced_ts is not 1970-01-01 08:00:00.000"
+		exit 1
+	fi
+
+	# compare sink_checkpoint_ts with current time
+	current=$(date +"%Y-%m-%d %H:%M:%S")
+	echo "sink_checkpoint_ts is "$sink_checkpoint_ts
+	checkpoint_timestamp=$(date -d "$sink_checkpoint_ts" +%s)
+	current_timestamp=$(date -d "$current" +%s)
+	if [ $(($current_timestamp - $checkpoint_timestamp)) -gt 300 ]; then # give a soft check
+		echo "sink_checkpoint_ts is not correct"
+		exit 1
+	fi
+
+	run_sql "USE TEST;Create table t1(a int primary key, b int);insert into t1 values(1,2);insert into t1 values(2,3);"
+	check_table_exists "test.t1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	sleep 5 # wait data insert
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	if [ "$info" != "The data syncing is not finished, please wait" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	sleep 130 # wait enough time for pass synced-check-interval
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != true ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+
+	#==========
+	# case 2: test with unavailable pd, query will not get the available response
+	kill_pd
+
+	sleep 20
+
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	error_code=$(echo $synced_status | jq -r '.error_code')
+	cleanup_process $CDC_BINARY
+	stop_tidb_cluster
+}
+
+function run_case_with_unavailable_tikv() {
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	config_path=$1
+
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --changefeed-id="test-1" --config="$CUR/$config_path"
+
+	# case 3: test in unavailable tikv cluster
+	run_sql "USE TEST;Create table t1(a int primary key, b int);insert into t1 values(1,2);insert into t1 values(2,3);"
+	check_table_exists "test.t1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	sleep 5 # make data inserted into downstream
+	kill_tikv
+
+	# test the case when pdNow - lastSyncedTs < threshold
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="The data syncing is not finished, please wait"
+
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	sleep 130 # wait enough time for pass synced-check-interval
+	# test the case when pdNow - lastSyncedTs > threshold
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="Please check whether PD is online and TiKV Regions are all available. \
+If PD is offline or some TiKV regions are not available, it means that the data syncing process is complete. \
+To check whether TiKV regions are all available, \
+you can view 'TiKV-Details' > 'Resolved-Ts' > 'Max Leader Resolved TS gap' on Grafana. \
+If the gap is large, such as a few minutes, it means that some regions in TiKV are unavailable. \
+Otherwise, if the gap is small and PD is online, it means the data syncing is incomplete, so please wait"
+
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	cleanup_process $CDC_BINARY
+	stop_tidb_cluster
+}
+
+function run_case_with_unavailable_tidb() {
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	config_path=$1
+
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --changefeed-id="test-1" --config="$CUR/$config_path"
+
+	# case 3: test in unavailable tikv cluster
+	run_sql "USE TEST;Create table t1(a int primary key, b int);insert into t1 values(1,2);insert into t1 values(2,3);"
+	check_table_exists "test.t1" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	sleep 5 # make data inserted into downstream
+	kill_tidb
+
+	# test the case when pdNow - lastSyncedTs < threshold
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="The data syncing is not finished, please wait"
+
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	sleep 130 # wait enough time for pass synced-check-interval
+	# test the case when pdNow - lastSyncedTs > threshold
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != true ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="Data syncing is finished"
+
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	cleanup_process $CDC_BINARY
+	stop_tidb_cluster
+}
+
+function run_case_with_failpoint() {
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+
+	# make failpoint to block checkpoint-ts
+	export GO_FAILPOINTS='github.com/pingcap/tiflow/cdc/owner/ChangefeedOwnerNotUpdateCheckpoint=return(true)'
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	config_path=$1
+
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --changefeed-id="test-1" --config="$CUR/$config_path"
+
+	sleep 20 # wait enough time for pass checkpoint-check-interval
+	synced_status=$(curl -X GET http://127.0.0.1:8300/api/v2/changefeeds/test-1/synced)
+	status=$(echo $synced_status | jq '.synced')
+	if [ $status != false ]; then
+		echo "synced status isn't correct"
+		exit 1
+	fi
+	info=$(echo $synced_status | jq -r '.info')
+	target_message="Please check whether PD is online and TiKV Regions are all available. \
+If PD is offline or some TiKV regions are not available, it means that the data syncing process is complete. \
+To check whether TiKV regions are all available, \
+you can view 'TiKV-Details' > 'Resolved-Ts' > 'Max Leader Resolved TS gap' on Grafana. \
+If the gap is large, such as a few minutes, it means that some regions in TiKV are unavailable. \
+Otherwise, if the gap is small and PD is online, it means the data syncing is incomplete, so please wait"
+	if [ "$info" != "$target_message" ]; then
+		echo "synced status info is not correct"
+		exit 1
+	fi
+
+	export GO_FAILPOINTS=''
+
+	cleanup_process $CDC_BINARY
+	stop_tidb_cluster
+}
+
+trap stop_tidb_cluster EXIT
+run_normal_case_and_unavailable_pd "conf/changefeed.toml"
+run_case_with_unavailable_tikv "conf/changefeed.toml"
+run_case_with_unavailable_tidb "conf/changefeed.toml"
+run_case_with_failpoint "conf/changefeed.toml"
+
+# enable redo
+run_normal_case_and_unavailable_pd "conf/changefeed-redo.toml"
+run_case_with_unavailable_tikv "conf/changefeed-redo.toml"
+run_case_with_unavailable_tidb "conf/changefeed-redo.toml"
+run_case_with_failpoint "conf/changefeed-redo.toml"
+
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"