*: implement auto analyze

ddl/ddl.go

@@ -33,6 +33,7 @@ import (
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/model"
 	"github.com/pingcap/tidb/mysql"
+	"github.com/pingcap/tidb/owner"
 	"github.com/pingcap/tidb/sessionctx/binloginfo"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/terror"
@@ -43,6 +44,9 @@ import (
 const (
 	// currentVersion is for all new DDL jobs.
 	currentVersion = 1
+	// DDLOwnerKey is the ddl owner path that is saved to etcd, and it's exported for testing.
+	DDLOwnerKey = "/tidb/ddl/fg/owner"
+	ddlPrompt   = "ddl"
 )
 
 var (
@@ -157,7 +161,7 @@ type DDL interface {
 	// SchemaSyncer gets the schema syncer.
 	SchemaSyncer() SchemaSyncer
 	// OwnerManager gets the owner manager, and it's used for testing.
-	OwnerManager() OwnerManager
+	OwnerManager() owner.Manager
 	// WorkerVars gets the session variables for DDL worker.
 	WorkerVars() *variable.SessionVars
 	// SetHook sets the hook. It's exported for testing.
@@ -195,7 +199,7 @@ type ddl struct {
 	hook         Callback
 	hookMu       sync.RWMutex
 	store        kv.Storage
-	ownerManager OwnerManager
+	ownerManager owner.Manager
 	schemaSyncer SchemaSyncer
 	// lease is schema seconds.
 	lease        time.Duration
@@ -263,15 +267,15 @@ func newDDL(ctx goctx.Context, etcdCli *clientv3.Client, store kv.Storage,
 
 	id := uuid.NewV4().String()
 	ctx, cancelFunc := goctx.WithCancel(ctx)
-	var manager OwnerManager
+	var manager owner.Manager
 	var syncer SchemaSyncer
 	if etcdCli == nil {
 		// The etcdCli is nil if the store is localstore which is only used for testing.
 		// So we use mockOwnerManager and mockSchemaSyncer.
-		manager = NewMockOwnerManager(id, cancelFunc)
+		manager = owner.NewMockManager(id, cancelFunc)
 		syncer = NewMockSchemaSyncer()
 	} else {
-		manager = NewOwnerManager(etcdCli, id, cancelFunc)
+		manager = owner.NewOwnerManager(etcdCli, ddlPrompt, id, DDLOwnerKey, cancelFunc)
 		syncer = NewSchemaSyncer(etcdCli, id)
 	}
 	d := &ddl{
@@ -400,7 +404,7 @@ func (d *ddl) SchemaSyncer() SchemaSyncer {
 }
 
 // OwnerManager implements DDL.OwnerManager interface.
-func (d *ddl) OwnerManager() OwnerManager {
+func (d *ddl) OwnerManager() owner.Manager {
 	return d.ownerManager
 }
 

ddl/mock.go

@@ -23,67 +23,8 @@ import (
 	goctx "golang.org/x/net/context"
 )
 
-var _ OwnerManager = &mockOwnerManager{}
 var _ SchemaSyncer = &mockSchemaSyncer{}
 
-// mockOwnerManager represents the structure which is used for electing owner.
-// It's used for local store and testing.
-// So this worker will always be the ddl owner and background owner.
-type mockOwnerManager struct {
-	ddlOwner int32
-	ddlID    string // id is the ID of DDL.
-	cancel   goctx.CancelFunc
-}
-
-// NewMockOwnerManager creates a new mock OwnerManager.
-func NewMockOwnerManager(id string, cancel goctx.CancelFunc) OwnerManager {
-	return &mockOwnerManager{
-		ddlID:  id,
-		cancel: cancel,
-	}
-}
-
-// ID implements mockOwnerManager.ID interface.
-func (m *mockOwnerManager) ID() string {
-	return m.ddlID
-}
-
-// IsOwner implements mockOwnerManager.IsOwner interface.
-func (m *mockOwnerManager) IsOwner() bool {
-	return atomic.LoadInt32(&m.ddlOwner) == 1
-}
-
-// SetOwner implements mockOwnerManager.SetOwner interface.
-func (m *mockOwnerManager) SetOwner(isOwner bool) {
-	if isOwner {
-		atomic.StoreInt32(&m.ddlOwner, 1)
-	} else {
-		atomic.StoreInt32(&m.ddlOwner, 0)
-	}
-}
-
-// Cancel implements mockOwnerManager.Cancel interface.
-func (m *mockOwnerManager) Cancel() {
-	m.cancel()
-}
-
-// GetOwnerID implements OwnerManager.GetOwnerID interface.
-func (m *mockOwnerManager) GetOwnerID(ctx goctx.Context, key string) (string, error) {
-	if key != DDLOwnerKey {
-		return "", errors.New("invalid owner key")
-	}
-	if m.IsOwner() {
-		return m.ID(), nil
-	}
-	return "", errors.New("no owner")
-}
-
-// CampaignOwner implements mockOwnerManager.CampaignOwner interface.
-func (m *mockOwnerManager) CampaignOwner(_ goctx.Context) error {
-	m.SetOwner(true)
-	return nil
-}
-
 const mockCheckVersInterval = 2 * time.Millisecond
 
 type mockSchemaSyncer struct {

ddl/syncer.go

@@ -23,6 +23,7 @@ import (
 	"github.com/coreos/etcd/clientv3/concurrency"
 	"github.com/juju/errors"
 	"github.com/ngaut/log"
+	"github.com/pingcap/tidb/owner"
 	goctx "golang.org/x/net/context"
 )
 
@@ -120,8 +121,8 @@ func (s *schemaVersionSyncer) Init(ctx goctx.Context) error {
 	if err != nil {
 		return errors.Trace(err)
 	}
-	s.session, err = newSession(ctx, s.selfSchemaVerPath, s.etcdCli,
-		newSessionDefaultRetryCnt, SyncerSessionTTL)
+	logPrefix := fmt.Sprintf("[%s] %s", ddlPrompt, s.selfSchemaVerPath)
+	s.session, err = owner.NewSession(ctx, logPrefix, s.etcdCli, owner.NewSessionDefaultRetryCnt, SyncerSessionTTL)
 	if err != nil {
 		return errors.Trace(err)
 	}
@@ -137,8 +138,8 @@ func (s *schemaVersionSyncer) Done() <-chan struct{} {
 
 // Restart implements SchemaSyncer.Restart interface.
 func (s *schemaVersionSyncer) Restart(ctx goctx.Context) error {
-	session, err := newSession(ctx, s.selfSchemaVerPath, s.etcdCli,
-		newSessionRetryUnlimited, SyncerSessionTTL)
+	logPrefix := fmt.Sprintf("[%s] %s", ddlPrompt, s.selfSchemaVerPath)
+	session, err := owner.NewSession(ctx, logPrefix, s.etcdCli, owner.NewSessionRetryUnlimited, SyncerSessionTTL)
 	if err != nil {
 		return errors.Trace(err)
 	}

domain/domain.go

@@ -29,6 +29,7 @@ import (
 	"github.com/pingcap/tidb/kv"
 	"github.com/pingcap/tidb/meta"
 	"github.com/pingcap/tidb/model"
+	"github.com/pingcap/tidb/owner"
 	"github.com/pingcap/tidb/perfschema"
 	"github.com/pingcap/tidb/privilege/privileges"
 	"github.com/pingcap/tidb/sessionctx/variable"
@@ -531,8 +532,9 @@ func (do *Domain) CreateStatsHandle(ctx context.Context) {
 	atomic.StorePointer(&do.statsHandle, unsafe.Pointer(statistics.NewHandle(ctx, do.statsLease)))
 }
 
-// UpdateTableStatsLoop creates a goroutine loads stats info and updates stats info in a loop. It
-// should be called only once in BootstrapSession.
+// UpdateTableStatsLoop creates a goroutine loads stats info and updates stats info in a loop.
+// It will also start a goroutine to analyze tables automatically.
+// It should be called only once in BootstrapSession.
 func (do *Domain) UpdateTableStatsLoop(ctx context.Context) error {
 	ctx.GetSessionVars().InRestrictedSQL = true
 	statsHandle := statistics.NewHandle(ctx, do.statsLease)
@@ -546,41 +548,70 @@ func (do *Domain) UpdateTableStatsLoop(ctx context.Context) error {
 	if lease <= 0 {
 		return nil
 	}
-	deltaUpdateDuration := lease * 5
-	go func(do *Domain) {
-		loadTicker := time.NewTicker(lease)
-		defer loadTicker.Stop()
-		deltaUpdateTicker := time.NewTicker(deltaUpdateDuration)
-		defer deltaUpdateTicker.Stop()
+	go do.updateStatsWorker(ctx, lease)
+	go do.autoAnalyzeWorker(lease)
+	return nil
+}
 
-		for {
-			select {
-			case <-loadTicker.C:
-				err = statsHandle.Update(do.InfoSchema())
-				if err != nil {
-					log.Error(errors.ErrorStack(err))
-				}
-			case <-do.exit:
-				return
+func (do *Domain) updateStatsWorker(ctx context.Context, lease time.Duration) {
+	deltaUpdateDuration := lease * 5
+	loadTicker := time.NewTicker(lease)
+	defer loadTicker.Stop()
+	deltaUpdateTicker := time.NewTicker(deltaUpdateDuration)
+	defer deltaUpdateTicker.Stop()
+	statsHandle := do.StatsHandle()
+	for {
+		select {
+		case <-loadTicker.C:
+			err := statsHandle.Update(do.InfoSchema())
+			if err != nil {
+				log.Error("[stats] update stats info fail: ", errors.ErrorStack(err))
+			}
+		case <-do.exit:
+			return
 			// This channel is sent only by ddl owner or the drop stats executor.
-			case t := <-statsHandle.DDLEventCh():
-				err = statsHandle.HandleDDLEvent(t)
+		case t := <-statsHandle.DDLEventCh():
+			err := statsHandle.HandleDDLEvent(t)
+			if err != nil {
+				log.Error("[stats] handle ddl event fail: ", errors.ErrorStack(err))
+			}
+		case t := <-statsHandle.AnalyzeResultCh():
+			for _, hg := range t.Hist {
+				err := hg.SaveToStorage(ctx, t.TableID, t.Count, t.IsIndex)
 				if err != nil {
-					log.Error(errors.ErrorStack(err))
-				}
-			case t := <-statsHandle.AnalyzeResultCh():
-				for _, hg := range t.Hist {
-					err = hg.SaveToStorage(t.Ctx, t.TableID, t.Count, t.IsIndex)
-					if err != nil {
-						log.Error(errors.ErrorStack(err))
-					}
+					log.Error("[stats] save histogram to storage fail: ", errors.ErrorStack(err))
 				}
-			case <-deltaUpdateTicker.C:
-				statsHandle.DumpStatsDeltaToKV()
 			}
+		case <-deltaUpdateTicker.C:
+			statsHandle.DumpStatsDeltaToKV()
 		}
-	}(do)
-	return nil
+	}
+}
+
+func (do *Domain) autoAnalyzeWorker(lease time.Duration) {
+	id := do.ddl.OwnerManager().ID()
+	cancelCtx, cancelFunc := goctx.WithCancel(goctx.Background())
+	var statsOwner owner.Manager
+	if do.etcdClient == nil {
+		statsOwner = owner.NewMockManager(id, cancelFunc)
+	} else {
+		statsOwner = owner.NewOwnerManager(do.etcdClient, statistics.StatsPrompt, id, statistics.StatsOwnerKey, cancelFunc)
+	}
+	// TODO: Need to do something when err is not nil.
+	err := statsOwner.CampaignOwner(cancelCtx)
+	if err != nil {
+		log.Warnf("[stats] campaign owner fail:", errors.ErrorStack(err))
+	}
+	statsHandle := do.StatsHandle()
+	for {
+		if statsOwner.IsOwner() {
+			err := statsHandle.HandleAutoAnalyze(do.InfoSchema())
+			if err != nil {
+				log.Error("[stats] auto analyze fail:", errors.ErrorStack(err))
+			}
+		}
+		time.Sleep(lease)
+	}
 }
 
 const privilegeKey = "/tidb/privilege"

executor/adapter.go

@@ -207,6 +207,9 @@ func (a *statement) buildExecutor(ctx context.Context) (Executor, error) {
 			}
 		}
 	}
+	if _, ok := a.plan.(*plan.Analyze); ok && ctx.GetSessionVars().InRestrictedSQL {
+		priority = kv.PriorityLow
+	}
 
 	b := newExecutorBuilder(ctx, a.is, priority)
 	e := b.build(a.plan)

executor/analyze.go

@@ -98,7 +98,6 @@ func (e *AnalyzeExec) Next() (Row, error) {
 				log.Error(errors.ErrorStack(err))
 				continue
 			}
-			result.Ctx = e.ctx
 			dom.StatsHandle().AnalyzeResultCh() <- &result
 		}
 		// We sleep two lease to make sure other tidb node has updated this node.

executor/builder.go

@@ -21,7 +21,6 @@ import (
 	"github.com/juju/errors"
 	"github.com/pingcap/tidb/ast"
 	"github.com/pingcap/tidb/context"
-	"github.com/pingcap/tidb/ddl"
 	"github.com/pingcap/tidb/distsql"
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/infoschema"
@@ -153,7 +152,7 @@ func (b *executorBuilder) buildShowDDL(v *plan.ShowDDL) Executor {
 	var err error
 	ownerManager := sessionctx.GetDomain(e.ctx).DDL().OwnerManager()
 	ctx, cancel := goctx.WithTimeout(goctx.Background(), 3*time.Second)
-	e.ddlOwnerID, err = ownerManager.GetOwnerID(ctx, ddl.DDLOwnerKey)
+	e.ddlOwnerID, err = ownerManager.GetOwnerID(ctx)
 	cancel()
 	if err != nil {
 		b.err = errors.Trace(err)