arrow-graph 0.2.0

Arrow-native graph processing engine with SQL interface
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

use arrow::record_batch::RecordBatch;
use arrow::datatypes::{DataType, Field, Schema};
use std::sync::Arc;
use std::path::Path;
use crate::error::{GraphError, Result};
use crate::graph::GraphIndexes;

#[derive(Debug, Clone)]
pub struct ArrowGraph {
    pub nodes: RecordBatch,
    pub edges: RecordBatch,
    pub indexes: GraphIndexes,
}

impl ArrowGraph {
    /// Create a new graph from nodes and edges RecordBatches
    pub fn new(nodes: RecordBatch, edges: RecordBatch) -> Result<Self> {
        let indexes = GraphIndexes::build(&nodes, &edges)?;
        
        Ok(ArrowGraph {
            nodes,
            edges,
            indexes,
        })
    }
    
    /// Create a graph from just edges (nodes will be inferred)
    pub fn from_edges(edges: RecordBatch) -> Result<Self> {
        // Create an empty nodes RecordBatch with proper schema
        let nodes_schema = Arc::new(Schema::new(vec![
            Field::new("id", DataType::Utf8, false),
        ]));
        
        let empty_nodes = RecordBatch::new_empty(nodes_schema);
        Self::new(empty_nodes, edges)
    }
    
    /// Load graph from Arrow/Parquet files
    pub async fn from_files<P: AsRef<Path>>(
        _nodes_path: P,
        _edges_path: P,
    ) -> Result<Self> {
        todo!("Implement loading from files - will read Arrow/Parquet files")
    }
    
    /// Create graph from RecordBatches (alias for new)
    pub fn from_tables(
        nodes: RecordBatch,
        edges: RecordBatch,
    ) -> Result<Self> {
        Self::new(nodes, edges)
    }
    
    /// Execute SQL query with graph functions
    pub async fn sql(&self, _query: &str) -> Result<RecordBatch> {
        todo!("Implement SQL execution using DataFusion with graph functions")
    }
    
    /// Get number of nodes in the graph
    pub fn node_count(&self) -> usize {
        self.indexes.node_count
    }
    
    /// Get number of edges in the graph  
    pub fn edge_count(&self) -> usize {
        self.indexes.edge_count
    }
    
    /// Calculate graph density (edges / max_possible_edges)
    pub fn density(&self) -> f64 {
        let n = self.node_count() as f64;
        let m = self.edge_count() as f64;
        
        if n <= 1.0 {
            0.0
        } else {
            m / (n * (n - 1.0))
        }
    }
    
    /// Get neighbors of a node
    pub fn neighbors(&self, node_id: &str) -> Option<&Vec<String>> {
        self.indexes.neighbors(node_id)
    }
    
    /// Get predecessors of a node (incoming edges)
    pub fn predecessors(&self, node_id: &str) -> Option<&Vec<String>> {
        self.indexes.predecessors(node_id)
    }
    
    /// Check if node exists in graph
    pub fn has_node(&self, node_id: &str) -> bool {
        self.indexes.has_node(node_id)
    }
    
    /// Get edge weight between two nodes
    pub fn edge_weight(&self, source: &str, target: &str) -> Option<f64> {
        self.indexes.edge_weight(source, target)
    }
    
    /// Get all node IDs
    pub fn node_ids(&self) -> impl Iterator<Item = &String> {
        self.indexes.all_nodes()
    }
}