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
15pub(crate) const RUNTIME_SECONDARY_INDEX_CODEC_VERSION: u32 = 3;
16
17#[cfg(test)]
18std::thread_local! {
19    static LEGACY_RUNTIME_SECONDARY_INDEX_IDENTITY: std::cell::Cell<bool> =
20        const { std::cell::Cell::new(false) };
21    static CHECKPOINT_SECONDARY_INDEX_SKIP_ENTITY:
22        std::cell::RefCell<Option<(String, u64)>> = const { std::cell::RefCell::new(None) };
23}
24
25pub(crate) fn runtime_secondary_index_codec_version() -> u32 {
26    #[cfg(test)]
27    if LEGACY_RUNTIME_SECONDARY_INDEX_IDENTITY.with(std::cell::Cell::get) {
28        return 2;
29    }
30    RUNTIME_SECONDARY_INDEX_CODEC_VERSION
31}
32
33#[cfg(test)]
34pub(crate) struct LegacyRuntimeSecondaryIndexIdentityGuard;
35
36#[cfg(test)]
37impl Drop for LegacyRuntimeSecondaryIndexIdentityGuard {
38    fn drop(&mut self) {
39        LEGACY_RUNTIME_SECONDARY_INDEX_IDENTITY.with(|legacy| legacy.set(false));
40    }
41}
42
43#[cfg(test)]
44pub(crate) fn use_legacy_runtime_secondary_index_identity(
45) -> LegacyRuntimeSecondaryIndexIdentityGuard {
46    assert!(
47        !LEGACY_RUNTIME_SECONDARY_INDEX_IDENTITY.with(|legacy| legacy.replace(true)),
48        "legacy runtime secondary-index identity is already active"
49    );
50    LegacyRuntimeSecondaryIndexIdentityGuard
51}
52
53#[cfg(test)]
54pub(crate) struct CheckpointSecondaryIndexSkipEntityGuard;
55
56#[cfg(test)]
57impl Drop for CheckpointSecondaryIndexSkipEntityGuard {
58    fn drop(&mut self) {
59        CHECKPOINT_SECONDARY_INDEX_SKIP_ENTITY.with(|skip| {
60            skip.borrow_mut().take();
61        });
62    }
63}
64
65#[cfg(test)]
66pub(crate) fn checkpoint_secondary_indices_without_entity(
67    collection: &str,
68    entity_id: u64,
69) -> CheckpointSecondaryIndexSkipEntityGuard {
70    CHECKPOINT_SECONDARY_INDEX_SKIP_ENTITY.with(|skip| {
71        let previous = skip.borrow_mut().replace((
72            UnifiedStore::canonical_collection(collection).to_string(),
73            entity_id,
74        ));
75        assert!(
76            previous.is_none(),
77            "secondary-index checkpoint skip is already active"
78        );
79    });
80    CheckpointSecondaryIndexSkipEntityGuard
81}
82
83#[cfg(test)]
84fn checkpoint_skips_secondary_index_entity(collection: &str, entity_id: u64) -> bool {
85    CHECKPOINT_SECONDARY_INDEX_SKIP_ENTITY.with(|skip| {
86        skip.borrow()
87            .as_ref()
88            .is_some_and(|(candidate, id)| candidate == collection && *id == entity_id)
89    })
90}
91
92#[derive(Clone, Debug, PartialEq, Eq)]
93pub struct SecondaryIndexIdentity {
94    pub codec_version: u32,
95    pub definition: [u8; 32],
96}
97
98/// The projector is immutable, deterministic and must not re-enter storage.
99/// It writes a canonical key into the supplied bounded scratch buffer. The
100/// store appends the physical identity. Current secondary indexes omit retired
101/// MVCC versions; historical reads use the query layer's authoritative fallback.
102pub trait SecondaryIndexProjector: Send + Sync {
103    fn project(
104        &self,
105        entity: &UnifiedEntity,
106        key: &mut [u8; SECONDARY_KEY_BYTES_MAX],
107    ) -> Result<Option<usize>, StoreError>;
108}
109
110#[derive(Clone)]
111pub(super) struct SecondaryIndexSpec {
112    pub(super) identity: SecondaryIndexIdentity,
113    pub(super) projector: Arc<dyn SecondaryIndexProjector>,
114    pub(super) deferred: bool,
115}
116
117pub(super) type SecondaryIndexSpecs = HashMap<String, BTreeMap<String, SecondaryIndexSpec>>;
118
119#[derive(Clone)]
120pub(super) struct PublishedSecondaryIndex {
121    pub identity: SecondaryIndexIdentity,
122    pub tree: Arc<BTree>,
123}
124
125impl PublishedSecondaryIndex {
126    pub(super) fn resume(&self) -> SecondaryIndexResume {
127        SecondaryIndexResume {
128            root_page: self.tree.root_page_id(),
129            codec_version: self.identity.codec_version,
130            definition: self.identity.definition,
131        }
132    }
133}
134
135impl PublishedKeyIndex {
136    pub(super) fn trees(&self) -> impl Iterator<Item = &Arc<BTree>> {
137        std::iter::once(&self.tree).chain(self.secondary.values().map(|index| &index.tree))
138    }
139}
140
141impl UnifiedStore {
142    #[cfg(test)]
143    pub(crate) fn rebuild_secondary_indices_on_next_checkpoint(
144        &self,
145        collection: &str,
146        names: &[&str],
147    ) {
148        let collection = Self::canonical_collection(collection);
149        let _publication = self.vault_publication_lock.lock();
150        let removed = {
151            let mut generations = self.key_indices.write();
152            let generation = generations
153                .get_mut(collection)
154                .expect("secondary-index checkpoint fixture generation");
155            names
156                .iter()
157                .filter_map(|name| generation.secondary.remove(*name))
158                .collect::<Vec<_>>()
159        };
160        assert_eq!(
161            removed.len(),
162            names.len(),
163            "secondary-index checkpoint fixture must remove every requested artifact"
164        );
165        for index in removed {
166            self.retire_btree(index.tree);
167        }
168        self.resident
169            .paged_registry_dirty
170            .store(true, Ordering::Release);
171    }
172
173    /// Check a registered index against its immutable checkpoint source. This
174    /// is an operator pass: O(rows * log(postings)), one bounded row at a time.
175    /// Source read failures propagate; only derived-tree failures are findings.
176    pub(crate) fn verify_persisted_secondary_index(
177        &self,
178        collection: &str,
179        name: &str,
180        limit_bytes: usize,
181    ) -> Result<Option<String>, StoreError> {
182        use reddb_engine::storage::collection_read::{CollectionCursor, PagedCollection};
183        let _publication = self.vault_publication_lock.lock();
184        let specs = self.secondary_index_specs.read();
185        let spec = specs
186            .get(collection)
187            .and_then(|indices| indices.get(name))
188            .ok_or_else(|| StoreError::Serialization("index has no persisted projection".into()))?;
189        let trees = self.btree_indices.read();
190        let Some(data) = trees.get(collection) else {
191            return Ok(Some("collection has no checkpoint generation".into()));
192        };
193        let keys = self.key_indices.read();
194        let index = keys
195            .get(collection)
196            .filter(|generation| generation.source.ptr_eq(&Arc::downgrade(data)))
197            .and_then(|generation| generation.secondary.get(name))
198            .filter(|index| index.identity == spec.identity);
199        let Some(index) = index else {
200            return Ok(Some("missing or incompatible persisted index".into()));
201        };
202        let source = PagedCollection::new(Arc::clone(data));
203        let mut cursor = source.scan()?;
204        let mut key = [0; SECONDARY_KEY_BYTES_MAX];
205        let mut expected = 0u64;
206        while let Some(record) = cursor.next_record(limit_bytes)? {
207            let (entity, _) = record.decode_with_budget(self.format_version(), limit_bytes)?;
208            let Some(length) = spec.projector.project(&entity, &mut key)? else {
209                continue;
210            };
211            if length > key.len() {
212                return Err(StoreError::Serialization(
213                    "secondary key exceeds its bounded buffer".into(),
214                ));
215            }
216            let mut posting = [0; SECONDARY_KEY_BYTES_MAX + 8];
217            posting[..length].copy_from_slice(&key[..length]);
218            posting[length..length + 8].copy_from_slice(&entity.id.raw().to_be_bytes());
219            match index
220                .tree
221                .get_with_budget(&posting[..length + 8], limit_bytes)
222            {
223                Ok(Some(value)) if value.is_empty() => expected += 1,
224                Ok(_) => {
225                    return Ok(Some(format!(
226                        "missing or invalid posting for physical entity {}",
227                        entity.id.raw()
228                    )))
229                }
230                Err(error) => return Ok(Some(format!("index read failed: {error}"))),
231            }
232        }
233        let handle = SecondaryIndexRead::with_source(Arc::clone(&index.tree), Arc::clone(data));
234        let mut actual = 0u64;
235        if let Err(error) = handle.try_for_each(|_, _| {
236            actual += 1;
237            Ok(())
238        }) {
239            return Ok(Some(format!("index inventory failed: {error}")));
240        }
241        Ok((actual != expected)
242            .then(|| format!("posting count mismatch: expected {expected}, found {actual}")))
243    }
244
245    /// Register a projection for subsequent checkpoints. The definition must
246    /// include method, ordered columns, uniqueness and any collation options.
247    /// A changed fingerprint is never allowed to reuse the previous tree.
248    pub fn register_secondary_index(
249        &self,
250        collection: &str,
251        name: &str,
252        identity: SecondaryIndexIdentity,
253        projector: Arc<dyn SecondaryIndexProjector>,
254    ) -> Result<(), StoreError> {
255        let collection = Self::canonical_collection(collection);
256        if name.is_empty() || name.len() > 1024 || identity.codec_version == 0 {
257            return Err(StoreError::Serialization(
258                "invalid secondary index identity".into(),
259            ));
260        }
261        let _publication = self.vault_publication_lock.lock();
262        if self.config.resident_memory_bytes_max.is_none()
263            || (self.commit.is_none() && !self.is_read_only())
264        {
265            return Err(StoreError::Serialization(
266                "secondary trees require disk residency and durable page ownership".into(),
267            ));
268        }
269        if !self.resident.collections.read().contains_key(collection) {
270            return Err(StoreError::CollectionNotFound(collection.into()));
271        }
272        let mut specs = self.secondary_index_specs.write();
273        let indices = specs.entry(collection.to_owned()).or_default();
274        if !indices.contains_key(name) && indices.len() >= 4096 {
275            return Err(StoreError::Serialization(
276                "too many secondary indexes".into(),
277            ));
278        }
279        indices.insert(
280            name.to_owned(),
281            SecondaryIndexSpec {
282                identity,
283                projector,
284                deferred: false,
285            },
286        );
287        Ok(())
288    }
289
290    /// Preparation is a full ownership/checkpoint epoch, not merely a catalog
291    /// flush. Reopened compatible roots need no source scan or publication.
292    pub(crate) fn checkpoint_missing_secondary_indices(&self) -> Result<(), StoreError> {
293        let missing = {
294            let _publication = self.vault_publication_lock.lock();
295            let specs = self.secondary_index_specs.read();
296            specs.iter().any(|(collection, indices)| {
297                indices.iter().any(|(name, spec)| {
298                    !spec.deferred
299                        && self
300                            .secondary_index(collection, name, &spec.identity)
301                            .is_none()
302                })
303            })
304        };
305        if missing {
306            self.persist()?;
307        }
308        Ok(())
309    }
310
311    /// Pin an immutable index only if both the data generation and definition
312    /// still match. A miss requires rebuilding or a correct scan fallback.
313    /// This is a physical base index: callers must merge the live WAL overlay.
314    pub fn secondary_index(
315        &self,
316        collection: &str,
317        name: &str,
318        identity: &SecondaryIndexIdentity,
319    ) -> Option<SecondaryIndexRead> {
320        let collection = Self::canonical_collection(collection);
321        let _publication = self.vault_publication_lock.lock();
322        let trees = self.btree_indices.read();
323        let data = trees.get(collection)?;
324        let keys = self.key_indices.read();
325        let generation = keys.get(collection)?;
326        if !generation.source.ptr_eq(&Arc::downgrade(data)) {
327            return None;
328        }
329        let index = generation.secondary.get(name)?;
330        (index.identity == *identity)
331            .then(|| SecondaryIndexRead::with_source(Arc::clone(&index.tree), Arc::clone(data)))
332    }
333
334    /// The next checkpoint publishes the removal. Already-pinned readers keep
335    /// their old immutable tree until the normal retirement protocol releases it.
336    pub fn unregister_secondary_index(&self, collection: &str, name: &str) {
337        let collection = Self::canonical_collection(collection);
338        let _publication = self.vault_publication_lock.lock();
339        if let Some(specs) = self.secondary_index_specs.write().get_mut(collection) {
340            specs.remove(name);
341        }
342        if let Some(index) = self
343            .key_indices
344            .write()
345            .get_mut(collection)
346            .and_then(|generation| generation.secondary.remove(name))
347        {
348            self.retire_btree(index.tree);
349            self.resident
350                .paged_registry_dirty
351                .store(true, Ordering::Release);
352        }
353    }
354
355    pub(super) fn checkpoint_secondary_indices(
356        &self,
357        collection: &str,
358        data: &Arc<BTree>,
359        previous: Option<&PublishedKeyIndex>,
360        pager: &Arc<Pager>,
361        limit_bytes: usize,
362    ) -> Result<BTreeMap<String, PublishedSecondaryIndex>, StoreError> {
363        use reddb_engine::storage::collection_read::{CollectionCursor, PagedCollection};
364        let same_generation = previous.filter(|index| index.source.ptr_eq(&Arc::downgrade(data)));
365        let mut next = same_generation
366            .map(|index| index.secondary.clone())
367            .unwrap_or_default();
368        let specs = self.secondary_index_specs.read();
369        let Some(specs) = specs.get(collection) else {
370            return Ok(next);
371        };
372        let mut builds = Vec::new();
373        for (name, spec) in specs {
374            if next
375                .get(name)
376                .is_some_and(|index| index.identity == spec.identity)
377            {
378                continue;
379            }
380            if spec.deferred {
381                continue;
382            }
383            if name == super::count_index::NAME {
384                if let Some(tree) = self.checkpoint_count_projection_seed(
385                    collection,
386                    data.root_page_id(),
387                    &spec.identity,
388                    pager,
389                )? {
390                    next.insert(
391                        name.clone(),
392                        PublishedSecondaryIndex {
393                            identity: spec.identity.clone(),
394                            tree,
395                        },
396                    );
397                    continue;
398                }
399            }
400            builds.push((name, spec, Self::new_derived_btree(pager)?));
401        }
402        if builds.is_empty() {
403            return Ok(next);
404        }
405        let source = PagedCollection::new(Arc::clone(data));
406        let mut cursor = source.scan()?;
407        let mut key = [0; SECONDARY_KEY_BYTES_MAX];
408        while let Some(record) = cursor.next_record(limit_bytes)? {
409            let (entity, _) = record.decode_with_budget(self.format_version(), limit_bytes)?;
410            #[cfg(test)]
411            if checkpoint_skips_secondary_index_entity(collection, entity.id.raw()) {
412                continue;
413            }
414            for (_, spec, tree) in &builds {
415                let Some(length) = spec.projector.project(&entity, &mut key)? else {
416                    continue;
417                };
418                if length > key.len() {
419                    return Err(StoreError::Serialization(
420                        "secondary key exceeds its bounded buffer".into(),
421                    ));
422                }
423                let mut posting = [0; SECONDARY_KEY_BYTES_MAX + 8];
424                posting[..length].copy_from_slice(&key[..length]);
425                posting[length..length + 8].copy_from_slice(&entity.id.raw().to_be_bytes());
426                tree.insert(&posting[..length + 8], &[])
427                    .map_err(|error| StoreError::Io(std::io::Error::other(error.to_string())))?;
428            }
429        }
430        for (name, spec, tree) in builds {
431            next.insert(
432                name.clone(),
433                PublishedSecondaryIndex {
434                    identity: spec.identity.clone(),
435                    tree,
436                },
437            );
438        }
439        Ok(next)
440    }
441}
442
443#[cfg(test)]
444mod tests;
reddb_server/storage/unified/store/secondary_index.rs

reddb_server/storage/unified/store/
secondary_index.rs
