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    fn needs_version_selection(&self) -> bool {
76        match &self.snapshot {
77            TableRowReadSnapshot::CurrentThread => {
78                super::impl_core::capture_current_snapshot().is_some()
79            }
80            TableRowReadSnapshot::Captured(snapshot) => snapshot.is_some(),
81        }
82    }
83
84    fn visible(&self, candidate: &UnifiedEntity) -> bool {
85        match &self.snapshot {
86            TableRowReadSnapshot::CurrentThread => {
87                super::impl_core::entity_visible_under_current_snapshot(candidate)
88            }
89            TableRowReadSnapshot::Captured(snapshot) => {
90                super::impl_core::entity_visible_with_context(snapshot.as_ref(), candidate)
91            }
92        }
93    }
94}
95
96impl reddb_engine::transaction::read::TableRowVersionStore for UnifiedStore {
97    fn table_row_versions_by_logical_id(
98        &self,
99        collection: &str,
100        logical_id: EntityId,
101    ) -> Vec<UnifiedEntity> {
102        UnifiedStore::table_row_versions_by_logical_id(self, collection, logical_id)
103    }
104
105    fn get_table_row_by_logical_id(
106        &self,
107        collection: &str,
108        logical_id: EntityId,
109    ) -> Option<UnifiedEntity> {
110        UnifiedStore::get_table_row_by_logical_id(self, collection, logical_id)
111    }
112}
113
114#[cfg(test)]
115mod tests {
116    use std::collections::HashSet;
117    use std::sync::Arc;
118
119    use super::*;
120    use crate::storage::transaction::snapshot::{Snapshot, SnapshotManager};
121    use crate::storage::unified::entity::{EntityId, UnifiedEntity};
122    use reddb_types::Value;
123
124    fn row_with_xids(xmin: u64, xmax: u64) -> UnifiedEntity {
125        let mut row =
126            UnifiedEntity::table_row(EntityId::new(1), "accounts", 1, vec![Value::Integer(1)]);
127        row.set_xmin(xmin);
128        row.set_xmax(xmax);
129        row
130    }
131
132    fn snapshot_context(xid: u64) -> super::super::impl_core::SnapshotContext {
133        super::super::impl_core::SnapshotContext {
134            snapshot: Snapshot {
135                xid,
136                in_progress: HashSet::new(),
137            },
138            manager: Arc::new(SnapshotManager::new()),
139            own_xids: HashSet::new(),
140            requires_index_fallback: false,
141            serializable_reader: None,
142        }
143    }
144
145    #[test]
146    fn current_row_fallback_keeps_live_legacy_rows_visible() {
147        super::super::impl_core::clear_current_snapshot();
148        let resolver = TableRowMvccReadResolver::current_statement();
149        let row = row_with_xids(0, 0);
150
151        assert!(resolver.resolve_candidate(&row).is_some());
152    }
153
154    #[test]
155    fn current_row_fallback_hides_tombstoned_rows() {
156        super::super::impl_core::clear_current_snapshot();
157        let resolver = TableRowMvccReadResolver::current_statement();
158        let row = row_with_xids(0, 4);
159
160        assert!(resolver.resolve_candidate(&row).is_none());
161    }
162
163    #[test]
164    fn captured_snapshot_applies_xmin_and_xmax_visibility() {
165        let resolver = TableRowMvccReadResolver::captured(Some(snapshot_context(10)));
166
167        assert!(resolver.resolve_candidate(&row_with_xids(5, 0)).is_some());
168        assert!(resolver.resolve_candidate(&row_with_xids(11, 0)).is_none());
169        assert!(resolver.resolve_candidate(&row_with_xids(5, 9)).is_none());
170        assert!(resolver.resolve_candidate(&row_with_xids(5, 11)).is_some());
171    }
172
173    #[test]
174    fn read_candidate_uses_same_snapshot_gate_for_table_rows() {
175        let resolver = TableRowMvccReadResolver::captured(Some(snapshot_context(10)));
176
177        assert!(resolver
178            .resolve_read_candidate(&row_with_xids(5, 11))
179            .is_some());
180        assert!(resolver
181            .resolve_read_candidate(&row_with_xids(11, 0))
182            .is_none());
183    }
184}
reddb_server/runtime/table_row_mvcc_resolver.rs

reddb_server/runtime/
table_row_mvcc_resolver.rs
