[stats] Rewrite stat management to use single threaded event loop

go/cmd/dolt/commands/engine/sqlengine.go

@@ -16,11 +16,6 @@ package engine
 
 import (
 	"context"
-	"fmt"
-	"os"
-	"strconv"
-	"strings"
-
 	gms "github.com/dolthub/go-mysql-server"
 	"github.com/dolthub/go-mysql-server/eventscheduler"
 	"github.com/dolthub/go-mysql-server/sql"
@@ -31,6 +26,9 @@ import (
 	_ "github.com/dolthub/go-mysql-server/sql/variables"
 	"github.com/dolthub/vitess/go/vt/sqlparser"
 	"github.com/sirupsen/logrus"
+	"os"
+	"strconv"
+	"strings"
 
 	"github.com/dolthub/dolt/go/cmd/dolt/cli"
 	"github.com/dolthub/dolt/go/libraries/doltcore/branch_control"
@@ -43,7 +41,6 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/kvexec"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/mysql_file_handler"
-	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/statsnoms"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/statspro"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/writer"
 	"github.com/dolthub/dolt/go/libraries/utils/config"
@@ -189,7 +186,13 @@ func NewSqlEngine(
 		"authentication_dolt_jwt": NewAuthenticateDoltJWTPlugin(config.JwksConfig),
 	})
 
-	statsPro := statspro.NewProvider(pro, statsnoms.NewNomsStatsFactory(mrEnv.RemoteDialProvider()))
+	var statsPro sql.StatsProvider
+	_, enabled, _ := sql.SystemVariables.GetGlobal(dsess.DoltStatsEnabled)
+	if enabled.(int8) == 1 {
+		statsPro = statspro.NewStatsCoord(pro, sqlEngine.NewDefaultContext, logrus.StandardLogger(), bThreads, mrEnv.GetEnv(mrEnv.GetFirstDatabase()))
+	} else {
+		statsPro = statspro.StatsNoop{}
+	}
 	engine.Analyzer.Catalog.StatsProvider = statsPro
 
 	engine.Analyzer.ExecBuilder = rowexec.NewOverrideBuilder(kvexec.Builder{})
@@ -202,8 +205,8 @@ func NewSqlEngine(
 
 	// configuring stats depends on sessionBuilder
 	// sessionBuilder needs ref to statsProv
-	if err = statsPro.Configure(ctx, sqlEngine.NewDefaultContext, bThreads, dbs); err != nil {
-		fmt.Fprintln(cli.CliErr, err)
+	if sc, ok := statsPro.(*statspro.StatsCoord); ok {
+		sc.Init(ctx, dbs)
 	}
 
 	// Load MySQL Db information

go/cmd/dolt/commands/sqlserver/server.go

@@ -19,6 +19,7 @@ import (
 	"crypto/tls"
 	"errors"
 	"fmt"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/statspro"
 	"net"
 	"net/http"
 	"os"
@@ -260,23 +261,23 @@ func ConfigureServices(
 	var sqlEngine *engine.SqlEngine
 	InitSqlEngine := &svcs.AnonService{
 		InitF: func(ctx context.Context) (err error) {
-			if statsOn, err := mrEnv.Config().GetString(env.SqlServerGlobalsPrefix + "." + dsess.DoltStatsAutoRefreshEnabled); err != nil {
-				// Auto-stats is off by default for every command except
-				// sql-server. Unless the config specifies a specific
-				// behavior, enable server stats collection.
-				sql.SystemVariables.SetGlobal(dsess.DoltStatsAutoRefreshEnabled, 1)
-			} else if statsOn != "0" {
-				// do not bootstrap if auto-stats enabled
-			} else if _, err := mrEnv.Config().GetString(env.SqlServerGlobalsPrefix + "." + dsess.DoltStatsBootstrapEnabled); err != nil {
-				// If we've disabled stats collection and config does not
-				// specify bootstrap behavior, enable bootstrapping.
-				sql.SystemVariables.SetGlobal(dsess.DoltStatsBootstrapEnabled, 1)
-			}
 			sqlEngine, err = engine.NewSqlEngine(
 				ctx,
 				mrEnv,
 				config,
 			)
+			if sc, ok := sqlEngine.GetUnderlyingEngine().Analyzer.Catalog.StatsProvider.(*statspro.StatsCoord); ok {
+				sqlCtx, err := sqlEngine.NewDefaultContext(ctx)
+				if err != nil {
+					return err
+				}
+				if sc == nil {
+					return fmt.Errorf("unexpected nil stats coord")
+				}
+				if err = sc.Restart(sqlCtx); err != nil {
+					return err
+				}
+			}
 			return err
 		},
 		StopF: func() error {

go/libraries/doltcore/schema/statistic.go

@@ -24,12 +24,12 @@ import (
 const StatsVersion int64 = 1
 
 const (
-	StatsQualifierColName     = "qualifier"
 	StatsDbColName            = "database_name"
 	StatsTableColName         = "table_name"
 	StatsIndexColName         = "index_name"
-	StatsPositionColName      = "position"
+	StatsBranchName           = "branch"
 	StatsCommitHashColName    = "commit_hash"
+	StatsPrefixLenName        = "prefix_len"
 	StatsRowCountColName      = "row_count"
 	StatsDistinctCountColName = "distinct_count"
 	StatsNullCountColName     = "null_count"
@@ -42,7 +42,7 @@ const (
 	StatsMcv2ColName          = "mcv2"
 	StatsMcv3ColName          = "mcv3"
 	StatsMcv4ColName          = "mcv4"
-	StatsMcvCountsColName     = "mcvCounts"
+	StatsMcvCountsColName     = "mcv_counts"
 	StatsVersionColName       = "version"
 )
 
@@ -52,6 +52,7 @@ const (
 	StatsIndexTag
 	StatsPositionTag
 	StatsVersionTag
+	StatsPrefixLenTag
 	StatsCommitHashTag
 	StatsRowCountTag
 	StatsDistinctCountTag
@@ -71,9 +72,9 @@ const (
 func StatsTableSqlSchema(dbName string) sql.PrimaryKeySchema {
 	return sql.PrimaryKeySchema{
 		Schema: sql.Schema{
-			&sql.Column{Name: StatsDbColName, Type: types.Text, PrimaryKey: true, DatabaseSource: dbName},
-			&sql.Column{Name: StatsTableColName, Type: types.Text, PrimaryKey: true, DatabaseSource: dbName},
-			&sql.Column{Name: StatsIndexColName, Type: types.Text, PrimaryKey: true, DatabaseSource: dbName},
+			&sql.Column{Name: StatsDbColName, Type: types.Text, DatabaseSource: dbName},
+			&sql.Column{Name: StatsTableColName, Type: types.Text, DatabaseSource: dbName},
+			&sql.Column{Name: StatsIndexColName, Type: types.Text, DatabaseSource: dbName},
 			&sql.Column{Name: StatsRowCountColName, Type: types.Int64, DatabaseSource: dbName},
 			&sql.Column{Name: StatsDistinctCountColName, Type: types.Int64, DatabaseSource: dbName},
 			&sql.Column{Name: StatsNullCountColName, Type: types.Int64, DatabaseSource: dbName},
@@ -88,28 +89,21 @@ func StatsTableSqlSchema(dbName string) sql.PrimaryKeySchema {
 			&sql.Column{Name: StatsMcv4ColName, Type: types.Text, DatabaseSource: dbName},
 			&sql.Column{Name: StatsMcvCountsColName, Type: types.Text, DatabaseSource: dbName},
 		},
-		PkOrdinals: []int{0, 1},
 	}
 }
 
 var StatsTableDoltSchema = StatsTableDoltSchemaGen()
 
 func StatsTableDoltSchemaGen() Schema {
 	colColl := NewColCollection(
-		NewColumn(StatsDbColName, StatsDbTag, stypes.StringKind, true, NotNullConstraint{}),
-		NewColumn(StatsTableColName, StatsTableTag, stypes.StringKind, true, NotNullConstraint{}),
-		NewColumn(StatsIndexColName, StatsIndexTag, stypes.StringKind, true, NotNullConstraint{}),
-		NewColumn(StatsPositionColName, StatsPositionTag, stypes.IntKind, true, NotNullConstraint{}),
+		NewColumn(StatsPrefixLenName, StatsPrefixLenTag, stypes.IntKind, true, NotNullConstraint{}),
+		NewColumn(StatsCommitHashColName, StatsCommitHashTag, stypes.StringKind, true, NotNullConstraint{}),
 		NewColumn(StatsVersionColName, StatsVersionTag, stypes.IntKind, false, NotNullConstraint{}),
-		NewColumn(StatsCommitHashColName, StatsCommitHashTag, stypes.StringKind, false, NotNullConstraint{}),
 		NewColumn(StatsRowCountColName, StatsRowCountTag, stypes.IntKind, false, NotNullConstraint{}),
 		NewColumn(StatsDistinctCountColName, StatsDistinctCountTag, stypes.IntKind, false, NotNullConstraint{}),
 		NewColumn(StatsNullCountColName, StatsNullCountTag, stypes.IntKind, false, NotNullConstraint{}),
-		NewColumn(StatsColumnsColName, StatsColumnsTag, stypes.StringKind, false, NotNullConstraint{}),
-		NewColumn(StatsTypesColName, StatsTypesTag, stypes.StringKind, false, NotNullConstraint{}),
 		NewColumn(StatsUpperBoundColName, StatsUpperBoundTag, stypes.StringKind, false, NotNullConstraint{}),
 		NewColumn(StatsUpperBoundCntColName, StatsUpperBoundCntTag, stypes.IntKind, false, NotNullConstraint{}),
-		NewColumn(StatsCreatedAtColName, StatsCreatedAtTag, stypes.TimestampKind, false, NotNullConstraint{}),
 		NewColumn(StatsMcv1ColName, StatsMcv1Tag, stypes.StringKind, false),
 		NewColumn(StatsMcv2ColName, StatsMcv2Tag, stypes.StringKind, false),
 		NewColumn(StatsMcv3ColName, StatsMcv3Tag, stypes.StringKind, false),

go/libraries/doltcore/sqle/clusterdb/database.go

@@ -162,6 +162,10 @@ func (db database) RequestedName() string {
 	return db.Name()
 }
 
+func (db database) AliasedName() string {
+	return db.Name()
+}
+
 type noopRepoStateWriter struct{}
 
 var _ env.RepoStateWriter = noopRepoStateWriter{}

go/libraries/doltcore/sqle/database.go

@@ -694,6 +694,9 @@ func (db Database) getTableInsensitive(ctx *sql.Context, head *doltdb.Commit, ds
 		if err != nil {
 			return nil, false, err
 		}
+		if branch == "" {
+			branch = db.Revision()
+		}
 		dt, found = dtables.NewStatisticsTable(ctx, db.Name(), db.schemaName, branch, tables), true
 	case doltdb.ProceduresTableName:
 		found = true

go/libraries/doltcore/sqle/database_provider.go

@@ -966,7 +966,7 @@ func (p *DoltDatabaseProvider) databaseForRevision(ctx *sql.Context, revisionQua
 			}
 		}
 
-		db, err := revisionDbForBranch(ctx, srcDb, resolvedRevSpec, requestedName)
+		db, err := RevisionDbForBranch(ctx, srcDb, resolvedRevSpec, requestedName)
 		// preserve original user case in the case of not found
 		if sql.ErrDatabaseNotFound.Is(err) {
 			return nil, false, sql.ErrDatabaseNotFound.New(revisionQualifiedName)
@@ -1507,8 +1507,8 @@ func isTag(ctx context.Context, db dsess.SqlDatabase, tagName string) (string, b
 	return "", false, nil
 }
 
-// revisionDbForBranch returns a new database that is tied to the branch named by revSpec
-func revisionDbForBranch(ctx context.Context, srcDb dsess.SqlDatabase, revSpec string, requestedName string) (dsess.SqlDatabase, error) {
+// RevisionDbForBranch returns a new database that is tied to the branch named by revSpec
+func RevisionDbForBranch(ctx context.Context, srcDb dsess.SqlDatabase, revSpec string, requestedName string) (dsess.SqlDatabase, error) {
 	static := staticRepoState{
 		branch:          ref.NewBranchRef(revSpec),
 		RepoStateWriter: srcDb.DbData().Rsw,

go/libraries/doltcore/sqle/dprocedures/init.go

@@ -47,12 +47,14 @@ var DoltProcedures = []sql.ExternalStoredProcedureDetails{
 	{Name: "dolt_tag", Schema: int64Schema("status"), Function: doltTag},
 	{Name: "dolt_verify_constraints", Schema: int64Schema("violations"), Function: doltVerifyConstraints},
 
-	{Name: "dolt_stats_drop", Schema: statsFuncSchema, Function: statsFunc(statsDrop)},
 	{Name: "dolt_stats_restart", Schema: statsFuncSchema, Function: statsFunc(statsRestart)},
 	{Name: "dolt_stats_stop", Schema: statsFuncSchema, Function: statsFunc(statsStop)},
-	{Name: "dolt_stats_status", Schema: statsFuncSchema, Function: statsFunc(statsStatus)},
-	{Name: "dolt_stats_prune", Schema: statsFuncSchema, Function: statsFunc(statsPrune)},
+	{Name: "dolt_stats_info", Schema: statsFuncSchema, Function: statsFunc(statsInfo)},
 	{Name: "dolt_stats_purge", Schema: statsFuncSchema, Function: statsFunc(statsPurge)},
+	{Name: "dolt_stats_wait", Schema: statsFuncSchema, Function: statsFunc(statsWait)},
+	{Name: "dolt_stats_gc", Schema: statsFuncSchema, Function: statsFunc(statsGc)},
+	{Name: "dolt_stats_sync", Schema: statsFuncSchema, Function: statsFunc(statsBranchSync)},
+	{Name: "dolt_stats_validate", Schema: statsFuncSchema, Function: statsFunc(statsValidate)},
 }
 
 // stringSchema returns a non-nullable schema with all columns as LONGTEXT.