1#![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
40pub 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
84pub 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) => ¤t != 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 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
151pub 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) => ¤t != 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
214pub 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) => ¤t != 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
299pub 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) => ¤t != 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
391pub 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; let base = sz / n;
449 let extra = sz % n;
450 let large_rows = extra * (base + 1);
452
453 let bucket = if row <= large_rows {
454 (row - 1) / (base + 1) + 1
456 } else {
457 let adjusted = row - large_rows;
459 if base == 0 {
460 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
482fn 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
583pub 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 }
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
662pub 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
741pub 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 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
791pub 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
839pub 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 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
936pub 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#[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 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 }
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 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 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
1440pub fn cmp_values(a: &SqliteValue, b: &SqliteValue) -> std::cmp::Ordering {
1442 a.cmp(b)
1443}
1444
1445pub 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 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#[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 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 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 for row in rows {
1545 func.step(&mut state, row).unwrap();
1546 }
1547 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 #[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 let r1 = run_window_partition(&RowNumberFunc, &[vec![], vec![], vec![]]);
1571 assert_eq!(r1, vec![int(1), int(2), int(3)]);
1572
1573 let r2 = run_window_partition(&RowNumberFunc, &[vec![], vec![]]);
1575 assert_eq!(r2, vec![int(1), int(2)]);
1576 }
1577
1578 #[test]
1581 fn test_rank_with_ties() {
1582 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 #[test]
1600 fn test_dense_rank_with_ties() {
1601 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 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 #[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 let results = run_window_two_pass(
1638 &PercentRankFunc,
1639 &[vec![int(1)], vec![int(2)], vec![int(2)], vec![int(3)]],
1640 );
1641 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 #[test]
1662 fn test_cume_dist_distinct() {
1663 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 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 #[test]
1705 fn test_ntile_even() {
1706 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 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 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 #[test]
1766 fn test_lag_default() {
1767 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 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 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 #[test]
1885 fn test_lead_default() {
1886 let func = LeadFunc;
1890 let mut state = func.initial_state();
1891 let rows = [int(10), int(20), int(30)];
1892
1893 for row in &rows {
1895 func.step(&mut state, std::slice::from_ref(row)).unwrap();
1896 }
1897
1898 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 #[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 assert_eq!(results, vec![int(10), int(10), int(10)]);
2041 }
2042
2043 #[test]
2046 fn test_last_value_default_frame() {
2047 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 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 #[test]
2071 fn test_nth_value_basic() {
2072 let func = NthValueFunc;
2073 let mut state = func.initial_state();
2074 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 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 #[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 #[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 #[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 #[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 #[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 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}