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

Code before:

func newAnalyzeSaveStatsWorker(
	resultsCh <-chan *statistics.AnalyzeResults,
	errCh chan<- error,
	killer *sqlkiller.SQLKiller) *analyzeSaveStatsWorker {
	worker := &analyzeSaveStatsWorker{
		resultsCh: resultsCh,
		errCh:     errCh,
		killer:    killer,
	}
	return worker
}

func (worker *analyzeSaveStatsWorker) run(ctx context.Context, statsHandle *handle.Handle, analyzeSnapshot bool) {
	errReported := false
	var drainErr error
	defer func() {
		if r := recover(); r != nil {
			logutil.BgLogger().Error("analyze save stats worker panicked", zap.Any("recover", r), zap.Stack("stack"))
			if !errReported {
				worker.errCh <- getAnalyzePanicErr(r)
				errReported = true
			}
		}
	}()
	for results := range worker.resultsCh {
		if drainErr != nil {
			finishJobWithLog(statsHandle, results.Job, drainErr)
			results.DestroyAndPutToPool()
			continue
		}
		failpoint.InjectCall("analyzeSaveWorkerBeforeHandleSignal")
		if err := worker.killer.HandleSignal(); err != nil {
			drainErr = err
			finishJobWithLog(statsHandle, results.Job, drainErr)
			results.DestroyAndPutToPool()
			if !errReported {
				worker.errCh <- drainErr
				errReported = true
			}
			continue
		}
		err := statsHandle.SaveAnalyzeResultToStorage(results, analyzeSnapshot, util.StatsMetaHistorySourceAnalyze)
		if err != nil {
			logutil.Logger(ctx).Warn("save table stats to storage failed", zap.Error(err))
			finishJobWithLog(statsHandle, results.Job, err)
			if !errReported {
				worker.errCh <- err
				errReported = true
			}
			// Keep draining results to avoid blocking analyze workers after a save failure.

Review comment (bug):
Trick: This error is used solely for handling the kill signal. This allows us to attempt saving the statistics of other partitions, even if the previous partition encountered an error while saving the result. I believe this is acceptable because, prior to this fix, we often ended up in this situation: the analyze command failed, but the statistics were partially stored. Therefore, I think this fix won't worsen the situation.

Code after:

func newAnalyzeSaveStatsWorker(
	resultsCh <-chan *statistics.AnalyzeResults,
	errCh chan<- error,
	killer *sqlkiller.SQLKiller) *analyzeSaveStatsWorker {
	worker := &analyzeSaveStatsWorker{
		resultsCh: resultsCh,
		errCh:     errCh,
		killer:    killer,
	}
	return worker
}

// run consumes analyze results and persists them. After a kill signal, it
// enters "drain mode": do not save more stats, just keep consuming resultsCh
// and finish jobs with the same error so upstream analyze workers are not
// blocked on sending.
func (worker *analyzeSaveStatsWorker) run(ctx context.Context, statsHandle *handle.Handle, analyzeSnapshot bool) {
	// Report at most one error per save worker. errCh is drained only after all
	// save workers exit, so sending every per-partition failure can fill errCh
	// and block this worker, which may deadlock the analyze pipeline.
	errReported := false
	// drainErr is only used for kill-signal handling. Save failures should not
	// switch to drain mode, so we can still try to persist stats for later
	// partitions. ANALYZE may already end up with partially persisted stats.
	var drainErr error
	defer func() {
		if r := recover(); r != nil {
			logutil.BgLogger().Error("analyze save stats worker panicked", zap.Any("recover", r), zap.Stack("stack"))
			if !errReported {
				worker.errCh <- getAnalyzePanicErr(r)
				errReported = true
			}
		}
	}()
	for results := range worker.resultsCh {
		if drainErr != nil {
			// Drain mode: consume the remaining results only to unblock producers.
			finishJobWithLog(statsHandle, results.Job, drainErr)
			results.DestroyAndPutToPool()
			continue
		}
		failpoint.InjectCall("analyzeSaveWorkerBeforeHandleSignal")
		if err := worker.killer.HandleSignal(); err != nil {
			drainErr = err
			finishJobWithLog(statsHandle, results.Job, drainErr)
			results.DestroyAndPutToPool()
			if !errReported {
				worker.errCh <- drainErr
				errReported = true
			}