laminar_core/mv/registry/
mod.rs

1//! Materialized view registry with dependency tracking.
2//!
3//! The registry maintains a directed acyclic graph (DAG) of materialized views,
4//! tracking dependencies between views and ensuring correct processing order.
5
6use super::error::{MvError, MvState};
7use arrow_schema::SchemaRef;
8use rustc_hash::{FxHashMap, FxHashSet};
9use std::collections::VecDeque;
10
11/// Materialized view definition.
12///
13/// A materialized view is a query result that is stored and incrementally
14/// maintained as its source data changes.
15#[derive(Debug, Clone)]
16pub struct MaterializedView {
17    /// Unique view name.
18    pub name: String,
19    /// SQL definition (for reference and introspection).
20    pub sql: String,
21    /// Input sources (base tables or other MVs).
22    pub sources: Vec<String>,
23    /// Output schema of the view.
24    pub schema: SchemaRef,
25    /// Associated operator ID for event routing.
26    pub operator_id: String,
27    /// Current execution state.
28    pub state: MvState,
29}
30
31impl MaterializedView {
32    /// Creates a new materialized view definition.
33    #[must_use]
34    pub fn new(
35        name: impl Into<String>,
36        sql: impl Into<String>,
37        sources: Vec<String>,
38        schema: SchemaRef,
39    ) -> Self {
40        let name = name.into();
41        let operator_id = format!("mv_{name}");
42        Self {
43            name,
44            sql: sql.into(),
45            sources,
46            schema,
47            operator_id,
48            state: MvState::Running,
49        }
50    }
51
52    /// Creates a simple view with no schema (for testing).
53    #[cfg(test)]
54    pub fn simple(name: impl Into<String>, sources: Vec<String>) -> Self {
55        use arrow_schema::{DataType, Field, Schema};
56        use std::sync::Arc;
57
58        let schema = Arc::new(Schema::new(vec![Field::new(
59            "value",
60            DataType::Int64,
61            false,
62        )]));
63        Self::new(name, "", sources, schema)
64    }
65
66    /// Returns true if this view depends on the given source.
67    #[must_use]
68    pub fn depends_on(&self, source: &str) -> bool {
69        self.sources.iter().any(|s| s == source)
70    }
71}
72
73/// Registry for managing materialized views: stores the dependency DAG,
74/// detects cycles on registration, and exposes a topological order for
75/// correct processing.
76#[derive(Debug, Default)]
77pub struct MvRegistry {
78    /// All registered MVs by name.
79    views: FxHashMap<String, MaterializedView>,
80    /// Base tables (sources that are not MVs).
81    base_tables: FxHashSet<String>,
82    /// Dependency graph: MV name -> MVs that depend on it.
83    dependents: FxHashMap<String, FxHashSet<String>>,
84    /// Reverse dependency graph: MV name -> MVs it depends on.
85    dependencies: FxHashMap<String, FxHashSet<String>>,
86    /// Topological order for processing (dependencies first).
87    topo_order: Vec<String>,
88}
89
90impl MvRegistry {
91    /// Creates an empty registry.
92    #[must_use]
93    pub fn new() -> Self {
94        Self::default()
95    }
96
97    /// Registers a base table (source that is not an MV).
98    ///
99    /// Base tables are assumed to exist and can be referenced as sources
100    /// by materialized views.
101    pub fn register_base_table(&mut self, name: impl Into<String>) {
102        self.base_tables.insert(name.into());
103    }
104
105    /// Remove a base-table identity after its dependents have been dropped.
106    pub fn unregister_base_table(&mut self, name: &str) -> bool {
107        if self
108            .dependents
109            .get(name)
110            .is_some_and(|dependents| !dependents.is_empty())
111        {
112            return false;
113        }
114        self.dependents.remove(name);
115        self.base_tables.remove(name)
116    }
117
118    /// Returns true if the given name is a registered base table.
119    #[must_use]
120    pub fn is_base_table(&self, name: &str) -> bool {
121        self.base_tables.contains(name)
122    }
123
124    /// Registers a new materialized view.
125    ///
126    /// # Errors
127    ///
128    /// Returns error if:
129    /// - View name already exists
130    /// - Source MV or base table doesn't exist
131    /// - Would create a dependency cycle
132    pub fn register(&mut self, view: MaterializedView) -> Result<(), MvError> {
133        // Check for duplicate name
134        if self.views.contains_key(&view.name) {
135            return Err(MvError::DuplicateName(view.name.clone()));
136        }
137
138        // Validate sources exist
139        for source in &view.sources {
140            if !self.views.contains_key(source) && !self.is_base_table(source) {
141                return Err(MvError::SourceNotFound(source.clone()));
142            }
143        }
144
145        // Check for cycles
146        if self.would_create_cycle(&view.name, &view.sources) {
147            return Err(MvError::CycleDetected(view.name.clone()));
148        }
149
150        // Update dependency graphs
151        for source in &view.sources {
152            self.dependents
153                .entry(source.clone())
154                .or_default()
155                .insert(view.name.clone());
156            self.dependencies
157                .entry(view.name.clone())
158                .or_default()
159                .insert(source.clone());
160        }
161
162        self.views.insert(view.name.clone(), view);
163        self.update_topo_order();
164
165        Ok(())
166    }
167
168    /// Unregisters a materialized view.
169    ///
170    /// # Errors
171    ///
172    /// Returns error if:
173    /// - View doesn't exist
174    /// - Other views depend on it (use `unregister_cascade` instead)
175    pub fn unregister(&mut self, name: &str) -> Result<MaterializedView, MvError> {
176        // Check if view exists
177        if !self.views.contains_key(name) {
178            return Err(MvError::ViewNotFound(name.to_string()));
179        }
180
181        // Check for dependents
182        if let Some(deps) = self.dependents.get(name) {
183            if !deps.is_empty() {
184                let dep_names: Vec<_> = deps.iter().cloned().collect();
185                return Err(MvError::HasDependents(name.to_string(), dep_names));
186            }
187        }
188
189        self.remove_view(name)
190    }
191
192    /// Unregisters a materialized view and all views that depend on it.
193    ///
194    /// Returns the views that were removed, in dependency order (dependents first).
195    ///
196    /// # Errors
197    ///
198    /// Returns error if the view doesn't exist.
199    pub fn unregister_cascade(&mut self, name: &str) -> Result<Vec<MaterializedView>, MvError> {
200        if !self.views.contains_key(name) {
201            return Err(MvError::ViewNotFound(name.to_string()));
202        }
203
204        // Collect all views to remove in dependency order (dependents first)
205        let mut to_remove = Vec::new();
206        self.collect_dependents_recursive(name, &mut to_remove);
207        to_remove.push(name.to_string());
208
209        // Remove in collected order (dependents first, then the view itself)
210        let mut removed = Vec::with_capacity(to_remove.len());
211        for view_name in to_remove {
212            if let Ok(view) = self.remove_view(&view_name) {
213                removed.push(view);
214            }
215        }
216
217        Ok(removed)
218    }
219
220    fn collect_dependents_recursive(&self, name: &str, result: &mut Vec<String>) {
221        if let Some(deps) = self.dependents.get(name) {
222            for dep in deps {
223                if !result.contains(dep) {
224                    self.collect_dependents_recursive(dep, result);
225                    result.push(dep.clone());
226                }
227            }
228        }
229    }
230
231    fn remove_view(&mut self, name: &str) -> Result<MaterializedView, MvError> {
232        let view = self
233            .views
234            .remove(name)
235            .ok_or_else(|| MvError::ViewNotFound(name.to_string()))?;
236
237        // Remove from dependency tracking
238        if let Some(sources) = self.dependencies.remove(name) {
239            for source in sources {
240                if let Some(deps) = self.dependents.get_mut(&source) {
241                    deps.remove(name);
242                }
243            }
244        }
245        self.dependents.remove(name);
246
247        // Update topological order
248        self.update_topo_order();
249
250        Ok(view)
251    }
252
253    /// Gets a view by name.
254    #[must_use]
255    pub fn get(&self, name: &str) -> Option<&MaterializedView> {
256        self.views.get(name)
257    }
258
259    /// Gets a mutable reference to a view by name.
260    #[must_use]
261    pub fn get_mut(&mut self, name: &str) -> Option<&mut MaterializedView> {
262        self.views.get_mut(name)
263    }
264
265    /// Returns all views in topological order (dependencies first).
266    #[must_use]
267    pub fn topo_order(&self) -> &[String] {
268        &self.topo_order
269    }
270
271    /// Returns all views that depend on the given source.
272    pub fn get_dependents(&self, source: &str) -> impl Iterator<Item = &str> {
273        self.dependents
274            .get(source)
275            .into_iter()
276            .flatten()
277            .map(String::as_str)
278    }
279
280    /// Returns all sources that the given view depends on.
281    pub fn get_dependencies(&self, view: &str) -> impl Iterator<Item = &str> {
282        self.dependencies
283            .get(view)
284            .into_iter()
285            .flatten()
286            .map(String::as_str)
287    }
288
289    /// Returns the number of registered views.
290    #[must_use]
291    pub fn len(&self) -> usize {
292        self.views.len()
293    }
294
295    /// Returns true if no views are registered.
296    #[must_use]
297    pub fn is_empty(&self) -> bool {
298        self.views.is_empty()
299    }
300
301    /// Returns an iterator over all registered views.
302    pub fn views(&self) -> impl Iterator<Item = &MaterializedView> {
303        self.views.values()
304    }
305
306    /// Returns the set of registered base tables.
307    #[must_use]
308    pub fn base_tables(&self) -> &FxHashSet<String> {
309        &self.base_tables
310    }
311
312    /// Returns the full dependency chain for a view (including transitive).
313    ///
314    /// The chain is returned in topological order (dependencies first).
315    #[must_use]
316    pub fn dependency_chain(&self, name: &str) -> Vec<String> {
317        let mut chain = Vec::new();
318        let mut visited = FxHashSet::default();
319        self.collect_dependencies_recursive(name, &mut chain, &mut visited);
320        chain
321    }
322
323    fn collect_dependencies_recursive(
324        &self,
325        name: &str,
326        result: &mut Vec<String>,
327        visited: &mut FxHashSet<String>,
328    ) {
329        if !visited.insert(name.to_string()) {
330            return;
331        }
332
333        if let Some(deps) = self.dependencies.get(name) {
334            for dep in deps {
335                self.collect_dependencies_recursive(dep, result, visited);
336            }
337        }
338
339        // Only add MVs, not base tables
340        if self.views.contains_key(name) {
341            result.push(name.to_string());
342        }
343    }
344
345    fn would_create_cycle(&self, new_name: &str, sources: &[String]) -> bool {
346        // DFS to check if any source transitively depends on new_name
347        let mut visited = FxHashSet::default();
348        let mut stack: Vec<_> = sources.to_vec();
349
350        while let Some(current) = stack.pop() {
351            if current == new_name {
352                return true;
353            }
354            if visited.insert(current.clone()) {
355                if let Some(deps) = self.dependencies.get(&current) {
356                    stack.extend(deps.iter().cloned());
357                }
358            }
359        }
360
361        false
362    }
363
364    fn update_topo_order(&mut self) {
365        // Kahn's algorithm for topological sort
366        let mut in_degree: FxHashMap<String, usize> = FxHashMap::default();
367        let mut queue: VecDeque<String> = VecDeque::new();
368
369        // Initialize in-degrees (count only MV dependencies, not base tables)
370        for name in self.views.keys() {
371            let deps = self.dependencies.get(name).map_or(0, |d| {
372                d.iter().filter(|dep| self.views.contains_key(*dep)).count()
373            });
374            in_degree.insert(name.clone(), deps);
375            if deps == 0 {
376                queue.push_back(name.clone());
377            }
378        }
379
380        // Process
381        self.topo_order.clear();
382        while let Some(name) = queue.pop_front() {
383            self.topo_order.push(name.clone());
384
385            if let Some(dependents) = self.dependents.get(&name) {
386                for dep in dependents {
387                    if let Some(count) = in_degree.get_mut(dep) {
388                        *count = count.saturating_sub(1);
389                        if *count == 0 {
390                            queue.push_back(dep.clone());
391                        }
392                    }
393                }
394            }
395        }
396    }
397}
398
399#[cfg(test)]
400mod tests;
laminar_core/mv/registry/mod.rs

laminar_core/mv/registry/
mod.rs
