linkrs_storage/client/
traits_writer.rs

1//! Mutation and commit operations.
2
3use linkrs_core::types::{InsertEdgeInfo, InsertVertexInfo, TransactionId, UpdateInfo, VertexId};
4use linkrs_core::{Edge, EdgeDeleteKey, StorageError, StorageResult, Vertex};
5
6/// Write operations for vertex and edge data.
7pub trait StorageWriter: Send + Sync + std::fmt::Debug {
8    fn insert_vertex(&mut self, space: &str, vertex: Vertex) -> Result<VertexId, StorageError>;
9    fn update_vertex(&mut self, space: &str, vertex: Vertex) -> Result<(), StorageError>;
10    fn update_vertex_replace(&mut self, space: &str, vertex: Vertex) -> Result<(), StorageError> {
11        let _ = (space, vertex);
12        Err(StorageError::not_supported(
13            "Row replacement is not supported by this storage implementation",
14        ))
15    }
16    /// Delete one vertex without touching its incident edges. The orphaned
17    /// edge rows stay visible until repaired: prefer
18    /// `delete_vertex_with_edges` for cascade deletes, or run
19    /// `repair_dangling_edges` afterwards.
20    fn delete_vertex(&mut self, space: &str, tag: &str, id: &VertexId) -> Result<(), StorageError>;
21    fn delete_vertex_with_edges(
22        &mut self,
23        space: &str,
24        tag: &str,
25        id: &VertexId,
26    ) -> Result<(), StorageError>;
27    fn batch_delete_vertices_with_edges(
28        &mut self,
29        space: &str,
30        tag: &str,
31        ids: &[VertexId],
32    ) -> Result<usize, StorageError>;
33    fn batch_insert_vertices(
34        &mut self,
35        space: &str,
36        vertices: Vec<Vertex>,
37    ) -> Result<Vec<VertexId>, StorageError>;
38    /// Minimal default: delegates when splitting is on and enforces the
39    /// single-request limit when off. The engine override chunks by label.
40    fn batch_insert_vertices_with_split(
41        &mut self,
42        space: &str,
43        vertices: Vec<Vertex>,
44        auto_split: bool,
45    ) -> Result<Vec<VertexId>, StorageError> {
46        if auto_split {
47            return self.batch_insert_vertices(space, vertices);
48        }
49        if vertices.len() > crate::vertex::MAX_WRITE_SCOPE_KEYS {
50            return Err(StorageError::new(
51                linkrs_core::error::storage::StorageErrorKind::CapacityExceeded,
52                format!(
53                    "batch holds {} rows above the single-request limit {}: enable split batching by label into smaller chunks instead of growing one request",
54                    vertices.len(),
55                    crate::vertex::MAX_WRITE_SCOPE_KEYS,
56                ),
57            ));
58        }
59        self.batch_insert_vertices(space, vertices)
60    }
61    fn insert_edge(&mut self, space: &str, edge: Edge) -> Result<(), StorageError>;
62    fn update_edge(&mut self, space: &str, edge: Edge) -> Result<(), StorageError>;
63    fn update_edge_replace(&mut self, space: &str, edge: Edge) -> Result<(), StorageError> {
64        let _ = (space, edge);
65        Err(StorageError::not_supported(
66            "Row replacement is not supported by this storage implementation",
67        ))
68    }
69    fn delete_edge(
70        &mut self,
71        space: &str,
72        src: &VertexId,
73        dst: &VertexId,
74        edge_type: &str,
75        rank: i64,
76    ) -> Result<(), StorageError>;
77    fn batch_insert_edges(&mut self, space: &str, edges: Vec<Edge>) -> Result<(), StorageError>;
78    fn batch_delete_edges(
79        &mut self,
80        space: &str,
81        deletes: &[EdgeDeleteKey],
82    ) -> Result<usize, StorageError>;
83
84    fn insert_vertex_data(
85        &mut self,
86        space: &str,
87        info: &InsertVertexInfo,
88    ) -> Result<bool, StorageError>;
89    fn insert_edge_data(
90        &mut self,
91        space: &str,
92        info: &InsertEdgeInfo,
93    ) -> Result<bool, StorageError>;
94    fn delete_vertex_data(
95        &mut self,
96        space: &str,
97        tag: &str,
98        vertex_id: &str,
99    ) -> Result<bool, StorageError>;
100    fn delete_edge_data(
101        &mut self,
102        space: &str,
103        src: &str,
104        dst: &str,
105        rank: i64,
106    ) -> Result<bool, StorageError>;
107    fn update_data(
108        &mut self,
109        space: &str,
110        space_id: u64,
111        info: &UpdateInfo,
112    ) -> Result<bool, StorageError>;
113}
114
115pub trait StorageCommitOps: Send + Sync + std::fmt::Debug {
116    fn commit_staged_writes(
117        &self,
118        transaction_id: TransactionId,
119        intents: &[linkrs_core::wal::OutboxIntent],
120    ) -> StorageResult<linkrs_core::types::CommitLsn>;
121
122    fn abort_staged_writes(&self, transaction_id: TransactionId) -> StorageResult<()>;
123
124    fn commit_staged_writes_with_durability(
125        &self,
126        transaction_id: TransactionId,
127        intents: &[linkrs_core::wal::OutboxIntent],
128        _durability: linkrs_core::types::DurabilityLevel,
129    ) -> StorageResult<linkrs_core::types::CommitLsn> {
130        self.commit_staged_writes(transaction_id, intents)
131    }
132
133    fn recover_outbox_projection(
134        &self,
135        sync_manager: &linkrs_sync::SyncManager,
136    ) -> StorageResult<usize>;
137}
linkrs_storage/client/traits_writer.rs

linkrs_storage/client/
traits_writer.rs
