ARROW-11883: [C++] Add ConcatMap, MergeMap, and an async-reentrant version of Map

[westonpace]

These items can all stand on their own and they are used by the async datasets conversion.

MergeMap - Given AsyncGenerator<AsyncGenerator> return AsyncGenerator. This method flattens a generator of generators into a generator of items. It may reorder the items.

ConcatMap - Same as MergeMap but it will only pull items from one inner subscription at a time. This reduced parallelism allows items to be returned in-order.

Async-reentrant Map - In some cases the map function is slow. Even if the source is not async-reentrant this map can still be async-reentrant by allowing multiple instances of the map function to run at once. The resulting mapped generator is async reentrant but it will not pull reentrantly from the source.

Vector utilities - In order to make migrating from Iterator code to vector code easier I added some map style utilities. These copy the vectors (where an iterator wouldn't) so some care should be taken but they can still be useful.

Moved Future/AsyncGenerator into top level type_fwd. This is needed for the RecordBatchGenerator alias in the same way Iterator is needed at the top level.

Added IsEnd to IterationTraits. This allows non-comparable types to be iterated on. It allows us to create an AsyncGenerator<AsyncGenerator> since AsyncGenerator is std::function and we can use an empty instance as an end token even though std::function is not comaprable.

[github-actions]

https://issues.apache.org/jira/browse/ARROW-11883

[lidavidm]

Is this max readahead or max subscriptions (as the implementation itself calls it?)

[westonpace]

Max subscriptions. I fixed up the naming. Maybe it's not the best name as subscriptions comes from Rx where there are actual "subscriptions" but the concept is pretty much the same.

[lidavidm]

LGTM. One comment about the SerialReadaheadGenerator, which is probably a misunderstanding on my part.

[lidavidm]

Hmm. In the case that max_readahead == 1, the first Callback will run and won't enqueue anything. Then the next time the caller gets a future from the generator, the readahead queue will be empty, even though the generator isn't finished.

But if we instead had last_available > 0, then we'd accidentally make spaces_available negative when the callback runs. I think we have to check last_available before decrementing it, which should be okay since it's not accessed concurrently. Not only that, but the check and decrement should be done in Pump(), which is what actually adds to the queue - otherwise you'd have the same problem again when operator() calls Pump().

[lidavidm]

And as written, max_readahead > 1 is safe, but will enqueue one fewer item than declared - maybe this is all intentional and I'm misunderstanding the intent here, given the SerialReadaheadSlowConsumer test below which explicitly checks the current behavior (num_delivered == 3 whereas I would expect 4). But if that's the case maybe we should enforce that max_readahead > 1, or add 1 to the given value.

[westonpace]

There might be an off-by-1 here that I'll test out tomorrow. I agree that max_readahead == 1 should mean the first callback pumps. It's maybe as simple as starting spaces_available with max_readahead + 1 (in which case spaces_available would be interpreted as the slots in the queue as well as the last item we returned to the caller, in other words, all requests currently in "flight"). This means max_readahead = 1 would allow you to have 2 requests in flight at once but I think that is what I would expect.

However, for your second paragraph, spaces_available IS accessed concurrently. It can be accessed by the caller (the end consuming the generator) as well as the producer (the end marking the futures complete and running the callbacks/continuations). I'm pretty sure it is correct that the caller end operator()() does not decrement spaces_available.

[westonpace]

Good catch. It was indeed an off-by-one. There were two issues, spaces_available was not accounting for the "in-flight" request and ProducerConsumerQueue actually has size-1 "usable" slots (which is a little misleading, size refers to the size in memory and not the functional size).

[pitrou]

This seems to be based on another PR?

[bkietz]

Is it worth maintaining arrow/util/type_fwd.h? Maybe it can be inlined into arrow/type_fwd.h. In any case, it can wait for follow up

[pitrou]

Just my 2 cents, but for readability it seems better not to dump everything in a single header.

[bkietz]

Suggested change

	Future<>AwaitAsyncGenerator(AsyncGenerator<T>generator) {
	Future<>DiscardAllFromAsyncGenerator(AsyncGenerator<T>generator) {

[westonpace]

I'm not the biggest fan of discard either as it implies to me the generator is doing nothing but I agree that "await" isn't very clear. I've switched to discard. This is used in the Scanner::ToTable method because the callbacks are side-effecting a table builder and after they've all run we can get a table out of the builder.

[bkietz]

LGTM

[bkietz]

@westonpace could you resolve merge conflicts?

[westonpace]

@bkietz Rebased

[westonpace]

These failures are from the Github Actions blackout. The rest of the jobs seem to be stuck. Feel free to restart them. I fear adding more commits would just add fuel to the issue.

[pitrou]

Just FTR, perhaps all_good.emplace_back(MoveOnlyDataType(1)) works?

[pitrou]

I've opened westonpace#6 for this.

[westonpace]

Thanks! I've moved this into ARROW-11998 and #9734 and converted this from a FIXME to a TODO.

[bkietz]

LGTM, thanks!

[pitrou]

+1, just one comment

[westonpace]

Assuming remaining jobs pass (Travis looks like it is backlogged ~12 hours) or are unrelated, please merge.