session/statistics: discard feedbacks from delete / update

executor/builder.go

@@ -2275,7 +2275,7 @@ func buildNoRangeTableReader(b *executorBuilder, v *plannercore.PhysicalTableRea
 	} else {
 		e.feedback = statistics.NewQueryFeedback(getPhysicalTableID(tbl), ts.Hist, int64(ts.StatsCount()), ts.Desc)
 	}
-	collect := (b.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl != nil) || e.feedback.CollectFeedback(len(ts.Ranges))
+	collect := statistics.CollectFeedback(b.ctx.GetSessionVars().StmtCtx, e.feedback, len(ts.Ranges))
 	if !collect {
 		e.feedback.Invalidate()
 	}
@@ -2358,7 +2358,7 @@ func buildNoRangeIndexReader(b *executorBuilder, v *plannercore.PhysicalIndexRea
 	} else {
 		e.feedback = statistics.NewQueryFeedback(e.physicalTableID, is.Hist, int64(is.StatsCount()), is.Desc)
 	}
-	collect := (b.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl != nil) || e.feedback.CollectFeedback(len(is.Ranges))
+	collect := statistics.CollectFeedback(b.ctx.GetSessionVars().StmtCtx, e.feedback, len(is.Ranges))
 	if !collect {
 		e.feedback.Invalidate()
 	}
@@ -2461,10 +2461,10 @@ func buildNoRangeIndexLookUpReader(b *executorBuilder, v *plannercore.PhysicalIn
 	} else {
 		e.feedback = statistics.NewQueryFeedback(getPhysicalTableID(tbl), is.Hist, int64(is.StatsCount()), is.Desc)
 	}
-	// do not collect the feedback for table request.
+	// Do not collect the feedback for table request.
 	collectTable := false
 	e.tableRequest.CollectRangeCounts = &collectTable
-	collectIndex := (b.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl != nil) || e.feedback.CollectFeedback(len(is.Ranges))
+	collectIndex := statistics.CollectFeedback(b.ctx.GetSessionVars().StmtCtx, e.feedback, len(is.Ranges))
 	if !collectIndex {
 		e.feedback.Invalidate()
 	}

executor/distsql.go

@@ -245,7 +245,9 @@ func (e *IndexReaderExecutor) Close() error {
 		copStats := e.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl.GetRootStats(e.plans[0].ExplainID().String())
 		copStats.SetRowNum(e.feedback.Actual())
 	}
-	e.ctx.StoreQueryFeedback(e.feedback)
+	if e.ctx.GetSessionVars().StmtCtx.InSelectStmt {
+		e.ctx.StoreQueryFeedback(e.feedback)
+	}
 	return err
 }
 
@@ -479,6 +481,8 @@ func (e *IndexLookUpExecutor) startIndexWorker(ctx context.Context, kvRanges []k
 	if worker.batchSize > worker.maxBatchSize {
 		worker.batchSize = worker.maxBatchSize
 	}
+	// Check if the query is SelectStmt before starting the new goroutine, otherwise data race may happen.
+	isSel := e.ctx.GetSessionVars().StmtCtx.InSelectStmt
 	e.idxWorkerWg.Add(1)
 	go func() {
 		ctx1, cancel := context.WithCancel(ctx)
@@ -496,7 +500,9 @@ func (e *IndexLookUpExecutor) startIndexWorker(ctx context.Context, kvRanges []k
 			copStats = e.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl.GetRootStats(e.tblPlans[0].ExplainID().String())
 			copStats.SetRowNum(int64(count))
 		}
-		e.ctx.StoreQueryFeedback(e.feedback)
+		if isSel {
+			e.ctx.StoreQueryFeedback(e.feedback)
+		}
 		close(workCh)
 		close(e.resultCh)
 		e.idxWorkerWg.Done()

executor/index_merge_reader.go

@@ -214,6 +214,7 @@ func (e *IndexMergeReaderExecutor) startPartialIndexWorker(ctx context.Context,
 		return errors.New("inject an error before start partialIndexWorker")
 	})
 
+	isSel := e.ctx.GetSessionVars().StmtCtx.InSelectStmt
 	go func() {
 		defer partialWorkerWg.Done()
 		ctx1, cancel := context.WithCancel(ctx)
@@ -231,7 +232,9 @@ func (e *IndexMergeReaderExecutor) startPartialIndexWorker(ctx context.Context,
 		if err := result.Close(); err != nil {
 			logutil.Logger(ctx).Error("close Select result failed:", zap.Error(err))
 		}
-		e.ctx.StoreQueryFeedback(e.feedbacks[workID])
+		if isSel {
+			e.ctx.StoreQueryFeedback(e.feedbacks[workID])
+		}
 	}()
 
 	return nil
@@ -271,6 +274,7 @@ func (e *IndexMergeReaderExecutor) startPartialTableWorker(ctx context.Context,
 	if worker.batchSize > worker.maxBatchSize {
 		worker.batchSize = worker.maxBatchSize
 	}
+	isSel := e.ctx.GetSessionVars().StmtCtx.InSelectStmt
 	go func() {
 		defer partialWorkerWg.Done()
 		ctx1, cancel := context.WithCancel(ctx)
@@ -288,7 +292,9 @@ func (e *IndexMergeReaderExecutor) startPartialTableWorker(ctx context.Context,
 		if err := worker.tableReader.Close(); err != nil {
 			logutil.Logger(ctx).Error("close Select result failed:", zap.Error(err))
 		}
-		e.ctx.StoreQueryFeedback(e.feedbacks[workID])
+		if isSel {
+			e.ctx.StoreQueryFeedback(e.feedbacks[workID])
+		}
 	}()
 	return nil
 }

executor/table_reader.go

@@ -186,7 +186,9 @@ func (e *TableReaderExecutor) Close() error {
 		copStats := e.ctx.GetSessionVars().StmtCtx.RuntimeStatsColl.GetRootStats(e.plans[0].ExplainID().String())
 		copStats.SetRowNum(e.feedback.Actual())
 	}
-	e.ctx.StoreQueryFeedback(e.feedback)
+	if e.ctx.GetSessionVars().StmtCtx.InSelectStmt {
+		e.ctx.StoreQueryFeedback(e.feedback)
+	}
 	return err
 }
 

statistics/feedback.go

@@ -98,10 +98,17 @@ func (q *QueryFeedback) CalcErrorRate() float64 {
 }
 
 // CollectFeedback decides whether to collect the feedback. It returns false when:
-// 1: the histogram is nil or has no buckets;
-// 2: the number of scan ranges exceeds the limit because it may affect the performance;
-// 3: it does not pass the probabilistic sampler.
-func (q *QueryFeedback) CollectFeedback(numOfRanges int) bool {
+// 1: the feedback is not generated by select query;
+// 2: the histogram is nil or has no buckets;
+// 3: the number of scan ranges exceeds the limit because it may affect the performance;
+// 4: it does not pass the probabilistic sampler.
+func CollectFeedback(sc *stmtctx.StatementContext, q *QueryFeedback, numOfRanges int) bool {
+	if sc.RuntimeStatsColl != nil {
+		return true
+	}
+	if !sc.InSelectStmt {
+		return false
+	}
 	if q.Hist == nil || q.Hist.Len() == 0 {
 		return false
 	}

statistics/handle/update_test.go

@@ -1674,3 +1674,30 @@ func (s *testStatsSuite) TestLoadHistCorrelation(c *C) {
 	c.Assert(len(result.Rows()), Equals, 1)
 	c.Assert(result.Rows()[0][9], Equals, "1")
 }
+
+func (s *testStatsSuite) TestDeleteUpdateFeedback(c *C) {
+	defer cleanEnv(c, s.store, s.do)
+	testKit := testkit.NewTestKit(c, s.store)
+
+	oriProbability := statistics.FeedbackProbability
+	defer func() {
+		statistics.FeedbackProbability = oriProbability
+	}()
+	statistics.FeedbackProbability.Store(1)
+
+	h := s.do.StatsHandle()
+	testKit.MustExec("use test")
+	testKit.MustExec("create table t (a bigint(64), b bigint(64), index idx_ab(a,b))")
+	for i := 0; i < 20; i++ {
+		testKit.MustExec(fmt.Sprintf("insert into t values (%d, %d)", i/5, i))
+	}
+	c.Assert(h.DumpStatsDeltaToKV(handle.DumpAll), IsNil)
+	testKit.MustExec("analyze table t with 3 buckets")
+
+	testKit.MustExec("delete from t where a = 1")
+	c.Assert(h.DumpStatsDeltaToKV(handle.DumpAll), IsNil)
+	c.Assert(len(h.GetQueryFeedback()), Equals, 0)
+	testKit.MustExec("update t set a = 6 where a = 2")
+	c.Assert(h.DumpStatsDeltaToKV(handle.DumpAll), IsNil)
+	c.Assert(len(h.GetQueryFeedback()), Equals, 0)
+}