GH-47385: Create BackpressureSource and BackpressureCombiner

[gitmodimo]

Rationale for this change

Provide infrastructure for unified backpressure control.

What changes are included in this PR?

BackpressureController is moved from asof_join and sorted_merge to common backpressure.
BackpressureCombiner with BackpressureCombiner::Source is added to handle multiple pause sources.

Are these changes tested?

Test is added.

Are there any user-facing changes?

New api is available

• GitHub Issue: [C++][Acero] Create BackpressureSource and BackpressureCombiner #47385

[github-actions]

⚠️ GitHub issue #47385 has been automatically assigned in GitHub to PR creator.

[github-actions]

⚠️ GitHub issue #47385 has no components, please add labels for components.

[gitmodimo]

I will simplify pause logic. It will be easier to reason with simple but composable pause propagation logic.

[zanmato1984]

Some minor suggestions regarding to formats. Now looking into more details.

[zanmato1984]

I think I get the idea of bridging multiple backpressure sources and propagating the backpressure between, in an implicit and automatic manner. But before we introduce this as a public API, I'd like to see a more concrete use case, e.g. a plan tree (possibly a hypothetical one with some new node types you are about to add in the future), to showcase some unclear things such as:

1. Who owns the life span of the combiner and the combiner sources?
2. When and where to establish the bridge between a combiner source and a particular combiner?
3. When and where to trigger a combiner source's Pause() or Resume()?

Current BackpressureCombiner exposes too much implementation detail as a public interface, e.g. we even know it uses std::mutex to manage synchronization. But this is another topic that we may talk about later.

[gitmodimo]

That is reasonable but I am not sure how to approach this. I was trying to keep the PRs relatively small. To showcase use of this component I would need Pipes that enable muli-output nodes. I can either add Pipe to this PR or update rebase Pipe draft to use that uses BackpressureCombiner internally. Also to think it would be relatively easy to showcase filter node with optional output providing filtered-out data. LMK what you think is best approach.

[zanmato1984]

Separation is good, you don't have to combine any PRs. I think you can just sketch the idea in text, if there isn't much trouble. (And I briefly viewed the Pipe PR and I get the idea.)

Thanks.