geoffreyclaude commented on code in PR #26242:
URL: https://github.com/apache/datafusion/pull/26242#discussion_r4282183201
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -98,6 +102,38 @@ impl ExternalSorterMetrics {
}
}
+/// Limits concurrent batch sorting across the partitions of one [`SortExec`].
+#[derive(Debug, Clone)]
+struct SortConcurrency {
+ /// Each partition can use the full worker budget, but queues at most this
+ /// many jobs so it cannot put all its runs ahead of other partitions.
+ per_partition_limit: usize,
+ /// Shared budget: a job holds a permit until its run has been sorted.
+ permits: Arc<Semaphore>,
+}
+
+impl SortConcurrency {
+ /// Returns the shared limit on multithreaded Tokio runtimes, initialized
+ /// from the current runtime's worker count on first use.
+ fn for_current_runtime(cell: &OnceLock<Self>) -> Option<Self> {
+ let handle = Handle::try_current().ok()?;
+ if handle.runtime_flavor() != RuntimeFlavor::MultiThread {
+ return None;
+ }
+ Some(
+ cell.get_or_init(|| {
+ // Sorting is CPU-bound; more jobs would only queue for a
worker.
+ let workers = handle.metrics().num_workers();
Review Comment:
Good point. I suspect the proper limit should actually be the min of the
two. Going over `available_parallelism` won't bring anything, but if the
runtime was configured with less workers than `available_parallelism` that's
probably a conscious decision and we should use that instead.
I'll check how it's done elsewhere in the codebase and align to that
behavior for consistency.
--
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]
