Merge phase23-running-frames: running frames + checked-SUM order-pin soundness fix

Author: Aly700

ARCHITECTURE.md

@@ -21,8 +21,9 @@
 19. Boolean-mask filter kernels. Delivered in Phase 19: vectorized Filter now evaluates WHERE/HAVING predicate trees as paired `is_true`/`is_known` byte masks using SQL Kleene 3VL algebra, with comparison leaves scanning pre-resolved int64 or string values column-at-a-time and `IS [NOT] NULL` leaves reading validity directly. Scan-adjacent identity selections use a dense full-batch domain, while filtered/non-identity inputs use dense selection-position masks that map back to the immutable input selection. Join residual predicates deliberately remain on the existing row-pair path. Release benchmark evidence in `docs/benchmarks.md` shows the three scan/filter vectorized medians improving with byte-identical correctness checks and no material non-target regression.
 20. LEFT/RIGHT OUTER JOIN arc. Delivered across Phase 20a/20b: the parser accepts `LEFT [OUTER] JOIN` and `RIGHT [OUTER] JOIN`; binding normalizes RIGHT to LEFT while stable identities preserve SQL-visible projection order; both interpreted and vectorized engines implement deterministic left-row-major NULL-extension, NULL-key non-matching, and residual-aware matchedness for hash and non-equi joins. `LeftJoinToInnerRule` additively simplifies a filtered LEFT join only under a proof-bearing conservative 3VL null-rejection test, unlocking the existing INNER commute/associate/pushdown exploration without weakening their guards. The deterministic fuzzer now mixes INNER/LEFT/RIGHT chains with actual NULL-bearing key data and verifies standalone rewrites, every memo alternative, and `extract_best` through both engines. The Phase 20a LEFT cardinality rule remains sufficient; no 20b cost-model change was required. The outer-join lowering guard and its verifier path are gone.
 21. Complete subquery arc. Delivered across Phase 21a/21b/21c: scalar-comparison, IN/NOT IN, and EXISTS/NOT EXISTS use immutable bound subplans with explicit lexical-depth correlation metadata. Empty correlation retains the eager once-per-query 3VL/cardinality model; correlated forms use the per-outer-row interpreted specification with nested/grandparent frames and row-ordered scalar errors. Explicit left SEMI/ANTI algebra has left-only output identity, deterministic interpreted/vectorized hash/NLJ execution, memo identity/printers/EXPLAIN/costing, and conservative guards. Uncorrelated and immediate equality-correlated top-level EXISTS, NOT EXISTS, and IN decorrelate through proof-bearing memo rules. Residual correlated forms remain oracle-defined and fail once at physical lowering; uncorrelated residuals still run vectorized recursively. Exact EXPLAIN goldens, eager/per-row error equivalence, and deterministic uncorrelated/correlated fuzz lanes close the story.
-22. Window functions arc. Delivered across Phase 22a/22b: whole-SELECT-item `ROW_NUMBER`, `RANK`, and `DENSE_RANK` plus whole-partition `SUM`, `COUNT`, `MIN`, and `MAX` parse and bind into one post-HAVING Window node; both interpreted and vectorized execution implement first-appearance partitioning, input-order preservation, stable ranking ties, NULL partition/peer equality, checked/NULL-skipping partition aggregates, and replicated results; Window participates in physical lowering, memo identity, EXPLAIN, extraction, conservative costing, golden coverage, deterministic fuzzing, and a Release benchmark workload while remaining a filter-pushdown barrier, with ROW_NUMBER and checked SUM additionally pinning order-changing child transforms because stable ties and intermediate overflow expose input order. Running aggregate windows and explicit `ROWS`/`RANGE` frames remain future work.
-23. Null-aware anti join (future). Extend algebra and execution before considering NOT IN decorrelation; ordinary Anti remains intentionally insufficient for a NULL-bearing right set, where unmatched NOT IN is UNKNOWN rather than TRUE.
+22. Window functions arc. Delivered across Phase 22a/22b: whole-SELECT-item `ROW_NUMBER`, `RANK`, and `DENSE_RANK` plus whole-partition `SUM`, `COUNT`, `MIN`, and `MAX` parse and bind into one post-HAVING Window node; both interpreted and vectorized execution implement first-appearance partitioning, input-order preservation, stable ranking ties, NULL partition/peer equality, checked/NULL-skipping partition aggregates, and replicated results; Window participates in physical lowering, memo identity, EXPLAIN, extraction, conservative costing, golden coverage, deterministic fuzzing, and a Release benchmark workload while remaining a filter-pushdown barrier, with ROW_NUMBER and whole-partition checked SUM pinning order-changing child transforms because stable ties and intermediate overflow expose input order.
+23. Running window frames. Delivered in Phase 23: aggregate OVER ORDER BY defaults to `RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW`, and the same explicit RANGE spelling plus `ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW` bind as distinct frame modes. RANGE publishes one result through each complete peer group and canonically orders checked SUM arguments within peers, making output and overflow category independent of incidental tie order. ROWS publishes each stable input-order prefix and pins order-changing child transforms; a checked Aggregate SUM below any Window pins independently. The oracle, vectorized operator, memo alternatives, deterministic fuzzer, goldens, and standalone benchmark cover the new paths while whole-partition behavior remains unchanged. Every other frame bound or modifier fails as a positioned unsupported frame.
+24. Null-aware anti join (future). Extend algebra and execution before considering NOT IN decorrelation; ordinary Anti remains intentionally insufficient for a NULL-bearing right set, where unmatched NOT IN is UNKNOWN rather than TRUE.
 
 ## Phase 1 first tasks
 
@@ -45,4 +46,4 @@ Do not optimize before correctness. Every phase should end with an executable de
 - Broaden fuzz coverage when the verifier can prove sortedness for ORDER BY keys that are not emitted in the final output.
 - Add value-width-aware string costing and string-specific performance work after collecting more benchmark evidence.
 - Investigate aggregation and sort kernels with benchmark evidence before changing their implementations.
-- Add running window frames (`ORDER BY` for aggregate windows and explicit `ROWS`/`RANGE BETWEEN`) only after frame bounds are represented explicitly in the algebra and both engines can share differential semantics.
+- Add bounded, following, `GROUPS`, and exclusion window frames only after their endpoints and exclusion rules are represented explicitly in the algebra and their order/error observability is proved across memo alternatives.

ROADMAP.md

@@ -366,6 +366,11 @@ std::vector<Workload> make_workloads(const execution::Catalog& catalog) {
         "ROW_NUMBER() OVER (PARTITION BY bucket ORDER BY sort_key) AS rn, "
         "SUM(value) OVER (PARTITION BY bucket) AS bucket_sum "
         "FROM fact");
+    add("window_running_sum",
+        "fact=200000,partitions=100",
+        "SELECT bucket, sort_key, value, "
+        "SUM(value) OVER (PARTITION BY bucket ORDER BY sort_key) AS running_sum "
+        "FROM fact");
     add("sort",
         "fact=200000",
         "SELECT a, value FROM fact ORDER BY sort_key, a LIMIT 1000");

benchmarks/sql_bench.cpp

@@ -248,3 +248,15 @@ phase.
 | workload | rows | correctness | interpreted min ms | interpreted median ms | vectorized min ms | vectorized median ms | median speedup |
 |---|---:|---|---:|---:|---:|---:|---:|
 | window_row_number_sum | fact=200000,partitions=100 | match rows=200000 checksum=0x1f5b8dc9ce10a38c | 725.385 | 732.916 | 108.253 | 109.029 | 6.722x |
+
+## Phase 23 Running Window Frame (2026-07-13)
+
+Same optimized build, five-repetition min/median, deterministic-data, and
+checksum-before-timing methodology as above. This workload measures the
+default cumulative `RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW` frame
+for `SUM(value) OVER (PARTITION BY bucket ORDER BY sort_key)` across the
+200,000-row `fact` table and its 100 partitions.
+
+| workload | rows | correctness | interpreted min ms | interpreted median ms | vectorized min ms | vectorized median ms | median speedup |
+|---|---:|---|---:|---:|---:|---:|---:|
+| window_running_sum | fact=200000,partitions=100 | match rows=200000 checksum=0xefdfa729da68f560 | 704.578 | 711.366 | 102.685 | 105.883 | 6.718x |

docs/benchmarks.md

@@ -156,6 +156,7 @@ struct WindowExpression {
     std::optional<BoundColumnRef> argument;
     std::vector<BoundColumnRef> partition_keys;
     std::vector<SortKey> order_keys;
+    sql::WindowFrame frame{sql::WindowFrame::WholePartition};
     std::size_t position{0};
     catalog::ColumnType type{catalog::ColumnType::Int64};
 };

include/plan/logical_plan.hpp

@@ -55,6 +55,7 @@ struct AggregateCall {
 enum class SortDirection { Asc, Desc };
 
 enum class WindowFunction { RowNumber, Rank, DenseRank, Count, Sum, Min, Max };
+enum class WindowFrame { WholePartition, RangeCumulative, RowsCumulative };
 
 using WindowInputExpr = std::variant<ColumnRef, AggregateCall>;
 
@@ -71,6 +72,8 @@ struct WindowCall {
     std::optional<WindowInputExpr> argument;
     std::vector<WindowInputExpr> partition_by;
     std::vector<WindowOrderKey> order_by;
+    std::optional<WindowFrame> explicit_frame;
+    std::size_t frame_position{0};
 };
 
 using SelectExpr = std::variant<ScalarExpr, AggregateCall, WindowCall>;
@@ -350,6 +353,18 @@ inline std::string window_function_name(WindowFunction function) {
     return "<unknown window function>";
 }
 
+inline std::string window_frame_name(WindowFrame frame) {
+    switch (frame) {
+    case WindowFrame::WholePartition:
+        return "WHOLE PARTITION";
+    case WindowFrame::RangeCumulative:
+        return "RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW";
+    case WindowFrame::RowsCumulative:
+        return "ROWS BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW";
+    }
+    return "<unknown window frame>";
+}
+
 inline std::string quote_string_literal(const std::string& value) {
     std::string quoted = "'";
     for (const auto ch : value) {
@@ -418,6 +433,7 @@ inline std::string output_name(const WindowCall& window) {
         name += window_input_output_name(*window.argument);
     }
     name += ") OVER (";
+    bool wrote_clause = false;
     if (!window.partition_by.empty()) {
         name += "PARTITION BY ";
         for (std::size_t i = 0; i < window.partition_by.size(); ++i) {
@@ -426,9 +442,10 @@ inline std::string output_name(const WindowCall& window) {
             }
             name += window_input_output_name(window.partition_by[i]);
         }
+        wrote_clause = true;
     }
     if (!window.order_by.empty()) {
-        if (!window.partition_by.empty()) {
+        if (wrote_clause) {
             name += " ";
         }
         name += "ORDER BY ";
@@ -439,6 +456,13 @@ inline std::string output_name(const WindowCall& window) {
             name += window_input_output_name(window.order_by[i].expression);
             name += window.order_by[i].direction == SortDirection::Asc ? " ASC" : " DESC";
         }
+        wrote_clause = true;
+    }
+    if (window.explicit_frame.has_value()) {
+        if (wrote_clause) {
+            name += " ";
+        }
+        name += window_frame_name(*window.explicit_frame);
     }
     name += ")";
     return name;

include/sql/ast.hpp

@@ -599,6 +599,15 @@ std::int64_t checked_sum(std::int64_t left, std::int64_t right, const std::strin
     return result;
 }
 
+void update_sum(AggregateValue& value, std::int64_t argument, const std::string& output_name) {
+    if (!value.has_value) {
+        value.sum = argument;
+        value.has_value = true;
+        return;
+    }
+    value.sum = checked_sum(value.sum, argument, output_name);
+}
+
 Cell aggregate_argument_value(const plan::AggregateExpression& aggregate,
                               const storage::ColumnarBatch& batch,
                               std::size_t row,
@@ -631,12 +640,7 @@ void update_aggregate(AggregateValue& value,
         if (argument.is_null) {
             return;
         }
-        if (!value.has_value) {
-            value.sum = argument.int_value;
-            value.has_value = true;
-            return;
-        }
-        value.sum = checked_sum(value.sum, argument.int_value, aggregate.output_name);
+        update_sum(value, argument.int_value, aggregate.output_name);
         return;
     }
     case sql::AggregateFunction::Min: {
@@ -1121,6 +1125,48 @@ std::vector<WindowPartition> build_window_partitions(const plan::WindowExpressio
     return partitions;
 }
 
+std::vector<std::size_t> ordered_window_rows(const plan::WindowExpression& window,
+                                             const WindowPartition& partition,
+                                             const storage::ColumnarBatch& input,
+                                             ExecutionContext& context) {
+    auto ordered_rows = partition.rows;
+    std::stable_sort(ordered_rows.begin(), ordered_rows.end(), [&](std::size_t left, std::size_t right) {
+        return window_row_less(window.order_keys, input, left, right, context);
+    });
+    return ordered_rows;
+}
+
+void update_range_peer(AggregateValue& state,
+                       const plan::AggregateExpression& aggregate,
+                       const std::vector<std::size_t>& ordered_rows,
+                       std::size_t begin,
+                       std::size_t end,
+                       const storage::ColumnarBatch& input,
+                       ExecutionContext& context) {
+    if (aggregate.function != sql::AggregateFunction::Sum) {
+        for (std::size_t index = begin; index < end; ++index) {
+            update_aggregate(state, aggregate, input, ordered_rows[index], context);
+        }
+        return;
+    }
+
+    // RANGE peers share one observable frame boundary. Canonicalizing the
+    // peer's non-NULL SUM arguments makes checked intermediate overflow a
+    // function of the peer-keyed bag rather than incidental stable tie order.
+    std::vector<std::int64_t> arguments;
+    arguments.reserve(end - begin);
+    for (std::size_t index = begin; index < end; ++index) {
+        const auto argument = aggregate_argument_value(aggregate, input, ordered_rows[index], context);
+        if (!argument.is_null) {
+            arguments.push_back(argument.int_value);
+        }
+    }
+    std::sort(arguments.begin(), arguments.end());
+    for (const auto argument : arguments) {
+        update_sum(state, argument, aggregate.output_name);
+    }
+}
+
 OutputColumn evaluate_window(const plan::WindowExpression& window,
                              const storage::ColumnarBatch& input,
                              ExecutionContext& context) {
@@ -1135,10 +1181,7 @@ OutputColumn evaluate_window(const plan::WindowExpression& window,
         case sql::WindowFunction::RowNumber:
         case sql::WindowFunction::Rank:
         case sql::WindowFunction::DenseRank: {
-            auto ordered_rows = partition.rows;
-            std::stable_sort(ordered_rows.begin(), ordered_rows.end(), [&](std::size_t left, std::size_t right) {
-                return window_row_less(window.order_keys, input, left, right, context);
-            });
+            const auto ordered_rows = ordered_window_rows(window, partition, input, context);
 
             std::size_t rank = 1;
             std::size_t dense_rank = 1;
@@ -1162,12 +1205,44 @@ OutputColumn evaluate_window(const plan::WindowExpression& window,
         case sql::WindowFunction::Max: {
             const auto aggregate = aggregate_expression_for_window(window);
             AggregateValue state;
-            for (const auto row : partition.rows) {
-                update_aggregate(state, aggregate, input, row, context);
+            if (window.frame == sql::WindowFrame::WholePartition) {
+                for (const auto row : partition.rows) {
+                    update_aggregate(state, aggregate, input, row, context);
+                }
+                const auto value = finalize_aggregate(state, aggregate);
+                for (const auto row : partition.rows) {
+                    values.at(row) = value;
+                }
+                break;
             }
-            const auto value = finalize_aggregate(state, aggregate);
-            for (const auto row : partition.rows) {
-                values.at(row) = value;
+
+            const auto ordered_rows = ordered_window_rows(window, partition, input, context);
+            if (window.frame == sql::WindowFrame::RowsCumulative) {
+                for (const auto row : ordered_rows) {
+                    update_aggregate(state, aggregate, input, row, context);
+                    values.at(row) = finalize_aggregate(state, aggregate);
+                }
+                break;
+            }
+
+            std::size_t peer_begin = 0;
+            while (peer_begin < ordered_rows.size()) {
+                auto peer_end = peer_begin + 1;
+                while (peer_end < ordered_rows.size() &&
+                       window_rows_are_peers(window.order_keys,
+                                             input,
+                                             ordered_rows[peer_begin],
+                                             ordered_rows[peer_end],
+                                             context)) {
+                    ++peer_end;
+                }
+                update_range_peer(
+                    state, aggregate, ordered_rows, peer_begin, peer_end, input, context);
+                const auto value = finalize_aggregate(state, aggregate);
+                for (auto index = peer_begin; index < peer_end; ++index) {
+                    values.at(ordered_rows[index]) = value;
+                }
+                peer_begin = peer_end;
             }
             break;
         }