distsql: change the contract of Next to minimize wasteful allocations and copies

[jordanlewis]

Currently, the contract of RowSource.Next requires that it return rows that are usable forever. This requires a lot of extra allocations and copies in ProcOutputHelper that may not be necessary.

It would be better to allow Next implementations to return containers that can be overwritten by the next call to Next, the way planNode.Next and many other components work. This would force changes in processors that need to retain references to the values they get from their inputs - those processors would need to start making copies of those values.

One complication is that these defensive copies themselves may not always be necessary, depending on the RowSource a processor is reading from. For example, if the RowSource is a RowChannel, the rows returned from Next() will not be ephemeral and therefore do not be copied.

To resolve this last situation, a RowSource should be queryable to determine whether or not it has this extra property of allowing Next values to be held on to. This could be a method on RowSource or just a big type switch in a function somewhere.

Ultimately, this change will allow processors to allocate their output EncDatumRow just once up front, and copy values into that output container when returning from Next. This will massively reduce allocations done per processor, shifting that allocation burden when strictly necessary, and removing it altogether otherwise.

[asubiotto]

I think there's an issue with this when using any sort of asynchronous communication between processors as described in #22462. Defensive copies won't solve this issue because if the upstream doesn't create a copy, it could overwrite rows before the downstream processor has had a chance to read & copy.

I think that this is a good idea in the general case but it's inevitable that we determine whether a processor is running asynchronously to others at plan time and swap out its ProcOutputHelper for one that allocates.

[petermattis]

My thinking is that Next as implemented by processors should never make a copy and we should create a CopyingRowSource that wraps around another RowSource and copies the row.

(I think this is similar to what @asubiotto is describing)

[jordanlewis]

Defensive copies won't solve this issue because if the upstream doesn't create a copy, it could overwrite rows before the downstream processor has had a chance to read & copy.

Such an upstream processor would need to do the copy ahead of time. Downstream processors would be informed of this special case by inspecting the upstream processor, like I was trying to describe in my top level comment. Am I still missing something?

As a strawman, let's say RowSource gets a new method bool SafeNext(). At init-time, processors will check all of their upstreams returned SafeNext() value. Rows returned from upstreams that have an unsafe next will be copied if they need to be retained.

My thinking is that Next as implemented by processors should never make a copy and we should create a CopyingRowSource that wraps around another RowSource and copies the row.

Yes, that could work.

[jordanlewis]

#24589 is a stab at fixing this.