fix deadlock

Author: JanKaul

datafusion/physical-plan/src/sorts/sort.rs

@@ -519,7 +519,11 @@ impl ExternalSorter {
         while let Some(batch) = sorted_stream.next().await {
             let batch = batch?;
             let sorted_size = get_reserved_bytes_for_record_batch(&batch)?;
-            if self.reservation.try_grow_async(sorted_size).await.is_err() {
+            // Sync `try_grow`, not `try_grow_async`: we are already in the
+            // spill path (freeing memory). A recursive reclaim here can
+            // close a cycle between two sorters that are each waiting on
+            // the other's spill to complete.
+            if self.reservation.try_grow(sorted_size).is_err() {
                 // Although the reservation is not enough, the batch is
                 // already in memory, so it's okay to combine it with previously
                 // sorted batches, and spill together.
@@ -1310,6 +1314,16 @@ impl ExecutionPlan for SortExec {
                             tokio::select! {
                                 biased;
                                 Some(resp_tx) = reclaim_rx.recv() => {
+                                    // A reclaim can be dequeued just after a
+                                    // prior spill drained `in_mem_batches`
+                                    // (sibling sent during the spill's awaits;
+                                    // pool's zero-byte filter can transiently
+                                    // miss us via split reservations). Nothing
+                                    // local to free — reply 0 and keep going.
+                                    if sorter.in_mem_batches.is_empty() {
+                                        let _ = resp_tx.send(0);
+                                        continue;
+                                    }
                                     let before = sorter.used();
                                     sorter.sort_and_spill_in_mem_batches().await?;
                                     let after = sorter.used();