Opensource internals

EXPLAIN ANALYZE Without the Overhead: How DataFusion Times Every Operator While the Query Runs

11 min read Jay Hubbert

How DataFusion's per-partition metric model, scoped timers, and deferred plan rendering turn a running query into a cost breakdown.

TL;DR: Metrics are plain atomics attached to an operator, tagged with the partition that produced them, and read after the query completes. elapsed_compute is not wall-clock: it accumulates only the nanoseconds spent inside poll(), so time suspended waiting for I/O is excluded. Timers are scoped tightly around the operator’s own work (a filter times its predicate evaluation, not its child), so the numbers are comparable rather than double-counted. EXPLAIN ANALYZE is not a separate measurement mode — it is the same plan with an AnalyzeExec on top that discards rows and then renders the tree with live metric values.

The path from SQL to a stream of batches

Before measuring, it helps to see what is being measured. A query in DataFusion is a chain of ExecutionPlan nodes, each of which produces a RecordBatchStream of Arrow batches.

text 
   ctx.sql("SELECT ...")
        │
        ▼
   LogicalPlan  ── optimizer rules ──▶  optimized LogicalPlan
        │
        ▼  DataFrame::create_physical_plan()   (dataframe/mod.rs:301-303)
   physical plan (Arc<dyn ExecutionPlan>)
        │
        ▼  execute_stream()                   (physical-plan/src/execution_plan.rs:1800-1815)
   ┌──────────────────────────┐
   │ 1 partition? plan.execute(0)                               │
   │ N partitions? CoalescePartitionsExec(plan).execute(0)      │
   └──────────────────────────┘
        │
        ▼  poll_next() → RecordBatch → RecordBatch → …
   SendableRecordBatchStream

The interesting structural decision is in execute_stream (datafusion/physical-plan/src/execution_plan.rs:1800-1815): a plan with more than one output partition gets a CoalescePartitionsExec wrapped around it so the caller always sees a single stream, but the parallelism is preserved underneath. The alternative — returning N streams and letting the caller merge — pushes fan-in into user code and makes cancellation semantics ambiguous.

collect_partitioned (:1822-1866) shows the other shape: one task per partition, all driven by a JoinSet, results re-sorted by partition index at the end:

rust 
// datafusion/physical-plan/src/execution_plan.rs:1830-1852 (abridged)
let mut join_set = JoinSet::new();
streams.into_iter().enumerate().for_each(|(idx, stream)| {
    join_set.spawn(async move {
        let result: Result<Vec<RecordBatch>> = stream.try_collect().await;
        (idx, result)
    });
});

while let Some(result) = join_set.join_next().await {
    match result {
        Ok((idx, res)) => batches.push((idx, res?)),
        Err(e) => {
            if e.is_panic() {
                std::panic::resume_unwind(e.into_panic());
            } else { unreachable!(); }
        }
    }
}
batches.sort_by_key(|(idx, _)| *idx);

Note the panic handling: a panic in one partition is re-raised on the collecting thread rather than being swallowed. Metrics are useless if the query dies before you can read them.

Everything an operator needs at runtime is bundled into TaskContext (datafusion/execution/src/task.rs:50-80): the session id, the SessionConfig, the function registry (scalar / aggregate / window / higher-order UDFs), and the RuntimeEnv that owns the memory pool. Operators get an Arc<TaskContext> and never touch global state — which is what makes it possible to run many queries, with different configs, on one runtime.

The metric model: cheap, partitioned, typed

The core type is MetricValue (datafusion/physical-expr-common/src/metrics/value.rs:635-718):

rust 
pub enum MetricValue {
    OutputRows(Count),
    ElapsedCompute(Time),
    SpillCount(Count),
    SpilledBytes(Count),
    OutputBytes(Count),
    OutputBatches(Count),
    SpilledRows(Count),
    CurrentMemoryUsage(Gauge),
    Count { name: Cow<'static, str>, count: Count },
    Gauge { name: Cow<'static, str>, gauge: Gauge },
    PeakMemoryUsage { name: Cow<'static, str>, gauge: Gauge },
    Time { name: Cow<'static, str>, time: Time },
    StartTimestamp(Timestamp),
    EndTimestamp(Timestamp),
    PruningMetrics { name: Cow<'static, str>, pruning_metrics: PruningMetrics },
    Ratio { name: Cow<'static, str>, ratio_metrics: RatioMetrics },
    Custom { name: Cow<'static, str>, value: Arc<dyn CustomMetricValue> },
}

Two things are worth noticing.

First, the built-in metrics are typed variants, not a HashMap<String, u64>. That means the engine can aggregate them intelligently: Count sums, Gauge takes the last value, Time sums nanoseconds, Timestamp takes the first/last. The Custom variant with a trait object is the escape hatch for operator-specific values that need custom merge semantics.

Second, the primitives are deliberately dumb. Count is an AtomicU64 with Ordering::Relaxed (value.rs:40); Time is an atomic nanosecond counter (value.rs:154); Gauge is an atomic with a load/store pair (value.rs:89). Timestamp is a Mutex<Option<DateTime<Utc>>> (value.rs:248), written exactly twice per partition.

The partition is not a label bolted on afterwards; it is part of the metric’s identity. Metric carries partition: Option<usize> (metrics/mod.rs:80-100), and ExecutionPlanMetricsSet is the per-operator registry that MetricBuilder writes into:

rust 
// datafusion/physical-expr-common/src/metrics/baseline.rs:79-100 (abridged)
pub fn new(metrics: &ExecutionPlanMetricsSet, partition: usize) -> Self {
    let start_time = MetricBuilder::new(metrics).start_timestamp(partition);
    start_time.record();

    Self {
        end_time: MetricBuilder::new(metrics).with_type(MetricType::Summary).end_timestamp(partition),
        elapsed_compute: MetricBuilder::new(metrics).with_type(MetricType::Summary).elapsed_compute(partition),
        output_rows: MetricBuilder::new(metrics).with_type(MetricType::Summary).output_rows(partition),
        output_bytes: MetricBuilder::new(metrics).with_type(MetricType::Summary).output_bytes(partition),
        output_batches: MetricBuilder::new(metrics).with_type(MetricType::Dev).output_batches(partition),
    }
}

Every operator gets these five for free by constructing a BaselineMetrics for its partition. output_rows/output_bytes/output_batches are Summary type; output_batches is Dev — the distinction exists so EXPLAIN ANALYZE can show a curated set by default and everything with EXPLAIN ANALYZE VERBOSE.

The elapsed_compute trick: measure the work, not the wait

This is the most misunderstood metric in DataFusion, and the doc comment is explicit (value.rs:638-657):

Note 1: Does not include time other operators spend computing input. Note 2: Does includes time when the thread could have made progress but the OS did not schedule it (e.g. due to CPU contention), thus making this value different than the classical definition of “cpu_time”.

So elapsed_compute is “wall-clock time this operator spent doing its own work”, not CLOCK_THREAD_CPUTIME_ID. Implementation-wise it is usually a scoped timer around a tight region of code:

rust 
// datafusion/physical-plan/src/filter.rs:1685-1690 (abridged)
Some(Ok(batch)) => {
    let timer = elapsed_compute.timer();
    let status = self.predicate.as_ref()
        .evaluate(&batch)
        ...

and Time::timer() returns a ScopedTimerGuard that adds the elapsed nanoseconds on drop (value.rs:223-228, :330-341). The timer covers the predicate evaluation and the projection — not the self.input.poll_next_unpin(cx) call, which happens outside the timer scope. This is why operator timings are comparable: a filter’s elapsed_compute is the cost of filtering, not the cost of waiting for the scan.

There is a second, more clever implementation for async operators: ElapsedComputeFuture (datafusion/physical-expr-common/src/metrics/elapsed_compute.rs:43-100).

rust 
// datafusion/physical-expr-common/src/metrics/elapsed_compute.rs:70-84 (abridged)
fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
    let self_projected = self.project();
    let start = Instant::now();
    let result = self_projected.inner.poll(cx);
    *self_projected.curr += start.elapsed();
    if result.is_ready() {
        self_projected.elapsed_compute.add_duration(*self_projected.curr);
        *self_projected.curr = Duration::default();
    }
    result
}

Three design details in there:

  1. Only the poll duration is counted. Time the future spends suspended (waiting for a socket, a channel, a lock) is excluded, because it is not inside poll(). Async I/O automatically stops being charged to the operator.

  2. A local accumulator avoids an atomic per poll. The nanos are summed in a plain Duration field and flushed to the atomic Time only when the future completes. The cost of instrumentation is one Instant::now() per poll and one atomic add per future, not per poll.

  3. PinnedDrop flushes on cancellation. If the future is dropped mid-flight (query cancelled), the accumulated time is still added:

    rust 
    // datafusion/physical-expr-common/src/metrics/elapsed_compute.rs:54-64
#[pinned_drop]
impl<T> PinnedDrop for ElapsedComputeFuture<T> {
    fn drop(self: Pin<&mut Self>) {
        if self.curr > Duration::default() {
            let self_projected = self.project();
            self_projected.elapsed_compute.add_duration(*self_projected.curr);
        }
    }
}

    A cancelled query still tells you where it was.

Operators that offload work to other tasks use BaselineMetrics::intermediate() (baseline.rs:107-117), which returns a BaselineMetrics that shares only the elapsed_compute counter and drops the rest — so background work is charged to the operator that owns it without inflating its row counts.

Recording output without boilerplate

The RecordOutput trait (baseline.rs:316-376) makes recording output a one-liner that composes with ?:

rust 
// datafusion/physical-expr-common/src/metrics/baseline.rs:331-341 (abridged)
impl RecordOutput for RecordBatch {
    fn record_output(self, bm: &BaselineMetrics) -> Self {
        bm.record_output(self.num_rows());
        let n_bytes = get_record_batch_memory_size(&self);
        bm.output_bytes.add(n_bytes);
        bm.output_batches.add(1);
        self
    }
}

Because it is implemented for RecordBatch, &RecordBatch, Option<...>, Result<...> and usize, an operator can write Ok(batch.record_output(&baseline_metrics)) and get rows, bytes and batch count recorded in the right order, with the batch size computed by DataFusion’s shared get_record_batch_memory_size (which knows about shared buffers).

Aggregation: turning N partitions into one number

Each partition registers its own copy of each metric. At display time, MetricsSet::aggregate_by_name (metrics/mod.rs:374-400) folds them:

rust 
// datafusion/physical-expr-common/src/metrics/mod.rs:374-390 (abridged)
pub fn aggregate_by_name(&self) -> Self {
    let mut map = HashMap::new();
    for metric in &self.metrics {
        let key = metric.value.name();
        map.entry(key)
            .and_modify(|accum: &mut Metric| {
                accum.value_mut().aggregate(metric.value());
            })
            .or_insert_with(|| { /* new empty metric with partition=None */ });
    }
    ...
}

MetricValue::aggregate knows the right merge per variant: counts add, gauges take the latest, timestamps take min/max, ratios recompute. The result has partition = None, i.e. “this number covers the whole operator”.

The display layer then chooses between aggregated and per-partition views (datafusion/physical-plan/src/display.rs:596-620):

rust 
// datafusion/physical-plan/src/display.rs:601-616 (abridged)
ShowMetrics::Aggregated => {
    if let Some(metrics) = plan.metrics() {
        let metrics = metrics
            .filter_by_metric_types(self.metric_types)
            .aggregate_by_name();
        let mut metrics = omit_idle_bloom_filter_pruning_metric(&metrics)
            .sorted_for_display()
            .timestamps_removed();
        ...
        write!(self.f, ", metrics=[{metrics}]")?;
    }
}

Note the chain: filter by type (Summary vs Dev), aggregate across partitions, drop the noisiest metric (a bloom-filter pruning metric that pruned nothing), sort for human readability, strip timestamps (they are ugly and rarely what you want), then optionally filter by category.

That last filter is a recent and useful addition. MetricCategory (datafusion/common/src/format.rs:307-336) splits metrics into Rows, Bytes, Timing and Uncategorized, and the semantics are documented honestly: rows and bytes are “mostly deterministic given the same plan and data”, while timing is “Non-deterministic — varies across runs even on the same hardware”. datafusion.explain.analyze_categories (datafusion/common/src/config.rs:1968) lets you ask for SET datafusion.explain.analyze_categories = 'rows, bytes' to get a profile that is comparable across runs — a genuinely useful property for regression testing.

A derived metric: output row skew

Not all metrics are recorded during execution; some are computed at display time. BaselineMetrics::output_rows_skew_metric (baseline.rs:146-190) computes how unevenly an operator’s output rows are spread across partitions:

text 
   effective_parallelism = (Σ rᵢ)² / Σ rᵢ²
   skew = (1 − (effective_parallelism − 1) / (partitions − 1)) × 100%

For rows [10, 10, 10, 10] the effective parallelism is 4 and the skew is 0%; for [40, 0, 0, 0] it is 100%. This is a diagnostic metric — it tells you that your repartition is producing nothing on three of four partitions, which is often the actual explanation for a slow query. It is attached by MetricsSet::with_output_rows_skew() and is a Ratio metric so it renders as a percentage.

EXPLAIN ANALYZE: the plan, annotated with live numbers

AnalyzeExec (datafusion/physical-plan/src/analyze.rs) is a pass-through operator with one job: run the plan, throw away the rows, and print the tree with metrics. Its execute (:275-345) runs every input partition through a RecordBatchReceiverStreamBuilder, sums the rows, and then renders:

rust 
// datafusion/physical-plan/src/analyze.rs:311-345 (abridged)
let mut input_stream = builder.build();
let output = async move {
    let mut total_rows = 0;
    while let Some(batch) = input_stream.next().await.transpose()? {
        total_rows += batch.num_rows();
    }
    drop(input_stream);

    let duration = start.elapsed();
    create_output_batch(
        verbose, show_statistics, total_rows, duration,
        &captured_input, &captured_schema, &metric_types,
        metric_categories.as_deref(), &format, &statistics_registry,
    )
};

and create_output_batch (:508-600) is where the tree is rendered:

rust 
// datafusion/physical-plan/src/analyze.rs:520-535 (abridged)
let annotated_plan = DisplayableExecutionPlan::with_metrics(input.as_ref())
    .set_metric_types(metric_types.to_vec())
    .set_metric_categories(metric_categories.map(|c| c.to_vec()))
    .set_show_statistics(show_statistics)
    .set_statistics_registry(statistics_registry.clone())
    .indent(verbose)
    .to_string();

The critical property: the plan is rendered after the query has run, so plan.metrics() returns the live metric values, not zeroed placeholders. There is no separate “profiling build” of the engine. EXPLAIN (without ANALYZE) renders the same tree with empty metric lists, because nothing ran.

The output is a two-column Arrow batch (type, plan) — EXPLAIN ANALYZE is a regular query that returns a string. And it supports ExplainFormat::PostgresJSON for machine consumption (:555-580), with "Actual Rows" / "Actual Total Time" mapped onto the PG-canonical keys.

Trade-offs and failure modes

  • Instrumentation is not free, but it is cheap. Every batch crossing an operator does a handful of relaxed atomic adds. Count::add is a fetch_add with Relaxed ordering; the elapsed time does one Instant::now() per timed region. This is why elapsed_compute uses a local accumulator for futures — an atomic per poll() would be measurable in hot loops.
  • elapsed_compute is not CPU time. Under CPU contention it over-counts (the thread was ready but not scheduled). Comparing operators within one run is meaningful; comparing across machines is not.
  • Metrics are opt-in per operator. ExecutionPlan::metrics() returns Option<MetricsSet>, and a operator that returns None simply shows metrics=[]. There is no enforcement that a new operator reports anything.
  • EXPLAIN ANALYZE measures a slightly different plan. AnalyzeExec wraps the plan, adds a row-counting pass, and disables result delivery. The plan’s own behaviour is unchanged, but the system is: one extra node, one extra stream, and the output batches are still produced and counted.
  • Some operators do not implement elapsed_compute at all. AsyncFuncExec carries an explicit TODO: “TODO: Track elapsed_compute in BaselineMetrics” (datafusion/physical-plan/src/async_func.rs:228). Its BaselineMetrics only tracks output rows/bytes/batches.
  • Timestamps are wall-clock, and are removed by default. start_timestamp/end_timestamp per partition give you the span, but they are stripped from the default rendering (timestamps_removed()), because they are noisy and rarely comparable.
  • Category filtering is a blunt instrument. Uncategorized catches custom metrics that never declared a category (format.rs:324-335), so SET datafusion.explain.analyze_categories = 'rows' silently hides operator-specific counters.

The takeaway

The design here is worth stealing: measure with the cheapest possible primitive (relaxed atomics), measure only the region you actually own (scoped timers around your own work, not around your child’s), tag everything with the partition that produced it, and render the report after the fact from live values. The result is that EXPLAIN ANALYZE in DataFusion is not a special instrumentation mode you turn on — it is the same execution with a renderer bolted on top, which is why the numbers are trustworthy and why there is no “profiling build” of the query engine.

DatafusionRuntimePerformanceInternals