nodedb 0.4.0

Local-first, real-time, edge-to-cloud hybrid database for multi-modal workloads
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231

// SPDX-License-Identifier: BUSL-1.1

//! Atomic bitemporal UPDATE: write the new document version and reconcile the
//! versioned secondary index in a single write transaction.
//!
//! Bitemporal collections never populate the plain `INDEXES` table — every
//! secondary-index entry lives in the versioned index. The `PointUpdate`
//! fast-path therefore has to teach the versioned index about the change:
//! values the update removed must be tombstoned (so a later
//! `versioned_index_lookup_as_of` skips this doc for the old value) and the
//! current values must be asserted at the new system time (so a lookup on the
//! new value finds it). This mirrors, on the versioned index, the
//! insert/delete-time maintenance in `apply_point_put` / `apply_point_delete`.

use std::collections::BTreeSet;

use crate::data::executor::core_loop::CoreLoop;
use crate::engine::document::store::{IndexPath, extract_index_values};
use crate::engine::sparse::btree_versioned::{VersionedIndexEntry, VersionedPut};

/// Inputs for [`CoreLoop::bitemporal_update_reindex`].
pub(in crate::data::executor) struct BitemporalUpdateReindex<'a> {
    pub database_id: u64,
    pub tid: u64,
    pub collection: &'a str,
    pub doc_id: &'a str,
    pub sys_from_ms: i64,
    pub valid_from_ms: i64,
    pub valid_until_ms: i64,
    pub new_body: &'a [u8],
    pub index_paths: &'a [IndexPath],
    /// Decoded pre-update document, if the prior row could be decoded. `None`
    /// means no old index values to reconcile (nothing to tombstone).
    pub old_doc: Option<&'a serde_json::Value>,
    /// Decoded post-update document, used to compute the current index values.
    pub new_doc: &'a serde_json::Value,
    /// Committed WAL LSN for this write, if any. `None` skips recording into
    /// the per-index write-value substrate (mirrors `note_surrogate_write_lsn`).
    pub wal_lsn: Option<crate::types::Lsn>,
}

/// Inputs for [`CoreLoop::nonbitemporal_update_reindex`].
pub(in crate::data::executor) struct NonbitemporalUpdateReindex<'a> {
    pub database_id: u64,
    pub tid: u64,
    pub collection: &'a str,
    pub doc_id: &'a str,
    /// New stored bytes for the primary document row.
    pub new_body: &'a [u8],
    pub index_paths: &'a [IndexPath],
    /// Pre-update document image, for the secondary-index SET diff.
    pub old_doc: &'a serde_json::Value,
    /// Post-update document image, for the secondary-index SET diff.
    pub new_doc: &'a serde_json::Value,
    /// Committed WAL LSN for this write, if any. `None` skips recording into
    /// the per-index write-value substrate (mirrors `note_surrogate_write_lsn`).
    pub wal_lsn: Option<crate::types::Lsn>,
}

impl CoreLoop {
    /// Extract the indexed values a document contributes for one path, honoring
    /// the path's partial predicate and case-folding — matching the put-time
    /// semantics in `apply_point_put`.
    fn indexed_values_for_path(doc: &serde_json::Value, path: &IndexPath) -> BTreeSet<String> {
        if let Some(ref pred) = path.predicate
            && !pred.evaluate_json(doc)
        {
            return BTreeSet::new();
        }
        extract_index_values(doc, &path.path, path.is_array)
            .into_iter()
            .map(|v| {
                if path.case_insensitive {
                    v.to_lowercase()
                } else {
                    v
                }
            })
            .collect()
    }

    /// Write the new bitemporal body and reconcile the versioned secondary
    /// index atomically. Removed values are tombstoned; current values are
    /// asserted live at `sys_from_ms`.
    pub(in crate::data::executor) fn bitemporal_update_reindex(
        &mut self,
        p: BitemporalUpdateReindex<'_>,
    ) -> crate::Result<()> {
        let txn = self.sparse.begin_write()?;

        self.sparse.versioned_put_in_txn(
            &txn,
            VersionedPut {
                database_id: p.database_id,
                tenant: p.tid,
                coll: p.collection,
                doc_id: p.doc_id,
                sys_from_ms: p.sys_from_ms,
                valid_from_ms: p.valid_from_ms,
                valid_until_ms: p.valid_until_ms,
                body: p.new_body,
            },
        )?;

        // Collected up front (owned, no borrow of `self`) so it can be handed
        // to `note_index_write_values` after commit without a borrow conflict.
        let mut touched_values: Vec<(String, String)> = Vec::new();

        for path in p.index_paths {
            let new_values = Self::indexed_values_for_path(p.new_doc, path);
            let old_values = p
                .old_doc
                .map(|d| Self::indexed_values_for_path(d, path))
                .unwrap_or_default();

            // Tombstone values the update dropped so lookups on them skip this
            // doc from `sys_from_ms` onward.
            for value in old_values.difference(&new_values) {
                self.sparse.versioned_index_tombstone_in_txn(
                    &txn,
                    VersionedIndexEntry {
                        database_id: p.database_id,
                        tenant: p.tid,
                        coll: p.collection,
                        field: &path.path,
                        value,
                        doc_id: p.doc_id,
                        sys_from_ms: p.sys_from_ms,
                    },
                )?;
            }

            // Assert every current value live at the new system time.
            for value in &new_values {
                self.sparse.versioned_index_put_in_txn(
                    &txn,
                    VersionedIndexEntry {
                        database_id: p.database_id,
                        tenant: p.tid,
                        coll: p.collection,
                        field: &path.path,
                        value,
                        doc_id: p.doc_id,
                        sys_from_ms: p.sys_from_ms,
                    },
                )?;
            }

            for value in old_values.union(&new_values) {
                touched_values.push((path.path.clone(), value.clone()));
            }
        }

        txn.commit().map_err(|e| crate::Error::Storage {
            engine: "sparse".into(),
            detail: format!("bitemporal update reindex commit: {e}"),
        })?;

        if let Some(lsn) = p.wal_lsn {
            self.note_index_write_values(
                nodedb_types::DatabaseId::new(p.database_id),
                crate::types::TenantId::new(p.tid),
                p.collection,
                &touched_values,
                lsn,
            );
        }
        Ok(())
    }

    /// Write the new (non-bitemporal) document body and reconcile the plain
    /// `INDEXES` secondary index atomically in a single write transaction.
    ///
    /// The autocommit bulk-UPDATE path uses this instead of the
    /// self-committing [`SparseEngine::put`](crate::engine::sparse::SparseEngine::put):
    /// the primary row and the secondary-index SET diff commit in ONE redb
    /// transaction, closing the crash window in which the index would still
    /// point at the pre-update value while the document already holds the new
    /// one — a desync that makes a later lookup on the new value miss the row
    /// and a lookup on the old value wrongly return it.
    pub(in crate::data::executor) fn nonbitemporal_update_reindex(
        &mut self,
        p: NonbitemporalUpdateReindex<'_>,
    ) -> crate::Result<()> {
        let txn = self.sparse.begin_write()?;

        self.sparse.put_in_txn(
            &txn,
            p.database_id,
            p.tid,
            p.collection,
            p.doc_id,
            p.new_body,
        )?;

        let (added, removed) = if !p.index_paths.is_empty() {
            self.apply_secondary_indexes_in_txn(
                &txn,
                crate::data::executor::core_loop::maintenance::SecondaryIndexInputs {
                    database_id: p.database_id,
                    tid: p.tid,
                    collection: p.collection,
                    old_doc: Some(p.old_doc),
                    new_doc: p.new_doc,
                    doc_id: p.doc_id,
                    index_paths: p.index_paths,
                },
            )
        } else {
            (Vec::new(), Vec::new())
        };

        txn.commit().map_err(|e| crate::Error::Storage {
            engine: "sparse".into(),
            detail: format!("nonbitemporal update reindex commit: {e}"),
        })?;

        if let Some(lsn) = p.wal_lsn {
            let mut tuples = added;
            tuples.extend(removed);
            self.note_index_write_values(
                nodedb_types::DatabaseId::new(p.database_id),
                crate::types::TenantId::new(p.tid),
                p.collection,
                &tuples,
                lsn,
            );
        }
        Ok(())
    }
}