Skip the tiflash job. (#941)

* Skip the tiflash job.

test:
1. create table t(id int);
2. ALTER TABLE t SET TIFLASH REPLICA 3 LOCATION LABELS "rack", "host", "abc";            // model.ActionSetTiFlashReplica
3. curl -X POST -d '{"id":45,"region_count":3,"flash_region_count":3}' https://127.0.0.1:10080/tiflash/replica  // model.ActionUpdateTiFlashReplicaStatus
4. performance write on table t and check can still replicate.

ref: pingcap/tidb#12453

* Specify status port for tidb

* Try repl SetReplicate

* check len

drainer/schema.go

@@ -262,6 +262,17 @@ func (s *Schema) handlePreviousDDLJobIfNeed(version int64) error {
 	return nil
 }
 
+func skipFlash(job *model.Job) bool {
+	switch job.Type {
+	case model.ActionUpdateTiFlashReplicaStatus: // empty job.Query
+		return true
+		// case model.ActionSetTiFlashReplica:
+		// 	return true
+	}
+
+	return false
+}
+
 // handleDDL has four return values,
 // the first value[string]: the schema name
 // the second value[string]: the table name

drainer/syncer.go

@@ -388,6 +388,10 @@ ForLoop:
 		} else if jobID > 0 {
 			log.Debug("get ddl binlog job", zap.Stringer("job", b.job))
 
+			if skipFlash(b.job) {
+				continue
+			}
+
 			// Notice: the version of DDL Binlog we receive are Monotonically increasing
 			// DDL (with version 10, commit ts 100) -> DDL (with version 9, commit ts 101) would never happen
 			s.schema.addJob(b.job)

pkg/loader/load.go

@@ -413,6 +413,10 @@ func (s *loaderImpl) execDDL(ddl *DDL) error {
 		return nil
 	})
 
+	if err != nil && isSetTiFlashReplica(ddl.SQL) {
+		return nil
+	}
+
 	return errors.Trace(err)
 }
 
@@ -861,3 +865,28 @@ func getAppliedTS(db *gosql.DB) int64 {
 	}
 	return appliedTS
 }
+
+func isSetTiFlashReplica(sql string) bool {
+	stmt, err := parser.New().ParseOneStmt(sql, "", "")
+	if err != nil {
+		log.Error("failed to parse", zap.Error(err), zap.String("sql", sql))
+		return false
+	}
+
+	n, ok := stmt.(*ast.AlterTableStmt)
+	if !ok {
+		return false
+	}
+
+	if len(n.Specs) > 1 {
+		return false
+	}
+
+	for _, spec := range n.Specs {
+		if spec.Tp == ast.AlterTableSetTiFlashReplica {
+			return true
+		}
+	}
+
+	return false
+}

pkg/loader/load_test.go

@@ -39,6 +39,16 @@ func (cs *LoadSuite) SetUpTest(c *check.C) {
 func (cs *LoadSuite) TearDownTest(c *check.C) {
 }
 
+func (cs *LoadSuite) TestTiFlash(c *check.C) {
+	sql := "ALTER TABLE t SET TIFLASH REPLICA 3 LOCATION LABELS \"rack\", \"host\", \"abc\""
+	res := isSetTiFlashReplica(sql)
+	c.Assert(res, check.IsTrue)
+
+	sql = "create table a(id int)"
+	res = isSetTiFlashReplica(sql)
+	c.Assert(res, check.IsFalse)
+}
+
 func (cs *LoadSuite) TestRemoveOrphanCols(c *check.C) {
 	dml := &DML{
 		Values: map[string]interface{}{

tests/run.sh

@@ -74,9 +74,11 @@ allow-auto-random = true
 EOF
 
     port=${1-4000}
+    sport=${2-10080}
     echo "Starting TiDB at port: $port..."
     tidb-server \
         -P $port \
+        -status $sport \
         -config "$OUT_DIR/tidb-config.toml" \
         --store tikv \
         --path 127.0.0.1:2379 \
@@ -194,8 +196,8 @@ EOF
 
     sleep 5
 
-    start_upstream_tidb 4000
-    start_upstream_tidb 4001
+    start_upstream_tidb 4000 10080
+    start_upstream_tidb 4001 10081
 
 
     cat - > "$OUT_DIR/down-tidb-config.toml" <<EOF