arete_server/
materialized_view.rs

1//! Materialized view evaluation for view pipelines.
2//!
3//! This module handles the runtime evaluation of ViewDef pipelines,
4//! maintaining materialized results that update as source data changes.
5
6use crate::cache::EntityCache;
7use crate::shared_entity::{lookup, EntityFields};
8use serde::Serialize;
9use serde_json::Value;
10use std::borrow::Cow;
11use std::collections::{HashMap, HashSet};
12use std::sync::Arc;
13use tokio::sync::RwLock;
14
15/// Result of evaluating whether an update affects a materialized view
16#[derive(Debug, Clone, PartialEq)]
17pub enum ViewEffect {
18    /// Update does not affect the view result
19    NoEffect,
20    /// Entity should be added to the view result
21    Add { key: String },
22    /// Entity should be removed from the view result
23    Remove { key: String },
24    /// Entity in view was updated
25    Update { key: String },
26    /// Entity replaces another in the view (for single-result views)
27    Replace { old_key: String, new_key: String },
28}
29
30/// Sort order for view evaluation
31#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
32pub enum SortOrder {
33    Asc,
34    Desc,
35}
36
37/// Comparison operators
38#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
39pub enum CompareOp {
40    Eq,
41    Ne,
42    Gt,
43    Gte,
44    Lt,
45    Lte,
46}
47
48/// A materialized view that tracks a subset of entities based on a pipeline
49#[derive(Debug)]
50pub struct MaterializedView {
51    /// View identifier
52    pub id: String,
53    /// Source view/entity this derives from
54    pub source_id: String,
55    /// Current set of entity keys in this view's result
56    current_keys: Arc<RwLock<HashSet<String>>>,
57    /// Pipeline configuration (simplified for now)
58    pipeline: ViewPipeline,
59}
60
61#[derive(Debug, Clone, Default, Serialize)]
62pub struct ViewPipeline {
63    /// Filter predicate (field path, op, value)
64    pub filter: Option<FilterConfig>,
65    /// Sort configuration
66    pub sort: Option<SortConfig>,
67    /// Limit (take N) - if Some(1), treated as single-result view for Replace effects
68    pub limit: Option<usize>,
69}
70
71#[derive(Debug, Clone, Serialize)]
72pub struct FilterConfig {
73    pub field_path: Vec<String>,
74    pub op: CompareOp,
75    pub value: Value,
76}
77
78impl FilterConfig {
79    /// Whether `entity` passes the filter. A missing field reads as null,
80    /// which passes only `Ne` against a non-null value.
81    pub fn matches<E: EntityFields + ?Sized>(&self, entity: &E) -> bool {
82        let field = lookup(entity, &self.field_path).unwrap_or(Cow::Owned(Value::Null));
83        let field: &Value = &field;
84        match self.op {
85            CompareOp::Eq => *field == self.value,
86            CompareOp::Ne => *field != self.value,
87            CompareOp::Gt => compare_values(field, &self.value) == std::cmp::Ordering::Greater,
88            CompareOp::Gte => compare_values(field, &self.value) != std::cmp::Ordering::Less,
89            CompareOp::Lt => compare_values(field, &self.value) == std::cmp::Ordering::Less,
90            CompareOp::Lte => compare_values(field, &self.value) != std::cmp::Ordering::Greater,
91        }
92    }
93}
94
95#[derive(Debug, Clone, Serialize)]
96pub struct SortConfig {
97    pub field_path: Vec<String>,
98    pub order: SortOrder,
99}
100
101impl MaterializedView {
102    /// Create a new materialized view
103    pub fn new(id: String, source_id: String, pipeline: ViewPipeline) -> Self {
104        Self {
105            id,
106            source_id,
107            current_keys: Arc::new(RwLock::new(HashSet::new())),
108            pipeline,
109        }
110    }
111
112    /// Get current keys in the view
113    pub async fn get_keys(&self) -> HashSet<String> {
114        self.current_keys.read().await.clone()
115    }
116
117    /// Evaluate initial state from cache
118    pub async fn evaluate_initial(&self, cache: &EntityCache) -> Vec<(String, Value)> {
119        let entities = cache.get_all(&self.source_id).await;
120        self.evaluate_pipeline(entities).await
121    }
122
123    /// Evaluate pipeline on a set of entities
124    async fn evaluate_pipeline(&self, mut entities: Vec<(String, Value)>) -> Vec<(String, Value)> {
125        // Apply filter
126        if let Some(ref filter) = self.pipeline.filter {
127            entities.retain(|(_, v)| self.matches_filter(v, filter));
128        }
129
130        // Apply sort
131        if let Some(ref sort) = self.pipeline.sort {
132            entities.sort_by(|(_, a), (_, b)| {
133                let a_val = extract_field(a, &sort.field_path);
134                let b_val = extract_field(b, &sort.field_path);
135                let cmp = compare_values(&a_val, &b_val);
136                match sort.order {
137                    SortOrder::Asc => cmp,
138                    SortOrder::Desc => cmp.reverse(),
139                }
140            });
141        }
142
143        // Apply limit
144        if let Some(limit) = self.pipeline.limit {
145            entities.truncate(limit);
146        }
147
148        // Update current keys
149        let keys: HashSet<String> = entities.iter().map(|(k, _)| k.clone()).collect();
150        *self.current_keys.write().await = keys;
151
152        entities
153    }
154
155    /// Check if an entity matches the filter
156    fn matches_filter(&self, entity: &Value, filter: &FilterConfig) -> bool {
157        filter.matches(entity)
158    }
159
160    /// Determine the effect of an entity update on this view
161    pub async fn compute_effect(
162        &self,
163        key: &str,
164        new_value: Option<&Value>,
165        _cache: &EntityCache,
166    ) -> ViewEffect {
167        let current_keys = self.current_keys.read().await;
168        let was_in_view = current_keys.contains(key);
169        drop(current_keys);
170
171        // Check if entity now matches filter
172        let matches_now = match new_value {
173            Some(v) => {
174                if let Some(ref filter) = self.pipeline.filter {
175                    self.matches_filter(v, filter)
176                } else {
177                    true
178                }
179            }
180            None => false, // Deleted
181        };
182
183        match (was_in_view, matches_now) {
184            (false, true) => {
185                if self.pipeline.limit == Some(1) {
186                    let current_keys = self.current_keys.read().await;
187                    if let Some(current_key) = current_keys.iter().next() {
188                        if current_key != key {
189                            return ViewEffect::Replace {
190                                old_key: current_key.clone(),
191                                new_key: key.to_string(),
192                            };
193                        }
194                    }
195                }
196                ViewEffect::Add {
197                    key: key.to_string(),
198                }
199            }
200            (true, false) => ViewEffect::Remove {
201                key: key.to_string(),
202            },
203            (true, true) => ViewEffect::Update {
204                key: key.to_string(),
205            },
206            (false, false) => ViewEffect::NoEffect,
207        }
208    }
209
210    /// Apply an effect to update the current keys
211    pub async fn apply_effect(&self, effect: &ViewEffect) {
212        let mut keys = self.current_keys.write().await;
213        match effect {
214            ViewEffect::Add { key } => {
215                keys.insert(key.clone());
216            }
217            ViewEffect::Remove { key } => {
218                keys.remove(key);
219            }
220            ViewEffect::Replace { old_key, new_key } => {
221                keys.remove(old_key);
222                keys.insert(new_key.clone());
223            }
224            ViewEffect::Update { .. } | ViewEffect::NoEffect => {}
225        }
226    }
227}
228
229/// Extract a field value from a JSON object using a path
230fn extract_field(value: &Value, path: &[String]) -> Value {
231    let mut current = value;
232    for segment in path {
233        match current.get(segment) {
234            Some(v) => current = v,
235            None => return Value::Null,
236        }
237    }
238    current.clone()
239}
240
241/// Compare two JSON values
242fn compare_values(a: &Value, b: &Value) -> std::cmp::Ordering {
243    match (a, b) {
244        (Value::Number(a), Value::Number(b)) => {
245            let a_f = a.as_f64().unwrap_or(0.0);
246            let b_f = b.as_f64().unwrap_or(0.0);
247            a_f.partial_cmp(&b_f).unwrap_or(std::cmp::Ordering::Equal)
248        }
249        (Value::String(a), Value::String(b)) => a.cmp(b),
250        (Value::Bool(a), Value::Bool(b)) => a.cmp(b),
251        _ => std::cmp::Ordering::Equal,
252    }
253}
254
255/// Registry of materialized views
256#[derive(Default)]
257pub struct MaterializedViewRegistry {
258    views: HashMap<String, Arc<MaterializedView>>,
259    /// Map from source view ID to dependent materialized views
260    dependencies: HashMap<String, Vec<String>>,
261}
262
263impl MaterializedViewRegistry {
264    pub fn new() -> Self {
265        Self::default()
266    }
267
268    /// Register a materialized view
269    pub fn register(&mut self, view: MaterializedView) {
270        let view_id = view.id.clone();
271        let source_id = view.source_id.clone();
272
273        self.dependencies
274            .entry(source_id)
275            .or_default()
276            .push(view_id.clone());
277
278        self.views.insert(view_id, Arc::new(view));
279    }
280
281    /// Get a materialized view by ID
282    pub fn get(&self, id: &str) -> Option<Arc<MaterializedView>> {
283        self.views.get(id).cloned()
284    }
285
286    /// Get all views that depend on a source
287    pub fn get_dependents(&self, source_id: &str) -> Vec<Arc<MaterializedView>> {
288        self.dependencies
289            .get(source_id)
290            .map(|ids| {
291                ids.iter()
292                    .filter_map(|id| self.views.get(id).cloned())
293                    .collect()
294            })
295            .unwrap_or_default()
296    }
297}
298
299#[cfg(test)]
300mod tests {
301    use super::*;
302    use serde_json::json;
303
304    #[tokio::test]
305    async fn test_filter_evaluation() {
306        let pipeline = ViewPipeline {
307            filter: Some(FilterConfig {
308                field_path: vec!["status".to_string()],
309                op: CompareOp::Eq,
310                value: json!("active"),
311            }),
312            sort: None,
313            limit: None,
314        };
315
316        let view =
317            MaterializedView::new("test/active".to_string(), "test/list".to_string(), pipeline);
318
319        let entities = vec![
320            ("1".to_string(), json!({"status": "active", "value": 10})),
321            ("2".to_string(), json!({"status": "inactive", "value": 20})),
322            ("3".to_string(), json!({"status": "active", "value": 30})),
323        ];
324
325        let result = view.evaluate_pipeline(entities).await;
326        assert_eq!(result.len(), 2);
327        assert_eq!(result[0].0, "1");
328        assert_eq!(result[1].0, "3");
329    }
330
331    #[tokio::test]
332    async fn test_sort_and_limit() {
333        let pipeline = ViewPipeline {
334            filter: None,
335            sort: Some(SortConfig {
336                field_path: vec!["value".to_string()],
337                order: SortOrder::Desc,
338            }),
339            limit: Some(2),
340        };
341
342        let view =
343            MaterializedView::new("test/top2".to_string(), "test/list".to_string(), pipeline);
344
345        let entities = vec![
346            ("1".to_string(), json!({"value": 10})),
347            ("2".to_string(), json!({"value": 30})),
348            ("3".to_string(), json!({"value": 20})),
349        ];
350
351        let result = view.evaluate_pipeline(entities).await;
352        assert_eq!(result.len(), 2);
353        assert_eq!(result[0].0, "2"); // value: 30
354        assert_eq!(result[1].0, "3"); // value: 20
355    }
356}
arete_server/materialized_view.rs

arete_server/
materialized_view.rs
