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
117
118
119
120
121
122

use crate::engine::graph_storage::GraphStorageContext;
use crate::index::{EdgeIndexOps, VertexIndexOps};
use linkrs_core::metadata::IndexMetadataManager;
use linkrs_core::types::{Index, Timestamp};
use linkrs_core::StorageResult;
use linkrs_transaction::wal::{
    CreateEdgeIndexRedo, CreateTagIndexRedo, DropEdgeIndexRedo, DropTagIndexRedo,
};

pub(crate) fn replay_create_tag_index(
    ctx: &GraphStorageContext,
    redo: &CreateTagIndexRedo,
    _ts: Timestamp,
) -> StorageResult<()> {
    let index = Index::new(linkrs_core::types::IndexConfig {
        id: 0,
        name: redo.index_name.clone(),
        space_id: redo.space_id,
        schema_name: String::new(),
        fields: redo
            .fields
            .iter()
            .map(|(name, _typ)| {
                linkrs_core::types::IndexField::new(
                    name.clone(),
                    linkrs_core::Value::string(""),
                    true,
                )
            })
            .collect(),
        properties: redo.properties.clone(),
        index_type: linkrs_core::types::IndexType::TagIndex,
        is_unique: redo.is_unique,
        covering: false,
        partial_condition: None,
    });
    match ctx
        .index_metadata_manager()
        .create_tag_index(redo.space_id, &index)
    {
        Ok(_) => {}
        Err(e) => log::warn!("create_tag_index replay: {}", e),
    }
    let stored = ctx
        .index_metadata_manager()
        .get_tag_index(redo.space_id, &redo.index_name)?
        .unwrap_or(index);
    let data_mgr = ctx.index_data_manager().write();
    let _ = data_mgr.register_native_index(redo.space_id, &stored);
    Ok(())
}

pub(crate) fn replay_drop_tag_index(
    ctx: &GraphStorageContext,
    redo: &DropTagIndexRedo,
    _ts: Timestamp,
) -> StorageResult<()> {
    let _ = ctx
        .index_metadata_manager()
        .drop_tag_index(redo.space_id, &redo.index_name);
    let data_mgr = ctx.index_data_manager().write();
    let _ = data_mgr.clear_tag_index(redo.space_id, &redo.index_name);
    data_mgr.unregister_native_index(redo.space_id, &redo.index_name);
    Ok(())
}

pub(crate) fn replay_create_edge_index(
    ctx: &GraphStorageContext,
    redo: &CreateEdgeIndexRedo,
    _ts: Timestamp,
) -> StorageResult<()> {
    let index = Index::new(linkrs_core::types::IndexConfig {
        id: 0,
        name: redo.index_name.clone(),
        space_id: redo.space_id,
        schema_name: String::new(),
        fields: redo
            .fields
            .iter()
            .map(|(name, _typ)| {
                linkrs_core::types::IndexField::new(
                    name.clone(),
                    linkrs_core::Value::string(""),
                    true,
                )
            })
            .collect(),
        properties: redo.properties.clone(),
        index_type: linkrs_core::types::IndexType::EdgeIndex,
        is_unique: redo.is_unique,
        covering: false,
        partial_condition: None,
    });
    match ctx
        .index_metadata_manager()
        .create_edge_index(redo.space_id, &index)
    {
        Ok(_) => {}
        Err(e) => log::warn!("create_edge_index replay: {}", e),
    }
    let stored = ctx
        .index_metadata_manager()
        .get_edge_index(redo.space_id, &redo.index_name)?
        .unwrap_or(index);
    let data_mgr = ctx.index_data_manager().write();
    let _ = data_mgr.register_native_index(redo.space_id, &stored);
    Ok(())
}

pub(crate) fn replay_drop_edge_index(
    ctx: &GraphStorageContext,
    redo: &DropEdgeIndexRedo,
    _ts: Timestamp,
) -> StorageResult<()> {
    let _ = ctx
        .index_metadata_manager()
        .drop_edge_index(redo.space_id, &redo.index_name);
    let data_mgr = ctx.index_data_manager().write();
    let _ = data_mgr.clear_edge_index(redo.space_id, &redo.index_name);
    data_mgr.unregister_native_index(redo.space_id, &redo.index_name);
    Ok(())
}