1use std::sync::Arc;
2
3use linkrs_core::types::{EdgeTypeInfo, Index, SpaceInfo, TagInfo, VertexId};
4use linkrs_core::{Edge, EdgeDirection, StorageError, Value, Vertex};
5
6use crate::cursor::{EdgeCursor, IndexCursor, IndexRow, IndexScanPlan, ScanOptions, VertexCursor};
7use crate::StorageReader;
8
9use super::GraphStorage;
10use super::{cursor_impl, index_manager, reader};
11
12impl StorageReader for GraphStorage {
13 fn get_vertex(
14 &self,
15 space: &str,
16 tag: &str,
17 id: &VertexId,
18 ) -> Result<Option<Vertex>, StorageError> {
19 reader::get_vertex(&self.ctx, space, tag, id)
20 }
21
22 fn layout_version(&self) -> u64 {
23 self.ctx.layout_version()
24 }
25
26 fn vertex_id_domain(&self, space: &str) -> Option<std::ops::Range<i64>> {
27 self.ctx.vertex_id_domain(space)
28 }
29
30 fn get_vertex_projected(
31 &self,
32 space: &str,
33 tag: &str,
34 id: &VertexId,
35 projection: &[std::sync::Arc<str>],
36 ) -> Result<Option<Vertex>, StorageError> {
37 reader::get_vertex_projected(&self.ctx, space, tag, id, projection)
38 }
39
40 fn scan_vertices(&self, space: &str) -> Result<Vec<Vertex>, StorageError> {
41 reader::scan_vertices(&self.ctx, space)
42 }
43
44 fn scan_vertices_by_tag(&self, space: &str, tag: &str) -> Result<Vec<Vertex>, StorageError> {
45 reader::scan_vertices_by_tag(&self.ctx, space, tag)
46 }
47
48 fn scan_vertices_by_tag_paginated(
49 &self,
50 space: &str,
51 tag: &str,
52 offset: usize,
53 limit: usize,
54 ) -> Result<Vec<Vertex>, StorageError> {
55 let options = crate::cursor::ScanOptions::new()
56 .with_tag(tag.to_string())
57 .with_offset(offset)
58 .with_limit(limit);
59 match self.create_vertex_cursor(space, &options) {
60 Ok(mut cursor) => cursor.next_batch(limit),
61 Err(e) if e.kind() == linkrs_core::error::storage::StorageErrorKind::NotSupported => {
62 let all = reader::scan_vertices_by_tag(&self.ctx, space, tag)?;
63 Ok(all.into_iter().skip(offset).take(limit).collect())
64 }
65 Err(e) => Err(e),
66 }
67 }
68
69 fn scan_vertices_by_prop(
70 &self,
71 space: &str,
72 tag: &str,
73 prop: &str,
74 value: &Value,
75 ) -> Result<Vec<Vertex>, StorageError> {
76 reader::scan_vertices_by_prop(&self.ctx, space, tag, prop, value)
77 }
78
79 fn get_edge(
80 &self,
81 space: &str,
82 src: &VertexId,
83 dst: &VertexId,
84 edge_type: &str,
85 rank: i64,
86 ) -> Result<Option<Edge>, StorageError> {
87 reader::get_edge(&self.ctx, space, src, dst, edge_type, rank)
88 }
89
90 fn get_edge_projected(
91 &self,
92 space: &str,
93 src: &VertexId,
94 dst: &VertexId,
95 edge_type: &str,
96 rank: i64,
97 projection: &[Arc<str>],
98 ) -> Result<Option<Edge>, StorageError> {
99 reader::get_edge_projected(&self.ctx, space, src, dst, edge_type, rank, projection)
100 }
101
102 fn get_node_edges(
103 &self,
104 space: &str,
105 node_id: &VertexId,
106 direction: EdgeDirection,
107 edge_types: &[String],
108 ) -> Result<Vec<Edge>, StorageError> {
109 reader::get_node_edges(&self.ctx, space, node_id, direction, edge_types)
110 }
111
112 fn get_node_edges_projected(
113 &self,
114 space: &str,
115 node_id: &VertexId,
116 direction: EdgeDirection,
117 edge_types: &[String],
118 projection: Option<&[Arc<str>]>,
119 limit: Option<usize>,
120 ) -> Result<Vec<Edge>, StorageError> {
121 reader::get_node_edges_projected(
122 &self.ctx, space, node_id, direction, edge_types, projection, limit,
123 )
124 }
125
126 fn get_vertices_batch(
127 &self,
128 space: &str,
129 tag: &str,
130 ids: &[VertexId],
131 ) -> Result<Vec<Option<Vertex>>, StorageError> {
132 reader::get_vertices_batch(&self.ctx, space, tag, ids)
133 }
134
135 fn get_vertices_projected_batch(
136 &self,
137 space: &str,
138 tag: &str,
139 ids: &[VertexId],
140 projection: &[Arc<str>],
141 ) -> Result<Vec<Option<Vertex>>, StorageError> {
142 reader::get_vertices_projected_batch(&self.ctx, space, tag, ids, projection)
143 }
144
145 fn neighbor_dst_ids_batch(
146 &self,
147 space: &str,
148 src_ids: &[VertexId],
149 direction: EdgeDirection,
150 edge_types: &[String],
151 ) -> Result<Vec<Vec<VertexId>>, StorageError> {
152 reader::neighbor_dst_ids_batch(&self.ctx, space, src_ids, direction, edge_types)
153 }
154
155 fn out_degree_batch(
156 &self,
157 space: &str,
158 src_ids: &[VertexId],
159 direction: EdgeDirection,
160 edge_types: &[String],
161 ) -> Result<Vec<usize>, StorageError> {
162 reader::out_degree_batch(&self.ctx, space, src_ids, direction, edge_types)
163 }
164
165 fn scan_edges_by_type(&self, space: &str, edge_type: &str) -> Result<Vec<Edge>, StorageError> {
166 reader::scan_edges_by_type(&self.ctx, space, edge_type)
167 }
168
169 fn scan_edges_by_type_parallel(
170 &self,
171 space: &str,
172 edge_type: &str,
173 shards: usize,
174 ) -> Result<Vec<Edge>, StorageError> {
175 reader::scan_edges_by_type_parallel(&self.ctx, space, edge_type, shards)
176 }
177
178 fn scan_all_edges(&self, space: &str) -> Result<Vec<Edge>, StorageError> {
179 reader::scan_all_edges(&self.ctx, space)
180 }
181
182 fn scan_edges_by_type_paginated(
183 &self,
184 space: &str,
185 edge_type: &str,
186 offset: usize,
187 limit: usize,
188 ) -> Result<Vec<Edge>, StorageError> {
189 reader::scan_edges_by_type_paginated(&self.ctx, space, edge_type, offset, limit)
190 }
191
192 fn count_vertices_by_tag(&self, space: &str, tag: &str) -> Result<u64, StorageError> {
193 reader::count_vertices_by_tag(&self.ctx, space, tag)
194 }
195
196 fn count_edges_by_type(&self, space: &str, edge_type: &str) -> Result<u64, StorageError> {
197 reader::count_edges_by_type(&self.ctx, space, edge_type)
198 }
199
200 fn lookup_index(
201 &self,
202 space: &str,
203 index_name: &str,
204 value: &Value,
205 ) -> Result<Vec<Value>, StorageError> {
206 index_manager::lookup_index(&self.ctx, space, index_name, value)
207 }
208
209 fn enable_edge_property_index(
210 &self,
211 space: &str,
212 edge_type: &str,
213 pool_capacity: u64,
214 ) -> Result<bool, StorageError> {
215 reader::enable_edge_property_index(&self.ctx, space, edge_type, pool_capacity)
216 }
217
218 fn has_edge_property_index(&self, space: &str, edge_type: &str) -> Result<bool, StorageError> {
219 reader::has_edge_property_index(&self.ctx, space, edge_type)
220 }
221
222 fn disable_edge_property_index(
223 &self,
224 space: &str,
225 edge_type: &str,
226 ) -> Result<(), StorageError> {
227 reader::disable_edge_property_index(&self.ctx, space, edge_type)
228 }
229
230 fn lookup_edges_by_property_range(
231 &self,
232 space: &str,
233 edge_type: &str,
234 prop_name: &str,
235 lower: Option<&Value>,
236 upper: Option<&Value>,
237 include_lower: bool,
238 include_upper: bool,
239 ) -> Result<Vec<Edge>, StorageError> {
240 reader::lookup_edges_by_property_range(
241 &self.ctx,
242 space,
243 edge_type,
244 prop_name,
245 lower,
246 upper,
247 include_lower,
248 include_upper,
249 )
250 }
251
252 fn get_vertex_with_schema(
253 &self,
254 space: &str,
255 tag: &str,
256 id: &Value,
257 ) -> Result<Option<(TagInfo, Vec<u8>)>, StorageError> {
258 reader::get_vertex_with_schema(&self.ctx, space, tag, id)
259 }
260
261 fn get_edge_with_schema(
262 &self,
263 space: &str,
264 edge_type: &str,
265 src: &Value,
266 dst: &Value,
267 ) -> Result<Option<(EdgeTypeInfo, Vec<u8>)>, StorageError> {
268 reader::get_edge_with_schema(&self.ctx, space, edge_type, src, dst)
269 }
270
271 fn scan_vertices_with_schema(
272 &self,
273 space: &str,
274 tag: &str,
275 ) -> Result<Vec<(TagInfo, Vec<u8>)>, StorageError> {
276 reader::scan_vertices_with_schema(&self.ctx, space, tag)
277 }
278
279 fn scan_edges_with_schema(
280 &self,
281 space: &str,
282 edge_type: &str,
283 ) -> Result<Vec<(EdgeTypeInfo, Vec<u8>)>, StorageError> {
284 reader::scan_edges_with_schema(&self.ctx, space, edge_type)
285 }
286
287 fn get_space(&self, space: &str) -> Result<Option<SpaceInfo>, StorageError> {
288 self.ctx.schema_manager().get_space(space)
289 }
290
291 fn get_space_by_id(&self, space_id: u64) -> Result<Option<SpaceInfo>, StorageError> {
292 self.ctx.schema_manager().get_space_by_id(space_id)
293 }
294
295 fn list_spaces(&self) -> Result<Vec<SpaceInfo>, StorageError> {
296 self.ctx.schema_manager().list_spaces()
297 }
298
299 fn get_space_id(&self, space: &str) -> Result<u64, StorageError> {
300 self.ctx.schema_manager().get_space_id(space)
301 }
302
303 fn space_exists(&self, space: &str) -> bool {
304 self.ctx
305 .schema_manager()
306 .get_space(space)
307 .ok()
308 .flatten()
309 .is_some()
310 }
311
312 fn get_tag(&self, space: &str, tag: &str) -> Result<Option<TagInfo>, StorageError> {
313 self.ctx.schema_manager().get_tag(space, tag)
314 }
315
316 fn list_tags(&self, space: &str) -> Result<Vec<TagInfo>, StorageError> {
317 self.ctx.schema_manager().list_tags(space)
318 }
319
320 fn get_edge_type(
321 &self,
322 space: &str,
323 edge_type: &str,
324 ) -> Result<Option<EdgeTypeInfo>, StorageError> {
325 self.ctx.schema_manager().get_edge_type(space, edge_type)
326 }
327
328 fn list_edge_types(&self, space: &str) -> Result<Vec<EdgeTypeInfo>, StorageError> {
329 self.ctx.schema_manager().list_edge_types(space)
330 }
331
332 fn get_tag_index(&self, space: &str, index_name: &str) -> Result<Option<Index>, StorageError> {
333 index_manager::get_tag_index(&self.ctx, space, index_name)
334 }
335
336 fn list_tag_indexes(&self, space: &str) -> Result<Vec<Index>, StorageError> {
337 index_manager::list_tag_indexes(&self.ctx, space)
338 }
339
340 fn get_edge_index(&self, space: &str, index_name: &str) -> Result<Option<Index>, StorageError> {
341 index_manager::get_edge_index(&self.ctx, space, index_name)
342 }
343
344 fn list_edge_indexes(&self, space: &str) -> Result<Vec<Index>, StorageError> {
345 index_manager::list_edge_indexes(&self.ctx, space)
346 }
347
348 fn get_vertex_version_history(
349 &self,
350 space: &str,
351 tag: &str,
352 ) -> Result<Option<crate::LabelVersionHistory>, StorageError> {
353 let tag_info = self.ctx.schema_manager().get_tag(space, tag)?;
354 let tag_info = tag_info.ok_or_else(|| StorageError::label_not_found(tag.to_string()))?;
355
356 let history = self.ctx.data_store().with_vertex_tables(|vertex_tables| {
357 let table = vertex_tables
358 .get(&tag_info.tag_id)
359 .ok_or_else(|| StorageError::label_not_found(tag.to_string()))?;
360 let version_history = table.version_history_ref();
361 let guard = version_history
362 .lock()
363 .map_err(|_| StorageError::db_error("Failed to lock version_history"))?;
364 Ok::<_, StorageError>(guard.clone())
365 })?;
366 Ok(Some(history))
367 }
368
369 fn get_edge_version_history(
370 &self,
371 space: &str,
372 edge_type: &str,
373 ) -> Result<Option<crate::LabelVersionHistory>, StorageError> {
374 let edge_info = self.ctx.schema_manager().get_edge_type(space, edge_type)?;
375 let edge_info =
376 edge_info.ok_or_else(|| StorageError::label_not_found(edge_type.to_string()))?;
377
378 let keys = {
379 self.ctx
380 .data_store()
381 .catalog_read_snapshot()
382 .with_edge_label_index(|edge_label_index| {
383 edge_label_index
384 .get(&edge_info.edge_type_id)
385 .cloned()
386 .ok_or_else(|| StorageError::label_not_found(edge_type.to_string()))
387 })?
388 };
389
390 if keys.is_empty() {
391 return Err(StorageError::label_not_found(edge_type.to_string()));
392 }
393
394 let key = &keys[0];
395
396 let history = self.ctx.data_store().with_single_edge_table(key, |table| {
397 let version_history = table.version_history_ref();
398 let guard = version_history
399 .lock()
400 .map_err(|_| StorageError::db_error("Failed to lock version_history"))?;
401 Ok::<_, StorageError>(guard.clone())
402 })?;
403 Ok(Some(history))
404 }
405
406 fn get_vertex_schema_changes(
407 &self,
408 space: &str,
409 tag: &str,
410 from_version: u64,
411 to_version: u64,
412 ) -> Result<Vec<crate::PropertyChange>, StorageError> {
413 let history = self.get_vertex_version_history(space, tag)?;
414 let history = history.ok_or_else(|| StorageError::label_not_found(tag.to_string()))?;
415
416 let mut changes = Vec::new();
417 for version in history.get_versions() {
418 if version > from_version && version <= to_version {
419 if let Some(version_changes) = history.change_log.get_version_changes(version) {
420 changes.extend(version_changes.iter().cloned());
421 }
422 }
423 }
424
425 Ok(changes)
426 }
427
428 fn get_edge_schema_changes(
429 &self,
430 space: &str,
431 edge_type: &str,
432 from_version: u64,
433 to_version: u64,
434 ) -> Result<Vec<crate::PropertyChange>, StorageError> {
435 let history = self.get_edge_version_history(space, edge_type)?;
436 let history =
437 history.ok_or_else(|| StorageError::label_not_found(edge_type.to_string()))?;
438
439 let mut changes = Vec::new();
440 for version in history.get_versions() {
441 if version > from_version && version <= to_version {
442 if let Some(version_changes) = history.change_log.get_version_changes(version) {
443 changes.extend(version_changes.iter().cloned());
444 }
445 }
446 }
447
448 Ok(changes)
449 }
450
451 fn detect_vertex_breaking_changes(
452 &self,
453 space: &str,
454 tag: &str,
455 from_version: u64,
456 to_version: u64,
457 ) -> Result<Vec<crate::PropertyChange>, StorageError> {
458 let changes = self.get_vertex_schema_changes(space, tag, from_version, to_version)?;
459 Ok(changes.into_iter().filter(|c| c.is_breaking()).collect())
460 }
461
462 fn detect_edge_breaking_changes(
463 &self,
464 space: &str,
465 edge_type: &str,
466 from_version: u64,
467 to_version: u64,
468 ) -> Result<Vec<crate::PropertyChange>, StorageError> {
469 let changes = self.get_edge_schema_changes(space, edge_type, from_version, to_version)?;
470 Ok(changes.into_iter().filter(|c| c.is_breaking()).collect())
471 }
472
473 fn create_vertex_cursor(
474 &self,
475 space: &str,
476 options: &ScanOptions,
477 ) -> Result<Box<dyn VertexCursor>, StorageError> {
478 let cursor =
479 cursor_impl::GraphVertexCursor::new(self.ctx.clone(), space.to_string(), options)?;
480 Ok(Box::new(cursor))
481 }
482
483 fn create_edge_cursor(
484 &self,
485 space: &str,
486 options: &ScanOptions,
487 ) -> Result<Box<dyn EdgeCursor>, StorageError> {
488 cursor_impl::create_edge_cursor(self.ctx.clone(), space, options)
489 }
490
491 fn create_index_cursor(
492 &self,
493 plan: &IndexScanPlan,
494 ) -> Result<Box<dyn IndexCursor<Row = IndexRow>>, StorageError> {
495 let tag_indexes = self.list_tag_indexes(&plan.space)?;
496 let edge_indexes = self.list_edge_indexes(&plan.space)?;
497 let index_id = plan.index_id;
498 if let Some(index) = tag_indexes.into_iter().find(|index| index.id == index_id) {
499 let space_id = self.ctx.schema_manager().get_space_id(&plan.space)?;
500 let space_name = plan.space.clone();
501 let tag_name = index.schema_name.clone();
502 let ctx = self.ctx.clone();
503 let stale_checker: Option<crate::index::types::StaleChecker> =
504 Some(Arc::new(move |entity_ref| match entity_ref {
505 linkrs_core::wal::EntityRef::Vertex(vid) => {
506 reader::get_vertex(&ctx, &space_name, &tag_name, vid)
507 .ok()
508 .flatten()
509 .is_some()
510 }
511 linkrs_core::wal::EntityRef::Edge { .. } => true,
512 }));
513 let cursor = self
514 .ctx
515 .index_data_manager()
516 .read()
517 .open_tag_index_cursor_full(space_id, &index, plan, stale_checker, None)?;
518 return Ok(Box::new(cursor));
519 }
520
521 if let Some(index) = edge_indexes.into_iter().find(|index| index.id == index_id) {
522 let space_id = self.ctx.schema_manager().get_space_id(&plan.space)?;
523 let cursor = self
524 .ctx
525 .index_data_manager()
526 .read()
527 .open_edge_index_cursor_full(space_id, &index, plan, None, None)?;
528 return Ok(Box::new(cursor));
529 }
530
531 Err(StorageError::not_found(format!(
532 "Index {} not found in space {}",
533 plan.index_id, plan.space
534 )))
535 }
536
537 fn list_migration_history(
538 &self,
539 space: &str,
540 label: &str,
541 is_edge: bool,
542 ) -> Result<Vec<crate::MigrationHistoryRecord>, StorageError> {
543 Ok(self.ctx.list_migration_history(space, label, is_edge))
544 }
545
546 fn get_applied_versions(
547 &self,
548 space: &str,
549 label: &str,
550 is_edge: bool,
551 ) -> Result<Vec<u64>, StorageError> {
552 Ok(self.ctx.get_applied_versions(space, label, is_edge))
553 }
554
555 fn record_migration_history(
556 &self,
557 record: crate::MigrationHistoryRecord,
558 ) -> Result<(), StorageError> {
559 self.ctx.record_migration_history(record)
560 }
561
562 fn list_all_migration_history(
563 &self,
564 ) -> Result<Vec<crate::MigrationHistoryRecord>, StorageError> {
565 Ok(self.ctx.list_all_migration_history())
566 }
567}