distributor_channels: zero-byte sends bypass the gate

Sentinel messages (end-of-stream `None`, error markers, spill markers)
all flow through `DistributionSender::send` with `bytes = 0`. Previously
they were subject to the same park-on-closed-gate logic as data sends,
which deadlocked the pipeline under skew:

  1. One output channel accumulates enough data to push total bytes
     above `max_buffered_bytes`. Gate B closes.
  2. An input pump finishes its source and `wait_for_task` sends a
     `None` sentinel to every output channel.
  3. The sentinel targeting the swollen non-empty channel parks behind
     the queued data (gate closed && !was_empty).
  4. The consumer of the swollen channel keeps draining, but the
     sentinel never reaches it, so `remaining_partitions` never
     decrements and the consumer never learns the producer is done.
  5. Result: AggFinal (or any downstream consumer) waits forever for
     EOF on partitions whose pumps completed long ago.

A zero-byte send contributes nothing to the byte budget, so throttling
it is semantically meaningless and operationally fatal. Allow it
through unconditionally, keeping the EOF/error/spill control plane
decoupled from the data-plane byte budget.

Empirically, this is the deadlock at trace_lines ~169k–288k in TPC-H
Q18 SF=100 with prefer_hash_join=false on the reclaimer composite:
all 16 Partial aggregates EOF, 8 of 16 Repartition pumps for the
l_orderkey hash partition signal output EOF; the other 8 are
indefinitely parked sending sentinels to the swollen channel.

Existing tests that used `.send(X, 0)` for data payloads were updated
to `.send(X, 1)` so they continue to exercise the gate; a new test
asserts the zero-byte bypass directly.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: JanKaul

datafusion/physical-plan/src/repartition/distributor_channels.rs

@@ -48,6 +48,11 @@
 //!
 //! As a soft cap, an empty channel is always allowed one push regardless of Gate B (the "overdraw escape"). Without
 //! this, a single batch larger than `max_buffered_bytes` would head-of-line block its consumer indefinitely.
+//!
+//! Zero-byte sends (end-of-stream `None`, error markers, spilled-batch markers) also bypass the gate
+//! unconditionally — they carry no in-memory bytes, so the byte budget cannot meaningfully back-pressure them, and
+//! parking them behind a swollen-channel backlog would starve the EOF/error/spill control plane and deadlock the
+//! pipeline.
 use std::{
     collections::VecDeque,
     future::Future,
@@ -252,12 +257,22 @@ impl<T> Future for SendFuture<'_, T> {
 
             let was_empty = data.is_empty();
 
-            // Park if the gate is closed *and* this channel is not empty.
+            // Park if the gate is closed *and* this channel is not empty *and* this push would
+            // actually consume gate budget (`bytes > 0`).
+            //
+            // Three escapes from parking:
+            //
+            // - Overdraw escape: if `was_empty` is true, this channel's consumer has nothing to chew on. Blocking the
+            //   push would starve the consumer. Gate A cannot be closed when `was_empty` is true (the empty-channels
+            //   count is at least 1), so `was_empty` only short-circuits Gate B.
             //
-            // Overdraw escape: if `was_empty` is true, this channel's consumer has nothing to chew on. Blocking the
-            // push would starve the consumer. Gate A cannot be closed when `was_empty` is true (the empty-channels
-            // count is at least 1), so `was_empty` only short-circuits Gate B.
-            if !was_empty && this.gate.is_closed() {
+            // - Zero-byte escape: sentinel sends (end-of-stream `None`, error markers, spill markers) carry no
+            //   in-memory bytes. Throttling them serves no purpose (they don't grow the budget) and parking them
+            //   deadlocks the pipeline: an EOF sentinel queued behind a backlog on a swollen channel never reaches
+            //   the consumer, who never decrements `remaining_partitions`, who never learns the producer finished.
+            //   Allowing zero-byte sends through unconditionally keeps the EOF/error/spill control plane independent
+            //   of the data-plane byte budget.
+            if this.bytes > 0 && !was_empty && this.gate.is_closed() {
                 // Re-check under the send_wakers lock to avoid a lost-wakeup race:
                 // - if the gate is still closed, park.
                 // - if the gate opened between our first read and acquiring this lock, fall through and push.
@@ -570,13 +585,13 @@ mod tests {
         let mut recv_fut = rxs[0].recv();
         let waker = poll_pending(&mut recv_fut);
 
-        poll_ready(&mut txs[0].send("foo", 0)).unwrap();
+        poll_ready(&mut txs[0].send("foo", 1)).unwrap();
         assert!(waker.woken());
         assert_eq!(poll_ready(&mut recv_fut), Some("foo"),);
 
-        poll_ready(&mut txs[0].send("bar", 0)).unwrap();
-        poll_ready(&mut txs[0].send("baz", 0)).unwrap();
-        poll_ready(&mut txs[0].send("end", 0)).unwrap();
+        poll_ready(&mut txs[0].send("bar", 1)).unwrap();
+        poll_ready(&mut txs[0].send("baz", 1)).unwrap();
+        poll_ready(&mut txs[0].send("end", 1)).unwrap();
         assert_eq!(poll_ready(&mut rxs[0].recv()), Some("bar"),);
         assert_eq!(poll_ready(&mut rxs[0].recv()), Some("baz"),);
 
@@ -594,8 +609,8 @@ mod tests {
 
         let tx_clone = txs[0].clone();
 
-        poll_ready(&mut txs[0].send("foo", 0)).unwrap();
-        poll_ready(&mut tx_clone.send("bar", 0)).unwrap();
+        poll_ready(&mut txs[0].send("foo", 1)).unwrap();
+        poll_ready(&mut tx_clone.send("bar", 1)).unwrap();
 
         assert_eq!(poll_ready(&mut rxs[0].recv()), Some("foo"),);
         assert_eq!(poll_ready(&mut rxs[0].recv()), Some("bar"),);
@@ -606,17 +621,17 @@ mod tests {
         let (txs, mut rxs) = channels(2, UNLIMITED);
 
         // gate initially open
-        poll_ready(&mut txs[0].send("0_a", 0)).unwrap();
+        poll_ready(&mut txs[0].send("0_a", 1)).unwrap();
 
         // gate still open because channel 1 is still empty
-        poll_ready(&mut txs[0].send("0_b", 0)).unwrap();
+        poll_ready(&mut txs[0].send("0_b", 1)).unwrap();
 
         // gate still open because channel 1 is still empty prior to this call, so this call still goes through
-        poll_ready(&mut txs[1].send("1_a", 0)).unwrap();
+        poll_ready(&mut txs[1].send("1_a", 1)).unwrap();
 
         // both channels non-empty => gate closed
 
-        let mut send_fut = txs[1].send("1_b", 0);
+        let mut send_fut = txs[1].send("1_b", 1);
         let waker = poll_pending(&mut send_fut);
 
         // drain channel 0
@@ -639,34 +654,34 @@ mod tests {
 
         let mut recv_fut = rxs[0].recv();
 
-        poll_ready(&mut tx1.send("a", 0)).unwrap();
+        poll_ready(&mut tx1.send("a", 1)).unwrap();
         let recv_waker = poll_pending(&mut recv_fut);
 
         // drop original sender
         drop(tx0);
 
         // not yet closed (there's a clone left)
         assert!(!recv_waker.woken());
-        poll_ready(&mut tx1.send("b", 0)).unwrap();
+        poll_ready(&mut tx1.send("b", 1)).unwrap();
         let recv_waker = poll_pending(&mut recv_fut);
 
         // create new clone
         let tx0_clone2 = tx0_clone.clone();
         assert!(!recv_waker.woken());
-        poll_ready(&mut tx1.send("c", 0)).unwrap();
+        poll_ready(&mut tx1.send("c", 1)).unwrap();
         let recv_waker = poll_pending(&mut recv_fut);
 
         // drop first clone
         drop(tx0_clone);
         assert!(!recv_waker.woken());
-        poll_ready(&mut tx1.send("d", 0)).unwrap();
+        poll_ready(&mut tx1.send("d", 1)).unwrap();
         let recv_waker = poll_pending(&mut recv_fut);
 
         // drop last clone
         drop(tx0_clone2);
 
         // channel closed => also close gate
-        poll_pending(&mut tx1.send("e", 0));
+        poll_pending(&mut tx1.send("e", 1));
         assert!(recv_waker.woken());
         assert_eq!(poll_ready(&mut recv_fut), None,);
     }
@@ -678,14 +693,14 @@ mod tests {
         let rx0 = rxs.remove(0);
         let _rx1 = rxs.remove(0);
 
-        poll_ready(&mut txs[1].send("a", 0)).unwrap();
+        poll_ready(&mut txs[1].send("a", 1)).unwrap();
 
         // drop receiver => also close gate
         drop(rx0);
 
-        poll_pending(&mut txs[1].send("b", 0));
+        poll_pending(&mut txs[1].send("b", 1));
         assert_eq!(
-            poll_ready(&mut txs[0].send("foo", 0)),
+            poll_ready(&mut txs[0].send("foo", 1)),
             Err(SendError("foo")),
         );
     }
@@ -698,11 +713,11 @@ mod tests {
         let mut rx1 = rxs.remove(0);
 
         // fill both channels
-        poll_ready(&mut txs[0].send("0_a", 0)).unwrap();
-        poll_ready(&mut txs[1].send("1_a", 0)).unwrap();
+        poll_ready(&mut txs[0].send("0_a", 1)).unwrap();
+        poll_ready(&mut txs[1].send("1_a", 1)).unwrap();
 
-        let mut send_fut0 = txs[0].send("0_b", 0);
-        let mut send_fut1 = txs[1].send("1_b", 0);
+        let mut send_fut0 = txs[0].send("0_b", 1);
+        let mut send_fut1 = txs[1].send("1_b", 1);
         let waker0 = poll_pending(&mut send_fut0);
         let waker1 = poll_pending(&mut send_fut1);
 
@@ -732,9 +747,9 @@ mod tests {
         let _rx2 = rxs.remove(0);
 
         // fill channels
-        poll_ready(&mut tx0.send("0_a", 0)).unwrap();
-        poll_ready(&mut tx1.send("1_a", 0)).unwrap();
-        poll_ready(&mut tx2.send("2_a", 0)).unwrap();
+        poll_ready(&mut tx0.send("0_a", 1)).unwrap();
+        poll_ready(&mut tx1.send("1_a", 1)).unwrap();
+        poll_ready(&mut tx2.send("2_a", 1)).unwrap();
 
         // drop / close one channel
         drop(rx1);
@@ -743,9 +758,9 @@ mod tests {
         assert_eq!(poll_ready(&mut rx0.recv()), Some("0_a"),);
 
         // use senders again
-        poll_ready(&mut tx0.send("0_b", 0)).unwrap();
-        assert_eq!(poll_ready(&mut tx1.send("1_b", 0)), Err(SendError("1_b")),);
-        poll_pending(&mut tx2.send("2_b", 0));
+        poll_ready(&mut tx0.send("0_b", 1)).unwrap();
+        assert_eq!(poll_ready(&mut tx1.send("1_b", 1)), Err(SendError("1_b")),);
+        poll_pending(&mut tx2.send("2_b", 1));
     }
 
     #[test]
@@ -757,7 +772,7 @@ mod tests {
         assert_eq!(counter.strong_count(), 1);
 
         // add object to channel
-        poll_ready(&mut txs[0].send(obj, 0)).unwrap();
+        poll_ready(&mut txs[0].send(obj, 1)).unwrap();
         assert_eq!(counter.strong_count(), 1);
 
         // drop receiver
@@ -778,14 +793,14 @@ mod tests {
         let mut recv_fut = rxs[0].recv();
         let waker_2 = poll_pending(&mut recv_fut);
 
-        poll_ready(&mut txs[0].send("a", 0)).unwrap();
+        poll_ready(&mut txs[0].send("a", 1)).unwrap();
         assert!(waker_1a.woken());
         assert!(waker_1b.woken());
         assert!(waker_2.woken());
         assert_eq!(poll_ready(&mut recv_fut), Some("a"),);
 
-        poll_ready(&mut txs[0].send("b", 0)).unwrap();
-        let mut send_fut = txs[0].send("c", 0);
+        poll_ready(&mut txs[0].send("b", 1)).unwrap();
+        let mut send_fut = txs[0].send("c", 1);
         let waker_3 = poll_pending(&mut send_fut);
         assert_eq!(poll_ready(&mut rxs[0].recv()), Some("b"),);
         assert!(waker_3.woken());
@@ -798,8 +813,8 @@ mod tests {
         let mut recv_fut = rxs[0].recv();
         let waker_5 = poll_pending(&mut recv_fut);
 
-        poll_ready(&mut txs[0].send("d", 0)).unwrap();
-        let mut send_fut = txs[0].send("e", 0);
+        poll_ready(&mut txs[0].send("d", 1)).unwrap();
+        let mut send_fut = txs[0].send("e", 1);
         let waker_6a = poll_pending(&mut send_fut);
         let waker_6b = poll_pending(&mut send_fut);
 
@@ -816,7 +831,7 @@ mod tests {
     #[should_panic(expected = "polled ready future")]
     fn test_panic_poll_send_future_after_ready_ok() {
         let (txs, _rxs) = channels(1, UNLIMITED);
-        let mut fut = txs[0].send("foo", 0);
+        let mut fut = txs[0].send("foo", 1);
         poll_ready(&mut fut).unwrap();
         poll_ready(&mut fut).ok();
     }
@@ -828,7 +843,7 @@ mod tests {
 
         drop(rxs);
 
-        let mut fut = txs[0].send("foo", 0);
+        let mut fut = txs[0].send("foo", 1);
         poll_ready(&mut fut).unwrap_err();
         poll_ready(&mut fut).ok();
     }
@@ -838,7 +853,7 @@ mod tests {
     fn test_panic_poll_recv_future_after_ready_some() {
         let (txs, mut rxs) = channels(1, UNLIMITED);
 
-        poll_ready(&mut txs[0].send("foo", 0)).unwrap();
+        poll_ready(&mut txs[0].send("foo", 1)).unwrap();
 
         let mut fut = rxs[0].recv();
         poll_ready(&mut fut).unwrap();
@@ -961,6 +976,31 @@ mod tests {
         assert!(waker_b.woken());
     }
 
+    /// Zero-byte sends (EOF sentinels, error markers, spill markers) bypass the gate even when the gate is
+    /// closed and the target channel is non-empty. Without this escape, EOF queued behind a data backlog on a
+    /// swollen channel would never reach the consumer.
+    #[test]
+    fn test_zero_byte_send_bypasses_closed_gate() {
+        let (txs, mut rxs) = channels(2, 100);
+
+        // Saturate the byte budget on ch0 (overdraw on the second push allowed because the first put us at the limit).
+        poll_ready(&mut txs[0].send("a", 50)).unwrap();
+        poll_ready(&mut txs[0].send("b", 50)).unwrap();
+        // Gate is now closed on Gate B (bytes >= max).
+
+        // A non-zero-byte send to non-empty ch0 parks.
+        let mut data_fut = txs[0].send("data", 1);
+        poll_pending(&mut data_fut);
+
+        // A zero-byte send to the same non-empty ch0 must NOT park — it carries no bytes, so throttling is
+        // meaningless and parking would block EOF/error/spill control plane progress.
+        poll_ready(&mut txs[0].send("eof", 0)).unwrap();
+
+        // Drain so the data future can also complete.
+        assert_eq!(poll_ready(&mut rxs[0].recv()), Some("a"));
+        poll_ready(&mut data_fut).unwrap();
+    }
+
     /// Dropping a receiver returns its bytes to the budget and wakes parked senders on surviving channels.
     #[test]
     fn test_drop_receiver_returns_bytes_to_budget() {