RepartitionExec channels grow unboundedly with one slow consumer
- Dominant language
- Rust
- Stars
- 9.3k
- Forks
- 2.4k
- Avg merge
- 3d 7h
- Merged PRs (30d)
- 344
Description
### Describe the bug
`RepartitionExec`'s distribution channels (`distributor_channels.rs`) only throttle producers when **every** output channel has at least one buffered item ("Gate A"). With balanced hash partitioning and a single slow consumer, the slow channel grows linearly per input batch.
### To reproduce
Trace with N=4 partitions, ch3's consumer slower than the others:
1. After the first input batch each channel has ≥1 item → Gate A closes.
2. ch0 drains → Gate A opens.
3. Producer pushes the next input batch's sub-batches: one `take`-materialized batch per partition. Push to ch0 succeeds (was empty); ch1, ch2, ch3 already had data, so the empty-counter never decrements during the round.
4. After the round all channels are non-empty again → Gate A re-closes. ch3 has gained 1.
Each cycle adds another batch to ch3 — linear growth in the number of input batches, until OOM (or until the per-partition `SharedMemoryReservation` triggers spilling).
### Cause
Gate A's invariant is "no live channel is empty." A single fast channel oscillating empty/non-empty keeps the gate open forever, and every gate-open window lets the producer push to **every** partition — including the slow one. Skewed input data is *not* required; even hash distribution + one lagging consumer is enough.
### Expected behavior
The producer should be throttled when total buffered memory crosses a configured threshold, regardless of how many channels are technically non-empty.
### Proposed fix
Add a second gate condition — total buffered bytes across all channels ≥ a configured budget — and close the gate when **either** condition fires (`A || B`). Gate A still gives O(1) per-channel depth for balanced workloads with even consumer rates; Gate B caps total memory whenever some channel never drains. Velox's `LocalExchangeMemoryManager` is the design template.
I'll send a PR shortly.
Contributor guide
Research direction
Start in distributor_channels.rs and trace how RepartitionExec tracks channel occupancy and buffered bytes. Compare the proposed total-memory gate with Velox's LocalExchangeMemoryManager design, then verify that the producer throttles when either gate condition is reached and that a slow consumer no longer permits unbounded growth.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- rust
- Domain
- data-engineering, distributed-systems, performance
- Issue type
- Bug
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Mostly clear
- Newbie friendliness
- 48/100