rushdb_storage/
owner_sets.rs

1//! Owner sets of declared-index leaves, cached per snapshot (READ-2).
2//!
3//! A page over an indexed filter used to read every index entry of each
4//! leaf on every request: millions of entries for a broad value on a large
5//! tenant. The set a leaf resolves to is a function of the snapshot's commit
6//! index, so it is kept here under `(key, commit index)` and reused by every
7//! reader of that snapshot; a commit moves readers to a new index and the
8//! older entries age out by LRU. Charged to the logical cache budget; a set
9//! that does not fit is simply not kept.
10
11use std::any::Any;
12use std::sync::{Arc, Mutex};
13
14use crate::cache::LogicalCache;
15use crate::resources::ByteBudget;
16use crate::StorageError;
17
18/// A cached owner set: opaque to storage (the query layer's set type).
19pub type OwnerSet = Arc<dyn Any + Send + Sync>;
20
21/// Bytes the cache may hold, a child of the logical cache budget.
22const OWNER_SET_CACHE_BYTES: usize = 64 * 1024 * 1024;
23
24/// The store's per-snapshot owner-set cache.
25pub struct OwnerSets {
26    cache: Mutex<Option<LogicalCache<(Vec<u8>, u64), OwnerSet>>>,
27}
28
29impl OwnerSets {
30    /// Build under `parent` (the logical cache budget); without one there is
31    /// no cache and every lookup misses.
32    ///
33    /// # Errors
34    /// Returns [`StorageError::InvalidConfiguration`] if the budget cannot be
35    /// created.
36    pub(crate) fn new(parent: Option<&ByteBudget>) -> Result<Self, StorageError> {
37        let invalid =
38            |e: crate::resources::ResourceError| StorageError::InvalidConfiguration(e.to_string());
39        let cache = match parent {
40            Some(parent) => {
41                let bytes = OWNER_SET_CACHE_BYTES.min(parent.snapshot().limit);
42                let budget = parent.child("owner-sets", bytes).map_err(invalid)?;
43                Some(LogicalCache::new(budget).map_err(invalid)?)
44            }
45            None => None,
46        };
47        Ok(Self {
48            cache: Mutex::new(cache),
49        })
50    }
51
52    /// The set cached for `key` at commit index `ci`.
53    #[must_use]
54    pub fn get(&self, key: &[u8], ci: u64) -> Option<OwnerSet> {
55        let mut guard = self.cache.lock().ok()?;
56        guard.as_mut()?.get(&(key.to_vec(), ci)).cloned()
57    }
58
59    /// Keep `set` (about `bytes` resident) for `key` at `ci`.
60    pub fn insert(&self, key: Vec<u8>, ci: u64, set: OwnerSet, bytes: usize) {
61        if let Ok(mut guard) = self.cache.lock() {
62            if let Some(cache) = guard.as_mut() {
63                // Sets of older snapshots are only read by readers still
64                // pinned there; the newest one is what later reads use.
65                cache.retain(|(_, at), _| *at >= ci);
66                cache.insert((key, ci), set, bytes);
67            }
68        }
69    }
70}
rushdb_storage/owner_sets.rs

rushdb_storage/
owner_sets.rs
