ExecNode::StartProducing()#

This method is called once at the start of the plan. Most nodes ignore this method (anynecessary initialization should happen in the constructor or Init). However, source nodeswill typically provide a custom implementation. Source nodes should schedule whatever tasksare needed to start reading and providing the data. Source nodes are usually the primarycreator of tasks in a plan.

ExecNode::InputReceived()#

This method is called many times during the execution of a plan. It is how nodes pass datato each other. An input node will call InputReceived on its output. Acero’s execution modelis push-based. Each node pushes data into its output by calling InputReceived and passing ina batch of data.

The InputReceived method is often where the actual work happens for the node. For example,a project node will execute its expressions and create a new expanded output batch. It will thencall InputReceived on its output. InputReceived will never be called on a source node. Sinknodes will never call InputReceived. All other nodes will experience both.

Some nodes (often called “pipeline breakers”) must accumulate input before they can generateany output. For example, a sort node must accumulate all input before it can sort the data andgenerate output. In these nodes the InputReceived method will typically place the data intosome kind of accumulation queue. If the node doesn’t have enough data to operate then it willnot call InputReceived. This will then be the end of the current task.

ExecNode::InputFinished()#

This method will be called once per input. A node will call InputFinished on its output once itknows how many batches it will be sending to that output. Normally this happens when the node isfinished working. For example, a scan node will call InputFinished once it has finished readingits files. However, it could call it earlier if it knows (maybe from file metadata) how manybatches will be created.

Some nodes will use this signal to trigger some processing. For example, a sort node need towait until it has received all of its input before it can sort the data. It relies on the InputFinishedcall to know this has happened.

Even if a node is not doing any special processing when finished (e.g. a project node or filter nodedoesn’t need to do any end-of-stream processing) that node will still call InputFinished on itsoutput.

ExecNode::PauseProducing() /ExecNode::ResumeProducing()#

These methods control backpressure. Some nodes may need to pause their input to avoid accumulatingtoo much data. For example, when the user is consuming the plan with a RecordBatchReader we use aSinkNode. The SinkNode places data in a queue that the RecordBatchReader pulls from (this is aconversion from a push-model to a pull-model). If the user is reading the RecordBatchReader slowly thenit is possible this queue will start to fill up. For another example we can consider the write node.This node writes data to a filesystem. If the writes are slow then data might accumulate at thewrite node. As a result, the write node would need to apply backpressure.

When a node realizes that it needs to apply some backpressure it will call PauseProducing on its input.Once the node has enough space to continue it will then call ResumeProducing on its input. For example,the SinkNode would pause when its queue gets too full. As the user continues to read from theRecordBatchReader we can expect the queue to slowly drain. Once the queue has drained enough then theSinkNode can call ResumeProducing.

Source nodes typically need to provide special behavior for PauseProducing and ResumeProducing. Forexample, a scan node that is reading from a file can pause reading the file. However, some source nodesmay not be able to pause in any meaningful way. There is not much point in a table source node pausingbecause its data is already in memory.

Nodes that are neither source or sink should still forward backpressure signals. For example, whenPauseProducing is called on a project node it should call PauseProducing on its input. If a node hasmultiple inputs then it should forward the signal to every input.

ExecNode::StopProducing()#

StopProducing is called when a plan needs to end early. This can happen because the user cancelledthe plan and it can happen because an error occurred. Most nodes do not need to do anything here.There is no expectation or requirement that a node sends any remaining data it has. Any node thatschedules tasks (e.g. a source node) should stop producing new data.

In addition to plan-wide cancellation, a node may call this method on its input if it has decidedthat it has received all the data that it needs. However, because of parallelism, a node may stillreceive a few calls toInputReceived after it has stopped its input.

If any external resources are used then cleanup should happen as part of this call.

Initialization / Construction / Destruction#

Simple initialization logic (that cannot error) can be done in the constructor. If the initializationlogic may return an invalid status then it can either be done in the exec node’s factory method ortheInit method. The factory method is preferred for simple validation. TheInit method ispreferred if the initialization might do expensive allocation or other resource consumption.Init willalways be called beforeStartProducing is called. Initialization could also be done inStartProducing but keep in mind that other nodes may have started by that point.

In addition, there is aValidate method that can be overloaded to provide custom validation. Thismethod is normally called beforeInit but after all inputs and outputs have been added.

Finalization happens today in the destructor. There are a few examples today where that might be slow.For example, in the write node, if there was an error during the plan, then we might close out some openfiles here. Should there be significant finalization that is either asynchronous or could potentiallytrigger an error then we could introduce a Finalize method to the ExecNode lifecycle. It hasn’t beendone yet only because it hasn’t been needed.

Summary#

Parallel Execution of Plans#

Users may want to execute multiple plans concurrently and they are welcome to do so. However, Acero has noconcept of inter-plan scheduling. Each plan will attempt to maximize its usage of compute resources andthere will likely be contention of CPU and memory and disk resources. If plans are using the default CPU &I/O thread pools this will be mitigated somewhat since they will share the same thread pool.

Locally Distributed Plans#

A common way to tackle multi-threading is to split the input into partitions and then create a plan foreach partition and then merge the results from these plans in some way. For example, let’s assume youhave 20 files and 10 cores and you want to read and sort all the data. You could create a plan for every2 files to read and sort those files. Then you could create one extra plan that takes the input from these10 child plans and merges the 10 input streams in a sorted fashion.

This approach is popular because it is how queries are distributed across multiple servers and so itis widely supported and well understood. Acero does not do this today but there is no reason to prevent it.Adding shuffle & partition nodes to Acero should be a high priority and would enable Acero to be used bydistributed systems. Once that has been done then it should be possible to do a local shuffle (localmeaning exchanging between multiple exec plan instances on a single system) if desired.

Pipeline Parallelism#

Acero attempts to maximize parallelism using pipeline parallelism. As each batch of data arrives from thesource we immediately create a task and start processing it. This means we will likely start processingbatch X before the processing of batch X-1 has completed. This is very flexible and powerful. However, it alsomeans that properly implementing an ExecNode is difficult.

For example, an ExecNode’s InputReceived method should be reentrant. In other words, it should be expectedthat InputReceived will be called before the previous call to InputReceived has completed. This means thatnodes with any kind of mutable state will need mutexes or similar mechanisms to protect that state from raceconditions. It also means that tasks can easily get out of order and nodes should not expect any particular orderingof their input (more on this later).

Asynchronicity#

Some operations take a long time and may not require the CPU. Reading data from the filesystem is one example. If weonly have one thread per core then time will be wasted while we wait for these operations to complete. Thereare two common solutions to this problem. A synchronous solution is often to create more threads than there arecores with the expectation that some of them will be blocked and that is ok. This approach tends to be simplerbut it can lead to excess thread contention and requires fine-tuning.

Another solution is to make the slow operations asynchronous. When the slow operation starts the caller gives upthe thread and allows other tasks to run in the meantime. Once the slow operation finishes then a new task iscreated to take the result and continue processing. This helps to minimize thread contention but tends to bemore complex to implement.

Due to a lack of standard C++ async APIs, Acero uses a combination of the two approaches. Acero has two thread pools.The first is the CPU thread pool. This thread pool has one thread per core. Tasks in this thread pool should neverblock (beyond minor delays for synchronization) and should generally be actively using CPU as much as possible. Threadson the I/O thread pool are expected to spend most of the time idle. They should avoid doing any CPU-intensive work.Their job is basically to wait for data to be available and schedule follow-up tasks on the CPU thread pool.

Task per Pipeline (and sometimes beyond)#

An engine could choose to create a thread task for every execution of a node. However, without careful scheduling,this leads to problems with cache locality. For example, let’s assume we have a basic plan consisting of threeexec nodes, scan, project, and then filter (this is a very common use case). Now let’s assume there are 100 batches.In a task-per-operator model we would have tasks like “Scan Batch 5”, “Project Batch 5”, and “Filter Batch 5”. Eachof those tasks is potentially going to access the same data. For example, maybe theproject andfilter nodes needto read the same column. A column which is intially created in a decode phase of thescan node. To maximize cacheutilization we would need to carefully schedule our tasks to ensure that all three of those tasks are run consecutivelyand assigned to the same CPU core.

To avoid this problem we design tasks that run through as many nodes as possible before the task ends. This sequenceof nodes is often referred to as a “pipeline” and the nodes that end the pipeline (and thus end the task) are oftencalled “pipeline breakers”. Some nodes might even fall somewhere in between. For example, in a hash join node, whenwe receive a batch on the probe side, and the hash table has been built, we do not need to end the task and instead keepon running. This means that tasks might sometimes end at the join node and might sometimes continue past the join node.

Thread Pools and Schedulers#

The CPU and I/O thread pools are a part of the core Arrow-C++ library. They contain a FIFO queue of tasks and willexecute them as a thread is available. For Acero we need additional capabilities. For this we use theAsyncTaskScheduler. In the simplest mode of operation the scheduler simply submits tasks to an underlying thread pool.However, it is also capable of creating sub-schedulers which can apply throttling, prioritization, and task tracking:

• A throttled scheduler associates a cost with each task. Tasks are only submitted to the underlying schedulerif there is room. If there is not then the tasks are placed in a queue. The write node uses a throttle of size1 to avoid reentrantly calling the dataset writer (the dataset writer does its own internal scheduling). A throttledscheduler can be manually paused and unpaused. When paused all tasks are queued and queued tasks will not be submittedeven if there is room. This can be useful in source nodes to implement PauseProducing and ResumeProducing.

• Priority can be applied to throttled schedulers to control the order in which queued tasks are submitted. Ifthere is room a task is submitted immediately (regardless of priority). However, if the throttle is full thenthe task is queued and subject to prioritization. The scan node throttles how many read requests it generatesand prioritizes reading a dataset in order, if possible.

• A task group can be used to keep track of a collection of tasks and run a finalization task when all of thetasks have completed. This is useful for fork-join style problems. The write node uses a task group to closea file once all outstanding write tasks for the file have completed.

There is research and examples out there for different ways to prioritize tasks in an execution engine. Acero has notyet had to address this problem. Let’s go through some common situations:

• Engines will often prioritize reading from the build side of a join node before reading from the probe side. Thiswould be more easily handled in Acero by applying backpressure.

• Another common use case is to control memory accumulation. Engines will prioritize tasks which are closer to thesink node in an effort to relieve memory pressure. However, Acero currently assumes that spilling will be addedat pipeline breakers and that memory usage in a plan will be more or less static (per core) and well below thelimits of the hardware. This might change if Acero needs to be used in an environment where there are many computeresources and limited memory (e.g. a GPU)

• Engines will often use work stealing algorithms to prioritize running tasks on the same core to improve cachelocality. However, since Acero uses a task-per-pipeline model there isn’t much lost opportunity for cacheparallelism that a scheduler could reclaim. Tasks only end when there is no more work that can be done with the data.

While there is not much prioritization in place in Acero today we do have the tools to apply it should we need to.

Intra-node Parallelism#

Some nodes can potentially exploit parallelism within a task. For example, in the scan node we can decodecolumns in parallel. In the hash join node, parallelism is sometimes exploited for complex tasks such asbuilding the hash table. This sort of parallelism is less common but not necessarily discouraged. Profiling shouldbe done first though to ensure that this extra parallelism will be helpful in your workload.

All Work Happens in Tasks#

All work in Acero happens as part of a task. When a plan is started the AsyncTaskScheduler is created and given aninitial task. This initial task calls StartProducing on the nodes. Tasks may schedule additional tasks. For example,source nodes will usually schedule tasks during the call to StartProducing. Pipeline breakers will often schedule taskswhen they have accumulated all the data they need. Once all tasks in a plan are finished then the plan is considereddone.

Some nodes use external threads. These threads must be registered as external tasks using the BeginExternalTask method.For example, the asof join node uses a dedicated processing thread to achieve serial execution. This dedicated threadis registered as an external task. External tasks should be avoided where possible because they require carefulhandling to avoid deadlock in error situations.

Public API#

The public API for Acero consists of Declaration and the various DeclarationToXyz methods. In addition theoptions classes for each node are part of the public API. However, nodes are extensible and so this API isextensible.

R (dplyr)#

Dplyr is an R library for programmatically building queries. The arrow-r package has dplyr bindings whichadapt the dplyr API to create Acero execution plans. In addition, there is a dplyr-substrait backend thatis in development which could eventually replace the Acero-aware binding.

Python#

The pyarrow library binds to Acero in two different ways. First, there is a direct binding in pyarrow.acerowhich directly binds to the public API. Second, there are a number of compute utilities likepyarrow.Table.group_by which uses Acero, though this is invisible to the user.

Java#

The Java implementation exposes some capabilities from Arrow datasets. These use Acero implicitly. Thereare no direct bindings to Acero or Substrait in the Java implementation today.

Engine Independent Compute#

If a node requires complex computation then it should encapsulate that work in abstractions that don’t depend onany particular engine design. For example, the hash join node uses utilities such as a row encoder, a hash table,and an exec batch builder. Other places share implementations of sequencing queues and row segmenters. The nodeitself should be kept minimal and simply maps from Acero to the abstraction.

This helps to decouple designs from Acero’s design details and allows them to be more resilient to changes in theengine. It also helps to promote these abstractions as capabilities on their own. Either for use in other enginesor for potential new additions to pyarrow as compute utilities.

Make Tasks not Threads#

If you need to run something in parallel then you should use thread tasks and not dedicated threads.

• This keeps the thread count down (reduces thread contention and context switches)

• This prevents deadlock (tasks get cancelled automatically in the event of a failure)

• This simplifies profiling (Tasks can be easily measured, easier to know where all the work is)

• This makes it possible to run without threads (sometimes users are doing their own threading andsometimes we need to run in thread-restricted environments like emscripten)

Note: we do not always follow this advice currently. There is a dedicated process thread in the asof joinnode. Dedicated threads are “ok” for experimental use but we’d like to migrate away from them.

Don’t Block on CPU Threads#

If you need to run a potentially long running activity that is not actively using CPU resources (e.g. reading fromdisk, network I/O, waiting on an external library using its own threads) then you should use asynchronous utilitiesto ensure that you do not block CPU threads.

Don’t Reinvent the Wheel#

Each node should not be a standalone island of utilities. Where possible, computation should be pushedeither into compute functions or into common shared utilities. This is the only way a project as large asthis can hope to be maintained.

Avoid Query Optimization#

Writing an efficient Acero plan can be challenging. For example, filter expressions and column selectionshould be pushed down into the scan node so that the data isn’t read from disk. Expressions should besimplified and common sub-expressions factored out. The build side of a hash join node should be thesmaller of the two inputs.

However, figuring these problems out is a challenge reserved for a query planner or a query optimizer.Creating a query optimizer is a challenging task beyond the scope of Acero. With adoption of Substraitwe hope utilities will eventually emerge that solve these problems. As a result, we generally avoid doingany kind of query optimization within Acero. Acero should interpret declarations as literally as possible.This helps reduce maintenance and avoids surprises.

We also realize that this is not always possible. For example, the hash join node currently detects if thereis a chain of hash join operators and, if there is, it configure bloom filters between the operators. This istechnically a task that could be left to a query optimizer. However, this behavior is rather specific to Aceroand fairly niche and so it is unlikely it will be introduced to an optimizer anytime soon.

Batch Size#

Perhaps the most discussed performance criteria is batch size. Acero was originallydesigned based on research to follow a morsel-batch model. Tasks are created based ona large batch of rows (a morsel). The goal is for the morsel to be large enough to justifythe overhead of a task. Within a task the data is further subdivided into batches.Each batch should be small enough to fit comfortable into CPU cache (often the L2 cache).

This sets up two loops. The outer loop is parallel and the inner loop is not:

formorselindataset:# parallelforbatchinmorsel:run_pipeline(batch)

The advantage of this style of execution is that successive nodes (or successive operationswithin an exec node) that access the same column are likely to benefit from cache. It alsois essential for functions that require random access to data. It maximizes parallelism whileminimizing the data transfer from main memory to CPU cache.

The morsel/batch model is reflected in a few places in Acero:

• In most source nodes we will try and grab batches of 1Mi rows. This is often configurable.

• In the source node we then iterate and slice off batches of 32Ki rows. This is not currentlyconfigurable.

• The hash join node currently requires that a batches contain at 32Ki rows or less as it uses16-bit signed integers as row indices in some places.

However, this guidance is debateable. Profiling has shown that we do not get any real benefitfrom moving to a smaller batch size. It seems any advantage we do get is lost in per-batchoverhead. Most of this overhead appears to be due to various per-batch allocations. In addition,depending on your hardware, it’s not clear that CPU Cache<->RAM will always be the bottleneck. Acombination of linear access, pre-fetch, and high CPU<->RAM bandwidth can alleviate the penaltyof cache misses.

As a result, this section is included in the guide to provide historical context, but should notbe considered binding.

Scanner v2#

The scanner is currently a node in the datasets module registered with the factory registry as “scan”.This node was written prior to Acero and made extensive use of AsyncGenerator to scan multiple filesin parallel. Unfortunately, the use of AsyncGenerator made the scan difficult to profile, difficultto debug, and impossible to cancel. A new scan node is in progress. It is currently registered withthe name “scan2”. The new scan node uses the AsyncTaskScheduler instead of AsyncGenerator and shouldprovide additional features such as the ability to skip rows and handle nested column projection (forformats that support it)

OrderBySink and SelectKSink#

These two exec nodes provided custom sink implementations. They were written before ordered executionwas added to Acero and were the only way to generate ordered output. However, they had to be placedat the end of a plan and the fact that they were custom sink nodes made them difficult to describe withDeclaration. The OrderByNode and FetchNode replace these. These are kept at the moment until existingbindings move away from them.