Merge pull request pingcap#5 from ti2sky/add-replay

fix some bug

replay_file

@@ -1,8 +1,2 @@
-5 3 test begin
-3 4 test begin
-5 5 test insert into t values(5)
-3 6 test insert into t values(6)
-5 7 test commit
-5 8 test select * from t
-3 8 test commit
-5 8 test commit
+3 1 test use `test`
+3 2 test select * from t where a BETWEEN ? AND ?  [arguments: (1, 55)]

session/session.go

@@ -2866,8 +2866,7 @@ func logGeneralQuery(execStmt *executor.ExecStmt, s *session, isPrepared bool) {
 	cfg := config.GetGlobalConfig()
 	vars := s.GetSessionVars()
 	if !s.isInternal() && cfg.EnableReplaySQL.Load() {
-		fmt.Println("print sql")
-		go func(sql, id string) {
+		go func(sql string) {
 			//TODO: We need to add a client col also.
 			var builder strings.Builder
 			builder.WriteString(fmt.Sprintf("%v", vars.ConnectionID))
@@ -2876,11 +2875,16 @@ func logGeneralQuery(execStmt *executor.ExecStmt, s *session, isPrepared bool) {
 			ts := strconv.FormatInt(s.sessionVars.StartTime.Unix()-cfg.ReplayMetaTS, 10)
 			builder.WriteString(ts)
 			builder.WriteString(" ")
-			text := strings.ReplaceAll(sql, "\n", " ")
+			builder.WriteString(vars.CurrentDB)
+			text := executor.QueryReplacer.Replace(sql)
 			builder.WriteString(text)
+			if vars.PreparedParams != nil {
+				builder.WriteString(" ")
+				builder.WriteString(vars.PreparedParams.String())
+			}
 			builder.WriteString("\n")
 			logutil.PutRecordOrDrop(builder.String())
-		}(vars.SQL, vars.TxnCtx.ID.String())
+		}(vars.SQL)
 	}
 
 	if variable.ProcessGeneralLog.Load() && !vars.InRestrictedSQL {

util/replayutil/record_replayer.go

@@ -9,14 +9,14 @@ import (
 	"strings"
 	"time"
 
-	"github.com/pingcap/errors"
 	"github.com/pingcap/log"
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/session"
 )
 
 // RecordReplayer is used to replay sql
 var RecordReplayer *recordReplayer
+
 // Sessions is a map
 var Sessions map[string]session.Session
 
@@ -72,7 +72,6 @@ func (r *recordReplayer) start() {
 		}
 		ts, _ := strconv.ParseFloat(record[1], 10)
 		if sleepTime := ts - time.Since(start).Seconds(); sleepTime > 0 {
-			fmt.Printf("sleep time:%v\n", sleepTime)
 			time.Sleep(time.Duration(sleepTime) * time.Second)
 		}
 		if s, exist := Sessions[record[0]]; !exist {
@@ -83,24 +82,43 @@ func (r *recordReplayer) start() {
 				return
 			}
 			Sessions[record[0]] = se
-			go replayExecuteSQL(record[3], se, record[0])
+			go replayExecuteSQL(record[3], se)
 		} else {
-			go replayExecuteSQL(record[3], s, record[0])
+			go replayExecuteSQL(record[3], s)
 		}
 	}
 }
 
-func replayExecuteSQL(sql string, s session.Session, connection string) error {
+func replayExecuteSQL(sql string, s session.Session) error {
 	ctx := context.Background()
+	fmt.Println(sql)
+	args := strings.Split(sql, "[arguments: (")
+	fmt.Println(args)
+	if len(args) > 1{
+		argument := strings.Split(args[1][:len(args[1])-2], ", ")
+		sql = helper(args[0], argument)
+	}
+	fmt.Println(sql)
 	stmts, err := s.Parse(ctx, sql)
 	if err != nil {
 		return err
 	}
 	for _, stmt := range stmts {
-		_, err := s.ExecuteStmt(ctx, stmt)
-		if err != nil {
-			return errors.Trace(err)
-		}
+		s.ExecuteStmt(ctx, stmt)
 	}
 	return nil
 }
+
+func helper(sql string, args []string) string {
+	newsql := ""
+	i := 0
+	for _, b := range []byte(sql) {
+		if b == byte('?'){
+			newsql += args[i]
+			i++
+		}else{
+			newsql += string(b)
+		}
+	}
+	return newsql
+}