reifydb_core/key/
retention_strategy.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_codec::key::{
5	deserializer::KeyDeserializer,
6	encoded::{EncodedKey, EncodedKeyRange},
7	serializer::KeySerializer,
8};
9use reifydb_value::value::dictionary::DictionaryId;
10use serde::{Deserialize, Serialize};
11
12use super::{EncodableKey, KeyKind};
13use crate::interface::catalog::{
14	flow::FlowNodeId,
15	id::{RingBufferId, SeriesId, TableId, ViewId},
16	shape::ShapeId,
17	vtable::VTableId,
18};
19
20#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
21pub struct ShapeRetentionStrategyKey {
22	pub shape: ShapeId,
23}
24
25impl ShapeRetentionStrategyKey {
26	pub fn encoded(shape: impl Into<ShapeId>) -> EncodedKey {
27		Self {
28			shape: shape.into(),
29		}
30		.encode()
31	}
32}
33
34impl EncodableKey for ShapeRetentionStrategyKey {
35	const KIND: KeyKind = KeyKind::ShapeRetentionStrategy;
36
37	fn encode(&self) -> EncodedKey {
38		let mut serializer = KeySerializer::with_capacity(10);
39		serializer.extend_u8(Self::KIND as u8);
40
41		match &self.shape {
42			ShapeId::Table(id) => {
43				serializer.extend_u8(0x01).extend_u64(id.0);
44			}
45			ShapeId::View(id) => {
46				serializer.extend_u8(0x02).extend_u64(id.0);
47			}
48			ShapeId::TableVirtual(id) => {
49				serializer.extend_u8(0x03).extend_u64(id.0);
50			}
51			ShapeId::RingBuffer(id) => {
52				serializer.extend_u8(0x04).extend_u64(id.0);
53			}
54			ShapeId::Dictionary(id) => {
55				serializer.extend_u8(0x06).extend_u64(id.0);
56			}
57			ShapeId::Series(id) => {
58				serializer.extend_u8(0x07).extend_u64(id.0);
59			}
60		}
61
62		serializer.to_encoded_key()
63	}
64
65	fn decode(key: &EncodedKey) -> Option<Self> {
66		let mut de = KeyDeserializer::from_bytes(key.as_slice());
67
68		let kind: KeyKind = de.read_u8().ok()?.try_into().ok()?;
69		if kind != Self::KIND {
70			return None;
71		}
72
73		let discriminator = de.read_u8().ok()?;
74		let id = de.read_u64().ok()?;
75
76		let object_id = match discriminator {
77			0x01 => ShapeId::Table(TableId(id)),
78			0x02 => ShapeId::View(ViewId(id)),
79			0x03 => ShapeId::TableVirtual(VTableId(id)),
80			0x04 => ShapeId::RingBuffer(RingBufferId(id)),
81			0x06 => ShapeId::Dictionary(DictionaryId(id)),
82			0x07 => ShapeId::Series(SeriesId(id)),
83			_ => return None,
84		};
85
86		Some(Self {
87			shape: object_id,
88		})
89	}
90}
91
92#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
93pub struct OperatorRetentionStrategyKey {
94	pub operator: FlowNodeId,
95}
96
97impl OperatorRetentionStrategyKey {
98	pub fn encoded(operator: impl Into<FlowNodeId>) -> EncodedKey {
99		Self {
100			operator: operator.into(),
101		}
102		.encode()
103	}
104}
105
106impl EncodableKey for OperatorRetentionStrategyKey {
107	const KIND: KeyKind = KeyKind::OperatorRetentionStrategy;
108
109	fn encode(&self) -> EncodedKey {
110		let mut serializer = KeySerializer::with_capacity(9);
111		serializer.extend_u8(Self::KIND as u8).extend_u64(self.operator);
112		serializer.to_encoded_key()
113	}
114
115	fn decode(key: &EncodedKey) -> Option<Self> {
116		let mut de = KeyDeserializer::from_bytes(key.as_slice());
117
118		let kind: KeyKind = de.read_u8().ok()?.try_into().ok()?;
119		if kind != Self::KIND {
120			return None;
121		}
122
123		Some(Self {
124			operator: FlowNodeId(de.read_u64().ok()?),
125		})
126	}
127}
128
129pub struct ShapeRetentionStrategyKeyRange;
130
131impl ShapeRetentionStrategyKeyRange {
132	pub fn full_scan() -> EncodedKeyRange {
133		EncodedKeyRange::start_end(Some(Self::start()), Some(Self::end()))
134	}
135
136	fn start() -> EncodedKey {
137		let mut serializer = KeySerializer::with_capacity(1);
138		serializer.extend_u8(ShapeRetentionStrategyKey::KIND as u8);
139		serializer.to_encoded_key()
140	}
141
142	fn end() -> EncodedKey {
143		let mut serializer = KeySerializer::with_capacity(1);
144		serializer.extend_u8(ShapeRetentionStrategyKey::KIND as u8 - 1);
145		serializer.to_encoded_key()
146	}
147}
148
149pub struct OperatorRetentionStrategyKeyRange;
150
151impl OperatorRetentionStrategyKeyRange {
152	pub fn full_scan() -> EncodedKeyRange {
153		EncodedKeyRange::start_end(Some(Self::start()), Some(Self::end()))
154	}
155
156	fn start() -> EncodedKey {
157		let mut serializer = KeySerializer::with_capacity(1);
158		serializer.extend_u8(OperatorRetentionStrategyKey::KIND as u8);
159		serializer.to_encoded_key()
160	}
161
162	fn end() -> EncodedKey {
163		let mut serializer = KeySerializer::with_capacity(1);
164		serializer.extend_u8(OperatorRetentionStrategyKey::KIND as u8 - 1);
165		serializer.to_encoded_key()
166	}
167}
168
169#[cfg(test)]
170pub mod tests {
171	use super::*;
172
173	#[test]
174	fn test_shape_retention_strategy_key_encoding() {
175		let key = ShapeRetentionStrategyKey {
176			shape: ShapeId::Table(TableId(42)),
177		};
178
179		let encoded = key.encode();
180		assert_eq!(encoded[0], 0xE8);
181
182		assert_eq!(encoded.len(), 3);
183		assert_eq!(encoded[1], 0xFE);
184		assert_eq!(encoded[2], 0xD5);
185
186		let decoded = ShapeRetentionStrategyKey::decode(&encoded).unwrap();
187		assert_eq!(key, decoded);
188	}
189
190	#[test]
191	fn test_operator_retention_strategy_key_encoding() {
192		let key = OperatorRetentionStrategyKey {
193			operator: FlowNodeId(12345),
194		};
195
196		let encoded = key.encode();
197		assert_eq!(encoded[0], 0xE7);
198
199		assert_eq!(encoded.len(), 3);
200		assert_eq!(encoded[1], 0x4F);
201		assert_eq!(encoded[2], 0xC6);
202
203		let decoded = OperatorRetentionStrategyKey::decode(&encoded).unwrap();
204		assert_eq!(key, decoded);
205	}
206}
reifydb_core/key/retention_strategy.rs

reifydb_core/key/
retention_strategy.rs
