geoffreyclaude commented on code in PR #26736:
URL: https://github.com/apache/datafusion/pull/26736#discussion_r4551689056
##########
datafusion/physical-optimizer/src/pushdown_sort.rs:
##########
@@ -167,11 +167,15 @@ impl PhysicalOptimizerRule for PushdownSort {
//
// If the SortExec carried a fetch (LIMIT), we must
preserve it.
// First try pushing the limit into the source via
`with_fetch()`.
- // If the source doesn't support `with_fetch`, fall back to
- // wrapping with GlobalLimitExec.
+ // If the source doesn't support `with_fetch`, preserve
whether
+ // the limit applies to each partition or to the entire
input.
if let Some(fetch) = sort_exec.fetch() {
let inner =
inner.with_fetch(Some(fetch)).unwrap_or_else(|| {
- Arc::new(GlobalLimitExec::new(inner, 0,
Some(fetch)))
+ if sort_exec.preserve_partitioning() {
+ Arc::new(LocalLimitExec::new(inner, fetch))
+ } else {
+ Arc::new(GlobalLimitExec::new(inner, 0,
Some(fetch)))
Review Comment:
Added `exact_pushdown_preserves_global_fetch`: it runs `PushdownSort` on a
standalone global TopK over one partition, then asserts the resulting physical
plan and the three expected rows. The plan assertion checks that
`GlobalLimitExec` remains, catching an incorrect local limit that would return
the same rows here.
--
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]
