reifydb_sub_flow/operator/scan/
ringbuffer.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_abi::operator::capabilities::OperatorCapability;
5use reifydb_core::{
6	interface::{
7		catalog::{flow::FlowNodeId, ringbuffer::RingBuffer},
8		change::{Change, Diff},
9	},
10	value::column::{ColumnWithName, buffer::ColumnBuffer, columns::Columns},
11};
12use reifydb_value::{Result, fragment::Fragment};
13
14use crate::{Operator, operator::sink::decode_dictionary_columns, transaction::FlowTransaction};
15
16pub struct PrimitiveRingBufferOperator {
17	node: FlowNodeId,
18	ringbuffer: RingBuffer,
19}
20
21impl PrimitiveRingBufferOperator {
22	pub fn new(node: FlowNodeId, ringbuffer: RingBuffer) -> Self {
23		Self {
24			node,
25			ringbuffer,
26		}
27	}
28}
29
30impl Operator for PrimitiveRingBufferOperator {
31	fn id(&self) -> FlowNodeId {
32		self.node
33	}
34
35	fn capabilities(&self) -> &[OperatorCapability] {
36		OperatorCapability::STANDARD
37	}
38
39	fn apply(&self, txn: &mut FlowTransaction, change: Change) -> Result<Change> {
40		let mut decoded_diffs = Vec::with_capacity(change.diffs.len());
41		for diff in change.diffs {
42			decoded_diffs.push(match diff {
43				Diff::Insert {
44					post,
45					..
46				} => {
47					let mut decoded = post;
48					decode_dictionary_columns(&mut decoded, txn)?;
49					Diff::insert(decoded)
50				}
51				Diff::Update {
52					pre,
53					post,
54					..
55				} => {
56					let mut decoded_pre = pre;
57					let mut decoded_post = post;
58					decode_dictionary_columns(&mut decoded_pre, txn)?;
59					decode_dictionary_columns(&mut decoded_post, txn)?;
60					Diff::update(decoded_pre, decoded_post)
61				}
62				Diff::Remove {
63					pre,
64					..
65				} => {
66					let mut decoded = pre;
67					decode_dictionary_columns(&mut decoded, txn)?;
68					Diff::remove(decoded)
69				}
70			});
71		}
72		Ok(Change::from_flow(self.node, change.version, decoded_diffs, change.changed_at))
73	}
74}
75
76impl PrimitiveRingBufferOperator {
77	pub fn output_schema(&self) -> Columns {
78		let columns: Vec<ColumnWithName> = self
79			.ringbuffer
80			.columns
81			.iter()
82			.map(|col| ColumnWithName {
83				name: Fragment::internal(&col.name),
84				data: ColumnBuffer::with_capacity(col.constraint.get_type(), 0),
85			})
86			.collect();
87		Columns::new(columns)
88	}
89}
reifydb_sub_flow/operator/scan/ringbuffer.rs

reifydb_sub_flow/operator/scan/
ringbuffer.rs
