linkrs_query/executor/expression/functions/builtin/
graph.rs

1//! Implementation of graph-related functions
2//!
3//! Provide functions for manipulating vertices and edges, including id, tags, labels, properties, type, src, dst, and rank.
4//! Also includes graph traversal functions: neighbors, degree, shortest_path.
5
6use crate::executor::expression::evaluation_context::graph_storage::GraphStorageRef;
7use crate::executor::expression::ExpressionError;
8use linkrs_core::types::VertexId;
9use linkrs_core::value::list::List;
10use linkrs_core::value::NullType;
11use linkrs_core::Value;
12
13/// Graph function enumeration
14#[derive(Debug, Clone, Copy, PartialEq, Eq)]
15pub enum GraphFunction {
16    Id,
17    Tags,
18    Labels,
19    Properties,
20    EdgeType,
21    Src,
22    Dst,
23    Rank,
24    StartNode,
25    EndNode,
26    Neighbors,
27    Degree,
28    OutEdges,
29    InEdges,
30    ShortestPath,
31    Bfs,
32    ConnectedComponents,
33    VariableLengthPath,
34    PageRank,
35}
36
37impl GraphFunction {
38    /// Obtain the function name
39    pub fn name(&self) -> &str {
40        match self {
41            Self::Id => "id",
42            Self::Tags => "tags",
43            Self::Labels => "labels",
44            Self::Properties => "properties",
45            Self::EdgeType => "type",
46            Self::Src => "src",
47            Self::Dst => "dst",
48            Self::Rank => "rank",
49            Self::StartNode => "startnode",
50            Self::EndNode => "endnode",
51            Self::Neighbors => "neighbors",
52            Self::Degree => "degree",
53            Self::OutEdges => "out_edges",
54            Self::InEdges => "in_edges",
55            Self::ShortestPath => "shortest_path",
56            Self::Bfs => "bfs",
57            Self::ConnectedComponents => "connected_components",
58            Self::VariableLengthPath => "variable_length_path",
59            Self::PageRank => "pagerank",
60        }
61    }
62
63    /// Determine the number of parameters
64    pub fn arity(&self) -> usize {
65        match self {
66            Self::Id => 1,
67            Self::Tags => 1,
68            Self::Labels => 1,
69            Self::Properties => 1,
70            Self::EdgeType => 1,
71            Self::Src => 1,
72            Self::Dst => 1,
73            Self::Rank => 1,
74            Self::StartNode => 1,
75            Self::EndNode => 1,
76            Self::Neighbors => 1,
77            Self::Degree => 1,
78            Self::OutEdges => 1,
79            Self::InEdges => 1,
80            Self::ShortestPath => 2,
81            Self::Bfs => 2,
82            Self::ConnectedComponents => 0,
83            Self::VariableLengthPath => 3,
84            Self::PageRank => 0,
85        }
86    }
87
88    /// Is it a function with variable parameters?
89    pub fn is_variadic(&self) -> bool {
90        matches!(self, Self::VariableLengthPath)
91    }
92
93    /// Obtain the function description
94    pub fn description(&self) -> &str {
95        match self {
96            Self::Id => "Get the ID of the vertex",
97            Self::Tags => "Get all labels of a vertex",
98            Self::Labels => "Get all labels (aliases) of a vertex",
99            Self::Properties => "Get all attributes of a vertex or edge",
100            Self::EdgeType => "Get the type of the edge",
101            Self::Src => "Get the starting vertex ID of the edge",
102            Self::Dst => "Get the target vertex ID of the edge",
103            Self::Rank => "Get the edge's rank value",
104            Self::StartNode => "Get the starting vertex of the edge",
105            Self::EndNode => "Get the target vertex of the edge",
106            Self::Neighbors => "Get all neighbor vertex IDs of a vertex",
107            Self::Degree => "Get the degree (number of edges) of a vertex",
108            Self::OutEdges => "Get all outgoing edges of a vertex",
109            Self::InEdges => "Get all incoming edges of a vertex",
110            Self::ShortestPath => "Find the shortest path between two vertices (returns path length)",
111            Self::Bfs => "Breadth-first search traversal (start_vid, max_depth)",
112            Self::ConnectedComponents => "Find all connected components in the graph",
113            Self::VariableLengthPath => "Find all paths between two vertices within a depth range (start_vid, end_vid, max_depth) or (start_vid, end_vid, min_depth, max_depth)",
114            Self::PageRank => "Calculate PageRank scores for all vertices in the graph",
115        }
116    }
117
118    pub fn execute(&self, args: &[Value]) -> Result<Value, ExpressionError> {
119        if !self.is_variadic() && args.len() != self.arity() {
120            return Err(ExpressionError::invalid_arity(
121                self.name(),
122                self.arity(),
123                args.len(),
124            ));
125        }
126        match self {
127            Self::Id => execute_id(args),
128            Self::Tags => execute_tags(args),
129            Self::Labels => execute_labels(args),
130            Self::Properties => execute_properties(args),
131            Self::EdgeType => execute_edge_type(args),
132            Self::Src => execute_src(args),
133            Self::Dst => execute_dst(args),
134            Self::Rank => execute_rank(args),
135            Self::StartNode => execute_startnode(args),
136            Self::EndNode => execute_endnode(args),
137            Self::Neighbors => execute_neighbors(args),
138            Self::Degree => execute_degree(args),
139            Self::OutEdges => execute_out_edges(args),
140            Self::InEdges => execute_in_edges(args),
141            Self::ShortestPath => execute_shortest_path(args),
142            Self::Bfs => execute_bfs(args),
143            Self::ConnectedComponents => execute_connected_components(args),
144            Self::VariableLengthPath => execute_variable_length_path(args),
145            Self::PageRank => execute_pagerank(args),
146        }
147    }
148
149    /// Execute with graph storage access.
150    /// Falls back to execute() if no results can be obtained from storage.
151    pub fn execute_with_storage(
152        &self,
153        args: &[Value],
154        storage: &GraphStorageRef,
155    ) -> Result<Value, ExpressionError> {
156        if !self.is_variadic() && args.len() != self.arity() {
157            return Err(ExpressionError::invalid_arity(
158                self.name(),
159                self.arity(),
160                args.len(),
161            ));
162        }
163        match self {
164            Self::StartNode => execute_startnode_with_storage(args, storage),
165            Self::EndNode => execute_endnode_with_storage(args, storage),
166            Self::Neighbors => execute_neighbors_with_storage(args, storage),
167            Self::Degree => execute_degree_with_storage(args, storage),
168            Self::OutEdges => execute_out_edges_with_storage(args, storage),
169            Self::InEdges => execute_in_edges_with_storage(args, storage),
170            Self::ShortestPath => execute_shortest_path_with_storage(args, storage),
171            Self::Bfs => execute_bfs_with_storage(args, storage),
172            Self::ConnectedComponents => execute_connected_components_with_storage(args, storage),
173            Self::VariableLengthPath => execute_variable_length_path_with_storage(args, storage),
174            Self::PageRank => execute_pagerank_with_storage(args, storage),
175            _ => self.execute(args),
176        }
177    }
178}
179
180fn execute_id(args: &[Value]) -> Result<Value, ExpressionError> {
181    match &args[0] {
182        Value::Vertex(v) => Ok(Value::from(v.vid)),
183        Value::Null(_) => Ok(Value::Null(NullType::Null)),
184        _ => Err(ExpressionError::type_error(
185            "The id function requires a vertex type",
186        )),
187    }
188}
189
190fn execute_tags(args: &[Value]) -> Result<Value, ExpressionError> {
191    match &args[0] {
192        Value::Vertex(v) => {
193            let tags: Vec<Value> = vec![Value::string(v.tag.name.clone())];
194            Ok(Value::list(List { values: tags }))
195        }
196        Value::Null(_) => Ok(Value::Null(NullType::Null)),
197        _ => Err(ExpressionError::type_error("tags requires vertex type")),
198    }
199}
200
201fn execute_labels(args: &[Value]) -> Result<Value, ExpressionError> {
202    execute_tags(args)
203}
204
205fn execute_properties(args: &[Value]) -> Result<Value, ExpressionError> {
206    match &args[0] {
207        Value::Vertex(v) => {
208            let mut props = std::collections::HashMap::new();
209            props.extend(
210                v.properties()
211                    .iter()
212                    .map(|(k, v)| (Value::string(k.clone()), v.clone())),
213            );
214            Ok(Value::map(props))
215        }
216        Value::Edge(e) => Ok(Value::string_map(e.props.clone())),
217        Value::EdgeHeader(_) => Err(ExpressionError::type_error(
218            "properties requires a materialized edge: the edge identity reference carries no properties",
219        )),
220        Value::Map(m) => Ok(Value::map((**m).clone())),
221        Value::Null(_) => Ok(Value::Null(NullType::Null)),
222        _ => Err(ExpressionError::type_error(
223            "properties requires vertex, edge or map type",
224        )),
225    }
226}
227
228fn execute_edge_type(args: &[Value]) -> Result<Value, ExpressionError> {
229    match &args[0] {
230        Value::Edge(e) => Ok(Value::string(e.edge_type.clone())),
231        Value::EdgeHeader(h) => Ok(Value::string(h.edge_type.clone())),
232        Value::Null(_) => Ok(Value::Null(NullType::Null)),
233        _ => Err(ExpressionError::type_error(
234            "The type function requires an edge type",
235        )),
236    }
237}
238
239fn execute_src(args: &[Value]) -> Result<Value, ExpressionError> {
240    match &args[0] {
241        Value::Edge(e) => Ok(Value::from(e.src)),
242        Value::EdgeHeader(h) => Ok(Value::from(h.src)),
243        Value::Null(_) => Ok(Value::Null(NullType::Null)),
244        _ => Err(ExpressionError::type_error(
245            "The src function requires the edge type",
246        )),
247    }
248}
249
250fn execute_dst(args: &[Value]) -> Result<Value, ExpressionError> {
251    match &args[0] {
252        Value::Edge(e) => Ok(Value::from(e.dst)),
253        Value::EdgeHeader(h) => Ok(Value::from(h.dst)),
254        Value::Null(_) => Ok(Value::Null(NullType::Null)),
255        _ => Err(ExpressionError::type_error(
256            "The dst function requires an edge type",
257        )),
258    }
259}
260
261fn execute_rank(args: &[Value]) -> Result<Value, ExpressionError> {
262    match &args[0] {
263        Value::Edge(e) => Ok(Value::BigInt(e.ranking)),
264        Value::EdgeHeader(h) => Ok(Value::BigInt(h.ranking)),
265        Value::Null(_) => Ok(Value::Null(NullType::Null)),
266        _ => Err(ExpressionError::type_error(
267            "The rank function requires an edge type",
268        )),
269    }
270}
271
272fn execute_startnode(args: &[Value]) -> Result<Value, ExpressionError> {
273    match &args[0] {
274        Value::Edge(_) | Value::EdgeHeader(_) => Err(ExpressionError::function_error(
275            "startnode() requires graph storage access to resolve the endpoint label; use within a query context",
276        )),
277        Value::Null(_) => Ok(Value::Null(NullType::Null)),
278        _ => Err(ExpressionError::type_error(
279            "The startnode function requires the edge type",
280        )),
281    }
282}
283
284fn execute_endnode(args: &[Value]) -> Result<Value, ExpressionError> {
285    match &args[0] {
286        Value::Edge(_) | Value::EdgeHeader(_) => Err(ExpressionError::function_error(
287            "endnode() requires graph storage access to resolve the endpoint label; use within a query context",
288        )),
289        Value::Null(_) => Ok(Value::Null(NullType::Null)),
290        _ => Err(ExpressionError::type_error(
291            "The endnode function requires an edge type",
292        )),
293    }
294}
295
296/// Resolve one edge endpoint to a tagged vertex through the edge type
297/// schema. Returns Null when the endpoint vertex no longer exists; errors
298/// when the edge type or its endpoint labels are unknown.
299fn endpoint_vertex_with_storage(
300    edge_type: &str,
301    vid: VertexId,
302    endpoint_src: bool,
303    storage: &GraphStorageRef,
304) -> Result<Value, ExpressionError> {
305    let reader = storage.storage.read();
306    let info = reader
307        .get_edge_type(&storage.space, edge_type)
308        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?
309        .ok_or_else(|| {
310            ExpressionError::function_error(format!("Edge type '{edge_type}' not found"))
311        })?;
312    let tag = if endpoint_src {
313        &info.src_tag_name
314    } else {
315        &info.dst_tag_name
316    };
317    if tag.is_empty() {
318        return Err(ExpressionError::function_error(format!(
319            "Edge type '{edge_type}' declares no endpoint label"
320        )));
321    }
322    let vertex = reader
323        .get_vertex(&storage.space, tag, &vid)
324        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
325    match vertex {
326        Some(vertex) => Ok(Value::Vertex(Box::new(vertex))),
327        None => Ok(Value::Null(NullType::Null)),
328    }
329}
330
331fn execute_startnode_with_storage(
332    args: &[Value],
333    storage: &GraphStorageRef,
334) -> Result<Value, ExpressionError> {
335    match &args[0] {
336        Value::Edge(e) => endpoint_vertex_with_storage(&e.edge_type, e.src, true, storage),
337        Value::EdgeHeader(h) => endpoint_vertex_with_storage(&h.edge_type, h.src, true, storage),
338        Value::Null(_) => Ok(Value::Null(NullType::Null)),
339        _ => Err(ExpressionError::type_error(
340            "The startnode function requires the edge type",
341        )),
342    }
343}
344
345fn execute_endnode_with_storage(
346    args: &[Value],
347    storage: &GraphStorageRef,
348) -> Result<Value, ExpressionError> {
349    match &args[0] {
350        Value::Edge(e) => endpoint_vertex_with_storage(&e.edge_type, e.dst, false, storage),
351        Value::EdgeHeader(h) => endpoint_vertex_with_storage(&h.edge_type, h.dst, false, storage),
352        Value::Null(_) => Ok(Value::Null(NullType::Null)),
353        _ => Err(ExpressionError::type_error(
354            "The endnode function requires an edge type",
355        )),
356    }
357}
358
359fn extract_vertex_id(value: &Value) -> Result<VertexId, ExpressionError> {
360    match value {
361        Value::Vertex(v) => Ok(v.vid),
362        Value::BigInt(id) => {
363            VertexId::try_from_int64(*id).map_err(|e| ExpressionError::type_error(e.to_string()))
364        }
365        Value::Int(id) => VertexId::try_from_int64(*id as i64)
366            .map_err(|e| ExpressionError::type_error(e.to_string())),
367        Value::Null(_) => Err(ExpressionError::type_error(
368            "Expected a vertex or vertex ID, got null",
369        )),
370        _ => Err(ExpressionError::type_error(
371            "Expected a vertex or vertex ID",
372        )),
373    }
374}
375
376fn execute_neighbors(args: &[Value]) -> Result<Value, ExpressionError> {
377    match &args[0] {
378        Value::Vertex(v) => Ok(Value::from(v.vid)),
379        Value::Null(_) => Ok(Value::Null(NullType::Null)),
380        _ => Err(ExpressionError::type_error(
381            "The neighbors function requires a vertex type",
382        )),
383    }
384}
385
386fn execute_degree(args: &[Value]) -> Result<Value, ExpressionError> {
387    match &args[0] {
388        Value::Vertex(v) => {
389            let degree = v
390                .properties()
391                .keys()
392                .filter(|k| k.starts_with("neighbor_"))
393                .count();
394            Ok(Value::BigInt(degree as i64))
395        }
396        Value::Null(_) => Ok(Value::Null(NullType::Null)),
397        _ => Err(ExpressionError::type_error(
398            "The degree function requires a vertex type",
399        )),
400    }
401}
402
403fn execute_out_edges(args: &[Value]) -> Result<Value, ExpressionError> {
404    match &args[0] {
405        Value::Vertex(v) => {
406            let edges: Vec<Value> = v
407                .properties()
408                .iter()
409                .filter(|(k, _)| k.starts_with("out_"))
410                .map(|(_, v)| v.clone())
411                .collect();
412            Ok(Value::list(List { values: edges }))
413        }
414        Value::Null(_) => Ok(Value::Null(NullType::Null)),
415        _ => Err(ExpressionError::type_error(
416            "The out_edges function requires a vertex type",
417        )),
418    }
419}
420
421fn execute_in_edges(args: &[Value]) -> Result<Value, ExpressionError> {
422    match &args[0] {
423        Value::Vertex(v) => {
424            let edges: Vec<Value> = v
425                .properties()
426                .iter()
427                .filter(|(k, _)| k.starts_with("in_"))
428                .map(|(_, v)| v.clone())
429                .collect();
430            Ok(Value::list(List { values: edges }))
431        }
432        Value::Null(_) => Ok(Value::Null(NullType::Null)),
433        _ => Err(ExpressionError::type_error(
434            "The in_edges function requires a vertex type",
435        )),
436    }
437}
438
439fn execute_shortest_path(args: &[Value]) -> Result<Value, ExpressionError> {
440    let start_vid = match &args[0] {
441        Value::BigInt(id) => *id,
442        Value::Int(id) => *id as i64,
443        _ => {
444            return Err(ExpressionError::type_error(
445                "shortest_path requires integer vertex IDs",
446            ))
447        }
448    };
449    let end_vid = match &args[1] {
450        Value::BigInt(id) => *id,
451        Value::Int(id) => *id as i64,
452        _ => {
453            return Err(ExpressionError::type_error(
454                "shortest_path requires integer vertex IDs",
455            ))
456        }
457    };
458    if start_vid < 0 || end_vid < 0 {
459        return Err(ExpressionError::type_error(
460            "shortest_path requires non-negative integer vertex IDs",
461        ));
462    }
463    if start_vid == end_vid {
464        return Ok(Value::BigInt(0));
465    }
466    Ok(Value::BigInt(-1))
467}
468
469fn execute_bfs(_args: &[Value]) -> Result<Value, ExpressionError> {
470    Err(ExpressionError::function_error(
471        "bfs() requires graph storage access; use within a query context".to_string(),
472    ))
473}
474
475fn execute_connected_components(_args: &[Value]) -> Result<Value, ExpressionError> {
476    Err(ExpressionError::function_error(
477        "connected_components() requires graph storage access; use within a query context"
478            .to_string(),
479    ))
480}
481
482// --- Storage-backed implementations ---
483
484fn execute_neighbors_with_storage(
485    args: &[Value],
486    storage: &GraphStorageRef,
487) -> Result<Value, ExpressionError> {
488    let vid = extract_vertex_id(&args[0])?;
489    let neighbors = storage
490        .get_neighbors(&vid)
491        .map_err(ExpressionError::function_error)?;
492    let neighbor_ids: Vec<Value> = neighbors
493        .into_iter()
494        .map(|(nid, _)| Value::from(nid))
495        .collect();
496    Ok(Value::list(List {
497        values: neighbor_ids,
498    }))
499}
500
501fn execute_degree_with_storage(
502    args: &[Value],
503    storage: &GraphStorageRef,
504) -> Result<Value, ExpressionError> {
505    let vid = extract_vertex_id(&args[0])?;
506    let neighbors = storage
507        .get_neighbors(&vid)
508        .map_err(ExpressionError::function_error)?;
509    Ok(Value::BigInt(neighbors.len() as i64))
510}
511
512fn execute_out_edges_with_storage(
513    args: &[Value],
514    storage: &GraphStorageRef,
515) -> Result<Value, ExpressionError> {
516    use linkrs_core::types::EdgeDirection;
517    let vid = extract_vertex_id(&args[0])?;
518    let reader = storage.storage.read();
519    let edges = reader
520        .get_node_edges(&storage.space, &vid, EdgeDirection::Out, &[])
521        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
522    drop(reader);
523    let edge_values: Vec<Value> = edges
524        .into_iter()
525        .map(|e| Value::Edge(Box::new(e)))
526        .collect();
527    Ok(Value::list(List {
528        values: edge_values,
529    }))
530}
531
532fn execute_in_edges_with_storage(
533    args: &[Value],
534    storage: &GraphStorageRef,
535) -> Result<Value, ExpressionError> {
536    use linkrs_core::types::EdgeDirection;
537    let vid = extract_vertex_id(&args[0])?;
538    let reader = storage.storage.read();
539    let edges = reader
540        .get_node_edges(&storage.space, &vid, EdgeDirection::In, &[])
541        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
542    drop(reader);
543    let edge_values: Vec<Value> = edges
544        .into_iter()
545        .map(|e| Value::Edge(Box::new(e)))
546        .collect();
547    Ok(Value::list(List {
548        values: edge_values,
549    }))
550}
551
552fn execute_shortest_path_with_storage(
553    args: &[Value],
554    storage: &GraphStorageRef,
555) -> Result<Value, ExpressionError> {
556    let start_vid = extract_vertex_id(&args[0])?;
557    let end_vid = extract_vertex_id(&args[1])?;
558
559    if start_vid == end_vid {
560        return Ok(Value::BigInt(0));
561    }
562
563    // Simple BFS to find shortest path length
564    use std::collections::{HashMap, VecDeque};
565    let mut visited: HashMap<VertexId, i64> = HashMap::new();
566    let mut queue: VecDeque<VertexId> = VecDeque::new();
567
568    visited.insert(start_vid, 0);
569    queue.push_back(start_vid);
570
571    while let Some(current) = queue.pop_front() {
572        let distance = visited[&current];
573
574        let neighbors = storage
575            .get_neighbors(&current)
576            .map_err(ExpressionError::function_error)?;
577
578        for (neighbor_id, _) in neighbors {
579            if neighbor_id == end_vid {
580                return Ok(Value::BigInt(distance + 1));
581            }
582            if let std::collections::hash_map::Entry::Vacant(e) = visited.entry(neighbor_id) {
583                e.insert(distance + 1);
584                queue.push_back(neighbor_id);
585            }
586        }
587    }
588
589    Ok(Value::BigInt(-1))
590}
591
592fn execute_bfs_with_storage(
593    args: &[Value],
594    storage: &GraphStorageRef,
595) -> Result<Value, ExpressionError> {
596    let start_vid = extract_vertex_id(&args[0])?;
597    let max_depth = match &args[1] {
598        Value::BigInt(d) => *d,
599        Value::Int(d) => *d as i64,
600        _ => {
601            return Err(ExpressionError::type_error(
602                "bfs max_depth must be an integer",
603            ))
604        }
605    };
606
607    use std::collections::{HashSet, VecDeque};
608    let mut visited: HashSet<VertexId> = HashSet::new();
609    let mut queue: VecDeque<(VertexId, i64)> = VecDeque::new();
610    let mut result: Vec<Value> = Vec::new();
611
612    visited.insert(start_vid);
613    queue.push_back((start_vid, 0));
614    result.push(Value::from(start_vid));
615
616    while let Some((current, depth)) = queue.pop_front() {
617        if depth >= max_depth {
618            continue;
619        }
620
621        let neighbors = storage
622            .get_neighbors(&current)
623            .map_err(ExpressionError::function_error)?;
624
625        for (neighbor_id, _) in neighbors {
626            if visited.insert(neighbor_id) {
627                queue.push_back((neighbor_id, depth + 1));
628                result.push(Value::from(neighbor_id));
629            }
630        }
631    }
632
633    Ok(Value::list(List { values: result }))
634}
635
636fn execute_connected_components_with_storage(
637    _args: &[Value],
638    storage: &GraphStorageRef,
639) -> Result<Value, ExpressionError> {
640    // Get all vertices from storage
641    let reader = storage.storage.read();
642    let all_vertices = reader
643        .scan_vertices(&storage.space)
644        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
645    drop(reader);
646
647    use std::collections::{HashSet, VecDeque};
648
649    let all_vids: HashSet<VertexId> = all_vertices.iter().map(|v| v.vid).collect();
650    let mut visited: HashSet<VertexId> = HashSet::new();
651    let mut components: Vec<Vec<Value>> = Vec::new();
652
653    for start_vid in &all_vids {
654        if visited.contains(start_vid) {
655            continue;
656        }
657
658        let mut component: Vec<Value> = Vec::new();
659        let mut queue: VecDeque<VertexId> = VecDeque::new();
660        visited.insert(*start_vid);
661        queue.push_back(*start_vid);
662        component.push(Value::from(*start_vid));
663
664        while let Some(current) = queue.pop_front() {
665            let neighbors = match storage.get_neighbors(&current) {
666                Ok(n) => n,
667                Err(_) => continue,
668            };
669            for (neighbor_id, _) in neighbors {
670                if visited.insert(neighbor_id) {
671                    queue.push_back(neighbor_id);
672                    component.push(Value::from(neighbor_id));
673                }
674            }
675        }
676
677        components.push(component);
678    }
679
680    let component_lists: Vec<Value> = components
681        .into_iter()
682        .map(|comp| Value::list(List { values: comp }))
683        .collect();
684
685    Ok(Value::list(List {
686        values: component_lists,
687    }))
688}
689
690fn execute_variable_length_path(_args: &[Value]) -> Result<Value, ExpressionError> {
691    Err(ExpressionError::type_error(
692        "variable_length_path() requires graph storage access; use within a query context",
693    ))
694}
695
696fn execute_pagerank(_args: &[Value]) -> Result<Value, ExpressionError> {
697    Err(ExpressionError::function_error(
698        "pagerank() requires graph storage access; use within a query context".to_string(),
699    ))
700}
701
702fn execute_variable_length_path_with_storage(
703    args: &[Value],
704    storage: &GraphStorageRef,
705) -> Result<Value, ExpressionError> {
706    let (start_vid, end_vid, min_depth, max_depth) = match args.len() {
707        3 => {
708            let sv = extract_vertex_id(&args[0])?;
709            let ev = extract_vertex_id(&args[1])?;
710            let md = match &args[2] {
711                Value::BigInt(d) => *d,
712                Value::Int(d) => *d as i64,
713                _ => {
714                    return Err(ExpressionError::type_error(
715                        "variable_length_path max_depth must be an integer",
716                    ))
717                }
718            };
719            (sv, ev, 1i64, md)
720        }
721        4 => {
722            let sv = extract_vertex_id(&args[0])?;
723            let ev = extract_vertex_id(&args[1])?;
724            let mind = match &args[2] {
725                Value::BigInt(d) => *d,
726                Value::Int(d) => *d as i64,
727                _ => {
728                    return Err(ExpressionError::type_error(
729                        "variable_length_path min_depth must be an integer",
730                    ))
731                }
732            };
733            let maxd = match &args[3] {
734                Value::BigInt(d) => *d,
735                Value::Int(d) => *d as i64,
736                _ => {
737                    return Err(ExpressionError::type_error(
738                        "variable_length_path max_depth must be an integer",
739                    ))
740                }
741            };
742            (sv, ev, mind, maxd)
743        }
744        _ => {
745            return Err(ExpressionError::type_error(
746                "variable_length_path takes 3 or 4 arguments (start_vid, end_vid, max_depth) or (start_vid, end_vid, min_depth, max_depth)",
747            ))
748        }
749    };
750
751    if min_depth < 1 || max_depth < min_depth {
752        return Err(ExpressionError::type_error(
753            "variable_length_path: min_depth must be >= 1 and max_depth must be >= min_depth",
754        ));
755    }
756
757    if start_vid == end_vid && min_depth <= 0 {
758        return Ok(Value::list(List {
759            values: vec![Value::list(List {
760                values: vec![Value::from(start_vid)],
761            })],
762        }));
763    }
764
765    // DFS with depth limit to find all paths from start_vid to end_vid
766    // within the depth range [min_depth, max_depth]
767    use std::collections::VecDeque;
768    // Each element: (current_vertex, path_so_far, depth)
769    // path_so_far is the list of vertex IDs visited (excluding current)
770    let mut results: Vec<Vec<Value>> = Vec::new();
771    let mut queue: VecDeque<(VertexId, Vec<Value>, i64)> = VecDeque::new();
772
773    queue.push_back((start_vid, vec![Value::from(start_vid)], 0));
774
775    while let Some((current, path, depth)) = queue.pop_front() {
776        if depth >= max_depth {
777            continue;
778        }
779
780        let neighbors = match storage.get_neighbors(&current) {
781            Ok(n) => n,
782            Err(_) => continue,
783        };
784
785        for (neighbor_id, _) in neighbors {
786            // Avoid cycles - skip if neighbor already in path
787            let neighbor_val = Value::from(neighbor_id);
788            if path.contains(&neighbor_val) {
789                continue;
790            }
791
792            let new_depth = depth + 1;
793            let mut new_path = path.clone();
794            new_path.push(neighbor_val.clone());
795
796            if neighbor_id == end_vid && new_depth >= min_depth {
797                results.push(new_path.clone());
798            }
799
800            // Limit total results to avoid excessive memory usage
801            if results.len() >= 1000 {
802                break;
803            }
804
805            queue.push_back((neighbor_id, new_path, new_depth));
806        }
807
808        if results.len() >= 1000 {
809            break;
810        }
811    }
812
813    let path_values: Vec<Value> = results
814        .into_iter()
815        .map(|path| Value::list(List { values: path }))
816        .collect();
817
818    Ok(Value::list(List {
819        values: path_values,
820    }))
821}
822
823fn execute_pagerank_with_storage(
824    _args: &[Value],
825    storage: &GraphStorageRef,
826) -> Result<Value, ExpressionError> {
827    let reader = storage.storage.read();
828    let all_vertices = reader
829        .scan_vertices(&storage.space)
830        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
831    drop(reader);
832
833    let vertex_count = all_vertices.len();
834    if vertex_count == 0 {
835        return Ok(Value::map(std::collections::HashMap::new()));
836    }
837
838    use std::collections::HashMap;
839
840    let vertex_ids: Vec<VertexId> = all_vertices.iter().map(|v| v.vid).collect();
841    let mut vid_to_index: HashMap<VertexId, usize> = HashMap::new();
842    for (i, vid) in vertex_ids.iter().enumerate() {
843        vid_to_index.insert(*vid, i);
844    }
845
846    let damping_factor: f64 = 0.85;
847    let max_iterations: usize = 100;
848    let tolerance: f64 = 1e-6;
849
850    let mut scores: Vec<f64> = vec![1.0 / vertex_count as f64; vertex_count];
851    let mut out_degrees: Vec<usize> = vec![0; vertex_count];
852
853    let mut adjacency: Vec<Vec<usize>> = vec![Vec::new(); vertex_count];
854
855    for vid in &vertex_ids {
856        let reader = storage.storage.read();
857        let edges = reader
858            .get_node_edges(
859                &storage.space,
860                vid,
861                linkrs_core::types::EdgeDirection::Out,
862                &[],
863            )
864            .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
865        drop(reader);
866
867        let src_index = vid_to_index[vid];
868        out_degrees[src_index] = edges.len();
869
870        for edge in edges {
871            if let Some(&dst_index) = vid_to_index.get(&edge.dst) {
872                adjacency[src_index].push(dst_index);
873            }
874        }
875    }
876
877    for _ in 0..max_iterations {
878        let mut new_scores: Vec<f64> = vec![0.0; vertex_count];
879        let mut dangling_sum: f64 = 0.0;
880
881        for i in 0..vertex_count {
882            if out_degrees[i] == 0 {
883                dangling_sum += scores[i];
884            }
885        }
886
887        let base = (1.0 - damping_factor) / vertex_count as f64
888            + damping_factor * dangling_sum / vertex_count as f64;
889        for score in new_scores.iter_mut() {
890            *score = base;
891        }
892
893        for i in 0..vertex_count {
894            if out_degrees[i] > 0 {
895                let share = scores[i] / out_degrees[i] as f64;
896                for &j in &adjacency[i] {
897                    new_scores[j] += damping_factor * share;
898                }
899            }
900        }
901
902        let mut diff: f64 = 0.0;
903        for i in 0..vertex_count {
904            diff += (new_scores[i] - scores[i]).abs();
905        }
906
907        scores = new_scores;
908
909        if diff < tolerance {
910            break;
911        }
912    }
913
914    let mut result: HashMap<String, Value> = HashMap::new();
915    for (i, vid) in vertex_ids.iter().enumerate() {
916        let vid_str = vid.to_string();
917        result.insert(vid_str, Value::Double(scores[i]));
918    }
919
920    Ok(Value::string_map(result))
921}
922
923#[cfg(test)]
924mod tests {
925    use super::*;
926    use linkrs_core::types::VertexId;
927    use linkrs_core::vertex_edge_path::{Edge, Tag, Vertex};
928    use std::collections::HashMap;
929
930    fn create_test_vertex() -> Vertex {
931        let mut props = HashMap::from([
932            ("name".into(), Value::string("Alice")),
933            ("age".into(), Value::Int(30)),
934        ]);
935        props.insert("dept".into(), Value::string("Engineering"));
936        let tag = Tag::new("person".into(), props);
937        Vertex::new(VertexId::try_from_int64(1).expect("valid vertex id"), tag)
938    }
939
940    fn create_test_edge() -> Edge {
941        Edge::new(
942            VertexId::try_from_int64(1).expect("valid vertex id"),
943            VertexId::try_from_int64(2).expect("valid vertex id"),
944            "knows".to_string(),
945            0,
946            HashMap::from([("since".into(), Value::Int(2020))]),
947        )
948    }
949
950    #[test]
951    fn test_id_function() {
952        let vertex = create_test_vertex();
953        let result = GraphFunction::Id
954            .execute(&[Value::Vertex(Box::new(vertex))])
955            .expect("The id function execution should succeed");
956        assert_eq!(result, Value::BigInt(1));
957    }
958
959    #[test]
960    fn test_tags_function() {
961        let vertex = create_test_vertex();
962        let result = GraphFunction::Tags
963            .execute(&[Value::Vertex(Box::new(vertex))])
964            .expect("The tags function should be executed successfully");
965        if let Value::List(tags) = result {
966            assert_eq!(tags.values.len(), 1);
967        } else {
968            panic!("The `tags` function should return a list.");
969        }
970    }
971
972    #[test]
973    fn test_properties_vertex() {
974        let vertex = create_test_vertex();
975        let result = GraphFunction::Properties
976            .execute(&[Value::Vertex(Box::new(vertex))])
977            .expect("The properties function should execute successfully");
978        if let Value::Map(props) = result {
979            assert!(props.contains_key(&Value::string("name")));
980            assert!(props.contains_key(&Value::string("age")));
981            assert!(props.contains_key(&Value::string("dept")));
982        } else {
983            panic!("The `properties` function should return a map.");
984        }
985    }
986
987    #[test]
988    fn test_type_function() {
989        let edge = create_test_edge();
990        let result = GraphFunction::EdgeType
991            .execute(&[Value::Edge(Box::new(edge))])
992            .expect("The type function execution should succeed");
993        assert_eq!(result, Value::string("knows"));
994    }
995
996    #[test]
997    fn test_src_function() {
998        let edge = create_test_edge();
999        let result = GraphFunction::Src
1000            .execute(&[Value::Edge(Box::new(edge))])
1001            .expect("The src function execution should succeed");
1002        assert_eq!(result, Value::BigInt(1));
1003    }
1004
1005    #[test]
1006    fn test_dst_function() {
1007        let edge = create_test_edge();
1008        let result = GraphFunction::Dst
1009            .execute(&[Value::Edge(Box::new(edge))])
1010            .expect("The dst function execution should succeed");
1011        assert_eq!(result, Value::BigInt(2));
1012    }
1013
1014    #[test]
1015    fn test_rank_function() {
1016        let edge = create_test_edge();
1017        let result = GraphFunction::Rank
1018            .execute(&[Value::Edge(Box::new(edge))])
1019            .expect("The execution of the rank function should succeed");
1020        assert_eq!(result, Value::BigInt(0));
1021    }
1022
1023    #[test]
1024    fn test_null_handling() {
1025        let null_value = Value::Null(NullType::Null);
1026
1027        assert_eq!(
1028            GraphFunction::Id
1029                .execute(std::slice::from_ref(&null_value))
1030                .expect("The id function should handle NULL"),
1031            Value::Null(NullType::Null)
1032        );
1033        assert_eq!(
1034            GraphFunction::Tags
1035                .execute(std::slice::from_ref(&null_value))
1036                .expect("The tags function should handle NULL"),
1037            Value::Null(NullType::Null)
1038        );
1039        assert_eq!(
1040            GraphFunction::Properties
1041                .execute(std::slice::from_ref(&null_value))
1042                .expect("The properties function should handle NULL."),
1043            Value::Null(NullType::Null)
1044        );
1045    }
1046}
linkrs_query/executor/expression/functions/builtin/graph.rs

linkrs_query/executor/expression/functions/builtin/
graph.rs
