feat: add window function support

AGENT.md

@@ -81,6 +81,8 @@ PRs that modify logic but leave obvious test gaps untouched may be rejected.
 
 ### 3.1 Prefer Simple Code
 
+- Implement functionality with the smallest reasonable change and the simplest correct design.
+- Do not over-engineer or introduce abstractions for hypothetical future requirements.
 - Prefer straightforward control flow
 - Avoid unnecessary abstractions
 - Avoid premature generalization
@@ -149,6 +151,8 @@ KiteSQL aims to stay easy to build, easy to audit, and easy to understand.
 
 A valid PR should:
 
+- Follow the repository pull request template
+- Describe test coverage and verified behavior instead of listing commands that were run
 - Compile cleanly
 - Pass all tests via make
 - Include new tests if behavior changes

docs/features.md

@@ -183,6 +183,11 @@ If `unsafe_txdb_checkpoint` is not enabled, `build_rocksdb()` returns an explici
 - [x] Exists
 - [x] Group By
 - [x] Having
+- [x] Window functions:
+  - `row_number()`, `rank()`, `dense_rank()`
+  - `count()`, `sum()`, `avg()`, `min()`, `max()` with `OVER`
+  - `PARTITION BY` and window `ORDER BY` with the default frame
+  - Explicit frames, named windows, and `QUALIFY` are not yet supported
 - [x] Order By
 - [x] Limit
 - [x] Show Tables

src/binder/aggregate.rs

@@ -16,7 +16,6 @@ use std::collections::HashSet;
 
 use super::{Binder, QueryBindStep};
 use crate::errors::DatabaseError;
-use crate::expression::function::scala::ScalarFunction;
 use crate::expression::visitor::{walk_expr, ExprVisitor};
 use crate::expression::visitor_mut::{walk_mut_expr, ExprVisitorMut};
 use crate::planner::LogicalPlan;
@@ -27,6 +26,22 @@ use crate::{
     planner::operator::{aggregate::AggregateOperator, sort::SortField},
 };
 
+struct AggregateCallCollector<'a> {
+    agg_calls: &'a mut Vec<ScalarExpression>,
+}
+
+impl<'expr> ExprVisitor<'expr> for AggregateCallCollector<'_> {
+    fn visit(&mut self, expr: &'expr ScalarExpression) -> Result<(), DatabaseError> {
+        match expr {
+            ScalarExpression::AggCall { .. } => self.agg_calls.push(expr.clone()),
+            ScalarExpression::Alias { expr, .. } => self.visit(expr)?,
+            ScalarExpression::Empty | ScalarExpression::TableFunction(_) => unreachable!(),
+            _ => walk_expr(self, expr)?,
+        }
+        Ok(())
+    }
+}
+
 impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A> {
     pub fn bind_aggregate(
         &mut self,
@@ -48,7 +63,7 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
         select_items: &mut [ScalarExpression],
     ) -> Result<(), DatabaseError> {
         for column in select_items {
-            self.visit_column_agg_expr(column)?;
+            self.collect_aggregate_calls(column)?;
         }
         Ok(())
     }
@@ -77,14 +92,14 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
         F: FnMut(&mut Self, I::Item) -> Result<SortField, DatabaseError>,
     {
         if let Some(having) = having.as_mut() {
-            self.visit_column_agg_expr(having)?;
+            self.collect_aggregate_calls(having)?;
         }
         let mut return_orderby = None;
         if let Some(orderby) = orderby {
             let mut fields = Vec::new();
             for orderby in orderby {
-                let mut field = bind_sort_field(self, orderby)?;
-                self.visit_column_agg_expr(&mut field.expr)?;
+                let field = bind_sort_field(self, orderby)?;
+                self.collect_aggregate_calls(&field.expr)?;
                 fields.push(field);
             }
             return_orderby = Some(fields);
@@ -119,119 +134,11 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
         Ok(())
     }
 
-    fn visit_column_agg_expr(&mut self, expr: &mut ScalarExpression) -> Result<(), DatabaseError> {
-        match expr {
-            ScalarExpression::AggCall { .. } => {
-                self.context.agg_calls.push(expr.clone());
-            }
-            ScalarExpression::TypeCast { expr, .. } => self.visit_column_agg_expr(expr)?,
-            ScalarExpression::IsNull { expr, .. } => self.visit_column_agg_expr(expr)?,
-            ScalarExpression::Unary { expr, .. } => self.visit_column_agg_expr(expr)?,
-            ScalarExpression::Alias { expr, .. } => self.visit_column_agg_expr(expr)?,
-            ScalarExpression::Binary {
-                left_expr,
-                right_expr,
-                ..
-            } => {
-                self.visit_column_agg_expr(left_expr)?;
-                self.visit_column_agg_expr(right_expr)?;
-            }
-            ScalarExpression::In { expr, args, .. } => {
-                self.visit_column_agg_expr(expr)?;
-                for arg in args {
-                    self.visit_column_agg_expr(arg)?;
-                }
-            }
-            ScalarExpression::Between {
-                expr,
-                left_expr,
-                right_expr,
-                ..
-            } => {
-                self.visit_column_agg_expr(expr)?;
-                self.visit_column_agg_expr(left_expr)?;
-                self.visit_column_agg_expr(right_expr)?;
-            }
-            ScalarExpression::SubString {
-                expr,
-                for_expr,
-                from_expr,
-            } => {
-                self.visit_column_agg_expr(expr)?;
-                if let Some(expr) = for_expr {
-                    self.visit_column_agg_expr(expr)?;
-                }
-                if let Some(expr) = from_expr {
-                    self.visit_column_agg_expr(expr)?;
-                }
-            }
-            ScalarExpression::Position { expr, in_expr } => {
-                self.visit_column_agg_expr(expr)?;
-                self.visit_column_agg_expr(in_expr)?;
-            }
-            ScalarExpression::Trim {
-                expr,
-                trim_what_expr,
-                ..
-            } => {
-                self.visit_column_agg_expr(expr)?;
-                if let Some(trim_what_expr) = trim_what_expr {
-                    self.visit_column_agg_expr(trim_what_expr)?;
-                }
-            }
-            ScalarExpression::Constant(_) | ScalarExpression::ColumnRef { .. } => (),
-            ScalarExpression::Empty => unreachable!(),
-            ScalarExpression::Tuple(args)
-            | ScalarExpression::ScalaFunction(ScalarFunction { args, .. })
-            | ScalarExpression::Coalesce { exprs: args, .. } => {
-                for expr in args {
-                    self.visit_column_agg_expr(expr)?;
-                }
-            }
-            ScalarExpression::If {
-                condition,
-                left_expr,
-                right_expr,
-                ..
-            } => {
-                self.visit_column_agg_expr(condition)?;
-                self.visit_column_agg_expr(left_expr)?;
-                self.visit_column_agg_expr(right_expr)?;
-            }
-            ScalarExpression::IfNull {
-                left_expr,
-                right_expr,
-                ..
-            }
-            | ScalarExpression::NullIf {
-                left_expr,
-                right_expr,
-                ..
-            } => {
-                self.visit_column_agg_expr(left_expr)?;
-                self.visit_column_agg_expr(right_expr)?;
-            }
-            ScalarExpression::CaseWhen {
-                operand_expr,
-                expr_pairs,
-                else_expr,
-                ..
-            } => {
-                if let Some(expr) = operand_expr {
-                    self.visit_column_agg_expr(expr)?;
-                }
-                for (expr_1, expr_2) in expr_pairs {
-                    self.visit_column_agg_expr(expr_1)?;
-                    self.visit_column_agg_expr(expr_2)?;
-                }
-                if let Some(expr) = else_expr {
-                    self.visit_column_agg_expr(expr)?;
-                }
-            }
-            ScalarExpression::TableFunction(_) => unreachable!(),
+    fn collect_aggregate_calls(&mut self, expr: &ScalarExpression) -> Result<(), DatabaseError> {
+        AggregateCallCollector {
+            agg_calls: &mut self.context.agg_calls,
         }
-
-        Ok(())
+        .visit(expr)
     }
 
     /// Validate select exprs must appear in the GROUP BY clause or be used in
@@ -269,6 +176,10 @@ impl<T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'_, '_, T, A>
             HashSet::from_iter(group_raw_exprs.iter().copied());
 
         for expr in select_items {
+            if expr.has_window_call()? {
+                HavingOrderByValidator::new(groupby, &self.context.agg_calls).visit(expr)?;
+                continue;
+            }
             if expr.has_agg_call()? {
                 continue;
             }

src/binder/expr.rs

@@ -476,76 +476,60 @@ impl<'a, T: Transaction, A: AsRef<[(&'static str, DataValue)]>> Binder<'a, '_, T
         })
     }
 
-    pub(crate) fn bind_function_call(
+    pub(crate) fn bind_aggregate_function(
         &mut self,
-        function_name: String,
-        mut args: Vec<ScalarExpression>,
+        kind: AggKind,
+        args: Vec<ScalarExpression>,
         is_distinct: bool,
         arena: &mut PlanArena,
     ) -> Result<ScalarExpression, DatabaseError> {
-        match function_name.as_str() {
-            "count" => {
+        let ty = match kind {
+            AggKind::Count => {
                 if args.len() != 1 {
                     return Err(DatabaseError::MisMatch("number of count() parameters", "1"));
                 }
-                return Ok(ScalarExpression::AggCall {
-                    distinct: is_distinct,
-                    kind: AggKind::Count,
-                    args,
-                    ty: LogicalType::Integer,
-                });
+                LogicalType::Integer
             }
-            "sum" => {
+            AggKind::Sum => {
                 if args.len() != 1 {
                     return Err(DatabaseError::MisMatch("number of sum() parameters", "1"));
                 }
-                let ty = args[0].return_type(arena).into_owned();
-
-                return Ok(ScalarExpression::AggCall {
-                    distinct: is_distinct,
-                    kind: AggKind::Sum,
-                    args,
-                    ty,
-                });
+                args[0].return_type(arena).into_owned()
             }
-            "min" => {
+            AggKind::Min => {
                 if args.len() != 1 {
                     return Err(DatabaseError::MisMatch("number of min() parameters", "1"));
                 }
-                let ty = args[0].return_type(arena).into_owned();
-
-                return Ok(ScalarExpression::AggCall {
-                    distinct: is_distinct,
-                    kind: AggKind::Min,
-                    args,
-                    ty,
-                });
+                args[0].return_type(arena).into_owned()
             }
-            "max" => {
+            AggKind::Max => {
                 if args.len() != 1 {
                     return Err(DatabaseError::MisMatch("number of max() parameters", "1"));
                 }
-                let ty = args[0].return_type(arena).into_owned();
-
-                return Ok(ScalarExpression::AggCall {
-                    distinct: is_distinct,
-                    kind: AggKind::Max,
-                    args,
-                    ty,
-                });
+                args[0].return_type(arena).into_owned()
             }
-            "avg" => {
+            AggKind::Avg => {
                 if args.len() != 1 {
                     return Err(DatabaseError::MisMatch("number of avg() parameters", "1"));
                 }
-
-                return Ok(ScalarExpression::AggCall {
-                    distinct: is_distinct,
-                    kind: AggKind::Avg,
-                    args,
-                    ty: LogicalType::Double,
-                });
+                LogicalType::Double
             }
+        };
+        Ok(ScalarExpression::AggCall {
+            distinct: is_distinct,
+            kind,
+            args,
+            ty,
+        })
+    }
+
+    pub(crate) fn bind_function_call(
+        &mut self,
+        function_name: String,
+        mut args: Vec<ScalarExpression>,
+        arena: &mut PlanArena,
+    ) -> Result<ScalarExpression, DatabaseError> {
+        match function_name.as_str() {
             "if" => {
                 if args.len() != 3 {
                     return Err(DatabaseError::MisMatch("number of if() parameters", "3"));

src/binder/mod.rs

@@ -48,6 +48,7 @@ mod show_table;
 mod show_view;
 mod truncate;
 mod update;
+mod window;
 
 #[cfg(feature = "parser")]
 pub use parser::{command_type, prepare, prepare_all, CommandType, Statement};
@@ -90,6 +91,7 @@ pub enum QueryBindStep {
     Where,
     Agg,
     Having,
+    Window,
     Distinct,
     Sort,
     Project,