[shahwazreza]

Which issue does this PR close?

• Closes ORDER BY ... LIMIT 0 panics in TopK (assertion failed: k > 0) when eliminate_limit does not run #26431

Rationale for this change

SELECT ... ORDER BY ... LIMIT 0 panics instead of returning an empty result when the limit makes it to the physical plan. Normally eliminate_limit removes LIMIT 0 before that happens, so most people won't run into it. But it does reach execution with datafusion.optimizer.max_passes = 0, or when someone builds the physical plan directly. In that case SortExec with fetch = Some(0) builds a TopK with k = 0, and TopKHeap::new hits assert!(k > 0).

What changes are included in this PR?

• SortExec::execute handles fetch = Some(0) on unsorted input before it gets to the TopK path, and returns an empty stream.
• That partition still gets marked as emitted on the shared TopK dynamic filter. Without this, the filter would wait forever for a partition that never reports in. mark_topk_emitted is now pub(crate) so SortExec can call it.

I left the assert!(k > 0) in TopKHeap alone. Removing it wouldn't be enough anyway, since TopKHeap::add also assumes k > 0 (it calls peek_mut().unwrap() when len == k), so handling the case in SortExec is simpler.

What is the testing strategy for this PR?

• Unit test test_sort_fetch_zero_returns_empty_and_completes_filter in sort.rs. It runs a two-partition SortExec with fetch = Some(0) and checks that both partitions return no rows. It also checks that the dynamic filter completes only after the last partition has run.
• sqllogictest case in limit.slt that sets max_passes = 0, so the LIMIT 0 survives into the physical plan. It checks the EXPLAIN output (SortExec: TopK(fetch=0)) and that the query returns no rows.

Are there any user-facing changes?

The query now returns an empty result instead of panicking.

[2010YOUY01]

Thank you for the fix.

I can understand why enter this branch and return empty set, but I don't get those three lines, for example why don't the above LimitStream branch have to special handle dynamic filter like this

[shahwazreza]

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.

[2010YOUY01]

See https://datafusion.apache.org/contributor-guide/testing.html#choosing-what-kind-of-test-to-write

This UT is asserting on low-level implementations (probing into the dynamic filter), make it hard to understand and maintain.

We could either:

• write it as only construct the SortExec and run
• or remove it, we already have slt I think it's good already

[shahwazreza]

Thanks! Since you're fine with the slt, I'll remove this.

[jayzhan211]

Thanks @shahwazreza @2010YOUY01 !