fix: saturate LimitPushPastWindows and ROWS frame bounds for huge FOLLOWING

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(_) => {

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<dyn ExecutionPlan>, ctx: &mut TraverseState) -> bool {
     if let Some(limit) = node.downcast_ref::<GlobalLimitExec>() {
-        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

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;