reifydb_sub_flow/operator/scan/
table.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, table::Table},
8		change::{Change, Diff},
9	},
10	value::column::columns::Columns,
11};
12use reifydb_value::Result;
13
14use crate::{Operator, operator::sink::decode_dictionary_columns, transaction::FlowTransaction};
15
16pub struct PrimitiveTableOperator {
17	node: FlowNodeId,
18	table: Table,
19}
20
21impl PrimitiveTableOperator {
22	pub fn new(node: FlowNodeId, table: Table) -> Self {
23		Self {
24			node,
25			table,
26		}
27	}
28}
29
30impl Operator for PrimitiveTableOperator {
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 PrimitiveTableOperator {
77	pub fn output_schema(&self) -> Columns {
78		Columns::from_catalog_columns(&self.table.columns)
79	}
80}
reifydb_sub_flow/operator/scan/table.rs

reifydb_sub_flow/operator/scan/
table.rs
