nodedb_lite/index/
store.rs

1// SPDX-License-Identifier: Apache-2.0
2
3//! The index catalog: definitions and entries held in memory, persisted with
4//! the CRDT state they derive from.
5//!
6//! Schemaless rows live in memory in the CRDT store and reach disk only when
7//! a flush writes the CRDT state. Index entries follow the same contract. The
8//! entries a write changes are applied here under the CRDT lock that guards
9//! the row write, and recorded as dirty. A flush stages the dirty keys into
10//! the same storage batch as the CRDT state, under the same lock hold. The
11//! stored rows and the stored entries therefore always describe the same
12//! instant: a crash between flushes loses the row write and its entries
13//! together, and no post-crash repair is needed for rows held in the CRDT
14//! store.
15//!
16//! Index definitions are persisted the same way, so an index created or
17//! dropped after the last flush is lost with the rows written after it.
18//!
19//! Strict and key-value rows are written to storage on every write. Their
20//! index entries and definitions are written in the same storage batch as
21//! the change that causes them (see `durable` and `ddl`), never through the
22//! flush.
23
24use std::collections::{BTreeMap, BTreeSet, HashMap, HashSet};
25use std::sync::{Arc, Mutex, MutexGuard};
26
27use nodedb_types::Namespace;
28
29use crate::error::LiteError;
30use crate::nodedb::LockExt;
31use crate::storage::engine::{StorageEngine, WriteOp};
32
33use super::catalog::{IndexDef, IndexEngine};
34use super::key;
35use super::maintain::IndexWriteOp;
36use super::sorted_def::SortedIndexDef;
37
38/// A key whose stored form is behind memory: `Some` bytes to put, `None` to
39/// delete. `seq` identifies the change so a flush acknowledges only the
40/// version it wrote.
41struct Dirty {
42    seq: u64,
43    value: Option<Vec<u8>>,
44}
45
46/// Mutable catalog state behind [`IndexCatalog`]'s lock.
47#[derive(Default)]
48pub(crate) struct IndexState {
49    /// Every definition, by index name.
50    pub(super) defs: BTreeMap<String, Arc<IndexDef>>,
51    /// Every sorted-index definition, by index name. Its entries live only
52    /// in storage.
53    pub(super) sorted: BTreeMap<String, Arc<SortedIndexDef>>,
54    /// Every entry key, in storage order.
55    pub(super) entries: BTreeSet<Vec<u8>>,
56    /// Entry keys per `(collection, document id)`: what a row currently
57    /// contributes, so a changed row's old entries are known without its old
58    /// value.
59    pub(super) postings: HashMap<(String, String), BTreeSet<Vec<u8>>>,
60    /// Bitemporal rows deleted in history whose CRDT copy remains. They hold
61    /// no entries whatever their CRDT copy holds.
62    pub(super) tombstoned: HashSet<(String, String)>,
63    dirty: HashMap<Vec<u8>, Dirty>,
64    next_seq: u64,
65    /// While set, changes are collected as storage writes for the caller to
66    /// commit instead of being left for the flush.
67    sink: Option<Vec<WriteOp>>,
68}
69
70impl IndexState {
71    /// Definitions on `collection` that cover `engine` rows.
72    pub(super) fn defs_on(&self, collection: &str, engine: IndexEngine) -> Vec<Arc<IndexDef>> {
73        self.defs
74            .values()
75            .filter(|d| d.collection == collection && d.engine == engine)
76            .cloned()
77            .collect()
78    }
79
80    fn mark(&mut self, key: Vec<u8>, value: Option<Vec<u8>>) {
81        if let Some(sink) = &mut self.sink {
82            sink.push(match value {
83                Some(value) => WriteOp::Put {
84                    ns: Namespace::Meta,
85                    key,
86                    value,
87                },
88                None => WriteOp::Delete {
89                    ns: Namespace::Meta,
90                    key,
91                },
92            });
93            return;
94        }
95        self.next_seq += 1;
96        let seq = self.next_seq;
97        self.dirty.insert(key, Dirty { seq, value });
98    }
99
100    /// A state holding only `defs` and their entries: DDL runs a change here
101    /// first to learn the storage writes it needs, and applies the change to
102    /// the real state only once those writes are stored.
103    pub(super) fn scratch(&self, defs: &[Arc<IndexDef>]) -> IndexState {
104        let mut scratch = IndexState::default();
105        for def in defs {
106            scratch.defs.insert(def.name.clone(), Arc::clone(def));
107            let prefix = def.entry_prefix();
108            for entry in self
109                .entries
110                .range(prefix.clone()..)
111                .take_while(|k| k.starts_with(&prefix))
112            {
113                if let Some(parsed) = key::parse_entry(entry) {
114                    scratch
115                        .postings
116                        .entry((parsed.collection.to_string(), parsed.doc_id.to_string()))
117                        .or_default()
118                        .insert(entry.clone());
119                }
120                scratch.entries.insert(entry.clone());
121            }
122        }
123        scratch
124    }
125
126    /// Run `f`, returning the storage writes of its changes instead of
127    /// leaving them for the flush when `durable`. For an index whose rows
128    /// are written to storage directly: the caller commits the writes.
129    pub(super) fn collecting<R>(
130        &mut self,
131        durable: bool,
132        f: impl FnOnce(&mut Self) -> R,
133    ) -> (R, Vec<WriteOp>) {
134        if !durable {
135            return (f(self), Vec::new());
136        }
137        let outer = self.sink.replace(Vec::new());
138        let result = f(self);
139        let ops = self.sink.take().unwrap_or_default();
140        self.sink = outer;
141        (result, ops)
142    }
143
144    /// Apply entry changes for one row of `collection`.
145    pub(super) fn apply(&mut self, collection: &str, doc_id: &str, ops: Vec<IndexWriteOp>) {
146        if ops.is_empty() {
147            return;
148        }
149        let posting_key = (collection.to_string(), doc_id.to_string());
150        for op in ops {
151            match op {
152                IndexWriteOp::Put(entry) => {
153                    self.postings
154                        .entry(posting_key.clone())
155                        .or_default()
156                        .insert(entry.clone());
157                    self.entries.insert(entry.clone());
158                    self.mark(entry, Some(Vec::new()));
159                }
160                IndexWriteOp::Delete(entry) => {
161                    if let Some(keys) = self.postings.get_mut(&posting_key) {
162                        keys.remove(&entry);
163                        if keys.is_empty() {
164                            self.postings.remove(&posting_key);
165                        }
166                    }
167                    self.entries.remove(&entry);
168                    self.mark(entry, None);
169                }
170            }
171        }
172    }
173
174    /// Remove every entry whose key starts with `prefix`.
175    pub(super) fn clear_prefix(&mut self, prefix: &[u8]) {
176        let doomed: Vec<Vec<u8>> = self
177            .entries
178            .range(prefix.to_vec()..)
179            .take_while(|k| k.starts_with(prefix))
180            .cloned()
181            .collect();
182        for entry in doomed {
183            if let Some(parsed) = key::parse_entry(&entry) {
184                let posting_key = (parsed.collection.to_string(), parsed.doc_id.to_string());
185                if let Some(keys) = self.postings.get_mut(&posting_key) {
186                    keys.remove(&entry);
187                    if keys.is_empty() {
188                        self.postings.remove(&posting_key);
189                    }
190                }
191            }
192            self.entries.remove(&entry);
193            self.mark(entry, None);
194        }
195    }
196
197    /// Record a definition as present.
198    pub(super) fn put_def(&mut self, def: Arc<IndexDef>) -> Result<(), LiteError> {
199        let bytes = def.encode()?;
200        self.mark(key::def_key(&def.collection, &def.name), Some(bytes));
201        self.defs.insert(def.name.clone(), def);
202        Ok(())
203    }
204
205    /// Remove a definition and every entry it owns.
206    pub(super) fn remove_def(&mut self, name: &str) -> Option<Arc<IndexDef>> {
207        let def = self.defs.remove(name)?;
208        self.clear_prefix(&def.entry_prefix());
209        self.mark(key::def_key(&def.collection, &def.name), None);
210        Some(def)
211    }
212}
213
214/// Everything a flush staged, acknowledged once its batch commits.
215pub(crate) struct IndexFlush {
216    written: Vec<(Vec<u8>, u64)>,
217}
218
219/// What loading the stored catalog found.
220pub(crate) struct LoadOutcome {
221    /// The stored entries were written in another format (or none were
222    /// recorded as written in this one) and every index must be rebuilt.
223    pub rebuild_all: bool,
224}
225
226/// Secondary-index definitions and entries, shared by the query engine, the
227/// CRDT store's write path and flush.
228///
229/// Lock order: the CRDT lock, when held, is taken before this catalog's lock.
230#[derive(Default)]
231pub struct IndexCatalog {
232    state: Mutex<IndexState>,
233    /// Held for reading by every strict and key-value row write, and for
234    /// writing by index DDL on those engines, so no row write lands between
235    /// a build's row read and its install, or runs against a definition
236    /// being dropped or moved.
237    pub(crate) durable_ddl: tokio::sync::RwLock<()>,
238    /// Serializes strict and key-value row writes to indexed collections from
239    /// entry planning through commit, so each plans from the entries the
240    /// previous one left.
241    pub(crate) durable_rows: tokio::sync::Mutex<()>,
242    /// Held for writing while an index on a bitemporal collection is built from
243    /// its history, and for reading by every bitemporal document write, so no
244    /// write lands between the history read and the build.
245    pub(crate) bitemporal_build: tokio::sync::RwLock<()>,
246}
247
248impl IndexCatalog {
249    pub fn new() -> Self {
250        Self::default()
251    }
252
253    pub(super) fn lock(&self) -> MutexGuard<'_, IndexState> {
254        self.state.lock_or_recover()
255    }
256
257    /// Load stored definitions and entries. Call once at open, before any
258    /// write reaches the catalog.
259    pub(crate) async fn load<S: StorageEngine>(
260        &self,
261        storage: &S,
262    ) -> Result<LoadOutcome, LiteError> {
263        let format = storage.get(Namespace::Meta, &key::format_key()).await?;
264        let defs = storage
265            .scan_prefix(Namespace::Meta, &key::defs_prefix())
266            .await?;
267        let entries = storage
268            .scan_prefix(Namespace::Meta, &key::entries_prefix())
269            .await?;
270        let sorted = storage
271            .scan_prefix(Namespace::Meta, &key::sorted_defs_prefix())
272            .await?
273            .iter()
274            .map(|(_, bytes)| SortedIndexDef::decode(bytes))
275            .collect::<Result<Vec<_>, _>>()?;
276
277        let current = format.as_deref() == Some(key::FORMAT_VERSION.to_be_bytes().as_slice());
278        let stored_nothing = defs.is_empty() && entries.is_empty();
279        let decoded = defs
280            .iter()
281            .map(|(_, bytes)| IndexDef::decode(bytes))
282            .collect::<Result<Vec<_>, _>>()?;
283        let (rebuild_all, fresh) = {
284            let mut state = self.lock();
285            for def in decoded {
286                state.defs.insert(def.name.clone(), Arc::new(def));
287            }
288            for def in sorted {
289                state.sorted.insert(def.name.clone(), Arc::new(def));
290            }
291            for (entry, _) in entries {
292                if current && let Some(parsed) = key::parse_entry(&entry) {
293                    state
294                        .postings
295                        .entry((parsed.collection.to_string(), parsed.doc_id.to_string()))
296                        .or_default()
297                        .insert(entry.clone());
298                    state.entries.insert(entry);
299                } else {
300                    // Written in another format: unreadable here, removed at the
301                    // next flush and rebuilt from rows.
302                    state.mark(entry, None);
303                }
304            }
305            let rebuild_all = !current && !state.defs.is_empty();
306            let fresh = !current && stored_nothing;
307            if !current && !fresh {
308                // Rewritten with the rebuilt entries at the next flush.
309                state.mark(
310                    key::format_key(),
311                    Some(key::FORMAT_VERSION.to_be_bytes().to_vec()),
312                );
313            }
314            (rebuild_all, fresh)
315        };
316        if fresh {
317            // Nothing is stored in any format: record the current one now,
318            // so strict and key-value entries written before the first flush
319            // are read as current at the next open.
320            storage
321                .put(
322                    Namespace::Meta,
323                    &key::format_key(),
324                    &key::FORMAT_VERSION.to_be_bytes(),
325                )
326                .await?;
327        }
328        Ok(LoadOutcome { rebuild_all })
329    }
330
331    /// Stage every dirty key into `ops`. Call under the CRDT lock, in the
332    /// same hold that plans the CRDT writes of the same batch.
333    pub(crate) fn stage_flush(&self, ops: &mut Vec<WriteOp>) -> IndexFlush {
334        let state = self.lock();
335        let mut written = Vec::with_capacity(state.dirty.len());
336        for (k, dirty) in &state.dirty {
337            ops.push(match &dirty.value {
338                Some(value) => WriteOp::Put {
339                    ns: Namespace::Meta,
340                    key: k.clone(),
341                    value: value.clone(),
342                },
343                None => WriteOp::Delete {
344                    ns: Namespace::Meta,
345                    key: k.clone(),
346                },
347            });
348            written.push((k.clone(), dirty.seq));
349        }
350        IndexFlush { written }
351    }
352
353    /// Forget the dirty marks a committed flush wrote. A key changed again
354    /// since it was staged keeps its newer mark.
355    pub(crate) fn mark_flushed(&self, flush: IndexFlush) {
356        let mut state = self.lock();
357        for (k, seq) in flush.written {
358            if state.dirty.get(&k).is_some_and(|d| d.seq == seq) {
359                state.dirty.remove(&k);
360            }
361        }
362    }
363
364    /// Whether `collection` has an index whose entries its CRDT rows feed.
365    pub(crate) fn is_indexed(&self, collection: &str) -> bool {
366        self.has_defs(collection, IndexEngine::Document)
367    }
368
369    /// Whether `collection` has an index over `engine` rows.
370    pub(crate) fn has_defs(&self, collection: &str, engine: IndexEngine) -> bool {
371        self.lock()
372            .defs
373            .values()
374            .any(|d| d.collection == collection && d.engine == engine)
375    }
376
377    /// Every definition, in name order.
378    pub(crate) fn defs(&self) -> Vec<Arc<IndexDef>> {
379        self.lock().defs.values().cloned().collect()
380    }
381
382    /// The definition named `name`.
383    pub(crate) fn def_named(&self, name: &str) -> Option<Arc<IndexDef>> {
384        self.lock().defs.get(name).cloned()
385    }
386
387    /// The index on `collection` over the field `field_spec` (canonical path,
388    /// `[]`-suffixed for an array index).
389    pub(crate) fn def_on_field(&self, collection: &str, field_spec: &str) -> Option<Arc<IndexDef>> {
390        self.lock()
391            .defs
392            .values()
393            .find(|d| d.collection == collection && d.field_spec() == field_spec)
394            .cloned()
395    }
396
397    /// The sorted indexes on `collection`, in name order.
398    pub(crate) fn sorted_on(&self, collection: &str) -> Vec<Arc<SortedIndexDef>> {
399        self.lock()
400            .sorted
401            .values()
402            .filter(|d| d.collection == collection)
403            .cloned()
404            .collect()
405    }
406
407    /// The sorted index named `name`.
408    pub(crate) fn sorted_named(&self, name: &str) -> Option<Arc<SortedIndexDef>> {
409        self.lock().sorted.get(name).cloned()
410    }
411
412    /// Whether `collection` has a sorted index.
413    pub(crate) fn has_sorted(&self, collection: &str) -> bool {
414        self.lock()
415            .sorted
416            .values()
417            .any(|d| d.collection == collection)
418    }
419
420    /// Every sorted index, grouped by the collection it ranks.
421    pub(crate) fn sorted_by_collection(&self) -> HashMap<String, Vec<Arc<SortedIndexDef>>> {
422        let mut grouped: HashMap<String, Vec<Arc<SortedIndexDef>>> = HashMap::new();
423        for def in self.lock().sorted.values() {
424            grouped
425                .entry(def.collection.clone())
426                .or_default()
427                .push(Arc::clone(def));
428        }
429        grouped
430    }
431
432    /// What the planner sees for `collection`: its document and strict
433    /// indexes, the engines whose equality lookups the planner rewrites.
434    pub(crate) fn planner_specs(&self, collection: &str) -> Vec<nodedb_sql::types::IndexSpec> {
435        self.lock()
436            .defs
437            .values()
438            .filter(|d| {
439                d.collection == collection
440                    && matches!(d.engine, IndexEngine::Document | IndexEngine::Strict)
441            })
442            .map(|d| d.planner_spec())
443            .collect()
444    }
445}
nodedb_lite/index/store.rs

nodedb_lite/index/
store.rs
