File tree Expand file tree Collapse file tree
datafusion/physical-plan/src/repartition Expand file tree Collapse file tree Original file line number Diff line number Diff line change 4040//! 1. *Gate A* — every output channel has at least one buffered item. This is the earliest legitimate backpressure
4141//! point: every consumer has work to do, so throttling the producer is safe.
4242//! 2. *Gate B* — total in-memory bytes buffered across all channels reaches the configured `max_buffered_bytes`. This
43- //! bounds memory under skewed fan-out, where Gate A would never fire because some channel is always empty.
43+ //! bounds memory whenever some channel is always empty, e.g. skewed fan-out, or even balanced fan-out where one
44+ //! consumer lags: the producer keeps adding to the slow channel each round Gate A briefly opens.
4445//!
4546//! While the gate is closed, sender futures are [pending](Poll::Pending) until either a receiver drains an item or
4647//! enough bytes are released to reopen the gate.
Original file line number Diff line number Diff line change @@ -1503,7 +1503,8 @@ impl RepartitionExec {
15031503 } ;
15041504
15051505 // Spilled markers occupy ~no in-channel memory; only count Memory variants against the gate's
1506- // byte budget.
1506+ // byte budget. `size` may over-count for shared buffers (e.g. dictionary values across hash
1507+ // partitions); over-counting is conservative — it just blocks producers slightly earlier.
15071508 let send_bytes = if is_memory_batch { size } else { 0 } ;
15081509
15091510 if channel
You can’t perform that action at this time.
0 commit comments