DataFusion Has No Runtime of Its Own: How common-runtime Borrows Yours and Keeps CPU Work Cooperative

TL;DR: DataFusion never creates a Tokio runtime. It uses the ambient one that the host application already installed, so every query in the process shares the same worker pool. Every task spawn in the codebase is funnelled through the tiny datafusion-common-runtime crate, enforced by a clippy lint that bans tokio::task::spawn outright. SpawnedTask and JoinSet add the one thing raw Tokio lacks: cancellation on drop, so a dropped query tears down its pipeline instead of leaking tasks. Counter-intuitively, DataFusion does not use spawn_blocking for CPU-bound work — it runs CPU work inline on the async worker threads and relies on cooperative yielding (coop.rs) to avoid starving the runtime.

The problem: a library that owns a runtime steals your threads

Most libraries that need concurrency create their own thread pool. That is fine for a CLI, and terrible for a library. If DataFusion created a Tokio runtime per SessionContext, then an application with ten sessions would have ten runtimes, each sized for the machine’s core count, and the host app’s own async tasks would be fighting DataFusion’s worker threads for CPU.

So DataFusion’s answer is: it doesn’t have a runtime. It borrows the one you already have.

rust
// datafusion/common-runtime/src/common.rs:40-50
pub fn spawn<T>(task: T) -> Self
where
    T: Future<Output = R>,
    T: Send + 'static,
    R: Send,
{
    // Ok to use spawn here as SpawnedTask handles aborting/cancelling the task on Drop
    #[expect(clippy::disallowed_methods)]
    let inner = tokio::task::spawn(trace_future(task));
    Self { inner }
}

tokio::task::spawn uses the current runtime handle, which Tokio stores in a thread-local. The example in datafusion-examples/examples/query_planning/thread_pools.rs:57-68 spells this out:

Each thread has “current” runtime that is installed in a thread local variable which is used by the tokio::spawn function. The #[tokio::main] macro creates a Runtime and installs it as the “current” runtime…

So the “one runtime shared by many users of the library” is the host application’s runtime, and every SessionContext, every query, every operator task in the process shares it. That is a deliberate design choice with a documented escape hatch: if you want DataFusion’s CPU work isolated from your latency-sensitive I/O, you create a second Tokio runtime and hand DataFusion a Handle (see spawn_on / spawn_blocking_on below, and the two-runtime ASCII diagram in datafusion/core/src/lib.rs:566-604).

Note what RuntimeEnv is not: datafusion/execution/src/runtime_env.rs:73-90 shows it holds a memory pool, a disk manager, a cache manager and an object-store registry. It is a resource manager, not a thread pool. The name is historical; the runtime is somebody else’s.

text

        Host application (e.g. tonic / axum / a batch job)
        ┌──────────────────────────────────────────────┐
        │  tokio::runtime::Runtime (multi-thread)      │
        │  ┌────────┬────────┬────────┬────────┐       │
        │  │worker 1│worker 2│worker 3│worker N│       │
        │  └───┬────┴───┬────┴────────┴────────┘       │
        │      │        │                               │
        │      │  tokio::task::spawn() uses the CURRENT │
        │      │  handle (thread-local)                │
        │      ▼        ▼                               │
        │  ┌──────────────────────────────────────┐    │
        │  │ DataFusion query A  (SessionContext) │    │
        │  │ DataFusion query B  (SessionContext) │    │
        │  │ DataFusion query C  (SessionContext) │    │
        │  └──────────────────────────────────────┘    │
        └──────────────────────────────────────────────┘

The choke point: a lint instead of a code review

The interesting engineering is not “use Tokio”. It is making sure that nobody in a 500k-line codebase spawns a task in a way that leaks. DataFusion does this with clippy configuration, not with reviewer discipline:

toml
# clippy.toml:1-5
disallowed-methods = [
    { path = "tokio::task::spawn", reason = "To provide cancel-safety, use `SpawnedTask::spawn` instead (https://github.com/apache/datafusion/issues/6513)" },
    { path = "tokio::task::spawn_blocking", reason = "To provide cancel-safety, use `SpawnedTask::spawn_blocking` instead (https://github.com/apache/datafusion/issues/6513)" },
    { path = "std::vec::Vec::reserve", reason = "Use `Vec::try_reserve` so allocation failures can be reported instead of panicking", replacement = "try_reserve" },
]

The tokio::task::spawn / tokio::task::spawn_blocking bans have essentially two escapes: the two calls inside SpawnedTask itself (datafusion/common-runtime/src/common.rs:47 and :59), plus one deliberate escape in the sort spill path where a detached task is genuinely wanted (datafusion/physical-plan/src/sorts/sort.rs:4155, commented “spawn allowed only in tests” — a comment that has since drifted from the code). Everything else in the tree — a dozen crates, hundreds of operators — goes through datafusion-common-runtime, a 588-line crate whose entire public surface is:

rust
// datafusion/common-runtime/src/lib.rs:28-36
pub mod common;
mod join_set;
mod trace_utils;

pub use common::SpawnedTask;
pub use join_set::JoinSet;
pub use trace_utils::{
    JoinSetTracer, JoinSetTracerError, set_join_set_tracer, trace_block, trace_future,
};

That is the whole trick: a library-level choke point with a compile-time enforcement mechanism. The crate is also the single place where instrumentation can be injected, via a global tracer:

rust
// datafusion/common-runtime/src/trace_utils.rs:92-118
static GLOBAL_TRACER: OnceCell<&'static dyn JoinSetTracer> = OnceCell::const_new();
...
pub fn set_join_set_tracer(
    tracer: &'static dyn JoinSetTracer,
) -> Result<(), JoinSetTracerError> {
    GLOBAL_TRACER
        .set(tracer)
        .map_err(|_set_err| JoinSetTracerError::AlreadySet)
}

trace_future boxes the future’s output type to Box<dyn Any + Send>, hands it to the tracer, and downcasts it back (trace_utils.rs:130-160). A tracer can therefore wrap every task DataFusion spawns — for OpenTelemetry spans, for a custom scheduler, for a test harness — without any operator knowing. The OnceCell means “set once, at startup”; a second attempt is a hard error rather than a silent race.

Cancel-safety: the actual reason the wrapper exists

Raw tokio::task::spawn returns a JoinHandle that, when dropped, detaches the task — the task keeps running to completion in the background. For a query engine that is a disaster: drop the result stream of a cancelled query and you keep burning CPU on a plan nobody is reading.

SpawnedTask fixes this with three lines of Drop:

rust
// datafusion/common-runtime/src/common.rs:108-112
impl<R> Drop for SpawnedTask<R> {
    fn drop(&mut self) {
        self.inner.abort();
    }
}

And it converts panics into something the caller can propagate rather than swallow:

rust
// datafusion/common-runtime/src/common.rs:87-104
pub async fn join_unwind_mut(&mut self) -> Result<R, JoinError> {
    self.await.map_err(|e| {
        // `JoinError` can be caused either by panic or cancellation. We have to handle panics:
        if e.is_panic() {
            std::panic::resume_unwind(e.into_panic());
        } else {
            log::warn!("SpawnedTask was polled during shutdown");
            e
        }
    })
}

Note the honest doc comment on the blocking variant (common.rs:29-31): “if the task was spawned with spawn_blocking, it will only be aborted if it hasn’t started yet.” You cannot kill a running OS thread safely in Rust; DataFusion says so instead of pretending.

JoinSet (datafusion/common-runtime/src/join_set.rs:29-35) is the fan-out counterpart: it forwards the whole Tokio JoinSet API and, crucially, abort_all() on drop cancels every outstanding task. This is how RecordBatchReceiverStreamBuilder works (datafusion/physical-plan/src/stream.rs:82-118): each input partition is a spawned task writing into an mpsc channel, and dropping the stream aborts all of them.

rust
// datafusion/physical-plan/src/stream.rs:118-160 (abridged)
pub fn build(self) -> BoxStream<'static, Result<O>> {
    ...
    // future that checks the result of the join set, and propagates panic if seen
    let check = async move {
        while let Some(result) = join_set.join_next().await {
            match result {
                Ok(Ok(())) => {}
                // This means a blocking task error
                Ok(Err(error)) => return Some(Err(error)),
                // This means a tokio task error, likely a panic
                Err(e) => {
                    if e.is_panic() {
                        // resume on the main thread
                        std::panic::resume_unwind(e.into_panic());
                    } else {
                        return Some(exec_err!("Non Panic Task error: {e}"));
                    }
                }
            }
        }
        None
    };
    ...
}

That check future is merged with the data channel using futures::stream::select, so a panic in any partition propagates into the output stream instead of vanishing into a JoinError nobody is watching. This is the difference between “a panic in partition 3 of 8 kills the query with a stack trace” and “the query hangs forever”.

Why `spawn_blocking` is not the answer for CPU work

The naive instinct is: CPU-bound work blocks the async worker, so push it to spawn_blocking and the blocking pool eats it. DataFusion explicitly refuses to do this:

rust
// datafusion/core/src/lib.rs:644-651
//! Note that DataFusion does not use [`tokio::task::spawn_blocking`] for
//! CPU-bounded work, because `spawn_blocking` is designed for blocking **IO**,
//! not designed CPU bound tasks. Among other challenges, spawned blocking
//! tasks can't yield waiting for input (can't call `await`) so they
//! can't be used to limit the number of concurrent CPU bound tasks or
//! keep the processing pipeline to the same core.

The reasons are worth internalising:

No await ⇒ no backpressure. A blocking task cannot suspend waiting for input. You cannot implement a bounded pipeline inside one, so you cannot limit concurrency or keep the pipeline on the same core.
A second unbounded pool. Tokio’s blocking pool defaults to 512 threads. Moving CPU work there means the machine now has two pools of workers; the “shared runtime” property is gone and you get double the context switching.
Different cancellation semantics. As common.rs:29-31 admits, spawn_blocking tasks that have started cannot be aborted — the exact opposite of the cancel-safety the rest of the crate is built around.

DataFusion’s actual answer is cooperative scheduling on the async workers. datafusion/physical-plan/src/coop.rs:22-40:

If a Stream runs for a long period of time without yielding back to the Tokio executor, it can starve other tasks waiting on that executor to execute them. Additionally, this prevents the query execution from being cancelled.

The implementation has three modes, selected at compile time (coop.rs:144-205). The default uses Tokio’s own cooperative budget:

rust
// datafusion/physical-plan/src/coop.rs:155-162
let coop = std::task::ready!(tokio::task::coop::poll_proceed(cx));
let value = self.inner.poll_next_unpin(cx);
if value.is_ready() {
    coop.made_progress();
}
value

The per_stream fallback shows the mechanism plainly (coop.rs:120, :190-201): a counter of 128 (Tokio’s own TASK_BUDGET) decremented per emitted batch; when it hits zero, the stream returns Poll::Pending and re-wakes itself, giving the scheduler a slot to run other tasks.

text

    poll_next() on a hot stream
    ────────────────────────────────────────────────────────────
    batch ─► batch ─► batch ─► ... 128 batches ... ─► budget==0
                                                      │
                                     return Poll::Pending + wake_by_ref()
                                                      │
                            scheduler runs another task (or the
                            cancellation token), then re-polls us

Crucially, this is injected automatically. The EnsureCooperative physical optimizer rule (coop.rs:63-70) inspects plans for “exchange-like” operators that pass data through their own channels rather than Tokio’s, and wraps them in CooperativeExec (coop.rs:66-72). A user writing a custom TableProvider gets cooperative behaviour without writing a single yield point.

The escape hatch: `spawn_on` and `spawn_blocking_on`

Because the runtime is ambient, DataFusion cannot choose a different one — but it can be told to use one:

rust
// datafusion/common-runtime/src/join_set.rs:73-81
/// [JoinSet::spawn_on](tokio::task::JoinSet::spawn_on) - Spawn a task on a provided runtime.
pub fn spawn_on<F>(&mut self, task: F, handle: &Handle) -> AbortHandle
where
    F: Future<Output = T>,
    F: Send + 'static,
    T: Send,
{
    self.inner.spawn_on(trace_future(task), handle)
}

The same Handle-based API exists on SpawnedTask’s consumers, e.g. RecordBatchReceiverStreamBuilder::spawn_blocking_on (datafusion/physical-plan/src/stream.rs:308-315). This is how the thread_pools example runs DataFusion’s CPU-heavy plan on a dedicated runtime while the host’s I/O runtime stays responsive — exactly the two-pool diagram in datafusion/core/src/lib.rs:566-604.

Trade-offs and failure modes

One runtime, many workloads. The default is convenient and fast for local files, but a latency-sensitive service reading from S3 will see p99 spikes because Parquet decoding and filtering share worker threads with network handling. The fix is a second runtime plus spawn_on/spawn_blocking_on, and it costs you the ergonomics of “just call df.collect()”.
Backpressure is cooperative, not preemptive. A single poll_next that ignores the coop budget (a badly written custom operator) can still block a worker for milliseconds-to-seconds. The optimizer rule catches known patterns; it cannot catch everything.
spawn_blocking is still used — for I/O and parsing. datafusion/datasource-json/src/source.rs:510 and datafusion/catalog/src/stream.rs:531 use SpawnedTask::spawn_blocking for JSON parsing and object-store writes, where the work genuinely blocks. The rule is “blocking I/O may go to the blocking pool; CPU work may not”.
Ambient runtime = ambient problems. If a user drops the last handle to the runtime, in-flight tasks are cancelled with a confusing JoinError. SpawnedTask::join_unwind_mut at least logs SpawnedTask was polled during shutdown instead of panicking (common.rs:92-94).
The tracer is process-global. set_join_set_tracer can only be called once (trace_utils.rs:110-118). In a library that hosts multiple DataFusion instances, the first one wins. That is fine for tracing, dangerous for anything that changes scheduling semantics.

The takeaway

The interesting part of DataFusion’s runtime design is not that it uses Tokio — it’s that it refuses to own one, and then spends three small files making that refusal safe: a clippy lint to ban raw spawns, a Drop-based SpawnedTask to make cancellation the default, a JoinSet to make fan-out cancellation the default, and a cooperative scheduler so that CPU-heavy streams don’t monopolise the workers they borrowed. All of it funnels through 588 lines in datafusion/common-runtime/, which is about the size of one operator file in the same repository.