stats: fix panic caused by outdated feedback

statistics/update.go

@@ -25,9 +25,9 @@ import (
 	"github.com/pingcap/tidb/infoschema"
 	"github.com/pingcap/tidb/metrics"
 	"github.com/pingcap/tidb/model"
-	"github.com/pingcap/tidb/mysql"
 	"github.com/pingcap/tidb/sessionctx/variable"
 	"github.com/pingcap/tidb/store/tikv/oracle"
+	"github.com/pingcap/tidb/types"
 	"github.com/pingcap/tidb/util/sqlexec"
 	log "github.com/sirupsen/logrus"
 	"golang.org/x/net/context"
@@ -473,79 +473,84 @@ func (h *Handle) HandleUpdateStats(is infoschema.InfoSchema) error {
 	if len(rows) == 0 || err != nil {
 		return errors.Trace(err)
 	}
-	tableID, histID, isIndex := int64(-1), int64(-1), int64(-1)
-	q := &QueryFeedback{}
-	var (
-		cms        *CMSketch
-		hist       *Histogram
-		col        *Column
-		idx        *Index
-		PKIsHandle bool
-	)
-	for _, row := range rows {
-		// merge into previous feedback
-		if row.GetInt64(0) == tableID && row.GetInt64(1) == histID && row.GetInt64(2) == isIndex {
-			err = decodeFeedback(row.GetBytes(3), q, cms)
-			if err != nil {
-				log.Debugf("decode feedback failed, err: %v", errors.ErrorStack(err))
-			}
-			continue
-		}
-		// dump the stats into kv
-		if hist != nil {
-			// Update the NDV of primary key column.
-			if col != nil && mysql.HasPriKeyFlag(col.Info.Flag) && PKIsHandle {
-				hist.NDV = int64(hist.totalRowCount())
-				col = nil
-			}
-			err = h.dumpStatsUpdateToKV(tableID, int(isIndex), q, hist, cms)
-			if err != nil {
-				return errors.Trace(err)
-			}
+
+	var groupedRows [][]types.Row
+	preIdx := 0
+	tableID, histID, isIndex := rows[0].GetInt64(0), rows[0].GetInt64(1), rows[0].GetInt64(2)
+	for i := 1; i < len(rows); i++ {
+		row := rows[i]
+		if row.GetInt64(0) != tableID || row.GetInt64(1) != histID || row.GetInt64(2) != isIndex {
+			groupedRows = append(groupedRows, rows[preIdx:i])
+			tableID, histID, isIndex = row.GetInt64(0), row.GetInt64(1), row.GetInt64(2)
+			preIdx = i
 		}
-		// initialize new feedback
-		tableID, histID, isIndex = row.GetInt64(0), row.GetInt64(1), row.GetInt64(2)
+	}
+	groupedRows = append(groupedRows, rows[preIdx:])
+
+	q := &QueryFeedback{}
+	for _, rows := range groupedRows {
+		tableID, histID, isIndex = rows[0].GetInt64(0), rows[0].GetInt64(1), rows[0].GetInt64(2)
 		table, ok := is.TableByID(tableID)
+		// The table has been deleted.
 		if !ok {
-			hist, cms = nil, nil
+			if err := h.deleteOutdatedFeedback(tableID, histID, isIndex); err != nil {
+				return errors.Trace(err)
+			}
 			continue
 		}
-		PKIsHandle = table.Meta().PKIsHandle
 		tbl := h.GetTableStats(table.Meta())
+		var cms *CMSketch
+		var hist *Histogram
 		if isIndex == 1 {
-			idx, ok = tbl.Indices[histID]
-			if !ok {
-				hist, cms = nil, nil
-				continue
+			idx, ok := tbl.Indices[histID]
+			if ok {
+				idxHist := idx.Histogram
+				hist = &idxHist
+				cms = idx.CMSketch.copy()
 			}
-			idxHist := idx.Histogram
-			hist = &idxHist
-			cms = idx.CMSketch.copy()
 		} else {
-			col, ok = tbl.Columns[histID]
-			if !ok {
-				hist, cms = nil, nil
-				continue
+			col, ok := tbl.Columns[histID]
+			if ok {
+				colHist := col.Histogram
+				hist = &colHist
 			}
-			colHist := col.Histogram
-			hist = &colHist
-			cms = nil
 		}
-		err = decodeFeedback(row.GetBytes(3), q, cms)
+		// The column or index has been deleted.
+		if hist == nil {
+			if err := h.deleteOutdatedFeedback(tableID, histID, isIndex); err != nil {
+				return errors.Trace(err)
+			}
+			continue
+		}
+		for _, row := range rows {
+			err = decodeFeedback(row.GetBytes(3), q, cms)
+			if err != nil {
+				log.Debugf("decode feedback failed, err: %v", errors.ErrorStack(err))
+			}
+		}
+		// Update the NDV of primary key column.
+		if table.Meta().PKIsHandle && isIndex == 0 {
+			hist.NDV = int64(hist.totalRowCount())
+		}
+		err = h.dumpStatsUpdateToKV(tableID, isIndex, q, hist, cms)
 		if err != nil {
-			log.Debugf("decode feedback failed, err: %v", errors.ErrorStack(err))
+			return errors.Trace(err)
 		}
 	}
-	// Update the NDV of primary key column.
-	if col != nil && mysql.HasPriKeyFlag(col.Info.Flag) && PKIsHandle {
-		hist.NDV = int64(hist.totalRowCount())
-	}
-	// dump the last feedback into kv
-	err = h.dumpStatsUpdateToKV(tableID, int(isIndex), q, hist, cms)
+	return nil
+}
+
+func (h *Handle) deleteOutdatedFeedback(tableID, histID, isIndex int64) error {
+	h.mu.Lock()
+	h.mu.ctx.GetSessionVars().BatchDelete = true
+	sql := fmt.Sprintf("delete from mysql.stats_feedback where table_id = %d and hist_id = %d and is_index = %d", tableID, histID, isIndex)
+	_, err := h.mu.ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
+	h.mu.ctx.GetSessionVars().BatchDelete = false
+	h.mu.Unlock()
 	return errors.Trace(err)
 }
 
-func (h *Handle) dumpStatsUpdateToKV(tableID int64, isIndex int, q *QueryFeedback, hist *Histogram, cms *CMSketch) (err error) {
+func (h *Handle) dumpStatsUpdateToKV(tableID, isIndex int64, q *QueryFeedback, hist *Histogram, cms *CMSketch) (err error) {
 	defer func() {
 		if err != nil {
 			metrics.UpdateStatsCounter.WithLabelValues(metrics.LblError).Inc()
@@ -554,16 +559,12 @@ func (h *Handle) dumpStatsUpdateToKV(tableID int64, isIndex int, q *QueryFeedbac
 		}
 	}()
 	hist = UpdateHistogram(hist, q)
-	err = h.SaveStatsToStorage(tableID, -1, isIndex, hist, cms, 0)
+	q.feedback = q.feedback[:0]
+	err = h.SaveStatsToStorage(tableID, -1, int(isIndex), hist, cms, 0)
 	if err != nil {
 		return errors.Trace(err)
 	}
-	h.mu.Lock()
-	h.mu.ctx.GetSessionVars().BatchDelete = true
-	sql := fmt.Sprintf("delete from mysql.stats_feedback where table_id = %d and hist_id = %d and is_index = %d", tableID, hist.ID, isIndex)
-	_, err = h.mu.ctx.(sqlexec.SQLExecutor).Execute(context.TODO(), sql)
-	h.mu.Unlock()
-	q.feedback = q.feedback[:0]
+	err = h.deleteOutdatedFeedback(tableID, hist.ID, isIndex)
 	return errors.Trace(err)
 }
 

statistics/update_test.go

@@ -619,6 +619,16 @@ func (s *testStatsUpdateSuite) TestQueryFeedback(c *C) {
 		feedback := h.GetQueryFeedback()
 		c.Assert(len(feedback), Equals, 0)
 	}
+
+	// Test that the outdated feedback won't cause panic.
+	statistics.FeedbackProbability = 1
+	for _, t := range tests {
+		testKit.MustQuery(t.sql)
+	}
+	c.Assert(h.DumpStatsDeltaToKV(statistics.DumpAll), IsNil)
+	c.Assert(h.DumpStatsFeedbackToKV(), IsNil)
+	testKit.MustExec("drop table t")
+	c.Assert(h.HandleUpdateStats(s.do.InfoSchema()), IsNil)
 }
 
 func (s *testStatsUpdateSuite) TestUpdateSystemTable(c *C) {