drainer: bugfix, handle "missing column" when a column is getting dropped

drainer/collector.go

@@ -283,9 +283,13 @@ func (c *Collector) syncBinlog(item *binlogItem) error {
 
  log.Info("get ddl job", zap.Stringer("job", job))
 
- if skipJob(job) {
+ isDelOnlyEvent := model.SchemaState(binlog.DdlSchemaState) == model.StateDeleteOnly
+ if skipJob(job) && !isDelOnlyEvent {
  return nil
  }
+ if isDelOnlyEvent {
+ job.SchemaState = model.StateDeleteOnly
+ }
  item.SetJob(job)
  ddlJobsCounter.Add(float64(1))
  }

drainer/schema.go

@@ -37,6 +37,7 @@ type Schema struct {
  tables map[int64]*model.TableInfo
 
  truncateTableID map[int64]struct{}
+ tblsDroppingCol map[int64]bool
 
  schemaMetaVersion int64
 
@@ -56,6 +57,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,
  }
 
@@ -231,25 +233,27 @@ 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.Debug("skip ddl job", zap.Stringer("job", s.jobs[i]))
+ job := s.jobs[i]
+
+ if job.BinlogInfo.SchemaVersion > version {
+ break
+ }
+
+ if job.BinlogInfo.SchemaVersion <= s.currentVersion {
+ log.Warn("ddl job schema version is less than current version, skip this ddl job",
+ zap.Stringer("job", job),
+ zap.Int64("currentVersion", s.currentVersion))
  continue
  }
 
- if s.jobs[i].BinlogInfo.SchemaVersion <= version {
- if s.jobs[i].BinlogInfo.SchemaVersion <= s.currentVersion {
- log.Warn("ddl job schema version is less than current version, skip this ddl job",
- zap.Stringer("job", s.jobs[i]),
- zap.Int64("currentVersion", s.currentVersion))
- 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)
- }
- } else {
- break
+ if job.SchemaState == model.StateDeleteOnly && job.Type == model.ActionDropColumn {
+ s.tblsDroppingCol[job.TableID] = true
+ log.Info("Got DeleteOnly Job", zap.Stringer("job", job))
+ continue
+ }
+ _, _, _, err := s.handleDDL(job)
+ if err != nil {
+ return errors.Annotatef(err, "handle ddl job %v failed, the schema info: %s", s.jobs[i], s)
  }
  }
 
@@ -264,12 +268,13 @@ func (s *Schema) handlePreviousDDLJobIfNeed(version int64) error {
 // the third value[string]: the sql that is corresponding to the job
 // the fourth value[error]: the handleDDL execution's err
 func (s *Schema) handleDDL(job *model.Job) (schemaName string, tableName string, sql string, err error) {
- log.Debug("handle job: ", zap.Stringer("job", job))
-
  if skipJob(job) {
+ log.Debug("Skip job", zap.Stringer("job", job))
  return "", "", "", nil
  }
 
+ log.Debug("Handle job", zap.Stringer("job", job))
+
  sql = job.Query
  if sql == "" {
  return "", "", "", errors.Errorf("[ddl job sql miss]%+v", job)
@@ -426,11 +431,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.Info("Finished dropping column", zap.Stringer("job", 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]

drainer/schema_test.go

@@ -56,7 +56,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)

drainer/syncer.go

@@ -362,7 +362,7 @@ ForLoop:
  lastAddComitTS = binlog.GetCommitTs()
  err = s.dsyncer.Sync(&dsync.Item{Binlog: binlog, PrewriteValue: preWrite})
  if err != nil {
- err = errors.Annotate(err, "add to dsyncer failed")
+ err = errors.Annotatef(err, "add to dsyncer, commit ts %d", binlog.CommitTs)
  break ForLoop
  }
  executeHistogram.Observe(time.Since(beginTime).Seconds())
@@ -383,6 +383,11 @@ ForLoop:
  break ForLoop
  }
 
+ if b.job.SchemaState == model.StateDeleteOnly && b.job.Type == model.ActionDropColumn {
+ log.Info("Syncer skips DeleteOnly DDL", zap.Stringer("job", b.job), zap.Int64("ts", b.GetCommitTs()))
+ continue
+ }
+
  sql := b.job.Query
  var schema, table string
  schema, table, err = s.schema.getSchemaTableAndDelete(b.job.BinlogInfo.SchemaVersion)
@@ -404,7 +409,7 @@ ForLoop:
 
  err = s.dsyncer.Sync(&dsync.Item{Binlog: binlog, PrewriteValue: nil, Schema: schema, Table: table})
  if err != nil {
- err = errors.Annotate(err, "add to dsyncer failed")
+ err = errors.Annotatef(err, "add to dsyncer, commit ts %d", binlog.CommitTs)
  break ForLoop
  }
  executeHistogram.Observe(time.Since(beginTime).Seconds())

drainer/translator/flash.go

@@ -42,6 +42,7 @@ func GenFlashSQLs(infoGetter TableInfoGetter, pv *tipb.PrewriteValue, commitTS i
  if !ok {
  return nil, nil, errors.Errorf("TableByID empty table id: %d", mut.GetTableId())
  }
+ isTblDroppingCol := infoGetter.IsDroppingColumn(mut.GetTableId())
 
  var schema string
  schema, _, ok = infoGetter.SchemaAndTableName(mut.GetTableId())
@@ -68,7 +69,7 @@ func GenFlashSQLs(infoGetter TableInfoGetter, pv *tipb.PrewriteValue, commitTS i
  return nil, nil, errors.Annotate(err, "gen insert sql fail")
  }
  case tipb.MutationType_Update:
- sql, args, err = GenFlashUpdateSQL(schema, info, row, commitTS)
+ sql, args, err = GenFlashUpdateSQL(schema, info, row, commitTS, isTblDroppingCol)
  if err != nil {
  return nil, nil, errors.Annotate(err, "gen update sql fail")
  }
@@ -140,15 +141,15 @@ func GenFlashInsertSQL(schema string, table *model.TableInfo, row []byte, commit
 }
 
 // GenFlashUpdateSQL generate the SQL need to execute syncing this update row to Flash
-func GenFlashUpdateSQL(schema string, table *model.TableInfo, row []byte, commitTS int64) (sql string, args []interface{}, err error) {
+func GenFlashUpdateSQL(schema string, table *model.TableInfo, row []byte, commitTS int64, isTblDroppingCol bool) (sql string, args []interface{}, err error) {
  schema = strings.ToLower(schema)
  pkColumn := pkHandleColumn(table)
  if pkColumn == nil {
  pkColumn = fakeImplicitColumn(table)
  }
  pkID := pkColumn.ID
 
- updtDecoder := newUpdateDecoder(table)
+ updtDecoder := newUpdateDecoder(table, isTblDroppingCol)
  version := makeInternalVersionValue(uint64(commitTS))
  delFlag := makeInternalDelmarkValue(false)