Merge phase22a-window-functions: window frontend + oracle

Author: Aly700

ARCHITECTURE.md

@@ -21,7 +21,8 @@
 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. 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. 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.
+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
 
@@ -44,3 +45,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.

ROADMAP.md

@@ -14,7 +14,7 @@ namespace optimizer {
 
 using GroupId = std::uint64_t;
 
-enum class MemoExpressionKind { Scan, Join, Filter, Project, Aggregate, Distinct, Sort, Limit, GroupRef };
+enum class MemoExpressionKind { Scan, Join, Filter, Project, Aggregate, Window, Distinct, Sort, Limit, GroupRef };
 
 struct MemoExpression {
     MemoExpressionKind kind{MemoExpressionKind::Scan};
@@ -26,6 +26,7 @@ struct MemoExpression {
     std::vector<plan::Projection> projections;
     std::vector<plan::BoundColumnRef> group_keys;
     std::vector<plan::AggregateExpression> aggregate_expressions;
+    std::vector<plan::WindowExpression> window_expressions;
     std::vector<plan::SortKey> sort_keys;
     std::vector<plan::BoundPredicate> predicates;
     plan::JoinKind join_kind{plan::JoinKind::Inner};

include/optimizer/memo.hpp

@@ -16,7 +16,7 @@ namespace plan {
 
 struct LogicalPlan;
 
-enum class LogicalKind { Scan, Join, Filter, Project, Aggregate, Distinct, Sort, Limit, Explain };
+enum class LogicalKind { Scan, Join, Filter, Project, Aggregate, Window, Distinct, Sort, Limit, Explain };
 enum class OrderPermission { Deterministic, Arbitrary };
 enum class JoinKind { Inner, Left, Semi, Anti };
 
@@ -149,6 +149,17 @@ struct AggregateExpression {
     catalog::ColumnType type{catalog::ColumnType::Int64};
 };
 
+struct WindowExpression {
+    std::string output_name;
+    sql::WindowFunction function{sql::WindowFunction::RowNumber};
+    bool count_star{false};
+    std::optional<BoundColumnRef> argument;
+    std::vector<BoundColumnRef> partition_keys;
+    std::vector<SortKey> order_keys;
+    std::size_t position{0};
+    catalog::ColumnType type{catalog::ColumnType::Int64};
+};
+
 struct LogicalPlan {
     LogicalKind kind{LogicalKind::Scan};
     OrderPermission order_permission{OrderPermission::Deterministic};
@@ -159,6 +170,7 @@ struct LogicalPlan {
     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;
     // Set only on a bound subplan root. Empty is the structural Phase 21a
@@ -228,6 +240,14 @@ struct LogicalPlan {
         return p;
     }
 
+    static LogicalPlan window(std::vector<WindowExpression> window_expressions, LogicalPlan child) {
+        LogicalPlan p;
+        p.kind = LogicalKind::Window;
+        p.window_expressions = std::move(window_expressions);
+        p.input = std::make_shared<LogicalPlan>(std::move(child));
+        return p;
+    }
+
     static LogicalPlan distinct(LogicalPlan child) {
         LogicalPlan p;
         p.kind = LogicalKind::Distinct;

include/plan/logical_plan.hpp

@@ -49,9 +49,31 @@ struct AggregateCall {
     bool nested_aggregate{false};
     AggregateFunction nested_function{AggregateFunction::Count};
     std::size_t nested_position{0};
+    std::shared_ptr<AggregateCall> nested_call;
 };
 
-using SelectExpr = std::variant<ScalarExpr, AggregateCall>;
+enum class SortDirection { Asc, Desc };
+
+enum class WindowFunction { RowNumber, Rank, DenseRank, Count, Sum, Min, Max };
+
+using WindowInputExpr = std::variant<ColumnRef, AggregateCall>;
+
+struct WindowOrderKey {
+    WindowInputExpr expression;
+    SortDirection direction{SortDirection::Asc};
+};
+
+struct WindowCall {
+    WindowFunction function{WindowFunction::RowNumber};
+    std::size_t position{0};
+    std::size_t over_position{0};
+    bool count_star{false};
+    std::optional<WindowInputExpr> argument;
+    std::vector<WindowInputExpr> partition_by;
+    std::vector<WindowOrderKey> order_by;
+};
+
+using SelectExpr = std::variant<ScalarExpr, AggregateCall, WindowCall>;
 using HavingExpr = std::variant<ColumnRef, IntLiteral, StringLiteral, NullLiteral, ScalarSubquery, AggregateCall>;
 using OrderByExpr = std::variant<ColumnRef, AggregateCall>;
 
@@ -201,8 +223,6 @@ struct HavingClause {
     std::vector<HavingPredicateExpr> conjuncts;
 };
 
-enum class SortDirection { Asc, Desc };
-
 struct OrderByKey {
     OrderByExpr expression;
     SortDirection direction{SortDirection::Asc};
@@ -264,7 +284,10 @@ inline std::size_t expression_position(const SelectExpr& expression) {
     if (const auto* scalar = std::get_if<ScalarExpr>(&expression)) {
         return expression_position(*scalar);
     }
-    return std::get<AggregateCall>(expression).position;
+    if (const auto* aggregate = std::get_if<AggregateCall>(&expression)) {
+        return aggregate->position;
+    }
+    return std::get<WindowCall>(expression).position;
 }
 
 inline std::size_t expression_position(const HavingExpr& expression) {
@@ -307,6 +330,26 @@ inline std::string aggregate_function_name(AggregateFunction function) {
     return "<unknown aggregate>";
 }
 
+inline std::string window_function_name(WindowFunction function) {
+    switch (function) {
+    case WindowFunction::RowNumber:
+        return "ROW_NUMBER";
+    case WindowFunction::Rank:
+        return "RANK";
+    case WindowFunction::DenseRank:
+        return "DENSE_RANK";
+    case WindowFunction::Count:
+        return "COUNT";
+    case WindowFunction::Sum:
+        return "SUM";
+    case WindowFunction::Min:
+        return "MIN";
+    case WindowFunction::Max:
+        return "MAX";
+    }
+    return "<unknown window function>";
+}
+
 inline std::string quote_string_literal(const std::string& value) {
     std::string quoted = "'";
     for (const auto ch : value) {
@@ -351,18 +394,64 @@ inline std::string output_name(const AggregateCall& aggregate) {
         name += "*";
     } else if (aggregate.argument.has_value()) {
         name += output_name(*aggregate.argument);
+    } else if (aggregate.nested_call != nullptr) {
+        name += output_name(*aggregate.nested_call);
     } else if (aggregate.nested_aggregate) {
         name += aggregate_function_name(aggregate.nested_function) + "(...)";
     }
     name += ")";
     return name;
 }
 
+inline std::string window_input_output_name(const WindowInputExpr& expression) {
+    if (const auto* column = std::get_if<ColumnRef>(&expression)) {
+        return output_name(*column);
+    }
+    return output_name(std::get<AggregateCall>(expression));
+}
+
+inline std::string output_name(const WindowCall& window) {
+    auto name = window_function_name(window.function) + "(";
+    if (window.count_star) {
+        name += "*";
+    } else if (window.argument.has_value()) {
+        name += window_input_output_name(*window.argument);
+    }
+    name += ") OVER (";
+    if (!window.partition_by.empty()) {
+        name += "PARTITION BY ";
+        for (std::size_t i = 0; i < window.partition_by.size(); ++i) {
+            if (i != 0) {
+                name += ", ";
+            }
+            name += window_input_output_name(window.partition_by[i]);
+        }
+    }
+    if (!window.order_by.empty()) {
+        if (!window.partition_by.empty()) {
+            name += " ";
+        }
+        name += "ORDER BY ";
+        for (std::size_t i = 0; i < window.order_by.size(); ++i) {
+            if (i != 0) {
+                name += ", ";
+            }
+            name += window_input_output_name(window.order_by[i].expression);
+            name += window.order_by[i].direction == SortDirection::Asc ? " ASC" : " DESC";
+        }
+    }
+    name += ")";
+    return name;
+}
+
 inline std::string output_name(const SelectExpr& expression) {
     if (const auto* scalar = std::get_if<ScalarExpr>(&expression)) {
         return output_name(*scalar);
     }
-    return output_name(std::get<AggregateCall>(expression));
+    if (const auto* aggregate = std::get_if<AggregateCall>(&expression)) {
+        return output_name(*aggregate);
+    }
+    return output_name(std::get<WindowCall>(expression));
 }
 
 inline std::string output_name(const HavingExpr& expression) {