fix(common): report a cancelled task as an error, not an unreachable arm

[claudespice]

Summary

A tokio::task::JoinError reports one of exactly two outcomes: the task panicked, or it was
cancelled. Seven JoinSet drain loops resumed the panic and then called unreachable!() for
everything else — so the arm written for "this cannot happen" is precisely the arm a cancelled task
lands in. A cancellation became a panic on a Tokio worker thread instead of an error the caller
could surface.

A task is cancelled when it is aborted, or when the runtime it was spawned on shuts down while the
task is still queued. Running the serving path and the query-execution path on separate runtimes —
a normal deployment shape — makes the second case ordinary rather than hypothetical: the drain loop
is still being polled while the tasks it is waiting on are cancelled underneath it.

Reported as a panic out of collect_partitioned:

thread 'tokio-runtime-worker' panicked at datafusion/physical-plan/src/execution_plan.rs:
internal error: entered unreachable code

Fixes spiceai/spiceai#7030.

Changes

DataFusionError::from_join_error resumes a panic on the calling thread — unchanged behaviour, so
a panicking task still surfaces with its payload and backtrace intact — and returns
DataFusionError::ExecutionJoin for a cancellation, keeping the JoinError as the error's source
so a caller can still ask is_cancelled().

ExecutionJoin(Box<JoinError>) already existed for exactly this case, and is already used for it a
few lines away from two of the sites this changes (datasource-arrow/src/file_format.rs:339,
datasource-parquet/src/sink.rs:391). The inconsistency within those two functions is the
clearest evidence this was an oversight rather than a decision.

All seven sites now read Err(e) => return Err(DataFusionError::from_join_error(e)):

crate	function
physical-plan	collect_partitioned (the reported one)
catalog	MemTable::load
datasource-arrow	ArrowFileSink::spawn_writer_tasks_and_join
datasource-parquet	ParquetSink::spawn_writer_tasks_and_join
datasource-parquet	plan_to_parquet
datasource-csv	plan_to_csv
datasource-json	plan_to_json

Two properties worth a reviewer's attention:

• The error returns immediately rather than continuing the drain. Every one of these loops is
  accumulating something — batches, a row count, writers — and continuing past a cancelled task
  would report a partial result as a success. return Err discards the partial accumulation, which
  is what the old panic also did.
• A shared helper rather than seven inline else branches. The duplication had already drifted:
  two other drains in the tree handle the same case correctly and differently
  (physical-plan/src/stream.rs returns exec_err!, common-runtime/src/common.rs logs and
  returns the JoinError). Putting the decision next to the error type is also what makes it
  unit-testable. datafusion-common-runtime::join_unwind is the existing precedent for a shared
  function that resumes panics on the caller's behalf.

The rest of the tree was checked for the same class and is already correct:
datasource/src/write/orchestration.rs handles a cancelled task deliberately ("Don't panic, instead
try to clean up as many writers as possible"), common-runtime/src/join_set.rs is a pass-through
wrapper, and the remaining unwrap()ing drains are all test code. No unreachable!() in an
is_panic() else-branch remains anywhere in the workspace.

Test plan

Three tests, each pinned by a neuter that makes it fail:

• collect_partitioned_reports_a_cancelled_task_as_an_error — calls the production
  collect_partitioned and reaches the arm the way it is reached in practice: the per-partition
  tasks are spawned on one runtime, that runtime is dropped, and a second runtime drives the
  collect. Restoring the unreachable!() fails it.
• from_join_error_reports_a_cancelled_task_as_execution_join — asserts the variant and that the
  JoinError survives as the error's source. Stringifying the cancellation instead fails it.
• from_join_error_resumes_a_panicking_task_instead_of_returning — asserts the panic is not
  downgraded to an error and the payload is carried through. Removing the resume_unwind fails it.

Each neuter fails exactly one of the three, so no test is standing in for another.

cargo test -p datafusion-common -p datafusion-catalog -p datafusion-physical-plan --lib
  8 passed / 462 passed / 1447 passed

cargo test -p datafusion-datasource-arrow -p datafusion-datasource-csv \
           -p datafusion-datasource-json -p datafusion-datasource-parquet --lib
  11 passed / 3 passed / 46 passed / 113 passed
  (16 pre-existing failures in datasource-parquet, all "failed to get parquet data dir:
   env PARQUET_TEST_DATA is undefined" — the parquet-testing submodule is not checked out
   locally; none reach the changed code)

cargo clippy --all-targets on all seven crates — no errors, and no warning in any changed file
cargo fmt on the seven touched crates

cargo fmt --all is not clean on this branch's base, so formatting was applied per-crate to keep
unrelated files out of the diff.

Notes

This is also a live bug upstream — apache/datafusion main still has the unreachable!() in
collect_partitioned.

The runtime side of spiceai/spiceai#7030 needs the datafusion rev in spiceai/spiceai's
Cargo.toml bumped after this merges; the issue stays open until that lands.

[Copilot]

Pull request overview

Fixes cancellation handling in seven JoinSet drain loops, addressing spiceai/spiceai#7030.

Changes:

• Converts cancelled tasks into ExecutionJoin errors while preserving panic propagation.
• Applies consistent handling across collection and file-writing paths.
• Adds unit and regression tests for cancellation and panic behavior.

Reviewed changes

Copilot reviewed 8 out of 8 changed files in this pull request and generated 1 comment.

Show a summary per file

File	Description
datafusion/common/src/error.rs	Adds the shared JoinError conversion helper and tests.
datafusion/physical-plan/src/execution_plan.rs	Handles cancelled partition tasks and adds a regression test.
datafusion/catalog/src/memory/table.rs	Handles cancellation while loading partitions.
datafusion/datasource-arrow/src/file_format.rs	Handles cancelled Arrow writer tasks.
datafusion/datasource-parquet/src/sink.rs	Handles cancelled Parquet sink tasks.
datafusion/datasource-parquet/src/writer.rs	Handles cancelled Parquet plan writers.
datafusion/datasource-csv/src/source.rs	Handles cancelled CSV writer tasks.
datafusion/datasource-json/src/source.rs	Handles cancelled JSON writer tasks.

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.

[Copilot]

Pull request overview

Copilot reviewed 8 out of 8 changed files in this pull request and generated no new comments.

Suppressed comments (2)

datafusion/common/src/error.rs:138

• Runtime shutdown cancels every outstanding asynchronous task, including tasks that have already started and yielded; it is not limited to tasks that are still queued. Narrowing this public contract to queued tasks could cause callers to misclassify a cancellation from an in-progress task. Please describe the condition as the runtime shutting down before the task completes.

    /// task that was **cancelled** does: it was aborted, or the runtime it was
    /// spawned on shut down while it was still queued. The `JoinError` stays

datafusion/common/src/error.rs:439

• This repeats the narrower “still queued” condition, but dropping a Tokio runtime cancels all outstanding async tasks, including ones that were previously polled and are pending. Please document cancellation as shutdown before completion so this helper's public contract covers all JoinError::is_cancelled() cases.

    /// * The task was **cancelled** — it was aborted, or the runtime it was spawned
    ///   on shut down while it was still queued. That is returned as