executor: add runtime information for DML statement in explain analyze

executor/adapter.go

@@ -387,19 +387,22 @@ func (a *ExecStmt) handleNoDelay(ctx context.Context, e Executor, isPessimistic
 	}()
 
 	toCheck := e
+	isExplainAnalyze := false
 	if explain, ok := e.(*ExplainExec); ok {
-		if explain.analyzeExec != nil {
-			toCheck = explain.analyzeExec
+		if analyze := explain.getAnalyzeExecToExecutedNoDelay(); analyze != nil {
+			toCheck = analyze
+			isExplainAnalyze = true
 		}
 	}
 
 	// If the executor doesn't return any result to the client, we execute it without delay.
 	if toCheck.Schema().Len() == 0 {
+		handled = !isExplainAnalyze
 		if isPessimistic {
-			return true, nil, a.handlePessimisticDML(ctx, e)
+			return handled, nil, a.handlePessimisticDML(ctx, toCheck)
 		}
-		r, err := a.handleNoDelayExecutor(ctx, e)
-		return true, r, err
+		r, err := a.handleNoDelayExecutor(ctx, toCheck)
+		return handled, r, err
 	} else if proj, ok := toCheck.(*ProjectionExec); ok && proj.calculateNoDelay {
 		// Currently this is only for the "DO" statement. Take "DO 1, @a=2;" as an example:
 		// the Projection has two expressions and two columns in the schema, but we should

executor/explain.go

@@ -28,6 +28,7 @@ type ExplainExec struct {
 
 	explain     *core.Explain
 	analyzeExec Executor
+	executed    bool
 	rows        [][]string
 	cursor      int
 }
@@ -79,7 +80,8 @@ func (e *ExplainExec) generateExplainInfo(ctx context.Context) (rows [][]string,
 			closed = true
 		}
 	}()
-	if e.analyzeExec != nil {
+	if e.analyzeExec != nil && !e.executed {
+		e.executed = true
 		chk := newFirstChunk(e.analyzeExec)
 		var nextErr, closeErr error
 		for {
@@ -111,3 +113,16 @@ func (e *ExplainExec) generateExplainInfo(ctx context.Context) (rows [][]string,
 	}
 	return e.explain.Rows, nil
 }
+
+// getAnalyzeExecToExecutedNoDelay gets the analyze DML executor to execute in handleNoDelay function.
+// For explain analyze insert/update/delete statement, the analyze executor should be executed in handleNoDelay
+// function and then commit transaction if needed.
+// Otherwise, in autocommit transaction, the table record change of analyze executor(insert/update/delete...)
+// will not be committed.
+func (e *ExplainExec) getAnalyzeExecToExecutedNoDelay() Executor {
+	if e.analyzeExec != nil && !e.executed && e.analyzeExec.Schema().Len() == 0 {
+		e.executed = true
+		return e.analyzeExec
+	}
+	return nil
+}

planner/core/integration_test.go

@@ -1525,6 +1525,28 @@ func (s *testIntegrationSerialSuite) TestExplainAnalyzePointGet(c *C) {
 	checkExplain("BatchGet")
 }
 
+func (s *testIntegrationSerialSuite) TestExplainAnalyzeDML(c *C) {
+	tk := testkit.NewTestKit(c, s.store)
+	tk.MustExec("use test")
+	tk.MustExec("drop table if exists t")
+	tk.MustExec(" create table t (a int, b int, unique index (a));")
+	tk.MustExec("insert into t values (1,1)")
+
+	res := tk.MustQuery("explain analyze select * from t where a=1;")
+	checkExplain := func(rpc string) {
+		resBuff := bytes.NewBufferString("")
+		for _, row := range res.Rows() {
+			fmt.Fprintf(resBuff, "%s\n", row)
+		}
+		explain := resBuff.String()
+		c.Assert(strings.Contains(explain, rpc+":{num_rpc:"), IsTrue, Commentf("%s", explain))
+		c.Assert(strings.Contains(explain, "total_time:"), IsTrue, Commentf("%s", explain))
+	}
+	checkExplain("Get")
+	res = tk.MustQuery("explain analyze insert ignore into t values (1,1),(2,2),(3,3),(4,4);")
+	checkExplain("BatchGet")
+}
+
 func (s *testIntegrationSuite) TestPartitionExplain(c *C) {
 	tk := testkit.NewTestKit(c, s.store)
 	tk.MustExec("use test")

session/session.go

@@ -1225,6 +1225,7 @@ func runStmt(ctx context.Context, se *session, s sqlexec.Statement) (rs sqlexec.
 				se.StmtCommit()
 			}
 		}
+		err = finishStmt(ctx, se, err, s)
 	}
 	if rs != nil {
 		return &execStmtResult{
@@ -1234,8 +1235,6 @@ func runStmt(ctx context.Context, se *session, s sqlexec.Statement) (rs sqlexec.
 		}, err
 	}
 
-	err = finishStmt(ctx, se, err, s)
-
 	// If it is not a select statement, we record its slow log here,
 	// then it could include the transaction commit time.
 	s.(*executor.ExecStmt).FinishExecuteStmt(origTxnCtx.StartTS, err == nil, false)