executor: add transaction commit runtime informati ... (#19366)

executor/adapter.go

@@ -849,6 +849,16 @@ func (a *ExecStmt) LogSlowQuery(txnTS uint64, succ bool, hasMoreResults bool) {
 		stmtDetail = *(stmtDetailRaw.(*execdetails.StmtExecDetails))
 	}
 	execDetail := sessVars.StmtCtx.GetExecDetails()
+
+	// Attach commit runtime stats to executor runtime stats.
+	if execDetail.CommitDetail != nil && sessVars.StmtCtx.RuntimeStatsColl != nil {
+		stats := sessVars.StmtCtx.RuntimeStatsColl.GetRootStats(a.Plan.ID())
+		statsWithCommit := &execdetails.RuntimeStatsWithCommit{
+			RuntimeStats: stats,
+			Commit:       execDetail.CommitDetail,
+		}
+		sessVars.StmtCtx.RuntimeStatsColl.RegisterStats(a.Plan.ID(), statsWithCommit)
+	}
 	copTaskInfo := sessVars.StmtCtx.CopTasksDetails()
 	statsInfos := plannercore.GetStatsInfo(a.Plan)
 	memMax := sessVars.StmtCtx.MemTracker.MaxConsumed()

util/execdetails/execdetails.go

@@ -14,6 +14,7 @@
 package execdetails
 
 import (
+	"bytes"
 	"fmt"
 	"sort"
 	"strconv"
@@ -491,3 +492,84 @@ func (e *RuntimeStatsWithConcurrencyInfo) String() string {
 	}
 	return result
 }
+
+// RuntimeStatsWithCommit is the RuntimeStats with commit detail.
+type RuntimeStatsWithCommit struct {
+	RuntimeStats
+	Commit *CommitDetails
+}
+
+func (e *RuntimeStatsWithCommit) String() string {
+	var result string
+	if e.RuntimeStats != nil {
+		result = e.RuntimeStats.String()
+	}
+	if e.Commit == nil {
+		return result
+	}
+	buf := bytes.NewBuffer(make([]byte, 0, len(result)+32))
+	buf.WriteString(result)
+	if e.Commit.PrewriteTime > 0 {
+		buf.WriteString(", prewrite:")
+		buf.WriteString(e.Commit.PrewriteTime.String())
+	}
+	if e.Commit.WaitPrewriteBinlogTime > 0 {
+		buf.WriteString(", wait_prewrite_binlog:")
+		buf.WriteString(e.Commit.WaitPrewriteBinlogTime.String())
+	}
+	if e.Commit.GetCommitTsTime > 0 {
+		buf.WriteString(", get_commit_ts:")
+		buf.WriteString(e.Commit.GetCommitTsTime.String())
+	}
+	if e.Commit.CommitTime > 0 {
+		buf.WriteString(", commit:")
+		buf.WriteString(e.Commit.CommitTime.String())
+	}
+	commitBackoffTime := atomic.LoadInt64(&e.Commit.CommitBackoffTime)
+	if commitBackoffTime > 0 {
+		buf.WriteString(", commit_backoff: {time: ")
+		buf.WriteString(time.Duration(commitBackoffTime).String())
+		tpMap := make(map[string]struct{})
+		tpArray := []string{}
+		e.Commit.Mu.Lock()
+		if len(e.Commit.Mu.BackoffTypes) > 0 {
+			for _, tp := range e.Commit.Mu.BackoffTypes {
+				tpStr := tp.String()
+				_, ok := tpMap[tpStr]
+				if ok {
+					continue
+				}
+				tpMap[tpStr] = struct{}{}
+				tpArray = append(tpArray, tpStr)
+			}
+			buf.WriteString(", type: ")
+			sort.Strings(tpArray)
+			buf.WriteString(fmt.Sprintf("%v", tpArray))
+		}
+		e.Commit.Mu.Unlock()
+		buf.WriteString("}")
+	}
+	if e.Commit.ResolveLockTime > 0 {
+		buf.WriteString(", resolve_lock: ")
+		buf.WriteString(time.Duration(e.Commit.ResolveLockTime).String())
+	}
+
+	prewriteRegionNum := atomic.LoadInt32(&e.Commit.PrewriteRegionNum)
+	if prewriteRegionNum > 0 {
+		buf.WriteString(", region_num:")
+		buf.WriteString(strconv.FormatInt(int64(prewriteRegionNum), 10))
+	}
+	if e.Commit.WriteKeys > 0 {
+		buf.WriteString(", write_keys:")
+		buf.WriteString(strconv.FormatInt(int64(e.Commit.WriteKeys), 10))
+	}
+	if e.Commit.WriteSize > 0 {
+		buf.WriteString(", write_byte:")
+		buf.WriteString(strconv.FormatInt(int64(e.Commit.WriteSize), 10))
+	}
+	if e.Commit.TxnRetry > 0 {
+		buf.WriteString(", txn_retry:")
+		buf.WriteString(strconv.FormatInt(int64(e.Commit.TxnRetry), 10))
+	}
+	return buf.String()
+}

util/execdetails/execdetails_test.go

@@ -121,6 +121,7 @@ func TestCopRuntimeStats(t *testing.T) {
 	}
 }
 
+<<<<<<< HEAD
 func TestCopRuntimeStatsForTiFlash(t *testing.T) {
 	stats := NewRuntimeStatsColl()
 	tableScanID := 1
@@ -156,5 +157,44 @@ func TestCopRuntimeStatsForTiFlash(t *testing.T) {
 	}
 	if stats.ExistsRootStats(tableReaderID) == false {
 		t.Fatal("table_reader not exists")
+=======
+func TestRuntimeStatsWithCommit(t *testing.T) {
+	basicStats := &BasicRuntimeStats{
+		loop:    1,
+		consume: int64(time.Second),
+	}
+	commitDetail := &CommitDetails{
+		GetCommitTsTime:   time.Second,
+		PrewriteTime:      time.Second,
+		CommitTime:        time.Second,
+		CommitBackoffTime: int64(time.Second),
+		Mu: struct {
+			sync.Mutex
+			BackoffTypes []fmt.Stringer
+		}{BackoffTypes: []fmt.Stringer{
+			stringutil.MemoizeStr(func() string {
+				return "backoff1"
+			}),
+			stringutil.MemoizeStr(func() string {
+				return "backoff2"
+			}),
+			stringutil.MemoizeStr(func() string {
+				return "backoff1"
+			}),
+		}},
+		ResolveLockTime:   int64(time.Second),
+		WriteKeys:         3,
+		WriteSize:         66,
+		PrewriteRegionNum: 5,
+		TxnRetry:          2,
+	}
+	stats := &RuntimeStatsWithCommit{
+		RuntimeStats: basicStats,
+		Commit:       commitDetail,
+	}
+	expect := "time:1s, loops:1, prewrite:1s, get_commit_ts:1s, commit:1s, commit_backoff: {time: 1s, type: [backoff1 backoff2]}, resolve_lock: 1s, region_num:5, write_keys:3, write_byte:66, txn_retry:2"
+	if stats.String() != expect {
+		t.Fatalf("%v != %v", stats.String(), expect)
+>>>>>>> db3c96b... executor: add transaction commit runtime information in slow log (#19366)
 	}
 }