diff --git a/datafusion/expr/src/window_state.rs b/datafusion/expr/src/window_state.rs index acba75268b50..d50476b03fe0 100644 --- a/datafusion/expr/src/window_state.rs +++ b/datafusion/expr/src/window_state.rs @@ -259,7 +259,7 @@ impl WindowFrameContext { ); } WindowFrameBound::Following(ScalarValue::UInt64(Some(n))) => { - std::cmp::min(idx + n as usize, length) + std::cmp::min(idx.saturating_add(n as usize), length) } // ERRONEOUS FRAMES WindowFrameBound::Preceding(_) | WindowFrameBound::Following(_) => { @@ -284,7 +284,7 @@ impl WindowFrameContext { // UNBOUNDED FOLLOWING WindowFrameBound::Following(ScalarValue::UInt64(None)) => length, WindowFrameBound::Following(ScalarValue::UInt64(Some(n))) => { - std::cmp::min(idx + n as usize + 1, length) + std::cmp::min(idx.saturating_add(n as usize).saturating_add(1), length) } // ERRONEOUS FRAMES WindowFrameBound::Preceding(_) | WindowFrameBound::Following(_) => { diff --git a/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs b/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs index 092570b05197..aad4f691deb5 100644 --- a/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs +++ b/datafusion/physical-optimizer/src/limit_pushdown_past_window.rs @@ -176,8 +176,8 @@ fn grow_limit(window: &BoundedWindowAggExec, ctx: &mut TraverseState) -> bool { ctx.max_lookahead(end_bound); } - // finish grow - ctx.max_lookahead(ctx.lookahead + max_rel); + // finish grow — saturate so a huge FOLLOWING bound cannot overflow + ctx.max_lookahead(ctx.lookahead.saturating_add(max_rel)); true } @@ -194,16 +194,22 @@ fn apply_limit( ctx.lookahead = 0; return Some(Transformed::no(Arc::clone(node))); }; + // Saturate like EnsureRequirements sort pushdown: a huge window + // FOLLOWING bound must not wrap the TopK fetch in release builds. let fetch = match node.fetch() { - None => fetch + ctx.lookahead, - Some(existing) => cmp::min(existing, fetch + ctx.lookahead), + None => fetch.saturating_add(ctx.lookahead), + Some(existing) => cmp::min(existing, fetch.saturating_add(ctx.lookahead)), }; Some(Transformed::complete(node.with_fetch(Some(fetch)).unwrap())) } fn get_limit(node: &Arc, ctx: &mut TraverseState) -> bool { if let Some(limit) = node.downcast_ref::() { - ctx.reset_limit(limit.fetch().map(|fetch| fetch + limit.skip())); + ctx.reset_limit( + limit + .fetch() + .map(|fetch| fetch.saturating_add(limit.skip())), + ); return true; } // In distributed execution, GlobalLimitExec becomes LocalLimitExec diff --git a/datafusion/sqllogictest/test_files/window_limits.slt b/datafusion/sqllogictest/test_files/window_limits.slt index 11c96904f4ea..480c503e6dee 100644 --- a/datafusion/sqllogictest/test_files/window_limits.slt +++ b/datafusion/sqllogictest/test_files/window_limits.slt @@ -862,3 +862,17 @@ 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 + +# Huge FOLLOWING bound must not overflow LimitPushPastWindows (issue #26658) +statement ok +CREATE TABLE t_huge_following(a INT, x INT) AS VALUES (1, 1), (2, 2), (3, 3), (4, 4), (5, 5); + +query II +SELECT a, sum(x) OVER (ORDER BY a ROWS BETWEEN CURRENT ROW AND 18446744073709551615 FOLLOWING) s +FROM t_huge_following ORDER BY a LIMIT 2; +---- +1 15 +2 14 + +statement ok +DROP TABLE t_huge_following;