statistics: gc the statistics correctly after drop the database

pkg/statistics/handle/ddl/BUILD.bazel

@@ -35,7 +35,7 @@ go_test(
     timeout = "short",
     srcs = ["ddl_test.go"],
     flaky = True,
-    shard_count = 19,
+    shard_count = 20,
     deps = [
         ":ddl",
         "//pkg/ddl/notifier",

pkg/statistics/handle/ddl/ddl.go

@@ -181,7 +181,18 @@ func (h *ddlHandlerImpl) HandleDDLEvent(s *notifier.SchemaChangeEvent) error {
 	case model.ActionAddIndex:
 		// No need to update the stats meta for the adding index event.
 	case model.ActionDropSchema:
-		// TODO: handle the drop schema event.
+		miniDBInfo := s.GetDropSchemaInfo()
+		intest.Assert(miniDBInfo != nil)
+		for _, table := range miniDBInfo.Tables {
+			// Try best effort to update the stats meta version for gc.
+			if err := h.statsWriter.UpdateStatsMetaVersionForGC(table.ID); err != nil {
+				logutil.StatsLogger().Error(
+					"Failed to update stats meta version for gc",
+					zap.Int64("tableID", table.ID),
+					zap.Error(err),
+				)
+			}
+		}
 	default:
 		intest.Assert(false)
 		logutil.StatsLogger().Error("Unhandled schema change event", zap.Stringer("type", s))

pkg/statistics/handle/ddl/ddl_test.go

@@ -1306,6 +1306,44 @@ func TestAddPartitioning(t *testing.T) {
 	)
 }
 
+func TestDropSchema(t *testing.T) {
+	store, dom := testkit.CreateMockStoreAndDomain(t)
+	tk := testkit.NewTestKit(t, store)
+
+	tk.MustExec("use test")
+	tk.MustExec("create table t (c1 int)")
+	h := dom.StatsHandle()
+	tk.MustExec("insert into t values (1)")
+	require.NoError(t, h.DumpStatsDeltaToKV(true))
+
+	is := dom.InfoSchema()
+	tbl, err := is.TableByName(context.Background(), pmodel.NewCIStr("test"), pmodel.NewCIStr("t"))
+	require.NoError(t, err)
+	tableInfo := tbl.Meta()
+	// Check the current stats meta version.
+	rows := tk.MustQuery(
+		"select version from mysql.stats_meta where table_id = ?",
+		tableInfo.ID,
+	).Rows()
+	require.Len(t, rows, 1)
+	version := rows[0][0].(string)
+
+	tk.MustExec("drop database test")
+
+	// Handle the drop schema event.
+	dropSchemaEvent := findEvent(h.DDLEventCh(), model.ActionDropSchema)
+	err = h.HandleDDLEvent(dropSchemaEvent)
+	require.NoError(t, err)
+
+	// Check the stats meta version after drop schema.
+	rows = tk.MustQuery(
+		"select version from mysql.stats_meta where table_id = ?",
+		tableInfo.ID,
+	).Rows()
+	require.Len(t, rows, 1)
+	require.NotEqual(t, version, rows[0][0].(string))
+}
+
 func findEvent(eventCh <-chan *notifier.SchemaChangeEvent, eventType model.ActionType) *notifier.SchemaChangeEvent {
 	// Find the target event.
 	for {