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

Jump to bottom

drainer: bugfix, handle "missing column" when a column is getting dro… #844

Merged
merged 5 commits into from
Dec 23, 2019
Merged

drainer: bugfix, handle "missing column" when a column is getting dro… #844

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
8f5207f
drainer: bugfix, handle "missing column" when a column is getting dro…
suzaku Dec 10, 2019
e803c1e
go mod tidy
suzaku Dec 10, 2019
32a7445
Merge branch 'release-2.1' into pick-drop-col-2.1
july2993 Dec 20, 2019
323db43
Make sure insert is executed before drop col
suzaku Dec 20, 2019
f3b5cac
Set DdlSchemaState
suzaku Dec 23, 2019
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
6 changes: 5 additions & 1 deletion drainer/collector.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -135,7 +135,11 @@ func (c *Collector) publishBinlogs(ctx context.Context) {
log.Warnf("unexpected job, job id %d state: %v", job.ID, job.State)
}

if !skipJob(job) {
isDelOnlyEvent := model.SchemaState(binlog.DdlSchemaState) == model.StateDeleteOnly
if !skipJob(job) || isDelOnlyEvent {
if isDelOnlyEvent {
job.SchemaState = model.StateDeleteOnly
}
item.SetJob(job)
c.syncer.Add(item)
ddlJobsCounter.Add(float64(1))
Expand Down
57 changes: 37 additions & 20 deletions drainer/schema.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@ type Schema struct {
tables map[int64]*model.TableInfo

truncateTableID map[int64]struct{}
tblsDroppingCol map[int64]bool

schemaMetaVersion int64

Expand All @@ -42,6 +43,7 @@ func NewSchema(jobs []*model.Job, hasImplicitCol bool) (*Schema, error) {
hasImplicitCol: hasImplicitCol,
version2SchemaTable: make(map[int64]TableName),
truncateTableID: make(map[int64]struct{}),
tblsDroppingCol: make(map[int64]bool),
jobs: jobs,
}

Expand Down Expand Up @@ -217,30 +219,32 @@ func (s *Schema) addJob(job *model.Job) {
func (s *Schema) handlePreviousDDLJobIfNeed(version int64) error {
var i int
for i = 0; i < len(s.jobs); i++ {
if skipJob(s.jobs[i]) {
log.Debugf("skip ddl job %v", s.jobs[i])
job := s.jobs[i]

if job.BinlogInfo.SchemaVersion > version {
break
}

if s.jobs[i].BinlogInfo.SchemaVersion <= s.currentVersion {
log.Warnf("ddl job %v schema version is less than current version %d, skip this ddl job", s.jobs[i], s.currentVersion)
continue
}

if s.jobs[i].BinlogInfo.SchemaVersion <= version {
if s.jobs[i].BinlogInfo.SchemaVersion <= s.currentVersion {
log.Warnf("ddl job %v schema version is less than current version %d, skip this ddl job", s.jobs[i], s.currentVersion)
continue
}

data, err := json.Marshal(s.jobs[i])
if err != nil {
log.Error(err)
} else {
log.Debugf("handle ddl job id(%d): %s", s.jobs[i].ID, string(data))
}

_, _, _, err = s.handleDDL(s.jobs[i])
if err != nil {
return errors.Annotatef(err, "handle ddl job %v failed, the schema info: %s", s.jobs[i], s)
}
data, err := json.Marshal(s.jobs[i])
if err != nil {
log.Error(err)
} else {
break
log.Debugf("handle ddl job id(%d): %s", s.jobs[i].ID, string(data))
}

if job.SchemaState == model.StateDeleteOnly && job.Type == model.ActionDropColumn {
s.tblsDroppingCol[job.TableID] = true
log.Infof("Got DeleteOnly Job [job: %+v]", job)
continue
}
_, _, _, err = s.handleDDL(s.jobs[i])
if err != nil {
return errors.Annotatef(err, "handle ddl job %v failed, the schema info: %s", s.jobs[i], s)
}
}

Expand All @@ -256,9 +260,12 @@ func (s *Schema) handlePreviousDDLJobIfNeed(version int64) error {
// the fourth value[error]: the handleDDL execution's err
func (s *Schema) handleDDL(job *model.Job) (schemaName string, tableName string, sql string, err error) {
if skipJob(job) {
log.Infof("Skip job: %+v", job)
return "", "", "", nil
}

log.Debugf("Handle job: %+v", job)

sql = job.Query
if sql == "" {
return "", "", "", errors.Errorf("[ddl job sql miss]%+v", job)
Expand Down Expand Up @@ -415,11 +422,21 @@ func (s *Schema) handleDDL(job *model.Job) (schemaName string, tableName string,
s.currentVersion = job.BinlogInfo.SchemaVersion
schemaName = schema.Name.O
tableName = tbInfo.Name.O

if job.Type == model.ActionDropColumn {
log.Infof("Finished dropping column [job: %+v]", job)
delete(s.tblsDroppingCol, job.TableID)
}
}

return
}

// IsDroppingColumn returns true if the table is in the middle of dropping a column
func (s *Schema) IsDroppingColumn(id int64) bool {
return s.tblsDroppingCol[id]
}

// IsTruncateTableID returns true if the table id have been truncated by truncate table DDL
func (s *Schema) IsTruncateTableID(id int64) bool {
_, ok := s.truncateTableID[id]
Expand Down
2 changes: 1 addition & 1 deletion drainer/schema_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,7 @@ func (t *schemaSuite) TestSchema(c *C) {
jobs = append(jobs, job)

// construct a rollbackdone job
jobs = append(jobs, &model.Job{ID: 5, State: model.JobStateRollbackDone})
jobs = append(jobs, &model.Job{ID: 5, State: model.JobStateRollbackDone, BinlogInfo: &model.HistoryInfo{}})

// reconstruct the local schema
schema, err := NewSchema(jobs, false)
Expand Down
9 changes: 8 additions & 1 deletion drainer/syncer.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -502,6 +502,11 @@ func (s *Syncer) run(jobs []*model.Job) error {
return errors.Trace(err)
}

if b.job.SchemaState == model.StateDeleteOnly && b.job.Type == model.ActionDropColumn {
log.Infof("Syncer skips DeleteOnly DDL [job: %+v] [ts: %d]", b.job, b.GetCommitTs())
continue
}

log.Debug("ddl query: ", b.job.Query)
sql := b.job.Query
schema, table, err := s.schema.getSchemaTableAndDelete(b.job.BinlogInfo.SchemaVersion)
Expand Down Expand Up @@ -552,6 +557,8 @@ func (s *Syncer) translateSqls(mutations []pb.TableMutation, commitTS int64, nod
continue
}

isTblDroppingCol := s.schema.IsDroppingColumn(mutation.GetTableId())

var (
safeMode bool

Expand Down Expand Up @@ -580,7 +587,7 @@ func (s *Syncer) translateSqls(mutations []pb.TableMutation, commitTS int64, nod
}

if len(mutation.GetUpdatedRows()) > 0 {
sqls[pb.MutationType_Update], keys[pb.MutationType_Update], args[pb.MutationType_Update], safeMode, err = s.translator.GenUpdateSQLs(schemaName, table, mutation.GetUpdatedRows(), commitTS)
sqls[pb.MutationType_Update], keys[pb.MutationType_Update], args[pb.MutationType_Update], safeMode, err = s.translator.GenUpdateSQLs(schemaName, table, mutation.GetUpdatedRows(), commitTS, isTblDroppingCol)
if err != nil {
return errors.Errorf("gen update sqls failed: %v, schema: %s, table: %s", err, schemaName, tableName)
}
Expand Down
6 changes: 3 additions & 3 deletions drainer/translator/flash.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -96,7 +96,7 @@ func (f *flashTranslator) GenInsertSQLs(schema string, table *model.TableInfo, r
return sqls, keys, values, nil
}

func (f *flashTranslator) GenUpdateSQLs(schema string, table *model.TableInfo, rows [][]byte, commitTS int64) ([]string, [][]string, [][]interface{}, bool, error) {
func (f *flashTranslator) GenUpdateSQLs(schema string, table *model.TableInfo, rows [][]byte, commitTS int64, isTblDroppingCol bool) ([]string, [][]string, [][]interface{}, bool, error) {
schema = strings.ToLower(schema)
pkColumn := pkHandleColumn(table)
if pkColumn == nil {
Expand All @@ -106,7 +106,7 @@ func (f *flashTranslator) GenUpdateSQLs(schema string, table *model.TableInfo, r
sqls := make([]string, 0, len(rows))
keys := make([][]string, 0, len(rows))
totalValues := make([][]interface{}, 0, len(rows))
colsTypeMap := util.ToColumnTypeMap(table.Columns)
cols := util.ToColumnMap(table.Columns)
version := makeInternalVersionValue(uint64(commitTS))
delFlag := makeInternalDelmarkValue(false)

Expand All @@ -115,7 +115,7 @@ func (f *flashTranslator) GenUpdateSQLs(schema string, table *model.TableInfo, r
var newValues []interface{}

// TODO: Make updating pk working
oldColumnValues, newColumnValues, err := decodeFlashOldAndNewRow(row, colsTypeMap, gotime.Local)
oldColumnValues, newColumnValues, err := DecodeOldAndNewRow(row, cols, gotime.Local, isTblDroppingCol)
newPkValue := newColumnValues[pkID]

if err != nil {
Expand Down
Loading