reifydb_sub_flow/operator/stateful/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_codec::{encoded::row::EncodedRow, key::encoded::EncodedKey};
5use reifydb_core::{common::CommitVersion, interface::store::MultiVersionRow};
6use reifydb_value::Result;
7
8pub mod counter;
9pub mod keyed;
10pub mod raw;
11pub mod row;
12pub mod single;
13pub mod test_utils;
14pub mod utils;
15pub mod window;
16
17use reifydb_core::key::{
18	EncodableKey, flow_node_internal_state::FlowNodeInternalStateKey, flow_node_state::FlowNodeStateKey,
19};
20
21pub struct StateIterator<'a> {
22	inner: Box<dyn Iterator<Item = Result<MultiVersionRow>> + Send + 'a>,
23}
24
25impl<'a> StateIterator<'a> {
26	pub fn new(inner: Box<dyn Iterator<Item = Result<MultiVersionRow>> + Send + 'a>) -> Self {
27		Self {
28			inner,
29		}
30	}
31}
32
33impl Iterator for StateIterator<'_> {
34	type Item = Result<(EncodedKey, EncodedRow)>;
35
36	fn next(&mut self) -> Option<Self::Item> {
37		match self.inner.next()? {
38			Ok(multi) => {
39				let pair = if let Some(state_key) = FlowNodeStateKey::decode(&multi.key) {
40					(EncodedKey::new(state_key.key), multi.row)
41				} else if let Some(internal_key) = FlowNodeInternalStateKey::decode(&multi.key) {
42					(EncodedKey::new(internal_key.key), multi.row)
43				} else {
44					(multi.key, multi.row)
45				};
46				Some(Ok(pair))
47			}
48			Err(e) => Some(Err(e)),
49		}
50	}
51}
52
53/// Like [`StateIterator`] but also yields the per-key `CommitVersion`. Used by TTL eviction, which
54/// is version-anchored: an entry is expired once its version is at or below the epoch cutoff.
55pub struct StateIteratorVersioned<'a> {
56	inner: Box<dyn Iterator<Item = Result<MultiVersionRow>> + Send + 'a>,
57}
58
59impl<'a> StateIteratorVersioned<'a> {
60	pub fn new(inner: Box<dyn Iterator<Item = Result<MultiVersionRow>> + Send + 'a>) -> Self {
61		Self {
62			inner,
63		}
64	}
65}
66
67impl Iterator for StateIteratorVersioned<'_> {
68	type Item = Result<(EncodedKey, CommitVersion, EncodedRow)>;
69
70	fn next(&mut self) -> Option<Self::Item> {
71		match self.inner.next()? {
72			Ok(multi) => {
73				let version = multi.version;
74				let key = if let Some(state_key) = FlowNodeStateKey::decode(&multi.key) {
75					EncodedKey::new(state_key.key)
76				} else if let Some(internal_key) = FlowNodeInternalStateKey::decode(&multi.key) {
77					EncodedKey::new(internal_key.key)
78				} else {
79					multi.key
80				};
81				Some(Ok((key, version, multi.row)))
82			}
83			Err(e) => Some(Err(e)),
84		}
85	}
86}
reifydb_sub_flow/operator/stateful/mod.rs

reifydb_sub_flow/operator/stateful/
mod.rs
