Add BufferExec execution plan

[gabotechs]

Which issue does this PR close?

• Closes #.

Rationale for this change

This is a PR from a batch of PRs that attempt to improve performance in hash joins:

• Do not require mut in memory reservation methods #19759
• This PR
• Hash join buffering on probe side #19761

It adds a building block that allows eagerly collecting data on the probe side of a hash join before the build side is finished.

Even if the intended use case is for hash joins, the new execution node is generic and is designed to work anywhere in the plan.

What changes are included in this PR?

Note

The new BufferExec node introduced in this PR is still not wired up automatically

Adds a new BufferExec node that can buffer up to a certain size in bytes for each partition eagerly performing work that otherwise would be delayed.

Schematically, it looks like this:

             ┌───────────────────────────┐
             │        BufferExec         │
             │                           │
             │┌────── Partition 0 ──────┐│
             ││            ┌────┐ ┌────┐││       ┌────┐
 ──background poll────────▶│    │ │    ├┼┼───────▶    │
             ││            └────┘ └────┘││       └────┘
             │└─────────────────────────┘│
             │┌────── Partition 1 ──────┐│
             ││     ┌────┐ ┌────┐ ┌────┐││       ┌────┐
 ──background poll─▶│    │ │    │ │    ├┼┼───────▶    │
             ││     └────┘ └────┘ └────┘││       └────┘
             │└─────────────────────────┘│
             │                           │
             │           ...             │
             │                           │
             │┌────── Partition N ──────┐│
             ││                   ┌────┐││       ┌────┐
 ──background poll───────────────▶│    ├┼┼───────▶    │
             ││                   └────┘││       └────┘
             │└─────────────────────────┘│
             └───────────────────────────┘

Are these changes tested?

yes, by new unit tests

Are there any user-facing changes?

users can import a new BufferExec execution plan in their codebase, but no internal usage is shipped yet in this PR.

[adriangb]

This seems in many ways quite similar to what RepartitionExec w/ spilling does. Have you had a chance to poke at that code? I wonder if we could share the code and give this operator spilling capabilities at the same time (would help with large batches? buffering on disk if there are no waiters on the other side?).

I'm curious to get @alamb's thoughts on this vs. his work on buffering / prefetching at the Parquet level. The advantage I see of buffering at the Parquet level is that the reader can do fancy things like planning to fetch a larger contiguous chunk of data from object storage. Maybe both are needed though? As in: you want buffering and prefetching.

[gabotechs]

This seems in many ways quite similar to what RepartitionExec w/ spilling does. Have you had a chance to poke at that code?

Yes, in fact a small chunk of the code there still shows my name in the git blames. It do is similar in the sense that there is some per-partition buffering, but it looks like that code is in a more difficult situation, as it needs to be able to buffer potentially indefinitely due to the unbounded nature of RepartitionExec (correct me if I'm wrong, it's been a while since I looked at that code), whether the code in this PR can afford to have bounded channels.

At first sight I do not see a lot of opportunities for reusing code in both places due to the different requirements, but happy to listen to ideas.

Maybe both are needed though? As in: you want buffering and prefetching.

Another difference with RepartitionExec is that BufferExec will eagerly poll its children regardless of whether its stream was polled or not, and RepartitionExec will wait for the first poll to start doing work. This means that RepartitionExec does not prefetch, but BufferExec does

The advantage I see of buffering at the Parquet level is that the reader can do fancy things like planning to fetch a larger contiguous chunk of data from object storage

👍 I can see this being beneficial. My intention was to first use this in #19761, but the BufferExec node is something you are supposed to be able to place wherever you want. In fact, we do use it in more scenarios at DataDog.