Repository: pingcap/tidb
File: pkg/executor/analyze_col_v2.go (Go)

Code before:

	resultCh <- &samplingMergeResult{collector: retCollector}
}

func (e *AnalyzeColumnsExecV2) subBuildWorker(resultCh chan error, taskCh chan *samplingBuildTask, hists []*statistics.Histogram, topns []*statistics.TopN, collectors []*statistics.SampleCollector, exitCh chan struct{}) {
	defer func() {
		if r := recover(); r != nil {
			logutil.BgLogger().Warn("analyze worker panicked", zap.Any("recover", r), zap.Stack("stack"))
			metrics.PanicCounter.WithLabelValues(metrics.LabelAnalyze).Inc()
			resultCh <- getAnalyzePanicErr(r)
		}
	}()
	failpoint.Inject("mockAnalyzeSamplingBuildWorkerPanic", func() {
		panic("failpoint triggered")
	})

	colLen := len(e.colsInfo)

workLoop:
	for {
		select {
		case task, ok := <-taskCh:
			if !ok {
				break workLoop
			}
			bufferedMemSize := int64(0)
			totalBuffered := int64(0)

			consumeBuffered := func(bytes int64) {
				totalBuffered += bytes
				e.memTracker.BufferedConsume(&bufferedMemSize, bytes)
			}
			flushBuffered := func(cum *int64) {
				if bufferedMemSize != 0 {
					e.memTracker.Consume(bufferedMemSize)
					bufferedMemSize = 0
				}
				*cum += totalBuffered
				totalBuffered = 0
			}
			var collector *statistics.SampleCollector
			if task.isColumn {
				if e.colsInfo[task.slicePos].IsGenerated() && !e.colsInfo[task.slicePos].GeneratedStored {
					hists[task.slicePos] = nil
					topns[task.slicePos] = nil
					continue
				}
				sampleNum := task.rootRowCollector.Base().Samples.Len()
				sampleItems := make([]*statistics.SampleItem, 0, sampleNum)
				// consume mandatory memory at the beginning, including empty SampleItems of all sample rows, if exceeds, fast fail
				// 8 means the pointer size of sampleItems slice.

Review comment (suggestion):
```suggestion
			// Track per-task allocations: currentBufferedMemSize is pending charges to the tracker,
			// totalBuffered accumulates bytes that become part of collector.MemSize.
			currentBufferedMemSize := int64(0)
```

Code after:

	resultCh <- &samplingMergeResult{collector: retCollector}
}

func (e *AnalyzeColumnsExecV2) subBuildWorker(resultCh chan error, taskCh chan *samplingBuildTask, hists []*statistics.Histogram, topns []*statistics.TopN, collectors []*statistics.SampleCollector, exitCh chan struct{}) {
	defer func() {
		if r := recover(); r != nil {
			logutil.BgLogger().Warn("analyze worker panicked", zap.Any("recover", r), zap.Stack("stack"))
			metrics.PanicCounter.WithLabelValues(metrics.LabelAnalyze).Inc()
			resultCh <- getAnalyzePanicErr(r)
		}
	}()
	failpoint.Inject("mockAnalyzeSamplingBuildWorkerPanic", func() {
		panic("failpoint triggered")
	})

	colLen := len(e.colsInfo)

workLoop:
	for {
		select {
		case task, ok := <-taskCh:
			if !ok {
				break workLoop
			}
			// Track per-task allocations: curBufferedMemSize is pending charges to the tracker,
			// totalBuffered accumulates bytes that become part of collector.MemSize.
			curBufferedMemSize := int64(0)
			totalBuffered := int64(0)

			consumeBuffered := func(bytes int64) {
				totalBuffered += bytes
				e.memTracker.BufferedConsume(&curBufferedMemSize, bytes)
			}
			flushBuffered := func(cum *int64) {
				if curBufferedMemSize != 0 {
					e.memTracker.Consume(curBufferedMemSize)
					curBufferedMemSize = 0
				}
				*cum += totalBuffered
				totalBuffered = 0
			}
			var collector *statistics.SampleCollector
			if task.isColumn {
				if e.colsInfo[task.slicePos].IsGenerated() && !e.colsInfo[task.slicePos].GeneratedStored {
					hists[task.slicePos] = nil
					topns[task.slicePos] = nil
					continue
				}
				sampleNum := task.rootRowCollector.Base().Samples.Len()
				sampleItems := make([]*statistics.SampleItem, 0, sampleNum)