executor: Add the HashAggExec runtime information

executor/aggregate.go

@@ -17,7 +17,10 @@ import (
 	"bytes"
 	"context"
 	"fmt"
+	"sort"
 	"sync"
+	"sync/atomic"
+	"time"
 
 	"github.com/cznic/mathutil"
 	"github.com/pingcap/errors"
@@ -43,18 +46,22 @@ type aggPartialResultMapper map[string][]aggfuncs.PartialResult
 
 // baseHashAggWorker stores the common attributes of HashAggFinalWorker and HashAggPartialWorker.
 type baseHashAggWorker struct {
-	ctx          sessionctx.Context
-	finishCh     <-chan struct{}
-	aggFuncs     []aggfuncs.AggFunc
-	maxChunkSize int
+	stats          *HashAggRuntimeStats
+	ctx            sessionctx.Context
+	finishCh       <-chan struct{}
+	aggFuncs       []aggfuncs.AggFunc
+	maxChunkSize   int
+	aggWorkerStats time.Duration
 }
 
-func newBaseHashAggWorker(ctx sessionctx.Context, finishCh <-chan struct{}, aggFuncs []aggfuncs.AggFunc, maxChunkSize int) baseHashAggWorker {
+func newBaseHashAggWorker(ctx sessionctx.Context, finishCh <-chan struct{}, aggFuncs []aggfuncs.AggFunc, maxChunkSize int, stat *HashAggRuntimeStats) baseHashAggWorker {
 	return baseHashAggWorker{
-		ctx:          ctx,
-		finishCh:     finishCh,
-		aggFuncs:     aggFuncs,
-		maxChunkSize: maxChunkSize,
+		stats:          stat,
+		ctx:            ctx,
+		finishCh:       finishCh,
+		aggFuncs:       aggFuncs,
+		maxChunkSize:   maxChunkSize,
+		aggWorkerStats: 0,
 	}
 }
 
@@ -171,6 +178,8 @@ type HashAggExec struct {
 	executed         bool
 
 	memTracker *memory.Tracker // track memory usage.
+
+	stats *HashAggRuntimeStats
 }
 
 // HashAggInput indicates the input of hash agg exec.
@@ -234,22 +243,6 @@ func (e *HashAggExec) Close() error {
 	for range e.finalOutputCh {
 	}
 	e.executed = false
-
-	if e.runtimeStats != nil {
-		var partialConcurrency, finalConcurrency int
-		if e.isUnparallelExec {
-			partialConcurrency = 0
-			finalConcurrency = 0
-		} else {
-			partialConcurrency = cap(e.partialWorkers)
-			finalConcurrency = cap(e.finalWorkers)
-		}
-		partialConcurrencyInfo := execdetails.NewConcurrencyInfo("PartialConcurrency", partialConcurrency)
-		finalConcurrencyInfo := execdetails.NewConcurrencyInfo("FinalConcurrency", finalConcurrency)
-		runtimeStats := &execdetails.RuntimeStatsWithConcurrencyInfo{}
-		runtimeStats.SetConcurrencyInfo(partialConcurrencyInfo, finalConcurrencyInfo)
-		e.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl.RegisterStats(e.id, runtimeStats)
-	}
 	return e.baseExecutor.Close()
 }
 
@@ -300,10 +293,12 @@ func (e *HashAggExec) initForParallelExec(ctx sessionctx.Context) {
 	e.partialWorkers = make([]HashAggPartialWorker, partialConcurrency)
 	e.finalWorkers = make([]HashAggFinalWorker, finalConcurrency)
 
+	e.initRuntimeStats()
+
 	// Init partial workers.
 	for i := 0; i < partialConcurrency; i++ {
 		w := HashAggPartialWorker{
-			baseHashAggWorker: newBaseHashAggWorker(e.ctx, e.finishCh, e.PartialAggFuncs, e.maxChunkSize),
+			baseHashAggWorker: newBaseHashAggWorker(e.ctx, e.finishCh, e.PartialAggFuncs, e.maxChunkSize, e.stats),
 			inputCh:           e.partialInputChs[i],
 			outputChs:         e.partialOutputChs,
 			giveBackCh:        e.inputCh,
@@ -328,7 +323,7 @@ func (e *HashAggExec) initForParallelExec(ctx sessionctx.Context) {
 	// Init final workers.
 	for i := 0; i < finalConcurrency; i++ {
 		e.finalWorkers[i] = HashAggFinalWorker{
-			baseHashAggWorker:   newBaseHashAggWorker(e.ctx, e.finishCh, e.FinalAggFuncs, e.maxChunkSize),
+			baseHashAggWorker:   newBaseHashAggWorker(e.ctx, e.finishCh, e.FinalAggFuncs, e.maxChunkSize, e.stats),
 			partialResultMap:    make(aggPartialResultMapper),
 			groupSet:            set.NewStringSet(),
 			inputCh:             e.partialOutputChs[i],
@@ -366,6 +361,7 @@ func recoveryHashAgg(output chan *AfFinalResult, r interface{}) {
 }
 
 func (w *HashAggPartialWorker) run(ctx sessionctx.Context, waitGroup *sync.WaitGroup, finalConcurrency int) {
+	start := time.Now()
 	needShuffle, sc := false, ctx.GetSessionVars().StmtCtx
 	defer func() {
 		if r := recover(); r != nil {
@@ -376,22 +372,34 @@ func (w *HashAggPartialWorker) run(ctx sessionctx.Context, waitGroup *sync.WaitG
 		}
 		w.memTracker.Consume(-w.chk.MemoryUsage())
 		waitGroup.Done()
+		w.aggWorkerStats = time.Since(start)
+		if w.stats != nil {
+			atomic.AddInt64(&w.stats.PartialTotalTime, int64(time.Since(start)))
+		}
 	}()
 	for {
+		waitStart := time.Now()
 		if !w.getChildInput() {
 			return
 		}
+		if w.stats != nil {
+			atomic.AddInt64(&w.stats.PartialWaitTime, int64(time.Since(waitStart)))
+		}
 		if err := w.updatePartialResult(ctx, sc, w.chk, len(w.partialResultsMap)); err != nil {
 			w.globalOutputCh <- &AfFinalResult{err: err}
 			return
 		}
+		if w.stats != nil {
+			atomic.AddInt64(&w.stats.PartialTaskNum, 1)
+		}
 		// The intermData can be promised to be not empty if reaching here,
 		// so we set needShuffle to be true.
 		needShuffle = true
 	}
 }
 
 func (w *HashAggPartialWorker) updatePartialResult(ctx sessionctx.Context, sc *stmtctx.StatementContext, chk *chunk.Chunk, finalConcurrency int) (err error) {
+	start := time.Now()
 	w.groupKey, err = getGroupKey(w.ctx, chk, w.groupKey, w.groupByItems)
 	if err != nil {
 		return err
@@ -408,6 +416,9 @@ func (w *HashAggPartialWorker) updatePartialResult(ctx sessionctx.Context, sc *s
 			}
 		}
 	}
+	if w.stats != nil {
+		atomic.AddInt64(&w.stats.PartialExecTime, int64(time.Since(start)))
+	}
 	return nil
 }
 
@@ -513,6 +524,9 @@ func (w *HashAggFinalWorker) consumeIntermData(sctx sessionctx.Context) (err err
 		if input, ok = w.getPartialInput(); !ok {
 			return nil
 		}
+		if w.stats != nil {
+			atomic.AddInt64(&w.stats.FinalTaskNum, 1)
+		}
 		if intermDataBuffer == nil {
 			intermDataBuffer = make([][]aggfuncs.PartialResult, 0, w.maxChunkSize)
 		}
@@ -580,11 +594,13 @@ func (w *HashAggFinalWorker) receiveFinalResultHolder() (*chunk.Chunk, bool) {
 }
 
 func (w *HashAggFinalWorker) run(ctx sessionctx.Context, waitGroup *sync.WaitGroup) {
+	start := time.Now()
 	defer func() {
 		if r := recover(); r != nil {
 			recoveryHashAgg(w.outputCh, r)
 		}
 		waitGroup.Done()
+		w.aggWorkerStats = time.Since(start)
 	}()
 	if err := w.consumeIntermData(ctx); err != nil {
 		w.outputCh <- &AfFinalResult{err: err}
@@ -660,20 +676,36 @@ func (e *HashAggExec) waitFinalWorkerAndCloseFinalOutput(waitGroup *sync.WaitGro
 
 func (e *HashAggExec) prepare4ParallelExec(ctx context.Context) {
 	go e.fetchChildData(ctx)
-
 	partialWorkerWaitGroup := &sync.WaitGroup{}
 	partialWorkerWaitGroup.Add(len(e.partialWorkers))
+	partialStart := time.Now()
 	for i := range e.partialWorkers {
 		go e.partialWorkers[i].run(e.ctx, partialWorkerWaitGroup, len(e.finalWorkers))
 	}
-	go e.waitPartialWorkerAndCloseOutputChs(partialWorkerWaitGroup)
-
+	go func() {
+		e.waitPartialWorkerAndCloseOutputChs(partialWorkerWaitGroup)
+		if e.stats != nil {
+			e.stats.PartialWallTime = time.Since(partialStart)
+			for i, worker := range e.partialWorkers {
+				e.stats.PartialWorkerTime[i] = worker.aggWorkerStats
+			}
+		}
+	}()
 	finalWorkerWaitGroup := &sync.WaitGroup{}
 	finalWorkerWaitGroup.Add(len(e.finalWorkers))
+	finalStart := time.Now()
 	for i := range e.finalWorkers {
 		go e.finalWorkers[i].run(e.ctx, finalWorkerWaitGroup)
 	}
-	go e.waitFinalWorkerAndCloseFinalOutput(finalWorkerWaitGroup)
+	go func() {
+		e.waitFinalWorkerAndCloseFinalOutput(finalWorkerWaitGroup)
+		if e.stats != nil {
+			e.stats.FinalWallTime = time.Since(finalStart)
+			for i, worker := range e.finalWorkers {
+				e.stats.FinalWorkerTime[i] = worker.aggWorkerStats
+			}
+		}
+	}()
 }
 
 // HashAggExec employs one input reader, M partial workers and N final workers to execute parallelly.
@@ -816,6 +848,114 @@ func (e *HashAggExec) getPartialResults(groupKey string) []aggfuncs.PartialResul
 	return partialResults
 }
 
+func (e *HashAggExec) initRuntimeStats() {
+	if e.runtimeStats != nil && e.stats == nil {
+		e.stats = &HashAggRuntimeStats{
+			PartialNum:        e.ctx.GetSessionVars().HashAggPartialConcurrency(),
+			FinalNum:          e.ctx.GetSessionVars().HashAggFinalConcurrency(),
+			PartialTaskNum:    0,
+			FinalTaskNum:      0,
+			PartialWallTime:   0,
+			FinalWallTime:     0,
+			PartialWorkerTime: make([]time.Duration, e.ctx.GetSessionVars().HashAggPartialConcurrency()),
+			FinalWorkerTime:   make([]time.Duration, e.ctx.GetSessionVars().HashAggFinalConcurrency()),
+		}
+		e.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl.RegisterStats(e.id, e.stats)
+	}
+}
+
+// HashAggRuntimeStats record the HashAggExec runtime stat
+type HashAggRuntimeStats struct {
+	PartialNum int
+	FinalNum   int
+
+	PartialTaskNum int64
+	FinalTaskNum   int64
+
+	PartialWallTime time.Duration
+	FinalWallTime   time.Duration
+
+	PartialWorkerTime []time.Duration
+	FinalWorkerTime   []time.Duration
+
+	PartialWaitTime  int64
+	PartialExecTime  int64
+	PartialTotalTime int64
+}
+
+func (e *HashAggRuntimeStats) String() string {
+	var result string
+	if e.PartialWallTime != 0 {
+		result += fmt.Sprintf("partial_worker:{wall_time:%s, concurrency:%d, task_num:%d, tot_wait:%s, tot_exec:%s, tot_time:%s", e.PartialWallTime, e.PartialNum, e.PartialTaskNum, time.Duration(e.PartialWaitTime), time.Duration(e.PartialExecTime), time.Duration(e.PartialTotalTime))
+	}
+	if e.PartialWorkerTime != nil {
+		sort.Slice(e.PartialWorkerTime, func(i, j int) bool { return e.PartialWorkerTime[i] < e.PartialWorkerTime[j] })
+		n := len(e.PartialWorkerTime)
+		if n != 0 {
+			result += fmt.Sprintf(", max:%v, p95:%v", e.PartialWorkerTime[n-1], e.PartialWorkerTime[n*19/20])
+		}
+	}
+	if len(result) > 0 {
+		result += "}, "
+	}
+	if e.FinalWallTime != 0 {
+		result += fmt.Sprintf("final_worker:{wall_time:%s, concurrency:%d, task_num:%d", e.FinalWallTime, e.FinalNum, e.FinalTaskNum)
+	}
+	if e.FinalWorkerTime != nil {
+		sort.Slice(e.FinalWorkerTime, func(i, j int) bool { return e.FinalWorkerTime[i] < e.FinalWorkerTime[j] })
+		m := len(e.FinalWorkerTime)
+		if m != 0 {
+			result += fmt.Sprintf(", max:%v, p95:%v", e.FinalWorkerTime[m-1], e.FinalWorkerTime[m*19/20])
+		}
+	}
+	if len(result) > 0 {
+		return result + "}"
+	}
+	return result
+}
+
+// Clone implements the RuntimeStats interface.
+func (e *HashAggRuntimeStats) Clone() execdetails.RuntimeStats {
+	newRs := &HashAggRuntimeStats{
+		PartialTaskNum:    e.PartialTaskNum,
+		FinalTaskNum:      e.FinalTaskNum,
+		PartialWallTime:   e.PartialWallTime,
+		FinalWallTime:     e.FinalWallTime,
+		PartialNum:        e.PartialNum,
+		FinalNum:          e.FinalNum,
+		PartialWorkerTime: e.PartialWorkerTime,
+		FinalWorkerTime:   e.FinalWorkerTime,
+		PartialExecTime:   e.PartialExecTime,
+		PartialTotalTime:  e.PartialTotalTime,
+		PartialWaitTime:   e.PartialWaitTime,
+	}
+	return newRs
+}
+
+// Merge implements the RuntimeStats interface.
+func (e *HashAggRuntimeStats) Merge(other execdetails.RuntimeStats) {
+	tmp, ok := other.(*HashAggRuntimeStats)
+	if !ok {
+		return
+	}
+	e.PartialWallTime += tmp.PartialWallTime
+	e.FinalWallTime += tmp.FinalWallTime
+	e.PartialTaskNum += tmp.PartialTaskNum
+	e.FinalTaskNum += tmp.FinalTaskNum
+	e.PartialNum += tmp.PartialNum
+	e.FinalNum += tmp.FinalNum
+	e.PartialWaitTime += tmp.PartialWaitTime
+	e.PartialTotalTime += tmp.PartialTotalTime
+	e.PartialExecTime += tmp.PartialExecTime
+	e.FinalWorkerTime = append(e.FinalWorkerTime, tmp.FinalWorkerTime...)
+	e.PartialWorkerTime = append(e.PartialWorkerTime, tmp.PartialWorkerTime...)
+}
+
+// Tp implements the RuntimeStats interface.
+func (e *HashAggRuntimeStats) Tp() int {
+	return execdetails.TpHashAggRuntimeStat
+}
+
 // StreamAggExec deals with all the aggregate functions.
 // It assumes all the input data is sorted by group by key.
 // When Next() is called, it will return a result for the same group.

executor/aggregate_test.go

@@ -15,10 +15,12 @@ package executor_test
 
 import (
 	"fmt"
+	"time"
 
 	. "github.com/pingcap/check"
 	"github.com/pingcap/errors"
 	"github.com/pingcap/parser/terror"
+	"github.com/pingcap/tidb/executor"
 	plannercore "github.com/pingcap/tidb/planner/core"
 	"github.com/pingcap/tidb/util/testkit"
 	"github.com/pingcap/tidb/util/testutil"
@@ -1143,28 +1145,22 @@ func (s *testSuiteAgg) TestIssue17216(c *C) {
 	tk.MustQuery("SELECT count(distinct col1) FROM t1").Check(testkit.Rows("48"))
 }
 
-func (s *testSuiteAgg) TestIssue19426(c *C) {
-	tk := testkit.NewTestKitWithInit(c, s.store)
-	tk.MustExec("use test")
-	tk.MustExec("drop table if exists t")
-	tk.MustExec("create table t(a int primary key, b int)")
-	tk.MustExec("insert into t values (1, 11), (4, 44), (2, 22), (3, 33)")
-	tk.MustQuery("select sum(case when a <= 0 or a > 1000 then 0.0 else b end) from t").
-		Check(testkit.Rows("110.0"))
-	tk.MustQuery("select avg(case when a <= 0 or a > 1000 then 0.0 else b end) from t").
-		Check(testkit.Rows("27.50000"))
-	tk.MustQuery("select distinct (case when a <= 0 or a > 1000 then 0.0 else b end) v from t order by v").
-		Check(testkit.Rows("11.0", "22.0", "33.0", "44.0"))
-	tk.MustQuery("select group_concat(case when a <= 0 or a > 1000 then 0.0 else b end order by -a) from t").
-		Check(testkit.Rows("44.0,33.0,22.0,11.0"))
-	tk.MustQuery("select group_concat(a, b, case when a <= 0 or a > 1000 then 0.0 else b end order by -a) from t").
-		Check(testkit.Rows("44444.0,33333.0,22222.0,11111.0"))
-	tk.MustQuery("select group_concat(distinct case when a <= 0 or a > 1000 then 0.0 else b end order by -a) from t").
-		Check(testkit.Rows("44.0,33.0,22.0,11.0"))
-	tk.MustQuery("select max(case when a <= 0 or a > 1000 then 0.0 else b end) from t").
-		Check(testkit.Rows("44.0"))
-	tk.MustQuery("select min(case when a <= 0 or a > 1000 then 0.0 else b end) from t").
-		Check(testkit.Rows("11.0"))
-	tk.MustQuery("select a, b, sum(case when a < 1000 then b else 0.0 end) over (order by a) from t").
-		Check(testkit.Rows("1 11 11.0", "2 22 33.0", "3 33 66.0", "4 44 110.0"))
+func (s *testSuiteAgg) TestHashAggRuntimeStat(c *C) {
+	stats := &executor.HashAggRuntimeStats{
+		PartialNum:        5,
+		FinalNum:          5,
+		PartialTaskNum:    5,
+		PartialWallTime:   1 * time.Second,
+		FinalTaskNum:      5,
+		FinalWallTime:     2 * time.Second,
+		PartialWorkerTime: []time.Duration{1 * time.Second, 1 * time.Second, 1 * time.Second, 2 * time.Second, 3 * time.Second},
+		FinalWorkerTime:   []time.Duration{1 * time.Second, 1 * time.Second, 1 * time.Second, 2 * time.Second, 4 * time.Second},
+		PartialWaitTime:   int64(2 * time.Second),
+		PartialTotalTime:  int64(2 * time.Second),
+		PartialExecTime:   int64(2 * time.Second),
+	}
+	c.Assert(stats.String(), Equals, "partial_worker:{wall_time:1s, concurrency:5, task_num:5, tot_wait:2s, tot_exec:2s, tot_time:2s, max:3s, p95:3s}, final_worker:{wall_time:2s, concurrency:5, task_num:5, max:4s, p95:4s}")
+	c.Assert(stats.String(), Equals, stats.Clone().String())
+	stats.Merge(stats.Clone())
+	c.Assert(stats.String(), Equals, "partial_worker:{wall_time:2s, concurrency:10, task_num:10, tot_wait:4s, tot_exec:4s, tot_time:4s, max:3s, p95:3s}, final_worker:{wall_time:4s, concurrency:10, task_num:10, max:4s, p95:4s}")
 }