Notable change is that now the materializer created when wrapping
row-execution processors are no longer added to the set of releasables
at the flow cleanup because in some cases it could be released before
being closed.

Is this a problem? What are the effects of this? Is this different from Closers?

Reviewed 57 of 57 files at r5, 1 of 1 files at r6, 13 of 13 files at r7, 8 of 8 files at r8, 9 of 9 files at r9, 9 of 9 files at r10, 1 of 1 files at r11, 11 of 11 files at r12, 48 of 48 files at r13.
Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @yuzefovich)

pkg/sql/conn_executor_internal_test.go, line 259 at r13 (raw file):

	distSQLMetrics := execinfra.MakeDistSQLMetrics(time.Hour /* histogramWindow */)
	gw := gossip.MakeOptionalGossip(nil)
	tempEngine, tempFS, err := storage.NewTempEngine(ctx, base.DefaultTestTempStorageConfig(st), base.DefaultTestStoreSpec)

Is this an in-mem engine? See storage.NewDefaultInMem if not. Also, why did you have to add this?

pkg/sql/colexec/columnarizer.go, line 37 at r12 (raw file):

	// phase.
	ColumnarizerDefaultMode ColumnarizerMode = iota
	// ColumnarizerStreamingMode is the mode of operation in which the

Why does making the columnarizer implement the streaming mode solve any problems? Isn't it an issue if we, say, have another operator reading from the columnarizer and it's a buffering processor? Or is it ok for those types of flows to exist since the equivalent row execution flow will do the same.

pkg/sql/colexec/columnarizer.go, line 66 at r12 (raw file):

// NewColumnarizer returns a new Columnarizer.
func NewColumnarizer(

Might be nicer to "hide" thismode option to avoid having to change always pass in a mode if we mostly want default behavior:

NewBufferingColumnarizer(opts...) {
    return newColumnarizer(opts, DefaultMode)
}

NewStreamingColumnarizer(opts...) {
    return newColumnarizer(opts, StreamingMode)
}

newColumnarizer(opts..., mode mode) {

}

Also, maybe we should rename ColumnarizerDetaultMode to ColumnarizerBufferingMode since it's more descriptive? (i.e. it avoids the question of what is the default mode?)

Notable change is that now the materializer created when wrapping
row-execution processors are no longer added to the set of releasables
at the flow cleanup because in some cases it could be released before
being closed.

Is this a problem? What are the effects of this? Is this different from Closers?

I don't think it is a problem - the code that closes the planNode tree is a bit broken in a sense that we cannot close the subquery's tree until after the main query runs, and releasing the materializer that wraps a LocalPlanNode is the only - I think - special case that outlives the flow that it was created by (as a reminder, we're putting all of the objects into their pools on the flow shutdown). So what happens is that subquery's flow is cleaned up yet the planNodes end up holding onto those materializers (via other adapters - planNodeToRowSource).

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto and @jordanlewis)

pkg/sql/conn_executor_internal_test.go, line 259 at r13 (raw file):

TestPortalsDestroyedOnTxnFinish fails with error setting up flow: FS unset on DiskQueueCfg, so we actually need to set the file system for all vectorized flows given that in vectorizedFlow.Setup we always create a DiskQueueCfg.

pkg/sql/colexec/columnarizer.go, line 37 at r12 (raw file):

I think there might be only one very special planNode that requires this streaming behavior - hookFnNode. In its Next method it waits for the context cancellation or an error for shutting down or for the next row to be sent on the channel. The buffering behavior of the columnarizer can block indefinitely long - for example, in case of the changefeeds.

In TestChangefeedBasics/sinkless if we make the columnarizer buffering, we observe the following situation: there are 5 total changes on the table, so 5 events are emitted; the columnarizer puts the 1st into the first batch, then 2nd and 3rd into the second batch, then 4th and 5th into the third batch of 4 capacity, and the columnarizer asks for more rows, but they never arrive.

At the moment, I believe making a "local" decision whether to make the columnarizer streaming or not works in all cases - AFAIK these hookFnNodes are the only things that could block indefinitely when Next is called, so we simply wanna make sure to not buffer anything that is coming out of them. I'm guessing that changefeeds might be the only special case, so we could customize the columnarizer to be streaming only on hookFnNodes or even more specifically on changefeed hookFnNode, but I think it's worth being on the safer side and make all LocalPlanNode cores wrapped with the streaming columnarizer.

I cannot say that I have 100% confidence in this approach, but I haven't encountered any other special cases apart from the changefeeds. I'm assuming that if there is something that I'm missing, it'll be exposed by the nightly and unit tests, especially since we are now randomizing coldata.BatchSize.

To test my theory about changefeed flow being the only special case, I'll open up a separate PR and see whether anything else fails.

pkg/sql/colexec/columnarizer.go, line 66 at r12 (raw file):

I initially also had "buffering" and "streaming" modes, but I like this suggestion. Done.

pkg/sql/sem/tree/casts.go, line 814 at r4 (raw file):

Done.

Reviewable status: complete! 0 of 0 LGTMs obtained (waiting on @asubiotto and @jordanlewis)

pkg/sql/colexec/columnarizer.go, line 37 at r12 (raw file):

Looks like my hypothesis might be correct since this build on a testing PR has only a couple of expected flakes (some linter issues and a benchmark data that needs to be updated in this PR), so I have a bit more confidence now in the current approach (of "local" decision on the mode of the columnarizer).

Reviewed 57 of 57 files at r14, 11 of 11 files at r15, 1 of 54 files at r16.
Reviewable status: complete! 1 of 0 LGTMs obtained (waiting on @asubiotto, @jordanlewis, and @yuzefovich)

pkg/sql/conn_executor_internal_test.go, line 259 at r13 (raw file):

👍 what about the in-mem part?

TFTR!

bors r+

Reviewable status: complete! 0 of 0 LGTMs obtained (and 1 stale) (waiting on @asubiotto and @jordanlewis)

pkg/sql/conn_executor_internal_test.go, line 259 at r13 (raw file):

The current invocation creates an in-memory engine and FS. Using NewDefaultInMem is not sufficient since it only returns the engine and not the FS, but we're only interested in the FS.