jayzhan211 commented on code in PR #26719:
URL: https://github.com/apache/datafusion/pull/26719#discussion_r4443461389
##########
datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/replace_with_order_preserving_variants.rs:
##########
@@ -97,6 +98,7 @@ pub fn plan_with_order_preserving_variants(
sort_input.data = false;
return Ok(sort_input);
}
+ let child_fetch = fetch.filter(|_|
sort_input.plan.supports_limit_pushdown());
Review Comment:
@Kristofferjoha thanks for the fix! Since #26660 merged,
`supports_limit_pushdown` is deprecated, so once this is merged with main,
clippy fails with `use of deprecated method` on this line. Main already has the
same "can a fetch go below this node" rule in sort pushdown
(`can_push_fetch_through`), plus `limit_node` for "remove the sort, keep its
fetch as a limit" (`enforce_sorting/mod.rs`). I tried this on top of main: the
full slt suite and the `physical_optimizer` tests pass with no plan changes.
```diff
---
a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs
+++
b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/sort_pushdown.rs
-fn can_push_fetch_through(plan: &Arc<dyn ExecutionPlan>) -> bool {
+pub(crate) fn can_push_fetch_through(plan: &Arc<dyn ExecutionPlan>) -> bool
{
---
a/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/replace_with_order_preserving_variants.rs
+++
b/datafusion/physical-optimizer/src/ensure_requirements/enforce_sorting/replace_with_order_preserving_variants.rs
+use super::sort_pushdown::can_push_fetch_through;
+use crate::row_bound::{RowBound, limit_node};
-use datafusion_physical_plan::limit::{GlobalLimitExec, LocalLimitExec};
@@
- let child_fetch = fetch.filter(|_|
sort_input.plan.supports_limit_pushdown());
+ let child_fetch = fetch.filter(|_|
can_push_fetch_through(&sort_input.plan));
@@
if let Some(fetch) = requirements.plan.fetch() {
- let plan = if
requirements.plan.output_partitioning().partition_count() == 1 {
- let limit =
- GlobalLimitExec::new(Arc::clone(&alternate_plan.plan), 0,
Some(fetch));
- Arc::new(limit) as _
- } else {
- let limit =
LocalLimitExec::new(Arc::clone(&alternate_plan.plan), fetch);
- Arc::new(limit) as _
- };
+ let plan = limit_node(
+ Arc::clone(&alternate_plan.plan),
+ RowBound::fetch(Some(fetch)),
+ );
alternate_plan = PlanContext::new(plan, false,
vec![alternate_plan]);
}
```
This also stops the fetch at limit nodes. That fixes a case that returns 0
rows today: an `OFFSET` subquery over a sorted multi-file table, like the one
below. It returns 1 row with the change, and `LimitPushdown` still puts
`fetch=3` on the merge. Could you add it to `window_limits.slt` (or
`limit.slt`) next to your cases?
```sql
-- sorted_t: two sorted CSV files, WITH ORDER (a ASC), target_partitions = 4
SELECT count(*) FROM (SELECT * FROM (SELECT a FROM sorted_t LIMIT 3 OFFSET
2) ORDER BY a LIMIT 1);
-- expected 1
```
--
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]
