func (w *worker) waitSchemaChanged(ctx context.Context, d *ddlCtx, waitTime time.Duration, latestSchemaVersion int64, job *model.Job) {
	if !job.IsRunning() && !job.IsRollingback() && !job.IsDone() && !job.IsRollbackDone() {
		return
	}
	if waitTime == 0 {
		return
	}
	timeStart := time.Now()
	var err error
	defer func() {
	}()
	if latestSchemaVersion == 0 {
		logutil.Logger(w.logCtx).Info("[ddl] schema version doesn't change")
		return
	}
	if ctx == nil {
		var cancelFunc context.CancelFunc
		ctx, cancelFunc = context.WithTimeout(context.Background(), waitTime)
		defer cancelFunc()
	}
	err = d.schemaSyncer.OwnerUpdateGlobalVersion(ctx, latestSchemaVersion)
	if err != nil {
		logutil.Logger(w.logCtx).Info("[ddl] update latest schema version failed", zap.Int64("ver", latestSchemaVersion), zap.Error(err))
		if terror.ErrorEqual(err, context.DeadlineExceeded) {
			return
		}
	}
	err = d.schemaSyncer.OwnerCheckAllVersions(ctx, latestSchemaVersion)
	if err != nil {
		logutil.Logger(w.logCtx).Info("[ddl] wait latest schema version to deadline", zap.Int64("ver", latestSchemaVersion), zap.Error(err))
		if terror.ErrorEqual(err, context.DeadlineExceeded) {
			return
		}
		d.schemaSyncer.NotifyCleanExpiredPaths()
		select {
		case <-ctx.Done():
			return
		}
	}
	logutil.Logger(w.logCtx).Info("[ddl] wait latest schema version changed",
		zap.Int64("ver", latestSchemaVersion),
		zap.Duration("take time", time.Since(timeStart)),
		zap.String("job", job.String()))
}
waitSchemaChanged waits for the completion of updating all servers' schema. In order to make sure that happens,
we wait 2 * lease time.
package ddl
import (
	"context"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/tidb/ddl/util"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/meta"
	"github.com/pingcap/tidb/parser/model"
	"github.com/pingcap/tidb/parser/terror"
	"github.com/pingcap/tidb/sessionctx"
	"github.com/pingcap/tidb/sessionctx/variable"
	tidbutil "github.com/pingcap/tidb/util"
	"github.com/pingcap/tidb/util/admin"
	"github.com/pingcap/tidb/util/logutil"
	"go.uber.org/zap"
)
var (
	ddlWorkerID = int32(0)
	WaitTimeWhenErrorOccurred = 1 * time.Second
)
type workerType byte
const (
	generalWorker workerType = 0
	addIdxWorker workerType = 1
	waitDependencyJobInterval = 200 * time.Millisecond
	noneDependencyJob = 0
)
type worker struct {
	id       int32
	tp       workerType
	ddlJobCh chan struct{}
	quitCh   chan struct{}
	wg       sync.WaitGroup
	sessPool *sessionPool  
	reorgCtx *reorgCtx     
	logCtx   context.Context
}
func newWorker(tp workerType, sessPool *sessionPool) *worker {
	worker := &worker{
		id:       atomic.AddInt32(&ddlWorkerID, 1),
		tp:       tp,
		ddlJobCh: make(chan struct{}, 1),
		quitCh:   make(chan struct{}),
		reorgCtx: &reorgCtx{notifyCancelReorgJob: 0},
		sessPool: sessPool,
	}
	worker.logCtx = logutil.WithKeyValue(context.Background(), "worker", worker.String())
	return worker
}
func (w *worker) typeStr() string {
	var str string
	switch w.tp {
	case generalWorker:
		str = "general"
	case addIdxWorker:
		str = model.AddIndexStr
	default:
		str = "unknow"
	}
	return str
}
func (w *worker) String() string {
	return fmt.Sprintf("worker %d, tp %s", w.id, w.typeStr())
}
func (w *worker) close() {
	startTime := time.Now()
	close(w.quitCh)
	w.wg.Wait()
	logutil.Logger(w.logCtx).Info("[ddl] DDL worker closed", zap.Duration("take time", time.Since(startTime)))
}
func (w *worker) start(d *ddlCtx) {
	logutil.Logger(w.logCtx).Info("[ddl] start DDL worker")
	defer w.wg.Done()
	checkTime := chooseLeaseTime(2*d.lease, 1*time.Second)
	ticker := time.NewTicker(checkTime)
	defer ticker.Stop()
	for {
		select {
		case <-ticker.C:
			logutil.Logger(w.logCtx).Debug("[ddl] wait to check DDL status again", zap.Duration("interval", checkTime))
		case <-w.ddlJobCh:
		case <-w.quitCh:
			return
		}
		err := w.handleDDLJobQueue(d)
		if err != nil {
			logutil.Logger(w.logCtx).Error("[ddl] handle DDL job failed", zap.Error(err))
		}
	}
}
func asyncNotify(ch chan struct{}) {
	select {
	case ch <- struct{}{}:
	default:
	}
}
func buildJobDependence(t *meta.Meta, curJob *model.Job) error {
	var jobs []*model.Job
	var err error
	switch curJob.Type {
	case model.ActionAddIndex, model.ActionAddPrimaryKey:
		jobs, err = t.GetAllDDLJobsInQueue(meta.DefaultJobListKey)
	default:
		jobs, err = t.GetAllDDLJobsInQueue(meta.AddIndexJobListKey)
	}
	if err != nil {
		return errors.Trace(err)
	}
	for _, job := range jobs {
		if curJob.ID < job.ID {
			continue
		}
		isDependent, err := curJob.IsDependentOn(job)
		if err != nil {
			return errors.Trace(err)
		}
		if isDependent {
			logutil.BgLogger().Info("[ddl] current DDL job depends on other job", zap.String("currentJob", curJob.String()), zap.String("dependentJob", job.String()))
			curJob.DependencyID = job.ID
			break
		}
	}
	return nil
}
func (d *ddl) addDDLJob(ctx sessionctx.Context, job *model.Job) error {
	job.Version = currentVersion
	job.Query, _ = ctx.Value(sessionctx.QueryString).(string)
	err := kv.RunInNewTxn(d.store, true, func(txn kv.Transaction) error {
		t := newMetaWithQueueTp(txn, job.Type.String())
		var err error
		job.ID, err = t.GenGlobalID()
		if err != nil {
			return errors.Trace(err)
		}
		job.StartTS = txn.StartTS()
		if err = buildJobDependence(t, job); err != nil {
			return errors.Trace(err)
		}
		err = t.EnQueueDDLJob(job)
		return errors.Trace(err)
	})
	return errors.Trace(err)
}
func (d *ddl) getHistoryDDLJob(id int64) (*model.Job, error) {
	var job *model.Job
	err := kv.RunInNewTxn(d.store, false, func(txn kv.Transaction) error {
		t := meta.NewMeta(txn)
		var err1 error
		job, err1 = t.GetHistoryDDLJob(id)
		return errors.Trace(err1)
	})
	return job, errors.Trace(err)
}
func (w *worker) getFirstDDLJob(t *meta.Meta) (*model.Job, error) {
	job, err := t.GetDDLJobByIdx(0)
	return job, errors.Trace(err)
}
func (w *worker) handleUpdateJobError(t *meta.Meta, job *model.Job, err error) error {
	if err == nil {
		return nil
	}
	if kv.ErrEntryTooLarge.Equal(err) {
		logutil.Logger(w.logCtx).Warn("[ddl] update DDL job failed", zap.String("job", job.String()), zap.Error(err))
		job.BinlogInfo.Clean()
		job.Error = toTError(err)
		job.SchemaState = model.StateNone
		job.State = model.JobStateCancelled
		err = w.finishDDLJob(t, job)
	}
	return errors.Trace(err)
}
func (w *worker) updateDDLJob(t *meta.Meta, job *model.Job, meetErr bool) error {
	updateRawArgs := true
	if meetErr && (job.RawArgs != nil && job.Args == nil) {
		logutil.Logger(w.logCtx).Info("[ddl] meet something wrong before update DDL job, shouldn't update raw args",
			zap.String("job", job.String()))
		updateRawArgs = false
	}
	return errors.Trace(t.UpdateDDLJob(0, job, updateRawArgs))
}
func (w *worker) finishDDLJob(t *meta.Meta, job *model.Job) (err error) {
	_, err = t.DeQueueDDLJob()
	if err != nil {
		return errors.Trace(err)
	}
	job.BinlogInfo.FinishedTS = t.StartTS
	logutil.Logger(w.logCtx).Info("[ddl] finish DDL job", zap.String("job", job.String()))
	updateRawArgs := true
	if job.Type == model.ActionAddPrimaryKey && !job.IsCancelled() {
		updateRawArgs = false
	}
	err = t.AddHistoryDDLJob(job, updateRawArgs)
	return errors.Trace(err)
}
func isDependencyJobDone(t *meta.Meta, job *model.Job) (bool, error) {
	if job.DependencyID == noneDependencyJob {
		return true, nil
	}
	historyJob, err := t.GetHistoryDDLJob(job.DependencyID)
	if err != nil {
		return false, errors.Trace(err)
	}
	if historyJob == nil {
		return false, nil
	}
	logutil.BgLogger().Info("[ddl] current DDL job dependent job is finished", zap.String("currentJob", job.String()), zap.Int64("dependentJobID", job.DependencyID))
	job.DependencyID = noneDependencyJob
	return true, nil
}
func newMetaWithQueueTp(txn kv.Transaction, tp string) *meta.Meta {
	if tp == model.AddIndexStr || tp == model.AddPrimaryKeyStr {
		return meta.NewMeta(txn, meta.AddIndexJobListKey)
	}
	return meta.NewMeta(txn)
}
func (w *worker) handleDDLJobQueue(d *ddlCtx) error {
	once := true
	waitDependencyJobCnt := 0
	for {
		if isChanClosed(w.quitCh) {
			return nil
		}
		var (
			job       *model.Job
			schemaVer int64
			runJobErr error
		)
		waitTime := 2 * d.lease
		err := kv.RunInNewTxn(d.store, false, func(txn kv.Transaction) error {
			if !d.isOwner() {
				return nil
			}
			var err error
			t := newMetaWithQueueTp(txn, w.typeStr())
			job, err = w.getFirstDDLJob(t)
			if job == nil || err != nil {
				return errors.Trace(err)
			}
			if isDone, err1 := isDependencyJobDone(t, job); err1 != nil || !isDone {
				return errors.Trace(err1)
			}
			if once {
				w.waitSchemaSynced(d, job, waitTime)
				once = false
				return nil
			}
			if job.IsDone() || job.IsRollbackDone() {
				if !job.IsRollbackDone() {
					job.State = model.JobStateSynced
				}
				err = w.finishDDLJob(t, job)
				return errors.Trace(err)
			}
			d.mu.RLock()
			d.mu.hook.OnJobRunBefore(job)
			d.mu.RUnlock()
			tidbutil.WithRecovery(func() {
				schemaVer, runJobErr = w.runDDLJob(d, t, job)
			}, func(r interface{}) {
				if r != nil {
					job.State = model.JobStateCancelling
				}
			})
			if job.IsCancelled() {
				txn.Reset()
				err = w.finishDDLJob(t, job)
				return errors.Trace(err)
			}
			err = w.updateDDLJob(t, job, runJobErr != nil)
			if err = w.handleUpdateJobError(t, job, err); err != nil {
				return errors.Trace(err)
			}
			return nil
		})
		if runJobErr != nil {
			logutil.Logger(w.logCtx).Info("[ddl] run DDL job failed, sleeps a while then retries it.",
				zap.Duration("waitTime", WaitTimeWhenErrorOccurred), zap.Error(runJobErr))
			time.Sleep(WaitTimeWhenErrorOccurred)
		}
		if err != nil {
			return errors.Trace(err)
		} else if job == nil {
			return nil
		}
		w.waitDependencyJobFinished(job, &waitDependencyJobCnt)
		d.mu.RLock()
		d.mu.hook.OnJobUpdated(job)
		d.mu.RUnlock()
		w.waitSchemaChanged(nil, d, waitTime, schemaVer, job)
		if job.IsSynced() || job.IsCancelled() {
			asyncNotify(d.ddlJobDoneCh)
		}
	}
}
func (w *worker) waitDependencyJobFinished(job *model.Job, cnt *int) {
	if job.DependencyID != noneDependencyJob {
		intervalCnt := int(3 * time.Second / waitDependencyJobInterval)
		if *cnt%intervalCnt == 0 {
			logutil.Logger(w.logCtx).Info("[ddl] DDL job need to wait dependent job, sleeps a while, then retries it.",
				zap.Int64("jobID", job.ID),
				zap.Int64("dependentJobID", job.DependencyID),
				zap.Duration("waitTime", waitDependencyJobInterval))
		}
		time.Sleep(waitDependencyJobInterval)
		*cnt++
	} else {
		*cnt = 0
	}
}
func chooseLeaseTime(t, max time.Duration) time.Duration {
	if t == 0 || t > max {
		return max
	}
	return t
}
func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, err error) {
	failpoint.Inject("mockPanicInRunDDLJob", func(val failpoint.Value) {})
	logutil.Logger(w.logCtx).Info("[ddl] run DDL job", zap.String("job", job.String()))
	if job.IsFinished() {
		return
	}
	if job.IsCancelling() {
		return convertJob2RollbackJob(w, d, t, job)
	}
	if !job.IsRollingback() && !job.IsCancelling() {
		job.State = model.JobStateRunning
	}
	switch job.Type {
	case model.ActionCreateSchema:
		ver, err = onCreateSchema(d, t, job)
	case model.ActionModifySchemaCharsetAndCollate:
		ver, err = onModifySchemaCharsetAndCollate(t, job)
	case model.ActionDropSchema:
		ver, err = onDropSchema(t, job)
	case model.ActionCreateTable:
		ver, err = onCreateTable(d, t, job)
	case model.ActionDropTable:
		ver, err = onDropTableOrView(t, job)
	case model.ActionAddColumn:
		ver, err = onAddColumn(d, t, job)
	case model.ActionDropColumn:
		ver, err = onDropColumn(t, job)
	case model.ActionModifyColumn:
		ver, err = w.onModifyColumn(t, job)
	case model.ActionSetDefaultValue:
		ver, err = onSetDefaultValue(t, job)
	case model.ActionAddIndex:
		ver, err = w.onCreateIndex(d, t, job, false)
	case model.ActionAddPrimaryKey:
		ver, err = w.onCreateIndex(d, t, job, true)
	case model.ActionDropIndex, model.ActionDropPrimaryKey:
		ver, err = onDropIndex(t, job)
	case model.ActionShardRowID:
		ver, err = w.onShardRowID(d, t, job)
	case model.ActionModifyTableComment:
		ver, err = onModifyTableComment(t, job)
	case model.ActionModifyTableCharsetAndCollate:
		ver, err = onModifyTableCharsetAndCollate(t, job)
	default:
		job.State = model.JobStateCancelled
		err = errInvalidDDLJob.GenWithStack("invalid ddl job type: %v", job.Type)
	}
	if err != nil {
		job.Error = toTError(err)
		job.ErrorCount++
		if job.State == model.JobStateCancelled {
			logutil.Logger(w.logCtx).Info("[ddl] DDL job is cancelled normally", zap.Error(err))
			return ver, nil
		}
		logutil.Logger(w.logCtx).Error("[ddl] run DDL job error", zap.Error(err))
		if err1 := loadDDLVars(w); err1 != nil {
			logutil.Logger(w.logCtx).Error("[ddl] load DDL global variable failed", zap.Error(err1))
		}
		if job.ErrorCount > variable.GetDDLErrorCountLimit() && job.State == model.JobStateRunning && admin.IsJobRollbackable(job) {
			logutil.Logger(w.logCtx).Warn("[ddl] DDL job error count exceed the limit, cancelling it now", zap.Int64("jobID", job.ID), zap.Int64("errorCountLimit", variable.GetDDLErrorCountLimit()))
			job.State = model.JobStateCancelling
		}
	}
	return
}
func loadDDLVars(w *worker) error {
	var ctx sessionctx.Context
	ctx, err := w.sessPool.get()
	if err != nil {
		return errors.Trace(err)
	}
	defer w.sessPool.put(ctx)
	return util.LoadDDLVars(ctx)
}
func toTError(err error) *terror.Error {
	originErr := errors.Cause(err)
	tErr, ok := originErr.(*terror.Error)
	if ok {
		return tErr
	}
	return terror.ClassDDL.New(terror.CodeUnknown, err.Error())
}