mmdb 3.2.4

The storage engine behind vsdb — a pure-Rust LSM-Tree key-value store
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105

//! Table cache: caches open SST file readers to avoid repeated file open/parse.

use std::path::{Path, PathBuf};
use std::sync::Arc;

use crate::cache::block_cache::BlockCache;
use crate::error::Result;
use crate::sst::table_reader::TableReader;
use crate::stats::DbStats;
use ruc::*;

/// Cache for open TableReader instances.
pub struct TableCache {
    db_path: PathBuf,
    inner: moka::sync::Cache<u64, Arc<TableReader>>,
    block_cache: Option<Arc<BlockCache>>,
    stats: Option<Arc<DbStats>>,
}

impl TableCache {
    /// Create a new table cache.
    /// `max_open_files` is the maximum number of SST files to keep open.
    pub fn new(db_path: &Path, max_open_files: u64, block_cache: Option<Arc<BlockCache>>) -> Self {
        Self::new_with_stats(db_path, max_open_files, block_cache, None)
    }

    /// Create a new table cache with optional stats.
    pub fn new_with_stats(
        db_path: &Path,
        max_open_files: u64,
        block_cache: Option<Arc<BlockCache>>,
        stats: Option<Arc<DbStats>>,
    ) -> Self {
        Self {
            db_path: db_path.to_path_buf(),
            inner: moka::sync::Cache::builder()
                .max_capacity(max_open_files)
                .build(),
            block_cache,
            stats,
        }
    }

    /// Get or open a table reader for the given file number.
    /// Uses moka's `try_get_with` to coalesce concurrent loads for the same file.
    pub fn get_reader(&self, file_number: u64) -> Result<Arc<TableReader>> {
        let db_path = self.db_path.clone();
        let block_cache = self.block_cache.clone();
        let stats = self.stats.clone();
        self.inner
            .try_get_with::<_, String>(file_number, || {
                let path = db_path.join(format!("{:06}.sst", file_number));
                let reader = Arc::new(
                    TableReader::open_with_all(&path, file_number, block_cache, stats)
                        .map_err(|e| format!("{}", e))?,
                );
                Ok(reader)
            })
            .map_err(|e| eg!(format!("table cache load failed: {}", e)))
    }

    /// Evict a file from the cache.
    pub fn evict(&self, file_number: u64) {
        self.inner.invalidate(&file_number);
    }

    /// Number of cached readers.
    pub fn entry_count(&self) -> u64 {
        self.inner.entry_count()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::sst::table_builder::{TableBuildOptions, TableBuilder};

    #[test]
    fn test_table_cache() {
        let dir = tempfile::tempdir().unwrap();

        // Create an SST file
        let path = dir.path().join("000001.sst");
        let mut builder = TableBuilder::new(&path, TableBuildOptions::default()).unwrap();
        builder.add(b"key", b"value").unwrap();
        builder.finish().unwrap();

        let cache = TableCache::new(dir.path(), 100, None);

        // First access opens the file
        let reader = cache.get_reader(1).unwrap();
        let val = reader.get(b"key").unwrap();
        assert_eq!(val, Some(b"value".to_vec()));

        // Second access should hit cache
        let reader2 = cache.get_reader(1).unwrap();
        assert!(Arc::ptr_eq(&reader, &reader2));

        // Evict and re-open
        cache.evict(1);
        cache.inner.run_pending_tasks();
        let reader3 = cache.get_reader(1).unwrap();
        assert!(!Arc::ptr_eq(&reader, &reader3));
    }
}