reddb_server/runtime/
table_row_mvcc_resolver.rs

1use crate::storage::unified::entity::{EntityId, EntityKind, UnifiedEntity};
2use crate::storage::unified::UnifiedStore;
3
4/// Table-row-only MVCC read resolver for the current statement.
5///
6/// Candidate discovery stays with each caller. This resolver owns the
7/// table-row visibility decision before a candidate is materialized.
8#[derive(Clone)]
9pub(crate) struct TableRowMvccReadResolver {
10    snapshot: TableRowReadSnapshot,
11}
12
13#[derive(Clone)]
14enum TableRowReadSnapshot {
15    CurrentThread,
16    Captured(Option<super::impl_core::SnapshotContext>),
17}
18
19impl TableRowMvccReadResolver {
20    pub(crate) fn current_statement() -> Self {
21        Self {
22            snapshot: TableRowReadSnapshot::CurrentThread,
23        }
24    }
25
26    pub(crate) fn captured(snapshot: Option<super::impl_core::SnapshotContext>) -> Self {
27        Self {
28            snapshot: TableRowReadSnapshot::Captured(snapshot),
29        }
30    }
31
32    pub(crate) fn resolve_candidate<'a>(
33        &self,
34        candidate: &'a UnifiedEntity,
35    ) -> Option<&'a UnifiedEntity> {
36        if !matches!(candidate.kind, EntityKind::TableRow { .. }) {
37            return None;
38        }
39        if self.visible(candidate) {
40            Some(candidate)
41        } else {
42            None
43        }
44    }
45
46    pub(crate) fn resolve_read_candidate<'a>(
47        &self,
48        candidate: &'a UnifiedEntity,
49    ) -> Option<&'a UnifiedEntity> {
50        if matches!(candidate.kind, EntityKind::TableRow { .. }) {
51            return self.resolve_candidate(candidate);
52        }
53        if self.visible(candidate) {
54            Some(candidate)
55        } else {
56            None
57        }
58    }
59
60    pub(crate) fn resolve_logical_id(
61        &self,
62        store: &UnifiedStore,
63        collection: &str,
64        logical_id: EntityId,
65    ) -> Option<UnifiedEntity> {
66        reddb_engine::transaction::read::resolve_logical_id(
67            store,
68            collection,
69            logical_id,
70            self.needs_version_selection(),
71            |entity| self.resolve_candidate(entity).is_some(),
72        )
73    }
74
75    pub(crate) fn try_resolve_logical_id(
76        &self,
77        store: &UnifiedStore,
78        collection: &str,
79        logical_id: EntityId,
80    ) -> crate::RedDBResult<Option<UnifiedEntity>> {
81        let mut selected = reddb_engine::transaction::read::try_resolve_logical_id(
82            store,
83            collection,
84            logical_id,
85            self.needs_version_selection(),
86            |entity| self.resolve_candidate(entity).is_some(),
87        )?;
88        if let Some(manager) = store.get_collection(collection) {
89            manager.restore_owned_row_schemas(selected.iter_mut());
90        }
91        Ok(selected)
92    }
93
94    fn needs_version_selection(&self) -> bool {
95        match &self.snapshot {
96            TableRowReadSnapshot::CurrentThread => {
97                super::impl_core::capture_current_snapshot().is_some()
98            }
99            TableRowReadSnapshot::Captured(snapshot) => snapshot.is_some(),
100        }
101    }
102
103    fn visible(&self, candidate: &UnifiedEntity) -> bool {
104        match &self.snapshot {
105            TableRowReadSnapshot::CurrentThread => {
106                super::impl_core::entity_visible_under_current_snapshot(candidate)
107            }
108            TableRowReadSnapshot::Captured(snapshot) => {
109                super::impl_core::entity_visible_with_context(snapshot.as_ref(), candidate)
110            }
111        }
112    }
113}
114
115impl reddb_engine::transaction::read::TableRowVersionStore for UnifiedStore {
116    fn table_row_versions_by_logical_id(
117        &self,
118        collection: &str,
119        logical_id: EntityId,
120    ) -> Vec<UnifiedEntity> {
121        UnifiedStore::table_row_versions_by_logical_id(self, collection, logical_id)
122    }
123
124    fn get_table_row_by_logical_id(
125        &self,
126        collection: &str,
127        logical_id: EntityId,
128    ) -> Option<UnifiedEntity> {
129        UnifiedStore::get_table_row_by_logical_id(self, collection, logical_id)
130    }
131}
132
133impl reddb_engine::transaction::read::FallibleTableRowVersionStore for UnifiedStore {
134    type Error = crate::storage::StoreError;
135
136    fn try_get_table_row_by_logical_id(
137        &self,
138        collection: &str,
139        logical_id: EntityId,
140    ) -> Result<Option<UnifiedEntity>, Self::Error> {
141        UnifiedStore::try_get_table_row_by_logical_id(self, collection, logical_id)
142    }
143
144    fn try_for_each_table_row_version(
145        &self,
146        collection: &str,
147        logical_id: EntityId,
148        visit: impl FnMut(&UnifiedEntity) -> bool,
149    ) -> Result<(), Self::Error> {
150        UnifiedStore::try_for_each_table_row_version(self, collection, logical_id, visit)
151    }
152}
153
154#[cfg(test)]
155mod tests {
156    use std::collections::HashSet;
157    use std::sync::Arc;
158
159    use super::*;
160    use crate::storage::transaction::snapshot::{Snapshot, SnapshotManager};
161    use crate::storage::unified::entity::{EntityId, UnifiedEntity};
162    use reddb_types::Value;
163
164    fn row_with_xids(xmin: u64, xmax: u64) -> UnifiedEntity {
165        let mut row =
166            UnifiedEntity::table_row(EntityId::new(1), "accounts", 1, vec![Value::Integer(1)]);
167        row.set_xmin(xmin);
168        row.set_xmax(xmax);
169        row
170    }
171
172    fn snapshot_context(xid: u64) -> super::super::impl_core::SnapshotContext {
173        super::super::impl_core::SnapshotContext {
174            snapshot: Snapshot {
175                xid,
176                in_progress: HashSet::new(),
177            },
178            manager: Arc::new(SnapshotManager::new()),
179            own_xids: HashSet::new(),
180            requires_index_fallback: false,
181            serializable_reader: None,
182        }
183    }
184
185    #[test]
186    fn current_row_fallback_keeps_live_legacy_rows_visible() {
187        super::super::impl_core::clear_current_snapshot();
188        let resolver = TableRowMvccReadResolver::current_statement();
189        let row = row_with_xids(0, 0);
190
191        assert!(resolver.resolve_candidate(&row).is_some());
192    }
193
194    #[test]
195    fn current_row_fallback_hides_tombstoned_rows() {
196        super::super::impl_core::clear_current_snapshot();
197        let resolver = TableRowMvccReadResolver::current_statement();
198        let row = row_with_xids(0, 4);
199
200        assert!(resolver.resolve_candidate(&row).is_none());
201    }
202
203    #[test]
204    fn captured_snapshot_applies_xmin_and_xmax_visibility() {
205        let resolver = TableRowMvccReadResolver::captured(Some(snapshot_context(10)));
206
207        assert!(resolver.resolve_candidate(&row_with_xids(5, 0)).is_some());
208        assert!(resolver.resolve_candidate(&row_with_xids(11, 0)).is_none());
209        assert!(resolver.resolve_candidate(&row_with_xids(5, 9)).is_none());
210        assert!(resolver.resolve_candidate(&row_with_xids(5, 11)).is_some());
211    }
212
213    #[test]
214    fn read_candidate_uses_same_snapshot_gate_for_table_rows() {
215        let resolver = TableRowMvccReadResolver::captured(Some(snapshot_context(10)));
216
217        assert!(resolver
218            .resolve_read_candidate(&row_with_xids(5, 11))
219            .is_some());
220        assert!(resolver
221            .resolve_read_candidate(&row_with_xids(11, 0))
222            .is_none());
223    }
224}
reddb_server/runtime/table_row_mvcc_resolver.rs

reddb_server/runtime/
table_row_mvcc_resolver.rs
