distsql: elide RowChannel when connecting local processors

[petermattis]

RowChannel is currently used for connecting local distsql Processors. A RowChannel is essentially a channel with a fixed size. #19288 identified RowChannel as a throughput bottleneck in distsql processing.

RowChannel implements both the RowSource and RowReceiver interfaces. Each Processor is run within a separate goroutine and the Processor.Run method loops, pulling rows from its inputs (RowSource), processing them and emitting them via ProcOutputHelper to a RowReceiver.

With a modest amount of refactoring, I think this could be restructured by having processor implementations also implement the RowSource interface. The internals of such a processor would be reorganized so that Next() would pull from its input(s), process and return the next row. This would be akin to the planNode interface. A processor which implements RowSource would either be run via a call to Run(), in which case Next() would never be called, or zero or more calls to Next(). That is, a processor would either act as a RowSource or as a Processor within a given flow, never both.

We can incrementally move towards this new API. Currently, Flow.setup() always joins processors together with a RowChannel or MultiplexedRowChannel. If a processor implements RowSource, creation of the RowChannel can be elided. We would have to mark processors joined in this way and only call Processor.Run on processors that are not acting as RowSources.

See also #19134 which mentions "fusing" multiple processors into a single goroutine. The above proposal would achieve this.

This proposal needs to be validated by benchmarking before significant work is done. There are undoubtedly complexities with regards to cancellation and the particulars of the RowSource interface that will make this challenging.

@arjunravinarayan, @asubiotto You've likely already been thinking in this direction, but I couldn't find an issue discussing this.

[rjnn]

@andreimatei brought up in #20584 the idea of "synchronous scheduling of co-located processors work". I'm not entirely sure what that means, but it seems to be an alternate design to this. I'd like to see it spelled out more explicitly.

My thoughts: two reasons I like the design outlined in this issue to maintaining/improving the current design using different goroutines for different processors is that:

1. direct function calls from one processor to the next have great cache locality.
2. When we know for sure that two processors are one after the other, we can use a single buffer to pass a pre-allocated batch of rows so that we don't create extra garbage when passing data between them.
3. in the longer run, we want to fuse the processors by JITting them into a single pipeline, to avoid interface indirection. The performance benefits of this is not as important when we already have row batches, but in the limit case JITting relatively small batches has the best cache locality[1]. JITting is probably not something we will do in 2018, but it bears keeping in mind what the optimal design looks like according to the fastest dataflow designs out there.

We definitely need benchmarks to convince anyone how direct function calls compare to using goroutines for separate processors. @danhhz has already done some preliminary work in this area, I'd also appreciate his thoughts here.

[1]: http://db.csail.mit.edu/pubs/abadi-column-stores.pdf section 4.1

[petermattis]

@arjunravinarayan I'd like to reiterate that fusing distsql processors together via JIT'ing is definitely not going to happen in 2018. There are lots of bits of lower hanging performance fruit to pluck first. And we also need to get distsql to feature parity with local sql. Definitely worthwhile to keep such fusing in mind, but I want to make sure any other spectators don't get too far ahead of the work immediately ahead.

[petermattis]

After a day spent looking at profiles of distsql processors, it is clear that the lowest hanging fruit to pluck other than the overhead of RowChannel, is to eliminate allocations. As @arjunravinarayan mentioned above, one of the advantages of eliding RowChannel is that we can avoid allocations between processors.

Once processors implement RowSource, another smaller opportunity is to elide the "post-processing" steps if they are not present. The filter/limit/projection post-processing could be implemented via separate nodes that also model RowSource. Experimentation would be necessary to determine if this is beneficial or not.

[jordanlewis]

Can this be closed, or are we waiting for a full solution?