linkrs_query/executor/streaming/helpers/
conversion.rs

1//! Data type conversions for streaming executor
2//!
3//! Converts between Vertex/Edge and row representations.
4
5use linkrs_core::Value;
6use linkrs_core::{Edge, Vertex};
7
8/// Render a vertex id as its raw external string without display quoting.
9fn vid_raw_string(vid: &Vertex) -> String {
10    if let Some(s) = vid.vid.as_str() {
11        s.to_string()
12    } else if let Some(i) = vid.vid.as_int64() {
13        i.to_string()
14    } else if let Some(u) = vid.vid.as_u64() {
15        u.to_string()
16    } else {
17        format!("{:?}", vid.vid.as_bytes())
18    }
19}
20
21fn edge_raw_string(vid: &linkrs_core::types::storage_ids::VertexId) -> String {
22    if let Some(s) = vid.as_str() {
23        s.to_string()
24    } else if let Some(i) = vid.as_int64() {
25        i.to_string()
26    } else if let Some(u) = vid.as_u64() {
27        u.to_string()
28    } else {
29        format!("{:?}", vid.as_bytes())
30    }
31}
32
33/// Convert a Vertex to row representation
34pub fn vertex_to_row(vertex: &Vertex) -> Vec<Value> {
35    let mut row = vec![
36        Value::from(vertex.vid),
37        Value::string(vid_raw_string(vertex)),
38    ];
39
40    row.push(Value::string(vertex.tag.name.clone()));
41
42    // Add first 3 properties from the single-tag view so tag-only
43    // rows are not silently dropped.
44    for value in vertex.properties().values().take(3) {
45        row.push(value.clone());
46    }
47
48    row
49}
50
51/// Convert an Edge to row representation
52pub fn edge_to_row(edge: &Edge) -> Vec<Value> {
53    let mut row = vec![
54        Value::string(edge_raw_string(&edge.src)),
55        Value::string(edge_raw_string(&edge.dst)),
56        Value::string(edge.edge_type.clone()),
57        Value::BigInt(edge.ranking),
58    ];
59
60    // Add first 2 properties (simplified)
61    for value in edge.props.values().take(2) {
62        row.push(value.clone());
63    }
64
65    row
66}
67
68/// Convert Vertex collection to rows, filtering by partition range.
69/// Range uses `i64` to match the real vertex ID type and avoid silent
70/// truncation of values >= 2^32 or negative IDs.
71pub fn vertices_to_rows(
72    vertices: Vec<Vertex>,
73    partition_range: &std::ops::Range<i64>,
74) -> Vec<Vec<Value>> {
75    vertices
76        .into_iter()
77        .filter(|v| {
78            v.vid
79                .as_int64()
80                .is_some_and(|id| id >= partition_range.start && id < partition_range.end)
81        })
82        .map(|v| vertex_to_row(&v))
83        .collect()
84}
85
86/// Convert Edge collection to rows, filtering by partition range.
87/// Only edges whose source ID can be parsed as `i64` are matched against
88/// the range; non-numeric source IDs are excluded.
89pub fn edges_to_rows(edges: Vec<Edge>, partition_range: &std::ops::Range<i64>) -> Vec<Vec<Value>> {
90    edges
91        .into_iter()
92        .filter(|e| {
93            edge_raw_string(&e.src)
94                .parse::<i64>()
95                .is_ok_and(|id| id >= partition_range.start && id < partition_range.end)
96        })
97        .map(|e| edge_to_row(&e))
98        .collect()
99}
100
101#[cfg(test)]
102mod tests {
103    use super::*;
104    use linkrs_core::types::storage_ids::VertexId;
105    use linkrs_core::vertex_edge_path::Tag;
106    use std::collections::HashMap;
107
108    #[test]
109    fn test_vertex_conversion() {
110        let vertex = Vertex::new(
111            VertexId::try_from_string("vertex_123").expect("valid vertex id"),
112            Tag::new("person".into(), HashMap::new()),
113        );
114
115        let row = vertex_to_row(&vertex);
116
117        // Verify row structure: vid, vid string and tag form the base layout.
118        assert_eq!(row.len(), 3);
119        assert_eq!(row[2], Value::string("person"));
120    }
121
122    #[test]
123    fn test_edge_conversion() {
124        let edge = Edge {
125            src: VertexId::try_from_string("src_1").expect("valid vertex id"),
126            dst: VertexId::try_from_string("dst_2").expect("valid vertex id"),
127            edge_type: "follows".to_string(),
128            ranking: 42,
129            props: std::collections::HashMap::new(),
130        };
131
132        let row = edge_to_row(&edge);
133
134        // Verify row structure: src, dst, edge_type, ranking.
135        assert_eq!(row.len(), 4);
136        assert_eq!(row[2], Value::string("follows"));
137        assert_eq!(row[3], Value::BigInt(42));
138    }
139
140    #[test]
141    fn test_partition_filtering() {
142        let vertices = vec![
143            Vertex::new(
144                VertexId::try_from_int64(10).expect("valid vertex id"),
145                Tag::new(String::new(), HashMap::new()),
146            ),
147            Vertex::new(
148                VertexId::try_from_int64(20).expect("valid vertex id"),
149                Tag::new(String::new(), HashMap::new()),
150            ),
151            Vertex::new(
152                VertexId::try_from_int64(30).expect("valid vertex id"),
153                Tag::new(String::new(), HashMap::new()),
154            ),
155        ];
156
157        // Filter for partition range [15, 35)
158        let partition_range = std::ops::Range {
159            start: 15i64,
160            end: 35,
161        };
162        let rows = vertices_to_rows(vertices, &partition_range);
163
164        // Should include vertices with id 20 and 30, but not 10
165        assert_eq!(rows.len(), 2);
166    }
167}
linkrs_query/executor/streaming/helpers/conversion.rs

linkrs_query/executor/streaming/helpers/
conversion.rs
