sql: enable resumption of a flow for pausable portals

[ZhouXing19]

This PR is part of the implementation of multiple active portals. (Extracted from #96358)

We now introduce a Resume() method for flow, and when a pausable portal is being re-executed, rather than generating a new flow, we resume the persisted flow to continue the previous execution.

---

sql: add telemetry MultipleActivePortalCounter

This commit added a telemetry counter MultipleActivePortalCounter that would
be incremented each time the cluster setting
sql.pgwire.multiple_active_portals.enabled is set to true

---

sql: add Resume method for flowinfra.Flow and execinfra.Processor

For pausable portals, each execution needs to resume the processor with new
output receiver. We don't need to restart the processors, and this Resume()
step can be called many times after Run() is called.

---

sql: reuse flow for pausable portal

To execute portals in an interleaving manner, we need to persist the flow and
queryID so that we can continue fetching the result when we come back to the same
portal.

We now introduce pauseInfo field in sql.PreparedPortal that stores this
metadata. It's set during the first-time execution of an engine, and all
following executions will reuse the flow and the queryID. This also implies that
these resources should not be cleaned up with the end of each execution.
Implementation for the clean-up steps is included in the next commit.

Also, in this commit we hang a *PreparedPortal to the planner, and how it is
set can be seen in the next commit as well.

Epic: CRDB-17622

Release note: None

[cockroach-teamcity]

This change is 

[ZhouXing19]

@yuzefovich since vectorizedFlow.Resume() will always call f.FlowBase.Resume(recv), maybe we don't need to have resumeCtx for vectorizedFlow?

[yuzefovich]

I have a few more comments, and it seems like the linter is unhappy about some of the unused stuff.

since vectorizedFlow.Resume() will always call f.FlowBase.Resume(recv), maybe we don't need to have resumeCtx for vectorizedFlow?

Yeah, sounds good.

Reviewed 3 of 3 files at r1, 11 of 11 files at r2, 6 of 6 files at r3, all commit messages.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @ZhouXing19)

---

pkg/sql/conn_executor.go line 1077 at r1 (raw file):

	ex.extraTxnState.createdSequences = make(map[descpb.ID]struct{})

	enableMultipleActivePortals.SetOnChange(&s.cfg.Settings.SV, func(ctx context.Context) {

I know I was the one who suggested to put this into newConnExecutor, but this is incorrect. newConnExecutor is called for each new session which results in adding a new callback to this cluster setting for each session, and we don't want that. We want to add a callback only once per SQL server, so ideally we'd do so when evaluating server.newSQLServer.

Perhaps we should define a new exported method next to where enableMultipleActivePortals is defined, and that method would be called by server.newSQLServer to add the callback. Another alternative is to export enableMultipleActivePortals itself so that we'd call SetOnChange directly from server/server_sql.go. Probably I like the latter option the most.

---

pkg/sql/colflow/vectorized_flow.go line 299 at r2 (raw file):

				Err: errors.AssertionFailedf(
					"batchFlowCoordinator should be nil for vectorizedFlow",
				)})

nit: we should call recv.ProducerDone here.

---

pkg/sql/physicalplan/aggregator_funcs_test.go line 87 at r3 (raw file):

		t.Fatal(err)
	}
	flow.Run(ctx, false /* noWait */)

nit: this change belongs in the second commit.

[ZhouXing19]

the linter is unhappy about some of the unused stuff.

I removed fields that are unused for now.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @yuzefovich)

---

pkg/sql/conn_executor.go line 1077 at r1 (raw file):

Previously, yuzefovich (Yahor Yuzefovich) wrote…

I know I was the one who suggested to put this into newConnExecutor, but this is incorrect. newConnExecutor is called for each new session which results in adding a new callback to this cluster setting for each session, and we don't want that. We want to add a callback only once per SQL server, so ideally we'd do so when evaluating server.newSQLServer.

Perhaps we should define a new exported method next to where enableMultipleActivePortals is defined, and that method would be called by server.newSQLServer to add the callback. Another alternative is to export enableMultipleActivePortals itself so that we'd call SetOnChange directly from server/server_sql.go. Probably I like the latter option the most.

Yup, I took the export solution.

[yuzefovich]

Also, we'll need to backport this to 23.1, so add the corresponding label and then request an approval in release-backports slack channel from Rafi (I'm not a TL).

Reviewed 20 of 20 files at r4, 13 of 13 files at r5, 6 of 6 files at r6, all commit messages.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @ZhouXing19)

[ZhouXing19]

TFTR!!

bors r=yuzefovich

[craig]

Build succeeded:

• Bazel Essential CI (Cockroach)

[blathers-crl]

Encountered an error creating backports. Some common things that can go wrong:

1. The backport branch might have already existed.
2. There was a merge conflict.
3. The backport branch contained merge commits.

You might need to create your backport manually using the backport tool.

---

Backport to branch 23.1.x failed. See errors above.

---

_{🦉 Hoot! I am a Blathers, a bot for CockroachDB. My owner is dev-inf.}