reddb_server/runtime/
continuous_materialized_view.rs

1//! Continuous (materialized) view descriptor + catalog persistence
2//! (issue #593, slice 9a of #575).
3//!
4//! Today the view registry on `RuntimeInner` is purely in-memory:
5//! every `CREATE [MATERIALIZED] VIEW` lands in
6//! `inner.views: RwLock<HashMap<String, Arc<CreateViewQuery>>>` and a
7//! restart loses every definition. This module introduces the
8//! descriptor type and the persistence layer that backs the registry
9//! onto the system collection [`CATALOG_COLLECTION`]. Read / write /
10//! refresh code paths are unchanged — the rehydrate hook at boot
11//! repopulates the in-memory registry from the persisted rows before
12//! the API opens.
13
14use crate::api::{RedDBError, RedDBResult};
15use crate::storage::unified::entity::{EntityData, EntityId, EntityKind, RowData, UnifiedEntity};
16use reddb_types::Value;
17
18/// Name of the system collection that stores one row per
19/// materialized-view definition. Bootstrapped alongside the other
20/// keyed system collections (`red.config`, `red.vault`) at boot.
21pub const CATALOG_COLLECTION: &str = "red.materialized_view.definitions";
22
23/// Persisted shape of a single `CREATE MATERIALIZED VIEW`.
24///
25/// The descriptor stores the original SQL source so the body AST can
26/// be recovered by re-parsing at boot — this avoids embedding a
27/// version-dependent AST serialization in the on-disk catalog, and
28/// keeps the rehydrate path symmetric with the user-facing CREATE.
29#[derive(Debug, Clone, PartialEq, Eq)]
30pub struct MaterializedViewDescriptor {
31    /// View name as declared in `CREATE MATERIALIZED VIEW <name>`.
32    pub name: String,
33    /// Verbatim SQL source of the `CREATE MATERIALIZED VIEW`
34    /// statement. The rehydrate path re-parses this string to recover
35    /// the body AST.
36    pub source_sql: String,
37    /// Source collections referenced by the view body — populated
38    /// from `collect_table_refs(&q.query)` at creation time.
39    pub source_collections: Vec<String>,
40    /// `REFRESH EVERY <duration>` clause in milliseconds, or `None`
41    /// for refresh-on-demand views.
42    pub refresh_every_ms: Option<u64>,
43    /// `WITH RETENTION <duration>` clause in milliseconds, or `None`
44    /// when no retention policy was declared on the view.
45    pub retention_duration_ms: Option<u64>,
46}
47
48impl MaterializedViewDescriptor {
49    /// Build a row entity suitable for `insert_auto(CATALOG_COLLECTION, …)`.
50    /// Each descriptor field maps to one named column on the row, in
51    /// the same shape `red_config` uses for its key/value entries —
52    /// keeps the storage layout introspectable from SQL without a
53    /// JSON parser.
54    fn to_row_entity(&self) -> UnifiedEntity {
55        let mut named = std::collections::HashMap::new();
56        named.insert("name".to_string(), Value::text(self.name.clone()));
57        named.insert(
58            "source_sql".to_string(),
59            Value::text(self.source_sql.clone()),
60        );
61        named.insert(
62            "source_collections".to_string(),
63            Value::Array(
64                self.source_collections
65                    .iter()
66                    .map(|s| Value::text(s.clone()))
67                    .collect(),
68            ),
69        );
70        named.insert(
71            "refresh_every_ms".to_string(),
72            match self.refresh_every_ms {
73                Some(ms) => Value::UnsignedInteger(ms),
74                None => Value::Null,
75            },
76        );
77        named.insert(
78            "retention_duration_ms".to_string(),
79            match self.retention_duration_ms {
80                Some(ms) => Value::UnsignedInteger(ms),
81                None => Value::Null,
82            },
83        );
84        UnifiedEntity::new(
85            EntityId::new(0),
86            EntityKind::TableRow {
87                table: std::sync::Arc::from(CATALOG_COLLECTION),
88                row_id: 0,
89            },
90            EntityData::Row(RowData {
91                columns: Vec::new(),
92                named: Some(named),
93                schema: None,
94            }),
95        )
96    }
97}
98
99/// Decode a stored row back into a descriptor. Returns `None` for
100/// rows that are missing the required `name` / `source_sql` columns
101/// — boot-time rehydrate logs and continues so a single malformed
102/// entry does not block startup.
103#[cfg(test)]
104fn decode_row(entity: &UnifiedEntity) -> Option<MaterializedViewDescriptor> {
105    decode_row_with_schema(entity, None)
106}
107
108fn decode_row_with_schema(
109    entity: &UnifiedEntity,
110    schema: Option<&Vec<String>>,
111) -> Option<MaterializedViewDescriptor> {
112    let EntityData::Row(row) = &entity.data else {
113        return None;
114    };
115    let name = match field(row, schema, "name")? {
116        Value::Text(s) => s.to_string(),
117        _ => return None,
118    };
119    let source_sql = match field(row, schema, "source_sql")? {
120        Value::Text(s) => s.to_string(),
121        _ => return None,
122    };
123    let source_collections = match field(row, schema, "source_collections") {
124        Some(Value::Array(values)) => values
125            .iter()
126            .filter_map(|v| match v {
127                Value::Text(s) => Some(s.to_string()),
128                _ => None,
129            })
130            .collect(),
131        _ => Vec::new(),
132    };
133    let refresh_every_ms = match field(row, schema, "refresh_every_ms") {
134        Some(Value::UnsignedInteger(ms)) => Some(*ms),
135        Some(Value::Integer(ms)) if *ms >= 0 => Some(*ms as u64),
136        _ => None,
137    };
138    let retention_duration_ms = match field(row, schema, "retention_duration_ms") {
139        Some(Value::UnsignedInteger(ms)) => Some(*ms),
140        Some(Value::Integer(ms)) if *ms >= 0 => Some(*ms as u64),
141        _ => None,
142    };
143    Some(MaterializedViewDescriptor {
144        name,
145        source_sql,
146        source_collections,
147        refresh_every_ms,
148        retention_duration_ms,
149    })
150}
151
152/// Persist a descriptor to the catalog collection, replacing any
153/// prior rows with the same `name`. Idempotent: re-persisting an
154/// existing view (e.g. `CREATE OR REPLACE`) leaves exactly one row
155/// behind, so the catalog never accumulates duplicates across
156/// repeated definition churn.
157pub(crate) fn persist_descriptor(
158    store: &crate::storage::unified::UnifiedStore,
159    descriptor: &MaterializedViewDescriptor,
160) -> RedDBResult<()> {
161    let _ = store.get_or_create_collection(CATALOG_COLLECTION);
162    remove_by_name(store, &descriptor.name)?;
163    let entity = descriptor.to_row_entity();
164    store
165        .insert_auto(CATALOG_COLLECTION, entity)
166        .map_err(|err| {
167            RedDBError::Internal(format!(
168                "persist materialized-view descriptor {}: {err}",
169                descriptor.name
170            ))
171        })?;
172    Ok(())
173}
174
175/// Remove every row whose `name` column matches `name`. Used on
176/// `DROP MATERIALIZED VIEW` and as the first half of
177/// [`persist_descriptor`]'s upsert semantics.
178pub(crate) fn remove_by_name(
179    store: &crate::storage::unified::UnifiedStore,
180    name: &str,
181) -> RedDBResult<()> {
182    let Some(manager) = store.get_collection(CATALOG_COLLECTION) else {
183        return Ok(());
184    };
185    let mut to_delete: Vec<EntityId> = Vec::new();
186    let schema = manager.column_schema();
187    manager
188        .try_for_each_entity(usize::MAX, |entity| {
189            let EntityData::Row(row) = &entity.data else {
190                return true;
191            };
192            if let Some(Value::Text(stored)) = field(row, schema.as_deref(), "name") {
193                if stored.as_ref() == name {
194                    to_delete.push(entity.id);
195                }
196            }
197            true
198        })
199        .map_err(|err| {
200            RedDBError::Internal(format!("scan materialized-view descriptors: {err}"))
201        })?;
202    for id in to_delete {
203        store.delete(CATALOG_COLLECTION, id).map_err(|err| {
204            RedDBError::Internal(format!(
205                "delete materialized-view descriptor row for {name}: {err}"
206            ))
207        })?;
208    }
209    Ok(())
210}
211
212/// Read every persisted descriptor. Returns an empty vector when the
213/// catalog collection doesn't exist (fresh datadir / first boot).
214pub(crate) fn load_all(
215    store: &crate::storage::unified::UnifiedStore,
216) -> RedDBResult<Vec<MaterializedViewDescriptor>> {
217    let Some(manager) = store.get_collection(CATALOG_COLLECTION) else {
218        return Ok(Vec::new());
219    };
220    let schema = manager.column_schema();
221    let mut descriptors = Vec::new();
222    manager
223        .try_for_each_entity(usize::MAX, |entity| {
224            if let Some(descriptor) = decode_row_with_schema(entity, schema.as_deref()) {
225                descriptors.push(descriptor);
226            }
227            true
228        })
229        .map_err(|err| {
230            RedDBError::Internal(format!("load materialized-view descriptors: {err}"))
231        })?;
232    Ok(descriptors)
233}
234
235fn field<'a>(row: &'a RowData, schema: Option<&Vec<String>>, name: &str) -> Option<&'a Value> {
236    if row.named.is_some() || row.schema.is_some() {
237        return row.get_field(name);
238    }
239    schema
240        .and_then(|columns| columns.iter().position(|column| column == name))
241        .and_then(|index| row.columns.get(index))
242}
243
244#[cfg(test)]
245mod tests {
246    use super::*;
247
248    #[test]
249    fn row_entity_roundtrips_through_decode() {
250        let d = MaterializedViewDescriptor {
251            name: "v".into(),
252            source_sql: "CREATE MATERIALIZED VIEW v AS SELECT * FROM t".into(),
253            source_collections: vec!["t".into(), "u".into()],
254            refresh_every_ms: Some(60_000),
255            retention_duration_ms: Some(7 * 24 * 3_600_000),
256        };
257        let entity = d.to_row_entity();
258        let back = decode_row(&entity).expect("decode");
259        assert_eq!(d, back);
260    }
261
262    #[test]
263    fn null_options_decode_to_none() {
264        let d = MaterializedViewDescriptor {
265            name: "v".into(),
266            source_sql: "CREATE MATERIALIZED VIEW v AS SELECT 1".into(),
267            source_collections: vec![],
268            refresh_every_ms: None,
269            retention_duration_ms: None,
270        };
271        let entity = d.to_row_entity();
272        let back = decode_row(&entity).expect("decode");
273        assert!(back.refresh_every_ms.is_none());
274        assert!(back.retention_duration_ms.is_none());
275    }
276}
reddb_server/runtime/continuous_materialized_view.rs

reddb_server/runtime/
continuous_materialized_view.rs
