yinli-systems commented on code in PR #24890:
URL: https://github.com/apache/datafusion/pull/24890#discussion_r3918690517
##########
datafusion/physical-plan/src/coalesce_partitions.rs:
##########
@@ -229,6 +232,27 @@ impl ExecutionPlan for CoalescePartitionsExec {
}
_ => {
let baseline_metrics = BaselineMetrics::new(&self.metrics,
partition);
+
+ // Single-threaded path: drain sequentially to avoid
+ // `JoinSet::spawn`, which needs a tokio reactor and panics
+ // on wasm32-unknown-unknown.
+ if context.session_config().target_partitions() == 1 {
+ // mirror the slow path so elapsed_compute isn't 0 here
+ let elapsed_compute =
baseline_metrics.elapsed_compute().clone();
+ let _timer = elapsed_compute.timer();
+
+ let input = Arc::clone(&self.input);
+ let ctx = Arc::clone(&context);
+ let stream = futures::stream::iter(0..input_partitions)
Review Comment:
Could we avoid sequential `try_flatten` here and scope the fallback to
`wasm32`? `target_partitions == 1` does not imply that the child has one
independent partition--this branch is reached precisely because it has more
than one. Polling partition 0 to completion before partition 1 can deadlock for
exchange-backed inputs; the existing test at `repartition/mod.rs:4473`
explicitly drains every output partition concurrently to avoid its
bounded-channel gate deadlock. This also changes native scheduling for every
session configured with `target_partitions = 1`, although the reactor failure
is wasm-specific. A `#[cfg(target_arch = "wasm32")]` path that wraps
`futures::stream::select_all(input_streams)` in `RecordBatchStreamAdapter`
avoids `JoinSet::spawn` while still polling all partitions cooperatively; the
existing `RecordBatchReceiverStream` path can remain unchanged for non-wasm
targets. I reproduced #24886 and verified that shape with 9/9 Chrome wasm-pack
tests, all 9 coalesce tes
ts, `rust_lint.sh`, and the full extended workspace suite.
##########
datafusion/wasmtest/src/lib.rs:
##########
@@ -234,6 +234,29 @@ 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();
+
+ assert_eq!(
+ batches_to_string(&result),
Review Comment:
`CoalescePartitionsExec` explicitly makes no output-order guarantee, so this
assertion couples the regression to the current partition order. Could this use
`assert_batches_sorted_eq!` instead? That still verifies both rows while
allowing valid interleavings.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
