sql: clarify cleanup queue for pausable portals

We have several cleanup "stacks" for pausable portals model, but in
reality we execute the functions in the "forward" direction, matching
the behavior of a "queue", so this commit renames the struct
accordingly. Only `execStmtInOpenState` and `dispatchToExecutionEngine`
stages have multiple cleanup functions.

Additionally, it fixes the inverted order of executing two cleanup
functions in `dispatchToExecutionEngine` stage. Namely, previously we
performed `planTop.Close` _before_ `recordStatementSummary` which
doesn't match the behavior on the main path, outside the pausable
portals. We now defer appending the former so that it's executed after
the latter.

Deferring `planTop.Close` was necessary for the following commit which
is what prompted this one.

Release note: None

Author: yuzefovich

pkg/sql/conn_executor_exec.go

@@ -1694,6 +1694,9 @@ func (ex *connExecutor) execStmtInOpenStateWithPausablePortal(
 			}
 		}
 
+		// Note that here we process the cleanup function (which will append it
+		// to the cleanup queue) without a defer since there is no more code
+		// relevant to pausable portals model below.
 		processCleanupFunc(func() {
 			cancelQueryCtx := ctx
 			if portal.isPausable() {
@@ -2772,9 +2775,11 @@ func (ex *connExecutor) dispatchToExecutionEngine(
 					defer planner.curPlan.close(ctx)
 				} else {
 					ppInfo.dispatchToExecutionEngine.planTop = planner.curPlan
-					ppInfo.dispatchToExecutionEngine.cleanup.appendFunc(func(ctx context.Context) {
-						ppInfo.dispatchToExecutionEngine.planTop.close(ctx)
-					})
+					defer func() {
+						ppInfo.dispatchToExecutionEngine.cleanup.appendFunc(func(ctx context.Context) {
+							ppInfo.dispatchToExecutionEngine.planTop.close(ctx)
+						})
+					}()
 				}
 			} else {
 				defer planner.curPlan.close(ctx)
@@ -2921,6 +2926,8 @@ func (ex *connExecutor) dispatchToExecutionEngine(
 		// We need to ensure that we're using the planner bound to the first-time
 		// execution of a portal.
 		curPlanner := *planner
+		// Note that here we append the cleanup function without a defer since
+		// there is no more code relevant to pausable portals model below.
 		ppInfo.dispatchToExecutionEngine.cleanup.appendFunc(func(ctx context.Context) {
 			populateQueryLevelStats(ctx, &curPlanner, ex.server.cfg, ppInfo.dispatchToExecutionEngine.queryStats, &ex.cpuStatsCollector)
 			ppInfo.dispatchToExecutionEngine.stmtFingerprintID = ex.recordStatementSummary(

pkg/sql/prepared_stmt.go

@@ -175,23 +175,28 @@ func (p *PreparedPortal) isPausable() bool {
 	return p != nil && p.pauseInfo != nil
 }
 
-// cleanupFuncStack stores cleanup functions for a portal. The clean-up
+// cleanupFuncQueue stores cleanup functions for a portal. The clean-up
 // functions are added during the first-time execution of a portal. When the
 // first-time execution is finished, we mark isComplete to true.
-type cleanupFuncStack struct {
-	stack      []func(context.Context)
+//
+// Generally, cleanup functions should be added in a defer (assuming that
+// originally they were deferred as well). The functions will be appended to the
+// end of the queue, which preserves the order of their execution if pausable
+// portals model wasn't used.
+type cleanupFuncQueue struct {
+	queue      []func(context.Context)
 	isComplete bool
 }
 
-func (n *cleanupFuncStack) appendFunc(f func(context.Context)) {
-	n.stack = append(n.stack, f)
+func (n *cleanupFuncQueue) appendFunc(f func(context.Context)) {
+	n.queue = append(n.queue, f)
 }
 
-func (n *cleanupFuncStack) run(ctx context.Context) {
-	for i := 0; i < len(n.stack); i++ {
-		n.stack[i](ctx)
+func (n *cleanupFuncQueue) run(ctx context.Context) {
+	for i := 0; i < len(n.queue); i++ {
+		n.queue[i](ctx)
 	}
-	*n = cleanupFuncStack{}
+	*n = cleanupFuncQueue{}
 }
 
 // instrumentationHelperWrapper wraps the instrumentation helper.
@@ -226,23 +231,23 @@ type portalPauseInfo struct {
 	// When closing a portal, we need to follow the reverse order of its execution,
 	// which means running the cleanup functions of the four structs in the
 	// following order:
-	//   - exhaustPortal.cleanup
-	//   - execStmtInOpenState.cleanup
-	//   - dispatchToExecutionEngine.cleanup
 	//   - resumableFlow.cleanup
+	//   - dispatchToExecutionEngine.cleanup
+	//   - execStmtInOpenState.cleanup
+	//   - exhaustPortal.cleanup
 	//
 	// If an error occurs in any of these functions, we run the cleanup functions of
 	// this layer and its children layers, and propagate the error to the parent
 	// layer. For example, if an error occurs in execStmtInOpenStateCleanup(), we
 	// run the cleanup functions in the following order:
-	//   - execStmtInOpenState.cleanup
-	//   - dispatchToExecutionEngine.cleanup
 	//   - resumableFlow.cleanup
+	//   - dispatchToExecutionEngine.cleanup
+	//   - execStmtInOpenState.cleanup
 	//
 	// When exiting connExecutor.execStmtInOpenState(), we finally run the
 	// exhaustPortal.cleanup function in connExecutor.execPortal().
 	exhaustPortal struct {
-		cleanup cleanupFuncStack
+		cleanup cleanupFuncQueue
 	}
 
 	// TODO(sql-session): replace certain fields here with planner.
@@ -266,7 +271,7 @@ type portalPauseInfo struct {
 		// retErr is needed for the cleanup steps as we will have to check the latest
 		// encountered error, so this field should be updated for each execution.
 		retErr  error
-		cleanup cleanupFuncStack
+		cleanup cleanupFuncQueue
 	}
 
 	dispatchReadCommittedStmtToExecutionEngine struct {
@@ -295,7 +300,7 @@ type portalPauseInfo struct {
 		// queryStats stores statistics on query execution. It is incremented for
 		// each execution of the portal.
 		queryStats *topLevelQueryStats
-		cleanup    cleanupFuncStack
+		cleanup    cleanupFuncQueue
 	}
 
 	resumableFlow struct {
@@ -307,7 +312,7 @@ type portalPauseInfo struct {
 		// We need this as when re-executing the portal, we are reusing the flow
 		// with the new receiver, but not re-generating the physical plan.
 		outputTypes []*types.T
-		cleanup     cleanupFuncStack
+		cleanup     cleanupFuncQueue
 	}
 }