fix comment

drainer/syncer.go

@@ -469,13 +469,15 @@ ForLoop:
 
 			if b.job.BinlogInfo.SchemaVersion == 0 {
 				log.Info("skip ddl due to the failed ddl", zap.String("sql", sql), zap.Int64("commit ts", commitTS))
-				shouldSkip = true
-			} else if s.cfg.SyncDDL {
-				schema, table, err = s.schema.getSchemaTableAndDelete(b.job.BinlogInfo.SchemaVersion)
-				if err != nil {
-					err = errors.Trace(err)
-					break ForLoop
-				}
+				appendFakeBinlogIfNeeded(nil, commitTS)
+				continue
+			}
+			schema, table, err = s.schema.getSchemaTableAndDelete(b.job.BinlogInfo.SchemaVersion)
+			if err != nil {
+				err = errors.Trace(err)
+				break ForLoop
+			}
+			if s.cfg.SyncDDL {
 				if s.filter.SkipSchemaAndTable(schema, table) {
 					log.Info("skip ddl by block allow filter", zap.String("schema", schema), zap.String("table", table),
 						zap.String("sql", sql), zap.Int64("commit ts", commitTS))

drainer/syncer_test.go

@@ -168,6 +168,34 @@ func (s *syncerSuite) TestNewSyncer(c *check.C) {
 		job:    job,
 	})
 
+	// Add failed ddl
+	commitTS++
+	jobID++
+	binlog = &pb.Binlog{
+		Tp:       pb.BinlogType_Commit,
+		CommitTs: commitTS,
+		DdlQuery: []byte("alter table test.test add column a int"),
+		DdlJobId: jobID,
+	}
+	job = &model.Job{
+		ID:       jobID,
+		SchemaID: 1, // must be the previous create schema id of `test`
+		Type:     model.ActionAddColumn,
+		State:    model.JobStateSynced,
+		Query:    "create table test.test(id int)",
+		BinlogInfo: &model.HistoryInfo{
+			SchemaVersion: 0,
+			TableInfo: &model.TableInfo{
+				ID:   testTableID,
+				Name: model.CIStr{O: "test", L: "test"},
+			},
+		},
+	}
+	syncer.Add(&binlogItem{
+		binlog: binlog,
+		job:    job,
+	})
+
 	// Add dml
 	commitTS++
 	binlog = &pb.Binlog{