linkrs_storage/edge/edge_table/
freeze.rs

1//! Explicit freeze and unfreeze of edge-table groups per direction.
2//!
3//! Freezing packs one direction's group into the frozen form through the
4//! shard-set freeze, promoting reclaimed deletions into the visibility
5//! authority exactly like the group compaction path. Unfreezing rebuilds the
6//! mutable variant with the same logical content and no authority traffic.
7
8use super::core::EdgeStore;
9use crate::edge::FreezeFeasibility;
10use linkrs_core::types::{EdgeId, Timestamp};
11use linkrs_core::StorageResult;
12
13impl EdgeStore {
14    /// Read-only freeze outcome for one group of one direction.
15    ///
16    /// `outgoing` selects the out or in shard set. Runs the same gate
17    /// `freeze_group` enforces without touching state.
18    pub fn freeze_feasibility(&self, outgoing: bool, gid: usize) -> FreezeFeasibility {
19        if outgoing {
20            self.out_csr.freeze_feasibility(gid)
21        } else {
22            self.in_csr.freeze_feasibility(gid)
23        }
24    }
25
26    /// Freeze one group of one direction, returning its packed live count.
27    ///
28    /// `outgoing` selects the out or in shard set. Reclaimable tombstones
29    /// below `cutoff` are dropped first and reported into the visibility
30    /// authority; a maximum cutoff packs verbatim. The frozen group rejects
31    /// writes until explicitly unfrozen.
32    pub fn freeze_group(
33        &mut self,
34        outgoing: bool,
35        gid: usize,
36        cutoff: Timestamp,
37        reserve_ratio: f32,
38    ) -> StorageResult<u64> {
39        let packed = if outgoing {
40            let shards = &mut self.out_csr;
41            let mvcc = &mut self.mvcc;
42            shards.freeze_group(
43                gid,
44                cutoff,
45                reserve_ratio,
46                &mut |edge_id: EdgeId, delete_ts: Timestamp| {
47                    mvcc.record_deletion(edge_id, delete_ts);
48                },
49            )?
50        } else {
51            let shards = &mut self.in_csr;
52            let mvcc = &mut self.mvcc;
53            shards.freeze_group(
54                gid,
55                cutoff,
56                reserve_ratio,
57                &mut |edge_id: EdgeId, delete_ts: Timestamp| {
58                    mvcc.record_deletion(edge_id, delete_ts);
59                },
60            )?
61        };
62        let drift = self.audit_copy_drift();
63        if !drift.is_empty() {
64            return Err(linkrs_core::StorageError::data_corruption(format!(
65                "freeze_group drift after pack: {}",
66                drift.join("; ")
67            )));
68        }
69        Ok(packed)
70    }
71
72    /// Unfreeze one group of one direction, returning its restored live count.
73    pub fn unfreeze_group(&mut self, outgoing: bool, gid: usize) -> StorageResult<u64> {
74        if outgoing {
75            self.out_csr.unfreeze_group(gid)
76        } else {
77            self.in_csr.unfreeze_group(gid)
78        }
79    }
80}
81
82#[cfg(test)]
83mod tests {
84    use super::*;
85    use crate::edge::{EdgeRecord, EdgeSchema, EdgeStrategy, RecordForm};
86    use crate::types::StoragePropertyDef;
87    use linkrs_core::types::{DataType, Timestamp, VertexId};
88    use linkrs_core::Value;
89
90    use super::super::config::EdgeTableConfig;
91
92    fn frozen_test_schema() -> EdgeSchema {
93        EdgeSchema {
94            label_id: 0,
95            label_name: "knows".into(),
96            src_label: 0,
97            dst_label: 0,
98            properties: vec![StoragePropertyDef {
99                name: "weight".into(),
100                data_type: DataType::Double,
101                nullable: false,
102                default_value: Some(Value::Double(0.0)),
103            }],
104            oe_strategy: EdgeStrategy::Multiple,
105            ie_strategy: EdgeStrategy::Multiple,
106            schema_version: 1,
107            record_form: RecordForm::default(),
108        }
109    }
110
111    fn sample_table() -> EdgeStore {
112        let mut table =
113            EdgeStore::with_config(frozen_test_schema(), EdgeTableConfig::default()).unwrap();
114        for src in 0..8u32 {
115            for k in 0..2u32 {
116                table
117                    .insert_edge(
118                        src,
119                        src + k + 1,
120                        0,
121                        &[("weight".into(), Value::Double(1.0))],
122                        100,
123                    )
124                    .unwrap();
125            }
126        }
127        assert!(table.delete_edge(0, 1, 0, 200).unwrap());
128        table
129    }
130
131    fn snapshot(table: &EdgeStore, ts: Timestamp) -> Vec<Vec<VertexId>> {
132        (0..8u32)
133            .map(|src| {
134                table
135                    .out_edges(src, ts)
136                    .into_iter()
137                    .map(|edge| edge.dst_vid)
138                    .collect()
139            })
140            .collect()
141    }
142
143    #[test]
144    fn freeze_keeps_reads_and_rejects_writes() {
145        let mut table = sample_table();
146        let before_freeze = snapshot(&table, 300);
147        let before_old = snapshot(&table, 150);
148
149        let out_packed = table.freeze_group(true, 0, Timestamp::MAX, 0.0).unwrap();
150        let in_packed = table.freeze_group(false, 0, Timestamp::MAX, 0.0).unwrap();
151        assert_eq!(out_packed, 15);
152        assert_eq!(in_packed, 15);
153
154        assert_eq!(snapshot(&table, 300), before_freeze);
155        assert_eq!(snapshot(&table, 150), before_old);
156        assert!(table.has_edge(0, 2, 0, 300));
157        assert!(!table.has_edge(0, 1, 0, 300));
158        assert!(table.has_edge(0, 1, 0, 150));
159
160        assert!(table.insert_edge(1, 9, 0, &[], 400).is_err());
161        assert!(table.delete_edge(1, 2, 0, 400).is_err());
162
163        assert_eq!(table.unfreeze_group(true, 0).unwrap(), 15);
164        assert_eq!(table.unfreeze_group(false, 0).unwrap(), 15);
165        assert_eq!(snapshot(&table, 300), before_freeze);
166        table.insert_edge(1, 9, 0, &[], 400).unwrap();
167        assert!(table.has_edge(1, 9, 0, 400));
168    }
169
170    #[test]
171    fn freeze_guards_bad_states() {
172        let mut table = sample_table();
173        assert!(table.freeze_group(true, 41, Timestamp::MAX, 0.0).is_err());
174        assert!(table.unfreeze_group(true, 0).is_err());
175        table.freeze_group(true, 0, Timestamp::MAX, 0.0).unwrap();
176        assert!(table.freeze_group(true, 0, Timestamp::MAX, 0.0).is_err());
177    }
178
179    #[test]
180    fn valued_bundled_freeze_preserves_values() {
181        use crate::edge::RecordFormPreference;
182        let config = EdgeTableConfig {
183            record_form: RecordFormPreference::Bundled,
184            ..Default::default()
185        };
186        let mut bundled =
187            EdgeStore::with_config(frozen_test_schema(), config).expect("bundled table builds");
188        assert_eq!(bundled.schema().record_form, RecordForm::Bundled);
189        bundled
190            .insert_edge(0, 1, 0, &[("weight".into(), Value::Double(1.0))], 100)
191            .expect("valued insert");
192        assert!(bundled.freeze_feasibility(true, 0).is_ready());
193        assert!(bundled.freeze_feasibility(false, 0).is_ready());
194
195        let before: Vec<EdgeRecord> =
196            crate::edge::edge_table::iterator::EdgeTableScanIterator::new(&bundled, 200).collect();
197        assert_eq!(before.len(), 1);
198        bundled.freeze_group(true, 0, Timestamp::MAX, 0.0).unwrap();
199        bundled.freeze_group(false, 0, Timestamp::MAX, 0.0).unwrap();
200        assert!(bundled.out_csr.is_frozen(0));
201        assert!(bundled.in_csr.is_frozen(0));
202
203        // Frozen reads serve the carried inline value on both legs.
204        let frozen: Vec<EdgeRecord> =
205            crate::edge::edge_table::iterator::EdgeTableScanIterator::new(&bundled, 200).collect();
206        assert_eq!(frozen.len(), 1);
207        assert_eq!(frozen[0].properties, before[0].properties);
208        assert_eq!(
209            bundled
210                .get_edge(0, 1, 0, 200)
211                .expect("frozen edge readable")
212                .properties,
213            vec![("weight".into(), Value::Double(1.0))]
214        );
215
216        // Writes stay rejected while frozen; unfreezing restores the live
217        // bundled group with values intact and writes working again.
218        assert!(bundled.insert_edge(1, 9, 0, &[], 400).is_err());
219        bundled.unfreeze_group(true, 0).unwrap();
220        bundled.unfreeze_group(false, 0).unwrap();
221        let after: Vec<EdgeRecord> =
222            crate::edge::edge_table::iterator::EdgeTableScanIterator::new(&bundled, 200).collect();
223        assert_eq!(after.len(), 1);
224        assert_eq!(after[0].properties, before[0].properties);
225        bundled
226            .insert_edge(1, 9, 0, &[("weight".into(), Value::Double(2.0))], 400)
227            .expect("writes resume after unfreeze");
228        assert_eq!(
229            bundled
230                .get_edge(1, 9, 0, 500)
231                .expect("new edge readable")
232                .properties,
233            vec![("weight".into(), Value::Double(2.0))]
234        );
235    }
236
237    #[test]
238    fn row_ordering_contract_mutable_unordered_frozen_sorted() {
239        use crate::edge::{EdgeSchema, EdgeStrategy, RecordForm};
240        use crate::types::StoragePropertyDef;
241        use linkrs_core::types::DataType;
242
243        let schema = EdgeSchema {
244            label_id: 0,
245            label_name: "ordered".into(),
246            src_label: 0,
247            dst_label: 0,
248            properties: vec![StoragePropertyDef {
249                name: "weight".into(),
250                data_type: DataType::Double,
251                nullable: false,
252                default_value: Some(Value::Double(0.0)),
253            }],
254            oe_strategy: EdgeStrategy::Multiple,
255            ie_strategy: EdgeStrategy::Multiple,
256            schema_version: 1,
257            record_form: RecordForm::default(),
258        };
259        let mut table = EdgeStore::with_config(schema, EdgeTableConfig::default()).unwrap();
260        for dst in [3u32, 1, 2] {
261            table
262                .insert_edge(0, dst, 0, &[], 100)
263                .expect("insert out of key order");
264        }
265        // Mutable rows promise no order: insertion order is observed but the
266        // query layer must not depend on it.
267        let mutable_order: Vec<i64> = table
268            .out_edges(0, 200)
269            .iter()
270            .map(|e| e.dst_vid.as_int64().unwrap_or(-1))
271            .collect();
272        assert_eq!(mutable_order, vec![3, 1, 2]);
273
274        // Frozen rows promise (endpoint, rank, edge_id) order.
275        table.freeze_group(true, 0, Timestamp::MAX, 0.0).unwrap();
276        let frozen_order: Vec<i64> = table
277            .out_edges(0, 200)
278            .iter()
279            .map(|e| e.dst_vid.as_int64().unwrap_or(-1))
280            .collect();
281        assert_eq!(frozen_order, vec![1, 2, 3]);
282
283        // Unfrozen rows promise no order again: only the content set is
284        // pinned, never the sequence.
285        table.unfreeze_group(true, 0).unwrap();
286        let mut restored: Vec<i64> = table
287            .out_edges(0, 200)
288            .iter()
289            .map(|e| e.dst_vid.as_int64().unwrap_or(-1))
290            .collect();
291        restored.sort_unstable();
292        assert_eq!(restored, vec![1, 2, 3]);
293    }
294}
linkrs_storage/edge/edge_table/freeze.rs

linkrs_storage/edge/edge_table/
freeze.rs
