server,session: make sure ResultSet.Close() errors return to the clie…

…nt (pingcap#48447)

ref pingcap#48411, close pingcap#48446

pkg/server/conn.go

@@ -2049,11 +2049,7 @@ func (cc *clientConn) handleStmt(ctx context.Context, stmt ast.StmtNode, warns [
 			execStmt.(*executor.ExecStmt).FinishExecuteStmt(0, err, false)
 		}
 	}
-	if err != nil {
-		return false, err
-	}
-
-	return false, nil
+	return false, err
 }
 
 func (cc *clientConn) handleFileTransInConn(ctx context.Context, status uint16) (bool, error) {
@@ -2285,6 +2281,9 @@ func (cc *clientConn) writeChunks(ctx context.Context, rs resultset.ResultSet, b
 			stmtDetail.WriteSQLRespDuration += time.Since(start)
 		}
 	}
+	if err := rs.Close(); err != nil {
+		return false, err
+	}
 
 	if stmtDetail != nil {
 		start = time.Now()

pkg/session/session.go

@@ -2475,16 +2475,20 @@ const ExecStmtVarKey ExecStmtVarKeyType = 0
 // RecordSet, so this struct exists and RecordSet.Close() is overrided handle that.
 type execStmtResult struct {
 	sqlexec.RecordSet
-	se  *session
-	sql sqlexec.Statement
+	se     *session
+	sql    sqlexec.Statement
+	closed bool
 }
 
 func (rs *execStmtResult) Close() error {
-	se := rs.se
-	if err := rs.RecordSet.Close(); err != nil {
-		return finishStmt(context.Background(), se, err, rs.sql)
+	if rs.closed {
+		return nil
 	}
-	return finishStmt(context.Background(), se, nil, rs.sql)
+	se := rs.se
+	err := rs.RecordSet.Close()
+	err = finishStmt(context.Background(), se, err, rs.sql)
+	rs.closed = true
+	return err
 }
 
 // rollbackOnError makes sure the next statement starts a new transaction with the latest InfoSchema.