linkrs-storage 0.1.0

Storage engine with CSR and memory-mapped containers
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

use linkrs_core::types::{LabelId, Timestamp, VertexId};
use linkrs_core::{StorageError, StorageResult, Value};
use std::sync::atomic::Ordering;
use std::sync::Arc;

use crate::vertex::WriteScope;

use super::super::GraphStorageContext;

impl GraphStorageContext {
    /// Scoped vertex insert staging the caller-owned row.
    ///
    /// Buffers the validated row in the scope without touching global
    /// state; the caller applies it through [`Self::commit_write_scope`].
    /// The scope travels by mutable borrow and is destroyed by the commit /
    /// rollback hooks or by crash drop.
    pub fn insert_vertex_with_scope(
        &self,
        label: LabelId,
        external_id: &str,
        properties: &[(std::sync::Arc<str>, Value)],
        ts: Timestamp,
        scope: &mut WriteScope,
    ) -> StorageResult<()> {
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return Err(StorageError::storage_not_open());
        }
        self.persistent
            .data_store
            .with_vertex_tables(|vertex_tables| {
                let table = vertex_tables.get(&label).ok_or_else(|| {
                    StorageError::label_not_found(format!("vertex label {}", label))
                })?;
                table.insert_with_scope(external_id, properties, ts, scope)
            })?;
        Ok(())
    }

    /// Integer-keyed scoped insert staging. Same contract as
    /// [`Self::insert_vertex_with_scope`].
    pub fn insert_vertex_by_i64_with_scope(
        &self,
        label: LabelId,
        external_id: i64,
        properties: &[(Arc<str>, Value)],
        ts: Timestamp,
        scope: &mut WriteScope,
    ) -> StorageResult<()> {
        VertexId::try_from_int64(external_id)?;
        if !self.persistent.is_open.load(Ordering::Acquire) {
            return Err(StorageError::storage_not_open());
        }
        self.persistent
            .data_store
            .with_vertex_tables(|vertex_tables| {
                let table = vertex_tables.get(&label).ok_or_else(|| {
                    StorageError::label_not_found(format!("vertex label {}", label))
                })?;
                table.insert_by_i64_with_scope(external_id, properties, ts, scope)
            })?;
        Ok(())
    }

    /// Commit hook for one label's staged rows: applies them to the table,
    /// then drops the label's staging records. Called before the timestamp
    /// commit; WAL durability stays the commit point. Returns the staged
    /// key to allocated global id mapping for the label.
    pub fn commit_write_scope(
        &self,
        label: LabelId,
        scope: &mut WriteScope,
        ts: Timestamp,
    ) -> StorageResult<Vec<(crate::vertex::IdKey, u32)>> {
        Ok(self.commit_write_scope_tracked(label, scope, ts)?.mapping)
    }

    /// Tracked commit hook reporting every applied mutation for later
    /// compensation of post-apply durability failures.
    pub fn commit_write_scope_tracked(
        &self,
        label: LabelId,
        scope: &mut WriteScope,
        ts: Timestamp,
    ) -> StorageResult<crate::vertex::vertex_table::sharded::CommitApplied> {
        self.persistent.data_store.with_vertex_tables(|tables| {
            tables
                .get(&label)
                .ok_or_else(|| StorageError::label_not_found(format!("vertex label {}", label)))
                .and_then(|table| table.commit_write_scope_tracked(scope, ts))
        })
    }

    /// Rollback hook for one label's staged rows. Discards the scope records
    /// without touching global state. Called before the timestamp abort and
    /// from the undo failure branches.
    pub fn rollback_write_scope(&self, label: LabelId, scope: &mut WriteScope, ts: Timestamp) {
        self.persistent.data_store.with_vertex_tables(|tables| {
            if let Some(table) = tables.get(&label) {
                table.rollback_write_scope(scope, ts);
            }
        });
    }

    /// Undo rows installed by a previous [`Self::commit_write_scope`] apply.
    ///
    /// Write entries that must fail the whole request after the table apply
    /// succeeded (secondary index maintenance, mutation recording) call this
    /// so a failed request leaves no applied row behind.
    pub(crate) fn undo_applied_scope_inserts(&self, label: LabelId, global_ids: &[u32]) {
        self.persistent.data_store.with_vertex_tables(|tables| {
            if let Some(table) = tables.get(&label) {
                table.undo_applied_ids(global_ids);
            }
        });
    }
}