reifydb_sdk/operator/context/
mod.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4pub mod ffi;
5
6use std::ops::Bound;
7
8use reifydb_codec::{
9	encoded::{
10		row::EncodedRow,
11		shape::{RowShape, fingerprint::RowShapeFingerprint},
12	},
13	key::encoded::EncodedKey,
14};
15use reifydb_core::{
16	common::CommitVersion,
17	interface::catalog::{
18		flow::FlowNodeId,
19		id::{NamespaceId, TableId},
20		namespace::Namespace,
21		table::Table,
22	},
23};
24use reifydb_value::value::{
25	Value,
26	dictionary::{DictionaryEntryId, DictionaryId},
27	row_number::RowNumber,
28};
29use serde::{Serialize, de::DeserializeOwned};
30
31use crate::{
32	error::Result,
33	operator::column::{row::Row, sink::RowSink},
34	state::StateEntry,
35};
36
37pub trait RowEmit {
38	type Sink: RowSink;
39	fn sink(&mut self) -> &mut Self::Sink;
40	fn finish(self, row_numbers: &[RowNumber]) -> Result<()>;
41}
42
43pub trait UpdateEmit {
44	type Sink: RowSink;
45	fn pre(&mut self) -> &mut Self::Sink;
46	fn post(&mut self) -> &mut Self::Sink;
47	fn finish(self, row_numbers: &[RowNumber]) -> Result<()>;
48}
49
50pub trait StateApi {
51	fn get<T: DeserializeOwned>(&self, key: &EncodedKey) -> Result<Option<T>>;
52	fn set<T: Serialize>(&mut self, key: &EncodedKey, value: &T) -> Result<()>;
53	fn remove(&mut self, key: &EncodedKey) -> Result<()>;
54	fn drop(&mut self, key: &EncodedKey) -> Result<()>;
55	fn contains(&self, key: &EncodedKey) -> Result<bool>;
56	fn clear(&mut self) -> Result<()>;
57	fn scan_prefix<T: DeserializeOwned>(&self, prefix: &EncodedKey) -> Result<Vec<(EncodedKey, T)>>;
58	fn get_many<T: DeserializeOwned>(&self, keys: &[EncodedKey]) -> Result<Vec<(EncodedKey, T)>>;
59	fn keys_with_prefix(&self, prefix: &EncodedKey) -> Result<Vec<EncodedKey>>;
60	fn range<T: DeserializeOwned>(
61		&self,
62		start: Bound<&EncodedKey>,
63		end: Bound<&EncodedKey>,
64	) -> Result<Vec<(EncodedKey, T)>>;
65	fn get_with_anchors<T: DeserializeOwned>(&self, key: &EncodedKey) -> Result<Option<StateEntry<T>>>;
66
67	fn get_many_visit<T: DeserializeOwned>(
68		&self,
69		keys: &[EncodedKey],
70		visit: &mut dyn FnMut(EncodedKey, T) -> Result<()>,
71	) -> Result<()> {
72		for (k, v) in self.get_many::<T>(keys)? {
73			visit(k, v)?;
74		}
75		Ok(())
76	}
77
78	fn range_visit<T: DeserializeOwned>(
79		&self,
80		start: Bound<&EncodedKey>,
81		end: Bound<&EncodedKey>,
82		visit: &mut dyn FnMut(EncodedKey, T) -> Result<()>,
83	) -> Result<()> {
84		for (k, v) in self.range::<T>(start, end)? {
85			visit(k, v)?;
86		}
87		Ok(())
88	}
89
90	fn scan_prefix_visit<T: DeserializeOwned>(
91		&self,
92		prefix: &EncodedKey,
93		visit: &mut dyn FnMut(EncodedKey, T) -> Result<()>,
94	) -> Result<()> {
95		for (k, v) in self.scan_prefix::<T>(prefix)? {
96			visit(k, v)?;
97		}
98		Ok(())
99	}
100}
101
102pub trait InternalStateApi {
103	fn get<T: DeserializeOwned>(&self, key: &EncodedKey) -> Result<Option<T>>;
104	fn get_many<T: DeserializeOwned>(&self, keys: &[EncodedKey]) -> Result<Vec<(EncodedKey, T)>>;
105	fn set<T: Serialize>(&mut self, key: &EncodedKey, value: &T) -> Result<()>;
106	fn remove(&mut self, key: &EncodedKey) -> Result<()>;
107	fn drop(&mut self, key: &EncodedKey) -> Result<()>;
108	fn contains(&self, key: &EncodedKey) -> Result<bool>;
109	fn range<T: DeserializeOwned>(
110		&self,
111		start: Bound<&EncodedKey>,
112		end: Bound<&EncodedKey>,
113	) -> Result<Vec<(EncodedKey, T)>>;
114
115	fn get_many_visit<T: DeserializeOwned>(
116		&self,
117		keys: &[EncodedKey],
118		visit: &mut dyn FnMut(EncodedKey, T) -> Result<()>,
119	) -> Result<()> {
120		for (k, v) in self.get_many::<T>(keys)? {
121			visit(k, v)?;
122		}
123		Ok(())
124	}
125
126	fn range_visit<T: DeserializeOwned>(
127		&self,
128		start: Bound<&EncodedKey>,
129		end: Bound<&EncodedKey>,
130		visit: &mut dyn FnMut(EncodedKey, T) -> Result<()>,
131	) -> Result<()> {
132		for (k, v) in self.range::<T>(start, end)? {
133			visit(k, v)?;
134		}
135		Ok(())
136	}
137}
138
139pub trait StoreApi {
140	fn get(&self, key: &EncodedKey) -> Result<Option<EncodedRow>>;
141	fn contains(&self, key: &EncodedKey) -> Result<bool>;
142	fn prefix(&self, prefix: &EncodedKey) -> Result<Vec<(EncodedKey, EncodedRow)>>;
143	fn range(&self, start: Bound<&EncodedKey>, end: Bound<&EncodedKey>) -> Result<Vec<(EncodedKey, EncodedRow)>>;
144
145	fn range_visit(
146		&self,
147		start: Bound<&EncodedKey>,
148		end: Bound<&EncodedKey>,
149		visit: &mut dyn FnMut(EncodedKey, EncodedRow) -> Result<()>,
150	) -> Result<()> {
151		for (k, v) in self.range(start, end)? {
152			visit(k, v)?;
153		}
154		Ok(())
155	}
156
157	fn prefix_visit(
158		&self,
159		prefix: &EncodedKey,
160		visit: &mut dyn FnMut(EncodedKey, EncodedRow) -> Result<()>,
161	) -> Result<()> {
162		for (k, v) in self.prefix(prefix)? {
163			visit(k, v)?;
164		}
165		Ok(())
166	}
167}
168
169pub trait CatalogApi {
170	fn find_namespace(&self, namespace: NamespaceId, version: CommitVersion) -> Result<Option<Namespace>>;
171	fn find_namespace_by_name(&self, namespace: &str, version: CommitVersion) -> Result<Option<Namespace>>;
172	fn find_table(&self, table: TableId, version: CommitVersion) -> Result<Option<Table>>;
173	fn find_table_by_name(
174		&self,
175		namespace: NamespaceId,
176		name: &str,
177		version: CommitVersion,
178	) -> Result<Option<Table>>;
179	fn find_row_shape(&self, fingerprint: RowShapeFingerprint) -> Result<Option<RowShape>>;
180}
181
182pub trait DictionaryApi {
183	fn id_by_name(&mut self, name: &str) -> Result<Option<DictionaryId>>;
184	fn find(&mut self, dictionary: DictionaryId, value: &Value) -> Result<Option<DictionaryEntryId>>;
185	fn get(&mut self, dictionary: DictionaryId, id: DictionaryEntryId) -> Result<Option<Value>>;
186}
187
188pub trait OperatorContext {
189	type InsertEmit<'a>: RowEmit
190	where
191		Self: 'a;
192	type UpdateEmit<'a>: UpdateEmit
193	where
194		Self: 'a;
195	type RemoveEmit<'a>: RowEmit
196	where
197		Self: 'a;
198
199	fn operator_id(&self) -> FlowNodeId;
200	fn clock_now_nanos(&self) -> u64;
201	fn state(&mut self) -> impl StateApi + '_;
202	fn internal_state(&mut self) -> impl InternalStateApi + '_;
203	fn store(&mut self) -> impl StoreApi + '_;
204	fn catalog(&mut self) -> impl CatalogApi + '_;
205	fn dictionary(&mut self) -> impl DictionaryApi + '_;
206	fn get_or_create_row_number(&mut self, key: &EncodedKey) -> Result<(RowNumber, bool)>;
207	fn get_or_create_row_numbers(&mut self, keys: &[EncodedKey]) -> Result<Vec<(RowNumber, bool)>>;
208	/// Reserve `count` fresh, globally-unique output row numbers for this operator and return the start
209	/// of the `[start, start + count)` range. Backed by the host's process-shared in-memory allocator so
210	/// it is immune to the committing transaction's MVCC snapshot (unlike a counter read from the store).
211	fn allocate_row_numbers(&mut self, count: u64) -> Result<RowNumber>;
212	fn shape_for_row(&mut self, row: &EncodedRow) -> Result<RowShape>;
213
214	fn insert_emit<R: Row>(&mut self, row_capacity: usize) -> Result<Self::InsertEmit<'_>>;
215	fn update_emit<R: Row>(&mut self, row_capacity: usize) -> Result<Self::UpdateEmit<'_>>;
216	fn remove_emit<R: Row>(&mut self, row_capacity: usize) -> Result<Self::RemoveEmit<'_>>;
217
218	fn emit_insert<R: Row>(&mut self, rows: &[R], row_numbers: &[RowNumber]) -> Result<()> {
219		if rows.is_empty() {
220			return Ok(());
221		}
222		let mut emit = self.insert_emit::<R>(rows.len())?;
223		for row in rows {
224			row.encode_into(emit.sink())?;
225		}
226		emit.finish(row_numbers)
227	}
228
229	fn emit_update<R: Row>(&mut self, pre: &[R], post: &[R], row_numbers: &[RowNumber]) -> Result<()> {
230		if row_numbers.is_empty() {
231			return Ok(());
232		}
233		let mut emit = self.update_emit::<R>(row_numbers.len())?;
234		for row in pre {
235			row.encode_into(emit.pre())?;
236		}
237		for row in post {
238			row.encode_into(emit.post())?;
239		}
240		emit.finish(row_numbers)
241	}
242
243	fn emit_remove<R: Row>(&mut self, rows: &[R], row_numbers: &[RowNumber]) -> Result<()> {
244		if rows.is_empty() {
245			return Ok(());
246		}
247		let mut emit = self.remove_emit::<R>(rows.len())?;
248		for row in rows {
249			row.encode_into(emit.sink())?;
250		}
251		emit.finish(row_numbers)
252	}
253}
reifydb_sdk/operator/context/mod.rs

reifydb_sdk/operator/context/
mod.rs
