use datafusion::error::Result as DataFusionResult;
use std::collections::HashMap;
pub struct MultiGraphProcessor;
#[derive(Debug, Clone, PartialEq)]
pub struct NamedGraph {
pub name: String,
pub edges: Vec<MultiGraphEdge>,
pub properties: HashMap<String, String>,
pub creation_time: Option<i64>,
}
#[derive(Debug, Clone, PartialEq)]
pub struct MultiGraphEdge {
pub source: String,
pub target: String,
pub weight: f64,
pub graph_name: String, pub edge_type: Option<String>, pub properties: HashMap<String, String>,
}
#[derive(Debug, Clone)]
pub struct MultiGraphConfig {
pub include_graph_metadata: bool,
pub merge_duplicate_edges: bool,
pub edge_weight_combination: EdgeWeightCombination,
}
#[derive(Debug, Clone)]
pub enum EdgeWeightCombination {
Sum,
Average,
Maximum,
Minimum,
KeepFirst,
KeepLast,
}
impl Default for MultiGraphConfig {
fn default() -> Self {
Self {
include_graph_metadata: true,
merge_duplicate_edges: false,
edge_weight_combination: EdgeWeightCombination::Average,
}
}
}
impl MultiGraphProcessor {
pub fn new() -> Self {
Self
}
pub fn create_graph(
&self,
name: &str,
edges: Vec<MultiGraphEdge>,
properties: Option<HashMap<String, String>>,
) -> DataFusionResult<NamedGraph> {
let graph = NamedGraph {
name: name.to_string(),
edges,
properties: properties.unwrap_or_default(),
creation_time: Some(std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs() as i64),
};
Ok(graph)
}
pub fn merge_graphs(
&self,
graphs: &[NamedGraph],
config: &MultiGraphConfig,
) -> DataFusionResult<NamedGraph> {
let mut merged_edges = Vec::new();
let mut merged_properties = HashMap::new();
for graph in graphs {
merged_edges.extend(graph.edges.clone());
for (key, value) in &graph.properties {
let prefixed_key = format!("{}_{}", graph.name, key);
merged_properties.insert(prefixed_key, value.clone());
}
}
if config.merge_duplicate_edges {
merged_edges = self.merge_duplicate_edges(merged_edges, &config.edge_weight_combination)?;
}
let merged_graph_name = graphs
.iter()
.map(|g| g.name.as_str())
.collect::<Vec<_>>()
.join("_merged_");
Ok(NamedGraph {
name: merged_graph_name,
edges: merged_edges,
properties: merged_properties,
creation_time: Some(std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs() as i64),
})
}
fn merge_duplicate_edges(
&self,
edges: Vec<MultiGraphEdge>,
combination: &EdgeWeightCombination,
) -> DataFusionResult<Vec<MultiGraphEdge>> {
let mut edge_groups: HashMap<(String, String), Vec<MultiGraphEdge>> = HashMap::new();
for edge in edges {
let key = (edge.source.clone(), edge.target.clone());
edge_groups.entry(key).or_insert_with(Vec::new).push(edge);
}
let mut merged_edges = Vec::new();
for ((source, target), group) in edge_groups {
if group.len() == 1 {
merged_edges.push(group.into_iter().next().unwrap());
} else {
let combined_weight = match combination {
EdgeWeightCombination::Sum => group.iter().map(|e| e.weight).sum(),
EdgeWeightCombination::Average => {
group.iter().map(|e| e.weight).sum::<f64>() / group.len() as f64
}
EdgeWeightCombination::Maximum => {
group.iter().map(|e| e.weight).fold(f64::NEG_INFINITY, f64::max)
}
EdgeWeightCombination::Minimum => {
group.iter().map(|e| e.weight).fold(f64::INFINITY, f64::min)
}
EdgeWeightCombination::KeepFirst => group.first().unwrap().weight,
EdgeWeightCombination::KeepLast => group.last().unwrap().weight,
};
let combined_graph_names = group
.iter()
.map(|e| e.graph_name.as_str())
.collect::<Vec<_>>()
.join(",");
let mut combined_properties = HashMap::new();
for edge in &group {
for (key, value) in &edge.properties {
let prefixed_key = format!("{}_{}", edge.graph_name, key);
combined_properties.insert(prefixed_key, value.clone());
}
}
let merged_edge = MultiGraphEdge {
source,
target,
weight: combined_weight,
graph_name: combined_graph_names,
edge_type: group.first().unwrap().edge_type.clone(),
properties: combined_properties,
};
merged_edges.push(merged_edge);
}
}
Ok(merged_edges)
}
pub fn query_multi_graph(
&self,
graphs: &[NamedGraph],
graph_names: Option<&[String]>,
edge_filter: Option<Box<dyn Fn(&MultiGraphEdge) -> bool>>,
) -> DataFusionResult<Vec<MultiGraphEdge>> {
let mut result_edges = Vec::new();
for graph in graphs {
if let Some(names) = graph_names {
if !names.contains(&graph.name) {
continue;
}
}
let filtered_edges = if let Some(ref filter) = edge_filter {
graph.edges.iter().filter(|e| filter(e)).cloned().collect()
} else {
graph.edges.clone()
};
result_edges.extend(filtered_edges);
}
Ok(result_edges)
}
pub fn find_cross_graph_connections(
&self,
graphs: &[NamedGraph],
) -> DataFusionResult<Vec<(String, String, Vec<String>)>> {
let mut node_to_graphs: HashMap<String, Vec<String>> = HashMap::new();
for graph in graphs {
for edge in &graph.edges {
node_to_graphs
.entry(edge.source.clone())
.or_insert_with(Vec::new)
.push(graph.name.clone());
node_to_graphs
.entry(edge.target.clone())
.or_insert_with(Vec::new)
.push(graph.name.clone());
}
}
let mut cross_graph_connections = Vec::new();
for (node, graph_list) in node_to_graphs {
if graph_list.len() > 1 {
let mut unique_graphs = graph_list;
unique_graphs.sort();
unique_graphs.dedup();
if unique_graphs.len() > 1 {
cross_graph_connections.push((
"cross_graph_node".to_string(),
node,
unique_graphs,
));
}
}
}
Ok(cross_graph_connections)
}
pub fn calculate_graph_similarity(
&self,
graph1: &NamedGraph,
graph2: &NamedGraph,
) -> DataFusionResult<f64> {
let edges1: std::collections::HashSet<(String, String)> = graph1
.edges
.iter()
.map(|e| (e.source.clone(), e.target.clone()))
.collect();
let edges2: std::collections::HashSet<(String, String)> = graph2
.edges
.iter()
.map(|e| (e.source.clone(), e.target.clone()))
.collect();
let intersection_size = edges1.intersection(&edges2).count();
let union_size = edges1.union(&edges2).count();
if union_size == 0 {
Ok(0.0)
} else {
Ok(intersection_size as f64 / union_size as f64)
}
}
pub fn get_multi_graph_statistics(
&self,
graphs: &[NamedGraph],
) -> DataFusionResult<HashMap<String, GraphStatistics>> {
let mut stats = HashMap::new();
for graph in graphs {
let graph_stats = self.calculate_graph_statistics(graph)?;
stats.insert(graph.name.clone(), graph_stats);
}
Ok(stats)
}
fn calculate_graph_statistics(&self, graph: &NamedGraph) -> DataFusionResult<GraphStatistics> {
let edge_count = graph.edges.len();
let mut nodes = std::collections::HashSet::new();
let mut total_weight = 0.0;
for edge in &graph.edges {
nodes.insert(edge.source.clone());
nodes.insert(edge.target.clone());
total_weight += edge.weight;
}
let node_count = nodes.len();
let average_weight = if edge_count > 0 {
total_weight / edge_count as f64
} else {
0.0
};
let density = if node_count > 1 {
edge_count as f64 / (node_count * (node_count - 1)) as f64
} else {
0.0
};
Ok(GraphStatistics {
node_count,
edge_count,
average_edge_weight: average_weight,
density,
total_weight,
})
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct GraphStatistics {
pub node_count: usize,
pub edge_count: usize,
pub average_edge_weight: f64,
pub density: f64,
pub total_weight: f64,
}
impl Default for MultiGraphProcessor {
fn default() -> Self {
Self::new()
}
}
pub struct MultiGraphQueries;
impl MultiGraphQueries {
pub fn find_largest_graph(graphs: &[NamedGraph]) -> Option<&NamedGraph> {
graphs.iter().max_by_key(|g| g.edges.len())
}
pub fn find_graphs_with_node<'a>(graphs: &'a [NamedGraph], node_id: &str) -> Vec<&'a NamedGraph> {
graphs
.iter()
.filter(|graph| {
graph.edges.iter().any(|edge|
edge.source == node_id || edge.target == node_id
)
})
.collect()
}
pub fn get_all_unique_nodes(graphs: &[NamedGraph]) -> Vec<String> {
let mut nodes = std::collections::HashSet::new();
for graph in graphs {
for edge in &graph.edges {
nodes.insert(edge.source.clone());
nodes.insert(edge.target.clone());
}
}
nodes.into_iter().collect()
}
pub fn find_overlapping_nodes(graph1: &NamedGraph, graph2: &NamedGraph) -> Vec<String> {
let nodes1: std::collections::HashSet<String> = graph1
.edges
.iter()
.flat_map(|e| vec![e.source.clone(), e.target.clone()])
.collect();
let nodes2: std::collections::HashSet<String> = graph2
.edges
.iter()
.flat_map(|e| vec![e.source.clone(), e.target.clone()])
.collect();
nodes1.intersection(&nodes2).cloned().collect()
}
}
#[cfg(test)]
mod tests {
use super::*;
fn create_test_graph(name: &str, edges: Vec<(String, String, f64)>) -> NamedGraph {
let multi_edges = edges
.into_iter()
.map(|(source, target, weight)| MultiGraphEdge {
source,
target,
weight,
graph_name: name.to_string(),
edge_type: None,
properties: HashMap::new(),
})
.collect();
NamedGraph {
name: name.to_string(),
edges: multi_edges,
properties: HashMap::new(),
creation_time: Some(1704067200),
}
}
#[test]
fn test_create_graph() {
let processor = MultiGraphProcessor::new();
let edges = vec![
MultiGraphEdge {
source: "A".to_string(),
target: "B".to_string(),
weight: 1.0,
graph_name: "test_graph".to_string(),
edge_type: None,
properties: HashMap::new(),
}
];
let graph = processor.create_graph("test_graph", edges, None).unwrap();
assert_eq!(graph.name, "test_graph");
assert_eq!(graph.edges.len(), 1);
}
#[test]
fn test_merge_graphs() {
let processor = MultiGraphProcessor::new();
let graph1 = create_test_graph("graph1", vec![
("A".to_string(), "B".to_string(), 1.0),
("B".to_string(), "C".to_string(), 2.0),
]);
let graph2 = create_test_graph("graph2", vec![
("C".to_string(), "D".to_string(), 3.0),
("D".to_string(), "A".to_string(), 4.0),
]);
let config = MultiGraphConfig::default();
let merged = processor.merge_graphs(&[graph1, graph2], &config).unwrap();
assert_eq!(merged.edges.len(), 4);
assert!(merged.name.contains("merged"));
}
#[test]
fn test_merge_duplicate_edges() {
let processor = MultiGraphProcessor::new();
let edges = vec![
MultiGraphEdge {
source: "A".to_string(),
target: "B".to_string(),
weight: 1.0,
graph_name: "graph1".to_string(),
edge_type: None,
properties: HashMap::new(),
},
MultiGraphEdge {
source: "A".to_string(),
target: "B".to_string(),
weight: 3.0,
graph_name: "graph2".to_string(),
edge_type: None,
properties: HashMap::new(),
},
];
let merged = processor.merge_duplicate_edges(edges, &EdgeWeightCombination::Average).unwrap();
assert_eq!(merged.len(), 1);
assert_eq!(merged[0].weight, 2.0); assert!(merged[0].graph_name.contains("graph1") && merged[0].graph_name.contains("graph2"));
}
#[test]
fn test_query_multi_graph() {
let processor = MultiGraphProcessor::new();
let graph1 = create_test_graph("graph1", vec![
("A".to_string(), "B".to_string(), 1.0),
("B".to_string(), "C".to_string(), 2.0),
]);
let graph2 = create_test_graph("graph2", vec![
("C".to_string(), "D".to_string(), 3.0),
("D".to_string(), "A".to_string(), 4.0),
]);
let result = processor.query_multi_graph(
&[graph1.clone(), graph2.clone()],
Some(&vec!["graph1".to_string()]),
None,
).unwrap();
assert_eq!(result.len(), 2);
let result_filtered = processor.query_multi_graph(
&[graph1, graph2],
None,
Some(Box::new(|edge| edge.weight > 2.0)),
).unwrap();
assert_eq!(result_filtered.len(), 2); }
#[test]
fn test_find_cross_graph_connections() {
let processor = MultiGraphProcessor::new();
let graph1 = create_test_graph("graph1", vec![
("A".to_string(), "B".to_string(), 1.0),
("B".to_string(), "C".to_string(), 2.0),
]);
let graph2 = create_test_graph("graph2", vec![
("C".to_string(), "D".to_string(), 3.0), ("D".to_string(), "E".to_string(), 4.0),
]);
let connections = processor.find_cross_graph_connections(&[graph1, graph2]).unwrap();
assert!(!connections.is_empty());
assert!(connections.iter().any(|(_, node, _)| node == "C"));
}
#[test]
fn test_calculate_graph_similarity() {
let processor = MultiGraphProcessor::new();
let graph1 = create_test_graph("graph1", vec![
("A".to_string(), "B".to_string(), 1.0),
("B".to_string(), "C".to_string(), 2.0),
]);
let graph2 = create_test_graph("graph2", vec![
("A".to_string(), "B".to_string(), 1.0), ("C".to_string(), "D".to_string(), 3.0), ]);
let similarity = processor.calculate_graph_similarity(&graph1, &graph2).unwrap();
assert!(similarity > 0.0 && similarity < 1.0);
}
#[test]
fn test_get_multi_graph_statistics() {
let processor = MultiGraphProcessor::new();
let graph1 = create_test_graph("graph1", vec![
("A".to_string(), "B".to_string(), 1.0),
("B".to_string(), "C".to_string(), 2.0),
]);
let graph2 = create_test_graph("graph2", vec![
("C".to_string(), "D".to_string(), 3.0),
]);
let stats = processor.get_multi_graph_statistics(&[graph1, graph2]).unwrap();
assert_eq!(stats.len(), 2);
assert!(stats.contains_key("graph1"));
assert!(stats.contains_key("graph2"));
let graph1_stats = &stats["graph1"];
assert_eq!(graph1_stats.edge_count, 2);
assert_eq!(graph1_stats.node_count, 3); }
#[test]
fn test_multi_graph_queries() {
let graph1 = create_test_graph("graph1", vec![
("A".to_string(), "B".to_string(), 1.0),
("B".to_string(), "C".to_string(), 2.0),
]);
let graph2 = create_test_graph("graph2", vec![
("C".to_string(), "D".to_string(), 3.0),
("D".to_string(), "E".to_string(), 4.0),
]);
let graphs = vec![graph1, graph2];
let largest = MultiGraphQueries::find_largest_graph(&graphs);
assert!(largest.is_some());
assert_eq!(largest.unwrap().edges.len(), 2);
let graphs_with_c = MultiGraphQueries::find_graphs_with_node(&graphs, "C");
assert_eq!(graphs_with_c.len(), 2);
let all_nodes = MultiGraphQueries::get_all_unique_nodes(&graphs);
assert!(all_nodes.len() >= 5);
let overlapping = MultiGraphQueries::find_overlapping_nodes(&graphs[0], &graphs[1]);
assert!(overlapping.contains(&"C".to_string()));
}
#[test]
fn test_edge_weight_combinations() {
let processor = MultiGraphProcessor::new();
let edges = vec![
MultiGraphEdge {
source: "A".to_string(),
target: "B".to_string(),
weight: 2.0,
graph_name: "graph1".to_string(),
edge_type: None,
properties: HashMap::new(),
},
MultiGraphEdge {
source: "A".to_string(),
target: "B".to_string(),
weight: 4.0,
graph_name: "graph2".to_string(),
edge_type: None,
properties: HashMap::new(),
},
];
let sum_result = processor.merge_duplicate_edges(edges.clone(), &EdgeWeightCombination::Sum).unwrap();
assert_eq!(sum_result[0].weight, 6.0);
let max_result = processor.merge_duplicate_edges(edges.clone(), &EdgeWeightCombination::Maximum).unwrap();
assert_eq!(max_result[0].weight, 4.0);
let min_result = processor.merge_duplicate_edges(edges.clone(), &EdgeWeightCombination::Minimum).unwrap();
assert_eq!(min_result[0].weight, 2.0);
let first_result = processor.merge_duplicate_edges(edges.clone(), &EdgeWeightCombination::KeepFirst).unwrap();
assert_eq!(first_result[0].weight, 2.0);
let last_result = processor.merge_duplicate_edges(edges, &EdgeWeightCombination::KeepLast).unwrap();
assert_eq!(last_result[0].weight, 4.0);
}
}