reifydb_sub_flow/operator/scan/
series.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use reifydb_abi::operator::capabilities::OperatorCapability;
5use reifydb_core::interface::{
6	catalog::flow::FlowNodeId,
7	change::{Change, Diff},
8};
9use reifydb_value::Result;
10
11use crate::{Operator, operator::sink::decode_dictionary_columns, transaction::FlowTransaction};
12
13pub struct PrimitiveSeriesOperator {
14	node: FlowNodeId,
15}
16
17impl PrimitiveSeriesOperator {
18	pub fn new(node: FlowNodeId) -> Self {
19		Self {
20			node,
21		}
22	}
23}
24
25impl Operator for PrimitiveSeriesOperator {
26	fn id(&self) -> FlowNodeId {
27		self.node
28	}
29
30	fn capabilities(&self) -> &[OperatorCapability] {
31		OperatorCapability::STANDARD
32	}
33
34	fn apply(&self, txn: &mut FlowTransaction, change: Change) -> Result<Change> {
35		let mut decoded_diffs = Vec::with_capacity(change.diffs.len());
36		for diff in change.diffs {
37			decoded_diffs.push(match diff {
38				Diff::Insert {
39					post,
40					..
41				} => {
42					let mut decoded = post;
43					decode_dictionary_columns(&mut decoded, txn)?;
44					Diff::insert(decoded)
45				}
46				Diff::Update {
47					pre,
48					post,
49					..
50				} => {
51					let mut decoded_pre = pre;
52					let mut decoded_post = post;
53					decode_dictionary_columns(&mut decoded_pre, txn)?;
54					decode_dictionary_columns(&mut decoded_post, txn)?;
55					Diff::update(decoded_pre, decoded_post)
56				}
57				Diff::Remove {
58					pre,
59					..
60				} => {
61					let mut decoded = pre;
62					decode_dictionary_columns(&mut decoded, txn)?;
63					Diff::remove(decoded)
64				}
65			});
66		}
67		Ok(Change::from_flow(self.node, change.version, decoded_diffs, change.changed_at))
68	}
69}
reifydb_sub_flow/operator/scan/series.rs

reifydb_sub_flow/operator/scan/
series.rs
