Repository: pingcap/tiflow
Pull request #11296 (2024-06-11 12:37:46)

puller(ticdc): fix wrong update splitting behavior after table scheduling
<!--
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 pingcap/tiflow#11219

### What is changed and how it works?
1. There are two cdc nodes `A` and `B`, and `B` start before `A`, that is `thresholdTSB` < `thresholdTSA`;
2. The sync task of table `t` is first on `A`;
3. Table `t` has an update event which `commitTS` is smaller than `thresholdTSA` and larger than `thresholdTSB`. So the update event is split to a delete event and an insert event on node `A`;
4. But the delete event and insert event cannot be send to the downstream in an atomic way. So if after the delete event is send to downstream and before the insert event being send, the table sync task is scheduling to node `B`, the update event are received by node `B` again;
5. The update event is not split by node `B` because its `commitTS` is larger than the `thresholdTSB`, and node `B` just send an update sql to downstream which cause data inconsistency;

And there is also another thing to notice that after scheduling, node `B` will send some events to downstream which are already send by node `A`; So node `B` must send these events in an idempotent way;
Previously, this is handled by getting a `replicateTS` in sink module when sink starts and split these events which `commitTS` is smaller than `replicateTS`. But this mechanism is also removed in https://github.com/pingcap/tiflow/pull/11030. So we need to handle this case in puller too.

In this pr, instead of maintaining a separate `thresholdTS` in sourcemanager, we try to get the `replicateTS` from sink when puller need to check whether to split the update event.
And since puller module starts working before sink module, so we give `replicateTS` a default value `MAXUInt64` which means to split all update events. After sink starts working, `replicateTS` will be set to the correct value.

The last thing to notice, when sink restarts due to some error, after restart, the sink may send some events downstream which are already send before restart. These events also need be send in an idempotent way. But these events are already in sorter, so just restart sink cannot accomplish this goal. So we forbid restarting sink in this pr and just restart the changefeed when meet error.

### Check List <!--REMOVE the items that are not applicable-->

#### Tests <!-- At least one of them must be included. -->

 - Manual test (add detailed scripts or steps below)
 1. deploy a cluster with three cdc nodes;
 2. kill all nodes occasionally while running workload and check whether the data is consistent;

#### Questions <!-- Authors should answer these questions and reviewers should consider these questions. -->

##### Will it cause performance regression or break compatibility?

##### Do you need to update user documentation, design documentation or monitoring documentation?

### Release note <!-- bugfixes or new features need a release note -->

```release-note
None
```


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

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

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


<!--Checker Notification Identifier-->

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

This pull-request has been approved by:
**Once this PR has been reviewed and has the lgtm label**, please ask for approval from [lidezhu](https://github.com/lidezhu), **ensuring that each of them provides their approval before proceeding**. For more information see [the Code Review Process](https://book.prow.tidb.net/#/workflows/pr).

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

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

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

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":["lidezhu"]} -->
codecov[bot]: ## [Codecov](https://app.codecov.io/gh/pingcap/tiflow/pull/11296?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) Report
Attention: Patch coverage is `35.29412%` with `33 lines` in your changes missing coverage. Please review.
> Project coverage is 57.6594%. Comparing base [(`e3b0bc7`)](https://app.codecov.io/gh/pingcap/tiflow/commit/e3b0bc705cd64f49260160107dd82213ddfd4696?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) to head [(`f25075d`)](https://app.codecov.io/gh/pingcap/tiflow/commit/f25075dcc9635326f2603aaf8e42162918ded5ee?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap).
> Report is 1 commits behind head on master.

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


| [Components](https://app.codecov.io/gh/pingcap/tiflow/pull/11296/components?src=pr&el=components&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | Coverage Δ | |
|---|---|---|
| [cdc](https://app.codecov.io/gh/pingcap/tiflow/pull/11296/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `61.4882% <53.8043%> (+0.0095%)` | :arrow_up: |
| [dm](https://app.codecov.io/gh/pingcap/tiflow/pull/11296/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `51.2091% <ø> (+0.0100%)` | :arrow_up: |
| [engine](https://app.codecov.io/gh/pingcap/tiflow/pull/11296/components?src=pr&el=component&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=pingcap) | `63.3879% <ø> (-0.0142%)` | :arrow_down: |

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

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

```diff
@@               Coverage Diff                @@
##             master     #11296        +/-   ##
================================================
+ Coverage   57.6493%   57.6594%   +0.0100%     
================================================
  Files           850        850                
  Lines        126095     126191        +96     
================================================
+ Hits          72693      72761        +68     
- Misses        48002      48032        +30     
+ Partials       5400       5398         -2     
```

</details>

Patch:
diff --git a/cdc/entry/mounter.go b/cdc/entry/mounter.go
index 82be98ea6..c9513f683 100644
--- a/cdc/entry/mounter.go
+++ b/cdc/entry/mounter.go
@@ -380,11 +380,11 @@ func parseJob(v []byte, startTs, CRTs uint64, fromHistoryTable bool) (*timodel.J
 	}
 
 	if fromHistoryTable {
-		// we only want to get `create table` ddl from tidb_ddl_history, so we just throw out others ddls.
+		// we only want to get `create table` and `create tables` ddl from tidb_ddl_history, so we just throw out others ddls.
 		// We only want the job with `JobStateSynced`, which is means the ddl job is done successfully.
 		// Besides, to satisfy the subsequent processing,
 		// We need to set the job to be Done to make it will replay in schemaStorage
-		if job.Type != timodel.ActionCreateTable || job.State != timodel.JobStateSynced {
+		if (job.Type != timodel.ActionCreateTable && job.Type != timodel.ActionCreateTables) || job.State != timodel.JobStateSynced {
 			return nil, nil
 		}
 		job.State = timodel.JobStateDone
diff --git a/cdc/entry/schema/snapshot.go b/cdc/entry/schema/snapshot.go
index e01821ef5..0af4f370b 100644
--- a/cdc/entry/schema/snapshot.go
+++ b/cdc/entry/schema/snapshot.go
@@ -43,7 +43,7 @@ func (s *Snapshot) PreTableInfo(job *timodel.Job) (*model.TableInfo, error) {
 	switch job.Type {
 	case timodel.ActionCreateSchema, timodel.ActionModifySchemaCharsetAndCollate, timodel.ActionDropSchema:
 		return nil, nil
-	case timodel.ActionCreateTable, timodel.ActionCreateView, timodel.ActionRecoverTable:
+	case timodel.ActionCreateTable, timodel.ActionCreateTables, timodel.ActionCreateView, timodel.ActionRecoverTable:
 		// no pre table info
 		return nil, nil
 	case timodel.ActionRenameTable, timodel.ActionDropTable, timodel.ActionDropView, timodel.ActionTruncateTable, timodel.ActionAlterTablePartitioning, timodel.ActionRemovePartitioning:
@@ -472,6 +472,15 @@ func (s *Snapshot) DoHandleDDL(job *timodel.Job) error {
 		if err != nil {
 			return errors.Trace(err)
 		}
+	case timodel.ActionCreateTables:
+		multiTableInfos := job.BinlogInfo.MultipleTableInfos
+		for _, tableInfo := range multiTableInfos {
+			err := s.inner.createTable(model.WrapTableInfo(job.SchemaID, job.SchemaName,
+				job.BinlogInfo.FinishedTS, tableInfo), job.BinlogInfo.FinishedTS)
+			if err != nil {
+				return errors.Trace(err)
+			}
+		}
 	case timodel.ActionCreateTable, timodel.ActionCreateView, timodel.ActionRecoverTable:
 		err := s.inner.createTable(getWrapTableInfo(job), job.BinlogInfo.FinishedTS)
 		if err != nil {
diff --git a/cdc/entry/schema_storage.go b/cdc/entry/schema_storage.go
index c4ba9c38d..faaa6d4fe 100644
--- a/cdc/entry/schema_storage.go
+++ b/cdc/entry/schema_storage.go
@@ -16,6 +16,7 @@ package entry
 import (
 	"context"
 	"sort"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -407,6 +408,20 @@ func (s *schemaStorage) BuildDDLEvents(
 		if err != nil {
 			return nil, errors.Trace(err)
 		}
+	case timodel.ActionCreateTables:
+		if job.BinlogInfo != nil && job.BinlogInfo.MultipleTableInfos != nil {
+			querys := strings.Split(job.Query, ";")
+			multiTableInfos := job.BinlogInfo.MultipleTableInfos
+			for index, tableInfo := range multiTableInfos {
+				newTableInfo := model.WrapTableInfo(job.SchemaID, job.SchemaName, job.BinlogInfo.FinishedTS, tableInfo)
+				job.Query = querys[index]
+				event := new(model.DDLEvent)
+				event.FromJob(job, nil, newTableInfo)
+				ddlEvents = append(ddlEvents, event)
+			}
+		} else {
+			return nil, errors.Errorf("there is no multiple table infos in the create tables job: %s", job)
+		}
 	default:
 		// parse preTableInfo
 		preSnap, err := s.GetSnapshot(ctx, job.BinlogInfo.FinishedTS-1)
diff --git a/cdc/model/kv.go b/cdc/model/kv.go
index 201c4b4f4..56aceac80 100644
--- a/cdc/model/kv.go
+++ b/cdc/model/kv.go
@@ -16,6 +16,7 @@
 package model
 
 import (
+	"errors"
 	"fmt"
 
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
@@ -110,3 +111,20 @@ func (v *RawKVEntry) String() string {
 func (v *RawKVEntry) ApproximateDataSize() int64 {
 	return int64(len(v.Key) + len(v.Value) + len(v.OldValue))
 }
+
+// ShouldSplitKVEntry checks whether the raw kv entry should be splitted.
+type ShouldSplitKVEntry func(raw *RawKVEntry) bool
+
+// SplitUpdateKVEntry splits the raw kv entry into a delete entry and an insert entry.
+func SplitUpdateKVEntry(raw *RawKVEntry) (*RawKVEntry, *RawKVEntry, error) {
+	if raw == nil {
+		return nil, nil, errors.New("nil event cannot be split")
+	}
+	deleteKVEntry := *raw
+	deleteKVEntry.Value = nil
+
+	insertKVEntry := *raw
+	insertKVEntry.OldValue = nil
+
+	return &deleteKVEntry, &insertKVEntry, nil
+}
diff --git a/cdc/owner/changefeed.go b/cdc/owner/changefeed.go
index f19716f69..18fc88bdb 100755
--- a/cdc/owner/changefeed.go
+++ b/cdc/owner/changefeed.go
@@ -631,7 +631,15 @@ LOOP2:
 
 	c.barriers = newBarriers()
 	if util.GetOrZero(cfInfo.Config.EnableSyncPoint) {
-		c.barriers.Update(syncPointBarrier, c.resolvedTs)
+		// firstSyncPointStartTs = k * syncPointInterval，
+		// which >= startTs, and choose the minimal k
+		syncPointInterval := util.GetOrZero(cfInfo.Config.SyncPointInterval)
+		k := oracle.GetTimeFromTS(c.resolvedTs).Sub(time.Unix(0, 0)) / syncPointInterval
+		if oracle.GetTimeFromTS(c.resolvedTs).Sub(time.Unix(0, 0))%syncPointInterval != 0 || oracle.ExtractLogical(c.resolvedTs) != 0 {
+			k += 1
+		}
+		firstSyncPointTs := oracle.GoTimeToTS(time.Unix(0, 0).Add(k * syncPointInterval))
+		c.barriers.Update(syncPointBarrier, firstSyncPointTs)
 	}
 	c.barriers.Update(finishBarrier, cfInfo.GetTargetTs())
 
diff --git a/cdc/processor/processor.go b/cdc/processor/processor.go
index 382d229d5..76bdbfb50 100644
--- a/cdc/processor/processor.go
+++ b/cdc/processor/processor.go
@@ -223,13 +223,13 @@ func (p *processor) AddTableSpan(
 			zap.Bool("isPrepare", isPrepare))
 	}
 
-	p.sinkManager.r.AddTable(
+	table := p.sinkManager.r.AddTable(
 		span, startTs, p.latestInfo.TargetTs)
 	if p.redo.r.Enabled() {
 		p.redo.r.AddTable(span, startTs)
 	}
-	p.sourceManager.r.AddTable(span, p.getTableName(ctx, span.TableID), startTs)
 
+	p.sourceManager.r.AddTable(span, p.getTableName(ctx, span.TableID), startTs, table.GetReplicaTs)
 	return true, nil
 }
 
@@ -488,18 +488,6 @@ func isProcessorIgnorableError(err error) bool {
 	return false
 }
 
-// needPullerSafeModeAtStart returns true if the scheme is mysql compatible.
-// pullerSafeMode means to split all update kv entries whose commitTS
-// is older then the start time of this changefeed.
-func needPullerSafeModeAtStart(sinkURIStr string) (bool, error) {
-	sinkURI, err := url.Parse(sinkURIStr)
-	if err != nil {
-		return false, cerror.WrapError(cerror.ErrSinkURIInvalid, err)
-	}
-	scheme := sink.GetScheme(sinkURI)
-	return sink.IsMySQLCompatibleScheme(scheme), nil
-}
-
 // Tick implements the `orchestrator.State` interface
 // the `info` parameter is sent by metadata store, the `info` must be the latest value snapshot.
 // the `status` parameter is sent by metadata store, the `status` must be the latest value snapshot.
@@ -600,6 +588,16 @@ func (p *processor) tick(ctx context.Context) (error, error) {
 	return nil, warning
 }
 
+// isMysqlCompatibleBackend returns true if the sinkURIStr is mysql compatible.
+func isMysqlCompatibleBackend(sinkURIStr string) (bool, error) {
+	sinkURI, err := url.Parse(sinkURIStr)
+	if err != nil {
+		return false, cerror.WrapError(cerror.ErrSinkURIInvalid, err)
+	}
+	scheme := sink.GetScheme(sinkURI)
+	return sink.IsMySQLCompatibleScheme(scheme), nil
+}
+
 // lazyInitImpl create Filter, SchemaStorage, Mounter instances at the first tick.
 func (p *processor) lazyInitImpl(etcdCtx context.Context) (err error) {
 	if p.initialized.Load() {
@@ -659,7 +657,7 @@ func (p *processor) lazyInitImpl(etcdCtx context.Context) (err error) {
 		return errors.Trace(err)
 	}
 
-	pullerSafeModeAtStart, err := needPullerSafeModeAtStart(p.latestInfo.SinkURI)
+	isMysqlBackend, err := isMysqlCompatibleBackend(p.latestInfo.SinkURI)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -667,14 +665,14 @@ func (p *processor) lazyInitImpl(etcdCtx context.Context) (err error) {
 		p.changefeedID, p.upstream, p.mg.r,
 		sortEngine, util.GetOrZero(cfConfig.BDRMode),
 		util.GetOrZero(cfConfig.EnableTableMonitor),
-		pullerSafeModeAtStart)
+		isMysqlBackend)
 	p.sourceManager.name = "SourceManager"
 	p.sourceManager.changefeedID = p.changefeedID
 	p.sourceManager.spawn(prcCtx)
 
 	p.sinkManager.r = sinkmanager.New(
 		p.changefeedID, p.latestInfo.SinkURI, cfConfig, p.upstream,
-		p.ddlHandler.r.schemaStorage, p.redo.r, p.sourceManager.r)
+		p.ddlHandler.r.schemaStorage, p.redo.r, p.sourceManager.r, isMysqlBackend)
 	p.sinkManager.name = "SinkManager"
 	p.sinkManager.changefeedID = p.changefeedID
 	p.sinkManager.spawn(prcCtx)
diff --git a/cdc/processor/sinkmanager/manager.go b/cdc/processor/sinkmanager/manager.go
index b1ac8e4a6..d7b986a86 100644
--- a/cdc/processor/sinkmanager/manager.go
+++ b/cdc/processor/sinkmanager/manager.go
@@ -128,6 +128,9 @@ type SinkManager struct {
 	// wg is used to wait for all workers to exit.
 	wg sync.WaitGroup
 
+	// isMysqlBackend indicates whether the backend is MySQL compatible.
+	isMysqlBackend bool
+
 	// Metric for table sink.
 	metricsTableSinkTotalRows prometheus.Counter
 
@@ -143,6 +146,7 @@ func New(
 	schemaStorage entry.SchemaStorage,
 	redoDMLMgr redo.DMLManager,
 	sourceManager *sourcemanager.SourceManager,
+	isMysqlBackend bool,
 ) *SinkManager {
 	m := &SinkManager{
 		changefeedID:        changefeedID,
@@ -156,7 +160,7 @@ func New(
 		sinkTaskChan:        make(chan *sinkTask),
 		sinkWorkerAvailable: make(chan struct{}, 1),
 		sinkRetry:           retry.NewInfiniteErrorRetry(),
-
+		isMysqlBackend:      isMysqlBackend,
 		metricsTableSinkTotalRows: tablesinkmetrics.TotalRowsCountCounter.
 			WithLabelValues(changefeedID.Namespace, changefeedID.ID),
 
@@ -298,6 +302,11 @@ func (m *SinkManager) Run(ctx context.Context, warnings ...chan<- error) (err er
 			if cerror.IsDupEntryError(err) {
 				return errors.Trace(err)
 			}
+
+			if m.isMysqlBackend {
+				// For MySQL backend, we should restart changefeed. Let owner to handle the error.
+				return errors.Trace(err)
+			}
 		}
 
 		// If the error is retryable, we should retry to re-establish the internal resources.
@@ -832,7 +841,7 @@ func (m *SinkManager) UpdateBarrierTs(globalBarrierTs model.Ts, tableBarrier map
 }
 
 // AddTable adds a table(TableSink) to the sink manager.
-func (m *SinkManager) AddTable(span tablepb.Span, startTs model.Ts, targetTs model.Ts) {
+func (m *SinkManager) AddTable(span tablepb.Span, startTs model.Ts, targetTs model.Ts) *tableSinkWrapper {
 	sinkWrapper := newTableSinkWrapper(
 		m.changefeedID,
 		span,
@@ -860,7 +869,6 @@ func (m *SinkManager) AddTable(span tablepb.Span, startTs model.Ts, targetTs mod
 			zap.String("namespace", m.changefeedID.Namespace),
 			zap.String("changefeed", m.changefeedID.ID),
 			zap.Stringer("span", &span))
-		return
 	}
 	m.sinkMemQuota.AddTable(span)
 	m.redoMemQuota.AddTable(span)
@@ -870,6 +878,7 @@ func (m *SinkManager) AddTable(span tablepb.Span, startTs model.Ts, targetTs mod
 		zap.Stringer("span", &span),
 		zap.Uint64("startTs", startTs),
 		zap.Uint64("version", sinkWrapper.version))
+	return sinkWrapper
 }
 
 // StartTable sets the table(TableSink) state to replicating.
diff --git a/cdc/processor/sinkmanager/redo_log_worker.go b/cdc/processor/sinkmanager/redo_log_worker.go
index 7ce4457e6..cb970afd0 100644
--- a/cdc/processor/sinkmanager/redo_log_worker.go
+++ b/cdc/processor/sinkmanager/redo_log_worker.go
@@ -127,7 +127,7 @@ func (w *redoWorker) handleTask(ctx context.Context, task *redoTask) (finalErr e
 		// NOTICE: The event can be filtered by the event filter.
 		if e.Row != nil {
 			// For all events, we add table replicate ts, so mysql sink can determine safe-mode.
-			e.Row.ReplicatingTs = task.tableSink.replicateTs
+			e.Row.ReplicatingTs = task.tableSink.replicateTs.Load()
 			x, size = handleRowChangedEvents(w.changefeedID, task.span, e)
 			advancer.appendEvents(x, size)
 		}
diff --git a/cdc/processor/sinkmanager/table_sink_worker.go b/cdc/processor/sinkmanager/table_sink_worker.go
index cff16a168..d1bd8b60d 100644
--- a/cdc/processor/sinkmanager/table_sink_worker.go
+++ b/cdc/processor/sinkmanager/table_sink_worker.go
@@ -204,7 +204,7 @@ func (w *sinkWorker) handleTask(ctx context.Context, task *sinkTask) (finalErr e
 		// NOTICE: The event can be filtered by the event filter.
 		if e.Row != nil {
 			// For all rows, we add table replicate ts, so mysql sink can determine safe-mode.
-			e.Row.ReplicatingTs = task.tableSink.replicateTs
+			e.Row.ReplicatingTs = task.tableSink.GetReplicaTs()
 			x, size := handleRowChangedEvents(w.changefeedID, task.span, e)
 			advancer.appendEvents(x, size)
 		}
diff --git a/cdc/processor/sinkmanager/table_sink_wrapper.go b/cdc/processor/sinkmanager/table_sink_wrapper.go
index f35437237..dddba30ea 100644
--- a/cdc/processor/sinkmanager/table_sink_wrapper.go
+++ b/cdc/processor/sinkmanager/table_sink_wrapper.go
@@ -15,6 +15,7 @@ package sinkmanager
 
 import (
 	"context"
+	"math"
 	"sort"
 	"sync"
 	"sync/atomic"
@@ -77,7 +78,7 @@ type tableSinkWrapper struct {
 	receivedSorterResolvedTs atomic.Uint64
 
 	// replicateTs is the ts that the table sink has started to replicate.
-	replicateTs    model.Ts
+	replicateTs    atomic.Uint64
 	genReplicateTs func(ctx context.Context) (model.Ts, error)
 
 	// lastCleanTime indicates the last time the table has been cleaned.
@@ -90,6 +91,11 @@ type tableSinkWrapper struct {
 	rangeEventCountsMu sync.Mutex
 }
 
+// GetReplicaTs returns the replicate ts of the table sink.
+func (t *tableSinkWrapper) GetReplicaTs() model.Ts {
+	return t.replicateTs.Load()
+}
+
 type rangeEventCount struct {
 	// firstPos and lastPos are used to merge many rangeEventCount into one.
 	firstPos sorter.Position
@@ -132,31 +138,34 @@ func newTableSinkWrapper(
 
 	res.receivedSorterResolvedTs.Store(startTs)
 	res.barrierTs.Store(startTs)
+	res.replicateTs.Store(math.MaxUint64)
 	return res
 }
 
 func (t *tableSinkWrapper) start(ctx context.Context, startTs model.Ts) (err error) {
-	if t.replicateTs != 0 {
+	if t.replicateTs.Load() != math.MaxUint64 {
 		log.Panic("The table sink has already started",
 			zap.String("namespace", t.changefeed.Namespace),
 			zap.String("changefeed", t.changefeed.ID),
 			zap.Stringer("span", &t.span),
 			zap.Uint64("startTs", startTs),
-			zap.Uint64("oldReplicateTs", t.replicateTs),
+			zap.Uint64("oldReplicateTs", t.replicateTs.Load()),
 		)
 	}
 
 	// FIXME(qupeng): it can be re-fetched later instead of fails.
-	if t.replicateTs, err = t.genReplicateTs(ctx); err != nil {
+	ts, err := t.genReplicateTs(ctx)
+	if err != nil {
 		return errors.Trace(err)
 	}
+	t.replicateTs.Store(ts)
 
 	log.Info("Sink is started",
 		zap.String("namespace", t.changefeed.Namespace),
 		zap.String("changefeed", t.changefeed.ID),
 		zap.Stringer("span", &t.span),
 		zap.Uint64("startTs", startTs),
-		zap.Uint64("replicateTs", t.replicateTs),
+		zap.Uint64("replicateTs", ts),
 	)
 
 	// This start ts maybe greater than the initial start ts of the table sink.
@@ -379,14 +388,16 @@ func (t *tableSinkWrapper) checkTableSinkHealth() (err error) {
 // committed at downstream but we don't know. So we need to update `replicateTs`
 // of the table so that we can re-send those events later.
 func (t *tableSinkWrapper) restart(ctx context.Context) (err error) {
-	if t.replicateTs, err = t.genReplicateTs(ctx); err != nil {
+	ts, err := t.genReplicateTs(ctx)
+	if err != nil {
 		return errors.Trace(err)
 	}
+	t.replicateTs.Store(ts)
 	log.Info("Sink is restarted",
 		zap.String("namespace", t.changefeed.Namespace),
 		zap.String("changefeed", t.changefeed.ID),
 		zap.Stringer("span", &t.span),
-		zap.Uint64("replicateTs", t.replicateTs))
+		zap.Uint64("replicateTs", ts))
 	return nil
 }
 
diff --git a/cdc/processor/sourcemanager/manager.go b/cdc/processor/sourcemanager/manager.go
index b252dc444..3afb21f26 100644
--- a/cdc/processor/sourcemanager/manager.go
+++ b/cdc/processor/sourcemanager/manager.go
@@ -17,7 +17,6 @@ import (
 	"context"
 	"time"
 
-	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tiflow/cdc/entry"
 	"github.com/pingcap/tiflow/cdc/kv"
@@ -28,13 +27,9 @@ import (
 	"github.com/pingcap/tiflow/cdc/processor/tablepb"
 	"github.com/pingcap/tiflow/cdc/puller"
 	"github.com/pingcap/tiflow/pkg/config"
-	cerrors "github.com/pingcap/tiflow/pkg/errors"
-	"github.com/pingcap/tiflow/pkg/retry"
 	"github.com/pingcap/tiflow/pkg/txnutil"
 	"github.com/pingcap/tiflow/pkg/upstream"
-	"github.com/tikv/client-go/v2/oracle"
 	"github.com/tikv/client-go/v2/tikv"
-	pd "github.com/tikv/pd/client"
 	"go.uber.org/zap"
 )
 
@@ -55,7 +50,8 @@ type SourceManager struct {
 	engine sorter.SortEngine
 	// Used to indicate whether the changefeed is in BDR mode.
 	bdrMode bool
-	startTs model.Ts
+
+	safeModeAtStart bool
 
 	enableTableMonitor bool
 	puller             *puller.MultiplexingPuller
@@ -92,21 +88,8 @@ func NewForTest(
 	}
 }
 
-func isOldUpdateKVEntry(raw *model.RawKVEntry, thresholdTs model.Ts) bool {
-	return raw != nil && raw.IsUpdate() && raw.CRTs < thresholdTs
-}
-
-func splitUpdateKVEntry(raw *model.RawKVEntry) (*model.RawKVEntry, *model.RawKVEntry, error) {
-	if raw == nil {
-		return nil, nil, errors.New("nil event cannot be split")
-	}
-	deleteKVEntry := *raw
-	deleteKVEntry.Value = nil
-
-	insertKVEntry := *raw
-	insertKVEntry.OldValue = nil
-
-	return &deleteKVEntry, &insertKVEntry, nil
+func isOldUpdateKVEntry(raw *model.RawKVEntry, getReplicaTs func() model.Ts) bool {
+	return raw != nil && raw.IsUpdate() && raw.CRTs < getReplicaTs()
 }
 
 func newSourceManager(
@@ -126,6 +109,7 @@ func newSourceManager(
 		engine:             engine,
 		bdrMode:            bdrMode,
 		enableTableMonitor: enableTableMonitor,
+		safeModeAtStart:    safeModeAtStart,
 	}
 
 	serverConfig := config.GetGlobalServerConfig()
@@ -138,15 +122,15 @@ func newSourceManager(
 
 	// consume add raw kv entry to the engine.
 	// It will be called by the puller when new raw kv entry is received.
-	consume := func(ctx context.Context, raw *model.RawKVEntry, spans []tablepb.Span) error {
+	consume := func(ctx context.Context, raw *model.RawKVEntry, spans []tablepb.Span, shouldSplitKVEntry model.ShouldSplitKVEntry) error {
 		if len(spans) > 1 {
 			log.Panic("DML puller subscribes multiple spans",
 				zap.String("namespace", mgr.changefeedID.Namespace),
 				zap.String("changefeed", mgr.changefeedID.ID))
 		}
 		if raw != nil {
-			if safeModeAtStart && isOldUpdateKVEntry(raw, mgr.startTs) {
-				deleteKVEntry, insertKVEntry, err := splitUpdateKVEntry(raw)
+			if shouldSplitKVEntry(raw) {
+				deleteKVEntry, insertKVEntry, err := model.SplitUpdateKVEntry(raw)
 				if err != nil {
 					return err
 				}
@@ -175,13 +159,17 @@ func newSourceManager(
 }
 
 // AddTable adds a table to the source manager. Start puller and register table to the engine.
-func (m *SourceManager) AddTable(span tablepb.Span, tableName string, startTs model.Ts) {
+func (m *SourceManager) AddTable(span tablepb.Span, tableName string, startTs model.Ts, getReplicaTs func() model.Ts) {
 	// Add table to the engine first, so that the engine can receive the events from the puller.
 	m.engine.AddTable(span, startTs)
 
+	shouldSplitKVEntry := func(raw *model.RawKVEntry) bool {
+		return m.safeModeAtStart && isOldUpdateKVEntry(raw, getReplicaTs)
+	}
+
 	// Only nil in unit tests.
 	if m.puller != nil {
-		m.puller.Subscribe([]tablepb.Span{span}, startTs, tableName)
+		m.puller.Subscribe([]tablepb.Span{span}, startTs, tableName, shouldSplitKVEntry)
 	}
 }
 
@@ -227,11 +215,6 @@ func (m *SourceManager) Run(ctx context.Context, _ ...chan<- error) error {
 	if m.puller == nil {
 		return nil
 	}
-	startTs, err := getCurrentTs(ctx, m.up.PDClient)
-	if err != nil {
-		return err
-	}
-	m.startTs = startTs
 	return m.puller.Run(ctx)
 }
 
@@ -273,23 +256,3 @@ func (m *SourceManager) Close() {
 func (m *SourceManager) Add(span tablepb.Span, events ...*model.PolymorphicEvent) {
 	m.engine.Add(span, events...)
 }
-
-func getCurrentTs(ctx context.Context, pdClient pd.Client) (model.Ts, error) {
-	backoffBaseDelayInMs := int64(100)
-	totalRetryDuration := 10 * time.Second
-	var replicateTs model.Ts
-	err := retry.Do(ctx, func() error {
-		phy, logic, err := pdClient.GetTS(ctx)
-		if err != nil {
-			return errors.Trace(err)
-		}
-		replicateTs = oracle.ComposeTS(phy, logic)
-		return nil
-	}, retry.WithBackoffBaseDelay(backoffBaseDelayInMs),
-		retry.WithTotalRetryDuratoin(totalRetryDuration),
-		retry.WithIsRetryableErr(cerrors.IsRetryableError))
-	if err != nil {
-		return model.Ts(0), errors.Trace(err)
-	}
-	return replicateTs, nil
-}
diff --git a/cdc/puller/ddl_puller.go b/cdc/puller/ddl_puller.go
index e50ad06bf..70323692b 100644
--- a/cdc/puller/ddl_puller.go
+++ b/cdc/puller/ddl_puller.go
@@ -17,6 +17,7 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
+	"strings"
 	"sync"
 	"sync/atomic"
 	"time"
@@ -121,7 +122,7 @@ func NewDDLJobPuller(
 
 	slots, hasher := 1, func(tablepb.Span, int) int { return 0 }
 	ddlJobPuller.mp = NewMultiplexingPuller(changefeed, client, up.PDClock, ddlJobPuller.Input, slots, hasher, 1)
-	ddlJobPuller.mp.Subscribe(ddlSpans, checkpointTs, memorysorter.DDLPullerTableName)
+	ddlJobPuller.mp.Subscribe(ddlSpans, checkpointTs, memorysorter.DDLPullerTableName, func(_ *model.RawKVEntry) bool { return false })
 
 	return ddlJobPuller
 }
@@ -174,6 +175,7 @@ func (p *ddlJobPullerImpl) Input(
 	ctx context.Context,
 	rawDDL *model.RawKVEntry,
 	_ []tablepb.Span,
+	_ model.ShouldSplitKVEntry,
 ) error {
 	p.sorter.AddEntry(ctx, model.NewPolymorphicEvent(rawDDL))
 	return nil
@@ -390,6 +392,28 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
 			return false, cerror.WrapError(cerror.ErrHandleDDLFailed,
 				errors.Trace(err), job.Query, job.StartTS, job.StartTS)
 		}
+	case timodel.ActionCreateTables:
+		// we only use multiTableInfos and Querys when we generate job event
+		// So if some table should be discard, we just need to delete the info from multiTableInfos and Querys
+		var newMultiTableInfos []*timodel.TableInfo
+		var newQuerys []string
+
+		multiTableInfos := job.BinlogInfo.MultipleTableInfos
+		querys := strings.Split(job.Query, ";")
+
+		for index, tableInfo := range multiTableInfos {
+			// judge each table whether need to be skip
+			if p.filter.ShouldDiscardDDL(job.Type, job.SchemaName, tableInfo.Name.O) {
+				continue
+			}
+			newMultiTableInfos = append(newMultiTableInfos, multiTableInfos[index])
+			newQuerys = append(newQuerys, querys[index])
+		}
+
+		skip = len(newMultiTableInfos) == 0
+
+		job.BinlogInfo.MultipleTableInfos = newMultiTableInfos
+		job.Query = strings.Join(newQuerys, ";")
 	case timodel.ActionRenameTable:
 		oldTable, ok := snap.PhysicalTableByID(job.TableID)
 		if !ok {
diff --git a/cdc/puller/multiplexing_puller.go b/cdc/puller/multiplexing_puller.go
index f9fc25d64..792be0bab 100644
--- a/cdc/puller/multiplexing_puller.go
+++ b/cdc/puller/multiplexing_puller.go
@@ -134,7 +134,7 @@ type MultiplexingPuller struct {
 	changefeed model.ChangeFeedID
 	client     *kv.SharedClient
 	pdClock    pdutil.Clock
-	consume    func(context.Context, *model.RawKVEntry, []tablepb.Span) error
+	consume    func(context.Context, *model.RawKVEntry, []tablepb.Span, model.ShouldSplitKVEntry) error
 	// inputChannelIndexer is used to determine which input channel to use for a given span.
 	inputChannelIndexer func(span tablepb.Span, workerCount int) int
 
@@ -172,7 +172,7 @@ func NewMultiplexingPuller(
 	changefeed model.ChangeFeedID,
 	client *kv.SharedClient,
 	pdClock pdutil.Clock,
-	consume func(context.Context, *model.RawKVEntry, []tablepb.Span) error,
+	consume func(context.Context, *model.RawKVEntry, []tablepb.Span, model.ShouldSplitKVEntry) error,
 	workerCount int,
 	inputChannelIndexer func(tablepb.Span, int) int,
 	resolvedTsAdvancerCount int,
@@ -197,16 +197,22 @@ func NewMultiplexingPuller(
 }
 
 // Subscribe some spans. They will share one same resolved timestamp progress.
-func (p *MultiplexingPuller) Subscribe(spans []tablepb.Span, startTs model.Ts, tableName string) {
+func (p *MultiplexingPuller) Subscribe(
+	spans []tablepb.Span,
+	startTs model.Ts,
+	tableName string,
+	shouldSplitKVEntry model.ShouldSplitKVEntry,
+) {
 	p.subscriptions.Lock()
 	defer p.subscriptions.Unlock()
-	p.subscribe(spans, startTs, tableName)
+	p.subscribe(spans, startTs, tableName, shouldSplitKVEntry)
 }
 
 func (p *MultiplexingPuller) subscribe(
 	spans []tablepb.Span,
 	startTs model.Ts,
 	tableName string,
+	shouldSplitKVEntry model.ShouldSplitKVEntry,
 ) {
 	for _, span := range spans {
 		// Base on the current design, a MultiplexingPuller is only used for one changefeed.
@@ -240,7 +246,7 @@ func (p *MultiplexingPuller) subscribe(
 		progress.consume.RLock()
 		defer progress.consume.RUnlock()
 		if !progress.consume.removed {
-			return p.consume(ctx, raw, spans)
+			return p.consume(ctx, raw, spans, shouldSplitKVEntry)
 		}
 		return nil
 	}
diff --git a/pkg/filter/filter.go b/pkg/filter/filter.go
index d7c73a729..3441322ae 100644
--- a/pkg/filter/filter.go
+++ b/pkg/filter/filter.go
@@ -37,6 +37,7 @@ var ddlWhiteListMap = map[timodel.ActionType]bf.EventType{
 
 	// table related DDLs
 	timodel.ActionCreateTable:                  bf.CreateTable,
+	timodel.ActionCreateTables:                 bf.CreateTable,
 	timodel.ActionDropTable:                    bf.DropTable,
 	timodel.ActionTruncateTable:                bf.TruncateTable,
 	timodel.ActionRenameTable:                  bf.RenameTable,


Test patch:
diff --git a/cdc/owner/changefeed_test.go b/cdc/owner/changefeed_test.go
index 0866638b0..c384ecd59 100644
--- a/cdc/owner/changefeed_test.go
+++ b/cdc/owner/changefeed_test.go
@@ -684,13 +684,7 @@ func TestBarrierAdvance(t *testing.T) {
 			require.Nil(t, err)
 			err = cf.handleBarrier(ctx, state.Info, state.Status, barrier)
 
-			nextSyncPointTs := oracle.GoTimeToTS(
-				oracle.GetTimeFromTS(state.Status.CheckpointTs + 10).
-					Add(util.GetOrZero(changefeedInfo.Config.SyncPointInterval)),
-			)
-
 			require.Nil(t, err)
-			require.Equal(t, nextSyncPointTs, barrier.GlobalBarrierTs)
 			require.Less(t, state.Status.CheckpointTs+10, barrier.GlobalBarrierTs)
 			require.Less(t, barrier.GlobalBarrierTs, cf.ddlManager.ddlResolvedTs)
 		}
diff --git a/cdc/processor/sinkmanager/manager_test.go b/cdc/processor/sinkmanager/manager_test.go
index 362ab7226..0532dcb3b 100644
--- a/cdc/processor/sinkmanager/manager_test.go
+++ b/cdc/processor/sinkmanager/manager_test.go
@@ -135,7 +135,7 @@ func TestAddTable(t *testing.T) {
 	require.Equal(t, 0, manager.sinkProgressHeap.len(), "Not started table shout not in progress heap")
 	err := manager.StartTable(span, 1)
 	require.NoError(t, err)
-	require.Equal(t, uint64(0x7ffffffffffbffff), tableSink.(*tableSinkWrapper).replicateTs)
+	require.Equal(t, uint64(0x7ffffffffffbffff), tableSink.(*tableSinkWrapper).replicateTs.Load())
 
 	progress := manager.sinkProgressHeap.pop()
 	require.Equal(t, span, progress.span)
@@ -359,7 +359,7 @@ func TestSinkManagerRunWithErrors(t *testing.T) {
 
 	span := spanz.TableIDToComparableSpan(1)
 
-	source.AddTable(span, "test", 100)
+	source.AddTable(span, "test", 100, func() model.Ts { return 0 })
 	manager.AddTable(span, 100, math.MaxUint64)
 	manager.StartTable(span, 100)
 	source.Add(span, model.NewResolvedPolymorphicEvent(0, 101))
diff --git a/cdc/processor/sinkmanager/manager_test_helper.go b/cdc/processor/sinkmanager/manager_test_helper.go
index 8f7fd65a3..3e1e483fc 100644
--- a/cdc/processor/sinkmanager/manager_test_helper.go
+++ b/cdc/processor/sinkmanager/manager_test_helper.go
@@ -71,7 +71,7 @@ func CreateManagerWithMemEngine(
 	sourceManager.WaitForReady(ctx)
 
 	sinkManager := New(changefeedID, changefeedInfo.SinkURI,
-		changefeedInfo.Config, up, schemaStorage, nil, sourceManager)
+		changefeedInfo.Config, up, schemaStorage, nil, sourceManager, false)
 	go func() { handleError(sinkManager.Run(ctx)) }()
 	sinkManager.WaitForReady(ctx)
 
@@ -92,6 +92,6 @@ func NewManagerWithMemEngine(
 	schemaStorage := &entry.MockSchemaStorage{Resolved: math.MaxUint64}
 	sourceManager := sourcemanager.NewForTest(changefeedID, up, mg, sortEngine, false)
 	sinkManager := New(changefeedID, changefeedInfo.SinkURI,
-		changefeedInfo.Config, up, schemaStorage, redoMgr, sourceManager)
+		changefeedInfo.Config, up, schemaStorage, redoMgr, sourceManager, false)
 	return sinkManager, sourceManager, sortEngine
 }
diff --git a/cdc/puller/ddl_puller_test.go b/cdc/puller/ddl_puller_test.go
index 2b08ddd3d..d34422d0f 100644
--- a/cdc/puller/ddl_puller_test.go
+++ b/cdc/puller/ddl_puller_test.go
@@ -66,12 +66,12 @@ func tsToRawKVEntry(_ *testing.T, ts model.Ts) *model.RawKVEntry {
 
 func inputDDL(t *testing.T, puller *ddlJobPullerImpl, job *timodel.Job) {
 	rawJob := jonToRawKVEntry(t, job)
-	puller.Input(context.Background(), rawJob, []tablepb.Span{})
+	puller.Input(context.Background(), rawJob, []tablepb.Span{}, func(_ *model.RawKVEntry) bool { return false })
 }
 
 func inputTs(t *testing.T, puller *ddlJobPullerImpl, ts model.Ts) {
 	rawTs := tsToRawKVEntry(t, ts)
-	puller.Input(context.Background(), rawTs, []tablepb.Span{})
+	puller.Input(context.Background(), rawTs, []tablepb.Span{}, func(_ *model.RawKVEntry) bool { return false })
 }
 
 func waitResolvedTs(t *testing.T, p DDLJobPuller, targetTs model.Ts) {
diff --git a/cdc/puller/multiplexing_puller_test.go b/cdc/puller/multiplexing_puller_test.go
index d3d495e94..6563c0f35 100644
--- a/cdc/puller/multiplexing_puller_test.go
+++ b/cdc/puller/multiplexing_puller_test.go
@@ -31,7 +31,7 @@ import (
 func newMultiplexingPullerForTest(outputCh chan<- *model.RawKVEntry) *MultiplexingPuller {
 	cfg := &config.ServerConfig{Debug: &config.DebugConfig{Puller: &config.PullerConfig{LogRegionDetails: false}}}
 	client := kv.NewSharedClient(model.ChangeFeedID{}, cfg, false, nil, nil, nil, nil, nil)
-	consume := func(ctx context.Context, e *model.RawKVEntry, _ []tablepb.Span) error {
+	consume := func(ctx context.Context, e *model.RawKVEntry, _ []tablepb.Span, _ model.ShouldSplitKVEntry) error {
 		select {
 		case <-ctx.Done():
 			return ctx.Err()
@@ -87,7 +87,10 @@ func TestMultiplexingPullerResolvedForward(t *testing.T) {
 
 	spans := []tablepb.Span{spanz.ToSpan([]byte("t_a"), []byte("t_e"))}
 	spans[0].TableID = 1
-	puller.subscribe(spans, 996, "test")
+	shouldSplitKVEntry := func(raw *model.RawKVEntry) bool {
+		return false
+	}
+	puller.subscribe(spans, 996, "test", shouldSplitKVEntry)
 	subID := puller.subscriptions.n.GetV(spans[0]).subID
 	for _, event := range events {
 		puller.inputChs[0] <- kv.MultiplexingEvent{RegionFeedEvent: event, SubscriptionID: subID}
diff --git a/pkg/filter/filter_test.go b/pkg/filter/filter_test.go
index 9ed19e04c..1d296eb0d 100644
--- a/pkg/filter/filter_test.go
+++ b/pkg/filter/filter_test.go
@@ -371,7 +371,7 @@ func TestShouldDiscardDDL(t *testing.T) {
 }
 
 func TestIsAllowedDDL(t *testing.T) {
-	require.Len(t, ddlWhiteListMap, 36)
+	require.Len(t, ddlWhiteListMap, 37)
 	type testCase struct {
 		timodel.ActionType
 		allowed bool
diff --git a/tests/integration_tests/batch_add_table/conf/changefeed.toml b/tests/integration_tests/batch_add_table/conf/changefeed.toml
new file mode 100644
index 000000000..c125b104e
--- /dev/null
+++ b/tests/integration_tests/batch_add_table/conf/changefeed.toml
@@ -0,0 +1,2 @@
+[filter]
+rules = ['*.*','!test.t_1']
\ No newline at end of file
diff --git a/tests/integration_tests/batch_add_table/data/prepare.sql b/tests/integration_tests/batch_add_table/data/prepare.sql
index 83674029f..1f37cf311 100644
--- a/tests/integration_tests/batch_add_table/data/prepare.sql
+++ b/tests/integration_tests/batch_add_table/data/prepare.sql
@@ -7,9 +7,31 @@ create table a2 (id int primary key);
 create table a3 (id int primary key);
 create table a4 (id int primary key);
 create table a5 (id int primary key);
+create table a6 (id int primary key);
+create table a7 (id int primary key);
+create table a8 (id int primary key);
+create table a9 (id int primary key);
+create table a10 (id int primary key);
+create table a11 (id int primary key);
+create table a12 (id int primary key);
+create table a13 (id int primary key);
+create table a14 (id int primary key);
+create table a15 (id int primary key);
+create table a16 (id int primary key);
 
 insert into a1 values (1);
 insert into a2 values (1);
 insert into a3 values (1);
 insert into a4 values (1);
-insert into a5 values (1);
\ No newline at end of file
+insert into a5 values (1);
+insert into a6 values (1);
+insert into a7 values (1);
+insert into a8 values (1);
+insert into a9 values (1);
+insert into a10 values (1);
+insert into a11 values (1);
+insert into a12 values (1);
+insert into a13 values (1);
+insert into a14 values (1);
+insert into a15 values (1);
+insert into a16 values (1);
\ No newline at end of file
diff --git a/tests/integration_tests/batch_add_table/data/test.sql b/tests/integration_tests/batch_add_table/data/test.sql
index 77afc7191..f51b65b34 100644
--- a/tests/integration_tests/batch_add_table/data/test.sql
+++ b/tests/integration_tests/batch_add_table/data/test.sql
@@ -7,6 +7,16 @@ insert into a2 values (2);
 insert into a3 values (2);
 insert into a4 values (2);
 insert into a5 values (2);
-
+insert into a6 values (2);
+insert into a7 values (2);
+insert into a8 values (2);
+insert into a9 values (2);
+insert into a10 values (2);
+insert into a11 values (2);
+insert into a12 values (2);
+insert into a13 values (2);
+insert into a14 values (2);
+insert into a15 values (2);
+insert into a16 values (2);
 
 create table finish_mark (id int primary key);
\ No newline at end of file
diff --git a/tests/integration_tests/batch_add_table/run.sh b/tests/integration_tests/batch_add_table/run.sh
index 9cce075c5..427a496e2 100644
--- a/tests/integration_tests/batch_add_table/run.sh
+++ b/tests/integration_tests/batch_add_table/run.sh
@@ -17,9 +17,6 @@ function run_with_fast_create_table() {
 
 	run_sql "set global tidb_enable_fast_create_table=on" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
-	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	run_sql_file $CUR/data/prepare.sql ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
 
 	TOPIC_NAME="ticdc-batch-add-table-test-$RANDOM"
@@ -32,16 +29,31 @@ function run_with_fast_create_table() {
 		;;
 	*) SINK_URI="mysql://normal:123456@127.0.0.1:3306/" ;;
 	esac
-	run_cdc_cli changefeed create --sink-uri="$SINK_URI"
+	run_cdc_cli changefeed create --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml"
 	case $SINK_TYPE in
 	kafka) run_kafka_consumer $WORK_DIR "kafka://127.0.0.1:9092/$TOPIC_NAME?protocol=open-protocol&partition-num=4&version=${KAFKA_VERSION}&max-message-bytes=10485760" ;;
 	storage) run_storage_consumer $WORK_DIR $SINK_URI "" "" ;;
 	pulsar) run_pulsar_consumer --upstream-uri $SINK_URI ;;
 	esac
 
+	## to generate batch create ddl. In changefeed.toml, we filter test.t_1
+	for ((i = 1; i <= 100; i++)); do
+		mysql -h ${UP_TIDB_HOST} -P ${UP_TIDB_PORT} -u root -D "test" -e "create table t_$i (a int primary key , b int)" &
+	done
+
+	check_table_exists test.t_100 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	for ((i = 1; i <= 100; i++)); do
+		mysql -h ${UP_TIDB_HOST} -P ${UP_TIDB_PORT} -u root -D "test" -e "insert into t_$i values(1,2)"
+	done
+
+	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql_file $CUR/data/test.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
 	check_table_exists batch_add_table.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	# check the ddl of this table is skipped
+	check_table_not_exists test.t_1 ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
 
 	cleanup_process $CDC_BINARY
@@ -56,9 +68,6 @@ function run_without_fast_create_table() {
 
 	run_sql "set global tidb_enable_fast_create_table=off" ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 
-	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
-	run_sql_file $CUR/data/prepare.sql ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
-
 	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
 
 	TOPIC_NAME="ticdc-batch-add-table-test-$RANDOM"
@@ -78,7 +87,9 @@ function run_without_fast_create_table() {
 	pulsar) run_pulsar_consumer --upstream-uri $SINK_URI ;;
 	esac
 
+	run_sql_file $CUR/data/prepare.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
 	run_sql_file $CUR/data/test.sql ${UP_TIDB_HOST} ${UP_TIDB_PORT}
+
 	# sync_diff can't check non-exist table, so we check expected tables are created in downstream first
 	check_table_exists batch_add_table.finish_mark ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
 	check_sync_diff $WORK_DIR $CUR/conf/diff_config.toml
diff --git a/tests/integration_tests/run_group.sh b/tests/integration_tests/run_group.sh
index ebcb41681..f8f15673e 100755
--- a/tests/integration_tests/run_group.sh
+++ b/tests/integration_tests/run_group.sh
@@ -11,7 +11,7 @@ group_num=${group#G}
 # Other tests that only support mysql: batch_update_to_no_batch ddl_reentrant
 # changefeed_fast_fail changefeed_resume_with_checkpoint_ts sequence
 # multi_cdc_cluster capture_suicide_while_balance_table
-mysql_only="bdr_mode capture_suicide_while_balance_table syncpoint hang_sink_suicide server_config_compatibility changefeed_dup_error_restart"
+mysql_only="bdr_mode capture_suicide_while_balance_table syncpoint syncpoint_check_ts hang_sink_suicide server_config_compatibility changefeed_dup_error_restart"
 mysql_only_http="http_api http_api_tls api_v2 http_api_tls_with_user_auth cli_tls_with_auth"
 mysql_only_consistent_replicate="consistent_replicate_ddl consistent_replicate_gbk consistent_replicate_nfs consistent_replicate_storage_file consistent_replicate_storage_file_large_value consistent_replicate_storage_s3 consistent_partition_table"
 
diff --git a/tests/integration_tests/syncpoint_check_ts/conf/changefeed.toml b/tests/integration_tests/syncpoint_check_ts/conf/changefeed.toml
new file mode 100644
index 000000000..d01100dc3
--- /dev/null
+++ b/tests/integration_tests/syncpoint_check_ts/conf/changefeed.toml
@@ -0,0 +1,2 @@
+enable-sync-point = true
+sync-point-interval = "30s"
diff --git a/tests/integration_tests/syncpoint_check_ts/run.sh b/tests/integration_tests/syncpoint_check_ts/run.sh
new file mode 100644
index 000000000..f6fb07cf9
--- /dev/null
+++ b/tests/integration_tests/syncpoint_check_ts/run.sh
@@ -0,0 +1,72 @@
+#!/bin/bash
+
+# [DISCRIPTION]:
+#   The test is used to check the first sync point ts value when we set different SyncPointInterval
+# [STEP]:
+#   1. Create changefeed with --sync-point --sync-interval=30s
+#   2. After test, get fisrt syncpoints from tidb_cdc.syncpoint_v1
+#   3. check the first syncpoint ts value
+
+set -eu
+
+CUR=$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)
+source $CUR/../_utils/test_prepare
+WORK_DIR=$OUT_DIR/$TEST_NAME
+CDC_BINARY=cdc.test
+SINK_TYPE=$1
+
+CDC_COUNT=3
+DB_COUNT=4
+
+function run() {
+	if [ "$SINK_TYPE" != "mysql" ]; then
+		echo "kafka downstream isn't support syncpoint record"
+		return
+	fi
+
+	rm -rf $WORK_DIR && mkdir -p $WORK_DIR
+
+	start_tidb_cluster --workdir $WORK_DIR
+
+	cd $WORK_DIR
+	run_sql "SET GLOBAL tidb_enable_external_ts_read = on;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+
+	start_ts=$(run_cdc_cli_tso_query ${UP_PD_HOST_1} ${UP_PD_PORT_1})
+	run_cdc_server --workdir $WORK_DIR --binary $CDC_BINARY
+
+	# this test contains `set global tidb_external_ts = ?` , which requires super privilege, so we
+	# can't use the normal user
+	SINK_URI="mysql://root@127.0.0.1:3306/?max-txn-row=1"
+	run_cdc_cli changefeed create --start-ts=$start_ts --sink-uri="$SINK_URI" --config="$CUR/conf/changefeed.toml"
+
+	run_sql "SET GLOBAL tidb_enable_external_ts_read = off;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	sleep 60
+
+	run_sql "SELECT primary_ts, secondary_ts FROM tidb_cdc.syncpoint_v1 order by primary_ts limit 1;" ${DOWN_TIDB_HOST} ${DOWN_TIDB_PORT}
+	echo "____________________________________"
+	cat "$OUT_DIR/sql_res.$TEST_NAME.txt"
+	primary_ts=($(grep primary_ts $OUT_DIR/sql_res.$TEST_NAME.txt | awk -F ": " '{print $2}'))
+	echo "primary_ts is " $primary_ts "start_ts is " $start_ts
+
+	shifted_primary_ts=$(($primary_ts >> 18))
+	if [ $(($shifted_primary_ts % 30000)) -eq 0 ]; then
+		pre_primary_ts=$(($primary_ts >> 18 - 30000))
+		pre_primary_ts=$(($pre_primary_ts << 18))
+		if [ $pre_primary_ts -lt $start_ts ]; then
+			echo "check successfully"
+		else
+			echo "primary ts is not correct, there should be a smaller primary ts as the first sync point ts"
+			exit 1
+		fi
+	else
+		echo "primary ts is not correct, the difference between it and the time.Unix(0,0) should be the integer multiples of the syncPointInterval "
+		exit 1
+	fi
+
+	cleanup_process $CDC_BINARY
+}
+
+trap stop_tidb_cluster EXIT
+run $*
+check_logs $WORK_DIR
+echo "[$(date)] <<<<<< run test case $TEST_NAME success! >>>>>>"