Better statistics handling

ddl/db_partition_test.go

@@ -4914,16 +4914,27 @@ func TestRemoveHashPartitioning(t *testing.T) {
 }
 
 func TestRemoveKeyPartitioning(t *testing.T) {
-	store := testkit.CreateMockStore(t)
+	store, dom := testkit.CreateMockStoreAndDomain(t)
 	tk := testkit.NewTestKit(t, store)
+	h := dom.StatsHandle()
 	tk.MustExec("create database RemovePartitioning")
 	tk.MustExec("use RemovePartitioning")
 	tk.MustExec(`create table t (a varchar(255), b varchar(255), key (a,b), key (b)) partition by key (a) partitions 7`)
+	require.NoError(t, h.HandleDDLEvent(<-h.DDLEventCh()))
 	// Fill the data with ascii strings
 	for i := 32; i <= 126; i++ {
 		tk.MustExec(fmt.Sprintf(`insert into t values (char(%d,%d,%d),char(%d,%d,%d,%d))`, i, i, i, i, i, i, i))
 	}
 	tk.MustExec(`analyze table t`)
+	tk.MustQuery(`show stats_meta where db_name = 'RemovePartitioning' and table_name = 't'`).Sort().CheckAt([]int{0, 1, 2, 4, 5}, [][]interface{}{
+		{"RemovePartitioning", "t", "global", "0", "95"},
+		{"RemovePartitioning", "t", "p0", "0", "9"},
+		{"RemovePartitioning", "t", "p1", "0", "11"},
+		{"RemovePartitioning", "t", "p2", "0", "12"},
+		{"RemovePartitioning", "t", "p3", "0", "13"},
+		{"RemovePartitioning", "t", "p4", "0", "16"},
+		{"RemovePartitioning", "t", "p5", "0", "23"},
+		{"RemovePartitioning", "t", "p6", "0", "11"}})
 	tk.MustQuery(`select partition_name, table_rows from information_schema.partitions where table_schema = 'RemovePartitioning' and table_name = 't'`).Sort().Check(testkit.Rows(""+
 		"p0 9",
 		"p1 11",
@@ -4940,4 +4951,14 @@ func TestRemoveKeyPartitioning(t *testing.T) {
 		"  KEY `a` (`a`,`b`),\n" +
 		"  KEY `b` (`b`)\n" +
 		") ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_bin"))
+	// Statistics are updated asynchronously
+	require.NoError(t, h.HandleDDLEvent(<-h.DDLEventCh()))
+	// And also cached and lazy loaded
+	h.Clear()
+	require.NoError(t, h.Update(dom.InfoSchema()))
+	tk.MustQuery(`show stats_meta where db_name = 'RemovePartitioning' and table_name = 't'`).Sort().CheckAt([]int{0, 1, 2, 4, 5}, [][]interface{}{
+		{"RemovePartitioning", "t", "", "0", "95"}})
+	tk.MustExec(`analyze table t`)
+	tk.MustQuery(`show stats_meta where db_name = 'RemovePartitioning' and table_name = 't'`).Sort().CheckAt([]int{0, 1, 2, 4, 5}, [][]interface{}{
+		{"RemovePartitioning", "t", "", "0", "95"}})
 }

ddl/partition.go

@@ -2283,8 +2283,8 @@ func (w *worker) onRemovePartitioning(d *ddlCtx, t *meta.Meta, job *model.Job) (
 		physicalTableIDs := getPartitionIDsFromDefinitions(tblInfo.Partition.DroppingDefinitions)
 		newIDs := getPartitionIDsFromDefinitions(partInfo.Definitions)
 		job.CtxVars = []interface{}{physicalTableIDs, newIDs}
-		definitionsToDrop := tblInfo.Partition.DroppingDefinitions
 		tblInfo.Partition = nil
+		oldTblID := tblInfo.ID
 		err = t.DropTableOrView(job.SchemaID, tblInfo.ID)
 		if err != nil {
 			job.State = model.JobStateCancelled
@@ -2316,7 +2316,9 @@ func (w *worker) onRemovePartitioning(d *ddlCtx, t *meta.Meta, job *model.Job) (
 		// How to handle this?
 		// Seems to only trigger asynchronous update of statistics.
 		// Should it actually be synchronous?
-		asyncNotifyEvent(d, &util.Event{Tp: job.Type, TableInfo: tblInfo, PartInfo: &model.PartitionInfo{Definitions: definitionsToDrop}})
+		// Include the old table ID, which may contain global statistics,
+		// so it can be reused for the new non-partitioned table.
+		asyncNotifyEvent(d, &util.Event{Tp: job.Type, TableInfo: tblInfo, PartInfo: &model.PartitionInfo{Definitions: []model.PartitionDefinition{{ID: oldTblID}}}})
 		// A background job will be created to delete old partition data.
 		job.Args = []interface{}{physicalTableIDs}
 

ddl/rollingback.go

@@ -380,7 +380,7 @@ func convertJob2RollbackJob(w *worker, d *ddlCtx, t *meta.Meta, job *model.Job)
 		ver, err = rollingbackAddIndex(w, d, t, job, true)
 	case model.ActionAddTablePartition:
 		ver, err = rollingbackAddTablePartition(d, t, job)
-	case model.ActionReorganizePartition:
+	case model.ActionReorganizePartition, model.ActionRemovePartitioning:
 		ver, err = rollingbackReorganizePartition(d, t, job)
 	case model.ActionDropColumn:
 		ver, err = rollingbackDropColumn(d, t, job)

parser/model/ddl.go

@@ -760,7 +760,8 @@ func (job *Job) NotStarted() bool {
 // MayNeedReorg indicates that this job may need to reorganize the data.
 func (job *Job) MayNeedReorg() bool {
 	switch job.Type {
-	case ActionAddIndex, ActionAddPrimaryKey, ActionReorganizePartition:
+	case ActionAddIndex, ActionAddPrimaryKey, ActionReorganizePartition,
+		ActionRemovePartitioning:
 		return true
 	case ActionModifyColumn:
 		if len(job.CtxVars) > 0 {

parser/model/ddl_test.go

@@ -76,6 +76,7 @@ func TestMayNeedReorg(t *testing.T) {
 	//TODO(bb7133): add more test cases for different ActionType.
 	reorgJobTypes := []model.ActionType{
 		model.ActionReorganizePartition,
+		model.ActionRemovePartitioning,
 		model.ActionAddIndex,
 		model.ActionAddPrimaryKey,
 	}

statistics/handle/ddl.go

@@ -16,7 +16,6 @@ package handle
 
 import (
 	"context"
-
 	"github.com/pingcap/errors"
 	"github.com/pingcap/tidb/ddl/util"
 	"github.com/pingcap/tidb/kv"
@@ -68,10 +67,13 @@ func (h *Handle) HandleDDLEvent(t *util.Event) error {
 				return err
 			}
 		}
-		// Do not update global stats, since the data have not changed!
+	// Do not update global stats, since the data have not changed!
+	case model.ActionRemovePartitioning:
+		// For now, we do not have access to the old table id, so we cannot just update the id.
+		// We highjacked the first partition's physical table id for the previous Table ID
+		return h.changeGlobalStatsID(t.PartInfo.Definitions[0].ID, t.TableInfo.ID)
 	case model.ActionFlashbackCluster:
 		return h.updateStatsVersion()
-		// TODO: For model.ActionRemovePartitioning move global stats to table stats?
 	}
 	return nil
 }
@@ -205,6 +207,27 @@ func (h *Handle) updateGlobalStats(tblInfo *model.TableInfo) error {
 	return nil
 }
 
+func (h *Handle) changeGlobalStatsID(from, to int64) (err error) {
+	h.mu.Lock()
+	defer h.mu.Unlock()
+	ctx := kv.WithInternalSourceType(context.Background(), kv.InternalTxnStats)
+	exec := h.mu.ctx.(sqlexec.SQLExecutor)
+	_, err = exec.ExecuteInternal(ctx, "begin pessimistic")
+	if err != nil {
+		return errors.Trace(err)
+	}
+	defer func() {
+		err = finishTransaction(ctx, exec, err)
+	}()
+	for _, table := range []string{"stats_meta", "stats_top_n", "stats_fm_sketch", "stats_buckets", "stats_histograms", "column_stats_usage"} {
+		_, err = exec.ExecuteInternal(ctx, "update mysql."+table+" set table_id = %? where table_id = %?", to, from)
+		if err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
 func (h *Handle) getInitStateTableIDs(tblInfo *model.TableInfo) (ids []int64) {
 	pi := tblInfo.GetPartitionInfo()
 	if pi == nil {