Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

ddl_puller.go(ticdc): fix DDLs are ignored when schema versions are out of order #11733

Merged
merged 5 commits into from
Nov 15, 2024
Merged

ddl_puller.go(ticdc): fix DDLs are ignored when schema versions are out of order #11733

Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
880e28f
ddl_puller.go: fix DDLs are ignored when schema versions are out of o…
wlwilliamx Nov 8, 2024
aef5ff6
ddl_puller.go: deprecate the `schemaVersion` field in `ddlJobPullerImpl`
wlwilliamx Nov 15, 2024
7a4eeef
DDLPuller: add a UT for handling out-of-order schema versions
wlwilliamx Nov 15, 2024
aab6542
ddl_puller.go: delete the unused field `schemaVersion` in `ddlJobPull…
wlwilliamx Nov 15, 2024
a45ec46
Revert "DDLPuller: add a UT for handling out-of-order schema versions"
wlwilliamx Nov 15, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 1 addition & 4 deletions cdc/puller/ddl_puller.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -73,7 +73,6 @@ type ddlJobPullerImpl struct {
kvStorage tidbkv.Storage
schemaStorage entry.SchemaStorage
resolvedTs uint64
schemaVersion int64
filter filter.Filter
// ddlTableInfo is initialized when receive the first concurrent DDL job.
ddlTableInfo *entry.DDLTableInfo
Expand Down Expand Up @@ -317,8 +316,7 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
return false, nil
}

if job.BinlogInfo.FinishedTS <= p.getResolvedTs() ||
job.BinlogInfo.SchemaVersion <= p.schemaVersion {
if job.BinlogInfo.FinishedTS <= p.getResolvedTs() {
log.Info("ddl job finishedTs less than puller resolvedTs,"+
"discard the ddl job",
zap.String("namespace", p.changefeedID.Namespace),
Expand Down Expand Up @@ -480,7 +478,6 @@ func (p *ddlJobPullerImpl) handleJob(job *timodel.Job) (skip bool, err error) {
errors.Trace(err), job.Query, job.StartTS, job.StartTS)
}
p.setResolvedTs(job.BinlogInfo.FinishedTS)
p.schemaVersion = job.BinlogInfo.SchemaVersion

return p.checkIneligibleTableDDL(snap, job)
}
Expand Down
Loading