reifydb_engine/vm/
flow_lineage.rs

1// SPDX-License-Identifier: Apache-2.0
2// Copyright (c) 2026 ReifyDB
3
4use std::{
5	collections::{BTreeMap, BTreeSet},
6	sync::Arc,
7};
8
9use reifydb_core::interface::catalog::{id::ViewId, object::ObjectId};
10use reifydb_runtime::sync::rwlock::RwLock;
11
12#[derive(Clone)]
13pub struct ViewLineage {
14	inner: Arc<RwLock<BTreeMap<ViewId, Arc<BTreeSet<ObjectId>>>>>,
15}
16
17impl Default for ViewLineage {
18	fn default() -> Self {
19		Self {
20			inner: Arc::new(RwLock::new(BTreeMap::new())),
21		}
22	}
23}
24
25impl ViewLineage {
26	pub fn publish(&self, map: BTreeMap<ViewId, BTreeSet<ObjectId>>) {
27		let map = map.into_iter().map(|(view, objects)| (view, Arc::new(objects))).collect();
28		*self.inner.write() = map;
29	}
30
31	pub fn upstream_of(&self, view: ViewId) -> Option<Arc<BTreeSet<ObjectId>>> {
32		self.inner.read().get(&view).cloned()
33	}
34}
35
36#[cfg(test)]
37mod tests {
38	use reifydb_core::interface::catalog::id::TableId;
39
40	use super::*;
41
42	#[test]
43	fn test_publish_replaces_and_upstream_of_looks_up() {
44		let lineage = ViewLineage::default();
45		assert!(lineage.upstream_of(ViewId(1)).is_none());
46
47		lineage.publish(BTreeMap::from([(ViewId(1), BTreeSet::from([ObjectId::Table(TableId(9))]))]));
48		assert_eq!(*lineage.upstream_of(ViewId(1)).unwrap(), BTreeSet::from([ObjectId::Table(TableId(9))]));
49
50		lineage.publish(BTreeMap::new());
51		assert!(lineage.upstream_of(ViewId(1)).is_none(), "publish must replace, not merge");
52	}
53}
reifydb_engine/vm/flow_lineage.rs

reifydb_engine/vm/
flow_lineage.rs
