shahwazreza commented on code in PR #27000:
URL: https://github.com/apache/datafusion/pull/27000#discussion_r4439016680
##########
datafusion/physical-plan/src/sorts/sort.rs:
##########
@@ -1794,6 +1794,18 @@ impl ExecutionPlan for SortExec {
BaselineMetrics::new(&self.metrics_set, partition),
))),
(true, None) => Ok(input),
+ // `TopK` requires `k > 0`. `LIMIT 0` is normally removed by the
+ // logical optimizer, but a `SortExec` with `fetch = Some(0)` can
+ // still reach execution, for example when the logical optimizer
+ // is disabled or the plan is built directly.
+ (false, Some(0)) => {
+ // Count this partition as emitted so the shared TopK dynamic
+ // filter completes once every partition has finished.
+ let filter = self.filter.as_ref();
+ let filter = unwrap_or_internal_err!(filter);
+ filter.read().mark_topk_emitted();
Review Comment:
I added them because the new arm takes the place of TopK, and TopK::emit()
calls mark_topk_emitted() for each partition so the shared filter counts down
and gets marked complete. I mirrored that so the filter ends in the same state
as a normal TopK run, since I was worried something waiting on wait_complete()
could hang otherwise.
Looking at it again, you have a point, nothing outside tests waits on
completion, and the LimitStream branch already leaves the filter incomplete
without issues, so it isn't needed. I'll remove those lines so this arm just
returns an empty stream, which also lets mark_topk_emitted go back to private.
--
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]
