修复指定开始时间从binlog最开始解析bug

Author: daiguadaidai

README.md

@@ -2,6 +2,21 @@
 
 史上最变态的MySQL DML 闪回工具
 
+* [MySQL Flashback](#mysql-flashback)
+  * [原理](#%E5%8E%9F%E7%90%86)
+  * [再次捣腾这功能原因](#%E5%86%8D%E6%AC%A1%E6%8D%A3%E8%85%BE%E8%BF%99%E5%8A%9F%E8%83%BD%E5%8E%9F%E5%9B%A0)
+  * [生成二进制](#%E7%94%9F%E6%88%90%E4%BA%8C%E8%BF%9B%E5%88%B6)
+  * [支持的功能](#%E6%94%AF%E6%8C%81%E7%9A%84%E5%8A%9F%E8%83%BD)
+    * [大家有我也有](#%E5%A4%A7%E5%AE%B6%E6%9C%89%E6%88%91%E4%B9%9F%E6%9C%89)
+    * [我的亮点](#%E6%88%91%E7%9A%84%E4%BA%AE%E7%82%B9)
+  * [生成回滚语句](#%E7%94%9F%E6%88%90%E5%9B%9E%E6%BB%9A%E8%AF%AD%E5%8F%A5)
+    * [溜溜的玩法](#%E6%BA%9C%E6%BA%9C%E7%9A%84%E7%8E%A9%E6%B3%95)
+    * [你可以指定多个 SQL](#%E4%BD%A0%E5%8F%AF%E4%BB%A5%E6%8C%87%E5%AE%9A%E5%A4%9A%E4%B8%AA-sql)
+    * [老旧玩法](#%E8%80%81%E6%97%A7%E7%8E%A9%E6%B3%95)
+  * [执行回滚SQL](#%E6%89%A7%E8%A1%8C%E5%9B%9E%E6%BB%9Asql)
+    * [回滚原理和注意事项](#%E5%9B%9E%E6%BB%9A%E5%8E%9F%E7%90%86%E5%92%8C%E6%B3%A8%E6%84%8F%E4%BA%8B%E9%A1%B9)
+    * [执行回滚就简单了](#%E6%89%A7%E8%A1%8C%E5%9B%9E%E6%BB%9A%E5%B0%B1%E7%AE%80%E5%8D%95%E4%BA%86)
+
 ## 原理
 
 在 `MySQL Binlog` 已经被玩烂掉的时代, 我想大家也都知道其中的原理. 无非就是解析 `Binlog` 反向生成需要回滚的 `SQL` 语句.
@@ -212,7 +227,7 @@ Flags:
 
 1. 由于生成回滚语句是顺序记入到文件中去的, 所以我们回滚的时候需要倒序读取文件 `SQL` 进行回滚.
 
-2. 在看回滚语句的时候你会发现, 每个`SQL`前面都有注释`/* crc32:xxx */`, 这里面的值`xxx`记录的是没条数据主键的一个`crc32`值. 主要是为了并发而记录的.
+2. 在看回滚语句的时候你会发现, 每个`SQL`前面都有注释`/* crc32:xxx */`, 这里面的值`xxx`记录的是每条数据主键的一个`crc32`值. 主要是为了并发而记录的.
 
 ```
 /* crc32:2313941001 */ INSERT INTO `employees`.`emp1`(`emp_no`, `birth_date`, `first_name`, `last_name`, `gender`, `hire_date`) VALUES(10008, "1958-02-19", "Saniya", "Kalloufi", 1, "1994-09-15");
@@ -252,4 +267,4 @@ Flags:
       --filepath string          指定执行的文件
   -h, --help                     help for execute
       --paraller int             回滚并发数 (default 1)
-```
+```

services/create/creator.go

@@ -18,6 +18,12 @@ import (
 	"time"
 )
 
+const (
+	START_POS_BY_NONE uint8 = iota
+	START_POS_BY_POS
+	START_POS_BY_TIME
+)
+
 type Creator struct {
 	CC               *config.CreateConfig
 	DBC              *config.DBConfig
@@ -26,8 +32,11 @@ type Creator struct {
 	StartPosition    *models.Position
 	EndPosition      *models.Position
 	CurrentPosition  *models.Position
+	GetStartPosType  uint8
 	CurrentTimestamp uint32
 	CurrentThreadID  uint32
+	StartTime        time.Time
+	StartTimestamp   uint32
 	HaveEndPosition  bool
 	EndTime          time.Time
 	HaveEndTime      bool
@@ -62,6 +71,23 @@ func NewFlashback(sc *config.CreateConfig, dbc *config.DBConfig, mTables []*visi
 	if err != nil {
 		return nil, err
 	}
+
+	// 获取获取位点的类型, 是通过 位点获取, 还是通过时间
+	if ct.CC.HaveStartPosInfo() { // 通过指定位点获取开始位点
+		ct.GetStartPosType = START_POS_BY_POS
+		seelog.Infof("解析binglog开始位点通过指定 开始binlog 位点获取. 开始位点: %s", ct.StartPosition.String())
+	} else if ct.CC.HaveStartTime() { // 通过时间获取开始位点
+		ct.GetStartPosType = START_POS_BY_TIME
+		seelog.Infof("解析binglog开始位点通过指定 开始时间 获取. 开始位点: %s", ct.StartPosition.String())
+		ct.StartTime, err = utils.NewTime(ct.CC.StartTime)
+		if err != nil {
+			return nil, fmt.Errorf("输入的开始时间有问题. %v", err)
+		}
+		ct.StartTimestamp = uint32(ct.StartTime.Unix())
+	} else {
+		return nil, fmt.Errorf("无法获取")
+	}
+
 	// 原sql文件
 	fileName := ct.getSqlFileName("origin_sql")
 	ct.OriSQLFile = fmt.Sprintf("%s/%s", ct.CC.GetSaveDir(), fileName)
@@ -202,6 +228,13 @@ func (this *Creator) runProduceEvent(wg *sync.WaitGroup) {
 	defer wg.Done()
 	defer this.Syncer.Close()
 
+	// 判断是否需要跳过, 位点
+	var isSkip bool
+	if this.GetStartPosType == START_POS_BY_TIME { // 如果开始位点是通过时间获取的就需要执行跳过
+		isSkip = true
+	}
+	seelog.Debugf("是否是需要跳过事件: %v", isSkip)
+
 	pos := mysql.Position{this.StartPosition.File, this.StartPosition.Position}
 	streamer, err := this.Syncer.StartSync(pos)
 	if err != nil {
@@ -222,6 +255,19 @@ produceLoop:
 				seelog.Error(err.Error())
 				this.quit()
 			}
+
+			// 过去掉还没到开始时间的事件
+			if isSkip {
+				// 判断是否到了开始时间
+				if ev.Header.Timestamp < this.StartTimestamp {
+					continue
+				} else {
+					isSkip = false
+					seelog.Infof("停止跳过, 开始生成回滚sql. 时间戳: %d, 时间: %s, 位点: %s:%d", ev.Header.Timestamp,
+						utils.TS2String(int64(ev.Header.Timestamp), utils.TIME_FORMAT), this.StartPosition.File, ev.Header.LogPos)
+				}
+			}
+
 			if err = this.handleEvent(ev); err != nil {
 				seelog.Error(err.Error())
 				this.quit()

services/create/helper.go

@@ -31,7 +31,7 @@ func GetStartPosition(cc *config.CreateConfig, dbc *config.DBConfig) (*models.Po
 		if err != nil {
 			return nil, err
 		}
-		return getPositionByTime(uint32(ts), dbc)
+		return getStartFileByTime(uint32(ts), dbc)
 	}
 
 	return nil, nil
@@ -67,7 +67,7 @@ func checkStartPosInRange(startPos *models.Position) error {
 }
 
 // 通过开始时间获取位点信息
-func getPositionByTime(ts uint32, dbc *config.DBConfig) (*models.Position, error) {
+func getStartFileByTime(ts uint32, dbc *config.DBConfig) (*models.Position, error) {
 	// 如果指定时间大于但前 返回错误.
 	nts := utils.NowTimestamp() // 但前时间戳
 	if int64(ts) > nts {