core: implement rank() window function

rank() assigns the same value to all rows in a peer group (rows with
equal ORDER BY values) and skips ranks by group size, so [a,a,b,c,c]
ordered ascending becomes [1,1,3,4,4].

Wire 'rank' name resolution to WindowFunc::Rank and add step/value
arms in op_window_step / op_window_value. The state machine mirrors
SQLite's rankStepFunc / rankValueFunc: payload[0] is the current rank
value (cleared by AggValue, latched by the next step), payload[1] is
the rows-seen counter (always increments). The 'latch on zero' trick
lets every peer in a group read the same rank without an explicit
peer-detection signal — AggValue's clear is what tells the next step
'you start a new peer group'.

Verification:
- New tests/window/rank.sqltest covers peers/no-peers, partition,
  no-ORDER-BY, NULL handling, empty input, multi-column ORDER BY,
  DESC, COLLATE NOCASE peer detection, partition reset with
  repeated values, alongside row_number, and two rank windows
  with different orderings.
- Updates two cases in tests/window/memory.sqltest that previously
  expected 'no such function: rank' — one becomes a passing test,
  the other now waits on dense_rank().

Author: jussisaurio

core/function.rs

@@ -518,7 +518,7 @@ impl WindowFunc {
     /// drifts ahead of the resolver and users get "no such function" when
     /// they try to call them.
     pub fn is_implemented(&self) -> bool {
-        matches!(self, Self::RowNumber)
+        matches!(self, Self::RowNumber | Self::Rank)
     }
 
     /// The hardcoded frame this built-in evaluates over, overriding any
@@ -1637,6 +1637,12 @@ impl Func {
                 }
                 Ok(Some(Self::Window(WindowFunc::RowNumber)))
             }
+            "rank" => {
+                if arg_count != 0 {
+                    crate::bail_parse_error!("wrong number of arguments to function {}()", name)
+                }
+                Ok(Some(Self::Window(WindowFunc::Rank)))
+            }
             "timediff" => {
                 if arg_count != 2 {
                     crate::bail_parse_error!("wrong number of arguments to function {}()", name)

core/vdbe/execute.rs

@@ -6418,6 +6418,43 @@ fn op_window_step(
             };
             *counter += 1;
         }
+        // rank() — mirrors SQLite's CallCount-based rankStepFunc.
+        //
+        // State (payload):
+        //   [0] = current rank value (what AggValue returns; cleared to 0 by
+        //         AggValue so the next step can detect that AggValue just ran).
+        //   [1] = rows-seen counter for the partition (always increments).
+        //
+        // The "latch on zero" trick: AggValue clears payload[0] every time it
+        // runs, and a flush only happens at peer-group / partition boundaries.
+        // So when this step observes payload[0] == 0, it knows AggValue just
+        // fired on the prior peer group — meaning this row begins a fresh peer
+        // group, and the current row position (rows_seen) is the new rank.
+        // Subsequent rows in the same peer group see payload[0] != 0 and leave
+        // the rank value untouched, so every peer reads the same rank.
+        WindowFunc::Rank => {
+            if let Register::Value(Value::Null) = state.registers[acc_reg] {
+                state.registers[acc_reg] = Register::Aggregate(AggContext::Builtin(vec![
+                    Value::from_i64(0),
+                    Value::from_i64(0),
+                ]));
+            }
+            let Register::Aggregate(AggContext::Builtin(payload)) = &mut state.registers[acc_reg]
+            else {
+                unreachable!("rank accumulator must be a Builtin payload");
+            };
+            let Value::Numeric(Numeric::Integer(rows_seen)) = &mut payload[1] else {
+                unreachable!("rank rows_seen counter must be Integer");
+            };
+            *rows_seen += 1;
+            let rows_seen = *rows_seen;
+            let Value::Numeric(Numeric::Integer(rank)) = &mut payload[0] else {
+                unreachable!("rank current value must be Integer");
+            };
+            if *rank == 0 {
+                *rank = rows_seen;
+            }
+        }
         other => {
             return Err(LimboError::InternalError(format!(
                 "window function {other} reached runtime dispatch but has no handler"
@@ -6444,6 +6481,18 @@ fn op_window_value(
                 )))
             }
         },
+        WindowFunc::Rank => {
+            // Read current rank, then clear it so the next peer group's first
+            // AggStep latches a fresh value (matches SQLite's rankValueFunc).
+            let Register::Aggregate(AggContext::Builtin(payload)) = &mut state.registers[acc_reg]
+            else {
+                return Err(LimboError::InternalError(format!(
+                    "rank accumulator in unexpected register state: {:?}",
+                    state.registers[acc_reg]
+                )));
+            };
+            std::mem::replace(&mut payload[0], Value::from_i64(0))
+        }
         other => {
             return Err(LimboError::InternalError(format!(
                 "window function {other} reached runtime dispatch but has no handler"

testing/sqltests/tests/pragma/default.sqltest

@@ -284,11 +284,18 @@ expect {
     row_number|w|0
 }
 
+test pragma-function-list-window-rank {
+    SELECT name, type, narg FROM pragma_function_list WHERE name = 'rank';
+}
+expect {
+    rank|w|0
+}
+
 @skip-if sqlite "sqlite implements these window functions; we don't yet, so they must not be advertised in pragma_function_list"
 test pragma-function-list-omits-unimplemented-window-functions {
     SELECT COUNT(*) FROM pragma_function_list
     WHERE name IN (
-        'rank','dense_rank','percent_rank','cume_dist','ntile',
+        'dense_rank','percent_rank','cume_dist','ntile',
         'lag','lead','first_value','last_value','nth_value'
     );
 }

testing/sqltests/tests/window/memory.sqltest

@@ -448,15 +448,15 @@ setup named_window_rank_data {
 }
 
 @setup named_window_rank_data
-@skip-if sqlite "TODO: rank() and dense_rank() are not yet implemented."
+@skip-if sqlite "TODO: dense_rank() is not yet implemented."
 test named-window-rank-dense-rank {
     SELECT a, rank() OVER w1, dense_rank() OVER w1
     FROM t_nw_rank
     WINDOW w1 AS (ORDER BY b)
     ORDER BY a;
 }
 expect error {
-    no such function: rank
+    no such function: dense_rank
 }
 
 setup named_window_ntile_data {
@@ -571,16 +571,17 @@ setup named_window_group_data {
 }
 
 @setup named_window_group_data
-@skip-if sqlite "TODO: rank() is not yet implemented."
 test named-window-with-group-by {
     SELECT a, sum(b) AS s, rank() OVER w1
     FROM t_nw_grp
     GROUP BY a
     WINDOW w1 AS (ORDER BY sum(b))
     ORDER BY a;
 }
-expect error {
-    no such function: rank
+expect {
+    x|3|1
+    y|7|3
+    z|5|2
 }
 
 @setup named_window_data

testing/sqltests/tests/window/rank.sqltest

@@ -0,0 +1,191 @@
+@database :memory:
+
+setup rank_base {
+    CREATE TABLE rank_items(id INTEGER PRIMARY KEY, g TEXT, v INTEGER);
+    INSERT INTO rank_items VALUES
+        (1, 'a', 10),
+        (2, 'a', 10),
+        (3, 'a', 20),
+        (4, 'b', 5),
+        (5, 'b', 5),
+        (6, 'b', 5),
+        (7, 'c', 100);
+}
+
+@setup rank_base
+test rank-basic-with-peers {
+    SELECT id, v, rank() OVER (ORDER BY v) FROM rank_items ORDER BY id;
+}
+expect {
+    1|10|4
+    2|10|4
+    3|20|6
+    4|5|1
+    5|5|1
+    6|5|1
+    7|100|7
+}
+
+@setup rank_base
+test rank-partition-by-group {
+    SELECT id, g, v, rank() OVER (PARTITION BY g ORDER BY v) FROM rank_items ORDER BY id;
+}
+expect {
+    1|a|10|1
+    2|a|10|1
+    3|a|20|3
+    4|b|5|1
+    5|b|5|1
+    6|b|5|1
+    7|c|100|1
+}
+
+@setup rank_base
+test rank-no-order-by {
+    SELECT id, rank() OVER (PARTITION BY g) FROM rank_items ORDER BY id;
+}
+expect {
+    1|1
+    2|1
+    3|1
+    4|1
+    5|1
+    6|1
+    7|1
+}
+
+setup rank_nulls {
+    CREATE TABLE rank_n(v INTEGER);
+    INSERT INTO rank_n VALUES (NULL), (NULL), (1), (2), (2);
+}
+
+@setup rank_nulls
+test rank-null-values-share-rank {
+    SELECT v, rank() OVER (ORDER BY v) FROM rank_n ORDER BY v;
+}
+expect {
+    |1
+    |1
+    1|3
+    2|4
+    2|4
+}
+
+setup rank_empty {
+    CREATE TABLE rank_e(v INTEGER);
+}
+
+@setup rank_empty
+test rank-empty-input {
+    SELECT v, rank() OVER (ORDER BY v) FROM rank_e;
+}
+expect {
+}
+
+@setup rank_base
+test rank-alongside-row-number {
+    SELECT id, v,
+           rank() OVER (ORDER BY v),
+           row_number() OVER (ORDER BY v)
+    FROM rank_items ORDER BY id;
+}
+expect {
+    1|10|4|4
+    2|10|4|5
+    3|20|6|6
+    4|5|1|1
+    5|5|1|2
+    6|5|1|3
+    7|100|7|7
+}
+
+setup rank_multikey {
+    CREATE TABLE rank_mk(id INTEGER PRIMARY KEY, a INTEGER, b INTEGER);
+    INSERT INTO rank_mk VALUES
+        (1, 1, 1),
+        (2, 1, 1),
+        (3, 1, 2),
+        (4, 2, 1);
+}
+
+# Peers are defined by the full ORDER BY tuple, not by any one column.
+@setup rank_multikey
+test rank-multi-column-order-by {
+    SELECT id, a, b, rank() OVER (ORDER BY a, b) FROM rank_mk ORDER BY id;
+}
+expect {
+    1|1|1|1
+    2|1|1|1
+    3|1|2|3
+    4|2|1|4
+}
+
+@setup rank_multikey
+test rank-order-by-desc {
+    SELECT id, a, rank() OVER (ORDER BY a DESC) FROM rank_mk ORDER BY id;
+}
+expect {
+    1|1|2
+    2|1|2
+    3|1|2
+    4|2|1
+}
+
+setup rank_collate {
+    CREATE TABLE rank_c(v TEXT COLLATE NOCASE);
+    INSERT INTO rank_c VALUES ('Apple'), ('apple'), ('Banana'), ('banana');
+}
+
+# Peer detection respects the column's collation: 'Apple' and 'apple'
+# compare equal under NOCASE, so they share rank and the next group
+# starts at 3 (rank gap reflects the two-row peer group).
+@setup rank_collate
+test rank-collate-nocase-peers {
+    SELECT v, rank() OVER (ORDER BY v) FROM rank_c ORDER BY v;
+}
+expect {
+    Apple|1
+    apple|1
+    Banana|3
+    banana|3
+}
+
+setup rank_partition_repeated {
+    CREATE TABLE rank_pr(g TEXT, v INTEGER);
+    INSERT INTO rank_pr VALUES
+        ('a', 1), ('a', 1), ('a', 2),
+        ('b', 1), ('b', 2), ('b', 2), ('b', 3);
+}
+
+# Values reappear across partitions; rank must reset to 1 at each
+# partition boundary regardless of what came before.
+@setup rank_partition_repeated
+test rank-resets-across-partitions-with-repeated-values {
+    SELECT g, v, rank() OVER (PARTITION BY g ORDER BY v) FROM rank_pr ORDER BY g, v;
+}
+expect {
+    a|1|1
+    a|1|1
+    a|2|3
+    b|1|1
+    b|2|2
+    b|2|2
+    b|3|4
+}
+
+@setup rank_base
+test two-rank-windows-different-orderings {
+    SELECT id, v,
+           rank() OVER (ORDER BY v),
+           rank() OVER (ORDER BY v DESC)
+    FROM rank_items ORDER BY id;
+}
+expect {
+    1|10|4|3
+    2|10|4|3
+    3|20|6|2
+    4|5|1|5
+    5|5|1|5
+    6|5|1|5
+    7|100|7|1
+}