geoffreyclaude opened a new pull request, #26242: URL: https://github.com/apache/datafusion/pull/26242
## Which issue does this PR close? Addresses the unbounded local-sort task population discussed in [#23174](https://github.com/apache/datafusion/issues/23174). ## Rationale for this change A sort receiving many small batches can delay unrelated queries for seconds by flooding the Tokio runnable queue. `ExternalSorter` starts a buffered producer for every sorted run. The one-batch buffer limits each producer's output, while the number of producers grows with the number of input batches. With TPC-H SF1 `lineitem` rebatch-sized to 32 rows, a four-column sort on four partitions and four Tokio workers reached **187,571 live DataFusion tasks** and approximately **186,000 queued tasks**. An unrelated, preplanned `SELECT 1` waited **4.33 seconds** before its first poll, then executed in microseconds. ## What changes are included in this PR? Prepare sorted runs through a bounded set of finite tasks. All partitions of one `SortExec` share a semaphore sized to the actual polling runtime's worker count. Acquire a permit before spawning a job, so pending batches remain ordinary buffered input rather than additional Tokio tasks. A single or skewed partition can use the whole shared budget. Each job polls the existing batch-sort stream through its first output batch, returns the prepared stream, and releases its permit. This preserves the first-batch memory handoff and allows spill workspace to be reused. The merge receives streams in their original run order. Dropping preparation cancels outstanding jobs through the existing traced `JoinSet`. Each run still uses one job; the bound applies to concurrent jobs. Sorting kernels, batch boundaries, coalescing policy and merge behavior are preserved. Current-thread and non-Tokio execution continue sorting inline. The concurrency state is initialized when execution is polled and reset with the operator's execution state. ## What is the testing strategy for this PR? Regression coverage checks task cleanup before output consumption, exact two-key sorted output, early output drop, preparation-error propagation, and reuse of constrained spill workspace between jobs. Existing sort, merge and spill tests exercise the surrounding paths. The same fragmented TPC-H workload reduced peak live traced tasks to **8** and the global runnable queue to **1**. Fixed runs returned all 6,001,215 rows; the longest unrelated-query start delays were **130–350 ms**. The remaining latency warrants separate investigation. The existing eleven-query `sort-tpch` benchmark matrix and interleaved comparisons were also run. The ordinary-throughput impact remains unresolved because host-load outliers affected both versions. Benchmark artifacts and the standalone reproducer are kept separate from this fix. The modified physical-plan crate passes all-target, all-feature Clippy. Full-workspace, all-feature Clippy encountered existing `mutable_key_type` warnings in `datafusion-optimizer/src/push_down_filter.rs` when Arrow memory tracking is enabled. That crate and its dependencies are unchanged by this PR. The broad test run was stopped after its core integration and randomized aggregation/join suites passed; the final focused sort suite was run separately. ## Are there any user-facing changes? Highly fragmented sorts have a bounded preparation task population shared across partitions. `SortExec` rustdoc documents this execution behavior. Public configuration and API signatures are unchanged. This draft was prepared with AI assistance and is ready for author and upstream design review. -- 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]
