use crate::error::Result;
use crate::streaming::incremental::IncrementalGraphProcessor;
use arrow::array::Array;
use std::collections::HashMap;
use std::hash::{Hash, Hasher};
#[derive(Debug, Clone)]
pub struct HyperLogLog {
buckets: Vec<u8>,
bucket_count: usize,
alpha: f64,
}
impl HyperLogLog {
pub fn new(precision: u8) -> Self {
let precision = precision.clamp(4, 16);
let bucket_count = 1 << precision;
let alpha = match bucket_count {
16 => 0.673,
32 => 0.697,
64 => 0.709,
_ => 0.7213 / (1.0 + 1.079 / bucket_count as f64),
};
Self {
buckets: vec![0; bucket_count],
bucket_count,
alpha,
}
}
pub fn add<T: Hash>(&mut self, element: &T) {
let hash = self.hash_element(element);
let bucket_index = (hash >> (64 - self.precision_bits())) as usize;
let leading_zeros = (hash << self.precision_bits()).leading_zeros() as u8 + 1;
self.buckets[bucket_index] = self.buckets[bucket_index].max(leading_zeros);
}
pub fn estimate(&self) -> f64 {
let raw_estimate = self.alpha * (self.bucket_count as f64).powi(2) /
self.buckets.iter().map(|&b| 2.0_f64.powf(-(b as f64))).sum::<f64>();
if raw_estimate <= 2.5 * self.bucket_count as f64 {
let zero_count = self.buckets.iter().filter(|&&b| b == 0).count();
if zero_count != 0 {
return (self.bucket_count as f64) * (self.bucket_count as f64 / zero_count as f64).ln();
}
}
if raw_estimate <= (1.0 / 30.0) * (1u64 << 32) as f64 {
raw_estimate
} else {
-((1u64 << 32) as f64) * (1.0 - raw_estimate / ((1u64 << 32) as f64)).ln()
}
}
pub fn merge(&mut self, other: &HyperLogLog) -> Result<()> {
if self.bucket_count != other.bucket_count {
return Err(crate::error::GraphError::graph_construction("HyperLogLog bucket counts must match for merge"));
}
for i in 0..self.bucket_count {
self.buckets[i] = self.buckets[i].max(other.buckets[i]);
}
Ok(())
}
fn hash_element<T: Hash>(&self, element: &T) -> u64 {
let mut hasher = std::collections::hash_map::DefaultHasher::new();
element.hash(&mut hasher);
hasher.finish()
}
fn precision_bits(&self) -> u32 {
(self.bucket_count as f64).log2() as u32
}
}
#[derive(Debug, Clone)]
pub struct CountMinSketch {
counters: Vec<Vec<u32>>,
hash_functions: Vec<HashFunction>,
width: usize,
depth: usize,
}
#[derive(Debug, Clone)]
struct HashFunction {
a: u64,
b: u64,
p: u64, }
impl CountMinSketch {
pub fn new(width: usize, depth: usize) -> Self {
let mut hash_functions = Vec::new();
let p = 2147483647u64;
for i in 0..depth {
hash_functions.push(HashFunction {
a: (i as u64 * 2 + 1) * 1000000007,
b: (i as u64 + 1) * 1000000009,
p,
});
}
Self {
counters: vec![vec![0; width]; depth],
hash_functions,
width,
depth,
}
}
pub fn add<T: Hash>(&mut self, element: &T, count: u32) {
let hash = self.hash_element(element);
for (i, hash_func) in self.hash_functions.iter().enumerate() {
let index = hash_func.hash(hash) % self.width as u64;
self.counters[i][index as usize] += count;
}
}
pub fn estimate<T: Hash>(&self, element: &T) -> u32 {
let hash = self.hash_element(element);
let mut min_count = u32::MAX;
for (i, hash_func) in self.hash_functions.iter().enumerate() {
let index = hash_func.hash(hash) % self.width as u64;
min_count = min_count.min(self.counters[i][index as usize]);
}
min_count
}
pub fn merge(&mut self, other: &CountMinSketch) -> Result<()> {
if self.width != other.width || self.depth != other.depth {
return Err(crate::error::GraphError::graph_construction("Count-Min Sketch dimensions must match for merge"));
}
for i in 0..self.depth {
for j in 0..self.width {
self.counters[i][j] += other.counters[i][j];
}
}
Ok(())
}
fn hash_element<T: Hash>(&self, element: &T) -> u64 {
let mut hasher = std::collections::hash_map::DefaultHasher::new();
element.hash(&mut hasher);
hasher.finish()
}
}
impl HashFunction {
fn hash(&self, x: u64) -> u64 {
((self.a.wrapping_mul(x).wrapping_add(self.b)) % self.p) % self.p
}
}
#[derive(Debug, Clone)]
pub struct BloomFilter {
bit_array: Vec<bool>,
size: usize,
hash_count: u32,
element_count: usize,
}
impl BloomFilter {
pub fn new(expected_elements: usize, false_positive_rate: f64) -> Self {
let size = Self::optimal_size(expected_elements, false_positive_rate);
let hash_count = Self::optimal_hash_count(size, expected_elements);
Self {
bit_array: vec![false; size],
size,
hash_count,
element_count: 0,
}
}
pub fn add<T: Hash>(&mut self, element: &T) {
let hashes = self.hash_element(element);
for i in 0..self.hash_count {
let index = (hashes.0.wrapping_add((i as u64).wrapping_mul(hashes.1))) % self.size as u64;
self.bit_array[index as usize] = true;
}
self.element_count += 1;
}
pub fn contains<T: Hash>(&self, element: &T) -> bool {
let hashes = self.hash_element(element);
for i in 0..self.hash_count {
let index = (hashes.0.wrapping_add((i as u64).wrapping_mul(hashes.1))) % self.size as u64;
if !self.bit_array[index as usize] {
return false;
}
}
true
}
pub fn false_positive_probability(&self) -> f64 {
let filled_ratio = self.bit_array.iter().filter(|&&b| b).count() as f64 / self.size as f64;
filled_ratio.powf(self.hash_count as f64)
}
pub fn union(&mut self, other: &BloomFilter) -> Result<()> {
if self.size != other.size || self.hash_count != other.hash_count {
return Err(crate::error::GraphError::graph_construction("Bloom Filter parameters must match for union"));
}
for i in 0..self.size {
self.bit_array[i] |= other.bit_array[i];
}
self.element_count += other.element_count;
Ok(())
}
fn optimal_size(expected_elements: usize, false_positive_rate: f64) -> usize {
let m = -(expected_elements as f64 * false_positive_rate.ln()) / (2.0_f64.ln().powi(2));
m.ceil() as usize
}
fn optimal_hash_count(size: usize, expected_elements: usize) -> u32 {
let k = (size as f64 / expected_elements as f64) * 2.0_f64.ln();
k.round().max(1.0) as u32
}
fn hash_element<T: Hash>(&self, element: &T) -> (u64, u64) {
let mut hasher1 = std::collections::hash_map::DefaultHasher::new();
let mut hasher2 = std::collections::hash_map::DefaultHasher::new();
element.hash(&mut hasher1);
(element, 1u8).hash(&mut hasher2);
(hasher1.finish(), hasher2.finish())
}
}
#[derive(Debug)]
pub struct ApproximateGraphProcessor {
unique_nodes: HyperLogLog,
unique_edges: HyperLogLog,
unique_triangles: HyperLogLog,
node_degrees: CountMinSketch,
edge_weights: CountMinSketch,
active_nodes: BloomFilter,
recent_edges: BloomFilter,
config: ApproximateConfig,
}
#[derive(Debug, Clone)]
pub struct ApproximateConfig {
pub hyperloglog_precision: u8,
pub count_min_width: usize,
pub count_min_depth: usize,
pub bloom_expected_elements: usize,
pub bloom_false_positive_rate: f64,
}
impl Default for ApproximateConfig {
fn default() -> Self {
Self {
hyperloglog_precision: 12, count_min_width: 1000,
count_min_depth: 5,
bloom_expected_elements: 10000,
bloom_false_positive_rate: 0.01, }
}
}
impl ApproximateGraphProcessor {
pub fn new(config: ApproximateConfig) -> Self {
Self {
unique_nodes: HyperLogLog::new(config.hyperloglog_precision),
unique_edges: HyperLogLog::new(config.hyperloglog_precision),
unique_triangles: HyperLogLog::new(config.hyperloglog_precision),
node_degrees: CountMinSketch::new(config.count_min_width, config.count_min_depth),
edge_weights: CountMinSketch::new(config.count_min_width, config.count_min_depth),
active_nodes: BloomFilter::new(config.bloom_expected_elements, config.bloom_false_positive_rate),
recent_edges: BloomFilter::new(config.bloom_expected_elements * 2, config.bloom_false_positive_rate),
config,
}
}
pub fn initialize(&mut self, processor: &IncrementalGraphProcessor) -> Result<()> {
let graph = processor.graph();
let nodes_batch = &graph.nodes;
let edges_batch = &graph.edges;
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() {
let node_id = node_ids.value(i);
self.unique_nodes.add(&node_id);
self.active_nodes.add(&node_id);
}
}
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"))?;
let weights = edges_batch.column(2)
.as_any()
.downcast_ref::<arrow::array::Float64Array>()
.ok_or_else(|| crate::error::GraphError::graph_construction("Expected float64 array for weights"))?;
for i in 0..source_ids.len() {
let source = source_ids.value(i);
let target = target_ids.value(i);
let weight = weights.value(i);
let edge = (source, target);
self.unique_edges.add(&edge);
self.recent_edges.add(&edge);
self.node_degrees.add(&source, 1);
self.node_degrees.add(&target, 1);
let weight_key = format!("{}→{}", source, target);
self.edge_weights.add(&weight_key, (weight * 1000.0) as u32); }
}
Ok(())
}
pub fn update(&mut self, processor: &IncrementalGraphProcessor) -> Result<ApproximateMetrics> {
self.initialize(processor)?;
Ok(self.get_metrics())
}
pub fn get_metrics(&self) -> ApproximateMetrics {
ApproximateMetrics {
estimated_unique_nodes: self.unique_nodes.estimate(),
estimated_unique_edges: self.unique_edges.estimate(),
estimated_unique_triangles: self.unique_triangles.estimate(),
active_nodes_bloom_fill_ratio: self.active_nodes.false_positive_probability(),
recent_edges_bloom_fill_ratio: self.recent_edges.false_positive_probability(),
}
}
pub fn is_node_active(&self, node_id: &str) -> bool {
self.active_nodes.contains(&node_id)
}
pub fn is_edge_recent(&self, source: &str, target: &str) -> bool {
let edge = (source, target);
self.recent_edges.contains(&edge)
}
pub fn estimate_node_degree(&self, node_id: &str) -> u32 {
self.node_degrees.estimate(&node_id)
}
pub fn estimate_edge_weight(&self, source: &str, target: &str) -> f64 {
let weight_key = format!("{}→{}", source, target);
self.edge_weights.estimate(&weight_key) as f64 / 1000.0 }
pub fn merge(&mut self, other: &ApproximateGraphProcessor) -> Result<()> {
self.unique_nodes.merge(&other.unique_nodes)?;
self.unique_edges.merge(&other.unique_edges)?;
self.unique_triangles.merge(&other.unique_triangles)?;
self.node_degrees.merge(&other.node_degrees)?;
self.edge_weights.merge(&other.edge_weights)?;
self.active_nodes.union(&other.active_nodes)?;
self.recent_edges.union(&other.recent_edges)?;
Ok(())
}
pub fn add_triangle(&mut self, node_a: &str, node_b: &str, node_c: &str) {
let mut nodes = vec![node_a, node_b, node_c];
nodes.sort();
let triangle = format!("{}→{}→{}", nodes[0], nodes[1], nodes[2]);
self.unique_triangles.add(&triangle);
}
pub fn reset(&mut self) {
self.unique_nodes = HyperLogLog::new(self.config.hyperloglog_precision);
self.unique_edges = HyperLogLog::new(self.config.hyperloglog_precision);
self.unique_triangles = HyperLogLog::new(self.config.hyperloglog_precision);
self.node_degrees = CountMinSketch::new(self.config.count_min_width, self.config.count_min_depth);
self.edge_weights = CountMinSketch::new(self.config.count_min_width, self.config.count_min_depth);
self.active_nodes = BloomFilter::new(self.config.bloom_expected_elements, self.config.bloom_false_positive_rate);
self.recent_edges = BloomFilter::new(self.config.bloom_expected_elements * 2, self.config.bloom_false_positive_rate);
}
}
#[derive(Debug, Clone)]
pub struct ApproximateMetrics {
pub estimated_unique_nodes: f64,
pub estimated_unique_edges: f64,
pub estimated_unique_triangles: f64,
pub active_nodes_bloom_fill_ratio: f64,
pub recent_edges_bloom_fill_ratio: f64,
}
#[cfg(test)]
mod tests {
use super::*;
use crate::graph::ArrowGraph;
use arrow::array::{StringArray, Float64Array};
use arrow::record_batch::RecordBatch;
use arrow::datatypes::{Schema, Field, DataType};
use std::sync::Arc;
#[test]
fn test_hyperloglog_basic() {
let mut hll = HyperLogLog::new(12);
for i in 0..1000 {
hll.add(&format!("element_{}", i));
}
let estimate = hll.estimate();
assert!(estimate > 950.0 && estimate < 1050.0);
}
#[test]
fn test_hyperloglog_merge() {
let mut hll1 = HyperLogLog::new(8);
let mut hll2 = HyperLogLog::new(8);
for i in 0..500 {
hll1.add(&format!("a_{}", i));
hll2.add(&format!("b_{}", i));
}
let estimate1 = hll1.estimate();
let estimate2 = hll2.estimate();
hll1.merge(&hll2).unwrap();
let merged_estimate = hll1.estimate();
assert!(merged_estimate > estimate1 + estimate2 - 100.0);
}
#[test]
fn test_count_min_sketch() {
let mut cms = CountMinSketch::new(100, 5);
cms.add(&"element1", 10);
cms.add(&"element2", 5);
cms.add(&"element1", 3);
assert!(cms.estimate(&"element1") >= 13);
assert!(cms.estimate(&"element2") >= 5);
assert_eq!(cms.estimate(&"nonexistent"), 0);
}
#[test]
fn test_bloom_filter() {
let mut bloom = BloomFilter::new(1000, 0.01);
bloom.add(&"element1");
bloom.add(&"element2");
bloom.add(&"element3");
assert!(bloom.contains(&"element1"));
assert!(bloom.contains(&"element2"));
assert!(bloom.contains(&"element3"));
assert!(!bloom.contains(&"nonexistent"));
}
#[test]
fn test_bloom_filter_union() {
let mut bloom1 = BloomFilter::new(1000, 0.01);
let mut bloom2 = BloomFilter::new(1000, 0.01);
bloom1.add(&"element1");
bloom2.add(&"element2");
bloom1.union(&bloom2).unwrap();
assert!(bloom1.contains(&"element1"));
assert!(bloom1.contains(&"element2"));
}
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", "D"]);
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", "C"]);
let targets = StringArray::from(vec!["B", "C", "D"]);
let weights = Float64Array::from(vec![1.0, 2.0, 3.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_approximate_graph_processor() {
let graph = create_test_graph().unwrap();
let processor = IncrementalGraphProcessor::new(graph).unwrap();
let mut approx = ApproximateGraphProcessor::new(ApproximateConfig::default());
approx.initialize(&processor).unwrap();
let metrics = approx.get_metrics();
assert!(metrics.estimated_unique_nodes > 3.0 && metrics.estimated_unique_nodes < 5.0);
assert!(metrics.estimated_unique_edges > 2.0 && metrics.estimated_unique_edges < 4.0);
}
#[test]
fn test_node_and_edge_queries() {
let graph = create_test_graph().unwrap();
let processor = IncrementalGraphProcessor::new(graph).unwrap();
let mut approx = ApproximateGraphProcessor::new(ApproximateConfig::default());
approx.initialize(&processor).unwrap();
assert!(approx.is_node_active("A"));
assert!(approx.is_node_active("B"));
assert!(!approx.is_node_active("nonexistent"));
assert!(approx.is_edge_recent("A", "B"));
assert!(approx.is_edge_recent("B", "C"));
assert!(approx.estimate_node_degree("A") > 0);
assert!(approx.estimate_node_degree("B") > 0);
}
#[test]
fn test_triangle_counting() {
let mut approx = ApproximateGraphProcessor::new(ApproximateConfig::default());
approx.add_triangle("A", "B", "C");
approx.add_triangle("B", "C", "D");
approx.add_triangle("A", "B", "C");
let metrics = approx.get_metrics();
assert!(metrics.estimated_unique_triangles > 1.5 && metrics.estimated_unique_triangles < 2.5);
}
#[test]
fn test_processor_merge() {
let config = ApproximateConfig::default();
let mut approx1 = ApproximateGraphProcessor::new(config.clone());
let mut approx2 = ApproximateGraphProcessor::new(config);
approx1.unique_nodes.add(&"A");
approx1.unique_nodes.add(&"B");
approx2.unique_nodes.add(&"C");
approx2.unique_nodes.add(&"D");
let estimate1 = approx1.get_metrics().estimated_unique_nodes;
let estimate2 = approx2.get_metrics().estimated_unique_nodes;
approx1.merge(&approx2).unwrap();
let merged_estimate = approx1.get_metrics().estimated_unique_nodes;
assert!(merged_estimate > estimate1);
assert!(merged_estimate > estimate2);
}
}