feat: support PERFORM SELECT statements

pkg/frontend/back_exec.go

@@ -821,6 +821,9 @@ func backSesOutputCallback(handle FeSession, execCtx *ExecCtx, dataSet *batch.Ba
 	if handle == nil || dataSet == nil {
 		return nil
 	}
+	if execCtx != nil && isPerformStatement(execCtx.stmt) {
+		return nil
+	}
 
 	// uncomment this to enable backExec export data to CSV file.
 	//back := handle.(*backSession)
@@ -939,6 +942,9 @@ func fakeDataSetFetcher2(handle FeSession, execCtx *ExecCtx, dataSet *batch.Batc
 	if handle == nil || dataSet == nil {
 		return nil
 	}
+	if execCtx != nil && isPerformStatement(execCtx.stmt) {
+		return nil
+	}
 
 	back := handle.(*backSession)
 	err := fillResultSet(execCtx.reqCtx, dataSet, back, back.mrs)
@@ -967,10 +973,13 @@ func fillResultSet(ctx context.Context, dataSet *batch.Batch, ses FeSession, mrs
 
 // batchFetcher2 gets the result batches from the pipeline and save the origin batches in the session.
 // It will not send the result to the client.
-func batchFetcher2(handle FeSession, _ *ExecCtx, dataSet *batch.Batch, _ *perfcounter.CounterSet) error {
+func batchFetcher2(handle FeSession, execCtx *ExecCtx, dataSet *batch.Batch, _ *perfcounter.CounterSet) error {
 	if handle == nil {
 		return nil
 	}
+	if execCtx != nil && isPerformStatement(execCtx.stmt) {
+		return nil
+	}
 	back := handle.(*backSession)
 	back.SaveResultSet()
 	if dataSet == nil {
@@ -981,10 +990,13 @@ func batchFetcher2(handle FeSession, _ *ExecCtx, dataSet *batch.Batch, _ *perfco
 
 // batchFetcher gets the result batches from the pipeline and save the origin batches in the session.
 // It will not send the result to the client.
-func batchFetcher(handle FeSession, _ *ExecCtx, dataSet *batch.Batch, _ *perfcounter.CounterSet) error {
+func batchFetcher(handle FeSession, execCtx *ExecCtx, dataSet *batch.Batch, _ *perfcounter.CounterSet) error {
 	if handle == nil {
 		return nil
 	}
+	if execCtx != nil && isPerformStatement(execCtx.stmt) {
+		return nil
+	}
 	ses := handle.(*Session)
 	ses.SaveResultSet()
 	if dataSet == nil {

pkg/frontend/back_status_stmt.go

@@ -35,6 +35,9 @@ func executeStatusStmtInBack(backSes *backSession,
 	if execCtx.runResult, err = execCtx.runner.Run(0); err != nil {
 		return
 	}
+	if isPerformStatement(execCtx.stmt) && execCtx.runResult != nil {
+		execCtx.runResult.AffectRows = 0
+	}
 
 	// only log if run time is longer than 1s
 	if time.Since(runBegin) > time.Second {

pkg/frontend/mysql_cmd_executor.go

@@ -762,7 +762,7 @@ func getDataFromPipeline(obj FeSession, execCtx *ExecCtx, bat *batch.Batch, crs
 	}
 	tTime := time.Since(begin)
 	n := 0
-	if bat != nil && bat.Vecs[0] != nil {
+	if !isPerformStatement(execCtx.stmt) && bat != nil && bat.Vecs[0] != nil {
 		n = bat.Vecs[0].Length()
 		ses.sentRows.Add(int64(n))
 	}
@@ -1441,6 +1441,7 @@ type analyzeDerivedResponder struct {
 }
 
 var _ Responser = (*analyzeDerivedResponder)(nil)
+var _ queryResultFinalizer = (*analyzeDerivedResponder)(nil)
 
 func (r *analyzeDerivedResponder) GetStr(id PropertyID) string { return r.live.GetStr(id) }
 func (r *analyzeDerivedResponder) GetU32(id PropertyID) uint32 { return r.live.GetU32(id) }
@@ -4941,22 +4942,26 @@ func ExecRequest(ses *Session, execCtx *ExecCtx, req *Request) (resp *Response,
 		// fall through to normal MO execution.
 		if isSidecar, useGPU := isSidecarQuery(query); isSidecar {
 			ses.addSqlCount(1)
-			err = handleSidecarOffload(ses, execCtx, query, useGPU)
-			if err == nil {
-				ses.resetDiagnostics()
-				setRowCount(ses, ses.GetProc(), -1)
-				mer := NewMysqlExecutionResult(0, 0, 0, 0, ses.GetMysqlResultSet())
-				resp = ses.SetNewResponse(ResultResponse, 0, int(COM_QUERY), mer, true)
-				return resp, nil
-			}
-			if err != errSidecarNotConfigured {
-				ses.resetDiagnostics()
-				markRowCountFailed(ses, ses.GetProc())
-				resp = NewGeneralErrorResponse(COM_QUERY, ses.GetTxnHandler().GetServerStatus(), err)
-				return resp, nil
+			if sidecarQueryMustRunLocally(execCtx.reqCtx, ses, query) {
+				query = stripSidecarHint(query)
+			} else {
+				err = handleSidecarOffload(ses, execCtx, query, useGPU)
+				if err == nil {
+					ses.resetDiagnostics()
+					setRowCount(ses, ses.GetProc(), -1)
+					mer := NewMysqlExecutionResult(0, 0, 0, 0, ses.GetMysqlResultSet())
+					resp = ses.SetNewResponse(ResultResponse, 0, int(COM_QUERY), mer, true)
+					return resp, nil
+				}
+				if err != errSidecarNotConfigured {
+					ses.resetDiagnostics()
+					markRowCountFailed(ses, ses.GetProc())
+					resp = NewGeneralErrorResponse(COM_QUERY, ses.GetTxnHandler().GetServerStatus(), err)
+					return resp, nil
+				}
+				// errSidecarNotConfigured: strip hint and fall through to normal execution
+				query = stripSidecarHint(query)
 			}
-			// errSidecarNotConfigured: strip hint and fall through to normal execution
-			query = stripSidecarHint(query)
 		} else {
 			ses.addSqlCount(1)
 		}