use crate::error::{Result, TdbError};
use parking_lot::RwLock;
use std::collections::{HashMap, HashSet};
use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
use std::sync::Arc;
use std::time::{Duration, Instant};
#[derive(Debug, Clone)]
pub struct MaterializedViewConfig {
pub max_views: usize,
pub default_refresh_strategy: RefreshStrategy,
pub enable_auto_selection: bool,
pub max_view_size: usize,
pub view_expiration: Duration,
}
impl Default for MaterializedViewConfig {
fn default() -> Self {
Self {
max_views: 100,
default_refresh_strategy: RefreshStrategy::Deferred,
enable_auto_selection: true,
max_view_size: 1_000_000,
view_expiration: Duration::from_secs(3600), }
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum RefreshStrategy {
Immediate,
Deferred,
Manual,
}
#[derive(Debug, Clone)]
pub struct MaterializedView {
pub id: u64,
pub name: String,
pub query_pattern: String,
pub refresh_strategy: RefreshStrategy,
pub is_valid: bool,
pub created_at: Instant,
pub last_refreshed: Instant,
pub result_count: usize,
pub data: Vec<Vec<u8>>,
}
impl MaterializedView {
pub fn new(id: u64, name: String, query_pattern: String, strategy: RefreshStrategy) -> Self {
let now = Instant::now();
Self {
id,
name,
query_pattern,
refresh_strategy: strategy,
is_valid: false,
created_at: now,
last_refreshed: now,
result_count: 0,
data: Vec::new(),
}
}
pub fn needs_refresh(&self, max_age: Duration) -> bool {
!self.is_valid || self.last_refreshed.elapsed() > max_age
}
pub fn invalidate(&mut self) {
self.is_valid = false;
}
pub fn refresh(&mut self, data: Vec<Vec<u8>>) {
self.data = data;
self.result_count = self.data.len();
self.last_refreshed = Instant::now();
self.is_valid = true;
}
pub fn age(&self) -> Duration {
self.last_refreshed.elapsed()
}
}
pub struct MaterializedViewManager {
config: MaterializedViewConfig,
views: RwLock<HashMap<u64, Arc<RwLock<MaterializedView>>>>,
name_to_id: RwLock<HashMap<String, u64>>,
pattern_index: RwLock<HashMap<String, HashSet<u64>>>,
next_view_id: AtomicU64,
stats: MaterializedViewStats,
}
impl MaterializedViewManager {
pub fn new(config: MaterializedViewConfig) -> Self {
Self {
config,
views: RwLock::new(HashMap::new()),
name_to_id: RwLock::new(HashMap::new()),
pattern_index: RwLock::new(HashMap::new()),
next_view_id: AtomicU64::new(1),
stats: MaterializedViewStats::default(),
}
}
pub fn create_view(
&self,
name: String,
query_pattern: String,
strategy: RefreshStrategy,
) -> Result<u64> {
if self.name_to_id.read().contains_key(&name) {
return Err(TdbError::Other(format!(
"View with name '{}' already exists",
name
)));
}
if self.views.read().len() >= self.config.max_views {
return Err(TdbError::Other(format!(
"Maximum number of views ({}) reached",
self.config.max_views
)));
}
let view_id = self.next_view_id.fetch_add(1, Ordering::Relaxed);
let view = MaterializedView::new(view_id, name.clone(), query_pattern.clone(), strategy);
self.views
.write()
.insert(view_id, Arc::new(RwLock::new(view)));
self.name_to_id.write().insert(name, view_id);
self.pattern_index
.write()
.entry(query_pattern)
.or_default()
.insert(view_id);
self.stats
.total_views_created
.fetch_add(1, Ordering::Relaxed);
Ok(view_id)
}
pub fn drop_view(&self, view_id: u64) -> Result<()> {
let views = self.views.read();
let view = views
.get(&view_id)
.ok_or_else(|| TdbError::Other(format!("View {} not found", view_id)))?;
let view_data = view.read();
let name = view_data.name.clone();
let pattern = view_data.query_pattern.clone();
drop(view_data);
drop(views);
self.views.write().remove(&view_id);
self.name_to_id.write().remove(&name);
if let Some(view_set) = self.pattern_index.write().get_mut(&pattern) {
view_set.remove(&view_id);
}
self.stats
.total_views_dropped
.fetch_add(1, Ordering::Relaxed);
Ok(())
}
pub fn refresh_view(&self, view_id: u64, data: Vec<Vec<u8>>) -> Result<()> {
let views = self.views.read();
let view = views
.get(&view_id)
.ok_or_else(|| TdbError::Other(format!("View {} not found", view_id)))?;
let mut view_data = view.write();
if data.len() > self.config.max_view_size {
return Err(TdbError::Other(format!(
"View size {} exceeds limit {}",
data.len(),
self.config.max_view_size
)));
}
view_data.refresh(data);
self.stats.total_refreshes.fetch_add(1, Ordering::Relaxed);
Ok(())
}
pub fn invalidate_views(&self, affected_patterns: &[String]) {
let pattern_index = self.pattern_index.read();
for pattern in affected_patterns {
if let Some(view_ids) = pattern_index.get(pattern) {
for &view_id in view_ids {
if let Some(view) = self.views.read().get(&view_id) {
let mut view_data = view.write();
if view_data.refresh_strategy == RefreshStrategy::Immediate {
view_data.invalidate();
self.stats
.total_invalidations
.fetch_add(1, Ordering::Relaxed);
} else {
view_data.invalidate();
self.stats
.total_invalidations
.fetch_add(1, Ordering::Relaxed);
}
}
}
}
}
}
pub fn find_applicable_views(&self, query_pattern: &str) -> Vec<u64> {
let pattern_index = self.pattern_index.read();
pattern_index
.get(query_pattern)
.map(|view_ids| view_ids.iter().copied().collect())
.unwrap_or_default()
}
pub fn get_view(&self, view_id: u64) -> Option<Arc<RwLock<MaterializedView>>> {
self.views.read().get(&view_id).cloned()
}
pub fn get_view_by_name(&self, name: &str) -> Option<Arc<RwLock<MaterializedView>>> {
let name_to_id = self.name_to_id.read();
let view_id = name_to_id.get(name)?;
self.views.read().get(view_id).cloned()
}
pub fn list_views(&self) -> Vec<ViewInfo> {
self.views
.read()
.values()
.map(|view| {
let v = view.read();
ViewInfo {
id: v.id,
name: v.name.clone(),
query_pattern: v.query_pattern.clone(),
refresh_strategy: v.refresh_strategy,
is_valid: v.is_valid,
result_count: v.result_count,
age: v.age(),
}
})
.collect()
}
pub fn cleanup_expired_views(&self) -> usize {
if self.config.view_expiration.is_zero() {
return 0;
}
let expired_views: Vec<u64> = self
.views
.read()
.values()
.filter_map(|view| {
let v = view.read();
if v.age() > self.config.view_expiration {
Some(v.id)
} else {
None
}
})
.collect();
let count = expired_views.len();
for view_id in expired_views {
let _ = self.drop_view(view_id);
}
count
}
pub fn stats(&self) -> MaterializedViewManagerStats {
MaterializedViewManagerStats {
total_views: self.views.read().len(),
total_views_created: self.stats.total_views_created.load(Ordering::Relaxed),
total_views_dropped: self.stats.total_views_dropped.load(Ordering::Relaxed),
total_refreshes: self.stats.total_refreshes.load(Ordering::Relaxed),
total_invalidations: self.stats.total_invalidations.load(Ordering::Relaxed),
total_hits: self.stats.total_hits.load(Ordering::Relaxed),
total_misses: self.stats.total_misses.load(Ordering::Relaxed),
}
}
pub fn record_hit(&self) {
self.stats.total_hits.fetch_add(1, Ordering::Relaxed);
}
pub fn record_miss(&self) {
self.stats.total_misses.fetch_add(1, Ordering::Relaxed);
}
}
#[derive(Debug, Clone)]
pub struct ViewInfo {
pub id: u64,
pub name: String,
pub query_pattern: String,
pub refresh_strategy: RefreshStrategy,
pub is_valid: bool,
pub result_count: usize,
pub age: Duration,
}
#[derive(Debug, Default)]
struct MaterializedViewStats {
total_views_created: AtomicU64,
total_views_dropped: AtomicU64,
total_refreshes: AtomicU64,
total_invalidations: AtomicU64,
total_hits: AtomicU64,
total_misses: AtomicU64,
}
#[derive(Debug, Clone)]
pub struct MaterializedViewManagerStats {
pub total_views: usize,
pub total_views_created: u64,
pub total_views_dropped: u64,
pub total_refreshes: u64,
pub total_invalidations: u64,
pub total_hits: u64,
pub total_misses: u64,
}
impl MaterializedViewManagerStats {
pub fn hit_rate(&self) -> f64 {
let total = self.total_hits + self.total_misses;
if total == 0 {
0.0
} else {
(self.total_hits as f64 / total as f64) * 100.0
}
}
pub fn avg_refreshes_per_view(&self) -> f64 {
if self.total_views_created == 0 {
0.0
} else {
self.total_refreshes as f64 / self.total_views_created as f64
}
}
}
#[cfg(test)]
#[allow(clippy::field_reassign_with_default)]
mod tests {
use super::*;
#[test]
fn test_materialized_view_creation() {
let view = MaterializedView::new(
1,
"test_view".to_string(),
"SELECT * WHERE { ?s ?p ?o }".to_string(),
RefreshStrategy::Deferred,
);
assert_eq!(view.id, 1);
assert_eq!(view.name, "test_view");
assert!(!view.is_valid);
assert_eq!(view.result_count, 0);
}
#[test]
fn test_view_refresh() {
let mut view = MaterializedView::new(
1,
"test".to_string(),
"pattern".to_string(),
RefreshStrategy::Deferred,
);
assert!(!view.is_valid);
let data = vec![vec![1, 2, 3], vec![4, 5, 6]];
view.refresh(data.clone());
assert!(view.is_valid);
assert_eq!(view.result_count, 2);
assert_eq!(view.data, data);
}
#[test]
fn test_view_invalidation() {
let mut view = MaterializedView::new(
1,
"test".to_string(),
"pattern".to_string(),
RefreshStrategy::Deferred,
);
view.refresh(vec![vec![1, 2, 3]]);
assert!(view.is_valid);
view.invalidate();
assert!(!view.is_valid);
}
#[test]
fn test_view_manager_creation() {
let config = MaterializedViewConfig::default();
let manager = MaterializedViewManager::new(config);
let stats = manager.stats();
assert_eq!(stats.total_views, 0);
assert_eq!(stats.total_views_created, 0);
}
#[test]
fn test_create_and_drop_view() {
let config = MaterializedViewConfig::default();
let manager = MaterializedViewManager::new(config);
let view_id = manager
.create_view(
"test_view".to_string(),
"SELECT * WHERE { ?s ?p ?o }".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
assert_eq!(view_id, 1);
let stats = manager.stats();
assert_eq!(stats.total_views, 1);
assert_eq!(stats.total_views_created, 1);
manager.drop_view(view_id).unwrap();
let stats = manager.stats();
assert_eq!(stats.total_views, 0);
assert_eq!(stats.total_views_dropped, 1);
}
#[test]
fn test_duplicate_view_name() {
let config = MaterializedViewConfig::default();
let manager = MaterializedViewManager::new(config);
manager
.create_view(
"dup".to_string(),
"pattern1".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
let result = manager.create_view(
"dup".to_string(),
"pattern2".to_string(),
RefreshStrategy::Deferred,
);
assert!(result.is_err());
}
#[test]
fn test_max_views_limit() {
let mut config = MaterializedViewConfig::default();
config.max_views = 2;
let manager = MaterializedViewManager::new(config);
manager
.create_view(
"view1".to_string(),
"pattern1".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
manager
.create_view(
"view2".to_string(),
"pattern2".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
let result = manager.create_view(
"view3".to_string(),
"pattern3".to_string(),
RefreshStrategy::Deferred,
);
assert!(result.is_err());
}
#[test]
fn test_refresh_view() {
let config = MaterializedViewConfig::default();
let manager = MaterializedViewManager::new(config);
let view_id = manager
.create_view(
"test".to_string(),
"pattern".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
let data = vec![vec![1, 2, 3], vec![4, 5, 6]];
manager.refresh_view(view_id, data.clone()).unwrap();
let view = manager.get_view(view_id).unwrap();
let view_data = view.read();
assert!(view_data.is_valid);
assert_eq!(view_data.result_count, 2);
}
#[test]
fn test_view_size_limit() {
let mut config = MaterializedViewConfig::default();
config.max_view_size = 2;
let manager = MaterializedViewManager::new(config);
let view_id = manager
.create_view(
"test".to_string(),
"pattern".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
let data = vec![vec![1], vec![2], vec![3]]; let result = manager.refresh_view(view_id, data);
assert!(result.is_err());
}
#[test]
fn test_invalidate_views() {
let config = MaterializedViewConfig::default();
let manager = MaterializedViewManager::new(config);
let view_id = manager
.create_view(
"test".to_string(),
"pattern1".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
manager.refresh_view(view_id, vec![vec![1, 2, 3]]).unwrap();
let view = manager.get_view(view_id).unwrap();
assert!(view.read().is_valid);
manager.invalidate_views(&["pattern1".to_string()]);
assert!(!view.read().is_valid);
let stats = manager.stats();
assert_eq!(stats.total_invalidations, 1);
}
#[test]
fn test_find_applicable_views() {
let config = MaterializedViewConfig::default();
let manager = MaterializedViewManager::new(config);
let view_id1 = manager
.create_view(
"view1".to_string(),
"pattern1".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
let view_id2 = manager
.create_view(
"view2".to_string(),
"pattern1".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
let applicable = manager.find_applicable_views("pattern1");
assert_eq!(applicable.len(), 2);
assert!(applicable.contains(&view_id1));
assert!(applicable.contains(&view_id2));
}
#[test]
fn test_get_view_by_name() {
let config = MaterializedViewConfig::default();
let manager = MaterializedViewManager::new(config);
manager
.create_view(
"myview".to_string(),
"pattern".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
let view = manager.get_view_by_name("myview");
assert!(view.is_some());
let view_arc = view.unwrap();
let view_data = view_arc.read();
assert_eq!(view_data.name, "myview");
}
#[test]
fn test_list_views() {
let config = MaterializedViewConfig::default();
let manager = MaterializedViewManager::new(config);
manager
.create_view(
"view1".to_string(),
"pattern1".to_string(),
RefreshStrategy::Deferred,
)
.unwrap();
manager
.create_view(
"view2".to_string(),
"pattern2".to_string(),
RefreshStrategy::Immediate,
)
.unwrap();
let views = manager.list_views();
assert_eq!(views.len(), 2);
let names: Vec<String> = views.iter().map(|v| v.name.clone()).collect();
assert!(names.contains(&"view1".to_string()));
assert!(names.contains(&"view2".to_string()));
}
#[test]
fn test_hit_miss_tracking() {
let config = MaterializedViewConfig::default();
let manager = MaterializedViewManager::new(config);
manager.record_hit();
manager.record_hit();
manager.record_miss();
let stats = manager.stats();
assert_eq!(stats.total_hits, 2);
assert_eq!(stats.total_misses, 1);
assert!((stats.hit_rate() - 66.67).abs() < 0.1);
}
#[test]
fn test_stats_calculations() {
let stats = MaterializedViewManagerStats {
total_views: 5,
total_views_created: 10,
total_views_dropped: 5,
total_refreshes: 30,
total_invalidations: 15,
total_hits: 80,
total_misses: 20,
};
assert_eq!(stats.hit_rate(), 80.0);
assert_eq!(stats.avg_refreshes_per_view(), 3.0);
}
#[test]
fn test_view_needs_refresh() {
let view = MaterializedView::new(
1,
"test".to_string(),
"pattern".to_string(),
RefreshStrategy::Deferred,
);
assert!(view.needs_refresh(Duration::from_secs(60)));
let mut view2 = view.clone();
view2.refresh(vec![vec![1, 2, 3]]);
assert!(!view2.needs_refresh(Duration::from_secs(3600)));
}
}