func (do *Domain) GetEtcdClient() *clientv3.Client {
	return do.etcdClient
}
GetEtcdClient returns the etcd client.
package domain
import (
	"context"
	"os"
	"sync"
	"sync/atomic"
	"time"
	"unsafe"
	"github.com/ngaut/pools"
	"github.com/pingcap/errors"
	"github.com/pingcap/failpoint"
	"github.com/pingcap/tidb/ddl"
	"github.com/pingcap/tidb/infoschema"
	"github.com/pingcap/tidb/kv"
	"github.com/pingcap/tidb/meta"
	"github.com/pingcap/tidb/parser/model"
	"github.com/pingcap/tidb/parser/mysql"
	"github.com/pingcap/tidb/parser/terror"
	"github.com/pingcap/tidb/sessionctx"
	"github.com/pingcap/tidb/sessionctx/variable"
	"github.com/pingcap/tidb/statistics"
	"github.com/pingcap/tidb/store/tikv"
	"github.com/pingcap/tidb/util"
	"github.com/pingcap/tidb/util/logutil"
	"go.etcd.io/etcd/clientv3"
	"go.uber.org/zap"
	"google.golang.org/grpc"
	"google.golang.org/grpc/keepalive"
)
type Domain struct {
	store           kv.Storage
	infoHandle      *infoschema.Handle
	statsHandle     unsafe.Pointer
	statsLease      time.Duration
	ddl             ddl.DDL
	m               sync.Mutex
	SchemaValidator SchemaValidator
	sysSessionPool  *sessionPool
	exit            chan struct{}
	etcdClient      *clientv3.Client
	gvc             GlobalVariableCache
	wg              sync.WaitGroup
}
func (do *Domain) loadInfoSchema(handle *infoschema.Handle, usedSchemaVersion int64, startTS uint64) (int64, []int64, bool, error) {
	var fullLoad bool
	snapshot, err := do.store.GetSnapshot(kv.NewVersion(startTS))
	if err != nil {
		return 0, nil, fullLoad, err
	}
	m := meta.NewSnapshotMeta(snapshot)
	neededSchemaVersion, err := m.GetSchemaVersion()
	if err != nil {
		return 0, nil, fullLoad, err
	}
	if usedSchemaVersion != 0 && usedSchemaVersion == neededSchemaVersion {
		return neededSchemaVersion, nil, fullLoad, nil
	}
	defer func() {
		if err != nil || neededSchemaVersion < do.InfoSchema().SchemaMetaVersion() {
			logutil.BgLogger().Info("do not update self schema version to etcd",
				zap.Int64("usedSchemaVersion", usedSchemaVersion),
				zap.Int64("neededSchemaVersion", neededSchemaVersion), zap.Error(err))
			return
		}
		err = do.ddl.SchemaSyncer().UpdateSelfVersion(context.Background(), neededSchemaVersion)
		if err != nil {
			logutil.BgLogger().Info("update self version failed",
				zap.Int64("usedSchemaVersion", usedSchemaVersion),
				zap.Int64("neededSchemaVersion", neededSchemaVersion), zap.Error(err))
		}
	}()
	startTime := time.Now()
	ok, tblIDs, err := do.tryLoadSchemaDiffs(m, usedSchemaVersion, neededSchemaVersion)
	if err != nil {
		logutil.BgLogger().Error("failed to load schema diff", zap.Error(err))
	}
	if ok {
		logutil.BgLogger().Info("diff load InfoSchema success",
			zap.Int64("usedSchemaVersion", usedSchemaVersion),
			zap.Int64("neededSchemaVersion", neededSchemaVersion),
			zap.Duration("start time", time.Since(startTime)),
			zap.Int64s("tblIDs", tblIDs))
		return neededSchemaVersion, tblIDs, fullLoad, nil
	}
	fullLoad = true
	schemas, err := do.fetchAllSchemasWithTables(m)
	if err != nil {
		return 0, nil, fullLoad, err
	}
	newISBuilder, err := infoschema.NewBuilder(handle).InitWithDBInfos(schemas, neededSchemaVersion)
	if err != nil {
		return 0, nil, fullLoad, err
	}
	logutil.BgLogger().Info("full load InfoSchema success",
		zap.Int64("usedSchemaVersion", usedSchemaVersion),
		zap.Int64("neededSchemaVersion", neededSchemaVersion),
		zap.Duration("start time", time.Since(startTime)))
	newISBuilder.Build()
	return neededSchemaVersion, nil, fullLoad, nil
}
func (do *Domain) fetchAllSchemasWithTables(m *meta.Meta) ([]*model.DBInfo, error) {
	allSchemas, err := m.ListDatabases()
	if err != nil {
		return nil, err
	}
	splittedSchemas := do.splitForConcurrentFetch(allSchemas)
	doneCh := make(chan error, len(splittedSchemas))
	for _, schemas := range splittedSchemas {
		go do.fetchSchemasWithTables(schemas, m, doneCh)
	}
	for range splittedSchemas {
		err = <-doneCh
		if err != nil {
			return nil, err
		}
	}
	return allSchemas, nil
}
const fetchSchemaConcurrency = 8
func (do *Domain) splitForConcurrentFetch(schemas []*model.DBInfo) [][]*model.DBInfo {
	groupSize := (len(schemas) + fetchSchemaConcurrency - 1) / fetchSchemaConcurrency
	splitted := make([][]*model.DBInfo, 0, fetchSchemaConcurrency)
	schemaCnt := len(schemas)
	for i := 0; i < schemaCnt; i += groupSize {
		end := i + groupSize
		if end > schemaCnt {
			end = schemaCnt
		}
		splitted = append(splitted, schemas[i:end])
	}
	return splitted
}
func (do *Domain) fetchSchemasWithTables(schemas []*model.DBInfo, m *meta.Meta, done chan error) {
	for _, di := range schemas {
		if di.State != model.StatePublic {
			continue
		}
		tables, err := m.ListTables(di.ID)
		if err != nil {
			done <- err
			return
		}
		di.Tables = make([]*model.TableInfo, 0, len(tables))
		for _, tbl := range tables {
			if tbl.State != model.StatePublic {
				continue
			}
			infoschema.ConvertCharsetCollateToLowerCaseIfNeed(tbl)
			di.Tables = append(di.Tables, tbl)
		}
	}
	done <- nil
}
const (
	initialVersion         = 0
	maxNumberOfDiffsToLoad = 100
)
func isTooOldSchema(usedVersion, newVersion int64) bool {
	if usedVersion == initialVersion || newVersion-usedVersion > maxNumberOfDiffsToLoad {
		return true
	}
	return false
}
func (do *Domain) tryLoadSchemaDiffs(m *meta.Meta, usedVersion, newVersion int64) (bool, []int64, error) {
	if isTooOldSchema(usedVersion, newVersion) {
		return false, nil, nil
	}
	var diffs []*model.SchemaDiff
	for usedVersion < newVersion {
		usedVersion++
		diff, err := m.GetSchemaDiff(usedVersion)
		if err != nil {
			return false, nil, err
		}
		if diff == nil {
			return false, nil, nil
		}
		diffs = append(diffs, diff)
	}
	builder := infoschema.NewBuilder(do.infoHandle).InitWithOldInfoSchema()
	tblIDs := make([]int64, 0, len(diffs))
	for _, diff := range diffs {
		ids, err := builder.ApplyDiff(m, diff)
		if err != nil {
			return false, nil, err
		}
		tblIDs = append(tblIDs, ids...)
	}
	builder.Build()
	return true, tblIDs, nil
}
func (do *Domain) InfoSchema() infoschema.InfoSchema {
	return do.infoHandle.Get()
}
func (do *Domain) DDL() ddl.DDL {
	return do.ddl
}
func (do *Domain) Store() kv.Storage {
	return do.store
}
func (do *Domain) GetScope(status string) variable.ScopeFlag {
	return variable.DefaultStatusVarScopeFlag
}
func (do *Domain) Reload() error {
	failpoint.Inject("ErrorMockReloadFailed", func(val failpoint.Value) {
		if val.(bool) {
			failpoint.Return(errors.New("mock reload failed"))
		}
	})
	do.m.Lock()
	defer do.m.Unlock()
	startTime := time.Now()
	var err error
	var neededSchemaVersion int64
	ver, err := do.store.CurrentVersion()
	if err != nil {
		return err
	}
	schemaVersion := int64(0)
	oldInfoSchema := do.infoHandle.Get()
	if oldInfoSchema != nil {
		schemaVersion = oldInfoSchema.SchemaMetaVersion()
	}
	var (
		fullLoad        bool
		changedTableIDs []int64
	)
	neededSchemaVersion, changedTableIDs, fullLoad, err = do.loadInfoSchema(do.infoHandle, schemaVersion, ver.Ver)
	if err != nil {
		return err
	}
	if fullLoad {
		logutil.BgLogger().Info("full load and reset schema validator")
		do.SchemaValidator.Reset()
	}
	do.SchemaValidator.Update(ver.Ver, schemaVersion, neededSchemaVersion, changedTableIDs)
	lease := do.DDL().GetLease()
	sub := time.Since(startTime)
	if sub > (lease/2) && lease > 0 {
		logutil.BgLogger().Warn("loading schema takes a long time", zap.Duration("take time", sub))
	}
	return nil
}
func (do *Domain) loadSchemaInLoop(lease time.Duration) {
	defer do.wg.Done()
	ticker := time.NewTicker(lease / 2)
	defer ticker.Stop()
	defer recoverInDomain("loadSchemaInLoop", true)
	syncer := do.ddl.SchemaSyncer()
	for {
		select {
		case <-ticker.C:
			err := do.Reload()
			if err != nil {
				logutil.BgLogger().Error("reload schema in loop failed", zap.Error(err))
			}
		case _, ok := <-syncer.GlobalVersionCh():
			err := do.Reload()
			if err != nil {
				logutil.BgLogger().Error("reload schema in loop failed", zap.Error(err))
			}
			if !ok {
				logutil.BgLogger().Warn("reload schema in loop, schema syncer need rewatch")
				syncer.WatchGlobalSchemaVer(context.Background())
			}
		case <-syncer.Done():
			logutil.BgLogger().Info("reload schema in loop, schema syncer need restart")
			do.SchemaValidator.Stop()
			err := do.mustRestartSyncer()
			if err != nil {
				logutil.BgLogger().Error("reload schema in loop, schema syncer restart failed", zap.Error(err))
				break
			}
			exitLoop := do.mustReload()
			if exitLoop {
				logutil.BgLogger().Error("domain is closed, exit loadSchemaInLoop")
				return
			}
			do.SchemaValidator.Restart()
			logutil.BgLogger().Info("schema syncer restarted")
		case <-do.exit:
			return
		}
	}
}
func (do *Domain) mustRestartSyncer() error {
	ctx := context.Background()
	syncer := do.ddl.SchemaSyncer()
	for {
		err := syncer.Restart(ctx)
		if err == nil {
			return nil
		}
		if do.isClose() {
			return err
		}
		time.Sleep(time.Second)
		logutil.BgLogger().Info("restart the schema syncer failed", zap.Error(err))
	}
}
func (do *Domain) mustReload() (exitLoop bool) {
	for {
		err := do.Reload()
		if err == nil {
			logutil.BgLogger().Info("mustReload succeed")
			return false
		}
		logutil.BgLogger().Info("reload the schema failed", zap.Error(err))
		if do.isClose() {
			return true
		}
		time.Sleep(200 * time.Millisecond)
	}
}
func (do *Domain) isClose() bool {
	select {
	case <-do.exit:
		logutil.BgLogger().Info("domain is closed")
		return true
	default:
	}
	return false
}
func (do *Domain) Close() {
	startTime := time.Now()
	if do.ddl != nil {
		terror.Log(do.ddl.Stop())
	}
	close(do.exit)
	if do.etcdClient != nil {
		terror.Log(errors.Trace(do.etcdClient.Close()))
	}
	do.sysSessionPool.Close()
	do.wg.Wait()
	logutil.BgLogger().Info("domain closed", zap.Duration("take time", time.Since(startTime)))
}
type ddlCallback struct {
	ddl.BaseCallback
	do *Domain
}
func (c *ddlCallback) OnChanged(err error) error {
	if err != nil {
		return err
	}
	logutil.BgLogger().Info("performing DDL change, must reload")
	err = c.do.Reload()
	if err != nil {
		logutil.BgLogger().Error("performing DDL change failed", zap.Error(err))
	}
	return nil
}
const resourceIdleTimeout = 3 * time.Minute  
func NewDomain(store kv.Storage, ddlLease time.Duration, statsLease time.Duration, factory pools.Factory) *Domain {
	capacity := 200  
	return &Domain{
		store:           store,
		SchemaValidator: NewSchemaValidator(ddlLease),
		exit:            make(chan struct{}),
		sysSessionPool:  newSessionPool(capacity, factory),
		statsLease:      statsLease,
		infoHandle:      infoschema.NewHandle(store),
	}
}
func (do *Domain) Init(ddlLease time.Duration, sysFactory func(*Domain) (pools.Resource, error)) error {
	if ebd, ok := do.store.(tikv.EtcdBackend); ok {
		if addrs := ebd.EtcdAddrs(); addrs != nil {
			cli, err := clientv3.New(clientv3.Config{
				Endpoints:        addrs,
				AutoSyncInterval: 30 * time.Second,
				DialTimeout:      5 * time.Second,
				DialOptions: []grpc.DialOption{
					grpc.WithBackoffMaxDelay(time.Second * 3),
					grpc.WithKeepaliveParams(keepalive.ClientParameters{
						Time:                time.Duration(10) * time.Second,
						Timeout:             time.Duration(3) * time.Second,
						PermitWithoutStream: true,
					}),
				},
				TLS: ebd.TLSConfig(),
			})
			if err != nil {
				return errors.Trace(err)
			}
			do.etcdClient = cli
		}
	}
	sysFac := func() (pools.Resource, error) {
		return sysFactory(do)
	}
	sysCtxPool := pools.NewResourcePool(sysFac, 2, 2, resourceIdleTimeout)
	ctx := context.Background()
	callback := &ddlCallback{do: do}
	d := do.ddl
	do.ddl = ddl.NewDDL(
		ctx,
		ddl.WithEtcdClient(do.etcdClient),
		ddl.WithStore(do.store),
		ddl.WithInfoHandle(do.infoHandle),
		ddl.WithHook(callback),
		ddl.WithLease(ddlLease),
		ddl.WithResourcePool(sysCtxPool),
	)
	failpoint.Inject("MockReplaceDDL", func(val failpoint.Value) {
		if val.(bool) {
			if err := do.ddl.Stop(); err != nil {
				logutil.BgLogger().Error("stop DDL failed", zap.Error(err))
			}
			do.ddl = d
		}
	})
	err := do.ddl.SchemaSyncer().Init(ctx)
	if err != nil {
		return err
	}
	err = do.Reload()
	if err != nil {
		return err
	}
	if ddlLease > 0 {
		do.wg.Add(1)
		go do.loadSchemaInLoop(ddlLease)
	}
	return nil
}
type sessionPool struct {
	resources chan pools.Resource
	factory   pools.Factory
	mu        struct {
		sync.RWMutex
		closed bool
	}
}
func newSessionPool(cap int, factory pools.Factory) *sessionPool {
	return &sessionPool{
		resources: make(chan pools.Resource, cap),
		factory:   factory,
	}
}
func (p *sessionPool) Get() (resource pools.Resource, err error) {
	var ok bool
	select {
	case resource, ok = <-p.resources:
		if !ok {
			err = errors.New("session pool closed")
		}
	default:
		resource, err = p.factory()
	}
	return
}
func (p *sessionPool) Put(resource pools.Resource) {
	p.mu.RLock()
	defer p.mu.RUnlock()
	if p.mu.closed {
		resource.Close()
		return
	}
	select {
	case p.resources <- resource:
	default:
		resource.Close()
	}
}
func (p *sessionPool) Close() {
	p.mu.Lock()
	if p.mu.closed {
		p.mu.Unlock()
		return
	}
	p.mu.closed = true
	close(p.resources)
	p.mu.Unlock()
	for r := range p.resources {
		r.Close()
	}
}
func (do *Domain) SysSessionPool() *sessionPool {
	return do.sysSessionPool
}