statistics: do not record historical stats meta if the table is locked

pkg/statistics/handle/handletest/lockstats/BUILD.bazel

@@ -19,6 +19,7 @@ go_test(
         "//pkg/statistics",
         "//pkg/testkit",
         "//pkg/testkit/testsetup",
+        "@com_github_pingcap_failpoint//:failpoint",
         "@com_github_stretchr_testify//require",
         "@org_uber_go_goleak//:goleak",
     ],

pkg/statistics/handle/handletest/lockstats/lock_table_stats_test.go

@@ -21,6 +21,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/config"
 	"github.com/pingcap/tidb/pkg/domain"
 	"github.com/pingcap/tidb/pkg/kv"
@@ -60,14 +61,25 @@ func TestLockAndUnlockTableStats(t *testing.T) {
 	require.Nil(t, err)
 	require.Equal(t, 1, len(lockedTables))
 
+	// Insert a new row to the table.
+	tk.MustExec("insert into t(a, b) values(3,'c')")
+	// Enable the failpoint to test the historical stats meta is not recorded.
+	err = failpoint.Enable(
+		"github.com/pingcap/tidb/pkg/statistics/handle/usage/panic-when-record-historical-stats-meta",
+		"1*return(true)",
+	)
+	require.NoError(t, err)
+	// Dump stats delta to KV.
+	require.NotPanics(t, func() { handle.DumpStatsDeltaToKV(true) })
+
 	tk.MustExec("unlock stats t")
 	rows = tk.MustQuery(selectTableLockSQL).Rows()
 	num, _ = strconv.Atoi(rows[0][0].(string))
 	require.Equal(t, num, 0)
 
 	tk.MustExec("analyze table test.t")
 	tblStats2 := handle.GetTableStats(tbl)
-	require.Equal(t, int64(2), tblStats2.RealtimeCount)
+	require.Equal(t, int64(3), tblStats2.RealtimeCount)
 }
 
 func TestLockAndUnlockPartitionedTableStats(t *testing.T) {

pkg/statistics/handle/usage/BUILD.bazel

@@ -25,6 +25,7 @@ go_library(
         "//pkg/util/intest",
         "//pkg/util/sqlescape",
         "@com_github_pingcap_errors//:errors",
+        "@com_github_pingcap_failpoint//:failpoint",
     ],
 )
 

pkg/statistics/handle/usage/session_stats_collect.go

@@ -23,6 +23,7 @@ import (
 	"time"
 
 	"github.com/pingcap/errors"
+	"github.com/pingcap/failpoint"
 	"github.com/pingcap/tidb/pkg/infoschema"
 	"github.com/pingcap/tidb/pkg/meta/model"
 	"github.com/pingcap/tidb/pkg/metrics"
@@ -129,8 +130,13 @@ func (s *statsUsageImpl) DumpStatsDeltaToKV(dumpAll bool) error {
 // For a partitioned table, we will update its global-stats as well.
 func (s *statsUsageImpl) dumpTableStatCountToKV(is infoschema.InfoSchema, physicalTableID int64, delta variable.TableDelta) (updated bool, err error) {
 	statsVersion := uint64(0)
+	isLocked := false
 	defer func() {
-		if err == nil && statsVersion != 0 {
+		// Only record the historical stats meta when the table is not locked because all stats meta are stored in the locked table.
+		if err == nil && statsVersion != 0 && !isLocked {
+			failpoint.Inject("panic-when-record-historical-stats-meta", func() {
+				panic("panic when record historical stats meta")
+			})
 			s.statsHandle.RecordHistoricalStatsMeta(physicalTableID, statsVersion, "flush stats", false)
 		}
 	}()
@@ -171,6 +177,7 @@ func (s *statsUsageImpl) dumpTableStatCountToKV(is infoschema.InfoSchema, physic
 				isPartitionLocked = true
 			}
 			tableOrPartitionLocked := isTableLocked || isPartitionLocked
+			isLocked = tableOrPartitionLocked
 			if err = storage.UpdateStatsMeta(utilstats.StatsCtx, sctx, statsVersion, delta,
 				physicalTableID, tableOrPartitionLocked); err != nil {
 				return err
@@ -201,6 +208,7 @@ func (s *statsUsageImpl) dumpTableStatCountToKV(is infoschema.InfoSchema, physic
 			if _, ok := lockedTables[physicalTableID]; ok {
 				isTableLocked = true
 			}
+			isLocked = isTableLocked
 			if err = storage.UpdateStatsMeta(utilstats.StatsCtx, sctx, statsVersion, delta,
 				physicalTableID, isTableLocked); err != nil {
 				return err