File tree Expand file tree Collapse file tree
ballista/core/src/execution_plans Expand file tree Collapse file tree Original file line number Diff line number Diff line change @@ -592,12 +592,6 @@ impl ShuffleWriterExec {
592592 } ) ;
593593 }
594594
595- // Joined as they complete, not in partition order: the K drains
596- // share one plan instance, so a panic while its `CollectLeft` build
597- // side (DataFusion's `OnceFut`) is polled leaves the others parked
598- // on that future forever — joining in order would block on one of
599- // them and hang the job instead of failing it. Dropping the
600- // `JoinSet` on the way out aborts whatever is still parked.
601595 let mut results = Vec :: with_capacity ( num_partitions) ;
602596 while let Some ( joined) = handles. join_next ( ) . await {
603597 let ( local_input_partition, stats) = joined. map_err ( |e| {
Original file line number Diff line number Diff line change @@ -1063,12 +1063,6 @@ async fn run_coordinator(
10631063 // Spawn per-input-partition write tasks concurrently — this matches the
10641064 // parallelism main's design achieved by having DataFusion drive
10651065 // `execute(0..M)` concurrently.
1066- //
1067- // Joined as they complete, not in partition order, and the first failure
1068- // abandons the rest: the M writes share one plan instance, so a panic while
1069- // its `CollectLeft` build side (DataFusion's `OnceFut`) is polled leaves the
1070- // others parked on that future forever — joining in order would block on one
1071- // of them and hang the job instead of failing it.
10721066 let mut writes = JoinSet :: new ( ) ;
10731067 for input_partition in 0 ..num_input_partitions {
10741068 let w = writer. clone ( ) ;
You can’t perform that action at this time.
0 commit comments