jayzhan211 commented on code in PR #26264:
URL: https://github.com/apache/datafusion/pull/26264#discussion_r4500714782
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -372,10 +376,25 @@ impl ExternalSorter {
batch_size,
sort_spill_reservation_bytes,
sort_in_place_threshold_bytes,
+ partitions_reading_input: Arc::new(AtomicUsize::new(1)),
sort_concurrency: Arc::new(OnceLock::new()),
})
}
+ /// Share the count of partitions still reading their input with the
+ /// other partitions of this sort. It must include this partition.
+ fn with_partitions_reading_input(mut self, reading: Arc<AtomicUsize>) ->
Self {
+ self.partitions_reading_input = reading;
+ self
+ }
+
+ /// Records that this partition has read all its input, and returns whether
+ /// another partition still reads its own. Executing partitions again
+ /// without `reset_state` can wrap the count, which only keeps headroom.
Review Comment:
Optional cleanup: on the first re-run the count is 0, so this partition gets
`false` and keeps no headroom, then the count wraps.
```suggestion
/// Records that this partition has read all its input, and returns
whether
/// another partition still reads its own. Executing partitions again
/// without `reset_state` leaves the count wrong, which changes how much
/// headroom a merge keeps but never its output.
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
