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. Current secondary indexes omit retired
24/// MVCC versions; historical reads use the query layer's authoritative fallback.
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    /// Check a registered index against its immutable checkpoint source. This
66    /// is an operator pass: O(rows * log(postings)), one bounded row at a time.
67    /// Source read failures propagate; only derived-tree failures are findings.
68    pub(crate) fn verify_persisted_secondary_index(
69        &self,
70        collection: &str,
71        name: &str,
72        limit_bytes: usize,
73    ) -> Result<Option<String>, StoreError> {
74        use reddb_engine::storage::collection_read::{CollectionCursor, PagedCollection};
75        let _publication = self.vault_publication_lock.lock();
76        let specs = self.secondary_index_specs.read();
77        let spec = specs
78            .get(collection)
79            .and_then(|indices| indices.get(name))
80            .ok_or_else(|| StoreError::Serialization("index has no persisted projection".into()))?;
81        let trees = self.btree_indices.read();
82        let Some(data) = trees.get(collection) else {
83            return Ok(Some("collection has no checkpoint generation".into()));
84        };
85        let keys = self.key_indices.read();
86        let index = keys
87            .get(collection)
88            .filter(|generation| generation.source.ptr_eq(&Arc::downgrade(data)))
89            .and_then(|generation| generation.secondary.get(name))
90            .filter(|index| index.identity == spec.identity);
91        let Some(index) = index else {
92            return Ok(Some("missing or incompatible persisted index".into()));
93        };
94        let source = PagedCollection::new(Arc::clone(data));
95        let mut cursor = source.scan()?;
96        let mut key = [0; SECONDARY_KEY_BYTES_MAX];
97        let mut expected = 0u64;
98        while let Some(record) = cursor.next_record(limit_bytes)? {
99            let (entity, _) = record.decode_with_budget(self.format_version(), limit_bytes)?;
100            let Some(length) = spec.projector.project(&entity, &mut key)? else {
101                continue;
102            };
103            if length > key.len() {
104                return Err(StoreError::Serialization(
105                    "secondary key exceeds its bounded buffer".into(),
106                ));
107            }
108            let mut posting = [0; SECONDARY_KEY_BYTES_MAX + 8];
109            posting[..length].copy_from_slice(&key[..length]);
110            posting[length..length + 8].copy_from_slice(&entity.id.raw().to_be_bytes());
111            match index
112                .tree
113                .get_with_budget(&posting[..length + 8], limit_bytes)
114            {
115                Ok(Some(value)) if value.is_empty() => expected += 1,
116                Ok(_) => {
117                    return Ok(Some(format!(
118                        "missing or invalid posting for physical entity {}",
119                        entity.id.raw()
120                    )))
121                }
122                Err(error) => return Ok(Some(format!("index read failed: {error}"))),
123            }
124        }
125        let handle = SecondaryIndexRead::with_source(Arc::clone(&index.tree), Arc::clone(data));
126        let mut actual = 0u64;
127        if let Err(error) = handle.try_for_each(|_, _| {
128            actual += 1;
129            Ok(())
130        }) {
131            return Ok(Some(format!("index inventory failed: {error}")));
132        }
133        Ok((actual != expected)
134            .then(|| format!("posting count mismatch: expected {expected}, found {actual}")))
135    }
136
137    /// Register a projection for subsequent checkpoints. The definition must
138    /// include method, ordered columns, uniqueness and any collation options.
139    /// A changed fingerprint is never allowed to reuse the previous tree.
140    pub fn register_secondary_index(
141        &self,
142        collection: &str,
143        name: &str,
144        identity: SecondaryIndexIdentity,
145        projector: Arc<dyn SecondaryIndexProjector>,
146    ) -> Result<(), StoreError> {
147        let collection = Self::canonical_collection(collection);
148        if name.is_empty() || name.len() > 1024 || identity.codec_version == 0 {
149            return Err(StoreError::Serialization(
150                "invalid secondary index identity".into(),
151            ));
152        }
153        let _publication = self.vault_publication_lock.lock();
154        if self.config.resident_memory_bytes_max.is_none()
155            || (self.commit.is_none() && !self.is_read_only())
156        {
157            return Err(StoreError::Serialization(
158                "secondary trees require disk residency and durable page ownership".into(),
159            ));
160        }
161        if !self.resident.collections.read().contains_key(collection) {
162            return Err(StoreError::CollectionNotFound(collection.into()));
163        }
164        let mut specs = self.secondary_index_specs.write();
165        let indices = specs.entry(collection.to_owned()).or_default();
166        if !indices.contains_key(name) && indices.len() >= 4096 {
167            return Err(StoreError::Serialization(
168                "too many secondary indexes".into(),
169            ));
170        }
171        indices.insert(
172            name.to_owned(),
173            SecondaryIndexSpec {
174                identity,
175                projector,
176                deferred: false,
177            },
178        );
179        Ok(())
180    }
181
182    /// Preparation is a full ownership/checkpoint epoch, not merely a catalog
183    /// flush. Reopened compatible roots need no source scan or publication.
184    pub(crate) fn checkpoint_missing_secondary_indices(&self) -> Result<(), StoreError> {
185        let missing = {
186            let _publication = self.vault_publication_lock.lock();
187            let specs = self.secondary_index_specs.read();
188            specs.iter().any(|(collection, indices)| {
189                indices.iter().any(|(name, spec)| {
190                    !spec.deferred
191                        && self
192                            .secondary_index(collection, name, &spec.identity)
193                            .is_none()
194                })
195            })
196        };
197        if missing {
198            self.persist()?;
199        }
200        Ok(())
201    }
202
203    /// Pin an immutable index only if both the data generation and definition
204    /// still match. A miss requires rebuilding or a correct scan fallback.
205    /// This is a physical base index: callers must merge the live WAL overlay.
206    pub fn secondary_index(
207        &self,
208        collection: &str,
209        name: &str,
210        identity: &SecondaryIndexIdentity,
211    ) -> Option<SecondaryIndexRead> {
212        let collection = Self::canonical_collection(collection);
213        let _publication = self.vault_publication_lock.lock();
214        let trees = self.btree_indices.read();
215        let data = trees.get(collection)?;
216        let keys = self.key_indices.read();
217        let generation = keys.get(collection)?;
218        if !generation.source.ptr_eq(&Arc::downgrade(data)) {
219            return None;
220        }
221        let index = generation.secondary.get(name)?;
222        (index.identity == *identity)
223            .then(|| SecondaryIndexRead::with_source(Arc::clone(&index.tree), Arc::clone(data)))
224    }
225
226    /// The next checkpoint publishes the removal. Already-pinned readers keep
227    /// their old immutable tree until the normal retirement protocol releases it.
228    pub fn unregister_secondary_index(&self, collection: &str, name: &str) {
229        let collection = Self::canonical_collection(collection);
230        let _publication = self.vault_publication_lock.lock();
231        if let Some(specs) = self.secondary_index_specs.write().get_mut(collection) {
232            specs.remove(name);
233        }
234        if let Some(index) = self
235            .key_indices
236            .write()
237            .get_mut(collection)
238            .and_then(|generation| generation.secondary.remove(name))
239        {
240            self.retire_btree(index.tree);
241            self.resident
242                .paged_registry_dirty
243                .store(true, Ordering::Release);
244        }
245    }
246
247    pub(super) fn checkpoint_secondary_indices(
248        &self,
249        collection: &str,
250        data: &Arc<BTree>,
251        previous: Option<&PublishedKeyIndex>,
252        pager: &Arc<Pager>,
253        limit_bytes: usize,
254    ) -> Result<BTreeMap<String, PublishedSecondaryIndex>, StoreError> {
255        use reddb_engine::storage::collection_read::{CollectionCursor, PagedCollection};
256        let same_generation = previous.filter(|index| index.source.ptr_eq(&Arc::downgrade(data)));
257        let mut next = same_generation
258            .map(|index| index.secondary.clone())
259            .unwrap_or_default();
260        let specs = self.secondary_index_specs.read();
261        let Some(specs) = specs.get(collection) else {
262            return Ok(next);
263        };
264        let mut builds = Vec::new();
265        for (name, spec) in specs {
266            if next
267                .get(name)
268                .is_some_and(|index| index.identity == spec.identity)
269            {
270                continue;
271            }
272            if spec.deferred {
273                continue;
274            }
275            if name == super::count_index::NAME {
276                if let Some(tree) = self.checkpoint_count_projection_seed(
277                    collection,
278                    data.root_page_id(),
279                    &spec.identity,
280                    pager,
281                )? {
282                    next.insert(
283                        name.clone(),
284                        PublishedSecondaryIndex {
285                            identity: spec.identity.clone(),
286                            tree,
287                        },
288                    );
289                    continue;
290                }
291            }
292            builds.push((name, spec, Self::new_derived_btree(pager)?));
293        }
294        if builds.is_empty() {
295            return Ok(next);
296        }
297        let source = PagedCollection::new(Arc::clone(data));
298        let mut cursor = source.scan()?;
299        let mut key = [0; SECONDARY_KEY_BYTES_MAX];
300        while let Some(record) = cursor.next_record(limit_bytes)? {
301            let (entity, _) = record.decode_with_budget(self.format_version(), limit_bytes)?;
302            for (_, spec, tree) in &builds {
303                let Some(length) = spec.projector.project(&entity, &mut key)? else {
304                    continue;
305                };
306                if length > key.len() {
307                    return Err(StoreError::Serialization(
308                        "secondary key exceeds its bounded buffer".into(),
309                    ));
310                }
311                let mut posting = [0; SECONDARY_KEY_BYTES_MAX + 8];
312                posting[..length].copy_from_slice(&key[..length]);
313                posting[length..length + 8].copy_from_slice(&entity.id.raw().to_be_bytes());
314                tree.insert(&posting[..length + 8], &[])
315                    .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
316            }
317        }
318        for (name, spec, tree) in builds {
319            next.insert(
320                name.clone(),
321                PublishedSecondaryIndex {
322                    identity: spec.identity.clone(),
323                    tree,
324                },
325            );
326        }
327        Ok(next)
328    }
329}
330
331#[cfg(test)]
332mod tests;
reddb_server/storage/unified/store/secondary_index.rs

reddb_server/storage/unified/store/
secondary_index.rs
