reifydb_core/key/
operator.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};
9
10use super::{EncodableKey, KeyKind};
11use crate::interface::catalog::flow::{FlowId, OperatorId};
12
13#[derive(Debug, Clone, PartialEq)]
14pub struct OperatorKey {
15	pub operator: OperatorId,
16}
17
18impl EncodableKey for OperatorKey {
19	const KIND: KeyKind = KeyKind::Operator;
20
21	fn encode(&self) -> EncodedKey {
22		let mut serializer = KeySerializer::with_capacity(9);
23		serializer.extend_u8(Self::KIND as u8).extend_u64(self.operator);
24		serializer.to_encoded_key()
25	}
26
27	fn decode(key: &EncodedKey) -> Option<Self> {
28		let mut de = KeyDeserializer::from_bytes(key.as_slice());
29
30		let kind: KeyKind = de.read_u8().ok()?.try_into().ok()?;
31		if kind != Self::KIND {
32			return None;
33		}
34
35		let operator = de.read_u64().ok()?;
36
37		Some(Self {
38			operator: OperatorId(operator),
39		})
40	}
41}
42
43impl OperatorKey {
44	pub fn encoded(operator: impl Into<OperatorId>) -> EncodedKey {
45		Self {
46			operator: operator.into(),
47		}
48		.encode()
49	}
50
51	pub fn full_scan() -> EncodedKeyRange {
52		EncodedKeyRange::start_end(Some(Self::start()), Some(Self::end()))
53	}
54
55	fn start() -> EncodedKey {
56		let mut serializer = KeySerializer::with_capacity(1);
57		serializer.extend_u8(Self::KIND as u8);
58		serializer.to_encoded_key()
59	}
60
61	fn end() -> EncodedKey {
62		let mut serializer = KeySerializer::with_capacity(1);
63		serializer.extend_u8((Self::KIND as u8) - 1);
64		serializer.to_encoded_key()
65	}
66}
67
68#[derive(Debug, Clone, PartialEq)]
69pub struct OperatorByFlowKey {
70	pub flow: FlowId,
71	pub operator: OperatorId,
72}
73
74impl EncodableKey for OperatorByFlowKey {
75	const KIND: KeyKind = KeyKind::OperatorByFlow;
76
77	fn encode(&self) -> EncodedKey {
78		let mut serializer = KeySerializer::with_capacity(17);
79		serializer.extend_u8(Self::KIND as u8).extend_u64(self.flow).extend_u64(self.operator);
80		serializer.to_encoded_key()
81	}
82
83	fn decode(key: &EncodedKey) -> Option<Self> {
84		let mut de = KeyDeserializer::from_bytes(key.as_slice());
85
86		let kind: KeyKind = de.read_u8().ok()?.try_into().ok()?;
87		if kind != Self::KIND {
88			return None;
89		}
90
91		let flow = de.read_u64().ok()?;
92		let operator = de.read_u64().ok()?;
93
94		Some(Self {
95			flow: FlowId(flow),
96			operator: OperatorId(operator),
97		})
98	}
99}
100
101impl OperatorByFlowKey {
102	pub fn encoded(flow: impl Into<FlowId>, operator: impl Into<OperatorId>) -> EncodedKey {
103		Self {
104			flow: flow.into(),
105			operator: operator.into(),
106		}
107		.encode()
108	}
109
110	pub fn full_scan(flow: FlowId) -> EncodedKeyRange {
111		EncodedKeyRange::start_end(Some(Self::start(flow)), Some(Self::end(flow)))
112	}
113
114	fn start(flow: FlowId) -> EncodedKey {
115		let mut serializer = KeySerializer::with_capacity(9);
116		serializer.extend_u8(Self::KIND as u8).extend_u64(flow);
117		serializer.to_encoded_key()
118	}
119
120	fn end(flow: FlowId) -> EncodedKey {
121		let mut serializer = KeySerializer::with_capacity(9);
122		serializer.extend_u8(Self::KIND as u8).extend_u64(FlowId(flow.0 - 1));
123		serializer.to_encoded_key()
124	}
125}
126
127#[cfg(test)]
128pub mod tests {
129	use super::{EncodableKey, OperatorByFlowKey, OperatorKey};
130	use crate::interface::catalog::flow::{FlowId, OperatorId};
131
132	#[test]
133	fn test_operator_key_encode_decode() {
134		let key = OperatorKey {
135			operator: OperatorId(0x1234),
136		};
137		let encoded = key.encode();
138		let decoded = OperatorKey::decode(&encoded).unwrap();
139		assert_eq!(decoded.operator, OperatorId(0x1234));
140		assert_eq!(key, decoded);
141	}
142
143	#[test]
144	fn test_operator_by_flow_key_encode_decode() {
145		let key = OperatorByFlowKey {
146			flow: FlowId(0x42),
147			operator: OperatorId(0x1234),
148		};
149		let encoded = key.encode();
150		let decoded = OperatorByFlowKey::decode(&encoded).unwrap();
151		assert_eq!(decoded.flow, FlowId(0x42));
152		assert_eq!(decoded.operator, OperatorId(0x1234));
153		assert_eq!(key, decoded);
154	}
155}
reifydb_core/key/operator.rs

reifydb_core/key/
operator.rs
