core/translate: split window step emission by per-source-row vs per-buffered-row pattern

Window functions fall into two categories with respect to when their
AggStep fires: aggregate-like functions (sum/count/avg as windows, and
the upcoming rank-family functions) count source rows as they arrive
from the input subquery and capture a single value per peer-group
flush; row_number and the upcoming positional functions (lag, lead,
first/last/nth_value) step per buffered row at flush time because
their value changes within a peer group.

Express this as a WindowFunc::steps_per_source_row() classifier and
split the two emit sites accordingly:

  - emit_aggregation_step (called per source row) handles aggregate
    window funcs and the rank-family;
  - emit_return_buffered_rows pre-loop captures AggValue once per
    flush for both categories above, and the per-row loop body steps
    + reads only the per-buffered-row category.

No behavioral change: the rank-family branch is currently dead because
none of those functions are wired in name resolution yet. row_number
continues to emit via the per-buffered-row path. Mirrors how SQLite
gates its xStep emission per function family rather than running a
runtime peer-detection flag.

Author: jussisaurio

core/function.rs

@@ -569,6 +569,66 @@ impl WindowFunc {
             ),
         }
     }
+
+    /// Whether the function produces a single value shared by every row in a
+    /// peer group (true) or a distinct value per row (false).
+    ///
+    /// This controls where `emit_aggregation_step` and
+    /// `emit_peer_group_flush` emit the function's `AggStep` and
+    /// `AggValue` opcodes:
+    ///
+    /// * true (rank-family): `AggStep` is emitted inside the input-scan loop
+    ///   in `emit_aggregation_step`, so it runs once for every row read from
+    ///   the subquery and advances the function's internal counter. `AggValue`
+    ///   is emitted once per peer-group flush in `emit_peer_group_flush`,
+    ///   before the loop over buffered rows, reading the accumulator into a
+    ///   result register. The loop then writes that one register's contents
+    ///   into the output for every buffered row in the group.
+    /// * false (row_number, ntile, lag, lead, first/last/nth_value): nothing
+    ///   is emitted in `emit_aggregation_step`. Both `AggStep` and `AggValue`
+    ///   are emitted inside the per-row loop in `emit_peer_group_flush`,
+    ///   so they run once per buffered row and produce a distinct value for
+    ///   each output row.
+    ///
+    /// Concrete example. Given `scores(name, score)`:
+    ///
+    ///   alice  100
+    ///   bob    100
+    ///   carol   90
+    ///
+    /// `rank() OVER (ORDER BY score DESC)` returns true — alice and bob are
+    /// peers on score=100 and share the same rank:
+    ///
+    ///   alice  100  1
+    ///   bob    100  1
+    ///   carol   90  3
+    ///
+    /// `row_number() OVER (ORDER BY score DESC)` returns false — every row
+    /// gets a distinct number even within the score=100 peer group:
+    ///
+    ///   alice  100  1
+    ///   bob    100  2
+    ///   carol   90  3
+    ///
+    /// Functions returning true: `rank`, `dense_rank`, `percent_rank`,
+    /// `cume_dist` — their value is determined entirely by where the peer
+    /// group sits in the partition's ordering.
+    /// Functions returning false: `row_number`, `ntile`, `lag`, `lead`,
+    /// `first_value`, `last_value`, `nth_value` — all depend on a row's
+    /// position within the partition/frame, not just its peer group.
+    pub fn one_value_per_peer_group(&self) -> bool {
+        match self {
+            Self::Rank | Self::DenseRank | Self::PercentRank | Self::CumeDist => true,
+            Self::RowNumber
+            | Self::Ntile
+            | Self::Lag
+            | Self::Lead
+            | Self::FirstValue
+            | Self::LastValue
+            | Self::NthValue
+            | Self::External(_) => false,
+        }
+    }
 }
 
 impl PartialEq for WindowFunc {

core/translate/window.rs

@@ -1040,57 +1040,86 @@ fn emit_aggregation_step(
     registers: &WindowRegisters,
 ) -> crate::Result<()> {
     for (i, func) in window.functions.iter().enumerate() {
-        let AccumulatorFunc::Agg(agg_func) = &func.func else {
-            continue;
-        };
-        // The aggregation step is performed incrementally as each row from the subquery is
-        // processed. Therefore, we don’t need to access the buffer table and can obtain argument
-        // values directly by evaluating the expressions that reference the subquery result columns.
-        // Use the rewritten form when available so the args reference the subquery
-        // rather than the (no-longer-visible) original tables.
-        let args = match func.current_expr() {
-            Expr::FunctionCall { args, .. } => args.iter().map(|a| (**a).clone()).collect(),
-            Expr::FunctionCallStar { .. } => vec![],
-            _ => unreachable!(
-                "All window functions should be either FunctionCall or FunctionCallStar expressions"
-            ),
-        };
-
         let reg_acc_start = registers.acc_start + i;
-        // FILTER controls whether the current input row contributes to the
-        // running aggregate; it does not suppress the output row itself.
-        let filter_skip_label = if let Some(filter_expr) =
-            func.rewritten.as_ref().and_then(|r| r.filter_expr.as_ref())
-        {
-            let label = program.allocate_label();
-            let filter_reg = program.alloc_register();
-            translate_expr(
-                program,
-                Some(&plan.table_references),
-                filter_expr,
-                filter_reg,
-                resolver,
-            )?;
-            program.emit_insn(Insn::IfNot {
-                reg: filter_reg,
-                target_pc: label,
-                jump_if_null: true,
-            });
-            Some(label)
-        } else {
-            None
-        };
+        match &func.func {
+            AccumulatorFunc::Agg(agg_func) => {
+                // The aggregation step is performed incrementally as each row from the subquery is
+                // processed. Therefore, we don’t need to access the buffer table and can obtain
+                // argument values directly by evaluating the expressions that reference the
+                // subquery result columns. Use the rewritten form when available so the args
+                // reference the subquery rather than the (no-longer-visible) original tables.
+                let args = match func.current_expr() {
+                    Expr::FunctionCall { args, .. } => {
+                        args.iter().map(|a| (**a).clone()).collect()
+                    }
+                    Expr::FunctionCallStar { .. } => vec![],
+                    _ => unreachable!(
+                        "All window functions should be either FunctionCall or FunctionCallStar expressions"
+                    ),
+                };
 
-        translate_aggregation_step(
-            program,
-            &plan.table_references,
-            AggArgumentSource::new_from_expression(agg_func, &args, &Distinctness::NonDistinct),
-            reg_acc_start,
-            resolver,
-            None,
-        )?;
-        if let Some(label) = filter_skip_label {
-            program.preassign_label_to_next_insn(label);
+                // FILTER controls whether the current input row contributes to the
+                // running aggregate; it does not suppress the output row itself.
+                let filter_skip_label = if let Some(filter_expr) =
+                    func.rewritten.as_ref().and_then(|r| r.filter_expr.as_ref())
+                {
+                    let label = program.allocate_label();
+                    let filter_reg = program.alloc_register();
+                    translate_expr(
+                        program,
+                        Some(&plan.table_references),
+                        filter_expr,
+                        filter_reg,
+                        resolver,
+                    )?;
+                    program.emit_insn(Insn::IfNot {
+                        reg: filter_reg,
+                        target_pc: label,
+                        jump_if_null: true,
+                    });
+                    Some(label)
+                } else {
+                    None
+                };
+
+                translate_aggregation_step(
+                    program,
+                    &plan.table_references,
+                    AggArgumentSource::new_from_expression(
+                        agg_func,
+                        &args,
+                        &Distinctness::NonDistinct,
+                    ),
+                    reg_acc_start,
+                    resolver,
+                    None,
+                )?;
+                if let Some(label) = filter_skip_label {
+                    program.preassign_label_to_next_insn(label);
+                }
+            }
+            AccumulatorFunc::Window(win_func) if win_func.one_value_per_peer_group() => {
+                // Rank-family functions take no arguments. Emit `AggStep`
+                // here, inside the input-scan loop, so the function's
+                // internal counter and peer-detection state advance once for
+                // every row read from the subquery. `AggValue` is NOT emitted
+                // here — the accumulator is read later, once per peer-group
+                // flush, in `emit_peer_group_flush`.
+                program.emit_insn(Insn::AggStep {
+                    acc_reg: reg_acc_start,
+                    col: 0,
+                    delimiter: 0,
+                    func: AccumulatorFunc::Window(win_func.clone()),
+                    comparator: None,
+                });
+            }
+            AccumulatorFunc::Window(_) => {
+                // row_number, ntile, lag, lead, first/last/nth_value: emit
+                // nothing here. Their value differs between rows of the same
+                // peer group, so both `AggStep` and `AggValue` are emitted
+                // later inside the loop over buffered rows in
+                // `emit_peer_group_flush`, once per output row.
+            }
         }
     }
 
@@ -1142,7 +1171,7 @@ pub fn emit_window_results(
         pc_if_empty: label_empty,
     });
 
-    emit_return_buffered_rows(program, window, t_ctx, plan)?;
+    emit_peer_group_flush(program, window, t_ctx, plan)?;
 
     program.preassign_label_to_next_insn(label_empty);
 
@@ -1159,7 +1188,7 @@ pub fn emit_window_results(
     Ok(())
 }
 
-fn emit_return_buffered_rows(
+fn emit_peer_group_flush(
     program: &mut ProgramBuilder,
     window: &Window,
     t_ctx: &mut TranslateCtx,
@@ -1173,15 +1202,23 @@ fn emit_return_buffered_rows(
         ..
     } = t_ctx.meta_window.as_ref().expect("missing window metadata");
 
-    // Aggregate window functions: capture the running accumulator value once
-    // per flush. Every buffered row in the just-finished peer group sees the
-    // same value (RANGE UNBOUNDED PRECEDING TO CURRENT ROW semantics).
+    // For aggregate window functions and the rank-family, the accumulator
+    // has already been advanced once per source row during the input scan.
+    // Emit one `AggValue` per such function here, before entering the loop
+    // over buffered rows: this reads the accumulator into the result
+    // register exactly once per peer-group flush. The per-row loop below
+    // will copy that register's contents into the output for every buffered
+    // row in the group (RANGE UNBOUNDED PRECEDING TO CURRENT ROW semantics).
     for (i, func) in window.functions.iter().enumerate() {
-        if let AccumulatorFunc::Agg(agg_func) = &func.func {
+        let value_pre_loop = match &func.func {
+            AccumulatorFunc::Agg(_) => true,
+            AccumulatorFunc::Window(w) => w.one_value_per_peer_group(),
+        };
+        if value_pre_loop {
             program.emit_insn(Insn::AggValue {
                 acc_reg: registers.acc_start + i,
                 dest_reg: registers.acc_result_start + i,
-                func: AccumulatorFunc::Agg(agg_func.clone()),
+                func: func.func.clone(),
             });
         }
     }
@@ -1197,13 +1234,20 @@ fn emit_return_buffered_rows(
         let reg_result = registers.result_columns_start + i;
         program.emit_column_or_rowid(cursors.csr_current, *col_idx, reg_result);
     }
-    // Pure window functions (e.g. row_number) advance their accumulator and
-    // read it out once per buffered row as the buffer is replayed. Aggregate
-    // window functions instead advance their accumulator once per source row
-    // (in emit_aggregation_step) and only their final value is read out per
-    // emitted row (the AggValue loop earlier in this function).
+    // For row_number, ntile, lag, lead and first/last/nth_value no
+    // `AggStep` ran during the input scan. Emit `AggStep` + `AggValue` here,
+    // inside the per-row loop over buffered rows: `AggStep` advances the
+    // function's state for the current row, then `AggValue` reads the
+    // resulting value into the result register that gets written to this
+    // row's output. This is the site that gives every buffered row in a
+    // peer group a distinct value. Functions where
+    // `one_value_per_peer_group()` is true were handled above the loop and
+    // are skipped here.
     for (i, func) in window.functions.iter().enumerate() {
         if let AccumulatorFunc::Window(win_func) = &func.func {
+            if win_func.one_value_per_peer_group() {
+                continue;
+            }
             let acc_reg = registers.acc_start + i;
             let dest_reg = registers.acc_result_start + i;
             program.emit_insn(Insn::AggStep {