linkrs_storage/edge/csr_with_properties/
write.rs

1use super::CsrWithProperties;
2use linkrs_core::types::{EdgeId, Timestamp};
3use linkrs_core::{StorageError, StorageResult, Value};
4use std::sync::Arc;
5
6impl CsrWithProperties {
7    /// Insert properties for an edge and associate the row with `edge_id`.
8    pub fn insert_for_edge(
9        &mut self,
10        edge_id: EdgeId,
11        values: &[(Arc<str>, Value)],
12        create_ts: Timestamp,
13    ) -> StorageResult<()> {
14        self.reject_inline()?;
15        let row_idx = self.allocate_row(values, create_ts)?;
16        self.map_insert(edge_id, row_idx)?;
17        self.ensure_row_aux_len(row_idx + 1);
18        self.row_to_edge[row_idx] = Some(edge_id);
19        Ok(())
20    }
21
22    /// Insert properties for an edge from pre-resolved column positions.
23    ///
24    /// Hot write-path entry: the caller resolved every column once through
25    /// the schema index, so this performs no name lookup, string clone or
26    /// string comparison per edge. Positions must be store column positions;
27    /// out-of-range entries fail loudly through `allocate_row_at`.
28    pub fn insert_for_edge_at(
29        &mut self,
30        edge_id: EdgeId,
31        positioned: &[(usize, Value)],
32        create_ts: Timestamp,
33    ) -> StorageResult<()> {
34        self.reject_inline()?;
35        let row_idx = self.allocate_row_at(positioned, create_ts)?;
36        self.map_insert(edge_id, row_idx)?;
37        self.ensure_row_aux_len(row_idx + 1);
38        self.row_to_edge[row_idx] = Some(edge_id);
39        Ok(())
40    }
41
42    /// Edge-aware property update: lookup row via `edge_id`.
43    pub fn set_property_for_edge(
44        &mut self,
45        edge_id: EdgeId,
46        name: &str,
47        value: Option<Value>,
48        ts: Timestamp,
49    ) -> StorageResult<()> {
50        self.reject_inline()?;
51        let pos = self
52            .mapped_row(edge_id)
53            .ok_or_else(|| StorageError::invalid_offset(0))?;
54        self.set_property_at_row(pos, name, value, ts)
55    }
56
57    pub fn set_property_by_id_for_edge(
58        &mut self,
59        edge_id: EdgeId,
60        prop_id: crate::types::PropertyId,
61        value: Option<Value>,
62        ts: Timestamp,
63    ) -> StorageResult<()> {
64        self.reject_inline()?;
65        let pos = self
66            .mapped_row(edge_id)
67            .ok_or_else(|| StorageError::invalid_offset(0))?;
68        if !linkrs_core::types::is_allocatable_timestamp(ts) {
69            return Err(StorageError::invalid_input(format!(
70                "property update carries reserved timestamp {}",
71                ts
72            )));
73        }
74        let idx = self
75            .prop_id_index
76            .get(&(prop_id.0 as i32))
77            .copied()
78            .ok_or_else(|| StorageError::column_not_found(format!("prop_id={}", prop_id.0)))?;
79        if pos >= self.visibility.len() || self.visibility[pos].is_virgin() {
80            return Err(StorageError::invalid_offset(pos as u32));
81        }
82        let col = &mut self.property_columns[idx];
83        col.set_versioned(pos, value.as_ref(), ts)?;
84        self.mark_column_dirty_at(idx);
85        Ok(())
86    }
87
88    pub fn set_property_at_row(
89        &mut self,
90        row_idx: usize,
91        name: &str,
92        value: Option<Value>,
93        ts: Timestamp,
94    ) -> StorageResult<()> {
95        if !linkrs_core::types::is_allocatable_timestamp(ts) {
96            return Err(StorageError::invalid_input(format!(
97                "property update carries reserved timestamp {}",
98                ts
99            )));
100        }
101        if row_idx >= self.visibility.len() || self.visibility[row_idx].is_virgin() {
102            return Err(StorageError::invalid_offset(row_idx as u32));
103        }
104        let col_idx = self
105            .column_index
106            .get(name)
107            .copied()
108            .ok_or_else(|| StorageError::column_not_found(name.to_string()))?;
109        let col = &mut self.property_columns[col_idx];
110        col.set_versioned(row_idx, value.as_ref(), ts)?;
111        self.mark_column_dirty_at(col_idx);
112        Ok(())
113    }
114
115    /// Mark one column position dirty without any name lookup or allocation.
116    /// Write paths resolve the position once through the schema indexes and
117    /// mark through this.
118    pub(crate) fn mark_column_dirty_at(&mut self, idx: usize) {
119        self.dirty_columns.insert(idx);
120    }
121}
linkrs_storage/edge/csr_with_properties/write.rs

linkrs_storage/edge/csr_with_properties/
write.rs
