rocksgraph 0.1.0

A Gremlin-inspired property graph query engine written in Rust, backed by RocksDB
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
123
124

// Physical steps: simplePath(), cyclicPath()

use crate::engine::volcano::steps::traits::ExplainNode;
use crate::types::PIPELINE_PRODUCE_SIZE;
use crate::{
    engine::{
        context::GraphCtx,
        traverser::Traverser,
        volcano::steps::traits::{CoreStep, StepRef},
    },
    types::{error::StoreError, gvalue::GValue},
};
use smallvec::SmallVec;
use std::rc::Rc;

/// Filters out traversers whose parent chain contains duplicates — keeps only simple paths.
#[derive(Debug, Default)]
pub struct SimplePathStep {
    upstream: Option<StepRef>,
}

impl CoreStep for SimplePathStep {
    fn add_upper(&mut self, upstream: StepRef) {
        self.upstream = Some(upstream);
    }
    fn reset(&mut self) {
        if let Some(u) = &self.upstream {
            u.reset();
        }
    }
    fn upper(&self) -> Option<StepRef> {
        self.upstream.clone()
    }

    fn produce(
        &mut self,
        ctx: &mut dyn GraphCtx,
    ) -> Result<Option<SmallVec<[Rc<Traverser>; PIPELINE_PRODUCE_SIZE]>>, StoreError> {
        let Some(upstream) = self.upstream.as_ref() else { return Ok(None) };
        let mut batch = SmallVec::with_capacity(PIPELINE_PRODUCE_SIZE);
        while batch.len() < PIPELINE_PRODUCE_SIZE {
            let Some(t) = upstream.next(ctx)? else { break };
            if has_duplicate_vertex(&t) {
                continue;
            }
            batch.push(t);
        }
        if batch.is_empty() {
            Ok(None)
        } else {
            Ok(Some(batch))
        }
    }

    fn explain(&self) -> ExplainNode {
        ExplainNode::new("SimplePathStep")
    }
}

/// Filters out traversers without duplicate vertices — keeps only cyclic paths.
#[derive(Debug, Default)]
pub struct CyclicPathStep {
    upstream: Option<StepRef>,
}

impl CoreStep for CyclicPathStep {
    fn add_upper(&mut self, upstream: StepRef) {
        self.upstream = Some(upstream);
    }
    fn reset(&mut self) {
        if let Some(u) = &self.upstream {
            u.reset();
        }
    }
    fn upper(&self) -> Option<StepRef> {
        self.upstream.clone()
    }

    fn produce(
        &mut self,
        ctx: &mut dyn GraphCtx,
    ) -> Result<Option<SmallVec<[Rc<Traverser>; PIPELINE_PRODUCE_SIZE]>>, StoreError> {
        let Some(upstream) = self.upstream.as_ref() else { return Ok(None) };
        let mut batch = SmallVec::with_capacity(PIPELINE_PRODUCE_SIZE);
        while batch.len() < PIPELINE_PRODUCE_SIZE {
            let Some(t) = upstream.next(ctx)? else { break };
            if !has_duplicate_vertex(&t) {
                continue;
            }
            batch.push(t);
        }
        if batch.is_empty() {
            Ok(None)
        } else {
            Ok(Some(batch))
        }
    }

    fn explain(&self) -> ExplainNode {
        ExplainNode::new("CyclicPathStep")
    }
}

/// Check whether the traverser's parent chain contains a duplicate vertex id.
fn has_duplicate_vertex(t: &Rc<Traverser>) -> bool {
    let current_id = match &t.value {
        GValue::Vertex(vk) => Some(*vk),
        _ => None,
    };
    let mut seen = std::collections::HashSet::new();
    if let Some(id) = current_id {
        seen.insert(id);
    }
    let mut cur = t.parent.as_deref();
    while let Some(ancestor) = cur {
        if let GValue::Vertex(vk) = &ancestor.value {
            if !seen.insert(*vk) {
                return true;
            }
        }
        cur = ancestor.parent.as_deref();
    }
    false
}