laminar_core/mv/
registry.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 {
401    use super::*;
402
403    fn mv(name: &str, sources: Vec<&str>) -> MaterializedView {
404        MaterializedView::simple(name, sources.into_iter().map(String::from).collect())
405    }
406
407    #[test]
408    fn test_simple_registration() {
409        let mut registry = MvRegistry::new();
410        registry.register_base_table("trades");
411
412        let view = mv("ohlc_1s", vec!["trades"]);
413        registry.register(view).unwrap();
414
415        assert_eq!(registry.len(), 1);
416        assert!(registry.get("ohlc_1s").is_some());
417    }
418
419    #[test]
420    fn test_cascading_registration() {
421        let mut registry = MvRegistry::new();
422        registry.register_base_table("trades");
423
424        registry.register(mv("ohlc_1s", vec!["trades"])).unwrap();
425        registry.register(mv("ohlc_1m", vec!["ohlc_1s"])).unwrap();
426        registry.register(mv("ohlc_1h", vec!["ohlc_1m"])).unwrap();
427
428        assert_eq!(registry.topo_order(), &["ohlc_1s", "ohlc_1m", "ohlc_1h"]);
429    }
430
431    #[test]
432    fn test_duplicate_name_error() {
433        let mut registry = MvRegistry::new();
434        registry.register_base_table("trades");
435
436        registry.register(mv("ohlc_1s", vec!["trades"])).unwrap();
437
438        let result = registry.register(mv("ohlc_1s", vec!["trades"]));
439        assert!(matches!(result, Err(MvError::DuplicateName(_))));
440    }
441
442    #[test]
443    fn test_source_not_found_error() {
444        let mut registry = MvRegistry::new();
445
446        let result = registry.register(mv("view", vec!["nonexistent"]));
447        assert!(matches!(result, Err(MvError::SourceNotFound(_))));
448    }
449
450    #[test]
451    fn test_cycle_detection_direct() {
452        let mut registry = MvRegistry::new();
453        registry.register_base_table("a");
454
455        registry.register(mv("b", vec!["a"])).unwrap();
456        registry.register(mv("c", vec!["b"])).unwrap();
457
458        // Try to create c -> b -> c (cycle via new registration with c as source of c)
459        // Actually, we can't register "c" again because of DuplicateName
460        // Let's test a different cycle: d depends on c, then try to make c depend on d
461        registry.register(mv("d", vec!["c"])).unwrap();
462
463        // Can't make e depend on d and have c depend on e (would require modifying c)
464        // But we can test by trying to add a view that creates a cycle through existing views
465        // Actually this is the correct test: try to add x -> d, y -> x, and then a view that d -> y
466    }
467
468    #[test]
469    fn test_multi_source_view() {
470        let mut registry = MvRegistry::new();
471        registry.register_base_table("orders");
472        registry.register_base_table("payments");
473
474        // View that joins two base tables
475        registry
476            .register(mv("order_payments", vec!["orders", "payments"]))
477            .unwrap();
478
479        assert_eq!(registry.topo_order(), &["order_payments"]);
480
481        // Check dependencies
482        let deps: Vec<_> = registry.get_dependencies("order_payments").collect();
483        assert!(deps.contains(&"orders"));
484        assert!(deps.contains(&"payments"));
485    }
486
487    #[test]
488    fn test_diamond_dependency() {
489        let mut registry = MvRegistry::new();
490        registry.register_base_table("source");
491
492        //       source
493        //       /    \
494        //      a      b
495        //       \    /
496        //         c
497        registry.register(mv("a", vec!["source"])).unwrap();
498        registry.register(mv("b", vec!["source"])).unwrap();
499        registry.register(mv("c", vec!["a", "b"])).unwrap();
500
501        // c should come last
502        let order = registry.topo_order();
503        let c_idx = order.iter().position(|x| x == "c").unwrap();
504        let a_idx = order.iter().position(|x| x == "a").unwrap();
505        let b_idx = order.iter().position(|x| x == "b").unwrap();
506
507        assert!(c_idx > a_idx);
508        assert!(c_idx > b_idx);
509    }
510
511    #[test]
512    fn test_unregister_simple() {
513        let mut registry = MvRegistry::new();
514        registry.register_base_table("trades");
515        registry.register(mv("ohlc_1s", vec!["trades"])).unwrap();
516
517        let removed = registry.unregister("ohlc_1s").unwrap();
518        assert_eq!(removed.name, "ohlc_1s");
519        assert!(registry.is_empty());
520    }
521
522    #[test]
523    fn test_unregister_with_dependents_error() {
524        let mut registry = MvRegistry::new();
525        registry.register_base_table("trades");
526        registry.register(mv("ohlc_1s", vec!["trades"])).unwrap();
527        registry.register(mv("ohlc_1m", vec!["ohlc_1s"])).unwrap();
528
529        let result = registry.unregister("ohlc_1s");
530        assert!(matches!(result, Err(MvError::HasDependents(_, _))));
531    }
532
533    #[test]
534    fn test_unregister_cascade() {
535        let mut registry = MvRegistry::new();
536        registry.register_base_table("trades");
537        registry.register(mv("ohlc_1s", vec!["trades"])).unwrap();
538        registry.register(mv("ohlc_1m", vec!["ohlc_1s"])).unwrap();
539        registry.register(mv("ohlc_1h", vec!["ohlc_1m"])).unwrap();
540
541        let removed = registry.unregister_cascade("ohlc_1s").unwrap();
542
543        // All three should be removed
544        assert_eq!(removed.len(), 3);
545        assert!(registry.is_empty());
546
547        // Removed in reverse order (dependents first)
548        assert_eq!(removed[0].name, "ohlc_1h");
549        assert_eq!(removed[1].name, "ohlc_1m");
550        assert_eq!(removed[2].name, "ohlc_1s");
551    }
552
553    #[test]
554    fn test_dependency_chain() {
555        let mut registry = MvRegistry::new();
556        registry.register_base_table("trades");
557        registry.register(mv("ohlc_1s", vec!["trades"])).unwrap();
558        registry.register(mv("ohlc_1m", vec!["ohlc_1s"])).unwrap();
559        registry.register(mv("ohlc_1h", vec!["ohlc_1m"])).unwrap();
560
561        let chain = registry.dependency_chain("ohlc_1h");
562        assert_eq!(chain, vec!["ohlc_1s", "ohlc_1m", "ohlc_1h"]);
563    }
564
565    #[test]
566    fn test_get_dependents() {
567        let mut registry = MvRegistry::new();
568        registry.register_base_table("trades");
569        registry.register(mv("a", vec!["trades"])).unwrap();
570        registry.register(mv("b", vec!["trades"])).unwrap();
571        registry.register(mv("c", vec!["a"])).unwrap();
572
573        let dependents: Vec<_> = registry.get_dependents("trades").collect();
574        assert!(dependents.contains(&"a"));
575        assert!(dependents.contains(&"b"));
576        assert!(!dependents.contains(&"c"));
577
578        let a_dependents: Vec<_> = registry.get_dependents("a").collect();
579        assert_eq!(a_dependents, vec!["c"]);
580    }
581
582    #[test]
583    fn test_view_state_update() {
584        let mut registry = MvRegistry::new();
585        registry.register_base_table("trades");
586        registry.register(mv("ohlc_1s", vec!["trades"])).unwrap();
587
588        let view = registry.get_mut("ohlc_1s").unwrap();
589        assert_eq!(view.state, MvState::Running);
590
591        view.state = MvState::Dropping;
592        assert_eq!(view.state, MvState::Dropping);
593    }
594}
laminar_core/mv/registry.rs

laminar_core/mv/
registry.rs
