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#[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#[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#[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#[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#[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#[derive(Debug, Clone)]
72pub enum EvictionPolicy {
73 LRU, LFU, ARC, Random, FIFO, }
79
80#[derive(Debug, Clone)]
82pub enum PrefetchStrategy {
83 None,
84 Sequential, Neighbors, RandomWalk, ML, }
89
90#[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>), }
98
99#[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#[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#[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#[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
141pub struct GraphAllocator {
143 pool: Arc<MemoryPool>,
144}
145
146impl MemoryMappedGraph {
147 pub fn new<P: AsRef<Path>>(storage_path: P) -> Result<Self> {
149 let path = storage_path.as_ref().to_path_buf();
150
151 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 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 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 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 self.serialize_nodes(&graph.nodes)?;
191
192 self.serialize_edges(&graph.edges)?;
194
195 self.save_metadata()?;
197
198 Ok(())
199 }
200
201 fn serialize_nodes(&mut self, nodes_batch: &RecordBatch) -> Result<()> {
203 let nodes_file_path = self.storage_path.join("nodes.mmap");
204
205 let estimated_size = self.estimate_nodes_size(nodes_batch)?;
207
208 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 let mut mmap = unsafe { MmapOptions::new().map_mut(&file)? };
218
219 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 let len = node_data.len() as u32;
233 mmap[offset..offset + 4].copy_from_slice(&len.to_le_bytes());
234 offset += 4;
235
236 mmap[offset..offset + node_data.len()].copy_from_slice(node_data);
238 offset += node_data.len();
239 }
240 }
241
242 mmap.flush()?;
244
245 *self.nodes_mmap.write() = Some(mmap.make_read_only()?);
247
248 self.metadata.node_size = estimated_size;
249
250 Ok(())
251 }
252
253 fn serialize_edges(&mut self, edges_batch: &RecordBatch) -> Result<()> {
255 let edges_file_path = self.storage_path.join("edges.mmap");
256
257 let estimated_size = self.estimate_edges_size(edges_batch)?;
259
260 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 let mut mmap = unsafe { MmapOptions::new().map_mut(&file)? };
270
271 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 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 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 mmap[offset..offset + 8].copy_from_slice(&weight.to_le_bytes());
308 offset += 8;
309 }
310 }
311
312 mmap.flush()?;
314
315 *self.edges_mmap.write() = Some(mmap.make_read_only()?);
317
318 self.metadata.edge_size = estimated_size;
319
320 Ok(())
321 }
322
323 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; size += node_ids.value(i).len(); }
337 }
338
339 Ok((size as f64 * 1.1) as usize)
341 }
342
343 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(); size += 4 + target_ids.value(i).len(); size += 8; }
362 }
363
364 Ok((size as f64 * 1.1) as usize)
366 }
367
368 fn map_existing_files(&mut self) -> Result<()> {
370 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 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 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 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 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 let mut offset = 0;
414 while offset < mmap.len() {
415 if offset + 4 > mmap.len() {
416 break;
417 }
418
419 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 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 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, pool_usage: HashMap::new(),
465 fragmentation_ratio: 0.0,
466 cache_hit_ratio: 0.95, }
468 }
469
470 pub fn prefetch_region(&self, offset: usize, size: usize) -> Result<()> {
472 #[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 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 pub fn allocate(&self, size: usize) -> Result<NonNull<u8>> {
507 let aligned_size = self.align_size(size);
508
509 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 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 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 pub fn deallocate(&mut self, ptr: NonNull<u8>, size: usize) {
540 let aligned_size = self.align_size(size);
541
542 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 let layout = Layout::from_size_align(aligned_size, self.alignment).unwrap();
552 unsafe { System.dealloc(ptr.as_ptr(), layout) };
553 }
554
555 let mut allocated = self.allocated_bytes.lock();
557 *allocated = allocated.saturating_sub(aligned_size);
558 }
559
560 fn align_size(&self, size: usize) -> usize {
562 (size + self.alignment - 1) & !(self.alignment - 1)
563 }
564
565 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, cache_hit_ratio: 0.85, }
582 }
583}
584
585impl CacheOptimizedStorage {
586 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 pub fn get(&self, key: &CacheKey) -> Option<CacheData> {
598 {
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 {
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 self.promote_to_l1(key.clone(), entry.clone());
615
616 return Some(entry.data.clone());
617 }
618 }
619
620 self.record_miss();
622 None
623 }
624
625 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 {
639 let mut l1 = self.l1_cache.write();
640
641 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 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 _ => {} }
664 }
665
666 pub fn get_cache_stats(&self) -> CacheStats {
668 self.hit_stats.lock().clone()
669 }
670
671 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 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 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 if let Some(key) = l1_cache.keys().next().cloned() {
719 l1_cache.remove(&key);
720 }
721 }
722 }
723
724 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 }
769
770 fn prefetch_neighbors(&self, _keys: Vec<CacheKey>) {
771 }
773
774 fn prefetch_random_walk(&self, _keys: Vec<CacheKey>) {
775 }
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
820impl 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 let ptr1 = pool.allocate(1024).unwrap();
1011 let ptr2 = pool.allocate(2048).unwrap();
1012
1013 let stats = pool.get_stats();
1015 assert!(stats.total_allocated > 0);
1016 assert!(stats.peak_usage > 0);
1017
1018 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 let key = CacheKey::Node("A".to_string());
1030 let data = CacheData::NodeData(vec![1, 2, 3, 4]);
1031
1032 assert!(cache.get(&key).is_none());
1034
1035 cache.put(key.clone(), data.clone());
1037
1038 assert!(cache.get(&key).is_some());
1040
1041 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 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 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 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 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}