fsqlite_func/
window.rs

1//! Window function trait with sliding-window support.
2//!
3//! Window functions extend aggregate semantics with the ability to
4//! efficiently process sliding window frames via the `inverse` method.
5//! This enables O(1) per-row computation for frames like
6//! `ROWS BETWEEN 1 PRECEDING AND 1 FOLLOWING`.
7#![allow(clippy::unnecessary_literal_bound)]
8
9use std::any::Any;
10
11use fsqlite_error::Result;
12use fsqlite_types::SqliteValue;
13
14use crate::FunctionArity;
15
16/// A window SQL function (e.g. `SUM() OVER (...)`, custom moving averages).
17///
18/// Window functions extend aggregates with:
19/// - [`inverse`](Self::inverse): remove a row from the frame (enables O(1) sliding windows)
20/// - [`value`](Self::value): peek at the current result without consuming state
21///
22/// This trait is **open** (user-implementable).
23///
24/// # State Lifecycle
25///
26/// 1. [`initial_state`](Self::initial_state) creates a fresh accumulator.
27/// 2. For each row in the frame: [`step`](Self::step) adds, [`inverse`](Self::inverse) removes.
28/// 3. After each step/inverse: [`value`](Self::value) returns the current result.
29/// 4. At partition end: [`finalize`](Self::finalize) consumes state and returns the final value.
30pub trait WindowFunction: Send + Sync {
31    /// The per-partition accumulator type.
32    type State: Send;
33
34    /// Create a fresh accumulator.
35    fn initial_state(&self) -> Self::State;
36
37    /// Add a row to the window frame.
38    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()>;
39
40    /// Remove a row from the window frame (sliding window support).
41    ///
42    /// This is the key difference from [`AggregateFunction`](crate::AggregateFunction):
43    /// by supporting removal, the engine can maintain a running window
44    /// in O(1) time per row rather than recomputing the entire frame.
45    fn inverse(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()>;
46
47    /// Return the current result without consuming state.
48    ///
49    /// Called after each step/inverse to provide the windowed value
50    /// for the current row. Must be callable multiple times.
51    fn value(&self, state: &Self::State) -> Result<SqliteValue>;
52
53    /// Consume the accumulator and produce the final result.
54    fn finalize(&self, state: Self::State) -> Result<SqliteValue>;
55
56    /// The number of arguments this function accepts (`-1` = variadic).
57    fn num_args(&self) -> i32;
58
59    /// Minimum accepted SQL argument count for a variadic function.
60    ///
61    /// Some built-in window functions receive ORDER BY values internally even
62    /// when their SQL call syntax accepts no visible arguments. These bounds
63    /// describe SQL-visible arity, not the runtime step argument slice. The
64    /// default is zero; fixed-arity functions do not consult this method.
65    fn min_args(&self) -> i32 {
66        0
67    }
68
69    /// Maximum accepted SQL argument count for a variadic function.
70    ///
71    /// The default is unbounded. Fixed-arity functions are matched directly
72    /// from [`Self::num_args`] and do not consult this method.
73    fn max_args(&self) -> Option<i32> {
74        None
75    }
76
77    /// Return the complete SQL-visible arity contract in one metadata call.
78    ///
79    /// Registries use this method exactly once before publication, preventing
80    /// a reentrant or stateful [`Self::num_args`] implementation from producing
81    /// a key and bounds from different observations.
82    fn arity(&self) -> FunctionArity {
83        let declared = self.num_args();
84        FunctionArity::from_declared_args(declared, || (self.min_args(), self.max_args()))
85    }
86
87    /// The function name, used in error messages and EXPLAIN output.
88    fn name(&self) -> &str;
89}
90
91/// Type-erased adapter for [`WindowFunction`], analogous to
92/// [`AggregateAdapter`](crate::AggregateAdapter).
93pub struct WindowAdapter<F> {
94    inner: F,
95}
96
97impl<F> WindowAdapter<F> {
98    /// Wrap a concrete window function for type-erased storage.
99    pub const fn new(inner: F) -> Self {
100        Self { inner }
101    }
102}
103
104impl<F> WindowFunction for WindowAdapter<F>
105where
106    F: WindowFunction,
107    F::State: 'static,
108{
109    type State = Box<dyn Any + Send>;
110
111    fn initial_state(&self) -> Self::State {
112        Box::new(self.inner.initial_state())
113    }
114
115    fn step(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
116        let concrete = state
117            .downcast_mut::<F::State>()
118            .expect("window state type mismatch");
119        self.inner.step(concrete, args)
120    }
121
122    fn inverse(&self, state: &mut Self::State, args: &[SqliteValue]) -> Result<()> {
123        let concrete = state
124            .downcast_mut::<F::State>()
125            .expect("window state type mismatch");
126        self.inner.inverse(concrete, args)
127    }
128
129    fn value(&self, state: &Self::State) -> Result<SqliteValue> {
130        let concrete = state
131            .downcast_ref::<F::State>()
132            .expect("window state type mismatch");
133        self.inner.value(concrete)
134    }
135
136    fn finalize(&self, state: Self::State) -> Result<SqliteValue> {
137        let concrete = *state
138            .downcast::<F::State>()
139            .expect("window state type mismatch");
140        self.inner.finalize(concrete)
141    }
142
143    fn num_args(&self) -> i32 {
144        self.inner.num_args()
145    }
146
147    fn min_args(&self) -> i32 {
148        self.inner.min_args()
149    }
150
151    fn max_args(&self) -> Option<i32> {
152        self.inner.max_args()
153    }
154
155    fn arity(&self) -> FunctionArity {
156        self.inner.arity()
157    }
158
159    fn name(&self) -> &str {
160        self.inner.name()
161    }
162}
163
164#[cfg(test)]
165mod tests {
166    use super::*;
167
168    // -- Mock: window sum --
169
170    struct WindowSum;
171
172    impl WindowFunction for WindowSum {
173        type State = i64;
174
175        fn initial_state(&self) -> i64 {
176            0
177        }
178
179        fn step(&self, state: &mut i64, args: &[SqliteValue]) -> Result<()> {
180            *state += args[0].to_integer();
181            Ok(())
182        }
183
184        fn inverse(&self, state: &mut i64, args: &[SqliteValue]) -> Result<()> {
185            *state -= args[0].to_integer();
186            Ok(())
187        }
188
189        fn value(&self, state: &i64) -> Result<SqliteValue> {
190            Ok(SqliteValue::Integer(*state))
191        }
192
193        fn finalize(&self, state: i64) -> Result<SqliteValue> {
194            Ok(SqliteValue::Integer(state))
195        }
196
197        fn num_args(&self) -> i32 {
198            1
199        }
200
201        fn name(&self) -> &str {
202            "window_sum"
203        }
204    }
205
206    #[test]
207    fn test_window_function_step_and_inverse() {
208        let f = WindowSum;
209        let mut state = f.initial_state();
210
211        // Simulate frame [10, 20, 30]
212        f.step(&mut state, &[SqliteValue::Integer(10)]).unwrap();
213        f.step(&mut state, &[SqliteValue::Integer(20)]).unwrap();
214        f.step(&mut state, &[SqliteValue::Integer(30)]).unwrap();
215        assert_eq!(f.value(&state).unwrap(), SqliteValue::Integer(60));
216
217        // Slide: remove 10, add 40 -> frame [20, 30, 40]
218        f.inverse(&mut state, &[SqliteValue::Integer(10)]).unwrap();
219        f.step(&mut state, &[SqliteValue::Integer(40)]).unwrap();
220        assert_eq!(f.value(&state).unwrap(), SqliteValue::Integer(90));
221
222        // Slide: remove 20 -> frame [30, 40]
223        f.inverse(&mut state, &[SqliteValue::Integer(20)]).unwrap();
224        assert_eq!(f.value(&state).unwrap(), SqliteValue::Integer(70));
225    }
226
227    #[test]
228    fn test_window_function_value_without_consuming() {
229        let f = WindowSum;
230        let mut state = f.initial_state();
231
232        f.step(&mut state, &[SqliteValue::Integer(42)]).unwrap();
233
234        // value() can be called multiple times without consuming state.
235        assert_eq!(f.value(&state).unwrap(), SqliteValue::Integer(42));
236        assert_eq!(f.value(&state).unwrap(), SqliteValue::Integer(42));
237        assert_eq!(f.value(&state).unwrap(), SqliteValue::Integer(42));
238
239        // State is still valid after multiple value() calls.
240        f.step(&mut state, &[SqliteValue::Integer(8)]).unwrap();
241        assert_eq!(f.value(&state).unwrap(), SqliteValue::Integer(50));
242    }
243
244    #[test]
245    fn test_window_function_finalize_consumes() {
246        let f = WindowSum;
247        let mut state = f.initial_state();
248
249        f.step(&mut state, &[SqliteValue::Integer(10)]).unwrap();
250        f.step(&mut state, &[SqliteValue::Integer(32)]).unwrap();
251
252        // finalize consumes state and produces final value.
253        let result = f.finalize(state).unwrap();
254        assert_eq!(result, SqliteValue::Integer(42));
255        // `state` is moved — cannot be used after finalize (enforced by Rust move semantics).
256    }
257}
fsqlite_func/window.rs

fsqlite_func/
window.rs
