Repository navigation
RepartitionExec channels grow unboundedly with one slow consumer #22090
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:
- After the first input batch each channel has ≥1 item → Gate A closes.
- ch0 drains → Gate A opens.
- 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. - 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.
Activity
alamb commentedon May 10, 2026
This is by design to avoid deadlocks
If you have a situation where one of the channels is empty, are you guaranteed that the other non empty channels can make progress? For example the classic diamond plan
If you have one consumer falling behind, I think better strategy might be to apply back pressure at the consumer end (rather than the Repartition)
What are the consumers in this case?
JanKaul commentedon May 10, 2026
Thanks for the quick reply! Let me try to defend my position for the sake of exploring the solution space.
With my proposal, the buffered bytes are capped at a certain value, but every time a record batch is processed by a consumer, its bytes are released, freeing up headroom. This essentially means the RepartitionExec stays "open" as long as it receives and releases memory at roughly the same rate. So yes, it can theoretically lead to more deadlocks.
That said, the situation under which it "locks" is generally an undesirable one to begin with: it's the case where a RepartitionExec continuously receives more bytes than it can release. This can of course be legitimately required by certain downstream consumers, as your example shows. But there is typically a limit to how much data downstream consumers need in order to make progress, and that's what this cap is meant to enforce. The cap could be rather large, allowing for the scenarios you mentioned. I'd argue that if the unhealthy state of receiving more than is being released persists for too long, it eventually leads to OOM, which isn't much better than a deadlock. The problem is that the higher memory consumption of RepartitionExec leads to even slower downstream consumers because they are forced to spill which leads to some kind of self-amplification effect.
Again, thanks a lot for taking the time to explain. I'll try to dig deeper into it and see if there are other possibilities to apply more backpressure. Like you suggested.
A bit of context of what I'm trying to achieve. So I'm trying to run TPCH query 17 sf=100 in a memory constrained environment with either 4gb or 8gb of memory. I'm using SortMergeJoin as the Join algorithm. And I'm experimenting with diffent(also custom) memory pools.
The query is:
I run into the problem that only one ExternalSorter partition continues working and 5.3gb of my 8gb memory pool is stuck in the RepartitionExec.
alamb commentedon May 11, 2026
No problem!
This is not just a theory -- we have seen such deadlocks in production at InfluxData -- typically on very skewed datasets
I agree the situation is undesirable, though I disagree with the root cause. The root cause is that there is some downstream operator consuming partitions at very different rates
I don't see how the cap will ensure that the downstream consumer can make progress
I wonder if the problem is that some of the SMJ partitions have chosen to spill but others have not (yet) resulting in different rates... maybe you could look into forcing all SMJ partitions to spill if any does
adriangb commentedon May 11, 2026
Could an a answer be to tweak how
RepartitionExecitself spills? I.e. instead of accumulating data in memory it spills when there's a lot of skew. I think it already spills in some conditions (memory pressure?) but maybe it should spill more aggressively (e.g. only keep 1 batch in memory at a time).Related: if the partition that is spilling is because it's upstream operator is slow because the upstream operator spilled, I wonder if it would be beneficial to have a
enum InFlightData { Memory(RecordBatch), Disk { file, start, end }or something like that. The point is: if we are going from one spilling operator to another maybe pushing the data around on disk instead of loading only to spill it again would make sense. But that'd be a big change.JanKaul commentedon May 11, 2026
All right, thanks for the input. I thought this would generally be a good idea, but I don't have as much experience with deadlocks. I'll focus on solving the problem downstream such that the consumers work more uniformly.
Passing the spilled files from one operator to another would only work if the operator doesn't change the data, which is probably not that often.