nodedb 0.2.1

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

// SPDX-License-Identifier: BUSL-1.1

//! In-memory registry of streaming materialized views.

use std::collections::HashMap;
use std::sync::RwLock;

use super::state::MvState;
use super::types::StreamingMvDef;

/// In-memory streaming MV registry.
pub struct MvRegistry {
    /// (tenant_id, mv_name) → definition.
    defs: RwLock<HashMap<(u64, String), StreamingMvDef>>,
    /// (tenant_id, mv_name) → live aggregate state.
    states: RwLock<HashMap<(u64, String), std::sync::Arc<MvState>>>,
}

impl MvRegistry {
    pub fn new() -> Self {
        Self {
            defs: RwLock::new(HashMap::new()),
            states: RwLock::new(HashMap::new()),
        }
    }

    /// Register a streaming MV and create its state.
    pub fn register(&self, def: StreamingMvDef) {
        let key = (def.tenant_id, def.name.clone());
        let state = std::sync::Arc::new(MvState::new(
            def.name.clone(),
            def.group_by_columns.clone(),
            def.aggregates.clone(),
        ));

        let mut defs = self.defs.write().unwrap_or_else(|p| p.into_inner());
        defs.insert(key.clone(), def);

        let mut states = self.states.write().unwrap_or_else(|p| p.into_inner());
        states.insert(key, state);
    }

    /// Unregister a streaming MV. Returns true if it existed.
    pub fn unregister(&self, tenant_id: u64, name: &str) -> bool {
        let key = (tenant_id, name.to_string());
        let mut defs = self.defs.write().unwrap_or_else(|p| p.into_inner());
        let existed = defs.remove(&key).is_some();

        let mut states = self.states.write().unwrap_or_else(|p| p.into_inner());
        states.remove(&key);

        existed
    }

    /// Get the definition of a streaming MV.
    pub fn get_def(&self, tenant_id: u64, name: &str) -> Option<StreamingMvDef> {
        let key = (tenant_id, name.to_string());
        let defs = self.defs.read().unwrap_or_else(|p| p.into_inner());
        defs.get(&key).cloned()
    }

    /// Get the live state of a streaming MV.
    pub fn get_state(&self, tenant_id: u64, name: &str) -> Option<std::sync::Arc<MvState>> {
        let key = (tenant_id, name.to_string());
        let states = self.states.read().unwrap_or_else(|p| p.into_inner());
        states.get(&key).cloned()
    }

    /// Find all MVs that source from a given stream.
    pub fn find_by_source(
        &self,
        tenant_id: u64,
        stream_name: &str,
    ) -> Vec<std::sync::Arc<MvState>> {
        let defs = self.defs.read().unwrap_or_else(|p| p.into_inner());
        let states = self.states.read().unwrap_or_else(|p| p.into_inner());

        defs.iter()
            .filter(|((tid, _), def)| *tid == tenant_id && def.source_stream == stream_name)
            .filter_map(|(key, _)| states.get(key).cloned())
            .collect()
    }

    /// Clear all entries and reload from catalog.
    /// Used by the recovery verifier repair path.
    pub fn clear_and_reload(
        &self,
        catalog: &crate::control::security::catalog::types::SystemCatalog,
    ) -> crate::Result<()> {
        let fresh = catalog.load_all_streaming_mvs()?;
        let mut defs = self.defs.write().unwrap_or_else(|p| p.into_inner());
        let mut states = self.states.write().unwrap_or_else(|p| p.into_inner());
        defs.clear();
        states.clear();
        for mv in fresh {
            let key = (mv.tenant_id, mv.name.clone());
            let state = std::sync::Arc::new(crate::event::streaming_mv::state::MvState::new(
                mv.name.clone(),
                mv.group_by_columns.clone(),
                mv.aggregates.clone(),
            ));
            defs.insert(key.clone(), mv);
            states.insert(key, state);
        }
        Ok(())
    }

    /// List all MV definitions (all tenants).
    pub fn list_all(&self) -> Vec<StreamingMvDef> {
        let defs = self.defs.read().unwrap_or_else(|p| p.into_inner());
        defs.values().cloned().collect()
    }

    /// List all MV definitions for a tenant.
    pub fn list_for_tenant(&self, tenant_id: u64) -> Vec<StreamingMvDef> {
        let defs = self.defs.read().unwrap_or_else(|p| p.into_inner());
        defs.values()
            .filter(|d| d.tenant_id == tenant_id)
            .cloned()
            .collect()
    }

    /// Load from catalog on startup.
    pub fn load_from_catalog(
        &self,
        catalog: &crate::control::security::catalog::types::SystemCatalog,
    ) {
        let mvs = match catalog.load_all_streaming_mvs() {
            Ok(m) => m,
            Err(e) => {
                tracing::warn!(error = %e, "failed to load streaming MVs from catalog");
                return;
            }
        };
        for mv in mvs {
            self.register(mv);
        }
    }
}

impl Default for MvRegistry {
    fn default() -> Self {
        Self::new()
    }
}