You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by gi...@apache.org on 2026/09/17 05:18:03 UTC
(datafusion) branch main updated: fix(physical-plan): CoalescePartitionsExec panic on wasm32-unknown-unknown (#24890)
This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion.git
The following commit(s) were added to refs/heads/main by this push:
new 0bef032a4a fix(physical-plan): CoalescePartitionsExec panic on wasm32-unknown-unknown (#24890)
0bef032a4a is described below
commit 0bef032a4a42a2d809c0090cc6703ca1788184a3
Author: Kent Wu <ke...@gmail.com>
AuthorDate: Thu Sep 17 05:09:30 2026 +0000
fix(physical-plan): CoalescePartitionsExec panic on wasm32-unknown-unknown (#24890)
## Which issue does this PR close?
- Closes #24886.
## Rationale for this change
Multi-partition queries panic on `wasm32-unknown-unknown` when they go
through `CoalescePartitionsExec`. The most common case is `UNION ALL`,
but any plan with more than one partition feeding the operator hits it
(e.g. a multi-file external table scan). The operator spawns a task per
input partition via `JoinSet::spawn`, which needs a tokio reactor that
isn't available under `wasm-bindgen-futures`:
```
panicked at datafusion/common-runtime/src/join_set.rs:69:20:
there is no reactor running, must be called from the context of a Tokio 1.x runtime
```
This is a follow-up to #24275, which fixed the same class of panic in
`collect_partitioned`.
## What is the testing strategy for this PR?
- On `wasm32` builds, `CoalescePartitionsExec::execute` no longer spawns
tasks. It polls all input partitions together with
`futures::stream::select_all` and merges their output into a single
stream.
- Native builds keep the existing spawn-based path unchanged.
## Are these changes tested?
- Adds a `test_union_all` regression test in `datafusion/wasmtest`. It
panics on unpatched `main` and passes with this change.
- CI runs it in the browser via `wasm-pack test --headless --firefox`
and `--chrome`. The native `tokio::test` run covers the unchanged spawn
path.
- The test sorts results before comparing, since
`CoalescePartitionsExec` makes no output-order guarantee.
## Are there any user-facing changes?
No API changes. On `wasm32` targets, queries that coalesce multiple
partitions no longer panic, regardless of `target_partitions`. Native
behavior is unchanged.
---
.../physical-plan/src/coalesce_partitions.rs | 22 ++++++++++++++++++++++
datafusion/wasmtest/src/lib.rs | 22 +++++++++++++++++++++-
2 files changed, 43 insertions(+), 1 deletion(-)
diff --git a/datafusion/physical-plan/src/coalesce_partitions.rs b/datafusion/physical-plan/src/coalesce_partitions.rs
index 7438f1c0ca..9e3811e0ad 100644
--- a/datafusion/physical-plan/src/coalesce_partitions.rs
+++ b/datafusion/physical-plan/src/coalesce_partitions.rs
@@ -21,6 +21,8 @@
use std::sync::Arc;
use super::metrics::{BaselineMetrics, ExecutionPlanMetricsSet, MetricsSet};
+#[cfg(target_arch = "wasm32")]
+use super::stream::RecordBatchStreamAdapter;
use super::stream::{ObservedStream, RecordBatchReceiverStream};
use super::{
DisplayAs, ExecutionPlanProperties, PlanProperties, SendableRecordBatchStream,
@@ -229,6 +231,26 @@ impl ExecutionPlan for CoalescePartitionsExec {
}
_ => {
let baseline_metrics = BaselineMetrics::new(&self.metrics, partition);
+
+ // wasm has no tokio reactor, so avoid `JoinSet::spawn`; poll
+ // input partitions cooperatively via `select_all` instead.
+ #[cfg(target_arch = "wasm32")]
+ {
+ let elapsed_compute = baseline_metrics.elapsed_compute().clone();
+ let _timer = elapsed_compute.timer();
+
+ let mut streams = Vec::with_capacity(input_partitions);
+ for i in 0..input_partitions {
+ streams.push(self.input.execute(i, Arc::clone(&context))?);
+ }
+ let merged = futures::stream::select_all(streams);
+ return Ok(Box::pin(ObservedStream::new(
+ Box::pin(RecordBatchStreamAdapter::new(self.schema(), merged)),
+ baseline_metrics,
+ self.fetch,
+ )));
+ }
+
// record the (very) minimal work done so that
// elapsed_compute is not reported as 0
let elapsed_compute = baseline_metrics.elapsed_compute().clone();
diff --git a/datafusion/wasmtest/src/lib.rs b/datafusion/wasmtest/src/lib.rs
index d8da5d4b4f..8e9a7160cd 100644
--- a/datafusion/wasmtest/src/lib.rs
+++ b/datafusion/wasmtest/src/lib.rs
@@ -90,7 +90,9 @@ mod test {
execution::context::SessionContext,
prelude::CsvReadOptions,
};
- use datafusion_common::{DataFusionError, test_util::batches_to_string};
+ use datafusion_common::{
+ DataFusionError, assert_batches_sorted_eq, test_util::batches_to_string,
+ };
use datafusion_execution::{
config::SessionConfig,
disk_manager::{DiskManagerBuilder, DiskManagerMode},
@@ -234,6 +236,24 @@ mod test {
);
}
+ #[wasm_bindgen_test(unsupported = tokio::test)]
+ async fn test_union_all() {
+ // Regression: UNION ALL used to panic on wasm via `JoinSet::spawn`.
+ let ctx = get_ctx();
+ let result = ctx
+ .sql("SELECT 1 AS n UNION ALL SELECT 2 AS n")
+ .await
+ .unwrap()
+ .collect()
+ .await
+ .unwrap();
+
+ // CoalescePartitionsExec makes no output-order guarantee; sort
+ // before comparing so the assertion tolerates any valid interleaving.
+ let expected = ["+---+", "| n |", "+---+", "| 1 |", "| 2 |", "+---+"];
+ assert_batches_sorted_eq!(expected, &result);
+ }
+
#[wasm_bindgen_test(unsupported = tokio::test)]
async fn test_parquet_write() {
let (schema, batch) = create_test_data();
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org