dbsp 0.287.0

Continuous streaming analytics engine
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
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280

use petgraph::graph::{EdgeIndex, NodeIndex, UnGraph};
use petgraph::visit::{Bfs, EdgeRef, VisitMap, Visitable};

/// Compute the connected components of an undirected graph.
///
/// Returns a vector of tuples, where each tuple contains a vector of node indices
/// and a vector of edge indices for a connected component.
pub fn components<N, E>(g: &UnGraph<N, E>) -> Vec<(Vec<NodeIndex>, Vec<EdgeIndex>)> {
    let mut visited_nodes = g.visit_map();
    let mut visited_edges = std::collections::HashSet::new();
    let mut result: Vec<(Vec<NodeIndex>, Vec<EdgeIndex>)> = Vec::new();

    for start in g.node_indices() {
        if visited_nodes.is_visited(&start) {
            continue;
        }

        let mut nodes = Vec::new();
        let mut bfs = Bfs::new(g, start);
        visited_nodes.visit(start);

        while let Some(n) = bfs.next(g) {
            visited_nodes.visit(n);
            nodes.push(n);
        }

        // Collect all edges where both endpoints are in this component
        let node_set: std::collections::HashSet<_> = nodes.iter().cloned().collect();
        let mut edges = Vec::new();
        for edge_ref in g.edge_references() {
            let edge_idx = edge_ref.id();
            if visited_edges.contains(&edge_idx) {
                continue;
            }
            let (a, b) = (edge_ref.source(), edge_ref.target());
            if node_set.contains(&a) && node_set.contains(&b) {
                visited_edges.insert(edge_idx);
                edges.push(edge_idx);
            }
        }

        result.push((nodes, edges));
    }

    result
}

#[cfg(test)]
mod tests {
    use super::components;
    use petgraph::graph::UnGraph;

    #[test]
    fn test_empty_graph() {
        let g: UnGraph<(), ()> = UnGraph::new_undirected();
        let comps = components(&g);
        assert_eq!(comps.len(), 0);
    }

    #[test]
    fn test_single_node() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        let n1 = g.add_node(1);
        let comps = components(&g);
        assert_eq!(comps.len(), 1);
        assert_eq!(comps[0].0.len(), 1);
        assert_eq!(comps[0].0[0], n1);
        assert_eq!(comps[0].1.len(), 0); // No edges for a single node
    }

    #[test]
    fn test_two_disconnected_nodes() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        let n1 = g.add_node(1);
        let n2 = g.add_node(2);
        let comps = components(&g);
        assert_eq!(comps.len(), 2);
        // Each component should have one node
        assert_eq!(comps.iter().map(|c| c.0.len()).sum::<usize>(), 2);
        // Check that both nodes are present
        let all_nodes: Vec<_> = comps.iter().flat_map(|c| &c.0).collect();
        assert!(all_nodes.contains(&&n1));
        assert!(all_nodes.contains(&&n2));
        // No edges in disconnected components
        assert_eq!(comps.iter().map(|c| c.1.len()).sum::<usize>(), 0);
    }

    #[test]
    fn test_two_connected_nodes() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        let n1 = g.add_node(1);
        let n2 = g.add_node(2);
        let e1 = g.add_edge(n1, n2, 0);
        let comps = components(&g);
        assert_eq!(comps.len(), 1);
        assert_eq!(comps[0].0.len(), 2);
        assert!(comps[0].0.contains(&n1));
        assert!(comps[0].0.contains(&n2));
        assert_eq!(comps[0].1.len(), 1);
        assert!(comps[0].1.contains(&e1));
    }

    #[test]
    fn test_three_nodes_two_components() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        let n1 = g.add_node(1);
        let n2 = g.add_node(2);
        let n3 = g.add_node(3);
        // n1 and n2 are connected, n3 is isolated
        let e1 = g.add_edge(n1, n2, 0);
        let comps = components(&g);
        assert_eq!(comps.len(), 2);
        // One component with 2 nodes, one with 1 node
        let sizes: Vec<usize> = comps.iter().map(|c| c.0.len()).collect();
        assert!(sizes.contains(&1));
        assert!(sizes.contains(&2));
        // Verify n3 is in a component by itself
        let n3_component = comps.iter().find(|c| c.0.contains(&n3)).unwrap();
        assert_eq!(n3_component.0.len(), 1);
        assert_eq!(n3_component.1.len(), 0); // Isolated node has no edges
        // Verify the connected component has the edge
        let connected_component = comps.iter().find(|c| c.0.contains(&n1)).unwrap();
        assert_eq!(connected_component.1.len(), 1);
        assert!(connected_component.1.contains(&e1));
    }

    #[test]
    fn test_three_nodes_one_component() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        let n1 = g.add_node(1);
        let n2 = g.add_node(2);
        let n3 = g.add_node(3);
        let e1 = g.add_edge(n1, n2, 0);
        let e2 = g.add_edge(n2, n3, 0);
        let comps = components(&g);
        assert_eq!(comps.len(), 1);
        assert_eq!(comps[0].0.len(), 3);
        assert_eq!(comps[0].1.len(), 2);
        assert!(comps[0].1.contains(&e1));
        assert!(comps[0].1.contains(&e2));
    }

    #[test]
    fn test_linear_path() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        let n1 = g.add_node(1);
        let n2 = g.add_node(2);
        let n3 = g.add_node(3);
        let n4 = g.add_node(4);
        // Create a path: n1 - n2 - n3 - n4
        let e1 = g.add_edge(n1, n2, 0);
        let e2 = g.add_edge(n2, n3, 0);
        let e3 = g.add_edge(n3, n4, 0);
        let comps = components(&g);
        assert_eq!(comps.len(), 1);
        assert_eq!(comps[0].0.len(), 4);
        assert_eq!(comps[0].1.len(), 3);
        assert!(comps[0].1.contains(&e1));
        assert!(comps[0].1.contains(&e2));
        assert!(comps[0].1.contains(&e3));
    }

    #[test]
    fn test_cycle() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        let n1 = g.add_node(1);
        let n2 = g.add_node(2);
        let n3 = g.add_node(3);
        // Create a triangle: n1 - n2 - n3 - n1
        let e1 = g.add_edge(n1, n2, 0);
        let e2 = g.add_edge(n2, n3, 0);
        let e3 = g.add_edge(n3, n1, 0);
        let comps = components(&g);
        assert_eq!(comps.len(), 1);
        assert_eq!(comps[0].0.len(), 3);
        assert_eq!(comps[0].1.len(), 3);
        assert!(comps[0].1.contains(&e1));
        assert!(comps[0].1.contains(&e2));
        assert!(comps[0].1.contains(&e3));
    }

    #[test]
    fn test_star_graph() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        let center = g.add_node(0);
        let n1 = g.add_node(1);
        let n2 = g.add_node(2);
        let n3 = g.add_node(3);
        let n4 = g.add_node(4);
        // Star: center connected to all others
        let e1 = g.add_edge(center, n1, 0);
        let e2 = g.add_edge(center, n2, 0);
        let e3 = g.add_edge(center, n3, 0);
        let e4 = g.add_edge(center, n4, 0);
        let comps = components(&g);
        assert_eq!(comps.len(), 1);
        assert_eq!(comps[0].0.len(), 5);
        assert_eq!(comps[0].1.len(), 4);
        assert!(comps[0].1.contains(&e1));
        assert!(comps[0].1.contains(&e2));
        assert!(comps[0].1.contains(&e3));
        assert!(comps[0].1.contains(&e4));
    }

    #[test]
    fn test_multiple_components() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        // Component 1: triangle
        let n1 = g.add_node(1);
        let n2 = g.add_node(2);
        let n3 = g.add_node(3);
        let e1 = g.add_edge(n1, n2, 0);
        let e2 = g.add_edge(n2, n3, 0);
        let e3 = g.add_edge(n3, n1, 0);

        // Component 2: two connected nodes
        let n4 = g.add_node(4);
        let n5 = g.add_node(5);
        let e4 = g.add_edge(n4, n5, 0);

        // Component 3: isolated node
        let n6 = g.add_node(6);

        let comps = components(&g);
        assert_eq!(comps.len(), 3);

        // Verify component node sizes
        let node_sizes: Vec<usize> = comps.iter().map(|c| c.0.len()).collect();
        assert!(node_sizes.contains(&3)); // triangle
        assert!(node_sizes.contains(&2)); // two connected
        assert!(node_sizes.contains(&1)); // isolated

        // Verify all nodes are present
        let all_nodes: Vec<_> = comps.iter().flat_map(|c| &c.0).collect();
        assert_eq!(all_nodes.len(), 6);

        // Verify n6 is in a component by itself
        let n6_component = comps.iter().find(|c| c.0.contains(&n6)).unwrap();
        assert_eq!(n6_component.0.len(), 1);
        assert_eq!(n6_component.1.len(), 0); // No edges

        // Verify triangle component has 3 edges
        let triangle_component = comps.iter().find(|c| c.0.contains(&n1)).unwrap();
        assert_eq!(triangle_component.1.len(), 3);
        assert!(triangle_component.1.contains(&e1));
        assert!(triangle_component.1.contains(&e2));
        assert!(triangle_component.1.contains(&e3));

        // Verify pair component has 1 edge
        let pair_component = comps.iter().find(|c| c.0.contains(&n4)).unwrap();
        assert_eq!(pair_component.1.len(), 1);
        assert!(pair_component.1.contains(&e4));
    }

    #[test]
    fn test_complete_graph() {
        let mut g: UnGraph<i32, i32> = UnGraph::new_undirected();
        let n1 = g.add_node(1);
        let n2 = g.add_node(2);
        let n3 = g.add_node(3);
        let n4 = g.add_node(4);
        // Complete graph K4
        let e1 = g.add_edge(n1, n2, 0);
        let e2 = g.add_edge(n1, n3, 0);
        let e3 = g.add_edge(n1, n4, 0);
        let e4 = g.add_edge(n2, n3, 0);
        let e5 = g.add_edge(n2, n4, 0);
        let e6 = g.add_edge(n3, n4, 0);
        let comps = components(&g);
        assert_eq!(comps.len(), 1);
        assert_eq!(comps[0].0.len(), 4);
        assert_eq!(comps[0].1.len(), 6); // K4 has 6 edges
        assert!(comps[0].1.contains(&e1));
        assert!(comps[0].1.contains(&e2));
        assert!(comps[0].1.contains(&e3));
        assert!(comps[0].1.contains(&e4));
        assert!(comps[0].1.contains(&e5));
        assert!(comps[0].1.contains(&e6));
    }
}