ddl, stats: handle add/trunctate partition events (#14500) (#14566)

ddl/ddl_worker.go

@@ -532,7 +532,7 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 	case model.ActionDropTablePartition:
 		ver, err = onDropTablePartition(t, job)
 	case model.ActionTruncateTablePartition:
-		ver, err = onTruncateTablePartition(t, job)
+		ver, err = onTruncateTablePartition(d, t, job)
 	case model.ActionAddColumn:
 		ver, err = onAddColumn(d, t, job)
 	case model.ActionDropColumn:
@@ -564,7 +564,7 @@ func (w *worker) runDDLJob(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64,
 	case model.ActionModifyTableComment:
 		ver, err = onModifyTableComment(t, job)
 	case model.ActionAddTablePartition:
-		ver, err = onAddTablePartition(t, job)
+		ver, err = onAddTablePartition(d, t, job)
 	case model.ActionModifyTableCharsetAndCollate:
 		ver, err = onModifyTableCharsetAndCollate(t, job)
 	case model.ActionRecoverTable:

ddl/partition.go

@@ -25,6 +25,7 @@ import (
 	"github.com/pingcap/parser/model"
 	"github.com/pingcap/parser/mysql"
 	"github.com/pingcap/parser/opcode"
+	"github.com/pingcap/tidb/ddl/util"
 	"github.com/pingcap/tidb/expression"
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/meta"
@@ -425,7 +426,7 @@ func onDropTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 }
 
 // onDropTablePartition truncates old partition meta.
-func onTruncateTablePartition(t *meta.Meta, job *model.Job) (int64, error) {
+func onTruncateTablePartition(d *ddlCtx, t *meta.Meta, job *model.Job) (int64, error) {
 	var ver int64
 	var oldID int64
 	if err := job.DecodeArgs(&oldID); err != nil {
@@ -441,7 +442,7 @@ func onTruncateTablePartition(t *meta.Meta, job *model.Job) (int64, error) {
 		return ver, errors.Trace(ErrPartitionMgmtOnNonpartitioned)
 	}
 
-	var find bool
+	var newPartition *model.PartitionDefinition
 	for i := 0; i < len(pi.Definitions); i++ {
 		def := &pi.Definitions[i]
 		if def.ID == oldID {
@@ -450,11 +451,11 @@ func onTruncateTablePartition(t *meta.Meta, job *model.Job) (int64, error) {
 				return ver, errors.Trace(err1)
 			}
 			def.ID = pid
-			find = true
+			newPartition = def
 			break
 		}
 	}
-	if !find {
+	if newPartition == nil {
 		return ver, table.ErrUnknownPartition.GenWithStackByArgs("drop?", tblInfo.Name.O)
 	}
 
@@ -465,6 +466,7 @@ func onTruncateTablePartition(t *meta.Meta, job *model.Job) (int64, error) {
 
 	// Finish this job.
 	job.FinishTableJob(model.JobStateDone, model.StateNone, ver, tblInfo)
+	asyncNotifyEvent(d, &util.Event{Tp: model.ActionTruncateTablePartition, TableInfo: tblInfo, PartInfo: &model.PartitionInfo{Definitions: []model.PartitionDefinition{*newPartition}}})
 	// A background job will be created to delete old partition data.
 	job.Args = []interface{}{oldID}
 	return ver, nil

ddl/table.go

@@ -747,7 +747,7 @@ func updateVersionAndTableInfo(t *meta.Meta, job *model.Job, tblInfo *model.Tabl
 }
 
 // TODO: It may have the issue when two clients concurrently add partitions to a table.
-func onAddTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
+func onAddTablePartition(d *ddlCtx, t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	partInfo := &model.PartitionInfo{}
 	err := job.DecodeArgs(&partInfo)
 	if err != nil {
@@ -785,6 +785,7 @@ func onAddTablePartition(t *meta.Meta, job *model.Job) (ver int64, _ error) {
 	}
 	// Finish this job.
 	job.FinishTableJob(model.JobStateDone, model.StatePublic, ver, tblInfo)
+	asyncNotifyEvent(d, &util.Event{Tp: model.ActionAddTablePartition, TableInfo: tblInfo, PartInfo: partInfo})
 	return ver, errors.Trace(err)
 }
 

ddl/util/event.go

@@ -23,6 +23,7 @@ import (
 type Event struct {
 	Tp         model.ActionType
 	TableInfo  *model.TableInfo
+	PartInfo   *model.PartitionInfo
 	ColumnInfo *model.ColumnInfo
 	IndexInfo  *model.IndexInfo
 }
@@ -33,6 +34,13 @@ func (e *Event) String() string {
 	if e.TableInfo != nil {
 		ret += fmt.Sprintf(", Table ID: %d, Table Name %s", e.TableInfo.ID, e.TableInfo.Name)
 	}
+	if e.PartInfo != nil {
+		ids := make([]int64, 0, len(e.PartInfo.Definitions))
+		for _, def := range e.PartInfo.Definitions {
+			ids = append(ids, def.ID)
+		}
+		ret += fmt.Sprintf(", Partition IDs: %v", ids)
+	}
 	if e.ColumnInfo != nil {
 		ret += fmt.Sprintf(", Column ID: %d, Column Name %s", e.ColumnInfo.ID, e.ColumnInfo.Name)
 	}

statistics/handle/ddl.go

@@ -43,6 +43,12 @@ func (h *Handle) HandleDDLEvent(t *util.Event) error {
 				return err
 			}
 		}
+	case model.ActionAddTablePartition, model.ActionTruncateTablePartition:
+		for _, def := range t.PartInfo.Definitions {
+			if err := h.insertTableStats2KV(t.TableInfo, def.ID); err != nil {
+				return err
+			}
+		}
 	}
 	return nil
 }

statistics/handle/ddl_test.go

@@ -214,4 +214,34 @@ PARTITION BY RANGE ( a ) (
 		c.Assert(statsTbl.Pseudo, IsFalse)
 		c.Check(statsTbl.Columns[tableInfo.Columns[2].ID].AvgColSize(statsTbl.Count), Equals, 3.0)
 	}
+
+	addPartition := "alter table t add partition (partition p4 values less than (26))"
+	testKit.MustExec(addPartition)
+	is = s.do.InfoSchema()
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t"))
+	c.Assert(err, IsNil)
+	tableInfo = tbl.Meta()
+	err = h.HandleDDLEvent(<-h.DDLEventCh())
+	c.Assert(err, IsNil)
+	c.Assert(h.Update(is), IsNil)
+	pi = tableInfo.GetPartitionInfo()
+	for _, def := range pi.Definitions {
+		statsTbl := h.GetPartitionStats(tableInfo, def.ID)
+		c.Assert(statsTbl.Pseudo, IsFalse)
+	}
+
+	truncatePartition := "alter table t truncate partition p4"
+	testKit.MustExec(truncatePartition)
+	is = s.do.InfoSchema()
+	tbl, err = is.TableByName(model.NewCIStr("test"), model.NewCIStr("t"))
+	c.Assert(err, IsNil)
+	tableInfo = tbl.Meta()
+	err = h.HandleDDLEvent(<-h.DDLEventCh())
+	c.Assert(err, IsNil)
+	c.Assert(h.Update(is), IsNil)
+	pi = tableInfo.GetPartitionInfo()
+	for _, def := range pi.Definitions {
+		statsTbl := h.GetPartitionStats(tableInfo, def.ID)
+		c.Assert(statsTbl.Pseudo, IsFalse)
+	}
 }