statistics: speed up the backgroud stats update worker (#58490)

ref #57868

Author: winoros

br/pkg/restore/snap_client/pipeline_items.go

@@ -297,7 +297,7 @@ func (rc *SnapClient) GoUpdateMetaAndLoadStats(
 			log.Info("start update metas", zap.Stringer("table", oldTable.Info.Name), zap.Stringer("db", oldTable.DB.Name))
 			// the total kvs contains the index kvs, but the stats meta needs the count of rows
 			count := int64(oldTable.TotalKvs / uint64(len(oldTable.Info.Indices)+1))
-			if statsErr = statsHandler.SaveMetaToStorage(tbl.Table.ID, count, 0, "br restore"); statsErr != nil {
+			if statsErr = statsHandler.SaveMetaToStorage(tbl.Table.ID, count, 0, "br restore", false); statsErr != nil {
 				log.Error("update stats meta failed", zap.Any("table", tbl.Table), zap.Error(statsErr))
 			}
 		}

br/pkg/restore/snap_client/systable_restore_test.go

@@ -117,5 +117,5 @@ func TestCheckSysTableCompatibility(t *testing.T) {
 
 // The above variables are in the file br/pkg/restore/systable_restore.go
 func TestMonitorTheSystemTableIncremental(t *testing.T) {
-	require.Equal(t, int64(246), session.CurrentBootstrapVersion)
+	require.Equal(t, int64(247), session.CurrentBootstrapVersion)
 }

pkg/session/bootstrap.go

@@ -222,11 +222,12 @@ const (
 
 	// CreateStatsMetaTable stores the meta of table statistics.
 	CreateStatsMetaTable = `CREATE TABLE IF NOT EXISTS mysql.stats_meta (
-		version 		BIGINT(64) UNSIGNED NOT NULL,
-		table_id 		BIGINT(64) NOT NULL,
-		modify_count	BIGINT(64) NOT NULL DEFAULT 0,
-		count 			BIGINT(64) UNSIGNED NOT NULL DEFAULT 0,
-		snapshot        BIGINT(64) UNSIGNED NOT NULL DEFAULT 0,
+		version 					BIGINT(64) UNSIGNED NOT NULL,
+		table_id 					BIGINT(64) NOT NULL,
+		modify_count				BIGINT(64) NOT NULL DEFAULT 0,
+		count 						BIGINT(64) UNSIGNED NOT NULL DEFAULT 0,
+		snapshot        			BIGINT(64) UNSIGNED NOT NULL DEFAULT 0,
+		last_stats_histograms_version 	BIGINT(64) UNSIGNED DEFAULT NULL,
 		INDEX idx_ver(version),
 		UNIQUE INDEX tbl(table_id)
 	);`
@@ -1271,19 +1272,23 @@ const (
 	// Add extra_params to tidb_global_task and tidb_global_task_history.
 	version243 = 243
 
-	// version242 add Max_user_connections into mysql.user.
+	// version244 add Max_user_connections into mysql.user.
 	version244 = 244
 
 	// version245 updates column types of mysql.bind_info.
 	version245 = 245
 
 	// version246 adds new unique index for mysql.bind_info.
 	version246 = 246
+
+	// version 247
+	// Add last_stats_histograms_version to mysql.stats_meta.
+	version247 = 247
 )
 
 // currentBootstrapVersion is defined as a variable, so we can modify its value for testing.
 // please make sure this is the largest version
-var currentBootstrapVersion int64 = version246
+var currentBootstrapVersion int64 = version247
 
 // DDL owner key's expired time is ManagerSessionTTL seconds, we should wait the time and give more time to have a chance to finish it.
 var internalSQLTimeout = owner.ManagerSessionTTL + 15
@@ -1465,6 +1470,7 @@ var (
 		upgradeToVer244,
 		upgradeToVer245,
 		upgradeToVer246,
+		upgradeToVer247,
 	}
 )
 
@@ -3406,7 +3412,6 @@ func upgradeToVer245(s sessiontypes.Session, ver int64) {
 	if ver >= version245 {
 		return
 	}
-
 	doReentrantDDL(s, "ALTER TABLE mysql.bind_info MODIFY COLUMN original_sql LONGTEXT NOT NULL")
 	doReentrantDDL(s, "ALTER TABLE mysql.bind_info MODIFY COLUMN bind_sql LONGTEXT NOT NULL")
 }
@@ -3465,6 +3470,13 @@ func upgradeToVer246(s sessiontypes.Session, ver int64) {
 	doReentrantDDL(s, "ALTER TABLE mysql.bind_info ADD UNIQUE INDEX digest_index(plan_digest, sql_digest)", dbterror.ErrDupKeyName)
 }
 
+func upgradeToVer247(s sessiontypes.Session, ver int64) {
+	if ver >= version247 {
+		return
+	}
+	doReentrantDDL(s, "ALTER TABLE mysql.stats_meta ADD COLUMN last_stats_histograms_version bigint unsigned DEFAULT NULL", infoschema.ErrColumnExists)
+}
+
 // initGlobalVariableIfNotExists initialize a global variable with specific val if it does not exist.
 func initGlobalVariableIfNotExists(s sessiontypes.Session, name string, val any) {
 	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnBootstrap)

pkg/statistics/handle/bootstrap.go

@@ -68,21 +68,26 @@ func (*Handle) initStatsMeta4Chunk(cache statstypes.StatsCache, iter *chunk.Iter
 		physicalID = row.GetInt64(1)
 		maxPhysicalID = max(physicalID, maxPhysicalID)
 		newHistColl := *statistics.NewHistColl(physicalID, row.GetInt64(3), row.GetInt64(2), 4, 4)
+		// During the initialization phase, we need to initialize LastAnalyzeVersion with the snapshot,
+		// which ensures that we don't duplicate the auto-analyze of a particular type of table.
+		// When the predicate columns feature is turned on, if a table has neither predicate columns nor indexes,
+		// then auto-analyze will only analyze the _row_id and refresh stats_meta,
+		// but since we don't have any histograms or topn's created for _row_id at the moment.
+		// So if we don't initialize LastAnalyzeVersion with the snapshot here,
+		// it will stay at 0 and auto-analyze won't be able to detect that the table has been analyzed.
+		// But in the future, we maybe will create some records for _row_id, see:
+		// https://github.com/pingcap/tidb/issues/51098
 		snapshot := row.GetUint64(4)
+		lastAnalyzeVersion, lastStatsHistUpdateVersion := snapshot, snapshot
+		if !row.IsNull(5) {
+			lastStatsHistUpdateVersion = max(lastStatsHistUpdateVersion, row.GetUint64(5))
+		}
 		tbl := &statistics.Table{
 			HistColl:              newHistColl,
 			Version:               row.GetUint64(0),
 			ColAndIdxExistenceMap: statistics.NewColAndIndexExistenceMapWithoutSize(),
-			// During the initialization phase, we need to initialize LastAnalyzeVersion with the snapshot,
-			// which ensures that we don't duplicate the auto-analyze of a particular type of table.
-			// When the predicate columns feature is turned on, if a table has neither predicate columns nor indexes,
-			// then auto-analyze will only analyze the _row_id and refresh stats_meta,
-			// but since we don't have any histograms or topn's created for _row_id at the moment.
-			// So if we don't initialize LastAnalyzeVersion with the snapshot here,
-			// it will stay at 0 and auto-analyze won't be able to detect that the table has been analyzed.
-			// But in the future, we maybe will create some records for _row_id, see:
-			// https://github.com/pingcap/tidb/issues/51098
-			LastAnalyzeVersion: snapshot,
+			LastAnalyzeVersion:    lastAnalyzeVersion,
+			LastStatsHistVersion:  lastStatsHistUpdateVersion,
 		}
 		cache.Put(physicalID, tbl) // put this table again since it is updated
 	}
@@ -95,7 +100,7 @@ func (*Handle) initStatsMeta4Chunk(cache statstypes.StatsCache, iter *chunk.Iter
 
 func (h *Handle) initStatsMeta(ctx context.Context) (statstypes.StatsCache, error) {
 	ctx = kv.WithInternalSourceType(ctx, kv.InternalTxnStats)
-	sql := "select HIGH_PRIORITY version, table_id, modify_count, count, snapshot from mysql.stats_meta"
+	sql := "select HIGH_PRIORITY version, table_id, modify_count, count, snapshot, last_stats_histograms_version from mysql.stats_meta"
 	rc, err := util.Exec(h.initStatsCtx, sql)
 	if err != nil {
 		return nil, errors.Trace(err)

pkg/statistics/handle/cache/statscache.go

@@ -133,7 +133,7 @@ func (s *StatsCacheImpl) Update(ctx context.Context, is infoschema.InfoSchema, t
 		err                       error
 	)
 	if err := util.CallWithSCtx(s.statsHandle.SPool(), func(sctx sessionctx.Context) error {
-		query := "SELECT version, table_id, modify_count, count, snapshot from mysql.stats_meta where version > %? "
+		query := "SELECT version, table_id, modify_count, count, snapshot, last_stats_histograms_version from mysql.stats_meta where version > %? "
 		args := []any{lastVersion}
 
 		if onlyForAnalyzedTables {
@@ -174,6 +174,10 @@ func (s *StatsCacheImpl) Update(ctx context.Context, is infoschema.InfoSchema, t
 		modifyCount := row.GetInt64(2)
 		count := row.GetInt64(3)
 		snapshot := row.GetUint64(4)
+		var latestHistUpdateVersion uint64
+		if !row.IsNull(5) {
+			latestHistUpdateVersion = row.GetUint64(5)
+		}
 
 		// Detect the context cancel signal, since it may take a long time for the loop.
 		// TODO: add context to TableInfoByID and remove this code block?
@@ -192,31 +196,44 @@ func (s *StatsCacheImpl) Update(ctx context.Context, is infoschema.InfoSchema, t
 		}
 		tableInfo := table.Meta()
 		// If the table is not updated, we can skip it.
-		if oldTbl, ok := s.Get(physicalID); ok &&
-			oldTbl.Version >= version &&
+
+		oldTbl, ok := s.Get(physicalID)
+		if ok && oldTbl.Version >= version &&
 			tableInfo.UpdateTS == oldTbl.TblInfoUpdateTS {
 			continue
 		}
-		tbl, err := s.statsHandle.TableStatsFromStorage(
-			tableInfo,
-			physicalID,
-			false,
-			0,
-		)
-		// Error is not nil may mean that there are some ddl changes on this table, we will not update it.
-		if err != nil {
-			statslogutil.StatsLogger().Error(
-				"error occurred when read table stats",
-				zap.String("table", tableInfo.Name.O),
-				zap.Error(err),
-			)
-			continue
+		var tbl *statistics.Table
+		needLoadColAndIdxStats := true
+		// If the column/index stats has not been updated, we can reuse the old table stats.
+		// Only need to update the count and modify count.
+		if ok && latestHistUpdateVersion > 0 && oldTbl.LastStatsHistVersion >= latestHistUpdateVersion {
+			tbl = oldTbl.Copy()
+			// count and modify count is updated in finalProcess
+			needLoadColAndIdxStats = false
 		}
-		if tbl == nil {
-			tblToUpdateOrDelete.addToDelete(physicalID)
-			continue
+		if needLoadColAndIdxStats {
+			tbl, err = s.statsHandle.TableStatsFromStorage(
+				tableInfo,
+				physicalID,
+				false,
+				0,
+			)
+			// Error is not nil may mean that there are some ddl changes on this table, we will not update it.
+			if err != nil {
+				statslogutil.StatsLogger().Error(
+					"error occurred when read table stats",
+					zap.String("table", tableInfo.Name.O),
+					zap.Error(err),
+				)
+				continue
+			}
+			if tbl == nil {
+				tblToUpdateOrDelete.addToDelete(physicalID)
+				continue
+			}
 		}
 		tbl.Version = version
+		tbl.LastStatsHistVersion = latestHistUpdateVersion
 		tbl.RealtimeCount = count
 		tbl.ModifyCount = modifyCount
 		tbl.TblInfoUpdateTS = tableInfo.UpdateTS

pkg/statistics/handle/ddl/ddl_test.go

@@ -299,16 +299,28 @@ func TestDDLHistogram(t *testing.T) {
 	<-h.DDLEventCh()
 	testKit.MustExec("insert into t values(1,2),(3,4)")
 	testKit.MustExec("analyze table t")
-
-	testKit.MustExec("alter table t add column c_null int")
-	err := statstestutil.HandleNextDDLEventWithTxn(h)
-	require.NoError(t, err)
 	is := do.InfoSchema()
 	require.Nil(t, h.Update(context.Background(), is))
 	tbl, err := is.TableByName(context.Background(), ast.NewCIStr("test"), ast.NewCIStr("t"))
 	require.NoError(t, err)
 	tableInfo := tbl.Meta()
 	statsTbl := do.StatsHandle().GetTableStats(tableInfo)
+	lastHistUpdateVersion1 := statsTbl.LastStatsHistVersion
+
+	testKit.MustExec("alter table t add column c_null int")
+	err = statstestutil.HandleNextDDLEventWithTxn(h)
+	require.NoError(t, err)
+
+	// Check that the last_stats_histograms_version has been updated.
+	is = do.InfoSchema()
+	tbl, err = is.TableByName(context.Background(), ast.NewCIStr("test"), ast.NewCIStr("t"))
+	require.NoError(t, err)
+	tableInfo = tbl.Meta()
+	require.Nil(t, h.Update(context.Background(), is))
+	statsTbl = do.StatsHandle().GetTableStats(tableInfo)
+	lastHistUpdateVersion2 := statsTbl.LastStatsHistVersion
+	require.Greater(t, lastHistUpdateVersion2, lastHistUpdateVersion1)
+
 	require.True(t, statsTbl.ColAndIdxExistenceMap.HasAnalyzed(2, false))
 	require.False(t, statsTbl.Pseudo)
 	require.True(t, statsTbl.GetCol(tableInfo.Columns[2].ID).IsStatsInitialized())

pkg/statistics/handle/ddl/subscriber.go

@@ -316,7 +316,7 @@ func (h subscriber) delayedDeleteStats4PhysicalID(
 	sctx sessionctx.Context,
 	id int64,
 ) error {
-	startTS, err2 := storage.UpdateStatsMetaVersion(ctx, sctx, id)
+	startTS, err2 := storage.UpdateStatsMetaVerAndLastHistUpdateVer(ctx, sctx, id)
 	if err2 != nil {
 		return errors.Trace(err2)
 	}

pkg/statistics/handle/storage/BUILD.bazel

@@ -59,7 +59,7 @@ go_test(
         "stats_read_writer_test.go",
     ],
     flaky = True,
-    shard_count = 27,
+    shard_count = 28,
     deps = [
         ":storage",
         "//pkg/domain",

pkg/statistics/handle/storage/dump_test.go

@@ -197,6 +197,33 @@ func TestLoadGlobalStats(t *testing.T) {
 	require.Equal(t, 3, len(loadedStats.Partitions)) // p0, p1, global
 }
 
+func TestLastStatsHistUpdateVersionAfterLoadStats(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+	tk.MustExec("use test")
+	tk.MustExec("set @@tidb_analyze_version = 2")
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec("create table t (a int, key(a))")
+	tk.MustExec("insert into t values (1), (2)")
+	tk.MustExec("analyze table t")
+
+	statsHandle := dom.StatsHandle()
+	table, err := dom.InfoSchema().TableByName(context.Background(), ast.NewCIStr("test"), ast.NewCIStr("t"))
+	require.NoError(t, err)
+	tableInfo := table.Meta()
+	statsTbl := statsHandle.GetTableStats(tableInfo)
+	require.Greater(t, statsTbl.LastStatsHistVersion, uint64(0))
+	origLastStatsHistVersion := statsTbl.LastStatsHistVersion
+
+	jsonTbl := getStatsJSON(t, dom, "test", "t")
+	dom.StatsHandle().Clear()
+	require.Nil(t, statsHandle.LoadStatsFromJSON(context.Background(), dom.InfoSchema(), jsonTbl, 0))
+	require.NoError(t, statsHandle.Update(context.Background(), dom.InfoSchema()))
+	statsTbl = statsHandle.GetTableStats(tableInfo)
+	require.Greater(t, statsTbl.LastStatsHistVersion, origLastStatsHistVersion)
+}
+
 func TestLoadPartitionStats(t *testing.T) {
 	store, dom := testkit.CreateMockStoreAndDomain(t)
 	tk := testkit.NewTestKit(t, store)

pkg/statistics/handle/storage/gc.go

@@ -141,8 +141,10 @@ func DeleteTableStatsFromKV(sctx sessionctx.Context, statsIDs []int64, soft bool
 		return errors.Trace(err)
 	}
 	for _, statsID := range statsIDs {
-		// We only update the version so that other tidb will know that this table is deleted.
-		if _, err = util.Exec(sctx, "update mysql.stats_meta set version = %? where table_id = %? ", startTS, statsID); err != nil {
+		// We update the version so that other tidb will know that this table is deleted.
+		// And we also update the last_stats_histograms_version to tell other tidb that the stats histogram is deleted
+		// and they should update their memory cache. It's mainly for soft delete triggered by DROP STATS.
+		if _, err = util.Exec(sctx, "update mysql.stats_meta set version = %?, last_stats_histograms_version = %? where table_id = %? ", startTS, startTS, statsID); err != nil {
 			return err
 		}
 		if soft {
@@ -245,7 +247,7 @@ func deleteHistStatsFromKV(sctx sessionctx.Context, physicalID int64, histID int
 		return errors.Trace(err)
 	}
 	// First of all, we update the version. If this table doesn't exist, it won't have any problem. Because we cannot delete anything.
-	if _, err = util.Exec(sctx, "update mysql.stats_meta set version = %? where table_id = %? ", startTS, physicalID); err != nil {
+	if _, err = util.Exec(sctx, "update mysql.stats_meta set version = %?, last_stats_histograms_version = %? where table_id = %? ", startTS, startTS, physicalID); err != nil {
 		return err
 	}
 	// delete histogram meta
@@ -434,7 +436,7 @@ func MarkExtendedStatsDeleted(sctx sessionctx.Context,
 	if err != nil {
 		return 0, errors.Trace(err)
 	}
-	if _, err = util.Exec(sctx, "UPDATE mysql.stats_meta SET version = %? WHERE table_id = %?", version, tableID); err != nil {
+	if _, err = util.Exec(sctx, "UPDATE mysql.stats_meta SET version = %?, last_stats_histograms_version = %? WHERE table_id = %?", version, version, tableID); err != nil {
 		return 0, err
 	}
 	statsVer = version