Merge phase22b-vectorized-windows: window arc complete

Author: Aly700

ARCHITECTURE.md

@@ -21,7 +21,7 @@
 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. Phase 22a is delivered: 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; the interpreted oracle defines first-appearance partitioning, input-order preservation, stable ranking ties, NULL partition/peer equality, and replicated NULL-aware aggregate results; Window participates in memo identity, EXPLAIN, extraction, and conservative costing 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; physical lowering fails with one stable unsupported error. Phase 22b will add the real vectorized Window operator and deterministic fuzzer generation. Running aggregate windows and explicit `ROWS`/`RANGE` frames remain future work after 22b.
+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.
 
 ## Phase 1 first tasks

ROADMAP.md

@@ -360,6 +360,12 @@ std::vector<Workload> make_workloads(const execution::Catalog& catalog) {
         "fact=200000,groups=50000",
         "SELECT group_many, SUM(value) AS total "
         "FROM fact GROUP BY group_many ORDER BY group_many");
+    add("window_row_number_sum",
+        "fact=200000,partitions=100",
+        "SELECT bucket, value, "
+        "ROW_NUMBER() OVER (PARTITION BY bucket ORDER BY sort_key) AS rn, "
+        "SUM(value) OVER (PARTITION BY bucket) AS bucket_sum "
+        "FROM fact");
     add("sort",
         "fact=200000",
         "SELECT a, value FROM fact ORDER BY sort_key, a LIMIT 1000");

benchmarks/sql_bench.cpp

@@ -235,3 +235,16 @@ The checksum match pins left-only output and duplicate-preserving Semi
 semantics before timing. The vectorized lookup-only hash implementation is
 2.301x faster on median than the interpreted left-row-major nested-loop oracle
 for this selective workload.
+
+## Phase 22b Window Baseline (2026-07-10)
+
+Same optimized build, five-repetition min/median, deterministic-data, and
+checksum-before-timing methodology as above. This adds one correctness-first
+Window workload over the 200,000-row `fact` table with 100 partitions:
+`ROW_NUMBER()` ordered inside each partition plus `SUM(value)` replicated over
+the partition. No Window-specific performance tuning was attempted in this
+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 |

docs/benchmarks.md

@@ -10,7 +10,7 @@
 
 namespace plan {
 
-enum class PhysicalKind { Scan, Join, Filter, Project, Aggregate, Distinct, Sort, Limit };
+enum class PhysicalKind { Scan, Join, Filter, Project, Aggregate, Window, Distinct, Sort, Limit };
 
 struct PhysicalPlan {
     PhysicalKind kind{PhysicalKind::Scan};
@@ -21,6 +21,7 @@ struct PhysicalPlan {
     std::vector<Projection> projections;
     std::vector<BoundColumnRef> group_keys;
     std::vector<AggregateExpression> aggregate_expressions;
+    std::vector<WindowExpression> window_expressions;
     std::vector<SortKey> sort_keys;
     std::vector<BoundPredicate> predicates;
     JoinKind join_kind{JoinKind::Inner};
@@ -81,6 +82,14 @@ struct PhysicalPlan {
         return p;
     }
 
+    static PhysicalPlan window(std::vector<WindowExpression> window_expressions, PhysicalPlan child) {
+        PhysicalPlan p;
+        p.kind = PhysicalKind::Window;
+        p.window_expressions = std::move(window_expressions);
+        p.input = std::make_shared<PhysicalPlan>(std::move(child));
+        return p;
+    }
+
     static PhysicalPlan distinct(PhysicalPlan child) {
         PhysicalPlan p;
         p.kind = PhysicalKind::Distinct;

include/plan/physical_plan.hpp

@@ -1471,6 +1471,32 @@ TypedCell finalize_aggregate(const AggregateValue& value, const plan::AggregateE
     throw std::logic_error("unreachable aggregate function");
 }
 
+sql::AggregateFunction aggregate_function_for_window(sql::WindowFunction function) {
+    switch (function) {
+    case sql::WindowFunction::Count:
+        return sql::AggregateFunction::Count;
+    case sql::WindowFunction::Sum:
+        return sql::AggregateFunction::Sum;
+    case sql::WindowFunction::Min:
+        return sql::AggregateFunction::Min;
+    case sql::WindowFunction::Max:
+        return sql::AggregateFunction::Max;
+    case sql::WindowFunction::RowNumber:
+    case sql::WindowFunction::Rank:
+    case sql::WindowFunction::DenseRank:
+        break;
+    }
+    throw std::logic_error("ranking window is not an aggregate");
+}
+
+plan::AggregateExpression aggregate_expression_for_window(const plan::WindowExpression& window) {
+    return plan::AggregateExpression{window.output_name,
+                                     aggregate_function_for_window(window.function),
+                                     window.argument,
+                                     window.position,
+                                     window.type};
+}
+
 AggregateGroupState make_aggregate_group(HashKey key, const plan::PhysicalPlan& plan) {
     AggregateGroupState group;
     group.key = std::move(key);
@@ -1721,6 +1747,7 @@ BatchView execute_filter(const plan::PhysicalPlan& plan, ExecutionContext& conte
 BatchView execute_join(const plan::PhysicalPlan& plan, ExecutionContext& context);
 BatchView execute_project(const plan::PhysicalPlan& plan, ExecutionContext& context);
 BatchView execute_aggregate(const plan::PhysicalPlan& plan, ExecutionContext& context);
+BatchView execute_window(const plan::PhysicalPlan& plan, ExecutionContext& context);
 BatchView execute_distinct(const plan::PhysicalPlan& plan, ExecutionContext& context);
 BatchView execute_sort(const plan::PhysicalPlan& plan, ExecutionContext& context);
 BatchView execute_limit(const plan::PhysicalPlan& plan, ExecutionContext& context);
@@ -1737,6 +1764,8 @@ BatchView execute_to_view(const plan::PhysicalPlan& plan, ExecutionContext& cont
         return execute_project(plan, context);
     case plan::PhysicalKind::Aggregate:
         return execute_aggregate(plan, context);
+    case plan::PhysicalKind::Window:
+        return execute_window(plan, context);
     case plan::PhysicalKind::Distinct:
         return execute_distinct(plan, context);
     case plan::PhysicalKind::Sort:
@@ -2065,6 +2094,156 @@ BatchView execute_aggregate(const plan::PhysicalPlan& plan, ExecutionContext& co
     return view;
 }
 
+struct WindowPartition {
+    std::vector<std::size_t> rows;
+};
+
+std::vector<WindowPartition> build_window_partitions(const plan::WindowExpression& window,
+                                                     const BatchView& input) {
+    validate_view(input);
+    const auto partition_keys = compile_columns(window.partition_keys, *input.batch);
+    std::unordered_map<HashKey, std::size_t, HashKeyHash> partition_index_by_key;
+    std::vector<WindowPartition> partitions;
+
+    for (auto row : *input.selection) {
+        auto key = make_key(row, partition_keys);
+        const auto found = partition_index_by_key.find(key);
+        if (found == partition_index_by_key.end()) {
+            const auto index = partitions.size();
+            partition_index_by_key.emplace(std::move(key), index);
+            partitions.push_back(WindowPartition{});
+            partitions.back().rows.push_back(row);
+        } else {
+            partitions.at(found->second).rows.push_back(row);
+        }
+    }
+    return partitions;
+}
+
+bool window_rows_are_peers(const std::vector<CompiledSortKey>& order_keys,
+                           std::size_t left,
+                           std::size_t right) {
+    for (const auto& key : order_keys) {
+        if (!(sort_key_cell(key, left) == sort_key_cell(key, right))) {
+            return false;
+        }
+    }
+    return true;
+}
+
+std::int64_t checked_window_ordinal(std::size_t ordinal, const std::string& output_name) {
+    if (ordinal > static_cast<std::size_t>(std::numeric_limits<std::int64_t>::max())) {
+        throw std::runtime_error(output_name + " overflowed int64");
+    }
+    return static_cast<std::int64_t>(ordinal);
+}
+
+void evaluate_ranking_window(const plan::WindowExpression& window,
+                             const storage::ColumnarBatch& batch,
+                             const std::vector<WindowPartition>& partitions,
+                             std::vector<TypedCell>& values) {
+    const auto order_keys = compile_sort_keys(window.order_keys, batch);
+    for (const auto& partition : partitions) {
+        auto ordered_rows = partition.rows;
+        std::stable_sort(ordered_rows.begin(), ordered_rows.end(), [&](std::size_t left, std::size_t right) {
+            for (const auto& key : order_keys) {
+                const auto left_value = sort_key_cell(key, left);
+                const auto right_value = sort_key_cell(key, right);
+                if (left_value == right_value) {
+                    continue;
+                }
+                return sort_cell_less(left_value, right_value, key.direction);
+            }
+            return false;
+        });
+
+        std::size_t rank = 1;
+        std::size_t dense_rank = 1;
+        for (std::size_t index = 0; index < ordered_rows.size(); ++index) {
+            if (index != 0 && !window_rows_are_peers(order_keys, ordered_rows[index - 1], ordered_rows[index])) {
+                rank = index + 1;
+                ++dense_rank;
+            }
+            const auto ordinal = window.function == sql::WindowFunction::RowNumber
+                                     ? index + 1
+                                     : (window.function == sql::WindowFunction::Rank ? rank : dense_rank);
+            values.at(ordered_rows[index]) = int_cell(checked_window_ordinal(ordinal, window.output_name));
+        }
+    }
+}
+
+void evaluate_aggregate_window(const plan::WindowExpression& window,
+                               const storage::ColumnarBatch& batch,
+                               const std::vector<WindowPartition>& partitions,
+                               std::vector<TypedCell>& values) {
+    const auto aggregate = aggregate_expression_for_window(window);
+    const auto compiled = compile_aggregate_expression(aggregate, batch);
+    for (const auto& partition : partitions) {
+        AggregateValue state;
+        for (auto row : partition.rows) {
+            update_aggregate(state, compiled, row);
+        }
+        const auto value = finalize_aggregate(state, aggregate);
+        for (auto row : partition.rows) {
+            values.at(row) = value;
+        }
+    }
+}
+
+void add_window_column(storage::ColumnarBatch& out,
+                       const plan::WindowExpression& window,
+                       const BatchView& input) {
+    const auto partitions = build_window_partitions(window, input);
+    std::vector<TypedCell> values(input.batch->row_count(), null_cell(window.type));
+
+    switch (window.function) {
+    case sql::WindowFunction::RowNumber:
+    case sql::WindowFunction::Rank:
+    case sql::WindowFunction::DenseRank:
+        evaluate_ranking_window(window, *input.batch, partitions, values);
+        break;
+    case sql::WindowFunction::Count:
+    case sql::WindowFunction::Sum:
+    case sql::WindowFunction::Min:
+    case sql::WindowFunction::Max:
+        evaluate_aggregate_window(window, *input.batch, partitions, values);
+        break;
+    }
+
+    TypedColumnBuilder column(window.type);
+    column.reserve(input.selection->size());
+    for (auto row : *input.selection) {
+        append_cell(column, values.at(row));
+    }
+    std::move(column).add_to(out, window.output_name);
+}
+
+storage::ColumnarBatch materialize_window(const plan::PhysicalPlan& plan, const BatchView& input) {
+    validate_view(input);
+
+    storage::ColumnarBatch out;
+    for (const auto& name : input.batch->column_names()) {
+        add_materialized_selected_column(out, input, name, name);
+    }
+    for (const auto& window : plan.window_expressions) {
+        add_window_column(out, window, input);
+    }
+    return out;
+}
+
+BatchView execute_window(const plan::PhysicalPlan& plan, ExecutionContext& context) {
+    auto input = execute_to_view(require_input(plan), context);
+    auto materialized = std::make_shared<const storage::ColumnarBatch>(materialize_window(plan, input));
+
+    BatchView view;
+    view.owned_batch = materialized;
+    view.batch = materialized.get();
+    view.selection = identity_selection(materialized->row_count());
+    view.selection_rows_match_positions = true;
+    validate_view(view);
+    return view;
+}
+
 BatchView execute_distinct(const plan::PhysicalPlan& plan, ExecutionContext& context) {
     auto input = execute_to_view(require_input(plan), context);
     validate_view(input);
@@ -2216,6 +2395,22 @@ bool plan_requires_typed_dispatch(const plan::PhysicalPlan& plan, const Catalog&
             return true;
         }
     }
+    for (const auto& window : plan.window_expressions) {
+        if (window.type == catalog::ColumnType::String ||
+            (window.argument.has_value() && window.argument->type == catalog::ColumnType::String)) {
+            return true;
+        }
+        for (const auto& key : window.partition_keys) {
+            if (key.type == catalog::ColumnType::String) {
+                return true;
+            }
+        }
+        for (const auto& key : window.order_keys) {
+            if (key.column.type == catalog::ColumnType::String) {
+                return true;
+            }
+        }
+    }
     for (const auto& key : plan.sort_keys) {
         if (key.column.type == catalog::ColumnType::String) {
             return true;
@@ -2236,6 +2431,7 @@ bool plan_requires_typed_dispatch(const plan::PhysicalPlan& plan, const Catalog&
     case plan::PhysicalKind::Filter:
     case plan::PhysicalKind::Project:
     case plan::PhysicalKind::Aggregate:
+    case plan::PhysicalKind::Window:
     case plan::PhysicalKind::Distinct:
     case plan::PhysicalKind::Sort:
     case plan::PhysicalKind::Limit: