Handle create/drop sequence job type (#950)

1. handle create/drop sequence job type
2. ref pingcap/tidb#14954
TiDB will write a fake ddl binlog like: select setval(`seq`.`sequence_name`, 1000)
when the value of sequence is changed for replicate the value of the sequence.
we CAN NOT query the job from the tikv according the job id.
just skip this kind of binlog now.

drainer/collector.go

@@ -14,6 +14,7 @@
 package drainer
 
 import (
+	"bytes"
 	"crypto/tls"
 	"fmt"
 	"net/http"
@@ -34,6 +35,7 @@ import (
 	"github.com/pingcap/tidb/store"
 	"github.com/pingcap/tidb/store/tikv"
 	"github.com/pingcap/tidb/store/tikv/oracle"
+	"github.com/pingcap/tipb/go-binlog"
 	"go.uber.org/zap"
 	"golang.org/x/net/context"
 )
@@ -260,9 +262,25 @@ func (c *Collector) reportErr(ctx context.Context, err error) {
 	}
 }
 
+// ref https://github.com/pingcap/tidb/pull/14954
+// TiDB will write a fake ddl binlog like: select setval(`seq`.`sequence_name`, 1000)
+// when the value of sequence is changed for replicate the value of the sequence.
+// we CAN NOT query the job from the tikv according the job id.
+// just skip this kind of binlog now.
+func skipQueryJob(binlog *binlog.Binlog) bool {
+	q := binlog.GetDdlQuery()
+	return bytes.HasPrefix(q, []byte("select setval"))
+}
+
 func (c *Collector) syncBinlog(item *binlogItem) error {
 	binlog := item.binlog
+	// DO NOT replicate the value of sequence now.
+	if skipQueryJob(binlog) {
+		return nil
+	}
+
 	if binlog.DdlJobId > 0 {
+		log.Info("start query job", zap.Int64("id", binlog.DdlJobId), zap.Stringer("binlog", binlog))
 		msgPrefix := fmt.Sprintf("get ddl job by id %d error", binlog.DdlJobId)
 		var job *model.Job
 		for {

drainer/schema.go

@@ -359,7 +359,7 @@ func (s *Schema) handleDDL(job *model.Job) (schemaName string, tableName string,
 		schemaName = schema.Name.O
 		tableName = table.Name.O
 
-	case model.ActionCreateTable, model.ActionCreateView, model.ActionRecoverTable:
+	case model.ActionCreateTable, model.ActionCreateView, model.ActionCreateSequence, model.ActionRecoverTable:
 		table := job.BinlogInfo.TableInfo
 		if table == nil {
 			return "", "", "", errors.NotFoundf("table %d", job.TableID)
@@ -380,7 +380,7 @@ func (s *Schema) handleDDL(job *model.Job) (schemaName string, tableName string,
 		schemaName = schema.Name.O
 		tableName = table.Name.O
 
-	case model.ActionDropTable, model.ActionDropView:
+	case model.ActionDropTable, model.ActionDropView, model.ActionDropSequence:
 		schema, ok := s.SchemaByID(job.SchemaID)
 		if !ok {
 			return "", "", "", errors.NotFoundf("schema %d", job.SchemaID)
@@ -422,7 +422,6 @@ func (s *Schema) handleDDL(job *model.Job) (schemaName string, tableName string,
 		schemaName = schema.Name.O
 		tableName = table.Name.O
 		s.truncateTableID[job.TableID] = struct{}{}
-
 	default:
 		binlogInfo := job.BinlogInfo
 		if binlogInfo == nil {

tests/sequence/drainer.toml

@@ -0,0 +1,17 @@
+data-dir = '/tmp/tidb_binlog_test/data.drainer'
+
+[syncer]
+txn-batch = 1
+worker-count = 1
+safe-mode = false
+db-type = 'mysql'
+replicate-do-db = ['seq']
+
+[syncer.to]
+host = '127.0.0.1'
+user = 'root'
+password = ''
+port = 3306
+
+[syncer.to.checkpoint]
+schema = "seq_checkpoint"

tests/sequence/run.sh

@@ -0,0 +1,29 @@
+#!/bin/sh
+
+set -e
+
+cd "$(dirname "$0")"
+
+run_drainer &
+
+sleep 3
+
+run_sql 'CREATE DATABASE seq;'
+run_sql 'CREATE SEQUENCE seq.sequence_name;'
+run_sql 'CREATE TABLE seq.table_name (id INT DEFAULT NEXT VALUE FOR seq.sequence_name, val int);'
+
+run_sql 'INSERT INTO seq.table_name(val) values(10);'
+
+sleep 3
+
+down_run_sql 'SELECT count(*), sum(id), sum(val) FROM seq.table_name;'
+check_contains 'count(*): 1'
+check_contains 'sum(id): 1'
+check_contains 'sum(val): 10'
+
+
+run_sql 'DROP TABLE seq.table_name;'
+run_sql 'DROP SEQUENCE seq.sequence_name;'
+
+
+killall drainer