fix sync-ddl control (#1130)

ref #1129
pingcap · Jan 19, 2022 · 4246eae · 4246eae
1 parent 194d4ac
commit 4246eae
Showing 1 changed file with 33 additions and 32 deletions.
diff --git a/drainer/syncer.go b/drainer/syncer.go
@@ -456,49 +456,50 @@ ForLoop:
 				continue
 			}
 
+			var (
+				shouldSkip    bool
+				schema, table string
+			)
 			sql := b.job.Query
-			var schema, table string
 			schema, table, err = s.schema.getSchemaTableAndDelete(b.job.BinlogInfo.SchemaVersion)
 			if err != nil {
 				err = errors.Trace(err)
 				break ForLoop
 			}
-
-			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))
-				appendFakeBinlogIfNeeded(nil, commitTS)
-				continue
-			}
-
-			// shouldSkip is used specially for database dsyncers like tidb/mysql/oracle
-			// although we skip some ddls, but we still need to update table info
-			// ignore means whether we should should this ddl event after binlogFilter
-			var (
-				shouldSkip, ignore bool
-				stmt               ast.StmtNode
-			)
-
-			if stmt, ignore, err = skipDDLEvent(sql, schema, table, p, s.binlogFilter); err != nil {
-				break ForLoop
-			} else if ignore {
-				log.Info("skip ddl by binlog event filter", zap.String("schema", schema), zap.String("table", table),
-					zap.String("sql", sql), zap.Int64("commit ts", commitTS))
-				// A empty sql force it to evict the downstream table info.
-				if s.cfg.DestDBType == "tidb" || s.cfg.DestDBType == "mysql" || s.cfg.DestDBType == "oracle" {
-					shouldSkip = true
-				} else {
+			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))
 					appendFakeBinlogIfNeeded(nil, commitTS)
 					continue
 				}
-			} else if !ignore && s.cfg.DestDBType == "oracle" {
-				if _, ok := stmt.(*ast.TruncateTableStmt); !ok {
-					err = errors.Errorf("unsupported ddl %s, you should skip commit ts %d", sql, commitTS)
+
+				// shouldSkip is used specially for database dsyncers like tidb/mysql/oracle
+				// although we skip some ddls, but we still need to update table info
+				// ignore means whether we should should this ddl event after binlogFilter
+				var (
+					ignore bool
+					stmt   ast.StmtNode
+				)
+				if stmt, ignore, err = skipDDLEvent(sql, schema, table, p, s.binlogFilter); err != nil {
 					break ForLoop
+				} else if ignore {
+					log.Info("skip ddl by binlog event filter", zap.String("schema", schema), zap.String("table", table),
+						zap.String("sql", sql), zap.Int64("commit ts", commitTS))
+					// A empty sql force it to evict the downstream table info.
+					if s.cfg.DestDBType == "tidb" || s.cfg.DestDBType == "mysql" || s.cfg.DestDBType == "oracle" {
+						shouldSkip = true
+					} else {
+						appendFakeBinlogIfNeeded(nil, commitTS)
+						continue
+					}
+				} else if !ignore && s.cfg.DestDBType == "oracle" {
+					if _, ok := stmt.(*ast.TruncateTableStmt); !ok {
+						err = errors.Errorf("unsupported ddl %s, you should skip commit ts %d", sql, commitTS)
+						break ForLoop
+					}
 				}
-			}
-
-			if !s.cfg.SyncDDL {
+			} else {
 				log.Info("skip ddl by SyncDDL setting to false", zap.String("schema", schema), zap.String("table", table),
 					zap.String("sql", sql), zap.Int64("commit ts", commitTS))
 				// A empty sql force it to evict the downstream table info.