Distributed DataFusion: Datadog Open-Sources a Framework for Running One Query Across Many Machines

Apache DataFusion is a fast, embeddable query engine built in Rust on top of Apache Arrow. It handles single-node query execution well — fast columnar processing, vectorized operators, good memory characteristics. What it doesn’t do natively is distribute a single query plan across multiple machines.

Datadog runs DataFusion in production for observability query workloads. Their engineers hit the single-node ceiling on long-running queries — the kind that scan large windows of telemetry data, aggregate across many series, or join multiple large datasets. Rather than switching to a different query engine, they extended DataFusion with distributed execution semantics and open-sourced the result under Apache 2.0.

The Three Primitives

Network Coalesce

The simplest distributed primitive: collect partial results from multiple machines to one machine for final aggregation.

A query that runs on N machines produces N partial result sets. Network Coalesce streams these partial results to a designated coordinator node, which merges them into the final output. This is the pattern that makes SUM, COUNT, and AVG distributable — each machine computes a partial aggregate, the coordinator applies the merge function, and the result is correct.

The implementation handles the backpressure problem: if the coordinator processes results slower than the worker machines produce them, Network Coalesce applies bounded buffering rather than unbounded queuing or dropped results.

Network Shuffle

The harder primitive: repartition data across machines by a key.

For operations where the result depends on having all rows with the same key on the same machine — GROUP BY, JOIN, ORDER BY — you can’t simply aggregate partial results. The data needs to be reshuffled so that each machine holds all rows for the keys it’s responsible for.

Network Shuffle implements hash partitioning: each source machine hashes each row’s key and sends the row to the machine responsible for that hash range. After the shuffle completes, each machine has all rows for its assigned key range, and can execute the final aggregation or join locally.

The efficiency tradeoff: Network Shuffle incurs network transfer proportional to data size, while Network Coalesce only transfers aggregated results. For queries where the shuffle can be avoided — where the data is already partitioned correctly, or where an approximate result is acceptable — Datadog’s framework uses Network Coalesce. Shuffle is reserved for operations that require exact correctness and can’t be expressed as commutative partial aggregates.

Adaptive Planning

The optimizer layer: the query planner observes runtime statistics (actual row counts, data distributions, network speeds) and adjusts the execution plan mid-flight.

Static query planners assume statistics collected at plan time are representative of actual data distribution. In observability workloads — telemetry data with highly variable cardinality across time windows, services, and metric names — statistics from one time window may be a poor predictor for another. Adaptive Planning collects runtime feedback and rewrites the query plan to use better strategies based on actual observed behavior.

The specific optimization that Datadog highlights: switching from hash join to broadcast join when runtime cardinality is smaller than the static planner estimated. A broadcast join sends the smaller table to all machines instead of shuffling both tables by key — it’s much faster when one side is small, but only when the size is actually small. Adaptive Planning makes this decision at runtime rather than at planning time.

Benchmark Results

Datadog benchmarked against single-node DataFusion, Apache Spark, Apache Ballista (the existing distributed DataFusion extension), and Trino on a synthetic observability workload.

10x improvement over single-node DataFusion on long-running queries. This is the ceiling case — the improvement on short queries is smaller (distribution overhead dominates on low-latency queries).

Faster than Spark, Ballista, and Trino in the benchmark suite. The comparison against Ballista is notable: Ballista is the existing distributed DataFusion project and uses a different approach to distributed execution. Datadog’s framework outperforms it on the specific benchmark workload, though the gap would vary significantly depending on query patterns and cluster configuration.

The benchmark is Datadog’s own internal comparison and the workload is their specific production use case. As with all vendor benchmarks: the results are meaningful for understanding the performance characteristics of the framework, but the absolute numbers depend heavily on the workload.

The Framework Design

The deliberate design choice: this is a framework, not a fixed query service.

Most distributed query engines (Spark, Trino, Presto) are systems you deploy, configure, and query via a specific interface. DataFusion-distributed is a library you embed and extend. The Network Coalesce, Network Shuffle, and Adaptive Planning primitives are implemented as DataFusion ExecutionPlan nodes — the same abstraction you’d use to write any custom DataFusion operator.

This means the distributed execution engine can be embedded in any Rust application that already uses DataFusion, extended with custom operators, and tuned for specific workload patterns. The tradeoff: you’re responsible for the cluster management, networking, and deployment that managed query services provide.

Why Datadog Built This vs. Adopting Existing Systems

The stated rationale: observability data has a specific access pattern — high-cardinality time series, variable window sizes, large fan-out across metrics and traces — that benefits from tight integration with the data storage layer. By building on DataFusion directly, Datadog’s query engine can leverage in-memory columnar formats, push down filters to the storage tier, and optimize for the specific cardinality characteristics of telemetry data.

Spark and Trino are general-purpose distributed query engines. They’re good at general-purpose workloads. Datadog’s framework is specifically tuned for the access patterns of observability data — which makes it faster on those patterns at the cost of being less general.

Source: Datadog

Content created with AI assistance and reviewed for accuracy.