Acero

Active contributors: William Ayd, Rossi Sun, gitmodimo, Antoine Pitrou

Acero is Arrow's streaming, push-based query execution engine. It lives in cpp/src/arrow/acero/ and was carved out of cpp/src/arrow/compute/exec/ in April 2023 to give the engine its own namespace and CMake target. Acero is the backbone of the dataset scanner, the dplyr backend in the R package, and the pandas-on-arrow query path.

Purpose

Take an ExecPlan — a directed graph of nodes — and stream record batches through it. Acero is designed for:

• Streaming. Data flows from sources to sinks in batches; the entire dataset never needs to be in memory at once.
• Push-based execution. Sources push batches to downstream nodes via callbacks, instead of nodes pulling. This pairs well with async IO.
• Concurrent execution. Each node can dispatch work to the CPU thread pool; backpressure is managed via BackpressureHandler.
• Composition with compute. Filter and project nodes evaluate arrow::compute::Expressions; aggregate nodes call hash-aggregate kernels; the engine reuses everything in cpp/src/arrow/compute/.

ExecPlan and ExecNode

cpp/src/arrow/acero/exec_plan.h defines the core types:

• ExecPlan — owns a graph of nodes and the execution context (thread pool, query options).
• ExecNode — an abstract operator with inputs(), outputs(), InputReceived(batch), Init, and StartProducing.
• ExecBatch — a batch flowing through the graph (a RecordBatch plus optional row indices).
• Declaration — a serializable "I want a node of type X with options Y" descriptor that builds into an actual ExecNode.

A query is built by composing Declarations:

auto plan = Declaration::Sequence({
    {"source", SourceNodeOptions{schema, batch_generator}},
    {"filter", FilterNodeOptions{filter_expr}},
    {"project", ProjectNodeOptions{project_exprs}},
    {"aggregate", AggregateNodeOptions{aggregates, keys}},
    {"order_by", OrderByNodeOptions{ordering}},
    {"sink", SinkNodeOptions{generator}},
});

ARROW_ASSIGN_OR_RAISE(auto reader, DeclarationToReader(plan));

Built-in nodes

Node	File	What it does
source	source_node.cc	Pulls from an AsyncGenerator<ExecBatch> or schema+iterator.
record_batch_source	source_node.cc	Pulls from a RecordBatchReader.
filter	filter_node.cc	Drops rows not matching a predicate.
project	project_node.cc	Computes new columns / rearranges columns.
aggregate	groupby_aggregate_node.cc, scalar_aggregate_node.cc	Hash and scalar aggregates.
hashjoin	hash_join_node.cc (52 KB), swiss_join.cc (122 KB)	Multi-strategy hash join with build/probe phases.
asofjoin	asof_join_node.cc (~61 KB)	Time-series asof join (each left row matches the most recent right row).
union	union_node.cc	Concatenates streams.
order_by	order_by_node.cc	Sort.
fetch	fetch_node.cc	Limit / offset.
pivot_longer	pivot_longer_node.cc	Wide → long reshape.
sorted_merge	sorted_merge_node.cc	Merges already-sorted streams.
sink	sink_node.cc	Terminal: emits batches to an AsyncGenerator or RecordBatchReader.
tpch	tpch_node.cc (~149 KB)	TPC-H data generator.

Each node is registered in cpp/src/arrow/acero/exec_plan.cc via RegisterFactory.

Hash join

hash_join_node.cc is the most complex node. It uses two implementations behind a common interface:

• Default (legacy) — based on key_hash_internal.h and key_map_internal.h from the compute layer.
• Swiss table — swiss_join.cc (122 KB) and swiss_join_avx2.cc (25 KB). Implements a high-throughput hash join using a Swiss table layout that keeps small fixed-width metadata in a separate vector for fast linear probing. Selected by HashJoinNodeOptions::should_use_swiss_join.

Both share hash_join.cc/.h for the build/probe orchestration. The probe side handles backpressure with concurrent_queue_internal.h to avoid overwhelming downstream nodes.

bloom_filter.cc implements a Bloom filter applied to the build side for quick row rejection on the probe side. AVX2 acceleration in bloom_filter_avx2.cc.

Aggregates

Hash aggregates use:

• groupby_aggregate_node.cc — the ExecNode.
• aggregate_internal.cc/.h — shared key extraction.
• The hash_* kernels in cpp/src/arrow/compute/kernels/hash_aggregate*.cc for the actual reductions.
• partition_util.cc — distributes groups across threads when running in parallel.

Scalar aggregates use scalar_aggregate_node.cc and the matching kernels (aggregate_basic.cc, aggregate_quantile.cc, etc.).

Asof join

asof_join_node.cc is one of the more domain-specific nodes. It joins time-series streams where each left row matches the right row whose timestamp is the most recent timestamp ≤ the left timestamp. The implementation uses a accumulation_queue.cc to buffer right-side rows and prune them as left-side time advances.

Backpressure and scheduling

• Nodes implement PauseProducing / ResumeProducing. The sink node throttles upstream producers when downstream consumers are slow.
• The CPU thread pool is arrow::internal::GetCpuThreadPool(). The IO pool is separate (arrow::io::default_io_context()). Nodes schedule work via task_util.cc.
• A QueryContext (query_context.h) carries plan-wide state.

Execution models

A plan can be executed in three ways:

• DeclarationToTable — runs to completion, returns an arrow::Table.
• DeclarationToReader — produces a RecordBatchReader for streaming consumption.
• DeclarationToBatches / DeclarationToBatchesAsync — collects all ExecBatches.

Each approach builds the same ExecPlan and starts the same ExecNode graph.

Test infrastructure

test_nodes.cc and test_util_internal.cc provide reusable plan-building helpers. plan_test.cc (75 KB) exercises plan composition. The hash join and aggregate tests are some of the largest in the project (hash_aggregate_test.cc is ~5,400 lines, hash_join_node_test.cc is ~5,400 lines).

Substrait integration

cpp/src/arrow/engine/substrait/ is the adapter that converts Substrait plans into Acero ExecPlans. See Substrait. This lets Acero serve as an execution backend for any frontend that produces Substrait, including DataFusion's planner and pandas-on-arrow.

Entry points for modification

• Adding a new node: subclass ExecNode, override InputReceived, StartProducing, Init, outputs. Register a factory. The simplest examples are filter_node.cc and project_node.cc.
• Optimizing the hash join: swiss_join.cc and swiss_join_avx2.cc are where the hot path lives. Benchmarks are in hash_join_benchmark.cc.
• Tuning the scheduler: task_util.cc and query_context.cc.

For how Acero is consumed from R see R. For how the dataset scanner builds Acero plans, see Dataset.