Future plans for Acero

[gitmodimo]

I would like to start a broad conversation on future of Acero development.
I have been using Acero for quite some time now and I think I have general understanding of current state. I do not use substrait nor python. I am strictly using Acero for streaming execution and I found Acero to be well designed and thought through. At first used concepts were overwhelming but later on I found them all useful and powerful. However along the way I spotted some hiccups and with my colleagues we fixed few of them.

With all my experience with Acero I came to conclusion that Acero needs some core changes in order to remain versatile and extensible library for streaming execution. This is general idea of this discussion. If/How such core changes can be introduced. I would like to split the discussion into few distinct topics.
In my understanding Acero was initially designed to execute queries on arrow datasets.  Usually queries are not executed in ordered fashion - ordering is optionally added as final step (hence OrderBySinkNode). As such Acero did not tackle source ordering. Later on additional concept of batch.index was introduced that paved the way for maintaining and leveraging source ordering (Topic 1).
All queries that I know of have only one output and as expected all ExecNodes have only one output. I am aware that Acero foundation does not prohibit multiple outputs in general, but as of now there are no multi-output nodes. (Topic 2).
With current state of Acero few coding patters occur. I think they should be considered for factoring out do remove code duplication and to simplify amintenance (Topic 3).

1. Ordering/Backpressure

Since introduction of batch.index not all exec nodes comply to this new semantics - even though multiple exec nodes do realy on data order. Most notable:
-asof_join - ordering done GH-41706. Backpressure pending merge GH-46421
-sorted_merge - PR ready GH-47269 not ready
-aggregate - PR ready GH-47269 - ordering needed conditionally not ready
In addition to those nodes all source and sink nodes need to account now ordering concept and user intention to maintain ordering of the source.

Solution to all those nodes that require ordering was to introduce SerialSequencingQueue. Although it fixes ordering it unfortunately breaks backpressure (SerialSequencingQueue does not limit how many items are queued). To fix backpressure SerialSequencingQueue has to produce its own pause signal and also propagate pause from downstream. I think the logic of this becomes to convoluted to replicate it in every ExecNode. So since ordering is now a global concept I think we should move validation_of_ordering + sequencing + backpressure logic out of specific ExecNodes and into ExecNode base. This would let implementer of new exec node focus on actual data processing and use already implemented access patterns of inputs and outputs, that have already emerged.

As extra feature Ordering could offer additional “stream” guarantees. Stream guarantee would hold condition that is guaranteed to be true for the rest of data stream (like “timestamp>x). This could be used to push timeline/segment in order leveraging execnodes.

2. Multiple outputs

Multiple outputs ExecNodes are mentioned in sevaral placed across documentation and issues, but no implementation of such node was ever implemented. In my application I found it neccesary to produce multiple outputs from single processing pipeline. In the beginning dataset tee node was enough, but along the way it turned out to be not flexible enough. A little bit inspired by implementation of tee node I created new pipe concept that fits Acero quite well and provides quite elegant alternative to multiple output nodes. In summary there are three new nodes:
-pipe_sink - node consumes all exec batches and replicates them to all pipe_source nodes
-pipe_source- is a source node that receives batches.
-pipe_tee - node replicates batches to pipe_source and output
All pipe nodes have names and sinks are connected by name with sources at init stage. Additionally pipe can be instantiated as an element in exec node to provide additional outputs (for example “filter” node can provide additional output of filtered out data). I have been using this concept in my processing pipeline extensively and now I completely stopped using tee node.
The reason why I think pipes are strong alternative to multi-output nodes is that pipes fit elegantly with entire Declaration infrastructure. With single output declaration always is a tree. Multiple outputs would create a directed graph that requires changing literary entire ExecPlan building infrastructure. With pipes we get effectively the same functionality that fits current ExecPlan building process.

3. Refactoring

I find several changes that would benefit Acero as a whole:

• Factor out handling of input. All not source exec nodes implement the same input_counter_ finish logic. I propose we move this to base ExecNode.
• Create ExecNodeInputAdapters that would sequence input when needed with two different access patterns push/pull based.
• Unify and move backpressure handling into input adapters
• Factor out outputing of ExecBatch into ExecNodeOutputAdapter.
• Factor out StopProducing and handle stream InputFinished logic ExecNodeOutputAdapter

I am not a maintainer of Arrow nor I am affiliated with Apache in any way, but I hope this discussion produces some kind of general roadmap for future development of Acero. All this might seem like extreme makeover but I honestly believe this is shortest path for fixing the current issues with backpressure and as a bonus really cleanup and simplify Acero codebase. I am ready to invest some time into it, but first I want to know whether this complies with maintainers plans and vision.

[kou]

Thanks for the proposal!

@zanmato1984 What do you think about this proposal?

[zanmato1984]

I think this is a thoughtful and reasonable proposal. Thanks for putting it all together! It’s also delightful to see people actively using Acero “the raw way,” and I really appreciate the contributions you and your colleagues have already made.

While Acero isn’t intended to be a cutting-edge query engine competing with systems like DuckDB or Velox, what we value most is enablement: giving users the ability to compose, orchestrate, and integrate Arrow-based data processing in customizable ways. From that perspective, I see your proposals as strengthening this core purpose, and none of them seem in conflict with it.

I’d be glad to help continue the discussion or review individual follow-up issues/PRs. I’ve noticed some of the ongoing work already (the pipe node family idea in particular is really cool!). That said, I do need to prioritize my engagement based on factors like criticality, author enthusiasm, PR size, and my familiarity with the relative code path. So please don’t hesitate to ping me directly if there’s something you’d like me to prioritize.

Overall, I really appreciate you bringing these ideas forward. It’s great to see thoughtful proposals like this shaping Acero’s future.

[pitrou]

I would be against any sort of long-lived branch that would end up in a monster PR with tens of thousands of added/changed lines.

[zanmato1984]

In Arrow repo, we tend to keep each PR reasonably sized and self-contained, and then merge them separately. (Besides, I don't think your proposed items require any large, long-lived, monolithic PRs.)

The issue filing is flexible. But make sure you have one issue for each PR. Use umbrella issue and sub-issue if you want them better organized or grouped, as Kou suggested.

[gitmodimo]

Thanks for the tips! I will start creating issues then.

[gitmodimo]

For some reason I cannot create subissues. @zanmato1984 can this be due to limited permissions?
#47383

[kou]

Oh, sorry. I'll set subissue metadata later. Please continue with the current style.

[gitmodimo]

I have created umbrella Issue #47383 and multiple subissues. Thank you @kou @pitrou @zanmato1984 for the help. Can you somehow expedite review of PRs #47386 and #47392. I really want to keep the momentum going and those two are blockers.

[zanmato1984]

Cool, I can help reviewing them shortly.

[gitmodimo]

I am path finding best way to unify access patterns of different ExecNodes. One important question arises that I am not sure of the answer. Is it legal to submit blocking task to io_executor? With blocking I mean waiting on condition to become true (not blocking as in waiting for read to complete).
Consider potential use case in asof_join:
InputReceived on input 0(left) blocks and waits until sufficient data is available on inputs 1-N(right). Current asof_join implementation uses additional thread for blocking wait. Is it due to blocking on io_executor is illegal or it is just implementation choice?

[zanmato1984]

It seems problematic if we put the the outstanding thread of asof join into any of the executor, but at this point it's hard to guess the intention of the original author.

But the general idea is as @pitrou said, there shouldn't be any kind of waiting (on things that are not guaranteed to happen) in an executor task. As this kind of waiting can easily cause deadlock within a limited number of threads.

[gitmodimo]

Got it! Thank you @zanmato1984 @pitrou.
According to Building Arrow C++ C++17 is minimal required version. Does that mean C++20 coroutines are off the table in foreseeable future?

[pitrou]

Even if we move to C++20 (which we plan to do at some point), there is no guarantee that all supported platforms would support coroutines anyway (not to mention potential compiler bugs).

So, yeah, definitely off the table IMHO.

[westonpace]

The sidecar thread in the asof join node was an implementation choice by the original author. It's possible to get rid of it but you would need to do a bit of work to refactor the asof join node implementation so that work is done by the push.

It's been a long while since I looked at it but I think the logic is basically that each side pushes its data into a queue and then the sidecar thread is constantly waking up to see if there is enough data to form rows and, if so, creates and publishes those rows.

You could have a solution where the work done by the sidecar thread is done after the batch is pushed into the queue.

[gitmodimo]

You could have a solution where the work done by the sidecar thread is done after the batch is pushed into the queue.

I think it is already there when ARROW_ENABLE_THREADING is not defined. Probably with some synchronization this would work with threading.