1use linkrs_core::types::{EdgeTypeInfo, Index, LabelId, PropertyDef, SpaceInfo, TagInfo};
2use linkrs_core::StorageError;
3use std::sync::Arc;
4
5use crate::{StorageOperationContext, StorageSchemaOps};
6
7use super::GraphStorage;
8use super::{index_manager, ops, reader, schema_engine, schema_writer};
9
10impl StorageSchemaOps for GraphStorage {
11 fn create_space(&mut self, space: &mut SpaceInfo) -> Result<bool, StorageError> {
12 self.ctx.check_write_admission()?;
13 schema_writer::create_space(&self.ctx, space)
14 }
15
16 fn drop_space(&mut self, space: &str) -> Result<bool, StorageError> {
17 self.ctx.check_write_admission()?;
18 schema_writer::drop_space(&self.ctx, space)
19 }
20
21 fn clear_space(&mut self, space: &str) -> Result<bool, StorageError> {
22 self.ctx.check_write_admission()?;
23 schema_writer::clear_space(&self.ctx, space)
24 }
25
26 fn create_tag(&mut self, space: &str, tag: &TagInfo) -> Result<u32, StorageError> {
27 self.ctx.check_write_admission()?;
28 schema_writer::create_tag(&self.ctx, space, tag)
29 }
30
31 fn create_tag_with_estimate(
32 &mut self,
33 space: &str,
34 tag: &TagInfo,
35 estimated_rows: Option<u64>,
36 ) -> Result<u32, StorageError> {
37 self.ctx.check_write_admission()?;
38 schema_writer::create_tag_with_estimate(&self.ctx, space, tag, estimated_rows)
39 }
40
41 fn alter_tag(
42 &mut self,
43 space: &str,
44 tag_name: &str,
45 additions: Vec<PropertyDef>,
46 deletions: Vec<String>,
47 new_comment: Option<String>,
48 ) -> Result<bool, StorageError> {
49 self.ctx.check_write_admission()?;
50 schema_writer::alter_tag(
51 &self.ctx,
52 space,
53 tag_name,
54 additions,
55 deletions.into_iter().map(Arc::from).collect(),
56 new_comment,
57 )
58 }
59
60 fn rename_vertex_property(
61 &mut self,
62 label: LabelId,
63 old_name: &str,
64 new_name: &str,
65 ) -> Result<(), StorageError> {
66 self.ctx.check_write_admission()?;
67 schema_engine::rename_vertex_property(&self.ctx, label, old_name, new_name)
68 }
69
70 fn rename_tag_property(
71 &mut self,
72 space: &str,
73 tag: &str,
74 old_name: &str,
75 new_name: &str,
76 ) -> Result<bool, StorageError> {
77 self.ctx.check_write_admission()?;
78 let renamed = self
79 .ctx
80 .schema_manager()
81 .rename_tag_property(space, tag, old_name, new_name)?;
82 if renamed {
83 let label_id = ops::tag_label_id(&self.ctx, space, tag)?
84 .ok_or_else(|| StorageError::label_not_found(tag.to_string()))?;
85 schema_engine::rename_vertex_property(&self.ctx, label_id, old_name, new_name)?;
86 }
87 Ok(renamed)
88 }
89
90 fn rename_tag(
91 &mut self,
92 space: &str,
93 old_name: &str,
94 new_name: &str,
95 ) -> Result<bool, StorageError> {
96 self.ctx.check_write_admission()?;
97 let label_id = ops::tag_label_id(&self.ctx, space, old_name)?
98 .ok_or_else(|| StorageError::label_not_found(old_name.to_string()))?;
99 let renamed = self
100 .ctx
101 .schema_manager()
102 .rename_tag(space, old_name, new_name)?;
103 if renamed {
104 schema_engine::rename_vertex_label(&self.ctx, label_id, new_name)?;
105 }
106 Ok(renamed)
107 }
108
109 fn drop_tag(&mut self, space: &str, tag: &str) -> Result<bool, StorageError> {
110 self.ctx.check_write_admission()?;
111 schema_writer::drop_tag(&self.ctx, space, tag)
112 }
113
114 fn create_edge_type(
115 &mut self,
116 space: &str,
117 edge_type: &EdgeTypeInfo,
118 ) -> Result<u32, StorageError> {
119 self.ctx.check_write_admission()?;
120 schema_writer::create_edge_type(&self.ctx, space, edge_type)
121 }
122
123 fn alter_edge_type(
124 &mut self,
125 space: &str,
126 edge_type_name: &str,
127 additions: Vec<PropertyDef>,
128 deletions: Vec<String>,
129 new_comment: Option<String>,
130 ) -> Result<bool, StorageError> {
131 self.ctx.check_write_admission()?;
132 schema_writer::alter_edge_type(
133 &self.ctx,
134 space,
135 edge_type_name,
136 additions,
137 deletions.into_iter().map(Arc::from).collect(),
138 new_comment,
139 )
140 }
141
142 fn rename_edge_type(
143 &mut self,
144 space: &str,
145 old_name: &str,
146 new_name: &str,
147 ) -> Result<bool, StorageError> {
148 self.ctx.check_write_admission()?;
149 let label_id = ops::edge_label_id(&self.ctx, space, old_name)?
150 .ok_or_else(|| StorageError::label_not_found(old_name.to_string()))?;
151 let renamed = self
152 .ctx
153 .schema_manager()
154 .rename_edge_type(space, old_name, new_name)?;
155 if renamed {
156 schema_engine::rename_edge_label(&self.ctx, label_id, new_name)?;
157 }
158 Ok(renamed)
159 }
160
161 fn rename_edge_property(
162 &mut self,
163 space: &str,
164 edge_type: &str,
165 old_name: &str,
166 new_name: &str,
167 ) -> Result<bool, StorageError> {
168 self.ctx.check_write_admission()?;
169 let renamed = self
170 .ctx
171 .schema_manager()
172 .rename_edge_property(space, edge_type, old_name, new_name)?;
173 if renamed {
174 let label_id = ops::edge_label_id(&self.ctx, space, edge_type)?
175 .ok_or_else(|| StorageError::label_not_found(edge_type.to_string()))?;
176 schema_engine::rename_edge_property(&self.ctx, label_id, old_name, new_name)?;
177 }
178 Ok(renamed)
179 }
180
181 fn update_edge_endpoints(
182 &mut self,
183 space: &str,
184 edge_type: &str,
185 src_tag_name: &str,
186 dst_tag_name: &str,
187 ) -> Result<bool, StorageError> {
188 self.ctx.check_write_admission()?;
189 schema_writer::update_edge_endpoints(
190 &self.ctx,
191 space,
192 edge_type,
193 src_tag_name,
194 dst_tag_name,
195 )
196 }
197
198 fn drop_edge_type(&mut self, space: &str, edge_type: &str) -> Result<bool, StorageError> {
199 self.ctx.check_write_admission()?;
200 schema_writer::drop_edge_type(&self.ctx, space, edge_type)
201 }
202
203 fn create_tag_index(&mut self, space: &str, index: &Index) -> Result<bool, StorageError> {
204 self.ctx.check_write_admission()?;
205 let created = schema_writer::create_tag_index(&self.ctx, space, index)?;
206 if created {
207 self.rebuild_tag_index(space, &index.name)?;
208 }
209 Ok(created)
210 }
211
212 fn drop_tag_index(&mut self, space: &str, index_name: &str) -> Result<bool, StorageError> {
213 self.ctx.check_write_admission()?;
214 schema_writer::drop_tag_index(&self.ctx, space, index_name)
215 }
216
217 fn rebuild_tag_index(&mut self, space: &str, index_name: &str) -> Result<bool, StorageError> {
218 self.ctx.check_write_admission()?;
219 let rebuild_gate = self.ctx.index_data_manager().read().rebuild_gate();
220 let _rebuild_guard = rebuild_gate.write();
221 let snapshot_timestamp = self.ctx.get_read_timestamp();
222 let start_lsn = match index_manager::current_wal_lsn(&self.ctx) {
223 linkrs_core::types::CommitLsn::ZERO => linkrs_core::types::CommitLsn::new(1),
224 lsn => lsn,
225 };
226 let snapshot_ctx = self.ctx.with_operation_context(StorageOperationContext {
227 transaction_id: None,
228 read_timestamp: snapshot_timestamp,
229 write_timestamp: None,
230 read_only: true,
231 auto_commit: false,
232 mutation_recorder: None,
233 auto_commit_group_start: None,
234 auto_commit_staging_start: None,
235 auto_commit_wal_start: 0,
236 });
237 let vertices = reader::scan_vertices(&snapshot_ctx, space)?;
238 let result = index_manager::rebuild_tag_index(
239 &self.ctx,
240 space,
241 index_name,
242 &vertices,
243 linkrs_core::types::SnapshotTimestamp::new(snapshot_timestamp),
244 start_lsn,
245 );
246 if let Some(stats) = self.ctx.stats_manager() {
247 if result.is_err() {
248 stats.record_generation_rebuild_failure();
249 }
250 }
251 result
252 }
253
254 fn create_edge_index(&mut self, space: &str, index: &Index) -> Result<bool, StorageError> {
255 self.ctx.check_write_admission()?;
256 let created = schema_writer::create_edge_index(&self.ctx, space, index)?;
257 if created {
258 self.rebuild_edge_index(space, &index.name)?;
259 }
260 Ok(created)
261 }
262
263 fn drop_edge_index(&mut self, space: &str, index_name: &str) -> Result<bool, StorageError> {
264 self.ctx.check_write_admission()?;
265 schema_writer::drop_edge_index(&self.ctx, space, index_name)
266 }
267
268 fn rebuild_edge_index(&mut self, space: &str, index_name: &str) -> Result<bool, StorageError> {
269 self.ctx.check_write_admission()?;
270 let rebuild_gate = self.ctx.index_data_manager().read().rebuild_gate();
271 let _rebuild_guard = rebuild_gate.write();
272 let snapshot_timestamp = self.ctx.get_read_timestamp();
273 let start_lsn = match index_manager::current_wal_lsn(&self.ctx) {
274 linkrs_core::types::CommitLsn::ZERO => linkrs_core::types::CommitLsn::new(1),
275 lsn => lsn,
276 };
277 let snapshot_ctx = self.ctx.with_operation_context(StorageOperationContext {
278 transaction_id: None,
279 read_timestamp: snapshot_timestamp,
280 write_timestamp: None,
281 read_only: true,
282 auto_commit: false,
283 mutation_recorder: None,
284 auto_commit_group_start: None,
285 auto_commit_staging_start: None,
286 auto_commit_wal_start: 0,
287 });
288 let edges = reader::scan_all_edges(&snapshot_ctx, space)?;
289 let result = index_manager::rebuild_edge_index(
290 &self.ctx,
291 space,
292 index_name,
293 &edges,
294 linkrs_core::types::SnapshotTimestamp::new(snapshot_timestamp),
295 start_lsn,
296 );
297 if let Some(stats) = self.ctx.stats_manager() {
298 if result.is_err() {
299 stats.record_generation_rebuild_failure();
300 }
301 }
302 result
303 }
304}