rgbuilds commented on code in PR #26432:
URL: https://github.com/apache/datafusion/pull/26432#discussion_r4301209947
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -1630,12 +1630,14 @@ impl ExecutionPlan for SortExec {
Ok(Box::pin(RecordBatchStreamAdapter::new(
self.schema(),
futures::stream::once(async move {
- while let Some(batch) = input.next().await {
- let batch = batch?;
- topk.insert_batch(batch)?;
- if topk.finished {
+ // Check `finished` before polling so that a TopK that
+ // is already complete (e.g. `fetch = 0`) does not read
+ // any input.
+ while !topk.finished {
Review Comment:
Non-blocking test suggestion: could we add a focused
`SortExec::with_fetch(Some(0))` execution test using `PanicExec` or another
first-poll-failing stream?
The unit test covers direct `TopK` construction, insertion, and emission,
while the SLT covers the plan and empty result, but neither would detect a
future change that polls the input once before checking `finished`. I verified
locally that such a `PanicExec` test passes with this implementation.
--
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]
