fsqlite_func/
window_builtins.rs

1//! Built-in window functions (S13.5, bd-14i6).
2//!
3//! Implements: row_number, rank, dense_rank, percent_rank, cume_dist,
4//! ntile, lag, lead, first_value, last_value, nth_value, plus
5//! aggregate-as-window variants including group_concat and string_agg.
6//!
7//! These functions implement the [`WindowFunction`] trait.  The VDBE is
8//! responsible for partitioning, ordering, and frame management; these
9//! implementations provide the per-row computation logic.
10//!
11//! # Design Notes
12//!
13//! Pure-numbering functions (row_number, rank, dense_rank) track position
14//! via step() and expose the current value through value().  The ORDER BY
15//! column is passed as args\[0\] so the function can detect peer-group
16//! boundaries.
17//!
18//! Buffer-based functions (lag, lead, first_value, last_value, nth_value)
19//! maintain an internal VecDeque of values and expose frame-relative
20//! access through value().
21#![allow(
22    clippy::unnecessary_literal_bound,
23    clippy::cast_possible_truncation,
24    clippy::cast_possible_wrap,
25    clippy::cast_precision_loss,
26    clippy::cast_sign_loss,
27    clippy::items_after_statements,
28    clippy::float_cmp,
29    clippy::match_same_arms,
30    clippy::similar_names
31)]
32
33use std::collections::VecDeque;
34
35use fsqlite_error::{FrankenError, Result};
36use fsqlite_types::SqliteValue;
37
38use crate::{FunctionRegistry, WindowFunction};
39
40// ═══════════════════════════════════════════════════════════════════════════
41// row_number()
42// ═══════════════════════════════════════════════════════════════════════════
43
44pub struct RowNumberState {
45    counter: i64,
46}
47
48pub struct RowNumberFunc;
49
50impl WindowFunction for RowNumberFunc {
51    type State = RowNumberState;
52
53    fn initial_state(&self) -> Self::State {
54        RowNumberState { counter: 0 }
55    }
56
57    fn step(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
58        state.counter += 1;
59        Ok(())
60    }
61
62    fn inverse(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
63        state.counter -= 1;
64        Ok(())
65    }
66
67    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
68        Ok(SqliteValue::Integer(state.counter))
69    }
70
71    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
72        Ok(SqliteValue::Integer(state.counter))
73    }
74
75    fn num_args(&self) -> i32 {
76        0
77    }
78
79    fn name(&self) -> &str {
80        "row_number"
81    }
82}
83
84// ═══════════════════════════════════════════════════════════════════════════
85// rank()
86// ═══════════════════════════════════════════════════════════════════════════
87
88pub struct RankState {
89    row_number: i64,
90    rank: i64,
91    last_order_value: Option<SqliteValue>,
92}
93
94pub struct RankFunc;
95
96impl WindowFunction for RankFunc {
97    type State = RankState;
98
99    fn initial_state(&self) -> Self::State {
100        RankState {
101            row_number: 0,
102            rank: 0,
103            last_order_value: None,
104        }
105    }
106
107    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
108        state.row_number += 1;
109        let current = args.first().cloned().unwrap_or(SqliteValue::Null);
110        let is_new_peer = match &state.last_order_value {
111            None => true,
112            Some(last) => &current != last,
113        };
114        if is_new_peer {
115            state.rank = state.row_number;
116            state.last_order_value = Some(current);
117        }
118        Ok(())
119    }
120
121    fn inverse(&self, _state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
122        // rank() uses UNBOUNDED PRECEDING to CURRENT ROW; inverse is a no-op.
123        Ok(())
124    }
125
126    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
127        Ok(SqliteValue::Integer(state.rank))
128    }
129
130    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
131        Ok(SqliteValue::Integer(state.rank))
132    }
133
134    fn num_args(&self) -> i32 {
135        -1
136    }
137
138    fn min_args(&self) -> i32 {
139        0
140    }
141
142    fn max_args(&self) -> Option<i32> {
143        Some(0)
144    }
145
146    fn name(&self) -> &str {
147        "rank"
148    }
149}
150
151// ═══════════════════════════════════════════════════════════════════════════
152// dense_rank()
153// ═══════════════════════════════════════════════════════════════════════════
154
155pub struct DenseRankState {
156    dense_rank: i64,
157    last_order_value: Option<SqliteValue>,
158}
159
160pub struct DenseRankFunc;
161
162impl WindowFunction for DenseRankFunc {
163    type State = DenseRankState;
164
165    fn initial_state(&self) -> Self::State {
166        DenseRankState {
167            dense_rank: 0,
168            last_order_value: None,
169        }
170    }
171
172    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
173        let current = args.first().cloned().unwrap_or(SqliteValue::Null);
174        let is_new_peer = match &state.last_order_value {
175            None => true,
176            Some(last) => &current != last,
177        };
178        if is_new_peer {
179            state.dense_rank += 1;
180            state.last_order_value = Some(current);
181        }
182        Ok(())
183    }
184
185    fn inverse(&self, _state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
186        Ok(())
187    }
188
189    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
190        Ok(SqliteValue::Integer(state.dense_rank))
191    }
192
193    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
194        Ok(SqliteValue::Integer(state.dense_rank))
195    }
196
197    fn num_args(&self) -> i32 {
198        -1
199    }
200
201    fn min_args(&self) -> i32 {
202        0
203    }
204
205    fn max_args(&self) -> Option<i32> {
206        Some(0)
207    }
208
209    fn name(&self) -> &str {
210        "dense_rank"
211    }
212}
213
214// ═══════════════════════════════════════════════════════════════════════════
215// percent_rank()
216// ═══════════════════════════════════════════════════════════════════════════
217
218/// State for `percent_rank()`.
219///
220/// The VDBE must step() all rows in the partition first (to compute ranks
221/// and partition size), then iterate by calling value() and inverse() for
222/// each output row.
223pub struct PercentRankState {
224    partition_size: i64,
225    ranks: Vec<i64>,
226    cursor: usize,
227    step_row_number: i64,
228    current_rank: i64,
229    last_order_value: Option<SqliteValue>,
230}
231
232pub struct PercentRankFunc;
233
234impl WindowFunction for PercentRankFunc {
235    type State = PercentRankState;
236
237    fn initial_state(&self) -> Self::State {
238        PercentRankState {
239            partition_size: 0,
240            ranks: Vec::new(),
241            cursor: 0,
242            step_row_number: 0,
243            current_rank: 0,
244            last_order_value: None,
245        }
246    }
247
248    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
249        state.step_row_number += 1;
250        state.partition_size += 1;
251        let current = args.first().cloned().unwrap_or(SqliteValue::Null);
252        let is_new_peer = match &state.last_order_value {
253            None => true,
254            Some(last) => &current != last,
255        };
256        if is_new_peer {
257            state.current_rank = state.step_row_number;
258            state.last_order_value = Some(current);
259        }
260        state.ranks.push(state.current_rank);
261        Ok(())
262    }
263
264    fn inverse(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
265        state.cursor += 1;
266        Ok(())
267    }
268
269    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
270        if state.partition_size <= 1 {
271            return Ok(SqliteValue::Float(0.0));
272        }
273        let rank = state.ranks.get(state.cursor).copied().unwrap_or(1);
274        let pr = (rank - 1) as f64 / (state.partition_size - 1) as f64;
275        Ok(SqliteValue::Float(pr))
276    }
277
278    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
279        self.value(&state)
280    }
281
282    fn num_args(&self) -> i32 {
283        -1
284    }
285
286    fn min_args(&self) -> i32 {
287        0
288    }
289
290    fn max_args(&self) -> Option<i32> {
291        Some(0)
292    }
293
294    fn name(&self) -> &str {
295        "percent_rank"
296    }
297}
298
299// ═══════════════════════════════════════════════════════════════════════════
300// cume_dist()
301// ═══════════════════════════════════════════════════════════════════════════
302
303/// State for `cume_dist()`.
304///
305/// The VDBE must step() all rows first, then iterate with value()+inverse().
306/// cume_dist = (last peer position) / partition_size.
307pub struct CumeDistState {
308    partition_size: i64,
309    current_row: usize,
310    cume_positions: Vec<i64>,
311    peer_start: usize,
312    last_order_value: Option<SqliteValue>,
313}
314
315pub struct CumeDistFunc;
316
317impl WindowFunction for CumeDistFunc {
318    type State = CumeDistState;
319
320    fn initial_state(&self) -> Self::State {
321        CumeDistState {
322            partition_size: 0,
323            current_row: 0,
324            cume_positions: Vec::new(),
325            peer_start: 0,
326            last_order_value: None,
327        }
328    }
329
330    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
331        let current = args.first().cloned().unwrap_or(SqliteValue::Null);
332        let is_new_peer = match &state.last_order_value {
333            None => true,
334            Some(last) => &current != last,
335        };
336        if is_new_peer {
337            let peer_end = state.partition_size;
338            if let Some(slots) = state.cume_positions.get_mut(state.peer_start..) {
339                for slot in slots {
340                    *slot = peer_end;
341                }
342            }
343            state.peer_start = state.cume_positions.len();
344            state.last_order_value = Some(current);
345        }
346        state.partition_size += 1;
347        state.cume_positions.push(0);
348        Ok(())
349    }
350
351    fn inverse(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
352        state.current_row += 1;
353        Ok(())
354    }
355
356    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
357        if state.partition_size == 0 {
358            return Ok(SqliteValue::Float(0.0));
359        }
360        let peer_end = state
361            .cume_positions
362            .get(state.current_row)
363            .copied()
364            .filter(|position| *position != 0)
365            .unwrap_or(state.partition_size);
366        let cd = peer_end as f64 / state.partition_size as f64;
367        Ok(SqliteValue::Float(cd))
368    }
369
370    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
371        self.value(&state)
372    }
373
374    fn num_args(&self) -> i32 {
375        -1
376    }
377
378    fn min_args(&self) -> i32 {
379        0
380    }
381
382    fn max_args(&self) -> Option<i32> {
383        Some(0)
384    }
385
386    fn name(&self) -> &str {
387        "cume_dist"
388    }
389}
390
391// ═══════════════════════════════════════════════════════════════════════════
392// ntile(N)
393// ═══════════════════════════════════════════════════════════════════════════
394
395/// State for `ntile(N)`.
396///
397/// The VDBE must step() all rows first, then iterate with value()+inverse().
398/// Distributes rows into N groups.  If partition_size % N != 0, the first
399/// (partition_size % N) groups get one extra row.
400pub struct NtileState {
401    partition_size: i64,
402    n: i64,
403    current_row: i64,
404}
405
406pub struct NtileFunc;
407
408const INVALID_NTILE_ARGUMENT: &str = "argument of ntile must be a positive integer";
409
410impl WindowFunction for NtileFunc {
411    type State = NtileState;
412
413    fn initial_state(&self) -> Self::State {
414        NtileState {
415            partition_size: 0,
416            n: 1,
417            current_row: 0,
418        }
419    }
420
421    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
422        if state.partition_size == 0 {
423            let n = args.first().map_or(0, SqliteValue::to_integer);
424            if n <= 0 {
425                return Err(FrankenError::function_error(INVALID_NTILE_ARGUMENT));
426            }
427            state.n = n;
428        }
429        state.partition_size += 1;
430        Ok(())
431    }
432
433    fn inverse(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
434        state.current_row += 1;
435        Ok(())
436    }
437
438    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
439        if state.partition_size == 0 {
440            return Ok(SqliteValue::Integer(1));
441        }
442        let n = state.n;
443        let sz = state.partition_size;
444        let row = state.current_row + 1; // 1-based
445
446        // Group size: first (sz % n) groups get (sz / n + 1) rows,
447        // remaining groups get (sz / n) rows.
448        let base = sz / n;
449        let extra = sz % n;
450        // Rows in "large" groups: extra * (base + 1).
451        let large_rows = extra * (base + 1);
452
453        let bucket = if row <= large_rows {
454            // In one of the first `extra` groups (each of size base+1).
455            (row - 1) / (base + 1) + 1
456        } else {
457            // In one of the remaining groups (each of size base).
458            let adjusted = row - large_rows;
459            if base == 0 {
460                // More buckets than rows; each remaining row gets its own bucket.
461                extra + adjusted
462            } else {
463                extra + (adjusted - 1) / base + 1
464            }
465        };
466        Ok(SqliteValue::Integer(bucket))
467    }
468
469    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
470        self.value(&state)
471    }
472
473    fn num_args(&self) -> i32 {
474        1
475    }
476
477    fn name(&self) -> &str {
478        "ntile"
479    }
480}
481
482// ═══════════════════════════════════════════════════════════════════════════
483// lag(X [, offset [, default]])
484// ═══════════════════════════════════════════════════════════════════════════
485
486fn numeric_prefix_len(bytes: &[u8]) -> usize {
487    let mut idx = 0;
488    if bytes
489        .get(idx)
490        .is_some_and(|byte| matches!(*byte, b'+' | b'-'))
491    {
492        idx += 1;
493    }
494
495    let mut saw_digit = false;
496    while bytes.get(idx).is_some_and(u8::is_ascii_digit) {
497        idx += 1;
498        saw_digit = true;
499    }
500
501    if bytes.get(idx) == Some(&b'.') {
502        idx += 1;
503        while bytes.get(idx).is_some_and(u8::is_ascii_digit) {
504            idx += 1;
505            saw_digit = true;
506        }
507    }
508
509    if !saw_digit {
510        return 0;
511    }
512
513    let mantissa_end = idx;
514    if bytes
515        .get(idx)
516        .is_some_and(|byte| matches!(*byte, b'e' | b'E'))
517    {
518        idx += 1;
519        if bytes
520            .get(idx)
521            .is_some_and(|byte| matches!(*byte, b'+' | b'-'))
522        {
523            idx += 1;
524        }
525        let exp_start = idx;
526        while bytes.get(idx).is_some_and(u8::is_ascii_digit) {
527            idx += 1;
528        }
529        if idx == exp_start {
530            return mantissa_end;
531        }
532    }
533
534    idx
535}
536
537fn trim_ascii_start(bytes: &[u8]) -> &[u8] {
538    let start = bytes
539        .iter()
540        .position(|byte| !byte.is_ascii_whitespace())
541        .unwrap_or(bytes.len());
542    bytes.get(start..).unwrap_or(&[])
543}
544
545fn lag_lead_bytes_offset(bytes: &[u8]) -> Option<i64> {
546    let trimmed = trim_ascii_start(bytes);
547    let prefix_len = numeric_prefix_len(trimmed);
548    if prefix_len == 0 {
549        return Some(0);
550    }
551    let prefix = trimmed
552        .get(..prefix_len)
553        .and_then(|bytes| std::str::from_utf8(bytes).ok())?;
554    if prefix
555        .as_bytes()
556        .iter()
557        .any(|byte| matches!(*byte, b'.' | b'e' | b'E'))
558    {
559        prefix.parse().ok().and_then(integral_f64_to_i64)
560    } else {
561        prefix
562            .parse()
563            .ok()
564            .or_else(|| prefix.parse().ok().and_then(integral_f64_to_i64))
565    }
566}
567
568fn lag_lead_text_offset(text: &str) -> Option<i64> {
569    lag_lead_bytes_offset(text.as_bytes())
570}
571
572fn lag_lead_offset_arg(value: Option<&SqliteValue>) -> Option<i64> {
573    match value {
574        None => Some(1),
575        Some(SqliteValue::Null) => None,
576        Some(SqliteValue::Integer(offset)) => Some(*offset),
577        Some(SqliteValue::Float(offset)) => integral_f64_to_i64(*offset),
578        Some(SqliteValue::Text(text)) => lag_lead_text_offset(text),
579        Some(SqliteValue::Blob(bytes)) => lag_lead_bytes_offset(bytes),
580    }
581}
582
583/// State for `lag()`: maintains a buffer of previous values.
584pub struct LagState {
585    buffer: Vec<SqliteValue>,
586    offsets: Vec<Option<i64>>,
587    defaults: Vec<SqliteValue>,
588    current_row: i64,
589}
590
591pub struct LagFunc;
592
593impl WindowFunction for LagFunc {
594    type State = LagState;
595
596    fn initial_state(&self) -> Self::State {
597        LagState {
598            buffer: Vec::new(),
599            offsets: Vec::new(),
600            defaults: Vec::new(),
601            current_row: 0,
602        }
603    }
604
605    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
606        let val = args.first().cloned().unwrap_or(SqliteValue::Null);
607        let offset = lag_lead_offset_arg(args.get(1));
608        let default_val = args.get(2).cloned().unwrap_or(SqliteValue::Null);
609        state.buffer.push(val);
610        state.offsets.push(offset);
611        state.defaults.push(default_val);
612        Ok(())
613    }
614
615    fn inverse(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
616        state.current_row += 1;
617        Ok(())
618    }
619
620    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
621        let current_index = usize::try_from(state.current_row).unwrap_or(usize::MAX);
622        let default_val = state
623            .defaults
624            .get(current_index)
625            .cloned()
626            .unwrap_or(SqliteValue::Null);
627        let Some(offset) = state.offsets.get(current_index).copied().flatten() else {
628            return Ok(default_val);
629        };
630        let target = state.current_row - offset;
631        let Ok(target_index) = usize::try_from(target) else {
632            return Ok(default_val);
633        };
634        Ok(state
635            .buffer
636            .get(target_index)
637            .cloned()
638            .unwrap_or(default_val))
639    }
640
641    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
642        self.value(&state)
643    }
644
645    fn num_args(&self) -> i32 {
646        -1 // 1, 2, or 3 args
647    }
648
649    fn min_args(&self) -> i32 {
650        1
651    }
652
653    fn max_args(&self) -> Option<i32> {
654        Some(3)
655    }
656
657    fn name(&self) -> &str {
658        "lag"
659    }
660}
661
662// ═══════════════════════════════════════════════════════════════════════════
663// lead(X [, offset [, default]])
664// ═══════════════════════════════════════════════════════════════════════════
665
666/// State for `lead()`: maintains a buffer and reads ahead.
667pub struct LeadState {
668    buffer: Vec<SqliteValue>,
669    offsets: Vec<Option<i64>>,
670    defaults: Vec<SqliteValue>,
671    current_row: i64,
672}
673
674pub struct LeadFunc;
675
676impl WindowFunction for LeadFunc {
677    type State = LeadState;
678
679    fn initial_state(&self) -> Self::State {
680        LeadState {
681            buffer: Vec::new(),
682            offsets: Vec::new(),
683            defaults: Vec::new(),
684            current_row: 0,
685        }
686    }
687
688    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
689        let val = args.first().cloned().unwrap_or(SqliteValue::Null);
690        let offset = lag_lead_offset_arg(args.get(1));
691        let default_val = args.get(2).cloned().unwrap_or(SqliteValue::Null);
692        state.buffer.push(val);
693        state.offsets.push(offset);
694        state.defaults.push(default_val);
695        Ok(())
696    }
697
698    fn inverse(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
699        state.current_row += 1;
700        Ok(())
701    }
702
703    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
704        let current_index = usize::try_from(state.current_row).unwrap_or(usize::MAX);
705        let default_val = state
706            .defaults
707            .get(current_index)
708            .cloned()
709            .unwrap_or(SqliteValue::Null);
710        let Some(offset) = state.offsets.get(current_index).copied().flatten() else {
711            return Ok(default_val);
712        };
713        let target = state.current_row + offset;
714        if target < 0 || target >= state.buffer.len() as i64 {
715            return Ok(default_val);
716        }
717        Ok(state.buffer[target as usize].clone())
718    }
719
720    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
721        self.value(&state)
722    }
723
724    fn num_args(&self) -> i32 {
725        -1
726    }
727
728    fn min_args(&self) -> i32 {
729        1
730    }
731
732    fn max_args(&self) -> Option<i32> {
733        Some(3)
734    }
735
736    fn name(&self) -> &str {
737        "lead"
738    }
739}
740
741// ═══════════════════════════════════════════════════════════════════════════
742// first_value(X)
743// ═══════════════════════════════════════════════════════════════════════════
744
745pub struct FirstValueState {
746    first: Option<SqliteValue>,
747}
748
749pub struct FirstValueFunc;
750
751impl WindowFunction for FirstValueFunc {
752    type State = FirstValueState;
753
754    fn initial_state(&self) -> Self::State {
755        FirstValueState { first: None }
756    }
757
758    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
759        if state.first.is_none() {
760            state.first = Some(args.first().cloned().unwrap_or(SqliteValue::Null));
761        }
762        Ok(())
763    }
764
765    fn inverse(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
766        // When the first row exits the frame, we need to clear so the next
767        // step() captures the new first value.  For simplicity, we use a
768        // VecDeque-based approach in FirstValueFrameFunc below.  This basic
769        // version handles the common UNBOUNDED PRECEDING case correctly.
770        state.first = None;
771        Ok(())
772    }
773
774    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
775        Ok(state.first.clone().unwrap_or(SqliteValue::Null))
776    }
777
778    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
779        Ok(state.first.unwrap_or(SqliteValue::Null))
780    }
781
782    fn num_args(&self) -> i32 {
783        1
784    }
785
786    fn name(&self) -> &str {
787        "first_value"
788    }
789}
790
791// ═══════════════════════════════════════════════════════════════════════════
792// last_value(X)
793// ═══════════════════════════════════════════════════════════════════════════
794
795pub struct LastValueState {
796    frame: VecDeque<SqliteValue>,
797}
798
799pub struct LastValueFunc;
800
801impl WindowFunction for LastValueFunc {
802    type State = LastValueState;
803
804    fn initial_state(&self) -> Self::State {
805        LastValueState {
806            frame: VecDeque::new(),
807        }
808    }
809
810    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
811        state
812            .frame
813            .push_back(args.first().cloned().unwrap_or(SqliteValue::Null));
814        Ok(())
815    }
816
817    fn inverse(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
818        state.frame.pop_front();
819        Ok(())
820    }
821
822    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
823        Ok(state.frame.back().cloned().unwrap_or(SqliteValue::Null))
824    }
825
826    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
827        Ok(state.frame.back().cloned().unwrap_or(SqliteValue::Null))
828    }
829
830    fn num_args(&self) -> i32 {
831        1
832    }
833
834    fn name(&self) -> &str {
835        "last_value"
836    }
837}
838
839// ═══════════════════════════════════════════════════════════════════════════
840// nth_value(X, N)
841// ═══════════════════════════════════════════════════════════════════════════
842
843pub struct NthValueState {
844    frame: VecDeque<SqliteValue>,
845    n: i64,
846}
847
848pub struct NthValueFunc;
849
850const INVALID_NTH_VALUE_ARGUMENT: &str = "second argument to nth_value must be a positive integer";
851
852fn integral_f64_to_i64(value: f64) -> Option<i64> {
853    const I64_MIN_AS_F64: f64 = -9_223_372_036_854_775_808.0;
854    const I64_MAX_EXCLUSIVE_AS_F64: f64 = 9_223_372_036_854_775_808.0;
855
856    if !value.is_finite()
857        || value.fract() != 0.0
858        || !(I64_MIN_AS_F64..I64_MAX_EXCLUSIVE_AS_F64).contains(&value)
859    {
860        return None;
861    }
862    Some(value as i64)
863}
864
865fn parse_integral_text(text: &str) -> Option<i64> {
866    let trimmed = text.trim();
867    if trimmed.is_empty() {
868        return None;
869    }
870    trimmed
871        .parse()
872        .ok()
873        .or_else(|| trimmed.parse().ok().and_then(integral_f64_to_i64))
874}
875
876fn nth_value_positive_integer_arg(value: Option<&SqliteValue>) -> Result<i64> {
877    let Some(value) = value else {
878        return Err(FrankenError::function_error(INVALID_NTH_VALUE_ARGUMENT));
879    };
880    let n = match value {
881        SqliteValue::Integer(n) => Some(*n),
882        SqliteValue::Float(n) => integral_f64_to_i64(*n),
883        SqliteValue::Text(text) => parse_integral_text(text),
884        SqliteValue::Null | SqliteValue::Blob(_) => None,
885    };
886    match n {
887        Some(n) if n > 0 => Ok(n),
888        _ => Err(FrankenError::function_error(INVALID_NTH_VALUE_ARGUMENT)),
889    }
890}
891
892impl WindowFunction for NthValueFunc {
893    type State = NthValueState;
894
895    fn initial_state(&self) -> Self::State {
896        NthValueState {
897            frame: VecDeque::new(),
898            n: 1,
899        }
900    }
901
902    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
903        let val = args.first().cloned().unwrap_or(SqliteValue::Null);
904        let n = nth_value_positive_integer_arg(args.get(1))?;
905        // Capture N from second arg on first call.
906        if state.frame.is_empty() {
907            state.n = n;
908        }
909        state.frame.push_back(val);
910        Ok(())
911    }
912
913    fn inverse(&self, state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
914        state.frame.pop_front();
915        Ok(())
916    }
917
918    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
919        let idx = (state.n - 1) as usize;
920        Ok(state.frame.get(idx).cloned().unwrap_or(SqliteValue::Null))
921    }
922
923    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
924        self.value(&state)
925    }
926
927    fn num_args(&self) -> i32 {
928        2
929    }
930
931    fn name(&self) -> &str {
932        "nth_value"
933    }
934}
935
936// ═══════════════════════════════════════════════════════════════════════════
937// Aggregate-as-window functions: SUM, AVG, COUNT, MIN, MAX, TOTAL
938// SQLite allows any aggregate function to be used as a window function.
939// ═══════════════════════════════════════════════════════════════════════════
940
941pub struct WindowSumState {
942    sum: f64,
943    err: f64,
944    has_value: bool,
945    is_int: bool,
946    int_sum: i64,
947    overflowed: bool,
948}
949
950/// Kahan-Babuska-Neumaier compensated summation step matching upstream
951/// aggregate precision behavior.
952#[inline]
953fn kbn_step(sum: &mut f64, err: &mut f64, value: f64) {
954    let s = *sum;
955    let t = s + value;
956    if s.abs() > value.abs() {
957        *err += (s - t) + value;
958    } else {
959        *err += (value - t) + s;
960    }
961    *sum = t;
962}
963
964pub struct WindowSumFunc;
965
966impl WindowFunction for WindowSumFunc {
967    type State = WindowSumState;
968
969    fn initial_state(&self) -> Self::State {
970        WindowSumState {
971            sum: 0.0,
972            err: 0.0,
973            has_value: false,
974            is_int: true,
975            int_sum: 0,
976            overflowed: false,
977        }
978    }
979
980    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
981        if args.is_empty() || args[0].is_null() {
982            return Ok(());
983        }
984        let value = args[0].to_sum_numeric_value();
985        if value.is_null() {
986            return Ok(());
987        }
988        state.has_value = true;
989        match value {
990            SqliteValue::Integer(i) => {
991                if state.is_int && !state.overflowed {
992                    match state.int_sum.checked_add(i) {
993                        Some(s) => state.int_sum = s,
994                        None => state.overflowed = true,
995                    }
996                }
997                kbn_step(&mut state.sum, &mut state.err, i as f64);
998            }
999            SqliteValue::Float(f) => {
1000                state.is_int = false;
1001                kbn_step(&mut state.sum, &mut state.err, f);
1002            }
1003            SqliteValue::Null | SqliteValue::Text(_) | SqliteValue::Blob(_) => {}
1004        }
1005        Ok(())
1006    }
1007
1008    fn inverse(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1009        if args.is_empty() || args[0].is_null() {
1010            return Ok(());
1011        }
1012        let value = args[0].to_sum_numeric_value();
1013        match value {
1014            SqliteValue::Integer(i) => {
1015                if state.is_int && !state.overflowed {
1016                    match state.int_sum.checked_sub(i) {
1017                        Some(s) => state.int_sum = s,
1018                        None => state.overflowed = true,
1019                    }
1020                }
1021                kbn_step(&mut state.sum, &mut state.err, -(i as f64));
1022            }
1023            SqliteValue::Float(f) => {
1024                state.is_int = false;
1025                kbn_step(&mut state.sum, &mut state.err, -f);
1026            }
1027            SqliteValue::Null | SqliteValue::Text(_) | SqliteValue::Blob(_) => {}
1028        }
1029        Ok(())
1030    }
1031
1032    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
1033        if !state.has_value {
1034            return Ok(SqliteValue::Null);
1035        }
1036        if state.is_int && state.overflowed {
1037            return Err(FrankenError::IntegerOverflow);
1038        }
1039        if state.is_int {
1040            Ok(SqliteValue::Integer(state.int_sum))
1041        } else {
1042            Ok(SqliteValue::Float(state.sum + state.err))
1043        }
1044    }
1045
1046    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
1047        self.value(&state)
1048    }
1049
1050    fn num_args(&self) -> i32 {
1051        1
1052    }
1053
1054    fn name(&self) -> &str {
1055        "SUM"
1056    }
1057}
1058
1059pub struct WindowTotalFunc;
1060
1061impl WindowFunction for WindowTotalFunc {
1062    type State = f64;
1063
1064    fn initial_state(&self) -> Self::State {
1065        0.0
1066    }
1067
1068    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1069        if !args.is_empty() && !args[0].is_null() {
1070            *state += args[0].to_float();
1071        }
1072        Ok(())
1073    }
1074
1075    fn inverse(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1076        if !args.is_empty() && !args[0].is_null() {
1077            *state -= args[0].to_float();
1078        }
1079        Ok(())
1080    }
1081
1082    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
1083        Ok(SqliteValue::Float(*state))
1084    }
1085
1086    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
1087        Ok(SqliteValue::Float(state))
1088    }
1089
1090    fn num_args(&self) -> i32 {
1091        1
1092    }
1093
1094    fn name(&self) -> &str {
1095        "TOTAL"
1096    }
1097}
1098
1099pub struct WindowCountState {
1100    count: i64,
1101}
1102
1103pub struct WindowCountFunc;
1104
1105impl WindowFunction for WindowCountFunc {
1106    type State = WindowCountState;
1107
1108    fn initial_state(&self) -> Self::State {
1109        WindowCountState { count: 0 }
1110    }
1111
1112    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1113        // COUNT(*) has 0 args → count all rows; COUNT(x) skips NULLs.
1114        if args.is_empty() || !args[0].is_null() {
1115            state.count += 1;
1116        }
1117        Ok(())
1118    }
1119
1120    fn inverse(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1121        if args.is_empty() || !args[0].is_null() {
1122            state.count -= 1;
1123        }
1124        Ok(())
1125    }
1126
1127    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
1128        Ok(SqliteValue::Integer(state.count))
1129    }
1130
1131    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
1132        Ok(SqliteValue::Integer(state.count))
1133    }
1134
1135    fn num_args(&self) -> i32 {
1136        -1 // variadic: COUNT(*) = 0 args, COUNT(x) = 1 arg
1137    }
1138
1139    fn min_args(&self) -> i32 {
1140        0
1141    }
1142
1143    fn max_args(&self) -> Option<i32> {
1144        Some(1)
1145    }
1146
1147    fn name(&self) -> &str {
1148        "COUNT"
1149    }
1150}
1151
1152pub struct WindowMinState {
1153    min: Option<SqliteValue>,
1154}
1155
1156pub struct WindowMinFunc;
1157
1158impl WindowFunction for WindowMinFunc {
1159    type State = WindowMinState;
1160
1161    fn initial_state(&self) -> Self::State {
1162        WindowMinState { min: None }
1163    }
1164
1165    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1166        if args.is_empty() || args[0].is_null() {
1167            return Ok(());
1168        }
1169        state.min = Some(match state.min.take() {
1170            None => args[0].clone(),
1171            Some(cur) => {
1172                if cmp_values(&args[0], &cur) == std::cmp::Ordering::Less {
1173                    args[0].clone()
1174                } else {
1175                    cur
1176                }
1177            }
1178        });
1179        Ok(())
1180    }
1181
1182    fn inverse(&self, _state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
1183        // MIN inverse is not efficiently invertible; no-op for unbounded frames.
1184        Ok(())
1185    }
1186
1187    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
1188        Ok(state.min.clone().unwrap_or(SqliteValue::Null))
1189    }
1190
1191    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
1192        Ok(state.min.unwrap_or(SqliteValue::Null))
1193    }
1194
1195    fn num_args(&self) -> i32 {
1196        1
1197    }
1198
1199    fn name(&self) -> &str {
1200        "MIN"
1201    }
1202}
1203
1204pub struct WindowMaxState {
1205    max: Option<SqliteValue>,
1206}
1207
1208pub struct WindowMaxFunc;
1209
1210impl WindowFunction for WindowMaxFunc {
1211    type State = WindowMaxState;
1212
1213    fn initial_state(&self) -> Self::State {
1214        WindowMaxState { max: None }
1215    }
1216
1217    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1218        if args.is_empty() || args[0].is_null() {
1219            return Ok(());
1220        }
1221        state.max = Some(match state.max.take() {
1222            None => args[0].clone(),
1223            Some(cur) => {
1224                if cmp_values(&args[0], &cur) == std::cmp::Ordering::Greater {
1225                    args[0].clone()
1226                } else {
1227                    cur
1228                }
1229            }
1230        });
1231        Ok(())
1232    }
1233
1234    fn inverse(&self, _state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
1235        Ok(())
1236    }
1237
1238    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
1239        Ok(state.max.clone().unwrap_or(SqliteValue::Null))
1240    }
1241
1242    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
1243        Ok(state.max.unwrap_or(SqliteValue::Null))
1244    }
1245
1246    fn num_args(&self) -> i32 {
1247        1
1248    }
1249
1250    fn name(&self) -> &str {
1251        "MAX"
1252    }
1253}
1254
1255pub struct WindowAvgState {
1256    sum: f64,
1257    err: f64,
1258    count: i64,
1259}
1260
1261pub struct WindowAvgFunc;
1262
1263impl WindowFunction for WindowAvgFunc {
1264    type State = WindowAvgState;
1265
1266    fn initial_state(&self) -> Self::State {
1267        WindowAvgState {
1268            sum: 0.0,
1269            err: 0.0,
1270            count: 0,
1271        }
1272    }
1273
1274    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1275        if args.is_empty() || args[0].is_null() {
1276            return Ok(());
1277        }
1278        kbn_step(&mut state.sum, &mut state.err, args[0].to_float());
1279        state.count += 1;
1280        Ok(())
1281    }
1282
1283    fn inverse(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1284        if args.is_empty() || args[0].is_null() {
1285            return Ok(());
1286        }
1287        kbn_step(&mut state.sum, &mut state.err, -args[0].to_float());
1288        state.count -= 1;
1289        Ok(())
1290    }
1291
1292    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
1293        if state.count == 0 {
1294            Ok(SqliteValue::Null)
1295        } else {
1296            #[allow(clippy::cast_precision_loss)]
1297            Ok(SqliteValue::Float(
1298                (state.sum + state.err) / state.count as f64,
1299            ))
1300        }
1301    }
1302
1303    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
1304        self.value(&state)
1305    }
1306
1307    fn num_args(&self) -> i32 {
1308        1
1309    }
1310
1311    fn name(&self) -> &str {
1312        "AVG"
1313    }
1314}
1315
1316pub struct WindowGroupConcatState {
1317    result: String,
1318    has_value: bool,
1319}
1320
1321fn window_group_concat_step(state: &mut WindowGroupConcatState, args: &[SqliteValue]) {
1322    if args.is_empty() || args[0].is_null() {
1323        return;
1324    }
1325    if state.has_value {
1326        match args.get(1) {
1327            Some(separator) if !separator.is_null() => {
1328                if let Some(text) = separator.as_text_str() {
1329                    state.result.push_str(text);
1330                } else {
1331                    state.result.push_str(&separator.to_text());
1332                }
1333            }
1334            Some(_) => {}
1335            None => state.result.push(','),
1336        }
1337    }
1338    if let Some(text) = args[0].as_text_str() {
1339        state.result.push_str(text);
1340    } else {
1341        state.result.push_str(&args[0].to_text());
1342    }
1343    state.has_value = true;
1344}
1345
1346fn window_group_concat_value(state: &WindowGroupConcatState) -> SqliteValue {
1347    if state.has_value {
1348        SqliteValue::Text(state.result.clone().into())
1349    } else {
1350        SqliteValue::Null
1351    }
1352}
1353
1354pub struct WindowGroupConcatFunc;
1355
1356impl WindowFunction for WindowGroupConcatFunc {
1357    type State = WindowGroupConcatState;
1358
1359    fn initial_state(&self) -> Self::State {
1360        WindowGroupConcatState {
1361            result: String::new(),
1362            has_value: false,
1363        }
1364    }
1365
1366    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1367        window_group_concat_step(state, args);
1368        Ok(())
1369    }
1370
1371    fn inverse(&self, _state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
1372        // Sliding ROWS/RANGE/GROUPS frames are recomputed by the connection-level
1373        // window executor; the remaining paths never evict rows from the left edge.
1374        Ok(())
1375    }
1376
1377    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
1378        Ok(window_group_concat_value(state))
1379    }
1380
1381    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
1382        Ok(window_group_concat_value(&state))
1383    }
1384
1385    fn num_args(&self) -> i32 {
1386        -1
1387    }
1388
1389    fn min_args(&self) -> i32 {
1390        1
1391    }
1392
1393    fn max_args(&self) -> Option<i32> {
1394        Some(2)
1395    }
1396
1397    fn name(&self) -> &str {
1398        "group_concat"
1399    }
1400}
1401
1402pub struct WindowStringAggFunc;
1403
1404impl WindowFunction for WindowStringAggFunc {
1405    type State = WindowGroupConcatState;
1406
1407    fn initial_state(&self) -> Self::State {
1408        WindowGroupConcatState {
1409            result: String::new(),
1410            has_value: false,
1411        }
1412    }
1413
1414    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
1415        window_group_concat_step(state, args);
1416        Ok(())
1417    }
1418
1419    fn inverse(&self, _state: &mut Self::State, _args: &[SqliteValue]) -> Result<()> {
1420        Ok(())
1421    }
1422
1423    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
1424        Ok(window_group_concat_value(state))
1425    }
1426
1427    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
1428        Ok(window_group_concat_value(&state))
1429    }
1430
1431    fn num_args(&self) -> i32 {
1432        2
1433    }
1434
1435    fn name(&self) -> &str {
1436        "string_agg"
1437    }
1438}
1439
1440/// Compare two SQLite values using type-aware ordering.
1441pub fn cmp_values(a: &SqliteValue, b: &SqliteValue) -> std::cmp::Ordering {
1442    a.cmp(b)
1443}
1444
1445// ── Registration ──────────────────────────────────────────────────────────
1446
1447/// Register all S13.5 window functions.
1448pub fn register_window_builtins(registry: &mut FunctionRegistry) {
1449    registry.register_window(RowNumberFunc);
1450    registry.register_window(RankFunc);
1451    registry.register_window(DenseRankFunc);
1452    registry.register_window(PercentRankFunc);
1453    registry.register_window(CumeDistFunc);
1454    registry.register_window(NtileFunc);
1455    registry.register_window(LagFunc);
1456    registry.register_window(LeadFunc);
1457    registry.register_window(FirstValueFunc);
1458    registry.register_window(LastValueFunc);
1459    registry.register_window(NthValueFunc);
1460
1461    // Aggregate-as-window functions (SQLite allows any aggregate as a window fn).
1462    registry.register_window(WindowSumFunc);
1463    registry.register_window(WindowTotalFunc);
1464    registry.register_window(WindowCountFunc);
1465    registry.register_window(WindowMinFunc);
1466    registry.register_window(WindowMaxFunc);
1467    registry.register_window(WindowAvgFunc);
1468    registry.register_window(WindowGroupConcatFunc);
1469    registry.register_window(WindowStringAggFunc);
1470}
1471
1472// ── Tests ─────────────────────────────────────────────────────────────────
1473
1474#[cfg(test)]
1475mod tests {
1476    use super::*;
1477
1478    fn int(v: i64) -> SqliteValue {
1479        SqliteValue::Integer(v)
1480    }
1481
1482    fn float(v: f64) -> SqliteValue {
1483        SqliteValue::Float(v)
1484    }
1485
1486    fn text(s: &str) -> SqliteValue {
1487        SqliteValue::Text(s.into())
1488    }
1489
1490    fn blob(bytes: &[u8]) -> SqliteValue {
1491        SqliteValue::Blob(std::sync::Arc::from(bytes))
1492    }
1493
1494    fn null() -> SqliteValue {
1495        SqliteValue::Null
1496    }
1497
1498    fn assert_function_error(err: FrankenError, expected: &str) {
1499        assert!(
1500            matches!(&err, FrankenError::FunctionError(message) if message == expected),
1501            "expected function error {expected:?}, got {err:?}"
1502        );
1503    }
1504
1505    fn assert_float_near(value: &SqliteValue, expected: f64) {
1506        assert!(
1507            matches!(value, SqliteValue::Float(_)),
1508            "expected Float, got {value:?}"
1509        );
1510        if let SqliteValue::Float(actual) = value {
1511            assert!(
1512                (*actual - expected).abs() < 1e-10,
1513                "expected {expected}, got {actual}"
1514            );
1515        }
1516    }
1517
1518    /// Simulate a partition by calling step() for each row, collecting
1519    /// value() after each step.  Returns the vector of per-row results.
1520    /// Suitable for progressive functions (row_number, rank, dense_rank).
1521    fn run_window_partition<F: WindowFunction>(
1522        func: &F,
1523        rows: &[Vec<SqliteValue>],
1524    ) -> Vec<SqliteValue> {
1525        let mut state = func.initial_state();
1526        let mut results = Vec::new();
1527        for row in rows {
1528            func.step(&mut state, row).unwrap();
1529            results.push(func.value(&state).unwrap());
1530        }
1531        results
1532    }
1533
1534    /// Two-pass partition simulation: step all rows first (pass 1), then
1535    /// iterate calling value()+inverse() for each row (pass 2).
1536    /// Required for functions that need full partition size (ntile,
1537    /// percent_rank, cume_dist).
1538    fn run_window_two_pass<F: WindowFunction>(
1539        func: &F,
1540        rows: &[Vec<SqliteValue>],
1541    ) -> Vec<SqliteValue> {
1542        let mut state = func.initial_state();
1543        // Pass 1: step all rows.
1544        for row in rows {
1545            func.step(&mut state, row).unwrap();
1546        }
1547        // Pass 2: read values, advance cursor via inverse().
1548        let mut results = Vec::new();
1549        for (i, _) in rows.iter().enumerate() {
1550            results.push(func.value(&state).unwrap());
1551            if i < rows.len() - 1 {
1552                func.inverse(&mut state, &[]).unwrap();
1553            }
1554        }
1555        results
1556    }
1557
1558    // ── row_number ───────────────────────────────────────────────────
1559
1560    #[test]
1561    fn test_row_number_basic() {
1562        let results =
1563            run_window_partition(&RowNumberFunc, &[vec![], vec![], vec![], vec![], vec![]]);
1564        assert_eq!(results, vec![int(1), int(2), int(3), int(4), int(5)]);
1565    }
1566
1567    #[test]
1568    fn test_row_number_partition_reset() {
1569        // Partition 1: 3 rows.
1570        let r1 = run_window_partition(&RowNumberFunc, &[vec![], vec![], vec![]]);
1571        assert_eq!(r1, vec![int(1), int(2), int(3)]);
1572
1573        // Partition 2: 2 rows (fresh state).
1574        let r2 = run_window_partition(&RowNumberFunc, &[vec![], vec![]]);
1575        assert_eq!(r2, vec![int(1), int(2)]);
1576    }
1577
1578    // ── rank ─────────────────────────────────────────────────────────
1579
1580    #[test]
1581    fn test_rank_with_ties() {
1582        // Values: [1, 2, 2, 3] -> ranks: [1, 2, 2, 4]
1583        let results = run_window_partition(
1584            &RankFunc,
1585            &[vec![int(1)], vec![int(2)], vec![int(2)], vec![int(3)]],
1586        );
1587        assert_eq!(results, vec![int(1), int(2), int(2), int(4)]);
1588    }
1589
1590    #[test]
1591    fn test_rank_no_ties() {
1592        let results =
1593            run_window_partition(&RankFunc, &[vec![int(10)], vec![int(20)], vec![int(30)]]);
1594        assert_eq!(results, vec![int(1), int(2), int(3)]);
1595    }
1596
1597    // ── dense_rank ───────────────────────────────────────────────────
1598
1599    #[test]
1600    fn test_dense_rank_with_ties() {
1601        // Values: [1, 2, 2, 3] -> dense_ranks: [1, 2, 2, 3]
1602        let results = run_window_partition(
1603            &DenseRankFunc,
1604            &[vec![int(1)], vec![int(2)], vec![int(2)], vec![int(3)]],
1605        );
1606        assert_eq!(results, vec![int(1), int(2), int(2), int(3)]);
1607    }
1608
1609    #[test]
1610    fn test_dense_rank_multiple_ties() {
1611        // Values: [1, 1, 2, 2, 3] -> dense_ranks: [1, 1, 2, 2, 3]
1612        let results = run_window_partition(
1613            &DenseRankFunc,
1614            &[
1615                vec![int(1)],
1616                vec![int(1)],
1617                vec![int(2)],
1618                vec![int(2)],
1619                vec![int(3)],
1620            ],
1621        );
1622        assert_eq!(results, vec![int(1), int(1), int(2), int(2), int(3)]);
1623    }
1624
1625    // ── percent_rank ─────────────────────────────────────────────────
1626
1627    #[test]
1628    fn test_percent_rank_single_row() {
1629        let results = run_window_two_pass(&PercentRankFunc, &[vec![int(1)]]);
1630        assert_eq!(results, vec![SqliteValue::Float(0.0)]);
1631    }
1632
1633    #[test]
1634    fn test_percent_rank_formula() {
1635        // 4 rows, values [1, 2, 2, 3] -> ranks [1, 2, 2, 4]
1636        // percent_rank = (rank - 1) / (N - 1) = (rank - 1) / 3
1637        let results = run_window_two_pass(
1638            &PercentRankFunc,
1639            &[vec![int(1)], vec![int(2)], vec![int(2)], vec![int(3)]],
1640        );
1641        // Row 1: (1-1)/3 = 0.0
1642        // Row 2: (2-1)/3 = 0.333...
1643        // Row 3: (2-1)/3 = 0.333... (same rank as row 2)
1644        // Row 4: (4-1)/3 = 1.0
1645        assert_float_near(&results[0], 0.0);
1646        assert_float_near(&results[1], 1.0 / 3.0);
1647        assert_float_near(&results[2], 1.0 / 3.0);
1648        assert_float_near(&results[3], 1.0);
1649    }
1650
1651    #[test]
1652    fn test_percent_rank_without_order_treats_partition_as_one_peer_group() {
1653        let results = run_window_two_pass(&PercentRankFunc, &[vec![], vec![], vec![]]);
1654        for value in results {
1655            assert_float_near(&value, 0.0);
1656        }
1657    }
1658
1659    // ── cume_dist ────────────────────────────────────────────────────
1660
1661    #[test]
1662    fn test_cume_dist_distinct() {
1663        // 4 distinct values: cume_dist = [0.25, 0.5, 0.75, 1.0]
1664        let results = run_window_two_pass(
1665            &CumeDistFunc,
1666            &[vec![int(1)], vec![int(2)], vec![int(3)], vec![int(4)]],
1667        );
1668        for (i, expected) in [0.25, 0.5, 0.75, 1.0].iter().enumerate() {
1669            assert_float_near(&results[i], *expected);
1670        }
1671    }
1672
1673    #[test]
1674    fn test_cume_dist_with_ties() {
1675        // Values [1, 2, 2, 3] -> last peer positions [1, 3, 3, 4].
1676        let results = run_window_two_pass(
1677            &CumeDistFunc,
1678            &[vec![int(1)], vec![int(2)], vec![int(2)], vec![int(3)]],
1679        );
1680        for (i, expected) in [0.25, 0.75, 0.75, 1.0].iter().enumerate() {
1681            assert_float_near(&results[i], *expected);
1682        }
1683    }
1684
1685    #[test]
1686    fn test_cume_dist_without_order_treats_partition_as_one_peer_group() {
1687        let results = run_window_two_pass(&CumeDistFunc, &[vec![], vec![], vec![]]);
1688        for value in results {
1689            assert_float_near(&value, 1.0);
1690        }
1691    }
1692
1693    #[test]
1694    fn test_cume_dist_null_peers_share_same_peer_group() {
1695        let results =
1696            run_window_two_pass(&CumeDistFunc, &[vec![null()], vec![null()], vec![int(1)]]);
1697        assert_float_near(&results[0], 2.0 / 3.0);
1698        assert_float_near(&results[1], 2.0 / 3.0);
1699        assert_float_near(&results[2], 1.0);
1700    }
1701
1702    // ── ntile ────────────────────────────────────────────────────────
1703
1704    #[test]
1705    fn test_ntile_even() {
1706        // ntile(4) over 8 rows: groups of 2 each -> [1,1,2,2,3,3,4,4]
1707        let rows: Vec<Vec<SqliteValue>> = (0..8).map(|_| vec![int(4)]).collect();
1708        let results = run_window_two_pass(&NtileFunc, &rows);
1709        assert_eq!(
1710            results,
1711            vec![
1712                int(1),
1713                int(1),
1714                int(2),
1715                int(2),
1716                int(3),
1717                int(3),
1718                int(4),
1719                int(4)
1720            ]
1721        );
1722    }
1723
1724    #[test]
1725    fn test_ntile_uneven() {
1726        // ntile(3) over 10 rows: groups of 4,3,3
1727        let rows: Vec<Vec<SqliteValue>> = (0..10).map(|_| vec![int(3)]).collect();
1728        let results = run_window_two_pass(&NtileFunc, &rows);
1729        assert_eq!(
1730            results,
1731            vec![
1732                int(1),
1733                int(1),
1734                int(1),
1735                int(1),
1736                int(2),
1737                int(2),
1738                int(2),
1739                int(3),
1740                int(3),
1741                int(3)
1742            ]
1743        );
1744    }
1745
1746    #[test]
1747    fn test_ntile_more_buckets_than_rows() {
1748        // ntile(10) over 3 rows: [1, 2, 3]
1749        let rows: Vec<Vec<SqliteValue>> = (0..3).map(|_| vec![int(10)]).collect();
1750        let results = run_window_two_pass(&NtileFunc, &rows);
1751        assert_eq!(results, vec![int(1), int(2), int(3)]);
1752    }
1753
1754    #[test]
1755    fn test_ntile_rejects_non_positive_argument() {
1756        for n in [0, -1] {
1757            let mut state = NtileFunc.initial_state();
1758            let err = NtileFunc.step(&mut state, &[int(n)]).unwrap_err();
1759            assert_function_error(err, INVALID_NTILE_ARGUMENT);
1760        }
1761    }
1762
1763    // ── lag ──────────────────────────────────────────────────────────
1764
1765    #[test]
1766    fn test_lag_default() {
1767        // lag(X) with default offset=1: previous row's value, NULL for first.
1768        let results = run_window_two_pass(&LagFunc, &[vec![int(10)], vec![int(20)], vec![int(30)]]);
1769        assert_eq!(results, vec![null(), int(10), int(20)]);
1770    }
1771
1772    #[test]
1773    fn test_lag_offset_3() {
1774        // lag(X, 3): 3 rows back.
1775        let results = run_window_two_pass(
1776            &LagFunc,
1777            &[
1778                vec![int(10), int(3)],
1779                vec![int(20), int(3)],
1780                vec![int(30), int(3)],
1781                vec![int(40), int(3)],
1782                vec![int(50), int(3)],
1783            ],
1784        );
1785        assert_eq!(results, vec![null(), null(), null(), int(10), int(20)]);
1786    }
1787
1788    #[test]
1789    fn test_lag_default_value() {
1790        // lag(X, 1, -1): returns -1 when no previous row.
1791        let results = run_window_two_pass(
1792            &LagFunc,
1793            &[
1794                vec![int(10), int(1), int(-1)],
1795                vec![int(20), int(1), int(-1)],
1796            ],
1797        );
1798        assert_eq!(results, vec![int(-1), int(10)]);
1799    }
1800
1801    #[test]
1802    fn test_lag_null_offset_returns_default_for_each_row() {
1803        let results = run_window_two_pass(
1804            &LagFunc,
1805            &[
1806                vec![int(10), null(), text("N/A")],
1807                vec![int(20), null(), text("N/A")],
1808            ],
1809        );
1810        assert_eq!(results, vec![text("N/A"), text("N/A")]);
1811    }
1812
1813    #[test]
1814    fn test_lag_uses_current_row_offset_and_default() {
1815        let results = run_window_two_pass(
1816            &LagFunc,
1817            &[
1818                vec![int(10), int(1), text("first")],
1819                vec![int(20), null(), text("null-offset")],
1820                vec![int(30), int(1), text("third")],
1821                vec![int(40), int(2), text("fourth")],
1822            ],
1823        );
1824        assert_eq!(
1825            results,
1826            vec![text("first"), text("null-offset"), int(20), int(20)]
1827        );
1828    }
1829
1830    #[test]
1831    fn test_lag_negative_offset_reads_following_row() {
1832        let results = run_window_two_pass(
1833            &LagFunc,
1834            &[
1835                vec![int(10), int(-1), text("N/A")],
1836                vec![int(20), int(-1), text("N/A")],
1837                vec![int(30), int(-1), text("N/A")],
1838            ],
1839        );
1840        assert_eq!(results, vec![int(20), int(30), text("N/A")]);
1841    }
1842
1843    #[test]
1844    fn test_lag_fractional_offset_uses_default() {
1845        let results = run_window_two_pass(
1846            &LagFunc,
1847            &[
1848                vec![int(10), float(1.5), text("N/A")],
1849                vec![int(20), float(1.5), text("N/A")],
1850                vec![int(30), float(1.5), text("N/A")],
1851            ],
1852        );
1853        assert_eq!(results, vec![text("N/A"), text("N/A"), text("N/A")]);
1854    }
1855
1856    #[test]
1857    fn test_lag_nonnumeric_text_offset_reads_current_row() {
1858        let results = run_window_two_pass(
1859            &LagFunc,
1860            &[
1861                vec![int(10), text("abc"), text("N/A")],
1862                vec![int(20), text("abc"), text("N/A")],
1863                vec![int(30), text("abc"), text("N/A")],
1864            ],
1865        );
1866        assert_eq!(results, vec![int(10), int(20), int(30)]);
1867    }
1868
1869    #[test]
1870    fn test_lag_integral_text_prefix_offset() {
1871        let results = run_window_two_pass(
1872            &LagFunc,
1873            &[
1874                vec![int(10), text("2.0x"), text("N/A")],
1875                vec![int(20), text("2e0x"), text("N/A")],
1876                vec![int(30), text("2x"), text("N/A")],
1877            ],
1878        );
1879        assert_eq!(results, vec![text("N/A"), text("N/A"), int(10)]);
1880    }
1881
1882    // ── lead ─────────────────────────────────────────────────────────
1883
1884    #[test]
1885    fn test_lead_default() {
1886        // lead(X): next row's value, NULL for last.
1887        // For lead, we need to step all rows first (to build the buffer),
1888        // then call inverse + value for each row.
1889        let func = LeadFunc;
1890        let mut state = func.initial_state();
1891        let rows = [int(10), int(20), int(30)];
1892
1893        // Step all rows to build the buffer.
1894        for row in &rows {
1895            func.step(&mut state, std::slice::from_ref(row)).unwrap();
1896        }
1897
1898        // Now iterate: first row is at current_row=0.
1899        let mut results = Vec::new();
1900        for _ in &rows {
1901            results.push(func.value(&state).unwrap());
1902            func.inverse(&mut state, &[]).unwrap();
1903        }
1904        assert_eq!(results, vec![int(20), int(30), null()]);
1905    }
1906
1907    #[test]
1908    fn test_lead_offset_2() {
1909        let func = LeadFunc;
1910        let mut state = func.initial_state();
1911        let rows = [int(10), int(20), int(30), int(40), int(50)];
1912
1913        for row in &rows {
1914            func.step(&mut state, &[row.clone(), int(2)]).unwrap();
1915        }
1916
1917        let mut results = Vec::new();
1918        for _ in &rows {
1919            results.push(func.value(&state).unwrap());
1920            func.inverse(&mut state, &[]).unwrap();
1921        }
1922        assert_eq!(results, vec![int(30), int(40), int(50), null(), null()]);
1923    }
1924
1925    #[test]
1926    fn test_lead_default_value() {
1927        let func = LeadFunc;
1928        let mut state = func.initial_state();
1929        let rows = [int(10), int(20)];
1930
1931        for row in &rows {
1932            func.step(&mut state, &[row.clone(), int(1), text("N/A")])
1933                .unwrap();
1934        }
1935
1936        let mut results = Vec::new();
1937        for _ in &rows {
1938            results.push(func.value(&state).unwrap());
1939            func.inverse(&mut state, &[]).unwrap();
1940        }
1941        assert_eq!(results, vec![int(20), text("N/A")]);
1942    }
1943
1944    #[test]
1945    fn test_lead_null_offset_returns_default_for_each_row() {
1946        let func = LeadFunc;
1947        let mut state = func.initial_state();
1948        let rows = [int(10), int(20)];
1949
1950        for row in &rows {
1951            func.step(&mut state, &[row.clone(), null(), text("N/A")])
1952                .unwrap();
1953        }
1954
1955        let mut results = Vec::new();
1956        for _ in &rows {
1957            results.push(func.value(&state).unwrap());
1958            func.inverse(&mut state, &[]).unwrap();
1959        }
1960        assert_eq!(results, vec![text("N/A"), text("N/A")]);
1961    }
1962
1963    #[test]
1964    fn test_lead_uses_current_row_offset_and_default() {
1965        let func = LeadFunc;
1966        let mut state = func.initial_state();
1967        let rows = [
1968            vec![int(10), int(1), text("first")],
1969            vec![int(20), null(), text("null-offset")],
1970            vec![int(30), int(1), text("third")],
1971        ];
1972
1973        for row in &rows {
1974            func.step(&mut state, row).unwrap();
1975        }
1976
1977        let mut results = Vec::new();
1978        for _ in &rows {
1979            results.push(func.value(&state).unwrap());
1980            func.inverse(&mut state, &[]).unwrap();
1981        }
1982        assert_eq!(results, vec![int(20), text("null-offset"), text("third")]);
1983    }
1984
1985    #[test]
1986    fn test_lead_negative_offset_reads_previous_row() {
1987        let func = LeadFunc;
1988        let mut state = func.initial_state();
1989        let rows = [int(10), int(20), int(30)];
1990
1991        for row in &rows {
1992            func.step(&mut state, &[row.clone(), int(-1), text("N/A")])
1993                .unwrap();
1994        }
1995
1996        let mut results = Vec::new();
1997        for _ in &rows {
1998            results.push(func.value(&state).unwrap());
1999            func.inverse(&mut state, &[]).unwrap();
2000        }
2001        assert_eq!(results, vec![text("N/A"), int(10), int(20)]);
2002    }
2003
2004    #[test]
2005    fn test_lead_fractional_offset_uses_default() {
2006        let results = run_window_two_pass(
2007            &LeadFunc,
2008            &[
2009                vec![int(10), float(1.5), text("N/A")],
2010                vec![int(20), float(1.5), text("N/A")],
2011                vec![int(30), float(1.5), text("N/A")],
2012            ],
2013        );
2014        assert_eq!(results, vec![text("N/A"), text("N/A"), text("N/A")]);
2015    }
2016
2017    #[test]
2018    fn test_lead_integral_blob_offset() {
2019        let results = run_window_two_pass(
2020            &LeadFunc,
2021            &[
2022                vec![int(10), blob(b"2.0x"), text("N/A")],
2023                vec![int(20), blob(b"2e0x"), text("N/A")],
2024                vec![int(30), blob(b"2x"), text("N/A")],
2025            ],
2026        );
2027        assert_eq!(results, vec![int(30), text("N/A"), text("N/A")]);
2028    }
2029
2030    // ── first_value ──────────────────────────────────────────────────
2031
2032    #[test]
2033    fn test_first_value_basic() {
2034        let results = run_window_partition(
2035            &FirstValueFunc,
2036            &[vec![int(10)], vec![int(20)], vec![int(30)]],
2037        );
2038        // With default frame (UNBOUNDED PRECEDING to CURRENT ROW),
2039        // first_value is always the first row's value.
2040        assert_eq!(results, vec![int(10), int(10), int(10)]);
2041    }
2042
2043    // ── last_value ───────────────────────────────────────────────────
2044
2045    #[test]
2046    fn test_last_value_default_frame() {
2047        // With default frame (RANGE BETWEEN UNBOUNDED PRECEDING AND CURRENT ROW),
2048        // last_value returns the current row's value.
2049        let results = run_window_partition(
2050            &LastValueFunc,
2051            &[vec![int(10)], vec![int(20)], vec![int(30)]],
2052        );
2053        assert_eq!(results, vec![int(10), int(20), int(30)]);
2054    }
2055
2056    #[test]
2057    fn test_last_value_unbounded_following() {
2058        // With UNBOUNDED FOLLOWING frame, step all rows first,
2059        // then value() returns the true last value.
2060        let func = LastValueFunc;
2061        let mut state = func.initial_state();
2062        func.step(&mut state, &[int(10)]).unwrap();
2063        func.step(&mut state, &[int(20)]).unwrap();
2064        func.step(&mut state, &[int(30)]).unwrap();
2065        assert_eq!(func.value(&state).unwrap(), int(30));
2066    }
2067
2068    // ── nth_value ────────────────────────────────────────────────────
2069
2070    #[test]
2071    fn test_nth_value_basic() {
2072        let func = NthValueFunc;
2073        let mut state = func.initial_state();
2074        // Step 5 rows; N=3.
2075        func.step(&mut state, &[int(10), int(3)]).unwrap();
2076        func.step(&mut state, &[int(20), int(3)]).unwrap();
2077        func.step(&mut state, &[int(30), int(3)]).unwrap();
2078        func.step(&mut state, &[int(40), int(3)]).unwrap();
2079        func.step(&mut state, &[int(50), int(3)]).unwrap();
2080        assert_eq!(func.value(&state).unwrap(), int(30));
2081    }
2082
2083    #[test]
2084    fn test_nth_value_out_of_range() {
2085        let func = NthValueFunc;
2086        let mut state = func.initial_state();
2087        func.step(&mut state, &[int(10), int(100)]).unwrap();
2088        func.step(&mut state, &[int(20), int(100)]).unwrap();
2089        // Frame has 2 rows but N=100.
2090        assert_eq!(func.value(&state).unwrap(), null());
2091    }
2092
2093    #[test]
2094    fn test_nth_value_n_zero() {
2095        let func = NthValueFunc;
2096        let mut state = func.initial_state();
2097        let err = func.step(&mut state, &[int(10), int(0)]).unwrap_err();
2098        assert_function_error(err, INVALID_NTH_VALUE_ARGUMENT);
2099    }
2100
2101    #[test]
2102    fn test_nth_value_rejects_negative_n() {
2103        let func = NthValueFunc;
2104        let mut state = func.initial_state();
2105        let err = func.step(&mut state, &[int(10), int(-1)]).unwrap_err();
2106        assert_function_error(err, INVALID_NTH_VALUE_ARGUMENT);
2107    }
2108
2109    #[test]
2110    fn test_nth_value_accepts_integral_real_n() {
2111        let func = NthValueFunc;
2112        let mut state = func.initial_state();
2113        func.step(&mut state, &[int(10), float(2.0)]).unwrap();
2114        func.step(&mut state, &[int(20), float(2.0)]).unwrap();
2115        assert_eq!(func.value(&state).unwrap(), int(20));
2116    }
2117
2118    #[test]
2119    fn test_nth_value_accepts_integral_text_n() {
2120        let func = NthValueFunc;
2121        let mut state = func.initial_state();
2122        func.step(&mut state, &[int(10), text("2e0")]).unwrap();
2123        func.step(&mut state, &[int(20), text("2.0")]).unwrap();
2124        assert_eq!(func.value(&state).unwrap(), int(20));
2125    }
2126
2127    #[test]
2128    fn test_nth_value_rejects_fractional_real_n() {
2129        let func = NthValueFunc;
2130        let mut state = func.initial_state();
2131        let err = func.step(&mut state, &[int(10), float(1.5)]).unwrap_err();
2132        assert_function_error(err, INVALID_NTH_VALUE_ARGUMENT);
2133    }
2134
2135    #[test]
2136    fn test_nth_value_rejects_fractional_text_n() {
2137        let func = NthValueFunc;
2138        let mut state = func.initial_state();
2139        let err = func.step(&mut state, &[int(10), text("1.5")]).unwrap_err();
2140        assert_function_error(err, INVALID_NTH_VALUE_ARGUMENT);
2141    }
2142
2143    #[test]
2144    fn test_nth_value_rejects_text_numeric_prefix_n() {
2145        let func = NthValueFunc;
2146        let mut state = func.initial_state();
2147        let err = func.step(&mut state, &[int(10), text("2x")]).unwrap_err();
2148        assert_function_error(err, INVALID_NTH_VALUE_ARGUMENT);
2149    }
2150
2151    #[test]
2152    fn test_nth_value_rejects_blob_integer_n() {
2153        let func = NthValueFunc;
2154        let mut state = func.initial_state();
2155        let err = func.step(&mut state, &[int(10), blob(b"2")]).unwrap_err();
2156        assert_function_error(err, INVALID_NTH_VALUE_ARGUMENT);
2157    }
2158
2159    #[test]
2160    fn test_nth_value_rejects_non_positive_n_after_first_row() {
2161        let func = NthValueFunc;
2162        let mut state = func.initial_state();
2163        func.step(&mut state, &[int(10), int(1)]).unwrap();
2164        let err = func.step(&mut state, &[int(20), int(0)]).unwrap_err();
2165        assert_function_error(err, INVALID_NTH_VALUE_ARGUMENT);
2166    }
2167
2168    // ── sum / total / avg ────────────────────────────────────────────
2169
2170    #[test]
2171    fn test_window_sum_text_integer_literals_stay_integer() {
2172        let results = run_window_partition(&WindowSumFunc, &[vec![text("1")], vec![text("2")]]);
2173        assert_eq!(results, vec![int(1), int(3)]);
2174    }
2175
2176    #[test]
2177    fn test_window_sum_non_numeric_text_returns_real_zero() {
2178        let results = run_window_partition(&WindowSumFunc, &[vec![text("abc")]]);
2179        assert_eq!(results, vec![float(0.0)]);
2180    }
2181
2182    #[test]
2183    fn test_window_sum_overflow_suppressed_after_float_input() {
2184        let func = WindowSumFunc;
2185        let mut state = func.initial_state();
2186
2187        func.step(&mut state, &[int(i64::MAX)]).unwrap();
2188        func.step(&mut state, &[int(1)]).unwrap();
2189        assert_eq!(
2190            func.value(&state).unwrap_err().to_string(),
2191            "integer overflow"
2192        );
2193
2194        func.step(&mut state, &[float(0.5)]).unwrap();
2195        assert!(matches!(func.value(&state).unwrap(), SqliteValue::Float(_)));
2196    }
2197
2198    // ── min / max ────────────────────────────────────────────────────
2199
2200    #[test]
2201    fn test_window_min_max_use_sqlite_storage_class_order() {
2202        let text_value = text("z");
2203        let blob_value = blob(b"\0");
2204
2205        let mut min_state = WindowMinFunc.initial_state();
2206        WindowMinFunc
2207            .step(&mut min_state, std::slice::from_ref(&blob_value))
2208            .unwrap();
2209        WindowMinFunc
2210            .step(&mut min_state, std::slice::from_ref(&text_value))
2211            .unwrap();
2212        assert_eq!(WindowMinFunc.value(&min_state).unwrap(), text_value);
2213
2214        let mut max_state = WindowMaxFunc.initial_state();
2215        WindowMaxFunc
2216            .step(&mut max_state, std::slice::from_ref(&text_value))
2217            .unwrap();
2218        WindowMaxFunc
2219            .step(&mut max_state, std::slice::from_ref(&blob_value))
2220            .unwrap();
2221        assert_eq!(WindowMaxFunc.value(&max_state).unwrap(), blob_value);
2222    }
2223
2224    // ── group_concat / string_agg ────────────────────────────────────
2225
2226    #[test]
2227    fn test_window_group_concat_running_default_separator() {
2228        let results = run_window_partition(
2229            &WindowGroupConcatFunc,
2230            &[vec![text("a")], vec![text("b")], vec![text("c")]],
2231        );
2232        assert_eq!(results, vec![text("a"), text("a,b"), text("a,b,c")]);
2233    }
2234
2235    #[test]
2236    fn test_window_group_concat_running_custom_separator() {
2237        let results = run_window_partition(
2238            &WindowGroupConcatFunc,
2239            &[
2240                vec![text("a"), text(" | ")],
2241                vec![text("b"), text(" | ")],
2242                vec![text("c"), text(" | ")],
2243            ],
2244        );
2245        assert_eq!(results, vec![text("a"), text("a | b"), text("a | b | c")]);
2246    }
2247
2248    #[test]
2249    fn test_window_group_concat_skips_null_and_uses_current_row_separator() {
2250        let results = run_window_partition(
2251            &WindowGroupConcatFunc,
2252            &[
2253                vec![text("a"), text("-")],
2254                vec![null(), text("?")],
2255                vec![text("b"), text("+")],
2256                vec![text("c"), text("*")],
2257            ],
2258        );
2259        assert_eq!(
2260            results,
2261            vec![text("a"), text("a"), text("a+b"), text("a+b*c")]
2262        );
2263    }
2264
2265    #[test]
2266    fn test_window_string_agg_alias_through_registry() {
2267        let mut reg = FunctionRegistry::new();
2268        register_window_builtins(&mut reg);
2269
2270        let sa = reg.find_window("string_agg", 2).unwrap();
2271        let mut state = sa.initial_state();
2272        sa.step(&mut state, &[text("a"), text(";")]).unwrap();
2273        assert_eq!(sa.value(&state).unwrap(), text("a"));
2274        sa.step(&mut state, &[text("b"), text(";")]).unwrap();
2275        assert_eq!(sa.value(&state).unwrap(), text("a;b"));
2276    }
2277
2278    // ── Registration ─────────────────────────────────────────────────
2279
2280    #[test]
2281    fn test_register_window_builtins_all_present() {
2282        let mut reg = FunctionRegistry::new();
2283        register_window_builtins(&mut reg);
2284
2285        let expected_variadic = [
2286            "row_number",
2287            "rank",
2288            "dense_rank",
2289            "percent_rank",
2290            "cume_dist",
2291            "lag",
2292            "lead",
2293        ];
2294        for name in expected_variadic {
2295            assert!(
2296                reg.find_window(name, 0).is_some()
2297                    || reg.find_window(name, 1).is_some()
2298                    || reg.find_window(name, -1).is_some(),
2299                "window function '{name}' not registered"
2300            );
2301        }
2302
2303        assert!(
2304            reg.find_window("ntile", 1).is_some(),
2305            "ntile(1) not registered"
2306        );
2307        assert!(
2308            reg.find_window("first_value", 1).is_some(),
2309            "first_value(1) not registered"
2310        );
2311        assert!(
2312            reg.find_window("last_value", 1).is_some(),
2313            "last_value(1) not registered"
2314        );
2315        assert!(
2316            reg.find_window("nth_value", 2).is_some(),
2317            "nth_value(2) not registered"
2318        );
2319        assert!(
2320            reg.find_window("group_concat", 1).is_some(),
2321            "group_concat(1) not registered"
2322        );
2323        assert!(
2324            reg.find_window("group_concat", 2).is_some(),
2325            "group_concat(2) not registered"
2326        );
2327        assert!(
2328            reg.find_window("string_agg", 2).is_some(),
2329            "string_agg(2) not registered"
2330        );
2331
2332        for (name, arity) in [
2333            ("rank", 1),
2334            ("dense_rank", 1),
2335            ("percent_rank", 1),
2336            ("cume_dist", 1),
2337            ("lag", 0),
2338            ("lag", 4),
2339            ("lead", 0),
2340            ("lead", 4),
2341            ("count", 2),
2342            ("group_concat", 0),
2343            ("group_concat", 3),
2344        ] {
2345            let f = reg
2346                .find_window(name, arity)
2347                .expect("known window with wrong arity returns erroring window");
2348            let mut state = f.initial_state();
2349            let err = f
2350                .step(&mut state, &[])
2351                .expect_err("invalid window arity should fail");
2352            let expected = format!("wrong number of arguments to function {name}()");
2353            assert!(
2354                matches!(&err, FrankenError::FunctionError(message) if message == &expected),
2355                "unexpected error for {name}/{arity}: {err:?}"
2356            );
2357        }
2358    }
2359
2360    // ── E2E: full lifecycle through registry ─────────────────────────
2361
2362    #[test]
2363    fn test_e2e_window_row_number_through_registry() {
2364        let mut reg = FunctionRegistry::new();
2365        register_window_builtins(&mut reg);
2366
2367        let rn = reg.find_window("row_number", 0).unwrap();
2368        let mut state = rn.initial_state();
2369        rn.step(&mut state, &[]).unwrap();
2370        assert_eq!(rn.value(&state).unwrap(), int(1));
2371        rn.step(&mut state, &[]).unwrap();
2372        assert_eq!(rn.value(&state).unwrap(), int(2));
2373        rn.step(&mut state, &[]).unwrap();
2374        assert_eq!(rn.value(&state).unwrap(), int(3));
2375    }
2376
2377    #[test]
2378    fn test_e2e_window_rank_through_registry() {
2379        let mut reg = FunctionRegistry::new();
2380        register_window_builtins(&mut reg);
2381
2382        let rank = reg.find_window("rank", 0).unwrap();
2383        let mut state = rank.initial_state();
2384        // [1, 2, 2, 3] -> [1, 2, 2, 4]
2385        rank.step(&mut state, &[int(1)]).unwrap();
2386        assert_eq!(rank.value(&state).unwrap(), int(1));
2387        rank.step(&mut state, &[int(2)]).unwrap();
2388        assert_eq!(rank.value(&state).unwrap(), int(2));
2389        rank.step(&mut state, &[int(2)]).unwrap();
2390        assert_eq!(rank.value(&state).unwrap(), int(2));
2391        rank.step(&mut state, &[int(3)]).unwrap();
2392        assert_eq!(rank.value(&state).unwrap(), int(4));
2393    }
2394}
fsqlite_func/window_builtins.rs

fsqlite_func/
window_builtins.rs
