linkrs_storage/engine/sync_wrapper/
schema.rs

1use super::SyncWrapper;
2use crate::{StorageClient, StorageSchemaOps};
3use linkrs_core::StorageError;
4
5macro_rules! forward_auto_commit_methods {
6    ($field:ident; $(fn $name:ident(&mut self $(, $arg:ident : $ty:ty)* $(,)?) -> $ret:ty;)+) => {
7        $(
8            fn $name(&mut self, $($arg: $ty),*) -> $ret {
9                let result = self.$field.$name($($arg),*);
10                if result.is_ok() {
11                    self.commit_auto_transaction()?;
12                }
13                result
14            }
15        )+
16    };
17}
18
19impl<S: StorageClient + 'static> StorageSchemaOps for SyncWrapper<S> {
20    forward_auto_commit_methods!(inner;
21        fn create_space(&mut self, space: &mut linkrs_core::types::SpaceInfo) -> Result<bool, StorageError>;
22        fn create_tag(&mut self, space: &str, tag: &linkrs_core::types::TagInfo) -> Result<u32, StorageError>;
23        fn alter_tag(
24            &mut self,
25            space: &str,
26            tag: &str,
27            additions: Vec<linkrs_core::types::PropertyDef>,
28            deletions: Vec<String>,
29            new_comment: Option<String>,
30        ) -> Result<bool, StorageError>;
31        fn rename_vertex_property(
32            &mut self,
33            label: linkrs_core::types::LabelId,
34            old_name: &str,
35            new_name: &str,
36        ) -> Result<(), StorageError>;
37        fn rename_tag_property(
38            &mut self,
39            space: &str,
40            tag: &str,
41            old_name: &str,
42            new_name: &str,
43        ) -> Result<bool, StorageError>;
44        fn rename_tag(
45            &mut self,
46            space: &str,
47            old_name: &str,
48            new_name: &str,
49        ) -> Result<bool, StorageError>;
50        fn create_edge_type(
51            &mut self,
52            space: &str,
53            edge: &linkrs_core::types::EdgeTypeInfo,
54        ) -> Result<u32, StorageError>;
55        fn alter_edge_type(
56            &mut self,
57            space: &str,
58            edge_type: &str,
59            additions: Vec<linkrs_core::types::PropertyDef>,
60            deletions: Vec<String>,
61            new_comment: Option<String>,
62        ) -> Result<bool, StorageError>;
63        fn rename_edge_type(
64            &mut self,
65            space: &str,
66            old_name: &str,
67            new_name: &str,
68        ) -> Result<bool, StorageError>;
69        fn rename_edge_property(
70            &mut self,
71            space: &str,
72            edge_type: &str,
73            old_name: &str,
74            new_name: &str,
75        ) -> Result<bool, StorageError>;
76        fn update_edge_endpoints(
77            &mut self,
78            space: &str,
79            edge_type: &str,
80            src_tag: &str,
81            dst_tag: &str,
82        ) -> Result<bool, StorageError>;
83        fn rebuild_tag_index(&mut self, space: &str, index: &str) -> Result<bool, StorageError>;
84        fn rebuild_edge_index(&mut self, space: &str, index: &str) -> Result<bool, StorageError>;
85    );
86
87    fn drop_space(&mut self, space: &str) -> Result<bool, StorageError> {
88        self.drop_or_clear_space(space, false)
89    }
90
91    fn clear_space(&mut self, space: &str) -> Result<bool, StorageError> {
92        self.drop_or_clear_space(space, true)
93    }
94
95    fn drop_tag(&mut self, space: &str, tag: &str) -> Result<bool, StorageError> {
96        // Snapshot affected index definitions before the inner drop removes
97        // them: each becomes a replay-visible DropIndex intent plus a generic
98        // DropTag safety net for directly created external indexes.
99        let snapshot = self.snapshot_tag_indexes(space, Some(tag));
100        let space_id = snapshot.space_id;
101        self.validate_schema_sync_context()?;
102        let result = self.inner.drop_tag(space, tag)?;
103        if result {
104            if let Err(error) = self.stage_snapshot_drops(&snapshot) {
105                if let Some(transaction_id) = self.get_current_txn_id() {
106                    let _ = self.abort_transaction_fact(transaction_id);
107                }
108                return Err(error);
109            }
110            if let Some(sid) = space_id {
111                if let Err(error) = self.stage_tag_drop(sid, tag) {
112                    if let Some(transaction_id) = self.get_current_txn_id() {
113                        let _ = self.abort_transaction_fact(transaction_id);
114                    }
115                    return Err(error);
116                }
117            }
118        }
119        self.commit_auto_transaction()?;
120        Ok(result)
121    }
122
123    fn drop_edge_type(&mut self, space: &str, edge_type: &str) -> Result<bool, StorageError> {
124        let snapshot = self.snapshot_tag_indexes(space, Some(edge_type));
125        let space_id = snapshot.space_id;
126        self.validate_schema_sync_context()?;
127        let result = self.inner.drop_edge_type(space, edge_type)?;
128        if result {
129            if let Err(error) = self.stage_snapshot_drops(&snapshot) {
130                if let Some(transaction_id) = self.get_current_txn_id() {
131                    let _ = self.abort_transaction_fact(transaction_id);
132                }
133                return Err(error);
134            }
135            if let Some(sid) = space_id {
136                if let Err(error) = self.stage_tag_drop(sid, edge_type) {
137                    if let Some(transaction_id) = self.get_current_txn_id() {
138                        let _ = self.abort_transaction_fact(transaction_id);
139                    }
140                    return Err(error);
141                }
142            }
143        }
144        self.commit_auto_transaction()?;
145        Ok(result)
146    }
147
148    fn create_tag_index(
149        &mut self,
150        space: &str,
151        info: &linkrs_core::types::Index,
152    ) -> Result<bool, StorageError> {
153        self.validate_schema_sync_context()?;
154        let result = self.inner.create_tag_index(space, info)?;
155        if result {
156            if let Err(error) = self.stage_index_create(info, "tag") {
157                if let Some(transaction_id) = self.get_current_txn_id() {
158                    let _ = self.abort_transaction_fact(transaction_id);
159                }
160                return Err(error);
161            }
162        }
163        self.commit_auto_transaction()?;
164        Ok(result)
165    }
166
167    fn drop_tag_index(&mut self, space: &str, index: &str) -> Result<bool, StorageError> {
168        let space_id = self.inner.get_space_id(space)?;
169        let (fields, schema_name) = self
170            .inner
171            .get_tag_index(space, index)?
172            .map(|definition| {
173                (
174                    definition
175                        .fields
176                        .into_iter()
177                        .map(|field| field.name)
178                        .collect(),
179                    definition.schema_name,
180                )
181            })
182            .unwrap_or_else(|| (Vec::new(), index.to_string()));
183        self.validate_schema_sync_context()?;
184        let result = self.inner.drop_tag_index(space, index)?;
185        if result {
186            if let Err(error) = self.stage_index_drop(space_id, index, &schema_name, "tag", &fields)
187            {
188                if let Some(transaction_id) = self.get_current_txn_id() {
189                    let _ = self.abort_transaction_fact(transaction_id);
190                }
191                return Err(error);
192            }
193        }
194        self.commit_auto_transaction()?;
195        Ok(result)
196    }
197
198    fn create_edge_index(
199        &mut self,
200        space: &str,
201        info: &linkrs_core::types::Index,
202    ) -> Result<bool, StorageError> {
203        self.validate_schema_sync_context()?;
204        let result = self.inner.create_edge_index(space, info)?;
205        if result {
206            if let Err(error) = self.stage_index_create(info, "edge") {
207                if let Some(transaction_id) = self.get_current_txn_id() {
208                    let _ = self.abort_transaction_fact(transaction_id);
209                }
210                return Err(error);
211            }
212        }
213        self.commit_auto_transaction()?;
214        Ok(result)
215    }
216
217    fn drop_edge_index(&mut self, space: &str, index: &str) -> Result<bool, StorageError> {
218        let space_id = self.inner.get_space_id(space)?;
219        let (fields, schema_name) = self
220            .inner
221            .get_edge_index(space, index)?
222            .map(|definition| {
223                (
224                    definition
225                        .fields
226                        .into_iter()
227                        .map(|field| field.name)
228                        .collect(),
229                    definition.schema_name,
230                )
231            })
232            .unwrap_or_else(|| (Vec::new(), index.to_string()));
233        self.validate_schema_sync_context()?;
234        let result = self.inner.drop_edge_index(space, index)?;
235        if result {
236            if let Err(error) =
237                self.stage_index_drop(space_id, index, &schema_name, "edge", &fields)
238            {
239                if let Some(transaction_id) = self.get_current_txn_id() {
240                    let _ = self.abort_transaction_fact(transaction_id);
241                }
242                return Err(error);
243            }
244        }
245        self.commit_auto_transaction()?;
246        Ok(result)
247    }
248}
linkrs_storage/engine/sync_wrapper/schema.rs

linkrs_storage/engine/sync_wrapper/
schema.rs
