oxirs_core/storage/

mod.rs

//! Next-Generation Storage Engine for OxiRS
//!
//! This module implements a quantum-ready storage architecture with:
//! - Tiered storage with intelligent data placement
//! - Columnar storage for analytical workloads  
//! - Time-series optimization for temporal RDF
//! - Immutable storage with content-addressable blocks
//! - Advanced compression (LZ4, ZSTD, custom RDF codecs)
//! - Storage virtualization with transparent migration
//! - Multi-Version Concurrency Control (MVCC) for high-concurrency operations

// #[cfg(feature = "columnar")]
// pub mod columnar; // TODO: Add 'columnar' feature to Cargo.toml when ready
pub mod compression;
pub mod immutable;
pub mod mmap_storage;
pub mod mvcc;
pub mod temporal;
pub mod tiered;
pub mod virtualization;

pub use mvcc::{IsolationLevel, MvccConfig, MvccStore, TransactionId as MvccTransactionId};
use parking_lot::RwLock;

use crate::OxirsError;
use std::path::Path;
use std::sync::Arc;

/// Storage configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct StorageConfig {
    /// Enable tiered storage
    pub enable_tiering: bool,
    /// Enable columnar storage for analytics
    pub enable_columnar: bool,
    /// Enable temporal optimization
    pub enable_temporal: bool,
    /// Compression algorithm
    pub compression: CompressionType,
    /// Storage tiers configuration
    pub tiers: TierConfig,
    /// Cache size in MB
    pub cache_size_mb: usize,
}

impl Default for StorageConfig {
    fn default() -> Self {
        StorageConfig {
            enable_tiering: true,
            enable_columnar: true,
            enable_temporal: true,
            compression: CompressionType::Zstd { level: 3 },
            tiers: TierConfig::default(),
            cache_size_mb: 1024,
        }
    }
}

/// Compression types
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum CompressionType {
    None,
    Lz4 { level: u32 },
    Zstd { level: i32 },
    RdfCustom { dictionary_size: usize },
}

/// Storage tier configuration
#[derive(Debug, Clone, Default, serde::Serialize, serde::Deserialize)]
pub struct TierConfig {
    /// Hot tier: in-memory with fastest access
    pub hot_tier: HotTierConfig,
    /// Warm tier: SSD-optimized storage
    pub warm_tier: WarmTierConfig,
    /// Cold tier: HDD/object storage
    pub cold_tier: ColdTierConfig,
    /// Archive tier: long-term immutable storage
    pub archive_tier: ArchiveTierConfig,
}

/// Hot tier configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct HotTierConfig {
    /// Maximum size in MB
    pub max_size_mb: usize,
    /// Eviction policy
    pub eviction_policy: EvictionPolicy,
    /// Time to live in seconds
    pub ttl_seconds: Option<u64>,
}

impl Default for HotTierConfig {
    fn default() -> Self {
        HotTierConfig {
            max_size_mb: 4096,
            eviction_policy: EvictionPolicy::Lru,
            ttl_seconds: Some(3600),
        }
    }
}

/// Warm tier configuration  
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct WarmTierConfig {
    /// Path to warm storage
    pub path: String,
    /// Maximum size in GB
    pub max_size_gb: usize,
    /// Promotion threshold (access count)
    pub promotion_threshold: u32,
    /// Demotion threshold (days since last access)
    pub demotion_threshold_days: u32,
}

impl Default for WarmTierConfig {
    fn default() -> Self {
        WarmTierConfig {
            path: "/var/oxirs/warm".to_string(),
            max_size_gb: 100,
            promotion_threshold: 10,
            demotion_threshold_days: 7,
        }
    }
}

/// Cold tier configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ColdTierConfig {
    /// Path to cold storage
    pub path: String,
    /// Maximum size in TB
    pub max_size_tb: usize,
    /// Compression level
    pub compression_level: i32,
    /// Archive threshold (days since last access)
    pub archive_threshold_days: u32,
}

impl Default for ColdTierConfig {
    fn default() -> Self {
        ColdTierConfig {
            path: "/var/oxirs/cold".to_string(),
            max_size_tb: 10,
            compression_level: 9,
            archive_threshold_days: 90,
        }
    }
}

/// Archive tier configuration
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub struct ArchiveTierConfig {
    /// Archive storage backend
    pub backend: ArchiveBackend,
    /// Retention policy
    pub retention_years: Option<u32>,
    /// Immutability guarantee
    pub immutable: bool,
}

impl Default for ArchiveTierConfig {
    fn default() -> Self {
        ArchiveTierConfig {
            backend: ArchiveBackend::Local("/var/oxirs/archive".to_string()),
            retention_years: Some(7),
            immutable: true,
        }
    }
}

/// Archive storage backend
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum ArchiveBackend {
    Local(String),
    S3 { bucket: String, prefix: String },
    GCS { bucket: String, prefix: String },
    Azure { container: String, prefix: String },
}

/// Eviction policy for hot tier
#[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
pub enum EvictionPolicy {
    Lru,
    Lfu,
    Fifo,
    Adaptive,
}

/// Storage engine trait
#[async_trait::async_trait]
pub trait StorageEngine: Send + Sync {
    /// Initialize the storage engine
    async fn init(&mut self, config: StorageConfig) -> Result<(), OxirsError>;

    /// Store a triple
    async fn store_triple(&self, triple: &crate::model::Triple) -> Result<(), OxirsError>;

    /// Store multiple triples
    async fn store_triples(&self, triples: &[crate::model::Triple]) -> Result<(), OxirsError>;

    /// Query triples by pattern
    async fn query_triples(
        &self,
        pattern: &crate::model::TriplePattern,
    ) -> Result<Vec<crate::model::Triple>, OxirsError>;

    /// Delete triples by pattern
    async fn delete_triples(
        &self,
        pattern: &crate::model::TriplePattern,
    ) -> Result<usize, OxirsError>;

    /// Get storage statistics
    async fn stats(&self) -> Result<StorageStats, OxirsError>;

    /// Optimize storage
    async fn optimize(&self) -> Result<(), OxirsError>;

    /// Backup storage
    async fn backup(&self, path: &Path) -> Result<(), OxirsError>;

    /// Restore from backup
    async fn restore(&self, path: &Path) -> Result<(), OxirsError>;
}

/// Storage statistics
#[derive(Debug, Clone)]
pub struct StorageStats {
    /// Total number of triples
    pub total_triples: u64,
    /// Storage size in bytes
    pub total_size_bytes: u64,
    /// Tier distribution
    pub tier_stats: TierStats,
    /// Compression ratio
    pub compression_ratio: f64,
    /// Query performance metrics
    pub query_metrics: QueryMetrics,
}

/// Tier statistics
#[derive(Debug, Clone)]
pub struct TierStats {
    /// Hot tier stats
    pub hot: TierStat,
    /// Warm tier stats
    pub warm: TierStat,
    /// Cold tier stats
    pub cold: TierStat,
    /// Archive tier stats
    pub archive: TierStat,
}

/// Individual tier statistics
#[derive(Debug, Clone)]
pub struct TierStat {
    /// Number of triples
    pub triple_count: u64,
    /// Size in bytes
    pub size_bytes: u64,
    /// Hit rate percentage
    pub hit_rate: f64,
    /// Average access time in microseconds
    pub avg_access_time_us: u64,
}

/// Query performance metrics
#[derive(Debug, Clone)]
pub struct QueryMetrics {
    /// Average query time in milliseconds
    pub avg_query_time_ms: f64,
    /// 99th percentile query time
    pub p99_query_time_ms: f64,
    /// Queries per second
    pub qps: f64,
    /// Cache hit rate
    pub cache_hit_rate: f64,
}

/// Create a new storage engine with the given configuration
pub async fn create_engine(config: StorageConfig) -> Result<Arc<dyn StorageEngine>, OxirsError> {
    let engine = SimpleStorageEngine::new(config).await?;
    Ok(Arc::new(engine))
}

/// Simple file-based storage engine implementation
pub struct SimpleStorageEngine {
    #[allow(dead_code)]
    config: StorageConfig,
    mvcc_store: MvccStore,
    stats: Arc<RwLock<StorageStats>>,
    #[allow(dead_code)]
    base_path: std::path::PathBuf,
}

impl SimpleStorageEngine {
    /// Create a new simple storage engine
    pub async fn new(config: StorageConfig) -> Result<Self, OxirsError> {
        let base_path = std::path::PathBuf::from("/tmp/oxirs_storage");
        std::fs::create_dir_all(&base_path)
            .map_err(|e| OxirsError::Store(format!("Failed to create storage directory: {e}")))?;

        let mvcc_config = MvccConfig {
            max_versions_per_triple: 100,
            gc_interval: std::time::Duration::from_secs(60),
            min_version_age: std::time::Duration::from_secs(30),
            enable_snapshot_isolation: true,
            enable_read_your_writes: true,
            conflict_detection: mvcc::ConflictDetection::OptimisticTwoPhase,
        };

        let mvcc_store = MvccStore::new(mvcc_config);

        let initial_stats = StorageStats {
            total_triples: 0,
            total_size_bytes: 0,
            tier_stats: TierStats {
                hot: TierStat {
                    triple_count: 0,
                    size_bytes: 0,
                    hit_rate: 0.0,
                    avg_access_time_us: 0,
                },
                warm: TierStat {
                    triple_count: 0,
                    size_bytes: 0,
                    hit_rate: 0.0,
                    avg_access_time_us: 0,
                },
                cold: TierStat {
                    triple_count: 0,
                    size_bytes: 0,
                    hit_rate: 0.0,
                    avg_access_time_us: 0,
                },
                archive: TierStat {
                    triple_count: 0,
                    size_bytes: 0,
                    hit_rate: 0.0,
                    avg_access_time_us: 0,
                },
            },
            compression_ratio: 1.0,
            query_metrics: QueryMetrics {
                avg_query_time_ms: 0.0,
                p99_query_time_ms: 0.0,
                qps: 0.0,
                cache_hit_rate: 0.0,
            },
        };

        Ok(Self {
            config,
            mvcc_store,
            stats: Arc::new(RwLock::new(initial_stats)),
            base_path,
        })
    }
}

#[async_trait::async_trait]
impl StorageEngine for SimpleStorageEngine {
    async fn init(&mut self, _config: StorageConfig) -> Result<(), OxirsError> {
        // Already initialized in new()
        Ok(())
    }

    async fn store_triple(&self, triple: &crate::model::Triple) -> Result<(), OxirsError> {
        let tx_id = self
            .mvcc_store
            .begin_transaction(IsolationLevel::Snapshot)
            .map_err(|e| OxirsError::Store(format!("Failed to begin transaction: {e}")))?;

        self.mvcc_store
            .insert(tx_id, triple.clone())
            .map_err(|e| OxirsError::Store(format!("Failed to insert triple: {e}")))?;

        self.mvcc_store
            .commit_transaction(tx_id)
            .map_err(|e| OxirsError::Store(format!("Failed to commit transaction: {e}")))?;

        // Update statistics
        let mut stats = self.stats.write();
        stats.total_triples += 1;
        stats.tier_stats.hot.triple_count += 1;

        Ok(())
    }

    async fn store_triples(&self, triples: &[crate::model::Triple]) -> Result<(), OxirsError> {
        let tx_id = self
            .mvcc_store
            .begin_transaction(IsolationLevel::Snapshot)
            .map_err(|e| OxirsError::Store(format!("Failed to begin transaction: {e}")))?;

        for triple in triples {
            self.mvcc_store
                .insert(tx_id, triple.clone())
                .map_err(|e| OxirsError::Store(format!("Failed to insert triple: {e}")))?;
        }

        self.mvcc_store
            .commit_transaction(tx_id)
            .map_err(|e| OxirsError::Store(format!("Failed to commit transaction: {e}")))?;

        // Update statistics
        let mut stats = self.stats.write();
        stats.total_triples += triples.len() as u64;
        stats.tier_stats.hot.triple_count += triples.len() as u64;

        Ok(())
    }

    async fn query_triples(
        &self,
        pattern: &crate::model::TriplePattern,
    ) -> Result<Vec<crate::model::Triple>, OxirsError> {
        let tx_id = self
            .mvcc_store
            .begin_transaction(IsolationLevel::Snapshot)
            .map_err(|e| OxirsError::Store(format!("Failed to begin transaction: {e}")))?;

        // Convert patterns to concrete terms for MVCC query
        let subject = Self::pattern_to_subject(pattern.subject());
        let predicate = Self::pattern_to_predicate(pattern.predicate());
        let object = Self::pattern_to_object(pattern.object());

        let results = self
            .mvcc_store
            .query(tx_id, subject.as_ref(), predicate.as_ref(), object.as_ref())
            .map_err(|e| OxirsError::Store(format!("Failed to query triples: {e}")))?;

        // Filter results to match the pattern (in case of variables)
        let filtered: Vec<_> = results
            .into_iter()
            .filter(|triple| pattern.matches(triple))
            .collect();

        Ok(filtered)
    }

    async fn delete_triples(
        &self,
        pattern: &crate::model::TriplePattern,
    ) -> Result<usize, OxirsError> {
        let tx_id = self
            .mvcc_store
            .begin_transaction(IsolationLevel::Snapshot)
            .map_err(|e| OxirsError::Store(format!("Failed to begin transaction: {e}")))?;

        // Convert patterns to concrete terms for MVCC query
        let subject = Self::pattern_to_subject(pattern.subject());
        let predicate = Self::pattern_to_predicate(pattern.predicate());
        let object = Self::pattern_to_object(pattern.object());

        // First query to find matching triples
        let matching_triples = self
            .mvcc_store
            .query(tx_id, subject.as_ref(), predicate.as_ref(), object.as_ref())
            .map_err(|e| OxirsError::Store(format!("Failed to query triples for deletion: {e}")))?;

        // Filter results to match the pattern exactly
        let filtered: Vec<_> = matching_triples
            .into_iter()
            .filter(|triple| pattern.matches(triple))
            .collect();

        let deleted_count = filtered.len();

        // Delete each matching triple
        for triple in &filtered {
            self.mvcc_store
                .delete(tx_id, triple)
                .map_err(|e| OxirsError::Store(format!("Failed to delete triple: {e}")))?;
        }

        self.mvcc_store
            .commit_transaction(tx_id)
            .map_err(|e| OxirsError::Store(format!("Failed to commit transaction: {e}")))?;

        // Update statistics
        let mut stats = self.stats.write();
        stats.total_triples = stats.total_triples.saturating_sub(deleted_count as u64);
        stats.tier_stats.hot.triple_count = stats
            .tier_stats
            .hot
            .triple_count
            .saturating_sub(deleted_count as u64);

        Ok(deleted_count)
    }

    async fn stats(&self) -> Result<StorageStats, OxirsError> {
        let stats = self.stats.read();
        Ok(stats.clone())
    }

    async fn optimize(&self) -> Result<(), OxirsError> {
        // Run garbage collection on MVCC store
        self.mvcc_store
            .garbage_collect()
            .map_err(|e| OxirsError::Store(format!("Failed to optimize storage: {e}")))?;
        Ok(())
    }

    async fn backup(&self, path: &Path) -> Result<(), OxirsError> {
        // Simple backup implementation - serialize current state to file
        let backup_path = path.join("oxirs_backup.json");

        // Get all triples using a query that matches everything
        let all_pattern = crate::model::TriplePattern::new(None, None, None);
        let triples = self.query_triples(&all_pattern).await?;

        let serialized = serde_json::to_string_pretty(&triples)
            .map_err(|e| OxirsError::Store(format!("Failed to serialize backup: {e}")))?;

        std::fs::write(&backup_path, serialized)
            .map_err(|e| OxirsError::Store(format!("Failed to write backup: {e}")))?;

        Ok(())
    }

    async fn restore(&self, path: &Path) -> Result<(), OxirsError> {
        // Simple restore implementation - deserialize from file
        let backup_path = path.join("oxirs_backup.json");

        let serialized = std::fs::read_to_string(&backup_path)
            .map_err(|e| OxirsError::Store(format!("Failed to read backup: {e}")))?;

        let triples: Vec<crate::model::Triple> = serde_json::from_str(&serialized)
            .map_err(|e| OxirsError::Store(format!("Failed to deserialize backup: {e}")))?;

        self.store_triples(&triples).await?;

        Ok(())
    }
}

impl SimpleStorageEngine {
    /// Convert a subject pattern to a concrete subject term
    fn pattern_to_subject(
        pattern: Option<&crate::model::pattern::SubjectPattern>,
    ) -> Option<crate::model::Subject> {
        pattern?.try_into().ok()
    }

    /// Convert a predicate pattern to a concrete predicate term
    fn pattern_to_predicate(
        pattern: Option<&crate::model::pattern::PredicatePattern>,
    ) -> Option<crate::model::Predicate> {
        pattern?.try_into().ok()
    }

    /// Convert an object pattern to a concrete object term
    fn pattern_to_object(
        pattern: Option<&crate::model::pattern::ObjectPattern>,
    ) -> Option<crate::model::Object> {
        pattern?.try_into().ok()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn test_default_config() {
        let config = StorageConfig::default();
        assert!(config.enable_tiering);
        assert!(config.enable_columnar);
        assert!(config.enable_temporal);
        assert_eq!(config.cache_size_mb, 1024);
    }
}