use crate::error::Result;
use crate::streaming::incremental::{IncrementalGraphProcessor, UpdateResult};
use arrow::array::Array;
use std::collections::{HashMap, VecDeque};
#[derive(Debug, Clone)]
pub struct SlidingWindowProcessor {
window_size: std::time::Duration,
max_snapshots: usize,
snapshots: VecDeque<GraphSnapshot>,
current_metrics: HashMap<String, f64>,
}
#[derive(Debug, Clone)]
pub struct GraphSnapshot {
pub timestamp: u64,
pub node_count: usize,
pub edge_count: usize,
pub density: f64,
pub avg_degree: f64,
pub largest_component_size: usize,
pub component_count: usize,
pub clustering_coefficient: f64,
pub custom_metrics: HashMap<String, f64>,
}
#[derive(Debug, Clone)]
pub struct TemporalConfig {
pub window_duration: std::time::Duration,
pub snapshot_interval: std::time::Duration,
pub max_snapshots: usize,
pub track_custom_metrics: Vec<String>,
}
#[derive(Debug, Clone)]
pub struct TemporalAnalytics {
pub trend_analysis: HashMap<String, TrendInfo>,
pub volatility_metrics: HashMap<String, f64>,
pub change_points: Vec<ChangePoint>,
pub evolution_patterns: Vec<EvolutionPattern>,
}
#[derive(Debug, Clone)]
pub struct TrendInfo {
pub metric_name: String,
pub direction: TrendDirection,
pub slope: f64,
pub correlation: f64,
pub volatility: f64,
pub recent_change: f64,
}
#[derive(Debug, Clone, PartialEq)]
pub enum TrendDirection {
Increasing,
Decreasing,
Stable,
Volatile,
}
#[derive(Debug, Clone)]
pub struct ChangePoint {
pub timestamp: u64,
pub metric_name: String,
pub before_value: f64,
pub after_value: f64,
pub change_magnitude: f64,
pub confidence: f64,
}
#[derive(Debug, Clone)]
pub struct EvolutionPattern {
pub pattern_type: PatternType,
pub start_time: u64,
pub end_time: u64,
pub affected_metrics: Vec<String>,
pub strength: f64,
}
#[derive(Debug, Clone, PartialEq)]
pub enum PatternType {
Growth, Decay, Burst, Oscillation, PhaseTransition, Stabilization, }
impl Default for TemporalConfig {
fn default() -> Self {
Self {
window_duration: std::time::Duration::from_secs(3600), snapshot_interval: std::time::Duration::from_secs(60), max_snapshots: 1000,
track_custom_metrics: vec![
"pagerank_entropy".to_string(),
"edge_density_variance".to_string(),
"community_modularity".to_string(),
],
}
}
}
impl SlidingWindowProcessor {
pub fn new(config: TemporalConfig) -> Self {
Self {
window_size: config.window_duration,
max_snapshots: config.max_snapshots,
snapshots: VecDeque::new(),
current_metrics: HashMap::new(),
}
}
pub fn default() -> Self {
Self::new(TemporalConfig::default())
}
pub fn initialize(&mut self, processor: &IncrementalGraphProcessor) -> Result<()> {
let snapshot = self.create_snapshot(processor)?;
self.snapshots.push_back(snapshot);
Ok(())
}
pub fn update(&mut self, processor: &IncrementalGraphProcessor, _changes: &UpdateResult) -> Result<()> {
let current_time = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
let snapshot = self.create_snapshot(processor)?;
self.snapshots.push_back(snapshot);
self.cleanup_old_snapshots(current_time);
while self.snapshots.len() > self.max_snapshots {
self.snapshots.pop_front();
}
Ok(())
}
fn create_snapshot(&self, processor: &IncrementalGraphProcessor) -> Result<GraphSnapshot> {
let graph = processor.graph();
let node_count = graph.node_count();
let edge_count = graph.edge_count();
let density = if node_count > 1 {
edge_count as f64 / (node_count * (node_count - 1)) as f64
} else {
0.0
};
let avg_degree = if node_count > 0 {
(2 * edge_count) as f64 / node_count as f64
} else {
0.0
};
let components = self.analyze_components(processor)?;
let largest_component_size = components.iter().map(|c| c.len()).max().unwrap_or(0);
let component_count = components.len();
let clustering_coefficient = self.estimate_clustering_coefficient(processor)?;
let timestamp = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap()
.as_secs();
Ok(GraphSnapshot {
timestamp,
node_count,
edge_count,
density,
avg_degree,
largest_component_size,
component_count,
clustering_coefficient,
custom_metrics: HashMap::new(), })
}
fn analyze_components(&self, processor: &IncrementalGraphProcessor) -> Result<Vec<Vec<String>>> {
let graph = processor.graph();
let edges_batch = &graph.edges;
let nodes_batch = &graph.nodes;
let mut components = Vec::new();
let mut visited = std::collections::HashSet::new();
let mut nodes = Vec::new();
if nodes_batch.num_rows() > 0 {
let node_ids = nodes_batch.column(0)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.ok_or_else(|| crate::error::GraphError::graph_construction("Expected string array for node IDs"))?;
for i in 0..node_ids.len() {
nodes.push(node_ids.value(i).to_string());
}
}
let mut adjacency: HashMap<String, Vec<String>> = HashMap::new();
if edges_batch.num_rows() > 0 {
let source_ids = edges_batch.column(0)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.ok_or_else(|| crate::error::GraphError::graph_construction("Expected string array for source IDs"))?;
let target_ids = edges_batch.column(1)
.as_any()
.downcast_ref::<arrow::array::StringArray>()
.ok_or_else(|| crate::error::GraphError::graph_construction("Expected string array for target IDs"))?;
for i in 0..source_ids.len() {
let source = source_ids.value(i).to_string();
let target = target_ids.value(i).to_string();
adjacency.entry(source.clone()).or_insert_with(Vec::new).push(target.clone());
adjacency.entry(target).or_insert_with(Vec::new).push(source);
}
}
for node in nodes {
if !visited.contains(&node) {
let mut component = Vec::new();
let mut stack = vec![node.clone()];
while let Some(current) = stack.pop() {
if visited.insert(current.clone()) {
component.push(current.clone());
if let Some(neighbors) = adjacency.get(¤t) {
for neighbor in neighbors {
if !visited.contains(neighbor) {
stack.push(neighbor.clone());
}
}
}
}
}
if !component.is_empty() {
components.push(component);
}
}
}
Ok(components)
}
fn estimate_clustering_coefficient(&self, _processor: &IncrementalGraphProcessor) -> Result<f64> {
Ok(0.3) }
fn cleanup_old_snapshots(&mut self, current_time: u64) {
let window_start = current_time.saturating_sub(self.window_size.as_secs());
while let Some(front) = self.snapshots.front() {
if front.timestamp < window_start {
self.snapshots.pop_front();
} else {
break;
}
}
}
pub fn analyze_temporal_patterns(&self) -> Result<TemporalAnalytics> {
if self.snapshots.len() < 2 {
return Ok(TemporalAnalytics {
trend_analysis: HashMap::new(),
volatility_metrics: HashMap::new(),
change_points: Vec::new(),
evolution_patterns: Vec::new(),
});
}
let trend_analysis = self.analyze_trends()?;
let volatility_metrics = self.calculate_volatility()?;
let change_points = self.detect_change_points()?;
let evolution_patterns = self.detect_evolution_patterns()?;
Ok(TemporalAnalytics {
trend_analysis,
volatility_metrics,
change_points,
evolution_patterns,
})
}
fn analyze_trends(&self) -> Result<HashMap<String, TrendInfo>> {
let mut trends = HashMap::new();
let metrics = ["node_count", "edge_count", "density", "avg_degree", "component_count"];
for metric in &metrics {
let values: Vec<f64> = self.snapshots.iter()
.map(|s| self.get_metric_value(s, metric))
.collect();
if values.len() >= 2 {
let trend_info = self.calculate_trend_info(metric, &values);
trends.insert(metric.to_string(), trend_info);
}
}
Ok(trends)
}
fn get_metric_value(&self, snapshot: &GraphSnapshot, metric: &str) -> f64 {
match metric {
"node_count" => snapshot.node_count as f64,
"edge_count" => snapshot.edge_count as f64,
"density" => snapshot.density,
"avg_degree" => snapshot.avg_degree,
"component_count" => snapshot.component_count as f64,
"clustering_coefficient" => snapshot.clustering_coefficient,
_ => snapshot.custom_metrics.get(metric).copied().unwrap_or(0.0),
}
}
fn calculate_trend_info(&self, metric_name: &str, values: &[f64]) -> TrendInfo {
let n = values.len() as f64;
let x_values: Vec<f64> = (0..values.len()).map(|i| i as f64).collect();
let x_mean = x_values.iter().sum::<f64>() / n;
let y_mean = values.iter().sum::<f64>() / n;
let numerator: f64 = x_values.iter().zip(values.iter())
.map(|(x, y)| (x - x_mean) * (y - y_mean))
.sum();
let denominator: f64 = x_values.iter()
.map(|x| (x - x_mean).powi(2))
.sum();
let slope = if denominator.abs() > 1e-10 {
numerator / denominator
} else {
0.0
};
let x_std = (x_values.iter().map(|x| (x - x_mean).powi(2)).sum::<f64>() / n).sqrt();
let y_std = (values.iter().map(|y| (y - y_mean).powi(2)).sum::<f64>() / n).sqrt();
let correlation = if x_std > 1e-10 && y_std > 1e-10 {
numerator / (n * x_std * y_std)
} else {
0.0
};
let volatility = y_std;
let direction = if slope.abs() < 0.01 {
if volatility > y_mean * 0.1 {
TrendDirection::Volatile
} else {
TrendDirection::Stable
}
} else if slope > 0.0 {
TrendDirection::Increasing
} else {
TrendDirection::Decreasing
};
let recent_change = if values.len() > 1 {
let first = values[0];
let last = values[values.len() - 1];
if first.abs() > 1e-10 {
(last - first) / first
} else {
0.0
}
} else {
0.0
};
TrendInfo {
metric_name: metric_name.to_string(),
direction,
slope,
correlation,
volatility,
recent_change,
}
}
fn calculate_volatility(&self) -> Result<HashMap<String, f64>> {
let mut volatility = HashMap::new();
let metrics = ["node_count", "edge_count", "density", "avg_degree"];
for metric in &metrics {
let values: Vec<f64> = self.snapshots.iter()
.map(|s| self.get_metric_value(s, metric))
.collect();
if values.len() > 1 {
let mean = values.iter().sum::<f64>() / values.len() as f64;
let variance = values.iter()
.map(|v| (v - mean).powi(2))
.sum::<f64>() / values.len() as f64;
let vol = variance.sqrt();
volatility.insert(metric.to_string(), vol);
}
}
Ok(volatility)
}
fn detect_change_points(&self) -> Result<Vec<ChangePoint>> {
let mut change_points = Vec::new();
let metrics = ["node_count", "edge_count", "density", "avg_degree"];
for metric in &metrics {
let values: Vec<f64> = self.snapshots.iter()
.map(|s| self.get_metric_value(s, metric))
.collect();
if values.len() < 10 {
continue; }
let window_size = values.len() / 4; if window_size < 2 {
continue;
}
for i in window_size..(values.len() - window_size) {
let before_mean = values[(i - window_size)..i].iter().sum::<f64>() / window_size as f64;
let after_mean = values[i..(i + window_size)].iter().sum::<f64>() / window_size as f64;
let change_magnitude = (after_mean - before_mean).abs();
let relative_change = if before_mean.abs() > 1e-10 {
change_magnitude / before_mean.abs()
} else {
0.0
};
if relative_change > 0.2 {
let confidence = (relative_change - 0.2) / 0.8; change_points.push(ChangePoint {
timestamp: self.snapshots[i].timestamp,
metric_name: metric.to_string(),
before_value: before_mean,
after_value: after_mean,
change_magnitude,
confidence: confidence.min(1.0),
});
}
}
}
Ok(change_points)
}
fn detect_evolution_patterns(&self) -> Result<Vec<EvolutionPattern>> {
let mut patterns = Vec::new();
if self.snapshots.len() < 5 {
return Ok(patterns);
}
let node_values: Vec<f64> = self.snapshots.iter()
.map(|s| s.node_count as f64)
.collect();
if self.is_growth_pattern(&node_values) {
patterns.push(EvolutionPattern {
pattern_type: PatternType::Growth,
start_time: self.snapshots[0].timestamp,
end_time: self.snapshots[self.snapshots.len() - 1].timestamp,
affected_metrics: vec!["node_count".to_string(), "edge_count".to_string()],
strength: self.calculate_pattern_strength(&node_values),
});
}
let edge_values: Vec<f64> = self.snapshots.iter()
.map(|s| s.edge_count as f64)
.collect();
if let Some((start_idx, end_idx)) = self.detect_burst_pattern(&edge_values) {
patterns.push(EvolutionPattern {
pattern_type: PatternType::Burst,
start_time: self.snapshots[start_idx].timestamp,
end_time: self.snapshots[end_idx].timestamp,
affected_metrics: vec!["edge_count".to_string(), "density".to_string()],
strength: self.calculate_burst_strength(&edge_values, start_idx, end_idx),
});
}
Ok(patterns)
}
fn is_growth_pattern(&self, values: &[f64]) -> bool {
if values.len() < 3 {
return false;
}
let increasing_count = values.windows(2)
.filter(|w| w[1] > w[0])
.count();
increasing_count as f64 / (values.len() - 1) as f64 > 0.7 }
fn detect_burst_pattern(&self, values: &[f64]) -> Option<(usize, usize)> {
if values.len() < 5 {
return None;
}
let mean = values.iter().sum::<f64>() / values.len() as f64;
let std = (values.iter().map(|v| (v - mean).powi(2)).sum::<f64>() / values.len() as f64).sqrt();
let threshold = mean + 2.0 * std;
for i in 1..(values.len() - 1) {
if values[i] > threshold && values[i - 1] < threshold {
for j in (i + 1)..values.len() {
if values[j] < threshold {
return Some((i, j));
}
}
}
}
None
}
fn calculate_pattern_strength(&self, values: &[f64]) -> f64 {
if values.len() < 2 {
return 0.0;
}
let first = values[0];
let last = values[values.len() - 1];
if first.abs() > 1e-10 {
((last - first) / first).abs().min(1.0)
} else {
0.0
}
}
fn calculate_burst_strength(&self, values: &[f64], start_idx: usize, end_idx: usize) -> f64 {
if start_idx >= values.len() || end_idx >= values.len() || start_idx >= end_idx {
return 0.0;
}
let baseline = values[0..start_idx].iter().sum::<f64>() / start_idx.max(1) as f64;
let peak = values[start_idx..=end_idx].iter().fold(0.0f64, |a, b| a.max(*b));
if baseline > 1e-10 {
((peak - baseline) / baseline).min(2.0) / 2.0 } else {
0.0
}
}
pub fn window_stats(&self) -> WindowStats {
let duration = if let (Some(first), Some(last)) = (self.snapshots.front(), self.snapshots.back()) {
last.timestamp - first.timestamp
} else {
0
};
WindowStats {
snapshot_count: self.snapshots.len(),
window_duration_seconds: duration,
oldest_timestamp: self.snapshots.front().map(|s| s.timestamp).unwrap_or(0),
newest_timestamp: self.snapshots.back().map(|s| s.timestamp).unwrap_or(0),
}
}
pub fn recent_snapshots(&self, count: usize) -> Vec<&GraphSnapshot> {
self.snapshots.iter().rev().take(count).collect()
}
pub fn metric_evolution(&self, metric_name: &str) -> Vec<(u64, f64)> {
self.snapshots.iter()
.map(|s| (s.timestamp, self.get_metric_value(s, metric_name)))
.collect()
}
}
#[derive(Debug, Clone)]
pub struct WindowStats {
pub snapshot_count: usize,
pub window_duration_seconds: u64,
pub oldest_timestamp: u64,
pub newest_timestamp: u64,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::graph::ArrowGraph;
use crate::streaming::incremental::IncrementalGraphProcessor;
use arrow::array::{StringArray, Float64Array};
use arrow::record_batch::RecordBatch;
use arrow::datatypes::{Schema, Field, DataType};
use std::sync::Arc;
fn create_test_graph() -> Result<ArrowGraph> {
let nodes_schema = Arc::new(Schema::new(vec![
Field::new("id", DataType::Utf8, false),
]));
let node_ids = StringArray::from(vec!["A", "B", "C"]);
let nodes_batch = RecordBatch::try_new(
nodes_schema,
vec![Arc::new(node_ids)],
)?;
let edges_schema = Arc::new(Schema::new(vec![
Field::new("source", DataType::Utf8, false),
Field::new("target", DataType::Utf8, false),
Field::new("weight", DataType::Float64, false),
]));
let sources = StringArray::from(vec!["A", "B"]);
let targets = StringArray::from(vec!["B", "C"]);
let weights = Float64Array::from(vec![1.0, 1.0]);
let edges_batch = RecordBatch::try_new(
edges_schema,
vec![Arc::new(sources), Arc::new(targets), Arc::new(weights)],
)?;
ArrowGraph::new(nodes_batch, edges_batch)
}
#[test]
fn test_sliding_window_initialization() {
let graph = create_test_graph().unwrap();
let processor = IncrementalGraphProcessor::new(graph).unwrap();
let mut window = SlidingWindowProcessor::default();
window.initialize(&processor).unwrap();
assert_eq!(window.snapshots.len(), 1);
let snapshot = &window.snapshots[0];
assert_eq!(snapshot.node_count, 3);
assert_eq!(snapshot.edge_count, 2);
assert!(snapshot.density > 0.0);
}
#[test]
fn test_sliding_window_update() {
let graph = create_test_graph().unwrap();
let mut processor = IncrementalGraphProcessor::new(graph).unwrap();
let mut window = SlidingWindowProcessor::default();
window.initialize(&processor).unwrap();
processor.add_edge("A".to_string(), "C".to_string(), 1.0).unwrap();
let update_result = UpdateResult {
vertices_added: 0,
vertices_removed: 0,
edges_added: 1,
edges_removed: 0,
affected_components: vec![],
recomputation_needed: false,
};
window.update(&processor, &update_result).unwrap();
assert_eq!(window.snapshots.len(), 2);
let latest = &window.snapshots[1];
assert_eq!(latest.edge_count, 2); }
#[test]
fn test_trend_analysis() {
let mut window = SlidingWindowProcessor::default();
for i in 0..5 {
let snapshot = GraphSnapshot {
timestamp: i,
node_count: (i + 1) as usize,
edge_count: i as usize,
density: (i as f64) * 0.1,
avg_degree: (i as f64) * 0.5,
largest_component_size: (i + 1) as usize,
component_count: 1,
clustering_coefficient: 0.3,
custom_metrics: HashMap::new(),
};
window.snapshots.push_back(snapshot);
}
let analytics = window.analyze_temporal_patterns().unwrap();
assert!(!analytics.trend_analysis.is_empty());
let node_trend = analytics.trend_analysis.get("node_count").unwrap();
assert_eq!(node_trend.direction, TrendDirection::Increasing);
assert!(node_trend.slope > 0.0);
}
#[test]
fn test_change_point_detection() {
let mut window = SlidingWindowProcessor::default();
let values = vec![1.0, 1.0, 1.0, 1.0, 1.0, 5.0, 5.0, 5.0, 5.0, 5.0];
for (i, &value) in values.iter().enumerate() {
let snapshot = GraphSnapshot {
timestamp: i as u64,
node_count: value as usize,
edge_count: 0,
density: 0.0,
avg_degree: 0.0,
largest_component_size: 1,
component_count: 1,
clustering_coefficient: 0.0,
custom_metrics: HashMap::new(),
};
window.snapshots.push_back(snapshot);
}
let analytics = window.analyze_temporal_patterns().unwrap();
assert!(!analytics.change_points.is_empty());
let change_point = &analytics.change_points[0];
assert_eq!(change_point.metric_name, "node_count");
assert!(change_point.change_magnitude > 0.0);
}
#[test]
fn test_evolution_pattern_detection() {
let mut window = SlidingWindowProcessor::default();
for i in 0..10 {
let snapshot = GraphSnapshot {
timestamp: i,
node_count: (i + 1) as usize,
edge_count: (i * 2) as usize,
density: 0.1,
avg_degree: 1.0,
largest_component_size: (i + 1) as usize,
component_count: 1,
clustering_coefficient: 0.3,
custom_metrics: HashMap::new(),
};
window.snapshots.push_back(snapshot);
}
let analytics = window.analyze_temporal_patterns().unwrap();
assert!(!analytics.evolution_patterns.is_empty());
let growth_pattern = analytics.evolution_patterns.iter()
.find(|p| p.pattern_type == PatternType::Growth);
assert!(growth_pattern.is_some());
let pattern = growth_pattern.unwrap();
assert!(pattern.strength > 0.0);
assert!(pattern.affected_metrics.contains(&"node_count".to_string()));
}
#[test]
fn test_metric_evolution() {
let mut window = SlidingWindowProcessor::default();
for i in 0..5 {
let snapshot = GraphSnapshot {
timestamp: i * 1000, node_count: ((i + 1) * 2) as usize,
edge_count: i as usize,
density: 0.1,
avg_degree: 1.0,
largest_component_size: 1,
component_count: 1,
clustering_coefficient: 0.3,
custom_metrics: HashMap::new(),
};
window.snapshots.push_back(snapshot);
}
let evolution = window.metric_evolution("node_count");
assert_eq!(evolution.len(), 5);
assert_eq!(evolution[0], (0, 2.0));
assert_eq!(evolution[1], (1000, 4.0));
assert_eq!(evolution[4], (4000, 10.0));
}
#[test]
fn test_window_stats() {
let mut window = SlidingWindowProcessor::default();
for i in 0..3 {
let snapshot = GraphSnapshot {
timestamp: i * 1000,
node_count: 1,
edge_count: 0,
density: 0.0,
avg_degree: 0.0,
largest_component_size: 1,
component_count: 1,
clustering_coefficient: 0.0,
custom_metrics: HashMap::new(),
};
window.snapshots.push_back(snapshot);
}
let stats = window.window_stats();
assert_eq!(stats.snapshot_count, 3);
assert_eq!(stats.window_duration_seconds, 2000);
assert_eq!(stats.oldest_timestamp, 0);
assert_eq!(stats.newest_timestamp, 2000);
}
}