jayzhan211 commented on code in PR #26662:
URL: https://github.com/apache/datafusion/pull/26662#discussion_r4369538964
##########
datafusion/physical-optimizer/src/limit_pushdown_past_window.rs:
##########
@@ -194,16 +197,21 @@ fn apply_limit(
ctx.lookahead = 0;
return Some(Transformed::no(Arc::clone(node)));
};
+ let fetch = fetch.saturating_add(ctx.lookahead);
Review Comment:
When this sum saturates, the sort below becomes `SortExec:
TopK(fetch=18446744073709551615)`. TopK keeps every row in memory and can't
spill. With 2M rows and a 24MB memory limit, the issue query fails with
`Resources exhausted ... TopK[0]`. The same query with `enable_window_limits =
false` returns the right rows, because the full sort spills. A fetch that
overflows can't drop any rows, so could we skip the pushdown instead of
saturating?
```sql
set datafusion.execution.target_partitions = 1;
CREATE TABLE big AS SELECT (value * 7919) % 1000003 AS a, value AS x FROM
generate_series(1, 2000000);
set datafusion.runtime.memory_limit = '24M';
set datafusion.execution.sort_spill_reservation_bytes = 1048576;
SELECT a, sum(x) OVER (ORDER BY a ROWS BETWEEN CURRENT ROW AND
18446744073709551615 FOLLOWING) s
FROM big ORDER BY a LIMIT 2;
```
```rs
let Some(fetch) = fetch.checked_add(ctx.lookahead) else {
ctx.lookahead = 0;
return Some(Transformed::no(Arc::clone(node)));
};
let fetch = match node.fetch() {
None => fetch,
Some(existing) => cmp::min(existing, fetch),
};
```
With this change the query succeeds, the plan shows a plain `SortExec`, and
`window_limits.slt` still passes.
##########
datafusion/sqllogictest/test_files/window_limits.slt:
##########
@@ -862,3 +862,26 @@ physical_plan
03)----BoundedWindowAggExec: wdw=[lead(employees.salary,Int64(2)) ORDER BY
[employees.empno ASC NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT
ROW: Field { "lead(employees.salary,Int64(2)) ORDER BY [employees.empno ASC
NULLS LAST] ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW": nullable UInt64
}, frame: ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW], mode=[Sorted]
04)------SortExec: TopK(fetch=5), expr=[empno@0 ASC NULLS LAST],
preserve_partitioning=[false]
05)--------DataSourceExec: file_groups={1 group:
[[WORKSPACE_ROOT/testing/data/csv/aggregate_test_100_with_dates.csv]]},
projection=[empno, salary], file_type=csv, has_header=true
+
+# A huge FOLLOWING bound must not overflow when the limit is grown by the
Review Comment:
These two cases cover the frame end and the `grow_limit` sum, but nothing
reaches the `Following` start bound in `calculate_range_rows`. On main that
line overflows for any row after the first. Could you add this case? It returns
the rows below with your change.
```sql
query II
SELECT a, sum(x) OVER (ORDER BY a ROWS BETWEEN 18446744073709551615
FOLLOWING AND 18446744073709551615 FOLLOWING) s
FROM huge_frame_t ORDER BY a LIMIT 2;
----
1 NULL
2 NULL
```
--
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]
