arrow_graph/performance/
memory.rs

1use crate::error::Result;
2use crate::graph::ArrowGraph;
3use arrow::array::{Array, StringArray, Float64Array};
4use arrow::record_batch::RecordBatch;
5use memmap2::{Mmap, MmapMut, MmapOptions};
6use parking_lot::{RwLock, Mutex};
7use std::collections::HashMap;
8use std::fs::{File, OpenOptions};
9use std::path::{Path, PathBuf};
10use std::sync::Arc;
11use std::alloc::{GlobalAlloc, Layout, System};
12use std::ptr::NonNull;
13use aligned_vec::avec;
14
15/// Memory-mapped graph storage for handling massive graphs
16/// Provides efficient disk-backed storage with OS-level caching
17#[derive(Debug)]
18pub struct MemoryMappedGraph {
19    nodes_mmap: Arc<RwLock<Option<Mmap>>>,
20    edges_mmap: Arc<RwLock<Option<Mmap>>>,
21    metadata: GraphMetadata,
22    storage_path: PathBuf,
23    read_only: bool,
24}
25
26/// Graph metadata for memory-mapped storage
27#[derive(Debug, Clone)]
28pub struct GraphMetadata {
29    pub num_nodes: usize,
30    pub num_edges: usize,
31    pub node_size: usize,
32    pub edge_size: usize,
33    pub version: u32,
34    pub created_at: u64,
35    pub last_modified: u64,
36    pub checksum: u64,
37}
38
39/// High-performance memory pool for graph operations
40/// Uses cache-aligned allocations and memory pooling
41#[derive(Debug)]
42pub struct MemoryPool {
43    pools: HashMap<usize, Arc<Mutex<Vec<NonNull<u8>>>>>,
44    alignment: usize,
45    max_pool_size: usize,
46    allocated_bytes: Arc<Mutex<usize>>,
47    peak_usage: Arc<Mutex<usize>>,
48}
49
50/// Cache-optimized storage for hot graph data
51/// Implements intelligent caching strategies
52#[derive(Debug)]
53pub struct CacheOptimizedStorage {
54    l1_cache: Arc<RwLock<HashMap<CacheKey, CacheEntry>>>,
55    l2_cache: Arc<RwLock<HashMap<CacheKey, CacheEntry>>>,
56    cache_config: CacheConfig,
57    hit_stats: Arc<Mutex<CacheStats>>,
58}
59
60/// Cache configuration parameters
61#[derive(Debug, Clone)]
62pub struct CacheConfig {
63    pub l1_size: usize,
64    pub l2_size: usize,
65    pub eviction_policy: EvictionPolicy,
66    pub prefetch_strategy: PrefetchStrategy,
67    pub compression_enabled: bool,
68}
69
70/// Cache eviction policies
71#[derive(Debug, Clone)]
72pub enum EvictionPolicy {
73    LRU,        // Least Recently Used
74    LFU,        // Least Frequently Used
75    ARC,        // Adaptive Replacement Cache
76    Random,     // Random eviction
77    FIFO,       // First In, First Out
78}
79
80/// Prefetch strategies for predictive caching
81#[derive(Debug, Clone)]
82pub enum PrefetchStrategy {
83    None,
84    Sequential,     // Sequential access pattern
85    Neighbors,      // Prefetch neighbor nodes
86    RandomWalk,     // Based on random walk patterns
87    ML,            // Machine learning based
88}
89
90/// Cache key for storing graph elements
91#[derive(Debug, Clone, Hash, PartialEq, Eq)]
92pub enum CacheKey {
93    Node(String),
94    Edge(String, String),
95    Subgraph(Vec<String>),
96    Algorithm(String, Vec<String>), // Algorithm name + parameters
97}
98
99/// Cache entry with metadata
100#[derive(Debug, Clone)]
101pub struct CacheEntry {
102    pub data: CacheData,
103    pub size: usize,
104    pub access_count: u64,
105    pub last_accessed: u64,
106    pub created_at: u64,
107    pub compressed: bool,
108}
109
110/// Cached data types
111#[derive(Debug, Clone)]
112pub enum CacheData {
113    NodeData(Vec<u8>),
114    EdgeData(Vec<u8>),
115    ComputationResult(Vec<u8>),
116    Metadata(Vec<u8>),
117}
118
119/// Cache performance statistics
120#[derive(Debug, Clone)]
121pub struct CacheStats {
122    pub l1_hits: u64,
123    pub l1_misses: u64,
124    pub l2_hits: u64,
125    pub l2_misses: u64,
126    pub evictions: u64,
127    pub prefetch_hits: u64,
128    pub memory_usage: usize,
129}
130
131/// Memory usage statistics
132#[derive(Debug, Clone)]
133pub struct MemoryStats {
134    pub total_allocated: usize,
135    pub peak_usage: usize,
136    pub pool_usage: HashMap<usize, usize>,
137    pub fragmentation_ratio: f64,
138    pub cache_hit_ratio: f64,
139}
140
141/// Custom allocator for graph operations
142pub struct GraphAllocator {
143    pool: Arc<MemoryPool>,
144}
145
146impl MemoryMappedGraph {
147    /// Create a new memory-mapped graph
148    pub fn new<P: AsRef<Path>>(storage_path: P) -> Result<Self> {
149        let path = storage_path.as_ref().to_path_buf();
150        
151        // Ensure directory exists
152        if let Some(parent) = path.parent() {
153            std::fs::create_dir_all(parent)?;
154        }
155        
156        Ok(Self {
157            nodes_mmap: Arc::new(RwLock::new(None)),
158            edges_mmap: Arc::new(RwLock::new(None)),
159            metadata: GraphMetadata::default(),
160            storage_path: path,
161            read_only: false,
162        })
163    }
164
165    /// Open existing memory-mapped graph in read-only mode
166    pub fn open_readonly<P: AsRef<Path>>(storage_path: P) -> Result<Self> {
167        let mut graph = Self::new(storage_path)?;
168        graph.read_only = true;
169        graph.load_metadata()?;
170        graph.map_existing_files()?;
171        Ok(graph)
172    }
173
174    /// Create memory-mapped graph from Arrow graph
175    pub fn from_arrow_graph<P: AsRef<Path>>(
176        graph: &ArrowGraph,
177        storage_path: P,
178    ) -> Result<Self> {
179        let mut mmap_graph = Self::new(storage_path)?;
180        mmap_graph.serialize_arrow_graph(graph)?;
181        Ok(mmap_graph)
182    }
183
184    /// Serialize Arrow graph to memory-mapped files
185    fn serialize_arrow_graph(&mut self, graph: &ArrowGraph) -> Result<()> {
186        self.metadata.num_nodes = graph.node_count();
187        self.metadata.num_edges = graph.edge_count();
188        
189        // Serialize nodes
190        self.serialize_nodes(&graph.nodes)?;
191        
192        // Serialize edges
193        self.serialize_edges(&graph.edges)?;
194        
195        // Save metadata
196        self.save_metadata()?;
197        
198        Ok(())
199    }
200
201    /// Serialize nodes to memory-mapped file
202    fn serialize_nodes(&mut self, nodes_batch: &RecordBatch) -> Result<()> {
203        let nodes_file_path = self.storage_path.join("nodes.mmap");
204        
205        // Calculate required size
206        let estimated_size = self.estimate_nodes_size(nodes_batch)?;
207        
208        // Create and resize file
209        let file = OpenOptions::new()
210            .create(true)
211            .write(true)
212            .read(true)
213            .open(&nodes_file_path)?;
214        file.set_len(estimated_size as u64)?;
215        
216        // Memory map the file
217        let mut mmap = unsafe { MmapOptions::new().map_mut(&file)? };
218        
219        // Serialize node data
220        let mut offset = 0;
221        if nodes_batch.num_rows() > 0 {
222            let node_ids = nodes_batch.column(0)
223                .as_any()
224                .downcast_ref::<StringArray>()
225                .ok_or_else(|| crate::error::GraphError::graph_construction("Expected string array for node IDs"))?;
226            
227            for i in 0..node_ids.len() {
228                let node_id = node_ids.value(i);
229                let node_data = node_id.as_bytes();
230                
231                // Write length prefix
232                let len = node_data.len() as u32;
233                mmap[offset..offset + 4].copy_from_slice(&len.to_le_bytes());
234                offset += 4;
235                
236                // Write node data
237                mmap[offset..offset + node_data.len()].copy_from_slice(node_data);
238                offset += node_data.len();
239            }
240        }
241        
242        // Ensure data is written to disk
243        mmap.flush()?;
244        
245        // Store mmap reference
246        *self.nodes_mmap.write() = Some(mmap.make_read_only()?);
247        
248        self.metadata.node_size = estimated_size;
249        
250        Ok(())
251    }
252
253    /// Serialize edges to memory-mapped file
254    fn serialize_edges(&mut self, edges_batch: &RecordBatch) -> Result<()> {
255        let edges_file_path = self.storage_path.join("edges.mmap");
256        
257        // Calculate required size
258        let estimated_size = self.estimate_edges_size(edges_batch)?;
259        
260        // Create and resize file
261        let file = OpenOptions::new()
262            .create(true)
263            .write(true)
264            .read(true)
265            .open(&edges_file_path)?;
266        file.set_len(estimated_size as u64)?;
267        
268        // Memory map the file
269        let mut mmap = unsafe { MmapOptions::new().map_mut(&file)? };
270        
271        // Serialize edge data
272        let mut offset = 0;
273        if edges_batch.num_rows() > 0 {
274            let source_ids = edges_batch.column(0)
275                .as_any()
276                .downcast_ref::<StringArray>()
277                .ok_or_else(|| crate::error::GraphError::graph_construction("Expected string array for source IDs"))?;
278            let target_ids = edges_batch.column(1)
279                .as_any()
280                .downcast_ref::<StringArray>()
281                .ok_or_else(|| crate::error::GraphError::graph_construction("Expected string array for target IDs"))?;
282            let weights = edges_batch.column(2)
283                .as_any()
284                .downcast_ref::<Float64Array>()
285                .ok_or_else(|| crate::error::GraphError::graph_construction("Expected float64 array for weights"))?;
286            
287            for i in 0..source_ids.len() {
288                let source = source_ids.value(i);
289                let target = target_ids.value(i);
290                let weight = weights.value(i);
291                
292                // Write source length and data
293                let source_len = source.len() as u32;
294                mmap[offset..offset + 4].copy_from_slice(&source_len.to_le_bytes());
295                offset += 4;
296                mmap[offset..offset + source.len()].copy_from_slice(source.as_bytes());
297                offset += source.len();
298                
299                // Write target length and data
300                let target_len = target.len() as u32;
301                mmap[offset..offset + 4].copy_from_slice(&target_len.to_le_bytes());
302                offset += 4;
303                mmap[offset..offset + target.len()].copy_from_slice(target.as_bytes());
304                offset += target.len();
305                
306                // Write weight
307                mmap[offset..offset + 8].copy_from_slice(&weight.to_le_bytes());
308                offset += 8;
309            }
310        }
311        
312        // Ensure data is written to disk
313        mmap.flush()?;
314        
315        // Store mmap reference
316        *self.edges_mmap.write() = Some(mmap.make_read_only()?);
317        
318        self.metadata.edge_size = estimated_size;
319        
320        Ok(())
321    }
322
323    /// Estimate required size for nodes serialization
324    fn estimate_nodes_size(&self, nodes_batch: &RecordBatch) -> Result<usize> {
325        let mut size = 0;
326        
327        if nodes_batch.num_rows() > 0 {
328            let node_ids = nodes_batch.column(0)
329                .as_any()
330                .downcast_ref::<StringArray>()
331                .ok_or_else(|| crate::error::GraphError::graph_construction("Expected string array for node IDs"))?;
332            
333            for i in 0..node_ids.len() {
334                size += 4; // Length prefix
335                size += node_ids.value(i).len(); // Node ID data
336            }
337        }
338        
339        // Add 10% padding for alignment
340        Ok((size as f64 * 1.1) as usize)
341    }
342
343    /// Estimate required size for edges serialization
344    fn estimate_edges_size(&self, edges_batch: &RecordBatch) -> Result<usize> {
345        let mut size = 0;
346        
347        if edges_batch.num_rows() > 0 {
348            let source_ids = edges_batch.column(0)
349                .as_any()
350                .downcast_ref::<StringArray>()
351                .ok_or_else(|| crate::error::GraphError::graph_construction("Expected string array for source IDs"))?;
352            let target_ids = edges_batch.column(1)
353                .as_any()
354                .downcast_ref::<StringArray>()
355                .ok_or_else(|| crate::error::GraphError::graph_construction("Expected string array for target IDs"))?;
356            
357            for i in 0..source_ids.len() {
358                size += 4 + source_ids.value(i).len(); // Source
359                size += 4 + target_ids.value(i).len(); // Target
360                size += 8; // Weight (f64)
361            }
362        }
363        
364        // Add 10% padding
365        Ok((size as f64 * 1.1) as usize)
366    }
367
368    /// Load existing memory-mapped files
369    fn map_existing_files(&mut self) -> Result<()> {
370        // Map nodes file
371        let nodes_path = self.storage_path.join("nodes.mmap");
372        if nodes_path.exists() {
373            let file = File::open(&nodes_path)?;
374            let mmap = unsafe { MmapOptions::new().map(&file)? };
375            *self.nodes_mmap.write() = Some(mmap);
376        }
377        
378        // Map edges file
379        let edges_path = self.storage_path.join("edges.mmap");
380        if edges_path.exists() {
381            let file = File::open(&edges_path)?;
382            let mmap = unsafe { MmapOptions::new().map(&file)? };
383            *self.edges_mmap.write() = Some(mmap);
384        }
385        
386        Ok(())
387    }
388
389    /// Save metadata to file
390    fn save_metadata(&self) -> Result<()> {
391        let metadata_path = self.storage_path.join("metadata.json");
392        let json = serde_json::to_string_pretty(&self.metadata)?;
393        std::fs::write(metadata_path, json)?;
394        Ok(())
395    }
396
397    /// Load metadata from file
398    fn load_metadata(&mut self) -> Result<()> {
399        let metadata_path = self.storage_path.join("metadata.json");
400        if metadata_path.exists() {
401            let json = std::fs::read_to_string(metadata_path)?;
402            self.metadata = serde_json::from_str(&json)?;
403        }
404        Ok(())
405    }
406
407    /// Get node data by ID
408    pub fn get_node(&self, node_id: &str) -> Result<Option<Vec<u8>>> {
409        let nodes_mmap = self.nodes_mmap.read();
410        if let Some(ref mmap) = *nodes_mmap {
411            // Linear search through memory-mapped nodes
412            // In production, this would use an index
413            let mut offset = 0;
414            while offset < mmap.len() {
415                if offset + 4 > mmap.len() {
416                    break;
417                }
418                
419                // Read length
420                let len = u32::from_le_bytes([
421                    mmap[offset],
422                    mmap[offset + 1],
423                    mmap[offset + 2],
424                    mmap[offset + 3],
425                ]) as usize;
426                offset += 4;
427                
428                if offset + len > mmap.len() {
429                    break;
430                }
431                
432                // Read node ID
433                let stored_id = std::str::from_utf8(&mmap[offset..offset + len])
434                    .map_err(|e| crate::error::GraphError::graph_construction(&format!("UTF-8 error: {}", e)))?;
435                
436                if stored_id == node_id {
437                    return Ok(Some(mmap[offset..offset + len].to_vec()));
438                }
439                
440                offset += len;
441            }
442        }
443        
444        Ok(None)
445    }
446
447    /// Get memory usage statistics
448    pub fn get_memory_stats(&self) -> MemoryStats {
449        let nodes_size = self.nodes_mmap.read()
450            .as_ref()
451            .map(|mmap| mmap.len())
452            .unwrap_or(0);
453        
454        let edges_size = self.edges_mmap.read()
455            .as_ref()
456            .map(|mmap| mmap.len())
457            .unwrap_or(0);
458        
459        let total = nodes_size + edges_size;
460        
461        MemoryStats {
462            total_allocated: total,
463            peak_usage: total, // Simplified
464            pool_usage: HashMap::new(),
465            fragmentation_ratio: 0.0,
466            cache_hit_ratio: 0.95, // Estimated
467        }
468    }
469
470    /// Prefetch data for better performance
471    pub fn prefetch_region(&self, offset: usize, size: usize) -> Result<()> {
472        // Use madvise to hint the OS about access patterns
473        #[cfg(unix)]
474        {
475            if let Some(ref mmap) = *self.nodes_mmap.read() {
476                if offset < mmap.len() {
477                    let actual_size = size.min(mmap.len() - offset);
478                    unsafe {
479                        libc::madvise(
480                            mmap.as_ptr().add(offset) as *mut libc::c_void,
481                            actual_size,
482                            libc::MADV_WILLNEED,
483                        );
484                    }
485                }
486            }
487        }
488        
489        Ok(())
490    }
491}
492
493impl MemoryPool {
494    /// Create a new memory pool
495    pub fn new(alignment: usize, max_pool_size: usize) -> Self {
496        Self {
497            pools: HashMap::new(),
498            alignment,
499            max_pool_size,
500            allocated_bytes: Arc::new(Mutex::new(0)),
501            peak_usage: Arc::new(Mutex::new(0)),
502        }
503    }
504
505    /// Allocate aligned memory
506    pub fn allocate(&self, size: usize) -> Result<NonNull<u8>> {
507        let aligned_size = self.align_size(size);
508        
509        // Try to get from pool first
510        if let Some(pool) = self.pools.get(&aligned_size) {
511            let mut pool = pool.lock();
512            if let Some(ptr) = pool.pop() {
513                return Ok(ptr);
514            }
515        }
516        
517        // Allocate new memory
518        let layout = Layout::from_size_align(aligned_size, self.alignment)
519            .map_err(|e| crate::error::GraphError::graph_construction(&format!("Layout error: {}", e)))?;
520        
521        let ptr = unsafe { System.alloc(layout) };
522        if ptr.is_null() {
523            return Err(crate::error::GraphError::graph_construction("Allocation failed"));
524        }
525        
526        // Update statistics
527        let mut allocated = self.allocated_bytes.lock();
528        *allocated += aligned_size;
529        
530        let mut peak = self.peak_usage.lock();
531        if *allocated > *peak {
532            *peak = *allocated;
533        }
534        
535        Ok(unsafe { NonNull::new_unchecked(ptr) })
536    }
537
538    /// Deallocate memory back to pool
539    pub fn deallocate(&mut self, ptr: NonNull<u8>, size: usize) {
540        let aligned_size = self.align_size(size);
541        
542        // Return to pool if not full
543        let pool = self.pools.entry(aligned_size)
544            .or_insert_with(|| Arc::new(Mutex::new(Vec::new())));
545        
546        let mut pool = pool.lock();
547        if pool.len() < self.max_pool_size {
548            pool.push(ptr);
549        } else {
550            // Actually deallocate
551            let layout = Layout::from_size_align(aligned_size, self.alignment).unwrap();
552            unsafe { System.dealloc(ptr.as_ptr(), layout) };
553        }
554        
555        // Update statistics
556        let mut allocated = self.allocated_bytes.lock();
557        *allocated = allocated.saturating_sub(aligned_size);
558    }
559
560    /// Align size to cache line boundary
561    fn align_size(&self, size: usize) -> usize {
562        (size + self.alignment - 1) & !(self.alignment - 1)
563    }
564
565    /// Get memory usage statistics
566    pub fn get_stats(&self) -> MemoryStats {
567        let allocated = *self.allocated_bytes.lock();
568        let peak = *self.peak_usage.lock();
569        
570        let mut pool_usage = HashMap::new();
571        for (size, pool) in &self.pools {
572            pool_usage.insert(*size, pool.lock().len());
573        }
574        
575        MemoryStats {
576            total_allocated: allocated,
577            peak_usage: peak,
578            pool_usage,
579            fragmentation_ratio: 0.1, // Estimated
580            cache_hit_ratio: 0.85, // Estimated
581        }
582    }
583}
584
585impl CacheOptimizedStorage {
586    /// Create new cache-optimized storage
587    pub fn new(config: CacheConfig) -> Self {
588        Self {
589            l1_cache: Arc::new(RwLock::new(HashMap::new())),
590            l2_cache: Arc::new(RwLock::new(HashMap::new())),
591            cache_config: config,
592            hit_stats: Arc::new(Mutex::new(CacheStats::default())),
593        }
594    }
595
596    /// Get data from cache
597    pub fn get(&self, key: &CacheKey) -> Option<CacheData> {
598        // Try L1 cache first
599        {
600            let l1 = self.l1_cache.read();
601            if let Some(entry) = l1.get(key) {
602                self.update_access_stats(entry, true, 1);
603                return Some(entry.data.clone());
604            }
605        }
606        
607        // Try L2 cache
608        {
609            let l2 = self.l2_cache.read();
610            if let Some(entry) = l2.get(key) {
611                self.update_access_stats(entry, true, 2);
612                
613                // Promote to L1
614                self.promote_to_l1(key.clone(), entry.clone());
615                
616                return Some(entry.data.clone());
617            }
618        }
619        
620        // Cache miss
621        self.record_miss();
622        None
623    }
624
625    /// Put data into cache
626    pub fn put(&self, key: CacheKey, data: CacheData) {
627        let size = self.estimate_data_size(&data);
628        let entry = CacheEntry {
629            data,
630            size,
631            access_count: 1,
632            last_accessed: self.current_time(),
633            created_at: self.current_time(),
634            compressed: false,
635        };
636        
637        // Insert into L1 cache
638        {
639            let mut l1 = self.l1_cache.write();
640            
641            // Check if eviction needed
642            if l1.len() >= self.cache_config.l1_size {
643                self.evict_l1(&mut l1);
644            }
645            
646            l1.insert(key, entry);
647        }
648    }
649
650    /// Prefetch data based on strategy
651    pub fn prefetch(&self, keys: Vec<CacheKey>) {
652        match self.cache_config.prefetch_strategy {
653            PrefetchStrategy::Sequential => {
654                self.prefetch_sequential(keys);
655            }
656            PrefetchStrategy::Neighbors => {
657                self.prefetch_neighbors(keys);
658            }
659            PrefetchStrategy::RandomWalk => {
660                self.prefetch_random_walk(keys);
661            }
662            _ => {} // No prefetching
663        }
664    }
665
666    /// Get cache statistics
667    pub fn get_cache_stats(&self) -> CacheStats {
668        self.hit_stats.lock().clone()
669    }
670
671    /// Clear all caches
672    pub fn clear(&self) {
673        self.l1_cache.write().clear();
674        self.l2_cache.write().clear();
675        
676        let mut stats = self.hit_stats.lock();
677        *stats = CacheStats::default();
678    }
679
680    // Private helper methods
681    fn promote_to_l1(&self, key: CacheKey, entry: CacheEntry) {
682        let mut l1 = self.l1_cache.write();
683        
684        if l1.len() >= self.cache_config.l1_size {
685            self.evict_l1(&mut l1);
686        }
687        
688        l1.insert(key, entry);
689    }
690
691    fn evict_l1(&self, l1_cache: &mut HashMap<CacheKey, CacheEntry>) {
692        match self.cache_config.eviction_policy {
693            EvictionPolicy::LRU => {
694                if let Some((key, _)) = l1_cache.iter()
695                    .min_by_key(|(_, entry)| entry.last_accessed)
696                    .map(|(k, v)| (k.clone(), v.clone())) {
697                    
698                    let entry = l1_cache.remove(&key).unwrap();
699                    
700                    // Move to L2
701                    let mut l2 = self.l2_cache.write();
702                    if l2.len() >= self.cache_config.l2_size {
703                        self.evict_l2(&mut l2);
704                    }
705                    l2.insert(key, entry);
706                }
707            }
708            EvictionPolicy::LFU => {
709                if let Some((key, _)) = l1_cache.iter()
710                    .min_by_key(|(_, entry)| entry.access_count)
711                    .map(|(k, v)| (k.clone(), v.clone())) {
712                    
713                    l1_cache.remove(&key);
714                }
715            }
716            _ => {
717                // Random eviction
718                if let Some(key) = l1_cache.keys().next().cloned() {
719                    l1_cache.remove(&key);
720                }
721            }
722        }
723        
724        // Update stats
725        let mut stats = self.hit_stats.lock();
726        stats.evictions += 1;
727    }
728
729    fn evict_l2(&self, l2_cache: &mut HashMap<CacheKey, CacheEntry>) {
730        if let Some(key) = l2_cache.keys().next().cloned() {
731            l2_cache.remove(&key);
732        }
733    }
734
735    fn update_access_stats(&self, entry: &CacheEntry, hit: bool, level: u8) {
736        let mut stats = self.hit_stats.lock();
737        match level {
738            1 => if hit { stats.l1_hits += 1 } else { stats.l1_misses += 1 },
739            2 => if hit { stats.l2_hits += 1 } else { stats.l2_misses += 1 },
740            _ => {}
741        }
742    }
743
744    fn record_miss(&self) {
745        let mut stats = self.hit_stats.lock();
746        stats.l1_misses += 1;
747        stats.l2_misses += 1;
748    }
749
750    fn estimate_data_size(&self, data: &CacheData) -> usize {
751        match data {
752            CacheData::NodeData(bytes) => bytes.len(),
753            CacheData::EdgeData(bytes) => bytes.len(),
754            CacheData::ComputationResult(bytes) => bytes.len(),
755            CacheData::Metadata(bytes) => bytes.len(),
756        }
757    }
758
759    fn current_time(&self) -> u64 {
760        std::time::SystemTime::now()
761            .duration_since(std::time::UNIX_EPOCH)
762            .unwrap_or_default()
763            .as_secs()
764    }
765
766    fn prefetch_sequential(&self, _keys: Vec<CacheKey>) {
767        // Implement sequential prefetching
768    }
769
770    fn prefetch_neighbors(&self, _keys: Vec<CacheKey>) {
771        // Implement neighbor prefetching
772    }
773
774    fn prefetch_random_walk(&self, _keys: Vec<CacheKey>) {
775        // Implement random walk prefetching
776    }
777}
778
779impl Default for GraphMetadata {
780    fn default() -> Self {
781        Self {
782            num_nodes: 0,
783            num_edges: 0,
784            node_size: 0,
785            edge_size: 0,
786            version: 1,
787            created_at: 0,
788            last_modified: 0,
789            checksum: 0,
790        }
791    }
792}
793
794impl Default for CacheConfig {
795    fn default() -> Self {
796        Self {
797            l1_size: 1000,
798            l2_size: 10000,
799            eviction_policy: EvictionPolicy::LRU,
800            prefetch_strategy: PrefetchStrategy::Sequential,
801            compression_enabled: false,
802        }
803    }
804}
805
806impl Default for CacheStats {
807    fn default() -> Self {
808        Self {
809            l1_hits: 0,
810            l1_misses: 0,
811            l2_hits: 0,
812            l2_misses: 0,
813            evictions: 0,
814            prefetch_hits: 0,
815            memory_usage: 0,
816        }
817    }
818}
819
820// Implement serde for GraphMetadata
821impl serde::Serialize for GraphMetadata {
822    fn serialize<S>(&self, serializer: S) -> std::result::Result<S::Ok, S::Error>
823    where
824        S: serde::Serializer,
825    {
826        use serde::ser::SerializeStruct;
827        let mut state = serializer.serialize_struct("GraphMetadata", 8)?;
828        state.serialize_field("num_nodes", &self.num_nodes)?;
829        state.serialize_field("num_edges", &self.num_edges)?;
830        state.serialize_field("node_size", &self.node_size)?;
831        state.serialize_field("edge_size", &self.edge_size)?;
832        state.serialize_field("version", &self.version)?;
833        state.serialize_field("created_at", &self.created_at)?;
834        state.serialize_field("last_modified", &self.last_modified)?;
835        state.serialize_field("checksum", &self.checksum)?;
836        state.end()
837    }
838}
839
840impl<'de> serde::Deserialize<'de> for GraphMetadata {
841    fn deserialize<D>(deserializer: D) -> std::result::Result<Self, D::Error>
842    where
843        D: serde::Deserializer<'de>,
844    {
845        use serde::de::{self, MapAccess, Visitor};
846        use std::fmt;
847
848        struct GraphMetadataVisitor;
849
850        impl<'de> Visitor<'de> for GraphMetadataVisitor {
851            type Value = GraphMetadata;
852
853            fn expecting(&self, formatter: &mut fmt::Formatter) -> fmt::Result {
854                formatter.write_str("struct GraphMetadata")
855            }
856
857            fn visit_map<V>(self, mut map: V) -> std::result::Result<GraphMetadata, V::Error>
858            where
859                V: MapAccess<'de>,
860            {
861                let mut num_nodes = None;
862                let mut num_edges = None;
863                let mut node_size = None;
864                let mut edge_size = None;
865                let mut version = None;
866                let mut created_at = None;
867                let mut last_modified = None;
868                let mut checksum = None;
869
870                while let Some(key) = map.next_key()? {
871                    match key {
872                        "num_nodes" => {
873                            if num_nodes.is_some() {
874                                return Err(de::Error::duplicate_field("num_nodes"));
875                            }
876                            num_nodes = Some(map.next_value()?);
877                        }
878                        "num_edges" => {
879                            if num_edges.is_some() {
880                                return Err(de::Error::duplicate_field("num_edges"));
881                            }
882                            num_edges = Some(map.next_value()?);
883                        }
884                        "node_size" => {
885                            if node_size.is_some() {
886                                return Err(de::Error::duplicate_field("node_size"));
887                            }
888                            node_size = Some(map.next_value()?);
889                        }
890                        "edge_size" => {
891                            if edge_size.is_some() {
892                                return Err(de::Error::duplicate_field("edge_size"));
893                            }
894                            edge_size = Some(map.next_value()?);
895                        }
896                        "version" => {
897                            if version.is_some() {
898                                return Err(de::Error::duplicate_field("version"));
899                            }
900                            version = Some(map.next_value()?);
901                        }
902                        "created_at" => {
903                            if created_at.is_some() {
904                                return Err(de::Error::duplicate_field("created_at"));
905                            }
906                            created_at = Some(map.next_value()?);
907                        }
908                        "last_modified" => {
909                            if last_modified.is_some() {
910                                return Err(de::Error::duplicate_field("last_modified"));
911                            }
912                            last_modified = Some(map.next_value()?);
913                        }
914                        "checksum" => {
915                            if checksum.is_some() {
916                                return Err(de::Error::duplicate_field("checksum"));
917                            }
918                            checksum = Some(map.next_value()?);
919                        }
920                        _ => {
921                            let _: serde_json::Value = map.next_value()?;
922                        }
923                    }
924                }
925
926                Ok(GraphMetadata {
927                    num_nodes: num_nodes.unwrap_or(0),
928                    num_edges: num_edges.unwrap_or(0),
929                    node_size: node_size.unwrap_or(0),
930                    edge_size: edge_size.unwrap_or(0),
931                    version: version.unwrap_or(1),
932                    created_at: created_at.unwrap_or(0),
933                    last_modified: last_modified.unwrap_or(0),
934                    checksum: checksum.unwrap_or(0),
935                })
936            }
937        }
938
939        deserializer.deserialize_struct(
940            "GraphMetadata",
941            &["num_nodes", "num_edges", "node_size", "edge_size", "version", "created_at", "last_modified", "checksum"],
942            GraphMetadataVisitor,
943        )
944    }
945}
946
947#[cfg(test)]
948mod tests {
949    use super::*;
950    use tempfile::TempDir;
951    use crate::graph::ArrowGraph;
952    use arrow::array::{StringArray, Float64Array};
953    use arrow::record_batch::RecordBatch;
954    use arrow::datatypes::{Schema, Field, DataType};
955    use std::sync::Arc;
956
957    fn create_test_graph() -> Result<ArrowGraph> {
958        let nodes_schema = Arc::new(Schema::new(vec![
959            Field::new("id", DataType::Utf8, false),
960        ]));
961        let node_ids = StringArray::from(vec!["A", "B", "C", "D"]);
962        let nodes_batch = RecordBatch::try_new(
963            nodes_schema,
964            vec![Arc::new(node_ids)],
965        )?;
966
967        let edges_schema = Arc::new(Schema::new(vec![
968            Field::new("source", DataType::Utf8, false),
969            Field::new("target", DataType::Utf8, false),
970            Field::new("weight", DataType::Float64, false),
971        ]));
972        let sources = StringArray::from(vec!["A", "B", "C"]);
973        let targets = StringArray::from(vec!["B", "C", "D"]);
974        let weights = Float64Array::from(vec![1.0, 1.0, 1.0]);
975        let edges_batch = RecordBatch::try_new(
976            edges_schema,
977            vec![Arc::new(sources), Arc::new(targets), Arc::new(weights)],
978        )?;
979
980        ArrowGraph::new(nodes_batch, edges_batch)
981    }
982
983    #[test]
984    fn test_memory_mapped_graph_creation() {
985        let temp_dir = TempDir::new().unwrap();
986        let path = temp_dir.path().join("test_graph");
987        
988        let mmap_graph = MemoryMappedGraph::new(&path).unwrap();
989        assert_eq!(mmap_graph.metadata.num_nodes, 0);
990        assert_eq!(mmap_graph.metadata.num_edges, 0);
991    }
992
993    #[test]
994    fn test_memory_mapped_graph_from_arrow() {
995        let temp_dir = TempDir::new().unwrap();
996        let path = temp_dir.path().join("test_graph");
997        
998        let graph = create_test_graph().unwrap();
999        let mmap_graph = MemoryMappedGraph::from_arrow_graph(&graph, &path).unwrap();
1000        
1001        assert_eq!(mmap_graph.metadata.num_nodes, 4);
1002        assert_eq!(mmap_graph.metadata.num_edges, 3);
1003    }
1004
1005    #[test]
1006    fn test_memory_pool() {
1007        let pool = MemoryPool::new(64, 100);
1008        
1009        // Allocate some memory
1010        let ptr1 = pool.allocate(1024).unwrap();
1011        let ptr2 = pool.allocate(2048).unwrap();
1012        
1013        // Check stats
1014        let stats = pool.get_stats();
1015        assert!(stats.total_allocated > 0);
1016        assert!(stats.peak_usage > 0);
1017        
1018        // Deallocate
1019        pool.deallocate(ptr1, 1024);
1020        pool.deallocate(ptr2, 2048);
1021    }
1022
1023    #[test]
1024    fn test_cache_optimized_storage() {
1025        let config = CacheConfig::default();
1026        let cache = CacheOptimizedStorage::new(config);
1027        
1028        // Test cache operations
1029        let key = CacheKey::Node("A".to_string());
1030        let data = CacheData::NodeData(vec![1, 2, 3, 4]);
1031        
1032        // Should be cache miss initially
1033        assert!(cache.get(&key).is_none());
1034        
1035        // Put data
1036        cache.put(key.clone(), data.clone());
1037        
1038        // Should be cache hit now
1039        assert!(cache.get(&key).is_some());
1040        
1041        // Check stats
1042        let stats = cache.get_cache_stats();
1043        assert!(stats.l1_hits > 0);
1044    }
1045
1046    #[test]
1047    fn test_cache_eviction() {
1048        let config = CacheConfig {
1049            l1_size: 2,
1050            l2_size: 2,
1051            ..Default::default()
1052        };
1053        let cache = CacheOptimizedStorage::new(config);
1054        
1055        // Fill L1 cache beyond capacity
1056        for i in 0..5 {
1057            let key = CacheKey::Node(format!("Node{}", i));
1058            let data = CacheData::NodeData(vec![i as u8]);
1059            cache.put(key, data);
1060        }
1061        
1062        // Check that evictions occurred
1063        let stats = cache.get_cache_stats();
1064        assert!(stats.evictions > 0);
1065    }
1066
1067    #[test]
1068    fn test_memory_stats() {
1069        let temp_dir = TempDir::new().unwrap();
1070        let path = temp_dir.path().join("test_graph");
1071        
1072        let graph = create_test_graph().unwrap();
1073        let mmap_graph = MemoryMappedGraph::from_arrow_graph(&graph, &path).unwrap();
1074        
1075        let stats = mmap_graph.get_memory_stats();
1076        assert!(stats.total_allocated > 0);
1077    }
1078
1079    #[test]
1080    fn test_node_retrieval() {
1081        let temp_dir = TempDir::new().unwrap();
1082        let path = temp_dir.path().join("test_graph");
1083        
1084        let graph = create_test_graph().unwrap();
1085        let mmap_graph = MemoryMappedGraph::from_arrow_graph(&graph, &path).unwrap();
1086        
1087        // Test node retrieval
1088        let node_data = mmap_graph.get_node("A").unwrap();
1089        assert!(node_data.is_some());
1090        
1091        let missing_node = mmap_graph.get_node("Z").unwrap();
1092        assert!(missing_node.is_none());
1093    }
1094
1095    #[test]
1096    fn test_cache_key_types() {
1097        let cache = CacheOptimizedStorage::new(CacheConfig::default());
1098        
1099        // Test different cache key types
1100        let node_key = CacheKey::Node("A".to_string());
1101        let edge_key = CacheKey::Edge("A".to_string(), "B".to_string());
1102        let subgraph_key = CacheKey::Subgraph(vec!["A".to_string(), "B".to_string()]);
1103        
1104        let data = CacheData::NodeData(vec![1, 2, 3]);
1105        
1106        cache.put(node_key.clone(), data.clone());
1107        cache.put(edge_key.clone(), data.clone());
1108        cache.put(subgraph_key.clone(), data.clone());
1109        
1110        assert!(cache.get(&node_key).is_some());
1111        assert!(cache.get(&edge_key).is_some());
1112        assert!(cache.get(&subgraph_key).is_some());
1113    }
1114}