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

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

@@ -16,10 +16,10 @@ package engine
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"strconv"
 	"strings"
+	"time"
 
 	gms "github.com/dolthub/go-mysql-server"
 	"github.com/dolthub/go-mysql-server/eventscheduler"
@@ -43,7 +43,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 +188,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{})
@@ -206,8 +211,28 @@ 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.Debug = true
+		_, memOnly, _ := sql.SystemVariables.GetGlobal(dsess.DoltStatsMemoryOnly)
+		sc.SetMemOnly(memOnly.(int8) == 1)
+
+		typ, jobI, _ := sql.SystemVariables.GetGlobal(dsess.DoltStatsJobInterval)
+		_, gcI, _ := sql.SystemVariables.GetGlobal(dsess.DoltStatsGCInterval)
+		_, brI, _ := sql.SystemVariables.GetGlobal(dsess.DoltStatsBranchInterval)
+
+		jobInterval, _, _ := typ.GetType().Convert(jobI)
+		gcInterval, _, _ := typ.GetType().Convert(gcI)
+		brInterval, _, _ := typ.GetType().Convert(brI)
+
+		sc.SetTimers(
+			jobInterval.(int64)*int64(time.Millisecond),
+			gcInterval.(int64)*int64(time.Millisecond),
+			brInterval.(int64)*int64(time.Millisecond))
+
+		err := sc.Init(ctx, dbs, false)
+		if err != nil {
+			return nil, err
+		}
 	}
 
 	// Load MySQL Db information

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

@@ -55,6 +55,7 @@ import (
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/cluster"
 	_ "github.com/dolthub/dolt/go/libraries/doltcore/sqle/dfunctions"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/dsess"
+	"github.com/dolthub/dolt/go/libraries/doltcore/sqle/statspro"
 	"github.com/dolthub/dolt/go/libraries/doltcore/sqlserver"
 	"github.com/dolthub/dolt/go/libraries/events"
 	"github.com/dolthub/dolt/go/libraries/utils/config"
@@ -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/go.mod

@@ -56,7 +56,7 @@ require (
 	github.com/cespare/xxhash/v2 v2.2.0
 	github.com/creasty/defaults v1.6.0
 	github.com/dolthub/flatbuffers/v23 v23.3.3-dh.2
-	github.com/dolthub/go-mysql-server v0.19.1-0.20250207201905-b3a4c87c4fdc
+	github.com/dolthub/go-mysql-server v0.19.1-0.20250210190204-a73f126157ef
 	github.com/dolthub/gozstd v0.0.0-20240423170813-23a2903bca63
 	github.com/dolthub/swiss v0.1.0
 	github.com/esote/minmaxheap v1.0.0
@@ -91,7 +91,6 @@ require (
 	golang.org/x/exp v0.0.0-20230522175609-2e198f4a06a1
 	golang.org/x/text v0.21.0
 	gonum.org/v1/plot v0.11.0
-	gopkg.in/errgo.v2 v2.1.0
 	gopkg.in/go-jose/go-jose.v2 v2.6.3
 	gopkg.in/yaml.v3 v3.0.1
 )

go/go.sum

@@ -179,8 +179,8 @@ github.com/dolthub/fslock v0.0.3 h1:iLMpUIvJKMKm92+N1fmHVdxJP5NdyDK5bK7z7Ba2s2U=
 github.com/dolthub/fslock v0.0.3/go.mod h1:QWql+P17oAAMLnL4HGB5tiovtDuAjdDTPbuqx7bYfa0=
 github.com/dolthub/go-icu-regex v0.0.0-20241215010122-db690dd53c90 h1:Sni8jrP0sy/w9ZYXoff4g/ixe+7bFCZlfCqXKJSU+zM=
 github.com/dolthub/go-icu-regex v0.0.0-20241215010122-db690dd53c90/go.mod h1:ylU4XjUpsMcvl/BKeRRMXSH7e7WBrPXdSLvnRJYrxEA=
-github.com/dolthub/go-mysql-server v0.19.1-0.20250207201905-b3a4c87c4fdc h1:SdN7GRPtaqmLwfi6cVcyF4Oc8FbFUJ+mwsFRV++6iH4=
-github.com/dolthub/go-mysql-server v0.19.1-0.20250207201905-b3a4c87c4fdc/go.mod h1:QQxZvPHOtycbC2bVmqmT6/Fov2g1/T1Rtm76wLd/Y1E=
+github.com/dolthub/go-mysql-server v0.19.1-0.20250210190204-a73f126157ef h1:vQ5zStRSgdem9R3BtUhkVa5Q8DhSrYs9ReRVFIq86so=
+github.com/dolthub/go-mysql-server v0.19.1-0.20250210190204-a73f126157ef/go.mod h1:QQxZvPHOtycbC2bVmqmT6/Fov2g1/T1Rtm76wLd/Y1E=
 github.com/dolthub/gozstd v0.0.0-20240423170813-23a2903bca63 h1:OAsXLAPL4du6tfbBgK0xXHZkOlos63RdKYS3Sgw/dfI=
 github.com/dolthub/gozstd v0.0.0-20240423170813-23a2903bca63/go.mod h1:lV7lUeuDhH5thVGDCKXbatwKy2KW80L4rMT46n+Y2/Q=
 github.com/dolthub/ishell v0.0.0-20240701202509-2b217167d718 h1:lT7hE5k+0nkBdj/1UOSFwjWpNxf+LCApbRHgnCA17XE=
@@ -1153,7 +1153,6 @@ gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b/go.mod h1:Co6ibVJAznAaIkqp8
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
 gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
 gopkg.in/cheggaaa/pb.v1 v1.0.25/go.mod h1:V/YB90LKu/1FcN3WVnfiiE5oMCibMjukxqG/qStrOgw=
-gopkg.in/errgo.v2 v2.1.0 h1:0vLT13EuvQ0hNvakwLuFZ/jYrLp5F3kcWHXdRggjCE8=
 gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI=
 gopkg.in/fsnotify.v1 v1.4.7/go.mod h1:Tz8NjZHkW78fSQdbUxIjBTcgA1z1m8ZHf0WmKUhAMys=
 gopkg.in/gcfg.v1 v1.2.3/go.mod h1:yesOnuUOFQAhST5vPY4nbZsb/huCgGGXlipJsBn0b3o=

go/libraries/doltcore/doltdb/doltdb.go

@@ -2052,7 +2052,7 @@ func (ddb *DoltDB) AddStash(ctx context.Context, head *Commit, stash RootValue,
 	return err
 }
 
-func (ddb *DoltDB) SetStatisics(ctx context.Context, branch string, addr hash.Hash) error {
+func (ddb *DoltDB) SetStatistics(ctx context.Context, branch string, addr hash.Hash) error {
 	statsDs, err := ddb.db.GetDataset(ctx, ref.NewStatsRef(branch).String())
 	if err != nil {
 		return err

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

@@ -970,7 +970,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)
@@ -1511,8 +1511,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,15 @@ 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)},
+	{Name: "dolt_stats_timers", Schema: statsFuncSchema, Function: statsFunc(statsTimers)},
 }
 
 // stringSchema returns a non-nullable schema with all columns as LONGTEXT.