reddb_server/storage/unified/store/
secondary_index.rs

1//! Generation-bound secondary trees, built inside the existing checkpoint epoch.
2//!
3//! Resource sketch: conversion reads N source rows once per collection, writes
4//! O(N * indexes) compact postings, and holds one decoded row plus one 1 KiB key.
5//! Reopen loads only root handles (O(indexes)); queries use the shared page cache.
6//! No network I/O or per-index resident posting map is introduced here.
7use super::*;
8use reddb_file::native_resume::SecondaryIndexResume;
9
10/// Leave eight bytes for the physical entity identity in each B-tree key.
11pub use reddb_engine::storage::secondary_index_read::SECONDARY_KEY_BYTES_MAX;
12pub type SecondaryIndexRead =
13    reddb_engine::storage::secondary_index_read::SecondaryIndexRead<Pager>;
14
15#[derive(Clone, Debug, PartialEq, Eq)]
16pub struct SecondaryIndexIdentity {
17    pub codec_version: u32,
18    pub definition: [u8; 32],
19}
20
21/// The projector is immutable, deterministic and must not re-enter storage.
22/// It writes a canonical key into the supplied bounded scratch buffer. The
23/// store appends the physical identity, retaining all MVCC versions. Visibility
24/// and logical uniqueness remain the query/transaction layer's responsibility.
25pub trait SecondaryIndexProjector: Send + Sync {
26    fn project(
27        &self,
28        entity: &UnifiedEntity,
29        key: &mut [u8; SECONDARY_KEY_BYTES_MAX],
30    ) -> Result<Option<usize>, StoreError>;
31}
32
33#[derive(Clone)]
34pub(super) struct SecondaryIndexSpec {
35    pub(super) identity: SecondaryIndexIdentity,
36    pub(super) projector: Arc<dyn SecondaryIndexProjector>,
37    pub(super) deferred: bool,
38}
39
40pub(super) type SecondaryIndexSpecs = HashMap<String, BTreeMap<String, SecondaryIndexSpec>>;
41
42#[derive(Clone)]
43pub(super) struct PublishedSecondaryIndex {
44    pub identity: SecondaryIndexIdentity,
45    pub tree: Arc<BTree>,
46}
47
48impl PublishedSecondaryIndex {
49    pub(super) fn resume(&self) -> SecondaryIndexResume {
50        SecondaryIndexResume {
51            root_page: self.tree.root_page_id(),
52            codec_version: self.identity.codec_version,
53            definition: self.identity.definition,
54        }
55    }
56}
57
58impl PublishedKeyIndex {
59    pub(super) fn trees(&self) -> impl Iterator<Item = &Arc<BTree>> {
60        std::iter::once(&self.tree).chain(self.secondary.values().map(|index| &index.tree))
61    }
62}
63
64impl UnifiedStore {
65    /// Register a projection for subsequent checkpoints. The definition must
66    /// include method, ordered columns, uniqueness and any collation options.
67    /// A changed fingerprint is never allowed to reuse the previous tree.
68    pub fn register_secondary_index(
69        &self,
70        collection: &str,
71        name: &str,
72        identity: SecondaryIndexIdentity,
73        projector: Arc<dyn SecondaryIndexProjector>,
74    ) -> Result<(), StoreError> {
75        let collection = Self::canonical_collection(collection);
76        if name.is_empty() || name.len() > 1024 || identity.codec_version == 0 {
77            return Err(StoreError::Serialization(
78                "invalid secondary index identity".into(),
79            ));
80        }
81        let _publication = self.vault_publication_lock.lock();
82        if self.config.resident_memory_bytes_max.is_none()
83            || (self.commit.is_none() && !self.is_read_only())
84        {
85            return Err(StoreError::Serialization(
86                "secondary trees require disk residency and durable page ownership".into(),
87            ));
88        }
89        if !self.resident.collections.read().contains_key(collection) {
90            return Err(StoreError::CollectionNotFound(collection.into()));
91        }
92        let mut specs = self.secondary_index_specs.write();
93        let indices = specs.entry(collection.to_owned()).or_default();
94        if !indices.contains_key(name) && indices.len() >= 4096 {
95            return Err(StoreError::Serialization(
96                "too many secondary indexes".into(),
97            ));
98        }
99        indices.insert(
100            name.to_owned(),
101            SecondaryIndexSpec {
102                identity,
103                projector,
104                deferred: false,
105            },
106        );
107        Ok(())
108    }
109
110    /// Preparation is a full ownership/checkpoint epoch, not merely a catalog
111    /// flush. Reopened compatible roots need no source scan or publication.
112    pub(crate) fn checkpoint_missing_secondary_indices(&self) -> Result<(), StoreError> {
113        let missing = {
114            let _publication = self.vault_publication_lock.lock();
115            let specs = self.secondary_index_specs.read();
116            specs.iter().any(|(collection, indices)| {
117                indices.iter().any(|(name, spec)| {
118                    !spec.deferred
119                        && self
120                            .secondary_index(collection, name, &spec.identity)
121                            .is_none()
122                })
123            })
124        };
125        if missing {
126            self.persist()?;
127        }
128        Ok(())
129    }
130
131    /// Pin an immutable index only if both the data generation and definition
132    /// still match. A miss requires rebuilding or a correct scan fallback.
133    /// This is a physical base index: callers must merge the live WAL overlay.
134    pub fn secondary_index(
135        &self,
136        collection: &str,
137        name: &str,
138        identity: &SecondaryIndexIdentity,
139    ) -> Option<SecondaryIndexRead> {
140        let collection = Self::canonical_collection(collection);
141        let _publication = self.vault_publication_lock.lock();
142        let trees = self.btree_indices.read();
143        let data = trees.get(collection)?;
144        let keys = self.key_indices.read();
145        let generation = keys.get(collection)?;
146        if !generation.source.ptr_eq(&Arc::downgrade(data)) {
147            return None;
148        }
149        let index = generation.secondary.get(name)?;
150        (index.identity == *identity)
151            .then(|| SecondaryIndexRead::with_source(Arc::clone(&index.tree), Arc::clone(data)))
152    }
153
154    /// The next checkpoint publishes the removal. Already-pinned readers keep
155    /// their old immutable tree until the normal retirement protocol releases it.
156    pub fn unregister_secondary_index(&self, collection: &str, name: &str) {
157        let collection = Self::canonical_collection(collection);
158        let _publication = self.vault_publication_lock.lock();
159        if let Some(specs) = self.secondary_index_specs.write().get_mut(collection) {
160            specs.remove(name);
161        }
162        if let Some(index) = self
163            .key_indices
164            .write()
165            .get_mut(collection)
166            .and_then(|generation| generation.secondary.remove(name))
167        {
168            self.retire_btree(index.tree);
169            self.resident
170                .paged_registry_dirty
171                .store(true, Ordering::Release);
172        }
173    }
174
175    pub(super) fn checkpoint_secondary_indices(
176        &self,
177        collection: &str,
178        data: &Arc<BTree>,
179        previous: Option<&PublishedKeyIndex>,
180        pager: &Arc<Pager>,
181        limit_bytes: usize,
182    ) -> Result<BTreeMap<String, PublishedSecondaryIndex>, StoreError> {
183        use reddb_engine::storage::collection_read::{CollectionCursor, PagedCollection};
184        let same_generation = previous.filter(|index| index.source.ptr_eq(&Arc::downgrade(data)));
185        let mut next = same_generation
186            .map(|index| index.secondary.clone())
187            .unwrap_or_default();
188        let specs = self.secondary_index_specs.read();
189        let Some(specs) = specs.get(collection) else {
190            return Ok(next);
191        };
192        let mut builds = Vec::new();
193        for (name, spec) in specs {
194            if next
195                .get(name)
196                .is_some_and(|index| index.identity == spec.identity)
197            {
198                continue;
199            }
200            if spec.deferred {
201                continue;
202            }
203            builds.push((name, spec, Arc::new(BTree::new(Arc::clone(pager)))));
204        }
205        if builds.is_empty() {
206            return Ok(next);
207        }
208        let source = PagedCollection::new(Arc::clone(data));
209        let mut cursor = source.scan()?;
210        let mut key = [0; SECONDARY_KEY_BYTES_MAX];
211        while let Some(record) = cursor.next_record(limit_bytes)? {
212            let (entity, _) = record.decode_with_budget(self.format_version(), limit_bytes)?;
213            for (_, spec, tree) in &builds {
214                let Some(length) = spec.projector.project(&entity, &mut key)? else {
215                    continue;
216                };
217                if length > key.len() {
218                    return Err(StoreError::Serialization(
219                        "secondary key exceeds its bounded buffer".into(),
220                    ));
221                }
222                let mut posting = [0; SECONDARY_KEY_BYTES_MAX + 8];
223                posting[..length].copy_from_slice(&key[..length]);
224                posting[length..length + 8].copy_from_slice(&entity.id.raw().to_be_bytes());
225                tree.insert(&posting[..length + 8], &[])
226                    .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
227            }
228        }
229        for (name, spec, tree) in builds {
230            next.insert(
231                name.clone(),
232                PublishedSecondaryIndex {
233                    identity: spec.identity.clone(),
234                    tree,
235                },
236            );
237        }
238        Ok(next)
239    }
240}
241
242#[cfg(test)]
243mod tests;
reddb_server/storage/unified/store/secondary_index.rs

reddb_server/storage/unified/store/
secondary_index.rs
