reifydb_core/flow/
dag.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{ops::Deref, sync::Arc};
5
6use reifydb_value::{Result, error::Error};
7
8use super::{
9	graph::DirectedGraph,
10	operator::{FlowNode, OperatorDef},
11};
12use crate::{
13	interface::catalog::{
14		flow::{FlowEdge, FlowId, OperatorId},
15		id::ViewId,
16	},
17	internal,
18};
19
20#[derive(Debug, Clone, PartialEq)]
21pub struct FlowDag {
22	inner: Arc<Inner>,
23}
24
25#[derive(Debug, PartialEq)]
26pub struct Inner {
27	pub id: FlowId,
28	pub graph: DirectedGraph<FlowNode>,
29	pub order: Vec<OperatorId>,
30	pub ephemeral: bool,
31}
32
33impl Deref for FlowDag {
34	type Target = Inner;
35
36	fn deref(&self) -> &Self::Target {
37		&self.inner
38	}
39}
40
41#[derive(Debug)]
42pub struct FlowBuilder {
43	id: FlowId,
44	graph: DirectedGraph<FlowNode>,
45	ephemeral: bool,
46}
47
48impl FlowBuilder {
49	pub fn new(id: impl Into<FlowId>) -> Self {
50		Self {
51			id: id.into(),
52			graph: DirectedGraph::new(),
53			ephemeral: false,
54		}
55	}
56
57	pub fn id(&self) -> FlowId {
58		self.id
59	}
60
61	pub fn ephemeral(&mut self) -> &mut Self {
62		self.ephemeral = true;
63		self
64	}
65
66	pub fn add_node(&mut self, node: FlowNode) -> OperatorId {
67		let node_id = node.id;
68		self.graph.add_node(node_id, node);
69		node_id
70	}
71
72	pub fn add_edge(&mut self, edge: FlowEdge) -> Result<()> {
73		let source = edge.source;
74		let target = edge.target;
75
76		if self.graph.get_node(&source).is_none() {
77			return Err(Error(Box::new(internal!(
78				"Flow edge references missing source node {:?}",
79				source
80			))));
81		}
82		if self.graph.get_node(&target).is_none() {
83			return Err(Error(Box::new(internal!(
84				"Flow edge references missing target node {:?}",
85				target
86			))));
87		}
88
89		self.graph.add_edge(edge);
90
91		if let Some(from_node) = self.graph.get_node_mut(&source) {
92			from_node.outputs.push(target);
93		}
94
95		if let Some(to_node) = self.graph.get_node_mut(&target) {
96			to_node.inputs.push(source);
97		}
98
99		Ok(())
100	}
101
102	pub fn build(self) -> FlowDag {
103		let order = self.graph.topological_sort();
104		FlowDag {
105			inner: Arc::new(Inner {
106				id: self.id,
107				graph: self.graph,
108				order,
109				ephemeral: self.ephemeral,
110			}),
111		}
112	}
113}
114
115impl FlowDag {
116	pub fn builder(id: impl Into<FlowId>) -> FlowBuilder {
117		FlowBuilder::new(id)
118	}
119
120	pub fn id(&self) -> FlowId {
121		self.inner.id
122	}
123
124	pub fn topological_order(&self) -> &[OperatorId] {
125		&self.inner.order
126	}
127
128	pub fn get_operator(&self, node_id: &OperatorId) -> Option<&FlowNode> {
129		self.inner.graph.get_node(node_id)
130	}
131
132	pub fn get_operator_ids(&self) -> impl Iterator<Item = OperatorId> + '_ {
133		self.inner.graph.nodes().map(|e| *e.0)
134	}
135
136	pub fn sink_views(&self) -> impl Iterator<Item = ViewId> + '_ {
137		self.get_operator_ids().filter_map(|id| match self.get_operator(&id).map(|operator| &operator.ty) {
138			Some(OperatorDef::SinkTableView {
139				view,
140				..
141			})
142			| Some(OperatorDef::SinkRingBufferView {
143				view,
144				..
145			})
146			| Some(OperatorDef::SinkSeriesView {
147				view,
148				..
149			}) => Some(*view),
150			_ => None,
151		})
152	}
153
154	pub fn node_count(&self) -> usize {
155		self.inner.graph.node_count()
156	}
157
158	pub fn edge_count(&self) -> usize {
159		self.inner.graph.edge_count()
160	}
161
162	pub fn ticks(&self) -> bool {
163		self.get_operator_ids().any(|id| self.get_operator(&id).is_some_and(|n| n.ty.ticks()))
164	}
165
166	pub fn has_timed_source(&self) -> bool {
167		self.get_operator_ids().any(|id| self.get_operator(&id).is_some_and(|n| n.ty.declares_time()))
168	}
169}
reifydb_core/flow/dag.rs

reifydb_core/flow/
dag.rs
