linkrs-storage 0.1.0

pub mod data;
pub mod deferred;
pub mod index;
pub mod schema;

use crate::engine::graph_storage::GraphStorageContext;
use linkrs_core::types::{LabelId, Timestamp, VertexId};
use linkrs_core::wal::traits::RecoveryApplier;
use linkrs_core::{StorageResult, Value};
use linkrs_transaction::wal::{
    AddEdgePropRedo, AddVertexPropRedo, ClearSpaceRedo, CreateEdgeIndexRedo, CreateEdgeTypeRedo,
    CreateSpaceRedo, CreateTagIndexRedo, CreateVertexTypeRedo, DeleteEdgePropRedo, DeleteEdgeRedo,
    DeleteEdgeTypeRedo, DeleteVertexPropRedo, DeleteVertexPropsRedo, DeleteVertexTypeRedo,
    DropEdgeIndexRedo, DropSpaceRedo, DropTagIndexRedo, InsertEdgeRedo, RenameEdgePropRedo,
    RenameEdgeTypeRedo, RenameTagRedo, RenameVertexPropRedo, UpdateEdgePropRedo,
    UpdateSequenceRedo,
};
use std::sync::Arc;

impl RecoveryApplier for GraphStorageContext {
    fn replay_insert_vertex(
        &self,
        label: LabelId,
        vid: VertexId,
        properties: &[(Arc<str>, Value)],
        ts: Timestamp,
    ) -> StorageResult<()> {
        data::replay_insert_vertex(self, label, vid, properties, ts)
    }

    fn replay_insert_edge(&self, redo: &InsertEdgeRedo, ts: Timestamp) -> StorageResult<()> {
        data::replay_insert_edge(self, redo, ts)
    }

    fn replay_delete_edge(&self, redo: &DeleteEdgeRedo, ts: Timestamp) -> StorageResult<()> {
        data::replay_delete_edge(self, redo, ts)
    }

    fn replay_delete_vertex_props(
        &self,
        redo: &DeleteVertexPropsRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        data::replay_delete_vertex_props(self, redo, ts)
    }

    fn replay_update_vertex_prop(
        &self,
        label: LabelId,
        vid: VertexId,
        prop_name: &str,
        value: &Value,
        ts: Timestamp,
    ) -> StorageResult<()> {
        data::replay_update_vertex_prop(self, label, vid, prop_name, value, ts)
    }

    fn replay_update_edge_prop(
        &self,
        redo: &UpdateEdgePropRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        data::replay_update_edge_prop(self, redo, ts)
    }

    fn replay_delete_vertex(
        &self,
        label: LabelId,
        vid: VertexId,
        ts: Timestamp,
    ) -> StorageResult<()> {
        data::replay_delete_vertex(self, label, vid, ts)
    }

    fn replay_create_space(&self, redo: &CreateSpaceRedo, ts: Timestamp) -> StorageResult<()> {
        schema::replay_create_space(self, redo, ts)
    }

    fn replay_drop_space(&self, redo: &DropSpaceRedo, ts: Timestamp) -> StorageResult<()> {
        schema::replay_drop_space(self, redo, ts)
    }

    fn replay_clear_space(&self, redo: &ClearSpaceRedo, ts: Timestamp) -> StorageResult<()> {
        schema::replay_clear_space(self, redo, ts)
    }

    fn replay_create_vertex_type(
        &self,
        redo: &CreateVertexTypeRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        schema::replay_create_vertex_type(self, redo, ts)
    }

    fn replay_create_edge_type(
        &self,
        redo: &CreateEdgeTypeRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        schema::replay_create_edge_type(self, redo, ts)
    }

    fn replay_delete_vertex_type(
        &self,
        redo: &DeleteVertexTypeRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        schema::replay_delete_vertex_type(self, redo, ts)
    }

    fn replay_delete_edge_type(
        &self,
        redo: &DeleteEdgeTypeRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        schema::replay_delete_edge_type(self, redo, ts)
    }

    fn replay_add_vertex_prop(&self, redo: &AddVertexPropRedo, ts: Timestamp) -> StorageResult<()> {
        schema::replay_add_vertex_prop(self, redo, ts)
    }

    fn replay_add_edge_prop(&self, redo: &AddEdgePropRedo, ts: Timestamp) -> StorageResult<()> {
        schema::replay_add_edge_prop(self, redo, ts)
    }

    fn replay_delete_vertex_prop(
        &self,
        redo: &DeleteVertexPropRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        schema::replay_delete_vertex_prop(self, redo, ts)
    }

    fn replay_delete_edge_prop(
        &self,
        redo: &DeleteEdgePropRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        schema::replay_delete_edge_prop(self, redo, ts)
    }

    fn replay_rename_vertex_prop(
        &self,
        redo: &RenameVertexPropRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        schema::replay_rename_vertex_prop(self, redo, ts)
    }

    fn replay_rename_edge_prop(
        &self,
        redo: &RenameEdgePropRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        schema::replay_rename_edge_prop(self, redo, ts)
    }

    fn replay_rename_tag(&self, redo: &RenameTagRedo, _ts: Timestamp) -> StorageResult<()> {
        schema::replay_rename_tag(self, redo)
    }

    fn replay_rename_edge_type(
        &self,
        redo: &RenameEdgeTypeRedo,
        _ts: Timestamp,
    ) -> StorageResult<()> {
        schema::replay_rename_edge_type(self, redo)
    }

    fn replay_create_tag_index(
        &self,
        redo: &CreateTagIndexRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        index::replay_create_tag_index(self, redo, ts)
    }

    fn replay_drop_tag_index(&self, redo: &DropTagIndexRedo, ts: Timestamp) -> StorageResult<()> {
        index::replay_drop_tag_index(self, redo, ts)
    }

    fn replay_create_edge_index(
        &self,
        redo: &CreateEdgeIndexRedo,
        ts: Timestamp,
    ) -> StorageResult<()> {
        index::replay_create_edge_index(self, redo, ts)
    }

    fn replay_drop_edge_index(&self, redo: &DropEdgeIndexRedo, ts: Timestamp) -> StorageResult<()> {
        index::replay_drop_edge_index(self, redo, ts)
    }

    fn replay_update_sequence(
        &self,
        redo: &UpdateSequenceRedo,
        _ts: Timestamp,
    ) -> StorageResult<()> {
        self.serial_allocator().seed(
            &crate::engine::graph_storage::SerialKey::new(redo.space_id, &redo.table_name),
            redo.next_value,
        );
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use crate::engine::graph_storage::GraphStorageContext;
    use crate::engine::{EdgeOperationParams, InsertEdgeParams};
    use linkrs_core::types::VertexId;
    use linkrs_core::wal::traits::RecoveryApplier;
    use linkrs_core::Value;
    use linkrs_transaction::wal::{
        AddEdgePropRedo, AddVertexPropRedo, CreateEdgeTypeRedo, CreateVertexTypeRedo,
        DeleteEdgePropRedo, DeleteVertexPropRedo, RenameEdgePropRedo, RenameVertexPropRedo,
    };

    #[test]
    fn test_schema_replay_roundtrip() {
        let ctx = GraphStorageContext::new();

        ctx.replay_create_vertex_type(
            &CreateVertexTypeRedo {
                space_name: "test_space".into(),
                label_id: Some(1),
                label_name: "Person".into(),
                schema: vec![
                    ("id".into(), "BIGINT".into(), true),
                    ("name".into(), "STRING".into(), false),
                ],
            },
            1,
        )
        .expect("Vertex type replay should succeed");

        ctx.replay_create_vertex_type(
            &CreateVertexTypeRedo {
                space_name: "test_space".into(),
                label_id: Some(2),
                label_name: "City".into(),
                schema: vec![
                    ("id".into(), "BIGINT".into(), false),
                    ("name".into(), "STRING".into(), false),
                ],
            },
            1,
        )
        .expect("Second vertex type replay should succeed");

        let person_label = ctx
            .data_store()
            .vertex_label_id("space_1:tag:Person")
            .expect("Person label should exist");
        let city_label = ctx
            .data_store()
            .vertex_label_id("space_1:tag:City")
            .expect("City label should exist");

        ctx.replay_add_vertex_prop(
            &AddVertexPropRedo {
                label: person_label,
                properties: vec![("age".into(), "INT".into(), false)],
                new_comment: None,
            },
            2,
        )
        .expect("Vertex property replay should succeed");

        ctx.replay_rename_vertex_prop(
            &RenameVertexPropRedo {
                label: person_label,
                old_name: "name".into(),
                new_name: "full_name".into(),
            },
            2,
        )
        .expect("Vertex rename replay should succeed");

        ctx.replay_delete_vertex_prop(
            &DeleteVertexPropRedo {
                label: person_label,
                prop_names: vec!["age".into()],
            },
            2,
        )
        .expect("Vertex delete replay should succeed");

        ctx.replay_create_edge_type(
            &CreateEdgeTypeRedo {
                space_name: "test_space".into(),
                label_id: Some(3),
                src_label: "Person".into(),
                dst_label: "City".into(),
                edge_label: "LIVES_IN".into(),
                schema: vec![("since".into(), "INT".into(), false)],
            },
            3,
        )
        .expect("Edge type replay should succeed");

        let lives_in_label = 3;

        ctx.replay_add_edge_prop(
            &AddEdgePropRedo {
                src_label: person_label,
                dst_label: city_label,
                edge_label: lives_in_label,
                properties: vec![("cost".into(), "INT".into(), false)],
                new_comment: None,
            },
            3,
        )
        .expect("Edge property replay should succeed");

        ctx.replay_rename_edge_prop(
            &RenameEdgePropRedo {
                src_label: person_label,
                dst_label: city_label,
                edge_label: lives_in_label,
                old_name: "since".into(),
                new_name: "started".into(),
            },
            3,
        )
        .expect("Edge rename replay should succeed");

        ctx.replay_delete_edge_prop(
            &DeleteEdgePropRedo {
                src_label: person_label,
                dst_label: city_label,
                edge_label: lives_in_label,
                prop_names: vec!["cost".into()],
            },
            3,
        )
        .expect("Edge delete replay should succeed");

        let person_tag = ctx
            .schema_manager()
            .find_tag_by_id(person_label)
            .expect("Person tag should exist")
            .1;
        assert_eq!(
            person_tag
                .properties
                .iter()
                .map(|prop| prop.name.as_str())
                .collect::<Vec<_>>(),
            vec!["id", "full_name"]
        );

        let lives_in_type = ctx
            .schema_manager()
            .find_edge_type_by_id(lives_in_label)
            .expect("Edge type should exist")
            .1;
        assert_eq!(
            lives_in_type
                .properties
                .iter()
                .map(|prop| prop.name.as_str())
                .collect::<Vec<_>>(),
            vec!["started"]
        );

        {
            let mut scope = crate::vertex::WriteScope::new(4);
            ctx.insert_vertex_by_i64_with_scope(
                person_label,
                1001,
                &[
                    ("id".into(), Value::BigInt(1001)),
                    ("full_name".into(), Value::string("Alice")),
                ],
                4,
                &mut scope,
            )
            .expect("Vertex stage should succeed after property replay");
            ctx.commit_write_scope(person_label, &mut scope, 4)
                .expect("Vertex apply should succeed after property replay");
        }

        {
            let mut scope = crate::vertex::WriteScope::new(4);
            ctx.insert_vertex_by_i64_with_scope(
                city_label,
                2001,
                &[
                    ("id".into(), Value::BigInt(2001)),
                    ("name".into(), Value::string("Shanghai")),
                ],
                4,
                &mut scope,
            )
            .expect("City stage should succeed");
            ctx.commit_write_scope(city_label, &mut scope, 4)
                .expect("City apply should succeed");
        }

        let vertex = ctx
            .get_vertex_by_i64(person_label, 1001, 5)
            .expect("Inserted vertex should be visible");
        assert_eq!(
            vertex
                .properties
                .iter()
                .find(|(name, _)| &**name == "full_name")
                .map(|(_, value)| value),
            Some(&Value::string("Alice"))
        );
        assert!(vertex.properties.iter().all(|(name, _)| &**name != "age"));

        ctx.insert_edge(InsertEdgeParams {
            edge_label: lives_in_label,
            src_label: person_label,
            src_id: VertexId::try_from_int64(1001).expect("test vertex id"),
            dst_label: city_label,
            dst_id: VertexId::try_from_int64(2001).expect("test vertex id"),
            rank: 0,
            properties: &[("started".into(), Value::Int(2012))],
            ts: 5,
        })
        .expect("Edge insert should succeed after property replay");

        let edge = ctx
            .get_edge(
                &EdgeOperationParams {
                    edge_label: lives_in_label,
                    src_label: person_label,
                    src_id: VertexId::try_from_int64(1001).expect("test vertex id"),
                    dst_label: city_label,
                    dst_id: VertexId::try_from_int64(2001).expect("test vertex id"),
                    rank: 0,
                },
                5,
            )
            .expect("Inserted edge should be visible");
        assert_eq!(
            edge.properties
                .iter()
                .find(|(name, _)| &**name == "started")
                .map(|(_, value)| value),
            Some(&Value::Int(2012))
        );
        assert!(edge.properties.iter().all(|(name, _)| &**name != "cost"));
        assert!(edge.properties.iter().all(|(name, _)| &**name != "since"));
    }
}