code: func (e *AnalyzeExec) Next(ctx context.Context, req *chunk.RecordBatch) error {
  	concurrency, err := getBuildStatsConcurrency(e.ctx)
  	if err != nil {
  		return err
  	}
  	taskCh := make(chan *analyzeTask, len(e.tasks))
  	resultCh := make(chan analyzeResult, len(e.tasks))
  	e.wg.Add(concurrency)
  	for i := 0; i < concurrency; i++ {
  		go e.analyzeWorker(taskCh, resultCh, i == 0)
  	}
  	for _, task := range e.tasks {
  		statistics.AddNewAnalyzeJob(task.job)
  	}
  	for _, task := range e.tasks {
  		taskCh <- task
  	}
  	close(taskCh)
  	statsHandle := domain.GetDomain(e.ctx).StatsHandle()
  	panicCnt := 0
  	for panicCnt < concurrency {
  		result, ok := <-resultCh
  		if !ok {
  			break
  		}
  		if result.Err != nil {
  			err = result.Err
  			if err == errAnalyzeWorkerPanic {
  				panicCnt++
  			} else {
  				logutil.Logger(ctx).Error("analyze failed", zap.Error(err))
  			}
  			result.job.Finish(true)
  			continue
  		}
  		for i, hg := range result.Hist {
  			err1 := statsHandle.SaveStatsToStorage(result.PhysicalTableID, result.Count, result.IsIndex, hg, result.Cms[i], 1)
  			if err1 != nil {
  				err = err1
  				logutil.Logger(ctx).Error("save stats to storage failed", zap.Error(err))
  				result.job.Finish(true)
  				continue
  			}
  		}
  		result.job.Finish(false)
  	}
  	for _, task := range e.tasks {
  		statistics.MoveToHistory(task.job)
  	}
  	if err != nil {
  		return err
  	}
  	return statsHandle.Update(GetInfoSchema(e.ctx))
  }
docs: // Next implements the Executor Next interface.