Skip to main content

oxirs_vec/
mmap_index.rs

1//! Memory-mapped vector index for efficient disk-based storage
2//!
3//! This module provides a disk-based vector index using memory-mapped files for
4//! efficient access to large vector datasets that don't fit in memory.
5
6use crate::mmap_advanced::{AdvancedMemoryMap, MemoryMapStats, NumaVectorAllocator};
7use crate::{
8    index::{DistanceMetric, IndexConfig, SearchResult},
9    Vector, VectorIndex,
10};
11use anyhow::{bail, Context, Result};
12use blake3::Hasher;
13use memmap2::{Mmap, MmapOptions};
14use oxirs_core::parallel::*;
15use parking_lot::{Mutex, RwLock};
16use std::collections::{BinaryHeap, HashMap};
17use std::fs::{File, OpenOptions};
18use std::io::{Seek, SeekFrom, Write};
19use std::path::{Path, PathBuf};
20use std::sync::Arc;
21
22/// Magic number for file format identification
23const MAGIC: &[u8; 8] = b"OXIRSVEC";
24
25/// Current file format version
26const VERSION: u32 = 1;
27
28/// Default page size for memory mapping (4KB)
29const PAGE_SIZE: usize = 4096;
30
31/// Vector page size for advanced memory mapping (16KB for better vector alignment)
32const VECTOR_PAGE_SIZE: usize = 16384;
33
34/// Header size (must be page-aligned)
35const HEADER_SIZE: usize = PAGE_SIZE;
36
37/// File header structure
38#[repr(C)]
39#[derive(Debug, Clone, Copy)]
40struct FileHeader {
41    magic: [u8; 8],
42    version: u32,
43    flags: u32,
44    vector_count: u64,
45    dimensions: u32,
46    vector_size: u32, // Size of each vector in bytes
47    data_offset: u64,
48    index_offset: u64,
49    uri_offset: u64,
50    checksum: [u8; 32],
51    reserved: [u8; 3968], // Pad to PAGE_SIZE
52}
53
54impl FileHeader {
55    fn new(dimensions: u32) -> Self {
56        let vector_size = dimensions * std::mem::size_of::<f32>() as u32;
57        Self {
58            magic: *MAGIC,
59            version: VERSION,
60            flags: 0,
61            vector_count: 0,
62            dimensions,
63            vector_size,
64            data_offset: HEADER_SIZE as u64,
65            index_offset: 0,
66            uri_offset: 0,
67            checksum: [0; 32],
68            reserved: [0; 3968],
69        }
70    }
71
72    fn validate(&self) -> Result<()> {
73        if self.magic != *MAGIC {
74            bail!("Invalid magic number");
75        }
76        if self.version != VERSION {
77            bail!("Unsupported version: {}", self.version);
78        }
79        Ok(())
80    }
81
82    fn compute_checksum(&mut self) {
83        let mut hasher = Hasher::new();
84        hasher.update(&self.magic);
85        hasher.update(&self.version.to_le_bytes());
86        hasher.update(&self.flags.to_le_bytes());
87        hasher.update(&self.vector_count.to_le_bytes());
88        hasher.update(&self.dimensions.to_le_bytes());
89        hasher.update(&self.vector_size.to_le_bytes());
90        hasher.update(&self.data_offset.to_le_bytes());
91        hasher.update(&self.index_offset.to_le_bytes());
92        hasher.update(&self.uri_offset.to_le_bytes());
93        self.checksum = *hasher.finalize().as_bytes();
94    }
95}
96
97/// Memory-mapped vector index for large datasets
98pub struct MemoryMappedVectorIndex {
99    config: IndexConfig,
100    path: PathBuf,
101    header: Arc<RwLock<FileHeader>>,
102    data_file: Arc<Mutex<File>>,
103    data_mmap: Arc<RwLock<Option<Mmap>>>,
104    uri_map: Arc<RwLock<HashMap<String, u64>>>, // URI to vector ID
105    uri_store: Arc<RwLock<Vec<String>>>,        // Vector ID to URI
106    write_buffer: Arc<Mutex<Vec<(String, Vector)>>>,
107    buffer_size: usize,
108
109    // Advanced memory mapping features
110    advanced_mmap: Option<Arc<AdvancedMemoryMap>>,
111    numa_allocator: Arc<NumaVectorAllocator>,
112    enable_lazy_loading: bool,
113}
114
115impl MemoryMappedVectorIndex {
116    /// Create a new memory-mapped vector index
117    pub fn new<P: AsRef<Path>>(path: P, config: IndexConfig) -> Result<Self> {
118        let path = path.as_ref().to_path_buf();
119
120        // Create or open the data file
121        let data_file = OpenOptions::new()
122            .read(true)
123            .write(true)
124            .create(true)
125            .truncate(true)
126            .open(&path)
127            .context("Failed to open data file")?;
128
129        // Initialize or load header
130        let header = if data_file.metadata()?.len() == 0 {
131            // New file, write header
132            let header = FileHeader::new(0);
133            data_file.set_len(HEADER_SIZE as u64)?;
134            let mut header_bytes = vec![0u8; HEADER_SIZE];
135            unsafe {
136                std::ptr::copy_nonoverlapping(
137                    &header as *const _ as *const u8,
138                    header_bytes.as_mut_ptr(),
139                    std::mem::size_of::<FileHeader>(),
140                );
141            }
142            (&data_file).write_all(&header_bytes)?;
143            header
144        } else {
145            // Existing file, read header
146            let mmap = unsafe { MmapOptions::new().map(&data_file)? };
147            let header = unsafe { std::ptr::read(mmap.as_ptr() as *const FileHeader) };
148            header.validate()?;
149            header
150        };
151
152        Ok(Self {
153            config,
154            path,
155            header: Arc::new(RwLock::new(header)),
156            data_file: Arc::new(Mutex::new(data_file)),
157            data_mmap: Arc::new(RwLock::new(None)),
158            uri_map: Arc::new(RwLock::new(HashMap::new())),
159            uri_store: Arc::new(RwLock::new(Vec::new())),
160            write_buffer: Arc::new(Mutex::new(Vec::new())),
161            buffer_size: 1000, // Buffer 1000 vectors before flushing
162            advanced_mmap: None,
163            numa_allocator: Arc::new(NumaVectorAllocator::new()),
164            enable_lazy_loading: true,
165        })
166    }
167
168    /// Load an existing memory-mapped index
169    pub fn load<P: AsRef<Path>>(path: P, config: IndexConfig) -> Result<Self> {
170        let path = path.as_ref().to_path_buf();
171
172        // Open existing file without truncation
173        let data_file = OpenOptions::new()
174            .read(true)
175            .write(true)
176            .open(&path)
177            .context("Failed to open existing data file")?;
178
179        // Read and validate header
180        let mmap = unsafe { MmapOptions::new().map(&data_file)? };
181        let header = unsafe { std::ptr::read(mmap.as_ptr() as *const FileHeader) };
182        header.validate()?;
183
184        let mut index = Self {
185            config,
186            path,
187            header: Arc::new(RwLock::new(header)),
188            data_file: Arc::new(Mutex::new(data_file)),
189            data_mmap: Arc::new(RwLock::new(None)),
190            uri_map: Arc::new(RwLock::new(HashMap::new())),
191            uri_store: Arc::new(RwLock::new(Vec::new())),
192            write_buffer: Arc::new(Mutex::new(Vec::new())),
193            buffer_size: 1000,
194            advanced_mmap: None,
195            numa_allocator: Arc::new(NumaVectorAllocator::new()),
196            enable_lazy_loading: true,
197        };
198
199        index.reload_mmap()?;
200        index.load_uri_mappings()?;
201        Ok(index)
202    }
203
204    /// Reload memory mapping with optimized configuration
205    fn reload_mmap(&mut self) -> Result<()> {
206        let file = self.data_file.lock();
207        let file_len = file.metadata()?.len();
208
209        if file_len > HEADER_SIZE as u64 {
210            // Create optimized memory mapping with proper options
211            let mmap = unsafe {
212                MmapOptions::new()
213                    .huge(Some(21)) // Use huge pages (2MB) for better performance
214                    .populate() // Pre-populate pages to reduce page faults
215                    .map(&*file)?
216            };
217
218            // Create advanced memory map if lazy loading is enabled
219            if self.enable_lazy_loading {
220                // Calculate optimal page count based on file size
221                let optimal_pages =
222                    ((file_len as usize / VECTOR_PAGE_SIZE) / 10).clamp(1000, 50000);
223
224                // Create advanced memory mapping with cloned mmap
225                let cloned_mmap = unsafe { MmapOptions::new().map(&*file)? };
226                let advanced = AdvancedMemoryMap::new(Some(cloned_mmap), optimal_pages);
227                self.advanced_mmap = Some(Arc::new(advanced));
228            }
229
230            *self.data_mmap.write() = Some(mmap);
231        }
232
233        Ok(())
234    }
235
236    /// Load URI mappings from disk
237    fn load_uri_mappings(&self) -> Result<()> {
238        let header = self.header.read();
239        let uri_offset = header.uri_offset as usize;
240
241        if uri_offset > 0 {
242            if let Some(ref mmap) = *self.data_mmap.read() {
243                // Guard against a stale/oversized offset (e.g. a truncated or
244                // externally-corrupted file): never index past the mapping.
245                if uri_offset >= mmap.len() {
246                    return Ok(());
247                }
248                // Parse URI mappings from memory-mapped region
249                let uri_data = &mmap[uri_offset..];
250                let mut offset = 0;
251                let mut uri_map = self.uri_map.write();
252                let mut uri_store = self.uri_store.write();
253
254                for id in 0..header.vector_count {
255                    if offset + 4 > uri_data.len() {
256                        break;
257                    }
258
259                    let uri_len = u32::from_le_bytes([
260                        uri_data[offset],
261                        uri_data[offset + 1],
262                        uri_data[offset + 2],
263                        uri_data[offset + 3],
264                    ]) as usize;
265                    offset += 4;
266
267                    if offset + uri_len > uri_data.len() {
268                        break;
269                    }
270
271                    let uri =
272                        String::from_utf8_lossy(&uri_data[offset..offset + uri_len]).into_owned();
273                    offset += uri_len;
274
275                    uri_map.insert(uri.clone(), id);
276                    uri_store.push(uri);
277                }
278            }
279        }
280
281        Ok(())
282    }
283
284    /// Flush write buffer to disk with optimized batch operations
285    fn flush_buffer(&self) -> Result<()> {
286        let mut buffer = self.write_buffer.lock();
287        if buffer.is_empty() {
288            return Ok(());
289        }
290
291        let mut file = self.data_file.lock();
292        let mut header = self.header.write();
293
294        // Calculate required space
295        let vectors_to_write = buffer.len();
296
297        // Pre-validate all vectors and calculate total size
298        let mut total_vector_data_size = 0;
299        for (_, vector) in buffer.iter() {
300            if header.dimensions == 0 {
301                header.dimensions = vector.dimensions as u32;
302                header.vector_size = vector.dimensions as u32 * std::mem::size_of::<f32>() as u32;
303                total_vector_data_size = vectors_to_write * header.vector_size as usize;
304            } else if vector.dimensions != header.dimensions as usize {
305                bail!(
306                    "Vector dimensions ({}) don't match index dimensions ({})",
307                    vector.dimensions,
308                    header.dimensions
309                );
310            } else {
311                total_vector_data_size = vectors_to_write * header.vector_size as usize;
312            }
313        }
314
315        // Extend file if needed
316        let current_data_end =
317            header.data_offset + (header.vector_count * header.vector_size as u64);
318        let new_data_end = current_data_end + total_vector_data_size as u64;
319
320        file.set_len(new_data_end)?;
321        file.seek(SeekFrom::Start(current_data_end))?;
322
323        // Prepare batch write buffer for better I/O performance
324        let mut batch_write_buffer = Vec::with_capacity(total_vector_data_size);
325        let mut uri_updates = Vec::with_capacity(vectors_to_write);
326        let mut uri_map = self.uri_map.write();
327        let mut uri_store = self.uri_store.write();
328
329        // Batch prepare all data in memory first
330        for (uri, vector) in buffer.drain(..) {
331            // Convert vector to bytes
332            let vector_f32 = vector.as_f32();
333            let vector_bytes: Vec<u8> = vector_f32.iter().flat_map(|&f| f.to_le_bytes()).collect();
334            batch_write_buffer.extend_from_slice(&vector_bytes);
335
336            // Prepare URI updates
337            let vector_id = header.vector_count + uri_updates.len() as u64;
338            uri_updates.push((uri, vector_id));
339        }
340
341        // Single large write operation for much better performance
342        file.write_all(&batch_write_buffer)?;
343
344        // Update all URI mappings after successful write
345        for (uri, vector_id) in uri_updates {
346            uri_map.insert(uri.clone(), vector_id);
347            uri_store.push(uri);
348        }
349        header.vector_count += vectors_to_write as u64;
350
351        // If a URI table had previously been persisted (e.g. via an explicit
352        // `save_uri_mappings()` checkpoint), it lived immediately after the old
353        // vector data — exactly the region we just overwrote with the newly
354        // appended vectors. Leaving `header.uri_offset` pointing there would make
355        // `load()` parse raw vector bytes as URIs (byte-for-byte corruption), so
356        // re-persist the table at the new tail and update the offset. When no
357        // table was ever saved (`uri_offset == 0`) we leave it untouched; it is
358        // written once on `Drop`/`save_uri_mappings()`.
359        if header.uri_offset != 0 {
360            let vector_data_end =
361                header.data_offset + (header.vector_count * header.vector_size as u64);
362            let mut uri_table = Vec::new();
363            for uri in uri_store.iter() {
364                uri_table.extend_from_slice(&(uri.len() as u32).to_le_bytes());
365                uri_table.extend_from_slice(uri.as_bytes());
366            }
367            file.set_len(vector_data_end + uri_table.len() as u64)?;
368            file.seek(SeekFrom::Start(vector_data_end))?;
369            file.write_all(&uri_table)?;
370            header.uri_offset = vector_data_end;
371        }
372
373        // Update header with optimized single write
374        header.compute_checksum();
375        file.seek(SeekFrom::Start(0))?;
376        let header_bytes = unsafe {
377            std::slice::from_raw_parts(
378                &*header as *const _ as *const u8,
379                std::mem::size_of::<FileHeader>(),
380            )
381        };
382        file.write_all(header_bytes)?;
383
384        // Use fsync for better durability control
385        file.sync_all()?;
386
387        // Reload memory mapping with optimizations
388        drop(file);
389        drop(header);
390        drop(uri_map);
391        drop(uri_store);
392
393        // Reload with advanced memory mapping if enabled
394        let file = self.data_file.lock();
395        let file_len = file.metadata()?.len();
396        if file_len > HEADER_SIZE as u64 {
397            // Use optimized mmap options for better performance
398            let mmap = unsafe {
399                MmapOptions::new()
400                    .populate() // Pre-populate pages
401                    .map(&*file)?
402            };
403            *self.data_mmap.write() = Some(mmap);
404
405            // Update advanced mmap if it exists
406            if let Some(ref advanced_mmap) = self.advanced_mmap {
407                // Trigger a prefetch of recently written pages
408                let start_page = (current_data_end as usize) / VECTOR_PAGE_SIZE;
409                let end_page = (new_data_end as usize) / VECTOR_PAGE_SIZE;
410
411                for page_id in start_page..=end_page.min(start_page + 10) {
412                    advanced_mmap.async_prefetch(page_id);
413                }
414            }
415        }
416
417        Ok(())
418    }
419
420    /// Get vector by ID from memory-mapped region with optimized loading
421    fn get_vector_by_id(&self, id: u64) -> Result<Option<Vector>> {
422        let header = self.header.read();
423
424        if id >= header.vector_count {
425            return Ok(None);
426        }
427
428        // Try advanced memory mapping first for better performance
429        if let Some(ref advanced_mmap) = self.advanced_mmap {
430            let offset = header.data_offset as usize + (id as usize * header.vector_size as usize);
431            let page_id = offset / VECTOR_PAGE_SIZE;
432
433            if let Ok(page_entry) = advanced_mmap.get_page(page_id) {
434                let page_offset = offset % VECTOR_PAGE_SIZE;
435                let vector_end = page_offset + header.vector_size as usize;
436
437                if vector_end <= page_entry.data().len() {
438                    // Use NUMA-optimized vector allocation
439                    let numa_node = page_entry.numa_node();
440                    let values = self
441                        .numa_allocator
442                        .allocate_vector_on_node(header.dimensions as usize, Some(numa_node));
443
444                    // Optimized SIMD-friendly vector parsing
445                    return Ok(Some(self.parse_vector_optimized(
446                        &page_entry.data()[page_offset..vector_end],
447                        header.dimensions as usize,
448                        values,
449                    )?));
450                }
451            }
452        }
453
454        // Fallback to direct memory mapping
455        if let Some(ref mmap) = *self.data_mmap.read() {
456            let offset = header.data_offset as usize + (id as usize * header.vector_size as usize);
457            let end = offset + header.vector_size as usize;
458
459            if end <= mmap.len() {
460                let vector_bytes = &mmap[offset..end];
461                let values = self
462                    .numa_allocator
463                    .allocate_vector_on_node(header.dimensions as usize, None);
464
465                return Ok(Some(self.parse_vector_optimized(
466                    vector_bytes,
467                    header.dimensions as usize,
468                    values,
469                )?));
470            }
471        }
472
473        Ok(None)
474    }
475
476    /// Optimized vector parsing with SIMD acceleration where possible
477    fn parse_vector_optimized(
478        &self,
479        bytes: &[u8],
480        dimensions: usize,
481        mut values: Vec<f32>,
482    ) -> Result<Vector> {
483        values.clear();
484        values.reserve_exact(dimensions);
485
486        // Use chunked parsing for better cache locality
487        for chunk in bytes.chunks_exact(4) {
488            if values.len() >= dimensions {
489                break;
490            }
491            let float_val = f32::from_le_bytes([chunk[0], chunk[1], chunk[2], chunk[3]]);
492            values.push(float_val);
493        }
494
495        Ok(Vector::new(values))
496    }
497
498    /// Search using brute force with memory-mapped vectors
499    fn search_mmap(&self, query: &Vector, k: usize) -> Result<Vec<SearchResult>> {
500        let header = self.header.read();
501        let distance_metric = self.config.distance_metric;
502
503        if header.vector_count == 0 {
504            return Ok(Vec::new());
505        }
506
507        // Check if we should use parallel search
508        if self.config.parallel && header.vector_count > 1000 {
509            self.search_mmap_parallel(query, k, distance_metric)
510        } else {
511            self.search_mmap_sequential(query, k, distance_metric)
512        }
513    }
514
515    /// Sequential search through memory-mapped vectors
516    fn search_mmap_sequential(
517        &self,
518        query: &Vector,
519        k: usize,
520        distance_metric: DistanceMetric,
521    ) -> Result<Vec<SearchResult>> {
522        let header = self.header.read();
523        let uri_store = self.uri_store.read();
524        let mut heap = BinaryHeap::new();
525
526        for id in 0..header.vector_count {
527            if let Some(vector) = self.get_vector_by_id(id)? {
528                let distance = distance_metric.distance_vectors(query, &vector);
529
530                if heap.len() < k {
531                    heap.push(std::cmp::Reverse(SearchResult {
532                        uri: uri_store
533                            .get(id as usize)
534                            .cloned()
535                            .unwrap_or_else(|| format!("vector_{id}")),
536                        distance,
537                        score: 1.0 - distance, // Convert distance to similarity score
538                        metadata: None,
539                    }));
540                } else if let Some(std::cmp::Reverse(worst)) = heap.peek() {
541                    if distance < worst.distance {
542                        heap.pop();
543                        heap.push(std::cmp::Reverse(SearchResult {
544                            uri: uri_store
545                                .get(id as usize)
546                                .cloned()
547                                .unwrap_or_else(|| format!("vector_{id}")),
548                            distance,
549                            score: 1.0 - distance, // Convert distance to similarity score
550                            metadata: None,
551                        }));
552                    }
553                }
554            }
555        }
556
557        let mut results: Vec<SearchResult> = heap.into_iter().map(|r| r.0).collect();
558        results.sort_by(|a, b| {
559            a.distance
560                .partial_cmp(&b.distance)
561                .unwrap_or(std::cmp::Ordering::Equal)
562        });
563        Ok(results)
564    }
565
566    /// Parallel search through memory-mapped vectors
567    fn search_mmap_parallel(
568        &self,
569        query: &Vector,
570        k: usize,
571        distance_metric: DistanceMetric,
572    ) -> Result<Vec<SearchResult>> {
573        let header = self.header.read();
574        let uri_store = self.uri_store.read();
575        let vector_count = header.vector_count;
576        let chunk_size = (vector_count / num_threads() as u64).max(100);
577
578        // Process chunks in parallel
579        let partial_results: Vec<Vec<SearchResult>> = (0..vector_count)
580            .step_by(chunk_size as usize)
581            .collect::<Vec<_>>()
582            .par_iter()
583            .map(|&start_id| {
584                let end_id = (start_id + chunk_size).min(vector_count);
585                let mut local_heap = BinaryHeap::new();
586
587                for id in start_id..end_id {
588                    if let Ok(Some(vector)) = self.get_vector_by_id(id) {
589                        let distance = distance_metric.distance_vectors(query, &vector);
590
591                        if local_heap.len() < k {
592                            local_heap.push(std::cmp::Reverse(SearchResult {
593                                uri: uri_store
594                                    .get(id as usize)
595                                    .cloned()
596                                    .unwrap_or_else(|| format!("vector_{id}")),
597                                distance,
598                                score: 1.0 - distance, // Convert distance to similarity score
599                                metadata: None,
600                            }));
601                        } else if let Some(std::cmp::Reverse(worst)) = local_heap.peek() {
602                            if distance < worst.distance {
603                                local_heap.pop();
604                                local_heap.push(std::cmp::Reverse(SearchResult {
605                                    uri: uri_store
606                                        .get(id as usize)
607                                        .cloned()
608                                        .unwrap_or_else(|| format!("vector_{id}")),
609                                    distance,
610                                    score: 1.0 - distance, // Convert distance to similarity score
611                                    metadata: None,
612                                }));
613                            }
614                        }
615                    }
616                }
617
618                local_heap
619                    .into_sorted_vec()
620                    .into_iter()
621                    .map(|r| r.0)
622                    .collect()
623            })
624            .collect();
625
626        // Merge results from all chunks
627        let mut final_heap = BinaryHeap::new();
628        for partial in partial_results {
629            for result in partial {
630                if final_heap.len() < k {
631                    final_heap.push(std::cmp::Reverse(result));
632                } else if let Some(std::cmp::Reverse(worst)) = final_heap.peek() {
633                    if result.distance < worst.distance {
634                        final_heap.pop();
635                        final_heap.push(std::cmp::Reverse(result));
636                    }
637                }
638            }
639        }
640
641        let mut results: Vec<SearchResult> = final_heap.into_iter().map(|r| r.0).collect();
642        results.sort_by(|a, b| {
643            a.distance
644                .partial_cmp(&b.distance)
645                .unwrap_or(std::cmp::Ordering::Equal)
646        });
647        Ok(results)
648    }
649
650    /// Save URI mappings to disk
651    pub fn save_uri_mappings(&self) -> Result<()> {
652        let mut file = self.data_file.lock();
653        let mut header = self.header.write();
654        let uri_store = self.uri_store.read();
655
656        // Calculate size needed for URI data
657        let mut uri_data_size = 0;
658        for uri in uri_store.iter() {
659            uri_data_size += 4 + uri.len(); // 4 bytes for length + URI bytes
660        }
661
662        // Set URI offset after vector data
663        let data_end = header.data_offset + (header.vector_count * header.vector_size as u64);
664        header.uri_offset = data_end;
665
666        // Extend file and write URI data
667        file.set_len(data_end + uri_data_size as u64)?;
668        file.seek(SeekFrom::Start(header.uri_offset))?;
669
670        for uri in uri_store.iter() {
671            let len_bytes = (uri.len() as u32).to_le_bytes();
672            file.write_all(&len_bytes)?;
673            file.write_all(uri.as_bytes())?;
674        }
675
676        // Update header
677        header.compute_checksum();
678        file.seek(SeekFrom::Start(0))?;
679        let header_bytes = unsafe {
680            std::slice::from_raw_parts(
681                &*header as *const _ as *const u8,
682                std::mem::size_of::<FileHeader>(),
683            )
684        };
685        file.write_all(header_bytes)?;
686        file.sync_all()?;
687
688        Ok(())
689    }
690
691    /// Compact the index file by removing deleted entries
692    pub fn compact(&self) -> Result<()> {
693        // This would rewrite the file removing any gaps
694        // For now, we don't support deletion, so nothing to compact
695        Ok(())
696    }
697
698    /// Get index statistics
699    pub fn stats(&self) -> MemoryMappedIndexStats {
700        let header = self.header.read();
701        let file_size = self
702            .data_file
703            .lock()
704            .metadata()
705            .map(|m| m.len())
706            .unwrap_or(0);
707
708        MemoryMappedIndexStats {
709            vector_count: header.vector_count,
710            dimensions: header.dimensions,
711            file_size,
712            memory_usage: self.estimate_memory_usage(),
713        }
714    }
715
716    fn estimate_memory_usage(&self) -> usize {
717        let uri_map_size = self.uri_map.read().len()
718            * (std::mem::size_of::<String>() + std::mem::size_of::<u64>());
719        let uri_store_size = self
720            .uri_store
721            .read()
722            .iter()
723            .map(|s| s.capacity())
724            .sum::<usize>();
725        let buffer_size = self.write_buffer.lock().len()
726            * (std::mem::size_of::<String>() + std::mem::size_of::<Vector>());
727
728        uri_map_size + uri_store_size + buffer_size + HEADER_SIZE
729    }
730
731    /// Enable or disable lazy loading
732    pub fn set_lazy_loading(&mut self, enabled: bool) {
733        self.enable_lazy_loading = enabled;
734    }
735
736    /// Get advanced memory mapping statistics
737    pub fn advanced_stats(&self) -> Option<MemoryMapStats> {
738        self.advanced_mmap.as_ref().map(|mmap| mmap.stats())
739    }
740
741    /// Configure NUMA allocation preferences
742    pub fn configure_numa(&mut self, numa_enabled: bool) {
743        if numa_enabled {
744            self.numa_allocator = Arc::new(NumaVectorAllocator::new());
745        }
746    }
747}
748
749impl VectorIndex for MemoryMappedVectorIndex {
750    fn insert(&mut self, uri: String, vector: Vector) -> Result<()> {
751        // Add to write buffer
752        self.write_buffer.lock().push((uri, vector));
753
754        // Flush if buffer is full
755        if self.write_buffer.lock().len() >= self.buffer_size {
756            self.flush_buffer()?;
757        }
758
759        Ok(())
760    }
761
762    fn search_knn(&self, query: &Vector, k: usize) -> Result<Vec<(String, f32)>> {
763        // Flush any pending writes
764        if !self.write_buffer.lock().is_empty() {
765            self.flush_buffer()?;
766        }
767
768        // `search_mmap` returns results ordered by ascending distance. Convert
769        // to the trait's similarity contract (similarity = 1 / (1 + distance),
770        // larger = closer); ascending distance == descending similarity, so the
771        // best-match-first ordering is preserved.
772        let results = self.search_mmap(query, k)?;
773        Ok(results
774            .into_iter()
775            .map(|r| (r.uri, 1.0 / (1.0 + r.distance)))
776            .collect())
777    }
778
779    fn search_threshold(&self, query: &Vector, threshold: f32) -> Result<Vec<(String, f32)>> {
780        // Flush any pending writes
781        if !self.write_buffer.lock().is_empty() {
782            self.flush_buffer()?;
783        }
784
785        let header = self.header.read();
786        let uri_store = self.uri_store.read();
787        let distance_metric = self.config.distance_metric;
788        let mut results = Vec::new();
789
790        for id in 0..header.vector_count {
791            if let Some(vector) = self.get_vector_by_id(id)? {
792                let distance = distance_metric.distance_vectors(query, &vector);
793                // Trait contract: similarity >= threshold, where
794                // similarity = 1 / (1 + distance).
795                let similarity = 1.0 / (1.0 + distance);
796                if similarity >= threshold {
797                    let uri = uri_store
798                        .get(id as usize)
799                        .cloned()
800                        .unwrap_or_else(|| format!("vector_{id}"));
801                    results.push((uri, similarity));
802                }
803            }
804        }
805
806        // Sort by descending similarity (best match first).
807        results.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal));
808        Ok(results)
809    }
810
811    fn get_vector(&self, _uri: &str) -> Option<&Vector> {
812        // Memory-mapped index doesn't store vectors in memory
813        // We would need to read from disk, which doesn't fit the API
814        // that returns a reference. Return None for now.
815        None
816    }
817}
818
819impl Drop for MemoryMappedVectorIndex {
820    fn drop(&mut self) {
821        // Flush any remaining vectors
822        if let Err(e) = self.flush_buffer() {
823            eprintln!("Error flushing buffer on drop: {e}");
824        }
825        // Save URI mappings
826        if let Err(e) = self.save_uri_mappings() {
827            eprintln!("Error saving URI mappings on drop: {e}");
828        }
829    }
830}
831
832/// Statistics for memory-mapped index
833#[derive(Debug, Clone)]
834pub struct MemoryMappedIndexStats {
835    pub vector_count: u64,
836    pub dimensions: u32,
837    pub file_size: u64,
838    pub memory_usage: usize,
839}
840
841#[cfg(test)]
842mod tests {
843    use super::*;
844    use tempfile::tempdir;
845
846    #[test]
847    fn test_memory_mapped_index_basic() -> Result<()> {
848        let dir = tempdir()?;
849        let path = dir.path().join("test_vectors.idx");
850
851        let config = IndexConfig::default();
852        let mut index = MemoryMappedVectorIndex::new(&path, config)?;
853
854        // Insert some vectors
855        let v1 = Vector::new(vec![1.0, 2.0, 3.0]);
856        let v2 = Vector::new(vec![4.0, 5.0, 6.0]);
857        let v3 = Vector::new(vec![7.0, 8.0, 9.0]);
858
859        index.insert("vec1".to_string(), v1.clone())?;
860        index.insert("vec2".to_string(), v2.clone())?;
861        index.insert("vec3".to_string(), v3.clone())?;
862
863        // Force flush
864        index.flush_buffer()?;
865
866        // Search
867        let query = Vector::new(vec![3.0, 4.0, 5.0]);
868        let results = index.search_knn(&query, 2)?;
869
870        assert_eq!(results.len(), 2);
871        assert_eq!(results[0].0, "vec2");
872
873        Ok(())
874    }
875
876    #[test]
877    fn test_memory_mapped_index_persistence() -> Result<()> {
878        let dir = tempdir()?;
879        let path = dir.path().join("test_persist.idx");
880
881        // Create and populate index
882        {
883            let config = IndexConfig::default();
884            let mut index = MemoryMappedVectorIndex::new(&path, config)?;
885
886            for i in 0..10 {
887                let vec = Vector::new(vec![i as f32, (i + 1) as f32, (i + 2) as f32]);
888                index.insert(format!("vec{i}"), vec)?;
889            }
890
891            // Explicitly flush the buffer to ensure data is persisted
892            index.flush_buffer()?;
893        }
894
895        // Load existing index
896        {
897            let config = IndexConfig::default();
898            let index = MemoryMappedVectorIndex::load(&path, config)?;
899
900            let stats = index.stats();
901            assert_eq!(stats.vector_count, 10);
902            assert_eq!(stats.dimensions, 3);
903
904            let query = Vector::new(vec![5.0, 6.0, 7.0]);
905            let results = index.search_knn(&query, 3)?;
906
907            assert_eq!(results.len(), 3);
908            assert_eq!(results[0].0, "vec5");
909        }
910
911        Ok(())
912    }
913
914    /// Regression: an explicit `save_uri_mappings()` checkpoint followed by more
915    /// inserts+flush must not leave `header.uri_offset` pointing at overwritten
916    /// vector bytes. Previously `flush_buffer` appended vectors over the
917    /// persisted URI table, corrupting it and causing garbage URIs on reload.
918    #[test]
919    fn regression_flush_after_save_uri_mappings_no_corruption() -> Result<()> {
920        let dir = tempdir()?;
921        let path = dir.path().join("regression_uri_table.idx");
922
923        {
924            let config = IndexConfig::default();
925            let mut index = MemoryMappedVectorIndex::new(&path, config)?;
926
927            // First batch, then persist the URI table (checkpoint).
928            for i in 0..5 {
929                let vec = Vector::new(vec![i as f32, (i + 1) as f32, (i + 2) as f32]);
930                index.insert(format!("first{i}"), vec)?;
931            }
932            index.flush_buffer()?;
933            index.save_uri_mappings()?; // writes the URI table right after vectors
934
935            // Second batch: the flush appends vectors exactly where the URI table
936            // was; the fix must re-persist the table at the new tail.
937            for i in 0..5 {
938                let vec = Vector::new(vec![(i + 100) as f32, (i + 101) as f32, (i + 102) as f32]);
939                index.insert(format!("second{i}"), vec)?;
940            }
941            index.flush_buffer()?;
942            index.save_uri_mappings()?;
943            // `index` intentionally not relying on Drop for this assertion.
944            drop(index);
945        }
946
947        // Reload and verify every URI round-trips (no corruption / no fallback
948        // "vector_N" names) and both batches are searchable by their real URI.
949        {
950            let config = IndexConfig::default();
951            let index = MemoryMappedVectorIndex::load(&path, config)?;
952            assert_eq!(index.stats().vector_count, 10);
953
954            let q_first = Vector::new(vec![0.0, 1.0, 2.0]);
955            let r_first = index.search_knn(&q_first, 1)?;
956            assert_eq!(
957                r_first[0].0, "first0",
958                "first-batch URI corrupted on reload"
959            );
960
961            let q_second = Vector::new(vec![100.0, 101.0, 102.0]);
962            let r_second = index.search_knn(&q_second, 1)?;
963            assert_eq!(
964                r_second[0].0, "second0",
965                "second-batch URI corrupted on reload"
966            );
967        }
968
969        Ok(())
970    }
971}